Skip to content

[muse] fix(server): recompute checkpoint turn count after slow git capture #5

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

Open
kvnloo wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[muse] fix(server): recompute checkpoint turn count after slow git capture #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
112 changes: 112 additions & 0 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 @@ -295,6 +295,10 @@ describe("CheckpointReactor", () => {
readonly checkpointLookupFailure?: (
cwd: string,
) => VcsProcessTimeoutError | VcsProcessSpawnError | undefined;
readonly captureCheckpointHook?: (input: {
readonly cwd: string;
readonly checkpointRef: CheckpointRef;
}) => Effect.Effect<void>;
readonly workspaceRefresh?: (cwd: string) => Effect.Effect<void>;
readonly hasSession?: boolean;
readonly seedFilesystemCheckpoints?: boolean;
Expand Down Expand Up @@ -386,6 +390,10 @@ describe("CheckpointReactor", () => {
const failure = options?.checkpointLookupFailure?.(input.cwd);
return failure ? Effect.fail(failure) : store.hasCheckpointRef(input);
},
captureCheckpoint: (input) =>
(options?.captureCheckpointHook?.(input) ?? Effect.void).pipe(
Effect.andThen(() => store.captureCheckpoint(input)),
),
})),
),
).pipe(Layer.provide(VcsDriverRegistry.layer)),
Expand Down Expand Up @@ -857,6 +865,110 @@ describe("CheckpointReactor", () => {
}),
);

effectIt.effect("recomputes the turn count when a concurrent turn claims it mid-capture", () =>
Effect.gen(function* () {
const createdAt = "2026-01-01T00:00:00.000Z";
const threadId = ThreadId.make("thread-1");
let harness: Awaited<ReturnType<typeof createHarness>> | undefined;
let placeholderInjected = false;
const inner = yield* Effect.promise(() =>
createHarness({
seedFilesystemCheckpoints: false,
captureCheckpointHook: (input) =>
Effect.gen(function* () {
// Only the completion capture for turn-1 races: while its slow
// git capture runs, a provider diff for turn-2 commits a
// placeholder claiming turn count 1.
if (placeholderInjected || !String(input.checkpointRef).endsWith("/turn/1")) {
return;
}
placeholderInjected = true;
if (harness === undefined) {
return yield* Effect.die(
new Error("checkpoint hook ran before the harness was ready"),
);
}
yield* harness.engine.dispatch({
type: "thread.turn.diff.complete",
commandId: CommandId.make("cmd-race-placeholder"),
threadId,
turnId: asTurnId("turn-2"),
completedAt: createdAt,
checkpointRef: CheckpointRef.make("provider-diff:evt-race"),
status: "missing",
files: [],
assistantMessageId: MessageId.make("assistant:turn-2"),
checkpointTurnCount: 1,
createdAt,
});
}),
}),
);
harness = inner;

yield* harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-race"),
threadId,
session: {
threadId,
status: "ready",
providerName: "codex",
runtimeMode: "approval-required",
activeTurnId: null,
lastError: null,
updatedAt: createdAt,
},
createdAt,
});

harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-race-started-1"),
provider: ProviderDriverKind.make("codex"),
createdAt,
threadId,
turnId: asTurnId("turn-1"),
});
expect(yield* harness.nextReceipt).toMatchObject({
type: "checkpoint.baseline.captured",
checkpointTurnCount: 0,
});
yield* Effect.promise(harness.drain);

NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "v2\n", "utf8");
harness.provider.emit({
type: "turn.completed",
eventId: EventId.make("evt-race-completed-1"),
provider: ProviderDriverKind.make("codex"),
createdAt,
threadId,
turnId: asTurnId("turn-1"),
payload: { state: "completed" },
});

expect(yield* harness.nextReceipt).toMatchObject({
type: "checkpoint.diff.finalized",
turnId: "turn-1",
checkpointTurnCount: 2,
});
yield* Effect.promise(harness.drain);
const thread = (yield* Effect.promise(harness.readModel)).threads.find(
(entry) => entry.id === "thread-1",
);
const counts = (thread?.checkpoints ?? []).map((entry) => entry.checkpointTurnCount);
expect(counts).toContain(1);
expect(counts).toContain(2);
expect(new Set(counts).size).toBe(counts.length);
expect(
thread?.checkpoints.find((entry) => entry.checkpointTurnCount === 2)?.turnId,
).toBe("turn-1");
expect(
gitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 2)),
).toBe(true);
}),
);

effectIt.effect("captures and reverts checkpoints from a nested Git workspace", () =>
Effect.gen(function* () {
const repositoryRoot = createGitRepository();
Expand Down
103 changes: 73 additions & 30 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 @@ -7,6 +7,7 @@ import {
ThreadId,
TurnId,
type OrchestrationEvent,
type OrchestrationCheckpointSummary,
type ProviderRuntimeEvent,
type VcsStatusLocalResult,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -239,6 +240,36 @@ const make = Effect.gen(function* () {
return cwd;
});

// The placeholder checkpoint for a turn, if provider diff ingestion recorded
// one mid-turn. Captures reuse its turn count instead of incrementing past it.
const findTurnPlaceholder = (
checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>,
turnId: TurnId,
) =>
checkpoints.find(
(checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",
);

// The next checkpoint turn count for a turn: reuse the turn's placeholder
// count when one exists, otherwise one past the current maximum. Computed
// both before the (slow) git capture and again after it, so a concurrent
// turn that committed in between gets a fresh count instead of a collision.
const nextCheckpointTurnCount = (
checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>,
turnId: TurnId,
) => {
const existingPlaceholder = findTurnPlaceholder(checkpoints, turnId);
if (existingPlaceholder !== undefined) {
return existingPlaceholder.checkpointTurnCount;
}
return (
checkpoints.reduce(
(maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
0,
) + 1
);
};

// Capture the completed turn's files, then publish its summary and receipts.
const captureAndDispatchCheckpoint = Effect.fn("captureAndDispatchCheckpoint")(function* (input: {
readonly threadId: ThreadId;
Expand All @@ -256,9 +287,41 @@ const make = Effect.gen(function* () {
readonly assistantMessageId: MessageId | undefined;
readonly createdAt: string;
}) {
const fromTurnCount = Math.max(0, input.turnCount - 1);
const checkpointRefForTurnCount = (turnCount: number) =>
checkpointRefForThreadTurn(input.threadId, turnCount);

yield* checkpointStore.captureCheckpoint({
cwd: input.cwd,
checkpointRef: checkpointRefForTurnCount(input.turnCount),
});

// Refresh the workspace entry index so the @-mention file picker
// reflects files created or deleted during this turn.
yield* refreshWorkspaceEntries(input.cwd);

// The git capture above is slow: a concurrent turn may have claimed the
// same count while it ran. Recompute from a fresh read and re-capture at
// the fresh ref so two turns never share a count (and its git ref). The
// decider rejects any residual collision at commit time.
const freshCheckpointContext = yield* projectionSnapshotQuery
.getThreadCheckpointContext(input.threadId)
.pipe(Effect.map(Option.getOrUndefined));
const freshTurnCount = nextCheckpointTurnCount(
freshCheckpointContext?.checkpoints ?? [],
input.turnId,
);
let turnCount = input.turnCount;
if (freshTurnCount !== turnCount) {
turnCount = freshTurnCount;
yield* checkpointStore.captureCheckpoint({
cwd: input.cwd,
checkpointRef: checkpointRefForTurnCount(turnCount),
});
}

const fromTurnCount = Math.max(0, turnCount - 1);
const fromCheckpointRef = checkpointRefForThreadTurn(input.threadId, fromTurnCount);
const targetCheckpointRef = checkpointRefForThreadTurn(input.threadId, input.turnCount);
const targetCheckpointRef = checkpointRefForTurnCount(turnCount);

const fromCheckpointExists = yield* checkpointStore
.hasCheckpointRef({
Expand All @@ -282,15 +345,6 @@ const make = Effect.gen(function* () {
});
}

yield* checkpointStore.captureCheckpoint({
cwd: input.cwd,
checkpointRef: targetCheckpointRef,
});

// Refresh the workspace entry index so the @-mention file picker
// reflects files created or deleted during this turn.
yield* refreshWorkspaceEntries(input.cwd);

// Git may have been initialized during this turn, leaving no pre-turn
// snapshot. Keep the completion checkpoint for future turns, but do not
// invent a baseline or attempt a diff against a ref that does not exist.
Expand Down Expand Up @@ -326,7 +380,7 @@ const make = Effect.gen(function* () {
Effect.logWarning("failed to derive checkpoint file summary", {
threadId: input.threadId,
turnId: input.turnId,
turnCount: input.turnCount,
turnCount,
detail: error.message,
}).pipe(Effect.as([])),
),
Expand All @@ -349,14 +403,14 @@ const make = Effect.gen(function* () {
status: input.status,
files,
assistantMessageId,
checkpointTurnCount: input.turnCount,
checkpointTurnCount: turnCount,
createdAt: input.createdAt,
});
yield* receiptBus.publish({
type: "checkpoint.diff.finalized",
threadId: input.threadId,
turnId: input.turnId,
checkpointTurnCount: input.turnCount,
checkpointTurnCount: turnCount,
checkpointRef: targetCheckpointRef,
status: input.status,
createdAt: input.createdAt,
Expand All @@ -365,7 +419,7 @@ const make = Effect.gen(function* () {
type: "turn.processing.quiesced",
threadId: input.threadId,
turnId: input.turnId,
checkpointTurnCount: input.turnCount,
checkpointTurnCount: turnCount,
createdAt: input.createdAt,
});

Expand All @@ -379,7 +433,7 @@ const make = Effect.gen(function* () {
kind: "checkpoint.captured",
summary: "Checkpoint captured",
payload: {
turnCount: input.turnCount,
turnCount,
status: input.status,
},
turnId: input.turnId,
Expand Down Expand Up @@ -429,19 +483,7 @@ const make = Effect.gen(function* () {
return;
}

// If a placeholder checkpoint exists for this turn, reuse its turn count
// instead of incrementing past it.
const existingPlaceholder = thread.checkpoints.find(
(checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",
);
const currentTurnCount = thread.checkpoints.reduce(
(maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
0,
);
const nextTurnCount = existingPlaceholder
? existingPlaceholder.checkpointTurnCount
: currentTurnCount + 1;

const nextTurnCount = nextCheckpointTurnCount(thread.checkpoints, turnId);
yield* captureAndDispatchCheckpoint({
threadId: thread.id,
turnId,
Expand All @@ -452,7 +494,8 @@ const make = Effect.gen(function* () {
event.type === "turn.aborted"
? "ready"
: checkpointStatusFromRuntime(event.payload.state),
assistantMessageId: existingPlaceholder?.assistantMessageId ?? undefined,
assistantMessageId:
findTurnPlaceholder(thread.checkpoints, turnId)?.assistantMessageId ?? undefined,
createdAt: event.createdAt,
});
},
Expand Down
18 changes: 18 additions & 0 deletions apps/server/src/orchestration/decider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2118,6 +2118,24 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
detail: `turn ${command.turnId} already has a captured checkpoint`,
});
}
// checkpointTurnCount doubles as the deterministic git ref suffix, so two
// turns must never share one. The dispatching sides compute it as max+1
// from a read taken before their (sometimes slow) capture, so a newer
// turn's placeholder can claim the count an older turn's capture is
// about to commit. The decider runs under the engine's command lock, so
// rejecting here closes that window; the loser recomputes from a fresh
// read and retries.
const turnCountClaimant = thread.checkpoints.find(
(checkpoint) =>
checkpoint.turnId !== command.turnId &&
checkpoint.checkpointTurnCount === command.checkpointTurnCount,
);
if (turnCountClaimant !== undefined) {
return yield* new OrchestrationCommandInvariantError({
commandType: command.type,
detail: `checkpoint turn count ${command.checkpointTurnCount} already claimed by turn ${turnCountClaimant.turnId}`,
});
}
return {
...(yield* withEventBase({
aggregateKind: "thread",
Expand Down
35 changes: 35 additions & 0 deletions apps/server/src/orchestration/decider.turnDiffComplete.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,4 +124,39 @@ it.layer(NodeServices.layer)("turn diff complete decider", (it) => {
expect(events[0]?.type).toBe("thread.turn-diff-completed");
}),
);

it.effect("rejects a checkpoint when another turn already claimed its turn count", () =>
Effect.gen(function* () {
const exit = yield* Effect.exit(
decideOrchestrationCommand({
command: {
...placeholderCommand(),
checkpointRef: CheckpointRef.make("refs/t3/checkpoints/turn-1"),
checkpointTurnCount: 1,
status: "ready",
},
readModel: makeReadModel([
{ ...makeCheckpoint("missing"), turnId: TurnId.make("turn-2") },
]),
}),
);
expect(Exit.isFailure(exit)).toBe(true);
}),
);

it.effect("accepts a capture that reuses its own turn's placeholder turn count", () =>
Effect.gen(function* () {
const event = yield* decideOrchestrationCommand({
command: {
...placeholderCommand(),
checkpointRef: CheckpointRef.make("refs/t3/checkpoints/turn-1"),
checkpointTurnCount: 1,
status: "ready",
},
readModel: makeReadModel([makeCheckpoint("missing")]),
});
const events = Array.isArray(event) ? event : [event];
expect(events[0]?.type).toBe("thread.turn-diff-completed");
}),
);
});
Loading