transcribe/sidecar/index.ts

import type { SidecarCommand, SidecarEvent } from '@@/sidecar/protocol';
import { createSidecarConnector } from '@@/sidecar/connect';
import { API_URL } from '@@/officer-url.mjs';
import { getConfig, setConfig, redactUrl, probeWhisper } from './config';
import { transcribeAudio } from './transcribe';
import { toConversation } from './conversation';
import { whisperQueue, workerQueue } from './queue';
import { diarizeAudio, probeAudio, probeWorker, type SpeakerTurn } from './worker';

// The officer-transcribe sidecar. Owns the whole whisper.cpp contract for Officer: the server URL and
// the request tuning that makes it usable for a person talking into a phone (see transcribe.ts — moved
// here unchanged from the platform, and every constant in it has a reason attached in its own comments).
//
// The platform API is a thin auth-gated forwarder (../api/router.ts) holding no Whisper knowledge at all.
//
// ─────────────────────────────────────────────────────────────────────────────────────────────────
// HTTP CONTRACT — the platform strips its /api/transcribe mount prefix before forwarding.
//
//   GET  /_health    { ok, configured, url?, worker, queues }
//   GET  /config      { configured, url, workerUrl }     — urls have userinfo stripped, never a secret
//   PUT  /config      { url, workerUrl? }                 — both validated against their servers first
//   POST /transcribe  multipart: file, spokenLanguages[]
//                     → { text, detectedLanguage, conversation: false }
//                     → { text: "Speaker 1: …", detectedLanguage, conversation: true, speakers }
//                     The conversation probe and diarization run on the torch worker (workerUrl,
//                     ~/dockers/torch-worker) behind their own queue; any worker failure degrades
//                     to the plain transcript rather than failing the request.
//
//   anything else     404
//
// Every route needs `X-Officer-User`, which the platform proxy sets after authenticating the caller. We
// bind loopback only, so its presence is the trust signal — a request without it did not come through
// the platform.
// ─────────────────────────────────────────────────────────────────────────────────────────────────

/** Grab an ephemeral free port by briefly binding one and releasing it. */
function getFreePort(): number {
  const probeServer = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: () => new Response('') });
  const p = probeServer.port;
  probeServer.stop(true);
  if (p == null) throw new Error('failed to acquire a free port');
  return p;
}

const port = getFreePort();

async function handleConfigGet(): Promise<Response> {
  const cfg = await getConfig();
  return Response.json({
    configured: !!cfg,
    url: cfg ? redactUrl(cfg.url) : null,
    workerUrl: cfg?.workerUrl ? redactUrl(cfg.workerUrl) : null,
  });
}

function isHttpUrl(url: string): boolean {
  return /^https?:\/\//i.test(url);
}

async function handleConfigPut(req: Request): Promise<Response> {
  const body = ((await req.json().catch(() => null)) as { url?: unknown; workerUrl?: unknown } | null) ?? {};
  const url = typeof body.url === 'string' ? body.url.trim() : '';
  if (!url) return Response.json({ error: 'url is required' }, { status: 400 });
  if (!isHttpUrl(url)) return Response.json({ error: 'url must start with http:// or https://' }, { status: 400 });

  const probe = await probeWhisper(url);
  if (!probe.ok) return Response.json({ error: probe.error }, { status: 400 });

  // The worker is optional, but a URL that is GIVEN must answer — storing a dead URL silently
  // would turn every future conversation back into an unlabeled wall of text with no error.
  const workerUrl = typeof body.workerUrl === 'string' ? body.workerUrl.trim() : '';
  if (workerUrl) {
    if (!isHttpUrl(workerUrl)) {
      return Response.json({ error: 'workerUrl must start with http:// or https://' }, { status: 400 });
    }
    const workerProbe = await probeWorker(workerUrl);
    if (!workerProbe.ok) return Response.json({ error: `diarization worker: ${workerProbe.error}` }, { status: 400 });
  }

  await setConfig({ url, ...(workerUrl ? { workerUrl } : {}) });
  return Response.json({
    configured: true,
    url: redactUrl(url),
    workerUrl: workerUrl ? redactUrl(workerUrl) : null,
  });
}

// The pipeline. Transcription and the conversation probe run CONCURRENTLY on their own
// resources (whisper.cpp on its GPU, the torch worker on CPU cores), each behind its own serial
// queue, so the probe costs a solo memo nothing — it finishes long before the transcript does.
//
// The probe and diarization are strictly best-effort: any worker failure logs and degrades to
// the plain transcript. A memo must never be lost to a speaker-labeling problem.
async function handleTranscribe(req: Request): Promise<Response> {
  const cfg = await getConfig();
  if (!cfg) {
    return Response.json({ error: 'Whisper not configured — PUT /api/transcribe/config first' }, { status: 400 });
  }

  const body = await req.formData().catch(() => null);
  const file = body?.get('file');
  if (!file || !(file instanceof File)) return Response.json({ error: 'file is required' }, { status: 400 });

  const spokenLanguages = (body?.getAll('spokenLanguages') ?? []).map(String);
  const workerUrl = cfg.workerUrl;

  const probePromise = workerUrl
    ? workerQueue.run(() => probeAudio(file, workerUrl)).catch((err) => {
        console.error('[transcribe] conversation probe failed — returning plain transcript', err);
        return null;
      })
    : Promise.resolve(null);

  try {
    const result = await whisperQueue.run(() => transcribeAudio({ file, whisperUrl: cfg.url, spokenLanguages }));
    const probe = await probePromise;

    if (probe && probe.verdict !== 'solo' && workerUrl) {
      // "unsure" diarizes too: a mislabeled monologue reads fine, a missed conversation is
      // information lost for good. Reuse the probe's turns when it already covered the file.
      const turns: SpeakerTurn[] | null = probe.turns
        ?? (await workerQueue.run(() => diarizeAudio(file, workerUrl)).then(
          (d) => d.turns,
          (err) => {
            console.error('[transcribe] diarization failed — returning plain transcript', err);
            return null;
          },
        ));

      if (turns && turns.length > 0) {
        const conversation = toConversation(result.segments, turns);
        // A "conversation" that clusters into one voice end-to-end is a monologue after all —
        // the paragraph rendering reads better than a wall of "Speaker 1:" prefixes.
        if (conversation.speakerCount > 1) {
          return Response.json({
            text: conversation.text,
            detectedLanguage: result.detectedLanguage,
            conversation: true,
            speakers: conversation.speakerCount,
          });
        }
      }
    }

    return Response.json({ text: result.text, detectedLanguage: result.detectedLanguage, conversation: false });
  } catch (err) {
    console.error('[transcribe] transcription failed', err);
    return Response.json(
      { error: err instanceof Error ? err.message : 'Transcription failed' },
      { status: 502 },
    );
  }
}

const server = Bun.serve({
  port,
  hostname: '127.0.0.1',
  // A phone's voice memo is small; this just needs to be well above any single recording.
  maxRequestBodySize: 1024 * 1024 * 1024,
  /*
    No idle timeout. This server holds a connection open while whisper works, and nothing flows on it in
    the meantime — which is exactly what an idle timeout kills.

    Unset, Bun applies its own ceiling and drops the platform's connection at around 255s. That was the
    last of three hops still doing it: the proxy's outbound fetch and the platform's inbound timeout were
    both fixed on 2026-08-24 and a 42-minute recording STILL died at 4m16s, because this one had not been.
    whisperx went on and finished it, and the result was thrown away.

    With diarization on, align runs at roughly 1.5x the audio duration, so an hour-long episode needs
    ninety minutes of processing. There is no ceiling that is both safe and high enough — 255s is Bun's
    maximum anyway, so "raise it" was never available. 0 disables it.
  */
  idleTimeout: 0,
  async fetch(req) {
    const url = new URL(req.url);

    const officerUser = req.headers.get('X-Officer-User');
    const userId = Number(officerUser);
    if (!officerUser || !Number.isInteger(userId) || userId <= 0) {
      return Response.json({ error: 'missing or invalid X-Officer-User' }, { status: 401 });
    }

    if (url.pathname === '/_health') {
      const cfg = await getConfig();
      if (!cfg) return Response.json({ ok: false, configured: false }, { status: 503 });
      const probe = await probeWhisper(cfg.url);
      if (!probe.ok) return Response.json({ ok: false, configured: true, error: probe.error }, { status: 502 });
      // The worker degrades rather than fails: an unreachable worker is reported but does not
      // cost the health check — transcription still works without it.
      const worker = cfg.workerUrl ? await probeWorker(cfg.workerUrl) : null;
      return Response.json({
        ok: true,
        configured: true,
        url: redactUrl(cfg.url),
        worker: worker === null ? 'not configured' : worker.ok ? 'ok' : `unreachable: ${worker.error}`,
        queues: { whisper: whisperQueue.pending, worker: workerQueue.pending },
      });
    }

    if (url.pathname === '/config') {
      try {
        if (req.method === 'GET') return await handleConfigGet();
        if (req.method === 'PUT' || req.method === 'POST') return await handleConfigPut(req);
        return Response.json({ error: 'method not allowed' }, { status: 405 });
      } catch (err) {
        console.error(`[transcribe] ${req.method} /config failed`, err);
        return Response.json({ error: 'internal error' }, { status: 500 });
      }
    }

    if (url.pathname === '/transcribe') {
      if (req.method !== 'POST') return Response.json({ error: 'method not allowed' }, { status: 405 });
      return handleTranscribe(req);
    }

    return Response.json({ error: 'not found' }, { status: 404 });
  },
});

console.log(`[transcribe] listening on 127.0.0.1:${port} (whisper.cpp URL configured from PUT /config)`);

type ReplyFn = (msg: SidecarEvent) => void;

function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
  switch (cmd.type) {
    case 'ping':
      reply({ type: 'pong', id: cmd.id });
      break;
    default:
      reply({
        type: 'error',
        id: (cmd as SidecarCommand).id,
        error: `Unknown command type: ${(cmd as Record<string, unknown>).type}`,
      });
  }
}

const connection = createSidecarConnector({
  apiUrl: `${API_URL}/api/sidecar/register`,
  name: 'transcribe',
  handles: ['transcribe'],
  onCommand(cmd, reply) {
    handleCommand(cmd as SidecarCommand, reply as ReplyFn);
  },
  onConnected() {
    connection.send({ type: 'transcribe:server', port });
    console.log(`[transcribe] reported server port ${port} to API`);
  },
});

function shutdown(signal: string) {
  console.log(`[transcribe] ${signal} received, shutting down...`);
  try {
    server.stop(true);
  } catch {
    /* already stopped */
  }
  connection.destroy();
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));