opencode 源码解析

graph LR
  CLI["cli（lildax）<br/>新 CLI：启动 server 守护进程 + TUI"] --> SRV
  SRV["server<br/>HTTP API：路由 + 中间件 + 处理器"] --> CORE
  CORE["core<br/>v2 内核：会话、执行、工具、权限、Effect 服务图"] --> LLM
  CORE --> SCHEMA
  LLM["llm<br/>统一请求/事件模型 + 各家协议"] --> SCHEMA
  SCHEMA["schema<br/>纯类型：消息、事件、权限、agent"]
  OLD["opencode（老 CLI，v1）<br/>自己的 server + SessionPrompt 单体"] -.依赖但不走 v2 runner.-> CORE
  OLD -.-> SRV

graph LR
  P["用户输入<br/>Prompt"] -->|"PromptAdmitted"| EV["会话事件<br/>event 表"]
  EV -->|"投影"| IN["待处理输入<br/>session_input 表"]
  IN -->|"runner 提升，发 Prompted"| EV
  EV -->|"投影"| MSG["会话消息<br/>session_message 表"]
  MSG -->|"toLLMMessages"| REQ["LLMRequest"]
  REQ -->|"LLMClient.stream"| LE["LLMEvent"]
  LE -->|"发布器转换"| EV
  EV -->|"SSE 推送"| APP["客户端 / TUI"]

classDiagram
  class Message {
    <<union>>
    id: ID
    type: Type
    metadata: Record
    time.created: DateTime
  }
  class User {
    text: string
    files: FileAttachment[]
    agents: AgentAttachment[]
  }
  class Assistant {
    agent: string
    model: Model.Ref
    content: AssistantContent[]
    snapshot.start: string
    snapshot.end: string
    snapshot.files: RelativePath[]
    finish: string
    cost: number
    tokens: Tokens
    error: UnknownError
    time.completed: DateTime
  }
  class Compaction {
    reason: auto or manual
    summary: string
    recent: string
  }
  class System {
    text: string
  }
  class Synthetic {
    sessionID: SessionID
    text: string
  }
  class Shell {
    callID: string
    command: string
    output: string
    time.completed: DateTime
  }
  class AgentSwitched {
    agent: string
  }
  class ModelSwitched {
    model: Model.Ref
  }
  class AssistantText {
    type: text
    id: string
    text: string
  }
  class AssistantReasoning {
    type: reasoning
    id: string
    text: string
    providerMetadata: ProviderMetadata
  }
  class AssistantTool {
    type: tool
    id: string
    name: string
    provider.executed: boolean
    provider.metadata: ProviderMetadata
    provider.resultMetadata: ProviderMetadata
    state: ToolState
    time.created ran completed pruned
  }
  class ToolState {
    <<union>>
    status: pending or running or completed or error
  }
  Message <|-- User
  Message <|-- Assistant
  Message <|-- Compaction
  Message <|-- System
  Message <|-- Synthetic
  Message <|-- Shell
  Message <|-- AgentSwitched
  Message <|-- ModelSwitched
  Assistant *-- AssistantText
  Assistant *-- AssistantReasoning
  Assistant *-- AssistantTool
  AssistantTool *-- ToolState

classDiagram
  class LLMRequest {
    id: string
    model: Model
    system: SystemPart[]
    messages: Message[]
    tools: ToolDefinition[]
    toolChoice: ToolChoice
    generation: GenerationOptions
    providerOptions: ProviderOptions
    http: HttpOptions
    responseFormat: ResponseFormat
    cache: CachePolicy
    metadata: Record
  }
  class Message {
    id: string
    role: system or user or assistant or tool
    content: ContentPart[]
    metadata: Record
    native: Record
  }
  class ContentPart {
    <<union>>
    TextPart
    MediaPart
    ToolCallPart
    ToolResultPart
    ReasoningPart
  }
  class ToolDefinition {
    name: string
    description: string
    inputSchema: JsonSchema
    outputSchema: JsonSchema
    cache: CacheHint
  }
  class ToolChoice {
    type: auto or none or required or tool
    name: string
  }
  class LLMEvent {
    <<union of 16>>
    step-start
    text-start delta end
    reasoning-start delta end
    tool-input-start delta end
    tool-call
    tool-result
    tool-error
    step-finish
    finish
    provider-error
  }
  class Model {
    id: string
    provider: ProviderID
    route: Route
  }
  class LLMClient {
    stream(request) Stream~LLMEvent, LLMError~
    generate(request) Effect~LLMResponse, LLMError~
  }
  LLMRequest *-- Message
  LLMRequest *-- ToolDefinition
  LLMRequest *-- ToolChoice
  LLMRequest *-- Model
  Message *-- ContentPart
  LLMClient ..> LLMRequest
  LLMClient ..> LLMEvent

classDiagram
  class Prompt {
    text: string
    files: FileAttachment[]
    agents: AgentAttachment[]
  }
  class Admitted {
    admittedSeq: int
    id: Message.ID
    sessionID: SessionID
    prompt: Prompt
    delivery: steer or queue
    timeCreated: DateTime
    promotedSeq: int
  }
  class SessionInfo {
    id: SessionID
    parentID: SessionID
    projectID: Project.ID
    agent: Agent.ID
    model: Model.Ref
    cost: number
    tokens: Tokens
    time: created updated archived
    title: string
    location: Location.Ref
    subpath: RelativePath
    revert: Revert.State
  }
  class AgentInfo {
    id: Agent.ID
    model: Model.Ref
    request: headers body
    system: string
    description: string
    mode: subagent or primary or all
    hidden: boolean
    steps: int
    permissions: Ruleset
  }
  class Rule {
    action: string
    resource: string
    effect: allow or deny or ask
  }
  Admitted *-- Prompt
  AgentInfo *-- Rule : permissions

classDiagram
  class session {
    id PK
    project_id FK
    workspace_id
    parent_id
    directory
    path
    title
    agent
    model json
    cost tokens_x5
    revert json
    permission json
    time_created updated compacting archived
  }
  class session_message {
    id PK
    session_id FK
    type
    seq  UNIQUE with session_id
    time_created
    data json
  }
  class session_input {
    id PK
    session_id FK
    prompt json
    delivery
    admitted_seq UNIQUE with session_id
    promoted_seq UNIQUE with session_id
    time_created
  }
  class session_context_epoch {
    session_id PK
    baseline text
    snapshot json
    baseline_seq
  }
  class event {
    id PK
    aggregate_id FK
    seq UNIQUE with aggregate_id
    type
    data json
  }
  class event_sequence {
    aggregate_id PK
    seq
    owner_id
  }
  session "1" --> "*" session_message
  session "1" --> "*" session_input
  session "1" --> "0..1" session_context_epoch
  event_sequence "1" --> "*" event

classDiagram
  class SessionV2Service {
    <<Service>>
    create(input) Info
    get(sessionID) Info
    list(input) Info[]
    messages(input) Message[]
    message(input) Message
    context(sessionID) Message[]
    events(input) Stream~DurableEvent~
    history(input) events hasMore
    prompt(input) Admitted
    switchAgent(input)
    switchModel(input)
    compact(input) OperationUnavailable
    wait(id) OperationUnavailable
    shell(input) OperationUnavailable
    skill(input) OperationUnavailable
    active Set~SessionID~
    resume(sessionID)
    interrupt(sessionID)
    revert.stage clear commit
  }
  class SessionStore {
    <<Service global>>
    get(sessionID) Info
    context(sessionID) Message[]
    runnerContext(sessionID, baselineSeq) Message[]
    message(messageID) sessionID message
  }
  class SessionInput {
    <<functions>>
    admit(db, events, input) Admitted
    find(db, id) Admitted
    hasPending(db, sessionID, delivery) boolean
    promoteSteers(db, events, sessionID, cutoff) int
    promoteNextQueued(db, events, sessionID) boolean
    projectAdmitted(db, input)
    projectPrompted(db, input)
    equivalent(admitted, expected) boolean
  }
  class SessionHistory {
    <<functions>>
    latestCompaction(db, sessionID) seq
    load(db, sessionID) Message[]
    entriesForRunner(db, sessionID, baselineSeq) Entry[]
  }
  class SessionProjector {
    <<Layer global>>
    run(db, event)
    insertMessage(db, event, message)
  }
  class SessionMessageUpdater {
    <<functions>>
    update(adapter, event)
    memory(state) Adapter
  }
  class SessionRevert {
    <<functions>>
    stage(input) Revert.State
    clear(session)
    commit(session)
  }
  class SessionExecution {
    <<Service global unbound>>
    active Set~SessionID~
    resume(sessionID)
    wake(sessionID)
    interrupt(sessionID)
  }
  SessionV2Service --> SessionStore : get context
  SessionV2Service --> SessionInput : admit
  SessionV2Service --> SessionExecution : wake resume interrupt
  SessionV2Service --> SessionRevert
  SessionV2Service ..> EventV2 : publish switchAgent switchModel
  SessionStore --> SessionHistory
  SessionProjector --> SessionMessageUpdater : 用 db 适配器
  SessionProjector --> SessionInput : projectAdmitted projectPrompted
  SessionProjector ..> EventV2 : events.project 注册

classDiagram
  class SessionExecutionLocal {
    <<Layer global>>
    store: SessionStore
    locations: LocationServiceMap
    coordinator: Coordinator
    drain(sessionID, force)
  }
  class SessionRunCoordinator {
    active: Map~Key, Entry~
    fork: FiberSet runtime
    run(key)
    wake(key)
    interrupt(key)
    active Set~Key~
    start(key, entry, force, successor)
    settle(key, entry, exit)
  }
  class Entry {
    done: Deferred
    owner: Fiber
    pendingWake: boolean
    stopping: boolean
  }
  class SessionRunner {
    <<Service location>>
    run(sessionID, force)
  }
  class SessionRunnerImpl {
    events llm agents tools models store location
    systemContext skillGuidance referenceGuidance config snapshots db
    compaction: SessionCompaction
    getSession(sessionID)
    failInterruptedTools(sessionID)
    loadSystemContext(agent)
    runTurnAttempt(sessionID, promotion, step, recoverOverflow)
    runTurn(sessionID, promotion, step)
    runAfterOverflowCompaction(sessionID, promotion, step)
    run(input)
  }
  class TurnTransitionError {
    transition: ContinueAfterCompaction or ContinueAfterOverflowCompaction
    step: int
  }
  class LLMEventPublisher {
    tools: Map~callID, ToolTrack~
    assistantMessageID: Message.ID
    assistantActive assistantFailed providerFailed: boolean
    stepSettlement: finish tokens
    publish(event, outputPaths)
    flush()
    startAssistant() Message.ID
    failAssistant(message)
    failUnsettledTools(message, hostedOnly)
    hasActiveAssistant() hasAssistantStarted() hasProviderError()
    stepSettlement()
    assistantMessageID(callID)
  }
  class SessionRunnerModel {
    <<Service location>>
    resolve(session) Model
  }
  class SessionContextEpoch {
    <<functions>>
    initialize(db, context, sessionID) Prepared
    prepare(db, events, context, sessionID) Prepared
    reset(db, sessionID)
  }
  class SessionCompaction {
    compactIfNeeded(input) boolean
    compactAfterOverflow(input) boolean
  }
  SessionExecutionLocal *-- SessionRunCoordinator
  SessionRunCoordinator *-- Entry
  SessionExecutionLocal ..> SessionRunner : drain 时 provide 目录服务后调 run
  SessionRunner <|.. SessionRunnerImpl
  SessionRunnerImpl ..> TurnTransitionError : die 抛出再 catchDefect
  SessionRunnerImpl ..> LLMEventPublisher : 每轮 new 一个
  SessionRunnerImpl --> SessionRunnerModel
  SessionRunnerImpl ..> SessionContextEpoch
  SessionRunnerImpl *-- SessionCompaction
  SessionRunnerImpl ..> SessionInput : promote hasPending
  SessionRunnerImpl ..> ToolRegistry : materialize settle

classDiagram
  class ToolDefinition_opaque {
    <<Tool.make 返回值 冻结空对象>>
  }
  class ToolRuntime {
    permission: string
    definition(name) ToolDefinition
    settle(call, context) Effect~ToolOutput, ToolFailure~
  }
  class ToolConfig {
    description: string
    input: Schema
    output: Schema
    structured: Schema
    toStructuredOutput(input, output)
    execute(input, context) Effect~Output, ToolFailure~
    toModelOutput(input, output) Content[]
  }
  class ToolContext {
    sessionID: SessionID
    agent: Agent.ID
    assistantMessageID: Message.ID
    toolCallID: string
  }
  class ToolRegistry {
    <<Service location>>
    local: Map~name, registration[]~
    applications: ApplicationTools
    resources: ToolOutputStore
    materialize(permissions) Materialization
    register(tools)
    settleWith(input, advertised) Settlement
  }
  class Materialization {
    definitions: ToolDefinition[]
    settle(input) Effect~Settlement~
  }
  class Settlement {
    result: ToolResultValue
    output: ToolOutput
    outputPaths: string[]
  }
  class ApplicationTools {
    <<Service global>>
    entries() Map~name, Entry~
    register(tools)
  }
  class ToolOutputStore {
    <<Service location>>
    MAX_LINES 2000
    MAX_BYTES 50KB
    RETENTION 7d
    bound(input) BoundResult
    cleanup()
  }
  class PermissionV2 {
    <<Service location>>
    pending: Map~ID, Pending~
    ask(input) AskResult
    assert(input)
    reply(input)
    get(id) Request
    forSession(sessionID) Request[]
    list() Request[]
    evaluate(action, resource, rulesets) Rule
  }
  class QuestionV2 {
    <<Service location>>
    pending: Map~ID, Pending~
    ask(input) Answer[]
    reply(input)
    reject(requestID)
  }
  class LocationMutation {
    <<Service location>>
    resolve(input) Target
  }
  class Target {
    canonical: string
    resource: string
    externalDirectory: action directory resource save
  }
  class BashTool
  class EditTool
  class ReadTool
  class WriteTool
  class QuestionTool
  ToolDefinition_opaque ..> ToolRuntime : WeakMap 关联
  ToolRuntime ..> ToolConfig : 闭包持有
  ToolRuntime ..> ToolContext
  ToolRegistry --> ApplicationTools : 兜底查
  ToolRegistry --> ToolOutputStore : bound
  ToolRegistry ..> Materialization : 返回
  Materialization ..> Settlement
  BashTool ..> PermissionV2 : assert
  BashTool ..> LocationMutation : resolve
  EditTool ..> PermissionV2
  EditTool ..> LocationMutation
  ReadTool ..> PermissionV2
  WriteTool ..> PermissionV2
  QuestionTool ..> QuestionV2 : ask
  QuestionTool ..> PermissionV2
  LocationMutation ..> Target

classDiagram
  class Node {
    kind: layer or unbound or group
    name: string
    service: Context.Service
    implementation: Layer
    dependencies: Node[]
    tag: Tag
  }
  class LayerNode {
    <<functions>>
    tags(config) Tags
    make(input) Node
    unbound(service, tag) Node
    group(nodes) Node
    hoist(root, tag, replacements) node hoisted
    compile(root, replacements) Layer
    hasUnbound(root, source) boolean
  }
  class AppNode {
    tags location depends global
    makeGlobalNode
    makeLocationNode
  }
  class AppNodeBuilder {
    build(root, replacements) Layer
  }
  class LocationServiceMap {
    <<Service global unbound>>
    get(ref) Layer
  }
  LayerNode ..> Node
  AppNode ..> LayerNode : tags
  AppNodeBuilder ..> LayerNode : compile
  AppNodeBuilder ..> LocationServiceMap : 需要时自动构造

sequenceDiagram
  participant CLI as lildax cli
  participant D as Daemon
  participant S as server routes.ts
  participant B as AppNodeBuilder
  participant LN as LayerNode
  participant LM as LocationServiceMap
  participant HTTP as HttpApiBuilder

  CLI->>D: 默认命令 daemon.transport()
  D->>D: 读 server.json 健康检查
  alt 没有健康的 server
    D->>CLI: spawn(execPath, [serve, --register])
    CLI->>S: serve 处理器 createRoutes(password)
  end
  S->>B: build(applicationServices, [[SessionExecution.node, SessionExecutionLocal.node]])
  B->>LN: hasUnbound(root, LocationServiceMap.node)
  LN-->>B: true
  B->>LM: buildLocationServiceMap(replacements)
  B->>LN: compile(root, replacements + LocationServiceMap 替换)
  LN-->>S: serviceLayer（Database, EventV2, SessionV2, SessionExecutionLocal ...）
  S->>HTTP: HttpApiBuilder.layer(Api) provide handlers, 中间件, auth, serviceLayer
  HTTP-->>CLI: 监听端口（4096 起找空闲）
  Note over LM: 目录服务此时还没建
  CLI->>HTTP: 第一个带 sessionID 的请求
  HTTP->>LM: sessionLocationLayer 查 session 表拿目录 → locations.get(ref)
  LM->>LN: hoist(locationServices, global, [Location 绑定 ref])
  LM->>LN: compile(目录级) fresh，provide compile(进程级)
  Note over LM: 34 个目录服务实例化，含 BuiltInTools 注册 12 个工具，SessionRunnerLLM
  LM-->>HTTP: 目录服务 Layer（缓存 60 分钟）

sequenceDiagram
  participant H as session.prompt 处理器
  participant SV as SessionV2.Service
  participant SI as SessionInput
  participant EV as EventV2
  participant PJ as SessionProjector
  participant EX as SessionExecutionLocal
  participant CO as SessionRunCoordinator
  participant RN as SessionRunner.run
  participant RT as runTurnAttempt
  participant CE as SessionContextEpoch
  participant TR as ToolRegistry
  participant LLM as LLMClient.stream
  participant PB as LLMEventPublisher

  H->>SV: prompt({ sessionID, id, prompt, delivery, resume })
  SV->>SV: get(sessionID) 存在？resolvePrompt，id 默认 msg_ 新建，delivery 默认 steer
  SV->>SI: admit(db, events, { id, sessionID, prompt, delivery })
  SI->>SI: find(id) 已有就返回
  SI->>EV: publish(PromptAdmitted)
  EV->>PJ: 事务内 projectAdmitted → INSERT session_input
  EV-->>SI: event.durable.seq → admittedSeq
  SI-->>SV: Admitted
  SV->>SV: equivalent(admitted, expected)？否则 PromptConflictError
  SV->>EX: wake(sessionID)（resume 不为 false 时）
  EX->>CO: wake(key)
  alt 该会话正在跑
    CO->>CO: entry.pendingWake = true
  else 空闲
    CO->>CO: 新 entry，fork drain(key, force=false)
  end
  SV-->>H: 返回 Admitted（HTTP 200，循环可能还没开始）
  CO->>EX: drain(sessionID, false)
  EX->>RN: provide(locations.get(session.location)) 后 run({ sessionID, force: false })
  RN->>SI: hasPending(steer)？hasPending(queue)？
  RN->>RN: 都没有且 force 为假 → return
  RN->>RN: failInterruptedTools(sessionID)
  loop while shouldRun（queue 外层）
    loop while needsContinuation（steer 与工具续接 内层）
      RN->>RT: runTurn → runTurnAttempt(sessionID, promotion, step, compactAfterOverflow)
      RT->>RT: 目录围栏：session.location 不等于本 runner 的 location → interrupt
      RT->>RT: agents.select(session.agent)
      RT->>CE: initialize（首次生成基线）
      RT->>SI: promoteSteers(cutoff) 或 promoteNextQueued + promoteSteers
      SI->>EV: publish(Prompted) 每条一次
      EV->>PJ: 事务内 projectPrompted（标 promoted_seq）+ 插 user 消息
      RT->>RT: promoted 大于 0 → currentStep = 1
      RT->>CE: prepare（比对系统上下文，变了发 ContextUpdated）
      RT->>RT: models.resolve(session)
      RT->>RT: SessionHistory.entriesForRunner(baselineSeq)
      RT->>TR: materialize(agent.permissions)（isLastStep 时跳过）
      RT->>RT: LLM.request({ model, http headers, providerOptions, system, messages, tools, toolChoice })
      RT->>RT: compaction.compactIfNeeded？→ die(ContinueAfterCompaction)
      RT->>RT: snapshots.capture() → startSnapshot
      RT->>PB: createLLMEventPublisher(events, { sessionID, agent, model, snapshot })
      RT->>LLM: stream(request)
      loop 每个 LLMEvent
        LLM-->>RT: event
        RT->>PB: publish(event)（Semaphore 串行）
        PB->>EV: publish(Step.Started / Text.* / Reasoning.* / Tool.Input.* / Tool.Called ...)
        RT->>RT: tool-call 且非 provider 执行 → needsContinuation = true，settle 进 FiberSet（§9）
      end
      RT->>PB: ensuring flush()
      RT->>RT: 溢出且没开始输出 → recoverOverflow → die(ContinueAfterOverflowCompaction)
      RT->>RT: LLMError → failUnsettledTools + failAssistant
      RT->>RT: awaitToolFibers（等全部工具结束）
      RT->>RT: 用户拒绝 → failUnsettledTools → interrupt
      RT->>EV: publish(Step.Ended { finish, tokens, snapshot: endSnapshot, files })
      RT-->>RN: { needsContinuation, step }
      RN->>RN: promotion = steer；不需续接时再查 hasPending(steer)
    end
    RN->>SI: hasPending(queue)？→ shouldRun，promotion = queue
  end
  RN-->>CO: 结束
  CO->>CO: settle：pendingWake 为真就起后继

sequenceDiagram
  participant RT as runTurnAttempt
  participant PB as LLMEventPublisher
  participant EV as EventV2
  participant FS as FiberSet toolFibers
  participant MT as Materialization.settle
  participant TR as ToolRegistry.settleWith
  participant TL as Tool.settle
  participant EX as edit.execute
  participant LM as LocationMutation
  participant PM as PermissionV2
  participant UI as 客户端
  participant OS as ToolOutputStore

  RT->>PB: publish(tool-call event)
  PB->>EV: publish(Tool.Called { callID, tool, input, provider })
  RT->>RT: needsContinuation = true
  RT->>PB: assistantMessageID(callID)
  RT->>FS: FiberSet.run(uninterruptibleMask(settle → publish toolResult))
  FS->>MT: settle({ sessionID, agent, assistantMessageID, call })
  MT->>TR: settleWith(input, registration.identity)
  TR->>TL: settle(tool, call, context)
  TL->>TL: 解码 input（失败 → ToolFailure）
  TL->>EX: execute(input, context)
  EX->>LM: resolve({ path, kind: file })
  LM-->>EX: Target { canonical, resource, externalDirectory? }
  opt 外部目录
    EX->>PM: assert({ action: external_directory, resources, save })
  end
  EX->>PM: assert({ action: edit, resources: [resource], save: [*], source })
  PM->>PM: evaluateInput：agent 规则 + 已保存规则 → allow / deny / ask
  alt ask
    PM->>EV: publish(permission.v2.asked)
    EV-->>UI: SSE 收到请求
    UI->>PM: reply({ requestID, reply: once/always/reject })
    PM-->>EX: Deferred 兑现（reject → die(DeclinedError)）
  else deny
    PM-->>EX: BlockedError → ToolFailure
  end
  EX->>EX: 读文件、精确匹配、writeIfUnchanged
  EX-->>TL: Output { files, replacements }
  TL->>TL: 编码 output，toModelOutput → content
  TL-->>TR: ToolOutput { structured, content }
  TR->>OS: bound({ sessionID, toolCallID, output })
  OS-->>TR: 超限则落盘 + 预览
  TR-->>MT: Settlement { result, output, outputPaths }
  MT-->>FS: settlement
  FS->>PB: publish(LLMEvent.toolResult, outputPaths)
  PB->>EV: publish(Tool.Success 或 Tool.Failed)
  Note over RT: 流结束后 awaitToolFibers 等所有 fiber

sequenceDiagram
  participant U as 用户
  participant SV as SessionV2.prompt
  participant DB as session_input 表
  participant CO as Coordinator
  participant RN as run 双层 while
  participant RT as runTurnAttempt

  U->>SV: prompt(A, steer)
  SV->>DB: A: admitted_seq=10, promoted_seq=NULL
  SV->>CO: wake → 空闲 → 起 drain
  CO->>RN: run(force=false)
  RN->>DB: hasPending(steer)=true → promotion=steer
  RN->>RT: runTurn(promotion=steer, step=1)
  RT->>DB: cutoff=latestSequence=10，promoteSteers(<=10) → A 变 user 消息
  Note over RT: 正在流式收响应
  U->>SV: prompt(B, steer)
  SV->>DB: B: admitted_seq=15, promoted_seq=NULL
  SV->>CO: wake → 正在跑 → pendingWake=true
  U->>SV: prompt(C, queue)
  SV->>DB: C: admitted_seq=16, promoted_seq=NULL
  RT-->>RN: { needsContinuation: true（有工具）, step: 1 }
  RN->>RN: promotion=steer, step=2
  RN->>RT: runTurn(steer, 2)
  RT->>DB: cutoff=20，promoteSteers → B 提升，promoted 大于 0 → currentStep=1
  RT-->>RN: { needsContinuation: false, step: 1 }
  RN->>DB: hasPending(steer)=false → 内层退出
  RN->>DB: hasPending(queue)=true → shouldRun=true, promotion=queue
  RN->>RT: runTurn(queue, 1)
  RT->>DB: promoteNextQueued → C 提升；再 promoteSteers（没有）
  RT-->>RN: { needsContinuation: false, step: 1 }
  RN->>DB: hasPending(queue)=false → 外层退出
  RN-->>CO: 结束
  CO->>CO: settle：pendingWake 为真 → 起后继 drain(force=false)
  CO->>RN: run → hasPending 都为 false → 直接 return

const hasSteer = yield* SessionInput.hasPending(db, sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, sessionID, "queue")  // 有 steer 就不看 queue
if (!input.force && !hasSteer && !hasQueue) return                                // wake 来的且没活 → 走人
yield* failInterruptedTools(input.sessionID)                                      // §12.3
let promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let shouldRun = input.force || hasSteer || hasQueue
while (shouldRun) {                                                               // 外层：queue
  let needsContinuation = true
  let step = 1
  while (needsContinuation) {                                                     // 内层：工具续接 + steer
    const result = yield* runTurn(sessionID, promotion, step)
    needsContinuation = result.needsContinuation
    step = result.step + 1                                                        // 注意 result.step 可能被重置成 1
    promotion = "steer"                                                           // 第二轮起只提升 steer
    if (!needsContinuation) needsContinuation = yield* hasPending(db, sessionID, "steer")   // 没工具了？看看有没有插话
  }
  shouldRun = yield* hasPending(db, sessionID, "queue")                           // 内层退出 → 看 queue
  promotion = shouldRun ? "queue" : undefined
}

sequenceDiagram
  participant RN as run
  participant T as runTurn
  participant A as runTurnAttempt
  participant C as SessionCompaction
  participant LLM as LLMClient
  participant EV as EventV2
  participant R as runAfterOverflowCompaction

  RN->>T: runTurn(sessionID, promotion, step)
  T->>A: runTurnAttempt(..., recoverOverflow=compactAfterOverflow)
  A->>A: 拼好 request
  A->>C: compactIfNeeded({ sessionID, entries, model, request })
  C->>C: auto 关？context 未知？估算 tokens 小于等于 context - max(output, buffer)？→ false
  alt 需要压缩
    C->>C: select(entries, keep.tokens) → head / recent
    C->>C: buildPrompt(previousSummary?, [prior.recent, head])
    C->>EV: publish(Compaction.Started { reason: auto })
    C->>LLM: stream(request{ model, messages: [user(summaryPrompt)], tools: [], maxTokens })
    LLM-->>C: text-delta 攒起来
    C->>EV: publish(Compaction.Ended { text: summary, recent })
    EV->>EV: 投影：插一条 compaction 消息
    C-->>A: true
    A->>A: die(ContinueAfterCompaction(step))
    T->>T: catchDefect 认出 → yieldNow → runTurn(sessionID, undefined, step)
    Note over T: 重新走一遍：entriesForRunner 从 compaction.seq 起读，prepare 发现新压缩 → replace 基线
  else 不需要
    A->>LLM: stream(request)
    LLM-->>A: provider-error classification=context-overflow（还没开始输出）
    A->>A: overflowFailure = event，不 publish
    A->>C: recoverOverflow = compactAfterOverflow(...)
    C-->>A: true
    A->>A: die(ContinueAfterOverflowCompaction(step))
    T->>R: runAfterOverflowCompaction(sessionID, undefined, step)
    R->>A: runTurnAttempt(..., recoverOverflow=undefined)
    A->>LLM: stream(request)
    alt 再次溢出
      A->>A: publish(overflowFailure) → Step.Failed
    end
  end

estimate({ system, messages, tools }) > context - max(generation.maxTokens ?? limits.output, buffer)

sequenceDiagram
  participant P as 发布方（发布器 / SessionInput / 门面）
  participant EV as EventV2.publish
  participant TX as SQLite 事务 immediate
  participant PJ as 投影函数们
  participant MU as SessionMessageUpdater
  participant SUB as 订阅者（SSE / durable 流）

  P->>EV: publish(定义, data, { commit? })
  EV->>EV: 补 id、location
  EV->>TX: begin
  TX->>TX: SELECT seq FROM event_sequence WHERE aggregate_id
  TX->>TX: seq = latest + 1，检查 event.id 不重复
  TX->>PJ: 逐个调 projectors[type]
  PJ->>MU: update(db 适配器, event)
  MU->>TX: UPDATE/INSERT session_message
  TX->>TX: options.commit(seq)（系统上下文 advance 快照时用）
  TX->>TX: UPSERT event_sequence，INSERT event
  TX-->>EV: commit
  EV->>SUB: PubSub.publish(durable wake) 叫醒该会话的 durable 流
  EV->>SUB: notify：listeners、typed PubSub、all PubSub
  EV-->>P: 带 durable.seq 的事件

classDiagram
  class SystemContextSource {
    key: Key
    codec: Schema
    load: Effect~A or Unavailable~
    baseline(current) string
    update(previous, current) string
    removed(previous) string
  }
  class SystemContext {
    <<opaque 一组 PackedSource>>
    make(source) SystemContext
    combine(contexts) SystemContext
    initialize(ctx) Generation
    reconcile(ctx, snapshot) Unchanged or Updated or ReplacementReady or Blocked
    replace(ctx, snapshot) ReplacementReady or Blocked
  }
  class Generation {
    baseline: string
    snapshot: Record~Key, SourceSnapshot~
  }
  class SystemContextRegistry {
    <<Service location>>
    entries: Ref~Entry[]~
    register(entry)
    load() SystemContext
  }
  class SystemContextBuiltIns {
    core/environment
    core/date
  }
  class SkillGuidance {
    load(agent) SystemContext
  }
  class ReferenceGuidance {
    load() SystemContext
  }
  SystemContext ..> SystemContextSource
  SystemContext ..> Generation
  SystemContextRegistry ..> SystemContext : combine
  SystemContextBuiltIns ..> SystemContextRegistry : register