feat(browser): retain cleanup evidence for failed Agent Tasks

CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to OriginWeave are documented in this file. The format follo
 
 ### Added
 
+- Failed ordinary and forced-close Agent Task browser trials now retain credential-free temporary-profile cleanup evidence after bounded browser errors, and separate aggregate compatibility gates require cleanup proof from every trial rather than filtering unsuccessful trials out; this does not attest adversarial filesystem erasure, process termination, or arbitrary browser recovery.
 - Rust workspace for independently reusable core, policy, destination, network, TLS, resource, and evidence modules.
 - Canonical HTTPS and loopback-origin boundary with case-normalized schemes and hosts, default-port normalization, IPv4/IPv6 handling, browser-special numeric-host rejection, and explicit malformed-input errors.
 - Typed browser actions, capabilities, risk classes, execution modes, robots decisions, secret-delivery contracts, immutable canonical action-intent digests, and intent-bound approval scopes.

scripts/ci/run_mv3_compatibility.py

@@ -1025,24 +1025,47 @@ def _run_agent_task_trial(
     fixture_url: str,
     trial_number: int,
 ) -> dict[str, Any]:
-    """Run one isolated Agent Task browser trial and prove its profile is removed."""
+    """Run one isolated Agent Task trial and retain cleanup evidence on failure."""
 
     trial_started = time.monotonic()
     profile_path: pathlib.Path
+    result: dict[str, Any] | None = None
+    failure_type: str | None = None
     with tempfile.TemporaryDirectory(
         prefix=f"originweave-agent-task-trial-{trial_number}-"
     ) as profile_dir:
         profile_path = pathlib.Path(profile_dir)
-        result = _run_agent_task_browser_pass(
-            chrome_bin,
-            chromedriver_bin,
-            fixture_url,
-            profile_dir,
-        )
+        try:
+            result = _run_agent_task_browser_pass(
+                chrome_bin,
+                chromedriver_bin,
+                fixture_url,
+                profile_dir,
+            )
+        except (
+            OSError,
+            ValueError,
+            RuntimeError,
+            json.JSONDecodeError,
+            subprocess.TimeoutExpired,
+        ) as exc:
+            failure_type = type(exc).__name__
     profile_cleaned = not profile_path.exists()
     if not profile_cleaned:
         raise RuntimeError(f"Agent Task profile cleanup failed in trial {trial_number}")
 
+    duration_ms = round((time.monotonic() - trial_started) * 1000)
+    if failure_type is not None:
+        return {
+            "trial_number": trial_number,
+            "passed": False,
+            "failure_type": failure_type,
+            "profile_cleaned": True,
+            "duration_ms": duration_ms,
+        }
+    if result is None:
+        raise RuntimeError("Agent Task browser pass returned no result")
+
     return {
         "trial_number": trial_number,
         "passed": True,
@@ -1068,8 +1091,8 @@ def _run_agent_task_trial(
         "semantic_observation_bytes": result["semantic_observation_bytes"],
         "action_latency_ms": result["action_latency_ms"],
         "task_duration_ms": result["task_duration_ms"],
-        "profile_cleaned": profile_cleaned,
-        "duration_ms": round((time.monotonic() - trial_started) * 1000),
+        "profile_cleaned": True,
+        "duration_ms": duration_ms,
     }
 
 
@@ -1274,34 +1297,57 @@ def _run_agent_task_forced_close_trial(
     fixture_url: str,
     trial_number: int,
 ) -> dict[str, Any]:
-    """Run one forced-close probe and prove its isolated browser profile is removed."""
+    """Run one forced-close trial and retain cleanup evidence on failure."""
 
     trial_started = time.monotonic()
     profile_path: pathlib.Path
+    result: dict[str, Any] | None = None
+    failure_type: str | None = None
     with tempfile.TemporaryDirectory(
         prefix=f"originweave-agent-task-forced-close-{trial_number}-"
     ) as profile_dir:
         profile_path = pathlib.Path(profile_dir)
-        result = _run_agent_task_forced_close_browser_pass(
-            chrome_bin,
-            chromedriver_bin,
-            fixture_url,
-            profile_dir,
-        )
+        try:
+            result = _run_agent_task_forced_close_browser_pass(
+                chrome_bin,
+                chromedriver_bin,
+                fixture_url,
+                profile_dir,
+            )
+        except (
+            OSError,
+            ValueError,
+            RuntimeError,
+            json.JSONDecodeError,
+            subprocess.TimeoutExpired,
+        ) as exc:
+            failure_type = type(exc).__name__
     profile_cleaned = not profile_path.exists()
     if not profile_cleaned:
         raise RuntimeError(
             f"Agent Task forced-close profile cleanup failed in trial {trial_number}"
         )
 
+    duration_ms = round((time.monotonic() - trial_started) * 1000)
+    if failure_type is not None:
+        return {
+            "trial_number": trial_number,
+            "passed": False,
+            "failure_type": failure_type,
+            "profile_cleaned": True,
+            "duration_ms": duration_ms,
+        }
+    if result is None:
+        raise RuntimeError("Agent Task forced-close browser pass returned no result")
+
     return {
         "trial_number": trial_number,
         "passed": True,
         "browser_version": result["browser_version"],
         "forced_close_detected": result["forced_close_detected"],
         "session_survived": result["session_survived"],
-        "profile_cleaned": profile_cleaned,
-        "duration_ms": round((time.monotonic() - trial_started) * 1000),
+        "profile_cleaned": True,
+        "duration_ms": duration_ms,
     }
 
 
@@ -1443,6 +1489,9 @@ def main() -> int:
         agent_task_trial_pass_rate = (
             agent_task_successful_trials / AGENT_TASK_REPEATABILITY_TRIALS
         )
+        agent_task_profiles_cleaned = all(
+            trial.get("profile_cleaned") is True for trial in agent_task_trials
+        )
         agent_task_isolation_complete = all(
             trial.get("profile_pristine_before_launch") is True
             and trial.get("ambient_cookies_absent") is True
@@ -1486,6 +1535,9 @@ def main() -> int:
         forced_close_successful_trials = sum(
             1 for trial in forced_close_trials if trial.get("passed") is True
         )
+        forced_close_profiles_cleaned = all(
+            trial.get("profile_cleaned") is True for trial in forced_close_trials
+        )
         forced_close_surfaces_complete = all(
             trial.get("forced_close_detected") is True
             and trial.get("session_survived") is True
@@ -1511,11 +1563,13 @@ def main() -> int:
                 "repeatability_trials": AGENT_TASK_REPEATABILITY_TRIALS,
                 "successful_trials": agent_task_successful_trials,
                 "trial_pass_rate": agent_task_trial_pass_rate,
+                "profiles_cleaned": agent_task_profiles_cleaned,
                 "isolation_complete": agent_task_isolation_complete,
                 "trial_results": agent_task_trials,
                 "forced_close": {
                     "repeatability_trials": AGENT_TASK_REPEATABILITY_TRIALS,
                     "successful_trials": forced_close_successful_trials,
+                    "profiles_cleaned": forced_close_profiles_cleaned,
                     "trial_results": forced_close_trials,
                 },
             },
@@ -1529,6 +1583,8 @@ def main() -> int:
             )
         if not common_surfaces or not all(common_surfaces.values()):
             raise RuntimeError("Manifest V3 repeatability surfaces were incomplete")
+        if not agent_task_profiles_cleaned:
+            raise RuntimeError("Agent Task profile cleanup gate failed")
         if agent_task_successful_trials != AGENT_TASK_REPEATABILITY_TRIALS:
             raise RuntimeError(
                 "Agent Task repeatability gate failed: "
@@ -1539,6 +1595,8 @@ def main() -> int:
             raise RuntimeError("Agent Task isolation gate failed")
         if not agent_task_surfaces_complete:
             raise RuntimeError("Agent Task repeatability surfaces were incomplete")
+        if not forced_close_profiles_cleaned:
+            raise RuntimeError("Agent Task forced-close profile cleanup gate failed")
         if (
             forced_close_successful_trials != AGENT_TASK_REPEATABILITY_TRIALS
             or not forced_close_surfaces_complete

tests/test_agent_task_failure_cleanup_contract.py

@@ -0,0 +1,132 @@
+"""Contract for Agent Task profile cleanup evidence on failed browser trials."""
+
+from __future__ import annotations
+
+import pathlib
+import runpy
+import subprocess
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py"
+
+
+class AgentTaskFailureCleanupContractTests(unittest.TestCase):
+    """Require failed trials to retain credential-free teardown evidence."""
+
+    def _namespace(self, name: str) -> dict[str, object]:
+        return runpy.run_path(str(RUNNER), run_name=name)
+
+    def test_failed_browser_pass_returns_profile_cleanup_evidence(self) -> None:
+        """A browser-pass failure must not discard proof that its task profile was removed."""
+
+        namespace = self._namespace("agent_task_failure_cleanup_behavior")
+        run_trial = namespace["_run_agent_task_trial"]
+
+        def fail_browser_pass(*_args: object, **_kwargs: object) -> dict[str, object]:
+            raise RuntimeError("synthetic controlled browser failure")
+
+        run_trial.__globals__["_run_agent_task_browser_pass"] = fail_browser_pass
+        result = run_trial(
+            pathlib.Path("controlled-chrome"),
+            pathlib.Path("controlled-chromedriver"),
+            "http://127.0.0.1/controlled-fixture",
+            7,
+        )
+
+        self.assertEqual(result["trial_number"], 7)
+        self.assertIs(result["passed"], False)
+        self.assertEqual(result["failure_type"], "RuntimeError")
+        self.assertIs(result["profile_cleaned"], True)
+        self.assertNotIn("synthetic controlled browser failure", repr(result))
+
+    def test_teardown_timeout_returns_profile_cleanup_evidence(self) -> None:
+        """A reviewed process teardown timeout must become one failed trial, not abort the run."""
+
+        namespace = self._namespace("agent_task_teardown_timeout_cleanup_behavior")
+        run_trial = namespace["_run_agent_task_trial"]
+
+        def timeout_browser_pass(*_args: object, **_kwargs: object) -> dict[str, object]:
+            raise subprocess.TimeoutExpired(
+                cmd="private-controlled-chromedriver-path",
+                timeout=5,
+            )
+
+        run_trial.__globals__["_run_agent_task_browser_pass"] = timeout_browser_pass
+        result = run_trial(
+            pathlib.Path("controlled-chrome"),
+            pathlib.Path("controlled-chromedriver"),
+            "http://127.0.0.1/controlled-fixture",
+            8,
+        )
+
+        self.assertEqual(result["trial_number"], 8)
+        self.assertIs(result["passed"], False)
+        self.assertEqual(result["failure_type"], "TimeoutExpired")
+        self.assertIs(result["profile_cleaned"], True)
+        self.assertNotIn("private-controlled-chromedriver-path", repr(result))
+
+    def test_failed_forced_close_pass_returns_profile_cleanup_evidence(self) -> None:
+        """A forced-close probe failure must still prove that its task profile was removed."""
+
+        namespace = self._namespace("agent_task_forced_close_cleanup_behavior")
+        run_trial = namespace["_run_agent_task_forced_close_trial"]
+
+        def fail_browser_pass(*_args: object, **_kwargs: object) -> dict[str, object]:
+            raise RuntimeError("synthetic forced-close browser failure")
+
+        run_trial.__globals__["_run_agent_task_forced_close_browser_pass"] = fail_browser_pass
+        result = run_trial(
+            pathlib.Path("controlled-chrome"),
+            pathlib.Path("controlled-chromedriver"),
+            "http://127.0.0.1/controlled-fixture",
+            9,
+        )
+
+        self.assertEqual(result["trial_number"], 9)
+        self.assertIs(result["passed"], False)
+        self.assertEqual(result["failure_type"], "RuntimeError")
+        self.assertIs(result["profile_cleaned"], True)
+        self.assertNotIn("synthetic forced-close browser failure", repr(result))
+
+    def test_forced_close_teardown_timeout_returns_profile_cleanup_evidence(self) -> None:
+        """Forced-close teardown timeout must retain cleanup evidence without raw command text."""
+
+        namespace = self._namespace("agent_task_forced_close_teardown_timeout_behavior")
+        run_trial = namespace["_run_agent_task_forced_close_trial"]
+
+        def timeout_browser_pass(*_args: object, **_kwargs: object) -> dict[str, object]:
+            raise subprocess.TimeoutExpired(
+                cmd="private-forced-close-chromedriver-path",
+                timeout=5,
+            )
+
+        run_trial.__globals__["_run_agent_task_forced_close_browser_pass"] = timeout_browser_pass
+        result = run_trial(
+            pathlib.Path("controlled-chrome"),
+            pathlib.Path("controlled-chromedriver"),
+            "http://127.0.0.1/controlled-fixture",
+            10,
+        )
+
+        self.assertEqual(result["trial_number"], 10)
+        self.assertIs(result["passed"], False)
+        self.assertEqual(result["failure_type"], "TimeoutExpired")
+        self.assertIs(result["profile_cleaned"], True)
+        self.assertNotIn("private-forced-close-chromedriver-path", repr(result))
+
+    def test_acceptance_gate_requires_cleanup_evidence_for_every_trial(self) -> None:
+        """Failed trials must not be filtered out of either profile-cleanup gate."""
+
+        runner = RUNNER.read_text(encoding="utf-8")
+        self.assertIn("agent_task_profiles_cleaned = all(", runner)
+        self.assertIn("forced_close_profiles_cleaned = all(", runner)
+        self.assertIn('trial.get("profile_cleaned") is True', runner)
+        self.assertIn('"profiles_cleaned": agent_task_profiles_cleaned', runner)
+        self.assertIn('"profiles_cleaned": forced_close_profiles_cleaned', runner)
+        self.assertIn("Agent Task profile cleanup gate failed", runner)
+        self.assertIn("Agent Task forced-close profile cleanup gate failed", runner)
+
+
+if __name__ == "__main__":
+    unittest.main()