fix: guard run completion boundary
This commit is contained in:
1 parent
245586a9aa
commit
d21edb244f
6 files changed
+268
-8
No files matched your search
@@ -90,11 +90,14 @@ async function startRun(job: RunJob) {
|
||||
await runner.run(job);
|
||||
} finally {
|
||||
cancelledRuns.delete(job.runId);
|
||||
active.delete(job.runId);
|
||||
tickQueue();
|
||||
}
|
||||
}
|
||||
|
||||
function acknowledgeRunCompletion(runId: string) {
|
||||
const released = active.delete(runId);
|
||||
if (released) tickQueue();
|
||||
}
|
||||
|
||||
function cancelRun(runId: string) {
|
||||
const queuedIndex = queue.findIndex((entry) => entry.runId === runId);
|
||||
if (queuedIndex >= 0) {
|
||||
@@ -129,6 +132,10 @@ parentPort.on("message", (event) => {
|
||||
cancelRun(message.runId);
|
||||
return;
|
||||
}
|
||||
if (message.type === "run-completion-ack") {
|
||||
acknowledgeRunCompletion(message.runId);
|
||||
return;
|
||||
}
|
||||
if (message.type === "steer-run") {
|
||||
const accepted = runner.steer(message.runId, message.message);
|
||||
post({
|
||||
|
||||
@@ -3,8 +3,10 @@ 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 { releaseCompletedRunWorkspaceLock } from "../handlers/run-finished-helpers";
|
||||
import { handleContextCompaction } from "../handlers/context-compaction-handler";
|
||||
import type { HandlerContext, HandlerDeps } from "../handlers/handler-context";
|
||||
import type { RunFinishedContext } from "../handlers/run-finished-helpers";
|
||||
|
||||
export async function handleWorkerMessage(
|
||||
message: unknown,
|
||||
@@ -47,11 +49,17 @@ export async function handleWorkerMessage(
|
||||
handleRunStatus(event, deps, context);
|
||||
return;
|
||||
case "run-finished": {
|
||||
let completion: RunFinishedContext | null = null;
|
||||
try {
|
||||
handleRunFinished(event, deps, context);
|
||||
completion = handleRunFinished(event, deps, context);
|
||||
} finally {
|
||||
deps.getRuntimeBridge?.()?.releaseRunResources(event.runId);
|
||||
}
|
||||
releaseCompletedRunWorkspaceLock(completion, deps);
|
||||
if (!worker) {
|
||||
throw new Error(`run-completion-ack worker missing for run ${event.runId}`);
|
||||
}
|
||||
worker.postMessage({ type: "run-completion-ack", runId: event.runId });
|
||||
return;
|
||||
}
|
||||
case "agent-event": {
|
||||
|
||||
@@ -6,17 +6,19 @@ import {
|
||||
triggerSelfWake,
|
||||
} from "./run-finished-helpers";
|
||||
import type { HandlerContext, HandlerDeps, WorkerMessageEvent } from "./handler-context";
|
||||
import type { RunFinishedContext } from "./run-finished-helpers";
|
||||
|
||||
export function handleRunFinished(
|
||||
event: Extract<WorkerMessageEvent, { type: "run-finished" }>,
|
||||
deps: HandlerDeps,
|
||||
context: HandlerContext,
|
||||
): void {
|
||||
): RunFinishedContext | null {
|
||||
const rfCtx = finalizeRun(event, deps, context);
|
||||
if (!rfCtx) return;
|
||||
if (!rfCtx) return null;
|
||||
|
||||
accumulateAndRotateSession(event, deps, rfCtx);
|
||||
emitAndNotifyCompletion(event, deps, rfCtx);
|
||||
const { hadDeferredWakes } = runPostCompletionAutomation(event, deps, context, rfCtx);
|
||||
triggerSelfWake(event, deps, context, rfCtx, hadDeferredWakes);
|
||||
return rfCtx;
|
||||
}
|
||||
@@ -71,9 +71,6 @@ export function finalizeRun(
|
||||
|
||||
leaseRenewalMap.delete(event.runId);
|
||||
const runOutcome = db.core.runs.finishRun({ ...event, unattributedCost });
|
||||
if (run.workspaceId) {
|
||||
db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id);
|
||||
}
|
||||
db.core.activity.addActivity({
|
||||
companyId: run.companyId,
|
||||
actor: run.agentId,
|
||||
@@ -86,6 +83,15 @@ export function finalizeRun(
|
||||
return { run, task, runOutcome };
|
||||
}
|
||||
|
||||
export function releaseCompletedRunWorkspaceLock(
|
||||
rfCtx: RunFinishedContext | null,
|
||||
deps: Pick<HandlerDeps, "db">,
|
||||
): void {
|
||||
const workspaceId = rfCtx?.run.workspaceId;
|
||||
if (!workspaceId) return;
|
||||
deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id);
|
||||
}
|
||||
|
||||
export function accumulateAndRotateSession(
|
||||
event: RunFinishedEvent,
|
||||
deps: HandlerDeps,
|
||||
|
||||
@@ -47,6 +47,7 @@ export type RunAgentLifecycleState = "agent_active" | "closing";
|
||||
export type ParentMessage =
|
||||
| { type: "enqueue-run"; payload: RunJob }
|
||||
| { type: "cancel-run"; runId: string }
|
||||
| { type: "run-completion-ack"; runId: string }
|
||||
| { type: "shutdown" }
|
||||
| { type: "steer-run"; runId: string; correlationId: string; message: AgentMessage }
|
||||
| { type: "rpc-response"; rpcId: string; response: { ok: true; result: unknown } | { ok: false; error: string } };
|
||||
|
||||
Reference in new issue
Block a user