feat(plugins): pipeline-runner — stage events + GenUI cache integration

Plan §3.A7 + §3.A8 / spec §10.1 / §10.2 / §10.3.3. New apps/daemon/src/plugins/pipeline-runner.ts is the wiring layer between the pure pipeline scheduler and a live run's event stream: - Walks the pipeline through runPipeline(), emitting pipeline_stage_started / pipeline_stage_completed events for SSE consumers and persisting one row per iteration into run_devloop_iterations. - Pre-stage GenUI: surfaces with no trigger.stageId (e.g. the __auto_connector_<id> oauth-prompts the apply path auto-derives) are raised once up-front via requestOrReuseSurface(). On a project-tier cache hit the registry emits genui_surface_response { respondedBy: 'cache' } WITHOUT broadcasting a new request — this is the e2e-5 anchor. - Stage-bound surfaces fire on stage entry only when the surface's trigger.stageId matches. The actual stage worker is supplied by the caller; in v1 the daemon plugs in a stub that records iteration counts. Phase 4 swaps in critique-theater scoring + build-test signals without touching this public contract. Daemon tests: 1434 → 1438 (added plugins-pipeline-runner with 4 cases: event sequence, devloop convergence, e2e-5 cross-conversation cache, stage-bound surface gating). Co-authored-by: Tom Huang <1043269994@qq.com>
2026-10-27 09:51:12 +01:00 · 2026-05-09 11:29:57 +00:00
parent b9d40094b5
commit 531df0e270
3 changed files with 370 additions and 0 deletions
@@ -8,6 +8,7 @@ export * from './doctor.js';
 export * from './installer.js';
 export * from './persistence.js';
 export * from './pipeline.js';
+export * from './pipeline-runner.js';
 export * from './registry.js';
 export * from './gc.js';
 export * from './resolve-snapshot.js';
@@ -0,0 +1,109 @@
+// Pipeline runner — bridges the pure pipeline scheduler in `pipeline.ts`
+// onto a live run's SSE event stream and the `genui_surfaces` cache.
+//
+// Plan §3.A7 / spec §10.1 / §10.2. Today the agent loop owns the actual
+// stage execution; this runner is the thin wrapper that:
+//
+//   - emits `pipeline_stage_started` / `pipeline_stage_completed`
+//     events through the run service so SSE / ND-JSON consumers see the
+//     stage timeline,
+//   - persists each iteration into `run_devloop_iterations`,
+//   - asks `requestOrReuseSurface()` for the matching GenUI surface
+//     before the stage runner fires (auto-derived `__auto_connector_*`
+//     prompts always go through the cross-conversation cache so a
+//     repeat-OAuth never re-broadcasts).
+//
+// The stage runner itself is supplied by the caller. In v1 the daemon
+// passes a stub that only records iteration counts; later phases can
+// plug in critique-theater scoring, build-test signals, etc. without
+// changing the public contract.
+
+import type Database from 'better-sqlite3';
+import type {
+  AppliedPluginSnapshot,
+  GenUISurfaceSpec,
+  PluginPipeline,
+} from '@open-design/contracts';
+import { runPipeline, type PipelineEnv, type PipelineEventSink, type StageOutcomeRecord, type StageRunner } from './pipeline.js';
+import { requestOrReuseSurface } from '../genui/registry.js';
+import type { GenUIEventSink } from '../genui/events.js';
+
+type SqliteDb = Database.Database;
+
+export interface PipelineRunnerInput {
+  db:        SqliteDb;
+  runId:     string;
+  projectId: string;
+  conversationId?: string | null | undefined;
+  snapshot:  AppliedPluginSnapshot;
+  pipeline:  PluginPipeline;
+  env:       PipelineEnv;
+  // The actual stage worker. Returns signals (`critique.score`,
+  // `iterations`, …) that drive `until` evaluation. Apps inject the
+  // real LLM-driven worker; tests pass a deterministic stub.
+  runStage:  StageRunner;
+  // Pluggable event sinks. The runs service injects the SSE forwarder
+  // for both pipeline + genui events; tests pass an array.push closure.
+  emitPipeline?: PipelineEventSink | undefined;
+  emitGenui?:    GenUIEventSink | undefined;
+}
+
+// Walk the pipeline + GenUI surfaces. Returns the per-stage outcomes
+// (iteration count + convergence) so the run service can decide whether
+// the run finished cleanly. The caller already started the run; we just
+// drive its event stream.
+export async function runPipelineForRun(
+  input: PipelineRunnerInput,
+): Promise<StageOutcomeRecord[]> {
+  // Pre-stage GenUI: surfaces with no `trigger.stageId` get raised once
+  // up-front (typical example: the auto-derived `__auto_connector_*`
+  // oauth-prompts that the apply path created when a required connector
+  // was not yet connected). The cache short-circuit here is what makes
+  // e2e-5 pass: a second conversation in the same project skips the
+  // broadcast.
+  const surfaces = input.snapshot.genuiSurfaces ?? [];
+  for (const surface of surfaces) {
+    if (surface.trigger?.stageId) continue;
+    raiseSurface(input, surface);
+  }
+
+  // Wrap the user-supplied stage runner so we can intercept stage entry
+  // for stage-bound surface raising before delegating to the real
+  // worker.
+  const wrapped: StageRunner = async ({ stage, iteration, snapshot }) => {
+    if (iteration === 0) {
+      for (const surface of surfaces) {
+        if (surface.trigger?.stageId === stage.id) raiseSurface(input, surface);
+      }
+    }
+    return input.runStage({ stage, iteration, snapshot });
+  };
+
+  return runPipeline({
+    db:       input.db,
+    runId:    input.runId,
+    snapshot: input.snapshot,
+    pipeline: input.pipeline,
+    env:      input.env,
+    runStage: wrapped,
+    ...(input.emitPipeline ? { emit: input.emitPipeline } : {}),
+  });
+}
+
+function raiseSurface(input: PipelineRunnerInput, surface: GenUISurfaceSpec): void {
+  try {
+    requestOrReuseSurface(input.db, {
+      projectId:        input.projectId,
+      conversationId:   input.conversationId ?? null,
+      runId:            input.runId,
+      pluginSnapshotId: input.snapshot.snapshotId,
+      surface,
+      ...(input.emitGenui ? { emit: input.emitGenui } : {}),
+    });
+  } catch (err) {
+    // Surface lookups should never crash the pipeline. The error path is
+    // surfaced to the run as a normal SSE 'error' event by the caller.
+    // eslint-disable-next-line no-console
+    console.warn(`[plugins] failed to raise surface ${surface.id}: ${(err as Error).message}`);
+  }
+}
@@ -0,0 +1,260 @@
+// Plan §3.A7 / §3.A8 — pipeline-runner integration test.
+//
+// Exercises the wiring that joins the pure pipeline scheduler
+// (`pipeline.ts`) onto the run service via:
+//   - emitted `pipeline_stage_started` / `pipeline_stage_completed`
+//     events on each stage entry / exit
+//   - `run_devloop_iterations` rows persisted per iteration (audit log)
+//   - GenUI surface raising via `requestOrReuseSurface`, including the
+//     F8 cross-conversation cache (e2e-5 anchor): a second conversation
+//     in the same project that re-applies the same `persist='project'`
+//     surface emits a `genui_surface_response { respondedBy: 'cache' }`
+//     event without broadcasting a new request.
+
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { mkdtemp, rm } from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import Database from 'better-sqlite3';
+import type {
+  AppliedPluginSnapshot,
+  GenUISurfaceEvent,
+  GenUISurfaceSpec,
+  PluginPipelineStageEvent,
+} from '@open-design/contracts';
+import { migratePlugins } from '../src/plugins/persistence.js';
+import { createSnapshot } from '../src/plugins/snapshots.js';
+import { runPipelineForRun } from '../src/plugins/pipeline-runner.js';
+import { listIterationsForRun } from '../src/plugins/pipeline.js';
+import { respondSurface } from '../src/genui/registry.js';
+import { findPendingByRunAndSurfaceId } from '../src/genui/store.js';
+
+let db: Database.Database;
+let tmpDir: string;
+
+beforeEach(async () => {
+  tmpDir = await mkdtemp(path.join(os.tmpdir(), 'od-runner-'));
+  db = new Database(path.join(tmpDir, 'test.sqlite'));
+  db.exec(`
+    CREATE TABLE projects (id TEXT PRIMARY KEY, name TEXT);
+    CREATE TABLE conversations (id TEXT PRIMARY KEY, project_id TEXT, title TEXT);
+  `);
+  migratePlugins(db);
+  db.prepare('INSERT INTO projects (id, name) VALUES (?, ?)').run('project-1', 'Project 1');
+  db.prepare('INSERT INTO conversations (id, project_id, title) VALUES (?, ?, ?)')
+    .run('conv-A', 'project-1', 'A');
+  db.prepare('INSERT INTO conversations (id, project_id, title) VALUES (?, ?, ?)')
+    .run('conv-B', 'project-1', 'B');
+});
+
+afterEach(async () => {
+  db.close();
+  await rm(tmpDir, { recursive: true, force: true });
+});
+
+function snapshotWith(extra: Partial<AppliedPluginSnapshot>): AppliedPluginSnapshot {
+  return createSnapshot(db, {
+    projectId: 'project-1',
+    conversationId: 'conv-A',
+    runId: null,
+    pluginId: 'sample-plugin',
+    pluginVersion: '1.0.0',
+    pluginTitle: 'Sample',
+    pluginDescription: '',
+    manifestSourceDigest: 'digest-1',
+    taskKind: 'new-generation' as const,
+    inputs: {},
+    resolvedContext: { items: [] },
+    pipeline: extra.pipeline,
+    genuiSurfaces: extra.genuiSurfaces,
+    capabilitiesGranted: ['prompt:inject'],
+    capabilitiesRequired: ['prompt:inject'],
+    assetsStaged: [],
+    connectorsRequired: [],
+    connectorsResolved: [],
+    mcpServers: [],
+    query: '',
+  });
+}
+
+describe('pipeline-runner: stage events + devloop persistence', () => {
+  it('emits started + completed events for every stage and persists iterations', async () => {
+    const snap = snapshotWith({
+      pipeline: {
+        stages: [
+          { id: 'discovery', atoms: ['discovery-question-form'] },
+          { id: 'plan', atoms: ['todo-write'] },
+        ],
+      },
+    });
+
+    const pipelineEvents: PluginPipelineStageEvent[] = [];
+    const outcomes = await runPipelineForRun({
+      db,
+      runId: 'run-1',
+      projectId: 'project-1',
+      conversationId: 'conv-A',
+      snapshot: snap,
+      pipeline: snap.pipeline!,
+      env: { maxIterations: 3 },
+      runStage: () => ({ signals: {} }),
+      emitPipeline: (e) => pipelineEvents.push(e),
+    });
+    expect(outcomes).toHaveLength(2);
+    expect(pipelineEvents.map((e) => `${e.kind}:${e.stageId}`)).toEqual([
+      'pipeline_stage_started:discovery',
+      'pipeline_stage_completed:discovery',
+      'pipeline_stage_started:plan',
+      'pipeline_stage_completed:plan',
+    ]);
+    const iterations = listIterationsForRun(db, 'run-1');
+    expect(iterations.map((i) => i.stageId)).toEqual(['discovery', 'plan']);
+  });
+
+  it('runs the devloop until a critique signal converges or hits the iteration cap', async () => {
+    const snap = snapshotWith({
+      pipeline: {
+        stages: [
+          {
+            id: 'critique',
+            atoms: ['critique-theater'],
+            repeat: true,
+            until: 'critique.score >= 4 || iterations >= 3',
+          },
+        ],
+      },
+    });
+
+    let iter = 0;
+    const outcomes = await runPipelineForRun({
+      db,
+      runId: 'run-2',
+      projectId: 'project-1',
+      conversationId: 'conv-A',
+      snapshot: snap,
+      pipeline: snap.pipeline!,
+      env: { maxIterations: 5 },
+      runStage: () => {
+        iter += 1;
+        // Make the third iteration converge by returning a critique
+        // score >= 4. The first two iterations come back with a
+        // score that fails the until predicate.
+        return {
+          signals: { 'critique.score': iter >= 3 ? 4 : 1 },
+        };
+      },
+    });
+    expect(outcomes[0]?.iterations).toBe(3);
+    expect(outcomes[0]?.converged).toBe(true);
+    expect(outcomes[0]?.termination).toBe('until-satisfied');
+    expect(listIterationsForRun(db, 'run-2')).toHaveLength(3);
+  });
+});
+
+describe('pipeline-runner: GenUI cross-conversation cache (e2e-5)', () => {
+  it('reuses a project-tier surface answer across conversations without re-broadcasting', async () => {
+    const surface: GenUISurfaceSpec = {
+      id: '__auto_connector_slack',
+      kind: 'oauth-prompt',
+      persist: 'project',
+      capabilitiesRequired: ['connector:slack'],
+      oauth: { route: 'connector', connectorId: 'slack' },
+    };
+    const snap = snapshotWith({
+      genuiSurfaces: [surface],
+      pipeline: { stages: [{ id: 'connect', atoms: ['todo-write'] }] },
+    });
+
+    // Conversation A runs and answers the surface.
+    const eventsA: GenUISurfaceEvent[] = [];
+    await runPipelineForRun({
+      db,
+      runId: 'run-A',
+      projectId: 'project-1',
+      conversationId: 'conv-A',
+      snapshot: snap,
+      pipeline: snap.pipeline!,
+      env: { maxIterations: 1 },
+      runStage: () => ({ signals: {} }),
+      emitGenui: (e) => {
+        if (e.kind.startsWith('genui_')) eventsA.push(e as GenUISurfaceEvent);
+      },
+    });
+    const requestA = eventsA.find((e) => e.kind === 'genui_surface_request');
+    expect(requestA).toBeDefined();
+    const pending = findPendingByRunAndSurfaceId(db, {
+      runId: 'run-A',
+      surfaceId: surface.id,
+    });
+    expect(pending).not.toBeNull();
+    respondSurface(db, {
+      rowId: pending!.id,
+      value: { connectorId: 'slack', accountLabel: 'Acme' },
+      respondedBy: 'user',
+      runId: 'run-A',
+    });
+
+    // Conversation B re-applies the same plugin in the same project.
+    // The runner asks the registry first; on cache hit it emits a
+    // `genui_surface_response { respondedBy: 'cache' }` event WITHOUT
+    // a fresh `genui_surface_request`. This is the e2e-5 anchor.
+    const eventsB: GenUISurfaceEvent[] = [];
+    await runPipelineForRun({
+      db,
+      runId: 'run-B',
+      projectId: 'project-1',
+      conversationId: 'conv-B',
+      snapshot: snap,
+      pipeline: snap.pipeline!,
+      env: { maxIterations: 1 },
+      runStage: () => ({ signals: {} }),
+      emitGenui: (e) => {
+        if (e.kind.startsWith('genui_')) eventsB.push(e as GenUISurfaceEvent);
+      },
+    });
+    const surfaceEventsB = eventsB.filter((e) => e.surfaceId === surface.id);
+    const requestB = surfaceEventsB.find((e) => e.kind === 'genui_surface_request');
+    const responseB = surfaceEventsB.find((e) => e.kind === 'genui_surface_response');
+    expect(requestB).toBeUndefined();
+    expect(responseB).toBeDefined();
+    if (responseB && responseB.kind === 'genui_surface_response') {
+      expect(responseB.respondedBy).toBe('cache');
+    }
+  });
+
+  it('raises a stage-bound surface only when the matching stage starts', async () => {
+    const surface: GenUISurfaceSpec = {
+      id: 'audience-clarify',
+      kind: 'form',
+      persist: 'conversation',
+      trigger: { stageId: 'discovery' },
+      schema: { type: 'object' },
+    };
+    const snap = snapshotWith({
+      genuiSurfaces: [surface],
+      pipeline: {
+        stages: [
+          { id: 'discovery', atoms: ['discovery-question-form'] },
+          { id: 'plan', atoms: ['todo-write'] },
+        ],
+      },
+    });
+    const events: GenUISurfaceEvent[] = [];
+    await runPipelineForRun({
+      db,
+      runId: 'run-stage',
+      projectId: 'project-1',
+      conversationId: 'conv-A',
+      snapshot: snap,
+      pipeline: snap.pipeline!,
+      env: { maxIterations: 1 },
+      runStage: () => ({ signals: {} }),
+      emitGenui: (e) => {
+        if (e.kind.startsWith('genui_')) events.push(e as GenUISurfaceEvent);
+      },
+    });
+    const requests = events.filter((e) => e.kind === 'genui_surface_request');
+    expect(requests).toHaveLength(1);
+    expect(requests[0]?.surfaceId).toBe('audience-clarify');
+  });
+});