每个插件都对着 Agent 句柄编程;loop 可整体替换 —— 本包零循环依赖。
ctx.agents + ctx.agent(DX accessor)
9 个 src 文件 · 1757 行
AgentRegistry 服务类 + agent-invariant 函数插件
下一章:agent-loop
分层原则(README:5):"Every plugin (UI, hooks, orchestrators) programs against the Agent handle defined here —
it has zero loop dependency, so the loop is swappable." 创建能力也留在本包接口上(AgentFactory),
消费者只依赖 ctx.agents 而不依赖具体 loop 包。
| 项 | 内容 |
|---|---|
| AgentRegistry | super(ctx, 'agents');注入 ctx.inject(['typert'], ...) 注册 typert lookup(agent ↔ agentId);无 Config |
| ctx.agent | DX accessor,默认 undefined;Agent.ctx 以自有属性遮蔽它(ctx.extend({ agent: this })) |
| 伴随插件 | agent-invariant:inject ['invariants'],把安装器注册进 ctx.invariants;根服务不隐式加载诊断 |
| 导出面 | runtime-types / types / inbox / consumed-work / model-selection 全 re-export;dispatch 导出 agentCarrier / agentEvents / assembleContextFor / emitAgentEvent |
interface Agent { readonly id: SessionId // 与 session 共享单一身份 readonly options: AgentOptions // provider 路由 + model + maxTokens readonly session: Session // 被驱动的 live session;日志是持久化真相 readonly inbox: Inbox // 持久化待处理工作的 agent 专属投影 readonly status: AgentStatus // 'idle' | 'running',每次转换镜像到 agent/status readonly ctx: Context // agent 作用域上下文(以 agent 为 key 的 dsh-scope) cancel(cause, options?): void // 清 inbox(除非 keepInbox)+ 中止活动 whenIdle(): Promise<void> // 整个 agent 到达静止 runMaintenance(task): Promise<T> // 真空闲阶段跑一次非 turn 维护任务 send(message, target, wakeup): void // 路由到 inbox 边界,可选唤醒 followup(message): void // next-turn 排队 + 唤醒 steer(message): void // next-step 排队 + 唤醒 inject(message): void // next-step 排队、不唤醒 }
interface AgentFactory { createAgent(ownerCtx, options): Promise<AgentHandle> resume(ownerCtx, options): Promise<AgentHandle> } interface CreateAgentOptions { sessionId; meta?; seed?; agentOptions?; signal?; // signal 仅创建期取消 setup?: (agentCtx) => AgentSetupCommit | Promise<...> // 未发布阶段组合 scoped world } interface AgentHandle { agent; dispose(): Promise<void> }
agent/session-start 之前执行。setup 只组合,不驱动:所有注册(工具、章节、变量、restrict、监听器、await 的子插件)
必须在发布与首次装配之前存在;成功后才同步调用可选的 setupCommit?.commit(),然后 publish。
scope 规则:所有事件都以 Scoped<Agent> 为 this —— agent-scoped 监听器只收到自己的 agent。
| 事件 | mode | 触发时机 |
|---|---|---|
agent/created | emit | 注册时;同步监听器失败否决发布,返回 promise 拒绝仅报告 |
agent/disposed | emit | 注销时;loop 在驱动静止后、session 分离前发 |
agent/status | emit | idle ⟷ running 转换时(仅变化时发) |
agent/inbox/inserted · claimed · discarded | emit | inbox 插入 / 认领(每条消息一次)/ 删除或 clear |
agent/session-start | emit | 发布时一次,第一个 turn 之前;通知非否决 |
agent/pre-step | waterfall | 每个 step 提议时;next() 保留当前 messages,返回 {kind:'enter', messages} 或 {kind:'reject'} |
agent/request | waterfall | 每次模型调用前;next() 得 LlmCallConfig,可整体替换;不能改 messages |
agent/request-error | waterfall | 流 finish 为 error/aborted 时;返回 {kind:'retry'} 且不调 next = 自拥恢复;默认 undefined = 终局 |
agent/turn-stopping | serial | turn 即将关闭时;先 await 再提交边界;监听器可 steer() 续跑 |
agent/error | emit | throwError 时;error 原样 |
session/event(turn/* step/* assistant/* user/message tool/*);
agent/* 只是进程内协调词汇,不可作为持久化事实。
注册表的 create 经 Reflect.apply(target.createAgent, getTraceable(ownerCtx, target), ...) 把工厂所有权跟随 caller;
并发同 id 创建时 enter 的权威检查只有一个能发布,失败者全量回滚自己的私有 scope/session/driver(index.ts:480-482)。
两条有序 pending 列表:next-turn 与 next-step。每次 mutation 先
this.session.append('agent/inbox/spliced', splice) 提交持久化、再改内存投影 —— 同步 session/event 观察者读到的仍是 splice 之前的列表。
重放时从 session.header.seedLength ?? 0 起重放全部 splice,损坏即抛。
agent/inbox/claimedoutcome:'canceled' splices + discarded 事件;{keepInbox:true} 只中止活动 turn
agentEvents(ctx, agent, carrier?) 返回 fused dispatcher:emit 不走 Cordis 原生 emit —— 自己解析过滤后的回调集、
逐个 try/catch(通知不能否决生命周期);payload 先 spread 再注入 agent(结构上杜绝调用者覆盖 subject)。
agentCarrier(agent) = scopeTarget(agent, agent) 静态路由对象,driver 构造一次复用,热路径零分配。
installModelSelection(agentCtx, selection) 注册两个监听器:system-prompt/assemble 在 await next() 后把
selection.current 存入 selection.assembled(装配时快照)并写进 assembly 变量;
agent/request 把快照的 provider/model 套回 config(effort 缺失时清除继承的 effort)。快照语义保证并发切换作用于下一个 step,不撕裂两个表面。
agent/status 无重复转换:WeakMap 记每个 agent 的 lastStatus,收到相同 status 即 fail(no-op transition 是 bug)。
agent.ctx(如 setup 内 agentCtx.on('agent/request', ...))只影响该 agent;
注册在全局 ctx 影响所有 agent;scope 过滤由 carrier 决定。
Agent 接口(runtime-types.ts:64-144)与自己的驱动;注册 ctx.agents.register(agent) 或高级路径 enter(agent, owner) + announce(agent)(自建 agent 需自己拥有 driver-ordering 契约)AgentFactory(createAgent / resume),ctx.agents.setFactory(this)(重复注册抛错)no agent factory registered (load an agent-loop plugin)ctx.agents.withInitiator(agent, () => this.kick()) 获得发起者传播(AsyncLocalStorage 进程内因果归因)ctx.agents.register() 一条路即可| 目标 | 机制 | 要点 |
|---|---|---|
| 改 messages(模型可见内容) | agent/pre-step 瀑布 | 返回自己的 {kind:'enter', messages:[...]} 或 {kind:'reject'};拒绝使 turn 以 blocked 关闭;模型可见内容必须走日志渠道 |
| 改调用配置(provider/model/采样) | agent/request 瀑布 | await next() 得到种子 config,返回替换的 LlmCallConfig;不能在此改 messages |
| 失败恢复 | agent/request-error 瀑布 | 返回 {kind:'retry'} 且不调 next = 自拥恢复;调 next 委托;未处理 = 终局(参考 dsh-llm-retry) |
waterfall 必须 next():不调 next() 直接返回值 = 短路下游并接管该层;request-error 例外:{kind:'retry'} 不调 next 是设计。
agent/turn-stopping(serial)在 turn/end 提交之前 await;想续跑就 agent.steer(...);无内置 turn 预算 —— 防失控循环的策略就在此挂 cancelagent.inject(message) 排队 next-step 不唤醒;模型可见 ⟺ 日志可重建(注入最终以 user/message 持久化);创建期种子监听 agent/session-startagent/status(状态流)/ agent/created|disposed(生命周期)/ agent/inbox/*(消息级);查询"已消费工作的结局"用 foldConsumedWork(session.events)(consumed-work.ts:68)Agent.id === Agent.session.id 由 enter 强制(index.ts:476-478);会话持久化身份 = 代理身份agent/session-start 不能 gate 启动(非否决通知);发布前完成的异步组合放 setup 事务