diff --git a/src/infermatrix_copilot/kb_service/init_knowledge.py b/src/infermatrix_copilot/kb_service/init_knowledge.py index b75ed535..292f8aa2 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 @@ -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)} @@ -231,15 +232,18 @@ 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) + 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 +251,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 @@ -317,7 +322,10 @@ def _build(self, tree: Path) -> InitRecord: 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 +338,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 +489,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 +517,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..110723ab 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,115 @@ def offered_ranges(items): return ranges +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 +225,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 +234,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 +283,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 +311,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 +325,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 +355,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 +370,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: @@ -323,6 +446,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 +472,64 @@ 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"]]} + 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 +542,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/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_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_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