P3

Author: rcholic

README.md

@@ -104,6 +104,22 @@ await runtime.enable_failure_artifacts(
 await runtime.record_action("CLICK")
 ```
 
+### Redaction callback (Phase 3)
+
+Provide a user-defined callback to redact snapshots and decide whether to persist frames. The SDK does not implement image/video redaction.
+
+```python
+from sentience.failure_artifacts import FailureArtifactsOptions, RedactionContext, RedactionResult
+
+def redact(ctx: RedactionContext) -> RedactionResult:
+    # Example: drop frames entirely, keep JSON only.
+    return RedactionResult(drop_frames=True)
+
+await runtime.enable_failure_artifacts(
+    FailureArtifactsOptions(on_before_persist=redact)
+)
+```
+
 **See examples:** [`examples/asserts/`](examples/asserts/)
 
 ## 🚀 Quick Start: Choose Your Abstraction Level

sentience/failure_artifacts.py

@@ -17,6 +17,27 @@ class FailureArtifactsOptions:
     fps: float = 0.0
     persist_mode: Literal["onFail", "always"] = "onFail"
     output_dir: str = ".sentience/artifacts"
+    on_before_persist: Callable[[RedactionContext], RedactionResult] | None = None
+    redact_snapshot_values: bool = True
+
+
+@dataclass
+class RedactionContext:
+    run_id: str
+    reason: str | None
+    status: Literal["failure", "success"]
+    snapshot: Any | None
+    diagnostics: Any | None
+    frame_paths: list[str]
+    metadata: dict[str, Any]
+
+
+@dataclass
+class RedactionResult:
+    snapshot: Any | None = None
+    diagnostics: Any | None = None
+    frame_paths: list[str] | None = None
+    drop_frames: bool = False
 
 
 @dataclass
@@ -99,6 +120,27 @@ def _write_json_atomic(self, path: Path, data: Any) -> None:
         tmp_path.write_text(json.dumps(data, indent=2))
         tmp_path.replace(path)
 
+    def _redact_snapshot_defaults(self, payload: Any) -> Any:
+        if not isinstance(payload, dict):
+            return payload
+        elements = payload.get("elements")
+        if not isinstance(elements, list):
+            return payload
+        redacted = []
+        for el in elements:
+            if not isinstance(el, dict):
+                redacted.append(el)
+                continue
+            input_type = (el.get("input_type") or "").lower()
+            if input_type in {"password", "email", "tel"} and "value" in el:
+                el = dict(el)
+                el["value"] = None
+                el["value_redacted"] = True
+            redacted.append(el)
+        payload = dict(payload)
+        payload["elements"] = redacted
+        return payload
+
     def persist(
         self,
         *,
@@ -118,25 +160,59 @@ def persist(
         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)
-
-        self._write_json_atomic(run_dir / "steps.json", self._steps)
-
         snapshot_payload = None
         if snapshot is not None:
             if hasattr(snapshot, "model_dump"):
                 snapshot_payload = snapshot.model_dump()
             else:
                 snapshot_payload = snapshot
-            self._write_json_atomic(run_dir / "snapshot.json", snapshot_payload)
+            if self.options.redact_snapshot_values:
+                snapshot_payload = self._redact_snapshot_defaults(snapshot_payload)
 
         diagnostics_payload = None
         if diagnostics is not None:
             if hasattr(diagnostics, "model_dump"):
                 diagnostics_payload = diagnostics.model_dump()
             else:
                 diagnostics_payload = diagnostics
+
+        frame_paths = [str(frame.path) for frame in self._frames]
+        drop_frames = False
+
+        if self.options.on_before_persist is not None:
+            try:
+                result = self.options.on_before_persist(
+                    RedactionContext(
+                        run_id=self.run_id,
+                        reason=reason,
+                        status=status,
+                        snapshot=snapshot_payload,
+                        diagnostics=diagnostics_payload,
+                        frame_paths=frame_paths,
+                        metadata=metadata or {},
+                    )
+                )
+                if result.snapshot is not None:
+                    snapshot_payload = result.snapshot
+                if result.diagnostics is not None:
+                    diagnostics_payload = result.diagnostics
+                if result.frame_paths is not None:
+                    frame_paths = result.frame_paths
+                drop_frames = result.drop_frames
+            except Exception:
+                drop_frames = True
+
+        if not drop_frames:
+            for frame_path in frame_paths:
+                src = Path(frame_path)
+                if not src.exists():
+                    continue
+                shutil.copy2(src, frames_out / src.name)
+
+        self._write_json_atomic(run_dir / "steps.json", self._steps)
+        if snapshot_payload is not None:
+            self._write_json_atomic(run_dir / "snapshot.json", snapshot_payload)
+        if diagnostics_payload is not None:
             self._write_json_atomic(run_dir / "diagnostics.json", diagnostics_payload)
 
         manifest = {
@@ -145,11 +221,15 @@ def persist(
             "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],
+            "frame_count": 0 if drop_frames else len(frame_paths),
+            "frames": (
+                [] if drop_frames else [{"file": Path(p).name, "ts": None} for p in frame_paths]
+            ),
             "snapshot": "snapshot.json" if snapshot_payload is not None else None,
             "diagnostics": "diagnostics.json" if diagnostics_payload is not None else None,
             "metadata": metadata or {},
+            "frames_redacted": not drop_frames and self.options.on_before_persist is not None,
+            "frames_dropped": drop_frames,
         }
         self._write_json_atomic(run_dir / "manifest.json", manifest)
 

tests/unit/test_failure_artifacts.py

@@ -2,7 +2,12 @@
 
 import json
 
-from sentience.failure_artifacts import FailureArtifactBuffer, FailureArtifactsOptions
+from sentience.failure_artifacts import (
+    FailureArtifactBuffer,
+    FailureArtifactsOptions,
+    RedactionContext,
+    RedactionResult,
+)
 
 
 def test_buffer_prunes_by_time(tmp_path) -> None:
@@ -34,7 +39,14 @@ def time_fn() -> float:
     buf.record_step(action="CLICK", step_id="s1", step_index=1, url="https://example.com")
     buf.add_frame(b"frame")
 
-    snapshot = {"status": "success", "url": "https://example.com", "elements": []}
+    snapshot = {
+        "status": "success",
+        "url": "https://example.com",
+        "elements": [
+            {"id": 1, "input_type": "password", "value": "secret"},
+            {"id": 2, "input_type": "email", "value": "user@example.com"},
+        ],
+    }
     diagnostics = {"confidence": 0.9, "reasons": ["ok"], "metrics": {"quiet_ms": 42}}
     run_dir = buf.persist(
         reason="assert_failed",
@@ -57,3 +69,24 @@ def time_fn() -> float:
     assert len(steps) == 1
     assert snap_json["url"] == "https://example.com"
     assert diag_json["confidence"] == 0.9
+    assert snap_json["elements"][0]["value"] is None
+    assert snap_json["elements"][0]["value_redacted"] is True
+    assert snap_json["elements"][1]["value"] is None
+    assert snap_json["elements"][1]["value_redacted"] is True
+
+
+def test_redaction_callback_can_drop_frames(tmp_path) -> None:
+    opts = FailureArtifactsOptions(output_dir=str(tmp_path))
+
+    def redactor(ctx: RedactionContext) -> RedactionResult:
+        return RedactionResult(drop_frames=True)
+
+    opts.on_before_persist = redactor
+    buf = FailureArtifactBuffer(run_id="run-3", options=opts)
+    buf.add_frame(b"frame")
+
+    run_dir = buf.persist(reason="fail", status="failure", snapshot={"status": "success"})
+    assert run_dir is not None
+    manifest = json.loads((run_dir / "manifest.json").read_text())
+    assert manifest["frame_count"] == 0
+    assert manifest["frames_dropped"] is True