diff --git a/.gitignore b/.gitignore index 01ada8bdf..79ba39b26 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,8 @@ coverage/ # Generated API docs (npm run docs:build) docs/api/ + +# LOU-N runtime control server's local run state (checkpoints/approvals/ +# saved agent specs) - see apps/agent-forge/server/{checkpointStore, +# approvalStore}.ts and src/persistence/fsAgentStore.ts. +.loushy/ diff --git a/apps/agent-forge/eslint.config.mjs b/apps/agent-forge/eslint.config.mjs index c9f01952c..2bb7627c7 100644 --- a/apps/agent-forge/eslint.config.mjs +++ b/apps/agent-forge/eslint.config.mjs @@ -11,7 +11,7 @@ export default tseslint.config( }, ...tseslint.configs.recommended, { - files: ['src/**/*.{ts,tsx}'], + files: ['src/**/*.{ts,tsx}', 'server/**/*.ts'], rules: { '@typescript-eslint/no-explicit-any': 'warn', '@typescript-eslint/no-unused-vars': 'warn', diff --git a/apps/agent-forge/package.json b/apps/agent-forge/package.json index dce307073..b011bfd0d 100644 --- a/apps/agent-forge/package.json +++ b/apps/agent-forge/package.json @@ -9,26 +9,39 @@ "build": "tsc --noEmit -p tsconfig.json && vite build", "preview": "vite preview", "test": "vitest", - "lint": "eslint src", - "typecheck": "tsc --noEmit -p tsconfig.json" + "lint": "eslint src server", + "typecheck": "tsc --noEmit -p tsconfig.json", + "server:dev": "tsx watch server/index.ts", + "server:start": "tsx server/index.ts", + "test:server": "vitest run -c server/vitest.config.ts", + "typecheck:server": "tsc --noEmit -p server/tsconfig.json" }, "dependencies": { "@loushy/build-ai-agent": "file:../..", "@xyflow/react": "^12.12.0", + "cors": "^2.8.5", "dagre": "^0.8.5", + "express": "^5.2.1", "react": "^18.3.1", "react-dom": "^18.3.1", + "ws": "^8.18.3", "yaml": "^2.9.1" }, "devDependencies": { "@ai-sdk/anthropic": "^0.0.42", "@ai-sdk/openai": "^0.0.42", + "@types/cors": "^2.8.19", "@types/dagre": "^0.7.52", + "@types/express": "^5.0.5", "@types/react": "^18.3.12", "@types/react-dom": "^18.3.1", + "@types/supertest": "^6.0.3", + "@types/ws": "^8.18.1", "@vitejs/plugin-react": "^4.3.4", "jsdom": "^25.0.1", "ollama-ai-provider": "^1.2.0", + "supertest": "^7.3.0", + "tsx": "^4.23.15", "typescript": "^5.9.3", "vite": "^5.4.11", "vitest": "^1.0.0" diff --git a/apps/agent-forge/server/__tests__/app.test.ts b/apps/agent-forge/server/__tests__/app.test.ts new file mode 100644 index 000000000..1f890ba13 --- /dev/null +++ b/apps/agent-forge/server/__tests__/app.test.ts @@ -0,0 +1,165 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import request from 'supertest'; +import type { AgentSpec } from '@loushy/build-ai-agent'; +import { createApp } from '../app'; +import { RunManager } from '../runRegistry'; +import { FileCheckpointStore } from '../checkpointStore'; +import { FileApprovalStore } from '../approvalStore'; +import { createFsAgentStore } from '../../src/persistence/fsAgentStore'; + +const SPEC: AgentSpec = { + name: 'test-agent', + prompt: 'You are a helpful agent.', + provider: { type: 'mock', model: 'mock-1' }, +}; + +function waitForStatus( + runManager: RunManager, + agentId: string, + predicate: (status: { status: string }) => boolean, + timeoutMs = 2000 +): Promise { + // The mock provider resolves near-instantly, so a run kicked off by an + // earlier `await request(app)...` call may already have reached its + // terminal status before this helper's listener is attached - check the + // current status first rather than only listening for a future + // transition that may never come. + const current = runManager.status(agentId); + if (predicate(current)) return Promise.resolve(current); + + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + runManager.off('status', onStatus); + reject(new Error('Timed out waiting for status')); + }, timeoutMs); + function onStatus(payload: any) { + if (payload.agentId !== agentId) return; + if (predicate(payload)) { + clearTimeout(timer); + runManager.off('status', onStatus); + resolve(payload); + } + } + runManager.on('status', onStatus); + }); +} + +describe('LOU-N HTTP API', () => { + let baseDir: string; + let app: ReturnType; + let runManager: RunManager; + + beforeEach(() => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-n-app-test-')); + const agentStore = createFsAgentStore(baseDir); + runManager = new RunManager({ + baseDir, + checkpointStore: new FileCheckpointStore(baseDir), + approvalStore: new FileApprovalStore(baseDir), + loadSpec: (id) => agentStore.load(id), + saveSpec: (id, spec) => agentStore.save(id, spec), + }); + app = createApp({ agentStore, runManager }); + }); + + afterEach(() => { + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + it('GET /health returns ok', async () => { + const res = await request(app).get('/health'); + expect(res.status).toBe(200); + expect(res.text).toBe('ok'); + }); + + it('GET /agents/:id/status returns idle before any run', async () => { + const res = await request(app).get('/agents/nope/status'); + expect(res.status).toBe(200); + expect(res.body.status).toBe('idle'); + }); + + it('rejects a path-traversal agent id instead of writing outside .loushy/agents', async () => { + // Express URL-decodes `:id` before handing it to the route, so a + // percent-encoded `..%2F..%2Fpwned` arrives as a plain string + // containing `/` and `..` - without the isValidAgentId guard in app.ts, + // this would let PUT /agents/:id escape `.loushy/agents/` via + // fsAgentStore.ts's path.join(agentsDir, `${id}.yaml`). + const traversalId = encodeURIComponent('../../pwned'); + const put = await request(app).put(`/agents/${traversalId}`).send(SPEC); + expect(put.status).toBe(400); + + const escapedFile = path.join(baseDir, '..', 'pwned.yaml'); + expect(fs.existsSync(escapedFile)).toBe(false); + + const run = await request(app).post(`/agents/${traversalId}/run`).send({ input: 'hi', spec: SPEC }); + expect(run.status).toBe(400); + + const status = await request(app).get(`/agents/${traversalId}/status`); + expect(status.status).toBe(400); + }); + + it('PUT then GET round-trips an AgentSpec', async () => { + const put = await request(app).put('/agents/foo').send(SPEC); + expect(put.status).toBe(204); + + const get = await request(app).get('/agents/foo'); + expect(get.status).toBe(200); + expect(get.body.name).toBe('test-agent'); + }); + + it('PUT /agents/:id rejects a malformed spec', async () => { + const res = await request(app).put('/agents/foo').send({ nonsense: true }); + expect(res.status).toBe(400); + }); + + it('POST /agents/:id/run starts a run and GET /status reflects it finishing', async () => { + const run = await request(app).post('/agents/bar/run').send({ spec: SPEC, input: 'hello' }); + expect(run.status).toBe(202); + expect(['running', 'stopped']).toContain(run.body.status); + + const final = await waitForStatus(runManager, 'bar', (s) => s.status === 'stopped'); + expect(final.resultText).toBeTruthy(); + + const status = await request(app).get('/agents/bar/status'); + expect(status.body.status).toBe('stopped'); + }); + + it('POST /agents/:id/run without input is rejected', async () => { + const res = await request(app).post('/agents/bar/run').send({ spec: SPEC }); + expect(res.status).toBe(400); + }); + + // The "already running" 409 guard itself is covered deterministically at + // the RunManager unit level (runRegistry.test.ts) by calling run() twice + // back-to-back with no intervening await. Reproducing that race through + // a real HTTP round-trip (supertest) against the near-instant mock + // provider is flaky: enough real event-loop turns elapse between the two + // requests that the first run can legitimately finish before the second + // one is even dispatched, so this is intentionally not re-tested here. + + it('POST /agents/:id/run for an agent with no saved spec and no spec in the body returns 404', async () => { + const res = await request(app).post('/agents/ghost/run').send({ input: 'hi' }); + expect(res.status).toBe(404); + }); + + it('POST /agents/:id/stop on an idle agent is a safe no-op (202)', async () => { + const res = await request(app).post('/agents/never-ran/stop'); + expect(res.status).toBe(202); + expect(res.body.status).toBe('idle'); + }); + + it('POST /agents/:id/approve with no pending approval returns 409', async () => { + const res = await request(app) + .post('/agents/none-pending/approve') + .send({ approvalId: 'x', approved: true }); + expect(res.status).toBe(409); + }); + + it('POST /agents/:id/approve with a bad body returns 400', async () => { + const res = await request(app).post('/agents/foo/approve').send({ approvalId: 'x' }); + expect(res.status).toBe(400); + }); +}); diff --git a/apps/agent-forge/server/__tests__/approvalFlow.test.ts b/apps/agent-forge/server/__tests__/approvalFlow.test.ts new file mode 100644 index 000000000..8209cb7f9 --- /dev/null +++ b/apps/agent-forge/server/__tests__/approvalFlow.test.ts @@ -0,0 +1,142 @@ +/** + * Exercises the exact approval-gate pause/resume machinery + * RunManager.approve() (runRegistry.ts) drives - AgentExecutor.execute() + * pausing on a tool that needs approval, FileApprovalStore persisting the + * pending decision, and resumeAfterApproval() continuing the run - using + * the public SDK API directly. RunManager itself only wires these together + * (buildAgentFromSpec + a thin dispatch), so this is where the actual + * pause -> approve -> resume round trip is verified; runRegistry.test.ts and + * app.test.ts cover the run()/stop() and HTTP-layer behavior around it. + * + * None of the LOU-N server's spec-resolvable tools (http/current-date/ + * day-name, see buildAgent.ts + specToAgent.ts's RESOLVABLE_BUILT_IN_TOOLS) + * set `needsApproval`, so a spec alone can't trigger this path end-to-end + * through the HTTP API in a test - a custom ToolDescriptor is used here + * instead, exactly mirroring what a real approval-gated tool would do. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import { + AgentExecutor, + ToolRegistry, + AgentBuilder, + AgentType, + createMockProvider, + resumeAfterApproval, + type ToolDescriptor, +} from '@loushy/build-ai-agent'; +import { FileApprovalStore } from '../approvalStore'; +import { FileCheckpointStore } from '../checkpointStore'; + +describe('approval-gate pause/resume (FileApprovalStore + FileCheckpointStore)', () => { + let baseDir: string; + let approvalStore: FileApprovalStore; + let checkpointStore: FileCheckpointStore; + + beforeEach(() => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-n-approval-test-')); + approvalStore = new FileApprovalStore(baseDir); + checkpointStore = new FileCheckpointStore(baseDir); + }); + + afterEach(() => { + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + function buildGatedAgentConfig(agentId: string) { + const dangerousTool: ToolDescriptor = { + displayName: 'Dangerous Tool', + tool: { + description: 'Does something dangerous', + parameters: { type: 'object', properties: {} }, + execute: async () => ({ done: true }), + } as any, + needsApproval: true, + }; + const toolRegistry = new ToolRegistry(); + toolRegistry.register('dangerous', dangerousTool); + + const provider = createMockProvider({ + name: 'mock', + responses: ['calling the dangerous tool', 'all done'], + }); + + const agent = AgentBuilder.create() + .setId(agentId) + .setType(AgentType.SmartAssistant) + .setName('gated-agent') + .setPrompt('You are a helpful agent.') + .setTools({ dangerous: { tool: 'dangerous' } }) + .build(); + + return { agent, provider, toolRegistry }; + } + + it('pauses on a needsApproval tool, persists the pending decision, and resumes on approve', async () => { + const agentId = 'gated-agent-1'; + const { agent, provider, toolRegistry } = buildGatedAgentConfig(agentId); + + const paused = await AgentExecutor.execute({ + agent, + input: 'please call the dangerous tool', + provider, + toolRegistry, + approvalStore, + sessionId: agentId, + checkpointStore, + }); + + expect(paused.finishReason).toBe('awaiting-approval'); + expect(paused.approvalId).toBeTruthy(); + + // Exactly what runRegistry.ts's handleRunSettled() reads to build the + // UI's pending-approval card. + const record = await approvalStore.peek(agentId, paused.approvalId!); + expect(record?.pending.toolName).toBe('dangerous'); + + const resumed = await resumeAfterApproval( + { id: paused.approvalId!, approved: true }, + approvalStore, + toolRegistry, + provider, + {}, + checkpointStore + ); + + expect(resumed.finishReason).not.toBe('awaiting-approval'); + expect(resumed.text).toBeTruthy(); + + // The approval record is delete-on-read. + expect(await approvalStore.peek(agentId, paused.approvalId!)).toBeNull(); + }); + + it('a rejected approval resumes the run with a rejection tool-result instead of executing the tool', async () => { + const agentId = 'gated-agent-2'; + const { agent, provider, toolRegistry } = buildGatedAgentConfig(agentId); + + const paused = await AgentExecutor.execute({ + agent, + input: 'please call the dangerous tool', + provider, + toolRegistry, + approvalStore, + sessionId: agentId, + checkpointStore, + }); + expect(paused.finishReason).toBe('awaiting-approval'); + + const resumed = await resumeAfterApproval( + { id: paused.approvalId!, approved: false, note: 'not today' }, + approvalStore, + toolRegistry, + provider, + {}, + checkpointStore + ); + + const toolMessage = resumed.messages.find((m) => m.role === 'tool'); + expect(toolMessage?.content).toContain('rejected'); + }); +}); diff --git a/apps/agent-forge/server/__tests__/runRegistry.test.ts b/apps/agent-forge/server/__tests__/runRegistry.test.ts new file mode 100644 index 000000000..290ffa846 --- /dev/null +++ b/apps/agent-forge/server/__tests__/runRegistry.test.ts @@ -0,0 +1,132 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import type { AgentSpec } from '@loushy/build-ai-agent'; +import { RunManager } from '../runRegistry'; +import { FileCheckpointStore } from '../checkpointStore'; +import { FileApprovalStore } from '../approvalStore'; + +const SPEC: AgentSpec = { + name: 'test-agent', + prompt: 'You are a helpful agent.', + provider: { type: 'mock', model: 'mock-1' }, + tools: ['current-date'], +}; + +function waitForStatus( + runManager: RunManager, + agentId: string, + predicate: (status: { status: string }) => boolean, + timeoutMs = 2000 +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + runManager.off('status', onStatus); + reject(new Error('Timed out waiting for status')); + }, timeoutMs); + function onStatus(payload: any) { + if (payload.agentId !== agentId) return; + if (predicate(payload)) { + clearTimeout(timer); + runManager.off('status', onStatus); + resolve(payload); + } + } + runManager.on('status', onStatus); + }); +} + +describe('RunManager', () => { + let baseDir: string; + let runManager: RunManager; + + beforeEach(() => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-n-test-')); + runManager = new RunManager({ + baseDir, + checkpointStore: new FileCheckpointStore(baseDir), + approvalStore: new FileApprovalStore(baseDir), + loadSpec: async () => undefined, + saveSpec: async () => {}, + }); + }); + + afterEach(() => { + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + it('reports idle status for an agent that has never run', () => { + const status = runManager.status('nope'); + expect(status.status).toBe('idle'); + }); + + it('runs an agent to completion and reports the result text', async () => { + await runManager.run('agent-1', 'please use current-date', SPEC); + const final = await waitForStatus(runManager, 'agent-1', (s) => s.status === 'stopped'); + expect(final.resultText).toBe('This is a mock response.'); + }); + + it('rejects a second run() call while one is already in flight', async () => { + await runManager.run('agent-2', 'please use current-date', SPEC); + await expect(runManager.run('agent-2', 'again', SPEC)).rejects.toThrow(/already running/i); + await waitForStatus(runManager, 'agent-2', (s) => s.status === 'stopped'); + }); + + it('throws when running an agent id with no spec available', async () => { + await expect(runManager.run('unknown-agent', 'hi')).rejects.toThrow(/no agent spec/i); + }); + + it('stop() then run() resumes from the last checkpoint instead of restarting', async () => { + const checkpointStore = new FileCheckpointStore(baseDir); + const agentId = 'agent-3'; + + // Abort the run right after the first tool result is recorded (and its + // checkpoint written), but before the loop's next provider.generate() + // call - deterministic because this handler runs synchronously within + // the same event-loop turn AgentExecutor's onEvent callback fires in, + // before the `continue` to the next step's generate() call is reached. + let stopped = false; + runManager.on('event', (id: string, event: any) => { + if (id === agentId && event.type === 'tool-result' && !stopped) { + stopped = true; + runManager.stop(agentId); + } + }); + + await runManager.run(agentId, 'please use current-date', SPEC); + await waitForStatus(runManager, agentId, (s) => s.status === 'stopped'); + + // The checkpoint from the aborted run must survive - AgentExecutor only + // deletes it on a successful terminal completion, never on an aborted + // run (see abortableProvider.ts's doc comment). + const checkpoint = await checkpointStore.load(agentId); + expect(checkpoint).not.toBeNull(); + expect(checkpoint!.stepIndex).toBeGreaterThanOrEqual(1); + + // Running again picks the same checkpoint back up (AgentExecutor + // rehydrates from it and ignores the new `input`) rather than starting + // a brand new conversation from scratch. + await runManager.run(agentId, 'this input is ignored on resume', SPEC); + const resumed = await waitForStatus(runManager, agentId, (s) => s.status === 'stopped'); + expect(resumed.resultText).toBe('This is a mock response.'); + + // And the checkpoint is cleared once that resumed run completes + // successfully. + const afterResume = await checkpointStore.load(agentId); + expect(afterResume).toBeNull(); + }); + + it('surfaces a thrown provider error as status "error" without crashing the process', async () => { + const errorSpec: AgentSpec = { + name: 'erroring-agent', + prompt: 'You are a helpful agent.', + // 'mock' provider type ignores model-specific errors, so simulate one + // via a spec whose provider type is unregistered - resolveSpecProvider + // throws synchronously inside buildAgentFromSpec(), which run() + // currently calls outside its own try/catch (see below). + provider: { type: 'definitely-not-a-real-provider', model: 'x' }, + }; + await expect(runManager.run('agent-4', 'hi', errorSpec)).rejects.toThrow(); + }); +}); diff --git a/apps/agent-forge/server/abortableProvider.ts b/apps/agent-forge/server/abortableProvider.ts new file mode 100644 index 000000000..aa1a13e7e --- /dev/null +++ b/apps/agent-forge/server/abortableProvider.ts @@ -0,0 +1,80 @@ +/** + * Cancellation support for `POST /agents/:id/stop`. + * + * `AgentExecutor.execute()` (src/execution/AgentExecutor.ts, read in full + * for this ticket) is a static, instance-free async call with NO + * cancellation-token option in `ExecuteOptions` at all - no `signal`, no + * `abortSignal`, nothing. Its internal `while (steps < maxSteps)` loop has + * exactly one await point per step that can take real wall-clock time: + * `provider.generate(generateRequest)`. Everything else in a step (pushing + * to arrays, JSON.stringify, checkpoint writes) is effectively synchronous. + * + * So rather than inventing an API AgentExecutor doesn't have, cancellation + * here is implemented the way the ticket's fallback describes: "kill the + * in-flight provider call". This wraps whatever LLMProvider the run was + * built with with one that checks an AbortSignal immediately before + * calling through to the real `generate()`/`stream()`, and throws an + * AbortError instead of calling through if the signal is already aborted. + * + * Effect: `stop()` (runRegistry.ts) aborts the controller; the NEXT + * provider call this wrapper makes (at the very next step boundary, or + * immediately if a call is currently pending model-side and the underlying + * provider is itself abort-aware) throws, which propagates out of + * AgentExecutor.execute()'s catch block (it only emits an 'error' event and + * rethrows - it does not swallow) as a rejected promise. Crucially, + * AgentExecutor only clears the CheckpointStore entry on a *successful* + * terminal completion (see the bottom of runAgentLoop(), reached only after + * the while-loop `break`s) - never from the catch block - so an abort + * leaves the last-saved checkpoint (written after each completed tool + * result) intact. The next `run()` call for the same agent reuses the same + * sessionId, so AgentExecutor.execute() rehydrates from exactly that + * checkpoint instead of starting over. This is what makes Stop-then-Run + * "resume from the last checkpoint" rather than "restart from scratch". + * + * Caveat (documented, not fixed here - real-provider wiring is LOU-R's + * job): for the MockLLMProvider used by default in this epic, `generate()` + * has no real network I/O to cancel, so this wrapper's pre-call check is + * the entire cancellation mechanism - a mock call already in flight when + * `stop()` fires will still resolve, but the loop will not continue to a + * further step. A real provider wired up later should thread the same + * AbortSignal into its underlying fetch/SDK call for true mid-request + * cancellation; this wrapper's shape (signal checked, forwarded if the + * real provider accepts one) is what that wiring would build on. + */ +import type { LLMProvider, GenerateOptions, GenerateResult, StreamResult } from '@loushy/build-ai-agent'; + +export class RunAbortedError extends Error { + constructor() { + super('Run was stopped'); + this.name = 'RunAbortedError'; + } +} + +export function withAbortSignal(provider: LLMProvider, signal: AbortSignal): LLMProvider { + function checkAborted(): void { + if (signal.aborted) throw new RunAbortedError(); + } + + return { + name: provider.name, + async generate(options: GenerateOptions): Promise { + checkAborted(); + const result = await provider.generate(options); + checkAborted(); + return result; + }, + async stream(options: GenerateOptions): Promise { + checkAborted(); + return provider.stream(options); + }, + supportsTools(model: string): boolean { + return provider.supportsTools(model); + }, + supportsStreaming(model: string): boolean { + return provider.supportsStreaming(model); + }, + async getModels(): Promise { + return provider.getModels(); + }, + }; +} diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts new file mode 100644 index 000000000..c4e35c848 --- /dev/null +++ b/apps/agent-forge/server/app.ts @@ -0,0 +1,176 @@ +/** + * LOU-N runtime control server - HTTP API (N1). + * + * Routes: + * GET /health -> 200 'ok' + * GET /agents -> AgentStoreEntry[] (saved agent specs on disk) + * GET /agents/:id -> AgentSpec | 404 + * PUT /agents/:id -> save an AgentSpec to disk (body: AgentSpec) + * POST /agents/:id/run -> body: { spec?: AgentSpec, input: string } + * POST /agents/:id/stop -> abort the in-flight run, if any + * GET /agents/:id/status -> AgentRunStatusPayload + * POST /agents/:id/approve -> body: { approvalId, approved, note? } + * + * `WS /agents/:id/stream` is wired separately in wsServer.ts (attached to + * the same underlying http.Server, since `ws` needs the raw HTTP upgrade + * event rather than an Express route). + * + * Built on Express (over Fastify): this SDK's only existing local server + * (src/cli/dev.ts) is hand-rolled `node:http`, so there's no established + * in-repo convention to match either way. Express is chosen for its + * ubiquity (every contributor already knows its routing/middleware model) + * and because `supertest` (the natural HTTP-layer test tool for this + * ticket) is built and documented around it; Fastify's JSON-schema-first + * validation wasn't worth the extra learning cost for ~6 small routes. + */ +import express, { type Express, type Request, type Response, type NextFunction } from 'express'; +import cors from 'cors'; +import type { AgentSpec } from '@loushy/build-ai-agent'; +import type { AgentStore } from '../src/persistence/AgentStore'; +import { RunManager, AgentNotFoundError, AlreadyRunningError, NoActiveRunError } from './runRegistry'; +import { isValidAgentId } from './types'; + +export interface CreateAppOptions { + agentStore: AgentStore; + runManager: RunManager; +} + +function asyncRoute(fn: (req: Request, res: Response) => Promise) { + return (req: Request, res: Response, next: NextFunction) => { + fn(req, res).catch(next); + }; +} + +/** + * Express 5's `req.params[name]` is typed `string | string[]` (a route + * param can repeat, e.g. `/:id+`). None of this file's routes use repeating + * params, so `:id` is always a single string in practice - this just + * narrows the type for TS rather than changing any behavior. + */ +function paramId(req: Request): string { + const id = req.params.id; + return Array.isArray(id) ? id[0] : id; +} + +export function createApp({ agentStore, runManager }: CreateAppOptions): Express { + const app = express(); + app.use(cors()); + app.use(express.json({ limit: '2mb' })); + + app.get('/health', (_req, res) => res.status(200).send('ok')); + + // Every `/agents/:id/**` route below eventually turns `:id` into a + // filesystem path segment (agent spec YAML, checkpoint/approval files - + // see fsAgentStore.ts/checkpointStore.ts/approvalStore.ts). Reject + // anything that isn't a safe single-segment token here, once, rather than + // trusting each store to sanitize it - closes off path traversal via a + // percent-encoded `..%2F..%2F...` id, which Express happily hands to + // `req.params.id` as a decoded string containing `/`/`..`. + app.use('/agents/:id', (req, res, next) => { + if (!isValidAgentId(paramId(req))) { + res.status(400).json({ error: 'Invalid agent id' }); + return; + } + next(); + }); + + app.get( + '/agents', + asyncRoute(async (_req, res) => { + res.json(await agentStore.list()); + }) + ); + + app.get( + '/agents/:id', + asyncRoute(async (req, res) => { + const spec = await agentStore.load(paramId(req)); + if (!spec) { + res.status(404).json({ error: `No saved agent '${paramId(req)}'` }); + return; + } + res.json(spec); + }) + ); + + app.put( + '/agents/:id', + asyncRoute(async (req, res) => { + const spec = req.body as AgentSpec; + if (!spec || typeof spec !== 'object' || !spec.name || !spec.prompt || !spec.provider) { + res.status(400).json({ error: 'Request body must be a valid AgentSpec' }); + return; + } + await agentStore.save(paramId(req), spec); + res.status(204).end(); + }) + ); + + app.post( + '/agents/:id/run', + asyncRoute(async (req, res) => { + const { spec, input } = req.body as { spec?: AgentSpec; input?: string }; + if (typeof input !== 'string' || !input) { + res.status(400).json({ error: "Request body must include a non-empty 'input' string" }); + return; + } + try { + await runManager.run(paramId(req), input, spec); + res.status(202).json(runManager.status(paramId(req))); + } catch (error) { + if (error instanceof AlreadyRunningError) { + res.status(409).json({ error: error.message }); + return; + } + if (error instanceof AgentNotFoundError) { + res.status(404).json({ error: error.message }); + return; + } + throw error; + } + }) + ); + + app.post('/agents/:id/stop', (req, res) => { + runManager.stop(paramId(req)); + res.status(202).json(runManager.status(paramId(req))); + }); + + app.get('/agents/:id/status', (req, res) => { + res.json(runManager.status(paramId(req))); + }); + + app.post( + '/agents/:id/approve', + asyncRoute(async (req, res) => { + const { approvalId, approved, note } = req.body as { + approvalId?: string; + approved?: boolean; + note?: string; + }; + if (typeof approvalId !== 'string' || typeof approved !== 'boolean') { + res + .status(400) + .json({ error: "Request body must include 'approvalId' (string) and 'approved' (boolean)" }); + return; + } + try { + await runManager.approve(paramId(req), approvalId, approved, note); + res.status(202).json(runManager.status(paramId(req))); + } catch (error) { + if (error instanceof NoActiveRunError || error instanceof AgentNotFoundError) { + res.status(409).json({ error: error.message }); + return; + } + throw error; + } + }) + ); + + // eslint-disable-next-line @typescript-eslint/no-unused-vars + app.use((err: Error, _req: Request, res: Response, _next: NextFunction) => { + res.status(500).json({ error: err.message }); + }); + + return app; +} diff --git a/apps/agent-forge/server/approvalStore.ts b/apps/agent-forge/server/approvalStore.ts new file mode 100644 index 000000000..4b6c50399 --- /dev/null +++ b/apps/agent-forge/server/approvalStore.ts @@ -0,0 +1,64 @@ +/** + * File-backed `ApprovalStore` (src/execution/ApprovalGate.ts, public SDK + * API) for the LOU-N runtime control server, persisting pending approvals + * under `/.loushy/agents//approvals/.json`. + * + * Sibling to FileCheckpointStore.ts - see its doc comment for why this is a + * separate store rather than extending fsAgentStore.ts. + */ +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import type { ApprovalStore, ExecutionSnapshot, PendingApproval, ResolvedApproval } from '@loushy/build-ai-agent'; + +export class FileApprovalStore implements ApprovalStore { + constructor(private readonly baseDir: string) {} + + private dir(agentId: string): string { + return path.join(this.baseDir, '.loushy', 'agents', agentId, 'approvals'); + } + + private filePath(agentId: string, approvalId: string): string { + return path.join(this.dir(agentId), `${approvalId}.json`); + } + + async save(pending: PendingApproval, snapshot: ExecutionSnapshot): Promise { + const agentId = pending.agentId || 'unknown'; + fs.mkdirSync(this.dir(agentId), { recursive: true }); + const record: ResolvedApproval = { pending, snapshot }; + fs.writeFileSync(this.filePath(agentId, pending.id), JSON.stringify(record), 'utf8'); + } + + /** + * `resolve()` is delete-on-read (matching StorageServiceApprovalStore's + * documented contract) but the on-disk layout is per-agent, so this scans + * every agent's approvals directory for a matching `.json`. + * The registry (runRegistry.ts) always calls this with the agentId it + * already knows in hand via a narrower `resolveFor()` below, which avoids + * the scan in the common case - this full scan is a fallback for the + * generic `ApprovalStore` interface contract (e.g. direct SDK use). + */ + async resolve(approvalId: string): Promise { + const agentsDir = path.join(this.baseDir, '.loushy', 'agents'); + if (!fs.existsSync(agentsDir)) return null; + for (const agentId of fs.readdirSync(agentsDir)) { + const found = await this.resolveFor(agentId, approvalId); + if (found) return found; + } + return null; + } + + async resolveFor(agentId: string, approvalId: string): Promise { + const file = this.filePath(agentId, approvalId); + if (!fs.existsSync(file)) return null; + const record = JSON.parse(fs.readFileSync(file, 'utf8')) as ResolvedApproval; + fs.unlinkSync(file); + return record; + } + + /** Reads a saved approval record without deleting it (runRegistry.ts uses this to render the pending-approval card). */ + async peek(agentId: string, approvalId: string): Promise { + const file = this.filePath(agentId, approvalId); + if (!fs.existsSync(file)) return null; + return JSON.parse(fs.readFileSync(file, 'utf8')) as ResolvedApproval; + } +} diff --git a/apps/agent-forge/server/buildAgent.ts b/apps/agent-forge/server/buildAgent.ts new file mode 100644 index 000000000..eba62bac1 --- /dev/null +++ b/apps/agent-forge/server/buildAgent.ts @@ -0,0 +1,54 @@ +/** + * Compiles an `AgentSpec` (the app's `graphToSpec()` output, or whatever a + * saved agent's spec file holds) into the `{agent, provider, toolRegistry}` + * triple `AgentExecutor.execute()` needs directly. + * + * `src/spec/specToAgent.ts`'s own `specToAgent()` does something similar, + * but it returns a `SimpleAgent` (`{send(message)}`) that hides the + * AgentConfig/provider/toolRegistry inside a closure and calls + * `AgentExecutor.execute()` with no `sessionId`/`checkpointStore`/ + * `approvalStore`/`onEvent`/abortable-provider hooks - exactly the things + * this server needs to wire up N2/N3. So this file composes the same + * public building blocks `specToAgent()` uses + * (`resolveSpecProvider`/`resolveSpecTool`/`AgentBuilder`/`ToolRegistry`, + * all public SDK exports) itself, rather than reaching past `specToAgent()` + * into SDK internals. + */ +import { + AgentBuilder, + AgentType, + ToolRegistry, + resolveSpecProvider, + resolveSpecTool, + type AgentConfig, + type AgentSpec, + type LLMProvider, +} from '@loushy/build-ai-agent'; + +export interface BuiltAgent { + agent: AgentConfig; + provider: LLMProvider; + toolRegistry?: ToolRegistry; +} + +export function buildAgentFromSpec(spec: AgentSpec, agentId: string): BuiltAgent { + const provider = resolveSpecProvider(spec.provider.type, spec.provider.model); + + let toolRegistry: ToolRegistry | undefined; + const toolsConfig: Record = {}; + for (const name of spec.tools || []) { + if (!toolRegistry) toolRegistry = new ToolRegistry(); + toolRegistry.register(name, resolveSpecTool(name)); + toolsConfig[name] = { tool: name }; + } + + const agent = AgentBuilder.create() + .setId(agentId) + .setType(AgentType.SmartAssistant) + .setName(spec.name) + .setPrompt(spec.prompt) + .setTools(toolsConfig) + .build(); + + return { agent, provider, toolRegistry }; +} diff --git a/apps/agent-forge/server/checkpointStore.ts b/apps/agent-forge/server/checkpointStore.ts new file mode 100644 index 000000000..e65fc10f8 --- /dev/null +++ b/apps/agent-forge/server/checkpointStore.ts @@ -0,0 +1,61 @@ +/** + * File-backed `CheckpointStore` (src/execution/checkpoint.ts, public SDK + * API) for the LOU-N runtime control server, persisting under + * `/.loushy/agents//checkpoints/.json`. + * + * This is exactly the seam LOU-L's `createFsAgentStore()` TODO pointed at + * ("wire this up behind the runtime control server"): a sibling store next + * to it, following the same `.loushy/agents//...` layout, rather than + * extending fsAgentStore.ts itself (agent specs and run checkpoints are + * different lifecycles/read-write patterns, so separate files/stores). + * + * Consumes only the public `CheckpointStore` interface - no SDK internals. + */ +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import type { Checkpoint, CheckpointStore } from '@loushy/build-ai-agent'; + +export class FileCheckpointStore implements CheckpointStore { + constructor(private readonly baseDir: string) {} + + private dir(agentId: string): string { + return path.join(this.baseDir, '.loushy', 'agents', agentId, 'checkpoints'); + } + + private filePath(agentId: string, sessionId: string): string { + return path.join(this.dir(agentId), `${sessionId}.json`); + } + + /** + * `CheckpointStore.save/load/delete` are keyed by `sessionId` alone (see + * src/execution/checkpoint.ts), but this store's on-disk layout is + * per-agent (`.loushy/agents//checkpoints/`). Sessions in this + * server are always created as `${agentId}:${runNumber}` (see + * runRegistry.ts), so the agentId is recovered by splitting on the first + * `:` - this keeps the public `CheckpointStore` interface unchanged while + * still getting the per-agent directory layout the ticket asks for. + */ + private agentIdFromSessionId(sessionId: string): string { + const idx = sessionId.indexOf(':'); + return idx === -1 ? sessionId : sessionId.slice(0, idx); + } + + async save(sessionId: string, checkpoint: Checkpoint): Promise { + const agentId = this.agentIdFromSessionId(sessionId); + fs.mkdirSync(this.dir(agentId), { recursive: true }); + fs.writeFileSync(this.filePath(agentId, sessionId), JSON.stringify(checkpoint), 'utf8'); + } + + async load(sessionId: string): Promise { + const agentId = this.agentIdFromSessionId(sessionId); + const file = this.filePath(agentId, sessionId); + if (!fs.existsSync(file)) return null; + return JSON.parse(fs.readFileSync(file, 'utf8')) as Checkpoint; + } + + async delete(sessionId: string): Promise { + const agentId = this.agentIdFromSessionId(sessionId); + const file = this.filePath(agentId, sessionId); + if (fs.existsSync(file)) fs.unlinkSync(file); + } +} diff --git a/apps/agent-forge/server/index.ts b/apps/agent-forge/server/index.ts new file mode 100644 index 000000000..bd1b06636 --- /dev/null +++ b/apps/agent-forge/server/index.ts @@ -0,0 +1,107 @@ +/** + * LOU-N runtime control server entry point. + * + * Started either directly (`tsx server/index.ts`, e.g. from `loushy studio` + * - see src/cli/studio.ts) or programmatically via `startStudioServer()` + * for tests/embedding. + */ +import * as http from 'node:http'; +import * as path from 'node:path'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import { createFsAgentStore } from '../src/persistence/fsAgentStore'; +import { createApp } from './app'; +import { attachWebSocketServer } from './wsServer'; +import { RunManager } from './runRegistry'; +import { FileCheckpointStore } from './checkpointStore'; +import { FileApprovalStore } from './approvalStore'; + +export interface StudioServerHandle { + server: http.Server; + port: number; + runManager: RunManager; + close: () => Promise; +} + +export interface StartStudioServerOptions { + /** Directory `.loushy/agents/**` is read from/written to. Defaults to process.cwd(). */ + baseDir?: string; + port?: number; + host?: string; +} + +export async function startStudioServer( + options: StartStudioServerOptions = {} +): Promise { + const baseDir = options.baseDir ?? process.cwd(); + const port = options.port ?? 4750; + const host = options.host ?? '127.0.0.1'; + + const agentStore = createFsAgentStore(baseDir); + const checkpointStore = new FileCheckpointStore(baseDir); + const approvalStore = new FileApprovalStore(baseDir); + + const runManager = new RunManager({ + baseDir, + checkpointStore, + approvalStore, + loadSpec: (agentId) => agentStore.load(agentId), + saveSpec: (agentId, spec) => agentStore.save(agentId, spec), + }); + + const app = createApp({ agentStore, runManager }); + const server = http.createServer(app); + attachWebSocketServer(server, runManager); + + await new Promise((resolve, reject) => { + const onError = (err: NodeJS.ErrnoException) => { + server.removeListener('listening', onListening); + if (err.code === 'EADDRINUSE') { + reject(new Error(`[loushy studio] API server port ${port} is already in use.`)); + } else { + reject(err); + } + }; + const onListening = () => { + server.removeListener('error', onError); + resolve(); + }; + server.once('error', onError); + server.once('listening', onListening); + server.listen(port, host); + }); + + return { + server, + port, + runManager, + close: () => new Promise((resolve, reject) => server.close((err) => (err ? reject(err) : resolve()))), + }; +} + +// Allow `tsx server/index.ts` (or `node --import tsx server/index.ts`) to +// boot the server directly, reading PORT/HOST/BASE_DIR from the +// environment - this is how `loushy studio` (src/cli/studio.ts) launches it +// as a child process. `apps/agent-forge/package.json` has `"type": "module"`, +// so this module runs as real ESM under tsx - there is no CJS `require`/ +// `module` to compare against, hence the `import.meta.url` check instead of +// the usual `require.main === module` idiom. +const isMainModule = + process.argv[1] !== undefined && import.meta.url === pathToFileURL(process.argv[1]).href; + +if (isMainModule) { + const port = process.env.PORT ? Number(process.env.PORT) : undefined; + const host = process.env.HOST; + const moduleDir = path.dirname(fileURLToPath(import.meta.url)); + // moduleDir is apps/agent-forge/server - the repo root (where `.loushy/` + // should live) is three levels up, unless BASE_DIR is set explicitly. + const baseDir = process.env.BASE_DIR ?? path.resolve(moduleDir, '..', '..', '..'); + + startStudioServer({ port, host, baseDir }) + .then((handle) => { + console.log(`[loushy studio] API server listening on http://${host ?? '127.0.0.1'}:${handle.port}`); + }) + .catch((error) => { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; + }); +} diff --git a/apps/agent-forge/server/runRegistry.ts b/apps/agent-forge/server/runRegistry.ts new file mode 100644 index 000000000..c518536c4 --- /dev/null +++ b/apps/agent-forge/server/runRegistry.ts @@ -0,0 +1,286 @@ +/** + * In-memory registry mapping agent IDs to live `AgentExecutor.execute()` + * invocations, tracked by `sessionId` (N1). One `RunEntry` per agent ID - + * only one live run per agent at a time, matching the app's Run/Stop/Debug + * buttons which operate on "the currently loaded agent". + * + * This is the piece that actually wires N2 (start/stop/status) and N3 + * (durable-execution resume + approval-gate pause/resume) together on top + * of the SDK's public execution API. + */ +import { EventEmitter } from 'node:events'; +import { + AgentExecutor, + ToolRegistry, + type ExecutionEvent, + type CheckpointStore, + resumeAfterApproval, + type AgentSpec, +} from '@loushy/build-ai-agent'; +import { buildAgentFromSpec } from './buildAgent'; +import { withAbortSignal, RunAbortedError } from './abortableProvider'; +import { FileApprovalStore } from './approvalStore'; +import type { AgentRunStatusPayload, RunStatus } from './types'; + +interface RunEntry { + status: RunStatus; + /** + * Stable for the lifetime of this agent id in the registry - always just + * `agentId` (see run()'s doc comment for why this must NOT change between + * a stop() and the next run() call). + */ + sessionId: string; + controller?: AbortController; + error?: string; + reason?: 'awaiting_approval'; + pendingApproval?: AgentRunStatusPayload['pendingApproval']; + resultText?: string; + updatedAt: string; + /** + * The AgentSpec this agent id was last run() with. approve() needs a spec + * to rebuild the provider/toolRegistry for the follow-up execute() call + * but the app doesn't resend one on `POST /agents/:id/approve` - it just + * resolves the pending decision - so the spec from the run that paused is + * kept here for reuse. Also used as the loadSpec() fallback's write-through + * cache is unnecessary since this is already in memory for the lifetime of + * the process. + */ + lastSpec?: AgentSpec; +} + +export interface RunManagerOptions { + baseDir: string; + checkpointStore: CheckpointStore; + /** Must be a FileApprovalStore (agent-scoped resolveFor) - see approve(). */ + approvalStore: FileApprovalStore; + /** + * Fallback spec loader used when run() is called without an explicit + * `spec` (e.g. a restarted server resuming via /status polling) - backed + * by the app's AgentStore, fsAgentStore, reading `.loushy/agents/.yaml`. + */ + loadSpec: (agentId: string) => Promise; + /** Persists a run's spec to disk (fsAgentStore.save) so it survives a server restart. */ + saveSpec?: (agentId: string, spec: AgentSpec) => Promise; +} + +export class AgentNotFoundError extends Error {} +export class AlreadyRunningError extends Error {} +export class NoActiveRunError extends Error {} + +export class RunManager extends EventEmitter { + private readonly entries = new Map(); + + constructor(private readonly opts: RunManagerOptions) { + super(); + this.setMaxListeners(0); + } + + status(agentId: string): AgentRunStatusPayload { + const entry = this.entries.get(agentId); + if (!entry) { + return { agentId, status: 'idle', updatedAt: new Date().toISOString() }; + } + return { + agentId, + status: entry.status, + sessionId: entry.sessionId, + reason: entry.reason, + pendingApproval: entry.pendingApproval, + error: entry.error, + resultText: entry.resultText, + updatedAt: entry.updatedAt, + }; + } + + private setEntry(agentId: string, patch: Partial, base?: RunEntry): RunEntry { + const prev = base ?? this.entries.get(agentId) ?? { + status: 'idle' as RunStatus, + sessionId: agentId, + updatedAt: new Date().toISOString(), + }; + const next: RunEntry = { ...prev, ...patch, updatedAt: new Date().toISOString() }; + this.entries.set(agentId, next); + this.emit('status', this.status(agentId)); + return next; + } + + private emitEvent(agentId: string, event: ExecutionEvent): void { + this.emit('event', agentId, { + type: event.type, + timestamp: event.timestamp, + text: event.text, + toolCall: event.toolCall, + toolResult: event.toolResult, + finishReason: event.finishReason, + error: event.error ? { message: event.error.message } : undefined, + }); + } + + /** + * Starts (or resumes) a run for `agentId`. + * + * `sessionId` is always just `agentId` - stable across every run() call + * for this agent id, including a run() that follows a stop(). This is + * what makes Stop-then-Run resume from the last checkpoint (N3) instead + * of starting over: AgentExecutor.execute() (see its doc comments) + * rehydrates from `checkpointStore.load(sessionId)` when a checkpoint + * exists under that id, and only clears it on a *successful* terminal + * completion - never when the run was aborted (see abortableProvider.ts). + * So after a stop(), the checkpoint from the last completed tool result + * is still there, and the very next run() call for this agentId picks it + * back up automatically. + * + * One consequence worth calling out: when a checkpoint exists, + * AgentExecutor.execute() ignores `input` entirely and continues the + * rehydrated conversation - this mirrors resumeAfterApproval()'s + * documented behavior for the approval-gate case and is intentional: a + * resumed run is a continuation of the same paused conversation, not a + * new question. `input` is only actually used to seed a genuinely fresh + * run (no prior checkpoint). + */ + async run(agentId: string, input: string, spec?: AgentSpec): Promise { + const existing = this.entries.get(agentId); + if (existing && existing.status === 'running') { + throw new AlreadyRunningError(`Agent '${agentId}' is already running`); + } + + const resolvedSpec = spec ?? existing?.lastSpec ?? (await this.opts.loadSpec(agentId)); + if (!resolvedSpec) { + throw new AgentNotFoundError(`No agent spec for id '${agentId}' (none supplied and none saved)`); + } + if (spec && this.opts.saveSpec) { + await this.opts.saveSpec(agentId, spec); + } + + // Build the agent/provider/toolRegistry BEFORE flipping status to + // 'running' - resolveSpecProvider()/resolveSpecTool() (inside + // buildAgentFromSpec) throw synchronously for a bad provider type or + // unknown tool name, and if that happened after setEntry('running') the + // entry would be stuck reporting 'running' forever (the .catch() chain + // that would otherwise flip it to 'error' is never reached, since the + // throw happens before AgentExecutor.execute() is even called). + const { agent, provider, toolRegistry } = buildAgentFromSpec(resolvedSpec, agentId); + + const sessionId = agentId; + const controller = new AbortController(); + + const entry = this.setEntry( + agentId, + { + status: 'running', + sessionId, + controller, + error: undefined, + reason: undefined, + pendingApproval: undefined, + resultText: undefined, + lastSpec: resolvedSpec, + }, + existing + ); + + const abortableProvider = withAbortSignal(provider, controller.signal); + + AgentExecutor.execute({ + agent, + input, + provider: abortableProvider, + toolRegistry, + sessionId, + checkpointStore: this.opts.checkpointStore, + approvalStore: this.opts.approvalStore, + onEvent: (event) => this.emitEvent(agentId, event), + }) + .then((result) => this.handleRunSettled(agentId, result)) + .catch((error: unknown) => this.handleRunFailed(agentId, error)); + + void entry; // status already broadcast by setEntry() above + } + + /** Shared success-path handling for both run() and approve()'s follow-up execute() call. */ + private async handleRunSettled( + agentId: string, + result: { finishReason: string; approvalId?: string; text: string } + ): Promise { + if (result.finishReason === 'awaiting-approval' && result.approvalId) { + const record = await this.opts.approvalStore.peek(agentId, result.approvalId); + this.setEntry(agentId, { + status: 'paused', + reason: 'awaiting_approval', + pendingApproval: { + approvalId: result.approvalId, + toolName: record?.pending.toolName ?? 'unknown', + args: record?.pending.args ?? {}, + createdAt: record?.pending.createdAt ?? new Date().toISOString(), + }, + }); + return; + } + this.setEntry(agentId, { status: 'stopped', resultText: result.text, controller: undefined }); + } + + /** Shared failure-path handling for both run() and approve()'s follow-up execute() call. */ + private handleRunFailed(agentId: string, error: unknown): void { + if (error instanceof RunAbortedError) { + // User-initiated stop() - a deliberate terminal state, not an error. + this.setEntry(agentId, { status: 'stopped', controller: undefined }); + return; + } + this.setEntry(agentId, { + status: 'error', + error: (error as Error)?.message ?? String(error), + controller: undefined, + }); + } + + /** + * Requests cancellation of the in-flight run for `agentId` (see + * abortableProvider.ts for exactly what this can and can't interrupt). + * No-ops (rather than throwing) if the agent isn't currently running - + * Stop is safe to click on an already-stopped/idle agent. + */ + stop(agentId: string): void { + const entry = this.entries.get(agentId); + if (!entry || entry.status !== 'running' || !entry.controller) { + return; + } + entry.controller.abort(); + } + + /** + * Resolves a pending approval (N3's human-in-the-loop round trip) via the + * SDK's public `resumeAfterApproval()`, then continues broadcasting + * status/events for the resumed run exactly like run() does. + */ + async approve(agentId: string, approvalId: string, approved: boolean, note?: string): Promise { + const entry = this.entries.get(agentId); + if (!entry || entry.status !== 'paused' || entry.pendingApproval?.approvalId !== approvalId) { + throw new NoActiveRunError( + `Agent '${agentId}' has no pending approval '${approvalId}' to resolve` + ); + } + + const spec = entry.lastSpec ?? (await this.opts.loadSpec(agentId)); + if (!spec) throw new AgentNotFoundError(`No saved agent spec for id '${agentId}'`); + const { provider, toolRegistry } = buildAgentFromSpec(spec, agentId); + + const controller = new AbortController(); + this.setEntry(agentId, { + status: 'running', + reason: undefined, + pendingApproval: undefined, + controller, + }); + + resumeAfterApproval( + { id: approvalId, approved, note }, + this.opts.approvalStore, + toolRegistry ?? new ToolRegistry(), + withAbortSignal(provider, controller.signal), + { onEvent: (event) => this.emitEvent(agentId, event) }, + this.opts.checkpointStore + ) + .then((result) => this.handleRunSettled(agentId, result)) + .catch((error: unknown) => this.handleRunFailed(agentId, error)); + } +} diff --git a/apps/agent-forge/server/tsconfig.json b/apps/agent-forge/server/tsconfig.json new file mode 100644 index 000000000..05b8915d7 --- /dev/null +++ b/apps/agent-forge/server/tsconfig.json @@ -0,0 +1,25 @@ +{ + "compilerOptions": { + "target": "ES2022", + "lib": ["ES2022"], + "module": "ESNext", + "moduleResolution": "bundler", + "declaration": false, + "sourceMap": true, + "composite": false, + "incremental": false, + "skipLibCheck": true, + "strict": true, + "esModuleInterop": true, + "resolveJsonModule": true, + "isolatedModules": true, + "noEmit": true, + "noUnusedLocals": true, + "noUnusedParameters": true, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": true, + "types": ["node", "vitest/globals"] + }, + "include": ["**/*.ts"], + "exclude": ["node_modules", "dist"] +} diff --git a/apps/agent-forge/server/types.ts b/apps/agent-forge/server/types.ts new file mode 100644 index 000000000..dc3a1ec8b --- /dev/null +++ b/apps/agent-forge/server/types.ts @@ -0,0 +1,77 @@ +/** + * LOU-N runtime control server - shared types. + * + * The server holds an in-memory registry mapping agent IDs to live + * `AgentExecutor.execute()` invocations (see runRegistry.ts). This file + * defines the wire shape of the status/events the server reports back to + * the app over `GET /agents/:id/status` and `WS /agents/:id/stream`. + */ + +/** + * Status pill states the app's agent list / topbar render (LOU-L/M's + * mockup: running/stopped/error/paused). + */ +export type RunStatus = 'idle' | 'running' | 'stopped' | 'error' | 'paused'; + +/** + * A tool call an in-progress run is paused on, awaiting a human decision + * via `POST /agents/:id/approve`. Mirrors the subset of + * `PendingApproval` (src/execution/ApprovalGate.ts) the UI needs to render + * an inline approve/reject card, without leaking the full ExecutionSnapshot. + */ +export interface PendingApprovalInfo { + approvalId: string; + toolName: string; + args: Record; + createdAt: string; +} + +/** + * The full status record the server keeps per agent ID and reports back to + * the client (REST snapshot + WS push). + */ +export interface AgentRunStatusPayload { + agentId: string; + status: RunStatus; + sessionId?: string; + /** Set when status === 'paused' because of a pending tool approval. */ + reason?: 'awaiting_approval'; + pendingApproval?: PendingApprovalInfo; + /** Set when status === 'error'. */ + error?: string; + /** Final assistant text, once a run completes successfully. */ + resultText?: string; + updatedAt: string; +} + +/** + * A single WS message pushed to `WS /agents/:id/stream` subscribers. + * `type: 'status'` carries the full AgentRunStatusPayload (sent on every + * status transition, and once immediately on connect). `type: 'event'` + * forwards a raw AgentExecutor ExecutionEvent (start/text-complete/tool-call/ + * tool-result/finish/error) for lightweight visibility into an in-progress + * run - the full log/trace UI is LOU-O's job, this is best-effort forwarding + * only, not a durable log. + */ +export type StreamMessage = + | { type: 'status'; payload: AgentRunStatusPayload } + | { type: 'event'; payload: Record }; + +/** + * Every `:id`/agentId this server touches ends up interpolated into a + * filesystem path (`.loushy/agents/.yaml`, `.loushy/agents//checkpoints/**`, + * `.loushy/agents//approvals/**` - see fsAgentStore.ts, checkpointStore.ts, + * approvalStore.ts) with no further sanitization there. Express route + * params are URL-decoded before `req.params.id` is populated, so a + * percent-encoded `..%2F..%2Fsomewhere` arrives as a plain string containing + * `/` and `..` segments - path.join() will happily walk it outside + * `.loushy/agents/`. Restrict every accepted id up front (here, and in + * wsServer.ts's WS upgrade handler, which parses `:id` itself rather than + * going through Express routing) to a safe, single-path-segment token + * instead of trying to sanitize/escape it later in each store. + */ +const AGENT_ID_RE = /^[A-Za-z0-9](?:[A-Za-z0-9._-]{0,127})$/; + +export function isValidAgentId(id: string): boolean { + return typeof id === 'string' && AGENT_ID_RE.test(id) && !id.includes('..'); +} diff --git a/apps/agent-forge/server/vitest.config.ts b/apps/agent-forge/server/vitest.config.ts new file mode 100644 index 000000000..0759b57ba --- /dev/null +++ b/apps/agent-forge/server/vitest.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from 'vitest/config'; + +// Separate from apps/agent-forge/vite.config.ts (which is jsdom-scoped to +// `src/**`, the browser app) - the LOU-N server is plain Node, and +// supertest/express/ws need real Node globals, not a DOM shim. +export default defineConfig({ + test: { + globals: true, + environment: 'node', + include: ['server/**/*.{test,spec}.ts'], + }, +}); diff --git a/apps/agent-forge/server/wsServer.ts b/apps/agent-forge/server/wsServer.ts new file mode 100644 index 000000000..1ab9c73cc --- /dev/null +++ b/apps/agent-forge/server/wsServer.ts @@ -0,0 +1,81 @@ +/** + * `WS /agents/:id/stream` (N1/N2) - pushes `StreamMessage`s + * (types.ts) to subscribers of one agent's run: a `status` message on every + * RunManager status transition (immediately on connect, too, so a client + * that connects mid-run sees the current state without waiting for the + * next transition) plus best-effort `event` messages forwarding raw + * AgentExecutor ExecutionEvents for lightweight in-progress visibility. + * + * Attached to the same underlying `http.Server` the Express app listens on + * (see index.ts) via a manual `upgrade` handler, since matching a path + * with a `:id` param needs a tiny bit of parsing `ws` doesn't do for you. + */ +import type { Server as HttpServer, IncomingMessage } from 'node:http'; +import type { Socket } from 'node:net'; +import { WebSocketServer, WebSocket } from 'ws'; +import type { RunManager } from './runRegistry'; +import { isValidAgentId, type StreamMessage } from './types'; + +const STREAM_PATH_RE = /^\/agents\/([^/]+)\/stream$/; + +export function attachWebSocketServer(server: HttpServer, runManager: RunManager): WebSocketServer { + const wss = new WebSocketServer({ noServer: true }); + // agentId -> connected sockets, so status/event broadcasts only fan out + // to clients watching that one agent's run. + const subscribers = new Map>(); + + function send(ws: WebSocket, message: StreamMessage): void { + if (ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify(message)); + } + } + + runManager.on('status', (payload) => { + const sockets = subscribers.get(payload.agentId); + if (!sockets) return; + for (const ws of sockets) send(ws, { type: 'status', payload }); + }); + + runManager.on('event', (agentId: string, payload: Record) => { + const sockets = subscribers.get(agentId); + if (!sockets) return; + for (const ws of sockets) send(ws, { type: 'event', payload }); + }); + + server.on('upgrade', (req: IncomingMessage, socket: Socket, head: Buffer) => { + const url = new URL(req.url ?? '', 'http://localhost'); + const match = STREAM_PATH_RE.exec(url.pathname); + if (!match) { + socket.destroy(); + return; + } + const agentId = decodeURIComponent(match[1]); + // Same path-traversal boundary as app.ts's `/agents/:id` middleware - + // this handler parses `:id` itself (see the module doc comment) rather + // than going through Express routing, so it needs its own check before + // `agentId` is used as a registry/status key alongside the HTTP routes. + if (!isValidAgentId(agentId)) { + socket.destroy(); + return; + } + wss.handleUpgrade(req, socket, head, (ws) => { + let sockets = subscribers.get(agentId); + if (!sockets) { + sockets = new Set(); + subscribers.set(agentId, sockets); + } + sockets.add(ws); + + // Send current status immediately so a client connecting mid-run (or + // after a pause) doesn't have to wait for the next transition. + send(ws, { type: 'status', payload: runManager.status(agentId) }); + + ws.on('close', () => { + sockets!.delete(ws); + if (sockets!.size === 0) subscribers.delete(agentId); + }); + }); + }); + + return wss; +} diff --git a/apps/agent-forge/src/components/LeftRail.tsx b/apps/agent-forge/src/components/LeftRail.tsx index c324cb55c..5a655f48f 100644 --- a/apps/agent-forge/src/components/LeftRail.tsx +++ b/apps/agent-forge/src/components/LeftRail.tsx @@ -34,8 +34,17 @@ const NODE_PALETTE: { section: string; items: { label: string; color: string; no * dragging in a node kind the graph model (and graphToSpec/validateGraph) * doesn't understand. */ +const STATUS_LABEL: Record = { + idle: 'idle', + running: 'running', + stopped: 'stopped', + error: 'error', + paused: 'awaiting approval', +}; + export function LeftRail() { - const { railTab, setRailTab, agents, agentId, graph, spec, switchAgent, createAgent } = useAppState(); + const { railTab, setRailTab, agents, agentId, graph, spec, switchAgent, createAgent, agentStatuses } = + useAppState(); const [creating, setCreating] = useState(false); const [newName, setNewName] = useState(''); const [newTemplate, setNewTemplate] = useState('blank'); @@ -85,27 +94,35 @@ export function LeftRail() { )} - {agents.map((entry) => ( -
void switchAgent(entry.id)} - onKeyDown={(e) => e.key === 'Enter' && void switchAgent(entry.id)} - > -
- {entry.id} - - - {entry.id === agentId ? 'active' : 'idle'} - -
-
- {entry.spec.provider.model} · {entry.spec.tools?.length ?? 0} tools + {agents.map((entry) => { + // LOU-N: real run status pushed over WS (see AppState's + // agentStatuses), replacing LOU-L/M's "active"/"idle" + // placeholder derived only from which agent is loaded in the + // canvas. An agent can be 'running' in the background even + // while a different agent is selected in the canvas. + const status = agentStatuses[entry.id]?.status ?? 'idle'; + return ( +
void switchAgent(entry.id)} + onKeyDown={(e) => e.key === 'Enter' && void switchAgent(entry.id)} + > +
+ {entry.id} + + + {STATUS_LABEL[status] ?? status} + +
+
+ {entry.spec.provider.model} · {entry.spec.tools?.length ?? 0} tools +
-
- ))} + ); + })} {creating ? (
diff --git a/apps/agent-forge/src/components/Topbar.tsx b/apps/agent-forge/src/components/Topbar.tsx index a18012447..8dda36914 100644 --- a/apps/agent-forge/src/components/Topbar.tsx +++ b/apps/agent-forge/src/components/Topbar.tsx @@ -1,10 +1,24 @@ -import { useRef } from 'react'; +import { useRef, useState } from 'react'; import { useAppState } from '../state/AppState'; import { downloadSpec, importSpecFile } from '../persistence/importExport'; +import { RuntimeApiError } from '../runtime/runtimeClient'; + +const STATUS_LABEL: Record = { + idle: 'idle', + running: 'running', + stopped: 'stopped', + error: 'error', + paused: 'awaiting approval', +}; export function Topbar() { - const { spec, setSpec, save, dirty, agentId } = useAppState(); + const { spec, setSpec, save, dirty, agentId, runStatus, runAgent, stopAgent, approveAgent } = useAppState(); const fileInputRef = useRef(null); + const [actionError, setActionError] = useState(undefined); + + const status = runStatus?.status ?? 'idle'; + const isRunning = status === 'running'; + const isPaused = status === 'paused' && !!runStatus?.pendingApproval; function handleExport() { downloadSpec(spec, `${spec.name || agentId}.yaml`); @@ -22,6 +36,37 @@ export function Topbar() { setSpec(() => imported); } + async function handleRun() { + setActionError(undefined); + try { + // A blank input is fine when resuming from a checkpoint (see + // runRegistry.ts's run() doc comment - AgentExecutor ignores `input` + // once a checkpoint exists), so this always sends *some* string + // rather than blocking Run on an empty prompt. + await runAgent('Run the agent.'); + } catch (error) { + setActionError(error instanceof RuntimeApiError ? error.message : (error as Error).message); + } + } + + async function handleStop() { + setActionError(undefined); + try { + await stopAgent(); + } catch (error) { + setActionError(error instanceof RuntimeApiError ? error.message : (error as Error).message); + } + } + + async function handleApprove(approved: boolean) { + setActionError(undefined); + try { + await approveAgent(approved); + } catch (error) { + setActionError(error instanceof RuntimeApiError ? error.message : (error as Error).message); + } + } + return (
@@ -40,19 +85,48 @@ export function Topbar() { {dirty && }
+ + {isPaused && ( +
+ + Approve {runStatus?.pendingApproval?.toolName}? + + + +
+ )} + + {status === 'error' && runStatus?.error && ( + + {runStatus.error} + + )} + {actionError && ( + + {actionError} + + )} + + + + {STATUS_LABEL[status] ?? status} + +
local · {spec.provider.type} provider
- {/* Debug/Stop/Run are chrome-only placeholders here - wiring them to a - real running agent needs LOU-N's runtime control server and - LOU-O's debug console, neither of which exist yet in this epic. */} + {/* Debug is a chrome-only placeholder still - the full log/trace UI is LOU-O's job. */} - -