From 3695296036023093dac210df8756dde40cc2a998 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 14:50:30 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(collab):=20add=20E1=20DM=20schema=20mi?= =?UTF-8?q?gration=20=E2=80=94=20remote=5Fmsg=5Fid,=20delivered=5Fat,=20pe?= =?UTF-8?q?er=5Foutbox?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add remote_msg_id + delivered_at columns to chat_messages via guarded _post_init migration (boot-brick safe) - Add UNIQUE index on (channel_id, remote_msg_id) for idempotent remote message delivery - Add send_remote_message() method with IntegrityError dedup - Add PeerOutboxStore with store-and-forward, exponential backoff, purge-on-revoke - Wire up PeerOutboxStore in app.py lifespan + state - Add 18 tests: upgrade migration, dedup, outbox operations Refs: #2020, epic #2012 --- tests/test_chat_remote_dm.py | 356 ++++++++++++++++++++++++++++++ tinyagentos/app.py | 5 + tinyagentos/chat/message_store.py | 80 +++++++ tinyagentos/chat/peer_outbox.py | 130 +++++++++++ 4 files changed, 571 insertions(+) create mode 100644 tests/test_chat_remote_dm.py create mode 100644 tinyagentos/chat/peer_outbox.py diff --git a/tests/test_chat_remote_dm.py b/tests/test_chat_remote_dm.py new file mode 100644 index 000000000..11a9ff967 --- /dev/null +++ b/tests/test_chat_remote_dm.py @@ -0,0 +1,356 @@ +"""Tests for remote DM message handling and peer outbox — E1 collab. + +Covers: +- ChatMessageStore migration: remote_msg_id / delivered_at columns +- send_remote_message dedup via unique constraint +- PeerOutboxStore enqueue / dequeue / retry / purge +- Upgrade from a pre-change chat.db (boot-brick regression gate) +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from pathlib import Path + +import pytest +import pytest_asyncio + +from tinyagentos.chat.message_store import ChatMessageStore +from tinyagentos.chat.peer_outbox import PeerOutboxStore + + +# --------------------------------------------------------------------------- +# ChatMessageStore — upgrade from pre-change DB +# --------------------------------------------------------------------------- + +MESSAGES_V0_SCHEMA = """\ +CREATE TABLE IF NOT EXISTS chat_messages ( + id TEXT PRIMARY KEY, + channel_id TEXT NOT NULL, + thread_id TEXT, + author_id TEXT NOT NULL, + author_type TEXT NOT NULL DEFAULT 'user', + content TEXT NOT NULL DEFAULT '', + content_type TEXT NOT NULL DEFAULT 'text', + content_blocks TEXT NOT NULL DEFAULT '[]', + embeds TEXT NOT NULL DEFAULT '[]', + components TEXT NOT NULL DEFAULT '[]', + attachments TEXT NOT NULL DEFAULT '[]', + reactions TEXT NOT NULL DEFAULT '{}', + state TEXT NOT NULL DEFAULT 'complete', + edited_at REAL, + deleted_at REAL, + expires_at REAL, + pinned INTEGER NOT NULL DEFAULT 0, + ephemeral INTEGER NOT NULL DEFAULT 0, + metadata TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_chat_messages_channel ON chat_messages(channel_id, created_at); +CREATE INDEX IF NOT EXISTS idx_chat_messages_thread ON chat_messages(thread_id); +""" + + +def _seed_db(db_path: Path, schema_sql: str) -> None: + conn = sqlite3.connect(str(db_path)) + conn.executescript(schema_sql) + conn.commit() + conn.close() + + +def _column_names(db_path: Path, table: str) -> set[str]: + conn = sqlite3.connect(str(db_path)) + rows = conn.execute(f"PRAGMA table_info({table})").fetchall() + conn.close() + return {r[1] for r in rows} + + +@pytest.mark.asyncio +class TestChatMessageStoreUpgradeE1: + """Upgrade from v0 (pre-collab) DB must add remote_msg_id + delivered_at.""" + + async def test_upgrade_adds_remote_dm_columns(self, tmp_path: Path): + db_path = tmp_path / "chat.db" + _seed_db(db_path, MESSAGES_V0_SCHEMA) + + store = ChatMessageStore(db_path) + await store.init() + try: + cols = _column_names(db_path, "chat_messages") + assert "remote_msg_id" in cols, "remote_msg_id missing after upgrade" + assert "delivered_at" in cols, "delivered_at missing after upgrade" + finally: + await store.close() + + async def test_upgrade_preserves_existing_data(self, tmp_path: Path): + db_path = tmp_path / "chat.db" + _seed_db(db_path, MESSAGES_V0_SCHEMA) + # Insert a row before the upgrade + conn = sqlite3.connect(str(db_path)) + now = time.time() + conn.execute( + """INSERT INTO chat_messages + (id, channel_id, author_id, author_type, content, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + ("msg-pre", "ch1", "user1", "user", "hello before", now), + ) + conn.commit() + conn.close() + + store = ChatMessageStore(db_path) + await store.init() + try: + cursor = await store._db.execute( + "SELECT id, content, remote_msg_id, delivered_at FROM chat_messages" + ) + rows = await cursor.fetchall() + assert len(rows) == 1 + assert rows[0][1] == "hello before" + assert rows[0][2] is None # remote_msg_id default + assert rows[0][3] is None # delivered_at default + finally: + await store.close() + + async def test_send_works_after_upgrade(self, tmp_path: Path): + db_path = tmp_path / "chat.db" + _seed_db(db_path, MESSAGES_V0_SCHEMA) + store = ChatMessageStore(db_path) + await store.init() + try: + msg = await store.send_message( + channel_id="ch1", + author_id="user1", + author_type="user", + content="post-upgrade", + ) + assert msg["id"] + assert msg["content"] == "post-upgrade" + # New columns default to NULL for local messages + assert "remote_msg_id" in msg + assert "delivered_at" in msg + assert msg["remote_msg_id"] is None + assert msg["delivered_at"] is None + finally: + await store.close() + + +# --------------------------------------------------------------------------- +# send_remote_message — dedup via unique constraint +# --------------------------------------------------------------------------- + +@pytest_asyncio.fixture +async def msg_store(tmp_path): + s = ChatMessageStore(tmp_path / "chat.db") + await s.init() + yield s + await s.close() + + +@pytest.mark.asyncio +class TestSendRemoteMessage: + async def test_first_insert_succeeds(self, msg_store): + msg = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-001", + content="hello from hogne", + ) + assert msg is not None + assert msg["channel_id"] == "dm-remote-1" + assert msg["author_id"] == "hub:hogne" + assert msg["author_type"] == "user" + assert msg["content"] == "hello from hogne" + assert msg["remote_msg_id"] == "rm-001" + assert msg["delivered_at"] is not None + + async def test_duplicate_returns_none(self, msg_store): + msg1 = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-001", + content="first", + ) + assert msg1 is not None + + msg2 = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-001", + content="second — different content, same remote_msg_id", + ) + assert msg2 is None # dedup + + async def test_same_remote_id_different_channel_both_succeed(self, msg_store): + """remote_msg_id dedup is scoped to channel_id.""" + m1 = await msg_store.send_remote_message( + channel_id="ch-a", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-shared", + content="in channel A", + ) + m2 = await msg_store.send_remote_message( + channel_id="ch-b", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-shared", + content="in channel B — same remote_msg_id, different channel", + ) + assert m1 is not None + assert m2 is not None + assert m1["channel_id"] == "ch-a" + assert m2["channel_id"] == "ch-b" + + async def test_remote_author_id_namespaced(self, msg_store): + msg = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:jaylfc", + contact_id="hub:jaylfc", + remote_msg_id="rm-002", + content="hello from jay", + ) + assert msg["author_id"] == "hub:jaylfc" + assert msg["author_type"] == "user" + + async def test_send_remote_message_with_metadata(self, msg_store): + msg = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="rm-meta", + content="with metadata", + metadata={"hops_since_user": 0, "delivered_via": "peer"}, + ) + assert msg["metadata"]["delivered_via"] == "peer" + + +# --------------------------------------------------------------------------- +# PeerOutboxStore +# --------------------------------------------------------------------------- + +@pytest_asyncio.fixture +async def outbox(tmp_path): + s = PeerOutboxStore(tmp_path / "outbox.db") + await s.init() + yield s + await s.close() + + +@pytest.mark.asyncio +class TestPeerOutboxStore: + async def test_enqueue_and_dequeue(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat", "body": {"content": "hello"}}, + ) + assert rid + + due = await outbox.dequeue_due("hub:hogne", limit=10) + assert len(due) == 1 + assert due[0]["id"] == rid + assert due[0]["contact_id"] == "hub:hogne" + env = json.loads(due[0]["envelope"]) + assert env["kind"] == "chat" + + async def test_mark_sent_removes(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + ) + ok = await outbox.mark_sent(rid) + assert ok is True + due = await outbox.dequeue_due("hub:hogne") + assert due == [] + + async def test_mark_sent_nonexistent(self, outbox): + ok = await outbox.mark_sent("does-not-exist") + assert ok is False + + async def test_mark_failed_backoff(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + ) + await outbox.mark_failed(rid) + + # After first failure, next_retry_at should be ~60s in the future + due_now = await outbox.dequeue_due("hub:hogne") + assert due_now == [] # not due yet + + # Check the row exists but with incremented attempts + count = await outbox.count_for_contact("hub:hogne") + assert count == 1 + + async def test_mark_failed_multiple_retries(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + ) + for _ in range(3): + await outbox.mark_failed(rid) + # Row should still exist with attempts=3 + count = await outbox.count_for_contact("hub:hogne") + assert count == 1 + + async def test_purge_for_contact(self, outbox): + await outbox.enqueue(contact_id="hub:a", envelope={"kind": "chat"}) + await outbox.enqueue(contact_id="hub:a", envelope={"kind": "ack"}) + await outbox.enqueue(contact_id="hub:b", envelope={"kind": "chat"}) + + deleted = await outbox.purge_for_contact("hub:a") + assert deleted == 2 + assert await outbox.count_for_contact("hub:a") == 0 + assert await outbox.count_for_contact("hub:b") == 1 + + async def test_count_for_contact(self, outbox): + for _ in range(3): + await outbox.enqueue(contact_id="hub:hogne", envelope={"kind": "chat"}) + assert await outbox.count_for_contact("hub:hogne") == 3 + assert await outbox.count_for_contact("hub:nonexistent") == 0 + + async def test_dequeue_due_respects_limit(self, outbox): + for i in range(5): + await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat", "seq": i}, + ) + due = await outbox.dequeue_due("hub:hogne", limit=2) + assert len(due) == 2 + + async def test_dequeue_due_future_retry_not_returned(self, outbox): + # Enqueue with a retry time far in the future + await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + next_retry_at=time.time() + 3600, + ) + due = await outbox.dequeue_due("hub:hogne") + assert due == [] + + +# --------------------------------------------------------------------------- +# Outbox upgrade — no columns added, but verify init is idempotent +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +class TestPeerOutboxUpgrade: + async def test_init_idempotent_on_existing_db(self, tmp_path: Path): + db_path = tmp_path / "outbox.db" + s1 = PeerOutboxStore(db_path) + await s1.init() + await s1.enqueue(contact_id="hub:hogne", envelope={"kind": "chat"}) + await s1.close() + + # Re-open — must work without error + s2 = PeerOutboxStore(db_path) + await s2.init() + try: + count = await s2.count_for_contact("hub:hogne") + assert count == 1 + finally: + await s2.close() diff --git a/tinyagentos/app.py b/tinyagentos/app.py index 32cc90d59..ba5a44390 100644 --- a/tinyagentos/app.py +++ b/tinyagentos/app.py @@ -92,6 +92,7 @@ async def get_response(self, path, scope): from tinyagentos.channel_hub.adapter_manager import AdapterManager from tinyagentos.chat.message_store import ChatMessageStore from tinyagentos.chat.channel_store import ChatChannelStore +from tinyagentos.chat.peer_outbox import PeerOutboxStore from tinyagentos.chat.hub import ChatHub from tinyagentos.chat.canvas import CanvasStore from tinyagentos.desktop_settings import DesktopSettingsStore @@ -389,6 +390,7 @@ async def _probe_backend(backend: dict) -> dict: adapter_manager = AdapterManager(channel_hub_router) chat_messages = ChatMessageStore(data_dir / "chat.db") chat_channels = ChatChannelStore(data_dir / "chat.db") + peer_outbox_store = PeerOutboxStore(data_dir / "peer_outbox.db") from tinyagentos.projects.project_store import ProjectStore from tinyagentos.projects.task_store import ProjectTaskStore from tinyagentos.projects.element_store import ProjectElementStore @@ -565,6 +567,7 @@ async def lifespan(app: FastAPI): await expert_agents.init() await chat_messages.init() await chat_channels.init() + await peer_outbox_store.init() await project_store.init() await project_invite_store.init() await board_audit_store.init() @@ -1441,6 +1444,7 @@ async def _web_push_sender(row: dict) -> None: await project_store.close() await chat_channels.close() await chat_messages.close() + await peer_outbox_store.close() await expert_agents.close() await streaming_sessions.close() await shared_folders.close() @@ -1626,6 +1630,7 @@ async def dispatch(self, request, call_next): app.state.song_store = song_store app.state.design_docs = design_docs app.state.contacts_store = contacts_store + app.state.peer_outbox = peer_outbox_store app.state.coding_workspaces = coding_workspaces_store app.state.install_registry = install_registry_store app.state.store_submissions = store_submissions diff --git a/tinyagentos/chat/message_store.py b/tinyagentos/chat/message_store.py index d0464ae31..83357d102 100644 --- a/tinyagentos/chat/message_store.py +++ b/tinyagentos/chat/message_store.py @@ -92,6 +92,28 @@ async def init(self) -> None: await self._db.commit() except Exception: pass + # E1 collab: remote message dedup columns + try: + await self._db.execute("ALTER TABLE chat_messages ADD COLUMN remote_msg_id TEXT") + await self._db.commit() + except Exception: + pass + try: + await self._db.execute("ALTER TABLE chat_messages ADD COLUMN delivered_at REAL") + await self._db.commit() + except Exception: + pass + # Dedup index MUST be created here (not in SCHEMA) because it references + # columns added by _post_init — per BaseStore boot-brick rule. + try: + await self._db.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_messages_remote_dedup " + "ON chat_messages(channel_id, remote_msg_id) " + "WHERE remote_msg_id IS NOT NULL" + ) + await self._db.commit() + except Exception: + pass async def send_message( self, @@ -134,6 +156,64 @@ async def send_message( await self._db.commit() return await self.get_message(msg_id) + async def send_remote_message( + self, + channel_id: str, + author_id: str, + contact_id: str, + remote_msg_id: str, + content: str, + *, + content_type: str = "text", + thread_id: str | None = None, + content_blocks: list | None = None, + metadata: dict | None = None, + created_at: float | None = None, + ) -> dict | None: + """Insert a DM received from a remote contact, deduplicating on + ``(channel_id, remote_msg_id)``. + + Returns the message dict on success, or ``None`` when the message + was already delivered (duplicate). ``author_id`` is namespaced as + ``hub:{username}`` per the collab spec. + + The caller MUST commit the message before sending the delivery ack. + """ + import sqlite3 + + msg_id = uuid.uuid4().hex[:12] + now = created_at if created_at is not None else time.time() + try: + await self._db.execute( + """INSERT INTO chat_messages + (id, channel_id, thread_id, author_id, author_type, content, + content_type, content_blocks, embeds, components, attachments, + reactions, state, pinned, ephemeral, metadata, created_at, + remote_msg_id, delivered_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?, ?, ?, ?)""", + ( + msg_id, channel_id, thread_id, author_id, "user", content, + content_type, + json.dumps(content_blocks or []), + json.dumps([]), # embeds + json.dumps([]), # components + json.dumps([]), # attachments + json.dumps({}), # reactions + "complete", + json.dumps(metadata or {}), + now, + remote_msg_id, + now, # delivered_at — mark delivered on receipt + ), + ) + await self._db.commit() + return await self.get_message(msg_id) + except sqlite3.IntegrityError: + # Duplicate — the (channel_id, remote_msg_id) unique constraint + # caught a replay. Treat as successful delivery. + await self._db.rollback() + return None + async def get_message(self, message_id: str) -> dict | None: async with self._db.execute( "SELECT * FROM chat_messages WHERE id = ?", (message_id,) diff --git a/tinyagentos/chat/peer_outbox.py b/tinyagentos/chat/peer_outbox.py new file mode 100644 index 000000000..8bcf3f6ab --- /dev/null +++ b/tinyagentos/chat/peer_outbox.py @@ -0,0 +1,130 @@ +"""peer_outbox — store-and-forward queue for DM delivery to remote contacts. + +Messages queued here are drained when the peer link becomes active +(``last_seen_at`` refresh on any peer request). Exponential backoff on +retry: 60s → 120s → 300s → 600s → 1800s cap. +""" + +from __future__ import annotations + +import json +import time +import uuid + +from tinyagentos.base_store import BaseStore + +PEER_OUTBOX_SCHEMA = """ +CREATE TABLE IF NOT EXISTS peer_outbox ( + id TEXT PRIMARY KEY, + contact_id TEXT NOT NULL, + envelope TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_retry_at REAL NOT NULL, + created_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_peer_outbox_retry ON peer_outbox(next_retry_at); +CREATE INDEX IF NOT EXISTS idx_peer_outbox_contact ON peer_outbox(contact_id); +""" + +# Exponential backoff schedule: 60, 120, 300, 600, 1800 (cap) +_BACKOFF_SECONDS = (0, 60, 120, 300, 600, 1800) + + +class PeerOutboxStore(BaseStore): + """Store-and-forward outbox for peer-to-peer message delivery. + + Each row is a serialized envelope queued for delivery to a remote + contact. The drain loop picks up rows where ``next_retry_at <= now`` + and attempts delivery; on failure it increments ``attempts`` and sets + the next retry time with exponential backoff. + """ + + SCHEMA = PEER_OUTBOX_SCHEMA + + async def enqueue( + self, + contact_id: str, + envelope: dict, + *, + next_retry_at: float | None = None, + ) -> str: + """Queue an envelope for delivery. Returns the outbox row id.""" + rid = uuid.uuid4().hex[:12] + now = time.time() + retry_at = next_retry_at if next_retry_at is not None else now + await self._db.execute( + """INSERT INTO peer_outbox (id, contact_id, envelope, attempts, next_retry_at, created_at) + VALUES (?, ?, ?, 0, ?, ?)""", + (rid, contact_id, json.dumps(envelope), retry_at, now), + ) + await self._db.commit() + return rid + + async def dequeue_due( + self, + contact_id: str, + limit: int = 10, + ) -> list[dict]: + """Return up to *limit* rows whose ``next_retry_at`` is due, oldest first.""" + now = time.time() + async with self._db.execute( + """SELECT * FROM peer_outbox + WHERE contact_id = ? AND next_retry_at <= ? + ORDER BY next_retry_at ASC LIMIT ?""", + (contact_id, now, limit), + ) as cursor: + rows = await cursor.fetchall() + columns = [d[0] for d in cursor.description] + return [_row_to_dict(columns, r) for r in rows] + + async def mark_sent(self, outbox_id: str) -> bool: + """Delete the row after successful delivery. Returns True if deleted.""" + cursor = await self._db.execute( + "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) + ) + await self._db.commit() + return cursor.rowcount > 0 + + async def mark_failed(self, outbox_id: str) -> None: + """Increment attempts and push ``next_retry_at`` with exponential backoff.""" + async with self._db.execute( + "SELECT attempts FROM peer_outbox WHERE id = ?", (outbox_id,) + ) as cursor: + row = await cursor.fetchone() + if row is None: + return + new_attempts = row[0] + 1 + delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] + next_at = time.time() + delay + await self._db.execute( + "UPDATE peer_outbox SET attempts = ?, next_retry_at = ? WHERE id = ?", + (new_attempts, next_at, outbox_id), + ) + await self._db.commit() + + async def purge_for_contact(self, contact_id: str) -> int: + """Delete all outbox rows for a contact (used on revoke).""" + cursor = await self._db.execute( + "DELETE FROM peer_outbox WHERE contact_id = ?", (contact_id,) + ) + await self._db.commit() + return cursor.rowcount + + async def count_for_contact(self, contact_id: str) -> int: + """Return the number of pending outbox rows for a contact.""" + async with self._db.execute( + "SELECT COUNT(*) FROM peer_outbox WHERE contact_id = ?", (contact_id,) + ) as cursor: + row = await cursor.fetchone() + return row[0] if row else 0 + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _row_to_dict(columns: list[str], row) -> dict: + if row is None: + return {} + return dict(zip(columns, row)) From a20e863acbe20abd13256cba5ece4b6b614b3ec9 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 14:57:35 +0200 Subject: [PATCH 2/4] fix(collab): resolve 6 Kilo findings on E1 PR #2047 - Log warning on dedup index creation failure instead of silently passing - Validate remote_msg_id non-empty before INSERT (raises ValueError) - Store contact_id in message metadata for downstream consumers - Narrow IntegrityError catch to only suppress dedup violations; re-raise structural violations (e.g. NOT NULL) - Add _MAX_ATTEMPTS=10 terminal-failure guard to peer_outbox; mark_failed returns bool, deletes row at cap - Document _BACKOFF_SECONDS sentinel index 0 --- tests/test_chat_remote_dm.py | 56 +++++++++++++++++++++++++++++++ tinyagentos/chat/message_store.py | 23 +++++++++++-- tinyagentos/chat/peer_outbox.py | 23 ++++++++++--- 3 files changed, 95 insertions(+), 7 deletions(-) diff --git a/tests/test_chat_remote_dm.py b/tests/test_chat_remote_dm.py index 11a9ff967..be6815279 100644 --- a/tests/test_chat_remote_dm.py +++ b/tests/test_chat_remote_dm.py @@ -227,6 +227,37 @@ async def test_send_remote_message_with_metadata(self, msg_store): metadata={"hops_since_user": 0, "delivered_via": "peer"}, ) assert msg["metadata"]["delivered_via"] == "peer" + assert msg["metadata"]["contact_id"] == "hub:hogne" + + async def test_contact_id_stored_in_metadata(self, msg_store): + msg = await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:jaylfc", + contact_id="hub:jaylfc", + remote_msg_id="rm-contact-test", + content="contact id test", + ) + assert msg["metadata"]["contact_id"] == "hub:jaylfc" + + async def test_empty_remote_msg_id_rejected(self, msg_store): + with pytest.raises(ValueError, match="remote_msg_id must be a non-empty string"): + await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id="", + content="empty remote_msg_id", + ) + + async def test_null_like_remote_msg_id_rejected(self, msg_store): + with pytest.raises(ValueError, match="remote_msg_id must be a non-empty string"): + await msg_store.send_remote_message( + channel_id="dm-remote-1", + author_id="hub:hogne", + contact_id="hub:hogne", + remote_msg_id=None, + content="None remote_msg_id", + ) # --------------------------------------------------------------------------- @@ -332,6 +363,31 @@ async def test_dequeue_due_future_retry_not_returned(self, outbox): due = await outbox.dequeue_due("hub:hogne") assert due == [] + async def test_mark_failed_returns_true_while_pending(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + ) + # First failure — still pending + ok = await outbox.mark_failed(rid) + assert ok is True + assert await outbox.count_for_contact("hub:hogne") == 1 + + async def test_mark_failed_terminal_after_max_attempts(self, outbox): + rid = await outbox.enqueue( + contact_id="hub:hogne", + envelope={"kind": "chat"}, + ) + # Retry up to _MAX_ATTEMPTS (10) + for attempt in range(10): + ok = await outbox.mark_failed(rid) + if attempt < 9: + assert ok is True, f"attempt {attempt+1} should still be pending" + else: + # 10th mark_failed should hit max attempts and delete + assert ok is False, f"attempt {attempt+1} should be terminal" + assert await outbox.count_for_contact("hub:hogne") == 0 + # --------------------------------------------------------------------------- # Outbox upgrade — no columns added, but verify init is idempotent diff --git a/tinyagentos/chat/message_store.py b/tinyagentos/chat/message_store.py index 83357d102..68564f605 100644 --- a/tinyagentos/chat/message_store.py +++ b/tinyagentos/chat/message_store.py @@ -113,7 +113,10 @@ async def init(self) -> None: ) await self._db.commit() except Exception: - pass + import logging + _log = logging.getLogger(__name__) + _log.warning("chat_messages: could not create remote dedup index — " + "pre-existing data may violate uniqueness constraint") async def send_message( self, @@ -181,8 +184,17 @@ async def send_remote_message( """ import sqlite3 + if not remote_msg_id or not isinstance(remote_msg_id, str): + raise ValueError("remote_msg_id must be a non-empty string") + msg_id = uuid.uuid4().hex[:12] now = created_at if created_at is not None else time.time() + + # Stash the contact_id in metadata so downstream consumers can + # resolve the sender without parsing author_id. + merged_meta = dict(metadata or {}) + merged_meta.setdefault("contact_id", contact_id) + try: await self._db.execute( """INSERT INTO chat_messages @@ -200,7 +212,7 @@ async def send_remote_message( json.dumps([]), # attachments json.dumps({}), # reactions "complete", - json.dumps(metadata or {}), + json.dumps(merged_meta), now, remote_msg_id, now, # delivered_at — mark delivered on receipt @@ -208,7 +220,12 @@ async def send_remote_message( ) await self._db.commit() return await self.get_message(msg_id) - except sqlite3.IntegrityError: + except sqlite3.IntegrityError as exc: + # Re-raise on structural integrity violations (e.g. NOT NULL on + # channel_id) — only the dedup index violation is benign. + if "remote_msg_id" not in str(exc): + await self._db.rollback() + raise # Duplicate — the (channel_id, remote_msg_id) unique constraint # caught a replay. Treat as successful delivery. await self._db.rollback() diff --git a/tinyagentos/chat/peer_outbox.py b/tinyagentos/chat/peer_outbox.py index 8bcf3f6ab..ba10c1b13 100644 --- a/tinyagentos/chat/peer_outbox.py +++ b/tinyagentos/chat/peer_outbox.py @@ -26,8 +26,11 @@ CREATE INDEX IF NOT EXISTS idx_peer_outbox_contact ON peer_outbox(contact_id); """ -# Exponential backoff schedule: 60, 120, 300, 600, 1800 (cap) +# Exponential backoff schedule (seconds): 60, 120, 300, 600, 1800 (cap). +# Index 0 is a sentinel so ``attempts=1`` maps to ``_BACKOFF_SECONDS[1]`` +# without an off-by-one adjustment. _BACKOFF_SECONDS = (0, 60, 120, 300, 600, 1800) +_MAX_ATTEMPTS = 10 # terminal failure after 10 retries (~6 hours total) class PeerOutboxStore(BaseStore): @@ -85,15 +88,26 @@ async def mark_sent(self, outbox_id: str) -> bool: await self._db.commit() return cursor.rowcount > 0 - async def mark_failed(self, outbox_id: str) -> None: - """Increment attempts and push ``next_retry_at`` with exponential backoff.""" + async def mark_failed(self, outbox_id: str) -> bool: + """Increment attempts and push ``next_retry_at`` with exponential backoff. + + Returns ``True`` if the row is still pending, ``False`` if it was + deleted (terminal failure after ``_MAX_ATTEMPTS`` retries). + """ async with self._db.execute( "SELECT attempts FROM peer_outbox WHERE id = ?", (outbox_id,) ) as cursor: row = await cursor.fetchone() if row is None: - return + return False new_attempts = row[0] + 1 + if new_attempts >= _MAX_ATTEMPTS: + # Terminal failure — drop the row so it never retries again. + await self._db.execute( + "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) + ) + await self._db.commit() + return False delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] next_at = time.time() + delay await self._db.execute( @@ -101,6 +115,7 @@ async def mark_failed(self, outbox_id: str) -> None: (new_attempts, next_at, outbox_id), ) await self._db.commit() + return True async def purge_for_contact(self, contact_id: str) -> int: """Delete all outbox rows for a contact (used on revoke).""" From 1869a297d2f723fb28ffec6f33d3fe6bb2886888 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 15:02:13 +0200 Subject: [PATCH 3/4] fix(collab): resolve 3 CodeRabbit findings on PR #2047 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Use time.time() for delivered_at (actual receipt time), separate from sender's created_at — per CR spec conformance - Narrow IntegrityError suppression to UNIQUE + remote_msg_id only (re-raise all other constraint violations like NOT NULL) - Wrap mark_failed read+write in transaction to eliminate read-then-write race on attempts counter --- tinyagentos/chat/message_store.py | 4 +-- tinyagentos/chat/peer_outbox.py | 47 ++++++++++++++++++------------- 2 files changed, 30 insertions(+), 21 deletions(-) diff --git a/tinyagentos/chat/message_store.py b/tinyagentos/chat/message_store.py index 68564f605..7f4db858f 100644 --- a/tinyagentos/chat/message_store.py +++ b/tinyagentos/chat/message_store.py @@ -215,7 +215,7 @@ async def send_remote_message( json.dumps(merged_meta), now, remote_msg_id, - now, # delivered_at — mark delivered on receipt + time.time(), # delivered_at — actual receipt time, not sender's created_at ), ) await self._db.commit() @@ -223,7 +223,7 @@ async def send_remote_message( except sqlite3.IntegrityError as exc: # Re-raise on structural integrity violations (e.g. NOT NULL on # channel_id) — only the dedup index violation is benign. - if "remote_msg_id" not in str(exc): + if "UNIQUE" not in str(exc) or "remote_msg_id" not in str(exc): await self._db.rollback() raise # Duplicate — the (channel_id, remote_msg_id) unique constraint diff --git a/tinyagentos/chat/peer_outbox.py b/tinyagentos/chat/peer_outbox.py index ba10c1b13..c764d05a1 100644 --- a/tinyagentos/chat/peer_outbox.py +++ b/tinyagentos/chat/peer_outbox.py @@ -93,29 +93,38 @@ async def mark_failed(self, outbox_id: str) -> bool: Returns ``True`` if the row is still pending, ``False`` if it was deleted (terminal failure after ``_MAX_ATTEMPTS`` retries). + + Wrapped in a transaction to avoid a read-then-write race on + ``attempts``. """ - async with self._db.execute( - "SELECT attempts FROM peer_outbox WHERE id = ?", (outbox_id,) - ) as cursor: - row = await cursor.fetchone() - if row is None: - return False - new_attempts = row[0] + 1 - if new_attempts >= _MAX_ATTEMPTS: - # Terminal failure — drop the row so it never retries again. + await self._db.execute("BEGIN") + try: + async with self._db.execute( + "SELECT attempts FROM peer_outbox WHERE id = ?", (outbox_id,) + ) as cursor: + row = await cursor.fetchone() + if row is None: + await self._db.execute("ROLLBACK") + return False + new_attempts = row[0] + 1 + if new_attempts >= _MAX_ATTEMPTS: + # Terminal failure — drop the row so it never retries again. + await self._db.execute( + "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) + ) + await self._db.commit() + return False + delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] + next_at = time.time() + delay await self._db.execute( - "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) + "UPDATE peer_outbox SET attempts = ?, next_retry_at = ? WHERE id = ?", + (new_attempts, next_at, outbox_id), ) await self._db.commit() - return False - delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] - next_at = time.time() + delay - await self._db.execute( - "UPDATE peer_outbox SET attempts = ?, next_retry_at = ? WHERE id = ?", - (new_attempts, next_at, outbox_id), - ) - await self._db.commit() - return True + return True + except Exception: + await self._db.execute("ROLLBACK") + raise async def purge_for_contact(self, contact_id: str) -> int: """Delete all outbox rows for a contact (used on revoke).""" From 5c2b0d38a3e7241086cc8fc44040436b33bbb16b Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 15:16:15 +0200 Subject: [PATCH 4/4] fix(collab): use atomic UPDATE ... RETURNING in peer_outbox mark_failed Replaces the read-then-write race pattern with an atomic SQLite UPDATE ... RETURNING increment so concurrent drain passes cannot undercount attempts. Keeps the _MAX_ATTEMPTS terminal-failure path. Resolves CodeRabbit finding on PR #2047. --- tinyagentos/chat/peer_outbox.py | 49 +++++++++++++++------------------ 1 file changed, 22 insertions(+), 27 deletions(-) diff --git a/tinyagentos/chat/peer_outbox.py b/tinyagentos/chat/peer_outbox.py index c764d05a1..01e15494d 100644 --- a/tinyagentos/chat/peer_outbox.py +++ b/tinyagentos/chat/peer_outbox.py @@ -94,37 +94,32 @@ async def mark_failed(self, outbox_id: str) -> bool: Returns ``True`` if the row is still pending, ``False`` if it was deleted (terminal failure after ``_MAX_ATTEMPTS`` retries). - Wrapped in a transaction to avoid a read-then-write race on - ``attempts``. + Uses ``UPDATE ... RETURNING`` for an atomic increment so concurrent + drain passes (or overlapping retry windows) cannot undercount + attempts. """ - await self._db.execute("BEGIN") - try: - async with self._db.execute( - "SELECT attempts FROM peer_outbox WHERE id = ?", (outbox_id,) - ) as cursor: - row = await cursor.fetchone() - if row is None: - await self._db.execute("ROLLBACK") - return False - new_attempts = row[0] + 1 - if new_attempts >= _MAX_ATTEMPTS: - # Terminal failure — drop the row so it never retries again. - await self._db.execute( - "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) - ) - await self._db.commit() - return False - delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] - next_at = time.time() + delay + async with self._db.execute( + "UPDATE peer_outbox SET attempts = attempts + 1 WHERE id = ? RETURNING attempts", + (outbox_id,), + ) as cursor: + row = await cursor.fetchone() + if row is None: + return False + new_attempts = row[0] + if new_attempts >= _MAX_ATTEMPTS: + # Terminal failure — drop the row so it never retries again. await self._db.execute( - "UPDATE peer_outbox SET attempts = ?, next_retry_at = ? WHERE id = ?", - (new_attempts, next_at, outbox_id), + "DELETE FROM peer_outbox WHERE id = ?", (outbox_id,) ) await self._db.commit() - return True - except Exception: - await self._db.execute("ROLLBACK") - raise + return False + delay = _BACKOFF_SECONDS[min(new_attempts, len(_BACKOFF_SECONDS) - 1)] + await self._db.execute( + "UPDATE peer_outbox SET next_retry_at = ? WHERE id = ?", + (time.time() + delay, outbox_id), + ) + await self._db.commit() + return True async def purge_for_contact(self, contact_id: str) -> int: """Delete all outbox rows for a contact (used on revoke)."""