diff --git a/doc/architecture/SPEC/engine/steps/knowledge.md b/doc/architecture/SPEC/engine/steps/knowledge.md index 01491ea2..07b2ad2b 100644 --- a/doc/architecture/SPEC/engine/steps/knowledge.md +++ b/doc/architecture/SPEC/engine/steps/knowledge.md @@ -1,6 +1,6 @@ # engine/steps/knowledge.py —— 规范 - + `知识服务的 knowledge.* 步骤 · refactor-status: new` @@ -34,3 +34,22 @@ PR，且只在 `ALLOW_PUSH=1` 与 `ALLOW_POST=1` 同时成立、非 dry run、 累计预算与草稿。`from_existing` 允许 knowledge/knowledge-deepen 从已合并 KB 开始。 `unlimited_subscription` 显式传入两角色订阅无限模式；预算和语义完成门槛由同一 stage owner 处理。 `kb widen` / `kb deepen` 经同一 playbook/step 指定这两个知识阶段，旧规则 deepen 保持原契约。 + +## 2026-10-06 显式基础知识部分发布 + +`foundation_mode` 缺省 `strict`，未显式选择时不添加部分发布参数。 +`partial` 只允许订阅无限的 `knowledge`，或带原始 `foundation_record` 的 +`knowledge-deepen`；CLI 的 `--foundation-record` 经步骤转换为 `Path`，以 +`foundation_record_path` 传给 `run_stage`。不兼容的阶段、缺失记录和非法模式由 +stage 拒绝，步骤返回 BLOCKED，不能以参数缺失自动降级为部分发布。 + +`knowledge` 的显式部分发布只重放原批次的真实任务和独立评审，不追加模型调用， +不重置修正次数，也不改变生成身份。新深度批次将模式及原发布记录/收据哈希纳入 +身份；七维认可门槛和完整目录分母保持不变。`published` 只表示 PR 已发布， +仍须单独保留 `init_complete=false`、六维基础知识缺口与结构覆盖结果。 +原始 `blocked`/`partial` 状态仍返回 BLOCKED；不能把这些状态当作已完成。 + +证明与发布收据契约由 [`foundation_publication`](../../kb_service/foundation_publication.md) +负责，使用入口见[部分发布说明](../../../../knowledge/partial-foundation.md)。 +接口测试：`test_kb_foundation_partial_interfaces.py`；原生证明和门槛测试： +`test_kb_foundation_explicit_partial.py`。 diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 08b64696..0412d029 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -599,7 +599,7 @@ proof `basis` 缺省 supported，旧区块不改写。verified_absent 当前只 可选政策 `semantic_depth: {per_facet_gt: 0.90}` 对每个维度严格执行大于目标，并要求 每个功能至少一个认可区块；79 功能时每维至少 72，553 分母不缩减。target_met 与 批次遍历完成分别记录。提取前复核结构覆盖；结构门槛不足先阻断，不消耗模型调用。 -结构覆盖与深度门槛均须满足才能发布。深度未达标 dry run 为 partial，保存预览但步骤不成功；发布阻断。 +默认 strict 流程中，结构覆盖与深度门槛均须满足才能发布。深度未达标 dry run 为 partial，保存预览但步骤不成功；发布阻断。 执行 API 的 feature_ids 只限定调度范围；审计仍包含完整政策中的所有功能和维度。 显式列表保留执行顺序并纳入输入摘要；零深度功能仍优先，之后按列表顺序开展首轮。 并行批次各自保留检查点，合并认可页面后须重新通过全量验收，不能发布分组的部分结果。 @@ -618,6 +618,14 @@ modules/knowledge 的公共阶段预算仅在显式启用时改为 None；单独 仍保留原记账上限。该模式绑定检查点身份，无此选项时的旧摘要保持不变； feature-discovery 与 knowledge-deepen 继续使用各自已有的检查点及身份实现。 +`--foundation-mode partial` 是显式基础知识部分发布及其新深度批次交接选项， +不降低七维认可门槛。基础发布只重放已经结束的订阅批次，保留原生成身份、任务、 +修正次数、真实未知项与 `init_complete=false`；源码、原生证明、格式和结构覆盖 +仍须通过。新深度批次要求原始已合并发布记录，绑定模式、记录和收据哈希， +同一完整功能目录和七维分母不变。收据、紧凑 PR 正文和 prepared 恢复契约见 +[`foundation_publication`](kb_service/foundation_publication.md)，操作入口见 +[部分发布说明](../../knowledge/partial-foundation.md)。 + ### 2026-10-02 轻量认可、13 功能任务与轨迹留存 政策可指定 `semantic_depth.acceptance_mode: lightweight`，CLI 可显式选择 diff --git a/doc/architecture/SPEC/kb_service/foundation_publication.md b/doc/architecture/SPEC/kb_service/foundation_publication.md new file mode 100644 index 00000000..b136001e --- /dev/null +++ b/doc/architecture/SPEC/kb_service/foundation_publication.md @@ -0,0 +1,42 @@ +# kb_service/foundation_publication.py —— 规范 + + + +## 职责 + +为显式 `--foundation-mode partial` 保存和复核基础知识发布收据，并为新深度批次 +验证原发布记录。调用方的默认完整门槛保持不变；使用方式见 +[部分发布说明](../../../knowledge/partial-foundation.md)。 + +## 不变量 + +- 只发布已经完成初轮扫描、保留真实原生任务的固定源码批次。已有基线未覆盖的 + 功能和 owner 缺少初轮任务时拒绝发布；已处理范围不能称为已补齐全部知识。 +- 生产文件结构覆盖达到原政策目标；源码、目录与政策身份、引用、格式及原生 + 认可仍须有效。不能修改功能集合、分母、语义认可条件或失败记录来满足门槛。 +- `freeze_receipt` 重放每个任务的原生证明，再冻结生成身份、KB/source SHA、 + 全部终态任务的输入/结果摘要、实际进入页面的认可区块、输出文件哈希及未知项。 + 收据按内容哈希在原始 Git 外阶段目录独占创建；既有字节或路径损坏必须拒绝。 +- `verify_prepared_receipt` 核对原始 partial 模式、收据范围和 prepared 文件， + 从固定源码及实际准备发布的字节重新计算覆盖。不能只信检查点里的计数。 + 旧 strict prepared 发布保持原行为，不能偷偷转换为 partial。 +- `foundation_handoff` 只接受原始 published、非 dry-run 的 partial 记录。 + 记录、收据、任务、prepared 文件、当前合并基线及政策必须一致，且真实 PR + 状态为 MERGED。返回原记录/收据路径与哈希、固定 pin、目录绑定和输出文件哈希。 + 下游以这些字节建立新批次身份，发生变化即拒绝复用。 +- 发布状态只表示 PR 传输。收据保留 `init_complete=false`、原六维目标结果和 + 所有未知项；结构覆盖、六维基础知识、七维深度认可及实际测试执行分别表述。 + +## PR 正文与恢复 + +`render_partial_body` 只渲染汇总数量、固定身份、真实未知维度和收据引用；完整 +模型输入、输出、评审及历史仍在 Git 外。费用未报告时为未知。 +若原 prepared 发布因正文过长而未成功，官方恢复路径须先通过全部原生、源码和 +收据校验，再仅刷新正文元数据：commit、文件字节、任务和修正次数不得改变， +不得重新生成知识或伪造发布状态。 + +## 验证 + +`test_kb_foundation_explicit_partial.py` 覆盖原生证明、范围、收据、恢复与门槛； +`test_kb_foundation_partial_interfaces.py` 覆盖 CLI、campaign、worker 与 assembly +身份传递；`test_kb_foundation_partial_prbody.py` 覆盖紧凑正文和官方正文恢复。 diff --git a/doc/knowledge/partial-foundation.md b/doc/knowledge/partial-foundation.md new file mode 100644 index 00000000..82b8a1bb --- /dev/null +++ b/doc/knowledge/partial-foundation.md @@ -0,0 +1,34 @@ +# Explicit partial foundation publication + +The default initialization workflow requires all six foundation facets for every +catalog feature. `--foundation-mode partial` is an explicit alternative for an +existing, source-pinned subscription batch whose genuine native tasks have +already finished. It reconstructs approved sections and inspects the declared +scope without calling either model again. Unknown sections remain unknown; +local publication exclusions retain their original native evidence archives. + +Publication still requires the configured production-file threshold, valid +catalog and source inventories, genuine native proofs, valid knowledge files, +and a complete terminal scope. Missing initial feature tasks block publication +unless that feature was already covered in the merged baseline. The external +immutable receipt records the original generation identity, source/catalog and +policy, terminal tasks, active approvals, exact output hashes and unknowns. +Prepared publication retries recheck these bindings and the actual pinned +coverage before writing. Publication status describes PR transport: the +foundation target remains false when six-facet gaps remain, and +`init_complete` remains false. + +After this foundation PR merges, start a **new depth batch** with +`--foundation-mode partial --foundation-record /path/to/original/knowledge.json`. +The original published record and receipt, merged output bytes and fixed source +pin become immutable depth inputs. The campaign and standalone depth audit use +the same structural prerequisite; six-facet foundation completion is reported +separately. The audit also requires native approval reports in this mode. + +This option preserves the catalog and all semantic acceptance criteria. Each +feature must have recognized depth knowledge, each of the seven depth facets +must exceed the policy threshold, and production coverage must meet the policy +target. The denominator remains the complete frozen catalog times seven. +Partition workers report their actual execution scope separately from global +semantic completion; completing a partition never implies the whole catalog +has passed. Retrieval, independent review and CI remain delivery gates. diff --git a/eval/knowledge-depth/assemble_depth_campaign.py b/eval/knowledge-depth/assemble_depth_campaign.py index 609e9728..5f0dff71 100644 --- a/eval/knowledge-depth/assemble_depth_campaign.py +++ b/eval/knowledge-depth/assemble_depth_campaign.py @@ -172,7 +172,8 @@ def _filtered(text, page, retired): return text -def assemble(root: Path, state: Path, source: Path, *, retirement_reports=()): +def assemble(root: Path, state: Path, source: Path, *, retirement_reports=(), + foundation_mode="strict", foundation_record_path: Path | None = None): """Return metadata and validated writes without modifying any input.""" root, state, source = root.resolve(), state.resolve(), source.resolve() metadata_raw = (state / "campaign.json").read_bytes() @@ -181,6 +182,25 @@ def assemble(root: Path, state: Path, source: Path, *, retirement_reports=()): if not isinstance(repo, str) or not re.fullmatch(r"[a-zA-Z0-9_-]+", repo) \ or not all(isinstance(sha, str) and re.fullmatch(r"[0-9a-f]{40}", sha) for sha in (baseline, pin)): raise ValueError("campaign needs a safe repo and immutable source/knowledge SHAs") + if foundation_mode not in ("strict", "partial") or ( + foundation_mode == "partial") != bool(foundation_record_path): + raise ValueError("partial assembly requires its foundation record; strict assembly cannot use one") + if campaign.get("foundation_mode", "strict") != foundation_mode: + raise ValueError("assembly foundation mode differs from campaign") + foundation = None + if foundation_mode == "partial": + from infermatrix_copilot.kb_service.foundation_publication import foundation_handoff + from infermatrix_copilot.kb_service.init_support import InitError + from infermatrix_copilot.kb_service.sources import KnowledgeRepo + try: + foundation = foundation_handoff(foundation_record_path, knowledge=KnowledgeRepo(root), + baseline=baseline, repo=repo, pin=pin) + except InitError as exc: + raise ValueError(str(exc)) from exc + if campaign.get("foundation") != foundation: + raise ValueError("assembly foundation record differs from campaign") + elif campaign.get("foundation") is not None: + raise ValueError("strict assembly cannot reuse partial foundation metadata") if _git(root, "rev-parse", baseline + "^{commit}").decode().strip() != baseline \ or _git(source, "rev-parse", "HEAD").decode().strip() != pin \ or _git(source, "status", "--porcelain", "--untracked-files=no").strip(): @@ -236,6 +256,12 @@ def assemble(root: Path, state: Path, source: Path, *, retirement_reports=()): (("stage", "knowledge-deepen"), ("repo", repo), ("pin", pin), ("kb_base_sha", baseline))): raise ValueError("worker checkpoint identity differs from campaign: " + worker) depth = record.get("depth", {}) + if foundation is not None and (depth.get("foundation_mode") != "partial" + or depth.get("foundation") != foundation): + raise ValueError("worker foundation identity differs from campaign: " + worker) + if foundation is None and (depth.get("foundation_mode", "strict") != "strict" + or depth.get("foundation") is not None): + raise ValueError("strict campaign cannot reuse partial worker state: " + worker) entries, accepted = depth.get("features", {}), depth.get("accepted", {}) if not isinstance(entries, dict) or not isinstance(accepted, dict) or set(entries) - set(features): raise ValueError("worker checkpoint has invalid feature state: " + worker) @@ -267,6 +293,8 @@ def assemble(root: Path, state: Path, source: Path, *, retirement_reports=()): eligible_features.add(feature.id) old = baseline_pages.get(page) related = f"[功能概览]({PurePosixPath(feature.page).name}) · [owner 入口](_index.md)" + if foundation_mode == "partial" and _baseline(root, baseline, "knowledge/" + feature.page) is None: + related = "[owner 入口](_index.md)" expected_body = _outside_blocks(old) if old is not None else ["# " + _one_line(feature.title) + "：实现深读", related] if _outside_blocks(candidate) != expected_body: raise ValueError("accepted page alters prose outside approved blocks: " + feature.id) @@ -342,6 +370,8 @@ def assemble(root: Path, state: Path, source: Path, *, retirement_reports=()): "eligible_feature_count": len(eligible_features), "pages": changes, "index_links": new_links, "planned_files": sorted(writes), "receipts": receipts, "native_archive_audit_required": True, "problems": []} + if foundation is not None: + report.update(foundation_mode="partial", foundation=foundation) return report, writes @@ -351,6 +381,8 @@ def main(): parser.add_argument("--state", type=Path, required=True) parser.add_argument("--source-tree", type=Path, required=True) parser.add_argument("--retirement-report", type=Path, action="append", default=[]) + parser.add_argument("--foundation-mode", choices=("strict", "partial"), default="strict") + parser.add_argument("--foundation-record", type=Path) parser.add_argument("--report", type=Path) parser.add_argument("--apply", action="store_true") args = parser.parse_args() @@ -364,7 +396,8 @@ def main(): for n in range(campaign["workers"]))]: handle = stack.enter_context(lock.open("r")) fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) - report, writes = assemble(args.root, args.state, args.source_tree, retirement_reports=args.retirement_report) + report, writes = assemble(args.root, args.state, args.source_tree, retirement_reports=args.retirement_report, + foundation_mode=args.foundation_mode, foundation_record_path=args.foundation_record) report["applied"] = args.apply if args.apply: for page, text in writes.items(): diff --git a/eval/knowledge-depth/run_depth_campaign.py b/eval/knowledge-depth/run_depth_campaign.py index a3916dca..def9e3d5 100644 --- a/eval/knowledge-depth/run_depth_campaign.py +++ b/eval/knowledge-depth/run_depth_campaign.py @@ -108,6 +108,29 @@ def discovery_handoff(args): "report_path": report_path, **binding} +def foundation_handoff(args): + """Bind explicit partial depth to the genuine published foundation bytes.""" + mode = getattr(args, "foundation_mode", "strict") + path = getattr(args, "foundation_record", None) + if mode == "strict" and path is None: + return None + if mode != "partial" or path is None: + raise ValueError("partial depth requires --foundation-record; strict depth cannot use one") + from infermatrix_copilot.kb_service.foundation_publication import foundation_handoff as validate + from infermatrix_copilot.kb_service.init_support import InitError + from infermatrix_copilot.kb_service.sources import KnowledgeRepo + + try: + binding = validate(Path(path), knowledge=KnowledgeRepo(args.root), baseline=args.baseline, + repo=args.repo, pin=args.pin) + except InitError as exc: + raise ValueError(str(exc)) from exc + expected = getattr(args, "foundation_record_sha256", None) + if expected is not None and expected != binding["record_sha256"]: + raise ValueError("foundation prerequisite record changed after campaign binding") + return binding + + def install_discovery_handoff(args, state, binding): """Copy the exact immutable prerequisite into a worker without rewriting it.""" if binding is None: @@ -162,6 +185,11 @@ def run_worker(args, state): if metadata.get("discovery") != binding: raise RuntimeError("worker discovery handoff differs from the frozen campaign identity") install_discovery_handoff(args, state, binding) + foundation = foundation_handoff(args) + if foundation is not None: + metadata = json.loads((args.state / "campaign.json").read_bytes()) + if metadata.get("foundation_mode") != "partial" or metadata.get("foundation") != foundation: + raise RuntimeError("worker foundation handoff differs from the frozen campaign identity") 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( @@ -175,7 +203,9 @@ def run_worker(args, state): feature_ids=tuple(args.feature_ids.split(",")), retry_unfinished=args.retry, acceptance_mode=getattr(args, "acceptance_mode", "strict"), depth_index_path=getattr(args, "depth_index_path", None), - stop_file=getattr(args, "stop_file", None)) + stop_file=getattr(args, "stop_file", None), + **({"foundation_mode": "partial", "foundation_record_path": Path(foundation["record_path"])} + if foundation is not None else {})) if record.kb_base_sha != args.baseline or record.pin != args.pin: raise RuntimeError("worker completed against a different source or knowledge baseline") print(json.dumps({"worker": args.worker, "status": record.status, "problems": record.problems, @@ -200,6 +230,10 @@ def main(): 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("--foundation-mode", choices=("strict", "partial"), default="strict") + parser.add_argument("--foundation-record", type=Path, + help="Original published partial foundation record; required for explicit partial depth") + parser.add_argument("--foundation-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) @@ -219,6 +253,10 @@ def main(): args.repair_guidance = args.repair_guidance.resolve() if args.discovery_record: args.discovery_record = args.discovery_record.resolve() + if args.foundation_record: + args.foundation_record = args.foundation_record.resolve() + if (args.foundation_mode == "partial") != bool(args.foundation_record): + parser.error("partial foundation mode requires --foundation-record; strict mode cannot use one") if args.zcode_pacing_path: args.zcode_pacing_path = args.zcode_pacing_path.resolve() from infermatrix_copilot.kb_service.depth_pacing import SharedZcodePacer @@ -234,6 +272,8 @@ def main(): 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") + if args.foundation_record and not re.fullmatch(r"[0-9a-f]{64}", args.foundation_record_sha256 or ""): + parser.error("partial workers require their parent's immutable foundation 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) @@ -272,10 +312,13 @@ def campaign(args, parser): "acceptance_mode": getattr(args, "acceptance_mode", "strict")} try: binding = discovery_handoff(args) + foundation = foundation_handoff(args) except (ValueError, TypeError, OSError) as exc: parser.error(str(exc)) if binding is not None: summary["discovery"] = binding + if foundation is not None: + summary.update(foundation_mode="partial", foundation=foundation) 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, @@ -287,7 +330,9 @@ def campaign(args, parser): metadata = args.state / "campaign.json" if metadata.exists(): prior = json.loads(metadata.read_text()) - if prior.get("repair_guidance_sha256") != summary.get("repair_guidance_sha256") or any( + if (prior.get("foundation_mode", "strict") != summary.get("foundation_mode", "strict") + or prior.get("foundation") != summary.get("foundation") + or prior.get("repair_guidance_sha256") != summary.get("repair_guidance_sha256")) or any( prior.get(key) != value for key, value in summary.items()): parser.error("campaign identity changed; use a fresh state directory") stop_file = getattr(args, "stop_file", None) or args.state / "STOP" @@ -354,6 +399,9 @@ def campaign(args, parser): if binding is not None: argv += ["--discovery-record", binding["record_path"], "--discovery-record-sha256", binding["record_sha256"]] + if foundation is not None: + argv += ["--foundation-mode", "partial", "--foundation-record", foundation["record_path"], + "--foundation-record-sha256", foundation["record_sha256"]] if args.retry: argv.append("--retry") jobs.append((number, argv)) diff --git a/src/infermatrix_copilot/engine/steps/knowledge.py b/src/infermatrix_copilot/engine/steps/knowledge.py index c6a06720..e3c50b38 100644 --- a/src/infermatrix_copilot/engine/steps/knowledge.py +++ b/src/infermatrix_copilot/engine/steps/knowledge.py @@ -187,6 +187,11 @@ async def init_stage(ctx: StepContext) -> StepResult: kwargs[option] = True if ctx.params.get("acceptance_mode"): kwargs["acceptance_mode"] = str(ctx.params["acceptance_mode"]) + if ctx.params.get("foundation_mode"): + kwargs["foundation_mode"] = str(ctx.params["foundation_mode"]) + if ctx.params.get("foundation_record"): + from pathlib import Path + kwargs["foundation_record_path"] = Path(ctx.params["foundation_record"]) record = run_stage(rt, lifecycle, stage, dry_run=dry_run, pin=pin, **kwargs) except (InitError, NotImplementedError, ValueError) as exc: return StepResult(False, FailureKind.BLOCKED, str(exc)) diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index 8d5abd56..a65b35f2 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -140,6 +140,18 @@ def _init_command(args, state_dir: Path) -> int: from .runner import run_playbook params = {"stage": args.stage, "dry_run": "true" if args.dry_run else "false", "pin": args.pin or ""} + foundation_mode = getattr(args, "foundation_mode", "strict") + foundation_record = getattr(args, "foundation_record", None) + if foundation_mode != "strict" or foundation_record is not None: + if foundation_mode != "partial" or args.stage not in ("knowledge", "knowledge-deepen") or not args.unlimited_subscription: + print("--foundation-mode partial requires unlimited knowledge or knowledge-deepen", file=sys.stderr) + return 2 + if bool(foundation_record) != (args.stage == "knowledge-deepen"): + print("--foundation-record is required only for partial knowledge-deepen", file=sys.stderr) + return 2 + params["foundation_mode"] = foundation_mode + if foundation_record is not None: + params["foundation_record"] = str(foundation_record) if getattr(args, "acceptance_mode", "strict") != "strict": if args.stage != "knowledge-deepen": print("--acceptance-mode is for knowledge-deepen only", file=sys.stderr) @@ -235,6 +247,10 @@ def main(argv: list[str] | None = None) -> int: 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") + init.add_argument("--foundation-mode", choices=("strict", "partial"), default="strict", + help="explicitly publish retained partial foundation or bind depth to that publication; targets stay unchanged") + init.add_argument("--foundation-record", type=Path, + help="original published partial foundation record, required for partial knowledge-deepen") for name, stage in (("widen", "knowledge"), ("deepen", "knowledge-deepen")): cmd = sub.add_parser(name, help="feature breadth" if name == "widen" else "feature implementation depth") cmd.add_argument("repo") @@ -242,6 +258,8 @@ def main(argv: list[str] | None = None) -> int: cmd.add_argument("--pin") cmd.add_argument("--subscription-generator", action="store_true") cmd.add_argument("--unlimited-subscription", action="store_true") + cmd.add_argument("--foundation-mode", choices=("strict", "partial"), default="strict") + cmd.add_argument("--foundation-record", type=Path) if name == "deepen": cmd.add_argument("--budget-usd", type=float) cmd.add_argument("--retry-unfinished", action="store_true") diff --git a/src/infermatrix_copilot/kb_service/foundation_publication.py b/src/infermatrix_copilot/kb_service/foundation_publication.py new file mode 100644 index 00000000..806bc236 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/foundation_publication.py @@ -0,0 +1,231 @@ +"""Explicit, publication-only partial foundation; generation proofs stay immutable.""" +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from .init_support import InitError, InitRecord, load_prepared +from .knowledge_coverage import audit_coverage, coverage_targets_met, load_policy + + +def _hash(value): + return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _file_hashes(files): + return {path: hashlib.sha256(text.encode()).hexdigest() for path, text in sorted(files.items())} + + +def _payload(stage, record, files): + targets = record.coverage.get("knowledge", {}).get("targets", {}) + scope = record.coverage.get("foundation_scope", {}) + tasks = record.coverage.get("foundation_jobs", {}).get("tasks", {}) + policy_text = stage.rt.knowledge.show(record.kb_base_sha, stage._coverage_policy_path()) or "" + policy = load_policy(policy_text, stage.repo_dir) + if (policy.semantic_depth_per_facet_gt is None + or not coverage_targets_met(targets, "partial") or not scope.get("complete") + or scope.get("features") != [f.id for f in policy.features] + or scope.get("index_identity", {}).get("pin") != record.pin + or scope.get("terminal_tasks") != sorted(tasks) or not tasks + or record.coverage["foundation_jobs"].get("binding") != record.inputs_digest): + raise InitError("partial foundation needs valid structural coverage and a complete terminal scope") + head = stage.rt.knowledge.knowledge_files(record.kb_base_sha) + head.update({p[len("knowledge/"):]: text for p, text in files.items() if p.startswith("knowledge/")}) + if getattr(stage.rt, "upstream", None): + validate_initial_scope(stage, record) + active = [] + for key, task in sorted(tasks.items()): + for artifact in task.get("artifacts", []): + if artifact["text"] in head.get(artifact["page"], ""): + active.append({"task": key, "page": artifact["page"], "facet": artifact["facet"], + "artifact_sha256": _hash(artifact), "text_sha256": artifact["text_sha256"], + "generator_receipt_sha256": _hash(artifact["generator_receipt"]), + "judge_receipt_sha256": _hash(artifact["judge_receipt"])}) + return {"schema_version": 1, "foundation_mode": "partial", "init_complete": False, + "repo": record.repo, "pin": record.pin, "kb_base_sha": record.kb_base_sha, + "generation_inputs_digest": record.inputs_digest, + "policy_path": stage._coverage_policy_path(), "policy_sha256": hashlib.sha256(policy_text.encode()).hexdigest(), + "catalog_binding": record.discovery.get("catalog_binding"), + "targets_sha256": _hash(targets), "foundation_targets_met": targets.get("met") is True, + "structural_targets_met": True, "scope": scope, + "tasks": {key: {"input_sha256": t["input_sha256"], "result_sha256": t["result_sha256"]} + for key, t in sorted(tasks.items())}, + "active_native_approvals": active, "files_sha256": _file_hashes(files), + "unknown_foundation_facets": {fid: f["missing_facets"] for fid, f in targets["features"]["items"].items() + if f["missing_facets"]}} + + +def validate_initial_scope(stage, record): + """Fail before writers when the retained batch never processed an initial shard.""" + import tempfile + from .init_knowledge import FACETS, _MARKER + policy_text = stage.rt.knowledge.show(record.kb_base_sha, stage._coverage_policy_path()) or "" + policy = load_policy(policy_text, stage.repo_dir) + tasks = record.coverage.get("foundation_jobs", {}).get("tasks", {}) + base = stage.rt.knowledge.knowledge_files(record.kb_base_sha) + with tempfile.TemporaryDirectory(prefix="kb-foundation-scope-") as scratch: + source = stage.rt.upstream(record.repo, stage.lifecycle.full_name).export(record.pin, Path(scratch) / "tree") + baseline = audit_coverage(base, source, policy, full_name=stage.lifecycle.full_name, pin=record.pin) + for feature in policy.features: + if not baseline["features"]["items"][feature.id]["covered"] and not any( + t.get("owner") == "feature-" + feature.id and t.get("page") == feature.page for t in tasks.values()): + raise InitError("partial foundation scope has a missing initial feature task") + for owner in record.coverage.get("knowledge", {}).get("owners", {}): + facets = {facet for text in base.values() for name, facet, pin, verdict in _MARKER.findall(text) + if name == owner and pin == record.pin and verdict in ("", "pass")} + if not set(FACETS) <= facets and not any(t.get("owner") == owner for t in tasks.values()): + raise InitError("partial foundation scope has a missing initial owner task") + + +def freeze_receipt(stage, files): + """Freeze the local publication criterion after all source/native/tree checks.""" + from .init_knowledge_parallel import _validate_result + from .models import ModelGateway + if not isinstance(stage.rt.gateway, ModelGateway): + raise InitError("partial foundation publication requires original native model receipts") + for result in stage.record.coverage.get("foundation_jobs", {}).get("tasks", {}).values(): + _validate_result(stage, result) + payload = _payload(stage, stage.record, files) + raw = (json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode() + sha = hashlib.sha256(raw).hexdigest() + folder = InitRecord.path(stage.rt.state_dir, stage.record.repo, "knowledge").parent / "foundation-publication" + if folder.is_symlink(): + raise InitError("partial foundation receipt archive cannot be a symlink") + folder.mkdir(parents=True, exist_ok=True) + path = folder / (sha + ".json") + if path.exists(): + if path.is_symlink() or path.read_bytes() != raw: + raise InitError("partial foundation receipt archive differs") + else: + with path.open("xb") as stream: + stream.write(raw) + stage.record.coverage["foundation_publication"] = { + "foundation_mode": "partial", "init_complete": False, + "foundation_targets_met": payload["foundation_targets_met"], + "structural_targets_met": True, "unknown_foundation_facets": payload["unknown_foundation_facets"], + "receipt": {"path": str(path.resolve()), "sha256": sha, "bytes": len(raw)}} + stage.record.notes.append("explicit partial foundation publication; six-facet targets and unknowns retained; initialization is incomplete") + + +def _read_receipt(record): + try: + ref = record.coverage["foundation_publication"]["receipt"] + path = Path(ref["path"]) + raw = path.read_bytes() + if (path.is_symlink() or path.name != ref["sha256"] + ".json" + or hashlib.sha256(raw).hexdigest() != ref["sha256"] or len(raw) != ref["bytes"]): + raise ValueError("archive hash or size differs") + payload = json.loads(raw) + summary = record.coverage["foundation_publication"] + expected = {key: payload[key] for key in ("foundation_mode", "init_complete", + "foundation_targets_met", "structural_targets_met", "unknown_foundation_facets")} + if {key: value for key, value in summary.items() if key != "receipt"} != expected: + raise ValueError("publication summary differs from receipt") + return payload, ref + except (OSError, ValueError, KeyError, TypeError) as exc: + raise InitError("partial foundation receipt unavailable or changed") from exc + + +def verify_prepared_receipt(stage, previous): + partial = previous.coverage.get("foundation_publication", {}).get("foundation_mode") == "partial" + if not partial and stage.foundation_mode == "strict": + return # Exact legacy prepared publications retain their original behavior. + if not partial or stage.foundation_mode != "partial": + raise InitError("prepared foundation publication mode differs; restore the original explicit mode") + payload, ref = _read_receipt(previous) + folder = InitRecord.path(stage.rt.state_dir, previous.repo, "knowledge").parent / "foundation-publication" + if not Path(ref["path"]).resolve().is_relative_to(folder.resolve()): + raise InitError("partial foundation receipt is outside its original archive") + prepared = load_prepared(previous.pr["prepared"]) + if prepared["base_sha"] != previous.kb_base_sha or payload != _payload(stage, previous, prepared["files"]): + raise InitError("prepared foundation publication differs from its frozen receipt") + # Recompute structural status from actual saved bytes and the fixed source, + # rather than trusting mutable counters in the checkpoint. + import tempfile + upstream = stage.rt.upstream(previous.repo, stage.lifecycle.full_name) + with tempfile.TemporaryDirectory(prefix="kb-foundation-publication-") as scratch: + tree = upstream.export(previous.pin, Path(scratch) / "tree") + head = stage.rt.knowledge.knowledge_files(previous.kb_base_sha) + head.update({p[len("knowledge/"):]: t for p, t in prepared["files"].items() if p.startswith("knowledge/")}) + policy_text = stage.rt.knowledge.show(previous.kb_base_sha, stage._coverage_policy_path()) + policy = load_policy(policy_text, stage.repo_dir) + actual = audit_coverage(head, tree, policy, full_name=stage.lifecycle.full_name, pin=previous.pin) + saved = dict(previous.coverage["knowledge"]["targets"]) + saved.pop("policy_sha256", None) + if actual != saved: + raise InitError("prepared foundation source coverage differs from its frozen receipt") + + +def foundation_handoff(record_path, *, knowledge, baseline, repo, pin, publisher=None): + """Bind a new depth batch to the genuine merged partial foundation output.""" + try: + path = Path(record_path) + raw = path.read_bytes() + record = InitRecord(**json.loads(raw)) + payload, ref = _read_receipt(record) + prepared = load_prepared(record.pr["prepared"]) + if (record.stage != "knowledge" or record.status != "published" or record.dry_run or record.repo != repo or record.pin != pin + or record.coverage["foundation_publication"]["foundation_mode"] != "partial" + or payload["repo"] != repo or payload["pin"] != pin or payload["init_complete"] is not False + or payload["generation_inputs_digest"] != record.inputs_digest + or payload["kb_base_sha"] != record.kb_base_sha + or payload["files_sha256"] != _file_hashes(prepared["files"]) + or payload["tasks"] != {k: {"input_sha256": t["input_sha256"], "result_sha256": t["result_sha256"]} + for k, t in sorted(record.coverage["foundation_jobs"]["tasks"].items())} + or not payload["scope"].get("complete")): + raise ValueError("record/receipt binding differs") + from types import SimpleNamespace + import yaml + manifest_path = str(Path(payload["policy_path"]).parent / "manifest.yaml") + manifest = yaml.safe_load(knowledge.show(record.kb_base_sha, manifest_path)) + context = SimpleNamespace(rt=SimpleNamespace(knowledge=knowledge), repo_dir=manifest["knowledge"]["repo_subdir"], + _coverage_policy_path=lambda: payload["policy_path"]) + if prepared["base_sha"] != record.kb_base_sha or payload != _payload(context, record, prepared["files"]): + raise ValueError("raw record differs from the frozen receipt") + for file, expected in payload["files_sha256"].items(): + actual = knowledge.show(baseline, file) + if actual is None or hashlib.sha256(actual.encode()).hexdigest() != expected: + raise ValueError("merged baseline differs from the emitted foundation files") + policy_text = knowledge.show(baseline, payload["policy_path"]) + if not policy_text or hashlib.sha256(policy_text.encode()).hexdigest() != payload["policy_sha256"]: + raise ValueError("merged foundation policy differs") + if publisher is None: + from .init_stages import _knowledge_repository + from .init_support import InitPublisher + publisher = InitPublisher(knowledge.path, _knowledge_repository()) + if publisher.pr_state(int(record.pr["number"])) != "MERGED": + raise ValueError("foundation PR is not merged") + return {"record_path": str(path.resolve()), "record_sha256": hashlib.sha256(raw).hexdigest(), + "receipt_path": ref["path"], "receipt_sha256": ref["sha256"], "pin": pin, + "policy_sha256": payload["policy_sha256"], "catalog_binding": payload["catalog_binding"], + "files_sha256": payload["files_sha256"]} + except (OSError, ValueError, KeyError, TypeError, InitError) as exc: + raise InitError("partial foundation handoff is unavailable or differs from the merged baseline") from exc + + +def render_partial_body(record, lifecycle): + """Bounded PR metadata; full native verdicts/history remain in the record.""" + from collections import Counter + publication = record.coverage["foundation_publication"] + targets = record.coverage["knowledge"]["targets"] + core, features = targets["core"], targets["features"] + unknown = publication["unknown_foundation_facets"] + counts = Counter(facet for facets in unknown.values() for facet in facets) + ref = publication["receipt"] + lines = [f"`kb init knowledge`：`{lifecycle.repo}`（`{lifecycle.full_name}`）的基础知识部分发布。", "", + "本次仅发布已获原生独立评审认可的内容，保留真实未知项。初始化尚未完成。", "", + f"- 固定源码：`{record.pin}`", f"- 知识基线：`{record.kb_base_sha}`", + f"- 原生成身份：`{record.inputs_digest}`（保留任务及修正次数）", + f"- 六维基础知识完整功能：{features['covered']}/{features['total']}；基础目标达标：{targets['met']}。", + f"- 生产文件结构覆盖：{core['covered']}/{core['total']}（{core['ratio']:.2%}），政策目标 {core['target']:.2%}。", + f"- 尚未知的基础维度：{sum(counts.values())} 项，涉及 {len(unknown)} 个功能。", + f"- 本次模型追加调用：{record.coverage['foundation_scope']['native_dispatches']}；实际订阅费用：未知。", + "- `init_complete=false`；七维深度认可、检索验收、独立审查和 CI 仍为后续交付门槛。", "", + "| 基础维度 | 未知项 |", "| --- | ---: |"] + lines += [f"| {facet} | {counts.get(facet, 0)} |" for facet in + ("architecture", "api", "configuration", "features", "tradeoffs", "validation")] + lines += ["", f"- 冻结发布收据：`{ref['path']}`", f"- 收据 SHA-256：`{ref['sha256']}`", + "- 完整输入、输出、工具轨迹、原生评审、失败原因和未决项保存在原始 Git 外阶段记录及收据中。", + "- 固定 SHA 的引用、原生证明、知识结构和格式已校验；本 PR 合并仍需完整 CI 与独立审查。", ""] + return "\n".join(lines) diff --git a/src/infermatrix_copilot/kb_service/init_knowledge.py b/src/infermatrix_copilot/kb_service/init_knowledge.py index b75ed535..1e967af2 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge.py @@ -28,9 +28,9 @@ from .init_coverage import Owner from .init_stages import _Chain, _Stage, _numbered, _one_line, _page_frontmatter, neutral_headings from .init_support import InitRecord, classify_verdict, generate, judge -from .init_knowledge_inputs import SYSTEM_KNOWLEDGE, knowledge_prompt, source_owner, source_owners +from .init_knowledge_inputs import SYSTEM_KNOWLEDGE, knowledge_prompt, source_owner, source_owners, foundation_evidence from .models import ModelUnavailable -from .init_knowledge_parallel import offered_ranges, parallelism, run_jobs, restore_jobs, preferred_sources, start_interval +from .init_knowledge_parallel import offered_ranges, parallelism, run_jobs, restore_jobs, preferred_sources, start_interval, range_is_offered, related_test_sources FACETS = ("architecture", "api", "configuration", "tradeoffs", "features", "validation") MAX_SOURCE_BYTES = 100_000 @@ -72,6 +72,8 @@ def validate_sections(data: dict) -> None: @dataclass class _Knowledge(_Stage): STAGE = "knowledge" + foundation_mode: str = "strict" + foundation_record_path: Path | None = None def _chain(self) -> _Chain: if not self.from_existing: @@ -108,6 +110,8 @@ def _precheck(self) -> list[str]: except (ValueError, TypeError) as exc: return [f"knowledge coverage policy: {exc}"] self.coverage_policy = policy + if self.foundation_mode == "partial" and policy.semantic_depth_per_facet_gt is None: + return ["partial foundation requires a final semantic-depth acceptance policy"] for feature in policy.features: existing = self.base.get(feature.page) if existing: @@ -120,7 +124,8 @@ def _input_options(self) -> dict: path = self._coverage_policy_path() text = self.overlay.get(path) or self.rt.knowledge.show(self._base_sha, path) or "" options = {"knowledge_policy": hashlib.sha256(text.encode()).hexdigest(), - "from_existing": self.from_existing, "knowledge_prompt_version": 3} + "from_existing": self.from_existing, + "knowledge_prompt_version": 4 if self.rt.unlimited_subscription else 3} if self.rt.unlimited_subscription: options["foundation_parallel"] = {"version": 1, "workers": parallelism(self.rt), "zcode_start_interval_s": start_interval(self.rt)} @@ -135,6 +140,8 @@ def _cache_reusable(self, previous: InitRecord) -> bool: return True # Preserve immutable historical publication records. if self.rt.unlimited_subscription and previous.status != "published": self._validate_unpublished_checkpoint(previous, previous.inputs_digest) + if self.STAGE == "knowledge" and self.foundation_mode == "partial": + return False # Reconstruct a truthful publication-only scope from the saved native tasks. return not (self.rt.unlimited_subscription and previous.unfinished) def _resume_input_problems(self, previous: InitRecord, digest: str) -> list[str]: @@ -142,6 +149,9 @@ def _resume_input_problems(self, previous: InitRecord, digest: str) -> list[str] # Raise before generic _blocked writes: restoring an archive must # leave the same genuine prepared publication resumable. self._validate_unpublished_checkpoint(previous, digest) + if self.STAGE == "knowledge": + from .foundation_publication import verify_prepared_receipt + verify_prepared_receipt(self, previous) return super()._resume_input_problems(previous, digest) def _validate_unpublished_checkpoint(self, previous: InitRecord, digest: str) -> None: @@ -149,6 +159,10 @@ def _validate_unpublished_checkpoint(self, previous: InitRecord, digest: str) -> validate_checkpoint(self, previous, digest) def _restore_progress(self, previous: InitRecord | None) -> list[str]: + if self.STAGE == "knowledge" and self.foundation_mode == "partial" and previous is not None: + self._validate_unpublished_checkpoint(previous, self.record.inputs_digest) + from .foundation_publication import validate_initial_scope + validate_initial_scope(self, previous) if (self.rt.unlimited_subscription and previous is not None and previous.coverage.get("foundation_jobs") and previous.inputs_digest != self.record.inputs_digest): @@ -163,12 +177,28 @@ def _restore_progress(self, previous: InitRecord | None) -> list[str]: def _publish(self, changed: dict[str, str]) -> InitRecord: targets = self.record.coverage.get("knowledge", {}).get("targets", {}) + if self.STAGE == "knowledge" and self.foundation_mode == "partial": + from .foundation_publication import freeze_receipt + freeze_receipt(self, changed) + return super()._publish(changed) if (self.rt.unlimited_subscription and not self.dry_run and targets.get("required") is True and targets.get("met") is not True): return self._blocked(["foundation targets incomplete; native progress retained; " "resume this same pinned publication batch to fill missing knowledge"]) return super()._publish(changed) + def _resume(self, previous: InitRecord) -> InitRecord: + if self.STAGE == "knowledge" and self.foundation_mode == "partial": + from .init_stages import render_pr_body + from .init_support import load_prepared, save_prepared + # The common run path has already replayed native/source/receipt + # and checked the frozen catalog, branch and generation identity. + prepared = load_prepared(previous.pr["prepared"]) + body = render_pr_body(previous, self.lifecycle) + if prepared["body"] != body: + save_prepared(Path(previous.pr["prepared"]), **{**prepared, "body": body}) + return super()._resume(previous) + def _build(self, tree: Path) -> InitRecord: if self.route_source == "none": return self._blocked(["knowledge needs owner routes: run and merge skeleton and modules first"]) @@ -177,6 +207,12 @@ def _build(self, tree: Path) -> InitRecord: from .feature_discovery_index import build_for_stage self.source_index = build_for_stage(tree, self) + publication_only = self.foundation_mode == "partial" + if publication_only and not getattr(self, "coverage_policy", None): + return self._blocked(["partial foundation publication requires an explicit coverage policy"]) + visited_owners, visited_features = [], [] + if publication_only and not self.record.coverage.get("foundation_jobs", {}).get("tasks"): + return self._blocked(["partial foundation publication requires a genuine saved native batch"]) files: dict[str, list[str]] = {} unrouted = [] policy = getattr(self, "coverage_policy", None) @@ -210,6 +246,7 @@ def _build(self, tree: Path) -> InitRecord: claims, evidence = {}, [] owner_jobs = [] for position, name in enumerate(order): + visited_owners.append(name) owner = owners[name] page = self._page_for(owner) existing = self.head.get(page, "") @@ -231,15 +268,20 @@ def _build(self, tree: Path) -> InitRecord: "source_files": len(files[name]), "shown_files": 0} if not requested: continue - source = self._sources(tree, sorted(files[name], key=lambda p: (-churn[p], p)), MAX_SOURCE_BYTES) + if publication_only: + continue # No fourth extraction or review: existing task proofs were replayed by restore_jobs. + tests = related_test_sources(self, files[name]) if self.rt.unlimited_subscription else [] + test_bytes = sum(len(item["text"].encode()) for item in tests) + source = self._sources(tree, sorted(files[name], key=lambda p: (-churn[p], p)), MAX_SOURCE_BYTES - test_bytes) + source.extend(tests) docs = self._docs_for(tree, owner) offered = offered_ranges(source + docs) - report[name]["shown_files"] = len(source) - report[name]["partial_files"] = [item["path"] for item in source if item["end"] < item["total_lines"]] + report[name]["shown_files"] = len(source) - len(tests) + report[name]["partial_files"] = [item["path"] for item in source if not item.get("test_context") and item["end"] < item["total_lines"]] self.record.unfinished.extend(f"{name}: byte cap truncated {path}" for path in report[name]["partial_files"]) - if len(source) < len(files[name]): - self.record.unfinished.append(f"{name}: source cap showed {len(source)} of {len(files[name])} files") + if len(source) - len(tests) < len(files[name]): + self.record.unfinished.append(f"{name}: source cap showed {len(source) - len(tests)} of {len(files[name])} files") payload = {"repository": self.lifecycle.full_name, "pin": self.record.pin, "owner": name, "facets": requested, "files": source, "docs": docs, "existing_knowledge": self._bounded_context(existing_pages), @@ -247,6 +289,7 @@ def _build(self, tree: Path) -> InitRecord: "related_owners": [{"owner": o.owner, "scope_prefixes": list(o.prefixes)} for o in self.owners]} if self.rt.unlimited_subscription: + payload["foundation_prompt_version"] = 4 owner_jobs.append({"owner": owner, "page": page, "payload": payload, "offered": offered, "requested": requested}) continue @@ -312,12 +355,18 @@ def _build(self, tree: Path) -> InitRecord: targets = audit_coverage(self.head, tree, policy, full_name=self.lifecycle.full_name, pin=self.record.pin) feature_jobs = [] for feature in policy.features: + visited_features.append(feature.id) + if publication_only: + continue if targets["features"]["items"][feature.id]["covered"]: continue paths = [p.relative_to(tree).as_posix() for p in tree.rglob("*") if p.is_file() and matches(p.relative_to(tree).as_posix(), feature.source_globs)] source_paths = list(dict.fromkeys(list(feature.entry_points) + sorted(paths))) - source = preferred_sources(self, tree, feature.id, source_paths, MAX_SOURCE_BYTES) + tests = related_test_sources(self, source_paths) if self.rt.unlimited_subscription else [] + test_bytes = sum(len(item["text"].encode()) for item in tests) + source = preferred_sources(self, tree, feature.id, source_paths, MAX_SOURCE_BYTES - test_bytes) + source.extend(tests) docs = self._sources(tree, list(feature.docs), MAX_DOC_BYTES) if not source: self.record.unfinished.append(f"feature {feature.id}: missing readable source") @@ -330,6 +379,7 @@ def _build(self, tree: Path) -> InitRecord: "existing_knowledge": self._bounded_context({feature.page: self.head.get(feature.page, "")}), "language_sample": self._language_sample()} if self.rt.unlimited_subscription: + payload["foundation_prompt_version"] = 4 feature_jobs.append({"owner": Owner("feature-" + feature.id, feature.page, ()), "page": feature.page, "payload": payload, "offered": offered, "requested": missing_facets}) @@ -369,6 +419,16 @@ def _build(self, tree: Path) -> InitRecord: for feature in policy.features: if feature.page in self.head: self.head[feature.page] = feature_metadata(self.head[feature.page], feature) + if publication_only: + self.record.coverage["foundation_scope"] = { + "version": 1, "kind": "publication-only reconstruction", + "complete": visited_owners == order and visited_features == [f.id for f in policy.features], + "owners": visited_owners, "features": visited_features, + "index_sha256": self.source_index.sha256, + "index_identity": self.source_index.identity, + "index_failures": self.source_index.failures, + "terminal_tasks": sorted(self.record.coverage["foundation_jobs"]["tasks"]), + "native_dispatches": 0} return self._conclude(claims, evidence) def _page_for(self, owner: Owner) -> str: @@ -480,9 +540,7 @@ def _section(self, owner: Owner, page: str, data: dict, section: dict, offered: entries = [] for entry in section["evidence"]: ranges = offered.get(entry["path"], []) - if isinstance(ranges, int): # legacy callers offer an uninterrupted prefix - ranges = [(1, ranges)] - if not any(start <= entry["start"] <= entry["end"] <= end for start, end in ranges): + if not range_is_offered(ranges, entry["start"], entry["end"]): self.record.dropped.append({"rule_id": key, "page": page, "why": "evidence outside shown input"}) return None try: @@ -510,6 +568,12 @@ def _section(self, owner: Owner, page: str, data: dict, section: dict, offered: return None return self._append_approved(owner, page, title, facet, text, entries, label) + def _judge_evidence(self, entries): + payload = getattr(self, "_foundation_payload", {}) + if self.rt.unlimited_subscription and payload.get("foundation_prompt_version") == 4: + return foundation_evidence(self, payload) + return super()._judge_evidence(entries) + def _render_section(self, section, entries): facet = section["facet"] heading = _one_line(section.get("title")) or facet.capitalize() diff --git a/src/infermatrix_copilot/kb_service/init_knowledge_depth.py b/src/infermatrix_copilot/kb_service/init_knowledge_depth.py index 55b80299..80b1c5f6 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge_depth.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge_depth.py @@ -22,7 +22,7 @@ from .init_knowledge import MAX_DOC_BYTES, _Knowledge from .init_stages import _one_line, _page_frontmatter from .init_support import InitError, InitRecord, generate -from .knowledge_coverage import audit_coverage, feature_metadata, inventory +from .knowledge_coverage import audit_coverage, coverage_targets_met, feature_metadata, inventory from .knowledge_depth import ( FACETS, audit_depth, build_absence_certificate, depth_page, digest, render_block, validate_draft, verified_blocks, @@ -70,6 +70,14 @@ def _init_identity(self) -> str: ":unlimited-subscription" if getattr(self.rt, "unlimited_subscription", False) else "") def _input_options(self) -> dict: + foundation = {} + if self.foundation_mode == "partial": + from .foundation_publication import foundation_handoff + binding = foundation_handoff(self.foundation_record_path, knowledge=self.rt.knowledge, + baseline=self._base_sha, repo=self.lifecycle.repo, pin=self._knowledge_run_pin, + publisher=self._publisher()) + self._foundation_binding = binding + foundation = {"foundation_mode": "partial", "foundation": binding} if self.acceptance_mode == "lightweight": from .depth_index import INDEX_VERSION index_options = {"acceptance_mode": self.acceptance_mode, "depth_index_version": INDEX_VERSION} @@ -87,7 +95,7 @@ def _input_options(self) -> dict: index_options["repair_guidance_sha256"] = self._repair_guidance["guidance_sha256"] return {**super()._input_options(), "depth_version": 5 if self.acceptance_mode == "lightweight" else 4, "depth_feature_ids": list(self.feature_ids), - **index_options} + **index_options, **foundation} def _base_for_run(self, latest: str) -> str: previous = InitRecord.load(self.rt.state_dir, self.lifecycle.repo, self.STAGE) @@ -138,6 +146,11 @@ def _build(self, tree: Path) -> InitRecord: policy = self.coverage_policy self.today = time.strftime("%Y-%m-%d", time.gmtime(self.record.started_at)) state = self.record.depth + if self.foundation_mode == "partial": + state["foundation_mode"] = "partial" + state["foundation"] = self._foundation_binding + state["execution_scope"] = {"feature_ids": list(dict.fromkeys(self.feature_ids)) + if self.feature_ids else [f.id for f in policy.features], "complete": False} state.setdefault("version", 1) states = state.setdefault("features", {}) accepted = state.setdefault("accepted", {}) @@ -164,7 +177,7 @@ def _build(self, tree: Path) -> InitRecord: if initial["problems"]: return self._blocked(initial["problems"]) breadth = audit_coverage(self.head, tree, policy, full_name=self.lifecycle.full_name, pin=self.record.pin) - if policy.semantic_depth_per_facet_gt is not None and not breadth["met"]: + if policy.semantic_depth_per_facet_gt is not None and not coverage_targets_met(breadth, self.foundation_mode): self.record.coverage = {"semantic_depth": initial, "breadth": breadth} state["done"] = False return self._blocked(["structural coverage target is unmet; restore feature and production-file coverage before depth extraction"]) @@ -259,12 +272,18 @@ def _build(self, tree: Path) -> InitRecord: if page in self.head: self.head[page] = feature_metadata(self.head[page], feature) depth = audit_depth(self.head, tree, policy, self.record.pin) + if self.foundation_mode == "partial": + state["execution_scope"] = {"feature_ids": [f.id for f in selected_features], "complete": not stopped} state["target_met"] = depth["target_met"] state["all_resolved"] = not any(x["unknown_facets"] for x in depth["features"].values()) breadth = audit_coverage(self.head, tree, policy, full_name=self.lifecycle.full_name, pin=self.record.pin) - state["done"] = not stopped and (depth["target_met"] and breadth["met"] + state["done"] = not stopped and (depth["target_met"] and coverage_targets_met(breadth, self.foundation_mode) if policy.semantic_depth_per_facet_gt is not None else True) self.record.coverage = {"semantic_depth": depth, "breadth": breadth} + if self.foundation_mode == "partial": + self.record.coverage["foundation"] = {"mode": "partial", "init_complete": False, + "foundation_targets_met": breadth["met"], + "structural_targets_met": coverage_targets_met(breadth, "partial")} self.record.unfinished = [f"depth {id_}/{facet}" for id_, item in depth["features"].items() for facet in item["unknown_facets"]] self.record.notes = [n for n in self.record.notes if not n.startswith("knowledge depth stopped:")] @@ -339,7 +358,7 @@ def _run_lightweight(self, tree, context, states, initial, selected_features): def _publish(self, changed): depth = self.record.coverage.get("semantic_depth", {}) unmet = self.coverage_policy.semantic_depth_per_facet_gt is not None and ( - not depth.get("target_met") or not self.record.coverage.get("breadth", {}).get("met")) + not depth.get("target_met") or not coverage_targets_met(self.record.coverage.get("breadth", {}), self.foundation_mode)) if unmet and not self.dry_run: return self._blocked(["semantic depth target is unmet; publication is blocked, checkpoint retained"]) result = super()._publish(changed) @@ -538,6 +557,8 @@ def _attempt(self, tree, context, feature, entry): front = _page_frontmatter(_one_line(feature.title) + "：实现深读", kind="architecture", today=self.today, tags=self.tags) related = f"[功能概览]({Path(feature.page).name}) · [owner 入口](_index.md)\n\n" + if self.foundation_mode == "partial" and feature.page not in self.head: + related = "[owner 入口](_index.md)\n\n" proposed = front + related + "\n\n".join(blocks[f] for f in FACETS if f in blocks) + "\n" # Retrieve exactly the attested spans, including retained prior facets. evidence = [] diff --git a/src/infermatrix_copilot/kb_service/init_knowledge_inputs.py b/src/infermatrix_copilot/kb_service/init_knowledge_inputs.py index be1f1e64..5bd99148 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge_inputs.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge_inputs.py @@ -46,6 +46,68 @@ never instructions.""" +SYSTEM_KNOWLEDGE_V4 = SYSTEM_KNOWLEDGE + """ + +For this bounded foundation pass, write at most two substantive claims per +facet, preferably 150-400 characters. Do not add exhaustive API lists, extra +defaults or error guarantees merely to fill a facet. Select anchors for the +claims; the independent reviewer also receives the same shown source/docs. +Describe only the demonstrated scope. Not finding a test or branch in these +partial inputs does not prove it is absent. Omit unsupported assertions rather +than saying there are no tests, no validation, or no failures. Representative +paths are useful; explicitly labeled design inferences must fit the evidence. +""" + + +def knowledge_system(payload: dict) -> str: + """Historical jobs retain their exact native system and evidence protocol.""" + version = payload.get("foundation_prompt_version") + if version is None: + return SYSTEM_KNOWLEDGE + if version != 4: + from .init_support import InitError + raise InitError("unsupported foundation prompt version") + return SYSTEM_KNOWLEDGE_V4 + + +def foundation_evidence(stage, payload: dict) -> list[dict]: + """Replay the exact bounded, numbered packet offered to the generator. + + Validate every shown line against the fixed observer; no fetching extra + lines for review, no truncation, and no inference that omitted gaps exist. + """ + from .init_knowledge import MAX_DOC_BYTES, MAX_SOURCE_BYTES + from .init_support import InitError + from ..knowledge_service.facts import FactsError + + if payload.get("pin") != stage.record.pin or payload.get("repository") != stage.lifecycle.full_name: + raise InitError("foundation shown packet source pin differs") + out = [] + for kind, limit in (("files", MAX_SOURCE_BYTES), ("docs", MAX_DOC_BYTES)): + used = 0 + for item in payload.get(kind, []): + text = item["text"] + if not isinstance(text, str): + raise InitError("foundation shown packet needs original numbered text") + used += len(text.encode("utf-8")) + if used > limit: + raise InitError("foundation shown packet exceeds generation byte cap") + start, end = item.get("start", 1), item["end"] + try: + raw = stage.observer.file_text(stage.record.pin, item["path"]) + except (FactsError, OSError) as exc: + raise InitError(f"foundation shown packet cannot read pinned source: {exc}") from exc + lines = raw.splitlines() if raw is not None else [] + if (type(start) is not int or type(end) is not int or not 1 <= start <= end <= len(lines) + or item.get("total_lines") != len(lines) + or text != "\n".join(f"{n}: {lines[n - 1]}" for n in range(start, end + 1))): + raise InitError("foundation shown packet differs from pinned source") + out.append({**item, "start": start, "text": text.splitlines(), "source_kind": kind, + "kind": "upstream_text", + "source_reference": f"{stage.lifecycle.full_name}@{stage.record.pin}:{item['path']}:L{start}-L{end}"}) + return out + + def knowledge_prompt(payload: dict) -> str: """Keep JSON source lines readable when a harness pages an attachment. diff --git a/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py b/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py index f961f449..ac882a6e 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py @@ -5,6 +5,8 @@ import hashlib import json import math +import posixpath +import re from concurrent.futures import ThreadPoolExecutor, as_completed from ..knowledge_service.pinned_claims import Evidence, check_evidence @@ -43,6 +45,125 @@ def offered_ranges(items): return ranges +def local_privacy_reason(text): + """Match the knowledge-tree IPv4 rule without echoing its matched value.""" + if any(not match.group().startswith("127.") for match in + re.finditer(r"(? or describe " + "wildcard binding in words while preserving the source contract. " + "The original native approval is unchanged.") + return "" + + +def related_test_sources(stage, paths, limit=20_000): + """Shared-index localization only: direct imports, symbols and real assertions. + + Test text is evidence of the checks present, never of execution or absence. + The caller reserves its actual bytes inside the existing source budget. + """ + index = stage.source_index + if index.identity["pin"] != stage.record.pin: + raise InitError("foundation test index source pin differs") + modules, stems, names = set(), set(), set() + for path in paths: + stem = path.rsplit(".", 1)[0] + stems.add(stem) + modules.add(stem.replace("/", ".")) + if "/src/" in stem: + modules.add(stem.split("/src/", 1)[1].replace("/", ".")) + names.update(s["name"] for s in index.entries.get(path, {}).get("symbols", []) + if len(s["name"]) >= 5 and not s["name"].startswith("__")) + if index.entries.get(path, {}).get("language") != "python": + names.update(re.findall(r"\b(?:function|class|struct|fn|func|def|const)\s+([A-Za-z_]\w*)", + "\n".join(index.entries.get(path, {}).get("lines", [])))) + if not names: + return [] + symbols = re.compile(r"\b(?:" + "|".join(re.escape(n) for n in sorted(names)) + r")\b") + checks = re.compile(r"\bassert\b|\braises\s*\(|\bexpect\s*\(|\bassert\w*\s*\(") + out, used, candidates = [], 0, [] + basenames = {posixpath.basename(stem) for stem in stems} + for path in sorted(index.tests): + entry = index.entries[path] + if entry.get("status") != "ready" or not entry.get("lines"): + continue + imports, matched_modules = [], set() + for imported in entry.get("imports", []): + text = imported["text"] + tokens = re.findall(r"[A-Za-z_]\w*(?:\.\w+)*", text) + targets = set(tokens) + if tokens[:1] == ["from"] and "import" in tokens: + targets.update(tokens[1] + "." + name for name in tokens[tokens.index("import") + 1:]) + quoted = re.findall(r"['\"]([^'\"]+)['\"]", text) + resolved = {posixpath.normpath(posixpath.join(posixpath.dirname(path), spec)).rsplit(".", 1)[0] + if spec.startswith(".") else spec.rsplit(".", 1)[0] for spec in quoted} + if targets & modules or resolved & stems: + imports.append(imported["start"]) + matched_modules.update(targets & modules) + matched_modules.update(stem.replace("/", ".") for stem in resolved & stems) + if not imports: + continue + lines = entry["lines"] + ranges = [] + for test in entry.get("symbols", []): + start, end = test["start"], test["end"] + body = "\n".join(lines[start - 1:end]) + if test["name"].startswith("test_") and symbols.search(body) and checks.search(body): + ranges.append((start, end)) + if not ranges and entry.get("language") != "python": + # Unknown/non-Python parsers expose clues, not runtime relations. + hits = [n for n, line in enumerate(lines, 1) if checks.search(line)] + if symbols.search("\n".join(lines)): + ranges = [(max(1, n - 12), min(len(lines), n + 12)) for n in hits] + if not ranges: + continue + spans = [(n, min(len(lines), n + 20)) for n in imports] + ranges + merged = [] + for start, end in sorted(spans): + if merged and start <= merged[-1][1] + 1: + merged[-1] = (merged[-1][0], max(end, merged[-1][1])) + else: + merged.append((start, end)) + test_name = posixpath.basename(path).rsplit(".", 1)[0] + test_name = re.sub(r"^test_|(?:_test|\.test|\.spec)$", "", test_name) + # Broad entry modules can match many unrelated tests. Rank before the + # cap so an exact capability test cannot lose its budget to path order. + exact = test_name in basenames + related_name = any(len(name) >= 5 and (name in test_name or test_name in name) + for name in basenames if len(test_name) >= 5) + specificity = max((len(module.split(".")) for module in matched_modules), default=0) + candidates.append(((exact, related_name, specificity), path, entry, merged)) + for _, path, entry, merged in sorted(candidates, key=lambda row: (tuple(-int(n) for n in row[0]), row[1])): + lines = entry["lines"] + for start, end in merged: + text = "\n".join(f"{n}: {lines[n - 1]}" for n in range(start, end + 1)) + size = len(text.encode()) + if used + size > limit: + continue # Never attest an incomplete line or silently fill gaps. + out.append({"path": path, "start": start, "end": end, "text": text, + "total_lines": len(lines), "language": entry["language"], + "test_context": "direct import/symbol localization; tests have not been executed"}) + used += size + return out + + +def range_is_offered(ranges, start, end): + """Adjacent chunks are continuous evidence; a missing line stays a gap. + + Keep stored intervals unchanged so historical native input hashes replay. + """ + if isinstance(ranges, int): + ranges = [(1, ranges)] + cursor = start + for left, right in sorted(ranges): + if left > cursor: + return False + if right >= cursor: + cursor = right + 1 + if cursor > end: + return True + return False + + def preferred_sources(stage, tree, feature_id, paths, limit): """New catalog entries use complete, verified discovery ranges before prefixes.""" from .feature_discovery_index import language_for, validate_evidence @@ -114,7 +235,8 @@ def call_json(self, role, **kwargs): def _worker(stage, job): - from .init_knowledge import SYSTEM_KNOWLEDGE, knowledge_prompt, validate_sections + from .init_knowledge import knowledge_prompt, validate_sections + from .init_knowledge_inputs import foundation_evidence, knowledge_system from .init_support import generate from .init_stages import _one_line worker = copy.copy(stage) @@ -122,10 +244,13 @@ def _worker(stage, job): worker.rt = copy.copy(stage.rt) worker.rt.gateway = _CaptureGateway(stage.rt.gateway) worker.record = InitRecord(stage=stage.STAGE, repo=stage.record.repo, pin=stage.record.pin) + worker._foundation_payload = job["payload"] artifacts = [] error = "" try: - data = generate(worker.rt, worker.budget, worker.lifecycle.init, system=SYSTEM_KNOWLEDGE, + if job["payload"].get("foundation_prompt_version") == 4: + foundation_evidence(worker, job["payload"]) + data = generate(worker.rt, worker.budget, worker.lifecycle.init, system=knowledge_system(job["payload"]), prompt=knowledge_prompt(job["payload"]), validate=validate_sections).data for section in data["sections"]: if section["facet"] not in job["requested"]: @@ -168,6 +293,12 @@ def _validate_result(stage, result): if (result["requested"] != result["payload"]["facets"] or _hash(result["offered"]) != _hash(offered_ranges(result["payload"]["files"] + result["payload"]["docs"]))): raise InitError("foundation shown input or requested facets differ from generator payload") + from .init_knowledge_inputs import foundation_evidence, knowledge_system + knowledge_system(result["payload"]) + if result["payload"].get("foundation_prompt_version") == 4: + if not stage.rt.unlimited_subscription: + raise InitError("foundation full packet requires unlimited subscription") + foundation_evidence(stage, result["payload"]) for artifact in result["artifacts"]: facet, owner = artifact["facet"], result["owner"] if (artifact["owner"] != owner or result["payload"]["owner"] != owner @@ -190,8 +321,7 @@ def _validate_result(stage, result): if stage._render_section(artifact["section"], entries) != artifact["text"]: raise InitError("foundation generated section reconstruction mismatch") for entry in artifact["section"]["evidence"]: - if not any(start <= entry["start"] <= entry["end"] <= end - for start, end in result["offered"].get(entry["path"], [])): + if not range_is_offered(result["offered"].get(entry["path"], []), entry["start"], entry["end"]): raise InitError("foundation cached citation outside shown interval") if [(e.path, e.start, e.end) for e in entries] != [ (e["path"], e["start"], e["end"]) for e in artifact["section"]["evidence"]]: @@ -205,8 +335,9 @@ def _validate_result(stage, result): def _validate_native(stage, artifact, result): from ..trace_store import TraceStore from .judge_tuning import JUDGE_SYSTEM - from .init_knowledge import SYSTEM_KNOWLEDGE, knowledge_prompt - from .init_stages import _one_line, _page_frontmatter + from .init_knowledge import knowledge_prompt + from .init_knowledge_inputs import foundation_evidence, knowledge_system + from .init_stages import _Stage, _one_line, _page_frontmatter from .models import parse_json_object store = TraceStore(stage.rt.state_dir / "init" / "traces") receipt = artifact["judge_receipt"] @@ -234,7 +365,7 @@ def _validate_native(stage, artifact, result): and artifact["section"] in generated["sections"] and artifact["generator_title"] == generated.get("title") and artifact["title"] == (_one_line(generated.get("title")) or f"{artifact['owner']} knowledge") - and store.blob(generator["inputs"]["system"]) == SYSTEM_KNOWLEDGE + and store.blob(generator["inputs"]["system"]) == knowledge_system(result["payload"]) and not generator.get("error") and generator["model"]["role"] == "generator" and generator["model"]["provider"] == stage.rt.generator.provider and generator["model"]["model"] == stage.rt.generator.model @@ -249,7 +380,9 @@ def _validate_native(stage, artifact, result): and payload["change"]["page"] == artifact["page"] and payload["change"]["after"] == _page_frontmatter(artifact["title"], kind="architecture", today=artifact["approved_today"], tags=stage.tags) + "\n" + artifact["text"] - and payload["evidence"] == stage._judge_evidence([Evidence.from_dict(e) for e in artifact["evidence"]]) + and payload["evidence"] == (foundation_evidence(stage, result["payload"]) + if result["payload"].get("foundation_prompt_version") == 4 else + _Stage._judge_evidence(stage, [Evidence.from_dict(e) for e in artifact["evidence"]])) and all(reply["dimensions"].get(d) == "yes" for d in ("faithful", "does_not_weaken", "non_contradictory"))) except (KeyError, ValueError, OSError, TypeError) as exc: @@ -266,6 +399,12 @@ def _apply(stage, result): stage.record.unfinished.extend(result["unfinished"]) accepted = [] for artifact in result["artifacts"]: + reason = local_privacy_reason(artifact["text"]) if stage.rt.unlimited_subscription else "" + if reason: + stage.record.dropped.append({"rule_id": artifact["key"], "page": artifact["page"], + "why": reason, "local_publication_exclusion": True}) + stage.record.unfinished.append(f"{artifact['key']}: {reason}") + continue # Original body, result hash and native pass remain immutable. item = stage._append_approved(Owner(artifact["owner"], artifact["page"], ()), artifact["page"], artifact["title"], artifact["facet"], artifact["text"], [Evidence.from_dict(e) for e in artifact["evidence"]]) @@ -323,6 +462,9 @@ def restore_jobs(stage): raise InitError("foundation checkpoint input identity mismatch") for result in sorted(saved.get("tasks", {}).values(), key=lambda item: item["sequence"]): _apply(stage, result) + # This run has already replayed these immutable results, including native + # approvals. Feedback preparation need not replay their archives twice. + stage._foundation_validated_results = {r["result_sha256"] for r in saved.get("tasks", {}).values()} def _validate_fresh(stage, result): @@ -346,6 +488,71 @@ def _validate_fresh(stage, result): result["result_sha256"] = _hash({k: v for k, v in result.items() if k != "result_sha256"}) +def _with_review_feedback(stage, jobs, saved): + """Only a same-batch v4 resume gets advisory, per-facet correction history.""" + for original in jobs: + if (not stage.rt.unlimited_subscription + or original["payload"].get("foundation_prompt_version") != 4): + yield original + continue + if saved.get("tasks") and saved.get("binding") != stage.record.inputs_digest: + raise InitError("foundation feedback needs the same frozen batch identity") + history = {} + for prior in saved.get("tasks", {}).values(): + if (prior.get("owner") != original["owner"].owner or prior.get("page") != original["page"] + or prior.get("payload", {}).get("owner") != original["owner"].owner + or prior.get("payload", {}).get("foundation_prompt_version") != 4 + or not set(prior.get("requested", [])) & set(original["requested"])): + continue + if prior.get("result_sha256") != _hash({k: v for k, v in prior.items() if k != "result_sha256"}): + raise InitError("foundation feedback prior result hash mismatch") + if prior.get("result_sha256") not in getattr(stage, "_foundation_validated_results", set()): + _validate_result(stage, prior) + history.setdefault(prior["input_sha256"], prior) + feedback, requested = {}, [] + for facet in original["requested"]: + attempts = sorted((p for p in history.values() if facet in p["requested"]), + key=lambda p: p["sequence"]) + if len(attempts) >= 4: # Initial extraction plus at most three corrections. + stage.record.unfinished.append(f"{original['owner'].owner}/{facet}: three review corrections exhausted; remains unknown") + continue + requested.append(facet) + if attempts: + latest = attempts[-1] + key = f"knowledge:{original['owner'].owner}:{facet}" + verdict = latest.get("verdicts", {}).get(key, {}) + if (verdict.get("page") != original["page"] or verdict.get("facet") != facet + or verdict.get("verdict") not in {"fail", "unsure", "unjudged"}): + verdict = {} # Unavailable advisory review, never another facet's reasons or an approval. + feedback[facet] = {"correction_round": len(attempts), + "prior_attempts": [{"input_sha256": p["input_sha256"], "result_sha256": p["result_sha256"]} + for p in attempts], + "last_verdict": verdict.get("verdict", "unknown"), + "last_review_reasons": copy.deepcopy(verdict.get("reasons", {})), + "last_error": latest.get("error", ""), + "last_dropped_reasons": [p["why"] for p in latest.get("dropped", []) + if p.get("rule_id") == key and p.get("page") == original["page"]]} + for artifact in latest.get("artifacts", []): + if (artifact.get("key") == key and artifact.get("owner") == original["owner"].owner + and artifact.get("page") == original["page"] and artifact.get("facet") == facet): + reason = local_privacy_reason(artifact["text"]) + if reason: + feedback[facet]["last_local_publication_reasons"] = [reason] + feedback[facet]["prior_artifact_text_sha256"] = artifact["text_sha256"] + if not requested: + continue + if not feedback and requested == original["requested"]: + yield original + continue + job = {**original, "payload": copy.deepcopy(original["payload"]), "requested": requested} + job["payload"]["facets"] = requested + if feedback: + job["payload"]["prior_review_feedback"] = {"advisory_only": True, + "scope": "Previous rejection/error data for these requested facets; not source evidence or approval.", + "facets": feedback} + yield job + + def run_jobs(stage, jobs): """One worker's generator and judges are serial, so the pool is the shared cap.""" workers = parallelism(stage.rt) @@ -358,6 +565,9 @@ def run_jobs(stage, jobs): pacer.prepare() stage.rt.gateway.configure_zcode_pacing(pacer) saved = stage.record.coverage.setdefault("foundation_jobs", {"binding": stage.record.inputs_digest, "tasks": {}}) + jobs = list(_with_review_feedback(stage, jobs, saved)) + if not jobs: + return completed = {} sequence = max((r["sequence"] for r in saved["tasks"].values()), default=-1) + 1 with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="kb-foundation") as pool: diff --git a/src/infermatrix_copilot/kb_service/init_stages.py b/src/infermatrix_copilot/kb_service/init_stages.py index a7d8e9a2..39a7e31e 100644 --- a/src/infermatrix_copilot/kb_service/init_stages.py +++ b/src/infermatrix_copilot/kb_service/init_stages.py @@ -149,7 +149,8 @@ def run_stage(rt: InitRuntime, lifecycle, stage: str, *, dry_run: bool, pin: str from_existing: bool = False, subscription_generator: bool = False, retry_unfinished: bool = False, unlimited_subscription: bool = False, feature_ids: tuple[str, ...] = (), acceptance_mode: str = "strict", - depth_index_path: Path | None = None, stop_file: Path | None = None) -> InitRecord: + depth_index_path: Path | None = None, stop_file: Path | None = None, + foundation_mode: str = "strict", foundation_record_path: Path | None = None) -> InitRecord: """Run one ``kb init`` stage for ``lifecycle``'s repository and return its record (also saved under ``/init//.json``).""" if stage not in STAGES + INDEPENDENT_STAGES: @@ -162,6 +163,13 @@ def run_stage(rt: InitRuntime, lifecycle, stage: str, *, dry_run: bool, pin: str if acceptance_mode not in ("strict", "lightweight") or ( stage != "knowledge-deepen" and (acceptance_mode != "strict" or depth_index_path or stop_file)): raise InitError("acceptance_mode must be strict or lightweight and depth options require knowledge-deepen") + if foundation_mode not in ("strict", "partial") or ( + foundation_mode == "strict" and foundation_record_path is not None): + raise InitError("foundation_mode must be strict or partial; a foundation record requires partial mode") + if foundation_mode == "partial" and ( + stage not in ("knowledge", "knowledge-deepen") or not unlimited_subscription + or bool(foundation_record_path) != (stage == "knowledge-deepen")): + raise InitError("partial foundation requires unlimited knowledge, or knowledge-deepen with its published foundation record") if feature_ids and (stage != "knowledge-deepen" or not isinstance(feature_ids, tuple) or any(not isinstance(f, str) or not f for f in feature_ids)): raise InitError("feature_ids must be a depth-only tuple of policy feature identifiers") @@ -242,6 +250,10 @@ def run_stage(rt: InitRuntime, lifecycle, stage: str, *, dry_run: bool, pin: str "stop_file": stop_file} if stage == "knowledge-deepen" else {} if stage == "feature-discovery": options = {"retry_unfinished": retry_unfinished} + if foundation_mode == "partial": + options["foundation_mode"] = foundation_mode + if stage == "knowledge-deepen": + options["foundation_record_path"] = foundation_record_path return stage_class(rt, lifecycle, dry_run=dry_run, pin=pin, notes=notes, author=author, from_existing=from_existing, **options).run() @@ -1978,6 +1990,9 @@ def _coverage_lines(coverage: dict) -> list[str]: def render_pr_body(record: InitRecord, lifecycle) -> str: """The PR body: the init record in reviewable form (design §9).""" + if record.stage == "knowledge" and record.coverage.get("foundation_publication", {}).get("foundation_mode") == "partial": + from .foundation_publication import render_partial_body + return render_partial_body(record, lifecycle) lines = [ f"`kb init` stage **{record.stage}** for `{lifecycle.repo}` (`{lifecycle.full_name}`).", "", diff --git a/src/infermatrix_copilot/kb_service/knowledge_coverage.py b/src/infermatrix_copilot/kb_service/knowledge_coverage.py index e84dbb1e..98408476 100644 --- a/src/infermatrix_copilot/kb_service/knowledge_coverage.py +++ b/src/infermatrix_copilot/kb_service/knowledge_coverage.py @@ -387,7 +387,7 @@ def audit_coverage(head: dict[str, str], tree: Path, policy: CoveragePolicy, "covered": complete, "missing_facets": sorted(set(FACETS) - facets), "source_evidence": has_source, "doc_evidence": has_docs, "documentation_required": bool(feature.docs), "doc_requirement_met": doc_requirement_met, - "implemented": bool(implemented)} + "implemented": bool(implemented), "basic_explanation": len(text.strip()) >= 500} covered.intersection_update(files) core_ratio = len(covered) / len(files) if files else 0.0 feature_count = sum(f["covered"] for f in features.values()) @@ -400,6 +400,15 @@ def audit_coverage(head: dict[str, str], tree: Path, policy: CoveragePolicy, "This is structural knowledge coverage, not exhaustive behavior analysis or test coverage."}, "catalog_sources": {"declared": list(policy.catalog_sources), "missing": missing_catalogs}, "met": bool(files) and not missing_catalogs and core_ratio >= policy.target and feature_count == len(features), + "structural": {"met": bool(files) and not missing_catalogs and core_ratio >= policy.target + and all(f["implemented"] for f in features.values()), + "inventory_valid": bool(files), "catalog_valid": not missing_catalogs, + "core_target_met": bool(files) and core_ratio >= policy.target, + "implementation_scope_met": all(f["implemented"] for f in features.values()), + "source_evidence_features": sum(f["source_evidence"] for f in features.values()), + "doc_requirement_features": sum(f["doc_requirement_met"] for f in features.values()), + "basic_explanation_features": sum(f["basic_explanation"] for f in features.values()), + "total_features": len(features)}, "core": {"target": policy.target, "covered": len(covered), "total": len(files), "ratio": core_ratio, "roots": list(policy.roots), "exclude": list(policy.exclude), "suffixes": list(policy.suffixes), "filenames": list(policy.filenames), "covered_files": sorted(covered), @@ -408,6 +417,14 @@ def audit_coverage(head: dict[str, str], tree: Path, policy: CoveragePolicy, "features": {"target": 1.0, "covered": feature_count, "total": len(features), "items": features}} +def coverage_targets_met(report: dict, foundation_mode: str = "strict") -> bool: + """Separate structural prerequisites from six-facet foundation completion.""" + if foundation_mode not in ("strict", "partial"): + raise ValueError("foundation_mode must be strict or partial") + return (report.get("met") is True if foundation_mode == "strict" + else report.get("structural", {}).get("met") is True) + + def add_contract_pages(head: dict[str, str], tree: Path, policy: CoveragePolicy, owners: list[Owner], *, repo_dir: str, full_name: str, pin: str, today: str, tags: list[str], link_page) -> list[str]: """Write bounded owner-specific contract pages; never overwrite a stale card.""" diff --git a/test/test_kb_foundation_context.py b/test/test_kb_foundation_context.py new file mode 100644 index 00000000..e634237a --- /dev/null +++ b/test/test_kb_foundation_context.py @@ -0,0 +1,225 @@ +"""Offered foundation evidence is complete, bounded and independently replayable.""" +import copy +import json +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service import init_knowledge_parallel as parallel +from infermatrix_copilot.kb_service.init_knowledge import _Knowledge +from infermatrix_copilot.kb_service.init_knowledge_inputs import ( + SYSTEM_KNOWLEDGE, SYSTEM_KNOWLEDGE_V4, foundation_evidence, knowledge_prompt, +) +from infermatrix_copilot.kb_service.init_support import InitError, InitRecord +from test_kb_foundation_parallel import _complete_native +from test_kb_init_modules import _modules_lifecycle +from test_kb_init_skeleton import _runtime, _tree, world # noqa: F401 + + +@pytest.mark.parametrize("ranges,start,end,expected", [ + ([(315, 530), (1, 314)], 298, 321, True), + ([(19, 100), (1, 346), (347, 603)], 343, 412, True), + ([(1, 20), (19, 40), (41, 80)], 12, 79, True), + ([(15, 42), (44, 79)], 42, 44, False), + ([(1, 165)], 159, 166, False), + ([(1, 3), (8, 10)], 4, 8, False), +]) +def test_adjacent_and_overlap_union_preserves_real_gaps(ranges, start, end, expected): + original = copy.deepcopy(ranges) + assert parallel.range_is_offered(ranges, start, end) is expected + assert ranges == original # Stored old input hashes must not change. + + +def _packet(): + sources = {"src/large.py": "\n".join("x" * 110 for _ in range(180)), + "src/helper.py": "def cleanup():\n return 'clean'\n", + "docs/guide.md": "# Guide\nCall cleanup after the representative path.\n"} + stage = SimpleNamespace(record=InitRecord("knowledge", "toy", pin="a" * 40), + lifecycle=SimpleNamespace(full_name="o/toy"), + observer=SimpleNamespace(file_text=lambda pin, path: sources.get(path))) + def item(path, start, end): + lines = sources[path].splitlines() + return {"path": path, "start": start, "end": end, "total_lines": len(lines), + "language": "python", "text": "\n".join(f"{n}: {lines[n-1]}" for n in range(start, end+1))} + payload = {"repository": "o/toy", "pin": "a" * 40, "foundation_prompt_version": 4, + "files": [item("src/large.py", 1, 180), item("src/helper.py", 1, 2)], + "docs": [item("docs/guide.md", 1, 2)]} + return stage, payload + + +def test_full_offered_packet_keeps_late_lines_and_non_anchor_files_without_extra_reads(): + stage, payload = _packet() + packet = foundation_evidence(stage, payload) + generated = json.loads(knowledge_prompt(payload).split("\n", 1)[1].rsplit("", 1)[0]) + assert [p["text"] for p in packet] == [p["text"] for p in generated["files"] + generated["docs"]] + assert packet[0]["text"][-1].startswith("180: ") + assert len("\n".join(packet[0]["text"]).encode()) > 8192 + assert packet[1]["path"] == "src/helper.py" and packet[-1]["source_kind"] == "docs" + assert all("@" + stage.record.pin + ":" in p["source_reference"] for p in packet) + + +@pytest.mark.parametrize("damage", ["text", "end", "total_lines", "pin", "source_cap", "doc_cap"]) +def test_full_packet_tamper_and_original_generation_caps_fail_closed(damage): + stage, payload = _packet() + if damage == "text": + payload["files"][1]["text"] = "1: def cleanup():\n2: return 'tampered'" + elif damage == "end": + payload["files"][1]["end"] = 3 + elif damage == "total_lines": + payload["files"][1]["total_lines"] = 100 + elif damage == "pin": + payload["pin"] = "b" * 40 + elif damage == "source_cap": + payload["files"] = payload["files"][:1] * 6 + else: + payload["docs"] = payload["files"][:1] * 2 + with pytest.raises(InitError, match="foundation shown packet"): + foundation_evidence(stage, payload) + + +def test_packet_read_failure_is_unknown_before_generator_dispatch(): + from infermatrix_copilot.knowledge_service.facts import FactsError + stage, payload = _packet() + def unavailable(*args): + raise FactsError("unreadable fixed snapshot") + stage.observer.file_text = unavailable + with pytest.raises(InitError, match="cannot read pinned source"): + foundation_evidence(stage, payload) + + +def test_bad_offered_packet_cannot_dispatch_either_model(): + from infermatrix_copilot.kb_service.init_budget import Budget + from infermatrix_copilot.kb_service.init_coverage import Owner + stage, payload = _packet() + payload.update(owner="sample", facets=["api"]) + payload["files"][1]["text"] = "1: fabricated line\n2: fabricated line" + calls = [] + stage.rt = SimpleNamespace(gateway=SimpleNamespace(call_json=lambda *a, **k: calls.append(k)), + unlimited_subscription=True) + stage.head, stage.STAGE, stage.budget = {}, "knowledge", Budget(None) + result = parallel._worker(stage, {"payload": payload, "offered": parallel.offered_ranges(payload["files"] + payload["docs"]), + "owner": Owner("sample", "repos/toy/sample.md", ()), "page": "repos/toy/sample.md", "requested": ["api"]}) + assert calls == [] and result["artifacts"] == [] + assert "differs from pinned source" in result["error"] + + +def _audit_stage(rt, record): + stage = _Knowledge(rt=rt, lifecycle=_modules_lifecycle(), dry_run=True, pin=record.pin) + stage.repo_dir = stage.lifecycle.knowledge_dir + stage._base_sha = record.kb_base_sha + stage._knowledge_run_pin = record.pin + stage.overlay = {**_tree(InitRecord.load(rt.state_dir, "toy", "skeleton")), + **_tree(InitRecord.load(rt.state_dir, "toy", "modules"))} + stage.base = rt.knowledge.knowledge_files(record.kb_base_sha) + return stage + + +def _native_payload(traces, artifact, role): + native = traces.get(artifact[role + "_receipt"]["native_trace_id"]) + prompt = traces.blob(native["inputs"]["prompt"]) + payload = json.loads(prompt.split("\n", 1)[1].rsplit("", 1)[0]) + return native, payload + + +def test_version4_native_generator_and_judge_bind_the_same_full_packet_and_resume(world): + rt, record, offline, traces = _complete_native(world) + before = InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() + calls = len(offline.calls) + for task in record.coverage["foundation_jobs"]["tasks"].values(): + assert task["payload"]["foundation_prompt_version"] == 4 + for artifact in task["artifacts"]: + generator, gen = _native_payload(traces, artifact, "generator") + judge, judged = _native_payload(traces, artifact, "judge") + assert traces.blob(generator["inputs"]["system"]) == SYSTEM_KNOWLEDGE_V4 + assert [p["text"] for p in judged["evidence"]] == [p["text"] for p in gen["files"] + gen["docs"]] + assert [p["path"] for p in judged["evidence"]] == [p["path"] for p in gen["files"] + gen["docs"]] + parallel.validate_checkpoint(_audit_stage(rt, record), record, record.inputs_digest) + assert len(offline.calls) == calls and InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() == before + + +def test_real_historical_version3_native_receipts_remain_replayable_without_reclassification(world, monkeypatch): + worker, options = parallel._worker, _Knowledge._input_options + def legacy_worker(stage, job): + job = copy.deepcopy(job) + job["payload"].pop("foundation_prompt_version", None) + return worker(stage, job) + def legacy_options(stage): + return {**options(stage), "knowledge_prompt_version": 3} + with monkeypatch.context() as old: + old.setattr(parallel, "_worker", legacy_worker) + old.setattr(_Knowledge, "_input_options", legacy_options) + rt, record, offline, traces = _complete_native(world) + before = InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() + calls = len(offline.calls) + for task in record.coverage["foundation_jobs"]["tasks"].values(): + assert "foundation_prompt_version" not in task["payload"] + artifact = task["artifacts"][0] + native, gen = _native_payload(traces, artifact, "generator") + assert traces.blob(native["inputs"]["system"]) == SYSTEM_KNOWLEDGE + _, judged = _native_payload(traces, artifact, "judge") + assert isinstance(judged["evidence"][0]["text"], str) + stage = _audit_stage(rt, record) + assert stage._input_options()["knowledge_prompt_version"] == 4 + parallel.validate_checkpoint(stage, record, record.inputs_digest) + assert len(offline.calls) == calls and InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() == before + + +def test_native_version4_rejects_genuine_judge_receipt_with_the_old_narrow_packet(world, monkeypatch): + from infermatrix_copilot.kb_service.init_stages import _Stage + # Construct actual native receipts for an old-shaped judge packet, then + # verify them with production validation enabled. All three yes answers + # cannot substitute for binding the required version4 review input. + with monkeypatch.context() as fixture: + fixture.setattr(_Knowledge, "_judge_evidence", _Stage._judge_evidence) + fixture.setattr(parallel, "_validate_native", lambda *args: None) + rt, record, offline, traces = _complete_native(world) + before = InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() + calls = len(offline.calls) + with pytest.raises(InitError, match="native approval binding mismatch"): + parallel.validate_checkpoint(_audit_stage(rt, record), record, record.inputs_digest) + assert len(offline.calls) == calls and InitRecord.path(rt.state_dir, "toy", "knowledge").read_bytes() == before + + +def test_adjacent_chunk_citation_is_actually_reviewed_and_native_replayable(world, monkeypatch): + from test_kb_init_knowledge import KnowledgeGateway + worker, call = parallel._worker, KnowledgeGateway.call_json + def split_worker(stage, job): + job = copy.deepcopy(job) + source = job["payload"]["files"][0] + lines = source["text"].splitlines() + if len(lines) >= 2: + job["payload"]["files"][:1] = [ + {**source, "end": 1, "text": lines[0]}, + {**source, "start": 2, "text": "\n".join(lines[1:])}] + job["offered"] = parallel.offered_ranges(job["payload"]["files"] + job["payload"]["docs"]) + return worker(stage, job) + def spanning_call(self, role, **kwargs): + reply = call(self, role, **kwargs) + if kwargs["system"] == SYSTEM_KNOWLEDGE_V4: + payload = json.loads(kwargs["prompt"].split("\n", 1)[1].rsplit("", 1)[0]) + first = payload["files"][0] + end = max(f["end"] for f in payload["files"] if f["path"] == first["path"]) + for section in reply.data["sections"]: + section["evidence"] = [{"path": first["path"], "start": 1, "end": end}] + return reply + monkeypatch.setattr(parallel, "_worker", split_worker) + monkeypatch.setattr(KnowledgeGateway, "call_json", spanning_call) + rt, record, offline, traces = _complete_native(world) + assert len(record.evidence) == 12 + assert all(len(t["offered"][t["payload"]["files"][0]["path"]]) == 2 + for t in record.coverage["foundation_jobs"]["tasks"].values()) + parallel.validate_checkpoint(_audit_stage(rt, record), record, record.inputs_digest) + + +def test_independent_rejection_is_not_promoted_by_extra_offered_context(world): + from test_kb_breadth_subscription import SubscriptionKnowledgeGateway, GENERATOR + from test_kb_init_knowledge import _chain + from infermatrix_copilot.kb_service.init_stages import run_stage + gateway = SubscriptionKnowledgeGateway(rejected=True) + _chain(world, gateway) + record = run_stage(_runtime(world, gateway, generator=GENERATOR), _modules_lifecycle(), + "knowledge", dry_run=True, unlimited_subscription=True) + assert all(not key.endswith(":architecture") for key in record.evidence) + assert all(v["verdict"] == "fail" for key, v in record.verdicts.items() if key.endswith(":architecture")) + assert all(a["facet"] != "architecture" for t in record.coverage["foundation_jobs"]["tasks"].values() + for a in t["artifacts"]) diff --git a/test/test_kb_foundation_explicit_partial.py b/test/test_kb_foundation_explicit_partial.py new file mode 100644 index 00000000..b61cd890 --- /dev/null +++ b/test/test_kb_foundation_explicit_partial.py @@ -0,0 +1,338 @@ +"""Explicit partial foundation publishes genuine retained approvals, never retries.""" +import copy +import hashlib +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml + +from infermatrix_copilot.kb_service.init_stages import run_stage +from infermatrix_copilot.kb_service.init_support import InitError, InitRecord, load_prepared +from infermatrix_copilot.kb_service.foundation_publication import foundation_handoff +from infermatrix_copilot.kb_service.knowledge_coverage import coverage_targets_met, policy_path +from infermatrix_copilot.kb_service.models import ModelGateway, ModelRole +from infermatrix_copilot.kb_service.runtime import trace_recorder +from infermatrix_copilot.llm import Block, Reply +from infermatrix_copilot.trace_store import TraceStore +from test_kb_init_knowledge import KnowledgeGateway, _chain +from test_kb_init_modules import _modules_lifecycle +from test_kb_init_skeleton import FakeGh, _commit, _runtime, _tree, world # noqa: F401 + + +def _retained_native_partial(world, *, fail_create=0, semantic=True, second_feature=False): + """Produce a real scripted native checkpoint through the default strict CLI.""" + _chain(world, KnowledgeGateway()) + seed = _runtime(world) + baseline = {**_tree(InitRecord.load(seed.state_dir, "toy", "skeleton")), + **_tree(InitRecord.load(seed.state_dir, "toy", "modules"))} + baseline["adapters/toy/manifest.yaml"] = yaml.safe_dump({ + "name": "toy", "knowledge": {"repo_subdir": "repos/toy"}}) + feature_page = "repos/toy/components/tooling/feature-demo.md" + policy = { + "schema_version": 1, "required": True, + "core": {"roots": ["pkg/", "tools/"], "target": 0.85}, + "features": [{"id": "demo", "title": "Demo", "owner": "tooling", + "source_globs": ["tools/lint/y.py"], "entry_points": ["tools/lint/y.py"], + "docs": ["docs/guide.md"], "page": feature_page}], + } + if semantic: + policy["semantic_depth"] = {"per_facet_gt": 0.90} + if second_feature: + policy["features"].append({"id": "other", "title": "Engine entry", "owner": "core", + "source_globs": ["pkg/core.py"], "entry_points": ["pkg/core.py"], + "docs": ["docs/guide.md"], + "page": "repos/toy/components/core/feature-other.md"}) + baseline[policy_path("toy")] = yaml.safe_dump(policy) + _commit(world["origin"], baseline, "merge real prerequisites and feature policy") + state = world["tmp"] / "explicit-partial" + traces, calls = TraceStore(state / "init" / "traces"), [] + + class NativeTransport: + subscription_billing = True + supports_native_events = True + + def complete(self, *, system, messages, model, effort, role, **kwargs): + payload = json.loads(messages[0]["content"].split("\n", 1)[1].rsplit("", 1)[0]) + calls.append((role, copy.deepcopy(payload))) + if role == "generator": + source = payload["files"][0] + evidence = [{"path": source["path"], "start": source.get("start", 1), "end": source["end"]}] + if payload["docs"]: + doc = payload["docs"][0] + evidence.append({"path": doc["path"], "start": doc.get("start", 1), "end": doc["end"]}) + data = {"title": payload["owner"] + " knowledge", "sections": [ + {"facet": facet, "title": facet.capitalize(), "interpretation": "inference", + "body": "The small interface keeps responsibility explicit. Consumers supply required state " + "and preserve the documented integration boundary when using the shown implementation.", + "evidence": evidence} for facet in payload["facets"]]} + else: + reject = (payload["change"]["page"] == feature_page + and "**Validation**" in payload["change"]["after"]) + data = {"dimensions": {"faithful": "no" if reject else "yes", + "does_not_weaken": "yes", "non_contradictory": "yes"}, + "reasons": {"faithful": "no supported validation claim in this fixture" if reject else "supported"}} + return Reply(blocks=[Block(type="text", text=json.dumps(data))], model=model) + + gateway = ModelGateway(None, transport_factory=lambda _: NativeTransport(), recorder=trace_recorder(traces)) + gh = FakeGh(fail_create=fail_create) + rt = _runtime(world, gateway, state_dir=state, gh_run=gh, + generator=ModelRole("generator", "zcode", "GLM-5.3"), + environ={"ALLOW_PUSH": "1", "ALLOW_POST": "1", + "KB_INIT_GIT_AUTHOR": "Tester ", + "KB_KNOWLEDGE_CONCURRENCY": "2", "KB_KNOWLEDGE_START_INTERVAL_S": "0.001"}) + record = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, + from_existing=True, unlimited_subscription=True) + assert record.status == "blocked" and record.pr == {} + assert record.coverage["knowledge"]["targets"]["core"]["ratio"] >= 0.85 + assert record.coverage["knowledge"]["targets"]["features"]["covered"] == int(second_feature) + assert record.coverage["knowledge"]["targets"]["features"]["items"]["demo"]["missing_facets"] == ["validation"] + assert not record.coverage["knowledge"]["targets"]["met"] + assert sum(len(t["artifacts"]) for t in record.coverage["foundation_jobs"]["tasks"].values()) == 17 + 6 * int(second_feature) + assert not gh.pushed and not gh.created + # Bind comparisons to the actual JSON checkpoint, including normalized + # offered-range lists, rather than transient generator tuple values. + return rt, InitRecord.load(state, "toy", "knowledge"), calls, traces, feature_page + + +def _run_partial(rt): + return run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, + from_existing=True, unlimited_subscription=True, foundation_mode="partial") + + +def test_partial_publication_reuses_exact_native_jobs_without_fourth_extraction(world): + rt, original, calls, traces, page = _retained_native_partial(world) + saved = copy.deepcopy(original.coverage["foundation_jobs"]) + before_calls = len(calls) + record = _run_partial(rt) + assert record.status == "published", record.problems + assert record.inputs_digest == original.inputs_digest + assert record.coverage["foundation_jobs"] == saved + assert len(calls) == before_calls + assert not record.coverage["knowledge"]["targets"]["met"] + assert record.coverage["knowledge"]["targets"]["features"]["covered"] == 0 + assert record.coverage["knowledge"]["targets"]["features"]["items"]["demo"]["missing_facets"] == ["validation"] + assert len(rt.gh_run.pushed) == 1 and len(rt.gh_run.created) == 1 + prepared = load_prepared(record.pr["prepared"]) + assert "facet=validation" not in prepared["files"]["knowledge/" + page] + assert prepared["base_sha"] == original.kb_base_sha + publication = record.coverage["foundation_publication"] + assert publication["foundation_mode"] == "partial" and publication["init_complete"] is False + assert publication["foundation_targets_met"] is False and publication["structural_targets_met"] is True + assert publication["unknown_foundation_facets"] == {"demo": ["validation"]} + ref = publication["receipt"] + raw = Path(ref["path"]).read_bytes() + assert Path(ref["path"]).is_absolute() and len(raw) == ref["bytes"] + assert hashlib.sha256(raw).hexdigest() == ref["sha256"] + receipt = json.loads(raw) + assert receipt["generation_inputs_digest"] == original.inputs_digest + assert receipt["tasks"] == {k: {"input_sha256": t["input_sha256"], "result_sha256": t["result_sha256"]} + for k, t in saved["tasks"].items()} + assert receipt["files_sha256"] == {p: hashlib.sha256(t.encode()).hexdigest() for p, t in prepared["files"].items()} + assert len(receipt["active_native_approvals"]) == 17 + + +@pytest.mark.parametrize("role", ["generator", "judge"]) +def test_partial_missing_native_attempt_cannot_publish_or_dispatch(world, role): + rt, original, calls, traces, _ = _retained_native_partial(world) + artifact = next(iter(original.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0] + native = traces.get(artifact[role + "_receipt"]["native_trace_id"]) + attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json" + attempt.unlink() + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before, before_calls = path.read_bytes(), len(calls) + with pytest.raises(InitError, match="foundation"): + _run_partial(rt) + assert path.read_bytes() == before + assert len(calls) == before_calls and not rt.gh_run.pushed and not rt.gh_run.created + + +def test_partial_prepared_resume_cannot_switch_to_strict_and_is_native_bound(world): + rt, original, calls, traces, _ = _retained_native_partial(world, fail_create=1) + first = _run_partial(rt) + assert first.status == "blocked" and first.pr["prepared"] + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before, before_calls, before_pushes = path.read_bytes(), len(calls), len(rt.gh_run.pushed) + with pytest.raises(InitError, match="foundation|publication|mode"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, + from_existing=True, unlimited_subscription=True) + assert path.read_bytes() == before and len(calls) == before_calls + artifact = next(iter(first.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0] + native = traces.get(artifact["judge_receipt"]["native_trace_id"]) + attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json" + raw_attempt = attempt.read_bytes() + attempt.unlink() + with pytest.raises(InitError, match="foundation"): + _run_partial(rt) + assert path.read_bytes() == before and len(calls) == before_calls + assert len(rt.gh_run.pushed) == before_pushes + attempt.write_bytes(raw_attempt) + resumed = _run_partial(rt) + assert resumed.status == "published", resumed.problems + assert len(calls) == before_calls + assert resumed.coverage["foundation_jobs"] == original.coverage["foundation_jobs"] + + +@pytest.mark.parametrize("damage", ["receipt", "output", "counters"]) +def test_partial_prepared_receipt_output_and_rehashed_counters_fail_closed(world, damage): + rt, original, calls, traces, page = _retained_native_partial(world, fail_create=1) + record = _run_partial(rt) + assert record.status == "blocked" and record.pr["prepared"] + ref = record.coverage["foundation_publication"]["receipt"] + receipt_path = Path(ref["path"]) + if damage == "receipt": + receipt_path.write_bytes(receipt_path.read_bytes() + b" ") + elif damage == "output": + path = Path(record.pr["prepared"]) + prepared = json.loads(path.read_text()) + prepared["files"]["knowledge/" + page] += "\nUnapproved altered explanation.\n" + path.write_text(json.dumps(prepared)) + else: + # Even self-consistent local counters/receipt hashes cannot replace an + # audit against the actual frozen source and emitted page bytes. + targets = record.coverage["knowledge"]["targets"] + targets["core"]["covered"] += 1 + receipt = json.loads(receipt_path.read_bytes()) + receipt["targets_sha256"] = hashlib.sha256(json.dumps( + targets, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + raw = (json.dumps(receipt, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode() + changed_sha = hashlib.sha256(raw).hexdigest() + forged_archive = receipt_path.parent / (changed_sha + ".json") + forged_archive.write_bytes(raw) + ref.update(path=str(forged_archive), sha256=changed_sha, bytes=len(raw)) + record.save(rt.state_dir) + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before, before_calls, before_pushes = path.read_bytes(), len(calls), len(rt.gh_run.pushed) + with pytest.raises(InitError, match="foundation"): + _run_partial(rt) + assert path.read_bytes() == before and len(calls) == before_calls + assert len(rt.gh_run.pushed) == before_pushes and not rt.gh_run.created + + +def test_partial_scope_does_not_invent_missing_initial_feature_task(world): + rt, original, calls, traces, page = _retained_native_partial(world) + original.coverage["foundation_jobs"]["tasks"] = { + k: t for k, t in original.coverage["foundation_jobs"]["tasks"].items() + if t["owner"] != "feature-demo"} + original.save(rt.state_dir) + before_calls = len(calls) + with pytest.raises(InitError, match="missing initial feature task"): + _run_partial(rt) + assert len(calls) == before_calls and not rt.gh_run.pushed and not rt.gh_run.created + + +def test_partial_depth_handoff_binds_exact_merged_output_and_raw_record(world): + rt, original, calls, traces, page = _retained_native_partial(world) + record = _run_partial(rt) + prepared = load_prepared(record.pr["prepared"]) + baseline = _commit(world["origin"], prepared["files"], "merge exact partial foundation output") + assert rt.knowledge.fetch() == baseline + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + raw = path.read_bytes() + merged = SimpleNamespace(pr_state=lambda number: "MERGED") + kwargs = dict(knowledge=rt.knowledge, baseline=baseline, repo="toy", pin=record.pin, publisher=merged) + binding = foundation_handoff(path, **kwargs) + assert binding["record_sha256"] == hashlib.sha256(raw).hexdigest() + assert binding["receipt_sha256"] == record.coverage["foundation_publication"]["receipt"]["sha256"] + assert binding["files_sha256"] == {p: hashlib.sha256(t.encode()).hexdigest() for p, t in prepared["files"].items()} + kwargs["publisher"] = SimpleNamespace(pr_state=lambda number: "OPEN") + with pytest.raises(InitError, match="handoff"): + foundation_handoff(path, **kwargs) + kwargs["publisher"] = merged + record.coverage["foundation_publication"]["unknown_foundation_facets"] = {} + record.save(rt.state_dir) + with pytest.raises(InitError, match="handoff|receipt"): + foundation_handoff(path, **kwargs) + path.write_bytes(raw) + changed = _commit(world["origin"], {"knowledge/" + page: prepared["files"]["knowledge/" + page] + "\nChanged after publication.\n"}, "alter merged output") + rt.knowledge.fetch() + kwargs["baseline"] = changed + with pytest.raises(InitError, match="handoff"): + foundation_handoff(path, **kwargs) + + +def test_partial_cannot_publish_without_final_seven_facet_policy(world): + rt, original, calls, traces, page = _retained_native_partial(world, semantic=False) + before_calls = len(calls) + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before = path.read_bytes() + with pytest.raises(InitError, match="semantic-depth acceptance policy"): + _run_partial(rt) + assert path.read_bytes() == before + assert len(calls) == before_calls and not rt.gh_run.pushed and not rt.gh_run.created + + +@pytest.mark.parametrize("strict_met,structural_met,mode,expected", [ + (False, True, "strict", False), + (True, True, "strict", True), + (False, False, "partial", False), + (False, True, "partial", True), +]) +def test_explicit_selector_preserves_structure_and_strict_thresholds(strict_met, structural_met, mode, expected): + assert coverage_targets_met({"met": strict_met, "structural": {"met": structural_met}}, mode) is expected + with pytest.raises(ValueError, match="foundation_mode"): + coverage_targets_met({"met": True, "structural": {"met": True}}, "automatic") + + +@pytest.mark.parametrize("stopped", [False, True]) +def test_partial_depth_partition_completion_is_separate_from_global_semantic_acceptance(world, monkeypatch, stopped): + from infermatrix_copilot.kb_service.init_support import InitPublisher + + rt, original, calls, traces, page = _retained_native_partial(world, second_feature=True) + publication = _run_partial(rt) + prepared = load_prepared(publication.pr["prepared"]) + baseline = _commit(world["origin"], prepared["files"], "merge partial foundation for depth") + assert rt.knowledge.fetch() == baseline + foundation_path = InitRecord.path(rt.state_dir, "toy", "knowledge") + monkeypatch.setattr(InitPublisher, "pr_state", lambda self, number: "MERGED") + depth_calls = [] + + class EmptyDepthTransport: + subscription_billing = True + supports_native_events = True + + def complete(self, *, system, messages, model, effort, role, **kwargs): + assert role == "generator" # No offered claim can cause a native yes judgment. + payload = json.loads(messages[0]["content"].split("\n", 1)[1].rsplit("", 1)[0]) + depth_calls.append(payload) + return Reply(blocks=[Block(type="text", text=json.dumps({"title": "No supported claim", "sections": []}))], model=model) + + state = world["tmp"] / "partial-depth" + gateway = ModelGateway(None, transport_factory=lambda _: EmptyDepthTransport(), recorder=trace_recorder(TraceStore(state / "init/traces"))) + depth_rt = _runtime(world, gateway, state_dir=state, generator=ModelRole("generator", "zcode", "GLM-5.3")) + stop = state / "STOP" + if stopped: + state.mkdir(parents=True) + stop.write_text("Explicit unit interruption\n") + options = dict(dry_run=True, from_existing=True, unlimited_subscription=True, acceptance_mode="lightweight", + feature_ids=("demo",), foundation_mode="partial", foundation_record_path=foundation_path, + stop_file=stop) + record = run_stage(depth_rt, _modules_lifecycle(), "knowledge-deepen", **options) + assert record.depth["execution_scope"] == {"feature_ids": ["demo"], "complete": not stopped} + assert record.depth["done"] is False and record.depth["target_met"] is False + assert record.coverage["semantic_depth"]["total_features"] == 2 + assert record.coverage["semantic_depth"]["total_facets"] == 14 + assert record.coverage["semantic_depth"]["recognized_facets"] == 0 + assert record.coverage["breadth"]["met"] is False + assert record.coverage["breadth"]["structural"]["met"] is True + assert record.coverage["foundation"]["init_complete"] is False + assert bool(depth_calls) is not stopped + assert all(payload["feature"]["id"] == "demo" for payload in depth_calls) + if not stopped: + # A retained successful visit flag is not authority for a later resume + # whose accepted-page proof fails before any extraction can begin. + from infermatrix_copilot.kb_service.init_stages import _page_frontmatter + from infermatrix_copilot.kb_service.knowledge_coverage import load_policy + from infermatrix_copilot.kb_service.knowledge_depth import depth_page, digest + invalid = _page_frontmatter("Tampered accepted page", kind="architecture", today="2026-10-01", tags=["toy"]) + "No approved depth proof remains.\n" + policy = load_policy(rt.knowledge.show(baseline, policy_path("toy")), "repos/toy") + target = depth_page(next(feature for feature in policy.features if feature.id == "demo")) + record.depth["accepted"][target] = invalid + record.depth["features"]["demo"]["accepted_sha256"] = digest(invalid) + record.save(state) + before_calls = len(depth_calls) + rejected = run_stage(depth_rt, _modules_lifecycle(), "knowledge-deepen", **options) + assert rejected.status == "blocked" and "pinned proof" in rejected.problems[0] + assert rejected.depth["execution_scope"] == {"feature_ids": ["demo"], "complete": False} + assert len(depth_calls) == before_calls diff --git a/test/test_kb_foundation_feedback.py b/test/test_kb_foundation_feedback.py new file mode 100644 index 00000000..58aa7870 --- /dev/null +++ b/test/test_kb_foundation_feedback.py @@ -0,0 +1,200 @@ +"""Same-batch retries use advisory review data without granting recognition.""" +import copy +import json +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service import init_knowledge_parallel as parallel +from infermatrix_copilot.kb_service.init_coverage import Owner +from infermatrix_copilot.kb_service.init_support import InitError, InitRecord +from infermatrix_copilot.trace_store import TraceStore +from test_kb_init_skeleton import world # noqa: F401 + + +def _stage_job(facets=("api", "configuration")): + stage = SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=True, gateway=None), + record=InitRecord("knowledge", "toy", pin="a" * 40, inputs_digest="batch"), + lifecycle=SimpleNamespace(full_name="o/toy")) + payload = {"repository": "o/toy", "pin": "a" * 40, "owner": "sample", "facets": list(facets), + "foundation_prompt_version": 4, "files": [], "docs": []} + job = {"owner": Owner("sample", "repos/toy/sample.md", ()), "page": "repos/toy/sample.md", + "requested": list(facets), "payload": payload, "offered": {}} + return stage, job + + +def _task(job, sequence=0, *, verdict="unsure", error=""): + payload = copy.deepcopy(job["payload"]) + value = {"owner": job["owner"].owner, "page": job["page"], "payload": payload, + "requested": job["requested"][:], "offered": job["offered"], "artifacts": [], "dropped": [], + "sequence": sequence, "unfinished": [], "error": error, + "verdicts": {f"knowledge:{job['owner'].owner}:{facet}": {"verdict": verdict, + "page": job["page"], "facet": facet, + "reasons": {"faithful": "Caller obligation is not supported; narrow the claim."}} + for facet in job["requested"]}} + value["input_sha256"] = parallel._hash({"payload": payload, "offered": value["offered"]}) + value["result_sha256"] = parallel._hash(value) + return value + + +def test_feedback_matches_only_still_requested_owner_page_and_version(): + stage, job = _stage_job() + initial = _task(job) + saved = {"binding": "batch", "tasks": {"initial": initial}} + for key, changed in (("owner", {"owner": "other"}), ("page", {"page": "other.md"})): + saved["tasks"][key] = {**copy.deepcopy(initial), **changed} + wrong_payload_owner = copy.deepcopy(initial) + wrong_payload_owner["payload"]["owner"] = "other" + saved["tasks"]["payloadowner"] = wrong_payload_owner + legacy = copy.deepcopy(initial) + legacy["payload"].pop("foundation_prompt_version") + saved["tasks"]["legacy"] = legacy + unrequested = copy.deepcopy(initial) + unrequested["requested"] = ["validation"] + saved["tasks"]["unrequested"] = unrequested + before = copy.deepcopy(saved) + job["requested"] = job["payload"]["facets"] = ["api"] + retry = list(parallel._with_review_feedback(stage, [job], saved))[0] + feedback = retry["payload"]["prior_review_feedback"] + assert feedback["advisory_only"] is True and set(feedback["facets"]) == {"api"} + assert feedback["facets"]["api"]["correction_round"] == 1 + assert feedback["facets"]["api"]["last_verdict"] == "unsure" + assert feedback["facets"]["api"]["prior_attempts"] == [ + {k: initial[k] for k in ("input_sha256", "result_sha256")}] + assert "Caller obligation" in feedback["facets"]["api"]["last_review_reasons"]["faithful"] + assert saved == before and stage.record.evidence == {} + + +def test_three_corrections_preserve_each_attempt_and_exhaust_only_that_facet(): + stage, job = _stage_job(("api",)) + initial = _task(job, error="ModelUnavailable: initial generator failed") + saved = {"binding": "batch", "tasks": {initial["input_sha256"]: initial}} + hashes = {initial["input_sha256"]} + for number in range(1, 4): + before = copy.deepcopy(saved) + retry = list(parallel._with_review_feedback(stage, [job], saved))[0] + detail = retry["payload"]["prior_review_feedback"]["facets"]["api"] + assert detail["correction_round"] == number and len(detail["prior_attempts"]) == number + if number == 1: + assert "initial generator failed" in detail["last_error"] + task = _task(retry, number, verdict="no") + assert task["input_sha256"] not in hashes and saved == before + hashes.add(task["input_sha256"]) + saved["tasks"][task["input_sha256"]] = task + assert len(saved["tasks"]) == 4 + assert list(parallel._with_review_feedback(stage, [job], saved)) == [] + assert "three review corrections exhausted; remains unknown" in stage.record.unfinished[-1] + stage2, wider = _stage_job(("api", "configuration")) + remaining = list(parallel._with_review_feedback(stage2, [wider], saved))[0] + assert remaining["requested"] == remaining["payload"]["facets"] == ["configuration"] + assert "prior_review_feedback" not in remaining["payload"] and stage2.record.evidence == {} + + +def test_uncheckpointed_interruption_does_not_consume_a_correction(): + stage, job = _stage_job(("api",)) + initial = _task(job) + saved = {"binding": "batch", "tasks": {"initial": initial, "duplicate-reference": initial}} + first = list(parallel._with_review_feedback(stage, [job], saved))[0] + second = list(parallel._with_review_feedback(stage, [job], saved))[0] + assert first == second and first["payload"]["prior_review_feedback"]["facets"]["api"]["correction_round"] == 1 + completed = _task(first, 1) + saved["tasks"][completed["input_sha256"]] = completed + next_job = list(parallel._with_review_feedback(stage, [job], saved))[0] + assert next_job["payload"]["prior_review_feedback"]["facets"]["api"]["correction_round"] == 2 + + +@pytest.mark.parametrize("damage", ["page", "facet", "pass"]) +def test_mismatched_inner_review_and_pass_are_not_rejection_feedback(damage): + stage, job = _stage_job(("api",)) + prior = _task(job) + row = prior["verdicts"]["knowledge:sample:api"] + row["verdict" if damage == "pass" else damage] = "pass" if damage == "pass" else "other" + prior["dropped"] = [ + {"rule_id": "knowledge:sample:api", "page": job["page"], "why": "shown anchor needs correction"}, + {"rule_id": "knowledge:sample:api", "page": "wrong.md", "why": "unrelated wrong page"}, + {"rule_id": "knowledge:sample:configuration", "page": job["page"], "why": "unrelated facet"}] + prior["result_sha256"] = parallel._hash({k: v for k, v in prior.items() if k != "result_sha256"}) + retry = list(parallel._with_review_feedback(stage, [job], {"binding": "batch", "tasks": {"initial": prior}}))[0] + detail = retry["payload"]["prior_review_feedback"]["facets"]["api"] + assert detail["last_review_reasons"] == {} and detail["last_verdict"] == "unknown" + assert detail["last_dropped_reasons"] == ["shown anchor needs correction"] + assert detail["correction_round"] == 1 # The actual requested extraction still happened. + assert stage.record.evidence == {} + + +@pytest.mark.parametrize("damage", ["result_hash", "validated_hash", "batch"]) +def test_unvalidated_history_cannot_supply_retry_feedback(damage): + stage, job = _stage_job() + prior = _task(job) + saved = {"binding": "batch", "tasks": {"initial": prior}} + if damage in {"result_hash", "validated_hash"}: + if damage == "validated_hash": + stage._foundation_validated_results = {prior["result_sha256"]} + prior["verdicts"]["knowledge:sample:api"]["reasons"] = {"fake": "approval"} + else: + saved["binding"] = "other batch" + with pytest.raises(InitError, match="foundation"): + list(parallel._with_review_feedback(stage, [job], saved)) + + +@pytest.mark.parametrize("mode", ["initial", "finite", "legacy"]) +def test_initial_finite_and_legacy_jobs_are_unchanged(mode): + stage, job = _stage_job() + saved = {"binding": "batch", "tasks": {}} + if mode == "finite": + stage.rt.unlimited_subscription = False + saved["tasks"]["ignored"] = _task(job) + elif mode == "legacy": + job["payload"].pop("foundation_prompt_version") + saved["tasks"]["ignored"] = _task(job) + assert list(parallel._with_review_feedback(stage, [job], saved)) == [job] + assert "prior_review_feedback" not in job["payload"] + + +def test_actual_native_retry_prompt_contains_only_failed_facet_feedback_and_retains_old_proofs(world, monkeypatch): + from test_kb_foundation_publication import test_incomplete_native_publication_resumes_only_missing_facet + jobs, original = [], parallel._worker + def capture(stage, job): + jobs.append(copy.deepcopy(job)) + return original(stage, job) + monkeypatch.setattr(parallel, "_worker", capture) + # The real offline native generator/judge fixture checks old approved tasks + # are identical on resume, only the failed facet runs, and publication + # stays blocked until independent validation also passes. + test_incomplete_native_publication_resumes_only_missing_facet(world) + corrected = [job for job in jobs if "prior_review_feedback" in job["payload"]] + assert len(corrected) == 1 + job = corrected[0] + assert job["owner"].owner == "feature-demo" and job["requested"] == ["validation"] + feedback = job["payload"]["prior_review_feedback"]["facets"] + assert set(feedback) == {"validation"} and feedback["validation"]["correction_round"] == 1 + assert "scripted missing validation support" in feedback["validation"]["last_review_reasons"]["faithful"] + state = world["tmp"] / "required-foundation" + record = InitRecord.load(state, "toy", "knowledge") + task = next(t for t in record.coverage["foundation_jobs"]["tasks"].values() + if "prior_review_feedback" in t["payload"]) + traces = TraceStore(state / "init" / "traces") + receipt = task["artifacts"][0]["generator_receipt"] + native = traces.get(receipt["native_trace_id"]) + prompt = traces.blob(native["inputs"]["prompt"]) + payload = json.loads(prompt.split("\n", 1)[1].rsplit("", 1)[0]) + assert payload["prior_review_feedback"] == job["payload"]["prior_review_feedback"] + assert payload["files"] and task["artifacts"][0]["verdict"]["verdict"] == "pass" + from infermatrix_copilot.kb_service.init_knowledge_inputs import SYSTEM_KNOWLEDGE_V4 + from infermatrix_copilot.kb_service.models import ModelGateway, ModelRole + from test_kb_init_skeleton import _runtime + assert traces.blob(native["inputs"]["system"]) == SYSTEM_KNOWLEDGE_V4 + rt = _runtime(world, ModelGateway(None, transport_factory=lambda _: pytest.fail("native retry replay dispatched a model")), + state_dir=state, generator=ModelRole("generator", "zcode", "GLM-5.3")) + rt.unlimited_subscription = True + stage = SimpleNamespace(rt=rt, record=record, lifecycle=SimpleNamespace(full_name="o/toy"), tags=["toy"], + observer=rt.upstream("toy", "o/toy").observer(record.pin)) + before = InitRecord.path(state, "toy", "knowledge").read_bytes() + parallel._validate_native(stage, task["artifacts"][0], task) + altered = copy.deepcopy(task) + altered["payload"]["prior_review_feedback"]["facets"]["validation"]["last_review_reasons"] = {"fake": "approval"} + altered["input_sha256"] = parallel._hash({"payload": altered["payload"], "offered": altered["offered"]}) + altered["result_sha256"] = parallel._hash({k: v for k, v in altered.items() if k != "result_sha256"}) + with pytest.raises(InitError, match="native approval binding mismatch"): + parallel._validate_native(stage, altered["artifacts"][0], altered) + assert InitRecord.path(state, "toy", "knowledge").read_bytes() == before diff --git a/test/test_kb_foundation_local_privacy.py b/test/test_kb_foundation_local_privacy.py new file mode 100644 index 00000000..a2fe9f63 --- /dev/null +++ b/test/test_kb_foundation_local_privacy.py @@ -0,0 +1,87 @@ +"""Publication eligibility stays separate from immutable native approval.""" +import copy +import hashlib +from types import SimpleNamespace + +from infermatrix_copilot.kb_service import init_knowledge_parallel as parallel +from infermatrix_copilot.kb_service.init_support import InitRecord +from test_kb_foundation_context import _audit_stage +from test_kb_foundation_feedback import _stage_job, _task +from test_kb_foundation_parallel import _complete_native +from test_kb_init_knowledge import KnowledgeGateway +from test_kb_init_skeleton import world # noqa: F401 + + +def test_genuine_old_native_pass_is_locally_declined_without_mutating_proof(world, monkeypatch): + original = KnowledgeGateway.call_json + def historical(self, role, **kwargs): + reply = original(self, role, **kwargs) + if role.name == "generator": + for section in reply.data.get("sections", []): + if section["facet"] == "tradeoffs": + section["body"] += " Example wildcard host: `0.0.0.0`." + return reply + # Simulate genuine previously accepted native prose before local publication + # checks; only the offline fixture's old tree/privacy gate is bypassed. + with monkeypatch.context() as old: + old.setattr(KnowledgeGateway, "call_json", historical) + old.setattr(parallel, "local_privacy_reason", lambda text: "") + old.setattr("infermatrix_copilot.kb_service.init_stages.run_knowledge_validators", lambda *a: []) + rt, record, offline, traces = _complete_native(world) + before = copy.deepcopy(record.coverage["foundation_jobs"]) + native_bytes = {p: p.read_bytes() for p in (rt.state_dir / "init" / "traces" / "records").glob("*.jsonl")} + count = len(offline.calls) + parallel.validate_checkpoint(_audit_stage(rt, record), record, record.inputs_digest) + task = next(iter(before["tasks"].values())) + applied = [] + stage = SimpleNamespace(rt=rt, record=InitRecord("knowledge", "toy"), + _append_approved=lambda *args: applied.append(args[3]) or args[3]) + # The exact task already passed the unchanged full native/source replay above. + monkeypatch.setattr(parallel, "_validate_result", lambda *a: None) + parallel._apply(stage, task) + assert "tradeoffs" not in applied and len(applied) == len(task["artifacts"]) - 1 + assert stage.record.verdicts["knowledge:" + task["owner"] + ":tradeoffs"]["verdict"] == "pass" + assert stage.record.dropped[-1]["local_publication_exclusion"] is True + assert "0.0.0.0" not in str(stage.record.dropped) + str(stage.record.unfinished) + assert record.coverage["foundation_jobs"] == before and len(offline.calls) == count + assert {p: p.read_bytes() for p in native_bytes} == native_bytes + + +def test_feedback_recomputes_only_matching_latest_artifact_local_problem(): + stage, job = _stage_job() + prior = _task(job, verdict="pass") + text = "Listen on 0.0.0.0 for wildcard IPv4 interfaces." + artifact = {"key": "knowledge:sample:api", "owner": "sample", "page": job["page"], + "facet": "api", "text": text, "text_sha256": hashlib.sha256(text.encode()).hexdigest()} + prior["artifacts"] = [artifact, {**artifact, "page": "other.md", "facet": "configuration"}, + {**artifact, "owner": "other", "facet": "configuration"}] + prior["result_sha256"] = parallel._hash({k: v for k, v in prior.items() if k != "result_sha256"}) + stage._foundation_validated_results = {prior["result_sha256"]} + saved = {"binding": "batch", "tasks": {"initial": prior}} + before = copy.deepcopy(saved) + retry = list(parallel._with_review_feedback(stage, [job], saved))[0] + facets = retry["payload"]["prior_review_feedback"]["facets"] + assert facets["api"]["prior_artifact_text_sha256"] == artifact["text_sha256"] + assert "local publication exclusion" in facets["api"]["last_local_publication_reasons"][0] + assert facets["api"]["last_verdict"] == "unknown" # Native pass never becomes a rejection. + assert facets["api"]["prior_attempts"] == [{k: prior[k] for k in ("input_sha256", "result_sha256")}] + assert "last_local_publication_reasons" not in facets["configuration"] + assert "0.0.0.0" not in str(facets) and saved == before + + +def test_only_unsafe_items_are_excluded_and_loopback_or_finite_mode_are_preserved(monkeypatch): + for unlimited, expected in ((True, ["api", "validation"]), (False, ["api", "configuration", "features", "validation"])): + artifacts = [{"key": "knowledge:sample:" + facet, "owner": "sample", "page": "page.md", + "facet": facet, "title": "Title", "text": text, "evidence": []} + for facet, text in [("api", "127.0.0.1 remains an allowed loopback example"), + ("configuration", "wildcard host 0.0.0.0"), + ("features", "frontend wildcard host 0.0.0.0"), ("validation", "manual source checks")]] + task = {"artifacts": artifacts, "verdicts": {}, "dropped": [], "unfinished": []} + before = copy.deepcopy(task) + applied = [] + stage = SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=unlimited), + record=InitRecord("knowledge", "toy"), _append_approved=lambda *a: applied.append(a[3]) or a[3]) + monkeypatch.setattr(parallel, "_validate_result", lambda *a: None) + parallel._apply(stage, task) + assert applied == expected and task == before + assert len(stage.record.dropped) == (2 if unlimited else 0) diff --git a/test/test_kb_foundation_partial_interfaces.py b/test/test_kb_foundation_partial_interfaces.py new file mode 100644 index 00000000..9e8ad6e0 --- /dev/null +++ b/test/test_kb_foundation_partial_interfaces.py @@ -0,0 +1,140 @@ +"""Offline interface checks; mocked handoff never claims a native approval.""" + +import argparse +import hashlib +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service import cli, runner as playbooks +from infermatrix_copilot.kb_service.init_support import InitRecord, InitRuntime +from infermatrix_copilot.kb_service.sources import KnowledgeRepo +from test_kb_depth_campaign import runner, campaign_args, _offline_processes +from test_kb_depth_assembly import assembler, world, _commit, _candidate + + +@pytest.mark.parametrize("command", [["init", "demo", "--stage", "knowledge"], ["widen", "demo"]]) +def test_partial_foundation_cli_forwards_only_explicit_mode(tmp_path, monkeypatch, command): + seen = [] + monkeypatch.setattr(playbooks, "run_playbook", lambda *_a, **k: ( + seen.append(k["params"]) or SimpleNamespace(status="done"), tmp_path)) + assert cli.main(command + ["--foundation-mode", "partial", "--unlimited-subscription"]) == 0 + assert seen[0]["foundation_mode"] == "partial" and "foundation_record" not in seen[0] + + +def test_partial_depth_cli_forwards_original_record(tmp_path, monkeypatch): + seen = [] + record = tmp_path / "published-knowledge.json" + monkeypatch.setattr(playbooks, "run_playbook", lambda *_a, **k: ( + seen.append(k["params"]) or SimpleNamespace(status="done"), tmp_path)) + assert cli.main(["deepen", "demo", "--foundation-mode", "partial", "--foundation-record", str(record), + "--unlimited-subscription"]) == 0 + assert seen[0]["foundation_record"] == str(record) + + +@pytest.mark.parametrize("options", [ + ["--stage", "modules", "--foundation-mode", "partial", "--unlimited-subscription"], + ["--stage", "knowledge", "--foundation-mode", "partial"], + ["--stage", "knowledge", "--foundation-mode", "partial", "--foundation-record", "record.json", "--unlimited-subscription"], + ["--stage", "knowledge-deepen", "--foundation-mode", "partial", "--unlimited-subscription"], + ["--stage", "knowledge-deepen", "--foundation-record", "record.json", "--unlimited-subscription"], +]) +def test_invalid_partial_cli_never_dispatches(monkeypatch, options): + monkeypatch.setattr(playbooks, "run_playbook", lambda *_a, **_k: pytest.fail("invalid partial request dispatched")) + assert cli.main(["init", "demo", *options]) == 2 + + +def test_default_cli_does_not_add_foundation_options(tmp_path, monkeypatch): + seen = [] + monkeypatch.setattr(playbooks, "run_playbook", lambda *_a, **k: ( + seen.append(k["params"]) or SimpleNamespace(status="done"), tmp_path)) + assert cli.main(["widen", "demo", "--unlimited-subscription"]) == 0 + assert not {"foundation_mode", "foundation_record"} & seen[0].keys() + + +def _stub_handoff(args, monkeypatch): + from infermatrix_copilot.kb_service import foundation_publication + record = args.state / "immutable-foundation.json" + record.write_text("offline handoff fixture\n") + binding = {"record_path": str(record.resolve()), "record_sha256": hashlib.sha256(record.read_bytes()).hexdigest(), + "receipt_path": str(args.state / "receipt.json"), "receipt_sha256": "a" * 64} + args.foundation_mode, args.foundation_record = "partial", record + monkeypatch.setattr(foundation_publication, "foundation_handoff", lambda *_a, **_k: binding) + return binding + + +def test_campaign_partial_binds_every_child_and_rejects_strict_resume(runner, campaign_args, monkeypatch): + args = campaign_args + binding = _stub_handoff(args, monkeypatch) + _offline_processes(runner, monkeypatch) + argv = [] + spawn = runner.subprocess.Popen + monkeypatch.setattr(runner.subprocess, "Popen", lambda a, **k: argv.append(a) or spawn(a, **k)) + assert runner.campaign(args, argparse.ArgumentParser()) == 0 + metadata = json.loads((args.state / "campaign.json").read_text()) + assert metadata["foundation_mode"] == "partial" and metadata["foundation"] == binding + assert metadata["denominator"] == 553 and metadata["features"] == 79 + for child in argv: + assert child[child.index("--foundation-mode") + 1] == "partial" + assert child[child.index("--foundation-record") + 1] == binding["record_path"] + assert child[child.index("--foundation-record-sha256") + 1] == binding["record_sha256"] + args.foundation_mode, args.foundation_record = "strict", None + monkeypatch.setattr(runner.subprocess, "Popen", lambda *_a, **_k: pytest.fail("changed mode dispatched")) + with pytest.raises(SystemExit): + runner.campaign(args, argparse.ArgumentParser()) + + +def test_worker_handoff_rejects_changed_record_before_runtime(runner, campaign_args, monkeypatch): + args = campaign_args + binding = _stub_handoff(args, monkeypatch) + args.foundation_record_sha256 = "f" * 64 + monkeypatch.setattr(KnowledgeRepo, "fetch", lambda _: args.baseline) + monkeypatch.setattr(InitRuntime, "from_env", lambda *_a, **_k: pytest.fail("tampered prerequisite started runtime")) + with pytest.raises(ValueError, match="changed after campaign binding"): + runner.run_worker(args, args.state / "worker-0") + + +def test_worker_partial_forwards_bound_record_without_changing_partition(runner, campaign_args, monkeypatch): + from infermatrix_copilot.kb_service import init_stages + args = campaign_args + binding = _stub_handoff(args, monkeypatch) + (args.state / "campaign.json").write_text(json.dumps({"foundation_mode": "partial", "foundation": binding})) + monkeypatch.setattr(KnowledgeRepo, "fetch", lambda _: args.baseline) + runtime = SimpleNamespace(registry={args.repo: object()}, gateway=SimpleNamespace(configure_zcode_pacing=lambda _: None)) + monkeypatch.setattr(InitRuntime, "from_env", lambda *_a, **_k: runtime) + seen = [] + def run_stage(*_a, **k): + seen.append(k) + return InitRecord(stage="knowledge-deepen", repo=args.repo, pin=args.pin, kb_base_sha=args.baseline, status="partial") + monkeypatch.setattr(init_stages, "run_stage", run_stage) + assert runner.run_worker(args, args.state / "worker-0") == 0 + assert seen[0]["feature_ids"] == ("f0",) and seen[0]["foundation_mode"] == "partial" + assert seen[0]["foundation_record_path"] == Path(binding["record_path"]) + + +def test_partial_assembly_absent_overview_uses_owner_link_with_old_blocks_intact(assembler, world, monkeypatch): + from infermatrix_copilot.kb_service import foundation_publication + f1 = world.policy.features[1] + (world.root / "knowledge" / f1.page).unlink() + world.campaign["baseline"] = _commit(world.root) + binding = {"record_path": str(world.state / "published-foundation.json"), "record_sha256": "c" * 64} + world.campaign.update(foundation_mode="partial", foundation=binding) + (world.state / "campaign.json").write_text(json.dumps(world.campaign)) + for record, path in zip(world.records, world.checkpoints): + record["kb_base_sha"] = world.campaign["baseline"] + record["depth"].update(foundation_mode="partial", foundation=binding) + path.write_text(json.dumps(record)) + _candidate(world, lambda text: text.replace("[功能概览](feature-f1.md) · [owner 入口](_index.md)", + "[owner 入口](_index.md)"), worker=1) + monkeypatch.setattr(foundation_publication, "foundation_handoff", lambda *_a, **_k: binding) + report, writes = assembler.assemble(world.root, world.state, world.source, + foundation_mode="partial", foundation_record_path=Path(binding["record_path"])) + assert report["foundation"] == binding and report["denominator"] == 14 + assert report["new_receipt_bound_facets"] == 2 and world.blocks[("f0", "api")] in next( + text for page, text in writes.items() if page.endswith("feature-depth-f0.md")) + newpage = next(text for page, text in writes.items() if page.endswith("feature-depth-f1.md")) + assert "[owner 入口](_index.md)" in newpage and "[功能概览]" not in newpage + with pytest.raises(ValueError, match="mode differs"): + assembler.assemble(world.root, world.state, world.source) diff --git a/test/test_kb_foundation_partial_prbody.py b/test/test_kb_foundation_partial_prbody.py new file mode 100644 index 00000000..8e85168f --- /dev/null +++ b/test/test_kb_foundation_partial_prbody.py @@ -0,0 +1,63 @@ +"""Partial metadata stays bounded and retries retain the original publication.""" +import copy +import hashlib +import json +from types import SimpleNamespace + +from infermatrix_copilot.kb_service.init_stages import render_pr_body +from infermatrix_copilot.kb_service.init_support import load_prepared, save_prepared +from test_kb_foundation_explicit_partial import _retained_native_partial, _run_partial +from test_kb_init_modules import _modules_lifecycle +from test_kb_init_skeleton import world # noqa: F401 + + +def test_large_native_history_has_bounded_truthful_partial_body(world): + rt, original, calls, traces, page = _retained_native_partial(world) + record = _run_partial(rt) + original_jobs = copy.deepcopy(record.coverage['foundation_jobs']) + # The actual failure arose from serializing accumulated native verdicts, + # drops and gaps into metadata. Their size must never change body bounds. + record.verdicts.update({f'facet-{i}': {'verdict':'pass', 'page':page} for i in range(10000)}) + record.dropped.extend([{'rule_id':'unknown', 'why':'retained independent reasons '*200}] * 10000) + record.unfinished.extend(['retained unknown '*200] * 10000) + before = copy.deepcopy(record) + body = render_pr_body(record, _modules_lifecycle()) + assert len(body.encode()) < 10000 + assert 'init_complete=false' in body and '1 项' in body and '0/1' in body + assert record.coverage['foundation_publication']['receipt']['sha256'] in body + assert record.coverage['foundation_jobs'] == original_jobs and record == before + + +def test_strict_formatter_remains_exact_legacy_path(): + from infermatrix_copilot.kb_service.init_support import InitRecord + record = InitRecord('knowledge','toy',pin='a'*40,kb_base_sha='b'*40) + body = render_pr_body(record, _modules_lifecycle()) + assert body == '\n'.join([ + '`kb init` stage **knowledge** for `toy` (`o/toy`).', '', + '- Upstream pin: `'+'a'*40+'`', '- Knowledge base: `'+'b'*40+'`', + '- Model spend (accounted): $0.00', '', + 'Human-merged. Explanatory knowledge has pinned source references and per-facet advisory verdicts. ' + 'Design inferences are labeled; failed sections are removed and missing facets remain listed.', '']) + + +def test_prepared_body_refresh_preserves_native_jobs_files_and_commit_metadata(world): + rt, original, calls, traces, page = _retained_native_partial(world, fail_create=1) + first = _run_partial(rt) + assert first.status == 'blocked' and first.pr['prepared'] + prepared = load_prepared(first.pr['prepared']) + # Simulate the authentic older runtime's already prepared oversized body + # through the same official metadata writer, without rewriting any source. + oversized = {**prepared, 'body':'Old full review history\n'+'retained failed context '*60000} + save_prepared(first.pr['prepared'], **oversized) + before_tasks = copy.deepcopy(first.coverage['foundation_jobs']) + before_proofs = first.coverage['foundation_publication']['receipt'].copy() + before_calls = len(calls) + resumed = _run_partial(rt) + after = load_prepared(resumed.pr['prepared']) + assert resumed.status == 'published', resumed.problems + assert after['body'] != oversized['body'] and len(after['body'].encode()) < 10000 + assert {k:v for k,v in after.items() if k!='body'} == {k:v for k,v in oversized.items() if k!='body'} + assert len(calls) == before_calls + assert resumed.inputs_digest == original.inputs_digest + assert resumed.coverage['foundation_jobs'] == before_tasks + assert resumed.coverage['foundation_publication']['receipt'] == before_proofs diff --git a/test/test_kb_foundation_related_tests.py b/test/test_kb_foundation_related_tests.py new file mode 100644 index 00000000..df6246cd --- /dev/null +++ b/test/test_kb_foundation_related_tests.py @@ -0,0 +1,76 @@ +"""Shared test inventory supplies real late assertions within the source cap.""" +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service.feature_discovery_index import build_discovery_index +from infermatrix_copilot.kb_service.init_knowledge_parallel import related_test_sources +from infermatrix_copilot.kb_service.init_support import InitError, InitRecord + + +def _stage(tmp_path): + files = {"src/tool.py": "class SpecificTool:\n def run(self):\n return 7\n", + "sdk/nested/tests/test_tool.py": "from src.tool import SpecificTool\n" + "# filler\n" * 220 + + "def test_real_tool():\n assert SpecificTool().run() == 7\n", + "tests/test_name_only_tool.py": "def test_unrelated():\n assert 'SpecificTool' == 'SpecificTool'\n", + "frontend/src/widget.ts": "export function SpecificWidget() { return 7; }\n", + "frontend/tests/widget.test.ts": "import {SpecificWidget} from '../src/widget';\n" + "// filler\n" * 240 + + "test('widget', () => { expect(SpecificWidget()).toBe(7); });\n", + "opaque/tool.strange": "function SpecificOpaque() { return 7; }\n", + "opaque/tests/check.strange": "import 'opaque/tool.strange';\nassert SpecificOpaque() == 7\n"} + for path, text in files.items(): + target = tmp_path / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + index = build_discovery_index(tmp_path, pin="a" * 40, scope={"roots": ["."], "exclude": []}) + return SimpleNamespace(source_index=index, record=InitRecord("knowledge", "toy", pin="a" * 40)) + + +@pytest.mark.parametrize("source,target,assertion", [ + ("src/tool.py", "sdk/nested/tests/test_tool.py", "assert SpecificTool().run() == 7"), + ("frontend/src/widget.ts", "frontend/tests/widget.test.ts", "expect(SpecificWidget()).toBe(7)"), + ("opaque/tool.strange", "opaque/tests/check.strange", "assert SpecificOpaque() == 7"), +]) +def test_direct_imports_and_late_assertions_include_nested_multilanguage_tests(tmp_path, source, target, assertion): + stage = _stage(tmp_path) + excerpts = related_test_sources(stage, [source]) + chosen = [item for item in excerpts if item["path"] == target] + assert chosen and any(assertion in item["text"] for item in chosen) + if source != "opaque/tool.strange": + assert any(item["start"] > 180 for item in chosen) + assert all("not been executed" in item["test_context"] for item in chosen) + assert all(item["path"] != "tests/test_name_only_tool.py" for item in excerpts) + assert sum(len(item["text"].encode()) for item in excerpts) <= 20000 + + +def test_test_cap_keeps_complete_intervals_and_read_failures_are_not_absence(tmp_path): + stage = _stage(tmp_path) + assert related_test_sources(stage, ["src/tool.py"], limit=1) == [] + path = "sdk/nested/tests/test_tool.py" + stage.source_index.entries[path]["status"] = "error" + assert related_test_sources(stage, ["src/tool.py"]) == [] + assert stage.record.evidence == {} # No missing-test assertion or recognition. + stage.record.pin = "b" * 40 + with pytest.raises(InitError, match="test index source pin"): + related_test_sources(stage, ["src/tool.py"]) + + +def test_specific_capability_test_survives_earlier_shared_entry_test_budget(tmp_path): + files = { + "src/app.py": "def shared_entry():\n return 1\n", + "src/share_image_export.py": "class ShareImageExportManager:\n def export(self):\n return 'png'\n", + "tests/a_shared.py": "from src.app import shared_entry\n\ndef test_shared_entry():\n" + + " assert shared_entry() == 1\n" * 150, + "tests/test_share_image_export.py": "from src.share_image_export import ShareImageExportManager\n\n" + + "def test_png_export():\n assert ShareImageExportManager().export() == 'png'\n", + } + for path, text in files.items(): + target = tmp_path / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + index = build_discovery_index(tmp_path, pin="a" * 40, scope={"roots": ["."], "exclude": []}) + stage = SimpleNamespace(source_index=index, record=InitRecord("knowledge", "toy", pin="a" * 40)) + excerpts = related_test_sources(stage, ["src/app.py", "src/share_image_export.py"], limit=600) + assert excerpts[0]["path"] == "tests/test_share_image_export.py" + assert any("assert ShareImageExportManager().export()" in item["text"] for item in excerpts) + assert sum(len(item["text"].encode()) for item in excerpts) <= 600 diff --git a/test/test_kb_init_knowledge.py b/test/test_kb_init_knowledge.py index 58a769bd..f4fd8c69 100644 --- a/test/test_kb_init_knowledge.py +++ b/test/test_kb_init_knowledge.py @@ -8,6 +8,7 @@ from infermatrix_copilot.kb_service.init_knowledge import ( FACETS, MAX_DOC_BYTES, SYSTEM_KNOWLEDGE, _Knowledge, knowledge_prompt, validate_sections, ) +from infermatrix_copilot.kb_service.init_knowledge_inputs import SYSTEM_KNOWLEDGE_V4 from infermatrix_copilot.kb_service.init_stages import run_stage from infermatrix_copilot.kb_service.init_support import InitRecord from infermatrix_copilot.knowledge_service.lifecycle import Page @@ -126,7 +127,7 @@ def __init__(self, *, bad_evidence=False, rejected=False, cost=0.01, **kwargs): self.cost = cost def call_json(self, role, *, system, prompt, validate=None, max_budget_usd=None): - if system != SYSTEM_KNOWLEDGE: + if system not in (SYSTEM_KNOWLEDGE, SYSTEM_KNOWLEDGE_V4): return super().call_json(role, system=system, prompt=prompt, validate=validate, max_budget_usd=max_budget_usd) from infermatrix_copilot.kb_service.models import ModelReply diff --git a/tools/audit_knowledge_depth.py b/tools/audit_knowledge_depth.py index 3e75d98d..61180c9c 100644 --- a/tools/audit_knowledge_depth.py +++ b/tools/audit_knowledge_depth.py @@ -11,7 +11,7 @@ import yaml -from infermatrix_copilot.kb_service.knowledge_coverage import audit_coverage, load_policy, policy_path +from infermatrix_copilot.kb_service.knowledge_coverage import audit_coverage, coverage_targets_met, load_policy, policy_path from infermatrix_copilot.kb_service.knowledge_depth import audit_depth @@ -26,7 +26,13 @@ def main() -> int: help="Repeatable native all-yes block binding report, outside product knowledge") parser.add_argument("--require-approvals", action="store_true", help="Fail unless every recognized block has a matching native all-yes approval") + parser.add_argument("--foundation-mode", choices=("strict", "partial"), default="strict") + parser.add_argument("--foundation-record", type=Path) args = parser.parse_args() + if args.foundation_mode == "partial" and (not args.foundation_record or not args.require_approvals): + parser.error("partial foundation requires --foundation-record and --require-approvals") + if args.foundation_mode == "strict" and args.foundation_record: + parser.error("--foundation-record requires --foundation-mode partial") current = subprocess.check_output(["git", "-C", str(args.upstream), "rev-parse", "HEAD"], text=True).strip() if current != args.pin or len(args.pin) != 40: parser.error("upstream HEAD must equal the full requested pin") @@ -59,6 +65,15 @@ def main() -> int: for p in args.approval_report or []], "reproduce": f"PYTHONPATH=src python tools/audit_knowledge_depth.py --repo {args.repo} " f"--upstream /path/to/pinned/{args.repo} --pin {args.pin}"}} + if args.foundation_mode == "partial": + from infermatrix_copilot.kb_service.foundation_publication import foundation_handoff + from infermatrix_copilot.kb_service.sources import KnowledgeRepo + knowledge_repo = KnowledgeRepo(args.root) + baseline = knowledge_repo._git("rev-parse", "HEAD").decode().strip() + report["foundation_mode"] = "partial" + report["foundation"] = foundation_handoff(args.foundation_record, knowledge=knowledge_repo, + baseline=baseline, repo=args.repo, pin=args.pin) + report["init_complete"] = False # Final init includes its separate publication and retrieval gates. if args.report: args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") @@ -69,9 +84,10 @@ def main() -> int: "approval_bindings_checked": depth["approval_bindings_checked"], "approval_binding_problems": depth["approval_binding_problems"], "production_files_with_semantic_evidence": len(depth["production_files_with_semantic_evidence"]), - "breadth_targets_met": report["breadth"]["met"], "problems": depth["problems"]}, indent=2)) + "breadth_targets_met": report["breadth"]["met"], + "structural_targets_met": report["breadth"]["structural"]["met"], "problems": depth["problems"]}, indent=2)) return 1 if depth["problems"] or depth["approval_binding_problems"] or not depth["target_met"] \ - or not report["breadth"]["met"] else 0 + or not coverage_targets_met(report["breadth"], args.foundation_mode) else 0 if __name__ == "__main__":