diff --git a/packages/llm/src/ProviderManager.ts b/packages/llm/src/ProviderManager.ts index f6bbae9..633e6fa 100755 --- a/packages/llm/src/ProviderManager.ts +++ b/packages/llm/src/ProviderManager.ts @@ -7,13 +7,11 @@ * @module packages/llm/src/ProviderManager */ -import type { ProviderAdapter } from '@aircoding/contracts' +import type { ProviderAdapter, ProviderCompletionInput, ProviderStreamEvent, ProviderCapabilityMatrix, ModelID, ProviderID } from '@aircoding/contracts' -// Local type definitions (contract types not yet finalized) -type CompleteOptions = { max_tokens?: number; temperature?: number; top_p?: number; system?: string } type StreamEvent = { type: 'text' | 'thinking' | 'done'; content?: string; reason?: string } type ModelRequirement = { model: string; provider?: string; min_output_tokens?: number; prefers_thinking?: boolean; requires_tools?: boolean } -type ModelAssignment = { provider: string; model: string; adapter: any; capabilities?: any } +type ModelAssignment = { provider: string; model: string; adapter: ProviderAdapter; capabilities?: ReturnType } import { ModelConfigLoader, createModelConfigLoader } from './ModelConfigLoader.js' import { CapabilityMatrixRegistry, createCapabilityMatrixRegistry } from './CapabilityMatrix.js' @@ -80,34 +78,118 @@ export class ProviderManager { /** * Complete a request with the current model. + * Returns AsyncIterable of ProviderStreamEvent per contracts §15. */ - async complete( + async *complete( + input: ProviderCompletionInput + ): AsyncGenerator { + const adapter = this.current_adapter || this.adapters.get(input.provider_id || 'anthropic') + if (!adapter) { + throw new Error('No adapter selected. Call select_model first.') + } + + yield* adapter.complete(input) + } + + /** + * Complete and collect all events into a single response. + */ + async complete_text( + messages: unknown[], + options: { model?: string; max_tokens?: number; temperature?: number; system?: string } = {} + ): Promise<{ content: string; usage?: { input_tokens: number; output_tokens: number } }> { + const adapter = this.current_adapter + if (!adapter) { + throw new Error('No adapter selected. Call select_model first.') + } + + const model_id = options.model || 'claude-haiku-4-5-20251001' + const input: ProviderCompletionInput = { + provider_id: 'anthropic', + model_id: model_id as ModelID, + canonical_format: 'anthropic', + messages, + max_output_tokens: options.max_tokens || 4096, + temperature: options.temperature, + system: options.system + } + + let content = '' + let usage: { input_tokens: number; output_tokens: number } | undefined + + for await (const event of adapter.complete(input)) { + if (event.type === 'content_delta') { + const payload = event.payload as { type: string; text?: string; thinking?: string } + if (payload.type === 'text_delta') { + content += payload.text || '' + } else if (payload.type === 'thinking_delta') { + // Accumulate thinking for reference but don't include in content + } + } else if (event.type === 'message_stop') { + const payload = event.payload as { usage?: { output_tokens: number } } + if (payload.usage) { + usage = { input_tokens: 0, output_tokens: payload.usage.output_tokens } + } + } + } + + return { content, usage } + } + + /** + * Stream a completion request (passthrough to adapter). + */ + async *stream_complete( + input: ProviderCompletionInput + ): AsyncGenerator { + const adapter = this.current_adapter + if (!adapter) { + throw new Error('No adapter selected. Call select_model first.') + } + + yield* adapter.complete(input) + } + + /** + * Legacy overload: select model then complete. + */ + async complete_after_select( messages: unknown[], assignment: ModelAssignment, - options: CompleteOptions = {} + options: { max_tokens?: number; temperature?: number } = {} ): Promise<{ content: string; usage?: { input_tokens: number; output_tokens: number } }> { const adapter = assignment.adapter || this.current_adapter if (!adapter) { throw new Error('No adapter selected. Call select_model first.') } - return adapter.complete(messages as any, { model: assignment.model }, options) - } - - /** - * Stream a completion request. - */ - async *stream_complete( - messages: unknown[], - assignment: ModelAssignment, - options: CompleteOptions = {} - ): AsyncGenerator { - const adapter = assignment.adapter || this.current_adapter - if (!adapter) { - throw new Error('No adapter selected. Call select_model first.') + const input: ProviderCompletionInput = { + provider_id: assignment.provider as ProviderID || 'anthropic', + model_id: assignment.model as ModelID, + canonical_format: 'anthropic', + messages, + max_output_tokens: options.max_tokens || 4096, + temperature: options.temperature } - yield* adapter.stream_complete(messages as any, { model: assignment.model }, options) + let content = '' + let usage: { input_tokens: number; output_tokens: number } | undefined + + for await (const event of adapter.complete(input)) { + if (event.type === 'content_delta') { + const payload = event.payload as { type: string; text?: string } + if (payload.type === 'text_delta') { + content += payload.text || '' + } + } else if (event.type === 'message_stop') { + const payload = event.payload as { usage?: { output_tokens: number } } + if (payload.usage) { + usage = { input_tokens: 0, output_tokens: payload.usage.output_tokens } + } + } + } + + return { content, usage } } /** diff --git a/packages/runtime/src/workers/WorkerManager.ts b/packages/runtime/src/workers/WorkerManager.ts index 38442cb..b676b96 100755 --- a/packages/runtime/src/workers/WorkerManager.ts +++ b/packages/runtime/src/workers/WorkerManager.ts @@ -13,6 +13,8 @@ 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 type { ToolRegistry } from '../tools/ToolRegistry.js' +import type { ProviderManager } from '@aircoding/llm' export interface WorkerConfig { entrypoint: string // Path to worker main.ts @@ -36,9 +38,36 @@ export interface WorkerHandle { export class WorkerManager { private protocol: WorkerProtocol private workers: Map = new Map() + private tool_registry?: ToolRegistry + private provider_manager?: ProviderManager + private execution_context?: { session_id: string; project_id: string; project_root: string } - constructor() { + 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 } /** diff --git a/packages/runtime/src/workers/WorkerProtocol.ts b/packages/runtime/src/workers/WorkerProtocol.ts index 155f7a6..3beb7c0 100755 --- a/packages/runtime/src/workers/WorkerProtocol.ts +++ b/packages/runtime/src/workers/WorkerProtocol.ts @@ -27,6 +27,7 @@ export type WorkerMessageType = | 'tool.result' | 'agent.cancel' | 'agent.ping' + | 'llm.response' // Worker → Parent | 'worker.ready' | 'tool.call' @@ -35,6 +36,7 @@ export type WorkerMessageType = | 'worker.heartbeat' | 'worker.error' | 'event' + | 'llm.request' const PROTOCOL_VERSION = 1 @@ -44,13 +46,15 @@ const DIRECTION_RULES: Record = { 'tool.result': 'parent_to_worker', 'agent.cancel': 'parent_to_worker', 'agent.ping': 'parent_to_worker', + 'llm.response': 'parent_to_worker', 'worker.ready': 'worker_to_parent', 'tool.call': 'worker_to_parent', 'worker.result': 'worker_to_parent', 'worker.checkpoint': 'worker_to_parent', 'worker.heartbeat': 'worker_to_parent', 'worker.error': 'worker_to_parent', - 'event': 'worker_to_parent' + 'event': 'worker_to_parent', + 'llm.request': 'worker_to_parent' } export class WorkerProtocol { diff --git a/packages/workers/src/WorkerRuntime.ts b/packages/workers/src/WorkerRuntime.ts index e7902fe..357915b 100755 --- a/packages/workers/src/WorkerRuntime.ts +++ b/packages/workers/src/WorkerRuntime.ts @@ -24,6 +24,20 @@ export interface ToolCallResult { content: Record } +export interface LLMRequest { + messages: Array<{ role: string; content: unknown }> + model?: string + max_tokens?: number + temperature?: number + tools?: unknown[] +} + +export interface LLMResponse { + content: string + usage?: { input_tokens: number; output_tokens: number } + tool_calls?: Array<{ name: string; arguments: Record }> +} + export class WorkerRuntime { private agent_id: string private session_id: string @@ -63,6 +77,36 @@ export class WorkerRuntime { return promise } + /** + * Call LLM through parent process via IPC. + * INV-3: This is the ONLY way workers access LLM. + */ + async call_llm(request: LLMRequest): Promise { + const call_id = crypto.randomUUID() + + const promise = new Promise((resolve, reject) => { + this.pending_calls.set(call_id, { resolve: resolve as any, reject }) + + // Set timeout (LLM calls can be long) + setTimeout(() => { + this.pending_calls.delete(call_id) + reject(new Error(`LLM call timeout: ${request.model || 'default'}`)) + }, 300000) // 5 minutes + }) + + // Send llm.request via IPC + this.send_message('llm.request', { + call_id, + messages: request.messages, + model: request.model || 'claude-haiku-4-5-20251001', + max_tokens: request.max_tokens || 4096, + temperature: request.temperature, + tools: request.tools + }) + + return promise + } + /** * Emit an event to the parent. */ @@ -118,6 +162,22 @@ export class WorkerRuntime { // Respond to ping this.send_message('worker.heartbeat', { timestamp: new Date().toISOString() }) break + + case 'llm.response': { + const call_id = payload.call_id as string + const pending = this.pending_calls.get(call_id) + if (pending) { + this.pending_calls.delete(call_id) + // Resolve with LLM response + const response: LLMResponse = { + content: (payload.content as string) || '', + usage: payload.usage as LLMResponse['usage'], + tool_calls: payload.tool_calls as LLMResponse['tool_calls'] + } + pending.resolve(response as any) + } + break + } } }