feat(04-02): worker lock release, wake promotion, IPC wakeup channel, and preload bridge
- Add releaseTaskLock to worker crash recovery after finishRun (EXEC-02) - Track affected agent company IDs for targeted wake promotion - Replace listWakeupRequests+loop with promoteWakeupRequests+wakeAgentIfPossible in run completion - Track hadDeferredWakes flag for circuit breaker self-wake decision - Add listWakeupRequests IPC channel to IPC_CHANNELS - Add wakeupQuerySchema with Zod validation and WakeupRequestRecord interface - Add listWakeupRequests to DesktopApi type - Add listWakeupRequests preload bridge method - Add registerHandle for listWakeupRequests IPC handler with company validation
This commit is contained in:
1 parent
8a424822a5
commit
930273ce87
4 files changed
+58
-15
No files matched your search
@@ -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,
|
||||
}));
|
||||
});
|
||||
}
|
||||
+20
-15
@@ -107,6 +107,7 @@ export function createWorkerService({
|
||||
|
||||
const snapshot = db.listSnapshot();
|
||||
const affectedAgentIds = new Set<string>();
|
||||
const affectedCompanyIds = new Map<string, string>();
|
||||
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) {
|
||||
|
||||
@@ -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) =>
|
||||
|
||||
@@ -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<T> =
|
||||
| { ok: true; data: T }
|
||||
| { ok: false; error: { code: string; message: string } };
|
||||
@@ -653,6 +677,7 @@ export type DesktopApi = {
|
||||
checkSkillUpdate: (companyId: string, skillId: string) => Promise<DesktopResult<SkillUpdateStatus>>;
|
||||
listSkillAgents: (companyId: string, skillId: string) => Promise<DesktopResult<string[]>>;
|
||||
listAgentSkills: (companyId: string, agentId: string) => Promise<DesktopResult<string[]>>;
|
||||
listWakeupRequests: (input: z.infer<typeof wakeupQuerySchema>) => Promise<DesktopResult<WakeupRequestRecord[]>>;
|
||||
recoveryRetry: (taskId: string, agentId: string, companyId: string) => Promise<DesktopResult<null>>;
|
||||
recoveryDismiss: (taskId: string) => Promise<DesktopResult<null>>;
|
||||
recoveryFail: (taskId: string) => Promise<DesktopResult<null>>;
|
||||
|
||||
Reference in new issue
Block a user