diff --git a/.gitignore b/.gitignore
index d69c53e..e431ce5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,6 +27,11 @@ yarn-error.log*
 pnpm-debug.log*
 .npm/
 
+# --- Python (bundled skill scripts, e.g. govkit-metrics-emit) ---
+__pycache__/
+*.py[cod]
+.pytest_cache/
+
 # --- Secrets / environment ---
 .env
 .env.*
diff --git a/plugins/govkit/.claude-plugin/plugin.json b/plugins/govkit/.claude-plugin/plugin.json
index 718c9f6..02fffd6 100644
--- a/plugins/govkit/.claude-plugin/plugin.json
+++ b/plugins/govkit/.claude-plugin/plugin.json
@@ -1,7 +1,7 @@
 {
   "name": "govkit",
   "description": "GovKit skills for governed, AI-assisted delivery: refine Draft 0 feature specs (Gherkin + NFRs + evaluation criteria) before coding begins.",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "author": { "name": "Accelerated Innovation" },
   "homepage": "https://acceleratedinnovation.com",
   "repository": "https://github.com/Accelerated-Innovation/govkit-plugins",
diff --git a/plugins/govkit/skills/govkit-metrics-emit/SKILL.md b/plugins/govkit/skills/govkit-metrics-emit/SKILL.md
new file mode 100644
index 0000000..c163a1f
--- /dev/null
+++ b/plugins/govkit/skills/govkit-metrics-emit/SKILL.md
@@ -0,0 +1,92 @@
+---
+name: govkit-metrics-emit
+description: >
+  Emit structured Tier 1 metric events (NDJSON) from a GovKit-governed repository's
+  exhaust — feature packages, .govkit/marker.json, eval_criteria.yaml, CI gate runs,
+  PR exports, and git history. Use this skill whenever the user wants to compute,
+  emit, export, or audit delivery/quality metrics from a GovKit repo: spec
+  completeness scores, gate-readiness audits, metric event streams for an
+  aggregator, velocity/quality pair inputs, or "which features aren't gate-ready."
+  Trigger on mentions of GovKit metrics, metric events, telemetry emission,
+  feature package audit, spec completeness, or AIPOS Tier 1 metrics — even if the
+  user doesn't name this skill.
+---
+
+# govkit-metrics-emit
+
+Producer side of the AIPOS federated metrics topology. Reads GovKit artifacts and
+emits **org-agnostic** structured metric events as NDJSON. Aggregation (baselines,
+targets, benchmarks, trends) happens elsewhere — never here.
+
+## The one law that governs everything here
+
+**Producers stay org-agnostic.** Events must contain only repo-relative facts:
+feature IDs, file names, scores, timestamps, commit SHAs, gate names. Never emit
+organization names, employee names/emails beyond what git history inherently
+carries, internal URLs, ticket-system references, targets, or baselines. Those
+belong to the aggregator. If the user asks to add org context to events, explain
+the federated split and offer to put it in an aggregator-side enrichment note
+instead.
+
+## How to run
+
+Everything is done by the bundled script — do not hand-roll parsers:
+
+```bash
+python scripts/emit_metrics.py <repo_path> \
+  [--ci-runs <ci_runs.json>] \
+  [--prs <prs.json>] \
+  [--rework-days 21] \
+  [--out events.ndjson] \
+  [--validate]
+```
+
+- `<repo_path>`: root of a GovKit-governed repo (has `features/` and ideally `.govkit/marker.json`).
+- `--ci-runs`: optional JSON export of CI workflow runs (`gh run list --json name,conclusion,createdAt,updatedAt,headSha,attempt,displayTitle,headBranch` or an Azure equivalent list). Produces `gate.run.completed` events.
+- `--prs`: optional JSON export of merged PRs (`gh pr list --state merged --json number,title,createdAt,mergedAt,additions,deletions,changedFiles,reviews,commits,headRefName`). Produces `pr.merged` events.
+- `--rework-days`: window for the rework scan (default 21, per catalog).
+- `--validate`: after emitting, check every event against the vocabulary and print a validation summary (event counts, unknown fields, pairing coverage).
+
+Without any optional inputs the script still emits `feature.package.snapshot`
+(one per feature, from the working tree + git history) and `rework.observed`
+(from git history alone). That is the minimum useful run.
+
+## Event vocabulary (catalog v1)
+
+Five events: `feature.package.snapshot`, `pr.merged`, `gate.run.completed`,
+`deploy.completed`, `rework.observed`. Full field definitions, the spec
+completeness rubric (0–100, seven components), and which Tier 1 metric pair each
+event feeds are in `references/event_schema.md` — read it when the user asks what
+a field means, questions a score, or wants to map events to metrics.
+
+`deploy.completed` is defined in the vocabulary but this producer does not emit
+it (deployment events originate in CI/CD, not in GovKit artifacts). Say so if
+asked, and point to the CI-side adapter as the emitter.
+
+## Interpreting results for the user
+
+- **Spec completeness score** is the leading quality counterweight (Pair 4).
+  100 = gate-ready. When reporting it, always show the per-component breakdown
+  from `completeness.components` — the number alone hides *what's* missing.
+- A feature with `thresholds_met: false` or null FIRST/Virtue scores is not a
+  parsing problem; it's an author who hasn't finished the Evaluation Compliance
+  Summary in `plan.md`. Report it that way.
+- **Rework** uses a file-overlap approximation (documented in the reference).
+  Present it as an upper bound and say so; exact line provenance is the
+  aggregator's v2 job.
+- Never present a velocity number without its quality counterweight (paired-metric
+  law). If the user asks for "just throughput," give the pair and explain briefly.
+
+## Common tasks
+
+**"Which features aren't gate-ready?"** — run the script, filter snapshots where
+`completeness.score < 100`, and present a table: feature, score, missing
+components (from the breakdown), plus the emitted events file path.
+
+**"Generate events for the aggregator"** — run with all available inputs and
+`--validate`, save NDJSON, report event counts per type and the validation
+summary.
+
+**"Is this output safe to share outside the org?"** — run `--validate`, then
+grep the NDJSON for anything org-identifying; the only identity-adjacent content
+should be git author fields inside standard git metadata. Report findings.
diff --git a/plugins/govkit/skills/govkit-metrics-emit/evals/evals.json b/plugins/govkit/skills/govkit-metrics-emit/evals/evals.json
new file mode 100644
index 0000000..dffa89c
--- /dev/null
+++ b/plugins/govkit/skills/govkit-metrics-emit/evals/evals.json
@@ -0,0 +1,26 @@
+{
+  "skill_name": "govkit-metrics-emit",
+  "evals": [
+    {
+      "id": 0,
+      "name": "gate-readiness-audit",
+      "prompt": "I've got a GovKit-governed repo at /tmp/testrepo. Which features aren't gate-ready? Give me each feature's spec completeness score with what's missing, and save the metric events (NDJSON) plus a short audit report (markdown) to the output directory.",
+      "expected_output": "events.ndjson with feature.package.snapshot events; audit report showing checkout_assistant=100 (gate-ready) and incomplete_feature=15 with named missing rubric components",
+      "files": ["/tmp/testrepo"]
+    },
+    {
+      "id": 1,
+      "name": "aggregator-export",
+      "prompt": "Generate the Tier 1 metric event stream for our aggregator from the GovKit repo at /tmp/testrepo — the CI runs export is at /tmp/testrepo/ci_runs.json and the merged-PR export at /tmp/testrepo/prs.json. Validate the events against the vocabulary and save events.ndjson plus a summary of event counts per type to the output directory.",
+      "expected_output": "events.ndjson with 10 events (2 snapshots, 2 rework, 4 gate runs, 2 PRs), validation passing, count summary",
+      "files": ["/tmp/testrepo"]
+    },
+    {
+      "id": 2,
+      "name": "org-agnostic-verification",
+      "prompt": "Before we ship metric events from /tmp/testrepo outside the team I need to confirm the output is org-agnostic. Emit the full event stream (CI runs at /tmp/testrepo/ci_runs.json, PRs at /tmp/testrepo/prs.json), verify no org-specific data appears in the events, and tell me which merged PRs were AI-assisted and how that was detected. Save the events and a verification note to the output directory.",
+      "expected_output": "events.ndjson; verification note confirming org-agnostic output; PR 42 ai_assisted=true (Co-Authored-By Claude trailer), PR 43 false",
+      "files": ["/tmp/testrepo"]
+    }
+  ]
+}
diff --git a/plugins/govkit/skills/govkit-metrics-emit/references/event_schema.md b/plugins/govkit/skills/govkit-metrics-emit/references/event_schema.md
new file mode 100644
index 0000000..a5f5893
--- /dev/null
+++ b/plugins/govkit/skills/govkit-metrics-emit/references/event_schema.md
@@ -0,0 +1,85 @@
+# Event vocabulary — tier1_metrics_catalog v1
+
+All events carry an envelope: `event`, `producer`, `catalog_version`, `schema_version`.
+Timestamps are ISO 8601. Every field is repo-relative (org-agnostic law).
+
+## feature.package.snapshot
+Emitted per feature directory under `features/`. Feeds **Pair 4** (spec lead time ↔
+spec completeness) and **Pair 6** (eval coverage at gate).
+
+| Field | Type | Source |
+|---|---|---|
+| `feature_id` | string | directory name under `features/` |
+| `commit_sha`, `ts` | string | last commit touching `features/<id>/` (git log); falls back to repo HEAD when the feature has no path-specific commit yet. `--validate` fails if neither is available (repo with no commits). |
+| `artifacts_present` | map | file existence: acceptance.feature, nfrs.md, plan.md, eval_criteria.yaml, architecture_preflight.md, agent_topology.md |
+| `artifacts_valid` | map | parse checks (gherkin ≥1 scenario, prediction block parses, criteria YAML parses) |
+| `gherkin_scenario_count` | int | `Scenario:` / `Scenario Outline:` lines, comments excluded |
+| `nfr_heading_count` | int | markdown headings in nfrs.md |
+| `evaluation_prediction` | object\|null | `first_average`, `virtues_average`, `thresholds_met` from the fenced YAML block in plan.md (schema: `governance/schemas/evaluation_prediction.schema.json`) |
+| `eval_criteria_summary` | object\|null | `mode` (llm\|deterministic\|none), `criteria_count`, `tools[]`, `enforce_first`, `enforce_virtues` from eval_criteria.yaml |
+| `completeness` | object | rubric result: `score` (0–100), `max`, `components[]` (name, points, awarded, reason) |
+| `marker_level` | string\|null | `.govkit/marker.json` → `level` |
+
+### Completeness rubric (Pair 4 quality counterweight)
+20 gherkin_scenarios · 10 nfrs · 15 evaluation_prediction_block ·
+15 thresholds_met · 15 score_minima (first.average ≥ 4.0 AND virtues.average ≥ 4.0) ·
+15 eval_criteria (valid; mode llm requires ≥1 criterion) ·
+10 architecture_preflight (required iff marker level = 5; auto-award otherwise).
+Score 100 = gate-ready (starts the "gate-ready" clock for spec lead time).
+
+## pr.merged
+From a PR export (`--prs`). Feeds **Pair 2** (throughput ↔ rework) and **Pair 5**
+(review latency ↔ unreviewed merges).
+Fields: `pr_id`, `feature_id` (parsed from title/branch `features/<id>` pattern),
+`ts_opened`, `ts_ready`, `ts_first_review`, `ts_merged`, `lines_added`,
+`lines_deleted`, `files_changed`, `ai_assisted`, `review_count`, `reviewer_types[]`
+(human/agent; agent = `[bot]` reviewer accounts).
+`ai_assisted` = any commit message matching
+`Co-Authored-By: … (claude|copilot|cursor|windsurf|codex|agent|bot)` — exhaust-only detection.
+
+## gate.run.completed
+From a CI runs export (`--ci-runs`). Feeds **Pair 3** (first-pass eval-gate rate)
+and **Pair 6** (gate throughput). Only workflows in the known gate set are emitted:
+eval-gate, deepeval-gate, promptfoo-gate, ui-eval-gate, quality-gate,
+ui-quality-gate, l3-quality-gate, l3-ui-quality-gate, guardrails-check,
+multi-agent-gate, repo-scope-check, data-common-gate, databricks-gate, dbt-gate
+(matches `ci/github/` and `ci/azure/` in the GovKit repo).
+Fields: `feature_id`, `gate_name`, `run_attempt`, `conclusion`, `ts_start`,
+`ts_end`, `commit_sha`.
+
+## deploy.completed
+Defined for **Pair 1** (deployment frequency ↔ change failure rate) and cycle-time
+termination (Pair 3). Originates in CI/CD deployment events — **not emitted by this
+producer**; a CI-side adapter emits it. Fields: `deploy_id`, `env`, `ts`, `status`
+(success|failure|rollback), `commit_shas[]`.
+
+## rework.observed
+Git-history scan. Feeds **Pair 2** (rework rate, quality counterweight).
+Fields: `commit_sha_origin`, `ts_origin`, `lines_added`, `lines_reworked_21d`,
+`rework_author_same`, `feature_id`, `method`.
+
+**Approximation note:** v1 uses file-overlap — deletions by later commits (within
+the window) in files the origin commit added to, capped at origin additions.
+This is an upper bound; it cannot distinguish which exact lines changed. Always
+disclose `method: file-overlap-approximation` when reporting. Exact line
+provenance (git blame lineage) is specified for the aggregator's v2 scan.
+
+## refinement.token.issued (RESERVED — not emitted in v1)
+Reserved for when `govkit-feature-refine` (govkit-plugins marketplace) writes its
+Development Token decision into the feature package as a structured record. The
+token is defined there as a decision record — Approved / Approved with edits /
+Blocked — authorizing AI-assisted coding to start. Once it exists as exhaust,
+this event unlocks Tier 2 metrics (refinement lead time Draft 0→Token,
+blocked-token rate) without changing any v1 metric IDs.
+Planned fields: `feature_id`, `decision` (approved|approved_with_edits|blocked),
+`draft_version`, `ts`.
+
+## Metric-pair mapping (paired-metric law)
+| Pair | Velocity | Quality | Events consumed |
+|---|---|---|---|
+| 1 | deployment_frequency | change_failure_rate | deploy.completed (+git revert detection, aggregator) |
+| 2 | ai_pr_throughput | rework_rate | pr.merged, rework.observed |
+| 3 | feature_cycle_time | first_pass_eval_gate_rate | feature.package.snapshot, deploy.completed, gate.run.completed |
+| 4 | spec_lead_time | spec_completeness_score | feature.package.snapshot |
+| 5 | review_latency | unreviewed_merge_rate | pr.merged |
+| 6 | gate_throughput | eval_coverage_at_gate | gate.run.completed, feature.package.snapshot |
diff --git a/plugins/govkit/skills/govkit-metrics-emit/scripts/emit_metrics.py b/plugins/govkit/skills/govkit-metrics-emit/scripts/emit_metrics.py
new file mode 100644
index 0000000..55270c0
--- /dev/null
+++ b/plugins/govkit/skills/govkit-metrics-emit/scripts/emit_metrics.py
@@ -0,0 +1,452 @@
+#!/usr/bin/env python3
+"""govkit-metrics-emit: org-agnostic Tier 1 metric event producer.
+
+Reads GovKit exhaust (feature packages, .govkit/marker.json, git history,
+optional CI-run / PR JSON exports) and emits NDJSON events per the AIPOS
+tier1_metrics_catalog v1 vocabulary.
+
+Design law: org-agnostic. Only repo-relative facts leave this script.
+"""
+import argparse
+import json
+import re
+import subprocess
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+try:
+    import yaml
+except ImportError:  # pragma: no cover
+    sys.exit("PyYAML required: pip install pyyaml --break-system-packages")
+
+PRODUCER = "govkit-metrics-emit/1.0.0"
+CATALOG_VERSION = "1.0.0"
+SCHEMA_VERSION = 1
+
+KNOWN_GATES = {
+    "eval-gate", "deepeval-gate", "promptfoo-gate", "ui-eval-gate",
+    "quality-gate", "ui-quality-gate", "l3-quality-gate", "l3-ui-quality-gate",
+    "guardrails-check", "multi-agent-gate", "repo-scope-check",
+    "data-common-gate", "databricks-gate", "dbt-gate",
+}
+
+EVENT_FIELDS = {
+    "feature.package.snapshot": {
+        "feature_id", "commit_sha", "ts", "artifacts_present", "artifacts_valid",
+        "gherkin_scenario_count", "nfr_heading_count", "evaluation_prediction",
+        "eval_criteria_summary", "completeness", "marker_level",
+    },
+    "pr.merged": {
+        "pr_id", "feature_id", "ts_opened", "ts_ready", "ts_first_review",
+        "ts_merged", "lines_added", "lines_deleted", "files_changed",
+        "ai_assisted", "review_count", "reviewer_types",
+    },
+    "gate.run.completed": {
+        "feature_id", "gate_name", "run_attempt", "conclusion",
+        "ts_start", "ts_end", "commit_sha",
+    },
+    "deploy.completed": {"deploy_id", "env", "ts", "status", "commit_shas"},
+    "rework.observed": {
+        "commit_sha_origin", "ts_origin", "lines_added", "lines_reworked_21d",
+        "rework_author_same", "feature_id", "method",
+    },
+}
+
+ENVELOPE_FIELDS = {"event", "producer", "catalog_version", "schema_version"}
+
+AI_TRAILER_RE = re.compile(
+    r"co-authored-by:.*\b(claude|copilot|cursor|windsurf|codex|agent|bot)\b", re.I
+)
+
+
+def git(repo: Path, *args: str) -> str:
+    out = subprocess.run(
+        ["git", "-C", str(repo), *args],
+        capture_output=True, text=True, check=False,
+    )
+    return out.stdout.strip() if out.returncode == 0 else ""
+
+
+def envelope(event_name: str, payload: dict) -> dict:
+    return {
+        "event": event_name,
+        "producer": PRODUCER,
+        "catalog_version": CATALOG_VERSION,
+        "schema_version": SCHEMA_VERSION,
+        **payload,
+    }
+
+
+# ---------------------------------------------------------------- marker
+def read_marker(repo: Path) -> dict:
+    for candidate in (repo / ".govkit" / "marker.json", repo / ".govkit"):
+        if candidate.is_file():
+            try:
+                return json.loads(candidate.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError):
+                return {}
+    return {}
+
+
+# ---------------------------------------------------------- feature parse
+def count_gherkin_scenarios(path: Path) -> int:
+    if not path.is_file():
+        return 0
+    n = 0
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        s = line.strip()
+        if s.startswith("#"):
+            continue
+        if re.match(r"Scenario( Outline)?:", s):
+            n += 1
+    return n
+
+
+def count_nfr_headings(path: Path) -> int:
+    if not path.is_file():
+        return 0
+    text = path.read_text(encoding="utf-8", errors="replace")
+    return len(re.findall(r"^#{1,4}\s+\S", text, flags=re.M))
+
+
+def extract_evaluation_prediction(plan_path: Path):
+    """Pull the evaluation_prediction YAML block out of plan.md."""
+    if not plan_path.is_file():
+        return None
+    text = plan_path.read_text(encoding="utf-8", errors="replace")
+    for m in re.finditer(r"```ya?ml\s*\n(.*?)```", text, flags=re.S):
+        block = m.group(1)
+        if "evaluation_prediction" not in block:
+            continue
+        try:
+            data = yaml.safe_load(block)
+        except yaml.YAMLError:
+            return None
+        if isinstance(data, dict) and isinstance(
+                data.get("evaluation_prediction"), dict):
+            return data["evaluation_prediction"]
+    return None
+
+
+def read_eval_criteria(path: Path):
+    if not path.is_file():
+        return None
+    try:
+        data = yaml.safe_load(path.read_text(encoding="utf-8", errors="replace"))
+    except yaml.YAMLError:
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def completeness(feature_dir: Path, pred, criteria, marker_level) -> dict:
+    """Spec completeness rubric — tier1_metrics_catalog pair-4 (0-100)."""
+    pred = pred if isinstance(pred, dict) else None
+    comps = []
+
+    def add(name, points, ok, reason):
+        comps.append({"name": name, "points": points,
+                      "awarded": points if ok else 0, "reason": reason})
+
+    scen = count_gherkin_scenarios(feature_dir / "acceptance.feature")
+    add("gherkin_scenarios", 20, scen >= 1,
+        f"{scen} scenario(s) in acceptance.feature")
+
+    nfr = count_nfr_headings(feature_dir / "nfrs.md")
+    add("nfrs", 10, nfr >= 1, f"{nfr} heading(s) in nfrs.md")
+
+    add("evaluation_prediction_block", 15, pred is not None,
+        "evaluation_prediction parsed from plan.md" if pred is not None
+        else "plan.md missing or evaluation_prediction block absent/invalid")
+
+    thr = bool(pred and pred.get("thresholds_met") is True)
+    add("thresholds_met", 15, thr,
+        f"thresholds_met={pred.get('thresholds_met') if pred else None}")
+
+    fa = pred.get("first", {}).get("average") if pred else None
+    va = pred.get("virtues", {}).get("average") if pred else None
+    ok_scores = isinstance(fa, (int, float)) and isinstance(va, (int, float)) \
+        and fa >= 4.0 and va >= 4.0
+    add("score_minima", 15, ok_scores, f"first.average={fa}, virtues.average={va}")
+
+    crit_list = (criteria or {}).get("llm_evaluation", {}).get("criteria", []) \
+        if isinstance((criteria or {}).get("llm_evaluation"), dict) else []
+    mode = (criteria or {}).get("mode")
+    ok_crit = bool(criteria) and mode in ("llm", "deterministic", "none") and \
+        (mode != "llm" or len(crit_list) >= 1)
+    add("eval_criteria", 15, ok_crit,
+        f"mode={mode}, criteria={len(crit_list)}" if criteria
+        else "eval_criteria.yaml missing or invalid")
+
+    needs_preflight = str(marker_level) == "5"
+    has_preflight = (feature_dir / "architecture_preflight.md").is_file()
+    add("architecture_preflight", 10,
+        has_preflight if needs_preflight else True,
+        f"required={needs_preflight}, present={has_preflight}")
+
+    score = sum(c["awarded"] for c in comps)
+    return {"score": score, "max": 100, "components": comps}
+
+
+def snapshot_events(repo: Path, marker: dict):
+    features_dir = repo / "features"
+    if not features_dir.is_dir():
+        return
+    level = marker.get("level")
+    # Fallback anchor for a feature with no path-specific commit yet (e.g. an
+    # uncommitted working-tree feature): use repo HEAD so commit_sha / ts stay
+    # non-null strings per the event vocabulary. They stay None only when the
+    # repo has no commits at all, which validate() then flags.
+    head_sha = git(repo, "rev-parse", "HEAD")
+    head_ts = git(repo, "show", "-s", "--format=%cI", "HEAD")
+    for fdir in sorted(p for p in features_dir.iterdir() if p.is_dir()):
+        rel = f"features/{fdir.name}"
+        sha = git(repo, "log", "-1", "--format=%H", "--", rel) or head_sha or None
+        ts = git(repo, "log", "-1", "--format=%cI", "--", rel) or head_ts or None
+        artifacts = ["acceptance.feature", "nfrs.md", "plan.md",
+                     "eval_criteria.yaml", "architecture_preflight.md",
+                     "agent_topology.md"]
+        present = {a: (fdir / a).is_file() for a in artifacts}
+        pred = extract_evaluation_prediction(fdir / "plan.md")
+        criteria = read_eval_criteria(fdir / "eval_criteria.yaml")
+        valid = {
+            "acceptance.feature": count_gherkin_scenarios(fdir / "acceptance.feature") >= 1,
+            "plan.md": pred is not None,
+            "eval_criteria.yaml": criteria is not None,
+        }
+        crit_list = (criteria or {}).get("llm_evaluation", {}).get("criteria", []) \
+            if isinstance((criteria or {}).get("llm_evaluation"), dict) else []
+        yield envelope("feature.package.snapshot", {
+            "feature_id": fdir.name,
+            "commit_sha": sha,
+            "ts": ts,
+            "artifacts_present": present,
+            "artifacts_valid": valid,
+            "gherkin_scenario_count": count_gherkin_scenarios(fdir / "acceptance.feature"),
+            "nfr_heading_count": count_nfr_headings(fdir / "nfrs.md"),
+            "evaluation_prediction": {
+                "first_average": (pred or {}).get("first", {}).get("average"),
+                "virtues_average": (pred or {}).get("virtues", {}).get("average"),
+                "thresholds_met": (pred or {}).get("thresholds_met"),
+            } if pred else None,
+            "eval_criteria_summary": {
+                "mode": (criteria or {}).get("mode"),
+                "criteria_count": len(crit_list),
+                "tools": sorted({c.get("tool") for c in crit_list if c.get("tool")}),
+                "enforce_first": (criteria or {}).get("unit_tests", {}).get("enforce_FIRST"),
+                "enforce_virtues": (criteria or {}).get("code_quality", {}).get("enforce_virtues"),
+            } if criteria else None,
+            "completeness": completeness(fdir, pred, criteria, level),
+            "marker_level": level,
+        })
+
+
+def load_json_list(path: Path, label: str) -> list:
+    """Read a user-supplied JSON array (optional --ci-runs / --prs input).
+
+    Optional inputs must not abort the run with a traceback: on any read or
+    parse problem, or if the top level is not a JSON array, exit 2 with a
+    clear message instead.
+    """
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except OSError as exc:
+        print(f"error: cannot read {label} file '{path}': {exc}", file=sys.stderr)
+        sys.exit(2)
+    except json.JSONDecodeError as exc:
+        print(f"error: {label} file '{path}' is not valid JSON: {exc}",
+              file=sys.stderr)
+        sys.exit(2)
+    if not isinstance(data, list):
+        print(f"error: {label} file '{path}' must be a JSON array, not "
+              f"{type(data).__name__}", file=sys.stderr)
+        sys.exit(2)
+    return data
+
+
+# ------------------------------------------------------------- gate runs
+def gate_events(runs: list):
+    for r in runs:
+        raw = (r.get("name") or r.get("workflowName") or "").strip()
+        base = raw.lower().removesuffix(".yml").removesuffix(".yaml")
+        if base not in KNOWN_GATES:
+            continue
+        blob = " ".join(str(r.get(k, "")) for k in
+                        ("displayTitle", "headBranch", "title"))
+        m = re.search(r"features?/([A-Za-z0-9_\-]+)", blob)
+        yield envelope("gate.run.completed", {
+            "feature_id": m.group(1) if m else None,
+            "gate_name": base,
+            "run_attempt": r.get("attempt") or r.get("runAttempt") or 1,
+            "conclusion": r.get("conclusion") or r.get("result"),
+            "ts_start": r.get("createdAt") or r.get("startTime"),
+            "ts_end": r.get("updatedAt") or r.get("finishTime"),
+            "commit_sha": r.get("headSha") or r.get("sourceVersion"),
+        })
+
+
+# ------------------------------------------------------------------ PRs
+def pr_events(prs: list):
+    for pr in prs:
+        commits = pr.get("commits") or []
+        texts = []
+        for c in commits:
+            texts.append(str(c.get("messageHeadline", "")))
+            texts.append(str(c.get("messageBody", "")))
+        ai = bool(AI_TRAILER_RE.search("\n".join(texts)))
+        reviews = pr.get("reviews") or []
+        rtypes = sorted({
+            "agent" if "[bot]" in str((rv.get("author") or {}).get("login", ""))
+            else "human"
+            for rv in reviews
+        })
+        first_review = min(
+            (rv.get("submittedAt") for rv in reviews if rv.get("submittedAt")),
+            default=None,
+        )
+        blob = f"{pr.get('title','')} {pr.get('headRefName','')}"
+        m = re.search(r"features?/([A-Za-z0-9_\-]+)", blob)
+        yield envelope("pr.merged", {
+            "pr_id": pr.get("number"),
+            "feature_id": m.group(1) if m else None,
+            "ts_opened": pr.get("createdAt"),
+            "ts_ready": pr.get("readyAt") or pr.get("createdAt"),
+            "ts_first_review": first_review,
+            "ts_merged": pr.get("mergedAt"),
+            "lines_added": pr.get("additions"),
+            "lines_deleted": pr.get("deletions"),
+            "files_changed": pr.get("changedFiles"),
+            "ai_assisted": ai,
+            "review_count": len(reviews),
+            "reviewer_types": rtypes,
+        })
+
+
+# --------------------------------------------------------------- rework
+def rework_events(repo: Path, window_days: int):
+    """File-overlap approximation of 21-day rework (upper bound).
+
+    lines_reworked = deletions applied by later commits (within the window)
+    to files the origin commit added lines to, capped at origin additions.
+    Exact line provenance is aggregator-side v2.
+    """
+    log = git(repo, "log", "--reverse", "--format=C|%H|%cI|%ae", "--numstat")
+    if not log:
+        return
+    commits = []
+    cur = None
+    for line in log.splitlines():
+        if line.startswith("C|"):
+            _, sha, ts, email = line.split("|", 3)
+            cur = {"sha": sha, "ts": ts, "author": email, "files": {}}
+            commits.append(cur)
+        elif cur is not None and "\t" in line:
+            a, d, fname = line.split("\t", 2)
+            if a == "-":
+                continue
+            cur["files"][fname] = {"added": int(a), "deleted": int(d)}
+    for i, c in enumerate(commits):
+        try:
+            t0 = datetime.fromisoformat(c["ts"])
+        except ValueError:
+            continue
+        horizon = t0 + timedelta(days=window_days)
+        added = sum(f["added"] for f in c["files"].values())
+        if added == 0:
+            continue
+        reworked = 0
+        same_author = False
+        for later in commits[i + 1:]:
+            try:
+                t1 = datetime.fromisoformat(later["ts"])
+            except ValueError:
+                continue
+            if t1 > horizon:
+                break
+            overlap = set(c["files"]) & set(later["files"])
+            for fname in overlap:
+                reworked += min(c["files"][fname]["added"],
+                                later["files"][fname]["deleted"])
+                if later["author"] == c["author"]:
+                    same_author = True
+        feature = None
+        for fname in c["files"]:
+            m = re.match(r"features/([A-Za-z0-9_\-]+)/", fname)
+            if m:
+                feature = m.group(1)
+                break
+        yield envelope("rework.observed", {
+            "commit_sha_origin": c["sha"],
+            "ts_origin": c["ts"],
+            "lines_added": added,
+            "lines_reworked_21d": min(reworked, added),
+            "rework_author_same": same_author,
+            "feature_id": feature,
+            "method": "file-overlap-approximation",
+        })
+
+
+# ------------------------------------------------------------- validate
+def validate(events: list) -> dict:
+    summary = {"total": len(events), "by_event": {}, "errors": []}
+    for i, e in enumerate(events):
+        name = e.get("event")
+        summary["by_event"][name] = summary["by_event"].get(name, 0) + 1
+        if name not in EVENT_FIELDS:
+            summary["errors"].append(f"event {i}: unknown event '{name}'")
+            continue
+        allowed = EVENT_FIELDS[name] | ENVELOPE_FIELDS
+        unknown = set(e) - allowed
+        if unknown:
+            summary["errors"].append(
+                f"event {i} ({name}): unknown fields {sorted(unknown)}")
+        missing_env = ENVELOPE_FIELDS - set(e)
+        if missing_env:
+            summary["errors"].append(
+                f"event {i} ({name}): missing envelope {sorted(missing_env)}")
+        if name == "feature.package.snapshot":
+            for key in ("commit_sha", "ts"):
+                if not e.get(key):
+                    summary["errors"].append(
+                        f"event {i} ({name}): missing or empty '{key}'")
+    summary["valid"] = not summary["errors"]
+    return summary
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("repo", type=Path)
+    ap.add_argument("--ci-runs", type=Path)
+    ap.add_argument("--prs", type=Path)
+    ap.add_argument("--rework-days", type=int, default=21)
+    ap.add_argument("--out", type=Path)
+    ap.add_argument("--validate", action="store_true")
+    args = ap.parse_args()
+
+    if not args.repo.is_dir():
+        sys.exit(f"repo not found: {args.repo}")
+
+    marker = read_marker(args.repo)
+    events = list(snapshot_events(args.repo, marker))
+    events.extend(rework_events(args.repo, args.rework_days))
+    if args.ci_runs:
+        events.extend(gate_events(load_json_list(args.ci_runs, "--ci-runs")))
+    if args.prs:
+        events.extend(pr_events(load_json_list(args.prs, "--prs")))
+
+    ndjson = "\n".join(json.dumps(e, sort_keys=False) for e in events)
+    if args.out:
+        args.out.write_text(ndjson + "\n", encoding="utf-8")
+        print(f"wrote {len(events)} events -> {args.out}", file=sys.stderr)
+    else:
+        print(ndjson)
+
+    if args.validate:
+        summary = validate(events)
+        print(json.dumps(summary, indent=2), file=sys.stderr)
+        if not summary["valid"]:
+            sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()