feat(pipeline): skill phase kind + PhaseRunner seam + halt/answer/resume

src/daemon/pipeline/answer.test.ts

@@ -0,0 +1,79 @@
+import { Database } from "bun:sqlite";
+import { describe, expect, test } from "bun:test";
+import type { PhaseDef } from "./interface";
+import { PipelineManager } from "./manager";
+import { PipelineStore } from "./store";
+
+const tenant = { accountId: "acct", projectId: "proj", createdBy: "user" };
+
+function mgr(): PipelineManager {
+  return new PipelineManager(new PipelineStore(new Database(":memory:")));
+}
+
+/** A pipeline whose first phase halts on the `manual` gate, followed by a phase
+ *  that passes — so answering resumes into the tail. */
+const halting: PhaseDef[] = [
+  { id: "gate", kind: "noop", gate: "manual" },
+  { id: "tail", kind: "noop", gate: "always" },
+];
+
+describe("PipelineManager.answer (halt → resume)", () => {
+  test("approve resolves the halted phase and resumes to done", async () => {
+    const m = mgr();
+    const p = m.create({ name: "REQ-1", phases: halting, ...tenant });
+    const halted = await m.advance(p.id);
+    expect(halted.status).toBe("halted");
+    expect(halted.phases[0].state.status).toBe("halted");
+
+    const done = await m.answer(p.id, "gate:gate", { approved: true, value: "LGTM" });
+    expect(done.status).toBe("done");
+    expect(done.phases[0].state).toMatchObject({ status: "passed", summary: "LGTM" });
+    expect(done.phases[1].state.status).toBe("passed");
+  });
+
+  test("reject fails the phase and the pipeline", async () => {
+    const m = mgr();
+    const p = m.create({ name: "REQ-1", phases: halting, ...tenant });
+    await m.advance(p.id);
+    const failed = await m.answer(p.id, "gate:gate", { approved: false, value: "nope" });
+    expect(failed.status).toBe("failed");
+    expect(failed.phases[0].state).toMatchObject({ status: "failed", reason: "nope" });
+  });
+
+  test("approving the final phase completes without a further advance", async () => {
+    const m = mgr();
+    const p = m.create({ name: "REQ-1", phases: [{ id: "only", kind: "noop", gate: "manual" }], ...tenant });
+    await m.advance(p.id);
+    const done = await m.answer(p.id, "gate:only", { approved: true });
+    expect(done.status).toBe("done");
+    expect(done.phases[0].state).toMatchObject({ status: "passed", summary: "approved" });
+  });
+
+  test("rejects a stale requestId", async () => {
+    const m = mgr();
+    const p = m.create({ name: "REQ-1", phases: halting, ...tenant });
+    await m.advance(p.id);
+    await expect(m.answer(p.id, "gate:wrong", { approved: true })).rejects.toThrow("stale requestId");
+  });
+
+  test("throws when the pipeline is not halted", async () => {
+    const m = mgr();
+    const p = m.create({ name: "REQ-1", phases: [{ id: "one", kind: "noop", gate: "always" }], ...tenant });
+    await m.advance(p.id); // runs straight to done
+    await expect(m.answer(p.id, "gate:one", { approved: true })).rejects.toThrow("not halted");
+  });
+
+  test("throws on an unknown pipeline id", async () => {
+    await expect(mgr().answer("nope", "gate:x", { approved: true })).rejects.toThrow("not found");
+  });
+
+  test("the resolved state persists across a restart", async () => {
+    const store = new PipelineStore(new Database(":memory:"));
+    const m = new PipelineManager(store);
+    const p = m.create({ name: "REQ-1", phases: halting, ...tenant });
+    await m.advance(p.id);
+    await m.answer(p.id, "gate:gate", { approved: true });
+    // Fresh manager over the same store sees the completed pipeline.
+    expect(new PipelineManager(store).get(p.id)?.status).toBe("done");
+  });
+});

src/daemon/pipeline/index.ts

@@ -15,3 +15,5 @@ export { PipelineEngine } from "./engine";
 export { PipelineStore } from "./store";
 export { type CreatePipelineOpts, PipelineManager } from "./manager";
 export { createPipelineManagerFromConfig, type PipelineWiringConfig } from "./wiring";
+export { makeSkillPhaseKind } from "./skill-kind";
+export type { PhaseRunner, PhaseRunRequest, PhaseRunOutput } from "./runner";

src/daemon/pipeline/manager.ts

@@ -78,6 +78,47 @@ export class PipelineManager {
     });
   }
 
+  /**
+   * Resolve a halted phase with a human decision, then resume the pipeline.
+   * `approved` marks the phase passed (its `value` becomes the summary) and
+   * advances; otherwise the phase — and the pipeline — fail. This is the
+   * daemon-side of the halt → answer-from-a-frontend → resume path (§4.1, §5.3):
+   * a `ui_response` lands here to unblock a pipeline that a gate parked.
+   */
+  async answer(
+    id: string,
+    requestId: string,
+    opts: { approved: boolean; value?: string },
+  ): Promise<PipelineState> {
+    const s = this.get(id);
+    if (!s) throw new Error(`pipeline "${id}" not found`);
+    if (s.status !== "halted") throw new Error(`pipeline "${id}" is not halted (status: ${s.status})`);
+    const current = s.phases[s.cursor];
+    if (!current || current.state.status !== "halted") {
+      throw new Error(`pipeline "${id}" has no halted phase at the cursor`);
+    }
+    if (current.state.requestId !== requestId) {
+      throw new Error(`stale requestId "${requestId}" for pipeline "${id}"`);
+    }
+
+    const next = cloneState(s);
+    const phase = next.phases[next.cursor];
+    if (opts.approved) {
+      phase.state = { status: "passed", summary: opts.value ?? "approved" };
+      next.cursor += 1;
+      next.status = next.cursor >= next.phases.length ? "done" : "running";
+    } else {
+      phase.state = { status: "failed", reason: opts.value ?? "rejected by human", attempts: 1 };
+      next.status = "failed";
+    }
+    next.updatedAt = Date.now();
+    this.#store.save(next);
+    this.#cache.set(id, next);
+
+    // Approving a non-final phase resumes the run to the next halt / terminal.
+    return next.status === "running" ? this.advance(id) : next;
+  }
+
   /** Mark a pipeline abandoned (terminal). Returns undefined if unknown. */
   abort(id: string): PipelineState | undefined {
     const s = this.get(id);

src/daemon/pipeline/runner.ts

@@ -0,0 +1,29 @@
+/**
+ * The seam between a phase and a backend. A `PhaseRunner` runs a phase's prompt
+ * on a (worker) session and returns its result. The concrete SessionManager-
+ * backed adapter lands in a later slice; abstracting it here keeps the pipeline
+ * package free of Session dependencies and lets the skill phase kind be tested
+ * with a fake runner.
+ */
+
+import type { PhaseDef, PipelineState } from "./interface";
+
+export interface PhaseRunRequest {
+  /** the resolved prompt / slash command to run for this phase. */
+  prompt: string;
+  /** per-phase backend override — enables cross-provider-per-phase routing. */
+  provider?: string;
+  /** per-phase model override. */
+  model?: string;
+  pipeline: PipelineState;
+  phase: PhaseDef;
+}
+
+export interface PhaseRunOutput {
+  summary?: string;
+  artifacts?: string[];
+}
+
+export interface PhaseRunner {
+  runPrompt(req: PhaseRunRequest): Promise<PhaseRunOutput>;
+}

src/daemon/pipeline/skill-kind.test.ts

@@ -0,0 +1,86 @@
+import { describe, expect, test } from "bun:test";
+import type { PhaseCtx, PhaseDef, PipelineState, SkillPlugin } from "./interface";
+import { createRegistries } from "./registry";
+import type { PhaseRunner, PhaseRunRequest } from "./runner";
+import { makeSkillPhaseKind } from "./skill-kind";
+
+function ctxFor(phase: PhaseDef, skills: SkillPlugin[]): PhaseCtx {
+  const registries = createRegistries();
+  for (const s of skills) registries.skills.register(s);
+  const pipeline: PipelineState = {
+    id: "p",
+    name: "p",
+    phases: [{ def: phase, state: { status: "running", startedAt: 1, attempts: 0 } }],
+    cursor: 0,
+    status: "running",
+    accountId: "a",
+    projectId: "p",
+    createdBy: "u",
+    createdAt: 1,
+    updatedAt: 1,
+  };
+  return { pipeline, phase, registries };
+}
+
+describe("skill phase kind", () => {
+  test("runs an fn skill natively and passes with its summary", async () => {
+    const skill: SkillPlugin = {
+      id: "hello",
+      kind: "fn",
+      async run() {
+        return { summary: "did the thing", artifacts: ["out.md"] };
+      },
+    };
+    const kind = makeSkillPhaseKind();
+    const res = await kind.run(ctxFor({ id: "one", kind: "skill", skill: "hello" }, [skill]));
+    expect(res).toEqual({ outcome: "passed", summary: "did the thing", artifacts: ["out.md"] });
+  });
+
+  test("a throwing fn skill fails gracefully", async () => {
+    const skill: SkillPlugin = {
+      id: "boom",
+      kind: "fn",
+      async run() {
+        throw new Error("fn-boom");
+      },
+    };
+    const res = await makeSkillPhaseKind().run(ctxFor({ id: "one", kind: "skill", skill: "boom" }, [skill]));
+    expect(res.outcome).toBe("failed");
+    if (res.outcome === "failed") expect(res.reason).toContain("fn-boom");
+  });
+
+  test("fails when the phase declares no skill id", async () => {
+    const res = await makeSkillPhaseKind().run(ctxFor({ id: "one", kind: "skill" }, []));
+    expect(res.outcome).toBe("failed");
+    if (res.outcome === "failed") expect(res.reason).toContain("no skill id");
+  });
+
+  test("fails on an unknown skill", async () => {
+    const res = await makeSkillPhaseKind().run(ctxFor({ id: "one", kind: "skill", skill: "nope" }, []));
+    expect(res.outcome).toBe("failed");
+    if (res.outcome === "failed") expect(res.reason).toContain("unknown skill");
+  });
+
+  test("a prompt/slash skill fails cleanly when no runner is configured", async () => {
+    const skill: SkillPlugin = { id: "spec", kind: "slash", command: "/spec" };
+    const res = await makeSkillPhaseKind().run(ctxFor({ id: "one", kind: "skill", skill: "spec" }, [skill]));
+    expect(res.outcome).toBe("failed");
+    if (res.outcome === "failed") expect(res.reason).toContain("needs a phase runner");
+  });
+
+  test("drives a prompt/slash skill through the runner with per-phase provider/model", async () => {
+    const seen: PhaseRunRequest[] = [];
+    const runner: PhaseRunner = {
+      async runPrompt(req) {
+        seen.push(req);
+        return { summary: "ran on backend" };
+      },
+    };
+    const skill: SkillPlugin = { id: "spec", kind: "slash", command: "/spec" };
+    const phase: PhaseDef = { id: "one", kind: "skill", skill: "spec", provider: "gemini", model: "flash" };
+    const res = await makeSkillPhaseKind(runner).run(ctxFor(phase, [skill]));
+    expect(res).toEqual({ outcome: "passed", summary: "ran on backend", artifacts: undefined });
+    expect(seen).toHaveLength(1);
+    expect(seen[0]).toMatchObject({ prompt: "/spec", provider: "gemini", model: "flash" });
+  });
+});

src/daemon/pipeline/skill-kind.ts

@@ -0,0 +1,58 @@
+/**
+ * The "skill" phase kind: resolve the phase's `skill` from the registry and run
+ * it. A `fn` skill runs natively (no backend); a `prompt` / `slash` skill is
+ * driven through an injected PhaseRunner (the daemon backend seam). Without a
+ * runner, a prompt/slash skill fails with a clear reason — so the kind stays
+ * usable in pure tests and degrades safely when no backend is wired.
+ */
+
+import type { PhaseCtx, PhaseKind, PhaseRunResult, SkillPlugin } from "./interface";
+import type { PhaseRunner } from "./runner";
+
+export function makeSkillPhaseKind(runner?: PhaseRunner): PhaseKind {
+  return {
+    id: "skill",
+    async run(ctx: PhaseCtx): Promise<PhaseRunResult> {
+      const skillId = ctx.phase.skill;
+      if (!skillId) {
+        return { outcome: "failed", reason: `phase "${ctx.phase.id}" has kind:"skill" but no skill id` };
+      }
+      const skill = ctx.registries.skills.resolve(skillId);
+      if (!skill) return { outcome: "failed", reason: `unknown skill "${skillId}"` };
+      return runSkill(skill, ctx, runner);
+    },
+  };
+}
+
+async function runSkill(
+  skill: SkillPlugin,
+  ctx: PhaseCtx,
+  runner: PhaseRunner | undefined,
+): Promise<PhaseRunResult> {
+  if (skill.kind === "fn") {
+    try {
+      const res = await skill.run(ctx);
+      return { outcome: "passed", summary: res.summary, artifacts: res.artifacts };
+    } catch (err) {
+      // A throwing native skill is a phase failure, not a crash. (The engine
+      // also guards this, but catching here attributes the error to the skill.)
+      const reason = err instanceof Error ? err.message : String(err);
+      return { outcome: "failed", reason: `skill "${skill.id}" threw: ${reason}` };
+    }
+  }
+  if (!runner) {
+    return {
+      outcome: "failed",
+      reason: `skill "${skill.id}" (${skill.kind}) needs a phase runner, but none is configured`,
+    };
+  }
+  const prompt = skill.kind === "slash" ? skill.command : skill.template;
+  const res = await runner.runPrompt({
+    prompt,
+    provider: ctx.phase.provider,
+    model: ctx.phase.model,
+    pipeline: ctx.pipeline,
+    phase: ctx.phase,
+  });
+  return { outcome: "passed", summary: res.summary, artifacts: res.artifacts };
+}