Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve message occurrence contracts

Browse Source
This commit is contained in:
_Kerman
2026-07-28 14:44:15 +08:00
parent 384535f0ae
commit 0225d598e3
7 changed files with 68 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -1042,8 +1042,8 @@ export const EVENT_API: readonly EventApiEntry[] = [
name: 'agent/inbox/discard',
mode: 'emit',
signature: '\'agent/inbox/discard\'(this: Scoped<Agent>, 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',

2
packages/core/agent/src/types.ts
View File

@@ -255,7 +255,7 @@ declare module 'cordis' {
'agent/inbox/dequeue'(this: Scoped<Agent>, 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.

12
packages/core/session/src/surface.ts
View File

@@ -224,8 +224,16 @@ function assertToolResultRewrite(
}
const originalRest = { ...original.data } as Record<string, unknown>
const replacementRest = { ...event.data } as Record<string, unknown>
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')
}

24
packages/core/session/tests/surface.spec.ts
View File

@@ -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<SessionEvent, { type: 'tool/result' }>['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)

22
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -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<SessionId, Map<MessageId, { message: UserMessage; steering: boolean }>>()
const queuedMirror = new Map<SessionId, { message: UserMessage; steering: boolean }[]>()
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<MessageId, { message: UserMessage; steering: boolean }>()
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,

18
packages/host/apiproxy/tests/api-proxy-commands.spec.ts
View File

@@ -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<MuxFrame>(
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)