fix(logstream): checkpoint the starting position once, whatever produced it

Fifth Codex round, one new P2 on dd9b123, and the sixth instance of one bug:
a starting cursor that never reaches disk, so the next launch calls itself a
first run and jumps to the tip.

This time --since-event-id with no state file skipped the immediate
checkpoint entirely, because that write lived inside the "resolve the tip"
branch and an explicit cursor never enters it. Nothing was persisted until
watch_events yielded, which by default is up to five minutes later; an
interrupt inside that window leaves no file at all.

Rather than add a third call site I hoisted it: one place resolves the
starting position, one place records it. The position can arrive three ways
— an explicit --since-event-id, the tip, or None from an empty log or
--from-start — and all three now take the same required write whenever no
state file exists yet. That removes the shape the last four rounds kept
finding new instances of, instead of removing one more instance.

Two test-quality notes, both the same lesson:

The first version of this test passed against the unfixed code. The helper
uses a 60ms poll timeout, so watch_events yielded immediately and the loop's
own checkpoint wrote the file — the startup window the bug lives in never
opened. It now stubs watch_events to interrupt before yielding, which is what
actually pins the startup write, and covers all three entry paths.

The hoist also broke test_match_does_not_checkpoint_if_output_fails, which
asserted that no state file existed after a failed match write. That was
pinning the symptom: a startup checkpoint now legitimately exists. The
invariant is that its cursor must still be the pre-match position so the
undelivered event replays, so it asserts the cursor value instead.

4407 passed, 31 skipped.

Author: igorls

mempalace/cli.py

@@ -1612,21 +1612,26 @@ def _logstream_watch(ls, args, as_json):
         # what arrives from now on. Never silent: say what was skipped.
         cursor = ls.latest_event_id()
         skipped_from = cursor
-        # Record the starting position immediately, even when the log is
-        # empty and that position is None. Without the file, the next launch
-        # is indistinguishable from a first run and would jump past anything
-        # that arrived while this watcher was stopped — so unlike later
-        # checkpoints this one must not fail quietly.
-        if args.state_file:
-            try:
-                write_watch_cursor(args.state_file, cursor, agent=args.agent, required=True)
-            except OSError as exc:
-                _logstream_fail(
-                    f"could not write the initial checkpoint to {args.state_file}: {exc}. "
-                    "Refusing to start: without it a restart would skip every event "
-                    "that arrives before then.",
-                    as_json,
-                )
+
+    # One place decides the starting position; one place records it. That
+    # position can arrive three ways — an explicit --since-event-id, the tip
+    # above, or None (an empty log, or --from-start) — and every one of them
+    # needs the same immediate checkpoint when no state file exists yet.
+    # Deferring to the first watch_events yield leaves a window of up to a
+    # full poll timeout in which an interrupt leaves no file behind, and the
+    # next launch calls itself a first run and jumps to the tip, skipping
+    # whatever arrived in between. Unlike later checkpoints this one cannot
+    # fail quietly: losing it costs a skipped event, not a replay.
+    if args.state_file and state_condition == WATCH_STATE_ABSENT:
+        try:
+            write_watch_cursor(args.state_file, cursor, agent=args.agent, required=True)
+        except OSError as exc:
+            _logstream_fail(
+                f"could not write the initial checkpoint to {args.state_file}: {exc}. "
+                "Refusing to start: without it a restart would skip every event "
+                "that arrives before then.",
+                as_json,
+            )
     if not as_json:
         where = args.agent or ", ".join(sorted(spec["to_agents"] or [])) or "everything"
         print(f"Watching {where} from {cursor or 'now'}; Ctrl-C to stop.", file=sys.stderr)

tests/test_cli_logstream.py

@@ -520,18 +520,29 @@ def test_idle_deadline_caps_the_poll(self, palace_path, capsys):
     def test_match_does_not_checkpoint_if_output_fails(
         self, palace_path, tmp_path, capsys, monkeypatch
     ):
-        """A broken pipe after a match must not persist the cursor."""
-        state = str(tmp_path / "watch.json")
+        """A broken pipe after a match must not advance the cursor past it.
+
+        Asserting "no state file" would be pinning the symptom: since the
+        starting position is now checkpointed before the first poll, a file
+        legitimately exists. What must hold is that its cursor is still the
+        pre-match position, so the undelivered event replays on restart —
+        a duplicate, never a skip.
+        """
+        state = tmp_path / "watch.json"
         cmd_logstream(_append_args(palace_path, to_agent="mac-claude", from_agent="windows-grok"))
-        capsys.readouterr()
+        matched_id = json.loads(capsys.readouterr().out)["id"]
 
         def boom(*_a, **_k):
             raise OSError("broken pipe")
 
         monkeypatch.setattr("json.dumps", boom)
         with pytest.raises(OSError, match="broken pipe"):
-            cmd_logstream(_watch_args(palace_path, agent="mac-claude", state_file=state))
-        assert not (tmp_path / "watch.json").exists()
+            cmd_logstream(_watch_args(palace_path, agent="mac-claude", state_file=str(state)))
+
+        monkeypatch.undo()
+        stored = json.loads(state.read_text(encoding="utf-8"))["cursor"]
+        assert stored != matched_id, "cursor advanced past an event that was never delivered"
+        assert stored is None, "cursor should still be the pre-match starting position"
 
     def test_fresh_watch_starts_at_the_tip_not_the_beginning(self, palace_path, capsys):
         """A first watch must not replay the whole log.
@@ -705,3 +716,48 @@ def denied(*_a, **kw):
             )
         assert exc.value.code == 1
         assert "initial checkpoint" in json.loads(capsys.readouterr().out)["error"]
+
+    def test_starting_cursor_is_checkpointed_before_the_first_poll(
+        self, palace_path, tmp_path, monkeypatch, capsys
+    ):
+        """Every entry path must checkpoint before polling, not after.
+
+        The starting position arrives three ways — an explicit
+        --since-event-id, the tip, or None (empty log / --from-start) — and
+        all three need the file on disk *before* the first poll. Deferring to
+        the first watch_events yield leaves a window of up to a full poll
+        timeout; an interrupt inside it leaves no file, so the next launch
+        calls itself a first run and jumps to the tip.
+
+        watch_events is stubbed to interrupt immediately, which is what makes
+        this pin the startup write. Letting the real loop run would pass on
+        the loop's own checkpoint instead, since the test poll timeout is
+        milliseconds rather than the five-minute default.
+        """
+        import mempalace.logstream as logstream_module
+
+        cmd_logstream(_append_args(palace_path, to_agent="mac-claude", from_agent="windows-grok"))
+        first_id = json.loads(capsys.readouterr().out)["id"]
+
+        def interrupt_before_yielding(*_a, **_k):
+            raise KeyboardInterrupt
+
+        monkeypatch.setattr(logstream_module.Logstream, "watch_events", interrupt_before_yielding)
+
+        for label, overrides, expected in (
+            ("explicit cursor", {"since_event_id": first_id}, first_id),
+            ("tip", {}, first_id),
+            ("from-start", {"from_start": True}, None),
+        ):
+            state = tmp_path / f"{label.replace(' ', '_')}.json"
+            kwargs = {
+                "agent": "mac-claude",
+                "state_file": str(state),
+                "from_start": False,
+                **overrides,
+            }
+            with pytest.raises(SystemExit) as exc:
+                cmd_logstream(_watch_args(palace_path, **kwargs))
+            assert exc.value.code == 130
+            assert state.exists(), f"{label}: interrupted before any checkpoint landed"
+            assert json.loads(state.read_text(encoding="utf-8"))["cursor"] == expected, label