vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-05 06:01:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(telegram): drive ingress through the core drain

Browse Source 
Both transports now enqueue durably and pump the shared drain: polling keeps
offset-after-committed-enqueue, webhook keeps 200-after-spool-write. Deletes
the reply fence, private claim-owner, retry policy, and per-transport
claim/replay/watchdog loops. Dispatch outcomes propagate: failed-retryable
releases for replay instead of tombstoning. Supersede policy stays
Telegram-owned, authorization-gated with pairing-store and owner identities
via the same resolver as normal ingress; room_event ambient pending remains
supersedable by a later user turn (#108656).
This commit is contained in:
Ayaan Zaidi
2026-07-16 21:18:10 +05:30
parent 16c14e5bbf
commit f7786a16cf
18 changed files with 974 additions and 2255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
extensions/telegram/src/bot-message-dispatch-fence.ts
View File

@@ -1,117 +0,0 @@
// Telegram plugin module owns pre-adoption reply-fence authority.
import type { TelegramMessageContext } from "./bot-message-context.js";
import type { DispatchTelegramMessageParams } from "./bot-message-dispatch.types.js";
import { getTelegramSequentialKey } from "./sequential-key.js";
import {
beginTelegramReplyFence,
buildTelegramNonInterruptingReplyFenceKey,
buildTelegramReplyFenceLaneKey,
endTelegramReplyFence,
isTelegramReplyFenceSuperseded,
releaseTelegramReplyFenceAbortController,
resolveTelegramReplyFenceKey,
shouldSupersedeTelegramReplyFence,
supersedeTelegramReplyFence,
} from "./telegram-reply-fence.js";
type CreateTelegramReplyFenceParams = Pick<
DispatchTelegramMessageParams,
"onTurnAdopted" | "onTurnDeferred" | "onTurnAbandoned" | "turnAbortSignal"
> & {
context: TelegramMessageContext;
};
export function createTelegramReplyFenceController(params: CreateTelegramReplyFenceParams) {
const { context } = params;
const replyFenceKey = resolveTelegramReplyFenceKey({
ctxPayload: context.ctxPayload,
chatId: context.chatId,
threadSpec: context.threadSpec,
});
const sequentialKey = getTelegramSequentialKey({
message: context.msg,
...(context.primaryCtx.me ? { me: context.primaryCtx.me } : {}),
});
const laneKey = buildTelegramReplyFenceLaneKey({
accountId: context.route.accountId,
sequentialKey,
});
const supersedes = shouldSupersedeTelegramReplyFence(context.ctxPayload);
const activeKey = supersedes
? replyFenceKey.activeKey
: buildTelegramNonInterruptingReplyFenceKey({
activeKey: replyFenceKey.activeKey,
laneKey,
});
// Ambient room-event work uses a separate fence key. Any non-room-event
// inbound may cancel it without owning abort authority over adopted user turns.
if (context.ctxPayload.InboundEventKind !== "room_event") {
supersedeTelegramReplyFence(replyFenceKey.roomEventKey);
}
const abortController = new AbortController();
const abortSignal = params.turnAbortSignal
? AbortSignal.any([abortController.signal, params.turnAbortSignal])
: abortController.signal;
let generation: number | undefined = beginTelegramReplyFence({
key: activeKey,
supersede: supersedes,
abortController,
laneKey,
});
let abortControllerQueued = false;
let queuedTurnAdmitted = false;
const isSuperseded = () =>
abortController.signal.aborted ||
(generation !== undefined && isTelegramReplyFenceSuperseded({ key: activeKey, generation }));
const release = () => {
if (generation === undefined) {
return;
}
endTelegramReplyFence(activeKey, abortControllerQueued ? undefined : abortController);
generation = undefined;
};
const adoptTurn = async () => {
await params.onTurnAdopted?.();
// Fence abort and supersession authority end after durable adoption.
// Core then owns all interruption of the adopted run.
release();
releaseTelegramReplyFenceAbortController(activeKey, abortController);
};
return {
abortSignal,
adoptTurn,
generation: () => generation,
isSuperseded,
release,
queuedFollowupLifecycle:
context.ctxPayload.InboundEventKind === "room_event" ||
params.onTurnAdopted ||
params.onTurnDeferred ||
params.onTurnAbandoned
? {
onEnqueued: () => {
abortControllerQueued = true;
params.onTurnDeferred?.();
},
onAdmitted: async () => {
await adoptTurn();
queuedTurnAdmitted = true;
},
onComplete: () => {
abortControllerQueued = false;
releaseTelegramReplyFenceAbortController(activeKey, abortController);
if (!queuedTurnAdmitted) {
params.onTurnAbandoned?.();
}
},
}
: undefined,
};
}
export type TelegramReplyFenceController = ReturnType<typeof createTelegramReplyFenceController>;

3
extensions/telegram/src/bot-message-dispatch-reply.ts
View File

@@ -13,7 +13,6 @@ import type { TelegramBotDeps } from "./bot-deps.js";
import type { TelegramMessageContext } from "./bot-message-context.js";
import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js";
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js";
import type { TelegramProgressController } from "./bot-message-dispatch-progress.js";
import { deduplicateBlockSentMedia } from "./bot-message-dispatch.media-dedup.js";
import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js";
@@ -59,7 +58,7 @@ export function createTelegramReplyDelivery(params: {
context: TelegramMessageContext;
delivery: TelegramDeliveryController;
draft: TelegramDraftController;
fence: Pick<TelegramReplyFenceController, "generation" | "isSuperseded">;
fence: { generation: () => number; isSuperseded: () => boolean };
progress: TelegramProgressController;
runtime: RuntimeEnv;
state: TelegramDispatchTurnState;

24
extensions/telegram/src/bot-message-dispatch-turn.ts
View File

@@ -12,7 +12,6 @@ import type { TelegramBotDeps } from "./bot-deps.js";
import type { TelegramMessageContext } from "./bot-message-context.js";
import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js";
import type { TelegramDraftController } from "./bot-message-dispatch-draft.js";
import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js";
import type { TelegramProgressController } from "./bot-message-dispatch-progress.js";
import type { TelegramReplyDelivery } from "./bot-message-dispatch-reply.js";
import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js";
@@ -26,7 +25,15 @@ export async function runTelegramDispatchTurn(params: {
context: TelegramMessageContext;
delivery: TelegramDeliveryController;
draft: TelegramDraftController;
fence: TelegramReplyFenceController;
/** Pre-adoption abort + lifecycle from durable ingress (optional for non-spooled). */
turnAdoptionLifecycle?: {
admission?: "exclusive" | "cancel-only";
onAdopted: () => void | Promise<void>;
onDeferred?: () => void;
onAbandoned?: () => void;
abortSignal?: AbortSignal;
};
isSuperseded: () => boolean;
progress: TelegramProgressController;
reply: TelegramReplyDelivery;
state: TelegramDispatchTurnState;
@@ -110,13 +117,20 @@ export async function runTelegramDispatchTurn(params: {
replyOptions: {
skillFilter: context.skillFilter,
disableBlockStreaming: params.draft.disableBlockStreaming,
abortSignal: params.fence.abortSignal,
onTurnAdopted: params.fence.adoptTurn,
abortSignal: params.turnAdoptionLifecycle?.abortSignal,
turnAdoptionLifecycle: params.turnAdoptionLifecycle
? {
admission: params.turnAdoptionLifecycle.admission ?? "exclusive",
onAdopted: params.turnAdoptionLifecycle.onAdopted,
onDeferred: params.turnAdoptionLifecycle.onDeferred,
onAbandoned: params.turnAdoptionLifecycle.onAbandoned,
abortSignal: params.turnAdoptionLifecycle.abortSignal,
}
: undefined,
sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined,
queuedDeliveryCorrelations: isRoomEvent
? [{ begin: beginDeliveryCorrelation }]
: undefined,
queuedFollowupLifecycle: params.fence.queuedFollowupLifecycle,
suppressTyping: isRoomEvent,
onPartialReply:
params.draft.answerLane.stream || params.draft.reasoningLane.stream

33
extensions/telegram/src/bot-message-dispatch.ts
View File

@@ -8,7 +8,6 @@ import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
import { resolveDispatchTelegramContext } from "./bot-message-dispatch-context.js";
import { createTelegramDeliveryController } from "./bot-message-dispatch-delivery.js";
import { createTelegramDraftController } from "./bot-message-dispatch-draft.js";
import { createTelegramReplyFenceController } from "./bot-message-dispatch-fence.js";
import { createTelegramProgressController } from "./bot-message-dispatch-progress.js";
import { createTelegramReplyDelivery } from "./bot-message-dispatch-reply.js";
import {
@@ -272,10 +271,7 @@ export const dispatchTelegramMessage = async ({
opts,
retryDispatchErrors = false,
suppressFailureFallback = false,
onTurnAdopted,
onTurnDeferred,
onTurnAbandoned,
turnAbortSignal,
turnAdoptionLifecycle,
}: DispatchTelegramMessageParams): Promise<TelegramDispatchResult> => {
const dispatchStartedAt = Date.now();
const dispatchContext = resolveDispatchTelegramContext({ context });
@@ -299,9 +295,9 @@ export const dispatchTelegramMessage = async ({
const forceBlockStreamingForReasoning =
resolvedReasoningLevel === "on" && streamMode !== "progress";
const quote = resolveTelegramQuoteContext({ context: dispatchContext, replyToMode });
// Controllers retain this callback but cannot run it before turn dispatch.
// Acquire the fence afterward so controller setup failures claim no ownership.
const isDispatchSuperseded = () => fence.isSuperseded();
// Pre-adoption abort is drain-owned via turnAdoptionLifecycle.abortSignal.
const isDispatchSuperseded = () => turnAdoptionLifecycle?.abortSignal?.aborted === true;
const dispatchGeneration = 0;
const draft = createTelegramDraftController({
accountId: dispatchContext.route.accountId,
bot,
@@ -369,7 +365,7 @@ export const dispatchTelegramMessage = async ({
delivery,
draft,
fence: {
generation: () => fence.generation(),
generation: () => dispatchGeneration,
isSuperseded: isDispatchSuperseded,
},
progress,
@@ -380,19 +376,12 @@ export const dispatchTelegramMessage = async ({
});
let isFirstTurnInSession = false;
let dispatchWasSuperseded: boolean;
let dispatchWasSuperseded = false;
let turnDispatched: boolean | undefined;
const isDmTopic =
!dispatchContext.isGroup &&
dispatchContext.threadSpec.scope === "dm" &&
dispatchContext.threadSpec.id != null;
const fence = createTelegramReplyFenceController({
context: dispatchContext,
onTurnAdopted,
onTurnDeferred,
onTurnAbandoned,
turnAbortSignal,
});
try {
await prepareTelegramSticker({ cfg, context: dispatchContext });
if (isDmTopic) {
@@ -418,7 +407,8 @@ export const dispatchTelegramMessage = async ({
context: dispatchContext,
delivery,
draft,
fence,
turnAdoptionLifecycle,
isSuperseded: isDispatchSuperseded,
progress,
reply,
state,
@@ -439,20 +429,19 @@ export const dispatchTelegramMessage = async ({
state.dispatchError ??= err;
runtime.error?.(danger(`telegram terminal block delivery failed: ${String(err)}`));
}
await draft.cleanup(fence.isSuperseded());
await draft.cleanup(isDispatchSuperseded());
if (
streamMode === "progress" &&
progress.sawProgressFinal() &&
!state.dispatchError &&
!state.hadErrorReplyFailureOrSkip &&
!fence.isSuperseded()
!isDispatchSuperseded()
) {
await delivery.deliverProgressCollapseSummary();
}
}
} finally {
dispatchWasSuperseded = fence.isSuperseded();
fence.release();
dispatchWasSuperseded = isDispatchSuperseded();
}
if (turnDispatched === false) {

19
extensions/telegram/src/bot-message-dispatch.types.ts
View File

@@ -26,14 +26,17 @@ export type DispatchTelegramMessageParams = {
opts: Pick<TelegramBotOptions, "token" | "mediaMaxMb">;
retryDispatchErrors?: boolean;
suppressFailureFallback?: boolean;
/** Fires after recovery-relevant session/run state is durably persisted. */
onTurnAdopted?: () => void | Promise<void>;
/** Marks a queued follow-up whose adoption will happen at reply-lane admission. */
onTurnDeferred?: () => void;
/** Releases a deferred turn that completed without ever owning the reply lane. */
onTurnAbandoned?: () => void;
/** Cancels queued/model work when ingress ownership fails before adoption. */
turnAbortSignal?: AbortSignal;
/**
* Canonical turn ownership lifecycle from the durable ingress drain
* (or a test double). Pre-adoption abort + adopt/defer/abandon.
*/
turnAdoptionLifecycle?: {
admission?: "exclusive" | "cancel-only";
onAdopted: () => void | Promise<void>;
onDeferred?: () => void;
onAbandoned?: () => void;
abortSignal?: AbortSignal;
};
};
export type TelegramDispatchResult =

84
extensions/telegram/src/bot-message.ts
View File

@@ -23,6 +23,7 @@ import {
createTelegramSpooledReplayParticipant,
createTelegramSpooledReplayDeferredParticipant,
getTelegramSpooledReplayDeferredParticipant,
getTelegramSpooledReplayLifecycle,
isTelegramSpooledReplayUpdate,
recordTelegramMessageProcessingResult,
type TelegramMessageProcessingResult,
@@ -34,7 +35,7 @@ import type { TelegramContext } from "./bot/types.js";
import type { TelegramReplyChainEntry } from "./message-cache.js";
import { TELEGRAM_TEXT_CHUNK_LIMIT } from "./outbound-adapter.js";
import { TELEGRAM_RICH_TEXT_LIMIT } from "./rich-message.js";
import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js";
import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./telegram-ingress-spool.js";
const telegramInboundLog = createSubsystemLogger("gateway/channels/telegram").child("inbound");
@@ -262,10 +263,13 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
await turnContext.onDispatchStart?.();
}
const runDispatch = async (params: {
onTurnAdopted?: () => void | Promise<void>;
onTurnDeferred?: () => void;
onTurnAbandoned?: () => void;
turnAbortSignal?: AbortSignal;
turnAdoptionLifecycle?: {
admission?: "exclusive" | "cancel-only";
onAdopted: () => void | Promise<void>;
onDeferred?: () => void;
onAbandoned?: () => void;
abortSignal?: AbortSignal;
};
}): Promise<TelegramMessageProcessingResult> => {
try {
const dispatchResult = await dispatchTelegramMessage({
@@ -281,10 +285,7 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
opts,
retryDispatchErrors: spooledReplay,
suppressFailureFallback: spooledReplay,
onTurnAdopted: params.onTurnAdopted,
onTurnDeferred: params.onTurnDeferred,
onTurnAbandoned: params.onTurnAbandoned,
turnAbortSignal: params.turnAbortSignal,
turnAdoptionLifecycle: params.turnAdoptionLifecycle,
});
if (dispatchResult?.kind === "failed-retryable") {
const result: TelegramMessageProcessingResult = {
@@ -389,34 +390,49 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
}
};
const run = async () => {
const turnAbortSignal = turnContext.spooledReplayAbortSignal
? AbortSignal.any([participant.abortSignal, turnContext.spooledReplayAbortSignal])
: participant.abortSignal;
const drainLifecycle = getTelegramSpooledReplayLifecycle();
// Participant always owns an AbortSignal on the spooled-replay path;
// merge optional drain/context signals without widening to undefined.
const turnAbortSignal: AbortSignal = (() => {
const extras = [turnContext.spooledReplayAbortSignal, drainLifecycle?.abortSignal].filter(
(signal): signal is AbortSignal => signal !== undefined,
);
if (extras.length === 0) {
return participant.abortSignal;
}
return AbortSignal.any([participant.abortSignal, ...extras]);
})();
const result = await runDispatch({
turnAbortSignal,
onTurnAdopted: async () => {
if (adopted) {
return;
}
adoptionAttempted = true;
const adoptedResult = await settle({ kind: "completed" }, "adopted");
if (adoptedResult.kind !== "completed") {
adoptionFinalizationError =
adoptedResult.kind === "failed-retryable"
turnAdoptionLifecycle: {
admission: "exclusive",
abortSignal: turnAbortSignal,
onAdopted: async () => {
if (adopted) {
return;
}
adoptionAttempted = true;
const adoptedResult = await settle({ kind: "completed" }, "adopted");
if (adoptedResult.kind !== "completed") {
adoptionFinalizationError =
adoptedResult.kind === "failed-retryable"
? adoptedResult.error
: new Error("telegram spooled turn adoption was not completed");
throw adoptedResult.kind === "failed-retryable"
? adoptedResult.error
: new Error("telegram spooled turn adoption was not completed");
throw adoptedResult.kind === "failed-retryable"
? adoptedResult.error
: new Error("telegram spooled turn adoption was not completed");
}
},
onTurnDeferred: () => {
deferred = true;
},
onTurnAbandoned: () => {
if (!adopted) {
void settle({ kind: "skipped" }, "terminal");
}
}
await drainLifecycle?.onAdopted();
},
onDeferred: () => {
deferred = true;
drainLifecycle?.onDeferred();
},
onAbandoned: () => {
if (!adopted) {
void settle({ kind: "skipped" }, "terminal");
}
drainLifecycle?.onAbandoned();
},
},
});
if (adopted) {

23
extensions/telegram/src/bot-processing-outcome.ts
View File

@@ -10,8 +10,18 @@ type TelegramUpdateProcessingFrame = {
result?: TelegramMessageProcessingResult;
};
export type TelegramSpooledReplayLifecycle = {
abortSignal: AbortSignal;
onAdopted: () => void | Promise<void>;
onDeferred: () => void;
/** Clears pre-adoption stall while durable adoption finalization is held. */
onAdoptionFinalizing?: () => void;
onAbandoned: () => void;
};
type TelegramSpooledReplayFrame = {
deferredWork?: TelegramSpooledReplayDeferredParticipant;
lifecycle?: TelegramSpooledReplayLifecycle;
};
export type TelegramSpooledReplayDeferredParticipant = {
@@ -95,6 +105,9 @@ export function createTelegramSpooledReplayParticipant(
return undefined;
}
settlementHeld = true;
// Timeout settlement must wait for durable adoption finalization: pause
// the drain stall watchdog while the hold is active.
telegramSpooledReplayFrames.getStore()?.lifecycle?.onAdoptionFinalizing?.();
let released = false;
return {
release: (mode) => {
@@ -145,8 +158,11 @@ export function getTelegramSpooledReplayDeferredParticipant():
export async function runWithTelegramSpooledReplayUpdate<T>(
update: object,
fn: () => Promise<T>,
lifecycle?: TelegramSpooledReplayLifecycle,
): Promise<{ value: T; deferredWork?: TelegramSpooledReplayDeferredParticipant }> {
const frame: TelegramSpooledReplayFrame = {};
const frame: TelegramSpooledReplayFrame = {
...(lifecycle ? { lifecycle } : {}),
};
telegramSpooledReplayUpdates.add(update);
try {
const value = await telegramSpooledReplayFrames.run(frame, fn);
@@ -156,6 +172,11 @@ export async function runWithTelegramSpooledReplayUpdate<T>(
}
}
/** Drain lifecycle for the active spooled-replay ALS frame, if any. */
export function getTelegramSpooledReplayLifecycle(): TelegramSpooledReplayLifecycle | undefined {
return telegramSpooledReplayFrames.getStore()?.lifecycle;
}
export function isTelegramSpooledReplayUpdate(update: unknown): boolean {
return (
telegramSpooledReplayFrames.getStore() !== undefined ||

885
extensions/telegram/src/polling-session.ts
View File

File diff suppressed because it is too large Load Diff

87
extensions/telegram/src/spooled-update-retry-policy.ts
View File

@@ -1,87 +0,0 @@
// Telegram plugin module shares spooled update retry policy.
import {
collectErrorGraphCandidates,
formatErrorMessage,
readErrorName,
} from "openclaw/plugin-sdk/error-runtime";
import type { BackoffPolicy } from "openclaw/plugin-sdk/runtime-env";
import { computeBackoff } from "openclaw/plugin-sdk/runtime-env";
import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-dedupe.js";
import type { TelegramSpooledUpdate } from "./telegram-ingress-spool.types.js";
export const TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS = 8;
const TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS = 24 * 60 * 60 * 1000;
const TELEGRAM_SPOOLED_RETRY_BASE_MS = 1_000;
const TELEGRAM_SPOOLED_RETRY_MAX_MS = 3 * 60_000;
const TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY: BackoffPolicy = {
initialMs: 250,
maxMs: 5_000,
factor: 2,
jitter: 0.2,
};
const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError";
const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u;
type NonRetryableSpooledUpdateFailure = {
reason: "missing-agent-harness" | "dispatch-dedupe-rollback-failed";
message: string;
};
export function resolveNonRetryableSpooledUpdateFailure(
err: unknown,
): NonRetryableSpooledUpdateFailure | null {
for (const candidate of collectErrorGraphCandidates(err, (current) => [
current.cause,
current.error,
])) {
const message = formatErrorMessage(candidate);
if (isTelegramMessageDispatchReplayForgetError(candidate)) {
// A committed dispatch key that cannot be rolled back makes retry unsafe:
// the next replay can be duplicate-suppressed and then deleted.
return { reason: "dispatch-dedupe-rollback-failed", message };
}
if (
readErrorName(candidate) === MISSING_AGENT_HARNESS_ERROR_NAME ||
MISSING_AGENT_HARNESS_MESSAGE_RE.test(message)
) {
return { reason: "missing-agent-harness", message };
}
}
return null;
}
export function resolveSpooledUpdateRetryDelayMs(
update: TelegramSpooledUpdate,
now = Date.now(),
): number {
const attempts = update.attempts ?? 0;
if (!update.lastError || update.lastAttemptAt === undefined || attempts <= 0) {
return 0;
}
const exponent = Math.min(attempts - 1, 8);
const delayMs = Math.min(
TELEGRAM_SPOOLED_RETRY_MAX_MS,
TELEGRAM_SPOOLED_RETRY_BASE_MS * 2 ** exponent,
);
return Math.max(0, update.lastAttemptAt + delayMs - now);
}
export function resolveSpooledUpdateAttemptNumber(update: TelegramSpooledUpdate): number {
return (update.attempts ?? 0) + 1;
}
export function resolveSpooledUpdatePersistenceRetryDelayMs(attempt: number): number {
return computeBackoff(TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY, attempt);
}
export function shouldDeadLetterRetryableSpooledUpdate(
update: TelegramSpooledUpdate,
attempt: number,
now = Date.now(),
): boolean {
return (
attempt >= TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS &&
now - update.receivedAt >= TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS
);
}

185
extensions/telegram/src/telegram-ingress-claim-owner.ts
View File

@@ -1,185 +0,0 @@
// Telegram plugin module implements telegram ingress claim-owner identity.
import childProcess from "node:child_process";
import { randomUUID } from "node:crypto";
import fsSync from "node:fs";
import type { ChannelIngressQueueCorruptClaim } from "openclaw/plugin-sdk/channel-outbound";
import type {
ClaimedTelegramSpooledUpdate,
TelegramSpooledUpdateClaimOwner,
} from "./telegram-ingress-spool.types.js";
// Liveness default: a claim older than its lease is never live-owner protected,
// so recovery can reclaim it even when the owner process still exists.
const TELEGRAM_SPOOLED_UPDATE_CLAIM_LEASE_MS = 30 * 60 * 1000;
type TelegramSpooledClaimLivenessOptions = {
maxAgeMs?: number;
now?: number;
/** Test seam for PID existence (including Linux TID impersonation). */
processExists?: (pid: number) => boolean;
/** Test seam for process start-time identity. */
readProcessStartTime?: (pid: number) => number | null;
};
function readProcessStartTime(pid: number): number | null {
if (!Number.isSafeInteger(pid) || pid <= 0) {
return null;
}
if (process.platform === "darwin") {
try {
const startedAt = childProcess
.execFileSync("/bin/ps", ["-o", "lstart=", "-p", String(pid)], {
encoding: "utf8",
env: { ...process.env, LC_ALL: "C", TZ: "UTC" },
stdio: ["ignore", "pipe", "ignore"],
})
.trim();
const startedAtMs = Date.parse(`${startedAt} UTC`);
return Number.isFinite(startedAtMs) ? Math.floor(startedAtMs / 1000) : null;
} catch {
return null;
}
}
if (process.platform !== "linux") {
return null;
}
try {
const stat = fsSync.readFileSync(`/proc/${pid}/stat`, "utf8");
const commEndIndex = stat.lastIndexOf(")");
if (commEndIndex < 0) {
return null;
}
const afterComm = stat.slice(commEndIndex + 1).trimStart();
const fields = afterComm.split(/\s+/);
// field 22 (starttime) = index 19 after the comm-split (field 3 is index 0).
const starttime = Number(fields[19]);
return Number.isInteger(starttime) && starttime >= 0 ? starttime : null;
} catch {
return null;
}
}
const TELEGRAM_SPOOLED_UPDATE_PROCESS_START_TIME = readProcessStartTime(process.pid);
// ownerId = pid:startToken:uuid. Starttime binds the PID to one process instance so
// Linux TIDs and recycled PIDs cannot impersonate a dead claim owner.
export const TELEGRAM_SPOOLED_UPDATE_PROCESS_ID = [
process.pid,
TELEGRAM_SPOOLED_UPDATE_PROCESS_START_TIME ?? "x",
randomUUID(),
].join(":");
export function processPidFromOwnerId(ownerId: string): number {
const pid = Number.parseInt(ownerId.split(":", 1)[0] ?? "", 10);
return Number.isSafeInteger(pid) && pid > 0 ? pid : -1;
}
// Canonical ownerId: pid:startToken:uuid. startToken is a numeric starttime, or
// the explicit "x" sentinel when the writer cannot supply one (win32).
type OwnerStartToken =
| { kind: "numeric"; value: number }
| { kind: "existence-only" }
| { kind: "missing" };
function parseOwnerStartToken(ownerId: string): OwnerStartToken {
const parts = ownerId.split(":");
// Legacy pid:uuid owners (pre start-token releases) carry no instance binding.
// Keep existence-based liveness for them: reclaiming a fresh claim from a live
// old-version worker during a rolling upgrade would double-dispatch its update.
if (parts.length === 2) {
return { kind: "existence-only" };
}
if (parts.length < 2) {
return { kind: "missing" };
}
const startField = parts[1] ?? "";
// Explicit "x": writer ran on a platform with no readable starttime (win32).
if (startField === "x") {
return { kind: "existence-only" };
}
const starttime = Number(startField);
if (Number.isSafeInteger(starttime) && starttime >= 0) {
return { kind: "numeric", value: starttime };
}
return { kind: "missing" };
}
function processExists(pid: number): boolean {
if (!Number.isSafeInteger(pid) || pid <= 0) {
return false;
}
try {
process.kill(pid, 0);
return true;
} catch (err) {
const code = (err as { code?: string }).code;
return code !== "ESRCH" && code !== "EINVAL";
}
}
function isFreshClaimOwner(
claim: TelegramSpooledUpdateClaimOwner,
options?: { maxAgeMs?: number; now?: number },
): boolean {
const now = options?.now ?? Date.now();
const maxAgeMs = options?.maxAgeMs ?? TELEGRAM_SPOOLED_UPDATE_CLAIM_LEASE_MS;
return now - claim.claimedAt < maxAgeMs;
}
function isClaimOwnerProcessInstanceLive(
claim: Pick<TelegramSpooledUpdateClaimOwner, "processId" | "processPid">,
options?: TelegramSpooledClaimLivenessOptions,
): boolean {
const exists = options?.processExists ?? processExists;
const readStart = options?.readProcessStartTime ?? readProcessStartTime;
if (!exists(claim.processPid)) {
return false;
}
const startToken = parseOwnerStartToken(claim.processId);
if (startToken.kind === "missing") {
// Legacy/malformed owner ids have no process-instance binding; reclaim.
return false;
}
if (startToken.kind === "existence-only") {
// Legacy or `x` owners cannot prove instance identity. Fall back to
// processExists-only liveness — the pre-starttime lease contract — instead
// of stealing a fresh claim from a possibly live worker.
return true;
}
const actualStart = readStart(claim.processPid);
if (actualStart === null) {
// Starttime unreadable while the PID appears live. Keep lease protection
// via process existence so a readable-starttime peer is not stolen mid-run.
return true;
}
return actualStart === startToken.value;
}
export function isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(
claim: ClaimedTelegramSpooledUpdate,
options?: TelegramSpooledClaimLivenessOptions,
): boolean {
return Boolean(
claim.claim &&
claim.claim.processId !== TELEGRAM_SPOOLED_UPDATE_PROCESS_ID &&
claim.claim.processPid !== process.pid &&
isFreshClaimOwner(claim.claim, options) &&
isClaimOwnerProcessInstanceLive(claim.claim, options),
);
}
export function isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess(
claim: ChannelIngressQueueCorruptClaim,
options?: TelegramSpooledClaimLivenessOptions,
): boolean {
const processId = claim.claim.ownerId;
const processPid = processPidFromOwnerId(processId);
const owner = { processId, processPid, claimedAt: claim.claim.claimedAt };
if (processId === TELEGRAM_SPOOLED_UPDATE_PROCESS_ID) {
return isFreshClaimOwner(owner, options);
}
return (
processPid !== process.pid &&
isFreshClaimOwner(owner, options) &&
isClaimOwnerProcessInstanceLive(owner, options)
);
}

65
extensions/telegram/src/telegram-ingress-drain-factory.ts Normal file
View File

@@ -0,0 +1,65 @@
// Telegram plugin module builds transport-shared durable ingress drains.
import type { ChannelIngressDrain } from "openclaw/plugin-sdk/channel-outbound";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import type { TelegramBotInfo } from "./bot-info.js";
import type { TelegramMessageProcessingResult } from "./bot-processing-outcome.js";
import {
createTelegramIngressDrain,
resolveTelegramAdoptionStallTimeoutMs,
type TelegramIngressDrainLifecycle,
} from "./telegram-ingress-drain.js";
import { openTelegramIngressQueue } from "./telegram-ingress-spool.js";
export type TelegramSpooledBot = {
handleUpdate: (update: never) => Promise<void>;
};
export type CreateTelegramTransportIngressDrainParams = {
spoolDir: string;
bot: TelegramSpooledBot;
cfg: OpenClawConfig;
accountId: string;
botInfo?: TelegramBotInfo;
adoptionStallTimeoutMs?: number;
onLog?: (message: string) => void;
abortSignal?: AbortSignal;
/**
* Optional override for full dispatch (tests). Default: bot.handleUpdate under
* the drain lifecycle via bot-message spooled replay path.
*/
dispatchUpdate?: (
update: unknown,
lifecycle: TelegramIngressDrainLifecycle,
) => Promise<TelegramMessageProcessingResult | void>;
};
/**
* One drain for polling + webhook: claim → dispatch with turnAdoptionLifecycle →
* complete at adoption. Transport code only enqueues then pumps drainOnce().
*/
export function createTelegramTransportIngressDrain(
params: CreateTelegramTransportIngressDrainParams,
): ChannelIngressDrain {
const queue = openTelegramIngressQueue(params.spoolDir);
const adoptionStallTimeoutMs = resolveTelegramAdoptionStallTimeoutMs({
configured: params.adoptionStallTimeoutMs,
env: process.env,
});
return createTelegramIngressDrain({
queue,
cfg: params.cfg,
accountId: params.accountId,
botInfo: params.botInfo,
adoptionStallTimeoutMs,
...(params.onLog ? { onLog: params.onLog } : {}),
...(params.abortSignal ? { abortSignal: params.abortSignal } : {}),
dispatch: async (update, lifecycle) => {
if (params.dispatchUpdate) {
return await params.dispatchUpdate(update, lifecycle);
}
// Lifecycle is also on the spooled ALS frame (runWithTelegramSpooledReplayUpdate).
// bot-message merges it into turnAdoptionLifecycle for complete-at-adoption.
await params.bot.handleUpdate(update as never);
},
});
}

176
extensions/telegram/src/telegram-ingress-drain.ts Normal file
View File

@@ -0,0 +1,176 @@
// Telegram plugin module owns the channel-side durable ingress drain adapter.
import {
bindIngressLifecycleToReplyOptions,
createChannelIngressDrain,
DEFAULT_INGRESS_ADOPTION_STALL_MS,
DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
type ChannelIngressDrain,
type ChannelIngressQueue,
} from "openclaw/plugin-sdk/channel-outbound";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { clampPositiveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
import type { TelegramBotInfo } from "./bot-info.js";
import {
runWithTelegramSpooledReplayUpdate,
type TelegramMessageProcessingResult,
} from "./bot-processing-outcome.js";
import { getTelegramSequentialKey } from "./sequential-key.js";
import { resolveTelegramIngressNonRetryableFailure } from "./telegram-ingress-non-retryable.js";
import type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js";
import { createShouldSupersedeTelegramSpooledPending } from "./telegram-ingress-supersede.js";
export const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS";
const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100;
const TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_SPOOLED_DRAIN_START_LIMIT * 10;
export function resolveTelegramAdoptionStallTimeoutMs(params: {
configured?: number;
env?: NodeJS.ProcessEnv;
}): number {
const candidates = [
params.configured,
Number(params.env?.[TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV]),
];
for (const candidate of candidates) {
const timeoutMs = clampPositiveTimerTimeoutMs(candidate);
if (timeoutMs !== undefined) {
return timeoutMs;
}
}
return DEFAULT_INGRESS_ADOPTION_STALL_MS;
}
export function telegramSpooledLaneKey(update: unknown, botInfo?: TelegramBotInfo): string {
return getTelegramSequentialKey({
update: update as Parameters<typeof getTelegramSequentialKey>[0]["update"],
...(botInfo ? { me: botInfo } : {}),
});
}
export type TelegramIngressDrainLifecycle = {
abortSignal: AbortSignal;
onAdopted: () => void | Promise<void>;
onDeferred: () => void;
onAdoptionFinalizing: () => void;
onAbandoned: () => void;
};
export type TelegramIngressDrainDispatch = (
update: unknown,
lifecycle: TelegramIngressDrainLifecycle,
) => Promise<TelegramMessageProcessingResult | void> | TelegramMessageProcessingResult | void;
export type CreateTelegramIngressDrainParams = {
queue: ChannelIngressQueue<TelegramSpooledUpdatePayload>;
/** Required for authorization-gated supersede (numeric allowlist). */
cfg: OpenClawConfig;
accountId: string;
botInfo?: TelegramBotInfo;
adoptionStallTimeoutMs?: number;
dispatch: TelegramIngressDrainDispatch;
onLog?: (message: string) => void;
abortSignal?: AbortSignal;
};
/**
* Shared polling/webhook drain over the core channel-ingress worker.
*
* room_event ambient work shares the sequential lane with the parent chat so a
* later user turn can supersede it pre-adoption; adopted user turns are never
* touched (core drain supersede is pre-adoption only).
*/
export function createTelegramIngressDrain(
params: CreateTelegramIngressDrainParams,
): ChannelIngressDrain {
return createChannelIngressDrain<TelegramSpooledUpdatePayload>({
queue: params.queue,
adoptionStallTimeoutMs: params.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS,
orderBy: "id",
scanLimit: TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT,
startLimit: TELEGRAM_SPOOLED_DRAIN_START_LIMIT,
retryPolicy: {
maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
},
resolveNonRetryableFailure: resolveTelegramIngressNonRetryableFailure,
shouldSupersedePending: createShouldSupersedeTelegramSpooledPending({
cfg: params.cfg,
accountId: params.accountId,
...(params.botInfo?.username ? { botUsername: params.botInfo.username } : {}),
}),
deriveLaneKey: (record) => telegramSpooledLaneKey(record.payload.update, params.botInfo),
...(params.onLog ? { onLog: params.onLog } : {}),
...(params.abortSignal ? { abortSignal: params.abortSignal } : {}),
dispatchClaimedEvent: async (event, lifecycle) => {
const bound = bindIngressLifecycleToReplyOptions(lifecycle);
const drainLifecycle: TelegramIngressDrainLifecycle = {
abortSignal: bound.turnAdoptionLifecycle.abortSignal,
onAdopted: bound.turnAdoptionLifecycle.onAdopted,
onDeferred: bound.turnAdoptionLifecycle.onDeferred,
onAdoptionFinalizing: lifecycle.onAdoptionFinalizing,
onAbandoned: bound.turnAdoptionLifecycle.onAbandoned,
};
try {
const result = await runWithTelegramSpooledReplayUpdate(
event.payload.update as object,
async () => await params.dispatch(event.payload.update, drainLifecycle),
drainLifecycle,
);
// Propagate explicit dispatch outcomes first.
const outcome = result.value;
if (outcome && typeof outcome === "object" && "kind" in outcome) {
if (outcome.kind === "failed-retryable") {
return { kind: "failed-retryable", error: outcome.error };
}
if (outcome.kind === "completed" || outcome.kind === "skipped") {
return { kind: "completed" };
}
}
// deferredWork exists for every spooled participant, not only genuine
// queued followups. Await the terminal participant outcome (or drain
// abort from guillotine/supersede) so failed-retryable releases and
// stalls do not hang the dispatch task forever.
const participant = result.deferredWork;
if (participant) {
const terminal = await new Promise<TelegramMessageProcessingResult>((resolve, reject) => {
if (drainLifecycle.abortSignal.aborted) {
reject(drainLifecycle.abortSignal.reason ?? new Error("ingress-aborted"));
return;
}
const onAbort = () => {
reject(drainLifecycle.abortSignal.reason ?? new Error("ingress-aborted"));
};
drainLifecycle.abortSignal.addEventListener("abort", onAbort, { once: true });
void participant.task.then(
(value) => {
drainLifecycle.abortSignal.removeEventListener("abort", onAbort);
resolve(value);
},
(error: unknown) => {
drainLifecycle.abortSignal.removeEventListener("abort", onAbort);
reject(error);
},
);
}).then(
(value) => value,
(error: unknown) => {
// Guillotine/supersede already own settleOnce — do not re-fail.
if (drainLifecycle.abortSignal.aborted) {
return { kind: "skipped" as const };
}
throw error;
},
);
if (terminal.kind === "failed-retryable") {
return { kind: "failed-retryable", error: terminal.error };
}
return { kind: "completed" };
}
return { kind: "completed" };
} catch (error) {
return { kind: "failed-retryable", error };
}
},
});
}

39
extensions/telegram/src/telegram-ingress-non-retryable.ts Normal file
View File

@@ -0,0 +1,39 @@
// Telegram plugin module classifies non-retryable spooled dispatch failures.
import {
collectErrorGraphCandidates,
formatErrorMessage,
readErrorName,
} from "openclaw/plugin-sdk/error-runtime";
import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-dedupe.js";
const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError";
const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u;
export type TelegramIngressNonRetryableFailure = {
reason: "missing-agent-harness" | "dispatch-dedupe-rollback-failed";
message: string;
};
/** Channel-owned non-retryable predicate for the core ingress drain. */
export function resolveTelegramIngressNonRetryableFailure(
err: unknown,
): TelegramIngressNonRetryableFailure | null {
for (const candidate of collectErrorGraphCandidates(err, (current) => [
current.cause,
current.error,
])) {
const message = formatErrorMessage(candidate);
if (isTelegramMessageDispatchReplayForgetError(candidate)) {
// A committed dispatch key that cannot be rolled back makes retry unsafe:
// the next replay can be duplicate-suppressed and then deleted.
return { reason: "dispatch-dedupe-rollback-failed", message };
}
if (
readErrorName(candidate) === MISSING_AGENT_HARNESS_ERROR_NAME ||
MISSING_AGENT_HARNESS_MESSAGE_RE.test(message)
) {
return { reason: "missing-agent-harness", message };
}
}
return null;
}

9
extensions/telegram/src/telegram-ingress-spool.payload.ts Normal file
View File

@@ -0,0 +1,9 @@
// Telegram plugin module defines durable ingress queue payload shape.
export type TelegramSpooledUpdatePayload = {
version: number;
updateId: number;
receivedAt: number;
update: unknown;
};
export const TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION = 1;

351
extensions/telegram/src/telegram-ingress-spool.ts
View File

@@ -1,24 +1,25 @@
// Telegram plugin module implements telegram ingress spool behavior.
// Telegram plugin module implements durable ingress enqueue + update_id mapping.
import os from "node:os";
import path from "node:path";
import type {
ChannelIngressQueue,
ChannelIngressQueueClaim,
ChannelIngressQueueClaimRef,
ChannelIngressQueueCorruptClaim,
ChannelIngressQueueRecord,
import {
INGRESS_CLAIM_PROCESS_ID,
processPidFromOwnerId,
type ChannelIngressQueue,
type ChannelIngressQueueClaim,
type ChannelIngressQueueClaimRef,
type ChannelIngressQueueCorruptClaim,
type ChannelIngressQueueRecord,
} from "openclaw/plugin-sdk/channel-outbound";
import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env";
import { computeBackoff, type BackoffPolicy } from "openclaw/plugin-sdk/runtime-env";
import { resolveStateDir } from "openclaw/plugin-sdk/state-paths";
import type { TelegramBotInfo } from "./bot-info.js";
import { getTelegramRuntime } from "./runtime.js";
import { getTelegramSequentialKey } from "./sequential-key.js";
import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js";
import { normalizeTelegramStateAccountId } from "./state-account-id.js";
import {
processPidFromOwnerId,
TELEGRAM_SPOOLED_UPDATE_PROCESS_ID,
} from "./telegram-ingress-claim-owner.js";
TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION,
type TelegramSpooledUpdatePayload,
} from "./telegram-ingress-spool.payload.js";
import type {
ClaimedTelegramSpooledUpdate,
TelegramSpooledUpdate,
@@ -28,29 +29,20 @@ export type {
ClaimedTelegramSpooledUpdate,
TelegramSpooledUpdate,
} from "./telegram-ingress-spool.types.js";
export type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js";
const SPOOL_VERSION = 1;
const TELEGRAM_INGRESS_SPOOL_PREFIX = "ingress-spool-";
const TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS = 6 * 60 * 60 * 1000;
const TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000;
const TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES = 1000;
const TELEGRAM_SPOOLED_UPDATE_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1000;
const TELEGRAM_SPOOLED_UPDATE_COMPLETED_MAX_ENTRIES = 1000;
type TelegramSpooledUpdatePayload = {
version: number;
updateId: number;
receivedAt: number;
update: unknown;
const TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY: BackoffPolicy = {
initialMs: 250,
maxMs: 5_000,
factor: 2,
jitter: 0.2,
};
class TelegramSpooledUpdateCompletionOwnershipError extends Error {
constructor(updateId: number) {
super(`Telegram spooled update ${updateId} lost claim ownership before completion.`);
this.name = "TelegramSpooledUpdateCompletionOwnershipError";
}
}
function isValidUpdateId(value: unknown): value is number {
return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
}
@@ -75,24 +67,12 @@ function resolveTelegramUpdateId(update: unknown): number | null {
return isValidUpdateId(value) ? value : null;
}
function spoolFileName(updateId: number): string {
return `${String(updateId).padStart(16, "0")}.json`;
}
function processingFileName(updateId: number): string {
return `${spoolFileName(updateId)}.processing`;
}
function queueEventId(updateId: number): string {
export function telegramQueueEventId(updateId: number): string {
return String(updateId).padStart(16, "0");
}
function pendingPath(spoolDir: string, updateId: number): string {
return path.join(spoolDir, spoolFileName(updateId));
}
function processingPath(spoolDir: string, updateId: number): string {
return path.join(spoolDir, processingFileName(updateId));
function spoolFileName(updateId: number): string {
return `${telegramQueueEventId(updateId)}.json`;
}
function resolveQueueParts(spoolDir: string): {
@@ -110,13 +90,11 @@ function resolveQueueParts(spoolDir: string): {
path.basename(path.dirname(spoolDir)) === "telegram"
? path.dirname(path.dirname(spoolDir))
: spoolDir;
return {
accountId,
stateDir,
};
return { accountId, stateDir };
}
function createTelegramIngressQueue(
/** Open the account-scoped durable ingress queue for this spool directory. */
export function openTelegramIngressQueue(
spoolDir: string,
): ChannelIngressQueue<TelegramSpooledUpdatePayload> {
const parts = resolveQueueParts(spoolDir);
@@ -126,6 +104,13 @@ function createTelegramIngressQueue(
});
}
export function telegramSpooledUpdateLaneKey(update: unknown, botInfo?: TelegramBotInfo): string {
return getTelegramSequentialKey({
update: update as Parameters<typeof getTelegramSequentialKey>[0]["update"],
...(botInfo ? { me: botInfo } : {}),
});
}
async function pruneTelegramIngressQueue(
queue: ChannelIngressQueue<TelegramSpooledUpdatePayload>,
now?: number,
@@ -139,62 +124,10 @@ async function pruneTelegramIngressQueue(
});
}
function parseQueueRecord(
spoolDir: string,
record: ChannelIngressQueueRecord<TelegramSpooledUpdatePayload>,
): TelegramSpooledUpdate | null {
const payload = record.payload;
if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) {
return null;
}
return {
updateId: payload.updateId,
path: pendingPath(spoolDir, payload.updateId),
update: payload.update,
receivedAt: payload.receivedAt,
attempts: record.attempts,
...(record.lastAttemptAt === undefined ? {} : { lastAttemptAt: record.lastAttemptAt }),
...(record.lastError === undefined ? {} : { lastError: record.lastError }),
};
}
function parseQueueClaim(
spoolDir: string,
record: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>,
): ClaimedTelegramSpooledUpdate | null {
const update = parseQueueRecord(spoolDir, record);
if (!update) {
return null;
}
return {
...update,
path: processingPath(spoolDir, update.updateId),
pendingPath: pendingPath(spoolDir, update.updateId),
claim: {
processId: record.claim.ownerId,
processPid: processPidFromOwnerId(record.claim.ownerId),
claimedAt: record.claim.claimedAt,
claimToken: record.claim.token,
},
};
}
function spooledUpdateLaneKey(update: unknown, botInfo?: TelegramBotInfo): string {
return getTelegramSequentialKey({
update: update as Parameters<typeof getTelegramSequentialKey>[0]["update"],
...(botInfo ? { me: botInfo } : {}),
});
}
function sortTelegramUpdates<T extends TelegramSpooledUpdate>(updates: T[]): T[] {
return updates.toSorted((a, b) => a.updateId - b.updateId);
}
function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef {
const id = queueEventId(update.updateId);
return update.claim?.claimToken ? { id, claim: { token: update.claim.claimToken } } : id;
}
/**
* Durable-before-ack accept path: commit the update to the ingress queue.
* Polling advances offset only after this returns; webhook returns 200 only after.
*/
export async function writeTelegramSpooledUpdate(params: {
spoolDir: string;
update: unknown;
@@ -206,19 +139,19 @@ export async function writeTelegramSpooledUpdate(params: {
throw new Error("Telegram update missing numeric update_id.");
}
const receivedAt = params.now ?? Date.now();
const queue = createTelegramIngressQueue(params.spoolDir);
const queue = openTelegramIngressQueue(params.spoolDir);
await pruneTelegramIngressQueue(queue, params.now);
await queue.enqueue(
queueEventId(updateId),
telegramQueueEventId(updateId),
{
version: SPOOL_VERSION,
version: TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION,
updateId,
receivedAt,
update: params.update,
},
{
receivedAt,
laneKey: params.laneKey ?? spooledUpdateLaneKey(params.update),
laneKey: params.laneKey ?? telegramSpooledUpdateLaneKey(params.update),
},
);
return updateId;
@@ -228,54 +161,78 @@ export async function listTelegramSpooledUpdates(params: {
spoolDir: string;
limit?: number | "all";
}): Promise<TelegramSpooledUpdate[]> {
const records = await createTelegramIngressQueue(params.spoolDir).listPending({
const records = await openTelegramIngressQueue(params.spoolDir).listPending({
limit: params.limit ?? 100,
orderBy: "id",
});
return sortTelegramUpdates(
records.flatMap((record) => {
const update = parseQueueRecord(params.spoolDir, record);
return records
.flatMap((record) => {
const update = parsePendingRecord(params.spoolDir, record);
return update ? [update] : [];
}),
);
})
.toSorted((a, b) => a.updateId - b.updateId);
}
async function completeTelegramSpooledUpdate(update: TelegramSpooledUpdate): Promise<boolean> {
const queue = createTelegramIngressQueue(path.dirname(update.path));
// Successful rows stay as bounded tombstones: Telegram can refetch an update
// after dispatch, and callbacks have side effects that plain delete would rerun.
return await queue.complete(queueMutationTarget(update));
function parsePendingRecord(
spoolDir: string,
record: ChannelIngressQueueRecord<TelegramSpooledUpdatePayload>,
): TelegramSpooledUpdate | null {
const payload = record.payload;
if (
payload.version !== TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION ||
!isValidUpdateId(payload.updateId)
) {
return null;
}
return {
updateId: payload.updateId,
path: path.join(spoolDir, spoolFileName(payload.updateId)),
update: payload.update,
receivedAt: payload.receivedAt,
attempts: record.attempts,
...(record.lastAttemptAt === undefined ? {} : { lastAttemptAt: record.lastAttemptAt }),
...(record.lastError === undefined ? {} : { lastError: record.lastError }),
};
}
export async function completeTelegramSpooledUpdateWithRetry(params: {
update: ClaimedTelegramSpooledUpdate;
abortSignal?: AbortSignal;
onRetry?: (retry: { attempt: number; delayMs: number; error: unknown }) => void;
}): Promise<void> {
if (!params.update.claim?.claimToken) {
throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId);
}
let attempt = 0;
while (true) {
try {
const completed = await completeTelegramSpooledUpdate(params.update);
if (!completed) {
throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId);
}
return;
} catch (err) {
if (
err instanceof TelegramSpooledUpdateCompletionOwnershipError ||
params.abortSignal?.aborted
) {
throw err;
}
attempt += 1;
const delayMs = resolveSpooledUpdatePersistenceRetryDelayMs(attempt);
params.onRetry?.({ attempt, delayMs, error: err });
await sleepWithAbort(delayMs, params.abortSignal);
}
/** Backoff for irrevocable-adoption completion retries (bot-message only). */
export function resolveSpooledUpdatePersistenceRetryDelayMs(attempt: number): number {
return computeBackoff(TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY, attempt);
}
// --- Thin queue claim helpers (transport tests + recovery tools) ---
// Drain loops live in core; these wrap openTelegramIngressQueue only.
function processingFileName(updateId: number): string {
return `${spoolFileName(updateId)}.processing`;
}
function parseQueueClaim(
spoolDir: string,
record: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>,
): ClaimedTelegramSpooledUpdate | null {
const update = parsePendingRecord(spoolDir, record);
if (!update) {
return null;
}
const claimRef = record.claim.token;
return {
...update,
path: path.join(spoolDir, processingFileName(update.updateId)),
pendingPath: path.join(spoolDir, spoolFileName(update.updateId)),
claim: {
processId: record.claim.ownerId,
processPid: processPidFromOwnerId(record.claim.ownerId),
claimedAt: record.claim.claimedAt,
claimToken: claimRef,
},
};
}
function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef {
const id = telegramQueueEventId(update.updateId);
const claimRef = update.claim?.claimToken;
return claimRef ? { id, claim: { token: claimRef } } : id;
}
export async function claimNextTelegramSpooledUpdate(params: {
@@ -285,16 +242,16 @@ export async function claimNextTelegramSpooledUpdate(params: {
candidateUpdateIds?: Iterable<number>;
scanLimit?: number;
}): Promise<ClaimedTelegramSpooledUpdate | null> {
const queue = createTelegramIngressQueue(params.spoolDir);
const queue = openTelegramIngressQueue(params.spoolDir);
const claimed = await queue.claimNext({
ownerId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID,
ownerId: INGRESS_CLAIM_PROCESS_ID,
blockedLaneKeys: params.blockedLaneKeys,
...(params.candidateUpdateIds === undefined
? {}
: { candidateIds: [...params.candidateUpdateIds].map(queueEventId) }),
: { candidateIds: [...params.candidateUpdateIds].map(telegramQueueEventId) }),
orderBy: "id",
scanLimit: params.scanLimit,
deriveLaneKey: (record) => spooledUpdateLaneKey(record.payload.update, params.botInfo),
deriveLaneKey: (record) => telegramSpooledUpdateLaneKey(record.payload.update, params.botInfo),
});
if (!claimed) {
return null;
@@ -310,68 +267,16 @@ export async function claimNextTelegramSpooledUpdate(params: {
return null;
}
export async function releaseTelegramSpooledUpdateClaim(
update: ClaimedTelegramSpooledUpdate,
options?: { lastError?: string; releasedAt?: number },
): Promise<void> {
await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
queueMutationTarget(update),
options,
);
}
export async function abandonTelegramSpooledUpdateClaim(
update: ClaimedTelegramSpooledUpdate,
): Promise<void> {
await createTelegramIngressQueue(path.dirname(update.pendingPath)).release(
queueMutationTarget(update),
{ recordAttempt: false },
);
}
export async function refreshTelegramSpooledUpdateClaim(
update: ClaimedTelegramSpooledUpdate,
options?: { refreshedAt?: number },
): Promise<boolean> {
const claimToken = update.claim?.claimToken;
if (!claimToken) {
return false;
}
const queue = createTelegramIngressQueue(path.dirname(update.pendingPath));
return (
(await queue.refreshClaim?.(
{ id: queueEventId(update.updateId), claim: { token: claimToken } },
options,
)) ?? false
);
}
export async function failTelegramSpooledUpdateClaim(params: {
update: ClaimedTelegramSpooledUpdate;
reason: string;
message: string;
now?: number;
}): Promise<boolean> {
const queue = createTelegramIngressQueue(path.dirname(params.update.pendingPath));
const failed = await queue.fail(queueMutationTarget(params.update), {
reason: params.reason,
message: params.message,
...(params.now === undefined ? {} : { failedAt: params.now }),
});
await pruneTelegramIngressQueue(queue, params.now);
return failed;
}
export async function listTelegramSpooledUpdateClaims(params: {
spoolDir: string;
}): Promise<ClaimedTelegramSpooledUpdate[]> {
const claims = await createTelegramIngressQueue(params.spoolDir).listClaims();
return sortTelegramUpdates(
claims.flatMap((claim) => {
const claims = await openTelegramIngressQueue(params.spoolDir).listClaims();
return claims
.flatMap((claim) => {
const update = parseQueueClaim(params.spoolDir, claim);
return update ? [update] : [];
}),
);
})
.toSorted((a, b) => a.updateId - b.updateId);
}
export async function recoverStaleTelegramSpooledUpdateClaims(params: {
@@ -383,12 +288,12 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: {
}): Promise<number> {
const shouldRecover = params.shouldRecover;
const shouldRecoverCorrupt = params.shouldRecoverCorrupt;
return await createTelegramIngressQueue(params.spoolDir).recoverStaleClaims({
staleMs: params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS,
return await openTelegramIngressQueue(params.spoolDir).recoverStaleClaims({
staleMs: params.staleMs ?? 0,
...(params.now === undefined ? {} : { now: params.now }),
...(shouldRecover
? {
shouldRecover: async (claim: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>) => {
shouldRecover: async (claim) => {
const update = parseQueueClaim(params.spoolDir, claim);
return update ? await shouldRecover(update) : false;
},
@@ -397,3 +302,29 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: {
...(shouldRecoverCorrupt ? { shouldRecoverCorrupt } : {}),
});
}
export async function releaseTelegramSpooledUpdateClaim(
update: ClaimedTelegramSpooledUpdate,
options?: { lastError?: string; releasedAt?: number },
): Promise<void> {
await openTelegramIngressQueue(path.dirname(update.pendingPath)).release(
queueMutationTarget(update),
options,
);
}
export async function failTelegramSpooledUpdateClaim(params: {
update: ClaimedTelegramSpooledUpdate;
reason: string;
message: string;
now?: number;
}): Promise<boolean> {
return await openTelegramIngressQueue(path.dirname(params.update.pendingPath)).fail(
queueMutationTarget(params.update),
{
reason: params.reason,
message: params.message,
...(params.now === undefined ? {} : { failedAt: params.now }),
},
);
}

374
extensions/telegram/src/telegram-ingress-supersede.ts Normal file
View File

@@ -0,0 +1,374 @@
import type {
ChannelIngressQueueClaim,
ChannelIngressQueueRecord,
} from "openclaw/plugin-sdk/channel-outbound";
import {
maybeResolveTextAlias,
normalizeCommandBody,
} from "openclaw/plugin-sdk/command-auth-native";
import {
isAbortRequestText,
isBtwRequestText,
} from "openclaw/plugin-sdk/command-primitives-runtime";
// Telegram plugin module owns pre-adoption supersede policy for durable ingress.
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { resolveTelegramDmAllow } from "./access-groups.js";
import { mergeTelegramAccountConfig } from "./account-config.js";
import {
resolveTelegramCommandAuthorization,
resolveTelegramGroupAllowFromContext,
resolveTelegramMessageForumFlagHint,
} from "./bot/helpers.js";
import { resolveTelegramScopedGroupConfig } from "./group-config-helpers.js";
import { resolveTelegramCommandIngressAuthorization } from "./ingress.js";
import { isTelegramReadOnlyControlLaneText } from "./sequential-key.js";
import type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js";
function isRecognizedTelegramTextCommand(rawText: string, botUsername?: string): boolean {
return (
maybeResolveTextAlias(
normalizeCommandBody(rawText, botUsername ? { botUsername } : undefined),
) != null
);
}
/**
* Whether a bot_command entity (or slash token) targets this bot.
* Same target rule as normalizeCommandBody: untargeted commands match any bot;
* @OtherBot is ignored when our identity is known.
*/
function isTelegramCommandTargetedAtBot(commandText: string, botUsername?: string): boolean {
const trimmed = commandText.trim();
if (!trimmed.startsWith("/")) {
return false;
}
// normalizeCommandBody only strips @bot when the target equals botUsername.
// A non-matching @target leaves the body as `/cmd@other`, which is not ours.
const normalized = normalizeCommandBody(
trimmed,
botUsername ? { botUsername } : undefined,
).trim();
if (!normalized.startsWith("/")) {
return false;
}
// Untargeted, or successfully stripped for this bot.
if (!/^\/[^\s@]+@/u.test(normalized)) {
return true;
}
// Identity unknown: keep untargeted-permissive behavior for pre-getMe drains.
return !botUsername?.trim();
}
/** True when the update carries a bot_command entity addressed to this bot. */
function updateHasBotCommandEntityForBot(update: unknown, botUsername?: string): boolean {
if (!update || typeof update !== "object") {
return false;
}
const root = update as Record<string, unknown>;
for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) {
const msg = root[key];
if (!msg || typeof msg !== "object") {
continue;
}
const message = msg as {
text?: unknown;
caption?: unknown;
entities?: unknown;
caption_entities?: unknown;
};
const body =
typeof message.text === "string"
? message.text
: typeof message.caption === "string"
? message.caption
: "";
for (const entities of [message.entities, message.caption_entities]) {
if (!Array.isArray(entities)) {
continue;
}
for (const entity of entities) {
if (!entity || typeof entity !== "object") {
continue;
}
const ent = entity as { type?: unknown; offset?: unknown; length?: unknown };
if (ent.type !== "bot_command") {
continue;
}
if (typeof ent.offset !== "number" || typeof ent.length !== "number") {
continue;
}
const commandText = body.slice(ent.offset, ent.offset + ent.length);
if (isTelegramCommandTargetedAtBot(commandText, botUsername)) {
return true;
}
}
}
}
return false;
}
function extractUpdateText(update: unknown): string {
if (!update || typeof update !== "object") {
return "";
}
const root = update as Record<string, unknown>;
for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) {
const msg = root[key];
if (msg && typeof msg === "object") {
const text = (msg as { text?: unknown; caption?: unknown }).text;
if (typeof text === "string") {
return text;
}
const caption = (msg as { caption?: unknown }).caption;
if (typeof caption === "string") {
return caption;
}
}
}
const callback = root.callback_query;
if (callback && typeof callback === "object") {
const data = (callback as { data?: unknown }).data;
if (typeof data === "string") {
return data;
}
}
return "";
}
type UpdateSenderFacts = {
senderId: string;
senderUsername?: string;
chatId: number;
chatType?: string;
isGroup: boolean;
isTopicMessage?: boolean;
isForum?: boolean;
messageThreadId?: number;
};
function extractUpdateSenderFacts(update: unknown): UpdateSenderFacts | null {
if (!update || typeof update !== "object") {
return null;
}
const root = update as Record<string, unknown>;
let message: Record<string, unknown> | undefined;
for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) {
const candidate = root[key];
if (candidate && typeof candidate === "object") {
message = candidate as Record<string, unknown>;
break;
}
}
if (!message) {
const callback = root.callback_query;
if (callback && typeof callback === "object") {
const cb = callback as Record<string, unknown>;
const from = cb.from;
const msg = cb.message;
if (from && typeof from === "object" && msg && typeof msg === "object") {
message = msg as Record<string, unknown>;
const chat = (msg as { chat?: { id?: unknown; type?: unknown; is_forum?: unknown } }).chat;
const fromObj = from as { id?: unknown; username?: unknown };
if (typeof chat?.id === "number" && typeof fromObj.id === "number") {
const chatType = typeof chat.type === "string" ? chat.type : "private";
return {
senderId: String(fromObj.id),
...(typeof fromObj.username === "string" ? { senderUsername: fromObj.username } : {}),
chatId: chat.id,
chatType,
isGroup: chatType !== "private",
...(typeof chat.is_forum === "boolean" ? { isForum: chat.is_forum } : {}),
...(typeof (msg as { is_topic_message?: unknown }).is_topic_message === "boolean"
? {
isTopicMessage: (msg as { is_topic_message: boolean }).is_topic_message,
}
: {}),
...(typeof (msg as { message_thread_id?: unknown }).message_thread_id === "number"
? {
messageThreadId: (msg as { message_thread_id: number }).message_thread_id,
}
: {}),
};
}
}
}
return null;
}
const chat = message.chat as { id?: unknown; type?: unknown; is_forum?: unknown } | undefined;
const from = message.from as { id?: unknown; username?: unknown } | undefined;
if (typeof chat?.id !== "number" || typeof from?.id !== "number") {
return null;
}
const chatType = typeof chat.type === "string" ? chat.type : "private";
return {
senderId: String(from.id),
...(typeof from.username === "string" ? { senderUsername: from.username } : {}),
chatId: chat.id,
chatType,
isGroup: chatType !== "private",
...(typeof chat.is_forum === "boolean" ? { isForum: chat.is_forum } : {}),
...(typeof message.is_topic_message === "boolean"
? { isTopicMessage: message.is_topic_message as boolean }
: {}),
...(typeof message.message_thread_id === "number"
? { messageThreadId: message.message_thread_id as number }
: {}),
};
}
/** Ambient room_event-shaped updates (no user text body) stay supersedable. */
export function isTelegramAmbientSpooledUpdate(update: unknown): boolean {
if (!update || typeof update !== "object") {
return false;
}
const root = update as Record<string, unknown>;
return (
root.message_reaction != null ||
root.message_reaction_count != null ||
root.chat_member != null ||
root.my_chat_member != null ||
root.chat_join_request != null ||
root.chat_boost != null ||
root.removed_chat_boost != null
);
}
export type TelegramSupersedeAuthContext = {
cfg: OpenClawConfig;
accountId: string;
/** Bot username for @bot command targeting (from getMe / botInfo). */
botUsername?: string;
/** Test seam / preloaded pairing-store ids; defaults to live pairing store. */
};
/**
* Whether the raw update's sender is command-authorized.
* Reuses resolveTelegramGroupAllowFromContext — same group/topic allowFrom
* overrides and access-group expansion as normal message ingress.
*/
export async function isTelegramSpooledUpdateSenderAuthorized(
update: unknown,
auth: TelegramSupersedeAuthContext,
): Promise<boolean> {
const facts = extractUpdateSenderFacts(update);
if (!facts) {
return false;
}
const accountCfg = mergeTelegramAccountConfig(auth.cfg, auth.accountId);
const dmPolicy = accountCfg.dmPolicy ?? "pairing";
const allowFrom = accountCfg.allowFrom;
const groupAllowFrom = accountCfg.groupAllowFrom ?? accountCfg.allowFrom;
const isForum =
resolveTelegramMessageForumFlagHint({
chatType: facts.chatType as "private" | "group" | "supergroup" | "channel" | undefined,
isForum: facts.isForum,
isTopicMessage: facts.isTopicMessage,
}) ?? false;
const groupAllowContext = await resolveTelegramGroupAllowFromContext({
cfg: auth.cfg,
chatId: facts.chatId,
accountId: auth.accountId,
dmPolicy,
allowFrom,
senderId: facts.senderId,
isGroup: facts.isGroup,
isForum,
messageThreadId: facts.messageThreadId,
groupAllowFrom,
resolveTelegramGroupConfig: (chatId, messageThreadId, cfg) => {
const telegramCfg = mergeTelegramAccountConfig(cfg, auth.accountId);
return resolveTelegramScopedGroupConfig(telegramCfg, chatId, messageThreadId);
},
});
const { resolvedThreadId, storeAllowFrom, groupAllowOverride, effectiveGroupAllow } =
groupAllowContext;
const dmAllow = await resolveTelegramDmAllow({
cfg: auth.cfg,
groupAllowOverride,
allowFrom,
accountId: auth.accountId,
senderId: facts.senderId,
storeAllowFrom: facts.isGroup ? [] : storeAllowFrom,
dmPolicy,
});
const ownerAccess = resolveTelegramCommandAuthorization({
cfg: auth.cfg,
accountId: auth.accountId,
chatId: facts.chatId,
isGroup: facts.isGroup,
...(resolvedThreadId !== undefined ? { resolvedThreadId } : {}),
senderId: facts.senderId,
...(facts.senderUsername !== undefined ? { senderUsername: facts.senderUsername } : {}),
});
const gate = await resolveTelegramCommandIngressAuthorization({
accountId: auth.accountId,
cfg: auth.cfg,
dmPolicy,
isGroup: facts.isGroup,
chatId: facts.chatId,
...(resolvedThreadId !== undefined ? { resolvedThreadId } : {}),
senderId: facts.senderId,
effectiveDmAllow: dmAllow.effectiveAllow,
effectiveGroupAllow,
ownerAccess,
eventKind: "message",
allowTextCommands: true,
hasControlCommand: true,
modeWhenAccessGroupsOff: "allow",
includeDmAllowForGroupCommands: false,
});
return gate.authorized;
}
/**
* Drain-level supersede predicate over raw spooled payloads.
* Authorization is resolved from the new event's numeric sender via the same
* ingress command gate as the old fence (CommandAuthorized).
*/
export function createShouldSupersedeTelegramSpooledPending(
auth: TelegramSupersedeAuthContext,
): (
newEvent: ChannelIngressQueueRecord<TelegramSpooledUpdatePayload>,
pendingEvent: ChannelIngressQueueClaim<TelegramSpooledUpdatePayload>,
) => boolean | Promise<boolean> {
return async (newEvent, pendingEvent) => {
const pendingUpdate = pendingEvent.payload.update;
const newUpdate = newEvent.payload.update;
// Ambient pending supersede still requires an authorized sender — same as the
// old fence (post-auth). Unauthorized strangers cannot cancel pre-adoption work.
if (
isTelegramAmbientSpooledUpdate(pendingUpdate) &&
!isTelegramAmbientSpooledUpdate(newUpdate)
) {
return await isTelegramSpooledUpdateSenderAuthorized(newUpdate, auth);
}
const text = extractUpdateText(newUpdate);
if (!text) {
return false;
}
const commandOptions = auth.botUsername ? { botUsername: auth.botUsername } : undefined;
if (
isBtwRequestText(text, commandOptions) ||
isTelegramReadOnlyControlLaneText({
rawText: text,
...(auth.botUsername ? { botUsername: auth.botUsername } : {}),
})
) {
return false;
}
// Abort, static text alias, or native bot_command entity (incl. skill commands)
// addressed to this bot. Never bare `/` prefixes without a bot_command entity.
const isAbort = isAbortRequestText(text, commandOptions);
const isCommand =
isRecognizedTelegramTextCommand(text, auth.botUsername) ||
updateHasBotCommandEntityForBot(newUpdate, auth.botUsername);
if (!isAbort && !isCommand) {
return false;
}
return await isTelegramSpooledUpdateSenderAuthorized(newUpdate, auth);
};
}

255
extensions/telegram/src/telegram-reply-fence.ts
View File

@@ -1,255 +0,0 @@
// Telegram plugin module implements telegram reply fence behavior.
import {
isExplicitCommandTurn,
type CommandTurnContext,
} from "openclaw/plugin-sdk/channel-inbound";
import {
maybeResolveTextAlias,
normalizeCommandBody,
} from "openclaw/plugin-sdk/command-auth-native";
import {
isAbortRequestText,
isBtwRequestText,
} from "openclaw/plugin-sdk/command-primitives-runtime";
import { isTelegramReadOnlyControlLaneText } from "./sequential-key.js";
type TelegramReplyFenceState = {
generation: number;
activeDispatches: number;
abortControllers?: Set<AbortController>;
laneKeys?: Set<string>;
};
type TelegramReplyFenceKey = {
activeKey: string;
roomEventKey: string;
};
// Newer accepted turns and authorized aborts can arrive ahead of older same-session reply work.
const TELEGRAM_REPLY_FENCE_STATE_KEY = Symbol.for("openclaw.telegram.replyFenceState");
function getTelegramReplyFenceState(): {
byKey: Map<string, TelegramReplyFenceState>;
keysByLane: Map<string, Set<string>>;
} {
const globalRecord = globalThis as Record<PropertyKey, unknown>;
const existing = globalRecord[TELEGRAM_REPLY_FENCE_STATE_KEY] as
| {
byKey: Map<string, TelegramReplyFenceState>;
keysByLane: Map<string, Set<string>>;
}
| undefined;
if (existing) {
return existing;
}
const created = {
byKey: new Map<string, TelegramReplyFenceState>(),
keysByLane: new Map<string, Set<string>>(),
};
globalRecord[TELEGRAM_REPLY_FENCE_STATE_KEY] = created;
return created;
}
const { byKey: telegramReplyFenceByKey, keysByLane: telegramReplyFenceKeysByLane } =
getTelegramReplyFenceState();
export function buildTelegramReplyFenceLaneKey(params: {
accountId: string;
sequentialKey: string;
}): string {
return `${params.accountId}\0${params.sequentialKey}`;
}
export function buildTelegramNonInterruptingReplyFenceKey(params: {
activeKey: string;
laneKey: string;
}): string {
return `${buildTelegramNonInterruptingReplyFenceKeyPrefix(params.activeKey)}${params.laneKey}`;
}
function buildTelegramNonInterruptingReplyFenceKeyPrefix(activeKey: string): string {
return `${activeKey}\0non-interrupting\0`;
}
function normalizeTelegramFenceKey(value: unknown): string | undefined {
if (typeof value !== "string") {
return undefined;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
}
export function resolveTelegramReplyFenceKey(params: {
ctxPayload: { SessionKey?: string; CommandTargetSessionKey?: string; InboundEventKind?: string };
chatId: number | string;
threadSpec: { id?: number | string | null; scope?: string };
}): TelegramReplyFenceKey {
const baseKey =
normalizeTelegramFenceKey(params.ctxPayload.CommandTargetSessionKey) ??
normalizeTelegramFenceKey(params.ctxPayload.SessionKey) ??
`telegram:${String(params.chatId)}:${params.threadSpec.scope ?? "default"}:${params.threadSpec.id ?? "root"}`;
const roomEventKey = `${baseKey}:room_event`;
return {
activeKey: params.ctxPayload.InboundEventKind === "room_event" ? roomEventKey : baseKey,
roomEventKey,
};
}
function abortTelegramReplyFenceControllers(state: TelegramReplyFenceState): void {
for (const controller of state.abortControllers ?? []) {
controller.abort();
}
state.abortControllers?.clear();
}
function deleteTelegramReplyFenceState(key: string, state: TelegramReplyFenceState): void {
telegramReplyFenceByKey.delete(key);
for (const laneKey of state.laneKeys ?? []) {
const keys = telegramReplyFenceKeysByLane.get(laneKey);
keys?.delete(key);
if (keys?.size === 0) {
telegramReplyFenceKeysByLane.delete(laneKey);
}
}
}
function maybeDeleteTelegramReplyFenceState(key: string, state: TelegramReplyFenceState): void {
if (state.activeDispatches <= 0 && (state.abortControllers?.size ?? 0) === 0) {
deleteTelegramReplyFenceState(key, state);
} else {
telegramReplyFenceByKey.set(key, state);
}
}
export function beginTelegramReplyFence(params: {
key: string;
supersede: boolean;
abortController?: AbortController;
laneKey?: string;
}): number {
const existing = telegramReplyFenceByKey.get(params.key);
const state: TelegramReplyFenceState = existing ?? {
generation: 0,
activeDispatches: 0,
};
if (params.supersede) {
state.generation += 1;
abortTelegramReplyFenceControllers(state);
supersedeTelegramNonInterruptingReplyFenceChildren(params.key);
}
if (params.abortController) {
(state.abortControllers ??= new Set()).add(params.abortController);
}
const laneKey = normalizeTelegramFenceKey(params.laneKey);
if (laneKey) {
(state.laneKeys ??= new Set()).add(laneKey);
const keys = telegramReplyFenceKeysByLane.get(laneKey) ?? new Set<string>();
keys.add(params.key);
telegramReplyFenceKeysByLane.set(laneKey, keys);
}
state.activeDispatches += 1;
telegramReplyFenceByKey.set(params.key, state);
return state.generation;
}
function supersedeTelegramReplyFenceState(key: string): boolean {
const state = telegramReplyFenceByKey.get(key);
if (!state) {
return false;
}
state.generation += 1;
abortTelegramReplyFenceControllers(state);
maybeDeleteTelegramReplyFenceState(key, state);
return true;
}
function supersedeTelegramNonInterruptingReplyFenceChildren(key: string): boolean {
let superseded = false;
const childPrefix = buildTelegramNonInterruptingReplyFenceKeyPrefix(key);
for (const childKey of telegramReplyFenceByKey.keys()) {
if (childKey.startsWith(childPrefix)) {
superseded = supersedeTelegramReplyFenceState(childKey) || superseded;
}
}
return superseded;
}
export function supersedeTelegramReplyFence(key: string): boolean {
let superseded = supersedeTelegramReplyFenceState(key);
superseded = supersedeTelegramNonInterruptingReplyFenceChildren(key) || superseded;
return superseded;
}
export function supersedeTelegramReplyFenceLane(laneKey: string): boolean {
const keys = [...(telegramReplyFenceKeysByLane.get(laneKey) ?? [])];
let superseded = false;
for (const key of keys) {
superseded = supersedeTelegramReplyFence(key) || superseded;
}
return superseded;
}
export function isTelegramReplyFenceSuperseded(params: {
key: string;
generation: number;
}): boolean {
return (telegramReplyFenceByKey.get(params.key)?.generation ?? 0) !== params.generation;
}
export function endTelegramReplyFence(key: string, abortController?: AbortController): void {
const state = telegramReplyFenceByKey.get(key);
if (!state) {
return;
}
if (abortController) {
state.abortControllers?.delete(abortController);
}
state.activeDispatches = Math.max(0, state.activeDispatches - 1);
maybeDeleteTelegramReplyFenceState(key, state);
}
export function releaseTelegramReplyFenceAbortController(
key: string,
abortController?: AbortController,
): void {
if (!abortController) {
return;
}
const state = telegramReplyFenceByKey.get(key);
if (!state) {
return;
}
state.abortControllers?.delete(abortController);
maybeDeleteTelegramReplyFenceState(key, state);
}
function isRecognizedTelegramTextCommand(rawText: string): boolean {
return maybeResolveTextAlias(normalizeCommandBody(rawText)) != null;
}
export function shouldSupersedeTelegramReplyFence(ctxPayload: {
Body?: string;
ChatType?: string;
RawBody?: string;
CommandBody?: string;
CommandAuthorized: boolean;
CommandTurn?: CommandTurnContext;
}): boolean {
const dispatchText = ctxPayload.CommandBody ?? ctxPayload.RawBody ?? ctxPayload.Body ?? "";
if (isAbortRequestText(dispatchText)) {
return ctxPayload.CommandAuthorized;
}
if (
isBtwRequestText(dispatchText) ||
isTelegramReadOnlyControlLaneText({ rawText: dispatchText })
) {
return false;
}
// One rule for all chat types: only authorized explicit/native commands
// supersede. Normal messages never abort an active turn at the transport
// fence; core queue policy owns steer/followup/interrupt after adoption.
return (
ctxPayload.CommandAuthorized &&
(isExplicitCommandTurn(ctxPayload.CommandTurn) || isRecognizedTelegramTextCommand(dispatchText))
);
}

500
extensions/telegram/src/webhook.ts
View File

@@ -33,43 +33,15 @@ import { readJsonBodyWithLimit } from "openclaw/plugin-sdk/webhook-request-guard
import { mergeTelegramAccountConfig } from "./account-config.js";
import { resolveTelegramAllowedUpdates } from "./allowed-updates.js";
import { withTelegramApiErrorLogging } from "./api-logging.js";
import {
runWithTelegramSpooledReplayUpdate,
type TelegramMessageProcessingResult,
type TelegramSpooledReplayDeferredParticipant,
} from "./bot-processing-outcome.js";
import { createTelegramBot } from "./bot.js";
import { resolveTelegramTransport } from "./fetch.js";
import { isRetryableTelegramApiError } from "./network-errors.js";
import { getTelegramSequentialKey } from "./sequential-key.js";
import { createTelegramTransportIngressDrain } from "./telegram-ingress-drain-factory.js";
import {
resolveNonRetryableSpooledUpdateFailure,
resolveSpooledUpdateAttemptNumber,
resolveSpooledUpdateRetryDelayMs,
shouldDeadLetterRetryableSpooledUpdate,
TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS,
} from "./spooled-update-retry-policy.js";
import {
isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess,
isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess,
} from "./telegram-ingress-claim-owner.js";
import {
claimNextTelegramSpooledUpdate,
completeTelegramSpooledUpdateWithRetry,
failTelegramSpooledUpdateClaim,
listTelegramSpooledUpdateClaims,
listTelegramSpooledUpdates,
recoverStaleTelegramSpooledUpdateClaims,
refreshTelegramSpooledUpdateClaim,
releaseTelegramSpooledUpdateClaim,
resolveTelegramIngressSpoolDir,
writeTelegramSpooledUpdate,
type ClaimedTelegramSpooledUpdate,
} from "./telegram-ingress-spool.js";
import {
buildTelegramReplyFenceLaneKey,
supersedeTelegramReplyFenceLane,
} from "./telegram-reply-fence.js";
import { createTelegramWebhookStatusPublisher } from "./webhook-status.js";
const TELEGRAM_WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
@@ -77,38 +49,12 @@ const TELEGRAM_WEBHOOK_BODY_TIMEOUT_MS = 30_000;
const TELEGRAM_WEBHOOK_ACCEPTED_HEADER = "x-openclaw-delivery-accepted";
const TELEGRAM_WEBHOOK_ACCEPTED_VALUE = "durable";
const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_INTERVAL_MS = 500;
const TELEGRAM_WEBHOOK_SPOOLED_CLAIM_REFRESH_INTERVAL_MS = 5 * 60 * 1000;
const TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS = 25 * 60_000;
const TELEGRAM_WEBHOOK_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000;
const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT = 100;
const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT * 10;
const TELEGRAM_WEBHOOK_REGISTRATION_RETRY_POLICY: BackoffPolicy = {
initialMs: 5_000,
maxMs: 60_000,
factor: 2,
jitter: 0.2,
};
type ActiveWebhookSpooledHandler = {
laneKey: string;
};
const activeWebhookSpooledHandlersByLane = new Map<string, ActiveWebhookSpooledHandler>();
function buildWebhookSpooledHandlerKey(params: { laneKey: string; spoolDir: string }): string {
return `${params.spoolDir}\0${params.laneKey}`;
}
function resolveActiveWebhookSpooledLaneKeys(spoolDir: string): Set<string> {
const laneKeys = new Set<string>();
const prefix = `${spoolDir}\0`;
for (const [handlerKey, handler] of activeWebhookSpooledHandlersByLane) {
if (handlerKey.startsWith(prefix)) {
laneKeys.add(handler.laneKey);
}
}
return laneKeys;
}
async function listenHttpServer(params: {
server: ReturnType<typeof createServer>;
port: number;
@@ -336,335 +282,6 @@ function resolveWebhookSpooledUpdateLaneKey(update: unknown): string {
});
}
async function releaseFailedWebhookSpooledUpdate(params: {
err: unknown;
log: (line: string) => void;
update: ClaimedTelegramSpooledUpdate;
}): Promise<void> {
const laneKey = resolveWebhookSpooledUpdateLaneKey(params.update.update);
const nonRetryable = resolveNonRetryableSpooledUpdateFailure(params.err);
if (nonRetryable) {
const failed = await failTelegramSpooledUpdateClaim({
update: params.update,
reason: nonRetryable.reason,
message: nonRetryable.message,
});
if (failed) {
params.log(
`[telegram][diag] webhook spooled update ${params.update.updateId} failed with non-retryable ${nonRetryable.reason}; dead-lettered: ${nonRetryable.message}`,
);
}
return;
}
const attempt = resolveSpooledUpdateAttemptNumber(params.update);
if (shouldDeadLetterRetryableSpooledUpdate(params.update, attempt)) {
const message = formatErrorMessage(params.err);
const failed = await failTelegramSpooledUpdateClaim({
update: params.update,
reason: "retry-limit-exceeded",
message,
});
if (failed) {
// Retryable poison updates must eventually become tombstones, but not
// during ordinary transient provider or state-store outages.
params.log(
`[telegram][warn] webhook spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit after ${attempt} attempts; dead-lettered: ${message}`,
);
}
return;
}
await releaseTelegramSpooledUpdateClaim(params.update, {
lastError: formatErrorMessage(params.err),
});
params.log(
`[telegram][diag] webhook spooled update ${params.update.updateId} failed; keeping for retry attempt ${attempt + 1}/${TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS}: ${formatErrorMessage(params.err)}`,
);
}
function startWebhookSpooledUpdateClaimRefresh(params: {
log: (line: string) => void;
update: ClaimedTelegramSpooledUpdate;
}): () => void {
let stopped = false;
let refreshing = false;
const refresh = async (): Promise<void> => {
if (stopped || refreshing) {
return;
}
refreshing = true;
try {
const refreshed = await refreshTelegramSpooledUpdateClaim(params.update);
if (!refreshed && !stopped) {
params.log(
`[telegram][diag] webhook spooled update ${params.update.updateId} claim refresh lost ownership`,
);
}
} catch (err) {
params.log(
`[telegram][diag] webhook spooled update ${params.update.updateId} claim refresh failed: ${formatErrorMessage(err)}`,
);
} finally {
refreshing = false;
}
};
const timer = setInterval(() => {
void refresh();
}, TELEGRAM_WEBHOOK_SPOOLED_CLAIM_REFRESH_INTERVAL_MS);
timer.unref?.();
return () => {
if (stopped) {
return;
}
stopped = true;
clearInterval(timer);
};
}
type WebhookSpooledDeferredWorkResult = TelegramMessageProcessingResult & {
timedOut?: boolean;
};
class WebhookSpooledHandlerTimeoutError extends Error {
constructor(
message: string,
readonly replayTask: Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }>,
) {
super(message);
this.name = "WebhookSpooledHandlerTimeoutError";
}
}
function formatWebhookSpooledHandlerTimeoutMessage(params: {
laneKey: string;
updateId: number;
}): string {
const age = formatDurationPrecise(TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS);
return `Telegram webhook spool processing timed out behind update ${params.updateId} on lane ${params.laneKey} after ${age}; marking the update failed.`;
}
async function failTimedOutWebhookSpooledUpdate(params: {
log: (line: string) => void;
message: string;
update: ClaimedTelegramSpooledUpdate;
}): Promise<void> {
const failed = await failTelegramSpooledUpdateClaim({
update: params.update,
reason: "handler-timeout",
message: params.message,
});
if (!failed) {
params.log(
`[telegram][diag] timed out webhook spooled update ${params.update.updateId} no longer had a processing marker to fail.`,
);
}
}
async function waitForTimedOutWebhookReplayGrace(params: {
log: (line: string) => void;
replayTask: Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }>;
updateId: number;
}): Promise<boolean> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
params.replayTask.then(
() => true,
(replayErr: unknown) => {
params.log(
`[telegram][diag] timed out webhook spooled update ${params.updateId} replay later failed: ${formatErrorMessage(replayErr)}`,
);
return true;
},
),
new Promise<boolean>((resolve) => {
timer = setTimeout(() => resolve(false), TELEGRAM_WEBHOOK_SPOOLED_HANDLER_ABORT_GRACE_MS);
timer.unref?.();
}),
]);
} finally {
if (timer) {
clearTimeout(timer);
}
}
}
type WebhookSpooledUpdateHandlerResult = {
retainLaneGuardTask?: Promise<unknown>;
};
async function runWebhookSpooledReplayWithTimeout(params: {
bot: ReturnType<typeof createTelegramBot>;
laneKey: string;
rawUpdate: object;
update: Parameters<ReturnType<typeof createTelegramBot>["handleUpdate"]>[0];
updateId: number;
}): Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }> {
let timer: ReturnType<typeof setTimeout> | undefined;
const replayTask = runWithTelegramSpooledReplayUpdate(params.rawUpdate, async () => {
await params.bot.handleUpdate(params.update);
});
replayTask.catch(() => undefined);
const timeout = new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => {
reject(
new WebhookSpooledHandlerTimeoutError(
formatWebhookSpooledHandlerTimeoutMessage({
laneKey: params.laneKey,
updateId: params.updateId,
}),
replayTask,
),
);
}, TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS);
timer.unref?.();
});
try {
return await Promise.race([replayTask, timeout]);
} finally {
if (timer) {
clearTimeout(timer);
}
}
}
async function waitForWebhookSpooledDeferredWork(params: {
deferredWork: TelegramSpooledReplayDeferredParticipant;
laneKey: string;
log: (line: string) => void;
update: ClaimedTelegramSpooledUpdate;
}): Promise<WebhookSpooledDeferredWorkResult> {
let timeoutError: Error | undefined;
const timer = setTimeout(() => {
const age = formatDurationPrecise(TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS);
const message = `Telegram webhook spool buffered processing timed out behind update ${params.update.updateId} on lane ${params.laneKey} after ${age}; marking the update failed.`;
params.log(`[telegram] ${message}`);
timeoutError = new Error(message);
params.deferredWork.settle({
kind: "failed-retryable",
error: timeoutError,
});
}, TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS);
timer.unref?.();
try {
const result = await params.deferredWork.task.catch(
(err: unknown): TelegramMessageProcessingResult => ({
kind: "failed-retryable",
error: err,
}),
);
// A durable-adoption hold can discard the timeout and settle completed.
// Only the exact timeout result owns the handler-timeout failure path.
return timeoutError !== undefined &&
result.kind === "failed-retryable" &&
result.error === timeoutError
? { ...result, timedOut: true }
: result;
} finally {
clearTimeout(timer);
}
}
async function handleWebhookSpooledUpdate(params: {
accountId: string;
abortSignal?: AbortSignal;
bot: ReturnType<typeof createTelegramBot>;
log: (line: string) => void;
update: ClaimedTelegramSpooledUpdate;
}): Promise<WebhookSpooledUpdateHandlerResult> {
let replay: { deferredWork?: TelegramSpooledReplayDeferredParticipant };
try {
const rawUpdate = params.update.update;
if (!rawUpdate || typeof rawUpdate !== "object") {
throw new Error("Telegram spooled webhook update payload was invalid.");
}
const laneKey = resolveWebhookSpooledUpdateLaneKey(rawUpdate);
const update = rawUpdate as Parameters<typeof params.bot.handleUpdate>[0];
replay = await runWebhookSpooledReplayWithTimeout({
bot: params.bot,
laneKey,
rawUpdate,
update,
updateId: params.update.updateId,
});
} catch (err) {
if (err instanceof WebhookSpooledHandlerTimeoutError) {
params.log(`[telegram] ${err.message}`);
const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({
accountId: params.accountId,
sequentialKey: resolveWebhookSpooledUpdateLaneKey(params.update.update),
});
const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey);
if (!abortedReplyWork) {
params.log(
`[telegram][diag] timed out webhook spooled update ${params.update.updateId} had no active reply fence on lane ${scopedReplyFenceLaneKey}.`,
);
}
await failTimedOutWebhookSpooledUpdate({
log: params.log,
message: err.message,
update: params.update,
});
const replaySettled = await waitForTimedOutWebhookReplayGrace({
log: params.log,
replayTask: err.replayTask,
updateId: params.update.updateId,
});
if (replaySettled) {
return {};
}
return {
retainLaneGuardTask: err.replayTask.catch((replayErr: unknown) => {
params.log(
`[telegram][diag] timed out webhook spooled update ${params.update.updateId} replay later failed: ${formatErrorMessage(replayErr)}`,
);
}),
};
}
await releaseFailedWebhookSpooledUpdate({
err,
log: params.log,
update: params.update,
});
return {};
}
if (replay.deferredWork) {
const result = await waitForWebhookSpooledDeferredWork({
deferredWork: replay.deferredWork,
laneKey: resolveWebhookSpooledUpdateLaneKey(params.update.update),
log: params.log,
update: params.update,
});
if (result.kind === "failed-retryable") {
if (result.timedOut) {
await failTimedOutWebhookSpooledUpdate({
log: params.log,
message: formatErrorMessage(result.error),
update: params.update,
});
return {};
}
await releaseFailedWebhookSpooledUpdate({
err: result.error,
log: params.log,
update: params.update,
});
return {};
}
}
await completeTelegramSpooledUpdateWithRetry({
update: params.update,
abortSignal: params.abortSignal,
onRetry: ({ attempt, delayMs, error }) => {
params.log(
`[telegram][diag] webhook spooled update ${params.update.updateId} completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`,
);
},
});
return {};
}
export async function startTelegramWebhook(opts: {
token: string;
accountId?: string;
@@ -703,9 +320,6 @@ export async function startTelegramWebhook(opts: {
const spoolDir = opts.spoolDir ?? resolveTelegramIngressSpoolDir({ accountId: opts.accountId });
let shutDown = false;
const shutdownAbortController = new AbortController();
const webhookAbortSignal = opts.abortSignal
? AbortSignal.any([shutdownAbortController.signal, opts.abortSignal])
: shutdownAbortController.signal;
const telegramAccountConfig = opts.config
? mergeTelegramAccountConfig(opts.config, opts.accountId ?? "default")
: undefined;
@@ -755,6 +369,7 @@ export async function startTelegramWebhook(opts: {
const log = (line: string) => runtime.log?.(line);
let drainActive = false;
let drainRequested = false;
let webhookIngressDrain: ReturnType<typeof createTelegramTransportIngressDrain> | undefined;
const drainWebhookSpool = async (): Promise<void> => {
if (shutDown || opts.abortSignal?.aborted) {
return;
@@ -766,105 +381,30 @@ export async function startTelegramWebhook(opts: {
drainActive = true;
drainRequested = false;
try {
const activeWebhookSpooledLaneKeys = resolveActiveWebhookSpooledLaneKeys(spoolDir);
await recoverStaleTelegramSpooledUpdateClaims({
// Shutdown must abort in-flight drain work (tombstone retries), not just
// stop the next claim; the composed signal carries webhook stop + caller abort.
const webhookAbortSignal = opts.abortSignal
? AbortSignal.any([shutdownAbortController.signal, opts.abortSignal])
: shutdownAbortController.signal;
webhookIngressDrain ??= createTelegramTransportIngressDrain({
spoolDir,
staleMs: 0,
shouldRecover: (claim) =>
!activeWebhookSpooledLaneKeys.has(resolveWebhookSpooledUpdateLaneKey(claim.update)) &&
!isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(claim),
shouldRecoverCorrupt: (claim) =>
!(claim.laneKey && activeWebhookSpooledLaneKeys.has(claim.laneKey)) &&
!isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess(claim),
bot,
cfg: opts.config ?? {},
accountId: opts.accountId ?? "default",
// Pre-migration product default: 25m claim→adoption stall for webhook.
adoptionStallTimeoutMs: 25 * 60_000,
abortSignal: webhookAbortSignal,
onLog: (message) => log(`webhook ${message}`),
});
const claimedLaneKeys = new Set(
(
await listTelegramSpooledUpdateClaims({
spoolDir,
})
).map((claim) => resolveWebhookSpooledUpdateLaneKey(claim.update)),
);
const updates = await listTelegramSpooledUpdates({
spoolDir,
limit: TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT,
await webhookIngressDrain.drainOnce({
shouldStop: () => shutDown || webhookAbortSignal.aborted,
});
const candidateUpdateIds = updates.map((update) => update.updateId);
const blockedLaneKeys = new Set([...activeWebhookSpooledLaneKeys, ...claimedLaneKeys]);
for (const update of updates) {
// Release stamps lastAttemptAt; block the lane until backoff expires so
// webhook replay cannot hot-loop a retryable poison update.
if (resolveSpooledUpdateRetryDelayMs(update) > 0) {
blockedLaneKeys.add(resolveWebhookSpooledUpdateLaneKey(update.update));
}
}
let started = 0;
while (started < TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT) {
if (shutDown || opts.abortSignal?.aborted) {
break;
}
const claimedUpdate = await claimNextTelegramSpooledUpdate({
spoolDir,
blockedLaneKeys,
candidateUpdateIds,
scanLimit: TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT,
});
if (!claimedUpdate) {
break;
}
const laneKey = resolveWebhookSpooledUpdateLaneKey(claimedUpdate.update);
const handlerKey = buildWebhookSpooledHandlerKey({ spoolDir, laneKey });
// Webhook HTTP requests and same-process restarts can overlap; keep
// one process-global active claim per spool lane to preserve ordering.
const handlerState: ActiveWebhookSpooledHandler = { laneKey };
activeWebhookSpooledHandlersByLane.set(handlerKey, handlerState);
blockedLaneKeys.add(laneKey);
// Claim ownership has a finite lease; refresh while the handler runs so
// another process cannot recover and replay this update concurrently.
const stopClaimRefresh = startWebhookSpooledUpdateClaimRefresh({
log,
update: claimedUpdate,
});
let retainLaneGuardTask: Promise<unknown> | undefined;
void handleWebhookSpooledUpdate({
accountId: opts.accountId ?? "default",
abortSignal: webhookAbortSignal,
bot,
log,
update: claimedUpdate,
})
.then((result) => {
retainLaneGuardTask = result.retainLaneGuardTask;
if (retainLaneGuardTask) {
void retainLaneGuardTask.finally(() => {
if (activeWebhookSpooledHandlersByLane.get(handlerKey) === handlerState) {
activeWebhookSpooledHandlersByLane.delete(handlerKey);
}
void Promise.resolve().then(drainWebhookSpool);
});
}
})
.catch((err: unknown) => {
runtime.log?.(
`[telegram][diag] webhook spooled update ${claimedUpdate.updateId} handler failed after claim: ${formatErrorMessage(err)}`,
);
})
.finally(() => {
stopClaimRefresh();
if (
!retainLaneGuardTask &&
activeWebhookSpooledHandlersByLane.get(handlerKey) === handlerState
) {
activeWebhookSpooledHandlersByLane.delete(handlerKey);
}
void Promise.resolve().then(drainWebhookSpool);
});
started += 1;
}
} catch (err) {
runtime.log?.(`[telegram][diag] webhook spool drain failed: ${formatErrorMessage(err)}`);
log(`[telegram][diag] webhook spool drain failed: ${formatErrorMessage(err)}`);
} finally {
drainActive = false;
if (drainRequested && !shutDown && !opts.abortSignal?.aborted) {
drainRequested = false;
void Promise.resolve().then(drainWebhookSpool);
}
}
@@ -1008,6 +548,8 @@ export async function startTelegramWebhook(opts: {
if (drainTimer) {
clearInterval(drainTimer);
}
webhookIngressDrain?.dispose();
webhookIngressDrain = undefined;
server.close();
await bot.stop();
// The webhook owns this transport because it resolved and injected it into