fix(listener): cancel a manager whose wake mail failed

apps/hub/src/manager-state.postgres.test.ts

@@ -4,7 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto";
 import { sql } from "drizzle-orm";
 import { createDB, runMigrations, schema } from "@intx/db";
 import type { Tx } from "./github/admin-github-grant.js";
-import { done, liveListener, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js";
+import { done, liveListener, liveManager, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js";
 import { migrateTriageTables } from "./triage-db.js";
 
 const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -449,6 +449,21 @@ test("PostgreSQL: a manager deployed inside the startup window, or one with a si
   });
 }, 120_000);
 
+test("PostgreSQL: a manager being cancelled is mailed nothing and takes nothing, but holds the pull request until it ends", async () => {
+  await withDatabase(async function cancelling(db) {
+    await recordEvent(db, TENANT, GITHUB, "d1", webhook("h1"));
+    await startManager(db, GITHUB, "run_cancelling");
+    await db.update(schema.workflowRun).set({ cancellationRequestedAt: new Date(), cancellationDeadline: new Date(), cancellationReason: "unwoken" }).where(sql`${schema.workflowRun.id} = 'run_cancelling'`);
+    await liveRun(db, "run_next");
+    expect(await liveManager(db, TENANT, GITHUB)).toBeUndefined();
+    expect(await take(db, TENANT, GITHUB, "run_cancelling")).toEqual({ outcome: "refused" });
+    expect(await take(db, TENANT, GITHUB, "run_next")).toEqual({ outcome: "refused" });
+    await db.update(schema.workflowRun).set({ status: "cancelled" }).where(sql`${schema.workflowRun.id} = 'run_cancelling'`);
+    expect(await take(db, TENANT, GITHUB, "run_next")).toMatchObject({ outcome: "pass", pass: { head: "h1" } });
+    expect(await stateOf(db, CLAIM)).toMatchObject({ status: "live", runId: "run_next" });
+  });
+}, 120_000);
+
 test("PostgreSQL: a listener counts as live as a manager does, and one deployed without a sidecar past the startup window is released", async () => {
   await withDatabase(async function listener(db) {
     await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" });

apps/hub/src/manager-state.ts

@@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto";
 import { and, desc, eq, inArray, isNull, like, or, sql, type SQL } from "drizzle-orm";
 import { jsonb, primaryKey, text } from "drizzle-orm/pg-core";
 import { type } from "arktype";
-import { schema, type DB } from "@intx/db";
+import { canExecuteWorkflowRun, schema, type DB } from "@intx/db";
 import { asRecord } from "@corbits/common";
 import { APPROVE_DO_KIND, REPO_SOURCES, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts";
 import type { Tx } from "./github/admin-github-grant.js";
@@ -195,13 +195,15 @@ export async function recordResult(db: DB["db"], tenantId: string, pr: PullReque
  */
 const MANAGER_START_GRACE_MS = 2 * 60_000;
 
-type ManagerRun = { status: string; createdAt: Date; allocationStatus: string | null; sidecarId: string | null };
+type ManagerRun = { status: string; createdAt: Date; expiresAt: Date | null; cancellationRequestedAt: Date | null; allocationStatus: string | null; sidecarId: string | null };
 
 async function managerRun(tx: Tx | DB["db"], runId: string): Promise<ManagerRun | undefined> {
   const [run] = await tx
     .select({
       status: schema.workflowRun.status,
       createdAt: schema.workflowRun.createdAt,
+      expiresAt: schema.workflowRun.expiresAt,
+      cancellationRequestedAt: schema.workflowRun.cancellationRequestedAt,
       allocationStatus: schema.sidecarAllocation.status,
       sidecarId: schema.sidecarAllocation.sidecarId,
     })
@@ -264,24 +266,28 @@ export async function liveListener(db: DB["db"], tenantId: string, source: RepoS
   return starting;
 }
 
-/** The run id of the pull request's manager while that run is up. */
+/** The run id of the pull request's manager while that run is up and stock still lets it execute, so one being cancelled is mailed nothing. */
 export async function liveManager(db: DB["db"], tenantId: string, pr: PullRequestRef): Promise<string | undefined> {
   return db.transaction(async function findLive(tx) {
     const now = new Date();
     const claim = await lockClaim(tx, tenantId, pr);
     if (claim?.status !== "live" || claim.runId === undefined) return undefined;
-    return (await runGone(tx, claim.runId, now)) ? undefined : claim.runId;
+    const run = await managerRun(tx, claim.runId);
+    return run && canExecuteWorkflowRun(run) && runIsUp(run, now) ? claim.runId : undefined;
   });
 }
 
 /**
  * The claim `runId` holds, taken for it when no up run holds it: it is missing or ended, or the run it names has gone,
- * whose open pass the caller resumes. Undefined while another up run holds it. Concurrent callers queue on the
- * claim's row lock, so exactly one takes it.
+ * whose open pass the caller resumes. Undefined while another up run holds it, a run being cancelled included until it
+ * ends, or once stock no longer lets `runId` execute. Concurrent callers queue on the claim's row lock, so exactly one
+ * takes it.
  */
 async function claimFor(tx: Tx, tenantId: string, pr: PullRequestRef, runId: string): Promise<Claim | undefined> {
   const key = claimKey(pr);
   const now = new Date();
+  const own = await managerRun(tx, runId);
+  if (own && !canExecuteWorkflowRun(own)) return undefined;
   const fresh = { status: "live" as const, claimId: randomUUID(), claimedAt: now.toISOString(), runId };
   await tx.insert(managerState).values({ tenantId, key, kind: "claim", state: claimOf(key, { ...fresh, reviewers: [], history: [], starts: 1 }) }).onConflictDoNothing();
   const claim = await lockClaim(tx, tenantId, pr);

docs/ARCHITECTURE.md

@@ -194,7 +194,17 @@ any other tenant or an unknown sender.
   hub publishes beside it in `release.json`, read as each body runs, and
   wakes it with a mail that names only the pull request. A pull request
   whose manager is up is mailed that manager instead, so a later webhook
-  does not deploy a second one. A mail that is not
+  does not deploy a second one. A wake mail waits up to two minutes for the
+  new manager to publish its key. When a wake mail fails for any reason, the
+  listener cancels that manager through the stock stop-run route under the
+  `manage` grant stock gives a deployment's creator, and its `unwoken` step's
+  output names each run it cancelled and why the mail failed. A run stock
+  no longer lets execute, as one whose cancellation is requested or whose
+  deadline has passed, is mailed nothing and its own take is refused; one
+  that holds the claim keeps it until it ends, so no second manager writes
+  to the pull request meanwhile. The pull request's next event, or the
+  reconciler re-queuing its head, which has no verdict, deploys a fresh
+  manager that claims it. A mail that is not
   the hub's, that names no pull request, or that finds no manager in
   `release.json` fails its body. Built with the release the module loads,
   its definition declares `workflow:*` `create` and `read` on exactly the

packages/triage-manager/README.md

@@ -4,7 +4,8 @@ A workspace's listener and a pull request's manager.
 
 - **Listener** (`triageListener`, `triage-listener-<host>`): the hub mails it the pull requests it recorded. For each, it
   deploys a manager through Interchange's deploy route with its run token and wakes it with a mail that names only the
-  pull request.
+  pull request. A manager whose wake mail fails is cancelled, and the pull request's next event, or the reconciler
+  re-queuing its head, deploys another.
 - **Manager** (`triageManager`, `triage-manager-<host>`): one live run per pull request, which runs only while it has
   work, and the only workflow that writes to the pull request. A mail starts a loop of passes: `take` claims the pull
   request and its pending events; each held Do approved since runs as pr-triage runs an approval; then pr-triage's own

packages/triage-manager/src/actions/index.ts

@@ -1,6 +1,6 @@
 export { execute, facts, mirror, rules, waiting } from "@corbits/triage-workflows/actions/index";
 export { executeApprovals } from "./approvals.js";
-export { forward } from "./listener.js";
+export { forward, unwoken } from "./listener.js";
 export { deployJudge, judgeMailed, judgeScope, verdict } from "./judge.js";
 export { done, stop, take } from "./manager.js";
 export { cleared } from "./review-clean.js";

packages/triage-manager/src/actions/listener.ts

@@ -1,9 +1,9 @@
 import type { EffectContext } from "@intx/workflow";
 import { type } from "arktype";
 import { isFromTriageSender, repoSourceOf, TRIAGE_LISTENERS } from "@corbits/triage-contracts";
-import type { MailInput } from "../mail.js";
+import { Mailed, MailInput } from "../mail.js";
 import { publishedFor } from "../release.js";
-import { deploy, mailOf, manageMail } from "./hub.js";
+import { deploy, endRun, mailOf, manageMail } from "./hub.js";
 
 /** How long a just-deployed manager has to reach a sidecar before its first mail fails. */
 const MANAGER_READY_MS = 2 * 60_000;
@@ -39,3 +39,27 @@ export async function forward(input: unknown, ctx: EffectContext, _signal: Abort
   }
   return { mails };
 }
+
+const UnwokenInput = type({ forward: { output: { mails: MailInput.array() } }, mail: { output: type({ reply: "string" }).array() } });
+
+/**
+ * Cancels each manager this body deployed whose wake mail did not go, through the stock stop-run route, so a manager
+ * that was never woken does not stay deployed; the output names each with why its mail failed. The pull request's next
+ * event, or the reconciler re-queuing its head, deploys a fresh one.
+ */
+export async function unwoken(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ cancelled: { runId: string; reason: string }[] }> {
+  const parsed = UnwokenInput(input);
+  if (parsed instanceof type.errors) throw new Error(`unwoken: ${parsed.summary}`);
+  const { mails } = parsed.forward.output;
+  const replies = parsed.mail.output;
+  if (replies.length !== mails.length) throw new Error(`unwoken: ${replies.length} mail replies for ${mails.length} managers`);
+  const cancelled: { runId: string; reason: string }[] = [];
+  for (const [index, mail] of mails.entries()) {
+    const reply = Mailed(replies[index]!.reply);
+    if (reply instanceof type.errors) throw new Error(`unwoken: ${reply.summary}`);
+    if (reply.mailed) continue;
+    await endRun(ctx, mail.tenantId, mail.to);
+    cancelled.push({ runId: mail.to, reason: typeof reply.error === "string" ? reply.error : JSON.stringify(reply.error) });
+  }
+  return { cancelled };
+}

packages/triage-manager/src/mail.ts

@@ -109,5 +109,5 @@ export function mailAgentOf(director: ReturnType<typeof mailDirectorOf>, inferen
   return defineAgent({ id: tolerate ? "triage-mail-tolerant" : "triage-mail", systemPrompt: "Send the mail the input names.", tools: [mailTool], capabilities: [], inference, director: director.build({ tolerate }) });
 }
 
-/** A tolerant mail agent's reply: whether the mail went. */
-export const Mailed = type("string.json.parse").to({ mailed: "boolean" });
+/** A tolerant mail agent's reply: whether the mail went, and why when it did not. */
+export const Mailed = type("string.json.parse").to(type({ mailed: "true" }).or({ mailed: "false", error: "unknown" }));

packages/triage-manager/src/triage-listener.ts

@@ -9,9 +9,10 @@ import { deployGrants, type Release } from "./release.js";
 
 /**
  * A workspace's listener for one host under `id`, its mail address, which the hub mails the pull requests it records:
- * it deploys each one's manager from the host's published release and wakes it. With `manager`, the release published
- * as the definition loads, it declares the grants to deploy that asset; without, none, and its body fails. It writes nothing to the pull request and holds
- * no grant on its state; a manager whose take finds another live one stops itself.
+ * it deploys each one's manager from the host's published release and wakes it, and cancels a manager whose wake mail
+ * did not go, so it is not left deployed. With `manager`, the release published as the definition loads, it declares the
+ * grants to deploy that asset; without, none, and its body fails. It writes nothing to the pull request and holds no
+ * grant on its state; a manager whose take finds another live one stops itself.
  */
 export function triageListener({ id, manager }: { id: string; manager: Release | undefined }) {
   const perMail = defineWorkflow({
@@ -24,7 +25,14 @@ export function triageListener({ id, manager }: { id: string; manager: Release |
         credentials: [HUB_CREDENTIAL_HANDLE],
         effect: { requires: [HUB_WRITE] },
       }),
-      mail: map({ over: { from: "steps.forward.output.mails" }, step: step({ agent: mailAgentOf(mailDirector, judgeAgent.inference, { tolerate: false }), input: { from: "trigger.payload" } }), after: ["forward"] }),
+      mail: map({ over: { from: "steps.forward.output.mails" }, step: step({ agent: mailAgentOf(mailDirector, judgeAgent.inference, { tolerate: true }), input: { from: "trigger.payload" } }), after: ["forward"] }),
+      unwoken: action({
+        handler: "unwoken",
+        input: { project: { from: "steps" }, fields: ["forward", "mail"] },
+        after: ["mail"],
+        credentials: [HUB_CREDENTIAL_HANDLE],
+        effect: { requires: [HUB_WRITE] },
+      }),
     },
   });
   return defineWorkflow({

packages/triage-manager/src/triage-manager.test.ts

@@ -101,7 +101,7 @@ function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[
 
 /** The host's credentials as triage-workflows wires them, plus the hub credential deploy binds to the hub actions. */
 function actionCredentials(bindings: Bindings, host: Fetch, hub: (url: URL, init?: RequestInit) => Promise<Response>) {
-  const consumers = ["take", "done", "deployReview", "deployJudge", "forward", "stop"].map(actionConsumer);
+  const consumers = ["take", "done", "deployReview", "deployJudge", "forward", "unwoken", "stop"].map(actionConsumer);
   const cell = delivered(bindings, [{ credentialId: "hub", providerKey: "http", origin: HUB_ORIGIN, secret: "run-token" }]);
   cell.current = { ...cell.current, bindings: [...cell.current.bindings, ...consumers.map((consumer) => ({ handle: HUB_CREDENTIAL_HANDLE, credentialId: "hub", consumer }))] };
   const hubGrants: GrantRule[] = consumers.map((tool, index) => ({ id: `hub-${index}`, resource: "credential:hub", action: "use", effect: "allow", origin: "system", conditions: { tool }, expiresAt: null, roleId: null, principalId: null }));
@@ -136,7 +136,7 @@ async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetc
     invokeStep: async function mailAgent({ agent, input }) {
       if (!agent.id.startsWith("triage-mail")) throw new Error(`no agent but the mail agent runs, not ${agent.id}`);
       mailed.push(input);
-      const reply = JSON.stringify({ mailed: delivers });
+      const reply = JSON.stringify(delivers ? { mailed: true } : { mailed: false, error: "public key was not published by its readyBy" });
       return { output: { reply, turn: { role: "assistant", content: [{ type: "text", text: reply }] } } };
     },
     actionResolver: (ref) => (actions as Record<string, ActionHandler>)[ref]!,
@@ -912,6 +912,15 @@ describe("triage-listener", () => {
     ]);
   });
 
+  test("a manager whose wake mail did not go is cancelled once, and the listener deploys no other for it", async () => {
+    const hub = fakeHub({ takes: [] });
+    const mail = triggerMail(PR, RULES, { tenantId: TENANT, items: [{ prNumber: 8, headSha: HEAD }] });
+    const { terminalStatus, outputs } = await runMail(listener, mail, githubHost().host, hub.hub, [], false);
+    expect(terminalStatus).toBe("completed");
+    expect(hub.calls.map(({ method, path }) => ({ method, path }))).toEqual([DEPLOY, { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${DEPLOYED}` }]);
+    expect(outputs["unwoken"]).toEqual({ cancelled: [{ runId: DEPLOYED, reason: expect.stringContaining("public key") }] });
+  });
+
   test("a mail that names no pull request, or is not from the hub, fails instead of being dropped", async () => {
     const hub = fakeHub({ takes: [] });
     const mail = triggerMail(PR, RULES, { tenantId: TENANT });