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

refactor(agent): unify agent-scoped event signatures as payload objects

Browse Source 
All agent/* and agent-loop/config-start-failed events take one payload
object carrying the agent subject; waterfall/serial payloads require a
signal and keep next as the final argument. PreStepContext and
RequestFailureContext are unfolded into payloads and retired.
goal/changed follows the same shape so agentEvents keeps its listener
error containment. ReactLoopAgent builds its scope carrier once in the
constructor. Regenerates scope resolvers, tool-cordis api catalog, and
docs catalogs; updates all affected listeners, tests, and the
core-data-structures docs (en + zh).
This commit is contained in:
_Kerman
2026-08-06 12:13:14 +08:00
parent bb53e25ed0
commit ccebba2349
91 changed files with 574 additions and 618 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/subagent/subagent-inprocess/src/index.ts
View File

@@ -75,7 +75,7 @@ function prePublicationAbort(): Error {
/** Append one one-shot descriptor inside the child's initial turn before its first request. */
function attachDescriptorAppend(childCtx: Context, descriptor: SubagentDescriptorData): void {
let appended = false
childCtx.on('agent/pre-step', async (agent, _messages, _context, next) => {
childCtx.on('agent/pre-step', async ({ agent }, next) => {
const decision = await next()
if (!appended && decision.kind === 'enter') {
appended = true

2
packages/subagent/subagent-spawn/tests/harness.ts
View File

@@ -42,7 +42,7 @@ export async function spawnHarness(workdir: string): Promise<Context> {
export function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()

6
packages/subagent/subagent/src/continuation.ts
View File

@@ -283,7 +283,7 @@ export class SubagentContinuationManager {
// child-first ordering.
const scope = ctx.plugin(function activationOwner() {})
this.ownerCtx = scope.ctx
ctx.on('agent/disposed', (agent) => {
ctx.on('agent/disposed', ({ agent }) => {
this.closingScopes.delete(agent)
})
ctx.effect(function* (this: SubagentContinuationManager) {
@@ -854,12 +854,12 @@ export class SubagentContinuationManager {
// quiet Agent from one whose accepted turn has not been admitted yet.
// Registered through the child's own scoped context, so scope filtering
// already restricts both listeners to this exact agent.
handle.agent.ctx.on('agent/inbox/claimed', (_agent, { message }) => {
handle.agent.ctx.on('agent/inbox/claimed', ({ message }) => {
/* v8 ignore next -- a claim of an id this manager never admitted needs
* another sender on the same child, which no current path allows. */
if (activation.accepted.delete(message.id)) this.wake(activation)
})
handle.agent.ctx.on('agent/inbox/discarded', (_agent, { message }) => {
handle.agent.ctx.on('agent/inbox/discarded', ({ message }) => {
if (activation.accepted.delete(message.id)) this.wake(activation)
})
// Agent creation committed setup at its publication boundary;

32
packages/subagent/subagent/tests/continuation.spec.ts
View File

@@ -159,10 +159,10 @@ describe('SubagentService.startContinuable', () => {
it('returns both identities at inbox acceptance, without waiting for the turn or the log', async () => {
const { ctx, parent, adapter } = await setup([textResponse('first answer')])
const enqueued: { id: MessageId; loggedYet: boolean }[] = []
ctx.on('agent/inbox/inserted', (agent, accepted) => {
ctx.on('agent/inbox/inserted', ({ agent, message }) => {
// Acceptance is the boundary `startContinuable` resolves at, so observe
// the log state exactly there rather than after later microtasks.
enqueued.push({ id: accepted.message.id, loggedYet: hasUserText(agent.session.events, 'child task') })
enqueued.push({ id: message.id, loggedYet: hasUserText(agent.session.events, 'child task') })
})
const started = await ctx.subagents.startContinuable(startSpec(parent))
@@ -231,7 +231,7 @@ describe('SubagentService.startContinuable', () => {
const { ctx, parent } = await setup([textResponse('unused')])
const controller = new AbortController()
// Abort inside the child's creation window: setup runs before publication.
ctx.on('agent/created', (child) => {
ctx.on('agent/created', ({ agent: child }) => {
if (child !== parent) controller.abort('caller gave up')
})
@@ -753,7 +753,7 @@ describe('continuable durability and teardown', () => {
await vi.waitFor(() => { expect(ctx.agents.get(grandchild.childId)).toBeDefined() })
const disposals: SessionId[] = []
ctx.on('agent/disposed', (agent) => { disposals.push(agent.id) })
ctx.on('agent/disposed', ({ agent }) => { disposals.push(agent.id) })
const drained = drainManager(ctx)
// Let the held model call observe its cancellation so quiescence can settle.
hold.resolve(undefined)
@@ -984,7 +984,7 @@ describe('continuable durability and teardown', () => {
const drains: Promise<void>[] = []
const accepted: MessageId[] = []
ctx.on('subagent/start', () => { drains.push(drainManager(ctx)) })
ctx.on('agent/inbox/inserted', (_agent, item) => { accepted.push(item.message.id) })
ctx.on('agent/inbox/inserted', ({ message }) => { accepted.push(message.id) })
await expect(ctx.subagents.startContinuable(startSpec(parent)))
.rejects.toMatchObject({ code: 'DRAINING' })
@@ -998,12 +998,12 @@ describe('continuable durability and teardown', () => {
const { ctx, parent } = await setup([])
const order: string[] = []
const drains: Promise<void>[] = []
ctx.on('agent/created', (child) => {
ctx.on('agent/created', ({ agent: child }) => {
if (child === parent) return
const draining = drainManager(ctx).then(() => { order.push('drain') })
drains.push(draining)
})
ctx.on('agent/disposed', (child) => {
ctx.on('agent/disposed', ({ agent: child }) => {
if (child !== parent) order.push('disposed')
})
@@ -1025,8 +1025,8 @@ describe('continuable durability and teardown', () => {
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
const child = ctx.agents.get(started.childId)!
const order: string[] = []
child.ctx.on('agent/inbox/inserted', (_agent, accepted) => {
if (accepted.message.content.some(block => block.type === 'text' && block.text === 'before drain')) {
child.ctx.on('agent/inbox/inserted', ({ message }) => {
if (message.content.some(block => block.type === 'text' && block.text === 'before drain')) {
order.push('enqueue')
}
})
@@ -1208,7 +1208,7 @@ describe('continuable review regressions', () => {
const ends: SubagentRunEndInfo[] = []
ctx.on('subagent/end', (info) => { ends.push(info) })
// Block the resumed prompt so this epoch produces nothing of its own.
ctx.on('agent/pre-step', async (subject, _messages, _context, next) => {
ctx.on('agent/pre-step', async ({ agent: subject }, next) => {
if (subject === parent) return next()
return { kind: 'reject' }
})
@@ -1356,8 +1356,8 @@ describe('continuable review regressions', () => {
// Cancel from the synchronous enqueue observer: the discard fires after the
// id is recorded but before `followup()` returns.
const off = child.ctx.on('agent/inbox/inserted', (_agent, accepted) => {
if (accepted.message.content.some(block => block.type === 'text' && block.text === 'doomed')) {
const off = child.ctx.on('agent/inbox/inserted', ({ message }) => {
if (message.content.some(block => block.type === 'text' && block.text === 'doomed')) {
child.cancel({ kind: 'user' })
}
})
@@ -1388,8 +1388,8 @@ describe('continuable review regressions', () => {
await followup(ctx, parent, started.childId, message('queued'))
expect(activation.accepted.size).toBe(1)
const off = child.ctx.on('agent/inbox/inserted', (_agent, accepted) => {
if (accepted.message.content.some(block => block.type === 'text' && block.text === 'doomed')) {
const off = child.ctx.on('agent/inbox/inserted', ({ message }) => {
if (message.content.some(block => block.type === 'text' && block.text === 'doomed')) {
child.cancel({ kind: 'user' })
}
})
@@ -1406,7 +1406,7 @@ describe('continuable review regressions', () => {
const ends: SubagentRunEndInfo[] = []
ctx.on('subagent/end', (info) => { ends.push(info) })
// Block admission so the child's only turn never opens.
ctx.on('agent/pre-step', async (subject, _messages, _context, next) => {
ctx.on('agent/pre-step', async ({ agent: subject }, next) => {
if (subject === parent) return next()
return { kind: 'reject' }
})
@@ -1428,7 +1428,7 @@ describe('continuable review regressions', () => {
const registeredAtEnqueue: boolean[] = []
// A synchronous inbox observer runs before the admitting microtask, the
// exact window where `Agent.status` is still idle.
ctx.on('agent/inbox/inserted', (agent) => {
ctx.on('agent/inbox/inserted', ({ agent }) => {
if (agent.session.header.parentSession !== undefined) {
registeredAtEnqueue.push(ctx.agents.get(agent.id) === agent)
}

8
packages/subagent/tool-subagent-report/tests/tool-subagent-report.spec.ts
View File

@@ -164,9 +164,9 @@ describe('dsh-tool-subagent-report', () => {
const { started, child } = await startChild(ctx, parent)
const parentRequests = adapter.requests.filter(request => request.sessionId === parent.id).length
const enqueues: string[] = []
ctx.on('agent/inbox/inserted', (agent, item) => {
ctx.on('agent/inbox/inserted', ({ agent, message }) => {
if (agent === parent) {
enqueues.push(agent.inbox.nextTurn.some(message => message.id === item.message.id) ? 'queued' : 'steering')
enqueues.push(agent.inbox.nextTurn.some(queued => queued.id === message.id) ? 'queued' : 'steering')
}
})
@@ -190,9 +190,9 @@ describe('dsh-tool-subagent-report', () => {
const { ctx, parent, adapter } = await setup({ config: { reportDelivery: 'wakeup' } })
const { child } = await startChild(ctx, parent)
const enqueues: string[] = []
ctx.on('agent/inbox/inserted', (agent, item) => {
ctx.on('agent/inbox/inserted', ({ agent, message }) => {
if (agent === parent) {
enqueues.push(agent.inbox.nextTurn.some(message => message.id === item.message.id) ? 'queued' : 'steering')
enqueues.push(agent.inbox.nextTurn.some(queued => queued.id === message.id) ? 'queued' : 'steering')
}
})