feat(studio): threads

apps/agent-forge/e2e/a11y.spec.ts

@@ -113,7 +113,7 @@ test('Ctrl+K opens the command palette; ? lists the shortcuts', async ({ page })
 });
 
 // One <main>, one h1, the banner and the primary nav on every screen, not just Agents.
-for (const route of ['#/traces', '#/traces?status=error', '#/traces/e2e-trace?obs=e2e-obs', '#/monitoring', '#/evals/scores', '#/settings', '#/playground']) {
+for (const route of ['#/traces', '#/traces?status=error', '#/traces/e2e-trace?obs=e2e-obs', '#/threads', '#/threads/s1', '#/threads?status=active', '#/monitoring', '#/evals/scores', '#/settings', '#/playground']) {
   test(`${route} keeps the landmarks and exactly one main`, async ({ page }) => {
     await page.goto(`/?token=e2e-studio-token${route}`);
     await expect(page.locator('.screen-main')).toBeVisible();

apps/agent-forge/src/observe/__tests__/observeClient.test.ts

@@ -27,6 +27,19 @@ describe('ObserveClient', () => {
     );
   });
 
+  it('lists sessions with their filter params and reads one session, undefined on 404', async () => {
+    const fetchFn = vi.fn(async (url: string) =>
+      url.includes('/sessions/')
+        ? jsonResponse(404, { error: 'No session', code: 'LOUSHO_REQUEST_INVALID' })
+        : jsonResponse(200, { sessions: [], total: 0, limit: 50, offset: 0 })
+    );
+    const client = clientWith(fetchFn as unknown as typeof fetch);
+    await client.listSessions({ agent_id: 'a1', status: 'active', from: 1, to: 2, limit: 50, offset: 0 });
+    expect(fetchFn).toHaveBeenCalledWith('/observe/api/v1/sessions?agent_id=a1&status=active&from=1&to=2&limit=50&offset=0', expect.anything());
+    expect(await client.session('s/1')).toBeUndefined();
+    expect(fetchFn).toHaveBeenLastCalledWith('/observe/api/v1/sessions/s%2F1', expect.anything());
+  });
+
   it('maps a proxy 502 and a network failure to ObserveDaemonDownError', async () => {
     const via502 = clientWith((async () => jsonResponse(502, { error: 'unreachable' })) as typeof fetch);
     await expect(via502.listTraces()).rejects.toBeInstanceOf(ObserveDaemonDownError);

apps/agent-forge/src/observe/observeClient.ts

@@ -16,6 +16,8 @@ import type {
   ObserveInsights,
   ObserveScorePassRate,
   ObserveScoreRow,
+  ObserveSessionDetail,
+  ObserveSessionList,
   ObserveTraceDetail,
   ObserveTraceList,
 } from './types';
@@ -85,6 +87,16 @@ export interface TraceListFilter extends TraceListQuery {
 /** The trace list's server-side sort keys. */
 export type TraceSort = 'start' | 'duration' | 'tokens' | 'cost';
 
+/** `GET /api/v1/sessions` params: sessions whose `started_at` is in the half-open `[from, to)` window (ms epoch). */
+export interface SessionListFilter {
+  agent_id?: string;
+  status?: string;
+  from?: number;
+  to?: number;
+  limit?: number;
+  offset?: number;
+}
+
 export interface ScoreListFilter {
   trace_id?: string;
   /** Exact score name (`<evalName>` is the eval-level row of each case). */
@@ -207,6 +219,21 @@ export class ObserveClient {
     }
   }
 
+  /** `GET /api/v1/sessions` - the paginated thread list, newest first. */
+  listSessions(filter: SessionListFilter = {}): Promise<ObserveSessionList> {
+    return this.request('/api/v1/sessions', { query: { ...filter } });
+  }
+
+  /** `GET /api/v1/sessions/:id` - one thread plus its traces, oldest first; `undefined` on the daemon's 404. */
+  async session(sessionId: string): Promise<ObserveSessionDetail | undefined> {
+    try {
+      return await this.request(`/api/v1/sessions/${encodeURIComponent(sessionId)}`);
+    } catch (error) {
+      if (error instanceof ObserveApiError && error.status === 404) return undefined;
+      throw error;
+    }
+  }
+
   /** `GET /api/v1/agents` - the fleet registry with read-time freshness. */
   async listAgents(project?: string): Promise<ObserveAgentRow[]> {
     const body = await this.request<{ agents: ObserveAgentRow[] }>('/api/v1/agents', { query: { project } });

apps/agent-forge/src/observe/types.ts

@@ -48,6 +48,39 @@ export interface ObserveTraceList {
   offset: number;
 }
 
+/** One row of `GET /api/v1/sessions`: a conversation thread with its run totals (`traces` = turns). */
+export interface ObserveSessionRow {
+  session_id: string;
+  agent_id: string | null;
+  /** The agent's registered name, or null. */
+  agent_name: string | null;
+  /** `active` until the agent's crash sweep marks it `interrupted`. */
+  status: string;
+  started_at: number;
+  /** The latest observation end seen in the session, or null while nothing has finished. */
+  ended_at: number | null;
+  /** Distinct traces (turns) in the session. */
+  traces: number;
+  input_tokens: number;
+  output_tokens: number;
+  cost_usd: number;
+  /** The latest turn's root input, first 200 chars. */
+  last_input_preview: string | null;
+}
+
+export interface ObserveSessionList {
+  sessions: ObserveSessionRow[];
+  total: number;
+  limit: number;
+  offset: number;
+}
+
+/** `GET /api/v1/sessions/:id`: the thread and its traces, oldest first (capped at 500). */
+export interface ObserveSessionDetail {
+  session: ObserveSessionRow;
+  traces: ObserveTraceRow[];
+}
+
 /** One flat observation row of `GET /api/v1/traces/:id` - the client nests them by `parent_obs_id`. */
 export interface ObserveObservationRow {
   obs_id: string;

apps/agent-forge/src/observe/useObserve.ts

@@ -6,13 +6,15 @@
  * duplicating effect logic.
  */
 import { useCallback, useEffect, useRef, useState } from 'react';
-import { observeClient, ObserveDaemonDownError, type InsightsFilter, type ScoreListFilter, type TraceListFilter } from './observeClient';
+import { observeClient, ObserveDaemonDownError, type InsightsFilter, type ScoreListFilter, type SessionListFilter, type TraceListFilter } from './observeClient';
 import type {
   ObserveAgentRow,
   ObserveHealth,
   ObserveInsights,
   ObserveScorePassRate,
   ObserveScoreRow,
+  ObserveSessionDetail,
+  ObserveSessionList,
   ObserveTraceDetail,
   ObserveTraceList,
 } from './types';
@@ -97,6 +99,14 @@ export function useTraceDetail(traceId: string | undefined, refreshKey?: unknown
   );
 }
 
+export function useSessions(filter: SessionListFilter, refreshKey?: unknown): ObserveQuery<ObserveSessionList> {
+  return useObserveQuery(() => observeClient.listSessions(filter), [JSON.stringify(filter), refreshKey]);
+}
+
+export function useSession(sessionId: string, refreshKey?: unknown): ObserveQuery<ObserveSessionDetail | undefined> {
+  return useObserveQuery(() => observeClient.session(sessionId), [sessionId, refreshKey]);
+}
+
 export function useAgents(project?: string, refreshKey?: unknown): ObserveQuery<ObserveAgentRow[]> {
   return useObserveQuery(() => observeClient.listAgents(project), [project, refreshKey]);
 }

apps/agent-forge/src/screens/listFilters.css

@@ -0,0 +1,9 @@
+/* Shared list-screen cells (#/traces, #/threads). Tokens only. */
+/* Preview cells: one line, ellipsis; the full text is in the title attribute. */
+.list-clip {
+  display: block;
+  overflow: hidden;
+  white-space: nowrap;
+  text-overflow: ellipsis;
+  min-width: 0;
+}

apps/agent-forge/src/screens/listFilters.tsx

@@ -0,0 +1,79 @@
+/**
+ * Filter-bar pieces shared by the list screens (`#/traces`, `#/threads`):
+ * single-value facets over FilterBar's multi-select, the agent facet, the
+ * URL-backed time range, and the one-line preview cell.
+ */
+import './listFilters.css';
+import { useRef } from 'react';
+import type { ObserveAgentRow } from '../observe/types';
+import type { FacetDef } from '../ui/FilterBar';
+import { formatTimeRange, toWindow, type TimeRange } from '../ui/TimeRangePicker';
+
+type Window = { from: number; to: number };
+type Preset = Extract<TimeRange, { preset: string }>['preset'];
+
+/** A preview cell: one line, ellipsised by CSS, the full text in `title`. */
+export function Clip({ text, className = 'list-clip' }: { text: string | null; className?: string }) {
+  if (!text) return null;
+  return (
+    <span className={className} title={text}>
+      {text}
+    </span>
+  );
+}
+
+/** Single-value facet over FilterBar's multi-select: the newest pick wins, clearing it drops the filter. */
+export function facet(
+  key: string,
+  label: string,
+  options: string[],
+  value: string | undefined,
+  set: (v: string | undefined) => void
+): FacetDef {
+  const all = value && !options.includes(value) ? [value, ...options] : options;
+  return {
+    key,
+    label,
+    options: all.map((o) => ({ value: o, label: o })),
+    value: value ? [value] : [],
+    onChange: (next) => set(next.find((v) => v !== value) ?? (value && next.includes(value) ? value : undefined)),
+  };
+}
+
+/** The agent facet: values are agent ids, labels the registered names. */
+export function agentFacet(agents: ObserveAgentRow[], value: string | undefined, set: (v: string | undefined) => void): FacetDef {
+  const def = facet('agent', 'Agent', agents.map((a) => a.agent_id), value, set);
+  const names = new Map(agents.map((a) => [a.agent_id, a.name ?? a.agent_id.slice(0, 8)]));
+  return { ...def, options: def.options.map((o) => ({ ...o, label: names.get(o.value) ?? o.value })) };
+}
+
+/**
+ * The time-range control for a list whose URL holds only absolute bounds.
+ * `defaultPreset` is the rolling window an empty URL means; picking it clears
+ * the bounds. Any other preset is snapshotted to bounds and remembered next to
+ * them, so the picker keeps showing e.g. "7d" until the window changes (a
+ * reload then shows it as Custom).
+ */
+export function useRangeFilter(
+  window: Window | undefined,
+  defaultPreset: Preset,
+  setWindow: (w: Window | undefined) => void
+): { value: TimeRange; onChange: (r: TimeRange) => void } {
+  const picked = useRef<{ range: TimeRange; key: string } | undefined>(undefined);
+  let value: TimeRange = { preset: defaultPreset };
+  if (window) {
+    const key = formatTimeRange(window);
+    value = picked.current?.key === key ? picked.current.range : window;
+  }
+  const onChange = (r: TimeRange) => {
+    if ('preset' in r && r.preset === defaultPreset) {
+      picked.current = undefined;
+      setWindow(undefined);
+      return;
+    }
+    const next = toWindow(r, Date.now());
+    picked.current = { range: r, key: formatTimeRange(next) };
+    setWindow(next);
+  };
+  return { value, onChange };
+}

apps/agent-forge/src/screens/routes.tsx

@@ -4,6 +4,7 @@ import { useRoute } from '../router/useRoute';
 import { Topbar } from '../components/Topbar';
 import { SettingsPanel } from '../components/SettingsPanel';
 import { TracesScreen } from './traces/TracesScreen';
+import { ThreadsScreen } from './threads/ThreadsScreen';
 import { AgentsScreen } from './agents/AgentsScreen';
 import { TraceScreen } from './traces/TraceScreen';
 import { MonitoringScreen } from './monitoring/MonitoringScreen';
@@ -66,7 +67,6 @@ export function useScreen(): ScreenId {
 }
 
 const PLACEHOLDERS: Partial<Record<ScreenId, { title: string; blurb: string }>> = {
-  threads: { title: 'Threads', blurb: 'Multi-turn conversations grouped by session, with every run in each thread.' },
   evals: { title: 'Evaluations', blurb: 'Datasets, evaluators and annotation queues for scoring runs.' },
   prompts: { title: 'Prompts', blurb: 'Versioned prompts you can compare, pin and pull into your agents.' },
   hub: { title: 'Context Hub', blurb: 'The shared context namespaces your agents read from, browsable by namespace.' },
@@ -102,6 +102,7 @@ export function Screens() {
   else if (screen === 'monitoring') content = <MonitoringScreen />;
   else if (screen === 'scores') content = <ScoresScreen />;
   else if (screen === 'playground') content = <PlaygroundRoute />;
+  else if (screen === 'threads') content = <ThreadsScreen />;
   else if (placeholder) content = <Placeholder {...placeholder} />;
   else
     content = (

apps/agent-forge/src/screens/threads/ThreadScreen.tsx

@@ -0,0 +1,92 @@
+/**
+ * `#/threads/:sessionId` - one conversation: every turn (one trace, oldest
+ * first) as its input and output messages, with a header line and a link to
+ * the turn's trace. Reads `GET /api/v1/sessions/:id` (traces capped at 500).
+ */
+import { useState } from 'react';
+import { useSession } from '../../observe/useObserve';
+import { ObserveError } from '../../observe/chrome';
+import type { ObserveTraceRow } from '../../observe/types';
+import { fmtCost, fmtDuration, fmtInt, fmtStarted } from '../../observe/format';
+import { EmptyState } from '../../ui/EmptyState';
+import { MessageView } from '../../ui/MessageView';
+import { StatusPill } from '../../ui/StatusPill';
+import { threadsListPath } from './threadsUrl';
+
+const BackLink = () => <a href={`#${threadsListPath()}`}>Back to threads</a>;
+
+/** One turn: time, status, duration and tokens, then the input and the output (or the run's error). */
+function Turn({ trace, index, now }: { trace: ObserveTraceRow; index: number; now: number }) {
+  return (
+    <li className="thread-turn" aria-label={`Turn ${index + 1}`}>
+      <div className="thread-turn-head">
+        <span className="thread-turn-index">#{index + 1}</span>
+        <time dateTime={new Date(trace.start).toISOString()}>{fmtStarted(trace.start, now)}</time>
+        <StatusPill status={trace.status} />
+        <span>{fmtDuration(trace.end - trace.start)}</span>
+        <span title={`${fmtInt(trace.input_tokens)} in / ${fmtInt(trace.output_tokens)} out`}>
+          {fmtInt(trace.input_tokens + trace.output_tokens)} tokens
+        </span>
+        <a className="thread-turn-link" href={`#/traces/${encodeURIComponent(trace.trace_id)}`}>
+          Open trace
+        </a>
+      </div>
+      <MessageView label="Input" value={trace.input_preview} />
+      {trace.error_message ? (
+        <p className="thread-error" role="note">
+          <span className="thread-error-label">Error</span> {trace.error_message}
+        </p>
+      ) : (
+        <MessageView label="Output" value={trace.output_preview} />
+      )}
+    </li>
+  );
+}
+
+/** `#/threads/:sessionId`; an unknown session shows an empty state with the way back. */
+export function ThreadScreen({ sessionId }: { sessionId: string }) {
+  const thread = useSession(sessionId);
+  const [now] = useState(() => Date.now());
+
+  const session = thread.data?.session;
+  const traces = thread.data?.traces ?? [];
+  let body;
+  if (thread.loading && !thread.data) body = <p className="thread-loading">Loading thread...</p>;
+  else if (thread.error) body = <ObserveError error={thread.error} />;
+  else if (!session) body = <EmptyState title="Thread not found" hint={`No session '${sessionId}'.`} action={<BackLink />} />;
+  else
+    body = (
+      <>
+        <p className="thread-summary">
+          <StatusPill status={session.status} />
+          <span>{session.agent_name ?? session.agent_id ?? 'unknown agent'}</span>
+          <span>
+            {fmtInt(session.traces)} {session.traces === 1 ? 'turn' : 'turns'}
+          </span>
+          <span>started {fmtStarted(session.started_at, now)}</span>
+          <span>{fmtInt(session.input_tokens + session.output_tokens)} tokens</span>
+          <span>{fmtCost(session.cost_usd)}</span>
+        </p>
+        {traces.length < session.traces && (
+          <p className="thread-note">Showing the first {fmtInt(traces.length)} turns of {fmtInt(session.traces)}.</p>
+        )}
+        <ol className="thread-turns">
+          {traces.map((t, i) => (
+            <Turn key={t.trace_id} trace={t} index={i} now={now} />
+          ))}
+        </ol>
+      </>
+    );
+
+  return (
+    <section className="thread-screen" aria-label="Thread">
+      <nav className="thread-nav" aria-label="Breadcrumb">
+        <BackLink />
+      </nav>
+      <h2 className="thread-title">
+        Thread <code title={sessionId}>{sessionId}</code>
+      </h2>
+      {body}
+    </section>
+  );
+}