Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -523,6 +523,20 @@ const McpServersSchema = z.record(z.string(), McpServerSchema).default({});
* `mcpServers` block. Normalized into an `McpServerSpec` by `McpRegistry`. */
export type RawMcpServerConfig = z.infer<typeof McpServerSchema>;

/**
* SDLC pipeline (docs/sdlc-pipeline.md). OFF by default — the daemon stays
* methodology-agnostic; nothing runs until this is enabled. When enabled, the
* daemon builds a PipelineManager sharing its DB and rehydrates non-terminal
* pipelines on boot. `defaultPack` is the methodology pack a pipeline runs when
* created without one (null = none / freestyle).
*/
const PipelineSchema = z
.object({
enabled: z.boolean().default(false),
defaultPack: z.string().nullable().default(null),
})
.default({ enabled: false, defaultPack: null });

const RootSchema = z.object({
daemonUrl: z.string().default("ws://127.0.0.1:7400"),
dbPath: z.string().default("codeoid.db"),
Expand All @@ -544,6 +558,7 @@ const RootSchema = z.object({
session: SessionSchema,
conductor: ConductorSchema,
dispatch: DispatchSchema,
pipeline: PipelineSchema,
providers: ProvidersSchema,
mcpServers: McpServersSchema,
hooks: HooksSchema,
Expand Down Expand Up @@ -678,6 +693,16 @@ export interface CodeoidConfig {
workerToolBudget: number;
retryBaseMs: number;
};
/**
* SDLC pipeline — OFF by default (docs/sdlc-pipeline.md). When enabled, a
* PipelineManager is constructed at boot sharing the daemon DB, and
* non-terminal pipelines are rehydrated (resume). Optional in the type so
* hand-built test configs stay minimal; loadConfig always populates it.
*/
pipeline?: {
enabled: boolean;
defaultPack: string | null;
};
/**
* Per-backend provider settings. Optional in the type so hand-built test
* configs stay minimal; loadConfig always populates it (schema defaults).
Expand Down Expand Up @@ -777,6 +802,10 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [
// without touching config.json. Other dispatch knobs are file-config only,
// matching the conductor block's convention.
{ env: "CODEOID_DISPATCH_ENABLED", path: "dispatch.enabled", kind: "boolean" },
// Pipeline enable/kill switch — turn the SDLC pipeline on/off per-invocation
// without touching config.json (off by default). Other pipeline knobs are
// file-config only, matching the dispatch/conductor convention.
{ env: "CODEOID_PIPELINE_ENABLED", path: "pipeline.enabled", kind: "boolean" },
{ env: "CODEOID_FALLBACK_MODEL", path: "session.fallbackModel", kind: "string" },
// Hooks kill switch — disable every configured hook per-invocation without
// touching config.json. Entries themselves are file-config only.
Expand Down Expand Up @@ -978,6 +1007,7 @@ export function loadConfig(opts: LoadOptions = {}): CodeoidConfig {
session: parsed.session,
conductor: parsed.conductor,
dispatch: parsed.dispatch,
pipeline: parsed.pipeline,
providers: parsed.providers,
mcpServers: parsed.mcpServers,
hooks: parsed.hooks,
Expand Down
79 changes: 79 additions & 0 deletions src/daemon/pipeline/answer.test.ts
Original file line number Diff line number Diff line change
@@ -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");
});
});
3 changes: 3 additions & 0 deletions src/daemon/pipeline/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,6 @@ export { alwaysGate, manualGate, noopPhaseKind, registerBuiltins } from "./built
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";
41 changes: 41 additions & 0 deletions src/daemon/pipeline/manager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
29 changes: 29 additions & 0 deletions src/daemon/pipeline/runner.ts
Original file line number Diff line number Diff line change
@@ -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>;
}
86 changes: 86 additions & 0 deletions src/daemon/pipeline/skill-kind.test.ts
Original file line number Diff line number Diff line change
@@ -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" });
});
});
58 changes: 58 additions & 0 deletions src/daemon/pipeline/skill-kind.ts
Original file line number Diff line number Diff line change
@@ -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 };
}
Loading