@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
+---
+
+Route verified webhooks into agent sessions with `chat.event({ endpoint, key })`, and connect external chat surfaces with `chat.channels.custom({ endpoint })`. Both subscribe to a shared webhook endpoint, so an agent and your other webhook tasks can use the same provider URL. Channels support reply delivery, session binding, conversation seeding, and lifecycle reactions, and a failed turn posts a short line instead of the error (`errorMessage` sets it, as a string or a function of the failure). Use `chat.createWatchToken()` to watch a session from another surface.
+
+A connector can also stream replies natively (`stream()`), mirror the session's state into the surface (`lifecycle()`), show tool calls as they run (`progress()`), stop a running turn from the surface (`onControl()`), post approvals and tool controls and resume the run from a verified click, download files only the surface can read (`download()`), and acknowledge or fold in messages that arrive mid-turn (`received()`, `pendingMessages`).
@@ -0,0 +1,25 @@
+---
+"@trigger.dev/core": minor
+"@trigger.dev/sdk": minor
+"trigger.dev": minor
+---
+
+Add hosted webhooks: declare an endpoint once with `webhooks.endpoint.define({ id, source })` and Trigger.dev hosts its URL, verifies every request, and routes each event to the endpoint's subscribers. Subscribe tasks with `webhook({ endpoint, filter, onEvent })`, each with its own type-safe filter, and pause a run until a matching webhook arrives with `webhooks.waiter.create()` and `webhooks.waiter.wait()`.
+
+```ts
+export const stripe = webhooks.endpoint.define({ id: "stripe", source: webhooks.stripe() });
+
+export const fulfill = webhook({
+  id: "fulfill",
+  endpoint: stripe,
+  filter: "event.type == 'checkout.session.completed'",
+  onEvent: async ({ event }) => {},
+});
+
+const waiter = await webhooks.waiter.create(stripe, {
+  match: { "event.data.object.metadata.orderId": order.id },
+});
+const { event } = await webhooks.waiter.wait(waiter).unwrap();
+```
+
+Sources cover presets (`webhooks.stripe()`, `webhooks.github()`, `webhooks.svix()`, `webhooks.discord()`, and others), named providers (`webhooks.clerk()`, `webhooks.shopify()`, and more) and `webhooks.custom<T>(config)`, and most carry a setup prompt an AI coding agent can follow to register the endpoint. Retries are deduplicated on the provider's own event or delivery id where it sends one. Filters, keys and waiter `match` paths share the `event.`, `header.` and `webhook.` namespaces, and filters accept `== null` / `!= null` presence checks and `event any (...)` over a payload that is an array of events. `webhook()` takes `task()` options such as `queue`, `machine` and `retry`. The HTTP API lists endpoints and deliveries, replays deliveries, and creates and cancels waiters, and both it and the MCP server can read an endpoint's setup prompt and generate or store its signing secret.
@@ -10,7 +10,7 @@ import {
   resolveSessionWithWriterFallback,
 } from "~/services/realtime/sessions.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
-import { stripClientWebhookActionSource } from "~/services/realtime/sanitizeSessionInput.server";
+import { stripClientIngressFields } from "~/services/realtime/sanitizeSessionInput.server";
 import {
   claimSessionStreamPart,
   releaseSessionStreamPart,
@@ -150,7 +150,7 @@ const { action, loader } = createActionApiRoute(
 
     let part = await request.text();
     if (params.io === "in") {
-      part = stripClientWebhookActionSource(part);
+      part = stripClientIngressFields(part);
     }
 
     const clientPartId = request.headers.get("X-Part-Id");
 
@@ -36,12 +36,13 @@ const { action } = createActionApiRoute(
     },
   },
   async ({ params, authentication }) => {
-    // `.out` is the agent→client channel. Only PRIVATE (secret key) auth —
-    // i.e. the agent run itself — may initialize it. Session-scoped JWTs carry
-    // `write:sessions:<key>` for `.in`; without this gate they could obtain
-    // credentials to forge assistant chunks on their own session's `.out`.
-    if (params.io === "out" && authentication.type !== "PRIVATE") {
-      return new Response("Initializing the out channel requires secret key authentication", {
+    // Only PRIVATE (secret key) auth may initialize a channel: the S2 credentials
+    // this returns write raw records. Session-scoped JWTs carry
+    // `write:sessions:<key>`; with them they could forge assistant chunks on
+    // `.out`, or skip the `.in` append route's removal of ingress-only fields.
+    // Clients write `.in` through that append route.
+    if (authentication.type !== "PRIVATE") {
+      return new Response("Initializing a session channel requires secret key authentication", {
         status: 403,
       });
     }
 
@@ -18,6 +18,7 @@ import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
 import { engine } from "~/v3/runEngine.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
+import { stripClientIngressFields } from "~/services/realtime/sanitizeSessionInput.server";
 
 const ParamsSchema = z.object({
   session: z.string(),
@@ -106,7 +107,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
 
-  const part = await request.text();
+  const body = await request.text();
+  const part = io === "in" ? stripClientIngressFields(body) : body;
   const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
 
   const [appendError, appendSeq] = await tryCatch(
 
@@ -1,17 +1,18 @@
 /**
- * Strip a client-forged `actionSource: "webhook"` from a session `.in` append part.
+ * Strip the fields only the hosted webhook ingress may set from a session `.in` append part.
  *
- * Only the hosted webhook ingress may claim webhook trust, and it appends server-side rather than
- * through the client append route. A client with session write access could otherwise send a record
- * carrying `actionSource: "webhook"`, which the run loop uses to skip action-schema validation. We
- * downgrade it here (delete the field) so the record is validated as a normal client action.
+ * The ingress appends server-side, never through a client append route. A client with session write
+ * access could otherwise send a record carrying `actionSource: "webhook"`, which the run loop uses to
+ * skip action-schema validation, or a `channelEvent`, which the run loop treats as a verified provider
+ * delivery and hands to the channel connector. We delete both, so an action is validated as a normal
+ * client action and a message is a normal client message.
  *
- * The part is always parsed rather than string-matched: JSON allows the property name to be spelled
+ * The part is always parsed rather than string-matched: JSON allows a property name to be spelled
  * with escapes (`"action\u0053ource"`), which a substring check misses and `JSON.parse` canonicalizes
  * to the same key the runtime reads. A part that does not parse is passed through unchanged; the
  * runtime rejects it as malformed.
  */
-export function stripClientWebhookActionSource(part: string): string {
+export function stripClientIngressFields(part: string): string {
   let record: unknown;
   try {
     record = JSON.parse(part);
@@ -20,14 +21,15 @@ export function stripClientWebhookActionSource(part: string): string {
   }
   if (!record || typeof record !== "object") return part;
 
-  const payload = (record as { payload?: unknown }).payload;
+  const payload = (record as { payload?: unknown }).payload as Record<string, unknown> | undefined;
   if (!payload || typeof payload !== "object") return part;
-  if (!Object.hasOwn(payload, "actionSource")) return part;
 
-  if ((payload as { actionSource?: unknown }).actionSource === "webhook") {
-    delete (payload as { actionSource?: unknown }).actionSource;
-    return JSON.stringify(record);
-  }
+  const forgedWebhookAction =
+    Object.hasOwn(payload, "actionSource") && payload.actionSource === "webhook";
+  const forgedChannelEvent = Object.hasOwn(payload, "channelEvent");
+  if (!forgedWebhookAction && !forgedChannelEvent) return part;
 
-  return part;
+  if (forgedWebhookAction) delete payload.actionSource;
+  if (forgedChannelEvent) delete payload.channelEvent;
+  return JSON.stringify(record);
 }
@@ -3013,7 +3013,7 @@ describe("API", () => {
         expect(res.status).not.toBe(403);
       });
 
-      it("PUT with write:sessions (was a superScope): auth passes", async () => {
+      it("PUT with write:sessions: authorized, but raw channel credentials need a secret key", async () => {
         const server = getTestServer();
         const seed = await seedTestEnvironment(server.prisma);
         const session = await seedTestApiSession(server.prisma, seed.environment);
@@ -3022,6 +3022,20 @@ describe("API", () => {
           method: "PUT",
           headers: { Authorization: `Bearer ${jwt}` },
         });
+        expect(res.status).toBe(403);
+        expect(await res.text()).toBe(
+          "Initializing a session channel requires secret key authentication"
+        );
+      });
+
+      it("PUT with a secret key: passes", async () => {
+        const server = getTestServer();
+        const seed = await seedTestEnvironment(server.prisma);
+        const session = await seedTestApiSession(server.prisma, seed.environment);
+        const res = await server.webapp.fetch(ioPath(session.friendlyId, "in"), {
+          method: "PUT",
+          headers: { Authorization: `Bearer ${seed.apiKey}` },
+        });
         expect(res.status).not.toBe(401);
         expect(res.status).not.toBe(403);
       });
 
@@ -1,9 +1,9 @@
 import { describe, expect, it } from "vitest";
-import { stripClientWebhookActionSource } from "~/services/realtime/sanitizeSessionInput.server";
+import { stripClientIngressFields } from "~/services/realtime/sanitizeSessionInput.server";
 
 const record = (payload: Record<string, unknown>) => JSON.stringify({ kind: "message", payload });
 
-describe("stripClientWebhookActionSource", () => {
+describe("stripClientIngressFields", () => {
   it("removes a client-forged webhook actionSource so the action is validated normally", () => {
     const forged = record({
       chatId: "c1",
@@ -12,33 +12,55 @@ describe("stripClientWebhookActionSource", () => {
       action: { type: "refund", amount: 9999 },
     });
 
-    const cleaned = JSON.parse(stripClientWebhookActionSource(forged));
+    const cleaned = JSON.parse(stripClientIngressFields(forged));
     expect(cleaned.payload.actionSource).toBeUndefined();
     expect(cleaned.payload.action).toEqual({ type: "refund", amount: 9999 });
     expect(cleaned.payload.trigger).toBe("action");
   });
 
   it("leaves a non-webhook actionSource untouched", () => {
     const part = record({ trigger: "action", actionSource: "client", action: { type: "ping" } });
-    expect(stripClientWebhookActionSource(part)).toBe(part);
+    expect(stripClientIngressFields(part)).toBe(part);
   });
 
   it("removes a webhook actionSource whose property name is spelled with JSON escapes", () => {
     const forged =
       '{"kind":"message","payload":{"chatId":"c1","trigger":"action","action\\u0053ource":"webhook","action":{"type":"refund"}}}';
-    const cleaned = JSON.parse(stripClientWebhookActionSource(forged));
+    const cleaned = JSON.parse(stripClientIngressFields(forged));
     expect(cleaned.payload.actionSource).toBeUndefined();
     expect(cleaned.payload.action).toEqual({ type: "refund" });
     expect(cleaned.payload.trigger).toBe("action");
   });
 
+  it("removes a client-forged channelEvent so the message is a normal client message", () => {
+    const forged = record({
+      chatId: "c1",
+      trigger: "submit-message",
+      message: { role: "user", parts: [{ type: "text", text: "hi" }] },
+      channelEvent: { connectorId: "slack", event: { type: "event_callback" } },
+    });
+
+    const cleaned = JSON.parse(stripClientIngressFields(forged));
+    expect(cleaned.payload.channelEvent).toBeUndefined();
+    expect(cleaned.payload.message).toEqual({
+      role: "user",
+      parts: [{ type: "text", text: "hi" }],
+    });
+  });
+
+  it("removes a channelEvent whose property name is spelled with JSON escapes", () => {
+    const forged =
+      '{"kind":"message","payload":{"trigger":"submit-message","channel\\u0045vent":{"connectorId":"slack","event":{}}}}';
+    expect(JSON.parse(stripClientIngressFields(forged)).payload.channelEvent).toBeUndefined();
+  });
+
   it("leaves a normal message part untouched", () => {
     const part = record({ trigger: "submit-message", message: { role: "user", parts: [] } });
-    expect(stripClientWebhookActionSource(part)).toBe(part);
+    expect(stripClientIngressFields(part)).toBe(part);
   });
 
   it("leaves a malformed part untouched", () => {
     const part = '{"kind":"message","payload":{"actionSource":"webhook"';
-    expect(stripClientWebhookActionSource(part)).toBe(part);
+    expect(stripClientIngressFields(part)).toBe(part);
   });
 });
@@ -68,6 +68,49 @@ describe("parseFilter", () => {
     }
   });
 
+  it("rejects empty path segments in fields, references, and quantifiers", () => {
+    for (const filter of [
+      "event == 1",
+      "event. == 1",
+      "event..amount >= 1",
+      "event.amount. >= 1",
+      "event.data.object..amount >= 1",
+      "header. == 'x'",
+      "header..name == 'x'",
+      "event.amount >= event..limit",
+      "event.amount >= webhook.",
+      "event..items any ( amount >= 1 )",
+      "event.items any ( amount. >= 1 )",
+      "event.items any ( item..amount >= 1 )",
+    ]) {
+      expect(() => parseFilter(filter), filter).toThrow(FilterParseError);
+    }
+  });
+
+  it("quantifies over the event itself when the body is an array", () => {
+    const ast = parseFilter("event any ( type == 'nodeCreated' )");
+    expect(ast).toEqual({
+      kind: "quantifier",
+      mode: "any",
+      path: "event",
+      clause: { path: "type", op: "eq", value: "nodeCreated" },
+    });
+    const event = [{ type: "nodeNeedsApproval" }, { type: "nodeCreated" }];
+    const ctx = { event, headers: {}, webhook: {} };
+    expect(evaluateFilter(ast, ctx).match).toBe(true);
+    expect(evaluateFilter(parseFilter("event all ( type == 'nodeCreated' )"), ctx).match).toBe(
+      false
+    );
+    expect(evaluateFilter(ast, { ...ctx, event: { type: "nodeCreated" } }).match).toBe(false);
+    for (const filter of [
+      "event == 'x'",
+      "header any ( type == 'x' )",
+      "webhook any ( id == 'x' )",
+    ]) {
+      expect(() => parseFilter(filter), filter).toThrow(FilterParseError);
+    }
+  });
+
   it("enforces the soft clause cap", () => {
     const big = Array.from({ length: 60 }, () => "event.action == 'created'").join(" && ");
     expect(() => parseFilter(big)).toThrow(/too complex/);
 
@@ -21,6 +21,16 @@ const WORD_OPS: Record<string, FilterOp> = {
 };
 const NAMESPACES = new Set(["event", "header", "webhook"]);
 
+function validatePath(path: string, namespaced = true): void {
+  const segments = path.split(".");
+  if (segments.some((segment) => segment.length === 0) || (namespaced && segments.length < 2)) {
+    throw new FilterParseError(`invalid field path "${path}"`);
+  }
+  if (namespaced && !NAMESPACES.has(segments[0])) {
+    throw new FilterParseError(`field must start with event./header./webhook., got "${path}"`);
+  }
+}
+
 type Tok = {
   t:
     | "lparen"
@@ -148,24 +158,24 @@ export function parseFilter(input: string): FilterAst {
     const pathTok = next();
     if (!pathTok || pathTok.t !== "word") throw new FilterParseError("expected a field path");
     const path = pathTok.v;
-    if (!NAMESPACES.has(path.split(".")[0])) {
-      throw new FilterParseError(`field must start with event./header./webhook., got "${path}"`);
-    }
     // Quantifier: `arrPath any|all ( subPath op value )` — sub-clause resolves against each element.
     const q = peek();
-    if (q?.t === "word" && (q.v === "any" || q.v === "all")) {
+    const mode = q?.t === "word" && (q.v === "any" || q.v === "all") ? q.v : undefined;
+    if (!(mode && path === "event")) validatePath(path);
+    if (mode) {
       next();
-      if (next()?.t !== "lparen") throw new FilterParseError(`expected "(" after "${q.v}"`);
+      if (next()?.t !== "lparen") throw new FilterParseError(`expected "(" after "${mode}"`);
       const subPathTok = next();
       if (!subPathTok || subPathTok.t !== "word") {
         throw new FilterParseError("expected a field path inside the quantifier");
       }
+      validatePath(subPathTok.v, false);
       const subOp = parseOp(subPathTok.v);
       const subValue = parseScalar();
       if (next()?.t !== "rparen") throw new FilterParseError('expected ")"');
       return {
         kind: "quantifier",
-        mode: q.v,
+        mode,
         path,
         clause: { path: subPathTok.v, op: subOp, value: subValue },
       };
@@ -175,6 +185,7 @@ export function parseFilter(input: string): FilterAst {
     const operand = peek();
     if (operand?.t === "word" && NAMESPACES.has(operand.v.split(".")[0])) {
       next();
+      validatePath(operand.v);
       return { kind: "clause", path, op, valueRef: operand.v };
     }
     return { kind: "clause", path, op, value: parseValue() };