From b522849ce2a0143f4f0319f1c76b3b8ecf101525 Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 12:41:30 +0200 Subject: [PATCH 1/4] refactor(conductor): delete the dead dev-job step coupling (epic #470 C5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Conductor's `dev.job` step was built in W3 and never wired: `conductor/index.ts` constructs `ConductorRunExecutor` without a `devJob` dep, so the dispatch branch was permanently false; the launch half had no implementation at all (`createConductorJob`, `setAwaitId`, `getAwaitId` existed only as interface members); `DevJobOutcomeEmitter.emit()` had no caller in `src/`; nothing scheduled the reconciliation sweep; and no bundled template referenced `dev.job` (which `listActions` rejects anyway). Per `specs/470-dev-platform-plugin/dormant-capabilities.md` §1 the verdict is DELETE, not genericise — nothing needs the generic version. Deleted whole: `conductor/devJobStepEffect.ts`, `devplatform/devJobConductorBridge.ts`, `test/conductorDevJobStep.test.ts`. Stripped from `runExecutor.ts`: both imports, the `devJob` field + ctor dep, `DevJobPortUnavailableError`, the dispatch branch, `resolveDevJobAwait`, `reconcileTerminalDevJobAwaits`, `openDevJobAwait`. `awaitStore.listWaiting` drops its `AND channel_type <> 'dev_job'` predicate outright rather than keeping a generic filter. `openHumanAwait` is now the single caller of `create`, so every await has a human holder by construction and the excluded set is provably empty; a filter whose complement no member can enter asserts an invariant in the wrong place, and a future non-human await kind would have to remember to add itself to survive it. New `test/conductorAwaitStore.test.ts` guards the channel-agnostic contract (mutation-checked: re-adding a channel predicate fails it). `dev_jobs.conductor_await_id` (migration 0024) is KEPT — migrations are forward-only here and a DROP is the one irreversible act. It is marked as an orphaned column so the schema is not misread as evidence of a live feature. Core-decoupling ratchet: 3303 -> 3164 (-139), no zone rose. --- .../migrations/0024_dev_platform_w3.sql | 11 +- middleware/src/conductor/awaitStore.ts | 32 +- middleware/src/conductor/devJobStepEffect.ts | 122 ------ middleware/src/conductor/routes.ts | 4 +- middleware/src/conductor/runExecutor.ts | 149 +------- .../src/devplatform/devJobConductorBridge.ts | 113 ------ middleware/test/conductorAwaitStore.test.ts | 121 ++++++ middleware/test/conductorDevJobStep.test.ts | 358 ------------------ 8 files changed, 145 insertions(+), 765 deletions(-) delete mode 100644 middleware/src/conductor/devJobStepEffect.ts delete mode 100644 middleware/src/devplatform/devJobConductorBridge.ts create mode 100644 middleware/test/conductorAwaitStore.test.ts delete mode 100644 middleware/test/conductorDevJobStep.test.ts diff --git a/middleware/migrations/0024_dev_platform_w3.sql b/middleware/migrations/0024_dev_platform_w3.sql index f90d4f34..d0c027f2 100644 --- a/middleware/migrations/0024_dev_platform_w3.sql +++ b/middleware/migrations/0024_dev_platform_w3.sql @@ -15,8 +15,15 @@ ALTER TABLE dev_repos ADD COLUMN IF NOT EXISTS policy_overrides jsonb NOT NULL DEFAULT '{}'::jsonb; -- --- link a parked job to its holding conductor await (W3 §5) --------------- --- Nullable: only Conductor-driven jobs park on an await. One-await invariant is --- enforced in conductorBridge, not by a DB constraint here. +-- ORPHANED COLUMN — do NOT read this as evidence of a live feature. +-- It was added for the Conductor `dev.job` step, which was built but never wired: +-- the run executor was always constructed without the port, so the dispatch branch +-- was permanently false and NOTHING EVER WROTE THIS COLUMN. That step was deleted +-- in epic #470 C5 (see the dormant-capabilities spec, §1). The column stays only +-- because migrations are forward-only here and a DROP is the one irreversible act +-- in that change. It has no reader and no writer in `src/`, and it is always NULL. +-- If the step is ever rebuilt (as a new feature, in the plugin repo) it may reclaim +-- this column; otherwise a future migration squash may drop it. ALTER TABLE dev_jobs ADD COLUMN IF NOT EXISTS conductor_await_id text; diff --git a/middleware/src/conductor/awaitStore.ts b/middleware/src/conductor/awaitStore.ts index 65c2cfe7..9679e8e6 100644 --- a/middleware/src/conductor/awaitStore.ts +++ b/middleware/src/conductor/awaitStore.ts @@ -114,38 +114,26 @@ export class ConductorAwaitStore { } /** - * All waiting HUMAN awaits (the operator inbox). Dev-job awaits (`channel_type='dev_job'`, - * Epic #470 W3) are excluded: they have no human holder and are resolved by a terminal dev-job - * outcome, not an operator response — surfacing them would show a phantom, un-actionable row - * whose `respond` can only ever be rejected by the authz gate. + * All waiting awaits — the operator inbox. + * + * Every await IS a human await by construction: `openHumanAwait` is the single caller of + * {@link create}, and it always writes a human principal and a human channel. An earlier + * `AND channel_type <> 'dev_job'` filter hid the never-built dev-job step's synthetic awaits; + * that step was deleted with its writer, so the excluded set is now empty and the filter is + * gone rather than kept as a generic one — a filter whose complement no member can enter is + * an invariant asserted in the wrong place, and a future non-human await kind would have to + * remember to add itself to survive it. If such a kind ever lands, it names itself here. */ async listWaiting(limit = 100): Promise { const r = await this.pool.query( `SELECT ${COLS} FROM conductor_awaits - WHERE status = 'waiting' AND channel_type <> 'dev_job' + WHERE status = 'waiting' ORDER BY created_at ASC LIMIT $1`, [Math.min(Math.max(1, limit), 500)], ); return r.rows.map(toAwait); } - /** - * Waiting dev-job awaits (`channel_type='dev_job'`) — the reconciliation sweep's input - * (Epic #470 W3). Deliberately the COMPLEMENT of {@link listWaiting}: these carry a synthetic - * `dev_job:` principal and are recovered by `reconcileTerminalDevJobAwaits`, which asks - * the dev-job port whether the bound job is already terminal (closing the terminal-before-bind - * lost-wakeup window). - */ - async listWaitingDevJobAwaits(limit = 200): Promise { - const r = await this.pool.query( - `SELECT ${COLS} FROM conductor_awaits - WHERE status = 'waiting' AND channel_type = 'dev_job' - ORDER BY created_at ASC LIMIT $1`, - [Math.min(Math.max(1, limit), 1000)], - ); - return r.rows.map(toAwait); - } - /** Waiting awaits whose deadline has passed (for the deadline worker). */ async listDue(now: Date): Promise { const r = await this.pool.query( diff --git a/middleware/src/conductor/devJobStepEffect.ts b/middleware/src/conductor/devJobStepEffect.ts deleted file mode 100644 index ca7dbfc6..00000000 --- a/middleware/src/conductor/devJobStepEffect.ts +++ /dev/null @@ -1,122 +0,0 @@ -import type { JsonObject, JsonValue, Step } from '@omadia/conductor-core'; - -/** - * Epic #470 W3 — the dev-job Conductor step. - * - * A dev job runs for minutes and is modelled as ONE opaque Conductor step: when the - * executor reaches it, it starts a single dev job, opens a durable await bound to that - * job, and parks the run `waiting` (mirroring `openHumanAwait`). When the job reaches a - * terminal state a resolver feeds the terminal outcome back through - * `ConductorRunExecutor.resolveDevJobAwait`, which resumes the run synchronously via - * `nextStep`. The terminal outcome becomes the step's `result`, so the WORKFLOW — not this - * effect — decides what happens next via its transitions/guards (a done/failed/gate/deny - * job all resume the run; only the branch taken differs). - * - * The step reuses the existing `action` step path (a reserved `actionId`) so no - * conductor-core schema change is needed. Every side effect (launch, await binding, event - * subscription) goes through the injected ports below, keeping the deterministic engine - * pure. The conductor layer never imports devplatform concretely — the devplatform binding - * (`devJobConductorBridge.ts`) implements these interfaces and the boot wiring binds them. - */ - -/** Reserved `actionId` that marks an `action` step as a dev-job step. Reusing the action - * path is deliberate: it avoids adding a new `StepKind` (and so a conductor-core schema - * migration) just to launch a job. */ -export const DEV_JOB_ACTION_ID = 'dev.job'; - -/** True when a step is a dev-job step — an `action` step carrying the reserved `actionId`. - * Backward compatible: without the port wired, such a step falls through to the normal - * action effect, so existing graphs are unaffected. */ -export function isDevJobStep(step: Step): boolean { - return step.kind === 'action' && step.actionId === DEV_JOB_ACTION_ID; -} - -/** Prefix of the synthetic `principalRef` a dev-job await carries (a dev job has no human - * holder). The jobId is recoverable from the await alone, so the reconciliation sweep needs - * no extra column. Defined once here and shared by open + reconcile. */ -export const DEV_JOB_PRINCIPAL_PREFIX = 'dev_job:'; - -/** Build the synthetic principalRef for a dev-job await. */ -export function buildDevJobPrincipalRef(jobId: string): string { - return `${DEV_JOB_PRINCIPAL_PREFIX}${jobId}`; -} - -/** Recover the jobId from a dev-job await's principalRef, or `null` if it is not one. */ -export function parseDevJobPrincipalRef(principalRef: string): string | null { - if (!principalRef.startsWith(DEV_JOB_PRINCIPAL_PREFIX)) return null; - const jobId = principalRef.slice(DEV_JOB_PRINCIPAL_PREFIX.length); - return jobId.length > 0 ? jobId : null; -} - -/** - * Terminal outcome of a dev job, fed back to the parked Conductor step as its `result`. - * `status` is a terminal DevJobStatus (`done`/`failed`/`cancelled`/`stalled`/ - * `budget_exceeded`) — kept as a plain `string` so the conductor layer needs no devplatform - * import. The effect reports these fields faithfully; guards read them (e.g. - * `stepResult.status == 'done'`, `stepResult.prUrl`) to choose the next transition. - */ -export interface DevJobTerminalOutcome { - jobId: string; - status: string; - prUrl?: string | null; - branch?: string | null; - /** the runner's `DevJobResult` (carries `outcome` + any gate/deny detail), verbatim. */ - result?: JsonValue | null; - error?: string | null; -} - -/** - * Conductor-side port: start a dev job for a step and link it to its holding await. - * - * `launch` MUST be idempotent per `(runId, stepId)`: a crash between `launch` and `park` - * re-drives this step, so a second `launch` for the same run+step must return the SAME - * `jobId` rather than starting a second job. That contract, paired with the idempotent - * `awaitStore.create`, is what keeps a dev-job step to exactly one job. - */ -export interface DevJobStepPort { - launch(input: { runId: string; stepId: string; step: Step; context: JsonObject }): Promise<{ jobId: string }>; - /** Persist the await↔job link (`dev_jobs.conductor_await_id`) so the resolver can find it. */ - bindAwait(jobId: string, awaitId: string): Promise; - /** The await bound to a job, or `null` when the job is not conductor-driven. */ - awaitIdForJob(jobId: string): Promise; - /** - * The job's terminal outcome if it is ALREADY terminal, else `null` (still running / unknown). - * The recovery path for the terminal-before-bind lost-wakeup: the `DevJobOutcomeEmitter` is - * edge-triggered and unbuffered, so a job that finished before its await existed (a crash, or - * the microsecond window between `create` and `bindAwait`) never re-emits. The reconciliation - * sweep polls this for every still-waiting dev-job await and resumes the ones already done. - */ - terminalOutcomeForJob(jobId: string): Promise; -} - -/** A source of terminal dev-job outcomes — an event-bus tail or a `finalizeDevJob` hook. - * It MUST emit only for TERMINAL jobs; emitting mid-job would resume the run prematurely. */ -export interface DevJobOutcomeSource { - onTerminal(handler: (outcome: DevJobTerminalOutcome) => void | Promise): () => void; -} - -/** The executor surface the resolver drives. Decouples the wiring factory from the concrete - * `ConductorRunExecutor` class (and keeps it trivially fakeable in tests). */ -export interface DevJobAwaitResolver { - resolveDevJobAwait(outcome: DevJobTerminalOutcome): Promise; -} - -/** - * Wiring factory: tie a terminal-outcome source to the executor. Returns an unsubscribe - * handle. The boot wiring calls this — it does NOT touch boot itself. A resolve failure is - * logged and swallowed, never thrown back into the source's emit (one job's resolve crash - * must not tear down the subscription or block sibling outcomes). - */ -export function subscribeDevJobResolver(deps: { - resolver: DevJobAwaitResolver; - source: DevJobOutcomeSource; - log?: (msg: string) => void; -}): () => void { - return deps.source.onTerminal((outcome) => { - void Promise.resolve(deps.resolver.resolveDevJobAwait(outcome)).catch((err) => { - deps.log?.( - `[conductor] dev-job resolve failed for job ${outcome.jobId}: ${err instanceof Error ? err.message : String(err)}`, - ); - }); - }); -} diff --git a/middleware/src/conductor/routes.ts b/middleware/src/conductor/routes.ts index 8dfe99ee..8109f6f5 100644 --- a/middleware/src/conductor/routes.ts +++ b/middleware/src/conductor/routes.ts @@ -329,8 +329,8 @@ export function createConductorRouter(deps: ConductorRouterDeps): Router { if (err instanceof AwaitNotPendingError) { res.status(409).json({ code: 'conductor.await_not_pending', message: err.message }); } else if (err instanceof AwaitResponderNotHolderError) { - // A non-holder tried to answer (incl. the phantom `dev_job:` principal an operator - // must never own). The authz gate already refused; surface it as 403, not a generic 500. + // A non-holder tried to answer. The authz gate already refused; surface it as 403, + // not a generic 500. res.status(403).json({ code: 'conductor.await_forbidden', message: err.message }); } else { console.error('[conductor] respond failed:', err); diff --git a/middleware/src/conductor/runExecutor.ts b/middleware/src/conductor/runExecutor.ts index c884fa3f..331ab270 100644 --- a/middleware/src/conductor/runExecutor.ts +++ b/middleware/src/conductor/runExecutor.ts @@ -8,8 +8,6 @@ import type { ConductorRun, ConductorRunStore, TriggerKind } from './runStore.js import { RunLeaseLostError } from './runStore.js'; import type { ConductorAwaitStore } from './awaitStore.js'; import type { StepEffects } from './stepEffects.js'; -import type { DevJobStepPort, DevJobTerminalOutcome } from './devJobStepEffect.js'; -import { isDevJobStep, buildDevJobPrincipalRef, parseDevJobPrincipalRef } from './devJobStepEffect.js'; import { canonicalizePrincipalId } from './principalId.js'; export class WorkflowNotFoundError extends Error {} @@ -18,8 +16,6 @@ export class WorkflowNotPublishedError extends Error {} export class AwaitNotPendingError extends Error {} /** A responder who is not a current holder tried to resolve an await (authorization gate). */ export class AwaitResponderNotHolderError extends Error {} -/** A dev-job step was reached, or a dev-job outcome arrived, but no dev-job port was wired. */ -export class DevJobPortUnavailableError extends Error {} export interface PreviewStep { stepId: string; @@ -81,9 +77,6 @@ export class ConductorRunExecutor { /** Late-bound role→holders resolver — the required responders for a quorum='all' role await. * Required (not optional) so a role-based 'all' can never silently degrade to 'any' when unwired. */ private readonly resolveRoleHolders: (roleKey: string) => Promise; - /** Optional dev-job port (Epic #470 W3). Absent ⇒ the feature is off: a dev-job step falls - * through to the normal action effect and `resolveDevJobAwait` throws if ever called. */ - private readonly devJob?: DevJobStepPort; /** Issue #437 — fired once a REAL (non-dry-run) run reaches a terminal status * ('completed'/'failed'). Feeds the outbound webhook dispatcher; best-effort and * never awaited inline — a slow/broken subscriber must not stall run driving. */ @@ -96,7 +89,6 @@ export class ConductorRunExecutor { awaitStore: ConductorAwaitStore; effects: StepEffects; resolveRoleHolders: (roleKey: string) => Promise; - devJob?: DevJobStepPort; notifyRunEnded?: (run: ConductorRun) => void; log?: (msg: string) => void; }) { @@ -105,7 +97,6 @@ export class ConductorRunExecutor { this.awaitStore = deps.awaitStore; this.effects = deps.effects; this.resolveRoleHolders = deps.resolveRoleHolders; - this.devJob = deps.devJob; this.notifyRunEnded = deps.notifyRunEnded; this.log = deps.log ?? (() => undefined); } @@ -195,16 +186,6 @@ export class ConductorRunExecutor { continue; } - // Dev-job step (Epic #470 W3) → launch one dev job, open a durable await bound to it, - // and park. The whole minutes-long job is ONE opaque step; `resolveDevJobAwait` resumes - // the run when the job reaches a terminal state, and the workflow branches on the - // outcome. Only active when a dev-job port is wired — otherwise it falls through to the - // normal action effect below (a `dev.job` actionId with no port simply fails there). - if (this.devJob && isDevJobStep(step)) { - await this.openDevJobAwait(runId, step, context, lease); - return (await this.runStore.get(runId)) ?? (await this.requireRun(runId)); - } - let exec; try { exec = step.kind === 'agent' @@ -232,8 +213,8 @@ export class ConductorRunExecutor { throw err; } - // The while loop's natural exit represents "this drive is genuinely done" — parked - // (human/dev-job await) and RunLeaseLostError both return earlier, above. + // The while loop's natural exit represents "this drive is genuinely done" — a parked + // human await and RunLeaseLostError both return earlier, above. return this.finalizeIfEnded((await this.runStore.get(runId)) ?? (await this.requireRun(runId))); } @@ -246,7 +227,7 @@ export class ConductorRunExecutor { * Centralizes the check used at every place a drive can stop without recursing back * into `driveFrom` — `driveFrom`'s own loop-exit (above) covers everything that * happens INSIDE a drive (including a direct in-loop terminal record, which never - * goes through `applyDecision`); `resolveAwait` / `resolveDevJobAwait` (a 'complete' + * goes through `applyDecision`); `resolveAwait` (a 'complete' * or 'stuck' decision that does not resume driving) and `expireAwait`'s no-fallback * branch each call this directly for the same reason. */ @@ -351,89 +332,6 @@ export class ConductorRunExecutor { return this.finalizeIfEnded((await this.runStore.get(aw.runId)) ?? run); } - /** - * A dev job reached a terminal state — resolve its holding await and resume the run - * SYNCHRONOUSLY (the redesign: the whole job is ONE opaque step; its terminal outcome is the - * step result fed to `nextStep`). Mirrors `resolveAwait` minus the human authorization gate — - * a dev job has no human responder, so there is nobody to authorize. - * - * Idempotent — a duplicate terminal event resolves the await AT MOST ONCE, so the run never - * double-advances: the `status !== 'waiting'` guard skips an already-resolved await, and the - * atomic `close` CAS makes the winner unique under a genuine race. An unknown/unbound job (no - * `conductor_await_id`) is a no-op — a non-Conductor job simply has no run to resume. - */ - async resolveDevJobAwait(outcome: DevJobTerminalOutcome): Promise { - const port = this.devJob; - if (!port) { - throw new DevJobPortUnavailableError(`dev-job outcome for job '${outcome.jobId}' but no dev-job port wired`); - } - const awaitId = await port.awaitIdForJob(outcome.jobId); - if (!awaitId) return null; // not a Conductor-driven job (or link missing) — nothing to resume - - const aw = await this.awaitStore.get(awaitId); - // Already resolved (a duplicate terminal event) or gone → idempotent no-op. Return the run's - // current state so a caller can observe where it landed, or null if the await is unknown. - if (!aw || aw.status !== 'waiting') { - return aw ? ((await this.runStore.get(aw.runId)) ?? null) : null; - } - - const stepResult: JsonValue = { - jobId: outcome.jobId, - status: outcome.status, - prUrl: outcome.prUrl ?? null, - branch: outcome.branch ?? null, - result: outcome.result ?? null, - error: outcome.error ?? null, - }; - - // Atomic waiting → resolved. If a concurrent resolver already won, `close` returns false and - // we must NOT advance the run a second time — return its current state instead. - const won = await this.awaitStore.close(awaitId, 'resolved'); - if (!won) return (await this.runStore.get(aw.runId)) ?? null; - - const { graph, run } = await this.loadRunGraph(aw.runId); - const lease = randomUUID(); - await this.runStore.acquireLease(aw.runId, lease); // take over the parked run's lease - const decision = nextStep(graph, aw.stepId, stepResult, run.context); - const context = this.accumulate(run.context, aw.stepId, stepResult); - const seq = (await this.runStore.stepsForRun(aw.runId)).length; - const next = await this.applyDecision( - aw.runId, seq, aw.stepId, { kind: 'dev_job', jobId: outcome.jobId, status: outcome.status }, decision, context, lease, - ); - if (next) return this.driveFrom(aw.runId, graph, next, context, lease); - return this.finalizeIfEnded((await this.runStore.get(aw.runId)) ?? run); - } - - /** - * Reconciliation sweep for the terminal-before-bind lost-wakeup (Epic #470 W3). The - * `DevJobOutcomeEmitter` is edge-triggered and unbuffered, so a job that reaches a terminal - * state BEFORE its await was bound — a crash between `launch` and `bindAwait`, or the - * microsecond window between `create` and `bindAwait` — never re-emits, and neither - * `claimResumableRuns` (only `running` runs) nor the deadline worker (dev-job awaits have no - * deadline) would ever recover the parked run. Left unrecovered the run hangs forever. - * - * The sweep re-derives the wakeup from durable state: for every still-waiting dev-job await it - * asks the port whether the bound job is already terminal, and if so feeds that outcome through - * the idempotent `resolveDevJobAwait`. Safe to run repeatedly and concurrently with a live emit - * — the await's status guard + close CAS make the winner unique. Returns the number resolved. - * Wire-nothing: W4 schedules this on a timer; it is a no-op until the dev-job port is present. - */ - async reconcileTerminalDevJobAwaits(limit = 200): Promise { - const port = this.devJob; - if (!port) return 0; - const waiting = await this.awaitStore.listWaitingDevJobAwaits(limit); - let resolved = 0; - for (const aw of waiting) { - const jobId = parseDevJobPrincipalRef(aw.principalRef); - if (!jobId) continue; // not a dev-job principal (defensive) — leave it for the human paths - const outcome = await port.terminalOutcomeForJob(jobId); - if (!outcome) continue; // still running / unknown — nothing to resume yet - await this.resolveDevJobAwait(outcome); - resolved += 1; - } - return resolved; - } - /** A deadline passed with no response — close the await and fire the in-graph fallback (FR-017). */ async expireAwait(awaitId: string): Promise { const aw = await this.awaitStore.get(awaitId); @@ -563,47 +461,6 @@ export class ConductorRunExecutor { return true; } - /** - * Launch ONE dev job for a dev-job step, open a durable await bound to it, and park the run — - * the launch-side mirror of `openHumanAwait`. All I/O goes through the injected `DevJobStepPort` - * so the deterministic engine stays pure. - * - * Exactly-one-job safety across a crash-and-resume: `port.launch` is contractually idempotent - * per (runId, stepId), and `awaitStore.create` is idempotent per (run, step) via its partial - * unique index — so re-driving this step (the run is still 'running' at this step until `park` - * commits) re-uses the same job and the same await rather than doubling either. The await binds - * to a synthetic `dev_job:` principal (a dev job has no human holder) and carries no - * deadline — the dev-job worker owns stall / wall-clock reaping, not Conductor. - */ - private async openDevJobAwait(runId: string, step: Step, context: JsonObject, lease: string): Promise { - const port = this.devJob; - if (!port) { - throw new DevJobPortUnavailableError(`run ${runId}: dev-job step '${step.id}' reached but no dev-job port wired`); - } - const { jobId } = await port.launch({ runId, stepId: step.id, step, context }); - const aw = await this.awaitStore.create({ - runId, - stepId: step.id, - principalKind: 'user', - principalRef: buildDevJobPrincipalRef(jobId), - channelType: 'dev_job', - message: '', - quorum: 'any', - reminderIntervalMs: null, - deadlineAt: null, - // Deliberately NULL (unlike openHumanAwait): a dev-job await carries no deadline, so - // `expireAwait` never fires and an await-level fallback could never be read. Failure - // branching for a dev-job step is expressed as ordinary GRAPH transitions on the outcome - // (`step.fallbackTransitionId` + guards), which `resolveDevJobAwait` honours through - // `nextStep`. Copying it onto the await too would be dead data that misleads authors into - // thinking the await-level field is what catches a failed job. - fallbackTransitionId: null, - }); - await port.bindAwait(jobId, aw.id); - await this.runStore.park(runId, step.id, context, lease); - this.log(`[conductor] run ${runId} launched dev job ${jobId} at step '${step.id}' (await ${aw.id})`); - } - private accumulate(context: JsonObject, stepId: string, result: JsonValue): JsonObject { const prev = asObject(context.steps); return { ...context, steps: { ...prev, [stepId]: result } }; diff --git a/middleware/src/devplatform/devJobConductorBridge.ts b/middleware/src/devplatform/devJobConductorBridge.ts deleted file mode 100644 index 7d08d0e8..00000000 --- a/middleware/src/devplatform/devJobConductorBridge.ts +++ /dev/null @@ -1,113 +0,0 @@ -/** - * Epic #470 W3 — devplatform binding for the dev-job Conductor step. - * - * The conductor layer defines the ports (`devJobStepEffect.ts`); this file implements them - * against the dev-platform primitives. It keeps the dependency arrow pointing the right way - * (devplatform → conductor port, never conductor → devplatform) and stays thin: the heavy - * wiring (which store, which event bus, which finalize) is injected by boot, which this file - * deliberately does NOT touch. Everything here is a factory the boot wiring calls. - */ - -import type { JsonObject, JsonValue, Step } from '@omadia/conductor-core'; - -import type { - DevJobStepPort, - DevJobOutcomeSource, - DevJobTerminalOutcome, -} from '../conductor/devJobStepEffect.js'; -import { isTerminalDevJobStatus, type DevJob } from './types.js'; - -/** - * Injected primitives for the launch port. `createConductorJob` MUST be idempotent per - * `(runId, stepId)` — a re-drive after a crash presents the same key and must return the - * SAME `jobId` (see `DevJobStepPort.launch`). `setAwaitId`/`getAwaitId` read and write - * `dev_jobs.conductor_await_id` (added in migration 0024). - */ -export interface DevJobLaunchDeps { - createConductorJob(input: { - runId: string; - stepId: string; - step: Step; - context: JsonObject; - }): Promise<{ jobId: string }>; - setAwaitId(jobId: string, awaitId: string): Promise; - getAwaitId(jobId: string): Promise; - /** Load a job's terminal-relevant fields (a `DevJobStore.getJob` slice). Backs the - * reconciliation sweep's `terminalOutcomeForJob`; `null` for an unknown job. */ - getJob(jobId: string): Promise; -} - -/** Build the `DevJobStepPort` the executor launches through. A thin adapter — the injected - * deps carry all the storage/queueing concerns. */ -export function createDevJobLaunchPort(deps: DevJobLaunchDeps): DevJobStepPort { - return { - launch: (input) => deps.createConductorJob(input), - bindAwait: (jobId, awaitId) => deps.setAwaitId(jobId, awaitId), - awaitIdForJob: (jobId) => deps.getAwaitId(jobId), - terminalOutcomeForJob: async (jobId) => { - const job = await deps.getJob(jobId); - // Only a genuinely terminal job produces an outcome — a still-active job returns null so the - // sweep leaves the run parked (no premature resume). - if (!job || !isTerminalDevJobStatus(job.status)) return null; - return toTerminalOutcome(job); - }, - }; -} - -/** The terminal-job fields the outcome carries. A finalized `DevJob` satisfies it. */ -export type TerminalDevJobView = Pick< - DevJob, - 'id' | 'status' | 'prUrl' | 'branch' | 'result' | 'error' ->; - -/** Build a `DevJobTerminalOutcome` from a terminal job row. */ -export function toTerminalOutcome(job: TerminalDevJobView): DevJobTerminalOutcome { - return { - jobId: job.id, - status: job.status, - prUrl: job.prUrl, - branch: job.branch, - result: (job.result ?? null) as JsonValue | null, - error: job.error, - }; -} - -/** - * In-process fan-in for terminal dev-job outcomes → the conductor resolver. - * - * The `DevJobEventBus` is keyed per jobId and has no "all jobs" tail, so the natural place to - * observe *any* job finishing is `finalizeDevJob` (the single terminal choke point). Boot - * pushes each finalized job in via {@link emit}; this class implements `DevJobOutcomeSource` - * for `subscribeDevJobResolver`. There is no buffering and there are no timers — a subscriber - * that attaches late simply misses in-flight emits (the run is still parked and can be - * re-driven from the durable await by a future emit or a resume sweep). - */ -export class DevJobOutcomeEmitter implements DevJobOutcomeSource { - private readonly handlers = new Set<(o: DevJobTerminalOutcome) => void | Promise>(); - - onTerminal(handler: (outcome: DevJobTerminalOutcome) => void | Promise): () => void { - this.handlers.add(handler); - return () => { - this.handlers.delete(handler); - }; - } - - /** - * Push a terminal job's outcome to every subscriber. NO-OP for a non-terminal job — this - * is the guard that stops a caller wiring the emitter to every state change from resuming - * the run mid-job. Each handler is invoked in isolation; one handler's rejection cannot - * suppress the others. - */ - emit(job: TerminalDevJobView): void { - if (!isTerminalDevJobStatus(job.status)) return; - const outcome = toTerminalOutcome(job); - for (const handler of this.handlers) { - void Promise.resolve(handler(outcome)).catch(() => undefined); - } - } - - /** Number of live subscribers (test/introspection only). */ - get subscriberCount(): number { - return this.handlers.size; - } -} diff --git a/middleware/test/conductorAwaitStore.test.ts b/middleware/test/conductorAwaitStore.test.ts new file mode 100644 index 00000000..63f0efe3 --- /dev/null +++ b/middleware/test/conductorAwaitStore.test.ts @@ -0,0 +1,121 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; + +import { ConductorAwaitStore } from '../src/conductor/awaitStore.js'; + +/** + * Guards the operator-inbox query (`listWaiting`). + * + * Epic #470 C5 removed a channel-exclusion predicate from this query when the never-wired + * Conductor job step it hid was deleted along with the only code that could ever have written + * such a row. That made the inbox channel-agnostic, and this suite is what keeps it so: + * `openHumanAwait` is now the single writer of `conductor_awaits`, so EVERY waiting await has a + * human holder and belongs in the inbox regardless of which channel notifies that holder. A new + * channel type must never have to register itself somewhere to become visible to an operator. + */ +describe('ConductorAwaitStore.listWaiting — the operator inbox', () => { + interface Row { + id: string; + run_id: string; + step_id: string; + principal_kind: 'user' | 'role'; + principal_ref: string; + channel_type: string; + message: string; + quorum: 'any' | 'all'; + reminder_interval_ms: string | null; + deadline_at: Date | null; + fallback_transition_id: string | null; + status: string; + unreachable: boolean; + created_at: Date; + } + + function awaitRow(id: string, channelType: string, status = 'waiting'): Row { + return { + id, + run_id: 'run1', + step_id: id, + principal_kind: 'role', + principal_ref: 'approvers', + channel_type: channelType, + message: '', + quorum: 'any', + reminder_interval_ms: null, + deadline_at: null, + fallback_transition_id: null, + status, + unreachable: false, + created_at: new Date(0), + }; + } + + /** Behavioural fake Pool over an in-memory `conductor_awaits`: it applies the query's OWN + * predicates, so re-introducing a channel filter in the SQL really does change the result. */ + function makePool(rows: Row[]) { + const seen: string[] = []; + const params: unknown[][] = []; + return { + seen, + params, + pool: { + async query(sql: string, args: unknown[] = []) { + seen.push(sql); + params.push(args); + let out = rows; + if (/status\s*=\s*'waiting'/.test(sql)) out = out.filter((r) => r.status === 'waiting'); + const excluded = /channel_type\s*<>\s*'([a-z_]+)'/.exec(sql); + if (excluded) out = out.filter((r) => r.channel_type !== excluded[1]); + const only = /channel_type\s*=\s*'([a-z_]+)'/.exec(sql); + if (only) out = out.filter((r) => r.channel_type === only[1]); + return { rows: out, rowCount: out.length }; + }, + }, + }; + } + + it('returns every waiting await regardless of channel type', async () => { + const { pool } = makePool([ + awaitRow('a1', 'teams'), + awaitRow('a2', 'telegram'), + awaitRow('a3', 'web'), + ]); + const store = new ConductorAwaitStore(pool as never); + + const inbox = await store.listWaiting(); + + // FAIL-IF-REVERTED: any channel_type predicate in listWaiting drops a holder's await out of + // the operator inbox, and the run it parks becomes invisible rather than actionable. + assert.equal(inbox.length, 3); + assert.deepEqual( + inbox.map((a) => a.channelType).sort(), + ['teams', 'telegram', 'web'], + ); + }); + + it('still filters on status — only waiting awaits reach the inbox', async () => { + const { pool } = makePool([ + awaitRow('a1', 'teams'), + awaitRow('a2', 'teams', 'resolved'), + awaitRow('a3', 'teams', 'cancelled'), + ]); + const store = new ConductorAwaitStore(pool as never); + + const inbox = await store.listWaiting(); + + assert.equal(inbox.length, 1); + assert.equal(inbox[0]!.id, 'a1'); + assert.equal(inbox[0]!.status, 'waiting'); + }); + + it('clamps the caller-supplied limit into the supported range', async () => { + const { pool, params } = makePool([awaitRow('a1', 'teams')]); + const store = new ConductorAwaitStore(pool as never); + + await store.listWaiting(10_000); // above the cap + await store.listWaiting(0); // below the floor + + assert.equal(params[0]![0], 500); + assert.equal(params[1]![0], 1); + }); +}); diff --git a/middleware/test/conductorDevJobStep.test.ts b/middleware/test/conductorDevJobStep.test.ts deleted file mode 100644 index bd87cf2f..00000000 --- a/middleware/test/conductorDevJobStep.test.ts +++ /dev/null @@ -1,358 +0,0 @@ -import { describe, it } from 'node:test'; -import { strict as assert } from 'node:assert'; - -import { ConductorRunExecutor } from '../src/conductor/runExecutor.js'; -import { ConductorAwaitStore } from '../src/conductor/awaitStore.js'; -import { subscribeDevJobResolver } from '../src/conductor/devJobStepEffect.js'; -import { DevJobOutcomeEmitter } from '../src/devplatform/devJobConductorBridge.js'; - -// Epic #470 W3 — the dev-job Conductor step effect. All fakes, no worker, no pg. -// -// Graph: a dev-job step `dj1` that branches on the terminal outcome — -// status == 'done' → happy transition t_ok → step `ok` -// otherwise → fallback transition t_denied → step `denied` -// Both `ok` and `denied` are ordinary (no-op) action steps with no outgoing transition, so -// once the run reaches either it executes a stub effect and completes. This proves the -// WORKFLOW — not the effect — decides what happens after the job, from the outcome alone. -const graphBranch = { - entryStepId: 'dj1', - steps: [ - { id: 'dj1', kind: 'action', actionId: 'dev.job', fallbackTransitionId: 't_denied' }, - { id: 'ok', kind: 'action', actionId: 'noop' }, - { id: 'denied', kind: 'action', actionId: 'noop' }, - ], - transitions: [ - { id: 't_ok', source: 'dj1', target: 'ok', guard: { op: 'eq', path: 'stepResult.status', value: 'done' } }, - { id: 't_denied', source: 'dj1', target: 'denied' }, - ], -}; - -interface RecordedStep { - stepId: string; - actor: { kind?: string; jobId?: string; status?: string } | null; - status: string; - nextStepId: string | null; - context: { steps?: Record> }; -} - -function makeWorld(graph: unknown) { - const run = { - id: 'run1', workflowVersionId: 'v1', status: 'running', currentStepId: null as string | null, - context: {} as Record, triggerKind: 'manual', triggerSource: null, - isDryRun: false, startedAt: new Date(0), endedAt: null, - }; - const steps: RecordedStep[] = []; - interface AwaitRow { - id: string; runId: string; stepId: string; status: string; - principalRef: string; channelType: string; fallbackTransitionId: string | null; - } - const awaits = new Map(); - const byRunStep = new Map(); - const devJobLink = new Map(); // jobId → awaitId (models dev_jobs.conductor_await_id) - const terminalJobs = new Map(); - const launchCalls: Array<{ runId: string; stepId: string }> = []; - const launchByKey = new Map(); // (runId,stepId) → jobId (launch idempotency) - let awaitCounter = 0; - let jobCounter = 0; - - const workflowStore = { - async getBySlug(slug: string) { return { id: 'w1', slug, status: 'active', activeVersionId: 'v1' }; }, - async getVersion() { return { id: 'v1', workflowId: 'w1', version: 1, graph }; }, - }; - const runStore = { - async create(input: { entryStepId: string; context: Record }) { - run.status = 'running'; run.currentStepId = input.entryStepId; run.context = input.context; return run; - }, - async get() { return run; }, - async stepsForRun() { return steps; }, - async acquireLease() {}, - async recordStepAndAdvance(input: RecordedStep) { - steps.push(input); - run.currentStepId = input.nextStepId; run.status = input.status; run.context = input.context; - }, - async park(_runId: string, stepId: string, context: Record) { - run.status = 'waiting'; run.currentStepId = stepId; run.context = context; - }, - }; - const awaitStore = { - async create(input: { - runId: string; stepId: string; principalRef: string; channelType: string; - fallbackTransitionId: string | null; - }) { - const key = `${input.runId}:${input.stepId}`; - const existingId = byRunStep.get(key); - if (existingId && awaits.get(existingId)!.status === 'waiting') return awaits.get(existingId)!; - const id = `aw${++awaitCounter}`; - const row: AwaitRow = { - id, runId: input.runId, stepId: input.stepId, status: 'waiting', - principalRef: input.principalRef, channelType: input.channelType, - fallbackTransitionId: input.fallbackTransitionId, - }; - awaits.set(id, row); byRunStep.set(key, id); return row; - }, - async get(id: string) { return awaits.get(id) ?? null; }, - async close(id: string, status: string) { - const row = awaits.get(id); - if (row && row.status === 'waiting') { row.status = status; return true; } - return false; - }, - async listWaitingDevJobAwaits() { - return [...awaits.values()].filter((a) => a.status === 'waiting' && a.channelType === 'dev_job'); - }, - }; - const port = { - async launch(input: { runId: string; stepId: string }) { - launchCalls.push({ runId: input.runId, stepId: input.stepId }); - const key = `${input.runId}:${input.stepId}`; - const existing = launchByKey.get(key); - if (existing) return { jobId: existing }; // idempotent per (runId, stepId) - const jobId = `job${++jobCounter}`; - launchByKey.set(key, jobId); return { jobId }; - }, - async bindAwait(jobId: string, awaitId: string) { devJobLink.set(jobId, awaitId); }, - async awaitIdForJob(jobId: string) { return devJobLink.get(jobId) ?? null; }, - async terminalOutcomeForJob(jobId: string) { return terminalJobs.get(jobId) ?? null; }, - }; - const effects = { - async runAgentStep(): Promise { throw new Error('unused'); }, - async runActionStep(step: { id: string; actionId?: string }) { - return { result: { executed: step.id }, actor: { kind: 'action', actionId: step.actionId ?? null } }; - }, - }; - - const executor = new ConductorRunExecutor({ - workflowStore: workflowStore as never, - runStore: runStore as never, - awaitStore: awaitStore as never, - effects: effects as never, - resolveRoleHolders: async () => [], - devJob: port as never, - }); - - const setJobTerminal = ( - jobId: string, - o: { status: string; prUrl?: string | null; result?: unknown; error?: string | null }, - ) => { terminalJobs.set(jobId, { jobId, ...o }); }; - - return { executor, run, steps, awaits, awaitStore, port, launchCalls, setJobTerminal }; -} - -describe('dev-job step — launch + park (guarantee a, d)', () => { - it('reaching the step launches exactly one job and parks the run waiting with an await bound to the jobId', async () => { - const w = makeWorld(graphBranch); - const result = await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - - // Exactly one job launched. FAIL-IF-REVERTED: a re-drive that double-launched, or a launch - // per tick, would push launchCalls past 1. - assert.equal(w.launchCalls.length, 1); - // The run is parked, not advanced. FAIL-IF-REVERTED: if openDevJobAwait skipped `park`, the - // run would still be 'running' (or would have executed the action effect and completed). - assert.equal(result.status, 'waiting'); - assert.equal(w.run.status, 'waiting'); - assert.equal(w.run.currentStepId, 'dj1'); - // No premature advance: nextStep never ran, so no step was recorded (guarantee d). - // FAIL-IF-REVERTED: resuming the run before the job finished would record dj1 here. - assert.equal(w.steps.length, 0); - // The await is durably bound to the launched job (dev_jobs.conductor_await_id link). - const awaitId = await w.port.awaitIdForJob('job1'); - assert.ok(awaitId, 'the launched job must be bound to an await'); - const aw = await w.awaitStore.get(awaitId!); - assert.equal(aw!.status, 'waiting'); - assert.equal(aw!.stepId, 'dj1'); - assert.equal(aw!.channelType, 'dev_job'); - // Footgun fix (Option B): the dev-job await carries NO fallbackTransitionId even though the - // STEP declares one (`t_denied`). Dev-job failure branching is graph-level (honoured by - // nextStep in resolveDevJobAwait), not an await-level field that would be dead data. - // FAIL-IF-REVERTED: copying step.fallbackTransitionId onto the await would make this 't_denied'. - assert.equal(aw!.fallbackTransitionId, null); - }); -}); - -describe('dev-job step — terminal-before-bind reconciliation (guarantee HIGH)', () => { - it('reconcileTerminalDevJobAwaits resolves an already-terminal job whose emit was lost, so the run advances (not hung)', async () => { - const w = makeWorld(graphBranch); - // Park normally: await is created + bound, run waiting. This is also the post-re-drive state of - // the terminal-before-bind race — the second openDevJobAwait binds the await, but the job is - // already terminal so the edge-triggered emitter will never fire again. - await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - assert.equal(w.run.status, 'waiting'); - assert.equal(w.steps.length, 0); - - // The job finished (done) but no emit reached resolveDevJobAwait — the lost wakeup. - w.setJobTerminal('job1', { status: 'done', prUrl: 'https://pr/7' }); - - const n = await w.executor.reconcileTerminalDevJobAwaits(); - - // The sweep recovered the parked run: it advanced down the happy path and completed. - // FAIL-IF-REVERTED: without the sweep the run stays 'waiting' forever (steps stays empty). - assert.equal(n, 1); - assert.equal(w.run.status, 'completed'); - assert.deepEqual(w.steps.map((s) => s.stepId), ['dj1', 'ok']); - assert.equal(w.steps.find((s) => s.stepId === 'dj1')!.context.steps!.dj1!.prUrl, 'https://pr/7'); - - // Idempotent: a second sweep finds no waiting dev-job await and does nothing. - const again = await w.executor.reconcileTerminalDevJobAwaits(); - assert.equal(again, 0); - }); - - it('reconcile leaves a still-running job parked (no premature resume)', async () => { - const w = makeWorld(graphBranch); - await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - // job1 is NOT terminal → terminalOutcomeForJob returns null. - const n = await w.executor.reconcileTerminalDevJobAwaits(); - // FAIL-IF-REVERTED: resuming a non-terminal job here would advance the run mid-job. - assert.equal(n, 0); - assert.equal(w.run.status, 'waiting'); - assert.equal(w.steps.length, 0); - }); -}); - -describe('dev-job step — terminal resume (guarantee b, c)', () => { - it('resolving with a terminal `done` outcome resumes the run and the step result carries pr_url', async () => { - const w = makeWorld(graphBranch); - await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - - await w.executor.resolveDevJobAwait({ - jobId: 'job1', status: 'done', prUrl: 'https://pr/1', result: { outcome: 'diff_ready' }, - }); - - // The run resumed and branched down the happy path (guard `status == 'done'`). - // FAIL-IF-REVERTED: if the outcome were not fed as the step result, the guard could not match - // and the run would fall through to 'denied'. - assert.deepEqual(w.steps.map((s) => s.stepId), ['dj1', 'ok']); - assert.equal(w.run.status, 'completed'); - - const dj1 = w.steps.find((s) => s.stepId === 'dj1')!; - assert.equal(dj1.context.steps!.dj1!.prUrl, 'https://pr/1'); - assert.equal(dj1.context.steps!.dj1!.status, 'done'); - assert.equal(dj1.actor?.kind, 'dev_job'); - assert.equal(dj1.actor?.jobId, 'job1'); - - // The holding await was closed exactly once. - const aw = await w.awaitStore.get((await w.port.awaitIdForJob('job1'))!); - assert.equal(aw!.status, 'resolved'); - }); - - for (const c of [ - { label: 'failed', outcome: { status: 'failed', error: 'apply failed' } }, - { label: 'cancelled', outcome: { status: 'cancelled' } }, - { label: 'gate/deny (failed + deny detail)', outcome: { status: 'failed', result: { outcome: 'failed', denied: true } } }, - ]) { - it(`a terminal '${c.label}' outcome also resumes the run carrying that outcome (no crash, no hang)`, async () => { - const w = makeWorld(graphBranch); - await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - - const out = await w.executor.resolveDevJobAwait({ jobId: 'job1', ...c.outcome }); - - // Resumed (did not hang) and branched down the fallback path (status != 'done'). - // FAIL-IF-REVERTED: a resolver that threw on a non-'done' outcome, or that never called - // nextStep, would leave `steps` short of ['dj1','denied'] and the run not 'completed'. - assert.deepEqual(w.steps.map((s) => s.stepId), ['dj1', 'denied']); - assert.equal(w.run.status, 'completed'); - assert.equal(out?.status, 'completed'); - const dj1 = w.steps.find((s) => s.stepId === 'dj1')!; - assert.equal(dj1.context.steps!.dj1!.status, c.outcome.status); - }); - } -}); - -describe('dev-job step — idempotency (guarantee e)', () => { - it('a duplicate terminal event resolves the await at most once (no double-advance)', async () => { - const w = makeWorld(graphBranch); - await w.executor.startRun({ slug: 'wf', payload: {}, awaitCompletion: true }); - - await w.executor.resolveDevJobAwait({ jobId: 'job1', status: 'done', prUrl: 'https://pr/1' }); - const stepsAfterFirst = w.steps.length; - const statusAfterFirst = w.run.status; - - // Duplicate terminal event for the same job. - await w.executor.resolveDevJobAwait({ jobId: 'job1', status: 'done', prUrl: 'https://pr/1' }); - - // The second resolve is a no-op: the await is already 'resolved', so no new steps are - // recorded and the run does not advance again. - // FAIL-IF-REVERTED: dropping the `status !== 'waiting'` guard (or the atomic close CAS) would - // let the duplicate re-run nextStep and re-record dj1, doubling the advance. - assert.equal(w.steps.length, stepsAfterFirst); - assert.equal(w.run.status, statusAfterFirst); - assert.equal(w.steps.filter((s) => s.stepId === 'dj1').length, 1); - }); -}); - -describe('operator inbox excludes dev_job awaits (guarantee MEDIUM)', () => { - // Behavioural fake Pool over an in-memory `conductor_awaits` table: it applies the query's own - // channel_type predicate, so it faithfully proves the SQL filter — remove `<> 'dev_job'` from - // listWaiting and the dev_job row leaks back into the result and the assertion fails. - function awaitRow(id: string, channelType: string) { - return { - id, run_id: 'run1', step_id: id, principal_kind: channelType === 'dev_job' ? 'user' : 'role', - principal_ref: channelType === 'dev_job' ? 'dev_job:job1' : 'approvers', channel_type: channelType, - message: '', quorum: 'any', reminder_interval_ms: null, deadline_at: null, - fallback_transition_id: null, status: 'waiting', unreachable: false, created_at: new Date(0), - }; - } - function makePool(rows: Array>) { - return { - async query(sql: string) { - const waiting = rows.filter((r) => r.status === 'waiting'); - let out = waiting; - if (/channel_type\s*<>\s*'dev_job'/.test(sql)) out = waiting.filter((r) => r.channel_type !== 'dev_job'); - else if (/channel_type\s*=\s*'dev_job'/.test(sql)) out = waiting.filter((r) => r.channel_type === 'dev_job'); - return { rows: out, rowCount: out.length }; - }, - }; - } - - it('listWaiting returns human awaits only; listWaitingDevJobAwaits returns the complement', async () => { - const store = new ConductorAwaitStore( - makePool([awaitRow('h1', 'teams'), awaitRow('dj1', 'dev_job')]) as never, - ); - - const inbox = await store.listWaiting(); - // FAIL-IF-REVERTED: dropping the `channel_type <> 'dev_job'` filter surfaces dj1 in the inbox. - assert.equal(inbox.length, 1); - assert.equal(inbox[0]!.stepId, 'h1'); - assert.ok(!inbox.some((a) => a.channelType === 'dev_job')); - - const sweep = await store.listWaitingDevJobAwaits(); - assert.equal(sweep.length, 1); - assert.equal(sweep[0]!.channelType, 'dev_job'); - assert.equal(sweep[0]!.principalRef, 'dev_job:job1'); - }); -}); - -describe('DevJobOutcomeEmitter + subscribeDevJobResolver', () => { - it('emits terminal outcomes and filters non-terminal jobs (no premature resume)', () => { - const emitter = new DevJobOutcomeEmitter(); - const seen: Array<{ jobId: string; status: string; prUrl?: string | null }> = []; - emitter.onTerminal((o) => { seen.push(o); }); - - // A still-running job must NOT be forwarded. - // FAIL-IF-REVERTED: removing the isTerminalDevJobStatus guard forwards this → premature resume. - emitter.emit({ id: 'job1', status: 'running', prUrl: null, branch: null, result: null, error: null } as never); - assert.equal(seen.length, 0); - - emitter.emit({ id: 'job1', status: 'done', prUrl: 'https://pr/1', branch: 'b', result: { outcome: 'diff_ready' }, error: null } as never); - assert.equal(seen.length, 1); - assert.equal(seen[0]!.jobId, 'job1'); - assert.equal(seen[0]!.status, 'done'); - assert.equal(seen[0]!.prUrl, 'https://pr/1'); - }); - - it('routes terminal outcomes to the executor resolver and stops on unsubscribe', async () => { - const emitter = new DevJobOutcomeEmitter(); - const calls: string[] = []; - const resolver = { resolveDevJobAwait: async (o: { jobId: string }) => { calls.push(o.jobId); } }; - const unsub = subscribeDevJobResolver({ resolver, source: emitter }); - - emitter.emit({ id: 'job2', status: 'failed', prUrl: null, branch: null, result: null, error: 'x' } as never); - await new Promise((r) => setImmediate(r)); - assert.deepEqual(calls, ['job2']); - - // After unsubscribe, no further delivery. - // FAIL-IF-REVERTED: an onTerminal that ignored its returned disposer would keep delivering. - unsub(); - emitter.emit({ id: 'job3', status: 'done', prUrl: null, branch: null, result: null, error: null } as never); - await new Promise((r) => setImmediate(r)); - assert.deepEqual(calls, ['job2']); - }); -}); From fa700fb087f67cac916434672245c7530bfb69e6 Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 13:35:23 +0200 Subject: [PATCH 2/4] refactor(conductor): delete the dead dev-job step coupling (C5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First change in this epic that moves the decoupling count DOWN. 3,303 → 3,167 (−136): middleware/src −96, middleware/test −43. WHAT THIS IS NOT: the Conductor is a live feature — 31 files, ~6,200 LOC backend, 23 UI files, 7 migrations, its own spec. Runs, steps, human approval gates, templates, the Designer canvas are all untouched. All 204 conductor tests pass. WHAT WAS DEAD: one step type inside it. `dev.job` was meant to let a workflow launch a Dev Platform job, park the run, and resume on its terminal outcome. It was built and never wired: - conductor/index.ts constructs the executor with NO devJob dep, so the dispatch branch was permanently false — and `git log -S` shows it was never wired in ANY commit - the launch half had no implementation at all: createConductorJob, setAwaitId and getAwaitId existed only as interface members, and dev_jobs.conductor_await_id was never selected or written - DevJobOutcomeEmitter.emit() had no caller in src/ - nothing scheduled the reconciliation sweep - no bundled template referenced dev.job, and listActions never included it, so the step could not pass validation Removed: conductor/devJobStepEffect.ts (122 LOC), devplatform/ devJobConductorBridge.ts (113), test/conductorDevJobStep.test.ts (358), and every dev-job reference in runExecutor.ts (31 → 0), awaitStore.ts (6 → 0) and routes.ts (1 → 0). KEPT: migration 0024's conductor_await_id column. Migrations are forward-only here and dropping a column is the one irreversible act in this change; it is marked orphaned instead. THE SUBTLE PART: awaitStore's human-inbox query excluded `channel_type = 'dev_job'`. That predicate is now gone rather than genericised, because after the delete `openHumanAwait` is the sole writer of conductor_awaits — a filter whose complement no code path can populate asserts an invariant in the wrong place, and a channel allow/denylist would fail open for a future await kind that forgot to register itself. The compensating test needed a fix the review caught: its fixtures were teams/telegram/web, so restoring `<> 'dev_job'` would have left it green — it could not detect the exact regression it exists for. Added a dev_job fixture row and mutation-checked it: with the predicate restored 1 fail, without it 3 pass. Also propagates the decision into the specs, which still described the step as a capability the extraction must carry and H2 as a registry to build. H2 needed no mechanism after all — the coupling was dead, so deleting it was the whole fix. --- middleware/test/conductorAwaitStore.test.ts | 8 ++++++-- specs/470-dev-platform-plugin/README.md | 2 +- specs/470-dev-platform-plugin/acceptance.md | 4 ++-- specs/470-dev-platform-plugin/decoupling-baseline.json | 6 +++--- specs/470-dev-platform-plugin/plan.md | 5 +++-- 5 files changed, 15 insertions(+), 10 deletions(-) diff --git a/middleware/test/conductorAwaitStore.test.ts b/middleware/test/conductorAwaitStore.test.ts index 63f0efe3..87febc8a 100644 --- a/middleware/test/conductorAwaitStore.test.ts +++ b/middleware/test/conductorAwaitStore.test.ts @@ -79,6 +79,10 @@ describe('ConductorAwaitStore.listWaiting — the operator inbox', () => { awaitRow('a1', 'teams'), awaitRow('a2', 'telegram'), awaitRow('a3', 'web'), + // The literal this query used to exclude. Without a row carrying it, the + // test cannot detect `AND channel_type <> 'dev_job'` being restored — it + // would stay green against the exact regression it exists to catch. + awaitRow('a4', 'dev_job'), ]); const store = new ConductorAwaitStore(pool as never); @@ -86,10 +90,10 @@ describe('ConductorAwaitStore.listWaiting — the operator inbox', () => { // FAIL-IF-REVERTED: any channel_type predicate in listWaiting drops a holder's await out of // the operator inbox, and the run it parks becomes invisible rather than actionable. - assert.equal(inbox.length, 3); + assert.equal(inbox.length, 4); assert.deepEqual( inbox.map((a) => a.channelType).sort(), - ['teams', 'telegram', 'web'], + ['dev_job', 'teams', 'telegram', 'web'], ); }); diff --git a/specs/470-dev-platform-plugin/README.md b/specs/470-dev-platform-plugin/README.md index 9331b330..756a8466 100644 --- a/specs/470-dev-platform-plugin/README.md +++ b/specs/470-dev-platform-plugin/README.md @@ -96,7 +96,7 @@ node scripts/check-core-decoupling.mjs --update # lower the baseline ``` The ratchet counts Dev Platform references across 14 disjoint zones and **fails if the count -rises, per zone**. Baseline **3,303**. It only ever falls; raising it needs a hand-edited baseline, so +rises, per zone**. Baseline **3,167**. It only ever falls; raising it needs a hand-edited baseline, so a new coupling shows up in review instead of slipping in. That is what makes the checklist's staleness survivable — a file inventory goes stale on diff --git a/specs/470-dev-platform-plugin/acceptance.md b/specs/470-dev-platform-plugin/acceptance.md index 94cb6b39..c94a375a 100644 --- a/specs/470-dev-platform-plugin/acceptance.md +++ b/specs/470-dev-platform-plugin/acceptance.md @@ -16,7 +16,7 @@ every row passes *and* the decoupling ratchet reads zero. | Guard | What it proves | Status | |---|---|---| -| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,303** across **14** zones, per-zone regression check | +| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,167** across **14** zones, per-zone regression check | | `middleware/test/devplatform/**` (54 files) | The behaviour itself, at unit/integration level. These **move with the plugin** and must stay green in the new repo | In place, moves in P4 | | §2 capability matrix below | Nothing is silently dropped in the move | **Written here; not yet automated** | | §3 install/uninstall | The result is genuinely installable | **Not yet built** — needs P3/P4 | @@ -155,7 +155,7 @@ depend on a plugin being able to declare a public path *and* own its prefix excl | `dev_job_start` / `dev_job_status` / `dev_job_list` | orchestrator native tools | tool call creates/reads a job | | **`ctx.devJobs`** (create/get/list/listEvents) → | plugin service | **needs the G8 contract decision** — a third-party plugin still reaches dev jobs, scoped to granted repos | | **Chat job card** → | `chat/page.tsx` | **needs H3** — either the rich card still renders, or the accepted `ToolRow` degradation is what ships | -| **Conductor `dev.job` step** → | workflow step kind | **needs H2** — a workflow parks on a dev job and resumes on its terminal outcome | +| ~~Conductor `dev.job` step~~ | — | **DELETED in C5.** Never wired, no demand, no template. Not a capability the extraction carries. See `dormant-capabilities.md` §1 | ### 2.7 Operator UI diff --git a/specs/470-dev-platform-plugin/decoupling-baseline.json b/specs/470-dev-platform-plugin/decoupling-baseline.json index d8190452..5eaa2f00 100644 --- a/specs/470-dev-platform-plugin/decoupling-baseline.json +++ b/specs/470-dev-platform-plugin/decoupling-baseline.json @@ -1,8 +1,8 @@ { - "total": 3303, + "total": 3167, "zones": { - "middleware/src": 1636, - "middleware/test": 965, + "middleware/src": 1540, + "middleware/test": 925, "middleware/packages": 97, "middleware/scripts": 8, "middleware/sidecars": 195, diff --git a/specs/470-dev-platform-plugin/plan.md b/specs/470-dev-platform-plugin/plan.md index 8a3ce294..7749828e 100644 --- a/specs/470-dev-platform-plugin/plan.md +++ b/specs/470-dev-platform-plugin/plan.md @@ -52,7 +52,8 @@ They sound small. They are not, and they invalidate two decisions in the first d Concretely: **276 hardcoded items across 18 zones, ≈49,100 LOC across ~200 files.** Full work-list in `core-decoupling-checklist.md`. Three of those items are not deletions at all — core has no extension point for them, so a new generic mechanism has to be built first -(H1 public paths, H2 conductor step kinds, H3 chat tool-card renderers). +(H1 public paths, H3 chat tool-card renderers). H2 turned out NOT to need one: the +conductor coupling was dead, so C5 deleted it rather than genericising it. --- @@ -488,7 +489,7 @@ single irreversible step moved last. | **P2a** | Decide the `ctx.devJobs` contract (§4.2) and the G8 public-contract break: `DevJob*` move to `@omadia/dev-platform-plugin-api`, `plugin-api` gets a SemVer-major bump, `dev_jobs` leaves the admin-v1 DTO. Add capability edges to `dynamicAgentRuntime` or document why agent plugins are excluded. | A written, versioned contract — before any code depends on it | | **P2b** | Decide **H3** (chat card): declarative card schema, or accept degradation to `ToolRow` for out-of-repo plugins. Decide **G7 option B vs E**. | Both answers written down before code moves | | **P2c** | Mechanical decoupling: break the `wireDevPlatform ↔ routes` cycle; collapse the 41 config keys into one namespaced object. ✅ `mintAppJwt` already moved to `src/services/githubAppJwt.ts`. | `index.ts` wiring reduced to one `assembleDevPlatform(cfg)` call | -| **P3** | The extension points. **H1** dynamic `publicPaths` + exclusive prefix ownership · **H2** generic conductor step-kind/channel-type registry · **G2** `auth: 'session'` composed *inside* the disposed guard · **G3** route-local raw parser · **G4** permission-gated `graphPool@1` + shared `runPluginMigrations`. | Any plugin can own routes, exemptions, raw bodies, tables, and long-running steps | +| **P3** | The extension points. **H1** dynamic `publicPaths` + exclusive prefix ownership · **G2** `auth: 'session'` composed *inside* the disposed guard · **G3** route-local raw parser · **G4** permission-gated `graphPool@1` + shared `runPluginMigrations`. | Any plugin can own routes, exemptions, raw bodies and tables | | **P3b** | **G7** (§4.3a): extract the `@theme inline` bridge out of `globals.css`; build the plugin Tailwind subset from it and serve it — replacing the 345 hand-written lines of `harness-admin-css.ts`; add a static-asset serving path for plugin SPA bundles; reject arbitrary-value classes at ingest. | Any plugin can ship a real UI in the house design system — the platform's weakest extension point today | | **P4** | Stand up `byte5ai/omadia-plugin-dev-platform`; move ~49,100 LOC per `core-decoupling-checklist.md`; port the UI; stand up the repo's own GHCR + SBOM + signing pipeline. **Do not delete the `publicPaths` exemptions until P3 is proven on the live runner phone-home path.** | Dev Platform installs and uninstalls from its own repo | | **P5** | Migration ownership handoff (no renumbering) + ledger seed, tested against a database restored from a production snapshot. Its own PR, its own rollback story. | Plugin owns its schema | From c21851966d7be3957bb7be56fcd53d8b3fd878d6 Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 14:33:03 +0200 Subject: [PATCH 3/4] =?UTF-8?q?chore(470):=20resync=20C5=20with=20main=20(?= =?UTF-8?q?#549)=20=E2=80=94=20baseline=203,167=20=E2=86=92=203,170?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel-api work added 3 dev-platform references (test/packages). Fourth legitimate raise: main ADDED dev-platform code; core did not re-acquire a dependency. --- specs/470-dev-platform-plugin/README.md | 2 +- specs/470-dev-platform-plugin/acceptance.md | 2 +- specs/470-dev-platform-plugin/decoupling-baseline.json | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/specs/470-dev-platform-plugin/README.md b/specs/470-dev-platform-plugin/README.md index 756a8466..193a3acc 100644 --- a/specs/470-dev-platform-plugin/README.md +++ b/specs/470-dev-platform-plugin/README.md @@ -96,7 +96,7 @@ node scripts/check-core-decoupling.mjs --update # lower the baseline ``` The ratchet counts Dev Platform references across 14 disjoint zones and **fails if the count -rises, per zone**. Baseline **3,167**. It only ever falls; raising it needs a hand-edited baseline, so +rises, per zone**. Baseline **3,170**. It only ever falls; raising it needs a hand-edited baseline, so a new coupling shows up in review instead of slipping in. That is what makes the checklist's staleness survivable — a file inventory goes stale on diff --git a/specs/470-dev-platform-plugin/acceptance.md b/specs/470-dev-platform-plugin/acceptance.md index c94a375a..bfa4e0b7 100644 --- a/specs/470-dev-platform-plugin/acceptance.md +++ b/specs/470-dev-platform-plugin/acceptance.md @@ -16,7 +16,7 @@ every row passes *and* the decoupling ratchet reads zero. | Guard | What it proves | Status | |---|---|---| -| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,167** across **14** zones, per-zone regression check | +| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,170** across **14** zones, per-zone regression check | | `middleware/test/devplatform/**` (54 files) | The behaviour itself, at unit/integration level. These **move with the plugin** and must stay green in the new repo | In place, moves in P4 | | §2 capability matrix below | Nothing is silently dropped in the move | **Written here; not yet automated** | | §3 install/uninstall | The result is genuinely installable | **Not yet built** — needs P3/P4 | diff --git a/specs/470-dev-platform-plugin/decoupling-baseline.json b/specs/470-dev-platform-plugin/decoupling-baseline.json index 5eaa2f00..4c4a2e0b 100644 --- a/specs/470-dev-platform-plugin/decoupling-baseline.json +++ b/specs/470-dev-platform-plugin/decoupling-baseline.json @@ -1,9 +1,9 @@ { - "total": 3167, + "total": 3170, "zones": { "middleware/src": 1540, - "middleware/test": 925, - "middleware/packages": 97, + "middleware/test": 926, + "middleware/packages": 99, "middleware/scripts": 8, "middleware/sidecars": 195, "middleware/migrations": 70, From 4a4f49c57feffef330e59b5f9f5ec2a0235fddf3 Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Fri, 31 Jul 2026 06:33:01 +0200 Subject: [PATCH 4/4] =?UTF-8?q?chore(470):=20resync=20C5=20with=20main=20(?= =?UTF-8?q?#552,=20#553)=20=E2=80=94=20baseline=203,167?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- specs/470-dev-platform-plugin/README.md | 2 +- specs/470-dev-platform-plugin/acceptance.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/specs/470-dev-platform-plugin/README.md b/specs/470-dev-platform-plugin/README.md index a780db4c..8cd535a4 100644 --- a/specs/470-dev-platform-plugin/README.md +++ b/specs/470-dev-platform-plugin/README.md @@ -97,7 +97,7 @@ node scripts/check-core-decoupling.mjs --update # lower the baseline ``` The ratchet counts Dev Platform references across 14 disjoint zones and **fails if the count -rises, per zone**. Baseline **3,170**. It only ever falls; raising it needs a hand-edited baseline, so +rises, per zone**. Baseline **3,167**. It only ever falls; raising it needs a hand-edited baseline, so a new coupling shows up in review instead of slipping in. That is what makes the checklist's staleness survivable — a file inventory goes stale on diff --git a/specs/470-dev-platform-plugin/acceptance.md b/specs/470-dev-platform-plugin/acceptance.md index bfa4e0b7..c94a375a 100644 --- a/specs/470-dev-platform-plugin/acceptance.md +++ b/specs/470-dev-platform-plugin/acceptance.md @@ -16,7 +16,7 @@ every row passes *and* the decoupling ratchet reads zero. | Guard | What it proves | Status | |---|---|---| -| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,170** across **14** zones, per-zone regression check | +| `scripts/check-core-decoupling.mjs` + CI job `core decoupling ratchet (#470)` | Core does not re-acquire Dev Platform references while the extraction is in flight | **In place.** Baseline **3,167** across **14** zones, per-zone regression check | | `middleware/test/devplatform/**` (54 files) | The behaviour itself, at unit/integration level. These **move with the plugin** and must stay green in the new repo | In place, moves in P4 | | §2 capability matrix below | Nothing is silently dropped in the move | **Written here; not yet automated** | | §3 install/uninstall | The result is genuinely installable | **Not yet built** — needs P3/P4 |