deepseek-harness/packages/agent-loop/src/index.ts

/**
 * THE concrete agent plugin: creates LoopAgents, 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.
 *
 * @module @deepseek-ai/dsh-agent-loop
 */

import { Context, Service } from 'cordis'
import { randomUUID } from 'node:crypto'
import z from 'schemastery'
import { AgentId } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentFactory, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import type {} from '@deepseek-ai/dsh-session-persistence'
import { LoopAgent } from './agent.ts'

export { LoopAgent } from './agent.ts'
export { Inbox, type InboxMessage } from './inbox.ts'
export { runLoop } from './loop.ts'

declare module 'cordis' {
  interface Context {
    agentLoop: AgentLoop
  }
}

export interface Config {
  /** Agents created from configuration at startup. */
  agents: (AgentOptions & { id: string })[]
}

/**
 * The agent-loop plugin (`ctx.agentLoop`): creates {@link LoopAgent}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.
 */
export class AgentLoop extends Service implements AgentFactory {
  static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']

  static Config: z<Config> = z.object({
    agents: z.array(z.object({
      id: z.string().required(),
      model: z.string(),
      systemPrompt: z.string(),
    })).default([]),
  })

  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()')
    for (const { id, ...options } of config.agents) {
      this.create(id, options)
    }
  }

  /**
   * Config-driven create: an agent on a FRESH, non-colliding session id per run
   * (`${id}-session-<uuid>`, no cwd). 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.
   *
   * TODO(sub-agents): spawn/fork land here — accept a parent agent reference;
   * fork seeds the new Session with the parent's event log, spawn starts
   * fresh; the child is returned as a regular Agent handle.
   */
  create(id: string, options: AgentOptions = {}): LoopAgent {
    this.assertAgentIdFree(id)
    const session = this.ctx.sessions.create(`${id}-session-${randomUUID()}`, { meta: {} })
    return this.start(AgentId(id), options, session)
  }

  /**
   * Programmatic factory create ({@link AgentFactory}): an agent on a
   * caller-supplied `sessionId` (NOT `${id}-session`), with optional session
   * metadata (validated `cwd`, lineage). The ACP bridge uses this so the
   * client-generated session id becomes the live/persisted session id.
   */
  createAgent(options: CreateAgentOptions): Agent {
    // Check the agent id BEFORE creating the session: register() would reject a
    // duplicate id only AFTER sessions.create(), leaving an orphaned live
    // session (and lazy persistence state) that blocks reuse of that id.
    this.assertAgentIdFree(options.agentId)
    const session = this.ctx.sessions.create(options.sessionId, { meta: options.meta ?? {} })
    return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session)
  }

  /**
   * 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`; throws a typed 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.
   */
  async resume(options: ResumeAgentOptions): Promise<Agent> {
    this.assertAgentIdFree(options.agentId)
    const persistence = this.ctx.sessionPersistence
    // `sessionPersistence` is declaration-merged onto Context as non-optional,
    // but the service is only present when a backend plugin is loaded — and
    // AgentLoop deliberately does NOT inject it (that would pend non-persistent
    // demos forever). So the runtime value can be undefined; the type cannot.
    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
    if (persistence === undefined) {
      throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
    }
    const { meta, events } = await persistence.load(SessionId(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 sessions.create() 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.
    const session = this.ctx.sessions.create(options.resumeSessionId, {
      seed: events,
      meta: {
        createdAt: meta.createdAt,
        ...meta.cwd !== undefined ? { cwd: meta.cwd } : {},
        ...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {},
      },
    })
    return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session)
  }

  /**
   * Reject a duplicate agent id BEFORE any session is created, so a failed
   * factory call never leaves an orphaned live session (and lazy persistence
   * state) behind. `register()` enforces the same uniqueness, but only after
   * `sessions.create()` has already run.
   */
  private assertAgentIdFree(id: string): void {
    if (this.ctx.agents.get(id) !== undefined) {
      throw new Error(`agent "${id}" is already registered`)
    }
  }

  /** Shared: construct a LoopAgent, register it, and start its loop (LIFO). */
  private start(id: AgentId, options: AgentOptions, session: Session): LoopAgent {
    const agent = new LoopAgent(this.ctx, id, options, session)
    // Generator effect: stop and unregister are independent disposables
    // (LIFO), so a throwing stop() cannot leak the registry entry.
    this.ctx.effect(function* (this: AgentLoop) {
      yield this.ctx.agents.register(agent)
      yield agent.start()
    }.bind(this), 'agentLoop.start()')
    return agent
  }
}

export default AgentLoop