feat(scheduled-tasks): event filters for webhook deliveries

apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx

@@ -13,6 +13,7 @@ import type {
 } from "@t3tools/contracts";
 import {
   MAX_WEBHOOK_DELIVERY_AGE_MINUTES,
+  MAX_WEBHOOK_FILTERS,
   resolveEnvironmentMachineKind,
 } from "@t3tools/contracts";
 import type { MenuAction } from "@react-native-menu/menu";
@@ -24,7 +25,11 @@ import {
 } from "@t3tools/client-runtime/state/runtime";
 import {
   DEFAULT_WEBHOOK_PROMPT,
+  EMPTY_WEBHOOK_FILTER_DRAFT,
   parseMaxDeliveryAge,
+  parseWebhookFilterDrafts,
+  WEBHOOK_FILTER_OPS,
+  type WebhookFilterDraft,
 } from "@t3tools/client-runtime/scheduled-task-webhook";
 import { useEffect, useLayoutEffect, useRef, useState, type ReactNode } from "react";
 import { useFocusEffect, useNavigation, usePreventRemove } from "@react-navigation/native";
@@ -222,6 +227,85 @@ function SelectRow(props: {
   );
 }
 
+function WebhookFilterRows(props: {
+  readonly filters: ReadonlyArray<WebhookFilterDraft>;
+  readonly disabled: boolean;
+  readonly onChange: (filters: ReadonlyArray<WebhookFilterDraft>) => void;
+}) {
+  const update = (index: number, patch: Partial<WebhookFilterDraft>) =>
+    props.onChange(
+      props.filters.map((filter, at) => (at === index ? { ...filter, ...patch } : filter)),
+    );
+  return (
+    <>
+      {props.filters.map((filter, index) => (
+        // Rows have no identity of their own; removing one re-renders the rest in order.
+        // oxlint-disable-next-line react/no-array-index-key
+        <View key={index}>
+          <View className="flex-row items-center gap-3 border-t border-border-subtle px-4 pt-3">
+            <Text className="min-w-0 flex-1 text-sm text-foreground-muted">Filter {index + 1}</Text>
+            <Pressable
+              accessibilityRole="button"
+              accessibilityLabel={`Remove filter ${index + 1}`}
+              disabled={props.disabled}
+              hitSlop={8}
+              onPress={() => props.onChange(props.filters.filter((_, at) => at !== index))}
+              className="active:opacity-70"
+            >
+              <SymbolView name="trash" size={16} tintColorClassName="accent-danger-foreground" />
+            </Pressable>
+          </View>
+          <FormField
+            label="Path"
+            value={filter.path}
+            placeholder="body.action"
+            disabled={props.disabled}
+            onChange={(path) => update(index, { path })}
+          />
+          <SelectRow
+            label="Test"
+            value={WEBHOOK_FILTER_OPS.find((option) => option.op === filter.op)?.label ?? ""}
+            actions={WEBHOOK_FILTER_OPS.map((option) => ({
+              id: option.op,
+              title: option.label,
+              state: option.op === filter.op ? ("on" as const) : undefined,
+            }))}
+            onSelect={(id) => {
+              const known = WEBHOOK_FILTER_OPS.find((option) => option.op === id);
+              if (known) update(index, { op: known.op });
+            }}
+          />
+          {filter.op === "exists" ? null : (
+            <FormField
+              label={filter.op === "in" ? "Values, comma-separated" : "Value"}
+              value={filter.value}
+              placeholder={filter.op === "in" ? "opened, reopened" : "opened"}
+              disabled={props.disabled}
+              onChange={(value) => update(index, { value })}
+            />
+          )}
+        </View>
+      ))}
+      {props.filters.length < MAX_WEBHOOK_FILTERS ? (
+        <Pressable
+          accessibilityRole="button"
+          disabled={props.disabled}
+          onPress={() => props.onChange([...props.filters, EMPTY_WEBHOOK_FILTER_DRAFT])}
+          className="min-h-14 flex-row items-center gap-3 border-t border-border-subtle px-4 py-3 active:opacity-70"
+        >
+          <Text className="min-w-0 flex-1 text-lg text-foreground">Add filter</Text>
+          <SymbolView name="plus" size={16} tintColorClassName="accent-icon" />
+        </Pressable>
+      ) : null}
+      {props.filters.length > 0 ? (
+        <Text className="px-4 pb-3 text-sm text-foreground-muted">
+          Runs only when every filter matches. Other requests are logged as filtered.
+        </Text>
+      ) : null}
+    </>
+  );
+}
+
 function PickerRow(props: {
   readonly label: string;
   readonly value: string;
@@ -627,6 +711,14 @@ function TaskForm({
       );
       return;
     }
+    const parsedFilters = parseWebhookFilterDrafts(draft.schedule.filters);
+    if (draft.schedule.mode === "webhook" && "invalidIndex" in parsedFilters) {
+      Dialog.alert(
+        "Invalid filter",
+        `Filter ${parsedFilters.invalidIndex + 1} needs a path under body, headers or query, and a value its test accepts.`,
+      );
+      return;
+    }
     if (
       !draft.title.trim() ||
       !draft.prompt.trim() ||
@@ -964,6 +1056,13 @@ function TaskForm({
                   setDraft({ ...draft, schedule: { ...draft.schedule, maxDeliveryAgeMinutes } })
                 }
               />
+              <WebhookFilterRows
+                filters={draft.schedule.filters}
+                disabled={saving}
+                onChange={(filters) =>
+                  setDraft({ ...draft, schedule: { ...draft.schedule, filters } })
+                }
+              />
             </>
           ) : (
             <>

apps/mobile/src/features/settings/scheduledTaskDraft.test.ts

@@ -65,6 +65,18 @@ describe("scheduleDraftForTask", () => {
     expect(saved).toEqual({ type: "webhook", signature, maxDeliveryAgeMinutes: null });
     expect(saved?.type === "webhook" && saved.signature && "secret" in saved.signature).toBe(false);
   });
+
+  it("round-trips webhook filters and refuses an invalid one", () => {
+    const filters = [
+      { path: "headers.x-github-event", op: "equals", value: "pull_request" },
+      { path: "body.label.name", op: "matches", value: "^agent:" },
+    ] as const;
+    const draft = scheduleDraftForTask({ schedule: { type: "webhook", signature: null, filters } });
+    expect(scheduleFromDraft(draft)).toMatchObject({ filters });
+    expect(
+      scheduleFromDraft({ ...draft, filters: [{ path: "", op: "exists", value: "" }] }),
+    ).toBeNull();
+  });
 });
 
 describe("hasScheduledTaskDraftChanges", () => {

apps/mobile/src/features/settings/scheduledTaskDraft.ts

@@ -9,7 +9,12 @@ import type {
 } from "@t3tools/contracts";
 
 import { DEFAULT_SERVER_SETTINGS } from "@t3tools/contracts";
-import { parseMaxDeliveryAge } from "@t3tools/client-runtime/scheduled-task-webhook";
+import {
+  parseMaxDeliveryAge,
+  parseWebhookFilterDrafts,
+  webhookFilterDrafts,
+  type WebhookFilterDraft,
+} from "@t3tools/client-runtime/scheduled-task-webhook";
 import {
   resolveProjectSettings,
   type LegacyProjectSettingsFields,
@@ -50,6 +55,7 @@ export type ScheduleDraft = {
   readonly intervalMinutes: string;
   readonly signature: ScheduledTaskWebhookSignature | null;
   readonly maxDeliveryAgeMinutes: string;
+  readonly filters: ReadonlyArray<WebhookFilterDraft>;
 };
 
 export const DEFAULT_SCHEDULE: ScheduleDraft = {
@@ -59,6 +65,7 @@ export const DEFAULT_SCHEDULE: ScheduleDraft = {
   intervalMinutes: "15",
   signature: null,
   maxDeliveryAgeMinutes: "",
+  filters: [],
 };
 
 export function scheduleDraftForTask(task: Pick<ScheduledTask, "schedule">): ScheduleDraft {
@@ -86,6 +93,7 @@ export function scheduleDraftForTask(task: Pick<ScheduledTask, "schedule">): Sch
           task.schedule.maxDeliveryAgeMinutes == null
             ? ""
             : String(task.schedule.maxDeliveryAgeMinutes),
+        filters: webhookFilterDrafts(task.schedule.filters),
       };
   }
 }
@@ -94,6 +102,8 @@ export function scheduleFromDraft(draft: ScheduleDraft): ScheduledTaskUpsertSche
   if (draft.mode === "webhook") {
     const maxDeliveryAgeMinutes = parseMaxDeliveryAge(draft.maxDeliveryAgeMinutes);
     if (maxDeliveryAgeMinutes === undefined) return null;
+    const parsedFilters = parseWebhookFilterDrafts(draft.filters);
+    if (!("filters" in parsedFilters)) return null;
     // No secret is sent, so the server keeps the stored one.
     return {
       type: "webhook",
@@ -106,6 +116,7 @@ export function scheduleFromDraft(draft: ScheduleDraft): ScheduledTaskUpsertSche
               prefix: draft.signature.prefix,
             },
       maxDeliveryAgeMinutes,
+      ...(parsedFilters.filters.length > 0 ? { filters: parsedFilters.filters } : {}),
     };
   }
   if (draft.mode === "interval") {
@@ -160,6 +171,7 @@ function draftSignature(draft: ScheduledTaskDraft): string {
     [...draft.schedule.weekdays].sort((a, b) => a - b),
     draft.schedule.intervalMinutes,
     draft.schedule.maxDeliveryAgeMinutes,
+    draft.schedule.filters,
     draft.workspace,
     draft.baseRef,
     draft.checkoutPath,

apps/server/src/mcp/toolkits/orchestrator/tools.ts

@@ -116,7 +116,7 @@ const WorkflowStopTool = Tool.make("t3_workflow_stop", {
 
 export const ScheduleTaskTool = Tool.make("schedule_task", {
   description:
-    "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). Each value arrives framed as untrusted request data, and $skill, @file or /command mentions inside it do not act; without a signature, anyone with the URL can trigger a run. For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.",
+    "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). Each value arrives framed as untrusted request data, and $skill, @file or /command mentions inside it do not act; without a signature, anyone with the URL can trigger a run. For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. To run only for some requests, add filters, all of which must match, e.g. filters:[{path:'headers.x-github-event', op:'equals', value:'pull_request'}, {path:'body.action', op:'in', value:['opened','reopened']}]; ops are equals, notEquals, in, startsWith, matches (regex) and exists. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.",
   parameters: OrchestratorMcpScheduleTaskInput,
   success: OrchestratorMcpScheduleTaskResult,
   failure: OrchestratorMcpFailure,
@@ -143,7 +143,7 @@ const ListScheduledTasksTool = Tool.make("list_scheduled_tasks", {
 
 const UpdateScheduledTaskTool = Tool.make("update_scheduled_task", {
   description:
-    "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run.",
+    "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run. A schedule replaces the whole trigger, so to change a webhook's filters pass its full schedule with the new filters.",
   parameters: OrchestratorMcpUpdateScheduledTaskInput,
   success: OrchestratorMcpScheduleTaskResult,
   failure: OrchestratorMcpFailure,

apps/server/src/scheduledTasks/ScheduledTaskService.ts

@@ -53,6 +53,7 @@ import {
   renderWebhookPrompt,
   type WebhookRequest,
 } from "./webhookTemplate.ts";
+import { describeFailedWebhookFilter, findFailedWebhookFilter } from "./webhookFilter.ts";
 import { constantTimeEquals, verifyWebhookSignature } from "./webhookVerification.ts";
 
 export const WEBHOOK_ROUTE_PREFIX = "/api/hooks";
@@ -114,14 +115,15 @@ export type WebhookDeliveryOutcome =
   | "disabled"
   | "rejected_signature"
   | "expired"
+  | "filtered"
   | "not_found"
   | "error";
 
 export type WebhookTriggerResult =
   | {
       readonly _tag: "accepted";
       readonly deliveryId: ScheduledTaskWebhookDeliveryId;
-      readonly outcome: "accepted" | "duplicate" | "prompt_too_long";
+      readonly outcome: "accepted" | "duplicate" | "prompt_too_long" | "filtered";
     }
   | { readonly _tag: "not_found" }
   | { readonly _tag: "rejected_signature" }
@@ -995,6 +997,7 @@ export const layer = Layer.effect(
                       prefix: input.schedule.signature.prefix,
                     },
               maxDeliveryAgeMinutes: input.schedule.maxDeliveryAgeMinutes ?? null,
+              ...(input.schedule.filters?.length ? { filters: input.schedule.filters } : {}),
             }
           : input.schedule;
       const webhook =
@@ -1560,6 +1563,18 @@ export const layer = Layer.effect(
             return { _tag: "expired" as const };
           }
 
+          // Filters: a request the task does not care about is accepted, so
+          // the sender does not retry it, and logged without starting a run.
+          const failedFilter = findFailedWebhookFilter(schedule.filters ?? [], request);
+          if (failedFilter !== null) {
+            yield* log("filtered", {
+              signatureVerified: signature !== null,
+              error: describeFailedWebhookFilter(failedFilter),
+            });
+            yield* observe("filtered");
+            return { _tag: "accepted" as const, deliveryId, outcome: "filtered" as const };
+          }
+
           const frameId = yield* crypto.randomBytes(6).pipe(
             Effect.map((bytes) => Buffer.from(bytes).toString("hex")),
             Effect.mapError((cause) => taskError("Could not generate frame id.", { cause })),

apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts

@@ -953,3 +953,80 @@ it.effect("counts each handled request by what happened to it", () =>
     }),
   ),
 );
+
+it.effect("checks filters after the signature and max age, and runs only matching deliveries", () =>
+  withService(({ service, launches }) =>
+    Effect.gen(function* () {
+      const { task } = yield* service.upsert(
+        yield* webhookTaskInput({
+          schedule: {
+            type: "webhook",
+            signature: {
+              header: "x-hub-signature-256",
+              encoding: "hex",
+              prefix: "sha256=",
+              secret: "s3cret",
+            },
+            maxDeliveryAgeMinutes: 30,
+            filters: [
+              { path: "headers.x-github-event", op: "equals", value: "pull_request" },
+              { path: "body.action", op: "in", value: ["opened", "reopened"] },
+            ],
+          },
+        }),
+      );
+      const delivery = (
+        event: string,
+        action: string,
+        options: { readonly signed?: boolean; readonly receivedAt?: string } = {},
+      ) => {
+        const body = new TextEncoder().encode(
+          JSON.stringify({ action, pull_request: { url: "https://github.com/org/repo/pull/45" } }),
+        );
+        const signature = `sha256=${NodeCrypto.createHmac("sha256", "s3cret").update(body).digest("hex")}`;
+        return requestFor(task, {
+          headers: {
+            "content-type": "application/json",
+            "x-github-event": event,
+            ...(options.signed === false ? {} : { "x-hub-signature-256": signature }),
+          },
+          body,
+          bodyText: new TextDecoder().decode(body),
+          ...(options.receivedAt === undefined
+            ? {}
+            : { relayDeliveryId: "old", receivedAt: options.receivedAt }),
+        });
+      };
+      const now = yield* DateTime.now;
+      const old = DateTime.formatIso(DateTime.subtract(now, { minutes: 31 }));
+
+      // A request failing a filter is still refused first for its signature or age.
+      const unsigned = yield* service.triggerWebhook(delivery("push", "opened", { signed: false }));
+      assert.equal(unsigned._tag, "rejected_signature");
+      const tooOld = yield* service.triggerWebhook(delivery("push", "opened", { receivedAt: old }));
+      assert.equal(tooOld._tag, "expired");
+
+      const wrongEvent = yield* service.triggerWebhook(delivery("push", "opened"));
+      const wrongAction = yield* service.triggerWebhook(delivery("pull_request", "closed"));
+      assert.deepInclude(wrongEvent, { _tag: "accepted", outcome: "filtered" });
+      assert.deepInclude(wrongAction, { _tag: "accepted", outcome: "filtered" });
+      assert.equal(yield* Queue.size(launches), 0);
+      const [listed] = (yield* service.list()).tasks;
+      assert.equal(listed?.runCount, 0);
+      assert.equal(listed?.lastRunStatus, "never");
+
+      const matched = yield* service.triggerWebhook(delivery("pull_request", "opened"));
+      assert.deepInclude(matched, { _tag: "accepted", outcome: "accepted" });
+      assert.include((yield* Queue.take(launches)).initialMessage?.text ?? "", "pull/45");
+
+      const deliveries = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
+      const filtered = deliveries
+        .filter((entry) => entry.outcome === "filtered")
+        .map((entry) => [entry.signatureVerified, entry.error]);
+      assert.sameDeepMembers(filtered, [
+        [true, "Filter 1 (headers.x-github-event) did not match."],
+        [true, "Filter 2 (body.action) did not match."],
+      ]);
+    }),
+  ),
+);