Reorganize worker entry layer into worker/entry/ (M8 part 1/3)

Pulls the worker process entry point, parent<->worker IPC plumbing, message validation schema, and crash-recovery handler into a single src/main/worker/entry/ module. No runtime behavior changes — only directory + file-name renames plus import-path updates: worker-entry.ts -> entry/index.ts, worker-message-handler.ts -> entry/message-handler.ts, worker-rpc.ts -> entry/rpc.ts, worker-crash-recovery.ts -> entry/crash-recovery.ts, worker-types.ts -> entry/types.ts. electron.vite.config.ts, worker-service.ts, handlers/*.ts, pi-agent-factory.ts, and worker-runner.ts updated to point at the new locations.

src/main/worker/entry/crash-recovery.ts

@@ -0,0 +1,138 @@
+import type { WorkerServiceDependencies } from "./types";
+import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace";
+
+export function handleWorkerExit(
+  code: number | null,
+  {
+    db,
+    logger,
+    isQuitting,
+    publishDomainChanged,
+    emitEvent: _emitEvent,
+    wakeAgentIfPossible,
+    dispatchRunToWorker,
+  }: Pick<WorkerServiceDependencies, "db" | "logger" | "isQuitting" | "publishDomainChanged" | "emitEvent" | "wakeAgentIfPossible" | "dispatchRunToWorker">,
+  createWorker: () => void,
+  getWorker: () => ReturnType<typeof import("electron").utilityProcess.fork> | null,
+) {
+  if (isQuitting()) return;
+
+  logger.warn(`[worker] Worker exited with code ${code}, recovering...`);
+
+  const snapshot = db.listSnapshot();
+  const affectedAgentIds = new Set<string>();
+  const affectedCompanyIds = new Map<string, string>();
+  const requeueCandidates: Array<{
+    id: string;
+    taskId: string | null;
+    agentId: string;
+    companyId: string;
+    workspaceId: string | null;
+    connectorId: string;
+  }> = [];
+
+  for (const run of snapshot.runs) {
+    if (run.status === "running") {
+      db.finishRun({
+        runId: run.id,
+        status: "interrupted",
+        summary: "Worker process crashed.",
+        errorMessage: "Worker process exited unexpectedly.",
+        exitCode: code,
+        signal: null,
+        model: null,
+        unattributedCost: true,
+      });
+      if (run.workspaceId) {
+        db.releaseWorkspaceLock(run.workspaceId, run.id);
+      }
+      // EXEC-02: Release execution lock on crash recovery
+      if (run.taskId) {
+        db.releaseTaskLock(run.taskId, run.id);
+      }
+      affectedAgentIds.add(run.agentId);
+      affectedCompanyIds.set(run.agentId, run.companyId);
+
+      const chain = db.getChainOfCommand(run.agentId);
+      if (chain.length > 0) {
+        wakeAgentIfPossible(chain[0].id, run.companyId, "report_failed");
+      }
+    } else if (run.status === "queued") {
+      requeueCandidates.push({
+        id: run.id,
+        taskId: run.taskId,
+        agentId: run.agentId,
+        companyId: run.companyId,
+        workspaceId: run.workspaceId,
+        connectorId: run.connectorId,
+      });
+    }
+  }
+
+  for (const agentId of affectedAgentIds) {
+    // EXEC-05: Promote deferred wakes after crash recovery
+    const cid = affectedCompanyIds.get(agentId) ?? "";
+    const promoted = db.promoteWakeupRequests(agentId, cid);
+    if (promoted > 0) {
+      logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${agentId} after crash recovery`);
+    }
+    const pendingWakeTrigger = db.getPreferredWakeTrigger(agentId, cid);
+    if (pendingWakeTrigger) {
+      wakeAgentIfPossible(agentId, cid, pendingWakeTrigger);
+    }
+  }
+
+  publishDomainChanged();
+
+  createWorker();
+
+  const worker = getWorker();
+  if (requeueCandidates.length > 0 && worker) {
+    logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`);
+    const recoverySnapshot = db.listSnapshot();
+    for (const candidate of requeueCandidates) {
+      try {
+        // The crashed run may still own a workspace lock acquired during the
+        // original dispatch; release it before re-dispatch so the replay
+        // does not self-collide with "Workspace is locked by another run".
+        if (candidate.workspaceId) {
+          db.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
+        }
+        const task = candidate.taskId ? db.getTask(candidate.taskId) : null;
+        const agent = db.getAgent(candidate.agentId);
+        const connector = db.getConnector(agent.connectorId);
+        const { workspace, lockMode } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId);
+        const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId);
+
+        dispatchRunToWorker({
+          runId: candidate.id,
+          companyId: candidate.companyId,
+          agent,
+          task,
+          connector,
+          workspace,
+          workspaceMode: lockMode,
+          companyName: recoveryCompany?.name ?? "Company",
+          companyDescription: recoveryCompany?.description ?? "",
+          snapshot: recoverySnapshot,
+          trigger: "recovery",
+        });
+      } catch (error) {
+        logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`);
+        db.finishRun({
+          runId: candidate.id,
+          status: "interrupted",
+          summary: "Could not re-enqueue after worker restart.",
+          errorMessage: error instanceof Error ? error.message : String(error),
+          exitCode: null,
+          signal: null,
+          model: null,
+          unattributedCost: true,
+        });
+        if (candidate.workspaceId) {
+          db.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
+        }
+      }
+    }
+  }
+}

src/main/worker/entry/index.ts

@@ -0,0 +1,159 @@
+import type { ChildMessage, ParentMessage, RunJob } from "../worker-helpers";
+import { WorkerRpcClient } from "./rpc";
+import { WorkerRunner } from "../worker-runner";
+
+const parentPort = process.parentPort;
+
+if (!parentPort) {
+  throw new Error("AgentCompany worker requires process.parentPort");
+}
+
+interface ActiveRun {
+  runId: string;
+  agentId: string;
+  workspaceKey: string | null;
+}
+
+/**
+ * Single source of truth for scheduling invariants: "one agent at a time"
+ * and "one exclusive-workspace lock at a time". Prior layout kept three
+ * parallel structures (`activeByAgent`, `activeByWorkspace`, queue) that
+ * had to be updated in lockstep. The Map-of-ActiveRun keyed by runId is
+ * the only authoritative record; agent/workspace busy checks iterate it.
+ */
+const active = new Map<string, ActiveRun>();
+const queue: RunJob[] = [];
+const cancelledRuns = new Set<string>();
+
+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): string | null {
+  if (job.workspaceMode !== "exclusive") return null;
+  return job.workspaceId ?? job.workspacePath ?? `task:${job.taskId}`;
+}
+
+function isAgentBusy(agentId: string): boolean {
+  for (const run of active.values()) {
+    if (run.agentId === agentId) return true;
+  }
+  return false;
+}
+
+function isWorkspaceBusy(lockKey: string): boolean {
+  for (const run of active.values()) {
+    if (run.workspaceKey === lockKey) return true;
+  }
+  return false;
+}
+
+function enqueue(job: RunJob) {
+  queue.push(job);
+  post({
+    type: "run-status",
+    runId: job.runId,
+    status: "queued",
+    message: "Run queued in orchestration worker.",
+  });
+  tickQueue();
+}
+
+/**
+ * Claim-before-dispatch: for each queued job whose agent + workspace are
+ * free, register the claim in `active` synchronously *before* calling
+ * `startRun`. No async window between claim and dispatch.
+ */
+function tickQueue() {
+  let i = 0;
+  while (i < queue.length) {
+    const job = queue[i];
+    if (isAgentBusy(job.agentId)) { i += 1; continue; }
+    const lockKey = workspaceKey(job);
+    if (lockKey && isWorkspaceBusy(lockKey)) { i += 1; continue; }
+    active.set(job.runId, { runId: job.runId, agentId: job.agentId, workspaceKey: lockKey });
+    queue.splice(i, 1);
+    void startRun(job);
+  }
+}
+
+async function startRun(job: RunJob) {
+  try {
+    await runner.run(job);
+  } finally {
+    cancelledRuns.delete(job.runId);
+    active.delete(job.runId);
+    tickQueue();
+  }
+}
+
+function cancelRun(runId: string) {
+  const queuedIndex = queue.findIndex((entry) => entry.runId === runId);
+  if (queuedIndex >= 0) {
+    queue.splice(queuedIndex, 1);
+    post({
+      type: "run-finished",
+      runId,
+      status: "cancelled",
+      summary: "Cancelled before execution.",
+      errorMessage: null,
+      exitCode: null,
+      signal: null,
+      model: null,
+      unattributedCost: true,
+      inputTokens: null,
+      outputTokens: null,
+    });
+    return;
+  }
+
+  cancelledRuns.add(runId);
+  runner.cancel(runId);
+}
+
+parentPort.on("message", (event) => {
+  const message = event.data as ParentMessage;
+  if (message.type === "enqueue-run") {
+    enqueue(message.payload);
+    return;
+  }
+  if (message.type === "cancel-run") {
+    cancelRun(message.runId);
+    return;
+  }
+  if (message.type === "steer-run") {
+    const accepted = runner.steer(message.runId, message.message);
+    post({
+      type: "steer-ack",
+      runId: message.runId,
+      correlationId: message.correlationId,
+      accepted,
+    });
+    return;
+  }
+  if (message.type === "rpc-response") {
+    rpc.resolve(message.rpcId, message.response);
+    return;
+  }
+  if (message.type === "shutdown") {
+    void (async () => {
+      // Graceful drain: abort every active run and await its promise
+      // settlement so that final `run-finished` messages and
+      // `persistMessage` RPCs reach the parent before we exit. Bounded by
+      // a 5s budget so a stuck run does not wedge the shutdown handshake
+      // (the parent has its own kill-timer as a safety net).
+      const drainBudgetMs = 5_000;
+      const timeout = new Promise<void>((resolve) => setTimeout(resolve, drainBudgetMs));
+      await Promise.race([runner.cancelAll(), timeout]);
+      rpc.cancelAll("Worker shutting down.");
+      process.exit(0);
+    })();
+  }
+});

src/main/worker/entry/message-handler.ts

@@ -0,0 +1,98 @@
+import type { RuntimeBridge, RuntimeMessageInjector } from "../../agent-runtime";
+import type { workerMessageSchema } from "./types";
+import { handleRunLog } from "../handlers/run-log-handler";
+import { handleRunStatus } from "../handlers/run-status-handler";
+import { handleRunFinished } from "../handlers/run-finished-handler";
+import type { HandlerContext, HandlerDeps } from "../handlers/handler-context";
+
+export async function handleWorkerMessage(
+  message: unknown,
+  worker: ReturnType<typeof import("electron").utilityProcess.fork> | null,
+  deps: HandlerDeps & {
+    getRuntimeBridge?: () => RuntimeBridge | null;
+    getRuntimeMessageInjector?: () => RuntimeMessageInjector | null;
+  },
+  workerMessageSchemaRef: typeof workerMessageSchema,
+  destroyed: boolean,
+  MAX_RETRY_ATTEMPTS: number,
+  RETRY_COOLDOWN_MS: number,
+  consecutiveWakeMap: Map<string, number>,
+  MAX_CONSECUTIVE_WAKES: number,
+  pendingRetryTimers: Set<ReturnType<typeof setTimeout>>,
+  leaseRenewalMap: Map<string, number>,
+) {
+  const parsed = workerMessageSchemaRef.safeParse(message);
+  if (!parsed.success) {
+    deps.logger.error("[worker] invalid message payload", parsed.error);
+    return;
+  }
+  const event = parsed.data;
+
+  const context: HandlerContext = {
+    destroyed,
+    MAX_RETRY_ATTEMPTS,
+    RETRY_COOLDOWN_MS,
+    MAX_CONSECUTIVE_WAKES,
+    consecutiveWakeMap,
+    pendingRetryTimers,
+    leaseRenewalMap,
+  };
+
+  switch (event.type) {
+    case "run-log":
+      await handleRunLog(event, deps, context);
+      return;
+    case "run-status":
+      handleRunStatus(event, deps, context);
+      return;
+    case "run-finished": {
+      try {
+        handleRunFinished(event, deps, context);
+      } finally {
+        deps.getRuntimeBridge?.()?.releaseRunResources(event.runId);
+      }
+      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;
+    }
+    case "run-agent-state": {
+      deps.getRuntimeBridge?.()?.setRunAgentState(event.runId, event.state);
+      return;
+    }
+    case "steer-ack": {
+      deps.getRuntimeMessageInjector?.()?.handleSteerAck(event.correlationId, event.accepted);
+      return;
+    }
+  }
+}

src/main/worker/entry/rpc.ts

@@ -0,0 +1,54 @@
+import { randomUUID } from "node:crypto";
+import type { ChildMessage } from "../worker-helpers";
+
+interface PendingRpc {
+  resolve: (value: { ok: true; result: unknown } | { ok: false; error: string }) => void;
+  timer: NodeJS.Timeout;
+}
+
+const DEFAULT_RPC_TIMEOUT_MS = 30_000;
+
+export class WorkerRpcClient {
+  private readonly pending = new Map<string, PendingRpc>();
+
+  constructor(
+    private readonly post: (message: ChildMessage) => void,
+    private readonly timeoutMs: number = DEFAULT_RPC_TIMEOUT_MS,
+  ) {}
+
+  invoke(
+    runId: string,
+    method: string,
+    payload: unknown,
+    options?: { timeoutMs?: number },
+  ): Promise<{ ok: true; result: unknown } | { ok: false; error: string }> {
+    const rpcId = randomUUID();
+    const effectiveTimeoutMs = options?.timeoutMs ?? this.timeoutMs;
+    return new Promise((resolve) => {
+      const timer = setTimeout(() => {
+        const pending = this.pending.get(rpcId);
+        if (!pending) return;
+        this.pending.delete(rpcId);
+        pending.resolve({ ok: false, error: `RPC ${method} timed out after ${effectiveTimeoutMs}ms` });
+      }, effectiveTimeoutMs);
+      this.pending.set(rpcId, { resolve, timer });
+      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;
+    clearTimeout(pending.timer);
+    this.pending.delete(rpcId);
+    pending.resolve(response);
+  }
+
+  cancelAll(reason: string): void {
+    for (const [id, pending] of this.pending) {
+      clearTimeout(pending.timer);
+      pending.resolve({ ok: false, error: reason });
+      this.pending.delete(id);
+    }
+  }
+}

src/main/worker/entry/types.ts

@@ -0,0 +1,111 @@
+import { z } from "zod";
+import type { AppDatabase } from "../../database";
+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, RuntimeMessageInjector } from "../../agent-runtime";
+
+const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]);
+
+export const workerMessageSchema = z.union([
+  z.object({
+    type: z.literal("run-log"),
+    runId: z.string().uuid(),
+    stream: z.enum(["stdout", "stderr"]),
+    chunk: z.string(),
+  }),
+  z.object({
+    type: z.literal("run-status"),
+    runId: z.string().uuid(),
+    status: runStatusSchema,
+    message: z.string(),
+  }),
+  z.object({
+    type: z.literal("run-finished"),
+    runId: z.string().uuid(),
+    status: runStatusSchema,
+    summary: z.string().nullable(),
+    errorMessage: z.string().nullable(),
+    exitCode: z.number().int().nullable(),
+    signal: z.string().nullable(),
+    model: z.string().nullable(),
+    unattributedCost: z.boolean(),
+    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([
+      "persistMessage",
+      "prepareRun",
+      "beforeToolCall",
+      "afterToolCall",
+      "transformContext",
+    ]),
+    payload: z.unknown(),
+  }),
+  z.object({
+    type: z.literal("run-agent-state"),
+    runId: z.string().uuid(),
+    state: z.enum(["agent_active", "closing"]),
+  }),
+  z.object({
+    type: z.literal("steer-ack"),
+    runId: z.string().uuid(),
+    correlationId: z.string(),
+    accepted: z.boolean(),
+  }),
+]);
+
+export interface WorkerServiceDependencies {
+  db: AppDatabase;
+  logger: {
+    debug: (...args: unknown[]) => void;
+    info: (...args: unknown[]) => void;
+    warn: (...args: unknown[]) => void;
+    error: (...args: unknown[]) => void;
+  };
+  workerPath: string;
+  isQuitting: () => boolean;
+  publishDomainChanged: () => void;
+  emitEvent: (event: DesktopEvent) => void;
+  wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
+  dispatchRunToWorker: (params: {
+    runId: string;
+    companyId: string;
+    agent: ReturnType<AppDatabase["getAgent"]>;
+    task: ReturnType<AppDatabase["getTask"]> | null;
+    connector: ReturnType<AppDatabase["getConnector"]>;
+    workspace: ReturnType<AppDatabase["getWorkspace"]>;
+    workspaceMode: "exclusive" | "shared" | "none";
+    companyName: string;
+    companyDescription: string;
+    snapshot: ReturnType<AppDatabase["listSnapshot"]>;
+    trigger: string;
+  }) => void;
+  processRunCompletion: (run: RunRecord, event: z.infer<typeof workerMessageSchema> & { type: "run-finished" }) => void;
+  onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void;
+  onRunCompleted?: (runId: string, taskId: string | null, agentId: string, companyId: string, status: string, summary: string | null, errorMessage: string | null) => void;
+  notify: (options: NotificationOptions) => void;
+  budgetService?: ReturnType<typeof createBudgetEnforcementService>;
+  sessionService?: ReturnType<typeof createSessionManagementService>;
+  /** Forwards `agent-event` from worker to renderer and routes RPC to main services. */
+  getRuntimeBridge?: () => RuntimeBridge | null;
+  /**
+   * Resolves the shared runtime message injector so `steer-ack` child
+   * messages can commit inbox delivery or trigger a wake fallback.
+   */
+  getRuntimeMessageInjector?: () => RuntimeMessageInjector | null;
+}