Harden generation commits, session expiry, and frontend builds

backend/cortex_backend/api/jobs.py

@@ -98,6 +98,7 @@ class _JobRecord:
     prepared: bool = False
     preparation_error: str | None = None
     cancel_event: Event = field(default_factory=Event)
+    commit_started: bool = False
     status: JobStatus = "queued"
     sequence: int = 0
     error: str | None = None
@@ -127,6 +128,39 @@ def publish_progress(
         payload = {"message": message, **dict(data or {})}
         self.publish_event("progress", phase=phase, data=payload)
 
+    def begin_commit(
+        self,
+        phase: str,
+        message: str,
+        *,
+        data: Mapping[str, Any] | None = None,
+    ) -> bool:
+        """Atomically cross the point after which cancellation is too late.
+
+        The worker must call this immediately before its first durable result
+        mutation.  Sharing the registry lock with :meth:`JobRegistry.cancel`
+        gives the two operations one unambiguous order: cancellation wins and
+        this returns ``False``, or commit wins and later cancellation is inert.
+        """
+        payload = {"message": message, **dict(data or {})}
+        with self._registry._lock:
+            if (
+                self._record.status != "running"
+                or self._record.cancel_event.is_set()
+            ):
+                return False
+            if self._record.commit_started:
+                return True
+            self._record.commit_started = True
+            self._registry._append_event(
+                self._record,
+                kind="progress",
+                status="running",
+                phase=phase,
+                data=payload,
+            )
+            return True
+
     def publish_event(
         self,
         kind: EventKind,
@@ -392,7 +426,11 @@ def active_snapshot(self, *, kind: JobKind) -> JobSnapshot | None:
     def cancel(self, job_id: str, *, owner: str) -> JobSnapshot:
         record = self._owned_record(job_id, owner)
         with self._lock:
-            if record.status not in TERMINAL_STATUSES and record.status != "cancelling":
+            if (
+                not record.commit_started
+                and record.status not in TERMINAL_STATUSES
+                and record.status != "cancelling"
+            ):
                 record.cancel_event.set()
                 self._append_event(
                     record,
@@ -443,6 +481,8 @@ async def shutdown(self) -> None:
                 if record.status not in TERMINAL_STATUSES
             ]
             for record in records:
+                if record.commit_started:
+                    continue
                 if record.status != "cancelling":
                     record.cancel_event.set()
                     self._append_event(
@@ -493,7 +533,9 @@ async def _run(
             with self._lock:
                 if record.status in TERMINAL_STATUSES:
                     return
-                if record.status == "cancelling" or record.cancel_event.is_set():
+                if not record.commit_started and (
+                    record.status == "cancelling" or record.cancel_event.is_set()
+                ):
                     self._finalize_cancellation(record)
                     return
                 data = dict(
@@ -510,6 +552,7 @@ async def _run(
             with self._lock:
                 if (
                     record.status not in TERMINAL_STATUSES
+                    and not record.commit_started
                     and (record.status == "cancelling" or record.cancel_event.is_set())
                 ):
                     self._finalize_cancellation(record)
@@ -518,7 +561,9 @@ async def _run(
             with self._lock:
                 if record.status in TERMINAL_STATUSES:
                     return
-                if record.status == "cancelling" or record.cancel_event.is_set():
+                if not record.commit_started and (
+                    record.status == "cancelling" or record.cancel_event.is_set()
+                ):
                     self._finalize_cancellation(record)
                     return
                 logging.error(

backend/cortex_backend/api/routes.py

@@ -1869,23 +1869,10 @@ def runner(sink, cancel_event):
                 history_messages=prepared_history,
             )
             # The generation service checks cancellation around its model work,
-            # but the API owns the following persistence and optional title work.
-            # Keep those side effects behind explicit checkpoints as well.
+            # while the API owns streaming and persistence. Keep everything
+            # cancellable until begin_commit atomically seals the durable result.
             if cancel_event.is_set():
                 return {"cancelled": True}
-            code_execution_job_id = _queue_code_proposal(
-                request,
-                principal,
-                settings,
-                generation_snapshot.job_id,
-                result,
-            )
-            if code_execution_job_id:
-                sink.publish_progress(
-                    "code_approval",
-                    "A local code task is waiting for your approval.",
-                    data={"execution_job_id": code_execution_job_id},
-                )
             if result.thoughts:
                 for delta in _chunks(result.thoughts):
                     if cancel_event.is_set():
@@ -1904,14 +1891,7 @@ def runner(sink, cancel_event):
                     data={"delta": delta},
                 )
 
-            for memo in result.memory_command.additions:
-                if cancel_event.is_set():
-                    return {"cancelled": True}
-                deps.memories.add_memo(memo)
-            if cancel_event.is_set():
-                return {"cancelled": True}
-            sink.publish_progress("persisting", "Saving the response.")
-            if cancel_event.is_set():
+            if not sink.begin_commit("persisting", "Saving the response."):
                 return {"cancelled": True}
             stats_payload = asdict(result.stats) if result.stats else None
             if target_message_id is None:
@@ -1934,13 +1914,40 @@ def runner(sink, cancel_event):
                 )
                 assistant_message_id = target_message_id
 
-            if cancel_event.is_set():
-                return {"cancelled": True}
+            # The assistant turn is the canonical generation result. Code and
+            # memory actions are optional derivatives: queue them only after the
+            # answer exists, and never invalidate that answer if they fail.
+            code_execution_job_id = None
+            try:
+                code_execution_job_id = _queue_code_proposal(
+                    request,
+                    principal,
+                    settings,
+                    generation_snapshot.job_id,
+                    result,
+                )
+            except Exception as exc:
+                logging.warning(
+                    "Cortex code proposal queueing failed (%s).", type(exc).__name__
+                )
+            if code_execution_job_id:
+                sink.publish_progress(
+                    "code_approval",
+                    "A local code task is waiting for your approval.",
+                    data={"execution_job_id": code_execution_job_id},
+                )
+
+            for memo in result.memory_command.additions:
+                try:
+                    deps.memories.add_memo(memo)
+                except Exception as exc:
+                    logging.warning(
+                        "Cortex memory update failed (%s).", type(exc).__name__
+                    )
+
             updated_chat = deps.chats.get_chat(thread_id) or {"messages": []}
             title = str(updated_chat.get("title") or "New Chat")
             if target_message_id is None and title == "New Chat":
-                if cancel_event.is_set():
-                    return {"cancelled": True}
                 raw_title = None
                 title_generator = getattr(
                     deps.generation, "generate_chat_title", None
@@ -1955,26 +1962,20 @@ def runner(sink, cancel_event):
                             "Cortex chat title generation failed (%s).",
                             type(exc).__name__,
                         )
-                if cancel_event.is_set():
-                    return {"cancelled": True}
                 generated_title = normalize_title(raw_title, fallback="")
                 if (
                     not generated_title
                     or generated_title.casefold() in {"new chat", "untitled chat"}
                 ):
                     generated_title = title_from_first_message(payload.user_input)
                 if generated_title != title:
-                    if cancel_event.is_set():
-                        return {"cancelled": True}
                     try:
                         deps.chats.rename_chat(thread_id, generated_title)
                         title = generated_title
                     except Exception as exc:
                         logging.warning(
                             "Cortex title update failed (%s).", type(exc).__name__
                         )
-            if cancel_event.is_set():
-                return {"cancelled": True}
             updated_chat = deps.chats.get_chat(thread_id) or updated_chat
             return {
                 "thread_id": thread_id,

backend/cortex_backend/launcher/frontend.py

@@ -200,14 +200,15 @@ def build_frontend(
     frontend_root = frontend_root.resolve()
     if not (frontend_root / "package.json").is_file():
         raise FrontendBuildError("frontend/package.json is missing from the source checkout.")
-    lock = lock_digest(frontend_root)
-    node_major = _major_version("node")
-    npm_major = _major_version("npm")
     build_root = _stage_frontend_source(frontend_root)
     staging = build_root / f".cortex-dist-staging-{uuid.uuid4().hex}"
     dist = frontend_root / "dist"
     backup = frontend_root / f".cortex-dist-backup-{uuid.uuid4().hex}"
     try:
+        lock = lock_digest(build_root)
+        source = source_digest(build_root)
+        node_major = _major_version("node")
+        npm_major = _major_version("npm")
         _install_if_needed(build_root, lock)
         _run(
             [_tool_name("npm"), "run", "build", "--", "--outDir", str(staging)],
@@ -217,7 +218,7 @@ def build_frontend(
             raise FrontendBuildError("Frontend build completed without index.html.")
         manifest = FrontendManifest(
             lock_digest=lock,
-            source_digest=source_digest(frontend_root),
+            source_digest=source,
             node_major=node_major,
             npm_major=npm_major,
             built_at=datetime.now(timezone.utc).isoformat(),

frontend/src/app/App.test.tsx

@@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import userEvent from "@testing-library/user-event";
 import { App } from "./App";
 import { CortexApi } from "../api/client";
+import { useChatStore } from "../stores/useChatStore";
 import { ToastProvider } from "./ToastProvider";
 
 describe("App", () => {
@@ -42,6 +43,40 @@ describe("App", () => {
     expect(screen.queryByLabelText(/token/i)).not.toBeInTheDocument();
   });
 
+  it("returns to onboarding and clears a persisted generation when its stream session expires", async () => {
+    window.sessionStorage.setItem("cortex.session.token", "local-session");
+    window.sessionStorage.setItem("cortex.active.generation", JSON.stringify({
+      jobId: "job-expired",
+      threadId: "thread-expired",
+      lastEventId: 3,
+    }));
+    window.history.replaceState({}, "", "/chat/thread-expired");
+    const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), {
+      status,
+      headers: { "Content-Type": "application/json" },
+    });
+    const fetcher = vi.fn<typeof fetch>(async (input) => {
+      const url = String(input);
+      if (url.endsWith("/system")) return json({ status: "ok", preview: true, session_required: true, started_at: "2026-07-21T18:00:00Z" });
+      if (url.endsWith("/chat-groups")) return json([]);
+      if (url.endsWith("/chats")) return json([{ id: "thread-expired", title: "Interrupted", timestamp: "2026-07-21T18:00:00Z" }]);
+      if (url.endsWith("/chats/thread-expired")) return json({ id: "thread-expired", title: "Interrupted", timestamp: "2026-07-21T18:00:00Z", revision: 1, messages: [] });
+      if (url.endsWith("/settings")) return json({ settings: { models: { chat: "model-a", title: null }, appearance: { theme: "dark" } } });
+      if (url.endsWith("/memories")) return json({ memos: [] });
+      if (url.endsWith("/models")) return json({ required_models: [], optional_models: [], installed_models: ["model-a"], connection: { success: true, status: "connected", message: "Ready" } });
+      if (url.endsWith("/generations/job-expired/events")) return json({ detail: "Local session expired." }, 401);
+      return json({ detail: "Unexpected test route." }, 404);
+    });
+
+    render(<ToastProvider><App api={new CortexApi("/api/v1", fetcher)} /></ToastProvider>);
+
+    expect(await screen.findByRole("heading", { name: "Start local workspace" })).toBeVisible();
+    expect(fetcher.mock.calls.filter(([input]) => String(input).endsWith("/generations/job-expired/events"))).toHaveLength(1);
+    expect(window.sessionStorage.getItem("cortex.session.token")).toBeNull();
+    expect(window.sessionStorage.getItem("cortex.active.generation")).toBeNull();
+    expect(useChatStore.getState().generation).toMatchObject({ jobId: null, phase: "idle" });
+  });
+
   it("opens the workspace when the model service is unavailable", async () => {
     window.sessionStorage.setItem("cortex.session.token", "local-session");
     const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), {

frontend/src/app/App.tsx

@@ -482,7 +482,7 @@ function AuthenticatedWorkspace({ api, onSessionExpired }: { api: CortexApi; onS
       <AppShell chats={chats} activeChatId={routeChatId} modelConnection={models.connection} theme={theme} executionTasks={visibleExecutionTasks} onCancelExecution={cancelExecution} onDecideExecutionApproval={decideExecutionApproval} onLoadCodeSource={loadCodeSource} onOpenSettings={() => { if (route.kind === "chat") setSettingsReturnChatId(route.threadId); }} onRenameChat={renameChat} onDeleteChat={deleteChat} groups={groups} onCreateGroup={createGroup} onRenameGroup={renameGroup} onDeleteGroup={deleteGroup} onToggleGroup={toggleGroup} onMoveChat={moveChat}>
         {route.kind === "settings"
           ? <SettingsRoute activeChatId={settingsReturnChatId} settings={settings} memos={memos} saving={saving} memoryBusy={memoryBusy} onSave={saveSettings} onAddMemory={addMemory} onReplaceMemory={replaceMemory} onClearMemory={clearMemory} models={models} modelBusy={modelBusy} modelProgress={modelProgress} setupUrl={system.ollama_setup_url ?? "https://ollama.com/download"} onCheckModels={checkModels} onPullModel={pullModel} llamacppStatus={llamacppStatus} onDownloadGGUF={downloadGGUFModel} />
-          : <ChatRoute threadId={routeChatId} api={api} runtimeReady={runtimeConnected && selectedModelAvailable} runtimeMessage={models.connection?.message ?? null} localModels={localModels} selectedModel={selectedModel} selectedModelSupportsVision={selectedModelSupportsVision} modelBusy={modelBusy || saving} onSelectModel={chooseLocalModel} onRescanModels={checkModels} onChatChanged={upsertChatSummary} onForked={upsertChatSummary} />}
+          : <ChatRoute threadId={routeChatId} api={api} runtimeReady={runtimeConnected && selectedModelAvailable} runtimeMessage={models.connection?.message ?? null} localModels={localModels} selectedModel={selectedModel} selectedModelSupportsVision={selectedModelSupportsVision} modelBusy={modelBusy || saving} onSelectModel={chooseLocalModel} onRescanModels={checkModels} onChatChanged={upsertChatSummary} onForked={upsertChatSummary} onSessionExpired={onSessionExpired} />}
       </AppShell>
       <CommandPalette
         chats={chats}
@@ -508,9 +508,9 @@ function updateModelProgress(event: SSEEvent, setProgress: (progress: ModelProgr
   setProgress({ model, status, percent });
 }
 
-function ChatRoute({ threadId, api, runtimeReady, runtimeMessage, localModels, selectedModel, selectedModelSupportsVision, modelBusy, onSelectModel, onRescanModels, onChatChanged, onForked }: { threadId: string | null; api: CortexApi; runtimeReady: boolean; runtimeMessage: string | null; localModels: readonly string[]; selectedModel: string | null; selectedModelSupportsVision: boolean | null; modelBusy: boolean; onSelectModel: (model: string) => Promise<boolean>; onRescanModels: () => Promise<void>; onChatChanged: (chat: ChatResponse) => void; onForked: (chat: ChatResponse) => void }) {
+function ChatRoute({ threadId, api, runtimeReady, runtimeMessage, localModels, selectedModel, selectedModelSupportsVision, modelBusy, onSelectModel, onRescanModels, onChatChanged, onForked, onSessionExpired }: { threadId: string | null; api: CortexApi; runtimeReady: boolean; runtimeMessage: string | null; localModels: readonly string[]; selectedModel: string | null; selectedModelSupportsVision: boolean | null; modelBusy: boolean; onSelectModel: (model: string) => Promise<boolean>; onRescanModels: () => Promise<void>; onChatChanged: (chat: ChatResponse) => void; onForked: (chat: ChatResponse) => void; onSessionExpired: () => void }) {
   const navigate = useNavigate();
-  return <ChatPage api={api} threadId={threadId} runtimeReady={runtimeReady} runtimeMessage={runtimeMessage} localModels={localModels} selectedModel={selectedModel} selectedModelSupportsVision={selectedModelSupportsVision} modelBusy={modelBusy} onSelectModel={onSelectModel} onRescanModels={onRescanModels} onThreadCreated={(id) => navigate(chatPath(id), { replace: true })} onChatChanged={onChatChanged} onForked={(chat) => { onForked(chat); navigate(chatPath(chat.id)); }} />;
+  return <ChatPage api={api} threadId={threadId} runtimeReady={runtimeReady} runtimeMessage={runtimeMessage} localModels={localModels} selectedModel={selectedModel} selectedModelSupportsVision={selectedModelSupportsVision} modelBusy={modelBusy} onSelectModel={onSelectModel} onRescanModels={onRescanModels} onThreadCreated={(id) => navigate(chatPath(id), { replace: true })} onChatChanged={onChatChanged} onForked={(chat) => { onForked(chat); navigate(chatPath(chat.id)); }} onSessionExpired={onSessionExpired} />;
 }
 
 function SettingsRoute({ activeChatId, ...props }: Omit<SettingsPanelProps, "onClose"> & { activeChatId: string | null }) {