mirror of
https://github.com/cline/cline.git
synced 2026-10-24 02:02:27 +08:00
fix(core): serialize session status writes so teardown's status lands last
A non-terminal status write (running from markTurnRunning, pending from a tool approval) that is already in flight when teardown starts has passed the shutting-down guard in updateStatus. The persistence layer retries a write whose status lock went stale, so when that write overlaps with teardown's cancelled write it re-reads the row and writes running on top of it, and the stopped session stays recorded as running. Chain status writes per session on the same promise queue the compaction state already uses, so teardown's write waits for the in-flight one; the guard moves to the point where a write actually starts, so a non-terminal write queued behind teardown's is still skipped.
This commit is contained in:
@@ -3818,15 +3818,20 @@ describe("LocalRuntimeHost", () => {
|
||||
await runningStatusHeld;
|
||||
|
||||
const stopping = manager.stopSession(sessionId);
|
||||
// Teardown must wait for the in-flight running write: the persistence
|
||||
// layer retries a write whose status lock went stale, so a cancelled
|
||||
// written underneath it would be overwritten with running.
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
const statuses = () =>
|
||||
sessionService.updateSessionStatus.mock.calls.map((call) => call[1]);
|
||||
expect(statuses()).toEqual(["running", "idle", "running"]);
|
||||
releaseRunningStatus?.();
|
||||
await stopping;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
|
||||
expect(sentPrompts).toHaveLength(1);
|
||||
expect(endedEvents).toBe(1);
|
||||
expect(
|
||||
sessionService.updateSessionStatus.mock.calls.map((call) => call[1]),
|
||||
).toEqual(["running", "idle", "running", "cancelled"]);
|
||||
expect(statuses()).toEqual(["running", "idle", "running", "cancelled"]);
|
||||
});
|
||||
|
||||
it("clears the remaining queue when a queue-initiated turn is aborted", async () => {
|
||||
|
||||
@@ -1676,18 +1676,30 @@ export class LocalRuntimeHost implements RuntimeHost {
|
||||
session: ActiveSession,
|
||||
action: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const previous = session.compactionStateWriteQueue ?? Promise.resolve();
|
||||
return this.enqueueSessionWrite(
|
||||
session,
|
||||
"compactionStateWriteQueue",
|
||||
action,
|
||||
);
|
||||
}
|
||||
|
||||
private async enqueueSessionWrite<T>(
|
||||
session: ActiveSession,
|
||||
queue: "compactionStateWriteQueue" | "statusWriteQueue",
|
||||
action: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const previous = session[queue] ?? Promise.resolve();
|
||||
const run = previous.catch(() => undefined).then(action);
|
||||
const tracked = run.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
session.compactionStateWriteQueue = tracked;
|
||||
session[queue] = tracked;
|
||||
try {
|
||||
return await run;
|
||||
} finally {
|
||||
if (session.compactionStateWriteQueue === tracked) {
|
||||
session.compactionStateWriteQueue = undefined;
|
||||
if (session[queue] === tracked) {
|
||||
session[queue] = undefined;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2668,6 +2680,21 @@ export class LocalRuntimeHost implements RuntimeHost {
|
||||
exitCode?: number | null,
|
||||
): Promise<void> {
|
||||
if (!session.artifacts) return;
|
||||
// Status writes are serialized per session so teardown's terminal
|
||||
// status lands after a non-terminal write that was already in flight
|
||||
// when teardown started. The persistence layer retries a write whose
|
||||
// status lock went stale, so an overlapping earlier write would
|
||||
// otherwise win.
|
||||
return this.enqueueSessionWrite(session, "statusWriteQueue", () =>
|
||||
this.writeStatus(session, status, exitCode),
|
||||
);
|
||||
}
|
||||
|
||||
private async writeStatus(
|
||||
session: ActiveSession,
|
||||
status: SessionStatus,
|
||||
exitCode?: number | null,
|
||||
): Promise<void> {
|
||||
// Turns aborted by teardown settle concurrently with it; their idle or
|
||||
// running transitions must not overwrite the status teardown persists.
|
||||
if (session.shuttingDown && isNonTerminalSessionStatus(status)) return;
|
||||
|
||||
@@ -33,6 +33,7 @@ export type ActiveSession = {
|
||||
persistedMessages?: LlmsProviders.MessageWithMetadata[];
|
||||
compactionState?: SessionCompactionState;
|
||||
compactionStateWriteQueue?: Promise<void>;
|
||||
statusWriteQueue?: Promise<void>;
|
||||
activeTeamRunIds: Set<string>;
|
||||
pendingTeamRunUpdates: TeamRunUpdate[];
|
||||
teamRunWaiters: Array<() => void>;
|
||||
|
||||
Reference in New Issue
Block a user