diff --git a/src/main/ipc-core-read-handlers.ts b/src/main/ipc-core-read-handlers.ts index fff3cd4a..fb658f8f 100644 --- a/src/main/ipc-core-read-handlers.ts +++ b/src/main/ipc-core-read-handlers.ts @@ -6,6 +6,7 @@ import { connectorInputSchema, metricsQuerySchema, settingsSchema, + wakeupQuerySchema, } from "@shared/contracts"; import { fail, ok, type RegisterCoreHandlersDependencies } from "./ipc-core-common"; @@ -142,4 +143,15 @@ export function registerCoreReadHandlers({ if (err) return err; return ok(db.getStandupReport(companyId)); }); + + registerHandle(IPC_CHANNELS.listWakeupRequests, (_event, input) => { + const parsed = wakeupQuerySchema.parse(input); + const err = ensureCompanyExists(parsed.companyId); + if (err) return err; + return ok(db.listWakeupRequests(parsed.companyId, { + status: parsed.status, + agentId: parsed.agentId, + limit: parsed.limit, + })); + }); } diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index d4cbb8d7..b8d43529 100644 --- a/src/main/worker-service.ts +++ b/src/main/worker-service.ts @@ -107,6 +107,7 @@ export function createWorkerService({ const snapshot = db.listSnapshot(); const affectedAgentIds = new Set(); + const affectedCompanyIds = new Map(); const requeueCandidates: Array<{ id: string; taskId: string; agentId: string; companyId: string; workspaceId: string | null; connectorId: string }> = []; for (const run of snapshot.runs) { @@ -126,7 +127,10 @@ export function createWorkerService({ if (run.workspaceId) { db.releaseWorkspaceLock(run.workspaceId, run.id); } + // EXEC-02: Release execution lock on crash recovery + 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) { @@ -145,15 +149,13 @@ export function createWorkerService({ } for (const agentId of affectedAgentIds) { - // Promote any deferred wakes back to queued, then process queued wakes - const agentRuns = snapshot.runs.filter((r) => r.agentId === agentId); - for (const companyId of new Set(agentRuns.map((r) => r.companyId))) { - db.promoteWakeupRequests(agentId, companyId); - const wakes = db.listWakeupRequests(companyId, { agentId, status: "queued" }); - for (const wake of wakes) { - wakeAgentIfPossible(agentId, wake.companyId, wake.source); - } + // 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`); } + // Let heartbeat scheduler pick up promoted wakes on next tick } publishDomainChanged(); @@ -386,12 +388,15 @@ export function createWorkerService({ } } - // Promote any deferred wakes back to queued after run completes - db.promoteWakeupRequests(run.agentId, run.companyId); - const queuedWakes = db.listWakeupRequests(run.companyId, { agentId: run.agentId, status: "queued" }); - if (queuedWakes.length > 0 && !isQuitting() && !destroyed) { - for (const wake of queuedWakes) { - wakeAgentIfPossible(run.agentId, wake.companyId, wake.source); + // EXEC-05: Promote deferred wakes after run completion + let hadDeferredWakes = false; + if (!isQuitting() && !destroyed) { + const promoted = db.promoteWakeupRequests(run.agentId, run.companyId); + if (promoted > 0) { + hadDeferredWakes = true; + logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); + // Trigger processing of newly-promoted wakes + wakeAgentIfPossible(run.agentId, run.companyId, "deferred_wake_promoted"); } } @@ -411,7 +416,7 @@ export function createWorkerService({ } // Self-wake with circuit breaker to prevent infinite loops - if (queuedWakes.length === 0 && !isQuitting() && !destroyed) { + if (!hadDeferredWakes && !isQuitting() && !destroyed) { try { const actionableTasks = db.listAgentTasks(run.agentId, run.companyId, ["todo", "in_progress"]); if (actionableTasks.length > 0) { diff --git a/src/preload/index.ts b/src/preload/index.ts index bd8afb60..1e139a5a 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -94,6 +94,7 @@ const api: DesktopApi = { ipcRenderer.invoke(IPC_CHANNELS.listSkillAgents, { companyId, skillId }), listAgentSkills: (companyId, agentId) => ipcRenderer.invoke(IPC_CHANNELS.listAgentSkills, { companyId, agentId }), + listWakeupRequests: (input) => ipcRenderer.invoke(IPC_CHANNELS.listWakeupRequests, input), recoveryRetry: (taskId: string, agentId: string, companyId: string) => ipcRenderer.invoke("recovery:retry", { taskId, agentId, companyId }), recoveryDismiss: (taskId: string) => diff --git a/src/shared/contracts.ts b/src/shared/contracts.ts index 375d569b..7087dddf 100644 --- a/src/shared/contracts.ts +++ b/src/shared/contracts.ts @@ -82,6 +82,7 @@ export const IPC_CHANNELS = { checkSkillUpdate: "agent-company:check-skill-update", listSkillAgents: "agent-company:list-skill-agents", listAgentSkills: "agent-company:list-agent-skills", + listWakeupRequests: "agent-company:list-wakeup-requests", } as const; const optionalId = z.preprocess( @@ -569,6 +570,29 @@ export const listAgentSkillsSchema = z.object({ agentId: z.string().min(1), }); +export const wakeupQuerySchema = z.object({ + companyId: z.string().uuid(), + status: z.enum(["queued", "claimed", "coalesced", "deferred", "completed"]).optional(), + agentId: z.string().uuid().optional(), + limit: z.number().int().min(1).max(100).default(50), +}); + +export interface WakeupRequestRecord { + id: string; + companyId: string; + agentId: string; + source: string; + triggerDetail: string | null; + status: string; + coalescedCount: number; + runId: string | null; + requestedAt: string; + claimedAt: string | null; + finishedAt: string | null; + createdAt: string; + updatedAt: string; +} + export type DesktopResult = | { ok: true; data: T } | { ok: false; error: { code: string; message: string } }; @@ -653,6 +677,7 @@ export type DesktopApi = { checkSkillUpdate: (companyId: string, skillId: string) => Promise>; listSkillAgents: (companyId: string, skillId: string) => Promise>; listAgentSkills: (companyId: string, agentId: string) => Promise>; + listWakeupRequests: (input: z.infer) => Promise>; recoveryRetry: (taskId: string, agentId: string, companyId: string) => Promise>; recoveryDismiss: (taskId: string) => Promise>; recoveryFail: (taskId: string) => Promise>;