fix(claude-channel): use opening cursor for first-run boundary

src/pepper_cli/claude_channel.py

@@ -424,6 +424,11 @@ def __init__(
         # so a server that ignored ``start`` and replayed its window is still told apart from a
         # resume and gets the first-run summary, never a push of every old message.
         self.head_pending = at_head
+        # The opening head of this connection, once a ``start=head`` cursor names it. Event ids
+        # above it are live, not replayed history, so the local cutoff must not judge them. It is
+        # reset for every connection: a reconnect resumes from the saved cursor and may replay old
+        # messages.
+        self.live_from: int | None = None
         # ``resumed``: the cursor file held a position. A saved head of 0 is a resume, not a
         # first run, or the messages posted while the channel was down would be summarised away.
         self.first_run = not resumed and stored is None and cursor == 0
@@ -653,11 +658,14 @@ def _is_first_run_history(self, event: dict[str, Any]) -> bool:
         created = _parse_time(event.get("created_at"))
         return self.first_run and created is not None and created < self.cutoff
 
-    def _handle_event(self, kind: str, event: dict[str, Any]) -> None:
+    def _handle_event(self, kind: str, event: dict[str, Any], event_id: int | None) -> None:
+        after_head = (
+            self.live_from is not None and event_id is not None and event_id > self.live_from
+        )
         if _is_known_silent(kind):
             return
         if kind not in MESSAGE_KINDS:
-            if self._filtered_out(kind) or self._is_first_run_history(event):
+            if self._filtered_out(kind) or (not after_head and self._is_first_run_history(event)):
                 return
             self._resync(f"an event of kind {sanitize_line(kind)[:_CAUSE_CLIP]}")
             return
@@ -678,7 +686,7 @@ def _handle_event(self, kind: str, event: dict[str, Any]) -> None:
             self._remember(message_id)
             return
         created = _parse_time(event.get("created_at"))
-        if created is not None and created < self.cutoff:
+        if not after_head and created is not None and created < self.cutoff:
             self.skipped[topic_id] = self.skipped.get(topic_id, 0) + 1
             self.skipped_times.append(created)
             self._remember(message_id)
@@ -845,8 +853,9 @@ def _handle_frame(self, frame: api_client.FeedFrame) -> str | None:
         if frame.kind != "cursor":
             if scoped and self.scope is not None and not _event_in_scope(frame.payload, self.scope):
                 raise self._ended("the server sent an event from outside the requested scope", 1)
-            self._handle_event(frame.kind, frame.payload)
+            self._handle_event(frame.kind, frame.payload, frame.id)
         elif self.head_pending:
+            self.live_from = new_cursor
             self._live_edge()
         self.head_pending = False
         self._advance(new_cursor)
@@ -892,6 +901,7 @@ def _follow(self) -> int:
         while not self.stop.is_set():
             problem: str | None = None
             self.scope = None
+            self.live_from = None
             try:
                 with self._open(reconnecting) as frames:
                     for frame in frames:

tests/test_claude_channel.py

@@ -1079,6 +1079,56 @@ def test_a_head_start_sends_start_head_and_tells_the_session_once(monkeypatch, t
     assert "Starting at the live edge" in " ".join(capsys.readouterr().err.split())
 
 
+def test_a_message_after_the_opening_head_ignores_the_local_clock(monkeypatch):
+    behind = _now() - timedelta(hours=1)
+    fake = _Fake(_CONNECTED + _scope() + _cursor(30) + _event(31, created=behind) + _PING)
+
+    result = run_channel(monkeypatch, fake, ["--room", _ROOM])
+
+    live, message = result.notifications
+    assert live["meta"]["live_edge"] == "true"
+    assert message["meta"]["kind"] == "message"
+
+
+def test_a_message_after_a_zero_head_ignores_the_local_clock(monkeypatch):
+    behind = _now() - timedelta(hours=1)
+    fake = _Fake(_CONNECTED + _scope() + _cursor(0) + _event(1, created=behind) + _PING)
+
+    result = run_channel(monkeypatch, fake, ["--room", _ROOM])
+
+    live, message = result.notifications
+    assert live["meta"]["live_edge"] == "true"
+    assert message["meta"]["kind"] == "message"
+
+
+def test_genuine_replay_history_is_still_suppressed(monkeypatch):
+    old = _now() - timedelta(hours=1)
+    fake = _Fake(
+        _CONNECTED + _scope() + _event(4, created=old) + _event(5, message=_MESSAGE_2) + _PING
+    )
+    fake.messages[_MESSAGE_2] = _message_json(_MESSAGE_2)
+
+    result = run_channel(monkeypatch, fake, ["--room", _ROOM, "--start", "beginning"])
+
+    assert result.kinds() == ["backlog", "message"]
+
+
+def test_a_reconnect_after_the_head_does_not_treat_replay_as_live(monkeypatch):
+    behind = _now() - timedelta(hours=1)
+    fake = _Fake(
+        _CONNECTED + _scope() + _cursor(30),
+        _CONNECTED + _scope() + _event(31, created=behind) + _PING,
+    )
+
+    result = run_channel(monkeypatch, fake, ["--room", _ROOM])
+
+    live, skipped = result.notifications
+    assert live["meta"]["live_edge"] == "true"
+    assert skipped["meta"]["kind"] == "backlog"
+    assert skipped["meta"]["count"] == "1"
+    assert fake.stream_requests()[1].headers["last-event-id"] == "30"
+
+
 def test_a_head_start_is_saved_even_when_no_event_follows(monkeypatch, tmp_path):
     cursor_file = tmp_path / "cursor"
     fake = _Fake(_CONNECTED + _scope() + _cursor(30))