Sessions keep running across navigation: turn liveness + write-ahead persist
The bug: answering a quiz and navigating away mid-turn, then returning, re-prompted the same question. Root causes: (1) the answer was durable only at onFinish — every mid-turn view read the stale pre-answer record; (2) the server had no notion of a turn in flight; (3) remount treated the stale record as current, with no re-attach path. Worse, the stale dock let the quiz be answered again, racing two turns into a last-writer-wins saveMessages clobber. Fix (four pieces): - Write-ahead persist in chat-handler: the request body IS the client's canonical state — saved to disk BEFORE the stream starts. The answer is durable the instant the request arrives; crashes recover via the existing mount continuation. - Turn liveness registry (src/server/turn-liveness.ts): in-memory chatId → in-flight map with a TTL leak guard; beginTurn rejects concurrent turns (409) so one turn ever owns a chat. endTurn runs in onFinish AFTER persistTurn, so pollers that see not-in-flight always read a finished record; a turn-start failure clears the registry (no lockout). - GET /api/chats/[id] reports turnInFlight; the chat page passes initiallyInFlight to the client. - Client re-attach (components/chat.tsx): while in flight, busy shimmer + composer gated + mount continuation suppressed; poll /api/chats/[id] until the turn finishes, then syncFromServer adopts the completed transcript. Stop stays local-stream only (a server turn has no client request to abort). Verified by scripted-model tests: concurrent turn 409 + registry lifecycle, write-ahead durability mid-turn, failed-turn lockout guard.
This commit is contained in:
@@ -4,6 +4,7 @@ import { createJobRunner } from "../agent/subagents/runner.ts";
|
||||
import { createExtractionPipeline } from "../agent/subagents/extraction.ts";
|
||||
import { createTeacherTools } from "../agent/tools.ts";
|
||||
import { stores } from "./stores.ts";
|
||||
import { beginTurn, endTurn } from "./turn-liveness.ts";
|
||||
import { buildTeacherContext, persistTurn, spawnTurnJobs, triggerExtraction } from "./turn-service.ts";
|
||||
|
||||
/**
|
||||
@@ -36,7 +37,29 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro
|
||||
return Response.json({ error: "messages must be an array" }, { status: 400 });
|
||||
}
|
||||
|
||||
return createAgentUIStreamResponse({
|
||||
// One turn per chat, ever. A second request while a turn runs means a
|
||||
// stale/remounted client raced a live one — its message list is outdated,
|
||||
// and letting it through would double-run the teacher and clobber the
|
||||
// first turn's persist (saveMessages is last-writer-wins).
|
||||
if (!beginTurn(chatId)) {
|
||||
console.info(`[chat ${chatId}] turn already in flight — request rejected`);
|
||||
return Response.json({ error: "A turn is already running for this session" }, { status: 409 });
|
||||
}
|
||||
|
||||
// Write-ahead persistence: the request body IS the client's canonical
|
||||
// state, so it goes on disk BEFORE the stream starts. The learner's answer
|
||||
// is durable the instant the request arrives — any remount, second tab, or
|
||||
// server crash sees truth instead of the pre-answer record. (A client that
|
||||
// returns mid-turn reads turnInFlight from the registry and re-attaches by
|
||||
// polling instead of resurrecting the stale transcript.)
|
||||
try {
|
||||
stores.chatStore.saveMessages(chatId, body.messages as UIMessage[]);
|
||||
} catch (err) {
|
||||
console.error(`[chat ${chatId}] write-ahead persist failed:`, err);
|
||||
}
|
||||
|
||||
try {
|
||||
return await createAgentUIStreamResponse({
|
||||
agent:
|
||||
agent ??
|
||||
createTeacherAgent(
|
||||
@@ -56,10 +79,20 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro
|
||||
if (InvalidToolInputError.isInstance(error) || NoSuchToolError.isInstance(error)) return error.message;
|
||||
return "An error occurred.";
|
||||
},
|
||||
onFinish: ({ messages, isAborted }) => {
|
||||
persistTurn(stores.chatStore, chatId, messages as UIMessage[], isAborted);
|
||||
spawnTurnJobs(jobRunner, chatId, messages as UIMessage[]);
|
||||
triggerExtraction(stores, extraction, chatId);
|
||||
},
|
||||
});
|
||||
onFinish: ({ messages, isAborted }) => {
|
||||
persistTurn(stores.chatStore, chatId, messages as UIMessage[], isAborted);
|
||||
spawnTurnJobs(jobRunner, chatId, messages as UIMessage[]);
|
||||
triggerExtraction(stores, extraction, chatId);
|
||||
// LAST: a poller that sees not-in-flight must read an already-persisted
|
||||
// record — endTurn only after persistTurn has done its job.
|
||||
endTurn(chatId);
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
// The stream never started, so no onFinish will fire — clear the registry
|
||||
// or a malformed request would lock the chat until the TTL eviction.
|
||||
endTurn(chatId);
|
||||
console.error(`[chat ${chatId}] turn failed to start:`, err);
|
||||
return Response.json({ error: "Failed to start the turn" }, { status: 500 });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,3 +9,4 @@ export * from "./chats.ts";
|
||||
export * from "./graph.ts";
|
||||
export * from "./jobs.ts";
|
||||
export * from "./stores.ts";
|
||||
export * from "./turn-liveness.ts";
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
/**
|
||||
* Server-side turn liveness: which chats currently have a teacher turn in
|
||||
* flight. The missing half of "sessions keep running in the background" —
|
||||
* together with the write-ahead persist in chat-handler, a returning client
|
||||
* can tell "the teacher is mid-turn — don't resurrect, don't start anything"
|
||||
* apart from "the record IS current".
|
||||
*
|
||||
* In-memory BY DESIGN: a process restart clears it, so there are no stale
|
||||
* flags to clean up — a crashed turn recovers via the write-ahead record +
|
||||
* mount continuation on the next visit. The TTL is a leak guard for the
|
||||
* pathological case where onFinish never fires (it runs even on aborts);
|
||||
* chat turns are minutes at most, so 30 minutes is far on the safe side.
|
||||
*/
|
||||
|
||||
const inFlight = new Map<string, number>(); // chatId → startedAt (epoch ms)
|
||||
const MAX_TURN_MS = 30 * 60_000;
|
||||
|
||||
/**
|
||||
* Mark a turn as starting. Returns false if one is already running for the
|
||||
* chat — the caller rejects the request (only one turn may own a chat).
|
||||
*/
|
||||
export function beginTurn(chatId: string): boolean {
|
||||
if (isTurnRunning(chatId)) return false;
|
||||
inFlight.set(chatId, Date.now());
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Mark the chat's turn as finished (runs in onFinish, aborted or not). */
|
||||
export function endTurn(chatId: string): void {
|
||||
inFlight.delete(chatId);
|
||||
}
|
||||
|
||||
/** True while the chat's turn is in flight; lazily evicts stale entries. */
|
||||
export function isTurnRunning(chatId: string): boolean {
|
||||
const since = inFlight.get(chatId);
|
||||
if (since === undefined) return false;
|
||||
if (Date.now() - since > MAX_TURN_MS) {
|
||||
inFlight.delete(chatId);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
Reference in New Issue
Block a user