diff --git a/app/api/chats/[id]/route.ts b/app/api/chats/[id]/route.ts index 5559ec5..5938fd8 100644 --- a/app/api/chats/[id]/route.ts +++ b/app/api/chats/[id]/route.ts @@ -1,8 +1,10 @@ -import { getChat } from "@/src/server"; +import { getChat, isTurnRunning } from "@/src/server"; export async function GET(_req: Request, { params }: { params: Promise<{ id: string }> }) { const { id } = await params; const chat = getChat(id); if (!chat) return Response.json({ error: "Not found" }, { status: 404 }); - return Response.json(chat); + // turnInFlight lets a returning client re-attach to a running turn + // (chat-handler write-ahead + registry — see src/server/turn-liveness.ts). + return Response.json({ ...chat, turnInFlight: isTurnRunning(id) }); } diff --git a/app/chat/[id]/page.tsx b/app/chat/[id]/page.tsx index e1b009a..627dc7a 100644 --- a/app/chat/[id]/page.tsx +++ b/app/chat/[id]/page.tsx @@ -1,7 +1,7 @@ import { notFound } from "next/navigation"; import { Chat } from "@/components/chat"; import type { TeacherUIMessage } from "@/src/core"; -import { getChat } from "@/src/server"; +import { getChat, isTurnRunning } from "@/src/server"; export const dynamic = "force-dynamic"; @@ -22,6 +22,9 @@ export default async function ChatPage({ title={chat.title} initialMessages={chat.messages as TeacherUIMessage[]} initialPrompt={prompt} + // A teacher turn that outlived its client: the remounted view must + // attach (busy shimmer + poll), never resurrect the stale record. + initiallyInFlight={isTurnRunning(chat.id)} /> ); } diff --git a/components/chat.tsx b/components/chat.tsx index d677b0c..ba5fa89 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -57,11 +57,15 @@ export function Chat({ title, initialMessages, initialPrompt, + initiallyInFlight, }: { chatId: string; title: string; initialMessages: TeacherUIMessage[]; initialPrompt?: string; + /** A teacher turn is running server-side but attached to no client + * (navigated away mid-turn, returned while in flight). */ + initiallyInFlight?: boolean; }) { const { messages, sendMessage, addToolOutput, setMessages, status, error, stop } = useChat({ @@ -82,27 +86,37 @@ export function Chat({ } }, [initialPrompt, initialMessages.length, sendMessage]); + // A turn running server-side but attached to no client of ours (navigated + // away mid-turn, returned while in flight): the record is write-ahead + // persisted (answer included), but the teacher's reaction only lands when + // the orphaned turn finishes — poll for that, don't resurrect. + const [turnInFlight, setTurnInFlight] = useState(initiallyInFlight ?? false); + // Mount continuation: if the stored transcript ends with a fulfilled tool // (a job result landed while no client was open), resume the lesson — the // SDK's sendAutomaticallyWhen only fires inside addToolOutput, so reloads // need this nudge. Once per mount; the turn that follows ends in text (or - // a pending question), so no loop. + // a pending question), so no loop. NEVER while a turn is already running + // server-side — that would double-run the teacher (the server would 409 + // it; the poll below resumes instead). const mountContinuation = useRef(false); useEffect(() => { if (mountContinuation.current) return; mountContinuation.current = true; if ( + !turnInFlight && status === "ready" && messages.length > 0 && lastAssistantMessageIsCompleteWithToolCalls({ messages }) ) { void sendMessage(); } - // mount-only by design + // mount-only by design (turnInFlight read at initial value) }, []); const pending = findPendingToolCall(messages); - const busy = status === "submitted" || status === "streaming"; + const localStream = status === "submitted" || status === "streaming"; + const busy = localStream || turnInFlight; // Teach-phase check (W5, vision §2 "two phases"): a pending question that // lands after plan_init starts collapsed behind a "Check your @@ -124,7 +138,7 @@ export function Chat({ // break provider conversion — contracts.ts "THE ASYNC PROTOCOL" §5), but // they don't dock: the running job's activity card sits in the transcript. const pendingJob = findPendingJobCall(messages); - const canSend = text.trim().length > 0 && !pending && !pendingJob; + const canSend = text.trim().length > 0 && !pending && !pendingJob && !turnInFlight; const syncFromServer = useCallback(async () => { const res = await fetch(`/api/chats/${chatId}`); @@ -134,8 +148,8 @@ export function Chat({ setMessages(record.messages); // setMessages does NOT fire sendAutomaticallyWhen (the SDK only checks // it inside addToolOutput) — without this the teacher would stay silent - // after a job result lands. sendMessage() with no args re-requests with - // the current list: the continuation. (contracts.ts protocol §4.) + // after a job result lands. sendMessage() with no args re-requests with + // the current list: the continuation. (contracts.ts protocol §4.) if ( status === "ready" && lastAssistantMessageIsCompleteWithToolCalls({ messages: record.messages }) @@ -144,6 +158,36 @@ export function Chat({ } }, [chatId, setMessages, sendMessage, status]); + // Re-attach after mid-turn navigation (the key fix): poll the record until + // the orphaned turn finishes (turnInFlight flips false server-side in + // onFinish, AFTER persist), then adopt the completed transcript — the + // syncFromServer above already runs the job-continuation check. Until then + // the session reads as busy: thinking shimmer, composer gated. + useEffect(() => { + if (!turnInFlight) return; + let cancelled = false; + let timer: ReturnType | undefined; + const poll = async () => { + if (cancelled) return; + try { + const res = await fetch(`/api/chats/${chatId}`); + if (res.ok && !((await res.json()) as { turnInFlight?: boolean }).turnInFlight) { + setTurnInFlight(false); + void syncFromServer(); + return; + } + } catch { + /* transient network hiccup — keep polling */ + } + if (!cancelled) timer = setTimeout(poll, 1500); + }; + timer = setTimeout(poll, 1000); + return () => { + cancelled = true; + if (timer) clearTimeout(timer); + }; + }, [turnInFlight, chatId, syncFromServer]); + const onSend = () => { const t = text.trim(); if (!t || busy || pending || pendingJob) return; @@ -223,7 +267,7 @@ export function Chat({ ))} - {status === "submitted" && Thinking…} + {(status === "submitted" || turnInFlight) && Thinking…} {questionStreaming && Writing a question…} {error && (
@@ -237,7 +281,9 @@ export function Chat({ requiresReveal={requiresReveal} composer={ { - 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 }); + } } diff --git a/src/server/index.ts b/src/server/index.ts index 2f27036..05b8e3a 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -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"; diff --git a/src/server/turn-liveness.ts b/src/server/turn-liveness.ts new file mode 100644 index 0000000..2f02369 --- /dev/null +++ b/src/server/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(); // 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; +} diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 8795aab..9c4694b 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -19,6 +19,7 @@ import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; +import { isTurnRunning } from "../src/server/turn-liveness.ts"; import { chatStore } from "../src/store/chats.ts"; import { buildAnswer, coerceCorrectAnswer, normalizeOptions, resolveCorrect, type QuestionData } from "../src/core/question-core.ts"; import { findPendingToolCall } from "../src/core/pending.ts"; @@ -326,3 +327,97 @@ describe("chat store", () => { assert.equal(loaded.title, "Store test"); }); }); + +describe("mid-turn navigation (turn liveness + write-ahead)", () => { + it("one turn per chat: a concurrent request is 409'd; the next one after finish is accepted", async () => { + const calls: string[][] = []; + const agent = createTeacherAgent(scriptedModel(calls)); + const chat = chatStore.createChat("Test: turn guard"); + createdChats.push(chat.id); + + const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me the solar system" }] }; + const res1 = await handleChatRequest(post(chat.id, [firstUser]), agent); // NOT drained → in flight + assert.equal(isTurnRunning(chat.id), true, "turn registered as in flight"); + + // A stale client (remount racing a live turn) tries to start a second turn. + const res2 = await handleChatRequest(post(chat.id, [firstUser]), agent); + assert.equal(res2.status, 409, "concurrent turn rejected — last-writer-wins persist would clobber it"); + + await drain(res1); // finish → onFinish → endTurn + assert.equal(isTurnRunning(chat.id), false, "registry cleared at finish"); + + const res3 = await handleChatRequest(post(chat.id, [firstUser]), agent); + assert.equal(res3.status, 200, "the next turn is accepted after the previous finished"); + await drain(res3); + }); + + it("write-ahead: the learner's answer is durable BEFORE the turn finishes", async () => { + const calls: string[][] = []; + const agent = createTeacherAgent(scriptedModel(calls)); + const chat = chatStore.createChat("Test: write-ahead"); + createdChats.push(chat.id); + + // Turn 1: the teacher asks the quiz (pending part in the record). + const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me the solar system" }] }; + await drain(await handleChatRequest(post(chat.id, [firstUser]), agent)); + const saved1 = chatStore.getChat(chat.id)!; + const assistantMsg = saved1.messages[1]!; + + // The learner answers; the client posts the graded transcript back. + const options = normalizeOptions(QUIZ_INPUT.options); + const qd: QuestionData = { + id: "call-1", + kind: "quiz", + question: QUIZ_INPUT.question, + mode: "single-select", + options, + correctIndices: resolveCorrect(coerceCorrectAnswer(QUIZ_INPUT.correctAnswer), options).indices, + explanation: QUIZ_INPUT.explanation, + createdAt: new Date().toISOString(), + }; + const answer = buildAnswer(qd, { values: ["Mercury"] }); + const messages2 = saved1.messages.map((m) => + m.id === assistantMsg.id + ? { + ...m, + parts: m.parts.map((p) => + p.type === "tool-quiz" ? { ...p, state: "output-available" as const, output: answer } : p, + ), + } + : m, + ); + + // NOT drained: the teacher turn is streaming, but the answer is already + // on disk — a remount (or crash) mid-turn sees the answered quiz, never + // the stale pre-answer question. + const res = await handleChatRequest(post(chat.id, messages2), agent); + const midTurn = chatStore.getChat(chat.id)!; + const midQuiz = midTurn.messages[1]!.parts.find((p) => p.type === "tool-quiz"); + assert.ok(midQuiz && midQuiz.type === "tool-quiz"); + assert.equal(midQuiz.state, "output-available", "write-ahead persisted the answer before finish"); + + await drain(res); // finish → persist the full turn → endTurn + const saved2 = chatStore.getChat(chat.id)!; + assert.equal(saved2.messages.length, 2, "full turn persisted as usual at finish"); + assert.equal(isTurnRunning(chat.id), false, "registry cleared at finish"); + }); + + it("a turn that fails to start (malformed messages) clears the registry — no lockout", async () => { + const chat = chatStore.createChat("Test: turn failed to start"); + createdChats.push(chat.id); + // A text part with no text: fails UIMessage validation before the stream + // exists — the path where no onFinish will ever run. + const bad = [{ id: "m1", role: "user", parts: [{ type: "text" }] }]; + const res = await handleChatRequest(post(chat.id, bad), createTeacherAgent(scriptedModel([]))); + assert.equal(res.status, 500, "turn-start failure is a 500, not a hang"); + assert.equal(isTurnRunning(chat.id), false, "registry cleared when the stream never started"); + + // And the chat is NOT locked: the next (valid) turn is accepted. + const ok = await handleChatRequest( + post(chat.id, [{ id: "m1", role: "user", parts: [{ type: "text", text: "teach me" }] }]), + createTeacherAgent(scriptedModel([])), + ); + assert.equal(ok.status, 200); + await drain(ok); + }); +}); diff --git a/tests/turn-liveness.test.ts b/tests/turn-liveness.test.ts new file mode 100644 index 0000000..c0c3533 --- /dev/null +++ b/tests/turn-liveness.test.ts @@ -0,0 +1,24 @@ +/** + * Turn liveness registry unit tests (src/server/turn-liveness.ts): one turn + * per chat, ever — begin/end round trip, double-begin rejection, restart + * after end. The 30-minute stale-eviction TTL is not tested (it is a leak + * guard, not a behavior). + */ + +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { beginTurn, endTurn, isTurnRunning } from "../src/server/turn-liveness.ts"; + +describe("turn liveness registry", () => { + it("begin/end round trip; a second begin is rejected until the turn ends", () => { + const id = "chat_liveness_unit"; + assert.equal(isTurnRunning(id), false, "initially idle"); + assert.equal(beginTurn(id), true, "first turn accepted"); + assert.equal(isTurnRunning(id), true, "in flight"); + assert.equal(beginTurn(id), false, "no two turns for one chat"); + endTurn(id); + assert.equal(isTurnRunning(id), false, "cleared at finish"); + assert.equal(beginTurn(id), true, "a new turn may start after the last ended"); + endTurn(id); + }); +});