diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index cefbdf18..6ebb7b02 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -1,6 +1,6 @@ # kb_service/ —— 规范 - + `知识服务核心：仓库配置、账本、outbox、CLI · refactor-status: new` @@ -886,3 +886,26 @@ Zcode 模型角色中的 `effort` 被 transport 忽略，不能代替该设置 未发布的旧检查点缺少此审计时只补审，复用完整扫描和首次评审；prepared 产物缺少审计时 阻止发布，保留原产物。已发布历史目录保持不可变。审计预算中断保存已完成批次，恢复时 不重复已完成的审计。显式 `--retry-unfinished` 不重置已用完的三次内容修正，仅恢复通道故障。 + +## 2026-10-06 订阅模式基础知识并发 + +仅显式 `knowledge --unlimited-subscription` 启用基础知识工作池。 +`KB_KNOWLEDGE_CONCURRENCY` 默认为 13，允许 1–13，并纳入阶段输入身份； +旧有有限预算模式保持顺序执行。每个 owner/feature 的 GLM 提取及原六个独立 +Codex 区块评审在同一 worker 中顺序运行，整个池同时最多 13 个模型调用。原生 Zcode 调用复用共享节流器，默认每 5 秒启动一次； +`KB_KNOWLEDGE_START_INTERVAL_S` 可选有限正数且不超过 60 秒，同样冻结在输入身份中。 +worker 使用独立知识视图、阶段记录和订阅传输；唯一协调器保存检查点， +按既定任务顺序追加精确获批区块，重新核对源码证据、正文摘要及原生评审记录， +不会重新调用模型来组装。单项通道失败保留其真实状态和已认可成果，其余任务继续。 +恢复绑定原阶段输入身份与原生归档，不生成任何替代前置阶段记录。 + +新发现的正式功能优先使用已冻结发现报告中的完整源码行段，仍受提示词字节 +预算限制；不可完整展示的行段显式保留为缺口。引用必须落在实际展示的单个区间 +中，不以最大行号冒充中间未读取的内容。旧功能仍使用原来源选择及覆盖判定， +已覆盖页不重新提取。预算预留/结算采用原子操作；订阅费用未报告仍为未知。 + +完成但尚未发布的订阅知识缓存及已准备发布的恢复也必须先做只读原生证明复核。 +复核绑定原输入身份、源码 pin、KB 基点、有效 owner、两个模型的完整归档及 +精确正文/证据。失败在任何记录或发布写入前拒绝，恢复原归档后可以继续同一 +已准备的发布；已发布的历史记录保持不变。已有订阅工作检查点的输入发生变化 +时须建立新批次，不能静默覆盖或重新抽取。 diff --git a/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md b/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md index fd043e89..81ef23df 100644 --- a/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md +++ b/plugins/infermatrix-copilot/skills/imkbinit/SKILL.md @@ -326,3 +326,22 @@ a dry run, otherwise the opened PR). Summarise: Without explicit authorization for a broader batch or merge, stop after one stage. The owner reviews and merges its PR before the next stage can run. Never run two stages in one invocation, and never merge a PR. + +### Foundation concurrency (2026-10-06) + +Explicit `--stage knowledge --unlimited-subscription` shares at most 13 worker +slots across extraction and all existing independent section judgments. Set +`KB_KNOWLEDGE_CONCURRENCY` to 1–13 before starting; it is frozen in the input +identity. Native Zcode starts share a default 5-second pacing interval; +`KB_KNOWLEDGE_START_INTERVAL_S` may override it with a finite positive value +up to 60 seconds and is also bound to the stage identity. The CLI coordinator alone saves real worker checkpoints and appends +approved sections in deterministic order. Resume the same command/state to reuse +completed native approvals; keep their external trace archives. Finite-budget +runs retain the previous serial flow. New discovery features preferentially read +the frozen report’s exact complete source spans; unread gaps cannot be cited. + +Completed unpublished caches and prepared-publication resumes also replay the +original native proofs before any write or model call. Missing or changed +archives reject reuse without changing the genuine record; restore the original +archive to continue. Changed frozen inputs require a new batch. Published +historical records remain unchanged. diff --git a/src/infermatrix_copilot/kb_service/init_budget.py b/src/infermatrix_copilot/kb_service/init_budget.py index 4cdef7c5..257c3640 100644 --- a/src/infermatrix_copilot/kb_service/init_budget.py +++ b/src/infermatrix_copilot/kb_service/init_budget.py @@ -28,6 +28,7 @@ import json import math import os +import threading from contextlib import contextmanager from dataclasses import dataclass from typing import Iterator, Mapping @@ -128,7 +129,7 @@ def charge(self, actual: float | None) -> None: class Budget: """A finite stage ceiling, or explicit None for uncapped accounting. - Not thread-safe (stages call models one at a time).""" + Reservations are atomic; independent native calls may share one budget.""" def __init__(self, limit_usd: float | None, *, spent_usd: float = 0.0): if limit_usd is not None and (not math.isfinite(limit_usd) or limit_usd <= 0): @@ -138,10 +139,12 @@ def __init__(self, limit_usd: float | None, *, spent_usd: float = 0.0): self.limit_usd = float(limit_usd) if limit_usd is not None else None self.spent_usd = float(spent_usd) self._reserved = 0.0 + self._lock = threading.RLock() @property def remaining_usd(self) -> float | None: - return None if self.limit_usd is None else self.limit_usd - self.spent_usd - self._reserved + with self._lock: + return None if self.limit_usd is None else self.limit_usd - self.spent_usd - self._reserved def can_reserve(self, amount: float) -> bool: """An explicit None ceiling keeps accounting without limiting calls.""" @@ -156,14 +159,16 @@ def reserve(self, amount: float) -> Iterator[Reservation]: call when it does not fit. On exit the charge (or, when the block did not charge, e.g. because the call raised, the whole reservation) is added to what was spent.""" - if not self.can_reserve(amount): - raise BudgetExhausted( - f"reserving ${amount:.4f} would exceed the budget " - f"(spent ${self.spent_usd:.4f} of ${self.limit_usd:.2f})") - reservation = Reservation(amount) - self._reserved += amount + with self._lock: + if not self.can_reserve(amount): + raise BudgetExhausted( + f"reserving ${amount:.4f} would exceed the budget " + f"(spent ${self.spent_usd:.4f} of ${self.limit_usd:.2f})") + reservation = Reservation(amount) + self._reserved += amount try: yield reservation finally: - self._reserved -= amount - self.spent_usd += reservation.charged if reservation.charged is not None else amount + with self._lock: + self._reserved -= amount + self.spent_usd += reservation.charged if reservation.charged is not None else amount diff --git a/src/infermatrix_copilot/kb_service/init_knowledge.py b/src/infermatrix_copilot/kb_service/init_knowledge.py index e651add4..03d2dc00 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge.py @@ -30,6 +30,7 @@ from .init_support import InitRecord, classify_verdict, generate, judge from .init_knowledge_inputs import SYSTEM_KNOWLEDGE, knowledge_prompt, source_owner, source_owners from .models import ModelUnavailable +from .init_knowledge_parallel import offered_ranges, parallelism, run_jobs, restore_jobs, preferred_sources, start_interval FACETS = ("architecture", "api", "configuration", "tradeoffs", "features", "validation") MAX_SOURCE_BYTES = 100_000 @@ -118,8 +119,47 @@ def _precheck(self) -> list[str]: 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 "" - return {"knowledge_policy": hashlib.sha256(text.encode()).hexdigest(), - "from_existing": self.from_existing, "knowledge_prompt_version": 3} + options = {"knowledge_policy": hashlib.sha256(text.encode()).hexdigest(), + "from_existing": self.from_existing, "knowledge_prompt_version": 3} + if self.rt.unlimited_subscription: + options["foundation_parallel"] = {"version": 1, "workers": parallelism(self.rt), + "zcode_start_interval_s": start_interval(self.rt)} + return options + + def _discovery_gate(self, chain: _Chain, pin: str) -> None: + self._knowledge_run_pin = pin + super()._discovery_gate(chain, pin) + + def _cache_reusable(self, previous: InitRecord) -> bool: + if previous.status == "published": + return True # Preserve immutable historical publication records. + if self.rt.unlimited_subscription and previous.status != "published": + self._validate_unpublished_checkpoint(previous, previous.inputs_digest) + return not (self.rt.unlimited_subscription and previous.unfinished) + + def _resume_input_problems(self, previous: InitRecord, digest: str) -> list[str]: + if self.rt.unlimited_subscription and previous.status != "published": + # Raise before generic _blocked writes: restoring an archive must + # leave the same genuine prepared publication resumable. + self._validate_unpublished_checkpoint(previous, digest) + return super()._resume_input_problems(previous, digest) + + def _validate_unpublished_checkpoint(self, previous: InitRecord, digest: str) -> None: + from .init_knowledge_parallel import validate_checkpoint + validate_checkpoint(self, previous, digest) + + def _restore_progress(self, previous: InitRecord | None) -> list[str]: + if (self.rt.unlimited_subscription and previous is not None + and previous.coverage.get("foundation_jobs") + and previous.inputs_digest != self.record.inputs_digest): + from .init_support import InitError + raise InitError("foundation inputs changed; create a new pinned batch") + if (self.rt.unlimited_subscription and previous is not None + and previous.inputs_digest == self.record.inputs_digest): + from copy import deepcopy + self.record.coverage["foundation_jobs"] = deepcopy(previous.coverage.get("foundation_jobs", {})) + self.budget.spent_usd = previous.spent_usd + return [] def _build(self, tree: Path) -> InitRecord: if self.route_source == "none": @@ -138,6 +178,8 @@ def _build(self, tree: Path) -> InitRecord: routes = self.owners owners = source_owners(paths, routes, policy) self.owners = list(owners.values()) + if self.rt.unlimited_subscription: + restore_jobs(self) for rel in paths: hits = source_owner(rel, routes, owners) if hits: @@ -158,6 +200,7 @@ def _build(self, tree: Path) -> InitRecord: return self._blocked([f"knowledge target {page} must be an explanatory architecture or guide page"]) report = {} claims, evidence = {}, [] + owner_jobs = [] for position, name in enumerate(order): owner = owners[name] page = self._page_for(owner) @@ -182,7 +225,7 @@ def _build(self, tree: Path) -> InitRecord: continue source = self._sources(tree, sorted(files[name], key=lambda p: (-churn[p], p)), MAX_SOURCE_BYTES) docs = self._docs_for(tree, owner) - offered = {item["path"]: item["end"] for item in source + docs} + 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"]] self.record.unfinished.extend(f"{name}: byte cap truncated {path}" @@ -195,6 +238,10 @@ def _build(self, tree: Path) -> InitRecord: "language_sample": self._language_sample(), "related_owners": [{"owner": o.owner, "scope_prefixes": list(o.prefixes)} for o in self.owners]} + if self.rt.unlimited_subscription: + owner_jobs.append({"owner": owner, "page": page, "payload": payload, + "offered": offered, "requested": requested}) + continue try: data = generate(self.rt, self.budget, self.lifecycle.init, system=SYSTEM_KNOWLEDGE, prompt=knowledge_prompt(payload), validate=validate_sections).data @@ -216,14 +263,26 @@ def _build(self, tree: Path) -> InitRecord: break except ModelUnavailable as exc: self.record.unfinished.append(f"knowledge owner {name}: unusable draft: {exc}") + if owner_jobs: + for job, results in run_jobs(self, owner_jobs): + name = job["owner"].owner + for key, text, entries, label in results: + facet = key.rsplit(":", 1)[-1] + claims[key] = text + evidence.extend(entries) + report[name]["facets"][facet] = "covered" + for facet in job["requested"]: + verdict = self.record.verdicts.get(f"knowledge:{name}:{facet}", {}) + if verdict.get("verdict") in ("unsure", "unjudged"): + report[name]["facets"][facet] = "needs_review" for name in order: report.setdefault(name, {"page": self._page_for(owners[name]), "facets": dict.fromkeys(FACETS, "missing"), "source_files": len(files[name]), "shown_files": 0}) totals = {f: sum(r["facets"][f] == "covered" for r in report.values()) for f in FACETS} - self.record.coverage = {"knowledge": {"owners": report, "covered_by_facet": totals, + self.record.coverage["knowledge"] = {"owners": report, "covered_by_facet": totals, "total_owners": len(order), "unrouted_files": unrouted, - "pin": self.record.pin}} + "pin": self.record.pin} missing = [f"{name}/{facet}" for name, r in report.items() for facet, status in r["facets"].items() if status != "covered"] self.record.unfinished.extend(f"missing knowledge: {item}" for item in missing) @@ -243,23 +302,30 @@ def _build(self, tree: Path) -> InitRecord: full_name=self.lifecycle.full_name, pin=self.record.pin, today=self.today, tags=self.tags, link_page=self._link_page) targets = audit_coverage(self.head, tree, policy, full_name=self.lifecycle.full_name, pin=self.record.pin) + feature_jobs = [] for feature in policy.features: 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 = self._sources(tree, list(dict.fromkeys(list(feature.entry_points) + sorted(paths))), MAX_SOURCE_BYTES) + source_paths = list(dict.fromkeys(list(feature.entry_points) + sorted(paths))) + source = preferred_sources(self, tree, feature.id, source_paths, MAX_SOURCE_BYTES) docs = self._sources(tree, list(feature.docs), MAX_DOC_BYTES) if not source: self.record.unfinished.append(f"feature {feature.id}: missing readable source") continue - offered = {f["path"]: f["end"] for f in source + docs} + offered = offered_ranges(source + docs) missing_facets = targets["features"]["items"][feature.id]["missing_facets"] or list(FACETS) payload = {"repository": self.lifecycle.full_name, "pin": self.record.pin, "owner": "feature-" + feature.id, "feature": feature.title, "facets": missing_facets, "files": source, "docs": docs, "existing_knowledge": self._bounded_context({feature.page: self.head.get(feature.page, "")}), "language_sample": self._language_sample()} + if self.rt.unlimited_subscription: + feature_jobs.append({"owner": Owner("feature-" + feature.id, feature.page, ()), + "page": feature.page, "payload": payload, + "offered": offered, "requested": missing_facets}) + continue try: data = generate(self.rt, self.budget, self.lifecycle.init, system=SYSTEM_KNOWLEDGE, prompt=knowledge_prompt(payload), validate=validate_sections).data @@ -277,6 +343,11 @@ def _build(self, tree: Path) -> InitRecord: break except ModelUnavailable as exc: self.record.unfinished.append(f"feature {feature.id}: unusable draft: {exc}") + if feature_jobs: + for job, results in run_jobs(self, feature_jobs): + for key, text, entries, label in results: + claims[key] = text + evidence.extend(entries) targets = audit_coverage(self.head, tree, policy, full_name=self.lifecycle.full_name, pin=self.record.pin) targets["policy_sha256"] = hashlib.sha256(policy_text.encode()).hexdigest() self.record.coverage["knowledge"]["targets"] = targets @@ -400,7 +471,10 @@ def _section(self, owner: Owner, page: str, data: dict, section: dict, offered: key = f"knowledge:{owner.owner}:{facet}" entries = [] for entry in section["evidence"]: - if entry["path"] not in offered or entry["end"] > offered[entry["path"]]: + 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): self.record.dropped.append({"rule_id": key, "page": page, "why": "evidence outside shown input"}) return None try: @@ -409,20 +483,11 @@ def _section(self, owner: Owner, page: str, data: dict, section: dict, offered: self.record.dropped.append({"rule_id": key, "page": page, "why": str(exc)}) return None title = _one_line(data.get("title")) or f"{owner.owner} knowledge" - heading = _one_line(section.get("title")) or facet.capitalize() - text = neutral_headings(f"## {heading}\n\n" + section["body"].strip()) - if section["interpretation"] == "inference": - text = text.replace("\n\n", "\n\nInference / 设计推断（非作者历史意图）：\n\n", 1) + text = self._render_section(section, entries) problems = check_rules({key: text}, self.observer) if problems: self.record.dropped.append({"rule_id": key, "page": page, "why": "; ".join(problems)}) return None - citations = [] - for entry in entries: - url = (f"https://github.com/{self.lifecycle.full_name}/blob/{self.record.pin}/" - f"{quote(entry.path, safe='/')}#L{entry.start}-L{entry.end}") - citations.append(f"[{entry.path}:L{entry.start}–L{entry.end}]({url})") - text += "\n\nSources / 来源：" + ", ".join(citations) + "\n" front = _page_frontmatter(title, kind="architecture", today=self.today, tags=self.tags) verdict = judge(self.rt, self.budget, self.lifecycle.init, Block("prose", page, "", "prose", hashlib.sha256(text.encode()).hexdigest()), base={}, @@ -435,6 +500,26 @@ def _section(self, owner: Owner, page: str, data: dict, section: dict, offered: if label != "fail": self.record.unfinished.append(f"{owner.owner}/{facet}: {label}; draft retained in model traces for review") return None + return self._append_approved(owner, page, title, facet, text, entries, label) + + def _render_section(self, section, entries): + facet = section["facet"] + heading = _one_line(section.get("title")) or facet.capitalize() + text = neutral_headings(f"## {heading}\n\n" + section["body"].strip()) + if section["interpretation"] == "inference": + text = text.replace("\n\n", "\n\nInference / 设计推断（非作者历史意图）：\n\n", 1) + citations = [] + for entry in entries: + url = (f"https://github.com/{self.lifecycle.full_name}/blob/{self.record.pin}/" + f"{quote(entry.path, safe='/')}#L{entry.start}-L{entry.end}") + citations.append(f"[{entry.path}:L{entry.start}–L{entry.end}]({url})") + text += "\n\nSources / 来源：" + ", ".join(citations) + "\n" + return text + + def _append_approved(self, owner, page, title, facet, text, entries, label="pass"): + """Coordinator assembly of the exact independently approved prose.""" + key = f"knowledge:{owner.owner}:{facet}" + front = _page_frontmatter(title, kind="architecture", today=self.today, tags=self.tags) marker = f"" current = self.head.get(page, front) proposed = current.rstrip() + "\n\n" + marker + "\n\n" + text + "\n" diff --git a/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py b/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py new file mode 100644 index 00000000..f961f449 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py @@ -0,0 +1,386 @@ +"""Bounded foundation workers; only their coordinator changes stage state.""" +from __future__ import annotations + +import copy +import hashlib +import json +import math +from concurrent.futures import ThreadPoolExecutor, as_completed + +from ..knowledge_service.pinned_claims import Evidence, check_evidence +from .init_support import InitError, InitRecord +from .models import ModelGateway, ModelUnavailable + + +def _hash(value): + return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True).encode()).hexdigest() + + +def parallelism(rt): + if not rt.unlimited_subscription: + return 1 + raw = rt.environ.get("KB_KNOWLEDGE_CONCURRENCY", "13") + if str(raw) not in {str(n) for n in range(1, 14)}: + raise InitError("KB_KNOWLEDGE_CONCURRENCY must be an integer from 1 through 13") + return int(raw) + + +def start_interval(rt): + raw = rt.environ.get("KB_KNOWLEDGE_START_INTERVAL_S", "5") + try: + value = float(raw) + except (TypeError, ValueError) as exc: + raise InitError("KB_KNOWLEDGE_START_INTERVAL_S must be finite and positive (at most 60)") from exc + if not math.isfinite(value) or not 0 < value <= 60: + raise InitError("KB_KNOWLEDGE_START_INTERVAL_S must be finite and positive (at most 60)") + return value + + +def offered_ranges(items): + ranges = {} + for item in items: + ranges.setdefault(item["path"], []).append((item.get("start", 1), item["end"])) + return ranges + + +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 + report = stage._discovery_catalog() + if report.get("full_artifact"): + from pathlib import Path + from .init_feature_discovery import verify_full_discovery_report + record = InitRecord.load(stage.rt.state_dir, stage.record.repo, "feature-discovery") + if record is None or record.pin != stage.record.pin or record.discovery.get("done") is not True: + raise InitError("preferred discovery ranges need their genuine completed stage record") + # A copied genuine prerequisite retains its original absolute archive paths. + origin = Path(report["full_artifact"]["path"]).parents[3] + if not hasattr(stage, "_foundation_full_discovery"): + stage._foundation_full_discovery = verify_full_discovery_report(origin, report, record.discovery) + report = stage._foundation_full_discovery + accepted = {r["id"]: r for r in report.get("candidates", []) if r.get("status") == "accepted"} + primary = accepted.get(feature_id) + if not primary or primary.get("relation") != "new": + return stage._sources(tree, paths, limit) + def canonical(row): + target, visited = row.get("related_id"), set() + while target and target not in visited: + if target == feature_id: + return target + visited.add(target) + target = accepted.get(target, {}).get("related_id") + return None + relations = {"implementation_supplement", "alias", "subcapability", "shared_component"} + rows = [primary] + [row for key, row in sorted(accepted.items()) + if row.get("relation") in relations and canonical(row) == feature_id] + out, used, seen = [], 0, set() + for row in rows: + for ref in row.get("evidence", []): + path, start, end = ref["path"], ref["start"], ref["end"] + if path not in paths or (path, start, end) in seen: + continue + if not validate_evidence(stage.source_index, ref): + raise InitError(f"frozen discovery evidence no longer matches {path}:L{start}-L{end}") + lines = stage.source_index.entries[path]["lines"] + text = "\n".join(f"{n}: {lines[n - 1]}" for n in range(start, end + 1)) + if used + len(text.encode()) > limit: + stage.record.unfinished.append(f"feature {feature_id}: discovery range omitted by byte cap: {path}:L{start}-L{end}") + continue # never attest a partially shown range + out.append({"path": path, "start": start, "end": end, "text": text, + "total_lines": len(lines), "language": language_for(path)}) + used += len(text.encode()) + seen.add((path, start, end)) + # Fill with other files; do not pretend gaps in an excerpted file were read. + out.extend(stage._sources(tree, [p for p in paths if p not in {r["path"] for r in out}], max(0, limit - used))) + return out + + +class _CaptureGateway: + def __init__(self, gateway): + self.gateway = copy.copy(gateway) if isinstance(gateway, ModelGateway) else gateway + if isinstance(self.gateway, ModelGateway): + self.gateway._subscription_transports = {} # each authenticated transport belongs to one worker + self.last = {} + + def subscription_billing(self, role): + return self.gateway.subscription_billing(role) + + def call_json(self, role, **kwargs): + reply = self.gateway.call_json(role, **kwargs) + self.last[role.name] = {"native_trace_id": reply.trace_id, + "native_reply_sha256": reply.reply_sha256, + "prompt_sha256": hashlib.sha256(kwargs["prompt"].encode()).hexdigest()} + return reply + + +def _worker(stage, job): + from .init_knowledge import SYSTEM_KNOWLEDGE, knowledge_prompt, validate_sections + from .init_support import generate + from .init_stages import _one_line + worker = copy.copy(stage) + worker.head = dict(stage.head) + 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) + artifacts = [] + error = "" + try: + data = generate(worker.rt, worker.budget, worker.lifecycle.init, system=SYSTEM_KNOWLEDGE, + prompt=knowledge_prompt(job["payload"]), validate=validate_sections).data + for section in data["sections"]: + if section["facet"] not in job["requested"]: + continue + result = worker._section(job["owner"], job["page"], data, section, job["offered"]) + if result: + key, text, entries, label = result + receipt = worker.rt.gateway.last.get("judge", {}) + if isinstance(stage.rt.gateway, ModelGateway) and not receipt.get("native_trace_id"): + worker.record.unfinished.append(f"{key}: native judge receipt missing; retained in traces") + continue + artifacts.append({"key": key, "owner": job["owner"].owner, "page": job["page"], + "title": _one_line(data.get("title")) or f"{job['owner'].owner} knowledge", + "facet": section["facet"], "text": text, + "section": copy.deepcopy(section), "generator_title": data.get("title"), + "approved_today": worker.today, + "text_sha256": hashlib.sha256(text.encode()).hexdigest(), + "evidence": [e.to_dict() for e in entries], "verdict": worker.record.verdicts[key], + "generator_receipt": worker.rt.gateway.last.get("generator", {}), + "judge_receipt": receipt}) + except Exception as exc: + error = f"{type(exc).__name__}: {exc}" + result = {"input_sha256": _hash({"payload": job["payload"], "offered": job["offered"]}), + "payload": job["payload"], "offered": job["offered"], "page": job["page"], + "owner": job["owner"].owner, "requested": job["requested"], + "artifacts": artifacts, "error": error, "verdicts": worker.record.verdicts, + "dropped": worker.record.dropped, "unfinished": worker.record.unfinished} + result["result_sha256"] = _hash(result) + return result + + +def _validate_result(stage, result): + if result.get("result_sha256") != _hash({k: v for k, v in result.items() if k != "result_sha256"}): + raise InitError("foundation checkpoint artifact hash mismatch") + if (result["payload"].get("pin") != stage.record.pin + or result["payload"].get("repository") != stage.lifecycle.full_name): + raise InitError("foundation generator repository or source pin differs") + if result["input_sha256"] != _hash({"payload": result["payload"], "offered": result["offered"]}): + raise InitError("foundation job input hash mismatch") + 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") + for artifact in result["artifacts"]: + facet, owner = artifact["facet"], result["owner"] + if (artifact["owner"] != owner or result["payload"]["owner"] != owner + or artifact["page"] != result["page"] or facet not in result["requested"] + or artifact["key"] != f"knowledge:{owner}:{facet}" + or artifact["section"].get("facet") != facet): + raise InitError("foundation artifact owner/page/facet binding mismatch") + if owner.startswith("feature-"): + feature = next((f for f in stage.coverage_policy.features if "feature-" + f.id == owner), None) + if (feature is None or feature.page != artifact["page"] + or result["payload"].get("feature") != feature.title): + raise InitError("foundation feature differs from frozen catalog") + elif not any(o.owner == owner and stage._page_for(o) == artifact["page"] for o in stage.owners): + raise InitError("foundation owner differs from frozen owner routes") + if artifact["text_sha256"] != hashlib.sha256(artifact["text"].encode()).hexdigest(): + raise InitError("foundation approved text hash mismatch") + if artifact["verdict"].get("verdict") != "pass": + raise InitError("foundation checkpoint contains unapproved prose") + entries = [Evidence.from_dict(e) for e in artifact["evidence"]] + 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"], [])): + 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"]]: + raise InitError("foundation generated evidence binding mismatch") + if check_evidence(entries, stage.observer): + raise InitError("foundation checkpoint pinned evidence mismatch") + if isinstance(stage.rt.gateway, ModelGateway): + _validate_native(stage, artifact, 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 .models import parse_json_object + store = TraceStore(stage.rt.state_dir / "init" / "traces") + receipt = artifact["judge_receipt"] + from .init_feature_discovery import _FeatureDiscovery + for name, role in (("generator", stage.rt.generator), ("judge", stage.rt.judge)): + proof = artifact[name + "_receipt"] + try: + _FeatureDiscovery._verify_archived_receipt(stage, { + "trace_id": proof["native_trace_id"], "reply_sha256": proof["native_reply_sha256"], + "requested": role.label()}, role) + except ModelUnavailable as exc: + raise InitError(f"foundation native approval unavailable: {exc}") from exc + try: + record = store.get(receipt["native_trace_id"]) + prompt = store.blob(record["inputs"]["prompt"]) + reply = parse_json_object(store.blob(record["outputs"]["reply"])) + payload = json.loads(prompt.split("\n", 1)[1].rsplit("", 1)[0]) + generator_receipt = artifact["generator_receipt"] + generator = store.get(generator_receipt["native_trace_id"]) + generator_prompt = store.blob(generator["inputs"]["prompt"]) + generated = parse_json_object(store.blob(generator["outputs"]["reply"])) + gen_payload = json.loads(generator_prompt.split("\n", 1)[1].rsplit("", 1)[0]) + expected_gen_payload = json.loads(knowledge_prompt(result["payload"]).split("\n", 1)[1].rsplit("", 1)[0]) + generator_valid = (gen_payload == expected_gen_payload + 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 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 + and generator["outputs"]["reply"] == "sha256:" + generator_receipt["native_reply_sha256"] + and hashlib.sha256(generator_prompt.encode()).hexdigest() == generator_receipt["prompt_sha256"]) + valid = (generator_valid and not record.get("error") and record["model"]["role"] == "judge" + and record["model"]["provider"] == stage.rt.judge.provider + and record["model"]["model"] == stage.rt.judge.model + and store.blob(record["inputs"]["system"]) == JUDGE_SYSTEM + and record["outputs"]["reply"] == "sha256:" + receipt["native_reply_sha256"] + and hashlib.sha256(prompt.encode()).hexdigest() == receipt["prompt_sha256"] + 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 all(reply["dimensions"].get(d) == "yes" for d in + ("faithful", "does_not_weaken", "non_contradictory"))) + except (KeyError, ValueError, OSError, TypeError) as exc: + raise InitError(f"foundation native approval unavailable: {exc}") from exc + if not valid: + raise InitError("foundation native approval binding mismatch") + + +def _apply(stage, result): + from .init_coverage import Owner + _validate_result(stage, result) + stage.record.verdicts.update(result["verdicts"]) + stage.record.dropped.extend(result["dropped"]) + stage.record.unfinished.extend(result["unfinished"]) + accepted = [] + for artifact in result["artifacts"]: + 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"]]) + if item: + accepted.append(item) + return accepted + + +def validate_checkpoint(stage, previous, digest): + """Read-only replay before cache reuse or any prepared publication write.""" + import tempfile + from pathlib import Path + from .feature_discovery_index import build_discovery_index, discovery_scope + from .init_knowledge_inputs import source_owners + + saved = previous.coverage.get("foundation_jobs", {}) + if (previous.stage != "knowledge" or previous.repo != stage.lifecycle.repo + or previous.pin != stage._knowledge_run_pin + or previous.kb_base_sha != stage._base_sha or previous.inputs_digest != digest + or (saved and saved.get("binding") != digest)): + raise InitError("foundation checkpoint frozen input identity differs") + if not saved.get("tasks"): + return # A genuine empty stage may reuse an already-covered merged KB. + context = copy.copy(stage) + context.record = copy.deepcopy(previous) + context.base = {**stage.rt.knowledge.knowledge_files(previous.kb_base_sha), + **{p[len("knowledge/"):]: text for p, text in stage.overlay.items() + if p.startswith("knowledge/")}} + context.tags = [stage.lifecycle.repo] + context.today = stage.rt.today() + problems = context._precheck() + if problems: + raise InitError("foundation checkpoint policy invalid: " + "; ".join(problems)) + upstream = stage.rt.upstream(stage.lifecycle.repo, stage.lifecycle.full_name) + context.observer = upstream.observer(previous.pin, pull=stage.rt.pull) + # Export and index live only in a temporary directory. Do not repair or + # rewrite stage records, previews, pacing journals or shared index caches. + with tempfile.TemporaryDirectory(prefix="kb-foundation-proof-") as scratch: + tree = upstream.export(previous.pin, Path(scratch) / "tree") + context._inputs(tree) + if context.route_problem: + raise InitError(context.route_problem) + policy = getattr(context, "coverage_policy", None) + docs = list(stage.lifecycle.init.doc_globs) + list(policy.catalog_sources if policy else ()) + index = build_discovery_index(tree, pin=previous.pin, + scope=discovery_scope(stage.lifecycle.init, policy), doc_globs=sorted(set(docs))) + context.owners = list(source_owners(list(index.production), context.owners, policy).values()) + for result in saved["tasks"].values(): + _validate_result(context, result) + + +def restore_jobs(stage): + saved = stage.record.coverage.get("foundation_jobs", {}) + if saved and saved.get("binding") != stage.record.inputs_digest: + raise InitError("foundation checkpoint input identity mismatch") + for result in sorted(saved.get("tasks", {}).values(), key=lambda item: item["sequence"]): + _apply(stage, result) + + +def _validate_fresh(stage, result): + """Keep each valid facet when another native proof cannot be used.""" + empty = {**result, "artifacts": []} + empty["result_sha256"] = _hash({k: v for k, v in empty.items() if k != "result_sha256"}) + _validate_result(stage, empty) + accepted = [] + for artifact in result["artifacts"]: + isolated = {**result, "artifacts": [artifact]} + isolated["result_sha256"] = _hash({k: v for k, v in isolated.items() if k != "result_sha256"}) + try: + _validate_result(stage, isolated) + accepted.append(artifact) + except InitError as exc: + result["unfinished"].append(f"{artifact['key']}: native approval unknown: {exc}") + result["dropped"].append({"rule_id": artifact["key"], "page": artifact["page"], "why": str(exc)}) + result["verdicts"][artifact["key"]] = {**artifact["verdict"], "verdict": "unjudged", + "reasons": {"native_binding": str(exc)}} + result["artifacts"] = accepted + result["result_sha256"] = _hash({k: v for k, v in result.items() if k != "result_sha256"}) + + +def run_jobs(stage, jobs): + """One worker's generator and judges are serial, so the pool is the shared cap.""" + workers = parallelism(stage.rt) + if isinstance(stage.rt.gateway, ModelGateway): + from .depth_pacing import SharedZcodePacer + pacer = stage.rt.gateway._zcode_pacer + if pacer is None: + pacer = SharedZcodePacer(stage.rt.state_dir / "init" / stage.record.repo / "foundation-zcode-pacing.json", + start_interval=start_interval(stage.rt), stop_file=stage.rt.state_dir / "STOP") + pacer.prepare() + stage.rt.gateway.configure_zcode_pacing(pacer) + saved = stage.record.coverage.setdefault("foundation_jobs", {"binding": stage.record.inputs_digest, "tasks": {}}) + 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: + futures = {pool.submit(_worker, stage, job): n for n, job in enumerate(jobs)} + for future in as_completed(futures): + n = futures[future] + result = future.result() + result["sequence"] = sequence + n + result["result_sha256"] = _hash({k: v for k, v in result.items() if k != "result_sha256"}) + try: + _validate_fresh(stage, result) + except InitError as exc: + result["error"] = str(exc) + result["artifacts"] = [] + result["result_sha256"] = _hash({k: v for k, v in result.items() if k != "result_sha256"}) + completed[n] = result + key = jobs[n]["owner"].owner + ":" + result["input_sha256"] + saved["tasks"][key] = result + stage.record.spent_usd = round(stage.budget.spent_usd, 6) + stage.record.save(stage.rt.state_dir) # only this coordinator writes checkpoints + # Completion order never affects pages, navigation or approval assembly. + for n, job in enumerate(jobs): + result = completed[n] + if result["error"]: + stage.record.unfinished.append(f"{job['owner'].owner}: unusable foundation draft: {result['error']}") + yield job, _apply(stage, result) diff --git a/test/test_kb_foundation_parallel.py b/test/test_kb_foundation_parallel.py new file mode 100644 index 00000000..b30a080f --- /dev/null +++ b/test/test_kb_foundation_parallel.py @@ -0,0 +1,348 @@ +"""Foundation workers retain native approval, evidence boundaries and recovery.""" +import copy +import hashlib +import json +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from pathlib import Path + +import pytest + +from infermatrix_copilot.kb_service import init_knowledge_parallel as parallel +from infermatrix_copilot.kb_service.init_budget import Budget, BudgetExhausted +from infermatrix_copilot.kb_service.init_coverage import Owner +from infermatrix_copilot.kb_service.init_knowledge import _Knowledge +from infermatrix_copilot.kb_service.init_stages import run_stage +from infermatrix_copilot.kb_service.init_support import InitError, InitRecord +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_breadth_subscription import SubscriptionKnowledgeGateway +from test_kb_init_knowledge import KnowledgeGateway, _chain +from test_kb_init_modules import _modules_lifecycle +from test_kb_init_skeleton import _runtime, _tree, world # noqa: F401 + + +def test_pool_shared_cap_deterministic_assembly_and_single_failure(tmp_path, monkeypatch): + lock, active, peak = threading.Lock(), 0, 0 + completed, applied, saves = [], [], [] + stage = SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=True, environ={}, state_dir=tmp_path, gateway=None), + record=InitRecord(stage="knowledge", repo="toy", inputs_digest="identity"), budget=Budget(None)) + jobs = [{"owner": Owner(f"owner-{i}", "same-page", ()), "payload": {}, "offered": {}} for i in range(32)] + + def worker(stage, job): + nonlocal active, peak + index = int(job["owner"].owner.split("-")[-1]) + with lock: + active += 1 + peak = max(peak, active) + # Generator followed by judges remains inside the same shared worker slot. + with stage.budget.reserve(1): + time.sleep((13 - index % 13) / 500) + with lock: + active -= 1 + completed.append(index) + return {"input_sha256": str(index), "error": "unavailable" if index == 4 else "", "index": index} + + monkeypatch.setattr(parallel, "_worker", worker) + monkeypatch.setattr(parallel, "_validate_result", lambda stage, result: None) + monkeypatch.setattr(parallel, "_validate_fresh", lambda stage, result: None) + monkeypatch.setattr(parallel, "_apply", lambda stage, result: applied.append(result["index"]) or []) + monkeypatch.setattr(InitRecord, "save", lambda record, path: saves.append(threading.current_thread().name)) + list(parallel.run_jobs(stage, jobs)) + assert 1 < peak <= 13 and stage.budget.spent_usd == 32 and stage.budget._reserved == 0 + assert completed != list(range(32)) and applied == list(range(32)) + assert len(saves) == 32 and set(saves) == {threading.current_thread().name} + assert any("owner-4" in item for item in stage.record.unfinished) + + +@pytest.mark.parametrize("value", ["0", "14", "1.5", "01", "bad"]) +def test_invalid_parallel_config_and_default_serial(value): + rt = SimpleNamespace(unlimited_subscription=True, environ={"KB_KNOWLEDGE_CONCURRENCY": value}) + with pytest.raises(InitError, match="1 through 13"): + parallel.parallelism(rt) + rt.unlimited_subscription = False + assert parallel.parallelism(rt) == 1 + + +def test_budget_atomic_ceiling_and_failure_drain(): + budget = Budget(3) + barrier = threading.Barrier(3) + accepted = [] + def call(): + try: + with budget.reserve(1): + accepted.append(1) + barrier.wait(timeout=2) + raise RuntimeError("native failed") + except (RuntimeError, BudgetExhausted): + pass + with ThreadPoolExecutor(max_workers=6) as pool: + list(pool.map(lambda _: call(), range(6))) + assert len(accepted) == 3 and budget.spent_usd == 3 and budget._reserved == 0 + + +def test_late_discovery_ranges_are_complete_and_gaps_are_not_offered(tmp_path): + lines = [f"line {i}" for i in range(1, 20001)] + (tmp_path / "giant.ts").write_text("\n".join(lines)) + refs = [{"path": "giant.ts", "start": 7000, "end": 7010, "pin": "a" * 40}, + {"path": "giant.ts", "start": 19000, "end": 19005, "pin": "a" * 40}] + stage = _Knowledge.__new__(_Knowledge) + stage.record = InitRecord("knowledge", "toy", pin="a" * 40) + stage.source_index = SimpleNamespace(identity={"pin": "a" * 40}, entries={ + "giant.ts": {"status": "ready", "lines": lines, "sha256": "digest"}}) + stage._frozen_discovery = {"candidates": [{"id": "new-feature", "relation": "new", "status": "accepted", "evidence": refs}]} + shown = parallel.preferred_sources(stage, tmp_path, "new-feature", ["giant.ts"], 100000) + assert [(item["start"], item["end"]) for item in shown] == [(7000, 7010), (19000, 19005)] + assert "19000: line 19000" in shown[1]["text"] + ranges = parallel.offered_ranges(shown) + stage._section(Owner("new", "page", ()), "page", {}, {"facet": "api", "evidence": [ + {"path": "giant.ts", "start": 7010, "end": 19000}]}, ranges) + assert stage.record.dropped[0]["why"] == "evidence outside shown input" + old = parallel.preferred_sources(stage, tmp_path, "old-feature", ["giant.ts"], 100) + assert old[0]["end"] < 7000 # legacy source selection is untouched + tiny = parallel.preferred_sources(stage, tmp_path, "new-feature", ["giant.ts"], 10) + assert all(item["end"] < 7000 for item in tiny) and len(stage.record.unfinished) == 2 + refs[0]["pin"] = "b" * 40 + with pytest.raises(InitError, match="no longer matches"): + parallel.preferred_sources(stage, tmp_path, "new-feature", ["giant.ts"], 100000) + + +def _complete_native(world, *, publish=False): + seed_gateway = KnowledgeGateway() + _chain(world, seed_gateway) + if publish: + from test_kb_init_skeleton import _commit + skeleton = InitRecord.load(_runtime(world).state_dir, "toy", "skeleton") + modules = InitRecord.load(_runtime(world).state_dir, "toy", "modules") + _commit(world["origin"], {**_tree(skeleton), **_tree(modules)}, "merge prerequisite pages") + state = _runtime(world).state_dir if not publish else world["tmp"] / "publish-state" + traces = TraceStore(state / "init" / "traces") + offline = KnowledgeGateway() + + class NativeTransport: + subscription_billing = True + supports_native_events = True + def complete(self, *, system, messages, model, effort, role, **kwargs): + result = offline.call_json(ModelRole(role, "zcode" if role == "generator" else "codex", model), + system=system, prompt=messages[0]["content"]) + if role == "generator": + prototype = result.data["sections"][0] + for facet in ("api", "configuration", "features", "validation"): + result.data["sections"].append({**prototype, "facet": facet}) + text = json.dumps(result.data) + else: + text = result.text + return Reply(blocks=[Block(type="text", text=text)], model=model) + + gateway = ModelGateway(None, transport_factory=lambda _: NativeTransport(), recorder=trace_recorder(traces)) + rt = _runtime(world, gateway, state_dir=state, generator=ModelRole("generator", "zcode", "GLM-5.3")) + if publish: + from test_kb_init_skeleton import FakeGh + rt.gh_run = FakeGh(fail_create=1) + rt.environ = {**rt.environ, "KB_KNOWLEDGE_CONCURRENCY": "2", "KB_KNOWLEDGE_START_INTERVAL_S": "0.001"} + if publish: + rt.environ.update(ALLOW_PUSH="1", ALLOW_POST="1", KB_INIT_GIT_AUTHOR="Tester ") + record = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=not publish, + from_existing=publish, unlimited_subscription=True) + assert record.status == ("blocked" if publish else "dry_run"), record.problems + assert not record.unfinished + return rt, record, offline, traces + + + +@pytest.mark.parametrize("damage", ["judge", "generator", "owner", "facet", "offered", "section"]) +def test_native_checkpoint_replay_exact_approval_and_tamper_rejected(world, damage): + rt, record, offline, traces = _complete_native(world) + artifacts = [a for task in record.coverage["foundation_jobs"]["tasks"].values() for a in task["artifacts"]] + assert len(artifacts) == 12 and all(a["judge_receipt"]["native_trace_id"] for a in artifacts) + first_tree = _tree(record) + call_count = len(offline.calls) + # Genuine saved worker approvals reconstruct exact pages without native rejudging. + record.status = "started" + record.save(rt.state_dir) + second = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + assert _tree(second) == first_tree and len(offline.calls) == call_count + # Crossing a day boundary keeps the exact original native approval usable. + second.status = "started" + second.save(rt.state_dir) + rt.clock = lambda: 1_790_000_000.0 + 86400 + next_day = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + assert len(offline.calls) == call_count and len(next_day.evidence) == 12 + saved = copy.deepcopy(record.coverage["foundation_jobs"]) + task = next(iter(saved["tasks"].values())) + artifact = task["artifacts"][0] + if damage in ("judge", "generator"): + artifact[damage + "_receipt"]["native_reply_sha256"] = "0" * 64 + elif damage == "owner": + artifact["owner"] = "another-owner" + elif damage == "facet": + artifact["facet"] = "api" if artifact["facet"] != "api" else "features" + elif damage == "offered": + task["offered"][artifact["evidence"][0]["path"]] = [[1, 100000]] + task["input_sha256"] = parallel._hash({"payload": task["payload"], "offered": task["offered"]}) + else: + artifact["section"]["interpretation"] = "inference" + task["result_sha256"] = parallel._hash({k: v for k, v in task.items() if k != "result_sha256"}) + second.coverage["foundation_jobs"] = saved + second.status = "started" + second.save(rt.state_dir) + with pytest.raises(InitError, match="foundation"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + assert len(offline.calls) == call_count + + +def test_compact_genuine_handoff_uses_original_archive_for_late_ranges(tmp_path): + from infermatrix_copilot.kb_service.init_feature_discovery import compact_discovery_report, write_full_discovery_report + from test_kb_feature_discovery_report import full_report + stage = _Knowledge.__new__(_Knowledge) + stage.record = InitRecord("knowledge", "toy", pin="a" * 40) + worker_state, origin_state = tmp_path / "worker", tmp_path / "original" + stage.rt = SimpleNamespace(state_dir=worker_state) + lines = ["readable"] * 20000 + stage.source_index = SimpleNamespace(identity={"pin": "a" * 40}, entries={ + "giant.ts": {"status": "ready", "lines": lines, "sha256": "digest"}}) + full = full_report() + full["repo"] = "toy" + full["candidates"] = [{"id": "late", "relation": "new", "status": "accepted", "evidence": [ + {"path": "giant.ts", "start": 7000, "end": 7002, "pin": "a" * 40}]}, + {"id": "frontend", "relation": "implementation_supplement", "related_id": "late", "status": "accepted", "evidence": [ + {"path": "giant.ts", "start": 19000, "end": 19002, "pin": "a" * 40}]}, + {"id": "alias", "relation": "alias", "related_id": "frontend", "status": "accepted", "evidence": [ + {"path": "giant.ts", "start": 19500, "end": 19502, "pin": "a" * 40}]}] + artifact = write_full_discovery_report(origin_state, full) + compact = compact_discovery_report(full, artifact) + stage._frozen_discovery = compact + genuine = InitRecord("feature-discovery", "toy", pin="a" * 40, status="published", discovery={ + **{k: full[k] for k in ("pin", "catalog_sha256", "index_sha256", "run_config")}, + "done": True, "report_format": compact["report_format"], "full_artifact": artifact}) + genuine.save(worker_state) + assert "candidates" not in compact + shown = parallel.preferred_sources(stage, tmp_path, "late", ["giant.ts"], 100000) + assert [(item["start"], item["end"]) for item in shown] == [(7000, 7002), (19500, 19502), (19000, 19002)] + assert Path(artifact["path"]).is_relative_to(origin_state) + del stage._foundation_full_discovery + genuine.discovery["full_artifact"] = {**artifact, "sha256": "0" * 64} + genuine.save(worker_state) + with pytest.raises(InitError, match="artifact bindings differ"): + parallel.preferred_sources(stage, tmp_path, "late", ["giant.ts"], 100000) + + +def test_titleless_section_uses_facet_heading_in_serial_mode(world): + gateway = KnowledgeGateway() + _chain(world, gateway) + original = gateway.call_json + def no_title(role, **kwargs): + reply = original(role, **kwargs) + if "sections" in reply.data: + for section in reply.data["sections"]: + section.pop("title", None) + return reply + gateway.call_json = no_title + record = run_stage(_runtime(world, gateway), _modules_lifecycle(), "knowledge", dry_run=True) + assert record.status == "dry_run" and len(record.evidence) == 4 + assert any("**Architecture**" in text for text in _tree(record).values()) + + +def test_fresh_bad_artifact_preserves_other_approved_facets(monkeypatch): + result = {"artifacts": [{"key": "good", "verdict": {"verdict": "pass"}}, + {"key": "bad", "page": "page", "verdict": {"verdict": "pass"}}], + "unfinished": [], "dropped": [], "verdicts": {}} + def validate(stage, isolated): + if any(item["key"] == "bad" for item in isolated["artifacts"]): + raise InitError("native archive incomplete") + monkeypatch.setattr(parallel, "_validate_result", validate) + parallel._validate_fresh(None, result) + assert [item["key"] for item in result["artifacts"]] == ["good"] + assert result["verdicts"]["bad"]["verdict"] == "unjudged" + assert "native archive incomplete" in result["unfinished"][0] + + +def test_shared_page_assembly_preserves_existing_sections_and_both_owners(): + from infermatrix_copilot.kb_service.init_stages import _page_frontmatter + from infermatrix_copilot.knowledge_service.pinned_claims import Evidence + stage = _Knowledge.__new__(_Knowledge) + stage.repo_dir = "repos/toy" + stage.today, stage.tags = "2026-10-06", ["toy"] + stage.lifecycle = SimpleNamespace(full_name="o/toy") + stage.record = InitRecord("knowledge", "toy", pin="a" * 40) + page = "repos/toy/components/shared/knowledge.md" + historical = "Existing accepted explanation remains byte-for-byte." + stage.head = {page: _page_frontmatter("Shared", kind="architecture", today=stage.today, tags=stage.tags) + historical + "\n"} + for owner, facet in (("left", "architecture"), ("right", "api")): + result = stage._append_approved(Owner(owner, page, ()), page, "Shared", facet, + f"**{facet}**\n\n{owner} independently approved body.\n", [Evidence("pkg/a.py", 1, 1, "b" * 64)]) + assert result is not None + assert historical in stage.head[page] + assert stage.head[page].index("owner=left") < stage.head[page].index("owner=right") + assert len(stage.record.evidence) == 2 + + +@pytest.mark.parametrize("damage", ["missing_generator", "missing_judge", "tampered_generator", "tampered_judge", "binding", "config", "role"]) +def test_completed_cache_cannot_skip_proof_replay_or_overwrite_on_failure(world, damage): + rt, record, offline, traces = _complete_native(world) + assert record.status == "dry_run" and record.unfinished == [] + artifact = next(iter(record.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0] + if damage.startswith("missing_") or damage.startswith("tampered_"): + role = damage.rsplit("_", 1)[1] + native = traces.get(artifact[role + "_receipt"]["native_trace_id"]) + attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json" + if damage.startswith("missing_"): + attempt.unlink() + else: + saved = json.loads(attempt.read_text()) + saved["status"] = "interrupted" + attempt.write_text(json.dumps(saved)) + elif damage == "binding": + record.coverage["foundation_jobs"]["binding"] = "different-inputs" + record.save(rt.state_dir) + elif damage == "config": + rt.gateway._settings = SimpleNamespace(zcode_reasoning_level="low") + else: + rt.judge = ModelRole("judge", "codex", "different-model", "medium") + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before, calls = path.read_bytes(), len(offline.calls) + with pytest.raises(InitError, match="foundation"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=True, unlimited_subscription=True) + assert path.read_bytes() == before and len(offline.calls) == calls + + +@pytest.mark.parametrize("damage", ["missing_generator", "missing_judge", "digest", "role", "config"]) +def test_prepared_publication_checks_original_proofs_before_resume(world, damage): + rt, record, offline, traces = _complete_native(world, publish=True) + assert record.pr["prepared"] and record.status == "blocked" and not record.unfinished + artifact = next(iter(record.coverage["foundation_jobs"]["tasks"].values()))["artifacts"][0] + restore = None + if damage.startswith("missing_"): + role = damage.rsplit("_", 1)[1] + native = traces.get(artifact[role + "_receipt"]["native_trace_id"]) + attempt = traces.root / "attempts" / native["result"]["native_attempt_id"] / "attempt.json" + restore = attempt, attempt.read_bytes() + attempt.unlink() + elif damage == "digest": + record.inputs_digest = "wrong-inputs" + record.save(rt.state_dir) + elif damage == "role": + rt.judge = ModelRole("judge", "codex", "different-model", "medium") + else: + rt.gateway._settings = SimpleNamespace(zcode_reasoning_level="low") + path = InitRecord.path(rt.state_dir, "toy", "knowledge") + before, calls, pushes = path.read_bytes(), len(offline.calls), len(rt.gh_run.pushed) + with pytest.raises(InitError, match="foundation"): + run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True) + assert path.read_bytes() == before and len(offline.calls) == calls and len(rt.gh_run.pushed) == pushes + if restore: + restore[0].write_bytes(restore[1]) + resumed = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True) + assert resumed.status == "published" and len(offline.calls) == calls + # A previously published record remains immutable historical output, + # including retained limitations; archive loss cannot trigger a rerun. + resumed.unfinished.append("retained historical scan limitation") + resumed.save(rt.state_dir) + restore[0].unlink() + published_bytes = path.read_bytes() + cached = run_stage(rt, _modules_lifecycle(), "knowledge", dry_run=False, from_existing=True, unlimited_subscription=True) + assert cached.status == "published" and path.read_bytes() == published_bytes and len(offline.calls) == calls