feat: complete WorkBuddy parity and Windows release workflow

src/main/chat-turn-service.ts

@@ -0,0 +1,584 @@
+/** Shared main-process turn execution. IPC and background automations use the same locks, native history and persistence. */
+import { app, type WebContents } from 'electron'
+import { randomUUID } from 'node:crypto'
+import { readFile, unlink, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import type { ChatEvent, ChatMessage, TokenUsage, ContextUsageCategory } from '@shared/types'
+import type { ActiveChatRun, SendPayload, SendResult } from '@shared/ipc'
+import { freezeAssistantModelMetadata } from '@shared/assistant-model'
+import { applyAssistantEvent } from '@shared/message-parts'
+import * as config from './config'
+import * as sessions from './sessions'
+import * as models from './models'
+import { Gateway } from './gateway'
+import { runTurn } from './agent/loop'
+import { OmpKernel, detectOmpPath } from './omp'
+import { runOmpTurn } from './agent/omp-turn'
+import { createMessageCheckpoint } from './message-checkpoint'
+import { classifyWorkspacePath } from './resources'
+import { createInteractionBroker } from './interaction-broker'
+import { attachmentPrompt, commitAttachments, getAttachment } from './attachments'
+import { rollbackEditedArtifacts } from './workspace-revert'
+import { normalizeUserContent } from '../shared/user-content'
+import { assistantWireHistory } from './agent/history'
+import { buildProjectPrompt, type ProjectRepository } from './projects'
+import * as generalSettings from './general-settings'
+import * as personalization from './personalization'
+import * as memorySettings from './memory-settings'
+import { prepareOmpModel, supportsOmpModel } from './omp-model'
+import { modelThinkingLevel } from '@shared/model-thinking'
+import type { ResolvedModel } from '@shared/models'
+import { createCapabilitySnapshot } from './capabilities/runtime'
+import { agentMode } from './agent/mode-policy'
+import { prepareContextObserver, writeContextRequest, observeOmpContextSamples } from './omp-context-observer'
+import { contextInputOrigins, estimateContextInput } from './context-usage-estimator'
+import { getAgentSettings, onAgentSettingsChanged } from './agent-settings'
+import { SubagentScheduler } from './subagents/scheduler'
+import { createSubagentWorker } from './subagents/worker'
+import { SUBAGENT_INSTRUCTIONS } from './subagents/tools'
+import { addUsage, zeroUsage } from '@shared/subagents'
+import { toolSpecs } from './agent/tools'
+import { modeToolSpecs } from './agent/mode-policy'
+
+export interface ChatTurnCompletion {
+  runId: string; sessionId: string; assistantMessageId: string
+  status: 'done' | 'cancelled' | 'error'; error?: string
+}
+export interface StartedChatTurn extends SendResult { completion: Promise<ChatTurnCompletion> }
+export interface ChatTurnServicePorts {
+  projects: ProjectRepository
+  sessionRoot(sessionId: string): Promise<string>
+  broadcastChat(channel: string, payload: unknown, sender?: WebContents): void
+  getDefaultRecipient?(): WebContents | undefined
+}
+
+/** Compensation for a managed task rejected before its first turn was persisted. */
+export async function removeEmptyManagedSession(sessionId: string): Promise<boolean> {
+  return withChatSessionLock(sessionId, async () => {
+    if ([...runTasks.values()].some(run => run.sessionId === sessionId)) return false
+    if ((await sessions.getMessages(sessionId)).length) return false
+    await disposeSessionKernel(sessionId)
+    await removeOmpSessionFile(sessionId)
+    await sessions.deleteSession(sessionId)
+    return true
+  })
+}
+
+/** runId → 取消控制器 */
+export const activeRuns = new Map<string, AbortController>()
+export const runTasks = new Map<string, { sessionId: string; promise: Promise<void>; snapshot: () => ActiveChatRun }>()
+export const runEmitters = new Map<string, (event: ChatEvent) => void>()
+const chatSessionLocks = new Map<string, Promise<void>>()
+export async function withChatSessionLock<T>(id: string, operation: () => Promise<T>): Promise<T> {
+  const previous = chatSessionLocks.get(id) ?? Promise.resolve()
+  let release!: () => void
+  const pending = new Promise<void>((resolve) => { release = resolve })
+  chatSessionLocks.set(id, pending)
+  await previous
+  try { return await operation() }
+  finally { release(); if (chatSessionLocks.get(id) === pending) chatSessionLocks.delete(id) }
+}
+
+interface OmpKernelEntry {
+  kernel: OmpKernel
+  workspaceRoot: string
+  appliedBranchId?: string
+  fingerprint?: string
+  stopContextObserver?: () => void
+}
+export async function applyPendingNativeBranch(sessionId: string, kernel: OmpKernel): Promise<void> {
+  const pendingId = await sessions.getPendingNativeBranch(sessionId)
+  const entry = ompKernels.get(sessionId)
+  if (!pendingId || entry?.appliedBranchId === pendingId) return
+  const response = await kernel.send({ type: 'branch', entryId: pendingId })
+  if (response.success === false) throw new Error('无法恢复已保存的原生会话分支')
+  if (entry) entry.appliedBranchId = pendingId
+}
+
+/** Each YCode session gets a native OMP transcript; existing YCode-only history is not imported. */
+export const ompKernels = new Map<string, OmpKernelEntry>()
+const ompKernelLocks = new Map<string, Promise<void>>()
+const ompRetirements = new Map<string, Promise<void>>()
+let ompShutdownStarted = false
+let disposeAllPromise: Promise<void> | null = null
+
+export function ompSessionPath(sessionId: string): string {
+  return join(
+    app.getPath('userData'),
+    'omp-sessions',
+    `session-${encodeURIComponent(sessionId)}.jsonl`
+  )
+}
+
+async function withOmpKernelLock<T>(sessionId: string, operation: () => Promise<T>): Promise<T> {
+  const previous = ompKernelLocks.get(sessionId) ?? Promise.resolve()
+  let release = (): void => {}
+  const current = new Promise<void>((resolve) => {
+    release = resolve
+  })
+  ompKernelLocks.set(sessionId, current)
+  await previous
+  try {
+    return await operation()
+  } finally {
+    release()
+    if (ompKernelLocks.get(sessionId) === current) ompKernelLocks.delete(sessionId)
+  }
+}
+
+function evictOmpKernel(sessionId: string, kernel: OmpKernel): void {
+  if (ompKernels.get(sessionId)?.kernel !== kernel) return
+  ompKernels.get(sessionId)?.stopContextObserver?.()
+  ompKernels.delete(sessionId)
+  const retirement = Promise.resolve()
+    .then(() => kernel.dispose())
+    .catch(() => undefined)
+  ompRetirements.set(sessionId, retirement)
+  void retirement.then(() => {
+    if (ompRetirements.get(sessionId) === retirement) ompRetirements.delete(sessionId)
+  })
+}
+
+async function waitForOmpRetirement(sessionId: string): Promise<void> {
+  await ompRetirements.get(sessionId)
+}
+
+export async function getOmpKernel(sessionId: string, workspaceRoot: string, model?: ResolvedModel | null): Promise<OmpKernel | null> {
+  return withOmpKernelLock(sessionId, async () => {
+    if (ompShutdownStarted) return null
+
+    if (model && !supportsOmpModel(model)) {
+      const existing = ompKernels.get(sessionId)
+      if (existing) { ompKernels.delete(sessionId); existing.stopContextObserver?.(); await existing.kernel.dispose() }
+      await waitForOmpRetirement(sessionId)
+      return null
+    }
+
+    const prepared = model ? await prepareOmpModel(model) : null
+    const existing = ompKernels.get(sessionId)
+    if (existing?.workspaceRoot === workspaceRoot && existing?.fingerprint === prepared?.fingerprint) return existing.kernel
+
+    if (existing) {
+      ompKernels.delete(sessionId)
+      existing.stopContextObserver?.()
+      await existing.kernel.dispose()
+    }
+    await waitForOmpRetirement(sessionId)
+
+    const exePath = detectOmpPath()
+    if (!exePath) return null
+
+    const observer = await prepareContextObserver(app.getPath('userData'),sessionId)
+    const kernel = new OmpKernel(exePath, workspaceRoot, prepared?.profile ?? 'ycode', ompSessionPath(sessionId), prepared?.model,
+      {...prepared?.env,YCODE_CONTEXT_OBSERVER_EXTENSION:observer.extensionPath,YCODE_CONTEXT_REQUEST_FILE:observer.configPath,YCODE_CONTEXT_SAMPLE_FILE:observer.samplePath})
+    const entry: OmpKernelEntry = { kernel, workspaceRoot, fingerprint: prepared?.fingerprint,
+      stopContextObserver: observeOmpContextSamples(kernel, observer.samplePath) }
+    const evict = (): void => evictOmpKernel(sessionId, kernel)
+    kernel.on('error', evict)
+    kernel.on('exit', evict)
+    ompKernels.set(sessionId, entry)
+    return kernel
+  })
+}
+
+export async function disposeOmpKernels(): Promise<void> {
+  if (disposeAllPromise) return disposeAllPromise
+  ompShutdownStarted = true
+  disposeAllPromise = (async () => {
+    for (const controller of activeRuns.values()) controller.abort()
+    while (ompKernelLocks.size > 0) {
+      await Promise.all([...ompKernelLocks.values()])
+    }
+    const entries = [...ompKernels.values()]
+    ompKernels.clear()
+    await Promise.all(
+      entries.map(async ({ kernel, stopContextObserver }) => {
+        stopContextObserver?.()
+        await kernel.dispose().catch(() => undefined)
+      })
+    )
+    await Promise.all([...ompRetirements.values()])
+    await Promise.all([...runTasks.values()].map((task) => task.promise))
+  })()
+  return disposeAllPromise
+}
+
+export async function disposeSessionKernel(sessionId: string): Promise<void> {
+  await withOmpKernelLock(sessionId, async () => {
+    const entry = ompKernels.get(sessionId)
+    if (entry) ompKernels.delete(sessionId)
+    if (entry) { entry.stopContextObserver?.(); await entry.kernel.dispose() }
+    await waitForOmpRetirement(sessionId)
+  })
+}
+
+export async function removeOmpSessionFile(sessionId: string): Promise<void> {
+  try {
+    await unlink(ompSessionPath(sessionId))
+  } catch (err) {
+    if ((err as NodeJS.ErrnoException).code !== 'ENOENT') throw err
+  }
+}
+
+export async function snapshotNativeSession(sessionId: string): Promise<() => Promise<void>> {
+  let before: Buffer | null = null
+  try { before = await readFile(ompSessionPath(sessionId)) }
+  catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error }
+  return async () => {
+    // Stop the writer before restoring the exact transcript consumed by --resume.
+    await disposeSessionKernel(sessionId)
+    if (before === null) await removeOmpSessionFile(sessionId)
+    else await writeFile(ompSessionPath(sessionId), before)
+  }
+}
+
+export async function compensateFailedAction(error: unknown, actions: Array<(() => Promise<void>) | undefined>): Promise<never> {
+  const failures: unknown[] = []
+  for (const action of actions) {
+    try { await action?.() } catch (failure) { failures.push(failure) }
+  }
+  if (failures.length) throw new AggregateError([error, ...failures], '操作失败，且原有会话或产物未能完全恢复，请检查工作区后重试。')
+  throw error
+}
+
+
+export function createChatTurnService({ projects, sessionRoot, broadcastChat, getDefaultRecipient }: ChatTurnServicePorts) {
+  const startTurn = async (payload: SendPayload, sender?: WebContents): Promise<StartedChatTurn> => withChatSessionLock(payload.sessionId, async () => {
+    if (ompShutdownStarted) throw new Error('应用正在退出')
+    const recipient = sender ?? getDefaultRecipient?.()
+    const { sessionId, text } = payload
+    if (payload.rollbackArtifacts !== undefined && typeof payload.rollbackArtifacts !== 'boolean') throw new Error('产物回退选项无效')
+    if (!(await sessions.listSessions()).some((session) => session.id === sessionId)) throw new Error('会话不存在')
+    if (typeof text !== 'string' || (!text.trim() && !payload.attachments?.length && !payload.quotes?.length && !payload.resources?.length && !payload.skillIds?.length)) throw new Error('消息为空')
+    if ([...runTasks.values()].some((run) => run.sessionId === sessionId)) throw new Error('该会话正在运行')
+    const attachments = await Promise.all((payload.attachments ?? []).map((item) => getAttachment(item.id)))
+    const promptAttachments = await attachmentPrompt(attachments)
+    const context = await sessions.getSessionContext(sessionId)
+    const project = context.projectId ? await projects.get(context.projectId) : undefined
+    const mode = agentMode(context.agentMode)
+    const selectedModel = await models.resolveModel(context.modelId)
+    if (!selectedModel) throw new Error('请先添加并选择自定义模型')
+    const snapshot = selectedModel?.config ?? await config.getGatewayConfig()
+    const workspaceRoot = await sessionRoot(sessionId)
+    const references = [...new Set(payload.resources ?? [])]
+    const capabilitySnapshot = await createCapabilitySnapshot(payload.skillIds ?? [], workspaceRoot, { sessionId, expertId: context.expertId, agentMode: mode })
+    const [personalizationView, memoryView, agentSettings] = await Promise.all([
+      personalization.getPersonalization(), memorySettings.getMemorySettings(), getAgentSettings()
+    ])
+    const personalizationPrompt = personalization.buildPersonalizationPrompt(personalizationView, {
+      includeLongTermMemory: memoryView.conversationMemory
+    })
+    const promptContext = [capabilitySnapshot.context, buildProjectPrompt(project), personalizationPrompt,
+      agentSettings.teamManagement && mode === 'craft' && selectedModel.profile.supportsToolCall !== false ? SUBAGENT_INSTRUCTIONS : ''].filter(Boolean).join('\n\n')
+    for (const path of references) {
+      if (typeof path !== 'string' || await classifyWorkspacePath(workspaceRoot, path) === 'unknown') throw new Error('引用资源不在工作区内或已不存在')
+    }
+    const kernel = await getOmpKernel(sessionId, workspaceRoot, selectedModel)
+    if (kernel) await applyPendingNativeBranch(sessionId, kernel)
+    const nativeBefore = kernel ? await kernel.send({ type: 'get_branch_messages' }) : null
+    const nativeBeforeIds = new Set((nativeBefore?.data as { messages?: Array<{ entryId?: string }> } | undefined)?.messages?.map((item) => item.entryId) ?? [])
+    await commitAttachments(attachments)
+    let history = [...await sessions.getMessages(sessionId)]
+    let selectedNativeBranch: string | undefined
+    let editUndo: (() => Promise<void>) | undefined
+    let undoNative: (() => Promise<void>) | undefined
+    if (payload.action) {
+      let targetIndex = history.findIndex((message) => message.id === payload.fromMessageId)
+      if (targetIndex < 0) throw new Error('未找到需要重发的消息')
+      if (payload.action === 'regenerate') {
+        if (history[targetIndex].role !== 'assistant') throw new Error('只能重新生成助手消息')
+        targetIndex = history.slice(0, targetIndex).findLastIndex((message) => message.role === 'user')
+      }
+      const target = history[targetIndex]
+      if (!target || target.role !== 'user') throw new Error('未找到对应用户消息')
+      if (kernel) undoNative = await snapshotNativeSession(sessionId)
+      // 原版编辑语义（userMessage.edit.hint）：产物回退至此消息前版本，再从此处重新开始对话。
+      editUndo = payload.action === 'edit' && payload.rollbackArtifacts !== false
+        ? await rollbackEditedArtifacts(workspaceRoot, history, payload.fromMessageId!) : undefined
+      try {
+        if (kernel) {
+        if (target.nativeSessionPath && target.nativeSessionPath !== ompSessionPath(sessionId)) throw new Error('这条历史消息没有当前 OMP 会话的分支记录')
+        let entryId = target.nativeEntryId
+        if (!entryId) {
+          const following = history[targetIndex + 1]
+          const prefixUsers = history.slice(0, targetIndex).filter((message) => message.role === 'user')
+          const nativeUsers = (nativeBefore?.data as { messages?: Array<{ entryId: string; text: string }> } | undefined)?.messages ?? []
+          const isLastInput = !history.slice(targetIndex + 1).some((message) => message.role === 'user')
+          const canRecover = isLastInput && following?.role === 'assistant' && ['cancelled', 'error'].includes(following.status ?? '') &&
+            prefixUsers.every((message, index) => !!message.nativeEntryId && message.nativeEntryId === nativeUsers[index]?.entryId)
+          if (!canRecover) throw new Error('这条历史消息没有当前 OMP 会话的分支记录')
+          const attached = await attachmentPrompt(target.attachments ?? [])
+          const nativePrompt = [target.content, target.quotes?.map((quote) => quote.markdown.trim()).filter(Boolean).join('\n\n'), attached.text, target.resources?.length ? `引用工作区资源：\n${target.resources.join('\n')}` : '', target.capabilityContext].filter(Boolean).join('\n\n')
+          if (nativeUsers.length === prefixUsers.length + 1 && nativeUsers.at(-1)?.text === nativePrompt) entryId = nativeUsers.at(-1)!.entryId
+          else if (nativeUsers.length !== prefixUsers.length) throw new Error('中断会话的原生历史与消息记录不一致')
+        }
+        if (entryId) {
+          try {
+            const result = await kernel.send({ type: 'branch', entryId })
+            if (result.success === false) throw new Error(String(result.error ?? '原生会话分支失败'))
+          } catch (error) { await disposeSessionKernel(sessionId); throw error }
+          selectedNativeBranch = entryId
+        }
+        const entry = ompKernels.get(sessionId)
+        if (entry && entryId) entry.appliedBranchId = entryId
+        }
+      } catch (error) {
+        await compensateFailedAction(error, [undoNative, editUndo])
+      }
+      history = history.slice(0, targetIndex)
+    }
+    const runId = randomUUID()
+
+    const userMessage: ChatMessage = {
+      id: randomUUID(),
+      role: 'user',
+      content: text,
+      requestId: runId,
+      planId: payload.planId,
+      attachments,
+      resources: references,
+      quotes: payload.quotes?.length ? payload.quotes : undefined,
+      userContent: normalizeUserContent(payload.userContent, text, { attachments, resources: references, quotes: payload.quotes, skills: capabilitySnapshot.references }),
+      skills: capabilitySnapshot.references,
+      capabilityContext: promptContext || undefined,
+      expertId: context.expertId ?? undefined,
+      createdAt: Date.now(),
+      status: 'done'
+    }
+    let assistant: ChatMessage = {
+      id: randomUUID(),
+      role: 'assistant',
+      content: '',
+      ...freezeAssistantModelMetadata(selectedModel.profile),
+      assistantDisplay: capabilitySnapshot.assistantDisplay,
+      requestId: runId,
+      tools: [],
+      parts: [],
+      createdAt: Date.now(),
+      turnStartedAt: userMessage.createdAt,
+      status: 'streaming'
+    }
+    try { await sessions.startTurn(sessionId, userMessage, assistant, history, payload.action, selectedNativeBranch) }
+    catch (error) {
+      await compensateFailedAction(error, [undoNative, editUndo])
+    }
+    editUndo = undefined
+    undoNative = undefined
+    let eventSequence = 0
+    let usage: TokenUsage | undefined
+    const declaredContextWindow = selectedModel.profile.contextWindowTokens ?? selectedModel.profile.maxInputTokens
+    const contextWindowAtStart = typeof declaredContextWindow === 'number' && Number.isFinite(declaredContextWindow) && declaredContextWindow > 0
+      ? declaredContextWindow : undefined
+
+    // 回合开始时冻结配置：运行中改设置不影响本回合
+    const controller = new AbortController()
+    activeRuns.set(runId, controller)
+    // 通用设置「运行时防止系统休眠」：任务运行期间持有 powerSaveBlocker。
+    generalSettings.updateActiveRuns(1)
+
+    const publish = (event: ChatEvent): void => {
+      broadcastChat('chat:event', { ...event, sessionId }, recipient)
+    }
+    const checkpoint = createMessageCheckpoint(
+      (message) => sessions.patchMessage(sessionId, message.id, {
+        content: message.content,
+        tools: message.tools,
+        parts: message.parts,
+        interactions: message.interactions,
+        status: message.status,
+        turnStartedAt: message.turnStartedAt,
+        turnDurationMs: message.turnDurationMs,
+        contextUsage: message.contextUsage,
+        contextBreakdown: message.contextBreakdown,
+        runUsage: message.runUsage,
+        error: message.error
+      }),
+      (error) => console.error('会话增量保存失败，将在回合结束时重试', error)
+    )
+    let terminalEvent: ChatEvent | null = null
+    let scheduler: SubagentScheduler | undefined
+    let mainUsage = zeroUsage(), childUsage = zeroUsage()
+    const publishReportedUsage = (): void => {
+      if (scheduler?.hasTasks) emit({ runId, type: 'run_usage', usage: { main: mainUsage, subagents: childUsage, total: addUsage(mainUsage, childUsage), reportedOnly: true } })
+    }
+    const emit = (event: ChatEvent): void => {
+      const isTerminal = event.type === 'done' || event.type === 'cancelled' || event.type === 'error'
+      let observedEvent: ChatEvent = isTerminal
+        ? { ...event, sequence: ++eventSequence, turnDurationMs: Math.max(0, Date.now() - userMessage.createdAt) }
+        : { ...event, sequence: ++eventSequence }
+      if (observedEvent.type === 'usage') {
+        mainUsage = addUsage(mainUsage, observedEvent.usage)
+        observedEvent = { ...observedEvent, contextWindow: contextWindowAtStart,source:'upstream',sampledAt:Date.now(),modelId:assistant.modelId }
+        usage = observedEvent.usage
+      }
+      if (observedEvent.type === 'context_breakdown') usage = undefined
+      assistant = applyAssistantEvent(assistant, observedEvent)
+      if (isTerminal) {
+        terminalEvent = observedEvent
+      } else {
+        checkpoint.schedule(assistant)
+        publish(observedEvent)
+      }
+      if (event.type === 'usage') publishReportedUsage()
+    }
+    runEmitters.set(runId, emit)
+    const interactionBroker = createInteractionBroker(runId, controller.signal, emit)
+    scheduler = new SubagentScheduler({ runId, workspaceRoot, modelName: selectedModel.profile.name,
+      enabled: agentSettings.teamManagement && mode === 'craft' && selectedModel.profile.supportsToolCall !== false,
+      signal: controller.signal, emit, capabilities: capabilitySnapshot,
+      onUsage: reported => { childUsage = addUsage(childUsage, reported); publishReportedUsage() },
+      worker: (task, tools) => createSubagentWorker({ task, tools, model: selectedModel, workspaceRoot, parentRunId: runId, parentSessionId: sessionId,
+        native: !!kernel, permissionMode: payload.permissionMode ?? 'default-sandbox',
+        thinkingLevel: modelThinkingLevel(selectedModel.profile, context.thinkingLevel), emitParent: emit })
+    })
+    const teamScheduler = scheduler
+    const unsubscribeAgents = onAgentSettingsChanged(next => teamScheduler.setEnabled(next.teamManagement && mode === 'craft' && selectedModel.profile.supportsToolCall !== false))
+    teamScheduler.onToolsChanged = () => {
+      if (!kernel || controller.signal.aborted) return
+      const available = modeToolSpecs([...toolSpecs(), ...teamScheduler.tools.specs], mode)
+      void kernel.setHostTools(available.map(spec => ({ ...spec.function, label: spec.function.name })))
+        .catch(error => console.error('更新智能体工具列表失败；宿主调用限制仍然生效', error))
+    }
+    // 对话中引用的内容块（原版 message-quote）随消息一起交给模型；表格 markdown 原样附带。
+    const quoteText = (payload.quotes ?? []).map((quote) => quote.markdown.trim()).filter(Boolean).join('\n\n')
+    // Freeze appended context with this message so fallback history and interrupted retries use the same input.
+    const promptText = [text, quoteText, promptAttachments.text, references.length ? `引用工作区资源：\n${references.join('\n')}` : '', promptContext].filter(Boolean).join('\n\n')
+    const origins = contextInputOrigins([...history.map(message=>message.capabilityContext??''),promptContext])
+    const toolCategories:Record<string,ContextUsageCategory> = {}
+    for (const spec of capabilitySnapshot.specs) {
+      if(capabilitySnapshot.metadata?.(spec.function.name)?.capability==='mcp')toolCategories[spec.function.name]='mcp'
+    }
+    for (const message of history) for(const tool of message.tools??[]) {
+      if(tool.metadata?.capability==='mcp')toolCategories[tool.name]='mcp'
+    }
+
+    // 回合在后台推进，立刻把 runId 还给渲染进程
+    const initialMessages = payload.action ? structuredClone([...history, userMessage, assistant]) : undefined
+    const runSnapshot = (): ActiveChatRun => ({
+      sessionId, runId, assistantMessageId: assistant.id, modelId: assistant.modelId, modelName: assistant.modelName,
+      assistantDisplay: assistant.assistantDisplay, userMessage: structuredClone(userMessage),
+      messages: structuredClone([...history, userMessage, assistant]), sequence: eventSequence,
+      ...(usage ? { usage: { ...usage } } : {})
+    })
+    // Publish the run identity before any sidecar can emit its first delta.
+    const task = Promise.resolve().then(async () => {
+      try {
+        // 内核选择：优先 OMP（外部 Agent 内核），不可用时回退自建循环
+        if (kernel) {
+          const observer = await prepareContextObserver(app.getPath('userData'),sessionId)
+          await writeContextRequest(observer.configPath,{runId,sessionPath:ompSessionPath(sessionId),modelId:assistant.modelId??'',
+            modelName:selectedModel.profile.model,contextWindow:contextWindowAtStart,origins,toolCategories})
+          const result = await runOmpTurn({
+            sessionId,
+            capabilityTools: teamScheduler.tools,
+            beforeFinish: () => teamScheduler.beforeFinish(),
+            refreshHostTools: true,
+            agentMode: mode,
+            thinkingLevel: modelThinkingLevel(selectedModel.profile, context.thinkingLevel),
+            runId,
+            kernel,
+            nativeSessionPath:ompSessionPath(sessionId),
+            supportsToolCall: selectedModel?.profile.supportsToolCall,
+            supportsImages: selectedModel.profile.supportsImages === true,
+            userText: promptText,
+            images: promptAttachments.images,
+            workspaceRoot,
+            permissionMode: payload.permissionMode ?? 'default-sandbox',
+            interactions: interactionBroker,
+            signal: controller.signal,
+            emit
+          })
+          assistant.content = result.content
+          assistant.tools = result.tools
+          // 终态由内核回合给出（abort 后到达的 agent_end 记为 cancelled）
+          assistant.status = result.status
+          const branch = await kernel.send({ type: 'get_branch_messages' })
+          const data = branch.data as { messages?: Array<{ entryId?: string; text?: string }> } | undefined
+          const native = data?.messages?.findLast(message => message.text === promptText && !!message.entryId && !nativeBeforeIds.has(message.entryId))
+          if (native?.entryId && !nativeBeforeIds.has(native.entryId) && native.text === promptText) {
+            await sessions.patchMessage(sessionId, userMessage.id, { nativeEntryId: native.entryId, nativeSessionPath: ompSessionPath(sessionId) })
+            await sessions.setPendingNativeBranch(sessionId, undefined)
+          }
+        } else {
+          const wireHistory = (await Promise.all(history.map(async (message) => {
+            if (message.role === 'assistant') return assistantWireHistory(message, { supportsImages: selectedModel.profile.supportsImages === true })
+            const attached = await attachmentPrompt(message.attachments ?? [])
+            const content = [message.content, message.quotes?.map((quote) => quote.markdown.trim()).filter(Boolean).join('\n\n'), attached.text, message.resources?.length ? `引用工作区资源：\n${message.resources.join('\n')}` : '', message.capabilityContext].filter(Boolean).join('\n\n')
+            return [{ role: message.role, content: attached.images.length ? [{ type: 'text' as const, text: content }, ...attached.images.map((image) => ({ type: 'image_url' as const, image_url: { url: `data:${image.mimeType};base64,${image.data}` } }))] : content }]
+          }))).flat()
+          const result = await runTurn({
+            sessionId,
+            supportsImages: selectedModel.profile.supportsImages === true,
+            capabilityTools: teamScheduler.tools,
+            beforeFinish: () => teamScheduler.beforeFinish(),
+            agentMode: mode,
+            runId,
+            history,
+            wireHistory,
+            userText: promptText,
+            images: promptAttachments.images,
+            interactions: interactionBroker,
+            permissionMode: payload.permissionMode ?? 'default-sandbox',
+            signal: controller.signal,
+            emit,
+            gateway: new Gateway({ ...snapshot, thinkingLevel: modelThinkingLevel(selectedModel.profile, context.thinkingLevel) },body=>{
+              const input=body as {messages?:unknown[];input?:unknown[];contents?:unknown[];instructions?:unknown;system?:unknown;systemInstruction?:{parts?:unknown[]};tools?:unknown[]}
+              const counts=estimateContextInput({messages:input.messages??input.input??input.contents??[],systemPrompt:input.instructions??input.system??input.systemInstruction?.parts,tools:input.tools??[],origins,toolCategories})
+              emit({runId,type:'context_breakdown',breakdown:{...counts,requestId:runId,sequence:(assistant.contextBreakdown?.sequence??0)+1,
+                sampledAt:Date.now(),modelId:assistant.modelId,contextWindow:contextWindowAtStart,method:'utf8-estimate',scope:'provider-request'}})
+            }),
+            supportsToolCall: selectedModel?.profile.supportsToolCall,
+            model: snapshot.model,
+            maxToolRounds: snapshot.maxToolRounds,
+            workspaceRoot
+          })
+          assistant.content = result.content
+          assistant.tools = result.tools
+          // 旧循环只在正常完成时走到这里；abort/error 走下面的 catch
+          assistant.status = 'done'
+        }
+      } catch (err) {
+        if ((err as Error).name === 'AbortError' || controller.signal.aborted) {
+          emit({ runId, type: 'cancelled' })
+        } else {
+          emit({ runId, type: 'error', message: err instanceof Error ? err.message : String(err) })
+        }
+      } finally {
+        controller.abort()
+        unsubscribeAgents()
+        await teamScheduler.close()
+        publishReportedUsage()
+        if(kernel)await unlink(join(app.getPath('userData'),'context-usage',`${encodeURIComponent(sessionId)}.json`)).catch(()=>undefined)
+        try {
+          // Child cleanup can publish final snapshots after the kernel has yielded.
+          // The terminal event must remain the last sequence consumed by live renderers.
+          if (terminalEvent) {
+            terminalEvent = { ...terminalEvent, sequence: ++eventSequence, turnDurationMs: Math.max(0, Date.now() - userMessage.createdAt) }
+            assistant = applyAssistantEvent(assistant, terminalEvent)
+          }
+          await checkpoint.finish(assistant)
+          if (terminalEvent) publish(terminalEvent)
+        } catch (error) {
+          const failure: ChatEvent = { runId, type: 'error', sequence: ++eventSequence,
+            message: `会话保存失败：${error instanceof Error ? error.message : String(error)}` }
+          assistant = applyAssistantEvent(assistant, failure)
+          publish(failure)
+        } finally {
+          activeRuns.delete(runId)
+          generalSettings.updateActiveRuns(-1)
+          runEmitters.delete(runId)
+        }
+      }
+    })
+    runTasks.set(runId, { sessionId, promise: task, snapshot: runSnapshot })
+    broadcastChat('chat:run-started', runSnapshot(), recipient)
+    void task.finally(() => runTasks.delete(runId))
+
+    const completion = task.then((): ChatTurnCompletion => ({ runId, sessionId, assistantMessageId: assistant.id,
+      status: assistant.status === 'done' ? 'done' : assistant.status === 'cancelled' ? 'cancelled' : 'error',
+      ...(assistant.error ? { error: assistant.error } : {}) }), (error): ChatTurnCompletion => ({
+        runId, sessionId, assistantMessageId: assistant.id, status: 'error', error: error instanceof Error ? error.message : String(error) }))
+    return { completion, runId, assistantMessageId: assistant.id, modelId: assistant.modelId, modelName: assistant.modelName,
+      assistantDisplay: assistant.assistantDisplay, userMessage, ...(initialMessages ? { messages: initialMessages } : {}) }  })
+  return { startTurn, cancel(runId: string) { const controller = activeRuns.get(runId); controller?.abort(); return { cancelled: !!controller } } }
+}