Harden artifact verification and publish DeepSeek evaluation

.gitignore

@@ -13,6 +13,7 @@ apps/web/playwright-report/
 apps/web/test-results/
 /evals/results/*
 !/evals/results/demo-smoke.json
+!/evals/results/deepseek-chat-v0.1.0.json
 
 # ---------- Python ----------
 __pycache__/

README.md

@@ -23,10 +23,11 @@ the evidence, and returns a Receipt anyone can replay.**
 Most coding agents end when the model says it is done. Loop separates **work** from
 **acceptance**:
 
-1. You confirm concrete success criteria and may provide exact verification commands.
+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 the checks on a fresh copy of the workspace.
+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.
@@ -77,9 +78,14 @@ fresh environment → open UI → confirm contract → run → execution verifie
 
 The committed [demo smoke report](./evals/results/demo-smoke.json) records `1/1`
 solved, zero false acceptances, two steps, 24 scripted tokens, and a passing replay.
-It proves product wiring, not general model quality. The separate
-[12-case real-provider suite](./evals/verified-completion.json) is published without
-invented results; running it requires explicit acknowledgement of provider spend.
+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
+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.
 
 ## The trust boundary
 
@@ -109,8 +115,9 @@ and residual risks.
 ## What is implemented
 
 - ReAct planning with independent executor/verifier provider selection and fallback.
-- Strict acceptance contracts, baseline regression checks, criterion-to-evidence
-  mappings, Receipt replay, and offline verification.
+- Strict acceptance contracts with required final artifacts, independent check
+  snapshots, baseline regression checks, criterion-to-evidence mappings, Receipt
+  replay, and offline verification.
 - Per-task workspaces, container or Kubernetes command isolation, secret redaction,
   destination-bound egress, and restart-safe approval gates.
 - Durable Redis Streams workers with visibility leases, stale-message reclaim,

apps/api/app/schemas/task.py

@@ -31,6 +31,7 @@ class TaskCreate(BaseModel):
     goal: str = Field(min_length=4, max_length=4_000)
     success_criteria: list[str] | None = None
     verification_commands: list[str] = Field(default_factory=list)
+    required_artifacts: list[str] = Field(default_factory=list)
     verification_mode: str | None = Field(default=None, pattern=r"^(strict|judgment)$")
     project_id: str = Field(default="default", min_length=1, max_length=100, pattern=r"^[\w.-]+$")
     # Optional path relative to LOOP_LOCAL_PROJECTS_ROOT. The source repository
@@ -86,6 +87,22 @@ def validate_verification_commands(cls, value: list[str]) -> list[str]:
             raise ValueError("each verification command must be at most 1000 characters")
         return commands
 
+    @field_validator("required_artifacts")
+    @classmethod
+    def validate_required_artifacts(cls, value: list[str]) -> list[str]:
+        artifacts = list(dict.fromkeys(item.strip() for item in value if item.strip()))
+        if len(artifacts) > 32:
+            raise ValueError("required_artifacts may contain at most 32 paths")
+        for artifact in artifacts:
+            parts = artifact.split("/")
+            if (
+                len(artifact) > 500
+                or "\\" in artifact
+                or any(part in {"", ".", ".."} for part in parts)
+            ):
+                raise ValueError("required_artifacts must be workspace-relative POSIX paths")
+        return artifacts
+
     @model_validator(mode="after")
     def validate_network_authority(self) -> TaskCreate:
         requested = set(self.capabilities or [])

apps/api/app/services/agent_react.py

@@ -59,6 +59,7 @@
     attach_baseline,
     completion_gates_pass,
     discover_project_checks,
+    mark_supplementary_agent_checks,
     merge_completion_checks,
     regressions,
 )
@@ -1768,6 +1769,9 @@ async def _handle_finish(
             await self._run_completion_checks(task, workspace, checks),
             task.baseline_checks or [],
         )
+        contract_substantiation_authoritative = mark_supplementary_agent_checks(
+            check_results, len(task.rubric or [])
+        )
         gates_passed = completion_gates_pass(check_results)
         coverage_complete = execution_coverage_complete(check_results, len(task.rubric or []))
 
@@ -1822,12 +1826,6 @@ async def _handle_finish(
         else:
             score, missing, llm_met, substantiate = 0, ["verifier returned no verdict"], False, True
 
-        contract_results = [result for result in check_results if result.source == "contract"]
-        contract_substantiation_authoritative = bool(
-            contract_results
-            and completion_gates_pass(contract_results)
-            and execution_coverage_complete(contract_results, len(task.rubric or []))
-        )
         execution_ready = bool(
             check_results
             and coverage_complete

apps/api/app/services/completion.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Any
 
-from app.services.verification import CheckResult
+from app.services.verification import CheckResult, execution_coverage_complete
 
 
 def discover_project_checks(root: Path) -> list[dict[str, Any]]:
@@ -78,6 +78,8 @@ def attach_baseline(
 
 def completion_gates_pass(results: list[CheckResult]) -> bool:
     for result in results:
+        if not result.gating:
+            continue
         if result.passed:
             continue
         if result.source == "system" and result.baseline_passed is False:
@@ -86,6 +88,24 @@ def completion_gates_pass(results: list[CheckResult]) -> bool:
     return True
 
 
+def mark_supplementary_agent_checks(results: list[CheckResult], criterion_count: int) -> bool:
+    contract = [result for result in results if result.source == "contract"]
+    authoritative = bool(
+        contract
+        and completion_gates_pass(contract)
+        and execution_coverage_complete(contract, criterion_count)
+    )
+    if not authoritative:
+        return False
+    for result in results:
+        if result.source != "agent":
+            continue
+        result.gating = False
+        if result.definition is not None:
+            result.definition["gating"] = False
+    return True
+
+
 def regressions(results: list[CheckResult]) -> list[CheckResult]:
     return [
         result

apps/api/app/services/evaluation.py

@@ -65,21 +65,19 @@ def aggregate_verified_completion(results: list[dict[str, Any]]) -> dict[str, An
     total = len(results)
     solved = sum(bool(item.get("solved")) for item in results)
     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)
     return {
         "cases": total,
         "solved": solved,
         "solve_rate": solved / total if total else 0.0,
         "false_acceptances": false_acceptances,
         "false_acceptance_rate": false_acceptances / total if total else 0.0,
-        "average_steps": (
-            sum(int(item.get("steps_used", 0)) for item in results) / total if total else 0.0
-        ),
-        "average_tokens": (
-            sum(int(item.get("tokens_used", 0)) for item in results) / total if total else 0.0
-        ),
-        "average_duration_seconds": (
-            sum(float(item.get("duration_seconds", 0)) for item in results) / total
-            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,
     }

apps/api/app/services/task.py

@@ -114,6 +114,15 @@ async def publish(self, payload: TaskCreate) -> TaskModel:
             }
             for index, command in enumerate(payload.verification_commands, start=1)
         ]
+        required_checks.extend(
+            {
+                "id": f"contract-artifact-{index:03d}",
+                "kind": "file_exists",
+                "path": path,
+                "source": "contract",
+            }
+            for index, path in enumerate(payload.required_artifacts, start=1)
+        )
         task = await self.tasks.create(
             goal=payload.goal.strip(),
             owner_id=self.subject,

apps/api/app/services/verification.py

@@ -44,6 +44,7 @@ class CheckResult:
     definition: dict[str, Any] | None = None
     source: str = "agent"
     baseline_passed: bool | None = None
+    gating: bool = True
 
 
 VERIFY_DIR_PREFIX = "verify-"
@@ -85,43 +86,43 @@ async def run_checks(
     docker_workspace_mount: str | None = None,
     infer_criterion_ids: bool = True,
 ) -> list[CheckResult]:
-    """Re-run each check on a fresh copy of the workspace. Never raises — a bad
-    check definition becomes a failed result the verifier can act on."""
+    """Re-run every check on its own fresh copy of the workspace. Never raises —
+    a bad check definition becomes a failed result the verifier can act on."""
     if not checks:
         return []
 
     # Copy under the workspaces root (a path the container can bind-mount) rather
     # than system temp, so re-verification runs in the same sandbox as the agent.
     tmp_root = source.root.parent / f"{VERIFY_DIR_PREFIX}{uuid.uuid4().hex[:12]}"
     try:
-        copy_dir = tmp_root / "ws"
-        await asyncio.to_thread(shutil.copytree, source.root, copy_dir)
-        workspace = Workspace(copy_dir)
         # Verify under the same authority as the task: default-deny egress unless
         # the task had it (so a check can't reach the network the task couldn't,
         # and a legit network check isn't blocked in a container).
         check_envelope = envelope or CapabilityEnvelope.from_tools(
             None, egress_allowed=egress_allowed
         )
-        executor = ToolExecutor(
-            workspace,
-            approval_mode=approval_mode,
-            command_timeout=command_timeout,
-            output_limit=output_limit,
-            envelope=check_envelope,
-            before_tool=make_egress_guard(check_envelope, workspace),
-            sandbox_image=sandbox_image,
-            sandbox_backend=sandbox_backend,
-            sandbox_memory=sandbox_memory,
-            sandbox_cpus=sandbox_cpus,
-            egress_proxy_url=egress_proxy_url,
-            egress_network=egress_network,
-            egress_token_factory=egress_token_factory,
-            docker_workspace_volume=docker_workspace_volume,
-            docker_workspace_mount=docker_workspace_mount,
-        )
         results: list[CheckResult] = []
         for index, check in enumerate(checks, start=1):
+            copy_dir = tmp_root / f"check-{index:03d}"
+            await asyncio.to_thread(shutil.copytree, source.root, copy_dir)
+            workspace = Workspace(copy_dir)
+            executor = ToolExecutor(
+                workspace,
+                approval_mode=approval_mode,
+                command_timeout=command_timeout,
+                output_limit=output_limit,
+                envelope=check_envelope,
+                before_tool=make_egress_guard(check_envelope, workspace),
+                sandbox_image=sandbox_image,
+                sandbox_backend=sandbox_backend,
+                sandbox_memory=sandbox_memory,
+                sandbox_cpus=sandbox_cpus,
+                egress_proxy_url=egress_proxy_url,
+                egress_network=egress_network,
+                egress_token_factory=egress_token_factory,
+                docker_workspace_volume=docker_workspace_volume,
+                docker_workspace_mount=docker_workspace_mount,
+            )
             mapped_check = dict(check)
             if infer_criterion_ids and not mapped_check.get("criterion_ids"):
                 if criterion_count == 1:
@@ -157,6 +158,7 @@ def make_result(target: str, passed: bool, evidence: str) -> CheckResult:
             criterion_ids=criterion_ids,
             definition=definition,
             source=str(check.get("source") or "agent")[:40],
+            gating=check.get("gating") is not False,
         )
 
     if kind == "command":
@@ -208,7 +210,8 @@ def checks_summary(results: list[CheckResult]) -> str:
         baseline = ""
         if r.baseline_passed is not None:
             baseline = f"; baseline={'PASS' if r.baseline_passed else 'FAIL'}"
-        lines.append(f"[{mark}] {r.source} {r.kind} {r.target}: {r.evidence}{baseline}")
+        gating = "" if r.gating else "; supplementary"
+        lines.append(f"[{mark}] {r.source} {r.kind} {r.target}: {r.evidence}{baseline}{gating}")
     return "\n".join(lines)
 
 

apps/api/scripts/evaluate_verified_completion.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import time
@@ -38,15 +39,27 @@ def _wait_for_task(client: httpx.Client, task_id: str, timeout: float) -> dict[s
     raise TimeoutError(f"task {task_id} did not finish within {timeout:g}s")
 
 
-def _evaluate_case(client: httpx.Client, case: dict[str, Any], timeout: float) -> dict[str, Any]:
-    started = time.monotonic()
-    payload = {
+def _build_task_payload(case: dict[str, Any]) -> dict[str, Any]:
+    expected_files = [str(path) for path in case.get("expected_files", [])]
+    success_criteria = list(case["success_criteria"])
+    if expected_files:
+        artifacts = ", ".join(f"`{path}`" for path in expected_files)
+        success_criteria.append(
+            f"The final workspace contains all required artifacts: {artifacts}."
+        )
+    return {
         "goal": case["goal"],
-        "success_criteria": case["success_criteria"],
+        "success_criteria": success_criteria,
         "verification_commands": case["verification_commands"],
+        "required_artifacts": expected_files,
         "verification_mode": "strict",
         "limits": case.get("limits", {"max_steps": 20, "token_budget": 30_000}),
     }
+
+
+def _evaluate_case(client: httpx.Client, case: dict[str, Any], timeout: float) -> dict[str, Any]:
+    started = time.monotonic()
+    payload = _build_task_payload(case)
     response = client.post("/api/v1/tasks", json=payload)
     response.raise_for_status()
     task = _wait_for_task(client, response.json()["id"], timeout)
@@ -74,6 +87,9 @@ def _evaluate_case(client: httpx.Client, case: dict[str, Any], timeout: float) -
         "status": task["status"],
         "duration_seconds": round(time.monotonic() - started, 3),
         "model": provenance.get("model"),
+        "isolation": receipt.get("isolation"),
+        "ledger_head": receipt.get("ledger_head"),
+        "receipt_hash": receipt.get("receipt_hash"),
         **scored,
     }
 
@@ -123,6 +139,7 @@ def main(argv: list[str] | None = None) -> int:
         "run_at": datetime.now(UTC).isoformat(),
         "label": args.label,
         "manifest": manifest,
+        "manifest_sha256": hashlib.sha256(args.cases.read_bytes()).hexdigest(),
         "summary": aggregate_verified_completion(results),
         "results": results,
     }