fix(server): scheduled runs record how their turn ended

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -272,6 +272,7 @@ function scheduledTaskSummary(task: ScheduledTask, mayRun: boolean): Orchestrato
     schedule: task.schedule,
     nextRunAt: task.nextRunAt,
     lastRunStatus: task.lastRunStatus,
+    ...(task.lastRunError === null ? {} : { lastRunError: task.lastRunError }),
     ...(task.webhook?.url == null || !mayRun || task.blockedByPolicy !== undefined
       ? {}
       : { webhookUrl: task.webhook.url }),

apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts

@@ -323,7 +323,7 @@ it.effect.each(deliveries)(
                 }),
               );
               const ran = yield* service.runNow({ id: task.id });
-              assert.equal(ran.task.lastRunStatus, "succeeded");
+              assert.equal(ran.task.lastRunStatus, "running");
             }).pipe(
               Effect.provide(
                 ScheduledTaskService.layer.pipe(

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -497,6 +497,10 @@ export interface OrchestratorV2Shape {
     readonly eventType?: OrchestrationV2DomainEvent["type"];
   }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestratorV2Error>;
   readonly streamDomainEvents: Stream.Stream<OrchestrationV2DomainEvent, OrchestratorV2Error>;
+  readonly openDomainEventStream: Effect.Effect<
+    Stream.Stream<OrchestrationV2DomainEvent, OrchestratorV2Error>,
+    OrchestratorV2Error
+  >;
 }
 
 export class OrchestratorV2 extends Context.Service<OrchestratorV2, OrchestratorV2Shape>()(
@@ -14484,6 +14488,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   const shellProjectionError = (cause: unknown) =>
     new OrchestratorProjectionError({ threadId: ThreadId.make("thread:shell"), cause });
 
+  const openDomainEventStream = Effect.fn("OrchestratorV2.openDomainEventStream")(
+    function* () {
+      const latest = yield* eventSink.latestSequence();
+      return eventSink.stream({ afterSequence: latest }).pipe(
+        Stream.map((stored) => stored.event),
+        Stream.mapError((cause) => new OrchestratorDomainEventStreamError({ cause })),
+      );
+    },
+    Effect.mapError((cause) => new OrchestratorDomainEventStreamError({ cause })),
+  );
+
   return OrchestratorV2.of({
     resumeQueuedRuns,
     recoverDelegatedTasks,
@@ -14689,20 +14704,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             }),
         ),
       ),
-    // Start at the latest sequence; awareness needs no historical replay.
-    streamDomainEvents: Stream.unwrap(
-      eventSink
-        .latestSequence()
-        .pipe(Effect.map((latest) => eventSink.stream({ afterSequence: latest }))),
-    ).pipe(
-      Stream.map((stored) => stored.event),
-      Stream.mapError(
-        (cause) =>
-          new OrchestratorDomainEventStreamError({
-            cause,
-          }),
-      ),
-    ),
+    streamDomainEvents: Stream.unwrap(openDomainEventStream()),
+    openDomainEventStream: openDomainEventStream(),
   });
 });
 

apps/server/src/orchestration-v2/ThreadLaunchService.test.ts

@@ -371,7 +371,7 @@ it.effect.each(
         creationSource: createdBy === "agent" ? "mcp" : "web",
       });
       const result = yield* tasks.runNow({ id: task.id });
-      assert.equal(result.task.lastRunStatus, "succeeded");
+      assert.equal(result.task.lastRunStatus, "running");
       const projectThreads = yield* threads.listProjectThreads({
         projectId,
         includeSubagents: false,
@@ -488,7 +488,7 @@ it.effect("runs a thread-bound automation on the provider the thread uses now",
     });
 
     const result = yield* tasks.runNow({ id: task.id });
-    assert.equal(result.task.lastRunStatus, "succeeded");
+    assert.equal(result.task.lastRunStatus, "running");
     const projection = yield* threads.getThreadProjection(launched.threadId);
     const message = projection.messages.find((item) => item.scheduledTaskId === task.id);
     const run = projection.runs.find((item) => item.id === message?.runId);
@@ -549,6 +549,75 @@ it.effect("retains automation and sender attribution while a message waits in th
   }).pipe(Effect.provide(harness.layer));
 });
 
+it.effect("an automation queued behind a running turn records how its own turn ends", () => {
+  const harness = makeHarness({ runSetup: () => Effect.never });
+  const layerScheduledTasks = ScheduledTasks.layer.pipe(
+    Layer.provide(
+      Layer.mergeAll(
+        ManagedPolicy.layerTest(),
+        harness.layer,
+        NodeCrypto.layer,
+        Scheduler.layer,
+        Layer.mock(SecretRequests.SecretRequests)({}),
+      ),
+    ),
+  );
+  return Effect.gen(function* () {
+    const tasks = yield* ScheduledTasks.ScheduledTaskService;
+    const launches = yield* ThreadLaunch.ThreadLaunchService;
+    const threads = yield* ThreadManagement.ThreadManagementService;
+    const launched = yield* launches.launch(
+      launchInput({
+        command: "command:automation:outcome",
+        thread: "thread:automation:outcome",
+        message: "First message",
+        workspace: { type: "worktree", baseRef: "main" },
+      }),
+    );
+    const { task } = yield* tasks.upsert({
+      id: ScheduledTaskId.make("scheduled-task:outcome"),
+      title: "Daily audit",
+      prompt: "Audit performance and crashes.",
+      enabled: false,
+      schedule: { type: "interval", everyMs: 60_000 },
+      projectId,
+      threadId: launched.threadId,
+      workspaceStrategy: { type: "root" },
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      createdBy: "user",
+      creationSource: "web",
+    });
+    const ran = yield* tasks.runNow({ id: task.id });
+    assert.equal(ran.task.lastRunStatus, "running");
+    const projection = yield* threads.getThreadProjection(launched.threadId);
+    const queued = projection.runs.find((run) => run.status === "queued");
+    assert.isDefined(queued);
+    const events = yield* threads.openDomainEventStream;
+
+    yield* threads.dispatch({
+      type: "queued-run.cancel",
+      commandId: CommandId.make("command:automation:outcome:cancel"),
+      threadId: launched.threadId,
+      runId: queued!.id,
+    });
+    const cancelled = yield* events.pipe(
+      Stream.filter((event) => event.type === "run.updated" && event.payload.id === queued!.id),
+      Stream.runHead,
+    );
+    assert.isTrue(Option.isSome(cancelled));
+    yield* tasks.subscribeList().pipe(
+      Stream.filter(({ tasks: listed }) => listed[0]?.lastRunStatus === "failed"),
+      Stream.take(1),
+      Stream.runDrain,
+    );
+    const [settled] = (yield* tasks.list()).tasks;
+    assert.equal(settled?.lastRunError, "The run was cancelled.");
+    assert.equal(settled?.runCount, 1);
+  }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerScheduledTasks)));
+});
+
 it.effect("provisions independent launches concurrently instead of behind a global semaphore", () =>
   Effect.gen(function* () {
     const setupCount = yield* Ref.make(0);
@@ -1051,7 +1120,7 @@ it.effect("runs a Scratch scheduled task saved with a worktree in its own folder
       WHERE task_id = ${task.id}
     `;
     const ran = yield* tasks.runNow({ id: task.id });
-    assert.equal(ran.task.lastRunStatus, "succeeded");
+    assert.equal(ran.task.lastRunStatus, "running");
     const [thread] = yield* threads.listProjectThreads({ projectId, includeSubagents: false });
     assert.isDefined(thread);
     const projection = yield* preparationSettled(thread!.id);

apps/server/src/orchestration-v2/ThreadManagementService.ts

@@ -497,6 +497,7 @@ export interface ThreadManagementServiceShape {
   readonly streamStoredEvents: Orchestrator.OrchestratorV2["Service"]["streamStoredEvents"];
   readonly streamStoredEventsFrom: Orchestrator.OrchestratorV2["Service"]["streamStoredEventsFrom"];
   readonly streamDomainEvents: Orchestrator.OrchestratorV2["Service"]["streamDomainEvents"];
+  readonly openDomainEventStream: Orchestrator.OrchestratorV2["Service"]["openDomainEventStream"];
 }
 
 export class ThreadManagementService extends Context.Service<
@@ -1632,6 +1633,7 @@ const make = Effect.gen(function* () {
     streamStoredEvents: orchestrator.streamStoredEvents,
     streamStoredEventsFrom: orchestrator.streamStoredEventsFrom,
     streamDomainEvents: orchestrator.streamDomainEvents,
+    openDomainEventStream: orchestrator.openDomainEventStream,
   });
 });
 

apps/server/src/relay/AgentAwarenessRelay.test.ts

@@ -265,6 +265,7 @@ const makeTestRelay = Effect.fnUntraced(function* (
     streamStoredEvents: Stream.empty,
     streamStoredEventsFrom: () => Stream.empty,
     streamDomainEvents: options.domainEvents ?? Stream.empty,
+    openDomainEventStream: Effect.succeed(options.domainEvents ?? Stream.empty),
   });
   const publications: Array<{
     readonly url: string;