From 7b586a97798719e48972b30d3d54430feaddca30 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Sat, 25 Jul 2026 22:29:44 +0000 Subject: [PATCH 1/5] Cache and progressively render transcripts --- backend/druks/extensions/base.py | 11 ++- backend/tests/test_api_runs.py | 30 ++++++- .../src/components/RunTranscript.test.tsx | 88 +++++++++++++++++++ frontend/src/components/RunTranscript.tsx | 19 ++-- 4 files changed, 140 insertions(+), 8 deletions(-) create mode 100644 frontend/src/components/RunTranscript.test.tsx diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index a7fb84d4..7f6f6353 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -273,11 +273,12 @@ 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 @@ -292,6 +293,7 @@ async def get_transcript( call_id: str, stream: Literal["stdout", "stderr"], engine: EngineDep, + response: Response, offset: int = 0, limit: int = default_limit, ) -> TranscriptChunk: @@ -304,6 +306,13 @@ async def get_transcript( chunk = reads.read_transcript_chunk(engine, call_id, stream, offset=offset, limit=limit) if not chunk: raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") + call = AgentCall.get(call_id) + if not call: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") + if call.get_live_status() == AgentCallStatus.RUNNING: + response.headers["Cache-Control"] = "no-store" + else: + response.headers["Cache-Control"] = "public, max-age=31536000, immutable" return chunk @router.get("/stream", response_class=StreamingResponse) diff --git a/backend/tests/test_api_runs.py b/backend/tests/test_api_runs.py index 5669bb8e..1f03d30a 100644 --- a/backend/tests/test_api_runs.py +++ b/backend/tests/test_api_runs.py @@ -21,7 +21,7 @@ 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) -> str: from conftest import finish_agent_run, seed_agent_run run = seed_agent_run() @@ -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,30 @@ def test_transcript_range_fetch_paginates( assert data["eof"] is True +def test_transcript_range_fetch_running_call_disables_cache( + client: TestClient, + tmp_path: Path, + db_session, +): + run_id = _seed_run(tmp_path=tmp_path, finished=False) + + response = client.get( + f"/api/build/transcripts/{run_id}", + params={"stream": "stdout", "limit": 5}, + ) + + assert response.status_code == 200 + assert response.headers["cache-control"] == "no-store" + assert response.json() == { + "callId": run_id, + "stream": "stdout", + "offset": 0, + "nextOffset": 5, + "eof": False, + "text": "hello", + } + + 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..493546a2 --- /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..e8c6eb8e 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,6 +36,7 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc let offset = 0 let text = '' let eof = false + let receivedChunk = false do { const res = await fetch( `${basePath}?stream=${stream}&offset=${offset}&limit=${TRANSCRIPT_CHUNK_LIMIT}`, @@ -46,8 +47,12 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc text += chunk.text offset = chunk.nextOffset eof = chunk.eof + receivedChunk = true + if (!cancelled) { + setInitial({ key: transcriptKey, text, nextOffset: offset, eof }) + } } while (!cancelled && !isLive && !eof) - if (!cancelled) { + if (!cancelled && !receivedChunk) { setInitial({ key: transcriptKey, text, nextOffset: offset, eof }) } })() @@ -60,6 +65,10 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc return 
loading transcript…
} + if (!isLive) { + return + } + return ( Date: Sat, 25 Jul 2026 22:38:43 +0000 Subject: [PATCH 2/5] Cover abandoned transcript caching and narrow SSE handlers --- backend/tests/test_api_runs.py | 33 +++++++++++++++++++ .../src/components/RunTranscript.test.tsx | 4 +-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/backend/tests/test_api_runs.py b/backend/tests/test_api_runs.py index 1f03d30a..ceab2308 100644 --- a/backend/tests/test_api_runs.py +++ b/backend/tests/test_api_runs.py @@ -111,6 +111,39 @@ def test_transcript_range_fetch_running_call_disables_cache( } +def test_transcript_range_fetch_derived_abandoned_call_uses_immutable_cache( + client: TestClient, + tmp_path: Path, + db_session, +): + from druks.durable.dbos_state import workflow_status + from sqlalchemy import update + + run_id = _seed_run(tmp_path=tmp_path, finished=False) + db_session.execute( + update(workflow_status) + .where(workflow_status.c.workflow_uuid == run_id) + .values(status="ERROR") + ) + db_session.expire_all() + + response = client.get( + f"/api/build/transcripts/{run_id}", + params={"stream": "stdout", "limit": 5}, + ) + + assert response.status_code == 200 + assert response.headers["cache-control"] == "public, max-age=31536000, immutable" + assert response.json() == { + "callId": run_id, + "stream": "stdout", + "offset": 0, + "nextOffset": 5, + "eof": False, + "text": "hello", + } + + 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 index 493546a2..71bbe951 100644 --- a/frontend/src/components/RunTranscript.test.tsx +++ b/frontend/src/components/RunTranscript.test.tsx @@ -72,13 +72,13 @@ describe('RunTranscript', () => { act(() => { const handlers = useSSEMock.mock.calls.at(-1)?.[1].handlers - handlers?.['transcript.chunk']({ text: 'streamed row\n' }) + 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']({}) + handlers?.['agent_call.finished']?.({}) }) expect(useSSEMock).toHaveBeenLastCalledWith( '/api/build/transcripts/call-2/stream?stream=stdout&offset=12', From df3af7872254681f0f1a5f853fb114546dbcf071 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Sat, 25 Jul 2026 22:43:59 +0000 Subject: [PATCH 3/5] Fix abandoned transcript cache test setup --- backend/tests/test_api_runs.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/backend/tests/test_api_runs.py b/backend/tests/test_api_runs.py index ceab2308..fa8d09e2 100644 --- a/backend/tests/test_api_runs.py +++ b/backend/tests/test_api_runs.py @@ -116,26 +116,29 @@ def test_transcript_range_fetch_derived_abandoned_call_uses_immutable_cache( tmp_path: Path, db_session, ): + from druks.durable import AgentCall from druks.durable.dbos_state import workflow_status from sqlalchemy import update - run_id = _seed_run(tmp_path=tmp_path, finished=False) + call_id = _seed_run(tmp_path=tmp_path, finished=False) + call = AgentCall.get(call_id) + assert call is not None db_session.execute( update(workflow_status) - .where(workflow_status.c.workflow_uuid == run_id) + .where(workflow_status.c.workflow_uuid == call.run_id) .values(status="ERROR") ) db_session.expire_all() response = client.get( - f"/api/build/transcripts/{run_id}", + 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" assert response.json() == { - "callId": run_id, + "callId": call_id, "stream": "stdout", "offset": 0, "nextOffset": 5, From 251b940cdf0dd743c75b4ef2aa3d2a0df0ba5d4b Mon Sep 17 00:00:00 2001 From: Paulo Date: Sun, 26 Jul 2026 01:23:28 +0200 Subject: [PATCH 4/5] Cache policy belongs to the call; the backfill needs no flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The route fetched the chunk, 404'd, then fetched the call and 404'd again with the same message — two queries guarding one condition. It now finds the call first and asks it how cacheable its transcript is, so read_transcript_chunk takes the call it already has and stops carrying a not-found arm nobody reads. receivedChunk existed only so a first-fetch failure still cleared "loading". The loop condition says that directly. The non-live path returns the static transcript, which retires RunTranscriptLive's isLive/initialEof plumbing. The abandoned-call test seeds a failed run instead of writing DBOS's status table by hand. --- backend/druks/durable/models.py | 8 ++++ backend/druks/durable/reads.py | 19 ++++----- backend/druks/extensions/base.py | 12 +----- backend/tests/conftest.py | 3 +- backend/tests/test_agent_routes.py | 4 +- backend/tests/test_api_runs.py | 44 ++++--------------- frontend/src/components/RunTranscript.tsx | 51 ++++++++++------------- 7 files changed, 52 insertions(+), 89 deletions(-) diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index f5f6f5a5..6600d08b 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -337,6 +337,14 @@ def get_live_status(self) -> str: return "running" return "abandoned" + @property + def transcript_cache_control(self) -> str: + # Only a running call still appends to its log. Every other state has + # written its last byte, so that byte range never changes again. + if self.get_live_status() == AgentCallStatus.RUNNING: + return "no-store" + return "public, max-age=31536000, immutable" + @property def artifact_dir(self) -> str: return str(load_settings().artifacts_dir / f"run-{self.run_id}") 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 7f6f6353..a8c88c83 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -278,7 +278,6 @@ def _get_transcript_routes(cls) -> "APIRouter": 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 @@ -292,7 +291,6 @@ 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, @@ -303,17 +301,11 @@ 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: - raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") call = AgentCall.get(call_id) if not call: raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") - if call.get_live_status() == AgentCallStatus.RUNNING: - response.headers["Cache-Control"] = "no-store" - else: - response.headers["Cache-Control"] = "public, max-age=31536000, immutable" - return chunk + response.headers["Cache-Control"] = call.transcript_cache_control + 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 fa8d09e2..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, finished: bool = True) -> 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 @@ -87,48 +87,30 @@ def test_transcript_range_fetch_paginates( assert data["eof"] is True -def test_transcript_range_fetch_running_call_disables_cache( +def test_transcript_of_a_running_call_is_never_cached( client: TestClient, tmp_path: Path, db_session, ): - run_id = _seed_run(tmp_path=tmp_path, finished=False) + call_id = _seed_run(tmp_path=tmp_path, finished=False) response = client.get( - f"/api/build/transcripts/{run_id}", + f"/api/build/transcripts/{call_id}", params={"stream": "stdout", "limit": 5}, ) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" - assert response.json() == { - "callId": run_id, - "stream": "stdout", - "offset": 0, - "nextOffset": 5, - "eof": False, - "text": "hello", - } -def test_transcript_range_fetch_derived_abandoned_call_uses_immutable_cache( +def test_transcript_of_an_abandoned_call_is_cached_immutably( client: TestClient, tmp_path: Path, db_session, ): - from druks.durable import AgentCall - from druks.durable.dbos_state import workflow_status - from sqlalchemy import update - - call_id = _seed_run(tmp_path=tmp_path, finished=False) - call = AgentCall.get(call_id) - assert call is not None - db_session.execute( - update(workflow_status) - .where(workflow_status.c.workflow_uuid == call.run_id) - .values(status="ERROR") - ) - db_session.expire_all() + # 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}", @@ -137,14 +119,6 @@ def test_transcript_range_fetch_derived_abandoned_call_uses_immutable_cache( assert response.status_code == 200 assert response.headers["cache-control"] == "public, max-age=31536000, immutable" - assert response.json() == { - "callId": call_id, - "stream": "stdout", - "offset": 0, - "nextOffset": 5, - "eof": False, - "text": "hello", - } def test_transcript_missing_file_returns_eof( diff --git a/frontend/src/components/RunTranscript.tsx b/frontend/src/components/RunTranscript.tsx index e8c6eb8e..552c5f88 100644 --- a/frontend/src/components/RunTranscript.tsx +++ b/frontend/src/components/RunTranscript.tsx @@ -36,25 +36,24 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc let offset = 0 let text = '' let eof = false - let receivedChunk = 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 - receivedChunk = true - if (!cancelled) { - setInitial({ key: transcriptKey, text, nextOffset: offset, eof }) + 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 } - } while (!cancelled && !isLive && !eof) - if (!cancelled && !receivedChunk) { + 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 @@ -65,40 +64,34 @@ export function RunTranscript({ basePath, stream = 'stdout', isLive }: RunTransc return 
loading transcript…
} - if (!isLive) { - return + if (isLive) { + return ( + + ) } - 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) => { From 7d98ede5615ee5aa5835f9dde9eb191578e2057e Mon Sep 17 00:00:00 2001 From: Paulo Date: Sun, 26 Jul 2026 01:45:06 +0200 Subject: [PATCH 5/5] Keep the cache directive at the wire, not on the call --- backend/druks/durable/models.py | 8 -------- backend/druks/extensions/base.py | 9 ++++++++- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index 6600d08b..f5f6f5a5 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -337,14 +337,6 @@ def get_live_status(self) -> str: return "running" return "abandoned" - @property - def transcript_cache_control(self) -> str: - # Only a running call still appends to its log. Every other state has - # written its last byte, so that byte range never changes again. - if self.get_live_status() == AgentCallStatus.RUNNING: - return "no-store" - return "public, max-age=31536000, immutable" - @property def artifact_dir(self) -> str: return str(load_settings().artifacts_dir / f"run-{self.run_id}") diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index a8c88c83..f76a2f44 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -278,12 +278,16 @@ def _get_transcript_routes(cls) -> "APIRouter": 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"]) @@ -304,7 +308,10 @@ async def get_transcript( call = AgentCall.get(call_id) if not call: raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.") - response.headers["Cache-Control"] = call.transcript_cache_control + 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)