@@ -0,0 +1,21 @@
+---
+"@trigger.dev/sdk": minor
+"@trigger.dev/core": patch
+---
+
+End a chat conversation from inside the agent with `chat.close({ reason })`. The session row is closed, further sends are refused with HTTP 409, and the run exits without scheduling a continuation, so a budget cap, a completed goal, or a signed-out user can stop the conversation rather than only the current run.
+
+```ts
+chat.agent({
+  id: "budgeted-agent",
+  run: async ({ messages, signal }) =>
+    streamText({ model: openai("gpt-4o"), messages, abortSignal: signal }),
+  onBeforeTurnComplete: async ({ chatId }) => {
+    if (await overBudget(chatId)) {
+      chat.close({ reason: "Monthly budget reached" });
+    }
+  },
+});
+```
+
+The current turn still streams in full. Decide the close before the turn ends (`run()`, `prepareStep`, `onBeforeTurnComplete`) so the closed state rides out on that turn's final record and the user sees it as soon as the answer finishes. `TriggerChatTransport` picks the close up from the response stream or from a refused send, exposes it as `transport.sessionStatus(chatId)` plus `transport.sessionClosedReason(chatId)`, and stops sending and reconnecting. Closing a session from outside with `sessions.close()` now also reaches a live run, so an idle or suspended agent exits on its next wake instead of waiting out its idle timeout. Writes to a closed session's named side channels are refused with the same 409.
@@ -1,12 +1,19 @@
 import { json } from "@remix-run/server-runtime";
+import { tryCatch } from "@trigger.dev/core/utils";
 import { CloseSessionRequestBody, type RetrieveSessionResponseBody } from "@trigger.dev/core/v3";
+import type { Session } from "@trigger.dev/database";
 import { z } from "zod";
 import { $replica, prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { appendServerSessionRecord } from "~/services/realtime/sessionChannelAppend.server";
 import {
+  canonicalSessionAddressingKey,
   resolveSessionByIdOrExternalId,
   serializeSessionWithFriendlyRunId,
 } from "~/services/realtime/sessions.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { runStore } from "~/v3/runStore.server";
 
 const ParamsSchema = z.object({
   session: z.string(),
@@ -62,8 +69,83 @@ const { action, loader } = createActionApiRoute(
 
     const updated = await prisma.session.findFirst({ where: { id: existing.id } });
     if (!updated) return json({ error: "Session not found" }, { status: 404 });
+
+    // Best effort in full: the row is closed and that is what the caller asked
+    // for. A failure resolving the calling run or building the stream client
+    // must not turn a successful close into a 500.
+    const [notifyError] = await tryCatch(
+      notifyLiveRunOfClose({
+        session: updated,
+        environment: authentication.environment,
+        paramSession: params.session,
+        callingRunId: body.callingRunId,
+        reason: body.reason,
+      })
+    );
+    if (notifyError) {
+      logger.error("Failed to notify the live run of a session close", {
+        sessionId: updated.id,
+        error: notifyError,
+      });
+    }
+
     return json<RetrieveSessionResponseBody>(await serializeSessionWithFriendlyRunId(updated));
   }
 );
 
+/**
+ * Put a `trigger: "close"` record on the session's `.in` channel so a live run
+ * leaves its loop now instead of sitting until its idle timeout. Appending also
+ * completes the run's `.in` waitpoint, which is what wakes a suspended run.
+ *
+ * Skipped when the close came from the session's own run (`chat.close()`), which
+ * is already exiting and would never consume the record. Best effort throughout:
+ * the row is closed either way, and further appends are refused.
+ */
+async function notifyLiveRunOfClose({
+  session,
+  environment,
+  paramSession,
+  callingRunId,
+  reason,
+}: {
+  session: Session;
+  environment: AuthenticatedEnvironment;
+  paramSession: string;
+  callingRunId?: string;
+  reason?: string;
+}): Promise<void> {
+  if (!session.currentRunId) {
+    return;
+  }
+
+  if (callingRunId) {
+    const callingRun = await runStore.findRun(
+      { friendlyId: callingRunId, runtimeEnvironmentId: environment.id },
+      { select: { id: true } },
+      $replica
+    );
+    if (callingRun?.id === session.currentRunId) {
+      return;
+    }
+  }
+
+  const record = JSON.stringify({
+    kind: "message",
+    payload: {
+      chatId: session.externalId ?? session.friendlyId,
+      trigger: "close",
+      ...(reason ? { closedReason: reason } : {}),
+    },
+  });
+
+  await appendServerSessionRecord({
+    environment,
+    session,
+    addressingKey: canonicalSessionAddressingKey(session, paramSession),
+    io: "in",
+    part: record,
+  });
+}
+
 export { action, loader };
@@ -13,11 +13,10 @@ import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.s
 import { stripClientWebhookActionSource } from "~/services/realtime/sanitizeSessionInput.server";
 import {
   claimSessionStreamPart,
-  drainSessionStreamWaitpoints,
   releaseSessionStreamPart,
 } from "~/services/sessionStreamWaitpointCache.server";
+import { completeSessionStreamWaitpoints } from "~/services/realtime/sessionChannelAppend.server";
 import { anyResource, createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { engine } from "~/v3/runEngine.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
 
 const ParamsSchema = z.object({
@@ -80,11 +79,22 @@ const { action, loader } = createActionApiRoute(
     }
 
     if (session.closedAt) {
-      return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
+      return json(
+        {
+          ok: false,
+          error: "Cannot append to a closed session",
+          code: "session_closed",
+          closedReason: session.closedReason,
+        },
+        { status: 409 }
+      );
     }
 
     if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
-      return json({ ok: false, error: "Cannot append to an expired session" }, { status: 400 });
+      return json(
+        { ok: false, error: "Cannot append to an expired session", code: "session_expired" },
+        { status: 400 }
+      );
     }
 
     // `.out` is the agent→client channel. Only PRIVATE (secret key) auth —
@@ -195,45 +205,12 @@ const { action, loader } = createActionApiRoute(
       }
     }
 
-    // Fire any run-scoped waitpoints registered against this channel. Best
-    // effort — a failure here must not fail the append (the record is
-    // durable in S2; the SSE tail will still deliver it). Waitpoints are
-    // keyed on the canonical addressing key the agent registered with via
-    // `sessions.open(...).in.wait()`, so writers and readers converge
-    // regardless of which URL form they used.
-    const [drainError, waitpointIds] = await tryCatch(
-      drainSessionStreamWaitpoints(authentication.environment.id, addressingKey, params.io)
+    await completeSessionStreamWaitpoints(
+      authentication.environment.id,
+      addressingKey,
+      params.io,
+      part
     );
-    if (drainError) {
-      logger.error("Failed to drain session stream waitpoints", {
-        addressingKey,
-        io: params.io,
-        error: drainError,
-      });
-    } else if (waitpointIds && waitpointIds.length > 0) {
-      await Promise.all(
-        waitpointIds.map(async (waitpointId) => {
-          const [completeError] = await tryCatch(
-            engine.completeWaitpoint({
-              id: waitpointId,
-              output: {
-                value: part,
-                type: "application/json",
-                isError: false,
-              },
-            })
-          );
-          if (completeError) {
-            logger.error("Failed to complete session stream waitpoint", {
-              addressingKey,
-              io: params.io,
-              waitpointId,
-              error: completeError,
-            });
-          }
-        })
-      );
-    }
 
     // `seq` lets the client correlate this send to the turn that consumes it.
     return json({ ok: true, seq: appendSeq }, { status: 200 });
 
@@ -55,7 +55,15 @@ const { action, loader } = createActionApiRoute(
     }
 
     if (session.closedAt) {
-      return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
+      return json(
+        {
+          ok: false,
+          error: "Cannot append to a closed session",
+          code: "session_closed",
+          closedReason: session.closedReason,
+        },
+        { status: 409 }
+      );
     }
 
     if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
 
@@ -62,7 +62,15 @@ export async function action({ request, params }: ActionFunctionArgs) {
   }
 
   if (session.closedAt) {
-    return json({ ok: false, error: "Cannot append to a closed session" }, { status: 400 });
+    return json(
+      {
+        ok: false,
+        error: "Cannot append to a closed session",
+        code: "session_closed",
+        closedReason: session.closedReason,
+      },
+      { status: 409 }
+    );
   }
 
   if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
 
@@ -0,0 +1,111 @@
+import { tryCatch } from "@trigger.dev/core/utils";
+import { nanoid } from "nanoid";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { drainSessionStreamWaitpoints } from "~/services/sessionStreamWaitpointCache.server";
+import { engine } from "~/v3/runEngine.server";
+
+type SessionChannelIO = "out" | "in";
+
+/**
+ * Fire any run-scoped waitpoints registered against a session channel, so a
+ * suspended run wakes on the record that just landed. Best effort: a failure
+ * here must not fail the append, because the record is durable in S2 and the
+ * SSE tail still delivers it. Waitpoints are keyed on the canonical addressing
+ * key the agent registered with via `sessions.open(...).in.wait()`, so writers
+ * and readers converge regardless of which URL form they used.
+ */
+export async function completeSessionStreamWaitpoints(
+  environmentId: string,
+  addressingKey: string,
+  io: SessionChannelIO,
+  part: string
+): Promise<void> {
+  const [drainError, waitpointIds] = await tryCatch(
+    drainSessionStreamWaitpoints(environmentId, addressingKey, io)
+  );
+
+  if (drainError) {
+    logger.error("Failed to drain session stream waitpoints", {
+      addressingKey,
+      io,
+      error: drainError,
+    });
+    return;
+  }
+
+  if (!waitpointIds || waitpointIds.length === 0) {
+    return;
+  }
+
+  await Promise.all(
+    waitpointIds.map(async (waitpointId) => {
+      const [completeError] = await tryCatch(
+        engine.completeWaitpoint({
+          id: waitpointId,
+          output: {
+            value: part,
+            type: "application/json",
+            isError: false,
+          },
+        })
+      );
+      if (completeError) {
+        logger.error("Failed to complete session stream waitpoint", {
+          addressingKey,
+          io,
+          waitpointId,
+          error: completeError,
+        });
+      }
+    })
+  );
+}
+
+/**
+ * Append one server-originated record to a session channel and wake anything
+ * waiting on it. Unlike the client-facing append route there is no idempotency
+ * claim: the server is the only writer of these records and the caller has
+ * already made its own state change conditional.
+ *
+ * Returns false when the environment isn't on the S2 backend (session channels
+ * don't exist there) or the append failed. Callers treat that as best effort.
+ */
+export async function appendServerSessionRecord({
+  environment,
+  session,
+  addressingKey,
+  io,
+  part,
+}: {
+  environment: AuthenticatedEnvironment;
+  session: { id: string; streamBasinName: string | null };
+  addressingKey: string;
+  io: SessionChannelIO;
+  part: string;
+}): Promise<boolean> {
+  const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
+
+  if (!(realtimeStream instanceof S2RealtimeStreams)) {
+    return false;
+  }
+
+  const [appendError] = await tryCatch(
+    realtimeStream.appendPartToSessionStream(part, nanoid(7), addressingKey, io)
+  );
+
+  if (appendError) {
+    logger.error("Failed to append server session record", {
+      sessionId: session.id,
+      addressingKey,
+      io,
+      error: appendError,
+    });
+    return false;
+  }
+
+  await completeSessionStreamWaitpoints(environment.id, addressingKey, io, part);
+  return true;
+}