fix(server): auto-settle no longer cuts off an agent whose awaited background task just finished

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

@@ -350,6 +350,7 @@ it.effect(
             thread: {
               ...expected,
               latestUserAuthoredMessageAt: candidate.latestUserAuthoredMessageAt,
+              heldWorkEndedAt: candidate.heldWorkEndedAt,
             },
           }),
         );
@@ -365,6 +366,66 @@ it.effect(
     }).pipe(Effect.provide(layerSql)),
 );
 
+it.effect("sql: records when a provider thread's roster stops holding work", () =>
+  Effect.gen(function* () {
+    const store = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* createThread("held-work");
+    yield* createRun(threadId);
+    const roster = (
+      name: string,
+      at: DateTime.Utc,
+      pendingBackgroundTasks: ReadonlyArray<{
+        readonly taskId: string;
+        readonly kind: "command";
+        readonly awaited?: true;
+      }>,
+    ) =>
+      store.apply({
+        id: EventId.make(`event:settlement:held-work:${name}`),
+        type: "provider-thread.updated",
+        threadId,
+        occurredAt: at,
+        payload: {
+          id: ProviderThreadId.make("provider-thread:settlement:held-work"),
+          appThreadId: threadId,
+          ownerNodeId: null,
+          driver: ProviderDriverKind.make("claudeAgent"),
+          providerInstanceId,
+          providerSessionId: null,
+          nativeThreadRef: null,
+          nativeConversationHeadRef: null,
+          status: "idle",
+          firstRunOrdinal: 1,
+          lastRunOrdinal: 1,
+          handoffIds: [],
+          forkedFrom: null,
+          createdAt: old,
+          updatedAt: at,
+          pendingBackgroundTasks,
+        },
+      });
+    const heldWorkEndedAt = Effect.map(store.getSettlementCandidates(threadId), ([candidate]) =>
+      candidate?.heldWorkEndedAt == null ? null : DateTime.formatIso(candidate.heldWorkEndedAt),
+    );
+    const ended = DateTime.subtract(now, { minutes: 1 });
+
+    yield* roster("created", DateTime.subtract(now, { minutes: 5 }), []);
+    assert.isNull(yield* heldWorkEndedAt);
+    // A command nobody waits on, such as a dev server, never held the thread.
+    yield* roster("dev-server", DateTime.subtract(now, { minutes: 4 }), [
+      { taskId: "dev", kind: "command" },
+    ]);
+    yield* roster("dev-server-ended", DateTime.subtract(now, { minutes: 3 }), []);
+    assert.isNull(yield* heldWorkEndedAt);
+    yield* roster("awaited", DateTime.subtract(now, { minutes: 2 }), [
+      { taskId: "tests", kind: "command", awaited: true },
+    ]);
+    yield* roster("awaited-ended", ended, []);
+    yield* roster("idle-again", now, []);
+    assert.equal(yield* heldWorkEndedAt, DateTime.formatIso(ended));
+  }).pipe(Effect.provide(layerSql)),
+);
+
 it.effect("sql: the user-authored message time ignores agent notifications", () =>
   Effect.gen(function* () {
     const store = yield* ProjectionStore.ProjectionStoreV2;

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

@@ -82,6 +82,7 @@ import {
   isOrchestrationV2SupersededInterrupt,
 } from "@t3tools/shared/orchestrationV2Timeline";
 import {
+  backgroundWorkHoldsCompletion,
   deriveBlockedOnThreadWait,
   derivePendingBackgroundWork,
 } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
@@ -224,6 +225,8 @@ export type ProjectionSettlementCandidate = Pick<
 > &
   Pick<OrchestrationV2AppThread, "lastSnoozeWakeAt"> & {
     readonly latestUserAuthoredMessageAt: DateTime.Utc | null;
+    /** When work last stopped holding one of the thread's provider threads. */
+    readonly heldWorkEndedAt: DateTime.Utc | null;
   };
 
 const ProjectionCheckpointContext = Schema.Struct({
@@ -1062,7 +1065,7 @@ type SettlementThreadRow = Pick<
   | "latest_run_completed_at"
   | "latest_user_message_at"
   | "latest_user_authored_message_at"
->;
+> & { readonly held_work_ended_at: string | null };
 
 type ShellRunItemCountRow = {
   readonly thread_id: string;
@@ -2376,6 +2379,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             case "provider-thread.updated": {
               const payloadJson = yield* encodeProviderThreadPayload(event.payload);
               const payload = parseEncodedPayload(payloadJson);
+              const occurredAt = yield* encodePayloadTimestamp(event.occurredAt);
+              // Settlement waits for the turn that reports work which just stopped holding the
+              // thread; recording that here commits it with the roster that dropped the work.
               yield* sql`
               INSERT INTO orchestration_v2_projection_provider_threads (
                 provider_thread_id,
@@ -2389,7 +2395,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 first_run_ordinal,
                 last_run_ordinal,
                 updated_at,
-                payload_json
+                payload_json,
+                holds_completion
               )
               VALUES (
                 ${event.payload.id},
@@ -2403,7 +2410,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 ${event.payload.firstRunOrdinal},
                 ${event.payload.lastRunOrdinal},
                 ${stringField(payload, "updatedAt")},
-                ${payloadJson}
+                ${payloadJson},
+                ${booleanInt(backgroundWorkHoldsCompletion(event.payload.pendingBackgroundTasks ?? []))}
               )
               ON CONFLICT(provider_thread_id)
               DO UPDATE SET
@@ -2417,7 +2425,14 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 first_run_ordinal = excluded.first_run_ordinal,
                 last_run_ordinal = excluded.last_run_ordinal,
                 updated_at = excluded.updated_at,
-                payload_json = excluded.payload_json
+                payload_json = excluded.payload_json,
+                held_work_ended_at = CASE
+                  WHEN orchestration_v2_projection_provider_threads.holds_completion = 1
+                    AND excluded.holds_completion = 0
+                  THEN ${occurredAt}
+                  ELSE orchestration_v2_projection_provider_threads.held_work_ended_at
+                END,
+                holds_completion = excluded.holds_completion
             `;
               if (
                 event.payload.appThreadId !== null &&
@@ -6596,7 +6611,12 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                   AND NOT ${isIdleCompactionMessage("message")}
                 ORDER BY message.updated_at DESC, message.message_id DESC
                 LIMIT 1
-              ) AS latest_user_authored_message_at
+              ) AS latest_user_authored_message_at,
+              (
+                SELECT MAX(provider_thread.held_work_ended_at)
+                FROM orchestration_v2_projection_provider_threads provider_thread
+                WHERE provider_thread.thread_id = t.thread_id
+              ) AS held_work_ended_at
             FROM orchestration_v2_projection_threads t
             LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = (
               SELECT latest.run_id FROM orchestration_v2_projection_runs latest
@@ -6679,6 +6699,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                     row.latest_user_authored_message_at === null
                       ? null
                       : DateTime.makeUnsafe(row.latest_user_authored_message_at),
+                  heldWorkEndedAt:
+                    row.held_work_ended_at === null
+                      ? null
+                      : DateTime.makeUnsafe(row.held_work_ended_at),
                   activityRunStatus: null,
                   activityRunStartedAt: null,
                   pendingRuntimeRequest: null,

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

@@ -53,7 +53,10 @@ function at(offsetMs: number): DateTime.Utc {
 }
 
 type SettlementShell = OrchestrationV2ThreadShell &
-  Pick<ProjectionStore.ProjectionSettlementCandidate, "latestUserAuthoredMessageAt">;
+  Pick<
+    ProjectionStore.ProjectionSettlementCandidate,
+    "latestUserAuthoredMessageAt" | "heldWorkEndedAt"
+  >;
 
 function shell(overrides: Partial<SettlementShell> = {}): SettlementShell {
   return {
@@ -93,6 +96,7 @@ function shell(overrides: Partial<SettlementShell> = {}): SettlementShell {
     latestRunCompletedAt: null,
     latestUserMessageAt: null,
     latestUserAuthoredMessageAt: overrides.latestUserMessageAt ?? null,
+    heldWorkEndedAt: null,
     createdAt: at(-30 * DAY_MS),
     updatedAt: at(-10 * DAY_MS),
     archivedAt: null,
@@ -1239,6 +1243,119 @@ describe("ThreadSettlementServiceV2 worker", () => {
     ),
   );
 
+  it.effect("full and merge sweeps wait for the turn that reports awaited work", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        const dispatched = yield* Queue.unbounded<ThreadId>();
+        // The roster that dropped the work committed, but its event has not reached this worker.
+        const held = makeThread("awaited-work", {
+          pullRequests: [
+            {
+              ...persistedLink("open", DateTime.makeUnsafe("2026-08-20T00:00:00.000Z")),
+              snapshot: {
+                state: "merged",
+                title: "Pull request",
+                headBranch: "feature",
+                baseBranch: "main",
+                isDraft: false,
+                updatedAt: NOW,
+                syncedAt: NOW,
+                mergedAt: NOW,
+              },
+            },
+          ],
+          heldWorkEndedAt: DateTime.makeUnsafe(NOW),
+        });
+        // Only the merge event settles this one, so its dispatch marks that sweep's decisions.
+        const mergedInApp = makeThread("merged-in-app", {
+          latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"),
+          linkedPullRequest: {
+            projectId: PROJECT_ID,
+            repository: "owner/repository",
+            number: 42,
+            url: "https://example.test/owner/repository/pull/42",
+          },
+        });
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([held, mergedInApp]),
+          onDispatch: (command) => Queue.offer(dispatched, command.threadId).pipe(Effect.asVoid),
+        });
+
+        yield* Effect.gen(function* () {
+          const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2;
+          yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
+          assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
+
+          // The report turn was requested but waits behind another run.
+          const reportAt = DateTime.makeUnsafe(Date.parse(NOW) + 1_000);
+          const report = {
+            ...held,
+            latestRunId: RunId.make("run:report"),
+            latestUserMessageAt: reportAt,
+            latestRunRequestedAt: reportAt,
+          };
+          yield* Ref.set(
+            fixture.snapshots,
+            makeSnapshot([{ ...report, status: "queued" }, mergedInApp]),
+          );
+          yield* fixture.publishMerge;
+          assert.strictEqual(yield* Queue.take(dispatched), mergedInApp.id);
+          assert.deepStrictEqual(
+            (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
+            [mergedInApp.id],
+          );
+
+          yield* Ref.set(
+            fixture.snapshots,
+            makeSnapshot([
+              {
+                ...report,
+                status: "completed",
+                latestRunStartedAt: reportAt,
+                latestRunCompletedAt: DateTime.makeUnsafe(Date.parse(NOW) + 20_000),
+              },
+            ]),
+          );
+          // The next full sweep is within the hold's grace, so only the report turn releases it.
+          yield* TestClock.adjust("1 minute");
+          assert.strictEqual(yield* Queue.take(dispatched), held.id);
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
+
+  it.effect("settles an idle thread whose awaited work ended once no report turn came", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        const settled = yield* Deferred.make<void>();
+        // Ended 90s ago, so the next full sweep, a minute later, falls past the hold's grace.
+        const thread = makeThread("awaited-work-idle", {
+          heldWorkEndedAt: DateTime.makeUnsafe(Date.parse(NOW) - 90_000),
+        });
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([thread]),
+          settings: { ...DEFAULT_SERVER_SETTINGS, sidebarAutoSettleAfterDays: 3 },
+          onDispatch: () => Deferred.succeed(settled, undefined),
+        });
+
+        yield* Effect.gen(function* () {
+          const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2;
+          yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
+          assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
+
+          yield* TestClock.adjust("1 minute");
+          yield* Deferred.await(settled);
+          assert.deepStrictEqual(
+            (yield* Ref.get(fixture.commands)).map((command) => command.threadId),
+            [thread.id],
+          );
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
+
   it.effect("settles an unlinked branch thread its agent merged from a shell, past the cache", () =>
     Effect.scoped(
       Effect.gen(function* () {

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

@@ -121,6 +121,34 @@ export function threadHasQueuedTurnStart(
   ].every((value) => value === null || value < messageAtMs);
 }
 
+/**
+ * Providers drop finished work from the roster before the turn that reports it is requested
+ * (Claude: about 100ms). Settling in that gap detaches the session, so the report never arrives.
+ */
+function awaitsHeldWorkReport(
+  thread: Pick<
+    ProjectionStore.ProjectionSettlementCandidate,
+    "heldWorkEndedAt" | "latestUserMessageAt" | "latestRunRequestedAt" | "status"
+  >,
+  nowMs: number,
+): boolean {
+  const heldWorkEndedAtMs = toMillis(thread.heldWorkEndedAt);
+  // The grace bounds the hold when no report turn comes, as when the run was stopped.
+  if (
+    heldWorkEndedAtMs === null ||
+    Math.abs(nowMs - heldWorkEndedAtMs) > QUEUED_TURN_START_GRACE_MS
+  ) {
+    return false;
+  }
+  // A report requested behind another run stays queued until that run ends.
+  if (thread.status === "queued") return true;
+  const turnRequestedAtMs = latestMillis([
+    toMillis(thread.latestUserMessageAt),
+    toMillis(thread.latestRunRequestedAt),
+  ]);
+  return turnRequestedAtMs === null || turnRequestedAtMs < heldWorkEndedAtMs;
+}
+
 function pullRequestSettles(
   thread: Pick<
     ProjectionStore.ProjectionSettlementCandidate,
@@ -145,7 +173,10 @@ function pullRequestSettles(
 }
 
 export function isAutoSettlementCandidate(
-  thread: Omit<ProjectionStore.ProjectionSettlementCandidate, "latestUserAuthoredMessageAt">,
+  thread: Omit<
+    ProjectionStore.ProjectionSettlementCandidate,
+    "latestUserAuthoredMessageAt" | "heldWorkEndedAt"
+  >,
   nowMs: number,
 ): boolean {
   if (thread.archivedAt !== null || thread.settledOverride !== null) return false;
@@ -329,7 +360,9 @@ export const make = Effect.gen(function* () {
     const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
     const projects = new Map(projectShells.map((project) => [project.id, project]));
     // Merge events resweep all candidates because unlinked threads settle through branch lookup.
-    const candidates = threads.filter((thread) => isAutoSettlementCandidate(thread, nowMs));
+    const candidates = threads.filter(
+      (thread) => isAutoSettlementCandidate(thread, nowMs) && !awaitsHeldWorkReport(thread, nowMs),
+    );
 
     // Runs per candidate thread, thousands per sweep on busy servers; the sweep owns the span.
     const settleThread = Effect.fnUntraced(

apps/server/src/persistence/Migrations.ts

@@ -83,6 +83,7 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts";
 import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts";
 import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts";
 import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts";
+import Migration0081 from "./Migrations/081_ProviderThreadHeldWorkEndedAt.ts";
 
 const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -166,6 +167,7 @@ const migrationEntries = [
   [78, "DropOrchestrationEventCorrelationIndex", Migration0078],
   [79, "DelegatedRecoveryIndexes", Migration0079],
   [80, "ScheduledTaskThreadMode", Migration0080],
+  [81, "ProviderThreadHeldWorkEndedAt", Migration0081],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/081_ProviderThreadHeldWorkEndedAt.ts

@@ -0,0 +1,9 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+// Rows start out holding nothing: restart recovery clears every roster before new turns run.
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  yield* sql`ALTER TABLE orchestration_v2_projection_provider_threads ADD COLUMN holds_completion INTEGER NOT NULL DEFAULT 0`;
+  yield* sql`ALTER TABLE orchestration_v2_projection_provider_threads ADD COLUMN held_work_ended_at TEXT`;
+});