Skip to content

Guard provider lifecycle and checkpoints to primary thread/active turn #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 4 commits into from
Feb 27, 2026
Merged

Guard provider lifecycle and checkpoints to primary thread/active turn #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d19b2f5
Guard provider lifecycle updates to primary thread and active turn
juliusmarminge Feb 27, 2026
7dab240
Inline provider thread and turn ID schemas in runtime events
juliusmarminge Feb 27, 2026
5b13d20
Handle provider turn IDs correctly in runtime ingestion
juliusmarminge Feb 27, 2026
4e3e689
Replace thread latestTurnId with structured latestTurn metadata
juliusmarminge Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions apps/server/integration/orchestrationEngine.integration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,7 @@ it.live("runs multi-turn file edits and persists checkpoint diffs", () =>
const secondTurnThread = yield* harness.waitForThread(
THREAD_ID,
(entry) =>
entry.latestTurnId === "turn-2" &&
entry.latestTurn?.turnId === "turn-2" &&
entry.checkpoints.length === 2 &&
entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 2),
);
Expand Down Expand Up @@ -653,7 +653,7 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git
yield* harness.waitForThread(
THREAD_ID,
(entry) =>
entry.latestTurnId === "turn-2" &&
entry.latestTurn?.turnId === "turn-2" &&
entry.checkpoints.length === 2 &&
entry.activities.some((activity) => activity.turnId === "turn-2"),
8000,
Expand Down
9 changes: 8 additions & 1 deletion apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,14 @@ function makeSnapshot(input: {
model: "gpt-5-codex",
branch: null,
worktreePath: input.worktreePath,
latestTurnId: TurnId.makeUnsafe("turn-1"),
latestTurn: {
turnId: TurnId.makeUnsafe("turn-1"),
state: "completed",
requestedAt: "2026-01-01T00:00:00.000Z",
startedAt: "2026-01-01T00:00:00.000Z",
completedAt: "2026-01-01T00:00:00.000Z",
assistantMessageId: null,
},
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
deletedAt: null,
Expand Down
99 changes: 88 additions & 11 deletions apps/server/src/orchestration/Layers/CheckpointReactor.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
ProjectId,
ProviderSessionId,
ProviderThreadId,
ProviderTurnId,
ThreadId,
TurnId,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -40,6 +41,7 @@ import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";
const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value);
const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);

function createProviderServiceHarness(cwd: string, hasSession = true, sessionCwd = cwd) {
Expand Down Expand Up @@ -91,15 +93,15 @@ function createProviderServiceHarness(cwd: string, hasSession = true, sessionCwd
async function waitForThread(
engine: OrchestrationEngineShape,
predicate: (thread: {
latestTurnId: string | null;
latestTurn: { turnId: string } | null;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;
activities: ReadonlyArray<{ kind: string }>;
}) => boolean,
timeoutMs = 2000,
) {
const deadline = Date.now() + timeoutMs;
const poll = async (): Promise<{
latestTurnId: string | null;
latestTurn: { turnId: string } | null;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;
activities: ReadonlyArray<{ kind: string }>;
}> => {
Expand Down Expand Up @@ -334,7 +336,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-1"),
turnId: asProviderTurnId("turn-1"),
});
await waitForGitRefExists(
harness.cwd,
Expand All @@ -349,14 +351,14 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-1"),
turnId: asProviderTurnId("turn-1"),
status: "completed",
});

await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed");
const thread = await waitForThread(
harness.engine,
(entry) => entry.latestTurnId === "turn-1" && entry.checkpoints.length === 1,
(entry) => entry.latestTurn?.turnId === "turn-1" && entry.checkpoints.length === 1,
);
expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
expect(
Expand All @@ -381,6 +383,81 @@ describe("CheckpointReactor", () => {
).toBe("v2\n");
});

it("ignores auxiliary thread turn completion while primary turn is active", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const createdAt = new Date().toISOString();

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.makeUnsafe("cmd-session-set-primary-running"),
threadId: ThreadId.makeUnsafe("thread-1"),
session: {
threadId: ThreadId.makeUnsafe("thread-1"),
status: "running",
providerName: "codex",
providerSessionId: asSessionId("sess-1"),
providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
approvalPolicy: "on-request",
sandboxMode: "workspace-write",
activeTurnId: asTurnId("turn-main"),
lastError: null,
updatedAt: createdAt,
},
createdAt,
}),
);

harness.provider.emit({
type: "turn.started",
eventId: EventId.makeUnsafe("evt-turn-started-main"),
provider: "codex",
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asProviderTurnId("turn-main"),
});
await waitForGitRefExists(
harness.cwd,
checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0),
);

fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8");

harness.provider.emit({
type: "turn.completed",
eventId: EventId.makeUnsafe("evt-turn-completed-aux"),
provider: "codex",
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-aux"),
turnId: asProviderTurnId("turn-aux"),
status: "completed",
});

await Effect.runPromise(Effect.sleep("40 millis"));
const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
expect(midThread?.checkpoints).toHaveLength(0);

Comment on lines +438 to +442

@coderabbitai coderabbitai Bot Feb 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Remove timing-based mid-assertion to avoid flaky/false-positive behavior.

Line 438 uses a fixed sleep, and Line 441 can pass before the auxiliary event is actually processed. This makes the test nondeterministic under load.

♻️ Suggested test hardening 
-    await Effect.runPromise(Effect.sleep("40 millis"));
-    const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
-    const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
-    expect(midThread?.checkpoints).toHaveLength(0);

Based on learnings: "For integration tests: Prefer deterministic inputs and explicit state checks; avoid relying on logs or timing assumptions."

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@apps/server/src/orchestration/Layers/CheckpointReactor.test.ts` around lines
438 - 442, Remove the timing-based mid-assertion that uses
Effect.runPromise(Effect.sleep(...)) and the immediate read-model check via
harness.engine.getReadModel(); instead synchronize deterministically: either
remove the mid-assertion entirely or replace the sleep+check with a wait-for
pattern that polls harness.engine.getReadModel() (or a harness-provided
processing-complete hook) until the expected state is observed with a bounded
timeout. Locate the usage of Effect.runPromise(Effect.sleep(...)), the
subsequent call to harness.engine.getReadModel(), and
ThreadId.makeUnsafe("thread-1") in the test and change the logic to
poll/getReadModel repeatedly (with a short interval and overall timeout) or use
a deterministic hook so the test does not rely on fixed sleep timing.

harness.provider.emit({
type: "turn.completed",
eventId: EventId.makeUnsafe("evt-turn-completed-main"),
provider: "codex",
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asProviderTurnId("turn-main"),
status: "completed",
});

const thread = await waitForThread(
harness.engine,
(entry) => entry.latestTurn?.turnId === "turn-main" && entry.checkpoints.length === 1,
);
expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
});

it("appends capture failure activity when turn diff summary cannot be derived", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const createdAt = new Date().toISOString();
Expand Down Expand Up @@ -413,7 +490,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-missing-baseline"),
turnId: asProviderTurnId("turn-missing-baseline"),
status: "completed",
});

Expand Down Expand Up @@ -505,7 +582,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-missing"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-missing"),
turnId: asTurnId("turn-missing-cwd"),
turnId: asProviderTurnId("turn-missing-cwd"),
status: "completed",
});

Expand Down Expand Up @@ -554,7 +631,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-3"),
turnId: asProviderTurnId("turn-3"),
turnCount: 3,
status: "completed",
});
Expand Down Expand Up @@ -607,7 +684,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-runtime-failure"),
turnId: asProviderTurnId("turn-runtime-failure"),
turnCount: 1,
status: "completed",
});
Expand All @@ -619,7 +696,7 @@ describe("CheckpointReactor", () => {
sessionId: asSessionId("sess-1"),
createdAt: new Date().toISOString(),
threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
turnId: asTurnId("turn-after-runtime-failure"),
turnId: asProviderTurnId("turn-after-runtime-failure"),
});

await waitForGitRefExists(
Expand Down Expand Up @@ -698,7 +775,7 @@ describe("CheckpointReactor", () => {
await waitForEvent(harness.engine, (event) => event.type === "thread.reverted");
const thread = await waitForThread(harness.engine, (entry) => entry.checkpoints.length === 1);

expect(thread.latestTurnId).toBe("turn-1");
expect(thread.latestTurn?.turnId).toBe("turn-1");
expect(thread.checkpoints).toHaveLength(1);
expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
expect(harness.provider.rollbackConversation).toHaveBeenCalledTimes(1);
Expand Down
27 changes: 27 additions & 0 deletions apps/server/src/orchestration/Layers/CheckpointReactor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import {
EventId,
MessageId,
ProviderSessionId,
ProviderThreadId,
ThreadId,
TurnId,
type OrchestrationEvent,
Expand Down Expand Up @@ -36,6 +37,17 @@ function toTurnId(value: string | undefined): TurnId | null {
return value === undefined ? null : TurnId.makeUnsafe(value);
}

function toProviderThreadId(value: string | undefined): ProviderThreadId | null {
return value === undefined ? null : ProviderThreadId.makeUnsafe(value);
}

function sameId(left: string | null | undefined, right: string | null | undefined): boolean {
if (left === null || left === undefined || right === null || right === undefined) {
return false;
}
return left === right;
}

function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" {
switch (status) {
case "failed":
Expand Down Expand Up @@ -173,6 +185,21 @@ const make = Effect.gen(function* () {
return;
}

const projectedProviderThreadId = thread.session?.providerThreadId ?? null;
const eventProviderThreadId = toProviderThreadId(event.threadId);
if (
projectedProviderThreadId !== null &&
eventProviderThreadId !== null &&
!sameId(projectedProviderThreadId, eventProviderThreadId)
) {
return;
}

// When a primary turn is active, only that turn may produce completion checkpoints.
if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {

@macroscopeapp macroscopeapp Bot Feb 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium Layers/CheckpointReactor.ts:199

Race condition: activeTurnId is read from the live read model, but if a new turn starts before this turn.completed event is processed, activeTurnId will have advanced, causing this check to incorrectly skip checkpoint capture for the completed turn. Consider removing this guard or comparing against the turn that was active when the completion event was emitted.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file apps/server/src/orchestration/Layers/CheckpointReactor.ts around line 199:

Race condition: `activeTurnId` is read from the live read model, but if a new turn starts before this `turn.completed` event is processed, `activeTurnId` will have advanced, causing this check to incorrectly skip checkpoint capture for the completed turn. Consider removing this guard or comparing against the turn that was active when the completion event was emitted.

Evidence trail:
apps/server/src/orchestration/Layers/CheckpointReactor.ts lines 180-202 (REVIEWED_COMMIT): Shows `getReadModel()` call at line 180, the guard at lines 199-201 comparing `thread.session?.activeTurnId` against `turnId` from the event. apps/server/src/orchestration/Layers/OrchestrationEngine.ts lines 219-220 (REVIEWED_COMMIT): Shows `getReadModel` implementation returning direct reference to the live `readModel` object via `Effect.sync((): OrchestrationReadModel => readModel)`. apps/server/src/orchestration/Layers/CheckpointReactor.ts lines 175-176: Shows `turnId` is extracted from the `turn.completed` event parameter.

return;
}

if (thread.checkpoints.some((checkpoint) => checkpoint.turnId === turnId)) {
return;
}
Expand Down
9 changes: 8 additions & 1 deletion apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,14 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
model: "gpt-5-codex",
branch: null,
worktreePath: null,
latestTurnId: asTurnId("turn-1"),
latestTurn: {
turnId: asTurnId("turn-1"),
state: "completed",
requestedAt: "2026-02-24T00:00:08.000Z",
startedAt: "2026-02-24T00:00:08.000Z",
completedAt: "2026-02-24T00:00:08.000Z",
assistantMessageId: asMessageId("message-1"),
},
createdAt: "2026-02-24T00:00:02.000Z",
updatedAt: "2026-02-24T00:00:03.000Z",
deletedAt: null,
Expand Down
Loading