feat(web): add Turn timeline and usage

apps/web/e2e/agents-lifecycle.spec.ts

@@ -5,6 +5,7 @@ const fixtureBaseUrl = `http://127.0.0.1:${process.env.AGENTS_FIXTURE_PORT ?? 18
 interface FixtureRequest {
   method: string;
   path: string;
+  query?: string;
   beta: string | null;
   authorizationPresent: boolean;
   idempotencyKeyPresent: boolean;
@@ -22,6 +23,11 @@ async function controlFixture(request: APIRequestContext, control: Record<string
   expect(response.ok()).toBe(true);
 }
 
+async function emitTurnFixture(request: APIRequestContext, status: "completed" | "failed" | "cancelled") {
+  const response = await request.post(`${fixtureBaseUrl}/__fixture/emit-turn`, { data: { status } });
+  expect(response.ok()).toBe(true);
+}
+
 async function fixtureRequests(request: APIRequestContext): Promise<FixtureRequest[]> {
   const response = await request.get(`${fixtureBaseUrl}/__fixture/requests`);
   expect(response.ok()).toBe(true);
@@ -502,6 +508,100 @@ test("applies a buffered live Environment event after an earlier durable snapsho
   await expect(panel).not.toContainText("Pending");
 });
 
+test("loads every Turn page, reconciles terminal events, and keeps failures beside conversation Items", async ({ page, request }, testInfo) => {
+  await resetFixture(request);
+  await controlFixture(request, {
+    turnsScenario: 1,
+    turnsPageSize: 2,
+  });
+  await page.goto("/");
+  await expect(page.getByText("listening", { exact: true })).toBeVisible();
+
+  const timeline = page.getByRole("region", { name: "Turn timeline" });
+  await expect(timeline).toContainText("7 observed Turns");
+  for (const status of ["Queued", "In progress", "Waiting", "Completed", "Failed", "Cancelled"]) {
+    await expect(timeline.getByRole("img", { name: `Turn status: ${status}` }).first()).toBeVisible();
+  }
+  await expect(timeline).toContainText("Running ·");
+  await expect(timeline.locator('[data-turn-id="turn_completed"]')).toContainText("7s");
+  await expect(timeline.getByRole("region", { name: "Session aggregate usage" })).toContainText("26");
+  await expect(timeline.locator('[data-turn-id="turn_completed"]').getByRole("group", { name: "Usage for Turn turn_completed" })).toContainText("13");
+  const failed = timeline.locator('[data-turn-id="turn_failed"]');
+  await expect(failed).toContainText("The execution could not complete.");
+  await expect(failed).toContainText("1 linked Item");
+  await expect(page.getByText("Persisted input before the Turn failed.")).toBeVisible();
+  await expect(timeline).toContainText("1 Item is not associated with an observed Turn yet.");
+
+  const readsBeforeTerminal = (await fixtureRequests(request)).filter((entry) => (
+    entry.method === "GET" && entry.path.endsWith("/turns")
+  ));
+  expect(readsBeforeTerminal.length).toBeGreaterThanOrEqual(4);
+  expect(readsBeforeTerminal.some((entry) => entry.query === "?limit=100&order=asc")).toBe(true);
+  expect(readsBeforeTerminal.some((entry) => entry.query?.includes("after=turn_in_progress"))).toBe(true);
+  expect(readsBeforeTerminal.every((entry) => entry.body === undefined)).toBe(true);
+
+  const terminal = timeline.locator('[data-turn-id="turn_terminal_refresh"]');
+  await expect(terminal).toHaveAttribute("data-turn-status", "in_progress");
+  await emitTurnFixture(request, "completed");
+  await expect(terminal).toHaveAttribute("data-turn-status", "completed");
+  await expect(terminal).toContainText("Turn usage");
+  await expect.poll(async () => (
+    await fixtureRequests(request)
+  ).filter((entry) => entry.method === "GET" && entry.path.endsWith("/turns")).length).toBeGreaterThan(readsBeforeTerminal.length);
+
+  await controlFixture(request, { turnsRetrieveStatus: 503 });
+  await page.getByRole("button", { name: "Recover durable state" }).click();
+  await expect(timeline.locator(".turn-timeline-failure")).toContainText("Couldn’t load Turn history");
+  await expect(timeline).toContainText("last observed Turn timeline remains visible");
+  await expect(page.getByText("Completed Turn output remains in the conversation.")).toBeVisible();
+  await expect(page.getByLabel("Message the Agent")).toBeVisible();
+
+  await page.setViewportSize({ width: 390, height: 844 });
+  await timeline.evaluate((element) => element.scrollIntoView({ block: "start" }));
+  const widths = await timeline.evaluate((element) => {
+    const box = element.getBoundingClientRect();
+    return {
+      viewport: innerWidth,
+      document: document.documentElement.scrollWidth,
+      body: document.body.scrollWidth,
+      left: box.left,
+      right: box.right,
+    };
+  });
+  expect(widths.document).toBeLessThanOrEqual(widths.viewport);
+  expect(widths.body).toBeLessThanOrEqual(widths.viewport);
+  expect(widths.left).toBeGreaterThanOrEqual(0);
+  expect(widths.right).toBeLessThanOrEqual(widths.viewport);
+  await attachElementScreenshot(timeline, testInfo, "narrow-turn-timeline");
+});
+
+test("drops a delayed Turn page after switching Sessions", async ({ page, request }) => {
+  await resetFixture(request);
+  await controlFixture(request, {
+    turnsScenario: 1,
+    turnsRetrieveDelayMs: 700,
+    turnsPageSize: 2,
+  });
+  await page.goto("/");
+  const timeline = page.getByRole("region", { name: "Turn timeline" });
+  await expect(page.getByText("Completed Turn output remains in the conversation.")).toBeVisible({ timeout: 1_500 });
+  await expect(timeline).toContainText("Loading every Turn page");
+  await page.getByRole("button", { name: "Agents" }).click();
+  await expect(page.getByRole("table", { name: "Agents" })).toBeVisible();
+  await page.getByRole("button", { name: /Start a Session with Second Agent/ }).click();
+
+  await expect(timeline).toContainText("No Turns reported yet.");
+  await page.waitForTimeout(3_000);
+  await expect(timeline).not.toContainText("turn_queued");
+  await expect(page.getByText("Completed Turn output remains in the conversation.")).toHaveCount(0);
+
+  const turnReads = (await fixtureRequests(request)).filter((entry) => (
+    entry.method === "GET" && entry.path.endsWith("/turns")
+  ));
+  expect(turnReads.some((entry) => entry.path.includes("session_snapshot"))).toBe(true);
+  expect(turnReads.some((entry) => entry.path.includes("session_created_"))).toBe(true);
+});
+
 test("renders Parsar patches as accessible read-only diffs in desktop and narrow themes", async ({ page, request }, testInfo) => {
   await resetFixture(request);
   await controlFixture(request, { itemsScenario: 1 });

apps/web/e2e/fixture-core.mjs

@@ -25,6 +25,26 @@ function patchItems() {
   ];
 }
 
+function observableTurns() {
+  return [
+    { id: "turn_queued", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "queued", created_at: baseline - 18, started_at: null, completed_at: null, error: null, usage: null },
+    { id: "turn_in_progress", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "in_progress", created_at: baseline - 17, started_at: baseline - 16, completed_at: null, error: null, usage: null },
+    { id: "turn_waiting", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "waiting", created_at: baseline - 15, started_at: baseline - 14, completed_at: null, error: null, usage: null },
+    { id: "turn_completed", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "completed", created_at: baseline - 13, started_at: baseline - 12, completed_at: baseline - 5, error: null, usage: { input_tokens: 10, output_tokens: 3, total_tokens: 13, input_tokens_details: { cached_tokens: 4 }, output_tokens_details: { reasoning_tokens: 2 } } },
+    { id: "turn_failed", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "failed", created_at: baseline - 4, started_at: baseline - 3, completed_at: baseline - 2, error: { code: "internal_error", message: "The execution could not complete." }, usage: null },
+    { id: "turn_cancelled", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "cancelled", created_at: baseline - 1, started_at: baseline, completed_at: baseline + 1, error: null, usage: null },
+    { id: "turn_terminal_refresh", agent_id: "agent_a", session_id: "session_snapshot", object: "agent.session.turn", status: "in_progress", created_at: baseline + 2, started_at: baseline + 3, completed_at: null, error: null, usage: null },
+  ];
+}
+
+function observableTurnItems() {
+  return [
+    { id: "turn_message", turn_id: "turn_completed", type: "message", status: "completed", role: "assistant", content: [{ type: "output_text", text: "Completed Turn output remains in the conversation." }] },
+    { id: "failed_input", turn_id: "turn_failed", type: "message", status: "completed", role: "user", content: [{ type: "input_text", text: "Persisted input before the Turn failed." }] },
+    { id: "unassociated", turn_id: "turn_not_loaded", type: "message", status: "completed", role: "assistant", content: [{ type: "output_text", text: "This Item is waiting for its Turn page." }] },
+  ];
+}
+
 function savedAgent(id, name, model, updatedAt) {
   return {
     id,
@@ -67,6 +87,7 @@ function initialState() {
       created_at: baseline - 20,
       last_active_at: baseline - 10,
     }],
+    turns: [],
     requests: [],
     controls: {
       retrieveDelayMs: 0,
@@ -78,6 +99,10 @@ function initialState() {
       sendStatus: 204,
       sendResponseLoss: 0,
       itemsScenario: 0,
+      turnsScenario: 0,
+      turnsRetrieveDelayMs: 0,
+      turnsRetrieveStatus: 200,
+      turnsPageSize: 2,
       environmentScenario: 0,
       environmentRetrieveDelayMs: 0,
       environmentRetrieveStatus: 200,
@@ -93,6 +118,24 @@ function initialState() {
   };
 }
 
+function applyTurnsScenario(value) {
+  const session = state.sessions[0];
+  if (!session) return;
+  if (value === 1) {
+    state.turns = observableTurns();
+    session.usage = {
+      input_tokens: 20,
+      output_tokens: 6,
+      total_tokens: 26,
+      input_tokens_details: { cached_tokens: 8 },
+      output_tokens_details: { reasoning_tokens: 4 },
+    };
+    return;
+  }
+  state.turns = [];
+  session.usage = null;
+}
+
 function applyEnvironmentScenario(value) {
   const session = state.sessions[0];
   if (!session) return;
@@ -126,6 +169,31 @@ function applyEnvironmentScenario(value) {
 }
 
 let state = initialState();
+const streamResponses = new Set();
+
+function emitTurnLifecycle(status) {
+  const index = state.turns.findIndex((turn) => turn.id === "turn_terminal_refresh");
+  const existing = state.turns[index];
+  if (!existing || !["completed", "failed", "cancelled"].includes(status)) return false;
+  const terminal = {
+    ...existing,
+    status,
+    completed_at: baseline + 10,
+    error: status === "failed" ? { code: "internal_error", message: "The execution could not complete." } : null,
+    usage: status === "completed" ? { input_tokens: 5, output_tokens: 2, total_tokens: 7, input_tokens_details: { cached_tokens: 1 }, output_tokens_details: { reasoning_tokens: 1 } } : null,
+  };
+  state.turns[index] = terminal;
+  state.sequence += 1;
+  const event = `id: turn_${state.sequence}\ndata: ${JSON.stringify({
+    type: `agent.session.turn.${status}`,
+    event_id: `turn_${state.sequence}`,
+    session_id: "session_snapshot",
+    turn_id: terminal.id,
+    turn: terminal,
+  })}\n\n`;
+  for (const stream of streamResponses) stream.write(event);
+  return true;
+}
 
 function sendJson(response, value, status = 200) {
   const body = JSON.stringify(value);
@@ -168,6 +236,7 @@ function recordRequest(request, url, body) {
   state.requests.push({
     method: request.method,
     path: url.pathname,
+    query: url.search,
     beta: request.headers["openai-beta"] ?? null,
     authorizationPresent: Boolean(request.headers.authorization),
     idempotencyKeyPresent: Boolean(request.headers["idempotency-key"]),
@@ -196,14 +265,23 @@ const server = http.createServer(async (request, response) => {
       return sendJson(response, { ready: true });
     }
     if (request.method === "POST" && url.pathname === "/__fixture/reset") {
+      for (const stream of streamResponses) stream.end();
+      streamResponses.clear();
       state = initialState();
       return sendJson(response, { reset: true });
     }
     if (request.method === "POST" && url.pathname === "/__fixture/control") {
       state.controls = { ...state.controls, ...await readJson(request) };
       applyEnvironmentScenario(state.controls.environmentScenario);
+      applyTurnsScenario(state.controls.turnsScenario);
       return sendJson(response, state.controls);
     }
+    if (request.method === "POST" && url.pathname === "/__fixture/emit-turn") {
+      const input = await readJson(request);
+      return emitTurnLifecycle(input.status)
+        ? sendJson(response, { emitted: true })
+        : sendError(response, 400, "Fixture terminal Turn is unavailable.");
+    }
     if (request.method === "GET" && url.pathname === "/__fixture/requests") {
       return sendJson(response, state.requests);
     }
@@ -319,7 +397,41 @@ const server = http.createServer(async (request, response) => {
     }
 
     const itemsMatch = url.pathname.match(/^\/v1\/agents\/sessions\/([^/]+)\/items$/);
-    if (request.method === "GET" && itemsMatch) return sendJson(response, page(state.controls.itemsScenario ? patchItems() : []));
+    if (request.method === "GET" && itemsMatch) {
+      const sessionId = decodeURIComponent(itemsMatch[1]);
+      const items = sessionId !== "session_snapshot"
+        ? []
+        : state.controls.itemsScenario
+          ? patchItems()
+          : state.controls.turnsScenario
+            ? observableTurnItems()
+            : [];
+      return sendJson(response, page(items));
+    }
+
+    const turnsMatch = url.pathname.match(/^\/v1\/agents\/sessions\/([^/]+)\/turns$/);
+    if (request.method === "GET" && turnsMatch) {
+      if (state.controls.turnsRetrieveDelayMs) await wait(state.controls.turnsRetrieveDelayMs);
+      if (state.controls.turnsRetrieveStatus !== 200) {
+        return sendError(response, state.controls.turnsRetrieveStatus, "Fixture Turns retrieve failed.");
+      }
+      const sessionId = decodeURIComponent(turnsMatch[1]);
+      if (!state.sessions.some((candidate) => candidate.id === sessionId)) {
+        return sendError(response, 404, "Fixture Session not found for Turns.");
+      }
+      const sessionTurns = state.turns.filter((turn) => turn.session_id === sessionId);
+      const after = url.searchParams.get("after");
+      const start = after ? sessionTurns.findIndex((turn) => turn.id === after) + 1 : 0;
+      if (after && start === 0) return sendError(response, 400, "Fixture Turn cursor not found.");
+      const requestedLimit = Number(url.searchParams.get("limit") ?? 20);
+      const size = Math.max(1, Math.min(requestedLimit, state.controls.turnsPageSize));
+      const data = sessionTurns.slice(start, start + size);
+      return sendJson(response, {
+        object: "list",
+        data,
+        has_more: start + data.length < sessionTurns.length,
+      });
+    }
 
     const eventsMatch = url.pathname.match(/^\/v1\/agents\/sessions\/([^/]+)\/events$/);
     if (request.method === "POST" && eventsMatch) {
@@ -345,6 +457,7 @@ const server = http.createServer(async (request, response) => {
         "cache-control": "no-cache, no-transform",
         connection: "keep-alive",
       });
+      streamResponses.add(response);
       response.write(": fixture stream open\n\n");
       const statuses = [null, "pending", "ready", "connected", "disconnected", "failed", "expired"];
       const environmentStatus = statuses[state.controls.environmentEventStatus] ?? null;
@@ -378,7 +491,10 @@ const server = http.createServer(async (request, response) => {
         setTimeout(() => response.end(), state.controls.streamCloseDelayMs);
       }
       const heartbeat = setInterval(() => response.write(": fixture heartbeat\n\n"), 10_000);
-      request.on("close", () => clearInterval(heartbeat));
+      request.on("close", () => {
+        clearInterval(heartbeat);
+        streamResponses.delete(response);
+      });
       return;
     }