/**
 * Telegram bus follower runtime
 * Zones: multi-instance bus, follower lifecycle, manual registration
 * Owns this Pi instance's follower-side bus behavior: manual registration,
 * heartbeat, forwarded-update receiving, and follower-routed API calls.
 * It must not spawn Pi processes or create hidden Telegram-originated instances.
 */

import { basename } from "node:path";

import * as Sync from "./sync.ts";
import * as Threads from "./threads.ts";
import { parseTelegramUpdateJournalQueueOwner } from "./journal.ts";
import type { TelegramLockEntry, TelegramLockState } from "./locks.ts";
import type {
  TelegramQueueHandoffPayload,
  TelegramQueueHandoffStageResult,
} from "./queue.ts";
import type { TelegramTarget } from "./target.ts";
import {
  isTelegramApiMethodRetrySafe,
  TelegramApiCommitUnknownError,
  TelegramApiStaleTargetError,
} from "./telegram-api.ts";
import {
  createTelegramBusFollowerDeliveryIdentity,
  createTelegramBusFollowerTargetController,
  createTelegramBusForeignOwnedUpdateForwarder,
  createTelegramBusLocalServer,
  createTelegramBusRequestIdFactory,
  createUnauthorizedBusAck,
  getTelegramBusProtocolCompatibility,
  getTelegramBusSocketPath,
  isTelegramBusEnvelopeAuthorized,
  resolveTelegramBusSocketPath,
  sendTelegramBusLocalEnvelope,
  type TelegramBusAgentMessage,
  type TelegramBusAgentTargetSelector,
  type TelegramBusEnvelope,
  type TelegramBusProtocolIdentity,
  type TelegramBusSocketPathSource,
} from "./bus.ts";
import {
  getTelegramBusTransportRetryPolicy,
  TELEGRAM_BUS_REGISTRATION_RETRY,
} from "./bus-transport.ts";

export const TELEGRAM_BUS_FOLLOWER_PROMOTION_GRACE_MS = 2_500;
export const TELEGRAM_FOLLOWER_SESSION_HANDOFF_TTL_MS = 30_000;
export const TELEGRAM_BUS_FOLLOWER_CLIENT_TIMEOUT_MS = 30_000;
export const TELEGRAM_BUS_FOLLOWER_REGISTRATION_WAIT_MS = 30_000;
export const TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_ATTEMPTS =
  TELEGRAM_BUS_REGISTRATION_RETRY.attempts;
export const TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_DELAY_MS =
  TELEGRAM_BUS_REGISTRATION_RETRY.delayMs;

const TELEGRAM_FOLLOWER_SESSION_HANDOFF_KEY =
  "__piTelegramFollowerSessionHandoff";

export interface TelegramFollowerSessionHandoff {
  pid: number;
  instanceId: string;
  createdAtMs: number;
  target: TelegramTarget;
  slot?: string;
  threadName?: string;
}

export function getTelegramFollowerSessionHandoff():
  TelegramFollowerSessionHandoff | undefined {
  const value = (globalThis as Record<string, unknown>)[
    TELEGRAM_FOLLOWER_SESSION_HANDOFF_KEY
  ];
  if (!value || typeof value !== "object") return undefined;
  const handoff = value as Partial<TelegramFollowerSessionHandoff>;
  if (
    typeof handoff.pid !== "number" ||
    typeof handoff.instanceId !== "string" ||
    typeof handoff.createdAtMs !== "number" ||
    !handoff.target ||
    typeof handoff.target !== "object" ||
    typeof handoff.target.chatId !== "number"
  ) {
    return undefined;
  }
  return handoff as TelegramFollowerSessionHandoff;
}

export function setTelegramFollowerSessionHandoff(
  handoff: TelegramFollowerSessionHandoff | undefined,
): void {
  const store = globalThis as Record<string, unknown>;
  if (!handoff) delete store[TELEGRAM_FOLLOWER_SESSION_HANDOFF_KEY];
  else store[TELEGRAM_FOLLOWER_SESSION_HANDOFF_KEY] = handoff;
}

export function isTelegramFollowerSessionHandoffFresh(
  handoff: TelegramFollowerSessionHandoff | undefined,
  options: { pid?: number; nowMs?: number; ttlMs?: number } = {},
): handoff is TelegramFollowerSessionHandoff {
  if (!handoff) return false;
  const pid = options.pid ?? process.pid;
  const nowMs = options.nowMs ?? Date.now();
  const ttlMs = options.ttlMs ?? TELEGRAM_FOLLOWER_SESSION_HANDOFF_TTL_MS;
  return handoff.pid === pid && nowMs - handoff.createdAtMs <= ttlMs;
}

export interface TelegramBusFollowerRegistrationRuntime<TContext> {
  registerWithLeader: (
    ctx: TContext,
    leader: { busSocketPath?: string; busSecret?: string },
    options?: { target?: TelegramTarget; previousInstanceId?: string },
  ) => Promise<boolean>;
  setContext: (ctx: TContext) => void;
  disconnectFromLeader?: () => Promise<boolean>;
  stop: () => void;
}

export interface TelegramBusFollowerSessionReplacementSuspenderDeps {
  registrationState: Pick<
    TelegramBusFollowerRegistrationState,
    "isRegistered" | "getTarget" | "getSlot" | "getThreadName"
  >;
  instanceId: string;
  suspendPolling: () => Promise<void>;
  isLeader?: () => boolean;
  getLeaderBinding?: () => TelegramBusFollowerPromotedBinding | undefined;
  getActiveContext?: () => { cwd?: string } | undefined;
  getActiveProfileName?: () => string | undefined;
  recordRuntimeEvent: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
  getNowMs?: () => number;
  getPid?: () => number;
}

export interface TelegramBusFollowerSessionRefreshHookDeps<TContext> {
  registrationState: Pick<TelegramBusFollowerRegistrationState, "isRegistered">;
  registrationRuntime: Pick<
    TelegramBusFollowerRegistrationRuntime<TContext>,
    "registerWithLeader" | "setContext"
  >;
  getLeaderState: () => TelegramLockState;
  isSessionActive?: (ctx: TContext) => boolean;
  updateStatus: (ctx: TContext) => void;
  recordRuntimeEvent: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
}

export type TelegramBusFollowerControlLifecyclePhase = "electing";

export interface TelegramBusFollowerControlState {
  getActiveAuthSecret: () => string | undefined;
  setActiveAuthSecret: (secret: string | undefined) => void;
  getLifecyclePhase: () => TelegramBusFollowerControlLifecyclePhase | undefined;
  setLifecyclePhase: (
    phase: TelegramBusFollowerControlLifecyclePhase | undefined,
  ) => void;
}

export interface TelegramBusFollowerRegistrationState {
  isRegistered: () => boolean;
  getTarget: () => TelegramTarget | undefined;
  getSlot: () => string | undefined;
  getThreadName: () => string | undefined;
  getGeneration: () => string | undefined;
  beginRecovery: () => number;
  cancelRecovery: () => void;
  waitForGeneration: (timeoutMs?: number) => Promise<string | undefined>;
  getLeaderProtocol: () => TelegramBusProtocolIdentity | undefined;
  getEligibleElectionSlots: () => readonly string[];
  setEligibleElectionSlots: (slots: readonly string[]) => void;
  setRegistered: (
    registered: boolean,
    target?: TelegramTarget,
    metadata?: {
      slot?: string;
      threadName?: string;
      generation?: string;
      leaderProtocol?: TelegramBusProtocolIdentity;
    },
  ) => void;
}

export interface TelegramBusForwardedUpdateReceiverRuntime {
  start: () => Promise<void>;
  stop: () => Promise<void>;
}

export interface TelegramBusFollowerDurableAdmissionResult {
  deliveryId: string;
  sourceUpdateId: number;
}

export interface TelegramBusFollowerDurableAdmissionPort<TContext> {
  admit: (
    envelope: Extract<
      TelegramBusEnvelope,
      {
        kind:
          | "leader.forwardCallback"
          | "leader.forwardReaction"
          | "leader.forwardMessage"
          | "leader.forwardEditedMessage";
      }
    >,
    ctx: TContext,
  ) => Promise<TelegramBusFollowerDurableAdmissionResult>;
}

export interface TelegramBusFollowerClientRuntimeDeps<TMessage = unknown> {
  socketPath: TelegramBusSocketPathSource;
  instanceId: string;
  getApiAuthSecret?: () => string | undefined;
  getForwardingAuthSecret?: () => string | undefined;
  getRegistrationGeneration: () => string | undefined;
  waitForRegistrationGeneration?: (
    timeoutMs?: number,
  ) => Promise<string | undefined>;
  getForwardCommentBatchPosition?: (
    message: TMessage,
  ) => "comment" | "forward" | undefined;
  recordRuntimeEvent?: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
  timeoutMs?: number;
}

export interface TelegramBusFollowerApiCallerDeps {
  socketPath: TelegramBusSocketPathSource;
  instanceId: string;
  createRequestId: () => string;
  getAuthSecret?: () => string | undefined;
  getRegistrationGeneration: () => string | undefined;
  waitForRegistrationGeneration?: (
    timeoutMs?: number,
  ) => Promise<string | undefined>;
  getNowMs?: () => number;
  timeoutMs?: number;
}

export interface TelegramBusFollowerRegistrationRuntimeDeps<
  TContext extends { cwd?: string },
> {
  instanceId: string;
  createRequestId: () => string;
  protocolIdentity: TelegramBusProtocolIdentity;
  getLeaderAuthSecret?: (leader: { busSecret?: string }) => string | undefined;
  setActiveAuthSecret?: (secret: string | undefined) => void;
  followerBusSocketPath?: string;
  getFollowerBusSocketPath?: () => string;
  getLeaderSocketPath?: () => string;
  startReceiving?: () => Promise<void>;
  stopReceiving?: () => Promise<void> | void;
  registrationState?: TelegramBusFollowerRegistrationState;
  isContextActive?: (ctx: TContext) => boolean;
  getProfileKey?: (ctx: TContext) => string | undefined;
  getThreadName?: (ctx: TContext) => string | undefined;
  getNowMs?: () => number;
  getPid?: () => number;
  getProcessBirthId?: () => string;
  getSessionGeneration?: () => number;
  timeoutMs?: number;
  registrationTimeoutMs?: number;
  registrationRetryAttempts?: number;
  registrationRetryDelayMs?: number;
  heartbeatMs?: number;
  recordRuntimeEvent?: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
  onHeartbeatFailure?: (error: unknown, ctx: TContext) => Promise<void> | void;
  onRegistered?: (ctx: TContext) => Promise<void> | void;
}

export function createTelegramManualFollowerProfileKeyResolver(input: {
  getActiveProfileName: () => string | undefined;
  manualFollowerOwnerId: string;
}): () => string {
  return () =>
    Threads.getTelegramThreadOwnerKey({
      kind: "manual-follower",
      instanceId: input.manualFollowerOwnerId,
      telegramProfile: input.getActiveProfileName(),
    });
}

export interface TelegramBusFollowerElection {
  expectedOwner?: TelegramLockEntry;
}

export type TelegramBusFollowerPromotionHandler<TContext> = (
  ctx: TContext,
  binding: TelegramBusFollowerPromotedBinding,
  election: TelegramBusFollowerElection,
) => Promise<boolean>;

export function createTelegramBusFollowerPromotionHandler<
  TContext extends { cwd: string },
>(input: {
  topicTargetStore: Threads.TelegramTopicTargetStore;
  instanceId: string;
  getActiveProfileName: () => string | undefined;
  startLeader: (
    ctx: TContext,
    election: TelegramBusFollowerElection,
    onAcquired: () => Promise<void>,
  ) => Promise<boolean>;
  recordRuntimeEvent: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
  getNowMs?: () => number;
  getPid?: () => number;
}): TelegramBusFollowerPromotionHandler<TContext> {
  return async (ctx, binding, election) => {
    const promoted = await input.startLeader(ctx, election, async () => {
      const promotedRecord =
        await Threads.promoteTelegramFollowerBindingToLeader({
          store: input.topicTargetStore,
          instanceId: input.instanceId,
          cwd: ctx.cwd,
          telegramProfile: input.getActiveProfileName(),
          target: binding.target,
          slot: binding.slot,
          threadName: binding.threadName,
        });
      if (promotedRecord) {
        input.recordRuntimeEvent(
          "bus",
          "Follower thread binding promoted to leader",
          {
            phase: "follower-promoted-binding",
            chatId: promotedRecord.target.chatId,
            threadId: promotedRecord.target.threadId,
            slot: promotedRecord.slot,
            threadName: promotedRecord.threadName,
          },
        );
      }
    });
    if (promoted && typeof binding.target?.threadId === "number") {
      const profileKey = Threads.getTelegramThreadOwnerKey({
        kind: "leader",
        cwd: ctx.cwd,
        instanceId: input.instanceId,
        telegramProfile: input.getActiveProfileName(),
      });
      Threads.setTelegramLeaderSessionHandoff({
        pid: input.getPid?.() ?? process.pid,
        instanceId: input.instanceId,
        createdAtMs: input.getNowMs?.() ?? Date.now(),
        profileKey,
        target: {
          chatId: binding.target.chatId,
          threadId: binding.target.threadId,
        },
        slot: binding.slot,
        threadName: binding.threadName,
      });
      input.recordRuntimeEvent(
        "bus",
        "Promoted leader binding retained for session replacement",
        {
          phase: "follower-promoted-session-handoff",
          chatId: binding.target.chatId,
          threadId: binding.target.threadId,
          slot: binding.slot,
          threadName: binding.threadName,
        },
      );
    }
    return promoted;
  };
}

export interface TelegramBusFollowerTargetReplacementHandlerDeps<TContext> {
  topicTargetStore: Pick<
    Threads.TelegramTopicTargetStore,
    "load" | "list" | "markStaleByTarget" | "upsert" | "persist"
  >;
  registrationState: Pick<
    TelegramBusFollowerRegistrationState,
    "getTarget" | "setRegistered"
  > &
    Partial<Pick<TelegramBusFollowerRegistrationState, "getGeneration">>;
  instanceId: string;
  getManualFollowerProfileKey: () => string;
  manualFollowerOwnerId: string;
  getSyncState: () => Sync.TelegramSyncState;
  setSyncState: (state: Sync.TelegramSyncState) => void;
  getNowMs?: () => number;
  updateStatus: (ctx: TContext) => void;
  recordRuntimeEvent?: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
}

export type TelegramBusFollowerLeaderState =
  | { kind: "inactive" }
  | { kind: "active-here"; lock: TelegramBusFollowerLeaderLock }
  | { kind: "active-elsewhere"; lock: TelegramBusFollowerLeaderLock }
  | { kind: "stale"; lock: TelegramBusFollowerLeaderLock };

export type TelegramBusFollowerLeaderLock = TelegramLockEntry;

export interface TelegramBusFollowerPromotedBinding {
  target?: TelegramTarget;
  slot?: string;
  threadName?: string;
}

export interface TelegramBusFollowerHeartbeatRecoveryHandlerDeps<TContext> {
  registrationState: Pick<
    TelegramBusFollowerRegistrationState,
    | "getTarget"
    | "getSlot"
    | "getThreadName"
    | "getEligibleElectionSlots"
    | "beginRecovery"
    | "setRegistered"
  >;
  getRegistrationRuntime: () => TelegramBusFollowerRegistrationRuntime<TContext>;
  getLeaderState: () => TelegramBusFollowerLeaderState;
  setLifecyclePhase: (phase: "electing" | undefined) => void;
  updateStatus: (ctx: TContext) => void;
  promoteToLeader: (
    ctx: TContext,
    binding: TelegramBusFollowerPromotedBinding,
    election: TelegramBusFollowerElection,
  ) => Promise<boolean>;
  sleep?: (ms: number) => Promise<void>;
  scheduleRetry?: (retry: () => void, delayMs: number) => void;
  getActiveContext?: () => TContext | undefined;
  promotionGraceMs?: number;
  recordRuntimeEvent: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
}

export interface TelegramBusForwardedUpdateReceiverRuntimeDeps<TContext> {
  socketPath: TelegramBusSocketPathSource;
  instanceId: string;
  getAuthSecret?: () => string | undefined;
  getRegistrationGeneration: () => string | undefined;
  getRecipientBindingKey: () => string | undefined;
  durableAdmission: TelegramBusFollowerDurableAdmissionPort<TContext>;
  getContext: () => TContext | undefined;
  handleQueueHandoff?: (
    envelope: Extract<
      TelegramBusEnvelope,
      { kind: "leader.offerQueueHandoff" }
    >,
    ctx: TContext,
  ) => Promise<TelegramQueueHandoffStageResult> | TelegramQueueHandoffStageResult;
  handleReplaceTarget?: (
    input: {
      target: TelegramTarget & { threadId: number };
      oldTarget?: TelegramTarget & { threadId: number };
      reason: "thread-restore";
    },
    ctx: TContext,
  ) => Promise<void> | void;
  recordRuntimeEvent?: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
}

export interface TelegramBusFollowerRuntimeAssembly<TContext> {
  receiver: TelegramBusForwardedUpdateReceiverRuntime;
  registration: TelegramBusFollowerRegistrationRuntime<TContext>;
}

export interface TelegramBusFollowerRuntimeAssemblyPorts<
  TContext extends { cwd?: string },
> {
  instanceId: string;
  registrationState: TelegramBusFollowerRegistrationState;
  recordRuntimeEvent: (
    category: string,
    error: unknown,
    details?: Record<string, unknown>,
  ) => void;
  receiver: Omit<
    TelegramBusForwardedUpdateReceiverRuntimeDeps<TContext>,
    | "handleReplaceTarget"
    | "instanceId"
    | "recordRuntimeEvent"
    | "getRegistrationGeneration"
  >;
  targetReplacement: Omit<
    TelegramBusFollowerTargetReplacementHandlerDeps<TContext>,
    "registrationState" | "instanceId" | "recordRuntimeEvent"
  >;
  recovery: Omit<
    TelegramBusFollowerHeartbeatRecoveryHandlerDeps<TContext>,
    "getRegistrationRuntime" | "registrationState" | "recordRuntimeEvent"
  >;
  registration: Omit<
    TelegramBusFollowerRegistrationRuntimeDeps<TContext>,
    | "startReceiving"
    | "stopReceiving"
    | "onHeartbeatFailure"
    | "instanceId"
    | "registrationState"
    | "recordRuntimeEvent"
    | "protocolIdentity"
  > & {
    protocolIdentity: TelegramBusProtocolIdentity;
  };
}

export function createTelegramBusFollowerRuntimeAssembly<
  TContext extends { cwd?: string },
>(
  ports: TelegramBusFollowerRuntimeAssemblyPorts<TContext>,
): TelegramBusFollowerRuntimeAssembly<TContext> {
  const sharedRuntimeDeps = {
    instanceId: ports.instanceId,
    registrationState: ports.registrationState,
    recordRuntimeEvent: ports.recordRuntimeEvent,
  };
  const receiver = createTelegramBusForwardedUpdateReceiverRuntime({
    ...ports.receiver,
    instanceId: ports.instanceId,
    recordRuntimeEvent: ports.recordRuntimeEvent,
    getRegistrationGeneration: ports.registrationState.getGeneration,
    handleReplaceTarget: createTelegramBusFollowerTargetReplacementHandler({
      ...ports.targetReplacement,
      ...sharedRuntimeDeps,
    }),
  });
  let registration: TelegramBusFollowerRegistrationRuntime<TContext>;
  const recovery = createTelegramBusFollowerHeartbeatRecoveryHandler({
    ...ports.recovery,
    registrationState: ports.registrationState,
    recordRuntimeEvent: ports.recordRuntimeEvent,
    getRegistrationRuntime: () => registration,
  });
  registration = createTelegramBusFollowerRegistrationRuntime({
    ...ports.registration,
    ...sharedRuntimeDeps,
    startReceiving: receiver.start,
    stopReceiving: receiver.stop,
    onHeartbeatFailure: recovery,
  });
  return { receiver, registration };
}

export function createTelegramBusFollowerTargetReplacementHandler<TContext>(
  deps: TelegramBusFollowerTargetReplacementHandlerDeps<TContext>,
): NonNullable<
  TelegramBusForwardedUpdateReceiverRuntimeDeps<TContext>["handleReplaceTarget"]
> {
  const getNowMs = deps.getNowMs ?? Date.now;
  return async (input, ctx) => {
    await deps.topicTargetStore.load();
    const nowMs = getNowMs();
    const currentRecord = Threads.findCurrentTelegramInstanceThreadRecord({
      records: deps.topicTargetStore.list(),
      instanceId: deps.instanceId,
      preferredTarget: input.oldTarget ?? deps.registrationState.getTarget(),
    });
    if (input.oldTarget) {
      deps.topicTargetStore.markStaleByTarget(
        input.oldTarget,
        "deleted",
        "Follower thread was replaced by thread restore.",
      );
    } else if (currentRecord) {
      deps.topicTargetStore.markStaleByTarget(
        currentRecord.target,
        "deleted",
        "Follower thread was replaced by thread restore.",
      );
    }
    const profileKey =
      currentRecord?.profileKey ?? deps.getManualFollowerProfileKey();
    deps.topicTargetStore.upsert({
      profileKey,
      owner: {
        kind: "manual-follower",
        instanceId: deps.manualFollowerOwnerId,
      },
      target: input.target,
      status: "active",
      syncStatus: "open",
      createdAtMs: currentRecord?.createdAtMs ?? nowMs,
      updatedAtMs: nowMs,
      lastSyncObservedAtMs: nowMs,
      lastReconcileAction: "follower-thread-restore",
      instanceId: deps.instanceId,
      slot: currentRecord?.slot,
      threadName: currentRecord?.threadName,
      rerouteConfirmedAtMs: nowMs,
    });
    deps.registrationState.setRegistered(true, input.target, {
      slot: currentRecord?.slot,
      threadName: currentRecord?.threadName,
      generation: deps.registrationState.getGeneration?.(),
    });
    deps.setSyncState(
      Sync.markTelegramSyncSliceFresh(deps.getSyncState(), "target-bindings", {
        nowMs,
        action: "follower-thread-restore",
      }),
    );
    await deps.topicTargetStore.persist();
    deps.updateStatus(ctx);
    deps.recordRuntimeEvent?.(
      "bus",
      "Telegram follower thread target replaced",
      {
        phase: "follower-thread-restore",
        chatId: input.target.chatId,
        threadId: input.target.threadId,
        oldThreadId:
          input.oldTarget?.threadId ?? currentRecord?.target.threadId,
        slot: currentRecord?.slot,
      },
    );
  };
}

export function createTelegramBusFollowerClientRuntime<
  TContext,
  TReactionUpdate,
  TCallbackQuery,
  TMessage = unknown,
>(deps: TelegramBusFollowerClientRuntimeDeps<TMessage>) {
  const createRequestId = createTelegramBusRequestIdFactory(deps.instanceId);
  const timeoutMs =
    deps.timeoutMs ?? TELEGRAM_BUS_FOLLOWER_CLIENT_TIMEOUT_MS;
  const sharedClientDeps = {
    socketPath: deps.socketPath,
    createRequestId,
    timeoutMs,
    waitForRegistrationGeneration: deps.waitForRegistrationGeneration,
  };
  return {
    createRequestId,
    callApi: createTelegramBusFollowerApiCaller({
      ...sharedClientDeps,
      instanceId: deps.instanceId,
      getAuthSecret: deps.getApiAuthSecret,
      getRegistrationGeneration: deps.getRegistrationGeneration,
    }),
    agentMessages: createTelegramBusAgentMessageClient({
      ...sharedClientDeps,
      instanceId: deps.instanceId,
      getAuthSecret: deps.getApiAuthSecret,
      getRegistrationGeneration: deps.getRegistrationGeneration,
    }),
    foreignOwnedUpdateForwarder: createTelegramBusForeignOwnedUpdateForwarder<
      TContext,
      TReactionUpdate,
      TCallbackQuery,
      TMessage
    >({
      ...sharedClientDeps,
      getAuthSecret: deps.getForwardingAuthSecret,
      getForwardCommentBatchPosition:
        deps.getForwardCommentBatchPosition,
      recordRuntimeEvent: deps.recordRuntimeEvent,
    }),
    queueHandoff: createTelegramBusFollowerQueueHandoffClient({
      ...sharedClientDeps,
      instanceId: deps.instanceId,
      getAuthSecret: deps.getApiAuthSecret,
      getRegistrationGeneration: deps.getRegistrationGeneration,
    }),
    targetController: createTelegramBusFollowerTargetController({
      ...sharedClientDeps,
      getAuthSecret: deps.getForwardingAuthSecret,
    }),
  };
}

export function createTelegramBusFollowerQueueHandoffClient(
  deps: TelegramBusFollowerApiCallerDeps,
): (input: {
  recipientInstanceId: string;
  recipientRegistrationGeneration: string;
  donorProcessId: number;
  donorProcessBirthId: string;
  donorSessionGeneration: number;
  donorAcquisitionId: string;
  donorAcquiredAtMs: number;
  handoffToken: string;
  payload: TelegramQueueHandoffPayload;
}) => Promise<TelegramQueueHandoffStageResult> {
  const getNowMs = deps.getNowMs ?? Date.now;
  const timeoutMs =
    deps.timeoutMs ?? TELEGRAM_BUS_FOLLOWER_CLIENT_TIMEOUT_MS;
  return async (input) => {
    const registration = await resolveTelegramBusFollowerRegistration(
      deps,
      timeoutMs,
    );
    const socketPath = resolveTelegramBusSocketPath(deps.socketPath);
    const response = await sendTelegramBusLocalEnvelope({
      socketPath,
      timeoutMs: registration.remainingTimeoutMs,
      retry: getTelegramBusTransportRetryPolicy({
        endpoint: socketPath,
        operation: "operation",
      }),
      envelope: {
        kind: "follower.offerQueueHandoff",
        requestId: deps.createRequestId(),
        auth: deps.getAuthSecret?.(),
        instanceId: deps.instanceId,
        registrationGeneration: registration.generation,
        ...input,
        sentAtMs: getNowMs(),
      },
    });
    const queueOwner =
      response?.kind === "bus.ack" && isRecord(response.result)
        ? parseTelegramUpdateJournalQueueOwner(response.result.queueOwner)
        : undefined;
    if (
      response?.kind === "bus.ack" &&
      response.ok &&
      isRecord(response.result) &&
      response.result.status === "staged" &&
      typeof response.result.receiptId === "string" &&
      Array.isArray(response.result.sourceUpdateIds) &&
      response.result.sourceUpdateIds.every(Number.isSafeInteger) &&
      input.payload.admissionReceipts.length === 1 &&
      response.result.receiptId ===
        input.payload.admissionReceipts[0]?.receiptId &&
      response.result.sourceUpdateIds.length ===
        input.payload.admissionReceipts[0].sourceUpdateIds.length &&
      response.result.sourceUpdateIds.every(
        (updateId, index) =>
          updateId === input.payload.admissionReceipts[0]!.sourceUpdateIds[index],
      ) &&
      queueOwner
    ) {
      return {
        status: "staged",
        receiptId: response.result.receiptId,
        sourceUpdateIds: response.result.sourceUpdateIds as number[],
        queueOwner,
      };
    }
    throw new Error(
      response?.kind === "bus.ack"
        ? response.message ?? "Telegram queue handoff was rejected."
        : "Telegram queue handoff did not return an acknowledgement.",
    );
  };
}

export function createTelegramBusAgentMessageClient(
  deps: TelegramBusFollowerApiCallerDeps,
): {
  resolveTarget: (
    selector: TelegramBusAgentTargetSelector,
  ) => Promise<TelegramTarget & { threadId: number }>;
  routeMessage: (message: TelegramBusAgentMessage) => Promise<void>;
} {
  const getNowMs = deps.getNowMs ?? Date.now;
  const timeoutMs =
    deps.timeoutMs ?? TELEGRAM_BUS_FOLLOWER_CLIENT_TIMEOUT_MS;
  const request = async (
    envelope:
      | Extract<TelegramBusEnvelope, { kind: "follower.resolveAgentTarget" }>
      | Extract<TelegramBusEnvelope, { kind: "follower.routeAgentMessage" }>,
    requestTimeoutMs = timeoutMs,
  ): Promise<unknown> => {
    const socketPath = resolveTelegramBusSocketPath(deps.socketPath);
    const response = await sendTelegramBusLocalEnvelope({
      socketPath,
      timeoutMs: requestTimeoutMs,
      retry: getTelegramBusTransportRetryPolicy({
        endpoint: socketPath,
        operation: "operation",
      }),
      envelope,
    });
    if (response?.kind === "bus.ack" && response.ok) return response.result;
    throw new Error(
      response?.kind === "bus.ack"
        ? response.message ?? "Telegram bus agent message failed."
        : "Telegram bus agent message did not return an acknowledgement.",
    );
  };
  const registrationFields = async () => {
    const registration = await resolveTelegramBusFollowerRegistration(
      deps,
      timeoutMs,
    );
    return {
      fields: {
        auth: deps.getAuthSecret?.(),
        instanceId: deps.instanceId,
        registrationGeneration: registration.generation,
      },
      remainingTimeoutMs: registration.remainingTimeoutMs,
    };
  };
  return {
    async resolveTarget(selector) {
      const registration = await registrationFields();
      const result = await request({
        kind: "follower.resolveAgentTarget",
        requestId: deps.createRequestId(),
        ...registration.fields,
        selector,
        sentAtMs: getNowMs(),
      }, registration.remainingTimeoutMs);
      if (!result || typeof result !== "object" || Array.isArray(result)) {
        throw new Error("Telegram bus returned an invalid agent target.");
      }
      const target = result as Record<string, unknown>;
      if (
        typeof target.chatId !== "number" ||
        typeof target.threadId !== "number"
      ) {
        throw new Error("Telegram bus returned an invalid agent target.");
      }
      return { chatId: target.chatId, threadId: target.threadId };
    },
    async routeMessage(message) {
      const registration = await registrationFields();
      await request({
        kind: "follower.routeAgentMessage",
        requestId: deps.createRequestId(),
        ...registration.fields,
        message,
        sentAtMs: getNowMs(),
      }, registration.remainingTimeoutMs);
    },
  };
}

export function createTelegramBusFollowerApiCaller(
  deps: TelegramBusFollowerApiCallerDeps,
): (method: string, args: unknown[]) => Promise<unknown> {
  const getNowMs = deps.getNowMs ?? Date.now;
  const timeoutMs =
    deps.timeoutMs ?? TELEGRAM_BUS_FOLLOWER_CLIENT_TIMEOUT_MS;
  return async (method, args) => {
    const registration = await resolveTelegramBusFollowerRegistration(
      deps,
      timeoutMs,
    );
    const socketPath = resolveTelegramBusSocketPath(deps.socketPath);
    let response: TelegramBusEnvelope | undefined;
    try {
      response = await sendTelegramBusLocalEnvelope({
        socketPath,
        timeoutMs: registration.remainingTimeoutMs,
        retry: getTelegramBusTransportRetryPolicy({
          endpoint: socketPath,
          operation: "operation",
        }),
        envelope: {
          kind: "follower.callApi",
          requestId: deps.createRequestId(),
          auth: deps.getAuthSecret?.(),
          instanceId: deps.instanceId,
          registrationGeneration: registration.generation,
          method,
          args,
          sentAtMs: getNowMs(),
        },
      });
    } catch (error) {
      const apiMethod =
        (method === "call" || method === "callMultipart") &&
        typeof args[0] === "string"
          ? args[0]
          : method;
      if (!isTelegramApiMethodRetrySafe(apiMethod)) {
        throw new TelegramApiCommitUnknownError(apiMethod, error);
      }
      throw error;
    }
    if (response?.kind === "bus.ack" && response.ok) return response.result;
    const message =
      response?.kind === "bus.ack"
        ? response.message
        : "Telegram bus API call did not return an acknowledgement.";
    if (
      response?.kind === "bus.ack" &&
      response.error?.code === "stale-target" &&
      response.error.chatId !== undefined &&
      response.error.threadId !== undefined
    ) {
      throw new TelegramApiStaleTargetError(
        message ?? "Telegram thread target is stale.",
        {
          chatId: response.error.chatId,
          threadId: response.error.threadId,
        },
      );
    }
    if (
      response?.kind === "bus.ack" &&
      response.error?.code === "commit-unknown"
    ) {
      throw new TelegramApiCommitUnknownError(
        response.error.method ?? method,
        new Error(message ?? "Telegram bus API call result is ambiguous."),
      );
    }
    throw new Error(message ?? "Telegram bus API call failed.");
  };
}

async function resolveTelegramBusFollowerRegistration(
  deps: Pick<
    TelegramBusFollowerApiCallerDeps,
    | "getRegistrationGeneration"
    | "waitForRegistrationGeneration"
    | "getNowMs"
  >,
  timeoutMs: number,
): Promise<{ generation: string; remainingTimeoutMs: number }> {
  const current = deps.getRegistrationGeneration();
  if (current) return { generation: current, remainingTimeoutMs: timeoutMs };
  const getNowMs = deps.getNowMs ?? Date.now;
  const startedAtMs = getNowMs();
  const restored = await deps.waitForRegistrationGeneration?.(
    timeoutMs,
  );
  const remainingTimeoutMs = Math.max(0, timeoutMs - (getNowMs() - startedAtMs));
  if (restored && remainingTimeoutMs > 0) {
    return { generation: restored, remainingTimeoutMs };
  }
  throw new Error("Telegram bus follower is not registered.");
}

function isTelegramStaleContextError(error: unknown): boolean {
  return (
    error instanceof Error &&
    (error.message.includes("stale after session") ||
      error.message.includes("stale ctx"))
  );
}

export function createTelegramBusFollowerSessionReplacementSuspender(
  deps: TelegramBusFollowerSessionReplacementSuspenderDeps,
): () => Promise<void> {
  const getNowMs = deps.getNowMs ?? Date.now;
  const getPid = deps.getPid ?? (() => process.pid);
  return async () => {
    const target = deps.registrationState.getTarget();
    if (deps.registrationState.isRegistered() && target) {
      setTelegramFollowerSessionHandoff({
        pid: getPid(),
        instanceId: deps.instanceId,
        createdAtMs: getNowMs(),
        target,
        slot: deps.registrationState.getSlot(),
        threadName: deps.registrationState.getThreadName(),
      });
      deps.recordRuntimeEvent(
        "bus",
        "Telegram follower registration suspended for session replacement",
        {
          phase: "follower-session-handoff",
          instanceId: deps.instanceId,
          chatId: target.chatId,
          threadId: target.threadId,
        },
      );
    } else if (deps.isLeader?.()) {
      const leaderBinding = deps.getLeaderBinding?.();
      if (typeof leaderBinding?.target?.threadId === "number") {
        const activeContext = deps.getActiveContext?.();
        const profileKey = Threads.getTelegramThreadOwnerKey({
          kind: "leader",
          cwd: activeContext?.cwd,
          instanceId: deps.instanceId,
          telegramProfile: deps.getActiveProfileName?.(),
        });
        Threads.setTelegramLeaderSessionHandoff({
          pid: getPid(),
          instanceId: deps.instanceId,
          createdAtMs: getNowMs(),
          profileKey,
          target: {
            chatId: leaderBinding.target.chatId,
            threadId: leaderBinding.target.threadId,
          },
          slot: leaderBinding.slot,
          threadName: leaderBinding.threadName,
        });
        deps.recordRuntimeEvent(
          "bus",
          "Telegram leader binding suspended for session replacement",
          {
            phase: "leader-session-handoff",
            instanceId: deps.instanceId,
            chatId: leaderBinding.target.chatId,
            threadId: leaderBinding.target.threadId,
            slot: leaderBinding.slot,
            threadName: leaderBinding.threadName,
          },
        );
      }
    }
    await deps.suspendPolling();
  };
}

export function createTelegramBusFollowerSessionRefreshHook<TContext>(
  deps: TelegramBusFollowerSessionRefreshHookDeps<TContext>,
): (_event: unknown, ctx: TContext) => Promise<void> {
  return async (_event, ctx) => {
    if (deps.isSessionActive && !deps.isSessionActive(ctx)) return;
    if (!deps.registrationState.isRegistered()) {
      const handoff = getTelegramFollowerSessionHandoff();
      const lockState = deps.getLeaderState();
      const handoffIsFresh = isTelegramFollowerSessionHandoffFresh(handoff);
      if (handoffIsFresh && lockState.kind === "active-elsewhere") {
        try {
          const restored = await deps.registrationRuntime.registerWithLeader(
            ctx,
            lockState.lock,
            {
              target: handoff.target,
              previousInstanceId: handoff.instanceId,
            },
          );
          if (deps.isSessionActive && !deps.isSessionActive(ctx)) return;
          if (restored) {
            setTelegramFollowerSessionHandoff(undefined);
            deps.updateStatus(ctx);
            deps.recordRuntimeEvent(
              "bus",
              "Telegram follower registration restored after session replacement",
              {
                phase: "follower-session-restore",
                previousInstanceId: handoff.instanceId,
              },
            );
          }
        } catch (error) {
          deps.recordRuntimeEvent("bus", error, {
            phase: "follower-session-restore",
            previousInstanceId: handoff?.instanceId,
          });
        }
      } else if (handoff) {
        setTelegramFollowerSessionHandoff(undefined);
      }
    }
    if (!deps.registrationState.isRegistered()) return;
    if (deps.isSessionActive && !deps.isSessionActive(ctx)) return;
    deps.registrationRuntime.setContext(ctx);
    deps.updateStatus(ctx);
    deps.recordRuntimeEvent(
      "bus",
      "Telegram follower session context refreshed",
      { phase: "follower-session-refresh" },
    );
  };
}

export function createTelegramBusFollowerControlState(): TelegramBusFollowerControlState {
  let activeAuthSecret: string | undefined;
  let lifecyclePhase: TelegramBusFollowerControlLifecyclePhase | undefined;
  return {
    getActiveAuthSecret: () => activeAuthSecret,
    setActiveAuthSecret(secret) {
      activeAuthSecret = secret;
    },
    getLifecyclePhase: () => lifecyclePhase,
    setLifecyclePhase(phase) {
      lifecyclePhase = phase;
    },
  };
}

export function createTelegramBusFollowerRegistrationState(
  options: { onAvailabilityChanged?: () => void } = {},
): TelegramBusFollowerRegistrationState {
  let registered = false;
  let target: TelegramTarget | undefined;
  let slot: string | undefined;
  let threadName: string | undefined;
  let generation: string | undefined;
  let leaderProtocol: TelegramBusProtocolIdentity | undefined;
  let eligibleElectionSlots: string[] = [];
  let recoveryEpoch = 0;
  let activeRecoveryEpoch: number | undefined;
  const generationWaiters = new Set<{
    epoch: number;
    settle: (value: string | undefined) => void;
  }>();
  const settleGenerationWaiters = (
    value: string | undefined,
    epoch?: number,
  ) => {
    for (const waiter of [...generationWaiters]) {
      if (epoch === undefined || waiter.epoch === epoch) waiter.settle(value);
    }
  };
  return {
    isRegistered: () => registered,
    getTarget: () => (target ? { ...target } : undefined),
    getSlot: () => slot,
    getThreadName: () => threadName,
    getGeneration: () => generation,
    beginRecovery: () => {
      if (activeRecoveryEpoch !== undefined) return activeRecoveryEpoch;
      activeRecoveryEpoch = ++recoveryEpoch;
      return activeRecoveryEpoch;
    },
    cancelRecovery: () => {
      const epoch = activeRecoveryEpoch;
      activeRecoveryEpoch = undefined;
      if (epoch !== undefined) settleGenerationWaiters(undefined, epoch);
    },
    waitForGeneration: (timeoutMs = TELEGRAM_BUS_FOLLOWER_REGISTRATION_WAIT_MS) => {
      if (generation) return Promise.resolve(generation);
      const epoch = activeRecoveryEpoch;
      if (epoch === undefined) return Promise.resolve(undefined);
      return new Promise((resolve) => {
        let timer: NodeJS.Timeout | undefined;
        const settle = (value: string | undefined) => {
          generationWaiters.delete(waiter);
          if (timer) clearTimeout(timer);
          resolve(value);
        };
        const waiter = { epoch, settle };
        generationWaiters.add(waiter);
        timer = setTimeout(() => settle(undefined), Math.max(0, timeoutMs));
      });
    },
    getLeaderProtocol: () =>
      leaderProtocol
        ? { ...leaderProtocol, capabilities: [...leaderProtocol.capabilities] }
        : undefined,
    getEligibleElectionSlots: () => [...eligibleElectionSlots],
    setEligibleElectionSlots: (slots) => {
      eligibleElectionSlots = Array.from(
        new Set(slots.filter((slot) => /^[A-Z]$/.test(slot))),
      ).sort();
    },
    setRegistered: (next, nextTarget, metadata) => {
      const availabilityChanged = registered !== next;
      registered = next;
      target = next ? (nextTarget ? { ...nextTarget } : undefined) : undefined;
      slot = next ? metadata?.slot : undefined;
      threadName = next ? metadata?.threadName : undefined;
      generation = next ? metadata?.generation : undefined;
      leaderProtocol =
        next && metadata?.leaderProtocol
          ? {
              ...metadata.leaderProtocol,
              capabilities: [...metadata.leaderProtocol.capabilities],
            }
          : undefined;
      if (availabilityChanged) options.onAvailabilityChanged?.();
      if (generation) {
        activeRecoveryEpoch = undefined;
        settleGenerationWaiters(generation);
      }
    },
  };
}

export function createTelegramBusFollowerHeartbeatRecoveryHandler<TContext>(
  deps: TelegramBusFollowerHeartbeatRecoveryHandlerDeps<TContext>,
): (error: unknown, ctx: TContext) => Promise<void> {
  const promotionGraceMs =
    deps.promotionGraceMs ?? TELEGRAM_BUS_FOLLOWER_PROMOTION_GRACE_MS;
  const sleep =
    deps.sleep ??
    ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
  const scheduleRetry =
    deps.scheduleRetry ??
    ((retry: () => void, delayMs: number) => {
      const timer = setTimeout(retry, delayMs);
      timer.unref?.();
    });
  let promotionPending = false;
  const safeUpdateStatus = (ctx: TContext) => {
    try {
      deps.updateStatus(ctx);
    } catch (error) {
      if (!isTelegramStaleContextError(error)) throw error;
      deps.recordRuntimeEvent("bus", error, {
        phase: "follower-stale-context-status",
      });
    }
  };
  const clearRegisteredState = (ctx: TContext) => {
    deps.registrationState.setRegistered(false);
    safeUpdateStatus(ctx);
  };
  const tryRegisterWithLeader = async (
    ctx: TContext,
    leader: TelegramBusFollowerLeaderLock,
    phase: string,
    binding?: TelegramBusFollowerPromotedBinding,
  ) => {
    try {
      const restored = await deps
        .getRegistrationRuntime()
        .registerWithLeader(
          ctx,
          leader,
          binding?.target ? { target: binding.target } : undefined,
        );
      if (!restored) return false;
      deps.setLifecyclePhase(undefined);
      safeUpdateStatus(ctx);
      deps.recordRuntimeEvent(
        "bus",
        "Telegram follower registration restored",
        {
          phase,
        },
      );
      return true;
    } catch (error) {
      clearRegisteredState(ctx);
      deps.recordRuntimeEvent("bus", error, { phase });
      return false;
    }
  };
  const snapshotBinding = (): TelegramBusFollowerPromotedBinding => ({
    target: deps.registrationState.getTarget(),
    slot: deps.registrationState.getSlot(),
    threadName: deps.registrationState.getThreadName(),
  });
  const scheduleRecovery = (
    reason: unknown,
    fallbackCtx: TContext,
    binding: TelegramBusFollowerPromotedBinding,
  ) => {
    const retry = () => {
      const activeCtx = deps.getActiveContext
        ? deps.getActiveContext()
        : fallbackCtx;
      if (!activeCtx) {
        scheduleRetry(retry, promotionGraceMs);
        return;
      }
      void recover(reason, activeCtx, binding);
    };
    scheduleRetry(retry, promotionGraceMs);
  };
  const promoteToLeader = async (
    reason: unknown,
    ctx: TContext,
    binding: TelegramBusFollowerPromotedBinding,
    election: TelegramBusFollowerElection,
  ) => {
    const activeCtx = deps.getActiveContext?.();
    if (deps.getActiveContext && activeCtx !== ctx) {
      scheduleRecovery(reason, ctx, binding);
      return;
    }
    deps.setLifecyclePhase("electing");
    safeUpdateStatus(ctx);
    deps.recordRuntimeEvent("bus", reason, {
      phase: "follower-promotion-electing",
    });
    deps.getRegistrationRuntime().stop();
    deps.setLifecyclePhase("electing");
    safeUpdateStatus(ctx);
    deps.recordRuntimeEvent("bus", "Telegram follower attempting promotion", {
      phase: "follower-promotion-electing",
    });
    const promoted = await deps.promoteToLeader(ctx, binding, election);
    deps.setLifecyclePhase(undefined);
    safeUpdateStatus(ctx);
    deps.recordRuntimeEvent(
      "bus",
      promoted
        ? "Telegram follower promotion completed"
        : "Telegram follower promotion lost election",
      {
        phase: promoted
          ? "follower-promotion-complete"
          : "follower-promotion-lost",
      },
    );
    if (!promoted) scheduleRecovery(reason, ctx, binding);
  };
  const attemptPreferredPromotion = async (
    reason: unknown,
    ctx: TContext,
    binding: TelegramBusFollowerPromotedBinding,
    candidateState: TelegramBusFollowerLeaderState,
  ): Promise<void> => {
    const slot = binding.slot;
    const lowerEligibleSlot = slot
      ? deps.registrationState
          .getEligibleElectionSlots()
          .find((candidate) => candidate < slot)
      : undefined;
    if (lowerEligibleSlot) {
      deps.recordRuntimeEvent(
        "bus",
        "Telegram follower deferring to a lower-slot election candidate",
        {
          phase: "follower-promotion-slot-priority",
          slot,
          lowerEligibleSlot,
        },
      );
      await sleep(promotionGraceMs);
      candidateState = deps.getLeaderState();
      if (candidateState.kind === "active-elsewhere") {
        if (
          !(await tryRegisterWithLeader(
            ctx,
            candidateState.lock,
            "follower-register-preferred-successor",
            binding,
          ))
        ) {
          scheduleRecovery(reason, ctx, binding);
        }
        return;
      }
    }
    if (candidateState.kind !== "stale" && candidateState.kind !== "inactive")
      return;
    await promoteToLeader(reason, ctx, binding, {
      expectedOwner:
        candidateState.kind === "stale" ? candidateState.lock : undefined,
    });
  };
  const recover = async (
    error: unknown,
    ctx: TContext,
    carriedBinding?: TelegramBusFollowerPromotedBinding,
  ): Promise<void> => {
    if (promotionPending) return;
    promotionPending = true;
    deps.registrationState.beginRecovery();
    try {
      const initialBinding = carriedBinding ?? snapshotBinding();
      const state = deps.getLeaderState();
      if (state.kind === "active-elsewhere") {
        clearRegisteredState(ctx);
        if (
          await tryRegisterWithLeader(
            ctx,
            state.lock,
            "follower-register-restore",
            initialBinding,
          )
        ) {
          return;
        }
        deps.setLifecyclePhase("electing");
        safeUpdateStatus(ctx);
        deps.recordRuntimeEvent(
          "bus",
          "Telegram follower waiting for leader reload recovery",
          { phase: "follower-promotion-grace" },
        );
        await sleep(promotionGraceMs);
        const graceState = deps.getLeaderState();
        if (graceState.kind === "active-elsewhere") {
          if (
            await tryRegisterWithLeader(
              ctx,
              graceState.lock,
              "follower-register-restore-grace",
              initialBinding,
            )
          ) {
            return;
          }
          deps.setLifecyclePhase(undefined);
          safeUpdateStatus(ctx);
          deps.recordRuntimeEvent(
            "bus",
            "Telegram follower promotion blocked by live leader lease",
            {
              phase: "follower-promotion-live-owner",
              leaderInstanceId: graceState.lock.instanceId,
              leaderEpoch: graceState.lock.leaderEpoch,
            },
          );
          scheduleRecovery(error, ctx, initialBinding);
          return;
        }
        await attemptPreferredPromotion(
          error,
          ctx,
          initialBinding,
          graceState,
        );
        return;
      }
      await attemptPreferredPromotion(error, ctx, initialBinding, state);
    } catch (promotionError) {
      deps.setLifecyclePhase(undefined);
      safeUpdateStatus(ctx);
      if (isTelegramStaleContextError(promotionError)) {
        deps.recordRuntimeEvent("bus", promotionError, {
          phase: "follower-heartbeat-stale-context",
        });
        return;
      }
      throw promotionError;
    } finally {
      promotionPending = false;
    }
  };
  return recover;
}

export function createTelegramBusFollowerRegistrationRuntime<
  TContext extends { cwd?: string },
>(
  deps: TelegramBusFollowerRegistrationRuntimeDeps<TContext>,
): TelegramBusFollowerRegistrationRuntime<TContext> {
  const getNowMs = deps.getNowMs ?? Date.now;
  const getPid = deps.getPid ?? (() => process.pid);
  const heartbeatMs = deps.heartbeatMs ?? 1000;
  const registrationTimeoutMs =
    deps.registrationTimeoutMs ?? deps.timeoutMs ?? 30000;
  const registrationRetryAttempts =
    deps.registrationRetryAttempts ??
    TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_ATTEMPTS;
  const registrationRetryDelayMs =
    deps.registrationRetryDelayMs ??
    TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_DELAY_MS;
  let heartbeatInterval: ReturnType<typeof setInterval> | undefined;
  let heartbeatPromise: Promise<void> | undefined;
  let heartbeatPromiseGeneration: string | undefined;
  let activeLeaderSocketPath: string | undefined;
  let activeAuthSecret: string | undefined;
  let activeRegistrationGeneration: string | undefined;
  let activeContext: TContext | undefined;
  let lastKnownTarget: TelegramTarget | undefined;
  let lastKnownSlot: string | undefined;
  let lastKnownThreadName: string | undefined;
  const stopHeartbeat = () => {
    if (!heartbeatInterval) return;
    clearInterval(heartbeatInterval);
    heartbeatInterval = undefined;
  };
  const stop = () => {
    stopHeartbeat();
    activeAuthSecret = undefined;
    activeRegistrationGeneration = undefined;
    heartbeatPromise = undefined;
    heartbeatPromiseGeneration = undefined;
    deps.setActiveAuthSecret?.(undefined);
    deps.registrationState?.cancelRecovery();
    deps.registrationState?.setRegistered(false);
    lastKnownTarget = undefined;
    lastKnownSlot = undefined;
    lastKnownThreadName = undefined;
    activeContext = undefined;
    void Promise.resolve(deps.stopReceiving?.()).catch((error) => {
      try {
        deps.recordRuntimeEvent?.("bus", error, {
          phase: "follower-receiver-stop",
        });
      } catch {
        // Stop diagnostics cannot create an unhandled Promise.
      }
    });
  };
  const sendHeartbeat = async () => {
    const leaderSocketPath = activeLeaderSocketPath;
    const registrationGeneration = activeRegistrationGeneration;
    const heartbeatContext = activeContext;
    if (!leaderSocketPath || !registrationGeneration) return;
    const isCurrentHeartbeat = (): boolean =>
      activeLeaderSocketPath === leaderSocketPath &&
      activeRegistrationGeneration === registrationGeneration &&
      activeContext === heartbeatContext;
    try {
      const response = await sendTelegramBusLocalEnvelope({
        socketPath: leaderSocketPath,
        timeoutMs: deps.timeoutMs,
        retry: getTelegramBusTransportRetryPolicy({
          endpoint: leaderSocketPath,
          operation: "operation",
        }),
        envelope: {
          kind: "follower.heartbeat",
          requestId: deps.createRequestId(),
          auth: activeAuthSecret,
          instanceId: deps.instanceId,
          registrationGeneration,
          sentAtMs: getNowMs(),
        },
      });
      if (!isCurrentHeartbeat()) return;
      if (response?.kind === "bus.ack" && response.ok) {
        const heartbeatResult = isRecord(response.result)
          ? response.result
          : undefined;
        const slots = Array.isArray(heartbeatResult?.eligibleElectionSlots)
          ? heartbeatResult.eligibleElectionSlots.filter(
              (slot): slot is string => typeof slot === "string",
            )
          : [];
        deps.registrationState?.setEligibleElectionSlots(slots);
      }
      if (response?.kind === "bus.ack" && !response.ok) {
        throw new Error(
          response.message ?? "Telegram bus follower heartbeat was rejected.",
        );
      }
    } catch (error) {
      if (!isCurrentHeartbeat()) return;
      try {
        deps.recordRuntimeEvent?.("bus", error, {
          phase: "follower-heartbeat",
        });
      } catch {
        // Diagnostics cannot replace heartbeat recovery.
      }
      if (!heartbeatContext) return;
      try {
        await deps.onHeartbeatFailure?.(error, heartbeatContext);
      } catch (recoveryError) {
        try {
          deps.recordRuntimeEvent?.("bus", recoveryError, {
            phase: "follower-heartbeat-recovery",
          });
        } catch {
          // A diagnostic sink cannot create an unhandled interval rejection.
        }
      }
    }
  };
  const requestHeartbeat = (): Promise<void> => {
    const generation = activeRegistrationGeneration;
    if (heartbeatPromise && heartbeatPromiseGeneration === generation) {
      return heartbeatPromise;
    }
    let tracked: Promise<void>;
    tracked = sendHeartbeat().finally(() => {
      if (heartbeatPromise === tracked) {
        heartbeatPromise = undefined;
        heartbeatPromiseGeneration = undefined;
      }
    });
    heartbeatPromise = tracked;
    heartbeatPromiseGeneration = generation;
    return tracked;
  };
  const startHeartbeat = (socketPath: string) => {
    stopHeartbeat();
    activeLeaderSocketPath = socketPath;
    heartbeatInterval = setInterval(() => {
      void requestHeartbeat();
    }, heartbeatMs);
    heartbeatInterval.unref?.();
  };
  return {
    registerWithLeader: async (ctx, leader, options) => {
      const pendingHandoff = options
        ? undefined
        : getTelegramFollowerSessionHandoff();
      const pendingHandoffOptions = isTelegramFollowerSessionHandoffFresh(
        pendingHandoff,
      )
        ? {
            target: pendingHandoff.target,
            previousInstanceId: pendingHandoff.instanceId,
          }
        : undefined;
      const registrationOptions = options ?? pendingHandoffOptions;
      const leaderSocketPath =
        leader.busSocketPath ??
        deps.getLeaderSocketPath?.() ??
        getTelegramBusSocketPath();
      await deps.startReceiving?.();
      activeAuthSecret = deps.getLeaderAuthSecret
        ? deps.getLeaderAuthSecret(leader)
        : leader?.busSecret;
      deps.setActiveAuthSecret?.(activeAuthSecret);
      const registrationGeneration = deps.createRequestId();
      const registrationEnvelope: Extract<
        TelegramBusEnvelope,
        { kind: "follower.register" }
      > = {
        kind: "follower.register",
        requestId: registrationGeneration,
        auth: activeAuthSecret,
        registration: {
          instanceId: deps.instanceId,
          ...(registrationOptions?.previousInstanceId
            ? { previousInstanceId: registrationOptions.previousInstanceId }
            : {}),
          profileKey:
            deps.getProfileKey?.(ctx) ??
            (ctx.cwd ? `cwd:${ctx.cwd}` : undefined),
          threadName:
            deps.registrationState?.getThreadName() ??
            lastKnownThreadName ??
            deps.getThreadName?.(ctx) ??
            (ctx.cwd ? basename(ctx.cwd) : undefined),
          ...((deps.registrationState?.getSlot() ?? lastKnownSlot)
            ? { slot: deps.registrationState?.getSlot() ?? lastKnownSlot }
            : {}),
          cwd: ctx.cwd,
          pid: getPid(),
          processBirthId: deps.getProcessBirthId?.(),
          sessionGeneration: deps.getSessionGeneration?.(),
          target:
            registrationOptions?.target ??
            deps.registrationState?.getTarget() ??
            lastKnownTarget,
          busSocketPath:
            deps.getFollowerBusSocketPath?.() ?? deps.followerBusSocketPath,
          registrationGeneration,
          protocol: deps.protocolIdentity,
          connectedAtMs: getNowMs(),
        },
      };
      let response: TelegramBusEnvelope | undefined;
      try {
        response = await sendTelegramBusLocalEnvelope({
          socketPath: leaderSocketPath,
          timeoutMs: registrationTimeoutMs,
          envelope: registrationEnvelope,
          retry: getTelegramBusTransportRetryPolicy({
            endpoint: leaderSocketPath,
            operation: "registration",
            overrides: {
              attempts: registrationRetryAttempts,
              delayMs: registrationRetryDelayMs,
            },
          }),
          recordTransportEvent(phase, details) {
            deps.recordRuntimeEvent?.("bus", `Telegram bus ${phase}`, {
              phase: `follower-register-${phase}`,
              ...details,
            });
          },
        });
      } catch (error) {
        stopHeartbeat();
        activeLeaderSocketPath = undefined;
        activeAuthSecret = undefined;
        deps.registrationState?.setRegistered(false);
        deps.setActiveAuthSecret?.(undefined);
        await deps.stopReceiving?.();
        throw error;
      }
      if (deps.isContextActive && !deps.isContextActive(ctx)) {
        stopHeartbeat();
        activeLeaderSocketPath = undefined;
        activeAuthSecret = undefined;
        deps.setActiveAuthSecret?.(undefined);
        await deps.stopReceiving?.();
        return false;
      }
      const compatibility = getTelegramBusProtocolCompatibility({
        local: deps.protocolIdentity,
        remote: response?.kind === "bus.ack" ? response.protocol : undefined,
      });
      if (!compatibility.compatible) {
        stopHeartbeat();
        activeLeaderSocketPath = undefined;
        activeAuthSecret = undefined;
        deps.registrationState?.setRegistered(false);
        deps.setActiveAuthSecret?.(undefined);
        await deps.stopReceiving?.();
        throw new Error(
          `Incompatible Telegram bus leader protocol: ${compatibility.reason}.`,
        );
      }
      if (response?.kind === "bus.ack" && !response.ok) {
        stopHeartbeat();
        activeLeaderSocketPath = undefined;
        activeAuthSecret = undefined;
        deps.registrationState?.setRegistered(false);
        deps.setActiveAuthSecret?.(undefined);
        await deps.stopReceiving?.();
        throw new Error(
          response.message ??
            "Telegram bus follower registration was rejected.",
        );
      }
      if (response?.kind === "bus.ack" && response.ok) {
        const registrationResult = parseRegistrationResult(response.result);
        deps.registrationState?.setRegistered(true, registrationResult.target, {
          ...registrationResult,
          generation: registrationGeneration,
          ...(response.protocol
            ? { leaderProtocol: response.protocol }
            : {}),
        });
        lastKnownTarget = registrationResult.target;
        lastKnownSlot = registrationResult.slot;
        lastKnownThreadName = registrationResult.threadName;
        activeLeaderSocketPath = leaderSocketPath;
        activeRegistrationGeneration = registrationGeneration;
        activeContext = ctx;
        try {
          await deps.onRegistered?.(ctx);
        } catch (error) {
          stopHeartbeat();
          activeLeaderSocketPath = undefined;
          activeAuthSecret = undefined;
          activeRegistrationGeneration = undefined;
          deps.registrationState?.setRegistered(false);
          deps.setActiveAuthSecret?.(undefined);
          await deps.stopReceiving?.();
          throw error;
        }
        await requestHeartbeat();
        startHeartbeat(leaderSocketPath);
        if (
          pendingHandoffOptions &&
          getTelegramFollowerSessionHandoff()?.instanceId ===
            pendingHandoffOptions.previousInstanceId
        ) {
          setTelegramFollowerSessionHandoff(undefined);
        }
        return true;
      }
      stopHeartbeat();
      activeLeaderSocketPath = undefined;
      activeAuthSecret = undefined;
      deps.registrationState?.setRegistered(false);
      deps.setActiveAuthSecret?.(undefined);
      await deps.stopReceiving?.();
      return false;
    },
    setContext(ctx) {
      activeContext = ctx;
    },
    async disconnectFromLeader() {
      if (!activeLeaderSocketPath || !activeRegistrationGeneration) {
        return false;
      }
      const response = await sendTelegramBusLocalEnvelope({
        socketPath: activeLeaderSocketPath,
        timeoutMs: registrationTimeoutMs,
        retry: getTelegramBusTransportRetryPolicy({
          endpoint: activeLeaderSocketPath,
          operation: "operation",
        }),
        envelope: {
          kind: "follower.disconnect",
          requestId: deps.createRequestId(),
          auth: activeAuthSecret,
          instanceId: deps.instanceId,
          registrationGeneration: activeRegistrationGeneration,
          sentAtMs: getNowMs(),
        },
      });
      if (response?.kind === "bus.ack" && response.ok) return true;
      throw new Error(
        response?.kind === "bus.ack"
          ? (response.message ?? "Telegram follower disconnect was rejected.")
          : "Telegram follower disconnect was not acknowledged.",
      );
    },
    stop,
  };
}

const TELEGRAM_FOLLOWER_FORWARD_BATCH_POSITION_FIELD =
  "pi_telegram_forward_comment_batch_position";

export function prepareTelegramBusFollowerJournaledUpdateForExecution<
  TUpdate extends { message?: unknown } & Record<string, unknown>,
>(
  update: TUpdate,
  prepareForwardedMessage: (
    message: NonNullable<TUpdate["message"]>,
    position: "comment" | "forward",
  ) => void,
): TUpdate {
  const position = update[TELEGRAM_FOLLOWER_FORWARD_BATCH_POSITION_FIELD];
  if (
    update.message !== undefined &&
    (position === "comment" || position === "forward")
  ) {
    prepareForwardedMessage(
      update.message as NonNullable<TUpdate["message"]>,
      position,
    );
  }
  if (position === undefined) return update;
  const prepared = { ...update };
  delete prepared[TELEGRAM_FOLLOWER_FORWARD_BATCH_POSITION_FIELD];
  return prepared;
}

export function createTelegramBusFollowerDurableAdmissionRuntime<TContext>(deps: {
  journal: {
    appendBatch(
      updates: readonly ({ update_id: number } & Record<string, unknown>)[],
    ): unknown;
  };
  signalWorker: (ctx: TContext) => void;
}): TelegramBusFollowerDurableAdmissionPort<TContext> {
  return {
    async admit(envelope, ctx) {
      const delivery = envelope.delivery;
      if (!delivery) {
        throw new Error("Telegram follower durable admission requires delivery identity.");
      }
      const expected = createTelegramBusFollowerDeliveryIdentity({
        kind: envelope.kind,
        recipientBindingKey: delivery.recipientBindingKey,
        sourceUpdateId: delivery.sourceUpdateId,
      });
      if (expected.deliveryId !== delivery.deliveryId) {
        throw new Error("Invalid Telegram follower delivery id.");
      }
      const carrier =
        envelope.kind === "leader.forwardCallback"
          ? envelope.query
          : envelope.kind === "leader.forwardReaction"
            ? envelope.reactionUpdate
            : envelope.message;
      if (
        !isRecord(carrier) ||
        carrier.pi_telegram_source_update_id !== delivery.sourceUpdateId
      ) {
        throw new Error("Telegram follower delivery source update id mismatch.");
      }
      const update = {
        update_id: delivery.sourceUpdateId,
        ...(envelope.kind === "leader.forwardMessage" &&
        envelope.forwardCommentBatchPosition
          ? {
              [TELEGRAM_FOLLOWER_FORWARD_BATCH_POSITION_FIELD]:
                envelope.forwardCommentBatchPosition,
            }
          : {}),
        ...(envelope.kind === "leader.forwardCallback"
          ? { callback_query: carrier }
          : envelope.kind === "leader.forwardReaction"
            ? { message_reaction: carrier }
            : envelope.kind === "leader.forwardMessage"
              ? { message: carrier }
              : { edited_message: carrier }),
      };
      deps.journal.appendBatch([update]);
      deps.signalWorker(ctx);
      return {
        deliveryId: delivery.deliveryId,
        sourceUpdateId: delivery.sourceUpdateId,
      };
    },
  };
}

export function createTelegramBusForwardedUpdateReceiverRuntime<TContext>(
  deps: TelegramBusForwardedUpdateReceiverRuntimeDeps<TContext>,
): TelegramBusForwardedUpdateReceiverRuntime {
  const server = createTelegramBusLocalServer({
    socketPath: deps.socketPath,
    recordTransportEvent(phase, details) {
      deps.recordRuntimeEvent?.("bus", `Telegram bus ${phase}`, {
        phase: `follower-receiver-${phase}`,
        ...details,
      });
    },
    async handleEnvelope(envelope) {
      const authSecret = deps.getAuthSecret?.();
      if (
        deps.getAuthSecret &&
        (!authSecret || !isTelegramBusEnvelopeAuthorized(envelope, authSecret))
      ) {
        return createUnauthorizedBusAck(envelope.requestId);
      }
      if (
        (envelope.kind !== "leader.forwardCallback" &&
          envelope.kind !== "leader.forwardReaction" &&
          envelope.kind !== "leader.forwardMessage" &&
          envelope.kind !== "leader.forwardEditedMessage" &&
          envelope.kind !== "leader.replaceFollowerTarget" &&
          envelope.kind !== "leader.offerQueueHandoff") ||
        envelope.recipientInstanceId !== deps.instanceId
      ) {
        return {
          kind: "bus.ack",
          requestId: envelope.requestId,
          ok: false,
          message: "Telegram bus receiver cannot handle this envelope.",
        };
      }
      const registrationGeneration = deps.getRegistrationGeneration();
      if (
        !registrationGeneration ||
        envelope.recipientRegistrationGeneration !== registrationGeneration
      ) {
        return {
          kind: "bus.ack",
          requestId: envelope.requestId,
          ok: false,
          message: "Stale Telegram bus follower registration generation.",
        };
      }
      if (
        envelope.kind !== "leader.replaceFollowerTarget" &&
        envelope.kind !== "leader.offerQueueHandoff" &&
        (!envelope.delivery ||
          envelope.delivery.recipientBindingKey !==
            deps.getRecipientBindingKey())
      ) {
        return {
          kind: "bus.ack",
          requestId: envelope.requestId,
          ok: false,
          message: "Mismatched Telegram follower delivery identity.",
        };
      }
      const ctx = deps.getContext();
      if (!ctx) {
        return {
          kind: "bus.ack",
          requestId: envelope.requestId,
          ok: false,
          message: "Telegram bus follower has no active context.",
        };
      }
      try {
        if (
          envelope.kind !== "leader.replaceFollowerTarget" &&
          envelope.kind !== "leader.offerQueueHandoff"
        ) {
          const receipt = await deps.durableAdmission.admit(envelope, ctx);
          return {
            kind: "bus.ack",
            requestId: envelope.requestId,
            ok: true,
            result: receipt,
          };
        }
        if (envelope.kind === "leader.offerQueueHandoff") {
          if (!deps.handleQueueHandoff) {
            throw new Error(
              "Telegram bus receiver cannot accept queue handoff payloads.",
            );
          }
          const result = await deps.handleQueueHandoff(envelope, ctx);
          const receipt = envelope.payload.admissionReceipts[0];
          if (
            envelope.payload.admissionReceipts.length !== 1 ||
            !receipt ||
            result.status !== "staged" ||
            result.receiptId !== receipt.receiptId ||
            result.sourceUpdateIds.length !== receipt.sourceUpdateIds.length ||
            result.sourceUpdateIds.some(
              (updateId, index) => updateId !== receipt.sourceUpdateIds[index],
            )
          ) {
            throw new Error(
              "Telegram queue handoff staging returned a mismatched receipt.",
            );
          }
          return {
            kind: "bus.ack",
            requestId: envelope.requestId,
            ok: true,
            result,
          };
        }
        {
          if (!deps.handleReplaceTarget) {
            throw new Error(
              "Telegram bus receiver cannot replace follower target.",
            );
          }
          await deps.handleReplaceTarget(
            {
              target: envelope.target,
              ...(envelope.oldTarget ? { oldTarget: envelope.oldTarget } : {}),
              reason: envelope.reason,
            },
            ctx,
          );
        }
        return { kind: "bus.ack", requestId: envelope.requestId, ok: true };
      } catch (error) {
        deps.recordRuntimeEvent?.("bus", error, { phase: "follower-forward" });
        return {
          kind: "bus.ack",
          requestId: envelope.requestId,
          ok: false,
          message:
            error instanceof Error
              ? error.message
              : "Telegram bus follower dispatch failed.",
        };
      }
    },
  });
  return server;
}

function parseRegistrationResult(value: unknown): {
  target?: TelegramTarget;
  slot?: string;
  threadName?: string;
} {
  if (!isRecord(value)) return {};
  const target = parseTarget(isRecord(value.target) ? value.target : value);
  return {
    ...(target ? { target } : {}),
    ...(typeof value.slot === "string" ? { slot: value.slot } : {}),
    ...(typeof value.threadName === "string"
      ? { threadName: value.threadName }
      : {}),
  };
}

function parseTarget(value: unknown): TelegramTarget | undefined {
  if (value === undefined) return undefined;
  if (!isRecord(value) || typeof value.chatId !== "number") return undefined;
  const target: TelegramTarget = { chatId: value.chatId };
  if (typeof value.threadId === "number") target.threadId = value.threadId;
  return target;
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}