opencode 源码解析

graph LR
  CLI["cli<br/>opencode 命令：拉起或连接服务、跑 TUI"] --> CLIENT
  TUI["tui<br/>终端界面"] --> CLIENT
  CLIENT["client<br/>生成的 HTTP 客户端 + 托管服务发现"] -->|HTTP / SSE| SERVER
  SERVER["server<br/>实现 protocol 定义的接口"] --> CORE
  PROTO["protocol<br/>HttpApi 接口定义"] -.-> SERVER
  PROTO -.-> CLIENT
  CORE["core<br/>会话、执行、工具、权限、事件、插件"] --> AI
  CORE --> CM["codemode<br/>JS 子集解释器"]
  CORE --> PLUGIN["plugin<br/>插件 API"]
  AI["ai<br/>统一请求/事件 + 各家协议"] --> SCHEMA
  CORE --> SCHEMA["schema<br/>纯类型：消息、事件、权限"]

flowchart TD
  P["Session.prompt：预处理输入"] --> A["SessionInbox.admit：写进 session_inbox"]
  A --> W["SessionExecution.wake"]
  W -->|空闲：起一个执行 fiber<br/>忙：按门铃| D["runner.drain"]
  D --> S["advanceToStep：处理控制项、投递输入、准备上下文"]
  S --> R["runStep：自动压缩检查、拼请求"]
  R --> T["SessionStep.attempt：读模型流，边读边并发跑工具"]
  T -->|Retry / Continue / Compacted / RecoverFull / OutputLimit| R
  T -->|Completed| N{"needsContinuation？"}
  N -->|是：有工具结果要交回模型| S
  N -->|否| S2["advanceToStep：还有待投递的输入？"]
  S2 -->|有| R
  S2 -->|没有| E["drain 返回 Complete"]
  E --> C{"执行期间有人按过门铃？"}
  C -->|是| D
  C -->|否| I["发 Execution.Succeeded，会话空闲"]

sequenceDiagram
  participant H as HTTP handler
  participant S as Session.prompt
  participant P as SessionPrompt.prepare
  participant IN as SessionInbox
  participant B as Bus / 投影
  participant EX as SessionExecution

  H->>S: prompt(sessionID, { id?, text, files, skills, delivery, resume })
  S->>IN: reconcile(id)：这条输入已经收过？
  alt 已经收过（客户端重试）
    IN-->>S: 返回已有记录
  else 新输入
    S->>P: prepare：插件 session.prompt 钩子、读附件、展开技能
    P-->>S: 一条 user 类型的收件箱条目
    S->>S: 会话上有暂存的 revert？先提交它
    S->>IN: admit
    IN->>B: publish(session.inbox.enqueued)
    B->>B: 同一事务：分配序号，投影插入 session_inbox 一行
  end
  opt resume 不是 false
    S->>EX: wake(sessionID)
  end
  S-->>H: 返回收件箱条目（不等模型跑完）

classDiagram
  class SessionHandle {
    <<Session.make().forSession(id)>>
    id: SessionID
    +get() / message(id)
    +prompt(input) SessionInbox.Info
    +synthetic(input) / skill(input) / shell(input)
    +compact() / revert(...)
    +inbox() / cancelInbox(id) / steerInbox(id) / queueInbox(id)
    +switchAgent / switchModel / rename / setPermissions
    +wait() / resume() / interrupt()
  }
  class SessionPrompt {
    <<模块>>
    +prepare(session, messageID, input) SessionInbox.Item
  }
  class SessionInbox {
    <<全局服务>>
    +reconcile(id, sessionID, type, delivery)
    +admit(id, sessionID, item)
    +admitCompaction(id, sessionID, delivery)
    +cancel / steer / queue(ref)
    +list(sessionID)
  }
  class SessionInboxModule {
    <<模块函数，runner 直接调>>
    +nextPromotable(db, sessionID, scope)
    +promote(db, bus, sessionID, scope) number
    +has(db, sessionID, scope)
    +serialized(sessionID, effect)
    +projectAdmitted / projectDelivered / projectCancelled
  }
  class SessionExecution {
    +wake(sessionID)
  }
  SessionHandle ..> SessionPrompt : 预处理
  SessionHandle ..> SessionInbox : 入库
  SessionHandle ..> SessionExecution : 叫醒
  SessionInbox ..> SessionInboxModule : 投影函数

// core/src/session/inbox.ts，promote 的主干
const steers = yield* pendingSteers(db, sessionID)
if (steers.length > 0 || scope === "steer") {
  const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move")
  if (control === 0) return undefined                  // 排在最前面的是控制项：交给 runner 先处理
  return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control))
}
// 没有 steer，且允许取 queue：只取最早的一条 queue
const queued = /* 最早一条 delivery = queue 的行 */
if (!queued) return 0
if (queued.type === "compaction" || queued.type === "move") return undefined
const promoted = yield* publish(db, bus, sessionID, [queued])
// 投递这条 queue 期间又来的 steer，跟着一起投
const arrivedSteers = yield* pendingSteers(db, sessionID)
...

classDiagram
  class SessionExecution {
    <<全局服务>>
    +active() Set~SessionID~
    +isActive(sessionID) boolean
    +resume(sessionID)
    +wake(sessionID)
    +interrupt(sessionID, options) boolean
    +awaitIdle(sessionID)
    -drain(sessionID, force, continuation, promotable)
  }
  class Coordinator {
    -executions: Map~SessionID, Execution~
    -fork: FiberSet 运行时
    +run(key)
    +wake(key, scope)
    +interrupt(key, reason)
    +awaitIdle(key)
    -start(key, force, scope)
    -loop(key, execution, force)
    -settle(key, execution, exit)
  }
  class Execution {
    done: Deferred
    owner?: Fiber
    scope: input 或 steer
    pendingWake?: input 或 steer
    stopping: boolean
    interruptionReason?
  }
  class SessionStore {
    +get(sessionID)
    +claim(sessionID) / release(sessionID)
    +listSuspended() / countResume(sessionID)
  }
  class SessionRunner {
    <<目录级服务>>
    +drain(sessionID, force, continuation, promotable) DrainResult
  }
  SessionExecution *-- Coordinator : make 时传入 started / drain / settled
  Coordinator o-- Execution : 每个忙碌的会话一个
  SessionExecution ..> SessionStore : 读会话、写声明
  SessionExecution ..> SessionRunner : 经 Instance.provide 取到

const wake = (key: Key, scope: Promotable = "input") =>
  Effect.sync(() => {
    const execution = executions.get(key)
    if (execution !== undefined) {
      // 合并多次 wake 时取范围大的："input" 包含 "steer"
      execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
      return
    }
    start(key, false, scope)
  })

const loop = (key, execution, force) =>
  Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
    Effect.andThen(Effect.suspend(() => {
      if (execution.stopping || execution.pendingWake === undefined) return Effect.void
      execution.scope = execution.pendingWake
      execution.pendingWake = undefined
      return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
    })),
  )

classDiagram
  class SessionRunner {
    <<目录级服务，runner/llm.ts>>
    -titles: FiberMap 生成标题的后台任务
    +drain(input) DrainResult
    -advanceToStep() Ready 或 DrainResult
    -prepareContext(sessionID) Selection
    -runStep(loaded, step) boolean
    -settleStaleCompactions(sessionID)
    -settleStaleToolCalls(sessionID)
  }
  class DrainResult {
    <<tagged enum>>
    Complete
    Moved(continuation?)
    Reloaded(force, continuation?)
  }
  class SessionContext {
    +select(sessionID) Selection
    +load(selection) Loaded
    +resolveModel(session)
    +request: SessionModelRequest
  }
  class Loaded {
    session
    agent
    model: Resolved
    initial: 指令基线文本
    messages: SessionMessage[]
    tools: Tool.Snapshot
  }
  class SessionModelRequest {
    +primary(input) Prepared
    +compaction(input) Prepared
    +generate(input) Prepared
    +title(input) Prepared
  }
  class Prepared {
    event
    request: LLMRequest
    options: http / webSocket
    retry(event)
    executeTool(call) NormalizedResult
  }
  class SessionStep {
    +attempt(input) Outcome
  }
  class SessionCompaction {
    +compact(trigger) skipped / completed / failed
  }
  class InstructionState {
    <<模块>>
    +prepare(db, bus, instructions, sessionID)
    +initial(db, sessionID, instructions)
  }
  SessionRunner ..> DrainResult : 返回
  SessionRunner --> SessionContext
  SessionRunner --> SessionStep
  SessionRunner --> SessionCompaction
  SessionRunner ..> InstructionState : prepareContext 时调
  SessionContext ..> Loaded : load 返回
  SessionContext --> SessionModelRequest
  SessionModelRequest ..> Prepared : 返回
  SessionStep ..> Prepared : 用它发请求、执行工具

sequenceDiagram
  participant U as 用户
  participant IN as session_inbox
  participant EX as Coordinator
  participant AD as advanceToStep
  participant RS as runStep

  U->>IN: A（steer）
  U->>EX: wake → 空闲，起执行
  EX->>AD: drain(force=false, scope=input)
  AD->>IN: promote("input") → A 变成用户消息
  AD->>RS: Step 1
  Note over RS: 模型回复里有工具调用，正在跑
  U->>IN: B（steer）、C（queue）
  U->>EX: wake → 忙，按门铃
  RS-->>AD: needsContinuation = true
  AD->>IN: 继续中：只能 promote("steer") → B 变成用户消息
  AD->>RS: Step 1（有新输入，步数重置）
  RS-->>AD: needsContinuation = false
  AD->>IN: 不再继续：promote("input") → 没有 steer，取一条 queue：C
  AD->>RS: Step 1
  RS-->>AD: needsContinuation = false
  AD->>IN: promote("input") → 没东西了
  AD-->>EX: Complete
  EX->>EX: 门铃响过 → 再 drain 一次 → 没事可做 → 结束

if (!force && !continuing) {
  const pending = yield* SessionInbox.nextPromotable(db, sessionID, "input")
  if (!pending) return DrainResult.Complete()                       // 没活就直接走
  const control = pending.type === "compaction" || pending.type === "move"
  if (promotable === "steer" && pending.delivery === "queue" && !control) return DrainResult.Complete()
}
yield* plugins.awaitActivation
yield* settleStaleCompactions(sessionID)   // 上个进程死时没收尾的压缩，标成失败
yield* settleStaleToolCalls(sessionID)     // 上个进程死时还在跑的工具调用，标成失败

while (true) {
  const next = yield* advanceToStep()
  if (next._tag !== "Ready") return next              // Complete / Moved / Reloaded
  continuing = yield* runStep(next.context, step)     // 返回 needsContinuation
  step++
  force = false
  entering = false
}

while (true) {
  const loaded = initial ?? (yield* prepareContext(sessionID).pipe(Effect.flatMap(context.load)))
  initial = undefined
  const compacted = yield* compaction.compact({ reason: "auto", context: loaded })   // 先看要不要自动压缩
  if (compacted.status === "failed") return yield* new StepFailedError(...)
  if (compacted.status === "completed") { assistantMessageID = SessionMessage.ID.create(); continue }

  const transcript = SessionModelRequest.baseTranscript({ ... })     // §6.4
  const prepared = yield* context.request.primary({ ... })           // §6.5
  const outcome = yield* steps.attempt({ ..., prepared, retry, recoverOverflow, recoverContinuation })  // §7
  const completed = yield* Outcome.$match(outcome, { ... })
  if (completed !== undefined) return completed
}

// core/src/session/context.ts，select 的主干
const session = yield* store.get(sessionID)
if (session.location.directory !== location.directory || ...) return yield* Effect.interrupt   // 会话已经搬走了
yield* mcpTools.flush
const agent = yield* agents.select(session.agent)
const permissions = Permission.merge(agent.info.permissions, session.permissions ?? [])
const loaded = yield* Effect.all({
  tools: registry.snapshot(permissions),          // 工具快照，§8
  builtins: builtins.load(),                      // 内置指令：日期、环境等
  discovery: discovery.load(),                    // 项目里发现的 AGENTS.md 之类
  skills: skillInstructions.load(permissions),    // 可用技能的说明
  references: referenceInstructions.load(),
  mcp: mcpInstructions.load(permissions),         // MCP 服务给的说明
  entries: entries.load(sessionID),               // 通过 API 给这个会话设置的指令条目
}, { concurrency: "unbounded" })
return {
  session, agent, tools: loaded.tools,
  instructions: Instructions.combine([
    CodeModeInstructions.make(loaded.tools.codeModeCatalog),
    loaded.mcp, loaded.references, loaded.skills, loaded.discovery, loaded.builtins, loaded.entries,
  ]),
}

system: [
  input.agent.system ? input.agent.system : SessionSystemPrompt.make(toolNames),  // agent 自带的提示词，或者默认提示词
  input.initial,                                                                   // 指令基线
].filter((part) => part.length > 0).map(SystemPart.make),
messages: toLLMMessages(input.messages, input.model.ref),

sequenceDiagram
  participant AT as attempt
  participant LLM as LLMClient.stream
  participant PB as publisher
  participant T as 工具 fiber × n
  participant B as Bus

  AT->>AT: snapshots.capture()：记下工作区快照
  AT->>LLM: stream(request)
  loop 每个模型事件
    LLM-->>AT: LLMEvent
    alt 输出开始前就报“上下文太长”
      AT->>AT: 记下来，先不发布
    else
      AT->>PB: publish(event)
      PB->>B: step.started / text.* / reasoning.* / tool.input.* ...
      opt 是一个完整的本地工具调用
        PB->>B: tool.called（持久化之后才开始执行）
        AT->>T: forkScoped：执行 → 截断输出 → tool.success / tool.failed
      end
    end
  end
  AT->>PB: flush：把攒着的增量发出去
  AT->>T: Fiber.awaitAll：等所有工具结束
  AT->>AT: 判断：压缩恢复？重试？续传？失败？
  AT->>B: step.ended（用量、费用、改动的文件）或 step.failed

classDiagram
  class SessionStep {
    +attempt(input) Outcome
  }
  class AttemptInput {
    <<step.ts 里的 Input>>
    sessionID
    assistantMessageID
    agent / model
    prepared: Prepared
    retry(cause, error, retryable) Decision
    recoverContinuation: boolean
    recoverOverflow: Effect~boolean~
    isLocationClosed()
  }
  class Publisher {
    <<createLLMEventPublisher 的返回值>>
    -tools: Map 调用 id → 状态
    -chunks: 攒着的增量
    +publish(event)
    +progress(callID, update)
    +toolExecution(id, name, result)
    +failTool(id, error)
    +failUnsettledTools(error)
    +failIncompleteTools()
    +failAssistant(error)
    +flush() / streamed()
    +startAssistant()
    +record() StepRecord
  }
  class StepRecord {
    outputStarted: boolean
    providerFailed: boolean
    failure?
    finish?: finish / tokens / providerState
    needsContinuation: boolean
  }
  class Outcome {
    <<tagged enum>>
    Completed(needsContinuation)
    Retry(error, decision)
    Continue(error, decision)
    Compacted
    RecoverFull
    OutputLimit(needsContinuation)
  }
  SessionStep ..> AttemptInput
  SessionStep *-- Publisher : 每次 attempt 新建
  Publisher ..> StepRecord : record()
  SessionStep ..> Outcome : 返回

classDiagram
  class ToolInfo {
    name
    description
    input: Schema
    output?: Schema
    options?: namespace / codemode / permission
    execute(input, context) Result
  }
  class ToolContext {
    sessionID
    agent
    messageID：发起调用的 assistant 消息
    id：调用 id
    progress(update)
  }
  class Result {
    output?：给 Code Mode 的机器值
    content：给模型看的内容
    metadata?：给 UI 的小块 JSON
  }
  class ToolService {
    <<Location 级>>
    transform(editor => ...)
    reload()
    list()
    snapshot(permissions) Snapshot
  }
  class Snapshot {
    definitions: ToolDefinition[]
    codeModeCatalog?
    execute(sessionID, agent, messageID, call, progress, definitions)
  }
  ToolService ..> Snapshot : snapshot
  ToolInfo ..> ToolContext : execute 收到
  ToolInfo ..> Result : execute 返回

tool.execute.before 钩子（可以修复参数）
  → 用输入 schema 校验参数（不合法就不执行）
  → 工具的 execute
  → 用输出 schema 编码 output
  → tool.execute.after 钩子（可以改结果）
  → 规整 content，压缩图片
  → runner 里的 ToolOutput.truncate
  → tool.success / tool.failed 事件

classDiagram
  class Permission {
    <<目录级服务>>
    -pending: Map~ID, Pending~
    -closed: boolean
    +ask(input) id 和 effect
    +assert(input)
    +reply(requestID, reply, message?)
    +get(id) / forSession(sessionID) / list()
    +close()
  }
  class Pending {
    request: Request
    agent?
    deferred: Deferred
  }
  class Request {
    id
    sessionID
    action
    resources: string[]
    save?: string[]
    metadata?
    source: 哪个工具调用
    message?
  }
  class Rule {
    action
    resource
    effect: allow / deny / ask
  }
  class PermissionSaved {
    +list(projectID)
    +add(projectID, action, resources)
  }
  Permission o-- Pending : 等待回复的请求
  Pending --> Request
  Permission ..> Rule : 求值
  Permission --> PermissionSaved : 总是允许的规则

const rules = merge(agent.permissions, session.permissions)        // agent 的规则 + 会话的规则
if (任一资源命中 deny) return "deny"
const all = [...rules, ...已保存的 always 规则]                       // 用户点过“总是允许”的
const effect = 任一资源是 ask ? "ask" : "allow"                       // 每个资源取最后一条匹配的规则；都没匹配算 ask
const event = yield* hooks.trigger("permission", "evaluate", { ..., effect })   // 插件可以改结论
return event.effect

第一次：[E1 … E8][E9 E10]               →  [S1 + "E9 E10"]
再一次：[S1 + "E9 E10"][E11 … E18][E19] →  [S2 + "E19"]    S2 覆盖 S1、E9 E10 和 E11–E18

classDiagram
  class Bus {
    <<全局服务>>
    -projectors: Map~事件类型, 投影器[]~
    -pubsub: live / durable / typed
    +publish(definition, data, options) Payload
    +publishAll(events)
    +project(definition, projector)
    +subscribe()
    +log(aggregateID, after, follow) Stream
    +replay(event, options)
  }
  class EventDefinition {
    type
    schema
    durable?: aggregate 字段名 / version
  }
  class PublishOptions {
    id?
    metadata?
    commit?(seq)
  }
  class SessionProjector {
    <<session/projector.ts>>
    为每种会话事件注册投影器
  }
  Bus ..> EventDefinition
  Bus ..> PublishOptions
  SessionProjector ..> Bus : project(...)

sequenceDiagram
  participant C as 调用方
  participant B as Bus.publish
  participant DB as SQLite 事务
  participant P as 投影器
  participant S as 订阅者

  C->>B: publish(definition, data, { commit? })
  B->>B: 按聚合 id（通常是会话 id）加锁
  B->>DB: 开事务
  DB->>DB: 读 event_sequence，分配下一个序号
  DB->>P: 依次跑这个事件类型的所有投影器
  P->>DB: 改 session_v2 / session_message / session_inbox / instruction_state ...
  DB->>DB: 执行 commit 回调（比如写执行声明）
  DB->>DB: 更新 event_sequence，插入 event 行
  DB-->>B: 提交
  B->>S: 推给在线订阅者（live / durable / typed 三种 PubSub）