diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py index ffae9f8c..38f11c30 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -146,28 +146,23 @@ def read_slice(path: Path, *, offset: int, limit: int) -> TextSlice: def read_transcript_chunk( - engine: Engine, - call_id: str, + call: AgentCall, stream: Literal["stdout", "stderr"], *, offset: int, limit: int, -) -> TranscriptChunk | None: +) -> TranscriptChunk: # One paginated slice of an agent call's stdout/stderr log — the initial - # backfill before the live tail takes over. None if the call is gone; an empty - # eof chunk if it has produced no log yet. The platform owns the log path. - with session_scope(engine): - call = AgentCall.get(call_id) - if not call: - return - path = call.get_stream_path(stream) + # backfill before the live tail takes over. An empty eof chunk while the call + # has produced no log yet. The platform owns the log path. + path = call.get_stream_path(stream) if not path: return TranscriptChunk( - call_id=call_id, stream=stream, offset=offset, next_offset=offset, eof=True, text="" + call_id=call.id, stream=stream, offset=offset, next_offset=offset, eof=True, text="" ) piece = read_slice(path, offset=offset, limit=limit) return TranscriptChunk( - call_id=call_id, + call_id=call.id, stream=stream, offset=piece.offset, next_offset=piece.next_offset, diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index a7fb84d4..f76a2f44 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -273,17 +273,21 @@ def _get_transcript_routes(cls) -> "APIRouter": import mimetypes from typing import Literal - from fastapi import APIRouter, HTTPException, status + from fastapi import APIRouter, HTTPException, Response, status from fastapi.responses import FileResponse, StreamingResponse from druks.api.dependencies import EngineDep from druks.durable import reads + from druks.durable.enums import AgentCallStatus from druks.durable.live import SSE_HEADERS from druks.durable.models import AgentCall from druks.durable.schemas import AgentCallFiles, TranscriptChunk default_limit = 64 * 1024 max_limit = 256 * 1024 + # A call that has stopped writing never appends to its log again, so the + # byte range this serves is permanent. + settled_cache = "public, max-age=31536000, immutable" router = APIRouter(prefix="/transcripts/{call_id}", tags=[f"{cls.name}:transcripts"]) @@ -291,7 +295,7 @@ def _get_transcript_routes(cls) -> "APIRouter": async def get_transcript( call_id: str, stream: Literal["stdout", "stderr"], - engine: EngineDep, + response: Response, offset: int = 0, limit: int = default_limit, ) -> TranscriptChunk: @@ -301,10 +305,14 @@ async def get_transcript( raise HTTPException( status.HTTP_400_BAD_REQUEST, f"limit must be in 1..{max_limit}." ) - chunk = reads.read_transcript_chunk(engine, call_id, stream, offset=offset, limit=limit) - if not chunk: + call = AgentCall.get(call_id) + if not call: raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") - return chunk + if call.get_live_status() == AgentCallStatus.RUNNING: + response.headers["Cache-Control"] = "no-store" + else: + response.headers["Cache-Control"] = settled_cache + return reads.read_transcript_chunk(call, stream, offset=offset, limit=limit) @router.get("/stream", response_class=StreamingResponse) async def stream_transcript( diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index c4ab984a..9c158a4f 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -464,6 +464,7 @@ def seed_agent_run( host_id: str | None = None, model: str | None = "gpt-5.5", workflow_id: str | None = None, + run_state: str = "running", ): """Create a build Run and an AgentCall on it, returning the AgentCall. @@ -482,7 +483,7 @@ def seed_agent_run( project = Project.create(name=repo) ProjectRepo.create(project_id=project.id, full_name=repo) work_item_id = WorkItem.create(project_id=project.id, title="x", repo=repo).id - run = seed_build_run(session, work_item_id=work_item_id) + run = seed_build_run(session, work_item_id=work_item_id, state=run_state) workflow_id = run.id else: run = Run.get(workflow_id) diff --git a/backend/tests/test_agent_routes.py b/backend/tests/test_agent_routes.py index c286c03d..80b9277c 100644 --- a/backend/tests/test_agent_routes.py +++ b/backend/tests/test_agent_routes.py @@ -192,7 +192,7 @@ def test_cancel_run_route(client: TestClient, db_session): assert again.json()["result"] == "already_cancelled" -def test_transcript_route_matches_the_read_machinery(client: TestClient, db_session, db_engine): +def test_transcript_route_matches_the_read_machinery(client: TestClient, db_session): call = seed_agent_run() call_dir = call.call_dir call_dir.mkdir(parents=True, exist_ok=True) @@ -202,7 +202,7 @@ def test_transcript_route_matches_the_read_machinery(client: TestClient, db_sess f"/api/build/transcripts/{call.id}", params={"stream": "stdout", "limit": 7} ) assert response.status_code == 200 - chunk = read_transcript_chunk(db_engine, call.id, "stdout", offset=0, limit=7) + chunk = read_transcript_chunk(call, "stdout", offset=0, limit=7) assert response.json() == chunk.model_dump(mode="json", by_alias=True) # The 7-byte cut lands mid-é; the window serves the seam's one �. assert response.json()["text"] == "hello �" diff --git a/backend/tests/test_api_runs.py b/backend/tests/test_api_runs.py index 5669bb8e..3f310a4a 100644 --- a/backend/tests/test_api_runs.py +++ b/backend/tests/test_api_runs.py @@ -21,10 +21,10 @@ def client(tmp_path: Path, db_session, monkeypatch): yield client -def _seed_run(*, tmp_path: Path) -> str: +def _seed_run(*, tmp_path: Path, finished: bool = True, run_state: str = "running") -> str: from conftest import finish_agent_run, seed_agent_run - run = seed_agent_run() + run = seed_agent_run(run_state=run_state) # The harness writes every file for a call into run-//. call_dir = run.call_dir @@ -34,7 +34,8 @@ def _seed_run(*, tmp_path: Path) -> str: (call_dir / "stdout.jsonl").write_bytes(b"hello stdout output") (call_dir / "stderr.log").write_bytes(b"warning text") - finish_agent_run(run) + if finished: + finish_agent_run(run) return run.id @@ -69,6 +70,7 @@ def test_transcript_range_fetch_paginates( params={"stream": "stdout", "limit": 5}, ) assert first.status_code == 200 + assert first.headers["cache-control"] == "public, max-age=31536000, immutable" data = first.json() assert data["offset"] == 0 assert data["nextOffset"] == 5 @@ -85,6 +87,40 @@ def test_transcript_range_fetch_paginates( assert data["eof"] is True +def test_transcript_of_a_running_call_is_never_cached( + client: TestClient, + tmp_path: Path, + db_session, +): + call_id = _seed_run(tmp_path=tmp_path, finished=False) + + response = client.get( + f"/api/build/transcripts/{call_id}", + params={"stream": "stdout", "limit": 5}, + ) + + assert response.status_code == 200 + assert response.headers["cache-control"] == "no-store" + + +def test_transcript_of_an_abandoned_call_is_cached_immutably( + client: TestClient, + tmp_path: Path, + db_session, +): + # The call never wrote finished_at, but its run died — nothing will append + # to that log again, so the chunk is as permanent as a finished call's. + call_id = _seed_run(tmp_path=tmp_path, finished=False, run_state="failed") + + response = client.get( + f"/api/build/transcripts/{call_id}", + params={"stream": "stdout", "limit": 5}, + ) + + assert response.status_code == 200 + assert response.headers["cache-control"] == "public, max-age=31536000, immutable" + + def test_transcript_missing_file_returns_eof( client: TestClient, tmp_path: Path, diff --git a/frontend/src/components/RunTranscript.test.tsx b/frontend/src/components/RunTranscript.test.tsx new file mode 100644 index 00000000..71bbe951 --- /dev/null +++ b/frontend/src/components/RunTranscript.test.tsx @@ -0,0 +1,88 @@ +import { act, cleanup, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { useSSE } from '../api/sse' +import { RunTranscript } from './RunTranscript' + +vi.mock('../api/sse', () => ({ + useSSE: vi.fn(), +})) + +const useSSEMock = vi.mocked(useSSE) + +function chunkResponse(text: string, nextOffset: number, eof: boolean): Response { + return new Response(JSON.stringify({ text, nextOffset, eof }), { status: 200 }) +} + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise + }) + return { promise, resolve } +} + +afterEach(() => { + cleanup() + vi.unstubAllGlobals() + vi.clearAllMocks() +}) + +describe('RunTranscript', () => { + it('renders each terminal transcript chunk as it arrives', async () => { + const secondResponse = deferred() + const fetchMock = vi + .fn<(url: string, init?: RequestInit) => Promise>() + .mockResolvedValueOnce(chunkResponse('first row\n', 10, false)) + .mockReturnValueOnce(secondResponse.promise) + vi.stubGlobal('fetch', fetchMock) + + render() + + expect(await screen.findByText('first row')).toBeTruthy() + expect(fetchMock).toHaveBeenCalledTimes(2) + + await act(async () => { + secondResponse.resolve(chunkResponse('second row\n', 21, true)) + await secondResponse.promise + }) + + expect(screen.getByText('first row')).toBeTruthy() + expect(await screen.findByText('second row')).toBeTruthy() + }) + + it('hands a live transcript from the first chunk to SSE until finished', async () => { + const fetchMock = vi.fn<(url: string, init?: RequestInit) => Promise>( + async () => chunkResponse('initial row\n', 12, false), + ) + vi.stubGlobal('fetch', fetchMock) + + render() + + expect(await screen.findByText('initial row')).toBeTruthy() + expect(fetchMock).toHaveBeenCalledTimes(1) + expect(fetchMock).toHaveBeenCalledWith( + '/api/build/transcripts/call-2?stream=stdout&offset=0&limit=262144', + { headers: { Accept: 'application/json' } }, + ) + expect(useSSEMock).toHaveBeenLastCalledWith( + '/api/build/transcripts/call-2/stream?stream=stdout&offset=12', + expect.objectContaining({ enabled: true }), + ) + + act(() => { + const handlers = useSSEMock.mock.calls.at(-1)?.[1].handlers + handlers?.['transcript.chunk']?.({ text: 'streamed row\n' }) + }) + expect(await screen.findByText('streamed row')).toBeTruthy() + + act(() => { + const handlers = useSSEMock.mock.calls.at(-1)?.[1].handlers + handlers?.['agent_call.finished']?.({}) + }) + expect(useSSEMock).toHaveBeenLastCalledWith( + '/api/build/transcripts/call-2/stream?stream=stdout&offset=12', + expect.objectContaining({ enabled: false }), + ) + }) +}) diff --git a/frontend/src/components/RunTranscript.tsx b/frontend/src/components/RunTranscript.tsx index 833780c8..552c5f88 100644 --- a/frontend/src/components/RunTranscript.tsx +++ b/frontend/src/components/RunTranscript.tsx @@ -16,10 +16,10 @@ interface RunTranscriptProps { } /** - * Shared live transcript: fetches the full log up-front (paginated), then — - * when ``isLive`` — opens an SSE stream pinned at the trailing offset and - * appends chunks until ``agent_call.finished``. Used by the agent-call and - * work-item pages; the only difference is ``basePath``. + * Shared transcript: progressively renders the paginated backfill, then — when + * ``isLive`` — opens an SSE stream pinned at the trailing offset and appends + * chunks until ``agent_call.finished``. Used by the agent-call and work-item + * pages; the only difference is ``basePath``. */ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTranscriptProps) { const transcriptKey = `${basePath}:${stream}` @@ -36,20 +36,24 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc let offset = 0 let text = '' let eof = false + let ok: boolean do { const res = await fetch( `${basePath}?stream=${stream}&offset=${offset}&limit=${TRANSCRIPT_CHUNK_LIMIT}`, { headers: { Accept: 'application/json' } }, ) - if (!res.ok) break - const chunk = (await res.json()) as { text: string; nextOffset: number; eof: boolean } - text += chunk.text - offset = chunk.nextOffset - eof = chunk.eof - } while (!cancelled && !isLive && !eof) - if (!cancelled) { + ok = res.ok + if (ok) { + const chunk = (await res.json()) as { text: string; nextOffset: number; eof: boolean } + text += chunk.text + offset = chunk.nextOffset + eof = chunk.eof + } + if (cancelled) return + // Publish after every chunk so a long backfill renders as it arrives, + // and once more on a failed fetch so the view stops saying "loading". setInitial({ key: transcriptKey, text, nextOffset: offset, eof }) - } + } while (ok && !isLive && !eof) })() return () => { cancelled = true @@ -60,36 +64,34 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc return
loading transcript…
} - return ( - - ) + if (isLive) { + return ( + + ) + } + + return } function RunTranscriptLive({ eventsUrl, initialText, - initialEof, - isLive, }: { eventsUrl: string initialText: string - initialEof: boolean - isLive: boolean }) { const [text, setText] = useState(initialText) - const [complete, setComplete] = useState(!isLive && initialEof) + const [complete, setComplete] = useState(false) // Gate ``enabled`` on ``!complete`` so useSSE closes the EventSource on // ``agent_call.finished`` — otherwise a native EventSource auto-reconnects to the // offset-pinned URL and replays the whole file, duplicating the transcript. useSSE(eventsUrl, { - enabled: isLive && !complete, + enabled: !complete, handlers: useMemo( () => ({ 'transcript.chunk': (payload) => {