Skip to content

Commit f78d13d

Browse files
pallyoungpallyoung
committed
fix: harden supervisor target transitions
1 parent 6b3cd13 commit f78d13d

2 files changed

Lines changed: 249 additions & 9 deletions

File tree

‎packages/server/src/__tests__/supervisor-manager.test.ts‎

Lines changed: 219 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -622,6 +622,189 @@ describe("SupervisorManager cycle triggers", () => {
622622
);
623623
});
624624

625+
it("keeps a superseded target meta state when an old in-flight cycle later stops", async () => {
626+
const supervisor = await manager.create({
627+
sessionId: "sess-objective-stop-race",
628+
workspaceId: "ws-1",
629+
objective: "Initial objective",
630+
evaluatorProviderId: "codex",
631+
});
632+
633+
const targetMetaById = new Map<string, Record<string, unknown>>([
634+
[
635+
supervisor.targetId,
636+
{
637+
targetId: supervisor.targetId,
638+
sessionId: supervisor.sessionId,
639+
workspaceId: supervisor.workspaceId,
640+
objective: supervisor.objective,
641+
status: "active",
642+
createdAt: 1,
643+
updatedAt: 1,
644+
supersededBy: null,
645+
completedAt: null,
646+
},
647+
],
648+
]);
649+
650+
deps.targetStore.readTargetMeta.mockImplementation(
651+
async (_workspacePath: string, targetId: string) => {
652+
const meta = targetMetaById.get(targetId);
653+
if (!meta) {
654+
throw new Error(`Missing target meta for ${targetId}`);
655+
}
656+
return { ...meta };
657+
}
658+
);
659+
deps.targetStore.createTargetFiles.mockImplementation(
660+
async (
661+
_workspacePath: string,
662+
input: {
663+
targetId: string;
664+
sessionId: string;
665+
workspaceId: string;
666+
objective: string;
667+
createdAt: number;
668+
}
669+
) => {
670+
targetMetaById.set(input.targetId, {
671+
targetId: input.targetId,
672+
sessionId: input.sessionId,
673+
workspaceId: input.workspaceId,
674+
objective: input.objective,
675+
status: "active",
676+
createdAt: input.createdAt,
677+
updatedAt: input.createdAt,
678+
supersededBy: null,
679+
completedAt: null,
680+
});
681+
}
682+
);
683+
deps.targetStore.markTargetSuperseded.mockImplementation(
684+
async (_workspacePath: string, targetId: string, nextTargetId: string, updatedAt: number) => {
685+
const current = targetMetaById.get(targetId);
686+
if (!current) {
687+
throw new Error(`Missing target meta for ${targetId}`);
688+
}
689+
targetMetaById.set(targetId, {
690+
...current,
691+
status: "superseded",
692+
supersededBy: nextTargetId,
693+
updatedAt,
694+
});
695+
}
696+
);
697+
deps.targetStore.saveTargetMeta.mockImplementation(
698+
async (_workspacePath: string, targetId: string, meta: Record<string, unknown>) => {
699+
targetMetaById.set(targetId, { ...meta, targetId });
700+
}
701+
);
702+
703+
let resolveEvaluation: ((result: SupervisorEvaluationResult) => void) | null = null;
704+
vi.spyOn(getManagerInternals().evaluator, "evaluate").mockImplementationOnce(
705+
async () =>
706+
await new Promise<SupervisorEvaluationResult>((resolve) => {
707+
resolveEvaluation = resolve;
708+
})
709+
);
710+
711+
const cycle = await manager.triggerEvaluation(supervisor.id);
712+
713+
await waitFor(() => {
714+
expect(resolveEvaluation).not.toBeNull();
715+
expect(manager.get(supervisor.id)?.state).toBe("evaluating");
716+
});
717+
718+
const rotated = await manager.update(supervisor.id, {
719+
objective: "New objective",
720+
});
721+
722+
resolveEvaluation?.({
723+
status: "stop",
724+
stopReason: "objective_complete",
725+
reason: "old objective is done",
726+
});
727+
728+
await waitFor(() => {
729+
const finished = manager.get(supervisor.id)?.cycles.find((entry) => entry.id === cycle.id);
730+
expect(finished?.status).toBe("completed");
731+
});
732+
733+
expect(targetMetaById.get(supervisor.targetId)).toMatchObject({
734+
status: "superseded",
735+
supersededBy: rotated.targetId,
736+
});
737+
expect(targetMetaById.get(rotated.targetId)).toMatchObject({
738+
status: "active",
739+
});
740+
});
741+
742+
it("rolls back a supersede mark when creating the next target files fails", async () => {
743+
const supervisor = await manager.create({
744+
sessionId: "sess-objective-create-fails",
745+
workspaceId: "ws-1",
746+
objective: "Initial objective",
747+
evaluatorProviderId: "codex",
748+
});
749+
750+
let latestMeta: Record<string, unknown> = {
751+
targetId: supervisor.targetId,
752+
sessionId: supervisor.sessionId,
753+
workspaceId: supervisor.workspaceId,
754+
objective: supervisor.objective,
755+
status: "active",
756+
createdAt: 1,
757+
updatedAt: 1,
758+
supersededBy: null,
759+
completedAt: null,
760+
};
761+
762+
deps.targetStore.readTargetMeta.mockImplementation(async () => ({ ...latestMeta }));
763+
deps.targetStore.markTargetSuperseded.mockImplementation(
764+
async (_workspacePath: string, targetId: string, nextTargetId: string, updatedAt: number) => {
765+
latestMeta = {
766+
...latestMeta,
767+
targetId,
768+
status: "superseded",
769+
supersededBy: nextTargetId,
770+
updatedAt,
771+
};
772+
}
773+
);
774+
deps.targetStore.saveTargetMeta.mockImplementation(
775+
async (_workspacePath: string, targetId: string, meta: Record<string, unknown>) => {
776+
latestMeta = { ...meta, targetId };
777+
}
778+
);
779+
780+
const createTargetFilesError = new Error("disk full");
781+
deps.targetStore.createTargetFiles.mockImplementation(async () => {
782+
throw createTargetFilesError;
783+
});
784+
785+
await expect(
786+
manager.update(supervisor.id, {
787+
objective: "New objective",
788+
})
789+
).rejects.toThrow("disk full");
790+
791+
expect(manager.get(supervisor.id)?.targetId).toBe(supervisor.targetId);
792+
expect(latestMeta).toMatchObject({
793+
targetId: supervisor.targetId,
794+
status: "active",
795+
supersededBy: null,
796+
});
797+
expect(deps.targetStore.saveTargetMeta).toHaveBeenCalledWith(
798+
expect.any(String),
799+
supervisor.targetId,
800+
expect.objectContaining({
801+
targetId: supervisor.targetId,
802+
status: "active",
803+
supersededBy: null,
804+
})
805+
);
806+
});
807+
625808
it("retries evaluator timeout up to the global retry budget", async () => {
626809
vi.useFakeTimers();
627810
deps.settingsRepo.get = vi.fn((key: string) => {
@@ -1049,6 +1232,42 @@ describe("SupervisorManager cycle triggers", () => {
10491232
expect(deps.logger.error).not.toHaveBeenCalled();
10501233
});
10511234

1235+
it("marks deleted active targets as cancelled with a completion timestamp", async () => {
1236+
const supervisor = await manager.create({
1237+
sessionId: "sess-delete-completed-at",
1238+
workspaceId: "ws-1",
1239+
objective: "Ship the fix",
1240+
evaluatorProviderId: "codex",
1241+
});
1242+
1243+
const activeMeta = {
1244+
targetId: supervisor.targetId,
1245+
sessionId: supervisor.sessionId,
1246+
workspaceId: supervisor.workspaceId,
1247+
objective: supervisor.objective,
1248+
status: "active",
1249+
createdAt: 1,
1250+
updatedAt: 1,
1251+
supersededBy: null,
1252+
completedAt: null,
1253+
};
1254+
deps.targetStore.readTargetMeta.mockResolvedValue(activeMeta);
1255+
1256+
await manager.delete(supervisor.id);
1257+
1258+
await waitFor(() => {
1259+
expect(deps.targetStore.saveTargetMeta).toHaveBeenCalledWith(
1260+
expect.any(String),
1261+
supervisor.targetId,
1262+
expect.objectContaining({
1263+
targetId: supervisor.targetId,
1264+
status: "cancelled",
1265+
completedAt: expect.any(Number),
1266+
})
1267+
);
1268+
});
1269+
});
1270+
10521271
it("pauses an in-flight evaluation by cancelling the cycle", async () => {
10531272
const supervisor = await manager.create({
10541273
sessionId: "sess-pause",

‎packages/server/src/supervisor/manager.ts‎

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -439,19 +439,32 @@ export class SupervisorManager {
439439

440440
if (objectiveChanged) {
441441
const nextTargetId = generateTargetId();
442+
const previousTargetMeta = await this.deps.targetStore.readTargetMeta(
443+
workspace.path,
444+
current.targetId
445+
);
442446
await this.deps.targetStore.markTargetSuperseded(
443447
workspace.path,
444448
current.targetId,
445449
nextTargetId,
446450
nextPatch.updatedAt ?? Date.now()
447451
);
448-
await this.deps.targetStore.createTargetFiles(workspace.path, {
449-
targetId: nextTargetId,
450-
sessionId: current.sessionId,
451-
workspaceId: current.workspaceId,
452-
objective: nextObjective,
453-
createdAt: nextPatch.updatedAt ?? Date.now(),
454-
});
452+
try {
453+
await this.deps.targetStore.createTargetFiles(workspace.path, {
454+
targetId: nextTargetId,
455+
sessionId: current.sessionId,
456+
workspaceId: current.workspaceId,
457+
objective: nextObjective,
458+
createdAt: nextPatch.updatedAt ?? Date.now(),
459+
});
460+
} catch (error) {
461+
await this.deps.targetStore.saveTargetMeta(
462+
workspace.path,
463+
current.targetId,
464+
previousTargetMeta
465+
);
466+
throw error;
467+
}
455468
nextPatch.targetId = nextTargetId;
456469
}
457470

@@ -1339,9 +1352,17 @@ export class SupervisorManager {
13391352
patch: Partial<Pick<SupervisorTargetMeta, "status" | "supersededBy" | "completedAt">>
13401353
): Promise<void> {
13411354
const current = await this.deps.targetStore.readTargetMeta(workspacePath, targetId);
1355+
const nextPatch =
1356+
current.status === "superseded" && patch.status && patch.status !== "superseded"
1357+
? {
1358+
...patch,
1359+
status: current.status,
1360+
supersededBy: current.supersededBy,
1361+
}
1362+
: patch;
13421363
await this.deps.targetStore.saveTargetMeta(workspacePath, targetId, {
13431364
...current,
1344-
...patch,
1365+
...nextPatch,
13451366
updatedAt: Date.now(),
13461367
});
13471368
}
@@ -1375,7 +1396,7 @@ export class SupervisorManager {
13751396
}
13761397
await this.updateTargetMetaStatus(workspacePath, supervisor.targetId, {
13771398
status: "cancelled",
1378-
completedAt: meta.completedAt,
1399+
completedAt: meta.completedAt ?? Date.now(),
13791400
});
13801401
}
13811402

0 commit comments

Comments
 (0)