fix(broker): self-heal route registration

README.md

@@ -465,6 +465,12 @@ pkill -f 'extensions/relay/broker/process.js'
 
 Then restart Pi or reload the package from the updated checkout.
 
+### A running Pi session appears offline in `/sessions`
+
+PiRelay tracks binding authority, broker transport, and broker route synchronization separately. A live Pi process and active binding do not by themselves prove that remote routing is online. Broker clients reconcile their complete live route map after connection and periodically, so a missing in-memory route should repair automatically within about 30 seconds. Local `/relay status` reports `Broker transport` and `Broker route` separately; `/relay doctor` can request the same safe idempotent repair for a proven live route. If recovery remains pending, temporarily enable communication diagnostics and inspect metadata-only `broker.connect`, `broker.reconcile`, and `broker.retry` events without sharing tokens, destinations, socket paths, or route payloads.
+
+`/relay restart` replaces the broker shared by all same-scope local Pi sessions. Clients reconnect autonomously, and the command reports recovered client/route counts. A partial timeout leaves recovered sessions available while missing clients continue bounded retries; run `/relay doctor` in a missing live session rather than re-pairing it.
+
 ### `/sessions` shows stale or offline duplicates
 
 When a newer online session exists for the same machine/workspace, PiRelay hides older offline same-workspace pairings from the default session list. Use `/sessions all` (or `relay sessions all` on Discord/Slack) to reveal hidden stale entries for diagnostics and cleanup, then run the appropriate forget command for your messenger to remove an offline pairing. PiRelay does not delete transcript files or automatically revoke live bindings when it hides superseded entries.

docs/testing.md

@@ -174,6 +174,17 @@ When a test fails, record:
 - whether the failure affects only local Pi input, only Telegram behavior, or both
 - the exact final assistant output if answer parsing behaved incorrectly
 
+## 9. Broker route recovery and shared restart
+
+1. Start three same-scope Pi sessions and verify all appear online in `/sessions`.
+2. Remove one route from broker memory without closing its client socket in the automated real-socket harness; verify periodic reconciliation restores it without `/reload`, `/relay status`, or another command.
+3. Run `/relay restart` from one session and verify the output explains shared impact and reports all three clients/routes recovered.
+4. Verify the pid/socket control files identify one broker, delayed clients can join later, and no old broker retains messenger ingress.
+5. While transport is reconnecting or a route is pending/missing, verify `/relay status`, `/relay doctor`, and the concise Telegram status segment do not imply full remote reachability.
+6. Revoke a binding during reconciliation and verify it is not resurrected, retargeted, or unpaused.
+
+Automated lifecycle tests must use public runtime behavior and real socket close/recreation. Do not invoke private `ensureConnected()` or rely on `/reload` as the recovery mechanism.
+
 ## 9. Optional communication diagnostics troubleshooting
 
 When investigating missing final assistant responses or broker delivery issues, enable `communicationDiagnostics.enabled` only for the reproduction, then inspect `logs/communication.jsonl` locally. Prefer sharing only redacted `agent_end.final_extraction` metadata. See `docs/communication-diagnostics.md`.

extensions/relay/broker/process.js

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import { unlink, readFile, writeFile, mkdir } from 'node:fs/promises';
 import net from 'node:net';
 import lockfile from 'proper-lockfile';
@@ -24,6 +25,7 @@ const [
   telegramRouteBindingModule,
   approvalGatesModule,
   communicationDiagnosticsModule,
+  brokerReconciliationModule,
   skillInvocationModule,
 ] = await Promise.all([
   jiti.import('../core/guided-answer.ts'),
@@ -43,6 +45,7 @@ const [
   jiti.import('./telegram-route-binding.ts'),
   jiti.import('../core/approval-gates.ts'),
   jiti.import('../diagnostics/communication.ts'),
+  jiti.import('./reconciliation.ts'),
   jiti.import('../core/skill-invocation.ts'),
 ]);
 
@@ -144,9 +147,11 @@ const isPendingSkillInputExpired = requiredFunction(skillInvocationModule, './sk
 const pendingSkillInputKey = requiredFunction(skillInvocationModule, './skill-invocation.ts', 'pendingSkillInputKey');
 const resolveRemoteSkill = requiredFunction(skillInvocationModule, './skill-invocation.ts', 'resolveRemoteSkill');
 const skillConfigForRelay = requiredFunction(skillInvocationModule, './skill-invocation.ts', 'skillConfigForRelay');
+const normalizeBrokerReconciliationRequest = requiredFunction(brokerReconciliationModule, './reconciliation.ts', 'normalizeBrokerReconciliationRequest');
 
 const socketPath = process.env.TELEGRAM_TUNNEL_BROKER_SOCKET_PATH;
 const pidPath = process.env.TELEGRAM_TUNNEL_BROKER_PID_PATH;
+const epochPath = process.env.PI_RELAY_BROKER_EPOCH_PATH;
 const config = JSON.parse(process.env.TELEGRAM_TUNNEL_BROKER_CONFIG_JSON || '{}');
 const diagnosticsConfig = JSON.parse(process.env.PI_RELAY_COMMUNICATION_DIAGNOSTICS_CONFIG_JSON || JSON.stringify(config.communicationDiagnostics || { enabled: false }));
 const skipPolling = process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING === '1';
@@ -171,7 +176,10 @@ let hasRegisteredTelegramCommands = false;
 let hasAttemptedTelegramBotCommandRegistration = false;
 const api = new Api(config.botToken);
 const clients = new Map();
+const socketGenerations = new Map();
+let nextClientGeneration = 0;
 const routes = new Map();
+const brokerEpoch = randomUUID();
 const pendingClientRequests = new Map();
 const activeSessionByChatId = new Map();
 const answerFlows = new Map();
@@ -729,12 +737,12 @@ async function activeBindingForRoute(route, options = {}) {
   return authorityOutcomeAllowsDelivery(outcome) ? outcome.binding : undefined;
 }
 
-async function stripRevokedBindingFromRoute(route) {
+async function stripRevokedBindingFromRoute(route, state = undefined) {
   if (!route?.binding) return route;
-  const snapshot = await loadStateSnapshot();
+  const snapshot = state ? bindingAuthorityStateFromData(state) : await loadStateSnapshot();
   const outcome = resolveTelegramBindingAuthority(
     snapshot,
-    { sessionKey: route.sessionKey, chatId: route.binding.chatId, userId: route.binding.userId, includePaused: true, allowVolatileFallback: true },
+    { sessionKey: route.sessionKey, chatId: route.binding.chatId, userId: route.binding.userId, includePaused: true, allowVolatileFallback: state === undefined },
     route.binding,
   );
   return authorityOutcomeAllowsDelivery(outcome) ? { ...route, binding: outcome.binding } : { ...route, binding: undefined };
@@ -2626,6 +2634,7 @@ async function pollLoop() {
 
 function removeClient(socket) {
   const client = clients.get(socket);
+  socketGenerations.delete(socket);
   if (!client) return;
   for (const sessionKey of client.routes) {
     const existing = routes.get(sessionKey);
@@ -2657,18 +2666,126 @@ async function handleClientRequest(socket, message) {
         respond(true, setup);
         return;
       }
+      case 'getBrokerHealth': {
+        recordDiagnostic({ component: 'broker', event: 'broker.health', outcome: 'reported', details: { brokerEpoch, connectedClientCount: [...clients.values()].filter((client) => client.routes.size > 0).length, registeredRouteCount: routes.size } });
+        respond(true, { brokerEpoch, connectedClientCount: [...clients.values()].filter((client) => client.routes.size > 0).length, registeredRouteCount: routes.size });
+        return;
+      }
+      case 'prepareBrokerRestart': {
+        const expectedClientCount = [...clients.values()].filter((client) => client.routes.size > 0).length;
+        const expectedRouteCount = routes.size;
+        recordDiagnostic({ component: 'broker', event: 'broker.restart.prepare', outcome: 'notified', details: { brokerEpoch, expectedClientCount, expectedRouteCount } });
+        for (const [clientSocket, client] of clients) {
+          if (clientSocket === socket || client.routes.size === 0 || clientSocket.destroyed) continue;
+          write(clientSocket, {
+            type: 'request',
+            requestId: `restart-${randomUUID()}`,
+            protocolVersion: BROKER_PROTOCOL_VERSION,
+            channel: 'telegram',
+            action: 'brokerRestartNotice',
+          });
+        }
+        respond(true, { brokerEpoch, expectedClientCount, expectedRouteCount });
+        return;
+      }
+      case 'reconcileRoutes': {
+        const validation = normalizeBrokerReconciliationRequest({
+          clientId: message.clientId,
+          observedBrokerEpoch: message.observedBrokerEpoch,
+          routes: message.routes,
+        }, brokerEpoch);
+        if (!validation.ok) {
+          recordDiagnostic({ component: 'broker', event: 'route.reconcile', outcome: 'rejected', details: { brokerEpoch, code: validation.rejection.code } });
+          respond(true, { brokerEpoch, acceptedSessionKeys: [], rejected: [validation.rejection] });
+          return;
+        }
+
+        const authoritySnapshot = await loadStateSnapshot();
+        if (authoritySnapshot.kind === 'state-unavailable') {
+          const rejection = { code: 'invalid-request', safeMessage: 'Broker binding authority is temporarily unavailable.' };
+          recordDiagnostic({ component: 'broker', event: 'route.reconcile', outcome: 'rejected', details: { brokerEpoch, code: rejection.code, category: 'state-unavailable' } });
+          respond(true, { brokerEpoch, acceptedSessionKeys: [], rejected: [rejection] });
+          return;
+        }
+        const state = authoritySnapshot.data;
+        const preparedRoutes = [];
+        for (const descriptor of validation.request.routes) {
+          preparedRoutes.push(await stripRevokedBindingFromRoute(routeWithPersistedTelegramBinding(descriptor, state), state));
+        }
+
+        if (!clients.has(socket)) clients.set(socket, { clientId: validation.request.clientId, routes: new Set(), generation: socketGenerations.get(socket) });
+        const client = clients.get(socket);
+        const previouslyOwned = new Set(client.routes);
+        const acceptedRoutes = [];
+        const rejected = [];
+        for (let index = 0; index < preparedRoutes.length; index += 1) {
+          const route = preparedRoutes[index];
+          const previousRoute = routes.get(route.sessionKey);
+          if (previousRoute?.socket !== socket && previousRoute?.ownerGeneration > client.generation) {
+            rejected.push({ index, code: 'stale-owner', safeMessage: 'A newer broker client owns this route.' });
+            continue;
+          }
+          acceptedRoutes.push(route);
+        }
+        const acceptedSessionKeys = acceptedRoutes.map((route) => route.sessionKey);
+        const accepted = new Set(acceptedSessionKeys);
+
+        for (const route of acceptedRoutes) {
+          const previousRoute = routes.get(route.sessionKey);
+          if (previousRoute?.socket && previousRoute.socket !== socket) clients.get(previousRoute.socket)?.routes.delete(route.sessionKey);
+          const nextRoute = { ...route, socket, ownerGeneration: client.generation };
+          if (previousRoute?.binding?.chatId !== nextRoute.binding?.chatId && previousRoute?.binding) {
+            clearActivityIndicator(previousRoute);
+            clearProgressState(previousRoute);
+          }
+          routes.set(route.sessionKey, nextRoute);
+          if (previousRoute && getCurrentTurnId(previousRoute) !== getCurrentTurnId(nextRoute)) {
+            clearAnswerStateForRoute(previousRoute);
+          } else if (!nextRoute.notification?.structuredAnswer && previousRoute) {
+            clearAnswerFlow(previousRoute);
+          }
+          clearStaleCustomAnswers(nextRoute);
+          client.routes.add(route.sessionKey);
+        }
+        for (const sessionKey of previouslyOwned) {
+          if (accepted.has(sessionKey)) continue;
+          client.routes.delete(sessionKey);
+          const existing = routes.get(sessionKey);
+          if (existing?.socket !== socket) continue;
+          clearAnswerStateForRoute(existing);
+          clearActivityIndicator(existing);
+          clearProgressState(existing);
+          routes.delete(sessionKey);
+        }
+        client.clientId = validation.request.clientId;
+
+        for (const route of acceptedRoutes) {
+          const nextRoute = routes.get(route.sessionKey);
+          if (!nextRoute) continue;
+          syncActivityIndicator(nextRoute);
+          syncProgressDelivery(nextRoute);
+          if (nextRoute.binding) await upsertBinding(nextRoute.binding);
+        }
+        recordDiagnostic({ component: 'broker', event: 'route.reconcile', outcome: rejected.length > 0 ? acceptedSessionKeys.length > 0 ? 'partial' : 'rejected' : 'ok', details: { clientId: validation.request.clientId, routeCount: acceptedSessionKeys.length, rejectedCount: rejected.length, brokerEpoch } });
+        respond(true, { brokerEpoch, acceptedSessionKeys, rejected });
+        return;
+      }
       case 'registerRoute': {
         const state = await loadState();
         const route = await stripRevokedBindingFromRoute(routeWithPersistedTelegramBinding(message.route, state));
-        if (!clients.has(socket)) clients.set(socket, { clientId: message.clientId, routes: new Set() });
+        if (!clients.has(socket)) clients.set(socket, { clientId: message.clientId, routes: new Set(), generation: socketGenerations.get(socket) });
         const client = clients.get(socket);
         client.clientId = message.clientId;
         client.routes.add(route.sessionKey);
         const previousRoute = routes.get(route.sessionKey);
+        if (previousRoute?.socket !== socket && previousRoute?.ownerGeneration > client.generation) {
+          respond(false, undefined, 'A newer broker client owns this route.');
+          return;
+        }
         if (previousRoute?.socket && previousRoute.socket !== socket) {
           clients.get(previousRoute.socket)?.routes.delete(route.sessionKey);
         }
-        const nextRoute = { ...route, socket };
+        const nextRoute = { ...route, socket, ownerGeneration: client.generation };
         if (previousRoute?.binding?.chatId !== nextRoute.binding?.chatId && previousRoute?.binding) {
           clearActivityIndicator(previousRoute);
           clearProgressState(previousRoute);
@@ -2757,8 +2874,10 @@ async function handleClientRequest(socket, message) {
 await mkdir(config.stateDir, { recursive: true, mode: 0o700 });
 try { await unlink(socketPath); } catch {}
 if (pidPath) await writeFile(pidPath, `${process.pid}\n`, { mode: 0o600 }).catch(() => undefined);
+if (epochPath) await writeFile(epochPath, `${brokerEpoch}\n`, { mode: 0o600 }).catch(() => undefined);
 
 const server = net.createServer((socket) => {
+  socketGenerations.set(socket, ++nextClientGeneration);
   recordDiagnostic({ component: 'broker', event: 'socket.connect', outcome: 'accepted', details: { clients: clients.size + 1 } });
   socket.setEncoding('utf8');
   let buffer = '';
@@ -2798,8 +2917,9 @@ const shutdown = async () => {
   clearAllActivityIndicators();
   clearAllProgressStates();
   server.close();
-  try { await unlink(socketPath); } catch {}
-  if (pidPath) { try { await unlink(pidPath); } catch {} }
+  // Scope control files are supervisor-owned. Leaving stale files is safer
+  // than an old/orphan broker unlinking a replacement broker's live files;
+  // the next supervised startup removes files after PID/epoch verification.
   process.exit(0);
 };
 

extensions/relay/broker/protocol.ts

@@ -3,6 +3,65 @@ import type { RelayInboundEvent, RelayOutboundPayload } from "../core/adapter-co
 import type { RelaySessionRouteDescriptor } from "../core/session-contracts.js";
 import type { ApprovalDecisionKind, ApprovalRiskCategory } from "../core/approval-gates.js";
 import type { RelayFileDeliveryRequester } from "../core/requester-file-delivery.js";
+import type { RelayRouteState } from "../core/relay-core.js";
+
+export type BrokerEpoch = string;
+
+export const MAX_BROKER_RECONCILIATION_ROUTES = 64;
+export const MAX_BROKER_RECONCILIATION_PAYLOAD_BYTES = 256 * 1024;
+
+export interface BrokerRouteReconciliationRequest {
+  clientId: string;
+  observedBrokerEpoch?: BrokerEpoch;
+  routes: RelayRouteState[];
+}
+
+export type BrokerRouteReconciliationRejectionCode =
+  | "invalid-request"
+  | "payload-too-large"
+  | "too-many-routes"
+  | "duplicate-route"
+  | "stale-epoch"
+  | "stale-owner";
+
+export interface BrokerRouteReconciliationRejection {
+  index?: number;
+  code: BrokerRouteReconciliationRejectionCode;
+  safeMessage: string;
+}
+
+export interface BrokerRouteReconciliationResponse {
+  brokerEpoch: BrokerEpoch;
+  acceptedSessionKeys: string[];
+  rejected: BrokerRouteReconciliationRejection[];
+}
+
+export interface BrokerHealthSnapshot {
+  brokerEpoch: BrokerEpoch;
+  connectedClientCount: number;
+  registeredRouteCount: number;
+}
+
+export interface BrokerRestartPreparation {
+  brokerEpoch: BrokerEpoch;
+  expectedClientCount: number;
+  expectedRouteCount: number;
+}
+
+export type BrokerRestartOutcome =
+  | { status: "complete"; expectedClientCount: number; connectedClientCount: number; registeredRouteCount: number }
+  | { status: "partial-timeout"; expectedClientCount: number; connectedClientCount: number; registeredRouteCount: number }
+  | { status: "in-progress"; expectedClientCount: number; connectedClientCount: number; registeredRouteCount: number }
+  | { status: "failed"; expectedClientCount: number; connectedClientCount: number; registeredRouteCount: number; safeMessage: string };
+
+export type BrokerTransportHealth = "disconnected" | "connecting" | "connected" | "reconnecting" | "unavailable";
+export type BrokerRouteSynchronizationHealth = "unknown" | "pending" | "synchronized" | "missing" | "rejected";
+
+export interface BrokerClientSynchronizationState {
+  transport: BrokerTransportHealth;
+  brokerEpoch?: BrokerEpoch;
+  routes: Readonly<Record<string, BrokerRouteSynchronizationHealth>>;
+}
 
 export type BrokerPeerAuthKind = "shared-secret" | "keypair";