From 0846434d06bb1f73e27deb50f52f3ce00e0bcd75 Mon Sep 17 00:00:00 2001 From: Tanner Date: Tue, 31 Mar 2026 10:15:12 -0400 Subject: [PATCH] Fast-forward last_poll_ts on startup to prevent re-processing old messages After resolving Slack identity in _resolve_startup_identity, fast-forward last_poll_ts for all active sessions to the current timestamp so the first poll cycle only sees messages posted after startup. Closes #74 Co-Authored-By: Claude Sonnet 4.6 --- claw/core/slack_poller.py | 13 +++++ .../changes/startup-fast-forward/proposal.md | 37 ++++++++++++++ .../changes/startup-fast-forward/tasks.md | 6 +++ tests/test_slack_poller.py | 49 +++++++++++++++++++ 4 files changed, 105 insertions(+) create mode 100644 openspec/changes/startup-fast-forward/proposal.md create mode 100644 openspec/changes/startup-fast-forward/tasks.md diff --git a/claw/core/slack_poller.py b/claw/core/slack_poller.py index adff160..98c40d5 100644 --- a/claw/core/slack_poller.py +++ b/claw/core/slack_poller.py @@ -74,6 +74,19 @@ async def _resolve_startup_identity(self) -> None: self._channel_id = await open_self_dm(self._ws, self._config, self._user_id) logger.info("Resolved self-DM channel_id=%s", self._channel_id) + # Fast-forward last_poll_ts for all active sessions to now, + # so we don't re-process messages from before this startup + startup_ts = f"{time.time():.6f}" + with DbSession(self._db_engine) as db: + active = get_active_sessions( + db, + days=self._config.ACTIVE_THREAD_WINDOW_DAYS, + limit=self._config.MAX_ACTIVE_THREADS, + ) + for session in active: + update_last_poll_ts(db, session.thread_ts, startup_ts) + logger.info("Fast-forwarded last_poll_ts for %d active sessions to %s", len(active), startup_ts) + async def _poll_loop(self) -> None: """Run _poll_once every POLL_INTERVAL_SECONDS until stopped.""" try: diff --git a/openspec/changes/startup-fast-forward/proposal.md b/openspec/changes/startup-fast-forward/proposal.md new file mode 100644 index 0000000..8cd2da6 --- /dev/null +++ b/openspec/changes/startup-fast-forward/proposal.md @@ -0,0 +1,37 @@ +# Fast-forward last_poll_ts on startup to prevent re-processing old messages + +## Problem +When the SlackPoller starts (or restarts), active sessions retain their old +`last_poll_ts`. The first poll cycle fetches every message posted since the +last poll — including messages that were already processed before the restart. +This causes duplicate agent responses. + +## Solution +In `SlackPoller._resolve_startup_identity()`, after resolving identity and +the self-DM channel, fast-forward `last_poll_ts` for all active sessions to +the current timestamp. This ensures the first poll cycle only sees messages +posted *after* startup. + +### Changes +1. **`claw/core/slack_poller.py`** — `_resolve_startup_identity()`: + - After resolving `_user_id` and `_channel_id`, open a `DbSession`. + - Call `get_active_sessions(db, days=config.ACTIVE_THREAD_WINDOW_DAYS, + limit=config.MAX_ACTIVE_THREADS)`. + - For each session, call `update_last_poll_ts(db, session.thread_ts, + startup_ts)` where `startup_ts = f"{time.time():.6f}"`. + - Log the count of fast-forwarded sessions. + +No new DB functions or config keys needed — reuses existing +`get_active_sessions` and `update_last_poll_ts`. + +## Dependencies +- `get_active_sessions` and `update_last_poll_ts` in `claw/core/sessions.py` + — already merged. + +## Acceptance Criteria +- `_resolve_startup_identity` fast-forwards `last_poll_ts` for all active + sessions after resolving identity. +- Uses existing `get_active_sessions` + `update_last_poll_ts`. +- Unit test: after calling `_resolve_startup_identity`, `update_last_poll_ts` + is called for each active session with a ts >= the startup time. +- All existing tests pass. diff --git a/openspec/changes/startup-fast-forward/tasks.md b/openspec/changes/startup-fast-forward/tasks.md new file mode 100644 index 0000000..6c1582d --- /dev/null +++ b/openspec/changes/startup-fast-forward/tasks.md @@ -0,0 +1,6 @@ +# Startup fast-forward tasks + +- [x] Write openspec proposal +- [x] RED: Write failing test `test_startup_fast_forwards_last_poll_ts` +- [x] GREEN: Implement fast-forward logic in `_resolve_startup_identity` +- [x] Verify all tests pass (284 existing + 1 new = 285) diff --git a/tests/test_slack_poller.py b/tests/test_slack_poller.py index c74519d..f47bb86 100644 --- a/tests/test_slack_poller.py +++ b/tests/test_slack_poller.py @@ -487,6 +487,55 @@ async def test_sets_user_id_and_channel_id(self, mock_resolve, mock_open_dm): mock_resolve.assert_called_once_with(ws, cfg) mock_open_dm.assert_called_once_with(ws, cfg, "U123") + @pytest.mark.asyncio + @patch("claw.core.slack_poller.open_self_dm", new_callable=AsyncMock, return_value="D789") + @patch( + "claw.core.slack_poller.resolve_identity", + new_callable=AsyncMock, + return_value={"user_id": "U123", "bot_id": "B456"}, + ) + @patch("claw.core.slack_poller.update_last_poll_ts") + @patch("claw.core.slack_poller.get_active_sessions") + @patch("claw.core.slack_poller.DbSession") + async def test_startup_fast_forwards_last_poll_ts( + self, + mock_db_session_cls, + mock_get_active, + mock_update_poll, + mock_resolve, + mock_open_dm, + ): + """After resolving identity, last_poll_ts is fast-forwarded for all active sessions.""" + # Two fake active sessions + s1 = MagicMock() + s1.thread_ts = "100.001" + s2 = MagicMock() + s2.thread_ts = "200.002" + mock_get_active.return_value = [s1, s2] + + ws = _make_ws({"ok": True, "messages": []}) + cfg = _make_config() + poller = SlackPoller(ws, cfg, _make_db_engine()) + + before = time.time() + await poller._resolve_startup_identity() + after = time.time() + + # update_last_poll_ts called once per active session + assert mock_update_poll.call_count == 2 + + # Each call uses the correct thread_ts and a ts within [before, after] + for call_args, session in zip(mock_update_poll.call_args_list, [s1, s2]): + _db, thread_ts, startup_ts = call_args[0] + assert thread_ts == session.thread_ts + assert before <= float(startup_ts) <= after + + # get_active_sessions called with config values + mock_get_active.assert_called_once() + call_kwargs = mock_get_active.call_args[1] + assert call_kwargs["days"] == cfg.ACTIVE_THREAD_WINDOW_DAYS + assert call_kwargs["limit"] == cfg.MAX_ACTIVE_THREADS + # --------------------------------------------------------------------------- # _handle_message — full agent flow