Merge remote-tracking branch 'origin/master' into parallel-tool-call
# Conflicts: # docs/architecture.md # docs/config-catalog.md # docs/cordis-catalog/services.md # docs/rfc/INDEX.md # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-loop/README.md # packages/core/agent-loop/src/index.ts # packages/core/agent-loop/src/loop.ts # packages/core/tools/README.md # packages/core/tools/src/index.ts # packages/subagent/subagent/src/types.ts # packages/subagent/tool-subagent/README.md # scripts/gen-cordis-catalog.ts # scripts/type-equiv.manifest.json
This commit is contained in:
Dudu-0223
@@ -1,27 +1,327 @@
|
||||
/**
|
||||
* THE concrete agent plugin: creates ReactLoopAgents, runs their loops, and
|
||||
* registers them in ctx.agents. Deliberately thin — every behavior beyond
|
||||
* "call the model, run the tools, repeat" belongs to plugins on the event
|
||||
* taxonomy.
|
||||
* Concrete agent-loop plugin: creates scoped ReactLoopAgents, publishes them
|
||||
* through the agent/session registries, and owns their ordered teardown.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-agent-loop
|
||||
*/
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import { Context, FiberState, Service } from 'cordis'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import z from 'schemastery'
|
||||
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions, SessionStartSource } from '@deepseek-ai/dsh-agent'
|
||||
import { createScope } from '@deepseek-ai/dsh-scope'
|
||||
import type { Scope } from '@deepseek-ai/dsh-scope'
|
||||
import { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import type {
|
||||
AgentFactory,
|
||||
AgentHandle,
|
||||
AgentId,
|
||||
AgentOptions,
|
||||
CreateAgentOptions,
|
||||
ResumeAgentOptions,
|
||||
SessionStartSource,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
import type {} from '@deepseek-ai/dsh-llm'
|
||||
import { SessionId, type SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { ReactLoopAgent } from './agent.ts'
|
||||
import {
|
||||
bindReactLoopAgentContext,
|
||||
prepareReactLoopAgent,
|
||||
ReactLoopAgent,
|
||||
} from './agent.ts'
|
||||
import type { PreparedReactLoopAgent } from './agent.ts'
|
||||
|
||||
export { ReactLoopAgent } from './agent.ts'
|
||||
export { Inbox, type InboxMessage } from './inbox.ts'
|
||||
export { runLoop } from './loop.ts'
|
||||
|
||||
/** Fiber states that cannot own or serve a new lifecycle. */
|
||||
const INACTIVE_STATES: ReadonlySet<FiberState> = new Set([
|
||||
FiberState.UNLOADING,
|
||||
FiberState.DISPOSED,
|
||||
FiberState.FAILED,
|
||||
])
|
||||
|
||||
/** Factory-level ownership of every preparing or live transaction. */
|
||||
class FactoryOwnership {
|
||||
private accepting = true
|
||||
private transactions = new Set<AgentCreationTransaction>()
|
||||
|
||||
constructor(private readonly fiber: Context['fiber']) {}
|
||||
|
||||
isActive(): boolean {
|
||||
return this.accepting && !INACTIVE_STATES.has(this.fiber.state)
|
||||
}
|
||||
|
||||
track(transaction: AgentCreationTransaction): () => void {
|
||||
this.transactions.add(transaction)
|
||||
return () => { this.transactions.delete(transaction) }
|
||||
}
|
||||
|
||||
async dispose(): Promise<void> {
|
||||
this.accepting = false
|
||||
const reason = new Error('agent loop is not active')
|
||||
await Promise.all(
|
||||
[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** Build the public cancellation error while preserving a caller-supplied cause. */
|
||||
function signalAbortError(id: AgentId, signal: AbortSignal): Error {
|
||||
if (signal.reason instanceof Error) return signal.reason
|
||||
return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
|
||||
}
|
||||
|
||||
/** Validate merge-extended options the loop owns before a session is published. */
|
||||
function validateAgentOptions(options: AgentOptions): void {
|
||||
const { maxParallelToolCalls } = options
|
||||
if (maxParallelToolCalls !== undefined && (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1)) {
|
||||
throw new Error('maxParallelToolCalls must be a positive integer')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One create/resume transaction from caller ownership through unpublished
|
||||
* setup, rollback-covered publication, and final quiescent teardown.
|
||||
*
|
||||
* The class deliberately owns the state machine in one place. Registries only
|
||||
* arbitrate identity at their final `enter()` calls; before that point every
|
||||
* resource is private to this transaction.
|
||||
*/
|
||||
class AgentCreationTransaction {
|
||||
private active = true
|
||||
private failure: Error | undefined
|
||||
private readonly deactivation = Promise.withResolvers<void>()
|
||||
private readonly publication = Promise.withResolvers<void>()
|
||||
private readonly torndown = Promise.withResolvers<void>()
|
||||
private readonly wrapperCompletion = Promise.withResolvers<void>()
|
||||
private preparing: Promise<void> | undefined
|
||||
private driver: PreparedReactLoopAgent | undefined
|
||||
private scope: Scope | undefined
|
||||
private session: Session | undefined
|
||||
private lifecycleDispose: (() => Promise<void> | void) | undefined
|
||||
private detachSession: (() => void) | undefined
|
||||
private detachAgent: (() => void) | undefined
|
||||
private publishing = false
|
||||
private cleanupTask: Promise<void> | undefined
|
||||
private ownerFollowing = true
|
||||
private readonly ownerDispose: () => Promise<void> | void
|
||||
private readonly untrackFactory: () => void
|
||||
private readonly abortListener: (() => void) | undefined
|
||||
readonly ownerAgent: Context['agent']
|
||||
readonly ownerFiber: Context['fiber']
|
||||
|
||||
constructor(
|
||||
private readonly loopCtx: Context,
|
||||
private readonly ownerCtx: Context,
|
||||
private readonly ownership: FactoryOwnership,
|
||||
readonly id: AgentId,
|
||||
signal?: AbortSignal,
|
||||
) {
|
||||
ownerCtx.fiber.assertActive()
|
||||
this.ownerAgent = ownerCtx.agent
|
||||
this.ownerFiber = ownerCtx.fiber
|
||||
if (!ownership.isActive()) throw new Error('agent loop is not active')
|
||||
this.ownerDispose = ownerCtx.effect(() => () => {
|
||||
if (!this.ownerFollowing) return
|
||||
return this.dispose(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
|
||||
}, `agentLoop.owner(${id})`)
|
||||
this.untrackFactory = ownership.track(this)
|
||||
if (signal === undefined) {
|
||||
this.abortListener = undefined
|
||||
} else {
|
||||
this.abortListener = () => {
|
||||
/* v8 ignore next 3 -- transaction teardown contains callback/driver failures; rejection is a future-drift backstop. */
|
||||
void this.dispose(signalAbortError(id, signal)).catch((error: unknown) => {
|
||||
this.loopCtx.logger.error(error)
|
||||
})
|
||||
}
|
||||
signal.addEventListener('abort', this.abortListener, { once: true })
|
||||
if (signal.aborted) this.deactivate(signalAbortError(id, signal))
|
||||
}
|
||||
this.signal = signal
|
||||
}
|
||||
|
||||
private readonly signal: AbortSignal | undefined
|
||||
|
||||
/** Whether caller, provider, and optional parent-agent ownership remain live. */
|
||||
isActive(): boolean {
|
||||
return this.active
|
||||
&& this.ownership.isActive()
|
||||
&& this.ownerFiber.uid !== null
|
||||
&& !INACTIVE_STATES.has(this.ownerFiber.state)
|
||||
&& this.ownerAgent?.status !== 'disposed'
|
||||
}
|
||||
|
||||
/** Fail synchronously at every real lifecycle boundary after deactivation. */
|
||||
assertActive(): void {
|
||||
if (this.isActive()) return
|
||||
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
||||
throw this.failure ?? new Error(`agent "${this.id}" setup aborted: owner disposed during setup`)
|
||||
}
|
||||
|
||||
/** Race an external async operation against structural/signal deactivation. */
|
||||
async waitFor<T>(operation: PromiseLike<T> | T): Promise<T> {
|
||||
this.assertActive()
|
||||
return await Promise.race([
|
||||
Promise.resolve(operation),
|
||||
this.deactivation.promise.then(() => {
|
||||
/* v8 ignore next -- deactivate() assigns failure before resolving deactivation. */
|
||||
throw this.failure ?? new Error(`agent "${this.id}" creation deactivated`)
|
||||
}),
|
||||
])
|
||||
}
|
||||
|
||||
/** Construct the driver and scope, then install their complete ordered lifecycle. */
|
||||
prepare(options: AgentOptions, session: Session): ReactLoopAgent {
|
||||
this.assertActive()
|
||||
const gate = Promise.withResolvers<void>()
|
||||
this.preparing = gate.promise
|
||||
try {
|
||||
this.session = session
|
||||
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session)
|
||||
this.driver = driver
|
||||
const agent = driver.agent
|
||||
const scope = createScope(this.loopCtx, agent)
|
||||
this.scope = scope
|
||||
bindReactLoopAgentContext(agent, scope.ctx.extend({ agent }))
|
||||
this.installLifecycle(scope, driver)
|
||||
this.assertActive()
|
||||
return agent
|
||||
} catch (error: unknown) {
|
||||
if (!this.isActive() && error instanceof Error && /inactive context/.test(error.message)) {
|
||||
throw this.failure ?? this.disposalReason()
|
||||
}
|
||||
throw error
|
||||
} finally {
|
||||
gate.resolve()
|
||||
this.preparing = undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Register the exact scope disposer inside the ordered transaction effect. */
|
||||
private installLifecycle(scope: Scope, driver: PreparedReactLoopAgent): void {
|
||||
this.lifecycleDispose = this.ownerCtx.effect(function* (this: AgentCreationTransaction) {
|
||||
// First yielded, disposed last.
|
||||
yield () => { this.finish() }
|
||||
yield scope.rawDispose
|
||||
yield () => {
|
||||
this.detachSession?.()
|
||||
this.detachSession = undefined
|
||||
}
|
||||
yield () => {
|
||||
this.detachAgent?.()
|
||||
this.detachAgent = undefined
|
||||
}
|
||||
// Last yielded, disposed first.
|
||||
yield () => {
|
||||
this.deactivate(this.disposalReason())
|
||||
if (this.publishing) {
|
||||
return this.publication.promise.then(() => driver.dispose())
|
||||
}
|
||||
return driver.dispose()
|
||||
}
|
||||
}.bind(this), `agentLoop.lifecycle(${this.id})`)
|
||||
}
|
||||
|
||||
/** Publish the exact prepared objects and start the driver. */
|
||||
publish(source: SessionStartSource): AgentHandle {
|
||||
this.assertActive()
|
||||
const driver = this.driver
|
||||
/* v8 ignore next -- publish() is private and every caller invokes prepare() first. */
|
||||
if (driver === undefined) throw new Error(`agent "${this.id}" is not prepared`)
|
||||
const agent = driver.agent
|
||||
const session = this.session
|
||||
/* v8 ignore next -- prepare() assigns the session before it can produce the driver above. */
|
||||
if (session === undefined) throw new Error(`agent "${this.id}" has no prepared session`)
|
||||
this.publishing = true
|
||||
try {
|
||||
this.detachSession = agent.ctx.sessions.enter(session)
|
||||
this.detachAgent = this.loopCtx.agents.enter(agent)
|
||||
|
||||
agent.ctx.sessions.announce(session)
|
||||
this.assertActive()
|
||||
this.loopCtx.agents.announce(agent)
|
||||
this.assertActive()
|
||||
|
||||
driver.markPublished()
|
||||
agentEvents(this.loopCtx, agent).emit('agent/session-start', source)
|
||||
this.assertActive()
|
||||
driver.startDriver()
|
||||
return { agent, dispose: () => this.dispose() }
|
||||
} finally {
|
||||
this.publishing = false
|
||||
this.publication.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
/** Mark the transaction inactive exactly once and wake load/setup races. */
|
||||
private deactivate(reason: Error): void {
|
||||
if (!this.active) return
|
||||
this.active = false
|
||||
this.failure = reason
|
||||
this.deactivation.resolve()
|
||||
}
|
||||
|
||||
/** Choose the structural cause when an owner/factory effect starts teardown first. */
|
||||
private disposalReason(): Error {
|
||||
if (this.failure !== undefined) return this.failure
|
||||
if (!this.ownership.isActive()) return new Error('agent loop is not active')
|
||||
if (this.ownerFiber.uid === null || INACTIVE_STATES.has(this.ownerFiber.state) || this.ownerAgent?.status === 'disposed') {
|
||||
return new Error(`agent "${this.id}" setup aborted: owner disposed during setup`)
|
||||
}
|
||||
return new Error(`agent "${this.id}" lifecycle disposed`)
|
||||
}
|
||||
|
||||
/** Complete ownership bookkeeping after every resource reached quiescence. */
|
||||
private finish(): void {
|
||||
this.untrackFactory()
|
||||
this.ownerFollowing = false
|
||||
void this.ownerDispose()
|
||||
this.torndown.resolve()
|
||||
}
|
||||
|
||||
/**
|
||||
* Deactivate and quiesce this transaction. The promise is memoized because
|
||||
* Cordis effect disposers are single-shot while handles promise shared
|
||||
* quiescence to every racing owner.
|
||||
*/
|
||||
dispose(reason = new Error(`agent "${this.id}" lifecycle disposed`)): Promise<void> {
|
||||
this.deactivate(reason)
|
||||
return (this.cleanupTask ??= (async () => {
|
||||
if (this.preparing !== undefined) await this.preparing
|
||||
if (this.lifecycleDispose !== undefined) {
|
||||
await this.lifecycleDispose()
|
||||
await this.torndown.promise
|
||||
return
|
||||
}
|
||||
try {
|
||||
await this.driver?.dispose()
|
||||
} finally {
|
||||
try {
|
||||
await this.scope?.dispose()
|
||||
} finally {
|
||||
this.finish()
|
||||
}
|
||||
}
|
||||
})())
|
||||
}
|
||||
|
||||
/** Mark the public create/resume continuation settled and detach its creation-only signal. */
|
||||
finishWrapper(): void {
|
||||
if (this.signal !== undefined && this.abortListener !== undefined) {
|
||||
this.signal.removeEventListener('abort', this.abortListener)
|
||||
}
|
||||
this.wrapperCompletion.resolve()
|
||||
}
|
||||
|
||||
/** Factory shutdown joins both resource teardown and the public wrapper's deactivation continuation. */
|
||||
async disposeForFactory(reason: Error): Promise<void> {
|
||||
await this.dispose(reason)
|
||||
await this.wrapperCompletion.promise
|
||||
}
|
||||
}
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
@@ -45,57 +345,29 @@ declare module '@deepseek-ai/dsh-agent' {
|
||||
|
||||
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
|
||||
|
||||
/**
|
||||
* Plugin config: the agents to create — or resume, via `resumeSessionId` —
|
||||
* declaratively at startup, so a cordis.yml deployment needs no code.
|
||||
*/
|
||||
/** Plugin configuration for declarative startup agents. */
|
||||
export interface Config {
|
||||
/** Agents created from configuration at startup. */
|
||||
/** Agents created or resumed at plugin startup. */
|
||||
agents: (AgentOptions & {
|
||||
/** Agent id to register under; also seeds the fresh per-run session id (`${id}-session-<uuid>`). */
|
||||
/** Registry identity for the live agent. */
|
||||
id: AgentId
|
||||
/** Optional workspace cwd for the config-created fresh session. */
|
||||
/** Optional workspace for a fresh session. */
|
||||
cwd?: string
|
||||
/**
|
||||
* Maximum parallel-safe tool calls to run concurrently within one assistant
|
||||
* step. Must be a positive integer; `1` preserves serial execution.
|
||||
*/
|
||||
maxParallelToolCalls?: number
|
||||
/**
|
||||
* If set, the config agent RESUMES this persisted session id instead of
|
||||
* starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
|
||||
* cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a
|
||||
* demo can continue a prior conversation without code changes. Requires a
|
||||
* `dsh-session-persistence` backend; the resume is deferred until that
|
||||
* service is available (via `ctx.inject`) and the loaded session's events
|
||||
* seed the live session so history continues.
|
||||
*
|
||||
* The schema accepts a plain string at runtime (cordis.yml values are
|
||||
* untyped); the brand is compile-time only — the config format is the
|
||||
* boundary where an id enters, so the TYPE declares the brand here.
|
||||
*/
|
||||
/** Persisted session to resume instead of creating a fresh session. */
|
||||
resumeSessionId?: SessionId
|
||||
})[]
|
||||
}
|
||||
|
||||
/**
|
||||
* The agent-loop plugin (`ctx.agentLoop`): creates {@link ReactLoopAgent}s, runs
|
||||
* their loops, and registers them in `ctx.agents`. Also implements the
|
||||
* {@link AgentFactory} seam, so plugins create/resume agents through
|
||||
* `ctx.agents` (the interface) without depending on this concrete package.
|
||||
*
|
||||
* The loop itself is deliberately thin — every behavior beyond "call the
|
||||
* model, run the tools, repeat" belongs to plugins listening on the event
|
||||
* taxonomy declared in @deepseek-ai/dsh-agent.
|
||||
*/
|
||||
/** Concrete ReactLoopAgent factory and driver service. */
|
||||
export class AgentLoop extends Service implements AgentFactory {
|
||||
static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']
|
||||
|
||||
// The schema validates plain strings (cordis.yml config values are untyped at
|
||||
// runtime); the {@link Config} TYPE declares the branded `id`/`resumeSessionId`
|
||||
// because the config format is the boundary where an id enters. The brand is a
|
||||
// zero-cost compile-time cast, so the runtime schema stays string-based and we
|
||||
// assert the branded view once here — the single schema boundary.
|
||||
/** Runtime schema for declarative agents. */
|
||||
static Config = z.object({
|
||||
agents: z.array(z.object({
|
||||
id: z.string().required(),
|
||||
@@ -108,281 +380,148 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
})).default([]),
|
||||
}) as unknown as z<Config>
|
||||
|
||||
private readonly ownership: FactoryOwnership
|
||||
/** Plain holder prevents Cordis from re-tracing the factory's dependency context through a caller shadow. */
|
||||
private readonly runtime: { ctx: Context }
|
||||
|
||||
constructor(ctx: Context, public config: Config) {
|
||||
super(ctx, 'agentLoop')
|
||||
// Provide the agent-creation factory to the registry (effect-scoped: the
|
||||
// slot is cleared on dispose).
|
||||
ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()')
|
||||
// The prompt variables the shipped loop provides, registered once. The
|
||||
// sections themselves (`harness:identity`, `deployment:persona`) belong to
|
||||
// dsh-system-prompt — they must survive a swapped loop plugin — but
|
||||
// `{{model}}`/`{{cwd}}` are runtime facts of the agents THIS loop drives:
|
||||
// it assembles with `{ agent }` each step (loop.ts), and the variables
|
||||
// project the agent's configured model and its session workspace from that
|
||||
// context. A provider returns undefined when the fact is absent
|
||||
// (renderPrompt then rejects a persona that claims it — fail loud).
|
||||
this.ownership = new FactoryOwnership(ctx.fiber)
|
||||
this.runtime = { ctx }
|
||||
ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
|
||||
ctx.effect(() => ctx.agents.setFactory(this), 'agentLoop.setFactory()')
|
||||
ctx.systemPrompt.variable('model', context => context.agent?.options.model)
|
||||
ctx.systemPrompt.variable('cwd', context => context.agent?.session.header.cwd)
|
||||
|
||||
for (const { id, cwd, resumeSessionId, ...options } of config.agents) {
|
||||
if (resumeSessionId !== undefined && resumeSessionId !== '') {
|
||||
// Resume a prior session instead of starting fresh. resume() needs
|
||||
// `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml
|
||||
// lists the backend later). `ctx.inject(['sessionPersistence'], cb)`
|
||||
// runs `cb` with a child ctx once the service exists; the child reads
|
||||
// the persistence and hands it to resumeWith (which uses this.ctx — the
|
||||
// parent — for sessions/registry, all in AgentLoop's static inject). A
|
||||
// failed resume is contained + logged: startup must not crash.
|
||||
ctx.effect(() => {
|
||||
const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => {
|
||||
void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options })
|
||||
.catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
|
||||
})
|
||||
})
|
||||
return () => void fiber.dispose()
|
||||
}, `agentLoop.resume(${id})`)
|
||||
} else {
|
||||
if (resumeSessionId === undefined || resumeSessionId === '') {
|
||||
this.create(id, options, cwd === undefined ? {} : { cwd })
|
||||
continue
|
||||
}
|
||||
ctx.effect(() => {
|
||||
const fiber = ctx.inject(['sessionPersistence'], (childCtx: Context) => {
|
||||
void this.resumeWith(ctx, childCtx.sessionPersistence, {
|
||||
agentId: id,
|
||||
resumeSessionId,
|
||||
agentOptions: options,
|
||||
}).catch((error: unknown) => {
|
||||
ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
|
||||
})
|
||||
})
|
||||
return fiber.dispose
|
||||
}, `agentLoop.resume(${id})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Config-driven create: an agent on a FRESH, non-colliding session id per run
|
||||
* (`${id}-session-<uuid>`). Used for `cordis.yml`-configured agents and as
|
||||
* the shared core for the programmatic factory {@link createAgent}.
|
||||
*
|
||||
* Why a per-run id, not a fixed `${id}-session`: once a durable persistence
|
||||
* backend is loaded, a fixed id collides on the second run — the backend
|
||||
* refuses to re-create an id whose log already exists on disk (the SessionId
|
||||
* is the identity). A fresh id means each run is a new session.
|
||||
*
|
||||
* TODO(demo): each run starting a brand-new session is fine for demos but is
|
||||
* NOT real conversation continuity. A production config-driven agent needs a
|
||||
* deliberate resume-or-create policy (resume the prior session if one exists,
|
||||
* else start fresh) or an explicit caller-chosen session id — revisit when the
|
||||
* UI/ACP path owns session selection.
|
||||
* @param id - the agent id; also seeds the generated session id.
|
||||
* @param options - loop options (model, limits, …); defaults applied per option.
|
||||
* @param meta - optional session metadata for the fresh session.
|
||||
* @returns the running agent, owned by the calling fiber (no handle).
|
||||
* Create an agent on a fresh per-run session, owned by the accessing fiber.
|
||||
* Constructor-driven config calls use the loop fiber itself.
|
||||
* @param id - agent registry id.
|
||||
* @param options - concrete loop options.
|
||||
* @param meta - optional fresh-session workspace metadata.
|
||||
* @returns the published running agent.
|
||||
*/
|
||||
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
|
||||
this.validateAgentOptions(options)
|
||||
this.assertAgentIdFree(id)
|
||||
// Config/programmatic path: prepare the session and let start() fold its
|
||||
// lifecycle into the agent's composite effect (so a fiber unload tears the
|
||||
// session + agent down as one ordered chain, capturing the loop's closing
|
||||
// flush). The whole effect is owned by THIS fiber; no AgentHandle is needed.
|
||||
const session = this.ctx.sessions.prepare(SessionId(`${id}-session-${randomUUID()}`), { meta })
|
||||
const { agent } = this.start(id, options, session, 'startup')
|
||||
return agent
|
||||
validateAgentOptions(options)
|
||||
const loopCtx = this.runtime.ctx
|
||||
const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
|
||||
try {
|
||||
const sessionId = SessionId(`${id}-session-${randomUUID()}`)
|
||||
const session = loopCtx.sessions.prepare(sessionId, { meta })
|
||||
const agent = transaction.prepare(options, session)
|
||||
transaction.publish('startup')
|
||||
return agent
|
||||
} catch (error: unknown) {
|
||||
void transaction.dispose(error instanceof Error ? error : new Error(String(error)))
|
||||
throw error
|
||||
} finally {
|
||||
transaction.finishWrapper()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Programmatic factory create ({@link AgentFactory}): an agent on a
|
||||
* caller-supplied `sessionId` (NOT `${id}-session`), with optional session
|
||||
* metadata (validated `cwd`, lineage) and an optional `seed` event prefix. The
|
||||
* ACP bridge uses this so the client-generated session id becomes the
|
||||
* live/persisted session id; the in-process FORK subagent backend passes a
|
||||
* `seed` (a balanced completed-turn prefix of the parent's log) so the child
|
||||
* starts with the parent's context. Returns an {@link AgentHandle} the owner
|
||||
* disposes to tear down exactly this agent.
|
||||
* @param options - agent id, caller-supplied session id, optional seed/meta,
|
||||
* and agent options.
|
||||
* @returns the handle whose dispose tears down exactly this agent.
|
||||
* Create an owned agent on a caller-supplied session id.
|
||||
* @param ownerCtx - caller context that structurally owns the transaction.
|
||||
* @param options - identities, session seed/metadata, loop options, setup, and cancellation.
|
||||
* @returns the published handle.
|
||||
*/
|
||||
createAgent(options: CreateAgentOptions): AgentHandle {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
// Check the agent id BEFORE preparing the session: register() would reject a
|
||||
// duplicate id only AFTER the session enters the store, leaving an orphaned
|
||||
// live session (and lazy persistence state) that blocks reuse of that id.
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const session = this.ctx.sessions.prepare(options.sessionId, {
|
||||
...options.seed !== undefined ? { seed: options.seed } : {},
|
||||
meta: options.meta ?? {},
|
||||
})
|
||||
// A seeded (forked) create is still a fresh start, NOT a resume — `resume`
|
||||
// is reserved for reloading a PERSISTED session via resume()/resumeWith().
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'startup')
|
||||
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
|
||||
validateAgentOptions(options.agentOptions ?? {})
|
||||
const transaction = new AgentCreationTransaction(
|
||||
this.runtime.ctx,
|
||||
ownerCtx,
|
||||
this.ownership,
|
||||
options.agentId,
|
||||
options.signal,
|
||||
)
|
||||
try {
|
||||
const session = this.runtime.ctx.sessions.prepare(options.sessionId, {
|
||||
...options.seed === undefined ? {} : { seed: options.seed },
|
||||
...options.meta === undefined ? {} : { meta: options.meta },
|
||||
})
|
||||
const agent = transaction.prepare(options.agentOptions ?? {}, session)
|
||||
await transaction.waitFor(options.setup?.(agent.ctx))
|
||||
transaction.assertActive()
|
||||
return transaction.publish('startup')
|
||||
} catch (error: unknown) {
|
||||
await transaction.dispose(error instanceof Error ? error : new Error(String(error)))
|
||||
throw error
|
||||
} finally {
|
||||
transaction.finishWrapper()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume an agent on a persisted session ({@link AgentFactory}). Loads the
|
||||
* session log + metadata via `ctx.sessionPersistence`, reconstructs the live
|
||||
* session with the loaded events (so `lastTurnNumber`/`deriveMessages`
|
||||
* continue), and starts a fresh agent on it. The live session id is the
|
||||
* resumed id, NOT `${agentId}-session`.
|
||||
*
|
||||
* Requires `ctx.sessionPersistence`; rejects with a clear error if it is not
|
||||
* configured. NOT hard-injected (that would make non-persistent demos pend
|
||||
* forever) — callers that need resume (ACP) inject `sessionPersistence`, so
|
||||
* by the time this runs the service exists.
|
||||
* @param options - the persisted session id to reload, plus agent id/options.
|
||||
* @returns the handle for the agent resumed on the reconstructed session.
|
||||
* Resume an owned agent from the configured persistence service.
|
||||
* @param ownerCtx - caller context that owns load, setup, and the live lifecycle.
|
||||
* @param options - persisted identity, loop options, setup, and cancellation.
|
||||
* @returns the published handle.
|
||||
*/
|
||||
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
// Read the service through `ctx.get('sessionPersistence')` — a direct
|
||||
// global-store lookup keyed by the isolate symbol — NOT
|
||||
// `this.ctx.sessionPersistence`. AgentLoop deliberately does NOT inject
|
||||
// `sessionPersistence` (injecting it would pend non-persistent demos
|
||||
// forever). The `ctx.<name>` property proxy resolves a service by an
|
||||
// ancestor-only walk of the current fiber's parent chain; from AgentLoop's
|
||||
// own fiber (which lacks the inject) that walk never reaches the sibling
|
||||
// backend fiber and throws "cannot get property … without inject". Worse,
|
||||
// when the call arrives via a traceable shadow (e.g. the ACP bridge child
|
||||
// fiber → `ctx.agents.resume()` → `this.factory.resume()`), the walk starts
|
||||
// at the shadow's origin fiber and fails the same way. `ctx.get(name)`
|
||||
// sidesteps the fiber walk entirely (a store lookup by the global isolate
|
||||
// key), so resume works from any caller fiber. It is strict by default: a
|
||||
// backend that is not ACTIVE (absent, or mid-teardown) reads as undefined
|
||||
// and we reject below, rather than handing back an unusable handle.
|
||||
const persistence = this.ctx.get('sessionPersistence')
|
||||
async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
validateAgentOptions(options.agentOptions ?? {})
|
||||
const persistence = this.runtime.ctx.get('sessionPersistence')
|
||||
if (persistence === undefined) {
|
||||
throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
|
||||
}
|
||||
return this.resumeWith(persistence, options)
|
||||
return this.resumeWith(ownerCtx, persistence, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume against an EXPLICIT persistence handle. Factored out of {@link resume}
|
||||
* so the config-driven path can pass the handle it obtained from a
|
||||
* `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the
|
||||
* service's own fiber) did not inject `sessionPersistence`, so reading it
|
||||
* there from inside the inject child trips the cordis inject guard. The
|
||||
* sessions store + registry are still read through `this.ctx` (both are in
|
||||
* AgentLoop's static inject, so they resolve fine).
|
||||
*/
|
||||
private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
this.validateAgentOptions(options.agentOptions ?? {})
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const { meta, events } = await persistence.load(options.resumeSessionId)
|
||||
// Re-check the agent id AFTER the await: the pre-load check above can go
|
||||
// stale while load() is pending (a concurrent resume/create may register the
|
||||
// same id). Re-checking immediately before prepare()/start keeps the
|
||||
// "no orphaned session on a duplicate id" guarantee under concurrency.
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
// Reconstruct the live session with the FULL persisted header (createdAt,
|
||||
// cwd, lineage) so resume preserves identity, not just the cwd. The seed
|
||||
// events make lastTurnNumber/deriveMessages continue; the backend already
|
||||
// has state (cursor) from the load above, so onCreated is a no-op and the
|
||||
// seed is not re-persisted. prepare() (not create()) so the session
|
||||
// lifecycle folds into the agent's composite effect (ordered teardown).
|
||||
const session = this.ctx.sessions.prepare(options.resumeSessionId, {
|
||||
seed: events,
|
||||
meta: {
|
||||
createdAt: meta.createdAt,
|
||||
...meta.cwd !== undefined ? { cwd: meta.cwd } : {},
|
||||
...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {},
|
||||
// Reconstruct the seed boundary from the persisted header, NOT from
|
||||
// `events.length` (the resume seeds the WHOLE stored log).
|
||||
...meta.seedLength !== undefined ? { seedLength: meta.seedLength } : {},
|
||||
},
|
||||
})
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'resume')
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject a duplicate agent id BEFORE the session is entered into the store, so
|
||||
* a failed factory call never leaves an orphaned live session (and lazy
|
||||
* persistence state) behind. `register()` enforces the same uniqueness, but
|
||||
* only after the session has already entered the store.
|
||||
*/
|
||||
private assertAgentIdFree(id: AgentId): void {
|
||||
if (this.ctx.agents.get(id) !== undefined) {
|
||||
throw new Error(`agent "${id}" is already registered`)
|
||||
/** Resume through an explicit persistence handle used by the deferred config path. */
|
||||
private async resumeWith(
|
||||
ownerCtx: Context,
|
||||
persistence: SessionPersistence,
|
||||
options: ResumeAgentOptions,
|
||||
): Promise<AgentHandle> {
|
||||
const transaction = new AgentCreationTransaction(
|
||||
this.runtime.ctx,
|
||||
ownerCtx,
|
||||
this.ownership,
|
||||
options.agentId,
|
||||
options.signal,
|
||||
)
|
||||
try {
|
||||
const loaded = await transaction.waitFor(persistence.load(options.resumeSessionId))
|
||||
transaction.assertActive()
|
||||
const session = this.runtime.ctx.sessions.prepare(options.resumeSessionId, {
|
||||
seed: loaded.events,
|
||||
meta: {
|
||||
createdAt: loaded.meta.createdAt,
|
||||
...loaded.meta.cwd === undefined ? {} : { cwd: loaded.meta.cwd },
|
||||
...loaded.meta.parentSession === undefined ? {} : { parentSession: loaded.meta.parentSession },
|
||||
...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
|
||||
},
|
||||
})
|
||||
const agent = transaction.prepare(options.agentOptions ?? {}, session)
|
||||
await transaction.waitFor(options.setup?.(agent.ctx))
|
||||
transaction.assertActive()
|
||||
return transaction.publish('resume')
|
||||
} catch (error: unknown) {
|
||||
await transaction.dispose(error instanceof Error ? error : new Error(String(error)))
|
||||
throw error
|
||||
} finally {
|
||||
transaction.finishWrapper()
|
||||
}
|
||||
}
|
||||
|
||||
/** Validate merge-extended options the loop owns before any session is prepared or loaded. */
|
||||
private validateAgentOptions(options: AgentOptions): void {
|
||||
const { maxParallelToolCalls } = options
|
||||
if (maxParallelToolCalls !== undefined && (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1)) {
|
||||
throw new Error('maxParallelToolCalls must be a positive integer')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared: construct a ReactLoopAgent over a PREPARED (not-yet-entered)
|
||||
* session, then build the ONE composite effect that owns the whole agent
|
||||
* lifecycle — session entry, registry registration, and the loop. Keeping all
|
||||
* three in a SINGLE effect (not sibling effects) is load-bearing: a fiber
|
||||
* unload disposes sibling effects CONCURRENTLY (`Promise.all`), which would
|
||||
* race the session detach against the loop's closing flush and drop the
|
||||
* closing `turn/end`. Inside one effect the disposers run as an ORDERED LIFO
|
||||
* chain — the runtime awaits each disposer's returned promise before the next:
|
||||
*
|
||||
* yield session-detach (disposed LAST — detach onAppend + remove entry)
|
||||
* yield register (disposed 2nd — unregister)
|
||||
* yield stop-and-drain (disposed FIRST — request loop stop, await agent.done)
|
||||
*
|
||||
* So on teardown: the loop is stopped and AWAITED to exit (its final
|
||||
* `session/flush` + `turn/end` fire through the still-attached `onAppend`),
|
||||
* THEN the agent is unregistered, THEN the session is detached — capturing the
|
||||
* closing events before detach, whether the trigger is the handle's `dispose()`
|
||||
* OR a fiber unload. Rollback safety: each yield runs before the next mutation,
|
||||
* so a throwing `session/created`/`agent/created` listener unwinds the
|
||||
* already-yielded disposers instead of leaking.
|
||||
*
|
||||
* `source` says why the session began ({@link SessionStartSource}); it is
|
||||
* emitted as `agent/session-start` once, AFTER the agent is registered (so a
|
||||
* listener can resolve the agent via `ctx.agents.get(id)` and `inject()` into
|
||||
* it) and BEFORE the loop starts its first turn. The emit is contained: a
|
||||
* throwing session-start listener must not abort agent construction — it is
|
||||
* logged, and the agent still starts. (Unlike a turn-boundary throw, there is
|
||||
* no open turn here to balance; the durable evidence of a session-start hook
|
||||
* is whatever it `inject()`ed.)
|
||||
*
|
||||
* Returns the agent plus the composite effect's disposer (`disposeAgent`).
|
||||
*/
|
||||
private start(
|
||||
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
|
||||
): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
|
||||
const agent = new ReactLoopAgent(this.ctx, id, options, session)
|
||||
const dispose = this.ctx.effect(function* (this: AgentLoop) {
|
||||
yield this.ctx.sessions.enter(session)
|
||||
this.ctx.sessions.announce(session)
|
||||
yield this.ctx.agents.register(agent)
|
||||
// Fire AFTER register (a listener can ctx.agents.get(id) + inject()) and
|
||||
// BEFORE the loop's first turn. Contained: a throwing listener is logged,
|
||||
// never aborts construction (no open turn to balance here).
|
||||
try {
|
||||
this.ctx.emit('agent/session-start', agent, source)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`agent "${id}": agent/session-start listener threw: ${String(error)}`)
|
||||
}
|
||||
const stop = agent.start()
|
||||
// Disposed FIRST (LIFO): request loop stop (sync), then AWAIT the loop's
|
||||
// actual exit so its closing flush lands while onAppend (yielded above,
|
||||
// disposed later) is still attached.
|
||||
yield async () => { stop(); await agent.done }
|
||||
}.bind(this), 'agentLoop.start()')
|
||||
return { agent, disposeAgent: async () => { await dispose() } }
|
||||
}
|
||||
|
||||
/**
|
||||
* Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The
|
||||
* handle's `dispose()` runs the composite effect's disposer (see
|
||||
* {@link start}) — which stops the loop, awaits its exit (final flush
|
||||
* captured), unregisters the agent, and detaches the session, in that order.
|
||||
* The same composite effect is what a fiber unload disposes, so both teardown
|
||||
* triggers honor the ordering identically.
|
||||
*
|
||||
* `dispose()` is MEMOIZED: the underlying cordis effect disposer is
|
||||
* single-shot (a second call returns immediately because the effect's epoch is
|
||||
* already cleared, NOT awaiting the in-flight teardown), so concurrent/repeated
|
||||
* `dispose()` calls would otherwise resolve before the first call's
|
||||
* `await agent.done` + final flush completed. Memoizing the promise makes every
|
||||
* caller observe the SAME quiescence boundary, honoring the
|
||||
* `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
|
||||
* helper).
|
||||
*/
|
||||
private startOwned(id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource): AgentHandle {
|
||||
const { agent, disposeAgent } = this.start(id, options, session, source)
|
||||
let disposing: Promise<void> | undefined
|
||||
return { agent, dispose: () => (disposing ??= disposeAgent()) }
|
||||
}
|
||||
}
|
||||
|
||||
export default AgentLoop
|
||||
|
||||
Reference in New Issue
Block a user