From e1120e5abbb79eee180608e3b5a8bb947f38b6bc Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 12 Sep 2026 11:41:20 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20centralize=20prepare?= =?UTF-8?q?d=20history=20publication=20and=20origin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ --- .../agentSession.preparedHistory.test.ts | 215 ++++++++++++++++++ src/node/services/agentSession.ts | 108 +++++---- src/node/services/messageQueue.test.ts | 35 +++ src/node/services/messageQueue.ts | 32 ++- src/node/services/taskService.test.ts | 74 +++--- src/node/services/taskService.ts | 50 +++- src/node/services/taskWorkspaceSeam.ts | 10 +- src/node/services/turnRequestBuilder.ts | 3 + src/node/services/workflowContinuation.ts | 1 + .../services/workspaceGoalService.test.ts | 34 +-- src/node/services/workspaceService.ts | 17 +- src/node/services/workspaceTurnManager.ts | 1 + 12 files changed, 484 insertions(+), 96 deletions(-) create mode 100644 src/node/services/agentSession.preparedHistory.test.ts diff --git a/src/node/services/agentSession.preparedHistory.test.ts b/src/node/services/agentSession.preparedHistory.test.ts new file mode 100644 index 00000000000..0094d50d7bb --- /dev/null +++ b/src/node/services/agentSession.preparedHistory.test.ts @@ -0,0 +1,215 @@ +import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"; +import { createMuxMessage, type MuxMessage } from "@/common/types/message"; +import { Err, Ok } from "@/common/types/result"; +import assert from "@/common/utils/assert"; +import type { CompactionMonitor } from "./compactionMonitor"; +import type { TurnAcceptanceOrigin } from "./taskWorkspaceSeam"; +import { + createAgentSessionHarness, + createStartedTurnHandle, + type AgentSessionHarness, +} from "./agentSession.testHarness"; + +const options = { model: "openai:gpt-4o", agentId: "exec" }; +const workspaceId = "prepared-history"; +const fixtures: AgentSessionHarness[] = []; + +interface PreparationInputs { + materializeFileAtMentionsSnapshot(): Promise<{ + snapshotMessage: MuxMessage; + materializedTokens: string[]; + fileStates: []; + }>; + materializeAgentSkillSnapshots(): Promise; + materializeMcpPromptSnapshots(metadata: unknown, invokingId: string): Promise; + compactionMonitor: CompactionMonitor; +} + +async function fixture() { + const h = await createAgentSessionHarness({ workspaceId }); + fixtures.push(h); + const inputs = h.session as unknown as PreparationInputs; + const file = createMuxMessage("file", "user", "file content", { + synthetic: true, + fileAtMentionSnapshot: ["@input.ts"], + }); + const skill = createMuxMessage("skill", "user", "skill content", { + synthetic: true, + agentSkillSnapshot: { skillName: "review", scope: "project", sha256: "skill-hash" }, + }); + spyOn(inputs, "materializeFileAtMentionsSnapshot").mockResolvedValue({ + snapshotMessage: file, + materializedTokens: ["@input.ts"], + fileStates: [], + }); + const skills = spyOn(inputs, "materializeAgentSkillSnapshots").mockResolvedValue([skill]); + const prompts = spyOn(inputs, "materializeMcpPromptSnapshots").mockImplementation( + (_metadata, invokingId) => + Promise.resolve([ + createMuxMessage("prompt", "user", "prompt content", { + synthetic: true, + mcpPromptSnapshot: { + serverName: "server", + promptName: "review", + commandKey: "review", + invokingMessageId: invokingId, + }, + }), + ]) + ); + const rows = async () => { + const result = await h.historyService.getHistoryFromLatestBoundary(workspaceId); + assert(result.success); + return result.data; + }; + return { ...h, inputs, skills, prompts, rows }; +} + +afterEach(async () => { + for (const h of fixtures.splice(0)) { + await h.session.dispose(); + await h.cleanup(); + } + mock.restore(); +}); + +describe("prepared history publication", () => { + test.each([false, true])( + "keeps prefix order before the trigger (pre-turn batch=%s)", + async (batch) => { + const h = await fixture(); + const payload = createMuxMessage("payload", "assistant", "delegated input", { + synthetic: true, + }); + const started = spyOn(h.aiService, "streamMessage").mockImplementation(async () => { + const rows = await h.rows(); + expect(rows.slice(0, -1).map((row) => row.id)).toEqual( + batch ? ["file", "skill", "prompt", "payload"] : ["file", "skill", "prompt"] + ); + expect(rows.at(-1)?.parts).toMatchObject([{ type: "text", text: "inspect input" }]); + expect(rows.map((row) => row.metadata?.historySequence)).toEqual( + rows.map((_row, index) => index) + ); + return Ok(createStartedTurnHandle(h.session.closingSignal)); + }); + expect( + await h.session.sendMessage("inspect input", options, { + ...(batch ? { preTurnMessages: [payload] } : {}), + }) + ).toEqual(Ok(undefined)); + expect(started).toHaveBeenCalledTimes(1); + } + ); + + test.each( + (["skill", "prompt", "trigger"] as const).flatMap((failure) => + (["result", "rejection"] as const).map((outcome) => ({ failure, outcome })) + ) + )( + "a failed $failure append ($outcome) rolls back only this attempt's previously published rows", + async ({ failure, outcome }) => { + const h = await fixture(); + const foreign = createMuxMessage("foreign", "assistant", "concurrent input"); + const earlier = ["file", "skill", "prompt"].slice( + 0, + ["skill", "prompt", "trigger"].indexOf(failure) + 1 + ); + const append = h.historyService.appendToHistory.bind(h.historyService); + const appends = spyOn(h.historyService, "appendToHistory").mockImplementationOnce( + async (...args) => { + const result = await append(...args); + expect(result).toEqual(Ok(undefined)); + // The real prefix released its lock; a foreign writer now lands before the failure. + expect(await append(workspaceId, foreign)).toEqual(Ok(undefined)); + return result; + } + ); + for (const _row of earlier.slice(1)) appends.mockImplementationOnce(append); + // Disk failures use Result Err; an unexpected service rejection must also retire + // already-published prefixes without deleting a concurrent writer's row. + if (outcome === "result") appends.mockResolvedValueOnce(Err("injected write failure")); + else appends.mockRejectedValueOnce(new Error("injected write failure")); + const start = spyOn(h.aiService, "streamMessage"); + const result = await h.session + .sendMessage("inspect input", options) + .catch((error: unknown) => error); + expect(appends.mock.calls.slice(0, -1).map(([, row]) => row.id)).toEqual(earlier); + expect(foreign.metadata?.historySequence).toBe(1); + expect(appends).toHaveBeenCalledTimes(earlier.length + 1); + expect((await h.rows()).map((row) => row.id)).toEqual([foreign.id]); + expect(result).toMatchObject({ success: false, error: { raw: "injected write failure" } }); + expect(start).not.toHaveBeenCalled(); + } + ); + + test.each(["file", "skill", "prompt", "trigger"])( + "cancellation after %s publication still runs the existing rollback checkpoint", + async (after) => { + const h = await fixture(); + const controller = new AbortController(); + const canceled = mock(() => undefined); + const accepted = mock(() => undefined); + const append = h.historyService.appendToHistory.bind(h.historyService); + spyOn(h.historyService, "appendToHistory").mockImplementation(async (id, row) => { + const result = await append(id, row); + if (row.id === after || (after === "trigger" && row.metadata?.synthetic !== true)) + controller.abort(); + return result; + }); + const start = spyOn(h.aiService, "streamMessage"); + expect( + await h.session.sendMessage("inspect input", options, { + cancelSignal: controller.signal, + onCanceled: canceled, + onAccepted: accepted, + }) + ).toEqual(Ok(undefined)); + expect(await h.rows()).toEqual([]); + expect(canceled).toHaveBeenCalledTimes(1); + expect(accepted).not.toHaveBeenCalled(); + expect(start).not.toHaveBeenCalled(); + } + ); + + test("on-send compaction publishes only the request carrying the deferred user input", async () => { + const h = await fixture(); + spyOn(h.inputs.compactionMonitor, "checkBeforeSend").mockReturnValue({ + shouldShowWarning: true, + shouldForceCompact: true, + usagePercentage: 99, + thresholdPercentage: 85, + contextTokens: 99_000, + maxTokens: 100_000, + }); + spyOn(h.inputs.compactionMonitor, "getThreshold").mockReturnValue(0.85); + expect(await h.session.sendMessage("inspect input", options)).toEqual(Ok(undefined)); + const rows = await h.rows(); + expect(rows).toHaveLength(1); + const request = rows[0].metadata?.muxMetadata; + assert(request?.type === "compaction-request"); + expect(request.parsed.followUpContent?.text).toBe("inspect input"); + expect(h.skills).not.toHaveBeenCalled(); + expect(h.prompts).not.toHaveBeenCalled(); + }); + + test("a manual add during queue admission updates the dispatched origin without splitting the entry", async () => { + const h = await fixture(); + const dispatched = Promise.withResolvers(); + // Observe the public dispatch argument while the real session/history implementation runs. + const send = h.session.sendMessage.bind(h.session); + spyOn(h.session, "sendMessage").mockImplementation((message, options, internal) => { + dispatched.resolve(internal?.acceptanceOrigin); + return send(message, options, internal); + }); + let added = false; + h.session.onChatEvent(({ message }) => { + if (added || message.type !== "stream-lifecycle" || message.phase !== "preparing") return; + added = true; + h.session.queueMessage("manual", options); + }); + h.session.queueMessage("automatic", options, { acceptanceOrigin: "automatic" }); + h.session.sendQueuedMessages(); + expect(await dispatched.promise).toBe("manual"); + expect(h.session.queuedMessageEntryCount()).toBe(0); + }); +}); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 5716440ccc5..4a05e0782d8 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -71,6 +71,7 @@ import { export type { StreamErrorRecoveryOutcome } from "./turnCoordinator"; import type { StreamMessageOptions } from "@/node/services/turnRequestBuilder"; import type { HistoryService } from "@/node/services/historyService"; +import type { TurnAcceptanceOrigin } from "./taskWorkspaceSeam"; import type { SessionUsageService } from "@/node/services/sessionUsageService"; import type { InitStateManager } from "@/node/services/initStateManager"; import type { MCPServerManager } from "@/node/services/mcpServerManager"; @@ -774,6 +775,7 @@ interface CachedMemoryContext { } interface SendMessageInternalOptions { + acceptanceOrigin?: TurnAcceptanceOrigin; preparation?: PreparationAttempt; /** A dequeued send keeps its admission owner through acceptance and startup failure. */ turnReservation?: TurnId; @@ -860,6 +862,7 @@ interface SendMessageInternalOptions { // Enqueueing creates no preparation attempt. Once dispatched, Promise success alone cannot // distinguish cancellation, a background transfer, and delivery to terminal policy. interface PreparationAttempt { + acceptanceOrigin: TurnAcceptanceOrigin; preparedRequest?: PreparedStreamMessage; owner?: TurnId; expectedTurn: TurnId; @@ -1684,6 +1687,7 @@ export class AgentSession { } const result = await this.resumeStream(request.options, { + acceptanceOrigin: "automatic", agentInitiated: request.agentInitiated === true ? true : undefined, goalKind: request.goalKind, goalId: request.goalId, @@ -3306,6 +3310,7 @@ export class AgentSession { if (internal?.preparation) return this.prepareMessage(message, options, internal, internal.preparation); const attempt: PreparationAttempt = { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", owner: internal?.turnReservation, expectedTurn: this.coordinator.turnId, outcome: "preparing", @@ -3417,6 +3422,24 @@ export class AgentSession { const cancelSignal = internal?.cancelSignal; const persistedCancelableMessageIds: string[] = []; + // All prepared rows share publication bookkeeping while retaining their current write + // order and rollback checkpoints. Prefixes alone never establish trigger acceptance. + const publishPreparedHistory = async ( + publication: + | { kind: "prefix"; message: MuxMessage } + | { kind: "trigger"; messages: MuxMessage[] } + ): Promise> => { + const messages = publication.kind === "prefix" ? [publication.message] : publication.messages; + // Preserve the same rollback checkpoints for unexpected service rejection as for + // ordinary Result failures; earlier prefixes may already be durable. + const result = await ( + messages.length === 1 + ? this.historyService.appendToHistory(this.workspaceId, messages[0]) + : this.historyService.appendManyToHistory(this.workspaceId, messages) + ).catch((error: unknown) => Err(getErrorMessage(error))); + if (result.success) persistedCancelableMessageIds.push(...messages.map((row) => row.id)); + return result; + }; // Roll back synthetic snapshots if the invoking user row fails to persist, or // later provider requests could consume orphaned context. /** @@ -4110,14 +4133,13 @@ export class AgentSession { ); // Persist compaction request (NOT the user message — it's the follow-up) - const appendCompactionResult = await this.historyService.appendToHistory( - this.workspaceId, - autoCompactionMessage - ); + const appendCompactionResult = await publishPreparedHistory({ + kind: "trigger", + messages: [autoCompactionMessage], + }); if (!appendCompactionResult.success) { return Err(createUnknownSendMessageError(appendCompactionResult.error)); } - persistedCancelableMessageIds.push(autoCompactionMessage.id); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4186,14 +4208,13 @@ export class AgentSession { } if (shouldPersistTurnSnapshots && !tokenBudgetActive && snapshotResult?.snapshotMessage) { - const snapshotAppendResult = await this.historyService.appendToHistory( - this.workspaceId, - snapshotResult.snapshotMessage - ); + const snapshotAppendResult = await publishPreparedHistory({ + kind: "prefix", + message: snapshotResult.snapshotMessage, + }); if (!snapshotAppendResult.success) { return Err(createUnknownSendMessageError(snapshotAppendResult.error)); } - persistedCancelableMessageIds.push(snapshotResult.snapshotMessage.id); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4201,15 +4222,14 @@ export class AgentSession { if (shouldPersistTurnSnapshots && !tokenBudgetActive && skillSnapshotMessages.length > 0) { for (const snapshotMessage of skillSnapshotMessages) { - const skillSnapshotAppendResult = await this.historyService.appendToHistory( - this.workspaceId, - snapshotMessage - ); + const skillSnapshotAppendResult = await publishPreparedHistory({ + kind: "prefix", + message: snapshotMessage, + }); if (!skillSnapshotAppendResult.success) { await rollbackPersistedTurnRows(); return Err(createUnknownSendMessageError(skillSnapshotAppendResult.error)); } - persistedCancelableMessageIds.push(snapshotMessage.id); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4218,15 +4238,14 @@ export class AgentSession { if (shouldPersistTurnSnapshots && !tokenBudgetActive && mcpPromptSnapshotMessages.length > 0) { for (const snapshotMessage of mcpPromptSnapshotMessages) { - const appendResult = await this.historyService.appendToHistory( - this.workspaceId, - snapshotMessage - ); + const appendResult = await publishPreparedHistory({ + kind: "prefix", + message: snapshotMessage, + }); if (!appendResult.success) { await rollbackPersistedTurnRows(); return Err(createUnknownSendMessageError(appendResult.error)); } - persistedCancelableMessageIds.push(snapshotMessage.id); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4342,16 +4361,14 @@ export class AgentSession { return Err(createUnknownSendMessageError(CONTEXT_MUTATION_SEND_BLOCKED_MESSAGE)); } // Ordinary sends stay append-only; only coupled snapshots/boundaries need an atomic batch. + const publish = () => publishPreparedHistory({ kind: "trigger", messages: batch }); const appended = contextRollover - ? await this.appendContextRolloverRows(batch) - : batch.length === 1 - ? await this.historyService.appendToHistory(this.workspaceId, userMessage) - : await this.historyService.appendManyToHistory(this.workspaceId, batch); + ? await this.appendContextRolloverRows(batch, publish) + : await publish(); if (!appended.success) return Err(createUnknownSendMessageError(appended.error)); } catch (error) { return Err(createUnknownSendMessageError(getErrorMessage(error))); } - persistedCancelableMessageIds.push(...batch.map((row) => row.id)); if (contextRollover) { const sequences = [batch[0], batch[1], userMessage].map( (row) => row.metadata?.historySequence @@ -4367,18 +4384,14 @@ export class AgentSession { } if (await cancelBeforeAcceptance()) return Ok(undefined); } else if (internal?.preTurnMessages != null && internal.preTurnMessages.length > 0) { - const batchAppendResult = await this.historyService.appendManyToHistory(this.workspaceId, [ - ...internal.preTurnMessages, - userMessage, - ]); + const batchAppendResult = await publishPreparedHistory({ + kind: "trigger", + messages: [...internal.preTurnMessages, userMessage], + }); if (!batchAppendResult.success) { await rollbackPersistedTurnRows(); return Err(createUnknownSendMessageError(batchAppendResult.error)); } - persistedCancelableMessageIds.push( - ...internal.preTurnMessages.map((message) => message.id), - userMessage.id - ); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4386,12 +4399,14 @@ export class AgentSession { // When on-send compaction triggers, the user message is NOT persisted to // history (it's sent as follow-up after compaction). Otherwise, persist // normally. - const appendResult = await this.historyService.appendToHistory(this.workspaceId, userMessage); + const appendResult = await publishPreparedHistory({ + kind: "trigger", + messages: [userMessage], + }); if (!appendResult.success) { await rollbackPersistedTurnRows(); return Err(createUnknownSendMessageError(appendResult.error)); } - persistedCancelableMessageIds.push(userMessage.id); if (await cancelBeforeAcceptance()) { return Ok(undefined); } @@ -4768,6 +4783,7 @@ export class AgentSession { async resumeStream( options: SendMessageOptions, internal?: { + acceptanceOrigin?: TurnAcceptanceOrigin; agentInitiated?: boolean; goalKind?: GoalSyntheticMessageKind; goalId?: string; @@ -4829,6 +4845,7 @@ export class AgentSession { } const attempt: PreparationAttempt = { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", expectedTurn: expectedTurnId, outcome: "preparing", durability: "accepted", @@ -5104,10 +5121,14 @@ export class AgentSession { } } - private async appendContextRolloverRows(rows: MuxMessage[]): Promise> { + private async appendContextRolloverRows( + rows: MuxMessage[], + publish: () => Promise> = () => + this.historyService.appendManyToHistory(this.workspaceId, rows) + ): Promise> { let appended: Result; try { - appended = await this.historyService.appendManyToHistory(this.workspaceId, rows); + appended = await publish(); } catch (error) { appended = Err(getErrorMessage(error)); } @@ -5963,6 +5984,7 @@ export class AgentSession { }, args.dedupeKey, { + acceptanceOrigin: "automatic", synthetic: true, agentInitiated: true, sealed: true, @@ -6806,6 +6828,7 @@ export class AgentSession { fallback?.messageText ?? followUp.text, fallback ? { ...fallback.sendOptions, muxMetadata: fallback.metadata } : context.options, { + acceptanceOrigin: "automatic", synthetic: true, agentInitiated: fallback?.agentInitiated ?? context.agentInitiated, goalKind: fallback ? undefined : context.goalKind, @@ -6891,7 +6914,11 @@ export class AgentSession { ...autoCompactionRequest.sendOptions, muxMetadata: autoCompactionRequest.metadata, }, - { synthetic: true, agentInitiated: autoCompactionRequest.agentInitiated } + { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: autoCompactionRequest.agentInitiated, + } ); if (!sendResult.success) { log.warn("Failed to dispatch mid-stream compaction request", { @@ -9200,6 +9227,7 @@ export class AgentSession { message: string, options?: SendMessageOptions & { fileParts?: FilePart[] }, internal?: { + acceptanceOrigin?: TurnAcceptanceOrigin; synthetic?: boolean; agentInitiated?: boolean; /** Request-entry authoring time captured before send preflight awaits (see MessageQueue). */ @@ -9710,6 +9738,7 @@ export class AgentSession { } const expectedTurnId = this.coordinator.turnId; const attempt: PreparationAttempt = { + acceptanceOrigin: candidate.acceptanceOrigin, expectedTurn: expectedTurnId, outcome: "preparing", durability: "rollback-eligible", @@ -9734,6 +9763,7 @@ export class AgentSession { if (this.messageQueue.peekNext()?.identity !== candidate.identity) return Ok(undefined); attempt.queued = true; const { message, options, internal, enqueuedAtMs } = this.messageQueue.dequeueNext(); + attempt.acceptanceOrigin = internal?.acceptanceOrigin ?? "manual"; attempt.onFailure = internal?.onAcceptedPreStreamFailure; this.dispatchingQueuedEntry = true; this.dispatchingQueuedEntryMuxMetadata = options?.muxMetadata; @@ -9750,6 +9780,7 @@ export class AgentSession { this.messageQueue.getNextDispatchableMode() === "tool-end" ); return this.sendMessage(message, options, { + acceptanceOrigin: attempt.acceptanceOrigin, ...internal, enqueuedAtMs, turnReservation: preparedTurn, @@ -10156,6 +10187,7 @@ export class AgentSession { // re-enable auto-retry after a user explicitly opted out. const sendResult = await this.sendMessage(finalText, options, { startStreamInBackground, + acceptanceOrigin: "automatic", synthetic: true, agentInitiated: followUp.agentInitiated, goalKind: persistedGoalKind, diff --git a/src/node/services/messageQueue.test.ts b/src/node/services/messageQueue.test.ts index d2c9de4613f..3e1474ce55a 100644 --- a/src/node/services/messageQueue.test.ts +++ b/src/node/services/messageQueue.test.ts @@ -10,6 +10,41 @@ describe("MessageQueue", () => { queue = new MessageQueue(); }); + describe("acceptance origin", () => { + it("preserves automatic origin across batching without changing visibility or billing", () => { + const internal = { acceptanceOrigin: "automatic" as const }; + queue.add("first", undefined, internal); + queue.add("second", undefined, internal); + expect(queue.getMessages()).toEqual(["first", "second"]); + const dispatched = queue.dequeueNext(); + expect(dispatched.message).toBe("first\nsecond"); + expect(dispatched.internal).toEqual(internal); + expect(queue.isEmpty()).toBe(true); + }); + + it("removing a keyed manual add restores the remaining automatic origin", () => { + const automatic = { acceptanceOrigin: "automatic" as const }; + queue.addOnce("automatic", undefined, "auto:1", automatic); + queue.addOnce("manual", undefined, "manual:1"); + expect(queue.peekNext()?.acceptanceOrigin).toBe("manual"); + expect(queue.removeByDedupeKeyPrefix("manual:").removedCount).toBe(1); + expect(queue.dequeueNext()).toMatchObject({ message: "automatic", internal: automatic }); + }); + + it("retains file-only origin and ignores duplicate adds that were never queued", () => { + const file = { type: "file" as const, url: "file:///input.txt", mediaType: "text/plain" }; + const automatic = { acceptanceOrigin: "automatic" as const }; + queue.addOnce("", { model: "test", agentId: "exec", fileParts: [file] }, "files", automatic); + expect(queue.addOnce("duplicate manual", undefined, "files")).toBe(false); + queue.add("automatic text", undefined, automatic); + expect(queue.dequeueNext()).toMatchObject({ + message: "automatic text", + options: { fileParts: [file] }, + internal: automatic, + }); + }); + }); + describe("authoredAtMs", () => { it("returns the request-entry authoring time from dequeueNext when provided", () => { // Codex P2 (PRRT_kwDOPxxmWM6b-orA): the sender captures authoring time diff --git a/src/node/services/messageQueue.ts b/src/node/services/messageQueue.ts index 53f368a6ed9..1ef64a2e194 100644 --- a/src/node/services/messageQueue.ts +++ b/src/node/services/messageQueue.ts @@ -7,6 +7,7 @@ import { getValidAgentPeerTriggerMeta } from "@/common/utils/agentMessageEnvelop import type { SendMessageError } from "@/common/types/errors"; import type { MuxMessage } from "@/common/types/message"; import type { ReviewNoteData } from "@/common/types/review"; +import type { TurnAcceptanceOrigin } from "./taskWorkspaceSeam"; // Type guard for compaction request metadata (for display text) interface CompactionMetadata { @@ -116,6 +117,7 @@ export type QueueCutCutter = | { stage: "queued"; muxMetadata: unknown; dispatchMode: QueueDispatchMode }; interface QueuedMessageInternalOptions { + acceptanceOrigin?: TurnAcceptanceOrigin; goalKind?: GoalSyntheticMessageKind; goalId?: string; synthetic?: boolean; @@ -204,6 +206,9 @@ interface QueueEntry { addCount: number; syntheticCount: number; agentInitiatedCount: number; + // Keep per-add origins so removing a keyed manual add cannot promote remaining + // automatic work into replacement authority. This does not change queue grouping. + acceptanceOrigins: Array<{ origin: TurnAcceptanceOrigin; dedupeKey?: string }>; /** * Timestamp of the latest add batched into this entry. Dispatch exposes it so * goal safety can tell messages typed before a goal existed (queued while the @@ -557,6 +562,7 @@ export class MessageQueue { const entry = this.addInternal(message, options, internal); if (entry != null && dedupeKey !== undefined) { entry.dedupeKeys.add(dedupeKey); + entry.acceptanceOrigins.at(-1)!.dedupeKey = dedupeKey; } return entry != null; } @@ -635,6 +641,7 @@ export class MessageQueue { addCount: 0, syntheticCount: 0, agentInitiatedCount: 0, + acceptanceOrigins: [], // 0, not Date.now(): every add (including the entry-creating one) // folds its authoring time in below via max(); seeding with the // creation wall clock would swallow an authoredAtMs captured before @@ -708,6 +715,7 @@ export class MessageQueue { } entry.addCount += 1; + entry.acceptanceOrigins.push({ origin: internal?.acceptanceOrigin ?? "manual" }); // Codex security P2 (PRRT_kwDOPxxmWM6b_OS9): batched sends can finish // preflight out of authoring order. Keep the NEWEST authoring time for // the entry — a plain overwrite would let an older pre-goal message mask @@ -915,6 +923,9 @@ export class MessageQueue { entry.addCount -= matchingKeys.length; entry.syntheticCount = Math.min(entry.syntheticCount, entry.addCount); entry.agentInitiatedCount = Math.min(entry.agentInitiatedCount, entry.addCount); + entry.acceptanceOrigins = entry.acceptanceOrigins.filter( + (add) => add.dedupeKey == null || !matchingKeySet.has(add.dedupeKey) + ); return [entry]; } if (entry.onCanceled != null || entry.onAcceptedPreStreamFailure != null) { @@ -948,10 +959,24 @@ export class MessageQueue { return true; } + private getAcceptanceOrigin(entry: QueueEntry): TurnAcceptanceOrigin { + return entry.acceptanceOrigins.every((add) => add.origin === "automatic") + ? "automatic" + : "manual"; + } + /** Capture before admission publication; observers may remove or reorder the head. */ - peekNext(): { identity: object; muxMetadata: unknown } | undefined { + peekNext(): + | { identity: object; muxMetadata: unknown; acceptanceOrigin: TurnAcceptanceOrigin } + | undefined { const entry = this.entries[0]; - return entry ? { identity: entry, muxMetadata: entry.muxMetadata } : undefined; + return entry + ? { + identity: entry, + muxMetadata: entry.muxMetadata, + acceptanceOrigin: this.getAcceptanceOrigin(entry), + } + : undefined; } /** @@ -991,7 +1016,9 @@ export class MessageQueue { const allAddsAreSynthetic = entry.addCount > 0 && entry.syntheticCount === entry.addCount; const allAddsAreAgentInitiated = entry.addCount > 0 && entry.agentInitiatedCount === entry.addCount; + const automaticAcceptance = this.getAcceptanceOrigin(entry) === "automatic"; const hasInternalOptions = + automaticAcceptance || allAddsAreSynthetic || allAddsAreAgentInitiated || entry.onAccepted != null || @@ -1002,6 +1029,7 @@ export class MessageQueue { (entry.preTurnMessages?.length ?? 0) > 0; const internal = hasInternalOptions ? { + ...(automaticAcceptance ? { acceptanceOrigin: "automatic" as const } : {}), ...(allAddsAreSynthetic ? { synthetic: true } : {}), ...(allAddsAreAgentInitiated ? { agentInitiated: true } : {}), ...(entry.goalKind != null ? { goalKind: entry.goalKind, goalId: entry.goalId } : {}), diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 4044f662853..ec5de7d6513 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -501,7 +501,7 @@ describe("TaskService", () => { childId, "Inspect the scratch files", expect.any(Object), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }); @@ -3499,6 +3499,7 @@ describe("TaskService", () => { expect(sends[index]?.[2].queueDispatchMode).toBe(entry.queueDispatchMode); expect(sends[index]?.[3]).toMatchObject({ restoreQueued: true, + acceptanceOrigin: "automatic", queueDedupeKey: entry.id, }); } @@ -3520,6 +3521,7 @@ describe("TaskService", () => { const later = findWorkspaceInConfig(config, childId)?.taskPendingGuidance?.[2]; assert(later != null); expect(sends[2]?.[3]?.queueDedupeKey).toBe(later.id); + expect(sends[2]?.[3]?.acceptanceOrigin).toBe("automatic"); await sends[0]?.[3]?.onAccepted?.(); expect(findWorkspaceInConfig(config, childId)?.taskPendingGuidance).toEqual([ guidance[1], @@ -6897,7 +6899,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const postCfg = config.loadConfigOrDefault(); @@ -6956,7 +6958,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const postCfg = config.loadConfigOrDefault(); @@ -7010,7 +7012,7 @@ describe("TaskService", () => { reasoningMode: "pro", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); // Persisted child settings carry it too, so queued/restart resumes @@ -7065,7 +7067,7 @@ describe("TaskService", () => { created.data.taskId, "run explore with base pro", expect.objectContaining({ agentId: "explore", reasoningMode: "pro" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7113,7 +7115,7 @@ describe("TaskService", () => { created.data.taskId, "run explore with parent pro mode", expect.objectContaining({ agentId: "explore", reasoningMode: "pro" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7168,7 +7170,7 @@ describe("TaskService", () => { created.data.taskId, "run with mapped alias max", expect.objectContaining({ model: "openai:team-sol", thinkingLevel: "max" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7214,7 +7216,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const postCfg = config.loadConfigOrDefault(); @@ -7276,7 +7278,7 @@ describe("TaskService", () => { thinkingLevel: "off", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const postCfg = config.loadConfigOrDefault(); @@ -7344,7 +7346,7 @@ describe("TaskService", () => { created.data.taskId, "run researcher with plan pro", expect.objectContaining({ agentId: "researcher", reasoningMode: "pro" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7405,7 +7407,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const postCfg = config.loadConfigOrDefault(); @@ -7478,7 +7480,7 @@ describe("TaskService", () => { thinkingLevel: "off", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7511,7 +7513,7 @@ describe("TaskService", () => { thinkingLevel: "medium", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.aiSettings).toEqual({ model: "openai:gpt-5.2", thinkingLevel: "medium" }); @@ -7549,7 +7551,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7585,7 +7587,7 @@ describe("TaskService", () => { thinkingLevel: "off", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7615,7 +7617,7 @@ describe("TaskService", () => { thinkingLevel: "high", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -7688,7 +7690,7 @@ describe("TaskService", () => { thinkingLevel: "medium", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe("openai:gpt-5.3-codex"); @@ -7728,7 +7730,7 @@ describe("TaskService", () => { thinkingLevel: "off", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe("anthropic:claude-haiku-4-5"); @@ -7770,7 +7772,7 @@ describe("TaskService", () => { thinkingLevel: expectedThinkingLevel, experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe(resolvedModel); @@ -7815,7 +7817,7 @@ describe("TaskService", () => { agentId: "exec", thinkingLevel: "high", }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const child = findWorkspaceInConfig(config, created.data.taskId); expect(child?.taskModelString).toBe("openai:gpt-6-astra"); @@ -7882,7 +7884,7 @@ describe("TaskService", () => { created.data.taskId, "check provenance", expect.objectContaining({ model: expected }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }); @@ -7998,7 +8000,7 @@ describe("TaskService", () => { grandchild.data.taskId, "grandchild", expect.objectContaining({ model: "openai:gpt-5.3-codex" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); expect(findWorkspaceInConfig(config, grandchild.data.taskId)?.taskModelString).toBe( "openai:gpt-5.3-codex" @@ -8038,7 +8040,7 @@ describe("TaskService", () => { created.data.taskId, "inherit Standard", expect.objectContaining({ reasoningMode: "standard" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); } ); @@ -8079,7 +8081,7 @@ describe("TaskService", () => { created.data.taskId, "keep explicit overrides", expect.objectContaining(expected), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }); @@ -8118,7 +8120,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe("openai:gpt-5.3-codex"); @@ -8159,7 +8161,7 @@ describe("TaskService", () => { thinkingLevel: "medium", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe("openai:gpt-5.2"); @@ -8198,7 +8200,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -8237,7 +8239,7 @@ describe("TaskService", () => { thinkingLevel: "xhigh", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -8279,7 +8281,7 @@ describe("TaskService", () => { thinkingLevel: expectedThinkingLevel, experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const childEntry = findWorkspaceInConfig(config, created.data.taskId); expect(childEntry?.taskModelString).toBe(resolvedModel); @@ -8322,7 +8324,7 @@ describe("TaskService", () => { thinkingLevel: "high", experiments: undefined, }, - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }, 20_000); @@ -10177,7 +10179,7 @@ describe("TaskService", () => { expect(resumeStream).toHaveBeenCalledWith( parentWorkspaceId, expect.objectContaining({ agentId: "plan" }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); const parentHistory = await collectFullHistory(historyService, parentWorkspaceId); @@ -17380,6 +17382,7 @@ describe("TaskService", () => { expect(remove).not.toHaveBeenCalled(); expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); expect(emit).toHaveBeenCalled(); @@ -18994,6 +18997,7 @@ describe("TaskService", () => { expect(remove).not.toHaveBeenCalled(); expect(sendMessageMock).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); }); @@ -19110,6 +19114,7 @@ describe("TaskService", () => { expect(remove).not.toHaveBeenCalled(); expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); }); @@ -23307,6 +23312,7 @@ describe("TaskService", () => { expect.objectContaining({ synthetic: true, agentInitiated: true }) ); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); // The failure details travel via the durable synthetic history message, @@ -23443,6 +23449,7 @@ describe("TaskService", () => { await flushTerminalAttentionDrains(taskService); expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); @@ -23580,6 +23587,7 @@ describe("TaskService", () => { await flushTerminalAttentionDrains(taskService); expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); @@ -23751,6 +23759,7 @@ describe("TaskService", () => { expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledTimes(1); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); }); @@ -23856,6 +23865,7 @@ describe("TaskService", () => { await flushTerminalAttentionDrains(taskService); expect(sendMessage).not.toHaveBeenCalled(); expect(resumeStream).toHaveBeenCalledWith(parentId, expect.any(Object), { + acceptanceOrigin: "automatic", agentInitiated: true, }); @@ -30652,7 +30662,7 @@ describe("TaskService", () => { expect.objectContaining({ muxMetadata: workspaceTurnMuxMetadata(parentId), }), - { agentInitiated: true } + { acceptanceOrigin: "automatic", agentInitiated: true } ); }); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 943bee0dc04..bd0b688ac5f 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -2559,7 +2559,7 @@ export class TaskService implements AgentTaskIntegration { "Xum restarted while this task was running. Continue where you left off. " + restartCompletionInstruction, sendOptions, - { synthetic: true, agentInitiated: true } + { acceptanceOrigin: "automatic", synthetic: true, agentInitiated: true } ); const durationMs = Date.now() - resumeStartedAt; if (!sendResult.success) { @@ -3650,10 +3650,12 @@ export class TaskService implements AgentTaskIntegration { const sendResult = plan.start.kind === "sendMessage" ? await this.workspaceService.sendMessage(plan.taskId, plan.start.prompt, startOptions, { + acceptanceOrigin: "automatic", allowQueuedAgentTask: true, agentInitiated: true, }) : await this.workspaceService.resumeStream(plan.taskId, startOptions, { + acceptanceOrigin: "automatic", allowQueuedAgentTask: true, agentInitiated: true, }); @@ -4355,7 +4357,10 @@ export class TaskService implements AgentTaskIntegration { reasoningMode: effectiveReasoningMode, experiments: args.experiments, }, - { agentInitiated: true } + { + acceptanceOrigin: "automatic", + agentInitiated: true, + } ) .catch((error: unknown) => Err(getErrorMessage(error))); if (!sendResult.success) { @@ -4756,6 +4761,7 @@ export class TaskService implements AgentTaskIntegration { const onCanceled = () => clear(true); // Live and restored guidance must share the same handoff and settlement lifecycle. return { + acceptanceOrigin: "automatic" as const, synthetic: true, agentInitiated: true, startStreamInBackground: true, @@ -5480,6 +5486,7 @@ export class TaskService implements AgentTaskIntegration { let accepted = false; const sendResult = await this.workspaceService.sendMessage(targetId, trigger, sendOptions, { + acceptanceOrigin: "automatic", admissionStale, synthetic: true, agentInitiated: true, @@ -7893,6 +7900,7 @@ export class TaskService implements AgentTaskIntegration { // suppressed ones first cannot go stale against a delivery. await markSuppressedSuperseded(); const resumeResult = await this.workspaceService.resumeStream(ownerWorkspaceId, sendOptions, { + acceptanceOrigin: "automatic", agentInitiated: true, }); if (!resumeResult.success) { @@ -7922,7 +7930,13 @@ export class TaskService implements AgentTaskIntegration { prompt, sendOptions, // Synthetic, idle-only auto-resume — same flags as the active-work auto-resume path. - { skipAutoResumeReset: true, synthetic: true, agentInitiated: true, requireIdle: true } + { + acceptanceOrigin: "automatic", + skipAutoResumeReset: true, + synthetic: true, + agentInitiated: true, + requireIdle: true, + } ); // Deferred until after the delivery attempt so no await separates the // batch revalidation above from sendMessage. Best-effort: an early return @@ -7970,6 +7984,7 @@ export class TaskService implements AgentTaskIntegration { prompt, sendOptions, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, @@ -8273,6 +8288,7 @@ export class TaskService implements AgentTaskIntegration { ...(workspaceTurnMuxMetadata != null ? { muxMetadata: workspaceTurnMuxMetadata } : {}), }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, @@ -8487,6 +8503,7 @@ export class TaskService implements AgentTaskIntegration { : {}), }, { + acceptanceOrigin: "automatic", synthetic: true, agentInitiated: true, startStreamInBackground: true, @@ -10872,7 +10889,11 @@ export class TaskService implements AgentTaskIntegration { ? { toolPolicy: [{ regex_match: "^propose_plan$", action: "require" as const }] } : {}), }, - { synthetic: true, agentInitiated: true } + { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: true, + } ); const durationMs = Date.now() - startedAt; if (!sendResult.success) { @@ -10933,7 +10954,11 @@ export class TaskService implements AgentTaskIntegration { reasoningMode: coerceOpenAIReasoningMode(entry.workspace.aiSettings?.reasoningMode), experiments: entry.workspace.taskExperiments, }, - { synthetic: true, agentInitiated: true } + { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: true, + } ); if (!sendResult.success) { log.error("Failed to prompt task for active background awaitables", { @@ -11189,7 +11214,13 @@ export class TaskService implements AgentTaskIntegration { prompt, sendOptions, // Skip auto-resume counter reset — this IS an auto-resume, not a user message. - { skipAutoResumeReset: true, synthetic: true, agentInitiated: true, requireIdle: true } + { + acceptanceOrigin: "automatic", + skipAutoResumeReset: true, + synthetic: true, + agentInitiated: true, + requireIdle: true, + } ); if (!sendResult.success && isWorkspaceBusyIdleOnlySend(sendResult.error)) { activeWorkspaceTurnIds = @@ -11239,6 +11270,7 @@ export class TaskService implements AgentTaskIntegration { }), sendOptions, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, @@ -12061,7 +12093,11 @@ export class TaskService implements AgentTaskIntegration { ...(effectiveReasoningMode != null ? { reasoningMode: effectiveReasoningMode } : {}), experiments: args.entry.workspace.taskExperiments, }, - { synthetic: true, agentInitiated: true } + { + acceptanceOrigin: "automatic", + synthetic: true, + agentInitiated: true, + } ); if (!sendKickoffResult.success) { // Keep status as "running" so the restart handler in initialize() can diff --git a/src/node/services/taskWorkspaceSeam.ts b/src/node/services/taskWorkspaceSeam.ts index 9b23d610c14..114a9358cc4 100644 --- a/src/node/services/taskWorkspaceSeam.ts +++ b/src/node/services/taskWorkspaceSeam.ts @@ -322,7 +322,11 @@ export interface WorkspaceLiveActivity { desktopViewers: boolean; } +/** Who requested acceptance, independent of transcript visibility and provider billing. */ +export type TurnAcceptanceOrigin = "manual" | "automatic"; + export interface SendMessageInternalOptions { + acceptanceOrigin?: TurnAcceptanceOrigin; allowQueuedAgentTask?: boolean; skipAutoResumeReset?: boolean; synthetic?: boolean; @@ -397,7 +401,11 @@ export interface WorkspaceTurnHost { resumeStream( workspaceId: string, options: SendMessageOptions, - internal?: { allowQueuedAgentTask?: boolean; agentInitiated?: boolean } + internal?: { + acceptanceOrigin?: TurnAcceptanceOrigin; + allowQueuedAgentTask?: boolean; + agentInitiated?: boolean; + } ): Promise>; clearQueue(workspaceId: string, options?: { cancelReason?: string }): Result; replaceHistory( diff --git a/src/node/services/turnRequestBuilder.ts b/src/node/services/turnRequestBuilder.ts index be3ff56dcbc..587039013e3 100644 --- a/src/node/services/turnRequestBuilder.ts +++ b/src/node/services/turnRequestBuilder.ts @@ -35,6 +35,7 @@ import { import type { DebugLlmRequestSnapshot } from "@/common/types/debugLlmRequest"; import type { SendMessageError } from "@/common/types/errors"; +import type { TurnAcceptanceOrigin } from "./taskWorkspaceSeam"; import type { GoalRecordV1 } from "@/common/types/goal"; import type { ModelMessage, MuxMessage, MuxMessageMetadata } from "@/common/types/message"; import { createMuxMessage } from "@/common/types/message"; @@ -510,6 +511,7 @@ interface WorkflowResultContinuationSender { message: string, options: SendMessageOptions, internal?: { + acceptanceOrigin?: TurnAcceptanceOrigin; skipAutoResumeReset?: boolean; synthetic?: boolean; agentInitiated?: boolean; @@ -2133,6 +2135,7 @@ export class TurnRequestBuilder { }, }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, diff --git a/src/node/services/workflowContinuation.ts b/src/node/services/workflowContinuation.ts index 94bbc88ef09..9304eeac08f 100644 --- a/src/node/services/workflowContinuation.ts +++ b/src/node/services/workflowContinuation.ts @@ -67,6 +67,7 @@ export async function sendWorkflowRunTerminalContinuation( }, }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, diff --git a/src/node/services/workspaceGoalService.test.ts b/src/node/services/workspaceGoalService.test.ts index a2ad52e592f..b56759ef801 100644 --- a/src/node/services/workspaceGoalService.test.ts +++ b/src/node/services/workspaceGoalService.test.ts @@ -464,27 +464,26 @@ describe("WorkspaceGoalService", () => { const publicationGate = new Promise((resolve) => { releasePublication = resolve; }); + let markPublicationEntered!: () => void; + const publicationEntered = new Promise((resolve) => { + markPublicationEntered = resolve; + }); const pushSnapshotSpy = spyOn( service as unknown as { pushSnapshot: (workspaceId: string, goal: unknown) => Promise }, "pushSnapshot" ).mockImplementation(async () => { + markPublicationEntered(); await publicationGate; }); + const pausePromise = service.setGoal({ + workspaceId, + status: "paused", + initiator: "user", + }); try { - const pausePromise = service.setGoal({ - workspaceId, - status: "paused", - initiator: "user", - }); - const goalPath = path.join(config.sessionsDir, workspaceId, "goal.json"); - await waitForCondition(async () => { - try { - const raw = JSON.parse(await fs.readFile(goalPath, "utf-8")) as { status?: string }; - return raw.status === "paused"; - } catch { - return false; - } - }); + // Atomic rename makes the file visible before write-file-atomic finishes + // cleanup. Publication entry proves writeGoal completed its generation bump. + await publicationEntered; // The durable pause has committed but publication (and the finalization // hold arming) has not — the captured probe must already be stale. expect(admission.admissionStale()).toBe(true); @@ -493,7 +492,12 @@ describe("WorkspaceGoalService", () => { expect(paused.success).toBe(true); } finally { releasePublication(); - pushSnapshotSpy.mockRestore(); + try { + // Drain the write even if an assertion fails, before fixture cleanup runs. + await pausePromise; + } finally { + pushSnapshotSpy.mockRestore(); + } } }); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 73fedd7d3d0..757ff24b9cc 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -341,6 +341,7 @@ import { type AgentTaskIntegration, type ArchiveWorkspaceOptions, type SendMessageInternalOptions, + type TurnAcceptanceOrigin, type WorkspaceHost, type WorkspaceLiveActivity, } from "@/node/services/taskWorkspaceSeam"; @@ -2695,6 +2696,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { muxMetadata: dispatch.muxMetadata, }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, @@ -11393,6 +11395,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { // ahead of the user's intervention while the fallback persists the // rejected row and applies goal safety. return await session.sendMessage(message, normalizedOptions, { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", synthetic: internal?.synthetic, agentInitiated: internal?.agentInitiated, goalKind: internal?.goalKind, @@ -11573,6 +11576,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { message, continuationSendState.options, { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", synthetic: internal?.synthetic, agentInitiated: internal?.agentInitiated, authoredAtMs, @@ -11690,6 +11694,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { // recovery admit an exec turn ahead of the accepted manual send. Refusal // paths never fire the callback; the scoped disposal releases on return. const result = await session.sendMessage(message, continuationSendState.options, { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", onTurnAdmissionCommitted: () => sessionInvisiblePreflight.release(), onContextWindowRollover: () => { this.advanceContextMutationEpoch(workspaceId); @@ -11810,7 +11815,11 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { async resumeStream( workspaceId: string, options: SendMessageOptions, - internal?: { allowQueuedAgentTask?: boolean; agentInitiated?: boolean } + internal?: { + acceptanceOrigin?: TurnAcceptanceOrigin; + allowQueuedAgentTask?: boolean; + agentInitiated?: boolean; + } ): Promise> { let resumedInterruptedTask = false; let previousTaskStatus: ReturnType; @@ -11965,6 +11974,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { // started (or refused), so no follow-up redispatched from within the // resumed turn itself can observe the reservation and self-veto. const result = await session.resumeStream(normalizedOptions, { + acceptanceOrigin: internal?.acceptanceOrigin ?? "manual", agentInitiated: internal?.agentInitiated, }); sessionInvisiblePreflight.release(); @@ -15421,6 +15431,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { editMessageId: undefined, }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, agentInitiated: true, @@ -15501,6 +15512,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { muxMetadata, }, { + acceptanceOrigin: "automatic", // Idle compaction runs in background; avoid mutating auto-resume counters. skipAutoResumeReset: true, // Backend-initiated maintenance turn: do not treat as explicit user re-engagement. @@ -15811,6 +15823,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { queueDispatchMode: whenBusy, }, { + acceptanceOrigin: "automatic", // Heartbeats run in background; avoid mutating auto-resume counters. skipAutoResumeReset: true, // Backend-initiated maintenance turn: do not treat as explicit user re-engagement. @@ -15849,6 +15862,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { ...(whenBusy !== "skip" ? { queueDispatchMode: whenBusy } : {}), }, { + acceptanceOrigin: "automatic", // Heartbeats run in background; avoid mutating auto-resume counters. skipAutoResumeReset: true, // Backend-initiated maintenance turn: do not treat as explicit user re-engagement. @@ -15896,6 +15910,7 @@ export class WorkspaceService extends EventEmitter implements WorkspaceHost { muxMetadata: compactionMuxMetadata, }, { + acceptanceOrigin: "automatic", skipAutoResumeReset: true, synthetic: true, requireIdle: true, diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts index 926971bf80e..8f5e9ed629d 100644 --- a/src/node/services/workspaceTurnManager.ts +++ b/src/node/services/workspaceTurnManager.ts @@ -1559,6 +1559,7 @@ export class WorkspaceTurnManager { : {}), }, { + acceptanceOrigin: "automatic", startStreamInBackground: true, requireIdle: !queuedForExistingWorkspace, onCanceled: async (reason) => {