diff --git a/src/pepper_cli/claude_channel.py b/src/pepper_cli/claude_channel.py index 2e117f2..9ae0a81 100644 --- a/src/pepper_cli/claude_channel.py +++ b/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: diff --git a/tests/test_claude_channel.py b/tests/test_claude_channel.py index ad7032f..89964fc 100644 --- a/tests/test_claude_channel.py +++ b/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))