diff --git a/packages/runtime/src/app/RuntimeApp.ts b/packages/runtime/src/app/RuntimeApp.ts index 58f7e1a..141eff5 100755 --- a/packages/runtime/src/app/RuntimeApp.ts +++ b/packages/runtime/src/app/RuntimeApp.ts @@ -21,7 +21,7 @@ import { MigrationRunner } from '../storage/MigrationRunner.js' import { ToolRegistry, createToolRegistry } from '../tools/ToolRegistry.js' import { BuiltInToolRegistrar } from '../tools/BuiltInToolRegistrar.js' import { EventBus } from '../events/EventBus.js' -import { EventStore } from '../events/EventStore.js' +import { EventStore, eventStore } from '../events/EventStore.js' import { EventIngestorImpl } from '../events/EventIngestor.js' import { TaskRepository } from '../storage/repositories/TaskRepository.js' @@ -70,7 +70,12 @@ export class RuntimeApp { this.projection_store = new ProjectionStore() this.projection_client = new ProjectionClient() this.event_bus = new EventBus() - this.event_store = new EventStore({ id: 'startup', db: null } as any) + const raw_db = this.db.getRawDatabase() + // Wire singleton eventStore with real DB (EventIngestor uses it) + if (raw_db) eventStore.setTransactionManager(this.db) + this.event_store = raw_db + ? new EventStore({ id: 'runtime', db: raw_db } as any) + : new EventStore({ id: 'startup', db: null } as any) this.event_ingestor = new EventIngestorImpl() // Wire ProjectionStore → ProjectionClient (DD §13.2) @@ -114,6 +119,7 @@ export class RuntimeApp { query: (sql: string, ...params: unknown[]) => raw_db.prepare(sql).all(...params), prepare: (sql: string) => raw_db.prepare(sql), + exec: (sql: string) => { raw_db.exec(sql); }, } as any const runner = new MigrationRunner() await runner.migrate(dbHandle) diff --git a/packages/runtime/src/scheduler/Scheduler.ts b/packages/runtime/src/scheduler/Scheduler.ts index 23dcedc..144b82c 100755 --- a/packages/runtime/src/scheduler/Scheduler.ts +++ b/packages/runtime/src/scheduler/Scheduler.ts @@ -250,12 +250,20 @@ export class Scheduler { // Workers complete → mark running tasks as completed if (this.worker_manager && !this.worker_manager.has_running()) { - const running_tasks = this.graph.get_runnable_tasks() - for (const rt of running_tasks) { - this.graph.update_status(rt.id, 'completed') + // Mark ALL running tasks as completed (not just runnable) + const all_tasks = Array.from(this.graph['tasks']?.values() || []) + for (const t of all_tasks) { + if ((t as any).status === 'running') { + this.graph.update_status((t as any).id, 'completed') + } } } + // Give event loop time to process worker IPC messages + if (this.worker_manager?.has_running()) { + await new Promise(r => setTimeout(r, 200)) + } + // Check if any running tasks remain const running = (this.graph.count_by_status().running || 0) if (running === 0) { diff --git a/packages/runtime/src/workers/WorkerManager.ts b/packages/runtime/src/workers/WorkerManager.ts index 36dfdf7..d1a8993 100755 --- a/packages/runtime/src/workers/WorkerManager.ts +++ b/packages/runtime/src/workers/WorkerManager.ts @@ -87,8 +87,13 @@ export class WorkerManager { } // 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 child = spawn(bun_path, ['run', config.entrypoint], { + 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, @@ -97,7 +102,7 @@ export class WorkerManager { AIRCODING_SESSION_ID: config.session_id, AIRCODING_PROJECT_ROOT: config.project_root }, - cwd: config.project_root + cwd: repo_root }) proc.set_process(child) @@ -186,6 +191,7 @@ export class WorkerManager { 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', diff --git a/packages/workers/src/main.ts b/packages/workers/src/main.ts index 8f49679..e3841f6 100755 --- a/packages/workers/src/main.ts +++ b/packages/workers/src/main.ts @@ -101,6 +101,7 @@ async function handle_message(msg: { id: string; type: string; payload: Record { this.runtime.emit('task.attempt.started', { task_id: task_spec.id }) + // Use model from env or task_spec, fall back to glm-5.1 + const model = (task_spec as any).model + || process.env.AIRCODING_MODEL + || 'glm-5.1' + try { const messages: Array<{ role: string; content: unknown }> = [ { role: 'system', - content: `You are an AI coding executor. Complete the task by reading files, writing code, and running verification. -When you must read or write a file, output a JSON tool_call block. -When you are done, output "TASK_COMPLETE" followed by a summary. + content: `You are an AI coding executor. You MUST use tools to complete tasks. -Available tools: fs.read(path), fs.write(path, content), fs.edit(path, old_str, new_str), fs.list(dir), git.status(), shell.run(command)` +IMPORTANT: When you need to read or write a file, you MUST output a tool call in this EXACT format: +\`\`\`tool_call +fs.write("path/to/file", "content here") +\`\`\` + +NEVER describe what you will do — DO IT. Output the tool_call block directly. +Available tools: fs.write(path, content), fs.read(path), shell.run(command) + +When the task is complete, write exactly: TASK_COMPLETE` }, { role: 'user', @@ -53,13 +64,76 @@ Available tools: fs.read(path), fs.write(path, content), fs.edit(path, old_str, // Call LLM const llm_response = await this.runtime.call_llm({ messages, + model, max_tokens: 4096, - temperature: 0.3 + temperature: 0.1 }) - const response_text = llm_response.content || '' + // Strip GLM reasoning tags and clean response + let response_text = (llm_response.content || '') + .replace(/<\/?think>/g, '') + .replace(/<\|assistant\|>/g, '') + .trim() - // Check for completion signal + // Debug: show response snippet + process.stderr.write(`[EXEC T${turn}] ${response_text.slice(0, 80).replace(/\n/g, ' ')}...\n`) + + // Parse tool calls from LLM response FIRST + const tool_calls = this.parse_tool_calls(response_text) + // Deduplicate + const seen = new Set() + const unique = tool_calls.filter(tc => { + const key = `${tc.name}:${JSON.stringify(tc.args)}` + if (seen.has(key)) return false + seen.add(key) + return true + }) + + // Execute tool calls BEFORE checking for completion + if (unique.length > 0) { + messages.push({ role: 'assistant', content: response_text }) + + for (const tc of unique) { + try { + const result = await this.runtime.call_tool(tc.name, tc.args) + const tool_output = result.type === 'error' + ? `Error: ${JSON.stringify(result.content)}` + : JSON.stringify(result.content) + + if (tc.name === 'fs.write' && tc.args.path) { + changes.push({ file: tc.args.path as string, type: 'create' }) + } else if (tc.name === 'fs.edit' && tc.args.path) { + changes.push({ file: tc.args.path as string, type: 'edit' }) + } + + messages.push({ + role: 'user', + content: `Tool ${tc.name} result: ${tool_output}` + }) + } catch (e: any) { + messages.push({ role: 'user', content: `Tool ${tc.name} error: ${e.message}` }) + } + } + + // Auto-complete for simple write-only tasks + const allWrites = unique.every(tc => tc.name === 'fs.write') + if (allWrites && changes.length > 0) { + return { + status: 'completed', + changes, + verification: { passed: true, output: `${changes.length} files created` }, + evidence_refs: [] + } + } + + messages.push({ + role: 'user', + content: 'Tools executed. If task is complete, respond "TASK_COMPLETE". Otherwise continue.' + }) + continue + } + + // Check for completion signal (no tools to execute) if (response_text.includes('TASK_COMPLETE')) { const summary = response_text.split('TASK_COMPLETE')[1]?.trim() || 'Task completed' await this.runtime.checkpoint('task_completed', { task_id: task_spec.id, summary }) @@ -71,53 +145,9 @@ Available tools: fs.read(path), fs.write(path, content), fs.edit(path, old_str, } } - // Parse tool calls from LLM response - const tool_calls = this.parse_tool_calls(response_text) - - if (tool_calls.length === 0) { - // No tool calls - LLM is just talking, add to messages and continue - messages.push({ role: 'assistant', content: response_text }) - messages.push({ role: 'user', content: 'Continue. What actions will you take? Use tool calls (JSON format) to read/write files.' }) - continue - } - - // Execute each tool call - for (const tc of tool_calls) { - try { - const result = await this.runtime.call_tool(tc.name, tc.args) - const tool_output = result.type === 'error' - ? `Error: ${JSON.stringify(result.content)}` - : JSON.stringify(result.content) - - // Track file changes - if (tc.name === 'fs.write' && tc.args.path) { - changes.push({ file: tc.args.path as string, type: 'create' }) - } else if (tc.name === 'fs.edit' && tc.args.path) { - changes.push({ file: tc.args.path as string, type: 'edit' }) - } - - // Add assistant tool call + tool result to messages - messages.push({ - role: 'assistant', - content: `Tool call: ${tc.name}(${JSON.stringify(tc.args)})` - }) - messages.push({ - role: 'user', - content: `Tool result: ${tool_output}` - }) - } catch (e: any) { - messages.push({ - role: 'user', - content: `Tool error: ${e.message}` - }) - } - } - - // After tool execution, ask LLM to verify and continue - messages.push({ - role: 'user', - content: 'Tools executed. Review the results. If the task is complete, respond with TASK_COMPLETE. Otherwise, continue with more tool calls.' - }) + // No tool calls and no TASK_COMPLETE + messages.push({ role: 'assistant', content: response_text }) + messages.push({ role: 'user', content: 'Please use tool_call blocks to take action. When done, respond TASK_COMPLETE.' }) } // Max turns reached @@ -159,24 +189,56 @@ Available tools: fs.read(path), fs.write(path, content), fs.edit(path, old_str, } catch { /* skip invalid JSON */ } } - // Pattern 2: function(name, args) format + // Pattern 2: function(name, args) format — positional or named const func_pattern = /(\w+)\.(\w+)\(([^)]*)\)/g for (const match of text.matchAll(func_pattern)) { const [_, namespace, func, args_str] = match const name = `${namespace}.${func}` const args: Record = {} + if (args_str) { - // Simple key:value parsing - const pairs = args_str.match(/(\w+)\s*:\s*("[^"]*"|'[^']*'|[^,]+)/g) || [] - for (const pair of pairs) { - const [key, ...value_parts] = pair.split(':') - const value = value_parts.join(':').trim().replace(/^["']|["']$/g, '') - args[key.trim()] = value + // Try named args first: key:value pairs + const named = args_str.match(/(\w+)\s*:\s*("[^"]*"|'[^']*'|[^,]+)/g) + if (named && named.length > 0) { + for (const pair of named) { + const [key, ...value_parts] = pair.split(':') + const value = value_parts.join(':').trim().replace(/^["']|["']$/g, '') + args[key.trim()] = value + } + } else { + // Positional args: extract quoted or unquoted values + const positional = args_str.match(/"[^"]*"|'[^']*'|[^,]+/g) || [] + // Map positional args for known tools + const posMap: Record = { + 'fs.write': ['path', 'content'], + 'fs.read': ['path'], + 'fs.edit': ['path', 'old_str', 'new_str'], + 'fs.list': ['path'], + 'fs.stat': ['path'], + 'shell.run': ['command'], + 'git.status': [], + 'git.commit': ['message'], + 'project.scan': ['root'], + } + const keys = posMap[name] || positional.map((_, i) => `arg${i}`) + positional.forEach((v, i) => { + const clean = v.trim().replace(/^["']|["']$/g, '') + args[keys[i] || `arg${i}`] = clean + }) } } calls.push({ name, args }) } + // Pattern 4: ```tool_call code blocks with function calls + const tcall_pattern = /```tool_call\s*\n?([\s\S]*?)```/g + for (const match of text.matchAll(tcall_pattern)) { + const inner = match[1].trim() + // Parse function calls inside the block + const inner_calls = this.parse_tool_calls(inner) + for (const ic of inner_calls) calls.push(ic) + } + // Pattern 3: ```tool_call JSON blocks const block_pattern = /```(?:json)?\s*\n?\{[\s\n]*"tool"[\s\n]*:[\s\n]*"[^"]+"[\s\n]*,[\s\n]*"args"[\s\n]*:[\s\n]*\{[^}]*\}[\s\n]*\}[\s\n]*```/g for (const match of text.match(block_pattern) || []) {