diff --git a/src/node/services/taskService.shared.testHarness.ts b/src/node/services/taskService.shared.testHarness.ts index 81a0fc6ea6e..73b910d01ce 100644 --- a/src/node/services/taskService.shared.testHarness.ts +++ b/src/node/services/taskService.shared.testHarness.ts @@ -422,8 +422,10 @@ export async function flushTerminalAttentionDrains( assert(passes > 0, "flushTerminalAttentionDrains needs at least one pass"); for (let i = 0; i < passes; i++) { const drains = ( - taskService as unknown as { pendingTerminalAttentionDrains: Set> } - ).pendingTerminalAttentionDrains; + taskService as unknown as { + terminalAttention: { pendingTerminalAttentionDrains: Set> }; + } + ).terminalAttention.pendingTerminalAttentionDrains; if (drains.size === 0) break; await Promise.all([...drains]); } diff --git a/src/node/services/taskService.terminalAttentionIdleWait.test.ts b/src/node/services/taskService.terminalAttentionIdleWait.test.ts index 04fa56e7b47..02b0f4857ec 100644 --- a/src/node/services/taskService.terminalAttentionIdleWait.test.ts +++ b/src/node/services/taskService.terminalAttentionIdleWait.test.ts @@ -97,10 +97,11 @@ describe("TaskService terminal attention idle waits", () => { const chatListeners = () => emitter.listenerCount("chat-event"); const baseline = chatListeners(); - const drainChains = Reflect.get(taskService, "pendingTerminalAttentionDrainsByOwner") as Map< - string, - Promise - >; + const drainChains = ( + taskService as unknown as { + terminalAttention: { pendingTerminalAttentionDrainsByOwner: Map> }; + } + ).terminalAttention.pendingTerminalAttentionDrainsByOwner; for (const [index, childId] of childIds.entries()) { const reportMarkdown = `Report ${index} from ${childId}`; diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 5b1713bce45..9550965060d 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -349,9 +349,12 @@ import { import { TerminalAttentionStore, type TerminalAttentionNotification, - type TerminalAttentionOutcome, } from "@/node/services/terminalAttentionStore"; import { WorkflowRunAttention } from "@/node/services/workflowRunAttention"; +import { + TerminalAttentionNotifier, + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS, +} from "@/node/services/terminalAttentionNotifier"; import { readAgentWorkflowRunReferences } from "@/node/services/agentWorkflowRunReferences"; import type { AgentWorkflowRunStrictPin } from "@/node/services/agentWorkflowRunReferences"; import { normalizeWorkflowAgentReportPayloadForHostSchema } from "@/common/utils/tools/workflowReportPayload"; @@ -1337,12 +1340,6 @@ function isWorkspaceBusyIdleOnlySend(error: unknown): boolean { const REMOVED_AGENT_TASKS_DIR = "removed-agent-tasks"; -// Level-triggered backstop for workflow terminal wakes: the sweep re-derives owed wakes from -// durable state (run records + settled markers), so lost terminal callbacks, deferred -// (transiently unreadable) evaluations, and crashes all recover here without any per-failure -// retry bookkeeping. -const WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS = 5 * 60_000; - /** Maximum consecutive auto-resumes before stopping. Prevents infinite loops when descendants are stuck. */ const MAX_CONSECUTIVE_PARENT_AUTO_RESUMES = 3; @@ -1919,29 +1916,13 @@ export class TaskService implements AgentTaskIntegration { // Git worktree creation touches per-repository metadata; serialize that narrow phase per project // while allowing post-fork init/send startup work for sibling tasks to overlap. private readonly reservedTaskLaunchByProjectPath = new Map>(); - // In-flight durable persistence of notify_on_terminal policy for backgrounded foreground waits. - // Awaited at the start of handleStreamEnd so a just-detached wait is treated as non-blocking. - private readonly pendingNotifyOnTerminalPersists = new Set>(); - // In-flight terminal attention drains (workspace-turn / sub-agent terminal wake-ups). Tracked so - // tests and shutdown can await them; drains are idempotent and re-triggered on owner idle events. - private readonly pendingTerminalAttentionDrainsByOwner = new Map>(); - private readonly pendingTerminalAttentionDrains = new Set>(); + // Terminal attention scheduling and bookkeeping: notify_on_terminal persists, drain chains, idle + // waits, fresh sub-agent edges and wake-group send backoff. The drain itself stays here. + private readonly terminalAttention: TerminalAttentionNotifier; /** Tail of each task's chain of deferred trusted messages (F54), in send order. */ private readonly deferredDescendantMessagesByTask = new Map>(); - private readonly pendingTerminalAttentionIdleWaitsByOwner = new Map>(); - // Sub-agent terminal notifications this process just enqueued while their owner was streaming, - // mapped to that owner turn's generation (owner -> notification ID -> turn). The next drain takes - // the whole map, so each report can cut at most one busy turn, and only the turn that was live - // at its delivery (see cutBusyOwnerTurnForSubagentAttention). - private readonly freshSubagentAttentionTurnByOwner = new Map>(); // Owed workflow terminal wakes and the per-run settlement chain (see WorkflowRunAttention). private readonly workflowRunAttention: WorkflowRunAttention; - // Owner -> wake-group key -> epoch ms before which the drain skips selecting the group. A - // group whose send was rejected for group-specific reasons (for example an unresolvable - // strictly pinned agent) would otherwise be re-selected newest-first by every drain and - // sweep, starving older deliverable groups. In-memory only: a restart retries every group, - // and entries expire on read. - private readonly workflowWakeGroupSendBackoffUntilMs = new Map>(); private workflowAttentionSweepTimer: ReturnType | null = null; /** The pending drain retry after a refused launch lease (QUEUED_LAUNCH_LEASE_RETRY_MS). */ private queuedLaunchLeaseRetryTimer: ReturnType | undefined; @@ -2617,7 +2598,7 @@ export class TaskService implements AgentTaskIntegration { }); return; } - await this.suppressTerminalAttention({ + await this.terminalAttention.suppressTerminalAttention({ ownerWorkspaceId: execution.ownerWorkspaceId, sourceKind: "workspace_turn", sourceId: execution.handleId, @@ -4680,6 +4661,15 @@ export class TaskService implements AgentTaskIntegration { sessionsDir: () => this.config.sessionsDir, terminalAttentionStore: () => this.terminalAttentionStore, }); + this.terminalAttention = new TerminalAttentionNotifier({ + terminalAttentionStore: () => this.terminalAttentionStore, + config: () => this.config, + aiService: () => this.aiService, + workspaceService: () => this.workspaceService, + turnManager: () => this.getWorkspaceTurnManager(), + // Through TaskService: the test harness wraps drainTerminalAttention on the instance. + drain: (ownerWorkspaceId) => this.drainTerminalAttention(ownerWorkspaceId), + }); this.gitPatchArtifactService = new GitPatchArtifactService(config, historyService.locks); this.aiService.on("stream-end", (payload: unknown) => { @@ -9843,13 +9833,13 @@ export class TaskService implements AgentTaskIntegration { log.warn("Deferred descendant message failed", { taskId: params.taskId, error }); }) .finally(() => { - this.pendingTerminalAttentionDrains.delete(promise); + this.terminalAttention.untrackDrain(promise); if (this.deferredDescendantMessagesByTask.get(params.taskId) === promise) { this.deferredDescendantMessagesByTask.delete(params.taskId); } }); this.deferredDescendantMessagesByTask.set(params.taskId, promise); - this.pendingTerminalAttentionDrains.add(promise); + this.terminalAttention.trackDrain(promise); } private taskGuidanceSendOptions( @@ -11084,7 +11074,7 @@ export class TaskService implements AgentTaskIntegration { ); this.markWorkspaceStopPersisted(id); if (parentWorkspaceId != null) { - await this.suppressTerminalAttention({ + await this.terminalAttention.suppressTerminalAttention({ ownerWorkspaceId: parentWorkspaceId, sourceKind: "agent_task", sourceId: id, @@ -11137,7 +11127,7 @@ export class TaskService implements AgentTaskIntegration { }); continue; } - await this.suppressTerminalAttention({ + await this.terminalAttention.suppressTerminalAttention({ ownerWorkspaceId: handle.ownerWorkspaceId, sourceKind: "workspace_turn", sourceId: handle.handleId, @@ -11867,61 +11857,16 @@ export class TaskService implements AgentTaskIntegration { return this.waiters.backgroundForegroundWaitsForWorkspace(workspaceId); } - /** - * Persist `notify_on_terminal` on a backgrounded handle (agent task config or - * workspace-turn record). Tracked per workspace so `handleStreamEnd` can await - * any in-flight persistence before it reads config to decide blocking work. - */ - /** - * Durably mark a still-active task/handle as `notify_on_terminal`. Used when a foreground wait - * detaches because it exceeded its foreground wait budget (timeout) and the work continues in the - * background: like queued-message detachment, the work must not re-force the owner to await it. - * Awaited so callers (e.g. the task tool) can rely on the policy before returning pending results. - */ - async markBackgroundWorkNotifyOnTerminal( - taskId: string, - ownerWorkspaceId: string - ): Promise { - await this.persistNotifyOnTerminalPolicy(taskId, ownerWorkspaceId); + /** See TerminalAttentionNotifier.markBackgroundWorkNotifyOnTerminal (task tool calls it). */ + markBackgroundWorkNotifyOnTerminal(taskId: string, ownerWorkspaceId: string): Promise { + return this.terminalAttention.markBackgroundWorkNotifyOnTerminal(taskId, ownerWorkspaceId); } private scheduleNotifyOnTerminalPersist( taskId: string, ownerWorkspaceId: string | undefined ): void { - const promise = this.persistNotifyOnTerminalPolicy(taskId, ownerWorkspaceId) - .catch((error: unknown) => { - log.error("Failed to persist notify_on_terminal policy for backgrounded wait", { - taskId, - error, - }); - }) - .finally(() => { - this.pendingNotifyOnTerminalPersists.delete(promise); - }); - this.pendingNotifyOnTerminalPersists.add(promise); - } - - private async persistNotifyOnTerminalPolicy( - taskId: string, - ownerWorkspaceId: string | undefined - ): Promise { - if (isWorkspaceTurnTaskId(taskId)) { - if (ownerWorkspaceId != null) { - await this.getWorkspaceTurnManager().markWorkspaceTurnBackgroundWorkNotifyOnTerminal( - taskId, - ownerWorkspaceId - ); - } - return; - } - await this.config.editConfig((config) => { - const found = findWorkspaceEntry(config, taskId); - if (found != null && found.workspace.taskAttentionPolicy !== "notify_on_terminal") { - found.workspace.taskAttentionPolicy = "notify_on_terminal"; - } - return config; - }); + this.terminalAttention.scheduleNotifyOnTerminalPersist(taskId, ownerWorkspaceId); } /** @@ -12095,20 +12040,9 @@ export class TaskService implements AgentTaskIntegration { this.scheduleTerminalAttentionDrain(params.ownerWorkspaceId); } - /** - * Level-triggered retry for outbox (sub-agent / workspace-turn) attention: pending records - * are the durable "wake owed" state, but unlike workflow runs they have no periodic - * re-derivation of their own, so a drain that failed transiently (for example an unreadable - * history for caller restrictions) would otherwise leave them stuck until restart. Startup - * and the sweep interval both re-poke their owners; drains are idempotent and no-op when - * nothing is deliverable. - */ - private async schedulePendingTerminalAttentionOwnerDrains(): Promise { - const ownerWorkspaceIds = await this.terminalAttentionStore.listPendingOwnerWorkspaceIds(); - for (const ownerWorkspaceId of ownerWorkspaceIds) { - this.scheduleTerminalAttentionDrain(ownerWorkspaceId); - } - return ownerWorkspaceIds.length; + // Startup and the sweep call this; tests cast-call it. See the notifier for the why. + private schedulePendingTerminalAttentionOwnerDrains(): Promise { + return this.terminalAttention.schedulePendingTerminalAttentionOwnerDrains(); } /** @@ -12141,21 +12075,6 @@ export class TaskService implements AgentTaskIntegration { return await this.workflowRunAttention.clearWorkflowRunDowngradeSettlement(params); } - private async suppressTerminalAttention(params: { - ownerWorkspaceId: string; - sourceKind: TerminalAttentionNotification["sourceKind"]; - sourceId: string; - }): Promise { - await this.terminalAttentionStore.enqueueIfAbsent({ - ...params, - terminalOutcome: "interrupted", - }); - await this.terminalAttentionStore.markSuperseded( - params.ownerWorkspaceId, - TerminalAttentionStore.notificationId(params.sourceKind, params.sourceId) - ); - } - private async getAgentTerminalAttentionGenerationId( ownerWorkspaceId: string, childTaskId: string @@ -12167,47 +12086,15 @@ export class TaskService implements AgentTaskIntegration { return execution?.record.handleId; } - async enqueueTerminalAttention(params: { - ownerWorkspaceId: string; - sourceKind: TerminalAttentionNotification["sourceKind"]; - terminalOutcome: TerminalAttentionOutcome; - sourceId: string; - generationId?: string; - }): Promise { - const created = await this.terminalAttentionStore.enqueueIfAbsent(params); - if (created == null) { - return; - } - const liveTurn = this.aiService.isStreaming(created.ownerWorkspaceId) - ? this.workspaceService.getActiveTurnGeneration(created.ownerWorkspaceId) - : undefined; - if (created.sourceKind === "agent_task" && liveTurn != null) { - let fresh = this.freshSubagentAttentionTurnByOwner.get(created.ownerWorkspaceId); - if (fresh == null) { - fresh = new Map(); - this.freshSubagentAttentionTurnByOwner.set(created.ownerWorkspaceId, fresh); - } - fresh.set(created.id, liveTurn); - } - this.scheduleTerminalAttentionDrain(params.ownerWorkspaceId); + // Spied on in tests, and called through TaskHost (taskWorkspaceSeam): stays a facade. + enqueueTerminalAttention( + params: Parameters[0] + ): Promise { + return this.terminalAttention.enqueueTerminalAttention(params); } scheduleTerminalAttentionDrain(ownerWorkspaceId: string): void { - const previous = this.pendingTerminalAttentionDrainsByOwner.get(ownerWorkspaceId); - const promise = (previous ?? Promise.resolve()) - .catch(() => undefined) - .then(() => this.drainTerminalAttention(ownerWorkspaceId)) - .catch((error: unknown) => { - log.error("Terminal attention drain failed", { ownerWorkspaceId, error }); - }) - .finally(() => { - this.pendingTerminalAttentionDrains.delete(promise); - if (this.pendingTerminalAttentionDrainsByOwner.get(ownerWorkspaceId) === promise) { - this.pendingTerminalAttentionDrainsByOwner.delete(ownerWorkspaceId); - } - }); - this.pendingTerminalAttentionDrainsByOwner.set(ownerWorkspaceId, promise); - this.pendingTerminalAttentionDrains.add(promise); + this.terminalAttention.scheduleTerminalAttentionDrain(ownerWorkspaceId); } /** @@ -12326,34 +12213,6 @@ export class TaskService implements AgentTaskIntegration { return { ...state.restrictions, ...withToolPolicy(state.trace.policy), ...state.pin }; } - private scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId: string): void { - // One idle wait per owner: each wait subscribes to the owner session's chat-event, and every - // report deferred by a busy owner lands here. The drain it schedules reads every pending - // notification from the store, so one wait delivers all of them, in store order. - if (this.pendingTerminalAttentionIdleWaitsByOwner.has(ownerWorkspaceId)) return; - const promise = this.workspaceService - .waitForIdleAndNoQueuedMessages(ownerWorkspaceId) - .catch((error: unknown) => { - log.debug("Terminal attention idle wait failed; retrying drain anyway", { - ownerWorkspaceId, - error, - }); - }) - .then(() => { - // Release the slot before scheduling: that drain can find the owner busy again and must - // be able to install the next idle wait. - if (this.pendingTerminalAttentionIdleWaitsByOwner.get(ownerWorkspaceId) === promise) { - this.pendingTerminalAttentionIdleWaitsByOwner.delete(ownerWorkspaceId); - } - this.scheduleTerminalAttentionDrain(ownerWorkspaceId); - }) - .finally(() => { - this.pendingTerminalAttentionDrains.delete(promise); - }); - this.pendingTerminalAttentionIdleWaitsByOwner.set(ownerWorkspaceId, promise); - this.pendingTerminalAttentionDrains.add(promise); - } - /** * A withdrawn correlated continuation settles nothing itself, but its turn's stream end may have * deferred to it (#5261). Once the target is idle the stale-turn sweep decides from live state; @@ -12390,9 +12249,9 @@ export class TaskService implements AgentTaskIntegration { log.warn("Withdrawn workspace-turn continuation reconcile failed", { targetId, error }); }) .finally(() => { - this.pendingTerminalAttentionDrains.delete(promise); + this.terminalAttention.untrackDrain(promise); }); - this.pendingTerminalAttentionDrains.add(promise); + this.terminalAttention.trackDrain(promise); } /** @@ -12844,25 +12703,6 @@ export class TaskService implements AgentTaskIntegration { return consumedIds; } - /** - * Back the given terminal-wake send batches off until the sweep cadence retries them and - * re-poke the drain so the remaining batches get their send this cycle instead of starving - * behind the failed one. - */ - private backOffTerminalWakeSends(ownerWorkspaceId: string, keys: readonly string[]): void { - assert(keys.length > 0, "backOffTerminalWakeSends requires keys"); - let ownerBackoff = this.workflowWakeGroupSendBackoffUntilMs.get(ownerWorkspaceId); - if (ownerBackoff == null) { - ownerBackoff = new Map(); - this.workflowWakeGroupSendBackoffUntilMs.set(ownerWorkspaceId, ownerBackoff); - } - const retryAt = Date.now() + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS; - for (const key of keys) { - ownerBackoff.set(key, retryAt); - } - this.scheduleTerminalAttentionDrain(ownerWorkspaceId); - } - /** * A sub-agent report or failure that lands while its owner is mid-turn is already a durable row * in the owner's history, but the terminal drain defers until idle and the running turn's model @@ -12974,9 +12814,7 @@ export class TaskService implements AgentTaskIntegration { private async drainTerminalAttention(ownerWorkspaceId: string): Promise { // Every drain consumes the fresh edges, whichever branch it takes: an idle drain delivers them // itself, and a busy drain may cut the live turn for them exactly once. - const freshSubagentAttentionTurns = - this.freshSubagentAttentionTurnByOwner.get(ownerWorkspaceId) ?? new Map(); - this.freshSubagentAttentionTurnByOwner.delete(ownerWorkspaceId); + const freshSubagentAttentionTurns = this.terminalAttention.takeFreshTurns(ownerWorkspaceId); const allPending = await this.terminalAttentionStore.listPending(ownerWorkspaceId); // Legacy pre-reconciler outbox records for workflow runs (no generation suffix) are dead // state now that workflow wakes are re-derived from run records + settled markers: delete @@ -13033,7 +12871,7 @@ export class TaskService implements AgentTaskIntegration { freshSubagentAttentionTurns ); if (ownerHasBusyQueuedOrRetry && !this.interruptedParentWorkspaceIds.has(ownerWorkspaceId)) { - this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + this.terminalAttention.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); } return; } @@ -13156,21 +12994,10 @@ export class TaskService implements AgentTaskIntegration { // A backed-off non-workflow batch (its conversation-identity send was rejected) sits out // the drain entirely, staying pending for the sweep-cadence retry, so agent-bound groups // are not starved behind a send that fails the same way on every drain. - const nonWorkflowSendBackoffUntil = this.workflowWakeGroupSendBackoffUntilMs - .get(ownerWorkspaceId) - ?.get(NON_WORKFLOW_WAKE_BACKOFF_KEY); - let nonWorkflowSendBackedOff = false; - if (nonWorkflowSendBackoffUntil != null) { - if (nonWorkflowSendBackoffUntil > Date.now()) { - nonWorkflowSendBackedOff = true; - } else { - const ownerBackoff = this.workflowWakeGroupSendBackoffUntilMs.get(ownerWorkspaceId); - ownerBackoff?.delete(NON_WORKFLOW_WAKE_BACKOFF_KEY); - if (ownerBackoff?.size === 0) { - this.workflowWakeGroupSendBackoffUntilMs.delete(ownerWorkspaceId); - } - } - } + const nonWorkflowSendBackedOff = this.terminalAttention.isWakeSendBackedOff( + ownerWorkspaceId, + NON_WORKFLOW_WAKE_BACKOFF_KEY + ); const hasNonWorkflowDeliverables = !nonWorkflowSendBackedOff && (deliverableAgentNotificationIds.size > 0 || workspaceTurnCandidates.length > 0); @@ -13260,20 +13087,12 @@ export class TaskService implements AgentTaskIntegration { ); // Group keys embed \u0000, so the empty string safely keys the unpinned group. const backoffKey = groupKey ?? ""; - const ownerBackoff = this.workflowWakeGroupSendBackoffUntilMs.get(ownerWorkspaceId); - const backoffUntil = ownerBackoff?.get(backoffKey); - if (backoffUntil != null) { - if (backoffUntil > Date.now()) { - // Recently rejected send: leave the group queued and give the next group its turn. - remainingWorkflowPrompts = remainingWorkflowPrompts.filter( - (candidate) => !groupCandidates.includes(candidate) - ); - continue; - } - ownerBackoff?.delete(backoffKey); - if (ownerBackoff?.size === 0) { - this.workflowWakeGroupSendBackoffUntilMs.delete(ownerWorkspaceId); - } + if (this.terminalAttention.isWakeSendBackedOff(ownerWorkspaceId, backoffKey)) { + // Recently rejected send: leave the group queued and give the next group its turn. + remainingWorkflowPrompts = remainingWorkflowPrompts.filter( + (candidate) => !groupCandidates.includes(candidate) + ); + continue; } // Resolve the send identity before the revalidation reread so the reread stays the last // await before dispatch: a history clear that completes during this history and @@ -13445,12 +13264,14 @@ export class TaskService implements AgentTaskIntegration { // Same starvation shape as the prompt path: agent-bound groups were excluded by this // batch, so back it off and re-poke to let them send under their own launch identity. if (deliverableWorkflowPrompts.some((candidate) => candidate.initiatingAgent != null)) { - this.backOffTerminalWakeSends(ownerWorkspaceId, [NON_WORKFLOW_WAKE_BACKOFF_KEY]); + this.terminalAttention.backOffTerminalWakeSends(ownerWorkspaceId, [ + NON_WORKFLOW_WAKE_BACKOFF_KEY, + ]); } return; } if (!resumeResult.data.started) { - this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + this.terminalAttention.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); return; } await markPendingDelivered(); @@ -13526,11 +13347,11 @@ export class TaskService implements AgentTaskIntegration { // Same wake as the primary send above (#4721). skipOnSendCompaction: true, onCanceled: () => { - this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + this.terminalAttention.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); }, onAcceptedPreStreamFailure: async () => { await markPendingForRetry(); - this.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); + this.terminalAttention.scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId); }, onAccepted: async () => { fallbackAccepted = true; @@ -13563,7 +13384,7 @@ export class TaskService implements AgentTaskIntegration { backoffKeys.push(NON_WORKFLOW_WAKE_BACKOFF_KEY); } if (backoffKeys.length > 0) { - this.backOffTerminalWakeSends(ownerWorkspaceId, backoffKeys); + this.terminalAttention.backOffTerminalWakeSends(ownerWorkspaceId, backoffKeys); } } // Busy rejection: the owner started work between the idle check and the send; leave @@ -17590,8 +17411,9 @@ export class TaskService implements AgentTaskIntegration { // Ensure any in-flight notify_on_terminal persistence (from a just-detached foreground wait) // has settled so the config we read below reflects the durable non-blocking policy. - if (this.pendingNotifyOnTerminalPersists.size > 0) { - await Promise.all([...this.pendingNotifyOnTerminalPersists]); + const notifyOnTerminalPersists = this.terminalAttention.inFlightNotifyOnTerminalPersists(); + if (notifyOnTerminalPersists.length > 0) { + await Promise.all(notifyOnTerminalPersists); } // The owner's own stream ending is the signal to retry any terminal wake-ups that were deferred diff --git a/src/node/services/terminalAttentionNotifier.test.ts b/src/node/services/terminalAttentionNotifier.test.ts new file mode 100644 index 00000000000..1e3e67dd171 --- /dev/null +++ b/src/node/services/terminalAttentionNotifier.test.ts @@ -0,0 +1,457 @@ +import { + afterEach, + beforeEach, + describe, + expect, + mock, + setSystemTime, + spyOn, + test, +} from "bun:test"; + +import type { ProjectsConfig } from "@/node/config"; +import { log } from "@/node/services/log"; +import { + TerminalAttentionNotifier, + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS, +} from "@/node/services/terminalAttentionNotifier"; +import { + TerminalAttentionStore, + type TerminalAttentionNotification, +} from "@/node/services/terminalAttentionStore"; + +function deferred(): { promise: Promise; resolve: () => void; reject: (e: Error) => void } { + let resolve!: () => void; + let reject!: (e: Error) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +async function flush(notifier: TerminalAttentionNotifier): Promise { + for (let i = 0; i < 5 && notifier.pendingTerminalAttentionDrains.size > 0; i++) { + await Promise.all([...notifier.pendingTerminalAttentionDrains]); + } +} + +function createHarness() { + const events: string[] = []; + let streaming = false; + const turn = Symbol("turn"); + let created: TerminalAttentionNotification | null = null; + let pendingOwners: string[] = []; + let config: ProjectsConfig = { projects: new Map() }; + const store = { + enqueueIfAbsent: mock((params: Parameters[0]) => { + events.push(`enqueue:${params.sourceKind}:${params.terminalOutcome ?? ""}`); + return Promise.resolve(created); + }), + markSuperseded: mock((owner: string, id: string) => { + events.push(`superseded:${owner}:${id}`); + return Promise.resolve(); + }), + listPendingOwnerWorkspaceIds: mock(() => Promise.resolve(pendingOwners)), + }; + const idleWaits: Array> = []; + const workspaceService = { + getActiveTurnGeneration: mock((_id: string) => turn), + waitForIdleAndNoQueuedMessages: mock((_id: string) => { + const wait = deferred(); + idleWaits.push(wait); + return wait.promise; + }), + }; + const turnManager = { + markWorkspaceTurnBackgroundWorkNotifyOnTerminal: mock((taskId: string, owner: string) => { + events.push(`turn-policy:${taskId}:${owner}`); + return Promise.resolve(); + }), + }; + const editConfig = mock(async (fn: (c: ProjectsConfig) => ProjectsConfig) => { + config = fn(config); + await Promise.resolve(); + }); + let drainImpl: (owner: string) => Promise = (owner) => { + events.push(`drain:${owner}`); + return Promise.resolve(); + }; + const drain = mock((owner: string) => drainImpl(owner)); + const notifier = new TerminalAttentionNotifier({ + terminalAttentionStore: () => store, + config: () => ({ editConfig }), + aiService: () => ({ isStreaming: () => streaming }), + workspaceService: () => workspaceService, + turnManager: () => turnManager, + drain, + }); + return { + notifier, + events, + turn, + store, + workspaceService, + turnManager, + editConfig, + drain, + idleWaits, + get config() { + return config; + }, + set config(value: ProjectsConfig) { + config = value; + }, + setStreaming(value: boolean) { + streaming = value; + }, + setCreated(value: TerminalAttentionNotification | null) { + created = value; + }, + setPendingOwners(value: string[]) { + pendingOwners = value; + }, + setDrain(impl: (owner: string) => Promise) { + drainImpl = impl; + }, + }; +} + +function notification( + sourceKind: TerminalAttentionNotification["sourceKind"], + id = "n1" +): TerminalAttentionNotification { + return { + id, + ownerWorkspaceId: "owner", + sourceKind, + sourceId: "child", + outputDelivery: "already_injected", + terminalOutcome: "completed", + status: "pending", + createdAt: new Date(0).toISOString(), + }; +} + +// Silence the expected failure logs; tests assert the log call itself. +let logError: ReturnType>; +beforeEach(() => { + logError = spyOn(log, "error").mockImplementation(() => undefined); +}); + +afterEach(() => { + setSystemTime(); + logError.mockRestore(); +}); + +describe("TerminalAttentionNotifier drain scheduling", () => { + test("serializes drains per owner and clears the tail only after the last one", async () => { + const h = createHarness(); + const first = deferred(); + let calls = 0; + h.setDrain((owner) => { + calls++; + h.events.push(`start:${owner}:${calls}`); + return calls === 1 ? first.promise : Promise.resolve(); + }); + + h.notifier.scheduleTerminalAttentionDrain("owner"); + h.notifier.scheduleTerminalAttentionDrain("owner"); + h.notifier.scheduleTerminalAttentionDrain("other"); + // Tracked in the same tick as the schedule call. + expect(h.notifier.pendingTerminalAttentionDrains.size).toBe(3); + await Promise.resolve(); + await Promise.resolve(); + expect(h.events).toEqual(["start:owner:1", "start:other:2"]); + + first.resolve(); + await flush(h.notifier); + expect(h.events).toEqual(["start:owner:1", "start:other:2", "start:owner:3"]); + expect(h.notifier.pendingTerminalAttentionDrainsByOwner.size).toBe(0); + expect(h.notifier.pendingTerminalAttentionDrains.size).toBe(0); + }); + + test("a failed drain does not block the owner's next drain", async () => { + const h = createHarness(); + let calls = 0; + h.setDrain(() => { + calls++; + return calls === 1 ? Promise.reject(new Error("boom")) : Promise.resolve(); + }); + + h.notifier.scheduleTerminalAttentionDrain("owner"); + h.notifier.scheduleTerminalAttentionDrain("owner"); + await flush(h.notifier); + expect(h.drain).toHaveBeenCalledTimes(2); + expect(logError).toHaveBeenCalledTimes(1); + }); + + test("an earlier chain link settling keeps the newer tail registered", async () => { + const h = createHarness(); + const second = deferred(); + let calls = 0; + h.setDrain(() => { + calls++; + return calls === 2 ? second.promise : Promise.resolve(); + }); + + h.notifier.scheduleTerminalAttentionDrain("owner"); + h.notifier.scheduleTerminalAttentionDrain("owner"); + const tail = h.notifier.pendingTerminalAttentionDrainsByOwner.get("owner"); + for (let i = 0; i < 10; i++) await Promise.resolve(); + expect(calls).toBe(2); + expect(h.notifier.pendingTerminalAttentionDrainsByOwner.get("owner")).toBe(tail); + + second.resolve(); + await flush(h.notifier); + expect(h.notifier.pendingTerminalAttentionDrainsByOwner.has("owner")).toBe(false); + }); + + test("re-pokes every owner with pending notifications", async () => { + const h = createHarness(); + h.setPendingOwners(["a", "b"]); + expect(await h.notifier.schedulePendingTerminalAttentionOwnerDrains()).toBe(2); + await flush(h.notifier); + expect(h.events).toEqual(["drain:a", "drain:b"]); + }); + + test("tracked work is awaited with the drains until untracked", () => { + const h = createHarness(); + const work = Promise.resolve(); + h.notifier.trackDrain(work); + expect([...h.notifier.pendingTerminalAttentionDrains]).toEqual([work]); + h.notifier.untrackDrain(work); + expect(h.notifier.pendingTerminalAttentionDrains.size).toBe(0); + }); +}); + +describe("TerminalAttentionNotifier idle waits", () => { + test("installs one idle wait per owner and drains once it settles", async () => { + const h = createHarness(); + h.notifier.scheduleTerminalAttentionDrainAfterIdle("owner"); + h.notifier.scheduleTerminalAttentionDrainAfterIdle("owner"); + expect(h.workspaceService.waitForIdleAndNoQueuedMessages).toHaveBeenCalledTimes(1); + expect(h.drain).not.toHaveBeenCalled(); + + h.idleWaits[0].resolve(); + await flush(h.notifier); + expect(h.events).toEqual(["drain:owner"]); + }); + + test("a failed idle wait still drains", async () => { + const h = createHarness(); + h.notifier.scheduleTerminalAttentionDrainAfterIdle("owner"); + h.idleWaits[0].reject(new Error("session gone")); + await flush(h.notifier); + expect(h.events).toEqual(["drain:owner"]); + }); + + test("releases the slot before the drain, so a busy drain can wait again", async () => { + const h = createHarness(); + let calls = 0; + h.setDrain((owner) => { + calls++; + if (calls === 1) h.notifier.scheduleTerminalAttentionDrainAfterIdle(owner); + return Promise.resolve(); + }); + h.notifier.scheduleTerminalAttentionDrainAfterIdle("owner"); + h.idleWaits[0].resolve(); + for (let i = 0; i < 10; i++) await Promise.resolve(); + expect(h.workspaceService.waitForIdleAndNoQueuedMessages).toHaveBeenCalledTimes(2); + + h.idleWaits[1].resolve(); + await flush(h.notifier); + expect(h.drain).toHaveBeenCalledTimes(2); + }); +}); + +describe("TerminalAttentionNotifier enqueue", () => { + test("records a fresh edge for an agent task delivered into a live turn", async () => { + const h = createHarness(); + h.setStreaming(true); + h.setCreated(notification("agent_task", "agent_task:child")); + h.setDrain((owner) => { + // The drain sees the fresh edge: it was written before scheduling. + h.events.push(`drain:${owner}:${h.notifier.takeFreshTurns(owner).size}`); + return Promise.resolve(); + }); + + await h.notifier.enqueueTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "agent_task", + terminalOutcome: "completed", + sourceId: "child", + }); + await flush(h.notifier); + expect(h.events).toEqual(["enqueue:agent_task:completed", "drain:owner:1"]); + // Taken once: the next take is empty. + expect(h.notifier.takeFreshTurns("owner").size).toBe(0); + }); + + test("keeps the live turn per notification for the next drain", async () => { + const h = createHarness(); + h.setStreaming(true); + h.setDrain(() => new Promise(() => undefined)); + for (const id of ["a", "b"]) { + h.setCreated(notification("agent_task", id)); + await h.notifier.enqueueTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "agent_task", + terminalOutcome: "completed", + sourceId: id, + }); + } + expect([...h.notifier.takeFreshTurns("owner")]).toEqual([ + ["a", h.turn], + ["b", h.turn], + ]); + }); + + test("records no fresh edge for an idle owner or a workspace turn", async () => { + const h = createHarness(); + h.setCreated(notification("agent_task")); + await h.notifier.enqueueTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "agent_task", + terminalOutcome: "completed", + sourceId: "child", + }); + h.setStreaming(true); + h.setCreated(notification("workspace_turn")); + await h.notifier.enqueueTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "workspace_turn", + terminalOutcome: "completed", + sourceId: "wst_child", + }); + await flush(h.notifier); + expect(h.notifier.takeFreshTurns("owner").size).toBe(0); + expect(h.drain).toHaveBeenCalledTimes(2); + }); + + test("schedules nothing when the notification already exists", async () => { + const h = createHarness(); + await h.notifier.enqueueTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "agent_task", + terminalOutcome: "completed", + sourceId: "child", + }); + expect(h.notifier.pendingTerminalAttentionDrains.size).toBe(0); + expect(h.drain).not.toHaveBeenCalled(); + }); + + test("suppression writes an interrupted record, then marks it superseded", async () => { + const h = createHarness(); + await h.notifier.suppressTerminalAttention({ + ownerWorkspaceId: "owner", + sourceKind: "workspace_turn", + sourceId: "wst_child", + }); + expect(h.events).toEqual([ + "enqueue:workspace_turn:interrupted", + `superseded:owner:${TerminalAttentionStore.notificationId("workspace_turn", "wst_child")}`, + ]); + expect(h.drain).not.toHaveBeenCalled(); + }); +}); + +describe("TerminalAttentionNotifier send backoff", () => { + test("backs keys off for one sweep interval, then expires them on read", async () => { + const h = createHarness(); + setSystemTime(new Date(1_000_000)); + h.notifier.backOffTerminalWakeSends("owner", ["group", "other"]); + // Backoff is written before the drain it schedules. + expect(h.notifier.pendingTerminalAttentionDrainsByOwner.has("owner")).toBe(true); + await flush(h.notifier); + expect(h.events).toEqual(["drain:owner"]); + + expect(h.notifier.isWakeSendBackedOff("owner", "group")).toBe(true); + expect(h.notifier.isWakeSendBackedOff("owner", "unknown")).toBe(false); + expect(h.notifier.isWakeSendBackedOff("stranger", "group")).toBe(false); + + setSystemTime(new Date(1_000_000 + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS)); + expect(h.notifier.isWakeSendBackedOff("owner", "group")).toBe(false); + // Expired entries are gone: a later clock rewind does not revive them. + setSystemTime(new Date(1_000_000)); + expect(h.notifier.isWakeSendBackedOff("owner", "group")).toBe(false); + expect(h.notifier.isWakeSendBackedOff("owner", "other")).toBe(true); + }); + + test("dropping the owner's last expired key does not drop a later backoff", () => { + const h = createHarness(); + setSystemTime(new Date(1_000_000)); + h.notifier.backOffTerminalWakeSends("owner", ["a"]); + setSystemTime(new Date(1_000_000 + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS + 1)); + expect(h.notifier.isWakeSendBackedOff("owner", "a")).toBe(false); + h.notifier.backOffTerminalWakeSends("owner", ["b"]); + expect(h.notifier.isWakeSendBackedOff("owner", "b")).toBe(true); + }); + + test("refuses an empty key list", () => { + const h = createHarness(); + expect(() => h.notifier.backOffTerminalWakeSends("owner", [])).toThrow(); + expect(h.drain).not.toHaveBeenCalled(); + }); +}); + +describe("TerminalAttentionNotifier notify_on_terminal policy", () => { + function configWith(id: string, policy?: "notify_on_terminal"): ProjectsConfig { + return { + projects: new Map([ + [ + "/repo", + { + workspaces: [ + { + path: "/repo/ws", + id, + ...(policy != null ? { taskAttentionPolicy: policy } : {}), + }, + ], + }, + ], + ]), + }; + } + + test("writes the policy onto a task workspace entry", async () => { + const h = createHarness(); + h.config = configWith("task-1"); + await h.notifier.markBackgroundWorkNotifyOnTerminal("task-1", "owner"); + expect(h.config.projects.get("/repo")?.workspaces[0].taskAttentionPolicy).toBe( + "notify_on_terminal" + ); + expect(h.turnManager.markWorkspaceTurnBackgroundWorkNotifyOnTerminal).not.toHaveBeenCalled(); + }); + + test("leaves config unchanged for an unknown task", async () => { + const h = createHarness(); + h.config = configWith("task-1"); + await h.notifier.markBackgroundWorkNotifyOnTerminal("task-2", "owner"); + expect(h.config.projects.get("/repo")?.workspaces[0].taskAttentionPolicy).toBeUndefined(); + }); + + test("routes a workspace-turn handle to the turn manager, and skips it without an owner", async () => { + const h = createHarness(); + await h.notifier.markBackgroundWorkNotifyOnTerminal("wst_handle", "owner"); + h.notifier.scheduleNotifyOnTerminalPersist("wst_handle", undefined); + await Promise.all(h.notifier.inFlightNotifyOnTerminalPersists()); + expect(h.events).toEqual(["turn-policy:wst_handle:owner"]); + expect(h.editConfig).not.toHaveBeenCalled(); + }); + + test("tracks a scheduled persist until it settles, also when it fails", async () => { + const h = createHarness(); + const write = deferred(); + h.editConfig.mockImplementationOnce(() => write.promise); + h.notifier.scheduleNotifyOnTerminalPersist("task-1", "owner"); + const inFlight = h.notifier.inFlightNotifyOnTerminalPersists(); + expect(inFlight).toHaveLength(1); + + write.reject(new Error("disk full")); + await Promise.all(inFlight); + expect(h.notifier.inFlightNotifyOnTerminalPersists()).toHaveLength(0); + expect(logError).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/node/services/terminalAttentionNotifier.ts b/src/node/services/terminalAttentionNotifier.ts new file mode 100644 index 00000000000..a12c1e7a08c --- /dev/null +++ b/src/node/services/terminalAttentionNotifier.ts @@ -0,0 +1,295 @@ +import assert from "node:assert/strict"; +import type { Config } from "@/node/config"; +import type { AIService } from "@/node/services/aiService"; +import { log } from "@/node/services/log"; +import { isWorkspaceTurnTaskId } from "@/node/services/taskHandleStore"; +import { findWorkspaceEntry } from "@/node/services/taskUtils"; +import type { WorkspaceHost } from "@/node/services/taskWorkspaceSeam"; +import { + TerminalAttentionStore, + type TerminalAttentionNotification, + type TerminalAttentionOutcome, +} from "@/node/services/terminalAttentionStore"; +import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager"; + +// Level-triggered backstop for workflow terminal wakes: the sweep re-derives owed wakes from +// durable state (run records + settled markers), so lost terminal callbacks, deferred +// (transiently unreadable) evaluations, and crashes all recover here without any per-failure +// retry bookkeeping. +export const WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS = 5 * 60_000; + +/** + * Everything the notifier reads from TaskService. Each entry is a call-time closure over + * TaskService, never a captured value or bound method: tests swap TaskService fields and patch + * its methods (the harness wraps drainTerminalAttention) after construction. + */ +export interface TerminalAttentionNotifierDeps { + terminalAttentionStore: () => Pick< + TerminalAttentionStore, + "enqueueIfAbsent" | "markSuperseded" | "listPendingOwnerWorkspaceIds" + >; + config: () => Pick; + aiService: () => Pick; + workspaceService: () => Pick< + WorkspaceHost, + "getActiveTurnGeneration" | "waitForIdleAndNoQueuedMessages" + >; + turnManager: () => Pick; + /** TaskService.drainTerminalAttention, which stays on TaskService. */ + drain: (ownerWorkspaceId: string) => Promise; +} + +/** + * Scheduling and bookkeeping for terminal attention (workspace-turn / sub-agent / workflow + * terminal wake-ups): the durable notify_on_terminal policy writes, the per-owner drain chains, + * the idle waits, the fresh sub-agent edges and the wake-group send backoff. The drain itself + * stays on TaskService and runs through deps.drain. + */ +export class TerminalAttentionNotifier { + // In-flight durable persistence of notify_on_terminal policy for backgrounded foreground waits. + // Awaited at the start of handleStreamEnd so a just-detached wait is treated as non-blocking. + private readonly pendingNotifyOnTerminalPersists = new Set>(); + // In-flight terminal attention drains (workspace-turn / sub-agent terminal wake-ups). Tracked so + // tests and shutdown can await them; drains are idempotent and re-triggered on owner idle events. + readonly pendingTerminalAttentionDrainsByOwner = new Map>(); + readonly pendingTerminalAttentionDrains = new Set>(); + private readonly pendingTerminalAttentionIdleWaitsByOwner = new Map>(); + // Sub-agent terminal notifications this process just enqueued while their owner was streaming, + // mapped to that owner turn's generation (owner -> notification ID -> turn). The next drain takes + // the whole map, so each report can cut at most one busy turn, and only the turn that was live + // at its delivery (see cutBusyOwnerTurnForSubagentAttention). + private readonly freshSubagentAttentionTurnByOwner = new Map>(); + // Owner -> wake-group key -> epoch ms before which the drain skips selecting the group. A + // group whose send was rejected for group-specific reasons (for example an unresolvable + // strictly pinned agent) would otherwise be re-selected newest-first by every drain and + // sweep, starving older deliverable groups. In-memory only: a restart retries every group, + // and entries expire on read. + private readonly workflowWakeGroupSendBackoffUntilMs = new Map>(); + + constructor(private readonly deps: TerminalAttentionNotifierDeps) {} + + /** + * Persist `notify_on_terminal` on a backgrounded handle (agent task config or + * workspace-turn record). Tracked per workspace so `handleStreamEnd` can await + * any in-flight persistence before it reads config to decide blocking work. + */ + /** + * Durably mark a still-active task/handle as `notify_on_terminal`. Used when a foreground wait + * detaches because it exceeded its foreground wait budget (timeout) and the work continues in the + * background: like queued-message detachment, the work must not re-force the owner to await it. + * Awaited so callers (e.g. the task tool) can rely on the policy before returning pending results. + */ + async markBackgroundWorkNotifyOnTerminal( + taskId: string, + ownerWorkspaceId: string + ): Promise { + await this.persistNotifyOnTerminalPolicy(taskId, ownerWorkspaceId); + } + + scheduleNotifyOnTerminalPersist(taskId: string, ownerWorkspaceId: string | undefined): void { + const promise = this.persistNotifyOnTerminalPolicy(taskId, ownerWorkspaceId) + .catch((error: unknown) => { + log.error("Failed to persist notify_on_terminal policy for backgrounded wait", { + taskId, + error, + }); + }) + .finally(() => { + this.pendingNotifyOnTerminalPersists.delete(promise); + }); + this.pendingNotifyOnTerminalPersists.add(promise); + } + + /** The notify_on_terminal writes still in flight, for handleStreamEnd to await. */ + inFlightNotifyOnTerminalPersists(): Array> { + return [...this.pendingNotifyOnTerminalPersists]; + } + + private async persistNotifyOnTerminalPolicy( + taskId: string, + ownerWorkspaceId: string | undefined + ): Promise { + if (isWorkspaceTurnTaskId(taskId)) { + if (ownerWorkspaceId != null) { + await this.deps + .turnManager() + .markWorkspaceTurnBackgroundWorkNotifyOnTerminal(taskId, ownerWorkspaceId); + } + return; + } + await this.deps.config().editConfig((config) => { + const found = findWorkspaceEntry(config, taskId); + if (found != null && found.workspace.taskAttentionPolicy !== "notify_on_terminal") { + found.workspace.taskAttentionPolicy = "notify_on_terminal"; + } + return config; + }); + } + + /** + * Level-triggered retry for outbox (sub-agent / workspace-turn) attention: pending records + * are the durable "wake owed" state, but unlike workflow runs they have no periodic + * re-derivation of their own, so a drain that failed transiently (for example an unreadable + * history for caller restrictions) would otherwise leave them stuck until restart. Startup + * and the sweep interval both re-poke their owners; drains are idempotent and no-op when + * nothing is deliverable. + */ + async schedulePendingTerminalAttentionOwnerDrains(): Promise { + const ownerWorkspaceIds = await this.deps + .terminalAttentionStore() + .listPendingOwnerWorkspaceIds(); + for (const ownerWorkspaceId of ownerWorkspaceIds) { + this.scheduleTerminalAttentionDrain(ownerWorkspaceId); + } + return ownerWorkspaceIds.length; + } + + async suppressTerminalAttention(params: { + ownerWorkspaceId: string; + sourceKind: TerminalAttentionNotification["sourceKind"]; + sourceId: string; + }): Promise { + await this.deps.terminalAttentionStore().enqueueIfAbsent({ + ...params, + terminalOutcome: "interrupted", + }); + await this.deps + .terminalAttentionStore() + .markSuperseded( + params.ownerWorkspaceId, + TerminalAttentionStore.notificationId(params.sourceKind, params.sourceId) + ); + } + + async enqueueTerminalAttention(params: { + ownerWorkspaceId: string; + sourceKind: TerminalAttentionNotification["sourceKind"]; + terminalOutcome: TerminalAttentionOutcome; + sourceId: string; + generationId?: string; + }): Promise { + const created = await this.deps.terminalAttentionStore().enqueueIfAbsent(params); + if (created == null) { + return; + } + const liveTurn = this.deps.aiService().isStreaming(created.ownerWorkspaceId) + ? this.deps.workspaceService().getActiveTurnGeneration(created.ownerWorkspaceId) + : undefined; + if (created.sourceKind === "agent_task" && liveTurn != null) { + let fresh = this.freshSubagentAttentionTurnByOwner.get(created.ownerWorkspaceId); + if (fresh == null) { + fresh = new Map(); + this.freshSubagentAttentionTurnByOwner.set(created.ownerWorkspaceId, fresh); + } + fresh.set(created.id, liveTurn); + } + this.scheduleTerminalAttentionDrain(params.ownerWorkspaceId); + } + + /** + * Take and clear the owner's fresh sub-agent edges (see freshSubagentAttentionTurnByOwner). + * Synchronous, so the drain consumes them before its first await. + */ + takeFreshTurns(ownerWorkspaceId: string): Map { + const freshSubagentAttentionTurns = + this.freshSubagentAttentionTurnByOwner.get(ownerWorkspaceId) ?? new Map(); + this.freshSubagentAttentionTurnByOwner.delete(ownerWorkspaceId); + return freshSubagentAttentionTurns; + } + + scheduleTerminalAttentionDrain(ownerWorkspaceId: string): void { + const previous = this.pendingTerminalAttentionDrainsByOwner.get(ownerWorkspaceId); + const promise = (previous ?? Promise.resolve()) + .catch(() => undefined) + .then(() => this.deps.drain(ownerWorkspaceId)) + .catch((error: unknown) => { + log.error("Terminal attention drain failed", { ownerWorkspaceId, error }); + }) + .finally(() => { + this.pendingTerminalAttentionDrains.delete(promise); + if (this.pendingTerminalAttentionDrainsByOwner.get(ownerWorkspaceId) === promise) { + this.pendingTerminalAttentionDrainsByOwner.delete(ownerWorkspaceId); + } + }); + this.pendingTerminalAttentionDrainsByOwner.set(ownerWorkspaceId, promise); + this.pendingTerminalAttentionDrains.add(promise); + } + + scheduleTerminalAttentionDrainAfterIdle(ownerWorkspaceId: string): void { + // One idle wait per owner: each wait subscribes to the owner session's chat-event, and every + // report deferred by a busy owner lands here. The drain it schedules reads every pending + // notification from the store, so one wait delivers all of them, in store order. + if (this.pendingTerminalAttentionIdleWaitsByOwner.has(ownerWorkspaceId)) return; + const promise = this.deps + .workspaceService() + .waitForIdleAndNoQueuedMessages(ownerWorkspaceId) + .catch((error: unknown) => { + log.debug("Terminal attention idle wait failed; retrying drain anyway", { + ownerWorkspaceId, + error, + }); + }) + .then(() => { + // Release the slot before scheduling: that drain can find the owner busy again and must + // be able to install the next idle wait. + if (this.pendingTerminalAttentionIdleWaitsByOwner.get(ownerWorkspaceId) === promise) { + this.pendingTerminalAttentionIdleWaitsByOwner.delete(ownerWorkspaceId); + } + this.scheduleTerminalAttentionDrain(ownerWorkspaceId); + }) + .finally(() => { + this.pendingTerminalAttentionDrains.delete(promise); + }); + this.pendingTerminalAttentionIdleWaitsByOwner.set(ownerWorkspaceId, promise); + this.pendingTerminalAttentionDrains.add(promise); + } + + /** Track other terminal-attention work (deferred messages, reconciles) with the drains. */ + trackDrain(promise: Promise): void { + this.pendingTerminalAttentionDrains.add(promise); + } + + untrackDrain(promise: Promise): void { + this.pendingTerminalAttentionDrains.delete(promise); + } + + /** + * Whether the owner's send batch under `key` is still backed off (see + * backOffTerminalWakeSends). An expired entry is deleted on this read, and the owner's map + * with it once empty. + */ + isWakeSendBackedOff(ownerWorkspaceId: string, key: string): boolean { + const ownerBackoff = this.workflowWakeGroupSendBackoffUntilMs.get(ownerWorkspaceId); + const backoffUntil = ownerBackoff?.get(key); + if (backoffUntil == null) { + return false; + } + if (backoffUntil > Date.now()) { + return true; + } + ownerBackoff?.delete(key); + if (ownerBackoff?.size === 0) { + this.workflowWakeGroupSendBackoffUntilMs.delete(ownerWorkspaceId); + } + return false; + } + + /** + * Back the given terminal-wake send batches off until the sweep cadence retries them and + * re-poke the drain so the remaining batches get their send this cycle instead of starving + * behind the failed one. + */ + backOffTerminalWakeSends(ownerWorkspaceId: string, keys: readonly string[]): void { + assert(keys.length > 0, "backOffTerminalWakeSends requires keys"); + let ownerBackoff = this.workflowWakeGroupSendBackoffUntilMs.get(ownerWorkspaceId); + if (ownerBackoff == null) { + ownerBackoff = new Map(); + this.workflowWakeGroupSendBackoffUntilMs.set(ownerWorkspaceId, ownerBackoff); + } + const retryAt = Date.now() + WORKFLOW_TERMINAL_ATTENTION_SWEEP_INTERVAL_MS; + for (const key of keys) { + ownerBackoff.set(key, retryAt); + } + this.scheduleTerminalAttentionDrain(ownerWorkspaceId); + } +}