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

fix(agent-loop): preserve deferred message boundaries

Browse Source
This commit is contained in:
_Kerman
2026-07-27 18:58:27 +08:00
parent 7c2cf5f0b3
commit 9152858fd7
17 changed files with 253 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
packages/core/agent-loop/src/agent.ts
View File

@@ -52,6 +52,8 @@ export class ReactLoopAgent implements Agent {
/** Whether observers see a running interval; consecutive turns share it. */
private busy = false
/** Whether an idle waking send has deferred driver admission. */
private wakeScheduled = false
/** Whether next-step input belongs to the current admission or open turn. */
acceptsNextStep = false
/** Abort owner for the current admission or turn. */
@@ -95,7 +97,7 @@ export class ReactLoopAgent implements Agent {
input: UserMessageData,
options: SendOptions,
): AgentMessageId {
const { content, source } = input
const { content, source } = deepFreeze(structuredClone(input))
const { target, wakeup } = options
const id = AgentMessageId(randomUUID())
if (target === 'next-step' && !wakeup) {
@@ -113,13 +115,17 @@ export class ReactLoopAgent implements Agent {
content,
source,
}
deepFreeze(message)
if (placement === 'steering') {
this.outbox.push(message)
} else {
this.queued.push({ message, wakeup })
}
// Preserve the routing decision for every send in this synchronous caller
// stack, while installing quiescence ownership before enqueue observers
// can cancel or dispose.
if (placement === 'queued' && wakeup) this.scheduleKick()
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message, placement)
if (placement === 'queued' && wakeup) this.kick()
return id
}
@@ -200,11 +206,33 @@ export class ReactLoopAgent implements Agent {
// but the waiter must not gamble quiescence on that: a future escape
// still counts as settled activity.
/* v8 ignore next 3 -- the catch arm backstops rejection paths that are all currently contained */
while (this.abort !== undefined || this.queued.some(item => item.wakeup)) {
while (this.wakeScheduled || this.abort !== undefined || this.queued.some(item => item.wakeup)) {
await this.done.catch(() => undefined)
}
}
/** Defer idle admission while keeping {@link done} as its quiescence owner. */
private scheduleKick(): void {
if (this.abort !== undefined || this.wakeScheduled) return
this.wakeScheduled = true
const pending = Promise.withResolvers<void>()
const scheduled = pending.promise
queueMicrotask(() => {
this.wakeScheduled = false
this.kick()
const activity = this.done
if (activity === scheduled) {
pending.resolve()
} else {
void activity.then(
() => { pending.resolve() },
() => { pending.resolve() },
)
}
})
this.done = scheduled
}
/** Claim and admit the next queued prompt, then start its turn. */
private kick(): void {
if (this.abort !== undefined || !this.queued.some(item => item.wakeup)) return
@@ -212,6 +240,7 @@ export class ReactLoopAgent implements Agent {
// assertion expresses that invariant.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const { message } = this.queued.shift()!
const inheritedOutboxLength = this.outbox.length
const admission = new AbortController()
this.abort = admission
@@ -274,7 +303,7 @@ export class ReactLoopAgent implements Agent {
this.continueOrIdle()
return
}
await this.run(trigger, admitted)
await this.run(trigger, admitted, inheritedOutboxLength)
})
// Published only after the abort owner and pending done are installed: a
// dequeue listener that cancels or disposes must find live cancellation
@@ -286,7 +315,11 @@ export class ReactLoopAgent implements Agent {
* Run one turn and any request-error retry. `admitted` input enters the log
* only after `turn/start` commits; until then it has no owner state to unwind.
*/
private async run(trigger: TurnTrigger, admitted: UserMessageData[] = []): Promise<void> {
private async run(
trigger: TurnTrigger,
admitted: UserMessageData[] = [],
inheritedOutboxLength = 0,
): Promise<void> {
// Both entries hold the invariant: kick() clears the admission slot before
// awaiting run(), and retry() returns early whenever a slot owner exists.
/* v8 ignore next -- unreachable guard: every caller clears or checks the abort slot first */
@@ -316,6 +349,9 @@ export class ReactLoopAgent implements Agent {
this.turnOpen = true
opened = true
this.lastTurn = turn
// Context or steering retained by an earlier rejected admission happened
// before this prompt and must occupy the same order in durable history.
this.drainOutbox(turn, inheritedOutboxLength)
for (const input of admitted) {
this.session.append('user/message', input, { surfaceOp: 'append' })
}
@@ -611,9 +647,9 @@ export class ReactLoopAgent implements Agent {
}
/** Commit the outbox and report whether it contained steering. */
private drainOutbox(turn: number): boolean {
private drainOutbox(turn: number, limit = this.outbox.length): boolean {
let steered = false
for (const message of this.outbox.splice(0)) {
for (const message of this.outbox.splice(0, limit)) {
if ('id' in message) {
steered = true
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)

2
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -928,7 +928,7 @@ describe('turn and step boundary recovery', () => {
})
send(agent, 'go')
await driverDone(agent)
await agent.whenIdle()
const e = [...agent.session.events]
// Balanced: one turn/start, one turn/end carrying disposed (NOT error).

28
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -348,7 +348,7 @@ describe('stream failure edges', () => {
})
describe('post-turn continuation edges', () => {
it('an agent/settled listener that enqueues a waking prompt preempts continueOrIdle', async () => {
it('an agent/settled listener that starts a retry preempts continueOrIdle', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('settled-preempt'), { provider: 'mock', model: 'mock' })
@@ -357,9 +357,9 @@ describe('post-turn continuation edges', () => {
if (subject !== agent || injected) return
expect(subject.status).toBe('running')
injected = true
// kick() installs the next admission synchronously, so the following
// continueOrIdle() sees an abort owner and yields to it.
send(agent, 'follow-up from settled listener')
// retry() installs the next run synchronously, so the following
// continueOrIdle() sees its abort owner and yields to it.
subject.retry()
})
send(agent, 'go')
@@ -524,6 +524,26 @@ describe('unrenderable failure settlement', () => {
})
describe('driver bookkeeping edges', () => {
it('a deferred wake settles when replacement activity rejects', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('rejected-deferred-wake'), {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/inbox/enqueue', (subject) => {
if (subject !== agent) return
subject.cancel({ kind: 'user' })
const mutable = subject as Agent & { done: Promise<void> }
mutable.done = Promise.reject(new Error('replacement rejected'))
})
send(agent, 'cancel before wake')
await expect(agent.whenIdle()).resolves.toBeUndefined()
expect(agent.session.events).toEqual([])
})
it('a whenIdle waiter survives a rejected driver promise', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)

123
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -1,10 +1,21 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
import SessionStore, {
SessionId,
type SessionEvent,
type TurnEndReason,
type UserMessageData,
} from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent, type InboxPlacement, type PromptDecision, type SessionStartSource } from '@deepseek-ai/dsh-agent'
import AgentRegistry, {
type Agent,
type AgentMessage,
type InboxPlacement,
type PromptDecision,
type SessionStartSource,
} from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
@@ -69,6 +80,57 @@ describe('agent/prompt-submit', () => {
expect(userMsg?.type === 'user/message' && userMsg.data.content).toEqual([{ type: 'text', text: 'hello' }])
})
it('snapshots and freezes input before publishing or awaiting admission', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('owned-input'), { provider: 'mock', model: 'mock' })
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PromptDecision>()
const observed: AgentMessage[] = []
ctx.on('agent/inbox/enqueue', (subject, message) => {
if (subject !== agent) return
expect(Object.isFrozen(message)).toBe(true)
expect(Object.isFrozen(message.content)).toBe(true)
expect(Object.isFrozen(message.content[0])).toBe(true)
expect(Object.isFrozen(message.source)).toBe(true)
expect(() => {
const block = message.content[0]
if (block?.type === 'text') block.text = 'listener mutation'
}).toThrow()
})
ctx.on('agent/inbox/enqueue', (subject, message) => {
if (subject === agent) observed.push(message)
})
ctx.on('agent/prompt-submit', async () => {
entered.resolve(undefined)
return decision.promise
})
const input: UserMessageData = {
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
}
const idle = waitForIdle(ctx, agent)
agent.followup(input)
await entered.promise
const block = input.content[0]
if (block?.type === 'text') block.text = 'caller mutation'
if (input.source.kind === 'plugin') input.source.plugin = 'caller mutation'
decision.resolve({ kind: 'allow' })
await idle
expect(observed).toHaveLength(1)
expect(observed[0]).toMatchObject({
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
})
const userMsg = events(agent).find(event => event.type === 'user/message')
expect(userMsg?.type === 'user/message' && userMsg.data).toEqual({
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
})
})
it('allow with content REWRITES the prompt before it is recorded', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
@@ -257,6 +319,54 @@ describe('agent/prompt-submit', () => {
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering')
})
it('orders rejected-admission outbox input before a later admitted prompt', async () => {
const adapter = new MockAdapter([textResponse('continued')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('rejected-admission-order'), {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next) => {
const decision = await next()
return content.some(block => block.type === 'text' && block.text === 'blocked prompt')
? { kind: 'block', reason: 'policy' }
: decision
})
ctx.on('agent/prompt-submit', async (subject, content, _source, _signal, next) => {
if (content.some(block => block.type === 'text' && block.text === 'blocked prompt')) {
subject.inject({
content: [{ type: 'text', text: 'earlier state change' }],
source: { kind: 'plugin', plugin: 'test' },
})
subject.steer({
content: [{ type: 'text', text: 'earlier steering' }],
source: { kind: 'user' },
})
}
return next()
})
const idle = waitForIdle(ctx, agent)
send(agent, 'blocked prompt')
send(agent, 'later prompt')
await idle
const staged = events(agent).filter(event =>
event.type === 'turn/start' || event.type === 'user/message' || event.type === 'steering/message')
expect(staged.map(event => event.type)).toEqual([
'turn/start',
'user/message',
'steering/message',
'user/message',
])
expect(staged[1]?.type === 'user/message' && staged[1].data.content)
.toEqual([{ type: 'text', text: 'earlier state change' }])
expect(staged[2]?.type === 'steering/message' && staged[2].data.content)
.toEqual([{ type: 'text', text: 'earlier steering' }])
expect(staged[3]?.type === 'user/message' && staged[3].data.content)
.toEqual([{ type: 'text', text: 'later prompt' }])
})
it('commits context-only injection when admission closes without a turn', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
@@ -296,13 +406,20 @@ describe('agent/prompt-submit', () => {
vi.spyOn(agent.session, 'append').mockImplementationOnce(() => {
throw new Error('append unavailable')
})
ctx.on('agent/prompt-submit', async () => ({ kind: 'block', reason: 'policy' }))
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PromptDecision>()
ctx.on('agent/prompt-submit', async () => {
entered.resolve(undefined)
return decision.promise
})
agent.followup({ content: [{ type: 'text', text: 'blocked prompt' }], source: { kind: 'user' } })
await entered.promise
agent.inject({
content: [{ type: 'text', text: 'retained context' }],
source: { kind: 'plugin', plugin: 'test' },
})
decision.resolve({ kind: 'block', reason: 'policy' })
await agent.whenIdle()
expect(events(agent)).toEqual([])

23
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -293,8 +293,8 @@ describe('agent loop', () => {
expect(flat).toContain('change of plans')
})
it('same-tick steering joins the prompt already in admission', async () => {
const adapter = new MockAdapter([textResponse('combined')])
it('same-tick idle steering preserves one turn per send', async () => {
const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -303,21 +303,18 @@ describe('agent loop', () => {
agent.steer({ content: [{ type: 'text', text: 'second idle steer' }], source: { kind: 'user' } })
await idle
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
expect(agent.session.events
.filter(event => event.type === 'user/message')
.map(event => event.data.content)).toEqual([
[{ type: 'text', text: 'first idle steer' }],
])
expect(agent.session.events
.filter(event => event.type === 'steering/message')
.map(event => event.data.content)).toEqual([
[{ type: 'text', text: 'second idle steer' }],
])
expect(adapter.requests).toHaveLength(1)
const request = JSON.stringify(adapter.requests[0]?.messages)
expect(request).toContain('first idle steer')
expect(request).toContain('second idle steer')
expect(agent.session.events.filter(event => event.type === 'steering/message')).toEqual([])
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('first idle steer')
expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('second idle steer')
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('second idle steer')
})
it('keeps steering staged after a failed step until retry', async () => {
@@ -428,7 +425,7 @@ describe('agent loop', () => {
expect(result.seq).toBeLessThan(contexts[0]!.seq)
expect(contexts.flatMap(event => event.type === 'user/message' ? event.data.content : []))
.toEqual([
{ type: 'text', text: 'mutated after inject' },
{ type: 'text', text: 'mid-turn notice' },
{ type: 'text', text: 'second notice' },
])
@@ -441,7 +438,7 @@ describe('agent loop', () => {
? [index]
: [])
expect(resultIndex).toBeGreaterThanOrEqual(0)
expect(contextIndexes).toHaveLength(1)
expect(contextIndexes).toHaveLength(2)
expect(contextIndexes.every(index => index > resultIndex)).toBe(true)
})