feat(collab): E1 — DM schema migration (remote_msg_id, delivered_at, peer_outbox)#2047
feat(collab): E1 — DM schema migration (remote_msg_id, delivered_at, peer_outbox)#2047hognek wants to merge 4 commits into
Conversation
…t, peer_outbox - 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: jaylfc#2020, epic jaylfc#2012
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
Next review available in: 34 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughChangesThe chat store now supports schema migration and deduplication for remotely received messages. A new SQLite-backed peer outbox provides queued delivery, retries, purging, and counting, and is initialized and exposed through the application lifecycle. Tests cover migrations, message handling, outbox operations, and reopening existing databases. Remote DM delivery
Estimated code review effort: 4 (Complex) | ~45 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
|
![kilo-code-bot[bot]](https://avatars.githubusercontent.com/in/2193792?s=60&v=4){kind=small}
| # 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( |
There was a problem hiding this comment.
WARNING: CREATE UNIQUE INDEX failure is silently swallowed by the bare except Exception: pass.
If the index cannot be created (e.g. a future backfill/migration introduces two rows with the same non-NULL remote_msg_id, or a partial-index corruption), the app boots without the dedup index. send_remote_message would then silently double-insert the same remote message with different ids, breaking idempotency. Consider logging the failure and/or verifying the index exists after _post_init so a missing index is a loud boot error rather than a silent correctness regression.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| ) | ||
| await self._db.commit() | ||
| return await self.get_message(msg_id) | ||
| except sqlite3.IntegrityError: |
There was a problem hiding this comment.
WARNING: The blanket except sqlite3.IntegrityError: return None conflates a genuine duplicate with any other constraint violation (e.g. a NULL in the NOT NULL columns channel_id/author_id/content, or a malformed value).
All of these paths are reported to the caller as "duplicate already delivered" (None), masking real insert failures and dropping the message without surfacing an error. The only realistic IntegrityError source today is the dedup index, but this is fragile: tightening any NOT NULL/column constraint later would turn hard errors into silent no-ops. Prefer catching only the dedup case (e.g. inspect e.sqlite_errorname == 'CONSTRAINT_UNIQUE' / the index name) and re-raising or returning an error sentinel for other violations.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| self, | ||
| channel_id: str, | ||
| author_id: str, | ||
| contact_id: str, |
There was a problem hiding this comment.
SUGGESTION: The contact_id parameter is accepted but never used — it is not persisted in the chat_messages row nor referenced anywhere in the method.
Callers may assume it is stored for later routing/audit, but it is silently discarded. Either persist it (e.g. in metadata or a dedicated column) or remove the parameter to avoid implying behavior that does not happen.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| "complete", | ||
| json.dumps(metadata or {}), | ||
| now, | ||
| remote_msg_id, |
There was a problem hiding this comment.
SUGGESTION: remote_msg_id is taken directly from the caller with no non-null/non-empty validation, yet the dedup index is partial (WHERE remote_msg_id IS NOT NULL).
If a caller passes remote_msg_id=None (the signature types it as str but enforces nothing at runtime), the value is stored as NULL and the unique index is bypassed — dedup is silently disabled and duplicate remote deliveries are inserted. Add a guard that rejects/raises on falsy remote_msg_id, or document loudly that None intentionally disables dedup.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| row = await cursor.fetchone() | ||
| if row is None: | ||
| return | ||
| new_attempts = row[0] + 1 |
There was a problem hiding this comment.
WARNING: There is no maximum-attempt / terminal-failure state. mark_failed increments attempts and pushes next_retry_at forward forever (the delay caps at 1800s but attempts grows unbounded).
A message whose peer is permanently unreachable will be retried indefinitely and the peer_outbox row is never removed or flagged dead, so the queue grows without bound and count_for_contact / dequeue_due keep returning it. Consider a max_attempts after which the row is moved to a dead-letter state (or purged) so the outbox does not accumulate undead rows.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| """ | ||
|
|
||
| # Exponential backoff schedule: 60, 120, 300, 600, 1800 (cap) | ||
| _BACKOFF_SECONDS = (0, 60, 120, 300, 600, 1800) |
There was a problem hiding this comment.
SUGGESTION: The leading 0 in _BACKOFF_SECONDS = (0, 60, 120, 300, 600, 1800) is dead code.
attempts starts at 0 and the first mark_failed sets new_attempts = 1, so index 0 is never read; the effective schedule is indices 1–5. The module docstring and the comment on line 29 both describe the schedule as "60, 120, 300, 600, 1800" (5 values), which does not match the 6-tuple here. Drop the 0 and align the tuple with the documented schedule to avoid confusion.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
Code Review SummaryStatus: No New Issues Found | Recommendation: Merge OverviewThe incremental diff ( Changes Verified (1 file)
Prior Findings Resolved by This Update
Files Reviewed (1 file, incremental)
Previous Review Summaries (3 snapshots, latest commit 1869a29)Current summary above is authoritative. Previous snapshots are kept for context only. Previous review (commit 1869a29)Status: No New Issues Found | Recommendation: Merge OverviewThe incremental diff ( Changes Verified (2 files)
Prior Findings Resolved by This Update
Files Reviewed (2 files, incremental)
Previous review (commit a20e863)Status: 1 Issue Found | Recommendation: Address before merge Overview
Issue Details (click to expand)SUGGESTION
Previously Reported Issues — Resolved in This UpdateAll 6 findings from the prior review commit (
Files Reviewed (4 files)
Fix these issues in Kilo Cloud Previous review (commit 3695296)Status: 6 Issues Found | Recommendation: Address before merge Overview
Issue Details (click to expand)WARNING
SUGGESTION
Files Reviewed (4 files)
Reviewed by hy3:free · Input: 42.1K · Output: 2.2K · Cached: 102K |
- 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
| 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): |
There was a problem hiding this comment.
SUGGESTION: Dedup detection relies on substring-matching the exception message ("remote_msg_id" not in str(exc)).
This is brittle: it assumes SQLite always renders the violating column name in the error string for the unique constraint. If the index/column is renamed, or a future SQLite build changes the message wording, a genuine duplicate would no longer be recognized as benign and would instead hit the raise branch (surfacing as an unhandled error), or a non-dedup IntegrityError whose message happens to contain the substring would be silently swallowed as a duplicate. Prefer inspecting exc.sqlite_errorname (e.g. == 'CONSTRAINT_UNIQUE') and/or checking the index name explicitly for a robust distinction.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
tinyagentos/chat/message_store.py (1)
159-216: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
contact_idparameter is accepted but never used.
contact_idis a required positional parameter (line 163) yet the method body never reads it — it's not stored, not validated againstauthor_id, and not used to scope anything. As written it's dead code that could mislead future maintainers into assuming some contact-based check happens here.Either remove the parameter or wire it into the intended behavior (e.g. validating
author_idcorresponds tocontact_id, or persisting it for audit/outbox correlation).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tinyagentos/chat/message_store.py` around lines 159 - 216, The send_remote_message method accepts contact_id but never uses it; remove this parameter if it is not required by the API, or implement the intended contact-based validation or persistence so it meaningfully affects message handling. Update all callers consistently and preserve the existing deduplication and commit behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tinyagentos/chat/message_store.py`:
- Around line 182-216: In the message insertion flow, decouple the timestamps
used by the created_at and delivered_at columns: preserve the caller-supplied
created_at value for historical creation time, but compute delivered_at from the
current wall-clock time when the hub receives the message. Update the variables
and INSERT parameters in the surrounding message-store method so the inline
receipt-time behavior remains accurate.
- Around line 95-116: Update the schema migration logic in _post_init to stop
swallowing failures from CREATE UNIQUE INDEX idx_chat_messages_remote_dedup,
allowing unexpected database errors to propagate. Narrow the exception handling
for the remote_msg_id and delivered_at ALTER TABLE statements to ignore only the
expected “column already exists” condition, while surfacing or logging all other
failures.
In `@tinyagentos/chat/peer_outbox.py`:
- Around line 88-103: Update mark_failed to perform the attempts increment
atomically in SQLite rather than reading attempts and writing a computed value
in separate statements. Use an UPDATE ... RETURNING flow to obtain the
incremented attempts value, calculate next_retry_at from that value, and
preserve the existing no-op behavior when outbox_id does not exist.
---
Nitpick comments:
In `@tinyagentos/chat/message_store.py`:
- Around line 159-216: The send_remote_message method accepts contact_id but
never uses it; remove this parameter if it is not required by the API, or
implement the intended contact-based validation or persistence so it
meaningfully affects message handling. Update all callers consistently and
preserve the existing deduplication and commit behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: f49b87ea-c6b6-4079-b007-4fc67af11c7f
📒 Files selected for processing (4)
tests/test_chat_remote_dm.pytinyagentos/app.pytinyagentos/chat/message_store.pytinyagentos/chat/peer_outbox.py
| # 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 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Silent failure of the dedup unique index defeats the PR's core guarantee.
The CREATE UNIQUE INDEX ... idx_chat_messages_remote_dedup is the sole mechanism enforcing idempotent remote delivery. Wrapping it in a blanket except Exception: pass means any failure other than "already exists" (disk/permission/syntax issue) silently leaves the index missing, and send_remote_message's dedup would then silently stop working with zero signal to operators. Same concern applies to the two ALTER TABLE guards above it.
🐛 Proposed fix: narrow the except and log unexpected failures
- 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
+ try:
+ await self._db.execute("ALTER TABLE chat_messages ADD COLUMN remote_msg_id TEXT")
+ await self._db.commit()
+ except sqlite3.OperationalError as exc:
+ if "duplicate column" not in str(exc).lower():
+ raise
+ try:
+ await self._db.execute("ALTER TABLE chat_messages ADD COLUMN delivered_at REAL")
+ await self._db.commit()
+ except sqlite3.OperationalError as exc:
+ if "duplicate column" not in str(exc).lower():
+ raise
# 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
+ 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()CREATE UNIQUE INDEX IF NOT EXISTS is already idempotent, so it doesn't need a try/except at all — a real failure here should surface loudly rather than be swallowed.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # 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 | |
| # 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 sqlite3.OperationalError as exc: | |
| if "duplicate column" not in str(exc).lower(): | |
| raise | |
| try: | |
| await self._db.execute("ALTER TABLE chat_messages ADD COLUMN delivered_at REAL") | |
| await self._db.commit() | |
| except sqlite3.OperationalError as exc: | |
| if "duplicate column" not in str(exc).lower(): | |
| raise | |
| # Dedup index MUST be created here (not in SCHEMA) because it references | |
| # columns added by _post_init — per BaseStore boot-brick rule. | |
| 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() |
🧰 Tools
🪛 Ruff (0.15.21)
[error] 99-100: try-except-pass detected, consider logging the exception
(S110)
[warning] 99-99: Do not catch blind exception: Exception
(BLE001)
[error] 104-105: try-except-pass detected, consider logging the exception
(S110)
[warning] 104-104: Do not catch blind exception: Exception
(BLE001)
[error] 115-116: try-except-pass detected, consider logging the exception
(S110)
[warning] 115-115: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/chat/message_store.py` around lines 95 - 116, Update the schema
migration logic in _post_init to stop swallowing failures from CREATE UNIQUE
INDEX idx_chat_messages_remote_dedup, allowing unexpected database errors to
propagate. Narrow the exception handling for the remote_msg_id and delivered_at
ALTER TABLE statements to ignore only the expected “column already exists”
condition, while surfacing or logging all other failures.
Source: Linters/SAST tools
- 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
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 jaylfc#2047.
1 participant
Summary
Implements E1 of the cross-user collaboration epic (#2012): human DM schema migration and peer outbox for store-and-forward delivery.
Closes #2020.
Changes
Schema migration (boot-brick safe)
remote_msg_id TEXTanddelivered_at REALcolumns via guarded_post_initALTER TABLE pattern (same as existingdeleted_at/expires_at)UNIQUE INDEXon(channel_id, remote_msg_id) WHERE remote_msg_id IS NOT NULLfor idempotent remote delivery_post_init(not SCHEMA) per BaseStore boot-brick rulesend_remote_message()
ChatMessageStore— atomic INSERT with dedupIntegrityErroron duplicate(channel_id, remote_msg_id)and returnsNonedelivered_aton receipt,author_idnamespaced ashub:{username}PeerOutboxStore
peer_outboxtable with exponential backoff retry schedule (60s → 120s → 300s → 600s → 1800s cap)enqueue,dequeue_due,mark_sent,mark_failed,purge_for_contact,count_for_contactapp.state.peer_outboxTests
tests/test_chat_remote_dm.pyAll tests pass
Summary by CodeRabbit
New Features
Bug Fixes