Merge latest origin/master into compact-post-step-overflow-recovery

# Conflicts: # docs/agent-lifecycle.md # docs/architecture.md # docs/cookbook/extension-cookbook.i18n.yaml # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/event-producer-consumer.md # docs/rfc/INDEX.md # docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md # docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.i18n.yaml # docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.md # docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.zh.md # docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md # docs/rfc/implemented/feature/2026-07-07-session-prefix.md # packages/compact/compact-basic/README.md # packages/compact/compact-basic/src/config.ts # packages/compact/compact-basic/src/index.ts # packages/compact/compact-basic/src/summarizer.ts # packages/compact/compact-basic/tests/compact-basic.spec.ts # packages/compact/compact-basic/tests/compact-loop-repro.spec.ts # packages/compact/compact/README.md # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-loop/src/loop.ts # packages/core/agent-loop/tests/cancel.spec.ts # packages/llm/llm-deepseek/src/adapter.ts # packages/llm/llm-pi-ai/README.md # packages/llm/llm-pi-ai/src/stream.ts # packages/llm/llm-pi-ai/tests/convert.spec.ts # packages/llm/llm/README.md # packages/llm/llm/src/index.ts # packages/llm/llm/tests/service.spec.ts # scripts/gen-doc-graphs.ts
2026-08-15 21:04:50 +00:00 · 2026-07-19 12:06:23 +08:00
parent 231aeabe55 383a305bd4
commit 7be78c7d4c
814 changed files with 42348 additions and 10067 deletions
--- a/packages/core/agent-loop/src/loop.ts
+++ b/packages/core/agent-loop/src/loop.ts
@@ -6,7 +6,8 @@
 */

 import type { Context } from 'cordis'
-import type { FinishReason, GenerateOptions, LlmCallConfig, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
+import { isDeepStrictEqual } from 'node:util'
 import { BlockAssembler, HarnessError, deepFreeze, isLlmAdapterFailure } 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'
@@ -16,7 +17,8 @@ import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
 import type { TransmissionLog } from './request-log.ts'
 import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
-import type { ToolExecutionResult } from '@deepseek-ai/dsh-tools'
+import type {} from '@deepseek-ai/dsh-tools'
+import { executeToolCalls } from './tool-calls.ts'
 import type { ReactLoopAgent } from './agent.ts'
 import type { Inbox } from './inbox.ts'

@@ -60,15 +62,6 @@ function errorData(err: RequestError): { message: string; code?: string } {
  return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
 }

-/** Build the durable result for a model-requested call skipped after cancellation. */
-function skippedToolResult(): ToolExecutionResult {
-  return {
-    content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
-    isError: true,
-    error: { name: 'AbortError', code: 'ABORTED' },
-  }
-}
-
 /** Map a successful max-token finish onto the turn reason; other successful finishes add nothing. */
 function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
  switch (finish.kind) {
@@ -86,6 +79,8 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
 export interface LoopHandle {
  /** Native-private agent inbox handed to the driver only at internal startup. */
  readonly inbox: Inbox
+  /** Maximum parallel-safe calls allowed in one step. */
+  readonly maxParallelToolCalls: number
  setStatus(status: 'idle' | 'running'): void
  setAbort(controller: AbortController | undefined): void
  /** Resolves when the agent is disposed — unblocks the idle wait. */
@@ -99,6 +94,8 @@ export interface LoopHandle {
  clearCancel(): void
  /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
  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>
 }

 /**
@@ -249,10 +246,15 @@ async function runTurn(
      // `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
      const content = decision.content ?? message.content
      session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
-      // `allow.additionalContext` is a SEPARATE context/message the next request
-      // also sees. The turn is open, so inject() appends it into THIS turn.
-      if (decision.additionalContext) {
-        agent.inject(decision.additionalContext.content, { source: decision.additionalContext.source })
+      // 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 ?? []) {
+        agent.inject(context.content, {
+          source: context.source,
+          ...context.envelope !== undefined ? { envelope: context.envelope } : {},
+          ...context.meta !== undefined ? { meta: context.meta } : {},
+        })
      }
    }

@@ -343,7 +345,7 @@ async function runTurn(
        | { error: RequestError }
      try {
        stepOutcome = await runStep(
-          ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
+          ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
      } catch (error: unknown) {
        if (isLlmAdapterFailure(error)) {
          stepOutcome = { requestError: error }
@@ -563,6 +565,7 @@ async function runStep(
  ctx: Context,
  events: AgentEventDispatch,
  agent: ReactLoopAgent,
+  handle: LoopHandle,
  turn: number,
  step: number,
  assembly: PromptAssembly,
@@ -578,12 +581,12 @@ async function runStep(
  const seedConfig: LlmCallConfig = deepFreeze(structuredClone(transmission.loggedHeader
    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- loggedHeader ⟹ a snapshot is in the log
    ? session.requestHeader()!.config
-    : { model: options.model ?? '' }))
+    : { provider: options.provider ?? '', model: options.model ?? '' }))

  // Listener replacements are recorded in the request header before dispatch.
  const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
-  if (!config.model) {
-    throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
+  if (!config.provider || !config.model) {
+    throw new Error(`agent "${agent.id}" has no provider/model: set AgentOptions.provider and AgentOptions.model or supply both via the agent/request waterfall`)
  }

  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
@@ -600,6 +603,7 @@ async function runStep(

  // Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
  const request: GenerateOptions = deepFreeze({
+    provider: header.config.provider,
    model: header.config.model,
    messages: [...header.messagePrefix ?? [], ...boundaryMessages],
    ...header.system !== undefined ? { system: header.system } : {},
@@ -627,95 +631,38 @@ async function runStep(
  if (stepError) throw new TerminalModelRequestFailure(stepError)

  if (assembler.finish.kind === 'max-tokens') {
-    let message: Message = withoutToolCalls(assembler.message())
+    const assembled = assembler.message()
+    const assembledContent = structuredClone(assembled.content)
+    let message: Message = withoutToolCalls(assembled)
    message = withoutToolCalls(await processStepResult(
-      events, session, turn, step, message, assembler.usage, chunkSeqs,
+      events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
    ))
-    appendAssistantCompletion(
-      session, turn, step, message.content, assembler.usage, chunkSeqs,
-    )
+    // Preserve usage even when max-token truncation produced no content.
+    recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
    return { hadToolCalls: false, finish: assembler.finish }
  }

  // Record the post-waterfall message that tool dispatch uses.
-  let message: Message = assembler.message()
+  const assembled = assembler.message()
+  const assembledContent = structuredClone(assembled.content)
+  let message: Message = assembled
  message = await processStepResult(
-    events, session, turn, step, message, assembler.usage, chunkSeqs,
+    events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
  )

-  // Every successful call records its completion anchor. A present empty
-  // source set means the provider stream was known to contain no chunks;
-  // omission remains the conservative legacy/unrecorded representation.
-  appendAssistantCompletion(
-    session, turn, step, message.content, assembler.usage, chunkSeqs,
-  )
+  // Every successful call records its completion anchor, including explicit
+  // empty chunk provenance for a contentless, usage-less provider response.
+  recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)

-  // Tool execution stays sequential; cancellation latches synthetic results for
-  // every remaining call while preserving one complete result batch.
+  // Dispatch may overlap; policy, durable results, and result context stay model-ordered.
  const toolCalls = message.content.filter(block => block.type === 'tool-call')
-  // Buffer context until all results are appended to preserve call/result adjacency.
-  const pendingContext: HookContext[] = []
-  let aborted = signal.aborted
-  for (const call of toolCalls) {
-    const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
-    let result: ToolExecutionResult
-    if (aborted || signal.aborted) {
-      aborted = true
-      result = skippedToolResult()
-    } else {
-      let parsedArguments: unknown
-      try {
-        parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
-      } catch {
-        parsedArguments = call.arguments
-      }
-      // TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
-      // see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
-      result = await ctx.tools.execute({
-        callId: call.id,
-        name: call.name,
-        arguments: parsedArguments,
-        agent,
-        signal,
-      })
-    }
-    session.append('tool/result', {
-      turn, step,
-      // Correlation comes from the immutable execution input; the result does
-      // not duplicate this authoritative transcript identity.
-      callId: call.id,
-      content: result.content,
-      isError: result.isError,
-      ...result.error ? { error: result.error } : {},
-      // Persist tool-owned presentation data for replay.
-      ...result.meta !== undefined ? { meta: result.meta } : {},
-    }, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
-    if (result.additionalContext) pendingContext.push(result.additionalContext)
-    if (signal.aborted) aborted = true
-  }
-
-  // Append buffered context after the complete result batch.
-  for (const context of pendingContext) {
-    agent.inject(context.content, { source: context.source })
-  }
-
-  return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
-}
-
-/** Append the single durable completion anchor for one successful provider call. */
-function appendAssistantCompletion(
-  session: Session,
-  turn: number,
-  step: number,
-  content: Message['content'],
-  usage: TokenUsage | undefined,
-  sourceEventSeqs: number[],
-): void {
-  session.append(
-    'assistant/message',
-    { turn, step, content, ...(usage ? { usage } : {}) },
-    { surfaceOp: 'append', sourceEventSeqs },
-  )
+  if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
+  return handle.withToolBatch(async (acceptContext) => {
+    await executeToolCalls(
+      ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
+    )
+    return { hadToolCalls: true, finish: assembler.finish }
+  })
 }

 /** Preserve successful-call accounting without retaining output that result processing rejected. */
@@ -724,20 +671,70 @@ async function processStepResult(
  session: Session,
  turn: number,
  step: number,
+  config: LlmCallConfig,
+  assembledContent: ContentBlock[],
  message: Message,
-  usage: TokenUsage | undefined,
-  sourceEventSeqs: number[],
+  assembler: BlockAssembler,
+  chunkSeqs: number[],
 ): Promise<Message> {
  try {
    return await events.waterfall(
      'agent/step-result', turn, step, message, () => Promise.resolve(message),
    )
  } catch (error: unknown) {
-    appendAssistantCompletion(session, turn, step, [], usage, sourceEventSeqs)
+    recordAssistantMessage(
+      session,
+      turn,
+      step,
+      config,
+      assembledContent,
+      { ...message, content: [] },
+      assembler,
+      chunkSeqs,
+      false,
+    )
    throw error
  }
 }

+/** Record one content-or-usage assistant message with replay-safe provenance. */
+function recordAssistantMessage(
+  session: Session,
+  turn: number,
+  step: number,
+  config: LlmCallConfig,
+  assembledContent: ContentBlock[],
+  message: Message,
+  assembler: BlockAssembler,
+  chunkSeqs: number[],
+  preserveReplayState = true,
+): void {
+  session.append(
+    'assistant/message',
+    {
+      turn,
+      step,
+      content: message.content,
+      provenance: assistantProvenance(
+        config,
+        assembler.replayState,
+        preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
+      ),
+      ...assembler.usage === undefined ? {} : { usage: assembler.usage },
+    },
+    { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
+  )
+}
+
+/** Build durable assistant provenance, dropping replay state after any content rewrite. */
+function assistantProvenance(config: LlmCallConfig, replayState: unknown, contentUnchanged: boolean): NonNullable<Message['provenance']> {
+  return {
+    provider: config.provider,
+    model: config.model,
+    ...contentUnchanged && replayState !== undefined ? { replayState } : {},
+  }
+}
+
 function withoutToolCalls(message: Message): Message {
  return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
 }