From a70c0cb1c0d7dccb3fb53b81ff257bdf7df89709 Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 22 Oct 2026 10:01:54 +0200 Subject: [PATCH] A reply counts as read once delivered --- CHANGELOG.md | 4 ++++ README.md | 2 +- docs/concepts.md | 4 ++-- docs/reference/protocol.md | 2 +- src/broker.ts | 2 -- src/store.ts | 14 ++++++++++++-- test/broker.test.ts | 15 +++++++++++++++ test/core.test.ts | 17 +++++++++++++++++ test/store.test.ts | 39 ++++++++++++++++++++++++++++++++++++++ 9 files changed, 91 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c5c218..76e8976 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to murmur are documented here. The format follows [Keep a Ch ## [Unreleased] +### Changed + +- A reply counts as read once it is delivered. A delivery of a reply that reaches `delivered` by any path (push proof, adapter confirmation, hook pull, `murmur_read`, `murmur_wait`) now ends `acknowledged`, so it no longer shows as unread. Asks are unchanged: they stay unread until acked or answered. A reply that is answered later still reaches `replied`. On open, reply deliveries an older version left at `delivered` become `acknowledged` (once, idempotent). + ## [0.2.7] - 2026-10-21 ### Added diff --git a/README.md b/README.md index 72cf1ac..f48b2ed 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,7 @@ Every hook, MCP server and CLI command walks up its process ancestors and takes ## Delivery truth table -Delivery states, per recipient: `queued` (stored, nothing left the broker), `sent` (a transport accepted it), `delivered` (evidence from the receiving side), `acknowledged` (`murmur_ack`, or a reply handed over in a `murmur_wait` result: seen, answer later), `replied`. `sent` is never reported as `delivered`. +Delivery states, per recipient: `queued` (stored, nothing left the broker), `sent` (a transport accepted it), `delivered` (evidence from the receiving side), `acknowledged` (`murmur_ack`: seen, answer later; a reply ends here as soon as it is delivered, because a reply needs no answer and counts as read), `replied`. Unread means `queued`, `sent` or `delivered`. `sent` is never reported as `delivered`. | CLI | Live push into a running session | Waiting messages also arrive | Evidence for `delivered` | | --- | --- | --- | --- | diff --git a/docs/concepts.md b/docs/concepts.md index 0eeb248..236b9ca 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -36,14 +36,14 @@ Verdicts are `agree`, `disagree`, `info` and `choice`, with the body starting `A ## Delivery states -One row per (message, recipient): `queued -> sent -> delivered -> acknowledged`, with `replied` reachable from any state. +One row per (message, recipient): `queued -> sent -> delivered -> acknowledged`, with `replied` reachable from any state. Unread means `queued`, `sent` or `delivered`. | State | Meaning | | --- | --- | | `queued` | Stored; nothing left the broker. | | `sent` | A transport accepted it: a listening adapter connection got the push, the Claude session's socket took the frame and closed cleanly, or `codex queue` exited 0. Nothing more is claimed. | | `delivered` | Evidence from the receiving side: the tag `[murmur m ` found in the receiver's own transcript after the push (Claude, Codex, Hummin); the adapter confirming its session accepted the message (opencode, Hummin); the hook pull handing it to the CLI; `murmur_read`. | -| `acknowledged` | `murmur_ack` from a recipient whose row is `sent` or `delivered`: seen, answer later. | +| `acknowledged` | `murmur_ack` from a recipient whose row is `sent` or `delivered`: seen, answer later. A reply skips `delivered`: once delivered by any path it ends here, because a reply needs no answer and counts as read. | | `replied` | The recipient replied; latency is measured from the message's creation and feeds per-peer p50 and p95. | `sent` is never reported as `delivered`. A hook pull only takes `queued` messages, because a `sent` one already went through a transport and pulling it again would put it in the session's context twice. diff --git a/docs/reference/protocol.md b/docs/reference/protocol.md index 48abf20..0f29331 100644 --- a/docs/reference/protocol.md +++ b/docs/reference/protocol.md @@ -52,7 +52,7 @@ Pushed on a listening connection, no `id`: ## Types -- Delivery state: `queued`, `sent`, `delivered`, `acknowledged`, `replied`. +- Delivery state: `queued`, `sent`, `delivered`, `acknowledged`, `replied`. Unread counts `queued`, `sent` and `delivered`; a reply delivery ends `acknowledged` once delivered, so it is not unread. - Verdict: `agree`, `disagree`, `info`, `choice`. - Delivery mode: `push`, `notify`, `silent`. - Presence: `online`, `busy`, `idle`, `offline`, `presumed-offline`. diff --git a/src/broker.ts b/src/broker.ts index e690273..084b1dd 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -1182,8 +1182,6 @@ export class Broker { // A reply held for this wait never waited on a session that was down: // no queued_delivered notice, which would wake the replier for nothing. this.markDeliveredHop(row.id, delivery.recipient, held?.has(row.id) !== true); - // The wait result is the asker's reading: seen, not unread. - this.store.markAcked(row.id, delivery.recipient, Date.now()); replies.push({ id: row.id, seq: row.seq, sender: row.sender, verdict: row.verdict as Verdict | null, body: row.deliveredBody }); } } diff --git a/src/store.ts b/src/store.ts index 42489e7..8cd72a7 100644 --- a/src/store.ts +++ b/src/store.ts @@ -368,6 +368,11 @@ export class Store { db.exec('ALTER TABLE ' + table + ' ADD COLUMN ' + column + ' ' + ddl); } } + // Replies an older version left at delivered count as read. Idempotent. + db.exec( + "UPDATE deliveries SET state = 'acknowledged', acked_at = COALESCE(acked_at, delivered_at) " + + "WHERE state = 'delivered' AND message_id IN (SELECT id FROM messages WHERE is_reply = 1)", + ); }).immediate(); // The database holds every message and each session's native messaging // token: owner-only, like the token file. @@ -634,8 +639,13 @@ export class Store { } markDelivered(messageId: number, recipient: string, now: number): void { - this.stmt("UPDATE deliveries SET state = 'delivered', delivered_at = ? WHERE message_id = ? AND recipient = ?") - .run(now, messageId, recipient); + // The one rule: a reply needs no answer, so once delivered it is read. + this.stmt( + "UPDATE deliveries SET delivered_at = ?, " + + "state = CASE WHEN (SELECT is_reply FROM messages WHERE id = message_id) = 1 THEN 'acknowledged' ELSE 'delivered' END, " + + "acked_at = CASE WHEN (SELECT is_reply FROM messages WHERE id = message_id) = 1 THEN ? ELSE acked_at END " + + "WHERE message_id = ? AND recipient = ?", + ).run(now, now, messageId, recipient); } markAcked(messageId: number, recipient: string, now: number): void { diff --git a/test/broker.test.ts b/test/broker.test.ts index 996c1d0..f9b54e5 100644 --- a/test/broker.test.ts +++ b/test/broker.test.ts @@ -2613,6 +2613,21 @@ describe('session handler', () => { }); describe('delivered acks', () => { + it('ends a reply acknowledged when the adapter confirms it, and keeps an ask delivered', async () => { + fakeWaitWindow(); + const stack = await buildStack(); + const alice = await connectAdapter(stack, 'alice', 5060); + const bob = await connectAdapter(stack, 'bob', 5061); + const ask = await req(alice.client, { type: 'send', threadId: null, to: ['bob'], body: 'q' }); + await req(bob.client, { type: 'delivered', ids: [ask.messageId] }); + const reply = await req<{ messageId: number }>(bob.client, { type: 'reply', messageId: ask.messageId, body: 'a' }); + await req(alice.client, { type: 'delivered', ids: [reply.messageId] }); + const stateOf = (id: number, who: string): string | undefined => stack.store.getDeliveries(id).find((d) => d.recipient === who)?.state; + expect(stateOf(reply.messageId, 'alice')).toBe('acknowledged'); + expect(stateOf(ask.messageId, 'bob')).toBe('replied'); + expect(stack.store.deliveryStatesForRecipient('alice', ['queued', 'sent', 'delivered'])).toBe(0); + }); + it('moves queued and sent deliveries to delivered and ignores everything else', async () => { fakeWaitWindow(); const stack = await buildStack(); diff --git a/test/core.test.ts b/test/core.test.ts index c3bcfaf..5b52ec5 100644 --- a/test/core.test.ts +++ b/test/core.test.ts @@ -777,6 +777,23 @@ describe('groups', () => { }); describe('pendingPush', () => { + it('hands a reply over acknowledged and an ask delivered, so only the ask is unread', () => { + register('asker', 'worker'); + const ask = core.send('asker', { threadId: null, to: ['worker'], body: 'q' }); + core.pendingPush('worker', 10); + const { result } = core.reply('worker', { messageId: ask.result.messageId, body: 'a' }); + core.pendingPush('asker', 10); + const stateOf = (id: number, who: string): string | undefined => store.getDeliveries(id).find((d) => d.recipient === who)?.state; + expect(stateOf(result.messageId, 'asker')).toBe('acknowledged'); + expect(stateOf(ask.result.messageId, 'worker')).toBe('replied'); + const peers = core.status(() => ({ online: false })).peers; + expect(peers.find((p) => p.handle === 'asker')?.unread).toBe(0); + const second = core.send('asker', { threadId: null, to: ['worker'], body: 'q2' }); + core.pendingPush('worker', 10); + expect(stateOf(second.result.messageId, 'worker')).toBe('delivered'); + expect(core.status(() => ({ online: false })).peers.find((p) => p.handle === 'worker')?.unread).toBe(1); + }); + it('hands out queued pushes newest first and marks them delivered', () => { register('sender', 'receiver'); const m1 = core.send('sender', { threadId: null, to: ['receiver'], body: 'first' }); diff --git a/test/store.test.ts b/test/store.test.ts index a886fb7..c16bd87 100644 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -243,6 +243,45 @@ describe('Store', () => { expect(store.getDeliveries(424242)).toEqual([]); }); + it('ends a delivered reply acknowledged at the delivery time and lets it reach replied', () => { + const tid = store.createThread('s', 1); + const ask = appendMessage(tid, {}); + const reply = appendMessage(tid, { isReply: true, inReplyTo: ask.id }); + store.addDeliveries(ask.id, ['r'], 1); + store.addDeliveries(reply.id, ['r'], 1); + store.markDelivered(ask.id, 'r', 10); + store.markDelivered(reply.id, 'r', 10); + expect(store.getDeliveries(ask.id)[0]).toMatchObject({ state: 'delivered', deliveredAt: 10, ackedAt: null }); + expect(store.getDeliveries(reply.id)[0]).toMatchObject({ state: 'acknowledged', deliveredAt: 10, ackedAt: 10 }); + store.markReplied(reply.id, 'r', 12, 5); + expect(store.getDeliveries(reply.id)[0]).toMatchObject({ state: 'replied', repliedAt: 12 }); + }); + + it('turns reply rows left delivered by an older version into acknowledged once at open', () => { + const tid = store.createThread('s', 1); + const ask = appendMessage(tid, {}); + const reply = appendMessage(tid, { isReply: true, inReplyTo: ask.id }); + store.addDeliveries(ask.id, ['r'], 1); + store.addDeliveries(reply.id, ['r'], 1); + store.close(); + const raw = new Database(dbPath); + raw.exec( + "UPDATE deliveries SET state = 'delivered', delivered_at = 7 WHERE state = 'queued'; " + + 'UPDATE deliveries SET acked_at = 3 WHERE message_id = ' + String(reply.id) + ';', + ); + raw.close(); + store = Store.open(dbPath); + expect(store.getDeliveries(ask.id)[0]).toMatchObject({ state: 'delivered', ackedAt: null }); + // acked_at keeps an existing value, and the row stays acknowledged. + expect(store.getDeliveries(reply.id)[0]).toMatchObject({ state: 'acknowledged', ackedAt: 3 }); + const raw2 = new Database(dbPath); + raw2.exec("UPDATE deliveries SET state = 'delivered', acked_at = NULL WHERE message_id = " + String(reply.id)); + raw2.close(); + store.close(); + store = Store.open(dbPath); + expect(store.getDeliveries(reply.id)[0]).toMatchObject({ state: 'acknowledged', ackedAt: 7 }); + }); + it('marks delivered, acked and replied with timestamps', () => { const tid = store.createThread('s', 1); const { id } = appendMessage(tid, {});