Merge pull request #10 from llblab/dev

0.6.2: fix telegram queue dispatch after reload

Author: llblab

AGENTS.md

@@ -59,7 +59,8 @@
 - Dispatch must still respect active turns, pending prompt dispatch, unsettled control-item execution, compaction, and pi pending-message state
 - Telegram `/stop` is a queue-reset command: clear pending model-switch state, all queued Telegram items, and aborted-turn history preservation before aborting the active run so the next Telegram message starts from a clean queue.
 - Telegram `reply_to_message` context is prompt-only: preserve raw new-message text/caption for slash-command parsing, then inject quoted text/caption as `[reply]` context only while building or editing queued prompt turns.
-- Long-lived timers, pollers, and ownership watchers must not depend on live pi context objects after session replacement. Snapshot primitive identity such as `cwd` when installing a watcher, stop local timers during session shutdown, and catch stale-context status updates during async cleanup.
+- Long-lived timers, pollers, and ownership watchers must not depend on live pi context objects after session replacement. Snapshot primitive identity such as `cwd` when installing a watcher, stop local timers during session shutdown, and prefer lifecycle-bound context ports over stale-context catch wrappers.
+- Deferred queue dispatch must be session-bound: bind on `session_start`, unbind/cancel on `session_shutdown`, and resolve the current bound context only when the deferred callback fires.
 - Prompt items should remain in the queue until `agent_start` consumes the dispatched turn; removing them earlier breaks active-turn binding, preview delivery, and end-of-turn follow-up behavior
 - In-flight `/model` switching is supported only for Telegram-owned active turns and is implemented as set-model plus synthetic continuation turn plus abort
 - If a tool call is active during in-flight `/model` switching, the abort is delayed until that tool finishes instead of interrupting the tool mid-flight

BACKLOG.md

@@ -1,3 +1,5 @@
 # Project Backlog
 
-No open backlog items.
+## Open Work
+
+No open work.

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.6.2: Reload-Stale Queue Dispatch Hotfix
+
+- `[Queue Dispatch]` Deferred post-agent-end queue dispatch is now session-bound and canceled on session shutdown. Impact: `/reload` and session replacement can no longer leave old queue timers calling stale `ExtensionContext` methods such as `ctx.isIdle()`.
+- `[Typing Safety]` Typing-loop transport failures now go only to runtime diagnostics instead of updating status through an interval-captured context. Impact: typing errors remain visible in diagnostics without retaining live `ExtensionContext` in timer error paths.
+- `[Lock Watcher Safety]` Ownership-loss watchers now retain only the snapshotted lock identity and stop polling without a captured live context. Impact: singleton takeover cleanup no longer needs stale-prone status refreshes from watcher callbacks.
+- `[Media Group Safety]` Media-group debounce timers now flush through controller state instead of closure-capturing the inbound context. Impact: album coalescing keeps the same behavior while reducing stale-context retention in timer callbacks.
+
 ## 0.6.1: Outbound Action & Command Timeout Hardening
 
 - `[Command Template Runtime]` Timed-out command templates now escalate from `SIGTERM` to `SIGKILL` when the child process does not exit. Impact: attachment and outbound-handler pipelines no longer hang forever on commands that ignore graceful termination.

docs/architecture.md

@@ -112,7 +112,7 @@ Dispatch is gated by:
 - `ctx.isIdle()` being true
 - `ctx.hasPendingMessages()` being false
 
-This prevents queue races around rapid follow-ups, `/compact`, and mixed local plus Telegram activity. Telegram `/status` and `/model` execute immediately; the dispatch controller still serializes any deferred control items so a queued control action must settle before the next queued action can dispatch.
+This prevents queue races around rapid follow-ups, `/compact`, and mixed local plus Telegram activity. Post-agent-end dispatch retries are scheduled through a session-bound deferred dispatcher that activates on session start, cancels timers on session shutdown, and skips callbacks from older generations before they touch `ExtensionContext`. Telegram `/status` and `/model` execute immediately; the dispatch controller still serializes any deferred control items so a queued control action must settle before the next queued action can dispatch.
 
 ### Abort Behavior
 

index.ts

@@ -46,10 +46,16 @@ export default function (pi: Pi.ExtensionAPI) {
   const runtimeEvents = Status.createTelegramRuntimeEventRecorder({
     getBotToken: configStore.getBotToken,
   });
-  const mediaGroupRuntime =
-    Media.createTelegramMediaGroupController<Api.TelegramMessage>();
+  const mediaGroupRuntime = Media.createTelegramMediaGroupController<
+    Api.TelegramMessage,
+    Pi.ExtensionContext
+  >();
   const telegramQueueStore =
     Queue.createTelegramQueueStore<Pi.ExtensionContext>();
+  const deferredQueueDispatchRuntime =
+    Queue.createTelegramDeferredQueueDispatchRuntime<Pi.ExtensionContext>({
+      recordRuntimeEvent: runtimeEvents.record,
+    });
   const pollingControllerState = Polling.createTelegramPollingControllerState();
   const { getStatusLines, updateStatus } =
     Status.createTelegramBridgeStatusRuntime<
@@ -88,12 +94,14 @@ export default function (pi: Pi.ExtensionAPI) {
       updateStatus,
     });
   const attachmentHandlerRuntime =
-    AttachmentHandlers.createTelegramAttachmentHandlerRuntime<Pi.ExtensionContext>({
-      getHandlers: configStore.getAttachmentHandlers,
-      execCommand: CommandTemplates.execCommandTemplate,
-      getCwd: Pi.getExtensionContextCwd,
-      recordRuntimeEvent: runtimeEvents.record,
-    });
+    AttachmentHandlers.createTelegramAttachmentHandlerRuntime<Pi.ExtensionContext>(
+      {
+        getHandlers: configStore.getAttachmentHandlers,
+        execCommand: CommandTemplates.execCommandTemplate,
+        getCwd: Pi.getExtensionContextCwd,
+        recordRuntimeEvent: runtimeEvents.record,
+      },
+    );
 
   // --- Telegram API ---
 
@@ -149,6 +157,7 @@ export default function (pi: Pi.ExtensionAPI) {
       hasDispatchPending: bridgeRuntime.lifecycle.hasDispatchPending,
       isIdle: Pi.isExtensionContextIdle,
       hasPendingMessages: Pi.hasExtensionContextPendingMessages,
+      hasDispatchContext: deferredQueueDispatchRuntime.isBound,
       updateStatus,
       sendTextReply,
       recordRuntimeEvent: runtimeEvents.record,
@@ -274,8 +283,10 @@ export default function (pi: Pi.ExtensionAPI) {
       setPendingModelSwitch: pendingModelSwitchStore.set,
       syncCounters: bridgeRuntime.queue.syncCounters,
       syncFlags: bridgeRuntime.lifecycle.syncFlags,
+      bindDeferredDispatchContext: deferredQueueDispatchRuntime.bind,
       prepareTempDir,
       updateStatus,
+      unbindDeferredDispatchContext: deferredQueueDispatchRuntime.unbind,
       clearPendingMediaGroups: mediaGroupRuntime.clear,
       clearModelMenuState: modelMenuRuntime.clear,
       getActiveTurnChatId: activeTurnRuntime.getChatId,
@@ -352,6 +363,8 @@ export default function (pi: Pi.ExtensionAPI) {
         clearDispatchPending: bridgeRuntime.lifecycle.clearDispatchPending,
       }),
       dispatchNextQueuedTelegramTurn,
+      requestDeferredDispatchNextQueuedTelegramTurn:
+        deferredQueueDispatchRuntime.request,
       clearPreview: previewRuntime.clear,
       setPreviewPendingText: previewRuntime.setPendingText,
       finalizeMarkdownPreview: previewRuntime.finalizeMarkdown,
@@ -362,16 +375,16 @@ export default function (pi: Pi.ExtensionAPI) {
         sendTextReply,
         recordRuntimeEvent: runtimeEvents.record,
       }),
-      planOutboundReply: OutboundHandlers.createTelegramOutboundReplyPlanner(
-        buttonActionStore,
-      ),
-      sendOutboundReplyArtifacts: OutboundHandlers.createTelegramOutboundReplyArtifactSender({
-        execCommand: CommandTemplates.execCommandTemplate,
-        sendMultipart: callMultipart,
-        sendTextReply,
-        getHandlers: configStore.getOutboundHandlers,
-        recordRuntimeEvent: runtimeEvents.record,
-      }),
+      planOutboundReply:
+        OutboundHandlers.createTelegramOutboundReplyPlanner(buttonActionStore),
+      sendOutboundReplyArtifacts:
+        OutboundHandlers.createTelegramOutboundReplyArtifactSender({
+          execCommand: CommandTemplates.execCommandTemplate,
+          sendMultipart: callMultipart,
+          sendTextReply,
+          getHandlers: configStore.getOutboundHandlers,
+          recordRuntimeEvent: runtimeEvents.record,
+        }),
       getActiveToolExecutions: bridgeRuntime.lifecycle.getActiveToolExecutions,
       setActiveToolExecutions: bridgeRuntime.lifecycle.setActiveToolExecutions,
       triggerPendingModelSwitchAbort: modelSwitchController.triggerPendingAbort,

lib/locks.ts

@@ -246,13 +246,6 @@ export function createTelegramLockedPollingRuntime<
     clearInterval(ownershipInterval);
     ownershipInterval = undefined;
   };
-  const updateStatusSafely = (ctx: TContext, phase: string) => {
-    try {
-      deps.updateStatus(ctx);
-    } catch (error) {
-      deps.recordRuntimeEvent?.("lock", error, { phase });
-    }
-  };
   const suspendPolling = async () => {
     stopOwnershipWatcher();
     if (ownershipStop) {
@@ -261,7 +254,7 @@ export function createTelegramLockedPollingRuntime<
     }
     await deps.stopPolling();
   };
-  const stopAfterOwnershipLoss = (ctx: TContext) => {
+  const stopAfterOwnershipLoss = () => {
     if (ownershipStop) return;
     stopOwnershipWatcher();
     ownershipStop = deps
@@ -271,15 +264,14 @@ export function createTelegramLockedPollingRuntime<
       )
       .finally(() => {
         ownershipStop = undefined;
-        updateStatusSafely(ctx, "ownership-loss-status");
       });
   };
   const startOwnershipWatcher = (ctx: TContext) => {
     const owner = snapshotLockContext(ctx);
     stopOwnershipWatcher();
     ownershipInterval = setInterval(() => {
       if (deps.lock.owns(owner)) return;
-      stopAfterOwnershipLoss(ctx);
+      stopAfterOwnershipLoss();
     }, ownershipCheckMs);
     ownershipInterval.unref?.();
   };

lib/media.ts

@@ -59,17 +59,20 @@ export interface TelegramMediaGroupMessage {
   media_group_id?: string;
 }
 
-export interface TelegramMediaGroupState<TMessage> {
+export interface TelegramMediaGroupState<TMessage, TContext = unknown> {
   messages: TMessage[];
+  context?: TContext;
   flushTimer?: ReturnType<typeof setTimeout>;
 }
 
 export interface TelegramMediaGroupController<
   TMessage extends TelegramMediaGroupMessage,
+  TContext = unknown,
 > {
   queueMessage: (options: {
     message: TMessage;
-    dispatchMessages: (messages: TMessage[]) => void;
+    context?: TContext;
+    dispatchMessages: (messages: TMessage[], ctx?: TContext) => void;
   }) => boolean;
   removeMessages: (messageIds: number[]) => number;
   clear: () => void;
@@ -79,7 +82,7 @@ export interface TelegramMediaGroupDispatchRuntimeDeps<
   TMessage extends TelegramMediaGroupMessage,
   TContext,
 > {
-  mediaGroups: TelegramMediaGroupController<TMessage>;
+  mediaGroups: TelegramMediaGroupController<TMessage, TContext>;
   dispatchMessages: (messages: TMessage[], ctx: TContext) => Promise<void>;
 }
 
@@ -249,7 +252,7 @@ export function getTelegramMediaGroupKey(
 export function removePendingTelegramMediaGroupMessages<
   TMessage extends TelegramMediaGroupMessage,
 >(
-  groups: Map<string, TelegramMediaGroupState<TMessage>>,
+  groups: Map<string, TelegramMediaGroupState<TMessage, unknown>>,
   messageIds: number[],
   clearTimer: (timer: ReturnType<typeof setTimeout>) => void,
 ): number {
@@ -273,45 +276,50 @@ export function removePendingTelegramMediaGroupMessages<
 
 export function queueTelegramMediaGroupMessage<
   TMessage extends TelegramMediaGroupMessage,
+  TContext = unknown,
 >(options: {
   message: TMessage;
-  groups: Map<string, TelegramMediaGroupState<TMessage>>;
+  context?: TContext;
+  groups: Map<string, TelegramMediaGroupState<TMessage, TContext>>;
   debounceMs: number;
   setTimer: (callback: () => void, ms: number) => ReturnType<typeof setTimeout>;
   clearTimer: (timer: ReturnType<typeof setTimeout>) => void;
-  dispatchMessages: (messages: TMessage[]) => void;
+  dispatchMessages: (messages: TMessage[], ctx?: TContext) => void;
 }): boolean {
   const key = getTelegramMediaGroupKey(options.message);
   if (!key) return false;
   const existing = options.groups.get(key) ?? { messages: [] };
   existing.messages.push(options.message);
+  existing.context = options.context;
   if (existing.flushTimer) options.clearTimer(existing.flushTimer);
   existing.flushTimer = options.setTimer(() => {
     const state = options.groups.get(key);
     options.groups.delete(key);
     if (!state) return;
-    options.dispatchMessages(state.messages);
+    options.dispatchMessages(state.messages, state.context);
   }, options.debounceMs);
   options.groups.set(key, existing);
   return true;
 }
 
 export function createTelegramMediaGroupController<
   TMessage extends TelegramMediaGroupMessage,
+  TContext = unknown,
 >(
   options: TelegramMediaGroupControllerOptions = {},
-): TelegramMediaGroupController<TMessage> {
-  const groups = new Map<string, TelegramMediaGroupState<TMessage>>();
+): TelegramMediaGroupController<TMessage, TContext> {
+  const groups = new Map<string, TelegramMediaGroupState<TMessage, TContext>>();
   const debounceMs = options.debounceMs ?? TELEGRAM_MEDIA_GROUP_DEBOUNCE_MS;
   const setTimer =
     options.setTimer ??
     ((callback: () => void, ms: number): ReturnType<typeof setTimeout> =>
       setTimeout(callback, ms));
   const clearTimer = options.clearTimer ?? clearTimeout;
   return {
-    queueMessage: ({ message, dispatchMessages }) =>
+    queueMessage: ({ message, context, dispatchMessages }) =>
       queueTelegramMediaGroupMessage({
         message,
+        context,
         groups,
         debounceMs,
         setTimer,
@@ -339,8 +347,11 @@ export function createTelegramMediaGroupDispatchRuntime<
     handleMessage: async (message, ctx) => {
       const queuedMediaGroup = deps.mediaGroups.queueMessage({
         message,
-        dispatchMessages: (messages) => {
-          void deps.dispatchMessages(messages, ctx);
+        context: ctx,
+        dispatchMessages: (messages, queuedCtx) => {
+          if (queuedCtx !== undefined) {
+            void deps.dispatchMessages(messages, queuedCtx);
+          }
         },
       });
       if (queuedMediaGroup) return;

lib/polling.ts

@@ -97,7 +97,7 @@ export interface TelegramPollingRuntimeDeps<TContext> {
   setPollingController: (controller: AbortController | undefined) => void;
   stopTypingLoop: () => unknown;
   runPollLoop: (ctx: TContext, signal: AbortSignal) => Promise<void>;
-  updateStatus: (ctx: TContext) => void;
+  updateStatus: (ctx: TContext, message?: string) => void;
   createAbortController?: () => AbortController;
 }