diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index 825bc5d7ce53c..1a901e4358fa2 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/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; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index dea68ee7723d1..4b60715d1e943 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/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 & { 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 = 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 = 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 = ${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 = 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 = 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 = 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, diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index f3c83eca70646..5d06b17fbd243 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -53,7 +53,10 @@ function at(offsetMs: number): DateTime.Utc { } type SettlementShell = OrchestrationV2ThreadShell & - Pick; + Pick< + ProjectionStore.ProjectionSettlementCandidate, + "latestUserAuthoredMessageAt" | "heldWorkEndedAt" + >; function shell(overrides: Partial = {}): SettlementShell { return { @@ -93,6 +96,7 @@ function shell(overrides: Partial = {}): 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(); + // 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(); + // 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* () { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 8c852426da323..3abb6d9ebc281 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/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, + 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( diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6a9c57e18cc43..3f2292fdc3804 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/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); diff --git a/apps/server/src/persistence/Migrations/081_ProviderThreadHeldWorkEndedAt.ts b/apps/server/src/persistence/Migrations/081_ProviderThreadHeldWorkEndedAt.ts new file mode 100644 index 0000000000000..5a56f21839cc2 --- /dev/null +++ b/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`; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 84fa4ddb44c77..3b2ca8ca19c03 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -60,6 +60,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProviderThreadHeldWorkEndedAt"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -163,6 +164,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProviderThreadHeldWorkEndedAt"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );