From be16e97fcd2b132a7b50cb3e98585129a3103db2 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Tue, 27 Oct 2026 07:25:52 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20move=20peer-send=20p?= =?UTF-8?q?arking=20into=20PeerSendParking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PeerSendParking owns the peer messages parked behind a root's delegated workspace turn (#4997), their per-target flush locks and the delegated-turn registration epochs (#5271). Move-only: the bodies of parkPeerSend, scheduleParkedPeerSendFlush, flushParkedPeerSends and onWorkspaceTurnRegistered are unchanged. TaskService keeps onWorkspaceTurnRegistered (TaskHost) and a private flushParkedPeerSends facade, because tests spy on it; the scheduled flush and the drain's sendTreeMessage call back through TaskService closures so spies see every call. sendTreeMessage, the broker's queue-cap count and the user-Stop drop use small accessors. --- src/node/services/peerSendParking.test.ts | 278 ++++++++++++++++++ src/node/services/peerSendParking.ts | 176 +++++++++++ .../taskService.peerLimitsRepro.test.ts | 6 +- src/node/services/taskService.ts | 151 ++-------- 4 files changed, 484 insertions(+), 127 deletions(-) create mode 100644 src/node/services/peerSendParking.test.ts create mode 100644 src/node/services/peerSendParking.ts diff --git a/src/node/services/peerSendParking.test.ts b/src/node/services/peerSendParking.test.ts new file mode 100644 index 00000000000..a9624ec1426 --- /dev/null +++ b/src/node/services/peerSendParking.test.ts @@ -0,0 +1,278 @@ +import { describe, expect, spyOn, test } from "bun:test"; + +import { Err, Ok } from "@/common/types/result"; +import { log } from "@/node/services/log"; +import { + PeerSendParking, + type ParkedPeerSend, + type PeerSendParkingDeps, +} from "@/node/services/peerSendParking"; +import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager"; + +/** Only its presence matters to the parking: a live registration holds the list. */ +const LIVE_REGISTRATION: NonNullable< + ReturnType +> = { handleId: "handle", ownerWorkspaceId: "owner", accepted: true }; + +const TARGET = "target"; + +function macrotask(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +function parkedSend(message: string, targetId = TARGET): ParkedPeerSend { + return { + relation: "peer", + senderWorkspaceId: `sender-${message}`, + targetId, + message, + targetRelation: "peer", + awaitedDelegatedTurn: { + unrelatedConsent: undefined, + registrationEpoch: 0, + stopEpochs: new Map([[TARGET, -1]]), + }, + }; +} + +function createHarness() { + const live = new Set(); + const stopping = new Set(); + const stopEpochs = new Map(); + const flushes: Array> = []; + const delivered: string[] = []; + const stopEpochReads: string[] = []; + let send: PeerSendParkingDeps["sendTreeMessage"] = (spec) => { + delivered.push(spec.message); + return Promise.resolve(Ok(undefined)); + }; + const parking: PeerSendParking = new PeerSendParking({ + flush: (targetId) => { + const flush = parking.flushParkedPeerSends(targetId); + flushes.push(flush); + return flush; + }, + getWorkspaceTurnManager: () => ({ + getLiveWorkspaceTurnRegistration: (workspaceId: string) => + live.has(workspaceId) ? LIVE_REGISTRATION : undefined, + }), + isWorkspaceStopInProgress: (workspaceId) => stopping.has(workspaceId), + getWorkspaceStopEpoch: (workspaceId) => { + stopEpochReads.push(workspaceId); + return stopEpochs.get(workspaceId) ?? 0; + }, + sendTreeMessage: (spec) => send(spec), + }); + return { + parking, + live, + stopping, + stopEpochs, + delivered, + stopEpochReads, + flushes, + setSend(next: PeerSendParkingDeps["sendTreeMessage"]) { + send = next; + }, + /** + * Wait until every scheduled flush settled, also ones scheduled meanwhile. The lock starts a + * flush after a microtask hop and the fake deps settle on microtasks, so one macrotask turn + * lets every started chain run. + */ + async settle() { + let seen = -1; + while (seen !== flushes.length) { + seen = flushes.length; + await macrotask(); + await Promise.all(flushes); + } + }, + }; +} + +describe("PeerSendParking", () => { + test("registration epochs start at zero and count each registration per target", () => { + const { parking } = createHarness(); + expect(parking.registrationEpoch(TARGET)).toBe(0); + parking.onWorkspaceTurnRegistered(TARGET); + parking.onWorkspaceTurnRegistered(TARGET); + parking.onWorkspaceTurnRegistered("other"); + expect(parking.registrationEpoch(TARGET)).toBe(2); + expect(parking.registrationEpoch("other")).toBe(1); + }); + + test("a scheduled flush with nothing parked never reaches the flush dep", () => { + const h = createHarness(); + h.parking.scheduleParkedPeerSendFlush(TARGET); + expect(h.flushes).toHaveLength(0); + }); + + test("parking schedules a flush through the dep that delivers in park order and unregisters", async () => { + const h = createHarness(); + h.live.add(TARGET); + h.parking.parkPeerSend(parkedSend("a")); + h.parking.parkPeerSend(parkedSend("b")); + await h.settle(); + // The live turn holds the list: nothing delivered, both still counted against the cap. + expect(h.delivered).toEqual([]); + expect(h.parking.countParked(TARGET)).toBe(2); + expect(h.parking.hasParked(TARGET)).toBe(true); + + h.live.delete(TARGET); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(h.delivered).toEqual(["a", "b"]); + expect(h.parking.hasParked(TARGET)).toBe(false); + expect(h.parking.countParked(TARGET)).toBe(0); + }); + + test("a stop in progress holds the list until a later flush", async () => { + const h = createHarness(); + h.stopping.add(TARGET); + h.parking.parkPeerSend(parkedSend("a")); + await h.settle(); + expect(h.delivered).toEqual([]); + expect(h.stopEpochReads).toEqual([]); + + h.stopping.delete(TARGET); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(h.delivered).toEqual(["a"]); + }); + + test("the drain takes one new target stop baseline for every waiting message", async () => { + const h = createHarness(); + h.live.add(TARGET); + const first = parkedSend("a"); + const second = parkedSend("b"); + h.parking.parkPeerSend(first); + h.parking.parkPeerSend(second); + await h.settle(); + h.live.delete(TARGET); + h.stopEpochs.set(TARGET, 7); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(h.stopEpochReads).toEqual([TARGET]); + expect(first.awaitedDelegatedTurn.stopEpochs.get(TARGET)).toBe(7); + expect(second.awaitedDelegatedTurn.stopEpochs.get(TARGET)).toBe(7); + }); + + test("a refused or throwing retry is dropped once and the rest still deliver", async () => { + const h = createHarness(); + const debug = spyOn(log, "debug").mockImplementation(() => undefined); + const error = spyOn(log, "error").mockImplementation(() => undefined); + try { + const attempts: string[] = []; + h.setSend((spec) => { + attempts.push(spec.message); + if (spec.message === "refused") return Promise.resolve(Err({ code: "interrupted" })); + if (spec.message === "throws") return Promise.reject(new Error("boom")); + return Promise.resolve(Ok(undefined)); + }); + h.live.add(TARGET); + h.parking.parkPeerSend(parkedSend("refused")); + h.parking.parkPeerSend(parkedSend("throws")); + h.parking.parkPeerSend(parkedSend("ok")); + await h.settle(); + h.live.delete(TARGET); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(attempts).toEqual(["refused", "throws", "ok"]); + expect(debug).toHaveBeenCalledTimes(1); + expect(error).toHaveBeenCalledTimes(1); + expect(h.parking.hasParked(TARGET)).toBe(false); + } finally { + debug.mockRestore(); + error.mockRestore(); + } + }); + + test("a retry that removed itself is not shifted twice", async () => { + const h = createHarness(); + h.setSend((spec) => { + h.delivered.push(spec.message); + // sendTreeMessage removes the retry under the admission lock. + h.parking.removeRetry(spec.targetId, spec); + return Promise.resolve(Ok(undefined)); + }); + h.live.add(TARGET); + h.parking.parkPeerSend(parkedSend("a")); + h.parking.parkPeerSend(parkedSend("b")); + await h.settle(); + h.live.delete(TARGET); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(h.delivered).toEqual(["a", "b"]); + }); + + test("removeRetry ignores a spec that is not the head", () => { + const h = createHarness(); + h.live.add(TARGET); + const head = parkedSend("a"); + h.parking.parkPeerSend(head); + h.parking.parkPeerSend(parkedSend("b")); + h.parking.removeRetry(TARGET, parkedSend("a")); + expect(h.parking.countParked(TARGET)).toBe(2); + h.parking.removeRetry(TARGET, head); + expect(h.parking.countParked(TARGET)).toBe(1); + h.parking.removeRetry("other", head); + expect(h.parking.countParked(TARGET)).toBe(1); + }); + + test("dropping the list mid-drain ends the drain", async () => { + const h = createHarness(); + h.setSend((spec) => { + h.delivered.push(spec.message); + expect(h.parking.dropParked(TARGET)).toBe(true); + return Promise.resolve(Ok(undefined)); + }); + h.live.add(TARGET); + h.parking.parkPeerSend(parkedSend("a")); + h.parking.parkPeerSend(parkedSend("b")); + await h.settle(); + h.live.delete(TARGET); + h.parking.scheduleParkedPeerSendFlush(TARGET); + await h.settle(); + expect(h.delivered).toEqual(["a"]); + expect(h.parking.dropParked(TARGET)).toBe(false); + }); + + test("a send parked during a drain is delivered after the waiting ones, one at a time", async () => { + const h = createHarness(); + let active = 0; + let maxActive = 0; + h.setSend(async (spec) => { + active++; + maxActive = Math.max(maxActive, active); + if (spec.message === "a") h.parking.parkPeerSend(parkedSend("fresh")); + await Promise.resolve(); + h.delivered.push(spec.message); + active--; + return Ok(undefined); + }); + h.parking.parkPeerSend(parkedSend("a")); + h.parking.parkPeerSend(parkedSend("b")); + await h.settle(); + expect(h.delivered).toEqual(["a", "b", "fresh"]); + expect(maxActive).toBe(1); + }); + + test("a failing scheduled flush is logged, not thrown", async () => { + const error = spyOn(log, "error").mockImplementation(() => undefined); + try { + const parking = new PeerSendParking({ + flush: () => Promise.reject(new Error("flush failed")), + getWorkspaceTurnManager: () => ({ getLiveWorkspaceTurnRegistration: () => undefined }), + isWorkspaceStopInProgress: () => false, + getWorkspaceStopEpoch: () => 0, + sendTreeMessage: () => Promise.resolve(Ok(undefined)), + }); + parking.parkPeerSend(parkedSend("a")); + await macrotask(); + expect(error).toHaveBeenCalledTimes(1); + expect(parking.countParked(TARGET)).toBe(1); + } finally { + error.mockRestore(); + } + }); +}); diff --git a/src/node/services/peerSendParking.ts b/src/node/services/peerSendParking.ts new file mode 100644 index 00000000000..28b18328b9a --- /dev/null +++ b/src/node/services/peerSendParking.ts @@ -0,0 +1,176 @@ +import type { Result } from "@/common/types/result"; +import { getErrorMessage } from "@/common/utils/errors"; +import { log } from "@/node/services/log"; +import type { PeerDelegatedTurnWait, TreeMessageSpec } from "@/node/services/taskService"; +import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager"; +import { MutexMap } from "@/node/utils/concurrency/mutexMap"; + +type PeerTreeMessageSpec = Extract; + +export type ParkedPeerSend = PeerTreeMessageSpec & { + awaitedDelegatedTurn: PeerDelegatedTurnWait; +}; + +/** + * Call-time closures into TaskService. Tests spy on TaskService.flushParkedPeerSends and + * TaskService.sendTreeMessage and swap TaskService fields, so every call goes back through it. + */ +export interface PeerSendParkingDeps { + /** TaskService.flushParkedPeerSends, which delegates back to flushParkedPeerSends here. */ + flush: (targetId: string) => Promise; + getWorkspaceTurnManager: () => Pick; + isWorkspaceStopInProgress: (workspaceId: string) => boolean; + getWorkspaceStopEpoch: (workspaceId: string) => number; + sendTreeMessage: (spec: ParkedPeerSend) => Promise>; +} + +/** Peer messages parked behind a root's delegated workspace turn (#4997, #5271). */ +export class PeerSendParking { + /** + * #4997: peer messages to a root whose running delegated workspace turn the sender does not + * own, in arrival order per target. The delegated turn stays owner-only, so they wait here and + * are delivered as ordinary new turns once its live registration is released (see + * flushParkedPeerSends). They wait outside the target's MessageQueue on purpose: a queued + * uncorrelated entry would supersede the delegated turn or strip the owner's continuations + * (MessageQueue.hasAllWorkspaceTurnContinuations), and a "hold" dispatch decision blocks every + * entry behind it, including the continuations that turn is waiting for. In-memory and + * process-local, like the queue. + */ + readonly parkedPeerSendsByTarget = new Map(); + private readonly parkedPeerSendFlushLocks = new MutexMap(); + /** #5271: new delegated-turn registrations per target (see onWorkspaceTurnRegistered). */ + readonly workspaceTurnRegistrationEpochs = new Map(); + + constructor(private readonly deps: PeerSendParkingDeps) {} + + /** + * WorkspaceTurnTaskHost: a new delegated turn registered (#5271). A message waits once, for the + * turn it was parked behind. Counting registrations, like the stop epochs, lets the retry's + * admission gate see a replacement turn even after it settled, at any point before admission. + */ + // dry-duplicate-ok: a per-key counter bump like the context-mutation and MCP option epochs, but + // each counts a different event and changes apart; moved here from TaskService unchanged. + onWorkspaceTurnRegistered(workspaceId: string): void { + this.workspaceTurnRegistrationEpochs.set( + workspaceId, + // mutation-equivalent: 0 || 0 and 0 ?? 0 are both 0, so || and ?? agree here + (this.workspaceTurnRegistrationEpochs.get(workspaceId) ?? 0) + 1 + ); + } + + /** The target's delegated-turn registration count (0 before the first registration). */ + registrationEpoch(targetId: string): number { + // mutation-equivalent: stored epochs start at 1, so a stored value is never 0 and || agrees + return this.workspaceTurnRegistrationEpochs.get(targetId) ?? 0; + } + + /** Parked sends count against the recipient's queue cap like queued ones. */ + countParked(targetId: string): number { + // mutation-equivalent: a length of 0 falls back to 0 too, so || and ?? agree + return this.parkedPeerSendsByTarget.get(targetId)?.length ?? 0; + } + + /** True while the target has a parked list, also while it drains. */ + hasParked(targetId: string): boolean { + return this.parkedPeerSendsByTarget.has(targetId); + } + + /** A retry leaves the parked list when it is still the head (see sendTreeMessage). */ + removeRetry(targetId: string, spec: PeerTreeMessageSpec): void { + const parked = this.parkedPeerSendsByTarget.get(targetId); + if (parked?.[0] === spec) parked.shift(); + } + + /** Drop every parked send for the target; true when a list existed. */ + dropParked(targetId: string): boolean { + return this.parkedPeerSendsByTarget.delete(targetId); + } + + /** + * Park a peer message behind the target's delegated turn, then schedule a flush. Park first, + * check second: the registration may have been released between the caller's synchronous + * check and this call, and that release found nothing to flush. Only first attempts park: a + * message that already waited is delivered once or dropped. + */ + parkPeerSend(send: ParkedPeerSend): void { + // mutation-equivalent: a stored list is an array, which is never falsy, so || and ?? agree + const parked = this.parkedPeerSendsByTarget.get(send.targetId) ?? []; + parked.push(send); + this.parkedPeerSendsByTarget.set(send.targetId, parked); + this.scheduleParkedPeerSendFlush(send.targetId); + } + + /** + * Triggered by the delegated turn's registration release and by the stop-latch release (an + * explicit interrupt releases the registration while its latch is still held), never by a + * timer. Flushes of one target run one at a time, in park order. + */ + scheduleParkedPeerSendFlush(targetId: string): void { + if (!this.parkedPeerSendsByTarget.has(targetId)) return; + this.parkedPeerSendFlushLocks + .withLock(targetId, () => this.deps.flush(targetId)) + .catch((error: unknown) => { + log.error("Failed to deliver peer messages that waited for a delegated turn", { + targetId, + error: getErrorMessage(error), + }); + }); + } + + async flushParkedPeerSends(targetId: string): Promise { + // While a registration or stop latch is held, stop: its release schedules the next flush. + // The list stays registered while it drains, so a fresh send parks behind it instead of + // overtaking the rest (see sendTreeMessage), and a user Stop that drops it ends the drain. + const parked = this.parkedPeerSendsByTarget.get(targetId); + let drainStarted = false; + while (parked != null && this.parkedPeerSendsByTarget.get(targetId) === parked) { + // Messages that waited for an earlier turn are dropped by their retry's admission gate. + if ( + this.deps.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(targetId) != null || + this.deps.isWorkspaceStopInProgress(targetId) + ) { + return; + } + if (!drainStarted) { + drainStarted = true; + // The owner's explicit interrupt of the delegated turn bumps the target's stop epoch (so + // sends admitted during its wind-down cannot outlive that stop), and these messages + // waited for the turn on purpose. Take a new target baseline once, here, not per retry: + // a user Stop after this point still refuses a retry that is already in flight, and one + // before it dropped the list (markParentWorkspaceInterrupted). + const epoch = this.deps.getWorkspaceStopEpoch(targetId); + for (const waiting of parked) waiting.awaitedDelegatedTurn.stopEpochs.set(targetId, epoch); + } + // The retry removes itself from the list under the target's admission lock (see + // sendTreeMessage), so it stays counted against the queue cap until then. + const spec = parked[0]; + if (spec == null) { + this.parkedPeerSendsByTarget.delete(targetId); + return; + } + // The retry re-runs the full peer path under the target's event lock. A withdrawn message + // (consent revoked, sender stopped, runtime change, target archived) is dropped and + // touches no turn state, and so is one that meets another delegated turn. + // Each message is handled on its own, so one failure cannot discard the rest. + try { + const result = await this.deps.sendTreeMessage(spec); + if (!result.success) { + log.debug("Dropped a peer message that waited for a delegated turn", { + senderWorkspaceId: spec.senderWorkspaceId, + targetId, + code: result.error.code, + }); + } + } catch (error) { + log.error("Failed to deliver a peer message that waited for a delegated turn", { + senderWorkspaceId: spec.senderWorkspaceId, + targetId, + error: getErrorMessage(error), + }); + } finally { + // A retry that never reached the lock (it threw first) must not be retried forever. + if (parked[0] === spec) parked.shift(); + } + } + } +} diff --git a/src/node/services/taskService.peerLimitsRepro.test.ts b/src/node/services/taskService.peerLimitsRepro.test.ts index 0ab8e2d64b2..bdd5659846f 100644 --- a/src/node/services/taskService.peerLimitsRepro.test.ts +++ b/src/node/services/taskService.peerLimitsRepro.test.ts @@ -310,8 +310,10 @@ describe("peer-message limits", () => { await fresh; // Let the drain deliver every waiting message (it unregisters the list once empty). const parkedByTarget = ( - taskService as unknown as { parkedPeerSendsByTarget: Map } - ).parkedPeerSendsByTarget; + taskService as unknown as { + peerSendParking: { parkedPeerSendsByTarget: Map }; + } + ).peerSendParking.parkedPeerSendsByTarget; for (let spin = 0; spin < 200 && parkedByTarget.has("target"); spin++) { await new Promise((resolve) => setTimeout(resolve, 10)); } diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index cf7d5de1f50..7e4b84f0105 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -110,6 +110,7 @@ import { TASK_FAMILY_MESSAGE_MAX_CHARS, } from "@/constants/taskMessages"; import { log } from "@/node/services/log"; +import { PeerSendParking } from "@/node/services/peerSendParking"; import { isProjectTrusted, isWorkspaceProjectTrustedIn } from "@/node/utils/projectTrust"; import { eventSpine } from "@/node/services/events/eventSpine"; import { sandboxHostService } from "@/node/services/sandbox/sandboxHostService"; @@ -1125,7 +1126,7 @@ function isInactiveAgentTaskStatus( /** After this many idle waits, a deferred message falls back to plain live delivery. */ const MAX_DEFERRED_DESCENDANT_MESSAGE_WAITS = 5; -type TreeMessageSpec = +export type TreeMessageSpec = | { relation: "descendant"; senderWorkspaceId: string; @@ -1169,7 +1170,7 @@ type TreeMessagePipelineError = SendAgentTreeMessageError | SendParentAgentMessa * What a peer message's first admission established, kept while it waits for the target's * delegated turn (#4997). Its retry re-runs the whole peer path against these baselines. */ -interface PeerDelegatedTurnWait { +export interface PeerDelegatedTurnWait { /** Recipient consent at first admission: an off/on cycle must not revive the message. */ unrelatedConsent: string | undefined; /** Both endpoints' tag generations at first admission: a clear/re-tag cycle is a new link. */ @@ -1187,10 +1188,6 @@ interface PeerDelegatedTurnWait { stopEpochs: Map; } -type ParkedPeerSend = Extract & { - awaitedDelegatedTurn: PeerDelegatedTurnWait; -}; - /** The caller-relative relationship tag on task_list scope:"tree" rows. */ export type TreeAgentRelationship = "self" | "ancestor" | "sibling" | "descendant"; @@ -2033,20 +2030,7 @@ export class TaskService implements AgentTaskIntegration { * flushParkedPeerSends), so parking needs this separate count to tell a user Stop apart. */ private readonly workspaceUserStopEpochs = new Map(); - /** - * #4997: peer messages to a root whose running delegated workspace turn the sender does not - * own, in arrival order per target. The delegated turn stays owner-only, so they wait here and - * are delivered as ordinary new turns once its live registration is released (see - * flushParkedPeerSends). They wait outside the target's MessageQueue on purpose: a queued - * uncorrelated entry would supersede the delegated turn or strip the owner's continuations - * (MessageQueue.hasAllWorkspaceTurnContinuations), and a "hold" dispatch decision blocks every - * entry behind it, including the continuations that turn is waiting for. In-memory and - * process-local, like the queue. - */ - private readonly parkedPeerSendsByTarget = new Map(); - private readonly parkedPeerSendFlushLocks = new MutexMap(); - /** #5271: new delegated-turn registrations per target (see onWorkspaceTurnRegistered). */ - private readonly workspaceTurnRegistrationEpochs = new Map(); + private readonly peerSendParking: PeerSendParking; /** * Child stream ends cut by a host-selected continuation (queued input or context-budget * hand-over) whose successor has not yet streamed or been withdrawn. Keyed by task; settled @@ -2452,106 +2436,17 @@ export class TaskService implements AgentTaskIntegration { /** WorkspaceTurnTaskHost: a delegated turn's registration release (#4997). */ onWorkspaceTurnRegistrationReleased(workspaceId: string): void { - this.scheduleParkedPeerSendFlush(workspaceId); + this.peerSendParking.scheduleParkedPeerSendFlush(workspaceId); } - /** - * WorkspaceTurnTaskHost: a new delegated turn registered (#5271). A message waits once, for the - * turn it was parked behind. Counting registrations, like the stop epochs, lets the retry's - * admission gate see a replacement turn even after it settled, at any point before admission. - */ + /** WorkspaceTurnTaskHost: a new delegated turn registered (#5271). */ onWorkspaceTurnRegistered(workspaceId: string): void { - this.workspaceTurnRegistrationEpochs.set( - workspaceId, - (this.workspaceTurnRegistrationEpochs.get(workspaceId) ?? 0) + 1 - ); + this.peerSendParking.onWorkspaceTurnRegistered(workspaceId); } - /** - * Park a peer message behind the target's delegated turn, then schedule a flush. Park first, - * check second: the registration may have been released between the caller's synchronous - * check and this call, and that release found nothing to flush. Only first attempts park: a - * message that already waited is delivered once or dropped. - */ - private parkPeerSend(send: ParkedPeerSend): void { - const parked = this.parkedPeerSendsByTarget.get(send.targetId) ?? []; - parked.push(send); - this.parkedPeerSendsByTarget.set(send.targetId, parked); - this.scheduleParkedPeerSendFlush(send.targetId); - } - - /** - * Triggered by the delegated turn's registration release and by the stop-latch release (an - * explicit interrupt releases the registration while its latch is still held), never by a - * timer. Flushes of one target run one at a time, in park order. - */ - private scheduleParkedPeerSendFlush(targetId: string): void { - if (!this.parkedPeerSendsByTarget.has(targetId)) return; - this.parkedPeerSendFlushLocks - .withLock(targetId, () => this.flushParkedPeerSends(targetId)) - .catch((error: unknown) => { - log.error("Failed to deliver peer messages that waited for a delegated turn", { - targetId, - error: getErrorMessage(error), - }); - }); - } - - private async flushParkedPeerSends(targetId: string): Promise { - // While a registration or stop latch is held, stop: its release schedules the next flush. - // The list stays registered while it drains, so a fresh send parks behind it instead of - // overtaking the rest (see sendTreeMessage), and a user Stop that drops it ends the drain. - const parked = this.parkedPeerSendsByTarget.get(targetId); - let drainStarted = false; - while (parked != null && this.parkedPeerSendsByTarget.get(targetId) === parked) { - // Messages that waited for an earlier turn are dropped by their retry's admission gate. - if ( - this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(targetId) != null || - this.isWorkspaceStopInProgress(targetId) - ) { - return; - } - if (!drainStarted) { - drainStarted = true; - // The owner's explicit interrupt of the delegated turn bumps the target's stop epoch (so - // sends admitted during its wind-down cannot outlive that stop), and these messages - // waited for the turn on purpose. Take a new target baseline once, here, not per retry: - // a user Stop after this point still refuses a retry that is already in flight, and one - // before it dropped the list (markParentWorkspaceInterrupted). - const epoch = this.getWorkspaceStopEpoch(targetId); - for (const waiting of parked) waiting.awaitedDelegatedTurn.stopEpochs.set(targetId, epoch); - } - // The retry removes itself from the list under the target's admission lock (see - // sendTreeMessage), so it stays counted against the queue cap until then. - const spec = parked[0]; - if (spec == null) { - this.parkedPeerSendsByTarget.delete(targetId); - return; - } - // The retry re-runs the full peer path under the target's event lock. A withdrawn message - // (consent revoked, sender stopped, runtime change, target archived) is dropped and - // touches no turn state, and so is one that meets another delegated turn. - // Each message is handled on its own, so one failure cannot discard the rest. - try { - const result = await this.sendTreeMessage(spec); - if (!result.success) { - log.debug("Dropped a peer message that waited for a delegated turn", { - senderWorkspaceId: spec.senderWorkspaceId, - targetId, - code: result.error.code, - }); - } - } catch (error) { - log.error("Failed to deliver a peer message that waited for a delegated turn", { - senderWorkspaceId: spec.senderWorkspaceId, - targetId, - error: getErrorMessage(error), - }); - } finally { - // A retry that never reached the lock (it threw first) must not be retried forever. - if (parked[0] === spec) parked.shift(); - } - } + /** Tests spy on this; PeerSendParking's scheduled flush calls back through it. */ + private flushParkedPeerSends(targetId: string): Promise { + return this.peerSendParking.flushParkedPeerSends(targetId); } /** @@ -2573,7 +2468,7 @@ export class TaskService implements AgentTaskIntegration { } this.notifyAttemptSettlementListeners(id); // An interrupted delegated turn releases its registration while this latch is held. - this.scheduleParkedPeerSendFlush(id); + this.peerSendParking.scheduleParkedPeerSendFlush(id); } else { this.workspaceStopsInProgress.set(id, count - 1); } @@ -4822,11 +4717,18 @@ export class TaskService implements AgentTaskIntegration { terminalAttentionStore?: TerminalAttentionStore, private readonly desktopInputCoordinator = new DesktopInputCoordinator(config) ) { + this.peerSendParking = new PeerSendParking({ + flush: (targetId) => this.flushParkedPeerSends(targetId), + getWorkspaceTurnManager: () => this.getWorkspaceTurnManager(), + isWorkspaceStopInProgress: (workspaceId) => this.isWorkspaceStopInProgress(workspaceId), + getWorkspaceStopEpoch: (workspaceId) => this.getWorkspaceStopEpoch(workspaceId), + sendTreeMessage: (spec) => this.sendTreeMessage(spec), + }); this.agentPeerMessageBroker = new AgentPeerMessageBroker({ // Parked sends count against the recipient's queue cap like queued ones. countQueuedAgentPeerMessages: (targetId) => workspaceService.countQueuedAgentPeerMessages(targetId) + - (this.parkedPeerSendsByTarget.get(targetId)?.length ?? 0), + this.peerSendParking.countParked(targetId), }); this.terminalAttentionStore = terminalAttentionStore ?? new TerminalAttentionStore(config); this.workflowRunAttention = new WorkflowRunAttention({ @@ -10398,8 +10300,7 @@ export class TaskService implements AgentTaskIntegration { // A retry leaves the parked list only here, under the target's admission lock: removed // earlier, it was counted nowhere while it waited for the lock, and a fresh send // admitted meanwhile could push the queue past its cap (see flushParkedPeerSends). - const parked = this.parkedPeerSendsByTarget.get(targetId); - if (parked?.[0] === spec) parked.shift(); + this.peerSendParking.removeRetry(targetId, spec); } const cfg = this.config.loadConfigOrDefault(); const targetEntry = findWorkspaceEntry(cfg, targetId); @@ -10545,7 +10446,7 @@ export class TaskService implements AgentTaskIntegration { // settled again before this check (#5271). if ( awaitedDelegatedTurn != null && - (this.workspaceTurnRegistrationEpochs.get(targetId) ?? 0) !== + this.peerSendParking.registrationEpoch(targetId) !== awaitedDelegatedTurn.registrationEpoch ) { return awaitDelegatedTurn; @@ -10884,7 +10785,7 @@ export class TaskService implements AgentTaskIntegration { const delegatedRootRefusal = getDelegatedRootRefusal(); if (delegatedRootRefusal != null) { if (delegatedRootRefusal === awaitDelegatedTurn) { - refusedAtRegistrationEpoch ??= this.workspaceTurnRegistrationEpochs.get(targetId) ?? 0; + refusedAtRegistrationEpoch ??= this.peerSendParking.registrationEpoch(targetId); } admissionRefusal = delegatedRootRefusal; return true; @@ -11009,13 +10910,13 @@ export class TaskService implements AgentTaskIntegration { return; } parkedThisAttempt = true; - this.parkPeerSend({ + this.peerSendParking.parkPeerSend({ ...spec, awaitedDelegatedTurn: { unrelatedConsent, tagGenerations, registrationEpoch: - refusedAtRegistrationEpoch ?? this.workspaceTurnRegistrationEpochs.get(targetId) ?? 0, + refusedAtRegistrationEpoch ?? this.peerSendParking.registrationEpoch(targetId), stopEpochs: new Map(capturedStopEpochs), }, }); @@ -11037,7 +10938,7 @@ export class TaskService implements AgentTaskIntegration { if ( firstAttempt && !targetIsAgentTask && - this.parkedPeerSendsByTarget.has(targetId) && + this.peerSendParking.hasParked(targetId) && this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(targetId) == null ) { return waitForDelegatedTurn(); @@ -17141,7 +17042,7 @@ export class TaskService implements AgentTaskIntegration { // A user Stop refuses agent messages that were not admitted yet, including those waiting // for a delegated turn: their retry takes a new baseline for the target's stop epoch (see // sendTreeMessage), so drop them here rather than let them run after a resume. - if (this.parkedPeerSendsByTarget.delete(workspaceId)) { + if (this.peerSendParking.dropParked(workspaceId)) { log.debug("Dropped peer messages waiting for a delegated turn after a user stop", { workspaceId, });