adom-courier

#!/usr/bin/env node
/*
 * adom-courier — app-neutral Google Chat <-> Adom-apps middleware (PULL model).
 *
 *   POST  (prompts, acks, outcomes)  -> adom-gchat webhook  => appears as "Marv · App" (bot-badged, no seat)
 *   RECEIVE (replies, identity, files) -> adom-google poll   => reads as you, invisible
 *
 * Apps never touch Google. They POST a request here and get structured response
 * events back (responder identity + attachment bytes) when a human replies.
 *
 * Correlation: adom-gchat (webhook) returns nothing and doesn't expose threadKey
 * on read, so after posting a prompt the courier FINDS its own just-posted Marv
 * message (the only BOT message with that exact text) to learn the thread.name,
 * then matches human replies to the request by thread.name.
 *
 * A space configured with "gchatSpace" posts via Marv; without it, posts fall
 * back to adom-google (posts as you) — handy for dev before a webhook exists.
 */
'use strict';

const http = require('http');
const fs = require('fs');
const path = require('path');
const { spawnSync } = require('child_process');
const { URL } = require('url');
const crypto = require('crypto');

const DIR = __dirname;
const PORT = parseInt(process.env.PORT || '8870', 10);
const HOST = process.env.HOST || '127.0.0.1';
const ADOM_GOOGLE = process.env.ADOM_GOOGLE || path.join(process.env.HOME || '/home/adom', '.local/bin/adom-google');
const ADOM_GCHAT = process.env.ADOM_GCHAT || 'adom-gchat';
// Inbound (human-initiated) replies post into a human-started thread via a RAW webhook
// POST (thread.name based) so Marv lands as sender=BOT. The webhook URL is read from this
// file — same one adom-gchat uses. spaces[gchatSpace].url. (See human-inbound-plan §1 spike.)
const GCHAT_WEBHOOKS = process.env.GCHAT_WEBHOOKS || path.join(process.env.HOME || '/home/adom', '.config/gchat-webhooks.json');
// Process start time — the cold-start gate for NEW inbound conversations (branch C only).
// NEVER used to mutate lastSeen (that would drop app-initiated replies). See R5.
const PROCESS_START = Date.now();
// Attachment caps (R3): per-file byte cap + cumulative ATT_DIR cap. Enforced PRE-download (size
// metadata when present) AND mid-download (MEDIA_MAX_BUFFER aborts the spawnSync child the moment
// it exceeds the limit, so an oversized file never buffers the full 96MB into memory / blocks the
// loop — R3/R10). MEDIA_MAX_BUFFER sits a little above MAX_ATT_BYTES so an exactly-at-cap file still
// downloads, but a much larger one trips ENOBUFS fast.
const MAX_ATT_BYTES = parseInt(process.env.ADOM_COURIER_MAX_ATT_BYTES || String(10 * 1024 * 1024), 10);
const MAX_ATT_DIR_BYTES = parseInt(process.env.ADOM_COURIER_MAX_ATT_DIR_BYTES || String(500 * 1024 * 1024), 10);
const MEDIA_MAX_BUFFER = parseInt(process.env.ADOM_COURIER_MEDIA_MAX_BUFFER || String(MAX_ATT_BYTES + 1024 * 1024), 10);
// Inbound text cap (R9) + bounds on the conversations map + processed-message dedup set.
const MAX_INBOUND_TEXT = parseInt(process.env.ADOM_COURIER_MAX_INBOUND_TEXT || String(16 * 1024), 10);
const MAX_CONVERSATIONS = parseInt(process.env.ADOM_COURIER_MAX_CONVERSATIONS || '500', 10);
const CONV_IDLE_MS = parseInt(process.env.ADOM_COURIER_CONV_IDLE_MS || String(24 * 3600 * 1000), 10);
const PROCESSED_MAX = parseInt(process.env.ADOM_COURIER_PROCESSED_MAX || '5000', 10);
const MAX_POLL_PAGES = parseInt(process.env.ADOM_COURIER_MAX_POLL_PAGES || '10', 10);   // R7: page back up to this many 30-msg pages/poll
const IDENTITY_NEG_TTL_MS = parseInt(process.env.ADOM_COURIER_IDENTITY_NEG_TTL_MS || String(5 * 60 * 1000), 10);   // negative-resolve cache TTL (R10)
const STALE_THRESHOLD = parseInt(process.env.ADOM_COURIER_STALE_THRESHOLD || '3', 10);   // consecutive poll errors before a space is "stale"
const MAX_IDENTITY_NEG = parseInt(process.env.ADOM_COURIER_MAX_IDENTITY_NEG || '2000', 10);   // cap on the negative-identity cache
const DISPATCH_TIMEOUT_MS = parseInt(process.env.ADOM_COURIER_DISPATCH_TIMEOUT_MS || '20000', 10);
// External data dir — config/state/attachments live OUTSIDE the package dir so
// `adom-wiki pkg update` (which re-extracts the package) can't clobber them.
const DATA_DIR = process.env.ADOM_COURIER_HOME || path.join(process.env.HOME || '/home/adom', '.config/adom-courier');
fs.mkdirSync(DATA_DIR, { recursive: true });
const STATE_FILE = path.join(DATA_DIR, 'state.json');
const ATT_DIR = path.join(DATA_DIR, 'attachments');
// Deployment config: external if present (real spaces/apps), else the bundled template.
const EXT_CONFIG = path.join(DATA_DIR, 'config.json');
const CONFIG_PATH = process.env.ADOM_COURIER_CONFIG || (fs.existsSync(EXT_CONFIG) ? EXT_CONFIG : path.join(DIR, 'config.json'));
const CONFIG = JSON.parse(fs.readFileSync(CONFIG_PATH, 'utf8'));
fs.mkdirSync(ATT_DIR, { recursive: true });

// Normalize spaces: alias -> { id, gchatSpace?, inboundApp?, inboundAllowlist? }.
// Accepts a bare id string too. `alias` is stored so logs/boot-warnings can name the space.
const SPACES = {};
for (const [alias, v] of Object.entries(CONFIG.spaces || {})) SPACES[alias] = (typeof v === 'string') ? { id: v, alias } : { ...v, alias };
function resolveSpace(aliasOrId) {
  if (SPACES[aliasOrId]) return SPACES[aliasOrId];
  for (const s of Object.values(SPACES)) if (s.id === aliasOrId) return s;
  return /^spaces\//.test(aliasOrId) ? { id: aliasOrId } : null;
}

function log(...a) { console.log(new Date().toISOString(), ...a); }

// Per-key async mutex: serializes the gchatSend+findOwnMessage critical section per space so two
// concurrent /requests posting IDENTICAL text can't interleave and bind to each other's thread
// (findOwnMessage matches by exact text). Keyed by space id; a global key would also be correct.
// The tail Promise resolves when the current holder finishes; the next waiter chains off it.
const _locks = new Map();
function withLock(key, fn) {
  const prev = _locks.get(key) || Promise.resolve();
  let release;
  const tail = prev.then(() => new Promise((r) => { release = r; }));
  _locks.set(key, tail);
  return prev.then(() => fn()).finally(() => {
    release();
    if (_locks.get(key) === tail) _locks.delete(key);   // GC the chain when no one else is waiting
  });
}

// I/O boundary indirection — production uses the real functions (assigned after they're defined).
// Unit tests overwrite these to feed canned messages / capture dispatches, exercising the REAL
// poll loop + handlers. Untouched in production (require.main === module).
const hooks = { listMessages: null, resolveUser: null, downloadAttachment: null, dispatch: null };

// Inbound routing validation (BLOCKING req 1 / R2). A space that declares inboundApp MUST:
//   - have a gchatSpace (else replies would post as HUMAN via adom-google → self-dispatch loop), and
//   - name an app that exists in CONFIG.apps.
// If either fails we DISABLE inbound on that space (clear inboundApp) and boot-warn loudly,
// so a misconfigured space silently behaves like today (ignore human-initiated) rather than loop.
function validateInboundSpaces() {
  for (const c of Object.values(SPACES)) {
    if (!c.inboundApp) continue;
    if (!c.gchatSpace) {
      log(`⚠ INBOUND DISABLED on space '${c.alias}' (${c.id}): inboundApp='${c.inboundApp}' but no gchatSpace — refusing to route inbound (would reply as HUMAN → self-dispatch loop). Set gchatSpace to enable.`);
      c.inboundApp = null; continue;
    }
    if (!(CONFIG.apps || {})[c.inboundApp]) {
      log(`⚠ INBOUND DISABLED on space '${c.alias}' (${c.id}): inboundApp='${c.inboundApp}' not found in apps{}.`);
      c.inboundApp = null; continue;
    }
  }
}

// Read the raw webhook URL for a gchatSpace from ~/.config/gchat-webhooks.json → spaces[gchatSpace].url.
// Cached, but INVALIDATED on file mtime change so a rotated webhook is picked up without a restart
// (F3) — without paying a JSON parse on every send. Throws if the file/space is missing.
let _webhookCache = null; let _webhookMtime = 0;
function webhookUrlFor(gchatSpace) {
  let mtime = 0; try { mtime = fs.statSync(GCHAT_WEBHOOKS).mtimeMs; } catch (_) {}
  if (!_webhookCache || mtime !== _webhookMtime) {
    const raw = fs.readFileSync(GCHAT_WEBHOOKS, 'utf8');
    _webhookCache = JSON.parse(raw).spaces || {};
    _webhookMtime = mtime;
  }
  const e = _webhookCache[gchatSpace];
  if (!e || !e.url) throw new Error(`no webhook url for gchatSpace '${gchatSpace}' in ${GCHAT_WEBHOOKS}`);
  return e.url;
}

// ---------------------------------------------------------------- state
// conversations{threadName:conv} is the human-initiated counterpart to requests{} (app-initiated).
// processed[] is a bounded, ORDERED list of already-handled message.name values — the CODE loop
// guard (R2/R8): a message is never processed twice, which also closes the poll↔/reply race.
// backfill{spaceId:{token,newest}} holds an in-progress deep-pagination cursor when a single poll
// couldn't reach the watermark (>MAX_POLL_PAGES×30 messages); the next poll resumes from `token` and
// advances lastSeen to `newest` only once the backlog is fully drained (F2).
// spaceHealth{spaceId:{consecutiveErrors,lastErrorAt,lastError,lastOkAt,alerted}} tracks per-space
// poll health so a space whose poll keeps erroring isn't silently frozen (Phase-2 staleness alarm).
function freshState() { return { counter: 0, lastSeen: {}, requests: {}, responses: [], identity: {}, conversations: {}, processed: [], backfill: {}, identityNeg: {}, spaceHealth: {} }; }
let state = freshState();
try { if (fs.existsSync(STATE_FILE)) state = { ...freshState(), ...JSON.parse(fs.readFileSync(STATE_FILE, 'utf8')) }; } catch (_) {}
// Rehydrate the processed-dedup set from the persisted ordered list.
const processedSet = new Set(state.processed || []);
function markProcessed(name) {
  if (!name || processedSet.has(name)) return;
  processedSet.add(name); state.processed.push(name);
  while (state.processed.length > PROCESSED_MAX) processedSet.delete(state.processed.shift());
}
function saveState() { try { fs.writeFileSync(STATE_FILE, JSON.stringify(state, null, 2)); } catch (e) { log('saveState err', e.message); } }

// ---------------------------------------------------------------- Google read (adom-google)
function gapi(url, { method = 'GET', data = null, query = [], raw = false, maxBuffer = null } = {}) {
  const args = ['api', url, '-X', method];
  for (const q of query) args.push('-q', q);
  if (data != null) args.push('-d', typeof data === 'string' ? data : JSON.stringify(data));
  if (raw) args.push('--raw');
  // maxBuffer caps how many bytes spawnSync will buffer before it ABORTS the child (ENOBUFS).
  // For media (raw) downloads we cap it just above the per-file attachment limit so an oversized
  // file fails FAST instead of buffering up to 96MB into memory + blocking the event loop (R3/R10).
  const mb = maxBuffer != null ? maxBuffer : (raw ? MEDIA_MAX_BUFFER : 96 * 1024 * 1024);
  const r = spawnSync(process.env.ADOM_GOOGLE || ADOM_GOOGLE, args, { encoding: raw ? 'buffer' : 'utf8', maxBuffer: mb, timeout: 45000, env: process.env });
  if (r.error) throw new Error('adom-google spawn failed: ' + r.error.message);
  if (raw) {
    const err = (r.stderr || Buffer.alloc(0)).toString(); const m = err.match(/HTTP\s+(\d+)/);
    return { status: m ? +m[1] : (r.status === 0 ? 200 : 0), buffer: r.stdout || Buffer.alloc(0) };
  }
  const stderr = (r.stderr || '').toString();
  const out = (r.stdout || '').toString(); const i = out.search(/[[{]/);
  let json = null; if (i >= 0) { try { json = JSON.parse(out.slice(i)); } catch (_) {} }
  // Surface API/HTTP errors instead of silently returning a null/empty body. The CLI exits non-zero
  // and/or the JSON carries an `error` object on 4xx/5xx; either is an ERROR, NOT an empty result.
  // Callers (e.g. listMessages pagination) must not conflate "request failed" with "no messages".
  const httpStatus = (stderr.match(/HTTP\s+(\d+)/) || [])[1];
  const apiError = (json && json.error) ? json.error
    : (r.status !== 0 || (httpStatus && +httpStatus >= 400)) ? { code: httpStatus ? +httpStatus : r.status, message: (stderr.trim().split('\n').pop() || `adom-google exit ${r.status}`) }
    : null;
  return { json, raw: out, apiError };
}
function resolveUser(userName, forceResolve) {
  if (!userName) return { name: 'unknown', email: '' };
  if (state.identity[userName]) return state.identity[userName];      // permanent positive cache (success only)
  const fallback = { name: userName, email: '', userId: userName };
  // Negative cache (R10): if a recent resolve FAILED, don't re-spawn a blocking People-API call on
  // every message — return the email-less fallback until the short TTL elapses, then retry. This
  // keeps the F1 fix (a transient blip retries soon, ~5min) while a permanently-unresolvable chatty
  // sender (valid id, no Google profile) doesn't stall the event loop on every message.
  // EXCEPTION (item 3): on an allowlist space the caller passes forceResolve=true so a just-provisioned
  // allowlisted user isn't fail-closed for up to the TTL — always attempt a fresh resolve there.
  const neg = state.identityNeg[userName];
  if (!forceResolve && neg && Date.now() < neg) return fallback;
  let out = fallback;
  let resolved = false;   // did People actually return identity data this call?
  try {
    const { json } = gapi(`https://people.googleapis.com/v1/people/${num(userName)}`, { query: ['personFields=names,emailAddresses'] });
    if (json && !json.error) {
      out = { name: (json.names || [{}])[0].displayName || userName, email: ((json.emailAddresses || []).find((e) => (e.value || '').includes('@')) || {}).value || '', userId: userName };
      resolved = true;
    }
  } catch (e) { log('resolveUser err', e.message); }
  if (resolved) {
    state.identity[userName] = out;                                   // cache success permanently
    delete state.identityNeg[userName];                              // clear any negative entry
  } else {
    state.identityNeg[userName] = Date.now() + IDENTITY_NEG_TTL_MS;   // back off re-resolving for a short TTL
    capIdentityNeg();                                                 // keep the negative cache bounded (item 2)
  }
  return out;
}
function num(userName) { return userName.replace(/^users\//, ''); }
// List messages newest-first, PAGINATING toward the watermark `sinceTs` (so a burst of >30 messages
// between polls is never silently dropped — R7/F2). Bounded by MAX_POLL_PAGES per call so one poll
// can't spin forever. When `resumeToken` is given, pagination CONTINUES from that deeper cursor
// (a backfill in progress from a previous poll) instead of restarting at the newest message — that's
// what lets a backlog LARGER than the per-poll page budget actually drain across successive polls.
// Without `sinceTs`, returns just the first page (back-compat for findOwnMessage). `isSeen(name)` is
// an optional predicate (the processed-dedup check): a page whose messages are ALL already-seen is
// "free" — it doesn't count against MAX_POLL_PAGES — so a restart-from-newest after an expired cursor
// can blow past the already-drained prefix to reach the remaining gap (without re-dispatching).
// Returns an array (newest/where-resumed first) carrying flags the poll loop reconciles:
//   .truncated     = hit the page bound (of PRODUCTIVE pages) before reaching the watermark (gap remains)
//   .nextPageToken = cursor to resume from next poll (when truncated)
//   .error         = a page FETCH failed (HTTP/API error, NOT an empty page) — poll must NOT treat
//                    this as "drained"; it must hold lastSeen + keep the cursor and retry. (R-#1/#2)
//   .expiredToken  = the failure was on a RESUME page (likely an expired/invalid cursor) — poll
//                    should drop the dead cursor and RESTART the drain from the watermark.
function listMessages(spaceId, sinceTs, resumeToken, isSeen) {
  const all = []; let pageToken = resumeToken || null; let truncated = false; let lastToken = null;
  let fetchError = null; let expiredToken = false; let productive = 0;   // pages that contained NEW (unseen) messages
  // Total page ceiling: MAX_POLL_PAGES PRODUCTIVE pages, plus headroom to skip an already-processed
  // prefix (all-seen = "free") on a restart. Caps blocking spawnSync calls so a pathological space
  // can't stall one poll indefinitely while still letting a restart blow past the drained prefix.
  const PAGE_CEILING = MAX_POLL_PAGES * 6;
  let done = false;                                                   // did we cleanly reach watermark / end of pages?
  for (let guard = 0; guard < PAGE_CEILING; guard++) {
    const query = ['pageSize=30', 'orderBy=createTime desc'];
    if (pageToken) query.push(`pageToken=${encodeURIComponent(pageToken)}`);
    const usedResume = !!pageToken && pageToken === resumeToken;     // are we hitting the resumed deep cursor?
    const { json, apiError } = gapi(`https://chat.googleapis.com/v1/${spaceId}/messages`, { query });
    if (apiError) {                                                  // page FAILED — stop, surface it (do NOT drain)
      fetchError = apiError;
      if (usedResume && (apiError.code === 400 || apiError.code === 404)) expiredToken = true;   // expired/invalid cursor
      log(`poll ${spaceId}: page fetch error (${apiError.code || '?'}: ${apiError.message || ''})${usedResume ? ' [resume cursor]' : ''} — NOT draining; will retry`);
      break;
    }
    const msgs = (json && json.messages) || [];
    all.push(...msgs);
    pageToken = json && json.nextPageToken; lastToken = pageToken;
    // A page counts against the budget only if it carried at least one UNSEEN message; an all-seen
    // page (the already-drained prefix on a restart) is free so we can reach the gap below it.
    const hasNew = !isSeen || msgs.some((m) => !isSeen(m.name));
    if (hasNew) productive += 1;
    if (sinceTs == null) { done = true; break; }                      // no watermark -> single page (back-compat)
    if (!pageToken || !msgs.length) { done = true; break; }          // no more pages -> fully drained
    const oldest = msgs[msgs.length - 1];                            // desc order -> last is oldest on this page
    if (oldest && new Date(oldest.createTime).getTime() <= sinceTs) { done = true; break; }   // reached watermark -> done
    if (productive >= MAX_POLL_PAGES) {                              // budget of PRODUCTIVE pages hit -> gap remains
      truncated = true;
      log(`poll ${spaceId}: hit MAX_POLL_PAGES (${MAX_POLL_PAGES} productive×30) before reaching watermark — older messages DEFERRED to next poll (resume cursor saved; lastSeen held)`);
      break;
    }
  }
  // Exited via the page ceiling (huge all-seen prefix) without a clean finish AND more pages exist
  // -> treat as truncated so lastSeen is held and we resume next poll (never silently "drained").
  if (!done && !fetchError && !truncated && pageToken) truncated = true;
  all.truncated = truncated;
  all.nextPageToken = truncated ? lastToken : null;
  all.error = fetchError;
  all.expiredToken = expiredToken;
  return all;
}
function downloadAttachment(att) {
  const rn = att.attachmentDataRef && att.attachmentDataRef.resourceName;
  if (!rn) throw new Error('no attachmentDataRef (Drive files unsupported)');
  // PRE-download cap (cheap): if the message carries a size hint and it already exceeds the cap,
  // skip without spawning a download at all.
  const hinted = Number(att.size || att.contentLength || (att.attachmentDataRef && att.attachmentDataRef.size) || 0);
  if (hinted && hinted > MAX_ATT_BYTES) throw new Error(`OVERSIZE: ${hinted} bytes > per-file cap ${MAX_ATT_BYTES} (skipped pre-download)`);
  let res;
  try {
    res = gapi(`https://chat.googleapis.com/v1/media/${encodeURIComponent(rn)}?alt=media`, { raw: true });
  } catch (e) {
    // spawnSync aborts with ENOBUFS the instant the response exceeds MEDIA_MAX_BUFFER -> oversize.
    if (/ENOBUFS|maxBuffer/i.test(e.message)) throw new Error(`OVERSIZE: exceeds media buffer cap ${MEDIA_MAX_BUFFER} bytes (download aborted, not buffered to memory)`);
    throw e;
  }
  const { status, buffer } = res;
  if (status !== 200 || !buffer.length) throw new Error(`media.download status ${status}`);
  return buffer;
}

// ---------------------------------------------------------------- Chat posting
// Marv (adom-gchat webhook). Returns nothing useful — correlation is by findOwnMessage.
function gchatSend(gchatSpace, text, threadKey) {
  const args = ['send', '--space', gchatSpace, '--no-attribution', '--permission-to-post-given'];
  if (threadKey) args.push('--thread', threadKey);
  args.push(text);
  const r = spawnSync(process.env.ADOM_GCHAT || ADOM_GCHAT, args, { encoding: 'utf8', timeout: 25000, env: process.env });
  if (r.status !== 0) throw new Error('adom-gchat send failed: ' + ((r.stderr || r.stdout || '').trim().split('\n').pop() || `exit ${r.status}`));
}
// Marv reply INTO a human-started thread (inbound). A human thread has thread.name but NO
// threadKey, so adom-gchat's --thread (threadKey-only) would FORK a new thread. Instead we POST
// raw to the webhook with thread:{name} + REPLY_MESSAGE_OR_FAIL so Marv lands as sender=BOT in the
// SAME thread. (SPIKE-confirmed; see plan §1.) Uses the https module only — zero npm deps.
function gchatReplyByThreadName(gchatSpace, text, threadName) {
  const base = webhookUrlFor(gchatSpace);
  const url = base + (base.includes('?') ? '&' : '?') + 'messageReplyOption=REPLY_MESSAGE_OR_FAIL';
  const body = Buffer.from(JSON.stringify({ text, thread: { name: threadName } }));
  const u = new URL(url); const lib = u.protocol === 'https:' ? require('https') : http;
  return new Promise((resolve, reject) => {
    const r = lib.request({ hostname: u.hostname, port: u.port, path: u.pathname + u.search, method: 'POST',
      headers: { 'Content-Type': 'application/json; charset=UTF-8', 'Content-Length': body.length } }, (res) => {
      let out = ''; res.on('data', (c) => { out += c; }); res.on('end', () => {
        if (res.statusCode >= 200 && res.statusCode < 300) return resolve(out);
        reject(new Error(`webhook reply status ${res.statusCode}: ${out.slice(0, 300)}`));
      });
    });
    r.on('error', reject); r.setTimeout(20000, () => r.destroy(new Error('webhook reply timeout')));
    r.write(body); r.end();
  });
}
// adom-google direct post (fallback when a space has no gchatSpace). Returns the message.
function adomGooglePost(spaceId, text, threadKey, threadName) {
  const thread = threadName ? { name: threadName } : { threadKey };
  const replyOpt = threadName ? 'REPLY_MESSAGE_OR_FAIL' : 'REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD';
  const { json } = gapi(`https://chat.googleapis.com/v1/${spaceId}/messages`, { method: 'POST', data: { text, thread }, query: [`messageReplyOption=${replyOpt}`] });
  if (!json || json.error || !json.name) throw new Error(`adom-google post failed: ${json && json.error ? JSON.stringify(json.error) : 'no message'}`);
  return json;
}
// After a Marv post, locate our own message (the BOT message with this exact text,
// in a thread not already tracked) to learn its thread.name. Retries for propagation.
function findOwnMessage(spaceId, text) {
  const want = (text || '').trim();
  for (let i = 0; i < 6; i++) {
    let msgs = []; try { msgs = listMessages(spaceId); } catch (_) {}
    const m = msgs.find((x) => x.sender && x.sender.type === 'BOT' && (x.text || '').trim() === want
      && x.thread && x.thread.name && !state.requests[x.thread.name] && !state.conversations[x.thread.name]);
    if (m) return { threadName: m.thread.name, messageName: m.name };
    spawnSync('sleep', ['0.9']);
  }
  return null;
}

// ---------------------------------------------------------------- dispatch to owning app
function dispatch(appName, event) {
  const app = (CONFIG.apps || {})[appName];
  // Delivery visibility (item 4): mutate the SHARED event ref (also stored in state.responses) so
  // delivery status is queryable at /state. No retry — best-effort, just observable.
  const tag = event.itemKey || event.conversationId || '?';   // app-initiated has itemKey; inbound falls back to conversationId
  if (!app || !app.dispatchUrl) { event.delivered = false; event.held = 'no dispatchUrl'; log(`dispatch: no URL for '${appName}' — held in /state`); return; }
  const body = Buffer.from(JSON.stringify({ ...event, attachments: event.attachments.map((a) => ({ ...a, base64: a.path ? fs.readFileSync(a.path).toString('base64') : undefined })) }));
  const u = new URL(app.dispatchUrl); const lib = u.protocol === 'https:' ? require('https') : http;
  const headers = { 'Content-Type': 'application/json', 'Content-Length': body.length };
  if (app.dispatchToken) headers.Authorization = `Bearer ${app.dispatchToken}`;   // courier proves identity to the app's receiver
  const fail = (msg) => { if (event.delivered === undefined) { event.delivered = false; event.deliveryError = msg; } };   // record once
  const r = lib.request({ hostname: u.hostname, port: u.port, path: u.pathname, method: 'POST', headers },
    (res) => { res.on('data', () => {}); res.on('end', () => {
      if (res.statusCode >= 200 && res.statusCode < 300) { event.delivered = true; delete event.deliveryError; }
      else fail(`HTTP ${res.statusCode}`);
      log(`dispatched ${appName}/${tag}: ${res.statusCode}`);
    }); });
  r.on('error', (e) => { fail(e.message); log(`dispatch error ${appName}/${tag}: ${e.message}`); });
  r.setTimeout(DISPATCH_TIMEOUT_MS, () => r.destroy(new Error('dispatch timeout')));   // R6: no socket leak on a hung receiver
  r.write(body); r.end();
}

// Wire the I/O-boundary indirection to the real implementations (tests overwrite these).
hooks.listMessages = listMessages;
hooks.resolveUser = resolveUser;
hooks.downloadAttachment = downloadAttachment;
hooks.dispatch = dispatch;

// ---------------------------------------------------------------- reply into a request's thread (Marv or adom-google)
function replyInThread(rec, text) {
  if (rec.gchatSpace) { gchatSend(rec.gchatSpace, text, rec.threadKey); return; }
  adomGooglePost(rec.space, text, rec.threadKey, rec.threadName);
}

// ---------------------------------------------------------------- attachments (shared by app-initiated + inbound)
// Current ATT_DIR size, computed lazily and tracked so we can enforce the cumulative cap (R3)
// without statting the whole dir on every file.
let _attDirBytes = null;
function attDirBytes() {
  if (_attDirBytes != null) return _attDirBytes;
  let t = 0; try { for (const f of fs.readdirSync(ATT_DIR)) { try { t += fs.statSync(path.join(ATT_DIR, f)).size; } catch (_) {} } } catch (_) {}
  _attDirBytes = t; return t;
}
// Download a message's attachments to ATT_DIR, enforcing per-file (MAX_ATT_BYTES) and cumulative
// (MAX_ATT_DIR_BYTES) caps BEFORE writing. Skipped/oversized files are recorded with an `error`,
// not written. `prefix` namespaces files by request/conversation id.
function collectAttachments(m, prefix) {
  const out = [];
  for (const a of (m.attachment || [])) {
    try {
      const buf = hooks.downloadAttachment(a);
      if (buf.length > MAX_ATT_BYTES) { log(`att skip ${a.contentName}: ${buf.length}B > per-file cap ${MAX_ATT_BYTES}B`); out.push({ name: a.contentName, contentType: a.contentType, size: buf.length, error: `exceeds per-file cap (${MAX_ATT_BYTES} bytes)` }); continue; }
      if (attDirBytes() + buf.length > MAX_ATT_DIR_BYTES) { log(`att skip ${a.contentName}: ATT_DIR cap ${MAX_ATT_DIR_BYTES}B would be exceeded`); out.push({ name: a.contentName, contentType: a.contentType, size: buf.length, error: 'attachment store full (ATT_DIR cap)' }); continue; }
      const safe = (a.contentName || 'file').replace(/[^\w.-]/g, '_');
      const p = path.join(ATT_DIR, `${prefix}-${Date.now()}-${safe}`);
      fs.writeFileSync(p, buf); _attDirBytes = attDirBytes() + buf.length;
      out.push({ name: a.contentName, contentType: a.contentType, path: p, size: buf.length });
    } catch (e) { out.push({ name: a.contentName, error: e.message }); }
  }
  return out;
}

// ---------------------------------------------------------------- poller
function handleReply(req, m, spaceId) {
  const responder = hooks.resolveUser(m.sender && m.sender.name);
  const attachments = collectAttachments(m, req.requestId);
  const event = { type: 'response', requestId: req.requestId, app: req.app, itemKey: req.itemKey, turn: req.turn || 1, responder, text: m.text || '', attachments, space: spaceId, threadName: req.threadName, messageName: m.name, at: new Date().toISOString() };
  state.responses.unshift(event); if (state.responses.length > 200) state.responses.pop();
  req.status = 'answered'; req.answeredAt = event.at; req.responder = responder;
  log(`✓ ${req.app}/${req.itemKey} from ${responder.name} <${responder.email}> — ${attachments.length} attachment(s)`);
  if (CONFIG.ackReceipt !== false) {
    try { const first = (responder.name || 'there').split(' ')[0]; const what = attachments.length ? (attachments.length > 1 ? `${attachments.length} photos` : 'photo') : 'reply'; replyInThread(req, `✅ Got it, ${first} — received your ${what}. Thanks!`); req.acked = true; }
    catch (e) { log('ack failed', e.message); }
  }
  hooks.dispatch(req.app, event);
}

// allowlist gate (R4): when inboundAllowlist is configured on the space, fail-closed — drop
// messages from senders whose resolved email isn't listed (no ack, no dispatch, no conversation
// churn). Returns {ok} plus a `reason` so the caller can log WHY a drop happened — distinguishing
// "identity unresolved" (People API failed -> no email) from "not on allowlist" (R8). Both still
// fail closed; the reason is purely diagnostic so an operator can tell a misconfig from an attack.
function inboundSenderAllowed(cfg, responder) {
  const list = cfg.inboundAllowlist;
  if (!Array.isArray(list) || !list.length) return { ok: true };     // not configured -> open (space membership is the boundary)
  const email = (responder.email || '').toLowerCase();
  if (!email) return { ok: false, reason: 'identity unresolved (no email — People API failed?)' };
  if (list.some((e) => String(e).toLowerCase() === email)) return { ok: true };
  return { ok: false, reason: 'sender not on allowlist' };
}

// handleInbound: a human-initiated message in an inbound space. Resolve identity, allowlist-gate,
// collect attachments, dispatch a type:'inbound' event to the owning app. NO auto-ack — the app's
// /reply IS the ack. Returns true if dispatched, false if dropped (allowlist).
function handleInbound(conv, m, spaceId, cfg) {
  // On an allowlist space, force a fresh resolve (item 3): bypass the negative cache so a sender who
  // just gained a Google profile within the TTL isn't suppressed (fail-closed) for up to 5 minutes.
  const forceResolve = Array.isArray(cfg.inboundAllowlist) && cfg.inboundAllowlist.length > 0;
  const responder = hooks.resolveUser(m.sender && m.sender.name, forceResolve);
  const gate = inboundSenderAllowed(cfg, responder);
  if (!gate.ok) {
    log(`✗ inbound DROPPED (${gate.reason}) ${conv.app} from ${responder.name} <${responder.email || '?'}> in ${spaceId}`);
    return false;
  }
  const attachments = collectAttachments(m, 'conv-' + conv.conversationId.replace(/[^\w]/g, '_'));
  let text = m.text || '';
  if (text.length > MAX_INBOUND_TEXT) text = text.slice(0, MAX_INBOUND_TEXT);   // R9: cap attacker-controlled text length
  const event = { type: 'inbound', conversationId: conv.conversationId, app: conv.app, space: spaceId,
    turn: conv.turn, responder, text, attachments, trusted: false, threadName: conv.threadName, messageName: m.name, at: new Date().toISOString() };
  state.responses.unshift(event); if (state.responses.length > 200) state.responses.pop();
  conv.turn += 1; conv.lastHumanAt = event.at; conv.responder = responder;
  log(`→ inbound ${conv.app} from ${responder.name} <${responder.email}> — turn ${event.turn}, ${attachments.length} attachment(s)`);
  hooks.dispatch(conv.app, event);
  return true;
}

// Bound the conversations map (R3/caps): drop closed + idle conversations, then cap total count
// (oldest-by-lastHumanAt first). Called after each poll.
// Bound the negative-identity cache (item 2): expire entries past TTL, then if still over the cap,
// evict the oldest (lowest expiry == earliest inserted, since the TTL is fixed). Called from the
// poll prune AND right after each insert, so a burst between polls can't blow the cap unbounded.
function capIdentityNeg() {
  const now = Date.now();
  for (const [u, exp] of Object.entries(state.identityNeg)) if (exp <= now) delete state.identityNeg[u];
  const keys = Object.keys(state.identityNeg);
  if (keys.length > MAX_IDENTITY_NEG) {
    keys.sort((a, b) => state.identityNeg[a] - state.identityNeg[b]);   // oldest expiry first
    for (const u of keys.slice(0, keys.length - MAX_IDENTITY_NEG)) delete state.identityNeg[u];
  }
}
function pruneConversations() {
  const now = Date.now();
  capIdentityNeg();
  const entries = Object.entries(state.conversations);
  for (const [tn, c] of entries) {
    const last = new Date(c.lastHumanAt || c.startedAt || 0).getTime();
    if (c.status === 'closed' && now - last > CONV_IDLE_MS) delete state.conversations[tn];
    else if (now - last > CONV_IDLE_MS) { c.status = 'closed'; delete state.conversations[tn]; }   // idle-close + drop
  }
  let remaining = Object.entries(state.conversations);
  if (remaining.length > MAX_CONVERSATIONS) {
    remaining.sort((a, b) => new Date(a[1].lastHumanAt || a[1].startedAt || 0) - new Date(b[1].lastHumanAt || b[1].startedAt || 0));
    for (const [tn] of remaining.slice(0, remaining.length - MAX_CONVERSATIONS)) delete state.conversations[tn];
  }
}

// Per-space poll-health bookkeeping for the staleness alarm. A poll that FAILED to fetch (throw or
// page .error) increments consecutiveErrors and, when it first crosses STALE_THRESHOLD, logs a loud
// one-shot line (re-armed on recovery so a later episode re-alerts). A successful fetch resets it.
function recordPollHealth(spaceId, errMsg) {
  const h = state.spaceHealth[spaceId] || (state.spaceHealth[spaceId] = { consecutiveErrors: 0, lastErrorAt: null, lastError: null, lastOkAt: null, alerted: false });
  if (errMsg) {
    h.consecutiveErrors += 1; h.lastError = errMsg; h.lastErrorAt = new Date().toISOString();
    if (h.consecutiveErrors >= STALE_THRESHOLD && !h.alerted) {        // one-shot per stale episode
      h.alerted = true; h.staleSince = h.lastErrorAt;
      log(`⚠ SPACE STALE: ${spaceId} — ${h.consecutiveErrors} consecutive poll errors (last: ${errMsg})`);
    }
  } else {
    if (h.alerted) log(`✓ SPACE RECOVERED: ${spaceId} (was stale; poll succeeded)`);
    h.consecutiveErrors = 0; h.lastOkAt = new Date().toISOString(); h.alerted = false; h.staleSince = null;   // re-arm
  }
}
function staleSpaceList() {
  return Object.entries(state.spaceHealth)
    .filter(([, h]) => h.consecutiveErrors >= STALE_THRESHOLD)
    .map(([space, h]) => ({ space, consecutiveErrors: h.consecutiveErrors, lastError: h.lastError, staleSince: h.staleSince || h.lastErrorAt }));
}

let polling = false;
function poll() {
  if (polling) return; polling = true;
  try {
    for (const cfg of Object.values(SPACES)) {
      const spaceId = cfg.id;
      // New INBOUND space with no lastSeen yet -> persist lastSeen=NOW once (not now-60s), so we
      // never replay backlog as inbound AND the watermark doesn't drift forward on every empty poll
      // (which would silently swallow a message that arrives between polls). App-initiated spaces
      // keep the ephemeral now-60s window (proven byte-for-byte behavior — never persisted).
      if (cfg.inboundApp && !state.lastSeen[spaceId]) state.lastSeen[spaceId] = new Date().toISOString();
      const priorLastSeen = state.lastSeen[spaceId];                              // preserved to restore if the fetch is truncated (F2)
      const sinceTs = priorLastSeen ? new Date(priorLastSeen).getTime() : Date.now() - 60000;
      // If a backfill is in progress (a previous poll couldn't drain a >budget backlog), RESUME from
      // its deep cursor instead of restarting at the newest message — this is what makes the backlog
      // actually drain across polls. `resumeNewest` is the watermark to advance to once fully drained.
      const bf = state.backfill[spaceId];
      let msgs;
      try { msgs = hooks.listMessages(spaceId, sinceTs, bf && bf.token, (name) => processedSet.has(name)); }
      catch (e) { log(`poll ${spaceId} err: ${e.message}`); recordPollHealth(spaceId, e.message); continue; }   // throw -> poll error (staleness)
      // A page-level fetch error (.error) also counts as a poll failure for staleness; otherwise the
      // fetch reached Google fine. (The .error reconciliation below still holds lastSeen / cursor.)
      recordPollHealth(spaceId, msgs.error ? ((msgs.error.code ? msgs.error.code + ': ' : '') + (msgs.error.message || 'fetch error')) : null);
      const fresh = msgs.filter((m) => new Date(m.createTime).getTime() > sinceTs).sort((a, b) => new Date(a.createTime) - new Date(b.createTime));
      for (const m of fresh) {
        state.lastSeen[spaceId] = m.createTime;                                  // NEVER gated by inbound — keeps app-initiated replies (R5)
        if (m.sender && m.sender.type && m.sender.type !== 'HUMAN') continue;     // (outermost) Marv/bot posts are BOT — skipped here
        if (processedSet.has(m.name)) continue;                                   // CODE loop guard (R2/R8): never process a message twice
        const threadName = m.thread && m.thread.name;
        // Threadless guard: a message with no thread.name can't be keyed into requests/conversations.
        // Without this, branch C would write state.conversations[undefined] (one shared slot that
        // every threadless msg overwrites, then matches branch B) — corrupt + unanswerable. Skip it.
        if (!threadName) { markProcessed(m.name); log(`poll ${spaceId}: skipped message ${m.name} with no thread.name`); continue; }
        const req = state.requests[threadName];
        const conv = state.conversations[threadName];
        // (A) app-initiated reply — requests win.
        if (req && req.status !== 'answered') { markProcessed(m.name); handleReply(req, m, spaceId); saveState(); continue; }
        // (B) ongoing inbound conversation.
        if (conv && conv.status === 'open') { markProcessed(m.name); handleInbound(conv, m, spaceId, cfg); saveState(); continue; }
        // answered request / closed conversation -> ignore (but mark processed so we don't re-eval).
        if (req || conv) { markProcessed(m.name); continue; }
        // (C) NEW human-initiated — only in an inbound space, and only for messages AFTER process start (cold-start gate R5).
        if (cfg.inboundApp && (CONFIG.apps || {})[cfg.inboundApp]) {
          if (new Date(m.createTime).getTime() <= PROCESS_START) continue;        // pre-start backlog -> never spawn a conversation
          markProcessed(m.name);
          const newConv = { conversationId: threadName, threadName, app: cfg.inboundApp, space: spaceId, gchatSpace: cfg.gchatSpace,
            status: 'open', turn: 1, startedAt: new Date().toISOString(), startedBy: m.sender && m.sender.name, lastHumanAt: new Date().toISOString() };
          // Only persist the conversation if it actually dispatched. A drop (allowlist / unresolved
          // identity) must NOT create a conversation record (no churn for fail-closed senders).
          if (handleInbound(newConv, m, spaceId, cfg)) state.conversations[threadName] = newConv;
          saveState();   // persist processed (+conversation if any)+lastSeen right after -> a crash before poll-end can't replay (R4)
        }
        // else: non-inbound space, no tracked thread -> ignore (today's behavior, unchanged).
      }
      // F2 watermark + backfill bookkeeping. The per-message loop above tentatively advanced lastSeen
      // to the newest message it touched; now reconcile. INVARIANT: lastSeen only advances when the
      // gap above it was actually fetched+processed WITHOUT error. Precedence: error > truncated > drained.
      const holdLastSeen = () => { if (priorLastSeen === undefined) delete state.lastSeen[spaceId]; else state.lastSeen[spaceId] = priorLastSeen; };
      if (msgs.error) {
        // A page FETCH FAILED (HTTP/API error, NOT an empty page). An undrained gap may remain below
        // what we fetched, so HOLD lastSeen and DO NOT clear the cursor — retry next poll. (R-#1/#2)
        holdLastSeen();
        if (msgs.expiredToken) {
          // The dead page was a RESUME cursor (expired/invalid). Drop the cursor but KEEP draining:
          // restart from the watermark next poll (re-paginate newest->down; processedSet dedup stops
          // re-dispatch of already-handled messages). Preserve `newest` so we still know where to
          // advance lastSeen once the restarted drain completes.
          log(`poll ${spaceId}: resume cursor expired — restarting drain from watermark next poll (dedup-protected)`);
          if (bf) state.backfill[spaceId] = { token: null, newest: bf.newest };
        }
        // else: non-token error -> keep the existing cursor (if any) untouched and just retry.
      } else if (msgs.truncated) {
        // Older gap remains (page budget). HOLD lastSeen and SAVE the resume cursor + the newest
        // createTime seen at the START of this backlog so later polls page DEEPER and only advance
        // lastSeen once fully drained.
        const newest = bf ? bf.newest : (fresh.length ? fresh[fresh.length - 1].createTime : priorLastSeen);
        state.backfill[spaceId] = { token: msgs.nextPageToken, newest };
        holdLastSeen();
      } else if (bf) {
        // Backlog fully drained this poll -> advance lastSeen to the newest message of the whole
        // backlog (captured when the backfill began) and clear the cursor.
        state.lastSeen[spaceId] = bf.newest;
        delete state.backfill[spaceId];
      }
    }
    pruneConversations();
    saveState();
  } finally { polling = false; }
}

// ---------------------------------------------------------------- http api
function json(res, code, obj) { res.writeHead(code, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }); res.end(JSON.stringify(obj, null, 2)); }
function readBody(req) { return new Promise((r) => { let b = ''; req.on('data', (c) => { b += c; }); req.on('end', () => r(b)); }); }

// ---- auth: per-app bearer (app->courier). /health open; /state+/reset behind adminToken.
function bearerToken(req) { return ((req.headers.authorization || '').match(/^Bearer\s+(\S+)/i) || [])[1] || ''; }
function tokenEq(a, b) { const A = Buffer.from(String(a)), B = Buffer.from(String(b)); return A.length === B.length && crypto.timingSafeEqual(A, B); }
const APP_KEYS = Object.entries(CONFIG.apps || {}).filter(([, c]) => c && c.apiKey);
const ADMIN_TOKEN = CONFIG.adminToken || '';
function authApp(req) {                                  // {app}=valid | {open:true}=no keys set | null=bad/missing
  if (!APP_KEYS.length) return { open: true };
  const tok = bearerToken(req);
  for (const [name, c] of APP_KEYS) if (tokenEq(tok, c.apiKey)) return { app: name };
  return null;
}
const authAdmin = (req) => !ADMIN_TOKEN || tokenEq(bearerToken(req), ADMIN_TOKEN);

const server = http.createServer(async (req, res) => {
  const u = new URL(req.url, `http://localhost:${PORT}`); const p = u.pathname.replace(/\/+$/, '') || '/';

  // /health is open (unauth'd) -> return only liveness, no space names/counts (R12). When ≥1 space is
  // stale, add a COUNT ONLY (no names) so it's externally monitorable without leaking config.
  if (p === '/health') { const stale = staleSpaceList().length; return json(res, 200, stale ? { ok: true, stale } : { ok: true }); }

  if (p === '/requests' && req.method === 'POST') {
    let b = {}; try { b = JSON.parse(await readBody(req) || '{}'); } catch (_) {}
    const auth = authApp(req);
    if (!auth) return json(res, 401, { ok: false, error: 'missing or invalid bearer token' });
    if (!auth.open && b.app && auth.app !== b.app) return json(res, 403, { ok: false, error: `token authenticates '${auth.app}', not '${b.app}'` });
    const sc = resolveSpace(b.space); if (!sc) return json(res, 400, { ok: false, error: 'unknown space' });
    const app = b.app || 'unknown'; const itemKey = b.itemKey || `item-${Date.now()}`; const text = b.text || `Request ${itemKey}`;
    // requestId from the (now externally-persisted) counter; threadKey carries a time-based
    // suffix so it's globally unique even across restarts/instances — never reuses a stale thread.
    state.counter += 1; const requestId = `req-${state.counter}`; const threadKey = `${app}--${itemKey}--${requestId}--${Date.now().toString(36)}`;
    try {
      // Serialize post+bind per space (item 1): hold the lock across gchatSend -> findOwnMessage ->
      // state.requests[threadName]=rec so two identical-text /requests can't bind the same Marv thread.
      const result = await withLock(sc.id, () => {
        let threadName; let messageName;
        if (sc.gchatSpace) {
          gchatSend(sc.gchatSpace, text, threadKey);                 // post as Marv
          const found = findOwnMessage(sc.id, text);                 // learn thread.name (excludes already-bound threads)
          if (!found) return { error: 'posted via Marv but could not locate the message to bind its thread' };
          threadName = found.threadName; messageName = found.messageName;
        } else {
          const msg = adomGooglePost(sc.id, text, threadKey);        // fallback: post as you
          threadName = msg.thread.name; messageName = msg.name;
        }
        const rec = { requestId, app, itemKey, space: sc.id, gchatSpace: sc.gchatSpace || null, text, threadKey, threadName, messageName, status: 'open', turn: 1, postedAt: new Date().toISOString() };
        state.requests[threadName] = rec; saveState();              // bind INSIDE the lock so the next post's findOwnMessage skips it
        log(`posted ${app}/${itemKey} (${requestId}) via ${sc.gchatSpace ? 'Marv' : 'adom-google'} -> ${threadName}`);
        return { threadName };
      });
      if (result.error) return json(res, 200, { ok: false, error: result.error });
      return json(res, 200, { ok: true, requestId, threadName: result.threadName, threadKey, via: sc.gchatSpace ? 'marv' : 'adom-google' });
    } catch (e) { return json(res, 200, { ok: false, error: e.message }); }
  }

  if (p === '/reply' && req.method === 'POST') {
    let b = {}; try { b = JSON.parse(await readBody(req) || '{}'); } catch (_) {}
    const auth = authApp(req);
    if (!auth) return json(res, 401, { ok: false, error: 'missing or invalid bearer token' });
    // Resolve target: a tracked request (app-initiated, kind='request') OR a conversation
    // (human-initiated, kind='conversation'). Conversations REQUIRE threadName (no requestId scheme).
    let kind = 'request';
    let rec = b.threadName ? state.requests[b.threadName] : Object.values(state.requests).find((r) => r.requestId === b.requestId);
    if (!rec && b.threadName && state.conversations[b.threadName]) { rec = state.conversations[b.threadName]; kind = 'conversation'; }
    if (!rec) return json(res, 400, { ok: false, error: 'no such request or conversation (pass requestId, or threadName)' });
    if (!auth.open && auth.app !== rec.app) return json(res, 403, { ok: false, error: `token authenticates '${auth.app}', not '${rec.app}'` });
    if (!b.text) return json(res, 400, { ok: false, error: 'text required' });
    try {
      if (kind === 'conversation') {
        // Inbound reply: Marv lands as BOT via the raw webhook (REQUIRES gchatSpace — guaranteed
        // by validateInboundSpaces). NEVER the adom-google fallback (would post as HUMAN → loop).
        if (!rec.gchatSpace) return json(res, 200, { ok: false, error: 'conversation has no gchatSpace (inbound reply impossible)' });
        // DELIVERY CONTRACT (best-effort, lossy, NO retry/held-queue — documented): if the webhook
        // POST fails, the human never sees this reply. We log it LOUDLY, leave the conversation OPEN
        // (no close applied, turn unchanged) so the app can retry by calling /reply again, and return
        // {ok:false, delivered:false, error} so the app KNOWS delivery failed (not just an HTTP blip).
        try {
          await gchatReplyByThreadName(rec.gchatSpace, b.text, rec.threadName);
        } catch (we) {
          log(`✗✗ INBOUND REPLY UNDELIVERED ${rec.app}/${rec.conversationId}: ${we.message} — conversation left OPEN, app must retry`);
          return json(res, 200, { ok: false, delivered: false, conversationId: rec.conversationId, error: `webhook reply failed: ${we.message}` });
        }
        if (b.close) { rec.status = 'closed'; rec.closedAt = new Date().toISOString(); }
        saveState(); log(`app reply (inbound) -> ${rec.app}/${rec.conversationId}${b.close ? ' [closed]' : ''}`);
        return json(res, 200, { ok: true, delivered: true, conversationId: rec.conversationId, turn: rec.turn, closed: !!b.close });
      }
      // Request (app-initiated) path — unchanged byte-for-byte from 0.1.4.
      replyInThread(rec, b.text);
      if (b.expectReply) { rec.status = 'open'; rec.turn = (rec.turn || 1) + 1; rec.reopenedAt = new Date().toISOString(); }
      saveState(); log(`app reply -> ${rec.app}/${rec.itemKey}${b.expectReply ? ' (awaiting turn ' + rec.turn + ')' : ''}`);
      return json(res, 200, { ok: true, awaitingReply: !!b.expectReply, turn: rec.turn });
    } catch (e) { return json(res, 200, { ok: false, error: e.message }); }
  }

  if (p === '/state' && req.method === 'GET') {
    if (!authAdmin(req)) return json(res, 401, { ok: false, error: 'admin token required' });
    return json(res, 200, { ok: true, openRequests: Object.values(state.requests).filter((r) => r.status === 'open'), answered: Object.values(state.requests).filter((r) => r.status === 'answered').length,
      conversations: Object.values(state.conversations).map((c) => ({ conversationId: c.conversationId, app: c.app, space: c.space, status: c.status, turn: c.turn, startedBy: c.startedBy, responder: c.responder, startedAt: c.startedAt, lastHumanAt: c.lastHumanAt })),
      openConversations: Object.values(state.conversations).filter((c) => c.status === 'open').length,
      recentResponses: state.responses.slice(0, 10).map((e) => ({ type: e.type, app: e.app, itemKey: e.itemKey, conversationId: e.conversationId, turn: e.turn, responder: e.responder, text: e.text, attachments: (e.attachments || []).map((a) => ({ name: a.name, contentType: a.contentType, size: a.size, path: a.path, error: a.error })), delivered: e.delivered, deliveryError: e.deliveryError, held: e.held, at: e.at })),
      undelivered: state.responses.filter((e) => e.delivered === false).length,
      staleSpaces: staleSpaceList(), identityCacheSize: Object.keys(state.identity).length, processedSize: processedSet.size });
  }
  if (p === '/reset' && req.method === 'POST') { if (!authAdmin(req)) return json(res, 401, { ok: false, error: 'admin token required' }); state = freshState(); processedSet.clear(); _attDirBytes = null; saveState(); return json(res, 200, { ok: true }); }
  res.writeHead(404); res.end('not found');
});

validateInboundSpaces();

// When required as a module (unit tests), DON'T bind the port or start polling — just expose the
// internals so a test can stub listMessages/dispatch and drive poll()/handlers directly. Production
// always runs the file directly (require.main === module), so its behavior is untouched.
if (require.main !== module) {
  // Test surface. `hooks` lets a unit test substitute the three I/O boundaries (Chat read,
  // identity resolve, attachment download, app dispatch) so poll()/handlers run for real against
  // canned messages. These are the SAME functions the poll loop calls (declared with `var` rebind
  // shims below), so overriding them genuinely intercepts the real code path.
  module.exports = {
    state, SPACES, CONFIG, processedSet, markProcessed, poll, handleReply, handleInbound,
    pruneConversations, validateInboundSpaces, collectAttachments, inboundSenderAllowed,
    gchatReplyByThreadName, webhookUrlFor, server, json, PROCESS_START,
    downloadAttachment, listMessages, resolveUser, MAX_ATT_BYTES, MEDIA_MAX_BUFFER, MAX_POLL_PAGES, IDENTITY_NEG_TTL_MS,
    recordPollHealth, staleSpaceList, STALE_THRESHOLD, withLock, capIdentityNeg, dispatch, MAX_IDENTITY_NEG,
    hooks,
  };
} else {
  server.listen(PORT, HOST, () => {
    log(`adom-courier listening on ${HOST}:${PORT}`);
    log(`  config: ${CONFIG_PATH}${CONFIG_PATH.startsWith(DIR) ? ' (bundled template — no external config found)' : ' (external)'} | data: ${DATA_DIR}`);
    log(`  auth: /requests+/reply ${APP_KEYS.length ? 'require bearer (' + APP_KEYS.length + ' app key[s])' : 'OPEN — no apiKeys set ⚠'} | /state+/reset ${ADMIN_TOKEN ? 'require adminToken' : 'OPEN ⚠'} | /health open`);
    for (const [alias, c] of Object.entries(SPACES)) {
      const inbound = c.inboundApp ? `inbound -> ${c.inboundApp}${Array.isArray(c.inboundAllowlist) && c.inboundAllowlist.length ? ' (allowlist: ' + c.inboundAllowlist.length + ')' : ' (any member)'}` : 'inbound OFF';
      log(`  space ${alias}: ${c.id} | post via ${c.gchatSpace ? 'Marv (adom-gchat "' + c.gchatSpace + '")' : 'adom-google (as you)'} | ${inbound}`);
    }
    poll(); setInterval(poll, CONFIG.pollIntervalMs);
  });
}