Durable workflow deploy source for every placement (CL-6581 phase 1)

apps/hub/package.json

@@ -47,6 +47,7 @@
     "@corbits/tasks": "workspace:*",
     "@corbits/tool-registry-publish": "workspace:*",
     "@corbits/turn-artifacts": "workspace:*",
+    "@corbits/workflow-deploy-source": "workspace:*",
     "@corbits/url-path": "workspace:*",
     "@corbits/webhook-triggers": "workspace:*",
     "@corbits/workflow-catalog": "workspace:*",

apps/hub/src/index.ts

@@ -206,6 +206,10 @@ import {
   createRunKeyHistoryRoutes,
   lookupRunKeyHistoryReconnectKey,
 } from "@corbits/run-key-history";
+import {
+  createDrizzleWorkflowDeploySourceStore,
+  withDeploySourceRecording,
+} from "@corbits/workflow-deploy-source";
 import {
   createMyraAgentDefinitionDrafting,
   createPlannerRoutes,
@@ -870,17 +874,28 @@ export async function createHub(config: HubConfig) {
     getSigningPublicKey: agentRepoStore.getSigningPublicKey,
     repoStore: launchCaches.repoStore,
   };
-  const sessionService = createSessionService({
-    sidecarRouter,
-    agentRepoStore: launchAgentRepoStore,
-    assetService: launchCaches.assetService,
-    db,
-    toolPackageRegistries: {
-      httpRegistries: REGISTRIES,
-      defaultRegistry: "npmjs",
-      scopeRouting: [{ scope: "@corbits", registry: CORBITS_TOOLS_REGISTRY }],
-    },
-  });
+  // Shared placement's code-sourced deploys previously left their
+  // `WorkflowDefinitionSource` durable nowhere on the hub -- only on the
+  // sidecar's local `deployment.json` (CL-6581). Wrapping the two deploy
+  // methods here, at the composition root, records that source into
+  // Postgres on every deploy without touching vendored
+  // `session-service.ts`; exclusive placement already persists its own via
+  // `workflow_run_launch_spec`, untouched.
+  const workflowDeploySourceStore = createDrizzleWorkflowDeploySourceStore(db);
+  const sessionService = withDeploySourceRecording(
+    createSessionService({
+      sidecarRouter,
+      agentRepoStore: launchAgentRepoStore,
+      assetService: launchCaches.assetService,
+      db,
+      toolPackageRegistries: {
+        httpRegistries: REGISTRIES,
+        defaultRegistry: "npmjs",
+        scopeRouting: [{ scope: "@corbits", registry: CORBITS_TOOLS_REGISTRY }],
+      },
+    }),
+    workflowDeploySourceStore,
+  );
   // Provisioner plugins are injected at the application composition
   // boundary, mirroring @intx/hub-sessions's own reference wiring: the
   // registry always exists, but ships with no provisioners (and no

packages/workflow-deploy-source/package.json

@@ -0,0 +1,28 @@
+{
+  "name": "@corbits/workflow-deploy-source",
+  "private": true,
+  "description": "Durable, hub-side record of a native workflow's deploy source (WorkflowDefinitionSource: an asset commit or a registry name@range pin), for every sidecar placement — so a deployment can be recreated from Postgres alone, with nothing owed to the sidecar's local disk.",
+  "version": "0.0.1",
+  "license": "LGPL-2.1-or-later",
+  "type": "module",
+  "exports": {
+    ".": "./src/index.ts",
+    "./migrations": "./src/migrations.ts"
+  },
+  "scripts": {
+    "typecheck": "tsc --noEmit",
+    "test": "bun test"
+  },
+  "dependencies": {
+    "@intx/hub-sessions": "workspace:*",
+    "@intx/log": "0.3.0",
+    "@intx/types": "0.3.0",
+    "arktype": "catalog:",
+    "drizzle-orm": "catalog:",
+    "postgres": "catalog:"
+  },
+  "devDependencies": {
+    "@types/bun": "catalog:",
+    "typescript": "catalog:"
+  }
+}

packages/workflow-deploy-source/src/index.ts

@@ -0,0 +1,12 @@
+export {
+  createDrizzleWorkflowDeploySourceStore,
+  type WorkflowDeploySourceDb,
+  type WorkflowDeploySourceRecord,
+  type WorkflowDeploySourceStore,
+} from "./store";
+export { workflowDeploySource, workflowDeploySourceSchema } from "./schema";
+export type { WorkflowDeploySourceRow } from "./schema";
+export {
+  withDeploySourceRecording,
+  type DeployWorkflowDeployer,
+} from "./record-on-deploy";

packages/workflow-deploy-source/src/migrations.ts

@@ -0,0 +1,99 @@
+// Package-owned migrations for @corbits/workflow-deploy-source.
+// Bookkeeping uses its own ledger table so the package can be extracted
+// without disentangling history from the platform drizzle journal. The
+// table this package owns lives in its own `workflow_deploy_source`
+// Postgres schema, never `public`; see docs/package-migrations.md.
+import postgres from "postgres";
+
+export interface WorkflowDeploySourceMigration {
+  name: string;
+  sql: string;
+}
+
+const SCHEMA = "workflow_deploy_source";
+
+export const workflowDeploySourceMigrations: readonly WorkflowDeploySourceMigration[] =
+  [
+    {
+      name: "0001_workflow_deploy_source",
+      sql: `
+      CREATE TABLE IF NOT EXISTS "workflow_deploy_source"."workflow_deploy_source" (
+        "anchor_run_id" text PRIMARY KEY,
+        "tenant_id" text NOT NULL,
+        "deployment_domain" text NOT NULL,
+        "source" jsonb NOT NULL,
+        "entry" text NOT NULL,
+        "pin" text,
+        "definition_asset_id" text NOT NULL,
+        "source_ref" text,
+        "source_authority_principal_id" text NOT NULL,
+        "recorded_at" timestamptz NOT NULL DEFAULT now()
+      );
+      CREATE INDEX IF NOT EXISTS "workflow_deploy_source_tenant_idx"
+        ON "workflow_deploy_source"."workflow_deploy_source" ("tenant_id");
+    `,
+    },
+  ];
+
+function quoteIdentifier(name: string): string {
+  return `"${name.replace(/"/g, '""')}"`;
+}
+
+function quoteQualified(schema: string, name: string): string {
+  return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`;
+}
+
+const LEDGER_TABLE = "workflow_deploy_source_migrations";
+
+export interface ApplyWorkflowDeploySourceMigrationsReport {
+  applied: string[];
+  alreadyApplied: string[];
+}
+
+export async function applyWorkflowDeploySourceMigrations(
+  databaseUrl: string,
+): Promise<ApplyWorkflowDeploySourceMigrationsReport> {
+  const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined });
+  try {
+    await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`);
+
+    await sql.unsafe(
+      `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` +
+        `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`,
+    );
+
+    const applied: string[] = [];
+    const alreadyApplied: string[] = [];
+
+    for (const migration of workflowDeploySourceMigrations) {
+      const existing = await sql.unsafe(
+        `SELECT 1 FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)} WHERE name = $1`,
+        [migration.name],
+      );
+      if (existing.length > 0) {
+        alreadyApplied.push(migration.name);
+        continue;
+      }
+      try {
+        await sql.begin(async (tx) => {
+          await tx.unsafe(migration.sql);
+          await tx.unsafe(
+            `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`,
+            [migration.name],
+          );
+        });
+        applied.push(migration.name);
+      } catch (err) {
+        const message = err instanceof Error ? err.message : String(err);
+        throw new Error(
+          `workflow_deploy_source migration ${migration.name} failed: ${message}`,
+          { cause: err },
+        );
+      }
+    }
+
+    return { applied, alreadyApplied };
+  } finally {
+    await sql.end({ timeout: 5 });
+  }
+}

packages/workflow-deploy-source/src/record-on-deploy.ts

@@ -0,0 +1,88 @@
+// The write-side seam: wraps a `SessionService & AdoptingWorkflowDeployer`
+// so every code-sourced workflow deploy on SHARED capacity -- the only
+// placement whose source previously existed nowhere in Postgres -- durably
+// records its `WorkflowDefinitionSource` before returning to the caller.
+//
+// This is a workbench-side decorator around `@intx/hub-sessions`'
+// `createSessionService(...)` output, composed at the app root
+// (apps/hub/src/index.ts) exactly where `createLaunchCaches` already wraps
+// `repoStore`/`assetService` the same way -- never a change to
+// vendor/intx/hub-sessions/src/session-service.ts, whose two deploy methods
+// this only wraps, not reimplements. Exclusive placement is unaffected: it
+// already persists its source durably via `workflow_run_launch_spec`
+// (vendor/intx/db/src/schema/workflow-run-launch-spec.ts), written from
+// vendored `workflow-allocation-service.ts`, which this package does not
+// touch.
+import { getLogger } from "@intx/log";
+import type {
+  AdoptingWorkflowDeployer,
+  DeployAdoptedWorkflowFromSourceParams,
+  DeployWorkflowFromSourceParams,
+  SessionService,
+} from "@intx/hub-sessions";
+
+import type { WorkflowDeploySourceStore } from "./store";
+
+const logger = getLogger(["workflow-deploy-source", "record-on-deploy"]);
+
+export type DeployWorkflowDeployer = SessionService & AdoptingWorkflowDeployer;
+
+function recordFromDeployParams(
+  params:
+    DeployWorkflowFromSourceParams | DeployAdoptedWorkflowFromSourceParams,
+) {
+  return {
+    anchorRunId: params.anchorRunId,
+    tenantId: params.tenantId,
+    deploymentDomain: params.deploymentDomain,
+    source: params.source,
+    entry: params.entry,
+    definitionAssetId: params.definitionAssetId,
+    sourceAuthorityPrincipalId: params.config.principalId,
+    ...(params.pin !== undefined ? { pin: params.pin } : {}),
+    ...(params.sourceRef !== undefined ? { sourceRef: params.sourceRef } : {}),
+  };
+}
+
+/**
+ * Wrap a session service so `deployWorkflowFromSource` and
+ * `deployAdoptedWorkflowFromSource` -- the shared-capacity code-sourced
+ * deploy entry points `POST /workflows/deployments` and the routine
+ * launcher's adopted-anchor deploy drive -- record their `source` durably
+ * AFTER the deploy itself succeeds. A recording failure is reported (never
+ * a bare catch) but does not fail the deploy: the sidecar agent is already
+ * live by the time this runs, so surfacing the recording failure as a
+ * deploy failure would strand a live deployment behind a 500.
+ */
+export function withDeploySourceRecording<T extends DeployWorkflowDeployer>(
+  sessionService: T,
+  store: WorkflowDeploySourceStore,
+): T {
+  return {
+    ...sessionService,
+    async deployWorkflowFromSource(params) {
+      const result = await sessionService.deployWorkflowFromSource(params);
+      await recordOrLog(store, recordFromDeployParams(params));
+      return result;
+    },
+    async deployAdoptedWorkflowFromSource(params) {
+      const result =
+        await sessionService.deployAdoptedWorkflowFromSource(params);
+      await recordOrLog(store, recordFromDeployParams(params));
+      return result;
+    },
+  };
+}
+
+async function recordOrLog(
+  store: WorkflowDeploySourceStore,
+  entry: ReturnType<typeof recordFromDeployParams>,
+): Promise<void> {
+  try {
+    await store.record(entry);
+  } catch (cause) {
+    logger.error`failed to record deploy source for anchor run ${entry.anchorRunId}: ${
+      cause instanceof Error ? cause.message : String(cause)
+    }`;
+  }
+}