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

fix(agent-loop): address second-round review — disposal discard, injection validation, frozen payloads

Browse Source 
Address the review bot's five genuinely-new findings on the current code:
- disposal now discards any still-pending inbox items before the loop
  exits, so every enqueued id gets a terminal lifecycle event.
- injection (next-step/no-wakeup) validates its payload up front, before
  opening the idle one-shot turn, honoring 'invalid input throws before
  any append'; and rejects attached contexts (which belong only to inbox
  messages) rather than silently dropping them.
- agentMessage() freezes the agent/inbox/* payload so a listener cannot
  mutate the shared correlation object mid-dispatch.
- refresh the package READMEs (compact, goal, guard, hook-protocol,
  plan-mode, time-context, workspace-context) that still referenced the
  removed context/message event, with the source-based user/message
  distinction.

The up-front injection validation makes two finally branches unreachable
(v8-ignored as the turn-enclosure backstop). Adds regression tests for
disposal discard, context rejection, up-front validation, and the frozen
payload; per-file coverage stays 100%.
This commit is contained in:
Turtle
2026-07-24 12:05:57 +08:00
parent 7b7f793ee5
commit c7c1b97501
14 changed files with 127 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -248,14 +248,22 @@ export class ReactLoopAgent extends Agent {
/** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */
private injectContext(content: ContentBlock[], options?: SendOptions): void {
// Injection is synthetic durable context, not an inbox message: attached
// contexts belong only to queued/steering sends, so reject them rather than
// silently dropping a value the option type structurally permits.
if (options?.contexts !== undefined && options.contexts.length > 0) {
throw new TypeError('agent inject (next-step/no-wakeup) does not accept attached contexts')
}
const source = options?.source ?? { kind: 'plugin', plugin: '' }
const context = {
// Detach and validate the payload BEFORE any append, so malformed input
// throws without opening a one-shot turn or mutating the session (the
// unified send contract: invalid input throws before any append).
const accepted = this.acceptContext({
content,
source,
...options?.meta !== undefined ? { meta: options.meta } : {},
}
})
if (isTurnOpen(this.session)) {
const accepted = this.acceptContext(context)
// Provider protocols require every assistant tool-call batch to be
// followed only by its tool results. Historical interrupted batches do
// not own new context; only the currently executing batch may defer it.
@@ -267,39 +275,35 @@ export class ReactLoopAgent extends Agent {
return
}
// No turn open: wrap the injection in a one-shot turn so every event stays
// turn-enclosed (the durability/replay boundary is the turn).
// turn-enclosed (the durability/replay boundary is the turn). The payload is
// validated above, so both appends commit together; the finally still owes
// a turn/end (the turn-enclosure invariant) even if a post-commit observer
// throws after turn/start.
const turn = lastTurnNumber(this.session) + 1
// Once turn/start enters the log, a turn/end is owed even if the message
// append fails acceptance or pre-commit validation. The finally re-checks
// the log and closes only a turn that actually opened; post-commit observers
// are contained by Session and cannot create a false append failure.
try {
this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
this.session.append('user/message', context, { surfaceOp: 'append' })
this.session.append('user/message', accepted, { surfaceOp: 'append' })
} finally {
// Close the turn if turn/start made it into the log. A pre-commit veto
// must escape rather than being mistaken for a committed turn/end.
// Close the turn if turn/start committed. With the payload validated up
// front both appends commit together, so the turn is always open here;
// the guard remains the turn-enclosure backstop.
/* v8 ignore next -- unopened turn is unreachable after up-front validation; kept as the enclosure backstop. */
if (isTurnOpen(this.session)) {
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
}
// Decide the durability checkpoint from the log: an accepted one-shot
// turn must be flushed even when its message append was the failing step.
const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
// Keep inject() synchronous: report checkpoint failures live instead of
// rejecting the caller, and track the task so disposal still drains it.
if (turnRecorded) {
// Through the store's flush (the carrier owner), never a raw parallel.
const flush = this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
const rendered = errorChain(error)
const err = error instanceof Error ? error : new Error(rendered)
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${rendered}`)
agentEvents(this.loopCtx, this).emit('agent/error', turn, 0, err)
})
this.pendingIdleFlushes.add(flush)
// Retire on either settlement path.
const retire = (): void => { this.pendingIdleFlushes.delete(flush) }
void flush.then(retire, retire)
}
// Flush the one-shot turn through the store (the carrier owner), never a
// raw parallel. Keep inject() synchronous: report checkpoint failures live
// instead of rejecting the caller, and track the task so disposal drains it.
const flush = this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
const rendered = errorChain(error)
const err = error instanceof Error ? error : new Error(rendered)
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${rendered}`)
agentEvents(this.loopCtx, this).emit('agent/error', turn, 0, err)
})
this.pendingIdleFlushes.add(flush)
// Retire on either settlement path.
const retire = (): void => { this.pendingIdleFlushes.delete(flush) }
void flush.then(retire, retire)
}
}
@@ -434,6 +438,18 @@ export class ReactLoopAgent extends Agent {
*/
private [stopDriver](): Promise<void> | void {
if (this._status !== 'disposed') {
// Discard any still-pending inbox items before disposal so every enqueued
// id gets a terminal lifecycle event; a disposed agent never dequeues
// them. Emitted while still published (before the status flip below), and
// only when there is a public lifecycle to observe it.
if (this.published) {
const discarded = this.#inbox.pending()
if (discarded.length > 0) {
const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
}
}
this.#inbox.clear()
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are

9
packages/core/agent-loop/src/inbox.ts
View File

@@ -29,7 +29,14 @@ export interface InboxMessage {
* @returns the live-event message for enqueue/dequeue/discard.
*/
export function agentMessage(message: InboxMessage, steering: boolean): AgentMessage {
return { id: message.id, content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
// Frozen: the fused emitter passes this exact object to every listener in
// turn, so one listener must not be able to mutate a field (`id`, `steering`,
// `content`, …) a later listener then observes. `message` is already a frozen
// inbox record, so its nested fields need no re-clone.
return Object.freeze({
id: message.id, content: message.content, source: message.source,
contexts: message.contexts, steering, wakeup: message.wakeup,
})
}
/**

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

@@ -98,6 +98,27 @@ describe('Agent', () => {
expect(() => { agent.send([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
})
it('disposal discards still-pending inbox items so every id gets a terminal event', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: Agent
const discarded: string[] = []
ctx.on('agent/inbox/discard', (subject, messages) => {
if (subject === agent) discarded.push(...messages.map(m => m.id))
})
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(SessionId('scoped'), { provider: 'mock', model: 'mock' })
}, { inject: ['agentLoop'] }))
// A quiet (non-waking) item stays parked in the inbox; disposal must drop it
// WITH a discard so its enqueued id is not left dangling forever.
const id = agent.send([{ type: 'text', text: 'never runs' }], { target: 'next-turn', wakeup: false })
await fiber.dispose()
await driverDone(agent)
expect(discarded).toEqual([id])
})
it('steer() throws after disposal', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
@@ -177,24 +198,22 @@ describe('Agent', () => {
warn.mockRestore()
})
it('idle inject() closes its one-shot turn AND still checkpoints even if the append throws', async () => {
it('idle inject() validates its payload BEFORE opening a turn, so invalid input appends nothing', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let flushes = 0
ctx.on('session/flush', () => { flushes += 1 })
// Non-serializable injected content makes Session.append throw AFTER
// turn/start was recorded. The turn/end must still be appended (finally),
// AND the durability checkpoint must still fire — the balanced turn is in
// memory and a crash before the next turn/dispose would otherwise lose it.
// Non-serializable injected content is rejected by the up-front snapshot
// BEFORE any append (the unified send contract: invalid input throws before
// mutating the log). No one-shot turn opens and no durability checkpoint fires.
expect(() => {
agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
}).toThrow(/non-JSON-serializable/)
const types = agent.session.events.map(e => e.type)
expect(types).toEqual(['turn/start', 'turn/end']) // balanced, no open turn
await new Promise(r => setTimeout(r, 10)) // let the fire-and-forget flush run
expect(flushes).toBe(1) // checkpoint fired despite the throw
}).toThrow(/losslessly JSON-serializable/)
expect(agent.session.events).toHaveLength(0)
await new Promise(r => setTimeout(r, 10)) // give any (erroneous) flush a chance
expect(flushes).toBe(0) // nothing was appended, so no checkpoint
})
it('idle inject() still checkpoints when a listener throws on the synthetic turn/end', async () => {
@@ -244,13 +263,27 @@ describe('Agent', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// A non-serializable source makes the turn/start append throw BEFORE the
// event is pushed (Session.append validates before push), so NO turn opens.
// The finally's isTurnOpen() guard sees no open turn and appends nothing —
// the log stays empty, not left with a dangling turn/start.
// A non-serializable source is rejected by the up-front snapshot BEFORE any
// append, so NO turn opens and the log stays empty.
expect(() => {
agent.inject([{ type: 'text', text: 'x' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
}).toThrow(/non-JSON-serializable/)
}).toThrow(/losslessly JSON-serializable/)
expect(agent.session.events).toHaveLength(0)
})
it('inject() rejects attached contexts (they belong to inbox messages, not injection)', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// contexts structurally compile on AliasSendOptions but injection cannot
// carry them, so they are rejected rather than silently dropped.
expect(() => {
agent.inject([{ type: 'text', text: 'x' }], {
source: { kind: 'plugin', plugin: 'p' },
contexts: [{ content: [{ type: 'text', text: 'ctx' }], source: { kind: 'plugin', plugin: 'p' } }],
} as never)
}).toThrow(/does not accept attached contexts/)
expect(agent.session.events).toHaveLength(0)
})

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

@@ -1,11 +1,20 @@
import { describe, expect, it } from 'vitest'
import { AgentMessageId } from '@deepseek-ai/dsh-agent'
import { Inbox } from '../src/inbox.ts'
import { Inbox, agentMessage } from '../src/inbox.ts'
function message(text: string) {
return { id: AgentMessageId(text), content: [{ type: 'text' as const, text }], source: { kind: 'user' as const }, contexts: [], wakeup: true }
}
describe('agentMessage', () => {
it('returns a frozen payload so a listener cannot mutate it for later listeners', () => {
const payload = agentMessage(message('m'), false)
expect(Object.isFrozen(payload)).toBe(true)
expect(() => { (payload as { id: string }).id = 'mutated' }).toThrow()
expect(payload.id).toBe(AgentMessageId('m'))
})
})
function resolverPair() {
let r!: () => void
const p = new Promise<void>((resolve) => { r = resolve })