From 0225d598e38c4c03f2bdf02868e0138948fb5223 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Tue, 28 Jul 2026 14:44:15 +0800 Subject: [PATCH] fix: preserve message occurrence contracts --- docs/cordis-catalog/events.md | 4 ++-- .../cordis/tool-cordis/src/api-catalog.ts | 4 ++-- packages/core/agent/src/types.ts | 2 +- packages/core/session/src/surface.ts | 12 ++++++++-- packages/core/session/tests/surface.spec.ts | 24 +++++++++++++++++++ packages/host/apiproxy/src/api-proxy.ts | 22 ++++++++--------- .../apiproxy/tests/api-proxy-commands.spec.ts | 18 ++++++++++++++ 7 files changed, 68 insertions(+), 18 deletions(-) diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 6163917869..85f0761f59 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -121,12 +121,12 @@ Source: [`packages/core/agent/src/types.ts:255`](../../packages/core/agent/src/t ### `agent/inbox/discard` — emit -Pending inbox items were dropped without delivering them, so every enqueued id receives exactly one terminal `agent/inbox/dequeue` OR `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal, emits this after `agent/cancel-requested` when applicable and before aborting the active work. Fires once per drop with every dropped item. +Pending inbox items were dropped without delivering them, so every enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal, emits this after `agent/cancel-requested` when applicable and before aborting the active work. Fires once per drop with every dropped item. ```ts cordis-catalog /** * Pending inbox items were dropped without delivering them, so every - * enqueued id receives exactly one terminal `agent/inbox/dequeue` OR + * enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR * `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal, * emits this after `agent/cancel-requested` when applicable and before * aborting the active work. Fires once per drop with every dropped item. diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 40782c39d1..9d261107fd 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -1042,8 +1042,8 @@ export const EVENT_API: readonly EventApiEntry[] = [ name: 'agent/inbox/discard', mode: 'emit', signature: '\'agent/inbox/discard\'(this: Scoped, agent: Agent, messages: UserMessage[]): void', - jsDoc: '/**\n * Pending inbox items were dropped without delivering them, so every\n * enqueued id receives exactly one terminal `agent/inbox/dequeue` OR\n * `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal,\n * emits this after `agent/cancel-requested` when applicable and before\n * aborting the active work. Fires once per drop with every dropped item.\n * @param agent - the agent whose inbox items were dropped.\n * @param messages - the discarded messages in FIFO order (queued then steering); never empty.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */', - summary: 'Pending inbox items were dropped without delivering them, so every enqueued id receives exactly one terminal `agent/inbox/dequeue` OR `agent/inbox/discard`.', + jsDoc: '/**\n * Pending inbox items were dropped without delivering them, so every\n * enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR\n * `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal,\n * emits this after `agent/cancel-requested` when applicable and before\n * aborting the active work. Fires once per drop with every dropped item.\n * @param agent - the agent whose inbox items were dropped.\n * @param messages - the discarded messages in FIFO order (queued then steering); never empty.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */', + summary: 'Pending inbox items were dropped without delivering them, so every enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR `agent/inbox/discard`.', }, { name: 'agent/inbox/enqueue', diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 924fc464c3..46ecc7a17f 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -255,7 +255,7 @@ declare module 'cordis' { 'agent/inbox/dequeue'(this: Scoped, agent: Agent, message: UserMessage): void /** * Pending inbox items were dropped without delivering them, so every - * enqueued id receives exactly one terminal `agent/inbox/dequeue` OR + * enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR * `agent/inbox/discard`. `cancel()` without `keepInbox`, including disposal, * emits this after `agent/cancel-requested` when applicable and before * aborting the active work. Fires once per drop with every dropped item. diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index 467273d544..9677c5ec2c 100644 --- a/packages/core/session/src/surface.ts +++ b/packages/core/session/src/surface.ts @@ -224,8 +224,16 @@ function assertToolResultRewrite( } const originalRest = { ...original.data } as Record const replacementRest = { ...event.data } as Record - originalRest['message'] = { ...original.data.message, content: null } - replacementRest['message'] = { ...event.data.message, content: null } + const originalResult = original.data.message.content[0] + const replacementResult = event.data.message.content[0] + originalRest['message'] = { + ...original.data.message, + content: [{ ...originalResult, content: null }], + } + replacementRest['message'] = { + ...event.data.message, + content: [{ ...replacementResult, content: null }], + } if (!isDeepEqualJson(originalRest, replacementRest)) { throw new Error('tool/result surface replacement may change only content') } diff --git a/packages/core/session/tests/surface.spec.ts b/packages/core/session/tests/surface.spec.ts index bbb16b4106..4078f526e6 100644 --- a/packages/core/session/tests/surface.spec.ts +++ b/packages/core/session/tests/surface.spec.ts @@ -171,6 +171,30 @@ describe('foldSurface tool-result rewrites', () => { expect(() => foldSurface(events)).toThrow(/may change only content/) }) + it.each([ + ['toolCallId', { toolCallId: CallId('changed') }], + ['isError', { isError: true }], + ] as const)('rejects a replacement that changes the result block %s', (_field, patch) => { + const original = toolResultEvent(0, 'original') + const data = original.data as Extract['data'] + const result = data.message.content[0] + const replacement = { + ...original, + seq: 1, + time: 1, + data: { + ...data, + message: freezeMessage({ + ...data.message, + content: [{ ...result, ...patch }] as [typeof result], + }), + }, + surfaceOp: { op: 'replace', start: 0, end: 0 }, + sourceEventSeqs: [0], + } as SessionEvent + expect(() => foldSurface([original, replacement])).toThrow(/may change only content/) + }) + it('compares array-valued rest fields structurally (meta arrays: equal accepted, drifted rejected)', () => { const withMeta = (seq: number, meta: unknown, surfaceOp: SurfaceEvent['surfaceOp'] = 'append', sourceEventSeqs?: number[]): SessionEvent => { const event = toolResultEvent(seq, 'c-meta', surfaceOp, sourceEventSeqs) diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 09d683b7a1..aa1478c938 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -421,29 +421,29 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro } /** - * Per-session inbox mirror serving the mux-open queue snapshot (the same - * refresh-recovery baseline as pending questions). Keyed by the stable - * MessageId: every enqueued id receives exactly one terminal - * `agent/inbox/dequeue` OR `agent/inbox/discard` (the inbox contract), so - * the mirror needs no consumption heuristics or sweeps beyond disposal. + * Per-session inbox occurrence mirror serving the mux-open queue snapshot + * (the same refresh-recovery baseline as pending questions). Each terminal + * inbox event retires one matching occurrence, so repeated sends of the same + * identified message remain visible until every occurrence is claimed. */ - const queuedMirror = new Map>() + const queuedMirror = new Map() ctx.effect(() => { const retire = (agent: Agent, id: MessageId): void => { const entries = queuedMirror.get(agent.id) if (entries === undefined) return - entries.delete(id) - if (entries.size === 0) queuedMirror.delete(agent.id) + const index = entries.findIndex(entry => entry.message.id === id) + if (index !== -1) entries.splice(index, 1) + if (entries.length === 0) queuedMirror.delete(agent.id) } const disposers = [ ctx.on('agent/inbox/enqueue', (agent: Agent, message: UserMessage, placement) => { let entries = queuedMirror.get(agent.id) if (entries === undefined) { - entries = new Map() + entries = [] queuedMirror.set(agent.id, entries) } const steering = placement === 'steering' - entries.set(message.id, { message, steering }) + entries.push({ message, steering }) broadcast({ type: 'session/queued', sessionId: agent.id, @@ -1119,7 +1119,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro // in arrival order per session; a reconnecting client rebuilds its // queue view from these alone. for (const [sessionId, entries] of queuedMirror) { - for (const entry of entries.values()) { + for (const entry of entries) { queue.push(frame({ type: 'session/queued', sessionId, diff --git a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts index 20f37c9c13..890d7aade2 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -293,6 +293,24 @@ describe('session/queued frames', () => { expect(frames.filter(f => f.type === 'session/queued')).toHaveLength(0) }) + it('retires repeated sends of one message identity by occurrence', async () => { + const ctx = await harness() + const api = createApiProxy(ctx, DEFAULTS) + const agent = stubAgent(ctx) + const repeated = inboxMessage('m-repeat', 'same prompt') + ctx.emit('agent/inbox/enqueue', agent, repeated, 'queued') + ctx.emit('agent/inbox/enqueue', agent, repeated, 'queued') + ctx.emit('agent/inbox/dequeue', agent, inboxMessage('unknown', 'not queued')) + ctx.emit('agent/inbox/dequeue', agent, repeated) + + const abort = new AbortController() + const frames = await collect( + api.events.mux({ rpcId: RpcId('t-mux-repeat'), payload: {} }, abort.signal), 2, abort) + expect(frames.filter(f => f.type === 'session/queued')).toEqual([ + { type: 'session/queued', sessionId: agent.id, message: repeated, steering: false }, + ]) + }) + it('retires mirror entries on a batch discard (cancel path)', async () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS)