diff --git a/VENDORED.md b/VENDORED.md index 2610e8398..0b1111182 100644 --- a/VENDORED.md +++ b/VENDORED.md @@ -24,7 +24,7 @@ never a convenience. | Vendored path | What was copied | Upstream repo @ commit | Why not a published package | Owner | Kill date | Kill-date test | | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------ | ---------- | ----------------- | -| `apps/sidecar` | Derived from upstream's own `apps/sidecar`: 11 shared modules, of which `signing-keypair.ts` is near-verbatim and the rest (`index.ts`, `config.ts`, `tool-materialization.ts`, `workflow-run-pack-client.ts`, …) are substantially rewritten, plus workbench-only modules. A living fork, not a frozen copy, so this row carries no tree hash. | [faremeter/interchange](https://github.com/faremeter/interchange) @ `59f5e7b9` | An app is never npm-published, so no publish can cover the execution host; retired by consuming an upstream-published host, or by renewing this row deliberately | sawyer | 2026-09-14 | `check:killdates` | +| `apps/sidecar` | Derived from upstream's own `apps/sidecar`: 11 shared modules, of which `signing-keypair.ts` is near-verbatim and the rest (`index.ts`, `config.ts`, `tool-materialization.ts`, `workflow-run-pack-client.ts`, …) are substantially rewritten, plus workbench-only modules. A living fork, not a frozen copy, so this row carries no tree hash. | [faremeter/interchange](https://github.com/faremeter/interchange) @ `4ed8baf4` | An app is never npm-published, so no publish can cover the execution host; retired by consuming an upstream-published host, or by renewing this row deliberately | sawyer | 2026-09-14 | `check:killdates` | | `vendor/intx/agent` | `@intx/agent` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `4ed8baf4` | npm 0.2.2 predates the folded model; retired by the next @intx npm publish covering it | sawyer | 2026-09-14 | `check:killdates` | | `vendor/intx/authz` | `@intx/authz` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `4ed8baf4` | npm 0.2.2 predates the folded model; retired by the next @intx npm publish covering it | sawyer | 2026-09-14 | `check:killdates` | | `vendor/intx/crypto` | `@intx/crypto` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `4ed8baf4` | npm 0.2.2 predates the folded model; retired by the next @intx npm publish covering it | sawyer | 2026-09-14 | `check:killdates` | @@ -65,9 +65,15 @@ records one commit rather than a mix. No published `@intx/*` version yet covers any vendored path: npm still tops out at `0.2.2`, which predates the folded model, so every row below stays vendored. -`apps/sidecar` stays pinned at `59f5e7b9`: workbench's execution host has -not yet been converted off the retired lineage (see CL-6324), so its row -records the last upstream commit its fork was reconciled against. +`apps/sidecar` now records `4ed8baf4`: its fork is converted onto the +closure-sourced lineage. Four modules are near-verbatim copies of upstream's +own at that commit — `workflow-probe-handler.ts`, +`workflow-closure-materialization.ts`, `workflow-closure-apply.ts`, +`source-asset-delivery.ts` — plus `bin/workflow-probe-child`; each is adapted +only where the fork's module layout differs (the host-platform resolution +lives in this fork's `tool-materialization.ts`, and the probe child's shebang +drops upstream's `intx-src` condition, which workbench forbids). The +remaining shared modules stay substantially rewritten, as the row records. Local modifications (all `vendor/intx/*` rows): each package's exports map is repointed from the upstream `intx-src` resolve condition to direct diff --git a/apps/sidecar/bin/workflow-probe-child b/apps/sidecar/bin/workflow-probe-child new file mode 100755 index 000000000..451b21541 --- /dev/null +++ b/apps/sidecar/bin/workflow-probe-child @@ -0,0 +1,21 @@ +#!/usr/bin/env bun +// One-shot workflow-probe child. The sidecar host spawns this by path +// (`Bun.spawn([binaryPath])`). +// +// The child reads the materialized package dir and IPC anchors from its +// fresh env, evaluates the workflow entry behind the airlock, and ships +// one HMAC-signed result frame on stdout. An evaluation failure is shipped +// as an `ok: false` frame (handled inside the runner), so a throw reaching +// here is a pre-evaluation defect (bad env, unwritable stdout) that exits +// non-zero -- the host then reaps and answers `workflow.probe.error`. +import { runWorkflowProbeChildFromProcessEnv } from "../src/workflow-probe-handler"; + +try { + await runWorkflowProbeChildFromProcessEnv(); + process.exit(0); +} catch (err) { + process.stderr.write( + `workflow probe child failed: ${err instanceof Error ? err.message : String(err)}\n`, + ); + process.exit(1); +} diff --git a/apps/sidecar/src/index.ts b/apps/sidecar/src/index.ts index f4b53bfe9..13c13f18d 100644 --- a/apps/sidecar/src/index.ts +++ b/apps/sidecar/src/index.ts @@ -33,7 +33,13 @@ import { createTarballCache } from "@intx/tool-packaging"; import { hexEncode } from "@intx/types"; import { readSidecarConfig } from "./config"; -import { DEFAULT_TOOL_REGISTRIES_JSON } from "./tool-materialization"; +import { + DEFAULT_TOOL_REGISTRIES_JSON, + parseToolRegistries, +} from "./tool-materialization"; +import { createWorkflowProbeExecutor } from "./workflow-probe-handler"; +import { createWorkflowClosureMaterializer } from "./workflow-closure-materialization"; +import { MAX_INLINE_ASSET_PAYLOAD_BYTES } from "./source-asset-delivery"; import { createDefaultHarnessBuilder } from "./default-harness"; import { createHubLinkWatchdog } from "./hub-link-watchdog"; import { drainWithTimeout } from "./shutdown"; @@ -181,6 +187,27 @@ if (config.tmpdir !== undefined) { // `canBuildSource` predicate against the one adapter registry. const buildHarness = createDefaultHarnessBuilder({ adapters }); +// Airlocked workflow-probe executor, assembled here and injected through the +// orchestrator so the sidecar answers `workflow.probe.request` with a real +// inert projection and its wire hash instead of the hub-link's rejecting +// placeholder. The materializer lays a probe frame's frozen closure out under +// a per-probe scratch dir (rooted in the sidecar data dir so it shares that +// dir's lifecycle); the executor spawns the one-shot child that evaluates the +// workflow entry against it. A probe delivers its source assets inline in one +// frame, capped by the shared inline-payload bound. +const workflowProbeExecutor = createWorkflowProbeExecutor({ + materialize: createWorkflowClosureMaterializer({ + cacheRoot: CACHE_ROOT, + cacheMaxBytes: CACHE_MAX_BYTES, + registryMaxTarballBytes: REGISTRY_MAX_TARBALL_BYTES, + maxAssetPayloadBytes: MAX_INLINE_ASSET_PAYLOAD_BYTES, + registries: parseToolRegistries( + config.toolRegistries ?? DEFAULT_TOOL_REGISTRIES_JSON, + ), + scratchRoot: path.join(config.dataDir, "workflow-probe", "closures"), + }), +}); + const watchdogLog = getLogger(["sidecar", "hub-link-watchdog"]); const watchdog = createHubLinkWatchdog({ stallDeadlineMs: 60_000, @@ -217,6 +244,7 @@ const orchestrator = createSidecarOrchestrator({ // supervisor spawns, against the unwrapped substrate so the restore is // never echoed back to the Hub as a new sidecar-authored update. applyWorkflowRunPack: restoreWorkflowRunPack, + workflowProbeExecutor, // Called from every connection's open handler -- the watchdog's // aliveness signal -- and from the close path, which immediately // re-schedules a reconnect that re-arms the deadline. diff --git a/apps/sidecar/src/source-asset-delivery.ts b/apps/sidecar/src/source-asset-delivery.ts new file mode 100644 index 000000000..dd2080b13 --- /dev/null +++ b/apps/sidecar/src/source-asset-delivery.ts @@ -0,0 +1,215 @@ +// Sidecar-side delivery of a workflow closure's source assets. +// +// A `WorkflowSourceAssetMount` carries a git pack for one hub asset. How that +// pack is materialized depends on how the closure references the asset: +// - a tarball-format entry reads a `.tgz` blob from a plain-file checkout +// (`applyAssetPack`), keyed by an `assetId -> mountPath` map; and +// - a source-format entry checks a subtree out of the git objects, so the +// pack is indexed into a RETAINED `.git` (a "gitDir"), keyed by an +// `assetId -> gitDir` map the loader hands `materializeGitEntry`. +// One asset can be referenced both ways; a source-format workflow closure +// references only source entries, so it produces only gitDirs. + +import fsp from "node:fs/promises"; +import path from "node:path"; + +import { getLogger } from "@intx/log"; +import { base64Decode } from "@intx/types"; +import { applyAssetPack } from "@intx/hub-agent"; +import { + DEFAULT_PACK_MATERIALIZATION_LIMITS, + indexPackIntoGitDir, +} from "@intx/storage-isogit/node"; +import type { WorkflowSourceAssetMount } from "@intx/types/sidecar"; +import type { ToolPackageManifest } from "@intx/types/tool-packages"; + +const logger = getLogger(["sidecar", "source-asset-delivery"]); + +const SAFE_ASSET_ID = /^[a-zA-Z0-9_.-]+$/; + +/** + * Index a delivered asset pack into `gitDir` and RETAIN the object store, so a + * source subtree can be checked out from it. Builds into a sibling temp `.git` + * and RENAMES it into place, so the durable store is complete-or-absent: a crash + * mid-materialization leaves only the temp, never a partial `gitDir` that the + * dir-exists check `resolveDeploymentAssetMounts` runs on restore would trust. + * The rename is same-filesystem (the temp is a sibling under `gitDir`'s parent). + * + * On a rename conflict (a stale `gitDir` from a torn prior attempt) the freshly + * built store wins: the existing dir is removed and the temp renamed over it, so + * a re-delivery at a new commit never keeps the old content. A secondary rm + * failure is logged so it does not silently mask state; the primary error is + * rethrown. + */ +export async function indexAssetPackIntoGitDir(args: { + pack: Uint8Array; + commitSha: string; + gitDir: string; +}): Promise { + const { pack, commitSha, gitDir } = args; + const parent = path.dirname(gitDir); + await fsp.mkdir(parent, { recursive: true }); + const tempDir = await fsp.mkdtemp(path.join(parent, ".indexing-")); + + const cleanupTemp = async (): Promise => { + await fsp.rm(tempDir, { recursive: true, force: true }).catch((rmErr) => { + const rmMsg = rmErr instanceof Error ? rmErr.message : String(rmErr); + logger.warn`source-asset temp gitdir cleanup failed at ${tempDir}: ${rmMsg}`; + }); + }; + + try { + await indexPackIntoGitDir( + tempDir, + pack, + commitSha, + DEFAULT_PACK_MATERIALIZATION_LIMITS, + ); + } catch (err) { + await cleanupTemp(); + throw err; + } + + try { + await fsp.rename(tempDir, gitDir); + } catch (err) { + // Only a "destination already exists" failure means a torn prior attempt we + // may supersede; any other rename error (EXDEV, EACCES, ENOSPC, EIO) must + // NOT destroy a possibly-good prior store -- clean up only the fresh temp + // and surface it. + if (!isDestinationExistsError(err)) { + await cleanupTemp(); + throw err; + } + // The final path already holds a store (a torn prior attempt): rebuild + // wins, so drop the stale store and rename the fresh one over it. + await fsp.rm(gitDir, { recursive: true, force: true }); + try { + await fsp.rename(tempDir, gitDir); + } catch (retryErr) { + await cleanupTemp(); + throw retryErr; + } + } +} + +/** Whether `err` is a rename failure caused by a non-empty destination. */ +function isDestinationExistsError(err: unknown): boolean { + if (err === null || typeof err !== "object" || !("code" in err)) return false; + const code = String(err.code); + return code === "ENOTEMPTY" || code === "EEXIST" || code === "EISDIR"; +} + +/** + * The materialization format(s) each asset id is referenced with in `closure`. + * An asset with any tarball entry needs a plain-file checkout; an asset with + * any source entry needs a gitDir. + */ +export function assetReferenceFormats( + closure: ToolPackageManifest, +): Map { + const byAsset = new Map(); + for (const entry of closure.entries) { + if (entry.source.kind !== "asset") continue; + const existing = byAsset.get(entry.source.assetId) ?? { + tarball: false, + source: false, + }; + if (entry.source.package.format === "tarball") existing.tarball = true; + else existing.source = true; + byAsset.set(entry.source.assetId, existing); + } + return byAsset; +} + +/** The absolute gitDir a source asset's objects are indexed into. */ +export function sourceAssetGitDir(gitDirRoot: string, assetId: string): string { + // Reject an all-dots assetId (".", "..", ...) before the join. SAFE_ASSET_ID + // permits "." as a character, so a bare ".." would otherwise escape the + // per-asset dir (`path.join(root, "..")` is root's parent) and "." would + // resolve to the shared root itself. Mirrors `applyAssetPack`'s all-dots + // segment guard. + if (!SAFE_ASSET_ID.test(assetId) || /^\.+$/.test(assetId)) { + throw new Error( + `source-asset delivery: unsafe assetId ${JSON.stringify(assetId)}`, + ); + } + return path.join(gitDirRoot, assetId); +} + +/** + * The single cap on the total inline (base64) source-asset payload a workflow + * closure may deliver in one frame. Both the probe and the deploy pass this to + * `materializeWorkflowAssets`; a git-sourced asset that grows past it is the + * signal to move that path's asset delivery to a streamed transfer. One + * constant so the two paths cannot drift. + */ +export const MAX_INLINE_ASSET_PAYLOAD_BYTES = 32 * 1024 * 1024; + +/** + * Materialize a workflow closure's delivered source assets: for each asset, + * check out plain tarball files under `assetRoot` (if the closure has tarball + * entries for it) and/or index the pack into a gitDir under `gitDirRoot` (if it + * has source entries). Returns both maps for the loader. + */ +export async function materializeWorkflowAssets(args: { + assets: readonly WorkflowSourceAssetMount[]; + closure: ToolPackageManifest; + assetRoot: string; + gitDirRoot: string; + maxAssetPayloadBytes: number; +}): Promise<{ + assetMounts: ReadonlyMap; + gitDirs: ReadonlyMap; +}> { + const formats = assetReferenceFormats(args.closure); + const assetMounts = new Map(); + const gitDirs = new Map(); + const seen = new Set(); + let totalPayloadBytes = 0; + for (const asset of args.assets) { + totalPayloadBytes += asset.pack.length; + if (totalPayloadBytes > args.maxAssetPayloadBytes) { + throw new Error( + `workflow source-asset materialization: inline asset payload exceeds the ${String(args.maxAssetPayloadBytes)}-byte cap`, + ); + } + if (seen.has(asset.assetId)) { + throw new Error( + `workflow source-asset materialization: asset ${JSON.stringify(asset.assetId)} is delivered more than once`, + ); + } + seen.add(asset.assetId); + // The frame delivers one mount per asset the closure references, so a + // delivered asset with no closure entry is a hub/frame inconsistency. + // Fail loud rather than silently ignore it (while still counting its + // payload toward the cap above). + const refs = formats.get(asset.assetId); + if (refs === undefined) { + throw new Error( + `workflow source-asset materialization: asset ${JSON.stringify(asset.assetId)} is delivered but referenced by no closure entry`, + ); + } + const pack = base64Decode(asset.pack); + if (refs.tarball) { + await applyAssetPack({ + workspaceRoot: args.assetRoot, + mountPath: asset.mountPath, + pack, + ref: asset.ref, + commitSha: asset.commitSha, + }); + assetMounts.set(asset.assetId, asset.mountPath); + } + if (refs.source) { + const gitDir = sourceAssetGitDir(args.gitDirRoot, asset.assetId); + await indexAssetPackIntoGitDir({ + pack, + commitSha: asset.commitSha, + gitDir, + }); + gitDirs.set(asset.assetId, gitDir); + } + } + return { assetMounts, gitDirs }; +} diff --git a/apps/sidecar/src/tool-materialization.ts b/apps/sidecar/src/tool-materialization.ts index 571e12b0b..aad9a8b08 100644 --- a/apps/sidecar/src/tool-materialization.ts +++ b/apps/sidecar/src/tool-materialization.ts @@ -21,6 +21,7 @@ import { type } from "arktype"; import { type AnnotatedPluginFactory } from "@intx/agent"; import { getLogger } from "@intx/log"; import { + type HostPlatform, type LoadedToolFactory, type LoadedToolPackage, type RegistryConfig, @@ -173,6 +174,18 @@ function assertKnownHostArch(arch: NodeJS.Architecture): void { } } +/** + * The host platform token pair the `@intx/tool-packaging` loader filters + * manifest entries against, asserted against npm's `os`/`cpu` namespaces + * first. Shared by the per-step tool apply below and the workflow-definition + * closure materializer so both filter against one resolution. + */ +export function resolveHostPlatform(): HostPlatform { + assertKnownHostPlatform(process.platform); + assertKnownHostArch(process.arch); + return { os: process.platform, cpu: process.arch }; +} + // Sentinel `previousDeployId` for an instance that has never applied // a deploy successfully. Encoded as a literal string so the value // travels through `applyAtomic`'s `ApplyAtomicFailure.previousDeployId` @@ -328,12 +341,10 @@ export async function materializeToolPackages(args: { rootDir: args.cacheRoot, maxBytes: args.cacheMaxBytes, }); - assertKnownHostPlatform(process.platform); - assertKnownHostArch(process.arch); const loader = createToolLoader({ cache, registries: args.registries, - host: { os: process.platform, cpu: process.arch }, + host: resolveHostPlatform(), maxRegistryTarballBytes: args.registryMaxTarballBytes, }); const result = await applyAtomic({ diff --git a/apps/sidecar/src/workflow-closure-apply.ts b/apps/sidecar/src/workflow-closure-apply.ts new file mode 100644 index 000000000..76a2aacfa --- /dev/null +++ b/apps/sidecar/src/workflow-closure-apply.ts @@ -0,0 +1,232 @@ +// Deploy-side application of a code-sourced workflow's frozen closure. +// +// When a deploy frame carries a `source` (the npm registry the workflow +// definition package is published to) plus the hub's frozen dependency +// `closure` (concrete versions + integrity SRIs), the sidecar materializes +// EXACTLY that closure and evaluates the pinned code to a validated +// `WorkflowDefinition` -- rather than trusting an inline serialized +// projection. The closure is applied byte-for-byte as the hub froze it; the +// sidecar never re-resolves the pin against the registry at apply time. +// +// This is the DURABLE deploy counterpart to the airlocked install-time probe: +// it reuses the same `@intx/tool-packaging` apply machinery +// (`createTarballCache` / `createToolLoader` / `applyAtomic`) that +// `tool-materialization.ts` uses for a step's tool closure, so the fetch + +// SRI-verify + extract + `node_modules` layout is not reimplemented here. +// +// A workflow-definition package declares `interchange.workflow` (the module +// whose evaluation produces the definition), NOT `interchange.tools`. +// `applyAtomic`'s load phase imports each TOP-LEVEL package's +// `interchange.tools` entry and rejects a package that has none +// (`package.entry.missing`), so the layout manifest handed to `applyAtomic` +// carries an EMPTY `topLevel`: every entry is still materialized and laid out +// (the dependency layout walks `entries`, and each dependency resolves against +// the frozen closure), but no tool factory is imported. The workflow entry +// itself is imported by `loadWorkflowDefinitionFromClosure` -- the correct load +// site for a workflow definition -- against the materialized package directory. + +import path from "node:path"; + +import { getLogger } from "@intx/log"; +import { + type RegistryConfig, + type TarballFetcher, + applyAtomic, + createTarballCache, + createToolLoader, + storeEntryDir, +} from "@intx/tool-packaging"; +import type { ToolPackageManifest } from "@intx/types/tool-packages"; +import { getToolPackageSourceContentIdentity } from "@intx/types/tool-packages"; +import type { WorkflowDefinitionSource } from "@intx/types/workflow-sources"; +import { loadWorkflowDefinitionFromClosure } from "@intx/workflow-host"; +import type { WorkflowDefinition } from "@intx/workflow/definition"; + +const logger = getLogger(["sidecar", "workflow-closure-apply"]); + +export interface ApplyFrozenWorkflowClosureArgs { + /** Names the registry the workflow definition package is published to. */ + readonly source: WorkflowDefinitionSource; + /** + * The hub's frozen dependency closure for the definition's pin: concrete + * versions and integrity SRIs. Applied byte-for-byte; never re-resolved. + */ + readonly closure: ToolPackageManifest; + /** + * Durable per-deployment directory the closure is staged under + * (`/packages//store/...`). + */ + readonly instanceDir: string; + /** Content-addressable tarball cache root shared across applies. */ + readonly cacheRoot: string; + /** Byte cap for the tarball cache. */ + readonly cacheMaxBytes: number; + /** Byte cap for a single HTTP-registry tarball fetch. */ + readonly registryMaxTarballBytes: number; + /** Registry identifier -> URL + credentials the loader resolves entries against. */ + readonly registries: ReadonlyMap; + /** + * Workspace root `kind: "asset"` closure entries mount against. A + * registry-sourced workflow definition closure carries no asset entries, so + * this defaults to `/workspace`. + */ + readonly assetRoot?: string; + /** `assetId` -> mount path for tarball `asset` entries; empty by default. */ + readonly assetMounts?: ReadonlyMap; + /** + * `assetId` -> absolute indexed git directory for source-format `asset` + * entries. A registry- or tarball-sourced closure carries no source + * entries, so this defaults to an empty map. + */ + readonly gitDirs?: ReadonlyMap; + /** + * Test seam for tarball fetching, forwarded to `createToolLoader`. + * Production omits it and the loader fetches from the configured registry. + */ + readonly fetchTarball?: TarballFetcher; + /** + * Test seam for the workflow entry's dynamic import, forwarded to + * `loadWorkflowDefinitionFromClosure`. Production omits it and the loader + * imports the materialized entry natively. + */ + readonly importModule?: (importUrl: string) => Promise; +} + +export interface AppliedWorkflowClosure { + /** The validated definition the pinned code evaluated to. */ + readonly definition: WorkflowDefinition; + /** Directory of the materialized workflow package within the closure. */ + readonly packageDir: string; + /** The staged, never-renamed deploy directory the closure was laid out under. */ + readonly deployDir: string; +} + +/** + * Materialize a workflow definition's frozen closure durably and load the + * pinned code to a validated `WorkflowDefinition`. + * + * The closure's single top-level pin IS the workflow definition package: the + * hub resolved the closure for exactly that pin. The frozen `entries` are + * applied verbatim (concrete versions + SRIs), so no registry re-resolution + * happens at apply time. + * + * @throws if the closure does not carry exactly one top-level pin, the source + * registry is not configured on this sidecar, the apply fails (integrity + * mismatch, fetch failure, extract failure, ...), or the pinned code does not + * evaluate to exactly one valid `WorkflowDefinition`. + */ +export async function applyFrozenWorkflowClosure( + args: ApplyFrozenWorkflowClosureArgs, +): Promise { + if (args.closure.topLevel.length !== 1) { + throw new Error( + `sidecar workflow-closure apply: the frozen closure must carry exactly one top-level pin (the workflow definition package), got ${String(args.closure.topLevel.length)}`, + ); + } + const workflowPin = args.closure.topLevel[0]; + if (workflowPin === undefined) { + throw new Error( + "sidecar workflow-closure apply: the frozen closure's single top-level pin is undefined", + ); + } + + // Boundary check on the definition's source. The `registry` arm surfaces a + // missing source registry loudly before any I/O (the per-entry registry + // gates fire again inside the loader). An `asset` closure materializes its + // entries from the durable stores the caller populated: tarball entries from + // `assetMounts`, source entries from `gitDirs`; the loader fails loud + // (`asset.mount.missing` / `git.materialization.failed`) if either is absent. + // The `never` default makes a future source kind a compile error rather than + // a silent fallthrough. + switch (args.source.kind) { + case "registry": + if (!args.registries.has(args.source.registry)) { + throw new Error( + `sidecar workflow-closure apply: source registry ${JSON.stringify(args.source.registry)} is not in the sidecar registry config`, + ); + } + break; + case "asset": + break; + default: { + const _exhaustive: never = args.source; + throw new Error( + `sidecar workflow-closure apply: unhandled workflow source kind ${String(_exhaustive)}`, + ); + } + } + + const cache = createTarballCache({ + rootDir: args.cacheRoot, + maxBytes: args.cacheMaxBytes, + }); + const loader = createToolLoader({ + cache, + registries: args.registries, + host: { os: process.platform, cpu: process.arch }, + maxRegistryTarballBytes: args.registryMaxTarballBytes, + ...(args.fetchTarball !== undefined + ? { fetchTarball: args.fetchTarball } + : {}), + }); + + // Apply EXACTLY the frozen entries. `topLevel` is emptied so `applyAtomic` + // imports no `interchange.tools` module (a workflow-definition package has + // none); the full `entries` set is still materialized and laid out. + const layoutManifest: ToolPackageManifest = { + schemaVersion: args.closure.schemaVersion, + topLevel: [], + entries: args.closure.entries, + }; + + const result = await applyAtomic({ + manifest: layoutManifest, + loader, + instanceDir: args.instanceDir, + assetRoot: args.assetRoot ?? path.join(args.instanceDir, "workspace"), + assetMounts: args.assetMounts ?? new Map(), + gitDirs: args.gitDirs ?? new Map(), + attemptId: crypto.randomUUID(), + // This apply stands alone per deployment: there is no prior deploy under + // `instanceDir` to retain, so the sentinel disables the retention window. + previousDeployId: "none", + newDeployId: crypto.randomUUID(), + }); + if (result.status === "failed") { + throw new Error( + `sidecar workflow-closure apply: materializing the frozen closure for ${workflowPin.name}@${workflowPin.version} failed (${result.category}): ${result.message}`, + ); + } + + const packageDir = storeEntryDir( + path.join(result.deployDir, "store"), + workflowPin.name, + workflowPin.version, + ); + + // The workflow package's own integrity is the natural ESM-cache-bust token: + // Node keys its module cache by resolved URL, so a re-apply of changed bytes + // under the same name@version reimports rather than resolving to the prior + // instance. + const workflowEntry = args.closure.entries.find( + (entry) => + entry.name === workflowPin.name && entry.version === workflowPin.version, + ); + + const definition = await loadWorkflowDefinitionFromClosure({ + packageDir, + ...(workflowEntry !== undefined + ? { + importCacheKey: getToolPackageSourceContentIdentity( + workflowEntry.source, + ), + } + : {}), + ...(args.importModule !== undefined + ? { importModule: args.importModule } + : {}), + }); + + logger.debug`applied frozen workflow closure ${workflowPin.name}@${workflowPin.version}: loaded definition ${definition.id}`; + return { definition, packageDir, deployDir: result.deployDir }; +} diff --git a/apps/sidecar/src/workflow-closure-materialization.ts b/apps/sidecar/src/workflow-closure-materialization.ts new file mode 100644 index 000000000..789d88db9 --- /dev/null +++ b/apps/sidecar/src/workflow-closure-materialization.ts @@ -0,0 +1,293 @@ +// Host-side materializer for a workflow-probe frame's frozen closure. +// +// The airlocked probe child (`workflow-probe-handler.ts`) evaluates a +// code-sourced workflow's `interchange.workflow` entry, but the frozen +// dependency closure it evaluates against is materialized on the sidecar +// HOST first -- fetch + SRI-verify + extract + `node_modules` layout is +// I/O, not author-code evaluation, so it stays out of the child. This +// module builds the production `MaterializeWorkflowClosure` the probe +// executor injects: it lays out the frame's frozen closure and returns +// the workflow package directory the child loads from, without importing +// any author code on the host. +// +// Layering (greybeard): the concrete materializer lives here in +// `apps/sidecar` so `@intx/workflow-host` stays free of a +// `@intx/tool-packaging` dependency and `workflow-probe-handler.ts` +// stays free of one too -- the materializer is an injected seam. The +// portable packages only see the `MaterializeWorkflowClosure` callback +// this module produces. +// +// Phases 1-2 only, no `applyAtomic`: a probe is ephemeral and inert, so +// the durable-deploy lifecycle bookkeeping (`active-deploy-id`, the +// per-deploy-id retention ladder) is the wrong semantics. The closure is +// laid out under a per-probe scratch dir that `cleanup` removes once the +// child has been reaped. `createToolLoader(...).loadManifest(...)` is +// invoked with an EMPTIED `topLevel`: the loader's phase-3 import loop +// only imports packages named in `topLevel`, so an empty `topLevel` +// fetches + extracts + lays out every closure entry (the full `entries` +// set) while importing NONE of them. That runs exactly the eval-free +// `materializeClosure` phases the probe needs, using the loader's +// production registry fetcher -- the tarball fetcher `@intx/tool-packaging` +// owns is only reachable through `createToolLoader`, so the layout is +// driven through the loader rather than by calling `materializeClosure` +// with a fetcher this package would otherwise have to build (and thereby +// reach for the npm-registry machinery that package exists to contain). + +import { promises as fs } from "node:fs"; +import path from "node:path"; + +import { type } from "arktype"; +import { getLogger } from "@intx/log"; +import { + type RegistryConfig, + type TarballFetcher, + createTarballCache, + createToolLoader, + storeEntryDir, +} from "@intx/tool-packaging"; +import { PackageJSON } from "@intx/types/package-json"; +import type { WorkflowProbeRequestFrame } from "@intx/types/sidecar"; +import type { ToolPackageManifest } from "@intx/types/tool-packages"; + +import { materializeWorkflowAssets } from "./source-asset-delivery"; +import { resolveHostPlatform } from "./tool-materialization"; +import type { + MaterializedWorkflowClosure, + MaterializeWorkflowClosure, +} from "./workflow-probe-handler"; + +const logger = getLogger(["sidecar", "workflow-closure-materialization"]); + +export interface WorkflowClosureMaterializerConfig { + /** Content-addressable tarball cache root shared across materializations. */ + readonly cacheRoot: string; + /** Byte cap for the tarball cache. */ + readonly cacheMaxBytes: number; + /** Byte cap for a single HTTP-registry tarball fetch. */ + readonly registryMaxTarballBytes: number; + /** + * Byte cap for the total base64-encoded asset payload a probe frame may + * deliver inline. Measured against the base64 wire length (a conservative + * upper bound on the decoded bytes), it is enforced before any pack is + * decoded so an oversized frame fails loud rather than being materialized. + */ + readonly maxAssetPayloadBytes: number; + /** Registry identifier -> URL + credentials the loader resolves entries against. */ + readonly registries: ReadonlyMap; + /** + * Root directory under which each probe's ephemeral closure scratch dir + * is created (one per probe, removed by the returned `cleanup`). + */ + readonly scratchRoot: string; + /** + * Test seam for tarball fetching, forwarded to `createToolLoader`. + * Production omits it and the loader fetches from the configured registry. + */ + readonly fetchTarball?: TarballFetcher; +} + +/** + * Build the production `MaterializeWorkflowClosure` the workflow-probe + * executor injects. The returned function lays out a probe frame's frozen + * closure under a fresh scratch dir and returns the workflow package + * directory plus a `cleanup` that removes the scratch dir. + * + * @throws (from the returned materializer) if the closure does not pin + * exactly one top-level package, the source registry is not configured, + * the layout fails (fetch / integrity / extract), or the frame's `entry` + * disagrees with the materialized package's `interchange.workflow`. + */ +export function createWorkflowClosureMaterializer( + config: WorkflowClosureMaterializerConfig, +): MaterializeWorkflowClosure { + const host = resolveHostPlatform(); + + return async function materialize( + frame: WorkflowProbeRequestFrame, + ): Promise { + // Gap 1: the closure's single top-level pin IS the workflow definition + // package (the hub resolved the closure for exactly that pin). Assert + // the cardinality and fail loud rather than silently picking `[0]`; a 0- + // or >1-pin closure is an incoherent request the materializer owns + // rejecting at this boundary. + const topLevel = frame.closure.topLevel; + if (topLevel.length !== 1) { + throw new Error( + `workflow-probe closure materialization: the frozen closure must pin exactly one top-level package (the workflow definition package), got ${String(topLevel.length)}`, + ); + } + const workflowPin = topLevel[0]; + if (workflowPin === undefined) { + throw new Error( + "workflow-probe closure materialization: the frozen closure's single top-level pin is undefined", + ); + } + + // Boundary check on the definition's source. The `registry` arm surfaces a + // missing source registry loudly before any I/O (the per-entry registry + // gates fire again inside the loader). An `asset` closure is materialized + // from the `assets` the frame delivers -- tarball entries from a plain-file + // mount, source entries from an indexed gitDir -- checked below. The + // `never` default makes a future source kind a compile error rather than a + // silent fallthrough. + switch (frame.source.kind) { + case "registry": + if (!config.registries.has(frame.source.registry)) { + throw new Error( + `workflow-probe closure materialization: source registry ${JSON.stringify(frame.source.registry)} is not in the sidecar registry config`, + ); + } + break; + case "asset": + break; + default: { + const _exhaustive: never = frame.source; + throw new Error( + `workflow-probe closure materialization: unhandled workflow source kind ${String(_exhaustive)}`, + ); + } + } + + const scratchDir = path.join(config.scratchRoot, crypto.randomUUID()); + await fs.mkdir(scratchDir, { recursive: true }); + const cleanup = async (): Promise => { + await fs.rm(scratchDir, { recursive: true, force: true }); + }; + + try { + const cache = createTarballCache({ + rootDir: config.cacheRoot, + maxBytes: config.cacheMaxBytes, + }); + const loader = createToolLoader({ + cache, + registries: config.registries, + host, + maxRegistryTarballBytes: config.registryMaxTarballBytes, + ...(config.fetchTarball !== undefined + ? { fetchTarball: config.fetchTarball } + : {}), + }); + + // Materialize any inline-delivered assets under the probe scratch: + // tarball entries as plain files under the workspace root, source entries + // as an indexed gitDir the loader checks subtrees out of. Registry-sourced + // closures deliver none; both maps stay empty and the loader fetches over + // HTTP. + const assetRoot = path.join(scratchDir, "workspace"); + const gitDirRoot = path.join(scratchDir, "gitdirs"); + const { assetMounts, gitDirs } = await materializeWorkflowAssets({ + assets: frame.assets ?? [], + closure: frame.closure, + assetRoot, + gitDirRoot, + maxAssetPayloadBytes: config.maxAssetPayloadBytes, + }); + + // Source-boundary check for the asset arm, the post-unpack analog of the + // registry arm's config check: the asset the definition is sourced from + // holds the workflow package, so it MUST be among the delivered assets -- + // as a gitDir for a source definition, as a mount for a tarball one. + // Surface a missing delivery here rather than as a downstream failure on + // the top-level entry. + if (frame.source.kind === "asset") { + const delivered = + frame.source.package.format === "source" + ? gitDirs.has(frame.source.assetId) + : assetMounts.has(frame.source.assetId); + if (!delivered) { + throw new Error( + `workflow-probe closure materialization: asset source ${JSON.stringify(frame.source.assetId)} was not among the delivered assets`, + ); + } + } + + // Lay out phases 1-2 only. `topLevel` is emptied so the loader's + // phase-3 loop imports nothing -- no author code is evaluated on the + // host; the airlocked child owns the single import of the workflow + // entry. The full `entries` set is still fetched, SRI-verified, + // extracted, and laid out with its `node_modules` graph. + const layoutManifest: ToolPackageManifest = { + schemaVersion: frame.closure.schemaVersion, + topLevel: [], + entries: frame.closure.entries, + }; + await loader.loadManifest({ + manifest: layoutManifest, + instanceScratchDir: scratchDir, + assetRoot, + assetMounts, + gitDirs, + }); + + const storeDir = path.join(scratchDir, "store"); + const packageDir = storeEntryDir( + storeDir, + workflowPin.name, + workflowPin.version, + ); + + await assertFrameEntryMatchesPackage(packageDir, frame.entry); + + logger.debug`materialized workflow-probe closure for ${workflowPin.name}@${workflowPin.version} at ${packageDir}`; + return { packageDir, cleanup }; + } catch (err) { + // On any failure before a closure handle is handed back, the executor + // never sees a `cleanup` to call, so the scratch dir is this function's + // to reclaim. + await cleanup(); + throw err; + } + }; +} + +/** + * Gap 2: cross-check the probe frame's `entry` against the materialized + * package's own `interchange.workflow`. The child loader reads the entry + * path from the package's `package.json`, ignoring the frame's `entry`; + * left unchecked the frame field is an input that travels but is never + * validated. Comparing them host-side -- a `package.json` read, no author + * code -- surfaces a tampered or incoherent request before the child is + * ever spawned, and fails loud on mismatch. + */ +async function assertFrameEntryMatchesPackage( + packageDir: string, + frameEntry: string, +): Promise { + const pkgJsonPath = path.join(packageDir, "package.json"); + let raw: string; + try { + raw = await fs.readFile(pkgJsonPath, "utf8"); + } catch (cause) { + throw new Error( + `workflow-probe closure materialization: cannot read package.json at ${packageDir} to cross-check the frame entry`, + { cause }, + ); + } + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (cause) { + throw new Error( + `workflow-probe closure materialization: malformed package.json at ${packageDir}`, + { cause }, + ); + } + const pkg = PackageJSON(parsed); + if (pkg instanceof type.errors) { + throw new Error( + `workflow-probe closure materialization: package.json at ${packageDir} failed validation: ${pkg.summary}`, + ); + } + const declaredEntry = pkg.interchange?.workflow; + if (declaredEntry === undefined) { + throw new Error( + `workflow-probe closure materialization: workflow package at ${packageDir} declares no "interchange.workflow" entry`, + ); + } + if (declaredEntry !== frameEntry) { + throw new Error( + `workflow-probe closure materialization: probe frame entry ${JSON.stringify(frameEntry)} does not match the materialized package's interchange.workflow ${JSON.stringify(declaredEntry)}`, + ); + } +} diff --git a/apps/sidecar/src/workflow-deployment-record.test.ts b/apps/sidecar/src/workflow-deployment-record.test.ts index 62a698e4c..f8e7fa308 100644 --- a/apps/sidecar/src/workflow-deployment-record.test.ts +++ b/apps/sidecar/src/workflow-deployment-record.test.ts @@ -20,6 +20,15 @@ const baseRecord: WorkflowDeploymentRecord = { agentAddress: "run_parked-test@example.com", definitionId: "def_1", sources: {}, + approvedWireHash: "d".repeat(64), + sourceRef: { + source: { kind: "registry", registry: "npm" }, + closure: { + schemaVersion: "1", + topLevel: [{ name: "@x/wf", version: "1.0.0" }], + entries: [], + }, + }, }; describe("markWorkflowDeploymentRecordParked", () => { diff --git a/apps/sidecar/src/workflow-deployment-record.ts b/apps/sidecar/src/workflow-deployment-record.ts index fb18b75f3..d9f3385aa 100644 --- a/apps/sidecar/src/workflow-deployment-record.ts +++ b/apps/sidecar/src/workflow-deployment-record.ts @@ -24,6 +24,7 @@ import { type } from "arktype"; import { getLogger } from "@intx/log"; import { InferenceSource } from "@intx/types/runtime"; +import { SourceRefPin } from "@intx/types/sidecar"; import { isErrnoNotFound } from "./conversation-state"; import { writeFileAtomicDurable } from "./atomic-write"; @@ -46,15 +47,16 @@ export const WorkflowDeploymentRecord = type({ }, "sessionId?": "string > 0", "hubPublicKey?": "string > 0", - // Hub-approved wire hash per referenced onTrigger body id, carried on the - // deploy frame's `referencedDefinitions[*].approvedWireHash`. Durable here - // (not re-derivable from the materialized body `workflow.json` alone -- - // that file has no hash field) so a boot-time restore can rebuild the - // spawned child's `REFERENCED_DEFINITION_HASHES` env without a hub - // round-trip. Absent for a deployment with no referenced bodies. - "referencedDefinitionHashes?": { - "[string]": "string > 0", - }, + // The hub-approved wire hash the restored child re-verifies its evaluated + // closure against, rather than a sidecar recompute of the inert projection + // -- the latter would collapse the out-of-band-pin property the re-verify + // barrier exists for. + approvedWireHash: "string > 0", + // The pin a restore re-runs the closure apply with: `source` names where the + // definition package comes from (no secret -- the registry token resolves + // from env at apply time), `closure` is the hub's frozen dependency set + // (concrete versions + integrity SRIs). Both rode the signed deploy frame. + sourceRef: SourceRefPin, // Written only on the state-preserving hibernate teardown // (`teardownDeployment({ reclaimDirs: false })`), never on deploy or // rotation. Its presence is the durable answer to "did the hub park this diff --git a/apps/sidecar/src/workflow-host-wiring/asset-materialization.ts b/apps/sidecar/src/workflow-host-wiring/asset-materialization.ts index c39733ed3..c40a16f0c 100644 --- a/apps/sidecar/src/workflow-host-wiring/asset-materialization.ts +++ b/apps/sidecar/src/workflow-host-wiring/asset-materialization.ts @@ -1,76 +1,20 @@ -// Workflow-asset materialization on the sidecar's local substrate: -// the deploy-time `workflow.json` / `sources.json` disk-convention -// writes the workflow-process child reads back, and the boot-time -// restore's re-read of the definition off the same convention. +// Workflow-asset materialization on the sidecar's local substrate: the +// deploy-time `sources.json` disk-convention write an in-process onTrigger +// body child reads its inference-source pins back from. The body DEFINITION +// is never staged -- it is resolved in-memory from the parent's re-verified +// closure. import { mkdir, readFile, writeFile } from "node:fs/promises"; import { dirname, join as pathJoin } from "node:path"; import type { AgentDeployFrame } from "@intx/types/sidecar"; -/** - * Materialize the workflow definition on the sidecar's local substrate so - * the workflow-process child's `loadWorkflowDefinition` can read - * `workflow.json` out of the workflow-asset repo's working tree. The - * destination mirrors the bare RepoStore's `getRepoDir` for - * `{ kind: "workflow", id }`: - * `${SIDECAR_DATA_DIR}/assets/workflow//workflow.json`. The child reads - * via `fs.readFile`, so writing the bytes outside git suffices. This is - * deploy-only durable state; the restore path finds it already on disk. - */ -export async function materializeWorkflowJson( - sidecarDataDir: string | undefined, - definition: NonNullable["definition"], -): Promise { - if (typeof sidecarDataDir !== "string" || sidecarDataDir.length === 0) { - throw new Error( - "sidecar deploy router: SIDECAR_DATA_DIR must be present in the multi-step substrate env; the workflow-process child resolves the workflow-asset repo dir against this data dir", - ); - } - const workflowAssetPath = pathJoin( - sidecarDataDir, - "assets", - "workflow", - definition.id, - "workflow.json", - ); - const workflowAssetBytes = JSON.stringify(definition, null, 2); - try { - await mkdir(dirname(workflowAssetPath), { recursive: true }); - // Idempotent: only rewrite when the on-disk content differs. Treats a - // missing file as different. - let existing: string | null = null; - try { - existing = await readFile(workflowAssetPath, "utf8"); - } catch (cause) { - if (!( - cause instanceof Error && - "code" in cause && - (cause as { code: unknown }).code === "ENOENT" - )) { - throw cause; - } - } - if (existing !== workflowAssetBytes) { - await writeFile(workflowAssetPath, workflowAssetBytes, "utf8"); - } - } catch (cause) { - const reason = cause instanceof Error ? cause.message : String(cause); - throw new Error( - `sidecar deploy router: failed to materialize workflow.json at ${workflowAssetPath}: ${reason}`, - { cause }, - ); - } -} - /** * Materialize an extracted onTrigger body's per-step inference-source pins to - * `${dataDir}/assets/workflow//sources.json`, co-located with the - * body's `workflow.json`. A body child runs in-process with no process env - * and loses its env across a restart, so its sources must be durable on disk - * beside the body definition; the body invoker reads this file to build the - * body's inference-source resolver. Mirrors `materializeWorkflowJson`: same - * per-body dir, idempotent content-compare write. + * `${dataDir}/assets/workflow//sources.json`. A body child runs + * in-process with no process env and loses its env across a restart, so its + * sources must be durable on disk; the body invoker reads this file to build + * the body's inference-source resolver. Idempotent content-compare write. */ export async function materializeWorkflowSources( sidecarDataDir: string | undefined, @@ -117,28 +61,3 @@ export async function materializeWorkflowSources( ); } } - -/** - * Read a workflow definition back off the sidecar's local substrate for a - * boot-time restore. Mirrors `materializeWorkflowJson`'s path derivation - * (`${dataDir}/assets/workflow//workflow.json`). Returns the - * parsed-but-unvalidated JSON: the on-disk file is untrusted at restore - * (partial write, corruption, tamper), so the caller re-validates it through - * the same wire + structural gates the deploy path applies. A missing file - * or unparseable JSON throws; the restore loop's per-record catch converts - * that into a warn-and-skip. - */ -export async function readWorkflowJson( - sidecarDataDir: string, - definitionId: string, -): Promise { - const workflowAssetPath = pathJoin( - sidecarDataDir, - "assets", - "workflow", - definitionId, - "workflow.json", - ); - const raw = await readFile(workflowAssetPath, "utf8"); - return JSON.parse(raw); -} diff --git a/apps/sidecar/src/workflow-host-wiring/closure-staging.ts b/apps/sidecar/src/workflow-host-wiring/closure-staging.ts new file mode 100644 index 000000000..0415bc922 --- /dev/null +++ b/apps/sidecar/src/workflow-host-wiring/closure-staging.ts @@ -0,0 +1,232 @@ +// Staging for a deployment's frozen workflow-definition closure: the +// durable per-deployment stores its source assets are checked out into, +// and the apply that lays the closure out and evaluates the pinned code +// to a `WorkflowDefinition`. Both the deploy path and the boot-time +// restore path route through here, so the two resolve identical mounts +// from the pin alone -- restore has only the pin, never a re-delivery. + +import { rm, stat } from "node:fs/promises"; +import { join as pathJoin } from "node:path"; + +import { workflowSourceAssetMountPath } from "@intx/hub-sessions"; +import type { SourceRefPin } from "@intx/types/sidecar"; + +import { + applyFrozenWorkflowClosure, + type AppliedWorkflowClosure, +} from "../workflow-closure-apply"; +import { sourceAssetGitDir } from "../source-asset-delivery"; +import { parseToolRegistries } from "../tool-materialization"; + +/** + * The durable per-deployment store the sidecar checks a deployment's source + * assets out into. A SIBLING of the closure instance dir, not a child: + * `materializeDeploymentClosure` reclaims the closure dir on every apply and + * restore, but never this store, so the checked-out assets survive a restart + * and re-materialization needs no re-delivery. The store is reclaimed on + * redeploy (at the deploy call site) and on undeploy. + */ +export function deploymentSourceAssetRoot( + dataDir: string, + deploymentId: string, +): string { + return pathJoin(dataDir, "workflow-definition-sources", deploymentId); +} + +/** + * The durable indexed-`.git` store root a pinned deployment's source-format + * asset entries are checked out from. Sibling of the plain-file source store; + * both survive restart so re-materialization needs no re-delivery. + */ +export function deploymentSourceGitRoot( + dataDir: string, + deploymentId: string, +): string { + return pathJoin(dataDir, "workflow-definition-source-gits", deploymentId); +} + +/** + * The per-deployment directory a deployment's closure is laid out under. + * Deterministic per deployment id, so a redeploy or a boot restore reuses it. + */ +export function deploymentClosureInstanceDir( + dataDir: string, + deploymentId: string, +): string { + return pathJoin(dataDir, "workflow-definition-closures", deploymentId); +} + +function deriveSourceAssetMounts(pin: SourceRefPin): Map { + const mounts = new Map(); + for (const entry of pin.closure.entries) { + if ( + entry.source.kind === "asset" && + entry.source.package.format === "tarball" + ) { + mounts.set( + entry.source.assetId, + workflowSourceAssetMountPath(entry.source.assetId), + ); + } + } + return mounts; +} + +function deriveSourceGitDirs( + pin: SourceRefPin, + gitRoot: string, +): Map { + const gitDirs = new Map(); + for (const entry of pin.closure.entries) { + if ( + entry.source.kind === "asset" && + entry.source.package.format === "source" + ) { + gitDirs.set( + entry.source.assetId, + sourceAssetGitDir(gitRoot, entry.source.assetId), + ); + } + } + return gitDirs; +} + +async function isExistingDir(dir: string): Promise { + try { + return (await stat(dir)).isDirectory(); + } catch (err) { + if (err instanceof Error && "code" in err && err.code === "ENOENT") { + return false; + } + throw err; + } +} + +/** + * Resolve the durable source-asset store root and the `assetId -> mountPath` / + * `assetId -> gitDir` maps a pinned deployment materializes its + * `kind: "asset"` closure entries from, asserting every referenced asset is + * present on disk. A cheap early gate for a missing checkout; the loader still + * SRI-verifies each tarball's bytes at materialization. A missing mount is a + * broken deployment the hub must re-drive, so it fails loud rather than + * materializing against an absent store. + */ +export async function resolveDeploymentAssetMounts( + dataDir: string, + deploymentId: string, + pin: SourceRefPin, +): Promise<{ + assetRoot: string; + assetMounts: ReadonlyMap; + gitDirs: ReadonlyMap; +}> { + const assetRoot = deploymentSourceAssetRoot(dataDir, deploymentId); + const assetMounts = deriveSourceAssetMounts(pin); + for (const [assetId, mountPath] of assetMounts) { + const mountDir = pathJoin(assetRoot, mountPath); + if (!(await isExistingDir(mountDir))) { + throw new Error( + `resolveDeploymentAssetMounts: source asset ${JSON.stringify(assetId)} for deployment ${deploymentId} is not present in the durable store at ${mountDir}; the deployment must be re-driven from the hub`, + ); + } + } + const gitRoot = deploymentSourceGitRoot(dataDir, deploymentId); + const gitDirs = deriveSourceGitDirs(pin, gitRoot); + for (const [assetId, gitDir] of gitDirs) { + if (!(await isExistingDir(gitDir))) { + throw new Error( + `resolveDeploymentAssetMounts: source asset ${JSON.stringify(assetId)} for deployment ${deploymentId} has no indexed git store at ${gitDir}; the deployment must be re-driven from the hub`, + ); + } + } + return { assetRoot, assetMounts, gitDirs }; +} + +/** + * Read a substrate-config byte cap (`SIDECAR_CACHE_MAX_BYTES` / + * `SIDECAR_REGISTRY_MAX_TARBALL_BYTES`) from the multi-step substrate env and + * parse it to a positive finite number. The boot edge resolves these once and + * threads them through the substrate env; the closure apply needs them to size + * the tarball cache and the per-fetch cap. A missing or non-numeric value is a + * boot-edge wiring bug, so it fails loud rather than defaulting. + */ +export function requireSubstrateByteCap( + env: Record, + key: string, +): number { + const raw = env[key]; + if (raw === undefined) { + throw new Error( + `sidecar deploy router: ${key} must be present in the multi-step substrate env to materialize a frozen workflow closure`, + ); + } + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed <= 0) { + throw new Error( + `sidecar deploy router: ${key} must be a positive finite number, got ${JSON.stringify(raw)}`, + ); + } + return parsed; +} + +/** + * Materialize a deployment's frozen closure to its per-deployment instance dir + * and evaluate the pinned code. The instance dir is force-reclaimed first: the + * id is deterministic per address, so a redeploy or a boot restore reuses the + * same dir and a prior soft-failed deploy can leave it half-materialized. Safe + * only because no live reader holds the dir when this runs -- a precondition + * each caller establishes. + */ +export async function materializeDeploymentClosure(args: { + dataDir: string; + deploymentId: string; + pin: SourceRefPin; + substrateEnv: Record; +}): Promise { + const instanceDir = deploymentClosureInstanceDir( + args.dataDir, + args.deploymentId, + ); + await rm(instanceDir, { recursive: true, force: true }); + + const { assetRoot, assetMounts, gitDirs } = + await resolveDeploymentAssetMounts( + args.dataDir, + args.deploymentId, + args.pin, + ); + + return applyFrozenWorkflowClosure({ + source: args.pin.source, + closure: args.pin.closure, + instanceDir, + cacheRoot: pathJoin(args.dataDir, "workflow-definition-closure-cache"), + cacheMaxBytes: requireSubstrateByteCap( + args.substrateEnv, + "SIDECAR_CACHE_MAX_BYTES", + ), + registryMaxTarballBytes: requireSubstrateByteCap( + args.substrateEnv, + "SIDECAR_REGISTRY_MAX_TARBALL_BYTES", + ), + registries: parseToolRegistries( + requireSubstrateEntry(args.substrateEnv, "SIDECAR_TOOL_REGISTRIES"), + ), + assetRoot, + assetMounts, + gitDirs, + }); +} + +function requireSubstrateEntry( + env: Record, + key: string, +): string { + const raw = env[key]; + if (raw === undefined) { + throw new Error( + `sidecar deploy router: ${key} must be present in the multi-step substrate env to materialize a frozen workflow closure`, + ); + } + return raw; +} diff --git a/apps/sidecar/src/workflow-host-wiring/index.ts b/apps/sidecar/src/workflow-host-wiring/index.ts index 9f2dd8f01..8fff6fcc5 100644 --- a/apps/sidecar/src/workflow-host-wiring/index.ts +++ b/apps/sidecar/src/workflow-host-wiring/index.ts @@ -37,9 +37,10 @@ import { type KeyPair, } from "@intx/types/runtime"; import { - AgentDeployWorkflow, + WorkflowProjectionDefinition, type AgentDeployFrame, } from "@intx/types/sidecar"; +import { projectLiveToInert } from "@intx/workflow"; import type { MultistepCredentialsRouter, @@ -80,11 +81,16 @@ import { } from "./step-strategy"; export { deriveDeploymentId }; +import { materializeWorkflowSources } from "./asset-materialization"; import { - materializeWorkflowJson, - materializeWorkflowSources, - readWorkflowJson, -} from "./asset-materialization"; + deploymentSourceAssetRoot, + deploymentSourceGitRoot, + materializeDeploymentClosure, +} from "./closure-staging"; +import { + MAX_INLINE_ASSET_PAYLOAD_BYTES, + materializeWorkflowAssets, +} from "../source-asset-delivery"; export { computeWireDefinitionHash, validateWorkflowProjection }; @@ -462,6 +468,13 @@ export function createSidecarDeployRouter(deps: { * it. */ writeWorkflowDeploymentRecord?: typeof writeWorkflowDeploymentRecord; + /** + * Closure materializer, injectable so a test can stand in for the real + * fetch + SRI-verify + layout + evaluate pass without publishing a package. + * Defaults to the real `materializeDeploymentClosure`; production never + * overrides it. + */ + materializeDeploymentClosure?: typeof materializeDeploymentClosure; }): SidecarDeployRouter { // Validate the signing seed at construction so a malformed key fails // sidecar boot rather than the first multi-step deploy, where the @@ -493,6 +506,8 @@ export function createSidecarDeployRouter(deps: { const stepStateDataDir = multistepSubstrateEnv.SIDECAR_DATA_DIR; const persistDeploymentRecord = deps.writeWorkflowDeploymentRecord ?? writeWorkflowDeploymentRecord; + const applyClosure = + deps.materializeDeploymentClosure ?? materializeDeploymentClosure; const multistepSpawner = deps.multistepSubprocessSpawner ?? defaultSubprocessSpawner; const multistepDeriveStepAddress: DeriveStepAddress = @@ -557,7 +572,14 @@ export function createSidecarDeployRouter(deps: { */ interface WorkflowDeploySpec { agentAddress: string; - definition: NonNullable["definition"]; + /** + * The runnable definition, projected to its inert wire shape. Always the + * closure evaluation (`projectLiveToInert(applied.definition)`), never a + * frame-carried inline definition -- the deploy frame carries none. Both + * the deploy path and the boot-time restore derive it the same way, from + * the materialized closure. + */ + definition: WorkflowProjectionDefinition; sources: NonNullable["sources"]; /** Correlates the child's inference events to the deploy's session. */ sessionId: string | undefined; @@ -570,14 +592,26 @@ export function createSidecarDeployRouter(deps: { */ hubPublicKey: string | undefined; /** - * Hub-approved wire hash per referenced onTrigger body id, sourced from - * the deploy frame's `referencedDefinitions[*].approvedWireHash`. Threaded - * to the spawned child as `REFERENCED_DEFINITION_HASHES` so a body spawn - * can re-verify against the parent's approval - * (`WorkflowSpawnSuspendableChildOpts.referencedDefinitionHashes`). - * Undefined for a deployment with no referenced bodies. + * The hub-approved wire hash the deploy frame carried. The child's + * `DEFINITION_HASH` is sourced from this hub authority, never a sidecar + * recompute, so a closure that no longer projects to the approved content + * cannot run. + */ + approvedWireHash: string; + /** + * Sidecar-local directory of the materialized closure. The spawn core + * threads it into the child's env so the run child evaluates the pinned + * code to a live definition. Never travels on the deploy frame and is + * never persisted -- a restore re-materializes it from `sourceRef`. */ - referencedDefinitionHashes: Record | undefined; + closurePackageDir: string; + /** + * The source-ref pin the deployment record persists so a boot-time restore + * can re-materialize the pinned code. Its `source` carries no secret (the + * registry token resolves from env at apply time); its `closure` is frozen + * versions + SRIs. + */ + sourceRef: NonNullable["sourceRef"]; /** * Decrypted credential material from the deploy frame's * `workflow.credentials`, threaded to the supervisor's @@ -613,43 +647,11 @@ export function createSidecarDeployRouter(deps: { ...(spec.hubPublicKey !== undefined ? { hubPublicKey: spec.hubPublicKey } : {}), - ...(spec.referencedDefinitionHashes !== undefined - ? { referencedDefinitionHashes: spec.referencedDefinitionHashes } - : {}), + approvedWireHash: spec.approvedWireHash, + sourceRef: spec.sourceRef, }; } - /** - * Derive the `bodyId -> approvedWireHash` map the spawn core threads to the - * child from the deploy frame's `referencedDefinitions`. Only a body whose - * entry actually carries `approvedWireHash` contributes -- the wire schema - * makes it optional for a frame built before the source-ref hand-off, and - * an unhashed body is exactly the misconfigured-deploy case the spawn-child - * adapter's `resolveVerifiedBody` fails closed on, so this must not paper - * over a missing hash with a fabricated one. Returns `undefined` for a - * deployment with no referenced bodies at all, matching the field's - * optional-when-absent shape on both the spec and the durable record. - */ - function deriveReferencedDefinitionHashes( - referencedDefinitions: NonNullable< - AgentDeployFrame["workflow"] - >["referencedDefinitions"], - ): Record | undefined { - if ( - referencedDefinitions === undefined || - referencedDefinitions.length === 0 - ) { - return undefined; - } - const hashes: Record = {}; - for (const referenced of referencedDefinitions) { - if (referenced.approvedWireHash !== undefined) { - hashes[referenced.definition.id] = referenced.approvedWireHash; - } - } - return hashes; - } - /** * The single owner of the workflow-deployment spawn sequence: construct * the supervisor, register the single-step agent's outbound key + head @@ -659,7 +661,8 @@ export function createSidecarDeployRouter(deps: { * step throws, so a failed spawn leaks nothing. Both the live deploy path * and the boot-time restore path route through here so the two can never * diverge on how a deployment is stood up. Callers materialize the - * deploy-only durable state (`workflow.json`, step grants) before calling. + * deploy-only durable state (the source closure, step grants) before + * calling. */ async function spawnWorkflowDeployment( spec: WorkflowDeploySpec, @@ -708,7 +711,9 @@ export function createSidecarDeployRouter(deps: { let hubKeyRecorded = false; let deploymentRegistered = false; try { - const definitionHash = await computeWireDefinitionHash(spec.definition); + // The hub-approved hash the frame carried, not a sidecar recompute: the + // child re-verifies its evaluated closure against the hub's authority. + const definitionHash = spec.approvedWireHash; // Warm-keep is the single-step launched-agent deploy: the sole step // IS the long-lived agent, so the child warm-keeps it across @@ -719,23 +724,17 @@ export function createSidecarDeployRouter(deps: { // Per-deployment substrate-config keys the workflow-substrate-factory // validator requires. The boot edge's `multistepSubstrateEnv` carries - // the boot-edge constants; the four workflow-definition / workflow-run - // identity keys are derived per-deploy here. + // the boot-edge constants; the definition identity, the workflow-run + // identity keys and the materialized closure dir are derived per-deploy + // here. `CLOSURE_PACKAGE_DIR` is what makes the run child EVALUATE the + // pinned code and re-verify by project-then-hash against + // `DEFINITION_HASH`. const substrateEnv: Record = { ...multistepSubstrateEnv, - WORKFLOW_DEFINITION_REPO_ID: spec.definition.id, - WORKFLOW_DEFINITION_REF: "refs/heads/main", + WORKFLOW_DEFINITION_ID: spec.definition.id, WORKFLOW_RUN_REPO_ID: deploymentId, WORKFLOW_RUN_REF: "refs/heads/main", - // Frozen for the deployment's lifetime, matching STEP_INFERENCE_SOURCES' - // sibling constants above -- unlike sources, a referenced body's - // approved hash never rotates independently of a redeploy. The - // workflow-host child parser (`parseSpawnTimeEnv`) treats an absent key - // as "no referenced bodies"; serializing `{}` here is equivalent and - // keeps this producer unconditional like its neighbors. - REFERENCED_DEFINITION_HASHES: JSON.stringify( - spec.referencedDefinitionHashes ?? {}, - ), + CLOSURE_PACKAGE_DIR: spec.closurePackageDir, }; // Live-rotatable per-step inference sources. Seeded from the deploy // spec, then revised in place by the single-step sources-rotation @@ -1127,29 +1126,6 @@ export function createSidecarDeployRouter(deps: { frame: AgentDeployFrame, projection: NonNullable, ): Promise { - // Boundary validation: a malformed projection is rejected at the - // router edge before the supervisor is constructed so the link - // surfaces a structured failure rather than a hung `starting` - // supervisor. - validateWorkflowProjection(projection); - - // Source-admission gate: reject a deploy where any step pins an - // inference provider this sidecar cannot build, BEFORE any state is - // claimed or the child is spawned. The throw propagates back through - // the deploy frame so the hub's `deployWorkflow` rejects synchronously - // at deploy time, rather than the child failing the run when the - // step's inference first resolves. Covers single- and multi-step: the - // projection's `narrow` guarantees every stepOrder entry has a - // `sources` entry. Every source in a step's failover chain must be - // buildable -- a chain with an unbuildable tail would fail only after - // the reactor failed over onto it -- so this iterates the whole list. - for (const stepId of projection.definition.stepOrder) { - const chain = projection.sources[stepId]; - if (chain !== undefined) { - for (const source of chain) deps.assertSourceBuildable(source); - } - } - // A re-deploy of an address with a live supervisor acks idempotently, // BEFORE touching any durable state: the resident deployment already // owns the address, its persisted key is what reconnect challenges @@ -1181,36 +1157,17 @@ export function createSidecarDeployRouter(deps: { const deploymentId = deriveDeploymentId(frame.agentAddress); - // Single-step launched-agent deploy vs. derived multi-step deploy. - // - // A one-step projection is the agent-launch identity path: the sole - // step keeps the deployment's own (legacy) mail address, and its - // grants live in the legacy agent-state repo keyed by the legacy - // instance id (`parseAgentId(frame.agentAddress)`). This preserves - // the identity the legacy agent-deploy path established -- the - // workflow-run repo stays keyed by `deriveWorkflowRunRepoId(legacy)` - // and `agent_instance.address` remains the `ins_` legacy shape. - // - // A multi-step projection derives `-` per step - // for both the mail address and the agent-state repo id, isolating - // each step's grants in its own repo. - const stepStrategy = createStepStrategy({ - legacyAddress: frame.agentAddress, - stepOrder: projection.definition.stepOrder, - multistepDeriveStepAddress, - }); - // Claim the deployment slug BEFORE any durable write so a colliding // deploymentId (two distinct addresses projecting to the same slug) is - // rejected before `workflow.json`, the step grants, or the supervisor - // touch disk -- the router's "no repo state touched before rejection" + // rejected before the closure, the step grants, or the supervisor touch + // disk -- the router's "no repo state touched before rejection" // guarantee. The claim is released on any failure below; a successful - // deploy keeps it (the undeploy hook releases it at teardown). The - // spawn core owns unwinding the supervisor and registrations it stands - // up; the slug is the caller's. - // Resolve the sidecar data dir once: the deployment record, workflow.json, - // and the per-step scratch all root under it. Required for any deployment - // that spawns a child. + // deploy keeps it (the undeploy hook releases it at teardown). The spawn + // core owns unwinding the supervisor and registrations it stands up; the + // slug is the caller's. + // + // Resolve the sidecar data dir once: the deployment record, the + // materialized closure, and the per-step scratch all root under it. const dataDir = stepStateDataDir; if (typeof dataDir !== "string" || dataDir.length === 0) { throw new Error( @@ -1218,27 +1175,6 @@ export function createSidecarDeployRouter(deps: { ); } - // The spec the shared spawn core consumes, and the durable record that - // lets a boot-time restore rebuild the SAME spec (definition re-read from - // workflow.json by id, grants from the step repos, and the record's - // frame/in-memory-only inputs: sources, session id, single-step hub key, - // referenced-body hashes). - const spec: WorkflowDeploySpec = { - agentAddress: frame.agentAddress, - definition: projection.definition, - sources: projection.sources, - sessionId: frame.config.sessionId, - hubPublicKey: - projection.definition.stepOrder.length === 1 - ? frame.hubPublicKey - : undefined, - referencedDefinitionHashes: deriveReferencedDefinitionHashes( - projection.referencedDefinitions, - ), - credentials: projection.credentials, - }; - const record = buildDeploymentRecord(spec, spec.sources); - claimSlug(deploymentId, frame.agentAddress); // Hold the single-flight reservation across the async body below and clear // it in the finally. Everything above is synchronous and throws before any @@ -1248,28 +1184,118 @@ export function createSidecarDeployRouter(deps: { // yield control before this point. reservingDeployAddresses.add(frame.agentAddress); try { + // Source-ref apply -- the only deploy lineage. Materialize EXACTLY the + // hub's frozen dependency closure and evaluate the PINNED CODE to the + // workflow definition; the frame carries no inline definition to trust. + // The closure is applied byte-for-byte (concrete versions + integrity + // SRIs) and never re-resolved here. + // + // Check the frame's inline source assets out into the durable + // per-deployment store the closure materializes from, reclaiming it + // first so a redeploy drops assets no longer referenced. This runs only + // on the DEPLOY path -- restore re-reads the store this deploy persisted + // -- so the checkout lives here rather than inside the shared + // materializer. A registry-sourced pin delivers no assets and only + // clears the store. + const assetStore = deploymentSourceAssetRoot(dataDir, deploymentId); + const gitStore = deploymentSourceGitRoot(dataDir, deploymentId); + await rm(assetStore, { recursive: true, force: true }); + await rm(gitStore, { recursive: true, force: true }); + if (projection.assets !== undefined && projection.assets.length > 0) { + await materializeWorkflowAssets({ + assets: projection.assets, + closure: projection.sourceRef.closure, + assetRoot: assetStore, + gitDirRoot: gitStore, + maxAssetPayloadBytes: MAX_INLINE_ASSET_PAYLOAD_BYTES, + }); + } + // Safe to reclaim the instance dir inside the helper: this deploy is + // single-flight-guarded by the reservation above and the child is not + // yet spawned, so no live reader holds it. + const applied = await applyClosure({ + dataDir, + deploymentId, + pin: projection.sourceRef, + substrateEnv: multistepSubstrateEnv, + }); + const validatedDefinition = WorkflowProjectionDefinition( + projectLiveToInert(applied.definition), + ); + if (validatedDefinition instanceof type.errors) { + throw new Error( + `sidecar deploy router: workflow definition loaded from the frozen closure failed projection validation: ${validatedDefinition.summary}`, + ); + } + const definition: WorkflowProjectionDefinition = validatedDefinition; + + // Structural invariants the wire arktype does not cover (non-empty + // stepOrder, every stepOrder entry backed by a `steps` entry AND a + // `sources` entry), checked against the closure-derived definition -- + // the frame carries none to cover. Mirrors the restore path. + validateWorkflowProjection({ definition, sources: projection.sources }); + + // Source-admission gate: reject a deploy where any step pins an + // inference provider this sidecar cannot build. Every source in a step's + // failover chain must be buildable -- a chain with an unbuildable tail + // would fail only after the reactor failed over onto it -- so this + // iterates the whole list. The throw propagates back through the deploy + // frame so the hub's `deployWorkflow` rejects synchronously. + for (const stepId of definition.stepOrder) { + const chain = projection.sources[stepId]; + if (chain !== undefined) { + for (const source of chain) deps.assertSourceBuildable(source); + } + } + + // Single-step launched-agent deploy vs. derived multi-step deploy. A + // one-step definition keeps the deployment's own mail address and its + // grants in the agent-state repo keyed by the instance id; a multi-step + // definition derives `-` per step for both the + // mail address and the agent-state repo id, isolating each step's + // grants in its own repo. + const stepStrategy = createStepStrategy({ + legacyAddress: frame.agentAddress, + stepOrder: definition.stepOrder, + multistepDeriveStepAddress, + }); + + // The child re-verifies its evaluated closure against the hub's approved + // hash. A frame that carried none has no anchor to verify against, so + // fail closed rather than substitute a sidecar recompute. + if (projection.approvedWireHash === undefined) { + throw new Error( + `sidecar deploy router: the deploy frame for ${frame.agentAddress} carries no approvedWireHash; the child has no hub-approved anchor to re-verify the evaluated closure against`, + ); + } + + const spec: WorkflowDeploySpec = { + agentAddress: frame.agentAddress, + definition, + sources: projection.sources, + sessionId: frame.config.sessionId, + hubPublicKey: + definition.stepOrder.length === 1 ? frame.hubPublicKey : undefined, + approvedWireHash: projection.approvedWireHash, + closurePackageDir: applied.packageDir, + sourceRef: projection.sourceRef, + credentials: projection.credentials, + }; + const record = buildDeploymentRecord(spec, spec.sources); + // Persist the deployment record BEFORE the spawn so a crash mid-spawn // leaves a record the boot scan re-drives (an idempotent re-spawn; the // child's in-flight-run discovery resumes any run). A soft-failed deploy // deletes it below, so only a crash-interrupted deploy leaves one. await persistDeploymentRecord(dataDir, deploymentId, record); - // Materialize the deploy-only durable state the spawned child and the - // supervisor read from disk: the workflow definition (`workflow.json`) - // and each step's grants. The restore path finds both already on disk - // and skips this; both land before the shared spawn core runs. - await materializeWorkflowJson(dataDir, projection.definition); - - // Materialize each extracted onTrigger section body as its own - // `assets/workflow//workflow.json` (the body id IS the ref) plus - // a co-located `sources.json`, so a body child's spawn-child resolves the - // body definition AND its inference sources off disk without a hub - // round-trip. The hub also stores each body, but that copy is not on the - // sidecar; the deploy frame carries them here for exactly this reason. The - // sources ride on disk (not through env) because the body child is - // in-process and loses its env across a restart. + // Materialize each extracted onTrigger section body's per-step inference + // sources. The body DEFINITION is not staged: the run child resolves each + // body in-memory from the parent's re-verified closure and hard-fails + // rather than reading a body definition off disk. The sources ride on + // disk (not through env) because the body child is in-process and loses + // its env across a restart. for (const referenced of projection.referencedDefinitions ?? []) { - await materializeWorkflowJson(dataDir, referenced.definition); await materializeWorkflowSources( dataDir, referenced.definition.id, @@ -1286,7 +1312,7 @@ export function createSidecarDeployRouter(deps: { await writeStepGrants({ repoStore: deps.repoStore, deploymentId, - stepOrder: projection.definition.stepOrder, + stepOrder: definition.stepOrder, deriveStepRepoId: stepStrategy.deriveStepRepoId, grants: frame.config.grants, }); @@ -1306,13 +1332,13 @@ export function createSidecarDeployRouter(deps: { // never disturbs the run's events. Guarded on `isRunAddress`: only // a run address names a self-anchored run id. if ( - projection.definition.stepOrder.length === 1 && + definition.stepOrder.length === 1 && isRunAddress(frame.agentAddress) ) { await writeStepGrants({ repoStore: deps.repoStore, deploymentId, - stepOrder: projection.definition.stepOrder, + stepOrder: definition.stepOrder, deriveStepRepoId: stepStrategy.deriveStepRepoId, grants: frame.config.grants, runId: parseAgentId(frame.agentAddress), @@ -1383,31 +1409,42 @@ export function createSidecarDeployRouter(deps: { return; } - // Re-read and RE-VALIDATE the definition off disk with the exact - // gates the deploy path applies: the wire arktype - // (`AgentDeployWorkflow`) to narrow the untrusted on-disk shape, - // then `validateWorkflowProjection` for the structural invariants - // the arktype does not cover. The on-disk `workflow.json` is - // untrusted at restore, so it must clear the same bar a fresh - // deploy frame clears -- no weaker. - const definitionRaw = await readWorkflowJson(dataDir, record.definitionId); - const projection = AgentDeployWorkflow({ - definition: definitionRaw, - sources: record.sources, + // Reconstruct this deployment's runnable definition: re-materialize the + // pinned closure and evaluate the pinned code, then project it to the + // inert wire shape -- the SAME computation the deploy path applies. The + // closure IS the source of truth; no on-disk definition is read. The + // helper reclaims the instance dir first, safe here because the prior + // process (the only reader) is dead and restore is serial before + // `hubLink.connect()`. Asset-sourced entries read from the durable source + // store the original deploy checked out, so no re-delivery is needed; a + // store miss soft-fails the record (kept for the next boot). + const applied = await applyClosure({ + dataDir, + deploymentId, + pin: record.sourceRef, + substrateEnv: multistepSubstrateEnv, }); - if (projection instanceof type.errors) { - logger.warn`skipping workflow deployment restore for ${record.agentAddress}: workflow.json failed validation: ${projection.summary}`; + const validatedDefinition = WorkflowProjectionDefinition( + projectLiveToInert(applied.definition), + ); + if (validatedDefinition instanceof type.errors) { + logger.warn`skipping workflow deployment restore for ${record.agentAddress}: workflow definition loaded from the frozen closure failed projection validation: ${validatedDefinition.summary}`; return; } - validateWorkflowProjection(projection); + const definition: WorkflowProjectionDefinition = validatedDefinition; + + // Structural invariants the wire arktype does not cover. The closure eval + // skips the deploy frame's coverage narrow, so this is where the + // definition-vs-sources coverage is checked. + validateWorkflowProjection({ definition, sources: record.sources }); // Re-run the source-admission gate: refuse to restore a deployment // whose pinned provider this sidecar can no longer build. Every // source in a step's failover chain must be buildable, so this // iterates the whole list. The record is KEPT (not deleted) so a // later boot with the provider restored retries it. - for (const stepId of projection.definition.stepOrder) { - const chain = projection.sources[stepId]; + for (const stepId of definition.stepOrder) { + const chain = record.sources[stepId]; if (chain !== undefined) { for (const source of chain) deps.assertSourceBuildable(source); } @@ -1415,22 +1452,27 @@ export function createSidecarDeployRouter(deps: { const spec: WorkflowDeploySpec = { agentAddress: record.agentAddress, - definition: projection.definition, - sources: projection.sources, + definition, + sources: record.sources, sessionId: record.sessionId, hubPublicKey: record.hubPublicKey, - referencedDefinitionHashes: record.referencedDefinitionHashes, - // Frame-only, never persisted: a restore (boot-time OR a CL-5477 - // wake) waits for the hub's next `credentials.update` push, exactly - // like a redeploy of a deployment that predates a credentials push. + approvedWireHash: record.approvedWireHash, + closurePackageDir: applied.packageDir, + // Re-carried so a post-restore source rotation -- which rebuilds the + // record from the spec -- re-persists the pin; without this a rotation + // would silently drop it and wedge the NEXT restart. + sourceRef: record.sourceRef, + // Frame-only, never persisted: a restore (boot-time OR a wake) waits for + // the hub's next `credentials.update` push, exactly like a redeploy of a + // deployment that predates a credentials push. credentials: undefined, }; // The slug is the caller's, matching `deployMultiStep`: claim before // the spawn, release on failure. Unlike deploy's soft-fail, restore - // does NOT delete the record and does NOT re-materialize - // `workflow.json` or the step grants -- all of that is already on - // disk from the original deploy. A failed restore just warns and + // does NOT delete the record and does NOT re-materialize the step grants + // or the onTrigger body sources -- both are already on disk from the + // original deploy. A failed restore just warns and // leaves the record for the next boot; there is deliberately no GC // of a permanently-unrestorable record here (an operator reclaims it // by undeploying the address). diff --git a/apps/sidecar/src/workflow-probe-handler.ts b/apps/sidecar/src/workflow-probe-handler.ts new file mode 100644 index 000000000..cc02450b6 --- /dev/null +++ b/apps/sidecar/src/workflow-probe-handler.ts @@ -0,0 +1,740 @@ +// Sidecar workflow-probe handler: the airlocked one-shot probe child. +// +// A `workflow.probe.request` frame asks this sidecar to inspect a +// code-sourced workflow WITHOUT deploying it. The inspection evaluates +// author code (the workflow package's `interchange.workflow` entry), so +// it must never run in the sidecar host's address space. This module +// spawns a ONE-SHOT child process behind the IPC airlock that loads and +// evaluates the entry, runs the capability walk plus the live->inert +// projector, and ships the inert projection + advisory grant set + wire +// hash back over an HMAC-authenticated result frame (reusing the same +// per-frame HMAC discipline `@intx/workflow-host`'s event channel uses). +// +// Reaping is sidecar-owned and independent of the hub's probe timeout. +// The hub timeout only rejects the hub-side promise; it does not kill +// the child. `runOneShotProbeChild` owns a self-contained lifecycle with +// its OWN deadline and reaps the child on every exit path -- eval +// success, eval throw, malformed code, and a probe that outruns the +// self-owned deadline -- so a wedged or runaway child can never survive +// the probe call. +// +// The frozen dependency closure is materialized sidecar-side (host, +// no eval) through the injected `MaterializeWorkflowClosure` seam; only +// the load+evaluate+walk+project step runs in the child. The child +// receives the materialized package directory in its fresh, minimal env +// -- no ambient inputs, no sidecar keys. + +import { fileURLToPath } from "node:url"; + +import { type } from "arktype"; + +import { getLogger } from "@intx/log"; +import { GrantWalkSnapshot, hexDecode, hexEncode } from "@intx/types"; +import type { GrantRequirement } from "@intx/types"; +import type { WorkflowProbeRequestFrame } from "@intx/types/sidecar"; +import { WorkflowProjectionDefinition } from "@intx/types/sidecar"; +import { computeWireDefinitionHash } from "@intx/types/wire-definition-hash"; +import { collectDeclaredPluginNames, projectLiveToInert } from "@intx/workflow"; +import { + walkCapabilities, + type CapabilityWalkResult, +} from "@intx/workflow-deploy"; +import { + DEFAULT_KILL_TIMEOUT_MS, + MacedEnvelope, + encodeEnvelope, + generateChannelId, + generateHmacKey, + loadWorkflowDefinitionFromClosure, + loadWorkflowDirectorRegistryFromClosure, + loadWorkflowPluginToolDefinitionsFromClosure, + signHmac, + verifyHmac, + type FrameEnvelope, +} from "@intx/workflow-host"; + +const logger = getLogger(["sidecar", "workflow-probe"]); + +const IPC_HMAC_KEY_BYTES = 32; + +/** + * Self-owned upper bound on how long a single probe child may run before + * the sidecar reaps it and fails the probe. Independent of the hub's + * `probeTimeoutMs`: the hub timeout only rejects the hub-side promise, + * whereas this deadline is what actually kills a wedged child. + */ +export const DEFAULT_PROBE_CHILD_TIMEOUT_MS = 30_000; + +/** + * Self-owned SIGTERM->SIGKILL escalation window when reaping the child. + * Mirrors the supervisor's `DEFAULT_KILL_TIMEOUT_MS` so a probe child + * that ignores SIGTERM is force-killed on the same schedule a supervised + * child is. + */ +export const DEFAULT_PROBE_CHILD_KILL_TIMEOUT_MS = DEFAULT_KILL_TIMEOUT_MS; + +// Env keys the host sets on the child's fresh spawn env. The child reads +// exactly these plus PATH/HOME/TMPDIR (for exec + tmp); nothing else +// crosses the airlock. +const PROBE_CHANNEL_ID_ENV = "PROBE_IPC_CHANNEL_ID"; +const PROBE_HMAC_KEY_ENV = "PROBE_IPC_HMAC_KEY"; +const PROBE_PACKAGE_DIR_ENV = "PROBE_PACKAGE_DIR"; + +/** + * The child's `bin/workflow-probe-child` entry, resolved statically at + * module load so the spawn surface does not depend on any runtime env + * override. Mirrors the supervisor's `bin/workflow-child` resolution. + */ +const DEFAULT_PROBE_CHILD_BINARY: string = fileURLToPath( + import.meta.resolve("../bin/workflow-probe-child"), +); + +// --------------------------------------------------------------------------- +// Result payload wire (child -> host) +// --------------------------------------------------------------------------- + +/** + * The child's single result payload, carried inside the HMAC-signed + * envelope. `ok: true` ships the inert projection, the advisory grant + * set, the un-flattened grant walk snapshot, and the wire hash; `ok: + * false` ships the failure reason (eval throw, malformed code) so the + * host can reject the probe with a meaningful message rather than a bare + * "child exited" surface. + */ +const ProbeResultPayload = type({ + ok: "true", + projection: "unknown", + grants: "string[]", + grantWalkSnapshot: GrantWalkSnapshot, + wireHash: "string > 0", +}).or({ + ok: "false", + error: "string", +}); +type ProbeResultPayload = typeof ProbeResultPayload.infer; + +/** + * The inert answer a probe execution produces: the workflow's inert + * needs-surface projection, the advisory grant set derived from it, the + * un-flattened grant walk snapshot the set is derived from, and the + * projection's content hash. Structurally the `WorkflowProbeResult` the + * hub-agent probe seam consumes. + * + * `grantWalkSnapshot` carries the per-step grant declarations (grant + * strings plus each step's tool-grant `grantEffects` map) and the + * definition's full `grantRequirements` -- the grouping and effect data + * the flattened `grants` union discards. + */ +export interface WorkflowProbeResult { + readonly projection: WorkflowProjectionDefinition; + readonly grants: string[]; + readonly grantWalkSnapshot: GrantWalkSnapshot; + readonly wireHash: string; +} + +// --------------------------------------------------------------------------- +// Closure materialization seam +// --------------------------------------------------------------------------- + +/** + * A materialized workflow package closure: the directory holding the + * workflow package's `package.json` (with its `node_modules/` laid out + * so the entry's bare-specifier imports resolve), plus a `cleanup` the + * handler always calls once the child has been reaped. + */ +export interface MaterializedWorkflowClosure { + readonly packageDir: string; + cleanup(): Promise; +} + +/** + * Host-side materializer for a probe frame's frozen closure. Fetches, + * verifies, extracts, and lays out the workflow package (and its + * dependency closure) into a resolvable tree, returning the package + * directory the child loads from. This runs on the sidecar host -- it is + * I/O, not author-code evaluation -- so the airlocked child only performs + * the load+evaluate step. The production materializer is host-supplied so + * `@intx/workflow-host` stays free of a `@intx/tool-packaging` dependency. + */ +export type MaterializeWorkflowClosure = ( + frame: WorkflowProbeRequestFrame, +) => Promise; + +// --------------------------------------------------------------------------- +// Child spawn seam +// --------------------------------------------------------------------------- + +/** + * Minimal handle over a spawned probe child. The probe needs only the + * child's stdout (the single result line), a kill primitive, and the + * `exited` promise for reaping -- no control/event channels, because the + * probe carries no bidirectional control traffic. + */ +export interface ProbeChildHandle { + readonly pid: number; + readonly stdout: ReadableStream; + kill(signal?: number | string): void; + readonly exited: Promise; +} + +/** + * Spawner the handler invokes to launch the one-shot probe child. + * Production injects the `Bun.spawn`-backed `defaultProbeChildSpawner`; + * tests inject a spawner that records the spawned pid so they can assert + * the child was reaped. + */ +export type ProbeChildSpawner = (args: { + binaryPath: string; + env: Record; +}) => ProbeChildHandle; + +/** + * Real `Bun.spawn`-backed probe-child spawner. Constructs a fresh env + * (the caller assembles it; no `process.env` spread), pipes stdout for + * the result line, ignores stdin, and inherits stderr so child + * diagnostics land on the sidecar's stderr. + */ +export const defaultProbeChildSpawner: ProbeChildSpawner = ({ + binaryPath, + env, +}): ProbeChildHandle => { + const proc = Bun.spawn([binaryPath], { + stdio: ["ignore", "pipe", "inherit"], + env, + }); + return { + pid: proc.pid, + stdout: proc.stdout, + kill(signal?: number | string): void { + if (signal === undefined) { + proc.kill(); + return; + } + if (typeof signal === "number") { + proc.kill(signal); + return; + } + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- the probe reaper passes "SIGTERM"/"SIGKILL"; Bun's runtime accepts the same "SIG*" strings, narrowed back at the boundary. + proc.kill(signal as NodeJS.Signals); + }, + exited: proc.exited, + }; +}; + +// --------------------------------------------------------------------------- +// Executor (host side) +// --------------------------------------------------------------------------- + +export interface WorkflowProbeExecutorOpts { + /** Host-side materializer for the frame's frozen closure. */ + materialize: MaterializeWorkflowClosure; + /** Override the child spawner (defaults to the Bun.spawn-backed one). */ + spawnProbeChild?: ProbeChildSpawner; + /** Override the `bin/workflow-probe-child` path. */ + binaryPath?: string; + /** + * Self-owned deadline before the child is reaped and the probe fails. + * Independent of the hub's `probeTimeoutMs`. + */ + childTimeoutMs?: number; + /** SIGTERM->SIGKILL escalation window when reaping. */ + killTimeoutMs?: number; +} + +/** + * Build the sidecar's workflow-probe executor. The returned object + * satisfies the hub-agent `WorkflowProbeExecutor` seam: `probe(frame)` + * returns the inert projection + advisory grant set + wire hash, and + * throws when any step fails so the link answers `workflow.probe.error`. + * + * `probe` materializes the frozen closure, spawns a one-shot airlocked + * child to evaluate the workflow, and reaps that child on every exit + * path independent of the hub's probe timeout. + */ +export function createWorkflowProbeExecutor(opts: WorkflowProbeExecutorOpts): { + probe(frame: WorkflowProbeRequestFrame): Promise; +} { + const spawnProbeChild = opts.spawnProbeChild ?? defaultProbeChildSpawner; + const binaryPath = opts.binaryPath ?? DEFAULT_PROBE_CHILD_BINARY; + const childTimeoutMs = opts.childTimeoutMs ?? DEFAULT_PROBE_CHILD_TIMEOUT_MS; + const killTimeoutMs = + opts.killTimeoutMs ?? DEFAULT_PROBE_CHILD_KILL_TIMEOUT_MS; + + async function probe( + frame: WorkflowProbeRequestFrame, + ): Promise { + const materialized = await opts.materialize(frame); + try { + return await runOneShotProbeChild({ + packageDir: materialized.packageDir, + spawnProbeChild, + binaryPath, + childTimeoutMs, + killTimeoutMs, + }); + } finally { + await materialized.cleanup(); + } + } + + return { probe }; +} + +interface RunOneShotProbeChildArgs { + readonly packageDir: string; + readonly spawnProbeChild: ProbeChildSpawner; + readonly binaryPath: string; + readonly childTimeoutMs: number; + readonly killTimeoutMs: number; +} + +/** + * Spawn a single probe child, drive it to its one result frame, and reap + * it on every exit path. The `finally` guarantees the child is killed + * whether the read succeeds, the child ships an error frame, the child + * exits without a frame (malformed code / crash), or the self-owned + * deadline fires first. + */ +async function runOneShotProbeChild( + args: RunOneShotProbeChildArgs, +): Promise { + const channelId = generateChannelId(); + const hmacKey = generateHmacKey(); + const env = buildProbeChildEnv({ + packageDir: args.packageDir, + channelId, + hmacKey, + }); + const handle = args.spawnProbeChild({ binaryPath: args.binaryPath, env }); + + let reaped = false; + async function reap(): Promise { + if (reaped) return; + reaped = true; + await reapChild(handle, args.killTimeoutMs); + } + + // Attach a catch so a post-reap stdout read error (the kill closes the + // pipe mid-read) resolves to null instead of surfacing as an unhandled + // rejection on the losing race branch. + const linePromise: Promise = readResultLine( + handle.stdout, + ).catch((err: unknown) => { + logger.debug`probe child ${String(handle.pid)} stdout read errored: ${errorMessage(err)}`; + return null; + }); + + const deadline = createDeadline(args.childTimeoutMs); + try { + // Race the result line against the deadline ONLY. Child exit is + // deliberately not a race arm: a child writes its result line and then + // exits promptly, so `handle.exited` and the buffered-line read both become + // ready, and an exit arm winning that race would discard an already-written + // result and fail the probe spuriously. Exit is not a distinct outcome the + // line read misses -- when the child exits its stdout write end closes, so + // `readResultLine` settles either with the trailing line (returned below) + // or null (the "closed its output" case). A child that neither writes nor + // exits is still caught by the deadline. + const outcome = await Promise.race([ + linePromise.then((line) => ({ kind: "line" as const, line })), + deadline.promise.then(() => ({ kind: "timeout" as const })), + ]); + + if (outcome.kind === "timeout") { + throw new Error( + `workflow probe child ${String(handle.pid)} did not produce a result within ${String(args.childTimeoutMs)}ms`, + ); + } + if (outcome.line === null) { + throw new Error( + `workflow probe child ${String(handle.pid)} closed its output without producing a result`, + ); + } + return await parseProbeResult(outcome.line, channelId, hmacKey); + } finally { + deadline.cancel(); + await reap(); + } +} + +function buildProbeChildEnv(args: { + packageDir: string; + channelId: string; + hmacKey: Uint8Array; +}): Record { + // A fresh, minimal env: exactly the IPC anchors and the materialized + // package dir, plus the OS handles the shebang needs to exec `bun` and + // land temp files on the host's temp root. No `process.env` spread, so + // no sidecar secret or ambient input crosses the airlock. + const env: Record = { + [PROBE_CHANNEL_ID_ENV]: args.channelId, + [PROBE_HMAC_KEY_ENV]: hexEncode(args.hmacKey), + [PROBE_PACKAGE_DIR_ENV]: args.packageDir, + }; + const path = process.env["PATH"]; + if (path !== undefined) env["PATH"] = path; + const home = process.env["HOME"]; + if (home !== undefined) env["HOME"] = home; + const tmpdir = process.env["TMPDIR"]; + if (tmpdir !== undefined) env["TMPDIR"] = tmpdir; + return env; +} + +/** + * Reap a probe child: SIGTERM, then SIGKILL if the exit does not land + * within `killTimeoutMs`. SIGKILL is unignorable, so `exited` is + * guaranteed to settle -- a child that traps or ignores SIGTERM cannot + * wedge this call. A kill against an already-exited child is a no-op. + */ +async function reapChild( + handle: ProbeChildHandle, + killTimeoutMs: number, +): Promise { + try { + handle.kill("SIGTERM"); + } catch (err) { + logger.debug`probe child ${String(handle.pid)} SIGTERM raised (already exited?): ${errorMessage(err)}`; + } + const deadline = createDeadline(killTimeoutMs); + const first = await Promise.race([ + handle.exited.then(() => "exited" as const), + deadline.promise.then(() => "deadline" as const), + ]); + deadline.cancel(); + if (first === "exited") return; + logger.warn`workflow probe child ${String(handle.pid)} did not exit on SIGTERM within ${String(killTimeoutMs)}ms; escalating to SIGKILL`; + try { + handle.kill("SIGKILL"); + } catch (err) { + logger.debug`probe child ${String(handle.pid)} SIGKILL raised (already exited?): ${errorMessage(err)}`; + } + await handle.exited.catch(() => { + // A non-zero exit on SIGKILL is the expected outcome; reaping treats + // child exit as success regardless of code. + }); +} + +/** + * Authenticate and parse the child's single result frame. Verifies the + * HMAC over the re-encoded envelope BEFORE trusting any field (mirroring + * the event channel's receiver), binds the frame to this spawn's + * channelId, then narrows the payload. A `ok: false` payload is turned + * into a throw so the probe fails with the child's reason. + */ +async function parseProbeResult( + line: string, + channelId: string, + hmacKey: Uint8Array, +): Promise { + let raw: unknown; + try { + raw = JSON.parse(line); + } catch (cause) { + throw new Error("workflow probe child result is not valid JSON", { cause }); + } + const maced = MacedEnvelope(raw); + if (maced instanceof type.errors) { + throw new Error( + `workflow probe child result envelope failed validation: ${maced.summary}`, + ); + } + const envelopeBytes = encodeEnvelope(maced.envelope); + const macBytes = hexDecode(maced.mac); + const ok = await verifyHmac(envelopeBytes, macBytes, hmacKey); + if (!ok) { + throw new Error( + `workflow probe child result HMAC did not verify (channelId=${maced.envelope.channelId})`, + ); + } + if (maced.envelope.channelId !== channelId) { + throw new Error( + `workflow probe child result carried a foreign channelId ${JSON.stringify(maced.envelope.channelId)}`, + ); + } + const payload = ProbeResultPayload(maced.envelope.payload); + if (payload instanceof type.errors) { + throw new Error( + `workflow probe child result payload failed validation: ${payload.summary}`, + ); + } + if (!payload.ok) { + throw new Error(`workflow probe evaluation failed: ${payload.error}`); + } + const projection = WorkflowProjectionDefinition(payload.projection); + if (projection instanceof type.errors) { + throw new Error( + `workflow probe child projection failed validation: ${projection.summary}`, + ); + } + return { + projection, + grants: payload.grants, + grantWalkSnapshot: payload.grantWalkSnapshot, + wireHash: payload.wireHash, + }; +} + +// --------------------------------------------------------------------------- +// Child side +// --------------------------------------------------------------------------- + +/** + * One line the child writes to a sink. Production wraps `process.stdout`; + * tests inject a capture. The bytes are handed to the OS before the child + * exits so the result is not truncated. + */ +export type ProbeChildLineWriter = (line: string) => Promise; + +export interface RunProbeChildOpts { + /** Raw env the child reads its anchors from (defaults to `process.env`). */ + rawEnv?: Readonly>; + /** Result-line sink (defaults to a drained `process.stdout` write). */ + writeLine?: ProbeChildLineWriter; +} + +/** + * The airlocked child's whole job: read the materialized package dir and + * IPC anchors from its fresh env, load+evaluate the workflow entry, run + * the capability walk plus the live->inert projector, and ship the inert + * projection + advisory grant set + wire hash back inside one + * HMAC-signed result frame. + * + * An evaluation failure (malformed code, an entry that throws, a package + * with no `interchange.workflow`) is caught and shipped as an `ok: false` + * frame so the host reaps cleanly and answers `workflow.probe.error` + * with the reason -- rather than the child crashing and the host seeing a + * bare "exited without result". + */ +export async function runWorkflowProbeChildFromProcessEnv( + opts: RunProbeChildOpts = {}, +): Promise { + const rawEnv = opts.rawEnv ?? process.env; + const writeLine = opts.writeLine ?? defaultStdoutWriteLine; + const { channelId, hmacKey, packageDir } = parseProbeChildEnv(rawEnv); + + let payload: ProbeResultPayload; + try { + payload = await computeProbePayload(packageDir); + } catch (err) { + payload = { ok: false, error: enrichProbeError(err) }; + } + + const envelope: FrameEnvelope = { seq: 0, channelId, payload }; + const envelopeBytes = encodeEnvelope(envelope); + const mac = hexEncode(await signHmac(envelopeBytes, hmacKey)); + await writeLine(`${JSON.stringify({ envelope, mac })}\n`); +} + +async function computeProbePayload( + packageDir: string, +): Promise { + const definition = await loadWorkflowDefinitionFromClosure({ packageDir }); + const projection = projectLiveToInert(definition); + const wireHash = await computeWireDefinitionHash(projection); + // Compose the director registry from the SAME closure the run-child will, + // so the `director:` grants advertised here match what the runtime + // resolves. Built-ins-only when the closure ships no `interchange.directors`. + const directors = await loadWorkflowDirectorRegistryFromClosure({ + packageDir, + }); + // Load the static tool `definitions` each declared plugin package + // contributes from the SAME closure, so the walk emits `tool:` + // grants for plugin-contributed tools (Tier-2 governance). A plugin + // package reaches an agent only through `env.plugins`, so its tool grant + // surface is invisible to the walk otherwise -- the run-child would then + // load the plugin from the closure and the reactor would fail closed on + // an un-approved `tool:`. Loading here (over the frozen closure the + // run-child also materializes from) keeps the approved snapshot and the + // runtime plugin in lockstep. + const pluginToolDefinitions = + await loadWorkflowPluginToolDefinitionsFromClosure({ + packageDir, + plugins: collectDeclaredPluginNames(definition), + }); + const walk = walkCapabilities(definition, directors, pluginToolDefinitions); + // Fail closed on an unresolved director: the runtime does not re-gate + // `director:` against the approved grant set, so this advertisement is + // the only approval checkpoint for a director. Shipping an ok probe whose + // grant set silently omits a director the runtime would still try to + // resolve would let the operator approve an incomplete manifest. Mirrors + // the live-authored approval gate (`createApprovalSetGate`). + const [unresolved] = walk.unresolvedDirectors; + if (unresolved !== undefined) { + return { ok: false, error: `unresolvable director: ${unresolved}` }; + } + return { + ok: true, + projection, + grants: collectDeploymentGrants(walk), + grantWalkSnapshot: buildGrantWalkSnapshot( + walk, + definition.grantRequirements, + ), + wireHash, + }; +} + +/** + * Flatten the per-step walk output into the deployment-wide advisory + * grant set: the deduplicated, sorted union of every step's grant + * strings. Sorting makes the shipped set order-independent. + */ +function collectDeploymentGrants(walk: CapabilityWalkResult): string[] { + const grants = new Set(); + for (const declarations of walk.perStep.values()) { + for (const grant of declarations.grants) { + grants.add(grant); + } + } + return [...grants].sort(); +} + +/** + * Serialize the un-flattened capability walk into a plain-data + * `GrantWalkSnapshot`: the per-step grant declarations (each step's grant + * strings plus its tool-grant `grantEffects` map, converted from the + * walk's `Map` to a plain object) and the definition's full, unfiltered + * `grantRequirements`. Unlike `collectDeploymentGrants`, this preserves + * the per-step grouping and the effect data the flattened set discards. + * A definition that declares no requirements snapshots an empty list. + */ +function buildGrantWalkSnapshot( + walk: CapabilityWalkResult, + grantRequirements: readonly GrantRequirement[] | undefined, +): GrantWalkSnapshot { + const perStep = [...walk.perStep].map(([stepId, declarations]) => ({ + stepId, + grants: [...declarations.grants], + grantEffects: Object.fromEntries(declarations.grantEffects), + })); + return { + perStep, + grantRequirements: [...(grantRequirements ?? [])], + }; +} + +interface ProbeChildEnv { + readonly channelId: string; + readonly hmacKey: Uint8Array; + readonly packageDir: string; +} + +const NonEmptyString = type("string > 0"); + +function parseProbeChildEnv( + rawEnv: Readonly>, +): ProbeChildEnv { + const channelId = requireEnv(rawEnv, PROBE_CHANNEL_ID_ENV); + const packageDir = requireEnv(rawEnv, PROBE_PACKAGE_DIR_ENV); + const hmacKeyHex = requireEnv(rawEnv, PROBE_HMAC_KEY_ENV); + const hmacKey = hexDecode(hmacKeyHex); + if (hmacKey.length !== IPC_HMAC_KEY_BYTES) { + throw new Error( + `workflow probe child env: ${PROBE_HMAC_KEY_ENV} must decode to ${String(IPC_HMAC_KEY_BYTES)} bytes, got ${String(hmacKey.length)}`, + ); + } + return { channelId, hmacKey, packageDir }; +} + +function requireEnv( + rawEnv: Readonly>, + key: string, +): string { + const value = NonEmptyString(rawEnv[key]); + if (value instanceof type.errors) { + throw new Error( + `workflow probe child env: required key ${key} is unset or empty`, + ); + } + return value; +} + +function defaultStdoutWriteLine(line: string): Promise { + return new Promise((resolve, reject) => { + process.stdout.write(line, (err) => { + if (err) reject(err); + else resolve(); + }); + }); +} + +// --------------------------------------------------------------------------- +// Shared helpers +// --------------------------------------------------------------------------- + +/** + * Read one newline-delimited line from a byte stream. Resolves the first + * complete line, or `null` when the stream closes without one (the child + * exited before writing). Releases the reader lock on every exit. + */ +async function readResultLine( + stream: ReadableStream, +): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder("utf-8"); + let pending = ""; + try { + for (;;) { + const { value, done } = await reader.read(); + if (value !== undefined) { + pending += decoder.decode(value, { stream: true }); + const nl = pending.indexOf("\n"); + if (nl >= 0) { + return pending.slice(0, nl).replace(/\r$/, ""); + } + } + if (done) { + const trailing = pending.replace(/\r?\n$/, ""); + return trailing.length > 0 ? trailing : null; + } + } + } finally { + reader.releaseLock(); + } +} + +function createDeadline(ms: number): { + promise: Promise; + cancel: () => void; +} { + let handle: ReturnType | undefined; + const promise = new Promise((resolve) => { + handle = setTimeout(resolve, ms); + }); + return { + promise, + cancel(): void { + if (handle !== undefined) clearTimeout(handle); + }, + }; +} + +function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +// Node/Bun's module-not-found message shape. The specifier is the missing +// package the workflow entry imported at evaluation time. +const MISSING_MODULE_RE = /Cannot find (?:module|package) ['"]([^'"]+)['"]/; + +/** + * Enrich a probe evaluation failure whose cause is a module that could not be + * resolved from the workflow's dependency closure. The evaluator is the layer + * that KNOWS what the workflow imports at run time (it actually ran the + * import), so a missing specifier here means the closure did not carry it -- + * the common cause is a runtime import declared only under `devDependencies` + * (which the closure does not materialize), whether a workspace-local member or + * an external package. Rewrite the opaque "Cannot find module" into that + * actionable diagnostic. A non-resolution failure passes through unchanged. + */ +export function enrichProbeError(err: unknown): string { + const message = errorMessage(err); + const match = MISSING_MODULE_RE.exec(message); + const specifier = match?.[1]; + if (specifier === undefined) return message; + return ( + `workflow entry could not resolve ${JSON.stringify(specifier)} from its dependency closure; ` + + `if the workflow imports it at run time, declare it under "dependencies" rather than "devDependencies" ` + + `(a devDependencies-only import is not materialized into the closure). ${message}` + ); +} diff --git a/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts b/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts index 53f19c3d7..c472fe6c4 100644 --- a/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts +++ b/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts @@ -21,6 +21,7 @@ import type { InferenceEvent } from "@intx/types/runtime"; import { createNoopDrainController, emptyState, + rewriteInlineChildWorkflowBodies, runtimeRun, type Scheduler, type StepInvokeRequest, @@ -35,7 +36,7 @@ import { createWorkflowHostSignalChannel, createWorkflowRunBlobSubstrate, createWorkflowRunRepoStore, - createWorkflowSpawnChild, + createInMemorySpawnChild, type RunChildWorkflow, type RunSuspendableChild, type SourcesSnapshotRef, @@ -100,18 +101,6 @@ export interface SidecarRunChildDeps { workflowRunRepoId: RepoId; /** Workflow-run ref the child reads/writes against. */ workflowRunRef: string; - /** - * Deploy ref the child env's recursive `spawnChild` resolves - * grandchild `definitionRef`s against. The runtime body's - * `runChildWorkflow` was designed for arbitrary depth; the child's - * env's `spawnChild` slot must itself be a `createWorkflowSpawnChild` - * adapter against this deploy ref so a grandchild spawn resolves - * the grandchild's `workflow.json` from the workflow asset substrate - * the same way the parent's spawn does. The sub-namespace scoping - * (`runs//...`) continues to work because each rung's - * runtime env routes through `runId`-keyed substrate adapters. - */ - workflowDefinitionRef: string; /** Principal the child presents on every substrate operation. */ principal: Principal; /** Host-process scheduler singleton; shared with the parent. */ @@ -210,9 +199,9 @@ export function createSidecarRunChild( // Self-referential `RunChildWorkflow` so a child env's recursive // `spawnChild` (wired inside `buildChildRunEnv`) can route grandchild // spawns back through the same adapter. Each invocation builds a - // per-runId env that itself wires a `spawnChild` slot whose `runChild` - // is this same `runChild` constant -- the recursion bottoms out when a - // rung's `WorkflowDefinition` has no `childWorkflow` primitive. + // per-runId env that itself wires an in-memory `spawnChild` resolver whose + // `runChild` is this same `runChild` constant -- the recursion bottoms out + // when a rung's `WorkflowDefinition` has no `childWorkflow` primitive. // Sub-namespace scoping continues to hold at every depth because // `childRunId` flows verbatim into the per-rung // `blobs`/`signalChannel`/`runtimeRun` calls, keeping every rung's @@ -223,7 +212,11 @@ export function createSidecarRunChild( input, signal, }) => { - const { env, signalChannel } = buildChildRunEnv({ + const { + env, + signalChannel, + definition: rewrittenDefinition, + } = buildChildRunEnv({ deps, directors, clock, @@ -234,7 +227,7 @@ export function createSidecarRunChild( childRunId, }); try { - const handle = runtimeRun(definition, env, { + const handle = runtimeRun(rewrittenDefinition, env, { runId: childRunId, triggerPayload: input, }); @@ -330,7 +323,11 @@ export function createSidecarSpawnSuspendableChild( { definition, childRunId, input, resumeFromEvents, signal }, onEvent, ) => { - const { env: baseEnv, signalChannel } = buildChildRunEnv({ + const { + env: baseEnv, + signalChannel, + definition: rewrittenDefinition, + } = buildChildRunEnv({ deps, directors, clock, @@ -360,7 +357,7 @@ export function createSidecarSpawnSuspendableChild( const bodySourcesRef: SourcesSnapshotRef = { current: await readBodyStepInferenceSources( deps.dataDir, - definition.id, + rewrittenDefinition.id, ), }; const bodyInvokeStep = deps.bodyInvokeStep; @@ -432,7 +429,7 @@ export function createSidecarSpawnSuspendableChild( // the grant via resume on the correlation it recovered from its own // log. On a fresh spawn, seed the run with the event's trigger payload. const handle = runtimeRun( - definition, + rewrittenDefinition, env, resumeFromEvents !== undefined ? { runId: childRunId, resumeFromEvents } @@ -536,10 +533,21 @@ function buildChildRunEnv(args: { }): { env: WorkflowRuntimeEnv; signalChannel: ReturnType; + definition: WorkflowDefinition; } { const { deps, directors, clock, newId, repoStore, runChild, definition } = args; const childRunId = args.childRunId; + // A rung may itself embed a grandchild as an inline `childWorkflow`. Lift + // each to an internal `{ ref }` and run the rewritten definition whose + // children are refs -- the shape the runtime dispatches -- keeping the + // lifted definitions in an in-memory map this rung's own resolver serves + // from, so a grandchild spawns with no on-disk read at any depth. + const { workflow: rewrittenDefinition, bodies: grandchildBodies } = + rewriteInlineChildWorkflowBodies(definition); + const grandchildMap = new Map( + grandchildBodies.map((body) => [body.ref, body.definition]), + ); const blobs = createWorkflowRunBlobSubstrate({ substrate: deps.substrate, repoId: deps.workflowRunRepoId, @@ -574,17 +582,14 @@ function buildChildRunEnv(args: { "sidecar runChild authorize: per-step credentials snapshot is not threaded through the spawn-child seam; the child runtime cannot resolve a workflow-typed authorize call", ); }; - const drain = createNoopDrainController(definition); - // Recursive `spawnChild`: a grandchild's `definitionRef` is resolved - // against the workflow-asset substrate the parent's spawn used, and - // the resolved `WorkflowDefinition` flows back into this same - // `runChild` callback. The runtime body's `runChildWorkflow` - // contract is depth-agnostic; the wiring here makes the sidecar's - // adapter depth-agnostic too. - const spawnChild = createWorkflowSpawnChild({ - substrate: deps.substrate, - principal: deps.principal, - deployRef: deps.workflowDefinitionRef, + const drain = createNoopDrainController(rewrittenDefinition); + // Recursive `spawnChild`: a grandchild embedded inline in this rung is + // resolved from the in-memory map lifted above and flows back into this + // same `runChild` callback. The runtime body's `runChildWorkflow` contract + // is depth-agnostic; the in-memory resolver makes the sidecar's adapter + // depth-agnostic too, with no on-disk read at any rung. + const spawnChild = createInMemorySpawnChild({ + bodies: grandchildMap, runChild, }); const env: WorkflowRuntimeEnv = { @@ -600,7 +605,7 @@ function buildChildRunEnv(args: { newId, drain, }; - return { env, signalChannel }; + return { env, signalChannel, definition: rewrittenDefinition }; } /** diff --git a/apps/sidecar/src/workflow-substrate-factory/config.ts b/apps/sidecar/src/workflow-substrate-factory/config.ts index bea2adc24..67205046c 100644 --- a/apps/sidecar/src/workflow-substrate-factory/config.ts +++ b/apps/sidecar/src/workflow-substrate-factory/config.ts @@ -25,8 +25,7 @@ import { InferenceSource } from "@intx/types/runtime"; */ export const SIDECAR_SUBSTRATE_CONFIG_KEYS = [ "SIDECAR_DATA_DIR", - "WORKFLOW_DEFINITION_REPO_ID", - "WORKFLOW_DEFINITION_REF", + "WORKFLOW_DEFINITION_ID", "WORKFLOW_RUN_REPO_ID", "WORKFLOW_RUN_REF", "SIDECAR_SIGNING_PUBLIC_KEY", @@ -43,8 +42,10 @@ export const SIDECAR_SUBSTRATE_CONFIG_KEYS = [ export const SubstrateConfig = type({ SIDECAR_DATA_DIR: "string > 0", - WORKFLOW_DEFINITION_REPO_ID: "string > 0", - WORKFLOW_DEFINITION_REF: "string > 0", + // The deployed definition's own id, for the workflow-run-authenticated + // capabilities route a step tool calls. Identity only: the definition + // itself is evaluated from the closure, never read from a repo. + WORKFLOW_DEFINITION_ID: "string > 0", WORKFLOW_RUN_REPO_ID: "string > 0", WORKFLOW_RUN_REF: "string > 0", SIDECAR_SIGNING_PUBLIC_KEY: "string > 0", diff --git a/apps/sidecar/src/workflow-substrate-factory/index.ts b/apps/sidecar/src/workflow-substrate-factory/index.ts index 8827b2239..6cae2c3da 100644 --- a/apps/sidecar/src/workflow-substrate-factory/index.ts +++ b/apps/sidecar/src/workflow-substrate-factory/index.ts @@ -48,8 +48,6 @@ import { adaptHostScheduler, createProxyWorkflowRunRepoStore, createWorkflowHostScheduler, - createWorkflowSpawnChild, - createWorkflowSpawnSuspendableChild, createWorkflowStepInvoker, type GrantEvaluator, type LoadParkedApproval, @@ -271,10 +269,6 @@ export function createSidecarSubstrateFactory( kind: "workflow-run" as const, id: validated.WORKFLOW_RUN_REPO_ID, }; - const workflowDefinitionRepoId = { - kind: "workflow" as const, - id: validated.WORKFLOW_DEFINITION_REPO_ID, - }; const principal: WorkflowRunWorkflowProcessPrincipal = { kind: "workflow-process", anchorRunId: env.spawn.anchorRunId, @@ -365,7 +359,7 @@ export function createSidecarSubstrateFactory( toolless: false, hubArtifactsUrl: deriveHubHttpUrl(validated.HUB_WS_URL), sidecarToken: validated.SIDECAR_TOKEN, - definitionId: workflowDefinitionRepoId.id, + definitionId: validated.WORKFLOW_DEFINITION_ID, }; const buildStepEnv = createSidecarStepBuildEnv( durableConversation !== undefined @@ -458,7 +452,7 @@ export function createSidecarSubstrateFactory( toolless: true, hubArtifactsUrl: deriveHubHttpUrl(validated.HUB_WS_URL), sidecarToken: validated.SIDECAR_TOKEN, - definitionId: workflowDefinitionRepoId.id, + definitionId: validated.WORKFLOW_DEFINITION_ID, }); const bodyInvokeStep: SidecarBodyStepInvoker = ( req, @@ -576,7 +570,6 @@ export function createSidecarSubstrateFactory( substrate, workflowRunRepoId, workflowRunRef: validated.WORKFLOW_RUN_REF, - workflowDefinitionRef: validated.WORKFLOW_DEFINITION_REF, principal, scheduler, invokeStep: childInvokeStep, @@ -585,32 +578,20 @@ export function createSidecarSubstrateFactory( bodyInvokeStep, dataDir: validated.SIDECAR_DATA_DIR, }; + // Terminal childWorkflow executor. `run-child` builds the in-memory + // resolver from this plus the lifted-body map it extracts after loading + // the parent's re-verified definition, so an owned inline child spawns + // with no on-disk asset read. const runChild = createSidecarRunChild(childRunDeps); - const spawnChild = createWorkflowSpawnChild({ - substrate, - principal, - deployRef: validated.WORKFLOW_DEFINITION_REF, - runChild, - }); - // An onTrigger section runs each event's body as a suspendable child. - // The resolving adapter maps the body's definition ref to a definition - // and delegates to the sidecar spawner, which returns the live handle - // `runOnTrigger` drives across the body's approval parks. - const spawnSuspendableChild = createWorkflowSpawnSuspendableChild({ - substrate, - principal, - deployRef: validated.WORKFLOW_DEFINITION_REF, - runSuspendableChild: createSidecarSpawnSuspendableChild(childRunDeps), - // Hub-approved wire hash per referenced onTrigger body id, carried on - // the parent's signed deploy frame and threaded here by the sidecar's - // deploy router (`REFERENCED_DEFINITION_HASHES` spawn-time env, parsed - // into `env.spawn.referencedDefinitionHashes` by the workflow-host - // child bootstrap). Not a sidecar recompute -- the hub is the - // authority the body path re-verifies against. - referencedDefinitionHashes: env.spawn.referencedDefinitionHashes, - }); + // `run-child` builds the in-memory body resolver from this raw executor + // plus the lifted-body map it extracts after re-evaluating the parent's + // closure, so a body resolves in-process with no on-disk read and no + // separate per-body re-verify -- the parent's re-verify already covers + // every inline body. + const runSuspendableChild = + createSidecarSpawnSuspendableChild(childRunDeps); // Per-run scratch reclamation for the cold (multi-step) path. The // run-loop fires this once each run reaches its terminal status; it @@ -700,12 +681,10 @@ export function createSidecarSubstrateFactory( workflowRunRepoId, workflowRunRef: validated.WORKFLOW_RUN_REF, principal, - workflowDefinitionRepoId, - workflowDefinitionRef: validated.WORKFLOW_DEFINITION_REF, invokeStep, initialSources: stepInferenceSources, - spawnChild, - spawnSuspendableChild, + runChild, + runSuspendableChild, scheduler, evaluateGrants: evaluateGrantsAdapter, loadParkedApproval, diff --git a/apps/sidecar/test/deploy-router.test.ts b/apps/sidecar/test/deploy-router.test.ts index 96da9eafc..f6adfc96f 100644 --- a/apps/sidecar/test/deploy-router.test.ts +++ b/apps/sidecar/test/deploy-router.test.ts @@ -25,6 +25,8 @@ import { hexEncode } from "@intx/types"; import type { HarnessConfig, InferenceSource } from "@intx/types/runtime"; import type { AgentDeployFrame } from "@intx/types/sidecar"; import type { SubprocessSpawner } from "@intx/workflow-host"; +import { defineWorkflow, step, type WorkflowDefinition } from "@intx/workflow"; +import { buildSingleStepAgentDefinition } from "@intx/workflow-deploy"; import { createSidecarDeployRouter, deriveDeploymentId, @@ -52,6 +54,14 @@ type RouterFixture = { rejectedSources: InferenceSource[]; }; +/** + * The closure a stubbed materializer evaluates to, keyed by the deployment id + * the router derives. A source-ref deploy has no inline definition on the wire, + * so a test that wants a specific definition registers it here rather than + * publishing a real package. + */ +const closureDefinitions = new Map(); + async function makeRouter(dataDir: string): Promise { const signingKey = await generateKeyPair(); const substrate = createAgentRepoStore({ dataDir, signingKey }); @@ -85,6 +95,22 @@ async function makeRouter(dataDir: string): Promise { registerDeployment: () => undefined, unregisterDeployment: () => undefined, multistepSubstrateEnv: { SIDECAR_DATA_DIR: dataDir }, + // Stand in for the real fetch + SRI-verify + layout + evaluate pass: a + // test cannot publish a package, so the pinned code's evaluation result is + // registered by deployment id instead. + materializeDeploymentClosure: ({ deploymentId }) => { + const definition = closureDefinitions.get(deploymentId); + if (definition === undefined) { + throw new Error( + `test closure materializer: no definition registered for ${deploymentId}`, + ); + } + return Promise.resolve({ + definition, + packageDir: path.join(dataDir, "closure-package", deploymentId), + deployDir: path.join(dataDir, "closure-deploy", deploymentId), + }); + }, multistepSubprocessSpawner: recordingSpawner, multistepBinaryPath: path.join(dataDir, "workflow-child-sentinel"), }); @@ -106,6 +132,46 @@ function makeHarnessConfig(agentAddress: string): HarnessConfig { }; } +/** + * The source-ref pin every workflow frame now carries. Its `closure` is never + * fetched in these tests -- the injected materializer answers from + * `closureDefinitions` -- so an empty frozen manifest is the honest fixture. + */ +const SOURCE_REF: NonNullable["sourceRef"] = { + source: { kind: "registry", registry: "npm" }, + closure: { schemaVersion: "1", topLevel: [], entries: [] }, +}; + +/** + * Register the definition the pinned closure evaluates to for `agentAddress` + * and return the live shape the router's projection gate runs against. + */ +function stageClosureDefinition( + agentAddress: string, + stepOrder: string[], +): void { + const steps: Record> = {}; + for (const stepId of stepOrder) { + steps[stepId] = step({ + agent: buildSingleStepAgentDefinition({ + id: stepId, + systemPrompt: "", + inferencePreferences: [], + toolFactories: [], + }), + triggers: "unbounded", + }); + } + closureDefinitions.set( + deriveDeploymentId(agentAddress), + defineWorkflow({ + id: "definition-1", + trigger: { type: "mail", to: "definition-1@example.com" }, + steps, + }), + ); +} + function makeSource(provider: string): InferenceSource { return { id: `source-${provider}`, @@ -166,23 +232,24 @@ test("an unbuildable inference provider rejects the deploy before any spawn", as config: makeHarnessConfig("ins_dep_1@example.com"), hubPublicKey: hexEncode(hubKey.publicKey), workflow: { - definition: { - id: "definition-1", - triggers: [], - stepOrder: ["step-1"], - steps: { "step-1": {} }, - }, sources: { "step-1": [makeSource("unbuildable")] }, + approvedWireHash: "d".repeat(64), + sourceRef: SOURCE_REF, }, }; + stageClosureDefinition("ins_dep_1@example.com", ["step-1"]); + await expect(router.deploy(frame)).rejects.toThrow(/not registered/); expect(rejectedSources).toHaveLength(1); expect(spawnedBinaries).toEqual([]); expect(router.activeAddresses()).toEqual([]); }); -test("a malformed workflow projection is refused at the router edge", async () => { +// The deploy frame no longer carries a definition, so its arktype `narrow` +// cannot check that the sources table covers every step. That coverage is now +// checked against the CLOSURE-derived definition, after the apply. +test("a closure-derived definition whose step has no sources entry is refused", async () => { const dataDir = await makeDataDir(); const { router, spawnedBinaries } = await makeRouter(dataDir); const hubKey = await generateKeyPair(); @@ -193,17 +260,15 @@ test("a malformed workflow projection is refused at the router edge", async () = config: makeHarnessConfig("ins_dep_1@example.com"), hubPublicKey: hexEncode(hubKey.publicKey), workflow: { - definition: { - id: "definition-1", - triggers: [], - stepOrder: [], - steps: {}, - }, sources: {}, + approvedWireHash: "d".repeat(64), + sourceRef: SOURCE_REF, }, }; - await expect(router.deploy(frame)).rejects.toThrow(/stepOrder/); + stageClosureDefinition("ins_dep_1@example.com", ["step-1"]); + + await expect(router.deploy(frame)).rejects.toThrow(/sources/); expect(spawnedBinaries).toEqual([]); }); @@ -238,16 +303,14 @@ test("a single-step deploy writes the self-anchored run's grants before spawning config: { ...makeHarnessConfig(agentAddress), grants: [grant] }, hubPublicKey: hexEncode(hubKey.publicKey), workflow: { - definition: { - id: "definition-1", - triggers: [], - stepOrder: ["step-1"], - steps: { "step-1": {} }, - }, sources: { "step-1": [makeSource("openai")] }, + approvedWireHash: "d".repeat(64), + sourceRef: SOURCE_REF, }, }; + stageClosureDefinition(agentAddress, ["step-1"]); + await expect(router.deploy(frame)).rejects.toThrow( /refuses to launch a real child/, ); diff --git a/apps/sidecar/test/support/workflow-lifecycle-fixture.ts b/apps/sidecar/test/support/workflow-lifecycle-fixture.ts index fece606a3..ce6745f1e 100644 --- a/apps/sidecar/test/support/workflow-lifecycle-fixture.ts +++ b/apps/sidecar/test/support/workflow-lifecycle-fixture.ts @@ -23,6 +23,8 @@ import { type SubprocessSpawner, } from "@intx/workflow-host"; import type { AgentDeployFrame } from "@intx/types/sidecar"; +import { defineWorkflow, step, type WorkflowDefinition } from "@intx/workflow"; +import { buildSingleStepAgentDefinition } from "@intx/workflow-deploy"; import { createSidecarDeployRouter, @@ -300,6 +302,15 @@ export async function makeLifecycleFixture(opts?: { multistepSubstrateEnv: { SIDECAR_DATA_DIR: dataDir, }, + // Stand in for the real fetch + SRI-verify + layout + evaluate pass: a + // fixture cannot publish a package, so every deploy evaluates to the one + // lifecycle definition below. + materializeDeploymentClosure: ({ deploymentId }) => + Promise.resolve({ + definition: LIFECYCLE_CLOSURE_DEFINITION, + packageDir: path.join(dataDir, "closure-package", deploymentId), + deployDir: path.join(dataDir, "closure-deploy", deploymentId), + }), multistepMailRouter, multistepSignalRouter, multistepDrainRouter, @@ -319,6 +330,30 @@ export async function makeLifecycleFixture(opts?: { }; } +/** + * The definition the fixture's stubbed closure evaluates to. Source-ref is the + * only deploy lineage, so a frame carries no inline definition and this is the + * single source of the deployment's shape. + */ +const LIFECYCLE_CLOSURE_DEFINITION: WorkflowDefinition = defineWorkflow({ + id: "wf-lifecycle", + trigger: { type: "mail", to: "wf-lifecycle@example.com" }, + steps: { + "step-1": step({ + agent: buildSingleStepAgentDefinition({ + id: "step-1", + systemPrompt: "", + inferencePreferences: [], + toolFactories: [], + }), + triggers: "unbounded", + }), + }, +}); + +/** The wire hash the hub approved for `LIFECYCLE_CLOSURE_DEFINITION`. */ +export const LIFECYCLE_APPROVED_WIRE_HASH = "d".repeat(64); + export function makeWorkflowFrame(agentAddress: string): AgentDeployFrame { return { type: "agent.deploy", @@ -331,11 +366,10 @@ export function makeWorkflowFrame(agentAddress: string): AgentDeployFrame { // Boundary type assertion: the multi-step branch does not read config config: {} as AgentDeployFrame["config"], workflow: { - definition: { - id: "wf-lifecycle", - triggers: [{ type: "manual" }], - stepOrder: ["step-1"], - steps: { "step-1": { kind: "step" } }, + approvedWireHash: LIFECYCLE_APPROVED_WIRE_HASH, + sourceRef: { + source: { kind: "registry", registry: "npm" }, + closure: { schemaVersion: "1", topLevel: [], entries: [] }, }, sources: { "step-1": [ diff --git a/apps/sidecar/test/workflow-deploy-lifecycle.test.ts b/apps/sidecar/test/workflow-deploy-lifecycle.test.ts index 920b0ed1d..04d540a81 100644 --- a/apps/sidecar/test/workflow-deploy-lifecycle.test.ts +++ b/apps/sidecar/test/workflow-deploy-lifecycle.test.ts @@ -19,10 +19,11 @@ import { answerReadyHandshake, makeLifecycleFixture, makeWorkflowFrame, + LIFECYCLE_APPROVED_WIRE_HASH, } from "./support/workflow-lifecycle-fixture"; describe("workflow deployment lifecycle through the deploy router", () => { - test("a deploy frame carrying referencedDefinitions materializes each body's workflow.json and sources.json", async () => { + test("a deploy frame carrying referencedDefinitions stages each body's sources.json and no body definition", async () => { const { router, spawns, dataDir } = await makeLifecycleFixture(); const frame = makeWorkflowFrame("run_lifecycle-bodies@example.com"); if (frame.workflow === undefined) throw new Error("unreachable"); @@ -51,103 +52,58 @@ describe("workflow deployment lifecycle through the deploy router", () => { await answerReadyHandshake(spawns, 0); await deployPromise; - // The top-level definition lands where the workflow-process child's - // loadWorkflowDefinition reads it... - const assetDir = (id: string) => - path.join(dataDir, "assets", "workflow", id); - const topLevel = JSON.parse( - await fs.readFile( - path.join(assetDir("wf-lifecycle"), "workflow.json"), - "utf8", - ), - ); - expect(topLevel).toEqual(frame.workflow.definition); - - // ...and each referenced onTrigger body lands beside it under its own - // ref -- the body id -- as the definition plus the co-located - // per-step source pins the in-process body child resolves off disk. - const bodyDir = assetDir(bodyDefinition.id); - expect( - JSON.parse( - await fs.readFile(path.join(bodyDir, "workflow.json"), "utf8"), - ), - ).toEqual(bodyDefinition); + // A body child runs in-process and loses its env across a restart, so its + // per-step source pins must be durable on disk. + const bodyDir = path.join(dataDir, "assets", "workflow", bodyDefinition.id); expect( JSON.parse(await fs.readFile(path.join(bodyDir, "sources.json"), "utf8")), ).toEqual(bodySources); + + // The body DEFINITION is never staged: the run child resolves each body + // in-memory from the parent's re-verified closure. A staged copy would be + // a second, un-verified source of the body's bytes. + await expect( + fs.stat(path.join(bodyDir, "workflow.json")), + ).rejects.toThrow(); + await expect( + fs.stat(path.join(dataDir, "assets", "workflow", "wf-lifecycle")), + ).rejects.toThrow(); }); - test("a deploy frame carrying a referenced body's approvedWireHash threads REFERENCED_DEFINITION_HASHES to the spawned child and persists it for restore", async () => { + test("a deploy threads the materialized closure dir and the hub-approved hash to the child, and persists the pin for restore", async () => { const { router, spawns, dataDir } = await makeLifecycleFixture(); - const frame = makeWorkflowFrame("run_lifecycle-hashes@example.com"); - if (frame.workflow === undefined) throw new Error("unreachable"); - const bodyDefinition = { - id: "wf-lifecycle-hashed-body", - triggers: [{ type: "manual" }], - stepOrder: ["body-step"], - steps: { "body-step": { kind: "step" } }, - }; - const bodySources = { - "body-step": [ - { - id: "body-step", - provider: "anthropic", - baseURL: "https://api.anthropic.com", - apiKey: "sk-body", - model: "claude-3-5", - }, - ], - }; - frame.workflow.referencedDefinitions = [ - { - definition: bodyDefinition, - sources: bodySources, - approvedWireHash: "sha256-approved-body-hash", - }, - ]; + const frame = makeWorkflowFrame("run_lifecycle-closure@example.com"); const deployPromise = router.deploy(frame); const spawn = await answerReadyHandshake(spawns, 0); await deployPromise; - // The spawned child's env carries the approved hash keyed by the body's - // definition id -- what `resolveVerifiedBody` in the workflow-host's - // spawn-child adapter re-verifies a body spawn's recompute against. - const referencedHashes = JSON.parse( - spawn.env.REFERENCED_DEFINITION_HASHES ?? "{}", + const deploymentId = deriveDeploymentId(frame.agentAddress); + // The child EVALUATES the pinned code from this dir rather than reading an + // inert definition off disk, and re-verifies its projection against the + // hub-approved hash -- never a sidecar recompute. + expect(spawn.env.CLOSURE_PACKAGE_DIR).toBe( + path.join(dataDir, "closure-package", deploymentId), ); - expect(referencedHashes).toEqual({ - [bodyDefinition.id]: "sha256-approved-body-hash", - }); + expect(spawn.env.DEFINITION_HASH).toBe(LIFECYCLE_APPROVED_WIRE_HASH); + expect(spawn.env.WORKFLOW_DEFINITION_REF).toBeUndefined(); + expect(spawn.env.REFERENCED_DEFINITION_HASHES).toBeUndefined(); - // ...and it survives a restart: the durable deployment record carries - // the same map so a boot-time restore rebuilds the identical spawn env - // without a hub round-trip. - const recordFile = path.join( - dataDir, - "workflow-runs", - deriveDeploymentId(frame.agentAddress), - "deployment.json", + // The record carries what a boot-time restore needs to re-materialize the + // same closure and re-verify it against the same anchor. + const record = JSON.parse( + await fs.readFile( + path.join(dataDir, "workflow-runs", deploymentId, "deployment.json"), + "utf8", + ), ); - const record = JSON.parse(await fs.readFile(recordFile, "utf8")); - expect(record.referencedDefinitionHashes).toEqual({ - [bodyDefinition.id]: "sha256-approved-body-hash", + expect(record.approvedWireHash).toBe(LIFECYCLE_APPROVED_WIRE_HASH); + expect(record.sourceRef.source).toEqual({ + kind: "registry", + registry: "npm", }); }); - test("a deploy frame with no referenced bodies threads an empty REFERENCED_DEFINITION_HASHES map", async () => { - const { router, spawns } = await makeLifecycleFixture(); - const frame = makeWorkflowFrame("run_lifecycle-no-bodies@example.com"); - - const deployPromise = router.deploy(frame); - const spawn = await answerReadyHandshake(spawns, 0); - await deployPromise; - - expect(JSON.parse(spawn.env.REFERENCED_DEFINITION_HASHES ?? "")).toEqual( - {}, - ); - }); - test("a workflow frame is accepted: the child spawns, the address goes live, and a durable record lands", async () => { const { router, spawns, dataDir } = await makeLifecycleFixture(); const frame = makeWorkflowFrame("run_lifecycle-accept@example.com"); diff --git a/apps/sidecar/test/workflow-substrate-factory-run-child.test.ts b/apps/sidecar/test/workflow-substrate-factory-run-child.test.ts index 25cae69dc..01b62319d 100644 --- a/apps/sidecar/test/workflow-substrate-factory-run-child.test.ts +++ b/apps/sidecar/test/workflow-substrate-factory-run-child.test.ts @@ -130,7 +130,6 @@ async function makeRunChild( substrate, workflowRunRepoId: WORKFLOW_RUN_REPO_ID, workflowRunRef: REF, - workflowDefinitionRef: REF, principal: PRINCIPAL, scheduler: createInMemoryScheduler({ repoStore: createInMemoryRepoStore(), diff --git a/apps/sidecar/test/workflow-substrate-factory-suspendable-child.test.ts b/apps/sidecar/test/workflow-substrate-factory-suspendable-child.test.ts index 0a850b823..3a89aaf54 100644 --- a/apps/sidecar/test/workflow-substrate-factory-suspendable-child.test.ts +++ b/apps/sidecar/test/workflow-substrate-factory-suspendable-child.test.ts @@ -149,7 +149,6 @@ function makeSpawner( substrate, workflowRunRepoId: WORKFLOW_RUN_REPO_ID, workflowRunRef: REF, - workflowDefinitionRef: REF, principal: PRINCIPAL, scheduler: createInMemoryScheduler({ repoStore: createInMemoryRepoStore(), @@ -240,7 +239,6 @@ describe("createSidecarSpawnSuspendableChild", () => { substrate, workflowRunRepoId: WORKFLOW_RUN_REPO_ID, workflowRunRef: REF, - workflowDefinitionRef: REF, principal: PRINCIPAL, scheduler: createInMemoryScheduler({ repoStore: createInMemoryRepoStore(), diff --git a/docs/revendor-inventory.md b/docs/revendor-inventory.md index 3c86ba89a..7dffe8808 100644 --- a/docs/revendor-inventory.md +++ b/docs/revendor-inventory.md @@ -368,12 +368,12 @@ one on the old pin leaves the frame contract split down the middle. Open conversion sites: -| Site | What it needs | -| --------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | -| ~~`packages/folded-runs/src/launch.ts` (`deployAtHead`), `wake.ts`~~ | **Done** — see "Conversion step 2" below. | -| `apps/sidecar/src/workflow-host-wiring/index.ts`, `asset-materialization.ts` | Stop writing `workflow.json` and stop reading `projection.definition`; stage the closure instead. | -| `apps/sidecar/src/workflow-substrate-factory/index.ts`, `child-runtime.ts`, `config.ts` | Drop `WORKFLOW_DEFINITION_REPO_ID`/`_REF`; in-memory child spawn; `closurePackageDir` plumbing. | -| `apps/sidecar/src/workflow-deployment-record.ts` | Drop `referencedDefinitionHashes`; carry the grant-walk snapshot. | +| Site | What it needs | +| ------------------------------------------------------------------------------------------- | ----------------------------------------- | +| ~~`packages/folded-runs/src/launch.ts` (`deployAtHead`), `wake.ts`~~ | **Done** — see "Conversion step 2" below. | +| ~~`apps/sidecar/src/workflow-host-wiring/index.ts`, `asset-materialization.ts`~~ | **Done** — see "Conversion step 3" below. | +| ~~`apps/sidecar/src/workflow-substrate-factory/index.ts`, `child-runtime.ts`, `config.ts`~~ | **Done** — see "Conversion step 3" below. | +| ~~`apps/sidecar/src/workflow-deployment-record.ts`~~ | **Done** — see "Conversion step 3" below. | Upstream's own diff over the same span is the reference implementation: `apps/sidecar/src/workflow-substrate-factory.ts` and @@ -478,3 +478,73 @@ Nothing caught this before because the in-memory definition was never parsed. Either the handle shape changes here (and with it the `env.credentials.resolve("mcp:")` key `@corbits/mcp-tools` uses) or upstream widens the handle grammar; it is not fixed in this change. + +#### Conversion step 3: the sidecar is on closures + +The execution host is converted. A deploy no longer writes a definition +into the deploy tree and reads it back: it materializes the frame's frozen +closure, evaluates the pinned code, and runs that. The boot-time restore +replays the same pin through the same helper, so both paths reach the +runnable definition by one computation rather than two that can drift. + +What moved: + +- **Closure staging.** `workflow-host-wiring/closure-staging.ts` owns the + durable per-deployment source stores (plain-file and indexed-git, + siblings of the reclaimed instance dir so they survive a restart with no + re-delivery), the `assetId -> mount` resolution both paths derive from + the pin alone, and the apply. It is an injectable router dependency, so + a test stands in for fetch + SRI-verify + layout + evaluate without + publishing a package. +- **`CLOSURE_PACKAGE_DIR` exists.** It is threaded on the frozen substrate + env, so the run child evaluates the pinned code and re-verifies its own + projection against `DEFINITION_HASH` — which is now the hub's + `approvedWireHash`, never a sidecar recompute. A frame carrying no + approved hash fails closed rather than substituting one. +- **The probe answers.** `createWorkflowProbeExecutor` is wired at the boot + edge against a closure materializer rooted in the sidecar data dir, so + `workflow.probe.request` returns a real inert projection, its advisory + grant set, and its wire hash instead of `workflow.probe.error`. +- **Child spawns are in-memory.** `createWorkflowSpawnChild` / + `createWorkflowSpawnSuspendableChild` are gone. A rung lifts its inline + children to refs and serves grandchildren from that map, so no rung reads + a definition off disk at any depth. An onTrigger body's `sources.json` is + still staged (a body child is in-process and loses its env across a + restart); its definition is not. +- **The deployment record carries the pin.** `referencedDefinitionHashes` + is gone; `approvedWireHash` and `sourceRef` are required, so a record + that cannot be restored fails at the scan boundary rather than + half-restoring. +- **`WORKFLOW_DEFINITION_REPO_ID`/`_REF` are gone.** What survives is + `WORKFLOW_DEFINITION_ID`: identity for the run-authenticated + capabilities route a step tool calls, never a repo to read from. + +Four modules are near-verbatim copies of upstream's own sidecar at +`4ed8baf4` — the probe handler, the closure materializer, the closure +apply, and the inline source-asset delivery — plus `bin/workflow-probe-child`. +`VENDORED.md` records them and the two adaptations the fork's module layout +forced. + +##### What is still unproven + +`bun run typecheck` is green repo-wide and the sidecar, folded-runs, and +chat suites pass, but **no run has executed end to end on these rails**. +The remaining wire, in order: + +1. **The MCP credential-handle defect from step 2 still stands.** Any + MCP-pinned launch fails closed at render time (`mcp:` is not a + legal `ToolCredentialHandle`). It gates a real chat launch, not the + deploy path itself. +2. **Nothing has published or committed a real per-run source package + through the deploy.** Every test injects the closure materializer, so + the fetch/SRI/layout/evaluate path itself is exercised only by + upstream's own tests at the vendored pin, never against a + `renderAgentRuntimeSourceTree` output. +3. **The probe has never been driven by a hub.** The executor is wired and + typechecks; nothing has sent it a `workflow.probe.request` frame. +4. **The pinned tool-package arm is untouched, deliberately.** Upstream + went all-source-tools (`req.agent.toolFactories`); workbench's + `agent-runtime` pins tool packages instead, so the sidecar keeps + `materializeStepTools`. Whether the source-format deploy still stages a + `tool-packages-manifest.json` for those pins is the first thing an + end-to-end run will answer.