Merge 68542eb into e964987

Totalsolutionsync · web-flow · commit 59917efdf995 · 2026-05-21T01:09:06.000Z
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -223,6 +223,11 @@ export class TelegramPollingSession {
   #spooledUpdateHandlerTimeoutMs: number;
   #spooledUpdateHandlerAbortGraceMs: number;
   #deliveryDrainInFlight = false;
+  // Callback supplied by the active isolated-ingress cycle to request the cycle
+  // abort itself when the bot has visibly lost its grammy initialized state
+  // mid-cycle. When set, calling it triggers a clean exit so the outer
+  // `runUntilAbort` loop creates a fresh bot and re-runs `bot.init()`.
+  #requestCycleRestartOnBotReinitNeeded: ((reason: string) => void) | null = null;
 
   constructor(private readonly opts: TelegramPollingSessionOpts) {
     this.#transportState = new TelegramPollingTransportState({
@@ -465,9 +470,20 @@ export class TelegramPollingSession {
       );
       return;
     }
+    const errMessage = formatErrorMessage(params.err);
     this.opts.log(
-      `[telegram][diag] spooled update ${params.update.updateId} failed; keeping for retry: ${formatErrorMessage(params.err)}`,
+      `[telegram][diag] spooled update ${params.update.updateId} failed; keeping for retry: ${errMessage}`,
     );
+    // If the grammy bot has lost its initialized state mid-cycle, every
+    // subsequent update handler fails with the same message in a tight retry
+    // loop. Ask the active cycle to abort itself so the outer runUntilAbort
+    // loop can create a fresh TelegramBot instance and re-run bot.init().
+    if (typeof errMessage === "string" && errMessage.includes("Bot not initialized")) {
+      const requestRestart = this.#requestCycleRestartOnBotReinitNeeded;
+      if (requestRestart) {
+        requestRestart(`spooled update ${params.update.updateId} hit Bot-not-initialized`);
+      }
+    }
   }
 
   async #waitForSpooledUpdateHandlers(): Promise<void> {
@@ -724,6 +740,20 @@ export class TelegramPollingSession {
       void worker.stop();
     };
     this.opts.abortSignal?.addEventListener("abort", stopOnAbort, { once: true });
+    // Install a one-shot callback that the spool failure path can use to ask
+    // this cycle to restart when it sees grammy's "Bot not initialized" error.
+    // Setting restartRequested and stopping the worker tears the cycle down via
+    // the existing try/finally cleanup below.
+    this.#requestCycleRestartOnBotReinitNeeded = (reason: string) => {
+      if (restartRequested) {
+        return;
+      }
+      restartRequested = true;
+      this.opts.log(
+        `[telegram][diag] requesting isolated polling cycle restart to re-run bot.init(): ${reason}`,
+      );
+      void worker.stop();
+    };
     const drainIntervalMs = Math.max(100, Math.floor(ingress.drainIntervalMs ?? 500));
     let drainActive = false;
     const stopBot = () => {
@@ -820,6 +850,9 @@ export class TelegramPollingSession {
       clearInterval(drainTimer);
       unsubscribe();
       this.opts.abortSignal?.removeEventListener("abort", stopOnAbort);
+      // Clear the restart-request callback so a future cycle is not
+      // accidentally talking to this cycle's local state.
+      this.#requestCycleRestartOnBotReinitNeeded = null;
       await worker.stop();
       if (!restartRequested) {
         await drainOnce();
diff --git a/src/agents/auth-profiles/store.ts b/src/agents/auth-profiles/store.ts
@@ -590,7 +590,12 @@ export function loadAuthProfileStoreForSecretsRuntime(agentDir?: string): AuthPr
   return loadAuthProfileStoreForRuntime(agentDir, {
     readOnly: true,
     allowKeychainPrompt: false,
-    resolveLegacyOAuthSidecars: false,
+    // Include legacy OAuth sidecar material when the runtime is resolving
+    // secrets for an agent turn. Without this, embedded agent runs cannot reach
+    // the access token for openai-codex profiles whose `oauthRef.source` is
+    // "openclaw-credentials", and resolveApiKeyForProfile() falls through to
+    // "No API key found".
+    resolveLegacyOAuthSidecars: true,
   });
 }
 
@@ -604,7 +609,10 @@ export function loadAuthProfileStoreWithoutExternalProfiles(
   const options: LoadAuthProfileStoreOptions = {
     readOnly: true,
     allowKeychainPrompt: loadOptions?.allowKeychainPrompt ?? false,
-    resolveLegacyOAuthSidecars: loadOptions?.resolveLegacyOAuthSidecars ?? false,
+    // Default sidecar resolution to true so callers that do not explicitly
+    // override still pick up legacy OAuth credential material for isolated and
+    // sub-agent auth resolution paths.
+    resolveLegacyOAuthSidecars: loadOptions?.resolveLegacyOAuthSidecars ?? true,
   };
   const store = loadAuthProfileStoreForAgent(agentDir, options);
   const authPath = resolveAuthStorePath(agentDir);
@@ -639,20 +647,31 @@ export function ensureAuthProfileStore(
 
 export function ensureAuthProfileStoreWithoutExternalProfiles(
   agentDir?: string,
-  options?: { allowKeychainPrompt?: boolean },
+  options?: { allowKeychainPrompt?: boolean; resolveLegacyOAuthSidecars?: boolean },
 ): AuthProfileStore {
-  const runtimeStore = resolveRuntimeAuthProfileStore(agentDir, options);
+  // Forward `resolveLegacyOAuthSidecars` through this entry point so embedded
+  // runner sub-agents and isolated session lanes can read legacy sidecar
+  // credential material. Default true to match
+  // `loadAuthProfileStoreWithoutExternalProfiles`.
+  const resolveLegacyOAuthSidecars = options?.resolveLegacyOAuthSidecars ?? true;
+  const effectiveOptions: LoadAuthProfileStoreOptions = {
+    ...(options ?? {}),
+    resolveLegacyOAuthSidecars,
+  };
+  const runtimeStore = resolveRuntimeAuthProfileStore(agentDir, effectiveOptions);
   if (runtimeStore) {
     return runtimeStore;
   }
-  const store = loadAuthProfileStoreForAgent(agentDir, options);
+  const store = loadAuthProfileStoreForAgent(agentDir, effectiveOptions);
   const authPath = resolveAuthStorePath(agentDir);
   const mainAuthPath = resolveAuthStorePath();
   if (!agentDir || authPath === mainAuthPath) {
     return store;
   }
 
-  const mainStore = loadAuthProfileStoreForAgent(undefined, options);
+  // Use the same options for the main fallback load; sub-agents that merge in
+  // the main store need the same legacy sidecar material.
+  const mainStore = loadAuthProfileStoreForAgent(undefined, effectiveOptions);
   return mergeAuthProfileStores(mainStore, store);
 }
 
diff --git a/src/logging/diagnostic-session-attention.test.ts b/src/logging/diagnostic-session-attention.test.ts
@@ -68,7 +68,10 @@ describe("classifySessionAttention", () => {
         reason: "queued_behind_terminal_active_work",
         classification: "stalled_agent_run",
         activeWorkKind: "embedded_run",
-        recoveryEligible: false,
+        // Terminal progress signal + queued items = lane is effectively
+        // done with the active turn; recovery coordinator should release
+        // the lane so the queue can drain.
+        recoveryEligible: true,
       },
     },
     {
diff --git a/src/logging/diagnostic-session-attention.ts b/src/logging/diagnostic-session-attention.ts
@@ -14,7 +14,7 @@ export type SessionAttentionClassification =
       reason: string;
       classification: "blocked_tool_call" | "stalled_agent_run";
       activeWorkKind?: DiagnosticSessionActiveWorkKind;
-      recoveryEligible: false;
+      recoveryEligible: boolean;
     }
   | {
       eventType: "session.stuck";
@@ -48,12 +48,17 @@ export function classifySessionAttention(params: {
       params.activity.activeWorkKind === "embedded_run" &&
       isTerminalDiagnosticProgressReason(params.activity.lastProgressReason)
     ) {
+      // The active embedded_run has emitted a terminal progress signal
+      // (e.g., `rawResponseItem/completed`, `embedded_run:ended`) yet the
+      // lane is still holding queued items. The work signal indicates the
+      // active turn is effectively done, so allow the recovery coordinator
+      // to release the lane and let the queue drain.
       return {
         eventType: "session.stalled",
         reason: "queued_behind_terminal_active_work",
         classification: "stalled_agent_run",
         activeWorkKind: params.activity.activeWorkKind,
-        recoveryEligible: false,
+        recoveryEligible: true,
       };
     }
     if ((params.activity.lastProgressAgeMs ?? 0) > params.staleMs) {
diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts
@@ -1,4 +1,5 @@
 import { monitorEventLoopDelay, performance } from "node:perf_hooks";
+import { resolveEmbeddedSessionLane } from "../agents/pi-embedded-runner/lanes.js";
 import { getRuntimeConfig } from "../config/config.js";
 import { resolveAllAgentSessionStoreTargetsSync } from "../config/sessions/targets.js";
 import type { OpenClawConfig } from "../config/types.openclaw.js";
@@ -9,6 +10,7 @@ import {
   type DiagnosticPhaseSnapshot,
   type DiagnosticLivenessWarningReason,
 } from "../infra/diagnostic-events.js";
+import { resetCommandLane } from "../process/command-queue.js";
 import { emitDiagnosticMemorySample, resetDiagnosticMemoryForTest } from "./diagnostic-memory.js";
 import {
   getCurrentDiagnosticPhase,
@@ -737,6 +739,22 @@ export function logSessionStateChange(
   if (params.state === "idle") {
     state.queueDepth = Math.max(0, state.queueDepth - 1);
     state.activeQueuedTurn = false;
+    // Belt-and-suspenders: if the lane returns to idle but still has queued
+    // items, force a pump. Normally `drainLane` re-fires recursively after
+    // each task completes, but in production we have observed lanes that go
+    // `idle` with `queueDepth > 0` and never dequeue (e.g., after an embedded
+    // run ends with terminal progress). Calling `resetCommandLane` bumps the
+    // lane generation, clears any stale `activeTaskIds`, and re-invokes
+    // `drainLane` — a no-op when the lane queue is already empty.
+    if (state.queueDepth > 0 && state.sessionKey) {
+      try {
+        resetCommandLane(resolveEmbeddedSessionLane(state.sessionKey));
+      } catch (err) {
+        diag.warn(
+          `lane-pump-on-idle failed: sessionKey=${state.sessionKey} error="${String(err)}"`,
+        );
+      }
+    }
   }
   if (!isProbeSession && diag.isEnabled("debug")) {
     diag.debug(
@@ -1116,6 +1134,14 @@ export function startDiagnosticHeartbeat(
           thresholdMs: stuckSessionWarnMs,
           abortThresholdMs: stuckSessionAbortMs,
         });
+        const activeAbortRecoveryEligible =
+          classification !== undefined &&
+          isActiveAbortRecoveryEligible({
+            classification,
+            activity,
+            ageMs: attentionAgeMs,
+            stuckSessionAbortMs,
+          });
         if (classification?.recoveryEligible) {
           requestStuckSessionRecovery({
             recover: opts?.recoverStuckSession ?? recoverStuckSession,
@@ -1125,19 +1151,12 @@ export function startDiagnosticHeartbeat(
               sessionKey: state.sessionKey,
               ageMs: attentionAgeMs,
               queueDepth: state.queueDepth,
+              ...(activeAbortRecoveryEligible ? { allowActiveAbort: true } : {}),
               expectedState: state.state,
               stateGeneration: state.generation,
             },
           });
-        } else if (
-          classification &&
-          isActiveAbortRecoveryEligible({
-            classification,
-            activity,
-            ageMs: attentionAgeMs,
-            stuckSessionAbortMs,
-          })
-        ) {
+        } else if (classification && activeAbortRecoveryEligible) {
           requestStuckSessionRecovery({
             recover: opts?.recoverStuckSession ?? recoverStuckSession,
             classification,
