diff --git a/.env.example b/.env.example index 18dff9826..ed60bb988 100644 --- a/.env.example +++ b/.env.example @@ -71,6 +71,13 @@ HUB_STATIC_DIR=../web/dist # personal bench gets the default workflow set deployed at first login. # ANTHROPIC_API_KEY= +# Set to 1 to make `workbench seed` also deploy the zero-cost +# catalog-test workflows (heartbeat, channel-digest), which exist only +# to exercise the platform's scheduling and channel-mail paths +# continuously. Leave unset for a real bench — these are dev/CI +# tooling, never part of a real user's workflow set. +# WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS= + # Behind a reverse proxy (e.g. tailscale serve), BASE_URL is the public # origin and PORT is the local port the proxy forwards to. # PORT=3000 diff --git a/bun.lock b/bun.lock index 8ed4fe1b3..5ad5c06b4 100644 --- a/bun.lock +++ b/bun.lock @@ -279,7 +279,9 @@ "version": "0.0.1", "dependencies": { "@corbits/assistant-workflow": "workspace:*", + "@corbits/channel-digest-workflow": "workspace:*", "@corbits/echo-workflow": "workspace:*", + "@corbits/heartbeat-workflow": "workspace:*", "@intx/inference": "workspace:*", "@intx/types": "workspace:*", "arktype": "catalog:", @@ -669,6 +671,18 @@ "typescript": "catalog:", }, }, + "workflows/channel-digest": { + "name": "@corbits/channel-digest-workflow", + "version": "0.0.1", + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/workflow": "workspace:*", + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:", + }, + }, "workflows/echo": { "name": "@corbits/echo-workflow", "version": "0.0.1", @@ -681,6 +695,18 @@ "typescript": "catalog:", }, }, + "workflows/heartbeat": { + "name": "@corbits/heartbeat-workflow", + "version": "0.0.1", + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/workflow": "workspace:*", + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:", + }, + }, }, "trustedDependencies": [ "@corbits/react-ui", @@ -771,6 +797,8 @@ "@corbits/bench-ui": ["@corbits/bench-ui@workspace:packages/bench-ui"], + "@corbits/channel-digest-workflow": ["@corbits/channel-digest-workflow@workspace:workflows/channel-digest"], + "@corbits/chat": ["@corbits/chat@workspace:packages/chat"], "@corbits/chat-ui": ["@corbits/chat-ui@workspace:packages/chat-ui"], @@ -781,6 +809,8 @@ "@corbits/folded-runs": ["@corbits/folded-runs@workspace:packages/folded-runs"], + "@corbits/heartbeat-workflow": ["@corbits/heartbeat-workflow@workspace:workflows/heartbeat"], + "@corbits/notify": ["@corbits/notify@workspace:packages/notify"], "@corbits/react-ui": ["@corbits/react-ui@github:corbitsdev/react-ui#bebe1ed", { "dependencies": { "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "tailwind-merge": "^3.3.1" }, "peerDependencies": { "@radix-ui/react-dialog": "^1.1.15", "@radix-ui/react-dropdown-menu": "^2.1.16", "@radix-ui/react-slot": "^1.2.3", "@tanstack/react-query": "^5.90.2", "lucide-react": "^0.545.0 || ^1.0.0", "react": "^18.2.0 || ^19.0.0", "react-dom": "^18.2.0 || ^19.0.0", "sonner": "^2.0.7" }, "optionalPeers": ["@tanstack/react-query"] }, "corbitsdev-react-ui-bebe1ed", "sha512-gQswotVhBuFuqiT8/Xy4vJXkJI0EBdxaXLnWGoH5fLH7M1O0OGuLDBdhSwN89SFb3gOYGDHwj9QngHOxssrBEg=="], diff --git a/packages/cli/src/config.ts b/packages/cli/src/config.ts index 6eabcf8a0..78cbadcd6 100644 --- a/packages/cli/src/config.ts +++ b/packages/cli/src/config.ts @@ -52,6 +52,9 @@ const SeedEnv = type({ "ANTHROPIC_API_KEY?": type("string > 0").describe( "your Anthropic API key; optional, but required for the tenant catalog to be launchable", ), + "WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS?": type("string").describe( + "set to 1 to also deploy the zero-cost catalog-test workflows (heartbeat, channel-digest); a dev/CI-only opt-in, never set for a real bench", + ), }); const DEFAULT_MODEL_PROVIDER = "anthropic"; @@ -80,6 +83,13 @@ export type SeedConfig = { * deploys. */ readonly modelSource: ModelSource; readonly anthropicApiKeyConfigured: boolean; + /** + * Opt-in, from WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS: also deploy + * the zero-cost catalog-test workflows (heartbeat, channel-digest) + * alongside the real default set. Unset for a real bench — these + * exist only to exercise the platform, not for a real user. + */ + readonly seedCatalogTestWorkflows: boolean; }; function environmentError(command: string, problems: string[]): CliError { @@ -170,6 +180,8 @@ export function readSeedConfig( apiKey: apiKey ?? PLACEHOLDER_CATALOG_API_KEY, }, anthropicApiKeyConfigured: apiKey !== undefined, + seedCatalogTestWorkflows: + parsed.WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS === "1", }; } diff --git a/packages/cli/src/seed.ts b/packages/cli/src/seed.ts index 68bf89ba8..977113a77 100644 --- a/packages/cli/src/seed.ts +++ b/packages/cli/src/seed.ts @@ -9,6 +9,7 @@ import { parseAs, seedCatalog, seedTenant, + CATALOG_TEST_WORKFLOWS, DEFAULT_WORKFLOWS, type ApiCall, type DefaultWorkflow, @@ -73,9 +74,25 @@ async function resolveTenant( }; } +/** + * The workflow set a plain `workbench seed` deploys: the real default + * set every tenant gets, plus the zero-cost catalog-test workflows + * only when the caller has explicitly opted in via + * `WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS`. A real bench never sets + * that variable, so it only ever gets `DEFAULT_WORKFLOWS` — the same + * set `provisionPersonalTenantIfNeeded` deploys on first login. + */ +export function resolveSeedWorkflows( + config: Pick, +): readonly DefaultWorkflow[] { + return config.seedCatalogTestWorkflows + ? [...DEFAULT_WORKFLOWS, ...CATALOG_TEST_WORKFLOWS] + : DEFAULT_WORKFLOWS; +} + export async function runSeed( deps: SeedDeps, - workflows: readonly DefaultWorkflow[] = DEFAULT_WORKFLOWS, + workflows?: readonly DefaultWorkflow[], ): Promise { const { config, api, log } = deps; if (config.adminDefaulted) { @@ -83,6 +100,12 @@ export async function runSeed( "using default admin alice@example.com — set HUB_ADMIN_EMAIL and HUB_ADMIN_PASSWORD for real deployments", ); } + const resolvedWorkflows = workflows ?? resolveSeedWorkflows(config); + if (config.seedCatalogTestWorkflows) { + log( + "WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS=1: also deploying the zero-cost catalog-test workflows (heartbeat, channel-digest)", + ); + } const session = await authenticate(api, { email: config.adminEmail, @@ -100,7 +123,7 @@ export async function runSeed( model: config.modelSource, pushWorkflow: deps.pushWorkflow, log, - workflows, + workflows: resolvedWorkflows, }; if (deps.sleep !== undefined) seedArgs.sleep = deps.sleep; if (deps.runStartTimeoutMs !== undefined) diff --git a/packages/cli/test/config.test.ts b/packages/cli/test/config.test.ts index 91deb4fa0..6da871d8b 100644 --- a/packages/cli/test/config.test.ts +++ b/packages/cli/test/config.test.ts @@ -95,4 +95,24 @@ describe("readSeedConfig", () => { expect(MODEL_CREDENTIAL_VARIABLES.length).toBe(1); expect(MODEL_CREDENTIAL_VARIABLES[0]).toContain("ANTHROPIC_API_KEY"); }); + + test("the catalog-test workflow opt-in defaults off", () => { + expect(readSeedConfig(VALID_SHARED).seedCatalogTestWorkflows).toBe(false); + }); + + test("WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS=1 opts into the catalog-test workflows", () => { + const config = readSeedConfig({ + ...VALID_SHARED, + WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS: "1", + }); + expect(config.seedCatalogTestWorkflows).toBe(true); + }); + + test("any other value for WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS stays opted out", () => { + const config = readSeedConfig({ + ...VALID_SHARED, + WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS: "true", + }); + expect(config.seedCatalogTestWorkflows).toBe(false); + }); }); diff --git a/packages/cli/test/seed.test.ts b/packages/cli/test/seed.test.ts index 727464e00..57c834313 100644 --- a/packages/cli/test/seed.test.ts +++ b/packages/cli/test/seed.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; import { CliError } from "@workbench/hub-client"; import type { SeedConfig } from "../src/config"; -import { runSeed, type SeedDeps } from "../src/seed"; +import { resolveSeedWorkflows, runSeed, type SeedDeps } from "../src/seed"; import { collector, fakeAPI, @@ -25,6 +25,7 @@ const CONFIG: SeedConfig = { apiKey: "placeholder-not-a-real-key", }, anthropicApiKeyConfigured: false, + seedCatalogTestWorkflows: false, }; function deps(overrides: Partial & Pick): SeedDeps { @@ -37,6 +38,22 @@ function deps(overrides: Partial & Pick): SeedDeps { }; } +describe("resolveSeedWorkflows", () => { + test("without the opt-in, only the real default workflow set is deployed", () => { + const names = resolveSeedWorkflows({ + seedCatalogTestWorkflows: false, + }).map((w) => w.assetName); + expect(names).toEqual(["echo", "assistant"]); + }); + + test("with the opt-in, the catalog-test workflows are appended", () => { + const names = resolveSeedWorkflows({ + seedCatalogTestWorkflows: true, + }).map((w) => w.assetName); + expect(names).toEqual(["echo", "assistant", "heartbeat", "channel-digest"]); + }); +}); + describe("runSeed", () => { test("authenticates, resolves the bench by slug, and starts seeding it", async () => { const { lines, log } = collector(); diff --git a/packages/hub-client/package.json b/packages/hub-client/package.json index e73bcb364..8ae7f2c32 100644 --- a/packages/hub-client/package.json +++ b/packages/hub-client/package.json @@ -17,7 +17,9 @@ "@intx/types": "workspace:*", "arktype": "catalog:", "@corbits/assistant-workflow": "workspace:*", - "@corbits/echo-workflow": "workspace:*" + "@corbits/channel-digest-workflow": "workspace:*", + "@corbits/echo-workflow": "workspace:*", + "@corbits/heartbeat-workflow": "workspace:*" }, "devDependencies": { "@types/bun": "catalog:", diff --git a/packages/hub-client/src/index.ts b/packages/hub-client/src/index.ts index 5f35ad0e5..254acf911 100644 --- a/packages/hub-client/src/index.ts +++ b/packages/hub-client/src/index.ts @@ -16,6 +16,7 @@ export type { WorkflowPusher, } from "./seed"; export { + CATALOG_TEST_WORKFLOWS, DEFAULT_WORKFLOWS, PLACEHOLDER_CATALOG_API_KEY, seedCatalog, diff --git a/packages/hub-client/src/seed.ts b/packages/hub-client/src/seed.ts index 4f492752c..1fb11e809 100644 --- a/packages/hub-client/src/seed.ts +++ b/packages/hub-client/src/seed.ts @@ -21,10 +21,18 @@ import { buildAssistantWorkflow, serializeAssistantWorkflow, } from "@corbits/assistant-workflow"; +import { + buildChannelDigestWorkflow, + serializeChannelDigestWorkflow, +} from "@corbits/channel-digest-workflow"; import { buildEchoWorkflow, serializeEchoWorkflow, } from "@corbits/echo-workflow"; +import { + buildHeartbeatWorkflow, + serializeHeartbeatWorkflow, +} from "@corbits/heartbeat-workflow"; import { CliError } from "./errors"; import { parseAs, type ApiCall } from "./hub"; import { catalogModel, catalogProvider } from "./catalog-seed-data"; @@ -32,6 +40,12 @@ import { catalogModel, catalogProvider } from "./catalog-seed-data"; const GIT_TOKEN_TTL_MS = 10 * 60 * 1000; const ECHO_TURN_TIMEOUT_MS = 2 * 60 * 1000; const ASSISTANT_TURN_TIMEOUT_MS = 2 * 60 * 1000; +// Short: these two run on a tight, continuous schedule to exercise +// scheduling itself, so a wedged noop-inference call should surface +// fast rather than tie up a run slot for the full two minutes the +// conversational workflows above allow. +const HEARTBEAT_TURN_TIMEOUT_MS = 30 * 1000; +const CHANNEL_DIGEST_TURN_TIMEOUT_MS = 30 * 1000; const RUN_START_TIMEOUT_MS = 30_000; const RUN_POLL_INTERVAL_MS = 1000; @@ -40,6 +54,34 @@ const RUN_POLL_INTERVAL_MS = 1000; // is exactly one honest value for it. const SEED_SOURCE_ID = "default"; +// The provider/model pair `noop-inference` (packages/chat/src/noop-inference.ts) +// answers for any request, regardless of what is actually sent — the +// route ignores its body and `x-api-key` entirely. Naming a distinct +// pair here (rather than reusing the tenant's real model id) keeps a +// noop-pinned deployment visually distinct from a real one in the hub's +// UI and logs. +const NOOP_PROVIDER = "anthropic"; +const NOOP_MODEL = "noop"; + +/** + * A `ModelSource` pointed at the hub's own `noop-inference` endpoint + * instead of a real provider — the same substitution + * `packages/chat/src/platform-adapter.ts`'s `noopSourcesOverride` makes + * for channel-host launches, reused here so a workflow deployed with + * this source resolves every turn instantly against a constant, + * locally served reply and never reaches a real model. `hubUrl` is the + * same base URL `seedTenant` already receives, so no new configuration + * is required to use it. + */ +export function NOOP_MODEL_SOURCE(hubUrl: string): ModelSource { + return { + provider: NOOP_PROVIDER, + model: NOOP_MODEL, + baseURL: `${hubUrl}/api/chat/noop-inference`, + apiKey: "noop", + }; +} + const GitTokenMintResponse = type({ id: "string", secret: "string" }); const WorkflowDeploymentResponse = type({ id: "string", @@ -74,12 +116,25 @@ export type DefaultWorkflow = { /** Asset name; lowercase-kebab so the smart-HTTP repo path is clean. */ assetName: string; buildJson: (tenantDomain: string, model: ModelSource) => string; + /** + * Overrides the deploy's inference source for this workflow only, + * given the hub's own base URL. Present on the catalog-test workflows + * `heartbeat` and `channel-digest`, which must stay free to run + * continuously: it names `NOOP_MODEL_SOURCE` instead of the tenant's + * real catalog model. Absent on every conversational workflow, which + * deploys against the tenant's real model as before. + */ + modelSource?: (hubUrl: string) => ModelSource; }; /** - * The workflow set a bench starts with: the echo walking-skeleton and - * the general-purpose assistant. Growing the set is adding an entry - * here, nothing more. + * The workflow set every real tenant starts with: the echo + * walking-skeleton and the general-purpose assistant. This is what + * `provisionPersonalTenantIfNeeded` (`@workbench/onboarding`) deploys + * on first login for every real user — growing it is adding an entry + * here, nothing more, but an entry here reaches every signup, so it is + * never the place for a workflow that exists only to exercise the + * platform itself. See `CATALOG_TEST_WORKFLOWS` for those. */ export const DEFAULT_WORKFLOWS: readonly DefaultWorkflow[] = [ { @@ -110,6 +165,48 @@ export const DEFAULT_WORKFLOWS: readonly DefaultWorkflow[] = [ }, ]; +/** + * Zero-cost workflows that exist to exercise the platform continuously + * — `heartbeat` proves the scheduling and mail-trigger paths, + * `channel-digest` proves the channel-mail-posting path — never to + * give a real user something to use. Both are pinned at + * `NOOP_MODEL_SOURCE` so running them on a tight schedule costs + * nothing. Deliberately absent from `DEFAULT_WORKFLOWS`: a real signup + * goes through `provisionPersonalTenantIfNeeded`, which never seeds + * this set. Only an explicit, dev/CI-specific caller (`workbench + * seed` with `WORKBENCH_SEED_CATALOG_TEST_WORKFLOWS` set) opts in. + */ +export const CATALOG_TEST_WORKFLOWS: readonly DefaultWorkflow[] = [ + { + assetName: "heartbeat", + buildJson: (tenantDomain, model) => + serializeHeartbeatWorkflow( + buildHeartbeatWorkflow({ + triggerAddress: `heartbeat@${tenantDomain}`, + inferencePreferences: [ + { provider: model.provider, model: model.model }, + ], + turnTimeoutMs: HEARTBEAT_TURN_TIMEOUT_MS, + }), + ), + modelSource: NOOP_MODEL_SOURCE, + }, + { + assetName: "channel-digest", + buildJson: (tenantDomain, model) => + serializeChannelDigestWorkflow( + buildChannelDigestWorkflow({ + triggerAddress: `channel-digest@${tenantDomain}`, + inferencePreferences: [ + { provider: model.provider, model: model.model }, + ], + turnTimeoutMs: CHANNEL_DIGEST_TURN_TIMEOUT_MS, + }), + ), + modelSource: NOOP_MODEL_SOURCE, + }, +]; + // The grants the deploy, trigger, and run-listing routes gate on, // planted at the wildcard scope the authz glob matcher resolves // against any concrete deployment (the deployment id is minted at @@ -468,6 +565,7 @@ export async function seedTenant(args: SeedTenantArgs): Promise { let confirmed = 0; for (const workflow of workflows) { + const workflowModel = workflow.modelSource?.(hubUrl) ?? model; const assetId = await ensureWorkflowAsset( api, cookies, @@ -479,7 +577,7 @@ export async function seedTenant(args: SeedTenantArgs): Promise { const outcome = await args.pushWorkflow({ remoteUrl: `${hubUrl}/api/tenants/${tenant.tenantId}/assets/workflow/${workflow.assetName}.git`, tokenSecret, - workflowJson: workflow.buildJson(tenant.domain, model), + workflowJson: workflow.buildJson(tenant.domain, workflowModel), }); log( outcome === "pushed" @@ -494,7 +592,7 @@ export async function seedTenant(args: SeedTenantArgs): Promise { tenantId: tenant.tenantId, assetId, assetName: workflow.assetName, - model, + model: workflowModel, }, log, ); diff --git a/packages/hub-client/test/seed.test.ts b/packages/hub-client/test/seed.test.ts index de82e1027..177827c2a 100644 --- a/packages/hub-client/test/seed.test.ts +++ b/packages/hub-client/test/seed.test.ts @@ -1,7 +1,9 @@ import { describe, expect, test } from "bun:test"; import { CliError } from "../src/errors"; import { + CATALOG_TEST_WORKFLOWS, DEFAULT_WORKFLOWS, + NOOP_MODEL_SOURCE, seedCatalog, seedTenant, type SeedTenantArgs, @@ -435,6 +437,247 @@ describe("seedTenant", () => { test("the default set also includes the assistant workflow", () => { expect(DEFAULT_WORKFLOWS.map((w) => w.assetName)).toContain("assistant"); }); + + test("NOOP_MODEL_SOURCE resolves to the hub's own noop-inference endpoint", () => { + const resolved = NOOP_MODEL_SOURCE("http://localhost:3000"); + expect(resolved.baseURL).toBe( + "http://localhost:3000/api/chat/noop-inference", + ); + expect(resolved.model).toBe("noop"); + }); + + test("echo and assistant carry no modelSource override, so they deploy against the tenant's real model", () => { + const realModelWorkflows = DEFAULT_WORKFLOWS.filter( + (w) => w.assetName === "echo" || w.assetName === "assistant", + ); + expect(realModelWorkflows).toHaveLength(2); + for (const workflow of realModelWorkflows) { + expect(workflow.modelSource).toBeUndefined(); + } + }); + + test("the default set consumed by real tenant provisioning is exactly echo and assistant", () => { + // provisionPersonalTenantIfNeeded (@workbench/onboarding) deploys + // DEFAULT_WORKFLOWS for every real signup. The catalog-test + // workflows exist only to exercise the platform continuously and + // must never reach a real user through this array — they are + // seeded only via the explicit CATALOG_TEST_WORKFLOWS opt-in. + expect(DEFAULT_WORKFLOWS.map((w) => w.assetName)).toEqual([ + "echo", + "assistant", + ]); + }); + + test("every non-conversational default declares a modelSource override", () => { + // echo and assistant deploy against the tenant's real model and + // must not declare one; a future addition to DEFAULT_WORKFLOWS + // that silently picks up real inference is exactly the class of + // regression this guards against. + for (const workflow of DEFAULT_WORKFLOWS) { + expect(workflow.modelSource).toBeUndefined(); + } + for (const workflow of CATALOG_TEST_WORKFLOWS) { + expect(workflow.modelSource).toBeDefined(); + } + }); + + test("the catalog-test set includes the heartbeat workflow", () => { + expect(CATALOG_TEST_WORKFLOWS.map((w) => w.assetName)).toContain( + "heartbeat", + ); + }); + + test("heartbeat pins its deploy source at noop-inference, never the tenant's real model", () => { + const heartbeat = CATALOG_TEST_WORKFLOWS.find( + (w) => w.assetName === "heartbeat", + ); + if (!heartbeat) throw new Error("expected the heartbeat workflow"); + const resolved = heartbeat.modelSource?.("http://localhost:3000"); + expect(resolved).toEqual(NOOP_MODEL_SOURCE("http://localhost:3000")); + }); + + test("fresh run pushes, deploys, and confirms the heartbeat workflow against the noop source", async () => { + const { lines, log } = collector(); + const { pushes, push } = recordingPusher(); + let runsCalls = 0; + let deployedSources: unknown; + const handler: FakeHandler = (method, path, body) => { + const base = baseRoutes(method, path); + if (base) return base; + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/assets`) + return { status: 201, data: assetRow("ast_3", "heartbeat") }; + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) + return { status: 200, data: [] }; + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) { + deployedSources = body; + return { status: 201, data: deploymentRow("dep_3", "ast_3", "active") }; + } + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/dep_3/runs` + ) { + runsCalls += 1; + return { + status: 200, + data: { runIds: runsCalls === 1 ? [] : ["run_1"] }, + }; + } + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/dep_3/mail` + ) + return { + status: 202, + data: { + deploymentId: "dep_3", + address: `ins_dep_3@${TENANT_DOMAIN}`, + messageId: "", + }, + }; + return undefined; + }; + + const heartbeatOnly = CATALOG_TEST_WORKFLOWS.filter( + (w) => w.assetName === "heartbeat", + ); + await seedTenant( + args({ + api: fakeAPI(handler), + pushWorkflow: push, + log, + workflows: heartbeatOnly, + }), + ); + + expect(pushes).toHaveLength(1); + const push0 = pushes[0]; + if (!push0) throw new Error("expected one workflow push"); + const definition = JSON.parse(push0.workflowJson) as { + id: string; + triggers: { type: string; to: string }[]; + stepOrder: string[]; + }; + expect(definition.id).toBe("wf_heartbeat"); + expect(definition.triggers[0]?.to).toBe(`heartbeat@${TENANT_DOMAIN}`); + expect(definition.stepOrder).toEqual(["heartbeat"]); + + // The deploy's own source, not the tenant's real MODEL, is what + // proves the noop pin took effect: it must name the noop provider + // fixture, not the ordinary anthropic/claude-sonnet-4-5 model this + // test file's `args()` helper hands every other workflow. + const deployedBody = deployedSources as { sources: { model: string }[] }; + expect(deployedBody.sources[0]?.model).toBe("noop"); + + const output = lines.join("\n"); + expect(output).toContain("deployed workflow heartbeat as dep_3"); + expect(output).toContain("confirmed workflow heartbeat: run run_1 started"); + }); + + test("the catalog-test set includes the channel-digest workflow", () => { + expect(CATALOG_TEST_WORKFLOWS.map((w) => w.assetName)).toContain( + "channel-digest", + ); + }); + + test("channel-digest pins its deploy source at noop-inference, never the tenant's real model", () => { + const channelDigest = CATALOG_TEST_WORKFLOWS.find( + (w) => w.assetName === "channel-digest", + ); + if (!channelDigest) throw new Error("expected the channel-digest workflow"); + const resolved = channelDigest.modelSource?.("http://localhost:3000"); + expect(resolved).toEqual(NOOP_MODEL_SOURCE("http://localhost:3000")); + }); + + test("fresh run pushes, deploys, and confirms the channel-digest workflow against the noop source", async () => { + const { lines, log } = collector(); + const { pushes, push } = recordingPusher(); + let runsCalls = 0; + let deployedSources: unknown; + const handler: FakeHandler = (method, path, body) => { + const base = baseRoutes(method, path); + if (base) return base; + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/assets`) + return { status: 201, data: assetRow("ast_4", "channel-digest") }; + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) + return { status: 200, data: [] }; + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) { + deployedSources = body; + return { status: 201, data: deploymentRow("dep_4", "ast_4", "active") }; + } + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/dep_4/runs` + ) { + runsCalls += 1; + return { + status: 200, + data: { runIds: runsCalls === 1 ? [] : ["run_1"] }, + }; + } + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/dep_4/mail` + ) + return { + status: 202, + data: { + deploymentId: "dep_4", + address: `ins_dep_4@${TENANT_DOMAIN}`, + messageId: "", + }, + }; + return undefined; + }; + + const channelDigestOnly = CATALOG_TEST_WORKFLOWS.filter( + (w) => w.assetName === "channel-digest", + ); + await seedTenant( + args({ + api: fakeAPI(handler), + pushWorkflow: push, + log, + workflows: channelDigestOnly, + }), + ); + + expect(pushes).toHaveLength(1); + const push0 = pushes[0]; + if (!push0) throw new Error("expected one workflow push"); + const definition = JSON.parse(push0.workflowJson) as { + id: string; + triggers: { type: string; to: string }[]; + stepOrder: string[]; + }; + expect(definition.id).toBe("wf_channel_digest"); + expect(definition.triggers[0]?.to).toBe(`channel-digest@${TENANT_DOMAIN}`); + expect(definition.stepOrder).toEqual(["channel-digest"]); + + // The deploy's own source, not the tenant's real MODEL, is what + // proves the noop pin took effect: it must name the noop provider + // fixture, not the ordinary anthropic/claude-sonnet-4-5 model this + // test file's `args()` helper hands every other workflow. + const deployedBody = deployedSources as { sources: { model: string }[] }; + expect(deployedBody.sources[0]?.model).toBe("noop"); + + const output = lines.join("\n"); + expect(output).toContain("deployed workflow channel-digest as dep_4"); + expect(output).toContain( + "confirmed workflow channel-digest: run run_1 started", + ); + }); }); const TIMESTAMP = "2026-01-01T00:00:00.000Z"; diff --git a/scripts/checks/kill-dates.txt b/scripts/checks/kill-dates.txt index 8c9c536f0..2897348b9 100644 --- a/scripts/checks/kill-dates.txt +++ b/scripts/checks/kill-dates.txt @@ -22,7 +22,7 @@ vendor/intx/hub-agent | sawyer | 2026-09-05 | 7c032e067f61059f623294ccbc48f4c7e7 vendor/intx/hub-api | sawyer | 2026-09-05 | 5eee9b1055e761e7e47db0b30db2764e6431b6666acf161d2713502f3d1f2fa4 vendor/intx/hub-common | sawyer | 2026-09-05 | bb0786ebf03cd0a6df71ca54f7e2bb845983bd406a7dac39190fb6601349f6af vendor/intx/hub-sessions | sawyer | 2026-09-05 | 5fbb566918a7d6147b6ba833277bf627ddbb45d2158c0400e88ed25dd70be66c -vendor/intx/inference | sawyer | 2026-09-05 | 8557d493be62c2bb35b87c992da5470f7d875882a68299ab425e0d0d36492a46 +vendor/intx/inference | sawyer | 2026-09-05 | 4fe554131a34b3d3a427648de16e0680b08548f95ebe9623a51818c55b9d19d2 vendor/intx/log | sawyer | 2026-09-05 | eda0b157097221cf03f286e226832e0607e9b12d2057b8d719b092ede05d6ee2 vendor/intx/mail-memory | sawyer | 2026-09-05 | dbc44f9812f59516f3660a9f9a7a14c7f9405c0c649f95bd10aff2823f32f58e vendor/intx/mime | sawyer | 2026-09-05 | 713e4e287f3916ce7f78a38b830da571eb6f8d70c03de0a6d7a84377fbbc0afd diff --git a/scripts/e2e/channel-digest.test.ts b/scripts/e2e/channel-digest.test.ts new file mode 100644 index 000000000..8724f6e0e --- /dev/null +++ b/scripts/e2e/channel-digest.test.ts @@ -0,0 +1,323 @@ +// A light end-to-end smoke test for the channel-digest workflow: the +// real hub and sidecar as spawned processes against a real Postgres, a +// channel-digest deployment whose inference source is the hub's own +// `noop-inference` endpoint (not a placeholder, not a real provider), +// and a trigger that runs the step to completion. +// +// This is the proof-by-construction that channel-digest costs nothing +// to run frequently: the deploy's source is a real, reachable endpoint +// (unlike the walking skeleton's `https://inference.invalid` +// placeholder), so a run started against it actually resolves its +// inference call — against `noop-inference`'s constant, locally +// served reply, never a real model. + +import { afterAll, describe, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { resetSchema, setupDatabase } from "../db-setup.ts"; +import { + CHANNEL_DIGEST_STEP_ID, + buildChannelDigestWorkflow, + serializeChannelDigestWorkflow, +} from "../../workflows/channel-digest/src/index.ts"; +import { + api, + e2eDatabaseUrl, + expectStatus, + expectStepCompleted, + freePort, + hop, + provisionSidecar, + pushWorkflowJson, + startHub, + startSidecar, + waitForRunCompletion, + type ApiResult, + type HubHandle, + type SpawnedApp, +} from "./harness.ts"; + +const databaseUrl = e2eDatabaseUrl(); +if (databaseUrl === undefined) { + console.warn( + "channel-digest: DATABASE_URL is not set; suite skipped. " + + "Set DATABASE_URL (see .env.example) to run it; " + + "CI sets E2E_REQUIRED=1 so this skip can never pass silently there.", + ); +} + +function stringField(data: unknown, field: string, what: string): string { + if (typeof data === "object" && data !== null && field in data) { + const value = (data as Record)[field]; + if (typeof value === "string" && value !== "") return value; + } + throw new Error( + `${what}: missing string field "${field}": ${JSON.stringify(data)}`, + ); +} + +function runIds(data: unknown): string[] { + if ( + typeof data === "object" && + data !== null && + "runIds" in data && + Array.isArray((data as Record)["runIds"]) + ) { + return (data as { runIds: unknown[] }).runIds.filter( + (id): id is string => typeof id === "string", + ); + } + throw new Error(`expected a runIds array: ${JSON.stringify(data)}`); +} + +const cleanups: (() => Promise)[] = []; + +afterAll(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function tempDir(prefix: string): Promise { + const dir = await mkdtemp(path.join(tmpdir(), prefix)); + cleanups.push(() => rm(dir, { recursive: true, force: true })); + return dir; +} + +function track(app: SpawnedApp): void { + cleanups.push(() => app.stop()); +} + +describe.skipIf(databaseUrl === undefined)("channel-digest workflow", () => { + test("launching channel-digest against the hub's own noop-inference endpoint completes a run", async () => { + const url = databaseUrl; + if (url === undefined) throw new Error("unreachable: suite is skipped"); + + await hop("database setup", async () => { + await resetSchema(url); + await setupDatabase(url); + }); + + const sidecarId = "sidecar-e2e-channel-digest"; + const sidecarToken = crypto.randomUUID(); + await hop("sidecar provisioning", () => + provisionSidecar(url, sidecarId, sidecarToken), + ); + + const hub: HubHandle = await hop("hub boot", async () => { + const handle = await startHub({ + databaseUrl: url, + port: freePort(), + sessionSecret: Buffer.from( + crypto.getRandomValues(new Uint8Array(32)), + ).toString("hex"), + dataDir: await tempDir("e2e-channel-digest-hub-data-"), + }); + track(handle); + return handle; + }); + + const sidecar = await hop("sidecar boot", async () => { + const app = startSidecar({ + hubPort: new URL(hub.baseUrl).port + ? Number(new URL(hub.baseUrl).port) + : 80, + sidecarId, + token: sidecarToken, + dataDir: await tempDir("e2e-channel-digest-sidecar-data-"), + }); + track(app); + return app; + }); + + const user = await hop("sign-up", async () => { + const res = await api(hub.baseUrl, "POST", "/api/auth/sign-up/email", { + name: "Channel Digest Tester", + email: `channel-digest-${crypto.randomUUID()}@example.invalid`, + password: `pw-${crypto.randomUUID()}`, + }); + expectStatus("sign-up", res, 200); + if (res.cookies.length === 0) { + throw new Error("sign-up returned no session cookie"); + } + return res; + }); + + const slug = `e2ecd${crypto.randomUUID().slice(0, 8)}`; + const tenantId = await hop("tenant creation", async () => { + const res = await api( + hub.baseUrl, + "POST", + "/api/tenants", + { name: "Channel Digest Smoke", slug }, + user.cookies, + ); + expectStatus("create tenant", res, 201); + return stringField(res.data, "id", "create tenant"); + }); + + const assetName = "channel-digest"; + const assetId = await hop("workflow asset publication", async () => { + const created = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/assets`, + { kind: "workflow", name: assetName }, + user.cookies, + ); + expectStatus("create workflow asset", created, 201); + const id = stringField(created.data, "id", "create workflow asset"); + + const minted = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/git-tokens`, + { + name: "e2e-channel-digest-push", + resource: "asset:*", + refPattern: "**", + actions: ["can_read", "can_push"], + expiresAt: new Date(Date.now() + 10 * 60_000).toISOString(), + }, + user.cookies, + ); + expectStatus("mint git token", minted, 201); + + const definition = buildChannelDigestWorkflow({ + triggerAddress: `channel-digest@${slug}.localhost`, + inferencePreferences: [{ provider: "anthropic", model: "noop" }], + turnTimeoutMs: 30_000, + }); + await pushWorkflowJson({ + baseUrl: hub.baseUrl, + tenantId, + assetName, + tokenSecret: stringField(minted.data, "secret", "mint git token"), + workflowJson: serializeChannelDigestWorkflow(definition), + }); + return id; + }); + + // The deploy's source is the hub's own, really-reachable + // noop-inference endpoint — not a placeholder like the walking + // skeleton's `https://inference.invalid`. That distinction is the + // whole point of this suite: a run started against this source + // actually completes an inference call, at zero cost, because + // noop-inference answers it locally without reaching a real model. + const deploymentId = await hop("workflow deploy", async () => { + const sourceId = "src-channel-digest-e2e"; + const body = { + assetId, + sources: [ + { + id: sourceId, + provider: "anthropic", + baseURL: `${hub.baseUrl}/api/chat/noop-inference`, + apiKey: "noop", + model: "noop", + }, + ], + defaultSource: sourceId, + }; + const deadline = Date.now() + 60_000; + let res: ApiResult; + for (;;) { + if (sidecar.exited()) { + throw new Error( + `sidecar exited before deploy; output:\n${sidecar.output()}`, + ); + } + res = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/workflows/instances`, + body, + user.cookies, + ); + if (res.status !== 502) break; + if (Date.now() > deadline) { + throw new Error( + `sidecar never became deployable (hub kept answering 502): ` + + `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, + ); + } + await Bun.sleep(1000); + } + expectStatus("deploy channel-digest workflow", res, 201); + return stringField(res.data, "id", "deploy channel-digest workflow"); + }); + + const startedRunId = await hop( + "channel-digest run starts against noop-inference", + async () => { + const before = new Set( + runIds( + ( + await api( + hub.baseUrl, + "GET", + `/api/tenants/${tenantId}/workflows/${deploymentId}/runs`, + undefined, + user.cookies, + ) + ).data, + ), + ); + + const triggered = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/workflows/${deploymentId}/mail`, + { content: "message count: 0" }, + user.cookies, + ); + expectStatus("trigger channel-digest mail", triggered, 202); + + const deadline = Date.now() + 30_000; + for (;;) { + const listed = await api( + hub.baseUrl, + "GET", + `/api/tenants/${tenantId}/workflows/${deploymentId}/runs`, + undefined, + user.cookies, + ); + const started = runIds(listed.data).find((id) => !before.has(id)); + if (started !== undefined) return started; + if (Date.now() > deadline) { + throw new Error( + "channel-digest trigger was accepted but no run started within 30s", + ); + } + await Bun.sleep(500); + } + }, + ); + + // The real gate: a run id proves only that the mail route accepted + // the trigger. Whether the deployment actually resolves — the + // step's agent launching, its turn completing against + // noop-inference, the run reaching a terminal state — is only + // proven by the run's own event log. A broken agent launch or a + // rejected inference call surfaces here as RunFailed (or no + // terminal event at all), failing this loudly instead of a + // "started" run standing in for a working platform. + const events = await hop("channel-digest run completes", () => + waitForRunCompletion( + hub.baseUrl, + tenantId, + deploymentId, + startedRunId, + user.cookies, + 30_000, + ), + ); + expectStepCompleted(events, CHANNEL_DIGEST_STEP_ID); + + console.log( + "channel-digest: gate achieved: a run completed against the " + + "real, reachable noop-inference source, proving the " + + "deployment resolves at zero cost.", + ); + }, 180_000); +}); diff --git a/scripts/e2e/harness.ts b/scripts/e2e/harness.ts index 6a0d7f0cd..4325a987b 100644 --- a/scripts/e2e/harness.ts +++ b/scripts/e2e/harness.ts @@ -341,6 +341,91 @@ export function expectStatus( } } +export type RunEvent = { seq: number; type: string; body: unknown }; + +function runEvents(data: unknown): RunEvent[] { + if ( + typeof data === "object" && + data !== null && + "events" in data && + Array.isArray((data as Record)["events"]) + ) { + return (data as { events: RunEvent[] }).events; + } + throw new Error(`expected a run events array: ${JSON.stringify(data)}`); +} + +const TERMINAL_EVENT_TYPES = ["RunCompleted", "RunFailed", "RunCancelled"]; + +/** + * Polls a run's event log until a terminal event lands, then requires + * it to be `RunCompleted` — not merely that the run started. A trigger + * accepted by the hub only proves the mail route works; a broken + * agent launch, a wedged inference call, or a rejected step surfaces + * as `RunFailed` (or no terminal event at all before the deadline), + * either of which fails this loudly instead of a workflow silently + * "succeeding" on nothing more than its own acceptance. + */ +export async function waitForRunCompletion( + baseUrl: string, + tenantId: string, + deploymentId: string, + runId: string, + cookies: string[], + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + const res = await api( + baseUrl, + "GET", + `/api/tenants/${tenantId}/workflows/${deploymentId}/runs/${runId}/events`, + undefined, + cookies, + ); + expectStatus("read run events", res, 200); + const events = runEvents(res.data); + const terminal = events.find((e) => TERMINAL_EVENT_TYPES.includes(e.type)); + if (terminal !== undefined) { + if (terminal.type !== "RunCompleted") { + throw new Error( + `run ${runId} ended in ${terminal.type}, not RunCompleted: ` + + JSON.stringify(terminal.body), + ); + } + return events; + } + if (Date.now() > deadline) { + throw new Error( + `run ${runId} reached no terminal event within ${Math.round(timeoutMs / 1000)}s; ` + + `events so far: ${JSON.stringify(events)}`, + ); + } + await Bun.sleep(500); + } +} + +/** + * Asserts the run's event log recorded a completed step with the + * given step id — the actual per-step execution, not just the run's + * own start/stop bookkeeping. + */ +export function expectStepCompleted(events: RunEvent[], stepId: string): void { + const completed = events.find( + (e) => + e.type === "StepCompleted" && + typeof e.body === "object" && + e.body !== null && + "stepId" in e.body && + (e.body as Record)["stepId"] === stepId, + ); + if (completed === undefined) { + throw new Error( + `no StepCompleted event for step "${stepId}"; events: ${JSON.stringify(events)}`, + ); + } +} + // --- workflow asset content over git smart-HTTP ----------------------- interface GitResult { diff --git a/scripts/e2e/heartbeat.test.ts b/scripts/e2e/heartbeat.test.ts new file mode 100644 index 000000000..b980632c0 --- /dev/null +++ b/scripts/e2e/heartbeat.test.ts @@ -0,0 +1,323 @@ +// A light end-to-end smoke test for the heartbeat workflow: the real +// hub and sidecar as spawned processes against a real Postgres, a +// heartbeat deployment whose inference source is the hub's own +// `noop-inference` endpoint (not a placeholder, not a real provider), +// and a trigger that starts a run. +// +// This is the proof-by-construction that heartbeat costs nothing to +// run frequently: the deploy's source is a real, reachable endpoint +// (unlike the walking skeleton's `https://inference.invalid` +// placeholder), so a run started against it actually resolves its +// inference call — against `noop-inference`'s constant, locally +// served reply, never a real model. + +import { afterAll, describe, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { resetSchema, setupDatabase } from "../db-setup.ts"; +import { + HEARTBEAT_STEP_ID, + buildHeartbeatWorkflow, + serializeHeartbeatWorkflow, +} from "../../workflows/heartbeat/src/index.ts"; +import { + api, + e2eDatabaseUrl, + expectStatus, + expectStepCompleted, + freePort, + hop, + provisionSidecar, + pushWorkflowJson, + startHub, + startSidecar, + waitForRunCompletion, + type ApiResult, + type HubHandle, + type SpawnedApp, +} from "./harness.ts"; + +const databaseUrl = e2eDatabaseUrl(); +if (databaseUrl === undefined) { + console.warn( + "heartbeat: DATABASE_URL is not set; suite skipped. " + + "Set DATABASE_URL (see .env.example) to run it; " + + "CI sets E2E_REQUIRED=1 so this skip can never pass silently there.", + ); +} + +function stringField(data: unknown, field: string, what: string): string { + if (typeof data === "object" && data !== null && field in data) { + const value = (data as Record)[field]; + if (typeof value === "string" && value !== "") return value; + } + throw new Error( + `${what}: missing string field "${field}": ${JSON.stringify(data)}`, + ); +} + +function runIds(data: unknown): string[] { + if ( + typeof data === "object" && + data !== null && + "runIds" in data && + Array.isArray((data as Record)["runIds"]) + ) { + return (data as { runIds: unknown[] }).runIds.filter( + (id): id is string => typeof id === "string", + ); + } + throw new Error(`expected a runIds array: ${JSON.stringify(data)}`); +} + +const cleanups: (() => Promise)[] = []; + +afterAll(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function tempDir(prefix: string): Promise { + const dir = await mkdtemp(path.join(tmpdir(), prefix)); + cleanups.push(() => rm(dir, { recursive: true, force: true })); + return dir; +} + +function track(app: SpawnedApp): void { + cleanups.push(() => app.stop()); +} + +describe.skipIf(databaseUrl === undefined)("heartbeat workflow", () => { + test("launching heartbeat against the hub's own noop-inference endpoint starts a run", async () => { + const url = databaseUrl; + if (url === undefined) throw new Error("unreachable: suite is skipped"); + + await hop("database setup", async () => { + await resetSchema(url); + await setupDatabase(url); + }); + + const sidecarId = "sidecar-e2e-heartbeat"; + const sidecarToken = crypto.randomUUID(); + await hop("sidecar provisioning", () => + provisionSidecar(url, sidecarId, sidecarToken), + ); + + const hub: HubHandle = await hop("hub boot", async () => { + const handle = await startHub({ + databaseUrl: url, + port: freePort(), + sessionSecret: Buffer.from( + crypto.getRandomValues(new Uint8Array(32)), + ).toString("hex"), + dataDir: await tempDir("e2e-heartbeat-hub-data-"), + }); + track(handle); + return handle; + }); + + const sidecar = await hop("sidecar boot", async () => { + const app = startSidecar({ + hubPort: new URL(hub.baseUrl).port + ? Number(new URL(hub.baseUrl).port) + : 80, + sidecarId, + token: sidecarToken, + dataDir: await tempDir("e2e-heartbeat-sidecar-data-"), + }); + track(app); + return app; + }); + + const user = await hop("sign-up", async () => { + const res = await api(hub.baseUrl, "POST", "/api/auth/sign-up/email", { + name: "Heartbeat Tester", + email: `heartbeat-${crypto.randomUUID()}@example.invalid`, + password: `pw-${crypto.randomUUID()}`, + }); + expectStatus("sign-up", res, 200); + if (res.cookies.length === 0) { + throw new Error("sign-up returned no session cookie"); + } + return res; + }); + + const slug = `e2ehb${crypto.randomUUID().slice(0, 8)}`; + const tenantId = await hop("tenant creation", async () => { + const res = await api( + hub.baseUrl, + "POST", + "/api/tenants", + { name: "Heartbeat Smoke", slug }, + user.cookies, + ); + expectStatus("create tenant", res, 201); + return stringField(res.data, "id", "create tenant"); + }); + + const assetName = "heartbeat"; + const assetId = await hop("workflow asset publication", async () => { + const created = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/assets`, + { kind: "workflow", name: assetName }, + user.cookies, + ); + expectStatus("create workflow asset", created, 201); + const id = stringField(created.data, "id", "create workflow asset"); + + const minted = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/git-tokens`, + { + name: "e2e-heartbeat-push", + resource: "asset:*", + refPattern: "**", + actions: ["can_read", "can_push"], + expiresAt: new Date(Date.now() + 10 * 60_000).toISOString(), + }, + user.cookies, + ); + expectStatus("mint git token", minted, 201); + + const definition = buildHeartbeatWorkflow({ + triggerAddress: `heartbeat@${slug}.localhost`, + inferencePreferences: [{ provider: "anthropic", model: "noop" }], + turnTimeoutMs: 30_000, + }); + await pushWorkflowJson({ + baseUrl: hub.baseUrl, + tenantId, + assetName, + tokenSecret: stringField(minted.data, "secret", "mint git token"), + workflowJson: serializeHeartbeatWorkflow(definition), + }); + return id; + }); + + // The deploy's source is the hub's own, really-reachable + // noop-inference endpoint — not a placeholder like the walking + // skeleton's `https://inference.invalid`. That distinction is the + // whole point of this suite: a run started against this source + // actually completes an inference call, at zero cost, because + // noop-inference answers it locally without reaching a real model. + const deploymentId = await hop("workflow deploy", async () => { + const sourceId = "src-heartbeat-e2e"; + const body = { + assetId, + sources: [ + { + id: sourceId, + provider: "anthropic", + baseURL: `${hub.baseUrl}/api/chat/noop-inference`, + apiKey: "noop", + model: "noop", + }, + ], + defaultSource: sourceId, + }; + const deadline = Date.now() + 60_000; + let res: ApiResult; + for (;;) { + if (sidecar.exited()) { + throw new Error( + `sidecar exited before deploy; output:\n${sidecar.output()}`, + ); + } + res = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/workflows/instances`, + body, + user.cookies, + ); + if (res.status !== 502) break; + if (Date.now() > deadline) { + throw new Error( + `sidecar never became deployable (hub kept answering 502): ` + + `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, + ); + } + await Bun.sleep(1000); + } + expectStatus("deploy heartbeat workflow", res, 201); + return stringField(res.data, "id", "deploy heartbeat workflow"); + }); + + const startedRunId = await hop( + "heartbeat run starts against noop-inference", + async () => { + const before = new Set( + runIds( + ( + await api( + hub.baseUrl, + "GET", + `/api/tenants/${tenantId}/workflows/${deploymentId}/runs`, + undefined, + user.cookies, + ) + ).data, + ), + ); + + const triggered = await api( + hub.baseUrl, + "POST", + `/api/tenants/${tenantId}/workflows/${deploymentId}/mail`, + { content: "heartbeat" }, + user.cookies, + ); + expectStatus("trigger heartbeat mail", triggered, 202); + + const deadline = Date.now() + 30_000; + for (;;) { + const listed = await api( + hub.baseUrl, + "GET", + `/api/tenants/${tenantId}/workflows/${deploymentId}/runs`, + undefined, + user.cookies, + ); + const started = runIds(listed.data).find((id) => !before.has(id)); + if (started !== undefined) return started; + if (Date.now() > deadline) { + throw new Error( + "heartbeat trigger was accepted but no run started within 30s", + ); + } + await Bun.sleep(500); + } + }, + ); + + // The real gate: a run id proves only that the mail route accepted + // the trigger. Whether the deployment actually resolves — the + // step's agent launching, its turn completing against + // noop-inference, the run reaching a terminal state — is only + // proven by the run's own event log. A broken agent launch or a + // rejected inference call surfaces here as RunFailed (or no + // terminal event at all), failing this loudly instead of a + // "started" run standing in for a working platform. + const events = await hop("heartbeat run completes", () => + waitForRunCompletion( + hub.baseUrl, + tenantId, + deploymentId, + startedRunId, + user.cookies, + 30_000, + ), + ); + expectStepCompleted(events, HEARTBEAT_STEP_ID); + + console.log( + "heartbeat: gate achieved: a run completed against the real, " + + "reachable noop-inference source, proving the deployment " + + "resolves at zero cost.", + ); + }, 180_000); +}); diff --git a/vendor/intx/inference/VENDORED-FROM b/vendor/intx/inference/VENDORED-FROM index ed76302d5..b25e95e52 100644 --- a/vendor/intx/inference/VENDORED-FROM +++ b/vendor/intx/inference/VENDORED-FROM @@ -1,4 +1,4 @@ Source: https://github.com/faremeter/interchange (packages/inference) Commit: cd7144eb4d5644f8e53680f56a664102355ade29 License: LGPL-2.1-only (see vendor/intx/LICENSE) -Local modifications: exports map repointed from the upstream intx-src condition to direct TypeScript source resolution (types/default -> ./src/...); dist references removed. +Local modifications: exports map repointed from the upstream intx-src condition to direct TypeScript source resolution (types/default -> ./src/...); dist references removed; default-director empty model turn emits reply("") instead of wait() so agent.send settles for workflow steps (channel hosts already filter empty connector.reply content). diff --git a/vendor/intx/inference/src/default-director.ts b/vendor/intx/inference/src/default-director.ts index e9c2dbd57..b511aab74 100644 --- a/vendor/intx/inference/src/default-director.ts +++ b/vendor/intx/inference/src/default-director.ts @@ -287,18 +287,20 @@ export class DefaultDirector implements ReactorDirector { } // Conversational agent: send reply via the connector. + // + // Empty text still goes through `reply("")` rather than `wait()`. + // `agent.send` only settles on `connector.reply` / gate / fatal + // error; a bare `wait()` after an empty model turn leaves the + // send promise hanging until an external timeout aborts it. + // Workflow step invokers (and any other send()-driven caller) + // need the turn to complete. Channel hosts that observe + // connector.reply already filter empty content before posting, + // so an empty reply is a no-op on the mailbox side. const replyContent = extractTextContent(event.turn); - if (replyContent.length > 0) { - return [ - capabilities.checkpoint("inference-done"), - capabilities.reply(replyContent), - ]; - } - - // Empty response (no text, no tool calls) — checkpoint and wait for - // the next inbound message. The reactor only shuts down on explicit - // stop (abort), never because the model produced an empty turn. - return [capabilities.checkpoint("inference-done"), capabilities.wait()]; + return [ + capabilities.checkpoint("inference-done"), + capabilities.reply(replyContent), + ]; } case "resume.execute_tools": { diff --git a/workflows/channel-digest/README.md b/workflows/channel-digest/README.md new file mode 100644 index 000000000..dbfaccdbc --- /dev/null +++ b/workflows/channel-digest/README.md @@ -0,0 +1,55 @@ +# @corbits/channel-digest-workflow + +A single mail-triggered step meant to be deployed against a channel's +own timeline address. It relays whatever deterministic summary line its +trigger mail already carries — a message count, a timestamp, any line a +scheduler computed ahead of time — straight back into the channel, +mirroring how a channel host's reply becomes a channel mail post (see +`packages/chat/src/channel-workflow.ts` and `platform-adapter.ts`). + +## What it does + +One step, one agent, a system prompt that instructs relaying the +trigger's exact text with no additions, no commentary, no formatting of +its own. The deterministic content — the actual digest line — is +computed by whatever sends the trigger mail; this definition never +computes it itself, so its output is exactly as deterministic as its +input. + +## Cost profile + +**Pinned to `noop-inference` (default, zero cost):** deployed the same +way as `@corbits/heartbeat-workflow` — `inferencePreferences` pointed at +the hub's `noop-inference` endpoint (see `NOOP_MODEL_SOURCE` in +`packages/hub-client/src/seed.ts`). Every run resolves against a +constant, locally served SSE response, so running this on a tight +schedule costs nothing. The trade-off: `noop-inference` always replies +with empty text by design (see its header comment), so under this pin +no visible digest line is actually posted — the run still proves the +scheduling and channel-mail-posting paths stay alive, just without +visible output. + +**Pinned to a real catalog model:** point `inferencePreferences` at any +configured model instead, and the relayed digest line is posted for +real, at that model's ordinary per-turn cost (a single short completion +per trigger — a few dozen tokens, not a real "reasoning" turn). + +## Usage + +```ts +import { + buildChannelDigestWorkflow, + serializeChannelDigestWorkflow, +} from "@corbits/channel-digest-workflow"; + +const definition = buildChannelDigestWorkflow({ + triggerAddress: "channel-digest@tenant.example", + inferencePreferences: [{ provider: "anthropic", model: "noop" }], + turnTimeoutMs: 60_000, +}); + +const json = serializeChannelDigestWorkflow(definition); +``` + +Seeded by default for every tenant — see `DEFAULT_WORKFLOWS` in +`packages/hub-client/src/seed.ts`. diff --git a/workflows/channel-digest/package.json b/workflows/channel-digest/package.json new file mode 100644 index 000000000..5c883f624 --- /dev/null +++ b/workflows/channel-digest/package.json @@ -0,0 +1,23 @@ +{ + "name": "@corbits/channel-digest-workflow", + "private": true, + "description": "Minimal mail-triggered workflow that posts a deterministic summary line, exercising the channel-mail path at zero inference cost", + "version": "0.0.1", + "license": "SEE LICENSE IN LICENSE.md", + "type": "module", + "exports": { + ".": "./src/index.ts" + }, + "scripts": { + "typecheck": "tsc --noEmit", + "test": "bun test" + }, + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/workflow": "workspace:*" + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:" + } +} diff --git a/workflows/channel-digest/src/index.ts b/workflows/channel-digest/src/index.ts new file mode 100644 index 000000000..bbc23c5a2 --- /dev/null +++ b/workflows/channel-digest/src/index.ts @@ -0,0 +1,146 @@ +// The channel-digest workflow: a single-step, mail-triggered definition +// meant to be deployed against a channel's own timeline address so its +// reply posts a deterministic summary line — the trigger's own message +// count and timestamp — back into the channel, the same way a channel +// host's reply becomes a channel mail post (see +// `packages/chat/src/channel-workflow.ts` and `platform-adapter.ts`'s +// `connectorReplyContent` handling). The digest line itself is +// computed by the caller that sends the trigger mail (a scheduler, in +// the common case) and carried in the trigger body; this definition's +// only job is to relay that already-deterministic line back out +// verbatim, so nothing about the reply's content is left to the model. +// +// Zero inference cost: like `@corbits/heartbeat-workflow`, this +// definition is deployed with its `inferencePreferences` pinned to the +// hub's `noop-inference` endpoint (see +// `packages/chat/src/noop-inference.ts` and +// `packages/hub-client/src/seed.ts`'s `NOOP_MODEL_SOURCE`). Under that +// pin the turn completes instantly against a constant, empty reply — +// by design, `noop-inference` never produces real text (see that +// file's header comment) — so this deployment proves the scheduling +// and channel-mail-posting paths stay alive at zero cost, without +// posting visible digest text. Pin `inferencePreferences` at a real +// catalog model instead to get an actual, human-visible digest line +// posted on every trigger, at that model's ordinary per-turn cost. + +import { defineAgent } from "@intx/agent"; +import type { InferencePreference } from "@intx/agent"; +import { defineWorkflow, step } from "@intx/workflow"; +import type { WorkflowDefinition } from "@intx/workflow"; + +export const CHANNEL_DIGEST_WORKFLOW_ID = "wf_channel_digest"; +export const CHANNEL_DIGEST_STEP_ID = "channel-digest"; + +export const CHANNEL_DIGEST_SYSTEM_PROMPT = + "You post a single deterministic summary line into a channel. The " + + "message you receive is already the exact summary line to post — " + + "reply with its exact text: nothing added, nothing removed, no " + + "commentary, no formatting of your own."; + +/** + * Everything the definition needs that is per-deployment data. The + * trigger address names a specific deployment's inbox — for this + * workflow, ordinarily a channel's own mail address — so a definition + * built here is per-deployment by construction. + */ +export interface ChannelDigestWorkflowInput { + /** The deployment's mail address; each inbound mail is one run. */ + readonly triggerAddress: string; + /** Provider/model preferences, in order; resolved at deploy time. */ + readonly inferencePreferences: readonly InferencePreference[]; + /** Per-turn timeout in milliseconds, enforced on the single step. */ + readonly turnTimeoutMs: number; +} + +/** + * Builds the channel-digest definition. Exactly one step, matching the + * shape every other definition in this repo commits to. + * + * The step always sets an explicit `timeout` — the singular `agent:` + * shorthand sets none, and a wedged inference call would then hang a + * run forever. Tools are never inlined on the definition: they arrive + * as packages on the deploy, keeping the definition pure data. + */ +export function buildChannelDigestWorkflow( + input: ChannelDigestWorkflowInput, +): WorkflowDefinition { + if (input.triggerAddress === "") { + throw new Error( + "buildChannelDigestWorkflow requires a non-empty triggerAddress", + ); + } + if (!Number.isInteger(input.turnTimeoutMs) || input.turnTimeoutMs <= 0) { + throw new Error( + "buildChannelDigestWorkflow requires turnTimeoutMs to be a positive integer", + ); + } + return defineWorkflow({ + id: CHANNEL_DIGEST_WORKFLOW_ID, + trigger: { type: "mail", to: input.triggerAddress }, + steps: { + "channel-digest": step({ + agent: defineAgent({ + id: CHANNEL_DIGEST_STEP_ID, + description: + "Relays an already-computed deterministic summary line " + + "back into the channel it is deployed against", + systemPrompt: CHANNEL_DIGEST_SYSTEM_PROMPT, + tools: [], + capabilities: [], + inference: { sources: input.inferencePreferences }, + }), + timeout: input.turnTimeoutMs, + }), + }, + }); +} + +/** + * Serializes a definition to the JSON a workflow asset carries. The + * definition must survive the asset round-trip byte-faithfully, so + * anything JSON would silently drop or mangle — functions, undefined, + * symbols, bigints, non-finite numbers, class instances — is a loud + * error naming the offending path instead of a corrupted asset. + */ +export function serializeChannelDigestWorkflow( + definition: WorkflowDefinition, +): string { + assertJsonPortable(definition, "definition"); + return JSON.stringify(definition); +} + +function assertJsonPortable(value: unknown, path: string): void { + if (value === null) return; + switch (typeof value) { + case "string": + case "boolean": + return; + case "number": + if (!Number.isFinite(value)) { + throw new Error(`${path} is a non-finite number; JSON drops it`); + } + return; + case "object": + break; + default: + throw new Error( + `${path} is a ${typeof value}, which does not survive JSON ` + + "serialization", + ); + } + if (Array.isArray(value)) { + value.forEach((element, index) => { + assertJsonPortable(element, `${path}[${index}]`); + }); + return; + } + const proto: unknown = Object.getPrototypeOf(value); + if (proto !== Object.prototype && proto !== null) { + throw new Error( + `${path} is a non-plain object; JSON would flatten it lossily`, + ); + } + for (const [key, entry] of Object.entries(value)) { + assertJsonPortable(entry, `${path}.${key}`); + } +} diff --git a/workflows/channel-digest/test/boundary.test.ts b/workflows/channel-digest/test/boundary.test.ts new file mode 100644 index 000000000..4b14b084f --- /dev/null +++ b/workflows/channel-digest/test/boundary.test.ts @@ -0,0 +1,51 @@ +// This package is installable data on the native workflow contract: +// its shipped sources import only published platform packages, so the +// source tree plus the package manifest deploys on any Interchange +// instance without a wrapper contract. + +import { readFile, readdir } from "node:fs/promises"; +import path from "node:path"; +import { expect, test } from "bun:test"; + +const ALLOWED_IMPORT_PREFIXES = ["@intx/", "./", "../"]; + +async function listFiles(dir: string): Promise { + const entries = await readdir(dir, { withFileTypes: true }); + const files: string[] = []; + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + files.push(...(await listFiles(full))); + } else { + files.push(full); + } + } + return files; +} + +async function shippedFiles(): Promise { + const packageRoot = path.join(import.meta.dir, ".."); + return [ + ...(await listFiles(path.join(packageRoot, "src"))), + path.join(packageRoot, "package.json"), + ]; +} + +test("shipped sources import only published platform packages", async () => { + const importPattern = /from\s+"([^"]+)"/g; + const violations: string[] = []; + for (const file of await shippedFiles()) { + if (!file.endsWith(".ts")) continue; + const content = await readFile(file, "utf8"); + for (const match of content.matchAll(importPattern)) { + const specifier = match[1] ?? ""; + const allowed = ALLOWED_IMPORT_PREFIXES.some((prefix) => + specifier.startsWith(prefix), + ); + if (!allowed) { + violations.push(`${path.basename(file)}: ${specifier}`); + } + } + } + expect(violations).toEqual([]); +}); diff --git a/workflows/channel-digest/test/definition.test.ts b/workflows/channel-digest/test/definition.test.ts new file mode 100644 index 000000000..9709dfdbc --- /dev/null +++ b/workflows/channel-digest/test/definition.test.ts @@ -0,0 +1,107 @@ +// Tests for this package's own contract: the shape our factory +// commits to, its serialization guarantees, and its boundary. The +// platform's own normalization and validation are its business, not +// re-proven here. + +import { expect, test } from "bun:test"; +import type { StepPrimitive, WorkflowDefinition } from "@intx/workflow"; + +import { + CHANNEL_DIGEST_STEP_ID, + CHANNEL_DIGEST_SYSTEM_PROMPT, + CHANNEL_DIGEST_WORKFLOW_ID, + buildChannelDigestWorkflow, + serializeChannelDigestWorkflow, +} from "../src/index"; + +const INPUT = { + triggerAddress: "ch_dep000000000000@example.test", + inferencePreferences: [{ provider: "anthropic", model: "claude-test" }], + turnTimeoutMs: 60000, +} as const; + +function digestStep(definition: WorkflowDefinition): StepPrimitive { + const primitive = definition.steps[CHANNEL_DIGEST_STEP_ID]; + if (primitive === undefined || primitive.kind !== "step") { + throw new Error( + `definition has no step primitive named ${CHANNEL_DIGEST_STEP_ID}`, + ); + } + return primitive; +} + +test("the definition has exactly one step", () => { + const definition = buildChannelDigestWorkflow(INPUT); + expect(definition.stepOrder).toEqual([CHANNEL_DIGEST_STEP_ID]); + expect(Object.keys(definition.steps)).toEqual([CHANNEL_DIGEST_STEP_ID]); +}); + +test("the step carries an explicit per-turn timeout", () => { + const definition = buildChannelDigestWorkflow(INPUT); + expect(digestStep(definition).timeout).toBe(INPUT.turnTimeoutMs); +}); + +test("the workflow is triggered by mail to the given deployment address", () => { + const definition = buildChannelDigestWorkflow(INPUT); + expect(definition.id).toBe(CHANNEL_DIGEST_WORKFLOW_ID); + expect(definition.triggers).toEqual([ + { type: "mail", to: INPUT.triggerAddress }, + ]); +}); + +test("the agent instructs relaying the exact summary line, carries the preferences, and inlines no tools", () => { + const agent = digestStep(buildChannelDigestWorkflow(INPUT)).agent; + expect(agent.systemPrompt).toBe(CHANNEL_DIGEST_SYSTEM_PROMPT); + expect(agent.inference.sources).toEqual([...INPUT.inferencePreferences]); + // Tools arrive as packages on the deploy, never inlined here: an + // inline factory is a function-valued field the asset cannot carry. + expect(agent.toolFactories).toEqual([]); +}); + +test("the definition survives the workflow-asset JSON round-trip", () => { + const definition = buildChannelDigestWorkflow(INPUT); + const revived: unknown = JSON.parse( + serializeChannelDigestWorkflow(definition), + ); + expect(revived).toEqual(definition); +}); + +test("serialization fails loud on a function-valued field, naming its path", () => { + const poisoned = { + id: CHANNEL_DIGEST_WORKFLOW_ID, + triggers: [{ type: "manual" }], + stepOrder: [CHANNEL_DIGEST_STEP_ID], + steps: { + "channel-digest": { + kind: "step", + id: CHANNEL_DIGEST_STEP_ID, + drainBehavior: "cancel", + agent: { + id: CHANNEL_DIGEST_STEP_ID, + systemPrompt: CHANNEL_DIGEST_SYSTEM_PROMPT, + toolFactories: [() => []], + capabilities: [], + inference: { sources: [] }, + }, + }, + }, + } as unknown as WorkflowDefinition; + expect(() => serializeChannelDigestWorkflow(poisoned)).toThrow( + /steps\.channel-digest\.agent\.toolFactories\[0\]/, + ); +}); + +test("an empty trigger address is rejected", () => { + expect(() => + buildChannelDigestWorkflow({ ...INPUT, triggerAddress: "" }), + ).toThrow(/triggerAddress/); +}); + +test("a non-positive or fractional turn timeout is rejected", () => { + expect(() => + buildChannelDigestWorkflow({ ...INPUT, turnTimeoutMs: 0 }), + ).toThrow(/turnTimeoutMs/); + expect(() => + buildChannelDigestWorkflow({ ...INPUT, turnTimeoutMs: 0.5 }), + ).toThrow(/turnTimeoutMs/); +}); diff --git a/workflows/channel-digest/tsconfig.json b/workflows/channel-digest/tsconfig.json new file mode 100644 index 000000000..e956ddd88 --- /dev/null +++ b/workflows/channel-digest/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "types": ["bun"] + }, + "include": ["src", "test"] +} diff --git a/workflows/heartbeat/README.md b/workflows/heartbeat/README.md new file mode 100644 index 000000000..2d80a92e5 --- /dev/null +++ b/workflows/heartbeat/README.md @@ -0,0 +1,51 @@ +# @corbits/heartbeat-workflow + +The smallest workflow definition in the catalog: a single mail-triggered +step that completes immediately on every trigger. It exists to give +Interchange's scheduling and mail-trigger paths a target that can run +continuously — every few seconds, if desired — without anyone worrying +about cost. + +## What it does + +One step, one agent, a fixed system prompt that forbids drafting a real +reply. Each inbound mail to the deployment's trigger address is one run; +the run's own lifecycle (when it started, when it completed) is the +"timestamp result" — nothing about the reply text carries information. + +## Cost profile: zero + +`@intx/workflow` has no deterministic, agent-free step primitive that +the shipped hub/sidecar host can execute today — the DSL's `action` +primitive exists, but no production host wires the `invokeAction` +callback it needs, so an `action` step throws at runtime. Reaching the +DSL's only invokable primitive that performs work at all — `step` — means +going through an agent, so this definition is deployed with its +`inferencePreferences` pinned to the hub's `noop-inference` endpoint +(`packages/chat/src/noop-inference.ts`), the same trick channel-host +anchors use to avoid burning a real model call on every message. See +`NOOP_MODEL_SOURCE` in `packages/hub-client/src/seed.ts` for the pin. + +Under that pin, every run resolves against a constant, locally served +SSE response — no request ever reaches a real provider, so triggering +this workflow as often as scheduling allows costs nothing. + +## Usage + +```ts +import { + buildHeartbeatWorkflow, + serializeHeartbeatWorkflow, +} from "@corbits/heartbeat-workflow"; + +const definition = buildHeartbeatWorkflow({ + triggerAddress: "heartbeat@tenant.example", + inferencePreferences: [{ provider: "anthropic", model: "noop" }], + turnTimeoutMs: 60_000, +}); + +const json = serializeHeartbeatWorkflow(definition); +``` + +Seeded by default for every tenant — see `DEFAULT_WORKFLOWS` in +`packages/hub-client/src/seed.ts`. diff --git a/workflows/heartbeat/package.json b/workflows/heartbeat/package.json new file mode 100644 index 000000000..6ef00837c --- /dev/null +++ b/workflows/heartbeat/package.json @@ -0,0 +1,23 @@ +{ + "name": "@corbits/heartbeat-workflow", + "private": true, + "description": "Minimal mail-triggered workflow that completes immediately, exercising Interchange's scheduling path at zero inference cost", + "version": "0.0.1", + "license": "SEE LICENSE IN LICENSE.md", + "type": "module", + "exports": { + ".": "./src/index.ts" + }, + "scripts": { + "typecheck": "tsc --noEmit", + "test": "bun test" + }, + "dependencies": { + "@intx/agent": "workspace:*", + "@intx/workflow": "workspace:*" + }, + "devDependencies": { + "@types/bun": "catalog:", + "typescript": "catalog:" + } +} diff --git a/workflows/heartbeat/src/index.ts b/workflows/heartbeat/src/index.ts new file mode 100644 index 000000000..b2ad47317 --- /dev/null +++ b/workflows/heartbeat/src/index.ts @@ -0,0 +1,146 @@ +// The heartbeat workflow: the smallest possible true consumer of the +// native workflow contract, meant to run on a tight, continuous +// schedule so that the platform's scheduling and mail-trigger paths +// stay exercised in the background. It is a single-step, mail-triggered +// definition whose agent replies with a fixed, zero-content +// acknowledgement — the definition carries no per-run computation at +// all, so the run's own recorded lifecycle (its trigger time and +// completion) is the "timestamp result" this workflow exists to +// produce, not anything the agent says. +// +// Zero inference cost: the DSL has no agent-free step primitive that +// runs on the deployed host today (the `action` primitive exists in +// `@intx/workflow`, but no shipped host wires an `invokeAction` +// callback to resolve it — see VENDORED.md-adjacent research notes in +// this package's README). A deployer therefore pins this definition's +// `inferencePreferences` to the hub's `noop-inference` endpoint (see +// `packages/chat/src/noop-inference.ts` and +// `packages/hub-client/src/seed.ts`'s `NOOP_MODEL_SOURCE`): the turn +// completes instantly against a constant, never reaching a real model +// provider, so running this workflow every few seconds costs nothing. +// +// This package is installable data. It imports only published platform +// packages, and nothing imports it statically: a host publishes the +// serialized definition as a workflow asset and deploys it through the +// platform's deploy machinery; the execution host materializes it at +// runtime from the deploy alone. + +import { defineAgent } from "@intx/agent"; +import type { InferencePreference } from "@intx/agent"; +import { defineWorkflow, step } from "@intx/workflow"; +import type { WorkflowDefinition } from "@intx/workflow"; + +export const HEARTBEAT_WORKFLOW_ID = "wf_heartbeat"; +export const HEARTBEAT_STEP_ID = "heartbeat"; + +export const HEARTBEAT_SYSTEM_PROMPT = + "You are a heartbeat check. You exist only to let a run complete; " + + "never draft a reply of your own."; + +/** + * Everything the definition needs that is per-deployment data. The + * trigger address names a specific deployment's inbox, so a definition + * built here is per-deployment by construction. + */ +export interface HeartbeatWorkflowInput { + /** The deployment's mail address; each inbound mail is one run. */ + readonly triggerAddress: string; + /** Provider/model preferences, in order; resolved at deploy time. */ + readonly inferencePreferences: readonly InferencePreference[]; + /** Per-turn timeout in milliseconds, enforced on the single step. */ + readonly turnTimeoutMs: number; +} + +/** + * Builds the heartbeat definition. Exactly one step, matching the + * shape every other definition in this repo commits to; a second step + * would give a heartbeat run more to fail on for no benefit. + * + * The step always sets an explicit `timeout` — the singular `agent:` + * shorthand sets none, and a wedged inference call would then hang a + * run forever. Tools are never inlined on the definition: they arrive + * as packages on the deploy, keeping the definition pure data. + */ +export function buildHeartbeatWorkflow( + input: HeartbeatWorkflowInput, +): WorkflowDefinition { + if (input.triggerAddress === "") { + throw new Error( + "buildHeartbeatWorkflow requires a non-empty triggerAddress", + ); + } + if (!Number.isInteger(input.turnTimeoutMs) || input.turnTimeoutMs <= 0) { + throw new Error( + "buildHeartbeatWorkflow requires turnTimeoutMs to be a positive integer", + ); + } + return defineWorkflow({ + id: HEARTBEAT_WORKFLOW_ID, + trigger: { type: "mail", to: input.triggerAddress }, + steps: { + heartbeat: step({ + agent: defineAgent({ + id: HEARTBEAT_STEP_ID, + description: + "Completes immediately on every trigger, proving the " + + "scheduling and mail-trigger paths are alive", + systemPrompt: HEARTBEAT_SYSTEM_PROMPT, + tools: [], + capabilities: [], + inference: { sources: input.inferencePreferences }, + }), + timeout: input.turnTimeoutMs, + }), + }, + }); +} + +/** + * Serializes a definition to the JSON a workflow asset carries. The + * definition must survive the asset round-trip byte-faithfully, so + * anything JSON would silently drop or mangle — functions, undefined, + * symbols, bigints, non-finite numbers, class instances — is a loud + * error naming the offending path instead of a corrupted asset. + */ +export function serializeHeartbeatWorkflow( + definition: WorkflowDefinition, +): string { + assertJsonPortable(definition, "definition"); + return JSON.stringify(definition); +} + +function assertJsonPortable(value: unknown, path: string): void { + if (value === null) return; + switch (typeof value) { + case "string": + case "boolean": + return; + case "number": + if (!Number.isFinite(value)) { + throw new Error(`${path} is a non-finite number; JSON drops it`); + } + return; + case "object": + break; + default: + throw new Error( + `${path} is a ${typeof value}, which does not survive JSON ` + + "serialization", + ); + } + if (Array.isArray(value)) { + value.forEach((element, index) => { + assertJsonPortable(element, `${path}[${index}]`); + }); + return; + } + const proto: unknown = Object.getPrototypeOf(value); + if (proto !== Object.prototype && proto !== null) { + throw new Error( + `${path} is a non-plain object; JSON would flatten it lossily`, + ); + } + for (const [key, entry] of Object.entries(value)) { + assertJsonPortable(entry, `${path}.${key}`); + } +} diff --git a/workflows/heartbeat/test/boundary.test.ts b/workflows/heartbeat/test/boundary.test.ts new file mode 100644 index 000000000..4b14b084f --- /dev/null +++ b/workflows/heartbeat/test/boundary.test.ts @@ -0,0 +1,51 @@ +// This package is installable data on the native workflow contract: +// its shipped sources import only published platform packages, so the +// source tree plus the package manifest deploys on any Interchange +// instance without a wrapper contract. + +import { readFile, readdir } from "node:fs/promises"; +import path from "node:path"; +import { expect, test } from "bun:test"; + +const ALLOWED_IMPORT_PREFIXES = ["@intx/", "./", "../"]; + +async function listFiles(dir: string): Promise { + const entries = await readdir(dir, { withFileTypes: true }); + const files: string[] = []; + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + files.push(...(await listFiles(full))); + } else { + files.push(full); + } + } + return files; +} + +async function shippedFiles(): Promise { + const packageRoot = path.join(import.meta.dir, ".."); + return [ + ...(await listFiles(path.join(packageRoot, "src"))), + path.join(packageRoot, "package.json"), + ]; +} + +test("shipped sources import only published platform packages", async () => { + const importPattern = /from\s+"([^"]+)"/g; + const violations: string[] = []; + for (const file of await shippedFiles()) { + if (!file.endsWith(".ts")) continue; + const content = await readFile(file, "utf8"); + for (const match of content.matchAll(importPattern)) { + const specifier = match[1] ?? ""; + const allowed = ALLOWED_IMPORT_PREFIXES.some((prefix) => + specifier.startsWith(prefix), + ); + if (!allowed) { + violations.push(`${path.basename(file)}: ${specifier}`); + } + } + } + expect(violations).toEqual([]); +}); diff --git a/workflows/heartbeat/test/definition.test.ts b/workflows/heartbeat/test/definition.test.ts new file mode 100644 index 000000000..b17f19d82 --- /dev/null +++ b/workflows/heartbeat/test/definition.test.ts @@ -0,0 +1,105 @@ +// Tests for this package's own contract: the shape our factory +// commits to, its serialization guarantees, and its boundary. The +// platform's own normalization and validation are its business, not +// re-proven here. + +import { expect, test } from "bun:test"; +import type { StepPrimitive, WorkflowDefinition } from "@intx/workflow"; + +import { + HEARTBEAT_STEP_ID, + HEARTBEAT_SYSTEM_PROMPT, + HEARTBEAT_WORKFLOW_ID, + buildHeartbeatWorkflow, + serializeHeartbeatWorkflow, +} from "../src/index"; + +const INPUT = { + triggerAddress: "ins_dep000000000000@example.test", + inferencePreferences: [{ provider: "anthropic", model: "claude-test" }], + turnTimeoutMs: 60000, +} as const; + +function heartbeatStep(definition: WorkflowDefinition): StepPrimitive { + const primitive = definition.steps[HEARTBEAT_STEP_ID]; + if (primitive === undefined || primitive.kind !== "step") { + throw new Error( + `definition has no step primitive named ${HEARTBEAT_STEP_ID}`, + ); + } + return primitive; +} + +test("the definition has exactly one step", () => { + const definition = buildHeartbeatWorkflow(INPUT); + expect(definition.stepOrder).toEqual([HEARTBEAT_STEP_ID]); + expect(Object.keys(definition.steps)).toEqual([HEARTBEAT_STEP_ID]); +}); + +test("the step carries an explicit per-turn timeout", () => { + const definition = buildHeartbeatWorkflow(INPUT); + expect(heartbeatStep(definition).timeout).toBe(INPUT.turnTimeoutMs); +}); + +test("the workflow is triggered by mail to the given deployment address", () => { + const definition = buildHeartbeatWorkflow(INPUT); + expect(definition.id).toBe(HEARTBEAT_WORKFLOW_ID); + expect(definition.triggers).toEqual([ + { type: "mail", to: INPUT.triggerAddress }, + ]); +}); + +test("the agent carries the fixed prompt, the preferences, and inlines no tools", () => { + const agent = heartbeatStep(buildHeartbeatWorkflow(INPUT)).agent; + expect(agent.systemPrompt).toBe(HEARTBEAT_SYSTEM_PROMPT); + expect(agent.inference.sources).toEqual([...INPUT.inferencePreferences]); + // Tools arrive as packages on the deploy, never inlined here: an + // inline factory is a function-valued field the asset cannot carry. + expect(agent.toolFactories).toEqual([]); +}); + +test("the definition survives the workflow-asset JSON round-trip", () => { + const definition = buildHeartbeatWorkflow(INPUT); + const revived: unknown = JSON.parse(serializeHeartbeatWorkflow(definition)); + expect(revived).toEqual(definition); +}); + +test("serialization fails loud on a function-valued field, naming its path", () => { + const poisoned = { + id: HEARTBEAT_WORKFLOW_ID, + triggers: [{ type: "manual" }], + stepOrder: [HEARTBEAT_STEP_ID], + steps: { + heartbeat: { + kind: "step", + id: HEARTBEAT_STEP_ID, + drainBehavior: "cancel", + agent: { + id: HEARTBEAT_STEP_ID, + systemPrompt: HEARTBEAT_SYSTEM_PROMPT, + toolFactories: [() => []], + capabilities: [], + inference: { sources: [] }, + }, + }, + }, + } as unknown as WorkflowDefinition; + expect(() => serializeHeartbeatWorkflow(poisoned)).toThrow( + /steps\.heartbeat\.agent\.toolFactories\[0\]/, + ); +}); + +test("an empty trigger address is rejected", () => { + expect(() => + buildHeartbeatWorkflow({ ...INPUT, triggerAddress: "" }), + ).toThrow(/triggerAddress/); +}); + +test("a non-positive or fractional turn timeout is rejected", () => { + expect(() => buildHeartbeatWorkflow({ ...INPUT, turnTimeoutMs: 0 })).toThrow( + /turnTimeoutMs/, + ); + expect(() => + buildHeartbeatWorkflow({ ...INPUT, turnTimeoutMs: 0.5 }), + ).toThrow(/turnTimeoutMs/); +}); diff --git a/workflows/heartbeat/tsconfig.json b/workflows/heartbeat/tsconfig.json new file mode 100644 index 000000000..e956ddd88 --- /dev/null +++ b/workflows/heartbeat/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "types": ["bun"] + }, + "include": ["src", "test"] +}