fix(kb): preserve complete replays without blocking baseline intake

Author: tzhouam

doc/architecture/SPEC/improve.md

@@ -1,6 +1,6 @@
 # improve/ —— 规范（元改进引擎）
 
-<!-- verified-against: 2026-10-08 -->
+<!-- verified-against: 2026-10-09 -->
 
 `设计：/data/zhoutaichang/copilot/meta-improvement-engine-design.md v1（GPT-6 sol 批准 2026-09-29） · refactor-status: building (P0–P4 已落地)`
 
@@ -14,6 +14,8 @@
 
 `objectives.py` 是受保护的父进程目标控制器（`objective/1`）：自动导入可信 replay 输入、生成有因果见证的故障／负例、固定一个 API 模型、预注册三次配对及 item 聚类功效、核验实际源码与输入、独立计算分数及护栏。见证不传生成器或候选；候选数值分数无效。PR 与知识指标仅证明结构契约、产物保留与资源收益，`semantic_quality_claim=false`。引擎候选每次只修改归因或 lint；lint 既有正例及负例受保护。功效不足会存档并等待新鲜样本，不重放完成的付费调用。
 
+Replay envelope 保留完整文件与引用，由 TraceStore 脱敏、压缩和按内容寻址；未压缩 UTF-8 JSON 上限为 64,000,000 bytes。超限只记录 `replay_capture_deferred` 的大小、上限和摘要并抛出专用 `ReplayCaptureLimit`，不保存截断输入，也不进入 replay 数据集。知识起草无活动进化工作流时继续原生固定生成器；活动工作流（含已禁用项）缺少完整 replay 时保持事件 pending。其他捕获错误不能走此延期路径。
+
 `runtime.py` 重新应用补丁核验实际评估树，原子采用本地不可变 release，原生 PR 评审、知识草稿、lint 和归因在活动制品的沙箱中运行。宿主记录运行指纹才更新部署基线。首次八个不同 item 与上一版比较，七日缺证据自动恢复；保留后每周最多八个新 item 持续比较。契约、凭据／隔离／预算、指纹失败或资源显著退化自动恢复上一版；恢复制品不可验证则禁用执行。执行／切换锁及持久化激活、回滚意图支持重启恢复。
 
 默认 `IMPROVE_EVALUATION_MODE=objective`、`IMPROVE_PROMOTION_MODE=automatic`，不要求人工标签、判官或 performance 模型。总开关仍关闭，固定模型取 `IMPROVE_EVOLVE_MODEL` 或 eco；每周一个候选、一次修复及共享 20 美元预算不变。`gold/pr` 模式保留原有双家族取证及人工流程。无隔离生产驱动的机械工作流仍可 lint，但不自动采用。当前主机命名空间关闭，代码候选实际执行延期。

doc/architecture/SPEC/kb_service.md

@@ -929,6 +929,7 @@ dry run 为旧快照建本地临时 baseline，再生成相同升级串与全量
 ## 自主进化接入
 
 自主进化开启时起草前捕获事件与知识快照；已采用制品通过隔离代理起草 typed operations，宿主沿用已有知识发布权限与门禁。进化实验只在内存应用，不写产品知识库。
+完整 replay 的未压缩 UTF-8 JSON 上限为 64,000,000 bytes；超过上限记录 `replay_capture_deferred`，不截断全局规则 ID、tombstone 或外部引用。只有无活动 `kb-intake.draft` 制品时可继续原生固定生成器起草；活动项（包括已禁用项）或无法读取的注册表保持 pending，不执行未捕获的候选。其他捕获错误照常阻断。质量门、校准和发布授权不受此存储延期影响。
 
 ## 2026-10-05 通用功能发现：文档基线与源码扩展
 

src/infermatrix_copilot/improve/objectives.py

@@ -22,6 +22,16 @@
 ENGINE = 'workflow-improve.improve.forensics'
 DRIVERS = {'pr-review': 'pr-review', 'kb-intake': 'kb-intake', 'meta': 'objective-meta', 'mechanical': 'objective-tests'}
 FAULTS = {'ceiling': ('S4', 'L01'), 'discard': ('S3', 'L09'), 'fallback': ('S7', 'L07')}
+# Complete KB trees and companion texts already exceed 16 MB. TraceStore
+# compresses these envelopes; retain a finite uncompressed memory/disk bound.
+REPLAY_INPUT_MAX_BYTES = 64_000_000
+
+
+class ReplayCaptureLimit(artifacts.ArtifactError):
+    """A complete replay exceeds the storage bound, before candidate execution."""
+    def __init__(self, size_bytes, limit_bytes):
+        self.size_bytes, self.limit_bytes = size_bytes, limit_bytes
+        super().__init__(f'replay input exceeds capture limit ({size_bytes} > {limit_bytes} bytes)')
 
 
 def enabled(settings):
@@ -81,9 +91,17 @@ def capture(store, workflow, item, payload):
     forbidden = {'labels', 'gold', 'scores', 'judgments', 'report'}
     if forbidden & set(payload): raise artifacts.ArtifactError('replay input includes evaluator material')
     raw = json.dumps(payload, ensure_ascii=False)
-    if len(raw.encode()) > 16_000_000: raise artifacts.ArtifactError('replay input exceeds capture limit')
-    store.append('decision', context={'workflow': workflow, 'item': item},
-                 inputs={'evolution_input': raw}, result={'type': 'replay_input', 'input_sha': artifacts.digest(raw.encode())})
+    encoded = raw.encode()
+    context = {'workflow': workflow, 'item': item}
+    input_sha = artifacts.digest(encoded)
+    if len(encoded) > REPLAY_INPUT_MAX_BYTES:
+        # A missing replay is explicit and never eligible for collection. Do
+        # not truncate files or references that affect the executable contract.
+        store.append('decision', context=context, result={'type': 'replay_capture_deferred', 'reason': 'size_limit',
+                     'input_bytes': len(encoded), 'limit_bytes': REPLAY_INPUT_MAX_BYTES, 'input_sha': input_sha})
+        raise ReplayCaptureLimit(len(encoded), REPLAY_INPUT_MAX_BYTES)
+    store.append('decision', context=context,
+                 inputs={'evolution_input': raw}, result={'type': 'replay_input', 'input_sha': input_sha})
 
 
 def collect(settings, store, workflow):

src/infermatrix_copilot/kb_service/runtime.py

@@ -256,16 +256,31 @@ def _run_intake_locked(rt: KbRuntime, lifecycle: RepoLifecycle, *, max_events: i
                 from ..improve import objectives
                 prediction = None
                 if objectives.enabled(draft_settings) and draft_settings.improve_enabled and draft_settings.improve_evolve_enabled and rt.traces is not None:
-                    from ..improve.runtime import execute
+                    from ..improve.runtime import active, execute
                     payload = {"repo": lifecycle.repo, "repo_dir": lifecycle.knowledge_dir, "event_id": event["id"],
                                "evidence": event["payload"], "files": base, "release": release, "today": today,
                                "external_texts": external, "generator_model": rt.generator.model}
-                    objectives.capture(rt.traces, "kb-intake.draft", f"{lifecycle.repo}#{event['id']}", payload)
-                    from ..llm import LLM
                     try:
-                        prediction = execute(draft_settings, rt.traces, "kb-intake.draft", payload, LLM(draft_settings))
-                    except Exception as exc:
-                        raise ModelUnavailable("autonomous draft rolled back: " + str(exc)) from exc
+                        objectives.capture(rt.traces, "kb-intake.draft", f"{lifecycle.repo}#{event['id']}", payload)
+                    except objectives.ReplayCaptureLimit as exc:
+                        # Ordinary pinned drafting does not require an evolution
+                        # replay. An active (even disabled) evolved workflow does.
+                        try:
+                            entry = active(draft_settings)
+                            workflows = entry.get("workflows", [])
+                            if not isinstance(workflows, list) or not all(isinstance(w, str) for w in workflows):
+                                raise ValueError("invalid active draft workflows")
+                            is_active = "kb-intake.draft" in workflows
+                        except Exception as registry_exc:
+                            raise ModelUnavailable("cannot verify active draft release: " + str(registry_exc)) from registry_exc
+                        if is_active:
+                            raise ModelUnavailable("active draft lacks a complete replay: " + str(exc)) from exc
+                    else:
+                        from ..llm import LLM
+                        try:
+                            prediction = execute(draft_settings, rt.traces, "kb-intake.draft", payload, LLM(draft_settings))
+                        except Exception as exc:
+                            raise ModelUnavailable("autonomous draft rolled back: " + str(exc)) from exc
                 if prediction is not None:
                     from .intake import Draft
                     from ..knowledge_service.ops import KnowledgeOperation

test/test_autonomous_evolution.py

@@ -198,6 +198,42 @@ def test_trace_inputs_are_automatically_frozen_and_clustered(tmp_path):
     with pytest.raises(artifacts.ArtifactError): objectives.rows(st, 'kb-intake.draft')
 
 
+def test_full_kb_replay_larger_than_old_limit_round_trips(tmp_path):
+    from infermatrix_copilot.trace_store import TraceStore
+    st = Settings(_env_file=None, improve_ledger_dir=str(tmp_path / 'ledger'))
+    store = TraceStore(tmp_path / 'trace')
+    payload = {'repo': 'demo', 'files': {'repos/demo/a.md': 'knowledge'},
+               'external_texts': {'adapters/demo/reference.md': '文' * 5_400_000}}
+    assert len(json.dumps(payload, ensure_ascii=False).encode()) > 16_000_000
+    objectives.capture(store, 'kb-intake.draft', 'demo#1', payload)
+    assert objectives.collect(st, store, 'kb-intake.draft') == 1
+    row = objectives.rows(st, 'kb-intake.draft')[0]
+    assert row['payload'] == payload
+
+
+def test_oversized_replay_is_explicit_and_ineligible(tmp_path, monkeypatch):
+    from infermatrix_copilot.trace_store import TraceStore
+    st = Settings(_env_file=None, improve_ledger_dir=str(tmp_path / 'ledger'))
+    store = TraceStore(tmp_path / 'trace')
+    monkeypatch.setattr(objectives, 'REPLAY_INPUT_MAX_BYTES', 80)
+    payload = {'files': {'other.md': '文' * 30}}
+    size = len(json.dumps(payload, ensure_ascii=False).encode())
+    assert len(json.dumps(payload, ensure_ascii=False)) < 80 < size
+    with pytest.raises(objectives.ReplayCaptureLimit) as exc:
+        objectives.capture(store, 'kb-intake.draft', 'demo#1', payload)
+    assert (exc.value.size_bytes, exc.value.limit_bytes) == (size, 80)
+    record, = store.query(workflow='kb-intake.draft')
+    assert record['inputs'] == {}
+    assert record['result'] == {'type': 'replay_capture_deferred', 'reason': 'size_limit',
+                                'input_bytes': size, 'limit_bytes': 80,
+                                'input_sha': artifacts.digest(json.dumps(payload, ensure_ascii=False).encode())}
+    assert objectives.collect(st, store, 'kb-intake.draft') == 0
+    assert objectives.rows(st, 'kb-intake.draft') == []
+    with pytest.raises(artifacts.ArtifactError, match='evaluator material') as exc:
+        objectives.capture(store, 'kb-intake.draft', 'demo#2', {'labels': []})
+    assert not isinstance(exc.value, objectives.ReplayCaptureLimit)
+
+
 def test_empty_kb_is_not_rewarded_as_an_error():
     row = {'payload': {'files': {}, 'repo_dir': 'repos/demo', 'release': 'v1', 'today': '2026-10-04'}}
     assert objectives.score('kb-intake', row, {'operations': [], 'rejected': False}) == 1