5e282a39b4
Round2 主线:
- A: 事件落库地基 (RuntimeApp EventStore 单例 + 14 repo wiring)
- B: 执行体对齐 (read-before-edit, verification-before-completion)
- C: 界面对齐 (@opentui/solid, 删除 runtime 依赖)
- D: 经验闭环 (ExperienceMiner, DebuggerRole, CompactorRole)
Round2 补充修复:
- fail-on-missing 反作弊门禁
- projection-store-apply.test.ts 补写
- 3个空壳测试转行为 (evidence-store, recovery-impl, knowledge-store)
- ask 项目根支持 AIRCODING_PROJECT_ROOT
- Worker 事件契约修复 (task.attempt.started → checkpoint)
Round3-F: cpp 工具切换
- 删除 BuiltInToolRegistrar cpp.* 闭包
- 接入 toolchain-cpp 真实 CppToolRegistrar
- canonical envelope {status/output/metadata}
- ExecutorRole system prompt 对齐新工具名
Round3-H: Doctor 5 类报告
- toolchain (cmake/ninja/cppcheck/clangd/g++)
- display (X11/Wayland + ImageMagick)
- network (internet connectivity)
- provider (api_key/base_url/model/connectivity)
Secret 脱敏:
- 状态交接.md: sk- → \${OPENAI_API_KEY}
- .gitignore: 添加 .air/ .claude/
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
511 lines
18 KiB
TypeScript
Executable File
511 lines
18 KiB
TypeScript
Executable File
/**
|
|
* WorkerManager - Spawns and manages worker child processes
|
|
*
|
|
* Implements DD §8.1. spawn (Bun child process + handshake), cancel.
|
|
* INV-1: WorkerManager never writes agents.status directly.
|
|
*
|
|
* @module packages/runtime/src/workers/WorkerManager
|
|
*/
|
|
|
|
import { spawn } from 'child_process'
|
|
import { existsSync } from 'fs'
|
|
import type { ChildProcess } from 'child_process'
|
|
import { WorkerProcess, type WorkerExitCode } from './WorkerProcess.js'
|
|
import { WorkerProtocol, type WorkerMessage, type WorkerMessageType } from './WorkerProtocol.js'
|
|
import type { WorkerResult, WorkerStatus, AgentType } from '@aircoding/contracts'
|
|
import { eventIngestor } from '../events/EventIngestor.js'
|
|
import { eventSchemaRegistry } from '../events/EventSchemaRegistry.js'
|
|
import type { ToolRegistry } from '../tools/ToolRegistry.js'
|
|
import type { ProviderManager } from '@aircoding/llm'
|
|
|
|
export interface WorkerConfig {
|
|
entrypoint: string // Path to worker main.ts
|
|
agent_id: string
|
|
session_id: string
|
|
project_root: string
|
|
timeout_ms?: number
|
|
env?: Record<string, string>
|
|
task_type?: string // DD §8.3: execute/review/debug/compact/mine_experience
|
|
task_spec?: Record<string, unknown> // DD §9: TaskSpec payload for worker
|
|
}
|
|
|
|
export interface WorkerHandle {
|
|
worker_id: string
|
|
process: WorkerProcess
|
|
config: WorkerConfig
|
|
state: 'starting' | 'ready' | 'running' | 'completed' | 'failed' | 'error' | 'cancelled'
|
|
started_at: string
|
|
completed_at?: string
|
|
result?: WorkerResult<unknown>
|
|
}
|
|
|
|
export class WorkerManager {
|
|
private protocol: WorkerProtocol
|
|
private workers: Map<string, WorkerHandle> = new Map()
|
|
private tool_registry?: ToolRegistry
|
|
private provider_manager?: ProviderManager
|
|
private execution_context?: { session_id: string; project_id: string; project_root: string }
|
|
|
|
constructor(tool_registry?: ToolRegistry, provider_manager?: ProviderManager) {
|
|
this.protocol = new WorkerProtocol()
|
|
this.tool_registry = tool_registry
|
|
this.provider_manager = provider_manager
|
|
}
|
|
|
|
/**
|
|
* Set execution context (session_id, project_id, project_root).
|
|
* Required for tool/LLM calls from workers.
|
|
*/
|
|
set_context(ctx: { session_id: string; project_id: string; project_root: string }): void {
|
|
this.execution_context = ctx
|
|
}
|
|
|
|
/**
|
|
* Set tool registry (can be injected after construction).
|
|
*/
|
|
set_tool_registry(registry: ToolRegistry): void {
|
|
this.tool_registry = registry
|
|
}
|
|
|
|
/**
|
|
* Set provider manager (can be injected after construction).
|
|
*/
|
|
set_provider_manager(pm: ProviderManager): void {
|
|
this.provider_manager = pm
|
|
}
|
|
|
|
/**
|
|
* Spawn a worker child process and perform handshake.
|
|
* INV-1: worker.ready handshake is a live signal, not a status write.
|
|
*/
|
|
async spawn(config: WorkerConfig): Promise<WorkerHandle> {
|
|
const proc = new WorkerProcess()
|
|
const handle: WorkerHandle = {
|
|
worker_id: config.agent_id,
|
|
process: proc,
|
|
config,
|
|
state: 'starting',
|
|
started_at: new Date().toISOString()
|
|
}
|
|
this.workers.set(config.agent_id, handle)
|
|
|
|
// Spawn worker process using Bun
|
|
// Worker must run from AirCoding repo root so Bun can resolve modules
|
|
const repo_root = process.env.AIRCODING_REPO_ROOT || config.project_root
|
|
const bun_path = this.find_bun()
|
|
const entrypoint = config.entrypoint.startsWith('/') ? config.entrypoint
|
|
: `${repo_root}/${config.entrypoint.replace(/^\.\//, '')}`
|
|
|
|
const child = spawn(bun_path, ['run', entrypoint], {
|
|
stdio: ['pipe', 'pipe', 'pipe'],
|
|
env: {
|
|
...process.env,
|
|
...config.env,
|
|
AIRCODING_AGENT_ID: config.agent_id,
|
|
AIRCODING_SESSION_ID: config.session_id,
|
|
AIRCODING_PROJECT_ROOT: config.project_root
|
|
},
|
|
cwd: repo_root
|
|
})
|
|
|
|
proc.set_process(child)
|
|
|
|
proc.on_exit((exit) => this.handle_worker_exit(config.agent_id, exit))
|
|
|
|
// Set up handlers for worker IPC messages
|
|
this.setup_worker_handlers(proc, config.agent_id)
|
|
|
|
// Wait for handshake: worker.ready
|
|
await this.wait_for_handshake(proc, config)
|
|
|
|
// Validate protocol version and dispatch task
|
|
const ready_msg = this.send_and_wait(proc, 'agent.start', {
|
|
protocol_version: this.protocol.get_version(),
|
|
agent_id: config.agent_id,
|
|
session_id: config.session_id,
|
|
project_root: config.project_root,
|
|
task_type: config.task_type || 'execute',
|
|
task_spec: config.task_spec || { id: `${config.agent_id}_task`, title: 'Execute task', description: '' }
|
|
})
|
|
|
|
handle.state = 'ready'
|
|
|
|
// Set up timeout
|
|
if (config.timeout_ms) {
|
|
setTimeout(() => this.cancel(config.agent_id, 'timeout'), config.timeout_ms)
|
|
}
|
|
|
|
return handle
|
|
}
|
|
|
|
/**
|
|
* Cancel a worker.
|
|
*/
|
|
async cancel(agent_id: string, reason: string): Promise<void> {
|
|
const handle = this.workers.get(agent_id)
|
|
if (!handle) return
|
|
|
|
const msg = this.protocol.create_message('agent.cancel', { reason }, 'parent_to_worker')
|
|
handle.process.send(msg)
|
|
handle.state = 'cancelled'
|
|
|
|
// Wait briefly then force kill
|
|
setTimeout(() => {
|
|
if (handle.process.is_alive()) {
|
|
handle.process.kill('SIGKILL')
|
|
}
|
|
}, 5000)
|
|
}
|
|
|
|
/**
|
|
* Set up IPC message handlers for a worker process.
|
|
* Handles tool.call and llm.request forwarded from worker to parent.
|
|
*/
|
|
private setup_worker_handlers(proc: WorkerProcess, agent_id: string): void {
|
|
// Handle tool.call from worker → execute via ToolRegistry
|
|
proc.on_message('tool.call', async (msg) => {
|
|
const call_id = msg.payload.call_id as string
|
|
const tool_name = msg.payload.name as string
|
|
const tool_args = (msg.payload.arguments || {}) as Record<string, unknown>
|
|
|
|
try {
|
|
if (!this.tool_registry) {
|
|
this.send_to_worker(agent_id, 'tool.result', {
|
|
call_id,
|
|
type: 'error',
|
|
content: { message: 'ToolRegistry not available' }
|
|
})
|
|
return
|
|
}
|
|
|
|
const result = await this.tool_registry.call(
|
|
{ call_id, name: tool_name, arguments: tool_args },
|
|
{
|
|
session_id: this.execution_context?.session_id || msg.session_id,
|
|
project_id: this.execution_context?.project_id || '',
|
|
project_root: this.execution_context?.project_root || process.cwd(),
|
|
agent_id,
|
|
agent_type: 'executor',
|
|
}
|
|
)
|
|
|
|
this.send_to_worker(agent_id, 'tool.result', {
|
|
call_id,
|
|
type: result.status === 'ok' ? 'text' : 'error',
|
|
content: result.output || result.error || {}
|
|
})
|
|
} catch (e: any) {
|
|
console.error('[WM] tool.call error:', e.message)
|
|
this.send_to_worker(agent_id, 'tool.result', {
|
|
call_id,
|
|
type: 'error',
|
|
content: { message: e.message || 'Tool execution failed' }
|
|
})
|
|
}
|
|
})
|
|
|
|
// Handle llm.request from worker → execute via ProviderManager
|
|
proc.on_message('llm.request', async (msg) => {
|
|
const call_id = msg.payload.call_id as string
|
|
try {
|
|
if (!this.provider_manager) {
|
|
this.send_to_worker(agent_id, 'llm.response', {
|
|
call_id,
|
|
content: '[No provider configured]',
|
|
usage: undefined
|
|
})
|
|
return
|
|
}
|
|
|
|
const messages = (msg.payload.messages || []) as Array<{ role: string; content: unknown }>
|
|
const response = await this.provider_manager.complete_text(messages, {
|
|
model: msg.payload.model as string,
|
|
max_tokens: msg.payload.max_tokens as number,
|
|
temperature: msg.payload.temperature as number,
|
|
tools: (msg.payload.tools as unknown[]) || this.tool_registry?.list?.() || []
|
|
})
|
|
|
|
this.send_to_worker(agent_id, 'llm.response', {
|
|
call_id,
|
|
content: response.content,
|
|
usage: response.usage,
|
|
tool_calls: response.tool_calls
|
|
})
|
|
} catch (e: any) {
|
|
this.send_to_worker(agent_id, 'llm.response', {
|
|
call_id,
|
|
content: `[LLM Error: ${e.message}]`,
|
|
usage: undefined
|
|
})
|
|
}
|
|
})
|
|
|
|
// Handle worker-emitted RuntimeEvent payloads through the single EventIngestor entry point.
|
|
proc.on_message('event', async (msg) => {
|
|
try {
|
|
const event_type = (msg.payload.event_type || msg.payload.type) as string
|
|
if (!event_type || !eventSchemaRegistry.isRegistered(event_type, 1)) {
|
|
console.error(`[WM] ignoring unregistered worker event: ${event_type || '(missing)'}`)
|
|
return
|
|
}
|
|
|
|
const { event_type: _eventType, ...restPayload } = msg.payload
|
|
const payload = _eventType ? restPayload : (() => {
|
|
const { type: _legacyType, ...legacyPayload } = restPayload
|
|
return legacyPayload
|
|
})()
|
|
const event = {
|
|
id: (payload.event_id as string) || msg.id,
|
|
type: event_type,
|
|
version: 1,
|
|
timestamp: msg.timestamp || new Date().toISOString(),
|
|
session_id: this.execution_context?.session_id || msg.session_id,
|
|
project_id: this.execution_context?.project_id || '',
|
|
source: { kind: 'agent', id: agent_id, agent_type: this.worker_agent_type(agent_id) },
|
|
route: ['worker', agent_id, event_type],
|
|
payload,
|
|
}
|
|
|
|
const persistence = eventSchemaRegistry.getPersistence(event_type, 1)
|
|
if (persistence === 'durable') await eventIngestor.ingest(event as any)
|
|
else if (persistence === 'ephemeral') await eventIngestor.ingest_ephemeral(event as any)
|
|
} catch (e: any) {
|
|
console.error('[WM] worker event ingest error:', e.message)
|
|
}
|
|
})
|
|
|
|
// Handle worker.result → update handle
|
|
proc.on_message('worker.result', (msg) => {
|
|
const handle = this.workers.get(agent_id)
|
|
if (handle) {
|
|
handle.result = this.wrap_worker_result(msg.payload, handle)
|
|
handle.state = handle.result.status === 'completed' ? 'completed'
|
|
: handle.result.status === 'cancelled' ? 'cancelled'
|
|
: 'failed'
|
|
handle.completed_at = new Date().toISOString()
|
|
}
|
|
})
|
|
|
|
// Handle worker.checkpoint
|
|
proc.on_message('worker.checkpoint', (msg) => {
|
|
const handle = this.workers.get(agent_id)
|
|
if (handle) {
|
|
handle.state = 'running'
|
|
}
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Send a message back to a specific worker.
|
|
*/
|
|
private send_to_worker(agent_id: string, type: string, payload: Record<string, unknown>): void {
|
|
const handle = this.workers.get(agent_id)
|
|
if (!handle) return
|
|
const msg = this.protocol.create_message(type as WorkerMessageType, payload, 'parent_to_worker', {
|
|
session_id: this.execution_context?.session_id,
|
|
agent_id
|
|
})
|
|
handle.process.send(msg)
|
|
}
|
|
|
|
/**
|
|
* Send a message to a worker.
|
|
*/
|
|
send(agent_id: string, type: WorkerMessageType, payload: Record<string, unknown>): void {
|
|
const handle = this.workers.get(agent_id)
|
|
if (!handle) throw new Error(`Worker not found: ${agent_id}`)
|
|
|
|
const msg = this.protocol.create_message(type, payload, 'parent_to_worker')
|
|
handle.process.send(msg)
|
|
}
|
|
|
|
/**
|
|
* Get a worker handle.
|
|
*/
|
|
get(agent_id: string): WorkerHandle | undefined {
|
|
return this.workers.get(agent_id)
|
|
}
|
|
|
|
/**
|
|
* List all workers.
|
|
*/
|
|
list(): WorkerHandle[] {
|
|
return Array.from(this.workers.values())
|
|
}
|
|
|
|
/**
|
|
* List workers by state.
|
|
*/
|
|
list_by_state(state: WorkerHandle['state']): WorkerHandle[] {
|
|
return this.list().filter(w => w.state === state)
|
|
}
|
|
|
|
/**
|
|
* Check if any workers are running.
|
|
*/
|
|
has_running(): boolean {
|
|
return this.list().some(w => w.state === 'running' || w.state === 'ready' || w.state === 'starting')
|
|
}
|
|
|
|
/**
|
|
* Get the stored WorkerResult for an agent.
|
|
*/
|
|
get_result(agent_id: string): WorkerResult<unknown> | undefined {
|
|
const handle = this.workers.get(agent_id)
|
|
if (!handle) return undefined
|
|
return handle.result
|
|
}
|
|
|
|
/**
|
|
* Get result for a task.
|
|
*/
|
|
get_result_for_task(task_id: string): WorkerResult<unknown> | undefined {
|
|
return this.list().find(w => w.config.task_spec?.id === task_id || w.config.task_spec?.task_id === task_id || w.worker_id === `agent_${task_id}`)?.result
|
|
}
|
|
|
|
/**
|
|
* Get handle for a task.
|
|
*/
|
|
get_handle_for_task(task_id: string): WorkerHandle | undefined {
|
|
return this.list().find(w => w.config.task_spec?.id === task_id || w.config.task_spec?.task_id === task_id || w.worker_id === `agent_${task_id}`)
|
|
}
|
|
|
|
// ============================================================================
|
|
// Private
|
|
// ============================================================================
|
|
|
|
private worker_agent_type(agent_id: string): AgentType {
|
|
const handle = this.workers.get(agent_id)
|
|
const task_type = handle?.config.task_type || 'execute'
|
|
switch (task_type) {
|
|
case 'review': return 'reviewer' as AgentType
|
|
case 'debug': return 'debugger' as AgentType
|
|
case 'compact': return 'compactor' as AgentType
|
|
case 'mine_experience': return 'experience_miner' as AgentType
|
|
default: return 'executor' as AgentType
|
|
}
|
|
}
|
|
|
|
private handle_worker_exit(agent_id: string, exit: { code: number | null; signal: NodeJS.Signals | null; semantic: string; description: string }): void {
|
|
const handle = this.workers.get(agent_id)
|
|
if (!handle) return
|
|
if (handle.result) return
|
|
|
|
const task_id = (handle.config.task_spec?.id as string) || `${agent_id}_task`
|
|
const cancelled = exit.semantic === 'parent_cancelled'
|
|
handle.state = cancelled ? 'cancelled' : 'failed'
|
|
handle.completed_at = new Date().toISOString()
|
|
handle.result = {
|
|
task_id: task_id as any,
|
|
agent_id: handle.config.agent_id as any,
|
|
agent_type: 'executor',
|
|
status: cancelled ? 'cancelled' : 'failed',
|
|
summary: `Worker exited without result: ${exit.semantic}`,
|
|
changed_files: [],
|
|
artifacts: [],
|
|
verification: [],
|
|
risks: [],
|
|
follow_up_tasks: [],
|
|
evidence_refs: [],
|
|
result: { exit },
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Wrap a raw worker payload into a properly typed WorkerResult envelope.
|
|
* Provides safe defaults for any missing fields.
|
|
*/
|
|
private wrap_worker_result(payload: Record<string, unknown>, handle: WorkerHandle): WorkerResult<unknown> {
|
|
const raw_status = (payload.status as string) || 'completed'
|
|
const status = raw_status === 'completed' || raw_status === 'cancelled' || raw_status === 'blocked' || raw_status === 'failed'
|
|
? raw_status
|
|
: raw_status === 'fixed' || raw_status === 'cannot_reproduce' || raw_status === 'pass' || raw_status === 'compacted' || raw_status === 'skipped' || raw_status === 'no_patterns'
|
|
? 'completed'
|
|
: raw_status === 'escalated'
|
|
? 'blocked'
|
|
: 'failed'
|
|
const changes = Array.isArray((payload as any).changes) ? (payload as any).changes : []
|
|
const changed_files = (payload.changed_files as string[] | undefined) || changes.map((c: any) => String(c.file)).filter(Boolean)
|
|
const verification_payload = payload.verification as any
|
|
const verification = Array.isArray(verification_payload) ? verification_payload
|
|
: verification_payload ? [{ command: 'worker verification', passed: Boolean(verification_payload.passed), output: String(verification_payload.output || '') }] as any[]
|
|
: []
|
|
const summary = (payload.summary as string)
|
|
|| (payload.summary_content as string)
|
|
|| (payload.root_cause as string)
|
|
|| (payload.error ? String(payload.error) : '')
|
|
|| (changed_files.length > 0 ? `Changed files: ${changed_files.join(', ')}` : `Worker ${status}`)
|
|
|
|
return {
|
|
task_id: (payload.task_id as string) || (handle.config.task_spec?.id as string) || (handle.config.task_spec?.task_id as string) || '' as any,
|
|
agent_id: (payload.agent_id as string) || handle.config.agent_id as any,
|
|
agent_type: (payload.agent_type as AgentType) || this.worker_agent_type(handle.config.agent_id),
|
|
status: status as WorkerStatus,
|
|
summary,
|
|
changed_files,
|
|
diff_ref: (payload.diff_ref as string | undefined) || undefined,
|
|
artifacts: (payload.artifacts as any[]) || [],
|
|
verification,
|
|
risks: (payload.risks as any[]) || [],
|
|
follow_up_tasks: (payload.follow_up_tasks as any[]) || [],
|
|
evidence_refs: (payload.evidence_refs as any[]) || [],
|
|
result: (payload.result as unknown) || payload,
|
|
}
|
|
}
|
|
|
|
private async wait_for_handshake(proc: WorkerProcess, config: WorkerConfig): Promise<void> {
|
|
return new Promise((resolve, reject) => {
|
|
const timeout = setTimeout(() => {
|
|
reject(new Error(`Worker handshake timeout: ${config.agent_id}`))
|
|
}, 30000)
|
|
|
|
proc.on_message('worker.ready', (msg) => {
|
|
clearTimeout(timeout)
|
|
const version = msg.payload.protocol_version as number
|
|
const check = this.protocol.check_version(version)
|
|
if (!check.compatible) {
|
|
reject(new Error(check.error))
|
|
return
|
|
}
|
|
resolve()
|
|
})
|
|
|
|
// Also handle worker.error
|
|
proc.on_message('worker.error', (msg) => {
|
|
clearTimeout(timeout)
|
|
reject(new Error(`Worker error during handshake: ${msg.payload.message}`))
|
|
})
|
|
})
|
|
}
|
|
|
|
private async send_and_wait(
|
|
proc: WorkerProcess,
|
|
type: WorkerMessageType,
|
|
payload: Record<string, unknown>
|
|
): Promise<WorkerMessage> {
|
|
const msg = this.protocol.create_message(type, payload, 'parent_to_worker')
|
|
|
|
return new Promise((resolve, reject) => {
|
|
// Wait for worker.send callback (worker acknowledges agent.start)
|
|
// For now just send and resolve after a short delay
|
|
proc.send(msg)
|
|
setTimeout(() => resolve(msg), 100)
|
|
})
|
|
}
|
|
|
|
private find_bun(): string {
|
|
// Try common paths first (no shell, no string interpolation)
|
|
const candidates = [
|
|
process.env.BUN_INSTALL ? `${process.env.BUN_INSTALL}/bin/bun` : null,
|
|
`${process.env.HOME || '/root'}/.bun/bin/bun`,
|
|
'/usr/local/bin/bun',
|
|
'/usr/bin/bun',
|
|
].filter((p): p is string => Boolean(p))
|
|
|
|
for (const candidate of candidates) {
|
|
if (existsSync(candidate)) return candidate
|
|
}
|
|
return 'bun' // PATH fallback
|
|
}
|
|
}
|