diff --git a/apps/server/src/oma.ts b/apps/server/src/oma.ts index 91ae8ff..149f7a8 100644 --- a/apps/server/src/oma.ts +++ b/apps/server/src/oma.ts @@ -5,33 +5,50 @@ import { formatStreamEvent, formatTraceEvent, } from './runs/format-trace.js' -import { currentRun } from './runs/state.js' +import { runRegistry } from './runs/registry.js' import type { ForgeTraceLine } from '@oma-forge/shared' -import { traceLog } from './runs/trace-log.js' -function publishTraceLine(entry: ForgeTraceLine | null): void { +function publishTraceLine(session: import('./runs/session.js').RunSession, entry: ForgeTraceLine | null): void { if (!entry) return - traceLog.append(entry) + session.appendTrace(entry) eventHub.publishTraceLine(entry) } +function getActiveSession() { + const session = runRegistry.getActive() + if (!session?.isRunning()) return null + return session +} + /** Shared OMA Core orchestrator for the Forge local API. */ export const oma = new OpenMultiAgent({ onProgress: (event) => { - publishTraceLine(formatProgressEvent(event)) - currentRun.applyProgress(event) + const session = getActiveSession() + if (!session) return + + publishTraceLine(session, formatProgressEvent(session.id, event)) + session.applyProgress(event) eventHub.publishProgress(event) - eventHub.publishSnapshot(currentRun.toSnapshot()) + eventHub.publishSnapshot(session.toSnapshot()) }, onPlanReady: async (tasks) => { - currentRun.setPlan(tasks) - eventHub.publishSnapshot(currentRun.toSnapshot()) + const session = getActiveSession() + if (!session) return true + + session.setPlan(tasks) + eventHub.publishSnapshot(session.toSnapshot()) return true }, onTrace: (event) => { - publishTraceLine(formatTraceEvent(event)) + const session = getActiveSession() + if (!session) return + + publishTraceLine(session, formatTraceEvent(session.id, event)) }, onAgentStream: (agentName, event) => { - publishTraceLine(formatStreamEvent(agentName, event)) + const session = getActiveSession() + if (!session) return + + publishTraceLine(session, formatStreamEvent(session.id, agentName, event)) }, }) diff --git a/apps/server/src/routes/runs.ts b/apps/server/src/routes/runs.ts index 8978822..def422e 100644 --- a/apps/server/src/routes/runs.ts +++ b/apps/server/src/routes/runs.ts @@ -1,16 +1,39 @@ import type { FastifyPluginAsync } from 'fastify' -import { currentRun } from '../runs/state.js' -import { DEFAULT_RUN_GOAL, startRun } from '../runs/service.js' -import { traceLog } from '../runs/trace-log.js' +import { DEFAULT_RUN_GOAL, cancelRun, startRun } from '../runs/service.js' +import { runRegistry } from '../runs/registry.js' type StartRunBody = { readonly goal?: string } export const runsRoutes: FastifyPluginAsync = async (fastify) => { - fastify.get('/api/runs/current', async () => currentRun.toSnapshot()) + fastify.get('/api/runs', async () => ({ + runs: runRegistry.listSummaries(), + activeRunId: runRegistry.getActive()?.id ?? null, + })) - fastify.get('/api/runs/trace', async () => traceLog.toSnapshot()) + fastify.get('/api/runs/current', async () => runRegistry.currentSnapshot()) + + fastify.get('/api/runs/trace', async () => { + const run = runRegistry.getActive() ?? runRegistry.getMostRecent() + return run?.traceSnapshot() ?? { lines: [] } + }) + + fastify.get<{ Params: { id: string } }>('/api/runs/:id', async (request, reply) => { + const run = runRegistry.get(request.params.id) + if (!run) { + return reply.status(404).send({ ok: false, error: 'not_found' }) + } + return run.toSnapshot() + }) + + fastify.get<{ Params: { id: string } }>('/api/runs/:id/trace', async (request, reply) => { + const run = runRegistry.get(request.params.id) + if (!run) { + return reply.status(404).send({ ok: false, error: 'not_found' }) + } + return run.traceSnapshot() + }) fastify.post<{ Body: StartRunBody }>('/api/runs', async (request, reply) => { const goal = request.body?.goal @@ -23,12 +46,29 @@ export const runsRoutes: FastifyPluginAsync = async (fastify) => { const result = await startRun({ goal }) if (!result.ok) { - return reply.status(409).send({ ok: false, error: result.error }) + return reply.status(409).send({ + ok: false, + error: result.error, + activeRunId: result.activeRunId, + }) } return reply.status(202).send({ ok: true, - goal: goal?.trim() || DEFAULT_RUN_GOAL, + runId: result.runId, + goal: result.goal, }) }) + + fastify.post<{ Params: { id: string } }>( + '/api/runs/:id/cancel', + async (request, reply) => { + const result = cancelRun(request.params.id) + if (!result.ok) { + const status = result.error === 'not_found' ? 404 : 409 + return reply.status(status).send({ ok: false, error: result.error }) + } + return { ok: true } + }, + ) } diff --git a/apps/server/src/runs/format-trace.ts b/apps/server/src/runs/format-trace.ts index 834494d..0168ca2 100644 --- a/apps/server/src/runs/format-trace.ts +++ b/apps/server/src/runs/format-trace.ts @@ -2,11 +2,13 @@ import type { OrchestratorEvent, StreamEvent, TraceEvent } from '@open-multi-age import type { ForgeTraceLine, TraceLineLevel } from '@oma-forge/shared' function line( + runId: string, level: TraceLineLevel, message: string, meta?: { readonly agent?: string; readonly taskId?: string }, ): ForgeTraceLine { return { + runId, at: Date.now(), level, message, @@ -14,33 +16,37 @@ function line( } } -export function formatProgressEvent(event: OrchestratorEvent): ForgeTraceLine | null { +export function formatProgressEvent( + runId: string, + event: OrchestratorEvent, +): ForgeTraceLine | null { const taskId = event.task const agent = event.agent switch (event.type) { case 'task_start': - return line('info', `Task started: ${taskId ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'info', `Task started: ${taskId ?? 'unknown'}`, { agent, taskId }) case 'task_complete': - return line('info', `Task completed: ${taskId ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'info', `Task completed: ${taskId ?? 'unknown'}`, { agent, taskId }) case 'task_skipped': - return line('warn', `Task skipped: ${taskId ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'warn', `Task skipped: ${taskId ?? 'unknown'}`, { agent, taskId }) case 'task_retry': - return line('warn', `Task retry: ${taskId ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'warn', `Task retry: ${taskId ?? 'unknown'}`, { agent, taskId }) case 'agent_start': - return line('info', `Agent started: ${agent ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'info', `Agent started: ${agent ?? 'unknown'}`, { agent, taskId }) case 'agent_complete': - return line('info', `Agent finished: ${agent ?? 'unknown'}`, { agent, taskId }) + return line(runId, 'info', `Agent finished: ${agent ?? 'unknown'}`, { agent, taskId }) case 'budget_exceeded': - return line('error', 'Token budget exceeded', { agent, taskId }) + return line(runId, 'error', 'Token budget exceeded', { agent, taskId }) case 'error': return line( + runId, 'error', typeof event.data === 'string' ? event.data : 'Orchestrator error', { agent, taskId }, ) case 'message': - return line('info', typeof event.data === 'string' ? event.data : 'Message', { + return line(runId, 'info', typeof event.data === 'string' ? event.data : 'Message', { agent, taskId, }) @@ -49,36 +55,41 @@ export function formatProgressEvent(event: OrchestratorEvent): ForgeTraceLine | } } -export function formatTraceEvent(event: TraceEvent): ForgeTraceLine | null { +export function formatTraceEvent(runId: string, event: TraceEvent): ForgeTraceLine | null { const meta = { agent: event.agent, taskId: event.taskId } switch (event.type) { case 'llm_call': return line( + runId, 'info', `LLM ${event.phase ?? 'turn'} #${event.turn} — ${event.tokens.input_tokens} in / ${event.tokens.output_tokens} out`, meta, ) case 'tool_call': return line( + runId, event.isError ? 'error' : 'info', `Tool ${event.tool}${event.isError ? ' (failed)' : ''}`, meta, ) case 'task': return line( + runId, event.success ? 'info' : 'error', `Task "${event.taskTitle}" ${event.success ? 'done' : 'failed'} (${event.retries} retries)`, { ...meta, taskId: event.taskId }, ) case 'agent': return line( + runId, 'info', `Agent run done — ${event.turns} turns, ${event.toolCalls} tool calls`, meta, ) case 'plan_ready': return line( + runId, 'info', `Plan ready: ${event.taskCount} tasks (${event.approved ? 'approved' : 'rejected'})`, meta, @@ -91,22 +102,23 @@ export function formatTraceEvent(event: TraceEvent): ForgeTraceLine | null { } export function formatStreamEvent( + runId: string, agent: string, event: StreamEvent, taskId?: string, ): ForgeTraceLine | null { if (event.type === 'text' && typeof event.data === 'string' && event.data.length > 0) { - return line('stream', event.data, { agent, taskId }) + return line(runId, 'stream', event.data, { agent, taskId }) } if (event.type === 'tool_use') { const data = event.data as { name?: string } | undefined const name = data?.name ?? 'tool' - return line('info', `Streaming tool: ${name}`, { agent, taskId }) + return line(runId, 'info', `Streaming tool: ${name}`, { agent, taskId }) } if (event.type === 'error') { const message = event.data instanceof Error ? event.data.message : 'Agent stream error' - return line('error', message, { agent, taskId }) + return line(runId, 'error', message, { agent, taskId }) } return null } diff --git a/apps/server/src/runs/registry.ts b/apps/server/src/runs/registry.ts new file mode 100644 index 0000000..4be606f --- /dev/null +++ b/apps/server/src/runs/registry.ts @@ -0,0 +1,94 @@ +import type { RunSnapshot, RunSummary } from '@oma-forge/shared' +import { RunSession } from './session.js' + +const MAX_HISTORY = 100 + +export type CreateRunResult = + | { readonly ok: true; readonly session: RunSession } + | { readonly ok: false; readonly error: 'run_in_progress'; readonly activeRunId: string } + +export type CancelRunResult = + | { readonly ok: true } + | { readonly ok: false; readonly error: 'not_found' | 'not_running' } + +export class RunRegistry { + private readonly runs = new Map() + private readonly order: string[] = [] + private activeId: string | null = null + + reset(): void { + for (const session of this.runs.values()) { + if (session.isRunning()) session.cancel() + } + this.runs.clear() + this.order.length = 0 + this.activeId = null + } + + create(goal: string): CreateRunResult { + const active = this.getActive() + if (active?.isRunning()) { + return { ok: false, error: 'run_in_progress', activeRunId: active.id } + } + + const session = new RunSession(crypto.randomUUID(), goal) + this.runs.set(session.id, session) + this.order.unshift(session.id) + this.activeId = session.id + this.trimHistory() + return { ok: true, session } + } + + get(id: string): RunSession | undefined { + return this.runs.get(id) + } + + getActive(): RunSession | null { + if (!this.activeId) return null + return this.runs.get(this.activeId) ?? null + } + + getMostRecent(): RunSession | null { + const id = this.order[0] + return id ? (this.runs.get(id) ?? null) : null + } + + /** Snapshot for “current” view: active run, else most recent, else idle. */ + currentSnapshot(): RunSnapshot { + const run = this.getActive() ?? this.getMostRecent() + if (!run) { + return { status: 'idle', tasks: [] } + } + return run.toSnapshot() + } + + listSummaries(): readonly RunSummary[] { + return this.order + .map((id) => this.runs.get(id)) + .filter((run): run is RunSession => run !== undefined) + .map((run) => run.toSummary()) + } + + cancel(id: string): CancelRunResult { + const session = this.runs.get(id) + if (!session) return { ok: false, error: 'not_found' } + if (!session.isRunning()) return { ok: false, error: 'not_running' } + session.cancel() + return { ok: true } + } + + private trimHistory(): void { + while (this.order.length > MAX_HISTORY) { + const removedId = this.order.pop() + if (!removedId) break + const removed = this.runs.get(removedId) + if (removed?.isRunning()) removed.cancel() + this.runs.delete(removedId) + if (this.activeId === removedId) { + this.activeId = this.order[0] ?? null + } + } + } +} + +export const runRegistry = new RunRegistry() diff --git a/apps/server/src/runs/service.ts b/apps/server/src/runs/service.ts index e776817..71a094f 100644 --- a/apps/server/src/runs/service.ts +++ b/apps/server/src/runs/service.ts @@ -3,47 +3,76 @@ import type { Team, TeamRunResult } from '@oma-forge/shared' import { eventHub } from '../events/hub.js' import { oma } from '../oma.js' import { forgeDemoTeam } from './demo-team.js' -import { currentRun } from './state.js' -import { traceLog } from './trace-log.js' +import { runRegistry } from './registry.js' +import type { RunSession } from './session.js' export { DEFAULT_RUN_GOAL } from '@oma-forge/shared' export type StartRunOptions = { readonly goal?: string readonly team?: Team - readonly runTeam?: (team: Team, goal: string) => Promise + readonly runTeam?: ( + team: Team, + goal: string, + options: { readonly abortSignal: AbortSignal }, + ) => Promise } -function publishSnapshot(): void { - eventHub.publishSnapshot(currentRun.toSnapshot()) +function publishSnapshot(session: RunSession): void { + eventHub.publishSnapshot(session.toSnapshot()) +} + +function isAbortError(error: unknown): boolean { + return ( + error instanceof Error && + (error.name === 'AbortError' || error.message.toLowerCase().includes('abort')) + ) } export async function startRun(options: StartRunOptions = {}): Promise< - | { readonly ok: true } - | { readonly ok: false; readonly error: 'run_in_progress' } + | { readonly ok: true; readonly runId: string; readonly goal: string } + | { readonly ok: false; readonly error: 'run_in_progress'; readonly activeRunId: string } > { - if (currentRun.isRunning()) { - return { ok: false, error: 'run_in_progress' } + const goal = options.goal?.trim() || DEFAULT_RUN_GOAL + const created = runRegistry.create(goal) + if (!created.ok) { + return { ok: false, error: created.error, activeRunId: created.activeRunId } } - const goal = options.goal?.trim() || DEFAULT_RUN_GOAL + const session = created.session const team = options.team ?? forgeDemoTeam - const runTeam = options.runTeam ?? ((t, g) => oma.runTeam(t, g)) + const runTeam = + options.runTeam ?? + ((t, g, opts) => oma.runTeam(t, g, { abortSignal: opts.abortSignal })) - currentRun.reset() - traceLog.clear() - currentRun.begin(goal) - publishSnapshot() + publishSnapshot(session) - void runTeam(team, goal) + void runTeam(team, goal, { abortSignal: session.abortSignal }) .then((result) => { - currentRun.finish(result) - publishSnapshot() + if (session.abortSignal.aborted) { + if (!session.isTerminal()) session.cancel() + } else { + session.finish(result) + } + publishSnapshot(session) }) - .catch(() => { - currentRun.fail() - publishSnapshot() + .catch((error: unknown) => { + if (isAbortError(error) || session.abortSignal.aborted) { + if (!session.isTerminal()) session.cancel() + } else { + session.fail() + } + publishSnapshot(session) }) - return { ok: true } + return { ok: true, runId: session.id, goal } +} + +export function cancelRun(runId: string): import('./registry.js').CancelRunResult { + const result = runRegistry.cancel(runId) + if (result.ok) { + const session = runRegistry.get(runId) + if (session) publishSnapshot(session) + } + return result } diff --git a/apps/server/src/runs/state.ts b/apps/server/src/runs/session.ts similarity index 51% rename from apps/server/src/runs/state.ts rename to apps/server/src/runs/session.ts index 966f936..04c110e 100644 --- a/apps/server/src/runs/state.ts +++ b/apps/server/src/runs/session.ts @@ -2,31 +2,43 @@ import type { OrchestratorEvent, RunSnapshot, RunStatus, + RunSummary, Task, TaskExecutionRecord, TeamRunResult, } from '@oma-forge/shared' +import { TraceLog } from './trace-log.js' import { taskToRecord } from './mapper.js' -export class CurrentRun { - private status: RunStatus = 'idle' - private goal: string | undefined +const TERMINAL_STATUSES = new Set(['completed', 'failed', 'cancelled']) + +export class RunSession { + private status: RunStatus = 'running' + private readonly trace = new TraceLog() private tasks: TaskExecutionRecord[] = [] + private finishedAt: number | undefined + + private goal: string - reset(): void { - this.status = 'idle' - this.goal = undefined - this.tasks = [] + constructor( + readonly id: string, + goal: string, + readonly startedAt: number = Date.now(), + readonly abortController: AbortController = new AbortController(), + ) { + this.goal = goal } isRunning(): boolean { return this.status === 'running' } - begin(goal: string): void { - this.goal = goal - this.tasks = [] - this.status = 'running' + isTerminal(): boolean { + return TERMINAL_STATUSES.has(this.status) + } + + get abortSignal(): AbortSignal { + return this.abortController.signal } setPlan(tasks: readonly Task[]): void { @@ -56,6 +68,7 @@ export class CurrentRun { } finish(result: TeamRunResult): void { + this.finishedAt = Date.now() this.status = result.success ? 'completed' : 'failed' if (result.goal !== undefined) { this.goal = result.goal @@ -66,16 +79,43 @@ export class CurrentRun { } fail(): void { + this.finishedAt = Date.now() this.status = 'failed' } + cancel(): void { + if (!this.isRunning()) return + this.abortController.abort() + this.finishedAt = Date.now() + this.status = 'cancelled' + } + + appendTrace(entry: import('@oma-forge/shared').ForgeTraceLine): void { + this.trace.append(entry) + } + + traceSnapshot(): import('@oma-forge/shared').TraceLogSnapshot { + return this.trace.toSnapshot() + } + toSnapshot(): RunSnapshot { return { + id: this.id, status: this.status, goal: this.goal, tasks: this.tasks, + startedAt: this.startedAt, + finishedAt: this.finishedAt, } } -} -export const currentRun = new CurrentRun() + toSummary(): RunSummary { + return { + id: this.id, + status: this.status as Exclude, + goal: this.goal, + startedAt: this.startedAt, + finishedAt: this.finishedAt, + } + } +} diff --git a/apps/server/src/runs/trace-log.ts b/apps/server/src/runs/trace-log.ts index 7f42aec..6df7f0e 100644 --- a/apps/server/src/runs/trace-log.ts +++ b/apps/server/src/runs/trace-log.ts @@ -23,5 +23,3 @@ export class TraceLog { return { lines: this.lines } } } - -export const traceLog = new TraceLog() diff --git a/apps/server/tests/format-trace.test.ts b/apps/server/tests/format-trace.test.ts index 2cbc092..28c59d5 100644 --- a/apps/server/tests/format-trace.test.ts +++ b/apps/server/tests/format-trace.test.ts @@ -7,7 +7,8 @@ import { describe('formatProgressEvent', () => { it('formats task lifecycle events', () => { - const line = formatProgressEvent({ type: 'task_start', task: 't1', agent: 'worker' }) + const line = formatProgressEvent('run-1', { type: 'task_start', task: 't1', agent: 'worker' }) + expect(line?.runId).toBe('run-1') expect(line?.message).toContain('Task started') expect(line?.taskId).toBe('t1') expect(line?.agent).toBe('worker') @@ -16,7 +17,7 @@ describe('formatProgressEvent', () => { describe('formatTraceEvent', () => { it('formats tool call traces', () => { - const line = formatTraceEvent({ + const line = formatTraceEvent('run-1', { type: 'tool_call', runId: 'r1', startMs: 0, @@ -35,7 +36,7 @@ describe('formatTraceEvent', () => { it('ignores agent_stream trace duplicates', () => { expect( - formatTraceEvent({ + formatTraceEvent('run-1', { type: 'agent_stream', runId: 'r1', startMs: 0, @@ -50,7 +51,8 @@ describe('formatTraceEvent', () => { describe('formatStreamEvent', () => { it('formats text stream deltas', () => { - const line = formatStreamEvent('writer', { type: 'text', data: 'Hello' }) + const line = formatStreamEvent('run-1', 'writer', { type: 'text', data: 'Hello' }) + expect(line?.runId).toBe('run-1') expect(line?.level).toBe('stream') expect(line?.message).toBe('Hello') expect(line?.agent).toBe('writer') diff --git a/apps/server/tests/runs-registry.test.ts b/apps/server/tests/runs-registry.test.ts new file mode 100644 index 0000000..d9d7dcd --- /dev/null +++ b/apps/server/tests/runs-registry.test.ts @@ -0,0 +1,43 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { RunRegistry } from '../src/runs/registry.js' + +describe('RunRegistry', () => { + const registry = new RunRegistry() + + afterEach(() => { + registry.reset() + }) + + it('lists runs newest first after multiple completed runs', () => { + const first = registry.create('First') + expect(first.ok).toBe(true) + if (first.ok) first.session.finish({ success: true, agentResults: new Map(), totalTokenUsage: { input_tokens: 0, output_tokens: 0 } }) + + const second = registry.create('Second') + expect(second.ok).toBe(true) + + const summaries = registry.listSummaries() + expect(summaries).toHaveLength(2) + expect(summaries[0]?.goal).toBe('Second') + expect(summaries[1]?.goal).toBe('First') + }) + + it('exposes current snapshot from most recent run when idle', () => { + const created = registry.create('Done') + expect(created.ok).toBe(true) + if (!created.ok) return + created.session.finish({ + success: true, + goal: 'Done', + tasks: [], + agentResults: new Map(), + totalTokenUsage: { input_tokens: 0, output_tokens: 0 }, + }) + + expect(registry.currentSnapshot()).toMatchObject({ + id: created.session.id, + status: 'completed', + goal: 'Done', + }) + }) +}) diff --git a/apps/server/tests/runs-service.test.ts b/apps/server/tests/runs-service.test.ts index c9b25eb..e332e1e 100644 --- a/apps/server/tests/runs-service.test.ts +++ b/apps/server/tests/runs-service.test.ts @@ -1,26 +1,31 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { eventHub } from '../src/events/hub.js' -import { startRun } from '../src/runs/service.js' -import { currentRun } from '../src/runs/state.js' -import { traceLog } from '../src/runs/trace-log.js' +import { cancelRun, startRun } from '../src/runs/service.js' +import { runRegistry } from '../src/runs/registry.js' describe('startRun', () => { afterEach(() => { - currentRun.reset() - traceLog.clear() + runRegistry.reset() vi.restoreAllMocks() }) it('returns run_in_progress when already running', async () => { - currentRun.begin('Busy') - - const result = await startRun({ goal: 'Another' }) + const first = await startRun({ + goal: 'Busy', + runTeam: () => new Promise(() => {}), + team: {} as import('@open-multi-agent/core').Team, + }) + expect(first.ok).toBe(true) - expect(result).toEqual({ ok: false, error: 'run_in_progress' }) + const second = await startRun({ goal: 'Another' }) + expect(second).toMatchObject({ ok: false, error: 'run_in_progress' }) + if (!second.ok) { + expect(second.activeRunId).toBe(first.runId) + } }) it('publishes snapshots from running to completed', async () => { - const snapshots: ReturnType[] = [] + const snapshots: import('@oma-forge/shared').RunSnapshot[] = [] vi.spyOn(eventHub, 'publishSnapshot').mockImplementation((snapshot) => { snapshots.push(snapshot) }) @@ -39,11 +44,39 @@ describe('startRun', () => { team: {} as import('@open-multi-agent/core').Team, }) - expect(result).toEqual({ ok: true }) + expect(result.ok).toBe(true) + if (result.ok) { + expect(result.runId).toBeDefined() + } await vi.waitFor(() => expect(runTeam).toHaveBeenCalled()) await vi.waitFor(() => expect(snapshots.at(-1)?.status).toBe('completed')) expect(snapshots[0]?.status).toBe('running') + expect(snapshots[0]?.id).toBe(result.ok ? result.runId : undefined) expect(snapshots.at(-1)?.goal).toBe('Done') }) + + it('marks run cancelled when abort signal fires', async () => { + const runTeam = vi.fn( + (_team, _goal, options: { abortSignal: AbortSignal }) => + new Promise((_resolve, reject) => { + options.abortSignal.addEventListener('abort', () => { + reject(Object.assign(new Error('Aborted'), { name: 'AbortError' })) + }) + }), + ) + + const started = await startRun({ + goal: 'Cancel me', + runTeam, + team: {} as import('@open-multi-agent/core').Team, + }) + expect(started.ok).toBe(true) + if (!started.ok) return + + cancelRun(started.runId) + await vi.waitFor(() => { + expect(runRegistry.get(started.runId)?.toSnapshot().status).toBe('cancelled') + }) + }) }) diff --git a/apps/server/tests/runs-state.test.ts b/apps/server/tests/runs-state.test.ts index 2fd5cbb..01dd03c 100644 --- a/apps/server/tests/runs-state.test.ts +++ b/apps/server/tests/runs-state.test.ts @@ -1,15 +1,19 @@ import { describe, expect, it } from 'vitest' -import { CurrentRun } from '../src/runs/state.js' +import { RunSession } from '../src/runs/session.js' -describe('CurrentRun', () => { - it('begins in idle with an empty snapshot', () => { - const run = new CurrentRun() - expect(run.toSnapshot()).toEqual({ status: 'idle', tasks: [] }) +describe('RunSession', () => { + it('begins in running with an empty task list', () => { + const run = new RunSession('run-1', 'Ship the feature') + expect(run.toSnapshot()).toMatchObject({ + id: 'run-1', + status: 'running', + goal: 'Ship the feature', + tasks: [], + }) }) it('maps coordinator plan tasks to records', () => { - const run = new CurrentRun() - run.begin('Ship the feature') + const run = new RunSession('run-1', 'Ship the feature') run.setPlan([ { id: 't1', @@ -30,8 +34,7 @@ describe('CurrentRun', () => { }) it('updates task status from progress events', () => { - const run = new CurrentRun() - run.begin('Run tests') + const run = new RunSession('run-1', 'Run tests') run.setPlan([ { id: 't1', @@ -51,8 +54,7 @@ describe('CurrentRun', () => { }) it('finishes with team run result tasks and metrics', () => { - const run = new CurrentRun() - run.begin('Goal') + const run = new RunSession('run-1', 'Goal') run.finish({ success: true, goal: 'Goal', @@ -77,6 +79,14 @@ describe('CurrentRun', () => { const snapshot = run.toSnapshot() expect(snapshot.status).toBe('completed') + expect(snapshot.finishedAt).toBeDefined() expect(snapshot.tasks[0]?.metrics).toBeDefined() }) + + it('cancels a running session', () => { + const run = new RunSession('run-1', 'Stop me') + run.cancel() + expect(run.toSnapshot().status).toBe('cancelled') + expect(run.abortSignal.aborted).toBe(true) + }) }) diff --git a/apps/server/tests/runs.test.ts b/apps/server/tests/runs.test.ts index 42d7b43..805aee0 100644 --- a/apps/server/tests/runs.test.ts +++ b/apps/server/tests/runs.test.ts @@ -1,25 +1,29 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { buildApp } from '../src/app.js' -import { currentRun } from '../src/runs/state.js' -import { traceLog } from '../src/runs/trace-log.js' +import { runRegistry } from '../src/runs/registry.js' import * as runService from '../src/runs/service.js' describe('runs API', () => { let app: Awaited> beforeEach(async () => { - currentRun.reset() - traceLog.clear() + runRegistry.reset() app = await buildApp() }) afterEach(async () => { vi.restoreAllMocks() - currentRun.reset() - traceLog.clear() + runRegistry.reset() await app.close() }) + it('GET /api/runs returns empty history when idle', async () => { + const response = await app.inject({ method: 'GET', url: '/api/runs' }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ runs: [], activeRunId: null }) + }) + it('GET /api/runs/trace returns empty log when idle', async () => { const response = await app.inject({ method: 'GET', url: '/api/runs/trace' }) @@ -34,8 +38,12 @@ describe('runs API', () => { expect(response.json()).toEqual({ status: 'idle', tasks: [] }) }) - it('POST /api/runs starts a run and returns 202', async () => { - vi.spyOn(runService, 'startRun').mockResolvedValue({ ok: true }) + it('POST /api/runs starts a run and returns 202 with runId', async () => { + vi.spyOn(runService, 'startRun').mockResolvedValue({ + ok: true, + runId: 'run-test-1', + goal: 'Test goal', + }) const response = await app.inject({ method: 'POST', @@ -44,7 +52,11 @@ describe('runs API', () => { }) expect(response.statusCode).toBe(202) - expect(response.json()).toEqual({ ok: true, goal: 'Test goal' }) + expect(response.json()).toEqual({ + ok: true, + runId: 'run-test-1', + goal: 'Test goal', + }) expect(runService.startRun).toHaveBeenCalledWith({ goal: 'Test goal' }) }) @@ -60,7 +72,7 @@ describe('runs API', () => { }) it('POST /api/runs returns 409 when a run is in progress', async () => { - currentRun.begin('Busy') + runRegistry.create('Busy') const response = await app.inject({ method: 'POST', @@ -69,6 +81,22 @@ describe('runs API', () => { }) expect(response.statusCode).toBe(409) - expect(response.json()).toEqual({ ok: false, error: 'run_in_progress' }) + const body = response.json() as { ok: boolean; error: string; activeRunId: string } + expect(body.ok).toBe(false) + expect(body.error).toBe('run_in_progress') + expect(body.activeRunId).toBeDefined() + }) + + it('GET /api/runs/:id returns 404 for unknown run', async () => { + const response = await app.inject({ method: 'GET', url: '/api/runs/missing' }) + expect(response.statusCode).toBe(404) + }) + + it('POST /api/runs/:id/cancel returns 404 for unknown run', async () => { + const response = await app.inject({ + method: 'POST', + url: '/api/runs/missing/cancel', + }) + expect(response.statusCode).toBe(404) }) }) diff --git a/apps/web/src/App.tsx b/apps/web/src/App.tsx index 6496575..89286fb 100644 --- a/apps/web/src/App.tsx +++ b/apps/web/src/App.tsx @@ -3,13 +3,27 @@ import { DEFAULT_RUN_GOAL } from '@oma-forge/shared' import { useForgeRun } from './hooks/useForgeRun.ts' export default function App() { - const { runStatus, result, traceLines, healthOk, startError, isStarting, startRun } = - useForgeRun() + const { + runStatus, + result, + traceLines, + runHistory, + activeRunId, + selectedRunId, + healthOk, + startError, + cancelError, + isStarting, + isCancelling, + startRun, + cancelRun, + selectRun, + } = useForgeRun() const isRunning = runStatus === 'running' return (
-
+
OMA Forge {healthOk === true ? 'API online' : healthOk === false ? 'API offline' : 'API …'} - run: {runStatus} + + run: {runStatus} + {selectedRunId ? ` · ${selectedRunId.slice(0, 8)}` : ''} + + + {runHistory.length > 0 ? ( + + ) : null}
{startError ? (

- {startError} + start: {startError} +

+ ) : null} + {cancelError ? ( +

+ cancel: {cancelError}

) : null} {result.tasks && result.tasks.length > 0 ? ( diff --git a/apps/web/src/hooks/useForgeRun.ts b/apps/web/src/hooks/useForgeRun.ts index b076705..9841b7e 100644 --- a/apps/web/src/hooks/useForgeRun.ts +++ b/apps/web/src/hooks/useForgeRun.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useState } from 'react' +import { useCallback, useEffect, useRef, useState } from 'react' import { DEFAULT_RUN_GOAL, emptyDashboardRun, @@ -8,11 +8,17 @@ import { type ForgeTraceLine, type RunSnapshot, type RunStatus, + type RunSummary, type TraceLogSnapshot, } from '@oma-forge/shared' const MAX_TRACE_LINES = 500 +type RunsListResponse = { + readonly runs: readonly RunSummary[] + readonly activeRunId: string | null +} + function appendTraceLine( lines: readonly ForgeTraceLine[], line: ForgeTraceLine, @@ -21,39 +27,77 @@ function appendTraceLine( return next.length > MAX_TRACE_LINES ? next.slice(-MAX_TRACE_LINES) : next } +async function fetchRunSnapshot(runId: string): Promise { + const res = await fetch(`/api/runs/${runId}`) + return res.ok ? ((await res.json()) as RunSnapshot) : null +} + +async function fetchRunTrace(runId: string): Promise { + const res = await fetch(`/api/runs/${runId}/trace`) + if (!res.ok) return [] + const body = (await res.json()) as TraceLogSnapshot + return body.lines ?? [] +} + export function useForgeRun() { const [runStatus, setRunStatus] = useState('idle') const [result, setResult] = useState(emptyDashboardRun) const [traceLines, setTraceLines] = useState([]) + const [runHistory, setRunHistory] = useState([]) + const [activeRunId, setActiveRunId] = useState(null) + const [selectedRunId, setSelectedRunId] = useState(null) const [healthOk, setHealthOk] = useState(null) const [startError, setStartError] = useState(null) + const [cancelError, setCancelError] = useState(null) const [isStarting, setIsStarting] = useState(false) + const [isCancelling, setIsCancelling] = useState(false) + + const selectedRunIdRef = useRef(null) + selectedRunIdRef.current = selectedRunId const applySnapshot = useCallback((snapshot: RunSnapshot) => { setRunStatus(snapshot.status) setResult(runSnapshotToDashboard(snapshot)) + if (snapshot.id) setSelectedRunId(snapshot.id) }, []) - const appendLine = useCallback((line: ForgeTraceLine) => { - setTraceLines((prev) => appendTraceLine(prev, line)) + const loadRun = useCallback(async (runId: string) => { + const [snapshot, lines] = await Promise.all([ + fetchRunSnapshot(runId), + fetchRunTrace(runId), + ]) + if (snapshot) applySnapshot(snapshot) + setTraceLines(lines) + setSelectedRunId(runId) + }, [applySnapshot]) + + const refreshRunsList = useCallback(async () => { + const res = await fetch('/api/runs') + if (!res.ok) return + const body = (await res.json()) as RunsListResponse + setRunHistory(body.runs) + setActiveRunId(body.activeRunId) + return body }, []) useEffect(() => { let cancelled = false - fetch('/api/runs/current') - .then((res) => (res.ok ? res.json() : null)) - .then((snapshot: RunSnapshot | null) => { - if (!cancelled && snapshot) applySnapshot(snapshot) - }) - .catch(() => {}) + void (async () => { + const list = await refreshRunsList() + if (cancelled) return - fetch('/api/runs/trace') - .then((res) => (res.ok ? res.json() : null)) - .then((body: TraceLogSnapshot | null) => { - if (!cancelled && body?.lines) setTraceLines(body.lines) - }) - .catch(() => {}) + const initialId = list?.activeRunId ?? list?.runs[0]?.id + if (initialId) { + await loadRun(initialId) + } else { + const currentRes = await fetch('/api/runs/current') + if (currentRes.ok) { + const snapshot = (await currentRes.json()) as RunSnapshot + if (!cancelled) applySnapshot(snapshot) + } + } + })() fetch('/api/health') .then((res) => (res.ok ? res.json() : null)) @@ -67,55 +111,116 @@ export function useForgeRun() { return () => { cancelled = true } - }, [applySnapshot]) + }, [applySnapshot, loadRun, refreshRunsList]) useEffect(() => { const source = new EventSource('/api/events') source.onmessage = (message) => { const event = parseForgeEvent(message.data) - if (event?.type === 'run_snapshot') { - applySnapshot(event.data) + if (!event) return + + const selected = selectedRunIdRef.current + + if (event.type === 'run_snapshot') { + const snapshot = event.data + void refreshRunsList() + if (!selected || snapshot.id === selected) { + applySnapshot(snapshot) + } } - if (event?.type === 'trace_line') { - appendLine(event.data) + + if (event.type === 'trace_line') { + const line = event.data + if (!selected || line.runId === selected) { + setTraceLines((prev) => appendTraceLine(prev, line)) + } } } return () => { source.close() } - }, [applySnapshot, appendLine]) + }, [applySnapshot, refreshRunsList]) - const startRun = useCallback(async (goal?: string) => { - setStartError(null) - setIsStarting(true) + const startRun = useCallback( + async (goal?: string) => { + setStartError(null) + setIsStarting(true) + try { + const response = await fetch('/api/runs', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ goal: goal ?? DEFAULT_RUN_GOAL }), + }) + const body = (await response.json()) as { + ok?: boolean + error?: string + runId?: string + } + if (!response.ok) { + setStartError(body.error ?? `request_failed_${response.status}`) + return + } + if (body.runId) { + setTraceLines([]) + setActiveRunId(body.runId) + setSelectedRunId(body.runId) + await loadRun(body.runId) + await refreshRunsList() + } + } catch { + setStartError('network_error') + } finally { + setIsStarting(false) + } + }, + [loadRun, refreshRunsList], + ) + + const cancelRun = useCallback(async () => { + const runId = activeRunId ?? selectedRunId + if (!runId) return + + setCancelError(null) + setIsCancelling(true) try { - const response = await fetch('/api/runs', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ goal: goal ?? DEFAULT_RUN_GOAL }), - }) + const response = await fetch(`/api/runs/${runId}/cancel`, { method: 'POST' }) const body = (await response.json()) as { ok?: boolean; error?: string } if (!response.ok) { - setStartError(body.error ?? `request_failed_${response.status}`) + setCancelError(body.error ?? `request_failed_${response.status}`) } else { - setTraceLines([]) + await refreshRunsList() + await loadRun(runId) } } catch { - setStartError('network_error') + setCancelError('network_error') } finally { - setIsStarting(false) + setIsCancelling(false) } - }, []) + }, [activeRunId, loadRun, refreshRunsList, selectedRunId]) + + const selectRun = useCallback( + (runId: string) => { + void loadRun(runId) + }, + [loadRun], + ) return { runStatus, result, traceLines, + runHistory, + activeRunId, + selectedRunId, healthOk, startError, + cancelError, isStarting, + isCancelling, startRun, + cancelRun, + selectRun, } } diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 513d974..da40109 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -3,7 +3,7 @@ export type { ForgeDashboardRun } from './dashboard.js' export { emptyDashboardRun, runSnapshotToDashboard } from './dashboard.js' export type { ForgeEvent } from './events.js' export { parseForgeEvent } from './events.js' -export type { RunSnapshot, RunStatus } from './run.js' +export type { RunSnapshot, RunStatus, RunSummary } from './run.js' export type { ForgeTraceLine, TraceLineLevel, diff --git a/packages/shared/src/run.ts b/packages/shared/src/run.ts index 3c01a24..f93819a 100644 --- a/packages/shared/src/run.ts +++ b/packages/shared/src/run.ts @@ -1,9 +1,26 @@ import type { TaskExecutionRecord } from '@open-multi-agent/core' -export type RunStatus = 'idle' | 'running' | 'completed' | 'failed' +export type RunStatus = + | 'idle' + | 'running' + | 'completed' + | 'failed' + | 'cancelled' export type RunSnapshot = { + readonly id?: string readonly status: RunStatus readonly goal?: string readonly tasks: readonly TaskExecutionRecord[] + readonly startedAt?: number + readonly finishedAt?: number +} + +/** Compact entry for run history lists. */ +export type RunSummary = { + readonly id: string + readonly status: Exclude + readonly goal?: string + readonly startedAt: number + readonly finishedAt?: number } diff --git a/packages/shared/src/trace.ts b/packages/shared/src/trace.ts index b5fb48a..74b2e17 100644 --- a/packages/shared/src/trace.ts +++ b/packages/shared/src/trace.ts @@ -1,6 +1,7 @@ export type TraceLineLevel = 'info' | 'warn' | 'error' | 'stream' export type ForgeTraceLine = { + readonly runId: string readonly at: number readonly level: TraceLineLevel readonly agent?: string