feat(supervisor): reclaim a run's checkpoint storage when it finishes (#4493)

When a run reaches a terminal state, ask the checkpoint service to
reclaim the storage its checkpoints occupied. Storage for finished runs
is not otherwise reclaimed, so nothing frees it today.

**Off by default** behind `DELETE_CHECKPOINTS_ON_COMPLETION`, and the
service-side handler ships separately, so merging this changes no
behaviour.

## Where the tenancy comes from

Addressing a run's checkpoints needs org, project, environment,
deployment version and run id. All five are already in hand at
`attempt.complete`, and three are **signed** by the deployment token:

| Value | Source | Trust |
| -- | -- | -- |
| org | claim `org_id` | signed |
| environment | claim `environment_id` | signed |
| deployment version | claim `deployment_version` | signed |
| project ref | `x-trigger-workload-project-ref` header |
runner-supplied |
| run | route param | runner-supplied |

`authorizeWorkloadRequest` previously returned only `environment_id`,
and only in enforce mode, so it now also returns the verified `claims`.
That difference is deliberate and documented on the method: claims are
used to address a run's **own** resources locally, never to scope the
platform, which is why `environmentId` stays enforce-only.

The two runner-supplied values are safe because the signed ones are
outermost - a runner lying about either can only name something inside
its own org and environment, and a project ref that doesn't pair with
its signed environment matches nothing. The run id is read from
`params.runFriendlyId`, the same value the platform just validated,
rather than from the body or a header. Where both a claim and a header
exist (`deployment_version`), the claim wins.

## Placement

The call sits after `reply.json(...)`, so the runner sees no added
latency - the same shape the suspend route already uses. The service
enqueues and returns 202, so it is one fast local hop.

Terminal means `RUN_FINISHED` **or `RUN_PENDING_CANCEL`** - a run
cancelled mid-execution never restores, and skipping it would leave its
storage behind. Retries are excluded deliberately: reclamation is
per-run, so a retry is covered by the final completion.

Also gated on `!snapshotService`, so it stays inert where checkpoints
aren't the kind this reclaims.

## Observability

`checkpoint_delete_requests_total{result}` counts `sent` **and every
reason we decide not to send**: `disabled`, `not_terminal`, `no_claims`,
`no_project_ref`, `http_error`.

The negative labels are the point - without them, "no requests are
happening" looks identical to the feature being switched off.
`no_claims` is reachable even under enforcement, since enforce only
rejects a *present-but-invalid* token; an absent or legacy id still
passes with no claims attached.

## Notes for review

- **No changeset**: `CheckpointClient` is `core/v3/serverOnly`, an
internal service-to-service API rather than customer-facing surface.
- **No `.server-changes/` note**: there is nothing a dashboard user
would notice here. Happy to add one if you disagree.
- `pnpm run typecheck` can't complete in my checkout -
`@trigger.dev/database` fails to build on a missing `tsc` in the pnpm
store, unrelated to this diff. Verified with `tsc --noEmit` against the
supervisor project instead: **zero errors in `apps/supervisor/src`**.
Worth noting it caught a real bug here - the completion response is
wrapped, so the status is `data.result.attemptStatus`.

refs TRI-12789

Author: nicktrn

apps/supervisor/src/env.ts

@@ -22,6 +22,7 @@ export const Env = z
     // also reject invalid tokens.
     WORKLOAD_TOKEN_SECRET: z.string().optional(),
     WORKLOAD_TOKEN_ENFORCEMENT: z.enum(["disabled", "log", "enforce"]).default("disabled"),
+    DELETE_CHECKPOINTS_ON_COMPLETION: BoolEnv.default(false), // irreversible; enable per cluster
     // Absolute expiry for minted deployment tokens. Deterministic (no wall-clock issued-at) so every
     // pod of a deployment carries an identical token; bump before this date. Must outlive any run.
     WORKLOAD_TOKEN_EXP: z.string().datetime().default("2032-01-01T00:00:00.000Z"),
@@ -326,6 +327,14 @@ export const Env = z
         path: ["WORKLOAD_TOKEN_SECRET"],
       });
     }
+    if (data.DELETE_CHECKPOINTS_ON_COMPLETION && data.WORKLOAD_TOKEN_ENFORCEMENT === "disabled") {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message:
+          "DELETE_CHECKPOINTS_ON_COMPLETION needs WORKLOAD_TOKEN_ENFORCEMENT set to log or enforce: the tenancy it deletes by comes from the deployment token, so with tokens disabled it would silently reclaim nothing",
+        path: ["DELETE_CHECKPOINTS_ON_COMPLETION"],
+      });
+    }
     if (
       data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED &&
       !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST

apps/supervisor/src/workloadServer/index.ts

@@ -23,13 +23,16 @@ import EventEmitter from "node:events";
 import type { IncomingMessage, ServerResponse } from "node:http";
 import { type Namespace, Server, type Socket } from "socket.io";
 import { z } from "zod";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { Counter } from "prom-client";
 import { env } from "../env.js";
 import { register } from "../metrics.js";
 import {
   verifyDeploymentIdHeader,
   workloadTokenEnforced,
   workloadTokensEnabled,
 } from "../workloadToken.js";
+import type { WorkloadDeploymentTokenClaims } from "@trigger.dev/core/v3";
 import {
   ComputeSnapshotService,
   type RunTraceContext,
@@ -50,6 +53,13 @@ interface DefaultEventsMap {
   [event: string]: (...args: any[]) => void;
 }
 
+const checkpointDeleteRequests = new Counter({
+  name: "checkpoint_delete_requests_total",
+  help: "Checkpoint delete requests attempted at run completion, by outcome",
+  labelNames: ["result"],
+  registers: [register],
+});
+
 const WorkloadActionParams = z.object({
   runFriendlyId: z.string(),
   snapshotFriendlyId: z.string(),
@@ -181,10 +191,15 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
    * environment_id to forward upstream. The env id is only forwarded in enforce mode: in log mode
    * we still verify + record metrics but attach no header (so the platform never scopes). Only
    * enforce fails a request, and only for a present-but-invalid token; absent and legacy ids pass.
+   *
+   * `claims` are returned on any valid token, for local use only - never to scope the platform,
+   * which is why environmentId stays gated on enforce.
    */
   private async authorizeWorkloadRequest(
     req: IncomingMessage
-  ): Promise<{ ok: true; environmentId?: string } | { ok: false }> {
+  ): Promise<
+    { ok: true; environmentId?: string; claims?: WorkloadDeploymentTokenClaims } | { ok: false }
+  > {
     if (!workloadTokensEnabled) {
       return { ok: true };
     }
@@ -201,9 +216,73 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         workloadTokenEnforced && result.outcome === "jwt_valid"
           ? result.claims.environment_id
           : undefined,
+      claims: result.outcome === "jwt_valid" ? result.claims : undefined,
     };
   }
 
+  /**
+   * reclaimCheckpoints asks the checkpoint service to delete a finished run's checkpoint storage.
+   * Must be called after the reply is sent: it never delays the runner.
+   */
+  private async reclaimCheckpoints(
+    req: IncomingMessage,
+    runFriendlyId: string,
+    attemptStatus: string,
+    claims: WorkloadDeploymentTokenClaims | undefined
+  ): Promise<void> {
+    if (!env.DELETE_CHECKPOINTS_ON_COMPLETION) {
+      checkpointDeleteRequests.inc({ result: "disabled" });
+      return;
+    }
+
+    if (!this.checkpointClient) {
+      checkpointDeleteRequests.inc({ result: "no_client" });
+      return;
+    }
+
+    if (this.snapshotService) {
+      checkpointDeleteRequests.inc({ result: "not_applicable" });
+      return;
+    }
+
+    if (attemptStatus !== "RUN_FINISHED" && attemptStatus !== "RUN_PENDING_CANCEL") {
+      checkpointDeleteRequests.inc({ result: "not_terminal" });
+      return;
+    }
+
+    if (!claims) {
+      checkpointDeleteRequests.inc({ result: "no_claims" });
+      return;
+    }
+
+    const projectRef = this.projectRefFromRequest(req);
+    if (!projectRef) {
+      checkpointDeleteRequests.inc({ result: "no_project_ref" });
+      this.logger.error("Cannot reclaim checkpoints without a project ref", { runFriendlyId });
+      return;
+    }
+
+    const [error, accepted] = await tryCatch(
+      this.checkpointClient.deleteCheckpoints({
+        runFriendlyId,
+        body: {
+          orgId: claims.org_id,
+          envId: claims.environment_id,
+          deploymentVersion: claims.deployment_version,
+          projectRef,
+        },
+      })
+    );
+
+    if (error || !accepted) {
+      checkpointDeleteRequests.inc({ result: "http_error" });
+      this.logger.error("Failed to request checkpoint reclaim", { runFriendlyId, error });
+      return;
+    }
+
+    checkpointDeleteRequests.inc({ result: "sent" });
+  }
+
   /**
    * Sets common route meta on the wide-event state from URL params.
    */
@@ -364,6 +443,13 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 }
 
                 reply.json(completeResponse.data satisfies WorkloadRunAttemptCompleteResponseBody);
+
+                await this.reclaimCheckpoints(
+                  req,
+                  params.runFriendlyId,
+                  completeResponse.data.result.attemptStatus,
+                  auth.claims
+                );
                 return;
               }
             ),

packages/core/src/v3/serverOnly/checkpointClient.ts

@@ -119,4 +119,42 @@ export class CheckpointClient {
 
     return true;
   }
+
+  /**
+   * Ask the checkpoint service to reclaim a finished run's checkpoint storage. Best-effort: the
+   * service enqueues and returns 202, so a `true` here means accepted, not deleted.
+   */
+  async deleteCheckpoints({
+    runFriendlyId,
+    body,
+  }: {
+    runFriendlyId: string;
+    body: {
+      orgId: string;
+      envId: string;
+      projectRef: string;
+      deploymentVersion: string;
+    };
+  }): Promise<boolean> {
+    const res = await fetch(
+      new URL(`/api/v1/runs/${runFriendlyId}/checkpoints/delete`, this.opts.apiUrl),
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify(body),
+      }
+    );
+
+    if (!res.ok) {
+      this.logger.error("[CheckpointClient] Delete checkpoints request failed", {
+        runFriendlyId,
+        status: res.status,
+      });
+      return false;
+    }
+
+    return true;
+  }
 }