Merge remote-tracking branch 'origin/master' into xtr/react-loop-simplification
# Conflicts: # .agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.i18n.yaml # .agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md # .agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md # .agents/notes/implemented/feature/2026-07-21-tui-skill-slash-command.i18n.yaml # .agents/notes/implemented/feature/2026-07-21-tui-skill-slash-command.md # .agents/notes/implemented/feature/2026-07-21-tui-skill-slash-command.zh.md # .agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.i18n.yaml # .agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md # .agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.zh.md # docs/architecture.i18n.yaml # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/core-data-structures/core.i18n.yaml # docs/core-data-structures/core.md # docs/core-data-structures/core.zh.md # docs/defensive-patterns.i18n.yaml # packages/client/runtime/src/client/sessions/session.ts # packages/client/runtime/tests/queue-store.spec.ts # packages/context/time-context/tests/time-context.spec.ts # packages/context/workspace-context/tests/workspace-context.spec.ts # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-loop/README.i18n.yaml # packages/core/agent-loop/README.md # packages/core/agent-loop/README.zh.md # packages/core/agent-loop/src/agent.ts # packages/core/agent/README.i18n.yaml # packages/core/agent/README.md # packages/core/agent/README.zh.md # packages/core/agent/src/types.ts # packages/core/agent/tests/agent.spec.ts # packages/core/scope/src/scoped-events.generated.ts # packages/goal/command-goal/tests/command-goal.spec.ts # packages/goal/goal-session/src/index.ts # packages/goal/goal-session/tests/goal-session.spec.ts # packages/goal/goal/tests/goal.spec.ts # packages/goal/goal/tests/projection.spec.ts # packages/goal/tool-goal/tests/tool-goal.spec.ts # packages/host/apiproxy/src/api-proxy.ts # packages/host/apiproxy/src/api/events.schema.ts # packages/host/apiproxy/src/api/events.ts # packages/host/apiproxy/tests/api-proxy-workspace.spec.ts # packages/llm/llm/README.i18n.yaml # packages/llm/llm/README.zh.md # packages/llm/llm/src/index.ts # packages/pty/pty-local/tests/index.spec.ts # packages/pty/pty-local/tests/local.spec.ts # packages/pty/pty/tests/service.spec.ts # packages/pty/tool-pty/tests/loader-composition.spec.ts # packages/pty/tool-pty/tests/tools.spec.ts # packages/skill/tool-skill/tests/tool-skill.spec.ts # packages/tasks/tasks-local/tests/tasks.spec.ts # packages/ui/tui/src/index.ts # packages/ui/tui/tests/harness.ts # packages/ui/tui/tests/tui.spec.ts # scripts/gen-cordis-catalog.ts # scripts/type-equiv.manifest.json
This commit is contained in:
_Kerman
@@ -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 packages/core/agent-loop/README.md
|
||||
README.md: 33e349f8945b45bf322171d4c02b9a940a68f2c2
|
||||
README.zh.md: 65a81ea82a02ea81bc3e0a8892fd23b281477df2
|
||||
README.md: a1617a1ef871f61157e0d70a06d055168170dced
|
||||
README.zh.md: 6ba945a41e700331929dabb557802c14256921fb
|
||||
|
||||
@@ -55,7 +55,9 @@ Configured agents start automatically. A model call requires both `provider` and
|
||||
|
||||
The concrete `ReactLoopAgent`, its queued input, outbox, and run controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
|
||||
|
||||
The concrete driver routes `followup()`/`steer()`/`inject()` through one private `send()` primitive. A follow-up joins the queued FIFO and wakes the driver; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue` with the same placement, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
|
||||
The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model.
|
||||
|
||||
Every FIFO acceptance mints an `InboxItemId` and publishes `agent/inbox/enqueue` with the complete occurrence. `updateInbox()` owns the synchronous queued-item boundary: edit freezes replacement content without changing message identity or position, while remove publishes discard. Edit publishes `agent/inbox/update`; steering and claimed occurrences return `not-found`. Claim publishes `agent/inbox/dequeue` and irrevocably removes the live address before prompt admission, so a racing update cannot rewrite durable history; `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
|
||||
|
||||
### Loop lifecycle (`agent.ts`)
|
||||
|
||||
@@ -67,7 +69,7 @@ After `agent/request` returns a provider/model call config, the loop asks `ctx.l
|
||||
|
||||
Plugin failure ends the current turn, not the loop. Final adapter selection, dispatch, and iteration failures arrive from `ctx.llm` as terminal error or aborted finishes and enter `agent/request-error`; middleware, result processing, tools, and other extension failures remain thrown and close directly. Recovery receives request coordinates, immutable provider facts, the immutable retry policy captured by the prepared adapter registration, and the turn signal; the policy is absent when middleware owns an unprepared route. A handling listener returns `{ kind: 'retry' }`; an unhandled failure is terminal. AgentLoop owns one cancellation signal for the current admission or turn. An effective `cancel(cause)` clears pending work unless `keepInbox` is set and cooperatively aborts that signal; idle cancellation is a no-op. Durable `turn/end` records `aborted` for `user` and `parent`, while disposal records `disposed`; undispatched model tool calls receive synthetic `tool/call` and `ABORTED_BEFORE_DISPATCH` result pairs. The cancellation cause changes reporting, not how result context finalized after cancellation is handled. Disposal waits for signal-ignoring work before registry removal. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns the lifecycle and race contract.
|
||||
|
||||
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, and retains their finalized result context without distinguishing the cancellation cause.
|
||||
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, and retains their finalized result context without distinguishing the cancellation cause. An internal scheduler failure stops new dispatches, waits for already-started dispatches, and reaches the turn error boundary without fabricating tool results.
|
||||
|
||||
### What belongs to plugins
|
||||
|
||||
|
||||
@@ -12,11 +12,11 @@
|
||||
|
||||
创建与恢复属于同一个受回滚保护的事务：构造私有会话、实体 agent 和带作用域的上下文；等待可选 setup；进入两个注册表；依次宣告 `session/created` 和 `agent/created`；发出 `agent/session-start`；此后才启动驱动器。Setup 接收完整的带作用域 `Context`，作为受信任的同进程组合代码，并且不得驱动尚未发布的 agent。普通的类型化身份与选项输入遵循只读契约以借用方式传入；seed 事件与会话元数据会跨越持久会话边界，因此系统会验证并快照它们。可选的 `AbortSignal` 只取消加载／setup／发布，并在返回的 handle 可见前分离。
|
||||
|
||||
调用方 fiber 与 AgentLoop 提供方共同拥有 agent。`AgentFactory.createAgent(ownerCtx, options)` 与 `resume(ownerCtx, options)` 显式接收调用方所有权，而工厂为 `sessions`/`llm`/`tools`/`systemPrompt` 保留自身的依赖上下文；这样，调用方可以只注入 `agents`，而不会缩减新 agent 的服务接口。调用方卸载、handle 释放或提供方卸载都会汇合到同一个记忆化的完全停稳边界。提供方关闭会同时等待资源 teardown，以及已经观测到停用的公开 create/resume 包装层，因此依赖消失后，任何 continuation 都无法继续发布。
|
||||
调用方 fiber 与 AgentLoop 提供方共同拥有 agent。`AgentFactory.createAgent(ownerCtx, options)` 与 `resume(ownerCtx, options)` 显式接收调用方所有权，而工厂为 `sessions`/`llm`/`tools`/`systemPrompt` 保留自身的依赖上下文；这样，调用方可以只注入 `agents`，而不会缩减新 agent 的服务接口。调用方卸载、handle dispose（资源释放）或提供方卸载都会汇合到同一个记忆化的完全停稳边界。提供方关闭会同时等待资源 teardown，以及已经观测到停用的公开 create/resume 包装层，因此依赖消失后，任何 continuation 都无法继续发布。
|
||||
|
||||
每个 agent 与其会话共享一个由调用方选择的 `SessionId`，并假设它在全局唯一；意外的 UUID 冲突不属于受支持模型。两个使用同一 id 的并发操作都可以进行准备，但最终的 `enter()` 调用会裁决发布，所有失败方都会回滚各自的私有资源。每次 detach 都绑定到确切进入的对象，因此陈旧 disposer 无法移除之后出现的同 id 替代项。在同步创建通知期间请求的 detach 会等待该次分发退栈，从而保留 created/disposed 配对。Teardown 顺序为停止并 drain → 撤销作用域 → detach agent → detach 会话；私有作用域清理完成后，该 id 即可复用。普通、不可 veto 的 `agent/*` 通知通过 `agentEvents(ctx, agent)` 发出；逐步骤组装通过 `assembleContextFor(agent)` 完成。
|
||||
|
||||
- `ctx.agentLoop.create(id: SessionId, options?: AgentOptions, meta?: { cwd?: string }): Agent`：在确切共享的 agent／会话 id 下同步创建，不运行 setup，并随调用 fiber 释放。声明式配置把 `agents[].id` 视为稳定 label，通常会先生成 `${label}-session-<uuid>`，再调用此边界。应用也可以提供稳定且确切的 `sessionId`：首次使用时创建；重新挂载且持久化内容已存在时，则恢复已经实体化的历史。`resumeSessionId` 要求并加载现有的持久化 id，且与 `sessionId` 互斥。这样，默认的全新重启不会冲突，也无需保留第二个实时路由身份。
|
||||
- `ctx.agentLoop.create(id: SessionId, options?: AgentOptions, meta?: { cwd?: string }): Agent`：在确切共享的 agent／会话 id 下同步创建，不运行 setup，并随调用 fiber dispose。声明式配置把 `agents[].id` 视为稳定 label，通常会先生成 `${label}-session-<uuid>`，再调用此边界。应用也可以提供稳定且确切的 `sessionId`：首次使用时创建；重新挂载且持久化内容已存在时，则恢复已经实体化的历史。`resumeSessionId` 要求并加载现有的持久化 id，且与 `sessionId` 互斥。这样，默认的全新重启不会冲突，也无需保留第二个实时路由身份。
|
||||
|
||||
`AgentLoop` 还实现 `AgentFactory` seam，并通过 `ctx.agents.setFactory(this)` 注册自身，因此插件会通过接口 `ctx.agents` 创建／恢复 agent：
|
||||
|
||||
@@ -49,31 +49,33 @@ interface Config {
|
||||
}
|
||||
```
|
||||
|
||||
通过配置创建的 agent 会自动启动。模型调用同时需要 `provider` 和 `model`；`agent/request` 可以在分发前补齐缺失的这一对值。可选的正整数 `maxTokens` 会为每次对话请求提供初始输出上限，并记录在请求 header 中。`maxParallelToolCalls` 限制每个 agent 针对并行安全调用使用的滚动池，默认值为 `10`。`cwd` 仅应用于全新会话，而 `resumeSessionId` 保留持久化元数据。通过配置创建的 agent 使用部署 persona；编程式 setup 可以按 agent 遮蔽它。该插件提供逐 agent 的 `provider`、`model` 和 `cwd` 提示词变量；harness 身份与部署 persona 属于 `dsh-system-prompt`。
|
||||
通过配置创建的 agent 会自动启动。模型调用同时需要 `provider` 和 `model`；`agent/request` 可以在分发前补齐缺失的这一对值。可选的正数 `maxTokens` 会为每次对话请求提供初始输出上限，并记录在请求 header 中。`maxParallelToolCalls` 限制每个 agent 针对并行安全调用使用的滚动池，默认值为 `10`。`cwd` 仅应用于全新会话，而 `resumeSessionId` 保留持久化元数据。通过配置创建的 agent 使用部署 persona；编程式 setup 可以按 agent 遮蔽它。该插件为每个 agent 提供 `provider`、`model` 和 `cwd` 提示词变量；harness 身份与部署 persona 属于 `dsh-system-prompt`。
|
||||
|
||||
### 包内部实体驱动器
|
||||
|
||||
实体 `ReactLoopAgent`、其排队输入、outbox 与运行控制均为包内部实现。包根只导出插件／服务／配置契约，包导出映射不提供 `./src/*` 逃逸路径；生命周期拥有方通过 `ctx.agents` 创建 agent，而不是点名、构造或启动驱动器内部组件。一个准备完成的会话只能由一个实体驱动器认领；所有可观测行为都通过会话事件和 `agent/*` 事件分类体系发生。
|
||||
|
||||
具体驱动器通过一个私有 `send()` 原语路由 `followup()`/`steer()`/`inject()`。后续消息加入排队 FIFO 并唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`，并携带相同的路由归类；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
|
||||
统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。
|
||||
|
||||
每次 FIFO 接受项时都会铸造一个 `InboxItemId`，并通过 `agent/inbox/enqueue` 发布完整的单次入队项。`updateInbox()` 持有同步 queued 项边界：编辑会冻结替换内容，但不改变消息标识或位置；移除会发布 discard。编辑会发布 `agent/inbox/update`；steering 项和已被认领的项会返回 `not-found`。认领操作会发布 `agent/inbox/dequeue`，并在提示词接纳前不可逆地移除实时寻址标识，因此竞态中的更新无法改写持久历史；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
|
||||
|
||||
### 循环生命周期（`agent.ts`）
|
||||
|
||||
驱动器在其整个生命周期内拥有一个 agent，并在 `ctx.agents.withInitiator(agent, ...)` 内运行。包私有的编排入口点会恢复确切的 Agent，一次性派生 `agent.session`，并让操作局部的辅助函数捕获它，而不是通过浅层接口继续传递实体驱动器或每次操作的 `Session`。如果显式 `Session` 正是辅助函数的实际接口，该辅助函数会保留它；创建、持久化加载、未发布 setup、服务、worker、进程、持久化和 wire 协议则继续保留各自的显式身份。[agent 服务](../agent/README.md#initiating-agent-scope)规定传播、teardown 和分离工作规则。
|
||||
|
||||
每次提供方调用成功结束时，都会恰好追加一个 `assistant/message` 完成锚点，包括无内容调用和以 `max-tokens` 结束的调用。该锚点原样记录组装后的内容，保留确切的 chunk 溯源（流没有 chunk 时为 `[]`），并在用量可用时包含用量；空内容不会进入派生消息历史。
|
||||
每次提供方调用成功结束时，都会恰好追加一个 `assistant/message` 完成锚点，包括无内容调用和以 `max-tokens` 结束的调用。该锚点原样记录组装后的内容，保留确切的分片溯源（流没有分片时为 `[]`），并在用量可用时包含用量；空内容不会进入派生消息历史。
|
||||
|
||||
在 `agent/request` 返回提供方／模型调用配置后，循环会调用 `ctx.llm.prepareCall()`，在活跃轮次信号的控制下校验由适配器持有的推理（reasoning）强度，并填入其配置默认值。准备完成的调用会在这次异步解析、`request/header` 日志记录和最终分派期间保留同一项确切的适配器注册，因此 HMR（热模块替换）不会把某个适配器的能力解析结果与另一适配器的请求混用。生效配置会在分派前写入日志，因此监听器可以在步骤之间更改推理强度，而不会产生未记录的请求变化。没有已注册适配器的路由会保留原定配置，使 `llm/stream` 监听器可以接管并短路该请求；最终分派仍会以 `NO_ADAPTER` 拒绝未得到处理的路由。新循环实例仅在初始提供方／模型路由与日志路由完全一致时恢复上次的推理强度；路由变化会丢弃由前一模型持有的不透明 ID，并单独解析新模型。
|
||||
|
||||
插件失败会结束当前轮次，而不是结束循环。最终适配器选择、分发与迭代失败会由 `ctx.llm` 作为终止 error 或 aborted finish 返回，并进入 `agent/request-error`；middleware、结果处理、工具及其他扩展失败仍会抛出并直接关闭轮次。恢复逻辑会接收请求坐标、不可变的提供方事实、准备完成的适配器注册所捕获的不可变重试策略以及轮次信号；middleware 接管未准备路由时，该策略缺失。处理失败的监听器返回 `{ kind: 'retry' }`；未被处理的失败是终态。AgentLoop 为当前接纳或轮次拥有一个取消信号。有效的 `cancel(cause)` 在未设置 `keepInbox` 时清除待处理工作，并以协作方式中止该信号；空闲取消是空操作。持久 `turn/end` 为 `user` 和 `parent` 记录 `aborted`，dispose（资源释放）则记录 `disposed`；未分发的模型工具调用会收到合成的 `tool/call` 与 `ABORTED_BEFORE_DISPATCH` 结果对。取消原因只改变报告方式，不改变对取消后已定案结果上下文的处理。dispose 会等待忽略信号的工作完成，然后才从注册表移除。[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)规定生命周期与竞态契约。
|
||||
|
||||
在步骤内，独占调用形成屏障；并行安全调用使用有界滚动池，并在启动前重新分类。只有分发／主体会重叠。策略、持久结果和结果上下文仍保持模型顺序。中止会停止新调用，drain 已启动的结果，并保留其已定案的结果上下文，不区分取消原因。
|
||||
在步骤内，独占调用形成屏障；并行安全调用使用有界滚动池，并在启动前重新分类。只有分发／主体会重叠。策略、持久结果和结果上下文仍保持模型顺序。中止会停止新调用，drain 已启动的结果，并保留其已定案的结果上下文，不区分取消原因。内部调度器故障会停止新的分发，等待已启动的分发，然后在不虚构工具结果的情况下到达轮次错误边界。
|
||||
|
||||
### 插件负责的内容
|
||||
|
||||
超出「调用模型、运行工具、重复」的所有内容，都属于监听事件分类体系的插件：
|
||||
- 钩子与策略：相关的 `agent/*` 检查点，加上受守卫保护的 `tools/pre-execute` → `tools/execute` → `tools/post-execute` → 定义拥有的 `finalizeContent` → `tools/result` 流水线；确切事件签名与 mode 位于生成的[事件目录](../../../docs/cordis-catalog/events.md)
|
||||
- 压缩（compaction）：在 `agent/step` 上观测压力；在 `agent/request-error` 上修复规范溢出
|
||||
- 压缩（compaction）：在 `agent/step` 上观测压力；在 `agent/request-error` 上进行规范的溢出修复
|
||||
- 模型请求恢复：`dsh-llm-retry` 在 `agent/request-error` 上记录并等待按确切提供方配置的 normal 或无界退避，发出不进入表层的 `llm/retry` 状态，然后返回重试动作
|
||||
- 沙箱、权限、计划模式：使用 `tools/pre-execute` 提供可扩展的拒绝／询问，使用 `tools.guard()` 提供单调拥有方策略，使用 `tools/post-execute` 处理结果决定，并使用 `tools/result` 进行最终观测
|
||||
- subagent：在循环外部实现为 `ctx.subagents` 提供方；进程内提供方使用 `ctx.agents.create()` 和拥有的 `AgentHandle` 进行 teardown，而通用的 [`ctx.tasks`](../../tasks/tasks/) 与 [`dsh-tool-subagent`](../../subagent/tool-subagent/) 负责后台收集。
|
||||
@@ -84,7 +86,7 @@ interface Config {
|
||||
|
||||
### 完整对话请求
|
||||
|
||||
#### 模型所见
|
||||
#### 模型看到的内容
|
||||
|
||||
每个步骤中，循环会发送针对该 agent 呈现的系统提示词、可见工具 schema 和会话派生消息。它提供 `provider`、`model` 与 `cwd` 变量值，但不添加固定文案。
|
||||
|
||||
@@ -98,7 +100,7 @@ interface Config {
|
||||
|
||||
### 保留的消息历史
|
||||
|
||||
#### 模型所见
|
||||
#### 模型看到的内容
|
||||
|
||||
已接纳的 user 消息、assistant 消息、工具调用与结果、注入上下文和 steering 都会记录，并在后续步骤中发送。原始流分片、生命周期边界和其他仅写入日志的事件会被排除。
|
||||
|
||||
@@ -108,11 +110,11 @@ interface Config {
|
||||
|
||||
#### KV Cache 影响
|
||||
|
||||
普通历史增长仅追加，并保留可复用条目。接口替换或压缩会从第一个被遮蔽的历史 token 起使复用失效。
|
||||
普通历史增长仅追加，并保留可复用条目。表层替换或压缩会从第一个被遮蔽的历史 token 起使复用失效。
|
||||
|
||||
### 取消后未分发的调用
|
||||
|
||||
#### 模型所见
|
||||
#### 模型看到的内容
|
||||
|
||||
如果后续请求回放一个中止的步骤，取消所阻止分发的每个工具调用都有错误码 `ABORTED_BEFORE_DISPATCH`，结果文本为 `Error: tool call aborted before dispatch`。
|
||||
|
||||
@@ -124,7 +126,7 @@ interface Config {
|
||||
|
||||
仅追加；每个合成结果都位于可复用请求前缀之后，不会使现有 KV-cache 条目失效。
|
||||
|
||||
## 已知限制与暂缓工作
|
||||
## 已知限制与暂缓事项
|
||||
|
||||
- **分类是一元的**：安全性取决于比较同级调用或资源的调用必须保持独占（参见[设计原理](../../../.agents/notes/implemented/feature/2026-07-10-parallel-tool-call-execution.md)）。
|
||||
- **配置 label 默认每次新建**：省略 `sessionId` 会在每次启动时创建全新的 `${id}-session-<uuid>`；确切的恢复或创建行为要求显式提供稳定的 `sessionId`，而 `resumeSessionId` 要求已有持久化历史。
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -249,7 +249,7 @@ describe('config-driven session id', () => {
|
||||
const failures: unknown[] = []
|
||||
ctx.on('agent-loop/config-start-failed', () => { throw unrenderable })
|
||||
// Deliberately violate the normal Error-only rejection rule to exercise the unknown boundary.
|
||||
// eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
|
||||
// oxlint-disable-next-line typescript/prefer-promise-reject-errors
|
||||
ctx.on('agent-loop/config-start-failed', () => Promise.reject(unrenderable) as never)
|
||||
ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) })
|
||||
vi.spyOn(ctx.sessionPersistence, 'list').mockRejectedValue(unrenderable)
|
||||
|
||||
@@ -4,7 +4,7 @@ import LlmService, { createUserMessage, CallId, MessageSource, ProviderRequestId
|
||||
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { type Agent, type InboxPlacement } from '@deepseek-ai/dsh-agent'
|
||||
import AgentRegistry, { type Agent, type InboxItem, type InboxPlacement } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { ReactLoopAgent } from '../src/agent.ts'
|
||||
import InvariantService from '@deepseek-ai/dsh-invariants'
|
||||
@@ -53,6 +53,111 @@ function send(agent: Agent, text: string) {
|
||||
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
|
||||
}
|
||||
|
||||
function inboxText(item: InboxItem): string {
|
||||
return item.message.content
|
||||
.flatMap(block => block.type === 'text' ? [block.text] : [])
|
||||
.join('')
|
||||
}
|
||||
|
||||
describe('addressable inbox operations', () => {
|
||||
it('edits in place and removes exactly one queued item', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
textResponse('first reply'),
|
||||
textResponse('edited reply'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('inbox-actions'), { provider: 'mock', model: 'mock' })
|
||||
const admission = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
ctx.on('agent/prompt-submit', async (_subject, message, _signal, next) => {
|
||||
if (message.content[0]?.type === 'text' && message.content[0].text === 'first') {
|
||||
admission.resolve(undefined)
|
||||
await release.promise
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
const pending: InboxItem[] = []
|
||||
const updates: { id: string; text: string }[] = []
|
||||
const discards: string[][] = []
|
||||
ctx.on('agent/inbox/enqueue', (subject, item) => {
|
||||
if (subject === agent && inboxText(item) !== 'first') pending.push(item)
|
||||
})
|
||||
ctx.on('agent/inbox/update', (subject, item) => {
|
||||
if (subject === agent) updates.push({ id: item.id, text: inboxText(item) })
|
||||
})
|
||||
ctx.on('agent/inbox/discard', (subject, items) => {
|
||||
if (subject === agent) discards.push(items.map(item => item.id))
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
await admission.promise
|
||||
send(agent, 'remove me')
|
||||
send(agent, 'edit me')
|
||||
expect(pending.map(inboxText)).toEqual(['remove me', 'edit me'])
|
||||
|
||||
const remove = pending[0]!
|
||||
const edit = pending[1]!
|
||||
expect(agent.updateInbox(edit.id, {
|
||||
kind: 'edit',
|
||||
content: [{ type: 'text', text: 'edited' }],
|
||||
})).toBe('applied')
|
||||
expect(agent.updateInbox(remove.id, { kind: 'remove' })).toBe('applied')
|
||||
expect(updates).toEqual([{ id: edit.id, text: 'edited' }])
|
||||
expect(discards).toEqual([[remove.id]])
|
||||
|
||||
const idle = waitForIdle(ctx, agent)
|
||||
release.resolve(undefined)
|
||||
await idle
|
||||
expect(agent.session.events
|
||||
.filter(event => event.type === 'user/message')
|
||||
.map(event => event.type === 'user/message'
|
||||
? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')
|
||||
: ''))
|
||||
.toEqual(['first', 'edited'])
|
||||
expect(agent.updateInbox(edit.id, { kind: 'remove' })).toBe('not-found')
|
||||
})
|
||||
|
||||
it('does not mutate steering occurrences', async () => {
|
||||
const adapter = new MockAdapter([textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('steering-inbox-actions'), { provider: 'mock', model: 'mock' })
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const decision = Promise.withResolvers<{ kind: 'allow' }>()
|
||||
ctx.on('agent/prompt-submit', async () => {
|
||||
entered.resolve(undefined)
|
||||
return decision.promise
|
||||
})
|
||||
|
||||
const pending: InboxItem[] = []
|
||||
ctx.on('agent/inbox/enqueue', (subject, item) => {
|
||||
if (subject === agent && item.placement === 'steering') pending.push(item)
|
||||
})
|
||||
|
||||
const idle = waitForIdle(ctx, agent)
|
||||
send(agent, 'admitted prompt')
|
||||
await entered.promise
|
||||
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'keep me' }], source: { kind: 'user' } }))
|
||||
expect(pending.map(inboxText)).toEqual(['keep me'])
|
||||
|
||||
const steering = pending[0]!
|
||||
expect(agent.updateInbox(steering.id, {
|
||||
kind: 'edit',
|
||||
content: [{ type: 'text', text: 'edited' }],
|
||||
})).toBe('not-found')
|
||||
expect(agent.updateInbox(steering.id, { kind: 'remove' })).toBe('not-found')
|
||||
|
||||
decision.resolve({ kind: 'allow' })
|
||||
await idle
|
||||
expect(agent.session.events
|
||||
.filter(event => event.type === 'steering/message')
|
||||
.map(event => event.type === 'steering/message'
|
||||
? event.data.message.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')
|
||||
: ''))
|
||||
.toEqual(['keep me'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('assistant replay provenance', () => {
|
||||
it('records adapter replay state with the assembled assistant content', async () => {
|
||||
const response = textResponse('unchanged')
|
||||
@@ -502,10 +607,10 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
|
||||
const queuedSources: MessageSource[] = []
|
||||
const queuedShapes: string[][] = []
|
||||
const placements: InboxPlacement[] = []
|
||||
ctx.on('agent/inbox/enqueue', (_agent, message, placement) => {
|
||||
queuedSources.push(message.source)
|
||||
queuedShapes.push(Object.keys(message).sort())
|
||||
placements.push(placement)
|
||||
ctx.on('agent/inbox/enqueue', (_agent, item) => {
|
||||
queuedSources.push(item.message.source)
|
||||
queuedShapes.push(Object.keys(item.message).sort())
|
||||
placements.push(item.placement)
|
||||
})
|
||||
|
||||
send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
|
||||
|
||||
@@ -86,8 +86,9 @@ describe('agent/prompt-submit', () => {
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const decision = Promise.withResolvers<PromptDecision>()
|
||||
const observed: UserMessage[] = []
|
||||
ctx.on('agent/inbox/enqueue', (subject, message) => {
|
||||
ctx.on('agent/inbox/enqueue', (subject, item) => {
|
||||
if (subject !== agent) return
|
||||
const message = item.message
|
||||
expect(Object.isFrozen(message)).toBe(true)
|
||||
expect(Object.isFrozen(message.content)).toBe(true)
|
||||
expect(Object.isFrozen(message.content[0])).toBe(true)
|
||||
@@ -97,8 +98,8 @@ describe('agent/prompt-submit', () => {
|
||||
if (block?.type === 'text') block.text = 'listener mutation'
|
||||
}).toThrow()
|
||||
})
|
||||
ctx.on('agent/inbox/enqueue', (subject, message) => {
|
||||
if (subject === agent) observed.push(message)
|
||||
ctx.on('agent/inbox/enqueue', (subject, item) => {
|
||||
if (subject === agent) observed.push(item.message)
|
||||
})
|
||||
ctx.on('agent/prompt-submit', async () => {
|
||||
entered.resolve(undefined)
|
||||
@@ -240,8 +241,8 @@ describe('agent/prompt-submit', () => {
|
||||
entered.resolve(undefined)
|
||||
return decision.promise
|
||||
})
|
||||
ctx.on('agent/inbox/enqueue', (subject, _message, placement) => {
|
||||
if (subject === agent) placements.push(placement)
|
||||
ctx.on('agent/inbox/enqueue', (subject, item) => {
|
||||
if (subject === agent) placements.push(item.placement)
|
||||
})
|
||||
|
||||
const idle = waitForIdle(ctx, agent)
|
||||
|
||||
@@ -59,6 +59,20 @@ describe('agent loop', () => {
|
||||
},
|
||||
)
|
||||
|
||||
it('seeds a valid AgentOptions.maxTokens into the first model request', async () => {
|
||||
const adapter = new MockAdapter([textResponse('bounded')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(
|
||||
SessionId('valid-max-tokens'),
|
||||
{ provider: 'mock', model: 'mock', maxTokens: 256 },
|
||||
)
|
||||
|
||||
send(agent, 'use the configured output limit')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests[0]?.maxTokens).toBe(256)
|
||||
})
|
||||
|
||||
it('runs a simple turn: queued message → model → idle, with ordered events', async () => {
|
||||
const adapter = new MockAdapter([textResponse('hello there')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
@@ -9,7 +9,7 @@ import { createUserMessage, CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import ToolRegistry, { defineContentToolFixture, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import ToolRegistry, { defineContentToolFixture, TOOL_ABORTED_BEFORE_DISPATCH, TOOL_REGISTRY_SCHEDULER, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
@@ -613,3 +613,66 @@ describe('tool-call scheduler: abort handling', () => {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('tool-call scheduler: failure quiescence', () => {
|
||||
it('stops new dispatches and drains started bodies before surfacing the first failure', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'p', args: { id: '1' } },
|
||||
{ id: 'c2', name: 'p', args: { id: '2' } },
|
||||
{ id: 'c3', name: 'p', args: { id: '3' } },
|
||||
]),
|
||||
])
|
||||
const ctx = await harness(adapter, 3)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
// The registry contains expected failures as results; replace its internal
|
||||
// view only to inject the invariant violation this boundary must contain.
|
||||
const scheduler = ctx.tools[TOOL_REGISTRY_SCHEDULER]
|
||||
const prepare = scheduler.prepare.bind(scheduler)
|
||||
const dispatch = scheduler.dispatch.bind(scheduler)
|
||||
const prepareGate = Promise.withResolvers<undefined>()
|
||||
let thirdPrepareEntered = false
|
||||
scheduler.prepare = async (exec) => {
|
||||
const prepared = await prepare(exec)
|
||||
if (exec.callId === CallId('c3')) {
|
||||
thirdPrepareEntered = true
|
||||
await prepareGate.promise
|
||||
}
|
||||
return prepared
|
||||
}
|
||||
const schedulerError = new Error('scheduler exploded')
|
||||
const drainedError = new Error('sibling failed while draining')
|
||||
let rejectFirst: ((error: Error) => void) | undefined
|
||||
scheduler.dispatch = exec => exec.callId === CallId('c1')
|
||||
? new Promise((_resolve, reject) => { rejectFirst = reject })
|
||||
: dispatch(exec).then(() => { throw drainedError })
|
||||
const agent = ctx.agentLoop.create(SessionId('scheduler-failure'), { provider: 'mock', model: 'mock' })
|
||||
const errors: unknown[] = []
|
||||
ctx.on('agent/error', (subject, _turn, _step, error) => {
|
||||
if (subject === agent) errors.push(error)
|
||||
})
|
||||
let idle = false
|
||||
const idlePromise = waitForIdle(ctx, agent).then(() => { idle = true })
|
||||
|
||||
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
|
||||
await until(() => gated.started.includes('2') && thirdPrepareEntered && rejectFirst !== undefined)
|
||||
rejectFirst?.(schedulerError)
|
||||
await new Promise<void>(resolve => setImmediate(resolve))
|
||||
prepareGate.resolve(undefined)
|
||||
await new Promise<void>(resolve => setImmediate(resolve))
|
||||
|
||||
const startedBeforeDrain = [...gated.started]
|
||||
const idleBeforeDrain = idle
|
||||
const errorsBeforeDrain = [...errors]
|
||||
for (const id of gated.pending()) gated.release(id)
|
||||
await idlePromise
|
||||
|
||||
expect(startedBeforeDrain).toEqual(['2'])
|
||||
expect(idleBeforeDrain).toBe(false)
|
||||
expect(errorsBeforeDrain).toEqual([])
|
||||
expect(gated.pending()).toEqual([])
|
||||
expect(errors).toEqual([schedulerError])
|
||||
expect(errors[0]).toBe(schedulerError)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user