From e80e0484dfd8e36b501ec98ef6119962cbd5e6bc Mon Sep 17 00:00:00 2001 From: tzhouam Date: Fri, 9 Oct 2026 16:28:21 +0800 Subject: [PATCH] fix(kb): preserve complete replays without blocking baseline intake --- doc/architecture/SPEC/improve.md | 4 +- doc/architecture/SPEC/kb_service.md | 1 + src/infermatrix_copilot/improve/objectives.py | 24 +- src/infermatrix_copilot/kb_service/runtime.py | 27 ++- test/test_autonomous_evolution.py | 36 +++ test/test_kb_replay_capture.py | 219 ++++++++++++++++++ test/test_replay_capture_pr_review.py | 58 +++++ 7 files changed, 359 insertions(+), 10 deletions(-) create mode 100644 test/test_kb_replay_capture.py create mode 100644 test/test_replay_capture_pr_review.py diff --git a/doc/architecture/SPEC/improve.md b/doc/architecture/SPEC/improve.md index 3ee21694..4041199c 100644 --- a/doc/architecture/SPEC/improve.md +++ b/doc/architecture/SPEC/improve.md @@ -1,6 +1,6 @@ # improve/ —— 规范（元改进引擎） - + `设计：/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，但不自动采用。当前主机命名空间关闭，代码候选实际执行延期。 diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index f126909a..b8fbd517 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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 通用功能发现：文档基线与源码扩展 diff --git a/src/infermatrix_copilot/improve/objectives.py b/src/infermatrix_copilot/improve/objectives.py index d948b6e3..cb6f24ca 100644 --- a/src/infermatrix_copilot/improve/objectives.py +++ b/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): diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index 1db9d98b..b5fe6170 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/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 diff --git a/test/test_autonomous_evolution.py b/test/test_autonomous_evolution.py index 23a818f0..076956a4 100644 --- a/test/test_autonomous_evolution.py +++ b/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 diff --git a/test/test_kb_replay_capture.py b/test/test_kb_replay_capture.py new file mode 100644 index 00000000..90e9e388 --- /dev/null +++ b/test/test_kb_replay_capture.py @@ -0,0 +1,219 @@ +"""Replay limits must neither truncate knowledge input nor bypass an active arm.""" +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.config import Settings +from infermatrix_copilot.improve import artifacts, objectives, runtime as improve_runtime +from infermatrix_copilot.kb_service import runtime as kb_runtime +from infermatrix_copilot.kb_service.config import RepoLifecycle +from infermatrix_copilot.kb_service.intake import Draft +from infermatrix_copilot.kb_service.ledger import Ledger +from infermatrix_copilot.kb_service.models import ModelRole +from infermatrix_copilot.trace_store import TraceStore + + +WORKFLOW = "kb-intake.draft" + + +@pytest.fixture +def intake(tmp_path, monkeypatch): + settings = Settings( + _env_file=None, + improve_enabled=True, + improve_evolve_enabled=True, + improve_evaluation_mode="objective", + improve_ledger_dir=str(tmp_path / "improve"), + playbooks_dir=tmp_path / "playbooks", + eco_model="fixture-model", + ) + ledger = Ledger(tmp_path / "kb.db") + ledger.ensure_repo("demo", "shadow") + evidence = {"source_reference": "PR #17", "body": "Keep the complete evidence.", + "changed_files": ["owner/module.py"], "diff_excerpt": "+complete source\n"} + event_id = ledger.record_event("demo", "merged_pr", "17", evidence) + # Ledger serializes evidence with sorted keys; pin the bytes the runtime + # actually reads rather than a semantically equal, differently ordered dict. + evidence = ledger.event(event_id)["payload"] + files = { + "repos/demo/rules.md": "owner start\n" + "x" * 4096 + "\nowner end 完整\n", + "repos/other/rules.md": "unrelated owner is also retained\n", + } + external = {"README.md": "complete external context\n"} + generator = ModelRole.parse("generator", "codex:pinned-fixture") + rt = SimpleNamespace( + settings=settings, + ledger=ledger, + traces=TraceStore(tmp_path / "trace", environ={}), + knowledge=SimpleNamespace(fetch=lambda: "a" * 40, + knowledge_files=lambda _: files, + external_texts=lambda _: external), + release_for=lambda _: "v1", + today=lambda: "2026-10-09", + lease_owner=None, + generator=generator, + gateway=object(), + ) + lifecycle = RepoLifecycle(repo="demo", full_name="org/demo", enabled=True, + mode="shadow", knowledge_dir="repos/demo") + drafts = [] + + def pinned_draft(**kwargs): + drafts.append(kwargs) + return Draft(event_ids=[kwargs["event_id"]], operations=[], result=None) + + def forbidden_execute(*args, **kwargs): + pytest.fail("an uncaptured input must never execute an autonomous artifact") + + monkeypatch.setattr(kb_runtime, "draft_changes", pinned_draft) + monkeypatch.setattr(improve_runtime, "execute", forbidden_execute) + # Constructing a model is also unnecessary on the capture-overflow path. + monkeypatch.setattr("infermatrix_copilot.llm.LLM", lambda _: object()) + payload = {"repo": "demo", "repo_dir": "repos/demo", "event_id": event_id, + "evidence": evidence, "files": files, "release": "v1", "today": "2026-10-09", + "external_texts": external, "generator_model": generator.model} + return SimpleNamespace(rt=rt, lifecycle=lifecycle, files=files, payload=payload, + event_id=event_id, drafts=drafts, settings=settings) + + +def assert_deferred_capture_only(world): + records = world.rt.traces.query(workflow=WORKFLOW) + assert len(records) == 1 + record = records[0] + assert record["kind"] == "decision" + assert record["context"]["item"] == f"demo#{world.event_id}" + assert record["inputs"] == record["outputs"] == {} + encoded = json.dumps(world.payload, ensure_ascii=False).encode() + assert record["result"] == { + "type": "replay_capture_deferred", "reason": "size_limit", + "input_bytes": len(encoded), "limit_bytes": objectives.REPLAY_INPUT_MAX_BYTES, + "input_sha": artifacts.digest(encoded), + } + assert not list((world.rt.traces.root / "blobs").glob("**/*.gz")) + assert objectives.collect(world.settings, world.rt.traces, WORKFLOW) == 0 + assert objectives.rows(world.settings, WORKFLOW) == [] + + +@pytest.mark.parametrize("entry", [ + {}, + {"workflows": ["pr-review.agent.review_diff"]}, + {"workflows": ["pr-review.agent.review_diff"], "disabled": True}, +]) +def test_oversize_capture_keeps_full_pinned_baseline_without_partial_replay(intake, monkeypatch, entry): + monkeypatch.setattr(objectives, "REPLAY_INPUT_MAX_BYTES", 1024) + artifacts.atomic_json(improve_runtime.registry(intake.settings), entry) + + assert kb_runtime.run_intake(intake.rt, intake.lifecycle) is None + + assert len(intake.drafts) == 1 + draft = intake.drafts[0] + assert draft["files"] is intake.files + assert draft["files"] == intake.payload["files"] + assert draft["evidence"] == intake.payload["evidence"] + assert draft["generator"] is intake.rt.generator + assert draft["gateway"] is intake.rt.gateway + assert draft["release"] == intake.payload["release"] + assert draft["today"] == intake.payload["today"] + assert intake.rt.ledger.event(intake.event_id)["status"] == "done" + assert_deferred_capture_only(intake) + + +@pytest.mark.parametrize("disabled", [False, True]) +def test_oversize_active_knowledge_arm_stays_pending_even_when_disabled(intake, monkeypatch, disabled): + monkeypatch.setattr(objectives, "REPLAY_INPUT_MAX_BYTES", 1024) + artifacts.atomic_json(improve_runtime.registry(intake.settings), + {"workflows": [WORKFLOW], "disabled": disabled}) + + assert kb_runtime.run_intake(intake.rt, intake.lifecycle) is None + + assert intake.drafts == [] + event = intake.rt.ledger.event(intake.event_id) + assert event["status"] == "pending" + assert event["payload"] == intake.payload["evidence"] + assert "generator unavailable" in event["detail"] + assert_deferred_capture_only(intake) + + +@pytest.mark.parametrize("raw_registry", [ + "{broken json", + "null", + "[]", + '{"workflows": null}', + '{"workflows": 7}', + '{"workflows": "kb-intake.draft"}', +]) +def test_oversize_capture_with_unverifiable_active_registry_stays_pending(intake, monkeypatch, raw_registry): + monkeypatch.setattr(objectives, "REPLAY_INPUT_MAX_BYTES", 1024) + path = improve_runtime.registry(intake.settings) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(raw_registry) + + assert kb_runtime.run_intake(intake.rt, intake.lifecycle) is None + + assert intake.drafts == [] + event = intake.rt.ledger.event(intake.event_id) + assert event["status"] == "pending" + assert "cannot verify active draft release" in event["detail"] + assert_deferred_capture_only(intake) + + +def test_snapshot_above_old_limit_is_captured_and_collected_without_omissions(intake, monkeypatch): + target_bytes = 16_080_000 + original_size = len(json.dumps(intake.payload, ensure_ascii=False).encode()) + intake.files["repos/demo/rules.md"] += "y" * (target_bytes - original_size) + encoded = json.dumps(intake.payload, ensure_ascii=False).encode() + assert len(encoded) == target_bytes + assert 16_000_000 < len(encoded) < objectives.REPLAY_INPUT_MAX_BYTES + executions = [] + + def baseline_execute(settings, store, workflow, payload, llm): + executions.append(payload) + assert improve_runtime.active(settings) == {} + return None + + monkeypatch.setattr(improve_runtime, "execute", baseline_execute) + assert kb_runtime.run_intake(intake.rt, intake.lifecycle) is None + + assert executions == [intake.payload] + assert len(intake.drafts) == 1 + assert intake.drafts[0]["files"] == intake.files + records = intake.rt.traces.query(workflow=WORKFLOW) + assert len(records) == 1 + record = records[0] + assert record["result"] == {"type": "replay_input", "input_sha": artifacts.digest(encoded)} + assert intake.rt.traces.blob(record["inputs"]["evolution_input"]).encode() == encoded + assert objectives.collect(intake.settings, intake.rt.traces, WORKFLOW) == 1 + rows = objectives.rows(intake.settings, WORKFLOW) + assert len(rows) == 1 + assert rows[0]["item"] == f"demo#{intake.event_id}" + assert rows[0]["payload"] == intake.payload + assert rows[0]["origin"] == {"kind": "trace-replay", "record": record["id"]} + assert intake.rt.ledger.event(intake.event_id)["status"] == "done" + + +@pytest.mark.parametrize("failure", ["evaluator_material", "trace_io"]) +def test_capture_failures_other_than_typed_size_limit_remain_hard(intake, monkeypatch, failure): + if failure == "evaluator_material": + original = objectives.capture + + def forbidden_capture(store, workflow, item, payload): + return original(store, workflow, item, {**payload, "labels": {"must_not_replay": True}}) + + monkeypatch.setattr(objectives, "capture", forbidden_capture) + error, message = artifacts.ArtifactError, "evaluator material" + else: + def failed_append(*args, **kwargs): + raise OSError("trace disk unavailable") + + monkeypatch.setattr(intake.rt.traces, "append", failed_append) + error, message = OSError, "trace disk unavailable" + + with pytest.raises(error, match=message): + kb_runtime.run_intake(intake.rt, intake.lifecycle) + + assert intake.drafts == [] + assert intake.rt.ledger.event(intake.event_id)["status"] == "pending" + assert intake.rt.traces.query(workflow=WORKFLOW) == [] diff --git a/test/test_replay_capture_pr_review.py b/test/test_replay_capture_pr_review.py new file mode 100644 index 00000000..ab45ddad --- /dev/null +++ b/test/test_replay_capture_pr_review.py @@ -0,0 +1,58 @@ +"""Oversize replay capture preserves the PR review release boundary.""" + +import asyncio +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.config import Settings +from infermatrix_copilot.engine.step import FailureKind +from infermatrix_copilot.improve import evolution, objectives, runtime +from infermatrix_copilot.trace_store import TraceStore, bind_store + + +WORKFLOW = 'pr-review.agent.review_diff' + + +@pytest.mark.parametrize('entry,blocked', [ + ({}, False), + ({'workflows': ['other.workflow'], 'candidate': 'other', 'disabled': True}, False), + ({'workflows': [WORKFLOW], 'candidate': 'review-candidate'}, True), + ({'workflows': [WORKFLOW], 'candidate': 'review-candidate', 'disabled': True}, True), +], ids=['baseline', 'unrelated-disabled-release', 'active-review', 'disabled-review']) +def test_oversize_review_replay_never_executes_or_becomes_dataset_input(tmp_path, monkeypatch, entry, blocked): + store = TraceStore(tmp_path / 'traces', environ={}) + settings = Settings(_env_file=None, improve_ledger_dir=str(tmp_path / 'ledger'), + improve_evolve_data_dir=str(tmp_path / 'inputs')) + frozen = {'repo': 'demo', 'pr': 1, 'base_sha': 'b' * 40, 'head_sha': 'a' * 40, + 'base_files': {'x.py': 'old\n'}, 'head_files': {'x.py': 'new\n'}, + 'diff': 'unabridged evidence\n' * 30, 'knowledge_files': {}} + candidate = {'id': 'review-candidate'} + rollbacks = [] + monkeypatch.setattr(objectives, 'REPLAY_INPUT_MAX_BYTES', 128) + monkeypatch.setattr(runtime, 'active', lambda _: entry) + monkeypatch.setattr(runtime, 'freeze_review', lambda *_: frozen) + monkeypatch.setattr(evolution, 'candidate', lambda _, identifier: candidate if identifier == candidate['id'] else pytest.fail('wrong release')) + monkeypatch.setattr(runtime, 'rollback', lambda *args: rollbacks.append(args)) + monkeypatch.setattr(runtime, 'execute', lambda *args, **kwargs: pytest.fail('uncaptured input reached evolved review')) + ctx = SimpleNamespace(settings=settings, state={}, llm=None) + + with bind_store(store): + result = asyncio.run(runtime.review_step(ctx)) + + if blocked: + assert result.ok is False and result.failure == FailureKind.BLOCKED + assert result.summary == 'active review lacks a safe frozen replay input' + assert len(rollbacks) == 1 and rollbacks[0][2] is candidate + else: + assert result is None # the ordinary review handler continues + assert rollbacks == [] + records = store.query(limit=20) + assert not any(r['result'].get('type') == 'replay_input' for r in records) + deferred = [r for r in records if r['result'].get('type') == 'replay_capture_deferred'] + assert len(deferred) == 1 and deferred[0]['inputs'] == {} + assert deferred[0]['context']['item'] == 'demo#1@' + 'a' * 40 + assert deferred[0]['result']['input_bytes'] > deferred[0]['result']['limit_bytes'] == 128 + assert any(r['result'].get('type') == 'replay_capture_missing' for r in records) + assert objectives.collect(settings, store, WORKFLOW) == 0 + assert objectives.rows(settings, WORKFLOW) == []