feat(sdk,webapp): add tags.delete() to remove tags from a run

.changeset/tags-delete.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+You can now remove tags from a run while it's running with `tags.delete("my-tag")` (or an array of tags). It only affects that run — every other run keeps the tag, and the tag stays available to filter by.

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -1,10 +1,10 @@
 import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
-import { AddTagsRequestBody } from "@trigger.dev/core/v3";
+import { AddTagsRequestBody, RemoveTagsRequestBody } from "@trigger.dev/core/v3";
 import type { BufferEntry } from "@trigger.dev/redis-worker";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
-import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { type AuthenticatedEnvironment, authenticateApiRequest } from "~/services/apiAuth.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { logger } from "~/services/logger.server";
 import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
@@ -31,23 +31,59 @@ const ParamsSchema = z.object({
   runId: z.string(),
 });
 
-export async function action({ request, params }: ActionFunctionArgs) {
-  if (request.method.toUpperCase() !== "POST") {
-    return { status: 405, body: "Method Not Allowed" };
-  }
+type ResolvedRequest =
+  | { kind: "ok"; environment: AuthenticatedEnvironment; runId: string }
+  | { kind: "error"; response: Response };
 
+// Auth + params, shared by both methods. Secret-key auth only, deliberately: adding
+// an RBAC `authorization` gate here would change behaviour for narrowly-scoped JWTs.
+async function resolveRequest(
+  request: Request,
+  params: ActionFunctionArgs["params"]
+): Promise<ResolvedRequest> {
   const authenticationResult = await authenticateApiRequest(request);
   if (!authenticationResult) {
-    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+    return {
+      kind: "error",
+      response: json({ error: "Invalid or Missing API Key" }, { status: 401 }),
+    };
   }
 
   const parsedParams = ParamsSchema.safeParse(params);
   if (!parsedParams.success) {
-    return json(
-      { error: "Invalid request parameters", issues: parsedParams.error.issues },
-      { status: 400 }
-    );
+    return {
+      kind: "error",
+      response: json(
+        { error: "Invalid request parameters", issues: parsedParams.error.issues },
+        { status: 400 }
+      ),
+    };
+  }
+
+  return {
+    kind: "ok",
+    environment: authenticationResult.environment,
+    runId: parsedParams.data.runId,
+  };
+}
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  switch (request.method.toUpperCase()) {
+    case "POST":
+      return addRunTags(request, params);
+    case "DELETE":
+      return removeRunTags(request, params);
+    default:
+      return json({ error: "Method Not Allowed" }, { status: 405 });
+  }
+}
+
+async function addRunTags(request: Request, params: ActionFunctionArgs["params"]) {
+  const resolved = await resolveRequest(request, params);
+  if (resolved.kind === "error") {
+    return resolved.response;
   }
+  const { environment: env, runId } = resolved;
 
   try {
     const anyBody = await request.json();
@@ -62,9 +98,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
       return json({ message: "No new tags to add" }, { status: 200 });
     }
 
-    const env = authenticationResult.environment;
     const outcome = await mutateWithFallback<Response>({
-      runId: parsedParams.data.runId,
+      runId,
       environmentId: env.id,
       organizationId: env.organizationId,
       bufferPatch: { type: "append_tags", tags: nonEmptyTags, maxTags: MAX_TAGS_PER_RUN },
@@ -141,3 +176,143 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }
 }
+
+// Removes tags from THIS run only. Tags aren't a shared entity — they're strings in
+// the run's own `runTags` array — so there is nothing org-wide to delete, and other
+// runs carrying the same tag are untouched.
+async function removeRunTags(request: Request, params: ActionFunctionArgs["params"]) {
+  const resolved = await resolveRequest(request, params);
+  if (resolved.kind === "error") {
+    return resolved.response;
+  }
+  const { environment: env, runId } = resolved;
+
+  try {
+    // A DELETE with no body at all is the natural thing to try by hand, and
+    // `request.json()` throws on it. That's a malformed request, not a server fault, so
+    // map it to 400 instead of letting it fall through to the catch-all 500 + error log.
+    let anyBody: unknown;
+    try {
+      anyBody = await request.json();
+    } catch {
+      return json({ error: "Invalid request body" }, { status: 400 });
+    }
+
+    const body = RemoveTagsRequestBody.safeParse(anyBody);
+    if (!body.success) {
+      return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 });
+    }
+    const bodyTags = typeof body.data.tags === "string" ? [body.data.tags] : body.data.tags;
+    const nonEmptyTags = bodyTags.filter((t) => t.trim().length > 0);
+
+    // Nothing asked for. No MAX_TAGS_PER_RUN check applies to a removal — it can
+    // only ever shrink the list.
+    if (nonEmptyTags.length === 0) {
+      return json({ message: "No tags to remove" }, { status: 200 });
+    }
+
+    const outcome = await mutateWithFallback<Response>({
+      runId,
+      environmentId: env.id,
+      organizationId: env.organizationId,
+      bufferPatch: { type: "remove_tags", tags: nonEmptyTags },
+      pgMutation: async (taskRun) => {
+        const doomed = new Set(nonEmptyTags);
+        let existing = taskRun.runTags ?? [];
+        let remaining = existing.filter((t) => !doomed.has(t));
+
+        if (existing.length === remaining.length) {
+          // `taskRun` normally comes from the READ REPLICA, so "the run has none of
+          // these tags" can just be replication lag rather than the truth — and
+          // `tags.add("x")` immediately followed by `tags.delete("x")` is a completely
+          // ordinary pattern. Confirm against the primary before concluding there's
+          // nothing to do, the same disambiguation mutateWithFallback does for a
+          // replica-lag 404. Passing the writer as the read client makes the routing
+          // store read the OWNING store's primary, so this is read-your-writes for a
+          // run on either database.
+          //
+          // The add path's equivalent skip needs no such check because a stale read is
+          // safe in that direction: it can only make the write bigger, never skip a
+          // real one. Skipping a real removal, by contrast, would silently drop the
+          // customer's mutation and still answer 200.
+          const fresh = await runStore.findRun(
+            { id: taskRun.id, runtimeEnvironmentId: env.id },
+            { select: { runTags: true } },
+            prisma
+          );
+
+          if (!fresh) {
+            return json({ error: "Run not found" }, { status: 404 });
+          }
+
+          existing = fresh.runTags ?? [];
+          remaining = existing.filter((t) => !doomed.has(t));
+        }
+
+        const removedCount = existing.length - remaining.length;
+
+        // Removing tags the run doesn't have is an idempotent success, not a 404.
+        // Skip the write entirely so we don't bump `updatedAt`, replicate a row that
+        // didn't change, or publish a no-op realtime record.
+        if (removedCount === 0) {
+          return json({ message: "Successfully removed 0 tags." }, { status: 200 });
+        }
+
+        const updated = await runStore.removeTags(
+          taskRun.id,
+          nonEmptyTags,
+          { runtimeEnvironmentId: env.id },
+          prisma
+        );
+
+        // The run vanished (or moved environment) between the read and this write.
+        if (!updated) {
+          return json({ error: "Run not found" }, { status: 404 });
+        }
+
+        // Publish a run-changed record with the REMAINING tag set so tag feeds
+        // reindex (no-op unless enabled). updatedAt is the read-your-writes
+        // watermark. Note this record no longer carries the removed tag, so a feed
+        // subscribed to that tag simply stops receiving this run — there is no
+        // "tag removed" un-subscribe signal.
+        publishChangeRecord({
+          runId: taskRun.id,
+          envId: env.id,
+          tags: remaining,
+          batchId: taskRun.batchId,
+          updatedAtMs: updated.updatedAt.getTime(),
+        });
+
+        return json({ message: `Successfully removed ${removedCount} tags.` }, { status: 200 });
+      },
+      // Buffer-applied patch path. The Lua removed the tags from the snapshot
+      // atomically. Count off the pre-mutation entry (already fetched by
+      // mutateWithFallback's env-auth pre-check, so no extra Redis read) so the
+      // message reports the same number the PG path would for the same input.
+      synthesisedResponse: ({ bufferEntry }) => {
+        // `null` here means the snapshot carried no readable `tags` array. For a
+        // buffered run that is simply the shape of a run triggered without tags, so
+        // there was nothing to remove — unlike the add path, whose unknown-existing
+        // fallback is the request count, falling back to the request count here would
+        // report removals that never happened on the most common buffered shape.
+        const existing = parseSnapshotTags(bufferEntry) ?? [];
+        const removedCount = existing.filter((t) => nonEmptyTags.includes(t)).length;
+        return json({ message: `Successfully removed ${removedCount} tags.` }, { status: 200 });
+      },
+      // No `rejectedResponse`: a `remove_tags` patch carries no cap, so the buffer
+      // never reports `limit_exceeded` for it.
+      abortSignal: getRequestAbortSignal(),
+    });
+
+    if (outcome.kind === "not_found") {
+      return json({ error: "Run not found" }, { status: 404 });
+    }
+    if (outcome.kind === "timed_out") {
+      return json({ error: "Run materialisation timed out" }, { status: 503 });
+    }
+    return outcome.response;
+  } catch (error) {
+    logger.error("Failed to remove run tags", { error });
+    return json({ error: "Something went wrong, please try again." }, { status: 500 });
+  }
+}

docs/docs.json

@@ -347,6 +347,7 @@
               "management/runs/reschedule",
               "management/runs/update-metadata",
               "management/runs/add-tags",
+              "management/runs/remove-tags",
               "management/runs/retrieve-events",
               "management/runs/retrieve-trace",
               "management/runs/retrieve-result"

docs/management/runs/remove-tags.mdx

@@ -0,0 +1,4 @@
+---
+title: "Remove tags from a run"
+openapi: "v3-openapi DELETE /api/v1/runs/{runId}/tags"
+---

docs/tags.mdx

@@ -18,11 +18,13 @@ We don't enforce prefixes but if you use them you'll find it easier to filter an
 
 ## How to add tags
 
-There are two ways to add tags to a run:
+You can add tags to a run in two ways:
 
 1. When triggering the run.
 2. Inside the `run` function, using `tags.add()`.
 
+You can also [remove tags](#removing-tags) from a run while it's running.
+
 ### 1. Adding tags when triggering the run
 
 You can add tags when triggering a run using the `tags` option. All the different [trigger](/triggering) methods support this.
@@ -96,6 +98,36 @@ export const myTask = task({
 });
 ```
 
+## Removing tags
+
+Use the `tags.delete()` function to remove tags from inside the `run` function. It takes a single string or an array of strings, just like `tags.add()`:
+
+```ts
+import { tags, task } from "@trigger.dev/sdk";
+
+export const myTask = task({
+  id: "my-task",
+  run: async (payload: { message: string }) => {
+    await tags.add(["status_processing", "user_123456"]);
+
+    // ...do the work...
+
+    // Remove a single tag, or an array of tags
+    await tags.delete("status_processing");
+    await tags.add("status_done");
+  },
+});
+```
+
+This only affects the run you call it from. Other runs keep the same tag, and the tag stays available to filter by in the dashboard.
+
+<Note>
+  Removing a tag the run doesn't have does nothing and won't throw, so you don't need to
+  check the run's current tags first.
+</Note>
+
+Tags can only be removed from inside the `run` function — there's no equivalent option when triggering.
+
 ## Filtering runs by tags
 
 You can filter runs by tags in the dashboard and in the SDK.