mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-11-09 20:07:43 +08:00
Merge remote-tracking branch 'origin/master' into worktree/dynamic-tool-activation-pr-64796e
This commit is contained in:
Yichen Jiang
@@ -21,8 +21,8 @@
|
||||
en: 20ef882e44b06a10
|
||||
zh: c10c3934ed5570e9
|
||||
/deepseek-harness-architecture/events:
|
||||
en: e89802f066e4955d
|
||||
zh: ea2fb4bef94e086d
|
||||
en: 693407efb233f961
|
||||
zh: 37b9dd0677d4405c
|
||||
/deepseek-harness-architecture/turn-flow:
|
||||
en: 1a79a70b62c04cdb
|
||||
zh: c86d765e21430405
|
||||
|
||||
@@ -72,7 +72,7 @@ Events are the extension points, and picking the right domain is the first decis
|
||||
- **Agent events** (`agent/*`) carry a live `Agent`: inbox, step, status, request, validation, continuation. Use one to observe or intercept work in flight.
|
||||
- **Capability events** attach policy and adapters to a seam (`fs/*`, `tools/*`, `telemetry/*`) without importing the loop.
|
||||
|
||||
AgentLoop awaits serial `agent/created` before queued work and rolls back failed initialization. Unload retains resources through Agent teardown; [agent-loop](../packages/core/agent-loop/README.md#understand-the-implementation) defines ordering.
|
||||
AgentLoop initializes Agents with rollback and awaits serial `agent/created` before queued work. Unload retains resources through Agent teardown; [agent-loop](../packages/core/agent-loop/README.md#understand-the-implementation) defines ordering.
|
||||
|
||||
The [event map](event-producer-consumer.md) lists every event's producers and consumers.
|
||||
|
||||
|
||||
@@ -74,7 +74,7 @@ Electron 使用 Electron Node 模式启动私有 Desktop Host。Host 调用共
|
||||
- **Agent 事件**（`agent/*`）携带活跃 `Agent`：inbox、步骤、状态、请求、验证、续跑。要观察或拦截进行中的工作时，使用它。
|
||||
- **能力事件**无需导入循环即可向某个 seam（`fs/*`、`tools/*`、`telemetry/*`）附加策略和适配器。
|
||||
|
||||
AgentLoop 在启动已排队工作前等待串行 `agent/created`，初始化失败则回滚。卸载在 Agent teardown 期间保留所需资源；[agent-loop](../packages/core/agent-loop/README.zh.md#understand-the-implementation)定义顺序。
|
||||
AgentLoop 初始化 Agent，失败则回滚，并在启动已排队工作前等待串行 `agent/created`。卸载在 Agent teardown 期间保留所需资源；[agent-loop](../packages/core/agent-loop/README.zh.md#understand-the-implementation)定义顺序。
|
||||
|
||||
[事件映射](event-producer-consumer.zh.md)列出每个事件的生产方与消费方。
|
||||
|
||||
|
||||
@@ -96,7 +96,7 @@ export interface Config {
|
||||
|
||||
- `inject`: `agents` · `sessions` · `llm` · `tools` · `systemPrompt` · `sessionProjections`
|
||||
- `refs`: [`AgentOptions`](subsystems/core.md) · [`SessionId`](subsystems/core.md) · `Volatile` (`@deepseek-ai/cosmokit`)
|
||||
- `source`: [`packages/core/agent-loop/src/index.ts:251`](../packages/core/agent-loop/src/index.ts)
|
||||
- `source`: [`packages/core/agent-loop/src/index.ts:253`](../packages/core/agent-loop/src/index.ts)
|
||||
|
||||
```ts config-catalog
|
||||
/** Agent-loop plugin configuration. */
|
||||
|
||||
@@ -98,7 +98,7 @@ export interface Config {
|
||||
|
||||
- `inject`: `agents` · `sessions` · `llm` · `tools` · `systemPrompt` · `sessionProjections`
|
||||
- `refs`: [`AgentOptions`](subsystems/core.zh.md) · [`SessionId`](subsystems/core.zh.md) · `Volatile` (`@deepseek-ai/cosmokit`)
|
||||
- `source`: [`packages/core/agent-loop/src/index.ts:251`](../packages/core/agent-loop/src/index.ts)
|
||||
- `source`: [`packages/core/agent-loop/src/index.ts:253`](../packages/core/agent-loop/src/index.ts)
|
||||
|
||||
```ts config-catalog
|
||||
/** Agent-loop plugin configuration. */
|
||||
|
||||
@@ -8,7 +8,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
<!-- BEGIN GENERATED event-producer-consumer:events -->
|
||||
| Event | Mode | Declared in | Dispatchers | Listeners |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:197`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | - |
|
||||
| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:199`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | - |
|
||||
| `agent-preset/selected` | `emit` | [`packages/preset/agent-preset-registry/src/types.ts:66`](../packages/preset/agent-preset-registry/src/types.ts) | [`agent-preset-registry`](../packages/preset/agent-preset-registry) (`emit`) | [`cua`](../packages/experimental/cua), `remotes` |
|
||||
| `agent/assistant-stream` | `emit` | [`packages/core/agent/src/runtime-types.ts:399`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`headless`](../packages/bundle/headless), `session-controller` |
|
||||
| `agent/created` | `serial` | [`packages/core/agent/src/runtime-types.ts:290`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`serial`) | `agent-team`, `browser-use-runtime`, `claude-code-mods`, [`cua`](../packages/experimental/cua), [`file-reference-local`](../packages/context/file-reference-local), [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/experimental/hooks-claude-code), [`hooks-codex`](../packages/experimental/hooks-codex), [`loader-smoke`](../packages/test-support/loader-smoke), [`permission-presets`](../packages/interaction/permission-presets), `session-controller`, `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
<!-- BEGIN GENERATED event-producer-consumer:events -->
|
||||
| Event | Mode | Declared in | Dispatchers | Listeners |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:197`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | - |
|
||||
| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:199`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | - |
|
||||
| `agent-preset/selected` | `emit` | [`packages/preset/agent-preset-registry/src/types.ts:66`](../packages/preset/agent-preset-registry/src/types.ts) | [`agent-preset-registry`](../packages/preset/agent-preset-registry) (`emit`) | [`cua`](../packages/experimental/cua), `remotes` |
|
||||
| `agent/assistant-stream` | `emit` | [`packages/core/agent/src/runtime-types.ts:399`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`headless`](../packages/bundle/headless), `session-controller` |
|
||||
| `agent/created` | `serial` | [`packages/core/agent/src/runtime-types.ts:290`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`serial`) | `agent-team`, `browser-use-runtime`, `claude-code-mods`, [`cua`](../packages/experimental/cua), [`file-reference-local`](../packages/context/file-reference-local), [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/experimental/hooks-claude-code), [`hooks-codex`](../packages/experimental/hooks-codex), [`loader-smoke`](../packages/test-support/loader-smoke), [`permission-presets`](../packages/interaction/permission-presets), `session-controller`, `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
|
||||
@@ -39,8 +39,8 @@
|
||||
en: ec2bf804bd1a5439
|
||||
zh: 316e767507e859ef
|
||||
/deepseek-ai-dsh-agent-loop/understand-the-implementation/creation-and-teardown:
|
||||
en: 1127b2c51e3f2d1a
|
||||
zh: 15a24452c18f0686
|
||||
en: 25be2920f1de4f6f
|
||||
zh: 279c60a6e1b1767a
|
||||
/deepseek-ai-dsh-agent-loop/understand-the-implementation/persistence-integration:
|
||||
en: 3b48a480036bd5df
|
||||
zh: 28cea2036b0dccee
|
||||
|
||||
@@ -107,7 +107,7 @@ The loop deep-freezes each derived message identity on its first request and reu
|
||||
|
||||
### Creation and teardown
|
||||
|
||||
Creation is one rollback-covered transaction: construct a session, concrete agent, and scoped context; store constructor history, enter the SessionStore, and complete `session/created`; await optional setup and its commit; enter the AgentRegistry; await serial `agent/created` listeners; then release queued input. A caller creating a runtime child sets `options.parentAgent`; the caller Context separately owns the transaction and live handle. Setup, commit, or listener failure and owner disposal roll back the prepared resources. The Session is writable during setup, and every removal emits `session/disposed`, including setup failure before Agent creation is announced. Failed creation preserves restored inbox input and any events already stored. Announcements already delivered remain observable and receive matching disposal notifications. Teardown stops and drains the driver, unwinds the scope, closes the session's write path, detaches the agent, then detaches the session. Every detach binds the exact entered object, so a stale disposer cannot remove a later same-id replacement.
|
||||
Fresh creation and resume share one rollback-covered initialization: construct the concrete agent and scoped context around the acquired Session; store constructor history, enter the SessionStore, and complete `session/created`; await optional setup and its commit; enter the AgentRegistry; await serial `agent/created` listeners; then release queued input. A caller creating a runtime child sets `options.parentAgent`; the caller Context separately owns the transaction and live handle. Setup, commit, or listener failure and owner disposal roll back the prepared resources. The Session is writable during setup, and every removal emits `session/disposed`, including setup failure before Agent creation is announced. Failed initialization preserves restored inbox input and any events already stored; it rejects only after releasing its scope and acquired write handle. Cancellation while persistence is still creating a write handle can reject earlier; that handle closes when creation settles. Announcements already delivered remain observable and receive matching disposal notifications. Teardown stops and drains the driver, unwinds the scope, closes the session's write path, detaches the agent, then detaches the session. Every detach binds the exact entered object, so a stale disposer cannot remove a later same-id replacement.
|
||||
|
||||
The loop awaits inbox and turn-boundary registration before publishing its factory and passes those handles into synchronous Agent construction. During factory unload, global registrations may disappear before teardown finishes; retained handles remain readable through scope cleanup, including cancellation from scoped abort listeners. Caller unload, factory unload, and handle disposal await the same teardown.
|
||||
|
||||
|
||||
@@ -107,7 +107,7 @@ const handle = await ctx.agents.create({
|
||||
|
||||
### 创建与拆除
|
||||
|
||||
创建是同一个受回滚保护的事务：构造会话、具象 agent 与带作用域上下文；存储构造历史、进入 SessionStore 并完成 `session/created`；等待可选 setup 及其 commit；进入 AgentRegistry；等待串行 `agent/created` 监听器；随后释放已排队输入。创建运行时子 Agent 的调用方设置 `options.parentAgent`；调用方 Context 则单独拥有事务和存活句柄。Setup、commit、监听器失败或所有者 dispose 都会回滚已准备的资源。setup 期间 Session 可写；每次移除都会发出 `session/disposed`，包括 Agent 创建宣告前的 setup 失败。创建失败保留恢复出的收件箱输入及已存储事件。已送达的宣告仍可被观察，并有配对的销毁通知。Teardown 停止并排空驱动器、撤销作用域、关闭会话写路径、detach agent，再 detach 会话。每次 detach 都绑定到确切进入的对象，因此陈旧 disposer 无法移除之后出现的同 id 替代项。
|
||||
新建和恢复共用一个受回滚保护的初始化流程：围绕已取得的 Session 构造具象 agent 与带作用域上下文；存储构造历史、进入 SessionStore 并完成 `session/created`；等待可选 setup 及其 commit；进入 AgentRegistry；等待串行 `agent/created` 监听器；随后释放已排队输入。创建运行时子 Agent 的调用方设置 `options.parentAgent`；调用方 Context 则单独拥有事务和存活句柄。Setup、commit、监听器失败或所有者 dispose 都会回滚已准备的资源。setup 期间 Session 可写；每次移除都会发出 `session/disposed`，包括 Agent 创建宣告前的 setup 失败。初始化失败保留恢复出的收件箱输入及已存储事件，并在释放自身作用域及已取得的写句柄后才拒绝。持久化层仍在创建写句柄时发生取消，可以更早拒绝；该句柄在创建完成后关闭。已送达的宣告仍可被观察，并有配对的销毁通知。Teardown 停止并排空驱动器、撤销作用域、关闭会话写路径、detach agent，再 detach 会话。每次 detach 都绑定到确切进入的对象，因此陈旧 disposer 无法移除之后出现的同 id 替代项。
|
||||
|
||||
loop 在发布工厂前等待 inbox 与轮次边界注册完成，并将句柄传入 Agent 的同步构造过程。工厂卸载时，全局注册可以先于 teardown 完成而消失；保留句柄在整个作用域清理期间仍可读取，包括作用域内 abort 监听器发起取消的时刻。调用方卸载、工厂卸载和句柄释放等待同一次 teardown。
|
||||
|
||||
|
||||
@@ -162,13 +162,15 @@ interface StoredSession {
|
||||
storedCount: number
|
||||
}
|
||||
|
||||
/** Prepared-but-unpublished agent resources sharing one memoized teardown. */
|
||||
interface PreparedAgent {
|
||||
agent: ReactLoopAgent
|
||||
/** Persist initial history, announce the Session, run setup/commit, then enter and announce the Agent. */
|
||||
initializeAndPublish(source: SessionStartSource, setup?: AgentSetup): Promise<AgentHandle>
|
||||
/** Reverse teardown: stop the machine, unregister, unwind the scope. Memoized. */
|
||||
dispose(): Promise<void>
|
||||
/** Acquired Session resources and composition inputs for one Agent initialization. */
|
||||
interface AgentInitializationOptions {
|
||||
readonly session: Session
|
||||
readonly agentOptions: AgentOptions
|
||||
readonly source: SessionStartSource
|
||||
readonly stored?: StoredSession | undefined
|
||||
readonly setup?: AgentSetup | undefined
|
||||
readonly signal?: AbortSignal | undefined
|
||||
readonly parentAgent?: Agent | undefined
|
||||
}
|
||||
|
||||
declare module '@deepseek-ai/cordis' {
|
||||
@@ -453,34 +455,20 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct the driver, scope, and one memoized reverse teardown for a new
|
||||
* agent. The teardown is registered with the factory and the owner fiber
|
||||
* BEFORE publication, so a mid-setup unload rolls everything back; `signal`
|
||||
* fuses caller cancellation with lifecycle teardown for setup awaits.
|
||||
* Own initialization and teardown from an acquired Session to a published Agent.
|
||||
* Registers factory teardown and caller cancellation before constructing the scope.
|
||||
* Failed initialization drains cleanup while retaining stored events and inbox input.
|
||||
* @param ownerCtx - caller context that owns the Agent lifecycle.
|
||||
* @param options - acquired Session/write handle, loop options, start source, parent Agent, setup, and cancellation.
|
||||
* @returns the initialized Agent and its shared disposer.
|
||||
*/
|
||||
private prepare(
|
||||
private async initializeAgent(
|
||||
ownerCtx: Context,
|
||||
id: SessionId,
|
||||
options: AgentOptions,
|
||||
session: Session,
|
||||
callerSignal?: AbortSignal,
|
||||
stored?: StoredSession,
|
||||
parentAgent?: Agent,
|
||||
): PreparedAgent {
|
||||
assertAgentOptions(options)
|
||||
ownerCtx.fiber.assertActive()
|
||||
// Every caller reaches prepare() synchronously from a service method
|
||||
// whose Cordis dispatch already requires the live factory fiber, or
|
||||
// re-checks ownership itself after its awaits (resume's load barrier).
|
||||
/* v8 ignore next -- unreachable backstop, see above */
|
||||
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
||||
if (callerSignal?.aborted) {
|
||||
throw callerSignal.reason instanceof Error
|
||||
? callerSignal.reason
|
||||
: new Error(`agent "${id}" creation aborted`, { cause: callerSignal.reason })
|
||||
}
|
||||
options: AgentInitializationOptions,
|
||||
): Promise<AgentHandle> {
|
||||
const { session, agentOptions, source, stored, setup, signal: callerSignal, parentAgent } = options
|
||||
const id = session.id
|
||||
const loopCtx = this.runtime.ctx
|
||||
const lastTurn = loopCtx.agents.turnBoundaryOf(session).lastTurn
|
||||
|
||||
// Deactivation fuses three owners, each with its own reason: the caller's
|
||||
// cancellation signal, the owner fiber's unload, and factory teardown.
|
||||
@@ -568,9 +556,20 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
const modelProjectionHandle = this.modelProjectionHandle
|
||||
const inboxProjectionHandle = this.inboxProjectionHandle
|
||||
try {
|
||||
assertAgentOptions(agentOptions)
|
||||
ownerCtx.fiber.assertActive()
|
||||
// Factory dispatch and resume's load barrier already require a live owner.
|
||||
/* v8 ignore next -- unreachable backstop for factory ownership */
|
||||
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
||||
if (callerSignal?.aborted) {
|
||||
throw callerSignal.reason instanceof Error
|
||||
? callerSignal.reason
|
||||
: new Error(`agent "${id}" creation aborted`, { cause: callerSignal.reason })
|
||||
}
|
||||
const lastTurn = loopCtx.agents.turnBoundaryOf(session).lastTurn
|
||||
unfollowOwner = ownerCtx.effect(function* () {
|
||||
const created = machine = new ReactLoopAgent(
|
||||
loopCtx, id, options, session,
|
||||
loopCtx, id, agentOptions, session,
|
||||
() => modelProjectionHandle.stateOf(session),
|
||||
() => inboxProjectionHandle.stateOf(session),
|
||||
lastTurn,
|
||||
@@ -593,15 +592,16 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
}
|
||||
}
|
||||
}, `agentLoop.lifecycle(${id})`)
|
||||
/* v8 ignore start -- ctx.effect throws only on an inactive fiber, which assertActive() above already rejected */
|
||||
} catch (error: unknown) {
|
||||
machineReady.resolve()
|
||||
untrack()
|
||||
callerSignal?.removeEventListener('abort', onCallerAbort)
|
||||
this.ownership.signal.removeEventListener('abort', onFactoryTeardown)
|
||||
await stored?.handle.close().catch((_closeError: unknown) => {
|
||||
// Initialization failure remains primary when releasing the acquired handle fails.
|
||||
})
|
||||
throw error
|
||||
}
|
||||
/* v8 ignore stop */
|
||||
|
||||
const assertLive = (): void => {
|
||||
if (!abort.signal.aborted) return
|
||||
@@ -617,39 +617,39 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
const agent = machine
|
||||
assertLive()
|
||||
|
||||
return {
|
||||
agent,
|
||||
initializeAndPublish: async (source, setup) => {
|
||||
publication = Promise.withResolvers<void>()
|
||||
try {
|
||||
assertLive()
|
||||
await this.appendUnstoredSuffix(stored, session)
|
||||
assertLive()
|
||||
detachSession = agent.ctx.sessions.enter(session)
|
||||
agent.ctx.sessions.announce(session)
|
||||
assertLive()
|
||||
const setupCommit = await raceAbort(setup?.(agent.ctx, agent), abort.signal, id)
|
||||
assertLive()
|
||||
setupCommit?.commit()
|
||||
assertLive()
|
||||
// The mounted backend routes announced live events into the active
|
||||
// write handle by session id; the loop only owns the handle itself.
|
||||
detachAgent = loopCtx.agents.enter(agent, parentAgent)
|
||||
await loopCtx.agents.announce(agent, source, abort.signal)
|
||||
assertLive()
|
||||
ready = true
|
||||
return { agent, dispose }
|
||||
} finally {
|
||||
publication.resolve()
|
||||
publication = undefined
|
||||
}
|
||||
},
|
||||
dispose,
|
||||
}
|
||||
return await agent.runMaintenance(async () => {
|
||||
publication = Promise.withResolvers<void>()
|
||||
try {
|
||||
assertLive()
|
||||
await this.appendUnstoredSuffix(stored, session)
|
||||
assertLive()
|
||||
detachSession = agent.ctx.sessions.enter(session)
|
||||
agent.ctx.sessions.announce(session)
|
||||
assertLive()
|
||||
const setupCommit = await raceAbort(setup?.(agent.ctx, agent), abort.signal, id)
|
||||
assertLive()
|
||||
setupCommit?.commit()
|
||||
assertLive()
|
||||
// The mounted backend routes announced live events into the active
|
||||
// write handle by session id; the loop only owns the handle itself.
|
||||
detachAgent = loopCtx.agents.enter(agent, parentAgent)
|
||||
await loopCtx.agents.announce(agent, source, abort.signal)
|
||||
assertLive()
|
||||
ready = true
|
||||
return { agent, dispose }
|
||||
} catch (error: unknown) {
|
||||
// Failed initialization must not release queued input into a new turn.
|
||||
agent.cancel({ kind: 'disposed' }, { keepInbox: true })
|
||||
throw error
|
||||
} finally {
|
||||
publication.resolve()
|
||||
publication = undefined
|
||||
}
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
machineReady.resolve()
|
||||
// Rollback swallows a disposal rejection: the setup failure is primary.
|
||||
void dispose().catch(() => {})
|
||||
await dispose().catch((_disposeError: unknown) => {
|
||||
// Initialization failure remains primary after cleanup reaches quiescence.
|
||||
})
|
||||
throw error
|
||||
}
|
||||
}
|
||||
@@ -667,14 +667,12 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
async create(id: SessionId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): Promise<Agent> {
|
||||
using preparation = SessionPreparation.create(this.runtime.ctx.sessions.prepare(id, { meta }))
|
||||
const stored = await this.createStoredSession(preparation.session)
|
||||
let prepared: PreparedAgent
|
||||
try {
|
||||
prepared = this.prepare(this.ctx, id, options, preparation.session, undefined, stored)
|
||||
} catch (error: unknown) {
|
||||
await stored?.handle.close().catch(() => {})
|
||||
throw error
|
||||
}
|
||||
return (await this.initializeAgent(prepared, () => prepared.initializeAndPublish('startup'))).agent
|
||||
return (await this.initializeAgent(this.ctx, {
|
||||
session: preparation.session,
|
||||
stored,
|
||||
agentOptions: options,
|
||||
source: 'startup',
|
||||
})).agent
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -722,81 +720,30 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
...options.inheritedEventCount === undefined ? {} : { inheritedEventCount: options.inheritedEventCount },
|
||||
}))
|
||||
const published = (async () => {
|
||||
let stored: StoredSession | undefined
|
||||
try {
|
||||
// raceAbortCall normalizes a pre-aborted or mid-create abort and
|
||||
// closes a handle that finishes creating after abandonment.
|
||||
stored = options.signal === undefined
|
||||
? await this.createStoredSession(preparation.session)
|
||||
: await raceAbortCall(
|
||||
() => this.createStoredSession(preparation.session, options.signal),
|
||||
options.signal,
|
||||
options.sessionId,
|
||||
(abandoned) => { void abandoned?.handle.close().catch(() => {}) },
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
preparation[Symbol.dispose]()
|
||||
throw error
|
||||
}
|
||||
return this.setupAndPublish(
|
||||
ownerCtx,
|
||||
options.sessionId,
|
||||
preparation,
|
||||
options.agentOptions ?? {},
|
||||
options.setup,
|
||||
options.signal,
|
||||
'startup',
|
||||
using ownedPreparation = preparation
|
||||
// raceAbortCall closes a handle that finishes creating after cancellation.
|
||||
const stored = options.signal === undefined
|
||||
? await this.createStoredSession(ownedPreparation.session)
|
||||
: await raceAbortCall(
|
||||
() => this.createStoredSession(ownedPreparation.session, options.signal),
|
||||
options.signal,
|
||||
options.sessionId,
|
||||
(abandoned) => { void abandoned?.handle.close().catch(() => {}) },
|
||||
)
|
||||
return await this.initializeAgent(ownerCtx, {
|
||||
session: ownedPreparation.session,
|
||||
stored,
|
||||
options.parentAgent,
|
||||
)
|
||||
agentOptions: options.agentOptions ?? {},
|
||||
setup: options.setup,
|
||||
signal: options.signal,
|
||||
source: 'startup',
|
||||
parentAgent: options.parentAgent,
|
||||
})
|
||||
})()
|
||||
this.ownership.trackWrapper(published)
|
||||
return published
|
||||
}
|
||||
|
||||
/** Prepare one Agent around an acquired Session, run setup, and publish it. */
|
||||
private async setupAndPublish(
|
||||
ownerCtx: Context,
|
||||
id: SessionId,
|
||||
preparation: SessionPreparation,
|
||||
agentOptions: AgentOptions,
|
||||
setup: AgentSetup | undefined,
|
||||
signal: AbortSignal | undefined,
|
||||
source: SessionStartSource,
|
||||
stored?: StoredSession,
|
||||
parentAgent?: Agent,
|
||||
): Promise<AgentHandle> {
|
||||
using ownedPreparation = preparation
|
||||
const session = ownedPreparation.session
|
||||
let prepared: PreparedAgent
|
||||
try {
|
||||
prepared = this.prepare(ownerCtx, id, agentOptions, session, signal, stored, parentAgent)
|
||||
} catch (error: unknown) {
|
||||
await stored?.handle.close().catch(() => {})
|
||||
throw error
|
||||
}
|
||||
return await this.initializeAgent(prepared, () => prepared.initializeAndPublish(source, setup))
|
||||
}
|
||||
|
||||
private async initializeAgent(prepared: PreparedAgent, initialize: () => Promise<AgentHandle>): Promise<AgentHandle> {
|
||||
try {
|
||||
return await prepared.agent.runMaintenance(async () => {
|
||||
try {
|
||||
return await initialize()
|
||||
} catch (error: unknown) {
|
||||
// Teardown owns inbox cleanup and may already have removed its projection.
|
||||
prepared.agent.cancel({ kind: 'disposed' }, { keepInbox: true })
|
||||
throw error
|
||||
}
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
// Rollback swallows a disposal rejection (a failing final handle close):
|
||||
// the setup failure is the primary error the caller must see.
|
||||
await prepared.dispose().catch(() => {})
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume an owned agent from the configured persistence service.
|
||||
* @param ownerCtx - caller context that owns load, setup, and the live lifecycle.
|
||||
@@ -867,18 +814,16 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
ownerCtx.fiber.assertActive()
|
||||
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
||||
const owned = stored
|
||||
handle = undefined // ownership passes to setupAndPublish/prepare
|
||||
return await this.setupAndPublish(
|
||||
ownerCtx,
|
||||
id,
|
||||
preparation,
|
||||
options.agentOptions ?? {},
|
||||
options.setup,
|
||||
options.signal,
|
||||
'resume',
|
||||
owned,
|
||||
options.parentAgent,
|
||||
)
|
||||
handle = undefined // initializeAgent owns the write handle through rollback and teardown.
|
||||
return await this.initializeAgent(ownerCtx, {
|
||||
session: preparation.session,
|
||||
stored: owned,
|
||||
agentOptions: options.agentOptions ?? {},
|
||||
setup: options.setup,
|
||||
signal: options.signal,
|
||||
source: 'resume',
|
||||
parentAgent: options.parentAgent,
|
||||
})
|
||||
} finally {
|
||||
preparation?.[Symbol.dispose]()
|
||||
await handle?.close().catch(() => {})
|
||||
|
||||
@@ -353,7 +353,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('rollback swallows a rejecting handle close after a prepare failure (create and createAgent)', async () => {
|
||||
it('rollback swallows a rejecting handle close after initialization validation fails (create and createAgent)', async () => {
|
||||
const { ctx } = await persistentHarness(new MockAdapter([]))
|
||||
const originalCreate = ctx.sessionPersistence.create.bind(ctx.sessionPersistence)
|
||||
const closeSpies: Array<{ mockRestore: () => void }> = []
|
||||
@@ -363,12 +363,10 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
|
||||
return handle
|
||||
}
|
||||
|
||||
// Config create path: prepare's option validation throws after the handle exists.
|
||||
await expect(ctx.agentLoop.create(SessionId('close-reject-config'), { maxTokens: -1 }))
|
||||
.rejects.toThrow('agent maxTokens must be a positive safe integer')
|
||||
expect(ctx.agents.get(SessionId('close-reject-config'))).toBeUndefined()
|
||||
|
||||
// Owned createAgent path: the same validation failure after the handle exists.
|
||||
await expect(ctx.agents.create({
|
||||
sessionId: SessionId('close-reject-owned'),
|
||||
agentOptions: { maxTokens: -1 },
|
||||
@@ -379,18 +377,20 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('a reentrant abort during preparation swallows a rejecting handle close', async () => {
|
||||
it('a reentrant abort during scope construction waits for a rejecting handle close', async () => {
|
||||
const { ctx } = await persistentHarness(new MockAdapter([]))
|
||||
const sessionId = SessionId('prepare-abort-close-reject')
|
||||
const sessionId = SessionId('scope-abort-close-reject')
|
||||
const originalCreate = ctx.sessionPersistence.create.bind(ctx.sessionPersistence)
|
||||
const spies: Array<{ mockRestore: () => void }> = []
|
||||
let closed: Promise<void> | undefined
|
||||
let closeCompleted = false
|
||||
ctx.sessionPersistence.create = async (header, options) => {
|
||||
const handle = await originalCreate(header, options)
|
||||
const realClose = handle.close.bind(handle)
|
||||
spies.push(vi.spyOn(handle, 'close').mockImplementation(async () => {
|
||||
closed = realClose()
|
||||
await closed
|
||||
closeCompleted = true
|
||||
throw new Error('close failed')
|
||||
}))
|
||||
return handle
|
||||
@@ -404,14 +404,15 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
|
||||
controller.abort(reason)
|
||||
})
|
||||
|
||||
await expect(ctx.agents.create({ sessionId, signal: controller.signal })).rejects.toBe(reason)
|
||||
// The rejecting close stays swallowed by the rollback; wait for the
|
||||
// rollback's real close so factory teardown finds a quiescent lifecycle.
|
||||
await vi.waitFor(() => { if (closed === undefined) throw new Error('close not reached') })
|
||||
await closed
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
for (const spy of spies.splice(0)) spy.mockRestore()
|
||||
await ctx.fiber.dispose()
|
||||
try {
|
||||
await expect(ctx.agents.create({ sessionId, signal: controller.signal })).rejects.toBe(reason)
|
||||
expect(closeCompleted).toBe(true)
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
} finally {
|
||||
await closed
|
||||
for (const spy of spies.splice(0)) spy.mockRestore()
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('config create rollback swallows a rejecting close after a publish failure', async () => {
|
||||
|
||||
@@ -189,20 +189,31 @@ describe('agent scope lifecycle', () => {
|
||||
const reason = new Error('cancelled while preparing')
|
||||
const controller = new AbortController()
|
||||
let aborted = false
|
||||
let scopeReleased = false
|
||||
ctx.on('internal/plugin', (fiber) => {
|
||||
if (aborted || fiber.name !== 'scope') return
|
||||
aborted = true
|
||||
fiber.ctx.effect(() => () => new Promise<void>((resolve) => {
|
||||
setImmediate(() => {
|
||||
scopeReleased = true
|
||||
resolve()
|
||||
})
|
||||
}))
|
||||
controller.abort(reason)
|
||||
})
|
||||
|
||||
await expect(ctx.agents.create({
|
||||
sessionId: SessionId('prepare-abort-s'),
|
||||
signal: controller.signal,
|
||||
})).rejects.toBe(reason)
|
||||
try {
|
||||
await expect(ctx.agents.create({
|
||||
sessionId: SessionId('prepare-abort-s'),
|
||||
signal: controller.signal,
|
||||
})).rejects.toBe(reason)
|
||||
|
||||
expect(ctx.agents.get(SessionId('prepare-abort-s'))).toBeUndefined()
|
||||
expect(ctx.sessions.get(SessionId('prepare-abort-s'))).toBeUndefined()
|
||||
await ctx.fiber.dispose()
|
||||
expect(scopeReleased).toBe(true)
|
||||
expect(ctx.agents.get(SessionId('prepare-abort-s'))).toBeUndefined()
|
||||
expect(ctx.sessions.get(SessionId('prepare-abort-s'))).toBeUndefined()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('normalizes non-Error create failures for rollback while rethrowing the original value', async () => {
|
||||
|
||||
@@ -18,8 +18,8 @@
|
||||
en: 7d1ae2375bd4fbfc
|
||||
zh: b6b4b2e943174cb2
|
||||
/deepseek-ai-dsh-agent/use-this-package/create-or-resume-an-agent:
|
||||
en: e5479ccebe6d7fd6
|
||||
zh: 191b9889e2d7699c
|
||||
en: 26d49c76bce41e30
|
||||
zh: 9710f54e24b190d1
|
||||
/deepseek-ai-dsh-agent/use-this-package/select-a-model:
|
||||
en: 85b0ff72a187110f
|
||||
zh: 3f61c1568d27b8bc
|
||||
|
||||
@@ -29,7 +29,7 @@ Mount `dsh-agent` with `dsh-session-projection`: it provides `ctx.agents`, the `
|
||||
|
||||
### Create or resume an agent
|
||||
|
||||
`ctx.agents.create()` builds a fresh agent and session under one identity; `ctx.agents.resume()` loads a persisted session and rebuilds the agent on it. Both delegate to the registered factory and return an `AgentHandle` — the only object that can tear that agent down. Set `parentAgent` in either operation's options to make the result a runtime child; omit it for a runtime root. `get(id)`, `list()`, and `roots()` find live agents, and `isOwnedBy(id, parent)` tests that exact live relation.
|
||||
`ctx.agents.create()` builds a fresh agent and session under one identity; `ctx.agents.resume()` loads a persisted session and rebuilds the agent on it. Both delegate to the registered factory and return an `AgentHandle` — the only object that can tear that agent down. See [`AgentFactory`](src/index.ts) for publication and failure-cleanup timing. Set `parentAgent` in either operation's options to make the result a runtime child; omit it for a runtime root. `get(id)`, `list()`, and `roots()` find live agents, and `isOwnedBy(id, parent)` tests that exact live relation.
|
||||
|
||||
```text
|
||||
const handle = await ctx.agents.create({
|
||||
|
||||
@@ -29,7 +29,7 @@ kind: "package-reference"
|
||||
|
||||
### 创建或恢复 agent
|
||||
|
||||
`ctx.agents.create()` 在一个身份下构建全新 agent 与会话；`ctx.agents.resume()` 加载持久化会话并在此基础上重建 agent。两者都委托给已注册工厂，并返回 `AgentHandle`——唯一能拆除该 agent 的对象。在任一操作的 options 中设置 `parentAgent`，可使结果成为运行时子级；省略它则得到运行时根级。`get(id)`、`list()` 与 `roots()` 用于查找实时 agent；`isOwnedBy(id, parent)` 用于检验这项确切的实时所有权关系。
|
||||
`ctx.agents.create()` 在一个身份下构建全新 agent 与会话；`ctx.agents.resume()` 加载持久化会话并在此基础上重建 agent。两者都委托给已注册工厂，并返回 `AgentHandle`——唯一能拆除该 agent 的对象。发布和失败清理的时序见 [`AgentFactory`](src/index.ts)。在任一操作的 options 中设置 `parentAgent`，可使结果成为运行时子级；省略它则得到运行时根级。`get(id)`、`list()` 与 `roots()` 用于查找实时 agent；`isOwnedBy(id, parent)` 用于检验这项确切的实时所有权关系。
|
||||
|
||||
```text
|
||||
const handle = await ctx.agents.create({
|
||||
|
||||
@@ -181,7 +181,10 @@ export interface AgentFactory {
|
||||
* rollback-covered, but notifications delivered before a later listener
|
||||
* failure remain observable; every agent or session creation announcement
|
||||
* that began is paired by `agent/disposed` or `session/disposed` during
|
||||
* rollback. The owner disposes the resolved handle to stop/drain,
|
||||
* rollback. Once Agent initialization begins, failures reject after its
|
||||
* scope and acquired write handle finish cleanup. Cancellation during
|
||||
* persistence acquisition may reject before a late handle is available to close.
|
||||
* The owner disposes the resolved handle to stop/drain,
|
||||
* unregister, remove the session, and unwind the scope.
|
||||
* The registry passes a context carrying the `create()` caller's fiber and
|
||||
* scope as `ownerCtx`. The implementation attaches the unpublished
|
||||
@@ -197,7 +200,7 @@ export interface AgentFactory {
|
||||
* persisted session for write, reads and repairs the log, publishes it, and
|
||||
* awaits the optional unpublished setup transaction; must be called after
|
||||
* `ctx.sessionPersistence` exists (consumers inject `sessionPersistence`).
|
||||
* Publication follows the same setup-commit and ordered boundary as
|
||||
* Initialization cleanup and publication follow the same timing as
|
||||
* {@link createAgent}.
|
||||
* @param ownerCtx - caller-bound context that owns load, setup, and the live handle.
|
||||
* @param options - persisted identity, configuration, optional live parent, and setup.
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Session, SessionId, SessionPreparation } from '@deepseek-ai/dsh-session'
|
||||
|
||||
describe('SessionPreparation', () => {
|
||||
it('retains the exact Session and releases provider state only once', () => {
|
||||
const session = Session.create(SessionId('prepared'))
|
||||
const release = vi.fn()
|
||||
{
|
||||
using preparation = SessionPreparation.create(session, { release })
|
||||
expect(preparation.session).toBe(session)
|
||||
preparation[Symbol.dispose]()
|
||||
expect(release).toHaveBeenCalledOnce()
|
||||
}
|
||||
expect(release).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('disposes a preparation without provider state', () => {
|
||||
using preparation = SessionPreparation.create(Session.create(SessionId('unretained')))
|
||||
expect(() => { preparation[Symbol.dispose]() }).not.toThrow()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user