diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 1e072578e..b1aa9b3c8 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -1036,7 +1036,6 @@ export async function createHub(config: HubConfig) { sidecarRouter, eventCollectors, credentialCipher, - hubPublicKey, toolGrantsForPins, mcpCredentialBindingsFor, noopInferenceBaseUrl: `${config.baseUrl}/api/chat/noop-inference`, @@ -1594,7 +1593,6 @@ export async function createHub(config: HubConfig) { assetService, sidecarRouter, eventCollectors, - hubPublicKey, toolGrantsForPins, mcpCredentialBindingsFor, cryptoProviderCache: foldedRunCryptoProviders, @@ -2002,7 +2000,6 @@ export async function createHub(config: HubConfig) { sidecarRouter, eventCollectors, credentialCipher, - hubPublicKey, toolGrantsForPins, mcpCredentialBindingsFor, cryptoProviderCache: foldedRunCryptoProviders, diff --git a/apps/hub/src/routine-launcher.test.ts b/apps/hub/src/routine-launcher.test.ts index 4b145d336..146aaf08b 100644 --- a/apps/hub/src/routine-launcher.test.ts +++ b/apps/hub/src/routine-launcher.test.ts @@ -114,7 +114,6 @@ function buildLauncher(overrides: { definition?: unknown } = {}) { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -232,7 +231,6 @@ describe("createHubRoutineLauncher — delivery workbench", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -259,7 +257,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -297,7 +294,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -324,7 +320,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -356,7 +351,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, @@ -378,7 +372,6 @@ describe("createHubRoutineLauncher — recurring-task bridge", () => { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never }, diff --git a/bun.lock b/bun.lock index 977cc66a2..d41711a93 100644 --- a/bun.lock +++ b/bun.lock @@ -229,6 +229,21 @@ "typescript": "catalog:", }, }, + "packages/agent-runtime": { + "name": "@corbits/agent-runtime", + "version": "0.0.1", + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/types": "workspace:*", + "@intx/workflow": "workspace:*", + "@intx/workflow-deploy": "workspace:*", + "arktype": "catalog:", + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:", + }, + }, "packages/api-query": { "name": "@corbits/api-query", "version": "0.0.1", @@ -527,7 +542,7 @@ }, "packages/connections-tools": { "name": "@corbits/connections-tools", - "version": "0.0.3", + "version": "0.0.4", "dependencies": { "@intx/agent": "workspace:*", "@intx/types": "workspace:*", @@ -635,6 +650,7 @@ "version": "0.0.1", "dependencies": { "@corbits/agent-lifecycle": "workspace:*", + "@corbits/agent-runtime": "workspace:*", "@intx/crypto": "workspace:*", "@intx/db": "workspace:*", "@intx/hub-api": "workspace:*", @@ -1842,6 +1858,8 @@ "@corbits/agent-lifecycle": ["@corbits/agent-lifecycle@workspace:packages/agent-lifecycle"], + "@corbits/agent-runtime": ["@corbits/agent-runtime@workspace:packages/agent-runtime"], + "@corbits/api-query": ["@corbits/api-query@workspace:packages/api-query"], "@corbits/approvals": ["@corbits/approvals@workspace:packages/approvals"], diff --git a/docs/revendor-inventory.md b/docs/revendor-inventory.md index 3c8cd099e..3c86ba89a 100644 --- a/docs/revendor-inventory.md +++ b/docs/revendor-inventory.md @@ -366,11 +366,11 @@ approval bundle, migrations 0082/0083) and `hub-api` (run trigger) all move together, and `apps/sidecar` reads the frame both sides write. Leaving any one on the old pin leaves the frame contract split down the middle. -Open conversion sites, all blocked on that one decision: +Open conversion sites: | Site | What it needs | | --------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | -| `packages/folded-runs/src/launch.ts` (`deployAtHead`), `wake.ts` | A code-sourced deploy for the folded single-step run — the root blocker. | +| ~~`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. | @@ -380,3 +380,101 @@ Upstream's own diff over the same span is the reference implementation: `workflow-host-wiring.ts` at `4ed8baf4` show every one of these conversions against the same contracts, and `apps/sidecar`'s `VENDORED.md` row stays at `59f5e7b9` until workbench's fork is reconciled with them. + +### Conversion step 1: `packages/agent-runtime` + +`packages/agent-runtime` holds the definition builder every workbench +agent run deploys. `AgentRuntimeConfig` is the arktype contract for +everything that differs per run — the mailbox it answers on, its system +prompt, its resolved inference chain, its tool-package pins, its +credential bindings — and the config's `mode` selects the shape: +`buildAgentRuntimeWorkflow` returns either the folded unbounded step or +the per-turn `onTrigger` section. Because the mode lives in the config, +the deploy front keeps one parameter set and no call site ever branches +on which shape it wants; `deployCodeSourcedWorkflow` already enumerates +inert `onTrigger` bodies on every deploy, so the section shape needs +nothing extra from the API. + +#### There is no out-of-band config channel — the config IS the bytes + +The obvious design, one static published package whose entry reads a +per-run config from its environment, does not work at this pin and +cannot be made to work by the sidecar. + +The approval probe and the run child each evaluate the entry module +independently, and the child refuses to run a definition whose recomputed +wire hash differs from the approved one +(`workflow-host/src/child/verified-definition-loader.ts`). The hashed +preimage (`workflow/src/live-inert-projector.ts`) covers the trigger +address, the agent's system prompt, its `(provider, model)` pairs, its +tool-package pins, and the definition's credential bindings — every field +of the config. So a config read from anywhere outside the closure's bytes +diverges between the two evaluations and fails closed. And there is +nowhere to read it from anyway: `WorkflowDefinitionSource` has no overlay +or params on any variant, `AgentDeployWorkflow` carries no config bag +(the code-sourced route builds `HarnessConfig` with an empty +`systemPrompt`, empty `tools`, empty `grants`), `SpawnTimeEnv` has no +config field, and `WorkflowProbeRequestFrame` carries no env at all — so +even a sidecar willing to inject one could not make the probe see it. + +`renderAgentRuntimeSourceTree` is the consequence: it renders a thin +per-run package — a `package.json` plus a four-line entry module that +pins `@corbits/agent-runtime` and calls `buildAgentRuntimeWorkflow` with +the run's config as a literal. All the behaviour stays in the one +versioned package; what varies per run is a JSON literal inside the +hashed bytes. A host commits that tree into a `workflow`-kind asset and +deploys `source.kind: "asset"`, `package.format: "source"`, `commitSha` — +the only source variant whose pin is cheap enough to mint per run, since +the registry and tarball variants each need a publish. + +Two in-tree prerequisites remain for any of this to execute, both already +on the conversion table above: nothing produces `CLOSURE_PACKAGE_DIR`, and +no `WorkflowProbeExecutor` is wired on the sidecar, so every probe +currently answers `workflow.probe.error`. Both are conversion step 2. + +#### Conversion step 2: `deployAtHead` is on the seam + +`deployAtHead` no longer synthesizes a definition. It renders the run's +config into a per-run `@corbits/agent-runtime` package, commits that tree +into the run's OWN `workflow`-kind definition asset on +`refs/heads/runs/`, and deploys the resulting `commitSha` through +`deployAdoptedWorkflowFromSource` — the adopting shared-capacity front, +the only one that accepts a pre-minted anchor row and threads a +`credentialCipher`. `wake.ts` takes the same path. The old +synthesize-in-memory branch is deleted, not gated. + +Reuse, not a second asset: one definition asset can back many runs (a +chat's workbench host, an invited agent's every launch), so each run gets +its own ref inside that asset rather than its own asset per deploy. The +pin is the `commitSha`, so the ref is bookkeeping. + +The step's input selector became the config's `mode`: `step` (with the +workbench host's optional `literalInput`, the CL-6164 pin) or `section` +with a per-turn timeout. The Phase 1.3 swap changes a caller's argument, +never a branch inside `deployAtHead`. Section mode authors +`onBodyFailure: "continue"`, which the vendored surface now carries +through the live→inert projection. + +##### What still blocks EXECUTION + +Deploying works at the type and call level; nothing has run it end to +end, because the two sidecar prerequisites are untouched: nothing +produces `CLOSURE_PACKAGE_DIR`, and no `WorkflowProbeExecutor` is wired, +so every probe still answers `workflow.probe.error`. The remaining +in-tree typecheck failures are exactly the sidecar rows in the table +above — `projection.definition` reads, `createWorkflowSpawnChild` / +`createWorkflowSpawnSuspendableChild`, `SpawnTimeEnv.referencedDefinitionHashes`, +and `RunWorkflowChildBindings.workflowDefinitionRepoId`. + +##### Defect surfaced by the conversion + +`renderAgentRuntimeSourceTree` parses the config before writing it, which +is the first time a folded run's credential bindings are validated +against the platform's `CredentialBinding` schema. `apps/hub`'s +`mcp-credential-bindings.ts` mints `handle: "mcp:"`, and +`ToolCredentialHandle` is `/^[a-z0-9][a-z0-9._-]*$/` — the colon is not +in it, so every MCP-pinned launch would now fail closed at render time. +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. diff --git a/packages/agent-runtime/package.json b/packages/agent-runtime/package.json new file mode 100644 index 000000000..e1a35d50b --- /dev/null +++ b/packages/agent-runtime/package.json @@ -0,0 +1,26 @@ +{ + "name": "@corbits/agent-runtime", + "private": true, + "description": "The versioned definition builder every workbench agent run deploys, plus the renderer that pins it into a per-run code-sourced workflow package", + "version": "0.0.1", + "license": "LGPL-2.1-or-later", + "type": "module", + "exports": { + ".": "./src/index.ts" + }, + "scripts": { + "typecheck": "tsc --noEmit", + "test": "bun test" + }, + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/types": "workspace:*", + "@intx/workflow": "workspace:*", + "@intx/workflow-deploy": "workspace:*", + "arktype": "catalog:" + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:" + } +} diff --git a/packages/agent-runtime/src/config.test.ts b/packages/agent-runtime/src/config.test.ts new file mode 100644 index 000000000..cbaad13f3 --- /dev/null +++ b/packages/agent-runtime/src/config.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, test } from "bun:test"; + +import { parseAgentRuntimeConfig, type AgentRuntimeConfig } from "./config"; + +const stepConfig: AgentRuntimeConfig = { + workflowId: "wf_run_a", + agentId: "run_a", + triggerAddress: "run_a@bench.example", + systemPrompt: "You are helpful.", + inferencePreferences: [{ provider: "acme", model: "acme-1" }], + toolPackagePins: [], + credentialBindings: [], + mode: { kind: "step" }, +}; + +describe("parseAgentRuntimeConfig", () => { + test("accepts a step-mode config", () => { + expect(parseAgentRuntimeConfig(stepConfig)).toEqual(stepConfig); + }); + + test("accepts a section-mode config with its turn timeout", () => { + const sectionConfig: AgentRuntimeConfig = { + ...stepConfig, + mode: { kind: "section", turnTimeoutMs: 60_000 }, + }; + expect(parseAgentRuntimeConfig(sectionConfig)).toEqual(sectionConfig); + }); + + test("rejects an empty inference chain rather than building a modelless agent", () => { + expect(() => + parseAgentRuntimeConfig({ ...stepConfig, inferencePreferences: [] }), + ).toThrow(/invalid agent-runtime config/); + }); + + test("rejects a section mode with no turn timeout", () => { + expect(() => + parseAgentRuntimeConfig({ ...stepConfig, mode: { kind: "section" } }), + ).toThrow(/invalid agent-runtime config/); + }); + + test("rejects an unknown mode", () => { + expect(() => + parseAgentRuntimeConfig({ ...stepConfig, mode: { kind: "swarm" } }), + ).toThrow(/invalid agent-runtime config/); + }); + + test("rejects an empty trigger address", () => { + expect(() => + parseAgentRuntimeConfig({ ...stepConfig, triggerAddress: "" }), + ).toThrow(/invalid agent-runtime config/); + }); +}); diff --git a/packages/agent-runtime/src/config.ts b/packages/agent-runtime/src/config.ts new file mode 100644 index 000000000..664345636 --- /dev/null +++ b/packages/agent-runtime/src/config.ts @@ -0,0 +1,84 @@ +// Everything about an agent run that its deployed definition must know: +// which mailbox it answers on, what it is told to be, which models it +// may use, which tool packages and credentials it carries, and which of +// the two shapes it takes. +// +// This config is DEPLOY-TIME data, and under the workflow.json +// retirement it has to live inside the deployed package's own bytes. +// The approval probe and the run child each evaluate the deployment's +// entry module independently and the child refuses to run a definition +// whose recomputed wire hash differs from the approved one; the hashed +// projection covers the system prompt, the trigger address, the model +// pairs, the tool package pins, and the credential bindings — every +// field here. So a config delivered out of band (an env var, a file the +// sidecar drops next to the entry) diverges between the two evaluations +// and fails closed. `./source-tree.ts` renders it into the bytes +// instead. +import { type } from "arktype"; +import { CredentialBinding } from "@intx/types"; +import { ToolPackagePin } from "@intx/types/tool-packages"; + +const InferencePreference = type({ + provider: "string > 0", + model: "string > 0", + "parameters?": "Record", +}); + +/** + * One unbounded agent step on the run's own address: the folded + * conversational shape. Every inbound mail is another turn of the same + * step, and the run never completes on its own. + */ +const StepMode = type({ + kind: "'step'", + /** + * When present, the step reads this fixed value instead of the + * triggering mail's `trigger.payload`. A run whose system prompt + * forbids acting on what it receives (the workbench host) pins a + * literal here so attachments-only mail — whose `content` is + * legitimately empty — cannot crash the step before it opens. + */ + "literalInput?": "unknown", +}); + +/** + * One long-lived `onTrigger` section on the run's own address: each + * inbound mail is one occurrence, run as its own child run with its own + * id and event log, so a reply is traceable. + */ +const SectionMode = type({ + kind: "'section'", + /** Per-occurrence timeout, enforced on the body's one step. */ + turnTimeoutMs: "number.integer > 0", +}); + +export const AgentRuntimeConfig = type({ + /** Definition id; also the base of the section body's id. */ + workflowId: "string > 0", + /** The step agent's id — the folded run's instance id. */ + agentId: "string > 0", + /** The mailbox this deployment answers on. */ + triggerAddress: "string > 0", + systemPrompt: "string", + /** Resolved catalog chain, in deploy order; the gate approves these. */ + inferencePreferences: InferencePreference.array().atLeastLength(1), + /** Tool packages the step agent carries; no inline tool factories. */ + toolPackagePins: ToolPackagePin.array(), + /** Definition-level bindings the host's per-step snapshot derives from. */ + credentialBindings: CredentialBinding.array(), + mode: StepMode.or(SectionMode), +}); +export type AgentRuntimeConfig = typeof AgentRuntimeConfig.infer; + +/** + * Parse `raw` as an `AgentRuntimeConfig`, throwing on any malformed + * shape. A deploy whose config does not parse must fail before a + * definition exists, not build a half-configured agent. + */ +export function parseAgentRuntimeConfig(raw: unknown): AgentRuntimeConfig { + const parsed = AgentRuntimeConfig(raw); + if (parsed instanceof type.errors) { + throw new Error(`invalid agent-runtime config: ${parsed.summary}`); + } + return parsed; +} diff --git a/packages/agent-runtime/src/definition.test.ts b/packages/agent-runtime/src/definition.test.ts new file mode 100644 index 000000000..27d2cb58e --- /dev/null +++ b/packages/agent-runtime/src/definition.test.ts @@ -0,0 +1,189 @@ +import { describe, expect, test } from "bun:test"; +import { projectLiveToInert } from "@intx/workflow"; + +import type { AgentRuntimeConfig } from "./config"; +import { + AGENT_RUNTIME_SECTION_ID, + AGENT_RUNTIME_STEP_ID, + AGENT_RUNTIME_TURN_STEP_ID, + agentRuntimeTurnRunId, + buildAgentRuntimeWorkflow, +} from "./definition"; + +const baseConfig: AgentRuntimeConfig = { + workflowId: "wf_run_a", + agentId: "run_a", + triggerAddress: "run_a@bench.example", + systemPrompt: "You are helpful.", + inferencePreferences: [ + { provider: "acme", model: "acme-1" }, + { provider: "acme", model: "acme-2" }, + ], + toolPackagePins: [], + credentialBindings: [], + mode: { kind: "step" }, +}; + +describe("buildAgentRuntimeWorkflow — step mode", () => { + test("builds one unbounded agent step on the config's own address", () => { + const definition = buildAgentRuntimeWorkflow(baseConfig); + + expect(definition.id).toBe("wf_run_a"); + expect(definition.stepOrder).toEqual([AGENT_RUNTIME_STEP_ID]); + const stepPrimitive = definition.steps[AGENT_RUNTIME_STEP_ID]; + expect(stepPrimitive?.kind).toBe("step"); + expect(definition.triggers).toEqual([ + { type: "mail", to: "run_a@bench.example" }, + ]); + }); + + test("the step's trigger budget is unbounded, so a run never goes silent after one reply", () => { + const definition = buildAgentRuntimeWorkflow(baseConfig); + const stepPrimitive = definition.steps[AGENT_RUNTIME_STEP_ID]; + + expect(stepPrimitive).toMatchObject({ triggers: "unbounded" }); + }); + + test("carries the config's system prompt and inference chain onto the step agent", () => { + const definition = buildAgentRuntimeWorkflow(baseConfig); + + expect(definition.steps[AGENT_RUNTIME_STEP_ID]).toMatchObject({ + agent: { + id: "run_a", + systemPrompt: "You are helpful.", + inference: { sources: baseConfig.inferencePreferences }, + }, + }); + }); + + test("carries the config's tool package pins onto the step agent", () => { + const definition = buildAgentRuntimeWorkflow({ + ...baseConfig, + toolPackagePins: [{ name: "@corbits/mcp-tools", version: "0.0.1" }], + }); + + expect(definition.steps[AGENT_RUNTIME_STEP_ID]).toMatchObject({ + agent: { + toolPackagePins: [{ name: "@corbits/mcp-tools", version: "0.0.1" }], + }, + }); + }); + + test("declares the config's credential bindings on the definition itself", () => { + const credentialBindings = [ + { + package: "@corbits/mcp-tools", + handle: "mcp:notion", + provider: "notion", + locator: "tenant" as const, + }, + ]; + const definition = buildAgentRuntimeWorkflow({ + ...baseConfig, + credentialBindings, + }); + + expect(definition.credentialBindings).toEqual(credentialBindings); + }); + + test("omits credentialBindings entirely when the config declares none", () => { + expect( + buildAgentRuntimeWorkflow(baseConfig).credentialBindings, + ).toBeUndefined(); + }); + + test("pins the step's input to a literal when the config supplies one", () => { + const definition = buildAgentRuntimeWorkflow({ + ...baseConfig, + mode: { kind: "step", literalInput: "wake up" }, + }); + + expect(definition.steps[AGENT_RUNTIME_STEP_ID]).toMatchObject({ + input: { literal: "wake up" }, + }); + }); + + test("leaves the step reading its real trigger payload by default", () => { + const stepPrimitive = + buildAgentRuntimeWorkflow(baseConfig).steps[AGENT_RUNTIME_STEP_ID]; + + expect(stepPrimitive).not.toMatchObject({ input: { literal: undefined } }); + }); +}); + +describe("buildAgentRuntimeWorkflow — section mode", () => { + const sectionConfig: AgentRuntimeConfig = { + ...baseConfig, + mode: { kind: "section", turnTimeoutMs: 45_000 }, + }; + + test("builds an onTrigger section on the config's address, not a plain step", () => { + const definition = buildAgentRuntimeWorkflow(sectionConfig); + + expect(definition.stepOrder).toEqual([AGENT_RUNTIME_SECTION_ID]); + expect(definition.steps[AGENT_RUNTIME_SECTION_ID]).toMatchObject({ + kind: "onTrigger", + on: { type: "mail", to: "run_a@bench.example" }, + }); + }); + + test("the section's body is one agent step carrying the per-turn timeout", () => { + const section = + buildAgentRuntimeWorkflow(sectionConfig).steps[AGENT_RUNTIME_SECTION_ID]; + + expect(section).toMatchObject({ + body: { + inline: { + id: "wf_run_a_body", + steps: { + [AGENT_RUNTIME_TURN_STEP_ID]: { + kind: "step", + timeout: 45_000, + agent: { systemPrompt: "You are helpful." }, + }, + }, + }, + }, + }); + }); + + test("authors onBodyFailure 'continue' so a failed turn re-arms the section", () => { + const section = + buildAgentRuntimeWorkflow(sectionConfig).steps[AGENT_RUNTIME_SECTION_ID]; + + expect(section).toMatchObject({ onBodyFailure: "continue" }); + }); + + test("the section's failure policy survives the live→inert projection", () => { + const projected = projectLiveToInert( + buildAgentRuntimeWorkflow(sectionConfig), + ); + const section = projected.steps[AGENT_RUNTIME_SECTION_ID]; + + expect(section).toMatchObject({ + kind: "onTrigger", + onBodyFailure: "continue", + }); + }); + + test("the mode alone selects the shape — same config fields, different definition", () => { + const asStep = buildAgentRuntimeWorkflow(baseConfig); + const asSection = buildAgentRuntimeWorkflow(sectionConfig); + + expect(asStep.steps[AGENT_RUNTIME_STEP_ID]?.kind).toBe("step"); + expect(asSection.steps[AGENT_RUNTIME_SECTION_ID]?.kind).toBe("onTrigger"); + expect(asStep.id).toBe(asSection.id); + }); +}); + +describe("agentRuntimeTurnRunId", () => { + test("matches the runtime's __ child-run scheme", () => { + expect(agentRuntimeTurnRunId(0)).toBe(`${AGENT_RUNTIME_SECTION_ID}__0`); + expect(agentRuntimeTurnRunId(7)).toBe(`${AGENT_RUNTIME_SECTION_ID}__7`); + }); + + test("rejects a non-integer or negative occurrence", () => { + expect(() => agentRuntimeTurnRunId(-1)).toThrow(); + expect(() => agentRuntimeTurnRunId(1.5)).toThrow(); + }); +}); diff --git a/packages/agent-runtime/src/definition.ts b/packages/agent-runtime/src/definition.ts new file mode 100644 index 000000000..d47c631de --- /dev/null +++ b/packages/agent-runtime/src/definition.ts @@ -0,0 +1,142 @@ +// Builds the workflow definition a workbench agent run executes, from +// the run's deploy-time config alone. +// +// Two shapes, selected by `config.mode`, never by a branch in the deploy +// API: the deploy front takes one parameter set and the shape is purely +// whatever this module evaluates to. +// +// `step` is the folded conversational run — one unbounded agent step +// servicing every inbound mail as another turn. Its `triggers: +// "unbounded"` budget is the whole reason this is authored rather than +// wrapped: the platform's default budget of 1 makes a run go silent +// after its first reply. +// +// `section` is the per-turn shape (CL-6329) — an `onTrigger` section +// whose body is one agent step, so every message becomes an occurrence +// with its own child run id and event log. +// +// Section mode authors `onBodyFailure: "continue"`, the failure edge +// that keeps a section subscribed after a failed turn: a conversation +// whose agent threw on one message must still answer the next, and the +// primitive's default (`"end"`) retires the whole run instead. The +// vendored surface carries the field through the live→inert projection, +// so the policy reaches the hub's frozen projection rather than being +// silently dropped before deploy. +import { buildSingleStepAgentDefinition } from "@intx/workflow-deploy"; +import { defineWorkflow, onTrigger, step } from "@intx/workflow"; +import type { WorkflowDefinition } from "@intx/workflow"; + +import type { AgentRuntimeConfig } from "./config"; + +/** The step id of the folded conversational run's one step. */ +export const AGENT_RUNTIME_STEP_ID = "default"; + +/** The section's step id in `section` mode. */ +export const AGENT_RUNTIME_SECTION_ID = "turn"; + +/** The body step inside one section occurrence — the agent that answers. */ +export const AGENT_RUNTIME_TURN_STEP_ID = "reply"; + +/** + * The child run id occurrence `occurrence` runs under. The runtime names + * an occurrence `__` (see `onTriggerBodyRef` in + * `@intx/workflow`), so the section id plus a zero-based occurrence + * index is the whole derivation. + */ +export function agentRuntimeTurnRunId(occurrence: number): string { + if (!Number.isInteger(occurrence) || occurrence < 0) { + throw new Error( + "agentRuntimeTurnRunId requires a non-negative integer occurrence", + ); + } + return `${AGENT_RUNTIME_SECTION_ID}__${String(occurrence)}`; +} + +function buildTurnAgent(config: AgentRuntimeConfig, id: string) { + return buildSingleStepAgentDefinition({ + id, + systemPrompt: config.systemPrompt, + inferencePreferences: config.inferencePreferences, + toolFactories: [], + toolPackagePins: config.toolPackagePins, + }); +} + +function buildFoldedStepWorkflow( + config: AgentRuntimeConfig, + literalInput: unknown, + hasLiteralInput: boolean, +): WorkflowDefinition { + const steps = { + [AGENT_RUNTIME_STEP_ID]: step({ + agent: buildTurnAgent(config, config.agentId), + triggers: "unbounded" as const, + ...(hasLiteralInput ? { input: { literal: literalInput } } : {}), + }), + }; + return config.credentialBindings.length > 0 + ? defineWorkflow({ + id: config.workflowId, + trigger: { type: "mail", to: config.triggerAddress }, + credentialBindings: config.credentialBindings, + steps, + }) + : defineWorkflow({ + id: config.workflowId, + trigger: { type: "mail", to: config.triggerAddress }, + steps, + }); +} + +function buildSectionWorkflow( + config: AgentRuntimeConfig, + turnTimeoutMs: number, +): WorkflowDefinition { + const body = defineWorkflow({ + id: `${config.workflowId}_body`, + trigger: { type: "mail", to: config.triggerAddress }, + steps: { + [AGENT_RUNTIME_TURN_STEP_ID]: step({ + agent: buildTurnAgent(config, AGENT_RUNTIME_TURN_STEP_ID), + timeout: turnTimeoutMs, + }), + }, + }); + const steps = { + [AGENT_RUNTIME_SECTION_ID]: onTrigger({ + on: { type: "mail" as const, to: config.triggerAddress }, + body, + onBodyFailure: "continue", + }), + }; + return config.credentialBindings.length > 0 + ? defineWorkflow({ + id: config.workflowId, + trigger: { type: "mail", to: config.triggerAddress }, + credentialBindings: config.credentialBindings, + steps, + }) + : defineWorkflow({ + id: config.workflowId, + trigger: { type: "mail", to: config.triggerAddress }, + steps, + }); +} + +/** + * Build the run's definition from its deploy-time config. The config's + * `mode` selects the shape; every other field is the same per-run data + * either shape needs. + */ +export function buildAgentRuntimeWorkflow( + config: AgentRuntimeConfig, +): WorkflowDefinition { + if (config.mode.kind === "section") { + return buildSectionWorkflow(config, config.mode.turnTimeoutMs); + } + return buildFoldedStepWorkflow( + config, + config.mode.literalInput, + "literalInput" in config.mode, + ); +} diff --git a/packages/agent-runtime/src/index.ts b/packages/agent-runtime/src/index.ts new file mode 100644 index 000000000..fb449cf06 --- /dev/null +++ b/packages/agent-runtime/src/index.ts @@ -0,0 +1,15 @@ +export { AgentRuntimeConfig, parseAgentRuntimeConfig } from "./config"; +export { + AGENT_RUNTIME_SECTION_ID, + AGENT_RUNTIME_STEP_ID, + AGENT_RUNTIME_TURN_STEP_ID, + agentRuntimeTurnRunId, + buildAgentRuntimeWorkflow, +} from "./definition"; +export { AGENT_RUNTIME_PACKAGE_NAME, AGENT_RUNTIME_PACKAGE_RANGE } from "./pin"; +export { + AGENT_RUNTIME_ENTRY_PATH, + renderAgentRuntimeSourceTree, + type AgentRuntimeSourceTree, + type RenderAgentRuntimeSourceTreeInput, +} from "./source-tree"; diff --git a/packages/agent-runtime/src/pin.ts b/packages/agent-runtime/src/pin.ts new file mode 100644 index 000000000..0a23175e9 --- /dev/null +++ b/packages/agent-runtime/src/pin.ts @@ -0,0 +1,16 @@ +/** + * The package name a rendered per-run workflow tree depends on and + * imports its builder from. Declared next to the package rather than + * re-typed in the renderer's template. + */ +export const AGENT_RUNTIME_PACKAGE_NAME = "@corbits/agent-runtime"; + +/** + * The dependency range a rendered per-run tree pins + * `@corbits/agent-runtime` at. The tree is materialized inside this + * monorepo's own closure by the sidecar, so the workspace protocol is + * the pin: every run deploys the one reviewed version in-tree, never a + * separately published copy that could drift from the builder the hub + * validated the config against. + */ +export const AGENT_RUNTIME_PACKAGE_RANGE = "workspace:*"; diff --git a/packages/agent-runtime/src/source-tree.test.ts b/packages/agent-runtime/src/source-tree.test.ts new file mode 100644 index 000000000..6a13c179e --- /dev/null +++ b/packages/agent-runtime/src/source-tree.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, test } from "bun:test"; + +import type { AgentRuntimeConfig } from "./config"; +import { AGENT_RUNTIME_PACKAGE_NAME } from "./pin"; +import { + AGENT_RUNTIME_ENTRY_PATH, + renderAgentRuntimeSourceTree, +} from "./source-tree"; + +const config: AgentRuntimeConfig = { + workflowId: "wf_run_a", + agentId: "run_a", + triggerAddress: "run_a@bench.example", + systemPrompt: "You are helpful.", + inferencePreferences: [{ provider: "acme", model: "acme-1" }], + toolPackagePins: [], + credentialBindings: [], + mode: { kind: "step" }, +}; + +function render(overrides: Partial = {}) { + return renderAgentRuntimeSourceTree({ + packageName: "run-a-workflow", + runtimeVersion: "0.0.1", + config: { ...config, ...overrides }, + }); +} + +describe("renderAgentRuntimeSourceTree", () => { + test("declares the entry the sidecar evaluates", () => { + const pkg = JSON.parse(render()["package.json"] ?? ""); + + expect(pkg.interchange).toEqual({ workflow: AGENT_RUNTIME_ENTRY_PATH }); + expect(Object.keys(render())).toContain("workflow.js"); + }); + + test("pins the versioned runtime package as the tree's one dependency", () => { + const pkg = JSON.parse(render()["package.json"] ?? ""); + + expect(pkg.dependencies).toEqual({ [AGENT_RUNTIME_PACKAGE_NAME]: "0.0.1" }); + }); + + test("renders the config into the entry module's own bytes", () => { + const entry = render()["workflow.js"] ?? ""; + + expect(entry).toContain(`from "${AGENT_RUNTIME_PACKAGE_NAME}"`); + expect(entry).toContain("buildAgentRuntimeWorkflow("); + expect(entry).toContain('"run_a@bench.example"'); + expect(entry).toContain('"You are helpful."'); + }); + + test("a differing per-run field produces differing bytes — the hash barrier's whole premise", () => { + const a = render()["workflow.js"]; + const b = render({ systemPrompt: "You are terse." })["workflow.js"]; + + expect(a).not.toBe(b); + }); + + test("the same config renders byte-identically, so probe and run agree", () => { + expect(render()).toEqual(render()); + }); + + test("the rendered entry's config round-trips back to the config it was given", () => { + const entry = render()["workflow.js"] ?? ""; + const literal = entry.slice( + entry.indexOf("buildAgentRuntimeWorkflow(") + + "buildAgentRuntimeWorkflow(".length, + entry.lastIndexOf(");"), + ); + + expect(JSON.parse(literal)).toEqual(config); + }); + + test("renders the section mode's turn timeout into the bytes too", () => { + const entry = + render({ mode: { kind: "section", turnTimeoutMs: 45_000 } })[ + "workflow.js" + ] ?? ""; + + expect(entry).toContain('"kind": "section"'); + expect(entry).toContain('"turnTimeoutMs": 45000'); + }); + + test("refuses to render a config the run child would reject", () => { + expect(() => render({ inferencePreferences: [] })).toThrow( + /invalid agent-runtime config/, + ); + }); +}); diff --git a/packages/agent-runtime/src/source-tree.ts b/packages/agent-runtime/src/source-tree.ts new file mode 100644 index 000000000..490e895f2 --- /dev/null +++ b/packages/agent-runtime/src/source-tree.ts @@ -0,0 +1,66 @@ +// Renders the source tree a code-sourced deploy actually deploys. +// +// Under the workflow.json retirement a deployment's definition is +// whatever its own pinned code closure evaluates to, and the approved +// wire hash covers every field that differs per run. So the per-run +// config cannot ride beside the bytes — it has to BE the bytes. +// +// The tree this renders is deliberately thin: a `package.json` and a +// four-line entry module that pins `@corbits/agent-runtime` and calls +// `buildAgentRuntimeWorkflow` with the run's config as a literal. All +// the behaviour stays in this one versioned package, reviewed and +// upgraded in one place; what varies per run is a JSON literal. A host +// commits the tree into a `workflow`-kind asset and deploys it with +// `source.kind: "asset"`, `package.format: "source"`, `commitSha` — the +// only source variant whose pin is cheap enough to mint per run (the +// registry and tarball variants would each need a publish). +import { parseAgentRuntimeConfig, type AgentRuntimeConfig } from "./config"; +import { AGENT_RUNTIME_PACKAGE_NAME } from "./pin"; + +/** The entry path the rendered `package.json` declares and the sidecar evaluates. */ +export const AGENT_RUNTIME_ENTRY_PATH = "./workflow.js"; + +export interface RenderAgentRuntimeSourceTreeInput { + /** + * The rendered package's own name. It never leaves the asset, so it + * only has to be a valid package name and stable for a given run. + */ + readonly packageName: string; + /** The `@corbits/agent-runtime` range the rendered package depends on. */ + readonly runtimeVersion: string; + /** The run's deploy-time config, rendered into the entry module. */ + readonly config: AgentRuntimeConfig; +} + +/** File contents keyed by path relative to the tree root. */ +export type AgentRuntimeSourceTree = Readonly>; + +/** + * Render the per-run workflow package. The config is validated before + * it is written, so a config the run child would reject fails at the + * deploying call site instead of inside the approval probe. + */ +export function renderAgentRuntimeSourceTree( + input: RenderAgentRuntimeSourceTreeInput, +): AgentRuntimeSourceTree { + const config = parseAgentRuntimeConfig(input.config); + const packageJson = { + name: input.packageName, + version: "0.0.0", + private: true, + type: "module", + interchange: { workflow: AGENT_RUNTIME_ENTRY_PATH }, + dependencies: { [AGENT_RUNTIME_PACKAGE_NAME]: input.runtimeVersion }, + }; + const entry = [ + `import { buildAgentRuntimeWorkflow } from ${JSON.stringify(AGENT_RUNTIME_PACKAGE_NAME)};`, + "", + `export default buildAgentRuntimeWorkflow(${JSON.stringify(config, null, 2)});`, + "", + ].join("\n"); + + return { + "package.json": `${JSON.stringify(packageJson, null, 2)}\n`, + "workflow.js": entry, + }; +} diff --git a/packages/agent-runtime/tsconfig.json b/packages/agent-runtime/tsconfig.json new file mode 100644 index 000000000..e956ddd88 --- /dev/null +++ b/packages/agent-runtime/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "types": ["bun"] + }, + "include": ["src", "test"] +} diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index a2ad83bb5..820e43219 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -22,12 +22,12 @@ import { sendFoldedMail, wakeFoldedRun, FoldedBodySchema, + type FoldedRunMode, type FoldedRunsDeps, type SendFoldedMailParams, type SourcesOverride, } from "@corbits/folded-runs"; import type { FoldedBody } from "@intx/workflow-deploy"; -import type { Selector } from "@intx/workflow"; import type { DB } from "@intx/db"; import { agentSession, @@ -50,7 +50,6 @@ import { ensureWorkflowDefinitionForAsset } from "@intx/hub-sessions"; import type { AssetService, EventCollectorRegistry, - SessionService, SidecarRouter, } from "@intx/hub-sessions"; import type { InferencePreference } from "@intx/agent"; @@ -71,11 +70,9 @@ import { export type CreateHubChatPlatformDeps = { db: DB["db"]; - sessionService: SessionService; + sessionService: FoldedRunsDeps["sessionService"]; assetService: AssetService; sidecarRouter: SidecarRouter; - /** See `FoldedRunsDeps.hubPublicKey`. */ - hubPublicKey: string; /** See `FoldedRunsDeps.toolGrantsForPins`. */ toolGrantsForPins: FoldedRunsDeps["toolGrantsForPins"]; /** See `FoldedRunsDeps.mcpCredentialBindingsFor`. */ @@ -202,8 +199,9 @@ function noopSourcesOverride( * value is never read by anything — the anchor's whole job is holding * the mailbox, not processing input. */ -const WORKBENCH_HOST_STEP_INPUT: Selector = { - literal: "workbench-host anchor turn", +const WORKBENCH_HOST_MODE: FoldedRunMode = { + kind: "step", + literalInput: "workbench-host anchor turn", }; /** @@ -246,7 +244,6 @@ export function createHubChatPlatform( assetService: deps.assetService, sidecarRouter: deps.sidecarRouter, eventCollectors: deps.eventCollectors, - hubPublicKey: deps.hubPublicKey, toolGrantsForPins: deps.toolGrantsForPins, ...(deps.credentialCipher !== undefined ? { credentialCipher: deps.credentialCipher } @@ -372,7 +369,7 @@ export function createHubChatPlatform( deps.noopInferenceBaseUrl, parsedFoldedBody, ), - stepInput: WORKBENCH_HOST_STEP_INPUT, + mode: WORKBENCH_HOST_MODE, } : { ...wakeParams, diff --git a/packages/chat/test/platform-adapter-activity.test.ts b/packages/chat/test/platform-adapter-activity.test.ts index 47527bace..b7e86432a 100644 --- a/packages/chat/test/platform-adapter-activity.test.ts +++ b/packages/chat/test/platform-adapter-activity.test.ts @@ -19,10 +19,10 @@ import { agentSession, sessionMail, workflowRun } from "@intx/db/schema"; import type { AssetService, EventCollectorRegistry, - SessionService, SidecarRouter, } from "@intx/hub-sessions"; import { createHubChatPlatform } from "../src/platform-adapter"; +import type { CreateHubChatPlatformDeps } from "../src/platform-adapter"; type Row = Record; @@ -73,11 +73,11 @@ function fakeDb(plan: { function buildPlatform(plan: Parameters[0]) { return createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: fakeDb(plan) as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", - sessionService: {} as unknown as SessionService, + sessionService: + {} as unknown as CreateHubChatPlatformDeps["sessionService"], assetService: {} as unknown as AssetService, sidecarRouter: {} as unknown as SidecarRouter, eventCollectors: {} as unknown as EventCollectorRegistry, diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index 255a30134..d67dc99a3 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -23,6 +23,7 @@ // exercised without a real Postgres. import { describe, expect, mock, test } from "bun:test"; +import type { FoldedRunsDeps } from "@corbits/folded-runs"; import { agentSession, asset, @@ -36,11 +37,7 @@ import { workbenchLaunch } from "../src/schema"; import { foldedRun } from "@corbits/folded-runs"; import { IDLE_HIBERNATE_UNDEPLOY_REASON } from "@corbits/agent-lifecycle"; import { SessionLaunchError } from "@intx/hub-sessions"; -import type { - EventCollectorRegistry, - SessionService, - SidecarRouter, -} from "@intx/hub-sessions"; +import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions"; import type { DefinitionSourceResolution } from "@intx/hub-api"; import { buildWorkbenchHostWorkflow, @@ -305,6 +302,13 @@ function createFakeDb(opts: { select(..._cols: unknown[]) { return { from(table: unknown) { + if (table === workflowRun) { + // `deployAtHead` joins the run to its definition asset — the + // asset its per-run workflow source tree is committed into. + return { + innerJoin: () => selectChain([{ assetId: "ast_definition1" }]), + }; + } if (table === asset) return selectChain([opts.assetRow]); if (table === workbenchLaunch) { const insertedLaunch = inserted.findLast( @@ -409,30 +413,41 @@ function createFakeEventCollectors( }; } -function createFakeSessionService(): SessionService & { - deployInstanceAtHeadCalls: unknown[]; +type AdoptedDeployCall = { + anchorRunId: string; + agentAddress: string; +}; + +type FakeSessionService = FoldedRunsDeps["sessionService"] & { + adoptedDeployCalls: unknown[]; sendUserMessageCalls: unknown[]; -} { - const deployInstanceAtHeadCalls: unknown[] = []; +}; + +function createFakeSessionService(): FakeSessionService { + const adoptedDeployCalls: unknown[] = []; const sendUserMessageCalls: unknown[] = []; return { - deployInstanceAtHeadCalls, + adoptedDeployCalls, sendUserMessageCalls, async stageWorkflowStep() {}, async deployInstanceAtHead() { throw new Error( "deployInstanceAtHead must not be called: a folded run deploys " + - "an explicit unbounded single-step workflow via deploySingleStepAtHead", + "its own rendered workflow source package", ); }, - async deploySingleStepAtHead(params: unknown) { - deployInstanceAtHeadCalls.push(params); - return { publicKey: "test-public-key" }; + async deployAdoptedWorkflowFromSource(params: AdoptedDeployCall) { + adoptedDeployCalls.push(params); + return { + anchorRunId: params.anchorRunId, + deploymentAddress: params.agentAddress, + publicKey: "test-public-key", + }; }, async deployWorkflowDefinition() { throw new Error( "deployWorkflowDefinition must not be called: launchWorkbench " + - "launches a folded instance via deployInstanceAtHead", + "launches a folded run through the adopting code-sourced front", ); }, async sendUserMessage(params: unknown) { @@ -440,10 +455,7 @@ function createFakeSessionService(): SessionService & { return new TextEncoder().encode("raw-mime-bytes"); }, async endSession() {}, - } as unknown as SessionService & { - deployInstanceAtHeadCalls: unknown[]; - sendUserMessageCalls: unknown[]; - }; + } as unknown as FakeSessionService; } function createFakeAssetService(opts: { assetBlob?: Uint8Array } = {}) { @@ -593,7 +605,6 @@ describe("createHubChatPlatform", () => { const eventCollectors = createFakeEventCollectors(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], // Fake db, not a real drizzle instance. db: db as never, @@ -615,7 +626,7 @@ describe("createHubChatPlatform", () => { expect(launched.instanceId).toBe("ins_workbench1"); expect(eventCollectors.createCalls).toEqual([]); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(0); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); expect(resolveDefinitionSourcesCalls).toHaveLength(0); await platform.ensureAwake("ins_workbench1@ten1.workbench.test"); @@ -644,11 +655,10 @@ describe("createHubChatPlatform", () => { expect(resolveDefinitionSourcesCalls).toHaveLength(0); // The folded launch path, never the native workflow-deploy path. - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as { agentAddress: string; - agentId: string; - runId: string; + anchorRunId: string; config: { systemPrompt: string; sources: { @@ -664,8 +674,7 @@ describe("createHubChatPlatform", () => { }; }; expect(deployed.agentAddress).toBe("ins_workbench1@ten1.workbench.test"); - expect(deployed.agentId).toBe("ins_workbench1"); - expect(deployed.runId).toBe("ins_workbench1"); + expect(deployed.anchorRunId).toBe("ins_workbench1"); expect(deployed.config.systemPrompt.length).toBeGreaterThan(0); expect(deployed.config.sources).toEqual([ { @@ -749,7 +758,7 @@ describe("createHubChatPlatform", () => { }); const sessionService = createFakeSessionService(); const deployError = new Error("sidecar unreachable"); - sessionService.deploySingleStepAtHead = async () => { + sessionService.deployAdoptedWorkflowFromSource = async () => { throw deployError; }; const assetService = createFakeAssetService(); @@ -757,7 +766,6 @@ describe("createHubChatPlatform", () => { const eventCollectors = createFakeEventCollectors(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -816,7 +824,7 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", }); const sessionService = createFakeSessionService(); - sessionService.deploySingleStepAtHead = async () => { + sessionService.deployAdoptedWorkflowFromSource = async () => { throw new SessionLaunchError("start", new Error("ack timeout"), true); }; const assetService = createFakeAssetService(); @@ -824,7 +832,6 @@ describe("createHubChatPlatform", () => { const eventCollectors = createFakeEventCollectors(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -899,7 +906,6 @@ describe("createHubChatPlatform", () => { const sidecarRouter = createFakeSidecarRouter(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -981,7 +987,6 @@ describe("createHubChatPlatform", () => { const eventCollectors = createFakeEventCollectors(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1004,17 +1009,14 @@ describe("createHubChatPlatform", () => { { assetId: "asst_echo", path: "workflow.json" }, ]); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(0); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); expect(resolveDefinitionSourcesCalls).toHaveLength(0); await platform.ensureAwake(launched.address); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - agentAddress: string; - runId: string; - }; + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as AdoptedDeployCall; expect(deployed.agentAddress).toBe(launched.address); - expect(deployed.runId).toBe(launched.instanceId); + expect(deployed.anchorRunId).toBe(launched.instanceId); const runInsert = db.inserted.find((row) => row.table === workflowRun); expect(runInsert?.values).toMatchObject({ @@ -1091,7 +1093,6 @@ describe("createHubChatPlatform", () => { }; const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1134,7 +1135,6 @@ describe("createHubChatPlatform", () => { tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1166,7 +1166,6 @@ describe("createHubChatPlatform", () => { tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1212,7 +1211,6 @@ describe("createHubChatPlatform", () => { tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1283,7 +1281,6 @@ describe("createHubChatPlatform", () => { tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1339,7 +1336,6 @@ describe("createHubChatPlatform", () => { tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1400,7 +1396,6 @@ describe("createHubChatPlatform", () => { ], }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1436,7 +1431,6 @@ describe("createHubChatPlatform", () => { const sidecarRouter = createFakeSidecarRouter(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1477,7 +1471,6 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1545,7 +1538,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1672,7 +1664,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1692,13 +1683,11 @@ describe("createHubChatPlatform", () => { expect(sent.id).toBeTruthy(); // The redeploy happened... - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - agentAddress: string; - runId: string; - }; + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService + .adoptedDeployCalls[0] as AdoptedDeployCall; expect(deployed.agentAddress).toBe("ins_workbench1@ten1.workbench.test"); - expect(deployed.runId).toBe("ins_workbench1"); + expect(deployed.anchorRunId).toBe("ins_workbench1"); // ...before the send. expect(sessionService.sendUserMessageCalls).toHaveLength(1); }); @@ -1771,7 +1760,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1790,7 +1778,7 @@ describe("createHubChatPlatform", () => { }); expect(sent.id).toBeTruthy(); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(0); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); expect(sidecarRouter.sendAgentUndeployCalls).toHaveLength(0); expect(sessionService.sendUserMessageCalls).toHaveLength(1); }); @@ -1852,7 +1840,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -1871,8 +1858,8 @@ describe("createHubChatPlatform", () => { }); expect(resolveDefinitionSourcesCalls).toHaveLength(0); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as { config: { sources: { id: string; @@ -1940,7 +1927,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2007,7 +1993,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2076,7 +2061,6 @@ describe("createHubChatPlatform", () => { }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2122,7 +2106,6 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", }); createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2158,7 +2141,6 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", }); createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2193,7 +2175,6 @@ describe("createHubChatPlatform", () => { }); const sessionService = createFakeSessionService(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2207,7 +2188,7 @@ describe("createHubChatPlatform", () => { await platform.ensureAwake(address); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(0); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); }); test("redeploys a non-routable address when lifecycle is configured", async () => { @@ -2245,7 +2226,6 @@ describe("createHubChatPlatform", () => { const sessionService = createFakeSessionService(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2258,7 +2238,7 @@ describe("createHubChatPlatform", () => { await platform.ensureAwake(address); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); + expect(sessionService.adoptedDeployCalls).toHaveLength(1); }); test("redeploys a non-routable address when lifecycle is not configured", async () => { @@ -2296,7 +2276,6 @@ describe("createHubChatPlatform", () => { const sessionService = createFakeSessionService(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2308,7 +2287,7 @@ describe("createHubChatPlatform", () => { await platform.ensureAwake(address); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); + expect(sessionService.adoptedDeployCalls).toHaveLength(1); }); test("rejects for an address this adapter has no folded run for", async () => { @@ -2322,7 +2301,6 @@ describe("createHubChatPlatform", () => { definitionId: "wfd_workbench1", }); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2405,7 +2383,6 @@ describe("createHubChatPlatform", () => { test("recomputes and persists the folded body from the definition's current asset", async () => { const db = buildRefreshableDb(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2454,7 +2431,6 @@ describe("createHubChatPlatform", () => { }); const sessionService = createFakeSessionService(); const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db: db as never, noopInferenceBaseUrl: "https://hub.invalid/api/chat/noop-inference", @@ -2483,8 +2459,8 @@ describe("createHubChatPlatform", () => { content: { content: "hello" }, }); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as { config: { systemPrompt: string }; }; expect(deployed.config.systemPrompt).toBe( diff --git a/packages/chat/test/reaction-message-lookup.drizzle.test.ts b/packages/chat/test/reaction-message-lookup.drizzle.test.ts index c40a898c0..9b4fddb6a 100644 --- a/packages/chat/test/reaction-message-lookup.drizzle.test.ts +++ b/packages/chat/test/reaction-message-lookup.drizzle.test.ts @@ -21,7 +21,6 @@ import { schema } from "@intx/db"; import type { AssetService, EventCollectorRegistry, - SessionService, SidecarRouter, } from "@intx/hub-sessions"; import type { TenantEnv } from "@intx/hub-api"; @@ -30,6 +29,7 @@ import { dbTargetFromUrl } from "../../../scripts/db-setup"; import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; import { createChatRoutes } from "../src/routes"; import { createHubChatPlatform } from "../src/platform-adapter"; +import type { CreateHubChatPlatformDeps } from "../src/platform-adapter"; import { createInMemoryChatStore } from "../src/store"; import { createInMemoryWorkbenchTenancyStore } from "../src/workbench-tenancy"; import { createInMemoryReactionStore } from "../src/reactions"; @@ -127,10 +127,10 @@ describeIfDb("reaction toggle: message lookup past the first mail page", () => { targetMessageId = target55Back.id; const platform = createHubChatPlatform({ - hubPublicKey: "hub-key", toolGrantsForPins: () => [], db, - sessionService: {} as unknown as SessionService, + sessionService: + {} as unknown as CreateHubChatPlatformDeps["sessionService"], assetService: {} as unknown as AssetService, sidecarRouter: {} as unknown as SidecarRouter, eventCollectors: {} as unknown as EventCollectorRegistry, diff --git a/packages/folded-runs/package.json b/packages/folded-runs/package.json index b2cb890ac..e354ac0e3 100644 --- a/packages/folded-runs/package.json +++ b/packages/folded-runs/package.json @@ -14,6 +14,7 @@ }, "dependencies": { "@corbits/agent-lifecycle": "workspace:*", + "@corbits/agent-runtime": "workspace:*", "@intx/crypto": "workspace:*", "@intx/db": "workspace:*", "@intx/hub-api": "workspace:*", diff --git a/packages/folded-runs/src/index.ts b/packages/folded-runs/src/index.ts index 7c95893b5..8f3b7d4d9 100644 --- a/packages/folded-runs/src/index.ts +++ b/packages/folded-runs/src/index.ts @@ -25,11 +25,13 @@ export { } from "./runs"; export { deployAtHead, + foldedRunSourceRef, launchFoldedRun, mintFoldedRun, parseSourcesOverride, SourcesOverride, InferenceResolutionError, + type FoldedRunMode, type LaunchFoldedRunParams, type MintFoldedRunParams, type LaunchedFoldedRun, diff --git a/packages/folded-runs/src/launch.ts b/packages/folded-runs/src/launch.ts index 0a3b5c458..44b0e0ea6 100644 --- a/packages/folded-runs/src/launch.ts +++ b/packages/folded-runs/src/launch.ts @@ -19,6 +19,7 @@ import type { CredentialBinding } from "@intx/types"; import { agentSession, principal as principalTable, + workflowDefinition, workflowRun, } from "@intx/db/schema"; import { foldedRun } from "./schema"; @@ -27,11 +28,13 @@ import { resolveDefinitionSources } from "@intx/hub-api"; import { generateId } from "@intx/hub-common"; import { InferenceSource } from "@intx/types/runtime"; import type { WireGrantRule } from "@intx/types/grant-wire"; +import type { FoldedBody } from "@intx/workflow-deploy"; import { - buildSingleStepAgentDefinition, - type FoldedBody, -} from "@intx/workflow-deploy"; -import { defineWorkflow, step, type Selector } from "@intx/workflow"; + AGENT_RUNTIME_ENTRY_PATH, + AGENT_RUNTIME_PACKAGE_RANGE, + renderAgentRuntimeSourceTree, + type AgentRuntimeConfig, +} from "@corbits/agent-runtime"; import type { FoldedRunsDeps } from "./types"; /** @@ -86,16 +89,85 @@ export function parseSourcesOverride( return parsed; } +/** + * The `mode` a folded run's deployed definition takes. `step` is the + * folded conversational shape every launcher gets today; `section` is + * CL-6329's per-turn `onTrigger` shape, selected by the caller alone — + * `deployAtHead` never branches on which one it is deploying, because + * the mode travels inside the rendered config. + */ +export type FoldedRunMode = AgentRuntimeConfig["mode"]; + +/** + * The ref a folded run's per-run workflow source tree is committed to + * inside its definition asset. Per-run rather than the asset's default + * branch because one definition asset backs many runs — a chat's + * workbench host, an invited agent's every launch — and each run's tree + * carries its OWN config in its bytes. The deploy pins the resulting + * `commitSha`, so the ref is bookkeeping, never the pin. + */ +export function foldedRunSourceRef(instanceId: string): string { + return `refs/heads/runs/${instanceId}`; +} + +/** The rendered per-run package's own name; it never leaves the asset. */ +function foldedRunPackageName(instanceId: string): string { + return `folded-run-${instanceId}`; +} + +/** + * The mail domain a run's deployment addresses live under. The deploy + * front re-derives `@` and refuses a pair + * that does not name the same run, so this must be the trigger + * address's own domain and nothing else. + */ +function domainOfAddress(address: string): string { + const domain = address.split("@")[1]; + if (domain === undefined || domain.length === 0) { + throw new Error(`folded run address "${address}" carries no mail domain`); + } + return domain; +} + +/** + * The `workflow`-kind asset backing this run's definition — the asset + * the launching host already minted for it (`@corbits/chat`'s + * `launchWorkbench`, `@corbits/agent-directory`'s create route). The + * per-run source tree is committed INTO that asset on its own ref + * rather than into a second asset minted per deploy. + */ +async function resolveRunDefinitionAssetId( + db: FoldedRunsDeps["db"], + instanceId: string, +): Promise { + const row = await db + .select({ assetId: workflowDefinition.assetId }) + .from(workflowRun) + .innerJoin( + workflowDefinition, + eq(workflowDefinition.id, workflowRun.definitionId), + ) + .where(eq(workflowRun.id, instanceId)) + .limit(1) + .then((rows) => rows[0]); + if (row === undefined || row.assetId === null) { + throw new Error( + `folded run ${instanceId} has no workflow-kind definition asset to commit its per-run source tree into`, + ); + } + return row.assetId; +} + /** * The deploy-only step shared by a fresh launch (`launchFoldedRun`) * and a wake (re-deploying an instance the sidecar no longer has * resident): resolve inference sources against the tenant catalog, - * (re)open the event collector, and call `deployInstanceAtHead`. - * Callers that just wrote new principal/session/run rows - * (`launchFoldedRun`) still own their own failure-path rollback of - * those rows — this function only throws. + * (re)open the event collector, render the run's own workflow source + * package, commit it, and deploy it onto the run's pre-minted anchor + * through the adopting code-sourced front. Callers that just wrote new + * principal/session/run rows (`launchFoldedRun`) still own their own + * failure-path rollback of those rows — this function only throws. */ -const FOLDED_STEP_ID = "default"; export async function deployAtHead( deps: Pick< @@ -105,7 +177,7 @@ export async function deployAtHead( | "sidecarRouter" | "eventCollectors" | "credentialCipher" - | "hubPublicKey" + | "assetService" | "toolGrantsForPins" | "mcpCredentialBindingsFor" >, @@ -138,23 +210,13 @@ export async function deployAtHead( */ fallbackModel?: string; /** - * Overrides the step's default input selector (`{ from: - * "trigger.payload" }`, `defineWorkflow`'s standard first-step - * default). The default reads the triggering mail's bare `content` - * verbatim and feeds it straight into `agent.send`, which throws on - * an empty string — and `content` is legitimately empty for - * attachments-only mail (an event-only send, e.g. - * `workbench.agent-joined`; see `@corbits/chat`'s `encodeParts`). - * A folded run that genuinely ignores its input (the workbench host: - * its system prompt forbids ever acting on what it receives) should - * pin a `{ literal: ... }` selector here instead of reading - * `trigger.payload`, so an attachments-only mail landing in its - * inbox — its very first message, in the common case — cannot crash - * the run before it ever opens (CL-6164). Absent, behavior is - * unchanged: the step reads the real trigger payload, as every - * inference-driven agent must. + * The shape the run's deployed definition takes. Defaults to the + * folded conversational step every launcher uses today; CL-6329's + * per-turn swap passes `{ kind: "section", turnTimeoutMs }` and + * nothing else about this call changes, because the mode is config + * data rendered into the deployed bytes rather than a branch here. */ - stepInput?: Selector; + mode?: FoldedRunMode; }, ): Promise { const sourcesOverride = parseSourcesOverride(params.sources); @@ -227,9 +289,11 @@ export async function deployAtHead( ...mcpBindings, ]; - let credentials: Parameters< - FoldedRunsDeps["sessionService"]["deploySingleStepAtHead"] - >[0]["credentials"]; + // The deploy front resolves the credential MATERIAL itself from the + // deployed definition's own bindings under `credentialCipher`. What it + // does not derive is the `credential:` use grants this run's principal + // needs in its own `grants.json`, so the delivery is still walked here + // — for `bindingGrants` alone. if (credentialBindings.length > 0) { if (deps.credentialCipher === undefined) { throw new Error( @@ -249,7 +313,6 @@ export async function deployAtHead( `${params.launchLabel}: credential binding resolution failed: ${delivery.reason.message}`, ); } - credentials = delivery.delivery; for (const bindingGrant of delivery.bindingGrants) { grants.push({ id: generateId("grant"), @@ -277,65 +340,74 @@ export async function deployAtHead( sources: resolution.sources, defaultSource: resolution.defaultSource, }; - const deployContent = { systemPrompt: params.foldedBody.systemPrompt }; - // A folded run is a conversation: its one step must service every - // inbound mail as another turn, never complete after the first. A wrap - // with the platform's default trigger budget of 1 (batch) is exactly what - // made every chat go silent after its first real reply — so the folded - // launch builds the single-step agent itself, with the budget declared, - // and deploys it through the same head deploy. The launch pins its tools - // as packages rather than factories, so the step agent carries none. - const foldedSteps = { - [FOLDED_STEP_ID]: step({ - agent: buildSingleStepAgentDefinition({ - id: config.agentId, - systemPrompt: deployContent.systemPrompt, - inferencePreferences: config.sources.map((source) => ({ - provider: source.provider, - model: source.model, - })), - toolFactories: [], - }), - triggers: "unbounded", - ...(params.stepInput !== undefined ? { input: params.stepInput } : {}), - }), - }; - // The workflow-host's per-step credential snapshot + // Everything that differs per run, in one literal. The deployed + // definition is whatever this run's own pinned bytes evaluate to, and + // the approved wire hash covers every field below — the trigger + // address, the system prompt, the (provider, model) pairs, the tool + // package pins, the credential bindings — so the config cannot ride + // beside the bytes as an env var or a staged file. It IS the bytes: + // `renderAgentRuntimeSourceTree` writes it into the entry module the + // approval probe and the run child each evaluate independently. + // + // The definition's own `credentialBindings` are what the workflow + // host's per-step credential snapshot // (`vendor/intx/workflow-host/src/supervisor/credentials.ts`) derives - // its bindings from the deployed *definition*'s own - // `credentialBindings`, not from `buildCredentialDelivery`'s output — - // that delivery only seeds the credential material itself. Mirror - // `buildAgentDefinitionWorkflow`'s same conditional shape so a folded - // run's synthesized definition carries the same combined bindings - // (the definition's own plus the pinned-package MCP bindings folded in - // above) the delivered material was resolved against; without this the - // sidecar's `consumerBindings` finds nothing for `mcp:` and every - // resolve() fails "not connected" even though the material was - // delivered. - const definition = - credentialBindings.length > 0 - ? defineWorkflow({ - id: `wf_${params.instanceId}`, - trigger: { type: "mail", to: params.triggerAddress }, - credentialBindings, - steps: foldedSteps, - }) - : defineWorkflow({ - id: `wf_${params.instanceId}`, - trigger: { type: "mail", to: params.triggerAddress }, - steps: foldedSteps, - }); - await deps.sessionService.deploySingleStepAtHead({ - agentAddress: params.triggerAddress, + // its consumer bindings from, which is why the pinned-package MCP + // bindings folded in above have to reach the rendered config and not + // just the delivery: without them `env.credentials.resolve("mcp:")` + // fails "not connected" even when the material was delivered. + const runtimeConfig: AgentRuntimeConfig = { + workflowId: `wf_${params.instanceId}`, agentId: params.instanceId, - runId: params.instanceId, + triggerAddress: params.triggerAddress, + systemPrompt: params.foldedBody.systemPrompt, + inferencePreferences: resolution.sources.map((source) => ({ + provider: source.provider, + model: source.model, + })), + toolPackagePins: [...params.foldedBody.toolPackagePins], + credentialBindings, + mode: params.mode ?? { kind: "step" }, + }; + const definitionAssetId = await resolveRunDefinitionAssetId( + deps.db, + params.instanceId, + ); + const { commitSha } = await deps.assetService.populateAsset({ + assetId: definitionAssetId, + ref: foldedRunSourceRef(params.instanceId), + principal: { kind: "hub" }, + tree: { + files: renderAgentRuntimeSourceTree({ + packageName: foldedRunPackageName(params.instanceId), + runtimeVersion: AGENT_RUNTIME_PACKAGE_RANGE, + config: runtimeConfig, + }), + message: `Deploy folded run ${params.instanceId}`, + }, + }); + + // The adopting front is the only code-sourced deploy a folded run can + // use: its anchor `workflow_run` row was minted before this call + // (`mintFoldedRun`), so the inserting front would collide on the + // primary key, and the prepared front hard-requires an exclusive + // allocation this run does not have. + await deps.sessionService.deployAdoptedWorkflowFromSource({ + tenantId: params.tenantId, + anchorRunId: params.instanceId, + deploymentDomain: domainOfAddress(params.triggerAddress), + agentAddress: params.triggerAddress, + source: { + kind: "asset", + assetId: definitionAssetId, + package: { format: "source", commitSha }, + }, + entry: AGENT_RUNTIME_ENTRY_PATH, + definitionAssetId, config, - deployContent, - definition, - sources: { [FOLDED_STEP_ID]: resolution.sources }, - hubPublicKey: deps.hubPublicKey, - toolPackagePins: params.foldedBody.toolPackagePins, - ...(credentials !== undefined ? { credentials } : {}), + ...(deps.credentialCipher !== undefined + ? { credentialCipher: deps.credentialCipher } + : {}), }); // Produce the run's `run.grants` frame, the same contract upstream's hub @@ -379,7 +451,7 @@ export type LaunchFoldedRunParams = { /** See `deployAtHead`'s own doc on the same field. */ fallbackModel?: string; /** See `deployAtHead`'s own doc on the same field. */ - stepInput?: Selector; + mode?: FoldedRunMode; /** * Invoked inside the same launch transaction, immediately after the * principal/session/run rows are written, so a caller-owned table @@ -548,9 +620,7 @@ export async function launchFoldedRun( ...(params.fallbackModel !== undefined ? { fallbackModel: params.fallbackModel } : {}), - ...(params.stepInput !== undefined - ? { stepInput: params.stepInput } - : {}), + ...(params.mode !== undefined ? { mode: params.mode } : {}), }); } catch (err) { // Mirrors the reference route's failure-path cleanup: a deploy diff --git a/packages/folded-runs/src/types.ts b/packages/folded-runs/src/types.ts index 861a01000..1622f6371 100644 --- a/packages/folded-runs/src/types.ts +++ b/packages/folded-runs/src/types.ts @@ -11,6 +11,7 @@ import type { } from "@intx/types"; import type { ToolPackagePin } from "@intx/types/tool-packages"; import type { + AdoptingWorkflowDeployer, AssetService, EventCollectorRegistry, SessionService, @@ -81,7 +82,13 @@ export type McpCredentialBindingsFor = ( export type FoldedRunsDeps = { db: DB["db"]; - sessionService: SessionService; + /** + * The session service, narrowed to include the adopting code-sourced + * deploy front `deployAtHead` uses: a folded run's anchor row is + * minted before any deployment attaches to it, which is the one + * combination the inserting and prepared fronts cannot serve. + */ + sessionService: SessionService & AdoptingWorkflowDeployer; assetService: AssetService; sidecarRouter: SidecarRouter; eventCollectors: EventCollectorRegistry; @@ -97,13 +104,6 @@ export type FoldedRunsDeps = { * ciphertext to the provider as its API key. */ credentialCipher?: CredentialCipher; - /** - * The hub's hex-encoded Ed25519 signing public key — the same value the - * sidecar router is created with. `deployAtHead` deploys a folded run - * as an explicit single-step workflow (so it can declare the step's - * `triggers: "unbounded"` budget) and that deploy carries the hub key. - */ - hubPublicKey: string; /** See `ToolGrantsForPins`'s own doc. */ toolGrantsForPins: ToolGrantsForPins; /** diff --git a/packages/folded-runs/src/wake.ts b/packages/folded-runs/src/wake.ts index 61151702c..12813ddce 100644 --- a/packages/folded-runs/src/wake.ts +++ b/packages/folded-runs/src/wake.ts @@ -8,11 +8,14 @@ // table of its own to read. import { eq } from "drizzle-orm"; import { sessionAsset } from "@intx/db/schema"; -import { deployAtHead, type SourcesOverride } from "./launch"; +import { + deployAtHead, + type FoldedRunMode, + type SourcesOverride, +} from "./launch"; import { resolveFoldedRunSessionId } from "./runs"; import type { FoldedRunsDeps } from "./types"; import type { FoldedBody } from "@intx/workflow-deploy"; -import type { Selector } from "@intx/workflow"; export type WakeFoldedRunParams = { tenantId: string; @@ -36,7 +39,7 @@ export type WakeFoldedRunParams = { * `trigger.payload` selector would silently restore the CL-6164 crash * on the very next mail this occurrence receives. */ - stepInput?: Selector; + mode?: FoldedRunMode; /** * See `deployAtHead`'s own doc on the same field. A definition that * declares no model of its own resolves a catalog default at every @@ -84,9 +87,7 @@ export async function wakeFoldedRun( await deployAtHead(deps, { ...deployAtHeadParams, ...(params.sources !== undefined ? { sources: params.sources } : {}), - ...(params.stepInput !== undefined - ? { stepInput: params.stepInput } - : {}), + ...(params.mode !== undefined ? { mode: params.mode } : {}), ...(params.fallbackModel !== undefined ? { fallbackModel: params.fallbackModel } : {}), diff --git a/packages/folded-runs/test/launch.test.ts b/packages/folded-runs/test/launch.test.ts index d0e651c33..929259c15 100644 --- a/packages/folded-runs/test/launch.test.ts +++ b/packages/folded-runs/test/launch.test.ts @@ -17,12 +17,9 @@ import { describe, expect, mock, test } from "bun:test"; import { agentSession, principal, workflowRun } from "@intx/db/schema"; import { foldedRun } from "../src/schema"; import { SessionLaunchError } from "@intx/hub-sessions"; -import type { - EventCollectorRegistry, - SessionService, - SidecarRouter, -} from "@intx/hub-sessions"; +import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions"; import type { DefinitionSourceResolution } from "@intx/hub-api"; +import type { FoldedRunsDeps } from "../src/types"; import type { FoldedBody } from "@intx/workflow-deploy"; const actualHubApi = await import("@intx/hub-api"); @@ -84,7 +81,7 @@ type InsertChain = { values(values: unknown): Promise; }; -function createFakeDb() { +function createFakeDb(assetId: string | null = "ast_definition1") { const inserted: { table: unknown; values: unknown }[] = []; const updated: { table: unknown; values: unknown }[] = []; const deleted: { table: unknown }[] = []; @@ -98,6 +95,19 @@ function createFakeDb() { } return { + // The one read `deployAtHead` does: the run's definition asset, the + // asset its per-run source tree is committed into. + select() { + return { + from: () => ({ + innerJoin: () => ({ + where: () => ({ + limit: async () => (assetId === null ? [] : [{ assetId }]), + }), + }), + }), + }; + }, insert(table: unknown) { return insertOn(table); }, @@ -165,34 +175,95 @@ function createFakeEventCollectors(): EventCollectorRegistry & { }; } -function createFakeSessionService(): SessionService & { - deployInstanceAtHeadCalls: unknown[]; -} { - const deployInstanceAtHeadCalls: unknown[] = []; +type FakeSessionService = FoldedRunsDeps["sessionService"] & { + adoptedDeployCalls: AdoptedDeployCall[]; +}; + +type AdoptedDeployCall = { + tenantId: string; + anchorRunId: string; + deploymentDomain: string; + agentAddress: string; + entry: string; + definitionAssetId: string; + source: { + kind: string; + assetId: string; + package: { format: string; commitSha: string }; + }; + config: { + sources: unknown[]; + defaultSource: string; + tenantId: string; + principalId: string; + grants: Record[]; + }; + credentialCipher?: unknown; +}; + +function createFakeSessionService(): FakeSessionService { + const adoptedDeployCalls: AdoptedDeployCall[] = []; return { - deployInstanceAtHeadCalls, + adoptedDeployCalls, async stageWorkflowStep() {}, async deployInstanceAtHead() { throw new Error( "deployInstanceAtHead must not be called: a folded run deploys " + - "an explicit unbounded single-step workflow via deploySingleStepAtHead", + "its own rendered workflow source package", + ); + }, + async deployWorkflowFromSource() { + throw new Error( + "deployWorkflowFromSource must not be called: it INSERTs an anchor " + + "row a folded run already owns", ); }, - async deploySingleStepAtHead(params: unknown) { - deployInstanceAtHeadCalls.push(params); - return { publicKey: "test-public-key" }; + async deployAdoptedWorkflowFromSource(params: AdoptedDeployCall) { + adoptedDeployCalls.push(params); + return { + anchorRunId: params.anchorRunId, + deploymentAddress: params.agentAddress, + publicKey: "test-public-key", + }; }, async deployWorkflowDefinition() { throw new Error( "deployWorkflowDefinition must not be called: launchFoldedRun " + - "launches a folded instance via deployInstanceAtHead", + "launches a folded run through the adopting code-sourced front", ); }, async sendUserMessage() { return new TextEncoder().encode("raw-mime-bytes"); }, async endSession() {}, - } as unknown as SessionService & { deployInstanceAtHeadCalls: unknown[] }; + } as unknown as FakeSessionService; +} + +type PopulateAssetCall = { + assetId: string; + ref: string; + tree: { files: Record; message: string }; +}; + +function createFakeAssetService(): FoldedRunsDeps["assetService"] & { + populateAssetCalls: PopulateAssetCall[]; +} { + const populateAssetCalls: PopulateAssetCall[] = []; + return { + populateAssetCalls, + async createAsset() { + throw new Error( + "createAsset must not be called: a folded run commits its per-run " + + "tree into the definition asset its host already minted", + ); + }, + async populateAsset(params: PopulateAssetCall) { + populateAssetCalls.push(params); + return { commitSha: "commit-sha-1" }; + }, + } as unknown as FoldedRunsDeps["assetService"] & { + populateAssetCalls: PopulateAssetCall[]; + }; } type RunGrantsCall = { @@ -224,6 +295,28 @@ function createFakeSidecarRouter(routable = true): SidecarRouter & { } as unknown as SidecarRouter & { runGrantsCalls: RunGrantsCall[] }; } +/** + * The `AgentRuntimeConfig` literal a rendered entry module carries. The + * config IS the deployed bytes under the workflow.json retirement, so a + * test that wants to know what was deployed reads it back out of them. + */ +function onlyCall(calls: readonly T[]): T { + const [call] = calls; + if (call === undefined) { + throw new Error("expected exactly one recorded call"); + } + return call; +} + +function entryConfigJSON(entry: string): string { + const open = entry.indexOf("buildAgentRuntimeWorkflow("); + const close = entry.lastIndexOf(");"); + if (open === -1 || close === -1) { + throw new Error(`rendered entry module has no config literal: ${entry}`); + } + return entry.slice(open + "buildAgentRuntimeWorkflow(".length, close); +} + const FOLDED_BODY: FoldedBody = { systemPrompt: "you are a workbench host", toolPackagePins: [], @@ -266,7 +359,7 @@ describe("mintFoldedRun", () => { // The whole point of a mint: an addressable run with no sidecar // traffic and no collector — the first mail wakes it instead. - expect(sessionService.deployInstanceAtHeadCalls).toEqual([]); + expect(sessionService.adoptedDeployCalls).toEqual([]); expect(eventCollectors.createCalls).toEqual([]); }); }); @@ -297,9 +390,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, }, @@ -339,24 +431,17 @@ describe("launchFoldedRun", () => { fallbackModel: "claude-sonnet-5", }); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - // The folded step must be unbounded: a conversation services every - // mail as another turn; the platform default (1) ends the run after - // the first reply and every later message is rejected as terminal. - const deployedDefinition = sessionService.deployInstanceAtHeadCalls[0] as { - definition: { steps: Record }; - hubPublicKey: string; - }; - expect( - Object.values(deployedDefinition.definition.steps)[0]?.triggers, - ).toBe("unbounded"); - expect(deployedDefinition.hubPublicKey).toBe("hub-key"); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - agentAddress: string; - agentId: string; - instanceId: string; - config: { sources: unknown[]; defaultSource: string; tenantId: string }; - }; + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + // The deploy adopts the anchor row `mintFoldedRun` already wrote, + // and pins the commit the run's own source tree was committed at. + const deployed = onlyCall(sessionService.adoptedDeployCalls); + expect(deployed.anchorRunId).toBe("ins_workbench1"); + expect(deployed.deploymentDomain).toBe("ten1.workbench.test"); + expect(deployed.source).toEqual({ + kind: "asset", + assetId: "ast_definition1", + package: { format: "source", commitSha: "commit-sha-1" }, + }); expect(deployed.agentAddress).toBe("ins_workbench1@ten1.workbench.test"); expect(deployed.config.defaultSource).toBe("off_1"); expect(deployed.config.tenantId).toBe("ten_1"); @@ -406,7 +491,9 @@ describe("launchFoldedRun", () => { // attachments-only mail. A caller that knows its run never reads its // input (the workbench host) must be able to pin a literal instead, so // an attachments-only first mail cannot crash the run before it opens. - test("stepInput overrides the step's default trigger.payload selector", async () => { + // The literal now travels in the rendered config, so it must show up + // in the committed bytes, not in a caller-supplied definition. + test("the caller's literal input reaches the deployed bytes", async () => { resolveDefinitionSourcesCalls.length = 0; resolveDefinitionSourcesResult = { ok: true, @@ -423,14 +510,14 @@ describe("launchFoldedRun", () => { }; const sessionService = createFakeSessionService(); + const assetService = createFakeAssetService(); await launchFoldedRun( { db: createFakeDb() as never, sessionService, - assetService: {} as never, + assetService, sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: createFakeEventCollectors(), }, @@ -441,16 +528,13 @@ describe("launchFoldedRun", () => { definitionId: "wfd_workbench1", foldedBody: FOLDED_BODY, launchLabel: "the workbench host", - stepInput: { literal: "workbench-host anchor turn" }, + mode: { kind: "step", literalInput: "workbench-host anchor turn" }, }, ); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - definition: { steps: Record }; - }; - expect(Object.values(deployed.definition.steps)[0]?.input).toEqual({ - literal: "workbench-host anchor turn", - }); + const entry = + assetService.populateAssetCalls[0]?.tree.files["workflow.js"] ?? ""; + expect(entry).toContain('"literalInput": "workbench-host anchor turn"'); }); // CL-6149: a pinned tool package's calls failed every call with @@ -490,9 +574,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: (pins) => { toolGrantsForPinsCalls.push(pins); return [ @@ -522,9 +605,7 @@ describe("launchFoldedRun", () => { expect(toolGrantsForPinsCalls).toEqual([pinnedFoldedBody.toolPackagePins]); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - config: { grants: unknown[]; principalId: string }; - }; + const deployed = onlyCall(sessionService.adoptedDeployCalls); expect(deployed.config.principalId).toBe(result.instancePrincipalId); expect(deployed.config.grants).toEqual([ { @@ -585,9 +666,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, credentialCipher, @@ -627,7 +707,7 @@ describe("launchFoldedRun", () => { const db = createFakeDb(); const sessionService = createFakeSessionService(); const deployError = new Error("sidecar unreachable"); - sessionService.deploySingleStepAtHead = async () => { + sessionService.deployAdoptedWorkflowFromSource = async () => { throw deployError; }; const eventCollectors = createFakeEventCollectors(); @@ -637,9 +717,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, }, @@ -687,7 +766,7 @@ describe("launchFoldedRun", () => { const db = createFakeDb(); const sessionService = createFakeSessionService(); - sessionService.deploySingleStepAtHead = async () => { + sessionService.deployAdoptedWorkflowFromSource = async () => { throw new SessionLaunchError("start", new Error("ack timeout"), true); }; const eventCollectors = createFakeEventCollectors(); @@ -697,9 +776,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, }, @@ -736,9 +814,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService: createFakeSessionService(), - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: createFakeEventCollectors(), }, @@ -795,9 +872,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, }, @@ -814,8 +890,8 @@ describe("launchFoldedRun", () => { expect(result.sessionId).toBeTruthy(); expect(resolveDefinitionSourcesCalls).toHaveLength(0); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { + expect(sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = sessionService.adoptedDeployCalls[0] as { config: { sources: unknown[]; defaultSource: string }; }; expect(deployed.config.sources).toEqual(override.sources); @@ -832,9 +908,8 @@ describe("launchFoldedRun", () => { { db: db as never, sessionService, - assetService: {} as never, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors, }, @@ -854,7 +929,7 @@ describe("launchFoldedRun", () => { ), ).rejects.toThrow(/invalid inference sources override/); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(0); + expect(sessionService.adoptedDeployCalls).toHaveLength(0); }); }); @@ -865,9 +940,16 @@ describe("wakeFoldedRun", () => { // with no conflict handling, so the previous occurrence's rows must // go first or the redeploy dies on the primary key. const db = createFakeDb(); + // Two reads share `select`: the session lookup (`.where().orderBy()`) + // and `deployAtHead`'s definition-asset join (`.innerJoin()`). const dbWithSelect = Object.assign(db, { select: () => ({ from: () => ({ + innerJoin: () => ({ + where: () => ({ + limit: () => Promise.resolve([{ assetId: "ast_definition1" }]), + }), + }), where: () => ({ orderBy: () => ({ limit: () => Promise.resolve([{ id: "ses_1" }]), @@ -881,6 +963,7 @@ describe("wakeFoldedRun", () => { { db: dbWithSelect as never, sessionService, + assetService: createFakeAssetService(), sidecarRouter: createFakeSidecarRouter(), eventCollectors: createFakeEventCollectors(), credentialCipher: {} as never, @@ -907,14 +990,14 @@ describe("wakeFoldedRun", () => { }, ); expect(db.deleted.map((d) => d.table)).toContain(sessionAsset); - expect(sessionService.deployInstanceAtHeadCalls).toHaveLength(1); + expect(sessionService.adoptedDeployCalls).toHaveLength(1); }); }); describe("deployAtHead — mcp credential bindings", () => { const MCP_BINDING = { package: "@corbits/mcp-tools", - handle: "mcp:exa", + handle: "mcp.exa", provider: "mcp:exa", locator: "tenant" as const, }; @@ -964,6 +1047,7 @@ describe("deployAtHead — mcp credential bindings", () => { const db = createFakeDb(); const sessionService = createFakeSessionService(); + const assetService = createFakeAssetService(); const eventCollectors = createFakeEventCollectors(); const mcpCredentialBindingsForCalls: string[] = []; @@ -971,10 +1055,10 @@ describe("deployAtHead — mcp credential bindings", () => { { db: db as never, sidecarRouter: createFakeSidecarRouter(), + assetService, sessionService, eventCollectors, credentialCipher: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], mcpCredentialBindingsFor: async (tenantId: string) => { mcpCredentialBindingsForCalls.push(tenantId); @@ -1002,14 +1086,11 @@ describe("deployAtHead — mcp credential bindings", () => { bindings: [MCP_BINDING], }); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { - credentials: unknown; - config: { grants: { resource: string; action: string }[] }; - definition: { credentialBindings?: readonly unknown[] }; - }; - expect(deployed.credentials).toEqual( - buildCredentialDeliveryResult.delivery, - ); + // The deploy front resolves the credential MATERIAL itself from the + // deployed definition's own bindings, so the cipher — not a + // pre-built delivery — is what crosses the boundary. + const deployed = onlyCall(sessionService.adoptedDeployCalls); + expect(deployed.credentialCipher).toBeDefined(); expect(deployed.config.grants).toContainEqual( expect.objectContaining({ resource: "credential:cred_1", @@ -1017,7 +1098,15 @@ describe("deployAtHead — mcp credential bindings", () => { conditions: { tool: "tool:@corbits/mcp-tools" }, }), ); - expect(deployed.definition.credentialBindings).toEqual([MCP_BINDING]); + // The workflow host derives its per-step consumer bindings from the + // DEFINITION's own `credentialBindings`, and the definition is now + // whatever the deployed bytes evaluate to — so the folded-in MCP + // binding has to be inside the committed tree. + const entry = + assetService.populateAssetCalls[0]?.tree.files["workflow.js"] ?? ""; + expect(JSON.parse(entryConfigJSON(entry)).credentialBindings).toEqual([ + MCP_BINDING, + ]); }); test("never calls mcpCredentialBindingsFor when @corbits/mcp-tools is not pinned", async () => { @@ -1045,10 +1134,10 @@ describe("deployAtHead — mcp credential bindings", () => { { db: db as never, sidecarRouter: createFakeSidecarRouter(), + assetService: createFakeAssetService(), sessionService, eventCollectors, credentialCipher: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], mcpCredentialBindingsFor: async () => { mcpCredentialBindingsForCallCount += 1; @@ -1068,7 +1157,7 @@ describe("deployAtHead — mcp credential bindings", () => { expect(mcpCredentialBindingsForCallCount).toBe(0); expect(buildCredentialDeliveryCalls).toHaveLength(0); - const deployed = sessionService.deployInstanceAtHeadCalls[0] as { + const deployed = sessionService.adoptedDeployCalls[0] as { credentials?: unknown; }; expect(deployed.credentials).toBeUndefined(); @@ -1101,9 +1190,9 @@ describe("deployAtHead — run.grants production", () => { return { db: createFakeDb() as never, sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), sidecarRouter, eventCollectors: createFakeEventCollectors(), - hubPublicKey: "hub-key", toolGrantsForPins: () => [ { resource: "tool:@corbits/mcp-tools:search", @@ -1152,9 +1241,7 @@ describe("deployAtHead — run.grants production", () => { await deployAtHead(deps, PARAMS); - const deployed = deps.sessionService.deployInstanceAtHeadCalls[0] as { - config: { grants: unknown[] }; - }; + const deployed = onlyCall(deps.sessionService.adoptedDeployCalls); expect(sidecarRouter.runGrantsCalls[0]?.stepGrants).toEqual( deployed.config.grants, ); @@ -1169,3 +1256,222 @@ describe("deployAtHead — run.grants production", () => { ); }); }); + +// The whole conversion in one test: under the workflow.json retirement a +// folded run's definition is no longer synthesized in memory and handed +// to the hub — it is RENDERED into a per-run source package, COMMITTED +// into the run's own definition asset, and DEPLOYED by pinning that +// commit onto the anchor row the run already owns. +describe("deployAtHead — the code-sourced round trip", () => { + const SOURCES: DefinitionSourceResolution = { + ok: true, + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + apiKey: "placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + + const PARAMS = { + tenantId: "ten_1", + instanceId: "run_rt1", + triggerAddress: "run_rt1@ten1.workbench.test", + principalId: "prn_1", + sessionId: "ses_1", + foldedBody: { + ...FOLDED_BODY, + systemPrompt: "you answer questions", + toolPackagePins: [{ name: "@corbits/mcp-tools", version: "*" }], + }, + launchLabel: "the invited agent", + }; + + function makeDeps() { + return { + db: createFakeDb() as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter(), + eventCollectors: createFakeEventCollectors(), + toolGrantsForPins: () => [], + }; + } + + test("commits the rendered tree into the run's own definition asset", async () => { + resolveDefinitionSourcesResult = SOURCES; + const deps = makeDeps(); + + await deployAtHead(deps, PARAMS); + + expect(deps.assetService.populateAssetCalls).toHaveLength(1); + const commit = onlyCall(deps.assetService.populateAssetCalls); + // Reuse, not a second asset: the tree lands in the asset the run's + // definition already points at, on a ref of its own so one asset can + // back many runs without their bytes colliding. + expect(commit.assetId).toBe("ast_definition1"); + expect(commit.ref).toBe("refs/heads/runs/run_rt1"); + expect(Object.keys(commit.tree.files).sort()).toEqual([ + "package.json", + "workflow.js", + ]); + }); + + test("renders the run's whole config into the deployed bytes", async () => { + resolveDefinitionSourcesResult = SOURCES; + const deps = makeDeps(); + + await deployAtHead(deps, PARAMS); + + const files = onlyCall(deps.assetService.populateAssetCalls).tree.files; + const config = JSON.parse(entryConfigJSON(files["workflow.js"] ?? "")); + // Every field the approved wire hash covers has to be inside the + // bytes: a config delivered out of band diverges between the + // approval probe's evaluation and the run child's and fails closed. + expect(config).toMatchObject({ + workflowId: "wf_run_rt1", + agentId: "run_rt1", + triggerAddress: "run_rt1@ten1.workbench.test", + systemPrompt: "you answer questions", + inferencePreferences: [ + { provider: "anthropic", model: "claude-sonnet-5" }, + ], + toolPackagePins: [{ name: "@corbits/mcp-tools", version: "*" }], + mode: { kind: "step" }, + }); + expect(files["package.json"]).toContain('"@corbits/agent-runtime"'); + }); + + test("deploys the committed pin through the adopting front", async () => { + resolveDefinitionSourcesResult = SOURCES; + const deps = makeDeps(); + + await deployAtHead(deps, PARAMS); + + expect(deps.sessionService.adoptedDeployCalls).toHaveLength(1); + const deployed = onlyCall(deps.sessionService.adoptedDeployCalls); + expect(deployed).toMatchObject({ + tenantId: "ten_1", + anchorRunId: "run_rt1", + deploymentDomain: "ten1.workbench.test", + agentAddress: "run_rt1@ten1.workbench.test", + entry: "./workflow.js", + definitionAssetId: "ast_definition1", + source: { + kind: "asset", + assetId: "ast_definition1", + package: { format: "source", commitSha: "commit-sha-1" }, + }, + }); + }); + + test("carries the caller's section mode into the bytes untouched", async () => { + resolveDefinitionSourcesResult = SOURCES; + const deps = makeDeps(); + + await deployAtHead(deps, { + ...PARAMS, + mode: { kind: "section", turnTimeoutMs: 45_000 }, + }); + + const config = JSON.parse( + entryConfigJSON( + onlyCall(deps.assetService.populateAssetCalls).tree.files[ + "workflow.js" + ] ?? "", + ), + ); + // The mode is config data, so nothing about the deploy call itself + // differs between the two shapes. + expect(config.mode).toEqual({ kind: "section", turnTimeoutMs: 45_000 }); + expect(deps.sessionService.adoptedDeployCalls).toHaveLength(1); + }); + + test("refuses a run whose definition has no workflow-kind asset", async () => { + resolveDefinitionSourcesResult = SOURCES; + const deps = { ...makeDeps(), db: createFakeDb(null) as never }; + + await expect(deployAtHead(deps, PARAMS)).rejects.toThrow( + /no workflow-kind definition asset/, + ); + expect(deps.sessionService.adoptedDeployCalls).toEqual([]); + }); +}); + +describe("wakeFoldedRun — the same code-sourced path", () => { + test("re-renders and re-commits the run's source tree, then adopts its anchor", async () => { + resolveDefinitionSourcesResult = { + ok: true, + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + apiKey: "placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + const db = Object.assign(createFakeDb(), { + select: () => ({ + from: () => ({ + innerJoin: () => ({ + where: () => ({ + limit: () => Promise.resolve([{ assetId: "ast_definition1" }]), + }), + }), + where: () => ({ + orderBy: () => ({ + limit: () => Promise.resolve([{ id: "ses_1" }]), + }), + }), + }), + }), + }); + const sessionService = createFakeSessionService(); + const assetService = createFakeAssetService(); + + await wakeFoldedRun( + { + db: db as never, + sessionService, + assetService, + sidecarRouter: createFakeSidecarRouter(), + eventCollectors: createFakeEventCollectors(), + toolGrantsForPins: () => [], + } as never, + { + tenantId: "ten_1", + instanceId: "ins_woken1", + triggerAddress: "ins_woken1@ten1.workbench.test", + principalId: "prn_1", + foldedBody: FOLDED_BODY, + // A wake must repin whatever the launch pinned; the literal + // input is a property of what the run IS. + mode: { kind: "step", literalInput: "workbench-host anchor turn" }, + }, + ); + + expect(assetService.populateAssetCalls[0]?.ref).toBe( + "refs/heads/runs/ins_woken1", + ); + const config = JSON.parse( + entryConfigJSON( + assetService.populateAssetCalls[0]?.tree.files["workflow.js"] ?? "", + ), + ); + expect(config.mode).toEqual({ + kind: "step", + literalInput: "workbench-host anchor turn", + }); + expect(sessionService.adoptedDeployCalls[0]).toMatchObject({ + anchorRunId: "ins_woken1", + source: { package: { format: "source", commitSha: "commit-sha-1" } }, + }); + }); +}); diff --git a/packages/webhook-triggers/test/launch.test.ts b/packages/webhook-triggers/test/launch.test.ts index 424eac839..445156b25 100644 --- a/packages/webhook-triggers/test/launch.test.ts +++ b/packages/webhook-triggers/test/launch.test.ts @@ -77,7 +77,6 @@ function baseDeps() { sessionService: {} as never, assetService: {} as never, sidecarRouter: {} as never, - hubPublicKey: "hub-key", toolGrantsForPins: () => [], eventCollectors: {} as never, cryptoProviderCache: { get: async () => ({}) as never },