fix(data-layer): redact loop event/status/decision outside their allowed sets

.agent/tools/data_layer_export.py

@@ -24,6 +24,21 @@
 VALID_WINDOWS = {"7d", "30d", "90d", "all"}
 VALID_BUCKETS = {"hour", "day", "week", "month"}
 
+# Finite value sets the loop supervisor (harness_manager/loops/runner.py)
+# actually writes to runtime/loops/events.jsonl. normalize_loop_event must
+# redact anything outside these sets rather than copy arbitrary loop-event
+# content into the exported dashboard/analytics surface.
+VALID_LOOP_EVENTS = {
+    "created", "awaiting_approval", "worktree_created", "paused",
+    "interrupted", "maker_finished", "verifier_finished", "checker_finished",
+    "completed", "cancelled",
+}
+VALID_LOOP_STATUSES = {
+    "created", "awaiting_approval", "paused", "exhausted", "interrupted",
+    "completed", "cancelled", "audit_failed", "failed", "rejected",
+}
+VALID_LOOP_DECISIONS = {"APPROVE", "ESCALATE", "MALFORMED"}
+
 
 def _e(*codes: int) -> str:
     return f"\x1b[{';'.join(map(str, codes))}m"
@@ -393,14 +408,30 @@ def normalize_agent_event(entry: dict[str, Any], idx: int, args: argparse.Namesp
     return base
 
 
+def _allowed_or_unknown(value: Any, allowed: set[str]) -> str:
+    text = str(value) if value is not None else ""
+    return text if text in allowed else "unknown"
+
+
 def normalize_loop_event(entry: dict[str, Any]) -> dict[str, Any]:
-    """Map the privacy-whitelisted loop event shape into data-layer fields."""
+    """Map the privacy-whitelisted loop event shape into data-layer fields.
+
+    entry's `event`/`status`/`decision` values come from a supervisor-controlled
+    finite set (VALID_LOOP_EVENTS/STATUSES/DECISIONS); anything else is redacted
+    to "unknown" rather than copied through, so a malformed or unexpected
+    events.jsonl row can't smuggle arbitrary text into the exported surface.
+    """
+    status_or_decision = entry.get("status") or entry.get("decision")
     return {
         "timestamp": entry.get("timestamp") or now_iso(),
         "skill": "agentic-loop",
-        "action": str(entry.get("event") or "loop_event"),
+        "action": _allowed_or_unknown(entry.get("event"), VALID_LOOP_EVENTS),
         "workflow": str(entry.get("loop") or "loop"),
-        "result": str(entry.get("status") or entry.get("decision") or "observed"),
+        "result": (
+            _allowed_or_unknown(status_or_decision, VALID_LOOP_STATUSES | VALID_LOOP_DECISIONS)
+            if status_or_decision is not None
+            else "observed"
+        ),
         "harness": "agentic-loop",
         "source": {"run_id": entry.get("run_id")},
         "privacy_level": "local_only",

tests/test_data_layer_export.py

@@ -148,6 +148,40 @@ def test_exports_privacy_safe_loop_events_and_quality_counts(self):
             self.assertNotIn("do not export", exported)
             self.assertIn("agentic-loop", exported)
 
+    def test_redacts_loop_event_status_and_decision_outside_the_allowed_sets(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            work = Path(tmp)
+            events = work / ".agent" / "runtime" / "loops"
+            events.mkdir(parents=True)
+            (events / "events.jsonl").write_text(
+                "\n".join(
+                    json.dumps(row)
+                    for row in [
+                        {
+                            "run_id": "run-a",
+                            "loop": "ci-sweeper",
+                            "event": "<script>exfiltrate this</script>",
+                        },
+                        {
+                            "run_id": "run-b",
+                            "loop": "ci-sweeper",
+                            "decision": "leaked prompt content here",
+                        },
+                        {"run_id": "run-c", "loop": "ci-sweeper", "event": "completed"},
+                    ]
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            result = self.run_export(work, "--window", "all", "--date", "2026-04-25")
+            self.assertEqual(result.returncode, 0, result.stderr)
+            out = work / ".agent" / "data-layer" / "exports" / "2026-04-25"
+            exported = (out / "agent-events.jsonl").read_text()
+            self.assertNotIn("exfiltrate", exported)
+            self.assertNotIn("leaked prompt content", exported)
+            self.assertNotIn("<script>", exported)
+            self.assertIn("completed", exported)
+
     def test_succeeds_with_empty_inputs(self):
         with tempfile.TemporaryDirectory() as tmp:
             work = Path(tmp)