From f44738129cb14c02b256ae6a17b2cf7d80a3326f Mon Sep 17 00:00:00 2001 From: Joseph Vidal Date: Tue, 6 Oct 2026 02:18:12 -0400 Subject: [PATCH 1/6] fix(server): cited assistant quotes reach providers expanded again The orchestrator v2 rewrite dropped the citation expansion that ran in ProviderService.sendTurn, so turn starts, steers, and context handoffs sent raw t3-citation:// links to providers. Provider-bound user text now goes through one helper that projects composer context and expands citations, and a message whose quotes expand past the 120,000-character provider limit fails its run with a validation error instead of being sent. Stored messages keep their clickable links. Fixes #16327 --- .../orchestration-v2/ProviderMessageText.ts | 38 ++++++ .../ProviderTurnControlService.test.ts | 114 ++++++++++++++++++ .../ProviderTurnControlService.ts | 8 +- .../ProviderTurnStartService.test.ts | 44 +++++++ .../ProviderTurnStartService.ts | 31 +++-- .../ProviderSwitch.integration.test.ts | 109 +++++++++++++++++ .../provider-core/src/server/handoffBudget.ts | 4 + 7 files changed, 334 insertions(+), 14 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ProviderMessageText.ts diff --git a/apps/server/src/orchestration-v2/ProviderMessageText.ts b/apps/server/src/orchestration-v2/ProviderMessageText.ts new file mode 100644 index 0000000000000..6f78a6ebf2e80 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderMessageText.ts @@ -0,0 +1,38 @@ +import { + type OrchestrationMessageContext, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, +} from "@t3tools/contracts"; +import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations"; +import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +class ProviderInputTooLongError extends Schema.TaggedError()( + "ProviderInputTooLongError", + { inputLength: Schema.Number }, +) { + override get message(): string { + const excess = this.inputLength - PROVIDER_SEND_TURN_MAX_INPUT_CHARS; + return `Message is ${excess.toLocaleString("en-US")} ${excess === 1 ? "character" : "characters"} over the ${PROVIDER_SEND_TURN_MAX_INPUT_CHARS.toLocaleString("en-US")}-character limit once assistant quotes are expanded. Shorten it or quote less.`; + } +} + +/** + * The text a provider receives for a persisted user message: composer context references + * become their envelope and assistant citations carry their saved quotes. The stored message + * keeps its links. A message that fit as links can exceed the provider input limit once its + * quotes expand, so the expanded text is checked again, as the composer does before sending. + */ +export const providerUserMessageText = (message: { + readonly text: string; + readonly context?: OrchestrationMessageContext | undefined; +}): Effect.Effect => { + const projected = projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }); + const text = expandAssistantCitationsForProvider(projected); + return text !== projected && text.length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS + ? Effect.fail(new ProviderInputTooLongError({ inputLength: text.length })) + : Effect.succeed(text); +}; diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 42b3943e71d0c..c3b1085bbf973 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -1,8 +1,12 @@ import { assert, it } from "@effect/vitest"; import { + EnvironmentId, + MessageId, type ModelSelection, NodeId, + type OrchestrationV2ConversationMessage, type OrchestrationV2ProviderThread, + type OrchestrationV2Run, type OrchestrationV2ThreadProjection, ProjectId, ProviderDriverKind, @@ -14,6 +18,10 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import { + expandAssistantCitationsForProvider, + serializeAssistantCitation, +} from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -327,3 +335,109 @@ it.effect( assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session"); }), ); + +it.effect("steers with cited assistant quotes expanded for the provider", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:steer-citation"); + const providerSessionId = ProviderSessionId.make("provider-session:steer-citation"); + const providerTurnId = ProviderTurnId.make("provider-turn:steer-citation"); + const attemptId = RunAttemptId.make("run-attempt:steer-citation"); + const providerThread: OrchestrationV2ProviderThread = { + id: ProviderThreadId.make("provider-thread:steer-citation"), + driver, + providerInstanceId, + providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-thread:steer-citation", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }; + const projection = makeProjection({ now, threadId, providerThread, providerTurnId, attemptId }); + const text = `Revisit ${serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment:steer-citation"), + threadId, + messageId: MessageId.make("message:steer-citation:source"), + text: "Ship it on Friday.", + start: 0, + end: 18, + prefix: "", + suffix: "", + })}`; + const message = { + id: MessageId.make("message:steer-citation"), + text, + attachments: [], + createdBy: "user", + creationSource: "web", + } as unknown as OrchestrationV2ConversationMessage; + const steered = yield* Ref.make>([]); + const runtime: ProviderAdapterV2SessionRuntime = { + instanceId: providerInstanceId, + driver, + providerSessionId, + providerSession: { + id: providerSessionId, + driver, + providerInstanceId, + status: "running", + cwd: "/workspace", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.empty, + ensureThread: () => Effect.die("unused ensureThread"), + resumeThread: () => Effect.die("unused resumeThread"), + startTurn: () => Effect.die("unused startTurn"), + steerTurn: (input) => Ref.update(steered, (texts) => [...texts, input.message.text]), + interruptTurn: () => Effect.die("unused interruptTurn"), + respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"), + readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"), + rollbackThread: () => Effect.die("unused rollbackThread"), + forkThread: () => Effect.die("unused forkThread"), + }; + const layerControl = ProviderTurnControlService.layer.pipe( + Layer.provide( + Layer.merge( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getProviderControlContext: () => + Effect.succeed({ + providerThread, + providerTurn: projection.providerTurns[0], + attempt: projection.attempts[0], + message, + run: { id: RunId.make("run:steer-citation") } as OrchestrationV2Run, + }), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + get: () => Effect.succeedSome(runtime), + }), + ), + ), + ); + + yield* Effect.gen(function* () { + const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2; + yield* control.steer({ + threadId, + providerSessionId, + providerThreadId: providerThread.id, + providerTurnId, + messageId: message.id, + }); + }).pipe(Effect.provide(layerControl)); + + assert.deepEqual(yield* Ref.get(steered), [expandAssistantCitationsForProvider(text)]); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index 4421da12da6ff..89d364f41d780 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -1,4 +1,3 @@ -import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { MessageId, ProviderSessionId, @@ -15,6 +14,7 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as ProjectionStore from "./ProjectionStore.ts"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; const yieldToRuntime = Effect.yieldNow.pipe( @@ -314,6 +314,7 @@ export const layer: Layer.Layer< cause: "The persisted steering message or target run is missing.", }); } + const text = yield* providerUserMessageText(message); yield* loaded.session.value .steerTurn({ threadId: input.threadId, @@ -322,10 +323,7 @@ export const layer: Layer.Layer< providerTurnId: loaded.providerTurn.id, message: { messageId: message.id, - text: projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }), + text, attachments: message.attachments, createdBy: message.createdBy, creationSource: message.creationSource, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 54e55c40a3ccf..338fa53a43026 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -2,6 +2,7 @@ import { expect, it, vi } from "vite-plus/test"; import { it as effectIt } from "@effect/vitest"; import { CheckpointScopeId, + EnvironmentId, MessageId, NodeId, ProviderSessionId, @@ -15,7 +16,12 @@ import { ProjectId, type OrchestrationV2ThreadProjection, OrchestrationV2DomainEvent, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, } from "@t3tools/contracts"; +import { + expandAssistantCitationsForProvider, + serializeAssistantCitation, +} from "@t3tools/shared/assistantCitations"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -855,3 +861,41 @@ for (const previousMessages of [[], ["/compact", " /COMPACT "]]) { }), ); } + +effectIt.effect("fails a run whose expanded assistant quotes exceed the provider input limit", () => + Effect.gen(function* () { + const citation = serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment-citation"), + threadId: ThreadId.make("thread-citation-source"), + messageId: MessageId.make("message-citation-source"), + text: "Quoted claim", + start: 0, + end: 12, + prefix: "", + suffix: "", + }); + // Fits as a link, which the composer allows, but not once the quote expands. + const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; + expect(text.length).toBe(PROVIDER_SEND_TURN_MAX_INPUT_CHARS); + expect(expandAssistantCitationsForProvider(text).length).toBeGreaterThan( + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + ); + const harness = makeLocalCommandHarness({ text }); + + yield* harness.start; + + expect(harness.open).not.toHaveBeenCalled(); + expect(harness.startRootRun).not.toHaveBeenCalled(); + const projection = harness.projection(); + expect(projection.runs.at(-1)?.status).toBe("failed"); + expect(projection.messages.at(-1)?.text).toBe(text); + expect(projection.turnItems).toMatchObject([ + { + type: "error", + title: "Message is too long", + failure: { class: "validation_error" }, + }, + ]); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index a928680b24024..bf7d36e5b9529 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1,5 +1,4 @@ import { modelSelectionsEqual } from "@t3tools/shared/model"; -import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { CommandId, latestProviderTurnForAttempt, @@ -47,6 +46,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import { @@ -375,6 +375,26 @@ export const layer: Layer.Layer< }); }, ); + const providerText = yield* Effect.result(providerUserMessageText(message)); + if (providerText._tag === "Failure") { + yield* settleRunBeforeStart({ + signal: "provider-input-too-long", + status: "failed", + now: yield* DateTime.now, + providerInstanceId: run.providerInstanceId, + itemProviderThreadId: providerThread.id, + item: { + type: "error", + title: "Message is too long", + failure: makeProviderFailure({ + class: "validation_error", + message: providerText.failure.message, + }), + }, + }); + return; + } + const userText = providerText.success; if (message.attachments.length === 0 && message.text.trimStart().startsWith("/")) { const isEmptyCompaction = message.text.trim().toLowerCase() === "/compact" && !projection.hasConversation; @@ -408,10 +428,7 @@ export const layer: Layer.Layer< : yield* Effect.result( providerAuth.tryHandlePromptCommand({ instanceId: authInstanceId, - text: projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }), + text: userText, hasAttachments: false, }), ); @@ -950,10 +967,6 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const userText = projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }); // Delivered once: this run's provider turn marks the work as told. A // restart continuation is prompted by its own text or resumes natively. const noteContinuation = isRestartNoteContinuation( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index a0de3392631c7..db9ac7cf2ef10 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -4,6 +4,7 @@ import { assert, describe, it } from "@effect/vitest"; import { CommandId, type ChatAttachment, + EnvironmentId, EventId, MessageId, type ModelSelection, @@ -21,6 +22,10 @@ import { TurnItemId, ProviderDriverKind, } from "@t3tools/contracts"; +import { + expandAssistantCitationsForProvider, + serializeAssistantCitation, +} from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -2961,6 +2966,110 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live("sends cited assistant quotes expanded in the turn and in the handoff", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("provider-switch-citations"); + const capturedTurns = yield* Ref.make>([]); + const citedPrompt = `Why ${serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment:citations"), + threadId, + messageId: MessageId.make("message:citations:source"), + text: "The release color is violet.", + comment: "Is this still true?", + start: 0, + end: 28, + prefix: "", + suffix: "", + })}`; + const expandedPrompt = expandAssistantCitationsForProvider(citedPrompt); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const send = (ordinal: number, text: string, modelSelection: ModelSelection) => + Effect.gen(function* () { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`command:citations:${ordinal}`), + threadId, + messageId: MessageId.make(`message:citations:${ordinal}`), + createdBy: "user", + creationSource: "web", + text, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + event.payload.status === "completed", + ), + Stream.runHead, + ); + yield* worker.drain(); + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("command:citations:create"), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Citations", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* send(1, citedPrompt, CODEX_MODEL_SELECTION); + yield* send(2, "Continue", CLAUDE_MODEL_SELECTION); + return yield* orchestrator.getThreadProjection(threadId); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "provider-switch-citations" }, + layerRegistry, + ), + ), + ); + const turns = yield* Ref.get(capturedTurns); + + assert.equal(turns[0]?.text, expandedPrompt); + assert.include(turns[1]?.text ?? "", "Context handoff (full_thread_summary):"); + assert.include(turns[1]?.text ?? "", expandedPrompt); + for (const turn of turns) assert.notInclude(turn.text, "t3-citation:"); + // The stored message keeps its clickable link. + assert.equal( + projection.turnItems.find((item) => item.type === "user_message")?.text, + citedPrompt, + ); + }), + ), + ); + it.live("resolves a Claude fork into portable Codex context on first dispatch", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/provider-core/src/server/handoffBudget.ts b/packages/provider-core/src/server/handoffBudget.ts index 042745f688263..298ea926a3166 100644 --- a/packages/provider-core/src/server/handoffBudget.ts +++ b/packages/provider-core/src/server/handoffBudget.ts @@ -8,6 +8,7 @@ import type { OrchestrationV2ProviderThread, OrchestrationV2TurnItem, } from "@t3tools/contracts"; +import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations"; import * as Config from "effect/Config"; @@ -142,7 +143,10 @@ export function historicalMessage( ): OrchestrationV2HistoricalMessage | null { let text: string; switch (item.type) { + // History is provider input; the stored message keeps its clickable citations. case "user_message": + text = expandAssistantCitationsForProvider(item.text); + break; case "assistant_message": text = item.text; break; From 48824b7aef6287eb3d63d04125ac2f2ec378a0b0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:48:00 +0000 Subject: [PATCH 2/6] fix(server): reject oversized citation messages before dispatch --- .../orchestration-v2/ProviderMessageText.ts | 8 +-- .../ProviderTurnControlService.test.ts | 53 ++++++++++++--- .../ThreadManagementService.test.ts | 64 ++++++++++++++++++ .../ThreadManagementService.ts | 23 +++++-- .../ProviderSwitch.integration.test.ts | 67 +++++++++++++++---- .../provider-core/src/server/handoffBudget.ts | 8 ++- 6 files changed, 189 insertions(+), 34 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderMessageText.ts b/apps/server/src/orchestration-v2/ProviderMessageText.ts index 6f78a6ebf2e80..59dde151d28f3 100644 --- a/apps/server/src/orchestration-v2/ProviderMessageText.ts +++ b/apps/server/src/orchestration-v2/ProviderMessageText.ts @@ -17,12 +17,8 @@ class ProviderInputTooLongError extends Schema.TaggedError +it.effect("steers with saved quotes expanded after composer context projection", () => Effect.gen(function* () { const now = yield* DateTime.now; const threadId = ThreadId.make("thread:steer-citation"); @@ -362,7 +360,7 @@ it.effect("steers with cited assistant quotes expanded for the provider", () => updatedAt: now, }; const projection = makeProjection({ now, threadId, providerThread, providerTurnId, attemptId }); - const text = `Revisit ${serializeAssistantCitation({ + const citation = serializeAssistantCitation({ version: 1, environmentId: EnvironmentId.make("environment:steer-citation"), threadId, @@ -372,16 +370,40 @@ it.effect("steers with cited assistant quotes expanded for the provider", () => end: 18, prefix: "", suffix: "", - })}`; - const message = { + }); + const text = `Revisit ${citation} with [Terminal](t3-context://v1/terminal/ctx_quote)`; + const message: OrchestrationV2ConversationMessage = { id: MessageId.make("message:steer-citation"), + threadId, + runId: RunId.make("run:steer-citation"), + nodeId: null, + role: "user", + streaming: false, + createdAt: now, + updatedAt: now, text, + context: { + version: 1, + records: [ + { + version: 1, + contextId: ComposerContextId.make("ctx_quote"), + kind: "terminal", + label: "Terminal", + terminalId: "term-1", + terminalLabel: "Terminal", + lineStart: 1, + lineEnd: 1, + text: citation, + }, + ], + }, attachments: [], createdBy: "user", creationSource: "web", - } as unknown as OrchestrationV2ConversationMessage; + }; const steered = yield* Ref.make>([]); - const runtime: ProviderAdapterV2SessionRuntime = { + const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { instanceId: providerInstanceId, driver, providerSessionId, @@ -439,6 +461,15 @@ it.effect("steers with cited assistant quotes expanded for the provider", () => }); }).pipe(Effect.provide(layerControl)); - assert.deepEqual(yield* Ref.get(steered), [expandAssistantCitationsForProvider(text)]); + const texts = yield* Ref.get(steered); + assert.lengthOf(texts, 1); + const sent = texts[0]!; + assert.include(sent, "Revisit [assistant-quote-1] with [Terminal: Terminal; ref=ctx_quote]"); + assert.include(sent, ''); + assert.include(sent, "1 | [assistant-quote-1]"); + assert.include(sent, '"text": "Ship it on Friday."'); + assert.notInclude(sent, "t3-citation:"); + assert.equal(message.text, text); + assert.include(JSON.stringify(message.context), citation); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 1f05d1c8d4dee..42d099dc8241b 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -1,6 +1,8 @@ import { expect, it } from "@effect/vitest"; import { CommandId, + EnvironmentId, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, MessageId, NodeId, type OrchestrationV2Command, @@ -18,6 +20,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; @@ -39,6 +42,7 @@ import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; +import { userFacingDispatchErrorMessage } from "./UserFacingErrors.ts"; import { DispatchModeLimit, type DispatchModeRefusal } from "./DispatchModeLimit.ts"; it("stamps authoritative provenance on commands that create threads or messages", () => { @@ -427,6 +431,66 @@ it.effect( }, ); +it.effect("rejects an oversized expanded steer to the caller before committing it", () => { + const dispatched: Array = []; + const layerTest = layerDispatchChokepointTest( + ServerSettings.layerTest({ + providerInstances: { + [ProviderInstanceId.make("codex_work")]: { driver: "codex", enabled: true }, + }, + }), + )(dispatched); + const citation = serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment:steer-citation"), + threadId: ThreadId.make("thread:steer-citation:source"), + messageId: MessageId.make("message:steer-citation:source"), + text: "Quoted claim", + start: 0, + end: 12, + prefix: "", + suffix: "", + }); + const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; + const command = { + type: "message.dispatch", + commandId: CommandId.make("command:steer-citation:oversized"), + threadId: ThreadId.make("thread:thread-management:chokepoint"), + messageId: MessageId.make("message:steer-citation:oversized"), + text, + attachments: [], + dispatchMode: { type: "steer_active", targetRunId: RunId.make("run:active") }, + createdBy: "user", + creationSource: "web", + } satisfies OrchestrationV2Command; + + return Effect.gen(function* () { + const threads = yield* ThreadManagementService.ThreadManagementService; + const result = yield* Effect.result(threads.dispatch(command)); + + expect(result._tag).toBe("Failure"); + if (result._tag === "Failure") { + expect(userFacingDispatchErrorMessage(result.failure)).toMatch( + /over the .*character limit.*quotes are expanded/, + ); + } + expect(dispatched).toEqual([]); + + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; + yield* receipts.insertIfAbsent({ + commandId: command.commandId, + threadId: command.threadId, + commandType: command.type, + acceptedAt: yield* DateTime.now, + resultSequence: 1, + status: "accepted", + error: null, + }); + yield* threads.dispatch(command); + expect(dispatched).toEqual(["message.dispatch"]); + }).pipe(Effect.provide(layerTest)); +}); + it.effect( "a replay of an already-accepted command id dispatches despite the instance now being disabled", () => { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d83dd9578c279..28949f5e767ed 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -51,6 +51,7 @@ import * as Orchestrator from "./Orchestrator.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts"; import { projectTurnItemForDetail } from "./WireProjection.ts"; @@ -670,10 +671,24 @@ const make = Effect.gen(function* () { }; const dispatch: ThreadManagementServiceShape["dispatch"] = (command) => - ensureCommandTranscripts(command).pipe( - Effect.andThen(assertDispatchTargetEnabled(command)), - Effect.andThen(orchestrator.dispatch(command)), - ); + Effect.gen(function* () { + yield* ensureCommandTranscripts(command); + yield* assertDispatchTargetEnabled(command); + // Receipts replay the original outcome regardless of the retry's text. + if (command.type === "message.dispatch" && !(yield* hasReceipt(command.commandId))) { + yield* providerUserMessageText(command).pipe( + Effect.mapError( + (cause) => + new Orchestrator.OrchestratorCommandRejectedError({ + commandId: command.commandId, + commandType: command.type, + cause, + }), + ), + ); + } + return yield* orchestrator.dispatch(command); + }); const getProjectThread: ThreadManagementServiceShape["getProjectThread"] = (input) => getThreadProjection(input.threadId).pipe( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index a90ce383535bf..61ed8cdaa5be7 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -3,11 +3,13 @@ import { historyResponseItems } from "@t3tools/provider-core/server/handoffBudge import { assert, describe, it } from "@effect/vitest"; import { CommandId, + ComposerContextId, type ChatAttachment, EnvironmentId, EventId, MessageId, type ModelSelection, + type OrchestrationMessageContext, type OrchestrationV2Command, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, @@ -22,10 +24,7 @@ import { TurnItemId, ProviderDriverKind, } from "@t3tools/contracts"; -import { - expandAssistantCitationsForProvider, - serializeAssistantCitation, -} from "@t3tools/shared/assistantCitations"; +import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -2959,8 +2958,33 @@ describe("orchestration v2 provider switching", () => { end: 28, prefix: "", suffix: "", - })}`; - const expandedPrompt = expandAssistantCitationsForProvider(citedPrompt); + })} with [Terminal](t3-context://v1/terminal/ctx_handoff_quote)`; + const context: OrchestrationMessageContext = { + version: 1, + records: [ + { + version: 1, + contextId: ComposerContextId.make("ctx_handoff_quote"), + kind: "terminal", + label: "Terminal", + terminalId: "term-1", + terminalLabel: "Terminal", + lineStart: 1, + lineEnd: 1, + text: serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment:citations"), + threadId, + messageId: MessageId.make("message:citations:context-source"), + text: "Context-only quote.", + start: 0, + end: 19, + prefix: "", + suffix: "", + }), + }, + ], + }; const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ makeTestAdapter({ instanceId: CODEX_MODEL_SELECTION.instanceId, @@ -2982,7 +3006,12 @@ describe("orchestration v2 provider switching", () => { const projection = yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const send = (ordinal: number, text: string, modelSelection: ModelSelection) => + const send = ( + ordinal: number, + text: string, + modelSelection: ModelSelection, + context?: OrchestrationMessageContext, + ) => Effect.gen(function* () { yield* orchestrator.dispatch({ type: "message.dispatch", @@ -2992,6 +3021,7 @@ describe("orchestration v2 provider switching", () => { createdBy: "user", creationSource: "web", text, + ...(context === undefined ? {} : { context }), attachments: [], modelSelection, dispatchMode: { type: "start_immediately" }, @@ -3021,7 +3051,7 @@ describe("orchestration v2 provider switching", () => { branch: null, worktreePath: cwd, }); - yield* send(1, citedPrompt, CODEX_MODEL_SELECTION); + yield* send(1, citedPrompt, CODEX_MODEL_SELECTION, context); yield* send(2, "Continue", CLAUDE_MODEL_SELECTION); return yield* orchestrator.getThreadProjection(threadId); }).pipe( @@ -3034,15 +3064,28 @@ describe("orchestration v2 provider switching", () => { ); const turns = yield* Ref.get(capturedTurns); - assert.equal(turns[0]?.text, expandedPrompt); + assert.lengthOf(turns, 2); + assert.include(turns[0]?.text ?? "", "Why [assistant-quote-1]"); assert.include(turns[1]?.text ?? "", "Context handoff (full_thread_summary):"); - assert.include(turns[1]?.text ?? "", expandedPrompt); - for (const turn of turns) assert.notInclude(turn.text, "t3-citation:"); - // The stored message keeps its clickable link. + for (const turn of turns) { + assert.include(turn.text, '"text": "The release color is violet."'); + assert.include(turn.text, '"comment": "Is this still true?"'); + assert.include(turn.text, '"text": "Context-only quote."'); + assert.include(turn.text, ''); + assert.notInclude(turn.text, "t3-citation:"); + assert.notInclude(turn.text, "t3-context:"); + } + assert.equal( + projection.messages.find( + (message) => message.id === MessageId.make("message:citations:1"), + )?.text, + citedPrompt, + ); assert.equal( projection.turnItems.find((item) => item.type === "user_message")?.text, citedPrompt, ); + assert.deepEqual(projection.messages[0]?.context, context); }), ), ); diff --git a/packages/provider-core/src/server/handoffBudget.ts b/packages/provider-core/src/server/handoffBudget.ts index 71ba389b2362e..4ebdfbe01bb1f 100644 --- a/packages/provider-core/src/server/handoffBudget.ts +++ b/packages/provider-core/src/server/handoffBudget.ts @@ -9,6 +9,7 @@ import type { OrchestrationV2TurnItem, } from "@t3tools/contracts"; import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations"; +import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import * as Config from "effect/Config"; @@ -143,7 +144,12 @@ export function historicalMessage( switch (item.type) { // History is provider input; the stored message keeps its clickable citations. case "user_message": - text = expandAssistantCitationsForProvider(item.text); + text = expandAssistantCitationsForProvider( + projectComposerContextForProvider({ + text: item.text, + records: item.context?.records ?? [], + }), + ); break; case "assistant_message": text = item.text; From d4f4cac558433a2ccc715726c683f878b15677d7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:00:06 +0000 Subject: [PATCH 3/6] fix(server): validate queued citation messages before steering --- .../ThreadManagementService.test.ts | 200 ++++++++++++++++++ .../ThreadManagementService.ts | 65 +++++- 2 files changed, 254 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 42d099dc8241b..87f3db585cb3a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -2,6 +2,7 @@ import { expect, it } from "@effect/vitest"; import { CommandId, EnvironmentId, + EventId, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, MessageId, NodeId, @@ -18,6 +19,7 @@ import { ProjectId, ProviderInstanceId, RunId, + RunAttemptId, ThreadId, } from "@t3tools/contracts"; import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; @@ -38,6 +40,7 @@ import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import { emptyProjection } from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -491,6 +494,203 @@ it.effect("rejects an oversized expanded steer to the caller before committing i }).pipe(Effect.provide(layerTest)); }); +it.effect( + "rejects oversized queue edits and promotion of older oversized messages before committing", + () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:queue-citation"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const providerThreadId = ProviderThreadId.make("provider-thread:queue-citation"); + const modelSelection = { instanceId: providerInstanceId, model: "test" }; + const runs: Array = (["running", "queued"] as const).map( + (status, index) => ({ + id: RunId.make(`run:queue-citation:${status}`), + threadId, + ordinal: index + 1, + providerInstanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make(`message:queue-citation:${status}`), + rootNodeId: NodeId.make(`node:queue-citation:${status}`), + activeAttemptId: RunAttemptId.make(`attempt:queue-citation:${status}`), + status, + queuePosition: status === "queued" ? 1 : null, + requestedAt: now, + startedAt: status === "running" ? now : null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }), + ); + let projection = emptyProjection({ + id: EventId.make("event:queue-citation:created"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + id: threadId, + projectId: ProjectId.make("project:queue-citation"), + title: "Queued citations", + providerInstanceId, + modelSelection, + createdBy: "user", + creationSource: "web", + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThreadId, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + projection = { + ...projection, + runs, + nodes: runs.map((run) => ({ + id: run.rootNodeId!, + threadId, + runId: run.id, + parentNodeId: null, + rootNodeId: run.rootNodeId!, + kind: "root_turn", + status: run.status === "running" ? "running" : "pending", + countsForRun: true, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: run.startedAt, + completedAt: null, + })), + attempts: runs.map((run) => ({ + id: run.activeAttemptId!, + runId: run.id, + attemptOrdinal: 1, + rootNodeId: run.rootNodeId!, + providerInstanceId, + providerThreadId, + providerTurnId: null, + reason: "initial", + status: run.status === "running" ? "running" : "pending", + startedAt: run.startedAt, + completedAt: null, + })), + messages: runs.map((run) => ({ + id: run.userMessageId, + threadId, + runId: run.id, + nodeId: run.rootNodeId, + role: "user", + text: "Keep this message", + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + })), + }; + const citation = serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("environment:queue-citation"), + threadId, + messageId: MessageId.make("message:queue-citation:source"), + text: "Quoted claim", + start: 0, + end: 12, + prefix: "", + suffix: "", + }); + const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; + const dispatched: Array = []; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadRecords: () => Effect.succeed(projection), + dispatch: (command) => + Effect.sync(() => { + dispatched.push(command.type); + return { sequence: 1, storedEvents: [] }; + }), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provideMerge(layerReceiptsTest), + ); + yield* Effect.gen(function* () { + const threads = yield* ThreadManagementService.ThreadManagementService; + const edit = { + type: "queued-run.edit", + commandId: CommandId.make("command:queue-citation:edit"), + threadId, + runId: runs[1]!.id, + text, + } satisfies OrchestrationV2Command; + const editError = yield* threads.dispatch(edit).pipe(Effect.flip); + expect(editError).toBeInstanceOf(Orchestrator.OrchestratorCommandRejectedError); + expect(userFacingDispatchErrorMessage(editError)).toMatch( + /over the .*character limit.*quotes are expanded/, + ); + expect(projection.runs).toEqual(runs); + expect(projection.messages.map((message) => message.text)).toEqual([ + "Keep this message", + "Keep this message", + ]); + expect(dispatched).toEqual([]); + + // Simulate a message persisted before citation validation existed. + projection = { + ...projection, + messages: projection.messages.map((message) => + message.runId === edit.runId ? { ...message, text } : message, + ), + }; + const beforePromotion = projection; + const promote = { + type: "queued-message.promote-to-steer", + commandId: CommandId.make("command:queue-citation:promote"), + threadId, + queuedRunId: edit.runId, + targetRunId: runs[0]!.id, + } satisfies OrchestrationV2Command; + const promoteError = yield* threads.dispatch(promote).pipe(Effect.flip); + expect(promoteError).toBeInstanceOf(Orchestrator.OrchestratorCommandRejectedError); + expect(userFacingDispatchErrorMessage(promoteError)).toMatch( + /over the .*character limit.*quotes are expanded/, + ); + expect(projection).toEqual(beforePromotion); + expect(projection.runs.map((run) => run.status)).toEqual(["running", "queued"]); + expect(dispatched).toEqual([]); + + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; + for (const command of [edit, promote]) { + yield* receipts.insertIfAbsent({ + commandId: command.commandId, + threadId, + commandType: command.type, + acceptedAt: now, + resultSequence: 1, + status: "accepted", + error: null, + }); + yield* threads.dispatch(command); + } + expect(dispatched).toEqual([edit.type, promote.type]); + }).pipe(Effect.provide(layerTest)); + }), +); + it.effect( "a replay of an already-accepted command id dispatches despite the instance now being disabled", () => { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 28949f5e767ed..16304b34b9e4d 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -675,17 +675,60 @@ const make = Effect.gen(function* () { yield* ensureCommandTranscripts(command); yield* assertDispatchTargetEnabled(command); // Receipts replay the original outcome regardless of the retry's text. - if (command.type === "message.dispatch" && !(yield* hasReceipt(command.commandId))) { - yield* providerUserMessageText(command).pipe( - Effect.mapError( - (cause) => - new Orchestrator.OrchestratorCommandRejectedError({ - commandId: command.commandId, - commandType: command.type, - cause, - }), - ), - ); + if ( + (command.type === "message.dispatch" || + command.type === "queued-run.edit" || + command.type === "queued-message.promote-to-steer") && + !(yield* hasReceipt(command.commandId)) + ) { + let message: Pick | undefined; + if (command.type === "message.dispatch") { + message = command; + } else if (command.type !== "queued-run.edit" || command.text.trim().length > 0) { + const projection = yield* orchestrator.getThreadRecords( + command.threadId, + command.type === "queued-run.edit" + ? ["runs", "messages"] + : ["runs", "messages", "nodes", "attempts"], + { messageRoles: ["user"] }, + ); + const runId = command.type === "queued-run.edit" ? command.runId : command.queuedRunId; + const run = projection.runs.find((candidate) => candidate.id === runId); + const storedMessage = projection.messages.find( + (candidate) => candidate.id === run?.userMessageId, + ); + // Leave invalid queue targets to the orchestrator's existing command errors. + if ( + run?.status === "queued" && + storedMessage !== undefined && + storedMessage.delegatedCompletion === undefined && + storedMessage.notification === undefined + ) { + if (command.type === "queued-run.edit") { + message = { text: command.text, context: command.context ?? storedMessage.context }; + } else if ( + command.type === "queued-message.promote-to-steer" && + projection.thread.archivedAt === null && + projection.thread.deletedAt === null && + projection.nodes.some((candidate) => candidate.id === run.rootNodeId) && + projection.attempts.some((candidate) => candidate.id === run.activeAttemptId) + ) { + message = storedMessage; + } + } + } + if (message !== undefined) { + yield* providerUserMessageText(message).pipe( + Effect.mapError( + (cause) => + new Orchestrator.OrchestratorCommandRejectedError({ + commandId: command.commandId, + commandType: command.type, + cause, + }), + ), + ); + } } return yield* orchestrator.dispatch(command); }); From 88988d529cbc14efca37c6c1c11a38b51512a4db Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:08:19 +0000 Subject: [PATCH 4/6] fix(server): expose citation validation to MCP queue tools --- apps/server/src/mcp/toolkits/core.test.ts | 4 ++++ .../ThreadManagementService.test.ts | 16 +++++++++++++--- .../orchestration-v2/ThreadManagementService.ts | 2 +- 3 files changed, 18 insertions(+), 4 deletions(-) diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index aaa5b176db468..184214dd420b8 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -238,6 +238,10 @@ it("bounds public command rejections and redacts internal dispatch causes", () = dispatchFailure(new OrchestratorDispatchError({ ...command, cause: "🙂".repeat(1001) })) .message, ).toBe("🙂".repeat(1000)); + expect( + dispatchFailure(new OrchestratorCommandRejectedError({ ...command, cause: "🙂".repeat(1001) })) + .message, + ).toBe("🙂".repeat(1000)); expect( dispatchFailure( new OrchestratorCommandRejectedError({ ...command, cause: "Run is not queued." }), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 87f3db585cb3a..989da142d74fb 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -38,6 +38,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { dispatchFailure } from "../mcp/threadAccess.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { emptyProjection } from "./ProjectionStore.ts"; @@ -642,7 +643,12 @@ it.effect( expect(userFacingDispatchErrorMessage(editError)).toMatch( /over the .*character limit.*quotes are expanded/, ); - expect(projection.runs).toEqual(runs); + expect(dispatchFailure(editError)).toMatchObject({ + code: "orchestration_error", + message: expect.stringMatching( + /over the .*character limit.*quotes are expanded.*Shorten it or quote less/, + ), + }); expect(projection.messages.map((message) => message.text)).toEqual([ "Keep this message", "Keep this message", @@ -656,7 +662,6 @@ it.effect( message.runId === edit.runId ? { ...message, text } : message, ), }; - const beforePromotion = projection; const promote = { type: "queued-message.promote-to-steer", commandId: CommandId.make("command:queue-citation:promote"), @@ -669,7 +674,12 @@ it.effect( expect(userFacingDispatchErrorMessage(promoteError)).toMatch( /over the .*character limit.*quotes are expanded/, ); - expect(projection).toEqual(beforePromotion); + expect(dispatchFailure(promoteError)).toMatchObject({ + code: "orchestration_error", + message: expect.stringMatching( + /over the .*character limit.*quotes are expanded.*Shorten it or quote less/, + ), + }); expect(projection.runs.map((run) => run.status)).toEqual(["running", "queued"]); expect(dispatched).toEqual([]); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 16304b34b9e4d..5d13bd8583db1 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -724,7 +724,7 @@ const make = Effect.gen(function* () { new Orchestrator.OrchestratorCommandRejectedError({ commandId: command.commandId, commandType: command.type, - cause, + cause: cause.message, }), ), ); From 64f9d72d344b713ae37b7b7070754eec8e38437d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:25:32 +0000 Subject: [PATCH 5/6] fix(server): validate citations under the thread dispatch lock --- .../src/orchestration-v2/Orchestrator.ts | 56 +++ .../ThreadManagementService.test.ts | 274 -------------- .../ThreadManagementService.ts | 57 --- .../src/orchestration-v2/runtimeLayer.test.ts | 342 ++++++++++++++++++ 4 files changed, 398 insertions(+), 331 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 29ba67bdf6efe..c92e7498b3eb5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -127,6 +127,7 @@ import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import * as RunExecutionService from "./RunExecutionService.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; import { RuntimePolicyV2 } from "./RuntimePolicy.ts"; import { @@ -11207,6 +11208,61 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (shell !== null) yield* refuseAboveDispatchModeLimit(command, threadId, shell); } + if ( + command.type === "message.dispatch" || + command.type === "queued-run.edit" || + command.type === "queued-message.promote-to-steer" + ) { + let message: Pick | undefined; + if (command.type === "message.dispatch") { + message = command; + } else if (command.type !== "queued-run.edit" || command.text.trim().length > 0) { + const projection = yield* loadProjectionForCommand( + command, + command.type === "queued-run.edit" + ? ["runs", "messages"] + : ["runs", "messages", "nodes", "attempts"], + { messageRoles: ["user"] }, + ); + const runId = command.type === "queued-run.edit" ? command.runId : command.queuedRunId; + const run = projection.runs.find((candidate) => candidate.id === runId); + const storedMessage = projection.messages.find( + (candidate) => candidate.id === run?.userMessageId, + ); + // Leave invalid queue targets to the orchestrator's existing command errors. + if ( + run?.status === "queued" && + storedMessage !== undefined && + storedMessage.delegatedCompletion === undefined && + storedMessage.notification === undefined + ) { + if (command.type === "queued-run.edit") { + message = { text: command.text, context: command.context ?? storedMessage.context }; + } else if ( + command.type === "queued-message.promote-to-steer" && + projection.thread.archivedAt === null && + projection.thread.deletedAt === null && + projection.nodes.some((candidate) => candidate.id === run.rootNodeId) && + projection.attempts.some((candidate) => candidate.id === run.activeAttemptId) + ) { + message = storedMessage; + } + } + } + if (message !== undefined) { + yield* providerUserMessageText(message).pipe( + Effect.mapError( + (cause) => + new OrchestratorCommandRejectedError({ + commandId: command.commandId, + commandType: command.type, + cause: cause.message, + }), + ), + ); + } + } + const plan = yield* dispatchOnce(command).pipe( Effect.flatMap((planned) => planned.events.length > 0 || diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 989da142d74fb..1f05d1c8d4dee 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -1,9 +1,6 @@ import { expect, it } from "@effect/vitest"; import { CommandId, - EnvironmentId, - EventId, - PROVIDER_SEND_TURN_MAX_INPUT_CHARS, MessageId, NodeId, type OrchestrationV2Command, @@ -19,10 +16,8 @@ import { ProjectId, ProviderInstanceId, RunId, - RunAttemptId, ThreadId, } from "@t3tools/contracts"; -import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; @@ -38,15 +33,12 @@ import * as TestClock from "effect/testing/TestClock"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; -import { dispatchFailure } from "../mcp/threadAccess.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; -import { emptyProjection } from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; -import { userFacingDispatchErrorMessage } from "./UserFacingErrors.ts"; import { DispatchModeLimit, type DispatchModeRefusal } from "./DispatchModeLimit.ts"; it("stamps authoritative provenance on commands that create threads or messages", () => { @@ -435,272 +427,6 @@ it.effect( }, ); -it.effect("rejects an oversized expanded steer to the caller before committing it", () => { - const dispatched: Array = []; - const layerTest = layerDispatchChokepointTest( - ServerSettings.layerTest({ - providerInstances: { - [ProviderInstanceId.make("codex_work")]: { driver: "codex", enabled: true }, - }, - }), - )(dispatched); - const citation = serializeAssistantCitation({ - version: 1, - environmentId: EnvironmentId.make("environment:steer-citation"), - threadId: ThreadId.make("thread:steer-citation:source"), - messageId: MessageId.make("message:steer-citation:source"), - text: "Quoted claim", - start: 0, - end: 12, - prefix: "", - suffix: "", - }); - const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; - const command = { - type: "message.dispatch", - commandId: CommandId.make("command:steer-citation:oversized"), - threadId: ThreadId.make("thread:thread-management:chokepoint"), - messageId: MessageId.make("message:steer-citation:oversized"), - text, - attachments: [], - dispatchMode: { type: "steer_active", targetRunId: RunId.make("run:active") }, - createdBy: "user", - creationSource: "web", - } satisfies OrchestrationV2Command; - - return Effect.gen(function* () { - const threads = yield* ThreadManagementService.ThreadManagementService; - const result = yield* Effect.result(threads.dispatch(command)); - - expect(result._tag).toBe("Failure"); - if (result._tag === "Failure") { - expect(userFacingDispatchErrorMessage(result.failure)).toMatch( - /over the .*character limit.*quotes are expanded/, - ); - } - expect(dispatched).toEqual([]); - - const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; - yield* receipts.insertIfAbsent({ - commandId: command.commandId, - threadId: command.threadId, - commandType: command.type, - acceptedAt: yield* DateTime.now, - resultSequence: 1, - status: "accepted", - error: null, - }); - yield* threads.dispatch(command); - expect(dispatched).toEqual(["message.dispatch"]); - }).pipe(Effect.provide(layerTest)); -}); - -it.effect( - "rejects oversized queue edits and promotion of older oversized messages before committing", - () => - Effect.gen(function* () { - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:queue-citation"); - const providerInstanceId = ProviderInstanceId.make("codex"); - const providerThreadId = ProviderThreadId.make("provider-thread:queue-citation"); - const modelSelection = { instanceId: providerInstanceId, model: "test" }; - const runs: Array = (["running", "queued"] as const).map( - (status, index) => ({ - id: RunId.make(`run:queue-citation:${status}`), - threadId, - ordinal: index + 1, - providerInstanceId, - modelSelection, - providerThreadId, - userMessageId: MessageId.make(`message:queue-citation:${status}`), - rootNodeId: NodeId.make(`node:queue-citation:${status}`), - activeAttemptId: RunAttemptId.make(`attempt:queue-citation:${status}`), - status, - queuePosition: status === "queued" ? 1 : null, - requestedAt: now, - startedAt: status === "running" ? now : null, - completedAt: null, - checkpointId: null, - contextHandoffId: null, - }), - ); - let projection = emptyProjection({ - id: EventId.make("event:queue-citation:created"), - type: "thread.created", - threadId, - occurredAt: now, - payload: { - id: threadId, - projectId: ProjectId.make("project:queue-citation"), - title: "Queued citations", - providerInstanceId, - modelSelection, - createdBy: "user", - creationSource: "web", - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: providerThreadId, - lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, - forkedFrom: null, - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, - }, - }); - projection = { - ...projection, - runs, - nodes: runs.map((run) => ({ - id: run.rootNodeId!, - threadId, - runId: run.id, - parentNodeId: null, - rootNodeId: run.rootNodeId!, - kind: "root_turn", - status: run.status === "running" ? "running" : "pending", - countsForRun: true, - providerThreadId, - providerTurnId: null, - nativeItemRef: null, - runtimeRequestId: null, - checkpointScopeId: null, - startedAt: run.startedAt, - completedAt: null, - })), - attempts: runs.map((run) => ({ - id: run.activeAttemptId!, - runId: run.id, - attemptOrdinal: 1, - rootNodeId: run.rootNodeId!, - providerInstanceId, - providerThreadId, - providerTurnId: null, - reason: "initial", - status: run.status === "running" ? "running" : "pending", - startedAt: run.startedAt, - completedAt: null, - })), - messages: runs.map((run) => ({ - id: run.userMessageId, - threadId, - runId: run.id, - nodeId: run.rootNodeId, - role: "user", - text: "Keep this message", - attachments: [], - streaming: false, - createdBy: "user", - creationSource: "web", - createdAt: now, - updatedAt: now, - })), - }; - const citation = serializeAssistantCitation({ - version: 1, - environmentId: EnvironmentId.make("environment:queue-citation"), - threadId, - messageId: MessageId.make("message:queue-citation:source"), - text: "Quoted claim", - start: 0, - end: 12, - prefix: "", - suffix: "", - }); - const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; - const dispatched: Array = []; - const layerTest = ThreadManagementService.layer.pipe( - Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), - Layer.provide( - Layer.mock(Orchestrator.OrchestratorV2)({ - getThreadRecords: () => Effect.succeed(projection), - dispatch: (command) => - Effect.sync(() => { - dispatched.push(command.type); - return { sequence: 1, storedEvents: [] }; - }), - }), - ), - Layer.provide(ServerSettings.layerTest()), - Layer.provideMerge(layerReceiptsTest), - ); - yield* Effect.gen(function* () { - const threads = yield* ThreadManagementService.ThreadManagementService; - const edit = { - type: "queued-run.edit", - commandId: CommandId.make("command:queue-citation:edit"), - threadId, - runId: runs[1]!.id, - text, - } satisfies OrchestrationV2Command; - const editError = yield* threads.dispatch(edit).pipe(Effect.flip); - expect(editError).toBeInstanceOf(Orchestrator.OrchestratorCommandRejectedError); - expect(userFacingDispatchErrorMessage(editError)).toMatch( - /over the .*character limit.*quotes are expanded/, - ); - expect(dispatchFailure(editError)).toMatchObject({ - code: "orchestration_error", - message: expect.stringMatching( - /over the .*character limit.*quotes are expanded.*Shorten it or quote less/, - ), - }); - expect(projection.messages.map((message) => message.text)).toEqual([ - "Keep this message", - "Keep this message", - ]); - expect(dispatched).toEqual([]); - - // Simulate a message persisted before citation validation existed. - projection = { - ...projection, - messages: projection.messages.map((message) => - message.runId === edit.runId ? { ...message, text } : message, - ), - }; - const promote = { - type: "queued-message.promote-to-steer", - commandId: CommandId.make("command:queue-citation:promote"), - threadId, - queuedRunId: edit.runId, - targetRunId: runs[0]!.id, - } satisfies OrchestrationV2Command; - const promoteError = yield* threads.dispatch(promote).pipe(Effect.flip); - expect(promoteError).toBeInstanceOf(Orchestrator.OrchestratorCommandRejectedError); - expect(userFacingDispatchErrorMessage(promoteError)).toMatch( - /over the .*character limit.*quotes are expanded/, - ); - expect(dispatchFailure(promoteError)).toMatchObject({ - code: "orchestration_error", - message: expect.stringMatching( - /over the .*character limit.*quotes are expanded.*Shorten it or quote less/, - ), - }); - expect(projection.runs.map((run) => run.status)).toEqual(["running", "queued"]); - expect(dispatched).toEqual([]); - - const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; - for (const command of [edit, promote]) { - yield* receipts.insertIfAbsent({ - commandId: command.commandId, - threadId, - commandType: command.type, - acceptedAt: now, - resultSequence: 1, - status: "accepted", - error: null, - }); - yield* threads.dispatch(command); - } - expect(dispatched).toEqual([edit.type, promote.type]); - }).pipe(Effect.provide(layerTest)); - }), -); - it.effect( "a replay of an already-accepted command id dispatches despite the instance now being disabled", () => { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 5d13bd8583db1..7bfead29062e7 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -51,7 +51,6 @@ import * as Orchestrator from "./Orchestrator.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; -import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts"; import { projectTurnItemForDetail } from "./WireProjection.ts"; @@ -674,62 +673,6 @@ const make = Effect.gen(function* () { Effect.gen(function* () { yield* ensureCommandTranscripts(command); yield* assertDispatchTargetEnabled(command); - // Receipts replay the original outcome regardless of the retry's text. - if ( - (command.type === "message.dispatch" || - command.type === "queued-run.edit" || - command.type === "queued-message.promote-to-steer") && - !(yield* hasReceipt(command.commandId)) - ) { - let message: Pick | undefined; - if (command.type === "message.dispatch") { - message = command; - } else if (command.type !== "queued-run.edit" || command.text.trim().length > 0) { - const projection = yield* orchestrator.getThreadRecords( - command.threadId, - command.type === "queued-run.edit" - ? ["runs", "messages"] - : ["runs", "messages", "nodes", "attempts"], - { messageRoles: ["user"] }, - ); - const runId = command.type === "queued-run.edit" ? command.runId : command.queuedRunId; - const run = projection.runs.find((candidate) => candidate.id === runId); - const storedMessage = projection.messages.find( - (candidate) => candidate.id === run?.userMessageId, - ); - // Leave invalid queue targets to the orchestrator's existing command errors. - if ( - run?.status === "queued" && - storedMessage !== undefined && - storedMessage.delegatedCompletion === undefined && - storedMessage.notification === undefined - ) { - if (command.type === "queued-run.edit") { - message = { text: command.text, context: command.context ?? storedMessage.context }; - } else if ( - command.type === "queued-message.promote-to-steer" && - projection.thread.archivedAt === null && - projection.thread.deletedAt === null && - projection.nodes.some((candidate) => candidate.id === run.rootNodeId) && - projection.attempts.some((candidate) => candidate.id === run.activeAttemptId) - ) { - message = storedMessage; - } - } - } - if (message !== undefined) { - yield* providerUserMessageText(message).pipe( - Effect.mapError( - (cause) => - new Orchestrator.OrchestratorCommandRejectedError({ - commandId: command.commandId, - commandType: command.type, - cause: cause.message, - }), - ), - ); - } - } return yield* orchestrator.dispatch(command); }); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1b6a7de8a944f..0fde0622141bd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -10,6 +10,10 @@ import { CheckpointId, CheckpointRef, CommandId, + ComposerContextId, + EnvironmentId, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + type OrchestrationV2Command, ContextTransferId, EventId, MessageId, @@ -29,6 +33,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -42,6 +47,9 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import { dispatchFailure } from "../mcp/threadAccess.ts"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as ServerConfig from "../config.ts"; @@ -268,6 +276,7 @@ const layerTest = Layer.mergeAll( ProjectStore.layer, ProjectionStore.layer, EffectOutbox.layer, + CommandReceiptStore.layer, ThreadCommandExecutor.layer, ).pipe( Layer.provide(McpSessionRegistryTestkit.layer), @@ -4937,6 +4946,339 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ); }); +it.layer(layerTest)("provider citation validation", (it) => { + const citation = serializeAssistantCitation({ + version: 1, + environmentId: EnvironmentId.make("citation-environment"), + threadId: ThreadId.make("citation-source-thread"), + messageId: MessageId.make("citation-source-message"), + text: "Quoted claim", + start: 0, + end: 12, + prefix: "", + suffix: "", + }); + const setupQueue = Effect.fnUntraced(function* (key: string) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + const threadId = ThreadId.make(`citation-${key}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Citation validation", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + for (const [index, text] of ["Active", "Queued"].entries()) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message:${index}`), + threadId, + messageId: MessageId.make(`${threadId}:message:${index}`), + text, + attachments: [], + dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" }, + }); + } + const initial = yield* orchestrator.getThreadProjection(threadId); + const active = initial.runs[0]!; + const queued = initial.runs[1]!; + const providerThread = initial.providerThreads[0]!; + const now = yield* DateTime.now; + const providerSession = { + id: providerThread.providerSessionId!, + driver, + providerInstanceId: modelSelection.instanceId, + status: "running" as const, + cwd: process.cwd(), + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${threadId}:running`), + type: "run.updated", + threadId, + runId: active.id, + occurredAt: now, + payload: { ...active, status: "running", startedAt: now }, + }, + { + id: EventId.make(`${threadId}:session`), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: providerSession, + }, + { + id: EventId.make(`${threadId}:turn`), + type: "provider-turn.updated", + threadId, + runId: active.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make(`${threadId}:turn`), + providerThreadId: providerThread.id, + nodeId: active.rootNodeId!, + runAttemptId: active.activeAttemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ], + }); + const sessionSpy = vi + .spyOn(sessions, "get") + .mockReturnValue( + Effect.succeed( + Option.some({ providerSession } as ProviderAdapter.ProviderAdapterV2SessionRuntime), + ), + ); + yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore())); + return { orchestrator, eventSink, threadId, active, queued }; + }); + + it.effect("rejects oversized citations before commit and replays accepted receipts", () => + Effect.gen(function* () { + const { orchestrator, eventSink, threadId, active, queued } = yield* setupQueue("rejection"); + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const text = `${"x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS - citation.length - 1)} ${citation}`; + const commands: Array = [ + { + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:steer`), + threadId, + messageId: MessageId.make(`${threadId}:steer`), + text, + attachments: [], + dispatchMode: { type: "steer_active", targetRunId: active.id }, + }, + { + type: "queued-run.edit", + commandId: CommandId.make(`${threadId}:edit`), + threadId, + runId: queued.id, + text, + }, + { + type: "queued-message.promote-to-steer", + commandId: CommandId.make(`${threadId}:promote`), + threadId, + queuedRunId: queued.id, + targetRunId: active.id, + }, + ]; + for (const command of commands) { + if (command.type === "queued-message.promote-to-steer") { + const projection = yield* orchestrator.getThreadProjection(threadId); + // Older databases can already contain messages that exceed the expanded limit. + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${threadId}:legacy-message`), + type: "message.updated", + threadId, + runId: queued.id, + occurredAt: yield* DateTime.now, + payload: { + ...projection.messages.find((row) => row.id === queued.userMessageId)!, + text, + }, + }, + ], + }); + } + const before = yield* orchestrator.getThreadProjection(threadId); + const result = yield* Effect.result(orchestrator.dispatch(command)); + assert.equal(result._tag, "Failure", command.type); + if (result._tag !== "Failure") return; + assert.instanceOf(result.failure, Orchestrator.OrchestratorCommandRejectedError); + const mcpError = dispatchFailure(result.failure); + assert.equal(mcpError.code, "orchestration_error"); + assert.match( + mcpError.message, + /over the .*character limit.*quotes are expanded.*Shorten it or quote less/, + ); + assert.deepEqual(yield* orchestrator.getThreadProjection(threadId), before); + assert.isTrue(Option.isNone(yield* receipts.getByCommandId(command.commandId))); + assert.deepEqual(yield* outbox.listByCommandId(command.commandId), []); + assert.deepEqual(yield* eventSink.readByCommandId(command).pipe(Stream.runCollect), []); + + if (command.type === "message.dispatch" || command.type === "queued-run.edit") { + const accepted = yield* orchestrator.dispatch({ ...command, text: "Shortened message" }); + const after = yield* orchestrator.getThreadProjection(threadId); + const effects = yield* outbox.listByCommandId(command.commandId); + assert.deepEqual(yield* orchestrator.dispatch(command), accepted); + assert.deepEqual(yield* orchestrator.getThreadProjection(threadId), after); + assert.deepEqual(yield* outbox.listByCommandId(command.commandId), effects); + } + } + }), + ); + + it.effect( + "rejects concurrent citation edits using the context committed under the thread lock", + () => + Effect.gen(function* () { + const { orchestrator, eventSink, threadId, queued } = yield* setupQueue("race"); + const threads = yield* ThreadManagementService.ThreadManagementService; + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const context = (length: number) => ({ + version: 1 as const, + records: [ + { + version: 1 as const, + contextId: ComposerContextId.make("ctx_quote"), + kind: "terminal" as const, + label: "Terminal", + terminalId: "term-1", + terminalLabel: "Terminal", + lineStart: 1, + lineEnd: 1, + text: "c".repeat(length), + }, + ], + }); + const shortText = "Read [Terminal](t3-context://v1/terminal/ctx_quote)"; + const oldContext = context(10_000); + const newContext = context(26_000); + const longText = `${"x".repeat(100_000)} ${citation} ${shortText}`; + yield* threads.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make(`${threadId}:initial-context`), + threadId, + runId: queued.id, + text: shortText, + context: oldContext, + }); + const before = yield* orchestrator.getThreadProjection(threadId); + assert.isBelow( + (yield* providerUserMessageText({ text: longText, context: oldContext })).length, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + ); + assert.isBelow( + (yield* providerUserMessageText({ text: shortText, context: newContext })).length, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + ); + const combinedError = yield* providerUserMessageText({ + text: longText, + context: newContext, + }).pipe(Effect.flip); + assert.isAbove(combinedError.inputLength, PROVIDER_SEND_TURN_MAX_INPUT_CHARS); + + const firstCommandId = CommandId.make(`${threadId}:context-edit`); + const secondCommandId = CommandId.make(`${threadId}:text-edit`); + const commitReady = yield* Deferred.make(); + const releaseCommit = yield* Deferred.make(); + const secondQueued = yield* Deferred.make(); + const commitCommand = eventSink.commitCommand; + const withLock = executor.withLock; + const commitSpy = vi + .spyOn(eventSink, "commitCommand") + .mockImplementation((input) => + input.commandId === firstCommandId + ? Deferred.succeed(commitReady, undefined).pipe( + Effect.andThen(Deferred.await(releaseCommit)), + Effect.andThen(commitCommand(input)), + ) + : commitCommand(input), + ); + let lockRequests = 0; + const observeLock: typeof withLock = (key, effect) => + key === threadId && ++lockRequests === 2 + ? Deferred.succeed(secondQueued, undefined).pipe(Effect.andThen(withLock(key, effect))) + : withLock(key, effect); + const lockSpy = vi.spyOn(executor, "withLock").mockImplementation(observeLock); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + commitSpy.mockRestore(); + lockSpy.mockRestore(); + }), + ); + const first = yield* threads + .dispatch({ + type: "queued-run.edit", + commandId: firstCommandId, + threadId, + runId: queued.id, + text: shortText, + context: newContext, + }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.raceFirst( + Deferred.await(commitReady), + Fiber.join(first).pipe( + Effect.andThen(Effect.die("Context edit bypassed the commit barrier.")), + ), + ); + const second = yield* threads + .dispatch({ + type: "queued-run.edit", + commandId: secondCommandId, + threadId, + runId: queued.id, + text: longText, + }) + .pipe(Effect.result, Effect.forkChild({ startImmediately: true })); + yield* Effect.raceFirst( + Deferred.await(secondQueued), + Fiber.join(second).pipe( + Effect.andThen(Effect.die("Text edit completed before reaching the occupied lock.")), + ), + ); + yield* Deferred.succeed(releaseCommit, undefined); + yield* Fiber.join(first); + const result = yield* Fiber.join(second); + assert.equal( + result._tag, + "Failure", + "the second edit must validate the newly committed context", + ); + if (result._tag !== "Failure") return; + assert.instanceOf(result.failure, Orchestrator.OrchestratorCommandRejectedError); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(after.runs, before.runs); + assert.deepEqual(after.nodes, before.nodes); + assert.deepEqual(after.attempts, before.attempts); + assert.deepEqual( + after.messages.find((row) => row.id === queued.userMessageId), + { + ...before.messages.find((row) => row.id === queued.userMessageId)!, + context: newContext, + }, + ); + assert.isTrue(Option.isNone(yield* receipts.getByCommandId(secondCommandId))); + assert.deepEqual(yield* outbox.listByCommandId(secondCommandId), []); + assert.deepEqual( + yield* eventSink.readByCommandId({ commandId: secondCommandId }).pipe(Stream.runCollect), + [], + ); + }), + ); +}); + it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", (it) => { it.effect("interrupts a pending provider start without launching provider work", () => Effect.gen(function* () { From 233f598e3bb482301643d0929c0d6a8fd59a9277 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 06:54:27 +0000 Subject: [PATCH 6/6] test(server): adopt main's receipt replay flag and succeedSome in citation tests AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/runtimeLayer.test.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 212f1ac8d3271..125a4438bc076 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -7221,9 +7221,7 @@ it.layer(layerTest)("provider citation validation", (it) => { const sessionSpy = vi .spyOn(sessions, "get") .mockReturnValue( - Effect.succeed( - Option.some({ providerSession } as ProviderAdapter.ProviderAdapterV2SessionRuntime), - ), + Effect.succeedSome({ providerSession } as ProviderAdapter.ProviderAdapterV2SessionRuntime), ); yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore())); return { orchestrator, eventSink, threadId, active, queued };