fix(pi-workflow): preserve recovery gate ownership
This commit is contained in:
@@ -1261,7 +1261,7 @@ export class WorkflowEngine {
|
||||
validateWorkflowCommand(command);
|
||||
if (command.action === "start") return this.start(command);
|
||||
if (command.action === "start-analysis") return this.startAnalysis(command);
|
||||
const run = await this.options.store.loadRun(command.runId);
|
||||
let run = await this.options.store.loadRun(command.runId);
|
||||
if (command.action === "inspect") return view(run);
|
||||
if (command.action === "consume-context-handoff") {
|
||||
return this.consumeContextHandoff(run, command);
|
||||
@@ -1269,6 +1269,7 @@ export class WorkflowEngine {
|
||||
if (command.action === "recover-context-intent") {
|
||||
return this.recoverContextIntent(run, command);
|
||||
}
|
||||
run = this.reclaimStaleRecoveryLease(run, command);
|
||||
this.assertActiveParent(run);
|
||||
if (command.action === "abort") return this.abort(run, command);
|
||||
if (command.action === "decide") return this.decide(run, command);
|
||||
@@ -1332,6 +1333,8 @@ export class WorkflowEngine {
|
||||
assignment.status === "running" ||
|
||||
(assignment.status === "queued" && Boolean(assignment.executionToken)),
|
||||
);
|
||||
const alreadyWaiting = activeGate(run)?.kind === "recovery";
|
||||
if (running.length === 0 && alreadyWaiting) continue;
|
||||
if (await this.hasActiveAssignmentLease(run.runId, running)) continue;
|
||||
if (!this.leaseIsStale(run)) continue;
|
||||
run = await this.transition(run, (next) => {
|
||||
@@ -1362,8 +1365,6 @@ export class WorkflowEngine {
|
||||
);
|
||||
}
|
||||
}
|
||||
const alreadyWaiting = activeGate(run)?.kind === "recovery";
|
||||
if (running.length === 0 && alreadyWaiting) continue;
|
||||
run = await this.transition(run, (next) => {
|
||||
const interruptedIds = new Set(running.map((assignment) => assignment.id));
|
||||
for (const assignment of next.assignments) {
|
||||
@@ -4519,6 +4520,31 @@ export class WorkflowEngine {
|
||||
});
|
||||
}
|
||||
|
||||
private reclaimStaleRecoveryLease(
|
||||
run: WorkflowRun,
|
||||
command: WorkflowCommand,
|
||||
) {
|
||||
if (
|
||||
command.action !== "decide" ||
|
||||
command.decision.kind !== "recover" ||
|
||||
command.decision.resolution !== "retry" ||
|
||||
activeGate(run)?.id !== command.gateId ||
|
||||
!run.parentSessionIds.includes(this.options.parentSessionId) ||
|
||||
!run.lease ||
|
||||
this.isProcessAlive(run.lease.processId)
|
||||
) {
|
||||
return run;
|
||||
}
|
||||
const claimed = structuredClone(run);
|
||||
claimed.lease = {
|
||||
processId: this.processId,
|
||||
activeParentSessionId: this.options.parentSessionId,
|
||||
heartbeatAt: this.now(),
|
||||
worktreeIdentity: claimed.workspace.root,
|
||||
};
|
||||
return claimed;
|
||||
}
|
||||
|
||||
private assertActiveParent(run: WorkflowRun) {
|
||||
if (
|
||||
run.lease &&
|
||||
|
||||
Reference in New Issue
Block a user