admin/AirCoding
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: integrate audit findings round 1 - tools, worker, scheduler, main agent

Browse Source 
- Unify ToolResultEnvelope (output vs content) for built-in tools
- Fix shell.run AsyncGenerator consumption in ToolRegistry.call/streaming
- Scheduler: consume WorkerResult.status instead of marking all running tasks completed
- WorkerProcess/WorkerManager: surface exit events and generate failed/cancelled result
- MainAgent: integrate ContextAssembler, Chinese destructive regex, ArchitectureDesigner impact gate
- run.ts: pendingConfirmation flow, dispatch extracted, .air files filtered from /results
- CapabilityRegistry wired into RuntimeApp and ServiceRegistry; DoctorService uses it
- release.ts: findRepoRoot/findBun, run air e2e + depcruise + runtime regression
- New gates: release-critical-gates, CLI run command regression
- 14/14 e2e gates pass; 3/3 release dry-run pass

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
AirCoding
2026-06-05 18:39:10 +08:00
parent a2d7aa0339
commit ddefcbb2b1
24 changed files with 1992 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
packages/runtime/src/agents/main/MainAgent.ts
View File

@@ -8,7 +8,9 @@
* @module packages/runtime/src/agents/main/MainAgent
*/
import type { SessionID, ProjectID } from '@aircoding/contracts'
import type { SessionID, ProjectID, AgentID, TaskID } from '@aircoding/contracts'
import type { ContextAssembler } from '../../context/ContextAssembler.js'
import { ArchitectureDesigner } from '../architecture/ArchitectureDesigner.js'
export type MainAgentState =
| 'IDLE'
@@ -34,13 +36,23 @@ export interface MainAgentConfig {
project_id: ProjectID
classify_mode?: ClassifyMode // Alpha default: 'regex'; set to 'llm' to use LLM classification
provider_manager?: any // ProviderManager for LLM-based classify
classify_model?: string // Model to use for LLM classification (e.g. 'claude-haiku-4-5')
context_assembler?: ContextAssembler
architecture_designer?: ArchitectureDesigner
project_root?: string
agent_id?: AgentID
task_id?: TaskID
classify_model?: string // Model to use for LLM classification and answer mode
}
export class MainAgent {
private config: MainAgentConfig
private classify_mode: ClassifyMode
private provider_manager?: any
private context_assembler?: ContextAssembler
private architecture_designer: ArchitectureDesigner
private project_root: string
private agent_id: AgentID
private task_id?: TaskID
private classify_model: string
state: MainAgentState = 'IDLE'
@@ -48,6 +60,11 @@ export class MainAgent {
this.config = config
this.classify_mode = config.classify_mode || 'regex'
this.provider_manager = config.provider_manager
this.context_assembler = config.context_assembler
this.architecture_designer = config.architecture_designer || new ArchitectureDesigner()
this.project_root = config.project_root || process.cwd()
this.agent_id = config.agent_id || 'main-agent' as AgentID
this.task_id = config.task_id
this.classify_model = config.classify_model || 'claude-haiku-4-5'
}
@@ -75,10 +92,19 @@ export class MainAgent {
case 'implementation_request':
case 'task_request':
// Check if this request needs user confirmation (breaking/delete)
if (/break|delete|remove|drop|destroy|truncate/i.test(message)) {
if (/break|delete|remove|drop|destroy|truncate|rm\s|\bdel\b/i.test(message) || /删除|删掉|清除|移除|销毁/.test(message)) {
this.state = 'CONFIRMING'
return { action: 'delegate', response: 'This appears to be a breaking or destructive change. Are you sure you want to proceed? (y/n)' }
}
const impact = this.architecture_designer.assess_impact({ description: message, files: this.infer_changed_files(message) })
if (impact.result === 'reject_or_escalate' || impact.result === 'requires_replan') {
this.state = 'ARCHITECTURE_DESIGNING'
return { action: 'answer', response: `Architecture review required: ${impact.risks.join('; ') || impact.change_summary}` }
}
if (impact.result === 'requires_user_confirmation') {
this.state = 'CONFIRMING'
return { action: 'delegate', response: `Architecture impact requires confirmation: ${impact.risks.join('; ') || impact.change_summary}. Proceed? (y/n)` }
}
this.state = 'DELEGATING'
return { action: 'delegate', tasks: ['task-1'] }
@@ -105,10 +131,25 @@ export class MainAgent {
}
try {
const messages = [
{ role: 'system', content: 'You are AirCoding, an AI coding assistant. Help the user with their coding tasks. Be concise and helpful.' },
{ role: 'user', content: user_message }
]
const assembled = this.context_assembler?.assemble({
session_id: this.config.session_id,
project_id: this.config.project_id,
project_root: this.project_root,
agent_id: this.agent_id,
agent_type: 'executor',
task_id: this.task_id,
token_budget: 200000,
})
const messages = assembled?.messages?.length
? [
...assembled.messages,
{ role: 'user', content: user_message }
]
: [
{ role: 'system', content: 'You are AirCoding, an AI coding assistant. Help the user with their coding tasks. Be concise and helpful.' },
{ role: 'user', content: user_message }
]
const result = await this.provider_manager.complete_text(messages, {
model: this.classify_model,
@@ -182,9 +223,9 @@ export class MainAgent {
].join('\n')
try {
const result = await this.provider_manager.complete(
const result = await this.provider_manager.complete_text(
[{ role: 'user', content: classification_prompt }],
{ model: this.classify_model }
{ model: this.classify_model, max_tokens: 32 }
)
const parsed = String(result.content || '').trim().toLowerCase()
if (parsed === 'simple_question' || parsed === 'implementation_request' || parsed === 'direct_command') {
@@ -198,6 +239,14 @@ export class MainAgent {
}
}
private infer_changed_files(message: string): string[] {
const files = Array.from(message.matchAll(/[\w./-]+\.(?:ts|tsx|js|jsx|json|md|cpp|c|h|hpp|cmake|txt|yaml|yml)/g)).map(m => m[0])
if (/contract|协议|契约/i.test(message)) files.push('packages/contracts/src/index.ts')
if (/runtime|调度|scheduler|worker/i.test(message)) files.push('packages/runtime/src/index.ts')
if (/tui|hud|界面/i.test(message)) files.push('packages/tui/src/index.ts')
return [...new Set(files)]
}
/**
* Handle confirmation from user.
*/

11
packages/runtime/src/app/RuntimeApp.ts
View File

@@ -24,6 +24,9 @@ import { EventBus } from '../events/EventBus.js'
import { EventStore, eventStore } from '../events/EventStore.js'
import { EventIngestorImpl } from '../events/EventIngestor.js'
import { TaskRepository } from '../storage/repositories/TaskRepository.js'
import { MessageRepository } from '../storage/repositories/MessageRepository.js'
import { EvidenceRepository } from '../storage/repositories/EvidenceRepository.js'
import { createCapabilityRegistry, type CapabilityRegistry } from '../capabilities/CapabilityRegistry.js'
export interface RuntimeAppConfig {
project_root: string
@@ -43,6 +46,7 @@ export class RuntimeApp {
logger: Logger
db: DatabaseManager
tool_registry: ToolRegistry
capability_registry: CapabilityRegistry
get session_id(): SessionID { return this.config.session_id }
get project_id(): ProjectID { return this.config.project_id }
@@ -64,9 +68,11 @@ export class RuntimeApp {
// Core services
this.tool_registry = createToolRegistry(config.project_root)
this.capability_registry = createCapabilityRegistry()
this.capability_registry.set_tool_registry(this.tool_registry)
this.worker_manager = new WorkerManager(this.tool_registry)
this.context_assembler = new ContextAssembler()
this.doctor = new DoctorService(config.project_root)
this.doctor = new DoctorService(config.project_root, this.capability_registry)
this.projection_store = new ProjectionStore()
this.projection_client = new ProjectionClient()
this.event_bus = new EventBus()
@@ -145,6 +151,9 @@ export class RuntimeApp {
const raw_db = this.db.getRawDatabase()
if (raw_db) {
const task_repo = new TaskRepository(raw_db as any)
const message_repo = new MessageRepository(raw_db as any)
const evidence_repo = new EvidenceRepository(raw_db as any)
this.context_assembler.set_data_sources({ message_repo, evidence_store: evidence_repo })
this.scheduler.set_task_repo(task_repo)
const rehydrated = await this.scheduler.rebuild_from_db()
this.logger.info('Scheduler recovery complete', { rehydrated })

13
packages/runtime/src/app/ServiceRegistry.ts
View File

@@ -14,6 +14,7 @@ import { ProjectionStore } from '../projection/ProjectionStore.js'
import { Logger } from '../logging/Logger.js'
import { Scheduler } from '../scheduler/Scheduler.js'
import { WorkerManager } from '../workers/WorkerManager.js'
import { createCapabilityRegistry, type CapabilityRegistry } from '../capabilities/CapabilityRegistry.js'
export interface ServiceGraph {
database: DatabaseManager
@@ -21,6 +22,7 @@ export interface ServiceGraph {
tool_registry: ToolRegistry
context_assembler: ContextAssembler
doctor: DoctorService
capability_registry: CapabilityRegistry
projection_store: ProjectionStore
logger: Logger
scheduler: Scheduler | null
@@ -38,11 +40,13 @@ export class ServiceRegistry {
const database = new DatabaseManager(`${project_root}/.air/sessions/${session_id}.db`)
const permission_engine = new PermissionEngine(project_root)
const tool_registry = new ToolRegistry(project_root)
const capability_registry = createCapabilityRegistry()
capability_registry.set_tool_registry(tool_registry)
const context_assembler = new ContextAssembler()
const doctor = new DoctorService(project_root)
const doctor = new DoctorService(project_root, capability_registry)
const projection_store = new ProjectionStore()
const worker_manager = new WorkerManager()
const scheduler = new Scheduler({ session_id, project_id, project_root })
const worker_manager = new WorkerManager(tool_registry)
const scheduler = new Scheduler({ session_id, project_id, project_root }, worker_manager)
// Register all services
this.services.set('database', database)
@@ -50,6 +54,7 @@ export class ServiceRegistry {
this.services.set('tool_registry', tool_registry)
this.services.set('context_assembler', context_assembler)
this.services.set('doctor', doctor)
this.services.set('capability_registry', capability_registry)
this.services.set('projection_store', projection_store)
this.services.set('logger', logger)
this.services.set('scheduler', scheduler)
@@ -57,7 +62,7 @@ export class ServiceRegistry {
return {
database, permission_engine, tool_registry,
context_assembler, doctor, projection_store,
context_assembler, doctor, capability_registry, projection_store,
logger, scheduler, worker_manager
}
}

49
packages/runtime/src/context/ContextAssembler.ts
View File

@@ -11,6 +11,8 @@ import type {
SessionID, ProjectID, AgentID, TaskID, ArtifactID, ISOTimeString
} from '@aircoding/contracts'
import { readdirSync, statSync } from 'fs'
import { join, relative } from 'path'
import { PromptLayerLoader, createPromptLayerLoader } from './PromptLayerLoader.js'
import { CompactionPolicy, createCompactionPolicy } from './CompactionPolicy.js'
@@ -104,6 +106,32 @@ export class ContextAssembler {
}
}
private build_project_files_snapshot(project_root: string): string {
const files: string[] = []
const ignored = new Set(['.air', '.git', 'node_modules', 'build', 'dist', '.claude'])
const walk = (dir: string, depth: number) => {
if (depth > 3 || files.length >= 200) return
let entries: string[] = []
try {
entries = readdirSync(dir)
} catch {
return
}
for (const entry of entries) {
if (ignored.has(entry)) continue
const full = join(dir, entry)
try {
const stat = statSync(full)
if (stat.isDirectory()) walk(full, depth + 1)
else files.push(relative(project_root, full))
} catch {}
if (files.length >= 200) return
}
}
walk(project_root, 0)
return ['# Project Files Snapshot (L3)', ...files.map(f => `- ${f}`)].join('\n')
}
/**
* Collect all layers in order L0-L9.
*/
@@ -132,6 +160,13 @@ export class ContextAssembler {
project_root: context.project_root
})
layers.push(...project_rules)
layers.push({
level: 'project_files' as any,
priority: 3,
content: this.build_project_files_snapshot(context.project_root),
token_estimate: 300,
source_ref: `project:${context.project_id}:files`
})
// L4: Architecture (if available)
if (context.additional_layers) {
@@ -162,7 +197,7 @@ export class ContextAssembler {
let evidence_content = ''
if (this.evidence_store && context.task_id) {
try {
const records = this.evidence_store.list_for_entity?.(context.task_id) || []
const records = this.evidence_store.list_for_entity?.('task_id', context.task_id) || []
if (records.length > 0) {
evidence_content = records.map((r: any) =>
`- [${r.type || 'evidence'}] ${r.summary || r.id}`).join('\n')
@@ -218,9 +253,9 @@ export class ContextAssembler {
if (this.message_repo) {
try {
const msgs = this.message_repo.list_by_session?.(context.session_id) || []
const tool_msgs = msgs.filter((m: any) => m.role === 'tool_result' || m.role === 'tool_use').slice(-10)
const tool_msgs = msgs.filter((m: any) => m.role === 'tool' || m.role === 'tool_result' || m.role === 'tool_use').slice(-10)
tool_content = tool_msgs.map((m: any) =>
`[${m.role}]: ${String(m.content_json || '').slice(0, 300)}`).join('\n')
`[${m.role}]: ${String(m.content_json || m.content || '').slice(0, 300)}`).join('\n')
} catch { /* fall through */ }
}
layers.push({
@@ -248,12 +283,6 @@ export class ContextAssembler {
})
}
// Add any additional layers
if (context.additional_layers) {
const others = context.additional_layers.filter(l => l.level !== 'architecture')
layers.push(...others)
}
return layers
}
@@ -265,7 +294,7 @@ export class ContextAssembler {
// System message: L0 + L1 + L2 + L3
const system_content = layers
.filter(l => ['runtime_invariant', 'role', 'safety', 'project_rules'].includes(l.level))
.filter(l => ['runtime_invariant', 'role', 'safety', 'project_rules', 'project_files'].includes(l.level))
.map(l => l.content)
.join('\n\n---\n\n')

80
packages/runtime/src/scheduler/Scheduler.ts
View File

@@ -248,13 +248,79 @@ export class Scheduler {
}
}
// Workers complete → mark running tasks as completed
if (this.worker_manager && !this.worker_manager.has_running()) {
// 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')
// Workers complete → consume explicit WorkerResult status
if (this.worker_manager) {
const running_tasks = this.graph.get_tasks_by_status('running')
for (const task of running_tasks) {
const handle = this.worker_manager.get_handle_for_task(task.id)
const result = this.worker_manager.get_result_for_task(task.id)
if (!handle || !result) continue
const attempt_id = `${task.id}_1`
if (result.status === 'completed') {
await eventIngestor.ingest({
id: `evt_${task.id}_completed`,
type: 'task.completed',
version: 1,
session_id: this.context.session_id,
project_id: this.context.project_id,
timestamp: new Date().toISOString(),
source: { kind: 'scheduler' },
route: ['scheduler', 'monitoring'],
payload: {
task_id: task.id,
agent_id: handle.worker_id,
attempt_id,
worker_result_json: result,
summary: result.summary,
changed_files: result.changed_files,
evidence_refs: result.evidence_refs,
}
})
this.graph.update_status(task.id, 'completed')
this.agent_monitor.remove(handle.worker_id)
} else if (result.status === 'blocked') {
await eventIngestor.ingest({
id: `evt_${task.id}_blocked`,
type: 'task.blocked',
version: 1,
session_id: this.context.session_id,
project_id: this.context.project_id,
timestamp: new Date().toISOString(),
source: { kind: 'scheduler' },
route: ['scheduler', 'monitoring'],
payload: { task_id: task.id, agent_id: handle.worker_id, reason: result.summary, blocker_kind: 'worker_blocked', evidence_refs: result.evidence_refs, suggested_next_step: 'Review worker blocker and retry with corrected plan' }
})
this.graph.update_status(task.id, 'blocked')
this.agent_monitor.remove(handle.worker_id)
} else if (result.status === 'cancelled') {
await eventIngestor.ingest({
id: `evt_${task.id}_cancelled_result`,
type: 'task.cancelled',
version: 1,
session_id: this.context.session_id,
project_id: this.context.project_id,
timestamp: new Date().toISOString(),
source: { kind: 'scheduler' },
route: ['scheduler', 'monitoring'],
payload: { task_id: task.id, reason: result.summary, cancelled_by: handle.worker_id }
})
this.graph.update_status(task.id, 'cancelled')
this.agent_monitor.remove(handle.worker_id)
} else {
await eventIngestor.ingest({
id: `evt_${task.id}_failed_result`,
type: 'task.failed',
version: 1,
session_id: this.context.session_id,
project_id: this.context.project_id,
timestamp: new Date().toISOString(),
source: { kind: 'scheduler' },
route: ['scheduler', 'monitoring'],
payload: { task_id: task.id, agent_id: handle.worker_id, attempt_id, error: { message: result.summary }, evidence_refs: result.evidence_refs, metadata: { worker_status: result.status } }
})
this.graph.update_status(task.id, 'failed')
this.agent_monitor.remove(handle.worker_id)
}
}
}

38
packages/runtime/src/tools/BuiltInToolRegistrar.ts
View File

@@ -178,7 +178,7 @@ export class BuiltInToolRegistrar {
const { path } = call.arguments as { path: string }
const s = statSync(resolve(project_root, path))
return { status: "ok", call_id: call.call_id, tool_name: 'fs.stat', type: 'text',
content: { path, exists: true, size: s.size, is_dir: s.isDirectory(), is_file: s.isFile(),
output: { path, exists: true, size: s.size, is_dir: s.isDirectory(), is_file: s.isFile(),
mode: s.mode, mtime: s.mtime.toISOString(), ctime: s.ctime.toISOString() },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name: 'fs.stat', type: 'ok' } }
} catch (e: any) {
@@ -192,7 +192,7 @@ export class BuiltInToolRegistrar {
const { pid, signal = 'SIGTERM' } = call.arguments as { pid: number; signal?: string }
process.kill(pid, signal as NodeJS.Signals)
return { status: "ok", call_id: call.call_id, tool_name: 'process.kill', type: 'text',
content: { pid, signal, killed: true },
output: { pid, signal, killed: true },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name: 'process.kill', type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: 'process.kill' },
@@ -204,7 +204,7 @@ export class BuiltInToolRegistrar {
try {
const { path, base_ref = 'HEAD' } = call.arguments as { path: string; base_ref?: string }
execFileSync('git', ['worktree', 'add', path, base_ref], { cwd: project_root, stdio: 'pipe', encoding: 'utf-8' })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', content: { path, base_ref, created: true },
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', output: { path, base_ref, created: true },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -216,7 +216,7 @@ export class BuiltInToolRegistrar {
try {
const { workspace_id } = call.arguments as { workspace_id: string; strategy?: string }
execFileSync('git', ['merge', '--no-ff', workspace_id], { cwd: project_root, stdio: 'pipe', encoding: 'utf-8' })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', content: { workspace_id, merged: true },
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', output: { workspace_id, merged: true },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -235,7 +235,7 @@ export class BuiltInToolRegistrar {
by_ext[ext] = (by_ext[ext] || 0) + 1
}
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { root: dir, total_files: entries.length, extensions: by_ext },
output: { root: dir, total_files: entries.length, extensions: by_ext },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -249,7 +249,7 @@ export class BuiltInToolRegistrar {
const dir = join(project_root, '.air', 'shared', 'profiles')
if (!existsSync(dir)) mkdirSync(dir, { recursive: true })
writeFileSync(join(dir, `${language}.json`), JSON.stringify(profile_json, null, 2))
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', content: { language, written: true },
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', output: { language, written: true },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -263,7 +263,7 @@ export class BuiltInToolRegistrar {
const cmake = existsSync(join(root, 'CMakeLists.txt'))
const makefile = existsSync(join(root, 'Makefile'))
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { has_cmake: cmake, has_makefile: makefile, build_system: cmake ? 'cmake' : makefile ? 'make' : 'none' },
output: { has_cmake: cmake, has_makefile: makefile, build_system: cmake ? 'cmake' : makefile ? 'make' : 'none' },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -277,7 +277,7 @@ export class BuiltInToolRegistrar {
const buildDir = join(project_root, 'build')
if (!existsSync(buildDir)) mkdirSync(buildDir, { recursive: true })
execFileSync('cmake', ['-G', generator, '-DCMAKE_BUILD_TYPE=' + build_type, '..'], { cwd: buildDir, stdio: 'pipe' })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', content: { generator, build_type, configured: true },
return { status: "ok", call_id: call.call_id, tool_name, type: 'text', output: { generator, build_type, configured: true },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -291,7 +291,7 @@ export class BuiltInToolRegistrar {
const args = target ? ['--build', '.', '--config', config, '--target', target] : ['--build', '.', '--config', config]
const out = execFileSync('cmake', args, { cwd: join(project_root, 'build'), stdio: 'pipe', timeout: 300000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { built: true, output: out.toString().slice(-500) },
output: { built: true, output: out.toString().slice(-500) },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -305,7 +305,7 @@ export class BuiltInToolRegistrar {
const args = filter ? ['--output-on-failure', '-R', filter] : ['--output-on-failure']
const out = execFileSync('ctest', args, { cwd: join(project_root, 'build'), stdio: 'pipe', timeout: 300000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { passed: true, output: out.toString().slice(-1000) },
output: { passed: true, output: out.toString().slice(-1000) },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -318,7 +318,7 @@ export class BuiltInToolRegistrar {
const { path = 'src' } = (call.arguments || {}) as any
const out = execFileSync('cppcheck', ['--enable=all', '--quiet', path], { cwd: project_root, stdio: 'pipe', encoding: 'utf-8', timeout: 120000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { output: out.toString().slice(-500), issues_found: 0 },
output: { output: out.toString().slice(-500), issues_found: 0 },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -331,7 +331,7 @@ export class BuiltInToolRegistrar {
const { file, line = 0, column = 0 } = (call.arguments || {}) as any
const out = execFileSync('clangd', ['--check=' + file], { cwd: project_root, stdio: 'pipe', encoding: 'utf-8', timeout: 30000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { file, line, column, diagnostics: out.toString().slice(-1000) },
output: { file, line, column, diagnostics: out.toString().slice(-1000) },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -344,7 +344,7 @@ export class BuiltInToolRegistrar {
const { target } = (call.arguments || {}) as any
const out = execFileSync('gdb', ['-batch', '-ex', 'run', '-ex', 'bt', '--', target], { cwd: project_root, stdio: 'pipe', encoding: 'utf-8', timeout: 60000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { target, backtrace: out.toString().slice(-2000) },
output: { target, backtrace: out.toString().slice(-2000) },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -358,7 +358,7 @@ export class BuiltInToolRegistrar {
const content = readFileSync(resolve(project_root, log_path), 'utf-8')
const errors = content.split('\n').filter(l => /error|fail|segfault|assert|abort|exception/i.test(l)).slice(0, 50)
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { log_path, error_count: errors.length, errors },
output: { log_path, error_count: errors.length, errors },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -373,7 +373,7 @@ export class BuiltInToolRegistrar {
const tmpFile = join(tmpDir, `screenshot-${Date.now()}.png`)
execFileSync('import', ['-window', 'root', tmpFile], { stdio: 'pipe', timeout: 10000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { captured: true, path: tmpFile },
output: { captured: true, path: tmpFile },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: `Screenshot not available: ${e.message}`, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -388,7 +388,7 @@ export class BuiltInToolRegistrar {
if (filter) args.push(filter)
const out = execFileSync('tcpdump', args, { stdio: 'pipe', timeout: (duration_sec + 5) * 1000 })
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { interface: iface, duration_sec, packets: (String(out) || '').split('\n').length },
output: { interface: iface, duration_sec, packets: (String(out) || '').split('\n').length },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: `Capture not available: ${e.message}`, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -399,7 +399,7 @@ export class BuiltInToolRegistrar {
'permission.request': async (call: any) => {
const { tool_name: tn, reason } = call.arguments as { tool_name: string; reason: string }
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { tool_name: tn, reason, status: 'allowed', message: `Permission granted for ${tn}: ${reason}` },
output: { tool_name: tn, reason, status: 'allowed', message: `Permission granted for ${tn}: ${reason}` },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
},
@@ -416,7 +416,7 @@ export class BuiltInToolRegistrar {
checks.push({ name: 'project_structure', passed: hasPkg, message: hasPkg ? 'Valid' : 'No package.json' })
const allPassed = checks.every(c => c.passed)
return { status: "ok", call_id: call.call_id, tool_name, type: 'text',
content: { checks, all_passed: allPassed, fixable_count: checks.filter(c => !c.passed).length },
output: { checks, all_passed: allPassed, fixable_count: checks.filter(c => !c.passed).length },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' } }
} catch (e: any) {
return { status: 'error', error: { error_id: call.call_id, kind: 'tool_error', severity: 'error', message: e.message, retryability: 'not_retryable', semantic_signature: tool_name },
@@ -432,7 +432,7 @@ export class BuiltInToolRegistrar {
call_id: call.call_id,
tool_name,
type: 'text',
content: { message: `Tool ${tool_name} not yet implemented` },
output: { message: `Tool ${tool_name} not yet implemented` },
metadata: { timestamp: new Date().toISOString(), call_id: call.call_id, tool_name, type: 'ok' }
})
}

72
packages/runtime/src/tools/ToolRegistry.ts
View File

@@ -12,8 +12,13 @@ import type { ToolDefinition, ToolCall, ToolResultEnvelope, ISOTimeString } from
import { PermissionEngine, createPermissionEngine, type PermissionContext, type PermissionDecision, type PermissionAction } from '../security/PermissionEngine.js'
import type { AgentType } from '@aircoding/contracts'
export type ToolExecutionReturn =
| ToolResultEnvelope
| Promise<ToolResultEnvelope>
| AsyncIterable<ToolResultEnvelope>
export interface ToolExecutor {
(call: ToolCall, context: ToolExecutionContext): Promise<ToolResultEnvelope>
(call: ToolCall, context: ToolExecutionContext): ToolExecutionReturn
}
export interface ToolExecutionContext {
@@ -140,28 +145,19 @@ export class ToolRegistry {
const permission_context = this.build_permission_context(call, context)
const decision = await this.permission_engine.evaluate(call, permission_context, definition)
if (decision.action !== 'allow') {
if (decision.action !== 'allow' && decision.action !== 'announce_then_run') {
yield create_error_result(call.call_id, 'permission_denied', decision.reason)
return
}
// Execute with streaming support
// The executor yields intermediate results, final result comes at end
let final_result: ToolResultEnvelope | undefined
let saw_final = false
for await (const chunk of this.execute_streaming(call, context, executor)) {
// Streaming signal: final result is the one with status='ok' whose metadata marks final
if (chunk.metadata && (chunk.metadata as any).is_final === true) {
final_result = chunk
} else {
yield chunk
}
if (chunk.metadata && (chunk.metadata as any).is_final === true) saw_final = true
yield chunk
}
// Yield final result exactly once
if (final_result) {
yield final_result
} else {
if (!saw_final) {
yield create_error_result(call.call_id, 'no_final_result', 'Streaming tool did not produce final result')
}
}
@@ -251,7 +247,7 @@ export class ToolRegistry {
if (!executor) {
return create_error_result(call.call_id, 'executor_not_found', 'Executor not registered')
}
return executor(call, ctx)
return this.execute_executor_final(executor, call, ctx)
}
case 'announce_then_run': {
@@ -260,7 +256,7 @@ export class ToolRegistry {
if (!executor) {
return create_error_result(call.call_id, 'executor_not_found', 'Executor not registered')
}
const result = await executor(call, ctx)
const result = await this.execute_executor_final(executor, call, ctx)
return {
...result,
metadata: { ...result.metadata, announced: true },
@@ -269,10 +265,10 @@ export class ToolRegistry {
case 'ask_user':
// Suspend; emit permission.prompt.requested
return create_error_result('', 'user_prompt_required', 'User confirmation required')
return create_error_result(call.call_id, 'user_prompt_required', 'User confirmation required')
case 'deny':
return create_error_result('', 'permission_denied', decision.reason)
return create_error_result(call.call_id, 'permission_denied', decision.reason)
case 'block': {
// Return blocked outcome → task.blocked upstream
@@ -289,6 +285,34 @@ export class ToolRegistry {
}
}
/**
* Execute a tool and return the final envelope.
* Streaming executors are consumed until their final result.
*/
private async execute_executor_final(
executor: ToolExecutor,
call: ToolCall,
context: ToolExecutionContext,
): Promise<ToolResultEnvelope> {
const result = executor(call, context)
if (this.is_async_iterable(result)) {
let final_result: ToolResultEnvelope | undefined
let last_chunk: ToolResultEnvelope | undefined
for await (const chunk of result) {
last_chunk = chunk
if (chunk.metadata && (chunk.metadata as any).is_final === true) {
final_result = chunk
}
}
return final_result || last_chunk || create_error_result(call.call_id, 'no_result', 'Tool produced no result')
}
return await result
}
private is_async_iterable(value: unknown): value is AsyncIterable<ToolResultEnvelope> {
return Boolean(value && typeof (value as any)[Symbol.asyncIterator] === 'function')
}
/**
* Execute streaming tool.
*/
@@ -297,10 +321,12 @@ export class ToolRegistry {
context: ToolExecutionContext,
executor: ToolExecutor
): AsyncGenerator<ToolResultEnvelope> {
// Tool-specific execution handler
// For now, just execute normally
const result = await executor(call, context)
yield result
const result = executor(call, context)
if (this.is_async_iterable(result)) {
for await (const chunk of result) yield chunk
return
}
yield await result
}
}

78
packages/runtime/src/tools/shell/index.ts
View File

@@ -43,14 +43,12 @@ export function createShellExecutor(project_root: string) {
const cwd = workdir || project_root
const timestamp = new Date().toISOString() as ISOTimeString
// Emit command.started event
yield {
status: 'ok',
output: { event: 'command.started', command, cwd },
metadata: { timestamp, streaming: true, call_id: call.call_id, tool_name: 'shell.run' }
}
// Execute command
const proc = spawn(command, [], {
cwd,
shell: true,
@@ -59,53 +57,81 @@ export function createShellExecutor(project_root: string) {
let stdout = ''
let stderr = ''
let final_code = 0
let timed_out = false
const chunks: ToolResultEnvelope[] = []
// Stream stdout
proc.stdout.on('data', (data) => {
const text = data.toString()
stdout += text
// Emit streaming stdout
// Note: In actual implementation, this would go through EventBus
chunks.push({
status: 'ok',
output: { event: 'command.stdout', text },
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: true, call_id: call.call_id, tool_name: 'shell.run' }
})
})
// Stream stderr
proc.stderr.on('data', (data) => {
const text = data.toString()
stderr += text
chunks.push({
status: 'ok',
output: { event: 'command.stderr', text },
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: true, call_id: call.call_id, tool_name: 'shell.run' }
})
})
// Wait for completion or timeout
let timed_out = false
const timeoutPromise = new Promise<number>((resolve) => {
setTimeout(() => {
timed_out = true
proc.kill('SIGKILL')
resolve(124) // standard timeout exit code
}, timeout)
const timeout_id = setTimeout(() => {
timed_out = true
proc.kill('SIGKILL')
}, timeout)
const exit_code = await new Promise<number>((resolve) => {
proc.on('exit', (code) => resolve(code ?? 0))
proc.on('error', () => resolve(1))
})
clearTimeout(timeout_id)
const exitCode = await Promise.race([
new Promise<number>((resolve) => proc.on('exit', (code) => resolve(code || 0))),
timeoutPromise
])
if (stdout) {
yield {
status: 'ok',
output: { event: 'command.stdout', text: stdout.slice(-50000) },
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: true, call_id: call.call_id, tool_name: 'shell.run' }
}
}
if (stderr) {
yield {
status: 'ok',
output: { event: 'command.stderr', text: stderr.slice(-10000) },
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: true, call_id: call.call_id, tool_name: 'shell.run' }
}
}
while (chunks.length > 0) {
yield chunks.shift()!
}
final_code = exitCode
if (timed_out) {
stderr += `\n[Command timed out after ${timeout}ms]`
}
// Emit command.completed event
yield {
status: final_code === 0 ? 'ok' : 'error',
status: exit_code === 0 ? 'ok' : 'error',
output: {
event: 'command.completed',
exit_code: final_code,
stdout: stdout.slice(-50000), // Last 50KB
stderr: stderr.slice(-10000), // Last 10KB
exit_code,
stdout: stdout.slice(-50000),
stderr: stderr.slice(-10000),
timed_out
},
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: false }
error: exit_code === 0 ? undefined : {
error_id: call.call_id,
kind: 'tool_error',
severity: 'error',
message: timed_out ? `Command timed out after ${timeout}ms` : `Command exited with code ${exit_code}`,
retryability: timed_out ? 'retryable' : 'not_retryable',
semantic_signature: 'shell.run'
},
metadata: { timestamp: new Date().toISOString() as ISOTimeString, streaming: false, is_final: true, call_id: call.call_id, tool_name: 'shell.run' }
}
}
}

82
packages/runtime/src/workers/WorkerManager.ts
View File

@@ -31,7 +31,7 @@ export interface WorkerHandle {
worker_id: string
process: WorkerProcess
config: WorkerConfig
state: 'starting' | 'ready' | 'running' | 'completed' | 'error' | 'cancelled'
state: 'starting' | 'ready' | 'running' | 'completed' | 'failed' | 'error' | 'cancelled'
started_at: string
completed_at?: string
result?: WorkerResult<unknown>
@@ -107,6 +107,8 @@ export class WorkerManager {
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)
@@ -179,10 +181,10 @@ export class WorkerManager {
{
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,
permission_template: 'executor',
cwd: this.execution_context?.project_root || process.cwd()
} as any
agent_type: 'executor',
}
)
this.send_to_worker(agent_id, 'tool.result', {
@@ -238,8 +240,11 @@ export class WorkerManager {
proc.on_message('worker.result', (msg) => {
const handle = this.workers.get(agent_id)
if (handle) {
handle.state = 'completed'
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()
}
})
@@ -313,29 +318,84 @@ export class WorkerManager {
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)?.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)
}
// ============================================================================
// Private
// ============================================================================
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 === 'pass'
? 'completed'
: '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.error ? String(payload.error) : '')
|| (changed_files.length > 0 ? `Changed files: ${changed_files.join(', ')}` : `Worker ${status}`)
return {
task_id: (payload.task_id as string) || '' as any,
task_id: (payload.task_id as string) || (handle.config.task_spec?.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) || 'executor',
status: (payload.status as WorkerStatus) || 'completed',
summary: (payload.summary as string) || '',
changed_files: (payload.changed_files as string[]) || [],
status: status as WorkerStatus,
summary,
changed_files,
diff_ref: (payload.diff_ref as string | undefined) || undefined,
artifacts: (payload.artifacts as any[]) || [],
verification: (payload.verification 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) || null,
result: (payload.result as unknown) || payload,
}
}

17
packages/runtime/src/workers/WorkerProcess.ts
View File

@@ -35,6 +35,7 @@ export class WorkerProcess {
private proc: ChildProcess | null = null
private protocol: WorkerProtocol
private message_handlers: Map<string, (msg: WorkerMessage) => void> = new Map()
private exit_handlers: Array<(info: { code: number | null; signal: NodeJS.Signals | null; semantic: string; description: string }) => void> = []
private buffer: string = ''
constructor() {
@@ -68,6 +69,13 @@ export class WorkerProcess {
this.message_handlers.set(type, handler)
}
/**
* Register an exit handler.
*/
on_exit(handler: (info: { code: number | null; signal: NodeJS.Signals | null; semantic: string; description: string }) => void): void {
this.exit_handlers.push(handler)
}
/**
* Get exit code info.
*/
@@ -124,8 +132,13 @@ export class WorkerProcess {
// Exit handler
this.proc.on('exit', (code, signal) => {
const info = this.get_exit_code_info(code || 1)
console.log(`[Worker] exited with code ${code} (${info?.semantic || 'unknown'}): ${info?.description || ''}`)
const info = this.get_exit_code_info(code ?? 1)
const semantic = info?.semantic || 'unknown'
const description = info?.description || ''
console.log(`[Worker] exited with code ${code} (${semantic}): ${description}`)
for (const handler of this.exit_handlers) {
handler({ code, signal: signal as NodeJS.Signals | null, semantic, description })
}
})
}