From 49662b22a2be8d87919ad79b695c9724981e5e77 Mon Sep 17 00:00:00 2001
From: Anthony Ettinger <anthony@profullstack.com>
Date: Wed, 11 Nov 2026 08:12:39 +0000
Subject: [PATCH] feat(api): agent broadcasts - an account agent goes live on a
 channel

POST /api/v1/agent/broadcasts (Bearer pux_ag_) opens a public SFU room on a
channel the agent's owner may use (can_use_resource), with the owner as host,
and returns a publish-only LiveKit token whose identity is that host, so the
webhook's host-left path ends the live when the publisher disconnects.
POST /:id is the heartbeat the listings need, DELETE /:id ends it. Go-live
notifications and channel restreams fire as for a person's live.

For streaming podcasts (TUI Weekly, Off The Record) to pairux.com/@moshcoding
as they are recorded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../v1/agent/broadcasts/[sessionId]/route.ts  |  44 ++++
 .../src/app/api/v1/agent/broadcasts/route.ts  |  63 +++++
 apps/web/src/app/docs/agents/page.tsx         |  14 +-
 apps/web/src/lib/agent-broadcast.test.ts      | 174 +++++++++++++
 apps/web/src/lib/agent-broadcast.ts           | 244 ++++++++++++++++++
 5 files changed, 536 insertions(+), 3 deletions(-)
 create mode 100644 apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts
 create mode 100644 apps/web/src/app/api/v1/agent/broadcasts/route.ts
 create mode 100644 apps/web/src/lib/agent-broadcast.test.ts
 create mode 100644 apps/web/src/lib/agent-broadcast.ts

diff --git a/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts b/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts
new file mode 100644
index 00000000..41762846
--- /dev/null
+++ b/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts
@@ -0,0 +1,44 @@
+import { z } from 'zod';
+import { errorResponse, handleApiError, successResponse } from '@/lib/api';
+import { AccountAgents } from '@/lib/account-agents';
+import { BroadcastError, endBroadcast, heartbeatBroadcast } from '@/lib/agent-broadcast';
+import { serviceClient } from '@/lib/supabase/service';
+
+export const runtime = 'nodejs';
+export const dynamic = 'force-dynamic';
+
+interface RouteParams {
+  params: Promise<{ sessionId: string }>;
+}
+
+const idSchema = z.guid('Invalid session ID');
+
+/** POST /api/v1/agent/broadcasts/:id — heartbeat; { live } says whether it still is. */
+export async function POST(request: Request, { params }: RouteParams) {
+  try {
+    const agent = await new AccountAgents().fromRequest(request);
+    if (!agent) return errorResponse('Send Authorization: Bearer <pux_ag_ agent key>.', 401);
+    const sessionId = idSchema.parse((await params).sessionId);
+    return successResponse(await heartbeatBroadcast(serviceClient(), agent, sessionId));
+  } catch (error) {
+    if (error instanceof BroadcastError) return errorResponse(error.message, error.status);
+    return handleApiError(error);
+  }
+}
+
+/** DELETE /api/v1/agent/broadcasts/:id — the broadcast is over. */
+export async function DELETE(request: Request, { params }: RouteParams) {
+  try {
+    const agent = await new AccountAgents().fromRequest(request);
+    if (!agent) return errorResponse('Send Authorization: Bearer <pux_ag_ agent key>.', 401);
+    const sessionId = idSchema.parse((await params).sessionId);
+    const result = await endBroadcast(serviceClient(), agent, sessionId);
+    void import('@/lib/channel-restream').then(({ stopChannelRestream }) =>
+      stopChannelRestream(sessionId)
+    );
+    return successResponse(result);
+  } catch (error) {
+    if (error instanceof BroadcastError) return errorResponse(error.message, error.status);
+    return handleApiError(error);
+  }
+}
diff --git a/apps/web/src/app/api/v1/agent/broadcasts/route.ts b/apps/web/src/app/api/v1/agent/broadcasts/route.ts
new file mode 100644
index 00000000..33503cb3
--- /dev/null
+++ b/apps/web/src/app/api/v1/agent/broadcasts/route.ts
@@ -0,0 +1,63 @@
+import { errorResponse, handleApiError, successResponse } from '@/lib/api';
+import { AccountAgents } from '@/lib/account-agents';
+import { BroadcastError, startBroadcast, startBroadcastSchema } from '@/lib/agent-broadcast';
+import { serviceClient } from '@/lib/supabase/service';
+import { getIceServers } from '@/lib/ice-servers';
+import { FixedWindowRateLimiter } from '@/lib/rate-limit';
+
+export const runtime = 'nodejs';
+export const dynamic = 'force-dynamic';
+
+const startsByAgent = new FixedWindowRateLimiter(10, 60 * 60_000);
+
+/**
+ * POST /api/v1/agent/broadcasts — an agent (Bearer pux_ag_...) goes live on
+ * one of its owner's channels: { channel: "moshcoding", title, description? }.
+ * Answers the room, its public page and a publish token for it.
+ */
+export async function POST(request: Request) {
+  try {
+    const agent = await new AccountAgents().fromRequest(request);
+    if (!agent) return errorResponse('Send Authorization: Bearer <pux_ag_ agent key>.', 401);
+    const limit = startsByAgent.check(agent.id);
+    if (!limit.success) {
+      return errorResponse(
+        `Too many broadcasts. Try again in ${String(limit.retryAfterSeconds)} seconds.`,
+        429
+      );
+    }
+    const apiKey = process.env.LIVEKIT_API_KEY;
+    const apiSecret = process.env.LIVEKIT_API_SECRET;
+    const url = process.env.NEXT_PUBLIC_LIVEKIT_URL;
+    if (!apiKey || !apiSecret || !url) return errorResponse('LiveKit not configured', 503);
+
+    const input = startBroadcastSchema.parse(await request.json().catch(() => ({})));
+    const db = serviceClient();
+    const { broadcast, channel, subject } = await startBroadcast(db, agent, input, {
+      apiKey,
+      apiSecret,
+      url,
+    });
+
+    // The same follow-ups a person's go-live gets: subscribers hear about it
+    // and the channel's restream destinations pick it up.
+    void import('@/lib/notify-live').then(({ notifyGoLive }) =>
+      notifyGoLive({
+        creator_id: agent.ownerUserId,
+        subject,
+        join_code: broadcast.joinCode,
+        channel_id: channel.id,
+        channel_handle: channel.handle,
+        channel_name: channel.name,
+      })
+    );
+    void import('@/lib/channel-restream').then(({ startChannelRestream }) =>
+      startChannelRestream(broadcast.sessionId, channel.id)
+    );
+
+    return successResponse({ ...broadcast, iceServers: await getIceServers() }, 201);
+  } catch (error) {
+    if (error instanceof BroadcastError) return errorResponse(error.message, error.status);
+    return handleApiError(error);
+  }
+}
diff --git a/apps/web/src/app/docs/agents/page.tsx b/apps/web/src/app/docs/agents/page.tsx
index 2447ff5d..ac160b77 100644
--- a/apps/web/src/app/docs/agents/page.tsx
+++ b/apps/web/src/app/docs/agents/page.tsx
@@ -58,6 +58,14 @@ const api: { route: string; what: string }[] = [
     what: 'Mint a 24-hour LiveKit token scoped to this session and microphone audio. The session must use SFU mode.',
   },
   { route: 'DELETE /api/v1/agents/:participantId', what: 'Leave the session.' },
+  {
+    route: 'POST /api/v1/agent/broadcasts',
+    what: 'Account agents (Bearer pux_ag_ key): { channel, title, description? } goes live on a channel the agent owner may use and returns { sessionId, liveUrl, url, token } to publish audio into. Used to stream podcasts as they are recorded.',
+  },
+  {
+    route: 'POST /api/v1/agent/broadcasts/:sessionId',
+    what: 'Heartbeat while publishing (a host unseen for 90 seconds drops off the listings). DELETE ends the broadcast.',
+  },
 ];
 
 export default function AgentDocsPage() {
@@ -173,9 +181,9 @@ pairux voice join --json`}</code>
               <MessageSquare className="h-6 w-6" aria-hidden="true" /> HTTP API
             </h2>
             <p className="mt-4 text-gray-600">
-              The CLI is a thin wrapper over five routes. Responses are <code>{'{ data }'}</code> or{' '}
-              <code>{'{ error }'}</code>. The participant id returned by join is the credential for
-              the other agent routes, so keep it private.
+              The CLI is a thin wrapper over the first five routes. Responses are{' '}
+              <code>{'{ data }'}</code> or <code>{'{ error }'}</code>. The participant id returned
+              by join is the credential for the other agent routes, so keep it private.
             </p>
             <div className="mt-6 overflow-hidden rounded-xl border border-gray-200">
               {api.map((a) => (
diff --git a/apps/web/src/lib/agent-broadcast.test.ts b/apps/web/src/lib/agent-broadcast.test.ts
new file mode 100644
index 00000000..6cfbb4aa
--- /dev/null
+++ b/apps/web/src/lib/agent-broadcast.test.ts
@@ -0,0 +1,174 @@
+import { describe, expect, it, vi } from 'vitest';
+import {
+  BroadcastError,
+  endBroadcast,
+  heartbeatBroadcast,
+  publisherGrant,
+  startBroadcast,
+  startBroadcastSchema,
+} from './agent-broadcast';
+import type { Db } from '@/lib/orgs';
+
+const tokenCtor = vi.fn();
+const tokenGrant = vi.fn();
+vi.mock('livekit-server-sdk', () => ({
+  TrackSource: { MICROPHONE: 2, CAMERA: 1 },
+  AccessToken: vi.fn().mockImplementation((...args: unknown[]) => {
+    tokenCtor(...args);
+    return { addGrant: tokenGrant, toJwt: () => Promise.resolve('jwt') };
+  }),
+}));
+vi.mock('@/lib/orgs', () => ({ resolveUserPlan: () => Promise.resolve('free') }));
+
+const OWNER = '11111111-1111-4111-8111-111111111111';
+const AGENT = { id: '22222222-2222-4222-8222-222222222222', name: 'riotcoder', ownerUserId: OWNER };
+const CHANNEL = {
+  id: '33333333-3333-4333-8333-333333333333',
+  handle: 'moshcoding',
+  name: 'Moshcoding',
+  owner_id: OWNER,
+  org_id: null,
+  team_id: null,
+};
+const LIVEKIT = {
+  apiKey: 'APIkey',
+  apiSecret: 'a-secret-long-enough-for-hs256-signing!!',
+  url: 'wss://sfu.test',
+};
+
+/** Just enough of the supabase builder for the calls the lib makes. */
+function fakeDb(opts: { channel?: unknown; allowed?: boolean; session?: unknown }) {
+  const calls: { table: string; op: string; payload?: unknown; filters: unknown[] }[] = [];
+  const builder = (table: string) => {
+    const call = { table, op: 'select', payload: undefined as unknown, filters: [] as unknown[] };
+    calls.push(call);
+    const b: Record<string, unknown> = {};
+    const chain =
+      (name: string) =>
+      (...args: unknown[]) => {
+        call.filters.push([name, ...args]);
+        return b;
+      };
+    for (const name of ['select', 'eq', 'ilike', 'not']) b[name] = chain(name);
+    b.insert = (payload: unknown) => {
+      call.op = 'insert';
+      call.payload = payload;
+      return b;
+    };
+    b.update = (payload: unknown) => {
+      call.op = 'update';
+      call.payload = payload;
+      return b;
+    };
+    const answer = () => {
+      if (table === 'channels') return { data: opts.channel ?? null };
+      if (call.op === 'insert')
+        return {
+          data: { id: '44444444-4444-4444-8444-444444444444', join_code: 'ABC234' },
+          error: null,
+        };
+      if (call.op === 'update') return { data: [{ id: 'x' }] };
+      return { data: opts.session ?? null };
+    };
+    b.maybeSingle = () => Promise.resolve(answer());
+    b.single = () => Promise.resolve(answer());
+    b.then = (resolve: (value: unknown) => void) => resolve(answer());
+    return b;
+  };
+  const db = { from: builder, rpc: () => Promise.resolve({ data: opts.allowed ?? false }) };
+  return { db: db as unknown as Db, calls };
+}
+
+describe('startBroadcastSchema', () => {
+  it('takes a handle with or without @', () => {
+    expect(startBroadcastSchema.parse({ channel: '@MoshCoding', title: 'Ep 1' }).channel).toBe(
+      'moshcoding'
+    );
+    expect(() => startBroadcastSchema.parse({ channel: 'bad handle', title: 'x' })).toThrow();
+    expect(() => startBroadcastSchema.parse({ channel: 'ok', title: '' })).toThrow();
+  });
+});
+
+describe('publisherGrant', () => {
+  it('publishes into one room and subscribes to nothing', () => {
+    const grant = publisherGrant('abc');
+    expect(grant.room).toBe('session-abc');
+    expect(grant.canSubscribe).toBe(false);
+    expect(grant.canPublishData).toBe(false);
+  });
+});
+
+describe('startBroadcast', () => {
+  it('opens a public live room on the channel with the owner as host', async () => {
+    const { db, calls } = fakeDb({ channel: CHANNEL, allowed: true });
+    const { broadcast } = await startBroadcast(
+      db,
+      AGENT,
+      { channel: 'moshcoding', title: 'TUI Weekly' },
+      LIVEKIT
+    );
+    const insert = calls.find((c) => c.op === 'insert');
+    expect(insert?.payload).toMatchObject({
+      channel_id: CHANNEL.id,
+      current_host_id: OWNER,
+      host_user_id: OWNER,
+      is_public: true,
+      mode: 'sfu',
+      subject: 'TUI Weekly',
+      settings: { broadcast: { agentId: AGENT.id } },
+    });
+    expect(broadcast.joinCode).toBe('ABC234');
+    expect(broadcast.liveUrl).toMatch(/\/l\/ABC234$/);
+    expect(broadcast.channelUrl).toMatch(/\/@moshcoding$/);
+    expect(broadcast.token).toBe('jwt');
+    // The host identity is what lets the webhook end the live when the publisher leaves.
+    expect(tokenCtor.mock.calls.at(-1)?.[2]).toMatchObject({ identity: OWNER });
+    expect(tokenGrant.mock.calls.at(-1)?.[0]).toMatchObject({
+      room: `session-${broadcast.sessionId}`,
+      canSubscribe: false,
+    });
+  });
+
+  it('refuses a channel the owner may not use, and an unknown one', async () => {
+    await expect(
+      startBroadcast(
+        fakeDb({ channel: CHANNEL, allowed: false }).db,
+        AGENT,
+        { channel: 'moshcoding', title: 'x' },
+        LIVEKIT
+      )
+    ).rejects.toMatchObject({ status: 403 });
+    await expect(
+      startBroadcast(fakeDb({}).db, AGENT, { channel: 'nope', title: 'x' }, LIVEKIT)
+    ).rejects.toBeInstanceOf(BroadcastError);
+  });
+});
+
+describe('heartbeat + end', () => {
+  const mine = { id: 's', status: 'active', settings: { broadcast: { agentId: AGENT.id } } };
+
+  it('only the agent that started it may ping or end it', async () => {
+    const other = { ...mine, settings: { broadcast: { agentId: 'someone-else' } } };
+    await expect(
+      heartbeatBroadcast(fakeDb({ session: other }).db, AGENT, 's')
+    ).rejects.toMatchObject({ status: 404 });
+    await expect(endBroadcast(fakeDb({ session: null }).db, AGENT, 's')).rejects.toMatchObject({
+      status: 404,
+    });
+  });
+
+  it('pings a live one and ends it', async () => {
+    expect(await heartbeatBroadcast(fakeDb({ session: mine }).db, AGENT, 's')).toEqual({
+      live: true,
+    });
+    expect(
+      await heartbeatBroadcast(fakeDb({ session: { ...mine, status: 'ended' } }).db, AGENT, 's')
+    ).toEqual({ live: false });
+    const { db, calls } = fakeDb({ session: mine });
+    await endBroadcast(db, AGENT, 's');
+    expect(calls.find((c) => c.op === 'update')?.payload).toMatchObject({
+      status: 'ended',
+      current_host_id: null,
+    });
+  });
+});
diff --git a/apps/web/src/lib/agent-broadcast.ts b/apps/web/src/lib/agent-broadcast.ts
new file mode 100644
index 00000000..88c80198
--- /dev/null
+++ b/apps/web/src/lib/agent-broadcast.ts
@@ -0,0 +1,244 @@
+/* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-explicit-any */
+/**
+ * Agent broadcasts: an account agent (Bearer pux_ag_ key) puts audio live on
+ * one of its owner's channels without a browser, e.g. a podcast episode the
+ * moment it is recorded.
+ *
+ * Start opens a public SFU room on the channel with the agent's OWNER as its
+ * host (so the existing listings, go-live notifications and restreams treat it
+ * like any live) and hands back a LiveKit token whose identity IS that host
+ * id: when the publisher disconnects, the webhook's "host left" path ends the
+ * live exactly as it does for a person. The publisher pings the heartbeat
+ * route while it plays (the listings drop a host unseen for 90 s) and ends the
+ * room with DELETE when it is done.
+ *
+ * The rows are written with the service client, so every rule a person's
+ * client would hit is checked here: the channel must be one the owner may go
+ * live on (can_use_resource), and only the agent that started a broadcast may
+ * ping or end it.
+ */
+import { z } from 'zod';
+import { AccessToken, TrackSource } from 'livekit-server-sdk';
+import type { VideoGrant } from 'livekit-server-sdk';
+import { maxListeners, participantCap } from '@pairux/shared-types';
+import { resolveUserPlan, type Db } from '@/lib/orgs';
+import { liveUrl, SITE_URL } from '@/lib/embed';
+
+export const startBroadcastSchema = z.object({
+  channel: z
+    .string()
+    .trim()
+    .transform((value) => value.replace(/^@/, '').toLowerCase())
+    .pipe(z.string().regex(/^[a-z0-9][a-z0-9_-]{0,63}$/, 'channel: a channel handle')),
+  title: z.string().trim().min(1).max(200),
+  description: z.string().trim().max(5000).optional(),
+});
+
+export type StartBroadcast = z.infer<typeof startBroadcastSchema>;
+
+export class BroadcastError extends Error {
+  constructor(
+    message: string,
+    readonly status: number
+  ) {
+    super(message);
+  }
+}
+
+export interface BroadcastAgent {
+  id: string;
+  name: string;
+  ownerUserId: string;
+}
+
+export interface Broadcast {
+  sessionId: string;
+  joinCode: string;
+  roomName: string;
+  /** LiveKit server URL (wss://). */
+  url: string;
+  /** Publish-only token for the room; identity = the channel owner (the host). */
+  token: string;
+  liveUrl: string;
+  channelUrl: string;
+}
+
+export const roomNameOf = (sessionId: string): string => `session-${sessionId}`;
+
+/**
+ * The host's grant: microphone (and camera, for a cover picture) into this one
+ * room, nothing to subscribe to, no data channel.
+ */
+export function publisherGrant(sessionId: string): VideoGrant {
+  return {
+    room: roomNameOf(sessionId),
+    roomJoin: true,
+    canPublish: true,
+    canPublishSources: [TrackSource.MICROPHONE, TrackSource.CAMERA],
+    canSubscribe: false,
+    canPublishData: false,
+  };
+}
+
+export async function publisherToken(input: {
+  apiKey: string;
+  apiSecret: string;
+  sessionId: string;
+  hostId: string;
+  name: string;
+  agentId: string;
+}): Promise<string> {
+  const token = new AccessToken(input.apiKey, input.apiSecret, {
+    identity: input.hostId,
+    name: input.name,
+    ttl: '24h',
+    // role host: viewers read this to tell the host from the audience.
+    metadata: JSON.stringify({ role: 'host', userId: input.hostId, broadcastBy: input.agentId }),
+  });
+  token.addGrant(publisherGrant(input.sessionId));
+  return token.toJwt();
+}
+
+interface ChannelRow {
+  id: string;
+  handle: string;
+  name: string;
+  owner_id: string;
+  org_id: string | null;
+  team_id: string | null;
+}
+
+async function channelFor(db: Db, agent: BroadcastAgent, handle: string): Promise<ChannelRow> {
+  const { data } = await (db as any)
+    .from('channels')
+    .select('id, handle, name, owner_id, org_id, team_id')
+    .ilike('handle', handle)
+    .maybeSingle();
+  const channel = data as ChannelRow | null;
+  if (!channel) throw new BroadcastError(`No channel @${handle}`, 404);
+  const { data: allowed } = await (db.rpc as any)('can_use_resource', {
+    p_owner: channel.owner_id,
+    p_org: channel.org_id,
+    p_team: channel.team_id,
+    p_user: agent.ownerUserId,
+  });
+  if (allowed !== true) {
+    throw new BroadcastError(`This agent's owner may not go live on @${channel.handle}`, 403);
+  }
+  return channel;
+}
+
+/** Opens the live room and returns where to publish into it. */
+export async function startBroadcast(
+  db: Db,
+  agent: BroadcastAgent,
+  input: StartBroadcast,
+  livekit: { apiKey: string; apiSecret: string; url: string }
+): Promise<{ broadcast: Broadcast; channel: ChannelRow; subject: string }> {
+  const channel = await channelFor(db, agent, input.channel);
+  const plan = await resolveUserPlan(db, agent.ownerUserId);
+  const now = new Date().toISOString();
+  const { data, error } = await (db as any)
+    .from('sessions')
+    .insert({
+      host_user_id: agent.ownerUserId,
+      creator_id: agent.ownerUserId,
+      current_host_id: agent.ownerUserId,
+      // The insert trigger fills an empty join code with a unique one.
+      join_code: '',
+      mode: 'sfu',
+      status: 'active',
+      settings: {
+        quality: 'medium',
+        allowControl: false,
+        maxParticipants: participantCap(plan),
+        maxListeners: maxListeners(plan),
+        broadcast: { agentId: agent.id },
+      },
+      channel_id: channel.id,
+      org_id: channel.org_id,
+      team_id: channel.team_id,
+      is_public: true,
+      subject: input.title,
+      description: input.description ?? null,
+      published_at: now,
+      host_last_seen_at: now,
+      live_notified_at: now,
+    })
+    .select('id, join_code')
+    .single();
+  if (error || !data) {
+    throw new BroadcastError(`Could not open the room: ${String(error?.message ?? 'no row')}`, 500);
+  }
+  const sessionId = String(data.id);
+  const joinCode = String(data.join_code);
+  const token = await publisherToken({
+    apiKey: livekit.apiKey,
+    apiSecret: livekit.apiSecret,
+    sessionId,
+    hostId: agent.ownerUserId,
+    name: channel.name,
+    agentId: agent.id,
+  });
+  return {
+    broadcast: {
+      sessionId,
+      joinCode,
+      roomName: roomNameOf(sessionId),
+      url: livekit.url,
+      token,
+      liveUrl: liveUrl(joinCode),
+      channelUrl: `${SITE_URL}/@${channel.handle}`,
+    },
+    channel,
+    subject: input.title,
+  };
+}
+
+/** A broadcast this agent started and has not ended, or a 404. */
+async function ownBroadcast(
+  db: Db,
+  agent: BroadcastAgent,
+  sessionId: string
+): Promise<{ id: string; status: string }> {
+  const { data } = await (db as any)
+    .from('sessions')
+    .select('id, status, settings')
+    .eq('id', sessionId)
+    .maybeSingle();
+  if (!data || data.settings?.broadcast?.agentId !== agent.id) {
+    throw new BroadcastError('No broadcast of yours by that id', 404);
+  }
+  return { id: String(data.id), status: String(data.status) };
+}
+
+/** Keeps the live listed: the listings drop a host unseen for 90 seconds. */
+export async function heartbeatBroadcast(
+  db: Db,
+  agent: BroadcastAgent,
+  sessionId: string
+): Promise<{ live: boolean }> {
+  const row = await ownBroadcast(db, agent, sessionId);
+  if (row.status === 'ended') return { live: false };
+  const { data } = await (db as any)
+    .from('sessions')
+    .update({ host_last_seen_at: new Date().toISOString() })
+    .eq('id', sessionId)
+    .not('current_host_id', 'is', null)
+    .select('id');
+  return { live: Array.isArray(data) && data.length > 0 };
+}
+
+/** Ends the live: no host, status ended. Restreams are stopped by the caller. */
+export async function endBroadcast(
+  db: Db,
+  agent: BroadcastAgent,
+  sessionId: string
+): Promise<{ ended: true }> {
+  await ownBroadcast(db, agent, sessionId);
+  await (db as any)
+    .from('sessions')
+    .update({ status: 'ended', current_host_id: null, restream_egress_id: null })
+    .eq('id', sessionId);
+  return { ended: true };
+}