fix(kb): include related test assertions in foundation evidence

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,22 +232,26 @@ 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),
                        "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:
+                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()

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.