fix(cron): clean up timed out agent runs

steipete · steipete · commit 61d53f98d314 · 2026-04-29T16:54:22.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -23,6 +23,7 @@ Docs: https://docs.openclaw.ai
 
 - Build/Gateway: route restart, shutdown, respawn, diagnostics, command-queue cleanup, and runtime cleanup through one stable gateway lifecycle runtime entry so rebuilt packages do not strand long-running gateways on stale hashed chunks. Carries forward #73964. Thanks @pashpashpash.
 - Memory/wiki: keep broad shared-source and generated related-link blocks from turning every page into a search hit, cap noisy backlinks, support all-term searches such as people-routing queries, and prefer readable page body snippets over generated metadata. Thanks @vincentkoc.
+- Cron/Gateway: abort and bounded-clean up timed-out isolated agent turns before recording the timeout, so stale cron sessions cannot leave Discord or other chat lanes stuck in `processing` after a timeout. Thanks @vincentkoc.
 - Agents/errors: suppress malformed streaming tool-call JSON fragments before they reach chat surfaces while preserving provider request-validation diagnostics. Fixes #59076; keeps #59080 as duplicate coverage. (#59118) Thanks @singleGanghood.
 - CLI/models: restore provider-filtered `models list --all --provider <id>` rows for providers without manifest/static catalog coverage, including Anthropic and Amazon Bedrock, while keeping the compatibility fallback off expensive availability and resolver paths. Thanks @shakkernerd.
 - CLI/models: move the OpenAI listable catalog into the plugin manifest so `models list --all --provider openai` uses the manifest fast path instead of loading provider runtime normalization hooks. Thanks @shakkernerd.
diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md
@@ -51,6 +51,7 @@ Cron is the Gateway's built-in scheduler. It persists jobs, wakes the agent at t
 - Isolated cron runs also guard against stale acknowledgement replies. If the first result is just an interim status update (`on it`, `pulling everything together`, and similar hints) and no descendant subagent run is still responsible for the final answer, OpenClaw re-prompts once for the actual result before delivery.
 - Isolated cron runs prefer structured execution-denial metadata from the embedded run, then fall back to known final summary/output markers such as `SYSTEM_RUN_DENIED` and `INVALID_REQUEST`, so a blocked command is not reported as a green run.
 - Isolated cron runs also treat run-level agent failures as job errors even when no reply payload is produced, so model/provider failures increment error counters and trigger failure notifications instead of clearing the job as successful.
+- When an isolated agent-turn job reaches `timeoutSeconds`, cron aborts the underlying agent run and gives it a short cleanup window. If the run does not drain, Gateway-owned cleanup force-clears that run's session ownership before cron records the timeout, so queued chat work is not left behind a stale processing session.
 
 <a id="maintenance"></a>
 
diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md
@@ -162,6 +162,7 @@ surfaces, while Codex native hooks remain a separate lower-level Codex mechanism
 
 - `agent.wait` default: 30s (just the wait). `timeoutMs` param overrides.
 - Agent runtime: `agents.defaults.timeoutSeconds` default 172800s (48 hours); enforced in `runEmbeddedPiAgent` abort timer.
+- Cron runtime: isolated agent-turn `timeoutSeconds` is owned by cron. The scheduler starts that timer when execution begins, aborts the underlying run at the configured deadline, then runs bounded cleanup before recording the timeout so a stale child session cannot keep the lane stuck.
 - Stuck-session recovery: with diagnostics enabled, `diagnostics.stuckSessionWarnMs` detects long `processing` sessions. Active embedded runs, active reply operations, and active session-lane tasks remain warning-only by default; if diagnostics show no active work for the session, the watchdog releases the affected session lane so queued startup work can drain.
 - Model idle timeout: OpenClaw aborts a model request when no response chunks arrive before the idle window. `models.providers.<id>.timeoutSeconds` extends this idle watchdog for slow local/self-hosted providers; otherwise OpenClaw uses `agents.defaults.timeoutSeconds` when configured, capped at 120s by default. Cron-triggered runs with no explicit model or agent timeout disable the idle watchdog and rely on the cron outer timeout.
 - Provider HTTP request timeout: `models.providers.<id>.timeoutSeconds` applies to that provider's model HTTP fetches, including connect, headers, body, SDK request timeout, total guarded-fetch abort handling, and model stream idle watchdog. Use this for slow local/self-hosted providers such as Ollama before raising the whole agent runtime timeout.
diff --git a/src/agents/pi-embedded-runner.ts b/src/agents/pi-embedded-runner.ts
@@ -21,6 +21,7 @@ export {
   runEmbeddedPiAgent as runEmbeddedAgent,
 } from "./pi-embedded-runner/run.js";
 export {
+  abortAndDrainEmbeddedPiRun,
   abortEmbeddedPiRun,
   abortEmbeddedPiRun as abortEmbeddedAgentRun,
   isEmbeddedPiRunActive,
diff --git a/src/agents/pi-embedded-runner/runs.test.ts b/src/agents/pi-embedded-runner/runs.test.ts
@@ -2,6 +2,7 @@ import { importFreshModule } from "openclaw/plugin-sdk/test-fixtures";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import {
   __testing,
+  abortAndDrainEmbeddedPiRun,
   abortEmbeddedPiRun,
   clearActiveEmbeddedRun,
   consumeEmbeddedRunModelSwitch,
@@ -65,6 +66,32 @@ describe("pi-embedded runner run registry", () => {
     expect(abortB).toHaveBeenCalledTimes(1);
   });
 
+  it("force-clears an aborted run that does not drain", async () => {
+    vi.useFakeTimers();
+    try {
+      const abortRun = vi.fn();
+      setActiveEmbeddedRun("session-stuck", createRunHandle({ abort: abortRun }), "agent:main");
+
+      const resultPromise = abortAndDrainEmbeddedPiRun({
+        sessionId: "session-stuck",
+        sessionKey: "agent:main",
+        settleMs: 100,
+        forceClear: true,
+        reason: "test_timeout",
+      });
+      await vi.advanceTimersByTimeAsync(100);
+      const result = await resultPromise;
+
+      expect(result).toEqual({ aborted: true, drained: false, forceCleared: true });
+      expect(abortRun).toHaveBeenCalledTimes(1);
+      expect(isEmbeddedPiRunHandleActive("session-stuck")).toBe(false);
+      expect(resolveActiveEmbeddedRunHandleSessionId("agent:main")).toBeUndefined();
+    } finally {
+      await vi.runOnlyPendingTimersAsync();
+      vi.useRealTimers();
+    }
+  });
+
   it("waits for active runs to drain", async () => {
     vi.useFakeTimers();
     try {
diff --git a/src/agents/pi-embedded-runner/runs.ts b/src/agents/pi-embedded-runner/runs.ts
@@ -310,6 +310,29 @@ export function waitForEmbeddedPiRunEnd(sessionId: string, timeoutMs = 15_000):
   });
 }
 
+export type AbortAndDrainEmbeddedPiRunResult = {
+  aborted: boolean;
+  drained: boolean;
+  forceCleared: boolean;
+};
+
+export async function abortAndDrainEmbeddedPiRun(params: {
+  sessionId: string;
+  sessionKey?: string;
+  settleMs?: number;
+  forceClear?: boolean;
+  reason?: string;
+}): Promise<AbortAndDrainEmbeddedPiRunResult> {
+  const settleMs = params.settleMs ?? 15_000;
+  const aborted = abortEmbeddedPiRun(params.sessionId);
+  const drained = aborted ? await waitForEmbeddedPiRunEnd(params.sessionId, settleMs) : false;
+  const forceCleared =
+    params.forceClear === true && (!aborted || !drained)
+      ? forceClearEmbeddedPiRun(params.sessionId, params.sessionKey, params.reason)
+      : false;
+  return { aborted, drained, forceCleared };
+}
+
 function notifyEmbeddedRunEnded(sessionId: string) {
   const waiters = EMBEDDED_RUN_WAITERS.get(sessionId);
   if (!waiters || waiters.size === 0) {
diff --git a/src/agents/pi-embedded.runtime.ts b/src/agents/pi-embedded.runtime.ts
@@ -1,4 +1,5 @@
 export {
+  abortAndDrainEmbeddedPiRun,
   abortEmbeddedPiRun,
   isEmbeddedPiRunActive,
   isEmbeddedPiRunStreaming,
diff --git a/src/agents/pi-embedded.ts b/src/agents/pi-embedded.ts
@@ -9,6 +9,7 @@ export type {
   EmbeddedPiRunResult,
 } from "./pi-embedded-runner.js";
 export {
+  abortAndDrainEmbeddedPiRun,
   abortEmbeddedAgentRun,
   abortEmbeddedPiRun,
   compactEmbeddedAgentSession,
diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts
@@ -10,6 +10,7 @@ import { stringifyRouteThreadId } from "../../plugin-sdk/channel-route.js";
 import { normalizeOptionalString } from "../../shared/string-coerce.js";
 import { resolveCronDeliveryPlan, type CronDeliveryPlan } from "../delivery-plan.js";
 import type {
+  CronAgentExecutionStarted,
   CronDeliveryTrace,
   CronDeliveryTraceMessageTarget,
   CronDeliveryTraceTarget,
@@ -424,7 +425,7 @@ type RunCronAgentTurnParams = {
   message: string;
   abortSignal?: AbortSignal;
   signal?: AbortSignal;
-  onExecutionStarted?: () => void;
+  onExecutionStarted?: (info?: CronAgentExecutionStarted) => void;
   sessionKey: string;
   agentId?: string;
   lane?: string;
@@ -1008,7 +1009,7 @@ export async function runCronIsolatedAgentTurn(params: {
   message: string;
   abortSignal?: AbortSignal;
   signal?: AbortSignal;
-  onExecutionStarted?: () => void;
+  onExecutionStarted?: (info?: CronAgentExecutionStarted) => void;
   sessionKey: string;
   agentId?: string;
   lane?: string;
@@ -1026,6 +1027,13 @@ export async function runCronIsolatedAgentTurn(params: {
   if (!prepared.ok) {
     return prepared.result;
   }
+  const notifyExecutionStarted = () =>
+    params.onExecutionStarted?.({
+      jobId: params.job.id,
+      agentId: prepared.context.agentId,
+      sessionId: prepared.context.runSessionId,
+      sessionKey: prepared.context.runSessionKey,
+    });
 
   try {
     const { executeCronRun } = await loadCronExecutorRuntime();
@@ -1054,7 +1062,7 @@ export async function runCronIsolatedAgentTurn(params: {
       commandBody: prepared.context.commandBody,
       persistSessionEntry: prepared.context.persistSessionEntry,
       abortSignal,
-      onExecutionStarted: params.onExecutionStarted,
+      onExecutionStarted: notifyExecutionStarted,
       abortReason,
       isAborted,
       thinkLevel: prepared.context.thinkLevel,
diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts
@@ -7,6 +7,7 @@ import type {
   CronJobCreate,
   CronJobPatch,
   CronMessageChannel,
+  CronAgentExecutionStarted,
   CronRunOutcome,
   CronRunStatus,
   CronRunTelemetry,
@@ -93,7 +94,7 @@ export type CronServiceDeps = {
     job: CronJob;
     message: string;
     abortSignal?: AbortSignal;
-    onExecutionStarted?: () => void;
+    onExecutionStarted?: (info?: CronAgentExecutionStarted) => void;
   }) => Promise<
     {
       summary?: string;
@@ -114,6 +115,11 @@ export type CronServiceDeps = {
     } & CronRunOutcome &
       CronRunTelemetry
   >;
+  cleanupTimedOutAgentRun?: (params: {
+    job: CronJob;
+    timeoutMs: number;
+    execution?: CronAgentExecutionStarted;
+  }) => Promise<void>;
   sendCronFailureAlert?: (params: {
     job: CronJob;
     text: string;
diff --git a/src/cron/service/timer.regression.test.ts b/src/cron/service/timer.regression.test.ts
@@ -13,7 +13,7 @@ import {
 } from "../../../test/helpers/cron/service-regression-fixtures.js";
 import type { HeartbeatRunResult } from "../../infra/heartbeat-wake.js";
 import * as schedule from "../schedule.js";
-import type { CronJob } from "../types.js";
+import type { CronAgentExecutionStarted, CronJob } from "../types.js";
 import { computeJobNextRunAtMs } from "./jobs.js";
 import { createCronServiceState, type CronEvent } from "./state.js";
 import {
@@ -1213,6 +1213,86 @@ describe("cron service timer regressions", () => {
     }
   });
 
+  it("cleans up timed-out isolated runs even when the runner ignores abort", async () => {
+    vi.useFakeTimers();
+    try {
+      const store = timerRegressionFixtures.makeStorePath();
+      const scheduledAt = Date.parse("2026-02-15T14:00:00.000Z");
+      const cronJob = createIsolatedRegressionJob({
+        id: "timeout-cleanup-stuck-run",
+        name: "timeout cleanup stuck run",
+        scheduledAt,
+        schedule: { kind: "at", at: new Date(scheduledAt).toISOString() },
+        payload: { kind: "agentTurn", message: "work", timeoutSeconds: 1 },
+        state: { nextRunAtMs: scheduledAt },
+      });
+      await writeCronJobs(store.storePath, [cronJob]);
+
+      vi.setSystemTime(scheduledAt);
+      let now = scheduledAt;
+      const started = createDeferred<void>();
+      let abortObserved = false;
+      const cleanupTimedOutAgentRun = vi.fn(async () => {});
+      const state = createCronServiceState({
+        cronEnabled: true,
+        storePath: store.storePath,
+        log: noopLogger,
+        nowMs: () => now,
+        enqueueSystemEvent: vi.fn(),
+        requestHeartbeatNow: vi.fn(),
+        cleanupTimedOutAgentRun,
+        runIsolatedAgentJob: vi.fn(
+          async ({
+            abortSignal,
+            onExecutionStarted,
+          }: {
+            abortSignal?: AbortSignal;
+            onExecutionStarted?: (info?: CronAgentExecutionStarted) => void;
+          }) => {
+            onExecutionStarted?.({
+              jobId: "timeout-cleanup-stuck-run",
+              agentId: "main",
+              sessionId: "cron-run-session",
+              sessionKey: "agent:main:cron:timeout-cleanup-stuck-run:run:cron-run-session",
+            });
+            started.resolve();
+            abortSignal?.addEventListener(
+              "abort",
+              () => {
+                abortObserved = true;
+              },
+              { once: true },
+            );
+            return await new Promise<never>(() => {});
+          },
+        ),
+      });
+
+      const timerPromise = onTimer(state);
+      await started.promise;
+      await vi.advanceTimersByTimeAsync(1_100);
+      now += 1_100;
+      await timerPromise;
+
+      expect(abortObserved).toBe(true);
+      expect(cleanupTimedOutAgentRun).toHaveBeenCalledWith({
+        job: expect.objectContaining({ id: "timeout-cleanup-stuck-run" }),
+        timeoutMs: 1_000,
+        execution: {
+          jobId: "timeout-cleanup-stuck-run",
+          agentId: "main",
+          sessionId: "cron-run-session",
+          sessionKey: "agent:main:cron:timeout-cleanup-stuck-run:run:cron-run-session",
+        },
+      });
+      const job = state.store?.jobs.find((entry) => entry.id === "timeout-cleanup-stuck-run");
+      expect(job?.state.lastStatus).toBe("error");
+      expect(job?.state.lastError).toContain("timed out");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("keeps state updates when cron next-run computation throws after a successful run (#30905)", () => {
     const startedAt = Date.parse("2026-03-02T12:00:00.000Z");
     const endedAt = startedAt + 50;
diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts
diff --git a/src/cron/types.ts b/src/cron/types.ts
diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts
diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts
diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.ts

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`export {`
	`2`	`+ abortAndDrainEmbeddedPiRun,`
`2`	`3`	`abortEmbeddedPiRun,`
`3`	`4`	`isEmbeddedPiRunActive,`
`4`	`5`	`isEmbeddedPiRunStreaming,`