fix: reconcile stale cron and chat-backed tasks (#60310) (thanks @lml2468)

src/tasks/task-registry.maintenance.issue-60299.test.ts

@@ -26,11 +26,15 @@ async function loadMaintenanceModule(params: {
   tasks: TaskRecord[];
   sessionStore?: Record<string, unknown>;
   acpEntry?: unknown;
+  activeCronJobIds?: string[];
+  activeRunIds?: string[];
 }) {
   vi.resetModules();
 
   const sessionStore = params.sessionStore ?? {};
   const acpEntry = params.acpEntry;
+  const activeCronJobIds = new Set(params.activeCronJobIds ?? []);
+  const activeRunIds = new Set(params.activeRunIds ?? []);
   const currentTasks = new Map(params.tasks.map((task) => [task.taskId, { ...task }]));
 
   vi.doMock("../acp/runtime/session-meta.js", () => ({
@@ -45,6 +49,15 @@ async function loadMaintenanceModule(params: {
     resolveStorePath: () => "",
   }));
 
+  vi.doMock("../cron/active-jobs.js", () => ({
+    isCronJobActive: (jobId: string) => activeCronJobIds.has(jobId),
+  }));
+
+  vi.doMock("../infra/agent-events.js", () => ({
+    getAgentRunContext: (runId: string) =>
+      activeRunIds.has(runId) ? { sessionKey: "main" } : undefined,
+  }));
+
   vi.doMock("./runtime-internal.js", () => ({
     deleteTaskRecordById: (taskId: string) => currentTasks.delete(taskId),
     ensureTaskRegistryReady: () => {},
@@ -90,22 +103,11 @@ async function loadMaintenanceModule(params: {
 }
 
 describe("task-registry maintenance issue #60299", () => {
-  it("marks cron tasks with no child session key lost after the grace period", async () => {
-    const task = makeStaleTask({
-      runtime: "cron",
-      childSessionKey: undefined,
-    });
-
-    const { mod, currentTasks } = await loadMaintenanceModule({ tasks: [task] });
-
-    expect(await mod.runTaskRegistryMaintenance()).toMatchObject({ reconciled: 1 });
-    expect(currentTasks.get(task.taskId)).toMatchObject({ status: "lost" });
-  });
-
-  it("marks cron tasks lost even if their transient child key still exists in the session store", async () => {
+  it("marks stale cron tasks lost once the runtime no longer tracks the job as active", async () => {
     const childSessionKey = "agent:main:slack:channel:test-channel";
     const task = makeStaleTask({
       runtime: "cron",
+      sourceId: "cron-job-1",
       childSessionKey,
     });
 
@@ -118,10 +120,28 @@ describe("task-registry maintenance issue #60299", () => {
     expect(currentTasks.get(task.taskId)).toMatchObject({ status: "lost" });
   });
 
-  it("treats cli tasks backed only by a persistent chat session as stale", async () => {
+  it("keeps active cron tasks live while the cron runtime still owns the job", async () => {
+    const task = makeStaleTask({
+      runtime: "cron",
+      sourceId: "cron-job-2",
+      childSessionKey: undefined,
+    });
+
+    const { mod, currentTasks } = await loadMaintenanceModule({
+      tasks: [task],
+      activeCronJobIds: ["cron-job-2"],
+    });
+
+    expect(await mod.runTaskRegistryMaintenance()).toMatchObject({ reconciled: 0 });
+    expect(currentTasks.get(task.taskId)).toMatchObject({ status: "running" });
+  });
+
+  it("marks chat-backed cli tasks lost after the owning run context disappears", async () => {
     const channelKey = "agent:main:slack:channel:C1234567890";
     const task = makeStaleTask({
       runtime: "cli",
+      sourceId: "run-chat-cli-stale",
+      runId: "run-chat-cli-stale",
       ownerKey: "agent:main:main",
       requesterSessionKey: channelKey,
       childSessionKey: channelKey,
@@ -136,18 +156,21 @@ describe("task-registry maintenance issue #60299", () => {
     expect(currentTasks.get(task.taskId)).toMatchObject({ status: "lost" });
   });
 
-  it("keeps subagent tasks live while their child session still exists", async () => {
-    const childKey = "agent:main:subagent:abc123";
+  it("keeps chat-backed cli tasks live while the owning run context is still active", async () => {
+    const channelKey = "agent:main:slack:channel:C1234567890";
     const task = makeStaleTask({
-      runtime: "subagent",
+      runtime: "cli",
+      sourceId: "run-chat-cli-live",
+      runId: "run-chat-cli-live",
       ownerKey: "agent:main:main",
-      requesterSessionKey: "agent:main:main",
-      childSessionKey: childKey,
+      requesterSessionKey: channelKey,
+      childSessionKey: channelKey,
     });
 
     const { mod, currentTasks } = await loadMaintenanceModule({
       tasks: [task],
-      sessionStore: { [childKey]: { updatedAt: Date.now() } },
+      sessionStore: { [channelKey]: { updatedAt: Date.now() } },
+      activeRunIds: ["run-chat-cli-live"],
     });
 
     expect(await mod.runTaskRegistryMaintenance()).toMatchObject({ reconciled: 0 });

src/tasks/task-registry.maintenance.ts

@@ -1,5 +1,7 @@
 import { readAcpSessionEntry } from "../acp/runtime/session-meta.js";
 import { loadSessionStore, resolveStorePath } from "../config/sessions.js";
+import { isCronJobActive } from "../cron/active-jobs.js";
+import { getAgentRunContext } from "../infra/agent-events.js";
 import { parseAgentSessionKey } from "../routing/session-key.js";
 import { deriveSessionChatType } from "../sessions/session-chat-type.js";
 import {
@@ -64,16 +66,25 @@ function hasLostGraceExpired(task: TaskRecord, now: number): boolean {
   return now - referenceAt >= TASK_RECONCILE_GRACE_MS;
 }
 
-/**
- * Returns false if the task's runtime is cron, since cron tasks do not maintain
- * a persistent child session after the job exits.
- *
- * For cli tasks, long-lived channel/group/direct session-store entries do not
- * imply task liveness, so only agent-scoped non-chat child sessions count.
- */
+function hasActiveCliRun(task: TaskRecord): boolean {
+  const candidateRunIds = [task.sourceId, task.runId];
+  for (const candidate of candidateRunIds) {
+    const runId = candidate?.trim();
+    if (runId && getAgentRunContext(runId)) {
+      return true;
+    }
+  }
+  return false;
+}
+
 function hasBackingSession(task: TaskRecord): boolean {
   if (task.runtime === "cron") {
-    return false;
+    const jobId = task.sourceId?.trim();
+    return jobId ? isCronJobActive(jobId) : false;
+  }
+
+  if (task.runtime === "cli" && hasActiveCliRun(task)) {
+    return true;
   }
 
   const childSessionKey = task.childSessionKey?.trim();
@@ -89,17 +100,12 @@ function hasBackingSession(task: TaskRecord): boolean {
     }
     return Boolean(acpEntry.entry);
   }
-  if (task.runtime === "subagent") {
-    const agentId = parseAgentSessionKey(childSessionKey)?.agentId;
-    const storePath = resolveStorePath(undefined, { agentId });
-    const store = loadSessionStore(storePath);
-    return Boolean(findSessionEntryByKey(store, childSessionKey));
-  }
-
-  if (task.runtime === "cli") {
-    const chatType = deriveSessionChatType(childSessionKey);
-    if (chatType === "channel" || chatType === "group" || chatType === "direct") {
-      return false;
+  if (task.runtime === "subagent" || task.runtime === "cli") {
+    if (task.runtime === "cli") {
+      const chatType = deriveSessionChatType(childSessionKey);
+      if (chatType === "channel" || chatType === "group" || chatType === "direct") {
+        return false;
+      }
     }
     const agentId = parseAgentSessionKey(childSessionKey)?.agentId;
     const storePath = resolveStorePath(undefined, { agentId });

src/tasks/task-registry.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { startAcpSpawnParentStreamRelay } from "../agents/acp-spawn-parent-stream.js";
+import { resetCronActiveJobsForTests } from "../cron/active-jobs.js";
 import {
   emitAgentEvent,
   registerAgentRunContext,
@@ -226,6 +227,7 @@ describe("task-registry", () => {
     resetSystemEventsForTest();
     resetHeartbeatWakeStateForTests();
     resetAgentRunContextForTest();
+    resetCronActiveJobsForTests();
     resetTaskRegistryDeliveryRuntimeForTests();
     resetTaskRegistryForTests({ persist: false });
     resetTaskFlowRegistryForTests({ persist: false });