From 531df0e270a8595f3971cebca4fedec8a60aa935 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 9 May 2026 11:29:57 +0000 Subject: [PATCH] =?UTF-8?q?feat(plugins):=20pipeline-runner=20=E2=80=94=20?= =?UTF-8?q?stage=20events=20+=20GenUI=20cache=20integration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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_ 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> --- apps/daemon/src/plugins/index.ts | 1 + apps/daemon/src/plugins/pipeline-runner.ts | 109 ++++++++ .../tests/plugins-pipeline-runner.test.ts | 260 ++++++++++++++++++ 3 files changed, 370 insertions(+) create mode 100644 apps/daemon/src/plugins/pipeline-runner.ts create mode 100644 apps/daemon/tests/plugins-pipeline-runner.test.ts diff --git a/apps/daemon/src/plugins/index.ts b/apps/daemon/src/plugins/index.ts index 8983087f78..2734e4c66a 100644 --- a/apps/daemon/src/plugins/index.ts +++ b/apps/daemon/src/plugins/index.ts @@ -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'; diff --git a/apps/daemon/src/plugins/pipeline-runner.ts b/apps/daemon/src/plugins/pipeline-runner.ts new file mode 100644 index 0000000000..607d6fe45d --- /dev/null +++ b/apps/daemon/src/plugins/pipeline-runner.ts @@ -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 { + // 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}`); + } +} diff --git a/apps/daemon/tests/plugins-pipeline-runner.test.ts b/apps/daemon/tests/plugins-pipeline-runner.test.ts new file mode 100644 index 0000000000..4352ba7122 --- /dev/null +++ b/apps/daemon/tests/plugins-pipeline-runner.test.ts @@ -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 { + 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'); + }); +});