From e25da3fd3df5969f5460b4c54bfd0caee320aab2 Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 21 Oct 2026 11:03:26 +0200 Subject: [PATCH] Timed-out waits tell the asker who is silent and to ask the user --- CHANGELOG.md | 4 +++ docs/consult.md | 2 +- docs/reference/protocol.md | 2 +- src/broker.ts | 29 +++++++++++++++++++++- src/protocol.ts | 2 ++ test/broker.test.ts | 50 ++++++++++++++++++++++++++++++++++++++ test/mcp.test.ts | 3 +++ 7 files changed, 89 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ba6e84..9e27f2f 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] +### Added + +- A timed-out `murmur_wait` (or `murmur_send` with `wait`) now carries a `note` when a recipient of the latest ask has not replied. It names who has the message but has not answered and who has not confirmed receipt (queued or sent), and tells the asker not to start work that could overlap theirs: ask the user, or wait again. It is absent when the wait condition was met and on a `timeoutMs: 0` check, which did not wait. + ## [0.2.5] - 2026-10-21 ### Added diff --git a/docs/consult.md b/docs/consult.md index 213f37b..9994921 100644 --- a/docs/consult.md +++ b/docs/consult.md @@ -12,7 +12,7 @@ murmur send claude-api,codex-web,@reviewers "Which of you owns the migration?" - ## Wait for the answers -`murmur_wait` returns for the thread's latest ask each recipient's state, verdict and reply latency (plus their historical p50 and p95), when everyone (`all_replied`) or anyone (`any_replied`) replied, or at the timeout, plus the text of the replies addressed to you. The default is 120 s and the most is 300 s; `timeoutMs: 0` returns now. See [MCP tools](reference/mcp.md). +`murmur_wait` returns for the thread's latest ask each recipient's state, verdict and reply latency (plus their historical p50 and p95), when everyone (`all_replied`) or anyone (`any_replied`) replied, or at the timeout, plus the text of the replies addressed to you. The default is 120 s and the most is 300 s; `timeoutMs: 0` returns now. A wait that times out with a recipient still unreplied (not a `timeoutMs: 0` check) adds a `note` naming who has the message but has not answered and who has not confirmed receipt, and telling you to ask the user (or wait again) rather than start work that could overlap theirs. See [MCP tools](reference/mcp.md). ## Read the thread diff --git a/docs/reference/protocol.md b/docs/reference/protocol.md index 57edd8c..0773069 100644 --- a/docs/reference/protocol.md +++ b/docs/reference/protocol.md @@ -24,7 +24,7 @@ Every client message carries `v` (protocol version) and `id` (a per-connection m | `inbox` | `unreadOnly` | `{handle, unreadTotal, threads}` | | `thread` | `threadId`, `sinceSeq` (or null), `full`, `messageId?` (only that message and its replies, full bodies; `bad_request` if not in the thread) | `{threadId, latestSeq, messages}` | | `collect` | `threadId` | `{threadId, askMessageId, recipients, verdictTally}` | -| `wait` | `threadId`, `until` (`all_replied`\|`any_replied`), `timeoutMs` | `{threadId, conditionMet, timedOut, matrix}` | +| `wait` | `threadId`, `until` (`all_replied`\|`any_replied`), `timeoutMs` | `{threadId, conditionMet, timedOut, matrix, replies, note?}` (`note` is set only on a timeout with a recipient still unreplied, never for `timeoutMs` 0: who has the message but is silent, whose receipt is unproven, and an instruction not to start overlapping work but to ask the user or wait again) | | `groups` | `action` (`create`\|`list`\|`add`\|`remove`\|`delete`), `name?`, `members?` | group records | | `status` | none | `{now, peers}` | | `prune` | none | `{removed}` (handles of the deleted peers) | diff --git a/src/broker.ts b/src/broker.ts index 511bc4f..350c482 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -103,6 +103,8 @@ interface Waiter { timer: NodeJS.Timeout; // Ids of replies held back for this wait; the handover returns them even if the latest ask moved on. held: Set; + // timeoutMs 0 asks for the state now; it never waited, so its result carries no timeout note. + snapshot: boolean; } interface NotifyEntry { @@ -1123,6 +1125,7 @@ export class Broker { threadId, until, held: new Set(), + snapshot: timeoutMs === 0, timer: setTimeout(() => this.resolveTimedOut(waiter), timeoutMs), }; conn.waiters.add(waiter); @@ -1217,11 +1220,14 @@ export class Broker { return; } try { + const matrix = this.matrixFor(waiter.threadId); + const note = waiter.snapshot ? null : timeoutNote(matrix); this.replyOk(waiter.conn.socket, waiter.id, { threadId: waiter.threadId, conditionMet: false, timedOut: true, - matrix: this.matrixFor(waiter.threadId), + matrix, + ...(note === null ? {} : { note }), replies: this.takeReplies(waiter.threadId, waiter.conn.handle, waiter.held), }); } catch { @@ -1344,3 +1350,24 @@ async function clearStaleSocket(socketPath: string): Promise { } unlinkSync(socketPath); } + +// What a timed-out asker is told: who has the message but is silent, whose receipt +// is unproven (queued or sent), and not to start work that could overlap theirs. +// Null when everyone replied. +function timeoutNote(matrix: RecipientMatrix[]): string | null { + const names = (states: string[]): string[] => matrix.filter((m) => states.includes(m.state)).map((m) => m.recipient); + const silent = names(['delivered', 'acknowledged']); + const unreached = names(['queued', 'sent']); + const clauses: string[] = []; + if (silent.length > 0) { + const v = silent.length > 1 ? 'have' : 'has'; + clauses.push(`${silent.join(', ')} ${v} the message but ${v} not replied`); + } + if (unreached.length > 0) { + clauses.push(`${unreached.join(', ')} ${unreached.length > 1 ? 'have' : 'has'} not confirmed receipt`); + } + if (clauses.length === 0) { + return null; + } + return `timed out: ${clauses.join('; ')}. Do not start work that could overlap theirs: ask the user, or murmur_wait again.`; +} diff --git a/src/protocol.ts b/src/protocol.ts index 057d302..c129f49 100644 --- a/src/protocol.ts +++ b/src/protocol.ts @@ -197,6 +197,8 @@ export interface WaitResult { timedOut: boolean; matrix: RecipientMatrix[]; replies: WaitReply[]; + // Only on a timed-out wait with a recipient still unreplied: who to wait for and what not to do. + note?: string; } export interface PushPayload { diff --git a/test/broker.test.ts b/test/broker.test.ts index 4faa573..ebc2002 100644 --- a/test/broker.test.ts +++ b/test/broker.test.ts @@ -1226,6 +1226,56 @@ describe('wait', () => { expect(result.matrix[0]).toMatchObject({ recipient: 'bob', state: 'queued' }); }); + it('tells a timed-out asker who has the message but is silent and whose receipt is unproven', async () => { + const stack = await buildStack(); + const alice = await connectPeer(stack, { handle: 'alice' }); + const bob = await connectPeer(stack, { handle: 'bob' }); + const carol = await connectPeer(stack, { handle: 'carol' }); + await connectPeer(stack, { handle: 'dave' }); + const ask = await req(alice.client, { type: 'send', threadId: null, to: ['bob', 'carol', 'dave'], body: 'slow' }); + for (const peer of [bob, carol]) { + await req(peer.client, { type: 'read', ids: [ask.messageId] }); + await req(peer.client, { type: 'ack', ids: [ask.messageId] }); + } + const result = await req(alice.client, { type: 'wait', threadId: ask.threadId, until: 'all_replied', timeoutMs: 100 }); + expect(result.note).toBe( + 'timed out: bob, carol have the message but have not replied; dave has not confirmed receipt. Do not start work that could overlap theirs: ask the user, or murmur_wait again.', + ); + }); + + it('words the note for a single silent recipient', async () => { + const stack = await buildStack(); + const alice = await connectPeer(stack, { handle: 'alice' }); + const bob = await connectPeer(stack, { handle: 'bob' }); + const ask = await req(alice.client, { type: 'send', threadId: null, to: ['bob'], body: 'slow' }); + await req(bob.client, { type: 'read', ids: [ask.messageId] }); + const result = await req(alice.client, { type: 'wait', threadId: ask.threadId, until: 'all_replied', timeoutMs: 100 }); + expect(result.note).toMatch(/^timed out: bob has the message but has not replied\. Do not start/); + }); + + it('omits the have-the-message clause when every silent recipient is still queued', async () => { + const stack = await buildStack(); + const alice = await connectPeer(stack, { handle: 'alice' }); + await connectPeer(stack, { handle: 'bob' }); + await connectPeer(stack, { handle: 'carol' }); + const ask = await req(alice.client, { type: 'send', threadId: null, to: ['bob', 'carol'], body: 'slow' }); + const result = await req(alice.client, { type: 'wait', threadId: ask.threadId, until: 'all_replied', timeoutMs: 100 }); + expect(result.note).toBe( + 'timed out: bob, carol have not confirmed receipt. Do not start work that could overlap theirs: ask the user, or murmur_wait again.', + ); + }); + + it('sets no note when the wait condition is met', async () => { + const stack = await buildStack(); + const alice = await connectPeer(stack, { handle: 'alice' }); + const bob = await connectPeer(stack, { handle: 'bob' }); + const ask = await req(alice.client, { type: 'send', threadId: null, to: ['bob'], body: 'quick' }); + await req(bob.client, { type: 'reply', messageId: ask.messageId, body: 'ok' }); + const result = await req(alice.client, { type: 'wait', threadId: ask.threadId, until: 'all_replied', timeoutMs: 100 }); + expect(result.conditionMet).toBe(true); + expect('note' in result).toBe(false); + }); + it('times out on a thread without an ask and returns an empty matrix', async () => { const stack = await buildStack(); const alice = await connectPeer(stack, { handle: 'alice' }); diff --git a/test/mcp.test.ts b/test/mcp.test.ts index b6f755e..2b18c23 100644 --- a/test/mcp.test.ts +++ b/test/mcp.test.ts @@ -292,11 +292,14 @@ describe('SDK conformance over InMemoryTransport', () => { const instant = await call(sdk, 'murmur_wait', { threadId: sent.threadId, until: 'all_replied', timeoutMs: 0 }); expect(Date.now() - now).toBeLessThan(500); expect(instant).toMatchObject({ threadId: sent.threadId, conditionMet: false, timedOut: true }); + // A snapshot did not wait, so it carries no advice about waiting. + expect('note' in instant).toBe(false); expect(instant.matrix[0]).toMatchObject({ recipient: 'peerone', verdict: null }); const started = Date.now(); const waited = await call(sdk, 'murmur_wait', { threadId: sent.threadId, until: 'all_replied', timeoutMs: 1000 }); expect(Date.now() - started).toBeGreaterThanOrEqual(900); expect(waited).toMatchObject({ conditionMet: false, timedOut: true }); + expect(waited.note).toContain('Do not start work that could overlap theirs: ask the user, or murmur_wait again.'); }); it('murmur_send with wait sends and returns the wait result in one call, resolving a display name', async () => {