Feature: run registry with history, per-run trace, and cancel

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))
   },
 })

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 }
+    },
+  )
 }

apps/server/src/runs/format-trace.ts

@@ -2,45 +2,51 @@ 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,
     ...meta,
   }
 }
 
-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
 }

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<string, RunSession>()
+  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()