Instrument Phase 2 and Phase 3 boundaries with clock-authored phase events (#1853)

.changeset/issue-1853-phase-event-instrumentation.md

@@ -0,0 +1,6 @@
+---
+bump: patch
+type: Added
+---
+
+- **Add a clock-authored `event` subcommand to `scripts/verification-flight.py` and instrument the Phase 2 and Phase 3 boundaries with it.** The subcommand appends a `{"event": …, "recorded_at": …}` record — timestamped from the helper's own clock — to an append-only JSONL log under `.prflow/logs/phase-events/`, and always exits 0 so a failed write only breadcrumbs and never blocks the run. The implement Phase 2 durability-checkpoint boundaries and the Phase 3 `/simplify`, reviewer-dispatch/return, and shadow-entry boundaries now emit one such event, so a long or expensive implement run's interior timeline is reconstructible from disk. (#1961)

.gitignore

@@ -19,6 +19,11 @@
 # per-checkout diagnostics, not a relayed record (see docs/internal/efficiency-trace.md) —
 # keep it out of the tree so a run's flight events are never committed.
 /.prflow/logs/verification-flight/
+# Phase-boundary event log (issue #1853): LOCAL per-run diagnostics written by
+# verification-flight.py's `event` subcommand. Must come AFTER the !/.prflow/logs/
+# negation, which would otherwise re-include it and let a run's transient phase
+# events be committed.
+/.prflow/logs/phase-events/
 # Per-run review workpads. These lived in the tree before issue #441 moved the
 # durable copies to the prflow-telemetry branch, which is now their canonical
 # home; lib/efficiency-trace.sh reads them from there and keeps the working-tree

docs/internal/DEVFLOW_SYSTEM_OVERVIEW.md

@@ -416,6 +416,15 @@ A PRFlow lifecycle can launch the same multi-minute verification profile many ti
 - **Unknown is never a pass.** States are exactly `claimed`, `running`, `passed`, `failed`, `timed_out`, `cancelled`, `stale`, `incomplete`. Only `passed` with complete matching bindings satisfies verification. Missing/partial/timed-out/unreadable/stale state is an attributable non-pass and never authorizes an automatic relaunch; a lease that expires before `mark-running` becomes `incomplete`; checkout drift becomes `stale`; missing terminal evidence becomes `incomplete`. A `wait` bound elapsing returns a non-mutating `wait_expired` observation — only the owner records a terminal `timed_out`.
 - **Consumers.** Implement persists the handle + terminal evidence in its durable state and re-anchors after nested work/compaction; Review-and-Fix persists them in its `iter-<N>.json`. On `wait_expired`, Implement and implement-driven Review-and-Fix take the existing **Blocked** path without relaunch; standalone Review-and-Fix records `verification_evidence.result: skipped` (reason `flight_wait_expired`) and cannot report a clean pass. **CI-grounded standalone `/prflow:review` creates no flight and runs no verification command.** Config lives under the versioned `verification_flight` namespace; the vendored helper is granted in `devflow.yml` and `devflow-implement.yml` only — `devflow-runner.yml` gains no grant.
 
+### Phase-boundary event instrumentation (issue #1853)
+
+Phase 2 and Phase 3 dominate a long implement run's wall clock, but historically neither wrote a clock-authored timing record, so an expensive run's interior — where the time and money went between phase boundaries — could not be reconstructed after the fact from anything but untimestamped `- [x]` ticks and agent-volunteered aggregates (an unestablished measurement under issue #1489). The `event <name>` subcommand on `scripts/verification-flight.py` closes that gap by **generalizing the same clock-authored event shape** the single-flight coordinator already uses — rather than adding a second recording idiom.
+
+- **The subcommand.** `verification-flight.py event <name> [--payload <json>] [--log-dir <dir>]` appends one record `{"event": <name>, "recorded_at": <iso>}` to an append-only JSONL log at `.prflow/logs/phase-events/phase-events.jsonl` (`O_APPEND`; the directory is created best-effort). The timestamp originates in the helper's **own clock** (`_now()`/`_iso()`), never the caller — so the record cannot be model-forged. An optional `--payload` object merges extra keys in, but `event` and `recorded_at` are the record's clock-authored identity and a payload key can never shadow them. A malformed or non-object `--payload` — a JSON `null` included — is dropped with a stderr breadcrumb and the **base event is still recorded**, while an empty `--payload` is treated as absent, like the flag's own default.
+- **Never blocking.** The append **always exits 0**; a failed write (e.g. an unwritable log directory) emits a stderr breadcrumb and the run continues. Instrumentation that can fail a run would be worse than the observability gap it closes, so every instrumented boundary invokes it best-effort, with the vendored literal as the command's leading token (no `VAR=` prefix, expansion, or redirect, which the cloud matcher would silently deny).
+- **The instrumented boundaries.** The implement Phase 2 durability-checkpoint boundaries emit `phase2-checkpoint` (`skills/implement/phases/phase-2-implement.md`, `phase-2-sweeps-quality.md`, at each sub-step boundary where `scripts/phase2-durability-checkpoint.sh` already runs); Phase 3 emits `phase3-simplify-start` / `phase3-simplify-end` around `/simplify` (`skills/implement/phases/phase-3-review.md`), `phase3-reviewers-dispatch` / `phase3-reviewers-return` around the Phase 3.3 `review-and-fix` dispatch and its return (`skills/implement/phases/phase-3-fix-loop.md`), and `phase3-shadow-entry` at shadow-review entry (`skills/review-and-fix/references/shadow-review.md`). The result is a run's interior timeline reconstructible from disk without trusting the agent's recollection.
+- **Scope.** No new agent-volunteered duration field is added and no existing telemetry field changes meaning. The helper is already granted with a wildcard over its arguments (`Bash(.prflow/vendor/prflow/scripts/verification-flight.py:*)` in `lib/capability-profiles.json`), so the new subcommand adds **no** grant and avoids the trigger-time grant bootstrap. The phase-prose additions are agent-executed prompt prose with no wording pin; the helper's own subcommand carries the automated boundary (record shape, clock-authored timestamp, exit-0 contract, failed-write breadcrumb), owned by the `harness-python-guards` module. Out of scope, each owned elsewhere: the whole-suite command's own elapsed time (#1808) and in-flight wait/stall detection (#1027).
+
 ### Receiving-review session artifacts (issue #668)
 
 The `receiving-code-review` skill's **Reception Preflight** now produces a machine-checkable session identity instead of only rendering in-chat facts. Two bundled helpers ship (by the `prflow_version` vendor fetch), and their cloud grants ship by `install.sh`'s workflow copy loop — two independently-upgraded artifacts, so the two-halves upgrade coupling applies.

lib/test/test_verification_flight.py

@@ -28,7 +28,7 @@
 import tempfile
 import unittest
 from unittest import mock
-from contextlib import redirect_stdout
+from contextlib import redirect_stdout, redirect_stderr
 from pathlib import Path
 
 #: Single source of truth for the no-execution sweep (issue #528).
@@ -1749,5 +1749,162 @@ def test_ineffective_backdate_raises(self):
         self.assertIn("backdate ineffective", str(ctx.exception))
 
 
+class TestPhaseEventAppend(Harness):
+    """Issue #1853: the `event` subcommand appends clock-authored phase-boundary
+    events to an append-only JSONL log, always exits 0, and breadcrumbs a failed
+    write instead of blocking the run."""
+
+    def _events_file(self, log_dir):
+        return Path(log_dir) / vf.PHASE_EVENTS_FILENAME
+
+    def test_appends_clock_authored_record(self):
+        log_dir = os.path.join(self.tmp, "phase-events")
+        os.environ["DEVFLOW_FLIGHT_NOW"] = "1000000000"
+        code, _ = self.run_cmd(["event", "simplify-start", "--log-dir", log_dir])
+        self.assertEqual(code, vf.EXIT_OK)
+        lines = self._events_file(log_dir).read_text(encoding="utf-8").splitlines()
+        self.assertEqual(len(lines), 1)
+        rec = json.loads(lines[0])
+        self.assertEqual(rec["event"], "simplify-start")
+        # recorded_at is the helper's own clock (via _now/_iso), never model-volunteered.
+        self.assertEqual(rec["recorded_at"], vf._iso(1000000000))
+        self.assertEqual(set(rec), {"event", "recorded_at"})
+
+    def test_append_only_accumulates(self):
+        log_dir = os.path.join(self.tmp, "phase-events")
+        for name in ("simplify-start", "simplify-end", "reviewer-dispatch"):
+            code, _ = self.run_cmd(["event", name, "--log-dir", log_dir])
+            self.assertEqual(code, vf.EXIT_OK)
+        lines = self._events_file(log_dir).read_text(encoding="utf-8").splitlines()
+        self.assertEqual(
+            [json.loads(line)["event"] for line in lines],
+            ["simplify-start", "simplify-end", "reviewer-dispatch"],
+        )
+
+    def test_default_location_under_prflow_logs(self):
+        # No --log-dir: the record must actually land under <cwd>/.prflow/logs/ per
+        # the AC — exercise the Path.cwd()/PHASE_EVENTS_DIRNAME default, not just the
+        # constant. chdir into the scratch tmp so the write never pollutes the repo.
+        self.assertIn(os.path.join(".prflow", "logs"), vf.PHASE_EVENTS_DIRNAME)
+        cwd = os.getcwd()
+        os.chdir(self.tmp)
+        try:
+            code, _ = self.run_cmd(["event", "phase2-checkpoint"])
+        finally:
+            os.chdir(cwd)
+        self.assertEqual(code, vf.EXIT_OK)
+        landed = Path(self.tmp) / vf.PHASE_EVENTS_DIRNAME / vf.PHASE_EVENTS_FILENAME
+        self.assertTrue(landed.is_file())
+        self.assertEqual(json.loads(landed.read_text(encoding="utf-8"))["event"], "phase2-checkpoint")
+
+    def test_optional_payload_merged_reserved_keys_protected(self):
+        log_dir = os.path.join(self.tmp, "phase-events")
+        os.environ["DEVFLOW_FLIGHT_NOW"] = "1000000000"
+        code, _ = self.run_cmd([
+            "event", "reviewer-dispatch", "--log-dir", log_dir,
+            "--payload", json.dumps(
+                {"agent": "code-reviewer", "event": "SPOOF", "recorded_at": "SPOOF"}
+            ),
+        ])
+        self.assertEqual(code, vf.EXIT_OK)
+        rec = json.loads(
+            self._events_file(log_dir).read_text(encoding="utf-8").splitlines()[0]
+        )
+        self.assertEqual(rec["agent"], "code-reviewer")
+        # event/recorded_at are clock-authored and never shadowed by a payload key.
+        self.assertEqual(rec["event"], "reviewer-dispatch")
+        self.assertEqual(rec["recorded_at"], vf._iso(1000000000))
+
+    def test_failed_write_breadcrumbs_and_exits_zero(self):
+        # A log-dir that cannot be created (a path under a regular file) drives the
+        # failed-write arm: exit 0, a specific stderr breadcrumb, run continues.
+        blocker = os.path.join(self.tmp, "not-a-dir")
+        Path(blocker).write_text("x", encoding="utf-8")
+        log_dir = os.path.join(blocker, "phase-events")
+        buf_out, buf_err = io.StringIO(), io.StringIO()
+        with redirect_stdout(buf_out), redirect_stderr(buf_err):
+            code = vf.main(["event", "simplify-start", "--log-dir", log_dir])
+        self.assertEqual(code, vf.EXIT_OK)
+        self.assertIn("phase event", buf_err.getvalue())
+        self.assertIn("simplify-start", buf_err.getvalue())
+
+    def test_unparseable_payload_breadcrumbs_but_still_records(self):
+        log_dir = os.path.join(self.tmp, "phase-events")
+        buf_err = io.StringIO()
+        with redirect_stderr(buf_err):
+            code, _ = self.run_cmd(
+                ["event", "shadow-entry", "--log-dir", log_dir, "--payload", "{not json"]
+            )
+        self.assertEqual(code, vf.EXIT_OK)
+        rec = json.loads(
+            self._events_file(log_dir).read_text(encoding="utf-8").splitlines()[0]
+        )
+        self.assertEqual(rec["event"], "shadow-entry")
+        self.assertIn("unparseable", buf_err.getvalue())
+        # exactly one breadcrumb: the parse-error arm must not also fall through to
+        # the non-object arm, which is what a shared None sentinel would have caused.
+        self.assertNotIn("non-object", buf_err.getvalue())
+
+    def test_non_object_payload_shapes_all_breadcrumb(self):
+        # The payload parser is a best-effort parser over caller-supplied JSON, so the
+        # non-object shapes are swept together rather than only the array row: a JSON
+        # `null` parses to None and must not be mistaken for the parse-error sentinel.
+        for label, raw in (
+            ("null", "null"),
+            ("number scalar", "42"),
+            ("string scalar", '"str"'),
+            ("array", "[1, 2]"),
+        ):
+            with self.subTest(payload=label):
+                log_dir = os.path.join(self.tmp, "phase-events-" + label.replace(" ", "-"))
+                buf_err = io.StringIO()
+                with redirect_stderr(buf_err):
+                    code, _ = self.run_cmd(
+                        ["event", "shadow-entry", "--log-dir", log_dir, "--payload", raw]
+                    )
+                self.assertEqual(code, vf.EXIT_OK)
+                rec = json.loads(
+                    self._events_file(log_dir).read_text(encoding="utf-8").splitlines()[0]
+                )
+                # the base event is still recorded, with no payload key merged in
+                self.assertEqual(rec["event"], "shadow-entry")
+                self.assertEqual(set(rec), {"event", "recorded_at"})
+                self.assertIn("non-object", buf_err.getvalue())
+                self.assertNotIn("unparseable", buf_err.getvalue())
+
+    def test_empty_payload_is_absent_and_empty_object_merges_nothing(self):
+        for label, raw in (("empty string", ""), ("empty object", "{}")):
+            with self.subTest(payload=label):
+                log_dir = os.path.join(self.tmp, "phase-events-" + label.replace(" ", "-"))
+                buf_err = io.StringIO()
+                with redirect_stderr(buf_err):
+                    code, _ = self.run_cmd(
+                        ["event", "phase2-checkpoint", "--log-dir", log_dir, "--payload", raw]
+                    )
+                self.assertEqual(code, vf.EXIT_OK)
+                rec = json.loads(
+                    self._events_file(log_dir).read_text(encoding="utf-8").splitlines()[0]
+                )
+                self.assertEqual(set(rec), {"event", "recorded_at"})
+                # neither shape is malformed or non-object, so neither breadcrumbs
+                self.assertEqual(buf_err.getvalue(), "")
+
+    def test_valid_falsy_payload_values_survive_the_merge(self):
+        # The repo's off-switch bug class: a real 0 / false / "" merged from a payload
+        # must reach the record, never be dropped by a truthiness test on the value.
+        log_dir = os.path.join(self.tmp, "phase-events")
+        code, _ = self.run_cmd([
+            "event", "phase3-simplify-end", "--log-dir", log_dir,
+            "--payload", json.dumps({"count": 0, "reused": False, "note": ""}),
+        ])
+        self.assertEqual(code, vf.EXIT_OK)
+        rec = json.loads(
+            self._events_file(log_dir).read_text(encoding="utf-8").splitlines()[0]
+        )
+        self.assertEqual(rec["count"], 0)
+        self.assertIs(rec["reused"], False)
+        self.assertEqual(rec["note"], "")
+
+
 if __name__ == "__main__":
     unittest.main()