diff --git a/apps/api/src/routes/reports.ts b/apps/api/src/routes/reports.ts index d6dd7aa32..093a6d65a 100644 --- a/apps/api/src/routes/reports.ts +++ b/apps/api/src/routes/reports.ts @@ -12,6 +12,7 @@ import { BulkTriggerReportsInputSchema, CreateReportInputSchema, InsightResponseSchema, + LogEventSchema, ReportResponseSchema, TriggerReportsInputSchema, evaluateTrigger, @@ -337,6 +338,31 @@ apiRoute(ctx.app, ctx.registry, { }, }); +apiRoute(ctx.app, ctx.registry, { + method: "get", + path: "/api/v1/reports/:id/logs/events", + tags: ["Reports"], + summary: "Get report log events (JSON)", + params: z.object({ id: z.string() }), + response: z.array(LogEventSchema), + errorResponses: { 404: { description: "Report not found" }, 502: { description: "Log storage unavailable" } }, + handler: async (req, res) => { + const { id } = req.params; + const report = await ctx.reportCollection.findOne({ _id: id }); + if (!report) { + res.status(404).json({ error: "Report not found" }); + return; + } + + try { + res.json(await ctx.blobStorage.getLogEvents(id)); + } catch (error) { + console.error(`Failed to read logs for report ${id}:`, error); + res.status(502).json({ error: "Cannot connect to log storage" }); + } + }, +}); + apiRoute(ctx.app, ctx.registry, { method: "get", path: "/api/v1/reports/:id/logs", diff --git a/apps/api/src/routes/requests/logs.test.ts b/apps/api/src/routes/requests/logs.test.ts index 44153cb70..0227cca14 100644 --- a/apps/api/src/routes/requests/logs.test.ts +++ b/apps/api/src/routes/requests/logs.test.ts @@ -63,6 +63,69 @@ describe("SSE log endpoints — blob replay", () => { _injectTestDependencies(mocks); }); + describe("JSON log endpoints", () => { + it("reads the current attempt's stored events without opening an SSE stream", async () => { + const logs = [makeLogEvent("completed")]; + (mocks.collection.findOne as any).mockResolvedValue({ + _id: "request-1", + run: { _id: "attempt-1", status: "done" }, + }); + (mocks.blobStorage.getLogEvents as any).mockResolvedValue(logs); + + const res = await request(testServer()).get("/api/v1/requests/request-1/logs/events"); + + expect(res.status).toBe(200); + expect(res.body).toEqual(logs); + expect(mocks.blobStorage.getLogEvents).toHaveBeenCalledWith("request-1", "attempt-1"); + expect(mocks.collection.watch).not.toHaveBeenCalled(); + }); + + it("uses the attempt URL for historical run logs", async () => { + const logs = [makeLogEvent("previous attempt")]; + (mocks.collection.findOne as any).mockResolvedValue({ + _id: "request-1", + run: { _id: "attempt-2", status: "done" }, + }); + (mocks.runsCollection.findOne as any).mockResolvedValue({ + _id: "attempt-1", + requestId: "request-1", + status: "done", + logsUrl: "https://logs/attempt-1", + }); + (mocks.blobStorage.getLogEvents as any).mockResolvedValue(logs); + + const res = await request(testServer()).get("/api/v1/requests/request-1/runs/attempt-1/logs/events"); + + expect(res.status).toBe(200); + expect(res.body).toEqual(logs); + expect(mocks.blobStorage.getLogEvents).toHaveBeenCalledWith("https://logs/attempt-1"); + }); + + it("returns a storage error for JSON log reads", async () => { + (mocks.collection.findOne as any).mockResolvedValue({ + _id: "request-1", + run: { _id: "attempt-1", status: "done" }, + }); + (mocks.blobStorage.getLogEvents as any).mockRejectedValue(new Error("blob unavailable")); + + const res = await request(testServer()).get("/api/v1/requests/request-1/logs/events"); + + expect(res.status).toBe(502); + }); + + it("reads stored report events as JSON", async () => { + const logs = [makeLogEvent("report complete")]; + (mocks.reportCollection.findOne as any).mockResolvedValue({ _id: "report-1", status: "completed" }); + (mocks.blobStorage.getLogEvents as any).mockResolvedValue(logs); + + const res = await request(testServer()).get("/api/v1/reports/report-1/logs/events"); + + expect(res.status).toBe(200); + expect(res.body).toEqual(logs); + expect(mocks.blobStorage.getLogEvents).toHaveBeenCalledWith("report-1"); + }); + }); + // ========================================================================= // GET /api/v1/requests/:id/logs // ========================================================================= diff --git a/apps/api/src/routes/requests/logs.ts b/apps/api/src/routes/requests/logs.ts index 3d732834b..189245154 100644 --- a/apps/api/src/routes/requests/logs.ts +++ b/apps/api/src/routes/requests/logs.ts @@ -2,7 +2,7 @@ // Licensed under the MIT License. import { z } from "zod"; -import type { RunState } from "shared"; +import { LogEventSchema, type LogEvent, type RunState } from "shared"; import type { Response } from "express"; import { apiRoute } from "../../openapi/api-route.js"; import type { RouteContext } from "../../route-context.js"; @@ -10,6 +10,13 @@ import { subscribeClient, unsubscribeClient } from "../../utils/sse.js"; import type { SSEClient } from "../../utils/sse.js"; import { resolveRunForRequest } from "./resolve-run.js"; +/** Read the same attempt log blob used by SSE replay. */ +function readRunLogs(ctx: RouteContext, id: string, run: RunState): Promise { + return run.logsUrl + ? ctx.blobStorage.getLogEvents(run.logsUrl) + : ctx.blobStorage.getLogEvents(id, run._id); +} + /** * Shared SSE log-streaming handler. * Works for both request-level (uses request.run) and per-run (uses resolved run) endpoints. @@ -32,10 +39,7 @@ async function handleLogs( // Replay logs from blob storage if (fromStart) { try { - const logsUrl = targetRun.logsUrl; - const pastLogs = logsUrl - ? await ctx.blobStorage.getLogEvents(logsUrl) - : await ctx.blobStorage.getLogEvents(id, targetRun._id); + const pastLogs = await readRunLogs(ctx, id, targetRun); for (const log of pastLogs) { res.write(`data: ${JSON.stringify(log)}\n\n`); } @@ -145,6 +149,58 @@ async function handleLogs( export function registerRequestsLogsRoutes(ctx: RouteContext): void { +// Plain JSON replay for completed requests avoids an SSE connection. +apiRoute(ctx.app, ctx.registry, { + method: "get", + path: "/api/v1/requests/:id/logs/events", + tags: ["Requests"], + summary: "Get request log events (JSON)", + params: z.object({ id: z.string() }), + response: z.array(LogEventSchema), + errorResponses: { 404: { description: "Request not found" }, 502: { description: "Log storage unavailable" } }, + handler: async (req, res) => { + const { id } = req.params; + const resource = await ctx.requestCollection.findOne({ _id: id }); + if (!resource?.run) { + res.status(404).json({ error: "Request not found" }); + return; + } + + try { + res.json(await readRunLogs(ctx, id, resource.run)); + } catch (error) { + console.error(`Failed to read logs for request ${id}:`, error); + res.status(502).json({ error: "Cannot connect to log storage" }); + } + }, +}); + +// Historical attempts use the same request ownership check as their SSE route. +apiRoute(ctx.app, ctx.registry, { + method: "get", + path: "/api/v1/requests/:id/runs/:runId/logs/events", + tags: ["Requests"], + summary: "Get log events for an attempt (JSON)", + params: z.object({ id: z.string(), runId: z.string() }), + response: z.array(LogEventSchema), + errorResponses: { 404: { description: "Run not found" }, 502: { description: "Log storage unavailable" } }, + handler: async (req, res) => { + const { id, runId } = req.params; + const resolved = await resolveRunForRequest(ctx, id, runId); + if ("error" in resolved) { + res.status(resolved.status).json({ error: resolved.error }); + return; + } + + try { + res.json(await readRunLogs(ctx, id, resolved.run)); + } catch (error) { + console.error(`Failed to read logs for request ${id}, attempt ${runId}:`, error); + res.status(502).json({ error: "Cannot connect to log storage" }); + } + }, +}); + // Request-level log streaming apiRoute(ctx.app, ctx.registry, { method: "get", diff --git a/apps/portal/src/components/LogViewer.stories.tsx b/apps/portal/src/components/LogViewer.stories.tsx index 959e18f06..30adbc661 100644 --- a/apps/portal/src/components/LogViewer.stories.tsx +++ b/apps/portal/src/components/LogViewer.stories.tsx @@ -57,8 +57,8 @@ export const StreamingLive: Story = { }, }; -/** Completed run: stream finished, no auto-scroll affordances shown. */ -export const StreamComplete: Story = { +/** Stored JSON events for a completed run; no SSE connection is needed. */ +export const CompletedRun: Story = { args: { runId: "demo-run", logs: sampleLogs, diff --git a/apps/portal/src/lib/api.ts b/apps/portal/src/lib/api.ts index 96cd39566..feb50f313 100644 --- a/apps/portal/src/lib/api.ts +++ b/apps/portal/src/lib/api.ts @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -import type { Run, RunState, CriteriaDocument, CriteriaGraphData, GeneratePromptResponse, AnalysisResponse, PromptFeatureDocument, Report, BulkReportStatus, BulkReportSummary, ReportTemplate, ReportTrigger, ReportTemplateSystemPrompt, KeyDocument, KeyValidationResult, CreateKeyRequest, UpdateKeyRequest, CodingAgent, AgentVersion, McpServerDocument, CreateMcpServerRequest, UpdateMcpServerRequest, BulkResubmitOverrides, Insight, InsightWithReference, TaskPrompt, TaskPromptFeatureExtractionResult, Model, FeatureFlag, SkillDocument, SkillSearchResult, SkillDiscoveryResult, SkillRevisionDocument, CodebaseDocument, CodebaseRevisionDocument, CodebaseSourceType, ResourceDocument, ResourceRevisionDocument, ResourceBindingSpec, CreateResourceBody, CreateResourceRevisionBody, ExtensionDocument, ExtensionSearchResult, ExtensionVersionInfo, MdpResponse, AccountDocument, CreateAccountRequest, UpdateAccountRequest, ProfileWithVersion, ProfileVersionDocument, ProfileDocument, RunGroup, RunFacetsResponse, CursorPaginatedResponse, IterationOp, GateConfig, GateId, PromptType, RunSortField, RunSortDir } from "@/types"; +import type { Run, RunState, LogEvent, CriteriaDocument, CriteriaGraphData, GeneratePromptResponse, AnalysisResponse, PromptFeatureDocument, Report, BulkReportStatus, BulkReportSummary, ReportTemplate, ReportTrigger, ReportTemplateSystemPrompt, KeyDocument, KeyValidationResult, CreateKeyRequest, UpdateKeyRequest, CodingAgent, AgentVersion, McpServerDocument, CreateMcpServerRequest, UpdateMcpServerRequest, BulkResubmitOverrides, Insight, InsightWithReference, TaskPrompt, TaskPromptFeatureExtractionResult, Model, FeatureFlag, SkillDocument, SkillSearchResult, SkillDiscoveryResult, SkillRevisionDocument, CodebaseDocument, CodebaseRevisionDocument, CodebaseSourceType, ResourceDocument, ResourceRevisionDocument, ResourceBindingSpec, CreateResourceBody, CreateResourceRevisionBody, ExtensionDocument, ExtensionSearchResult, ExtensionVersionInfo, MdpResponse, AccountDocument, CreateAccountRequest, UpdateAccountRequest, ProfileWithVersion, ProfileVersionDocument, ProfileDocument, RunGroup, RunFacetsResponse, CursorPaginatedResponse, IterationOp, GateConfig, GateId, PromptType, RunSortField, RunSortDir } from "@/types"; import type { Project, CreateProjectRequest, UpdateProjectRequest } from "@/types"; import { qs } from "./url"; @@ -394,6 +394,9 @@ export const api = { return `${BASE}/requests/${id}/logs?fromStart=${fromStart}`; }, + /** Fetch stored request logs as JSON for a terminal run. */ + getLogEvents: (id: string): Promise => request(`/requests/${id}/logs/events`), + // ─── Per-run artifact URLs (for historical attempts) ───────────────────── /** SSE endpoint URL for log streaming of a specific attempt */ @@ -401,6 +404,10 @@ export const api = { return `${BASE}/requests/${requestId}/runs/${runId}/logs?fromStart=${fromStart}`; }, + /** Fetch stored logs for a historical execution attempt. */ + getRunLogEvents: (requestId: string, runId: string): Promise => + request(`/requests/${requestId}/runs/${runId}/logs/events`), + /** HAR file download URL for a specific attempt */ runHarUrl: (requestId: string, runId: string, iteration?: number): string => { const qs = iteration ? `?iteration=${iteration}` : ""; @@ -786,6 +793,9 @@ export const api = { return `${BASE}/reports/${id}/logs?fromStart=${fromStart}`; }, + /** Fetch stored report logs as JSON for a terminal report. */ + getReportLogEvents: (id: string): Promise => request(`/reports/${id}/logs/events`), + // ─── Report Templates ───────────────────────────────────────────────────── /** Get the default system prompt used when no template override is set */ diff --git a/apps/portal/src/pages/ReportDetail.tsx b/apps/portal/src/pages/ReportDetail.tsx index 685e69792..298b01ff0 100644 --- a/apps/portal/src/pages/ReportDetail.tsx +++ b/apps/portal/src/pages/ReportDetail.tsx @@ -44,18 +44,27 @@ export function ReportDetail() { const isActive = report?.status === "pending" || report?.status === "generating"; - // The SSE endpoint handles completed reports by replaying blob logs then closing. + const isTerminal = report?.status === "completed" || report?.status === "failed"; + const { data: storedLogs, error: storedLogsError } = useQuery({ + queryKey: ["report-log-events", id], + queryFn: () => api.getReportLogEvents(id!), + enabled: !!id && isTerminal, + }); + + // Active reports stream; terminal reports use the JSON log response. const logStream = useLogStream({ id: report?._id ?? "", - enabled: !!report, + enabled: !!report && !isTerminal, fromStart: true, urlBuilder: api.reportLogsUrl, }); - const effectiveLogs = logStream.logs; + const effectiveLogs = isTerminal ? (storedLogs ?? logStream.logs) : logStream.logs; const effectiveIsConnected = isActive ? logStream.isConnected : false; - const effectiveIsDone = logStream.isDone; - const effectiveError = logStream.error; + const effectiveIsDone = isTerminal || logStream.isDone; + const effectiveError = isTerminal + ? (storedLogsError instanceof Error ? storedLogsError.message : null) + : logStream.error; const copyId = () => { navigator.clipboard.writeText(id ?? ""); diff --git a/apps/portal/src/pages/RunDetail.tsx b/apps/portal/src/pages/RunDetail.tsx index 77ac6b0fa..9b99b49b8 100644 --- a/apps/portal/src/pages/RunDetail.tsx +++ b/apps/portal/src/pages/RunDetail.tsx @@ -312,11 +312,18 @@ export function RunDetail() { const isActive = activeRun?.status === "pending" || activeRun?.status === "processing"; // Note: "done" is terminal — not active, no log streaming needed - // Lift the log stream so it can be shared between LogViewer and CriteriaGraphView - // Must be called unconditionally (before any early returns) per Rules of Hooks - // The SSE endpoint handles completed runs by replaying blob logs then closing. + // Share live logs between LogViewer and CriteriaGraphView. Terminal attempts + // read the same stored events over JSON without opening an SSE connection. // When viewing a historical run, use the per-run logs endpoint; otherwise use the // default endpoint with attemptNumber for reconnection on retry. + const isTerminal = activeRun?.status === "done"; + const { data: storedLogs, error: storedLogsError } = useQuery({ + queryKey: ["run-log-events", run?._id, activeRun?._id], + queryFn: () => isViewingHistorical + ? api.getRunLogEvents(run!._id, activeRun!._id) + : api.getLogEvents(run!._id), + enabled: !!run && !!activeRun && isTerminal, + }); const logStreamUrlBuilder = useMemo(() => { if (isViewingHistorical && activeRun?._id && run?._id) { const requestId = run._id; @@ -328,16 +335,20 @@ export function RunDetail() { const logStream = useLogStream({ id: run?._id ?? "", - enabled: !!run, + enabled: !!run && !isTerminal, fromStart: true, attemptNumber: activeRun?.attemptNumber, urlBuilder: logStreamUrlBuilder, }); - const effectiveLogs = logStream.logs; + const effectiveLogs = isTerminal + ? (storedLogs ?? (isViewingHistorical ? [] : logStream.logs)) + : logStream.logs; const effectiveIsConnected = isActive ? logStream.isConnected : false; - const effectiveIsDone = logStream.isDone; - const effectiveError = logStream.error; + const effectiveIsDone = isTerminal || logStream.isDone; + const effectiveError = isTerminal + ? (storedLogsError instanceof Error ? storedLogsError.message : null) + : logStream.error; // Fetch linked task prompt (if present) — provides prompt features const taskPromptId = run?.taskPromptId; diff --git a/docs/research/realtime-data-flow.md b/docs/research/realtime-data-flow.md index abe77fd3d..2d563237b 100644 --- a/docs/research/realtime-data-flow.md +++ b/docs/research/realtime-data-flow.md @@ -13,7 +13,16 @@ The portal uses **two mechanisms** for real-time updates: GET /api/v1/requests/:id/logs?fromStart=true ``` -Defined in `apps/api/src/index.ts`. Accepts an optional `fromStart=true` query param to replay existing logs from MongoDB before switching to live streaming. +Defined in `apps/api/src/routes/requests/logs.ts`. Accepts an optional `fromStart=true` query param to replay existing logs from Blob Storage before switching to live streaming. + +The Portal uses SSE only while a request attempt or report is active. Once it is +terminal, it fetches the stored events as JSON from +`GET /api/v1/requests/:id/logs/events`, +`GET /api/v1/requests/:id/runs/:runId/logs/events` (historical attempt), or +`GET /api/v1/reports/:id/logs/events`. These endpoints return the same ordered +`LogEvent[]` as SSE replay without holding a stream open. If an active stream +finishes while the page is open, its events remain visible until the JSON read +completes. The SSE endpoints remain available to other clients for replay. ### Server-Side Flow @@ -60,7 +69,7 @@ read lazily by clients (no SSE stream — fetched on demand by ID/iteration): | Artifact | Container | Path | Producer | Consumer | |----------|-----------|------|----------|----------| -| Log events | `logs` | `{requestId}/runs/{runId}/run.jsonl` | `BlobStorage.appendLogEvent` (workers) | SSE replay + `GET /api/v1/requests/:id/logs` | +| Log events | `logs` | `{requestId}/runs/{runId}/run.jsonl` | `BlobStorage.appendLogEvent` (workers) | Active SSE replay; terminal JSON log-events endpoints | | Per-iteration tool calls | `snapshots` | `{requestId}/runs/{runId}/iteration-{n}/tool-calls.jsonl` | `BlobStorage.writeToolCalls` (`multi-turn-loop`) | `GET /api/v1/requests/:id/tool-calls?iteration=N`, consumed by `useAllTurnsToolCalls` | Tool-call JSONL was introduced (#812) to keep unbounded per-iteration tool-call