Merge pull request #88 from llblab/dev

0.18.4: Windows threaded mode hotfix

Author: llblab

BACKLOG.md

@@ -36,8 +36,11 @@ Open work:
 
 - [ ] Live smoke Threaded Mode on native Windows without WSL.
   - Scope: leader/follower `/telegram-connect`, follower heartbeat, forwarded Bot API calls, restore flows, lifecycle announcements, shutdown cleanup, and reconnect/reload behavior.
+  - Observed: a same-directory follower can see a live leader lock but fail registration with `connect ENOENT \\.\\pipe\\...`, leaving the follower disconnected during leader reload/hot activation timing.
+  - Observed: Windows/QEMU live polling/dispatch can lag until reload; inbound messages appear to increase the extension queue count, but the next queued item is not dispatched promptly.
   - Baseline: deterministic path tests run everywhere, and a Windows-only named-pipe roundtrip regression runs when the suite executes on `win32`. Live Windows smoke remains unavailable in this environment.
-- [ ] If Windows live smoke exposes pipe-specific behavior, add a minimized regression at the bus transport boundary before changing higher-level Threaded Mode logic.
+- [x] Add a minimized registration-boundary regression for transient leader endpoint startup races.
+- [x] Add a session-bound queue dispatch watchdog so queued Telegram work can recover if a one-shot wakeup/timer is missed.
 
 Done when: Threaded Mode leader/follower operation works on native Windows with the same safety guarantees as Unix-like systems, and unsupported transport assumptions are covered by tests/docs.
 

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Unreleased
 
+## 0.18.4: Windows Threaded Mode hotfix
+
+- `[Windows IPC]` Follower registration now retries transient local bus connection failures while the leader named pipe/socket is still coming online. Impact: a same-directory Windows follower is less likely to fail `/telegram-connect` with `connect ENOENT \\.\\pipe\\...` during leader reload or hot Threaded Mode activation.
+- `[Queue]` A session-bound queue dispatch watchdog now retries dispatch while Telegram work remains queued. Impact: if a platform drops the one-shot deferred dispatch wakeup, queued Telegram messages can resume without waiting for a manual `/reload`.
+
 ## 0.18.3: Threaded Mode live hotfix
 
 - `[Threaded Mode]` Inbound Telegram prompts now request an immediate dispatch and a session-bound deferred retry. Impact: hosts where Pi is not yet dispatch-ready at update handling time no longer need a later `/reload` or command to process the queued prompt.

index.ts

@@ -454,6 +454,12 @@ export default function (pi: Pi.ExtensionAPI) {
       ...promptDispatchRuntime,
       sendUserMessage,
     }).dispatchNext;
+  const queueDispatchWatchdogRuntime =
+    Queue.createTelegramQueueDispatchWatchdogRuntime({
+      hasQueuedItems: telegramQueueStore.hasQueuedItems,
+      dispatchNextQueuedTelegramTurn,
+      recordRuntimeEvent,
+    });
   const nativeMarkdownDraftSender =
     TelegramApi.createTelegramNativeMarkdownDraftSender({
       sendMessageDraft,
@@ -1168,8 +1174,10 @@ export default function (pi: Pi.ExtensionAPI) {
       async onSessionStart(event, ctx) {
         await lockedPollingRuntime.onSessionStart(event, ctx);
         telegramThreadCapabilityMonitor.start(ctx);
+        queueDispatchWatchdogRuntime.start(ctx);
       },
       async onSessionShutdown() {
+        queueDispatchWatchdogRuntime.stop();
         telegramThreadCapabilityMonitor.stop();
       },
     },

lib/bus-follower.ts

@@ -22,6 +22,8 @@ import {
 
 export const TELEGRAM_BUS_FOLLOWER_PROMOTION_GRACE_MS = 2_500;
 export const TELEGRAM_FOLLOWER_SESSION_HANDOFF_TTL_MS = 30_000;
+export const TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_ATTEMPTS = 10;
+export const TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_DELAY_MS = 150;
 
 const TELEGRAM_FOLLOWER_SESSION_HANDOFF_KEY =
   "__piTelegramFollowerSessionHandoff";
@@ -150,6 +152,8 @@ export interface TelegramBusFollowerRegistrationRuntimeDeps<
   getPid?: () => number;
   timeoutMs?: number;
   registrationTimeoutMs?: number;
+  registrationRetryAttempts?: number;
+  registrationRetryDelayMs?: number;
   heartbeatMs?: number;
   recordRuntimeEvent?: (
     category: string,
@@ -364,6 +368,19 @@ function isTelegramStaleContextError(error: unknown): boolean {
   );
 }
 
+function isRetryableTelegramBusRegistrationError(error: unknown): boolean {
+  if (!(error instanceof Error)) return false;
+  const code = (error as NodeJS.ErrnoException).code;
+  return code === "ENOENT" || code === "ECONNREFUSED" || code === "EPIPE";
+}
+
+function delayTelegramBusFollowerRegistration(ms: number): Promise<void> {
+  return new Promise((resolve) => {
+    const timer = setTimeout(resolve, ms);
+    timer.unref?.();
+  });
+}
+
 export function createTelegramBusFollowerSessionReplacementSuspender(
   deps: TelegramBusFollowerSessionReplacementSuspenderDeps,
 ): () => Promise<void> {
@@ -588,6 +605,12 @@ export function createTelegramBusFollowerRegistrationRuntime<
   const heartbeatMs = deps.heartbeatMs ?? 1000;
   const registrationTimeoutMs =
     deps.registrationTimeoutMs ?? deps.timeoutMs ?? 30000;
+  const registrationRetryAttempts =
+    deps.registrationRetryAttempts ??
+    TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_ATTEMPTS;
+  const registrationRetryDelayMs =
+    deps.registrationRetryDelayMs ??
+    TELEGRAM_BUS_FOLLOWER_REGISTRATION_RETRY_DELAY_MS;
   let heartbeatInterval: ReturnType<typeof setInterval> | undefined;
   let activeLeaderSocketPath: string | undefined;
   let activeAuthSecret: string | undefined;
@@ -646,30 +669,54 @@ export function createTelegramBusFollowerRegistrationRuntime<
       await deps.startReceiving?.();
       activeAuthSecret = deps.getLeaderAuthSecret?.(leader);
       deps.setActiveAuthSecret?.(activeAuthSecret);
+      const createRegistrationEnvelope = (): Extract<
+        TelegramBusEnvelope,
+        { kind: "follower.register" }
+      > => ({
+        kind: "follower.register",
+        requestId: deps.createRequestId(),
+        auth: activeAuthSecret,
+        registration: {
+          instanceId: deps.instanceId,
+          profileKey:
+            deps.getProfileKey?.(ctx) ??
+            (ctx.cwd ? `cwd:${ctx.cwd}` : undefined),
+          threadName:
+            deps.getThreadName?.(ctx) ??
+            (ctx.cwd ? basename(ctx.cwd) : undefined),
+          cwd: ctx.cwd,
+          pid: getPid(),
+          busSocketPath: deps.followerBusSocketPath,
+          connectedAtMs: getNowMs(),
+        },
+      });
       let response: TelegramBusEnvelope | undefined;
       try {
-        response = await sendTelegramBusLocalEnvelope({
-          socketPath: leaderSocketPath,
-          timeoutMs: registrationTimeoutMs,
-          envelope: {
-            kind: "follower.register",
-            requestId: deps.createRequestId(),
-            auth: activeAuthSecret,
-            registration: {
-              instanceId: deps.instanceId,
-              profileKey:
-                deps.getProfileKey?.(ctx) ??
-                (ctx.cwd ? `cwd:${ctx.cwd}` : undefined),
-              threadName:
-                deps.getThreadName?.(ctx) ??
-                (ctx.cwd ? basename(ctx.cwd) : undefined),
-              cwd: ctx.cwd,
-              pid: getPid(),
-              busSocketPath: deps.followerBusSocketPath,
-              connectedAtMs: getNowMs(),
-            },
-          },
-        });
+        for (let attempt = 1; ; attempt += 1) {
+          try {
+            response = await sendTelegramBusLocalEnvelope({
+              socketPath: leaderSocketPath,
+              timeoutMs: registrationTimeoutMs,
+              envelope: createRegistrationEnvelope(),
+            });
+            break;
+          } catch (error) {
+            if (
+              attempt >= registrationRetryAttempts ||
+              !isRetryableTelegramBusRegistrationError(error)
+            ) {
+              throw error;
+            }
+            deps.recordRuntimeEvent?.("bus", error, {
+              phase: "follower-register-retry",
+              attempt,
+              socketPath: leaderSocketPath,
+            });
+            await delayTelegramBusFollowerRegistration(
+              registrationRetryDelayMs,
+            );
+          }
+        }
       } catch (error) {
         stopHeartbeat();
         activeLeaderSocketPath = undefined;

lib/queue.ts

@@ -1920,6 +1920,76 @@ export function createTelegramDeferredQueueDispatchRuntime<TContext = unknown>(
   };
 }
 
+// --- Dispatch Watchdog Runtime ---
+
+export interface TelegramQueueDispatchWatchdogRuntime<TContext = unknown> {
+  start: (ctx: TContext) => void;
+  stop: () => void;
+  poke: () => void;
+}
+
+export interface TelegramQueueDispatchWatchdogRuntimeDeps<
+  TContext = unknown,
+> extends TelegramRuntimeEventRecorderPort {
+  hasQueuedItems: () => boolean;
+  dispatchNextQueuedTelegramTurn: (ctx: TContext) => void;
+  intervalMs?: number;
+  setInterval?: (
+    callback: () => void,
+    ms: number,
+  ) => ReturnType<typeof setInterval>;
+  clearInterval?: (timer: ReturnType<typeof setInterval>) => void;
+}
+
+export function createTelegramQueueDispatchWatchdogRuntime<
+  TContext = unknown,
+>(
+  deps: TelegramQueueDispatchWatchdogRuntimeDeps<TContext>,
+): TelegramQueueDispatchWatchdogRuntime<TContext> {
+  const intervalMs = deps.intervalMs ?? 1000;
+  const setIntervalFn: NonNullable<
+    TelegramQueueDispatchWatchdogRuntimeDeps<TContext>["setInterval"]
+  > = deps.setInterval ?? ((callback, ms) => setInterval(callback, ms));
+  const clearIntervalFn: NonNullable<
+    TelegramQueueDispatchWatchdogRuntimeDeps<TContext>["clearInterval"]
+  > = deps.clearInterval ?? ((timer) => clearInterval(timer));
+  let ctx: TContext | undefined;
+  let interval: ReturnType<typeof setInterval> | undefined;
+  let dispatchInFlight = false;
+  const tick = (): void => {
+    if (ctx === undefined || dispatchInFlight || !deps.hasQueuedItems()) return;
+    dispatchInFlight = true;
+    try {
+      deps.dispatchNextQueuedTelegramTurn(ctx);
+    } catch (error) {
+      deps.recordRuntimeEvent?.("dispatch", error, {
+        phase: "queue-watchdog",
+      });
+    } finally {
+      dispatchInFlight = false;
+    }
+  };
+  const stop = (): void => {
+    ctx = undefined;
+    if (!interval) return;
+    clearIntervalFn(interval);
+    interval = undefined;
+  };
+  return {
+    start: (nextCtx) => {
+      ctx = nextCtx;
+      if (!interval) {
+        const nextInterval = setIntervalFn(tick, intervalMs);
+        interval = nextInterval;
+        nextInterval.unref?.();
+      }
+      tick();
+    },
+    stop,
+    poke: tick,
+  };
+}
+
 // --- Dispatch Runtime ---
 
 export interface TelegramPromptDeliveryOptions {

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@llblab/pi-telegram",
-  "version": "0.18.3",
+  "version": "0.18.4",
   "private": false,
   "publishConfig": {
     "access": "public"

package.json

@@ -398,6 +398,57 @@ test("Bus follower registration state tracks successful registration and stop",
   }
 });
 
+test("Bus follower registration runtime retries while leader endpoint is starting", async () => {
+  const dir = mkdtempSync(join(tmpdir(), "pi-telegram-bus-follower-retry-"));
+  const socketPath = join(dir, "bus.sock");
+  const registry = createTelegramBusFollowerRegistry();
+  const state = createTelegramBusFollowerRegistrationState();
+  const events: Array<Record<string, unknown> | undefined> = [];
+  const server = createTelegramBusLocalServer({
+    socketPath,
+    handleEnvelope: createTelegramBusLeaderEnvelopeHandler({
+      followerRegistry: registry,
+      provisionFollowerTarget() {
+        return { chatId: -1007, threadId: 42 };
+      },
+    }),
+  });
+  const follower = createTelegramBusFollowerRegistrationRuntime({
+    instanceId: "inst-a",
+    createRequestId: () => "inst-a:1",
+    getNowMs: () => 1000,
+    registrationState: state,
+    registrationTimeoutMs: 50,
+    registrationRetryAttempts: 10,
+    registrationRetryDelayMs: 10,
+    recordRuntimeEvent(_category, _error, details) {
+      events.push(details);
+    },
+  });
+  try {
+    setTimeout(() => {
+      void server.start();
+    }, 25).unref?.();
+    assert.equal(
+      await follower.registerWithLeader(
+        { cwd: "/repo" },
+        { busSocketPath: socketPath },
+      ),
+      true,
+    );
+    assert.equal(state.isRegistered(), true);
+    assert.deepEqual(state.getTarget(), { chatId: -1007, threadId: 42 });
+    assert.equal(
+      events.some((event) => event?.phase === "follower-register-retry"),
+      true,
+    );
+  } finally {
+    follower.stop();
+    await server.stop();
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
 test("Bus follower registration runtime waits for slow target provisioning", async () => {
   const dir = mkdtempSync(
     join(tmpdir(), "pi-telegram-bus-follower-slow-register-"),

tests/bus-follower.test.ts

@@ -34,6 +34,7 @@ import {
   createTelegramPromptEnqueueController,
   createTelegramQueueDispatchController,
   createTelegramQueueDispatchRuntime,
+  createTelegramQueueDispatchWatchdogRuntime,
   createTelegramQueueMutationController,
   createTelegramQueueStore,
   createTelegramSessionLifecycleHooks,
@@ -2806,6 +2807,33 @@ test("Deferred queue dispatch uses only the bound session context", () => {
   assert.deepEqual(events, ["dispatch:new"]);
 });
 
+test("Queue dispatch watchdog retries dispatch for queued work", () => {
+  const events: string[] = [];
+  let hasQueuedItems = false;
+  let intervalCallback: (() => void) | undefined;
+  const watchdog = createTelegramQueueDispatchWatchdogRuntime<{ id: string }>({
+    hasQueuedItems: () => hasQueuedItems,
+    dispatchNextQueuedTelegramTurn: (ctx) => {
+      events.push(`dispatch:${ctx.id}`);
+    },
+    setInterval: (callback) => {
+      intervalCallback = callback;
+      return 1 as unknown as ReturnType<typeof setInterval>;
+    },
+    clearInterval: () => {
+      events.push("clear");
+    },
+  });
+  watchdog.start({ id: "ctx" });
+  assert.deepEqual(events, []);
+  hasQueuedItems = true;
+  watchdog.poke();
+  intervalCallback?.();
+  assert.deepEqual(events, ["dispatch:ctx", "dispatch:ctx"]);
+  watchdog.stop();
+  assert.deepEqual(events, ["dispatch:ctx", "dispatch:ctx", "clear"]);
+});
+
 test("Dispatch controller skips inactive stale contexts before readiness checks", () => {
   const events: string[] = [];
   const controller = createTelegramQueueDispatchController<string>({