Routines target a workflow asset, resolved at use (CL-7350)

apps/hub/src/index.ts

@@ -196,6 +196,7 @@ import {
   createMyraRoutineDrafting,
   createRoutineRoutes,
   createWorkflowRoutineRoutes,
+  resolveLaunchableDefinition,
   routine as routineTable,
   routineRun as routineRunTable,
   type RoutineDraftInventoryWorkflow,
@@ -2768,10 +2769,11 @@ export async function createHub(config: HubConfig) {
     const out: RoutineDraftInventoryWorkflow[] = [];
     for (const row of rows) {
       if (!isAutomatableWorkflowName(row.name)) continue;
+      if (row.assetId === null) continue;
       const entry = workflowCatalogEntry(row.name);
       if (entry === undefined) continue;
       const workflow = {
-        definitionId: row.id,
+        definitionAssetId: row.assetId,
         assetName: row.name,
         displayName: workflowDisplayName(row.name, row.description),
         deliveryMode: entry.deliveryMode,
@@ -2807,6 +2809,28 @@ export async function createHub(config: HubConfig) {
     joinDeliveryWorkbench: (input) =>
       joinRunParticipant({ store: chatStore }, input),
   });
+  // A `{kind: "webhook"}` routine trigger and the `@corbits/webhook-
+  // triggers` row it names are two views of one binding: the row still
+  // points at a `workflow_definition` row id (`workflowDefinitionId`),
+  // while a routine now names its stable `definitionAssetId` — so
+  // agreement means the definition row's own `assetId` equals the
+  // routine's `definitionAssetId`, not a direct id comparison.
+  const webhookTriggerInTenant = async (
+    tenantId: string,
+    webhookTriggerId: string,
+    definitionAssetId: string,
+  ): Promise<boolean> => {
+    const trigger = await webhookTriggerStore.get(tenantId, webhookTriggerId);
+    if (trigger === undefined) return false;
+    const definitionRow = await db.query.workflowDefinition.findFirst({
+      where: and(
+        eq(workflowDefinition.id, trigger.workflowDefinitionId),
+        eq(workflowDefinition.tenantId, tenantId),
+      ),
+      columns: { assetId: true },
+    });
+    return definitionRow?.assetId === definitionAssetId;
+  };
   const routineWorkbenchNotice = {
     postWorkbenchNotice: (input: {
       tenantId: string;
@@ -2881,29 +2905,14 @@ export async function createHub(config: HubConfig) {
       // thread; see `@corbits/routines`' `RoutineLauncher` doc comment
       // for the multi-message contract.
       runSummaryResolver: createHubRunSummaryResolver(db),
-      definitionInTenant: async (tenantId, definitionId) => {
-        const row = await db.query.workflowDefinition.findFirst({
-          where: and(
-            eq(workflowDefinition.id, definitionId),
-            eq(workflowDefinition.tenantId, tenantId),
-          ),
-          columns: { id: true },
-        });
-        return row !== undefined;
-      },
+      resolveTarget: (tenantId, definitionAssetId) =>
+        resolveLaunchableDefinition({ db, tenantId, definitionAssetId }),
       // A `{kind: "webhook"}` trigger's `webhookTriggerId` must resolve
       // to a real `webhook_trigger` row in this tenant, pointed at the
       // exact same workflow definition the routine itself runs — see
       // `webhookTriggerValid`'s doc comment in
       // `@corbits/routines`' routes.ts for why the two ids must agree.
-      webhookTriggerInTenant: async (
-        tenantId,
-        webhookTriggerId,
-        definitionId,
-      ) => {
-        const row = await webhookTriggerStore.get(tenantId, webhookTriggerId);
-        return row !== undefined && row.workflowDefinitionId === definitionId;
-      },
+      webhookTriggerInTenant,
       deliveryWorkbenchRequired: routineDeliveryWorkbenchRequired,
       validateRoutineInput: routineInputValid,
     }),
@@ -2922,50 +2931,9 @@ export async function createHub(config: HubConfig) {
       launcher: routineLauncher,
       workbenchNotice: routineWorkbenchNotice,
       authenticator: createWorkflowRunAuthenticator({ db }),
-      definitionInTenant: async (tenantId, definitionId) => {
-        const row = await db.query.workflowDefinition.findFirst({
-          where: and(
-            eq(workflowDefinition.id, definitionId),
-            eq(workflowDefinition.tenantId, tenantId),
-          ),
-          columns: { id: true },
-        });
-        return row !== undefined;
-      },
-      // Myra's `routine_create` tool receives a definition's NAME from
-      // `list_agents`, not its `wfd_` id — resolve an exact, deployed-only
-      // name match within the tenant before the `definitionInTenant`
-      // check above runs a second time against the resolved id.
-      resolveDefinitionId: async (tenantId, idOrName) => {
-        const rows = await db.query.workflowDefinition.findMany({
-          where: and(
-            eq(workflowDefinition.name, idOrName),
-            eq(workflowDefinition.tenantId, tenantId),
-            eq(workflowDefinition.status, "deployed"),
-          ),
-          columns: { id: true },
-        });
-        return rows.length === 1 ? rows[0]?.id : undefined;
-      },
-      listDefinitionCandidates: async (tenantId) => {
-        const rows = await db.query.workflowDefinition.findMany({
-          where: and(
-            eq(workflowDefinition.tenantId, tenantId),
-            eq(workflowDefinition.status, "deployed"),
-          ),
-          columns: { id: true, name: true },
-          limit: 8,
-        });
-        return rows;
-      },
-      webhookTriggerInTenant: async (
-        tenantId,
-        webhookTriggerId,
-        definitionId,
-      ) => {
-        const row = await webhookTriggerStore.get(tenantId, webhookTriggerId);
-        return row !== undefined && row.workflowDefinitionId === definitionId;
-      },
+      resolveTarget: (tenantId, definitionAssetId) =>
+        resolveLaunchableDefinition({ db, tenantId, definitionAssetId }),
+      webhookTriggerInTenant,
       deliveryWorkbenchRequired: routineDeliveryWorkbenchRequired,
       // A routine created from inside a workbench delivers into that
       // workbench: the creating run is a workbench participant, so its

apps/hub/src/routine-launcher.test.ts

@@ -94,10 +94,56 @@ const TENANT_ROW = {
   domain: "acme.workbench.test",
 };
 
+// One `workflow_definition` candidate for `resolveLaunchableDefinition`'s
+// asset-id lookup — deployed and frozen, so it resolves to `wfd_1`
+// (`DEFINITION_ROW`'s id) exactly like every existing fixture expects.
+const DEFINITION_CANDIDATE_ROW = {
+  id: DEFINITION_ROW.id,
+  tenantId: DEFINITION_ROW.tenantId,
+  status: DEFINITION_ROW.status,
+  createdAt: new Date("2024-01-01T00:00:00.000Z"),
+  approvedWireHash: "hash_1",
+  grantSnapshot: {},
+  wireProjection: {},
+};
+
+// A stand-in `.select().from()...` chain covering both drizzle queries
+// this launcher and `triggerNativeWorkflowRoutineRun` run: the native
+// anchor-run lookup (`.orderBy().limit()`) and
+// `resolveLaunchableDefinition`'s asset lookup (`.leftJoin()...orderBy()`,
+// awaited directly, no `.limit()`). A `.leftJoin()` call is what tells
+// the two apart.
+function makeSelectMock(
+  nativeRows: readonly unknown[],
+  candidateRows: readonly unknown[],
+) {
+  return () => {
+    let sawLeftJoin = false;
+    const chain = {
+      from: () => chain,
+      leftJoin: () => {
+        sawLeftJoin = true;
+        return chain;
+      },
+      where: () => chain,
+      orderBy: () => {
+        const rows = sawLeftJoin ? candidateRows : nativeRows;
+        const result = Promise.resolve(rows) as Promise<unknown> & {
+          limit: (n: number) => Promise<unknown>;
+        };
+        result.limit = async () => nativeRows;
+        return result;
+      },
+    };
+    return chain;
+  };
+}
+
 function createFakeDb(
   overrides: {
     definition?: unknown;
     tenant?: unknown;
+    candidateRows?: readonly unknown[];
   } = {},
 ) {
   return {
@@ -114,15 +160,10 @@ function createFakeDb(
     // Drives `triggerNativeWorkflowRoutineRun`'s real anchor-run
     // lookup for the multi-step tests below — see that module's own
     // test file for coverage of its query shape in isolation.
-    select: () => ({
-      from: () => ({
-        where: () => ({
-          orderBy: () => ({
-            limit: async () => [NATIVE_ANCHOR_ROW],
-          }),
-        }),
-      }),
-    }),
+    select: makeSelectMock(
+      [NATIVE_ANCHOR_ROW],
+      overrides.candidateRows ?? [DEFINITION_CANDIDATE_ROW],
+    ),
     // `recordSourcesDigest` writes the deployed inference chain's digest
     // onto the launch row once `launchFoldedRun` returns (CL-6687).
     update: () => ({
@@ -139,7 +180,7 @@ function baseInput(input: Record<string, unknown>) {
   return {
     tenantId: "ten_1",
     principalId: "usr_1",
-    definitionId: "wfd_1",
+    definitionAssetId: "ast_1",
     input,
   };
 }
@@ -413,13 +454,7 @@ describe("createHubRoutineLauncher — multi-step native routing", () => {
           workflowDefinition: { findFirst: async () => DEFINITION_ROW },
           tenant: { findFirst: async () => TENANT_ROW },
         },
-        select: () => ({
-          from: () => ({
-            where: () => ({
-              orderBy: () => ({ limit: async () => [] }),
-            }),
-          }),
-        }),
+        select: makeSelectMock([], [DEFINITION_CANDIDATE_ROW]),
       } as never,
       sessionService: {} as never,
       assetService: {} as never,

apps/hub/src/routine-launcher.ts

@@ -33,6 +33,7 @@
 import { and, eq } from "drizzle-orm";
 import type { DB } from "@intx/db";
 import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema";
+import { reportError } from "@corbits/error-sink";
 import {
   domainOf,
   launchFoldedRun,
@@ -53,7 +54,12 @@ import {
   recordSourcesDigest,
   workbenchLaunchPersistExtra,
 } from "@corbits/chat";
-import { renderRoutineInput, type RoutineLauncher } from "@corbits/routines";
+import {
+  renderRoutineInput,
+  resolveLaunchableDefinition,
+  RoutineTargetUnresolvableError,
+  type RoutineLauncher,
+} from "@corbits/routines";
 import { triggerNativeWorkflowRoutineRun } from "./native-workflow-routine-launch";
 
 const log = getLogger(["hub", "routine-launcher"]);
@@ -87,26 +93,54 @@ export function createHubRoutineLauncher(
 ): RoutineLauncher {
   return {
     async launchRoutineRun(input) {
+      const resolution = await resolveLaunchableDefinition({
+        db: deps.db,
+        tenantId: input.tenantId,
+        definitionAssetId: input.definitionAssetId,
+      });
+      if (!resolution.ok) {
+        reportError(
+          new RoutineTargetUnresolvableError(
+            input.definitionAssetId,
+            resolution.reason,
+          ),
+          {
+            operation: "routine-launcher.launchRoutineRun",
+            tenantId: input.tenantId,
+            extra: { definitionAssetId: input.definitionAssetId },
+          },
+        );
+        throw new RoutineTargetUnresolvableError(
+          input.definitionAssetId,
+          resolution.reason,
+        );
+      }
+      const definitionId = resolution.definitionId;
+
       const definitionRow = await deps.db.query.workflowDefinition.findFirst({
         where: and(
-          eq(workflowDefinition.id, input.definitionId),
+          eq(workflowDefinition.id, definitionId),
           eq(workflowDefinition.tenantId, input.tenantId),
         ),
       });
-      if (definitionRow === undefined) {
-        throw new Error(
-          `no definition "${input.definitionId}" for this tenant`,
-        );
-      }
-      if (definitionRow.status !== "deployed") {
-        throw new Error(
-          `definition "${input.definitionId}" is not in a launchable ` +
-            `state (status: ${definitionRow.status})`,
+      if (
+        definitionRow === undefined ||
+        definitionRow.status !== "deployed" ||
+        definitionRow.assetId === null
+      ) {
+        const reason =
+          definitionRow === undefined ? "not_found" : "not_deployed";
+        reportError(
+          new RoutineTargetUnresolvableError(input.definitionAssetId, reason),
+          {
+            operation: "routine-launcher.launchRoutineRun",
+            tenantId: input.tenantId,
+            extra: { definitionAssetId: input.definitionAssetId, definitionId },
+          },
         );
-      }
-      if (definitionRow.assetId === null) {
-        throw new Error(
-          `definition "${input.definitionId}" has not been materialized`,
+        throw new RoutineTargetUnresolvableError(
+          input.definitionAssetId,
+          reason,
         );
       }
 
@@ -142,7 +176,7 @@ export function createHubRoutineLauncher(
         const content = renderRoutineInput(input.input);
         const triggered = await triggerNativeWorkflowRoutineRun(deps, {
           tenantId: input.tenantId,
-          definitionId: input.definitionId,
+          definitionId,
           principalId: input.principalId,
           fromDomain: tenantRow.domain,
           // A native run only starts on its first trigger mail — unlike
@@ -185,7 +219,7 @@ export function createHubRoutineLauncher(
         tenantId: input.tenantId,
         instanceId,
         triggerAddress,
-        definitionId: input.definitionId,
+        definitionId,
         foldedBody,
         launchLabel: "a routine",
         // The same `onTrigger` section shape and stable-id → current-run

apps/hub/src/routine-scheduler.ts

@@ -41,7 +41,7 @@ export type RoutineSchedulerDeps = {
    * workbench-required-or-not rule a manual "run now" does. */
   deliveryWorkbenchRequired?: (
     tenantId: string,
-    definitionId: string,
+    definitionAssetId: string,
   ) => Promise<boolean>;
   /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */
   now?: () => Date;