diff --git a/backend/cortex_backend/api/routes.py b/backend/cortex_backend/api/routes.py index 8dc781c..99217bc 100644 --- a/backend/cortex_backend/api/routes.py +++ b/backend/cortex_backend/api/routes.py @@ -1313,7 +1313,21 @@ def cancel_generation( _raise_job_error(exc) return _job_response(snapshot) - @router.get("/generations/{job_id}/events", response_model=GenerationEvent) + @router.get( + "/generations/{job_id}/events", + response_model=GenerationEvent, + response_class=StreamingResponse, + responses={ + 200: { + "description": "Server-sent generation events.", + "content": { + "text/event-stream": { + "schema": {"$ref": "#/components/schemas/GenerationEvent"} + } + }, + } + }, + ) async def generation_events( job_id: str, request: Request, diff --git a/backend/cortex_backend/llamacpp/server_manager.py b/backend/cortex_backend/llamacpp/server_manager.py index cbffe89..2693ce5 100644 --- a/backend/cortex_backend/llamacpp/server_manager.py +++ b/backend/cortex_backend/llamacpp/server_manager.py @@ -552,35 +552,43 @@ def _start_with_backend( base_url = f"http://127.0.0.1:{port}" deadline = time.monotonic() + self._health_timeout_seconds last_status_at = time.monotonic() - while time.monotonic() < deadline: - exit_code = process.poll() - if exit_code is not None: - if backend == "vulkan": - self._mark_backend_bad("vulkan") - raise ServerLaunchError( - "The local model runtime exited before it became ready.\n" - + "\n".join(stderr_tail[-20:]) - ) - if self._probe_health(base_url): - with self._state_lock: - self._process = process - self._loaded_model_path = model_path - self._loaded_num_ctx = num_ctx - self._base_url = base_url - self._state = "ready" - self._last_error = None - self._active_backend = backend - self._last_health_check = time.monotonic() - self._stderr_tail = stderr_tail - return ServerHandle(base_url=base_url, model_path=model_path) - now = time.monotonic() - if on_status is not None and now - last_status_at >= _STATUS_REPEAT_SECONDS: - on_status(f"Still loading the model ({model_path.name})... this can take a while for large files.") - last_status_at = now - time.sleep(_HEALTH_POLL_INTERVAL_SECONDS) - - process.terminate() - raise ServerStartTimeoutError("The local model runtime did not become ready in time.") + ready = False + try: + while time.monotonic() < deadline: + exit_code = process.poll() + if exit_code is not None: + if backend == "vulkan": + self._mark_backend_bad("vulkan") + raise ServerLaunchError( + "The local model runtime exited before it became ready.\n" + + "\n".join(stderr_tail[-20:]) + ) + if self._probe_health(base_url): + with self._state_lock: + self._process = process + self._loaded_model_path = model_path + self._loaded_num_ctx = num_ctx + self._base_url = base_url + self._state = "ready" + self._last_error = None + self._active_backend = backend + self._last_health_check = time.monotonic() + self._stderr_tail = stderr_tail + ready = True + return ServerHandle(base_url=base_url, model_path=model_path) + now = time.monotonic() + if on_status is not None and now - last_status_at >= _STATUS_REPEAT_SECONDS: + on_status(f"Still loading the model ({model_path.name})... this can take a while for large files.") + last_status_at = now + time.sleep(_HEALTH_POLL_INTERVAL_SECONDS) + + raise ServerStartTimeoutError("The local model runtime did not become ready in time.") + finally: + # The manager does not publish the process into ``self._process`` + # until health succeeds. Reap every failed startup here so a + # timeout or callback error cannot leave an unowned model process. + if not ready and process.poll() is None: + self._terminate_process(process) def _probe_health(self, base_url: str) -> bool: try: diff --git a/contracts/openapi.json b/contracts/openapi.json index 7ce8e97..fa1438b 100644 --- a/contracts/openapi.json +++ b/contracts/openapi.json @@ -4599,13 +4599,13 @@ "responses": { "200": { "content": { - "application/json": { + "text/event-stream": { "schema": { "$ref": "#/components/schemas/GenerationEvent" } } }, - "description": "Successful Response" + "description": "Server-sent generation events." }, "422": { "content": { diff --git a/frontend/src/features/chat/ChatPage.test.tsx b/frontend/src/features/chat/ChatPage.test.tsx index 6049c07..5dd892a 100644 --- a/frontend/src/features/chat/ChatPage.test.tsx +++ b/frontend/src/features/chat/ChatPage.test.tsx @@ -262,6 +262,140 @@ describe("ChatPage composer integration", () => { expect(screen.getByRole("button", { name: "Stop generating" })).toBeInTheDocument(); }); + it("moves drafts created during new-chat acceptance into the accepted thread", async () => { + const user = userEvent.setup(); + const lateAttachment: ChatAttachment = { + attachment_id: "late-doc", + filename: "next-turn.md", + mime_type: "text/markdown", + size: 9, + sha256: "1".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + let accept!: (value: { job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }) => void; + const accepted = new Promise<{ job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }>((resolve) => { accept = resolve; }); + const api = chatApi({ + chat: vi.fn(async (id: string) => emptyChat(id)), + generate: vi.fn(() => accepted), + stageChatAttachment: vi.fn().mockResolvedValue(lateAttachment), + }); + function RoutedChat() { + const [threadId, setThreadId] = useState(null); + return ( + true} + onRescanModels={async () => undefined} + onThreadCreated={setThreadId} + onChatChanged={vi.fn()} + onForked={vi.fn()} + /> + ); + } + render(); + + const composer = await screen.findByLabelText("Message Cortex"); + const attachmentInput = screen.getByLabelText("Attach images or documents"); + await user.type(composer, "First turn"); + await user.click(screen.getByRole("button", { name: "Send message" })); + await waitFor(() => expect(api.generate).toHaveBeenCalledTimes(1)); + + await user.clear(composer); + await user.type(composer, "Draft for the next turn"); + await user.upload(attachmentInput, new File(["next turn"], "next-turn.md", { type: "text/markdown" })); + expect(await screen.findByRole("button", { name: "Remove next-turn.md" })).toBeInTheDocument(); + + accept({ + job_id: "job-new", + kind: "generation", + status: "queued", + thread_id: "thread-new", + user_message_id: "message-new", + }); + + await waitFor(() => expect(screen.getByLabelText("Message Cortex")).toHaveValue("Draft for the next turn")); + expect(screen.getByRole("button", { name: "Remove next-turn.md" })).toBeInTheDocument(); + expect(window.sessionStorage.getItem("cortex.composer.draft.new")).toBeNull(); + expect(window.sessionStorage.getItem("cortex.composer.draft.thread-new")).toBe("Draft for the next turn"); + expect(window.sessionStorage.getItem("cortex.composer.attachments.new")).toBeNull(); + expect(JSON.parse(window.sessionStorage.getItem("cortex.composer.attachments.thread-new") ?? "[]")).toEqual([lateAttachment]); + }); + + it("retargets an in-flight new-chat attachment when acceptance wins the race", async () => { + const user = userEvent.setup(); + const stagedAttachment: ChatAttachment = { + attachment_id: "inverse-order-doc", + filename: "after-acceptance.md", + mime_type: "text/markdown", + size: 16, + sha256: "2".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + let accept!: (value: { job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }) => void; + let finishStaging!: (value: ChatAttachment) => void; + const accepted = new Promise<{ job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }>((resolve) => { accept = resolve; }); + const staging = new Promise((resolve) => { finishStaging = resolve; }); + const api = chatApi({ + chat: vi.fn(async (id: string) => emptyChat(id)), + generate: vi.fn(() => accepted), + stageChatAttachment: vi.fn(() => staging), + }); + function RoutedChat() { + const [threadId, setThreadId] = useState(null); + return ( + true} + onRescanModels={async () => undefined} + onThreadCreated={setThreadId} + onChatChanged={vi.fn()} + onForked={vi.fn()} + /> + ); + } + render(); + + const composer = await screen.findByLabelText("Message Cortex"); + await user.type(composer, "First turn"); + await user.click(screen.getByRole("button", { name: "Send message" })); + await waitFor(() => expect(api.generate).toHaveBeenCalledTimes(1)); + await user.upload( + screen.getByLabelText("Attach images or documents"), + new File(["after acceptance"], "after-acceptance.md", { type: "text/markdown" }), + ); + await waitFor(() => expect(api.stageChatAttachment).toHaveBeenCalledTimes(1)); + + accept({ + job_id: "job-inverse", + kind: "generation", + status: "queued", + thread_id: "thread-inverse", + user_message_id: "message-inverse", + }); + await waitFor(() => expect(api.chat).toHaveBeenCalledWith("thread-inverse")); + expect(window.sessionStorage.getItem("cortex.composer.attachments.new")).toBeNull(); + + act(() => finishStaging(stagedAttachment)); + + expect(await screen.findByRole("button", { name: "Remove after-acceptance.md" })).toBeInTheDocument(); + expect(window.sessionStorage.getItem("cortex.composer.attachments.new")).toBeNull(); + expect(JSON.parse(window.sessionStorage.getItem("cortex.composer.attachments.thread-inverse") ?? "[]")).toEqual([stagedAttachment]); + }); + it("replays an active generation from the beginning after a remount", async () => { window.sessionStorage.setItem("cortex.active.generation", JSON.stringify({ jobId: "job-replay", threadId: "thread-a", lastEventId: 7 })); const streamCalls: Array<{ afterEventId?: number }> = []; @@ -429,6 +563,94 @@ describe("ChatPage composer integration", () => { expect(screen.getByLabelText("Message Cortex")).toBeEnabled(); }); + it("regenerates from the selected user turn instead of stale cross-chat or composer state", async () => { + const user = userEvent.setup(); + const originalAttachment: ChatAttachment = { + attachment_id: "original-doc", + filename: "original.md", + mime_type: "text/markdown", + size: 12, + sha256: "d".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + const draftAttachment: ChatAttachment = { + attachment_id: "next-draft-doc", + filename: "next-draft.md", + mime_type: "text/markdown", + size: 14, + sha256: "e".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + const threadA = emptyChat("thread-a"); + const threadB: ChatResponse = { + ...emptyChat("thread-b"), + revision: 2, + messages: [ + { id: "user-b", role: "user", content: "Prompt from B", attachments: [originalAttachment] }, + { id: "assistant-b", role: "assistant", content: "Answer from B" }, + ], + }; + const api = chatApi({ + chat: vi.fn(async (id: string) => id === "thread-b" ? threadB : threadA), + generate: vi.fn().mockResolvedValue({ + job_id: "job-a", kind: "generation", status: "queued", thread_id: "thread-a", user_message_id: "user-a", + }), + regenerate: vi.fn().mockResolvedValue({ + job_id: "job-b", kind: "generation", status: "queued", thread_id: "thread-b", + }), + stageChatAttachment: vi.fn().mockResolvedValue(draftAttachment), + streamGeneration: vi.fn(async (jobId, onEvent) => { + const completedThreadId = jobId === "job-a" ? "thread-a" : "thread-b"; + onEvent({ + event_id: 1, + event: "generation.completed", + job_id: jobId, + thread_id: completedThreadId, + data: {}, + }); + }), + }); + const view = renderChat(api, "thread-a"); + const composer = await screen.findByLabelText("Message Cortex"); + await user.type(composer, "Prompt from A"); + await user.click(screen.getByRole("button", { name: "Send message" })); + await waitFor(() => expect(window.sessionStorage.getItem("cortex.active.generation")).toBeNull()); + + view.rerender( + true} + onRescanModels={async () => undefined} + onThreadCreated={vi.fn()} + onChatChanged={vi.fn()} + onForked={vi.fn()} + />, + ); + expect(await screen.findByText("Answer from B")).toBeInTheDocument(); + await user.upload( + screen.getByLabelText("Attach images or documents"), + new File(["next"], "next-draft.md", { type: "text/markdown" }), + ); + expect(await screen.findByRole("button", { name: "Remove next-draft.md" })).toBeInTheDocument(); + + await user.click(screen.getByRole("button", { name: "Regenerate response" })); + + await waitFor(() => expect(api.regenerate).toHaveBeenCalledWith("thread-b", expect.objectContaining({ + message_id: "assistant-b", + user_input: "Prompt from B", + attachments: [originalAttachment], + }))); + expect(screen.getByRole("button", { name: "Remove next-draft.md" })).toBeInTheDocument(); + }); + it("stages a document without putting its contents into the composer and sends its opaque metadata", async () => { const user = userEvent.setup(); const attachment: ChatAttachment = { @@ -458,6 +680,51 @@ describe("ChatPage composer integration", () => { }))); }); + it("keeps attachments staged while generation acceptance is pending", async () => { + const user = userEvent.setup(); + const firstAttachment: ChatAttachment = { + attachment_id: "doc-first", + filename: "first.md", + mime_type: "text/markdown", + size: 5, + sha256: "f".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + const nextAttachment: ChatAttachment = { + attachment_id: "doc-next", + filename: "next.md", + mime_type: "text/markdown", + size: 4, + sha256: "0".repeat(64), + kind: "document", + expires_at: "2099-01-01T00:00:00Z", + }; + let accept!: (value: { job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }) => void; + const accepted = new Promise<{ job_id: string; kind: "generation"; status: "queued"; thread_id: string; user_message_id: string }>((resolve) => { accept = resolve; }); + const api = chatApi({ + stageChatAttachment: vi.fn() + .mockResolvedValueOnce(firstAttachment) + .mockResolvedValueOnce(nextAttachment), + generate: vi.fn(() => accepted), + }); + renderChat(api); + + const attachmentInput = await screen.findByLabelText("Attach images or documents"); + await user.upload(attachmentInput, new File(["first"], "first.md", { type: "text/markdown" })); + await screen.findByRole("button", { name: "Remove first.md" }); + await user.click(screen.getByRole("button", { name: "Send message" })); + await waitFor(() => expect(api.generate).toHaveBeenCalledTimes(1)); + + await user.upload(attachmentInput, new File(["next"], "next.md", { type: "text/markdown" })); + expect(await screen.findByRole("button", { name: "Remove next.md" })).toBeInTheDocument(); + + accept({ job_id: "job-1", kind: "generation", status: "queued", thread_id: "thread-a", user_message_id: "message-1" }); + + await waitFor(() => expect(screen.queryByRole("button", { name: "Remove first.md" })).not.toBeInTheDocument()); + expect(screen.getByRole("button", { name: "Remove next.md" })).toBeInTheDocument(); + }); + it("explains the image capability mismatch before a generation request is made", async () => { const user = userEvent.setup(); const attachment: ChatAttachment = { diff --git a/frontend/src/features/chat/ChatPage.tsx b/frontend/src/features/chat/ChatPage.tsx index 92de760..595c132 100644 --- a/frontend/src/features/chat/ChatPage.tsx +++ b/frontend/src/features/chat/ChatPage.tsx @@ -48,6 +48,15 @@ type ChatLoadState = { error: string | null; }; +type StartedGeneration = { + threadId: string; +}; + +type AttachmentDraftTarget = { + scope: string; + threadId: string | null; +}; + const MAX_CHAT_ATTACHMENT_BYTES = 10 * 1024 * 1024; const MAX_CHAT_ATTACHMENT_TOTAL_BYTES = 24 * 1024 * 1024; const MAX_CHAT_ATTACHMENTS = 8; @@ -102,6 +111,7 @@ export function ChatPage({ const initialMountRef = useRef(true); const draftsRef = useRef(drafts); const attachmentDraftsRef = useRef(attachmentDrafts); + const attachmentDraftTargetsRef = useRef(new Set()); const reportGenerationFailure = useCallback((failedThreadId: string, message: string) => { setGenerationError({ threadId: failedThreadId, message }); @@ -243,15 +253,15 @@ export function ChatPage({ // eslint-disable-next-line react-hooks/exhaustive-deps }, [threadId, loadChat]); - const startGeneration = async (prompt: string, regenerateMessageId?: string, suppliedAttachments: readonly ChatAttachment[] = attachments): Promise => { + const startGeneration = async (prompt: string, regenerateMessageId?: string, suppliedAttachments: readonly ChatAttachment[] = attachments): Promise => { const input = prompt.trim() || (suppliedAttachments.length ? "Please review the attached file(s)." : ""); - if (!input || generation.jobId || startingRef.current) return false; + if (!input || generation.jobId || startingRef.current) return null; if (!runtimeReady) { setGenerationError({ threadId, message: runtimeMessage ?? "The local runtime is unavailable. Rescan local models after it is running.", }); - return false; + return null; } startingRef.current = true; @@ -298,14 +308,13 @@ export function ChatPage({ error: null, }); } - if (!threadId) onThreadCreated(jobThreadId); - return true; + return { threadId: jobThreadId }; } catch (requestError) { setGenerationError({ threadId, message: requestError instanceof ApiError ? requestError.detail : "The response could not be started. Your message is still here.", }); - return false; + return null; } finally { startingRef.current = false; setStarting(false); @@ -316,20 +325,66 @@ export function ChatPage({ const submittedDraft = draft; const submittedAttachments = attachments; const submittedScope = draftScope; + const submittedAttachmentScope = attachmentScope; const submittedThreadId = threadId; - const accepted = await startGeneration(submittedDraft, undefined, submittedAttachments); + const started = await startGeneration(submittedDraft, undefined, submittedAttachments); + if (!started) return false; + + const destinationThreadId = submittedThreadId ?? started.threadId; + const destinationDraftScope = composerDraftKey(destinationThreadId); + const destinationAttachmentScope = composerAttachmentKey(destinationThreadId); + if (submittedAttachmentScope !== destinationAttachmentScope) { + // Retarget only batches that were already staging into this submitted + // draft. Each batch owns its mutable target, so a later /chat/new never + // inherits a stale redirect to this accepted thread. + for (const target of attachmentDraftTargetsRef.current) { + if (target.scope === submittedAttachmentScope) { + target.scope = destinationAttachmentScope; + target.threadId = destinationThreadId; + } + } + } const currentDraft = draftsRef.current[submittedScope] ?? readComposerDraft(submittedThreadId); - if (accepted && currentDraft === submittedDraft) { - const nextDrafts = { ...draftsRef.current, [submittedScope]: "" }; + const retainedDraft = currentDraft === submittedDraft ? "" : currentDraft; + if (submittedScope === destinationDraftScope) { + const nextDrafts = { ...draftsRef.current, [submittedScope]: retainedDraft }; + draftsRef.current = nextDrafts; + setDrafts(nextDrafts); + writeComposerDraft(submittedThreadId, retainedDraft); + } else { + const nextDrafts = { + ...draftsRef.current, + [submittedScope]: "", + [destinationDraftScope]: retainedDraft, + }; draftsRef.current = nextDrafts; setDrafts(nextDrafts); writeComposerDraft(submittedThreadId, ""); - const nextAttachments = { ...attachmentDraftsRef.current, [attachmentScope]: [] }; + writeComposerDraft(destinationThreadId, retainedDraft); + } + if (submittedAttachments.length || submittedAttachmentScope !== destinationAttachmentScope) { + const submittedAttachmentIds = new Set(submittedAttachments.map((attachment) => attachment.attachment_id)); + const currentAttachments = attachmentDraftsRef.current[submittedAttachmentScope] + ?? readComposerAttachments(submittedThreadId); + const retainedAttachments = currentAttachments.filter( + (attachment) => !submittedAttachmentIds.has(attachment.attachment_id), + ); + const nextAttachments = submittedAttachmentScope === destinationAttachmentScope + ? { ...attachmentDraftsRef.current, [submittedAttachmentScope]: retainedAttachments } + : { + ...attachmentDraftsRef.current, + [submittedAttachmentScope]: [], + [destinationAttachmentScope]: retainedAttachments, + }; attachmentDraftsRef.current = nextAttachments; setAttachmentDrafts(nextAttachments); - writeComposerAttachments(submittedThreadId, []); + if (submittedAttachmentScope !== destinationAttachmentScope) { + writeComposerAttachments(submittedThreadId, []); + } + writeComposerAttachments(destinationThreadId, retainedAttachments); } - return accepted; + if (!submittedThreadId) onThreadCreated(started.threadId); + return true; }; const cancel = async (): Promise => { @@ -355,7 +410,9 @@ export function ChatPage({ const retryLastPrompt = async (): Promise => { if (!lastPrompt) return false; - return startGeneration(lastPrompt, undefined, lastAttachments); + const started = await startGeneration(lastPrompt, undefined, lastAttachments); + if (started && !threadId) onThreadCreated(started.threadId); + return Boolean(started); }; const fork = async (message: ChatMessage) => { @@ -383,6 +440,8 @@ export function ChatPage({ const addAttachments = async (files: File[]): Promise => { if (attachmentsBusy || !files.length) return; + const target: AttachmentDraftTarget = { scope: attachmentScope, threadId }; + attachmentDraftTargetsRef.current.add(target); setAttachmentsBusy(true); setAttachmentError(null); try { @@ -406,14 +465,25 @@ export function ChatPage({ staged.push(attachment); totalBytes += attachment.size; } - const next = [...attachments, ...staged]; - const nextAttachments = { ...attachmentDraftsRef.current, [attachmentScope]: next }; + // The generation request and attachment staging can finish in either + // order. Merge into the latest scoped draft instead of the render-time + // `attachments` snapshot, which may contain files that were submitted + // and cleared while these new files were still uploading. + const currentAttachments = attachmentDraftsRef.current[target.scope] + ?? readComposerAttachments(target.threadId); + const currentAttachmentIds = new Set(currentAttachments.map((attachment) => attachment.attachment_id)); + const next = [ + ...currentAttachments, + ...staged.filter((attachment) => !currentAttachmentIds.has(attachment.attachment_id)), + ]; + const nextAttachments = { ...attachmentDraftsRef.current, [target.scope]: next }; attachmentDraftsRef.current = nextAttachments; setAttachmentDrafts(nextAttachments); - writeComposerAttachments(threadId, next); + writeComposerAttachments(target.threadId, next); } catch (error) { setAttachmentError(error instanceof ApiError ? error.detail : error instanceof Error ? error.message : "The attachment could not be uploaded."); } finally { + attachmentDraftTargetsRef.current.delete(target); setAttachmentsBusy(false); } }; @@ -456,7 +526,14 @@ export function ChatPage({ finalAssistantId={finalAssistantId} busy={Boolean(generation.jobId) || starting} forkingMessageId={forkingMessage} - onRegenerate={(message, index) => void startGeneration(lastPrompt || messages[index - 1]?.content || "", message.id ?? undefined)} + onRegenerate={(message, index) => { + const userTurn = messages[index - 1]; + void startGeneration( + userTurn?.role === "user" ? userTurn.content : "", + message.id ?? undefined, + userTurn?.role === "user" ? userTurn.attachments ?? [] : [], + ); + }} onFork={(message) => void fork(message)} onNearEndChange={handleNearEndChange} trailingContent={ diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index def0534..96d0ee8 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -53,6 +53,20 @@ def _events(body: str) -> list[dict]: ] +def test_generation_stream_openapi_declares_sse_media_type(): + app = create_app(allowed_hosts=ALLOWED_HOSTS) + response = app.openapi()["paths"]["/api/v1/generations/{job_id}/events"][ + "get" + ]["responses"]["200"] + + assert response["description"] == "Server-sent generation events." + assert response["content"] == { + "text/event-stream": { + "schema": {"$ref": "#/components/schemas/GenerationEvent"} + } + } + + def test_api_factory_is_headless_and_session_exchange_is_one_time(): app, client = _client() with client: diff --git a/tests/test_llamacpp_server_manager.py b/tests/test_llamacpp_server_manager.py index 2a6a7ac..967e956 100644 --- a/tests/test_llamacpp_server_manager.py +++ b/tests/test_llamacpp_server_manager.py @@ -8,6 +8,7 @@ import io import json +import subprocess import threading import time from pathlib import Path @@ -44,6 +45,20 @@ def wait(self, timeout=None): return 0 +class _UncooperativePopen(_FakePopen): + """Stay alive after terminate() until the manager escalates to kill().""" + + def __init__(self) -> None: + super().__init__() + self.wait_calls = 0 + + def wait(self, timeout=None): + self.wait_calls += 1 + if not self.killed: + raise subprocess.TimeoutExpired("llama-server", timeout) + return 0 + + class _QueueLauncher: """Returns pre-built fake processes in order, one per launch() call.""" @@ -281,6 +296,25 @@ def test_slow_but_alive_process_times_out_without_gpu_fallback(tmp_path: Path) - assert not (tmp_path / "preferred_gpu_backend.json").exists() +def test_start_timeout_kills_and_reaps_process_that_ignores_terminate(tmp_path: Path) -> None: + process = _UncooperativePopen() + manager = _manager( + tmp_path, + fetcher=_FakeFetcher(), + launcher=_QueueLauncher([process]), + http_client=_AlwaysUnhealthyClient(), + gpu_backend="vulkan", + health_timeout_seconds=0.05, + ) + + with pytest.raises(ServerStartTimeoutError): + manager.ensure_ready(tmp_path / "model.gguf", num_ctx=4096) + + assert process.terminated is True + assert process.killed is True + assert process.wait_calls == 2 + + def test_unconfigured_release_fails_cleanly(tmp_path: Path) -> None: fetcher = _FakeFetcher() launcher = _QueueLauncher([])