Fix #85: Duplicate chat message submissions produce inconsistent AI-corrected content #119

Open

typosaurus wants to merge 2 commits from typosaurus/ticket-85 into master

typosaurus commented

2026-07-23 03:33:26 +02:00

Collaborator

Resolves #85.

Plan

Implementation Plan: Cross-Worker Dedup for Chat Messages

1. Add Database-Level Unique Index

File: devplacepy/database/schema.py
Action: After the existing table creation, add a migration function that creates a unique composite index on (sender_uid, receiver_uid, content) in the messages table. Use raw SQL via the dataset connection to ensure the index exists. The index name: idx_messages_unique_sender_receiver_content.

If the table already has rows with duplicates, the migration must either:

Remove duplicates first (keep the oldest row), or
Use INSERT OR IGNORE only going forward (safer, no data loss).
Given the ticket, assume no production data loss is acceptable. Strategy: Add the index with CREATE UNIQUE INDEX IF NOT EXISTS ... – if duplicates exist, the migration will fail. To avoid this, first delete duplicates with a SQL query keeping the oldest id per group. This is safe because duplicates were unintended.

def _add_unique_message_index(db):
    # Remove duplicates: keep the row with smallest id for each (sender_uid, receiver_uid, content)
    db.query("""
        DELETE FROM messages WHERE id NOT IN (
            SELECT MIN(id) FROM messages GROUP BY sender_uid, receiver_uid, content
        )
    """)
    db.query("""
        CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_unique_sender_receiver_content
        ON messages(sender_uid, receiver_uid, content)
    """)

Call this function after the schema is initialised.

2. Modify `persist_message` to Use DB Constraint

File: devplacepy/persist.py
Action: In the persist_message function (around lines 88–106), replace the in-memory dedup check with a try/except that attempts an insert and falls back to selecting the existing row on IntegrityError.

Remove or keep the in-memory cache as a first-pass speed optimisation, but always fall through to DB insert so the constraint is the authority.
The key change: instead of returning from the cache, attempt to insert the row using dataset (which uses SQLAlchemy). If an IntegrityError (SQLite raises sqlite3.IntegrityError) is caught, query for the existing row with same sender_uid, receiver_uid, content and return that row's UID.

Implementation sketch:

from sqlalchemy.exc import IntegrityError

try:
    # existing insert logic (lines 111-120)
    messages.insert(dict(...))
except IntegrityError:
    # Duplicate: fetch the existing row
    existing = messages.find_one(sender_uid=sender_uid, receiver_uid=receiver_uid, content=content)
    if existing:
        msg_uid = existing['uid']
        # do NOT call schedule_correction again
        return msg_uid
    else:
        raise  # unexpected

Important: After catching the IntegrityError, we must not call schedule_correction or schedule_modification for the duplicate. The correction was already scheduled for the first insert. So move those calls outside the try block, only in the non-duplicate path.

3. Update `_content_cache` Usage

The existing in-memory cache can remain as an optimisation, but it should not be relied upon for correctness. Ensure that the DB constraint is always the final check. If _content_cache returns a hit, still proceed to the DB insert to confirm. This adds minimal overhead and guarantees correctness across workers.

4. Adjust `schedule_correction` Call Location

Currently schedule_correction is called after insert regardless. Move it inside the successful insert branch only.

5. Migrate Existing Duplicate Rows

Implement a migration step (as described in step 1) that deduplicates existing data before adding the unique index. This ensures the migration doesn't fail.

6. Review and Update Existing Test

File: tests/api/messages/send.py – test test_duplicate_message_returns_same_uid already expects same UID. It should still pass because the DB constraint will cause the second insert to return the existing row with the same UID. No change needed unless the test uses a single worker; it will now work across workers too.

Add a new test (optional but recommended): test_duplicate_message_different_workers – simulate two concurrent requests via threads with separate sessions. This is not mandatory for the fix but would improve coverage.

7. Verify No Side Effects

Run the full test suite and ensure no regressions, especially the previously failing tests noted in the lessons:

tests/unit/services/game/store.py::test_advance_quests_increments_and_caps
tests/unit/services/deepsearch/chat.py::test_answer_is_grounded_and_cited

These failures were likely caused by previous incorrect changes; this plan should not affect them.

Definition of Done

A migration function is added to schema.py that removes duplicate (sender_uid, receiver_uid, content) rows (keeping the oldest) and creates a UNIQUE index on those columns.
persist_message in persist.py catches IntegrityError on insert, fetches the existing row, returns its UID without scheduling correction/modification a second time.
schedule_correction and schedule_modification are called only for new inserts, not for duplicates.
The in-memory _content_cache is retained as a best-effort optimisation but a DB insert is always attempted.
The existing test test_duplicate_message_returns_same_uid passes (unchanged expectations).
The complete test suite passes: pip install -e '.[dev]' -q && python -m pytest
No lint violations are introduced (run ruff . or equivalent linter and fix any new issues).
The fix works in multi-worker mode: tested manually or via integration test (e.g., two simultaneous POSTs to different workers return same UID and same corrected content).

Opened automatically by Typosaurus.

Resolves #85. ## Plan ## Implementation Plan: Cross-Worker Dedup for Chat Messages ### 1. Add Database-Level Unique Index **File:** `devplacepy/database/schema.py` **Action:** After the existing table creation, add a migration function that creates a unique composite index on `(sender_uid, receiver_uid, content)` in the `messages` table. Use raw SQL via the `dataset` connection to ensure the index exists. The index name: `idx_messages_unique_sender_receiver_content`. If the table already has rows with duplicates, the migration must either: - Remove duplicates first (keep the oldest row), or - Use `INSERT OR IGNORE` only going forward (safer, no data loss). Given the ticket, assume no production data loss is acceptable. **Strategy:** Add the index with `CREATE UNIQUE INDEX IF NOT EXISTS ...` – if duplicates exist, the migration will fail. To avoid this, first delete duplicates with a SQL query keeping the oldest `id` per group. This is safe because duplicates were unintended. ```python def _add_unique_message_index(db): # Remove duplicates: keep the row with smallest id for each (sender_uid, receiver_uid, content) db.query(""" DELETE FROM messages WHERE id NOT IN ( SELECT MIN(id) FROM messages GROUP BY sender_uid, receiver_uid, content ) """) db.query(""" CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_unique_sender_receiver_content ON messages(sender_uid, receiver_uid, content) """) ``` Call this function after the schema is initialised. ### 2. Modify `persist_message` to Use DB Constraint **File:** `devplacepy/persist.py` **Action:** In the `persist_message` function (around lines 88–106), replace the in-memory dedup check with a try/except that attempts an insert and falls back to selecting the existing row on `IntegrityError`. - Remove or keep the in-memory cache as a first-pass speed optimisation, but **always** fall through to DB insert so the constraint is the authority. - The key change: instead of returning from the cache, attempt to insert the row using `dataset` (which uses SQLAlchemy). If an `IntegrityError` (SQLite raises `sqlite3.IntegrityError`) is caught, query for the existing row with same `sender_uid, receiver_uid, content` and return that row's UID. Implementation sketch: ```python from sqlalchemy.exc import IntegrityError try: # existing insert logic (lines 111-120) messages.insert(dict(...)) except IntegrityError: # Duplicate: fetch the existing row existing = messages.find_one(sender_uid=sender_uid, receiver_uid=receiver_uid, content=content) if existing: msg_uid = existing['uid'] # do NOT call schedule_correction again return msg_uid else: raise # unexpected ``` **Important:** After catching the IntegrityError, we must **not** call `schedule_correction` or `schedule_modification` for the duplicate. The correction was already scheduled for the first insert. So move those calls outside the try block, only in the non-duplicate path. ### 3. Update `_content_cache` Usage The existing in-memory cache can remain as an optimisation, but it should not be relied upon for correctness. Ensure that the DB constraint is always the final check. If `_content_cache` returns a hit, still proceed to the DB insert to confirm. This adds minimal overhead and guarantees correctness across workers. ### 4. Adjust `schedule_correction` Call Location Currently `schedule_correction` is called after insert regardless. Move it inside the successful insert branch only. ### 5. Migrate Existing Duplicate Rows Implement a migration step (as described in step 1) that deduplicates existing data before adding the unique index. This ensures the migration doesn't fail. ### 6. Review and Update Existing Test **File:** `tests/api/messages/send.py` – test `test_duplicate_message_returns_same_uid` already expects same UID. It should still pass because the DB constraint will cause the second insert to return the existing row with the same UID. No change needed unless the test uses a single worker; it will now work across workers too. Add a new test (optional but recommended): `test_duplicate_message_different_workers` – simulate two concurrent requests via threads with separate sessions. This is not mandatory for the fix but would improve coverage. ### 7. Verify No Side Effects Run the full test suite and ensure no regressions, especially the previously failing tests noted in the lessons: - `tests/unit/services/game/store.py::test_advance_quests_increments_and_caps` - `tests/unit/services/deepsearch/chat.py::test_answer_is_grounded_and_cited` These failures were likely caused by previous incorrect changes; this plan should not affect them. --- ## Definition of Done - [ ] A migration function is added to `schema.py` that removes duplicate `(sender_uid, receiver_uid, content)` rows (keeping the oldest) and creates a UNIQUE index on those columns. - [ ] `persist_message` in `persist.py` catches `IntegrityError` on insert, fetches the existing row, returns its UID without scheduling correction/modification a second time. - [ ] `schedule_correction` and `schedule_modification` are called only for new inserts, not for duplicates. - [ ] The in-memory `_content_cache` is retained as a best-effort optimisation but a DB insert is always attempted. - [ ] The existing test `test_duplicate_message_returns_same_uid` passes (unchanged expectations). - [ ] The complete test suite passes: `pip install -e '.[dev]' -q && python -m pytest` - [ ] No lint violations are introduced (run `ruff .` or equivalent linter and fix any new issues). - [ ] The fix works in multi-worker mode: tested manually or via integration test (e.g., two simultaneous POSTs to different workers return same UID and same corrected content). Opened automatically by Typosaurus.

typosaurus added 2 commits 2026-07-23 03:33:27 +02:00