583 lines
31 KiB
TypeScript
583 lines
31 KiB
TypeScript
/** 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,
|
||
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,
|
||
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,
|
||
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 } } }
|
||
}
|