p1

rcholic · rcholic · commit 4ce424695578 · 2026-01-18T10:04:30.000-08:00
diff --git a/README.md b/README.md
@@ -89,6 +89,21 @@ async def main():
 asyncio.run(main())
 ```
 
+### Failure Artifact Buffer (Phase 1)
+
+Capture a short ring buffer of screenshots and persist them when a required assertion fails.
+
+```python
+from sentience.failure_artifacts import FailureArtifactsOptions
+
+await runtime.enable_failure_artifacts(
+    FailureArtifactsOptions(buffer_seconds=15, capture_on_action=True, fps=0.0)
+)
+
+# After each action, record it (best-effort).
+await runtime.record_action("CLICK")
+```
+
 **See examples:** [`examples/asserts/`](examples/asserts/)
 
 ## 🚀 Quick Start: Choose Your Abstraction Level
diff --git a/sentience/agent_runtime.py b/sentience/agent_runtime.py
@@ -70,6 +70,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from .failure_artifacts import FailureArtifactBuffer, FailureArtifactsOptions
 from .models import Snapshot, SnapshotOptions
 from .verification import AssertContext, AssertOutcome, Predicate
 
@@ -138,6 +139,10 @@ def __init__(
         # Snapshot state
         self.last_snapshot: Snapshot | None = None
 
+        # Failure artifacts (Phase 1)
+        self._artifact_buffer: FailureArtifactBuffer | None = None
+        self._artifact_timer_task: asyncio.Task | None = None
+
         # Cached URL (updated on snapshot or explicit get_url call)
         self._cached_url: str | None = None
 
@@ -250,6 +255,90 @@ async def snapshot(self, **kwargs: Any) -> Snapshot:
         self.last_snapshot = await backend_snapshot(self.backend, options=options)
         return self.last_snapshot
 
+    async def enable_failure_artifacts(
+        self,
+        options: FailureArtifactsOptions | None = None,
+    ) -> None:
+        """
+        Enable failure artifact buffer (Phase 1).
+        """
+        opts = options or FailureArtifactsOptions()
+        self._artifact_buffer = FailureArtifactBuffer(
+            run_id=self.tracer.run_id,
+            options=opts,
+        )
+        if opts.fps > 0:
+            self._artifact_timer_task = asyncio.create_task(self._artifact_timer_loop())
+
+    def disable_failure_artifacts(self) -> None:
+        """
+        Disable failure artifact buffer and stop background capture.
+        """
+        if self._artifact_timer_task:
+            self._artifact_timer_task.cancel()
+            self._artifact_timer_task = None
+
+    async def record_action(
+        self,
+        action: str,
+        *,
+        url: str | None = None,
+    ) -> None:
+        """
+        Record an action in the artifact timeline and capture a frame if enabled.
+        """
+        if not self._artifact_buffer:
+            return
+        self._artifact_buffer.record_step(
+            action=action,
+            step_id=self.step_id,
+            step_index=self.step_index,
+            url=url,
+        )
+        if self._artifact_buffer.options.capture_on_action:
+            await self._capture_artifact_frame()
+
+    async def _capture_artifact_frame(self) -> None:
+        if not self._artifact_buffer:
+            return
+        try:
+            image_bytes = await self.backend.screenshot_png()
+        except Exception:
+            return
+        self._artifact_buffer.add_frame(image_bytes, fmt="png")
+
+    async def _artifact_timer_loop(self) -> None:
+        if not self._artifact_buffer:
+            return
+        interval = 1.0 / max(0.001, self._artifact_buffer.options.fps)
+        try:
+            while True:
+                await self._capture_artifact_frame()
+                await asyncio.sleep(interval)
+        except asyncio.CancelledError:
+            return
+
+    def finalize_run(self, *, success: bool) -> None:
+        """
+        Finalize artifact buffer at end of run.
+        """
+        if not self._artifact_buffer:
+            return
+        if success:
+            if self._artifact_buffer.options.persist_mode == "always":
+                self._artifact_buffer.persist(reason="success", status="success")
+            self._artifact_buffer.cleanup()
+        else:
+            self._persist_failure_artifacts(reason="finalize_failure")
+
+    def _persist_failure_artifacts(self, *, reason: str) -> None:
+        if not self._artifact_buffer:
+            return
+        self._artifact_buffer.persist(reason=reason, status="failure")
+        self._artifact_buffer.cleanup()
+        if self._artifact_buffer.options.persist_mode == "onFail":
+            self.disable_failure_artifacts()
+
     def begin_step(self, goal: str, step_index: int | None = None) -> str:
         """
         Begin a new step in the verification loop.
@@ -309,6 +398,8 @@ def assert_(
             kind="assert",
             record_in_step=True,
         )
+        if required and not outcome.passed:
+            self._persist_failure_artifacts(reason=f"assert_failed:{label}")
         return outcome.passed
 
     def check(self, predicate: Predicate, label: str, required: bool = False) -> AssertionHandle:
@@ -619,6 +710,10 @@ async def eventually(
                                     "vision_fallback": True,
                                 },
                             )
+                            if self.required and not passed:
+                                self.runtime._persist_failure_artifacts(
+                                    reason=f"assert_eventually_failed:{self.label}"
+                                )
                             return passed
                         except Exception as e:
                             # If vision fallback fails, fall through to snapshot_exhausted.
@@ -649,6 +744,10 @@ async def eventually(
                             "exhausted": True,
                         },
                     )
+                    if self.required:
+                        self.runtime._persist_failure_artifacts(
+                            reason=f"assert_eventually_failed:{self.label}"
+                        )
                     return False
 
                 if time.monotonic() >= deadline:
@@ -666,6 +765,10 @@ async def eventually(
                             "timeout": True,
                         },
                     )
+                    if self.required:
+                        self.runtime._persist_failure_artifacts(
+                            reason=f"assert_eventually_timeout:{self.label}"
+                        )
                     return False
 
                 await asyncio.sleep(poll_s)
@@ -705,6 +808,10 @@ async def eventually(
                     record_in_step=True,
                     extra={"eventually": True, "attempt": attempt, "final": True, "timeout": True},
                 )
+                if self.required:
+                    self.runtime._persist_failure_artifacts(
+                        reason=f"assert_eventually_timeout:{self.label}"
+                    )
                 return False
 
             await asyncio.sleep(poll_s)
diff --git a/sentience/failure_artifacts.py b/sentience/failure_artifacts.py
@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import json
+import shutil
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Literal
+
+
+@dataclass
+class FailureArtifactsOptions:
+    buffer_seconds: float = 15.0
+    capture_on_action: bool = True
+    fps: float = 0.0
+    persist_mode: Literal["onFail", "always"] = "onFail"
+    output_dir: str = ".sentience/artifacts"
+
+
+@dataclass
+class _FrameRecord:
+    ts: float
+    file_name: str
+    path: Path
+
+
+class FailureArtifactBuffer:
+    """
+    Ring buffer of screenshots with minimal persistence on failure.
+    """
+
+    def __init__(
+        self,
+        *,
+        run_id: str,
+        options: FailureArtifactsOptions,
+        time_fn: Callable[[], float] = time.time,
+    ) -> None:
+        self.run_id = run_id
+        self.options = options
+        self._time_fn = time_fn
+        self._temp_dir = Path(tempfile.mkdtemp(prefix="sentience-artifacts-"))
+        self._frames_dir = self._temp_dir / "frames"
+        self._frames_dir.mkdir(parents=True, exist_ok=True)
+        self._frames: list[_FrameRecord] = []
+        self._steps: list[dict] = []
+        self._persisted = False
+
+    @property
+    def temp_dir(self) -> Path:
+        return self._temp_dir
+
+    def record_step(
+        self,
+        *,
+        action: str,
+        step_id: str | None,
+        step_index: int | None,
+        url: str | None,
+    ) -> None:
+        self._steps.append(
+            {
+                "ts": self._time_fn(),
+                "action": action,
+                "step_id": step_id,
+                "step_index": step_index,
+                "url": url,
+            }
+        )
+
+    def add_frame(self, image_bytes: bytes, *, fmt: str = "png") -> None:
+        ts = self._time_fn()
+        file_name = f"frame_{int(ts * 1000)}.{fmt}"
+        path = self._frames_dir / file_name
+        path.write_bytes(image_bytes)
+        self._frames.append(_FrameRecord(ts=ts, file_name=file_name, path=path))
+        self._prune()
+
+    def frame_count(self) -> int:
+        return len(self._frames)
+
+    def _prune(self) -> None:
+        cutoff = self._time_fn() - max(0.0, self.options.buffer_seconds)
+        keep: list[_FrameRecord] = []
+        for frame in self._frames:
+            if frame.ts >= cutoff:
+                keep.append(frame)
+            else:
+                try:
+                    frame.path.unlink(missing_ok=True)
+                except Exception:
+                    pass
+        self._frames = keep
+
+    def persist(self, *, reason: str | None, status: Literal["failure", "success"]) -> Path | None:
+        if self._persisted:
+            return None
+
+        output_dir = Path(self.options.output_dir)
+        output_dir.mkdir(parents=True, exist_ok=True)
+        ts = int(self._time_fn() * 1000)
+        run_dir = output_dir / f"{self.run_id}-{ts}"
+        frames_out = run_dir / "frames"
+        frames_out.mkdir(parents=True, exist_ok=True)
+
+        for frame in self._frames:
+            shutil.copy2(frame.path, frames_out / frame.file_name)
+
+        steps_path = run_dir / "steps.json"
+        steps_path.write_text(json.dumps(self._steps, indent=2))
+
+        manifest = {
+            "run_id": self.run_id,
+            "created_at_ms": ts,
+            "status": status,
+            "reason": reason,
+            "buffer_seconds": self.options.buffer_seconds,
+            "frame_count": len(self._frames),
+            "frames": [
+                {"file": frame.file_name, "ts": frame.ts} for frame in self._frames
+            ],
+        }
+        manifest_path = run_dir / "manifest.json"
+        manifest_path.write_text(json.dumps(manifest, indent=2))
+
+        self._persisted = True
+        return run_dir
+
+    def cleanup(self) -> None:
+        if self._temp_dir.exists():
+            shutil.rmtree(self._temp_dir, ignore_errors=True)
diff --git a/tests/unit/test_failure_artifacts.py b/tests/unit/test_failure_artifacts.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import json
+
+from sentience.failure_artifacts import FailureArtifactBuffer, FailureArtifactsOptions
+
+
+def test_buffer_prunes_by_time(tmp_path) -> None:
+    now = {"t": 0.0}
+
+    def time_fn() -> float:
+        return now["t"]
+
+    opts = FailureArtifactsOptions(buffer_seconds=1.0, output_dir=str(tmp_path))
+    buf = FailureArtifactBuffer(run_id="run-1", options=opts, time_fn=time_fn)
+
+    buf.add_frame(b"first")
+    assert buf.frame_count() == 1
+
+    now["t"] = 2.0
+    buf.add_frame(b"second")
+    assert buf.frame_count() == 1
+
+
+def test_persist_writes_manifest_and_steps(tmp_path) -> None:
+    now = {"t": 10.0}
+
+    def time_fn() -> float:
+        return now["t"]
+
+    opts = FailureArtifactsOptions(output_dir=str(tmp_path))
+    buf = FailureArtifactBuffer(run_id="run-2", options=opts, time_fn=time_fn)
+
+    buf.record_step(action="CLICK", step_id="s1", step_index=1, url="https://example.com")
+    buf.add_frame(b"frame")
+
+    run_dir = buf.persist(reason="assert_failed", status="failure")
+    assert run_dir is not None
+    manifest = json.loads((run_dir / "manifest.json").read_text())
+    steps = json.loads((run_dir / "steps.json").read_text())
+
+    assert manifest["run_id"] == "run-2"
+    assert manifest["frame_count"] == 1
+    assert len(steps) == 1
