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

Fix async injection tool-result ordering

Browse Source
This commit is contained in:
Hypatia May
2026-07-15 11:12:05 +08:00
parent 34bb75b2d2
commit 4ce32807e1
10 changed files with 104 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,7 +12,7 @@ import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import { snapshotJsonValue, type Session } from '@deepseek-ai/dsh-session'
import { isToolPairingBalanced, snapshotJsonValue, type Session } from '@deepseek-ai/dsh-session'
import { Inbox, type InboxMessage } from './inbox.ts'
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
@@ -149,6 +149,8 @@ export class ReactLoopAgent implements Agent {
* this set before the lifecycle unregisters the agent or detaches its session.
*/
private pendingIdleFlushes = new Set<Promise<void>>()
/** Open-turn injections waiting for the active assistant tool-call batch to close. */
private deferredInjections: InboxMessage[] = []
constructor(
private loopCtx: Context,
@@ -189,12 +191,11 @@ export class ReactLoopAgent implements Agent {
}
/**
* Accept one public send/steer payload as the exact detached record shared by
* the live notification and inbox. Lossless-JSON materialization reads every
* nested field once; deep freeze prevents an observer from rewriting queued
* work before the loop drains it.
* Accept one public message payload as a detached record. Lossless-JSON
* materialization reads every nested field once; deep freeze prevents later
* caller mutation before an inbox or deferred-injection queue drains it.
*/
private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const accepted = snapshotJsonValue({ content, source })
if (accepted === undefined) {
@@ -210,7 +211,7 @@ export class ReactLoopAgent implements Agent {
send(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -219,7 +220,7 @@ export class ReactLoopAgent implements Agent {
steer(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
if (this._status !== 'running') { this.send(content, options); return }
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -227,14 +228,21 @@ export class ReactLoopAgent implements Agent {
inject(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
const source = this.resolveSource(options)
if (isTurnOpen(this.session)) {
// A turn is open in the LOG (decided from the log, not agent status —
// status can be `running` with no turn open): the context/message is
// turn-enclosed by that turn, so append it directly.
this.session.append('context/message', { content, source }, { surfaceOp: 'append' })
const accepted = this.acceptMessage(content, options)
// Provider protocols require every assistant tool-call batch to be
// followed only by its tool results. Queue arbitrary asynchronous context
// until the tail cut is balanced; an existing queue preserves FIFO in the
// narrow window after the last result and before the loop drains it.
if (this.deferredInjections.length > 0
|| !isToolPairingBalanced(this.session.surface.nodes, this.session.events, null)) {
this.deferredInjections.push(accepted)
return
}
this.session.append('context/message', accepted, { surfaceOp: 'append' })
return
}
const source = this.resolveSource(options)
// No turn open: wrap the injection in a one-shot turn so every event stays
// turn-enclosed (the durability/replay boundary is the turn).
const turn = lastTurnNumber(this.session) + 1
@@ -272,6 +280,14 @@ export class ReactLoopAgent implements Agent {
}
}
/** Append deferred open-turn injections after the loop closes a tool-result batch. */
private drainDeferredInjections(): void {
const pending = this.deferredInjections.splice(0)
for (const accepted of pending) {
this.session.append('context/message', accepted, { surfaceOp: 'append' })
}
}
cancel(reason?: string): void {
// Arm only for current work; an idle marker would cancel the next prompt.
if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
@@ -336,6 +352,7 @@ export class ReactLoopAgent implements Agent {
isCancelled: () => this.cancelRequested,
cancelReason: () => this.cancelReason,
clearCancel: () => { this.cancelRequested = false },
drainDeferredInjections: () => { this.drainDeferredInjections() },
// Pre-step cancellation re-parks without emitting a status transition.
settleIdle: () => { this.settleIdleWaiters() },
})