Feature: start team run via API and stream live task state to dashboard

apps/server/src/app.ts

@@ -1,6 +1,7 @@
 import Fastify, { type FastifyInstance } from 'fastify'
 import { eventsRoutes } from './routes/events.js'
 import { healthRoutes } from './routes/health.js'
+import { runsRoutes } from './routes/runs.js'
 
 export type BuildAppOptions = {
   readonly logger?: boolean
@@ -10,5 +11,6 @@ export async function buildApp(options: BuildAppOptions = {}): Promise<FastifyIn
   const app = Fastify({ logger: options.logger ?? false })
   await app.register(healthRoutes)
   await app.register(eventsRoutes)
+  await app.register(runsRoutes)
   return app
 }

apps/server/src/events/hub.ts

@@ -1,8 +1,10 @@
 import type { OrchestratorEvent } from '@open-multi-agent/core'
+import type { RunSnapshot } from '../runs/types.js'
 
 export type ForgeEvent =
   | { readonly type: 'connected'; readonly data: { readonly ok: true } }
   | { readonly type: 'progress'; readonly data: OrchestratorEvent }
+  | { readonly type: 'run_snapshot'; readonly data: RunSnapshot }
 
 type Listener = (event: ForgeEvent) => void
 
@@ -19,6 +21,10 @@ export class EventHub {
     this.publish({ type: 'progress', data: event })
   }
 
+  publishSnapshot(snapshot: RunSnapshot): void {
+    this.publish({ type: 'run_snapshot', data: snapshot })
+  }
+
   publish(event: ForgeEvent): void {
     for (const listener of this.listeners) {
       listener(event)

apps/server/src/index.ts

@@ -1,3 +1,4 @@
+import './runs/demo-team.js'
 import { buildApp } from './app.js'
 
 const PORT = Number(process.env.PORT ?? 3001)

apps/server/src/oma.ts

@@ -1,7 +1,17 @@
 import { OpenMultiAgent } from '@open-multi-agent/core'
 import { eventHub } from './events/hub.js'
+import { currentRun } from './runs/state.js'
 
 /** Shared OMA Core orchestrator for the Forge local API. */
 export const oma = new OpenMultiAgent({
-  onProgress: (event) => eventHub.publishProgress(event),
+  onProgress: (event) => {
+    currentRun.applyProgress(event)
+    eventHub.publishProgress(event)
+    eventHub.publishSnapshot(currentRun.toSnapshot())
+  },
+  onPlanReady: async (tasks) => {
+    currentRun.setPlan(tasks)
+    eventHub.publishSnapshot(currentRun.toSnapshot())
+    return true
+  },
 })

apps/server/src/routes/runs.ts

@@ -0,0 +1,31 @@
+import type { FastifyPluginAsync } from 'fastify'
+import { currentRun } from '../runs/state.js'
+import { DEFAULT_RUN_GOAL, startRun } from '../runs/service.js'
+
+type StartRunBody = {
+  readonly goal?: string
+}
+
+export const runsRoutes: FastifyPluginAsync = async (fastify) => {
+  fastify.get('/api/runs/current', async () => currentRun.toSnapshot())
+
+  fastify.post<{ Body: StartRunBody }>('/api/runs', async (request, reply) => {
+    const goal = request.body?.goal
+    if (goal !== undefined && typeof goal !== 'string') {
+      return reply.status(400).send({ ok: false, error: 'invalid_goal' })
+    }
+    if (goal !== undefined && goal.trim().length === 0) {
+      return reply.status(400).send({ ok: false, error: 'empty_goal' })
+    }
+
+    const result = await startRun({ goal })
+    if (!result.ok) {
+      return reply.status(409).send({ ok: false, error: result.error })
+    }
+
+    return reply.status(202).send({
+      ok: true,
+      goal: goal?.trim() || DEFAULT_RUN_GOAL,
+    })
+  })
+}

apps/server/src/runs/demo-team.ts

@@ -0,0 +1,24 @@
+import type { TeamConfig } from '@open-multi-agent/core'
+import { oma } from '../oma.js'
+
+export const FORGE_DEMO_TEAM_NAME = 'forge-demo'
+
+export const forgeDemoTeamConfig: TeamConfig = {
+  name: FORGE_DEMO_TEAM_NAME,
+  agents: [
+    {
+      name: 'researcher',
+      model: 'claude-opus-4-6',
+      systemPrompt: 'You research topics thoroughly and report findings concisely.',
+    },
+    {
+      name: 'engineer',
+      model: 'claude-opus-4-6',
+      systemPrompt: 'You implement technical tasks and write clear summaries.',
+    },
+  ],
+  sharedMemory: true,
+}
+
+/** Demo team registered with the shared orchestrator at startup. */
+export const forgeDemoTeam = oma.createTeam(FORGE_DEMO_TEAM_NAME, forgeDemoTeamConfig)

apps/server/src/runs/mapper.ts

@@ -0,0 +1,11 @@
+import type { Task, TaskExecutionRecord } from '@open-multi-agent/core'
+
+export function taskToRecord(task: Task): TaskExecutionRecord {
+  return {
+    id: task.id,
+    title: task.title,
+    assignee: task.assignee,
+    status: task.status,
+    dependsOn: task.dependsOn ?? [],
+  }
+}

apps/server/src/runs/service.ts

@@ -0,0 +1,47 @@
+import type { Team, TeamRunResult } from '@open-multi-agent/core'
+import { eventHub } from '../events/hub.js'
+import { oma } from '../oma.js'
+import { forgeDemoTeam } from './demo-team.js'
+import { currentRun } from './state.js'
+
+export const DEFAULT_RUN_GOAL =
+  'Outline the steps to connect OMA Forge live execution to the team-run dashboard'
+
+export type StartRunOptions = {
+  readonly goal?: string
+  readonly team?: Team
+  readonly runTeam?: (team: Team, goal: string) => Promise<TeamRunResult>
+}
+
+function publishSnapshot(): void {
+  eventHub.publishSnapshot(currentRun.toSnapshot())
+}
+
+export async function startRun(options: StartRunOptions = {}): Promise<
+  | { readonly ok: true }
+  | { readonly ok: false; readonly error: 'run_in_progress' }
+> {
+  if (currentRun.isRunning()) {
+    return { ok: false, error: 'run_in_progress' }
+  }
+
+  const goal = options.goal?.trim() || DEFAULT_RUN_GOAL
+  const team = options.team ?? forgeDemoTeam
+  const runTeam = options.runTeam ?? ((t, g) => oma.runTeam(t, g))
+
+  currentRun.reset()
+  currentRun.begin(goal)
+  publishSnapshot()
+
+  void runTeam(team, goal)
+    .then((result) => {
+      currentRun.finish(result)
+      publishSnapshot()
+    })
+    .catch(() => {
+      currentRun.fail()
+      publishSnapshot()
+    })
+
+  return { ok: true }
+}

apps/server/src/runs/state.ts

@@ -0,0 +1,80 @@
+import type {
+  OrchestratorEvent,
+  Task,
+  TaskExecutionRecord,
+  TeamRunResult,
+} from '@open-multi-agent/core'
+import { taskToRecord } from './mapper.js'
+import type { RunSnapshot, RunStatus } from './types.js'
+
+export class CurrentRun {
+  private status: RunStatus = 'idle'
+  private goal: string | undefined
+  private tasks: TaskExecutionRecord[] = []
+
+  reset(): void {
+    this.status = 'idle'
+    this.goal = undefined
+    this.tasks = []
+  }
+
+  isRunning(): boolean {
+    return this.status === 'running'
+  }
+
+  begin(goal: string): void {
+    this.goal = goal
+    this.tasks = []
+    this.status = 'running'
+  }
+
+  setPlan(tasks: readonly Task[]): void {
+    this.tasks = tasks.map(taskToRecord)
+  }
+
+  applyProgress(event: OrchestratorEvent): void {
+    if (!event.task) return
+
+    this.tasks = this.tasks.map((task) => {
+      if (task.id !== event.task) return task
+
+      switch (event.type) {
+        case 'task_start':
+        case 'task_retry':
+          return { ...task, status: 'in_progress' }
+        case 'task_complete':
+          return { ...task, status: 'completed' }
+        case 'task_skipped':
+          return { ...task, status: 'skipped' }
+        case 'error':
+          return { ...task, status: 'failed' }
+        default:
+          return task
+      }
+    })
+  }
+
+  finish(result: TeamRunResult): void {
+    this.status = result.success ? 'completed' : 'failed'
+    if (result.goal !== undefined) {
+      this.goal = result.goal
+    }
+    if (result.tasks !== undefined) {
+      this.tasks = [...result.tasks]
+    }
+  }
+
+  fail(): void {
+    this.status = 'failed'
+  }
+
+  toSnapshot(): RunSnapshot {
+    return {
+      status: this.status,
+      goal: this.goal,
+      tasks: this.tasks,
+    }
+  }
+}
+
+export const currentRun = new CurrentRun()

apps/server/src/runs/types.ts

@@ -0,0 +1,9 @@
+import type { TaskExecutionRecord } from '@open-multi-agent/core'
+
+export type RunStatus = 'idle' | 'running' | 'completed' | 'failed'
+
+export type RunSnapshot = {
+  readonly status: RunStatus
+  readonly goal?: string
+  readonly tasks: readonly TaskExecutionRecord[]
+}

apps/server/tests/events-sse.test.ts

@@ -51,4 +51,27 @@ describe('GET /api/events (SSE)', () => {
 
     await reader.cancel()
   })
+
+  it('forwards run snapshot events to subscribers', async () => {
+    const response = await fetch(`${baseUrl}/api/events`)
+    const reader = response.body!.getReader()
+    const decoder = new TextDecoder()
+
+    await reader.read()
+
+    eventHub.publishSnapshot({
+      status: 'running',
+      goal: 'SSE goal',
+      tasks: [{ id: 't1', title: 'Task', status: 'in_progress', dependsOn: [] }],
+    })
+
+    const { value } = await reader.read()
+    const chunk = decoder.decode(value)
+
+    expect(chunk).toContain('"type":"run_snapshot"')
+    expect(chunk).toContain('SSE goal')
+    expect(chunk).toContain('in_progress')
+
+    await reader.cancel()
+  })
 })

apps/server/tests/runs-service.test.ts

@@ -0,0 +1,47 @@
+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'
+
+describe('startRun', () => {
+  afterEach(() => {
+    currentRun.reset()
+    vi.restoreAllMocks()
+  })
+
+  it('returns run_in_progress when already running', async () => {
+    currentRun.begin('Busy')
+
+    const result = await startRun({ goal: 'Another' })
+
+    expect(result).toEqual({ ok: false, error: 'run_in_progress' })
+  })
+
+  it('publishes snapshots from running to completed', async () => {
+    const snapshots: ReturnType<typeof currentRun.toSnapshot>[] = []
+    vi.spyOn(eventHub, 'publishSnapshot').mockImplementation((snapshot) => {
+      snapshots.push(snapshot)
+    })
+
+    const runTeam = vi.fn().mockResolvedValue({
+      success: true,
+      goal: 'Done',
+      tasks: [{ id: 'a', title: 'A', status: 'completed', dependsOn: [] }],
+      agentResults: new Map(),
+      totalTokenUsage: { input_tokens: 0, output_tokens: 0 },
+    })
+
+    const result = await startRun({
+      goal: 'Live test',
+      runTeam,
+      team: {} as import('@open-multi-agent/core').Team,
+    })
+
+    expect(result).toEqual({ ok: true })
+    await vi.waitFor(() => expect(runTeam).toHaveBeenCalled())
+    await vi.waitFor(() => expect(snapshots.at(-1)?.status).toBe('completed'))
+
+    expect(snapshots[0]?.status).toBe('running')
+    expect(snapshots.at(-1)?.goal).toBe('Done')
+  })
+})