mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
Merge pull request #384 from deepseek-harness/codex/simplify-agent-identity-plumbing
refactor(core): simplify AgentLoop identity plumbing with ctx.agents
This commit is contained in:
@@ -120,7 +120,7 @@ Every live agent owns a scoped `agent.ctx`. Its registrations shadow globals, re
|
||||
|
||||
### Initiating Agent Scope
|
||||
|
||||
`AgentLoop` runs each process-local driver inside `ctx.agents.withInitiator()`; the [decision](rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md) owns boundary and explicit-identity rules.
|
||||
`AgentLoop` runs each driver inside `ctx.agents.withInitiator()`; private code derives `agent.session`, while other identities stay explicit ([decision](rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md)).
|
||||
|
||||
## State
|
||||
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
2026-07-15-agent-initiator-scope.md: b3c9be0be1dea29568dfcdeb0578e643734486e8
|
||||
2026-07-15-agent-initiator-scope.zh.md: 55494977b8ade0d380fa21b25171bce65a46a9fb
|
||||
2026-07-15-agent-initiator-scope.md: a9df15beb8744216e020c259934db9fdf8b28b79
|
||||
2026-07-15-agent-initiator-scope.zh.md: 4198f066ef27042bda0d12fbcaf86f143482d596
|
||||
|
||||
@@ -16,7 +16,9 @@ The mandatory `ctx.agents` service uses Node `AsyncLocalStorage` to carry the in
|
||||
|
||||
`currentInitiator()` reads optionally, `requireInitiator()` throws `no initiating agent is active`, and `withInitiator(agent, operation)` preserves the operation's exact synchronous value or Promise. `withoutInitiator(operation)` establishes a clearing boundary for work that must not inherit an Agent. Session remains derived as `agent.session`; turn, step, tool call, `signal`, model, `cwd`, sandbox, and authorization stay with their existing owners.
|
||||
|
||||
`AgentLoop` already injects `ctx.agents` and wraps each concrete driver's complete `runLoop` lifetime in `agents.withInitiator(agent, ...)`. Concurrent drivers therefore receive independent stores. A child driver's continuations carry the child, while the caller resumes in its prior store as soon as `withInitiator()` returns; active-run tracking keeps the returned Promise in the teardown drain until it settles. Creation, persistence load, and unpublished `setup(agentCtx)` remain outside the child's driver boundary: creation initiated by a parent runs under the parent identity, while `agentCtx.agent` explicitly identifies the child.
|
||||
`AgentLoop` already injects `ctx.agents` and wraps each concrete driver's complete `runLoop` lifetime in `agents.withInitiator(agent, ...)`. Its package-private loop, turn, step, and tool-call orchestration entries recover the exact Agent from `ctx.agents`, derive `agent.session` once, and let operation-local helpers capture it instead of forwarding the concrete driver or `Session` through shallow interfaces. A leaf helper keeps a narrow `Session` parameter when that is its actual interface rather than accepting a broader `Context` only for an ambient lookup.
|
||||
|
||||
Concurrent drivers receive independent stores. A child driver's continuations carry the child, while the caller resumes in its prior store as soon as `withInitiator()` returns; active-run tracking keeps the returned Promise in the teardown drain until it settles. Creation, persistence load, and unpublished `setup(agentCtx)` remain outside the child's driver boundary: creation initiated by a parent runs under the parent identity, while `agentCtx.agent` explicitly identifies the child.
|
||||
|
||||
Ambient identity does not replace explicit contracts. `ToolExecution.agent`, `AssembleContext.agent`, `GenerateOptions.sessionId`, task ownership, parent/child requests, `ctx.agent`, `agentCtx.agent`, approval and hook subjects, `cwd` selection, cancellation, worker/process messages, persistence records, and wire identity remain explicit. A remote boundary materializes the identity it needs into its typed request because ALS is process-local.
|
||||
|
||||
@@ -30,9 +32,9 @@ This decision extends the [Agent registration-scope contract](2026-07-08-agent-s
|
||||
|
||||
## Verification
|
||||
|
||||
Agent service tests pin optional and required reads, exact synchronous and cross-realm Promise identity, intrinsic Promise settlement observation, overlapping, nested, and cleared boundaries, restoration after throws or rejection, ordinary and reentrant drain ordering, and retained-reference errors. AgentLoop integration pins concurrent and nested drivers, agentless calls, AgentRegistry restart, and root teardown. Composition, module-graph, build, and runtime-closure checks keep `ctx.agents` wired through the default bundle, SDK spine, Python runtime closure, and direct AgentLoop harnesses without another provider.
|
||||
Agent service tests pin optional and required reads, exact synchronous and cross-realm Promise identity, intrinsic Promise settlement observation, overlapping, nested, and cleared boundaries, restoration after throws or rejection, ordinary and reentrant drain ordering, and retained-reference errors. AgentLoop integration pins concurrent and nested drivers, agentless calls, AgentRegistry restart, root teardown, and package-private loop and tool scheduling through the ambient lookup. Composition, module-graph, build, and runtime-closure checks keep `ctx.agents` wired through the default bundle, SDK spine, Python runtime closure, and direct AgentLoop harnesses without another provider.
|
||||
|
||||
Only a test-double host-aware transport consumes ambient identity; it derives `X-Harness-Session-Id` internally and verifies that tool schema and logged arguments contain no identity field. The service deliberately does not drain async work omitted from the Promise returned by the boundary operation; that work remains subject to its owner's explicit stop contract.
|
||||
A test-double host-aware transport derives `X-Harness-Session-Id` internally and verifies that tool schema and logged arguments contain no identity field. The service deliberately does not drain async work omitted from the Promise returned by the boundary operation; that work remains subject to its owner's explicit stop contract.
|
||||
|
||||
## Alternatives considered
|
||||
|
||||
|
||||
@@ -16,7 +16,9 @@ Harness 中存在两种有用但不同的上下文概念。Cordis `Context` 负
|
||||
|
||||
`currentInitiator()` 用于可选读取,`requireInitiator()` 抛出 `no initiating agent is active`,`withInitiator(agent, operation)` 保留操作返回的同步值或 Promise 本身。`withoutInitiator(operation)` 会建立清空边界,供不得继承 Agent 的工作使用。会话仍通过 `agent.session` 推导;轮次、步骤、工具调用、`signal`、模型、`cwd`、沙箱和授权继续由现有归属方管理。
|
||||
|
||||
`AgentLoop` 已经注入 `ctx.agents`,并用 `agents.withInitiator(agent, ...)` 包裹每个具体驱动的完整 `runLoop` 生命周期。因此,并发驱动使用彼此独立的存储。子驱动的异步延续携带子 Agent;`withInitiator()` 返回后,调用方立即恢复之前的存储,而活动运行计数仍持续跟踪返回的 Promise,直到其结束。创建、持久化加载和尚未发布的 `setup(agentCtx)` 位于子驱动边界之外:由父 Agent 发起的创建使用父身份,而 `agentCtx.agent` 显式标识子 Agent。
|
||||
`AgentLoop` 已经注入 `ctx.agents`,并用 `agents.withInitiator(agent, ...)` 包裹每个具体驱动的完整 `runLoop` 生命周期。循环、轮次、步骤和工具调用的包内私有入口从 `ctx.agents` 恢复同一个 Agent,一次推导 `agent.session`,再由操作内辅助函数捕获该值,避免在浅层接口中转发具体驱动或 `Session`。若 `Session` 本身就是底层辅助函数的实际接口,该函数会保留狭窄的 `Session` 参数,而不会只为隐式查找而接收更宽泛的 `Context`。
|
||||
|
||||
因此,并发驱动使用彼此独立的存储。子驱动的异步延续携带子 Agent;`withInitiator()` 返回后,调用方立即恢复之前的存储,而活动运行计数仍持续跟踪返回的 Promise,直到其结束。创建、持久化加载和尚未发布的 `setup(agentCtx)` 位于子驱动边界之外:由父 Agent 发起的创建使用父身份,而 `agentCtx.agent` 显式标识子 Agent。
|
||||
|
||||
隐式身份不会取代显式契约。`ToolExecution.agent`、`AssembleContext.agent`、`GenerateOptions.sessionId`、任务归属、父子请求、`ctx.agent`、`agentCtx.agent`、审批与 hook 主体、`cwd` 选择、取消、worker 和进程消息、持久化记录及协议身份都保持显式传递。远程边界会把所需身份写入类型化请求,因为 ALS 只在进程内有效。
|
||||
|
||||
@@ -30,9 +32,9 @@ Harness 中存在两种有用但不同的上下文概念。Cordis `Context` 负
|
||||
|
||||
## 验证
|
||||
|
||||
Agent 服务测试锁定可选与必需读取、同步值和跨 realm Promise 的引用身份、内建 Promise 结束状态观察、并发、嵌套及清空边界、同步抛错或 Promise 拒绝后的恢复、普通与重入排空顺序及保留引用的错误。AgentLoop 集成测试锁定并发与嵌套驱动、无 Agent 调用、AgentRegistry 重启及根 Context 销毁。组合、模块图、构建及运行时闭包检查确保默认组合包、SDK 主干、Python 运行时闭包及直接 AgentLoop harness 通过 `ctx.agents` 完成接线,无需其他提供方。
|
||||
Agent 服务测试锁定可选与必需读取、同步值和跨 realm Promise 的引用身份、内建 Promise 结束状态观察、并发、嵌套及清空边界、同步抛错或 Promise 拒绝后的恢复、普通与重入排空顺序及保留引用的错误。AgentLoop 集成测试锁定并发与嵌套驱动、无 Agent 调用、AgentRegistry 重启、根 Context 销毁,以及包内私有的循环和工具调度通过隐式查找完成。组合、模块图、构建及运行时闭包检查确保默认组合包、SDK 主干、Python 运行时闭包及直接 AgentLoop harness 通过 `ctx.agents` 完成接线,无需其他提供方。
|
||||
|
||||
只有测试替身形式的宿主感知传输层消费隐式身份;它在内部推导 `X-Harness-Session-Id`,并验证工具 schema 与记录参数都不包含身份字段。服务有意不排空边界操作所返回 Promise 之外的异步工作;这类工作仍由所属方的显式停止契约管理。
|
||||
测试替身形式的宿主感知传输层在内部推导 `X-Harness-Session-Id`,并验证工具 schema 与记录参数都不包含身份字段。服务有意不排空边界操作所返回 Promise 之外的异步工作;这类工作仍由所属方的显式停止契约管理。
|
||||
|
||||
## 考虑过的替代方案
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ These package-specific rules supplement the repo-wide [conventions](../AGENTS.md
|
||||
- **Optional services use `ctx.get(name)`.** Reserve `ctx.<name>` for declared injections; the property proxy is topology-sensitive, while strict `ctx.get` reads the global service store ([postmortem](../docs/postmortem/0001-acp-default-export-drops-inject.md)).
|
||||
- **Product-visible plugins require a non-unit REAL-composition test.** Hand-built `ctx.plugin(...)` suites are insufficient. Boot test-only `cordis.yml` through the Loader and app/process; mock only external/nondeterministic boundaries and assert model-visible, durable, or user-visible output. Keep opt-ins out of shipped defaults. [Policy](../docs/testing.md).
|
||||
- **Typed same-process service and plugin calls are contracts, not serialization boundaries.** Prefer readonly borrowed values; materialize or defensively validate only at parser/config, queued, model/tool JSON, durable/file, worker, process, or wire boundaries.
|
||||
- **Initiator-owned private chains derive, then capture.** Under `ctx.agents.withInitiator()`, recover the Agent at each orchestration entry, derive `agent.session`, and let operation-local helpers close over it. Keep `Agent` and `Session` explicit at lifecycle, session-log, service, authority, worker/process, persistence, and wire interfaces; do not widen a leaf helper from `Session` to `Context` merely to hide a parameter ([rationale](../docs/rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md)).
|
||||
- **Represent one asynchronous operation with one lifecycle controller or transaction.** Separate readiness, cancellation, disposal, reservation, or sentinel state requires an independent owner or settlement boundary; otherwise fold it while preserving rollback, callback containment, and quiescence.
|
||||
|
||||
Naming notes:
|
||||
|
||||
@@ -50,7 +50,7 @@ The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publicati
|
||||
|
||||
### Loop lifecycle (`loop.ts`)
|
||||
|
||||
The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`, so process-local asynchronous continuations can recover the initiating Agent. Creation, persistence load, and unpublished setup stay outside the driver boundary; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. The [agent service](../agent/README.md#initiating-agent-scope) owns propagation, teardown, and detached-work rules.
|
||||
The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`. Package-private orchestration entry points recover the exact Agent, derive `agent.session` once, and let operation-local helpers capture it instead of forwarding the concrete driver or per-operation `Session` through shallow interfaces. A helper keeps an explicit `Session` when that is its actual interface, while creation, persistence load, unpublished setup, services, workers, processes, persistence, and wire protocols retain their explicit identities. The [agent service](../agent/README.md#initiating-agent-scope) owns propagation, teardown, and detached-work rules.
|
||||
|
||||
Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
|
||||
|
||||
|
||||
@@ -387,7 +387,7 @@ export class ReactLoopAgent implements Agent {
|
||||
[startDriver](): void {
|
||||
if (this._status === 'disposed') return
|
||||
this.driverStarted = true
|
||||
this.done = this.loopCtx.agents.withInitiator(this, () => runLoop(this.loopCtx, this, {
|
||||
this.done = this.loopCtx.agents.withInitiator(this, () => runLoop(this.loopCtx, {
|
||||
inbox: this.#inbox,
|
||||
maxParallelToolCalls: this.maxParallelToolCalls,
|
||||
setStatus: (status) => { this.setStatus(status) },
|
||||
|
||||
@@ -19,7 +19,6 @@ import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
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'
|
||||
|
||||
/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
|
||||
@@ -95,12 +94,17 @@ export interface LoopHandle {
|
||||
|
||||
/**
|
||||
* Drive queued batches as durable turns until disposal. Plugin failures end the
|
||||
* current turn without terminating the driver.
|
||||
* @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
|
||||
* @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
|
||||
* 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)
|
||||
* through.
|
||||
* @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
|
||||
* @throws when no initiating Agent is active.
|
||||
*/
|
||||
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
|
||||
export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
|
||||
const agent = ctx.agents.requireInitiator()
|
||||
// Per-instance prefix and request-header state; conversation history remains in the session log.
|
||||
const transmission = createTransmissionLog()
|
||||
|
||||
@@ -138,7 +142,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
const turn = lastTurnNumber(session) + 1
|
||||
let terminalStopped = false
|
||||
try {
|
||||
terminalStopped = await runTurn(ctx, events, agent, handle, turn, transmission)
|
||||
terminalStopped = await runTurn(ctx, events, handle, turn, transmission)
|
||||
} catch (error: unknown) {
|
||||
// Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
|
||||
const err = toError(error)
|
||||
@@ -161,9 +165,17 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
}
|
||||
|
||||
async function runTurn(
|
||||
ctx: Context, events: AgentEventDispatch, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
|
||||
ctx: Context, events: AgentEventDispatch, handle: LoopHandle, turn: number, transmission: TransmissionLog,
|
||||
): Promise<boolean> {
|
||||
const agent = ctx.agents.requireInitiator()
|
||||
const { session } = agent
|
||||
const drainSteering = (): boolean => {
|
||||
const messages = handle.inbox.drainSteering()
|
||||
for (const message of messages) {
|
||||
session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
|
||||
}
|
||||
return messages.length > 0
|
||||
}
|
||||
|
||||
// Drain before opening the turn, but append only after `turn/start`.
|
||||
const queued = handle.inbox.drainQueued()
|
||||
@@ -262,7 +274,7 @@ async function runTurn(
|
||||
|
||||
// Steering from the previous round's continuation listeners joins before
|
||||
// the request.
|
||||
drainSteering(agent, handle.inbox, turn)
|
||||
drainSteering()
|
||||
|
||||
// The step's AbortController exists BEFORE any async pre-step work so a
|
||||
// dispose() or cancel() — in a synchronous turn-start listener or an
|
||||
@@ -336,7 +348,7 @@ async function runTurn(
|
||||
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
|
||||
try {
|
||||
stepOutcome = await runStep(
|
||||
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
|
||||
ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
|
||||
} catch (error: unknown) {
|
||||
stepOutcome = { error: toError(error) }
|
||||
} finally {
|
||||
@@ -365,7 +377,7 @@ async function runTurn(
|
||||
if (stepReason) reason = stepReason
|
||||
|
||||
// Steering that arrived during streaming/tool execution.
|
||||
const steered = drainSteering(agent, handle.inbox, turn)
|
||||
const steered = drainSteering()
|
||||
|
||||
closeStep()
|
||||
|
||||
@@ -454,15 +466,6 @@ async function runTurn(
|
||||
return terminalStopped
|
||||
}
|
||||
|
||||
/** Drain the steering queue into the session. Returns whether any arrived. */
|
||||
function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boolean {
|
||||
const messages = inbox.drainSteering()
|
||||
for (const message of messages) {
|
||||
agent.session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
|
||||
}
|
||||
return messages.length > 0
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one committed step: transform call config, log the request header, build
|
||||
* the request from the cached prefix plus the step-boundary snapshot, stream and
|
||||
@@ -472,7 +475,6 @@ function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boole
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
events: AgentEventDispatch,
|
||||
agent: ReactLoopAgent,
|
||||
handle: LoopHandle,
|
||||
turn: number,
|
||||
step: number,
|
||||
@@ -482,6 +484,7 @@ async function runStep(
|
||||
transmission: TransmissionLog,
|
||||
signal: AbortSignal,
|
||||
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
|
||||
const agent = ctx.agents.requireInitiator()
|
||||
const { session, options } = agent
|
||||
|
||||
// Seed the first request from agent options and later requests from the logged header;
|
||||
@@ -538,15 +541,47 @@ async function runStep(
|
||||
const stepError = finishError(assembler.finish)
|
||||
if (stepError) throw stepError
|
||||
|
||||
const recordAssistantMessage = (
|
||||
assembledContent: ContentBlock[],
|
||||
message: Message,
|
||||
preserveReplayState = true,
|
||||
): void => {
|
||||
session.append(
|
||||
'assistant/message',
|
||||
{
|
||||
turn,
|
||||
step,
|
||||
content: message.content,
|
||||
provenance: assistantProvenance(
|
||||
header.config,
|
||||
assembler.replayState,
|
||||
preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
|
||||
),
|
||||
...assembler.usage === undefined ? {} : { usage: assembler.usage },
|
||||
},
|
||||
{ surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
|
||||
)
|
||||
}
|
||||
|
||||
// A rejected result still records the successful provider call without retaining rejected output.
|
||||
const processStepResult = async (assembledContent: ContentBlock[], message: Message): Promise<Message> => {
|
||||
try {
|
||||
return await events.waterfall(
|
||||
'agent/step-result', turn, step, message, () => Promise.resolve(message),
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
recordAssistantMessage(assembledContent, { ...message, content: [] }, false)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
if (assembler.finish.kind === 'max-tokens') {
|
||||
const assembled = assembler.message()
|
||||
const assembledContent = structuredClone(assembled.content)
|
||||
let message: Message = withoutToolCalls(assembled)
|
||||
message = withoutToolCalls(await processStepResult(
|
||||
events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
|
||||
))
|
||||
message = withoutToolCalls(await processStepResult(assembledContent, message))
|
||||
// Preserve usage even when max-token truncation produced no content.
|
||||
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
|
||||
recordAssistantMessage(assembledContent, message)
|
||||
return { hadToolCalls: false, finish: assembler.finish }
|
||||
}
|
||||
|
||||
@@ -554,86 +589,23 @@ async function runStep(
|
||||
const assembled = assembler.message()
|
||||
const assembledContent = structuredClone(assembled.content)
|
||||
let message: Message = assembled
|
||||
message = await processStepResult(
|
||||
events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
|
||||
)
|
||||
message = await processStepResult(assembledContent, message)
|
||||
|
||||
// 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)
|
||||
recordAssistantMessage(assembledContent, message)
|
||||
|
||||
// Dispatch may overlap; policy, durable results, and result context stay model-ordered.
|
||||
const toolCalls = message.content.filter(block => block.type === 'tool-call')
|
||||
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,
|
||||
ctx, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
|
||||
)
|
||||
return { hadToolCalls: true, finish: assembler.finish }
|
||||
})
|
||||
}
|
||||
|
||||
/** Preserve successful-call accounting without retaining output that result processing rejected. */
|
||||
async function processStepResult(
|
||||
events: AgentEventDispatch,
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
config: LlmCallConfig,
|
||||
assembledContent: ContentBlock[],
|
||||
message: Message,
|
||||
assembler: BlockAssembler,
|
||||
chunkSeqs: number[],
|
||||
): Promise<Message> {
|
||||
try {
|
||||
return await events.waterfall(
|
||||
'agent/step-result', turn, step, message, () => Promise.resolve(message),
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
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 {
|
||||
|
||||
@@ -12,9 +12,7 @@
|
||||
import type { Context } from 'cordis'
|
||||
import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { HookContext } from '@deepseek-ai/dsh-agent'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import { TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
|
||||
import type { ReactLoopAgent } from './agent.ts'
|
||||
|
||||
/** One tool call after argument parsing, ready to schedule. */
|
||||
interface PlannedCall {
|
||||
@@ -33,9 +31,10 @@ interface Slot {
|
||||
* Schedule one assistant step's tool calls by their live concurrency mode.
|
||||
* Started calls receive ordered results. Abort drains them and rethrows after
|
||||
* accepting their context into the batch FIFO owned by the caller.
|
||||
* The committed step's AgentLoop driver boundary supplies the initiating Agent
|
||||
* that becomes each explicit {@link ToolExecutionInput.agent}.
|
||||
*
|
||||
* @param ctx - loop context that owns the tool registry.
|
||||
* @param agent - agent and session receiving the call lifecycle.
|
||||
* @param ctx - loop context that owns the tool registry and carries the initiating Agent.
|
||||
* @param turn - current turn number.
|
||||
* @param step - current step number.
|
||||
* @param toolCalls - assistant calls in model order.
|
||||
@@ -45,7 +44,6 @@ interface Slot {
|
||||
*/
|
||||
export async function executeToolCalls(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
toolCalls: ToolCallBlock[],
|
||||
@@ -53,7 +51,7 @@ export async function executeToolCalls(
|
||||
maxParallel: number,
|
||||
acceptContext: (context: HookContext) => void,
|
||||
): Promise<void> {
|
||||
const { session } = agent
|
||||
const agent = ctx.agents.requireInitiator()
|
||||
|
||||
// Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
|
||||
const planned: PlannedCall[] = toolCalls.map(block => ({
|
||||
@@ -74,7 +72,7 @@ export async function executeToolCalls(
|
||||
const first = planned[next]!
|
||||
const mode = ctx.tools.executionMode(first.exec).kind
|
||||
const group = mode === 'parallel' ? planned.slice(next) : [first]
|
||||
next += await runGroup(ctx, session, turn, step, group, mode, signal, maxParallel, acceptContext)
|
||||
next += await runGroup(ctx, turn, step, group, mode, signal, maxParallel, acceptContext)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -96,7 +94,6 @@ function parseArguments(raw: string): unknown {
|
||||
*/
|
||||
async function runGroup(
|
||||
ctx: Context,
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
group: PlannedCall[],
|
||||
@@ -105,6 +102,30 @@ async function runGroup(
|
||||
maxParallel: number,
|
||||
acceptContext: (context: HookContext) => void,
|
||||
): Promise<number> {
|
||||
const { session } = ctx.agents.requireInitiator()
|
||||
const appendToolCall = (block: ToolCallBlock): number => {
|
||||
return session.append('tool/call', {
|
||||
turn,
|
||||
step,
|
||||
callId: block.id,
|
||||
name: block.name,
|
||||
arguments: block.arguments,
|
||||
}).seq
|
||||
}
|
||||
const appendToolResult = (block: ToolCallBlock, result: ToolExecutionResult, callSeq: number): void => {
|
||||
session.append('tool/result', {
|
||||
turn,
|
||||
step,
|
||||
// Correlation stays with the loop's authoritative model-transcript call id;
|
||||
// registry results deliberately do not duplicate it.
|
||||
callId: block.id,
|
||||
content: result.content,
|
||||
isError: result.isError,
|
||||
...result.error ? { error: result.error } : {},
|
||||
// Persist presentation payloads so UI bridges reproduce result cards on replay.
|
||||
...result.meta !== undefined ? { meta: result.meta } : {},
|
||||
}, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
|
||||
}
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
const slots: (Slot | undefined)[] = group.map(() => undefined)
|
||||
@@ -125,7 +146,7 @@ async function runGroup(
|
||||
? 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
|
||||
appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
|
||||
appendToolResult(call!.block, result, callSeqs[committed]!)
|
||||
for (const context of result.additionalContexts ?? []) acceptContext(context)
|
||||
committed++
|
||||
}
|
||||
@@ -136,7 +157,7 @@ async function runGroup(
|
||||
const startCall = async (index: number): Promise<void> => {
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
|
||||
const call = group[index]!
|
||||
callSeqs[index] = appendToolCall(session, turn, step, call.block)
|
||||
callSeqs[index] = appendToolCall(call.block)
|
||||
started++
|
||||
const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
|
||||
switch (prepared.kind) {
|
||||
@@ -198,32 +219,3 @@ async function runGroup(
|
||||
if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
|
||||
return started
|
||||
}
|
||||
|
||||
/** Append a started call and return its provenance sequence. */
|
||||
function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
|
||||
const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
|
||||
return event.seq
|
||||
}
|
||||
|
||||
/** Append a model-ordered result linked to its call event. */
|
||||
function appendToolResult(
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
block: ToolCallBlock,
|
||||
result: ToolExecutionResult,
|
||||
callSeq: number,
|
||||
): void {
|
||||
session.append('tool/result', {
|
||||
turn, step,
|
||||
// Correlation stays with the loop's authoritative model-transcript call id;
|
||||
// registry results deliberately do not duplicate it.
|
||||
callId: block.id,
|
||||
content: result.content,
|
||||
isError: result.isError,
|
||||
...result.error ? { error: result.error } : {},
|
||||
// The tool's private presentation payload (e.g. a result-time diff),
|
||||
// persisted so a UI bridge reproduces the card on replay.
|
||||
...result.meta !== undefined ? { meta: result.meta } : {},
|
||||
}, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
|
||||
}
|
||||
|
||||
@@ -6,6 +6,6 @@
|
||||
"docs/defensive-patterns.md": 550,
|
||||
"docs/testing.md": 960,
|
||||
"examples/AGENTS.md": 310,
|
||||
"packages/AGENTS.md": 290,
|
||||
"packages/AGENTS.md": 370,
|
||||
"packages/README.md": 760
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user