Skip already-answered messages in Phase 2 using messages table

alembic/versions/005_add_message_ts.py

@@ -0,0 +1,24 @@
+"""add message_ts column to messages table
+
+Revision ID: 005
+Revises: 004
+Create Date: 2026-03-31
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "005"
+down_revision: Union[str, None] = "004"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("messages", sa.Column("message_ts", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("messages", "message_ts")

claw/core/sessions.py

@@ -128,15 +128,26 @@ def get_sessions_to_evict(
     return list(rows)
 
 
-def save_message(db: DbSession, thread_ts: str, role: str, content: str) -> Message:
+def save_message(
+    db: DbSession, thread_ts: str, role: str, content: str, message_ts: Optional[str] = None
+) -> Message:
     """Persist a user or assistant message for a thread."""
-    msg = Message(thread_ts=thread_ts, role=role, content=content)
+    msg = Message(thread_ts=thread_ts, role=role, content=content, message_ts=message_ts)
     db.add(msg)
     db.commit()
     db.refresh(msg)
     return msg
 
 
+def is_message_handled(db: DbSession, thread_ts: str, message_ts: str) -> bool:
+    """Return True if we already saved a user message with this slack ts for this thread."""
+    stmt = select(Message).where(
+        Message.thread_ts == thread_ts,
+        Message.message_ts == message_ts,
+    ).limit(1)
+    return db.exec(stmt).first() is not None
+
+
 def get_recent_messages(db: DbSession, thread_ts: str, limit: int = 20) -> list[Message]:
     """Return the most recent *limit* messages for *thread_ts* in chronological order."""
     all_msgs = db.exec(

claw/core/slack_poller.py

@@ -12,7 +12,7 @@
 from claw.core.agent import create_client, iter_response, send_message
 from claw.models import Session as SessionModel
 from claw.core.mcp_mapper import build_mcp_servers
-from claw.core.sessions import close_session, get_active_sessions, get_recent_messages, get_session, get_sessions_to_evict, save_message, update_last_poll_ts, upsert_session
+from claw.core.sessions import close_session, get_active_sessions, get_recent_messages, get_session, get_sessions_to_evict, is_message_handled, save_message, update_last_poll_ts, upsert_session
 from claw.core.slack_client import fetch_history, fetch_thread_replies, open_self_dm, reply_to_thread, resolve_identity
 
 logger = logging.getLogger(__name__)
@@ -179,6 +179,11 @@ async def _poll_once(self) -> None:
                 if reply_ts in self._posted_ts:
                     continue
 
+                # Skip already-handled messages (echo prevention)
+                with DbSession(self._db_engine) as db:
+                    if is_message_handled(db, session.thread_ts, reply_ts):
+                        continue
+
                 # Track max ts
                 if float(reply_ts) > float(max_reply_ts):
                     max_reply_ts = reply_ts
@@ -274,8 +279,9 @@ async def _handle_message(self, message: dict) -> None:
 
             # Send prompt and spawn background task for response collection
             await send_message(client, prompt)
+            msg_ts = message.get("ts", "")
             self._response_tasks[thread_ts] = asyncio.create_task(
-                self._run_response_loop(thread_ts, channel, text)
+                self._run_response_loop(thread_ts, channel, text, slack_ts=msg_ts)
             )
 
         except Exception:
@@ -335,7 +341,7 @@ async def _evict_lra_client(self) -> None:
         except Exception:
             logger.exception("Failed to close session for evicted thread %s", lra_ts)
 
-    async def _run_response_loop(self, thread_ts: str, channel: str, original_text: str) -> None:
+    async def _run_response_loop(self, thread_ts: str, channel: str, original_text: str, slack_ts: str = "") -> None:
         """Background task: collect iter_response chunks, reply, update DB."""
         try:
             client = self._clients[thread_ts]
@@ -415,7 +421,7 @@ async def _run_response_loop(self, thread_ts: str, channel: str, original_text:
 
             # Persist user + assistant messages
             with DbSession(self._db_engine) as db:
-                save_message(db, thread_ts, "user", original_text)
+                save_message(db, thread_ts, "user", original_text, message_ts=slack_ts or None)
                 save_message(db, thread_ts, "assistant", full_response)
 
             # Update session with new sdk_session_id

claw/models.py

@@ -31,4 +31,5 @@ class Message(SQLModel, table=True):
     thread_ts: str = Field(index=True)
     role: str  # "user" | "assistant"
     content: str
+    message_ts: Optional[str] = Field(default=None)
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

openspec/changes/echo-prevention/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-03-31

openspec/changes/echo-prevention/proposal.md

@@ -0,0 +1,26 @@
+# Echo Prevention — Skip Already-Answered Messages in Phase 2
+
+## Why
+
+Phase 2 (active thread replies) can re-process messages that the bot already
+responded to in a previous poll cycle.  When the poller restarts or
+`last_poll_ts` drifts, a user message that was already saved and answered gets
+picked up again, producing a duplicate agent run.  Tracking the Slack `ts` of
+each handled user message in the `messages` table lets us skip duplicates
+cheaply.
+
+## What Changes
+
+- `alembic/versions/005_add_message_ts.py` — migration adding nullable
+  `message_ts` column to the `messages` table.
+- `claw/models.py` — `Message.message_ts` field (str | None).
+- `claw/core/sessions.py` — `save_message()` gains optional `message_ts`
+  parameter; new `is_message_handled(db, thread_ts, message_ts)` function.
+- `claw/core/slack_poller.py` — Phase 2 calls `is_message_handled` before
+  `_handle_message`; `_run_response_loop` passes `slack_ts` to `save_message`.
+
+## Impact
+
+- Backend: models, sessions CRUD, poller loop
+- Tests: new unit tests in test_sessions.py and test_slack_poller.py
+- DB: new Alembic migration (005)

openspec/changes/echo-prevention/specs/skip-handled-messages/spec.md

@@ -0,0 +1,42 @@
+# Skip Handled Messages
+
+## ADDED Requirements
+
+### Requirement: Track Slack message_ts in messages table
+
+Each saved user message can optionally record the Slack `ts` of the triggering
+message so we can detect duplicates.
+
+#### Scenario: save_message stores message_ts when provided
+- **WHEN** `save_message` is called with `message_ts="123.456"`
+- **THEN** the persisted `Message` row has `message_ts == "123.456"`
+
+#### Scenario: save_message works without message_ts
+- **WHEN** `save_message` is called without `message_ts`
+- **THEN** the persisted `Message` row has `message_ts == None`
+
+### Requirement: is_message_handled detects duplicates
+
+#### Scenario: returns True for already-saved message
+- **WHEN** a message with `thread_ts="T1"` and `message_ts="M1"` exists
+- **THEN** `is_message_handled(db, "T1", "M1")` returns `True`
+
+#### Scenario: returns False for unseen message
+- **WHEN** no message with `thread_ts="T1"` and `message_ts="M1"` exists
+- **THEN** `is_message_handled(db, "T1", "M1")` returns `False`
+
+### Requirement: Phase 2 skips already-handled replies
+
+#### Scenario: handled reply is skipped
+- **WHEN** Phase 2 encounters a reply whose `ts` is already in the messages table
+- **THEN** `_handle_message` is NOT called for that reply
+
+#### Scenario: unhandled reply is processed
+- **WHEN** Phase 2 encounters a reply whose `ts` is NOT in the messages table
+- **THEN** `_handle_message` IS called for that reply
+
+### Requirement: _run_response_loop records slack_ts
+
+#### Scenario: user message saved with slack_ts
+- **WHEN** `_run_response_loop` saves the user message
+- **THEN** it passes `message_ts=<original slack ts>` to `save_message`

openspec/changes/echo-prevention/tasks.md

@@ -0,0 +1,31 @@
+# Tasks — Echo Prevention
+
+## 1. Message model + sessions CRUD
+
+### RED
+- [x] 1.1 Write test: `test_save_message_stores_message_ts` — call save_message with message_ts, assert it is persisted
+- [x] 1.2 Write test: `test_save_message_defaults_message_ts_none` — call save_message without message_ts, assert None
+- [x] 1.3 Write test: `test_is_message_handled_true` — save a message with message_ts, assert is_message_handled returns True
+- [x] 1.4 Write test: `test_is_message_handled_false` — assert is_message_handled returns False for unknown ts
+
+### GREEN
+- [x] 1.5 Create migration `005_add_message_ts.py`
+- [x] 1.6 Add `message_ts` field to `Message` model
+- [x] 1.7 Update `save_message()` to accept and store `message_ts`
+- [x] 1.8 Implement `is_message_handled()`
+- [x] 1.9 Run `uv run pytest tests/test_sessions.py -q` — all pass
+
+## 2. Slack poller echo prevention
+
+### RED
+- [x] 2.1 Write test: `test_phase2_skips_already_handled_message` — mock is_message_handled to return True, assert _handle_message not called
+- [x] 2.2 Write test: `test_phase2_processes_unhandled_message` — mock is_message_handled to return False, assert _handle_message called
+
+### GREEN
+- [x] 2.3 Update Phase 2 in `_poll_once` to call `is_message_handled` before `_handle_message`
+- [x] 2.4 Update `_run_response_loop` signature to accept `slack_ts`, pass to `save_message`
+- [x] 2.5 Update `_handle_message` to pass `msg_ts` through to `_run_response_loop`
+- [x] 2.6 Run `uv run pytest tests/test_slack_poller.py -q` — all pass
+
+## 3. Full regression
+- [x] 3.1 Run `uv run pytest tests/ -q` — 0 failures

tests/test_sessions.py

@@ -366,6 +366,39 @@ def test_multiple_messages_same_thread(self, db):
         rows = db.exec(select(Message).where(Message.thread_ts == "thread.1")).all()
         assert len(rows) == 3
 
+    def test_save_message_stores_message_ts(self, db):
+        from claw.core.sessions import save_message, upsert_session
+
+        upsert_session(db, "thread.1", "C001")
+        msg = save_message(db, "thread.1", "user", "hello", message_ts="123.456")
+        assert msg.message_ts == "123.456"
+
+    def test_save_message_defaults_message_ts_none(self, db):
+        from claw.core.sessions import save_message, upsert_session
+
+        upsert_session(db, "thread.1", "C001")
+        msg = save_message(db, "thread.1", "user", "hello")
+        assert msg.message_ts is None
+
+
+# ---------------------------------------------------------------------------
+# is_message_handled
+# ---------------------------------------------------------------------------
+class TestIsMessageHandled:
+    """is_message_handled detects duplicate messages by thread_ts + message_ts."""
+
+    def test_returns_true_for_existing_message(self, db):
+        from claw.core.sessions import is_message_handled, save_message, upsert_session
+
+        upsert_session(db, "T1", "C001")
+        save_message(db, "T1", "user", "hi", message_ts="M1")
+        assert is_message_handled(db, "T1", "M1") is True
+
+    def test_returns_false_for_unknown_message(self, db):
+        from claw.core.sessions import is_message_handled
+
+        assert is_message_handled(db, "T1", "M1") is False
+
 
 # ---------------------------------------------------------------------------
 # get_recent_messages

tests/test_slack_poller.py

@@ -235,6 +235,7 @@ async def test_polls_active_thread_replies(self):
              patch("claw.core.slack_poller.get_active_sessions", return_value=[mock_session]), \
              patch("claw.core.slack_poller.fetch_thread_replies", new_callable=AsyncMock) as mock_fetch, \
              patch("claw.core.slack_poller.update_last_poll_ts"), \
+             patch("claw.core.slack_poller.is_message_handled", return_value=False), \
              patch("claw.core.slack_poller.DbSession"):
             mock_fetch.return_value = [
                 {"ts": "50.002", "user": "U_HUMAN", "text": "thread reply"},
@@ -261,6 +262,7 @@ async def test_updates_last_poll_ts_after_processing_replies(self):
              patch("claw.core.slack_poller.get_active_sessions", return_value=[mock_session]), \
              patch("claw.core.slack_poller.fetch_thread_replies", new_callable=AsyncMock) as mock_fetch, \
              patch("claw.core.slack_poller.update_last_poll_ts") as mock_update_ts, \
+             patch("claw.core.slack_poller.is_message_handled", return_value=False), \
              patch("claw.core.slack_poller.DbSession") as mock_db_cls:
             mock_db = MagicMock()
             mock_db_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
@@ -299,6 +301,73 @@ async def test_skips_own_messages_in_thread_replies(self):
             mock_handle.assert_not_called()
 
 
+# ---------------------------------------------------------------------------
+# Phase 2 — echo prevention (is_message_handled)
+# ---------------------------------------------------------------------------
+
+class TestPhase2EchoPrevention:
+    """Phase 2 skips messages already recorded in the messages table."""
+
+    @pytest.mark.asyncio
+    async def test_phase2_skips_already_handled_message(self):
+        """When is_message_handled returns True, _handle_message is NOT called."""
+        ws = _make_ws({"ok": True, "messages": []})
+        cfg = _make_config()
+        poller = SlackPoller(ws, cfg, _make_db_engine())
+        poller._last_seen_ts = "99.0"
+        poller._channel_id = "C123"
+        poller._user_id = "U_BOT"
+
+        mock_session = MagicMock()
+        mock_session.thread_ts = "50.001"
+        mock_session.last_poll_ts = "50.001"
+
+        with patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \
+             patch("claw.core.slack_poller.get_active_sessions", return_value=[mock_session]), \
+             patch("claw.core.slack_poller.fetch_thread_replies", new_callable=AsyncMock) as mock_fetch, \
+             patch("claw.core.slack_poller.update_last_poll_ts"), \
+             patch("claw.core.slack_poller.is_message_handled", return_value=True) as mock_handled, \
+             patch("claw.core.slack_poller.DbSession") as mock_db_cls:
+            mock_db = MagicMock()
+            mock_db_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
+            mock_db_cls.return_value.__exit__ = MagicMock(return_value=False)
+            mock_fetch.return_value = [
+                {"ts": "50.002", "user": "U_HUMAN", "text": "already handled reply"},
+            ]
+            await poller._poll_once()
+            mock_handled.assert_called_once_with(mock_db, "50.001", "50.002")
+            mock_handle.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_phase2_processes_unhandled_message(self):
+        """When is_message_handled returns False, _handle_message IS called."""
+        ws = _make_ws({"ok": True, "messages": []})
+        cfg = _make_config()
+        poller = SlackPoller(ws, cfg, _make_db_engine())
+        poller._last_seen_ts = "99.0"
+        poller._channel_id = "C123"
+        poller._user_id = "U_BOT"
+
+        mock_session = MagicMock()
+        mock_session.thread_ts = "50.001"
+        mock_session.last_poll_ts = "50.001"
+
+        with patch.object(poller, "_handle_message", new_callable=AsyncMock) as mock_handle, \
+             patch("claw.core.slack_poller.get_active_sessions", return_value=[mock_session]), \
+             patch("claw.core.slack_poller.fetch_thread_replies", new_callable=AsyncMock) as mock_fetch, \
+             patch("claw.core.slack_poller.update_last_poll_ts"), \
+             patch("claw.core.slack_poller.is_message_handled", return_value=False), \
+             patch("claw.core.slack_poller.DbSession") as mock_db_cls:
+            mock_db = MagicMock()
+            mock_db_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
+            mock_db_cls.return_value.__exit__ = MagicMock(return_value=False)
+            mock_fetch.return_value = [
+                {"ts": "50.002", "user": "U_HUMAN", "text": "new reply"},
+            ]
+            await poller._poll_once()
+            mock_handle.assert_called_once()
+
+
 # ---------------------------------------------------------------------------
 # Client registry — per-thread ClaudeSDKClient reuse
 # ---------------------------------------------------------------------------