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

fix pre-step lifecycle regressions

Browse Source
This commit is contained in:
_Kerman
2026-07-31 19:40:59 +08:00
parent fcc2b5e282
commit 8e88b17c9f
37 changed files with 331 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
packages/goal/goal-session/src/index.ts
View File

@@ -10,7 +10,7 @@ import type { Agent, PreStepDecision } from '@deepseek-ai/dsh-agent'
import type { GoalMessageSource, GoalRef, GoalView } from '@deepseek-ai/dsh-goal'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageId, MessageSource } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, UserMessage } from '@deepseek-ai/dsh-session'
import { renderGoalRoundPrompt } from './prompt.ts'
export { renderGoalRoundPrompt } from './prompt.ts'
@@ -25,11 +25,11 @@ interface RoundIdentity {
readonly round: number
}
/** One queued or admitted goal message retained until whole-agent quiescence. */
/** One queued, claimed, or admitted goal message retained until whole-agent quiescence. */
interface RoundAttempt extends RoundIdentity {
readonly messageId: MessageId
readonly content: ContentBlock[]
phase: 'queued' | 'admitted'
phase: 'queued' | 'claimed' | 'admitted'
cancelled: boolean
stale: boolean
}
@@ -123,6 +123,16 @@ export function apply(ctx: Context): void {
}
}
/** Preserve claimed step context when this driver drops only its own round. */
function restoreOtherClaimed(agent: Agent, messages: UserMessage[], messageId: MessageId): void {
const retained = messages.filter(message => message.id !== messageId)
for (const message of retained.toReversed()) {
if (agent.inbox.nextStep.some(candidate => candidate.id === message.id)
|| agent.inbox.nextTurn.some(candidate => candidate.id === message.id)) continue
agent.inbox.prepend('next-step', message)
}
}
/** Process admitted work at quiescence, then reserve at most one next round. */
async function drive(state: DriverState): Promise<void> {
const { agent } = state
@@ -144,7 +154,7 @@ export function apply(ctx: Context): void {
const attempt = state.attempt
if (attempt !== undefined) {
if (attempt.phase === 'queued') return
if (attempt.phase === 'queued' || attempt.phase === 'claimed') return
state.attempt = undefined
state.needsCheckpoint = true
state.requested = true
@@ -252,7 +262,7 @@ export function apply(ctx: Context): void {
state.competingQueued = false
const attempt = state.attempt
const goal = currentGoal(state)
if ((attempt?.phase === 'queued' || attempt?.cancelled)
if ((attempt?.phase === 'queued' || attempt?.phase === 'claimed' || attempt?.cancelled)
&& goal?.phase === 'active' && goal.activation === 'armed') {
state.attempt = undefined
try {
@@ -271,21 +281,34 @@ export function apply(ctx: Context): void {
requestDrive(state)
})
ctx.on('agent/inbox/inserted', (agent, { message }) => {
if (!agent.inbox.nextTurn.some(candidate => candidate.id === message.id)) return
const state = stateFor(agent)
const attempt = state.attempt
if (attempt !== undefined && sameQueued(message.content, message.source, attempt)) return
state.competingQueued = true
if (attempt?.phase === 'queued') attempt.stale = true
})
ctx.on('agent/inbox/claimed', (agent, { message }) => {
const state = stateFor(agent)
const attempt = state.attempt
if (attempt !== undefined && sameQueued(message.content, message.source, attempt)) {
attempt.phase = 'claimed'
}
})
ctx.on('agent/inbox/discarded', (agent, { message }) => {
const state = stateFor(agent)
const attempt = state.attempt
if (attempt !== undefined && sameQueued(message.content, message.source, attempt)) {
attempt.cancelled = true
}
})
ctx.on('session/event', (session: Session, event: SessionEvent) => {
const agent = ctx.agents.get(session.id)
if (agent === undefined || agent.session !== session) return
const state = stateFor(agent)
switch (event.type) {
case 'agent/inbox/spliced': {
if (event.data.target !== 'next-turn') return
const attempt = state.attempt
for (const message of event.data.inserted) {
if (attempt !== undefined && sameQueued(message.content, message.source, attempt)) continue
state.competingQueued = true
if (attempt?.phase === 'queued') attempt.stale = true
}
return
}
case 'user/message':
if (state.attempt !== undefined && event.data.id === state.attempt.messageId) {
state.attempt.phase = 'admitted'
@@ -314,7 +337,7 @@ export function apply(ctx: Context): void {
const attempt = state.attempt
const goal = currentGoal(state)
return ctx.fiber.state === FiberState.ACTIVE
&& !state.stopping && attempt !== undefined && attempt.phase === 'queued'
&& !state.stopping && attempt !== undefined && attempt.phase === 'claimed'
&& !attempt.stale && sameQueued(content, source, attempt)
&& goal !== undefined && goal.id === source.goalId && goal.revision === source.revision
&& goal.phase === 'active' && goal.activation === 'armed'
@@ -340,6 +363,7 @@ export function apply(ctx: Context): void {
attempt.stale = true
state.attempt = undefined
}
restoreOtherClaimed(agent, messages, submitted.id)
requestDrive(state)
return { kind: 'reject' }
}
@@ -353,13 +377,16 @@ export function apply(ctx: Context): void {
// starve every later drive pass. Clear it and let the driver
// reschedule the round.
const attempt = state.attempt
if (attempt !== undefined && sameRound(source, attempt) && attempt.phase === 'queued') {
if (attempt !== undefined && sameRound(source, attempt) && attempt.phase === 'claimed') {
state.attempt = undefined
requestDrive(state)
}
throw error
}
if (signal.aborted) return decision
if (signal.aborted) {
if (decision.kind === 'enter') restoreOtherClaimed(agent, decision.messages, submitted.id)
return decision
}
if (decision.kind === 'reject') {
const attempt = state.attempt
if (attempt !== undefined && sameRound(source, attempt)) state.attempt = undefined
@@ -386,6 +413,7 @@ export function apply(ctx: Context): void {
attempt.stale = true
state.attempt = undefined
}
restoreOtherClaimed(agent, decision.messages, submitted.id)
requestDrive(state)
return { kind: 'reject' }
}
@@ -409,7 +437,8 @@ export function apply(ctx: Context): void {
const attempt = state.attempt
if (attempt !== undefined) {
attempt.stale = true
if (attempt.phase === 'admitted' && state.agent.status === 'running') {
if ((attempt.phase === 'claimed' || attempt.phase === 'admitted')
&& state.agent.status === 'running') {
state.agent.cancel({ kind: 'parent' })
waits.push(state.agent.whenIdle())
}

17
packages/goal/goal-session/tests/goal-session.spec.ts
View File

@@ -113,6 +113,17 @@ function onInboxMessage(
})
}
/** Observe one claimed message at its exclusive pre-step ownership transfer. */
function onClaimedMessage(
ctx: Context,
agent: Agent,
listener: (message: UserMessage) => void,
): () => void {
return ctx.on('agent/inbox/claimed', (subject, { message }) => {
if (subject === agent) listener(message)
})
}
/** Await a stable goal projection selected by the caller. */
async function waitForGoal(
ctx: Context,
@@ -273,8 +284,8 @@ describe('same-session goal driving', () => {
it('pauses and drops a reserved round when cancellation lands before pre-step', async () => {
const test = await harness([])
const cancel = onInboxMessage(test.ctx, test.agent, (message) => {
if (message.source.kind === 'goal') {
const cancel = onClaimedMessage(test.ctx, test.agent, (message) => {
if (message.source.kind === 'goal' && message.source.round > 0) {
cancel()
test.agent.cancel({ kind: 'user' })
}
@@ -616,7 +627,7 @@ describe('same-session goal driving', () => {
it('fails an initial pre-step read closed even when the first disarm attempt throws', async () => {
const test = await harness([textResponse('retry after containment')])
let armed = true
onInboxMessage(test.ctx, test.agent, (message) => {
onClaimedMessage(test.ctx, test.agent, (message) => {
if (message.source.kind !== 'goal' || message.source.round <= 0 || !armed) return
armed = false
vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => {