/** * Forum-topic registry for eligible orchestration sessions on the threaded * Telegram surface. * * Each eligible orchestration session owns one active Telegram forum topic. * Remote archive closes daemon-created topics without deleting their durable * records; rotated successors move inactive records into retained history before * creating a new active authority. The registry also tracks whether the one-time * identity header has already been pinned. * * State is a plain serialisable map persisted beside the daemon state files; * topic creation is injected so this module is pure and unit-testable without a * live Bot API. */ /** Persisted record for one session's topic. */ export type TopicLifecycleState = "active" | "disconnect_grace" | "archive_pending" | "archive_exhausted" | "inactive" | "legacy_quarantined" /** Read-only input compatibility; normalized to archive_pending during load. */ | "delete_pending"; /** Persisted record for one immutable session identity's topic. */ export interface TopicRecord { /** Telegram forum topic id (message_thread_id). */ topicId: string; /** Whether Telegram created this topic for the daemon or it was explicitly adopted from a user. */ topicOrigin: "daemon_created" | "user_created"; /** Immutable UUID record identity; never derive authority from a title or PID. */ sessionUuid?: string; /** Whether the one-time identity header has been sent/pinned. */ identitySent: boolean; /** Creation timestamp (ms epoch). */ createdAt: number; /** First positive observation that the owning endpoint is stale, dead, or missing. */ orphanedAt?: number; /** Last applied or observed Telegram topic title. */ name?: string; /** Naming authority. Missing values are legacy daemon-owned records. */ nameOwner?: "user"; /** Whether a user-owned name still needs a best-effort Telegram re-assert. */ nameReconcilePending?: boolean; /** Last accepted Telegram update id for a user-owned name. */ userNameUpdateId?: number; /** Stable repo/branch identity used when topic names are user-owned or customized. */ identityKey?: string; /** Last SDK event generation durably consumed by the notification daemon. */ replayGeneration?: number; /** Last SDK event sequence durably consumed within replayGeneration. */ replaySeq?: number; /** Serialized authority epoch; a late create may commit only in its starting epoch. */ authorityEpoch?: number; /** Immutable authority epoch held when this remote topic create began. */ creationLeaseEpoch?: number; /** Durable non-destructive lifecycle state. */ authorityState?: TopicLifecycleState; /** Telegram chat and endpoint authority last proven to use this topic. */ chatId?: string; /** Stable provider binding; never contains SDK endpoint or attachment identity. */ telegramBinding?: TelegramTopicBinding; /** Why this record is archive-fenced; absence means no remote archive is authorized. */ archiveReason?: "session_closed" | "notification_subscription_removed" | "create_compensation"; /** Canonical endpoint tuple (URL + token) that currently holds the lease. */ endpointKey?: string; /** Authenticated endpoint authority digest, excluding transport presentation. */ endpointDigest?: string; /** SDK event generation associated with the current endpoint lease. */ endpointGeneration?: number; /** Monotonic authenticated endpoint handoffs; legacy bindings begin at zero. */ endpointIncarnation?: number; /** Shared-authority lease owner (installation UUID), heartbeat, and expiry. */ leaseOwner?: string; leaseHeartbeatAt?: number; leaseExpiresAt?: number; /** Durable archive initiator; a live foreign owner cannot be displaced. */ archiveHostId?: string; /** Authority epoch captured by the archive initiator when it published the fence. */ archiveLeaseEpoch?: number; disconnectGraceExpiresAt?: number; /** True when persisted binding fields were present but malformed; recovery must fail closed. */ bindingMalformed?: true; } /** Durable claim published before invoking createForumTopic. */ export interface TopicCreateClaim { sessionId: string; hostId?: string; leaseOwner?: string; authorityEpoch: number; createdAt: number; telegramBinding?: TelegramTopicBinding; /** Legacy v2 input only; never emitted by serialize(). */ binding?: TopicEndpointBinding; } export interface ArchiveJob { sessionId: string; topicId: string; /** Number of archive calls that have returned an ambiguous outcome. */ attempt: number; /** First ambiguous result; bounds retry lifetime. */ firstAttemptAt?: number; /** Compatibility read for pre-journal snapshots; normalized to `attempt`. */ retryCount?: number; backoffMs: number; nextAttemptAt: number; safeDiagnostic?: string; } /** Serialisable shape persisted to disk. */ export interface TopicRegistryState { /** Writer format. Missing is the quarantined legacy format; future versions fail closed. */ version?: 2; /** Monotonically increasing snapshot generation used by shared CAS stores. */ registryGeneration?: number; /** sessionId -> record. */ topics: Record; /** Durable lifecycle epochs retained after an archive starts. */ fences?: Record; /** Closed durable Telegram bindings; values intentionally contain no SDK endpoint identity. */ closedEndpoints?: Record; /** Persistent host identity used to distinguish concurrent installations. */ installationHostId?: string; /** Bounded durable archive work; no topic record is physically deleted. */ archiveJobs?: Record; /** Durable create claims. A claim fences concurrent creators before remote I/O. */ createClaims?: Record; /** Retained inactive predecessors keyed by logical session id. */ retiredTopics?: Record; } /** * Shared registry authority. Filesystem atomic rename is sufficient for a single * installation, but it cannot serialize two hosts sharing a state volume. */ export interface TopicRegistryCasAuthority { read(): Promise; compareAndSet(expectedGeneration: number, next: TopicRegistryState): Promise; } /** Durable Telegram authority. This is the only binding shape writers persist. */ export interface TelegramTopicBinding { chatId: string; transport: "telegram"; } export type TopicArchiveReason = NonNullable; /** Authenticated runtime binding for a durable topic lease. */ export interface TopicEndpointBinding { chatId: string; transport?: "telegram"; logicalSessionId?: string; /** Legacy input-only fields. They are never authority or persistence fields. */ endpointKey?: string; endpointDigest?: string; endpointGeneration?: number; } /** Discriminated durable endpoint authority for identity-less replay admission. */ export type TopicEndpointAuthority = { state: "none"; } | { state: "unique"; sessionId: string; } | { state: "ambiguous"; }; /** Conditional rollback token for a delete fence publication. */ export interface TopicArchiveAuthoritySnapshot { sessionId: string; topicId?: string; authorityEpoch?: number; authorityState?: TopicRecord["authorityState"]; fenceEpoch?: number; /** Exact fenced record, retained to restore an in-memory tombstone after a failed clear publication. */ record?: TopicRecord; } export declare function emptyTopicRegistryState(): TopicRegistryState; /** * Reject snapshots written by a newer daemon. Missing versions are preserved as * evidence but quarantined: legacy records must never route or mutate remotely. */ export declare function parseTopicRegistryState(value: unknown): TopicRegistryState | undefined; /** * In-memory registry over a serialisable state. Topic creation is injected via * `getOrCreateTopic`'s `create` callback (the daemon supplies a real * `createForumTopic` call); reuse-on-resume is automatic when a record exists. */ export declare class TopicRegistry { #private; private readonly topics; /** Maps topicId -> sessionId for fast inbound routing. */ private readonly byTopic; /** In-flight create promises, keyed by session, to dedupe concurrent creates. */ private readonly inflight; /** Newly-created records being durably published; never routable until committed. */ private readonly staged; /** Socket-specific provenance for transient endpoint claims. */ private readonly transientClaimants; /** Endpoint claims registered before a remote topic create can publish a record. */ private readonly creatingBindings; /** Monotonic authority epochs, including deletion fences for absent records. */ private readonly epochs; /** Archive work is retained and retryable; records are never physically removed. */ private readonly archiveJobs; /** Durable pre-create claims; remote creation is forbidden until published. */ private readonly createClaims; /** Inactive predecessor evidence retained across successor generations. */ private readonly retiredTopics; /** Generation of the last loaded/published snapshot. */ private registryGeneration; constructor(state?: TopicRegistryState); /** Replace all runtime state after a successfully persisted staged publication. */ replace(state: TopicRegistryState): void; /** Merge serialized state and normalize authority fields from older releases. */ load(state: TopicRegistryState): void; private rebuildInboundRoutes; /** Resolve the owning session for a topic id (for fail-closed inbound routing). */ sessionForTopic(topicId: string): string | undefined; /** All session ids with a persisted topic record. */ sessionIds(): string[]; /** The existing topic record for a session, if any. */ get(sessionId: string): TopicRecord | undefined; /** Durable claims restored from disk require explicit authoritative reconciliation. */ pendingCreateClaims(): TopicCreateClaim[]; /** * Resolve a durable pre-create claim only with authoritative topic evidence. * An absent or malformed topic deliberately leaves the claim in place, fencing * subsequent creators after a crash. */ reconcileCreateClaim(sessionId: string, topic?: TopicRecord): boolean; /** Clear a claim only when the local create attempt proved no remote topic was accepted. */ abandonCreateClaim(sessionId: string, authorityEpoch: number): boolean; /** Current immutable authority epoch for a creation lease. */ authorityEpoch(sessionId: string): number; /** Whether this session has an active, unambiguous topic authority. */ isActiveUnambiguous(sessionId: string): boolean; /** * Pure read-only availability check for user-topic adoption. Rejects invalid * ids and any id already committed, delete-pending/fenced, staged, or * ambiguous, without mutating maps, epochs, or persistence. Used inside the * {@link getOrCreateTopic} create callback to fail-closed before adopting a * user-created topic id. */ isTopicIdAvailable(topicId: string): boolean; /** * Resolve endpoint authority for identity-less replay. An endpoint may bootstrap * only when no committed, staged, or pre-create claim can own it; malformed * partial bindings and deletion fences deliberately fail closed. */ endpointAuthority(binding: TopicEndpointBinding, excludedTransientClaimant?: object): TopicEndpointAuthority; /** Resolve a uniquely bound logical owner for callers that only need the owner. */ uniqueSessionForEndpoint(binding: TopicEndpointBinding): string | undefined; /** Whether this exact session owns the complete durable endpoint binding. */ matchesEndpoint(sessionId: string, binding: TopicEndpointBinding): boolean; /** * Retire a remotely settled inactive topic only when an authenticated * successor proves a different endpoint authority. The inactive predecessor * is serialized into retained history before the active slot is released. */ retireInactiveEndpointForSuccessor(sessionId: string, binding: TopicEndpointBinding): boolean; /** * Rebind an existing topic to an authenticated successor endpoint. The exact * logical session id is proved by replay before this method is called. A * rotated credential may replace only an inactive incumbent; concurrent * incumbents, cross-chat records, malformed evidence, collisions, and delete * fences remain fail-closed. */ bindEndpoint(sessionId: string, binding: TopicEndpointBinding, _activeEndpointKeys?: ReadonlySet, _allowEndpointRotation?: boolean): "bound" | "unchanged" | "rejected"; /** Undo a failed durable endpoint migration without disturbing concurrent metadata writers. */ restoreEndpointBinding(sessionId: string, expected: TopicEndpointBinding, previous: Pick): boolean; /** * Return the existing active topic for `sessionId`, or create one via * `create` (called only on first use). */ getOrCreateTopic(sessionId: string, create: () => Promise, now?: () => number, name?: string, binding?: TopicEndpointBinding, commit?: () => Promise, transientClaimant?: object, topicOrigin?: TopicRecord["topicOrigin"]): Promise; /** Mark the identity header as sent for a session. Idempotent. */ markIdentitySent(sessionId: string): void; /** Generation used as the compare value for a shared CAS publication. */ registryVersion(): number; /** Advance only after a successful shared compare-and-set publication. */ markRegistryPublished(generation: number): void; /** * Acquire or renew a host lease. Another unexpired host is never displaced; * a disconnected owner may resume the exact topic during its grace window. */ acquireLease(sessionId: string, hostId: string, now: number, ttlMs: number, graceMs: number): boolean; /** Record a disconnect without losing the topic identity needed for a grace resume. */ releaseLeaseToGrace(sessionId: string, hostId: string, now: number, graceMs: number): boolean; /** Whether the identity header still needs sending for this session. */ needsIdentity(sessionId: string): boolean; /** Remember stable repo/branch identity independently of the displayed name. */ markIdentityKey(sessionId: string, identityKey: string): boolean; /** Start the orphan grace clock on the first positive liveness-loss observation. */ markOrphaned(sessionId: string, now: number): boolean; /** Clear a prior orphan observation after the endpoint is positively live again. */ clearOrphaned(sessionId: string): boolean; /** Last durably consumed SDK event cursor for reconnect replay. */ replayCursor(sessionId: string): { generation: number; seq: number; } | undefined; /** Advance the durable reconnect cursor without allowing stale responses to move it backwards. */ markReplayCursor(sessionId: string, generation: number, seq: number): boolean; /** Whether daemon identity reconciliation should apply `name`. */ needsRename(sessionId: string, name: string): boolean; /** The user-owned name that must be preserved, when one exists. */ userOwnedName(sessionId: string): string | undefined; /** A user-owned name whose Telegram reconciliation is still pending. */ userNameToReconcile(sessionId: string): string | undefined; /** Record an explicit Telegram-side user rename, rejecting stale update ids. */ markUserName(sessionId: string, name: string, updateId: number): "updated" | "duplicate" | "stale"; /** Mark the matching preserved user name as reconciled with Telegram. */ markUserNameReconciled(sessionId: string, name: string): boolean; /** Restore retryable reconciliation after a failed pending-clear persistence. */ markUserNamePending(sessionId: string, name: string): boolean; /** Commit a successfully-applied daemon topic title. */ markNameApplied(sessionId: string, name: string): void; /** Capture only authority fields that a failed archive-fence publication may restore. */ captureArchiveAuthority(sessionId: string): TopicArchiveAuthoritySnapshot; /** Restore a failed archive fence only while its exact authority mutation remains current. */ restoreArchiveAuthority(snapshot: TopicArchiveAuthoritySnapshot): boolean; /** Restore the exact archive fence after a failed compensation publication. */ restoreArchiveFence(snapshot: TopicArchiveAuthoritySnapshot): boolean; /** Fence new work before the remote archive starts, including an absent in-flight create. */ beginArchive(sessionId: string, hostId: string | undefined, now: number, reason: TopicArchiveReason): TopicRecord | undefined; /** Verify the durable archive initiator immediately before remote dispatch. */ archiveAuthorityAllows(sessionId: string, hostId: string, pairedChatId: string, now: number): boolean; /** @deprecated Production dispatch must provide the current paired chat id. */ archiveAuthorityAllows(sessionId: string, hostId: string, now: number): boolean; /** Retain an accepted create as deletion-fenced before remote compensation can begin. */ fenceAcceptedCreate(sessionId: string, topicId: string, now?: () => number, name?: string, binding?: TopicEndpointBinding, topicOrigin?: TopicRecord["topicOrigin"], archiveChatId?: string, reason?: TopicArchiveReason): TopicRecord; /** Fence an accepted create only when its exact creator lease still owns the record. */ fenceAcceptedCreateForLease(sessionId: string, topicId: string, creationLeaseEpoch: number, hostId: string, now?: () => number, name?: string, binding?: TopicEndpointBinding, topicOrigin?: TopicRecord["topicOrigin"], archiveChatId?: string, reason?: TopicArchiveReason): TopicRecord | undefined; /** Wait for a revoked create to settle before admitting a later lifecycle epoch. */ awaitInflight(sessionId: string): Promise; /** * Retain a topic record after a definite remote archive only while the exact * dispatched authority epoch is still current. */ settleArchive(sessionId: string, topicId: string, dispatchedAuthorityEpoch: number, reason: TopicArchiveReason): boolean; /** Durable archive jobs that are eligible for a retry at `now`. */ archivePendingSessionIds(now?: number): string[]; /** Durable/manual recovery candidates that exceeded the automatic archive retry budget. */ archiveExhaustedSessionIds(): string[]; /** Persist an indefinitely discoverable retry after an ambiguous archive result. */ scheduleArchiveRetry(sessionId: string, now: number, reason: TopicArchiveReason, diagnostic?: string): ArchiveJob | undefined; /** Serialise active records plus unpublished staged creates for atomic commit. */ serialize(): TopicRegistryState; }