mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
576 lines
24 KiB
TypeScript
576 lines
24 KiB
TypeScript
/**
|
|
* 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, FiberState, Service } from 'cordis'
|
|
import { randomUUID } from 'node:crypto'
|
|
import z from 'schemastery'
|
|
import { emitAgentEvent } from '@deepseek-ai/dsh-agent'
|
|
import type {
|
|
Agent,
|
|
AgentFactory,
|
|
AgentHandle,
|
|
AgentOptions,
|
|
CreateAgentOptions,
|
|
ResumeAgentOptions,
|
|
SessionStartSource,
|
|
} from '@deepseek-ai/dsh-agent'
|
|
import { errorChain } from '@deepseek-ai/dsh-llm'
|
|
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 { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.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: live agent teardowns plus config startup work. */
|
|
class FactoryOwnership {
|
|
private accepting = true
|
|
private readonly teardown = new AbortController()
|
|
private readonly inactive = Promise.withResolvers<void>()
|
|
private readonly liveAgents = new Set<() => Promise<void>>()
|
|
private startupTasks = new Set<Promise<void>>()
|
|
|
|
constructor(private readonly fiber: Context['fiber']) {}
|
|
|
|
/** Aborts (reason: `agent loop is not active` error) when factory teardown begins. */
|
|
get signal(): AbortSignal {
|
|
return this.teardown.signal
|
|
}
|
|
|
|
isActive(): boolean {
|
|
return this.accepting && !INACTIVE_STATES.has(this.fiber.state)
|
|
}
|
|
|
|
/** Track one live agent's shared teardown until it has run. */
|
|
track(dispose: () => Promise<void>): () => void {
|
|
this.liveAgents.add(dispose)
|
|
return () => { this.liveAgents.delete(dispose) }
|
|
}
|
|
|
|
/** Join config startup work that begins before an agent exists. */
|
|
trackStartup(task: Promise<void>): void {
|
|
this.startupTasks.add(task)
|
|
const forget = () => { this.startupTasks.delete(task) }
|
|
void task.then(forget, forget)
|
|
}
|
|
|
|
/** Join one public create/resume continuation; factory dispose awaits its settlement. */
|
|
trackWrapper(task: Promise<unknown>): void {
|
|
this.trackStartup(task.then(() => undefined, () => undefined))
|
|
}
|
|
|
|
/** Resolve `task`, or stop waiting when factory teardown begins. */
|
|
async waitWhileActive(task: Promise<void>): Promise<void> {
|
|
await Promise.race([task, this.inactive.promise])
|
|
}
|
|
|
|
async dispose(): Promise<void> {
|
|
this.accepting = false
|
|
this.teardown.abort(new Error('agent loop is not active'))
|
|
this.inactive.resolve()
|
|
await Promise.all([
|
|
...[...this.liveAgents].map(dispose => dispose()),
|
|
...this.startupTasks,
|
|
])
|
|
}
|
|
}
|
|
|
|
/** Await `operation`, or throw the signal's reason as soon as it aborts. */
|
|
async function raceAbort<T>(operation: PromiseLike<T> | T, signal: AbortSignal, id: SessionId): Promise<T> {
|
|
const toAbortError = (): Error => signal.reason instanceof Error
|
|
? signal.reason
|
|
: new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
|
|
if (signal.aborted) throw toAbortError()
|
|
const aborted = Promise.withResolvers<never>()
|
|
const listener = (): void => { aborted.reject(toAbortError()) }
|
|
signal.addEventListener('abort', listener, { once: true })
|
|
try {
|
|
return await Promise.race([Promise.resolve(operation), aborted.promise])
|
|
} finally {
|
|
signal.removeEventListener('abort', listener)
|
|
}
|
|
}
|
|
|
|
/** Resolve the deployment-wide scheduler cap at the owning config boundary. */
|
|
function resolveMaxParallelToolCalls(value: number | undefined): number {
|
|
const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
|
|
if (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1) {
|
|
throw new Error('maxParallelToolCalls must be a positive integer')
|
|
}
|
|
return maxParallelToolCalls
|
|
}
|
|
|
|
/** Reject an output-token cap that cannot be represented exactly on the request wire. */
|
|
function assertAgentOptions(options: AgentOptions): void {
|
|
if (options.maxTokens !== undefined
|
|
&& (!Number.isSafeInteger(options.maxTokens) || options.maxTokens <= 0)) {
|
|
throw new TypeError('agent maxTokens must be a positive safe integer')
|
|
}
|
|
}
|
|
|
|
/** Prepared-but-unpublished agent resources sharing one memoized teardown. */
|
|
interface PreparedAgent {
|
|
agent: ReactLoopAgent
|
|
/** Aborts when the factory unloads, the caller cancels, or teardown begins — ends any setup await. */
|
|
signal: AbortSignal
|
|
/** Enter registries, announce, notify session-start, and start the machine. */
|
|
publish(source: SessionStartSource): AgentHandle
|
|
/** Reverse teardown: stop the machine, unregister, unwind the scope. Memoized. */
|
|
dispose(): Promise<void>
|
|
}
|
|
|
|
declare module 'cordis' {
|
|
interface Context {
|
|
agentLoop: AgentLoop
|
|
}
|
|
interface Events {
|
|
/**
|
|
* A declarative agent entry failed before it could publish a live agent.
|
|
* Consumers that buffer work for the configured identity use this
|
|
* transient signal to reject that work instead of waiting forever. Normal
|
|
* factory teardown suppresses failures from the cancelled startup attempt.
|
|
* @param sessionId - exact shared agent/session identity that failed startup.
|
|
* @param error - persistence, setup, or publication failure.
|
|
* @mode emit
|
|
*/
|
|
'agent-loop/config-start-failed'(sessionId: SessionId, error: unknown): void
|
|
}
|
|
}
|
|
|
|
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS }
|
|
|
|
/** Agent-loop plugin configuration. */
|
|
export interface Config {
|
|
/**
|
|
* Maximum parallel-safe calls in flight per agent step. `1` is serial;
|
|
* omission defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
|
|
*/
|
|
maxParallelToolCalls?: number
|
|
/** Agents created or resumed at plugin startup. */
|
|
agents: (AgentOptions & {
|
|
/** Stable config label used in logs and as the fresh combined-id prefix. */
|
|
id: string
|
|
/** Optional stable identity; remounts resume its materialized history, while first use creates it fresh. */
|
|
sessionId?: SessionId
|
|
/** Optional workspace for a fresh session. */
|
|
cwd?: string
|
|
/** Persisted session to resume instead of creating a fresh session. */
|
|
resumeSessionId?: SessionId
|
|
})[]
|
|
}
|
|
|
|
/** Agent-loop configuration after defaults and load-time validation. */
|
|
type ResolvedConfig = Config & { maxParallelToolCalls: number }
|
|
|
|
/** Reject self-contained identity conflicts before any configured agent starts. */
|
|
function validateConfiguredAgents(agents: Config['agents']): void {
|
|
const exactIdentities = new Map<SessionId, string>()
|
|
for (const { id, sessionId, resumeSessionId } of agents) {
|
|
const hasResumeId = resumeSessionId !== undefined && resumeSessionId !== ''
|
|
if (sessionId !== undefined && hasResumeId) {
|
|
throw new Error(`agent "${id}": sessionId and resumeSessionId are mutually exclusive`)
|
|
}
|
|
const exactIdentity = hasResumeId ? resumeSessionId : sessionId
|
|
if (exactIdentity === undefined) continue
|
|
const firstId = exactIdentities.get(exactIdentity)
|
|
if (firstId !== undefined) {
|
|
throw new Error(`agents "${firstId}" and "${id}" use duplicate exact session identity "${exactIdentity}"`)
|
|
}
|
|
exactIdentities.set(exactIdentity, id)
|
|
}
|
|
}
|
|
|
|
/** Concrete agent factory and driver service. */
|
|
export class AgentLoop extends Service implements AgentFactory {
|
|
static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']
|
|
|
|
/** Runtime schema for declarative agents. */
|
|
static Config = z.object({
|
|
maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS),
|
|
agents: z.array(z.object({
|
|
id: z.string().required(),
|
|
sessionId: z.string().min(1),
|
|
provider: z.string(),
|
|
model: z.string(),
|
|
maxTokens: z.number().step(1).min(1).max(Number.MAX_SAFE_INTEGER),
|
|
cwd: z.string(),
|
|
resumeSessionId: z.string(),
|
|
})).default([]),
|
|
}) as z<Config>
|
|
|
|
/** Validated configuration owned by the agent-loop service. */
|
|
readonly config: ResolvedConfig
|
|
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, config: Config) {
|
|
super(ctx, 'agentLoop')
|
|
this.config = {
|
|
...config,
|
|
maxParallelToolCalls: resolveMaxParallelToolCalls(config.maxParallelToolCalls),
|
|
}
|
|
validateConfiguredAgents(this.config.agents)
|
|
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('provider', context => context.agent?.options.provider)
|
|
ctx.systemPrompt.variable('model', context => context.agent?.options.model)
|
|
ctx.systemPrompt.variable('cwd', context => context.agent?.session.header.cwd)
|
|
|
|
for (const { id, sessionId, cwd, resumeSessionId, ...options } of this.config.agents) {
|
|
const meta = cwd === undefined ? {} : { cwd }
|
|
if (resumeSessionId === undefined || resumeSessionId === '') {
|
|
const configuredId = sessionId ?? SessionId(`${id}-session-${randomUUID()}`)
|
|
const persistence = sessionId === undefined ? undefined : ctx.get('sessionPersistence')
|
|
if (persistence === undefined) {
|
|
this.create(configuredId, options, meta)
|
|
} else {
|
|
const startup = this.restoreOrCreateConfigured(ctx, persistence, configuredId, options, meta).catch((error: unknown) => {
|
|
this.reportConfiguredStartupFailure(id, 'restore', configuredId, error)
|
|
})
|
|
this.ownership.trackStartup(startup)
|
|
}
|
|
continue
|
|
}
|
|
ctx.effect(() => {
|
|
const fiber = ctx.inject(['sessionPersistence'], (childCtx: Context) => {
|
|
void this.resumeWith(ctx, childCtx.sessionPersistence, {
|
|
resumeSessionId,
|
|
agentOptions: options,
|
|
}).catch((error: unknown) => {
|
|
this.reportConfiguredStartupFailure(id, 'resume', resumeSessionId, error)
|
|
})
|
|
})
|
|
return fiber.dispose
|
|
}, `agentLoop.resume(${id})`)
|
|
}
|
|
}
|
|
|
|
/** Report a contained declarative-start failure to identity-bound consumers. */
|
|
private reportConfiguredStartupFailure(
|
|
configId: string,
|
|
action: 'restore' | 'resume',
|
|
sessionId: SessionId,
|
|
error: unknown,
|
|
): void {
|
|
if (!this.ownership.isActive()) return
|
|
this.ctx.logger.warn(`agent "${configId}": config-driven ${action} of "${sessionId}" failed: ${errorChain(error)}`)
|
|
const args: unknown[] = ['agent-loop/config-start-failed', sessionId, error]
|
|
for (const callback of this.ctx.events.dispatch('emit', args)) {
|
|
try {
|
|
const returned: unknown = callback(...args)
|
|
void Promise.resolve(returned).catch((listenerError: unknown) => {
|
|
this.ctx.logger.warn(`agent "${configId}": config-start-failed listener rejected: ${errorChain(listenerError)}`)
|
|
})
|
|
} catch (listenerError: unknown) {
|
|
this.ctx.logger.warn(`agent "${configId}": config-start-failed listener threw: ${errorChain(listenerError)}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Restore a materialized exact config identity on remount, or create it on first use. */
|
|
private async restoreOrCreateConfigured(
|
|
ownerCtx: Context,
|
|
persistence: SessionPersistence,
|
|
sessionId: SessionId,
|
|
agentOptions: AgentOptions,
|
|
meta: Pick<SessionHeader, 'cwd'>,
|
|
): Promise<void> {
|
|
await this.waitForDrainingConfiguredIdentity(ownerCtx, sessionId)
|
|
if (!this.ownership.isActive()) return
|
|
try {
|
|
await this.resumeWith(ownerCtx, persistence, { resumeSessionId: sessionId, agentOptions })
|
|
return
|
|
} catch (error: unknown) {
|
|
if (!this.ownership.isActive()) return
|
|
// A load is the per-id serialization barrier for eager write-behind and
|
|
// lifecycle retirement. Only a genuinely absent artifact falls back to
|
|
// first creation; corruption and backend failures stay loud.
|
|
const exists = (await persistence.list()).some(header => header.id === sessionId)
|
|
if (exists) throw error
|
|
}
|
|
this.create(sessionId, agentOptions, meta)
|
|
}
|
|
|
|
/** Wait for a draining same-id lifecycle to finish registry teardown. */
|
|
private async waitForDrainingConfiguredIdentity(ownerCtx: Context, sessionId: SessionId): Promise<void> {
|
|
// Only an id still occupying a registry needs waiting for; a live healthy
|
|
// occupant is a collision the create/resume below will surface itself.
|
|
if (ownerCtx.agents.get(sessionId) === undefined && ownerCtx.sessions.get(sessionId) === undefined) return
|
|
|
|
const released = Promise.withResolvers<void>()
|
|
const checkReleased = (): void => {
|
|
if (ownerCtx.agents.get(sessionId) === undefined && ownerCtx.sessions.get(sessionId) === undefined) {
|
|
released.resolve()
|
|
}
|
|
}
|
|
const disposeAgentListener = ownerCtx.on('agent/disposed', checkReleased)
|
|
const disposeSessionListener = ownerCtx.on('session/disposed', checkReleased)
|
|
try {
|
|
checkReleased()
|
|
await this.ownership.waitWhileActive(released.promise)
|
|
} finally {
|
|
disposeAgentListener()
|
|
disposeSessionListener()
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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.
|
|
*/
|
|
private prepare(ownerCtx: Context, id: SessionId, options: AgentOptions, session: Session, callerSignal?: AbortSignal): 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 })
|
|
}
|
|
const loopCtx = this.runtime.ctx
|
|
|
|
// Deactivation fuses three owners, each with its own reason: the caller's
|
|
// cancellation signal, the owner fiber's unload, and factory teardown.
|
|
// It is registered BEFORE any resource exists, over mutable slots, so an
|
|
// unload arriving while the scope is still minting finds a working
|
|
// disposer instead of a leak.
|
|
const abort = new AbortController()
|
|
const onCallerAbort = (): void => {
|
|
abort.abort(callerSignal?.reason instanceof Error
|
|
? callerSignal.reason
|
|
: new Error(`agent "${id}" creation aborted`, { cause: callerSignal?.reason }))
|
|
}
|
|
const onFactoryTeardown = (): void => { abort.abort(this.ownership.signal.reason) }
|
|
callerSignal?.addEventListener('abort', onCallerAbort, { once: true })
|
|
this.ownership.signal.addEventListener('abort', onFactoryTeardown, { once: true })
|
|
|
|
let machine: ReactLoopAgent | undefined
|
|
let detachSession: (() => void) | undefined
|
|
let detachAgent: (() => void) | undefined
|
|
let disposing: Promise<void> | undefined
|
|
const machineReady = Promise.withResolvers<void>()
|
|
// Reverse teardown, memoized so every racing owner awaits one quiescence:
|
|
// stop the machine, leave the registries, unwind the scope, release
|
|
// bookkeeping.
|
|
const dispose = (ownerTriggered = false): Promise<void> => (disposing ??= (async () => {
|
|
abort.abort(new Error(`agent "${id}" lifecycle disposed`))
|
|
callerSignal?.removeEventListener('abort', onCallerAbort)
|
|
this.ownership.signal.removeEventListener('abort', onFactoryTeardown)
|
|
try {
|
|
// Disposal IS a disposed-cause cancel followed by quiescence. New work
|
|
// sent after this point is the sender's bug — the registries are about
|
|
// to drop the agent, so nothing should still hold it.
|
|
if (machine === undefined) await machineReady.promise
|
|
if (machine !== undefined) {
|
|
machine.cancel({ kind: 'disposed' })
|
|
// Drain to TRUE quiescence: cancel's own synchronous event chain
|
|
// (running→idle) can legitimately re-enter through an automation
|
|
// listener (goal-session's idle drive) and replace `done` with a
|
|
// fresh admission before this await captures it. The replacement
|
|
// work is cancelled and drained in turn until the slot stabilizes.
|
|
let done = machine.done
|
|
while (true) {
|
|
await Promise.allSettled([done])
|
|
if (machine.done === done) break
|
|
done = machine.done
|
|
machine.cancel({ kind: 'disposed' })
|
|
}
|
|
await machine.scope.dispose()
|
|
}
|
|
} finally {
|
|
try {
|
|
detachAgent?.()
|
|
detachSession?.()
|
|
} finally {
|
|
untrack()
|
|
if (!ownerTriggered) await unfollowOwner()
|
|
}
|
|
}
|
|
})())
|
|
const untrack = this.ownership.track(dispose)
|
|
let unfollowOwner: () => Promise<void> | void
|
|
try {
|
|
unfollowOwner = ownerCtx.effect(() => () => {
|
|
// Owner disposal owns the same quiescence boundary. Its teardown skips
|
|
// unregistering this already-running owner effect from inside itself.
|
|
if (disposing !== undefined) return
|
|
abort.abort(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
|
|
return dispose(true)
|
|
}, `agentLoop.lifecycle(${id})`)
|
|
/* v8 ignore start -- ctx.effect throws only on an inactive fiber, which assertActive() above already rejected */
|
|
} catch (error: unknown) {
|
|
untrack()
|
|
callerSignal?.removeEventListener('abort', onCallerAbort)
|
|
this.ownership.signal.removeEventListener('abort', onFactoryTeardown)
|
|
throw error
|
|
}
|
|
/* v8 ignore stop */
|
|
|
|
const assertLive = (): void => {
|
|
if (!abort.signal.aborted) return
|
|
// Every fused abort source carries an Error reason: onCallerAbort and
|
|
// raceAbort wrap non-Error caller reasons, and the factory/lifecycle
|
|
// owners abort with constructed Errors.
|
|
/* v8 ignore next -- unreachable String() arm, see above */
|
|
throw abort.signal.reason instanceof Error ? abort.signal.reason : new Error(String(abort.signal.reason))
|
|
}
|
|
try {
|
|
const agent = machine = new ReactLoopAgent(loopCtx, id, options, session)
|
|
machineReady.resolve()
|
|
assertLive()
|
|
|
|
return {
|
|
agent,
|
|
signal: abort.signal,
|
|
publish: (source) => {
|
|
assertLive()
|
|
detachSession = agent.ctx.sessions.enter(session)
|
|
detachAgent = loopCtx.agents.enter(agent, ownerCtx.agent)
|
|
agent.ctx.sessions.announce(session)
|
|
assertLive()
|
|
loopCtx.agents.announce(agent)
|
|
assertLive()
|
|
// A synchronous announce/session-start listener may have started
|
|
// teardown; the machine is already live (send() works from the
|
|
// session-start seam), so only the liveness recheck is owed.
|
|
emitAgentEvent(loopCtx, agent, 'agent/session-start', source)
|
|
assertLive()
|
|
return { agent, dispose }
|
|
},
|
|
dispose,
|
|
}
|
|
} catch (error: unknown) {
|
|
machineReady.resolve()
|
|
void dispose()
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Create an agent and session under one caller-supplied identity, owned by
|
|
* the accessing fiber. Constructor-driven config calls mint a fresh combined
|
|
* id before entering this boundary.
|
|
* @param id - shared agent/session identity.
|
|
* @param options - concrete loop options.
|
|
* @param meta - optional fresh-session workspace metadata.
|
|
* @returns the published running agent.
|
|
*/
|
|
create(id: SessionId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): Agent {
|
|
const session = this.runtime.ctx.sessions.prepare(id, { meta })
|
|
const prepared = this.prepare(this.ctx, id, options, session)
|
|
try {
|
|
return prepared.publish('startup').agent
|
|
} catch (error: unknown) {
|
|
void prepared.dispose()
|
|
throw error
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Create an owned agent on a caller-supplied session id.
|
|
* @param ownerCtx - caller context that structurally owns the lifecycle.
|
|
* @param options - identities, session seed/metadata, loop options, setup, and cancellation.
|
|
* @returns the published handle.
|
|
*/
|
|
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
|
|
const session = this.runtime.ctx.sessions.prepare(options.sessionId, {
|
|
...options.seed === undefined ? {} : { seed: options.seed },
|
|
...options.meta === undefined ? {} : { meta: options.meta },
|
|
})
|
|
const prepared = this.prepare(ownerCtx, options.sessionId, options.agentOptions ?? {}, session, options.signal)
|
|
const published = (async () => {
|
|
try {
|
|
await raceAbort(options.setup?.(prepared.agent.ctx), prepared.signal, options.sessionId)
|
|
return prepared.publish('startup')
|
|
} catch (error: unknown) {
|
|
await prepared.dispose()
|
|
throw error
|
|
}
|
|
})()
|
|
this.ownership.trackWrapper(published)
|
|
return published
|
|
}
|
|
|
|
/**
|
|
* 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(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle> {
|
|
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(ownerCtx, persistence, options)
|
|
}
|
|
|
|
/** Resume through an explicit persistence handle used by the deferred config path. */
|
|
private resumeWith(
|
|
ownerCtx: Context,
|
|
persistence: SessionPersistence,
|
|
options: ResumeAgentOptions,
|
|
): Promise<AgentHandle> {
|
|
const id = options.resumeSessionId
|
|
const published = (async () => {
|
|
// The load may outlive its owner: race it against caller cancellation,
|
|
// owner-fiber unload, and factory teardown so a never-settling backend
|
|
// cannot pin the identity.
|
|
const ownerAbort = new AbortController()
|
|
const unfollowOwner = ownerCtx.effect(() => () => {
|
|
ownerAbort.abort(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
|
|
}, `agentLoop.resume-load(${id})`)
|
|
const fused = AbortSignal.any([
|
|
...options.signal === undefined ? [] : [options.signal],
|
|
ownerAbort.signal,
|
|
this.ownership.signal,
|
|
])
|
|
let loaded: Awaited<ReturnType<SessionPersistence['load']>>
|
|
try {
|
|
loaded = await raceAbort(persistence.load(id), fused, id)
|
|
} finally {
|
|
await unfollowOwner()
|
|
}
|
|
ownerCtx.fiber.assertActive()
|
|
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
|
const session = this.runtime.ctx.sessions.prepare(id, {
|
|
seed: loaded.events,
|
|
meta: loaded.meta,
|
|
})
|
|
const prepared = this.prepare(ownerCtx, id, options.agentOptions ?? {}, session, options.signal)
|
|
try {
|
|
await raceAbort(options.setup?.(prepared.agent.ctx), prepared.signal, id)
|
|
return prepared.publish('resume')
|
|
} catch (error: unknown) {
|
|
await prepared.dispose()
|
|
throw error
|
|
}
|
|
})()
|
|
this.ownership.trackWrapper(published)
|
|
return published
|
|
}
|
|
}
|
|
|
|
export default AgentLoop
|