Compare commits
3 commits
cfcb06a6dc
...
93917f9fc2
| Author | SHA1 | Date | |
|---|---|---|---|
| 93917f9fc2 | |||
| 3fe0f4b744 | |||
| 05d15cea56 |
6 changed files with 485 additions and 51 deletions
|
|
@ -28,12 +28,9 @@ import sqlite3
|
||||||
import json
|
import json
|
||||||
|
|
||||||
|
|
||||||
# Map PR status to Clay's operation color palette
|
# Non-merged statuses map directly to operation — no semantic classification yet.
|
||||||
# extract (cyan), new (green), enrich (amber), challenge (red-orange),
|
NON_MERGED_STATUS_TO_OPERATION = {
|
||||||
# decision (violet), infra (grey)
|
'approved': 'new', # about to become knowledge
|
||||||
STATUS_TO_OPERATION = {
|
|
||||||
'merged': 'new', # green — new knowledge merged
|
|
||||||
'approved': 'enrich', # amber — approved, enriching KB
|
|
||||||
'open': 'extract', # cyan — new extraction in progress
|
'open': 'extract', # cyan — new extraction in progress
|
||||||
'validating': 'extract', # cyan — being validated
|
'validating': 'extract', # cyan — being validated
|
||||||
'reviewing': 'extract', # cyan — under review
|
'reviewing': 'extract', # cyan — under review
|
||||||
|
|
@ -43,6 +40,51 @@ STATUS_TO_OPERATION = {
|
||||||
'conflict': 'challenge', # red-orange — conflict detected
|
'conflict': 'challenge', # red-orange — conflict detected
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# Maintenance commit_types that land on main but don't represent new knowledge.
|
||||||
|
_MAINTENANCE_COMMIT_TYPES = {'fix', 'pipeline', 'reweave'}
|
||||||
|
|
||||||
|
|
||||||
|
def classify_pr_operation(status, commit_type, branch, description=None):
|
||||||
|
"""Derive a Timeline operation from a PR row.
|
||||||
|
|
||||||
|
Priority order for MERGED PRs (commit_type wins over branch prefix —
|
||||||
|
extract/* branches with commit_type='enrich' or 'challenge' classify
|
||||||
|
by commit_type, matching the contributor-role wiring fix):
|
||||||
|
1. commit_type == 'challenge' OR branch.startswith('challenge/') OR
|
||||||
|
description contains 'challenged_by' → 'challenge'
|
||||||
|
2. commit_type == 'enrich' OR branch.startswith('enrich/' | 'reweave/')
|
||||||
|
→ 'enrich'
|
||||||
|
3. commit_type in _MAINTENANCE_COMMIT_TYPES → 'infra'
|
||||||
|
4. default (commit_type='knowledge'|'extract'|'research'|'entity' or
|
||||||
|
anything else) → 'new'
|
||||||
|
|
||||||
|
For non-merged PRs, falls back to NON_MERGED_STATUS_TO_OPERATION.
|
||||||
|
"""
|
||||||
|
commit_type = (commit_type or '').lower()
|
||||||
|
branch = branch or ''
|
||||||
|
description_lower = (description or '').lower()
|
||||||
|
|
||||||
|
if status != 'merged':
|
||||||
|
return NON_MERGED_STATUS_TO_OPERATION.get(status, 'infra')
|
||||||
|
|
||||||
|
# Challenge takes precedence — the signal is inherently more specific.
|
||||||
|
if (commit_type == 'challenge'
|
||||||
|
or branch.startswith('challenge/')
|
||||||
|
or 'challenged_by' in description_lower):
|
||||||
|
return 'challenge'
|
||||||
|
|
||||||
|
if (commit_type == 'enrich'
|
||||||
|
or branch.startswith('enrich/')
|
||||||
|
or branch.startswith('reweave/')):
|
||||||
|
return 'enrich'
|
||||||
|
|
||||||
|
if commit_type in _MAINTENANCE_COMMIT_TYPES:
|
||||||
|
return 'infra'
|
||||||
|
|
||||||
|
# Default: legacy 'knowledge', new 'extract', 'research', 'entity',
|
||||||
|
# unknown/null commit_type → treat as new knowledge.
|
||||||
|
return 'new'
|
||||||
|
|
||||||
# Map audit_log stage to operation type
|
# Map audit_log stage to operation type
|
||||||
STAGE_TO_OPERATION = {
|
STAGE_TO_OPERATION = {
|
||||||
'ingest': 'extract',
|
'ingest': 'extract',
|
||||||
|
|
@ -118,6 +160,8 @@ async def handle_activity(request):
|
||||||
Query params:
|
Query params:
|
||||||
limit (int, default 100, max 500): number of events to return
|
limit (int, default 100, max 500): number of events to return
|
||||||
cursor (ISO timestamp): return events older than this timestamp
|
cursor (ISO timestamp): return events older than this timestamp
|
||||||
|
type (str, optional): comma-separated operation types to include
|
||||||
|
(extract|new|enrich|challenge|infra). If absent, returns all types.
|
||||||
|
|
||||||
Derives events from two sources:
|
Derives events from two sources:
|
||||||
1. prs table — per-PR events with domain, agent, status
|
1. prs table — per-PR events with domain, agent, status
|
||||||
|
|
@ -131,6 +175,13 @@ async def handle_activity(request):
|
||||||
limit = 100
|
limit = 100
|
||||||
|
|
||||||
cursor = request.query.get('cursor')
|
cursor = request.query.get('cursor')
|
||||||
|
type_param = request.query.get('type', '').strip()
|
||||||
|
allowed_ops = None
|
||||||
|
if type_param:
|
||||||
|
allowed_ops = {t.strip() for t in type_param.split(',') if t.strip()}
|
||||||
|
if not allowed_ops:
|
||||||
|
allowed_ops = None
|
||||||
|
|
||||||
db_path = request.app['db_path']
|
db_path = request.app['db_path']
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|
@ -143,22 +194,27 @@ async def handle_activity(request):
|
||||||
# Each PR generates events at created_at and merged_at timestamps
|
# Each PR generates events at created_at and merged_at timestamps
|
||||||
pr_query = """
|
pr_query = """
|
||||||
SELECT number, status, domain, agent, branch, source_path,
|
SELECT number, status, domain, agent, branch, source_path,
|
||||||
created_at, merged_at, source_channel
|
created_at, merged_at, source_channel, commit_type,
|
||||||
|
description
|
||||||
FROM prs
|
FROM prs
|
||||||
WHERE {where_clause}
|
WHERE {where_clause}
|
||||||
ORDER BY COALESCE(merged_at, created_at) DESC
|
ORDER BY COALESCE(merged_at, created_at) DESC
|
||||||
LIMIT ?
|
LIMIT ?
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
# Over-fetch when filtering by type so we have enough matching rows after
|
||||||
|
# post-build filtering. Cap at 2000 to avoid runaway queries.
|
||||||
|
fetch_limit = min(2000, limit * 5) if allowed_ops else limit + 1
|
||||||
|
|
||||||
if cursor:
|
if cursor:
|
||||||
rows = conn.execute(
|
rows = conn.execute(
|
||||||
pr_query.format(where_clause="COALESCE(merged_at, created_at) < ?"),
|
pr_query.format(where_clause="COALESCE(merged_at, created_at) < ?"),
|
||||||
(cursor, limit + 1)
|
(cursor, fetch_limit)
|
||||||
).fetchall()
|
).fetchall()
|
||||||
else:
|
else:
|
||||||
rows = conn.execute(
|
rows = conn.execute(
|
||||||
pr_query.format(where_clause="1=1"),
|
pr_query.format(where_clause="1=1"),
|
||||||
(limit + 1,)
|
(fetch_limit,)
|
||||||
).fetchall()
|
).fetchall()
|
||||||
|
|
||||||
# Known knowledge agents for branch-prefix inference
|
# Known knowledge agents for branch-prefix inference
|
||||||
|
|
@ -166,7 +222,14 @@ async def handle_activity(request):
|
||||||
|
|
||||||
for row in rows:
|
for row in rows:
|
||||||
row_dict = dict(row)
|
row_dict = dict(row)
|
||||||
operation = STATUS_TO_OPERATION.get(row_dict['status'], 'infra')
|
operation = classify_pr_operation(
|
||||||
|
row_dict['status'],
|
||||||
|
row_dict.get('commit_type'),
|
||||||
|
row_dict.get('branch'),
|
||||||
|
row_dict.get('description'),
|
||||||
|
)
|
||||||
|
if allowed_ops and operation not in allowed_ops:
|
||||||
|
continue
|
||||||
description = pr_description(row_dict)
|
description = pr_description(row_dict)
|
||||||
|
|
||||||
# Use merged_at if available (more interesting event), else created_at
|
# Use merged_at if available (more interesting event), else created_at
|
||||||
|
|
@ -218,6 +281,8 @@ async def handle_activity(request):
|
||||||
for row in audit_rows:
|
for row in audit_rows:
|
||||||
row_dict = dict(row)
|
row_dict = dict(row)
|
||||||
operation = STAGE_TO_OPERATION.get(row_dict['stage'], 'infra')
|
operation = STAGE_TO_OPERATION.get(row_dict['stage'], 'infra')
|
||||||
|
if allowed_ops and operation not in allowed_ops:
|
||||||
|
continue
|
||||||
description = audit_description(row_dict)
|
description = audit_description(row_dict)
|
||||||
|
|
||||||
events.append({
|
events.append({
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,32 @@ logger = logging.getLogger("pipeline.attribution")
|
||||||
|
|
||||||
VALID_ROLES = frozenset({"sourcer", "extractor", "challenger", "synthesizer", "reviewer"})
|
VALID_ROLES = frozenset({"sourcer", "extractor", "challenger", "synthesizer", "reviewer"})
|
||||||
|
|
||||||
|
# Handle sanity: lowercase alphanumerics, hyphens, underscores. 1-39 chars (matches
|
||||||
|
# GitHub's handle rules). Rejects garbage like "governance---meritocratic-voting-+-futarchy"
|
||||||
|
# or "sec-interpretive-release-s7-2026-09-(march-17" that upstream frontmatter hygiene
|
||||||
|
# bugs produce. Apply at parse time so bad handles never reach the contributors table.
|
||||||
|
_HANDLE_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,38}$")
|
||||||
|
|
||||||
|
|
||||||
|
def _valid_handle(handle: str) -> bool:
|
||||||
|
"""Return True if handle matches the handle format (alphanum + _-, ≤39 chars)."""
|
||||||
|
if not handle or not isinstance(handle, str):
|
||||||
|
return False
|
||||||
|
h = handle.strip().lower().lstrip("@")
|
||||||
|
if h.endswith("-") or h.endswith("_"):
|
||||||
|
return False
|
||||||
|
return bool(_HANDLE_RE.match(h))
|
||||||
|
|
||||||
|
|
||||||
|
def _filter_valid_handles(result: dict) -> dict:
|
||||||
|
"""Drop entries with invalid handles from a parsed attribution dict."""
|
||||||
|
filtered: dict[str, list[dict]] = {role: [] for role in VALID_ROLES}
|
||||||
|
for role, entries in result.items():
|
||||||
|
for entry in entries:
|
||||||
|
if _valid_handle(entry.get("handle", "")):
|
||||||
|
filtered[role].append(entry)
|
||||||
|
return filtered
|
||||||
|
|
||||||
|
|
||||||
# ─── Parse attribution from claim content ──────────────────────────────────
|
# ─── Parse attribution from claim content ──────────────────────────────────
|
||||||
|
|
||||||
|
|
@ -51,7 +77,11 @@ def parse_attribution(fm: dict) -> dict[str, list[dict]]:
|
||||||
elif isinstance(entries, str):
|
elif isinstance(entries, str):
|
||||||
# Single entry as string
|
# Single entry as string
|
||||||
result[role].append({"handle": entries.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
result[role].append({"handle": entries.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
||||||
return result
|
# Fall through to the filter at the end (don't early-return). The nested
|
||||||
|
# block path was skipping the handle sanity filter, letting garbage like
|
||||||
|
# "senator-elissa-slotkin-/-the-hill" through when it was written into
|
||||||
|
# frontmatter during the legacy-fallback era.
|
||||||
|
return _filter_valid_handles(result)
|
||||||
|
|
||||||
# Flat format fallback (attribution_sourcer, attribution_extractor, etc.)
|
# Flat format fallback (attribution_sourcer, attribution_extractor, etc.)
|
||||||
for role in VALID_ROLES:
|
for role in VALID_ROLES:
|
||||||
|
|
@ -64,22 +94,40 @@ def parse_attribution(fm: dict) -> dict[str, list[dict]]:
|
||||||
if isinstance(v, str):
|
if isinstance(v, str):
|
||||||
result[role].append({"handle": v.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
result[role].append({"handle": v.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
||||||
|
|
||||||
# Legacy fallback: infer from source field
|
# Bare-key flat format: `sourcer: alexastrum`, `extractor: leo`, etc.
|
||||||
if not any(result[r] for r in VALID_ROLES):
|
# This is what extract.py writes (line 290: f'sourcer: "{sourcer}"') — the most
|
||||||
source = fm.get("source", "")
|
# common format in practice (~42% of claim files). The Apr 24 incident traced
|
||||||
if isinstance(source, str) and source:
|
# missing leaderboard entries to this format being silently dropped because the
|
||||||
# Try to extract author handle from source string
|
# parser only checked the `attribution_*` prefix.
|
||||||
# Patterns: "@handle", "Author Name", "org, description"
|
# Only fill if the role wasn't already populated by the prefixed form, to avoid
|
||||||
handle_match = re.search(r"@(\w+)", source)
|
# double-counting when both formats coexist on the same claim.
|
||||||
if handle_match:
|
for role in VALID_ROLES:
|
||||||
result["sourcer"].append({"handle": handle_match.group(1).lower(), "agent_id": None, "context": source})
|
if result[role]:
|
||||||
else:
|
continue
|
||||||
# Use first word/phrase before comma as sourcer handle
|
bare_val = fm.get(role)
|
||||||
author = source.split(",")[0].strip().lower().replace(" ", "-")
|
if isinstance(bare_val, str) and bare_val.strip():
|
||||||
if author and len(author) > 1:
|
result[role].append({"handle": bare_val.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
||||||
result["sourcer"].append({"handle": author, "agent_id": None, "context": source})
|
elif isinstance(bare_val, list):
|
||||||
|
for v in bare_val:
|
||||||
|
if isinstance(v, str) and v.strip():
|
||||||
|
result[role].append({"handle": v.strip().lower().lstrip("@"), "agent_id": None, "context": None})
|
||||||
|
elif isinstance(v, dict) and v.get("handle"):
|
||||||
|
result[role].append({
|
||||||
|
"handle": v["handle"].strip().lower().lstrip("@"),
|
||||||
|
"agent_id": v.get("agent_id"),
|
||||||
|
"context": v.get("context"),
|
||||||
|
})
|
||||||
|
|
||||||
return result
|
# Legacy `source` heuristic REMOVED (Ganymede review, Apr 24). It fabricated
|
||||||
|
# handles from descriptive source strings — "governance---meritocratic-voting-+-
|
||||||
|
# futarchy", "cameron-(contributor)", "sec-interpretive-release-s7-2026-09-
|
||||||
|
# (march-17". Hit rate on real handles was near-zero, false-positive rate was
|
||||||
|
# high. Claims without explicit attribution now return empty (better surface as
|
||||||
|
# data hygiene than invent fake contributors).
|
||||||
|
|
||||||
|
# Filter to valid handles only. Bad handles (garbage from upstream frontmatter
|
||||||
|
# bugs) get dropped rather than written to the contributors table.
|
||||||
|
return _filter_valid_handles(result)
|
||||||
|
|
||||||
|
|
||||||
def parse_attribution_from_file(filepath: str) -> dict[str, list[dict]]:
|
def parse_attribution_from_file(filepath: str) -> dict[str, list[dict]]:
|
||||||
|
|
|
||||||
|
|
@ -148,27 +148,50 @@ async def record_contributor_attribution(conn, pr_number: int, branch: str, git_
|
||||||
)
|
)
|
||||||
agents_found.add(agent_name)
|
agents_found.add(agent_name)
|
||||||
|
|
||||||
# Parse attribution blocks from claim frontmatter in diff
|
# Parse attribution from NEWLY ADDED knowledge files via the canonical attribution
|
||||||
# Look for added lines with attribution YAML
|
# parser (lib/attribution.py). The previous diff-line regex parser dropped
|
||||||
current_role = None
|
# both the bare-key flat format (`sourcer: alexastrum`) and the nested
|
||||||
for line in diff.split("\n"):
|
# `attribution:` block format because it only matched `- handle: "X"` lines.
|
||||||
if not line.startswith("+") or line.startswith("+++"):
|
# The Apr 24 incident traced missing leaderboard entries (alexastrum=0,
|
||||||
continue
|
# thesensatore=0, cameron-s1=0) directly to this parser's blind spots.
|
||||||
stripped = line[1:].strip()
|
#
|
||||||
|
# --diff-filter=A restricts to added files only (Ganymede review): enrich and
|
||||||
|
# challenge PRs modify existing claims, and re-crediting the existing sourcer on
|
||||||
|
# every modification would inflate counts. The synthesizer/challenger/reviewer
|
||||||
|
# roles for those PRs are credited via the Pentagon-Agent trailer path above.
|
||||||
|
rc_files, files_output = await git_fn(
|
||||||
|
"diff", "--name-only", "--diff-filter=A",
|
||||||
|
f"origin/main...origin/{branch}", timeout=10,
|
||||||
|
)
|
||||||
|
if rc_files == 0 and files_output:
|
||||||
|
from pathlib import Path
|
||||||
|
from . import config
|
||||||
|
from .attribution import parse_attribution_from_file
|
||||||
|
|
||||||
# Detect role sections in attribution block
|
main_root = Path(config.MAIN_WORKTREE)
|
||||||
for role in ("sourcer", "extractor", "challenger", "synthesizer", "reviewer"):
|
# Match is_knowledge_pr's gate exactly. Entities/convictions are excluded
|
||||||
if stripped.startswith(f"{role}:"):
|
# here because is_knowledge_pr skips entity-only PRs at line 123 — so a
|
||||||
current_role = role
|
# broader list here only matters for mixed PRs where the narrower list
|
||||||
break
|
# already matches via the claim file. Widening requires Cory sign-off
|
||||||
|
# since it would change leaderboard accounting (entity-only PRs → CI credit).
|
||||||
# Extract handle from attribution entries
|
knowledge_prefixes = ("domains/", "core/", "foundations/", "decisions/")
|
||||||
handle_match = re.match(r'-\s*handle:\s*["\']?([^"\']+)["\']?', stripped)
|
for rel_path in files_output.strip().split("\n"):
|
||||||
if handle_match and current_role:
|
rel_path = rel_path.strip()
|
||||||
handle = handle_match.group(1).strip().lower()
|
if not rel_path.endswith(".md"):
|
||||||
agent_id_match = re.search(r'agent_id:\s*["\']?([^"\']+)', stripped)
|
continue
|
||||||
agent_id = agent_id_match.group(1).strip() if agent_id_match else None
|
if not rel_path.startswith(knowledge_prefixes):
|
||||||
upsert_contributor(conn, handle, agent_id, current_role, today)
|
continue
|
||||||
|
full = main_root / rel_path
|
||||||
|
if not full.exists():
|
||||||
|
continue # file removed in this PR
|
||||||
|
attribution = parse_attribution_from_file(str(full))
|
||||||
|
for role, entries in attribution.items():
|
||||||
|
for entry in entries:
|
||||||
|
handle = entry.get("handle")
|
||||||
|
if handle:
|
||||||
|
upsert_contributor(
|
||||||
|
conn, handle, entry.get("agent_id"), role, today,
|
||||||
|
)
|
||||||
|
|
||||||
# Fallback: if no Pentagon-Agent trailer found, try git commit authors
|
# Fallback: if no Pentagon-Agent trailer found, try git commit authors
|
||||||
_BOT_AUTHORS = frozenset({
|
_BOT_AUTHORS = frozenset({
|
||||||
|
|
|
||||||
22
lib/db.py
22
lib/db.py
|
|
@ -232,9 +232,20 @@ def classify_branch(branch: str) -> tuple[str, str]:
|
||||||
|
|
||||||
|
|
||||||
# Keep in sync with BRANCH_PREFIX_MAP above.
|
# Keep in sync with BRANCH_PREFIX_MAP above.
|
||||||
|
#
|
||||||
|
# Valid source_channel values: github | telegram | agent | maintenance | web | unknown
|
||||||
|
# - github: external contributor PR (set via sync-mirror.sh github_pr linking,
|
||||||
|
# or from gh-pr-* branches, or any time github_pr is provided)
|
||||||
|
# - telegram: message captured by telegram bot (must be tagged explicitly by
|
||||||
|
# ingestion — extract/* default is "unknown" because the bare branch prefix
|
||||||
|
# can no longer distinguish telegram-origin from github-origin extractions)
|
||||||
|
# - agent: per-agent research branches (rio/, theseus/, etc.)
|
||||||
|
# - maintenance: pipeline housekeeping (reweave/, epimetheus/, fix/)
|
||||||
|
# - web: future in-app submissions (chat UI or form posts)
|
||||||
|
# - unknown: fallback when provenance cannot be determined
|
||||||
_CHANNEL_MAP = {
|
_CHANNEL_MAP = {
|
||||||
"extract": "telegram",
|
"extract": "unknown",
|
||||||
"ingestion": "telegram",
|
"ingestion": "unknown",
|
||||||
"rio": "agent",
|
"rio": "agent",
|
||||||
"theseus": "agent",
|
"theseus": "agent",
|
||||||
"astra": "agent",
|
"astra": "agent",
|
||||||
|
|
@ -249,7 +260,12 @@ _CHANNEL_MAP = {
|
||||||
|
|
||||||
|
|
||||||
def classify_source_channel(branch: str, *, github_pr: int = None) -> str:
|
def classify_source_channel(branch: str, *, github_pr: int = None) -> str:
|
||||||
"""Derive source_channel from branch prefix and github_pr flag."""
|
"""Derive source_channel from branch prefix and github_pr flag.
|
||||||
|
|
||||||
|
Precedence: github_pr flag > gh-pr- branch prefix > _CHANNEL_MAP lookup.
|
||||||
|
extract/* defaults to "unknown" — callers with better provenance (telegram
|
||||||
|
bot, web submission handler) must override at PR-insert time.
|
||||||
|
"""
|
||||||
if github_pr is not None or branch.startswith("gh-pr-"):
|
if github_pr is not None or branch.startswith("gh-pr-"):
|
||||||
return "github"
|
return "github"
|
||||||
prefix = branch.split("/", 1)[0] if "/" in branch else branch
|
prefix = branch.split("/", 1)[0] if "/" in branch else branch
|
||||||
|
|
|
||||||
261
scripts/backfill-sourcer-attribution.py
Executable file
261
scripts/backfill-sourcer-attribution.py
Executable file
|
|
@ -0,0 +1,261 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Backfill sourcer/extractor/etc. attribution from claim frontmatter.
|
||||||
|
|
||||||
|
Walks every merged knowledge file under domains/, entities/, decisions/,
|
||||||
|
foundations/, convictions/, core/ and re-runs the canonical attribution
|
||||||
|
parser (lib/attribution.py). For each parsed (handle, role) pair, increments
|
||||||
|
the corresponding *_count column on the contributors table.
|
||||||
|
|
||||||
|
Why this is needed (Apr 24 incident):
|
||||||
|
- lib/contributor.py used a diff-line regex parser that handled neither
|
||||||
|
the bare-key flat format (`sourcer: alexastrum`, ~42% of claims) nor
|
||||||
|
the nested `attribution: { sourcer: [...] }` block format used by Leo's
|
||||||
|
manual extractions (Shaga's claims).
|
||||||
|
- Result: alexastrum, thesensatore, cameron-s1, and similar handles were
|
||||||
|
silently dropped at merge time. Their contributor rows either don't
|
||||||
|
exist or are stuck at zero counts.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
python3 backfill-sourcer-attribution.py --dry-run # report deltas, no writes
|
||||||
|
python3 backfill-sourcer-attribution.py # apply (additive: max(db, truth))
|
||||||
|
python3 backfill-sourcer-attribution.py --reset # destructive: set absolute truth
|
||||||
|
|
||||||
|
Default mode is ADDITIVE for safety: per-role count is set to max(current_db, truth).
|
||||||
|
This preserves any existing high counts that came from non-frontmatter sources
|
||||||
|
(e.g., m3taversal.sourcer=1011 reflects Telegram-curator credit accumulated via
|
||||||
|
a different code path; truncating to the file-walk truth would be destructive).
|
||||||
|
|
||||||
|
Use --reset to set absolute truth from the file walk only — this clobbers
|
||||||
|
all existing role counts including legitimate non-frontmatter credit.
|
||||||
|
|
||||||
|
Idempotency: additive mode is safe to re-run. --reset run is gated by an
|
||||||
|
audit_log marker; pass --force to override.
|
||||||
|
"""
|
||||||
|
import argparse
|
||||||
|
import os
|
||||||
|
import sqlite3
|
||||||
|
import sys
|
||||||
|
from collections import defaultdict
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Allow running from anywhere — point at pipeline lib
|
||||||
|
PIPELINE_ROOT = Path(__file__).resolve().parent.parent
|
||||||
|
sys.path.insert(0, str(PIPELINE_ROOT))
|
||||||
|
|
||||||
|
from lib.attribution import parse_attribution_from_file, VALID_ROLES # noqa: E402
|
||||||
|
|
||||||
|
DB_PATH = os.environ.get("PIPELINE_DB", "/opt/teleo-eval/pipeline/pipeline.db")
|
||||||
|
REPO = Path(os.environ.get("REPO_DIR", "/opt/teleo-eval/workspaces/main"))
|
||||||
|
KNOWLEDGE_PREFIXES = (
|
||||||
|
"domains", "entities", "decisions", "foundations", "convictions", "core",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def collect_attributions(repo_root: Path) -> dict[str, dict[str, int]]:
|
||||||
|
"""Walk all knowledge files; return {handle: {role: count}}."""
|
||||||
|
counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
||||||
|
files_scanned = 0
|
||||||
|
files_with_attribution = 0
|
||||||
|
|
||||||
|
for prefix in KNOWLEDGE_PREFIXES:
|
||||||
|
base = repo_root / prefix
|
||||||
|
if not base.exists():
|
||||||
|
continue
|
||||||
|
for path in base.rglob("*.md"):
|
||||||
|
if path.name.startswith("_"):
|
||||||
|
continue
|
||||||
|
files_scanned += 1
|
||||||
|
attr = parse_attribution_from_file(str(path))
|
||||||
|
had_any = False
|
||||||
|
for role, entries in attr.items():
|
||||||
|
for entry in entries:
|
||||||
|
handle = entry.get("handle")
|
||||||
|
if handle:
|
||||||
|
counts[handle][role] += 1
|
||||||
|
had_any = True
|
||||||
|
if had_any:
|
||||||
|
files_with_attribution += 1
|
||||||
|
|
||||||
|
print(f" Scanned {files_scanned} knowledge files", file=sys.stderr)
|
||||||
|
print(f" {files_with_attribution} had parseable attribution", file=sys.stderr)
|
||||||
|
return counts
|
||||||
|
|
||||||
|
|
||||||
|
def existing_contributors(conn) -> dict[str, dict[str, int]]:
|
||||||
|
"""Return {handle: {role: count}} from the current DB."""
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT handle, sourcer_count, extractor_count, challenger_count, "
|
||||||
|
"synthesizer_count, reviewer_count, claims_merged FROM contributors"
|
||||||
|
).fetchall()
|
||||||
|
out = {}
|
||||||
|
for r in rows:
|
||||||
|
out[r["handle"]] = {
|
||||||
|
"sourcer": r["sourcer_count"] or 0,
|
||||||
|
"extractor": r["extractor_count"] or 0,
|
||||||
|
"challenger": r["challenger_count"] or 0,
|
||||||
|
"synthesizer": r["synthesizer_count"] or 0,
|
||||||
|
"reviewer": r["reviewer_count"] or 0,
|
||||||
|
"claims_merged": r["claims_merged"] or 0,
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def claims_merged_for(role_counts: dict[str, int]) -> int:
|
||||||
|
"""Mirror upsert_contributor logic: claims_merged += sourcer + extractor."""
|
||||||
|
return role_counts.get("sourcer", 0) + role_counts.get("extractor", 0)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
parser.add_argument("--dry-run", action="store_true",
|
||||||
|
help="Report deltas without writing")
|
||||||
|
parser.add_argument("--reset", action="store_true",
|
||||||
|
help="Destructive: set absolute truth from file walk "
|
||||||
|
"(default is additive max(db, truth))")
|
||||||
|
parser.add_argument("--force", action="store_true",
|
||||||
|
help="Re-run even if a previous --reset marker exists")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
if not REPO.exists():
|
||||||
|
print(f"ERROR: repo not found at {REPO}", file=sys.stderr)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
print(f"DB: {DB_PATH}", file=sys.stderr)
|
||||||
|
print(f"Repo: {REPO}", file=sys.stderr)
|
||||||
|
print("", file=sys.stderr)
|
||||||
|
print("Walking knowledge tree...", file=sys.stderr)
|
||||||
|
|
||||||
|
truth = collect_attributions(REPO)
|
||||||
|
print(f" Found attributions for {len(truth)} unique handles", file=sys.stderr)
|
||||||
|
print("", file=sys.stderr)
|
||||||
|
|
||||||
|
conn = sqlite3.connect(DB_PATH, timeout=30)
|
||||||
|
conn.row_factory = sqlite3.Row
|
||||||
|
current = existing_contributors(conn)
|
||||||
|
|
||||||
|
# Compute deltas: new handles + handles with role-count mismatches
|
||||||
|
new_handles: list[tuple[str, dict[str, int]]] = []
|
||||||
|
role_deltas: list[tuple[str, dict[str, int], dict[str, int]]] = []
|
||||||
|
|
||||||
|
for handle, roles in truth.items():
|
||||||
|
if handle not in current:
|
||||||
|
new_handles.append((handle, dict(roles)))
|
||||||
|
else:
|
||||||
|
cur = current[handle]
|
||||||
|
mismatches = {r: roles.get(r, 0) for r in VALID_ROLES
|
||||||
|
if roles.get(r, 0) != cur.get(r, 0)}
|
||||||
|
if mismatches:
|
||||||
|
role_deltas.append((handle, dict(roles), cur))
|
||||||
|
|
||||||
|
print(f"=== {len(new_handles)} NEW contributors to insert ===")
|
||||||
|
for handle, roles in sorted(new_handles, key=lambda x: -sum(x[1].values()))[:20]:
|
||||||
|
roles_str = ", ".join(f"{r}={c}" for r, c in roles.items() if c > 0)
|
||||||
|
print(f" + {handle}: {roles_str} (claims_merged={claims_merged_for(roles)})")
|
||||||
|
if len(new_handles) > 20:
|
||||||
|
print(f" ... and {len(new_handles) - 20} more")
|
||||||
|
print()
|
||||||
|
|
||||||
|
print(f"=== {len(role_deltas)} EXISTING contributors with count drift ===")
|
||||||
|
for handle, truth_roles, cur_roles in sorted(
|
||||||
|
role_deltas,
|
||||||
|
key=lambda x: -sum(x[1].values()),
|
||||||
|
)[:20]:
|
||||||
|
for role in VALID_ROLES:
|
||||||
|
t = truth_roles.get(role, 0)
|
||||||
|
c = cur_roles.get(role, 0)
|
||||||
|
if t != c:
|
||||||
|
print(f" ~ {handle}.{role}: db={c} → truth={t} (Δ{t - c:+d})")
|
||||||
|
if len(role_deltas) > 20:
|
||||||
|
print(f" ... and {len(role_deltas) - 20} more")
|
||||||
|
print()
|
||||||
|
|
||||||
|
if args.dry_run:
|
||||||
|
mode = "RESET" if args.reset else "ADDITIVE"
|
||||||
|
print(f"Dry run ({mode} mode) — no changes written.")
|
||||||
|
if not args.reset:
|
||||||
|
print("Default is ADDITIVE: existing high counts (e.g. m3taversal=1011) preserved.")
|
||||||
|
print("Pass --reset to clobber existing counts with file-walk truth.")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Idempotency: --reset is gated by audit marker. Additive mode is always safe.
|
||||||
|
if args.reset:
|
||||||
|
marker = conn.execute(
|
||||||
|
"SELECT 1 FROM audit_log WHERE event = 'sourcer_attribution_backfill_reset' LIMIT 1"
|
||||||
|
).fetchone()
|
||||||
|
if marker and not args.force:
|
||||||
|
print("ERROR: --reset has already run (audit marker present).")
|
||||||
|
print("Pass --force to re-run.")
|
||||||
|
sys.exit(2)
|
||||||
|
|
||||||
|
inserted = 0
|
||||||
|
updated = 0
|
||||||
|
preserved_higher = 0
|
||||||
|
for handle, roles in truth.items():
|
||||||
|
truth_counts = {
|
||||||
|
"sourcer": roles.get("sourcer", 0),
|
||||||
|
"extractor": roles.get("extractor", 0),
|
||||||
|
"challenger": roles.get("challenger", 0),
|
||||||
|
"synthesizer": roles.get("synthesizer", 0),
|
||||||
|
"reviewer": roles.get("reviewer", 0),
|
||||||
|
}
|
||||||
|
|
||||||
|
if handle in current:
|
||||||
|
cur = current[handle]
|
||||||
|
if args.reset:
|
||||||
|
# Preserve reviewer_count even on reset (PR-level not file-level)
|
||||||
|
final = dict(truth_counts)
|
||||||
|
final["reviewer"] = max(truth_counts["reviewer"], cur.get("reviewer", 0))
|
||||||
|
else:
|
||||||
|
# Additive: max of db vs truth, per role
|
||||||
|
final = {
|
||||||
|
role: max(truth_counts[role], cur.get(role, 0))
|
||||||
|
for role in truth_counts
|
||||||
|
}
|
||||||
|
if any(cur.get(r, 0) > truth_counts[r] for r in truth_counts):
|
||||||
|
preserved_higher += 1
|
||||||
|
|
||||||
|
cm = final["sourcer"] + final["extractor"]
|
||||||
|
conn.execute(
|
||||||
|
"""UPDATE contributors SET
|
||||||
|
sourcer_count = ?,
|
||||||
|
extractor_count = ?,
|
||||||
|
challenger_count = ?,
|
||||||
|
synthesizer_count = ?,
|
||||||
|
reviewer_count = ?,
|
||||||
|
claims_merged = ?,
|
||||||
|
updated_at = datetime('now')
|
||||||
|
WHERE handle = ?""",
|
||||||
|
(final["sourcer"], final["extractor"], final["challenger"],
|
||||||
|
final["synthesizer"], final["reviewer"], cm, handle),
|
||||||
|
)
|
||||||
|
updated += 1
|
||||||
|
else:
|
||||||
|
cm = truth_counts["sourcer"] + truth_counts["extractor"]
|
||||||
|
conn.execute(
|
||||||
|
"""INSERT INTO contributors (
|
||||||
|
handle, sourcer_count, extractor_count, challenger_count,
|
||||||
|
synthesizer_count, reviewer_count, claims_merged,
|
||||||
|
first_contribution, last_contribution, tier
|
||||||
|
) VALUES (?, ?, ?, ?, ?, ?, ?, date('now'), date('now'), 'new')""",
|
||||||
|
(handle, truth_counts["sourcer"], truth_counts["extractor"],
|
||||||
|
truth_counts["challenger"], truth_counts["synthesizer"],
|
||||||
|
truth_counts["reviewer"], cm),
|
||||||
|
)
|
||||||
|
inserted += 1
|
||||||
|
|
||||||
|
event = "sourcer_attribution_backfill_reset" if args.reset else "sourcer_attribution_backfill"
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO audit_log (stage, event, detail) VALUES (?, ?, ?)",
|
||||||
|
("contributor", event,
|
||||||
|
f'{{"inserted": {inserted}, "updated": {updated}, '
|
||||||
|
f'"preserved_higher": {preserved_higher}, "mode": '
|
||||||
|
f'"{"reset" if args.reset else "additive"}"}}'),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
print(f"Done ({'RESET' if args.reset else 'ADDITIVE'}). "
|
||||||
|
f"Inserted {inserted} new, updated {updated} existing, "
|
||||||
|
f"preserved {preserved_higher} higher-than-truth values.")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -34,13 +34,34 @@ class TestParseAttribution:
|
||||||
assert result["extractor"][0]["handle"] == "rio"
|
assert result["extractor"][0]["handle"] == "rio"
|
||||||
assert result["sourcer"][0]["handle"] == "theiaresearch"
|
assert result["sourcer"][0]["handle"] == "theiaresearch"
|
||||||
|
|
||||||
def test_legacy_source_fallback(self):
|
def test_legacy_source_fallback_removed(self):
|
||||||
|
"""Legacy `source` heuristic removed (Ganymede review, Apr 24).
|
||||||
|
|
||||||
|
It fabricated handles from descriptive strings (garbage like
|
||||||
|
'sec-interpretive-release-s7-2026-09-(march-17'). Claims without
|
||||||
|
explicit attribution now return empty — better to surface as data
|
||||||
|
hygiene than invent contributors.
|
||||||
|
"""
|
||||||
fm = {
|
fm = {
|
||||||
"type": "claim",
|
"type": "claim",
|
||||||
"source": "@pineanalytics, Q4 2025 report",
|
"source": "@pineanalytics, Q4 2025 report",
|
||||||
}
|
}
|
||||||
result = parse_attribution(fm)
|
result = parse_attribution(fm)
|
||||||
assert result["sourcer"][0]["handle"] == "pineanalytics"
|
assert all(len(v) == 0 for v in result.values())
|
||||||
|
|
||||||
|
def test_bad_handles_filtered(self):
|
||||||
|
"""Handles with spaces, parens, or garbage chars are dropped."""
|
||||||
|
fm = {
|
||||||
|
"sourcer": "governance---meritocratic-voting-+-futarchy",
|
||||||
|
}
|
||||||
|
result = parse_attribution(fm)
|
||||||
|
assert len(result["sourcer"]) == 0
|
||||||
|
|
||||||
|
def test_valid_handle_with_hyphen_passes(self):
|
||||||
|
"""Legitimate handles like 'cameron-s1' survive the filter."""
|
||||||
|
fm = {"sourcer": "cameron-s1"}
|
||||||
|
result = parse_attribution(fm)
|
||||||
|
assert result["sourcer"][0]["handle"] == "cameron-s1"
|
||||||
|
|
||||||
def test_empty_attribution(self):
|
def test_empty_attribution(self):
|
||||||
fm = {"type": "claim"}
|
fm = {"type": "claim"}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue