Restrict Daytona verification to computer use

Author: homanp

pithos/runtime_plugins.py

@@ -962,7 +962,7 @@ def _run_pi_agent(
                         env=_local_pi_env(app, agent_cfg, computer_use_bin=computer_use_bin),
                         transcript_path=str(transcript),
                         progress_prefix=f"[verify {finding.id}]",
-                        tools=["read", "bash"],
+                        tools=_live_agent_tools(computer_use_mode),
                         system_prompt=_localize_live_agent_paths(
                             _live_agent_system_prompt(computer_use_mode),
                             repo_path=app.repo_copy,
@@ -1010,7 +1010,7 @@ def _run_pi_agent(
                         model=model,
                         transcript_path=str(transcript),
                         progress_prefix=f"[verify:{finding.id}]",
-                        tools=["read", "bash"],
+                        tools=_live_agent_tools(computer_use_mode),
                         system_prompt=_live_agent_system_prompt(computer_use_mode),
                         event_sink=getattr(self, "event_sink", NULL_EVENT_SINK),
                     )
@@ -1049,25 +1049,27 @@ def _verdict_from_agent_result(
             )
         verdict = _verdict_from_payload(finding, plan, verdict_data, plugin=self.name)
         verdict.artifacts["transcript"] = str(finding_dir / "live-agent-transcript.jsonl")
-        if _profile_computer_use_mode(self.profile) == COMPUTER_USE_DAYTONA and not (
-            _agent_result_has_daytona_command(result)
-        ):
-            return RuntimeVerdict(
-                finding_id=finding.id,
-                title=finding.title,
-                plugin=self.name,
-                status=RUNTIME_BLOCKED,
-                confidence="low",
-                evidence=[
-                    "Daytona computer use was requested, but the live-agent transcript did not "
-                    "include a bash tool call invoking any cu-* command.",
-                    "The verifier requires computer-use evidence before accepting a Daytona "
-                    "live-agent verdict.",
-                ],
-                error="missing Daytona computer-use evidence",
-                artifacts={"transcript": str(finding_dir / "live-agent-transcript.jsonl")},
-                plan=plan.to_dict(),
-            )
+        if _profile_computer_use_mode(self.profile) == COMPUTER_USE_DAYTONA:
+            daytona_policy = _daytona_computer_only_policy(result)
+            if daytona_policy:
+                details = daytona_policy[:5]
+                if len(daytona_policy) > len(details):
+                    details.append(f"... {len(daytona_policy) - len(details)} more violation(s)")
+                return RuntimeVerdict(
+                    finding_id=finding.id,
+                    title=finding.title,
+                    plugin=self.name,
+                    status=RUNTIME_BLOCKED,
+                    confidence="low",
+                    evidence=[
+                        "Daytona computer use was requested, but verification did not happen "
+                        "exclusively through cu-* computer-use commands.",
+                        *details,
+                    ],
+                    error="missing or invalid Daytona computer-use evidence",
+                    artifacts={"transcript": str(finding_dir / "live-agent-transcript.jsonl")},
+                    plan=plan.to_dict(),
+                )
         return verdict
 
     def _blocked(
@@ -1294,13 +1296,21 @@ def _profile_computer_use_mode(profile: AppRuntimeProfile) -> str:
         return COMPUTER_USE_NONE
 
 
+def _live_agent_tools(computer_use: str) -> list[str]:
+    if normalize_computer_use_mode(computer_use) == COMPUTER_USE_DAYTONA:
+        return ["bash"]
+    return ["read", "bash"]
+
+
 def _localize_live_agent_paths(text: str, *, repo_path: Path, artifacts_path: Path) -> str:
     return text.replace("/work/repo", str(repo_path)).replace(
         "/work/artifacts", str(artifacts_path)
     )
 
 
-def _agent_result_has_daytona_command(result) -> bool:
+def _daytona_computer_only_policy(result) -> list[str]:
+    violations: list[str] = []
+    saw_computer_use = False
     for msg in getattr(result, "messages", []) or []:
         if msg.get("role") != "assistant":
             continue
@@ -1310,13 +1320,33 @@ def _agent_result_has_daytona_command(result) -> bool:
         for block in content:
             if not isinstance(block, dict) or block.get("type") != "toolCall":
                 continue
-            if block.get("name") != "bash":
+            name = str(block.get("name") or "")
+            if name != "bash":
+                violations.append(f"non-computer tool used: {name}")
                 continue
             args = block.get("arguments")
             command = args.get("command") if isinstance(args, dict) else ""
-            if any(cmd in str(command) for cmd in DAYTONA_COMMANDS):
-                return True
-    return False
+            if _bash_command_is_computer_use_only(str(command)):
+                saw_computer_use = True
+            else:
+                violations.append(f"non-computer bash command used: {command}")
+    if not saw_computer_use:
+        violations.insert(0, "no cu-* computer-use command was run")
+    return violations
+
+
+def _bash_command_is_computer_use_only(command: str) -> bool:
+    if not command.strip():
+        return False
+    if any(token in command for token in ("|", "`", "$(")):
+        return False
+    for segment in re.split(r"\s*(?:&&|\|\||;|\n)\s*", command):
+        segment = segment.strip()
+        if not segment:
+            continue
+        if segment.split()[0] not in DAYTONA_COMMANDS:
+            return False
+    return True
 
 
 def _extract_verdict_payload(text: str) -> dict[str, object] | None:
@@ -1398,8 +1428,10 @@ def _live_agent_system_prompt(computer_use: str = COMPUTER_USE_NONE) -> str:
 - Do not start Xvfb, XFCE, x11vnc, noVNC, or a nested Daytona sandbox yourself.
 - Use bash-accessible local commands to control the existing desktop:
   `cu-info`, `cu-screenshot`, `cu-click`, `cu-type`, `cu-key`, `cu-scroll`, and `cu-drag`.
-- Your first bash actions must run `cu-info` and `cu-screenshot` before reading source, installing
-  dependencies, running tests, or writing reproduction scripts.
+- Verification must happen only through these `cu-*` computer-use commands. Do not use the read tool,
+  do not inspect source files through shell commands, do not install dependencies, do not run tests,
+  and do not write reproduction scripts.
+- Your first bash actions must run `cu-info` and `cu-screenshot`.
 - Take another screenshot after desktop/UI actions, and use the image evidence to decide whether the
   action affected the expected screen.
 - Include the `cu-*` commands you ran and their artifact paths in the final runtime verdict evidence.

tests/test_runtime_verifier.py

@@ -716,8 +716,12 @@ def test_daytona_computer_use_reaches_local_live_agent(tmp_path, monkeypatch):
     monkeypatch.delenv("DAYTONA_JWT_TOKEN", raising=False)
 
     async def fake_run_agent_process(prompt, **kwargs):
-        assert kwargs["tools"] == ["read", "bash"]
+        assert kwargs["tools"] == ["bash"]
         assert "Daytona Computer Use" in kwargs["system_prompt"]
+        assert (
+            "Verification must happen only through these `cu-*` computer-use commands"
+            in kwargs["system_prompt"]
+        )
         assert "cu-screenshot" in kwargs["system_prompt"]
         assert "Do not start Xvfb" in kwargs["system_prompt"]
         assert kwargs["env"]["PITHOS_COMPUTER_USE"] == "daytona"
@@ -848,7 +852,91 @@ async def fake_run_agent_process(prompt, **kwargs):
     )
 
     assert result.verdicts[0].status == "blocked"
-    assert result.verdicts[0].error == "missing Daytona computer-use evidence"
+    assert result.verdicts[0].error == "missing or invalid Daytona computer-use evidence"
+
+
+def test_daytona_computer_use_blocks_mixed_source_verification(tmp_path, monkeypatch):
+    repo = tmp_path / "app"
+    repo.mkdir()
+    profile = repo / ".pithos" / "runtime.yaml"
+    profile.parent.mkdir()
+    profile.write_text(
+        """
+verification:
+  execute_app: true
+environment:
+  sandbox: local
+""",
+        encoding="utf-8",
+    )
+    results = tmp_path / "results" / "app" / "run"
+    results.mkdir(parents=True)
+    (results / "run-summary.json").write_text(json.dumps({"repo": {"path": str(repo)}}))
+    (results / "VULN-FINDINGS.json").write_text(
+        json.dumps(
+            [
+                {
+                    "id": "F001",
+                    "title": "Generic live authorization bypass",
+                    "category": "authorization bypass",
+                    "files": ["src/server.ts"],
+                }
+            ]
+        )
+    )
+    (results / "TRIAGE.json").write_text(
+        json.dumps({"findings": [{"id": "F001", "title": "Generic live authorization bypass"}]})
+    )
+
+    async def fake_run_agent_process(prompt, **kwargs):
+        payload = json.dumps(
+            {
+                "status": "not_reproduced",
+                "confidence": "high",
+                "evidence": ["ran screenshot and source verification"],
+            }
+        )
+        return SimpleNamespace(
+            error=None,
+            messages=[
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "toolCall",
+                            "name": "bash",
+                            "arguments": {"command": "cu-info && cu-screenshot before.png"},
+                        },
+                        {
+                            "type": "toolCall",
+                            "name": "bash",
+                            "arguments": {"command": "cat src/server.ts"},
+                        },
+                    ],
+                }
+            ],
+            last_assistant_message=payload,
+            find_tagged_message=lambda _tag: payload,
+        )
+
+    monkeypatch.setattr(runtime_plugins, "run_agent_process", fake_run_agent_process)
+    monkeypatch.setattr(
+        runtime_plugins.docker_ops,
+        "run",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("docker was called")),
+    )
+
+    result = run_verify_repo(
+        triage_path=results,
+        execute_app=True,
+        provider="anthropic",
+        model="claude-sonnet-4-5",
+        sandbox_mode="local",
+        computer_use="daytona",
+    )
+
+    assert result.verdicts[0].status == "blocked"
+    assert any("cat src/server.ts" in e for e in result.verdicts[0].evidence)
 
 
 def test_verify_repo_confirms_polar_style_source_oracles(tmp_path):