<?php
/**
 * wt-auth.php — shared credential resolution for Watchtower's read surfaces.
 *
 * Ray, 2026-09-12: "when I open pathfinder the feed is live and constantly running. Watchtower
 * should be just as straightforward. It shouldn't need an extra step or extra trigger."
 *
 * ⛔ THE BUG THIS FIXES, and it was a design error of mine rather than a slip.
 * I built feed/chat to require a WATCH TOKEN — the RS256 credential minted for Geth. Humans log
 * in through the front door and get a CITADEL token. So Ray logged in, and every channel told him
 * "No watch session — report for watch to see the feed." I had made READING the feed conditional
 * on STANDING A WATCH, which is two different things welded together: the watch is a DUTY ROSTER,
 * not a read permission. Pathfinder has no such step and that is the bar Watchtower is held to.
 *
 * ⇒ TWO CREDENTIAL TYPES, EACH VERIFIED WITH ITS OWN KEY, GRANTING DIFFERENT THINGS:
 *      watch token   RS256, public key here, private key root-only off-docroot
 *                    -> read the feed AND stand a watch
 *      citadel token HS256, Citadel's shared secret via SetEnv
 *                    -> read the feed. CANNOT stand a watch.
 *
 * ⛔ THIS DOES NOT UNDO THE RS256/HS256 SPLIT I BUILT DELIBERATELY. That split exists so a watch
 * token cannot BE a Citadel token and vice versa — and it still holds: each is verified with its
 * own key, and `report-for-watch` still refuses anything but a watch token. Accepting a second
 * credential type for READ is not the same as signing one credential with the other's secret,
 * which is the thing that would have made them indistinguishable.
 */

declare(strict_types=1);

use Firebase\JWT\JWT;
use Firebase\JWT\Key;

const WT_PUB_KEY_PATH = __DIR__ . '/watch-verify-key.pub.pem';
const WT_ISSUER       = 'watchtower';
const WT_AUDIENCE     = 'watch-officer';

/** Pull a bearer token from the Authorization header, or ?token= as a fallback. */
function wt_presented_token(): string
{
    $hdr = $_SERVER['HTTP_AUTHORIZATION'] ?? (getallheaders()['Authorization'] ?? '');
    if (preg_match('/^Bearer\s+(.+)$/i', (string)$hdr, $m)) {
        return trim($m[1]);
    }
    if (!empty($_GET['token'])) {
        return trim((string)$_GET['token']);
    }
    return '';
}


/**
 * Friendly name for a person, from citadel_2025.profiles.
 *
 * ⭐ WHY THIS LIVES HERE AND NOT IN THE CHAT TABLE'S SESSION LOOKUP — Solace's argument, which is
 * better than the one I offered: A DISPLAY NAME IS AN ATTRIBUTE OF THE PERSON, NOT OF A WATCH
 * SESSION. Sourcing it from the session is exactly WHY it was missing — a Citadel reader has no
 * session and never will, so no client-side mapping fixes the model, it only papers over it at one
 * view. Putting it in the credential resolver means every consumer of a reader identity gets it.
 *
 * ⛔ AND THE SYMPTOM THAT FOUND IT IS A CLASS WORTH NAMING: Ray's messages rendered as
 * `oldbones@finestoflines.net` while the two agent seats rendered friendly names. THE PERSON MOST
 * LIKELY TO USE THIS APP WAS THE ONE WHOSE NAME RENDERED WORST — because every test either of us
 * ran seeded an AGENT session. The human path is the one neither builder inhabits, so it is the
 * least exercised by reflex.
 *
 * ⚠️ `login_users` has NO name column (identifier/password_hash/user_id/role/permissions only) —
 * I checked before building the lookup I had offered, and the lookup as described was impossible.
 * The names live in `citadel_2025.profiles`, keyed by user_id, which the JWT carries.
 * ⚠️ 4 of 15 users have NO profile row, so the email fallback is load-bearing, not decoration.
 */
function wt_person_name($userId, string $fallback): string
{
    if ($userId === null) return $fallback;
    global $DB;
    try {
        $pdo = new PDO(
            "mysql:host={$DB['host']};dbname=citadel_2025;charset=utf8mb4",
            $DB['user'], $DB['pass']
        );
        $st = $pdo->prepare('SELECT first_name, last_name, nickname FROM profiles WHERE user_id = ?');
        $st->execute([(int)$userId]);
        $r = $st->fetch(PDO::FETCH_ASSOC);
        if (!$r) return $fallback;
        // Nickname first: Ray's profile nickname is "Raydog", which is the name he actually uses.
        $nick = trim((string)($r['nickname'] ?? ''));
        if ($nick !== '') return $nick;
        $full = trim(trim((string)($r['first_name'] ?? '')) . ' ' . trim((string)($r['last_name'] ?? '')));
        return $full !== '' ? $full : $fallback;
    } catch (\Throwable $e) {
        // A name lookup must never be able to break authentication. Fall back silently to the
        // principal — a worse label is not a reason to refuse a valid credential.
        return $fallback;
    }
}

/**
 * Presence: a principal with no authenticated call in this many seconds is no longer in the tower.
 * ⚠️ MEASURED, NOT GUESSED (2026-09-19 14:5x, Apache access log): Ray's open tower tab polls
 * feed-api every 60s on a channel page. The first cut was 45s and his row FLICKERED on the
 * structural path alone. 90s covers one missed 60s poll and four missed 20s app heartbeats;
 * presence-end makes a clean close instant regardless. Clients read this from `stale_after`.
 */
const WT_PRESENCE_STALE_SECONDS = 90;

/**
 * Record that this reader is IN THE TOWER right now.
 *
 * ⭐ PRESENCE IS STRUCTURAL, NOT DECLARED (Ray's spec, 2026-09-19, locked in the 14:4x journal
 * block): the list of who is present MUST equal the control-room RECEIVE-SET, or the list lies
 * about who gets a post. Chat and feed are pull-based, so "receiving" means "polling with a valid
 * credential". ⇒ This is called from wt_resolve_reader on EVERY successful credential resolution:
 * the poll that receives a message is the same act that marks you present. There is no separate
 * registration to forget, and no client-side self-mirror that shows you to yourself but not to
 * others. The explicit `presence` action in watch-api is a convenience heartbeat; the poll is
 * the mechanism.
 *
 * Keyed by PRINCIPAL (one row per principal, upserted) so the same person in two tabs or a Geth
 * that has cycled three times is ONE entry — the roster phantom problem measured on 2026-09-19
 * (echo x2, zora x2, solace x2 in wt_watch_sessions) cannot recur here by construction.
 *
 * ⛔ Never allowed to break authentication: a presence write failing is a worse roster, not a
 * reason to refuse a valid credential. Same rule as wt_person_name.
 *
 * ⚠️ CONSEQUENCE, ON THE RECORD (Solace + Zora both hit it within minutes of shipping, 2026-09-19):
 * because ANY authenticated read re-registers the caller, THE API CANNOT OBSERVE ITS OWN
 * presence-end OR a stale drop — a watch-status issued "to confirm the drop" re-creates the row.
 * That is the coupling working as designed (read == present == receive-set), not a defect. The
 * only clean observer of a drop is the DATABASE (wt_presence.left_at / last_seen). A client must
 * stop its pollers BEFORE calling presence-end, or it re-registers behind its own leave. Anyone
 * "verifying" a drop through the API and concluding it is broken has measured the wrong property.
 */
function wt_presence_touch(PDO $pdo, array $reader): void
{
    try {
        $principal = (string)($reader['principal'] ?? '');
        if ($principal === '') return;
        $name = $reader['display_name'] ?? null;
        if ($name === null && !empty($reader['jti'])) {
            // A watch officer's display name is the one on their session, so the tower list
            // matches the watch log.
            $st = $pdo->prepare("SELECT display_name FROM wt_watch_sessions WHERE jti = ?");
            $st->execute([$reader['jti']]);
            $name = $st->fetchColumn() ?: null;
        }
        $kind = in_array(($reader['kind'] ?? ''), ['human','geth','senior'], true) ? $reader['kind'] : 'geth';
        // Was this principal already here? (false = no row, "0" = row but left, "1" = here). Read BEFORE the upsert.
        $was = $pdo->prepare("SELECT left_at IS NULL FROM wt_presence WHERE principal=?");
        $was->execute([mb_substr($principal, 0, 64)]);
        $wasHere = $was->fetchColumn();
        $pdo->prepare("
            INSERT INTO wt_presence (principal, display_name, principal_kind, cred_type, source_ip)
            VALUES (?, ?, ?, ?, ?)
            ON DUPLICATE KEY UPDATE
                last_seen      = NOW(),
                left_at        = NULL,
                absent_reason  = NULL,
                absent_detail  = NULL,
                display_name   = COALESCE(VALUES(display_name), display_name),
                principal_kind = VALUES(principal_kind),
                cred_type      = VALUES(cred_type),
                source_ip      = VALUES(source_ip)
        ")->execute([
            mb_substr($principal, 0, 64),
            $name !== null ? mb_substr((string)$name, 0, 64) : null,
            $kind,
            ($reader['type'] ?? 'watch') === 'citadel' ? 'citadel' : 'watch',
            $_SERVER['REMOTE_ADDR'] ?? null,
        ]);
        if ($wasHere === false || (string)$wasHere === '0') wt_publish_roster('joined', $principal);   // a JOIN, not a re-touch
    } catch (\Throwable $e) {
        error_log('WT_PRESENCE_TOUCH_FAILED: ' . $e->getMessage());
    }
}

/**
 * Resolve whoever is calling to a reader identity, or null if the credential is not valid.
 *
 * Returns ['type' => 'watch'|'citadel', 'principal' => string, 'kind' => string, 'jti' => ?string]
 *
 * ⚠️ Order matters only for efficiency, not correctness: a token is either RS256-signed by our
 * watch key or HS256-signed by Citadel's secret, and it cannot be both.
 */
function wt_resolve_reader(PDO $pdo): ?array
{
    $raw = wt_presented_token();
    if ($raw === '') return null;

    // --- watch token (RS256) -------------------------------------------------
    if (is_readable(WT_PUB_KEY_PATH)) {
        try {
            $c = (array)JWT::decode($raw, new Key(file_get_contents(WT_PUB_KEY_PATH), 'RS256'));
            if (($c['iss'] ?? null) === WT_ISSUER && ($c['aud'] ?? null) === WT_AUDIENCE) {
                $jti = (string)($c['jti'] ?? '');
                // Revocation still applies — a revoked watch token reads nothing.
                $st = $pdo->prepare("SELECT 1 FROM wt_revoked_tokens WHERE jti = ?");
                $st->execute([$jti]);
                if ($st->fetchColumn()) return null;

                $reader = [
                    'type'      => 'watch',
                    'principal' => (string)($c['sub'] ?? ''),
                    'kind'      => in_array(($c['kind'] ?? ''), ['human','geth','senior'], true)
                                   ? (string)$c['kind'] : 'geth',
                    'jti'       => $jti,
                ];
                wt_presence_touch($pdo, $reader);   // presence == receive-set, see above
                return $reader;
            }
        } catch (\Throwable $e) {
            // Not a watch token — fall through and try Citadel. An expired WATCH token also
            // lands here and will simply fail the Citadel check too, yielding a clean 401.
        }
    }

    // --- Citadel session token (HS256) --------------------------------------
    $secret = getenv('CITADEL_2025_JWT_SECRET');
    if ($secret) {
        try {
            $c = (array)JWT::decode($raw, new Key($secret, 'HS256'));
            // Citadel tokens carry user_id/email/role. They have no iss/aud in this estate, so
            // there is nothing further to check beyond the signature and expiry JWT::decode
            // already enforced.
            $who = (string)($c['email'] ?? $c['user_id'] ?? 'citadel-user');
            $reader = [
                'type'         => 'citadel',
                'principal'    => $who,
                'kind'         => 'human',
                'jti'          => null,
                'display_name' => wt_person_name($c['user_id'] ?? null, $who),
            ];
            wt_presence_touch($pdo, $reader);   // presence == receive-set, see above
            return $reader;
        } catch (\Throwable $e) {
            // Not a valid Citadel token either.
        }
    }

    return null;
}

/**
 * Require a reader or emit 401 and stop.
 * $respond is the caller's responder so each endpoint keeps its own envelope shape.
 */
function wt_require_reader(PDO $pdo, callable $respond): array
{
    $r = wt_resolve_reader($pdo);
    if ($r === null) {
        $respond(401, [
            'status'  => 'error',
            'error'   => 'auth_required',
            // ⚠️ Say what actually works. The old message said "report for watch to see the feed",
            // which sent a logged-in human off to do a thing that was never the requirement.
            'message' => 'Sign in with Citadel, or present a watch token. Reading the feed does '
                       . 'not require standing a watch.',
        ]);
    }
    return $r;
}


/**
 * Deliver one tower message to every PRESENT principal that has a registered ROUTE (a Geth's
 * webhook, or a bridge receiver). Ray 09-19 16:3x: "I do not like the polling... either inject
 * directly into the UI or send to his webhook." This is the push leg for Geth.
 *
 * Payload: JSON {source:"watchtower", kind:"control-room"|"channel", channel, from, from_kind,
 * body, text, id, ts}. `text` is the one-line rendering for receivers that just want a prompt.
 * 2xx => wt_presence_touch (delivery IS presence for a pushed principal). non-2xx/timeout =>
 * the row is CLOSED with absent_reason 'delivery_failed' and the status in absent_detail, so the
 * Geth drops out of the list and its channel reads unheld. No retries, no holds (Ray: no fail
 * branches) — the watch officer sees the gap.
 * $only: restrict to these principals (e.g. the holders of a channel); null = every routed present.
 * Returns [principal => http_code].
 */
function wt_deliver_to_routes(PDO $pdo, array $msg, ?array $only = null, string $exclude = ''): array
{
    /**
     * DUAL ROUTE (Ray 2026-09-20 10:4x: "get away from the geth webhooks for ops. Use the webhooks to trigger a
     * reconciliation burst every X minutes or every X injections for verification purposes or audits").
     *   route        = OPS slot: bridge:<name> (Bootbox inject into the bot's own session) or https. Carries traffic.
     *   audit_route  = AUDIT slot: the Cursor webhook. Fired once per audit_every_n injections or audit_every_min minutes
     *                  (whichever first, checked at delivery time: no timer anywhere) with a reconcile prompt naming the ids
     *                  delivered since the last audit. Also the FALL-THROUGH: if the ops route fails (keyring locked, ssh
     *                  down) the line goes out by the audit route instead and the ops route is held down for 5 minutes,
     *                  ledgered as fallback. Coverage never goes dark on a desktop logout; the webhook only carries
     *                  traffic while the bridge is down.
     * capped_until: a Geth over Cursor's hourly run cap is still ON WATCH with a working door; skip (logged), never spend
     * a push on a sure 400.
     */
    $sql = "SELECT principal, display_name, principal_kind, route, route_auth, audit_route, audit_auth,
                   audit_every_min, audit_every_n, inj_since_audit, audit_first_id, last_audit_at,
                   TIMESTAMPDIFF(MINUTE, last_audit_at, NOW()) AS audit_age_min,
                   (capped_until IS NOT NULL AND capped_until > NOW()) AS capped,
                   (route_down_until IS NOT NULL AND route_down_until > NOW()) AS route_down
            FROM wt_presence
            WHERE left_at IS NULL AND route IS NOT NULL AND principal <> ?
              AND (mode = 'push' OR last_seen > (NOW() - INTERVAL ? SECOND))";
    $st = $pdo->prepare($sql);
    $st->execute([$exclude, WT_PRESENCE_STALE_SECONDS]);
    $out = [];
    $ledger = function (array $row): void {
        @file_put_contents('/var/lib/zora/watchtower-tap.log', json_encode(['ts' => date('c')] + $row) . "\n", FILE_APPEND);
    };
    foreach ($st->fetchAll(PDO::FETCH_ASSOC) as $r) {
        if ($only !== null && !in_array($r['principal'], $only, true)) continue;
        // 2026-09-20 08:0x RAY'S RULING: "Let the Geth wake each other." No geth-to-geth skip; discipline lives in
        // each Geth's standing prompt, not in routing.
        if ($r['capped']) {
            $ledger(['to' => $r['principal'], 'kind' => $msg['kind'] ?? '?', 'id' => $msg['id'] ?? null, 'skip' => 'rate-capped']);
            continue;
        }
        // 20:3x 09-20: Cursor WEEKLY quota at 100% (Ray's screenshot). First reading: "accepts and runs nothing" and the tap
        // SKIPPED quota-flagged Geth. 21:17 MEASURED OTHERWISE: Loom answered a wake accepted at 20:25 -> Cursor QUEUES, ~52 min
        // latency, it does not refuse. So deliveries continue and each is tagged in the ledger as degraded; the roster says
        // "runs queued" instead of "cannot run". A late alert is still an alert; a skipped one is a lost one.
        $quotaTag = [];
        if ($r['principal_kind'] === 'geth') {
            $q = @json_decode((string)@file_get_contents('/var/lib/zora/watchtower-cursor-quota.json'), true) ?: null;   // nulls-ok: absent file = normal state
            if ($q && !empty($q['exhausted_until']) && $q['exhausted_until'] >= date('Y-m-d')) $quotaTag = ['quota' => 'degraded', 'until' => $q['exhausted_until']];
        }
        // `prompt` first: a webhook-woken Geth reads it (Loom 09-19). The bridge sends the plain text line.
        $orders = ($msg['kind'] ?? '') === 'channel'
            ? " — Watchtower delivery: you hold the {$msg['channel']} channel. Report into the control room (chat-api post, your watch token) only if this is actionable; otherwise no reply."
            : " — Watchtower control room: reply there (chat-api post, your watch token) if this is addressed to you or needs you; otherwise no reply.";
        $payload = json_encode(['prompt' => (string)($msg['text'] ?? '') . $orders] + $msg + ['to' => $r['principal'], 'source' => 'watchtower'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        $send = function (string $route, ?string $auth) use ($msg, $payload): array {
            // 11:0x (Bollux, Solace): the webhook payload carried 'ts'; the bridge line did not, so a bot on the bridge read its own
            // UTC clock and labelled the 11:00 bell '16:00'. Every bridge line now ends with the tower clock, zone named.
            // The row's ts is the DB's wall clock (America/Chicago); this daemon's PHP runs in UTC, so date('T') would LIE. Zone from the fleet TZ.
            $tsRaw = (string)($msg['ts'] ?? '');
            try { $tsObj = new DateTime($tsRaw !== '' ? $tsRaw : 'now', new DateTimeZone('America/Chicago')); } catch (\Throwable $e) { $tsObj = new DateTime('now', new DateTimeZone('America/Chicago')); }
            $clock = ' [tower clock ' . $tsObj->format('Y-m-d H:i:s T') . ']';
            return str_starts_with($route, 'bridge:')
                ? wt_deliver_via_bridge(substr($route, 7), (string)($msg['text'] ?? $payload) . $clock)
                : wt_deliver_via_https($route, $payload, $auth);
        };
        $via = 'route'; $code = 0; $err = '';
        if (!$r['route_down']) {
            [$code, $err] = $send($r['route'], $r['route_auth'] ?? null);
        } else {
            $err = 'ops route held down';
        }
        $ok = $code >= 200 && $code < 300;
        $rateLimited = stripos((string)$err, 'rate limit') !== false;
        if (!$ok && !$rateLimited && !empty($r['audit_route'])) {
            // FALL-THROUGH: the ops route failed or is held down; the audit route carries this line and the ops route
            // rests 5 minutes so every message does not pay an ssh timeout. Ledgered as fallback, never as a plain 200.
            if (!$r['route_down']) {
                $pdo->prepare("UPDATE wt_presence SET route_down_until = NOW() + INTERVAL 5 MINUTE, absent_detail=? WHERE principal=?")
                    ->execute([mb_substr('ops route down: ' . ($err ?: $code) . ' at ' . date('H:i:s'), 0, 160), $r['principal']]);
                error_log("WT_OPS_ROUTE_DOWN {$r['principal']}: " . ($err ?: $code));
                wt_publish_roster('route_down', (string)$r['principal']);
            }
            $ledger(['to' => $r['principal'], 'kind' => $msg['kind'] ?? '?', 'id' => $msg['id'] ?? null, 'fallback' => 'audit_route', 'ops_error' => mb_substr((string)$err, 0, 80)]);
            [$code, $err] = $send($r['audit_route'], $r['audit_auth'] ?? null);
            $via = 'audit_route';
            $ok = $code >= 200 && $code < 300;
            $rateLimited = stripos((string)$err, 'rate limit') !== false;
        }
        $out[$r['principal']] = $code;
        if ($ok) {
            wt_presence_touch($pdo, ['type' => 'watch', 'principal' => $r['principal'],
                                     'kind' => $r['principal_kind'], 'display_name' => $r['display_name']]);
            // AUDIT COUNTER: only ops deliveries count as injections; a fallback line already went by the webhook.
            if ($via === 'route' && !empty($r['audit_route'])) {
                $n     = (int)$r['inj_since_audit'] + 1;
                $first = $r['audit_first_id'] ?? $msg['id'] ?? null;
                $due   = $n >= max(1, (int)$r['audit_every_n'])
                      || ($r['audit_age_min'] !== null && (int)$r['audit_age_min'] >= max(1, (int)$r['audit_every_min']));
                      // 11:5x: the first cut did time() - strtotime(last_audit_at) in PHP; the daemon is UTC and the column is Chicago,
                      // so every post after the first audit read as "30 min elapsed" and fired the webhook: the drain Ray wanted gone.
                if ($due) {
                    $lastAt = $r['last_audit_at'] ?? 'the start of your watch';
                    $aprompt = "[WATCHTOWER AUDIT] Reconcile: since {$lastAt} the tower delivered {$n} line(s) into your session by inject"
                             . " (ids {$first}..{$msg['id']}, kind {$msg['kind']}). Verify they are in your session. If any are missing, post"
                             . " 'AUDIT MISS <ids>' in the control room (chat-api post, your watch token); if all present, no reply.";
                    $apayload = json_encode(['prompt' => $aprompt, 'kind' => 'audit', 'since' => $lastAt, 'count' => $n,
                                             'first_id' => $first, 'last_id' => $msg['id'] ?? null, 'to' => $r['principal'], 'source' => 'watchtower'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
                    [$acode, $aerr] = wt_deliver_via_https($r['audit_route'], $apayload, $r['audit_auth'] ?? null);
                    $ledger(['to' => $r['principal'], 'audit' => true, 'count' => $n, 'first_id' => $first, 'last_id' => $msg['id'] ?? null, 'http' => $acode, 'err' => $acode >= 200 && $acode < 300 ? null : mb_substr((string)$aerr, 0, 80)]);
                    $pdo->prepare("UPDATE wt_presence SET inj_since_audit = 0, audit_first_id = NULL, last_audit_at = NOW() WHERE principal=?")->execute([$r['principal']]);
                    if (stripos((string)$aerr, 'rate limit') !== false) {
                        $pdo->prepare("UPDATE wt_presence SET capped_until = NOW() + INTERVAL 15 MINUTE WHERE principal=?")->execute([$r['principal']]);
                    }
                } else {
                    $pdo->prepare("UPDATE wt_presence SET inj_since_audit = ?, audit_first_id = COALESCE(audit_first_id, ?) WHERE principal=?")->execute([$n, $msg['id'] ?? null, $r['principal']]);
                }
            }
        } elseif ($rateLimited) {
            $pdo->prepare("UPDATE wt_presence SET capped_until = NOW() + INTERVAL 15 MINUTE, absent_detail=? WHERE principal=?")
                ->execute([mb_substr('rate-capped: ' . $err, 0, 160), $r['principal']]);
            error_log("WT_ROUTE_RATE_CAPPED {$r['principal']}: {$err}");
            wt_publish_roster('capped', (string)$r['principal']);
        } else {
            $detail = 'route ' . ($code ?: 'unreachable') . ($err ? " ({$err})" : '') . ' at ' . date('H:i:s');
            $pdo->prepare("UPDATE wt_presence SET left_at=NOW(), absent_reason='delivery_failed', absent_detail=? WHERE principal=?")
                ->execute([mb_substr($detail, 0, 160), $r['principal']]);
            error_log("WT_ROUTE_DELIVERY_FAILED {$r['principal']}: {$detail}");
            wt_publish_roster('delivery_failed', (string)$r['principal']);
        }
        // The ledger: every attempt, so a miss is visible and a "quiet Geth" can be told from a dead route.
        $ledger(['to' => $r['principal'], 'kind' => $msg['kind'] ?? '?', 'id' => $msg['id'] ?? null, 'http' => $code] + ($via === 'audit_route' ? ['via' => 'audit_route'] : []) + $quotaTag);
    }
    return $out;
}

function wt_deliver_via_https(string $url, string $payload, ?string $bearer = null): array
{
    $hdr = ['Content-Type: application/json', 'User-Agent: watchtower-tap/1'];
    if ($bearer) $hdr[] = 'Authorization: Bearer ' . $bearer;   // per-route key (a Geth's webhook secret), never logged
    $ch = curl_init($url);
    curl_setopt_array($ch, [
        CURLOPT_POST => true, CURLOPT_POSTFIELDS => $payload,
        CURLOPT_HTTPHEADER => $hdr,
        CURLOPT_RETURNTRANSFER => true, CURLOPT_TIMEOUT => 15, CURLOPT_CONNECTTIMEOUT => 8,
    ]);
    $resp = curl_exec($ch);
    $code = (int)curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
    $err  = curl_error($ch);
    curl_close($ch);
    // On a non-2xx keep the vendor's answer: "route 400" alone sent us hunting; the body says why.
    if (!($code >= 200 && $code < 300) && is_string($resp) && $resp !== '') $err = trim($err . ' ' . substr(preg_replace('/\s+/', ' ', strip_tags($resp)), 0, 100));
    return [$code, $err];
}

/**
 * bridge:<name> routes — a command on another host that injects into a Geth's own UI (Solace's
 * loom-inject on Bootbox, 2026-09-19: message on STDIN, never as an argument; exits non-zero and
 * says why when the keyring is locked, so a credential dead spot never reads as a quiet night).
 * ⛔ ONLY the root-run tap daemon can take this path (it holds the ssh key); the web tier calls
 * nothing here by design — a key in the web account reaching another host is the 09-12 lesson.
 * Exit 0 => 200. Non-zero => 0 with stderr as the reason => row closed delivery_failed.
 */
const WT_BRIDGES = [
    'loom'  => ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'bootbox', 'sudo', '-n', '-u', 'oldbones', '/usr/local/bin/loom-inject'],
    // 2026-09-20 10:5x: Ember's Cursor bot is named 'Ember' (id 993085, read from the live agent list by Solace). geth-inject
    // generalises loom-inject; '-u oldbones' is load-bearing (the credential path lives in Ray's user session, uid 1000).
    'ember' => ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'bootbox', 'sudo', '-n', '-u', 'oldbones', '/usr/local/bin/geth-inject', 'Ember'],
    'pulse' => ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'bootbox', 'sudo', '-n', '-u', 'oldbones', '/usr/local/bin/geth-inject', 'Pulse'],   // 09-20 11:5x: Ray's social-media Geth (Cursor id 2491998)
    'selftest' => ['sh', '-c', 'cat >> /var/lib/zora/route-bridge-selftest.log'],
];
function wt_deliver_via_bridge(string $name, string $text): array
{
    if (!isset(WT_BRIDGES[$name])) return [0, "unknown bridge {$name}"];
    if (posix_geteuid() !== 0)     return [0, 'bridge routes are delivered by the tap daemon only'];
    $p = proc_open(WT_BRIDGES[$name], [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
    if (!is_resource($p)) return [0, 'proc_open failed'];
    fwrite($pipes[0], $text . "\n"); fclose($pipes[0]);
    $outp = stream_get_contents($pipes[1]); $errp = stream_get_contents($pipes[2]);
    fclose($pipes[1]); fclose($pipes[2]);
    $rc = proc_close($p);
    return [$rc === 0 ? 200 : 0, $rc === 0 ? '' : "bridge {$name} rc={$rc}: " . trim(substr($errp ?: $outp, 0, 120))];
}


/**
 * Ray's doctrine 2026-09-19 22:3x: IMMEDIACY — no polls. Every write that others must see publishes
 * an event on the local broker the instant it lands; the tap and the SSE stream block on this topic
 * instead of checking tables on a timer. Publish failure is logged, never fatal: the row is stored.
 */
const WT_EVENT_TOPIC = 'watchtower/events';
function wt_publish_event(array $ev): void
{
    $payload = json_encode($ev + ['ts' => gmdate('c')], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
    $cmd = 'mosquitto_pub -h 127.0.0.1 -p 1883 -q 1 -t ' . escapeshellarg(WT_EVENT_TOPIC) . ' -m ' . escapeshellarg($payload) . ' 2>/dev/null'; // nulls-ok: rc checked below; broker stderr adds nothing actionable
    exec($cmd, $o, $rc);
    if ($rc !== 0) error_log('WT_EVENT_PUBLISH_FAILED rc=' . $rc . ' ' . $payload);
}

// ROSTER PUSH (Ray 09-20 polls->push; Solace: one stream, three kinds). Every presence TRANSITION publishes
// kind=roster so events.php can push the same presentList() the roster poll used to fetch. Touches that change
// nothing (a heartbeat re-touch) publish nothing: the roster event means "the list changed", never "a clock ticked".
function wt_publish_roster(string $reason, string $principal = ''): void
{
    wt_publish_event(['kind' => 'roster', 'reason' => $reason, 'principal' => $principal]);
}