From 47371bbf6700e1e6e580b98fc1e0c4c93d2b567e Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 6 Oct 2026 05:55:29 +0800 Subject: [PATCH] fix(kb): review foundation with its full offered evidence packet --- .../kb_service/init_knowledge.py | 19 +- .../kb_service/init_knowledge_inputs.py | 62 +++++ .../kb_service/init_knowledge_parallel.py | 46 +++- test/test_kb_foundation_context.py | 225 ++++++++++++++++++ test/test_kb_init_knowledge.py | 3 +- 5 files changed, 340 insertions(+), 15 deletions(-) create mode 100644 test/test_kb_foundation_context.py diff --git a/src/infermatrix_copilot/kb_service/init_knowledge.py b/src/infermatrix_copilot/kb_service/init_knowledge.py index b75ed535..9660868a 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 FACETS = ("architecture", "api", "configuration", "tradeoffs", "features", "validation") MAX_SOURCE_BYTES = 100_000 @@ -120,7 +120,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)} @@ -247,6 +248,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 @@ -330,6 +332,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}) @@ -480,9 +483,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 +511,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_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..1e5c0aa7 100644 --- a/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py +++ b/src/infermatrix_copilot/kb_service/init_knowledge_parallel.py @@ -43,6 +43,24 @@ def offered_ranges(items): return ranges +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 +132,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 +141,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 +190,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 +218,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 +232,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 +262,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 +277,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: 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_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