Harden verified completion loop

README.md

@@ -2,8 +2,8 @@
 
 # Loop
 
-**Hand off a goal and an acceptance contract. Loop works in isolation, re-runs
-the evidence, and returns a Receipt anyone can replay.**
+**Give Loop one instruction. It keeps working, testing, and repairing its own
+result until it can prove the work is ready to deliver.**
 
 [![CI](https://github.com/chriswu727/loop-agent/actions/workflows/ci.yml/badge.svg)](https://github.com/chriswu727/loop-agent/actions/workflows/ci.yml)
 [![Desktop](https://github.com/chriswu727/loop-agent/actions/workflows/desktop.yml/badge.svg)](https://github.com/chriswu727/loop-agent/actions/workflows/desktop.yml)
@@ -18,36 +18,60 @@ the evidence, and returns a Receipt anyone can replay.**
   <img src="./docs/images/task.png" alt="A completed Loop task with re-executed checks, a verified Receipt, and downloadable artifacts" width="840" />
 </p>
 
-## What makes Loop different
+## The core idea
 
-Most coding agents end when the model says it is done. Loop separates **work** from
-**acceptance**:
+Using an autonomous agent should not mean repeatedly telling it to continue, checking
+whether it really ran the tests, or discovering that “done” meant “I wrote some code.”
+The user should state the goal once. Loop owns the work between that instruction and a
+verified delivery.
 
-1. You confirm concrete success criteria, required final artifacts, and optional exact
-   verification commands.
-2. Loop works inside a per-task workspace under a server-enforced capability and
-   token/step envelope.
-3. An independent verifier re-runs each check on its own fresh workspace snapshot.
-4. Every criterion must map to passing execution evidence in strict mode.
-5. Loop emits a content-addressed Receipt with the contract, checks, model/runtime
-   provenance, output hashes, and the head of a hash-chained step ledger.
-6. The Receipt can be replayed later through the API, CLI, or bundled GitHub Action.
+Loop turns that idea into a bounded execution protocol:
 
-If the evidence fails, the task does not become `completed` merely because the model
-called `finish`. Limit, stuck, cancelled, and error outcomes still receive an
-explicitly unverified Receipt for auditability.
+1. Translate the instruction into a concrete acceptance contract.
+2. Plan one useful action, execute it, and observe the real result.
+3. Feed failures back into the next decision instead of handing them to the user.
+4. Repeat until the required artifacts exist and every acceptance check passes.
+5. Re-run the evidence independently on fresh workspace snapshots.
+6. Deliver the artifacts with a replayable Receipt—or return an explicit, auditable
+   failure when the limits are exhausted.
 
 ```mermaid
 flowchart LR
-    G(["Goal + contract"]) --> P["Plan one action"]
-    P --> A["Act inside authority envelope"]
-    A --> O["Observe"]
-    O --> P
-    P -. "finish claim" .-> V{"Re-run checks\non a fresh copy"}
-    V -. "fails" .-> P
-    V == "passes + full coverage" ==> R(["Replayable Receipt"])
+    U(["One user instruction"]) --> C["Acceptance contract"]
+    C --> P["Plan one action"]
+    P --> A["Act inside bounded authority"]
+    A --> O["Observe real output"]
+    O --> V{"Independent checks pass?"}
+    V -- "No: failure becomes evidence" --> P
+    V -- "Yes" --> D(["Verified delivery + Receipt"])
 ```
 
+The loop is autonomous, not unbounded. Capabilities, filesystem scope, egress policy,
+approvals, step limits, token budgets, no-progress detection, and a verification reserve
+are enforced by the runtime rather than left to the model's discretion.
+
+## What makes Loop different
+
+Most agent frameworks help a model call tools. Loop supervises the complete journey from
+an instruction to an accepted result. It separates **work** from **acceptance**:
+
+- **The model proposes; the runtime decides.** Tool access and resource limits come from
+  a server-enforced authority envelope.
+- **A claim is not completion.** Strict tasks complete only when every criterion maps to
+  passing execution evidence and every required artifact is present.
+- **Failure drives the next loop.** Test output, verifier feedback, and blocked actions are
+  preserved as evidence while repeated or evidence-free branches are stopped.
+- **Verification is isolated.** Each check runs on its own fresh workspace snapshot, so
+  one check cannot manufacture state for another.
+- **Delivery is auditable.** The final Receipt records the contract, checks, model and
+  runtime provenance, output hashes, and the head of a hash-chained step ledger.
+- **Proof is portable.** The Receipt can be replayed later through the API, CLI, or bundled
+  GitHub Action.
+
+If the evidence fails, the task does not become `completed` merely because the model
+called `finish`. Limit, stuck, cancelled, and error outcomes still receive an
+explicitly unverified Receipt for auditability.
+
 ## Run the verified demo
 
 Requirements: Node 22.13+, pnpm 11+ (or Corepack), and Python 3.12+. No provider
@@ -82,7 +106,7 @@ It proves product wiring, not general model quality.
 
 A recorded [DeepSeek `deepseek-chat` run](./evals/results/deepseek-chat-v0.1.0.json)
 of the [12-case real-provider suite](./evals/verified-completion.json) solved `12/12`
-with zero false acceptances: 39 steps, 64,447 provider-reported tokens, and 94.954
+with zero false acceptances: 30 steps, 42,403 provider-reported tokens, and 65.795
 seconds. Every case was execution-verified, fully covered, artifact-complete,
 integrity-valid, and replayable. This is one clean local run using visibly reduced
 `inline` isolation—not a cross-model quality claim or a production-sandbox result.

apps/api/app/services/agent_react.py

@@ -58,6 +58,7 @@
 from app.services.completion import (
     attach_baseline,
     completion_gates_pass,
+    declared_contract_coverage_complete,
     discover_project_checks,
     mark_supplementary_agent_checks,
     merge_completion_checks,
@@ -1380,10 +1381,12 @@ async def _run_loop(
             consecutive_no_progress = guard.no_progress
 
             workspace_changed = workspace.state_marker() != workspace_before
+            auto_finish_candidate = (
+                tool in {"write_file", "edit_file"} and workspace_changed
+            ) or tool == "run_command"
             if (
-                tool in {"write_file", "edit_file"}
+                auto_finish_candidate
                 and status is ToolStatus.OK
-                and workspace_changed
                 and number < task.max_steps
                 and await self._contract_evidence_ready(task, workspace)
             ):
@@ -1760,11 +1763,20 @@ async def _handle_finish(
             if isinstance(raw_checks, list)
             else []
         )
-        checks = merge_completion_checks(
+        required_checks = merge_completion_checks(
             task.required_checks or [],
-            proposed_checks,
+            [],
             criterion_count=len(task.rubric or []),
         )
+        checks = (
+            required_checks
+            if declared_contract_coverage_complete(required_checks, len(task.rubric or []))
+            else merge_completion_checks(
+                required_checks,
+                proposed_checks,
+                criterion_count=len(task.rubric or []),
+            )
+        )
         check_results = attach_baseline(
             await self._run_completion_checks(task, workspace, checks),
             task.baseline_checks or [],
@@ -1996,15 +2008,26 @@ def _history_view(self) -> str:
         compacted into durable decisions/evidence so a long run stays bounded
         without forgetting failed branches."""
         if len(self._history) <= _HISTORY_WINDOW:
-            return "\n".join(entry.render() for entry in self._history) or "(nothing yet)"
+            rendered = [entry.render() for entry in self._history[:-1]]
+            if self._history:
+                rendered.append(self._render_latest_history(self._history[-1]))
+            return "\n".join(rendered) or "(nothing yet)"
         older = self._history[:-_HISTORY_WINDOW]
         recent = self._history[-_HISTORY_WINDOW:]
         return (
             compact_history(older)
             + "\n\n[RECENT STEPS]\n"
-            + "\n".join(entry.render() for entry in recent)
+            + "\n".join(
+                self._render_latest_history(entry) if index == len(recent) - 1 else entry.render()
+                for index, entry in enumerate(recent)
+            )
         )
 
+    @staticmethod
+    def _render_latest_history(entry: HistoryEntry) -> str:
+        limit = 2_400 if entry.tool in {"run_command", "read_file"} else None
+        return entry.render(observation_limit=limit)
+
     @staticmethod
     def _required_checks_view(checks: list[dict[str, Any]]) -> str:
         lines = []

apps/api/app/services/completion.py

@@ -67,6 +67,19 @@ def merge_completion_checks(
     return list(merged.values())
 
 
+def declared_contract_coverage_complete(checks: list[dict[str, Any]], criterion_count: int) -> bool:
+    if criterion_count <= 0:
+        return False
+    expected = {f"criterion-{index:03d}" for index in range(1, criterion_count + 1)}
+    covered = {
+        str(criterion)
+        for check in checks
+        if check.get("source") == "contract"
+        for criterion in check.get("criterion_ids", [])
+    }
+    return expected <= covered
+
+
 def attach_baseline(
     results: list[CheckResult], baseline: list[dict[str, Any]]
 ) -> list[CheckResult]:

apps/api/app/services/evaluation.py

@@ -67,17 +67,19 @@ def aggregate_verified_completion(results: list[dict[str, Any]]) -> dict[str, An
     false_acceptances = sum(bool(item.get("false_acceptance")) for item in results)
     total_steps = sum(int(item.get("steps_used", 0)) for item in results)
     total_tokens = sum(int(item.get("tokens_used", 0)) for item in results)
-    total_duration_seconds = sum(float(item.get("duration_seconds", 0)) for item in results)
+    total_duration_seconds = round(
+        sum(float(item.get("duration_seconds", 0)) for item in results), 3
+    )
     return {
         "cases": total,
         "solved": solved,
-        "solve_rate": solved / total if total else 0.0,
+        "solve_rate": round(solved / total, 4) if total else 0.0,
         "false_acceptances": false_acceptances,
-        "false_acceptance_rate": false_acceptances / total if total else 0.0,
+        "false_acceptance_rate": round(false_acceptances / total, 4) if total else 0.0,
         "total_steps": total_steps,
         "total_tokens": total_tokens,
         "total_duration_seconds": total_duration_seconds,
-        "average_steps": total_steps / total if total else 0.0,
-        "average_tokens": total_tokens / total if total else 0.0,
-        "average_duration_seconds": total_duration_seconds / total if total else 0.0,
+        "average_steps": round(total_steps / total, 3) if total else 0.0,
+        "average_tokens": round(total_tokens / total, 3) if total else 0.0,
+        "average_duration_seconds": round(total_duration_seconds / total, 3) if total else 0.0,
     }

apps/api/app/services/progress.py

@@ -25,13 +25,16 @@ class HistoryEntry:
     observation: str
     status: ToolStatus
 
-    def render(self) -> str:
+    def render(self, *, observation_limit: int | None = None) -> str:
         arg_preview = ", ".join(f"{key}={str(value)[:60]!r}" for key, value in self.args.items())
-        observation_limit = 1_600 if self.tool.startswith(("sibyl_", "argus_", "browser_")) else 600
+        if observation_limit is None:
+            observation_limit = (
+                1_600 if self.tool.startswith(("sibyl_", "argus_", "browser_")) else 600
+            )
         obs = (
             self.observation
             if len(self.observation) <= observation_limit
-            else self.observation[:observation_limit] + " …[truncated]"
+            else self.observation[:observation_limit] + " …[observation preview truncated]"
         )
         return (
             f"Step {self.number} [{self.tool}] ({self.status.value}): {self.thought}\n"

apps/api/app/services/prompts.py

@@ -95,9 +95,13 @@ def plan_prompts(
         "action should usually run it (run_command) — never rewrite the same file "
         "twice in a row without running it in between.\n"
         "- Never read_file — or re-inspect with cat/head/tail/wc via run_command — a "
-        "file you just wrote; write_file already echoed its full contents. Spend steps "
-        "on progress (run it, write the next file, finish), not on re-reading what you "
-        "already have. read_file is only for files you did not create.\n"
+        "file you just wrote. write_file confirms the complete write but may echo only "
+        "a bounded preview; a 'preview truncated' marker NEVER means the source file "
+        "was truncated. Spend steps on progress (run it, write the next file, finish). "
+        "read_file is only for files you did not create.\n"
+        "- A test command that reports zero discovered tests is a failed test run, even "
+        "when its exit code is 0. Fix discovery instead of rewriting working source; for "
+        "Python unittest, put TestCase classes in a test_*.py file.\n"
         "- To create a missing file, call write_file with its COMPLETE requested "
         "content. edit_file only changes a file that already exists. Tool names are "
         "JSON actions, never shell commands.\n"

apps/api/app/services/verification.py

@@ -25,6 +25,7 @@
 
 from app.tools import CapabilityEnvelope, ToolExecutor, ToolStatus, Workspace
 from app.tools.guards import make_egress_guard
+from app.tools.shell import empty_test_suite_reason
 
 
 def _leading_exit_code(out: str) -> int | None:
@@ -176,8 +177,17 @@ def make_result(target: str, passed: bool, evidence: str) -> CheckResult:
         code = _leading_exit_code(out)  # exact code, so "exit code 1" != "exit code 10"
         exit_ok = code == want_exit
         stdout_ok = (expect_stdout is None) or (str(expect_stdout) in out)
-        passed = bool(exit_ok and stdout_ok and tool_result.status is not ToolStatus.BLOCKED)
-        return make_result(command, passed, out[:300])
+        empty_suite = empty_test_suite_reason(command, out)
+        passed = bool(
+            exit_ok
+            and stdout_ok
+            and empty_suite is None
+            and tool_result.status is not ToolStatus.BLOCKED
+        )
+        evidence = out[:300]
+        if empty_suite is not None:
+            evidence = f"{out[:220]}\nrejected: {empty_suite}"[:300]
+        return make_result(command, passed, evidence)
 
     if kind == "file_exists":
         path = str(check.get("path", "")).strip()

apps/api/app/tools/shell.py

@@ -11,6 +11,7 @@
 import asyncio
 import contextlib
 import os
+import re
 import signal
 
 from app.tools.base import ToolResult, ToolStatus
@@ -31,6 +32,22 @@
 _READ_CAP_MULT = 4  # bytes kept before a runaway-output command is killed
 
 
+def empty_test_suite_reason(command: str, observation: str) -> str | None:
+    if re.search(r"(?mi)^Ran 0 tests?\b", observation):
+        return "the test runner executed zero tests"
+    runner = re.search(
+        r"(?i)(?:^|[\s;&|])(?:pytest|py\.test|vitest|jest|mocha)(?:[\s;&|]|$)"
+        r"|(?:npm|pnpm|yarn|bun)\s+(?:run\s+)?test(?:[\s;&|]|$)",
+        command,
+    )
+    if runner and re.search(
+        r"(?i)\b(?:no tests ran|collected 0 items|no tests found|no test files found|0 passing)\b",
+        observation,
+    ):
+        return "the test runner reported an empty test suite"
+    return None
+
+
 def _safe_env() -> dict[str, str]:
     env = {k: os.environ[k] for k in _SAFE_ENV_KEYS if k in os.environ}
     env.setdefault("PATH", "/usr/local/bin:/usr/bin:/bin")
@@ -115,4 +132,14 @@ async def run_command(
     raw, code = await collect_output(
         proc, timeout_seconds=timeout_seconds, output_limit=output_limit
     )
-    return format_result(raw, code, timeout_seconds=timeout_seconds, output_limit=output_limit)
+    result = format_result(raw, code, timeout_seconds=timeout_seconds, output_limit=output_limit)
+    empty_suite = empty_test_suite_reason(command, result.observation)
+    if empty_suite is not None:
+        return ToolResult(
+            result.observation
+            + "\n[INVALID TEST RUN: zero tests were executed. This is not evidence that the "
+            "source file was truncated. For unittest discovery, put TestCase classes in a "
+            "test_*.py file; for other runners, add a discoverable test file.]",
+            ToolStatus.ERROR,
+        )
+    return result

apps/api/app/tools/workspace.py

@@ -32,7 +32,9 @@ def _preview(content: str, *, max_lines: int = 20, max_chars: int = 1000) -> str
     """A bounded echo of written content — enough to confirm the write."""
     snippet = "\n".join(content.splitlines()[:max_lines])
     truncated = len(snippet) > max_chars or snippet != content.rstrip("\n")
-    return snippet[:max_chars] + ("\n… (truncated)" if truncated else "")
+    return snippet[:max_chars] + (
+        "\n… [preview truncated; the write completed with the full content]" if truncated else ""
+    )
 
 
 class Workspace:
@@ -95,7 +97,9 @@ def read(self, relative: str, *, limit: int = 6000) -> str:
             raise ToolError(f"No such file: {relative}")
         text = target.read_text(encoding="utf-8", errors="replace")
         if len(text) > limit:
-            return text[:limit] + f"\n... [truncated, {len(text)} chars total]"
+            return text[:limit] + (
+                f"\n... [preview truncated; the file is unchanged and has {len(text)} chars total]"
+            )
         return text
 
     def list_files(self, *, max_entries: int = 500) -> list[tuple[str, int]]: