fix(harness): save whether each proof's solution ran and report proved counts (TH-8752)

src/fi/alk/harness/builder.py

@@ -162,7 +162,7 @@ def suite_server(
 ) -> ToolServer:
     """Whole-suite operations the single-scenario tools would take one call per scenario to do."""
     from .catalogue import load_catalogue
-    from .proofs import record_proof
+    from .proofs import proof_key, record_proof
     from .prove import prove
     from .scenario_tools import write_scenarios
 
@@ -205,7 +205,7 @@ async def prove_scenarios(args: dict[str, Any]) -> dict[str, Any]:
                 lines.append(f"{name}: not in the suite")
                 continue
             proof = prove(scenario, catalogue, out)
-            record_proof(out, scenario.scenario_key, proof)
+            record_proof(out, proof_key(scenario), proof, solution_steps=len(scenario.solution))
             gates = ", ".join(f"{gate} {'yes' if held else 'no'}" for gate, held in proof.gates().items())
             lines.append(
                 f"{name}: {'holds' if proof.holds else 'does not hold'} ({gates})"

src/fi/alk/harness/hosted_chat_entrypoint.py

@@ -6,6 +6,7 @@
 import traceback
 import json
 import os
+import re
 import signal
 import time
 import uuid
@@ -50,6 +51,79 @@
 _NOT_MESSAGES = {"interrupt", "cancel_operation", "user_response", "approval"}
 _PENDING_QUESTION = ".futureagi-pending-question.json"
 _BUILD_FINISHED = ".futureagi-build-finished.json"
+
+
+def _proof_summary(proofs: dict[str, dict[str, Any]], keys: list[str], counts: dict[str, int]) -> str:
+    """The proof counts the closing summary must repeat, with the tools and reasons behind them."""
+    tools = sorted(
+        {
+            tool
+            for key in keys
+            if (proofs.get(key) or {}).get("kept_as_written") is True
+            for tool in (proofs.get(key) or {}).get("assumed") or []
+        }
+    )
+    reasons = sorted(
+        {
+            str((proofs.get(key) or {}).get("blocked_reason") or "not_validated")
+            for key in keys
+            if (proofs.get(key) or {}).get("proved") is not True
+            and (proofs.get(key) or {}).get("kept_as_written") is not True
+        }
+    )
+    kept = f"{counts['kept_as_written']} kept as written" + (f" ({', '.join(tools)})" if tools else "")
+    unproved = f"{counts['not_proved']} not proved" + (f" ({', '.join(reasons)})" if reasons else "")
+    return (
+        f"{counts['proved']} proved, {kept}, {unproved}. Report exactly these counts; call only the "
+        f"{counts['proved']} proved scenarios proved."
+    )
+_NUMBER_PROVED = re.compile(
+    r"\b(\d+)\s+((?:[A-Za-z-]+\s+){0,2})proved\b", re.IGNORECASE
+)
+_ALL_PROVED = re.compile(r"\b(?:all|every)\b[^.]{0,40}\bproved\b", re.IGNORECASE)
+
+
+def _proof_claim_mismatch(text: str, counts: dict[str, int]) -> str | None:
+    """A correction when the reply claims more scenarios proved than the proofs hold, else None."""
+    proved = int(counts.get("proved") or 0)
+    total = (
+        proved
+        + int(counts.get("kept_as_written") or 0)
+        + int(counts.get("not_proved") or 0)
+    )
+    claims = [
+        int(match.group(1))
+        for match in _NUMBER_PROVED.finditer(text)
+        if not re.search(r"\b(?:not|un)", match.group(2), re.IGNORECASE)
+    ]
+    overclaimed = any(claim != proved for claim in claims) or (
+        proved < total
+        and any(
+            not re.search(
+                r"\bnot\b",
+                text[max(0, match.start() - 12) : match.start()],
+                re.IGNORECASE,
+            )
+            for match in _ALL_PROVED.finditer(text)
+        )
+    )
+    return (
+        f"Correction: only {proved} of {total} scenarios are proved."
+        if overclaimed
+        else None
+    )
+
+
+def _with_proof_counts(text: str, counts: dict[str, int]) -> str:
+    """The closing reply with the proof counts written by code, and a correction if the reply overclaims."""
+    correction = _proof_claim_mismatch(text, counts)
+    line = (
+        f"Proof counts: {int(counts.get('proved') or 0)} proved, {int(counts.get('kept_as_written') or 0)} kept as "
+        f"written, {int(counts.get('not_proved') or 0)} not proved."
+    )
+    return "\n\n".join(part for part in (text, line, correction) if part)
+
+
 _PROVIDER_PROFILE = "provider-import-profile.json"
 _JOURNAL = ".futureagi-conversation.json"
 _PROVIDER_SESSIONS = ".futureagi-provider-sessions.json"
@@ -849,6 +923,7 @@ async def remember(args: dict[str, Any]) -> dict[str, Any]:
         )
         async def finish_build(args: dict[str, Any]) -> dict[str, Any]:
             from .builder import suite_health
+            from .proofs import load_proofs, proof_counts, proof_key
             from .scenario_tools import load_scenarios
             from .understand import load as load_contract
 
@@ -876,12 +951,19 @@ async def finish_build(args: dict[str, Any]) -> dict[str, Any]:
                 return text("Not finished:\n- " + "\n- ".join(missing[:60]), error=True)
             # The bundle is sealed from the world as it is now, after every change made during the build.
             await asyncio.to_thread(self._after_saved, world_changed=True)
+            proofs = load_proofs(self.workspace)
+            keys = [proof_key(one) for one in kept]
+            counts = proof_counts(proofs, keys)
             (self.workspace / _BUILD_FINISHED).write_text(
-                json.dumps({"scenarios": len(kept), "asked": asked, "note": str(args.get("note") or "")}),
+                json.dumps(
+                    {"scenarios": len(kept), "asked": asked, "note": str(args.get("note") or ""), "proofs": counts}
+                ),
                 encoding="utf-8",
             )
             return text(
-                f"Finished: {len(kept)} scenarios (the build asked for {asked}). It is now being "
+                f"Finished: {len(kept)} scenarios (the build asked for {asked}): "
+                + _proof_summary(proofs, keys, counts)
+                + " It is now being "
                 "validated against the running agent and sealed; the environment opens when that "
                 "is done, and this conversation carries on there."
             )
@@ -1476,6 +1558,7 @@ async def _handle_command(self, command: dict[str, Any]) -> None:
         )
         emitted_text: list[str] = []
         turn_failures: list[str] = []
+        finished_before = self._build_finished_stamp()
         queue: asyncio.Queue[Event | None] = asyncio.Queue()
 
         def receive(event: Event) -> None:
@@ -1615,6 +1698,7 @@ def receive(event: Event) -> None:
             ).strip()
         self._last_health = ""
         self._report_again_if_written()
+        final_text = self._close_with_proof_counts(final_text, finished_before)
         if final_text:
             await self.client.emit(
                 "assistant_message",
@@ -1646,6 +1730,26 @@ def receive(event: Event) -> None:
         if self._restart_after_turn:
             self.stopping.set()
 
+    def _build_finished_stamp(self) -> int | None:
+        """When the build-finished file was last written, or None when there is none."""
+        try:
+            return (self.workspace / _BUILD_FINISHED).stat().st_mtime_ns
+        except OSError:
+            return None
+
+    def _close_with_proof_counts(self, text: str, finished_before: int | None) -> str:
+        """Append code-written proof counts when this turn finished the build, so the reply cannot overclaim."""
+        stamp = self._build_finished_stamp()
+        if stamp is None or stamp == finished_before:
+            return text
+        try:
+            counts = json.loads(
+                (self.workspace / _BUILD_FINISHED).read_text(encoding="utf-8")
+            ).get("proofs")
+        except (OSError, ValueError, AttributeError):
+            return text
+        return _with_proof_counts(text, counts) if isinstance(counts, dict) else text
+
     async def _fail_turn(
         self,
         journal: dict[str, str],

src/fi/alk/harness/nudge.py

@@ -415,7 +415,7 @@ def suite_gate(
             record_unprovable(workspace, key, type(exc).__name__)
             holds, why = False, type(exc).__name__
         else:
-            record_proof(workspace, key, proof)
+            record_proof(workspace, key, proof, solution_steps=len(scenario.solution))
             if proof.deferred and not proof.broken:
                 # Only the live runtime can prove it; that is not a failure.
                 not_checked.append(key)

src/fi/alk/harness/proofs.py

@@ -41,29 +41,86 @@ def _now() -> str:
     return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
 
 
-def record_proof(destination: Path, scenario_key: str, proof: Proof) -> None:
-    """Save the three gates for one scenario, replacing its earlier proof."""
-    proofs = load_proofs(destination)
-    proofs[scenario_key] = {
+def proof_key(scenario: Any) -> str:
+    """The key every proof of a scenario is saved under."""
+    from .scenario import _slug
+
+    return getattr(scenario, "scenario_key", "") or _slug(scenario.name)
+
+
+def _entry(proof: Proof, solution_steps: int) -> dict[str, Any]:
+    blocked = proof.blocked_reason()
+    if proof.proved:
+        why = ""
+    elif not proof.holds:
+        why = proof.why()
+    elif proof.assumed:
+        why = (
+            "kept as written: these tools have no endpoint here, so their solution steps never ran: "
+            + ", ".join(proof.assumed)
+        )
+    else:
+        why = proof.why()
+    return {
+        "format": 2,
         "holds": proof.holds,
+        "proved": proof.proved,
+        "kept_as_written": proof.kept_as_written,
         "ready": proof.ready,
         "solvable": proof.solvable,
         "not_vacuous": not proof.vacuous,
-        "why": "" if proof.holds else proof.why(),
+        "why": why,
         "proved_at": _now(),
+        "assumed": list(proof.assumed),
+        "solution_steps": solution_steps,
+        "deferred": proof.deferred,
+        "judged_only": proof.judged_only,
+        "blocked_reason": blocked,
     }
-    _save(destination, proofs)
 
 
-def record_unprovable(destination: Path, scenario_key: str, why: str) -> None:
-    """Save that a scenario could not be proved at all."""
+def record_proof(destination: Path, scenario_key: str, proof: Proof, *, solution_steps: int = 0) -> None:
+    """Save one scenario's proof, replacing its earlier one: what held, what ran and what was assumed."""
     proofs = load_proofs(destination)
-    proofs[scenario_key] = {
+    proofs[scenario_key] = _entry(proof, solution_steps)
+    _save(destination, proofs)
+
+
+def unprovable_entry(why: str, blocked_reason: str = "broken_check") -> dict[str, Any]:
+    return {
+        "format": 2,
         "holds": False,
+        "proved": False,
+        "kept_as_written": False,
         "ready": False,
         "solvable": False,
         "not_vacuous": False,
         "why": why,
         "proved_at": _now(),
+        "assumed": [],
+        "solution_steps": 0,
+        "deferred": False,
+        "judged_only": False,
+        "blocked_reason": blocked_reason,
     }
+
+
+def record_unprovable(destination: Path, scenario_key: str, why: str) -> None:
+    """Save that a scenario could not be proved at all."""
+    proofs = load_proofs(destination)
+    proofs[scenario_key] = unprovable_entry(why)
     _save(destination, proofs)
+
+
+def proof_counts(proofs: dict[str, dict[str, Any]], keys: list[str]) -> dict[str, int]:
+    """Proved, kept as written and not proved, over `keys`. A key with no proof is not proved."""
+    counts = {"proved": 0, "kept_as_written": 0, "not_proved": 0}
+    for key in keys:
+        entry = proofs.get(key) or {}
+        if entry.get("proved") is True:
+            counts["proved"] += 1
+        elif entry.get("kept_as_written") is True:
+            counts["kept_as_written"] += 1
+        else:
+            counts["not_proved"] += 1
+    return counts

src/fi/alk/harness/prove.py

@@ -41,6 +41,10 @@
 from .world.snapshot import read_manifest, restore
 
 
+# The lane could not execute something; the case runs labelled "kept as written", never as proved.
+LANE_CODES = frozenset({"proof_deferred", "solution_assumed", "control_not_run"})
+
+
 @dataclass
 class Proof:
     """Whether a scenario holds up, and what happened when it was tried."""
@@ -87,6 +91,44 @@ def holds(self) -> bool:
             and not self.unchecked
         )
 
+    @property
+    def proved(self) -> bool:
+        """Holds, and the reference solution really ran: nothing assumed, nothing deferred."""
+        return self.holds and not self.assumed and not self.deferred
+
+    @property
+    def kept_as_written(self) -> bool:
+        """Not proved only because this lane could not execute something; nothing that ran failed."""
+        return not self.proved and self.blocked_reason() in LANE_CODES
+
+    def blocked_reason(self) -> str | None:
+        """The first reason this proof is not proved, faults before the lane's own limits."""
+        if self.proved:
+            return None
+        if self.broken:
+            return "broken_check"
+        if not self.deferred:
+            # A deferred proof returns before any gate runs, so its gates read unset, not failed.
+            if not self.ready:
+                return "not_ready"
+            if not self.solvable:
+                return "not_solvable"
+            if self.vacuous:
+                return "vacuous"
+            if self.weak:
+                return "weak_check"
+            if self.failed_attempts:
+                return "credits_failed_attempt"
+            if self.too_strict:
+                return "too_strict"
+            if self.unchecked:
+                return "unchecked"
+        if self.deferred:
+            return "proof_deferred"
+        if self.assumed:
+            return "solution_assumed"
+        return None
+
     def gates(self) -> dict[str, bool]:
         """The three answers, for anything that wants to show them."""
         return {

src/fi/alk/harness/scenario_tools.py

@@ -53,7 +53,7 @@
     write_index,
 )
 from .checks import reference_of, run_check
-from .proofs import record_proof, record_unprovable
+from .proofs import proof_key, record_proof, record_unprovable
 from .prove import (
     executes,
     native_runtime,
@@ -173,10 +173,10 @@ def _no_longer_hold(
         try:
             proof = prove(scenario, catalogue, world_root)
         except Exception as exc:  # noqa: BLE001 - a scenario that cannot be proved is the finding
-            record_unprovable(destination, scenario.scenario_key, type(exc).__name__)
+            record_unprovable(destination, proof_key(scenario), type(exc).__name__)
             failed.append((scenario.name, type(exc).__name__))
             continue
-        record_proof(destination, scenario.scenario_key, proof)
+        record_proof(destination, proof_key(scenario), proof, solution_steps=len(scenario.solution))
         if not proof.holds and not (proof.deferred and not proof.broken):
             why = "not ready" if not proof.ready else (
                 "the reference solution no longer passes it"
@@ -1106,7 +1106,7 @@ def accept_scenario(
     # Saved on both paths: parallel writers share this process, so the proof file is safe to
     # write even where the folders are not. A rename is not proved again: its key keeps its proof.
     if proof is not None:
-        record_proof(world_root, scenario.scenario_key or _slug(scenario.name), proof)
+        record_proof(world_root, proof_key(scenario), proof, solution_steps=len(scenario.solution))
     if persist:
         write_scenarios(kept, world_root, catalogue)
     else:

src/fi/alk/harness/skills/builder/SKILL.md

@@ -132,6 +132,12 @@ yourself. When you mention a scenario, write its number and name, `#12 some_name
 find it. A message can start with a block naming rows the person selected on the Scenarios tab; that
 selection is what "these" and "the selected ones" mean.
 
+## Finishing a build
+
+`finish_build` reports how many scenarios are proved, kept as written and not proved. Say 'proved' only
+for the count finish_build reports as proved; call the rest 'kept as written' or 'not proved', with the
+reason it gives.
+
 ## Changing the environment
 
 Load `write-scenarios` before you write or rewrite any scenario in this conversation; its rules are