Retain bounded private pilot events on explicit opt-in

PILOT.md

@@ -1782,3 +1782,7 @@ Both arms observed the initial focused failure, repaired only the allowed source
 The treatment inspected all configured evidence and made **zero bridge requests and zero Jev transport calls**. Local resolution was permitted; no model ranking or diagnostic recommendation was delivered. The exact workflow acknowledgment was invalid, so protocol delivery failed and the overall receipt remains **incomplete**, despite correct repairs. The timing difference is descriptive single-pair evidence only: treatment-first order, unequal observed token/cache usage and extra workflow instructions prevent attribution to Jev. It does not prove native Desktop delivery, repeatable speed or quality improvement. Raw events were not durably archived by this runner; retained measurements and task receipts cannot establish the cause of the missing acknowledgment.
 
 Next: preserve this outcome unchanged, add bounded private event retention for future diagnostic trials, and evaluate genuinely ambiguous cases separately. Do not force remote advice on locally resolved contracts.
+
+## VCR398 — Optional private event retention for future trials
+
+The runner now exposes --retain-private-events to retain bounded raw CLI events in mode-0600 local files before task-specific parsing. Default execution retains its current behavior. Only capture status and byte count belong in receipts; raw prompts, source and outputs must never be copied into public documentation, logs or decision-service requests. The feature must reject oversize archives and existing/symlink targets, preserve timeout/parser-failure evidence, and leave historical VCR396 unchanged. Verification: 39 focused/regression tests pass; the complete local suite passes 818 tests in 37.686 seconds. Green hosted CI remains the merge gate.

TASKS.md

@@ -1186,3 +1186,7 @@ Both arms observed the initial focused failure, repaired only the allowed source
 The treatment inspected all configured evidence and made **zero bridge requests and zero Jev transport calls**. Local resolution was permitted; no model ranking or diagnostic recommendation was delivered. The exact workflow acknowledgment was invalid, so protocol delivery failed and the overall receipt remains **incomplete**, despite correct repairs. The timing difference is descriptive single-pair evidence only: treatment-first order, unequal observed token/cache usage and extra workflow instructions prevent attribution to Jev. It does not prove native Desktop delivery, repeatable speed or quality improvement. Raw events were not durably archived by this runner; retained measurements and task receipts cannot establish the cause of the missing acknowledgment.
 
 Next: preserve this outcome unchanged, add bounded private event retention for future diagnostic trials, and evaluate genuinely ambiguous cases separately. Do not force remote advice on locally resolved contracts.
+
+## VCR398 — Optional private event retention for future trials
+
+The runner now exposes --retain-private-events to retain bounded raw CLI events in mode-0600 local files before task-specific parsing. Default execution retains its current behavior. Only capture status and byte count belong in receipts; raw prompts, source and outputs must never be copied into public documentation, logs or decision-service requests. The feature must reject oversize archives and existing/symlink targets, preserve timeout/parser-failure evidence, and leave historical VCR396 unchanged. Verification: 39 focused/regression tests pass; the complete local suite passes 818 tests in 37.686 seconds. Green hosted CI remains the merge gate.

scripts/pilot_contract_triage_pair.py

@@ -840,6 +840,33 @@ def _empty_arm(failure: str) -> dict[str, Any]:
         "codex_billing_estimate": None,
     }
 
+def _write_private_event_archive(
+    lines: list[str], archive_path: Path,
+) -> dict[str, Any]:
+    """Write bounded raw CLI events once to a private local-only JSONL file."""
+    payload = ("\n".join(lines) + ("\n" if lines else "")).encode("utf-8")
+    if len(payload) > core.MAX_EVENT_BYTES:
+        return {
+            "private_event_archive_captured": False,
+            "private_event_archive_bytes": None,
+        }
+    directory_fd = None
+    try:
+        directory_fd, _ = core._private_directory_fd(archive_path.parent)
+        core._write_new_file_at(directory_fd, archive_path.name, payload, 0o600)
+    except (OSError, ValueError):
+        return {
+            "private_event_archive_captured": False,
+            "private_event_archive_bytes": None,
+        }
+    finally:
+        if directory_fd is not None:
+            os.close(directory_fd)
+    return {
+        "private_event_archive_captured": True,
+        "private_event_archive_bytes": len(payload),
+    }
+
 
 def _run_arm(
     *, codex: str, model: str, reasoning_effort: str, prompt: str,
@@ -851,6 +878,7 @@ def _run_arm(
     accepted_triage_statuses: tuple[str, ...] | None = None,
     allow_network: bool = False,
     max_tokens: int | None = None,
+    retain_private_events: bool = False,
 ) -> tuple[dict[str, Any], str | None]:
     (home / ".codex").mkdir(mode=0o700, parents=True, exist_ok=True)
     measurement_path = measurement_path or (home / ".codex" / "agent-measurement.json")
@@ -959,6 +987,12 @@ def observe(event: dict[str, Any]) -> None:
     measurement = build_agent_measurement_receipt(
         lines, times, started, ended, process.returncode, failure,
     )
+    private_archive = (
+        _write_private_event_archive(
+            lines, measurement_path.with_name(measurement_path.stem + "-events.jsonl"),
+        )
+        if retain_private_events else {}
+    )
     # Preserve the bounded event measurement before task-specific parsing.
     common._private_write(
         measurement_path,
@@ -994,6 +1028,7 @@ def observe(event: dict[str, Any]) -> None:
         failed["cli_exit_code"] = process.returncode
         failed["agent_measurement"] = measurement
         failed["event_count"] = len(lines)
+        failed.update(private_archive)
         if bridge_summary is not None:
             failed["profile_triage_bridge"] = bridge_summary
             failed["profile_triage_typed_receipt"] = typed_bridge_receipt
@@ -1016,6 +1051,7 @@ def observe(event: dict[str, Any]) -> None:
             "profile_triage_bridge": bridge_summary,
             "profile_triage_typed_receipt": typed_bridge_receipt,
             **bridge_metadata,
+            **private_archive,
         })
         return failed, None
 
@@ -1026,6 +1062,7 @@ def observe(event: dict[str, Any]) -> None:
         "completion_ms": wall_ms if wall_ms <= MAX_TIMEOUT * 1000 else None,
         "agent_measurement": measurement,
         **observed,
+        **private_archive,
     }
     if bridge_summary is not None:
         result["profile_triage_bridge"] = bridge_summary
@@ -1192,6 +1229,7 @@ def run_pair(
     remote_profile_triage: bool = False,
     accepted_remote_statuses: tuple[str, ...] = (),
     max_tokens: int | None = None,
+    retain_private_events: bool = False,
 ) -> dict[str, Any]:
     """Run both local-only CLI arms and save private blind artifacts/receipts."""
     if case_profile is not None:
@@ -1225,6 +1263,8 @@ def run_pair(
         or not 1 <= max_tokens <= core.MAX_EVENT_TOKEN_BUDGET
     ):
         raise ValueError("max_tokens is outside the supported event-token budget")
+    if type(retain_private_events) is not bool:
+        raise ValueError("retain_private_events must be boolean")
     if not _verify_codex_version(codex):
         raise ValueError("Codex CLI 0.157.0 is required")
 
@@ -1325,6 +1365,7 @@ def run_pair(
                 ),
                 allow_network=remote_profile_triage,
                 max_tokens=max_tokens,
+                **({"retain_private_events": True} if retain_private_events else {}),
             )
             if repair_profile:
                 arms[label].setdefault("agent_git_diff_check_invocation_observed", False)
@@ -1605,6 +1646,8 @@ def main(argv: list[str] | None = None) -> int:
                         help="explicit supervisor acceptance status (repeat as needed)")
     parser.add_argument("--max-tokens", type=int,
                         help="observed completed-turn token cap; not a provider-side limit")
+    parser.add_argument("--retain-private-events", action="store_true",
+                        help="archive bounded raw CLI JSONL locally in each private arm directory")
     parser.add_argument("--output-dir", type=Path, required=True,
                         help="new private directory for blind receipts and artifacts")
     args = parser.parse_args(argv)
@@ -1624,6 +1667,7 @@ def main(argv: list[str] | None = None) -> int:
             remote_profile_triage=args.remote_profile_triage,
             accepted_remote_statuses=tuple(args.accept_profile_triage_status),
             max_tokens=args.max_tokens,
+            retain_private_events=args.retain_private_events,
         )
     except (OSError, ValueError, RuntimeError):
         print(json.dumps({"status": "failed", "failure": "runner_setup_failed"}))

tests/test_pilot_profile_event_retention.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import io
+import json
+import os
+from pathlib import Path
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path[:0] = [str(ROOT / "scripts"), str(ROOT / "src")]
+
+import pilot_contract_triage_pair as runner
+import pilot_cli_core as core
+
+
+class PrivateEventRetentionTests(unittest.TestCase):
+    def _fixture(self, root: Path) -> Path:
+        fixture = root / "fixture"
+        fixture.mkdir()
+        return fixture
+
+    def _run(self, root: Path, *, source: str, retain: bool, timeout: int = 5,
+             parser_error: bool = False):
+        fixture = self._fixture(root)
+        output = root / "private-output"
+        output.mkdir(mode=0o700)
+        measurement = output / "agent-measurement.json"
+        with (
+            mock.patch.object(
+                runner.common, "_cli_command",
+                return_value=[sys.executable, "-c", source],
+            ),
+            mock.patch.object(runner, "_event_receipts",
+                              side_effect=RuntimeError("parser failed")
+                              if parser_error else lambda *a, **k: ({"parsed": True}, "answer")),
+        ):
+            return runner._run_arm(
+                codex="fake", model="test-model", reasoning_effort="low",
+                prompt="synthetic", fixture=fixture, home=root / "home",
+                timeout=timeout, treatment=False, measurement_path=measurement,
+                retain_private_events=retain,
+            ), measurement, output
+
+    def test_success_archives_private_jsonl_and_receipt_exposes_metadata_only(self):
+        raw_line = json.dumps({"event": "private-sentinel", "secret": "do-not-copy-to-receipt"})
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            source = f"print({raw_line!r}, flush=True)"
+            (result, answer), measurement, output = self._run(
+                root, source=source, retain=True,
+            )
+            archive = output / "agent-measurement-events.jsonl"
+            self.assertEqual(answer, "answer")
+            self.assertTrue(archive.is_file())
+            self.assertEqual(archive.read_text(encoding="utf-8"), raw_line + "\n")
+            self.assertEqual(stat.S_IMODE(archive.stat().st_mode), 0o600)
+            self.assertTrue(result["private_event_archive_captured"])
+            self.assertEqual(result["private_event_archive_bytes"], archive.stat().st_size)
+            self.assertNotIn("private-sentinel", json.dumps(result))
+            self.assertNotIn("private-sentinel", measurement.read_text(encoding="utf-8"))
+            self.assertEqual(json.loads(measurement.read_text())["status"], "completed")
+
+    def test_parser_error_keeps_archive_and_redacted_measurement(self):
+        raw_line = json.dumps({"event": "parser-failure-private"})
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            (result, answer), measurement, output = self._run(
+                root, source=f"print({raw_line!r}, flush=True)",
+                retain=True, parser_error=True,
+            )
+            archive = output / "agent-measurement-events.jsonl"
+            self.assertEqual(result["failure"], "event_parser_error")
+            self.assertIsNone(answer)
+            self.assertEqual(archive.read_text(encoding="utf-8"), raw_line + "\n")
+            self.assertTrue(result["private_event_archive_captured"])
+            self.assertNotIn("parser-failure-private", json.dumps(result))
+            self.assertTrue(measurement.is_file())
+
+    def test_timeout_keeps_events_collected_before_timeout(self):
+        raw_line = json.dumps({"event": "timeout-private"})
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            source = f"import time; print({raw_line!r}, flush=True); time.sleep(3)"
+            (result, answer), measurement, output = self._run(
+                root, source=source, retain=True, timeout=1,
+            )
+            archive = output / "agent-measurement-events.jsonl"
+            self.assertEqual(result["failure"], "timeout")
+            self.assertIsNone(answer)
+            self.assertEqual(archive.read_text(encoding="utf-8"), raw_line + "\n")
+            self.assertTrue(result["private_event_archive_captured"])
+            self.assertTrue(measurement.is_file())
+
+    def test_over_limit_archive_is_not_created(self):
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "oversized.jsonl"
+            result = runner._write_private_event_archive(
+                ["x" * (core.MAX_EVENT_BYTES + 1)], path,
+            )
+            self.assertEqual(result, {
+                "private_event_archive_captured": False,
+                "private_event_archive_bytes": None,
+            })
+            self.assertFalse(path.exists())
+
+    def test_existing_or_symlink_target_is_never_overwritten(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            target = root / "events.jsonl"
+            target.write_text("original\n", encoding="utf-8")
+            result = runner._write_private_event_archive(["new"], target)
+            self.assertFalse(result["private_event_archive_captured"])
+            self.assertEqual(target.read_text(encoding="utf-8"), "original\n")
+
+            outside = root / "outside.txt"
+            outside.write_text("untouched\n", encoding="utf-8")
+            link = root / "linked-events.jsonl"
+            link.symlink_to(outside)
+            result = runner._write_private_event_archive(["new"], link)
+            self.assertFalse(result["private_event_archive_captured"])
+            self.assertEqual(outside.read_text(encoding="utf-8"), "untouched\n")
+
+    def test_default_path_does_not_create_archive_or_add_metadata(self):
+        raw_line = json.dumps({"event": "default-private"})
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            (result, answer), _measurement, output = self._run(
+                root, source=f"print({raw_line!r}, flush=True)", retain=False,
+            )
+            self.assertEqual(answer, "answer")
+            self.assertFalse((output / "agent-measurement-events.jsonl").exists())
+            self.assertNotIn("private_event_archive_captured", result)
+            self.assertNotIn("private_event_archive_bytes", result)
+
+    def test_cli_exposes_explicit_archive_optin(self):
+        stdout = io.StringIO()
+        with mock.patch("sys.stdout", stdout):
+            with self.assertRaises(SystemExit) as exit_info:
+                runner.main(["--help"])
+        self.assertEqual(exit_info.exception.code, 0)
+        self.assertIn("--retain-private-events", stdout.getvalue())
+
+    def test_pair_explicit_optin_is_forwarded_to_both_arms(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            seen = []
+
+            def strict_arm(*, retain_private_events, **kwargs):
+                seen.append((kwargs["treatment"], retain_private_events))
+                return runner._empty_arm("synthetic_offline"), None
+
+            with (
+                mock.patch.object(runner, "_verify_codex_version", return_value=True),
+                mock.patch.object(core, "_copy_auth", return_value=True),
+                mock.patch.object(runner, "_run_arm", side_effect=strict_arm),
+            ):
+                runner.run_pair(
+                    codex="fake", model="test-model", reasoning_effort="low",
+                    timeout=5, seed=1, output_dir=root / "pair",
+                    fixture_source=runner.FIXTURE,
+                    retain_private_events=True,
+                )
+            self.assertEqual(sorted(seen), [(False, True), (True, True)])
+
+
+if __name__ == "__main__":
+    unittest.main()