feat(web): pause and resume a routine through @corbits/cron (CL-9623)

apps/hub/package.json

@@ -17,7 +17,7 @@
     "@corbits/approvals": "workspace:*",
     "@corbits/artifacts": "github:corbitsdev/corbits-artifacts#4a2a3c22187991a6ce598faf794908fedd252e98",
     "@corbits/codex-provider": "github:corbitsdev/corbits-codex-provider#a1f475f341261c05fa8dade0c82f1246c175b5b3",
-    "@corbits/cron": "github:corbitsdev/corbits-cron#872360f4bb55f140e6462208bfb0888f4a7e395a",
+    "@corbits/cron": "0.2.1",
     "@corbits/error-sink": "workspace:*",
     "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#65590a85fa143251b3ac20ba2eca92fc35e70e51",
     "@corbits/mcp": "github:corbitsdev/corbits-mcp#9df1ab81d2f50d798bca2d2f44eb4759c92fac0e",

apps/hub/src/migrate.ts

@@ -9,7 +9,7 @@ import { runMigrations } from "@intx/db";
 import { sql } from "drizzle-orm";
 import postgres from "postgres";
 import { applyAgentTokenMigrations } from "@corbits/agent-token/migrations";
-import { applyCronMigrations } from "@corbits/cron";
+import { runCronMigrations } from "@corbits/cron/migrations";
 import { createMailboxDb, runMailboxMigrations } from "@corbits/mailbox";
 import { runArtifactMigrations } from "@corbits/artifacts";
 import { runMemoryMigrations } from "@corbits/memory";
@@ -83,7 +83,7 @@ async function applyMigrations(
   if (!(await platformSchemaPresent(db, schema))) {
     await runMigrations(config, { schema });
   }
-  await applyCronMigrations(databaseUrl, { tenantSchema: schema });
+  await runCronMigrations(config, { schema });
   await applyAgentTokenMigrations(databaseUrl, { tenantSchema: schema });
 
   const mailboxDb = createMailboxDb(databaseUrl);

apps/hub/src/server.ts

@@ -86,7 +86,7 @@ import {
   refreshXaiTokens,
 } from "@corbits/xai-provider";
 import { createAgentTokenVerifier, mountAgentTokens } from "@corbits/agent-token";
-import { createCronTicker, createRunTriggerCronDeliver, mountCron } from "@corbits/cron";
+import { createCronTicker, createCronRoutes, createRunTriggerCronDeliver } from "@corbits/cron";
 import {
   createHubMailboxAuthorizeSender,
   createHubPersistMailWithSessionEnsure,
@@ -642,15 +642,13 @@ export async function createHubServer({
   let cronTicker: { start(): void; stop(): void } | undefined;
   let oauthTokenRefresher: { start(): void; stop(): void } | undefined;
   {
-    const cronApp = new Hono<TenantEnv>();
-    mountCron(cronApp, {
-      db,
-      requireTenantMember: (ctx, tenantId) => {
-        const c = ctx as { get(key: "tenant"): { id: string } };
-        return c.get("tenant").id === tenantId;
-      },
-    });
-    app.route("/", cronApp);
+    app.route(
+      `${TENANT_PREFIX}/cron`,
+      createCronRoutes({
+        db,
+        requireGrant: createRequireGrant({ grantStore, conditionRegistry: grantConditionRegistry }),
+      }),
+    );
 
     cronTicker = createCronTicker({
       db,

apps/web/src/bench/workflows-tab.tsx

@@ -44,6 +44,9 @@ export function WorkflowsTab({ workbenchTenantId }: { readonly workbenchTenantId
                 <b>{definition.name}</b>
                 <span>{scheduleSentence(definition.schedule)}</span>
               </Link>
+              {definition.status === "deployed" && !definition.scheduleEnabled ? (
+                <span className="drawer-li-m">Paused</span>
+              ) : null}
               {definition.status === "deployed" ? (
                 <RunNowButton
                   variant="outline"

apps/web/src/pages/routine-detail-page.tsx

@@ -10,7 +10,13 @@ import { useNavigate } from "../navigation";
 import { WORKFLOWS_PATH_PREFIX } from "../path-ids";
 import { PageLayout } from "../shell/page-layout";
 import { StageTopBar } from "../shell/stage-top-bar";
-import { RoutinePill, formatWhen, routineState, useRoutineRuns } from "./routine-ui";
+import {
+  PauseResumeButton,
+  RoutinePill,
+  formatWhen,
+  routineState,
+  useRoutineRuns,
+} from "./routine-ui";
 import type { RunRow } from "./routine-ui";
 import { scheduleSentence } from "./routines-page";
 
@@ -187,6 +193,7 @@ export function RoutineDetailPage({
         ]}
         actions={
           <div className="flex items-center gap-2">
+            <PauseResumeButton row={row} />
             <RunNowButton variant="outline" size="sm" onRun={onRunNow} />
           </div>
         }

apps/web/src/pages/routine-ui.tsx

@@ -1,6 +1,13 @@
 import "./routines.css";
 
+import { Button } from "@corbits/react-ui";
+import { toast } from "@corbits/react-ui/ui/toast";
+import { reportError } from "@corbits/error-sink";
 import { WorkflowRunResponse, paginatedSchema } from "@intx/types";
+import { useMutation, useQueryClient } from "@tanstack/react-query";
+import { describeApiError } from "@/lib/api-query";
+import { tenantKeys } from "../query-client";
+import { pauseSchedule, resumeSchedule } from "../routines-api";
 import { useAPIQuery } from "../api";
 import type { APIQuery } from "@/lib/api-query";
 import type { GlobalRoutineRow } from "../global-routines";
@@ -40,7 +47,8 @@ export function routineState(
   lastRun: RunRow | undefined,
 ): { readonly tone: Tone; readonly label: string } {
   if (lastRun?.status === "error") return { tone: "failed", label: "failed" };
-  if (row.definition.status === "stopped") return { tone: "idle", label: "paused" };
+  if (row.definition.status === "stopped" || !row.definition.scheduleEnabled)
+    return { tone: "idle", label: "paused" };
   if (lastRun?.status === "running" || lastRun?.status === "updating") {
     return { tone: "info", label: "running" };
   }
@@ -60,3 +68,25 @@ export function RoutinePill({
     </span>
   );
 }
+
+/** Pauses or resumes the routine's cron schedule; absent when none exists. */
+export function PauseResumeButton({ row }: { readonly row: GlobalRoutineRow }) {
+  const queryClient = useQueryClient();
+  const { scheduleId, scheduleEnabled, name } = row.definition;
+  const toggle = useMutation({
+    mutationFn: () =>
+      (scheduleEnabled ? pauseSchedule : resumeSchedule)(row.tenantId, scheduleId ?? ""),
+    onSuccess: () => toast(`${name} ${scheduleEnabled ? "paused" : "resumed"}`),
+    onError: (cause) => {
+      reportError(cause, { operation: "scheduled_workflow_toggle_pause", tenantId: row.tenantId });
+      toast(`Couldn't update ${name}: ${describeApiError(cause, "updating this routine")}`);
+    },
+    onSettled: () => queryClient.invalidateQueries({ queryKey: tenantKeys.routines(row.tenantId) }),
+  });
+  if (scheduleId === null || row.definition.status === "stopped") return null;
+  return (
+    <Button variant="outline" size="sm" disabled={toggle.isPending} onClick={() => toggle.mutate()}>
+      {scheduleEnabled ? "Pause" : "Resume"}
+    </Button>
+  );
+}

apps/web/src/pages/routines-page.tsx

@@ -12,7 +12,13 @@ import { Link } from "../navigation";
 import { benchLink, useFromBench } from "../shell/page-crumbs";
 import { PageLayout } from "../shell/page-layout";
 import { StageTopBar } from "../shell/stage-top-bar";
-import { RoutinePill, formatWhen, routineState, useRoutineRuns } from "./routine-ui";
+import {
+  PauseResumeButton,
+  RoutinePill,
+  formatWhen,
+  routineState,
+  useRoutineRuns,
+} from "./routine-ui";
 
 export type { GlobalRoutineRow } from "../global-routines";
 
@@ -52,7 +58,10 @@ function RoutineListRow({
         {runs.kind === "ready" ? formatWhen(lastRun?.createdAt) : "…"}
       </span>
       <RoutinePill tone={state.tone}>{state.label}</RoutinePill>
-      <RunNowButton variant="outline" size="sm" onRun={() => onRunNow(row)} />
+      <div className="flex items-center gap-2">
+        <PauseResumeButton row={row} />
+        <RunNowButton variant="outline" size="sm" onRun={() => onRunNow(row)} />
+      </div>
     </li>
   );
 }

apps/web/src/routines-api.ts

@@ -21,6 +21,10 @@ export const ScheduledWorkflowDefinition = type({
   /** The cron expression firing this deployment's live run, or null when no
    * `@corbits/cron` row is addressed at it. */
   schedule: "string | null",
+  /** The cron row's id, or null when none is addressed at this deployment. */
+  scheduleId: "string | null",
+  /** False while the cron row is paused; true when it fires or none exists. */
+  scheduleEnabled: "boolean",
 });
 
 export type ScheduledWorkflowDefinition = typeof ScheduledWorkflowDefinition.infer;
@@ -42,6 +46,8 @@ export const CronSchedule = type({
    * never fires again, and a later redeploy does not resume it. */
   stoppedAt: "string | null",
   stoppedReason: "string | null",
+  /** False while the schedule is paused: the ticker skips it until resumed. */
+  enabled: "boolean",
   createdAt: "string",
 });
 
@@ -94,6 +100,32 @@ export async function createCronSchedule(
   return parsed.schedule;
 }
 
+async function setScheduleEnabled(
+  tenantId: string,
+  id: string,
+  action: "pause" | "resume",
+): Promise<CronSchedule> {
+  const path = `${cronPath(tenantId)}/${encodeURIComponent(id)}/${action}`;
+  const response = await fetch(path, { method: "POST", headers: { accept: "application/json" } });
+  if (response.status === 401) throw new UnauthenticatedError();
+  if (!response.ok) {
+    throw new ApiQueryError(`The server answered ${response.status}.`, response.status, path);
+  }
+  const parsed = CreatedCronSchedule(await response.json());
+  if (parsed instanceof type.errors) {
+    throw new ApiQueryError(`Unexpected response shape: ${parsed.summary}`, undefined, path);
+  }
+  return parsed.schedule;
+}
+
+export function pauseSchedule(tenantId: string, id: string): Promise<CronSchedule> {
+  return setScheduleEnabled(tenantId, id, "pause");
+}
+
+export function resumeSchedule(tenantId: string, id: string): Promise<CronSchedule> {
+  return setScheduleEnabled(tenantId, id, "resume");
+}
+
 export async function deleteCronSchedule(tenantId: string, id: string): Promise<void> {
   const path = `${cronPath(tenantId)}/${encodeURIComponent(id)}`;
   const response = await fetch(path, { method: "DELETE" });
@@ -158,10 +190,8 @@ export async function listScheduledWorkflows(
   const nameByAssetId = new Map(assets.map((asset) => [asset.id, asset.name]));
   // A schedule names its target by definition name, which is the source
   // asset's name — so a redeploy keeps the row joined to the same row here.
-  const expressionByDefinitionName = new Map(
-    schedules
-      .filter((row) => row.stoppedAt === null)
-      .map((row) => [row.definitionName, row.expression]),
+  const scheduleByDefinitionName = new Map(
+    schedules.filter((row) => row.stoppedAt === null).map((row) => [row.definitionName, row]),
   );
   return deployments
     .filter((deployment) => {
@@ -170,6 +200,7 @@ export async function listScheduledWorkflows(
     })
     .map((deployment) => {
       const name = nameByAssetId.get(deployment.definitionAssetId);
+      const row = name === undefined ? undefined : scheduleByDefinitionName.get(name);
       return {
         definitionId: deployment.id,
         assetId: deployment.definitionAssetId,
@@ -178,7 +209,9 @@ export async function listScheduledWorkflows(
         status: deployment.status === "deployed" ? "deployed" : "stopped",
         createdAt: deployment.createdAt,
         updatedAt: deployment.createdAt,
-        schedule: name === undefined ? null : (expressionByDefinitionName.get(name) ?? null),
+        schedule: row?.expression ?? null,
+        scheduleId: row?.id ?? null,
+        scheduleEnabled: row?.enabled ?? true,
       };
     });
 }