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

Merge latest master into codex/llm-error-recovery-rfc

Browse Source 
# Conflicts:
#	docs/architecture.md
#	docs/config-catalog.md
#	docs/cordis-catalog/events.md
#	docs/event-producer-consumer.md
#	docs/module-graph.md
#	docs/persistence-catalog.md
#	packages/cordis/tool-cordis/src/api-catalog.ts
#	packages/core/agent-loop/src/loop.ts
#	packages/core/agent/src/types.ts
#	packages/examples/agent-spine-demo/README.md
#	packages/examples/stdio-demo/README.md
#	packages/examples/stdio-demo/src/index.ts
#	packages/llm/llm-deepseek/README.md
#	packages/llm/llm-deepseek/src/adapter.ts
#	packages/llm/llm-deepseek/tests/adapter.spec.ts
#	packages/llm/llm/README.md
#	packages/llm/llm/src/error.ts
#	packages/llm/llm/tests/service.spec.ts
#	packages/sandbox/sandbox-policy/tsconfig.json
#	packages/ui/stdio/README.md
#	packages/ui/stdio/src/index.ts
#	packages/ui/stdio/tests/stdio.spec.ts
#	packages/ui/tui/src/index.ts
#	pnpm-lock.yaml
#	website/zh-CN/api/harness/events.md
#	website/zh-CN/api/harness/llm.md
This commit is contained in:
Tianyi Cui
2026-07-20 20:59:11 +08:00
parent 1777785986 7b7d6f38e8
commit 8c12f8945e
672 changed files with 18585 additions and 13005 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/agent-loop/src/loop.ts
+52 -55
View File
@@ -8,7 +8,7 @@
import type { Context } from 'cordis'
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
import { isDeepStrictEqual } from 'node:util'
import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, llmFailureOf } from '@deepseek-ai/dsh-llm'
import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf } from '@deepseek-ai/dsh-llm'
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
@@ -61,9 +61,12 @@ function finishError(finish: FinishReason): { error: RequestError; failure: LlmF
/**
* Build the `{ message, code? }` part of an error payload, omitting the
* `code` key entirely when absent (exactOptionalPropertyTypes-correct).
* The durable message renders the full cause chain: `turn/end` is the single
* durable record of an in-turn failure, so a wrapper message alone (e.g.
* `fetch failed`) would lose the diagnosis the session log exists to keep.
*/
function errorData(err: RequestError): { message: string; code?: string } {
return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
return { message: errorChain(err), ...typeof err.code === 'string' ? { code: err.code } : {} }
}
/** Map a successful max-token finish onto the turn reason; other successful finishes add nothing. */
@@ -96,16 +99,16 @@ export interface LoopHandle {
cancelReason(): string
/** Clear the cancel marker (called once per iteration after the turn returns). */
clearCancel(): void
/** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
/** Settle idle waiters before pre-running cancellation publishes idle. */
settleIdle(): void
/** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
}
/**
* Drive queued batches as durable turns until disposal. Plugin failures end the
* current turn without terminating the driver. The caller establishes the
* `ctx.agents.withInitiator()` boundary before entry; package-private
* Drive queued messages as independent durable turns until disposal. Plugin
* failures end the current turn without terminating the driver. The caller
* establishes the `ctx.agents.withInitiator()` boundary before entry; package-private
* orchestration recovers that exact Agent and captures its Session locally.
* @param ctx - the plugin context the loop reaches its initiating Agent,
* events (agent/…, session/flush), and services (systemPrompt, llm, tools)
@@ -123,20 +126,35 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
const events = agentEvents(ctx, agent)
while (!handle.isDisposed()) {
await handle.inbox.waitForQueued(handle.disposed)
if (handle.isDisposed()) break
// Cancellation between wake and `running` skips only the cancelled work;
// a replacement prompt still runs and owns the eventual idle transition.
// An idle listener can enqueue and cancel replacement work before the next
// wait is installed. Consume that empty marker before parking the driver.
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
handle.settleIdle()
handle.setStatus('idle')
continue
}
}
await handle.inbox.waitForQueued(handle.disposed)
if (handle.isDisposed()) break
// Cancellation between wake and `running` skips only the cancelled work;
// a replacement prompt still runs before the eventual idle transition.
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
// Settle before publishing idle: the already-idle path has no status
// transition, while an idle listener can register waiters for new work.
handle.settleIdle()
handle.setStatus('idle')
continue
}
}
handle.setStatus('running')
if (handle.isDisposed()) break
// A synchronous `running` listener can cancel before `runTurn`; balance the
// status only when no replacement prompt was queued by that listener.
@@ -156,7 +174,7 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
} catch (error: unknown) {
// Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${errorChain(err)}`)
try {
events.emit('agent/error', turn, 0, err)
} catch { /* contained: a throwing agent/error listener must not kill the driver */ }
@@ -187,12 +205,11 @@ async function runTurn(
return messages.length > 0
}
// Drain before opening the turn, but append only after `turn/start`.
const queued = handle.inbox.drainQueued()
const first = queued[0]
// Claim one queued message before opening its turn, but append it only after `turn/start`.
const message = handle.inbox.dequeueQueued()
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
if (!first) throw new Error('runTurn invariant violated: no queued message at turn start')
const trigger: TurnTrigger = { kind: 'message', source: first.source }
if (!message) throw new Error('runTurn invariant violated: no queued message at turn start')
const trigger: TurnTrigger = { kind: 'message', source: message.source }
let reason: TurnEndReason = { kind: 'completed' }
let step = 0
@@ -214,7 +231,7 @@ async function runTurn(
errorReported = true
reason = failure === undefined
? { kind: 'error', step, ...errorData(err) }
: { kind: 'error', step, failure }
: { kind: 'error', step, failure: { ...failure, message: errorChain(err) } }
try {
events.emit('agent/error', turn, step, err)
} catch {
@@ -233,56 +250,36 @@ async function runTurn(
// matter what throws below; the catch + closeTurn guarantee it. A pre-commit
// veto leaves no turn/start in the log and therefore owes no turn/end.
session.append('turn/start', { turn, trigger })
// Each drained queued message runs the `agent/prompt-submit` waterfall before
// it becomes a `user/message` — a hook can rewrite the prompt or block it.
// The claimed message runs the `agent/prompt-submit` waterfall before it
// becomes a `user/message` — a hook can rewrite the prompt or block it.
// Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed;
// turn/end is now owed, so a throwing prompt-submit listener (the waterfall
// throws) is caught below and the turn still closes.
let anyAllowed = false
// Seeded with a floor (only observable if the batch were empty, which
// runTurn never allows — it is called with ≥1 queued message); each `block`
// decision carries a required `reason` and overwrites it, so a fully-blocked
// batch always reports the last vetoing reason.
let lastBlockReason = 'prompt blocked by hook'
for (const message of queued) {
const decision = await events.waterfall(
'agent/prompt-submit', message.content, message.source,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
)
if (decision.kind === 'block') {
lastBlockReason = decision.reason
// Record the veto durably: `PromptDecision.reason` is the durable record
// of why a prompt was blocked, but a fully-blocked batch's `rejected`
// turn/end only preserves the LAST reason, and a MIXED batch (this prompt
// blocked, another allowed) does not end `rejected` at all — so without
// this append a blocked prompt would vanish from the log whenever any
// sibling prompt is allowed. `prompt/blocked` sits in the open turn in
// place of the `user/message` this prompt would have become.
session.append('prompt/blocked', { content: message.content, source: message.source, reason: decision.reason })
continue
}
anyAllowed = true
const promptDecision = await events.waterfall(
'agent/prompt-submit', message.content, message.source,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
)
if (promptDecision.kind === 'block') {
session.append('prompt/blocked', { content: message.content, source: message.source, reason: promptDecision.reason })
reason = { kind: 'rejected', reason: promptDecision.reason }
} else {
// `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
const content = decision.content ?? message.content
const content = promptDecision.content ?? message.content
session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
// Every `allow.additionalContexts` entry is a separate context/message the
// next request also sees. The turn is open, so inject() appends each one
// into THIS turn without flattening provenance, framing, or metadata.
for (const context of decision.additionalContexts ?? []) {
// into THIS turn without flattening provenance or metadata.
for (const context of promptDecision.additionalContexts ?? []) {
agent.inject(context.content, {
source: context.source,
...context.envelope !== undefined ? { envelope: context.envelope } : {},
...context.meta !== undefined ? { meta: context.meta } : {},
})
}
}
while (true) {
// A fully blocked batch closes its zero-step turn as rejected.
if (!anyAllowed) {
reason = { kind: 'rejected', reason: lastBlockReason }
break
}
// A blocked prompt closes its zero-step turn as rejected.
if (promptDecision.kind === 'block') break
step += 1
// Steering from the previous round's continuation listeners joins before
@@ -395,7 +392,7 @@ async function runTurn(
)
} catch (recoveryError: unknown) {
ctx.logger.warn(
`agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${toError(recoveryError).message}`,
`agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`,
)
}
handle.setAbort(undefined)
@@ -559,7 +556,7 @@ async function runTurn(
} catch (error: unknown) {
// The turn is closed, so report the failed flush live rather than append outside a turn.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${errorChain(err)}`)
try {
events.emit('agent/error', turn, step, err)
} catch {