From 660f50c1ca5b72249bfae03fdc8654d4e38ff491 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Mon, 5 Oct 2026 22:33:03 +0800 Subject: [PATCH] Preserve catalog gates through init reruns and parallel deepening --- doc/architecture/SPEC/kb_service.md | 21 +- eval/feature-discovery/README.md | 6 + eval/knowledge-depth/README.md | 2 + eval/knowledge-depth/run_depth_campaign.py | 135 +++++++- .../skills/imkbinit/SKILL.md | 24 +- src/infermatrix_copilot/kb_service/cli.py | 10 +- .../kb_service/init_feature_discovery.py | 266 ++++++++++++++- .../kb_service/init_stages.py | 11 +- .../kb_service/init_support.py | 5 + test/test_kb_breadth_subscription.py | 162 +++++++++ ...est_kb_depth_campaign_discovery_handoff.py | 199 +++++++++++ test/test_kb_feature_discovery_boundary.py | 314 ++++++++++++++++++ test/test_kb_unlimited_subscription.py | 4 +- 13 files changed, 1128 insertions(+), 31 deletions(-) create mode 100644 test/test_kb_breadth_subscription.py create mode 100644 test/test_kb_depth_campaign_discovery_handoff.py create mode 100644 test/test_kb_feature_discovery_boundary.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 8afa8c16..cefbdf18 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -596,11 +596,15 @@ proof `basis` 缺省 supported，旧区块不改写。verified_absent 当前只 判定时的编号源码正文须完整覆盖每个区块证明并匹配内容哈希，缺失区块还须绑定相同的 重放证书；显式报告非 Codex 裁判身份则拒绝，未报告身份仍保留未知。 -`--unlimited-subscription` 仅用于 knowledge-deepen/deepen，显式要求生成和裁判两角色 +`--unlimited-subscription` 用于 feature-discovery、modules、knowledge、knowledge-deepen +及 widen/deepen 别名，显式要求生成和裁判两角色 分别使用 Zcode GLM-5.3 与独立 Codex 的已认证订阅，拒绝其他模型协议、budget_usd 冲突 及 configured fallback。预算上限 None 表示 该模式，不使用无穷或零价格伪装；累计固定记账和原生调用用量仍记录，实际费用未报告 保持未知。每次调用重新绑定已经检查的订阅传输，不启用 API 付费回退。 +modules/knowledge 的公共阶段预算仅在显式启用时改为 None；单独选择订阅生成器 +仍保留原记账上限。该模式绑定检查点身份，无此选项时的旧摘要保持不变； +feature-discovery 与 knowledge-deepen 继续使用各自已有的检查点及身份实现。 ### 2026-10-02 轻量认可、13 功能任务与轨迹留存 @@ -628,6 +632,12 @@ Zcode 调度文件。默认启动间隔 15 秒、原生 429/account-1302 退避 调度等待、开始、冷却与停止事件与原生输入输出一起脱敏保存于每个 worker 的 `init/traces/attempts/`，最终关联唯一 model_call；未完成调用及未报告账单费用保持未知。 监控读取错误记录诊断、沿用最后安全快照，不停止其它任务。 +发现目录合并后的并行深度批次使用 `--discovery-record` 显式传入原始完成记录。 +父任务复用正式阶段的前置门禁，检查真实 PR 已合并或原记录为 empty、done、repo/pin， +以及基线中的目录与报告哈希；原 JSON 字节摘要进入批次身份，逐 worker 原样复制， +不重写状态或外部原生档案路径。基线已有发现报告时，即使旧适配器未启用必需标记， +也不能省略交接。worker 在模型初始化前再核对父任务绑定，记录变化或既有副本冲突 +拒绝恢复；没有发现报告的旧流程保持原行为。 新增轻量区块的原生审计还须绑定实际 GLM-5.3 草稿的对应维度、正文和已展示引用，以及 Zcode 身份和无 fallback 记录；附近的无关生成调用不能证明该区块的提取归属。 @@ -867,3 +877,12 @@ Zcode 模型角色中的 `effort` 被 transport 忽略，不能代替该设置 只能称“声明库存已处理”，不能证明所有功能均已发现。新目录的深度分母为 `N × 7`； 历史目录的认可率独立保留，不能直接混用。功能发现、知识认可、生产文件结构覆盖、 测试覆盖和真实 PR 评审效果是不同指标。 + +首次评审支持的生产源码新候选，进入目录前必须再经一次批量 Codex 目录边界审计。 +该审计展示完整的正式功能 ID、标题和 owner，不让正式功能与候选竞争 16000 字符预算； +有剩余空间才展示相关候选。保留首次评审，另存真实原生审计回执、审计版本、种子哈希、 +候选/首次评审集合哈希及每项候选哈希。拒绝、通道失败、别名循环或未解决的重复标题/别名 +保持未知，不触发另一轮提取修正。跨候选别名只可归属最终已认可的正式或新增父功能。 +未发布的旧检查点缺少此审计时只补审，复用完整扫描和首次评审；prepared 产物缺少审计时 +阻止发布，保留原产物。已发布历史目录保持不可变。审计预算中断保存已完成批次，恢复时 +不重复已完成的审计。显式 `--retry-unfinished` 不重置已用完的三次内容修正，仅恢复通道故障。 diff --git a/eval/feature-discovery/README.md b/eval/feature-discovery/README.md index 0be5490a..963f3fed 100644 --- a/eval/feature-discovery/README.md +++ b/eval/feature-discovery/README.md @@ -32,6 +32,12 @@ infermatrix-copilot kb init REPO --stage feature-discovery --from-existing \ 字段时使用新状态目录，保留旧轨迹。扩大包不会证明已发现所有功能，也不取消独立评审。 恰好达到每包 24 个候选上限的任务另列数量和 ID，保留为潜在遗漏线索。 +新增正式功能在首次认可后，还须经过一次目录边界审计。独立 Codex 同时看到所有既有 +功能的 ID、标题和 owner，并读取候选的真实源码引用；子能力、别名和实现补充归属父功能。 +首次评审及轨迹保留，边界审计另绑定版本、种子和候选集合哈希。重复标题/别名仍无法归属 +时保持未知，不增加分母。未发布旧批次可以复用扫描与首次评审补做这一步；内容修正已经 +达到三次上限的候选不会因显式重试再获得三次机会。 + 检查点绑定源码、范围、种子目录、索引版本、提示词与模型配置。成功分片、评审和待审修正稿可复用；预算中断不标为完成。竞争启动同一批次不会覆盖检查点。单项最多三次修正；通道故障保持未知，不重做已成功的提取。仅有文档的未知声明不通过重复改写变成实现证明。 紧凑摘要写入 `eval/feature-discovery/`。完整输入、原生输出、工具事件、配置、用量和修正过程留在 Git 外。恢复时核验每个完成分片的原生归档，包括返回空候选的分片。Zcode 保存原生流式事件；Codex 保存 CLI 结束后按原顺序转发的事件。 diff --git a/eval/knowledge-depth/README.md b/eval/knowledge-depth/README.md index 858ac379..8341f8b2 100644 --- a/eval/knowledge-depth/README.md +++ b/eval/knowledge-depth/README.md @@ -73,6 +73,8 @@ JiuwenSwarm explicitly selects `semantic_depth.acceptance_mode: lightweight`; ot The campaign runner defaults to 13 isolated workers and accepts `--workers 13`. Each worker owns its checkpoint, model gateway and trace archive. The parent builds one immutable source/test/document index bound to the source pin, policy and index version. Initial extraction packets offer up to 24 KB of source/test lines, 8 KB of relevant document lines and 4 KB of existing knowledge; subsequent corrections retrieve evidence for the specific gap. Format, references, hashes, receipts and counts are checked incrementally; comprehensive audits, retrieval acceptance and full CI run at delivery. +For a catalog produced by feature discovery, pass `--discovery-record /absolute/path/to/init//feature-discovery.json` to `run_depth_campaign.py`. Use the original completed `published` record after its PR merges, or the original `empty` record when the catalog was already current. The native prerequisite gate checks the real PR state, source pin and exact catalog/report hashes at the campaign baseline. The campaign binds the record's byte hash and copies its exact bytes into every worker before dispatch; native trace paths stay unchanged outside Git. A baseline containing a discovery report requires this explicit handoff, including repositories whose older manifest did not require discovery. Changed records or mismatched worker prerequisites require a fresh campaign; statuses and prerequisite records are never synthesized. Legacy baselines without discovery retain their existing behavior. + Each missing facet receives an initial attempt and at most three corrections. The round is saved before dispatch, including review-only retries. Restarting or `--retry-unfinished` does not reset this lifetime limit. Passed facets are frozen; failure in another facet does not discard them. After exhausted repairs, the facet remains unknown with its reason. Creating the campaign's STOP file stops new dispatch and drains active calls; the supervisor records progress in `drain.json`. Validation knowledge labels runtime assertions, source-text assertions, helper-unit assertions or existing documented manual checks. Manual entries include their documented action and expected observation and explicitly say this batch did not execute them. Missing tests never become verified absence, and knowledge about assertions does not mean tests passed. diff --git a/eval/knowledge-depth/run_depth_campaign.py b/eval/knowledge-depth/run_depth_campaign.py index 119d9a43..a3916dca 100644 --- a/eval/knowledge-depth/run_depth_campaign.py +++ b/eval/knowledge-depth/run_depth_campaign.py @@ -25,6 +25,109 @@ def git(*args): subprocess.run(["git", *map(str, args)], check=True, stdout=subprocess.DEVNULL) +def baseline_inputs(args): + """Resolve the adapter from the immutable baseline, including aliased dirs.""" + import yaml + from infermatrix_copilot.kb_service.knowledge_coverage import policy_path + + names = subprocess.check_output(["git", "-C", str(args.root), "ls-tree", "-r", "--name-only", + args.baseline, "adapters"], text=True).splitlines() + found = [] + for name in names: + if not re.fullmatch(r"adapters/[A-Za-z0-9_-]+/manifest\.yaml", name): + continue + text = subprocess.check_output(["git", "-C", str(args.root), "show", args.baseline + ":" + name], text=True) + manifest = yaml.safe_load(text) + if isinstance(manifest, dict) and isinstance(manifest.get("knowledge"), dict) \ + and manifest["knowledge"].get("repo_subdir") == "repos/" + args.repo: + found.append((name, manifest)) + if len(found) != 1: + raise ValueError("campaign baseline must contain exactly one adapter for this knowledge repository") + name, manifest = found[0] + relative = policy_path(args.repo, Path(name).parent) + text = subprocess.check_output(["git", "-C", str(args.root), "show", args.baseline + ":" + relative], text=True) + return manifest, text, relative + + +def discovery_handoff(args): + """Validate a genuine completed record against the merged frozen catalog.""" + from types import SimpleNamespace + from infermatrix_copilot.kb_service.init_stages import _Chain, _Stage + from infermatrix_copilot.kb_service.init_support import InitRecord + from infermatrix_copilot.kb_service.sources import KnowledgeRepo + + manifest, _, relative = baseline_inputs(args) + report_path = f"eval/feature-discovery/{args.repo}-{args.pin[:12]}.json" + knowledge = KnowledgeRepo(args.root) + report = knowledge.show(args.baseline, report_path) + lifecycle = manifest.get("knowledge_lifecycle") or {} + configured = lifecycle.get("init") or {} if isinstance(lifecycle, dict) else None + if not isinstance(configured, dict): + raise ValueError("campaign baseline lifecycle/init must be a mapping") + path = getattr(args, "discovery_record", None) + if path is None: + if report is not None or configured.get("feature_discovery_required"): + raise ValueError("this campaign baseline requires --discovery-record; the frozen catalog gate cannot be omitted") + return None + path = Path(path).resolve() + raw = path.read_bytes() + record_hash = hashlib.sha256(raw).hexdigest() + expected = getattr(args, "discovery_record_sha256", None) + if expected is not None and record_hash != expected: + raise ValueError("discovery prerequisite record changed after campaign binding") + data = json.loads(raw) + if not isinstance(data, dict) or not isinstance(data.get("discovery"), dict): + raise ValueError("discovery handoff record must contain a discovery mapping") + record = InitRecord(**data) + if record.stage != "feature-discovery" or record.repo != args.repo or record.pin != args.pin \ + or record.status not in ("published", "empty") or record.discovery.get("done") is not True \ + or record.status == "published" and record.dry_run: + raise ValueError("discovery handoff requires the original published/empty completed record for this repository and pin") + # Reuse the production gate: verify real PR state, complete report, feature + # IDs, owners and the exact catalog/report hashes at the merged baseline. + # The temporary file is the original record's bytes, never a replacement + # status or a fabricated prerequisite. Native archives stay at their origin. + with tempfile.TemporaryDirectory(prefix="discovery-handoff-") as scratch: + state = Path(scratch) + copied = InitRecord.path(state, args.repo, "feature-discovery") + copied.parent.mkdir(parents=True) + copied.write_bytes(raw) + stage = _Stage(SimpleNamespace(state_dir=state, knowledge=knowledge, gh_run=subprocess.run), + SimpleNamespace(repo=args.repo, knowledge_dir="repos/" + args.repo, + adapter_dir=Path(relative).parent, + init=SimpleNamespace(feature_discovery_required=True)), + dry_run=False, pin=args.pin) + stage.STAGE = "knowledge-deepen" + stage.repo_dir, stage._base_sha = "repos/" + args.repo, args.baseline + chain = _Chain() + stage._discovery_gate(chain, args.pin) + if chain.problems: + raise ValueError("discovery handoff failed the merged catalog gate: " + "; ".join(chain.problems)) + binding = stage._discovery_binding() + return {"record_path": str(path), "record_sha256": record_hash, + "report_path": report_path, **binding} + + +def install_discovery_handoff(args, state, binding): + """Copy the exact immutable prerequisite into a worker without rewriting it.""" + if binding is None: + return + from infermatrix_copilot.kb_service.init_support import InitRecord + + raw = Path(binding["record_path"]).read_bytes() + if hashlib.sha256(raw).hexdigest() != binding["record_sha256"]: + raise ValueError("discovery prerequisite record changed before worker handoff") + dest = InitRecord.path(state, args.repo, "feature-discovery") + if dest.exists(): + if dest.read_bytes() != raw: + raise ValueError("worker discovery prerequisite differs from the immutable campaign record") + return + dest.parent.mkdir(parents=True, exist_ok=True) + temporary = dest.with_suffix(".json.tmp") + temporary.write_bytes(raw) + os.replace(temporary, dest) + + def worker(args): state = args.state / f"worker-{args.worker}" state.mkdir(parents=True, exist_ok=True) @@ -53,6 +156,12 @@ def run_worker(args, state): previous = InitRecord.load(state, args.repo, "knowledge-deepen") if previous and (previous.kb_base_sha != args.baseline or previous.pin != args.pin): raise RuntimeError("worker checkpoint source or knowledge baseline differs from campaign") + binding = discovery_handoff(args) + if binding is not None: + metadata = json.loads((args.state / "campaign.json").read_bytes()) + if metadata.get("discovery") != binding: + raise RuntimeError("worker discovery handoff differs from the frozen campaign identity") + install_discovery_handoff(args, state, binding) runtime = InitRuntime.from_env(Settings(_env_file=None), state_dir=state) from infermatrix_copilot.kb_service.depth_pacing import SharedZcodePacer runtime.gateway.configure_zcode_pacing(SharedZcodePacer( @@ -88,6 +197,9 @@ def main(): parser.add_argument("--depth-index-path", type=Path) parser.add_argument("--repair-guidance", type=Path, help="Pinned, immutable localization hints for a newly authorized gap-repair batch") parser.add_argument("--repair-guidance-sha256", help=argparse.SUPPRESS) + parser.add_argument("--discovery-record", type=Path, + help="Original completed, merged feature-discovery record; required for a discovered catalog") + parser.add_argument("--discovery-record-sha256", help=argparse.SUPPRESS) parser.add_argument("--zcode-pacing-path", type=Path) parser.add_argument("--zcode-start-interval", type=float, default=15.0) parser.add_argument("--zcode-rate-cooldown", type=float, default=90.0) @@ -105,6 +217,8 @@ def main(): args.depth_index_path = args.depth_index_path.resolve() if args.repair_guidance: args.repair_guidance = args.repair_guidance.resolve() + if args.discovery_record: + args.discovery_record = args.discovery_record.resolve() if args.zcode_pacing_path: args.zcode_pacing_path = args.zcode_pacing_path.resolve() from infermatrix_copilot.kb_service.depth_pacing import SharedZcodePacer @@ -118,6 +232,8 @@ def main(): parser.error("worker must belong to the campaign and have feature IDs") if args.repair_guidance and not re.fullmatch(r"[0-9a-f]{64}", args.repair_guidance_sha256 or ""): parser.error("guided workers require their parent's immutable guidance hash") + if args.discovery_record and not re.fullmatch(r"[0-9a-f]{64}", args.discovery_record_sha256 or ""): + parser.error("discovery workers require their parent's immutable prerequisite hash") return worker(args) args.baseline = subprocess.check_output(["git", "-C", str(args.root), "rev-parse", args.baseline + "^{commit}"], text=True).strip() args.state.mkdir(parents=True, exist_ok=True) @@ -130,13 +246,11 @@ def main(): def campaign(args, parser): - from infermatrix_copilot.kb_service.knowledge_coverage import load_policy, policy_path + from infermatrix_copilot.kb_service.knowledge_coverage import load_policy from infermatrix_copilot.kb_service.sources import KnowledgeRepo - import yaml from infermatrix_copilot.kb_service.outbox import atomic_write_json - manifest = yaml.safe_load((args.root / f"adapters/{args.repo}/manifest.yaml").read_text()) - text = subprocess.check_output(["git", "-C", str(args.root), "show", args.baseline + ":" + policy_path(args.repo)], text=True) + manifest, text, _ = baseline_inputs(args) policy = load_policy(text, manifest["knowledge"]["repo_subdir"]) if policy.semantic_depth_per_facet_gt is None: parser.error("campaign baseline must declare its global semantic completion target") @@ -156,6 +270,12 @@ def campaign(args, parser): "features": len(policy.features), "denominator": len(policy.features) * 7, "workers": args.workers, "partitions": partitions, "acceptance_mode": getattr(args, "acceptance_mode", "strict")} + try: + binding = discovery_handoff(args) + except (ValueError, TypeError, OSError) as exc: + parser.error(str(exc)) + if binding is not None: + summary["discovery"] = binding if getattr(args, "repair_guidance", None): from infermatrix_copilot.kb_service.depth_inputs import load_repair_guidance guidance = load_repair_guidance(args.repair_guidance, pin=args.pin, @@ -206,6 +326,10 @@ def campaign(args, parser): jobs, checkpoints = [], [] for number in range(args.workers): state = args.state / f"worker-{number}" + try: + install_discovery_handoff(args, state, binding) + except (ValueError, OSError) as exc: + parser.error(str(exc)) clone = state / "init/knowledge-repo" if not clone.exists(): clone.parent.mkdir(parents=True, exist_ok=True) @@ -227,6 +351,9 @@ def campaign(args, parser): if getattr(args, "repair_guidance", None): argv += ["--repair-guidance", str(args.repair_guidance), "--repair-guidance-sha256", summary["repair_guidance_sha256"]] + if binding is not None: + argv += ["--discovery-record", binding["record_path"], + "--discovery-record-sha256", binding["record_sha256"]] if args.retry: argv.append("--retry") jobs.append((number, argv)) diff --git a/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md b/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md index 1f2a586e..fd043e89 100644 --- a/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md +++ b/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md @@ -96,6 +96,12 @@ checks the merged index and every owner route. Existing stage records still keep their review and merge gates; never manufacture records to skip them. An enabled or started feature-discovery batch must still finish and its catalog PR must merge before publishing modules or knowledge. +When using `eval/knowledge-depth/run_depth_campaign.py` after discovery, +pass `--discovery-record` with the original completed record. Its byte hash, +merged catalog and report are checked before any worker dispatch; each worker +receives the exact original JSON, retaining external archive paths. A baseline +with a discovery report requires this handoff even for older adapters. Never +rewrite a prior status or create a substitute record to satisfy the gate. For implementation knowledge in an existing KB, use the independent stage `knowledge-deepen` (`kb deepen ` is its `--from-existing` alias). @@ -125,11 +131,12 @@ this mode; default API spend-cap checks remain in place. Report generator USD as unreported and subscription fees outside stage USD accounting, alongside token usage and the judge's accounted spend. Never claim a measured zero cost. -When the user explicitly selects uncapped subscription extraction, the independent -`knowledge-deepen` stage and `kb deepen` accept `--unlimited-subscription`: +When the user explicitly selects uncapped subscription extraction, +`feature-discovery`, `modules`, `knowledge`, `knowledge-deepen`, `kb widen` +and `kb deepen` accept `--unlimited-subscription`: ```text -KB_GENERATOR=zcode:GLM-5.3:low KB_JUDGE=codex:gpt-6-sol:medium kb deepen --pin --subscription-generator --unlimited-subscription --dry-run +ZCODE_REASONING_LEVEL=low KB_GENERATOR=zcode:GLM-5.3 KB_JUDGE=codex:gpt-6.1-sol:medium kb deepen --pin --subscription-generator --unlimited-subscription --dry-run ``` The generator must be Zcode GLM-5.3 and the independent judge must be Codex; @@ -139,6 +146,11 @@ accounting ceiling, and preserves call, served-model and token receipts. It does not authorize API spending or fallback to another provider; unavailable subscription roles stop with their reason. Actual USD remains unknown when the provider does not report it. Keep the requested source pin and reviewed policy. +For breadth reruns, explicitly set the same `KB_GENERATOR` and `KB_JUDGE` +roles and pass this flag to both `modules --from-existing` and +`knowledge --from-existing`; `--subscription-generator` alone preserves the +normal stage ceiling for judge accounting. The unlimited mode has a distinct +checkpoint identity; old runs with no flag keep their existing identity. Only publish when the user explicitly asks (`--publish`, "open the PR"). Publishing needs `ALLOW_PUSH=1`, `ALLOW_POST=1` and @@ -163,6 +175,12 @@ What each stage opens: Changed settings or old checkpoints without them require a fresh state directory; retain the previous archive. Larger packets do not prove all features were found. + Before promoting implemented new IDs, run the supplemental independent + catalog boundary audit against every formal feature's ID, title and owner. + Keep primary reviews and native receipts; unsupported or unresolved duplicate + titles/aliases stay unknown without another generator repair loop. Unpublished + old previews reuse their scan and primary reviews to acquire this audit. + Exhausted content repairs remain exhausted on `--retry-unfinished`. `--from-existing` checks merged skeleton routes, while `--retry-unfinished` resumes supported work. `--budget-usd` controls cumulative accounting; `--unlimited-subscription` authenticates both roles and removes that ceiling. diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index 7edaa200..8d5abd56 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -118,8 +118,8 @@ def _init_command(args, state_dir: Path) -> int: if args.suggest_seeds == bool(args.stage): print("kb init: pass exactly one of --stage or --suggest-seeds", file=sys.stderr) return 2 - if args.unlimited_subscription and (args.stage not in ("feature-discovery", "knowledge-deepen") or args.budget_usd is not None): - print("--unlimited-subscription is for feature-discovery or knowledge-deepen only and conflicts with --budget-usd", file=sys.stderr) + if args.unlimited_subscription and (args.stage not in ("feature-discovery", "modules", "knowledge", "knowledge-deepen") or args.budget_usd is not None): + print("--unlimited-subscription is for feature-discovery, modules, knowledge or knowledge-deepen only and conflicts with --budget-usd", file=sys.stderr) return 2 if args.suggest_seeds: from .init_stages import suggest_seeds @@ -231,7 +231,7 @@ def main(argv: list[str] | None = None) -> int: init.add_argument("--subscription-generator", action="store_true", help="explicit subscription generator; unreported fees outside stage USD accounting") init.add_argument("--unlimited-subscription", action="store_true", - help="uncapped discovery/depth with authenticated subscription generator and judge; no fallback") + help="uncapped discovery/modules/knowledge/depth with authenticated subscription generator and judge; no fallback") init.add_argument("--retry-unfinished", action="store_true", help="retry unfinished discovery/depth work, preserving prior spend") init.add_argument("--acceptance-mode", choices=("strict", "lightweight"), default="strict", help="depth recognition standard; lightweight uses pinned citations and one independent feature review") @@ -241,13 +241,13 @@ def main(argv: list[str] | None = None) -> int: cmd.add_argument("--dry-run", action="store_true") cmd.add_argument("--pin") cmd.add_argument("--subscription-generator", action="store_true") + cmd.add_argument("--unlimited-subscription", action="store_true") if name == "deepen": cmd.add_argument("--budget-usd", type=float) cmd.add_argument("--retry-unfinished", action="store_true") - cmd.add_argument("--unlimited-subscription", action="store_true") cmd.add_argument("--acceptance-mode", choices=("strict", "lightweight"), default="strict") cmd.set_defaults(stage=stage, from_existing=True, suggest_seeds=False, pr_count=None, - **({"budget_usd": None, "retry_unfinished": False, "unlimited_subscription": False} + **({"budget_usd": None, "retry_unfinished": False} if name == "widen" else {})) publish = sub.add_parser("publish") where = publish.add_mutually_exclusive_group(required=True) diff --git a/src/infermatrix_copilot/kb_service/init_feature_discovery.py b/src/infermatrix_copilot/kb_service/init_feature_discovery.py index 74fa5ac9..daaca788 100644 --- a/src/infermatrix_copilot/kb_service/init_feature_discovery.py +++ b/src/infermatrix_copilot/kb_service/init_feature_discovery.py @@ -33,6 +33,7 @@ MAX_CANDIDATES = 24 MAX_ATTEMPTS = 4 MAX_BASELINE_CHARS = 16000 +CATALOG_BOUNDARY_VERSION = "catalog-boundary-v1" RELATIONS = {"new", "implementation_supplement", "alias", "subcapability", "shared_component", "outdated", "unknown"} _SLUG = re.compile(r"[a-z0-9][a-z0-9-]{0,39}\Z") @@ -159,6 +160,58 @@ def _bounded_baseline(rows, text, preferred=()): return chosen +def _formal_catalog(seeds): + return [{"catalog_status": "existing_feature", **{k: row.get(k) for k in ("id", "title", "owner")}} + for row in seeds] + + +def _catalog_names(row): + return {name for value in [row["id"], row["title"], *row.get("aliases", [])] + if (name := unicodedata.normalize("NFKC", str(value)).casefold().strip())} + + +def _boundary_identity(seeds, state, *, repository, pin): + return {"version": CATALOG_BOUNDARY_VERSION, "repository": repository, "pin": pin, + "scan_identity": state.get("identity"), "seed_sha256": _hash(seeds), + "index_sha256": state.get("index_sha256"), + "formal_catalog_sha256": _hash(_formal_catalog(seeds)), + "candidate_set_sha256": _hash([{ "id": key, "candidate_sha256": _hash(row), + "primary_review_sha256": _hash(state.get("reviews", {}).get(key, {}))} + for key, row in sorted(state.get("candidates", {}).items())])} + + +def _boundary_context(marker): + return {"identity": marker["identity"], "candidate_ids": marker["candidate_ids"]} + + +def _boundary_current(row, primary, checked): + return bool(checked) and _decision_valid(checked) \ + and checked.get("candidate_sha256") == _hash(row) \ + and checked.get("primary_review_sha256") == _hash(primary) \ + and (checked.get("supported") != "yes" or bool(checked.get("judge_receipt"))) + + +def _boundary_ready(seeds, state, *, repository, pin, production_paths=None): + marker = state.get("catalog_boundary_audit", {}) + if marker.get("identity") != _boundary_identity(seeds, state, repository=repository, pin=pin) \ + or marker.get("done") is not True: + return False + keys = marker.get("candidate_ids") + if not isinstance(keys, list) or any(not isinstance(k, str) for k in keys) or len(set(keys)) != len(keys): + return False + if production_paths is not None: + production = set(production_paths) + eligible = {key for key, row in state.get("candidates", {}).items() + if not row.get("invalid_reason") and any(ref["path"] in production for ref in row["evidence"]) + and row.get("generator_receipts") and state.get("reviews", {}).get(key, {}).get("judge_receipt") + and state["reviews"][key].get("supported") == "yes" and state["reviews"][key].get("relation") == "new" + and state["reviews"][key].get("candidate_sha256") == _hash(row)} + if set(keys) != eligible: + return False + return all(k in state.get("candidates", {}) and _boundary_current(state["candidates"][k], + state.get("reviews", {}).get(k, {}), state.get("boundary_reviews", {}).get(k, {})) for k in keys) + + def _decision_valid(row): return isinstance(row, dict) and isinstance(row.get("supported"), str) and row["supported"] in ("yes", "no", "unsure") \ and isinstance(row.get("relation"), str) and row["relation"] in RELATIONS and isinstance(row.get("related_id", ""), str) \ @@ -385,11 +438,28 @@ def _context(self, rows): # Preserve complete supplied cited ranges, chunking review batches by caller. return [evidence_excerpt(self.index, ref) for ref in {_json(r): r for r in refs}.values()] - def _review(self, rows, baseline): + def _review(self, rows, baseline, *, audit=None): + if audit is not None: + formal = _formal_catalog(self.seeds) + # Formal IDs cannot compete with proposals for the baseline budget. + # Always include every formal entry, even for large repositories. + room = max(0, MAX_BASELINE_CHARS - sum(len(_json(r)) for r in formal)) + neighbors = _bounded_baseline([r for r in baseline if r.get("catalog_status") != "existing_feature"], + _json(rows), [r["id"] for r in rows] + [r.get("related_id") for r in rows]) + chosen = [] + for neighbor in neighbors: + size = len(_json(neighbor)) + if size <= room: + chosen.append(neighbor); room -= size + offered_baseline = formal + chosen + else: + offered_baseline = _bounded_baseline(baseline, _json(rows), + [r["id"] for r in rows] + [r.get("related_id") for r in rows]) payload = {"repository": self.repository, "pin": self.pin, "candidates": rows, - "baseline": _bounded_baseline(baseline, _json(rows), - [r["id"] for r in rows] + [r.get("related_id") for r in rows]), + "baseline": offered_baseline, "evidence": self._context(rows)} + if audit is not None: + payload["catalog_boundary_audit"] = audit reply = self.call("judge", SYSTEM_REVIEW, _prompt(payload), lambda d: None) proof = receipt(reply) decisions = reply.data.get("decisions", {}) @@ -488,14 +558,87 @@ def review(self): self.save() return True - def catalog(self, repo_dir): + def _implemented_approval(self, row, judgment): + from .feature_discovery_index import validate_evidence + return not row.get("invalid_reason") and any(ref["path"] in self.index.production for ref in row["evidence"]) \ + and all(validate_evidence(self.index, ref) for ref in row["evidence"]) \ + and judgment.get("supported") == "yes" and bool(row.get("generator_receipts")) \ + and bool(judgment.get("judge_receipt")) and judgment.get("candidate_sha256") == _hash(row) + + def audit_catalog_boundaries(self): + """One additional independent audit before promoting implemented new IDs. + + Retain primary reviews. Rejections and missing replies become unknown; + they do not enter the generator repair loop. A changed candidate set + invalidates the audit without invalidating successful scan tasks. + """ + rows = [row for key, row in sorted(self.state["candidates"].items()) + if self.state["reviews"].get(key, {}).get("relation") == "new" + and self._implemented_approval(row, self.state["reviews"][key])] + identity = _boundary_identity(self.seeds, self.state, repository=self.repository, pin=self.pin) + marker = {"identity": identity, "candidate_ids": [r["id"] for r in rows], "done": False} + previous = self.state.get("catalog_boundary_audit", {}) + if previous.get("identity") != identity or previous.get("candidate_ids") != marker["candidate_ids"]: + if previous: + self.state.setdefault("boundary_history", []).append({"audit": deepcopy(previous), + "reviews": deepcopy(self.state.get("boundary_reviews", {}))}) + self.state["boundary_reviews"] = {} + self.state["catalog_boundary_audit"] = marker + decisions = self.state.setdefault("boundary_reviews", {}) + pending = [r for r in rows if not _boundary_current(r, self.state["reviews"][r["id"]], decisions.get(r["id"], {}))] + # Collision families stay adjacent so the reviewer can classify aliases + # against one another, while the complete formal baseline remains visible. + pending.sort(key=lambda r: (re.sub(r"-[a-f0-9]{8}$", "", r["id"]), r["id"])) + batches = [pending[n:n + 12] for n in range(0, len(pending), 12)] + baseline = self._baseline() + self.save() + with ThreadPoolExecutor(max_workers=self.concurrency) as pool: + for start in range(0, len(batches), self.concurrency): + jobs = {pool.submit(self._review, batch, baseline, audit=_boundary_context(marker)): batch + for batch in batches[start:start + self.concurrency]} + stopped = False + for future in as_completed(jobs): + batch = jobs[future] + try: + results = future.result() + except BudgetExhausted: + stopped = True + continue + except (ModelUnavailable, ValueError) as exc: + results = {r["id"]: {"supported": "unsure", "relation": "unknown", "related_id": "", + "reason": str(exc), "repair_blocked": True} for r in batch} + for row in batch: + decisions[row["id"]] = {**results[row["id"]], "candidate_sha256": _hash(row), + "primary_review_sha256": _hash(self.state["reviews"][row["id"]])} + self.save() + if stopped: + return False + marker["done"] = True + self.save() + return True + + def catalog(self, repo_dir, *, require_boundary=False): from .feature_discovery_index import validate_evidence seeds = deepcopy(self.seeds) by_id = {r["id"]: r for r in seeds} outcomes = [] deferred = [] + boundary_valid = _boundary_ready(self.seeds, self.state, repository=self.repository, pin=self.pin, + production_paths=self.index.production) + proposed_names = Counter() + formal_names = {name for seed in seeds for name in _catalog_names(seed)} + for key, row in self.state["candidates"].items(): + checked = self.state.get("boundary_reviews", {}).get(key, {}) + if boundary_valid and checked.get("supported") == "yes" and checked.get("relation") == "new" \ + and _boundary_current(row, self.state["reviews"].get(key, {}), checked): + proposed_names.update(_catalog_names(row)) for key, row in sorted(self.state["candidates"].items()): - judgment = self.state["reviews"].get(key, {}) + primary = self.state["reviews"].get(key, {}) + judgment = primary + if require_boundary and primary.get("relation") == "new" and self._implemented_approval(row, primary): + checked = self.state.get("boundary_reviews", {}).get(key, {}) + judgment = checked if boundary_valid and _boundary_current(row, primary, checked) else { + "supported": "unsure", "relation": "unknown", "reason": "complete independent catalog boundary audit is missing"} refs = row["evidence"] source = sorted({r["path"] for r in refs if r["path"] in self.index.production}) docs = sorted({r["path"] for r in refs if r["path"] in self.index.docs}) @@ -509,6 +652,14 @@ def catalog(self, repo_dir): "status": "accepted" if accepted else "unknown", "evidence": refs, "origin_rounds": row.get("origin_rounds", []), "reason": judgment.get("reason", "not independently reviewed"), "generator_receipts": row.get("generator_receipts", []), "judge_receipt": judgment.get("judge_receipt"), "candidate_sha256": judgment.get("candidate_sha256")} + if judgment is not primary: + result["primary_judge_receipt"] = primary.get("judge_receipt") + result["boundary_review"] = deepcopy(judgment) + names = _catalog_names(row) + if accepted and require_boundary and relation == "new" and any( + name in formal_names or proposed_names[name] > 1 for name in names): + accepted = False + result.update(status="unknown", reason="unresolved duplicate catalog title or alias; independent classification needed") if accepted and relation == "new" and key in by_id: accepted = False result.update(status="unknown", reason="new capability collides with an existing stable ID") @@ -525,6 +676,19 @@ def catalog(self, repo_dir): outcomes.append(result) for row, judgment, source, docs, result in deferred: target = judgment.get("related_id") or row["id"] + if require_boundary: + visited = {row["id"]} + while target not in by_id and target not in visited: + visited.add(target) + target_row = self.state["candidates"].get(target) + primary = self.state["reviews"].get(target, {}) + target_check = self.state.get("boundary_reviews", {}).get(target, primary) + if target_row is None or not self._implemented_approval(target_row, primary) \ + or (primary.get("relation") == "new" and not _boundary_current(target_row, primary, target_check)) \ + or target_check.get("supported") != "yes" \ + or target_check.get("relation") not in ("implementation_supplement", "alias", "subcapability", "shared_component"): + break + target = target_check.get("related_id") or target if target not in by_id: result.update(status="unknown", reason="related feature is not accepted in the frozen catalog") continue @@ -570,15 +734,51 @@ def _chain(self): def _cache_reusable(self, previous): if previous.discovery.get("done") and self._validate_saved_archives(previous.discovery): raise InitError("completed discovery native archives are unavailable; restore the archive before reuse") - return bool(previous.discovery.get("done")) and not self.retry_unfinished + if not previous.discovery.get("done") or self.retry_unfinished: + return False + # Published historical catalogs remain immutable. Unpublished previews + # must acquire the additive audit before publication, reusing their scan. + return previous.status == "published" or self._boundary_record_ready(previous) + + def _boundary_record_ready(self, previous): + state = previous.discovery + if not state.get("catalog_boundary_audit", {}).get("done") or not state.get("index_sha256"): + return False + from .feature_discovery_index import load_discovery_index + # A checkpoint cannot declare an empty audit subset for implemented new + # candidates. Recover the verified inventory, independent of proposals. + for path in (Path(self.rt.state_dir) / "feature-discovery-index").glob("*.json"): + try: + index = load_discovery_index(path) + except ValueError: + continue + if index.sha256 == state["index_sha256"] and index.identity["pin"] == previous.pin: + return _boundary_ready(self._boundary_seeds(previous.kb_base_sha), state, + repository=self.lifecycle.full_name, pin=previous.pin, production_paths=index.production) + return False + + def _boundary_seeds(self, base_sha): + text = self.rt.knowledge.show(base_sha, self._coverage_policy_path()) + return (yaml.safe_load(text) or {}).get("features", []) if text else [] def _mode_identity(self): return False def _resume_input_problems(self, previous, digest): - return [] if previous.inputs_digest == digest else [ - "discovery inputs changed; restore the original configuration to resume this prepared publication, " - "or preserve its checkpoint and use a new state directory"] + if previous.inputs_digest != digest: + return ["discovery inputs changed; restore the original configuration to resume this prepared publication, " + "or preserve its checkpoint and use a new state directory"] + # Prepared publication resumes before the normal cache/archive gate. + # Audit an isolated record so a missing receipt blocks publication + # without rewriting the immutable prepared change or retained proofs. + audited = deepcopy(previous) + if self._validate_saved_archives(audited.discovery): + return ["prepared discovery native archives could not be replayed; restore the original archives " + "before resuming this exact publication"] + if not self._boundary_record_ready(audited): + return ["prepared discovery has no complete catalog boundary audit; preserve its publication and checkpoint " + "and audit a copy before creating a new publication"] + return [] def _init_identity(self): return repr(replace(self.lifecycle.init, budget_usd=0.0)) + ( @@ -682,6 +882,25 @@ def _verify_saved_approval(self, key, row, checked): except (KeyError, IndexError, TypeError, AttributeError, ValueError, OSError) as exc: raise ModelUnavailable(f"native approval binding differs: {exc}") from exc + def _verify_boundary_approval(self, key, row, checked, state): + self._verify_saved_approval(key, row, checked) + from ..trace_store import TraceStore + try: + store = TraceStore(Path(self.rt.state_dir) / "init" / "traces") + record = store.get(checked["judge_receipt"]["trace_id"]) + prompt = store.blob(record["inputs"]["prompt"]) + payload = json.loads(prompt.split("\n", 1)[1].rsplit("\n", 1)[0]) + marker = state["catalog_boundary_audit"] + if payload.get("catalog_boundary_audit") != _boundary_context(marker): + raise ValueError("native catalog boundary audit identity differs") + formal = [r for r in payload["baseline"] if r.get("catalog_status") == "existing_feature"] + if _hash(formal) != marker["identity"]["formal_catalog_sha256"]: + raise ValueError("native catalog boundary review omitted or changed formal entries") + if checked.get("primary_review_sha256") != _hash(state["reviews"][key]): + raise ValueError("catalog boundary approval differs from its primary review") + except (KeyError, IndexError, TypeError, AttributeError, ValueError, OSError) as exc: + raise ModelUnavailable(f"native catalog boundary binding differs: {exc}") from exc + def _validate_saved_archives(self, state): failures = [] for key, task in state.get("tasks", {}).items(): @@ -710,6 +929,18 @@ def _validate_saved_archives(self, state): "related_id": "", "reason": str(exc), "attempts": MAX_ATTEMPTS} state.get("repair_drafts", {}).pop(key, None) failures.append(key) + for key, checked in state.get("boundary_reviews", {}).items(): + if checked.get("supported") != "yes": + continue + try: + self._verify_archived_receipt(checked["judge_receipt"], self.rt.judge) + self._verify_boundary_approval(key, state["candidates"][key], checked, state) + except (ModelUnavailable, KeyError) as exc: + checked.update(supported="unsure", relation="unknown", related_id="", reason=str(exc), + repair_blocked=True) + state.get("catalog_boundary_audit", {})["done"] = False + # Keep the valid primary proof and scan; this candidate simply + # cannot promote without a usable supplemental approval. return failures def _build(self, tree): @@ -748,6 +979,7 @@ def _build_catalog(self, tree): if state.get("identity") and state["identity"] != identity: return self._blocked(["discovery index or catalog identity changed; create a new batch"]) state["identity"] = identity + state["index_sha256"] = index.sha256 state["run_config"] = self.discovery_run_config if self.retry_unfinished: for task in state.get("tasks", {}).values(): @@ -759,8 +991,9 @@ def _build_catalog(self, tree): checked.pop("repair_blocked") else: state["reviews"].pop(key) - elif checked.get("supported") != "yes": - checked["attempts"] = 1 # repair using previous rejection, not unchanged resampling + # Content rejections retain their cumulative attempt count. + # Only a blocked native channel resumes; exhausted content does + # not acquire three more repairs on every explicit retry. budget = _ConcurrentBudget(self.budget, Path(self.rt.state_dir) / "init" / self.lifecycle.repo / "discovery-budget.json", identity) if hasattr(self.rt.gateway, "configure_zcode_pacing") and self.rt.generator.provider == "zcode": from .depth_pacing import SharedZcodePacer @@ -796,7 +1029,14 @@ def save(): if not scanned or not engine.review(): state["done"] = False return self._blocked(["discovery incomplete; checkpoint saved; resume the same pinned batch"]) - features, outcomes = engine.catalog(self.repo_dir) + if not engine.audit_catalog_boundaries(): + state["done"] = False + return self._blocked(["catalog boundary audit incomplete; successful scan and primary reviews retained"]) + self._validate_saved_archives(state) + # Archive failures are terminal unknowns, never synthetic approvals. + if not state["catalog_boundary_audit"].get("done"): + engine.audit_catalog_boundaries() + features, outcomes = engine.catalog(self.repo_dir, require_boundary=True) from .init_coverage import most_specific seed_ids = {f["id"] for f in seeds} outcome_map = {r["id"]: r for r in outcomes} @@ -846,6 +1086,8 @@ def save(): "invalid_candidates": sum(len(t.get("invalid_candidates", [])) for t in state["tasks"].values())} report = {"schema_version": 1, "repo": self.lifecycle.repo, "pin": self.record.pin, "run_config": self.discovery_run_config, + "catalog_boundary_audit": {**state["catalog_boundary_audit"], + "counts": dict(Counter(r["supported"] for r in state.get("boundary_reviews", {}).values()))}, "complete": True, "done": True, "catalog_sha256": catalog_sha, "feature_ids": [f["id"] for f in features], "features": [{"id": f["id"], "owner": f["owner"], "title": f["title"]} for f in features], diff --git a/src/infermatrix_copilot/kb_service/init_stages.py b/src/infermatrix_copilot/kb_service/init_stages.py index f8e14e49..a7d8e9a2 100644 --- a/src/infermatrix_copilot/kb_service/init_stages.py +++ b/src/infermatrix_copilot/kb_service/init_stages.py @@ -167,8 +167,8 @@ def run_stage(rt: InitRuntime, lifecycle, stage: str, *, dry_run: bool, pin: str raise InitError("feature_ids must be a depth-only tuple of policy feature identifiers") if type(unlimited_subscription) is not bool: raise InitError("unlimited_subscription must be a boolean") - if unlimited_subscription and (stage not in ("feature-discovery", "knowledge-deepen") or budget_usd is not None): - raise InitError("--unlimited-subscription is for feature-discovery or knowledge-deepen only and conflicts with --budget-usd") + if unlimited_subscription and (stage not in ("feature-discovery", "modules", "knowledge", "knowledge-deepen") or budget_usd is not None): + raise InitError("--unlimited-subscription is for feature-discovery, modules, knowledge or knowledge-deepen only and conflicts with --budget-usd") if stage == "feature-discovery": from dataclasses import replace from .models import ModelRole @@ -224,7 +224,8 @@ def run_stage(rt: InitRuntime, lifecycle, stage: str, *, dry_run: bool, pin: str if any(role.fallback is not None for role in (rt.generator, rt.judge)): raise InitError("--unlimited-subscription requires pinned generator and judge without fallback") if rt.generator.provider != "zcode" or rt.generator.model.casefold() != "glm-5.3" \ - or rt.judge.provider != "codex": + or rt.judge.provider != "codex" \ + or rt.generator.model.casefold().split("-")[0] == rt.judge.model.casefold().split("-")[0]: raise InitError("--unlimited-subscription requires Zcode GLM-5.3 extraction and an independent Codex judge") for role in (rt.generator, rt.judge): try: @@ -532,7 +533,7 @@ def run(self) -> InitRecord: self.record.discovery["catalog_binding"] = self._discovery_binding() if chain.pin and pin != chain.pin: self.record.notes.append(f"pinned at {pin[:12]}, not at the earlier stages' {chain.pin[:12]}") - self.budget = Budget(init.budget_usd) + self.budget = Budget(None if rt.unlimited_subscription else init.budget_usd) self.base = {**main, **chain.knowledge} problems = chain.problems + self._restore_progress(previous) + self._precheck() if adapter_missing_problem: @@ -680,6 +681,8 @@ def _init_identity(self) -> str: identity = identity.replace(", feature_discovery_required=False", "") if self.rt.subscription_generator: identity += ":subscription-generator" + if self.rt.unlimited_subscription: + identity += ":unlimited-subscription" return identity def _mode_identity(self) -> bool: diff --git a/src/infermatrix_copilot/kb_service/init_support.py b/src/infermatrix_copilot/kb_service/init_support.py index ee6ade03..4213356b 100644 --- a/src/infermatrix_copilot/kb_service/init_support.py +++ b/src/infermatrix_copilot/kb_service/init_support.py @@ -356,6 +356,11 @@ def judge(rt: InitRuntime, budget: Budget, init, block, *, base: Mapping[str, st as verdict ``human`` with ``reasons["model"]`` set.""" from .gate import judge_block + if rt.unlimited_subscription: + if rt.judge.fallback is not None: + raise ModelUnavailable("unlimited subscription judgment must not configure a fallback", allow_fallback=False) + if not rt.gateway.subscription_billing(rt.judge): + raise ModelUnavailable("subscription judge needs an authenticated subscription backend", allow_fallback=False) with budget.reserve(init.judge_call_usd) as reservation: verdict = judge_block(block, base=dict(base), head=dict(head), evidence=evidence, gateway=rt.gateway, judge=rt.judge) diff --git a/test/test_kb_breadth_subscription.py b/test/test_kb_breadth_subscription.py new file mode 100644 index 00000000..17c96fe2 --- /dev/null +++ b/test/test_kb_breadth_subscription.py @@ -0,0 +1,162 @@ +"""Explicit breadth subscription runs preserve legacy caps and fail closed.""" + +import json +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service import cli, runner +from infermatrix_copilot.kb_service.init_budget import Budget +from infermatrix_copilot.kb_service.init_modules import _Modules +from infermatrix_copilot.kb_service.init_stages import run_stage +from infermatrix_copilot.kb_service.init_support import InitError, judge +from infermatrix_copilot.kb_service.models import ModelGateway, ModelRole, ModelUnavailable +from infermatrix_copilot.knowledge_service.l1 import Block +from infermatrix_copilot.llm import Block as ReplyBlock, Reply +from test_kb_init_knowledge import KnowledgeGateway, _chain +from test_kb_init_modules import CardGateway, _modules_lifecycle, _skeleton, _world_with_tools +from test_kb_init_skeleton import _runtime, world # noqa: F401 + + +GENERATOR = ModelRole("generator", "zcode", "GLM-5.3") + + +class SubscriptionTransport: + def __init__(self, *, subscription=True): + self.subscription_billing = subscription + self.calls = [] + + def complete(self, **kwargs): + self.calls.append(kwargs) + data = {"dimensions": {k: "yes" for k in ("faithful", "does_not_weaken", "non_contradictory")}} + return Reply(blocks=[ReplyBlock(type="text", text=json.dumps(data))], usage={"input_tokens": 12}) + + +@pytest.mark.parametrize("command,stage", [ + (["init", "toy", "--stage", "modules"], "modules"), + (["init", "toy", "--stage", "knowledge"], "knowledge"), + (["widen", "toy"], "knowledge"), +]) +def test_breadth_cli_forwards_explicit_unlimited_subscription(tmp_path, monkeypatch, command, stage): + seen = [] + + def run(settings, playbook, repo, *, state_dir, params): + seen.append(params) + return SimpleNamespace(status="done"), tmp_path + + monkeypatch.setattr(runner, "run_playbook", run) + assert cli.main(command + ["--unlimited-subscription", "--dry-run"]) == 0 + assert seen[0]["unlimited_subscription"] == "true" and seen[0]["stage"] == stage + assert "budget_usd" not in seen[0] + + +@pytest.mark.parametrize("stage", ["modules", "knowledge"]) +@pytest.mark.parametrize("unavailable_role", ["generator", "judge"]) +def test_breadth_requires_both_authenticated_subscriptions_before_dispatch(world, stage, unavailable_role): + transports = {"zcode": SubscriptionTransport(subscription=unavailable_role != "generator"), + "codex": SubscriptionTransport(subscription=unavailable_role != "judge")} + rt = _runtime(world, ModelGateway(None, transport_factory=transports.__getitem__), generator=GENERATOR) + with pytest.raises(InitError, match=unavailable_role + ":.*authenticated subscription"): + run_stage(rt, _modules_lifecycle(), stage, dry_run=True, from_existing=True, unlimited_subscription=True) + assert not any(transport.calls for transport in transports.values()) + + +@pytest.mark.parametrize("stage", ["modules", "knowledge"]) +def test_breadth_unlimited_budget_conflict_is_rejected_before_auth(world, stage): + rt = _runtime(world, ModelGateway(None, transport_factory=lambda _: pytest.fail("invalid options authenticated"))) + with pytest.raises(InitError, match="conflicts with --budget-usd"): + run_stage(rt, _modules_lifecycle(), stage, dry_run=True, unlimited_subscription=True, budget_usd=100) + + +@pytest.mark.parametrize("fallback_role", ["generator", "judge"]) +def test_breadth_unlimited_refuses_fallback_before_auth(world, fallback_role): + fallback = ModelRole(fallback_role, "other", "paid-model") + generator = ModelRole("generator", "zcode", "GLM-5.3", fallback=fallback if fallback_role == "generator" else None) + judge_role = ModelRole("judge", "codex", "gpt-6.1-sol", fallback=fallback if fallback_role == "judge" else None) + rt = _runtime(world, ModelGateway(None, transport_factory=lambda _: pytest.fail("fallback config authenticated")), + generator=generator, judge=judge_role) + with pytest.raises(InitError, match="without fallback"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + + +def test_unlimited_breadth_requires_an_independent_model_even_across_providers(world): + rt = _runtime(world, ModelGateway(None, transport_factory=lambda _: pytest.fail("same model authenticated")), + generator=GENERATOR, judge=ModelRole("judge", "codex", "GLM-5.3")) + with pytest.raises(InitError, match="independent Codex"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + + +class SubscriptionKnowledgeGateway(KnowledgeGateway): + def subscription_billing(self, role): + return True + + +def test_breadth_unlimited_exceeds_sixty_dollar_accounting_and_legacy_cap_remains(world): + gateway = SubscriptionKnowledgeGateway() + _chain(world, gateway) + lifecycle = _modules_lifecycle(budget_usd=60, judge_call_usd=20) + rt = _runtime(world, gateway, generator=GENERATOR) + capped = run_stage(rt, lifecycle, "knowledge", dry_run=True, subscription_generator=True) + assert capped.status == "dry_run", capped.problems + assert capped.spent_usd == 60 and len(capped.verdicts) == 3 + assert any("budget exhausted" in note for note in capped.notes) + calls_before = len(gateway.calls) + uncapped = run_stage(rt, lifecycle, "knowledge", dry_run=True, subscription_generator=True, + unlimited_subscription=True) + assert uncapped.inputs_digest != capped.inputs_digest + assert len(gateway.calls) > calls_before # a capped result cannot be reused as unlimited + assert uncapped.status == "dry_run", uncapped.problems + assert uncapped.spent_usd == 80 and len(uncapped.verdicts) == 4 + assert not any("budget exhausted" in note for note in uncapped.notes) + assert any("observability only" in note and "remain unknown" in note for note in uncapped.notes) + + +def test_modules_selects_uncapped_budget_only_when_explicit(world, monkeypatch): + _world_with_tools(world) + _skeleton(world) + gateway = CardGateway() + gateway.subscription_billing = lambda role: True + original = _Modules._build + limits = [] + + def build(stage, tree): + limits.append(stage.budget.limit_usd) + return original(stage, tree) + + monkeypatch.setattr(_Modules, "_build", build) + rt = _runtime(world, gateway, generator=GENERATOR) + legacy = run_stage(rt, _modules_lifecycle(budget_usd=60), "modules", dry_run=True, subscription_generator=True) + unlimited = run_stage(rt, _modules_lifecycle(budget_usd=60), "modules", dry_run=True, + subscription_generator=True, unlimited_subscription=True) + assert legacy.status == unlimited.status == "dry_run" + assert limits == [60, None] + assert legacy.inputs_digest != unlimited.inputs_digest + + +def test_unlimited_breadth_identity_preserves_legacy_default(world): + rt = _runtime(world) + lifecycle = _modules_lifecycle() + stage = _Modules(rt, lifecycle, dry_run=True, pin=None) + expected = repr(lifecycle.init).replace(f", pr_history_count={lifecycle.init.pr_history_count}", "") \ + .replace(", feature_discovery_required=False", "") + assert stage._init_identity() == expected + rt.unlimited_subscription = True + assert stage._init_identity() == expected + ":unlimited-subscription" + rt.subscription_generator = True + assert stage._init_identity() == expected + ":subscription-generator:unlimited-subscription" + + +def test_unlimited_judge_rechecks_and_binds_subscription_transport_before_every_call(world): + subscription, paid = SubscriptionTransport(), SubscriptionTransport(subscription=False) + transports = iter([subscription, paid]) + rt = _runtime(world, ModelGateway(None, transport_factory=lambda _: next(transports)), + unlimited_subscription=True) + budget = Budget(None) + block = Block("prose", "repos/toy/components/core/knowledge.md", "", "prose", "digest") + verdict = judge(rt, budget, _modules_lifecycle().init, block, base={}, head={}, evidence=[]) + assert verdict.verdict == "pass" and len(subscription.calls) == 1 + assert budget.spent_usd == _modules_lifecycle().init.judge_call_usd + with pytest.raises(ModelUnavailable, match="subscription judge.*authenticated subscription"): + judge(rt, budget, _modules_lifecycle().init, block, base={}, head={}, evidence=[]) + assert paid.calls == [] + assert budget.spent_usd == _modules_lifecycle().init.judge_call_usd diff --git a/test/test_kb_depth_campaign_discovery_handoff.py b/test/test_kb_depth_campaign_discovery_handoff.py new file mode 100644 index 00000000..8c377428 --- /dev/null +++ b/test/test_kb_depth_campaign_discovery_handoff.py @@ -0,0 +1,199 @@ +"""Real prerequisite handoff uses original bytes and the native merged gate.""" + +import argparse +import hashlib +import json +from dataclasses import asdict +from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml + +from infermatrix_copilot.kb_service.init_support import InitPublisher, InitRecord, InitRuntime +from infermatrix_copilot.kb_service.sources import KnowledgeRepo +from test_kb_depth_campaign import campaign_args, runner, _git, _offline_processes # noqa: F401 + + +def _frozen_discovery(args, *, status="empty", adapter="demo"): + """An original completed record plus its real committed catalog/report.""" + if adapter != "demo": + (args.root / "adapters/demo").rename(args.root / "adapters" / adapter) + policy_path = args.root / "adapters" / adapter / "knowledge-coverage.yaml" + policy = policy_path.read_text() + features = yaml.safe_load(policy)["features"] + report_path = f"eval/feature-discovery/{args.repo}-{args.pin[:12]}.json" + report = {"schema_version": 1, "repo": args.repo, "pin": args.pin, "complete": True, "done": True, + "feature_ids": [row["id"] for row in features], "owner_requests": [], + "features": [{key: row[key] for key in ("id", "title", "owner")} for row in features], + "catalog_sha256": hashlib.sha256(policy.encode()).hexdigest()} + target = args.root / report_path + target.parent.mkdir(parents=True) + report_text = json.dumps(report, indent=2) + "\n" + target.write_text(report_text) + routes = args.root / "knowledge/repos/demo/_routes.yaml" + routes.parent.mkdir(parents=True) + routes.write_text(yaml.safe_dump({"schema_version": 1, "owners": [ + {"owner": "core", "path": "repos/demo/components/core/_index.md", "prefixes": ["src/"]}]})) + old_baseline = args.baseline + _git(args.root, "add", ".") + _git(args.root, "commit", "--quiet", "-m", "Freeze discovered feature catalog") + args.baseline = _git(args.root, "rev-parse", "HEAD") + record = InitRecord(stage="feature-discovery", repo=args.repo, pin=args.pin, kb_base_sha=old_baseline, + status=status, dry_run=status != "published", pr={"number": 42} if status == "published" else {}, + notes=["native traces remain /outside/git/archive"], discovery={ + "done": True, "report_path": report_path, "catalog_sha256": report["catalog_sha256"], + "report_sha256": hashlib.sha256(report_text.encode()).hexdigest()}) + args.discovery_record = args.state.parent / "original-feature-discovery.json" + raw = json.dumps(asdict(record), ensure_ascii=False, indent=3).encode() + b"\n" + args.discovery_record.write_bytes(raw) + return raw + + +def test_campaign_copies_exact_prerequisite_and_binds_every_worker(runner, campaign_args, monkeypatch): + args = campaign_args + raw = _frozen_discovery(args) + processes, _ = _offline_processes(runner, monkeypatch) + assert runner.campaign(args, argparse.ArgumentParser()) == 0 + summary = json.loads((args.state / "campaign.json").read_bytes()) + binding = summary["discovery"] + assert binding["record_sha256"] == hashlib.sha256(raw).hexdigest() + assert binding["pin"] == args.pin and summary["denominator"] == 553 + assert len(processes) == 2 + for number in range(args.workers): + copied = InitRecord.path(args.state / f"worker-{number}", args.repo, "feature-discovery") + assert copied.read_bytes() == raw + assert b"/outside/git/archive" in copied.read_bytes() + + +def test_aliased_adapter_is_resolved_from_baseline(runner, campaign_args): + _frozen_discovery(campaign_args, adapter="demo_package") + assert runner.baseline_inputs(campaign_args)[2] == "adapters/demo_package/knowledge-coverage.yaml" + assert runner.discovery_handoff(campaign_args)["pin"] == campaign_args.pin + + +def test_catalog_report_requires_explicit_handoff_before_dispatch(runner, campaign_args, monkeypatch): + _frozen_discovery(campaign_args) + campaign_args.discovery_record = None + processes, _ = _offline_processes(runner, monkeypatch) + with pytest.raises(SystemExit) as error: + runner.campaign(campaign_args, argparse.ArgumentParser()) + assert error.value.code == 2 and not processes + assert not (campaign_args.state / "campaign.json").exists() + + +def test_missing_original_record_blocks_before_dispatch(runner, campaign_args, monkeypatch): + _frozen_discovery(campaign_args) + campaign_args.discovery_record.unlink() + processes, _ = _offline_processes(runner, monkeypatch) + with pytest.raises(SystemExit): + runner.campaign(campaign_args, argparse.ArgumentParser()) + assert not processes + + +def test_committed_policy_change_cannot_reuse_record(runner, campaign_args): + _frozen_discovery(campaign_args) + policy = campaign_args.root / "adapters/demo/knowledge-coverage.yaml" + policy.write_text(policy.read_text() + "# changed after discovery\n") + _git(campaign_args.root, "add", ".") + _git(campaign_args.root, "commit", "--quiet", "-m", "Change frozen catalog") + campaign_args.baseline = _git(campaign_args.root, "rev-parse", "HEAD") + with pytest.raises(ValueError, match="catalog hash differs"): + runner.discovery_handoff(campaign_args) + + +@pytest.mark.parametrize("field,value", [ + ("stage", "modules"), ("repo", "other"), ("pin", "b" * 40), + ("status", "dry_run"), ("discovery.done", False), + ("discovery.catalog_sha256", "b" * 64), ("discovery.report_sha256", "b" * 64), +]) +def test_invalid_record_is_rejected_before_model_runtime(runner, campaign_args, monkeypatch, field, value): + _frozen_discovery(campaign_args) + path = campaign_args.discovery_record + data = json.loads(path.read_bytes()) + if "." in field: + parent, key = field.split(".") + data[parent][key] = value + else: + data[field] = value + path.write_text(json.dumps(data)) + monkeypatch.setattr(InitRuntime, "from_env", lambda *_a, **_k: pytest.fail("invalid handoff started models")) + with pytest.raises(ValueError, match="handoff"): + runner.discovery_handoff(campaign_args) + + +@pytest.mark.parametrize("merged", [False, True]) +def test_published_record_requires_actual_merged_pr(runner, campaign_args, monkeypatch, merged): + _frozen_discovery(campaign_args, status="published") + checked = [] + monkeypatch.setattr(InitPublisher, "pr_state", lambda self, number: checked.append(number) or ("MERGED" if merged else "OPEN")) + if merged: + assert runner.discovery_handoff(campaign_args)["pin"] == campaign_args.pin + else: + with pytest.raises(ValueError, match="merge the feature-discovery PR"): + runner.discovery_handoff(campaign_args) + assert checked == [42] + + +def test_changed_original_record_cannot_resume_campaign(runner, campaign_args, monkeypatch): + raw = _frozen_discovery(campaign_args) + _offline_processes(runner, monkeypatch) + assert runner.campaign(campaign_args, argparse.ArgumentParser()) == 0 + data = json.loads(raw) + data["notes"].append("changed after dispatch") + campaign_args.discovery_record.write_text(json.dumps(data)) + monkeypatch.setattr(runner.subprocess, "Popen", lambda *_a, **_k: pytest.fail("changed record dispatched")) + with pytest.raises(SystemExit) as error: + runner.campaign(campaign_args, argparse.ArgumentParser()) + assert error.value.code == 2 + assert InitRecord.path(campaign_args.state / "worker-0", campaign_args.repo, "feature-discovery").read_bytes() == raw + + +def test_worker_record_cannot_be_overwritten(runner, campaign_args): + raw = _frozen_discovery(campaign_args) + binding = runner.discovery_handoff(campaign_args) + state = campaign_args.state / "worker-0" + runner.install_discovery_handoff(campaign_args, state, binding) + path = InitRecord.path(state, campaign_args.repo, "feature-discovery") + path.write_bytes(raw + b" ") + with pytest.raises(ValueError, match="worker discovery prerequisite differs"): + runner.install_discovery_handoff(campaign_args, state, binding) + assert path.read_bytes() == raw + b" " + + +@pytest.mark.parametrize("mismatch", ["hash", "metadata"]) +def test_worker_rechecks_hash_and_parent_binding_before_models(runner, campaign_args, monkeypatch, mismatch): + raw = _frozen_discovery(campaign_args) + campaign_args.discovery_record_sha256 = "b" * 64 if mismatch == "hash" else hashlib.sha256(raw).hexdigest() + (campaign_args.state / "campaign.json").write_text(json.dumps({"discovery": {}})) + monkeypatch.setattr(KnowledgeRepo, "fetch", lambda _: campaign_args.baseline) + monkeypatch.setattr(InitRuntime, "from_env", lambda *_a, **_k: pytest.fail("unbound worker started runtime")) + with pytest.raises((ValueError, RuntimeError), match="binding|identity"): + runner.run_worker(campaign_args, campaign_args.state / "worker-0") + + +def test_genuine_worker_dispatch_reuses_original_record(runner, campaign_args, monkeypatch): + from infermatrix_copilot.kb_service import init_stages + + args = campaign_args + raw = _frozen_discovery(args) + binding = runner.discovery_handoff(args) + args.discovery_record_sha256 = binding["record_sha256"] + (args.state / "campaign.json").write_text(json.dumps({"discovery": binding})) + monkeypatch.setattr(KnowledgeRepo, "fetch", lambda _: args.baseline) + state = args.state / "worker-0" + runtime = SimpleNamespace(registry={args.repo: object()}, gateway=SimpleNamespace(configure_zcode_pacing=lambda _: None)) + + def get_runtime(*_a, **_k): + assert InitRecord.path(state, args.repo, "feature-discovery").read_bytes() == raw + return runtime + + def run_stage(*_a, **kwargs): + assert kwargs["from_existing"] and kwargs["feature_ids"] == ("f0",) + assert InitRecord.load(state, args.repo, "feature-discovery").status == "empty" + return InitRecord(stage="knowledge-deepen", repo=args.repo, pin=args.pin, kb_base_sha=args.baseline, status="dry_run") + + monkeypatch.setattr(InitRuntime, "from_env", get_runtime) + monkeypatch.setattr(init_stages, "run_stage", run_stage) + assert runner.run_worker(args, state) == 0 + assert InitRecord.path(state, args.repo, "feature-discovery").read_bytes() == raw diff --git a/test/test_kb_feature_discovery_boundary.py b/test/test_kb_feature_discovery_boundary.py new file mode 100644 index 00000000..b48ae087 --- /dev/null +++ b/test/test_kb_feature_discovery_boundary.py @@ -0,0 +1,314 @@ +"""Catalog admission requires an independent audit against all formal IDs.""" +from copy import deepcopy +import json + +import pytest + +from infermatrix_copilot.kb_service.init_budget import BudgetExhausted +from infermatrix_copilot.kb_service.init_feature_discovery import ( + DiscoveryEngine, SYSTEM_REVIEW, _hash, _boundary_ready, +) +from infermatrix_copilot.kb_service.models import ModelUnavailable +from infermatrix_copilot.llm import Block, Reply + +from test_kb_feature_discovery import PIN, candidate, index, reply +from test_kb_feature_discovery_stage import NativeScript, _setup, _run, _report, _policy +from test_kb_init_skeleton import world # noqa: F401 + + +def approved(row, relation='new', related_id=''): + return {'supported': 'yes', 'relation': relation, 'related_id': related_id, + 'reason': 'Pinned implementation supports the independently classified contract.', + 'candidate_sha256': _hash(row), 'judge_receipt': {'trace_id': 'primary'}} + + +def engine(tmp_path, rows, call, seeds=(), concurrency=13): + idx = index(tmp_path) + for row in rows: + row['generator_receipts'] = [{'trace_id': 'generator'}] + state = {'candidates': {r['id']: r for r in rows}, + 'reviews': {r['id']: approved(r) for r in rows}, + 'tasks': {'source:preserved': {'status': 'complete', 'receipt': 'preserved'}}, + 'identity': 'original-pinned-scan'} + return DiscoveryEngine(idx, seeds=list(seeds), owners=['runtime'], state=state, call=call, + save=lambda: None, concurrency=concurrency, repository='o/library', pin=PIN) + + +def decisions(prompt, classify=None): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + results = {} + for row in payload['candidates']: + relation, target = classify(row) if classify else ('new', '') + results[row['id']] = {'supported': 'yes', 'relation': relation, 'related_id': target, + 'reason': 'Independent boundary classification with the complete formal catalog.'} + return payload, {'decisions': results} + + +def test_all_formal_features_remain_visible_despite_large_proposal_competition(tmp_path): + seeds = [{'id': f'formal-{n}', 'title': 'Formal ' + str(n) + ' ' + 'behavior ' * 80, + 'owner': 'runtime', 'source_globs': ['src/session.py']} for n in range(79)] + captures = [] + def call(role, system, prompt, validate): + assert role == 'judge' and system == SYSTEM_REVIEW + payload, data = decisions(prompt, lambda r: ('subcapability', 'formal-78')) + captures.append(payload) + assert [r['id'] for r in payload['baseline'] if r['catalog_status'] == 'existing_feature'] == [s['id'] for s in seeds] + return reply(role, data) + e = engine(tmp_path, [candidate(key='microfeature')], call, seeds) + original = deepcopy(e.state['reviews']) + assert e.audit_catalog_boundaries() + features, outcomes = e.catalog('repos/library', require_boundary=True) + assert len(features) == 79 + assert outcomes[0]['status'] == 'accepted' and outcomes[0]['relation'] == 'subcapability' + assert outcomes[0]['related_id'] == 'formal-78' + assert e.state['reviews'] == original + assert outcomes[0]['primary_judge_receipt'] == original['microfeature']['judge_receipt'] + assert len(captures) == 1 + + +def test_missing_audit_cannot_promote_a_primary_yes(tmp_path): + e = engine(tmp_path, [candidate()], None) + assert not e.catalog('repos/library', require_boundary=True)[0] + assert e.catalog('repos/library', require_boundary=True)[1][0]['status'] == 'unknown' + + +def test_test_and_document_only_candidates_are_not_boundary_repaired(tmp_path): + calls = [] + e = engine(tmp_path, [candidate(path='docs/session.md')], lambda *args: calls.append(args)) + assert e.audit_catalog_boundaries() + assert e.state['catalog_boundary_audit']['candidate_ids'] == [] + assert not calls and not e.catalog('repos/library', require_boundary=True)[0] + + +@pytest.mark.parametrize('failure', ['rejected', 'unavailable']) +def test_boundary_failure_is_unknown_without_generator_repairs_or_resampling(tmp_path, failure): + calls = [] + def call(role, system, prompt, validate): + calls.append(role) + if failure == 'unavailable': + raise ModelUnavailable('native channel unavailable') + _, data = decisions(prompt) + for row in data['decisions'].values(): + row.update(supported='no', relation='unknown', reason='Boundary overlaps an existing capability.') + return reply(role, data) + e = engine(tmp_path, [candidate()], call) + original = deepcopy(e.state['reviews']) + assert e.audit_catalog_boundaries() and e.review() and e.audit_catalog_boundaries() + assert calls == ['judge'] + assert e.state['reviews'] == original + assert not e.catalog('repos/library', require_boundary=True)[0] + + +def test_budget_resume_reuses_scan_primary_and_successful_boundary_batches(tmp_path): + calls = [] + def call(role, system, prompt, validate): + payload, data = decisions(prompt) + calls.append([r['id'] for r in payload['candidates']]) + if len(calls) == 2: + raise BudgetExhausted('boundary interrupted') + return reply(role, data) + e = engine(tmp_path, [candidate(key=f'feature-{n:02d}', title=f'Capability {n}') for n in range(13)], call, concurrency=1) + original_tasks, original_reviews = deepcopy(e.state['tasks']), deepcopy(e.state['reviews']) + assert not e.audit_catalog_boundaries() + assert len(e.state['boundary_reviews']) == 12 + assert e.audit_catalog_boundaries() + assert len(calls) == 3 and calls[1] == calls[2] + assert len(e.catalog('repos/library', require_boundary=True)[0]) == 13 + assert e.state['tasks'] == original_tasks and e.state['reviews'] == original_reviews + + +def test_changed_candidate_set_invalidates_only_boundary_audit(tmp_path): + calls = [] + def call(role, system, prompt, validate): + calls.append(role); _, data = decisions(prompt) + return reply(role, data) + e = engine(tmp_path, [candidate()], call) + assert e.audit_catalog_boundaries() + previous = deepcopy(e.state['catalog_boundary_audit']) + e.state['candidates']['session']['description'] += ' With a newly reviewed explicit close contract.' + e.state['reviews']['session'] = approved(e.state['candidates']['session']) + assert not _boundary_ready(e.seeds, e.state, repository=e.repository, pin=e.pin) + assert e.audit_catalog_boundaries() + assert calls == ['judge', 'judge'] + assert e.state['boundary_history'][0]['audit'] == previous + assert e.state['tasks']['source:preserved']['status'] == 'complete' + + +def test_unresolved_same_title_new_candidates_stay_unknown(tmp_path): + def call(role, system, prompt, validate): + _, data = decisions(prompt); return reply(role, data) + e = engine(tmp_path, [candidate(key='first'), candidate(key='second')], call) + assert e.audit_catalog_boundaries() + features, outcomes = e.catalog('repos/library', require_boundary=True) + assert not features + assert all(r['status'] == 'unknown' and 'duplicate' in r['reason'] for r in outcomes) + + +def test_aliases_can_resolve_to_an_independently_accepted_new_parent(tmp_path): + def call(role, system, prompt, validate): + mapping = {'parent': ('new', ''), 'alias-one': ('alias', 'parent'), 'alias-two': ('alias', 'alias-one')} + _, data = decisions(prompt, lambda r: mapping[r['id']]); return reply(role, data) + e = engine(tmp_path, [candidate(key=k, title=k) for k in ('parent', 'alias-one', 'alias-two')], call) + assert e.audit_catalog_boundaries() + features, outcomes = e.catalog('repos/library', require_boundary=True) + assert [f['id'] for f in features] == ['parent'] + assert all(r['status'] == 'accepted' for r in outcomes) + + +def test_alias_cycle_cannot_create_a_formal_feature(tmp_path): + def call(role, system, prompt, validate): + _, data = decisions(prompt, lambda r: ('alias', 'second' if r['id'] == 'first' else 'first')) + return reply(role, data) + e = engine(tmp_path, [candidate(key='first', title='First'), candidate(key='second', title='Second')], call) + assert e.audit_catalog_boundaries() + features, outcomes = e.catalog('repos/library', require_boundary=True) + assert not features and all(r['status'] == 'unknown' for r in outcomes) + + +class BoundaryNative(NativeScript): + def __init__(self, supported=True): + super().__init__(); self.boundary_supported = supported + + def complete(self, **kwargs): + response = super().complete(**kwargs) + payload = self.calls[-1]['payload'] + if 'catalog_boundary_audit' not in payload: + return response + data = {'decisions': {r['id']: {'supported': 'yes' if self.boundary_supported else 'no', + 'relation': 'new' if self.boundary_supported else 'unknown', 'related_id': '', + 'reason': 'Native supplemental boundary decision.'} for r in payload['candidates']}} + return Reply(blocks=[Block(type='text', text=json.dumps(data))], stop_reason=response.stop_reason, + usage=response.usage, model=response.model) + + +def test_legacy_unpublished_preview_reuses_all_scan_and_primary_records(world): + from infermatrix_copilot.kb_service.init_support import InitRecord + rt, lifecycle, script, _ = _setup(world, script=BoundaryNative()) + previous = _run(rt, lifecycle) + assert previous.status == 'dry_run', previous.problems + original_tasks = deepcopy(previous.discovery['tasks']) + original_reviews = deepcopy(previous.discovery['reviews']) + generator_calls = sum(c['role'] == 'generator' for c in script.calls) + judge_calls = sum(c['role'] == 'judge' for c in script.calls) + previous.discovery.pop('catalog_boundary_audit') + previous.discovery.pop('boundary_reviews') + previous.save(rt.state_dir) + current = _run(rt, lifecycle) + assert current.status == 'dry_run', current.problems + assert current.discovery['tasks'] == original_tasks and current.discovery['reviews'] == original_reviews + assert sum(c['role'] == 'generator' for c in script.calls) == generator_calls + assert sum(c['role'] == 'judge' for c in script.calls) == judge_calls + 1 + assert _report(current)[1]['catalog_boundary_audit']['done'] + + +def test_forged_empty_audit_subset_is_rejected_by_cached_production_inventory(world): + from infermatrix_copilot.kb_service.init_feature_discovery import _FeatureDiscovery + rt, lifecycle, _, _ = _setup(world, script=BoundaryNative()) + record = _run(rt, lifecycle) + assert record.status == 'dry_run', record.problems + record.discovery['catalog_boundary_audit']['candidate_ids'] = [] + record.discovery['boundary_reviews'] = {} + stage = _FeatureDiscovery(rt, lifecycle, dry_run=True, pin=None) + assert not stage._boundary_record_ready(record) + + +def test_explicit_retry_does_not_restart_exhausted_content_repairs(world): + from infermatrix_copilot.kb_service.init_stages import run_stage + rt, lifecycle, script, _ = _setup(world, policy=_policy(), script=NativeScript(supported=False)) + first = _run(rt, lifecycle) + assert first.status == 'dry_run', first.problems + assert first.discovery['reviews']['engine-step']['attempts'] == 4 + original = deepcopy(first.discovery['reviews']) + calls = len(script.calls) + second = run_stage(rt, lifecycle, 'feature-discovery', dry_run=True, from_existing=True, + unlimited_subscription=True, retry_unfinished=True) + assert second.status == 'dry_run', second.problems + assert second.discovery['reviews'] == original + assert len(script.calls) == calls + + +def test_prepared_catalog_without_boundary_proof_cannot_publish(world): + from infermatrix_copilot.kb_service.init_stages import run_stage + from test_kb_init_skeleton import FakeGh + rt, lifecycle, script, _ = _setup(world, script=BoundaryNative()) + rt.environ.update(ALLOW_PUSH='1', ALLOW_POST='1', KB_INIT_GIT_AUTHOR='t ') + gh = FakeGh(fail_create=1); rt.gh_run = gh + first = run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + assert first.status == 'blocked' and first.pr.get('prepared'), first.problems + first.discovery.pop('catalog_boundary_audit') + first.save(rt.state_dir) + calls, pushes = len(script.calls), list(gh.pushed) + second = run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + assert second.status == 'blocked' and any('boundary audit' in p for p in second.problems) + assert len(script.calls) == calls and gh.pushed == pushes + + +@pytest.mark.parametrize('review', ['primary', 'boundary']) +@pytest.mark.parametrize('damage', ['missing', 'tampered']) +def test_prepared_catalog_replays_native_archives_without_rewriting_proofs(world, review, damage): + from pathlib import Path + from infermatrix_copilot.kb_service.init_stages import run_stage + from test_kb_init_skeleton import FakeGh + + rt, lifecycle, script, store = _setup(world, script=BoundaryNative()) + rt.environ.update(ALLOW_PUSH='1', ALLOW_POST='1', KB_INIT_GIT_AUTHOR='t ') + gh = FakeGh(fail_create=1); rt.gh_run = gh + first = run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + assert first.status == 'blocked' and first.pr.get('prepared'), first.problems + original_discovery = deepcopy(first.discovery) + prepared = Path(first.pr['prepared']) + prepared_bytes = prepared.read_bytes() + checked = next(iter(first.discovery['reviews' if review == 'primary' else 'boundary_reviews'].values())) + native = store.get(checked['judge_receipt']['trace_id']) + reply_hash = native['outputs']['reply'].removeprefix('sha256:') + blob = store.root / 'blobs' / reply_hash[:2] / f'{reply_hash}.gz' + original_blob = blob.read_bytes() + if damage == 'missing': + blob.unlink() + else: + blob.write_bytes(b'tampered native archive') + calls, pushes = len(script.calls), list(gh.pushed) + rejected = run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + assert rejected.status == 'blocked' + assert any('native archives' in p or 'boundary audit' in p for p in rejected.problems) + assert rejected.discovery == original_discovery + assert prepared.read_bytes() == prepared_bytes + assert len(script.calls) == calls and gh.pushed == pushes + # Restoring the genuine archive resumes the same prepared publication, + # preserving both review records instead of synthesizing replacements. + blob.write_bytes(original_blob) + resumed = run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + assert resumed.status == 'published', resumed.problems + assert resumed.discovery == original_discovery + assert prepared.read_bytes() == prepared_bytes + assert len(script.calls) == calls + + +@pytest.mark.parametrize('tamper', ['verdict', 'formal_catalog', 'candidate']) +def test_supplemental_native_approval_binds_verdict_full_catalog_and_candidate(world, tamper): + from infermatrix_copilot.kb_service.init_feature_discovery import _FeatureDiscovery + from infermatrix_copilot.kb_service.models import ModelRole + from dataclasses import replace + rt, lifecycle, _, _ = _setup(world, policy=_policy(), script=BoundaryNative(supported=tamper != 'verdict')) + record = _run(rt, lifecycle) + assert record.status == 'dry_run', record.problems + state = record.discovery; checked = state['boundary_reviews']['engine-step'] + row = state['candidates']['engine-step'] + if tamper == 'verdict': + checked.update(supported='yes', relation='new') + elif tamper == 'formal_catalog': + state['catalog_boundary_audit']['identity']['formal_catalog_sha256'] = '0' * 64 + else: + row['description'] = 'Claim that the supplemental native judge never read.' + checked['candidate_sha256'] = _hash(row) + stage = _FeatureDiscovery(replace(rt, generator=ModelRole.parse('generator', row['generator_receipts'][0]['requested']), + judge=ModelRole.parse('judge', checked['judge_receipt']['requested'])), lifecycle, dry_run=True, pin=None) + stage._validate_saved_archives(state) + assert state['boundary_reviews']['engine-step']['supported'] == 'unsure' + assert state['catalog_boundary_audit']['done'] is False + assert 'binding' in state['boundary_reviews']['engine-step']['reason'] diff --git a/test/test_kb_unlimited_subscription.py b/test/test_kb_unlimited_subscription.py index f1d852bf..ef53cb07 100644 --- a/test/test_kb_unlimited_subscription.py +++ b/test/test_kb_unlimited_subscription.py @@ -64,7 +64,7 @@ def run(settings, playbook, repo, *, state_dir, params): @pytest.mark.parametrize("command", [ - ["init", "toy", "--stage", "knowledge"], + ["init", "toy", "--stage", "skeleton"], ["init", "toy", "--stage", "deepen"], ["init", "toy", "--suggest-seeds"], ["deepen", "toy", "--budget-usd", "100"], @@ -134,7 +134,7 @@ def test_unlimited_stage_rejects_a_different_model_protocol_before_auth_or_calls @pytest.mark.parametrize("stage, options", [ - ("knowledge", {"unlimited_subscription": True}), + ("skeleton", {"unlimited_subscription": True}), ("knowledge-deepen", {"unlimited_subscription": True, "budget_usd": 100.0}), ("knowledge-deepen", {"unlimited_subscription": "true"}), ])