feat(workflows): run workflow instances in the driver, shadow mode only

The engine could queue and claim instances, but nothing executed a
claimed node. The driver now runs one node at a time: functions in the
function-runtime sandbox, component operations (registry, host backup,
verify), owner decisions, read-only privileged actions through the
executor; wait/terminal/structural nodes are recorded as they are.
A mutating privileged action (backup, apply, rollback, model repair)
is never dispatched: the node fails with 'shadow_mode'.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
homelab-updaterandClaude Opus 5 committed 2026-09-22 18:50:43 +02:00
1 parent 36b8f1261e
commit 1f166025b9
4 files changed
+389 -2

No files matched your search

+169
View File
@@ -0,0 +1,169 @@
import { describe, expect, it } from 'vitest';
import { openDatabase } from '../infra/sqlite/open.ts';
import { createSqliteWorkflowCatalogStore } from '../infra/sqlite/workflow-catalog-store.ts';
import { createSqliteWorkflowFunctionCatalogStore } from '../infra/sqlite/workflow-function-catalog-store.ts';
import { createSqlitePrivilegedActionStore } from '../infra/sqlite/privileged-action-store.ts';
import { createSqliteAppConfigStore } from '../infra/sqlite/app-config-store.ts';
import { createSqliteSkillStore } from '../infra/sqlite/skill-store.ts';
import { createSqliteWorkflowInstanceStore } from '../infra/sqlite/workflow-instance-store.ts';
import { createSqliteModelProfileStore } from '../infra/sqlite/model-profile-store.ts';
import { createIsolatedFunctionRuntime } from '../infra/runtime/isolated-function-runtime.ts';
import { composeWorkflowExtensionRegistries } from '../domain/workflow-graph.ts';
import { createFunctionExtensionRegistry } from '../domain/workflow-function-extension.ts';
import { createPrivilegedActionExtensionRegistry } from '../domain/workflow-privileged-action-extension.ts';
import { createOwnerDecisionExtensionRegistry } from '../domain/owner-decision-extension.ts';
import { createComponentOperationExtensionRegistry } from '../domain/workflow-component-operation-extension.ts';
import { buildComponentUpdateSnapshot } from '../domain/workflow-snapshot-builder.ts';
import type { PrivilegedActionExecutor } from '../domain/ports.ts';
import type { PrivilegedActionRequest, PrivilegedActionResult } from '../domain/privileged-action.ts';
import type { JsonValue } from '../domain/steps.ts';
import { FakeClock } from '../testing/fake-clock.ts';
import { createPrivilegedActionDispatcher } from './privileged-action-dispatcher.ts';
import { ensureAppWorkflows, ensureUpdatePolicy } from './workflow-seed.ts';
import { startShadowRun } from './workflow-run.ts';
import { createWorkflowRunner, type ComponentOperationRunner } from './workflow-runner.ts';
const now = '2026-09-22T12:00:00.000Z';
const target = { component: 'jackett', project: 'jackett', service: 'jackett' };
const container = { state: 'running', health: 'healthy', image: 'linuxserver/jackett:version-v0.22.0' };
const observed = { ...target, manifestSha256: 'a'.repeat(64), envSha256: 'b'.repeat(64), image: container.image, container };
/** Docker stand-in: answers only read-only actions and records every request it gets. */
function fakeDocker() {
const requests: PrivilegedActionRequest[] = [];
const executor: PrivilegedActionExecutor = {
execute(request) {
requests.push(request);
if (request.action === 'docker.inspect') return Promise.resolve({ action: 'docker.inspect', version: 2, ...observed } as PrivilegedActionResult);
if (request.action === 'docker.preflight') {
return Promise.resolve({
action: 'docker.preflight', version: 2, ...observed,
storage: { path: '/var/lib/docker', availableBytes: 50_000_000_000, requiredFreeBytes: 2_000_000_000 }, mounts: [],
} as PrivilegedActionResult);
}
return Promise.reject(new Error(`unexpected ${request.action}`));
},
};
return { executor, requests };
}
/** Registry and host stand-in: a patch update is available and the host backup is fresh. */
function fakeOperations(tags: 'candidate_within_auto' | 'candidate_above_auto' = 'candidate_within_auto'): ComponentOperationRunner {
const results: Record<string, { event: string; result: JsonValue }> = {
image_version: { event: 'version_detected', result: { event: 'version_detected', detectedVersion: '0.22.0', imageVersionState: 'detected' } },
image_tags: { event: tags, result: { event: tags, current: '0.22.0', candidate: 'version-v0.22.1', change: tags === 'candidate_within_auto' ? 'patch' : 'major' } },
release_notes: { event: 'no_source', result: { sources: [] } },
image_resolve: {
event: 'resolved',
result: { event: 'resolved', current: '0.22.0', candidate: 'version-v0.22.1', pinned: `linuxserver/jackett:version-v0.22.1@sha256:${'e'.repeat(64)}`, versionLabel: '0.22.1' },
},
host_backup: { event: 'ok', result: { event: 'ok', hostBackup: { ok: true, file: 'vzdump-103.vma.zst', ageSeconds: 120 } } },
};
return { execute: (operation) => Promise.resolve(results[operation] ?? { event: 'unexpected', result: null }) };
}
async function setup(operations: ComponentOperationRunner = fakeOperations()) {
const db = openDatabase(':memory:');
const clock = new FakeClock(now);
const workflowCatalog = createSqliteWorkflowCatalogStore(db);
const functionCatalog = createSqliteWorkflowFunctionCatalogStore(db);
const privilegedActionStore = createSqlitePrivilegedActionStore(db);
const appConfigStore = createSqliteAppConfigStore(db);
const skillStore = createSqliteSkillStore(db);
const modelProfileStore = createSqliteModelProfileStore(db);
ensureUpdatePolicy(privilegedActionStore, now);
const extensions = composeWorkflowExtensionRegistries(
createFunctionExtensionRegistry(functionCatalog),
createPrivilegedActionExtensionRegistry(privilegedActionStore.getLatestPolicy()?.policy),
createOwnerDecisionExtensionRegistry(),
createComponentOperationExtensionRegistry(),
);
appConfigStore.upsert({
app: 'jackett', project: 'jackett', enabled: true, position: 1, service: 'jackett', imageRepo: 'linuxserver/jackett',
tagPattern: 'version-*', auto: 'patch', httpUrl: null, expectStatus: 200, releaseNotesExecutor: 'script',
migrateExecutor: null, promptHint: null, postUpgradeExecutor: null, notes: null, apply: true,
});
skillStore.upsert({ project: 'jackett', content: 'Jackett torrent proxy', updatedAt: now });
// The real sandbox: the template functions run in bwrap exactly as on VM103.
const functionRuntime = createIsolatedFunctionRuntime();
const seeded = await ensureAppWorkflows({ workflowCatalog, functionCatalog, functionRuntime, extensions, newId: () => crypto.randomUUID(), now: () => now }, ['jackett']);
expect(seeded.failures).toEqual([]);
const instanceStore = createSqliteWorkflowInstanceStore(db, extensions);
const docker = fakeDocker();
const runner = createWorkflowRunner({
instanceStore,
functionCatalog,
functionRuntime,
componentOperations: operations,
privilegedActions: createPrivilegedActionDispatcher({ store: privilegedActionStore, executor: docker.executor, clock, appConfig: appConfigStore }),
policyStore: privilegedActionStore,
clock,
});
const runDeps = { workflowCatalog, functionCatalog, privilegedActionStore, appConfigStore, skillStore, modelProfileStore, instanceStore, newId: () => 'run-1', now: () => now };
return { runner, instanceStore, docker, runDeps };
}
async function runUntilIdle(runner: ReturnType<typeof createWorkflowRunner>) {
const nodes: string[] = [];
for (let i = 0; i < 50; i++) {
const tick = await runner.tick();
if (tick.kind === 'idle') return nodes;
if (tick.kind === 'finalize_failed') throw new Error(`finalize failed at ${tick.nodeId}: ${tick.reason}`);
nodes.push(tick.outcome.type === 'failure' ? `${tick.nodeId}!${JSON.stringify(tick.outcome.error)}` : tick.nodeId);
}
throw new Error('runner did not go idle');
}
describe('workflow runner', () => {
it('runs a shadow run of the update workflow to its dry-run end without touching the app', async () => {
const { runner, instanceStore, docker, runDeps } = await setup();
expect(startShadowRun(runDeps, 'jackett').ok).toBe(true);
const nodes = await runUntilIdle(runner);
expect(nodes).toEqual(['observe', 'image_version', 'image_tags', 'image_resolve', 'assess_dry_run', 'terminal_dry_run']);
expect(instanceStore.get('run-1')?.projection.status).toBe('succeeded');
expect(docker.requests.map((request) => request.action)).toEqual(['docker.inspect']);
});
it('stops at an owner gate and continues after the decision', async () => {
const { runner, instanceStore, runDeps } = await setup(fakeOperations('candidate_above_auto'));
startShadowRun(runDeps, 'jackett');
expect(await runUntilIdle(runner)).toEqual(['observe', 'image_version', 'image_tags', 'release_notes', 'gate_above_auto']);
expect(instanceStore.get('run-1')?.projection.status).toBe('waiting');
const instance = instanceStore.get('run-1')!;
const decided = instanceStore.submitOwnerDecision({
decision: {
schemaVersion: 1, decisionId: crypto.randomUUID(), signal: 'owner-decision.v1', instanceId: 'run-1', waitNodeId: 'gate_above_auto',
action: 'proceed', note: null,
},
expectedLifecycleSeq: instance.projection.seq,
expectedExecutionSeq: instanceStore.getReplayStatus('run-1').replaySeq,
now,
});
expect(decided.ok).toBe(true);
expect(await runUntilIdle(runner)).toEqual(['resolve_above_auto', 'image_resolve', 'assess_dry_run', 'terminal_dry_run']);
expect(instanceStore.get('run-1')?.projection.status).toBe('succeeded');
});
it('never dispatches a mutating action, even when a run is not a dry run', async () => {
const { runner, instanceStore, docker, runDeps } = await setup();
const policy = runDeps.privilegedActionStore.getLatestPolicy()!;
const snapshot = buildComponentUpdateSnapshot(
{ componentId: 'jackett', rootWorkflowId: 'update-jackett', rootWorkflowVersion: 1, policyVersion: policy.policyVersion, policyRevision: policy.revision, now },
runDeps,
);
if (!snapshot.ok) throw new Error(snapshot.reason);
instanceStore.create({ id: 'real', workflowId: 'update-jackett', workflowVersion: 1, snapshot: snapshot.snapshot, dueAt: now, now });
instanceStore.initializeExecution({ instanceId: 'real', input: { ...target, inspectRequest: target, preflightRequest: target, backupRequest: { project: 'jackett' }, dryRun: false }, now });
const nodes = await runUntilIdle(runner);
expect(nodes.at(-1)).toMatch(/^backup!.*shadow_mode/);
expect(instanceStore.get('real')?.projection.status).toBe('failed');
expect(docker.requests.map((request) => request.action)).toEqual(['docker.inspect', 'docker.preflight']);
});
});
+156
View File
@@ -0,0 +1,156 @@
import type { Clock, FunctionTestExecutor, PrivilegedActionStore, WorkflowExecutionClaim, WorkflowFunctionCatalogStore, WorkflowInstanceStore } from '../domain/ports.ts';
import type { JsonValue } from '../domain/steps.ts';
import type { ExecutionFinalizeOutcome } from '../domain/workflow-execution-ledger.ts';
import { resolveOwnerDecision } from '../domain/owner-decision.ts';
import { ComponentOperationNodeConfigV1, type ComponentOperationId } from '../domain/workflow-component-operation.ts';
import { FunctionNodeConfig, FunctionNodeConfigV2, FunctionOutcome } from '../domain/workflow-function.ts';
import { resultEvent } from '../domain/workflow-privileged-action-extension.ts';
import { describeError } from '../domain/mask.ts';
import type { DispatchPrivilegedActionInput, DispatchPrivilegedActionResult } from './privileged-action-dispatcher.ts';
export interface ComponentOperationRunner {
execute(operation: ComponentOperationId, version: number, input: unknown, signal?: AbortSignal): Promise<{ readonly event: string; readonly result: JsonValue }>;
}
export interface WorkflowRunnerDeps {
readonly instanceStore: WorkflowInstanceStore;
readonly functionCatalog: WorkflowFunctionCatalogStore;
readonly functionRuntime: FunctionTestExecutor;
readonly componentOperations: ComponentOperationRunner;
readonly privilegedActions: { dispatch(input: DispatchPrivilegedActionInput): Promise<DispatchPrivilegedActionResult> };
readonly policyStore: PrivilegedActionStore;
readonly clock: Clock;
}
export type WorkflowTickResult =
| { readonly kind: 'idle' }
| { readonly kind: 'finalized'; readonly instanceId: string; readonly nodeId: string; readonly outcome: ExecutionFinalizeOutcome }
| { readonly kind: 'finalize_failed'; readonly instanceId: string; readonly nodeId: string; readonly reason: string };
const failure = (code: string, message: string): ExecutionFinalizeOutcome => ({ type: 'failure', error: { code, message } });
/**
* Runs workflow instances one node at a time: claims the next due node, executes it, records the outcome. Shadow
* mode only for now (docs/IMPLEMENTATION.md, stage 8): a mutating privileged action (backup, apply, rollback, model
* repair) is never dispatched, and the node fails instead. The update template's dry run never reaches one.
*/
export function createWorkflowRunner(deps: WorkflowRunnerDeps) {
async function execute(claim: WorkflowExecutionClaim, signal?: AbortSignal): Promise<ExecutionFinalizeOutcome> {
const command = claim.command;
switch (command.type) {
case 'wait':
return { type: 'wait', signal: command.signal };
case 'terminal':
return { type: 'terminal', status: command.status, result: command.result };
case 'structural':
return { type: 'continue' };
case 'attempt':
return command.node.type === 'emit' ? { type: 'event', event: command.node.event } : failure('unsupported_node', `node type ${command.node.type} has no executor`);
case 'extension':
return executeExtension(claim, command.node.kind, command.node.config, signal);
case 'privilegedAction':
return executePrivilegedAction(claim, command.action, command.version, command.request, signal);
}
}
async function executeExtension(claim: WorkflowExecutionClaim, kind: string, config: JsonValue, signal?: AbortSignal): Promise<ExecutionFinalizeOutcome> {
const input = claim.attempt.input;
if (kind === 'function') {
const parsed = FunctionNodeConfigV2.safeParse(config);
const legacy = FunctionNodeConfig.safeParse(config);
const ref = parsed.success ? parsed.data : legacy.success ? legacy.data : null;
if (ref === null) return failure('invalid_config', 'function node config is invalid');
const fn = deps.functionCatalog.getPublishedVersion(ref.functionId, ref.version);
if (fn === null) return failure('function_not_found', `function ${ref.functionId}@${String(ref.version)} is not published`);
const outcome = FunctionOutcome.safeParse(await deps.functionRuntime.execute({
language: fn.document.language, source: fn.document.source, capabilities: fn.document.capabilities, input,
caseId: `${claim.id}#${String(claim.attempt.ordinal)}`,
}));
if (!outcome.success) return failure('invalid_function_outcome', 'the function returned no valid {event, summary, data}');
if (outcome.data.event === 'error') return failure('function_error', outcome.data.summary);
return { type: 'event', event: outcome.data.event, result: outcome.data };
}
if (kind === 'componentOperation') {
const parsed = ComponentOperationNodeConfigV1.safeParse(config);
if (!parsed.success) return failure('invalid_config', 'component operation node config is invalid');
const done = await deps.componentOperations.execute(parsed.data.operation, parsed.data.version, input, signal);
return { type: 'event', event: done.event, result: done.result };
}
if (kind === 'ownerDecision') {
// The decision lives in the execution projection (the lifecycle projection on the claim does not carry it).
const position = deps.instanceStore.getExecutionPosition(claim.id);
const resolved = resolveOwnerDecision(config, position?.schemaVersion === 2 ? position.ownerDecision : null);
return resolved.ok ? { type: 'event', event: resolved.event } : failure('owner_decision_unresolved', 'no owner decision matches this node');
}
return failure('unsupported_extension', `extension ${kind} has no executor`);
}
async function executePrivilegedAction(claim: WorkflowExecutionClaim, action: string, version: number, request: JsonValue, signal?: AbortSignal): Promise<ExecutionFinalizeOutcome> {
const policy = deps.policyStore.getLatestPolicy()?.policy;
const rule = policy?.rules.find((item) => item.action === action && item.version === version);
if (rule === undefined) return failure('action_not_allowed', `${action}@${String(version)} is not in the policy`);
if (rule.class === 'mutating') return failure('shadow_mode', `${action} changes the app; shadow runs never dispatch it`);
const dispatched = await deps.privilegedActions.dispatch({
instanceId: claim.id,
expectedExecutionSeq: deps.instanceStore.getReplayStatus(claim.id).replaySeq,
nodeAttemptOrdinal: claim.attempt.ordinal,
nodeAttemptToken: claim.attempt.token,
policy,
request,
componentId: typeof claim.command === 'object' && 'componentId' in claim.command && typeof claim.command.componentId === 'string' ? claim.command.componentId : undefined,
startedAt: deps.clock.now().toISOString(),
...(signal === undefined ? {} : { signal }),
});
if (!dispatched.ok) return failure(dispatched.reason, `${action} failed: ${dispatched.reason}`);
const result = dispatched.invocation.result;
if (result === null) return failure('no_result', `${action} returned no result`);
return { type: 'event', event: resultEvent(result), result: result as JsonValue };
}
return {
/** Claims and runs at most one node. */
async tick(signal?: AbortSignal): Promise<WorkflowTickResult> {
const now = deps.clock.now().toISOString();
const claim = deps.instanceStore.claimNext(now);
if (claim === null || !('attempt' in claim)) return { kind: 'idle' };
let outcome: ExecutionFinalizeOutcome;
try {
outcome = await execute(claim, signal);
} catch (cause) {
outcome = failure('executor_failed', describeError(cause));
}
const finalized = deps.instanceStore.finalizeExecution({
instanceId: claim.id,
expectedLifecycleSeq: claim.projection.seq,
expectedExecutionSeq: deps.instanceStore.getReplayStatus(claim.id).replaySeq,
token: claim.attempt.token,
outcome,
now: deps.clock.now().toISOString(),
});
return finalized.ok
? { kind: 'finalized', instanceId: claim.id, nodeId: claim.attempt.nodeId, outcome }
: { kind: 'finalize_failed', instanceId: claim.id, nodeId: claim.attempt.nodeId, reason: finalized.reason };
},
};
}
/** Runs nodes back to back while there is work, and polls every `idleMs` otherwise, until `signal` aborts. */
export async function runWorkflowLoop(
runner: ReturnType<typeof createWorkflowRunner>,
options: { readonly signal: AbortSignal; readonly idleMs: number; readonly onTick: (result: WorkflowTickResult) => void },
): Promise<void> {
while (!options.signal.aborted) {
let result: WorkflowTickResult;
try {
result = await runner.tick(options.signal);
} catch (cause) {
result = { kind: 'finalize_failed', instanceId: '', nodeId: '', reason: describeError(cause) };
}
options.onTick(result);
if (result.kind === 'finalized') continue;
await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, options.idleMs);
options.signal.addEventListener('abort', () => { clearTimeout(timer); resolve(); }, { once: true });
});
}
}
@@ -104,7 +104,8 @@ function eventsFor(action: string, version: number): readonly string[] {
return ['completed'];
}
function resultEvent(result: ActionResult): string {
/** The route a privileged-action result takes; the executor uses it to report the event the reducer expects. */
export function resultEvent(result: ActionResult): string {
if (result.action === 'docker.apply') {
// A diff wider than the intended image change (DESIGN.md: "diff szerszy niż obraz → STOP") always
// wins over the normal dry-run/applied/verify_required routing, on both the dry run and the real apply.
+62 -1
View File
@@ -19,6 +19,19 @@ import { createSqliteAppConfigStore } from '../infra/sqlite/app-config-store.ts'
import { createSqliteSkillStore } from '../infra/sqlite/skill-store.ts';
import { createSqliteStepStore } from '../infra/sqlite/step-store.ts';
import { systemClock } from '../infra/system-clock.ts';
import { createComponentOperationExecutor } from '../app/component-operation-executor.ts';
import { createPrivilegedActionDispatcher } from '../app/privileged-action-dispatcher.ts';
import { createWorkflowRunner, runWorkflowLoop } from '../app/workflow-runner.ts';
import { composeWorkflowExtensionRegistries } from '../domain/workflow-graph.ts';
import { createFunctionExtensionRegistry } from '../domain/workflow-function-extension.ts';
import { createPrivilegedActionExtensionRegistry } from '../domain/workflow-privileged-action-extension.ts';
import { createOwnerDecisionExtensionRegistry } from '../domain/owner-decision-extension.ts';
import { createComponentOperationExtensionRegistry } from '../domain/workflow-component-operation-extension.ts';
import { createExecutorBackedPrivilegedActionExecutor } from '../infra/privileged-action-executor-adapter.ts';
import { createFunctionRuntimeClient } from '../infra/runtime/function-runtime-client.ts';
import { createSqlitePrivilegedActionStore } from '../infra/sqlite/privileged-action-store.ts';
import { createSqliteWorkflowFunctionCatalogStore } from '../infra/sqlite/workflow-function-catalog-store.ts';
import { createSqliteWorkflowInstanceStore } from '../infra/sqlite/workflow-instance-store.ts';
import { hasDemoData, missingApps, parseAppsFile } from './apps-file.ts';
import { loadConfig, secretValues } from './config.ts';
import { missingSkills, readSkillFiles } from './skills-file.ts';
@@ -113,5 +126,53 @@ const driver = createDriver({
const shutdown = new AbortController();
for (const signal of ['SIGINT', 'SIGTERM'] as const) process.once(signal, () => shutdown.abort());
await driver.run(shutdown.signal);
// Workflow instances (the new engine), started by hand from the web app. Shadow runs only: the runner never dispatches
// a mutating action; real updates stay with the step chain above (docs/IMPLEMENTATION.md, stage 8).
const functionRuntimeUrl = process.env.FUNCTION_RUNTIME_URL;
const workflows = config.demo || functionRuntimeUrl === undefined ? null : (() => {
const functionCatalog = createSqliteWorkflowFunctionCatalogStore(db);
const policyStore = createSqlitePrivilegedActionStore(db);
const extensions = composeWorkflowExtensionRegistries(
createFunctionExtensionRegistry(functionCatalog),
createPrivilegedActionExtensionRegistry(policyStore.getLatestPolicy()?.policy),
createOwnerDecisionExtensionRegistry(),
createComponentOperationExtensionRegistry(),
);
const instanceStore = createSqliteWorkflowInstanceStore(db, extensions);
const recovered = instanceStore.recoverStale(systemClock.now().toISOString(), systemClock.now().toISOString());
if (recovered > 0) logger.info('workflows.recovered', { instances: recovered });
return createWorkflowRunner({
instanceStore,
functionCatalog,
functionRuntime: createFunctionRuntimeClient({ baseUrl: functionRuntimeUrl }),
componentOperations: createComponentOperationExecutor({
registry: createHttpRegistryClient(),
hostBackup,
backupVmid: config.backupVmid,
clock: systemClock,
skills,
...(executorClient === undefined ? {} : { executor: executorClient }),
}),
privilegedActions: createPrivilegedActionDispatcher({
store: policyStore,
executor: executorClient === undefined
? { execute: () => Promise.reject(new Error('EXECUTOR_URL is not set')) }
: createExecutorBackedPrivilegedActionExecutor({ executor: executorClient }),
clock: systemClock,
appConfig,
}),
policyStore,
clock: systemClock,
});
})();
const workflowLoop = workflows === null ? Promise.resolve() : runWorkflowLoop(workflows, {
signal: shutdown.signal,
idleMs: 2000,
onTick: (result) => {
if (result.kind === 'finalized') logger.info('workflow.node', { instance: result.instanceId, node: result.nodeId, outcome: result.outcome.type, ...(result.outcome.type === 'event' ? { event: result.outcome.event } : {}), ...(result.outcome.type === 'failure' ? { error: result.outcome.error } : {}) });
if (result.kind === 'finalize_failed') logger.error('workflow.finalize_failed', { instance: result.instanceId, node: result.nodeId, reason: result.reason });
},
});
await Promise.all([driver.run(shutdown.signal), workflowLoop]);
db.close();