Add trace schema v1, replay and dataset export

doc/architecture/SPEC/kb_service.md

@@ -81,3 +81,17 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv
   既不执行也不回执。每轮与每个决定都写入 `traces/publisher.jsonl`。
 - 测试：`test_kb_publisher.py`（端到端 open_pr → ack → pr_open、dry-run、崩溃后不重复执行、伪造/未配置/过期、
   越界路径、verdict/入队/暂停/关闭、暂停期间仍执行暂停、SSH 引号与回执、ack 字段）。
+
+## 2026-09-28 留痕、回放与数据集
+- 运行时持有 `TraceStore(state_dir/traces)`：`ModelGateway` 的每次调用（含失败）写为 `model_call`；
+  `gate_and_stage` 先分配变更集 ID，判定期间绑定 `changeset_id`/`rule_ids`，暂存后写 `decision`；
+  `merge.advance` 写 `outcome`（merged / closed_unmerged / head_changed），退役写 `rule_retired`。
+  intake、sweep、calibration 分别绑定 playbook `kb-intake`/`kb-sweep`/`kb-calibrate`。留痕失败从不中断服务。
+- `replay`：用记录中的原始 system/prompt 询问替代模型，比较决定性字段（`dimensions` 或 `verdict`），写
+  `replay` 记录；替代模型的调用本身也记为 playbook `kb-replay`。只调用模型，不触及 GitHub、outbox 或账本。
+- `export_dataset`：按角色导出（输入、高成本模型输出、变更集判定、事后结果、规则后续结果）；丢弃校准与回放
+  调用、失败调用（截断、空、无法解析或不符 schema 的回复都记为失败），以及 prompt 中任意位置出现校准用例证据
+  引用（如 `PR #8107`，按边界匹配）或标题的调用。生成器调用发生在变更集之前：每次起草有唯一 `draft_key`（intake 为事件、
+  巡检为发版 + 页面，均带随机后缀），只有被采纳的那次尝试（`draft_key#attempt`）列入之后 `decision` 记录的
+  `draft_keys`，被拒绝的尝试与失败的重试不继承判定。记录的每个字符串（含键、model、usage）都经过脱敏。
+- CLI：`kb traces`、`kb replay --record ID --model P:M[:E]`、`kb export --out FILE`。

doc/architecture/SPEC/sdk.md

@@ -124,3 +124,7 @@ Direct 1.1 / Strict 1.3 accept typed `CarriedFinding` inputs (unique IDs, source
 
 ## 2026-09-28 退役规则不再提供
 `DirectClient` 返回的文档内容与 `sha256` 均基于去掉退役规则后的服务文本（无退役规则时与磁盘字节一致）。
+
+## 2026-09-28 trace/1 导出
+`sdk.v1` 导出 `TraceStore`、`TRACE_SCHEMA`、`trace_context`、`redact`（见 `trace_store.md`），RB 用同一 schema
+记录 review 的模型调用、判定与结果；导入它们不加载 `config`/`contract`/`mcp_server` 等私有模块。

doc/architecture/SPEC/trace_store.md

@@ -0,0 +1,24 @@
+# trace_store.py —— 规范
+
+<!-- verified-against: 2026-09-28 -->
+
+`LOC ~260 · trace/1：Copilot 与 RB 共用的模型调用/判定/结果记录 · refactor-status: stable`
+
+## 职责
+- `TraceStore(root)`：追加式 `records/<日期>.jsonl`（事实来源）+ 内容寻址 blob
+  （`blobs/<aa>/<sha256>.gz`，gzip，读取时复核哈希）+ SQLite 索引 `index.db`
+  （按 kind、run、仓库、变更集、PR、规则 ID、角色、模型、结果查询；`rebuild_index` 可从 JSONL 重建）。
+- 记录（`schema: trace/1`）：`kind` ∈ `model_call`/`decision`/`outcome`/`replay`；`context`（run_id、
+  playbook、step、repo、changeset_id、pr、rule_ids…）、`model`（role、provider、model、effort、served_model）、
+  `usage`、`seconds`、`inputs`/`outputs`（只存 blob 引用，从不内联正文）、`result`、`error`、`env`
+  （copilot 版本与 SHA）。`outcome` 是事后金标（merged、closed_unmerged、head_changed、rule_retired）。
+- `trace_context(**fields)`：用 contextvars 绑定上下文，嵌套叠加，线程/协程安全。
+
+## 不变量
+- 写入前统一脱敏：已知令牌形态（GitHub、Anthropic、OpenAI、AWS、Slack、Bearer、私钥块）以及名字表明是
+  密钥的环境变量的值（路径形式的值除外）。脱敏发生在哈希之前，blob 与 JSONL 中都不会出现密钥。
+- 只依赖标准库；经 `sdk.v1` 导出（`TraceStore`、`TRACE_SCHEMA`、`trace_context`、`redact`）时不加载私有模块。
+
+## 测试
+`test_trace_store.py`（schema 与去重 blob、索引与重建、脱敏、网关记录成功与失败调用、回放、导出的判定/结果
+关联与校准集泄漏过滤、SDK 导出、知识服务的 decision/outcome 记录）。

doc/features/knowledge-service.md

@@ -106,3 +106,14 @@ ALLOW_POST=1 ALLOW_PUSH=1 infermatrix-copilot kb publish --remote bot-host:/path
 
 没有 `ALLOW_POST=1` 时只把将要执行的动作写入 `$KB_PUBLISHER_STATE/traces/publisher.jsonl`；推送分支还需要
 `ALLOW_PUSH=1`。控制记录超过 10 分钟未更新（服务可能宕机）时整轮不执行任何动作。
+
+## 留痕、回放与数据集（trace/1）
+
+知识服务的每次模型调用（含失败）、每个质量门判定与每个事后结果都写入 `$KB_STATE_DIR/traces/`
+（JSONL + 内容寻址 blob + SQLite 索引，写入前脱敏）。RB 通过 `sdk.v1.TraceStore` 使用同一 schema。
+
+```bash
+infermatrix-copilot kb traces --changeset <id>                      # 查询记录
+infermatrix-copilot kb replay --record <id> --model codex:gpt-6-mini:low   # 用便宜模型重放一次调用
+infermatrix-copilot kb export --out judge.jsonl                     # 导出数据集（自动剔除校准集相关调用）
+```

src/infermatrix_copilot/kb_service/calibration.py

@@ -67,6 +67,14 @@ def load_cases(directory: str | Path) -> list[dict]:
 
 
 def run_calibration(directory: str | Path, *, gateway, judge) -> CalibrationReport:
+    """Traced as playbook ``kb-calibrate``: dataset export never uses these calls."""
+    from ..trace_store import trace_context
+
+    with trace_context(playbook="kb-calibrate", run_id=f"calibrate-{Path(directory).name}"):
+        return _run_calibration(directory, gateway=gateway, judge=judge)
+
+
+def _run_calibration(directory: str | Path, *, gateway, judge) -> CalibrationReport:
     details = []
     bad_total = bad_caught = good_total = good_rejected = 0
     for case in load_cases(directory):

src/infermatrix_copilot/kb_service/cli.py

@@ -10,6 +10,9 @@
     kb serve [--once]                 the scheduler (holds the single-writer lease)
     kb activate                       activate the knowledge snapshot of the repository's main now
     kb rollback --to SHA              point `active` back at an earlier snapshot
+    kb traces [--kind K] [--changeset ID] [--rule ID] [--limit N]   query trace/1 records
+    kb replay --record ID --model PROVIDER:MODEL[:EFFORT]           re-ask a recorded call
+    kb export --out FILE [--role judge|generator]                   dataset (calibration-safe)
     kb publish (--remote HOST:/STATE_DIR | --local DIR) [--once]
                                       the publisher (GPU box, owner's gh login): perform
                                       the signed outbox items; writes need ALLOW_POST=1
@@ -110,6 +113,18 @@ def main(argv: list[str] | None = None) -> int:
     sub.add_parser("activate")
     rollback = sub.add_parser("rollback")
     rollback.add_argument("--to", required=True)
+    traces = sub.add_parser("traces")
+    traces.add_argument("--kind")
+    traces.add_argument("--changeset")
+    traces.add_argument("--rule")
+    traces.add_argument("--run")
+    traces.add_argument("--limit", type=int, default=50)
+    replay = sub.add_parser("replay")
+    replay.add_argument("--record", required=True)
+    replay.add_argument("--model", required=True, help="provider:model[:effort], e.g. codex:gpt-6-mini:low")
+    export = sub.add_parser("export")
+    export.add_argument("--out", required=True)
+    export.add_argument("--role", default="judge", choices=("judge", "generator"))
     publish = sub.add_parser("publish")
     where = publish.add_mutually_exclusive_group(required=True)
     where.add_argument("--remote", help="host:/absolute/path of the service state directory (over ssh)")
@@ -129,6 +144,8 @@ def main(argv: list[str] | None = None) -> int:
 
     if args.command == "publish":
         return _publish(args)
+    if args.command in {"traces", "replay", "export"}:
+        return _traces_command(args, _state_dir(args.state_dir))
 
     state_dir = _state_dir(args.state_dir)
     ledger = _ledger(state_dir)
@@ -271,3 +288,37 @@ def _publish(args) -> int:
         return 0
     publisher.serve(interval=args.interval)
     return 0
+
+
+def _traces_command(args, state_dir: Path) -> int:
+    """Read-only over the trace store (replay also calls the substitute model)."""
+    from ..trace_store import TraceStore
+
+    store = TraceStore(state_dir / "traces")
+    if args.command == "traces":
+        records = store.query(kind=args.kind, changeset_id=args.changeset, rule_id=args.rule,
+                              run_id=args.run, limit=args.limit)
+        for record in records:
+            print(json.dumps({k: record[k] for k in ("id", "kind", "at", "context", "model", "result", "error")},
+                             ensure_ascii=False, sort_keys=True))
+        return 0
+    if args.command == "replay":
+        from ..config import Settings
+        from .models import ModelGateway, ModelRole
+        from .replay import replay
+        from .runtime import trace_recorder
+
+        role_name = (store.get(args.record).get("model") or {}).get("role") or "judge"
+        gateway = ModelGateway(Settings(), recorder=trace_recorder(store))  # the replay call is traced too
+        result = replay(store, args.record, gateway, ModelRole.parse(role_name, args.model))
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
+        print()
+        return 0
+    from ..sdk._resources import adapters_root
+    from .replay import export_dataset
+
+    adapters = Path(os.environ.get("ADAPTERS_DIR") or adapters_root())
+    calibration = [p for p in sorted(adapters.glob("*/kb-calibration")) if p.is_dir()]
+    json.dump(export_dataset(store, args.out, role=args.role, calibration_dirs=calibration), sys.stdout)
+    print()
+    return 0

src/infermatrix_copilot/kb_service/intake.py

@@ -166,10 +166,13 @@ def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict,
     prompt = draft_prompt(repo, evidence, files, repo_dir)
     attempts: list[dict] = []
     feedback = ""
+    from ..trace_store import accept_attempt, trace_context
+
     for attempt in range(MAX_REPAIRS + 1):
         try:
-            reply = gateway.call_json(generator, system=SYSTEM, prompt=prompt + feedback,
-                                      validate=_validate_reply)
+            with trace_context(attempt=attempt):
+                reply = gateway.call_json(generator, system=SYSTEM, prompt=prompt + feedback,
+                                          validate=_validate_reply)
         except ModelUnavailable as exc:
             if "failed its schema" not in str(exc):
                 raise  # the model itself is unavailable: the event waits
@@ -194,6 +197,7 @@ def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict,
                             "Fix exactly that and answer again with the full JSON object.")
                 attempts.append({"attempt": attempt, "error": str(exc)})
                 continue
+            accept_attempt(attempt)  # only this call's reply became the change
             return Draft([event_id], operations, result, str(reply.data.get("rationale") or ""), attempts)
         attempts.append({"attempt": attempt, "error": feedback.strip()})
     return Draft([event_id], [], None, "rejected after repairs", attempts, rejected=True)

src/infermatrix_copilot/kb_service/merge.py

@@ -159,14 +159,17 @@ def advance(rt, lifecycle) -> list[str]:
             rt.ledger.update_changeset(changeset["id"], status="merged",
                                        merge_sha=str(pr.get("merge_commit_sha") or ""))
             record_retirements(rt, changeset)
+            _outcome(rt, changeset, "merged", merge_sha=str(pr.get("merge_commit_sha") or ""))
             events.append(f"merged {changeset['id']}")
             continue
         if pr.get("state") == "closed":
             rt.ledger.update_changeset(changeset["id"], status="closed")
+            _outcome(rt, changeset, "closed_unmerged")  # overturned by people: a negative label
             events.append(f"closed {changeset['id']}")
             continue
         head = str((pr.get("head") or {}).get("sha") or "")
         if head != changeset["head_sha"]:
+            _outcome(rt, changeset, "head_changed", head_sha=head)
             rt.ledger.update_changeset(changeset["id"], status="head_changed")
             rt.ledger.enqueue_human(lifecycle.repo, f"PR #{number} head changed after signing", changeset["id"])
             events.append(f"head_changed {changeset['id']}")
@@ -192,11 +195,31 @@ def advance(rt, lifecycle) -> list[str]:
     return events
 
 
+def _rule_ids(changeset: dict) -> list[str]:
+    return sorted({op.get("new_rule_id") or op["rule_id"] for op in changeset["detail"].get("operations", [])
+                   if op.get("new_rule_id") or op.get("rule_id")})
+
+
+def _outcome(rt, changeset: dict, outcome: str, **result) -> None:
+    """What finally happened to a change set: the gold label its decision is scored against."""
+    trace = getattr(rt, "trace", None)
+    if trace is not None:
+        trace("outcome", context={"repo": changeset["repo"], "changeset_id": changeset["id"],
+                                  "pr": changeset.get("pr_number"), "rule_ids": _rule_ids(changeset)},
+              result={"outcome": outcome, **result})
+
+
 def record_retirements(rt, changeset: dict) -> None:
     release = changeset["detail"].get("release", "")
     for op in changeset["detail"].get("operations", []):
         if op["kind"] in ("retire", "replace"):
             rt.ledger.record_retirement(changeset["repo"], op["rule_id"], op["page"], release)
+            trace = getattr(rt, "trace", None)
+            if trace is not None:  # the rule's earlier admission now has a later-retired label
+                trace("outcome", context={"repo": changeset["repo"], "changeset_id": changeset["id"],
+                                          "rule_ids": [op["rule_id"]]},
+                      result={"outcome": "rule_retired", "rule_id": op["rule_id"], "release": release,
+                              "reason": op.get("reason", "")})
         elif op["kind"] == "purge":
             rt.ledger.mark_purged(changeset["repo"], op["rule_id"])
 

src/infermatrix_copilot/kb_service/models.py

@@ -104,29 +104,46 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str,
                   validate: Callable[[dict], None] | None = None) -> ModelReply:
         transport = self._transport(role.provider)
         started = time.time()
+        identity = {"role": role.name, "requested": role.label(), "provider": role.provider,
+                    "model": role.model, "effort": role.effort}
         try:
             reply = transport.complete(
                 system=system, messages=[{"role": "user", "content": prompt}],
                 model=role.model, effort=role.effort, role=role.name)
         except Exception as exc:  # the transport's own failure modes
+            if self._recorder is not None:
+                self._recorder({**identity, "served_model": "", "stop_reason": "", "usage": {},
+                                "seconds": round(time.time() - started, 3), "system": system,
+                                "prompt": prompt, "reply": "", "error": str(exc)[:2000]})
             raise ModelUnavailable(f"{role.label()} failed: {exc}") from exc
         seconds = time.time() - started
         text = "".join(getattr(block, "text", "") or "" for block in getattr(reply, "blocks", []))
         record = {
-            "role": role.name, "requested": role.label(),
+            **identity,
             "served_model": getattr(reply, "model", "") or "",
             "stop_reason": getattr(reply, "stop_reason", ""),
             "usage": dict(getattr(reply, "usage", {}) or {}),
             "seconds": round(seconds, 3), "system": system, "prompt": prompt, "reply": text,
         }
-        if self._recorder is not None:
-            self._recorder(record)
+        # the record carries the call's FINAL verdict: a truncated, empty,
+        # unparseable or schema-failing reply is recorded as a failure, so it
+        # never becomes a training example
+        failure: ModelUnavailable | None = None
+        data: dict = {}
         if getattr(reply, "stop_reason", "") == "max_tokens" or not text.strip():
-            raise ModelUnavailable(f"{role.label()} timed out or returned nothing")
-        data = parse_json_object(text)
-        if validate is not None:
+            failure = ModelUnavailable(f"{role.label()} timed out or returned nothing")
+        else:
             try:
-                validate(data)
-            except Exception as exc:  # any malformed shape is a controlled refusal
-                raise ModelUnavailable(f"{role.label()} reply failed its schema: {exc!r}") from exc
+                data = parse_json_object(text)
+                if validate is not None:
+                    try:
+                        validate(data)
+                    except Exception as exc:  # any malformed shape is a controlled refusal
+                        raise ModelUnavailable(f"{role.label()} reply failed its schema: {exc!r}") from exc
+            except ModelUnavailable as exc:
+                failure = exc
+        if self._recorder is not None:
+            self._recorder({**record, "error": str(failure) if failure else ""})
+        if failure is not None:
+            raise failure
         return ModelReply(role, data, text, record["served_model"], record["usage"], seconds)