diff --git a/extensions/telegram/src/bot-message-dispatch.test-harness.ts b/extensions/telegram/src/bot-message-dispatch.test-harness.ts
index 8a9981538b71..b4e23c5c4b73 100644
--- a/extensions/telegram/src/bot-message-dispatch.test-harness.ts
+++ b/extensions/telegram/src/bot-message-dispatch.test-harness.ts
@@ -558,17 +558,6 @@ export function createDirectSessionPayload(): TelegramMessageContext["ctxPayload
   } as TelegramMessageContext["ctxPayload"];
 }
 
-export function observeDeliveredReply(text: string): Promise<void> {
-  return new Promise((resolve) => {
-    deliverReplies.mockImplementation(async (params: { replies?: Array<{ text?: string }> }) => {
-      if (params.replies?.some((reply) => reply.text === text)) {
-        resolve();
-      }
-      return { delivered: true };
-    });
-  });
-}
-
 export function createBot(): Bot {
   return {
     api: {
diff --git a/extensions/telegram/src/bot-processing-outcome.ts b/extensions/telegram/src/bot-processing-outcome.ts
index b3b24729871e..310757e472ee 100644
--- a/extensions/telegram/src/bot-processing-outcome.ts
+++ b/extensions/telegram/src/bot-processing-outcome.ts
@@ -10,7 +10,7 @@ type TelegramUpdateProcessingFrame = {
   result?: TelegramMessageProcessingResult;
 };
 
-export type TelegramSpooledReplayLifecycle = {
+type TelegramSpooledReplayLifecycle = {
   abortSignal: AbortSignal;
   onAdopted: () => void | Promise<void>;
   onDeferred: () => void;
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
index e40378c85248..60a3ed2fac8a 100644
--- a/extensions/telegram/src/polling-session.test.ts
+++ b/extensions/telegram/src/polling-session.test.ts
@@ -42,7 +42,7 @@ const shouldDeadLetterRetryableSpooledUpdate = (
   attempt: number,
   now?: number,
 ) => shouldDeadLetterRetryableIngressEvent(update, attempt, undefined, now);
-import type { TelegramSpooledUpdate } from "./telegram-ingress-spool.types.js";
+import type { TelegramSpooledUpdate } from "./telegram-ingress-spool.test-support.js";
 import type { TelegramIngressWorkerMessage } from "./telegram-ingress-worker.js";
 
 const runMock = vi.hoisted(() => vi.fn());
@@ -110,10 +110,10 @@ const pollingSessionTesting = {
 };
 // Mirrors core INGRESS_CLAIM_LEASE_MS (ingress-claim-owner).
 const telegramSpooledUpdateClaimLeaseMs = 30 * 60 * 1000;
-let claimNextTelegramSpooledUpdate: typeof import("./telegram-ingress-spool.js").claimNextTelegramSpooledUpdate;
-let listTelegramSpooledUpdateClaims: typeof import("./telegram-ingress-spool.js").listTelegramSpooledUpdateClaims;
-let listTelegramSpooledUpdates: typeof import("./telegram-ingress-spool.js").listTelegramSpooledUpdates;
-let recoverStaleTelegramSpooledUpdateClaims: typeof import("./telegram-ingress-spool.js").recoverStaleTelegramSpooledUpdateClaims;
+let claimNextTelegramSpooledUpdate: typeof import("./telegram-ingress-spool.test-support.js").claimNextTelegramSpooledUpdate;
+let listTelegramSpooledUpdateClaims: typeof import("./telegram-ingress-spool.test-support.js").listTelegramSpooledUpdateClaims;
+let listTelegramSpooledUpdates: typeof import("./telegram-ingress-spool.test-support.js").listTelegramSpooledUpdates;
+let recoverStaleTelegramSpooledUpdateClaims: typeof import("./telegram-ingress-spool.test-support.js").recoverStaleTelegramSpooledUpdateClaims;
 let writeTelegramSpooledUpdate: typeof import("./telegram-ingress-spool.js").writeTelegramSpooledUpdate;
 let createTelegramSpooledReplayDeferredParticipant: typeof import("./bot-processing-outcome.js").createTelegramSpooledReplayDeferredParticipant;
 type TelegramMessageProcessingResult =
@@ -739,13 +739,13 @@ function startIsolatedIngressSession(params: {
 describe("TelegramPollingSession", () => {
   beforeAll(async () => {
     ({ TelegramPollingSession } = await import("./polling-session.js"));
+    ({ writeTelegramSpooledUpdate } = await import("./telegram-ingress-spool.js"));
     ({
       claimNextTelegramSpooledUpdate,
       listTelegramSpooledUpdateClaims,
       listTelegramSpooledUpdates,
       recoverStaleTelegramSpooledUpdateClaims,
-      writeTelegramSpooledUpdate,
-    } = await import("./telegram-ingress-spool.js"));
+    } = await import("./telegram-ingress-spool.test-support.js"));
     ({ createTelegramSpooledReplayDeferredParticipant } =
       await import("./bot-processing-outcome.js"));
   });
diff --git a/extensions/telegram/src/telegram-ingress-drain-factory.ts b/extensions/telegram/src/telegram-ingress-drain-factory.ts
index d95ed753cb14..16da16f7a21a 100644
--- a/extensions/telegram/src/telegram-ingress-drain-factory.ts
+++ b/extensions/telegram/src/telegram-ingress-drain-factory.ts
@@ -10,11 +10,11 @@ import {
 } from "./telegram-ingress-drain.js";
 import { openTelegramIngressQueue } from "./telegram-ingress-spool.js";
 
-export type TelegramSpooledBot = {
+type TelegramSpooledBot = {
   handleUpdate: (update: never) => Promise<void>;
 };
 
-export type CreateTelegramTransportIngressDrainParams = {
+type CreateTelegramTransportIngressDrainParams = {
   spoolDir: string;
   bot: TelegramSpooledBot;
   cfg: OpenClawConfig;
diff --git a/extensions/telegram/src/telegram-ingress-drain.ts b/extensions/telegram/src/telegram-ingress-drain.ts
index ad4fef94fae7..cb76b4a07885 100644
--- a/extensions/telegram/src/telegram-ingress-drain.ts
+++ b/extensions/telegram/src/telegram-ingress-drain.ts
@@ -20,7 +20,7 @@ import { resolveTelegramIngressNonRetryableFailure } from "./telegram-ingress-no
 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_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;
 
@@ -41,7 +41,7 @@ export function resolveTelegramAdoptionStallTimeoutMs(params: {
   return DEFAULT_INGRESS_ADOPTION_STALL_MS;
 }
 
-export function telegramSpooledLaneKey(update: unknown, botInfo?: TelegramBotInfo): string {
+function telegramSpooledLaneKey(update: unknown, botInfo?: TelegramBotInfo): string {
   return getTelegramSequentialKey({
     update: update as Parameters<typeof getTelegramSequentialKey>[0]["update"],
     ...(botInfo ? { me: botInfo } : {}),
@@ -56,12 +56,12 @@ export type TelegramIngressDrainLifecycle = {
   onAbandoned: () => void;
 };
 
-export type TelegramIngressDrainDispatch = (
+type TelegramIngressDrainDispatch = (
   update: unknown,
   lifecycle: TelegramIngressDrainLifecycle,
 ) => Promise<TelegramMessageProcessingResult | void> | TelegramMessageProcessingResult | void;
 
-export type CreateTelegramIngressDrainParams = {
+type CreateTelegramIngressDrainParams = {
   queue: ChannelIngressQueue<TelegramSpooledUpdatePayload>;
   /** Required for authorization-gated supersede (numeric allowlist). */
   cfg: OpenClawConfig;
diff --git a/extensions/telegram/src/telegram-ingress-non-retryable.ts b/extensions/telegram/src/telegram-ingress-non-retryable.ts
index 933e4eccac47..4ac89affc5f8 100644
--- a/extensions/telegram/src/telegram-ingress-non-retryable.ts
+++ b/extensions/telegram/src/telegram-ingress-non-retryable.ts
@@ -9,7 +9,7 @@ import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-d
 const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError";
 const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u;
 
-export type TelegramIngressNonRetryableFailure = {
+type TelegramIngressNonRetryableFailure = {
   reason: "missing-agent-harness" | "dispatch-dedupe-rollback-failed";
   message: string;
 };
diff --git a/extensions/telegram/src/telegram-ingress-spool.test-support.ts b/extensions/telegram/src/telegram-ingress-spool.test-support.ts
new file mode 100644
index 000000000000..d920a97f03e6
--- /dev/null
+++ b/extensions/telegram/src/telegram-ingress-spool.test-support.ts
@@ -0,0 +1,186 @@
+// Test-only queue inspection/claim wrappers over openTelegramIngressQueue.
+// Prod claiming lives in the core ingress drain; transport tests use these to
+// assert spool contents and drive claim states without a running drain.
+import path from "node:path";
+import {
+  INGRESS_CLAIM_PROCESS_ID,
+  processPidFromOwnerId,
+  type ChannelIngressQueueClaim,
+  type ChannelIngressQueueCorruptClaim,
+  type ChannelIngressQueueRecord,
+} from "openclaw/plugin-sdk/channel-outbound";
+import type { TelegramBotInfo } from "./bot-info.js";
+import {
+  openTelegramIngressQueue,
+  telegramSpooledUpdateLaneKey,
+} from "./telegram-ingress-spool.js";
+import {
+  TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION,
+  type TelegramSpooledUpdatePayload,
+} from "./telegram-ingress-spool.payload.js";
+
+type TelegramSpooledUpdateClaimOwner = {
+  processId: string;
+  processPid: number;
+  claimedAt: number;
+  claimToken?: string;
+};
+
+export type TelegramSpooledUpdate = {
+  updateId: number;
+  path: string;
+  update: unknown;
+  receivedAt: number;
+  attempts?: number;
+  lastAttemptAt?: number;
+  lastError?: string;
+  claim?: TelegramSpooledUpdateClaimOwner;
+};
+
+export type ClaimedTelegramSpooledUpdate = TelegramSpooledUpdate & {
+  pendingPath: string;
+};
+
+export function telegramQueueEventId(updateId: number): string {
+  return String(updateId).padStart(16, "0");
+}
+
+function spoolFileName(updateId: number): string {
+  return `${telegramQueueEventId(updateId)}.json`;
+}
+
+function processingFileName(updateId: number): string {
+  return `${spoolFileName(updateId)}.processing`;
+}
+
+function isValidUpdateId(value: unknown): value is number {
+  return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
+}
+
+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 }),
+  };
+}
+
+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,
+    },
+  };
+}
+
+export async function listTelegramSpooledUpdates(params: {
+  spoolDir: string;
+  limit?: number | "all";
+}): Promise<TelegramSpooledUpdate[]> {
+  const records = await openTelegramIngressQueue(params.spoolDir).listPending({
+    limit: params.limit ?? 100,
+    orderBy: "id",
+  });
+  return records
+    .flatMap((record) => {
+      const update = parsePendingRecord(params.spoolDir, record);
+      return update ? [update] : [];
+    })
+    .toSorted((a, b) => a.updateId - b.updateId);
+}
+
+export async function claimNextTelegramSpooledUpdate(params: {
+  spoolDir: string;
+  blockedLaneKeys?: Iterable<string>;
+  botInfo?: TelegramBotInfo;
+  candidateUpdateIds?: Iterable<number>;
+  scanLimit?: number;
+}): Promise<ClaimedTelegramSpooledUpdate | null> {
+  const queue = openTelegramIngressQueue(params.spoolDir);
+  const claimed = await queue.claimNext({
+    ownerId: INGRESS_CLAIM_PROCESS_ID,
+    blockedLaneKeys: params.blockedLaneKeys,
+    ...(params.candidateUpdateIds === undefined
+      ? {}
+      : { candidateIds: [...params.candidateUpdateIds].map(telegramQueueEventId) }),
+    orderBy: "id",
+    scanLimit: params.scanLimit,
+    deriveLaneKey: (record) => telegramSpooledUpdateLaneKey(record.payload.update, params.botInfo),
+  });
+  if (!claimed) {
+    return null;
+  }
+  const update = parseQueueClaim(params.spoolDir, claimed);
+  if (update) {
+    return update;
+  }
+  await queue.fail(claimed, {
+    reason: "invalid-spooled-update",
+    message: "Telegram spooled update payload was invalid.",
+  });
+  return null;
+}
+
+export async function listTelegramSpooledUpdateClaims(params: {
+  spoolDir: string;
+}): Promise<ClaimedTelegramSpooledUpdate[]> {
+  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: {
+  spoolDir: string;
+  staleMs?: number;
+  now?: number;
+  shouldRecover?: (claim: ClaimedTelegramSpooledUpdate) => boolean | Promise<boolean>;
+  shouldRecoverCorrupt?: (claim: ChannelIngressQueueCorruptClaim) => boolean | Promise<boolean>;
+}): Promise<number> {
+  const shouldRecover = params.shouldRecover;
+  const shouldRecoverCorrupt = params.shouldRecoverCorrupt;
+  return await openTelegramIngressQueue(params.spoolDir).recoverStaleClaims({
+    staleMs: params.staleMs ?? 0,
+    ...(params.now === undefined ? {} : { now: params.now }),
+    ...(shouldRecover
+      ? {
+          shouldRecover: async (claim) => {
+            const update = parseQueueClaim(params.spoolDir, claim);
+            return update ? await shouldRecover(update) : false;
+          },
+        }
+      : {}),
+    ...(shouldRecoverCorrupt ? { shouldRecoverCorrupt } : {}),
+  });
+}
diff --git a/extensions/telegram/src/telegram-ingress-spool.test.ts b/extensions/telegram/src/telegram-ingress-spool.test.ts
index 0b70bff32386..898ef761e86e 100644
--- a/extensions/telegram/src/telegram-ingress-spool.test.ts
+++ b/extensions/telegram/src/telegram-ingress-spool.test.ts
@@ -10,13 +10,15 @@ import { afterEach, describe, expect, it } from "vitest";
 import { setTelegramRuntime } from "./runtime.js";
 import { clearTelegramRuntimeForTest } from "./runtime.test-support.js";
 import {
-  listTelegramSpooledUpdates,
   openTelegramIngressQueue,
   resolveTelegramIngressSpoolDir,
-  telegramQueueEventId,
   telegramSpooledUpdateLaneKey,
   writeTelegramSpooledUpdate,
 } from "./telegram-ingress-spool.js";
+import {
+  listTelegramSpooledUpdates,
+  telegramQueueEventId,
+} from "./telegram-ingress-spool.test-support.js";
 
 async function withTempState<T>(
   fn: (stateDir: string, spoolDir: string) => Promise<T>,
diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts
index 0c03bf582e9f..9073f9ba5d27 100644
--- a/extensions/telegram/src/telegram-ingress-spool.ts
+++ b/extensions/telegram/src/telegram-ingress-spool.ts
@@ -1,15 +1,7 @@
 // Telegram plugin module implements durable ingress enqueue + update_id mapping.
 import os from "node:os";
 import path from "node:path";
-import {
-  INGRESS_CLAIM_PROCESS_ID,
-  processPidFromOwnerId,
-  type ChannelIngressQueue,
-  type ChannelIngressQueueClaim,
-  type ChannelIngressQueueClaimRef,
-  type ChannelIngressQueueCorruptClaim,
-  type ChannelIngressQueueRecord,
-} from "openclaw/plugin-sdk/channel-outbound";
+import type { ChannelIngressQueue } from "openclaw/plugin-sdk/channel-outbound";
 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";
@@ -20,17 +12,6 @@ import {
   TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION,
   type TelegramSpooledUpdatePayload,
 } from "./telegram-ingress-spool.payload.js";
-import type {
-  ClaimedTelegramSpooledUpdate,
-  TelegramSpooledUpdate,
-} from "./telegram-ingress-spool.types.js";
-
-export type {
-  ClaimedTelegramSpooledUpdate,
-  TelegramSpooledUpdate,
-} from "./telegram-ingress-spool.types.js";
-export type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js";
-
 const TELEGRAM_INGRESS_SPOOL_PREFIX = "ingress-spool-";
 const TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000;
 const TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES = 1000;
@@ -67,14 +48,10 @@ function resolveTelegramUpdateId(update: unknown): number | null {
   return isValidUpdateId(value) ? value : null;
 }
 
-export function telegramQueueEventId(updateId: number): string {
+function telegramQueueEventId(updateId: number): string {
   return String(updateId).padStart(16, "0");
 }
 
-function spoolFileName(updateId: number): string {
-  return `${telegramQueueEventId(updateId)}.json`;
-}
-
 function resolveQueueParts(spoolDir: string): {
   accountId: string;
   stateDir: string;
@@ -157,174 +134,7 @@ export async function writeTelegramSpooledUpdate(params: {
   return updateId;
 }
 
-export async function listTelegramSpooledUpdates(params: {
-  spoolDir: string;
-  limit?: number | "all";
-}): Promise<TelegramSpooledUpdate[]> {
-  const records = await openTelegramIngressQueue(params.spoolDir).listPending({
-    limit: params.limit ?? 100,
-    orderBy: "id",
-  });
-  return records
-    .flatMap((record) => {
-      const update = parsePendingRecord(params.spoolDir, record);
-      return update ? [update] : [];
-    })
-    .toSorted((a, b) => a.updateId - b.updateId);
-}
-
-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 }),
-  };
-}
-
 /** 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: {
-  spoolDir: string;
-  blockedLaneKeys?: Iterable<string>;
-  botInfo?: TelegramBotInfo;
-  candidateUpdateIds?: Iterable<number>;
-  scanLimit?: number;
-}): Promise<ClaimedTelegramSpooledUpdate | null> {
-  const queue = openTelegramIngressQueue(params.spoolDir);
-  const claimed = await queue.claimNext({
-    ownerId: INGRESS_CLAIM_PROCESS_ID,
-    blockedLaneKeys: params.blockedLaneKeys,
-    ...(params.candidateUpdateIds === undefined
-      ? {}
-      : { candidateIds: [...params.candidateUpdateIds].map(telegramQueueEventId) }),
-    orderBy: "id",
-    scanLimit: params.scanLimit,
-    deriveLaneKey: (record) => telegramSpooledUpdateLaneKey(record.payload.update, params.botInfo),
-  });
-  if (!claimed) {
-    return null;
-  }
-  const update = parseQueueClaim(params.spoolDir, claimed);
-  if (update) {
-    return update;
-  }
-  await queue.fail(claimed, {
-    reason: "invalid-spooled-update",
-    message: "Telegram spooled update payload was invalid.",
-  });
-  return null;
-}
-
-export async function listTelegramSpooledUpdateClaims(params: {
-  spoolDir: string;
-}): Promise<ClaimedTelegramSpooledUpdate[]> {
-  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: {
-  spoolDir: string;
-  staleMs?: number;
-  now?: number;
-  shouldRecover?: (claim: ClaimedTelegramSpooledUpdate) => boolean | Promise<boolean>;
-  shouldRecoverCorrupt?: (claim: ChannelIngressQueueCorruptClaim) => boolean | Promise<boolean>;
-}): Promise<number> {
-  const shouldRecover = params.shouldRecover;
-  const shouldRecoverCorrupt = params.shouldRecoverCorrupt;
-  return await openTelegramIngressQueue(params.spoolDir).recoverStaleClaims({
-    staleMs: params.staleMs ?? 0,
-    ...(params.now === undefined ? {} : { now: params.now }),
-    ...(shouldRecover
-      ? {
-          shouldRecover: async (claim) => {
-            const update = parseQueueClaim(params.spoolDir, claim);
-            return update ? await shouldRecover(update) : false;
-          },
-        }
-      : {}),
-    ...(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 }),
-    },
-  );
-}
diff --git a/extensions/telegram/src/telegram-ingress-spool.types.ts b/extensions/telegram/src/telegram-ingress-spool.types.ts
deleted file mode 100644
index 38903a7ac9aa..000000000000
--- a/extensions/telegram/src/telegram-ingress-spool.types.ts
+++ /dev/null
@@ -1,23 +0,0 @@
-// Telegram plugin module defines ingress spool record contracts.
-
-export type TelegramSpooledUpdateClaimOwner = {
-  processId: string;
-  processPid: number;
-  claimedAt: number;
-  claimToken?: string;
-};
-
-export type TelegramSpooledUpdate = {
-  updateId: number;
-  path: string;
-  update: unknown;
-  receivedAt: number;
-  attempts?: number;
-  lastAttemptAt?: number;
-  lastError?: string;
-  claim?: TelegramSpooledUpdateClaimOwner;
-};
-
-export type ClaimedTelegramSpooledUpdate = TelegramSpooledUpdate & {
-  pendingPath: string;
-};
diff --git a/extensions/telegram/src/telegram-ingress-supersede-auth.ts b/extensions/telegram/src/telegram-ingress-supersede-auth.ts
new file mode 100644
index 000000000000..47d4798da1fc
--- /dev/null
+++ b/extensions/telegram/src/telegram-ingress-supersede-auth.ts
@@ -0,0 +1,200 @@
+// Telegram plugin module owns supersede sender authorization policy.
+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";
+
+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;
+}
diff --git a/extensions/telegram/src/telegram-ingress-supersede.test.ts b/extensions/telegram/src/telegram-ingress-supersede.test.ts
index 72ae74fd0a40..87d1f42f2173 100644
--- a/extensions/telegram/src/telegram-ingress-supersede.test.ts
+++ b/extensions/telegram/src/telegram-ingress-supersede.test.ts
@@ -27,10 +27,10 @@ afterEach(() => {
 });
 import type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js";
 import {
-  createShouldSupersedeTelegramSpooledPending,
   isTelegramAmbientSpooledUpdate,
   isTelegramSpooledUpdateSenderAuthorized,
-} from "./telegram-ingress-supersede.js";
+} from "./telegram-ingress-supersede-auth.js";
+import { createShouldSupersedeTelegramSpooledPending } from "./telegram-ingress-supersede.js";
 
 const OWNER_ID = "111";
 const STRANGER_ID = "999";
diff --git a/extensions/telegram/src/telegram-ingress-supersede.ts b/extensions/telegram/src/telegram-ingress-supersede.ts
index 4c0f03b0fd7a..fb45a1575d52 100644
--- a/extensions/telegram/src/telegram-ingress-supersede.ts
+++ b/extensions/telegram/src/telegram-ingress-supersede.ts
@@ -11,18 +11,13 @@ import {
   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";
+import {
+  isTelegramAmbientSpooledUpdate,
+  isTelegramSpooledUpdateSenderAuthorized,
+  type TelegramSupersedeAuthContext,
+} from "./telegram-ingress-supersede-auth.js";
 
 function isRecognizedTelegramTextCommand(rawText: string, botUsername?: string): boolean {
   return (
@@ -135,195 +130,6 @@ function extractUpdateText(update: unknown): string {
   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
diff --git a/extensions/telegram/src/webhook.test.ts b/extensions/telegram/src/webhook.test.ts
index 6d3b15e7508b..19eadf2f5901 100644
--- a/extensions/telegram/src/webhook.test.ts
+++ b/extensions/telegram/src/webhook.test.ts
@@ -20,11 +20,11 @@ import {
 import { setTelegramRuntime } from "./runtime.js";
 import { clearTelegramRuntimeForTest as clearTelegramRuntime } from "./runtime.test-support.js";
 import type { TelegramRuntime } from "./runtime.types.js";
+import { writeTelegramSpooledUpdate } from "./telegram-ingress-spool.js";
 import {
   listTelegramSpooledUpdateClaims,
   listTelegramSpooledUpdates,
-  writeTelegramSpooledUpdate,
-} from "./telegram-ingress-spool.js";
+} from "./telegram-ingress-spool.test-support.js";
 
 const telegramSpooledRetryDeadLetterMinAgeMs = 24 * 60 * 60 * 1000;
 
diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts
index 711753bc3378..fce2b817c9d9 100644
--- a/src/auto-reply/get-reply-options.types.ts
+++ b/src/auto-reply/get-reply-options.types.ts
@@ -50,7 +50,7 @@ export type QueuedReplyDeliveryCorrelation = {
  * Exclusive: each lifecycle is its own collect-admission identity.
  * Cancel-only: share collect identity via ownerKey (gateway chat.send).
  */
-export type TurnAdoptionAdmission = "exclusive" | "cancel-only";
+type TurnAdoptionAdmission = "exclusive" | "cancel-only";
 
 /**
  * Canonical turn-ownership lifecycle (adopt / defer / abandon / settle).
diff --git a/src/channels/message/index.ts b/src/channels/message/index.ts
index 43303fefd0cb..0d2596e16cd2 100644
--- a/src/channels/message/index.ts
+++ b/src/channels/message/index.ts
@@ -40,11 +40,7 @@ export {
   createTypingCallbacks,
   resolveChannelSourceReplyDeliveryMode,
 } from "./reply-pipeline.js";
-export type {
-  ChannelIngressDispatchLifecycle,
-  ChannelIngressDrain,
-  ChannelIngressDrainDispatchResult,
-} from "./ingress-drain.js";
+export type { ChannelIngressDrain } from "./ingress-drain.js";
 export type {
   ChannelIngressQueue,
   ChannelIngressQueueClaim,
@@ -53,7 +49,6 @@ export type {
   ChannelIngressQueueRecord,
 } from "./ingress-queue.js";
 export type { MessageAckPolicy, MessageReceiveContext } from "./receive.js";
-export type { IngressNonRetryableFailure } from "./ingress-retry-policy.js";
 export type {
   ChannelMessageAdapterShape,
   ChannelMessageDurableFinalAdapter,
diff --git a/src/channels/message/ingress-claim-owner.ts b/src/channels/message/ingress-claim-owner.ts
index d2f50c2ce0b7..8b3d08f6cac5 100644
--- a/src/channels/message/ingress-claim-owner.ts
+++ b/src/channels/message/ingress-claim-owner.ts
@@ -13,13 +13,13 @@ import type { ChannelIngressQueueClaim, ChannelIngressQueueCorruptClaim } from "
 // so recovery can reclaim it even when the owner process still exists.
 export const INGRESS_CLAIM_LEASE_MS = 30 * 60 * 1000;
 
-export type IngressClaimOwnerIdentity = {
+type IngressClaimOwnerIdentity = {
   processId: string;
   processPid: number;
   claimedAt: number;
 };
 
-export type IngressClaimLivenessOptions = {
+type IngressClaimLivenessOptions = {
   maxAgeMs?: number;
   now?: number;
   /** Test seam for PID existence (including Linux TID impersonation). */
@@ -87,7 +87,7 @@ export function processPidFromOwnerId(ownerId: string): number {
 }
 
 /** Instance UUID from ownerId `pid:startToken:uuid`. */
-export function processInstanceIdFromOwnerId(ownerId: string): string | null {
+function processInstanceIdFromOwnerId(ownerId: string): string | null {
   const parts = ownerId.split(":");
   if (parts.length < 3) {
     return null;
diff --git a/src/channels/message/ingress-drain.test.ts b/src/channels/message/ingress-drain.test.ts
index 89ba121629e3..5ac2c53637d5 100644
--- a/src/channels/message/ingress-drain.test.ts
+++ b/src/channels/message/ingress-drain.test.ts
@@ -10,8 +10,12 @@ import {
   createChannelIngressDrain,
   DEFAULT_INGRESS_ADOPTION_STALL_MS,
   isIngressAdoptionLostError,
-  type ChannelIngressDispatchLifecycle,
 } from "./ingress-drain.js";
+
+// Module-private in ingress-drain.ts; derive from the factory signature.
+type ChannelIngressDispatchLifecycle = Parameters<
+  Parameters<typeof createChannelIngressDrain>[0]["dispatchClaimedEvent"]
+>[1];
 import { createChannelIngressQueue } from "./ingress-queue.js";
 import {
   DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
diff --git a/src/channels/message/ingress-drain.ts b/src/channels/message/ingress-drain.ts
index 2114650bd8e8..24801cb4b0cd 100644
--- a/src/channels/message/ingress-drain.ts
+++ b/src/channels/message/ingress-drain.ts
@@ -49,14 +49,14 @@ class IngressAdoptionLostError extends Error {
   }
 }
 
-export type IngressAdoptionLost = IngressAdoptionLostError;
+type IngressAdoptionLost = IngressAdoptionLostError;
 
 export function isIngressAdoptionLostError(error: unknown): error is IngressAdoptionLostError {
   return error instanceof IngressAdoptionLostError;
 }
 
 /** Full pre-adoption → adoption ownership lifecycle for one claimed event. */
-export type ChannelIngressDispatchLifecycle = {
+type ChannelIngressDispatchLifecycle = {
   /** Pre-adoption only. After adopt the drain treats this signal as inert. */
   abortSignal: AbortSignal;
   /**
@@ -83,7 +83,7 @@ export type ChannelIngressDispatchLifecycle = {
   onAbandoned: () => void;
 };
 
-export type ChannelIngressDrainDispatchResult =
+type ChannelIngressDrainDispatchResult =
   | { kind: "completed" }
   | { kind: "deferred" }
   | { kind: "failed-retryable"; error: unknown };
diff --git a/src/channels/message/ingress-retry-policy.test.ts b/src/channels/message/ingress-retry-policy.test.ts
index 7319f28448fa..73d4743776de 100644
--- a/src/channels/message/ingress-retry-policy.test.ts
+++ b/src/channels/message/ingress-retry-policy.test.ts
@@ -3,19 +3,12 @@ import { describe, expect, it } from "vitest";
 import {
   DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
   DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
-  resolveIngressAttemptNumber,
   resolveIngressFailureDisposition,
   resolveIngressRetryDelayMs,
   shouldDeadLetterRetryableIngressEvent,
 } from "./ingress-retry-policy.js";
 
 describe("ingress retry policy", () => {
-  it("resolves attempt numbers from stored attempts", () => {
-    expect(resolveIngressAttemptNumber({ receivedAt: 0 })).toBe(1);
-    expect(resolveIngressAttemptNumber({ receivedAt: 0, attempts: 0 })).toBe(1);
-    expect(resolveIngressAttemptNumber({ receivedAt: 0, attempts: 3 })).toBe(4);
-  });
-
   it.each([
     {
       name: "no prior error → immediate",
diff --git a/src/channels/message/ingress-retry-policy.ts b/src/channels/message/ingress-retry-policy.ts
index fc0e39ff1677..c0734049c84b 100644
--- a/src/channels/message/ingress-retry-policy.ts
+++ b/src/channels/message/ingress-retry-policy.ts
@@ -16,7 +16,7 @@ export type IngressRetryPolicyConfig = {
   maxMs?: number;
 };
 
-export type IngressRetryEventFacts = {
+type IngressRetryEventFacts = {
   receivedAt: number;
   attempts?: number;
   lastAttemptAt?: number;
@@ -28,7 +28,7 @@ export type IngressNonRetryableFailure = {
   message: string;
 };
 
-export type IngressFailureDisposition =
+type IngressFailureDisposition =
   | {
       kind: "fail";
       reason: string;
@@ -51,7 +51,7 @@ function resolveConfig(config?: IngressRetryPolicyConfig) {
 }
 
 /** Next attempt number after a failed dispatch (1-based for the attempt just finished). */
-export function resolveIngressAttemptNumber(event: IngressRetryEventFacts): number {
+function resolveIngressAttemptNumber(event: IngressRetryEventFacts): number {
   return (event.attempts ?? 0) + 1;
 }