import './env-resolve.ts'; // resolve named .env file (must be first — before any config read)
import './env-sanitize.ts'; // strip unresolved ${VAR} placeholders before any config is read

const SUPPRESSED_ERRORS = /NGHTTP2|h2 is not supported|socket disconnected before secure TLS/i;
process.on('uncaughtException', (err) => {
  if (SUPPRESSED_ERRORS.test(err.message ?? '')) return;
  console.error('[server] Uncaught exception (kept alive):', err.message ?? err);
});
process.on('unhandledRejection', (reason) => {
  const msg = (reason as Error)?.message ?? String(reason);
  if (SUPPRESSED_ERRORS.test(msg)) return;
  console.error('[server] Unhandled rejection (kept alive):', msg);
});
import { createServer, STATUS_CODES } from 'node:http';
import { createHmac, timingSafeEqual } from 'node:crypto';
import { execSync } from 'node:child_process';
import { existsSync, readFileSync, writeFileSync, mkdirSync, statSync } from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import express from 'express';
import { WebSocketServer, WebSocket } from 'ws';
import { requireAuth, verifyBearer, authenticateToken, AUTH_PROVIDER, localLogin, addLocalUser, localUserCount, deniedPrincipalOf, publishInternalToken } from './auth.ts';
import { getMcpConfig, getRawMcpConfig, getResolvedMcpConfig, getGlobalMcpConfig, saveMcpConfig, maskEnvValues, mergeWithOriginal, type McpConfig } from './mcp.ts';
import { streamChat, consumeStream, getAgentConfig, saveAgentConfig, getClaudeAuthSource, type AgentConfig, type PermissionHandler, type QuestionHandler, type QuestionAnswers, type AttachmentMeta, type WsEvent, MAX_TURNS_NOTICE } from './claude.ts';
import { mountFeatures, registerFeature, resumeFeatureSession, collectFeatureFlags, collectSidecarRoutes } from './features.ts';
import { registerSpaCatchAll } from './spa-catchall.ts';
import { slackFeature } from './slack/feature.ts';
import { dataPath } from './paths.ts';
import { getAllSessions, getSession, getSessionHistory, upsertSession, appendMessage, saveConversation, loadConversation, setSessionDirectives, getAutoApprove, setAutoApprove, getSessionsByScheduleId, getSessionsVisibleTo, isSessionVisibleTo, setRunStatus, incrementRetryCount, getRunningSessions, getActiveLockCount, updateScheduledSessionStatus, setShuttingDown, backfillSessionVisibility, writePartial, readPartial, clearPartial, registerLivePartial, unregisterLivePartial, readLivePartial, acquireSessionLock, releaseSessionLock, replaceSessionLock, isSessionLocked, getSessionAbortController, forkSession, generateSessionTitle, toListItem, pageSessions, isOwnSession, type ConvBlock, type ConvMessage, type SessionMeta } from './sessions.ts';
import { setBroadcaster, emitToSession } from './session-bus.ts';
import { TURN_CONTROL_EVENTS } from './turn-stream.ts';
import * as scheduler from './scheduler/index.ts';
import { initPolls } from './polls.ts';
import { isScheduledSession } from './scheduler/runner.ts';
import { initBackgroundJobs } from './background-jobs.ts';
import { pushEnabled } from './push/push.ts';
import { upsertToken, removeToken } from './push/store.ts';
import { initPushTriggers, pushTurnDone, pushQuestion } from './push/triggers.ts';
import type { Schedule } from './scheduler/index.ts';
import { listSkills, listMcpCommands, getSkill, saveSkill, deleteSkill, duplicateSkill, renameSkill, getDefaultSkills, setDefaultSkills, resolveDefaultSkillsContent, purgeExpiredSkills, lintSkills } from './skills.ts';
import { listWorkspaceTree, listWorkspaceDir, readWorkspaceFile, safeResolve as resolveWorkspacePath, watchWorkspace, ensureDir as ensureWorkspaceDir } from './workspace.ts';
import { seedDefaults, getBuiltinSkillNames } from './seed.ts';
import { registerModuleRoutes, reconcileInstalledModules } from './modules/index.ts';
import { hydrateSlackUserToken } from './slack/oauth.ts';
import { registerMcpOAuthRoutes } from './mcp-oauth.ts';
import { registerEventRoutes } from './events/routes.ts';
import { reclaimStalePort } from './port-reclaim.ts';
import { listenWithRetry } from './listen-retry.ts';
import { startHeartbeat, recordBootGap, buildReport } from './downtime.ts';
import { registerWebhook } from './events/webhook.ts';
import { startEventDispatcher } from './events/dispatcher.ts';
import { seedOperators } from './contacts.ts';
import { dataSync } from './data-sync.ts';
import { mountMcpServer } from './mcp-server.ts';
import { fromInternal, type Principal } from './security/principal.ts';
import { initSecurity, security, admitTurn, requestIp } from './security/runtime.ts';
import type { PolicyOptions } from './security/policy.ts';
import { writeDenial } from './security/guard.ts';
import { requireOwner } from './security/owner-only.ts';
import { flushEscalations } from './security/escalate.ts';
import { lookupIdempotent, rememberIdempotent } from './idempotency.ts';
import { apiKeyRouter } from './api-key-routes.ts';
import { addUnread, markRead as markUnread, getUnreads } from './unread.ts';

import { loadShragaConfig, getPublicOrigin } from './shraga-config.ts';
import { startSidecars, stopSidecars } from './mcp-sidecar.ts';
import { syncVendorRepos } from './vendor-sync.ts';
import { SelfUpgrade } from './self-upgrade/index.ts';
import { initEngines, getAvailableEngines, getEngine } from './engine/index.ts';
import { statsSampler } from './stats.ts';
import { claudeUsageFor } from './claude-usage.ts';
import { claudeAccountDir } from './claude-account.ts';
import { ClaudeLogin, ClaudeLoginError } from './claude-login.ts';
import { getAll as getAllContacts } from './contacts.ts';
import { artifactsRouter } from './artifacts/artifacts.routes.ts';
import { setSharedFileHeaders } from './share-file.ts';
import { ownerRouter } from './security/owner-routes.ts';
import { handleArtifactToolUse } from './artifacts/artifacts.handler.ts';
import { registerEngine } from './engine/index.ts';
import { subscribeEvent } from './events/bus.ts';
import type { AgentEngine } from './engine/types.ts';
import type { ServerFeature } from './features.ts';
import type { Server as HttpServer } from 'node:http';
import type { Express } from 'express';
import { emitEvent } from './events/bus.ts';
import type { ExtRegisterFn } from './extensions.ts';
import type { WebhookOptions } from './events/webhook.ts';
import type { ShragaEvent, PayloadOf } from './events/types.ts';

export interface BootRegistrations {
  features?: ServerFeature[];
  engines?: AgentEngine[];
  extensions?: ExtRegisterFn[];
  eventSubs?: Array<{ source: string; handler: (payload: any, evt: any) => void }>;
  /** ShragaOptions.security — threaded into the process-wide Policy. */
  security?: { migrate?: PolicyOptions['migrate'] };
}

export interface ServerHandle {
  app: Express;
  server: HttpServer;
  port: number;
  url: string;
  /** Publish an event onto the in-process bus (same fn extensions get as ctx.emitEvent). */
  emitEvent: typeof emitEvent;
  /** Register an extension AFTER start() — mounts onto the live extension Router (before the SPA
   *  catch-all), the same seam file-based *.ext.ts drop-ins hot-load through. OPT-IN: throws unless
   *  ShragaOptions.runtimeRegistration is enabled. */
  registerExtension: (fn: ExtRegisterFn) => Promise<void>;
  /** Declare a verified vendor webhook AFTER start() (sugar over registerExtension — a webhook IS an
   *  extension). OPT-IN: throws unless runtimeRegistration is enabled. */
  registerWebhook: <K extends string>(opts: WebhookOptions<K>) => Promise<void>;
  /** Subscribe to a typed event source AFTER start(). Returns an unsubscribe fn. OPT-IN: throws
   *  unless runtimeRegistration is enabled. */
  on: <K extends string>(source: K, handler: (payload: PayloadOf<K>, evt: ShragaEvent<K>) => void) => () => void;
  /** Drain in-flight streams, stop consumers, close the server. Does NOT exit the process. */
  stop: () => Promise<void>;
}

export async function bootServer(__reg: BootRegistrations = {}): Promise<ServerHandle> {
// Passive mode: HTTP serving only — no schedulers, event consumers, or background writers.
// Used by shadow-verify instances and warm-standby twins that share a live DATA_DIR
// (single-active-writer rule: the active instance is the only one mutating data/).
// `UNCLAW_PASSIVE` is the legacy name — still honoured so existing deploy recipes keep working.
const PASSIVE_FLAG = process.env.SHRAGA_PASSIVE ?? process.env.UNCLAW_PASSIVE;
const PASSIVE = PASSIVE_FLAG === '1' || PASSIVE_FLAG === 'true';
if (PASSIVE) console.log('[server] PASSIVE mode — schedulers, consumers and background writers disabled');
// Flipped by activateConsumers() (blue-green promotion). Declared up here: initSecurity reads it synchronously.
let activated = !PASSIVE;

// Local-only prep (git config, gitignore, untracking). The NETWORK half — fetch/merge — is
// deliberately NOT awaited here; it runs via dataSync.syncOnBoot() AFTER listen(). See syncOnBoot().
if (!PASSIVE) await dataSync.init();

// Kick the boot-time data sync detached, AFTER the port is bound. syncOnBoot() never rejects, and the
// extra .catch() is belt-and-braces: an unhandled rejection here would hit the process-level handler.
function bootDataSync(): void {
  dataSync.syncOnBoot().catch(err => console.error('[data-sync] boot sync error:', (err as Error).message));
}
await loadShragaConfig();
// Process-wide policy + audit (shadow mode: decide + audit, never deny). Audit writes only on the
// active instance — a PASSIVE standby shares DATA_DIR. `activated` is declared above.
// A syncing instance that isn't the audit writer (data-sync mirror) logs to an untracked dir: the shared data/audit
// chain belongs to the writer alone.
const auditMirror = dataSync.isEnabled() && !dataSync.options.auditWriter;
initSecurity({ isActive: () => !PASSIVE || activated, policy: { migrate: __reg.security?.migrate }, audit: auditMirror ? { dir: dataPath('.tmp/audit') } : {} });
// Programmatic engines register through the same seam an overlay uses — BEFORE initEngines() so
// getAvailableEngines() includes them and a directive can resolve to one immediately.
for (const e of __reg.engines ?? []) registerEngine(e);
// Programmatic event subscribers land before the dispatcher starts, symmetric with an overlay.
for (const s of __reg.eventSubs ?? []) subscribeEvent(s.source, s.handler);
await initEngines();
if (!PASSIVE) syncVendorRepos().catch(err => console.warn('[vendor-sync] error:', (err as Error).message));
const selfUpgrade = new SelfUpgrade();
seedDefaults();
const purged = purgeExpiredSkills();
if (purged.length) console.log(`[skills] Purged ${purged.length} expired skill(s): ${purged.join(', ')}`);
for (const w of lintSkills()) console.warn(`[skills] lint: ${w}`);
hydrateSlackUserToken();

// Seed operator contacts from whitelist
try {
  const wl = JSON.parse(readFileSync(dataPath('whitelist.json'), 'utf-8'));
  if (Array.isArray(wl)) seedOperators(wl);
} catch (err) { console.warn('[contacts] Could not seed operators:', (err as Error).message); }

// Backfill visibleTo on legacy bot sessions (idempotent — skips already-patched)
backfillSessionVisibility(({ name }) => {
  if (!name) return null;
  const lower = name.toLowerCase();
  return getAllContacts().find((c) => c.name.toLowerCase() === lower) ?? null;
});

const __dirname = path.dirname(fileURLToPath(import.meta.url));
// The client build to serve. Default = shraga's shipped dist/client; a consumer shipping its own UI
// (e.g. the EE client) sets SHRAGA_CLIENT_DIR (via the clientDir option or env) to its own dist.
const distPath = process.env.SHRAGA_CLIENT_DIR
  ? path.resolve(process.env.SHRAGA_CLIENT_DIR)
  : path.resolve(__dirname, '../../dist/client');

// Only the server publishes its internal-token secret (auth.ts no longer does it on import).
publishInternalToken();

const app = express();
app.use(express.json({
  limit: '20mb',
  verify: (req, _res, buf) => { (req as any).rawBody = buf; },
}));
// Slack interactivity posts application/x-www-form-urlencoded; capture rawBody for signature verification.
app.use(express.urlencoded({
  extended: true,
  limit: '5mb',
  verify: (req, _res, buf) => { (req as any).rawBody = buf; },
}));

// High-frequency poll routes (pty cwd, list refreshes) are quiet on success — they'd otherwise
// drown real request logs. Errors and everything else still logs unconditionally.
const QUIET_POLL_RE = /\/(cwd|ptys|sessions|workspace\/(pty-owners|layout))(\?|$)/;

/** Package version, read once. Served as the `X-Shraga-Version` identity header on EVERY response
 *  (see the middleware below) — a response WITHOUT it proves the responder is not this server, which
 *  is what lets the client tell "shraga 404'd" apart from "something else owns this port". */
const SHRAGA_VERSION: string = (() => {
  try {
    return JSON.parse(readFileSync(path.resolve(__dirname, '../../package.json'), 'utf8')).version ?? 'unknown';
  } catch { return 'unknown'; }
})();

app.use((req, _res, next) => {
  // Set BEFORE any route runs, so it rides on error responses (404/401/5xx) too — not just successes.
  _res.setHeader('X-Shraga-Version', SHRAGA_VERSION);
  const start = Date.now();
  const orig = _res.end.bind(_res);
  (_res as any).end = (...args: any[]) => {
    const quiet = req.method === 'GET' && _res.statusCode < 400 && QUIET_POLL_RE.test(req.url);
    if (!quiet) {
      console.log(`[http] ${req.method} ${req.url} → ${_res.statusCode} (${Date.now() - start}ms)`);
    }
    return orig(...args);
  };
  next();
});

const SERVER_BUILD_ID = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;

// ── REST routes ───────────────────────────────────────────────────────────────

app.get('/api/version', (_req, res) => {
  res.json({ version: SHRAGA_VERSION });
});

// ── Self-upgrade (owner only) ────────────────────────────────────────────────
// Upgrading replaces the code serving this request and restarts the process, so it is strictly an
// owner action, and the response can only ever be "started" — the OUTCOME arrives after the restart,
// via the self-upgrade.finished event (see deliverPendingReport) and GET /api/self-upgrade.

app.get('/api/self-upgrade', requireAuth, async (req, res) => {
  if (!(req as any).user?.isOwner) return void res.status(403).json({ error: 'Only the owner can manage upgrades' });
  const current = selfUpgrade.currentVersion();
  let latest: string | null = null;
  let latestError: string | undefined;
  try { latest = await selfUpgrade.latestVersion(); }
  catch (err) { latestError = (err as Error).message; }
  res.json({
    current,
    latest,
    latestError,
    upToDate: !!latest && latest === current,
    blockers: selfUpgrade.blockers(),
    inFlight: selfUpgrade.inFlight(),
    lastReport: selfUpgrade.lastReport(),
  });
});

app.post('/api/self-upgrade', requireAuth, async (req, res) => {
  if (!(req as any).user?.isOwner) return void res.status(403).json({ error: 'Only the owner can manage upgrades' });
  try {
    const plan = await selfUpgrade.start({ version: req.body?.version, force: req.body?.force === true });
    // 409, not 500: a refusal is a well-formed answer about the deployment's state, and the caller
    // (often the agent, relaying to a human) needs the reason, not a stack trace.
    res.status(plan.started ? 202 : 409).json(plan);
  } catch (err) {
    res.status(500).json({ error: (err as Error).message });
  }
});

// Cached host stats — returns the in-memory ring buffer (does NOT sample on request).
app.get('/api/stats', requireAuth, (_req, res) => {
  res.json({ samples: statsSampler.getStats() });
});

// Claude Code subscription usage for this box. 204 (not an error, not an empty object) is the
// deliberate answer whenever we cannot PROVE a subscription — no credentials anywhere, or no
// user:profile scope — so the client renders nothing rather than a misleading gauge. A LATER upstream
// failure does not 204: the reader keeps serving its last known-good reading marked `stale`, with
// `fetchedAt`, so the gauge stays put and states its age instead of blinking out.
// Deliberately NOT gated on the active engine: the question is whether Claude Code is configured
// with a subscription on this host, not which runtime happens to be selected right now.
// Per viewer: a caller routed to a personal login (claude-account.ts) sees THAT account's usage.
app.get('/api/claude-usage', requireAuth, async (req, res) => {
  const usage = await claudeUsageFor(claudeAccountDir((req as any).user?.email)).get();
  res.set('Cache-Control', 'private, no-store'); // per-viewer answer on a shared URL
  if (!usage) return void res.status(204).end();
  res.json(usage);
});

// Claude subscription login from the web UI (claude-login.ts). `me` = the caller's per-user login
// (workspace/users/<contactId>/.claude); `global` = the box's shared login, owner only.
const claudeLogin = new ClaudeLogin({ onChange: t => claudeUsageFor(t.name === 'global' ? null : t.dir).invalidate() });
const claudeLoginRoute = (fn: (req: express.Request, res: express.Response, user: any) => Promise<unknown>) =>
  async (req: express.Request, res: express.Response) => {
    try { await fn(req, res, (req as any).user); }
    catch (err) {
      const status = err instanceof ClaudeLoginError ? err.status : 500;
      if (status === 500) console.error('[claude-login]', err);
      res.status(status).json({ error: (err as Error).message });
    }
  };
app.get('/api/claude-account', requireAuth, claudeLoginRoute(async (_req, res, user) => {
  const me = await claudeLogin.status(claudeLogin.resolve('me', user)).catch((err: Error) => {
    console.debug('[claude-login] no personal target:', err.message);
    return null;
  });
  const global = user?.isOwner ? await claudeLogin.status(claudeLogin.resolve('global', user)) : undefined;
  // Cached readings only (the reader dedupes + TTLs upstream) — shared usage is visible to every member.
  const meDir = me?.connected ? claudeLogin.resolve('me', user).dir : null;
  const [meUsage, globalUsage] = await Promise.all([meDir ? claudeUsageFor(meDir).get() : null, claudeUsageFor(null).get()]);
  res.json({ me, global, usage: { me: meUsage, global: globalUsage } });
}));
app.put('/api/claude-account/:target/use-shared', requireAuth, claudeLoginRoute(async (req, res, user) => {
  await claudeLogin.setUseShared(claudeLogin.resolve(String(req.params.target), user), !!req.body?.on);
  res.json({ ok: true });
}));
app.post('/api/claude-account/:target/login', requireAuth, claudeLoginRoute(async (req, res, user) => {
  res.json({ url: await claudeLogin.start(claudeLogin.resolve(String(req.params.target), user)) });
}));
app.post('/api/claude-account/:target/login/code', requireAuth, claudeLoginRoute(async (req, res, user) => {
  res.json(await claudeLogin.submitCode(claudeLogin.resolve(String(req.params.target), user), String(req.body?.code ?? '')));
}));
app.delete('/api/claude-account/:target', requireAuth, claudeLoginRoute(async (req, res, user) => {
  await claudeLogin.disconnect(claudeLogin.resolve(String(req.params.target), user));
  res.json({ ok: true });
}));

/**
 * Conversation list — PAGED and TRIMMED.
 *
 * It used to serialize the entire session index: 6.76 MB and ~226 ms of blocking work per call on
 * the Circles box, most of it `triggeredSkills` + `seenSlackTs`, which are server bookkeeping the
 * browser never reads. The full record is still one `/api/sessions/:id/meta` away.
 *
 * Two modes, both returning `{ sessions, nextCursor }`:
 *   ?filter=mine|all&limit=&before=<cursor>  — a page of the list, newest first.
 *   ?ids=a,b,c                               — exactly these sessions, at any age.
 *
 * `ids` is what keeps the client's filters honest instead of "searches the first page only": the
 * unread map arrives over the socket in full (per-user, `unread_sync`), so the Sidebar knows every
 * unread id and hydrates the ones outside the loaded window by id. The active session is hydrated
 * the same way. `mine` cannot be done that way — it is a predicate over the whole 12k index — so
 * it is a server-side filter applied BEFORE paging.
 */
const SESSIONS_PAGE_MAX = 200;
app.get('/api/sessions', requireAuth, async (req, res) => {
  const user = (req as any).user;
  // Exclude PTY-only sessions — a standalone/terminal-first shell is not a conversation.
  const all = getSessionsVisibleTo(user.uid, user.isOwner, user.email).filter((s) => s.kind !== 'terminal');

  const email = String(user.email ?? '').toLowerCase();
  const mine = (s: SessionMeta) => isOwnSession(s, user.uid, email);

  const idsParam = String(req.query.ids ?? '').trim();
  if (idsParam) {
    const want = new Set(idsParam.split(',').map((s) => s.trim()).filter(Boolean).slice(0, SESSIONS_PAGE_MAX));
    // Deliberately NOT narrowed to `mine`: this is how the ACTIVE conversation gets a row, and the
    // open session must render under either filter. Each row carries `mine` so the client can scope
    // the parts that should be scoped (the unread view and its dot) without losing the active one.
    return void res.json({ sessions: all.filter((s) => want.has(s.sessionId)).map((s) => toListItem(s, mine(s))), nextCursor: null });
  }

  const scoped = String(req.query.filter ?? 'all') === 'mine' ? all.filter(mine) : all;

  const limit = Math.min(Math.max(Number(req.query.limit) || 50, 1), SESSIONS_PAGE_MAX);
  res.json(pageSessions(scoped, { limit, before: String(req.query.before ?? '') || undefined, mine }));
});

app.get('/api/sessions/:id/meta', requireAuth, async (req, res) => {
  const user = (req as any).user;
  const meta = getSession(String(req.params.id));
  if (!meta) return res.status(404).json({ error: 'not found' });
  if (!isSessionVisibleTo(meta, user.uid, user.isOwner, user.email)) return res.status(404).json({ error: 'not found' });
  res.json(meta);
});

// Per-session runtime directives (engine/model/turns/thinking). The Agent Config panel writes these
// for the active session so a change applies to THIS conversation — session directives shadow the
// global agent-config at send time (see claude.ts), so editing the global config alone never affects
// an already-started session.
app.put('/api/sessions/:id/directives', requireAuth, (req, res) => {
  const user = (req as any).user;
  const sid = String(req.params.id);
  const meta = getSession(sid);
  if (!meta) return void res.status(404).json({ error: 'not found' });
  if (!isSessionVisibleTo(meta, user.uid, user.isOwner, user.email)) return void res.status(404).json({ error: 'not found' });
  // thinking is an untrusted request field; type it to the valid set (invalid strings fall through
  // the `|| undefined` below). Keys the core doesn't own (an add-on's, e.g. voice model directives)
  // pass through OPAQUELY — the core names none of them: '' clears, `false` is preserved, else stored.
  const CORE_KEYS = new Set(['engine', 'model', 'turns', 'thinking']);
  const body = (req.body ?? {}) as Record<string, unknown> & { engine?: string; model?: string; turns?: number; thinking?: 'enabled' | 'adaptive' | 'disabled' };
  const passthrough: Record<string, unknown> = {};
  for (const [k, v] of Object.entries(body)) {
    if (CORE_KEYS.has(k)) continue;
    passthrough[k] = v === '' ? undefined : v; // '' = unset → fall back to default; `false` preserved (e.g. a tier off)
  }
  const next = {
    ...meta.directives,
    ...(body.engine !== undefined ? { engine: body.engine || undefined } : {}),
    ...(body.model !== undefined ? { model: body.model || undefined } : {}),
    ...(body.turns !== undefined ? { turns: (body.turns as unknown) === null || (body.turns as unknown) === '' ? undefined : body.turns } : {}),
    ...(body.thinking !== undefined ? { thinking: body.thinking || undefined } : {}),
    ...passthrough,
  };
  setSessionDirectives(sid, next);
  res.json({ directives: next });
});

app.get('/api/sessions/:id/messages', requireAuth, async (req, res) => {
  const sid = String(req.params.id);
  console.log(`[http] loading messages for session ${sid.slice(0, 8)}…`);
  const session = getSession(sid);
  const conv = loadConversation(sid);
  if (conv.length > 0) {
    const partial = readLivePartial(sid) ?? readPartial(sid);
    if (partial?.length) {
      conv.push({ id: `partial-${sid}`, role: 'assistant', blocks: partial, ts: Date.now() });
      console.log(`[http] loaded ${conv.length} messages (incl. partial) from own store for ${sid.slice(0, 8)}`);
    } else {
      console.log(`[http] loaded ${conv.length} messages from own store for ${sid.slice(0, 8)}`);
    }
    const senders = new Set(conv.filter(m => m.role === 'user' && m.senderName).map(m => m.senderName));
    if (session?.userName) senders.add(session.userName);
    return res.json({ format: 'conv', messages: conv, busy: isSessionBusy(sid), participants: [...senders] });
  }
  const messages = await getSessionHistory(sid);
  console.log(`[http] loaded ${messages.length} messages from Claude JSONL for ${sid.slice(0, 8)}`);
  res.json({ format: 'jsonl', messages, busy: isSessionBusy(sid) });
});

app.post('/api/sessions/:id/fork', requireAuth, (req, res) => {
  const user = (req as any).user as import('./auth.ts').AuthUser;
  const sourceId = String(req.params.id);
  const source = getSession(sourceId);
  if (!source) return void res.status(404).json({ error: 'not found' });
  if (!isSessionVisibleTo(source, user.uid, user.isOwner, user.email)) return void res.status(404).json({ error: 'not found' });
  const { truncateAtIndex } = req.body as { truncateAtIndex?: number };
  const newId = forkSession(sourceId, { uid: user.uid, email: user.email, name: user.email.split('@')[0] }, truncateAtIndex);
  if (!newId) return void res.status(400).json({ error: 'nothing to fork' });
  console.log(`[http] forked session ${sourceId.slice(0, 8)} → ${newId.slice(0, 8)} for ${user.email}`);
  res.json({ sessionId: newId });
});

app.post('/api/sessions/:id/push', requireAuth, (req, res) => {
  const sid = String(req.params.id);
  const meta = getSession(sid);
  if (!meta) return void res.status(404).json({ error: 'not found' });
  const { message, source } = req.body as { message?: string; source?: 'proactive' | 'schedule' };
  if (!message) return void res.status(400).json({ error: 'message required' });
  appendMessage(sid, { id: crypto.randomUUID(), role: 'assistant', blocks: [{ type: 'text', text: message }], ts: Date.now() });
  upsertSession(sid, meta.title, { uid: meta.uid, email: meta.userEmail });
  notifyUnread(meta.uid, sid, message.slice(0, 120), source || 'proactive', meta.title);
  broadcast({ type: 'session_messages_changed', sessionId: sid });
  console.log(`[http] pushed message to ${sid.slice(0, 8)} for ${meta.userEmail}`);
  res.json({ ok: true });
});

app.get('/api/mcps', requireAuth, (req, res) => {
  const user = (req as any).user;
  const globalNames = new Set(Object.keys(getGlobalMcpConfig()));
  const resolved = maskEnvValues(getResolvedMcpConfig(user.uid));
  const entries: Record<string, McpConfig[string] & { readonly?: boolean }> = {};
  for (const [name, config] of Object.entries(resolved)) {
    entries[name] = { ...config, readonly: globalNames.has(name) };
  }
  res.json(entries);
});

// Owner-only: an MCP entry can carry a stdio `command` the server spawns.
app.put('/api/mcps', requireAuth, requireOwner('Only an owner can change MCP servers'), (req, res) => {
  const user = (req as any).user;
  const globalNames = new Set(Object.keys(getGlobalMcpConfig()));
  const incoming = req.body as McpConfig;
  const userOnly: McpConfig = {};
  for (const [name, config] of Object.entries(incoming)) {
    if (!globalNames.has(name)) userOnly[name] = config;
  }
  const original = getRawMcpConfig(user.uid);
  const merged = mergeWithOriginal(userOnly, original);
  saveMcpConfig(user.uid, merged);
  res.json({ ok: true });
});

app.get('/api/config', requireAuth, (req, res) => {
  // `claudeAuthSource` is derived server state (not persisted config) — the spread always overrides
  // any stale value, so it can never round-trip into agent-config.json even if a client echoes it back.
  // `isOwner` is the same kind of derived field: the client hides owner-only controls on it (PUT strips it).
  res.json({ ...getAgentConfig(), isOwner: !!(req as any).user?.isOwner, claudeAuthSource: getClaudeAuthSource() });
});

app.put('/api/config', requireAuth, requireOwner('Only an owner can change the agent config'), (req, res) => {
  const { claudeAuthSource: _drop, isOwner: _dropOwner, ...config } = (req.body ?? {}) as AgentConfig & { claudeAuthSource?: string; isOwner?: boolean };
  saveAgentConfig(config);
  res.json({ ok: true });
});

app.get('/api/engines', requireAuth, (_req, res) => {
  const engines = getAvailableEngines();
  const result = engines.map(name => {
    const engine = getEngine(name);
    return { name, models: engine.getModels() };
  });
  res.json({ engines: result, multiEngine: engines.length > 1 });
});


app.get('/api/skills', requireAuth, (_req, res) => {
  res.json({ skills: [...listSkills(), ...listMcpCommands(), 'compact'], builtins: getBuiltinSkillNames() });
});

app.get('/api/skills/:name', requireAuth, (req, res) => {
  const skill = getSkill(String(req.params.name));
  if (!skill) return res.status(404).json({ error: 'Not found' });
  res.json(skill);
});

app.put('/api/skills/:name', requireAuth, requireOwner('Only an owner can change skills'), (req, res) => {
  try {
    const { content } = req.body as { content: string };
    saveSkill(String(req.params.name), content ?? '');
    res.json({ ok: true });
  } catch (e: any) { res.status(400).json({ error: e.message }); }
});

app.delete('/api/skills/:name', requireAuth, requireOwner('Only an owner can change skills'), (req, res) => {
  try {
    deleteSkill(String(req.params.name));
    res.json({ ok: true });
  } catch (e: any) { res.status(400).json({ error: e.message }); }
});

app.post('/api/skills/:name/duplicate', requireAuth, requireOwner('Only an owner can change skills'), (req, res) => {
  try {
    const { newName } = req.body as { newName: string };
    const skill = duplicateSkill(String(req.params.name), newName);
    res.json(skill);
  } catch (e: any) { res.status(400).json({ error: e.message }); }
});

app.post('/api/skills/:name/rename', requireAuth, requireOwner('Only an owner can change skills'), (req, res) => {
  try {
    const { newName } = req.body as { newName: string };
    renameSkill(String(req.params.name), newName);
    res.json({ ok: true });
  } catch (e: any) { res.status(400).json({ error: e.message }); }
});

app.get('/api/skills-defaults', requireAuth, (_req, res) => {
  res.json(getDefaultSkills());
});

app.put('/api/skills-defaults', requireAuth, requireOwner('Only an owner can change skills'), (req, res) => {
  setDefaultSkills(req.body);
  res.json({ ok: true });
});

// ── Data-plane modules ───────────────────────────────────────────────────────
registerModuleRoutes(app, requireAuth);

// ── Schedules ────────────────────────────────────────────────────────────────

function scheduleIfVisible(id: string, uid: string, isOwner = false): Schedule | undefined {
  const s = scheduler.getSchedule(id);
  if (!s) return undefined;
  if (isOwner || s.scope === 'system' || s.createdBy.uid === uid) return s;
  return undefined;
}

// "What did I miss?" — the on-demand downtime report. Read-only and INERT: it returns outage
// ranges, the windows phase 1 deliberately did not replay (`missedRun`), and the Slack messages
// that arrived while we were down, each with a PROPOSAL. Running any of it is a separate,
// explicit call (POST /api/schedules/:id/run). `?slack=0` skips the Slack read.
app.get('/api/downtime', requireAuth, async (req, res) => {
  try {
    const slack = String(req.query.slack ?? '1') !== '0';
    res.json(await buildReport({ slack }));
  } catch (err) {
    console.error('[downtime] report failed:', err);
    res.status(500).json({ error: (err as Error).message });
  }
});

app.get('/api/schedules', requireAuth, (req, res) => {
  const user = (req as any).user;
  const schedules = scheduler.listSchedules().filter((s) => user.isOwner || s.scope === 'system' || s.createdBy.uid === user.uid);
  const runningIds = scheduler.getRunningIds();
  res.json({ schedules, runningIds });
});

app.get('/api/schedules/:id', requireAuth, (req, res) => {
  const user = (req as any).user;
  const s = scheduleIfVisible(String(req.params.id), user.uid, user.isOwner);
  if (!s) return res.status(404).json({ error: 'Not found' });
  res.json(s);
});

app.post('/api/schedules', requireAuth, (req, res) => {
  const user = (req as any).user;
  const body = req.body as Partial<Schedule>;
  const now = Date.now();
  const schedule: Schedule = {
    id: crypto.randomUUID(),
    name: body.name || 'Untitled schedule',
    enabled: body.enabled ?? true,
    trigger: body.trigger as Schedule['trigger'],
    task: body.task as Schedule['task'],
    onMissed: body.onMissed,
    scope: 'user',
    createdBy: { uid: user.uid, email: user.email },
    createdAt: now,
    updatedAt: now,
    runCount: 0,
  };
  const result = scheduler.upsertSchedule(schedule);
  if (!result.ok) return res.status(400).json({ error: result.error });
  res.json(result.schedule);
});

app.put('/api/schedules/:id', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  const existing = scheduleIfVisible(id, user.uid, user.isOwner);
  if (!existing) return res.status(404).json({ error: 'Not found' });
  if (existing.createdBy.uid !== user.uid && !user.isOwner) return res.status(403).json({ error: 'Only the owner can edit this schedule' });
  const body = req.body as Partial<Schedule>;
  const updated: Schedule = {
    ...existing,
    name: body.name ?? existing.name,
    enabled: body.enabled ?? existing.enabled,
    trigger: (body.trigger ?? existing.trigger) as Schedule['trigger'],
    task: (body.task ?? existing.task) as Schedule['task'],
    onMissed: body.onMissed ?? existing.onMissed,
  };
  const result = scheduler.upsertSchedule(updated);
  if (!result.ok) return res.status(400).json({ error: result.error });
  res.json(result.schedule);
});

app.delete('/api/schedules/:id', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  const existing = scheduleIfVisible(id, user.uid, user.isOwner);
  if (!existing) return res.status(404).json({ error: 'Not found' });
  if (existing.createdBy.uid !== user.uid && !user.isOwner) return res.status(403).json({ error: 'Only the owner can delete this schedule' });
  const ok = scheduler.deleteSchedule(id);
  if (!ok) return res.status(404).json({ error: 'Not found' });
  res.json({ ok: true });
});

app.post('/api/schedules/:id/toggle', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  if (!scheduleIfVisible(id, user.uid, user.isOwner)) return res.status(404).json({ error: 'Not found' });
  const s = scheduler.toggleSchedule(id, !!req.body.enabled);
  if (!s) return res.status(404).json({ error: 'Not found' });
  res.json(s);
});

app.post('/api/schedules/:id/run', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  if (!scheduleIfVisible(id, user.uid, user.isOwner)) return res.status(404).json({ error: 'Not found' });
  const override = typeof req.body?.override === 'string' ? req.body.override.trim() || undefined : undefined;
  const outcome = scheduler.runNow(id, override);
  // A refusal here is a CONFLICT with the schedule's current state (a run already holds the
  // cross-restart lock, the period is already done), not a missing resource — 404 sent operators
  // hunting for a schedule that plainly exists.
  if (!outcome.ok) {
    if (outcome.reason === 'unknown-schedule') return res.status(404).json({ error: 'Not found' });
    return res.status(409).json({ error: outcome.message, reason: outcome.reason });
  }
  res.json({ sessionId: outcome.sessionId, queued: outcome.queued ?? false });
});

app.post('/api/schedules/:id/cancel', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  if (!scheduleIfVisible(id, user.uid, user.isOwner)) return res.status(404).json({ error: 'Not found' });
  const ok = scheduler.cancelRun(id);
  res.json({ ok });
});

app.get('/api/schedules/:id/runs', requireAuth, (req, res) => {
  const user = (req as any).user;
  const id = String(req.params.id);
  if (!scheduleIfVisible(id, user.uid, user.isOwner)) return res.status(404).json({ error: 'Not found' });
  res.json(getSessionsByScheduleId(id));
});

// Cap on a tool_result pushed to viewers; the transcript keeps the full output.
const STREAMED_RESULT_MAX = 2000;
// ── REST chat endpoint (for automation / CLI triggers / agent-to-agent) ──────
/**
 * Run a single chat turn with all its side effects (session lock, message
 * persistence, run-status, unread notify, broadcast). Shared by the /api/chat
 * route and the MCP streaming handler. Pass `hooks.onEvent` to observe the live
 * agent stream (progress streaming). Returns a discriminated result so callers
 * map it to their own transport (HTTP status / MCP frame).
 */
type RunChatTurnResult =
  | { status: 'busy' }
  | { sessionId: string; text: string; blocks: ConvBlock[]; stopReason?: string }
  | { sessionId: string; error: string };

async function runChatTurn(
  opts: {
    prompt: string;
    sessionId?: string;
    uid: string;
    userEmail: string;
    userName?: string;
    abortController?: AbortController;
    context?: Record<string, string>;
    principal: Principal;
  },
  hooks?: { onEvent?: (ev: WsEvent) => void },
): Promise<RunChatTurnResult> {
  const { prompt, sessionId: reqSid, uid, userEmail } = opts;
  const userName = opts.userName ?? userEmail.split('@')[0];
  const sid = reqSid || `api-${crypto.randomUUID()}`;
  const abortController = opts.abortController ?? new AbortController();

  if (reqSid && !acquireSessionLock(sid, 'api', abortController)) {
    return { status: 'busy' };
  }
  if (!reqSid) acquireSessionLock(sid, 'api', abortController);
  upsertSession(sid, prompt, { uid, email: userEmail });
  appendMessage(sid, { id: crypto.randomUUID(), role: 'user', blocks: [{ type: 'text', text: prompt }], channel: 'api', senderName: userName });
  setRunStatus(sid, 'running', 'web');

  let stopReason: string | undefined;
  const onEvent = (ev: WsEvent) => {
    // consumeStream() drops the `done` event's stopReason, so every non-WS transport lost the fact
    // that the turn was TRUNCATED and returned the partial answer as if it were complete.
    if (ev.type === 'done') stopReason = (ev as any).stopReason;
    hooks?.onEvent?.(ev);
  };

  try {
    const blocks = await consumeStream(streamChat({
      principal: opts.principal,
      prompt,
      sessionId: sid,
      uid,
      userEmail,
      userName,
      mcpServers: getMcpConfig(uid),
      abortController,
      context: opts.context ?? { source: 'api', user: userEmail },
      onPermissionRequest: async () => ({ allow: true }),
    }), onEvent, {
      // Same reason as the wake lane: an /api/chat or MCP-driven turn used to surface nothing to a
      // web viewer until it ended, and dropped an add-on engine's subagent events entirely.
      maxResultChars: STREAMED_RESULT_MAX,
      onDelta: (event) => broadcast({ type: 'session_stream', sessionId: sid, event }),
      onPassthrough: (event) => emitToSession(sid, event),
    });
    if (blocks.length) {
      appendMessage(sid, { id: crypto.randomUUID(), role: 'assistant', blocks });
    }
    const text = blocks.filter(b => b.type === 'text').map(b => b.text).join('\n');
    const meta = getSession(sid);
    notifyUnread(uid, sid, text.slice(0, 120) || '(completed)', 'response', meta?.title);
    broadcast({ type: 'session_messages_changed', sessionId: sid });
    return { sessionId: sid, text, blocks, stopReason };
  } catch (err: any) {
    console.error(`[chat-turn] error:`, err.message);
    return { sessionId: sid, error: err.message };
  } finally {
    if (releaseSessionLock(sid, abortController)) {
      setRunStatus(sid, 'idle');
    }
  }
}

app.post('/api/chat', requireAuth, async (req, res) => {
  const user = (req as any).user as import('./auth.ts').AuthUser;
  const { prompt, sessionId: reqSid, callbackUrl, sync, clientRequestId } = req.body as {
    prompt?: string; sessionId?: string; callbackUrl?: string; sync?: boolean; clientRequestId?: string;
  };
  if (!prompt) return void res.status(400).json({ error: 'prompt required' });
  if (callbackUrl) {
    try { const u = new URL(callbackUrl); if (!['http:', 'https:'].includes(u.protocol)) throw 0; }
    catch { return void res.status(400).json({ error: 'callbackUrl must be a valid HTTP(S) URL' }); }
  }

  // Idempotency: a retried submit with the same key reuses the session that first
  // handled it (within TTL) instead of spawning a duplicate.
  const idemKey = clientRequestId || (req.get('idempotency-key') || undefined);
  if (idemKey) {
    const existing = lookupIdempotent(user.uid, idemKey);
    if (existing) return void res.json({ sessionId: existing, status: 'duplicate' });
  }

  // Guard before any spend: blocklist → rate → concurrency (shadow unless SECURITY_ENFORCE).
  const admission = admitTurn(user.principal, { ip: requestIp(req), channel: 'api' });
  if (!admission.ok) return void writeDenial(res, admission);

  const sid = reqSid || `api-${crypto.randomUUID()}`;
  if (idemKey) rememberIdempotent(user.uid, idemKey, sid);
  const apiAbortController = new AbortController();
  const run = async () => {
    try {
      return await runChatTurn({
        prompt,
        sessionId: sid,
        uid: user.uid,
        userEmail: user.email,
        principal: user.principal,
        abortController: apiAbortController,
        context: { source: 'api', user: user.email },
      });
    } finally { admission.release(); }
  };

  if (sync) {
    const result = await run();
    if ('status' in result) return void res.status(409).json({ error: 'Session is already processing a request' });
    if ('error' in result) return void res.status(500).json(result);
    res.json(result);
  } else {
    // Reject a duplicate before responding 'accepted' (lock is acquired inside run()).
    if (reqSid && isSessionLocked(sid)) {
      admission.release();
      return void res.status(409).json({ error: 'Session is already processing a request' });
    }
    res.json({ sessionId: sid, status: 'accepted' });
    const result = await run();
    if (callbackUrl) {
      try {
        await fetch(callbackUrl, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify(result),
        });
      } catch (err: any) {
        console.error(`[api-chat] callback failed (${callbackUrl}):`, err.message);
      }
    }
  }
});

app.get('/api/workspace', requireAuth, (_req, res) => {
  res.json({ entries: listWorkspaceTree() });
});

app.get('/api/workspace/ls', requireAuth, (req, res) => {
  const dir = String(req.query.path ?? '');
  res.json({ entries: listWorkspaceDir(dir) });
});

app.get('/api/workspace/file', requireAuth, (req, res) => {
  const rel = String(req.query.path ?? '');
  if (!rel) return res.status(400).json({ error: 'path required' });
  const result = readWorkspaceFile(rel);
  if (!result) return res.status(404).json({ error: 'Not found or invalid path' });
  res.json(result);
});

app.get('/api/workspace/raw', requireAuth, (req, res) => {
  const rel = String(req.query.path ?? '');
  if (!rel) return res.status(400).json({ error: 'path required' });
  const resolved = resolveWorkspacePath(rel);
  if (!resolved || !existsSync(resolved)) return res.status(404).json({ error: 'Not found' });
  try { if (!statSync(resolved).isFile()) return res.status(400).json({ error: 'Not a file' }); }
  catch { return res.status(404).json({ error: 'Not found' }); }
  if (req.query.dl) {
    const filename = (rel.split('/').pop() || 'download').replace(/"/g, '\\"');
    res.setHeader('Content-Disposition', `attachment; filename="${filename}"`);
  }
  res.sendFile(resolved);
});

app.use('/uploads/shared', express.static(dataPath('uploads/shared'), { dotfiles: 'deny', index: false, setHeaders: setSharedFileHeaders }));
app.use('/uploads', requireAuth, express.static(dataPath('uploads'), { dotfiles: 'deny', index: false }));

app.post('/api/upload', requireAuth, express.raw({ type: '*/*', limit: '50mb' }), (req, res) => {
  const sid = (req.headers['x-session-id'] as string) || 'shared';
  const uploadsDir = dataPath(`uploads/${sid}`);
  mkdirSync(uploadsDir, { recursive: true });
  const raw = (req.headers['x-filename'] as string) || 'upload';
  const safeName = path.basename(decodeURIComponent(raw));
  const id = crypto.randomUUID().slice(0, 8);
  const filename = `${id}-${safeName}`;
  const dest = path.join(uploadsDir, filename);
  writeFileSync(dest, req.body as Buffer);
  const mimeType = (req.headers['content-type'] as string) || 'application/octet-stream';
  res.json({ url: `/uploads/${sid}/${filename}`, path: dest, name: safeName, mimeType });
});

registerMcpOAuthRoutes(app);

app.get('/api/data-sync/log', requireAuth, async (_req, res) => {
  const log = await dataSync.getLog();
  res.json(log);
});

app.use(artifactsRouter);
app.use(ownerRouter);

// Runtime feature flags for the client (env-gated, never persisted to agent-config.json).
// ── Auth mode + local login (PUBLIC — no requireAuth) ────────────────────────
// The client asks /api/auth/mode to decide which login UI to render (local form vs
// Firebase Google). Local login/register only exist when AUTH_PROVIDER=local (default).
app.get('/api/auth/mode', (_req, res) => {
  res.json({ provider: AUTH_PROVIDER, needsSetup: AUTH_PROVIDER === 'local' && localUserCount() === 0 });
});
app.post('/api/auth/login', (req, res) => {
  if (AUTH_PROVIDER !== 'local') return void res.status(404).json({ error: 'local auth disabled' });
  const { email, password } = (req.body ?? {}) as { email?: string; password?: string };
  const token = email && password ? localLogin(email, password) : null;
  if (!token) return void res.status(401).json({ error: 'Invalid credentials' });
  res.json({ token, user: { uid: email, email } });
});
app.post('/api/auth/register', (req, res) => {
  if (AUTH_PROVIDER !== 'local') return void res.status(404).json({ error: 'local auth disabled' });
  // First-run bootstrap: allow creating the first user; after that require SHRAGA_ALLOW_SIGNUP=1.
  if (localUserCount() > 0 && process.env.SHRAGA_ALLOW_SIGNUP !== '1') return void res.status(403).json({ error: 'Signup disabled' });
  const { email, password } = (req.body ?? {}) as { email?: string; password?: string };
  if (!email || !password) return void res.status(400).json({ error: 'email + password required' });
  try { addLocalUser(email, password); } catch (e: any) { return void res.status(409).json({ error: e.message }); }
  res.json({ token: localLogin(email, password), user: { uid: email, email } });
});

// Feature gates. Add-on surfaces ship OFF; enable per-deployment via SHRAGA_FEAT_* env
// (an optional add-on / downstream distribution sets them on). Single source of truth for the web UI.
const featEnabled = (k: string, def = false): boolean => {
  const v = process.env[`SHRAGA_FEAT_${k}`];
  return v === undefined ? def : v === '1' || v === 'true';
};
app.get('/api/features', requireAuth, (_req, res) => {
  // Core flags (SHRAGA_FEAT_* env), then merge feature-contributed flags OVER them. The core names no
  // add-on surface: add-on features declare their own capability flags through the seam (collectFeatureFlags).
  res.json({
    push: pushEnabled(),
    workspace: featEnabled('WORKSPACE'), // multi-tab workspace (FlexLayout). Default = chat-only.
    instances: featEnabled('INSTANCES'), // multi-instance (fleet) switcher
    ...collectFeatureFlags(),            // add-on surfaces declare their own flags here.
  });
});

// ── Remote push (native appwrap wrappers register device tokens here) ──────────
// Gated by PUSH_ENABLED + provider creds; register is a no-op-OK when disabled.
app.post('/api/push/register', requireAuth, (req, res) => {
  const uid = (req as any).user.uid as string;
  const { token, platform, topic } = (req.body || {}) as { token?: string; platform?: string; topic?: string };
  if (!pushEnabled()) return void res.json({ ok: true, enabled: false });
  if (!token || (platform !== 'apns' && platform !== 'fcm')) {
    return void res.status(400).json({ error: 'token + platform(apns|fcm) required' });
  }
  upsertToken(uid, token, platform, topic);
  res.json({ ok: true });
});
app.post('/api/push/unregister', requireAuth, (req, res) => {
  const uid = (req as any).user.uid as string;
  const { token } = (req.body || {}) as { token?: string };
  if (token) removeToken(uid, token);
  res.json({ ok: true });
});

// ── API Keys ──────────────────────────────────────────────────────────────────
app.use(apiKeyRouter({ base: '/api/api-keys', gate: [requireAuth] }));

// ── MCP Server ────────────────────────────────────────────────────────────────
mountMcpServer(app, { runChatTurn });

// Mount deployment drop-in routes from data/extensions/*.ext.ts (hot-reload, before catch-all).
const { loadExtensions, registerExtension } = await import('./extensions.ts');
// Programmatic extensions funnel through the SAME router+ctx as file-based *.ext.ts drop-ins
// (mounted before the SPA catch-all). Queue them before loadExtensions so it flushes them.
for (const fn of __reg.extensions ?? []) await registerExtension(fn);
await loadExtensions(app);

// `index: false` so `/` falls through to the SPA catch-all, which injects the runtime web-config
// into index.html. Static assets (JS/CSS/etc.) are still served directly from here.
if (existsSync(distPath)) app.use(express.static(distPath, { index: false }));

// The SPA catch-all (`app.get('*')`) is registered LATER — after mountFeatures() — via
// registerSpaCatchAll(), so feature/extension GET routes are matched before falling through to
// index.html. It is re-registered on passive→active promotion (mountFeatures runs again there),
// each call splicing out the prior catch-all layer so it stays truly LAST in the router stack.
// In dev (no dist/) it is skipped entirely — Vite serves the SPA and a catch-all would 404-shadow it.
// Implementation + predicate live in ./spa-catchall.ts (unit-tested there).

// ── WebSocket + Server ───────────────────────────────────────────────────────

const server = createServer(app);

// ── WebSocket ────────────────────────────────────────────────────────────────

const wss = new WebSocketServer({ noServer: true });

interface WsSession {
  uid: string;
  email: string;
  /** Set on ws auth; undefined until then. */
  principal?: Principal;
  /** Client IP from the upgrade request (TRUSTED_PROXIES-aware). */
  ip?: string;
  busySessions: Set<string>;
  autoApprove: boolean;
  abortControllers: Map<string, AbortController>;
  pendingPermissions: Map<string, { resolve: (result: { allow: boolean }) => void; destructive?: boolean; tool?: string; input?: unknown; sessionId?: string }>;
  pendingQuestions: Map<string, { resolve: (answers: QuestionAnswers | null) => void }>;
  steerPending: Map<string, string>;
  lastSessionId: string | null;
  viewingSessionId: string | null;
  focused: boolean;
}

function isUserViewingSession(uid: string, sessionId: string, excludeWs?: WebSocket): boolean {
  for (const [ws, s] of activeConnections) {
    if (ws === excludeWs) continue;
    if (s.uid === uid && s.viewingSessionId === sessionId && s.focused) return true;
  }
  return false;
}

function isUserConnected(uid: string): boolean {
  for (const [, s] of activeConnections) {
    if (s.uid === uid) return true;
  }
  return false;
}

function notifyUnread(uid: string, sessionId: string, preview: string, source: 'response' | 'proactive' | 'schedule', title?: string, senderWs?: WebSocket) {
  if (senderWs) {
    const senderSession = activeConnections.get(senderWs);
    if (senderSession?.viewingSessionId === sessionId && senderSession.focused) return;
  }
  if (isUserViewingSession(uid, sessionId)) return;
  const entry = addUnread(uid, sessionId, preview, source, title);
  console.log(`[unread] ${source} notification for ${uid.slice(0, 8)} session=${sessionId.slice(0, 8)} count=${entry.count}`);
  for (const [ws, s] of activeConnections) {
    if (s.uid === uid) {
      send(ws, { type: 'unread', sessionId, count: entry.count, preview, source, title });
    }
  }
}

function sendUnreadSync(ws: WebSocket, uid: string) {
  const unreads = getUnreads(uid);
  if (Object.keys(unreads.sessions).length > 0) {
    send(ws, { type: 'unread_sync', sessions: unreads.sessions });
  }
}

// ── Sidecar WebSocket proxy ─────────────────────────────────────────────────

// Sidecar WS proxy routes (url-prefix → localhost port). CE owns NONE — every route is an add-on's
// own daemon (its prefix → its port), folded in from the `sidecarRoutes` feature seam after feature
// registration below (see `Object.assign(WS_PROXY_ROUTES, collectSidecarRoutes())`).
const WS_PROXY_ROUTES: Record<string, number> = {};

function resolveSidecarPort(urlPath: string): number | null {
  const prefix = urlPath.split('/').filter(Boolean)[0];
  return prefix ? WS_PROXY_ROUTES[prefix] ?? null : null;
}

/** Handshake subprotocol marker carrying the caller's token: `[WS_AUTH_PROTOCOL, <token>]`.
 * Keep in sync with `WS_AUTH_PROTOCOL` in src/client/lib/ws.ts (the client half). */
const WS_AUTH_PROTOCOL = 'shraga.auth';

// handleProtocols is explicit so the accepted subprotocol is always the marker — never the token
// itself (ws's default picks the first offered, and echoing a credential back in a response header
// would put it in any intermediary's log). Only consulted when the client offered protocols at all,
// so a header-authenticated non-browser client is unaffected.
const sidecarWss = new WebSocketServer({ noServer: true, handleProtocols: (protocols) => (protocols.has(WS_AUTH_PROTOCOL) ? WS_AUTH_PROTOCOL : false) });

/** Authenticate a WS upgrade. Token comes from the subprotocol list (the only browser-settable
 * handshake field), falling back to a bearer header for non-browser clients (curl/CLI/tests). */
async function authenticateWsUpgrade(req: import('node:http').IncomingMessage): Promise<import('./auth.ts').AuthUser | null> {
  const protocols = (req.headers['sec-websocket-protocol'] as string | undefined)?.split(',').map((s) => s.trim()) ?? [];
  const token = protocols[0] === WS_AUTH_PROTOCOL && protocols[1]
    ? protocols[1]
    : req.headers.authorization?.replace('Bearer ', '');
  return authenticateToken(token);
}

/** Reject a WS upgrade with a real HTTP response before destroying the socket — same shape as ws's own
 * internal `abortHandshake`. A bare `socket.destroy()` writes ZERO bytes, so the client sees a generic
 * "connection failed" and an expired token is indistinguishable from "server is down".
 *
 * CAVEAT, measured, do not assume this reaches the client today: under Bun (1.3.10, the runtime this
 * server actually runs on) the socket handed to a `node:http` 'upgrade' listener is write-dead — it
 * reports `writable: true` but has no `_handle`, and NOTHING written to it is ever delivered. Probed in
 * an isolated two-runtime rig (plain `createServer` + 'upgrade' listener, raw TCP client): Node 24
 * delivered the 401 for every variant (end / write / paused / unpaused); Bun 1.3.10 delivered zero bytes
 * for all of them, including a direct `_handle.write`. ws's own `abortHandshake` would fail identically —
 * it is the same `socket.end(...)`. So on Bun the observable behaviour is still a bare close.
 * This is kept because it is the correct shape, costs nothing, still closes the socket, and starts
 * working the moment Bun implements it (or the process runs under Node). Until then, do NOT build a
 * client-side "auth error" affordance on top of it — the client cannot see this response. */
function abortUpgrade(socket: import('node:stream').Duplex, code: number, message?: string) {
  if (socket.destroyed || !socket.writable) return;
  const body = message ?? STATUS_CODES[code] ?? '';
  const headers = ['Connection: close', 'Content-Type: text/plain', `Content-Length: ${Buffer.byteLength(body)}`];
  socket.once('finish', () => socket.destroy());
  // The socket was paused for the auth gap; pause() only stops READS, so it is still writable here.
  try { socket.end(`HTTP/1.1 ${code} ${STATUS_CODES[code] ?? 'Error'}\r\n${headers.join('\r\n')}\r\n\r\n${body}`); }
  catch { socket.destroy(); }
}

/** Claude Code speaks the kitty keyboard protocol, where `CSI <code>;<mods> u` IS a modified keypress —
 *  `CSI 107;9u` is cmd+k, which clears its screen and, arriving twice, runs `/clear` and destroys the
 *  conversation. Our terminal is xterm.js 5.5, which has no kitty support and can never emit that byte,
 *  so anything shaped like it on this socket is spurious, not a keystroke the user made. Drop it and say
 *  where it came from — repeated unexplained `/clear`s cost real sessions. Remove once a client that
 *  genuinely speaks the protocol attaches here. */
const CSI_U_KEY = /\x1b\[\d+;\d+u/;
function isSpuriousKittyKey(data: import('ws').RawData, isBinary: boolean): string | null {
  if (isBinary) return null;
  try {
    const msg = JSON.parse(data.toString());
    if (msg?.type !== 'input' || typeof msg.data !== 'string') return null;
    const input = Buffer.from(msg.data, 'base64').toString();
    return CSI_U_KEY.test(input) ? input : null;
  } catch { return null; }
}

function proxySidecarWebSocket(req: import('node:http').IncomingMessage, socket: import('node:stream').Duplex, head: Buffer, port: number, authed: Promise<import('./auth.ts').AuthUser | null>) {
  const targetUrl = `ws://127.0.0.1:${port}${req.url}`;
  sidecarWss.handleUpgrade(req, socket as any, head, (clientWs) => {
    // Belt-and-braces: nothing pauses this socket any more (the auth gap moved past the handshake, see
    // AUTH GATE below), but an explicitly paused socket does not re-enter flowing mode just because a
    // listener was added — so if anything upstream ever pauses again, resuming here is what keeps every
    // buffered byte and every subsequent keystroke from sitting unread forever.
    socket.resume();

    // AUTH GATE — deliberately AFTER the handshake, not before it. The upgrade must complete in the
    // SAME tick as the 'upgrade' event: on Bun (reproduced on 1.4.0) a `handleUpgrade` deferred past a
    // macrotask — exactly what awaiting a token verification costs on a cold key cache — lands in ws's
    // `completeUpgrade` abort path, and ws's own `abortHandshake` then throws
    // (`undefined is not an object (evaluating 'message')`). The browser sees a bare close, caddy logs
    // `502 EOF`, and every terminal pane stays blank. Isolated repro: a microtask-deferred handleUpgrade
    // succeeds, a `setTimeout(…, 0)`-deferred one throws — which is why this only bit when auth actually
    // hit the network. Security is unchanged: NOTHING is dialed or forwarded until the token verifies,
    // and a failed check closes the socket with 1008 before a single byte reaches the sidecar.
    let targetWs: WebSocket;
    let opened = false;
    // Anything typed between the 101 and the auth verdict. Bounded like `pending` below; on rejection it
    // is dropped with the socket.
    const preAuth: Array<{ data: import('ws').RawData; isBinary: boolean }> = [];
    let gateOpen = false;
    clientWs.on('message', (data, isBinary) => {
      if (gateOpen || preAuth.length >= 256) return; // bounded: never buffer unboundedly
      // Same kitty-key filter the bridged handler applies. These frames are flushed straight into
      // `pending`, so without it a spurious `CSI …u` sent during the auth window would reach the shell
      // by the one path that skips the guard.
      if (isSpuriousKittyKey(data, isBinary)) {
        console.warn('[ws-proxy] BLOCKED kitty key input (pre-auth)', JSON.stringify({ url: req.url?.split('?')[0], ua: req.headers['user-agent'] }));
        return;
      }
      preAuth.push({ data, isBinary });
    });
    void authed.then((user) => {
      if (clientWs.readyState !== WebSocket.OPEN) return;
      if (!user) {
        console.warn(`[ws-proxy] rejected unauthenticated upgrade for ${req.url?.split('?')[0]}`);
        clientWs.close(1008, 'unauthorized');
        return;
      }
      gateOpen = true;
      startBridge();
    });

    function startBridge() {
      targetWs = new WebSocket(targetUrl);

      // Bound IMMEDIATELY, not inside targetWs 'open': the BROWSER's socket is OPEN the moment
      // handleUpgrade returns, so it reports "connected" and starts sending while we're still dialing the
      // sidecar. Registering the listener on open discarded everything typed in that window — silently.
      // Queue instead, and flush in order once upstream is up.
      const pending: Array<{ data: import('ws').RawData; isBinary: boolean }> = preAuth.splice(0);
      clientWs.on('message', (data, isBinary) => {
        // App-level liveness probe (Layer 2): the client can't read protocol pongs from JS, so it sends
        // `{type:'ping'}` and expects `{type:'pong'}`. We RELAY it — we must not answer it here. A
        // proxy-local reply only proves THIS hop is alive: if the proxy→sidecar leg is half-open, or the
        // sidecar has already dropped this client from its subscriber set, the browser still gets pongs,
        // keeps `readyState === OPEN`, shows a green "connected" dot, and every keystroke disappears.
        // The probe is only worth anything end-to-end, so the sidecar owns the reply (para-pty answers in
        // its ws message handler); a sidecar that doesn't reply fails the probe, which is the honest
        // outcome — the client then reconnects rather than trusting a dead pipe.
        const spurious = isSpuriousKittyKey(data, isBinary);
        if (spurious) {
          console.warn('[ws-proxy] BLOCKED kitty key input', JSON.stringify({
            url: req.url?.split('?')[0], seq: spurious, ua: req.headers['user-agent'], ref: req.headers.referer,
          }));
          return;
        }
        if (targetWs.readyState === WebSocket.OPEN) targetWs.send(data, { binary: isBinary });
        else if (!opened && pending.length < 256) pending.push({ data, isBinary }); // bounded: never buffer unboundedly
      });

      targetWs.on('open', () => {
        opened = true;
        for (const m of pending) {
          if (targetWs.readyState === WebSocket.OPEN) targetWs.send(m.data, { binary: m.isBinary });
        }
        pending.length = 0;
        targetWs.on('message', (data, isBinary) => {
          if (clientWs.readyState === WebSocket.OPEN) clientWs.send(data, { binary: isBinary });
        });
      });

      // Keepalive: a proxied sidecar socket carries no app-level heartbeat, so an idle WS gets silently dropped
      // by an intermediary (Cloudflare tunnel idles WS at ~100s) leaving the BROWSER half-open — readyState
      // stays OPEN, no onclose fires, the "connected" dot stays green and keystrokes vanish into a dead pipe.
      // Ping the client (browsers auto-pong at the protocol level) to keep intermediaries from idling us out,
      // and terminate a peer that misses a pong so the client gets a real close → its reconnect kicks in.
      // Tolerate ONE missed pong before terminating (~2 intervals of grace): a backgrounded mobile tab is
      // JS/network-frozen and can't auto-pong for a cycle, so a 1-strike policy force-closed it every 30s and
      // churned reconnects. Two strikes lets a brief freeze ride through; a truly dead pipe still gets cut.
      let missedPongs = 0;
      clientWs.on('pong', () => { missedPongs = 0; });
      const pingInterval = setInterval(() => {
        if (clientWs.readyState !== WebSocket.OPEN) return;
        if (missedPongs >= 2) { console.warn('[ws-proxy] client missed pongs — terminating (likely backgrounded/frozen client)'); clientWs.terminate(); return; }
        missedPongs++;
        clientWs.ping();
      }, WS_PING_INTERVAL);

      targetWs.on('close', () => { clearInterval(pingInterval); clientWs.close(); });
      targetWs.on('error', (e) => {
        // Pre-open failure = the sidecar daemon is unreachable (e.g. it idle-exited, or is not up yet
        // after a restart). This is TRANSIENT: the daemon (and its shells) survive a server/proxy blip, and
        // we revive it right below — so flag `fatal:false`. The client must keep the pane alive and re-attach
        // (a mobile client that backgrounded for minutes recovers its still-running sidecar on resume), NOT show a
        // permanent "session unavailable". Only the daemon's own `session not found` (post-open) is fatal.
        if (!opened) {
          if (clientWs.readyState === WebSocket.OPEN) {
            try { clientWs.send(JSON.stringify({ type: 'error', message: 'sidecar daemon unavailable', fatal: false })); } catch { /* socket gone */ }
          }
        } else {
          console.warn('[ws-proxy] target error:', e.message);
        }
        clientWs.close();
      });
      clientWs.on('close', () => { clearInterval(pingInterval); if (targetWs.readyState === WebSocket.OPEN) targetWs.close(); });
      clientWs.on('error', (e) => { console.error(`[ws-proxy] client error:`, e.message); targetWs.close(); });
    }
  });
}
server.on('upgrade', (req, socket, head) => {
  if (req.url === '/ws') {
    wss.handleUpgrade(req, socket as any, head, (ws) => {
      const session: WsSession = { uid: '', email: '', ip: requestIp(req), busySessions: new Set(), autoApprove: false, abortControllers: new Map(), pendingPermissions: new Map(), pendingQuestions: new Map(), steerPending: new Map(), lastSessionId: null, viewingSessionId: null, focused: true };
      handleConnection(ws, session);
    });
  } else {
    const port = req.url ? resolveSidecarPort(req.url) : null;
    if (port) {
      // A sidecar socket is a LIVE ATTACHED SHELL (a terminal daemon) — read/write on the user's terminals. It
      // must be authenticated BEFORE we bridge it, like the HTTP routes (`requireAuth`) and the `/ws`
      // control socket (its `auth` message). NOTE the limit of this gate: it is IDENTITY-ONLY. It proves
      // the token belongs to a valid user; it does NOT check that the user owns (or may access) the
      // `sessionId` in the URL, so it is weaker than routes that additionally authorize — e.g. the HTTP
      // pty kill route, which checks group membership. Any authenticated user can currently attach to any
      // session id. Unlike those, there's no place to put a bearer header:
      // a browser can't set headers on a WebSocket handshake. The one field it CAN set is the
      // subprotocol list, so the client sends `[WS_AUTH_PROTOCOL, <token>]` (see ptyWsUrl's consumer) and
      // we verify entry 1 with the same token seam requireAuth uses. Works identically over the `.local`
      // HTTPS origin and the cloudflared tunnel — Sec-WebSocket-Protocol is a standard handshake header
      // both pass through, and unlike `?token=` it never lands in an access log or a Referer.
      // Pause for the async gap. Node hands us the raw socket with its own parser detached, so nothing
      // is reading it while we verify; pausing makes that explicit and guarantees bytes the client sends
      // between the handshake and our decision are buffered, not dropped. proxySidecarWebSocket resumes
      // it once ws owns the socket.
      // Fail CLOSED on anything: a rejected/thrown/slow auth must destroy the socket, never leave it
      // dangling. Without the catch a throw in authenticateWsUpgrade (token store unavailable, malformed
      // header) produced an unhandled rejection AND an open, unauthenticated, un-proxied socket; without
      // the timeout a hung verifier held the socket and its fd open indefinitely.
      // The catch sits on the auth promise itself rather than on the race, so a rejection that arrives
      // AFTER the timeout already won is visibly logged instead of vanishing into the race's own
      // (already-settled) internal handler.
      const authed = authenticateWsUpgrade(req)
        .catch((err) => { console.error('[ws-proxy] auth threw for upgrade:', (err as Error)?.message ?? err); return null; });
      // Cleared on settle: an uncleared 5s timer pins this socket/req/head closure alive for 5s per
      // upgrade even when auth resolved in milliseconds.
      let authTimer: ReturnType<typeof setTimeout> | undefined;
      const timedOut = new Promise<null>((resolve) => { authTimer = setTimeout(() => resolve(null), WS_AUTH_TIMEOUT_MS); authTimer.unref?.(); });
      const verdict = Promise.race([authed, timedOut]).then((user) => { if (authTimer) clearTimeout(authTimer); return user; });
      // The handshake is NOT deferred behind that verification: on Bun (reproduced on 1.4.0) a
      // `handleUpgrade` deferred past a macrotask lands in ws's `completeUpgrade` abort path, and ws's
      // own `abortHandshake` then throws (`undefined is not an object (evaluating 'message')`). The
      // browser sees a bare close, caddy logs `502 EOF`, and every terminal pane stays blank — and an
      // UNAUTHENTICATED upgrade hangs instead of being refused. Complete the upgrade now, hand the
      // in-flight verification down, and let it decide whether to bridge or close (see the AUTH GATE
      // note in proxySidecarWebSocket).
      proxySidecarWebSocket(req, socket, head, port, verdict);
    } else {
      // No sidecar registered for this path — a client error, not an auth failure.
      abortUpgrade(socket, 400, 'Unknown upgrade path');
    }
  }
});

function send(ws: WebSocket, data: object) {
  if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(data));
}

const DESTRUCTIVE_PERMISSION_TTL = 10 * 60_000;
const WS_PING_INTERVAL = 30_000;
const WS_AUTH_TIMEOUT_MS = 5_000; // upper bound on sidecar-upgrade auth; past it the socket is destroyed
const activeConnections = new Map<WebSocket, WsSession>();
const globalPendingPermissions = new Map<string, { resolve: (r: { allow: boolean }) => void; sessionId: string; tool: string; input: unknown; uid: string }>();
function isSessionBusy(sid: string): boolean {
  return isSessionLocked(sid);
}
function broadcast(data: object, exclude?: WebSocket) {
  for (const ws of activeConnections.keys()) {
    if (ws !== exclude) send(ws, data);
  }
}

setBroadcaster(broadcast); // let session-bus push async events (e.g. an add-on's background worker output) to clients
ensureWorkspaceDir();
watchWorkspace((event) => broadcast({ type: 'workspace_change', ...event }));
if (!PASSIVE) {
  // BEFORE anything that could take time: the gap is measured as (now - last heartbeat), so it must
  // be read and re-stamped while "now" still means boot — a slow start would otherwise inflate the
  // gap, or (once startHeartbeat is running) erase it. Order vs the scheduler's catch-up doesn't
  // matter for the join: missedSchedules() reads the two files independently, at report time.
  // Passive twins never write here — single-active-writer, same rule as the scheduler.
  recordBootGap();
  startHeartbeat();
  scheduler.start(broadcast);
  startEventDispatcher();
  // Modules reconcile MUST follow scheduler.start(): upsertSchedule mutates the engine's
  // in-memory list, which is empty (and would be saved over schedules.json) before start.
  // Skipped when passive (single-active-writer: no data/ mutations from a standby twin).
  try { reconcileInstalledModules(); } catch (err) { console.error('[modules] boot reconcile failed:', (err as Error).message); }
}
// Host telemetry is read-only (no persistence, unref'd timer) — not a writer or consumer, so it
// runs in passive too. Otherwise a standby instance reports empty stats and /api/stats is a lie.
statsSampler.start(broadcast);
registerEventRoutes(app, requireAuth);
/** How long an out-of-band wake turn waits for a busy session before giving up (see runTurn). */
const WAKE_LOCK_WAIT_MS = 5 * 60_000;
initPolls({
  broadcast,
  // The out-of-band turn runner (polls + background-job follow-ups). It TAKES THE SESSION LOCK, and
  // that is load-bearing: `streamChat` never acquires one itself, so before this every guard written
  // against `isSessionLocked` — including background-jobs.ts's "never start a turn on top of a live
  // one" — was reading a lock that this path never took. Two jobs finishing minutes apart then ran
  // two concurrent wake turns in one session, each free to dispatch the next leg of the same
  // workflow: two writers on one browser instance, which is the failure these workflows are built to
  // prevent. Waiting (bounded) rather than failing fast, because the whole point of a wake is that
  // the outcome gets told: a busy session usually means another wake is mid-turn and will be done in
  // seconds.
  runTurn: async ({ prompt, sessionId, uid, userEmail }) => {
    const abortController = new AbortController();
    const deadline = Date.now() + WAKE_LOCK_WAIT_MS;
    while (!acquireSessionLock(sessionId, 'api', abortController)) {
      if (Date.now() >= deadline) {
        // Give up by returning NOTHING, never by throwing. A throw propagates out of wake.ts's
        // unguarded `await runTurn` into background-jobs' catch, which records `reported: 'failed'`
        // — the one delivery path with no raw fallback, so the job's outcome would reach the user in
        // no form at all, after wake.ts had already appended the trigger prompt (a question with no
        // answer in the transcript). Empty blocks are the 'no-output' contract callers already
        // handle: the job store then delivers its raw report instead. Degraded, but never silent.
        console.warn(`[wake] session ${sessionId} stayed busy for ${Math.round(WAKE_LOCK_WAIT_MS / 1000)}s — skipping the turn; the caller falls back to a raw report`);
        return [];
      }
      await new Promise((r) => setTimeout(r, 2_000));
    }
    try {
      // Stream it. A woken turn (a background job reporting back, a poll firing) used to surface
      // NOTHING until it ended — and an add-on engine's subagent events were dropped outright — so a
      // turn that dispatched a worker was indistinguishable from one that only claimed to.
      return await consumeStream(
        streamChat({ principal: fromInternal({ uid, email: userEmail, lane: 'wake' }), prompt, sessionId, uid, userEmail, mcpServers: getMcpConfig(uid), abortController, onPermissionRequest: async () => ({ allow: true }), scheduled: isScheduledSession(sessionId) }),
        undefined,
        {
          maxResultChars: STREAMED_RESULT_MAX,
          onDelta: (event) => broadcast({ type: 'session_stream', sessionId, event }),
          onPassthrough: (event) => emitToSession(sessionId, event),
        },
      );
    } finally {
      if (releaseSessionLock(sessionId, abortController)) setRunStatus(sessionId, 'idle');
    }
  },
});
// Background jobs outlive the turn that started them, so their follow-up must too. Must run AFTER
// initPolls (which wires wake.ts's turn runner) — boot adoption can report a job that finished
// while we were down, and that report runs a turn. Skipped when passive: a standby twin must not
// adopt the active instance's children or double-report them.
if (!PASSIVE) initBackgroundJobs();
// Remote-push triggers: subscribe to schedule.finished and expose turn-done/question
// hooks. isForeground reuses the existing presence tracking (see isUserViewingSession).
initPushTriggers({
  origin: getPublicOrigin(),
  isForeground: (uid, sessionId) => isUserViewingSession(uid, sessionId),
});
// Optional add-ons (voice, github, gmail, fleet, …) mount here through the feature seam.
// The core registers none; an optional add-on calls registerFeature(...) before startup.
// SHRAGA_OVERLAY points at an external add-on module (outside the core tree) that imports
// features.ts and registerFeature(...)s its add-ons at import time. Guarded so a missing/broken
// add-on logs and never crashes the core.
if (process.env.SHRAGA_OVERLAY) {
  try {
    // Resolve relative to CWD (not this module) so a path like ../my-extensions/index.ts works as typed.
    const overlaySpec = path.isAbsolute(process.env.SHRAGA_OVERLAY)
      ? process.env.SHRAGA_OVERLAY
      : path.resolve(process.cwd(), process.env.SHRAGA_OVERLAY);
    await import(overlaySpec);
    console.log(`[overlay] loaded ${process.env.SHRAGA_OVERLAY}`);
  } catch (err) {
    console.error(`[overlay] failed to load ${process.env.SHRAGA_OVERLAY}:`, (err as Error)?.stack || err);
  }
}
// Slack ships in this app — register it through the same feature seam add-ons use.
// Programmatic features register through the same seam an overlay uses — BEFORE mountFeatures()
// so their routes mount ahead of the SPA fallback, identical to the overlay path.
for (const f of __reg.features ?? []) registerFeature(f);
registerFeature(slackFeature);
mountFeatures({ app, requireAuth, broadcast, passive: PASSIVE });
// Fold in feature-contributed sidecar WS proxy routes (the core names none; each add-on adds its own).
Object.assign(WS_PROXY_ROUTES, collectSidecarRoutes());

// SPA fallback — MUST be the last GET route so it never shadows real API/feature/extension routes.
registerSpaCatchAll(app, distPath);

// ── Runtime promotion (blue-green flip) ──────────────────────────────────────
// A passive instance can be promoted to active once traffic has been flipped to it:
// starts every consumer/writer that passive boot skipped. One-way; idempotent-guarded.
async function activateConsumers() {
  activated = true;
  console.log('[server] ACTIVATING — starting consumers and background writers');
  security()?.activate(); // passive Policy never wrote; now re-load from disk as trusted (migrates if still missing)
  await dataSync.init();
  bootDataSync();
  syncVendorRepos().catch(err => console.warn('[vendor-sync] error:', (err as Error).message));
  recordBootGap();
  startHeartbeat();
  scheduler.start(broadcast);
  startEventDispatcher();
  mountFeatures({ app, requireAuth, broadcast, passive: false });
  // Re-place the SPA catch-all AFTER the promotion's feature mount so newly-added GET routes win.
  registerSpaCatchAll(app, distPath);
  startSidecars().catch(err => console.error('[sidecar] startup error:', err));
  recoverInterruptedSessions().catch(err => console.error('[recovery] failed:', err));
}
app.post('/internal/activate', async (req, res) => {
  const token = req.headers['x-internal-token'] as string | undefined;
  if (!token || token !== process.env.INTERNAL_API_TOKEN) return res.sendStatus(403);
  if (activated) return res.status(409).json({ error: 'already active' });
  await activateConsumers();
  res.json({ ok: true });
});

// Constant-time equality for two strings (avoids leaking length/content via timing).
function safeStrEqual(a: string, b: string): boolean {
  const ab = Buffer.from(a), bb = Buffer.from(b);
  return ab.length === bb.length && timingSafeEqual(ab, bb);
}

// Verify GitHub's HMAC signature (x-hub-signature-256 = "sha256=" + HMAC-SHA256(secret, RAW BODY)).
// Must run over the EXACT bytes GitHub sent, captured as req.rawBody by express.json's verify hook —
// re-serializing req.body would change the bytes and never match.
function validGithubSignature(req: express.Request, secret: string): boolean {
  const header = req.headers['x-hub-signature-256'];
  if (typeof header !== 'string') return false;
  const raw = (req as any).rawBody as Buffer | undefined;
  if (!raw) return false;
  const expected = 'sha256=' + createHmac('sha256', secret).update(raw).digest('hex');
  return safeStrEqual(header, expected);
}

// Data-sync webhook, ported to shraga's first-class registerWebhook convention: the framework mounts
// the PUBLIC route, captures rawBody, runs `verify` (the ONLY per-vendor piece — the HMAC below),
// rejects on a falsy result, and on a pass emits the 'data-sync.pull' event. The subscribeEvent below
// (gated on active/non-passive) consumes it and calls dataSync.pull(). Mounted on `app` at the
// SAME position the bespoke route occupied — before the SPA catch-all — via the `path` option, so the
// GitHub webhook needs no reconfig.
//
// Semantic deltas vs the old inline route (documented, deliberate):
//   • bad signature → 400 (was 403): registerWebhook's convention is 400 for any falsy verify. GitHub
//     treats any non-2xx as a failed delivery, so redelivery/behavior is unchanged.
//   • not-activated / data-sync disabled → 200 (was 404): the route now always accepts + emits; the
//     PULL is what's gated. In passive/standby no pull handler is subscribed (single-active-writer),
//     and pull() itself no-ops when disabled — so no data is mutated. The webhook only ever targets
//     the active instance, so the visible-status change is inert in practice.
//   • response no longer awaits the pull: 200 is returned on emit and pull() runs async. This is the
//     convention (and safer — a git pull must not block GitHub's ~10s webhook timeout).
registerWebhook(app, {
  source: 'data-sync.pull',
  path: '/api/data-sync/webhook',
  // Reuse the verified-correct crypto below verbatim — signature-only; activation/enablement gating
  // lives on the pull handler, not here.
  verify: (req) => {
    const secret = process.env.DATA_SYNC_WEBHOOK_SECRET;
    if (!secret) return true; // secret unset → open (current live behavior)
    // Accept EITHER a valid GitHub HMAC signature OR the legacy plain header (backward-compat).
    const plain = req.headers['x-webhook-secret'];
    return validGithubSignature(req, secret) || (typeof plain === 'string' && safeStrEqual(plain, secret));
  },
});

// Consume the verified webhook's event → pull. The PULL is gated on active (non-passive) via the live
// `activated`/`PASSIVE` closure: a passive/standby twin shares DATA_DIR and must NOT mutate data/
// (single-active-writer), and before promotion there's nothing to pull into serving. `activated` flips
// true in activateConsumers(), so a promoted instance starts pulling with no extra wiring. pull() also
// self-no-ops when data-sync is disabled.
subscribeEvent('data-sync.pull', () => {
  if (PASSIVE || !activated) return;
  dataSync.pull().catch((err) => console.error('[data-sync] webhook pull failed:', (err as Error).message));
});

async function runStream(ws: WebSocket, session: WsSession, sid: string, promptText: string, attachments: AttachmentMeta[] | undefined, mcpServers: McpConfig, isSteerRestart = false, conversationReset = false, turnHints?: Record<string, unknown>) {
  const abortController = new AbortController();
  if (!isSteerRestart) {
    if (!acquireSessionLock(sid, 'web', abortController)) {
      console.warn(`[ws] Session ${sid.slice(0, 8)} already locked, rejecting`);
      send(ws, { type: 'error', message: 'Session is already processing a request (from another source)', sessionId: sid });
      session.busySessions.delete(sid);
      return;
    }
  } else {
    replaceSessionLock(sid, 'web', abortController);
  }
  session.abortControllers.set(sid, abortController);
  session.steerPending.delete(sid);
  send(ws, { type: 'session_busy', sessionId: sid, busy: true });
  broadcast({ type: 'session_busy', sessionId: sid, busy: true }, ws);

  // An unattended turn (e.g. a voice-originated one) has nobody watching the UI to click Allow, so
  // auto-approve for the whole run. Generic hint off the opaque bag — the core names no add-on concept.
  const unattended = turnHints?.unattended === true;

  const onPermissionRequest: PermissionHandler = (id, tool, input) => {
    if (session.autoApprove || unattended) {
      console.log(`[ws] Auto-approved ${tool} id=${id}${unattended ? ' (unattended)' : ''}`);
      return Promise.resolve({ allow: true });
    }
    if (ws.readyState !== WebSocket.OPEN) {
      console.log(`[ws] Auto-approved ${tool} id=${id} (client disconnected)`);
      return Promise.resolve({ allow: true });
    }
    return new Promise<{ allow: boolean }>((resolve) => {
      session.pendingPermissions.set(id, { resolve });
      send(ws, { type: 'permission_request', id, tool, input, sessionId: sid });
      console.log(`[ws] Permission request for ${tool} id=${id}`);
    });
  };

  const onUserQuestion: QuestionHandler = (id, questions) => {
    if (ws.readyState !== WebSocket.OPEN) {
      console.log(`[ws] Question id=${id} skipped (client disconnected) — agent self-decides`);
      return Promise.resolve(null);
    }
    try { pushQuestion(getSession(sid)?.uid || session.uid, sid); }
    catch (err) { console.error('[push] question trigger failed:', err); }
    return new Promise<QuestionAnswers | null>((resolve) => {
      session.pendingQuestions.set(id, { resolve });
      send(ws, { type: 'question_request', id, questions, sessionId: sid });
      console.log(`[ws] Question request id=${id} (${questions.length}q)`);
    });
  };

  const isFirstTurn = loadConversation(sid).length === 0;
  let assistantText = '';
  let thinkingText = '';
  const assistantBlocks: ConvBlock[] = [];
  let saved = false;

  const collectPartialBlocks = () => [
    ...assistantBlocks,
    ...(thinkingText ? [{ type: 'thinking' as const, text: thinkingText }] : []),
    ...(assistantText ? [{ type: 'text' as const, text: assistantText }] : []),
  ];

  const flushAssistant = () => {
    if (saved) return;
    saved = true;
    clearPartial(sid);
    if (thinkingText) { assistantBlocks.push({ type: 'thinking', text: thinkingText }); thinkingText = ''; }
    if (assistantText) assistantBlocks.push({ type: 'text', text: assistantText });
    if (assistantBlocks.length === 0) return;
    appendMessage(sid, { id: crypto.randomUUID(), role: 'assistant', blocks: assistantBlocks });
    console.log(`[ws] Saved assistant (${assistantBlocks.length} blocks) for ${sid.slice(0, 8)}`);
  };

  registerLivePartial(sid, collectPartialBlocks);
  const partialInterval = setInterval(() => {
    const blocks = collectPartialBlocks();
    if (blocks.length) writePartial(sid, blocks);
  }, 5_000);

  setRunStatus(sid, 'running', 'web');

  let stopReason = '';
  try {
    let eventCount = 0;
    for await (const event of streamChat({
      principal: session.principal ?? fromInternal({ uid: session.uid, email: session.email, lane: 'ws-unresolved' }),
      prompt: promptText,
      attachments,
      sessionId: sid,
      uid: session.uid,
      userEmail: session.email,
      userName: session.email.split('@')[0],
      mcpServers,
      abortController,
      conversationReset,
      turnHints,
      context: { source: 'web', user: session.email },
      onPermissionRequest,
      onUserQuestion,
      onDestructiveApproval: (id, tool, input) => {
        if (unattended) {
          console.log(`[ws] Auto-approved destructive ${tool} id=${id} (unattended)`);
          return Promise.resolve({ allow: true });
        }
        return new Promise<{ allow: boolean }>((resolve) => {
          const ttl = ws.readyState !== WebSocket.OPEN ? DESTRUCTIVE_PERMISSION_TTL : undefined;
          session.pendingPermissions.set(id, { resolve, destructive: true, tool, input, sessionId: sid });
          if (ws.readyState === WebSocket.OPEN) {
            send(ws, { type: 'permission_request', id, tool, input, sessionId: sid });
          } else {
            globalPendingPermissions.set(id, { resolve, sessionId: sid, tool, input, uid: session.uid });
          }
          if (ttl) setTimeout(() => {
            if (globalPendingPermissions.delete(id)) {
              console.log(`[ws] Destructive ${tool} id=${id} denied after TTL (client didn't reconnect)`);
              resolve({ allow: false });
            }
          }, ttl);
          console.log(`[ws] Destructive op approval required for ${tool} id=${id}${ws.readyState !== WebSocket.OPEN ? ` (queued for reconnect, ${ttl! / 1000}s TTL)` : ''}`);
        });
      },
    })) {
      eventCount++;
      if (event.type !== 'done' && event.type !== 'error') send(ws, { ...event, sessionId: sid });

      if (event.type === 'thinking_delta') {
        thinkingText += event.text;
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'thinking_delta', text: event.text } }, ws);
      } else if (event.type === 'text_delta') {
        if (thinkingText) { assistantBlocks.push({ type: 'thinking', text: thinkingText }); thinkingText = ''; }
        assistantText += event.text;
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'text_delta', text: event.text } }, ws);
      } else if (event.type === 'tool_use') {
        if (thinkingText) { assistantBlocks.push({ type: 'thinking', text: thinkingText }); thinkingText = ''; }
        if (assistantText) {
          assistantBlocks.push({ type: 'text', text: assistantText });
          assistantText = '';
        }
        assistantBlocks.push({ type: 'tool_use', tool: event.tool, toolUseId: event.toolUseId, input: event.input });
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_use', tool: event.tool, toolUseId: event.toolUseId, input: event.input } }, ws);
        const artifactEvent = handleArtifactToolUse(sid, event.tool, event.input);
        if (artifactEvent) send(ws, artifactEvent);
      } else if (event.type === 'tool_use_input') {
        const existing = assistantBlocks.find((b: any) => b.type === 'tool_use' && b.toolUseId === event.toolUseId) as any;
        if (existing) existing.input = event.input;
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_use_input', toolUseId: event.toolUseId, input: event.input } }, ws);
        if (existing?.tool) {
          const artifactEvent = handleArtifactToolUse(sid, existing.tool, event.input);
          if (artifactEvent) send(ws, artifactEvent);
        }
      } else if (event.type === 'tool_result') {
        assistantBlocks.push({ type: 'tool_result', toolUseId: event.toolUseId, output: event.output });
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_result', toolUseId: event.toolUseId, output: event.output } }, ws);
      } else if (event.type === 'tool_result_image') {
        assistantBlocks.push({ type: 'image', src: event.dataUrl });
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_result_image', toolUseId: event.toolUseId, dataUrl: event.dataUrl } }, ws);
      } else if (event.type === 'done') {
        stopReason = event.stopReason ?? 'end_turn';
        if (stopReason === 'max_turns_reached') {
          assistantBlocks.push({ type: 'text', text: MAX_TURNS_NOTICE });
        }
        if (!assistantText && !thinkingText && assistantBlocks.length === 0 && !event.builtinHandled) {
          const fallback = '⚠️ No response was generated. Try rephrasing or sending again.';
          assistantText = fallback;
          send(ws, { type: 'text_delta', text: fallback, sessionId: sid });
          broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'text_delta', text: fallback } }, ws);
        }
        flushAssistant();
        console.log(`[ws] Done for ${session.email}: sessionId=${sid.slice(0, 8)} events=${eventCount} stopReason=${stopReason}`);
        upsertSession(sid, promptText, { uid: session.uid, email: session.email });
        send(ws, { type: 'done', sessionId: sid, stopReason });
        broadcast({ type: 'session_messages_changed', sessionId: sid }, ws);

        // Notify owner if they're not viewing this session
        const meta = getSession(sid);
        const preview = assistantText.slice(0, 120) || '(completed)';
        notifyUnread(session.uid, sid, preview, 'response', meta?.title, ws);

        // Generate a better title after first turn
        if (isFirstTurn && assistantText) {
          generateSessionTitle(sid, promptText, assistantText).then((title) => {
            if (title) {
              send(ws, { type: 'session_title_updated', sessionId: sid, title });
              broadcast({ type: 'session_title_updated', sessionId: sid, title }, ws);
            }
          });
        }
        break;
      } else if (event.type === 'error') {
        if (session.steerPending.has(sid)) {
          console.log(`[ws] Suppressing error during steer for ${sid.slice(0, 8)}`);
        } else {
          console.error(`[ws] Error event for ${session.email}: ${event.message}`);
          // Persist alongside whatever partial output the turn produced — flushAssistant() saves it.
          if (thinkingText) { assistantBlocks.push({ type: 'thinking', text: thinkingText }); thinkingText = ''; }
          if (assistantText) { assistantBlocks.push({ type: 'text', text: assistantText }); assistantText = ''; }
          assistantBlocks.push({ type: 'error', text: event.message });
          send(ws, { type: 'error', message: event.message, sessionId: sid });
        }
        break;
      } else if (!TURN_CONTROL_EVENTS.has(event.type)) {
        // Not core-owned — an add-on engine's event (subagent pills, worker cards). The owning
        // socket already got it verbatim above; mirror it to the session's OTHER viewers too, so a
        // second device watching the same turn sees the same thing.
        broadcast({ ...event, sessionId: sid }, ws);
      }
    }
    if (eventCount === 0) {
      console.warn(`[ws] Stream yielded 0 events for ${session.email}`);
      send(ws, { type: 'error', message: 'No response from agent — check server logs', sessionId: sid });
    }
  } catch (err: any) {
    if (session.steerPending.has(sid)) {
      console.log(`[ws] Stream aborted for steer in ${sid.slice(0, 8)}`);
    } else {
      stopReason = 'error';
      const msg = err.message || String(err);
      console.error(`[ws] Stream error for ${session.email}:`, msg);
      // Flush partial output first — flushAssistant() appends it in the finally, which would
      // otherwise order the failure ahead of the text it followed.
      if (thinkingText) { assistantBlocks.push({ type: 'thinking', text: thinkingText }); thinkingText = ''; }
      if (assistantText) { assistantBlocks.push({ type: 'text', text: assistantText }); assistantText = ''; }
      assistantBlocks.push({ type: 'error', text: msg });
      send(ws, { type: 'error', message: msg, sessionId: sid });
    }
  } finally {
    clearInterval(partialInterval);
    unregisterLivePartial(sid);
    flushAssistant();
    session.abortControllers.delete(sid);

    const steerText = session.steerPending.get(sid);
    session.steerPending.delete(sid);
    if (steerText) {
      appendMessage(sid, { id: crypto.randomUUID(), role: 'user', blocks: [{ type: 'text', text: steerText }], channel: 'web', senderName: session.email.split('@')[0] });
      console.log(`[ws] Restarting stream with steer for ${sid.slice(0, 8)}`);
      // Preserve the turn hints (e.g. unattended auto-approve) across a steer-restart, else auto-approve is lost mid-turn and prompts hang.
      await runStream(ws, session, sid, steerText, undefined, mcpServers, true, false, turnHints);
      return;
    }

    releaseSessionLock(sid);
    const okReasons = ['', 'end_turn', 'success'];
    const resolvedStop = stopReason === 'max_turns_reached' ? 'max_turns_reached'
      : (!okReasons.includes(stopReason)) ? 'error' : undefined;
    setRunStatus(sid, 'idle', undefined, resolvedStop);
    session.busySessions.delete(sid);
    broadcast({ type: 'session_busy', sessionId: sid, busy: false });
    // Turn-done remote push (owner only; suppressed if they're foregrounding this session).
    try { pushTurnDone(getSession(sid)?.uid || session.uid, sid); }
    catch (err) { console.error('[push] turn-done trigger failed:', err); }
    for (const [id, perm] of globalPendingPermissions) {
      if (perm.sessionId === sid) globalPendingPermissions.delete(id);
    }
  }
}

function handleConnection(ws: WebSocket, session: WsSession) {
  console.log('[ws] New connection');

  // Tolerate ONE missed pong before terminating (~2 intervals of grace) — see the ws-proxy keepalive note:
  // a 1-strike policy force-closed backgrounded/frozen mobile clients every 30s, triggering reconnect churn
  // (and, paired with an OS tab-discard, the reload + "Verifying" + sidecar re-attach the user saw).
  let missedPongs = 0;
  ws.on('pong', () => { missedPongs = 0; });
  const pingInterval = setInterval(() => {
    if (missedPongs >= 2) { console.warn(`[ws] client missed pongs — terminating (${session.email || 'unauth'})`); ws.terminate(); return; }
    missedPongs++;
    ws.ping();
  }, WS_PING_INTERVAL);

  ws.on('message', async (raw) => {
    let msg: any;
    try {
      msg = JSON.parse(raw.toString());
    } catch {
      return send(ws, { type: 'error', message: 'Invalid JSON' });
    }

    if (msg.type === 'auth') {
      try {
        const user = await verifyBearer(msg.token); // pluggable (local|firebase), not firebase-only
        session.uid = user.uid;
        session.email = user.email;
        session.principal = user.principal;
        security()?.authAllow(user.principal, 'ws:bearer');
        session.autoApprove = getAutoApprove(user.uid);
        console.log(`[ws] Authenticated: ${user.email} (${user.uid}) autoApprove=${session.autoApprove}`);
        send(ws, { type: 'auth_ok', uid: user.uid, email: user.email, buildId: SERVER_BUILD_ID });
        activeConnections.set(ws, session);
        sendUnreadSync(ws, user.uid);
        // Re-send any orphaned permission requests waiting for this user
        for (const [id, perm] of globalPendingPermissions) {
          if (perm.uid === user.uid) {
            session.pendingPermissions.set(id, { resolve: perm.resolve, destructive: true });
            send(ws, { type: 'permission_request', id, tool: perm.tool, input: perm.input, sessionId: perm.sessionId });
            globalPendingPermissions.delete(id);
            console.log(`[ws] Re-sent orphaned permission ${id} (${perm.tool}) to reconnected ${user.email}`);
          }
        }
      } catch (err: any) {
        security()?.authDeny('ws:bearer', err?.message?.includes('whitelist') ? 'not-whitelisted' : err?.message?.includes('revoked') ? 'token-revoked' : 'invalid-bearer', undefined, deniedPrincipalOf(err));
        console.error(`[ws] Auth failed:`, err.message);
        send(ws, { type: 'auth_error', message: err.message });
        ws.close();
      }
      return;
    }

    if (!session.uid) return send(ws, { type: 'error', message: 'Not authenticated' });

    if (msg.type === 'permission_response') {
      if (msg.allowAll) {
        session.autoApprove = true;
        setAutoApprove(session.uid, true);
        console.log(`[ws] Auto-approve enabled and persisted for ${session.email}`);
      }
      const entry = session.pendingPermissions.get(msg.id);
      if (entry) {
        session.pendingPermissions.delete(msg.id);
        entry.resolve({ allow: !!msg.allow });
      }
      return;
    }

    if (msg.type === 'question_response') {
      const entry = session.pendingQuestions.get(msg.id);
      if (entry) {
        session.pendingQuestions.delete(msg.id);
        entry.resolve((msg.answers as QuestionAnswers) ?? null);
        console.log(`[ws] Question ${msg.id} answered`);
      }
      return;
    }

    if (msg.type === 'presence') {
      session.viewingSessionId = msg.sessionId || null;
      session.focused = !!msg.focused;
      return;
    }

    // Native wrapper foreground signal (push suppression). Reuses the same presence
    // fields so triggers' isUserViewingSession() sees a native client like a web one.
    if (msg.type === 'client_presence') {
      session.viewingSessionId = msg.visible ? (msg.sessionId || null) : null;
      session.focused = !!msg.visible;
      return;
    }

    if (msg.type === 'mark_read') {
      const sid = msg.sessionId;
      if (sid) {
        markUnread(session.uid, sid);
        for (const [otherWs, s] of activeConnections) {
          if (s.uid === session.uid && otherWs !== ws) {
            send(otherWs, { type: 'unread_cleared', sessionId: sid });
          }
        }
      }
      return;
    }

    if (msg.type === 'steer') {
      const steerSid = msg.sessionId || session.lastSessionId;
      if (!steerSid) return;
      const localAc = session.abortControllers.get(steerSid);
      const globalAc = !localAc ? getSessionAbortController(steerSid) : null;
      const ac = localAc || globalAc;
      if (!ac) return;
      const steerText = msg.text ?? '';
      console.log(`[ws] Steer from ${session.email}: "${steerText.slice(0, 80)}" session=${steerSid.slice(0, 8)}`);
      if (localAc) {
        session.steerPending.set(steerSid, steerText);
        ac.abort();
      } else {
        // A takeover starts a NEW turn on this connection — same guard as a normal WS turn, checked BEFORE the abort
        // so a denied steer leaves the running turn alone.
        const admission = admitTurn(session.principal ?? fromInternal({ uid: session.uid, email: session.email, lane: 'ws-unresolved' }), { ip: session.ip, channel: 'ws' });
        if (!admission.ok) {
          return send(ws, { type: 'error', sessionId: steerSid, reason: admission.reason, retryAfter: admission.retryAfter,
            message: admission.status === 403 ? 'Access blocked' : `Too many requests — try again in ${admission.retryAfter ?? 60}s` });
        }
        try {
          ac.abort();
          appendMessage(steerSid, { id: crypto.randomUUID(), role: 'user', blocks: [{ type: 'text', text: steerText }], channel: 'web', senderName: session.email.split('@')[0] });
          console.log(`[ws] External steer takeover for ${steerSid.slice(0, 8)}`);
          session.busySessions.add(steerSid);
          session.lastSessionId = steerSid;
          await runStream(ws, session, steerSid, steerText, undefined, getMcpConfig(session.uid), true);
        } finally { admission.release(); }
      }
      return;
    }

    if (msg.type === 'message') {
      if (_draining) return send(ws, { type: 'error', message: 'Server is restarting — please retry in a moment' });
      let sid = msg.sessionId || crypto.randomUUID();
      const wantsFork = typeof msg.truncateAt === 'number' && msg.truncateAt >= 0 && (session.busySessions.has(sid) || isSessionLocked(sid));
      if (!wantsFork && (session.busySessions.has(sid) || isSessionLocked(sid))) return send(ws, { type: 'error', message: 'Already processing a request', sessionId: sid });
      // Guard before any spend: blocklist → rate → concurrency (shadow unless SECURITY_ENFORCE).
      const admission = admitTurn(session.principal ?? fromInternal({ uid: session.uid, email: session.email, lane: 'ws-unresolved' }), { ip: session.ip, channel: 'ws' });
      if (!admission.ok) {
        return send(ws, { type: 'error', sessionId: sid, reason: admission.reason, retryAfter: admission.retryAfter,
          message: admission.status === 403 ? 'Access blocked' : `Too many requests — try again in ${admission.retryAfter ?? 60}s` });
      }
      try {
        if (!wantsFork) session.busySessions.add(sid);
        const mcpServers = getMcpConfig(session.uid);
        // Opaque per-send bag from the client's send-options slot. The core forwards it verbatim to the
        // turn-context/engine seams and interprets no add-on key (a plain object only — never an array/primitive).
        const turnHints = msg.turnHints && typeof msg.turnHints === 'object' && !Array.isArray(msg.turnHints)
          ? (msg.turnHints as Record<string, unknown>) : undefined;
        // Ephemeral user turn (generic hint): a synthetic, add-on-originated opener (e.g. a voice greeting).
        // Run it, but DON'T persist it as a user message — the reply is the real turn, not a user turn.
        const ephemeralUser = turnHints?.ephemeralUser === true;
        const promptText = msg.text ?? '';
        session.lastSessionId = sid;
        session.viewingSessionId = sid;
        session.focused = true;
        const isNew = !msg.sessionId;
        if (isNew) {
          upsertSession(sid, promptText, { uid: session.uid, email: session.email });
          send(ws, { type: 'session_id', sessionId: sid });
          // Immediate prompt-derived title so the tab renames off "New Chat" now; the LLM title refines it later.
          const t0 = getSession(sid)?.title;
          if (t0) send(ws, { type: 'session_title_updated', sessionId: sid, title: t0 });
          console.log(`[ws] New session ${sid.slice(0, 8)} for ${session.email}`);
        }

        console.log(`[ws] Message from ${session.email}: "${promptText.slice(0, 100)}" session=${sid.slice(0, 8)}`);

        // Truncate or fork conversation if replaying/editing a previous message
        if (typeof msg.truncateAt === 'number' && msg.truncateAt >= 0) {
          if (wantsFork) {
            // Session is busy — fork instead of destructive truncate to avoid race conditions
            let forkedId: string | null = null;
            if (msg.truncateAt > 0) {
              // forkSession truncateAtIndex is inclusive (slices to index+1), truncateAt is message count to keep
              forkedId = forkSession(sid, { uid: session.uid, email: session.email, name: session.email.split('@')[0] }, msg.truncateAt - 1);
            }
            if (!forkedId) {
              // truncateAt=0 (restart from scratch) or forkSession failed — create a fresh session
              forkedId = crypto.randomUUID();
              upsertSession(forkedId, promptText, { uid: session.uid, email: session.email });
            }
            console.log(`[ws] Forked busy session ${sid.slice(0, 8)} → ${forkedId.slice(0, 8)} (truncateAt=${msg.truncateAt})`);
            session.busySessions.add(forkedId);
            sid = forkedId;
            session.lastSessionId = forkedId;
            send(ws, { type: 'forked', sourceSessionId: msg.sessionId, sessionId: forkedId });
          } else {
            const existing = loadConversation(sid);
            saveConversation(sid, existing.slice(0, msg.truncateAt));
            console.log(`[ws] Truncated conversation ${sid.slice(0, 8)} to ${msg.truncateAt} messages`);
          }
        }

        // Save user message to disk immediately
        const attachments: AttachmentMeta[] = msg.attachments ?? [];
        const attBlocks: ConvBlock[] = attachments.map((a: any) =>
          a.mimeType.startsWith('image/')
            ? { type: 'image' as const, src: a.url }
            : { type: 'file' as const, src: a.url, name: a.name, mimeType: a.mimeType }
        );
        if (!ephemeralUser) appendMessage(sid, { id: crypto.randomUUID(), role: 'user', blocks: [...attBlocks, { type: 'text', text: promptText }], channel: 'web', senderName: session.email.split('@')[0] });

        const wasReset = typeof msg.truncateAt === 'number' && msg.truncateAt >= 0;
        await runStream(ws, session, sid, promptText, attachments, mcpServers, false, wasReset, turnHints);
      } finally { admission.release(); }
    }

    if (msg.type === 'interruption_marker') {
      const sid = msg.sessionId || session.lastSessionId;
      if (sid && msg.revealedText) {
        const tail = msg.revealedText.length > 120
          ? '…' + msg.revealedText.slice(-120)
          : msg.revealedText;
        appendMessage(sid, {
          id: crypto.randomUUID(),
          role: 'system',
          blocks: [{ type: 'text', text: `[User interrupted playback — only heard up to: "${tail}"]` }],
        });
      }
    }

    if (msg.type === 'cancel') {
      const cancelSid = msg.sessionId || session.lastSessionId;
      if (!cancelSid) return;
      const localAc = session.abortControllers.get(cancelSid);
      const globalAc = getSessionAbortController(cancelSid);
      const ac = localAc || globalAc;
      if (ac) {
        console.log(`[ws] Cancel requested by ${session.email} session=${cancelSid.slice(0, 8)}`);
        ac.abort();
        session.abortControllers.delete(cancelSid);
      }
      session.busySessions.delete(cancelSid);
    }
  });

  ws.on('close', () => {
    clearInterval(pingInterval);
    console.log(`[ws] Disconnected: ${session.email || 'unauthenticated'}`);
    activeConnections.delete(ws);
    // Don't abort running processes — let them finish and save results.
    // Auto-approve regular permissions; queue destructive ones for reconnect.
    for (const [id, entry] of session.pendingPermissions) {
      if (entry.destructive && entry.tool && entry.sessionId) {
        globalPendingPermissions.set(id, { resolve: entry.resolve, sessionId: entry.sessionId, tool: entry.tool, input: entry.input, uid: session.uid });
        setTimeout(() => {
          if (globalPendingPermissions.delete(id)) {
            console.log(`[ws] Destructive ${entry.tool} id=${id} denied after TTL (client didn't reconnect)`);
            entry.resolve({ allow: false });
          }
        }, DESTRUCTIVE_PERMISSION_TTL);
        console.log(`[ws] Destructive permission ${id} (${entry.tool}) queued for reconnect (${DESTRUCTIVE_PERMISSION_TTL / 1000}s TTL)`);
      } else {
        console.log(`[ws] Auto-approving orphaned permission ${id} (client disconnected)`);
        entry.resolve({ allow: true });
      }
    }
    session.pendingPermissions.clear();
    // Orphaned questions: resolve null so the agent proceeds with its own judgement.
    for (const [id, entry] of session.pendingQuestions) {
      console.log(`[ws] Resolving orphaned question ${id} as null (client disconnected)`);
      entry.resolve(null);
    }
    session.pendingQuestions.clear();
    session.abortControllers.clear();
  });

  ws.on('error', (err) => console.error(`[ws] Error (${session.email}):`, err.message));
}

// ── Deploy recovery ──────────────────────────────────────────────────────────

async function retryWebSession(session: SessionMeta, prompt: string) {
  const sid = session.sessionId;
  console.log(`[recovery] retrying web session ${sid.slice(0, 8)}`);
  const recoveryAc = new AbortController();
  if (!acquireSessionLock(sid, 'web', recoveryAc)) {
    console.warn(`[recovery] session ${sid.slice(0, 8)} already locked, skipping`);
    return;
  }
  setRunStatus(sid, 'running', 'web');
  broadcast({ type: 'session_busy', sessionId: sid, busy: true });

  try {
    let assistantText = '';
    const assistantBlocks: ConvBlock[] = [];
    const collectPartial = () => [
      ...assistantBlocks,
      ...(assistantText ? [{ type: 'text' as const, text: assistantText }] : []),
    ];
    registerLivePartial(sid, collectPartial);
    for await (const ev of streamChat({
      // Recovery of an interrupted web turn: no live human, the session's owner marked internal.
      principal: fromInternal({ uid: session.uid, email: session.userEmail, lane: 'web-retry' }),
      prompt,
      sessionId: sid,
      uid: session.uid,
      userEmail: session.userEmail,
      userName: session.userName || session.userEmail.split('@')[0],
      mcpServers: getMcpConfig(session.uid),
      abortController: recoveryAc,
      context: { source: 'web', user: session.userEmail },
      onPermissionRequest: async () => ({ allow: true }),
    })) {
      if (ev.type === 'text_delta') {
        assistantText += ev.text;
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'text_delta', text: ev.text } });
      } else if (ev.type === 'tool_use') {
        if (assistantText) { assistantBlocks.push({ type: 'text', text: assistantText }); assistantText = ''; }
        assistantBlocks.push({ type: 'tool_use', tool: ev.tool, toolUseId: ev.toolUseId, input: ev.input });
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_use', tool: ev.tool, toolUseId: ev.toolUseId, input: ev.input } });
      } else if (ev.type === 'tool_use_input') {
        const existing = assistantBlocks.find((b: any) => b.type === 'tool_use' && b.toolUseId === ev.toolUseId) as any;
        if (existing) existing.input = ev.input;
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_use_input', toolUseId: ev.toolUseId, input: ev.input } });
      } else if (ev.type === 'tool_result') {
        assistantBlocks.push({ type: 'tool_result', toolUseId: ev.toolUseId, output: ev.output });
      } else if (ev.type === 'tool_result_image') {
        assistantBlocks.push({ type: 'image', src: ev.dataUrl });
        broadcast({ type: 'session_stream', sessionId: sid, event: { type: 'tool_result_image', toolUseId: ev.toolUseId, dataUrl: ev.dataUrl } });
      } else if (ev.type === 'done') {
        break;
      } else if (ev.type === 'error') {
        if (assistantText) { assistantBlocks.push({ type: 'text', text: assistantText }); assistantText = ''; }
        assistantBlocks.push({ type: 'error', text: ev.message });
        break;
      }
    }
    if (assistantText) assistantBlocks.push({ type: 'text', text: assistantText });
    if (assistantBlocks.length) {
      appendMessage(sid, { id: crypto.randomUUID(), role: 'assistant', blocks: assistantBlocks });
      broadcast({ type: 'session_messages_changed', sessionId: sid });
      const errBlock = assistantBlocks.find((b) => b.type === 'error');
      const preview = assistantText.slice(0, 120)
        || (errBlock?.type === 'error' ? `⚠️ ${errBlock.text}`.slice(0, 120) : '(completed)');
      notifyUnread(session.uid, sid, preview, 'response', session.title);
    }
  } catch (err: any) {
    console.error(`[recovery] web session error ${sid.slice(0, 8)}:`, err.message);
  } finally {
    unregisterLivePartial(sid);
    if (releaseSessionLock(sid, recoveryAc)) {
      setRunStatus(sid, 'idle');
      broadcast({ type: 'session_busy', sessionId: sid, busy: false });
    }
  }
}

async function recoverInterruptedSessions() {
  const interrupted = getRunningSessions();
  if (!interrupted.length) return;
  console.log(`[recovery] Found ${interrupted.length} interrupted session(s)`);

  for (const s of interrupted) {
    // Scheduler sessions: resume in-place on the SAME conversation (symmetric with web/slack),
    // instead of marking error and letting startup catch-up spawn a fresh convo — that re-fire
    // produced duplicate side-effects (e.g. a second Slack post).
    if (s.runOrigin === 'scheduler') {
      const partialBlocks = readPartial(s.sessionId);
      clearPartial(s.sessionId);
      const partialTexts = partialBlocks?.filter(b => b.type === 'text').map(b => (b as any).text) ?? [];

      // Bail to the old mark-error path if we can't resume safely:
      // no schedule id to re-run, or we've already retried this run twice.
      const canResume = !!s.scheduleId && (s.runRetryCount ?? 0) < 2;
      if (!canResume) {
        if (partialTexts.length) {
          appendMessage(s.sessionId, {
            id: crypto.randomUUID(),
            role: 'assistant',
            blocks: [{ type: 'text', text: partialTexts.join('\n') + '\n\n[...interrupted by server restart]' }],
          });
        }
        setRunStatus(s.sessionId, 'idle');
        updateScheduledSessionStatus(s.sessionId, 'error');
        if (s.scheduleId) scheduler.clearRunningMarker(s.scheduleId);
        console.log(`[recovery] scheduler session ${s.sessionId.slice(0, 8)} (${s.scheduleId ?? '?'}) — not resumable (no schedule / retries exhausted), marked error`);
        continue;
      }

      incrementRetryCount(s.sessionId);
      // Preserve the partial response with a cutoff marker, then continue in-place.
      const cutoff = partialTexts.length
        ? partialTexts.join('\n') + '\n\n[...response interrupted by server restart]'
        : '[...response interrupted by server restart]';
      appendMessage(s.sessionId, {
        id: crypto.randomUUID(),
        role: 'assistant',
        blocks: [{ type: 'text', text: cutoff }],
      });
      const resumePrompt = 'Your previous response was cut off by a server restart. The partial response has been preserved above. Continue from where you left off, and avoid repeating any side-effects (e.g. messages already sent) that may have completed before the interruption.';
      setRunStatus(s.sessionId, 'idle');
      const outcome = scheduler.resumeRun(s.scheduleId!, s.sessionId, resumePrompt);
      if (!outcome.ok) {
        // The refusal is now the EXPECTED outcome after a long outage (stale window / onMissed).
        // Close the session out exactly like the not-resumable branch above — otherwise
        // scheduleRunStatus stays 'running', the sidebar shows it busy forever, and it can never
        // self-heal because getRunningSessions() filters on runStatus, which is already idle.
        appendMessage(s.sessionId, {
          id: crypto.randomUUID(),
          role: 'assistant',
          blocks: [{ type: 'error', text: `Not resumed after the server restart — ${outcome.message} (${outcome.reason}).` }],
        });
        updateScheduledSessionStatus(s.sessionId, 'error');
        scheduler.clearRunningMarker(s.scheduleId!);
        console.log(`[recovery] scheduler session ${s.sessionId.slice(0, 8)} (${s.scheduleId}) — resume refused (${outcome.reason}): ${outcome.message}`);
        continue;
      }
      console.log(`[recovery] resuming scheduler session ${s.sessionId.slice(0, 8)} (${s.scheduleId}) in-place`);
      continue;
    }

    if (s.sessionId.startsWith('api-')) {
      console.log(`[recovery] Skip API session ${s.sessionId.slice(0, 12)} — no user waiting`);
      setRunStatus(s.sessionId, 'idle');
      continue;
    }

    if ((s.runRetryCount ?? 0) >= 2) {
      console.log(`[recovery] Skip ${s.sessionId.slice(0, 8)} — max retries reached`);
      const partialBlocks = readPartial(s.sessionId);
      clearPartial(s.sessionId);
      const partialTexts = partialBlocks?.filter(b => b.type === 'text').map(b => (b as any).text) ?? [];
      appendMessage(s.sessionId, {
        id: crypto.randomUUID(),
        role: 'assistant',
        blocks: [{ type: 'text', text: (partialTexts.length ? partialTexts.join('\n') + '\n\n' : '') + '[Interrupted by server restart — please send a message to continue]' }],
      });
      setRunStatus(s.sessionId, 'idle');
      continue;
    }

    const conv = loadConversation(s.sessionId);
    const lastUser = [...conv].reverse().find(m =>
      m.role === 'user' && m.blocks.some(b => b.type === 'text' && !b.text.startsWith('[Resumed'))
    );
    const originalPrompt = lastUser?.blocks.find(b => b.type === 'text')?.text;
    if (!originalPrompt) {
      setRunStatus(s.sessionId, 'idle');
      continue;
    }

    incrementRetryCount(s.sessionId);

    // Recover partial assistant response saved before crash
    const partialBlocks = readPartial(s.sessionId);
    clearPartial(s.sessionId);

    let prompt: string;
    if (partialBlocks?.length) {
      const partialTexts = partialBlocks.filter(b => b.type === 'text').map(b => (b as any).text);
      const cutoffText = partialTexts.length
        ? partialTexts.join('\n') + '\n\n[...response interrupted by server restart]'
        : '[...response interrupted by server restart]';
      appendMessage(s.sessionId, {
        id: crypto.randomUUID(),
        role: 'assistant',
        blocks: [{ type: 'text', text: cutoffText }],
      });
      prompt = 'Your previous response was cut off by a server restart. The partial response has been preserved above. Continue from where you left off.';
      console.log(`[recovery] Restored partial response (${partialBlocks.length} blocks) for ${s.sessionId.slice(0, 8)}`);
    } else {
      appendMessage(s.sessionId, {
        id: crypto.randomUUID(),
        role: 'user',
        blocks: [{ type: 'text', text: '[Resumed after server restart]' }],
      });
      prompt = originalPrompt;
    }

    if (s.runOrigin === 'slack') {
      if (!resumeFeatureSession('slack', s, prompt)) {
        console.warn(`[recovery] slack feature not available to resume ${s.sessionId.slice(0, 8)}`);
        setRunStatus(s.sessionId, 'idle');
      }
    } else {
      retryWebSession(s, prompt).catch(err => console.error(`[recovery] web retry failed:`, err.message));
    }
  }

  // Periodic sweep: clean up scheduler sessions not tracked by the engine
  const orphanSweep = setInterval(() => {
    const stillRunning = getRunningSessions().filter(s => s.runOrigin === 'scheduler');
    if (!stillRunning.length) { clearInterval(orphanSweep); return; }
    const engineRunning = new Set(scheduler.getRunningIds());
    for (const s of stillRunning) {
      if (s.scheduleId && engineRunning.has(s.scheduleId)) continue;
      console.log(`[recovery] orphaned scheduler session ${s.sessionId.slice(0, 30)}… — not in engine, marking idle/error`);
      setRunStatus(s.sessionId, 'idle');
      updateScheduledSessionStatus(s.sessionId, 'error');
    }
  }, 30_000);
}

let _draining = false;
const DRAIN_MS = Number(process.env.SHRAGA_DRAIN_MS) || 8_000;

async function gracefulShutdown(signal: string, opts: { exit?: boolean } = {}) {
  const exit = opts.exit ?? true;
  if (_draining) return;
  _draining = true;
  console.log(`[server] ${signal} — draining (up to ${Math.round(DRAIN_MS / 1000)}s)…`);
  setShuttingDown();
  stopSidecars();

  for (const ws of activeConnections.keys()) send(ws, { type: 'server_restarting' });

  // Agent turns routinely run for minutes; give in-flight streams time to finish before we force-kill
  // their Claude Code child processes (which would otherwise surface as `exited with code 143`).
  // Keep this under the service manager's stop timeout (e.g. systemd TimeoutStopSec) so the drain wins, not SIGKILL.
  // Drain on the LIVE lock count, not the persisted index: setRunStatus freezes `running` in the
  // index at shutdown (the crash-recovery marker), so index entries can never go idle mid-drain —
  // polling the index made every restart with an active session (or a stale marker from a prior
  // crash) sit out the full 90s.
  // Bounded by what the service manager ACTUALLY grants, not by what we'd like. launchd's default
  // ExitTimeOut is 20s and our power-guard wrapper SIGKILLs the child 10s after SIGTERM, so a 90s
  // drain was fiction: we never reached the clean exit below, and the in-flight turn was shredded
  // mid-write instead of aborted and persisted as a resumable partial. Override with
  // SHRAGA_DRAIN_MS where the manager really does grant longer.
  const deadline = Date.now() + DRAIN_MS;
  while (Date.now() < deadline) {
    const running = getActiveLockCount();
    if (running === 0) break;
    console.log(`[server] waiting for ${running} active stream(s)…`);
    await new Promise((r) => setTimeout(r, 2000));
  }

  console.log(`[server] drain complete — ${getActiveLockCount()} stream(s) still active, closing`);

  // Batched escalation digests live only in memory: hand them to the owners before exiting. Best-effort, bounded.
  let flushTimer: ReturnType<typeof setTimeout> | undefined;
  await Promise.race([
    flushEscalations().then((n) => { if (n) console.log(`[server] flushed ${n} pending escalation digest(s)`); }),
    new Promise<void>((r) => { flushTimer = setTimeout(() => { console.warn('[server] escalation flush timed out'); r(); }, 3000); }),
  ]).catch((e) => console.error('[server] escalation flush failed:', e?.message ?? e));
  clearTimeout(flushTimer);

  for (const [ws, session] of activeConnections) {
    for (const ac of session.abortControllers.values()) ac.abort();
    ws.close();
  }

  // Optional engines that hold OS resources (warm subprocesses, h2 connections) register their own
  // teardown on SIGTERM/SIGINT from the overlay — the core names none of them here.

  // Exit 0 on the fallback too: by this point we've drained and aborted cleanly, so a slow
  // server.close() is not a failure. Exiting 1 here made systemd log `status=1/FAILURE` on every
  // normal restart — a false alarm. (Restart=always brings us back regardless of code.)
  // Library embedders call stop() (exit:false): close the sockets but leave the host process alive.
  if (!exit) {
    wss.close();
    sidecarWss.close();
    await new Promise<void>((resolve) => server.close(() => resolve()));
    return;
  }
  server.close(() => process.exit(0));
  setTimeout(() => process.exit(0), 5000);
}

// Install process-level signal handlers for the standalone server/CLI path (default). A pure
// library embedder that owns its own lifecycle sets SHRAGA_INSTALL_SIGNALS=0 and uses handle.stop().
if (process.env.SHRAGA_INSTALL_SIGNALS !== '0') {
  process.on('SIGTERM', () => gracefulShutdown('SIGTERM').catch((e) => { console.error('[server] shutdown error:', e); process.exit(1); }));
  process.on('SIGINT', () => gracefulShutdown('SIGINT').catch((e) => { console.error('[server] shutdown error:', e); process.exit(1); }));
}

// ── Start ─────────────────────────────────────────────────────────────────────

// Kill orphaned vendor MCP processes from a previous server crash.
// Only targets processes reparented to init (ppid=1) — safe for multi-tenant.
try {
  const out = execSync(
    "pgrep -f 'vendor/mcp-.*--stdio' | xargs -I{} sh -c 'ppid=$(ps -o ppid= -p {} 2>/dev/null | tr -d \" \"); [ \"$ppid\" = \"1\" ] && echo {}' 2>/dev/null || true",
    { encoding: 'utf-8' },
  ).trim();
  if (out) {
    const pids = out.split('\n').filter(Boolean).map(Number);
    console.log(`[server] killing ${pids.length} orphaned vendor MCP process(es): ${pids.join(',')}`);
    for (const pid of pids) { try { process.kill(pid, 'SIGTERM'); } catch {} }
  }
} catch {}

const PORT = Number(process.env.PORT) || 3032;
await new Promise<void>((resolve) => {
  // A listen failure MUST kill the process. Without this the error reaches the `uncaughtException`
  // handler above, which by design keeps us alive — so the promise never settles and we sit there
  // forever: process running, port unbound, nothing served. The service manager sees a healthy job
  // and a port watchdog sees a dead one, so it kickstarts on a loop and shreds in-flight runs.
  // EADDRINUSE is the common case: `kickstart -k` starts the replacement while the old process is
  // still draining (up to 90s). Exiting used to be the answer, but the manager respawns on a ~10s
  // throttle and meets the same owner: MEASURED 2026-09-06, one watchdog kick became ~20 start/exit
  // cycles across five minutes, and every scheduled run in that window died with them. Wait the
  // drain out here instead — one process holding still, rather than N racing (see listen-retry.ts).
  //
  // EXCEPT when the holder is an ORPHANED copy of ourselves: nothing will ever signal it, so it
  // holds the port forever and waiting alone cannot fix it — reclaim it, then the next retry binds.
  let reclaimed = false;
  listenWithRetry(server, PORT, {
    onBusy: () => { if (!reclaimed) reclaimed = reclaimStalePort(PORT); },
    onWaiting: (ms) => console.warn(`[server] port ${PORT} is held (previous instance draining?) — waiting up to ${Math.round(ms / 1000)}s for it`),
  }).then(() => {
    console.log(`[server] Running on http://0.0.0.0:${PORT}`);
    resolve();
    if (PASSIVE) return; // no sidecars, recovery, or MCP warmers in passive mode
    // Deliver here, not earlier in boot: an upgrade's outcome is only knowable once we are the
    // process that came back, and the event has to land after the dispatcher is subscribed or the
    // owner never hears how it went.
    try { selfUpgrade.deliverPendingReport(); }
    catch (err) { console.warn('[self-upgrade] could not deliver report:', (err as Error).message); }
    startSidecars().catch(err => console.error('[sidecar] startup error:', err));
    recoverInterruptedSessions().catch(err => console.error('[recovery] failed:', err));
    bootDataSync();
    // The disk MCP catalog is warmed off the turn path by whichever engine consumes it — the CE default
    // (Claude Code) hands MCP servers straight to its SDK and needs no catalog. An add-on engine that
    // uses the shared catalog registers its own boot/interval warm-up through the overlay.
  }, (err: NodeJS.ErrnoException) => {
    const why = err.code === 'EADDRINUSE'
      ? `port ${PORT} is still in use after waiting for the previous instance to drain`
      : (err.message ?? String(err));
    console.error(`[server] FATAL: cannot listen on ${PORT} — ${why}. Exiting so the service manager restarts us.`);
    process.exit(1);
  });
});

// OPT-IN post-start plug-and-play. Only extensions/webhooks/events are runtime-safe (they mount on
// the persistent extRouter / in-process bus); features & engines mount at boot and are NOT re-entrant.
const RT_FLAG = process.env.SHRAGA_RUNTIME_REGISTRATION;
const RUNTIME_REG = RT_FLAG === '1' || RT_FLAG === 'true';
const runtimeGuard = (what: string) => {
  if (!RUNTIME_REG) throw new Error(
    `[shraga] ${what}() at runtime is disabled. Enable it with createShraga({ runtimeRegistration: true }) ` +
    `(or SHRAGA_RUNTIME_REGISTRATION=1) before start().`,
  );
};

return {
  app,
  server,
  port: PORT,
  url: `http://localhost:${PORT}`,
  emitEvent,
  registerExtension: (fn) => { runtimeGuard('registerExtension'); return registerExtension(fn); },
  // A webhook is just an extension that mounts a verified route on the extension Router — reuse the seam.
  registerWebhook: (opts) => { runtimeGuard('registerWebhook'); return registerExtension((_r, ctx) => { ctx.registerWebhook(opts); }); },
  on: (source, handler) => { runtimeGuard('on'); return subscribeEvent(source, handler); },
  stop: () => gracefulShutdown('stop', { exit: false }),
};

}