fix(kb): use bounded review feedback on foundation resumes

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()

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.
 

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,18 +132,22 @@ 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)
     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)
+    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:
@@ -323,6 +353,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 +379,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 +449,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: