diff --git a/README.md b/README.md index b12de2a..95b8f78 100644 --- a/README.md +++ b/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. diff --git a/docs/testing.md b/docs/testing.md index 7561453..d56acbb 100644 --- a/docs/testing.md +++ b/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`. diff --git a/extensions/relay/broker/process.js b/extensions/relay/broker/process.js index 92d059a..dab76b5 100644 --- a/extensions/relay/broker/process.js +++ b/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); }; diff --git a/extensions/relay/broker/protocol.ts b/extensions/relay/broker/protocol.ts index 25057c4..cd82d48 100644 --- a/extensions/relay/broker/protocol.ts +++ b/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>; +} export type BrokerPeerAuthKind = "shared-secret" | "keypair"; diff --git a/extensions/relay/broker/reconciliation.ts b/extensions/relay/broker/reconciliation.ts new file mode 100644 index 0000000..3760088 --- /dev/null +++ b/extensions/relay/broker/reconciliation.ts @@ -0,0 +1,172 @@ +import { Buffer } from "node:buffer"; +import type { RelayRouteState } from "../core/relay-core.js"; +import { + MAX_BROKER_RECONCILIATION_PAYLOAD_BYTES, + MAX_BROKER_RECONCILIATION_ROUTES, + type BrokerClientSynchronizationState, + type BrokerEpoch, + type BrokerRouteReconciliationRejection, + type BrokerRouteReconciliationRequest, + type BrokerRouteReconciliationResponse, + type BrokerRouteSynchronizationHealth, +} from "./protocol.js"; + +const MAX_CLIENT_ID_CHARS = 128; +const MAX_SESSION_KEY_CHARS = 8_192; +const MAX_SESSION_ID_CHARS = 512; +const MAX_SESSION_LABEL_CHARS = 512; +const MAX_SESSION_FILE_CHARS = 16_384; + +export type BrokerReconciliationValidationResult = + | { ok: true; request: BrokerRouteReconciliationRequest } + | { ok: false; rejection: BrokerRouteReconciliationRejection }; + +export function normalizeBrokerReconciliationRequest( + value: unknown, + currentBrokerEpoch: BrokerEpoch, +): BrokerReconciliationValidationResult { + if (!isRecord(value)) return invalidRequest("Invalid broker reconciliation request."); + if (!isBoundedNonEmptyString(value.clientId, MAX_CLIENT_ID_CHARS) || !Array.isArray(value.routes)) { + return invalidRequest("Invalid broker reconciliation request."); + } + if (value.observedBrokerEpoch !== undefined && !isBoundedNonEmptyString(value.observedBrokerEpoch, 128)) { + return invalidRequest("Invalid broker reconciliation epoch."); + } + if (value.observedBrokerEpoch !== undefined && value.observedBrokerEpoch !== currentBrokerEpoch) { + return { + ok: false, + rejection: { code: "stale-epoch", safeMessage: "Broker epoch changed; route reconciliation must be retried." }, + }; + } + if (value.routes.length > MAX_BROKER_RECONCILIATION_ROUTES) { + return { + ok: false, + rejection: { code: "too-many-routes", safeMessage: "Broker route reconciliation exceeds the route limit." }, + }; + } + if (serializedByteSize(value) > MAX_BROKER_RECONCILIATION_PAYLOAD_BYTES) { + return { + ok: false, + rejection: { code: "payload-too-large", safeMessage: "Broker route reconciliation exceeds the payload limit." }, + }; + } + + const routes: RelayRouteState[] = []; + const sessionKeys = new Set(); + for (let index = 0; index < value.routes.length; index += 1) { + const route = value.routes[index]; + if (!isRelayRouteState(route)) { + return { ok: false, rejection: { index, code: "invalid-request", safeMessage: "Broker route descriptor is invalid." } }; + } + if (sessionKeys.has(route.sessionKey)) { + return { ok: false, rejection: { index, code: "duplicate-route", safeMessage: "Broker route reconciliation contains a duplicate route." } }; + } + sessionKeys.add(route.sessionKey); + routes.push(route); + } + + return { + ok: true, + request: { + clientId: value.clientId, + observedBrokerEpoch: value.observedBrokerEpoch, + routes, + }, + }; +} + +export function normalizeBrokerReconciliationResponse(value: unknown): BrokerRouteReconciliationResponse | undefined { + if (!isRecord(value) || !isBoundedNonEmptyString(value.brokerEpoch, 128)) return undefined; + if (!Array.isArray(value.acceptedSessionKeys) || !value.acceptedSessionKeys.every((key) => isBoundedNonEmptyString(key, MAX_SESSION_KEY_CHARS))) return undefined; + if (new Set(value.acceptedSessionKeys).size !== value.acceptedSessionKeys.length || !Array.isArray(value.rejected)) return undefined; + const rejected: BrokerRouteReconciliationRejection[] = []; + for (const rejection of value.rejected) { + if (!isRecord(rejection) || !isRejectionCode(rejection.code) || typeof rejection.safeMessage !== "string") return undefined; + if (rejection.index !== undefined && (!Number.isInteger(rejection.index) || Number(rejection.index) < 0)) return undefined; + rejected.push({ + index: rejection.index === undefined ? undefined : Number(rejection.index), + code: rejection.code, + safeMessage: rejection.safeMessage, + }); + } + return { brokerEpoch: value.brokerEpoch, acceptedSessionKeys: value.acceptedSessionKeys, rejected }; +} + +export function synchronizationStateAfterConnect( + sessionKeys: readonly string[], + previous: BrokerClientSynchronizationState = { transport: "disconnected", routes: {} }, +): BrokerClientSynchronizationState { + return { + transport: "connected", + brokerEpoch: previous.brokerEpoch, + routes: routeHealthRecord(sessionKeys, "pending"), + }; +} + +export function synchronizationStateAfterDisconnect( + sessionKeys: readonly string[], + transport: "disconnected" | "reconnecting" | "unavailable" = "reconnecting", +): BrokerClientSynchronizationState { + return { transport, routes: routeHealthRecord(sessionKeys, "unknown") }; +} + +export function synchronizationStateAfterResponse( + sessionKeys: readonly string[], + response: BrokerRouteReconciliationResponse, +): BrokerClientSynchronizationState { + const accepted = new Set(response.acceptedSessionKeys); + const hasGeneralRejection = response.rejected.some((rejection) => rejection.index === undefined); + const rejectedIndices = new Set(response.rejected.flatMap((rejection) => rejection.index === undefined ? [] : [rejection.index])); + const routes: Record = {}; + for (let index = 0; index < sessionKeys.length; index += 1) { + const sessionKey = sessionKeys[index]!; + routes[sessionKey] = accepted.has(sessionKey) ? "synchronized" : hasGeneralRejection || rejectedIndices.has(index) ? "rejected" : "missing"; + } + return { transport: "connected", brokerEpoch: response.brokerEpoch, routes }; +} + +function routeHealthRecord( + sessionKeys: readonly string[], + health: BrokerRouteSynchronizationHealth, +): Record { + return Object.fromEntries(sessionKeys.map((sessionKey) => [sessionKey, health])); +} + +function invalidRequest(safeMessage: string): BrokerReconciliationValidationResult { + return { ok: false, rejection: { code: "invalid-request", safeMessage } }; +} + +function serializedByteSize(value: unknown): number { + try { + return Buffer.byteLength(JSON.stringify(value), "utf8"); + } catch { + return Number.POSITIVE_INFINITY; + } +} + +function isRelayRouteState(value: unknown): value is RelayRouteState { + if (!isRecord(value)) return false; + return isBoundedNonEmptyString(value.channel, 64) + && isBoundedNonEmptyString(value.sessionKey, MAX_SESSION_KEY_CHARS) + && isBoundedNonEmptyString(value.sessionId, MAX_SESSION_ID_CHARS) + && (value.sessionFile === undefined || isBoundedNonEmptyString(value.sessionFile, MAX_SESSION_FILE_CHARS)) + && isBoundedNonEmptyString(value.sessionLabel, MAX_SESSION_LABEL_CHARS) + && typeof value.busy === "boolean" + && isRecord(value.notification) + && (value.binding === undefined || isRecord(value.binding)) + && (value.modelId === undefined || typeof value.modelId === "string") + && (value.imageInputSupported === undefined || typeof value.imageInputSupported === "boolean") + && (value.lastActivityAt === undefined || typeof value.lastActivityAt === "number" && Number.isFinite(value.lastActivityAt)); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isRejectionCode(value: unknown): value is BrokerRouteReconciliationRejection["code"] { + return value === "invalid-request" || value === "payload-too-large" || value === "too-many-routes" || value === "duplicate-route" || value === "stale-epoch" || value === "stale-owner"; +} + +function isBoundedNonEmptyString(value: unknown, maxChars: number): value is string { + return typeof value === "string" && value.length > 0 && value.length <= maxChars; +} diff --git a/extensions/relay/broker/supervisor.ts b/extensions/relay/broker/supervisor.ts index 450f850..39b4d97 100644 --- a/extensions/relay/broker/supervisor.ts +++ b/extensions/relay/broker/supervisor.ts @@ -8,6 +8,7 @@ export interface LocalBrokerControlPaths { namespace?: string; socketPath: string; pidPath: string; + epochPath: string; lockPath: string; } @@ -52,6 +53,7 @@ function controlPathsForBasename(stateDir: string, basename: string, namespace?: namespace, socketPath: join(stateDir, `${basename}.sock`), pidPath: join(stateDir, `${basename}.pid`), + epochPath: join(stateDir, `${basename}.epoch`), lockPath: join(stateDir, `${basename}.lock`), }; } @@ -82,12 +84,13 @@ export function isProcessAlive(pid: number): boolean { export async function readBrokerPid(pidPath: string): Promise { try { await access(pidPath, constants.R_OK); - } catch { - return undefined; + const raw = (await readFile(pidPath, "utf8")).trim(); + const pid = Number(raw); + return Number.isInteger(pid) && pid > 0 ? pid : undefined; + } catch (error) { + if (typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT") return undefined; + throw error; } - const raw = (await readFile(pidPath, "utf8")).trim(); - const pid = Number(raw); - return Number.isInteger(pid) && pid > 0 ? pid : undefined; } export async function cleanupStaleBrokerControlFiles(paths: LocalBrokerControlPaths, isAlive: (pid: number) => boolean = isProcessAlive): Promise { @@ -95,6 +98,7 @@ export async function cleanupStaleBrokerControlFiles(paths: LocalBrokerControlPa if (pid && isAlive(pid)) return false; await Promise.all([ rm(paths.pidPath, { force: true }), + rm(paths.epochPath, { force: true }), rm(paths.socketPath, { force: true }), rm(paths.lockPath, { force: true, recursive: true }), ]); @@ -166,6 +170,7 @@ export async function ensureScopedBroker(options: EnsureScopedBrokerOptions): Pr await Promise.all([ rm(paths.pidPath, { force: true }), + rm(paths.epochPath, { force: true }), rm(paths.socketPath, { force: true }), ]); const started = await options.startBroker(paths); @@ -176,3 +181,78 @@ export async function ensureScopedBroker(options: EnsureScopedBrokerOptions): Pr await release(); } } + +export interface ReplaceScopedBrokerOptions extends BrokerScopeOptions { + selectedPid?: number; + selectedEpoch?: string; + stopBroker: (pid: number) => Promise; + startBroker: (paths: LocalBrokerControlPaths) => Promise; + probeSocket: (paths: LocalBrokerControlPaths) => Promise; + waitForSocketReady: (paths: LocalBrokerControlPaths) => Promise; + isAlive?: (pid: number) => boolean; +} + +export type ReplaceScopedBrokerResult = + | { status: "replaced"; paths: LocalBrokerControlPaths; oldPid?: number; pid: number } + | { status: "joined"; paths: LocalBrokerControlPaths; pid: number }; + +export async function replaceScopedBroker(options: ReplaceScopedBrokerOptions): Promise { + const paths = brokerScopeControlPaths(options); + const alive = options.isAlive ?? isProcessAlive; + await mkdir(options.stateDir, { recursive: true, mode: 0o700 }); + await removeLegacyBrokerLockFile(paths.lockPath); + const lockTargetPath = paths.lockPath.endsWith(".lock") ? paths.lockPath.slice(0, -".lock".length) : `${paths.lockPath}.target`; + const release = await lockfile.lock(lockTargetPath, { + lockfilePath: paths.lockPath, + realpath: false, + stale: 60_000, + retries: { retries: 200, minTimeout: 50, maxTimeout: 500 }, + }); + try { + let authoritativePid = await readBrokerPid(paths.pidPath); + const authoritativeEpoch = await readBrokerEpoch(paths.epochPath); + if (options.selectedEpoch && authoritativeEpoch && authoritativeEpoch !== options.selectedEpoch && authoritativePid && alive(authoritativePid) && await options.probeSocket(paths)) { + return { status: "joined", paths, pid: authoritativePid }; + } + if (options.selectedPid && authoritativePid && authoritativePid !== options.selectedPid && alive(authoritativePid) && await options.probeSocket(paths)) { + return { status: "joined", paths, pid: authoritativePid }; + } + + const oldPid = options.selectedPid ?? authoritativePid; + const selectedEpochMatches = Boolean(options.selectedEpoch && authoritativeEpoch === options.selectedEpoch); + const selectedBrokerProven = Boolean(oldPid && authoritativePid === oldPid && alive(oldPid) && (selectedEpochMatches || await options.probeSocket(paths))); + if (oldPid && authoritativePid === oldPid && alive(oldPid) && !selectedBrokerProven) { + throw new Error("PiRelay broker PID is live but its scoped socket is unavailable; refusing unsafe replacement."); + } + if (oldPid && authoritativePid === oldPid && selectedBrokerProven) await options.stopBroker(oldPid); + + authoritativePid = await readBrokerPid(paths.pidPath); + if (authoritativePid && authoritativePid !== oldPid && alive(authoritativePid) && await options.probeSocket(paths)) { + return { status: "joined", paths, pid: authoritativePid }; + } + if (authoritativePid === oldPid || !authoritativePid || !alive(authoritativePid)) { + await Promise.all([rm(paths.pidPath, { force: true }), rm(paths.epochPath, { force: true }), rm(paths.socketPath, { force: true })]); + } else if (await options.probeSocket(paths)) { + return { status: "joined", paths, pid: authoritativePid }; + } else { + throw new Error("A different live PiRelay broker owns the scope but its socket is unavailable."); + } + + const started = await options.startBroker(paths); + await writeFile(paths.pidPath, `${started.pid}\n`, { mode: 0o600 }); + await options.waitForSocketReady(paths); + return { status: "replaced", paths, oldPid, pid: started.pid }; + } finally { + await release(); + } +} + +export async function readBrokerEpoch(epochPath: string): Promise { + try { + const value = (await readFile(epochPath, "utf8")).trim(); + return value || undefined; + } catch (error) { + if (typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT") return undefined; + throw error; + } +} diff --git a/extensions/relay/broker/tunnel-runtime.ts b/extensions/relay/broker/tunnel-runtime.ts index 18a01d4..05e350b 100644 --- a/extensions/relay/broker/tunnel-runtime.ts +++ b/extensions/relay/broker/tunnel-runtime.ts @@ -1,6 +1,6 @@ import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { readFile, unlink } from "node:fs/promises"; +import { readFile } from "node:fs/promises"; import { createConnection, type Socket } from "node:net"; import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; @@ -13,10 +13,23 @@ import { relayRouteStateForRoute, statusSnapshotForRoute, type RelayRouteState } import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, newSessionRouteSafely, routeActionDisplayMessage, routeImageByPathSafely, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../core/route-actions.js"; import { relayPipelineProtocolVersion } from "../middleware/pipeline.js"; import { sha256 } from "../core/utils.js"; -import { brokerScopeControlPaths, ensureScopedBroker, normalizeBrokerNamespace } from "./supervisor.js"; +import { brokerScopeControlPaths, ensureScopedBroker, normalizeBrokerNamespace, replaceScopedBroker } from "./supervisor.js"; +import { normalizeBrokerReconciliationResponse, synchronizationStateAfterConnect, synchronizationStateAfterDisconnect, synchronizationStateAfterResponse } from "./reconciliation.js"; +import type { BrokerClientSynchronizationState, BrokerHealthSnapshot, BrokerRestartOutcome, BrokerRestartPreparation } from "./protocol.js"; +import { createCommunicationDiagnosticsLogger, type CommunicationDiagnosticsLogger } from "../diagnostics/communication.js"; const BROKER_PROTOCOL_VERSION = 1; const BROKER_CHANNEL = "telegram" as const; +const DEFAULT_RECONCILIATION_INTERVAL_MS = 20_000; +const MAX_RECONCILIATION_JITTER_MS = 10_000; + +export interface BrokerTunnelRuntimeOptions { + reconciliationIntervalMs?: number; + restartConvergenceTimeoutMs?: number; + requestTimeoutMs?: number; + random?: () => number; + onSynchronizationStateChange?: (state: BrokerClientSynchronizationState) => void; +} type BrokerRouteState = RelayRouteState; @@ -55,6 +68,7 @@ export class BrokerTunnelRuntime implements TunnelRuntime { private readonly clientId = randomUUID(); private readonly socketPath: string; private readonly pidPath: string; + private readonly epochPath: string; private readonly brokerNamespace?: string; private readonly routes = new Map(); private readonly pending = new Map void; reject: (error: Error) => void }>(); @@ -62,11 +76,22 @@ export class BrokerTunnelRuntime implements TunnelRuntime { private buffer = ""; private connecting?: Promise; private reconnectTimer?: ReturnType; + private reconciliationTimer?: ReturnType; + private reconciliationQueue: Promise = Promise.resolve(); + private operationGeneration = 0; + private reconnectNeeded = false; + private restartInProgress = false; private reconnectDelayMs = 250; private started = false; private setupCache?: SetupCache; - - constructor(private readonly config: TelegramTunnelConfig) { + private synchronizationState: BrokerClientSynchronizationState = { transport: "disconnected", routes: {} }; + private reconciliationSupported?: boolean; + private synchronizationListener?: (state: BrokerClientSynchronizationState) => void; + private readonly diagnostics: CommunicationDiagnosticsLogger; + private lastFailureDiagnostic?: string; + + constructor(private readonly config: TelegramTunnelConfig, private readonly options: BrokerTunnelRuntimeOptions = {}) { + this.diagnostics = createCommunicationDiagnosticsLogger(config.communicationDiagnostics); this.brokerNamespace = normalizeBrokerNamespace(config.brokerNamespace ?? process.env.PI_RELAY_BROKER_NAMESPACE); const tokenHash = sha256(config.botToken).slice(0, 16); // Broker scope is intentionally machine-local and secret-safe: @@ -75,6 +100,7 @@ export class BrokerTunnelRuntime implements TunnelRuntime { const paths = brokerScopeControlPaths({ stateDir: config.stateDir, tokenHash, namespace: this.brokerNamespace }); this.socketPath = paths.socketPath; this.pidPath = paths.pidPath; + this.epochPath = paths.epochPath; } get setup(): SetupCache | undefined { @@ -83,27 +109,82 @@ export class BrokerTunnelRuntime implements TunnelRuntime { async start(): Promise { this.started = true; - await this.ensureConnected(); + try { + await this.ensureConnected(); + } catch (error) { + this.scheduleReconnect(); + throw error; + } finally { + this.schedulePeriodicReconciliation(); + } } async stop(): Promise { + this.operationGeneration += 1; this.started = false; this.disconnectClient(new Error("Broker runtime stopped.")); } - async restartBrokerProcess(): Promise { - this.started = false; - this.disconnectClient(new Error("Broker runtime restarted.")); - const pid = await this.readBrokerPid(); - if (pid !== undefined && this.isProcessAlive(pid)) { + async restartBrokerProcess(): Promise { + if (this.restartInProgress) { + return { status: "in-progress", expectedClientCount: 1, connectedClientCount: this.socket && !this.socket.destroyed ? 1 : 0, registeredRouteCount: Object.values(this.synchronizationState.routes).filter((health) => health === "synchronized").length }; + } + this.restartInProgress = true; + try { + const restartStartedAt = Date.now(); + const fallbackExpected = this.routes.size; + let preparation: BrokerRestartPreparation = { brokerEpoch: this.synchronizationState.brokerEpoch ?? "legacy", expectedClientCount: fallbackExpected > 0 ? 1 : 0, expectedRouteCount: fallbackExpected }; + if (this.socket && !this.socket.destroyed) { try { - process.kill(pid, "SIGTERM"); + const value = await this.requestOnce("prepareBrokerRestart", { clientId: this.clientId }); + if (isBrokerRestartPreparation(value)) preparation = value; } catch (error) { - if (!isNoSuchProcessError(error)) throw new Error(`Failed to stop PiRelay broker process ${pid}: ${error instanceof Error ? error.message : String(error)}`); + if (!isUnsupportedRestartAction(error)) throw error; } - await this.waitForBrokerProcessExit(pid); } - await this.unlinkBrokerFiles(); + const selectedPid = await this.readBrokerPid(); + this.recordDiagnostic("broker.restart", "prepared", { expectedClientCount: preparation.expectedClientCount, expectedRouteCount: preparation.expectedRouteCount }); + this.operationGeneration += 1; + this.started = false; + this.disconnectClient(new Error("Broker runtime restarted.")); + const tokenHash = sha256(this.config.botToken).slice(0, 16); + const replacement = await replaceScopedBroker({ + stateDir: this.config.stateDir, + tokenHash, + namespace: this.brokerNamespace, + selectedPid, + selectedEpoch: preparation.brokerEpoch === "legacy" ? undefined : preparation.brokerEpoch, + isAlive: (pid) => this.isProcessAlive(pid), + probeSocket: async () => this.probeSocket().then(() => true, () => false), + waitForSocketReady: async () => this.waitForSocketReady(), + stopBroker: async (pid) => { + try { + process.kill(pid, "SIGTERM"); + } catch (error) { + if (!isNoSuchProcessError(error)) throw error; + } + await this.waitForBrokerProcessExit(pid); + }, + startBroker: async () => this.spawnBroker(), + }); + this.recordDiagnostic("broker.restart", replacement.status, {}); + this.started = true; + try { + await this.ensureConnected(); + const outcome = await this.waitForBrokerConvergence(preparation); + this.recordDiagnostic("broker.restart", outcome.status, { expectedClientCount: outcome.expectedClientCount, connectedClientCount: outcome.connectedClientCount, registeredRouteCount: outcome.registeredRouteCount, durationMs: Date.now() - restartStartedAt }); + return outcome; + } catch { + this.scheduleReconnect(); + return { status: "failed", expectedClientCount: preparation.expectedClientCount, connectedClientCount: 0, registeredRouteCount: 0, safeMessage: "Broker restart did not converge; automatic retry remains active." }; + } + } catch { + this.started = true; + this.scheduleReconnect(); + return { status: "failed", expectedClientCount: 1, connectedClientCount: 0, registeredRouteCount: 0, safeMessage: "Broker restart failed safely; automatic retry remains active." }; + } finally { + this.restartInProgress = false; + } } async ensureSetup(): Promise { @@ -114,21 +195,46 @@ export class BrokerTunnelRuntime implements TunnelRuntime { async registerRoute(route: SessionRoute): Promise { this.routes.set(route.sessionKey, route); - await this.request("registerRoute", { - clientId: this.clientId, - route: this.serializeRoute(route), - }); - route.actions.clearLocalStatus?.("relay-sync"); + this.publishSynchronizationState(this.synchronizationStatePendingForCurrentRoutes()); + try { + await this.reconcileRoutes(); + route.actions.clearLocalStatus?.("relay-sync"); + } finally { + this.schedulePeriodicReconciliation(); + } } async unregisterRoute(sessionKey: string): Promise { this.routes.delete(sessionKey); - await this.request("unregisterRoute", { clientId: this.clientId, sessionKey }); + this.publishSynchronizationState({ + ...this.synchronizationState, + routes: Object.fromEntries(Object.entries(this.synchronizationState.routes).filter(([key]) => key !== sessionKey)), + }); if (this.routes.size === 0) { + this.operationGeneration += 1; + this.cancelPeriodicReconciliation(); + if (this.socket && !this.socket.destroyed) { + await this.requestOnce("unregisterRoute", { clientId: this.clientId, sessionKey }).catch(() => undefined); + } await this.stop(); + } else { + await this.reconcileRoutes(); } } + getBrokerSynchronizationState(): BrokerClientSynchronizationState { + return this.synchronizationState; + } + + setBrokerSynchronizationListener(listener: (state: BrokerClientSynchronizationState) => void): void { + this.synchronizationListener = listener; + } + + async reconcileBrokerRoutes(): Promise { + if (this.routes.size > 0) await this.reconcileRoutes("doctor"); + return this.synchronizationState; + } + getStatus(sessionKey: string): SessionStatusSnapshot | undefined { const route = this.routes.get(sessionKey); if (!route) return undefined; @@ -146,15 +252,18 @@ export class BrokerTunnelRuntime implements TunnelRuntime { private disconnectClient(error: Error): void { if (this.reconnectTimer) clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; + this.cancelPeriodicReconciliation(); this.rejectPending(error); this.socket?.destroy(); this.socket = undefined; + this.publishSynchronizationState(synchronizationStateAfterDisconnect([...this.routes.keys()], this.started ? "reconnecting" : "disconnected")); } private async ensureConnected(): Promise { if (this.socket && !this.socket.destroyed) return; if (this.connecting) return this.connecting; + const generation = this.operationGeneration; this.connecting = (async () => { await ensureStateDir(this.config.stateDir); if (!await this.tryConnectSocket()) { @@ -170,6 +279,11 @@ export class BrokerTunnelRuntime implements TunnelRuntime { }); await this.connectSocket(); } + if (generation !== this.operationGeneration) { + this.socket?.destroy(); + this.socket = undefined; + return; + } await this.resyncRoutes(); })(); @@ -177,6 +291,10 @@ export class BrokerTunnelRuntime implements TunnelRuntime { await this.connecting; } finally { this.connecting = undefined; + if (this.started && (this.reconnectNeeded || !this.socket || this.socket.destroyed)) { + this.reconnectNeeded = false; + this.scheduleReconnect(); + } } } @@ -210,6 +328,10 @@ export class BrokerTunnelRuntime implements TunnelRuntime { this.reconnectTimer = undefined; this.reconnectDelayMs = 250; this.socket = socket; + this.reconciliationSupported = undefined; + this.reconnectNeeded = false; + this.publishSynchronizationState(synchronizationStateAfterConnect([...this.routes.keys()], this.synchronizationState)); + this.recordDiagnostic("broker.connect", "connected", { routeCount: this.routes.size }); this.buffer = ""; socket.setEncoding("utf8"); socket.on("data", (chunk: string) => { @@ -219,13 +341,17 @@ export class BrokerTunnelRuntime implements TunnelRuntime { const line = this.buffer.slice(0, newlineIndex).trim(); this.buffer = this.buffer.slice(newlineIndex + 1); if (line) { - void this.handleMessage(line); + void this.handleMessage(line, socket); } newlineIndex = this.buffer.indexOf("\n"); } }); socket.on("close", () => { - if (this.socket === socket) this.socket = undefined; + if (this.socket !== socket) return; + this.socket = undefined; + if (this.started) this.reconnectNeeded = true; + this.publishSynchronizationState(synchronizationStateAfterDisconnect([...this.routes.keys()], this.started ? "reconnecting" : "disconnected")); + this.recordDiagnostic("broker.disconnect", this.started ? "reconnecting" : "stopped", { routeCount: this.routes.size }); this.rejectPending(new Error("Broker connection closed.")); this.scheduleReconnect(); }); @@ -234,7 +360,8 @@ export class BrokerTunnelRuntime implements TunnelRuntime { }); } - private async handleMessage(line: string): Promise { + private async handleMessage(line: string, sourceSocket?: Socket): Promise { + if (sourceSocket && sourceSocket !== this.socket) return; const message = JSON.parse(line) as BrokerProtocolRequest | BrokerProtocolResponse; if (message.type === "response") { const pending = this.pending.get(message.requestId); @@ -271,6 +398,20 @@ export class BrokerTunnelRuntime implements TunnelRuntime { return; } + if (request.action === "brokerRestartNotice") { + await respond({ ok: true }); + this.reconnectNeeded = this.started; + this.publishSynchronizationState(synchronizationStateAfterDisconnect([...this.routes.keys()], this.started ? "reconnecting" : "disconnected")); + setTimeout(() => { + if (sourceSocket) { + if (this.socket === sourceSocket) sourceSocket.destroy(); + } else { + this.socket?.destroy(); + } + }, 0).unref?.(); + return; + } + const sessionKey = String(request.sessionKey ?? ""); const route = this.routes.get(sessionKey); if (!route) { @@ -423,6 +564,7 @@ export class BrokerTunnelRuntime implements TunnelRuntime { private async request(action: string, payload: Record): Promise { await this.ensureConnected(); return this.requestOnce(action, payload).catch(async (error) => { + if (isBrokerRequestTimeout(error)) throw error; this.socket?.destroy(); this.socket = undefined; await this.ensureConnected(); @@ -445,10 +587,24 @@ export class BrokerTunnelRuntime implements TunnelRuntime { action, pipeline: { protocolVersion: relayPipelineProtocolVersion, channel: BROKER_CHANNEL, action }, }; + const timeoutMs = this.options.requestTimeoutMs ?? (action === "getBrokerHealth" ? 2_000 : action === "prepareBrokerRestart" || action === "reconcileRoutes" ? 10_000 : 30_000); const result = new Promise((resolvePromise, rejectPromise) => { - this.pending.set(requestId, { resolve: resolvePromise, reject: rejectPromise }); + const timeout = setTimeout(() => { + if (!this.pending.delete(requestId)) return; + rejectPromise(new Error(`Broker ${action} request timed out.`)); + }, timeoutMs); + timeout.unref?.(); + this.pending.set(requestId, { + resolve: (value) => { clearTimeout(timeout); resolvePromise(value); }, + reject: (error) => { clearTimeout(timeout); rejectPromise(error); }, + }); }); - this.writeMessage(message); + try { + this.writeMessage(message); + } catch (error) { + this.pending.get(requestId)?.reject(error instanceof Error ? error : new Error("Broker request write failed.")); + this.pending.delete(requestId); + } return result; } @@ -468,13 +624,178 @@ export class BrokerTunnelRuntime implements TunnelRuntime { } private async resyncRoutes(): Promise { - for (const route of this.routes.values()) { - await this.requestOnce("registerRoute", { + const startedAt = Date.now(); + if (this.routes.size === 0) return; + let result: unknown; + try { + result = await this.requestOnce("reconcileRoutes", { clientId: this.clientId, - route: this.serializeRoute(route), + routes: [...this.routes.values()].map((route) => this.serializeRoute(route)), }); - route.actions.clearLocalStatus?.("relay-sync"); + } catch (error) { + if (!isUnsupportedReconciliationAction(error)) throw error; + this.reconciliationSupported = false; + await this.resyncRoutesLegacy(); + for (const route of this.routes.values()) route.actions.clearLocalStatus?.("relay-sync"); + return; } + if (!this.applyReconciliationResponse(result, startedAt, "connect")) await this.resyncRoutesLegacy(); + for (const route of this.routes.values()) route.actions.clearLocalStatus?.("relay-sync"); + } + + private reconcileRoutes(trigger: "route-mutation" | "periodic" | "doctor" = "route-mutation"): Promise { + const generation = this.operationGeneration; + const operation = this.reconciliationQueue + .catch(() => undefined) + .then(async () => { + if (generation !== this.operationGeneration || this.routes.size === 0) return; + await this.reconcileRoutesOnce(trigger); + }); + this.reconciliationQueue = operation.catch(() => undefined); + return operation; + } + + private async reconcileRoutesOnce(trigger: "route-mutation" | "periodic" | "doctor"): Promise { + const startedAt = Date.now(); + this.recordDiagnostic("broker.reconcile", "attempt", { routeCount: this.routes.size, trigger }); + if (this.reconciliationSupported === false) { + await this.reconcileRoutesLegacy(); + return; + } + this.publishSynchronizationState(this.synchronizationStatePendingForCurrentRoutes()); + let result: unknown; + try { + await this.ensureConnected(); + result = await this.requestOnce("reconcileRoutes", { + clientId: this.clientId, + observedBrokerEpoch: this.synchronizationState.brokerEpoch, + routes: [...this.routes.values()].map((route) => this.serializeRoute(route)), + }); + } catch (error) { + if (!isUnsupportedReconciliationAction(error)) throw error; + this.reconciliationSupported = false; + await this.reconcileRoutesLegacy(); + return; + } + if (!this.applyReconciliationResponse(result, startedAt, trigger)) await this.reconcileRoutesLegacy(); + } + + private applyReconciliationResponse(value: unknown, startedAt = Date.now(), trigger = "direct"): boolean { + const previous = this.synchronizationState; + const response = normalizeBrokerReconciliationResponse(value); + if (!response) { + if (value === undefined || value === true) return false; + throw new Error("Invalid broker route reconciliation response."); + } + this.reconciliationSupported = true; + this.publishSynchronizationState(synchronizationStateAfterResponse([...this.routes.keys()], response)); + const synchronizedCount = Object.values(this.synchronizationState.routes).filter((health) => health === "synchronized").length; + const repairedCount = Object.entries(this.synchronizationState.routes).filter(([sessionKey, health]) => health === "synchronized" && previous.routes[sessionKey] !== "synchronized").length; + this.recordDiagnostic("broker.reconcile", response.rejected.length > 0 ? "rejected" : repairedCount > 0 ? "repaired" : "acknowledged", { + routeCount: this.routes.size, + synchronizedCount, + rejectedCount: response.rejected.length, + repairedCount, + durationMs: Date.now() - startedAt, + trigger, + }); + return true; + } + + private async resyncRoutesLegacy(): Promise { + for (const route of this.routes.values()) { + await this.requestOnce("registerRoute", { clientId: this.clientId, route: this.serializeRoute(route) }); + } + this.markLegacyRoutesSynchronized(); + } + + private async reconcileRoutesLegacy(): Promise { + for (const route of this.routes.values()) { + await this.request("registerRoute", { clientId: this.clientId, route: this.serializeRoute(route) }); + } + this.markLegacyRoutesSynchronized(); + } + + private markLegacyRoutesSynchronized(): void { + this.publishSynchronizationState({ + transport: "connected", + routes: Object.fromEntries([...this.routes.keys()].map((sessionKey) => [sessionKey, "synchronized" as const])), + }); + } + + private synchronizationStatePendingForCurrentRoutes(): BrokerClientSynchronizationState { + return { + transport: "connected", + brokerEpoch: this.synchronizationState.brokerEpoch, + routes: Object.fromEntries([...this.routes.keys()].map((sessionKey) => [ + sessionKey, + this.synchronizationState.routes[sessionKey] === "synchronized" ? "synchronized" : "pending", + ])), + }; + } + + private recordDiagnostic(event: string, outcome: string, details: Record): void { + if (!this.diagnostics.config.enabled) return; + const failureSignature = outcome === "error" || outcome === "rejected" ? `${event}:${outcome}:${JSON.stringify(details)}` : undefined; + if (failureSignature) { + if (failureSignature === this.lastFailureDiagnostic) return; + this.lastFailureDiagnostic = failureSignature; + } else if (outcome === "acknowledged" || outcome === "repaired" || outcome === "connected") { + this.lastFailureDiagnostic = undefined; + } + void this.diagnostics.record({ component: "runtime", event, outcome, details }); + } + + private publishSynchronizationState(state: BrokerClientSynchronizationState): void { + this.synchronizationState = state; + try { + (this.synchronizationListener ?? this.options.onSynchronizationStateChange)?.(state); + } catch { + // Status rendering must not interfere with broker recovery. + } + } + + private schedulePeriodicReconciliation(): void { + this.cancelPeriodicReconciliation(); + if (!this.started || this.routes.size === 0) return; + const baseDelay = Math.max(10, this.options.reconciliationIntervalMs ?? DEFAULT_RECONCILIATION_INTERVAL_MS); + const random = this.options.random ?? Math.random; + const jitter = this.options.reconciliationIntervalMs === undefined + ? Math.floor(Math.max(0, Math.min(1, random())) * MAX_RECONCILIATION_JITTER_MS) + : 0; + this.reconciliationTimer = setTimeout(() => { + this.reconciliationTimer = undefined; + if (!this.started || this.routes.size === 0) return; + const reconciliation = this.reconcileRoutes("periodic") + .catch(() => { + this.recordDiagnostic("broker.reconcile", "error", { category: "request-failed" }); + if (this.started) this.scheduleReconnect(); + }) + .finally(() => { + this.schedulePeriodicReconciliation(); + }); + }, baseDelay + jitter); + this.reconciliationTimer.unref?.(); + } + + private cancelPeriodicReconciliation(): void { + if (this.reconciliationTimer) clearTimeout(this.reconciliationTimer); + this.reconciliationTimer = undefined; + } + + private async waitForBrokerConvergence(preparation: BrokerRestartPreparation): Promise { + const deadline = Date.now() + (this.options.restartConvergenceTimeoutMs ?? 10_000); + let latest: BrokerHealthSnapshot = { brokerEpoch: this.synchronizationState.brokerEpoch ?? "unknown", connectedClientCount: 1, registeredRouteCount: this.routes.size }; + while (Date.now() < deadline) { + const value = await this.requestOnce("getBrokerHealth", { clientId: this.clientId }); + if (isBrokerHealthSnapshot(value)) latest = value; + const ownRoutesSynchronized = [...this.routes.keys()].every((sessionKey) => this.synchronizationState.routes[sessionKey] === "synchronized"); + if (ownRoutesSynchronized && latest.connectedClientCount >= preparation.expectedClientCount && latest.registeredRouteCount >= preparation.expectedRouteCount) { + return { status: "complete", expectedClientCount: preparation.expectedClientCount, connectedClientCount: latest.connectedClientCount, registeredRouteCount: latest.registeredRouteCount }; + } + await new Promise((resolvePromise) => setTimeout(resolvePromise, 100)); + } + return { status: "partial-timeout", expectedClientCount: preparation.expectedClientCount, connectedClientCount: latest.connectedClientCount, registeredRouteCount: latest.registeredRouteCount }; } private async readBrokerPid(): Promise { @@ -508,18 +829,6 @@ export class BrokerTunnelRuntime implements TunnelRuntime { throw new Error(`PiRelay broker process ${pid} did not stop in time.`); } - private async unlinkBrokerFiles(): Promise { - await Promise.all([this.unlinkBrokerFile(this.socketPath), this.unlinkBrokerFile(this.pidPath)]); - } - - private async unlinkBrokerFile(path: string): Promise { - try { - await unlink(path); - } catch (error) { - if (!isMissingFileError(error)) throw new Error(`Could not remove PiRelay broker file ${path}: ${error instanceof Error ? error.message : String(error)}`); - } - } - private async spawnBroker(): Promise<{ pid: number }> { const brokerPath = fileURLToPath(new URL("./process.js", import.meta.url)); const child = spawn(process.execPath, [brokerPath], { @@ -530,6 +839,7 @@ export class BrokerTunnelRuntime implements TunnelRuntime { TELEGRAM_TUNNEL_BROKER_CONFIG_JSON: JSON.stringify(this.config), TELEGRAM_TUNNEL_BROKER_SOCKET_PATH: this.socketPath, TELEGRAM_TUNNEL_BROKER_PID_PATH: this.pidPath, + PI_RELAY_BROKER_EPOCH_PATH: this.epochPath, PI_RELAY_BROKER_NAMESPACE: this.brokerNamespace ?? "", PI_RELAY_COMMUNICATION_DIAGNOSTICS_CONFIG_JSON: JSON.stringify(this.config.communicationDiagnostics ?? { enabled: false }), }, @@ -569,19 +879,52 @@ export class BrokerTunnelRuntime implements TunnelRuntime { } private scheduleReconnect(): void { - if (!this.started || this.reconnectTimer || this.connecting) return; + if (!this.started || this.reconnectTimer) return; + if (this.connecting) { + this.reconnectNeeded = true; + return; + } this.reconnectTimer = setTimeout(() => { this.reconnectTimer = undefined; if (!this.started) return; void this.ensureConnected().catch(() => { this.reconnectDelayMs = Math.min(this.reconnectDelayMs * 2, 5_000); this.scheduleReconnect(); + }).then(() => { + this.schedulePeriodicReconciliation(); }); }, this.reconnectDelayMs); + this.recordDiagnostic("broker.retry", "scheduled", { delayMs: this.reconnectDelayMs }); this.reconnectTimer.unref?.(); } } +function isBrokerRestartPreparation(value: unknown): value is BrokerRestartPreparation { + return isRecord(value) + && typeof value.brokerEpoch === "string" + && Number.isInteger(value.expectedClientCount) && Number(value.expectedClientCount) >= 0 + && Number.isInteger(value.expectedRouteCount) && Number(value.expectedRouteCount) >= 0; +} + +function isBrokerHealthSnapshot(value: unknown): value is BrokerHealthSnapshot { + return isRecord(value) + && typeof value.brokerEpoch === "string" + && Number.isInteger(value.connectedClientCount) && Number(value.connectedClientCount) >= 0 + && Number.isInteger(value.registeredRouteCount) && Number(value.registeredRouteCount) >= 0; +} + +function isUnsupportedRestartAction(error: unknown): boolean { + return error instanceof Error && error.message === "Unknown client action: prepareBrokerRestart"; +} + +function isBrokerRequestTimeout(error: unknown): boolean { + return error instanceof Error && /^Broker [A-Za-z0-9_-]+ request timed out\.$/.test(error.message); +} + +function isUnsupportedReconciliationAction(error: unknown): boolean { + return error instanceof Error && (error.message === "Unknown client action: reconcileRoutes" || error.message === "Unknown broker action: reconcileRoutes"); +} + function isMissingFileError(error: unknown): boolean { return typeof error === "object" && error !== null && "code" in error && (error as { code?: unknown }).code === "ENOENT"; } diff --git a/extensions/relay/core/types.ts b/extensions/relay/core/types.ts index 50a6079..8fef675 100644 --- a/extensions/relay/core/types.ts +++ b/extensions/relay/core/types.ts @@ -389,11 +389,28 @@ export interface TelegramOutboundChunk { total: number; } +export interface BrokerRestartResult { + status: "complete" | "partial-timeout" | "in-progress" | "failed"; + expectedClientCount: number; + connectedClientCount: number; + registeredRouteCount: number; + safeMessage?: string; +} + +export interface BrokerSynchronizationSnapshot { + transport: "disconnected" | "connecting" | "connected" | "reconnecting" | "unavailable"; + brokerEpoch?: string; + routes: Readonly>; +} + export interface TunnelRuntime { readonly setup?: SetupCache; start(): Promise; stop(): Promise; - restartBrokerProcess?(): Promise; + restartBrokerProcess?(): Promise; + getBrokerSynchronizationState?(): BrokerSynchronizationSnapshot; + reconcileBrokerRoutes?(): Promise; + setBrokerSynchronizationListener?(listener: (state: BrokerSynchronizationSnapshot) => void): void; ensureSetup(): Promise; registerRoute(route: SessionRoute): Promise; unregisterRoute(sessionKey: string): Promise; diff --git a/extensions/relay/runtime/extension-runtime.ts b/extensions/relay/runtime/extension-runtime.ts index 82a7447..46c7de8 100644 --- a/extensions/relay/runtime/extension-runtime.ts +++ b/extensions/relay/runtime/extension-runtime.ts @@ -326,9 +326,11 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { } async function ensureRuntime(ctx?: ExtensionContext, interactiveNotice = false): Promise { - if (runtime) return runtime; - const config = await ensureConfig(ctx, interactiveNotice); - runtime = getOrCreateTunnelRuntime(config); + if (!runtime) { + const config = await ensureConfig(ctx, interactiveNotice); + runtime = getOrCreateTunnelRuntime(config); + } + runtime.setBrokerSynchronizationListener?.(() => refreshRelayStatusesSoon()); return runtime; } @@ -392,17 +394,18 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { return runtimes; } - async function stopAndClearRuntimes(ctx: ExtensionContext, options: { restartBrokerProcess?: boolean } = {}): Promise<{ telegramStopped: boolean; discordStopped: string[]; slackStopped: string[]; brokerRestarted: boolean }> { + async function stopAndClearRuntimes(ctx: ExtensionContext, options: { restartBrokerProcess?: boolean } = {}): Promise<{ telegramStopped: boolean; discordStopped: string[]; slackStopped: string[]; brokerRestarted: boolean; brokerRestartResult?: Awaited>> }> { resetToolProgress(); let telegramStopped = false; let brokerRestarted = false; + let brokerRestartResult: Awaited>> | undefined; const discordStopped: string[] = []; const slackStopped: string[] = []; const failures: unknown[] = []; if (runtime) { try { if (options.restartBrokerProcess && runtime.restartBrokerProcess) { - await runtime.restartBrokerProcess(); + brokerRestartResult = await runtime.restartBrokerProcess(); brokerRestarted = true; } else { await runtime.stop(); @@ -437,7 +440,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { const message = first instanceof Error ? first.message : String(first); ctx.ui.notify(`Stopped PiRelay runtimes with ${failures.length} warning(s): ${redactSecrets(message)}`, "warning"); } - return { telegramStopped, discordStopped, slackStopped, brokerRestarted }; + return { telegramStopped, discordStopped, slackStopped, brokerRestarted, brokerRestartResult }; } function statusKeyForChannel(channel: RelayStatusLineChannel, instanceId = "default"): string { @@ -512,11 +515,13 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { const config = await ensureConfig(ctx, false); const configured = statusConfiguredForChannel(config, channel, instanceId); const binding = runtimeStatus?.error ? undefined : await currentStatusBinding(config, channel, instanceId); + const brokerState = channel === "telegram" ? runtime?.getBrokerSynchronizationState?.() : undefined; await setMessengerStatus(ctx, channel, { configured, runtimeStarted: channel === "telegram" ? Boolean(runtime) : runtimeStatus?.started, error: runtimeStatus?.error ? redactSecrets(runtimeStatus.error) : undefined, binding, + broker: brokerState && currentRoute ? { transport: brokerState.transport, route: brokerState.routes[currentRoute.sessionKey] ?? "unknown" } : undefined, }, instanceId); } @@ -1500,7 +1505,17 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { configCache = undefined; await resetStoppedRuntimeStatuses(ctx, stopped); await syncRoute(ctx); - ctx.ui.notify(stopped.brokerRestarted ? "PiRelay broker process and runtimes restarted for this session." : "PiRelay runtimes restarted for this session.", "info"); + const convergence = stopped.brokerRestartResult || undefined; + const restartMessage = convergence + ? convergence.status === "complete" + ? `Shared PiRelay broker restarted; recovered ${convergence.connectedClientCount}/${convergence.expectedClientCount} client(s) and ${convergence.registeredRouteCount} route(s).` + : convergence.status === "failed" + ? "Shared PiRelay broker restart failed safely; automatic recovery remains active. Use /relay doctor if this session stays offline." + : convergence.status === "in-progress" + ? "A shared PiRelay broker restart is already in progress; active sessions will reconnect automatically." + : `Shared PiRelay broker restart reached a partial timeout: recovered ${convergence.connectedClientCount}/${convergence.expectedClientCount} client(s) and ${convergence.registeredRouteCount} route(s). Other live sessions will keep retrying automatically; use /relay doctor locally if one remains offline.` + : stopped.brokerRestarted ? "Shared PiRelay broker and runtimes restarted for this session." : "PiRelay runtimes restarted for this session."; + ctx.ui.notify(restartMessage, convergence && convergence.status !== "complete" ? "warning" : "info"); } function renderRelayStatusDiagnostics(): string | undefined { @@ -1537,7 +1552,23 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { const config = await ensureConfig(ctx, true); const facts = await collectRelaySetupFacts(config); const diagnostics = renderRelayStatusDiagnostics(); - ctx.ui.notify([renderRelayDoctorReport(config, relaySetupDiagnostics(config, facts)), diagnostics].filter(Boolean).join("\n\n"), "info"); + let brokerRepair: string | undefined; + const brokerState = runtime?.getBrokerSynchronizationState?.(); + const authoritativeBinding = currentRoute ? await currentStatusBinding(config, "telegram") : undefined; + if (currentRoute && authoritativeBinding && brokerState?.routes[currentRoute.sessionKey] !== "synchronized") { + try { + const repaired = await runtime?.reconcileBrokerRoutes?.(); + const routeHealth = repaired?.routes[currentRoute.sessionKey] ?? "unknown"; + brokerRepair = routeHealth === "synchronized" + ? "Broker route repair: synchronized." + : `Broker route repair: ${routeHealth}; automatic retry remains active.`; + } catch { + brokerRepair = "Broker route repair: failed; automatic retry remains active. Check opt-in communication diagnostics locally for metadata-only details."; + } + } else if (currentRoute && brokerState) { + brokerRepair = `Broker route: ${brokerState.routes[currentRoute.sessionKey] ?? "unknown"}; transport: ${brokerState.transport}.`; + } + ctx.ui.notify([renderRelayDoctorReport(config, relaySetupDiagnostics(config, facts)), brokerRepair, diagnostics].filter(Boolean).join("\n\n"), "info"); } catch (error) { const message = error instanceof Error ? error.message : String(error); ctx.ui.notify(redactSecrets([ @@ -2032,16 +2063,19 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { return; } const busy = !ctx.isIdle(); - const binding = currentRoute.binding; + const config = await ensureConfig(ctx, false); + const binding = await currentStatusBinding(config, "telegram"); + const brokerState = runtime?.getBrokerSynchronizationState?.(); + const routeState = brokerState?.routes[currentRoute.sessionKey] ?? "unknown"; const lines = [ `Session: ${currentRoute.sessionLabel}`, - `Session key: ${currentRoute.sessionKey}`, `Busy: ${busy ? "yes" : "no"}`, - `Binding: ${binding ? `${binding.chatId}/${binding.userId}` : "not paired"}`, - `Paused: ${binding?.paused ? "yes" : "no"}`, + `Binding: ${binding ? binding.paused ? "paused" : "active" : "not paired"}`, + `Broker transport: ${brokerState?.transport ?? "unknown"}`, + `Broker route: ${routeState}`, `Last status: ${currentRoute.notification.lastStatus ?? "unknown"}`, ]; - ctx.ui.notify(lines.join("\n"), "info"); + ctx.ui.notify(lines.join("\n"), routeState === "synchronized" || !binding ? "info" : "warning"); } type LocalRelayCommand = Parameters[1]; diff --git a/extensions/relay/runtime/status-line.ts b/extensions/relay/runtime/status-line.ts index 694e43f..93cec7e 100644 --- a/extensions/relay/runtime/status-line.ts +++ b/extensions/relay/runtime/status-line.ts @@ -12,6 +12,10 @@ export interface RelayStatusLineState { runtimeStarted?: boolean; error?: string; binding?: RelayStatusLineBindingState; + broker?: { + transport: "disconnected" | "connecting" | "connected" | "reconnecting" | "unavailable"; + route?: "unknown" | "pending" | "synchronized" | "missing" | "rejected"; + }; } export interface RelayStatusLineFormatOptions { @@ -35,6 +39,11 @@ export function conversationKindIcon(kind: string | undefined): "✉" | "◉" | function relayStatusLineSegment(state: RelayStatusLineState): { icon: string; detail?: string; tone: RelayStatusLineTone } { if (!state.configured) return { icon: "○", tone: "dim" }; if (state.error) return { icon: "✖", tone: "error" }; + if (state.binding && state.broker && (state.broker.transport !== "connected" || state.broker.route !== "synchronized")) { + return state.broker.route === "rejected" || state.broker.route === "missing" + ? { icon: "!", tone: "warning" } + : { icon: "◌", tone: "warning" }; + } if (state.binding) { const detail = conversationKindIcon(state.binding.conversationKind); return state.binding.paused diff --git a/openspec/changes/self-heal-broker-route-registration/.openspec.yaml b/openspec/changes/self-heal-broker-route-registration/.openspec.yaml new file mode 100644 index 0000000..5e6d53a --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-24 diff --git a/openspec/changes/self-heal-broker-route-registration/design.md b/openspec/changes/self-heal-broker-route-registration/design.md new file mode 100644 index 0000000..169212e --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/design.md @@ -0,0 +1,136 @@ +## Context + +PiRelay keeps live session routes only in the machine-local broker's memory. Each `BrokerTunnelRuntime` also keeps its own live `SessionRoute` objects and re-registers them when it connects. Existing recovery tests prove that `resyncRoutes()` works when tests explicitly reconnect clients, but they do not exercise autonomous recovery after a real shared-broker restart or the case where a client socket remains connected while the broker no longer has the corresponding route. + +During the Sigma incident, the Pi process was alive and idle, persisted Telegram authority was active, and a Unix client socket was open, yet the broker session list reported Sigma offline. Restarting the shared broker also showed that clients converged at different times and one session needed an extension reload/status interaction before all three routes were present again. Local `/relay status` could not expose the mismatch because it reported only local route and binding state. + +Investigation also found a concrete restart race. `restartBrokerProcess()` kills the PID read from shared control state and then unconditionally unlinks the shared pid/socket files outside the supervisor lock. During that wait, another disconnected client can acquire supervision and start a replacement broker; the restarting client can then delete the replacement broker's control files and start another process. Clients may split between an unreachable orphan broker and the broker named by the new pid/socket files. Reconnect scheduling can also be skipped when a socket closes while `connecting` is set, without an unconditional retry after that connection attempt settles. + +The design must preserve authorization-before-side-effect rules, persisted binding authority, one broker per scope, current namespace isolation, and bounded secret-safe state. It must not persist live route actions or full route descriptors because those contain process-local behavior and become stale across process lifetimes. + +## Goals / Non-Goals + +**Goals:** + +- Repair a missing broker registration automatically while the owning Pi session remains live. +- Make autonomous multi-client recovery after broker restart an exercised lifecycle, not a path that requires callers to invoke private connection helpers. +- Keep route ownership idempotent when old and replacement sockets overlap or disconnect out of order. +- Make local status and doctor output distinguish binding, transport, and broker-registration health. +- Produce enough opt-in, secret-safe evidence to diagnose registration loss and convergence failures. +- Bound retry rate, timers, payloads, and recovery time without weakening binding authority. + +**Non-Goals:** + +- Persisting executable route actions or reconstructing a live Pi route after the Pi process exits. +- Treating an active persisted binding as proof that a session is online. +- Starting an offline Pi process from the broker. +- Changing cross-machine federation, messenger authorization, pairing, or route selection semantics. +- Adding a new dependency or requiring a state-schema migration. + +## Decisions + +### 1. Add an idempotent client route-reconciliation request + +Each connected broker client will periodically send one bounded reconciliation request containing its `clientId` and serialized descriptors for its current live route map. The broker will atomically upsert those routes for the requesting socket, remove only registrations still owned by that same socket that are no longer in the client's declared set, and return a broker epoch plus the accepted route keys. + +The client will consider a route synchronized only after its key is acknowledged for the current broker epoch. Missing acknowledgements cause safe local unsynchronized state and a bounded retry; they do not cause prompt delivery, binding mutation, or optimistic online reporting. + +Reconciliation will run: + +- immediately after the socket connects, +- after a broker epoch change or reconnect, +- after local route add/remove or material route-state changes through the existing publish path, +- periodically with an unref'ed, jittered interval while live routes exist. + +The broker will continue applying persisted binding authority to every reconciled descriptor before storing or using its binding. + +**Alternative considered:** Send only a lightweight `hasRoute` heartbeat. Rejected because recovering a missing route would require a second protocol round trip and separate race handling, while the existing serialized route descriptor is already bounded and registration is idempotent. + +**Alternative considered:** Persist live route descriptors in `state.json`. Rejected because route actions and availability belong to a live client process; persisted online descriptors would create zombie routes and stale action authority. + +### 2. Use broker epochs and socket ownership for stale-event safety + +A broker process will generate a non-secret random epoch at startup. Reconciliation responses and client synchronization state are scoped to that epoch. A client clears synchronized status when its socket closes or an epoch changes. + +The broker route owner remains the concrete client socket. A replacement socket may take ownership by reconciling the same route. Later unregister or disconnect events from an older socket MUST NOT remove the replacement registration. Atomic reconciliation removes only routes whose current owner is the requesting socket. + +**Alternative considered:** Use timestamps to choose the newest registration. Rejected because clock order is unnecessary on one host and socket ownership provides a stronger causal boundary. + +### 3. Make broker replacement atomic before coordinating convergence + +Before an intentional restart, the initiating client will request a bounded restart snapshot from the old broker. The broker will return a non-secret expected connected-client count and notify connected clients that the broker is restarting. Clients retain their local route maps, clear current synchronization, and enter reconnecting state immediately. + +Kill, cleanup, and replacement startup will execute under the broker-scope supervisor lock. Cleanup will be conditional: the client may remove control files only when they still identify the broker PID/epoch selected for replacement. If another reconnecting client has already installed a live replacement broker, the restart path will join that broker instead of deleting its socket/pid files or spawning a competitor. A caller connected to an orphan or superseded broker will never kill the different live broker currently named by authoritative control state merely because both share a scope. + +The initiating client will stop the old broker, start or connect to the replacement, reconcile its routes, and wait for a bounded convergence result from broker health: expected client count, connected client count, and registered route count. Success requires the initiating route to be acknowledged and all still-running clients to have had an opportunity to reconnect; timeout reports partial convergence without marking missing routes online. + +Unexpected broker death follows the same autonomous reconnect and periodic reconciliation path but has no prior expected-client snapshot. Socket closure during an in-flight connection or resynchronization attempt will set a deferred reconnect requirement; after `connecting` settles, the client MUST either be connected and synchronized or have a bounded retry scheduled. + +The local restart message will state that the broker is shared across local sessions and report safe convergence counts or a partial-recovery warning. + +**Alternative considered:** Restart each messenger runtime or force `/reload` in every Pi session. Rejected because a broker restart must not require user coordination across sessions and extension reloads invalidate unrelated contexts. + +**Alternative considered:** Unconditionally remove the known socket and pid paths after the selected PID exits. Rejected because those paths are scope-owned mutable coordination state and may already belong to a replacement broker started by another client. + +### 4. Track three separate health dimensions locally + +The runtime will maintain bounded broker health state: + +1. local binding authority (`active`, `paused`, `revoked`, or absent), +2. broker transport (`connected`, `reconnecting`, or unavailable), +3. route registration (`synchronized`, `pending`, `missing/rejected`, or unknown) for the current broker epoch. + +`/relay status`, `/relay doctor`, and concise status-line rendering will use these dimensions. An active local binding with a missing broker route will never be described simply as connected. Doctor may trigger the same idempotent reconciliation request and report whether repair succeeded; it will not synthesize a route if the runtime has no live local route. + +Raw session keys, session files, socket paths, chat/user ids, tokens, and route payloads remain excluded from normal status UX. + +### 5. Extend opt-in communication diagnostics + +When communication diagnostics are enabled, clients and broker will record metadata-only events for connection open/close, broker epoch change, reconciliation request/ack/rejection, missing-route repair, intentional restart notice, convergence success/timeout, and retry scheduling. Fields are bounded to safe counts, outcomes, namespace labels, hashed or existing safe correlation identifiers, and durations. + +Diagnostics remain disabled by default and MUST NOT contain serialized route payloads, prompt/output content, messenger secrets, pairing data, or raw destination ids. + +### 6. Test real autonomous lifecycle sequences + +Tests will use real socket close/recreation and public runtime methods/timers rather than manually invoking private `ensureConnected()` as the recovery mechanism under test. Fake timers or configurable short intervals will keep periodic reconciliation deterministic. + +A multi-client integration sequence will cover: + +```text +broker epoch A: routes A, B, C + │ intentional or unexpected restart + ▼ +broker epoch B: clients reconnect independently + │ reconciliation acknowledgements + ▼ +routes A, B, C online without /reload or local commands +``` + +Separate tests will delete or omit one broker route while leaving its socket connected, then assert that periodic reconciliation restores it and session lists become online again. + +## Risks / Trade-offs + +- **[Periodic reconciliation increases local IPC and state checks]** → Use one bounded batch per client, jittered intervals, no timer when no routes exist, and idempotent broker handling. +- **[A malicious or buggy client repeatedly overwrites a route]** → Scope ownership to the socket/client, retain persisted binding authority, bound payload size/count, and preserve replacement-socket stale-disconnect checks. +- **[Restart convergence waits for a client that exited concurrently]** → Bound the wait and report partial convergence; do not block the replacement broker or claim missing clients are online. +- **[A reconnecting client starts a broker during intentional cleanup]** → Hold the supervisor lock across ownership verification and cleanup; compare PID/epoch before deletion and join any already-authoritative replacement. +- **[A close event occurs while connection setup suppresses reconnect scheduling]** → Persist a reconnect-needed flag and enforce the post-connection invariant that a started client is connected or has a retry timer. +- **[Status can become stale between acknowledgements]** → Clear synchronized state immediately on socket close/epoch change and display unknown/reconnecting conservatively. +- **[Reconciliation races with session replacement or disconnect]** → Serialize against local route-map mutations, snapshot descriptors per request, and ensure a later snapshot removes only routes still owned by that client socket. +- **[Always-on observability could expose identifiers]** → Keep detailed records within the existing opt-in diagnostics capability and expose only safe categorical health in normal UX. + +## Migration Plan + +1. Add protocol types and broker epoch/reconciliation handling while preserving existing `registerRoute`/`unregisterRoute` compatibility during the change. +2. Move client startup and reconnect resync onto the reconciliation primitive. +3. Add periodic reconciliation and health-state tracking. +4. Add coordinated restart snapshot/notice/convergence reporting. +5. Update local status, doctor, status-line copy, diagnostics, and documentation. +6. Remove redundant one-off resync code only after direct, reconnect, and restart paths share tested reconciliation behavior. + +Rollback is code-only: older clients continue using existing registration requests, and no persisted state migration is required. New clients must treat a broker that lacks reconciliation support as an explicit unsupported/legacy condition and fall back to existing reconnect registration without reporting enhanced convergence guarantees. + +## Open Questions + +- Choose the final default reconciliation interval and restart convergence timeout from tests and local IPC measurements; initial targets are 15–30 seconds and 10 seconds respectively. +- Decide whether normal status-line space permits a distinct `syncing` glyph or whether detailed broker health remains in `/relay status` and `/relay doctor` while the status line shows only a warning state. diff --git a/openspec/changes/self-heal-broker-route-registration/proposal.md b/openspec/changes/self-heal-broker-route-registration/proposal.md new file mode 100644 index 0000000..62e5bab --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/proposal.md @@ -0,0 +1,33 @@ +## Why + +A live Pi session can retain an active messenger binding and an open broker socket while its route is missing from the broker's in-memory registry, causing `/sessions` to report the session offline even though local `/relay status` appears healthy. The recent Sigma incident also showed that restarting the shared broker from one session does not reliably or observably converge every other local client without manual reloads. + +## What Changes + +- Add bounded automatic reconciliation so connected broker clients detect and repair missing or stale registrations for every live local route. +- Make broker restart and socket recovery converge all local clients without callers manually invoking connection internals or reloading Pi extensions. +- Make broker replacement atomic under supervisor ownership so a restarting client cannot delete the pid/socket files of a replacement broker started by another reconnecting client or split clients across orphan brokers. +- Define route registration ownership and generation rules so late disconnects, duplicate clients, and restart races cannot remove a newer registration. +- Expose broker socket and route-registration health through local status and doctor surfaces without raw destination identifiers or secrets. +- Record secret-safe broker connection, reconciliation, and restart-convergence diagnostics when communication diagnostics are enabled. +- Add real multi-client recovery tests covering autonomous reconnect, missing-route repair, restart races, delayed clients, and partial failures. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `relay-broker-topology`: Require autonomous route reconciliation, multi-client restart convergence, idempotent registration ownership, and bounded recovery behavior. +- `relay-runtime-status-line`: Distinguish local binding health from broker socket and route-registration health, and surface reconnecting or unsynchronized states safely. +- `relay-communication-diagnostics`: Trace broker client connection, route reconciliation, restart generation, and convergence outcomes with bounded secret-safe metadata. + +## Impact + +- Broker client and process code under `extensions/relay/broker/`, especially connection supervision, route registration, client ownership, and restart behavior. +- Extension runtime status and doctor reporting under `extensions/relay/runtime/`, `extensions/relay/commands/`, and status helpers. +- Communication diagnostic event contracts and troubleshooting documentation. +- Broker namespace, integration, runtime, and multi-client process tests. +- No state schema migration, messenger API change, new dependency, or authorization relaxation is intended. diff --git a/openspec/changes/self-heal-broker-route-registration/specs/relay-broker-topology/spec.md b/openspec/changes/self-heal-broker-route-registration/specs/relay-broker-topology/spec.md new file mode 100644 index 0000000..158f875 --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/specs/relay-broker-topology/spec.md @@ -0,0 +1,107 @@ +## ADDED Requirements + +### Requirement: Broker clients autonomously reconcile live route registrations +Broker clients SHALL reconcile every live local session route with the authoritative broker after connection and at a bounded interval while routes remain live. + +#### Scenario: Initial connection reconciles all client routes +- **WHEN** a broker client connects with one or more live local routes +- **THEN** it sends a bounded reconciliation snapshot for all routes it currently owns +- **AND** it does not report a route synchronized until the broker acknowledges that route for the current broker process epoch + +#### Scenario: Connected client route is missing from broker memory +- **WHEN** a client socket remains connected and the client retains a live route that is absent from the broker route registry +- **THEN** periodic reconciliation restores the route without requiring `/reload`, `/relay restart`, another local command, or messenger re-pairing +- **AND** broker-backed session lists report the route online after acknowledgement + +#### Scenario: Broker restarts unexpectedly +- **WHEN** a broker client loses its socket because the broker exits or its socket is recreated unexpectedly +- **THEN** the client clears synchronized route status, reconnects with bounded backoff, and reconciles every still-live route automatically +- **AND** it does not require callers or tests to invoke private connection helpers to recover + +#### Scenario: Socket closes during an in-flight connection attempt +- **WHEN** a broker socket closes while the client is connecting, resynchronizing routes, or otherwise suppressing duplicate reconnect timers +- **THEN** the client records that reconnect is still required +- **AND** after the in-flight attempt settles it is either connected and reconciled or has a bounded reconnect attempt scheduled + +#### Scenario: Client has no live routes +- **WHEN** a broker client has no live local routes +- **THEN** it does not run an unnecessary periodic route-reconciliation timer +- **AND** it does not keep the broker alive solely for an empty reconciliation snapshot + +#### Scenario: Reconciled descriptor contains stale binding metadata +- **WHEN** a reconciliation snapshot contains binding metadata that persisted binding authority marks revoked, moved, paused, missing, or destination-mismatched +- **THEN** the broker applies binding authority before accepting protected delivery state +- **AND** reconciliation does not resurrect, retarget, unpause, or otherwise weaken the persisted authority outcome + +### Requirement: Route reconciliation is atomic and ownership-safe +The broker SHALL process one client's route reconciliation snapshot atomically with respect to that client's socket ownership and SHALL reject stale ownership effects. + +#### Scenario: Replacement socket takes route ownership +- **WHEN** a replacement socket reconciles a route previously owned by an older socket for the same live client/session +- **THEN** the broker assigns the route to the replacement socket +- **AND** a later unregister, stale snapshot, or disconnect from the older socket does not remove the replacement registration + +#### Scenario: Client removes a route from its snapshot +- **WHEN** a connected client reconciles a snapshot that no longer contains a route previously owned by that same socket +- **THEN** the broker removes that socket's stale route registration +- **AND** it does not remove an equivalent route currently owned by another socket + +#### Scenario: Broker process epoch changes +- **WHEN** a client receives a reconciliation response from a new broker process epoch +- **THEN** it discards synchronization acknowledgements from the previous epoch +- **AND** it requires current-epoch acknowledgement before presenting its routes as synchronized + +#### Scenario: Reconciliation request is oversized or malformed +- **WHEN** a client sends a reconciliation snapshot exceeding configured route-count or payload bounds or containing invalid route descriptors +- **THEN** the broker rejects the invalid request with a secret-safe error +- **AND** it does not partially mutate route ownership or persisted binding state + +#### Scenario: Reconciliation repeatedly fails +- **WHEN** broker connection or reconciliation attempts repeatedly fail +- **THEN** the client retries with bounded backoff and jitter rather than a tight loop +- **AND** local Pi operation continues while broker-backed remote actions remain conservatively unavailable + +### Requirement: Intentional broker restart converges active local clients +An intentional restart of a shared machine-local broker SHALL coordinate bounded reconnection and route reconciliation for active clients in the same broker scope. + +#### Scenario: One session restarts a broker shared by multiple sessions +- **WHEN** a local session invokes `/relay restart` while multiple live Pi clients share that broker scope +- **THEN** the old broker provides a bounded non-secret restart snapshot and notifies connected clients that restart is beginning +- **AND** clients retain their local route maps, clear old synchronization state, and reconnect to the replacement broker automatically + +#### Scenario: All clients recover after restart +- **WHEN** all previously connected clients remain alive and can reach the replacement broker +- **THEN** each client reconciles its routes without manual `/reload` or unrelated local commands +- **AND** the replacement broker reports convergence with every recovered route present in broker-backed session lists + +#### Scenario: One client reconnects late +- **WHEN** one live client reconnects after other clients have already registered with the replacement broker +- **THEN** its late reconciliation adds its routes without replacing or removing routes owned by earlier clients +- **AND** convergence health updates when the late client is acknowledged + +#### Scenario: Restart convergence times out partially +- **WHEN** one or more expected clients do not reconnect before the bounded convergence timeout +- **THEN** the replacement broker remains available for clients that did recover +- **AND** `/relay restart` reports a safe partial-convergence warning with counts and repair guidance +- **AND** missing clients and routes remain offline rather than being synthesized from persisted bindings + +#### Scenario: Concurrent restart requests occur +- **WHEN** two clients request restart for the same broker scope concurrently +- **THEN** broker supervision serializes the restart lifecycle +- **AND** at most one replacement broker becomes authoritative +- **AND** both callers receive a consistent success, in-progress, or safe failure outcome without deleting the live replacement socket + +#### Scenario: Reconnecting client starts replacement during restart cleanup +- **WHEN** one client is waiting for the selected old broker PID to exit and another reconnecting client starts or discovers a replacement broker for the same scope +- **THEN** restart cleanup verifies the authoritative PID and broker epoch under the supervisor lock before deleting control files +- **AND** it joins the valid replacement instead of deleting its socket or pid file or spawning a competing broker + +#### Scenario: Caller is connected to a superseded broker +- **WHEN** a client connected to an orphaned or superseded broker requests restart while authoritative control state names a different live broker +- **THEN** the client does not kill the different authoritative broker based only on the shared scope paths +- **AND** it disconnects from the superseded process and converges on the authoritative broker safely + +#### Scenario: Restart leaves no orphan broker +- **WHEN** intentional restart completes or reports partial convergence +- **THEN** exactly one live broker owns the authoritative pid/socket files for the broker scope +- **AND** no displaced broker continues messenger ingress or retains reachable route ownership for that scope diff --git a/openspec/changes/self-heal-broker-route-registration/specs/relay-communication-diagnostics/spec.md b/openspec/changes/self-heal-broker-route-registration/specs/relay-communication-diagnostics/spec.md new file mode 100644 index 0000000..3603353 --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/specs/relay-communication-diagnostics/spec.md @@ -0,0 +1,29 @@ +## ADDED Requirements + +### Requirement: Broker reconciliation and restart convergence are diagnosable +PiRelay SHALL record bounded secret-safe broker client reconciliation and restart-convergence events when communication diagnostics are explicitly enabled. + +#### Scenario: Client connection lifecycle is traced +- **WHEN** a broker client connects, disconnects, begins reconnecting, observes a broker epoch change, or exhausts a retry interval +- **THEN** diagnostics record the event kind, outcome, bounded attempt/delay metadata, component, and safe broker-scope correlation +- **AND** they do not include socket paths, tokens, raw route descriptors, prompts, outputs, or destination ids + +#### Scenario: Route reconciliation is traced +- **WHEN** a client sends a reconciliation snapshot and the broker accepts, partially accepts, or rejects it +- **THEN** diagnostics record bounded route counts, acknowledgement counts, duration, trigger category, and outcome +- **AND** a missing-route repair is distinguishable from initial registration and ordinary route-state publication + +#### Scenario: Intentional restart convergence is traced +- **WHEN** `/relay restart` snapshots connected clients, notifies them, starts a replacement broker, and waits for recovery +- **THEN** diagnostics record restart phase, old/new epoch transition category, expected client count, recovered client and route counts, duration, and success or timeout outcome +- **AND** diagnostics remain bounded if multiple clients reconnect concurrently + +#### Scenario: Reconciliation is suppressed safely +- **WHEN** reconciliation does not run because the client is stopped, has no live routes, lacks a socket, is backing off, or has stale binding authority +- **THEN** diagnostics may record a deduplicated suppression category useful for troubleshooting +- **AND** they do not repeatedly emit unbounded records for the same timer condition + +#### Scenario: Doctor reports diagnostic correlation +- **WHEN** `/relay doctor` detects or repairs an unsynchronized route while communication diagnostics are enabled +- **THEN** it reports a safe event time or correlation label that helps locate the relevant diagnostic records +- **AND** it does not print raw diagnostic records or upload logs automatically diff --git a/openspec/changes/self-heal-broker-route-registration/specs/relay-runtime-status-line/spec.md b/openspec/changes/self-heal-broker-route-registration/specs/relay-runtime-status-line/spec.md new file mode 100644 index 0000000..49facac --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/specs/relay-runtime-status-line/spec.md @@ -0,0 +1,35 @@ +## ADDED Requirements + +### Requirement: Relay status distinguishes local binding from broker route health +PiRelay SHALL report local binding authority, broker transport health, and current-route registration health as separate states without exposing raw routing or destination identifiers. + +#### Scenario: Active binding and synchronized broker route +- **WHEN** the current session has an active binding, its broker transport is connected, and the broker acknowledged its route for the current broker epoch +- **THEN** local status may report the messenger as paired and broker-synchronized +- **AND** concise status-line rendering remains consistent with an online remotely reachable route + +#### Scenario: Active binding but broker route is missing +- **WHEN** the current session has an active local binding but its route is missing, rejected, or not yet acknowledged by the broker +- **THEN** `/relay status` and `/relay doctor` distinguish the active binding from the unsynchronized broker route +- **AND** the status line shows a warning, syncing, or unavailable state instead of implying fully connected remote control + +#### Scenario: Broker socket is reconnecting +- **WHEN** the broker client socket closes or the broker epoch changes +- **THEN** PiRelay clears prior synchronized-route status immediately +- **AND** local status reports reconnecting or unavailable until current-epoch reconciliation succeeds + +#### Scenario: Automatic reconciliation repairs status +- **WHEN** route reconciliation restores and acknowledges a previously missing route +- **THEN** PiRelay refreshes local status through a live extension context +- **AND** the warning or syncing state returns to paired/synchronized without requiring an unrelated user command + +#### Scenario: Doctor repairs a missing registration +- **WHEN** `/relay doctor` finds a live local route with active binding authority but no acknowledged broker registration +- **THEN** it may invoke the same idempotent reconciliation operation used by automatic recovery +- **AND** it reports whether repair succeeded, remains pending, or failed with actionable local guidance +- **AND** it does not synthesize a route when no live local route exists + +#### Scenario: Status output remains secret-safe +- **WHEN** local status or doctor reports broker transport, route synchronization, epoch change, restart convergence, or repair outcome +- **THEN** normal output excludes raw session keys, transcript paths, socket paths, bot tokens, token hashes, chat ids, user ids, and serialized route payloads +- **AND** it uses safe labels, categorical states, and bounded counts instead diff --git a/openspec/changes/self-heal-broker-route-registration/tasks.md b/openspec/changes/self-heal-broker-route-registration/tasks.md new file mode 100644 index 0000000..f1acee8 --- /dev/null +++ b/openspec/changes/self-heal-broker-route-registration/tasks.md @@ -0,0 +1,68 @@ +## 1. Reconciliation Protocol and Domain Model + +- [x] 1.1 Define typed broker epoch, route-reconciliation request/response, accepted-route, and broker-health contracts with bounded route counts and payload sizes. +- [x] 1.2 Add pure validation and normalization helpers for reconciliation snapshots and secret-safe rejection outcomes. +- [x] 1.3 Define client-side transport and per-route synchronization states scoped to a broker epoch. +- [x] 1.4 Add unit tests for valid, malformed, oversized, duplicate, and stale-epoch reconciliation payloads. + +## 2. Broker Route Reconciliation + +- [x] 2.1 Generate a non-secret broker epoch for each broker process lifetime and expose it only through local client protocol responses and diagnostics. +- [x] 2.2 Implement atomic reconciliation that upserts the requesting socket's declared routes after binding-authority validation. +- [x] 2.3 Remove omitted routes only when they are still owned by the requesting socket, preserving replacement-socket registrations. +- [x] 2.4 Return acknowledged and rejected route outcomes without raw route descriptors, destinations, prompts, outputs, or secrets. +- [x] 2.5 Add broker tests for missing-route restoration, stale binding stripping, omitted-route removal, malformed snapshot rollback, and replacement-socket ownership races. + +## 3. Autonomous Broker Client Recovery + +- [x] 3.1 Replace one-off reconnect resync with a shared reconciliation operation over the client's current live route map. +- [x] 3.2 Reconcile immediately after initial connection, reconnect, broker epoch change, and local route-map mutation. +- [x] 3.3 Add an unref'ed jittered periodic reconciliation timer that runs only while the client is started with live routes. +- [x] 3.4 Clear current-epoch synchronization immediately on socket close, broker epoch change, route removal, or reconciliation rejection. +- [x] 3.5 Persist reconnect-needed state when close occurs during `connecting`, and enforce that every started client is connected or has a bounded retry scheduled after the attempt settles. +- [x] 3.6 Retry connection and reconciliation failures with bounded backoff while preserving local Pi operation and conservative offline routing. +- [x] 3.7 Ensure stop, explicit disconnect, session replacement, and runtime reload cancel timers and cannot re-register removed routes. +- [x] 3.8 Add client unit tests for timer lifecycle, autonomous reconnect, epoch reset, delayed acknowledgement, overlapping close/connect, backoff, and stop/remove races. + +## 4. Shared Broker Restart Convergence + +- [x] 4.1 Add an intentional-restart preparation request that captures bounded expected client/route counts and notifies connected clients without persisting live descriptors. +- [x] 4.2 Make notified clients retain live route maps, clear synchronization, and enter autonomous reconnect before the old socket closes. +- [x] 4.3 Move broker kill, ownership verification, control-file cleanup, and replacement startup under broker-scope supervisor serialization. +- [x] 4.4 Make pid/socket cleanup compare the selected old PID/epoch with current authoritative control state and join any replacement already started by another client. +- [x] 4.5 Prevent a client connected to a superseded broker from killing the different live broker named by authoritative scope state. +- [x] 4.6 Reconcile the initiating client's routes and poll bounded broker health for expected client/route convergence. +- [x] 4.7 Return explicit complete, partial-timeout, in-progress, and failed restart outcomes while keeping recovered routes available. +- [x] 4.8 Update `/relay restart` copy to explain shared-machine impact and report secret-safe convergence counts and repair guidance. +- [x] 4.9 Add multi-client process tests for complete recovery, reconnect-during-cleanup, delayed/exited clients, concurrent restart callers, no orphan brokers, and no-manual-reload behavior. + +## 5. Broker-Aware Status and Repair + +- [x] 5.1 Track local binding authority, broker transport, broker epoch, and per-route synchronization as separate runtime health fields. +- [x] 5.2 Extend `/relay status` with safe connected, reconnecting, synchronized, pending, missing/rejected, and unknown broker-route states. +- [x] 5.3 Update concise Telegram, Discord, and Slack status-line rendering so an active binding with an unsynchronized route does not appear fully connected. +- [x] 5.4 Extend `/relay doctor` to detect active-binding/missing-route mismatches and invoke idempotent reconciliation only for a proven live local route. +- [x] 5.5 Refresh status through the latest matching live extension context after reconciliation success, failure, disconnect, or epoch change. +- [x] 5.6 Add status and doctor tests for healthy, missing, reconnecting, repaired, stale-context, no-live-route, and secret-redaction outcomes. + +## 6. Communication Diagnostics and Documentation + +- [x] 6.1 Add opt-in metadata-only events for broker connect/disconnect, epoch changes, reconciliation triggers, acknowledgements, rejection, repair, and retry scheduling. +- [x] 6.2 Add restart-phase and convergence diagnostics with bounded expected/recovered counts, durations, outcomes, and safe broker-scope correlation. +- [x] 6.3 Deduplicate periodic suppression/failure diagnostics so reconciliation timers cannot create unbounded diagnostic noise. +- [x] 6.4 Update broker troubleshooting documentation with the active-binding/missing-route symptom, status interpretation, automatic recovery, and safe diagnostics workflow. +- [x] 6.5 Update README and testing guidance for shared restart behavior and expected autonomous multi-session convergence. + +## 7. End-to-End Regression Coverage + +- [x] 7.1 Add a real-socket regression that removes a broker route while its client socket remains connected and verifies periodic reconciliation restores online session listing. +- [x] 7.2 Add a three-client broker restart regression that races autonomous reconnect against restart cleanup, relies only on public runtime behavior, and verifies one authoritative broker plus every route without `/reload`, `/relay status`, or private `ensureConnected()` calls. +- [x] 7.3 Cover identifiers present and missing, repeated reconciliation, partial and delayed acknowledgements, stale disconnects, authorization denial, binding revocation, and state-unavailable behavior. +- [x] 7.4 Verify reconciliation and restart recovery do not duplicate prompt injection, completion delivery, progress timers, approvals, lifecycle notifications, or messenger ingress ownership. +- [x] 7.5 Run a structured pre-review focused on broker lifecycle, authorization, stale ownership, bounded timers, adapter parity, diagnostics, and restart failure modes. + +## 8. Validation + +- [x] 8.1 Run `npm run typecheck`. +- [x] 8.2 Run `npm test`. +- [x] 8.3 Run `openspec validate self-heal-broker-route-registration --strict`. diff --git a/tests/broker-namespace.test.ts b/tests/broker-namespace.test.ts index 998026b..822c1b9 100644 --- a/tests/broker-namespace.test.ts +++ b/tests/broker-namespace.test.ts @@ -1,7 +1,7 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { createServer, type Server, type Socket } from "node:net"; +import { createServer, Socket, type Server } from "node:net"; import { afterEach, describe, expect, it, vi } from "vitest"; import { BrokerTunnelRuntime } from "../extensions/relay/broker/tunnel-runtime.js"; import type { SessionRoute, TelegramTunnelConfig } from "../extensions/relay/core/types.js"; @@ -62,6 +62,134 @@ describe("broker namespace isolation", () => { await runtime.stop(); }); + it("periodically reconciles live routes without a local command", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-periodic-")); + tempDirs.push(stateDir); + const messages: Array> = []; + const runtime = new BrokerTunnelRuntime(config(stateDir), { reconciliationIntervalMs: 20 }); + await listenJsonBroker(socketPath(runtime), messages); + + await runtime.start(); + await runtime.registerRoute(route("periodic-session")); + const initialCount = messages.filter((message) => message.action === "reconcileRoutes").length; + + await waitForCondition(() => messages.filter((message) => message.action === "reconcileRoutes").length > initialCount); + await new Promise((resolve) => setTimeout(resolve, 30)); + + await runtime.stop(); + const stoppedCount = messages.filter((message) => message.action === "reconcileRoutes").length; + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(messages.filter((message) => message.action === "reconcileRoutes")).toHaveLength(stoppedCount); + }); + + it("autonomously reconnects and republishes routes after socket loss", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-autonomous-reconnect-")); + tempDirs.push(stateDir); + const firstMessages: Array> = []; + const runtime = new BrokerTunnelRuntime(config(stateDir), { reconciliationIntervalMs: 20 }); + const firstServer = await listenJsonBroker(socketPath(runtime), firstMessages); + await runtime.start(); + await runtime.registerRoute(route("autonomous-session")); + + for (const socket of sockets.splice(0)) socket.destroy(); + await closeServer(firstServer); + await rm(socketPath(runtime), { force: true }); + const recoveredMessages: Array> = []; + await listenJsonBroker(socketPath(runtime), recoveredMessages); + + await waitForCondition(() => recoveredMessages.some((message) => message.action === "registerRoute") && runtime.getBrokerSynchronizationState().routes["autonomous-session"] === "synchronized", 3_000); + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ transport: "connected", routes: { "autonomous-session": "synchronized" } }); + await runtime.stop(); + }); + + it("records reconciliation diagnostics without route payloads or secrets", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-diagnostics-")); + tempDirs.push(stateDir); + const logPath = join(stateDir, "communication.jsonl"); + const runtime = new BrokerTunnelRuntime({ + ...config(stateDir), + communicationDiagnostics: { enabled: true, logPath, maxFileBytes: 100_000, maxFiles: 2, includeContentPreview: false, previewChars: 0, redactionPatterns: [] }, + }); + await listenJsonBroker(socketPath(runtime), [], { brokerEpoch: "diagnostic-epoch" }); + const sensitiveRoute = route("secret-session-key"); + sensitiveRoute.notification.lastAssistantText = "sensitive assistant output"; + await runtime.registerRoute(sensitiveRoute); + await waitForCondition(async () => readFile(logPath, "utf8").then((value) => value.includes("broker.reconcile"), () => false)); + const diagnostics = await readFile(logPath, "utf8"); + expect(diagnostics).toContain("broker.reconcile"); + expect(diagnostics).not.toContain("secret-session-key"); + expect(diagnostics).not.toContain("sensitive assistant output"); + expect(diagnostics).not.toContain(config(stateDir).botToken); + await runtime.stop(); + }); + + it("bounds a delayed reconciliation request instead of leaving recovery stuck", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-timeout-")); + tempDirs.push(stateDir); + const runtime = new BrokerTunnelRuntime(config(stateDir), { requestTimeoutMs: 25 }); + await listenJsonBroker(socketPath(runtime), [], { dropReconciliationResponse: true }); + + await expect(runtime.registerRoute(route("timeout-session"))).rejects.toThrow("Broker reconcileRoutes request timed out."); + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ transport: "connected", routes: { "timeout-session": "pending" } }); + await runtime.stop(); + }); + + it("serializes overlapping reconciliation and route-map mutations", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-overlap-")); + tempDirs.push(stateDir); + const messages: Array> = []; + const runtime = new BrokerTunnelRuntime(config(stateDir)); + await listenJsonBroker(socketPath(runtime), messages, { brokerEpoch: "epoch-1", reconciliationDelayMs: 30 }); + await runtime.registerRoute(route("first-overlap")); + + const earlier = runtime.reconcileBrokerRoutes(); + const mutation = runtime.registerRoute(route("second-overlap")); + await Promise.all([earlier, mutation]); + + const snapshots = messages.filter((message) => message.action === "reconcileRoutes").map((message) => (message.routes as Array<{ sessionKey: string }>).map((entry) => entry.sessionKey)); + expect(snapshots.at(-1)).toEqual(["first-overlap", "second-overlap"]); + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ routes: { "first-overlap": "synchronized", "second-overlap": "synchronized" } }); + await runtime.stop(); + }); + + it("keeps routes pending until delayed acknowledgement and resets them for a new epoch", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-epoch-")); + tempDirs.push(stateDir); + const messages: Array> = []; + const broker = { brokerEpoch: "epoch-1", reconciliationDelayMs: 30 }; + const runtime = new BrokerTunnelRuntime(config(stateDir)); + await listenJsonBroker(socketPath(runtime), messages, broker); + const pendingRegistration = runtime.registerRoute(route("epoch-session")); + expect(runtime.getBrokerSynchronizationState().routes["epoch-session"]).toBe("pending"); + await pendingRegistration; + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ brokerEpoch: "epoch-1", routes: { "epoch-session": "synchronized" } }); + + broker.brokerEpoch = "epoch-2"; + const pendingEpoch = runtime.reconcileBrokerRoutes(); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(runtime.getBrokerSynchronizationState().routes["epoch-session"]).toBe("synchronized"); + await pendingEpoch; + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ brokerEpoch: "epoch-2", routes: { "epoch-session": "synchronized" } }); + await runtime.stop(); + }); + + it("falls back to legacy registration when an existing broker lacks reconciliation", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-legacy-reconcile-")); + tempDirs.push(stateDir); + const runtime = new BrokerTunnelRuntime(config(stateDir)); + const messages: Array> = []; + await listenJsonBroker(socketPath(runtime), messages, { rejectReconciliation: true }); + + await runtime.start(); + await runtime.registerRoute(route("legacy-session")); + + expect(messages.some((message) => message.action === "reconcileRoutes")).toBe(true); + expect(messages.filter((message) => message.action === "reconcileRoutes")).toHaveLength(1); + expect(messages.some((message) => message.action === "registerRoute" && (message.route as { sessionKey?: string } | undefined)?.sessionKey === "legacy-session")).toBe(true); + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ routes: { "legacy-session": "synchronized" } }); + await runtime.stop(); + }); + it("re-registers routes from multiple clients after broker socket recovery", async () => { const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-multi-reconnect-")); tempDirs.push(stateDir); @@ -99,6 +227,66 @@ describe("broker namespace isolation", () => { await secondRuntime.stop(); }); + it("tracks reconciliation rejection without failing the live runtime", () => { + const runtime = new BrokerTunnelRuntime(config("/tmp/pirelay-reconciliation-rejection")); + const rejectedRoute = route("rejected-session"); + (runtime as unknown as { routes: Map }).routes.set(rejectedRoute.sessionKey, rejectedRoute); + + expect(() => (runtime as unknown as { applyReconciliationResponse(value: unknown): boolean }).applyReconciliationResponse({ + brokerEpoch: "epoch-1", + acceptedSessionKeys: [], + rejected: [{ index: 0, code: "invalid-request", safeMessage: "Invalid route." }], + })).not.toThrow(); + expect(runtime.getBrokerSynchronizationState()).toMatchObject({ routes: { "rejected-session": "rejected" } }); + }); + + it("does not connect or spawn a broker only to unregister the final route", async () => { + const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-final-unregister-")); + tempDirs.push(stateDir); + const runtime = new BrokerTunnelRuntime(config(stateDir)); + const finalRoute = route("final-session"); + (runtime as unknown as { routes: Map }).routes.set(finalRoute.sessionKey, finalRoute); + const ensureConnected = vi.fn(async () => { throw new Error("must not connect"); }); + (runtime as unknown as { ensureConnected(): Promise }).ensureConnected = ensureConnected; + + await runtime.unregisterRoute(finalRoute.sessionKey); + + expect(ensureConnected).not.toHaveBeenCalled(); + }); + + it("ignores a stale socket close after a replacement socket is attached", () => { + const runtime = new BrokerTunnelRuntime(config("/tmp/pirelay-stale-close")); + const stale = new Socket(); + const replacement = new Socket(); + (runtime as unknown as { attachSocket(socket: Socket): void }).attachSocket(stale); + (runtime as unknown as { attachSocket(socket: Socket): void }).attachSocket(replacement); + + stale.emit("close"); + + expect((runtime as unknown as { socket?: Socket }).socket).toBe(replacement); + expect(runtime.getBrokerSynchronizationState().transport).toBe("connected"); + stale.destroy(); + replacement.destroy(); + }); + + it("ignores late restart notices from a superseded client socket", async () => { + const runtime = new BrokerTunnelRuntime(config("/tmp/pirelay-stale-client-socket")); + const current = new Socket(); + const stale = new Socket(); + (runtime as unknown as { socket?: Socket }).socket = current; + const responses: Array> = []; + (runtime as unknown as { writeMessage(message: Record): void }).writeMessage = (message) => { responses.push(message); }; + + await (runtime as unknown as { handleMessage(line: string, sourceSocket?: Socket): Promise }).handleMessage(JSON.stringify({ + type: "request", requestId: "stale-restart", action: "brokerRestartNotice", + }), stale); + + expect(responses).toEqual([]); + expect(current.destroyed).toBe(false); + current.destroy(); + stale.destroy(); + }); + it("bridges stale skill command lookups as an empty metadata list", async () => { const stateDir = await mkdtemp(join(shortSocketTmpdir(), "pirelay-broker-skills-")); tempDirs.push(stateDir); @@ -152,7 +340,7 @@ function socketPath(runtime: BrokerTunnelRuntime): string { return (runtime as unknown as { socketPath: string }).socketPath; } -async function listenJsonBroker(path: string, messages: Array>): Promise { +async function listenJsonBroker(path: string, messages: Array>, options: { rejectReconciliation?: boolean; brokerEpoch?: string; reconciliationDelayMs?: number; dropReconciliationResponse?: boolean } = {}): Promise { const server = createServer((socket) => { sockets.push(socket); socket.setEncoding("utf8"); @@ -166,7 +354,17 @@ async function listenJsonBroker(path: string, messages: Array; messages.push(message); - socket.write(`${JSON.stringify({ type: "response", requestId: message.requestId, ok: true })}\n`); + if (message.action === "reconcileRoutes" && options.dropReconciliationResponse) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const unsupported = options.rejectReconciliation && message.action === "reconcileRoutes"; + const result = message.action === "reconcileRoutes" && options.brokerEpoch + ? { brokerEpoch: options.brokerEpoch, acceptedSessionKeys: (message.routes as Array<{ sessionKey: string }> | undefined)?.map((route) => route.sessionKey) ?? [], rejected: [] } + : undefined; + const respond = () => socket.write(`${JSON.stringify({ type: "response", requestId: message.requestId, ok: !unsupported, result, error: unsupported ? "Unknown client action: reconcileRoutes" : undefined })}\n`); + if (message.action === "reconcileRoutes" && options.reconciliationDelayMs) setTimeout(respond, options.reconciliationDelayMs); + else respond(); } newlineIndex = buffer.indexOf("\n"); } @@ -189,10 +387,10 @@ function closeServer(server: Server): Promise { return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); } -async function waitForCondition(predicate: () => boolean, timeoutMs = 2_000): Promise { +async function waitForCondition(predicate: () => boolean | Promise, timeoutMs = 2_000): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { - if (predicate()) return; + if (await predicate()) return; await new Promise((resolve) => setTimeout(resolve, 25)); } throw new Error("Timed out waiting for condition."); diff --git a/tests/broker-process.test.ts b/tests/broker-process.test.ts index 6f66af8..57067e9 100644 --- a/tests/broker-process.test.ts +++ b/tests/broker-process.test.ts @@ -122,6 +122,65 @@ describe("telegram broker process", () => { expect(updated.bindings?.["revoked-session:memory"]).toMatchObject({ status: "revoked", revokedAt: revokedBinding.revokedAt, lastSeenAt: revokedBinding.lastSeenAt }); }); + it("reconciles route snapshots atomically with binding and replacement-socket safety", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "pirelay-broker-reconcile-")); + tempDirs.push(stateDir); + const statePath = join(stateDir, "state.json"); + const activeBinding = { + sessionKey: "live-session:memory", sessionId: "live-session", sessionLabel: "Live", chatId: 123, userId: 456, + boundAt: new Date(0).toISOString(), lastSeenAt: new Date(1).toISOString(), status: "active", + }; + const revokedBinding = { + sessionKey: "revoked-session:memory", sessionId: "revoked-session", sessionLabel: "Revoked", chatId: 123, userId: 456, + boundAt: new Date(0).toISOString(), lastSeenAt: new Date(2).toISOString(), revokedAt: new Date(3).toISOString(), status: "revoked", + }; + await writeFile(statePath, JSON.stringify({ pendingPairings: {}, bindings: { [activeBinding.sessionKey]: activeBinding, [revokedBinding.sessionKey]: revokedBinding }, channelBindings: {} })); + const socketPath = join(stateDir, "broker.sock"); + const brokerPath = fileURLToPath(new URL("../extensions/relay/broker/entry.js", import.meta.url)); + const child = spawn(process.execPath, [brokerPath], { env: { + ...process.env, + TELEGRAM_TUNNEL_BROKER_SOCKET_PATH: socketPath, + TELEGRAM_TUNNEL_BROKER_CONFIG_JSON: JSON.stringify({ botToken: "123456:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456", stateDir, pollingTimeoutSeconds: 1 }), + TELEGRAM_TUNNEL_BROKER_SKIP_POLLING: "1", + } }); + children.push(child); + await waitForSocket(socketPath, child); + const first = await openBrokerClient(socketPath); + const second = await openBrokerClient(socketPath); + const reconcile = (clientId: string, routes: Record[]) => ({ type: "request", action: "reconcileRoutes", clientId, routes }); + const liveRoute = brokerRoute(activeBinding); + const revokedRoute = brokerRoute({ ...revokedBinding, status: "active", revokedAt: undefined }); + const missingBinding = { ...activeBinding, sessionKey: "missing-session:memory", sessionId: "missing-session", sessionLabel: "Missing" }; + const missingRoute = brokerRoute(missingBinding); + + const initial = await first.request(reconcile("first", [liveRoute, revokedRoute, missingRoute])) as { acceptedSessionKeys?: string[]; rejected?: unknown[] }; + expect(initial.acceptedSessionKeys).toEqual([activeBinding.sessionKey, revokedBinding.sessionKey, missingBinding.sessionKey]); + expect(initial.rejected).toEqual([]); + expect(await first.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 3 }); + const persistedAfterReconciliation = (JSON.parse(await readFile(statePath, "utf8")) as { bindings: Record }).bindings; + expect(persistedAfterReconciliation[revokedBinding.sessionKey]?.status).toBe("revoked"); + expect(persistedAfterReconciliation[missingBinding.sessionKey]).toBeUndefined(); + + await first.request(reconcile("first", [liveRoute])); + expect(await first.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 1 }); + const malformed = await first.request(reconcile("first", [liveRoute, { sessionKey: "invalid" }])) as { acceptedSessionKeys?: string[]; rejected?: Array<{ index?: number }> }; + expect(malformed.acceptedSessionKeys).toEqual([]); + expect(malformed.rejected?.[0]?.index).toBe(1); + expect(await first.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 1 }); + + await second.request(reconcile("second", [liveRoute])); + const stale = await first.request(reconcile("first", [liveRoute])) as { acceptedSessionKeys?: string[]; rejected?: Array<{ code?: string }> }; + expect(stale.acceptedSessionKeys).toEqual([]); + expect(stale.rejected).toEqual([{ index: 0, code: "stale-owner", safeMessage: "A newer broker client owns this route." }]); + await first.close(); + expect(await second.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 1 }); + await second.request(reconcile("second", [])); + expect(await second.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 0 }); + await second.request(reconcile("second", [liveRoute])); + expect(await second.request({ type: "request", action: "getBrokerHealth" })).toMatchObject({ registeredRouteCount: 1 }); + await second.close(); + }); + it("preserves concurrent store updates while registering routes", async () => { const stateDir = await mkdtemp(join(tmpdir(), "pirelay-broker-process-")); tempDirs.push(stateDir); @@ -293,7 +352,8 @@ describe("telegram broker process", () => { children.push(child); await waitForSocket(socketPath, child); - await sendBrokerRequest(socketPath, { + const client = await openBrokerClient(socketPath); + await client.request({ type: "request", requestId: "corrupt-state-route", action: "registerRoute", @@ -317,7 +377,14 @@ describe("telegram broker process", () => { }, }); + const reconciliation = await client.request({ + type: "request", requestId: "corrupt-state-reconcile", action: "reconcileRoutes", clientId: "test-client", + routes: [{ channel: "telegram", sessionKey: "corrupt-session:memory", sessionId: "corrupt-session", sessionLabel: "Corrupt Docs", busy: false, notification: {} }], + }) as { acceptedSessionKeys?: string[]; rejected?: Array<{ code?: string; safeMessage?: string }> }; + expect(reconciliation.acceptedSessionKeys).toEqual([]); + expect(reconciliation.rejected).toEqual([{ code: "invalid-request", safeMessage: "Broker binding authority is temporarily unavailable." }]); expect(await readFile(statePath, "utf8")).toBe("{not-json"); + await client.close(); }); it("hydrates registered routes with persisted Telegram bindings when the client route is stale", async () => { @@ -1511,6 +1578,7 @@ function waitForSocket(socketPath: string, child: ChildProcessWithoutNullStreams function brokerRoute(binding: Record): Record { return { sessionKey: binding.sessionKey, + channel: "telegram", sessionId: binding.sessionId, sessionLabel: binding.sessionLabel, online: true, diff --git a/tests/broker-restart-convergence.test.ts b/tests/broker-restart-convergence.test.ts new file mode 100644 index 0000000..3615664 --- /dev/null +++ b/tests/broker-restart-convergence.test.ts @@ -0,0 +1,214 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createConnection } from "node:net"; +import { afterEach, describe, expect, it } from "vitest"; +import { BrokerTunnelRuntime } from "../extensions/relay/broker/tunnel-runtime.js"; +import type { SessionRoute, TelegramTunnelConfig } from "../extensions/relay/core/types.js"; + +const dirs: string[] = []; +const runtimes: BrokerTunnelRuntime[] = []; + +afterEach(async () => { + await Promise.all(runtimes.splice(0).map((runtime) => runtime.stop().catch(() => undefined))); + await new Promise((resolve) => setTimeout(resolve, 100)); + await Promise.all(dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +describe("shared broker restart convergence", () => { + it("repairs a missing route while the owning client socket remains connected", async () => { + const stateDir = await mkdtemp(join(tmpdir().length <= 40 ? tmpdir() : "/tmp", "pirelay-broker-route-repair-")); + dirs.push(stateDir); + const previousSkip = process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = "1"; + try { + const config = testConfig(stateDir); + const target = new BrokerTunnelRuntime(config, { reconciliationIntervalMs: 100 }); + const keeper = new BrokerTunnelRuntime(config, { reconciliationIntervalMs: 100 }); + runtimes.push(target, keeper); + await Promise.all([target.start(), keeper.start()]); + await target.registerRoute(route("target-session")); + await keeper.registerRoute(route("keeper-session")); + const socketPath = (target as unknown as { socketPath: string }).socketPath; + const thief = await openClient(socketPath); + await thief.request({ type: "request", action: "reconcileRoutes", clientId: "replacement", routes: [serializedRoute("target-session")] }); + await thief.close(); + + await waitFor(async () => { + const observer = await openClient(socketPath); + try { + const health = await observer.request({ type: "request", action: "getBrokerHealth" }) as { registeredRouteCount?: number }; + return health.registeredRouteCount === 2 && target.getBrokerSynchronizationState().routes["target-session"] === "synchronized"; + } finally { + await observer.close(); + } + }); + expect(target.getBrokerSynchronizationState()).toMatchObject({ transport: "connected", routes: { "target-session": "synchronized" } }); + } finally { + if (previousSkip === undefined) delete process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + else process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = previousSkip; + } + }, 20_000); + + it("reports bounded partial convergence when an expected client exits", async () => { + const stateDir = await mkdtemp(join(tmpdir().length <= 40 ? tmpdir() : "/tmp", "pirelay-broker-partial-")); + dirs.push(stateDir); + const previousSkip = process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = "1"; + try { + const config = testConfig(stateDir); + const initiator = new BrokerTunnelRuntime(config, { reconciliationIntervalMs: 25, restartConvergenceTimeoutMs: 300 }); + const exiting = new BrokerTunnelRuntime(config, { reconciliationIntervalMs: 25 }); + runtimes.push(initiator, exiting); + await Promise.all([initiator.start(), exiting.start()]); + await initiator.registerRoute(route("remaining-session")); + await exiting.registerRoute(route("exiting-session")); + + let stoppedAfterNotice: Promise | undefined; + exiting.setBrokerSynchronizationListener((state) => { + if (state.transport === "reconnecting" && !stoppedAfterNotice) stoppedAfterNotice = exiting.stop(); + }); + const restarting = initiator.restartBrokerProcess(); + await waitFor(async () => Boolean(stoppedAfterNotice)); + await stoppedAfterNotice; + const result = await restarting; + + expect(result).toMatchObject({ status: "partial-timeout", expectedClientCount: 2, connectedClientCount: 1, registeredRouteCount: 1 }); + expect(initiator.getBrokerSynchronizationState().routes["remaining-session"]).toBe("synchronized"); + } finally { + if (previousSkip === undefined) delete process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + else process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = previousSkip; + } + }, 20_000); + + it("recovers three clients and routes without private reconnect helpers", async () => { + const stateDir = await mkdtemp(join(tmpdir().length <= 40 ? tmpdir() : "/tmp", "pirelay-broker-convergence-")); + dirs.push(stateDir); + const previousSkip = process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = "1"; + try { + const config = testConfig(stateDir); + const clients = [0, 1, 2].map(() => new BrokerTunnelRuntime(config, { reconciliationIntervalMs: 25 })); + runtimes.push(...clients); + await Promise.all(clients.map((client) => client.start())); + await Promise.all(clients.map((client, index) => client.registerRoute(route(`session-${index + 1}`)))); + + const restarting = clients[0]!.restartBrokerProcess(); + const concurrent = await clients[0]!.restartBrokerProcess(); + const result = await restarting; + + expect(concurrent).toMatchObject({ status: "in-progress" }); + expect(result).toMatchObject({ status: "complete", expectedClientCount: 3, connectedClientCount: 3, registeredRouteCount: 3 }); + await waitFor(async () => clients.every((client, index) => client.getBrokerSynchronizationState().routes[`session-${index + 1}`] === "synchronized")); + for (let index = 0; index < clients.length; index += 1) { + expect(clients[index]!.getBrokerSynchronizationState()).toMatchObject({ + transport: "connected", + routes: { [`session-${index + 1}`]: "synchronized" }, + }); + } + const pidPath = (clients[0] as unknown as { pidPath: string }).pidPath; + const pid = Number((await readFile(pidPath, "utf8")).trim()); + expect(Number.isInteger(pid) && pid > 0).toBe(true); + } finally { + if (previousSkip === undefined) delete process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING; + else process.env.TELEGRAM_TUNNEL_BROKER_SKIP_POLLING = previousSkip; + } + }, 20_000); +}); + +function route(sessionKey: string): SessionRoute { + return { + sessionKey, + sessionId: sessionKey, + sessionLabel: sessionKey, + notification: { lastStatus: "idle" }, + actions: { + context: { isIdle: () => true } as never, + getModel: () => undefined, + sendUserMessage: () => undefined, + getLatestImages: async () => [], + getImageByPath: async () => ({ ok: false, error: "not-found" }), + appendAudit: () => undefined, + persistBinding: () => undefined, + promptLocalConfirmation: async () => true, + abort: () => undefined, + compact: async () => undefined, + }, + }; +} + +function testConfig(stateDir: string): TelegramTunnelConfig { + return { + botToken: "123456:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456", + stateDir, + pairingExpiryMs: 300_000, + busyDeliveryMode: "followUp", + allowUserIds: [], + summaryMode: "deterministic", + maxTelegramMessageChars: 3900, + sendRetryCount: 1, + sendRetryBaseMs: 1, + pollingTimeoutSeconds: 1, + redactionPatterns: [], + maxInboundImageBytes: 1024, + maxOutboundImageBytes: 1024, + maxLatestImages: 4, + allowedImageMimeTypes: ["image/png"], + }; +} + +function serializedRoute(sessionKey: string): Record { + return { channel: "telegram", sessionKey, sessionId: sessionKey, sessionLabel: sessionKey, busy: false, notification: {} }; +} + +async function openClient(socketPath: string): Promise<{ request(payload: Record): Promise; close(): Promise }> { + const socket = createConnection(socketPath); + await new Promise((resolve, reject) => { socket.once("connect", resolve); socket.once("error", reject); }); + socket.setEncoding("utf8"); + let buffer = ""; + const pending = new Map(); + socket.on("data", (chunk: string) => { + buffer += chunk; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + const line = buffer.slice(0, newline).trim(); + buffer = buffer.slice(newline + 1); + if (line) { + const message = JSON.parse(line) as { type?: string; requestId?: string; ok?: boolean; result?: unknown; error?: string }; + if (message.type === "request" && message.requestId) { + socket.write(`${JSON.stringify({ type: "response", requestId: message.requestId, ok: true })}\n`); + } else if (message.requestId) { + const waiter = pending.get(message.requestId); + if (waiter) { + pending.delete(message.requestId); + if (message.ok) waiter.resolve(message.result); + else waiter.reject(new Error(message.error ?? "Broker request failed.")); + } + } + } + newline = buffer.indexOf("\n"); + } + }); + return { + request(payload) { + const requestId = `${Date.now()}-${Math.random()}`; + return new Promise((resolve, reject) => { + pending.set(requestId, { resolve, reject }); + socket.write(`${JSON.stringify({ ...payload, requestId })}\n`); + }); + }, + close() { + if (socket.destroyed) return Promise.resolve(); + return new Promise((resolve) => { socket.once("close", resolve); socket.end(); }); + }, + }; +} + +async function waitFor(predicate: () => Promise, timeoutMs = 3_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error("Timed out waiting for broker route repair."); +} diff --git a/tests/integration.test.ts b/tests/integration.test.ts index 87a4546..7acf4ac 100644 --- a/tests/integration.test.ts +++ b/tests/integration.test.ts @@ -1426,11 +1426,61 @@ describe("PiRelay integration behavior", () => { expect(fakeSlackRuntime.stop).toHaveBeenCalledTimes(1); expect(fakeRuntime.registerRoute).toHaveBeenCalledTimes(2); expect(fakeSlackRuntime.start).toHaveBeenCalledTimes(2); - expect(notifications.at(-1)?.message).toBe("PiRelay broker process and runtimes restarted for this session."); + expect(notifications.at(-1)?.message).toBe("Shared PiRelay broker and runtimes restarted for this session."); expect(statuses).toContainEqual({ key: "relay", value: "tg ◌" }); expect(statuses).toContainEqual({ key: "slack-relay", value: "sl ◌" }); }); + it("reports and repairs broker route health without exposing routing identifiers", async () => { + const config = await createRuntimeConfig("pi-relay-broker-health-"); + vi.stubEnv("TELEGRAM_BOT_TOKEN", config.botToken); + vi.stubEnv("PI_TELEGRAM_TUNNEL_STATE_DIR", config.stateDir); + let registeredRoute: SessionRoute | undefined; + let health = "missing" as "missing" | "synchronized"; + const fakeRuntime: TunnelRuntime = { + setup: undefined, + start: vi.fn(async () => undefined), + stop: vi.fn(async () => undefined), + ensureSetup: vi.fn(async () => ({ botId: 123456, botUsername: "pi_test_bot", botDisplayName: "Pi Test Bot", validatedAt: new Date().toISOString() })), + registerRoute: vi.fn(async (route) => { + registeredRoute = route; + route.binding = createBinding(route.sessionId, 987654, 456789); + route.binding.sessionKey = route.sessionKey; + await new TunnelStateStore(config.stateDir).upsertBinding(route.binding); + }), + unregisterRoute: vi.fn(async () => undefined), + getStatus: vi.fn(() => undefined), + getBrokerSynchronizationState: vi.fn(() => ({ transport: "connected" as const, routes: registeredRoute ? { [registeredRoute.sessionKey]: health } : {} })), + reconcileBrokerRoutes: vi.fn(async () => { + health = "synchronized"; + return { transport: "connected" as const, routes: registeredRoute ? { [registeredRoute.sessionKey]: health } : {} }; + }), + sendToBoundChat: vi.fn(async () => undefined), + }; + vi.doMock("../extensions/relay/adapters/telegram/runtime.js", () => ({ + getOrCreateTunnelRuntime: () => fakeRuntime, + sendSessionNotification: vi.fn(async () => undefined), + })); + const { default: relayExtension } = await import("../extensions/relay/index.js"); + const pi = createMockPi(); + const { context, notifications } = createMockContext("broker-health-session"); + relayExtension(pi.api as any); + await pi.emit("session_start", {}, context); + + await pi.runCommand("relay", "status", context); + const status = notifications.at(-1)?.message ?? ""; + expect(status).toContain("Binding: active"); + expect(status).toContain("Broker transport: connected"); + expect(status).toContain("Broker route: missing"); + expect(status).not.toContain("987654"); + expect(status).not.toContain("456789"); + expect(status).not.toContain(registeredRoute!.sessionKey); + + await pi.runCommand("relay", "doctor", context); + expect(fakeRuntime.reconcileBrokerRoutes).toHaveBeenCalledTimes(1); + expect(notifications.at(-1)?.message).toContain("Broker route repair: synchronized."); + }); + it("does not show Slack ready when required credentials are incomplete", async () => { const config = await createRuntimeConfig("pi-slack-incomplete-status-"); await writeFile(config.configPath!, JSON.stringify({ diff --git a/tests/relay/broker-reconciliation.test.ts b/tests/relay/broker-reconciliation.test.ts new file mode 100644 index 0000000..8f331ce --- /dev/null +++ b/tests/relay/broker-reconciliation.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from "vitest"; +import { + normalizeBrokerReconciliationRequest, + normalizeBrokerReconciliationResponse, + synchronizationStateAfterConnect, + synchronizationStateAfterDisconnect, + synchronizationStateAfterResponse, +} from "../../extensions/relay/broker/reconciliation.js"; +import { MAX_BROKER_RECONCILIATION_ROUTES } from "../../extensions/relay/broker/protocol.js"; +import type { RelayRouteState } from "../../extensions/relay/core/relay-core.js"; + +function route(sessionKey = "session-1"): RelayRouteState { + return { + channel: "telegram", + sessionKey, + sessionId: sessionKey, + sessionLabel: sessionKey, + busy: false, + notification: {}, + }; +} + +describe("broker route reconciliation contracts", () => { + it("normalizes a bounded valid route snapshot", () => { + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", routes: [route()] }, "epoch-1")).toEqual({ + ok: true, + request: { clientId: "client-1", observedBrokerEpoch: undefined, routes: [route()] }, + }); + }); + + it.each([ + undefined, + {}, + { clientId: "", routes: [] }, + { clientId: "client-1", routes: "not-an-array" }, + { clientId: "client-1", routes: [{ ...route(), busy: "no" }] }, + { clientId: "client-1", routes: [{ ...route(), notification: null }] }, + ])("rejects malformed snapshots without reflecting input", (input) => { + const result = normalizeBrokerReconciliationRequest(input, "epoch-1"); + expect(result).toMatchObject({ ok: false, rejection: { code: "invalid-request" } }); + if (!result.ok) expect(result.rejection.safeMessage).not.toContain("session-1"); + }); + + it("accepts bounded non-Telegram channel descriptors", () => { + const discordRoute = { ...route(), channel: "discord" as const }; + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", routes: [discordRoute] }, "epoch-1")).toMatchObject({ ok: true }); + }); + + it("rejects snapshots over the route count limit", () => { + const routes = Array.from({ length: MAX_BROKER_RECONCILIATION_ROUTES + 1 }, (_, index) => route(`session-${index}`)); + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", routes }, "epoch-1")).toMatchObject({ + ok: false, + rejection: { code: "too-many-routes" }, + }); + }); + + it("rejects oversized payloads before accepting descriptors", () => { + const oversized = route(); + oversized.notification = { lastAssistantText: "x".repeat(300_000) }; + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", routes: [oversized] }, "epoch-1")).toMatchObject({ + ok: false, + rejection: { code: "payload-too-large" }, + }); + }); + + it("rejects duplicate routes atomically", () => { + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", routes: [route(), route()] }, "epoch-1")).toMatchObject({ + ok: false, + rejection: { index: 1, code: "duplicate-route" }, + }); + }); + + it("rejects stale observed broker epochs", () => { + expect(normalizeBrokerReconciliationRequest({ clientId: "client-1", observedBrokerEpoch: "epoch-old", routes: [route()] }, "epoch-new")).toMatchObject({ + ok: false, + rejection: { code: "stale-epoch" }, + }); + }); + + it("normalizes only bounded typed reconciliation responses", () => { + expect(normalizeBrokerReconciliationResponse({ brokerEpoch: "epoch-1", acceptedSessionKeys: ["session-1"], rejected: [] })).toEqual({ brokerEpoch: "epoch-1", acceptedSessionKeys: ["session-1"], rejected: [] }); + expect(normalizeBrokerReconciliationResponse({ brokerEpoch: "epoch-1", acceptedSessionKeys: [], rejected: [{ index: 0, code: "stale-owner", safeMessage: "A newer broker client owns this route." }] })).toMatchObject({ rejected: [{ code: "stale-owner" }] }); + expect(normalizeBrokerReconciliationResponse({ brokerEpoch: "epoch-1", acceptedSessionKeys: ["session-1", "session-1"], rejected: [] })).toBeUndefined(); + expect(normalizeBrokerReconciliationResponse({ brokerEpoch: "epoch-1", acceptedSessionKeys: [], rejected: [{ code: "unknown", safeMessage: "no" }] })).toBeUndefined(); + }); + + it("tracks transport and per-route health within the acknowledged epoch", () => { + const pending = synchronizationStateAfterConnect(["session-1", "session-2"]); + expect(pending).toEqual({ transport: "connected", brokerEpoch: undefined, routes: { "session-1": "pending", "session-2": "pending" } }); + + const synchronized = synchronizationStateAfterResponse(["session-1", "session-2"], { + brokerEpoch: "epoch-2", + acceptedSessionKeys: ["session-1"], + rejected: [], + }); + expect(synchronized).toEqual({ transport: "connected", brokerEpoch: "epoch-2", routes: { "session-1": "synchronized", "session-2": "missing" } }); + + expect(synchronizationStateAfterDisconnect(["session-1", "session-2"])).toEqual({ + transport: "reconnecting", + routes: { "session-1": "unknown", "session-2": "unknown" }, + }); + }); + + it("maps indexed rejections back to request-order routes", () => { + expect(synchronizationStateAfterResponse(["session-1", "session-2"], { + brokerEpoch: "epoch-2", + acceptedSessionKeys: ["session-1"], + rejected: [{ index: 1, code: "invalid-request", safeMessage: "Invalid route." }], + })).toMatchObject({ routes: { "session-1": "synchronized", "session-2": "rejected" } }); + }); + + it("marks unacknowledged routes rejected for a general safe rejection", () => { + expect(synchronizationStateAfterResponse(["session-1"], { + brokerEpoch: "epoch-2", + acceptedSessionKeys: [], + rejected: [{ code: "invalid-request", safeMessage: "Invalid broker reconciliation request." }], + })).toMatchObject({ routes: { "session-1": "rejected" } }); + }); +}); diff --git a/tests/relay/broker-supervisor.test.ts b/tests/relay/broker-supervisor.test.ts index 69dc3e9..96621ad 100644 --- a/tests/relay/broker-supervisor.test.ts +++ b/tests/relay/broker-supervisor.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import { mkdtemp, readFile, stat, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { tmpdir } from "node:os"; -import { brokerControlPaths, brokerScopeControlPaths, cleanupStaleBrokerControlFiles, ensureLocalBroker, ensureScopedBroker, normalizeBrokerNamespace, readBrokerPid } from "../../extensions/relay/broker/index.js"; +import { brokerControlPaths, brokerScopeControlPaths, cleanupStaleBrokerControlFiles, ensureLocalBroker, ensureScopedBroker, normalizeBrokerNamespace, readBrokerPid, replaceScopedBroker } from "../../extensions/relay/broker/index.js"; describe("local broker supervisor", () => { it("starts a broker when no live pid exists", async () => { @@ -187,6 +187,85 @@ describe("local broker supervisor", () => { expect(starts).toBe(0); }); + it("joins a replacement instead of killing a different authoritative broker", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "pirelay-supervisor-replace-")); + const paths = brokerScopeControlPaths({ stateDir, tokenHash: "abc123" }); + await writeFile(paths.pidPath, "2222\n", { mode: 0o600 }); + let stopped: number | undefined; + let starts = 0; + + const result = await replaceScopedBroker({ + stateDir, + tokenHash: "abc123", + selectedPid: 1111, + isAlive: (pid) => pid === 2222, + probeSocket: async () => true, + stopBroker: async (pid) => { stopped = pid; }, + startBroker: async () => { starts += 1; return { pid: 3333 }; }, + waitForSocketReady: async () => undefined, + }); + + expect(result).toMatchObject({ status: "joined", pid: 2222 }); + expect(stopped).toBeUndefined(); + expect(starts).toBe(0); + expect(await readBrokerPid(paths.pidPath)).toBe(2222); + }); + + it("does not signal a live pid when the scoped broker socket is not reachable", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "pirelay-supervisor-unproven-pid-")); + const paths = brokerScopeControlPaths({ stateDir, tokenHash: "abc123" }); + await writeFile(paths.pidPath, "1111\n", { mode: 0o600 }); + let stopped = false; + const replacement = replaceScopedBroker({ + stateDir, tokenHash: "abc123", selectedPid: 1111, + isAlive: () => true, probeSocket: async () => false, + stopBroker: async () => { stopped = true; }, + startBroker: async () => ({ pid: 2222 }), waitForSocketReady: async () => undefined, + }); + await expect(replacement).rejects.toThrow("refusing unsafe replacement"); + expect(stopped).toBe(false); + }); + + it("does not replace a reused pid belonging to a different broker epoch", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "pirelay-supervisor-epoch-")); + const paths = brokerScopeControlPaths({ stateDir, tokenHash: "abc123" }); + await writeFile(paths.pidPath, "1111\n", { mode: 0o600 }); + await writeFile(paths.epochPath, "new-epoch\n", { mode: 0o600 }); + let stopped = false; + const result = await replaceScopedBroker({ + stateDir, tokenHash: "abc123", selectedPid: 1111, selectedEpoch: "old-epoch", + isAlive: () => true, probeSocket: async () => true, + stopBroker: async () => { stopped = true; }, + startBroker: async () => ({ pid: 2222 }), waitForSocketReady: async () => undefined, + }); + expect(result).toMatchObject({ status: "joined", pid: 1111 }); + expect(stopped).toBe(false); + }); + + it("serializes replacement cleanup and publishes one authoritative pid", async () => { + const stateDir = await mkdtemp(join(tmpdir(), "pirelay-supervisor-replace-")); + const paths = brokerScopeControlPaths({ stateDir, tokenHash: "abc123" }); + await writeFile(paths.pidPath, "1111\n", { mode: 0o600 }); + let livePid = 1111; + let nextPid = 2000; + let starts = 0; + const replace = () => replaceScopedBroker({ + stateDir, + tokenHash: "abc123", + selectedPid: 1111, + isAlive: (pid) => pid === livePid, + probeSocket: async () => true, + stopBroker: async (pid) => { if (livePid === pid) livePid = 0; }, + startBroker: async () => { starts += 1; livePid = ++nextPid; return { pid: livePid }; }, + waitForSocketReady: async () => undefined, + }); + + const [first, second] = await Promise.all([replace(), replace()]); + expect(starts).toBe(1); + expect([first.status, second.status].sort()).toEqual(["joined", "replaced"]); + expect(await readBrokerPid(paths.pidPath)).toBe(livePid); + }); + it("reuses and cleans up only the selected namespace", async () => { const stateDir = await mkdtemp(join(tmpdir(), "pirelay-supervisor-")); const alpha = brokerControlPaths(stateDir, "alpha"); diff --git a/tests/relay/status-line.test.ts b/tests/relay/status-line.test.ts index 1dc4b31..897cc9b 100644 --- a/tests/relay/status-line.test.ts +++ b/tests/relay/status-line.test.ts @@ -16,6 +16,12 @@ describe("relay status line formatting", () => { expect(formatRelayStatusLine({ channel: "slack", configured: true, runtimeStarted: true, binding: { conversationKind: "C123456" } })).toBe("sl ●"); }); + it("does not present an active binding as reachable while broker synchronization is unhealthy", () => { + expect(formatRelayStatusLine({ channel: "telegram", configured: true, binding: {}, broker: { transport: "reconnecting", route: "unknown" } })).toBe("tg ◌"); + expect(formatRelayStatusLine({ channel: "telegram", configured: true, binding: {}, broker: { transport: "connected", route: "missing" } })).toBe("tg !"); + expect(formatRelayStatusLine({ channel: "telegram", configured: true, binding: {}, broker: { transport: "connected", route: "synchronized" } })).toBe("tg ●"); + }); + it("applies status tones to the full segment", () => { const colorize = (tone: string, text: string) => `<${tone}>${text}`;