Skip to content
Merged
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
7 changes: 7 additions & 0 deletions apps/agent-forge/e2e/studioUx.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ test('a new agent is named what you typed; a finished run says done and labels i
await page.getByRole('button', { name: 'Run', exact: true }).click();
await expect(page.locator('.topbar [role="status"]')).toContainText('done', { timeout: 15_000 });
await expect(page.locator('.topbar [role="status"]')).not.toContainText('stopped');
// The finished run links to its trace (builder runs report to the observe daemon).
await expect(page.getByRole('link', { name: 'Open trace' })).toHaveAttribute('href', /^#\/traces\/[^/]+$/);

const runBubble = page.locator('.chat-msg.run-input');
await expect(runBubble).toBeVisible();
Expand Down Expand Up @@ -88,6 +90,11 @@ test('the Output node is drawn once and the minimap shows the graph', async ({ p
await page.goto(`/?token=${TOKEN}`);
await createAgent(page, id);

// Node cards: the LLM node shows its provider/model chip and an idle status before any run.
const llm = page.locator('[data-node-type="llm"]');
await expect(llm.locator('.node-model')).toHaveText('mock/mock-1');
await expect(llm.locator('.status-pill')).toHaveText('idle');

// React Flow's built-in `output` type used to add its own border/padding box around ours.
await expect(page.locator('.react-flow__node-output')).toHaveCount(0);
const output = page.locator('.react-flow__node:has([data-node-type="output"])');
Expand Down
17 changes: 16 additions & 1 deletion apps/agent-forge/playwright.config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,14 @@ const baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-forge-e2e-'));
const port = Number(process.env.LOUSHO_STUDIO_E2E_PORT ?? 4799);
/** The API needs the per-launch token; pin one so the spec can open `/?token=...`. */
const E2E_STUDIO_TOKEN = 'e2e-studio-token';
/**
* Builder and playground runs report to the observe daemon: point them at a
* closed loopback port so e2e runs never reach a real daemon on this machine
* (the reporter only drops-and-counts), and ignore an inherited LOUSHO_OBSERVE.
*/
const inheritedEnv = { ...process.env };
delete inheritedEnv.LOUSHO_OBSERVE;
const E2E_OBSERVE_URL = process.env.LOUSHO_STUDIO_E2E_OBSERVE_URL ?? 'http://127.0.0.1:1';

export default defineConfig({
testDir: './e2e',
Expand All @@ -39,7 +47,14 @@ export default defineConfig({
webServer: {
command: 'node dist-server/index.cjs',
cwd: moduleDir,
env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', BASE_DIR: baseDir, LOUSHO_STUDIO_TOKEN: E2E_STUDIO_TOKEN },
env: {
...inheritedEnv,
PORT: String(port),
HOST: '127.0.0.1',
BASE_DIR: baseDir,
LOUSHO_STUDIO_TOKEN: E2E_STUDIO_TOKEN,
LOUSHO_OBSERVE_URL: E2E_OBSERVE_URL,
},
url: `http://127.0.0.1:${port}/health`,
reuseExistingServer: false,
timeout: 20_000,
Expand Down
2 changes: 1 addition & 1 deletion apps/agent-forge/server/__tests__/auth.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ describe('studio access control', () => {
baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'studio-auth-test-'));
staticDir = fs.mkdtempSync(path.join(os.tmpdir(), 'studio-auth-static-'));
fs.writeFileSync(path.join(staticDir, 'index.html'), '<!doctype html><title>forge</title>');
handle = await startStudioServer({ baseDir, port: 0, staticDir, token: TOKEN });
handle = await startStudioServer({ baseDir, port: 0, staticDir, token: TOKEN, observe: false });
origin = `http://127.0.0.1:${handle.port}`;
});

Expand Down
140 changes: 140 additions & 0 deletions apps/agent-forge/server/__tests__/builderReporting.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
/**
* Builder runs (`RunManager.run()`) report to the observe daemon the Studio
* reads, as agent `<agentId>`, through the same reporter factory the
* playground uses (observeReporting.ts) - so the top bar's "Open trace" link
* (`#/traces/<traceId>`) resolves. The run status carries that trace id.
*/
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
import * as fs from 'node:fs';
import * as path from 'node:path';
import * as os from 'node:os';
import type { AgentSpec } from '@lousho/build-ai-agent';
import { RunManager } from '../runRegistry';
import { agentTraceDir } from '../traceStore';
import { FileCheckpointStore } from '../checkpointStore';
import { FileApprovalStore } from '../approvalStore';

type Ingested = { type: string; data: Record<string, unknown> };

const SPEC: AgentSpec = {
name: 'builder-agent',
prompt: 'You are a helpful agent.',
provider: { type: 'mock', model: 'mock-1' },
};

function waitForDone(runManager: RunManager, agentId: string): Promise<{ status: string; traceId?: string }> {
return new Promise((resolve) => {
const onStatus = (payload: { agentId: string; status: string; traceId?: string }) => {
if (payload.agentId !== agentId || payload.status !== 'done') return;
runManager.off('status', onStatus);
resolve(payload);
};
runManager.on('status', onStatus);
});
}

describe('builder run reporting', () => {
let baseDir: string;
let ingested: Ingested[];
const savedObserve = process.env.LOUSHO_OBSERVE;

function makeManager(observe: ConstructorParameters<typeof RunManager>[0]['observe']): RunManager {
return new RunManager({
baseDir,
checkpointStore: new FileCheckpointStore(baseDir),
approvalStore: new FileApprovalStore(baseDir),
loadSpec: async () => undefined,
...(observe ? { observe } : {}),
});
}

const fetchFn = () =>
(async (_url: string, init?: RequestInit) => {
const body = JSON.parse(String(init?.body ?? '{}')) as { batch?: Ingested[] };
ingested.push(...(body.batch ?? []));
return new Response(JSON.stringify({ accepted: body.batch?.length ?? 0 }), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}) as typeof fetch;

beforeEach(() => {
delete process.env.LOUSHO_OBSERVE;
baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'builder-report-'));
ingested = [];
});

afterEach(() => {
if (savedObserve === undefined) delete process.env.LOUSHO_OBSERVE;
else process.env.LOUSHO_OBSERVE = savedObserve;
fs.rmSync(baseDir, { recursive: true, force: true });
});

const rootEnd = () =>
ingested.find((e) => e.type === 'observation' && e.data.parent_obs_id === undefined && e.data.end !== undefined);

it("reports a builder run's spans as agent <agentId>, and the status names the trace", async () => {
const runManager = makeManager({ url: 'http://127.0.0.1:7998', fetchFn: fetchFn(), registerExitHooks: false, flushIntervalMs: 10 });
const done = waitForDone(runManager, 'support-desk');
await runManager.run('support-desk', 'hello', SPEC);
const final = await done;
expect(typeof final.traceId).toBe('string');
await vi.waitFor(() => expect(rootEnd()).toBeDefined());
expect(ingested.find((e) => e.type === 'agent.heartbeat')?.data.name).toBe('support-desk');
expect(rootEnd()!.data.trace_id).toBe(final.traceId);
expect(runManager.status('support-desk').traceId).toBe(final.traceId);
});

it('captures message content for a loopback daemon only', async () => {
const reportTo = async (url: string, agentId: string) => {
ingested = [];
const runManager = makeManager({ url, fetchFn: fetchFn(), registerExitHooks: false, flushIntervalMs: 10 });
const done = waitForDone(runManager, agentId);
await runManager.run(agentId, 'hello', SPEC);
await done;
await vi.waitFor(() => expect(rootEnd()).toBeDefined());
return JSON.stringify(ingested);
};
expect(await reportTo('http://127.0.0.1:7998', 'local-agent')).toContain('gen_ai.input.messages');
expect(await reportTo('http://observe.internal:7777', 'remote-agent')).not.toContain('gen_ai.input.messages');
});

it('keeps captured content observe-only: the trace file and live feed get none', async () => {
const runManager = makeManager({ url: 'http://127.0.0.1:7998', fetchFn: fetchFn(), registerExitHooks: false, flushIntervalMs: 10 });
const live: string[] = [];
runManager.on('span', (payload: unknown) => live.push(JSON.stringify(payload)));
const done = waitForDone(runManager, 'private-agent');
await runManager.run('private-agent', 'hello', SPEC);
await done;
await vi.waitFor(() => expect(rootEnd()).toBeDefined());
expect(JSON.stringify(ingested)).toContain('gen_ai.input.messages');

const traceFiles = (fs.readdirSync(agentTraceDir(baseDir, 'private-agent'), { recursive: true }) as string[])
.filter((f) => f.endsWith('.jsonl'))
.map((f) => fs.readFileSync(path.join(agentTraceDir(baseDir, 'private-agent'), f), 'utf8'));
expect(traceFiles.join('')).toContain('invoke_agent');
expect(traceFiles.join('')).not.toContain('gen_ai.input.messages');
expect(live.length).toBeGreaterThan(0);
expect(live.join('')).not.toContain('gen_ai.input.messages');
});

it('a daemon that is down never fails the run', async () => {
const down = (async () => {
throw new Error('ECONNREFUSED');
}) as typeof fetch;
const runManager = makeManager({ url: 'http://127.0.0.1:7998', fetchFn: down, registerExitHooks: false, flushIntervalMs: 10 });
const done = waitForDone(runManager, 'offline');
await runManager.run('offline', 'hello', SPEC);
expect((await done).status).toBe('done');
});

it('without the observe option the manager makes no reporter of its own', async () => {
const runManager = makeManager(undefined);
const done = waitForDone(runManager, 'quiet');
await runManager.run('quiet', 'hello', SPEC);
const final = await done;
// The status still names the run's trace (the local trace file has it too).
expect(typeof final.traceId).toBe('string');
expect(ingested).toEqual([]);
});
});
52 changes: 52 additions & 0 deletions apps/agent-forge/server/__tests__/observeReporting.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
/**
* The reporter factory shared by the playground and builder runs: an env
* reporter (`LOUSHO_OBSERVE`) wins so a run is not reported twice, and
* content is captured only for a loopback target.
*/
import { describe, it, expect, afterEach } from 'vitest';
import { observeExporterFromEnv, type ObserveReporter } from '@lousho/build-ai-agent/traces';
import { isLoopbackUrl, observeReporting } from '../observeReporting';

const noFetch = (async () => new Response('{}', { status: 200 })) as typeof fetch;
const overrides = (url: string) => ({ url, fetchFn: noFetch, registerExitHooks: false });

describe('observeReporting', () => {
const saved = process.env.LOUSHO_OBSERVE;
afterEach(() => {
if (saved === undefined) delete process.env.LOUSHO_OBSERVE;
else process.env.LOUSHO_OBSERVE = saved;
});

it('knows a loopback URL', () => {
expect(isLoopbackUrl('http://127.0.0.1:7777')).toBe(true);
expect(isLoopbackUrl('http://localhost:7777')).toBe(true);
expect(isLoopbackUrl('http://[::1]:7777')).toBe(true);
expect(isLoopbackUrl('http://observe.internal:7777')).toBe(false);
expect(isLoopbackUrl('not a url')).toBe(false);
});

it('captures content only for a loopback target', () => {
const local = observeReporting({ agentName: 'a', baseDir: process.cwd(), overrides: overrides('http://127.0.0.1:7998') });
const remote = observeReporting({ agentName: 'a', baseDir: process.cwd(), overrides: overrides('http://observe.internal:7777') });
expect(local.captureContent()).toBe(true);
expect(remote.captureContent()).toBe(false);
});

it('makes one reporter per agent name, on first use', async () => {
const reporting = observeReporting({ agentName: 'once', baseDir: process.cwd(), overrides: overrides('http://127.0.0.1:7998') });
const first = reporting.exporter() as ObserveReporter;
expect(reporting.exporter()).toBe(first);
expect(first.agentId).toBeTruthy();
await first.close();
});

it('uses the LOUSHO_OBSERVE reporter when set, so nothing is reported twice', async () => {
// A closed loopback port: the env reporter drops-and-counts, nothing real is reached.
process.env.LOUSHO_OBSERVE = 'http://127.0.0.1:1';
const reporting = observeReporting({ agentName: 'env-wins', baseDir: process.cwd() });
const fromEnv = observeExporterFromEnv({ agentName: 'env-wins' }) as ObserveReporter;
expect(reporting.exporter()).toBe(fromEnv);
expect(reporting.captureContent()).toBe(true);
await fromEnv.close();
});
});
9 changes: 9 additions & 0 deletions apps/agent-forge/server/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,13 @@ export interface StartStudioServerOptions {
* minted random one.
*/
token?: string;
/**
* Report builder runs to the observe daemon (`LOUSHO_OBSERVE_URL`, see
* server/observeReporting.ts) so the top bar's "Open trace" resolves.
* Defaults to true; pass `false` to make no reporter (an embedding or a
* test that must not reach a daemon).
*/
observe?: boolean;
}

async function listen(server: http.Server, port: number, host: string): Promise<void> {
Expand Down Expand Up @@ -95,6 +102,8 @@ export async function startStudioServer(
saveSpec: (agentId, spec) => agentStore.save(agentId, spec),
secretsStore,
settingsStore,
// Builder runs report to the observe daemon, so "Open trace" resolves.
...(options.observe === false ? {} : { observe: true as const }),
});

const token = options.token ?? (process.env[STUDIO_TOKEN_ENV] || mintStudioToken());
Expand Down
105 changes: 105 additions & 0 deletions apps/agent-forge/server/observeReporting.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
/**
* The observe reporter a Studio run reports through - shared by the
* playground (agent `playground`, playground.ts) and builder runs (agent =
* the agent id, runRegistry.ts).
*
* Runs report to the daemon the Studio reads (`LOUSHO_OBSERVE_URL`, default
* `http://127.0.0.1:7777`, the same target as observeProxy.ts) through the
* SDK's own `observeExporter()`, so a daemon that is down only means the
* reporter drops-and-counts; a run never waits on it or fails because of it.
* When `LOUSHO_OBSERVE` is set for the studio process, its reporter is used
* instead, so the run is not reported twice. Content (`gen_ai.*` messages,
* tool arguments and results) is captured only for a loopback target.
*/
import {
observeExporter,
observeExporterFromEnv,
type ObserveExporterOptions,
type Span,
type TraceExporter,
} from '@lousho/build-ai-agent/traces';
import { observeTarget } from './observeProxy';

/** One agent name's reporter and whether its runs may capture content. */
export interface ObserveReporting {
/** The reporter for the next run; made on first use (no heartbeat until a run). */
exporter: () => TraceExporter;
/** True only when the reporter's target is a loopback daemon. */
captureContent: () => boolean;
}

export interface ObserveReportingOptions {
/** Every run reports under this agent name. */
agentName: string;
/** Project dir the reporter's `.lousho/identity/` lives under. */
baseDir: string;
/**
* Reporter overrides (tests: `url`, `fetchFn`, `registerExitHooks`,
* `flushIntervalMs`). Given, the `LOUSHO_OBSERVE` reporter is ignored. The
* agent name stays `agentName`.
*/
overrides?: Partial<ObserveExporterOptions>;
}

/** True for a `127.0.0.1` / `localhost` / `[::1]` URL. */
export function isLoopbackUrl(url: string): boolean {
try {
return ['127.0.0.1', 'localhost', '[::1]'].includes(new URL(url).hostname.toLowerCase());
} catch {
return false;
}
}

/** Content capture defaults on for a loopback collector, as `createAgent` does under `LOUSHO_OBSERVE`. */
function envTargetIsLoopback(): boolean {
const value = process.env.LOUSHO_OBSERVE?.trim() ?? '';
return /^(1|true|on|yes)$/i.test(value) || isLoopbackUrl(value);
}

/** The observe reporting for runs under `options.agentName`. */
export function observeReporting(options: ObserveReportingOptions): ObserveReporting {
let own: TraceExporter | undefined;
const { agentName, overrides } = options;
const url = () => overrides?.url ?? observeTarget();
const fromEnv = () => (overrides ? undefined : observeExporterFromEnv({ agentName }));
return {
exporter: () =>
fromEnv() ??
(own ??= observeExporter({
cwd: options.baseDir,
// Parity with the LOUSHO_OBSERVE reporter: a daemon bound beyond loopback demands the bearer.
...(process.env.LOUSHO_OBSERVE_TOKEN ? { token: process.env.LOUSHO_OBSERVE_TOKEN } : {}),
...overrides,
url: url(),
agentName,
})),
captureContent: () => (fromEnv() ? envTargetIsLoopback() : isLoopbackUrl(url())),
};
}

/** The opt-in `gen_ai.*` content attributes `captureContent` records (the SDK's CAPTURED_CONTENT_KEYS). */
const CAPTURED_CONTENT_KEYS: ReadonlySet<string> = new Set([
'gen_ai.system_instructions',
'gen_ai.input.messages',
'gen_ai.output.messages',
'gen_ai.tool.call.arguments',
'gen_ai.tool.call.result',
]);

function stripContent(span: Span): Span {
const kept = Object.entries(span.attributes).filter(([key]) => !CAPTURED_CONTENT_KEYS.has(key));
return kept.length === Object.keys(span.attributes).length ? span : { ...span, attributes: Object.fromEntries(kept) };
}

/**
* `exporter` seeing every span without its captured content - for the
* exporters next to the observe reporter (trace file, live feed), so content
* captured for a loopback daemon stays observe-only, as the SDK's
* `capturedContentForObserveOnly` does for `createAgent`.
*/
export function withoutCapturedContent(exporter: TraceExporter): TraceExporter {
return {
onSpanStart: (span) => exporter.onSpanStart(stripContent(span)),
onSpanEnd: (span) => exporter.onSpanEnd(stripContent(span)),
};
}
Loading
Loading