From 4de9cd0339d1fe0b8498fb0256a2f4427c15b841 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Fri, 17 Apr 2026 02:21:12 -0400 Subject: [PATCH] refactor(worker): use pi-mono Agent directly Rewrites the utility-process worker to drive pi-agent-core's Agent class against pi-ai models and pi-coding-agent tools. The entry file keeps the queue and cancellation state, delegates runs to WorkerRunner, and routes RPC responses through WorkerRpcClient so beforeToolCall/afterToolCall approvals and prepareRun credential resolution travel back to main's RuntimeBridge. Worker message schema gains `agent-event` + `rpc-request` envelopes, and run-finished no longer depends on the deleted sessionCodec. --- src/main/worker-service.ts | 3 +- .../worker/handlers/run-finished-helpers.ts | 18 +- src/main/worker/worker-entry.ts | 261 ++---------------- src/main/worker/worker-helpers.ts | 96 +++++++ src/main/worker/worker-message-handler.ts | 38 ++- src/main/worker/worker-rpc.ts | 34 +++ src/main/worker/worker-runner.ts | 192 +++++++++++++ src/main/worker/worker-types.ts | 19 ++ 8 files changed, 404 insertions(+), 257 deletions(-) create mode 100644 src/main/worker/worker-helpers.ts create mode 100644 src/main/worker/worker-rpc.ts create mode 100644 src/main/worker/worker-runner.ts diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index 2de19f16..2031bc20 100644 --- a/src/main/worker-service.ts +++ b/src/main/worker-service.ts @@ -20,6 +20,7 @@ export function createWorkerService({ notify, budgetService, sessionService, + getRuntimeBridge, }: WorkerServiceDependencies) { const MAX_RETRY_ATTEMPTS = 2; const RETRY_COOLDOWN_MS = 30_000; // 30 seconds @@ -110,7 +111,7 @@ export function createWorkerService({ await handleWorkerMessage( message, worker, - { db, logger, isQuitting, emitEvent, wakeAgentIfPossible, processRunCompletion, onRunStarted, onRunCompleted, notify, budgetService, sessionService }, + { db, logger, isQuitting, emitEvent, wakeAgentIfPossible, processRunCompletion, onRunStarted, onRunCompleted, notify, budgetService, sessionService, getRuntimeBridge }, workerMessageSchema, destroyed, MAX_RETRY_ATTEMPTS, diff --git a/src/main/worker/handlers/run-finished-helpers.ts b/src/main/worker/handlers/run-finished-helpers.ts index 76cfe4ff..4f4362c8 100644 --- a/src/main/worker/handlers/run-finished-helpers.ts +++ b/src/main/worker/handlers/run-finished-helpers.ts @@ -1,5 +1,4 @@ import type { z } from "zod"; -import { getConnectorDefinition } from "../../connectors"; import { createCircuitBreaker } from "../circuit-breaker"; import { handleFailureEscalation, handleSuccessReset } from "../failure-escalation"; import { persistRunEvent } from "../run-event-recorder"; @@ -53,8 +52,6 @@ export function finalizeRun( }, }); - const connector = getConnectorDefinition(run.connectorId); - let task: TaskRecord | null = null; try { task = db.getTask(run.taskId); @@ -63,17 +60,10 @@ export function finalizeRun( } if (task) { - if (event.sessionParams && connector.module.sessionCodec) { - const serialized = connector.module.sessionCodec.serialize(event.sessionParams); - db.updateTaskSession( - task.id, - serialized, - connector.module.sessionCodec.getDisplayId?.(serialized ?? null) ?? event.sessionDisplayId ?? null, - ); - } - if (!event.sessionParams) { - db.updateTaskSession(task.id, null, null); - } + // Pi-mono sessions are persisted through SqliteSessionStore via RuntimeBridge; + // worker-reported sessionParams is an opaque JSON snapshot of whatever the + // orchestrator wants to stamp onto the task_sessions row. + db.updateTaskSession(task.id, event.sessionParams, event.sessionDisplayId ?? null); } leaseRenewalMap.delete(event.runId); diff --git a/src/main/worker/worker-entry.ts b/src/main/worker/worker-entry.ts index 446c5fdd..766d4bbf 100644 --- a/src/main/worker/worker-entry.ts +++ b/src/main/worker/worker-entry.ts @@ -1,74 +1,6 @@ -import { runningProcesses, shutdownAllRunning } from "@agentcompany/adapter-utils/server-utils"; -import { getConnectorDefinition } from "@main/connectors"; -import { buildSanitizedRuntimeEnv } from "@main/runtime-env"; -import type { ConnectorId, RunStatus } from "@shared/types"; - -function buildRedactor(env: Record): (text: string) => string { - const sensitivePatterns: string[] = []; - for (const [key, value] of Object.entries(env)) { - if (value.length < 8) continue; - if (["PATH", "HOME", "USER", "SHELL", "LANG", "TMPDIR", "TERM"].includes(key)) continue; - sensitivePatterns.push(value); - } - if (sensitivePatterns.length === 0) return (text) => text; - sensitivePatterns.sort((a, b) => b.length - a.length); - return (text: string) => { - let result = text; - for (const pattern of sensitivePatterns) { - if (result.includes(pattern)) { - result = result.replaceAll(pattern, "[REDACTED]"); - } - } - return result; - }; -} - -interface RunJob { - runId: string; - companyId: string; - taskId: string; - taskTitle: string; - taskDescription: string; - agentId: string; - agentName: string; - connectorId: ConnectorId; - workspaceId: string | null; - workspacePath: string | null; - workspaceMode: "exclusive" | "shared" | "none"; - command: string; - model: string | null; - thinkingEffort: string | null; - env: Record; - sessionParams: Record | null; - systemPrompt: string | null; - trigger: string | null; - managedSkills: Array<{ runtimeName: string; sourcePath: string }>; -} - -type ParentMessage = - | { type: "enqueue-run"; payload: RunJob } - | { type: "cancel-run"; runId: string } - | { type: "shutdown" }; - -type ChildMessage = - | { type: "run-log"; runId: string; stream: "stdout" | "stderr"; chunk: string } - | { type: "run-status"; runId: string; status: RunStatus; message: string } - | { - type: "run-finished"; - runId: string; - status: RunStatus; - summary: string | null; - errorMessage: string | null; - exitCode: number | null; - signal: string | null; - model: string | null; - sessionParams: Record | null; - sessionDisplayId: string | null; - costUsd: number | null; - unattributedCost: boolean; - inputTokens: number | null; - outputTokens: number | null; - }; +import type { ChildMessage, ParentMessage, RunJob } from "./worker-helpers"; +import { WorkerRpcClient } from "./worker-rpc"; +import { WorkerRunner } from "./worker-runner"; const parentPort = process.parentPort; @@ -80,74 +12,23 @@ const activeByAgent = new Map(); const activeByWorkspace = new Map(); const queue: RunJob[] = []; const cancelledRuns = new Set(); -// S-01: Bypass is opt-in only. Packaged builds hard-refuse the dangerous CLI -// flags unless the operator sets BOTH `AGENTCOMPANY_ALLOW_UNSAFE_CONNECTOR_BYPASS=true` -// AND is running a dev/E2E build (AGENTCOMPANY_PACKAGED !== "true"). Main forks -// the worker with AGENTCOMPANY_PACKAGED=. -const isPackaged = process.env.AGENTCOMPANY_PACKAGED === "true"; -const allowUnsafeConnectorBypass = - !isPackaged && process.env.AGENTCOMPANY_ALLOW_UNSAFE_CONNECTOR_BYPASS === "true"; function post(message: ChildMessage) { parentPort.postMessage(message); } +const rpc = new WorkerRpcClient(post); +const runner = new WorkerRunner({ + post, + rpc, + isCancelled: (runId) => cancelledRuns.has(runId), +}); + function workspaceKey(job: RunJob) { - if (job.workspaceMode !== "exclusive") { - return null; - } + if (job.workspaceMode !== "exclusive") return null; return job.workspaceId ?? job.workspacePath ?? `task:${job.taskId}`; } -function buildConnectorExecutionConfig(job: RunJob, sanitizedEnv: Record) { - const baseConfig: Record = { - command: job.command, - cwd: job.workspacePath ?? process.cwd(), - model: job.model, - env: sanitizedEnv, - managedSkills: job.managedSkills, - }; - - switch (job.connectorId) { - case "codex_local": - return { - ...baseConfig, - modelReasoningEffort: job.thinkingEffort ?? "", - dangerouslyBypassApprovalsAndSandbox: allowUnsafeConnectorBypass, - extraArgs: [ - "--skip-git-repo-check", - "--sandbox", - allowUnsafeConnectorBypass ? "danger-full-access" : "workspace-write", - ], - }; - case "claude_local": - case "claude_compatible": - case "claude_compat_2": - case "claude_compat_3": - case "claude_compat_4": - case "claude_compat_5": - case "claude_compat_6": - case "claude_compat_7": - case "claude_compat_8": - case "claude_compat_9": - case "claude_compat_10": - return { - ...baseConfig, - effort: job.thinkingEffort ?? "", - dangerouslySkipPermissions: allowUnsafeConnectorBypass, - }; - case "gemini_local": - return { - ...baseConfig, - yolo: allowUnsafeConnectorBypass, - approvalMode: allowUnsafeConnectorBypass ? "yolo" : "suggest", - sandbox: true, - }; - default: - return baseConfig; - } -} - function enqueue(job: RunJob) { queue.push(job); post({ @@ -172,115 +53,16 @@ async function maybeStartQueuedRuns() { } async function startRun(job: RunJob) { - const definition = getConnectorDefinition(job.connectorId); const lockKey = workspaceKey(job); activeByAgent.set(job.agentId, job.runId); - if (lockKey) { - activeByWorkspace.set(lockKey, job.runId); - } - post({ - type: "run-status", - runId: job.runId, - status: "running", - message: `Running with ${definition.label}.`, - }); - - const sanitizedEnv = buildSanitizedRuntimeEnv( - Object.fromEntries( - Object.entries(process.env).filter((e): e is [string, string] => typeof e[1] === "string"), - ), - job.env, - ); - - const redact = buildRedactor(job.env); + if (lockKey) activeByWorkspace.set(lockKey, job.runId); try { - const result = await definition.module.execute({ - runId: job.runId, - agent: { - id: job.agentId, - companyId: job.companyId, - name: job.agentName, - adapterType: job.connectorId, - adapterConfig: {}, - }, - runtime: { - sessionId: - typeof job.sessionParams?.sessionId === "string" ? job.sessionParams.sessionId : null, - sessionParams: job.sessionParams, - sessionDisplayId: - typeof job.sessionParams?.sessionId === "string" ? job.sessionParams.sessionId : null, - taskKey: job.taskId, - }, - config: buildConnectorExecutionConfig(job, sanitizedEnv), - context: { - taskId: job.taskId, - taskTitle: job.systemPrompt ?? job.taskTitle, - taskDescription: job.systemPrompt ? "" : job.taskDescription, - agentCompanyWorkspace: job.workspaceId || job.workspacePath ? { - workspaceId: job.workspaceId ?? "", - cwd: job.workspacePath ?? "", - source: "agent_company", - } : undefined, - }, - onLog: async (stream, chunk) => { - post({ - type: "run-log", - runId: job.runId, - stream, - chunk: redact(chunk), - }); - }, - }); - - const cancelled = cancelledRuns.has(job.runId); - const finalStatus = cancelled - ? "cancelled" - : result.timedOut - ? "timed_out" - : result.errorMessage || result.exitCode - ? "failed" - : "succeeded"; - - post({ - type: "run-finished", - runId: job.runId, - status: finalStatus, - summary: result.summary ?? null, - errorMessage: result.errorMessage ?? null, - exitCode: result.exitCode, - signal: result.signal, - model: result.model ?? job.model, - sessionParams: result.sessionParams ?? null, - sessionDisplayId: result.sessionDisplayId ?? null, - costUsd: typeof result.costUsd === "number" ? result.costUsd : null, - unattributedCost: !result.costUsd, - inputTokens: result.usage?.inputTokens ?? null, - outputTokens: result.usage?.outputTokens ?? null, - }); - } catch (error) { - post({ - type: "run-finished", - runId: job.runId, - status: cancelledRuns.has(job.runId) ? "cancelled" : "failed", - summary: null, - errorMessage: error instanceof Error ? error.message : String(error), - exitCode: null, - signal: null, - model: job.model, - sessionParams: null, - sessionDisplayId: null, - costUsd: null, - unattributedCost: true, - inputTokens: null, - outputTokens: null, - }); + await runner.run(job); } finally { cancelledRuns.delete(job.runId); activeByAgent.delete(job.agentId); - if (lockKey) { - activeByWorkspace.delete(lockKey); - } + if (lockKey) activeByWorkspace.delete(lockKey); void maybeStartQueuedRuns(); } } @@ -309,10 +91,7 @@ function cancelRun(runId: string) { } cancelledRuns.add(runId); - const running = runningProcesses.get(runId); - if (running) { - running.child.kill("SIGTERM"); - } + runner.cancel(runId); } parentPort.on("message", (event) => { @@ -325,11 +104,13 @@ parentPort.on("message", (event) => { cancelRun(message.runId); return; } + if (message.type === "rpc-response") { + rpc.resolve(message.rpcId, message.response); + return; + } if (message.type === "shutdown") { - // Kill every running CLI process (claude, codex, gemini, etc.) and exit. - // This is the only reliable way to ensure child processes don't survive app quit, - // because the main process cannot kill grandchildren directly. - shutdownAllRunning(); + runner.cancelAll(); + rpc.cancelAll("Worker shutting down."); process.exit(0); } }); diff --git a/src/main/worker/worker-helpers.ts b/src/main/worker/worker-helpers.ts new file mode 100644 index 00000000..06458045 --- /dev/null +++ b/src/main/worker/worker-helpers.ts @@ -0,0 +1,96 @@ +import type { RunStatus } from "@shared/types"; + +export interface RunJob { + runId: string; + companyId: string; + taskId: string; + taskTitle: string; + taskDescription: string; + agentId: string; + agentName: string; + connectorId: string; + workspaceId: string | null; + workspacePath: string | null; + workspaceMode: "exclusive" | "shared" | "none"; + command: string; + model: string | null; + thinkingEffort: string | null; + env: Record; + sessionParams: Record | null; + systemPrompt: string | null; + trigger: string | null; + managedSkills: Array<{ runtimeName: string; sourcePath: string }>; + taskSessionId: string; + executionMode: "ai_ceo" | "ai_cofounder"; +} + +export type ParentMessage = + | { type: "enqueue-run"; payload: RunJob } + | { type: "cancel-run"; runId: string } + | { type: "shutdown" } + | { type: "rpc-response"; rpcId: string; response: { ok: true; result: unknown } | { ok: false; error: string } }; + +export type ChildMessage = + | { type: "run-log"; runId: string; stream: "stdout" | "stderr"; chunk: string } + | { type: "run-status"; runId: string; status: RunStatus; message: string } + | { + type: "run-finished"; + runId: string; + status: RunStatus; + summary: string | null; + errorMessage: string | null; + exitCode: number | null; + signal: string | null; + model: string | null; + sessionParams: Record | null; + sessionDisplayId: string | null; + costUsd: number | null; + unattributedCost: boolean; + inputTokens: number | null; + outputTokens: number | null; + } + | { + type: "agent-event"; + runId: string; + agentId: string; + companyId: string; + taskId: string | null; + emittedAt: string; + event: { type: string } & Record; + } + | { type: "rpc-request"; rpcId: string; runId: string; method: string; payload: unknown }; + +export function buildRedactor(env: Record): (text: string) => string { + const sensitivePatterns: string[] = []; + for (const [key, value] of Object.entries(env)) { + if (value.length < 8) continue; + if (["PATH", "HOME", "USER", "SHELL", "LANG", "TMPDIR", "TERM"].includes(key)) continue; + sensitivePatterns.push(value); + } + if (sensitivePatterns.length === 0) return (text) => text; + sensitivePatterns.sort((a, b) => b.length - a.length); + return (text: string) => { + let result = text; + for (const pattern of sensitivePatterns) { + if (result.includes(pattern)) { + result = result.replaceAll(pattern, "[REDACTED]"); + } + } + return result; + }; +} + +/** Pushes extra env vars onto process.env and returns a restore function. */ +export function applyEnvOverrides(overrides: Record): () => void { + const saved: Record = {}; + for (const [key, value] of Object.entries(overrides)) { + saved[key] = process.env[key]; + process.env[key] = value; + } + return () => { + for (const [key, previous] of Object.entries(saved)) { + if (previous === undefined) delete process.env[key]; + else process.env[key] = previous; + } + }; +} diff --git a/src/main/worker/worker-message-handler.ts b/src/main/worker/worker-message-handler.ts index dec030b4..91352b14 100644 --- a/src/main/worker/worker-message-handler.ts +++ b/src/main/worker/worker-message-handler.ts @@ -1,3 +1,4 @@ +import type { RuntimeBridge } from "../agent-runtime"; import type { workerMessageSchema } from "./worker-types"; import { handleRunLog } from "./handlers/run-log-handler"; import { handleRunStatus } from "./handlers/run-status-handler"; @@ -6,8 +7,8 @@ import type { HandlerContext, HandlerDeps } from "./handlers/handler-context"; export async function handleWorkerMessage( message: unknown, - _worker: ReturnType | null, - deps: HandlerDeps, + worker: ReturnType | null, + deps: HandlerDeps & { getRuntimeBridge?: () => RuntimeBridge | null }, workerMessageSchemaRef: typeof workerMessageSchema, destroyed: boolean, MAX_RETRY_ATTEMPTS: number, @@ -44,5 +45,38 @@ export async function handleWorkerMessage( case "run-finished": handleRunFinished(event, deps, context); return; + case "agent-event": { + const bridge = deps.getRuntimeBridge?.(); + if (bridge) { + bridge.handleAgentEvent({ + runId: event.runId, + agentId: event.agentId, + companyId: event.companyId, + taskId: event.taskId, + emittedAt: event.emittedAt, + event: event.event, + }); + } + return; + } + case "rpc-request": { + const bridge = deps.getRuntimeBridge?.(); + const response = bridge + ? await bridge.handleRpc({ + rpcId: event.rpcId, + runId: event.runId, + method: event.method, + payload: event.payload, + }) + : { ok: false as const, error: "RuntimeBridge not initialised." }; + try { + worker?.postMessage({ type: "rpc-response", rpcId: event.rpcId, response }); + } catch (error) { + deps.logger.error( + `[worker] rpc-response postMessage failed: ${error instanceof Error ? error.message : String(error)}`, + ); + } + return; + } } } diff --git a/src/main/worker/worker-rpc.ts b/src/main/worker/worker-rpc.ts new file mode 100644 index 00000000..719ac3bb --- /dev/null +++ b/src/main/worker/worker-rpc.ts @@ -0,0 +1,34 @@ +import { randomUUID } from "node:crypto"; +import type { ChildMessage } from "./worker-helpers"; + +interface PendingRpc { + resolve: (value: { ok: true; result: unknown } | { ok: false; error: string }) => void; +} + +export class WorkerRpcClient { + private readonly pending = new Map(); + + constructor(private readonly post: (message: ChildMessage) => void) {} + + invoke(runId: string, method: string, payload: unknown): Promise<{ ok: true; result: unknown } | { ok: false; error: string }> { + const rpcId = randomUUID(); + return new Promise((resolve) => { + this.pending.set(rpcId, { resolve }); + this.post({ type: "rpc-request", rpcId, runId, method, payload }); + }); + } + + resolve(rpcId: string, response: { ok: true; result: unknown } | { ok: false; error: string }): void { + const pending = this.pending.get(rpcId); + if (!pending) return; + this.pending.delete(rpcId); + pending.resolve(response); + } + + cancelAll(reason: string): void { + for (const [id, pending] of this.pending) { + pending.resolve({ ok: false, error: reason }); + this.pending.delete(id); + } + } +} diff --git a/src/main/worker/worker-runner.ts b/src/main/worker/worker-runner.ts new file mode 100644 index 00000000..06f3d262 --- /dev/null +++ b/src/main/worker/worker-runner.ts @@ -0,0 +1,192 @@ +import { Agent } from "@mariozechner/pi-agent-core"; +import type { AgentMessage } from "@mariozechner/pi-agent-core"; +import { getModel, streamSimple } from "@mariozechner/pi-ai"; +import { createCodingTools } from "@mariozechner/pi-coding-agent"; +import type { RunStatus } from "@shared/types"; +import type { ChildMessage, RunJob } from "./worker-helpers"; +import { applyEnvOverrides, buildRedactor } from "./worker-helpers"; +import type { WorkerRpcClient } from "./worker-rpc"; + +interface PrepareRunResult { + provider: string; + apiKey: string; + extraEnv: Record; + priorMessages: AgentMessage[]; +} + +interface RunnerDeps { + post: (message: ChildMessage) => void; + rpc: WorkerRpcClient; + isCancelled: (runId: string) => boolean; +} + +export class WorkerRunner { + private readonly abortControllers = new Map(); + + constructor(private readonly deps: RunnerDeps) {} + + cancel(runId: string): void { + const controller = this.abortControllers.get(runId); + if (controller) controller.abort(); + } + + cancelAll(): void { + for (const controller of this.abortControllers.values()) { + controller.abort(); + } + this.abortControllers.clear(); + } + + async run(job: RunJob): Promise { + const { post, rpc, isCancelled } = this.deps; + const redact = buildRedactor(job.env); + const abortController = new AbortController(); + this.abortControllers.set(job.runId, abortController); + + post({ type: "run-status", runId: job.runId, status: "running", message: `Running agent ${job.agentName}.` }); + + let resolvedModelId: string | null = job.model; + let finalUsage: { inputTokens: number | null; outputTokens: number | null; costUsd: number | null } = { + inputTokens: null, + outputTokens: null, + costUsd: null, + }; + let errorMessage: string | null = null; + + try { + if (!job.model) { + throw new Error(`Run ${job.runId} has no model selected.`); + } + + const prepareResponse = await rpc.invoke(job.runId, "prepareRun", { + connectorId: job.connectorId, + taskSessionId: job.taskSessionId, + agentId: job.agentId, + companyId: job.companyId, + taskId: job.taskId, + modelId: job.model, + }); + if (!prepareResponse.ok) { + throw new Error(prepareResponse.error); + } + const prepared = prepareResponse.result as PrepareRunResult; + + const restoreEnv = applyEnvOverrides(prepared.extraEnv); + try { + const model = getModel(prepared.provider as never, job.model as never); + resolvedModelId = model.id; + + const cwd = job.workspacePath ?? process.cwd(); + const tools = createCodingTools(cwd) as never; + + const agent = new Agent({ + initialState: { + systemPrompt: job.systemPrompt ?? "", + model, + tools, + messages: prepared.priorMessages, + }, + streamFn: (m, messages, options) => + streamSimple(m, messages, { ...options, apiKey: prepared.apiKey }), + beforeToolCall: async (context) => { + const verdict = await rpc.invoke(job.runId, "beforeToolCall", { + agentId: job.agentId, + companyId: job.companyId, + taskId: job.taskId, + toolCallId: context.toolCall.id, + toolName: context.toolCall.name, + args: context.args, + }); + if (verdict.ok) { + return verdict.result as { block?: boolean; reason?: string } | undefined; + } + return { block: true, reason: verdict.error }; + }, + afterToolCall: async (context) => { + await rpc.invoke(job.runId, "afterToolCall", { + toolCallId: context.toolCall.id, + toolName: context.toolCall.name, + isError: context.isError, + }); + return undefined; + }, + }); + + const unsubscribe = agent.subscribe(async (event) => { + const timestamp = new Date().toISOString(); + post({ + type: "agent-event", + runId: job.runId, + agentId: job.agentId, + companyId: job.companyId, + taskId: job.taskId, + emittedAt: timestamp, + event: event as { type: string } & Record, + }); + + if (event.type === "message_end" && event.message) { + await rpc.invoke(job.runId, "persistMessage", { + taskSessionId: job.taskSessionId, + message: event.message, + }); + const maybeAssistant = event.message as { role?: string; usage?: { input?: number; output?: number; cost?: { total?: number } } }; + if (maybeAssistant.role === "assistant" && maybeAssistant.usage) { + finalUsage = { + inputTokens: typeof maybeAssistant.usage.input === "number" ? maybeAssistant.usage.input : finalUsage.inputTokens, + outputTokens: typeof maybeAssistant.usage.output === "number" ? maybeAssistant.usage.output : finalUsage.outputTokens, + costUsd: typeof maybeAssistant.usage.cost?.total === "number" ? maybeAssistant.usage.cost.total : finalUsage.costUsd, + }; + } + } + }); + + try { + const userMessage = job.taskTitle + ? `${job.taskTitle}\n\n${job.taskDescription ?? ""}`.trim() + : (job.taskDescription ?? ""); + const promptContent = userMessage.length > 0 ? userMessage : "Continue."; + const promptMessage: AgentMessage = { + role: "user", + content: [{ type: "text", text: promptContent }], + timestamp: Date.now(), + }; + const runPromise = agent.prompt(promptMessage); + abortController.signal.addEventListener("abort", () => agent.abort(), { once: true }); + await runPromise; + } finally { + unsubscribe(); + } + } finally { + restoreEnv(); + } + } catch (error) { + errorMessage = error instanceof Error ? error.message : String(error); + post({ type: "run-log", runId: job.runId, stream: "stderr", chunk: redact(errorMessage) }); + } finally { + this.abortControllers.delete(job.runId); + } + + const status: RunStatus = isCancelled(job.runId) + ? "cancelled" + : errorMessage + ? "failed" + : "succeeded"; + + post({ + type: "run-finished", + runId: job.runId, + status, + summary: null, + errorMessage, + exitCode: null, + signal: null, + model: resolvedModelId, + sessionParams: job.sessionParams, + sessionDisplayId: null, + costUsd: finalUsage.costUsd, + unattributedCost: finalUsage.costUsd === null, + inputTokens: finalUsage.inputTokens, + outputTokens: finalUsage.outputTokens, + }); + } +} diff --git a/src/main/worker/worker-types.ts b/src/main/worker/worker-types.ts index a617bbbd..2b7f74a4 100644 --- a/src/main/worker/worker-types.ts +++ b/src/main/worker/worker-types.ts @@ -4,6 +4,7 @@ import type { DesktopEvent, RunRecord } from "@shared/types"; import type { NotificationOptions } from "../notification-service"; import type { createBudgetEnforcementService } from "../budget-enforcement-service"; import type { createSessionManagementService } from "../session-management-service"; +import type { RuntimeBridge } from "../agent-runtime"; const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]); @@ -36,6 +37,22 @@ export const workerMessageSchema = z.union([ inputTokens: z.number().nullable(), outputTokens: z.number().nullable(), }), + z.object({ + type: z.literal("agent-event"), + runId: z.string().uuid(), + agentId: z.string(), + companyId: z.string(), + taskId: z.string().nullable(), + emittedAt: z.string(), + event: z.record(z.string(), z.unknown()).and(z.object({ type: z.string() })), + }), + z.object({ + type: z.literal("rpc-request"), + rpcId: z.string(), + runId: z.string().uuid(), + method: z.enum(["beforeToolCall", "afterToolCall", "persistMessage", "prepareRun"]), + payload: z.unknown(), + }), ]); export interface WorkerServiceDependencies { @@ -70,4 +87,6 @@ export interface WorkerServiceDependencies { notify: (options: NotificationOptions) => void; budgetService?: ReturnType; sessionService?: ReturnType; + /** Forwards `agent-event` from worker to renderer and routes RPC to main services. */ + getRuntimeBridge?: () => RuntimeBridge | null; }