diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index de54a97ebebf8..d86e87d546c35 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -827,6 +827,105 @@ it.effect("record reads filter notification sources without dropping matching me }).pipe(Effect.provide(layerTest)), ); +it.effect("shells flag a latest run triggered by a provider background wake", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("background-wake"); + const initial = yield* store.getThreadProjection(threadId); + const at = initial.updatedAt; + let ordinal = 1; + for (const [id, source, outcome, expected] of [ + ["background-task", { kind: "background_task" }, "updated", true], + ["monitor", { kind: "monitor" }, "updated", true], + ["command", { kind: "command" }, "updated", true], + ["command-finished", { kind: "command" }, "completed", false], + ["monitor-stopped", { kind: "monitor" }, "cancelled", false], + ["subagent", { kind: "subagent" }, "updated", false], + ["delegated", { kind: "delegated_task", taskIds: [] }, "updated", false], + ["steered", { kind: "background_task" }, "updated", false], + ] as const) { + const messageId = MessageId.make(`message:background-wake:${id}`); + yield* store.apply({ + id: EventId.make(`event:background-wake:${id}:message`), + type: "message.updated", + threadId, + occurredAt: at, + payload: { + id: messageId, + threadId, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "server", + role: "user", + text: id, + attachments: [], + streaming: false, + createdAt: at, + updatedAt: at, + notification: { source, outcome, summary: "Background activity updated" }, + ...(id === "delegated" + ? { + delegatedCompletion: { + parentRunId: initial.runs[0]!.id, + generation: 1, + taskIds: [], + }, + } + : {}), + }, + }); + ordinal += 1; + yield* store.apply({ + id: EventId.make(`event:background-wake:${id}:run`), + type: "run.created", + threadId, + occurredAt: at, + payload: { + ...initial.runs[0]!, + id: RunId.make(`run:background-wake:${id}`), + ordinal, + userMessageId: messageId, + status: "completed", + completedAt: at, + }, + }); + if (id === "steered") { + yield* store.apply({ + id: EventId.make("event:background-wake:steered:follow-up"), + type: "message.updated", + threadId, + occurredAt: at, + payload: { + id: MessageId.make("message:background-wake:steered:follow-up"), + threadId, + runId: RunId.make("run:background-wake:steered"), + nodeId: null, + createdBy: "user", + creationSource: "web", + role: "user", + text: "also fix X", + attachments: [], + streaming: false, + createdAt: at, + updatedAt: at, + }, + }); + } + const shells = [ + yield* store.getThreadShell(threadId), + ProjectionStore.threadShellFromProjection(yield* store.getThreadProjection(threadId)), + ]; + assert.deepEqual( + shells.map((shell) => shell?.latestRunBackgroundWake), + [expected, expected], + id, + ); + yield* assertShellIndexCoversShells(); + } + }).pipe(Effect.provide(layerTest)), +); + it.effect("SQL queue readiness skips held runs without bypassing active work", () => 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 7c7bbecafe5a0..9dd5cdfe3e4f8 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1032,6 +1032,7 @@ type ShellThreadRow = { readonly latest_run_started_at: string | null; readonly latest_run_completed_at: string | null; readonly latest_run_stop_requested: number; + readonly latest_run_background_wake: number; readonly active_run_id: string | null; readonly active_run_status: string | null; readonly activity_run_status: string | null; @@ -1043,6 +1044,7 @@ type ShellThreadRow = { readonly blocking_run_started_at: string | null; readonly blocking_run_completed_at: string | null; readonly blocking_run_stop_requested: number; + readonly blocking_run_background_wake: number; readonly blocking_failure_payload_json: string | null; readonly pending_request_payload_json: string | null; readonly pending_secret_request_payload_json: string | null; @@ -1658,6 +1660,8 @@ export function threadShellFromProjection( projection.turnItems.some( (item) => item.runId === latestRun.id && item.type === "run_interrupt_request", ), + latestRunBackgroundWake: + latestRun != null && isBackgroundWakeRunForShell(latestRun, projection), activeRunId: activeRun?.id ?? null, blockedOnThreadWait: deriveBlockedOnThreadWait({ activeRun, turnItems: projection.turnItems }), activityRunStatus: activityRun?.status ?? null, @@ -1753,6 +1757,25 @@ function isSilentRunForShell( ); } +const BACKGROUND_WAKE_SOURCE_KINDS = new Set(["background_task", "command", "monitor"]); + +function isBackgroundWakeRunForShell( + run: OrchestrationV2ThreadProjection["runs"][number], + projection: Pick, +): boolean { + const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); + return ( + message?.delegatedCompletion === undefined && + // Only progress wakes; a terminal one ("Command … finished") is the work reporting its end. + message?.notification?.outcome === "updated" && + BACKGROUND_WAKE_SOURCE_KINDS.has(message.notification.source.kind) && + // A follow-up the user steered into the wake makes the run theirs. + !projection.messages.some( + (candidate) => candidate.runId === run.id && candidate.createdBy === "user", + ) + ); +} + function isInterruptibleRunForShell(run: OrchestrationV2ThreadProjection["runs"][number]): boolean { return run.status === "preparing" || run.status === "starting" || run.status === "running"; } @@ -1775,6 +1798,7 @@ type ShellThreadState = { readonly latestRunStartedAt: DateTime.Utc | null; readonly latestRunCompletedAt: DateTime.Utc | null; readonly latestRunStopRequested: boolean; + readonly latestRunBackgroundWake: boolean; readonly activeRunId: RunId | null; readonly blockedOnThreadWait: boolean; readonly activityRunStatus: ShellActivityRunStatus | null; @@ -1935,6 +1959,7 @@ function shellFromState(input: { latestRunStartedAt: input.state.latestRunStartedAt, latestRunCompletedAt: input.state.latestRunCompletedAt, latestRunStopRequested: input.state.latestRunStopRequested, + latestRunBackgroundWake: input.state.latestRunBackgroundWake, activeRunId: input.state.activeRunId, blockedOnThreadWait: input.state.blockedOnThreadWait, activityRunStatus: input.state.activityRunStatus, @@ -6267,6 +6292,25 @@ export const layer: Layer.Layer = LIMIT 1 )`; + // Mirrors isBackgroundWakeRunForShell. Stored sources keep the legacy encoding, where a + // command is background_command and a subagent is background_task with work "subagent". + const isBackgroundWakeShellRun = (alias: string) => { + const run = sql.literal(alias); + return sql`EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages wake + WHERE wake.message_id = json_extract(${run}.payload_json, '$.userMessageId') + AND json_extract(wake.payload_json, '$.notification.source.kind') + IN ('background_task', 'background_command', 'command', 'monitor') + AND json_extract(wake.payload_json, '$.notification.source.work') IS NULL + AND json_extract(wake.payload_json, '$.notification.outcome') = 'updated' + AND json_extract(wake.payload_json, '$.delegatedCompletion') IS NULL + ) AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages steer + WHERE steer.run_id = ${run}.run_id + AND json_extract(steer.payload_json, '$.createdBy') = 'user' + )`; + }; + const selectShellThreadRows = ( threadId?: ThreadId, { @@ -6299,6 +6343,7 @@ export const layer: Layer.Layer = WHERE stop_request.thread_id = t.thread_id AND stop_request.run_id = presented.run_id AND stop_request.type = 'run_interrupt_request' ) AS latest_run_stop_requested, + ${isBackgroundWakeShellRun("presented")} AS latest_run_background_wake, active.run_id AS active_run_id, active.status AS active_run_status, ( @@ -6356,6 +6401,7 @@ export const layer: Layer.Layer = WHERE stop_request.thread_id = t.thread_id AND stop_request.run_id = blocked.run_id AND stop_request.type = 'run_interrupt_request' ) AS blocking_run_stop_requested, + ${isBackgroundWakeShellRun("blocked")} AS blocking_run_background_wake, ( SELECT item.payload_json FROM orchestration_v2_projection_turn_items item @@ -6928,6 +6974,7 @@ export const layer: Layer.Layer = ? null : DateTime.makeUnsafe(row.latest_run_completed_at); let latestRunStopRequested = row.latest_run_stop_requested === 1; + let latestRunBackgroundWake = row.latest_run_background_wake === 1; if (row.blocking_run_id !== null && row.blocking_failure_payload_json !== null) { const blockingFailure = yield* decodeTurnItemPayload( row.blocking_failure_payload_json, @@ -6941,6 +6988,7 @@ export const layer: Layer.Layer = terminalFailureItem = blockingFailure; latestRunId = RunId.make(row.blocking_run_id); latestRunStopRequested = row.blocking_run_stop_requested === 1; + latestRunBackgroundWake = row.blocking_run_background_wake === 1; latestRunStatus = "failed"; latestRunRequestedAt = row.blocking_run_requested_at === null @@ -6981,6 +7029,7 @@ export const layer: Layer.Layer = latestRunStartedAt, latestRunCompletedAt, latestRunStopRequested, + latestRunBackgroundWake, activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id), blockedOnThreadWait: deriveBlockedOnThreadWait({ activeRun: @@ -7251,6 +7300,7 @@ export const layer: Layer.Layer = activityRunStatus: state.activityRunStatus, latestRunCompletedAt: state.latestRunCompletedAt, latestRunStopRequested: state.latestRunStopRequested, + latestRunBackgroundWake: state.latestRunBackgroundWake, pendingRuntimeRequest: state.pendingRuntimeRequest, pendingBackgroundTasks: state.pendingBackgroundTasks, })), diff --git a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts index 257994ccc76de..aab1de3db6726 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts @@ -5,7 +5,8 @@ import * as Option from "effect/Option"; import { Atom, AtomRegistry } from "effect/reactivity"; import { expect, it, vi } from "vite-plus/test"; -import { stubGlobal } from "../../test/render"; +import { fakeTimers, stubGlobal } from "../../test/render"; +import { RELEASED_COMPLETION_GRACE_MS } from "./ThreadNotificationCoordinator.logic"; const state = vi.hoisted(() => ({ mode: "notifications", @@ -267,6 +268,69 @@ it("presents completion notifications and sound as soon as the shell completes", expect(state.sound).toHaveBeenCalledWith("completion", expect.any(Function)); }); +it.each([false, true])( + "presents one completion when a monitor ends with a final wake (disconnected: %s)", + async (disconnect) => { + await using app = setup(); + using _timers = fakeTimers(); + state.mode = "notifications-and-sound"; + app.start(); + const watching = ( + turnId: string, + turnState: string, + completedAt: string | null, + latestRunBackgroundWake: boolean, + pendingBackgroundTasks: ReadonlyArray<{ taskId: string; kind: "monitor" }>, + ) => ({ + status: "live", + snapshot: Option.some({ + threads: [ + { + ...toV2ThreadShell({ + ...thread, + latestTurn: { turnId, state: turnState, completedAt }, + }), + latestRunBackgroundWake, + pendingBackgroundTasks, + }, + ], + }), + }); + const monitor = [{ taskId: "watch", kind: "monitor" as const }]; + const titles = () => TestNotification.sent.map(({ title }) => title); + state.shells.set( + "one", + watching("turn", "completed", "2026-09-13T07:00:01.000Z", false, monitor), + ); + expect(titles()).toEqual(["Thread completed"]); + + state.shells.set("one", watching("wake", "running", null, true, monitor)); + state.shells.set( + "one", + watching("wake", "completed", "2026-09-13T07:00:09.000Z", true, monitor), + ); + expect(titles()).toEqual(["Thread completed"]); + + // The adapter drains the roster before the final wake starts. + state.shells.set("one", watching("wake", "completed", "2026-09-13T07:00:09.000Z", true, [])); + expect(titles()).toEqual(["Thread completed"]); + if (disconnect) { + state.shells.set("one", { + ...watching("wake", "completed", "2026-09-13T07:00:09.000Z", true, []), + status: "cached", + }); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(titles()).toEqual(["Thread completed"]); + } + + state.shells.set("one", watching("final", "running", null, false, [])); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + state.shells.set("one", watching("final", "completed", "2026-09-13T07:00:11.000Z", false, [])); + expect(titles()).toEqual(["Thread completed", "Thread completed"]); + expect(state.sound).toHaveBeenCalledTimes(2); + }, +); + it.each(["hasPendingApprovals", "hasPendingUserInput"] as const)( "notifies for offline completion and %s once across reconnects", async (flag) => { diff --git a/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts b/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts index 1c98f89aa56ca..965dba45ce8a2 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vite-plus/test"; import { createThreadNotificationGate, + RELEASED_COMPLETION_GRACE_MS, type ThreadNotificationObservation, } from "./ThreadNotificationCoordinator.logic"; @@ -10,15 +11,27 @@ const working: ThreadNotificationObservation = { requestId: null, settled: false, completedAt: null, + backgroundWake: false, }; const completed = (at: number): ThreadNotificationObservation => ({ attention: null, requestId: null, settled: true, completedAt: at, + backgroundWake: false, +}); +const waking: ThreadNotificationObservation = { ...working, backgroundWake: true }; +const woke = (at: number): ThreadNotificationObservation => ({ + ...completed(at), + backgroundWake: true, }); const visible = { archived: false, quiet: false }; +function fakeTimers(): Disposable { + vi.useFakeTimers(); + return { [Symbol.dispose]: () => void vi.useRealTimers() }; +} + function harness() { const notify = vi.fn(); const gate = createThreadNotificationGate(notify); @@ -59,12 +72,7 @@ describe("createThreadNotificationGate", () => { it("notifies each attention once per stretch and still notifies the subsequent completion", () => { const { notify, observe } = harness(); observe(working); - const attention = { - attention: "run-1:approval:a", - requestId: "a", - settled: false, - completedAt: null, - }; + const attention = { ...working, attention: "run-1:approval:a", requestId: "a" }; observe(attention); expect(notify.mock.calls).toEqual([["thread", "input"]]); observe({ ...attention }); @@ -96,7 +104,7 @@ describe("createThreadNotificationGate", () => { const { notify, observe } = harness(); observe(working); observe(completed(1)); - observe({ attention: "run-1:limited", requestId: null, settled: false, completedAt: null }); + observe({ ...working, attention: "run-1:limited" }); observe(completed(1)); expect(notify.mock.calls).toEqual([ ["thread", "completion"], @@ -132,4 +140,58 @@ describe("createThreadNotificationGate", () => { expect(notify.mock.calls).toEqual([["thread", "completion"]]); } }); + + it("waits out the grace before notifying a wake's completion, and a final wake replaces it", () => { + using _timers = fakeTimers(); + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + // Held until the monitor left, or completing after it already had: either way unheld here. + observe(waking); + observe(woke(2)); + expect(notify).toHaveBeenCalledTimes(1); + observe(working); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(notify).toHaveBeenCalledTimes(1); + observe(completed(3)); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(notify.mock.calls).toEqual([ + ["thread", "completion"], + ["thread", "completion"], + ]); + }); + + it("notifies a wake's completion once after the grace when no run follows, even across a pause", () => { + using _timers = fakeTimers(); + const { gate, notify, observe } = harness(); + observe(waking); + observe(woke(1)); + gate.pause(); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(notify).not.toHaveBeenCalled(); + observe({ ...woke(1) }); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS - 1); + observe({ ...woke(1) }); + expect(notify).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(notify.mock.calls).toEqual([["thread", "completion"]]); + observe({ ...woke(1) }); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(notify).toHaveBeenCalledTimes(1); + }); + + it("drops a pending wake completion when the thread is removed or the gate is cleared", () => { + using _timers = fakeTimers(); + for (const reset of [ + (gate: ReturnType>) => gate.retain(new Set()), + (gate: ReturnType>) => gate.clear(), + ]) { + const { gate, notify, observe } = harness(); + observe(waking); + observe(woke(1)); + reset(gate); + vi.advanceTimersByTime(RELEASED_COMPLETION_GRACE_MS); + expect(notify).not.toHaveBeenCalled(); + } + }); }); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.logic.ts b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts index 6414f2e88f1a3..c8f0ca7df7a9a 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.logic.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts @@ -9,16 +9,38 @@ export interface ThreadNotificationObservation { readonly settled: boolean; /** When the latest run completed, if it did. */ readonly completedAt: number | null; + /** The latest run answers a background-work progress wake, like a Monitor line. */ + readonly backgroundWake: boolean; } +/** Matches the push relay's completion confirmation: a final wake can start right after a progress wake. */ +export const RELEASED_COMPLETION_GRACE_MS = 5_000; + export function createThreadNotificationGate( notify: (key: Key, kind: ThreadNotificationKind) => void, ) { // The newest completion already notified, or deliberately skipped. const handled = new Map(); + // Wake completions waiting out the grace, by completion time; no timer while paused. + const releases = new Map< + Key, + { readonly at: number; readonly timer: ReturnType | undefined } + >(); + const release = (key: Key, at: number) => { + const timer = setTimeout(() => { + releases.delete(key); + handled.set(key, at); + notify(key, "completion"); + }, RELEASED_COMPLETION_GRACE_MS); + releases.set(key, { at, timer }); + }; const skip = (key: Key, at: number | null) => { if (at !== null && at > (handled.get(key) ?? -Infinity)) handled.set(key, at); }; + const cancel = (key: Key) => { + clearTimeout(releases.get(key)?.timer); + releases.delete(key); + }; // Attention keys with a request id seen since the thread last stopped needing attention. const attended = new Map>(); const attend = ( @@ -44,6 +66,13 @@ export function createThreadNotificationGate( next: ThreadNotificationObservation, options: { readonly archived: boolean; readonly quiet: boolean }, ): void { + // The grace survives updates that keep the wake's completion; anything else cancels it. + const pending = releases.get(key); + if (pending?.at === next.completedAt && !options.archived && !options.quiet) { + if (pending.timer === undefined) release(key, pending.at); + return; + } + cancel(key); // The first sighting, archived threads, and stopped or silent runs never notify completion. if (!prior || options.archived || options.quiet) skip(key, next.completedAt); const fresh = attend(key, prior, next); @@ -54,15 +83,28 @@ export function createThreadNotificationGate( } const at = next.completedAt; if (!next.settled || at === null || at <= (handled.get(key) ?? -Infinity)) return; + if (next.backgroundWake) { + release(key, at); + return; + } handled.set(key, at); notify(key, "completion"); }, + /** Stops grace timers while the snapshot is stale; the next observation restarts or drops them. */ + pause(): void { + for (const [key, { at, timer }] of releases) { + clearTimeout(timer); + releases.set(key, { at, timer: undefined }); + } + }, /** Forgets threads that left the snapshot. */ retain(keys: ReadonlySet): void { for (const key of handled.keys()) if (!keys.has(key)) handled.delete(key); + for (const key of releases.keys()) if (!keys.has(key)) cancel(key); for (const key of attended.keys()) if (!keys.has(key)) attended.delete(key); }, clear(): void { + for (const key of releases.keys()) cancel(key); handled.clear(); attended.clear(); }, diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index 4d772458a028e..2051d4ad633a8 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -7,6 +7,7 @@ import { isShownUnderParent } from "@t3tools/client-runtime/state/threads"; import type { AnyRouter } from "@tanstack/react-router"; import { threadKey } from "@t3tools/client-runtime/state/entities"; import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; +import { monitorHoldsWakeCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import type { ClientSettings } from "@t3tools/contracts/settings"; import * as Option from "effect/Option"; import type { AtomRegistry } from "effect/reactivity"; @@ -263,8 +264,10 @@ export function startThreadNotifications( const attention = needsAttention ? `${thread.latestRun?.runId ?? ""}:${status}:${requestId ?? ""}` : null; - const completedAt = Date.parse(thread.latestRun?.completedAt ?? ""); - const settled = status === "ready" && thread.latestRun?.status === "completed"; + const completed = status === "ready" && thread.latestRun?.status === "completed"; + const settled = completed && !monitorHoldsWakeCompletion(rawThread); + const parsedCompletedAt = Date.parse(thread.latestRun?.completedAt ?? ""); + const completedAt = Number.isFinite(parsedCompletedAt) ? parsedCompletedAt : null; const state = { raw: rawThread, subagentRequest, @@ -273,7 +276,8 @@ export function startThreadNotifications( attention, requestId, settled, - completedAt: settled && Number.isFinite(completedAt) ? completedAt : null, + completedAt: settled ? completedAt : null, + backgroundWake: rawThread.latestRunBackgroundWake === true, }; next.set(thread.id, state); changed.push([prior, state]); @@ -312,6 +316,7 @@ export function startThreadNotifications( return; } entry.reconnecting = true; + entry.gate.pause(); }, { immediate: true }, ); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 856cf89bafcb6..f7b582dd9f854 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2121,6 +2121,8 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ latestRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), latestRunCompletedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), latestRunStopRequested: Schema.optional(Schema.Boolean), + /** The latest run answers a background-work progress wake, like a Monitor line, not a terminal one. Absent from older servers. */ + latestRunBackgroundWake: Schema.optional(Schema.Boolean), activeRunId: Schema.NullOr(RunId), activityRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), activityRunStatus: Schema.optional( diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index d5f77e763622e..71f4f21bd1e2a 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -24,6 +24,7 @@ describe("projectThreadAwarenessV2", () => { Pick< OrchestrationV2ThreadShell, | "activityRunStatus" + | "latestRunBackgroundWake" | "latestRunStopRequested" | "status" | "pendingBackgroundTasks" @@ -63,6 +64,44 @@ describe("projectThreadAwarenessV2", () => { ).toBeNull(); }); + it.each([ + ["holds a wake run", true, [{ taskId: "watch", kind: "monitor" as const }], "running"], + ["completes a wake run once its monitor ends", true, [], "completed"], + [ + "holds a wake run while a monitor remains beside other work", + true, + [ + { taskId: "build", kind: "command" as const }, + { taskId: "watch", kind: "monitor" as const }, + ], + "running", + ], + [ + "completes a wake run with only a command pending", + true, + [{ taskId: "build", kind: "command" as const }], + "completed", + ], + [ + "completes a run the user started", + false, + [{ taskId: "watch", kind: "monitor" as const }], + "completed", + ], + ] as const)("%s while a monitor wakes the thread", (_, latestRunBackgroundWake, tasks, phase) => { + expect( + projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + thread: v2Thread({ + status: "completed", + latestRunBackgroundWake, + pendingBackgroundTasks: [...tasks], + }), + })?.phase, + ).toBe(phase); + }); + it.each([ ["running", "subagent"], ["completed", "subagent"], @@ -97,12 +136,21 @@ describe("projectThreadAwarenessV2", () => { ).toMatchObject({ phase: "running", headline: "Agent is working" }); }); - it.each(["completed", "failed"] as const)("removes settled %s work", (status) => { + it.each([ + ["completed", "completed", false], + ["failed", "failed", false], + ["monitor-held", "completed", true], + ] as const)("removes settled %s work", (_, status, latestRunBackgroundWake) => { expect( projectThreadAwarenessV2({ environmentId: "env-1" as EnvironmentId, project, - thread: v2Thread({ status, settledOverride: "settled" }), + thread: v2Thread({ + status, + settledOverride: "settled", + latestRunBackgroundWake, + pendingBackgroundTasks: [{ taskId: "watch", kind: "monitor" }], + }), }), ).toBeNull(); }); diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index e6b377c5b0ddf..55ad807957d64 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -9,7 +9,10 @@ import { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; -import { backgroundWorkHoldsCompletion } from "./orchestrationV2PendingBackgroundWork.ts"; +import { + backgroundWorkHoldsCompletion, + monitorHoldsWakeCompletion, +} from "./orchestrationV2PendingBackgroundWork.ts"; export type AgentAwarenessPhase = | "starting" @@ -73,6 +76,7 @@ export interface ProjectThreadAwarenessV2Input { | "activityRunStatus" | "id" | "lineage" + | "latestRunBackgroundWake" | "latestRunStopRequested" | "modelSelection" | "pendingBackgroundTasks" @@ -98,9 +102,12 @@ export function projectThreadAwarenessV2( } if (phase === "completed" && thread.latestRunStopRequested === true) return null; // Resolve the phase first so live or waiting work stays visible after settlement. + const runStatus = thread.activityRunStatus ?? thread.status; if ( thread.settledOverride === "settled" && - (phase === "completed" || phase === "failed") && + (phase === "completed" || + phase === "failed" || + (phase === "running" && runStatus === "completed")) && !backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) ) { return null; @@ -153,7 +160,8 @@ function resolveThreadAwarenessPhaseV2( case "waiting": return "running"; case "completed": - return backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) + return backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) || + monitorHoldsWakeCompletion(thread) ? "running" : "completed"; case "failed": diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 6dc635b5153a6..dfdf9a24ace1e 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -2,6 +2,7 @@ import type { OrchestrationV2PendingBackgroundTask, OrchestrationV2ProviderThread, OrchestrationV2Run, + OrchestrationV2ThreadShell, OrchestrationV2TurnItem, ThreadId, ThreadPullRequestLink, @@ -107,6 +108,16 @@ function backgroundWorkKindHoldsCompletion(kind: PendingBackgroundWorkTask["kind } } +/** A monitor wakes the agent once per line, so a progress wake run's completion waits for the monitor to end. */ +export function monitorHoldsWakeCompletion( + thread: Pick, +): boolean { + return ( + thread.latestRunBackgroundWake === true && + (thread.pendingBackgroundTasks ?? []).some((task) => task.kind === "monitor") + ); +} + type PendingBackgroundWorkRun = Pick; type PendingBackgroundWorkPullRequest = Pick<