Merge pull request #1833 from razzant/codex/reminder-history-parity-20261014

Fix duplicate chat rows across live and history delivery

docs/CLIENT_CONTRACT.md

@@ -242,7 +242,7 @@ In a Project response, `window.latest_message` names the latest eligible non-own
 
 ### Merging live frames and history
 
-The core writes each stored row and sends its live frame with the same `ts` string (the same characters, not only the same instant), the same `task_id`, `role`, `system_type` and `is_progress`, and for a text row the same words. So a live frame and a history row are the same row when:
+The core writes each stored row and sends its live frame with an identical `ts` string, the same `task_id`, `role`, `system_type`, `is_progress` and `source` (when provided), and for a text row the same words. So a live frame and a history row are the same row when:
 
 | Live frame | History row | Same row when |
 |---|---|---|

supervisor/message_bus.py

@@ -1480,6 +1480,7 @@ def send_with_budget(chat_id: int, text: str, log_text: Optional[str] = None,
             text if log_text is None else log_text,
             ts=msg_ts,
             fmt=fmt,
+            source=str(meta.get("source") or ""),
             task_id=task_id,
             record_type=system_type,
             message_meta=progress_meta,

tests/test_bugfixes_airi.py

@@ -108,7 +108,8 @@ def test_reconnect_merges_user_rows_without_clearing_visible_history():
     # Keyed reconciliation replaces the former clear-and-rebuild requirement:
     # physical rows dedupe and an offline/local echo is adopted in place.
     assert "opts.historyId ? `history:${opts.historyId}` : legacyKey" in add
-    assert "if (messageKey && seenMessageKeys.has(messageKey))" in add
+    # A retained DOM row or a remembered key suppresses a late live copy.
+    assert "if (prior || (messageKey && seenMessageKeys.has(messageKey))) return false;" in add
     assert "node.dataset.clientMessageId === clientMessageId" in add
     assert "stampHistoryNode(prior, opts.historyId, opts.historyPosition);" in add
     assert "seenMessageKeys.clear();" not in sync + replay

tests/test_client_contract_history.py

@@ -52,6 +52,10 @@ def _produce(chat) -> None:
     message_bus.send_with_budget(1, "Here is the plan.", task_id="t1")
     message_bus.send_with_budget(1, "Here is the plan.", task_id="t2")  # same words, another task
     message_bus.send_with_budget(1, "Review settled.", task_id="t1", role="system", system_type="custody_notice")
+    for fmt in ("", "markdown"):
+        for index, meta in enumerate(({"source": "custom-service"}, {}, {"source": ""})):
+            message_bus.send_with_budget(1, f"**Notice {index}** ({fmt or 'plain'})", fmt=fmt,
+                                         task_id="t1", progress_meta=meta)
     bridge.send_document(1, b"%PDF-1.4 data", filename="plan.pdf", caption="the plan",
                          mime="application/pdf", task_id="t1")
     bridge.send_photo(1, b"\x89PNG\r\n\x1a\n" + b"0" * 16, caption="map", task_id="t1")
@@ -115,6 +119,7 @@ def test_every_stored_live_frame_finds_exactly_its_history_row(chat):
         assert len(matches) == 1, (frame, matches)
         # The facts the rule relies on: the stored row keeps the live frame's exact time string.
         assert matches[0]["ts"] == frame["ts"]
+        assert matches[0].get("source", "") == frame.get("source", ""), (frame, matches[0])
     live_keys = [_live_key(frame) for frame in stored]
     assert len(live_keys) == len(set(live_keys)), "the same words on another task stay distinct"
 

tests/test_host_service_notify.py

@@ -61,6 +61,8 @@ def test_identity_advertises_the_notify_route(tmp_path: pathlib.Path) -> None:
 
 
 def test_a_notice_is_one_signed_system_row_in_the_owners_chat(tmp_path: pathlib.Path, chat) -> None:
+    from ouroboros.gateway.history import make_chat_history_endpoint
+
     client, _app = _client(tmp_path)
     response = client.post("/notify", headers=HEADERS,
                            json={"text": "  ⏰ Meeting with Ivan in 15 min ", "source": "Ouroboros"})
@@ -73,6 +75,12 @@ def test_a_notice_is_one_signed_system_row_in_the_owners_chat(tmp_path: pathlib.
     assert row["ts"] == body["ts"]
     [frame] = [f for f in chat.frames if f.get("type") == "chat"]
     assert (frame["role"], frame["system_type"], frame["source"]) == ("system", "skill_notice", "cal")
+    history = asyncio.run(make_chat_history_endpoint(tmp_path)(types.SimpleNamespace(query_params={"chat_id": "1"})))
+    [replayed] = json.loads(history.body)["messages"]
+    assert replayed["source"] == row["source"] == frame["source"] == "cal", "the body cannot spoof the skill's source"
+    assert replayed["text"] == row["text"] == frame["content"]
+    assert replayed["ts"] == row["ts"] == frame["ts"]
+    assert replayed["role"] == "system" and replayed["system_type"] == "skill_notice" and replayed["task_id"] == ""
     assert [topic for topic, _ in chat.published] == ["chat.outbound"]
 
 

tests/test_schedule_notes.py

@@ -69,6 +69,7 @@ def test_a_due_note_is_one_signed_system_row_and_never_a_task(host):
     assert body == "Call mother", "the mind's words ride verbatim"
     [frame] = [f for f in host.frames if f.get("type") == "chat"]
     assert (frame["role"], frame["system_type"], frame["source"]) == ("system", "reminder", "Ouroboros")
+    assert row["source"] == frame["source"]
     assert frame["set_at"] == "1999-12-31T22:00:00+00:00"
     assert frame["scheduled_for"] == ONCE_PAST["run_at"] and frame["delivered_at"]
     assert [topic for topic, _ in host.published] == ["chat.outbound"], "transports see it like any System row"
@@ -249,6 +250,11 @@ def test_history_shows_the_row_and_only_live_frames_ring(host):
     messages = json.loads(response.body)["messages"]
     [replayed] = [m for m in messages if m.get("system_type") == "reminder"]
     assert replayed["role"] == "system" and replayed["text"].endswith("\nCall mother")
+    [row] = _chat_rows(host.root)
+    [frame] = [f for f in host.frames if f.get("type") == "chat"]
+    assert replayed["source"] == row["source"] == frame["source"] == "Ouroboros"
+    assert replayed["text"] == row["text"] == frame["content"]
+    assert replayed["ts"] == row["ts"] == frame["ts"]
 
 
 def test_the_next_turn_reads_the_note_as_a_host_fact(host):
@@ -272,7 +278,10 @@ def test_a_note_from_another_source_keeps_its_own_author(host):
     """The voice follows authorship: only the mind's own follow-up is signed Ouroboros."""
     _note(host.queue, source="owner")
     host.queue.check_scheduled_tasks()
-    assert _chat_rows(host.root)[0]["text"].startswith("Reminder · owner · written ")
+    [row] = _chat_rows(host.root)
+    assert row["text"].startswith("Reminder · owner · written ")
+    [frame] = [f for f in host.frames if f.get("type") == "chat"]
+    assert row["source"] == frame["source"] == "owner"
 
 
 def test_note_label_reads_both_times_in_the_schedule_zone():

tests/test_ui_owner_notes_browser.py

@@ -4,7 +4,9 @@ A running supervisor shows a due note in Main as one System row whose host signa
 stands on its own line above the mind's words; reopening replays it from history
 without a banner; Activity names a waiting note by its words; and with notifications
 on, a note that fires while the page is open rings once in the existing messages
-category, titled by its author. No model is called for any of it.
+category, titled by its author. Two reconnects retain that live row and its text
+selection while history gives it a durable identity, without ringing again.
+No model is called for any of it.
 """
 from __future__ import annotations
 
@@ -106,6 +108,61 @@ def test_notes_reach_main_activity_and_the_live_banner(direct_server_with_data,
             assert page.locator("#toast-stack .toast").filter(has_text="Reminder").count() == 1
             page.screenshot(path=str(evidence / "main-live-banner.png"))
 
+            live = page.locator(f"#chat-messages {BUBBLE}").filter(has_text=LIVE)
+            live.evaluate("""(node, words) => {
+                window.__liveReminder = node;
+                const body = node.querySelector('.message').firstChild;
+                const start = body.textContent.lastIndexOf(words);
+                const range = document.createRange();
+                range.setStart(body, start); range.setEnd(body, start + words.length);
+                const selection = getSelection();
+                selection.removeAllRanges(); selection.addRange(range);
+                window.__reminderSelection = { anchorNode: selection.anchorNode,
+                    anchorOffset: selection.anchorOffset, focusNode: selection.focusNode,
+                    focusOffset: selection.focusOffset };
+                window.__extraReminderBanners = 0;
+                window.__reminderBannerObserver = new MutationObserver(records => {
+                    for (const record of records) for (const added of record.addedNodes) {
+                        if (added.nodeType === 1 && added.matches('.toast')
+                            && added.textContent.includes('Reminder from Ouroboros')) {
+                            window.__extraReminderBanners += 1;
+                        }
+                    }
+                });
+                window.__reminderBannerObserver.observe(document.querySelector('#toast-stack'), { childList: true });
+            }""", LIVE)
+            assert page.evaluate("getSelection().toString()") == LIVE
+            for attempt in (1, 2):
+                reconnected = page.locator('#chat-messages .chat-bubble[data-system-type="reconnect"]')
+                before = reconnected.count()
+                with page.expect_response(lambda response: "/api/chat/history" in response.url
+                                          and response.request.method == "GET", timeout=30_000) as history:
+                    page.evaluate("window.__ouroWs.ws.close()")
+                assert history.value.status == 200
+                rows = history.value.json()["messages"]
+                row = next(row for row in rows if row.get("system_type") == "reminder"
+                           and row.get("text", "").endswith(LIVE))
+                # The reconnect row is appended only after syncHistory has painted.
+                # Wait for that independent marker, so a duplicate still gets a RED capture.
+                page.wait_for_function("""before => document.querySelectorAll(
+                    '#chat-messages .chat-bubble[data-system-type="reconnect"]').length > before""",
+                    arg=before, timeout=30_000)
+                page.evaluate(SETTLED)
+                page.screenshot(path=str(evidence / f"main-live-reconnect-{attempt}.png"))
+                assert live.count() == 1, "reconnect history must adopt the live reminder"
+                assert live.evaluate("node => node === window.__liveReminder")
+                assert live.get_attribute("data-history-id") == row["history_id"]
+                assert live.get_attribute("data-history-source") == row["history_position"]["source"]
+                assert live.get_attribute("data-history-offset") == str(row["history_position"]["offset"])
+                assert page.evaluate("""() => {
+                    const current = getSelection(), saved = window.__reminderSelection;
+                    return ['anchorNode', 'anchorOffset', 'focusNode', 'focusOffset']
+                        .every(key => current[key] === saved[key]);
+                }""")
+                assert page.evaluate("getSelection().toString()") == LIVE
+                assert page.evaluate("window.__extraReminderBanners") == 0, "history replay must not ring again"
+            page.evaluate("window.__reminderBannerObserver.disconnect()")
+
             page.click('[data-nav-page="dashboard"]')
             page.click('[data-dashboard-tab="activity"]')
             schedules = page.locator('[data-activity-section="schedules"]')

web/modules/chat.js

@@ -2206,19 +2206,16 @@ export function createChatInstance({
             taskId,
         });
         const messageKey = opts.historyId ? `history:${opts.historyId}` : legacyKey;
-        if (opts.historyId || _historyReplayActive) {
-            const prior = opts.historyId
-                ? historyNodes(opts.historyId).find(node => node.classList.contains('chat-bubble'))
-                : Array.from(messagesDiv.querySelectorAll('.chat-bubble')).find(node => node.dataset.messageKey === legacyKey);
-            if (prior) {
-                rememberMessageKey(messageKey);
-                const originChanged = syncSavedProjectContext(prior, opts.originProjected, opts.originId);
-                return chatDecision.renderRoutingDecision(prior, opts.chatAnnotation) || originChanged;
-            }
-        }
-        if (messageKey && seenMessageKeys.has(messageKey)) {
-            return false;
+        const prior = opts.historyId
+            ? historyNodes(opts.historyId).find(node => node.classList.contains('chat-bubble'))
+            : Array.from(messagesDiv.querySelectorAll('.chat-bubble')).find(node => node.dataset.messageKey === legacyKey);
+        if (prior && (opts.historyId || _historyReplayActive)) {
+            rememberMessageKey(messageKey);
+            const originChanged = syncSavedProjectContext(prior, opts.originProjected, opts.originId);
+            return chatDecision.renderRoutingDecision(prior, opts.chatAnnotation) || originChanged;
         }
+        // History may land first; a live copy adds no new facts.
+        if (prior || (messageKey && seenMessageKeys.has(messageKey))) return false;
         if (opts.historyId) {
             const prior = Array.from(messagesDiv.querySelectorAll('.chat-bubble')).find(node =>
                 (clientMessageId && node.dataset.clientMessageId === clientMessageId)

web/tests/chat_history_integration.test.js

@@ -313,9 +313,40 @@ test('live message adopts its physical history identity without replacing the vi
     assert.deepEqual(f.bubbles().filter((node) => node.innerHTML.includes('A useful answer')), [live]);
 });
 
+for (const systemType of ['reminder', 'skill_notice']) for (const historyFirst of [false, true]) {
+    test(`${systemType} retains one node across ${historyFirst ? 'history then live' : 'live then history'} and repeats`, async t => {
+        const message = row('chat:100', 'A scheduled notice', {
+            role: 'system', system_type: systemType, source: 'Ouroboros',
+        });
+        const live = { role: message.role, content: message.text, ts: message.ts,
+            system_type: message.system_type, source: message.source };
+        const f = fixture(t, page(historyFirst ? [message] : []));
+        await f.refresh();
+        if (!historyFirst) f.emit(live);
+        assert.equal(f.bubbles().length, 1);
+        const original = f.bubbles()[0];
+        if (historyFirst) f.emit(live);
+        else await f.refresh(page([message]));
+        for (let repeat = 0; repeat < 2; repeat += 1) {
+            assert.equal(f.bubbles().length, 1, 'the live frame and saved row describe one occurrence');
+            assert.equal(f.bubbles()[0], original, 'joining the row preserves the original node');
+            assert.equal(original.dataset.historyId, 'chat:100');
+            assert.equal(original.dataset.historySource, 'chat:rotation-1');
+            assert.equal(original.dataset.historyOffset, '100');
+            const saved = sessionStorage.getItem('ouro_chat:2');
+            f.emit(live);
+            assert.equal(sessionStorage.getItem('ouro_chat:2'), saved, 'a late live copy is not persisted again');
+            await f.refresh(page([message]));
+        }
+        assert.equal(f.bubbles().length, 1);
+        assert.equal(f.bubbles()[0], original);
+    });
+}
+
 test('repeated physical history identity updates the routing annotation on the same user bubble', async (t) => {
     const message = row('chat:200', 'Please investigate this', {
         role: 'user', client_message_id: 'owner-message', chat_annotation: { status: 'pending' },
+        source: 'web', sender_session_id: 'another-tab',
     });
     const f = fixture(t, page([message]));
     await f.refresh();
@@ -333,6 +364,42 @@ test('repeated physical history identity updates the routing annotation on the s
     assert.equal(bubble.dataset.chatAnnotationStatus, 'delivered');
     assert.notEqual(note.textContent, 'Choosing the right destination…');
     assert.match(note.textContent, /Investigation/);
+    const annotationText = note.textContent, metadata = { ...bubble.dataset };
+    const saved = sessionStorage.getItem('ouro_chat:2');
+    f.emit({ role: message.role, content: message.text, ts: message.ts,
+        client_message_id: message.client_message_id, source: message.source,
+        sender_session_id: message.sender_session_id });
+    assert.equal(f.bubbles().length, 1, 'a remote owner echo does not append a second bubble');
+    assert.equal(f.bubbles()[0], bubble);
+    assert.equal(bubble.querySelector('.msg-routing-annotation'), note);
+    assert.equal(note.textContent, annotationText, 'the live echo cannot clear the saved routing receipt');
+    assert.deepEqual(bubble.dataset, metadata);
+    assert.equal(sessionStorage.getItem('ouro_chat:2'), saved);
+});
+
+test('a live echo preserves the projected saved-context note and origin metadata', async t => {
+    const origin = row('chat:210', 'Original request', {
+        role: 'user', source: 'web', sender_session_id: 'another-tab',
+        history_id: undefined, history_position: undefined,
+        origin_projected: true, origin_id: 'binding-source-ref',
+    });
+    const f = fixture(t, page([origin]));
+    await f.refresh();
+    assert.equal(f.bubbles().length, 1);
+    const original = f.bubbles()[0], note = original.querySelector('.saved-project-context');
+    assert.ok(note);
+    assert.equal(original.dataset.originId, origin.origin_id);
+    assert.equal(original.dataset.originProjected, '1');
+    const metadata = { ...original.dataset }, noteText = note.textContent;
+    const saved = sessionStorage.getItem('ouro_chat:2');
+    f.emit({ role: origin.role, content: origin.text, ts: origin.ts,
+        source: origin.source, sender_session_id: origin.sender_session_id });
+    assert.equal(f.bubbles().length, 1);
+    assert.equal(f.bubbles()[0], original);
+    assert.equal(original.querySelector('.saved-project-context'), note);
+    assert.equal(note.textContent, noteText);
+    assert.deepEqual(original.dataset, metadata);
+    assert.equal(sessionStorage.getItem('ouro_chat:2'), saved);
 });
 
 test('a replayed refusal receipt shows the host cause and a later scheduled receipt patches the same note', async (t) => {
@@ -363,8 +430,9 @@ test('a replayed refusal receipt shows the host cause and a later scheduled rece
     assert.equal(bubble.querySelector('.msg-routing-actions'), null, 'the current Project is already displayed');
 });
 
-test('two physical rows with identical timestamp and body remain two messages across refresh', async (t) => {
-    const rows = [row('chat:300', 'Same words'), row('chat:400', 'Same words')];
+test('two physical rows with one full legacy key remain two messages after refresh and late live delivery', async (t) => {
+    const metadata = { role: 'system', system_type: 'reminder', source: 'Ouroboros' };
+    const rows = [row('chat:300', 'Same words', metadata), row('chat:400', 'Same words', metadata)];
     const f = fixture(t, page(rows));
     await f.refresh();
     const bubbles = f.bubbles().filter((node) => node.innerHTML.includes('Same words'));
@@ -372,6 +440,28 @@ test('two physical rows with identical timestamp and body remain two messages ac
     assert.deepEqual(bubbles.map((node) => node.dataset.historyId), ['chat:300', 'chat:400']);
     await f.refresh(page(rows));
     assert.deepEqual(f.bubbles().filter((node) => node.innerHTML.includes('Same words')), bubbles);
+    assert.equal(bubbles[0].dataset.messageKey, bubbles[1].dataset.messageKey);
+    f.emit({ ...metadata, content: 'Same words', ts: TS });
+    assert.equal(f.bubbles().length, 2, 'the matching live copy cannot become a third physical row');
+    assert.equal(f.bubbles()[0], bubbles[0]);
+    assert.equal(f.bubbles()[1], bubbles[1]);
+    assert.deepEqual(bubbles.map(node => node.dataset.historyId), ['chat:300', 'chat:400']);
+});
+
+test('standalone notice joins still distinguish source, text and timestamp', async t => {
+    const message = row('chat:500', 'A scheduled notice', {
+        role: 'system', system_type: 'reminder', source: 'Ouroboros',
+    });
+    const f = fixture(t, page([message]));
+    await f.refresh();
+    const original = f.bubbles()[0];
+    const live = { role: message.role, content: message.text, ts: message.ts,
+        system_type: message.system_type, source: message.source };
+    for (const difference of [{ source: 'Calendar skill' }, { content: 'A different notice' },
+        { ts: '2026-09-12T12:00:01.000Z' }]) f.emit({ ...live, ...difference });
+    assert.equal(f.bubbles().length, 4, 'each changed field identifies another standalone message');
+    assert.equal(new Set(f.bubbles().map(node => node.dataset.messageKey)).size, 4);
+    assert.equal(f.bubbles().find(node => node.dataset.historyId === message.history_id), original);
 });
 
 test('one older-navigation action crosses a sparse page and displays the next physical row', async (t) => {
@@ -398,6 +488,9 @@ test('retained history nodes still dedupe after navigation exceeds the live key
     for (let i = 0; i < 16; i += 1) await f.clickOlder();
     const mounted = f.bubbles().length;
     assert.ok(mounted < 1000, 'distant pages actually evicted rather than all remaining protected');
+    f.emit({ role: 'assistant', content: 'Retained answer 2499', ts: TS });
+    assert.equal(f.bubbles().length, mounted, 'the mounted node suppresses a late frame after FIFO turnover');
+    assert.equal(f.bubbles().find(node => node.dataset.historyId === 'chat:2499'), recent);
     await f.refresh(pack(0));
     assert.deepEqual(f.bubbles().filter(node => node.dataset.historyId === 'chat:2499'), [recent]);
     assert.equal(f.bubbles().length, mounted, 'refresh adds no duplicate retained recent page');