feat(webapp,db,core): Sessions become run manager

Session is now the run manager for chat.agent and any future task-bound
session. Atomically creates the row + triggers the first run + tracks
the current run via optimistic claim, with a SessionRun audit log for
provenance.

Schema:
- Session gains `taskIdentifier`, `triggerConfig` (JSON), `currentRunId`
  (non-FK), `currentRunVersion` (monotonic int for optimistic claim).
- New SessionRun audit table — one row per run a session triggers,
  with `reason: "initial" | "continuation" | "upgrade" | "manual"`.

Lifecycle:
- `POST /api/v1/sessions`: idempotent on `(env, externalId)`, refreshes
  triggerConfig on cache hit, runs `ensureRunForSession` (probe +
  optimistic claim), returns a session-scoped PAT. JWT auth path
  dropped — secret-key only. The customer's server is the only entry
  point for session creation.
- `POST /api/v1/sessions/:s/end-and-continue`: server-orchestrated
  handoff (cancels current run, triggers a fresh one, swaps
  currentRunId via `updateMany where currentRunVersion`). Powers
  `chat.requestUpgrade()` from inside the agent runtime.
- `POST /realtime/v1/sessions/:s/:io/append`: probe + ensureRunForSession
  before append so messages arriving while no run is alive boot one
  transparently.

Cross-form addressing on write paths:
- `createActionApiRoute` now runs `findResource` before `authorization`,
  matching `createLoaderApiRoute`. Action routes get an optional
  `resource` argument on `authorization.resource()` —
  backwards-compatible (existing 4-arg callbacks unchanged).
- Append + end-and-continue use the new ordering to authorize against
  `{paramSession, friendlyId, externalId}` so a JWT minted for either
  form authorizes either URL form.

Helpers:
- `mintSessionToken.server.ts`: server-side session-PAT factory
  (`read:sessions:{key} + write:sessions:{key}`, 1h TTL).
- `sessionRunManager.server.ts`: `ensureRunForSession` (probe + claim)
  and `swapSessionRun` (force handoff with optimistic claim +
  cancel-on-loss).

Pre-mutation existence reads switched to `$replica` (close, end-and-
continue, PATCH).

Author: ericallam

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -7,7 +7,11 @@ import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { createWaitpointTag, MAX_TAGS_PER_WAITPOINT } from "~/models/waitpointTag.server";
-import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
+import {
+  canonicalSessionAddressingKey,
+  isSessionFriendlyIdForm,
+  resolveSessionByIdOrExternalId,
+} from "~/services/realtime/sessions.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import {
@@ -50,16 +54,23 @@ const { action, loader } = createActionApiRoute(
         return json({ error: "Run not found" }, { status: 404 });
       }
 
-      const session = await resolveSessionByIdOrExternalId(
+      // Row-optional addressing — see the .out / .in.append handlers.
+      // The waitpoint cache + S2 stream key derive from the row's
+      // canonical identity (externalId if set, else friendlyId), so
+      // the agent's wait registration and the append-side drain
+      // converge regardless of which URL form each side used.
+      const maybeSession = await resolveSessionByIdOrExternalId(
         $replica,
         authentication.environment.id,
         body.session
       );
 
-      if (!session) {
+      if (!maybeSession && isSessionFriendlyIdForm(body.session)) {
         return json({ error: "Session not found" }, { status: 404 });
       }
 
+      const addressingKey = canonicalSessionAddressingKey(maybeSession, body.session);
+
       const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
         ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
         : undefined;
@@ -95,11 +106,13 @@ const { action, loader } = createActionApiRoute(
       });
 
       // Step 2: Register the waitpoint on the session channel so the next
-      // append fires it. Keyed by (sessionFriendlyId, io) — both runs on a
-      // multi-tab session wake on the same record.
+      // append fires it. Keyed by (addressingKey, io) — the canonical
+      // string for the row. The append handler drains by the same
+      // canonical key, so writers and readers converge regardless of
+      // which URL form the agent vs. the appending caller used.
       const ttlMs = timeout ? timeout.getTime() - Date.now() : undefined;
       await addSessionStreamWaitpoint(
-        session.friendlyId,
+        addressingKey,
         body.io,
         result.waitpoint.id,
         ttlMs && ttlMs > 0 ? ttlMs : undefined
@@ -117,7 +130,7 @@ const { action, loader } = createActionApiRoute(
 
           if (realtimeStream instanceof S2RealtimeStreams) {
             const records = await realtimeStream.readSessionStreamRecords(
-              session.friendlyId,
+              addressingKey,
               body.io,
               body.lastSeqNum
             );
@@ -135,7 +148,7 @@ const { action, loader } = createActionApiRoute(
               });
 
               await removeSessionStreamWaitpoint(
-                session.friendlyId,
+                addressingKey,
                 body.io,
                 result.waitpoint.id
               );
@@ -146,7 +159,7 @@ const { action, loader } = createActionApiRoute(
           // will complete the waitpoint via the append handler path. Log so
           // a broken race-check doesn't silently degrade to timeout-only.
           logger.warn("session-stream wait race-check failed", {
-            sessionFriendlyId: session.friendlyId,
+            addressingKey,
             io: body.io,
             waitpointId: WaitpointId.toFriendlyId(result.waitpoint.id),
             error,

apps/webapp/app/routes/api.v1.sessions.$session.close.ts

@@ -4,7 +4,7 @@ import {
   type RetrieveSessionResponseBody,
 } from "@trigger.dev/core/v3";
 import { z } from "zod";
-import { prisma } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import {
   resolveSessionByIdOrExternalId,
   serializeSession,
@@ -31,7 +31,7 @@ const { action, loader } = createActionApiRoute(
   },
   async ({ authentication, params, body }) => {
     const existing = await resolveSessionByIdOrExternalId(
-      prisma,
+      $replica,
       authentication.environment.id,
       params.session
     );

apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts

@@ -0,0 +1,132 @@
+import { json } from "@remix-run/server-runtime";
+import {
+  EndAndContinueSessionRequestBody,
+  type EndAndContinueSessionResponseBody,
+} from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { swapSessionRun } from "~/services/realtime/sessionRunManager.server";
+import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  session: z.string(),
+});
+
+// POST /api/v1/sessions/:session/end-and-continue
+//
+// Generic "the running run is exiting; please trigger a fresh one for
+// this session and swap `currentRunId` to it" endpoint. The agent calls
+// this from `chat.requestUpgrade` and other planned-handoff paths. The
+// transport's `.out` SSE keeps streaming across the swap because S2 is
+// keyed on the session, not the run — v1's last chunks land, v2's new
+// chunks land on the same stream.
+//
+// Auth: `write:sessions:{ext}` — the running agent's internal API key
+// (PRIVATE) bypasses authorization; a browser holding the session PAT
+// can also reach this endpoint, which is fine: if you have the session
+// PAT, you own the chat.
+const { action } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    body: EndAndContinueSessionRequestBody,
+    method: "POST",
+    maxContentLength: 1024,
+    allowJWT: true,
+    corsStrategy: "all",
+    // Resolved before authorization so the auth scope can expand to both
+    // addressing forms (friendlyId + externalId). Handler reads the row
+    // from `resource` instead of re-fetching.
+    findResource: async (params, auth) =>
+      resolveSessionByIdOrExternalId($replica, auth.environment.id, params.session),
+    authorization: {
+      action: "write",
+      resource: (params, _, __, ___, session) => {
+        const ids = new Set<string>([params.session]);
+        if (session) {
+          ids.add(session.friendlyId);
+          if (session.externalId) ids.add(session.externalId);
+        }
+        return { sessions: [...ids] };
+      },
+      superScopes: ["write:sessions", "write:all", "admin"],
+    },
+  },
+  async ({ authentication, params, body, resource: session }) => {
+    if (!session) {
+      // Unreachable — `findResource` 404s before this runs. Type narrow.
+      return json({ error: "Session not found" }, { status: 404 });
+    }
+
+    if (session.closedAt) {
+      return json(
+        { error: "Cannot end-and-continue a closed session" },
+        { status: 400 }
+      );
+    }
+
+    if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
+      return json(
+        { error: "Cannot end-and-continue an expired session" },
+        { status: 400 }
+      );
+    }
+
+    // The wire `callingRunId` is a friendlyId (that's what the agent
+    // SDK exposes via `ctx.run.id`). Internally `Session.currentRunId`
+    // stores the TaskRun.id cuid, so resolve before handing to the
+    // optimistic-claim service.
+    const callingRun = await $replica.taskRun.findFirst({
+      where: {
+        friendlyId: body.callingRunId,
+        runtimeEnvironmentId: authentication.environment.id,
+      },
+      select: { id: true },
+    });
+    if (!callingRun) {
+      return json({ error: "callingRunId not found in this environment" }, { status: 404 });
+    }
+
+    try {
+      // Body's `reason` is free-form for forward-compat (audit metadata
+      // only); narrow into the closed `EnsureRunReason` set, defaulting
+      // to `"manual"` for unknown labels.
+      const reason: "initial" | "continuation" | "upgrade" | "manual" =
+        body.reason === "upgrade" ||
+        body.reason === "continuation" ||
+        body.reason === "initial" ||
+        body.reason === "manual"
+          ? body.reason
+          : "manual";
+
+      const result = await swapSessionRun({
+        session,
+        callingRunId: callingRun.id,
+        environment: authentication.environment,
+        reason,
+      });
+
+      // The swap stored a TaskRun.id (cuid) in `currentRunId`; surface
+      // the friendlyId for parity with the rest of the public API.
+      const run = await $replica.taskRun.findFirst({
+        where: { id: result.runId },
+        select: { friendlyId: true },
+      });
+
+      const responseBody: EndAndContinueSessionResponseBody = {
+        runId: run?.friendlyId ?? result.runId,
+        swapped: result.swapped,
+      };
+      return json<EndAndContinueSessionResponseBody>(responseBody);
+    } catch (error) {
+      logger.error("Failed end-and-continue", {
+        sessionId: session.id,
+        error,
+      });
+      return json({ error: "Failed to swap session run" }, { status: 500 });
+    }
+  }
+);
+
+export { action };

apps/webapp/app/routes/api.v1.sessions.$session.ts

@@ -54,7 +54,7 @@ const { action } = createActionApiRoute(
   },
   async ({ authentication, params, body }) => {
     const existing = await resolveSessionByIdOrExternalId(
-      prisma,
+      $replica,
       authentication.environment.id,
       params.session
     );