dsh-runtime.ts

import type {
  AgentCompactionResult,
  AgentForkResult,
  AgentRuntime,
  AgentRuntimeContext,
  AgentRuntimeEvent,
  PromptInput,
} from "../agent-adapter.ts"
import type { HistoryEntry, MuxFrame, RpcRequest } from "@deepseek-ai/dsh-host-apiproxy/api"
import { SessionId, type SessionEvent, type SessionId as DshSessionId } from "@deepseek-ai/dsh-session/types"
import { createPartId } from "../../id/index.ts"
import { DshEventMapper } from "./dsh-event-mapper.ts"
import type { DshConnection } from "./connection.ts"
import { DshConversationStore, type DshMessageCursor } from "./dsh-conversation-store.ts"
import { DshSubagentLinkRegistry } from "./dsh-subagent-link-registry.ts"
import { DshSubagentProjector } from "./dsh-subagent-projector.ts"
import type { DshSkillCatalog } from "./dsh-skill-catalog.ts"
import { DshRpcResultError } from "./errors.ts"

const COMPACTION_TERMINAL_GRACE_MS = 500
const SETTLEMENT_WATCHDOG_INTERVAL_MS = 5_000
const HISTORY_PROBE_MAX_FAILURES = 5

export interface DshRuntimeOptions {
  connection: DshConnection
  store: DshConversationStore
  provider?: string
  model?: string
  /** Optional adapter-owned reasoning effort forwarded to `session.selectModel`. */
  reasoningEffort?: string
  /**
   * Desired host plan mode for this runtime (the `/plan` host command).
   * The runtime aligns the session's recorded plan state with this value on
   * start and after fork switches: plan variants set it on, regular variants
   * set it off. Fork children inherit the forked prefix's state.
   */
  planMode?: boolean
  /** Settlement watchdog cadence; primarily for tests. */
  watchdogIntervalMs?: number
  /**
   * Shared (parentSessionId, callId) → OpenCode child session link registry
   * for native subagent projection; absent runtimes skip the binding step.
   */
  subagentLinks?: DshSubagentLinkRegistry
  /** Shared host skill catalog mirror; refreshed whenever a runtime starts. */
  skillCatalog?: DshSkillCatalog
  /** Subagent child discovery retry cadence; primarily for tests. */
  subagentPollIntervalMs?: number
}

interface ActivePrompt {
  generation: number
  rpcId: string
  assistantMessageId: string
  messageId: string
  /** Turn opened for this prompt; undefined until the first turn/start lands. */
  turn: number | undefined
  settlement: {
    resolve: () => void
    reject: (error: Error) => void
    failure?: Error
  }
  settled: boolean
}

interface CompactionTerminalOutcome {
  sequence: number
  result?: AgentCompactionResult
  error?: string
}

/**
 * Session-level Runtime for one OpenCode session backed by a DSH session on
 * the shared connection. Events are translated to the unified
 * AgentRuntimeEvent vocabulary; prompt settlement rides the DSH turn
 * lifecycle (`turn/start` → `turn/end`), with a history-based catch-up
 * watchdog so a dropped downlink never strands an active prompt.
 */
export class DshRuntime implements AgentRuntime {
  private readonly context: AgentRuntimeContext
  private readonly connection: DshConnection
  private readonly store: DshConversationStore
  private readonly provider: string | undefined
  private readonly model: string | undefined
  private readonly reasoningEffort: string | undefined
  private readonly planMode: boolean
  private readonly watchdogIntervalMs: number
  private readonly subagentLinks: DshSubagentLinkRegistry | undefined
  private readonly skillCatalog: DshSkillCatalog | undefined
  private readonly subagentPollIntervalMs: number | undefined
  private readonly listeners = new Set<(event: AgentRuntimeEvent) => void>()
  private readonly logger: AgentRuntimeContext["logger"]

  private dshSessionId: string | undefined
  private started = false
  private projector: DshSubagentProjector | null = null
  private lastSeq = -1
  private currentTurn = 0
  private lastTurnEndSeq: number | undefined
  private readonly mapper: DshEventMapper

  private promptGeneration = 0
  private activePrompt: ActivePrompt | null = null
  private watchdogTimer: ReturnType<typeof setTimeout> | null = null
  private historyProbeFailures = 0

  private eventChain: Promise<void> = Promise.resolve()
  private readonly unsubscribes: Array<() => void> = []

  private compactionEndSeen = 0
  private compactionTerminalOutcome: CompactionTerminalOutcome | null = null

  constructor(context: AgentRuntimeContext, options: DshRuntimeOptions) {
    this.context = context
    this.connection = options.connection
    this.store = options.store
    this.provider = options.provider
    this.model = options.model
    this.reasoningEffort = options.reasoningEffort?.trim() || undefined
    this.planMode = options.planMode ?? false
    this.watchdogIntervalMs = options.watchdogIntervalMs ?? SETTLEMENT_WATCHDOG_INTERVAL_MS
    this.subagentLinks = options.subagentLinks
    this.skillCatalog = options.skillCatalog
    this.subagentPollIntervalMs = options.subagentPollIntervalMs
    this.logger = context.logger
    this.mapper = new DshEventMapper({
      sessionId: context.sessionId,
      generatePartId: () => createPartId(),
      onUnmapped: (type, reason) => this.warnUnprojectedEvent(type, "mapping", reason),
    })
  }

  async start(): Promise<void> {
    if (this.started) throw new Error("Runtime already started")
    await this.connection.start()

    const persisted = this.store.getSession(this.context.sessionId)
    let sessionId = persisted.activeSessionId
    if (sessionId) {
      const exists = await this.verifySession(sessionId)
      if (!exists) {
        this.logger.warn("Stored DSH session no longer exists; creating a fresh backend session", {
          sessionId: this.context.sessionId,
          dshSessionId: sessionId,
        })
        sessionId = undefined
      }
    }
    if (!sessionId) {
      sessionId = await this.createBackendSession()
      this.store.setSession(this.context.sessionId, { activeSessionId: sessionId })
    }
    this.dshSessionId = sessionId

    // Refresh the host skill mirror while the backend session id is at hand;
    // best-effort and never blocks runtime startup.
    if (this.skillCatalog) void this.skillCatalog.refresh(SessionId(sessionId), this.context.directory)

    await this.applyModelSelection()

    // Align the host session's plan mode with this runtime's desired state
    // (plan variants ON, regular variants OFF) — covers fresh creation,
    // restore after restart, and agent switches between the two variants.
    await this.syncPlanMode(sessionId)

    this.installSubscriptions()

    // Native subagent projection: discover DSH child sessions spawned by this
    // session's model and replay their transcripts into the mirrored OpenCode
    // child sessions (see DshSubagentProjector).
    this.projector = new DshSubagentProjector({
      connection: this.connection,
      store: this.store,
      links: this.subagentLinks ?? new DshSubagentLinkRegistry(),
      parentSessionId: this.context.sessionId,
      parentDshSessionId: sessionId,
      logger: this.logger,
      schedule: (fn) => this.enqueueEvent(fn),
      emit: (event) => this.emit(event),
      pollIntervalMs: this.subagentPollIntervalMs,
    })
    this.projector.start()

    // Establish the replay boundary: everything older than the current log
    // tail is OpenCode's persisted history, never re-projected.
    await this.catchUpLastSeq()

    this.started = true
    this.emit({ type: "session_started", sessionId: this.context.sessionId })
  }

  async stop(): Promise<void> {
    this.clearWatchdog()
    this.projector?.dispose()
    this.projector = null
    const active = this.activePrompt
    if (active && !active.settled) {
      const error = active.settlement.failure ?? new Error("DSH Runtime stopped before the current prompt settled")
      active.settlement.failure = error
      active.settlement.reject(error)
    }
    this.activePrompt = null
    for (const unsubscribe of this.unsubscribes.splice(0)) {
      try {
        unsubscribe()
      } catch {
        // Unsubscribes are plain set removals.
      }
    }
    this.started = false
    this.emit({ type: "session_stopped", sessionId: this.context.sessionId })
  }

  async prompt(input: PromptInput): Promise<void> {
    if (!this.started || !this.dshSessionId) throw new Error("Runtime is not started")
    if (this.activePrompt) throw new Error("A prompt is already active on this Runtime")

    this.promptGeneration++
    const generation = this.promptGeneration
    this.mapper.beginPrompt(input.assistantMessageId)
    this.historyProbeFailures = 0

    const activePrompt: ActivePrompt = {
      generation,
      rpcId: "",
      messageId: input.messageId,
      assistantMessageId: input.assistantMessageId,
      turn: undefined,
      settlement: { resolve: () => undefined, reject: () => undefined },
      settled: false,
    }
    this.activePrompt = activePrompt
    const settled = new Promise<void>((resolve, reject) => {
      activePrompt.settlement = { resolve, reject }
    })
    void settled.catch(() => undefined)

    try {
      const response = await this.connection.client.sessions.prompt({
        sessionId: this.sid(),
        mode: "queue",
        content: [{ type: "text", text: input.text }],
      })
      if (!response.result.ok) {
        throw DshRpcResultError.fromRpcError(response.result.error)
      }
      if (activePrompt.generation !== generation) {
        throw new Error("Runtime state changed while dispatching the DSH prompt")
      }
      activePrompt.rpcId = String(response.rpcId)
      this.scheduleSettlementWatchdog()

      // A leading-/ prompt executes as a host slash command and never opens a
      // model turn. Settle the OpenCode message immediately.
      if (response.result.value.command) {
        this.logger.info("DSH prompt executed as a host command; no model turn will run", {
          sessionId: this.context.sessionId,
          commandKind: response.result.value.command.kind,
        })
        this.finishCurrentPrompt(generation, "stop")
      }

      await settled
    } catch (error) {
      if (this.activePrompt === activePrompt && !activePrompt.settled) {
        this.activePrompt = null
        const failure = error instanceof Error ? error : new Error(String(error))
        activePrompt.settlement.failure = failure
        activePrompt.settlement.reject(failure)
      }
      throw error
    } finally {
      this.clearWatchdog()
    }
  }

  async abort(): Promise<void> {
    if (!this.started || !this.dshSessionId) return
    const generation = this.activePrompt?.generation
    try {
      const response = await this.connection.client.sessions.cancel({ sessionId: this.sid() })
      if (!response.result.ok) {
        throw DshRpcResultError.fromRpcError(response.result.error)
      }
    } catch (error) {
      const active = this.activePrompt
      if (active) {
        const message = `DSH cancel failed: ${error instanceof Error ? error.message : String(error)}`
        this.failRuntime(new Error(message), active)
        return
      }
      throw error
    }
    if (generation !== undefined && this.activePrompt?.generation === generation) {
      // Settle immediately; the host's turn/end (reason aborted) arriving
      // later is deduplicated by the mapper's terminal guard.
      this.finishCurrentPrompt(generation, "aborted")
    } else {
      this.emit({ type: "session_idle", sessionId: this.context.sessionId })
    }
  }

  async compact(input?: { customInstructions?: string }): Promise<AgentCompactionResult> {
    if (!this.started || !this.dshSessionId) throw new Error("Runtime is not started")
    if (this.activePrompt) throw new Error("The DSH session is busy; compaction is only available while idle")
    if (input?.customInstructions?.trim()) {
      throw new Error("The DSH backend does not support custom compaction instructions")
    }

    // Host commands are a Typert Remote (`commands/execute`), not prompts: the
    // `/compact` command runs host-side compaction and never reaches the model.
    // It blocks until compaction settles, so it needs no bounded-call timeout.
    const endSeenBefore = this.compactionEndSeen
    const result = await this.connection.client.rpc.call(
      "commands/execute",
      { args: { agentId: this.dshSessionId, line: "/compact", images: [] } },
      { timeoutMs: null },
    )
    if (!result.ok) {
      throw new Error(`DSH /compact command failed: ${result.error.code}: ${result.error.message}`)
    }
    const execution = result.value as { result?: { kind?: string; text?: string } } | undefined
    if (execution === undefined) {
      throw new Error("The DSH host does not expose a /compact command")
    }
    if (execution.result?.kind === "error") {
      throw new Error(`DSH /compact command failed: ${execution.result.text ?? "unknown error"}`)
    }

    // The command RPC resolves after the host finished compacting; wait a
    // short grace for the durable lifecycle events to drain through the
    // shared serialized event chain.
    await this.drainEvents()
    await Bun.sleep(COMPACTION_TERMINAL_GRACE_MS)
    await this.drainEvents()

    const outcome =
      this.compactionTerminalOutcome && this.compactionEndSeen > endSeenBefore
        ? this.compactionTerminalOutcome
        : undefined
    if (outcome) {
      if (outcome.result) return outcome.result
      throw new Error(outcome.error ?? "DSH compaction failed")
    }
    // No durable lifecycle events: the host reported a no-op ("No compactable
    // history yet."). The command success text is the summary.
    return { summary: execution.result?.text ?? "" }
  }

  async fork(input: { messageId: string }): Promise<AgentForkResult> {
    if (!this.started || !this.dshSessionId) throw new Error("Runtime is not started")
    const cursor = this.requireMessageCursor(input.messageId)
    const previous = this.dshSessionId
    // Re-forking on top of a staged revert abandons the previously tracked
    // original branch; archive it so it never lingers on the host.
    const existing = this.store.getSession(this.context.sessionId)
    if (existing.revert?.previousSessionId) {
      await this.archiveAbandonedSession(existing.revert.previousSessionId, "re-fork")
    }
    const forked = await this.forkBackendSession(cursor, false)
    this.store.setSession(this.context.sessionId, {
      activeSessionId: forked,
      revert: {
        previousSessionId: previous,
        forkedSessionId: forked,
      },
    })
    this.dshSessionId = forked
    // Fork children inherit the forked prefix's plan state; a fresh-session
    // fallback has no plan/mode record, so align it with this variant.
    await this.syncPlanMode(forked)
    return { backendSessionId: forked }
  }

  async createSessionFork(input: { targetSessionId: string; messageId?: string }): Promise<AgentForkResult> {
    if (!this.started || !this.dshSessionId) throw new Error("Runtime is not started")
    const cursor = input.messageId ? this.requireMessageCursor(input.messageId) : undefined
    // No messageId means "fork the entire conversation": the DSH fork API's
    // omitted-atSeq fallback (source's last completed turn) is the exact
    // whole-session semantic. The source runtime stays bound to the source.
    const forked = await this.forkBackendSession(cursor, input.messageId === undefined)
    this.store.setSession(input.targetSessionId, { activeSessionId: forked })
    return { backendSessionId: forked }
  }

  async restoreFork(): Promise<AgentForkResult> {
    const persisted = this.store.getSession(this.context.sessionId)
    if (!persisted.revert) throw new Error("No DSH conversation fork is available to restore")
    // The forked branch is abandoned by unrevert; archive it on the host.
    await this.archiveAbandonedSession(persisted.revert.forkedSessionId, "unrevert")
    this.store.setSession(this.context.sessionId, { activeSessionId: persisted.revert.previousSessionId })
    this.dshSessionId = persisted.revert.previousSessionId
    return { backendSessionId: persisted.revert.previousSessionId }
  }

  async commitFork(): Promise<void> {
    const persisted = this.store.getSession(this.context.sessionId)
    if (!persisted.revert) return
    // Commit keeps the fork and abandons the original branch; archive it.
    await this.archiveAbandonedSession(persisted.revert.previousSessionId, "commit")
    this.store.setSession(this.context.sessionId, { activeSessionId: persisted.activeSessionId })
  }

  subscribe(listener: (event: AgentRuntimeEvent) => void): () => void {
    this.listeners.add(listener)
    return () => this.listeners.delete(listener)
  }

  private sid(): DshSessionId {
    return SessionId(this.dshSessionId!)
  }

  private sameSession(id: unknown): boolean {
    return typeof id === "string" && id === this.dshSessionId
  }

  private installSubscriptions(): void {
    this.unsubscribes.push(
      this.connection.subscribeMux((envelope) => {
        this.routeMuxEnvelope(envelope)
      }),
    )
    this.unsubscribes.push(
      this.connection.subscribeHost((envelope) => {
        const frame = envelope.payload
        if (frame.type === "host/agent-error" && this.sameSession(frame.sessionId)) {
          this.emit({
            type: "session_error",
            sessionId: this.context.sessionId,
            ...(this.activePrompt ? { messageId: this.activePrompt.assistantMessageId } : {}),
            error: { type: "dsh_agent_error", message: frame.message },
            fatal: false,
          })
        }
      }),
    )
    this.unsubscribes.push(
      this.connection.subscribeState((state) => {
        if (state === "reconnecting") {
          this.logger.warn("DSH connection lost; active prompts will catch up from history", {
            sessionId: this.context.sessionId,
          })
        } else if (state === "connected") {
          if (this.started && this.activePrompt) {
            this.enqueueEvent(() => this.catchUpHistory())
          }
        }
      }),
    )
  }

  private routeMuxEnvelope(envelope: RpcRequest<MuxFrame>): void {
    const frame = envelope.payload
    if (frame.type === "session/subscribed") {
      if (this.sameSession(frame.sessionId)) {
        this.lastSeq = Math.max(this.lastSeq, frame.lastSeq)
      }
      return
    }
    if (frame.type !== "session/event") return
    if (!this.sameSession(frame.sessionId)) return
    this.enqueueEvent(() => this.handleSessionEvent(frame.event, false))
  }

  private enqueueEvent(fn: () => Promise<void>): void {
    this.eventChain = this.eventChain.then(fn).catch((error) =>
      this.logger.error("DSH event processing failed", {
        sessionId: this.context.sessionId,
        error: error instanceof Error ? error.message : String(error),
      }),
    )
  }

  private async drainEvents(): Promise<void> {
    await this.eventChain
  }

  /** Live delivery and history replay share one serialized path. */
  private async handleSessionEvent(event: SessionEvent, replayed: boolean): Promise<void> {
    if (event.seq > this.lastSeq) this.lastSeq = event.seq
    const loose = event as unknown as { type: string; seq: number; data: Record<string, unknown> }

    switch (loose.type) {
      case "turn/start": {
        const turn = typeof loose.data.turn === "number" ? loose.data.turn : 0
        this.currentTurn = turn
        const active = this.activePrompt
        if (active && active.turn === undefined) active.turn = turn
        break
      }
      case "user/message": {
        this.correlateUserMessage(loose)
        break
      }
      case "turn/end": {
        const turn = typeof loose.data.turn === "number" ? loose.data.turn : 0
        this.lastTurnEndSeq = event.seq
        const active = this.activePrompt
        // The first turn/end after prompt acceptance is ours even when the
        // preceding turn/start was lost to a disconnect window.
        if (active && !active.settled && (active.turn === turn || active.turn === undefined)) {
          active.settled = true
        }
        break
      }
      case "compaction/end": {
        this.compactionEndSeen++
        const snapshot = this.mapper.compactionSnapshot
        const failed = typeof loose.data.error === "string" && loose.data.error.length > 0
        this.compactionTerminalOutcome = {
          sequence: this.compactionEndSeen,
          result: !failed
            ? {
                summary: snapshot?.summary ?? "",
                tokensBefore: snapshot?.shadowedTokenCount,
                usage: snapshot?.usage,
                details: snapshot?.details,
              }
            : undefined,
          error: failed ? (loose.data.error as string) : undefined,
        }
        break
      }
      default:
        break
    }

    const events = this.mapper.map(event)
    let delivered = false
    for (const mapped of events) {
      delivered = this.emit(mapped) || delivered
    }
    if (events.length > 0 && !delivered) {
      this.warnUnprojectedEvent(loose.type, "runtime_delivery", "DSH Runtime has no accepting event subscriber")
    }

    // Route task tool lifecycle to the subagent projector. This runs AFTER
    // delivery: AgentService mirrors the OpenCode child session synchronously
    // and records the call → child link before the projector discovers the
    // DSH child session.
    if (this.activePrompt && this.projector) {
      for (const mapped of events) {
        if (mapped.type === "tool_call_started" && mapped.tool === "task") {
          this.projector.onToolCall(
            mapped.callId,
            mapped.input,
            mapped.messageId,
            mapped.partId,
            this.mapper.rawToolName(mapped.callId),
          )
        } else if (mapped.type === "tool_call_running" && mapped.tool === "task") {
          // The finalized argument snapshot supersedes the partial one.
          this.projector.onToolCall(
            mapped.callId,
            mapped.input,
            mapped.messageId,
            mapped.partId,
            this.mapper.rawToolName(mapped.callId),
          )
        } else if (
          (mapped.type === "tool_call_completed" || mapped.type === "tool_call_error") &&
          mapped.tool === "task"
        ) {
          this.projector.onToolResult(mapped.callId)
        }
      }
    }

    // Settlement only after the mapped terminal events reached the projector.
    const active = this.activePrompt
    if (
      active &&
      active.settled &&
      active.generation === this.promptGeneration &&
      events.some((mapped) => mapped.type === "session_idle")
    ) {
      this.settlePrompt(active)
    }
    void replayed
  }

  private settlePrompt(active: ActivePrompt): void {
    this.clearWatchdog()
    this.activePrompt = null
    active.settlement.resolve()
  }

  private correlateUserMessage(loose: { data: Record<string, unknown> }): void {
    const source = loose.data.source as Record<string, unknown> | undefined
    const active = this.activePrompt
    if (!active || !source) return
    if (source.kind !== "user" || typeof source.rpcId !== "string") return
    if (source.rpcId !== active.rpcId) return
    active.turn = this.currentTurn
    const cursor: DshMessageCursor = {
      seq: this.lastSeq,
      messageId: typeof loose.data.id === "string" ? loose.data.id : undefined,
    }
    // Fork anchor: the previous turn's end (so the fork boundary is the first
    // turn/end at-or-after it, i.e. everything before this message's turn).
    if (this.lastTurnEndSeq !== undefined) {
      cursor.prevTurnEndSeq = this.lastTurnEndSeq
    }
    try {
      this.store.setMessage(active.messageId, cursor)
    } catch (error) {
      // Subtask backends may reference messages outside the store; cursors are
      // only ever read for conversation forks on ordinary sessions.
      this.logger.debug("Could not persist the DSH message cursor", {
        sessionId: this.context.sessionId,
        error: error instanceof Error ? error.message : String(error),
      })
    }
  }

  private requireMessageCursor(messageId: string): DshMessageCursor {
    const cursor = this.store.getMessageCursor(messageId)
    if (cursor.seq === undefined) {
      throw new Error(`DSH message cursor is missing for OpenCode message '${messageId}'`)
    }
    return cursor
  }

  /**
   * Create the backend side of a conversation fork.
   *
   * - With a turn anchor (`cursor.prevTurnEndSeq`): `session.fork({ atSeq })`,
   *   cutting at the turn/end immediately before the boundary message's turn.
   * - Whole-session fork (`wholeSession`): omit `atSeq` so the host falls back
   *   to the source's last completed turn — the official full-conversation
   *   fork. A source with no completed turn has nothing to fork, so an empty
   *   backend session is the honest equivalent.
   * - Fork before the first message: no prior turn/end anchor exists, so an
   *   empty backend session is the honest equivalent.
   */
  private async forkBackendSession(cursor: DshMessageCursor | undefined, wholeSession: boolean): Promise<string> {
    if (!this.dshSessionId) throw new Error("Runtime is not started")
    if (cursor?.prevTurnEndSeq !== undefined) {
      const response = await this.connection.client.sessions.fork({
        sessionId: this.sid(),
        atSeq: cursor.prevTurnEndSeq,
      })
      if (!response.result.ok) {
        throw DshRpcResultError.fromRpcError(response.result.error)
      }
      return String(response.result.value.sessionId)
    }
    if (wholeSession) {
      // Probe the source: a session without any turn has nothing to fork, and
      // the host's error for that edge is not part of the stable contract.
      const probe = await this.connection.client.sessions.history({
        sessionId: this.sid(),
        maxMessages: 1,
      })
      if (probe.result.ok && probe.result.value.events.length === 0) {
        return this.createFreshBackendSession()
      }
      const response = await this.connection.client.sessions.fork({ sessionId: this.sid() })
      if (response.result.ok) return String(response.result.value.sessionId)
      const error = DshRpcResultError.fromRpcError(response.result.error)
      // The host rejects a source whose only in-log turn is still open (or a
      // source it cannot fork from); an empty backend session is the honest
      // equivalent for that case.
      if (error.code === "fork-unavailable") {
        this.logger.info("DSH whole-session fork fell back to a fresh session", {
          sessionId: this.context.sessionId,
          dshSessionId: this.dshSessionId,
          reason: error.message,
        })
        return this.createFreshBackendSession()
      }
      throw error
    }
    return this.createFreshBackendSession()
  }

  private async createFreshBackendSession(): Promise<string> {
    const fresh = await this.createBackendSession()
    // Select on the fresh session explicitly: `this.dshSessionId` still
    // points at the source until the fork caller swaps it in.
    await this.applyModelSelection(fresh)
    return fresh
  }

  /**
   * Archive a DSH conversation branch that no OpenCode state references
   * anymore (re-fork, unrevert, commit). The host has no session.delete;
   * archiving is its native removal gesture and is idempotent. Best-effort:
   * a host outage must never break the conversation operation itself.
   */
  private async archiveAbandonedSession(dshSessionId: string, reason: string): Promise<void> {
    if (!dshSessionId) return
    try {
      await this.connection.start()
      const response = await this.connection.client.workspace.archiveSession({
        sessionId: SessionId(dshSessionId),
      })
      if (response.result.ok) {
        this.logger.info("Archived an abandoned DSH conversation branch", {
          sessionId: this.context.sessionId,
          dshSessionId,
          reason,
        })
      } else {
        const error = response.result.error
        if (error.code !== "session-not-found") {
          this.logger.warn("Could not archive an abandoned DSH conversation branch", {
            sessionId: this.context.sessionId,
            dshSessionId,
            reason,
            code: error.code,
            error: error.message,
          })
        }
      }
    } catch (error) {
      this.logger.warn("Could not archive an abandoned DSH conversation branch", {
        sessionId: this.context.sessionId,
        dshSessionId,
        reason,
        error: error instanceof Error ? error.message : String(error),
      })
    }
  }

  private async createBackendSession(): Promise<string> {
    const response = await this.connection.client.sessions.create({
      cwd: this.context.directory,
    })
    if (!response.result.ok) {
      throw DshRpcResultError.fromRpcError(response.result.error)
    }
    // Plan state is aligned by syncPlanMode right after the runtime binds the
    // session (fresh sessions carry no plan/mode record yet).
    return String(response.result.value.sessionId)
  }

  /**
   * Align the host session's plan mode with this runtime's desired state.
   * The current state reads the host's `plan` projection (the history tail
   * page baseline); when it differs, one directed `/plan` host command sets
   * it. Plan state lives in the DSH session log, so forked children inherit
   * the forked prefix; only state that disagrees with the variant is set.
   */
  private async syncPlanMode(sessionId: string): Promise<void> {
    // Absent plan/mode records mean plan mode is off (the host fold treats an
    // empty prefix as inactive), so an unreadable state falls back to off.
    const current = (await this.readPlanState(sessionId)) ?? false
    if (current === this.planMode) return
    await this.setPlanMode(sessionId, this.planMode)
  }

  private async readPlanState(sessionId: string): Promise<boolean | undefined> {
    try {
      const response = await this.connection.client.sessions.history({
        sessionId: SessionId(sessionId),
        maxMessages: 1,
      })
      if (!response.result.ok) return undefined
      const projections = response.result.value.projections?.values as { plan?: { active?: boolean } } | undefined
      if (typeof projections?.plan?.active === "boolean") return projections.plan.active
      // Fallback: fold plan/mode from the returned events (hosts without a
      // projection registry).
      let active: boolean | undefined
      for (const entry of response.result.value.events) {
        const event = entry.event as unknown as { type: string; data?: { active?: boolean } }
        if (event.type === "plan/mode") active = event.data?.active === true
      }
      return active
    } catch (error) {
      this.logger.warn("Could not read the DSH plan state", {
        sessionId: this.context.sessionId,
        dshSessionId: sessionId,
        error: error instanceof Error ? error.message : String(error),
      })
      return undefined
    }
  }

  /**
   * Set the host's plan mode with the directed `/plan` command (`/plan off`
   * to leave). Best-effort: a host without the commands Remote simply keeps
   * its current plan state.
   */
  private async setPlanMode(sessionId: string, active: boolean): Promise<void> {
    try {
      const result = await this.connection.client.rpc.call("commands/execute", {
        args: { agentId: sessionId, line: active ? "/plan" : "/plan off", images: [] },
      })
      if (!result.ok) {
        this.logger.warn("Could not set DSH plan mode", {
          sessionId: this.context.sessionId,
          dshSessionId: sessionId,
          desired: active,
          code: result.error.code,
          error: result.error.message,
        })
        return
      }
      const execution = result.value as { result?: { kind?: string; text?: string } } | undefined
      if (execution === undefined) {
        this.logger.warn("The DSH host does not expose a /plan command; the plan state is unchanged", {
          sessionId: this.context.sessionId,
          dshSessionId: sessionId,
        })
        return
      }
      this.logger.info(`DSH plan mode ${active ? "enabled" : "disabled"}`, {
        sessionId: this.context.sessionId,
        dshSessionId: sessionId,
        text: execution.result?.text,
      })
    } catch (error) {
      this.logger.warn("Could not set DSH plan mode", {
        sessionId: this.context.sessionId,
        dshSessionId: sessionId,
        desired: active,
        error: error instanceof Error ? error.message : String(error),
      })
    }
  }

  private async verifySession(sessionId: string): Promise<boolean> {
    const response = await this.connection.client.sessions.history({
      sessionId: SessionId(sessionId),
      maxMessages: 1,
    })
    if (response.result.ok) return true
    // Only an explicit session-not-found means the mapping is stale; other
    // business errors still identify an existing session.
    return response.result.error.code !== "session-not-found"
  }

  /**
   * Apply this runtime's provider/model/reasoning-effort selection. With an
   * explicit provider/model the complete selection is sent directly. When
   * only a reasoning effort is configured, read the DSH session's current
   * provider/model selection and re-submit it with the requested effort so
   * `DSH_REASONING_EFFORT` works even without `DSH_PROVIDER`/`DSH_MODEL`.
   */
  private async applyModelSelection(sessionId?: string): Promise<void> {
    const target = sessionId ?? this.dshSessionId
    if (!target) return
    if (this.provider && this.model) {
      await this.selectModel(target, this.provider, this.model, this.reasoningEffort)
      return
    }
    if (!this.reasoningEffort) return
    const response = await this.connection.client.sessions.models({ sessionId: SessionId(target) })
    if (!response.result.ok) {
      throw DshRpcResultError.fromRpcError(response.result.error)
    }
    const current = response.result.value.current
    await this.selectModel(target, current.provider, current.model, this.reasoningEffort)
  }

  private async selectModel(
    sessionId: string,
    provider: string,
    model: string,
    reasoningEffort?: string,
  ): Promise<void> {
    const response = await this.connection.client.sessions.selectModel({
      sessionId: SessionId(sessionId),
      provider,
      model,
      ...(reasoningEffort ? { reasoningEffort } : {}),
    })
    if (!response.result.ok) {
      throw DshRpcResultError.fromRpcError(response.result.error)
    }
  }

  /** Fetch the log tail and advance lastSeq without re-projecting history. */
  private async catchUpLastSeq(): Promise<void> {
    if (!this.dshSessionId) return
    try {
      const response = await this.connection.client.sessions.history({
        sessionId: this.sid(),
        maxMessages: 10,
      })
      if (!response.result.ok) return
      for (const entry of response.result.value.events) {
        if (entry.event.seq > this.lastSeq) this.lastSeq = entry.event.seq
      }
    } catch (error) {
      this.logger.warn("Could not establish the DSH replay boundary", {
        sessionId: this.context.sessionId,
        error: error instanceof Error ? error.message : String(error),
      })
    }
  }

  /** Replay missed events (seq > lastSeq) from history in serialized order. */
  private async catchUpHistory(): Promise<void> {
    if (!this.dshSessionId) return
    let entries: HistoryEntry[] = []
    try {
      const response = await this.connection.client.sessions.history({
        sessionId: this.sid(),
        maxMessages: 30,
      })
      if (!response.result.ok) {
        this.logger.warn("DSH history catch-up failed", {
          sessionId: this.context.sessionId,
          code: response.result.error.code,
          error: response.result.error.message,
        })
        return
      }
      entries = response.result.value.events
    } catch (error) {
      this.logger.warn("DSH history catch-up failed", {
        sessionId: this.context.sessionId,
        error: error instanceof Error ? error.message : String(error),
      })
      return
    }
    let replayedCount = 0
    for (const entry of entries) {
      if (entry.event.seq <= this.lastSeq) continue
      replayedCount++
      await this.handleSessionEvent(entry.event, true)
    }
    if (replayedCount > 0) {
      this.logger.info("Replayed missed DSH events from history", {
        sessionId: this.context.sessionId,
        replayedCount,
      })
    }
  }

  private finishCurrentPrompt(generation: number, finish: string): void {
    const active = this.activePrompt
    if (!active || active.generation !== generation) return
    const events = this.mapper.finishPromptTurn(finish)
    let delivered = false
    for (const event of events) {
      delivered = this.emit(event) || delivered
    }
    if (events.length > 0 && !delivered) {
      this.warnUnprojectedEvent(
        "message_completed",
        "runtime_delivery",
        "DSH Runtime has no accepting event subscriber",
      )
    }
    this.settlePrompt(active)
  }

  private failRuntime(error: Error, active: ActivePrompt): void {
    this.clearWatchdog()
    active.settlement.failure = error
    this.emit({
      type: "runtime_fault",
      sessionId: this.context.sessionId,
      messageId: active.assistantMessageId,
      error: { type: "dsh_transport", message: error.message },
    })
    active.settlement.reject(error)
    this.activePrompt = null
  }

  private scheduleSettlementWatchdog(): void {
    this.clearWatchdog()
    const generation = this.promptGeneration
    this.watchdogTimer = setTimeout(() => {
      this.watchdogTimer = null
      this.enqueueEvent(() => this.runSettlementWatchdog(generation))
    }, this.watchdogIntervalMs)
  }

  private clearWatchdog(): void {
    if (!this.watchdogTimer) return
    clearTimeout(this.watchdogTimer)
    this.watchdogTimer = null
  }

  private async runSettlementWatchdog(generation: number): Promise<void> {
    const active = this.activePrompt
    if (!active || active.generation !== generation || active.settled) return
    try {
      await this.catchUpHistory()
      this.historyProbeFailures = 0
    } catch (error) {
      this.historyProbeFailures++
      this.logger.warn("DSH history probe failed while waiting for prompt settlement", {
        sessionId: this.context.sessionId,
        attempt: this.historyProbeFailures,
        error: error instanceof Error ? error.message : String(error),
      })
      if (this.historyProbeFailures >= HISTORY_PROBE_MAX_FAILURES) {
        const activeNow = this.activePrompt
        if (activeNow && activeNow.generation === generation && !activeNow.settled) {
          this.failRuntime(
            new Error(
              `DSH history probe failed ${this.historyProbeFailures} times: ${
                error instanceof Error ? error.message : String(error)
              }`,
            ),
            activeNow,
          )
        }
        return
      }
    }
    if (this.activePrompt?.generation === generation && !this.activePrompt.settled) {
      this.scheduleSettlementWatchdog()
    }
  }

  private emit(event: AgentRuntimeEvent): boolean {
    let delivered = 0
    for (const listener of this.listeners) {
      try {
        listener(event)
        delivered++
      } catch {
        // Listener isolation mirrors the pool's own containment.
      }
    }
    return delivered > 0
  }

  private warnUnprojectedEvent(type: string, stage: string, reason: string): void {
    this.logger.warn("DSH event produced no OpenCode event", {
      sessionId: this.context.sessionId,
      dshEvent: type,
      stage,
      reason,
    })
  }
}