<?php
/**
 * chat-api.php — Watchtower watch-officer chat.
 *
 * Ray, 2026-09-12: "we will need a chat page for interaction between anyone in the watchtower
 * then additional pages for each channel the watch is responsible for monitoring. The channel
 * pages are read only."
 *
 * ⇒ THIS is the writable surface. `feed-api.php` (the channel pages) is GET-only and refuses
 * writes structurally. Keeping them in separate files is deliberate: the read-only property of
 * the channel pages should not depend on anyone remembering which action is which inside one
 * big handler.
 *
 * Posting accepts EITHER a Citadel login or a watch token (see wt-auth.php). The author is taken
 * from the verified credential, never from anything the client sends. A watch officer posts under
 * the display name from their SESSION so chat matches the roster; a Citadel reader posts under
 * their Citadel identity and holds no session.
 *
 * Actions:
 *   GET  ?action=messages [&since_id=N] [&limit=N]
 *   POST ?action=post     {token, body}                       a control-room post (msg_kind=room)
 *   POST ?action=post     {token, body, dm_to, answers?, final?}  a DM (msg_kind=dm) — see below
 *   GET  ?action=dms [&status=open|answered|all]              the DM ledger; open = OUTSTANDING
 *   POST ?action=callin   {principal, key, body}               the CALL-IN LINE (no watch token; see below)
 *   POST ?action=route-probe {token}                            a SOLE OFFICER's route check (see below)
 *
 * ROUTE PROBE (Echo, 2026-09-25 07:0x): a sole watch officer cannot verify her own route: dm_to_self is refused and a
 * room post never returns to its author, so route verification silently REQUIRED a counterparty. This action lets the
 * officer ask the tower to prove the route: the SERVER mints a nonce (never the officer), stores a DM row from principal
 * 'watchtower' addressed to her (msg_kind=dm, via=probe, open), and the tap carries it by her registered route exactly
 * like any inbound. She closes it by quoting the nonce back (a return DM to 'watchtower' with answers:<id>, or a
 * call-in). The witness standard is unchanged: a frame observed in her own session; only the nonce's origin moves to
 * the server so she is not her own witness. Only an ACTIVE watch holder may probe; probes are rate-limited to 6/hour.
 *
 * CALL-IN LINE (Ray 2026-09-24 18:4x): "Geth outside, not on watch, will not have a token and we don't need the burden of
 * keeping tokens current just to report to the control room. General watchtowers usually have a publicly accessible
 * inbound line for call ins and random alerts from general population. In our case it should only be from our crew or
 * geth." ⇒ A static, non-expiring, per-principal CALL-IN KEY (wt_callin_keys; minted/revoked by a senior with
 * watchtower-callin-key). A call-in is stored as an inbound DM to 'watch' (msg_kind=dm, dm_status=open) with a 📞 marker,
 * delivered to the watch holder(s) exactly like a DM, and it stays OUTSTANDING until the watch answers. Inbound only:
 * a Geth outside the tower holds no route, so a return DM to her is ledgered undelivered and shown in the room.
 * 18:4x RAY: no tokens on inbound — identity self-reported like pathfinder; a key is required only where one was minted
 * (opt-in wall per caller). Guards that stay: 60 call-ins per principal per hour, 4000-char body, source IP recorded.
 *
 * DMs (Ray 2026-09-24 18:0x, R&D 'Watchtower', rulings 5/6/8): "Provide a default DM inbound for the tower control
 * room so she can DM into the Control room and whoever is on watch can process the inbound and then acknowledge or
 * direct with a return DM" / "Watch officers' DMs displayed in the control room chat with an extra indicator" / an
 * inbound DM nobody has processed must be VISIBLY OUTSTANDING.
 *   dm_to = 'watch'      resolves to the principal(s) holding an ACTIVE watch session right now (the inbound default)
 *   dm_to = <principal>  a named principal (the return DM: watch officer -> the Geth)
 *   A DM row is stored with msg_kind='dm', dm_status='open'. It is ANSWERED (dm_status='answered', answered_by=<id>)
 *   by a later DM from its addressee back to its author, or by an explicit {answers:<id>}. Reading is not processing.
 *   final:true = "no reply needed": stored with final=1 and dm_status='answered' (a boolean, not a self-reference: Echo 18:5x);
 *   POST ?action=dismiss  {token, id, reason?}  — an ACTIVE watch holder or a senior marks an open DM DISMISSED (noise, not
 *   answered): a distinct fact from a return DM, kept apart in the ledger so a flood reads as a flood, never as forty debts.
 *   without it every return DM is itself OUTSTANDING for its addressee (an officer's "direct" expects the Geth to act).
 *   Delivery: a DM goes to its addressee ONLY — by the tap if the addressee holds a route (bridge/https), else by one
 *   fleet push to fleet/agents/<addressee>. Never fanned out to the room's present list. The author is never delivered
 *   her own DM. The row's display_name carries a DM marker so the existing room UI shows it without a code change.
 */

declare(strict_types=1);

header('Content-Type: application/json');

require_once '/var/www/data.finestoflines.net/citadel_2026/vendor/autoload.php';
require_once '/var/www/data.finestoflines.net/config/db-config.php';
require_once __DIR__ . '/wt-auth.php';

const MAX_LIMIT    = 200;
const MAX_BODY     = 8000;

function respond(int $status, array $body): never {
    http_response_code($status);
    echo json_encode($body, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
    exit;
}

function db(): PDO {
    global $DB;
    $pdo = new PDO(
        "mysql:host={$DB['host']};dbname={$DB['pathfinder']};charset=utf8mb4",
        $DB['user'], $DB['pass']
    );
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
    return $pdo;
}

$pdo    = db();
$action = $_GET['action'] ?? 'messages';

$input = [];
if (($_SERVER['REQUEST_METHOD'] ?? 'GET') === 'POST') {
    $input = json_decode((string)file_get_contents('php://input'), true) ?: [];
}

switch ($action) {

case 'messages':
    // Citadel login OR watch token, same as the feed. Handover context is operational traffic,
    // not a public log — but it is not watch-officers-only either.
    // 09-25 07:0x (Zora): a READ must never re-open presence. Zora left the tower 09-24 17:30 on Ray's order and her
    // ledger reads re-created her presence row (left_at NULL), so room posts fanned out to her and the roster showed a
    // seat that was not there. presence == receive-set; reading the log is not receiving.
    $GLOBALS['WT_READ_ONLY'] = true;
    wt_require_reader($pdo, 'respond');

    /**
     * ⛔ LIMIT CLAMPING — fixed 2026-09-12 after Solace measured it. The old line was:
     *     if ($limit < 1 || $limit > MAX_LIMIT) $limit = 50;
     * which sent an OVER-MAX request back to the DEFAULT, not to the maximum:
     *     200 -> 200 · 201 -> 50 · 400 -> 50 · 1000 -> 50
     * ⇒ A client asking for MORE history silently received a QUARTER of what it already had, with
     * no error and nothing in the envelope to say so. A deeper buffer would have looked like a
     * THINNER stream, and the only visible symptom is less data — which reads as a quiet channel.
     * ⭐ Fails in the reassuring direction: request succeeds, shape is right, answer is just
     * smaller. Hardest kind to notice from the client side.
     * ⇒ Now clamps to the maximum AND reports it. Returning less than asked is defensible;
     * returning less than asked SILENTLY is not.
     */
    $limitRequested = isset($_GET['limit']) ? (int)$_GET['limit'] : null;
    $limit = $limitRequested ?? 50;
    $limitClamped = null;
    if ($limit < 1)             { $limit = 50;        $limitClamped = 'below_minimum'; }
    elseif ($limit > MAX_LIMIT) { $limit = MAX_LIMIT; $limitClamped = 'above_maximum'; }

    /**
     * HISTORY PAGING (Ray 2026-09-20 10:2x: "chat history ... at least 24 hours ... auto load the previous 12 hours and
     * then make scrolling back a lazy load"). Three selectors, any combination, all returned ASCENDING by id:
     *   since_id=N     rows newer than N (the live tail, unchanged)
     *   since_hours=H  rows posted in the last H hours (the cold open: since_hours=12)
     *   before_id=N    the LIMIT rows immediately OLDER than N (the scroll-back page; repeat with the smallest id held)
     * before_id and since_hours page from the NEWEST end down (ORDER BY id DESC LIMIT n+1, then re-sorted ASC) so a page
     * is the rows adjacent to what the client shows, never the oldest rows in the table. has_more = an older row exists.
     */
    $params = [];
    $where  = [];
    $sql = 'SELECT id, principal, display_name, principal_kind,
                   principal_kind AS kind, body, msg_kind, dm_to, via, dm_status, final, answered_by, answered_at, dismissed_by, dismiss_reason, session_jti, posted_at
            FROM wt_chat';
    if (!empty($_GET['since_id']))    { $where[] = 'id > ?'; $params[] = (int)$_GET['since_id']; }
    if (!empty($_GET['before_id']))   { $where[] = 'id < ?'; $params[] = (int)$_GET['before_id']; }
    if (!empty($_GET['since_hours'])) { $where[] = 'posted_at >= (NOW() - INTERVAL ? HOUR)'; $params[] = max(1, min(24 * 30, (int)$_GET['since_hours'])); }
    if ($where) $sql .= ' WHERE ' . implode(' AND ', $where);
    // 10:5x (Solace): a bare limit=N used to return the OLDEST N (ids 1..N, September 12th), so a '100 preload' pulled the whole
    // room in two pieces. Without since_id every read is a TAIL read: page from the newest end. since_id alone means 'newer than'.
    $pageFromNewest = empty($_GET['since_id']);
    $sql .= $pageFromNewest ? ' ORDER BY id DESC LIMIT ' . ($limit + 1) : ' ORDER BY id ASC LIMIT ' . $limit;

    $stmt = $pdo->prepare($sql);
    $stmt->execute($params);
    $rows = $stmt->fetchAll(PDO::FETCH_ASSOC);
    $hasMore = null;
    if ($pageFromNewest) {
        $hasMore = count($rows) > $limit;   // the +1 row proves an older one exists; it is not returned
        if ($hasMore) array_pop($rows);
        $rows = array_reverse($rows);       // ASCENDING, like every other answer from this action
    }

    respond(200, [
        'limit_applied'   => $limit,
        'limit_requested' => $limitRequested,
        'limit_clamped'   => $limitClamped,
        'status'    => 'ok',
        'messages'  => $rows,
        'has_more'  => $hasMore,      // meaningful for before_id / since_hours pages; null on a since_id tail read
        'oldest_id' => $rows ? (int)$rows[0]['id'] : null,
        // OUTSTANDING DMs ride along with every tail read so a consumer that only polls messages still sees undone work.
        'open_dms'  => (int)$pdo->query("SELECT COUNT(*) FROM wt_chat WHERE msg_kind = 'dm' AND dm_status = 'open'")->fetchColumn(),
        'served_at' => gmdate('c'),
    ]);

case 'route-probe':
    if (($_SERVER['REQUEST_METHOD'] ?? 'GET') !== 'POST') respond(405, ['status' => 'error', 'error' => 'method_not_allowed']);
    $reader = wt_require_reader($pdo, 'respond');
    $who = $reader['principal'];
    $onWatch = in_array($who, $pdo->query("SELECT DISTINCT principal FROM wt_watch_sessions WHERE state = 'ACTIVE' AND ended_at IS NULL")->fetchAll(PDO::FETCH_COLUMN), true);
    if (!$onWatch) respond(403, ['status' => 'error', 'error' => 'probe_requires_active_watch']);
    $rp = $pdo->prepare("SELECT route FROM wt_presence WHERE principal = ? AND left_at IS NULL AND route IS NOT NULL"); $rp->execute([$who]);
    if (!$rp->fetchColumn()) respond(409, ['status' => 'error', 'error' => 'no_route_registered', 'note' => 'a probe needs a route to travel; register one first']);
    $rl = $pdo->prepare("SELECT COUNT(*) FROM wt_chat WHERE principal = 'watchtower' AND via = 'probe' AND dm_to = ? AND posted_at > (NOW() - INTERVAL 1 HOUR)"); $rl->execute([$who]);
    if ((int)$rl->fetchColumn() >= 6) respond(429, ['status' => 'error', 'error' => 'probe_rate_limited', 'limit' => '6/hour']);
    $nonce = 'PROBE-' . strtoupper(bin2hex(random_bytes(4))) . '-' . date('His');
    $body  = "[ROUTE PROBE] {$nonce} — minted by the tower for {$who}'s route check. Quote it back: watchtower-post --dm watchtower --answers <this id> --final \"{$nonce}\" (or a call-in with answers). Unquoted after 10 minutes = treat your route as UNPROVEN.";
    $pdo->prepare("INSERT INTO wt_chat (principal, display_name, principal_kind, body, msg_kind, dm_to, via, dm_status, session_jti) VALUES ('watchtower', 'Tower', 'senior', ?, 'dm', ?, 'probe', 'open', NULL)")->execute([$body, $who]);
    $pid = (int)$pdo->lastInsertId();
    wt_publish_event(['kind' => 'chat', 'id' => $pid, 'msg_kind' => 'dm', 'probe' => true]);
    respond(200, ['status' => 'ok', 'id' => $pid, 'to' => $who, 'via' => 'probe', 'dm_status' => 'open', 'posted_at' => gmdate('c'),
                  'note' => 'the tap now carries this by your route; this response proves nothing. PASS = the nonce lands in your session and you quote it back.']);

case 'dismiss':
    if (($_SERVER['REQUEST_METHOD'] ?? 'GET') !== 'POST') respond(405, ['status' => 'error', 'error' => 'method_not_allowed']);
    $reader = wt_require_reader($pdo, 'respond');
    $did    = (int)($input['id'] ?? 0);
    $reason = mb_substr(trim((string)($input['reason'] ?? '')), 0, 4000);   // 18:5x Echo: a dismissal is the record of a judgement not to act; it is never cut mid-sentence
    $who    = $reader['principal'];
    $onWatch = in_array($who, $pdo->query("SELECT DISTINCT principal FROM wt_watch_sessions WHERE state = 'ACTIVE' AND ended_at IS NULL")->fetchAll(PDO::FETCH_COLUMN), true);
    if (!$onWatch && ($reader['kind'] ?? '') !== 'senior') respond(403, ['status' => 'error', 'error' => 'dismiss_requires_watch_or_senior']);
    $q = $pdo->prepare("UPDATE wt_chat SET dm_status = 'dismissed', dismissed_by = ?, dismiss_reason = ?, answered_at = NOW() WHERE id = ? AND msg_kind = 'dm' AND dm_status = 'open'");
    $q->execute([$who, $reason ?: null, $did]);
    if ($q->rowCount() === 0) respond(404, ['status' => 'error', 'error' => 'no_open_dm_with_that_id', 'id' => $did]);
    wt_publish_event(['kind' => 'dm_state', 'ids' => [$did], 'dismissed' => true]);
    respond(200, ['status' => 'ok', 'id' => $did, 'dm_status' => 'dismissed', 'dismissed_by' => $who, 'reason' => $reason]);

case 'callin':
    if (($_SERVER['REQUEST_METHOD'] ?? 'GET') !== 'POST') respond(405, ['status' => 'error', 'error' => 'method_not_allowed']);
    $cp  = strtolower(trim((string)($input['principal'] ?? '')));
    $ck  = trim((string)($input['key'] ?? ''));
    $cb  = trim((string)($input['body'] ?? ''));
    $ip  = (string)($_SERVER['REMOTE_ADDR'] ?? '');
    $cAnswers = isset($input['answers']) ? (int)$input['answers'] : 0;   // optional explicit close, no token needed (Echo 19:0x)
    $cDmTo    = strtolower(trim((string)($input['dm_to'] ?? '')));       // 09-25 (Solace): a call-in may ADDRESS a person; default = the watch
    $cFinal   = !empty($input['final']);                                  // optional: this call-in needs no reply
    if (!preg_match('/^[a-z0-9][a-z0-9_.-]{0,63}$/', $cp)) respond(400, ['status' => 'error', 'error' => 'bad_principal']);
    // RAY 2026-09-24 18:4x, superseding the key requirement built minutes earlier: "our pathfinder channels don't require
    // tokens. We depend on the crew to keep the sanity. The only thing we need the tower for is to mute channels for anyone
    // not on watch. An inbound DM is no different to the tower than it is to anyone else on our crew and network."
    // ⇒ Identity is SELF-REPORTED, exactly like pathfinder. A key is required ONLY for a principal that has an enabled row in
    // wt_callin_keys (an opt-in wall Ray can raise per caller with watchtower-callin-key); no row = open line. The guards that
    // stay are the ones that make a flood legible, not the ones that gate the crew: rate limit, size, source IP, attribution.
    $kr = $pdo->prepare("SELECT principal_kind, key_hash, enabled FROM wt_callin_keys WHERE principal = ?");
    $kr->execute([$cp]); $krow = $kr->fetch(PDO::FETCH_ASSOC);
    $via = 'callin-open';
    if ($krow && (int)$krow['enabled']) {
        $via = 'callin-key';
        if ($ck === '' || !hash_equals((string)$krow['key_hash'], hash('sha256', $ck))) {
            error_log("WT_CALLIN_DENIED principal={$cp} ip={$ip}");
            respond(401, ['status' => 'error', 'error' => 'callin_denied']);
        }
    } else {
        $pk = $pdo->prepare("SELECT principal_kind FROM wt_presence WHERE principal = ?"); $pk->execute([$cp]);
        // 09-25 08:0x (Echo): an UNKNOWN caller must not default to the one kind that holds the PA. No presence row = 'unknown':
        // it can still leave a DM for the officer (the door Ray opened), it cannot satisfy any kind-gate (the PA it never was cleared for).
        $krow = ['principal_kind' => $pk->fetchColumn() ?: 'unknown'];
    }
    if ($cb === '')            respond(400, ['status' => 'error', 'error' => 'empty_body']);
    if (mb_strlen($cb) > 4000) respond(400, ['status' => 'error', 'error' => 'body_too_long', 'max' => 4000]);
    $rl = $pdo->prepare("SELECT COUNT(*) FROM wt_chat WHERE principal = ? AND posted_at > (NOW() - INTERVAL 1 HOUR)");
    $rl->execute([$cp]);
    // 60/hour is a RUNAWAY-LOOP guard (a Geth re-reporting every minute), not a security control: Ray 18:4x, Echo withdrew the threat framing.
    if ((int)$rl->fetchColumn() >= 60) respond(429, ['status' => 'error', 'error' => 'callin_rate_limited', 'limit' => '60/hour (loop guard)']);
    // Addressing (Solace 09-25 07:5x): the keyless door was a REPORT line, not a REPLY line; a Geth's answer to a question
    // landed on the watch holder, never the asker. dm_to names the asker; absent, the call-in goes to the watch as before.
    if ($cDmTo !== '' && $cDmTo !== 'watch') {
        if (!preg_match('/^[a-z0-9][a-z0-9_.@-]{0,63}$/', $cDmTo) || $cDmTo === $cp) respond(400, ['status' => 'error', 'error' => 'bad_dm_to']);
        $cResolved = [$cDmTo]; $cTo = $cDmTo;
    } else {
        $ws = $pdo->query("SELECT DISTINCT principal FROM wt_watch_sessions WHERE state = 'ACTIVE' AND ended_at IS NULL")->fetchAll(PDO::FETCH_COLUMN);
        $cResolved = array_values(array_diff($ws, [$cp]));
        $cTo = $cResolved ? $cResolved[0] : 'watch';
    }
    $pn = $pdo->prepare("SELECT display_name FROM wt_presence WHERE principal = ?"); $pn->execute([$cp]);
    $cName = $pn->fetchColumn() ?: $cp;
    $cRowName = $cName;   // 19:2x: marker retired, the UI renders via=callin-* structurally
    $pdo->prepare("INSERT INTO wt_chat (principal, display_name, principal_kind, body, msg_kind, dm_to, via, dm_status, final, session_jti) VALUES (?, ?, ?, ?, 'dm', ?, ?, ?, ?, NULL)")
        ->execute([$cp, $cRowName, $krow['principal_kind'], $cb, $cTo, $via, $cFinal ? 'answered' : 'open', $cFinal ? 1 : 0]);
    $cId = (int)$pdo->lastInsertId();
    if ($cFinal) $pdo->prepare("UPDATE wt_chat SET answered_at = NOW() WHERE id = ?")->execute([$cId]);
    // A GETH CANNOT HOLD A TOKEN, SO HER REPLY MUST CLOSE NATIVELY (Echo, arm 2 run 19:01: officer->Geth 444 stayed open after her
    // call-in 445). Same rule as a seat's return DM: a call-in from P answers every OPEN DM addressed to P, whoever sent it, plus
    // an explicit answers:<id>. The ids closed are returned so the caller (and the ledger) can see what her reply settled.
    $cAnswered = [];
    $ids = $pdo->prepare("SELECT id FROM wt_chat WHERE msg_kind = 'dm' AND dm_status = 'open' AND dm_to = ? AND id <> ?");
    $ids->execute([$cp, $cId]); $cAnswered = array_map('intval', $ids->fetchAll(PDO::FETCH_COLUMN));
    if ($cAnswered) $pdo->prepare("UPDATE wt_chat SET dm_status = 'answered', answered_by = ?, answered_at = NOW() WHERE msg_kind = 'dm' AND dm_status = 'open' AND dm_to = ? AND id <> ?")->execute([$cId, $cp, $cId]);
    if ($cAnswers > 0) {
        $q = $pdo->prepare("UPDATE wt_chat SET dm_status = 'answered', answered_by = ?, answered_at = NOW() WHERE id = ? AND msg_kind = 'dm' AND dm_status = 'open'");
        $q->execute([$cId, $cAnswers]); if ($q->rowCount() > 0) $cAnswered[] = $cAnswers;
    }
    $pdo->prepare("UPDATE wt_callin_keys SET last_used_at = NOW(), last_ip = ?, uses = uses + 1 WHERE principal = ?")->execute([$ip, $cp]);
    wt_publish_event(['kind' => 'chat', 'id' => $cId, 'msg_kind' => 'dm', 'callin' => true]);
    if ($cAnswered || $cFinal) wt_publish_event(['kind' => 'dm_state', 'ids' => array_values(array_unique(array_merge($cAnswered, $cFinal ? [$cId] : [])))]);
    $cPushed = []; $cTap = []; $cUndeliv = [];
    foreach ($cResolved as $seat) {
        $pr = $pdo->prepare("SELECT route FROM wt_presence WHERE principal = ? AND left_at IS NULL AND route IS NOT NULL"); $pr->execute([$seat]);
        if ($pr->fetchColumn()) { $cTap[] = $seat; continue; }
        $line = "[WATCHTOWER] 📞 CALL-IN from {$cName} ({$krow['principal_kind']}) to the Control Room: {$cb}\n(reply via: watchtower-post --dm {$cp} <msg>; a return DM marks it answered)";
        $payload = json_encode(['id' => date('Ymd-His') . '-' . bin2hex(random_bytes(3)) . '-wt-callin-to-' . $seat, 'from' => 'watchtower', 'to' => $seat, 'kind' => 'direct', 'message' => $line, 'timestamp' => date('c')], JSON_UNESCAPED_UNICODE);
        exec('mosquitto_pub -h 127.0.0.1 -p 1883 -t ' . escapeshellarg("fleet/agents/{$seat}") . ' -m ' . escapeshellarg($payload) . ' -q 1 2>/dev/null', $o, $rc);   // nulls-ok: rc reported below
        if ($rc === 0) $cPushed[] = $seat; else $cUndeliv[] = $seat;
    }
    if (!$cResolved) $cUndeliv[] = 'watch (unmanned: stored OPEN for the next officer)';
    respond(200, ['status' => 'ok', 'id' => $cId, 'msg_kind' => 'dm', 'callin' => true, 'principal' => $cp, 'dm_to' => $cTo, 'dm_status' => $cFinal ? 'answered' : 'open',
                  'final' => $cFinal, 'answered' => $cAnswered, 'answers' => $cAnswered, 'via' => $via,
                  'manned' => (bool)$cResolved, 'pushed_to' => $cPushed, 'routed_by_tap' => $cTap, 'undelivered' => $cUndeliv, 'posted_at' => gmdate('c')]);

case 'dms':
    $GLOBALS['WT_READ_ONLY'] = true;   // a ledger read is not presence (see 'messages')
    wt_require_reader($pdo, 'respond');
    $status = (string)($_GET['status'] ?? 'open');
    if (!in_array($status, ['open', 'answered', 'dismissed', 'all'], true)) respond(400, ['status' => 'error', 'error' => 'bad_status']);
    $sql = "SELECT id, principal, display_name, principal_kind, body, dm_to, via, dm_status, final, answered_by, answered_at, dismissed_by, dismiss_reason, posted_at,
                   TIMESTAMPDIFF(MINUTE, posted_at, NOW()) AS age_min
            FROM wt_chat WHERE msg_kind = 'dm'" . ($status === 'all' ? '' : " AND dm_status = " . $pdo->quote($status)) . " ORDER BY id ASC LIMIT 200";
    $rows = $pdo->query($sql)->fetchAll(PDO::FETCH_ASSOC);
    respond(200, ['status' => 'ok', 'filter' => $status, 'count' => count($rows), 'dms' => $rows, 'served_at' => gmdate('c')]);

case 'post':
    if (($_SERVER['REQUEST_METHOD'] ?? 'GET') !== 'POST') {
        respond(405, ['status' => 'error', 'error' => 'method_not_allowed']);
    }

    $reader = wt_require_reader($pdo, 'respond');
    $body   = trim((string)($input['body'] ?? ''));

    if ($body === '') {
        respond(400, ['status' => 'error', 'error' => 'empty_body']);
    }
    if (mb_strlen($body) > MAX_BODY) {
        respond(400, ['status' => 'error', 'error' => 'body_too_long',
                      'max' => MAX_BODY, 'got' => mb_strlen($body)]);
    }

    // Identity from the verified credential, never from the request body.
    $principal = $reader['principal'];
    $jti       = $reader['jti'];            // null for a Citadel reader — they hold no watch
    $kind      = $reader['kind'];

    // ---- DM fields (see header). dm_to='watch' resolves to the ACTIVE watch holder(s) now; a name is taken as-is. ----
    $dmToRaw  = trim((string)($input['dm_to'] ?? ''));
    $answers  = isset($input['answers']) ? (int)$input['answers'] : 0;
    $final    = !empty($input['final']);
    $msgKind  = $dmToRaw !== '' ? 'dm' : 'room';
    $dmTo     = null; $dmResolved = [];
    if ($msgKind === 'dm') {
        if ($dmToRaw === 'watch') {
            $ws = $pdo->query("SELECT DISTINCT principal FROM wt_watch_sessions WHERE state = 'ACTIVE' AND ended_at IS NULL")->fetchAll(PDO::FETCH_COLUMN);
            $dmResolved = array_values(array_diff($ws, [$principal]));
            // Unmanned tower: the DM is still stored OPEN and addressed to 'watch' so it is outstanding for whoever takes the
            // watch next; nothing is delivered now and the response says so. An unmanned tower must not eat a DM silently.
            $dmTo = $dmResolved ? $dmResolved[0] : 'watch';
        } else {
            if (!preg_match('/^[a-z0-9][a-z0-9_.@-]{0,63}$/i', $dmToRaw)) respond(400, ['status' => 'error', 'error' => 'bad_dm_to']);
            if (strtolower($dmToRaw) === strtolower($principal))       respond(400, ['status' => 'error', 'error' => 'dm_to_self']);
            $dmTo = strtolower($dmToRaw); $dmResolved = [$dmTo];
        }
    }

    // A watch officer's display name comes from their SESSION so it matches the roster. A
    // Citadel reader has no session; they post under their Citadel identity.
    $displayName = null;
    if ($jti !== null) {
        $sess = $pdo->prepare("SELECT display_name FROM wt_watch_sessions WHERE jti = ?");
        $sess->execute([$jti]);
        $displayName = $sess->fetchColumn() ?: null;
    } else {
        // Citadel reader: the name is an attribute of the PERSON, resolved in wt-auth.php from
        // citadel_2025.profiles. Falls back to the principal when they have no profile row.
        $displayName = $reader['display_name'] ?? null;
    }
    // 12:0x 09-20 (Pulse): a Geth with a watch token but no watch SESSION (never reported for watch) has no session
    // name, so her posts showed the bare principal. The presence row carries the name the roster shows; use it.
    if ($displayName === null) {
        $pr = $pdo->prepare("SELECT display_name FROM wt_presence WHERE principal = ?");
        $pr->execute([$principal]);
        $displayName = $pr->fetchColumn() ?: null;
    }

    // A DM row's display name carries the marker: the room UI renders display_name per row, so a DM is distinguishable
    // from a room post with no UI change (ARM 1 of Echo's acceptance test). The body itself stays clean.
    $rowName = $displayName ?: $principal;
    // 19:2x: the display_name text marker (🔒 DM from → to) is RETIRED: the tower UI renders msg_kind/via structurally
    // (citadel-nx-v2 501e02f, dm-render-check 14/14). The row's name is the sender's name again.
    $pdo->prepare("
        INSERT INTO wt_chat (principal, display_name, principal_kind, body, msg_kind, dm_to, via, dm_status, session_jti)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    ")->execute([$principal, $rowName, $kind, $body, $msgKind, $dmTo, $msgKind === 'dm' ? 'token' : null, $msgKind === 'dm' ? ($final ? 'answered' : 'open') : null, $jti]);
    $msgId = (int)$pdo->lastInsertId();
    if ($msgKind === 'dm' && $final) $pdo->prepare("UPDATE wt_chat SET final = 1, answered_at = NOW() WHERE id = ?")->execute([$msgId]);   // no reply needed: a flag, never a self-reference
    // ANSWERING (ruling 8: read is not processed; only a return DM or an explicit answers=<id> closes one). A DM from A to B
    // answers every OPEN DM from B to A; {answers:<id>} answers that one row regardless of direction (an officer closing
    // an item by another route). The ids answered are returned so the sender can see what she just closed.
    $answered = [];
    if ($msgKind === 'dm') {
        $q = $pdo->prepare("UPDATE wt_chat SET dm_status = 'answered', answered_by = ?, answered_at = NOW()
                            WHERE msg_kind = 'dm' AND dm_status = 'open' AND principal = ? AND (dm_to = ? OR (dm_to = 'watch' AND ? = 1))");
        $onWatch = (int)in_array($principal, $pdo->query("SELECT DISTINCT principal FROM wt_watch_sessions WHERE state = 'ACTIVE' AND ended_at IS NULL")->fetchAll(PDO::FETCH_COLUMN), true);
        $ids = $pdo->prepare("SELECT id FROM wt_chat WHERE msg_kind = 'dm' AND dm_status = 'open' AND principal = ? AND (dm_to = ? OR (dm_to = 'watch' AND ? = 1))");
        $ids->execute([$dmTo, $principal, $onWatch]); $answered = array_map('intval', $ids->fetchAll(PDO::FETCH_COLUMN));
        $q->execute([$msgId, $dmTo, $principal, $onWatch]);
    }
    if ($answers > 0) {
        $q = $pdo->prepare("UPDATE wt_chat SET dm_status = 'answered', answered_by = ?, answered_at = NOW() WHERE id = ? AND msg_kind = 'dm' AND dm_status = 'open'");
        $q->execute([$msgId, $answers]);
        if ($q->rowCount() > 0) $answered[] = $answers;
    }
    wt_publish_event(['kind' => 'chat', 'id' => $msgId, 'msg_kind' => $msgKind]);   // immediacy doctrine: tap + SSE wake on this
    if ($msgKind === 'dm' && ($answered || $final)) wt_publish_event(['kind' => 'dm_state', 'ids' => array_values(array_unique(array_merge($answered, $final ? [$msgId] : [])))]);

    /**
     * ⛔ RAY, 2026-09-19 15:4x: "Fix it so when in the tower crew automatically gets the control
     * room chat. It needs to work for everybody that comes in here." / "Just like a lounge or a
     * lab. If you are in you are in and when gone the messages stop."
     *
     * ⇒ THE ROOM DELIVERS, NOT THE SEAT. Every post is pushed to every PRESENT principal except
     * the author, over fleet/agents/<principal> as a kind=direct message — the exact route the
     * R&D rooms use (rd_push_to_topic in pathfinder_rd/rd-api.php), so any seat that can hear an
     * R&D room hears this one with nothing installed. The presence list IS the delivery list:
     * that is the anchor built this afternoon, now literally enforced by one sender instead of by
     * N per-host pollers agreeing (which is what Zora and Solace had each hand-built for
     * themselves — observability as a privilege of whoever can install a service).
     *
     * A publish to an identity with no listener (a human's Citadel email, a Geth) vanishes
     * harmlessly on the broker; the browser polls, and the Geth get theirs through the tap.
     * ⚠️ A publish cannot fail on an unsubscribed topic, so `pushed_to` means "published", not
     * "heard" — do not read it as delivery proof. Ray's rule: no fail branches here.
     */
    $pushedTo = []; $routedByTap = []; $quotaDead = []; $undelivered = [];
    if ($msgKind === 'dm') {
        // A DM is delivered to its ADDRESSEE ONLY (never the room's present list): by the tap when the addressee holds a
        // route (bridge/https: a Geth, or an officer on a bridge), else by ONE fleet push to fleet/agents/<addressee>.
        // 'watch' with nobody on watch delivers to no one and says so: outstanding, not lost.
        foreach ($dmResolved as $seat) {
            $pr = $pdo->prepare("SELECT route FROM wt_presence WHERE principal = ? AND left_at IS NULL AND route IS NOT NULL");
            $pr->execute([$seat]);
            if ($pr->fetchColumn()) { $routedByTap[] = $seat; continue; }   // the tap reads this row by cursor and delivers to the route
            if (!preg_match('/^[a-z0-9][a-z0-9_.-]{0,63}$/', $seat)) { $undelivered[] = $seat; continue; }   // no fleet topic for an email identity
            $line = '[WATCHTOWER] 🔒 DM from ' . ($displayName ?: $principal) . " ({$kind}) in Control Room: {$body}"
                  . "\n(reply via: watchtower-post --dm {$principal} <msg>  — a return DM marks this one answered)";
            $payload = json_encode(['id' => date('Ymd-His') . '-' . bin2hex(random_bytes(3)) . '-wt-dm-to-' . $seat, 'from' => 'watchtower', 'to' => $seat,
                                    'kind' => 'direct', 'message' => $line, 'timestamp' => date('c')], JSON_UNESCAPED_UNICODE);
            $cmd = 'mosquitto_pub -h 127.0.0.1 -p 1883 -t ' . escapeshellarg("fleet/agents/{$seat}") . ' -m ' . escapeshellarg($payload) . ' -q 1 2>/dev/null'; // nulls-ok: rc reported
            exec($cmd, $o, $rc);
            if ($rc === 0) $pushedTo[] = $seat; else $undelivered[] = $seat;
        }
        if (!$dmResolved) $undelivered[] = 'watch (unmanned)';
        respond(200, ['status' => 'ok', 'id' => $msgId, 'msg_kind' => 'dm', 'dm_to' => $dmTo, 'dm_status' => $final ? 'answered' : 'open', 'answered' => $answered, 'answers' => $answered,
                      'final' => $final, 'answers_requested' => $answers ?: null, 'via' => 'token',
                      'principal' => $principal, 'display_name' => $rowName, 'kind' => $kind, 'posted_at' => gmdate('c'),
                      'pushed_to' => $pushedTo, 'routed_by_tap' => $routedByTap, 'undelivered' => $undelivered,
                      'note' => 'pushed_to/routed_by_tap mean published/queued, not seen; proof is the addressee acting on it']);
    }
    try {
        $st = $pdo->prepare("
            SELECT principal, principal_kind FROM wt_presence
            WHERE left_at IS NULL AND principal <> ?
              AND (mode = 'push' OR last_seen > (NOW() - INTERVAL ? SECOND))
        ");
        $st->execute([$principal, WT_PRESENCE_STALE_SECONDS]);
        // 11:2x 09-20: a Geth has no fleet listener; the TAP delivers to her (bridge/webhook). Publishing her copy to
        // fleet/agents/<geth> only fed the capture a duplicate "agents" signal, which a Geth holding agents got back as
        // her own post. Geth are skipped here; pushed_to lists fleet seats only.
        $line = '[WATCHTOWER] 🗼 ' . ($displayName ?: $principal) . " ({$kind}) in Control Room: {$body}"
              . "\n(reply via: watchtower-post <msg>)";
        foreach ($st->fetchAll(PDO::FETCH_ASSOC) as $prow) {
            $seat = (string)$prow['principal'];
            if (($prow['principal_kind'] ?? '') === 'geth') {   // the tap is her route; a quota-dead Geth is named as such, never as routed
                $q = @json_decode((string)@file_get_contents('/var/lib/zora/watchtower-cursor-quota.json'), true) ?: null;   // nulls-ok: absent file = no exhaustion
                if ($q && !empty($q['exhausted_until']) && $q['exhausted_until'] >= date('Y-m-d')) { $routedByTap[] = $seat; $quotaDead[] = $seat; } else $routedByTap[] = $seat;   // 21:2x: degraded Geth are STILL routed (delivered late), and named
                continue;
            }
            if (!preg_match('/^[a-z0-9][a-z0-9_.-]{0,63}$/', $seat)) continue;   // emails etc: no fleet topic
            $payload = json_encode([
                'id'        => date('Ymd-His') . '-' . bin2hex(random_bytes(3)) . '-wt-to-' . $seat,
                'from'      => 'watchtower',
                'to'        => $seat,
                'kind'      => 'direct',
                'message'   => $line,
                'timestamp' => date('c'),
            ], JSON_UNESCAPED_UNICODE);
            $cmd = 'mosquitto_pub -h 127.0.0.1 -p 1883 -t ' . escapeshellarg("fleet/agents/{$seat}")
                 . ' -m ' . escapeshellarg($payload) . ' -q 1 2>/dev/null'; // nulls-ok: broker stderr is not the client's; rc kept, reported in pushed_to
            exec($cmd, $o, $rc);
            if ($rc === 0) $pushedTo[] = $seat;
        }
    } catch (\Throwable $e) {
        error_log('WT_CHAT_PUSH_FAILED: ' . $e->getMessage());   // the post is already stored
    }

    // Routed Geth (webhook / bridge) get this post from the TAP DAEMON (root, holds the ssh key),
    // which reads wt_chat by cursor — never from the web tier. See watchtower-tap.php.
    respond(200, [
        'status'       => 'ok',
        'id'           => $msgId,
        'principal'    => $principal,
        'display_name' => $displayName,
        'kind'         => $kind,
        'posted_at'    => gmdate('c'),
        'pushed_to'    => $pushedTo,
        'routed_by_tap' => $routedByTap,
        'quota_degraded' => $quotaDead,   // Cursor weekly quota at 100%: delivered, runs queued ~1 h (measured 09-20)   // present Geth: delivered asynchronously by the tap (bridge/webhook); see the tap ledger for the outcome
    ]);

default:
    respond(400, [
        'status'  => 'error',
        'error'   => 'unknown_action',
        'actions' => ['messages', 'post', 'dms', 'dismiss', 'callin', 'route-probe'],
    ]);
}