Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/workflows/agentex-ui-lint-typecheck.yml
Original file line number Diff line number Diff line change
Expand Up @@ -40,3 +40,7 @@ jobs:
- name: Run lint
run: npm run lint
working-directory: ./agentex-ui

- name: Run unit tests
run: npm run test:run
working-directory: ./agentex-ui
5 changes: 2 additions & 3 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,7 @@ The backend (`agentex/src/`) follows a clean architecture with strict layer sepa
```
src/
├── api/ # FastAPI routes, middleware, request/response schemas
│ ├── routes/ # API endpoints (agents, tasks, messages, spans, etc.)
│ ├── routes/ # API endpoints (agents, tasks, messages, states, etc.)
│ ├── schemas/ # Pydantic request/response models
│ ├── authentication_middleware.py
│ └── app.py # FastAPI application setup
Expand Down Expand Up @@ -285,7 +285,6 @@ Tests are organized by type and use different strategies:
- **Agents**: Autonomous entities that execute tasks, managed via ACP protocol
- **Tasks**: Work units with lifecycle states (pending → running → completed/failed). Identified by a UUID `id`; the human-readable `name` is **optional** (nullable) and, when set, globally unique. `task/create` is get-or-create keyed on `name`, so reusing an existing name returns that task with its prior history instead of creating a new one — omit `name` (or make it unique) whenever each call should produce a fresh task.
- **Messages**: Communication between system and agents (stored in MongoDB)
- **Spans**: Execution traces for observability (OpenTelemetry-style)
- **Events**: Domain events for async communication
- **States**: Key-value state storage for agents
- **Deployment History**: Track agent deployment versions and changes
Expand Down Expand Up @@ -347,7 +346,7 @@ For any migration that adds a backfilled column with an FK and an index on a lar
| Step | What | Why |
|---|---|---|
| **M1 (Alembic)** | `ADD COLUMN` (nullable) + `ADD CONSTRAINT ... NOT VALID` + `CREATE INDEX CONCURRENTLY` (in `autocommit_block()`) | Schema-only, all metadata-cheap or non-blocking. Each operation is idempotent (`IF NOT EXISTS` / `pg_constraint` guard) so the migration is safe to re-run on environments that already ran a previous (broken) version. |
| **Out-of-band runbook** | Chunked backfill script with `lock_timeout`, small batches, `COMMIT` between batches, `pg_sleep` between batches | Operator-driven; runs during a low-traffic window, can be cancelled cleanly, doesn't block pod startup. Pattern: `agentex/docs/runbooks/spans-task-id-backfill.md`. |
| **Out-of-band runbook** | Chunked backfill script with `lock_timeout`, small batches, `COMMIT` between batches, `pg_sleep` between batches | Operator-driven; runs during a low-traffic window, can be cancelled cleanly, doesn't block pod startup. |
| **M2 (Alembic)** | `ALTER TABLE ... VALIDATE CONSTRAINT` (only if a fully validated FK state is actually needed) | Runs after the backfill so the scan finds no violations. `ShareUpdateExclusiveLock` is non-blocking against reads/writes but still scans the table — usually optional. |

The application should also tolerate the partially-backfilled state at read time (e.g. ORing the new column against the legacy column where they overlap) so deployment of M1 is decoupled from the backfill's completion.
Expand Down
4 changes: 2 additions & 2 deletions agentex-ui/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ A modern web interface for building, testing, and monitoring intelligent agents.

### Observability

- **Execution Traces** - View OpenTelemetry-style spans for task execution
- **Execution Traces** - View a task's spans from Scale GenAI Platform (needs `SGP_API_URL` or `NEXT_PUBLIC_SGP_APP_URL`)
- **Span Visualization** - Hierarchical view of execution flow
- **Performance Metrics** - Timing and duration information for each execution step
- **Error Tracking** - Detailed error information when tasks fail
Expand Down Expand Up @@ -178,7 +178,7 @@ For Docker-related commands, see the Docker section in `build.ps1 help`.
- `hooks/use-tasks.ts` - Task list with infinite scroll pagination
- `hooks/use-task-messages.ts` - Message fetching and sending with message streaming for sync agents
- `hooks/use-task-subscription.ts` - Real-time task updates via WebSocket for async agents
- `hooks/use-spans.ts` - Execution trace data
- `hooks/use-spans.ts` - Execution trace data (via `/api/traces`)

**Components:**

Expand Down
161 changes: 161 additions & 0 deletions agentex-ui/app/api/traces/[traceId]/spans/route.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,161 @@
import { afterEach, describe, expect, it, vi } from 'vitest';

import { GET } from './route';

const bff = vi.hoisted(() => ({
baseURL: 'https://sgp.example/api' as string | undefined,
applyBffCredentials: vi.fn(async (_req: Request, headers: Headers) => {
headers.set('authorization', 'Bearer server-side');
}),
}));

vi.mock('@/app/api/_lib/bff', () => ({
get SGP_BASE_URL() {
return bff.baseURL;
},
applyBffCredentials: bff.applyBffCredentials,
}));

function call(traceId: string, init?: RequestInit, search = '') {
return GET(
new Request(`http://ui.local/api/traces/${traceId}/spans${search}`, init),
{ params: Promise.resolve({ traceId }) }
);
}

function upstreamURL(fetchMock: ReturnType<typeof vi.fn>) {
return new URL(fetchMock.mock.calls[0]![0] as string);
}

describe('GET /api/traces/[traceId]/spans', () => {
afterEach(() => {
vi.unstubAllGlobals();
bff.baseURL = 'https://sgp.example/api';
});

it('searches the platform for the trace with server-attached credentials', async () => {
const page = { items: [{ id: 's1', trace_id: 't1' }], has_more: false };
const fetchMock = vi.fn().mockResolvedValue(
new Response(JSON.stringify(page), {
status: 200,
headers: { 'content-type': 'application/json' },
})
);
vi.stubGlobal('fetch', fetchMock);

const res = await call('t1');

expect(res.status).toBe(200);
expect(await res.json()).toEqual(page);
expect(bff.applyBffCredentials).toHaveBeenCalledTimes(1);
const [url, init] = fetchMock.mock.calls[0]!;
expect(url).toBe(
'https://sgp.example/api/v5/spans/search?limit=100&sort_by=start_timestamp&sort_order=asc&allow_short_pages=true'
);
expect(init.method).toBe('POST');
expect(JSON.parse(init.body)).toEqual({ trace_ids: ['t1'] });
expect(new Headers(init.headers).get('authorization')).toBe(
'Bearer server-side'
);
expect(init.signal).toBeInstanceOf(AbortSignal);
});

it('answers 499 when the browser aborts before the platform replies', async () => {
const controller = new AbortController();
vi.stubGlobal(
'fetch',
vi.fn((_url: string, init: RequestInit) => {
const signal = init.signal as AbortSignal;
return new Promise<Response>((_resolve, reject) => {
if (signal.aborted) reject(signal.reason);
signal.addEventListener('abort', () => reject(signal.reason));
});
})
);

const pending = call('t1', { signal: controller.signal });
controller.abort();
const res = await pending;

expect(res.status).toBe(499);
});

it('anchors the search window on the task creation time', async () => {
const fetchMock = vi
.fn()
.mockResolvedValue(new Response('{"items":[]}', { status: 200 }));
vi.stubGlobal('fetch', fetchMock);
const from = '2026-01-01T00:00:00.000Z';

await call('t1', undefined, `?from=${encodeURIComponent(from)}`);

const params = upstreamURL(fetchMock).searchParams;
const fromTs = Date.parse(params.get('from_ts')!);
const toTs = Date.parse(params.get('to_ts')!);
expect(fromTs).toBe(Date.parse(from) - 5 * 60 * 1000);
expect(toTs - fromTs).toBe(90 * 24 * 60 * 60 * 1000 - 60 * 1000);
});

it('caps the window at now for a recent task', async () => {
const fetchMock = vi
.fn()
.mockResolvedValue(new Response('{"items":[]}', { status: 200 }));
vi.stubGlobal('fetch', fetchMock);
const from = new Date(Date.now() - 60 * 60 * 1000).toISOString();

await call('t1', undefined, `?from=${encodeURIComponent(from)}`);

const toTs = Date.parse(upstreamURL(fetchMock).searchParams.get('to_ts')!);
expect(Date.now() - toTs).toBeLessThan(5000);
});

it('sends no window without a creation time', async () => {
const fetchMock = vi
.fn()
.mockResolvedValue(new Response('{"items":[]}', { status: 200 }));
vi.stubGlobal('fetch', fetchMock);

await call('t1');

const params = upstreamURL(fetchMock).searchParams;
expect(params.has('from_ts')).toBe(false);
expect(params.has('to_ts')).toBe(false);
});

it('rejects a creation time that is not a timestamp', async () => {
const fetchMock = vi.fn();
vi.stubGlobal('fetch', fetchMock);

const res = await call('t1', undefined, '?from=yesterday');

expect(res.status).toBe(400);
expect(fetchMock).not.toHaveBeenCalled();
});

it('passes the upstream status through', async () => {
vi.stubGlobal(
'fetch',
vi
.fn()
.mockResolvedValue(
new Response(JSON.stringify({ detail: 'forbidden' }), { status: 403 })
)
);

const res = await call('t1');

expect(res.status).toBe(403);
expect(await res.json()).toEqual({ detail: 'forbidden' });
});

it('returns 503 when the platform API is not configured', async () => {
bff.baseURL = undefined;
const fetchMock = vi.fn();
vi.stubGlobal('fetch', fetchMock);

const res = await call('t1');

expect(res.status).toBe(503);
expect(fetchMock).not.toHaveBeenCalled();
});
});
84 changes: 84 additions & 0 deletions agentex-ui/app/api/traces/[traceId]/spans/route.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
import { NextResponse } from 'next/server';

import { applyBffCredentials, SGP_BASE_URL } from '@/app/api/_lib/bff';

/**
* Scoped BFF proxy for one trace's spans, read from the platform's span search with the
* credentials attached server-side. Only this path is exposed, not a catch-all, so the
* browser can't reach arbitrary platform endpoints with those credentials.
*/
export const dynamic = 'force-dynamic';

// The sidebar shows one page and reports the rest through has_more.
const PAGE_SIZE = 100;
// The platform refuses a window wider than 90 days, and it defaults an omitted window to the
// last 90 days, which hides older tasks. Anchor the window on the task's creation instead.
const WINDOW_MS = 90 * 24 * 60 * 60 * 1000 - 60 * 1000;
const SKEW_MS = 5 * 60 * 1000;

function searchWindow(from: string | null): Record<string, string> | null {
if (from === null) return {};
const start = Date.parse(from);
if (Number.isNaN(start)) return null;
const fromTs = start - SKEW_MS;
const toTs = Math.min(Date.now(), fromTs + WINDOW_MS);
return {
from_ts: new Date(fromTs).toISOString(),
to_ts: new Date(toTs).toISOString(),
};
}

export async function GET(
request: Request,
ctx: { params: Promise<{ traceId: string }> }
): Promise<Response> {
if (!SGP_BASE_URL) {
return NextResponse.json(
{ error: 'SGP traces are not configured. Set SGP_API_URL.' },
{ status: 503 }
);
}

const { traceId } = await ctx.params;
const window = searchWindow(new URL(request.url).searchParams.get('from'));
if (window === null) {
return NextResponse.json(
{ error: 'from must be an ISO timestamp' },
{ status: 400 }
);
}
const headers = new Headers({
'Content-Type': 'application/json',
accept: 'application/json',
});
await applyBffCredentials(request, headers);

const query = new URLSearchParams({
Comment thread
mohammadatallah-scale marked this conversation as resolved.
limit: String(PAGE_SIZE),
sort_by: 'start_timestamp',
sort_order: 'asc',
// Over the byte budget the platform shortens the page instead of refusing it.
allow_short_pages: 'true',
...window,
});
let upstream: Response;
try {
upstream = await fetch(`${SGP_BASE_URL}/v5/spans/search?${query}`, {
method: 'POST',
headers,
body: JSON.stringify({ trace_ids: [traceId] }),
signal: request.signal,
});
} catch (error) {
// A browser that navigated away aborts the request, and the abort reason is what the
// fetch rejects with, so the same identity check the Agentex proxy uses applies here.
if (error === request.signal.reason) {
return new Response(null, { status: 499 });
}
throw error;
}
return new Response(upstream.body, {
status: upstream.status,
headers: { 'content-type': 'application/json' },
});
}
2 changes: 1 addition & 1 deletion agentex-ui/components/providers/agentex-provider.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import {
// Hitting /api/auth/session runs the jwt-callback refresh and rotates the cookie. Deduped
// so a burst of 401s (e.g. a refocused tab) shares one refresh.
let sessionRefresh: Promise<unknown> | null = null;
function refreshSession(): Promise<unknown> {
export function refreshSession(): Promise<unknown> {
sessionRefresh ??= fetch('/api/auth/session', { credentials: 'include' })
.catch(() => {})
.finally(() => {
Expand Down
6 changes: 5 additions & 1 deletion agentex-ui/components/providers/index.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
export { AgentexProvider, useAgentexClient } from './agentex-provider';
export {
AgentexProvider,
refreshSession,
useAgentexClient,
} from './agentex-provider';
export { TaskProvider } from './task-provider';
export { ThemeProvider } from './theme-provider';
export { QueryProvider } from './query-provider';
5 changes: 2 additions & 3 deletions agentex-ui/components/task-header/task-header.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import {
SelectValue,
} from '@/components/ui/select';
import { useSafeSearchParams } from '@/hooks/use-safe-search-params';
import { useSpans } from '@/hooks/use-spans';

import type { Agent } from 'agentex/resources';

Expand All @@ -37,8 +36,8 @@ export function TaskHeader({
}: TaskHeaderProps) {
const displayTaskId = taskId ? taskId.split('-')[0] : '';
const { agentName: selectedAgentName } = useSafeSearchParams();
const { spans } = useSpans(taskId);
const traceId = spans[0]?.trace_id ?? taskId;
// Agents trace under the task id, so the task is the trace.
const traceId = taskId;

const copyTaskId = async () => {
if (taskId) {
Expand Down
22 changes: 20 additions & 2 deletions agentex-ui/components/traces-sidebar/traces-sidebar.tsx
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { AnimatePresence, motion } from 'framer-motion';

import { useAgentexClient } from '@/components/providers';
import { JsonViewer } from '@/components/ui/json-viewer';
import { ResizableSidebar } from '@/components/ui/resizable-sidebar';
import {
Expand All @@ -10,6 +11,7 @@ import {
} from '@/components/ui/tooltip';
import { useSafeSearchParams } from '@/hooks/use-safe-search-params';
import { useSpans } from '@/hooks/use-spans';
import { useTask } from '@/hooks/use-tasks';

const MIN_SIDEBAR_WIDTH = 350;
const DEFAULT_SIDEBAR_WIDTH = 350;
Expand All @@ -20,7 +22,15 @@ type TracesSidebarProps = {

export function TracesSidebar({ isOpen }: TracesSidebarProps) {
const { taskID } = useSafeSearchParams();
const { spans, isLoading, error } = useSpans(taskID);
const { agentexClient, sgpAppURL } = useAgentexClient();
const { data: task, isError: taskUnavailable } = useTask({
agentexClient,
taskId: taskID ?? '',
});
// The task's creation time anchors the search window. Without it the query waits, unless
// the task itself cannot be read, in which case the platform's default window is used.
const createdAt = task?.created_at ?? (taskUnavailable ? null : undefined);
Comment thread
mohammadatallah-scale marked this conversation as resolved.
const { spans, hasMore, isLoading, error } = useSpans(taskID, createdAt);

return (
<AnimatePresence>
Expand Down Expand Up @@ -76,8 +86,16 @@ export function TracesSidebar({ isOpen }: TracesSidebarProps) {
</div>
)}

{hasMore && (
<div className="text-muted-foreground text-sm">
Showing the first {spans.length} spans.
{sgpAppURL &&
' Use Investigate traces for the full trace.'}
</div>
)}

{spans.map(span => {
const startTime = new Date(span.start_time);
const startTime = new Date(span.start_timestamp);

return (
<div key={span.id}>
Expand Down
2 changes: 1 addition & 1 deletion agentex-ui/example.env.development
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
AGENTEX_API_URL=http://localhost:5003 # /api/agentex → agentex API
# ENABLE_AGENT_RUN_SCHEDULES=true # enables scheduled tasks in both API and UI
# SGP_API_URL=<your_sgp_api_url> # optional: /api/feedback & /api/user-info → SGP API
# SGP_API_URL=<your_sgp_api_url> # optional: /api/feedback, /api/user-info & /api/traces → SGP API
# NEXT_PUBLIC_SGP_APP_URL=<your_sgp_app_url> # optional: links to SGP traces

#---- OIDC login (opt-in) — set AGENTEX_UI_AUTH_PROVIDER_ID to enable login. Its value must
Expand Down
Loading
Loading