fix(ui): preserve active sessions on recoverable errors

scripts/tui-e2e-permission.tsx

@@ -91,6 +91,7 @@ class MockGateway implements Gateway {
   describeServer = stub({ mode: "in_process" as const });
   cronCreate = stub({ taskId: "c", task: {} as any, created: true }) as unknown as Gateway["cronCreate"];
   cronList = stub({ tasks: [] }) as Gateway["cronList"];
+  cronUpdate = stub({ updated: false, reason: "not_found" }) as Gateway["cronUpdate"];
   cronDelete = stub({ deleted: true }) as Gateway["cronDelete"];
   cronStop = stub({ stopped: true }) as Gateway["cronStop"];
   cronRunNow = stub({ triggered: true }) as unknown as Gateway["cronRunNow"];

src/cli/pilotdeck.ts

@@ -88,6 +88,9 @@ async function main(argv = process.argv.slice(2)): Promise<void> {
         sessionOverrides,
         logger: cronLogger,
         telemetry,
+        onTurnEvent: (sessionKey, channelKey, event) => {
+          deferredBroadcast?.("always-on:turn-event", { sessionKey, channelKey, event });
+        },
         onResultDelivery: (delivery) => {
           void serverRef?.deliverCronResult(delivery)
             .then((delivered) => {
@@ -783,6 +786,9 @@ function createFallbackGateway(): Gateway {
     cronList: async () => {
       throw new Error("Cron runtime is not configured.");
     },
+    cronUpdate: async () => {
+      throw new Error("Cron runtime is not configured.");
+    },
     cronDelete: async () => {
       throw new Error("Cron runtime is not configured.");
     },

src/cron/index.ts

@@ -33,4 +33,6 @@ export type {
   CronStopResult,
   CronTask,
   CronTaskStatus,
+  CronUpdateInput,
+  CronUpdateResult,
 } from "./protocol/types.js";

src/cron/protocol/types.ts

@@ -38,6 +38,7 @@ export type CronTask = {
   updatedAt: string;
   nextRunAt?: string;
   lastRunId?: string;
+  revision?: number;
   scheduleComputationVersion?: 2;
   originSessionKey?: string;
   originChannelKey?: GatewayChannelKey;
@@ -92,6 +93,25 @@ export type CronCreateResult = {
   task: CronTask;
 };
 
+export type CronUpdateInput = {
+  taskId: string;
+  projectKey: string;
+  expectedRevision: number;
+  message: string;
+  schedule: CronTaskSchedule;
+  timezone?: string;
+};
+
+export type CronUpdateResult =
+  | {
+      updated: true;
+      task: CronTask;
+    }
+  | {
+      updated: false;
+      reason: "not_found" | "running" | "conflict";
+    };
+
 export type CronListInput = {
   projectKey?: string;
   includeHistory?: boolean;

src/cron/runtime/CronFire.ts

@@ -22,6 +22,8 @@ export type CronPhaseEventCallback = (event: {
   error?: { code: string; message: string };
 }) => void;
 
+export type CronTurnEventHandler = (sessionKey: string, channelKey: string, event: GatewayEvent) => void;
+
 export type CronFireDependencies = {
   gateway: Gateway;
   store: CronTaskStore;
@@ -33,6 +35,7 @@ export type CronFireDependencies = {
   defaultTimezone: string;
   releaseTaskSession: (task: CronTask) => Promise<void>;
   onResultDelivery?: CronResultDeliveryHandler;
+  onTurnEvent?: CronTurnEventHandler;
   logger?: {
     warn: (message: string, data?: Record<string, unknown>) => void;
   };
@@ -42,34 +45,44 @@ export type CronFireDependencies = {
 export class CronFire {
   constructor(private readonly deps: CronFireDependencies) {}
 
-  async runTask(task: CronTask, runId: string): Promise<void> {
+  async runTask(taskSnapshot: CronTask, runId: string): Promise<void> {
     const startedAt = this.deps.now();
     const activeRun: CronActiveRun = {
       runId,
-      taskId: task.taskId,
-      sessionKey: task.sessionKey,
-      scheduleType: task.schedule.type,
+      taskId: taskSnapshot.taskId,
+      sessionKey: taskSnapshot.sessionKey,
+      scheduleType: taskSnapshot.schedule.type,
       stopRequested: false,
     };
     this.deps.registerActiveRun(activeRun);
 
+    let task = taskSnapshot;
     let outcome: CronRunOutcome = "completed";
     let error: CronRunRecord["error"];
     let forcedFailure = false;
     let abortRequested = false;
     let startedRun = false;
     let assistantText = "";
     try {
-      const started = await this.deps.store.replaceTask({
-        ...task,
-        status: "running",
-        lastRunId: runId,
-        updatedAt: startedAt.toISOString(),
+      let claimed = false;
+      const currentTask = await this.deps.store.updateTask(taskSnapshot.taskId, (current) => {
+        if (!matchesScheduledSnapshot(current, taskSnapshot)) {
+          return current;
+        }
+        claimed = true;
+        return {
+          ...current,
+          status: "running",
+          lastRunId: runId,
+          revision: (current.revision ?? 0) + 1,
+          updatedAt: startedAt.toISOString(),
+        };
       });
-      if (!started) {
+      if (!claimed || !currentTask) {
         outcome = "aborted";
         return;
       }
+      task = currentTask;
       startedRun = true;
       this.deps.onPhaseEvent?.({
         phase: "cron_started",
@@ -89,6 +102,7 @@ export class CronFire {
         timeoutMs: this.deps.runTimeoutMs,
       })) {
         await this.deps.store.appendRunEvent(runId, event);
+        this.forwardTurnEvent(task, event);
         if (event.type === "assistant_text_delta") {
           assistantText += event.text;
         }
@@ -177,7 +191,7 @@ export class CronFire {
           error: deliveryError instanceof Error ? deliveryError.message : String(deliveryError),
         });
       });
-      await this.updateTaskAfterRun(task, finishedAt, outcome).catch((updateError: unknown) => {
+      await this.updateTaskAfterRun(task, runId, finishedAt, outcome).catch((updateError: unknown) => {
         this.deps.logger?.warn("cron task post-run update failed", {
           taskId: task.taskId,
           runId,
@@ -187,6 +201,18 @@ export class CronFire {
     }
   }
 
+  private forwardTurnEvent(task: CronTask, event: GatewayEvent): void {
+    try {
+      this.deps.onTurnEvent?.(task.sessionKey, task.channelKey, event);
+    } catch (error) {
+      this.deps.logger?.warn("cron turn event delivery failed", {
+        taskId: task.taskId,
+        runId: task.lastRunId,
+        error: error instanceof Error ? error.message : String(error),
+      });
+    }
+  }
+
   private async deliverResult(
     task: CronTask,
     runId: string,
@@ -212,11 +238,17 @@ export class CronFire {
     });
   }
 
-  private async updateTaskAfterRun(task: CronTask, finishedAt: Date, outcome: CronRunOutcome): Promise<void> {
+  private async updateTaskAfterRun(task: CronTask, runId: string, finishedAt: Date, outcome: CronRunOutcome): Promise<void> {
     if (task.schedule.type === "once") {
-      try {
-        await this.deps.store.deleteTask(task.taskId);
-      } finally {
+      let deleted = false;
+      await this.deps.store.updateTask(task.taskId, (current) => {
+        if (!matchesRunningTask(current, task, runId)) {
+          return current;
+        }
+        deleted = true;
+        return undefined;
+      });
+      if (deleted) {
         await this.deps.releaseTaskSession(task);
       }
       return;
@@ -228,15 +260,34 @@ export class CronFire {
     );
     const schedule = { ...task.schedule, timezone };
     const nextRunAt = computeNextRunAt(schedule, finishedAt, timezone)?.toISOString();
-    await this.deps.store.updateTask(task.taskId, (current) => ({
-      ...current,
-      schedule,
-      timezone,
-      status: "scheduled",
-      nextRunAt,
-      scheduleComputationVersion: 2,
-      updatedAt: finishedAt.toISOString(),
-    }));
+    await this.deps.store.updateTask(task.taskId, (current) => {
+      if (!matchesRunningTask(current, task, runId)) {
+        return current;
+      }
+      return {
+        ...current,
+        schedule,
+        timezone,
+        status: "scheduled",
+        nextRunAt,
+        revision: (current.revision ?? 0) + 1,
+        scheduleComputationVersion: 2,
+        updatedAt: finishedAt.toISOString(),
+      };
+    });
     void outcome;
   }
 }
+
+function matchesScheduledSnapshot(current: CronTask, snapshot: CronTask): boolean {
+  return current.status === "scheduled"
+    && (current.revision ?? 0) === (snapshot.revision ?? 0)
+    && current.nextRunAt === snapshot.nextRunAt
+    && current.lastRunId === snapshot.lastRunId;
+}
+
+function matchesRunningTask(current: CronTask, claimedTask: CronTask, runId: string): boolean {
+  return current.status === "running"
+    && current.lastRunId === runId
+    && (current.revision ?? 0) === (claimedTask.revision ?? 0);
+}

src/cron/runtime/CronManager.ts

@@ -20,6 +20,8 @@ import type {
   CronStopInput,
   CronStopResult,
   CronTask,
+  CronUpdateInput,
+  CronUpdateResult,
 } from "../protocol/types.js";
 import { resolveCronPaths } from "../storage/CronPaths.js";
 import { createCronCreateTool } from "../tool/CronCreateTool.js";
@@ -28,6 +30,7 @@ import { createCronListTool } from "../tool/CronListTool.js";
 import { createCronStopTool } from "../tool/CronStopTool.js";
 import { migrateCronStores } from "../storage/CronStoreMigration.js";
 import { CronRuntime, type CronRuntimeLogger } from "./CronRuntime.js";
+import type { CronTurnEventHandler } from "./CronFire.js";
 
 export type CreateCronManagerOptions = {
   config: CronConfig;
@@ -38,6 +41,7 @@ export type CreateCronManagerOptions = {
   logger?: CronRuntimeLogger;
   telemetry?: TelemetryClient;
   onResultDelivery?: CronResultDeliveryHandler;
+  onTurnEvent?: CronTurnEventHandler;
 };
 
 export class CronManager {
@@ -131,6 +135,12 @@ export class CronManager {
     return result;
   }
 
+  async updateTask(input: CronUpdateInput): Promise<CronUpdateResult> {
+    const runtime = await this.resolveTaskRuntime(input.taskId, input.projectKey);
+    if (!runtime) return { updated: false, reason: "not_found" };
+    return runtime.updateTask(input);
+  }
+
   async deleteTask(input: CronDeleteInput): Promise<CronDeleteResult> {
     const runtime = await this.resolveTaskRuntime(input.taskId, input.projectKey);
     if (!runtime) return { deleted: false };
@@ -180,6 +190,7 @@ export class CronManager {
       logger: this.options.logger,
       telemetry: this.options.telemetry,
       onResultDelivery: this.options.onResultDelivery,
+      onTurnEvent: this.options.onTurnEvent,
       activeRunCount: () => this.activeRunCount(),
       skipToolCreation: true,
     });