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

Merge remote-tracking branch 'origin/master' into codex/status-bar-token-metrics

Browse Source 
Conflict resolution notes:

- StatsLine: master redesigned the row into pipe-separated groups with LLM and
  tool wall times. Kept that design and swapped only the token accounting
  source, so counts and durations stay window-scoped while billing and context
  occupancy read the durable projections.
- Generated artifacts (cordis catalog, module graph, event producer/consumer,
  i18n pairing hashes) and web snapshots took master's side; they are
  regenerated and re-recorded after this merge.
- Web e2e goldens and details-panel/timeline assertions took master's side:
  that evolution is unrelated to this branch.
- ui-conversation package.json: kept master's devDependency ordering, re-adding
  only the token-meter entry this branch needs.
This commit is contained in:
Hypatia May
2026-07-30 13:56:39 +08:00
parent 4819210142 1db976210f
commit 7f30a7e30c
1091 changed files with 45715 additions and 7873 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/agent-loop/src/agent.ts
+58 -14
View File
@@ -8,13 +8,18 @@
*/
import type { Context } from 'cordis'
import { agentCarrier, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
import { randomUUID } from 'node:crypto'
import { agentCarrier, assembleContextFor, emitAgentEvent, InboxItemId } from '@deepseek-ai/dsh-agent'
import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import type {
Agent,
CancelOptions,
AgentInterruptReason,
InboxAction,
InboxActionResult,
InboxItem,
InboxItemId as InboxItemIdType,
InboxPlacement,
AgentOptions,
AgentStatus,
@@ -55,9 +60,9 @@ type StepOutcome =
*/
export class ReactLoopAgent implements Agent {
/** Prompts awaiting individual turns. */
private queued: { message: UserMessage; wakeup: boolean }[] = []
private queued: { item: InboxItem; wakeup: boolean }[] = []
/** Input taken into the session log at step boundaries. */
private outbox: { message: UserMessage; steering: boolean }[] = []
private outbox: { message: UserMessage; steering: boolean; item?: InboxItem }[] = []
/** Whether observers see a running interval; consecutive turns share it. */
private busy = false
@@ -115,16 +120,52 @@ export class ReactLoopAgent implements Agent {
}
const placement: InboxPlacement = target === 'next-step' && this.acceptsNextStep ? 'steering' : 'queued'
const item: InboxItem = Object.freeze({
id: InboxItemId(randomUUID()),
message,
placement,
})
if (placement === 'steering') {
this.outbox.push({ message, steering: true })
this.outbox.push({ message, steering: true, item })
} else {
this.queued.push({ message, wakeup })
this.queued.push({ item, wakeup })
}
// Preserve the routing decision for every send in this synchronous caller
// stack, while installing quiescence ownership before enqueue observers
// can cancel or dispose.
if (placement === 'queued' && wakeup) this.scheduleKick()
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message, placement)
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', item)
}
/** Apply one synchronous mutation to a still-pending queued occurrence. */
updateInbox(id: InboxItemIdType, action: InboxAction): InboxActionResult {
const queuedIndex = this.queued.findIndex(candidate => candidate.item.id === id)
if (queuedIndex === -1) return 'not-found'
const pending = this.queued[queuedIndex]
/* v8 ignore next -- the index was resolved from this array without an async boundary. */
if (pending === undefined) throw new Error(`agent "${this.id}" queued item disappeared during update`)
/* v8 ignore next -- InboxAction is a closed discriminated union; all variants are covered below. */
switch (action.kind) {
case 'edit': {
const item: InboxItem = Object.freeze({
...pending.item,
message: freezeMessage({ ...pending.item.message, content: action.content }),
})
this.queued[queuedIndex] = { ...pending, item }
emitAgentEvent(this.loopCtx, this, 'agent/inbox/update', item)
return 'applied'
}
case 'remove': {
this.queued.splice(queuedIndex, 1)
emitAgentEvent(this.loopCtx, this, 'agent/inbox/discard', [pending.item])
return 'applied'
}
default:
/* v8 ignore next -- InboxAction is a closed discriminated union. */
return assertNever(action)
}
}
/** Queue one ordinary prompt turn and wake the driver. */
@@ -169,9 +210,9 @@ export class ReactLoopAgent implements Agent {
if (cause.kind !== 'disposed') emitAgentEvent(this.loopCtx, this, 'agent/cancel-requested', cause)
}
if (!options.keepInbox) {
const discarded = this.queued.map(item => item.message)
const discarded = this.queued.map(item => item.item)
for (const item of this.outbox) {
if (item.steering) discarded.push(item.message)
if (item.steering && item.item !== undefined) discarded.push(item.item)
}
// Clear before abort observers run: replacement work belongs to the next turn.
this.queued.length = 0
@@ -221,8 +262,9 @@ export class ReactLoopAgent implements Agent {
if (this.abort !== undefined || !this.queued.some(item => item.wakeup)) return
// The some() guard above proves the queue is non-empty; the non-null
// assertion expresses that invariant.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const { message } = this.queued.shift()!
// oxlint-disable-next-line typescript/no-non-null-assertion
const { item } = this.queued.shift()!
const { message } = item
const inheritedOutboxLength = this.outbox.length
const admission = new AbortController()
@@ -293,7 +335,7 @@ export class ReactLoopAgent implements Agent {
// Published only after the abort owner and pending done are installed: a
// dequeue listener that cancels or disposes must find live cancellation
// and quiescence ownership, not the previous activity's settled state.
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message, 'queued')
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item)
}
/**
@@ -368,7 +410,7 @@ export class ReactLoopAgent implements Agent {
outcome.failure, requestFailureHistory, outcome.retryPolicy, signal,
() => Promise.resolve<RequestErrorAction>(undefined),
)
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- signal can abort while recovery is awaited.
// oxlint-disable-next-line typescript/no-unnecessary-condition -- signal can abort while recovery is awaited.
if (action?.kind === 'retry' && !signal.aborted) {
retryFailures = Object.freeze([...requestFailureHistory, outcome.failure])
}
@@ -584,7 +626,7 @@ export class ReactLoopAgent implements Agent {
const maxTokens = this.options.maxTokens
const seedConfig = deepFreeze(structuredClone(
this.requestHeaderLogged
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- the instance logged the header it now folds
// oxlint-disable-next-line typescript/no-non-null-assertion -- the instance logged the header it now folds
? persistedConfig!
: {
...route,
@@ -660,7 +702,9 @@ export class ReactLoopAgent implements Agent {
for (const item of this.outbox.splice(0, limit)) {
if (item.steering) {
steered = true
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.message, 'steering')
/* v8 ignore next -- only inbox-backed steer entries carry steering:true. */
if (item.item === undefined) throw new Error(`agent "${this.id}" steering outbox item has no inbox identity`)
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.item)
this.session.append(
'steering/message',
{ turn, message: item.message },
packages/core/agent-loop/src/tool-calls.ts
+54 -27
View File
@@ -2,10 +2,12 @@
* Schedules one assistant step's tool calls. Exclusive calls form barriers;
* parallel calls use a bounded rolling pool and are reclassified before start.
* Dispatch may overlap, while policy, results, and result context remain
* model-ordered. Abort stops replenishment and drains started calls.
* model-ordered. Abort or an internal scheduler failure stops replenishment
* and drains started calls.
*
* Each advertised call records a balanced `tool/call`/`tool/result` pair. Calls
* skipped after abort receive synthetic error results so replay stays valid.
* Abort records synthetic error results for skipped calls so replay stays
* valid. A terminal scheduler failure preserves already-recorded `tool/call`
* events without fabricating results.
* @module dsh-agent-loop/tool-calls
*/
@@ -37,10 +39,13 @@ interface GroupOutcome {
/**
* Schedule one assistant step's tool calls by their live concurrency mode.
* Started calls receive ordered results. Abort drains them, records synthetic
* results for unstarted calls, and returns with the signal still aborted after
* accepting started-call context through the caller-supplied acceptor (the
* machine stages it on its outbox for the next step boundary).
* Ordinary completion and abort commit started-call results in order. Abort
* drains them, records synthetic results for unstarted calls, and returns with
* the signal still aborted after accepting started-call context through the
* caller-supplied acceptor (the machine stages it on its outbox for the next
* step boundary). An internal scheduler failure stops new dispatches, drains
* already-started dispatches, and rejects with the first failure without
* fabricating tool results.
* The committed step's AgentLoop driver boundary supplies the initiating Agent
* that becomes each explicit {@link ToolExecutionInput.agent}.
*
@@ -78,7 +83,7 @@ export async function executeToolCalls(
let concluded = false
while (next < planned.length) {
// Commit before classifying again so registry changes affect unstarted calls.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
// oxlint-disable-next-line typescript/no-non-null-assertion -- bounded by the loop condition
const first = planned[next]!
const mode = ctx.tools.executionMode(first.exec).kind
const group = mode === 'parallel' ? planned.slice(next) : [first]
@@ -110,7 +115,8 @@ function parseArguments(raw: string): unknown {
* drain and remains for the caller's next barrier. Results and contexts commit
* in model order. Abort stops starts, drains and commits started calls, accepts
* their contexts into the owning batch, records results for skipped calls, and
* returns an aborted outcome.
* returns an aborted outcome. Scheduler failure drains dispatches without
* committing synthetic recovery results.
*/
async function runGroup(
ctx: Context,
@@ -131,6 +137,10 @@ async function runGroup(
let started = 0
let aborted: boolean = signal.aborted
let concluded = false
let schedulerFailure: { error: unknown } | undefined
const throwSchedulerFailure = (): void => {
if (schedulerFailure !== undefined) throw schedulerFailure.error
}
// `committed` advances only across contiguous model-order slots.
const commitReady = async (): Promise<void> => {
@@ -141,7 +151,7 @@ async function runGroup(
const result = slot.needsPost
? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(slot.exec, slot.result)
: ctx.tools[TOOL_REGISTRY_SCHEDULER].finish(slot.exec, slot.result)
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
// oxlint-disable-next-line typescript/no-non-null-assertion -- bounded index
appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
for (const context of result.additionalContexts ?? []) acceptContext(context)
concluded ||= result.concludesTurn === true
@@ -152,17 +162,24 @@ async function runGroup(
const inFlight = new Map<number, Promise<number>>()
const startCall = async (index: number): Promise<void> => {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
// oxlint-disable-next-line typescript/no-non-null-assertion -- bounded index
const call = group[index]!
callSeqs[index] = appendToolCall(session, turn, step, call.block)
started++
const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
throwSchedulerFailure()
switch (prepared.kind) {
case 'dispatch': {
const promise = ctx.tools[TOOL_REGISTRY_SCHEDULER].dispatch(prepared.exec).then((outcome) => {
slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
return index
})
const promise = ctx.tools[TOOL_REGISTRY_SCHEDULER].dispatch(prepared.exec).then(
(outcome) => {
slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
return index
},
(error: unknown) => {
schedulerFailure ??= { error }
return index
},
)
inFlight.set(index, promise)
break
}
@@ -181,30 +198,40 @@ async function runGroup(
const fillPool = async (): Promise<void> => {
while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {
// Re-read later modes after ordered commits so registry changes can create a barrier.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
// oxlint-disable-next-line typescript/no-non-null-assertion -- bounded by the loop condition
const nextCall = group[nextToStart]!
if (nextToStart > 0 && mode === 'parallel'
&& ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
await startCall(nextToStart)
nextToStart++
throwSchedulerFailure()
await commitReady()
throwSchedulerFailure()
// Abort may arrive while pre-execute awaits.
if (signal.aborted) aborted = true
}
}
// Ordered pre-execute may await; only dispatch/body overlaps.
// TODO: Drain every started call before rethrowing a scheduler error; tool
// bodies must not outlive the failed turn.
await fillPool()
while (inFlight.size > 0) {
const settledIndex = await Promise.race(inFlight.values())
inFlight.delete(settledIndex)
await commitReady()
// Abort may arrive while a tool or ordered commit awaits.
if (signal.aborted) aborted = true
// Ordered pre-execute may await; only dispatch/body overlaps. A scheduler
// failure stops new dispatches and reaches the turn boundary after every
// already-started dispatch settles.
try {
await fillPool()
while (inFlight.size > 0) {
const settledIndex = await Promise.race(inFlight.values())
inFlight.delete(settledIndex)
throwSchedulerFailure()
await commitReady()
throwSchedulerFailure()
// Abort may arrive while a tool or ordered commit awaits.
if (signal.aborted) aborted = true
await fillPool()
}
} catch (error: unknown) {
schedulerFailure ??= { error }
await Promise.allSettled(inFlight.values())
throw schedulerFailure.error
}
if (aborted) {