diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 07c1a544798be..83403156753df 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -175,6 +175,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 { isRuntimePolicyAboveCeilingError, RuntimePolicyV2 } from "./RuntimePolicy.ts"; import { @@ -13992,6 +13993,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, sideChatIds).pipe( Effect.tap((planned) => refuseAboveRuntimeModeCeiling(command, planned.events)), Effect.filterOrFail( diff --git a/apps/server/src/orchestration-v2/ProviderMessageText.ts b/apps/server/src/orchestration-v2/ProviderMessageText.ts new file mode 100644 index 0000000000000..59dde151d28f3 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderMessageText.ts @@ -0,0 +1,34 @@ +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.`; + } +} + +// Project context before expanding quotes, including citations inside context records. +// The durable message keeps its clickable links. +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 8b30a961c099d..3376e8a3a9870 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -1,8 +1,13 @@ import { assert, it } from "@effect/vitest"; import { + ComposerContextId, + EnvironmentId, + MessageId, type ModelSelection, NodeId, + type OrchestrationV2ConversationMessage, type OrchestrationV2ProviderThread, + type OrchestrationV2Run, type OrchestrationV2ThreadProjection, ProjectId, ProviderDriverKind, @@ -14,6 +19,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -347,3 +353,142 @@ it.effect( assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session"); }), ); + +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"); + 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 citation = 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 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", + }; + const steered = yield* Ref.make>([]); + const runtime: ProviderAdapter.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)); + + 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/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index bd225bb41773b..13656f9e12894 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, type NodeId, @@ -18,6 +17,7 @@ import * as Schema from "effect/Schema"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; +import { providerUserMessageText } from "./ProviderMessageText.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; const yieldToRuntime = Effect.yieldNow.pipe( @@ -366,6 +366,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, @@ -374,10 +375,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 7c58e6ad079e8..216d0cda697cb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -6,6 +6,7 @@ import { CheckpointScopeId, ContextHandoffId, ContextTransferId, + EnvironmentId, GitCommandError, MessageId, NodeId, @@ -21,7 +22,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 Deferred from "effect/Deferred"; @@ -1873,3 +1879,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 98652036118fc..bb36e5f1e59d9 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, @@ -55,6 +54,7 @@ import { knownProviderFailureMessage, 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 { @@ -548,6 +548,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; @@ -580,10 +600,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, }), ); @@ -837,10 +854,6 @@ export const layer: Layer.Layer< }); let effectiveHandoffs = handoffs; let resumeFellBack = false; - const userText = projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }); // Native goal commands go verbatim; handoff history waits a turn. const goalDriver = session.driver === "codex" || diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a49e323770cc1..125a4438bc076 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -13,6 +13,10 @@ import { CheckpointId, CheckpointRef, CommandId, + ComposerContextId, + EnvironmentId, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + type OrchestrationV2Command, ContextTransferId, EventId, MessageId, @@ -36,6 +40,7 @@ import { ScheduledTaskId, 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"; @@ -48,6 +53,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"; @@ -316,6 +324,7 @@ const layerTest = Layer.mergeAll( ProjectStore.layer, ProjectionStore.layer, EffectOutbox.layer, + CommandReceiptStore.layer, ThreadCommandExecutor.layer, ).pipe( Layer.provide(McpSessionRegistryTestkit.layer), @@ -7111,6 +7120,337 @@ 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.succeedSome({ 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, replayed: true }); + 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* () { 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 83eee8f66a5dc..c7628d6d29ba7 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -3,10 +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,6 +25,7 @@ import { TurnItemId, ProviderDriverKind, } from "@t3tools/contracts"; +import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -3792,4 +3796,152 @@ describe("orchestration v2 provider switching", () => { "reissues imported v1 context when a queued provider starts after the first provider fails", () => importedFailureScenario(true), ); + + 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: "", + })} 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, + 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, + context?: OrchestrationMessageContext, + ) => + 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, + ...(context === undefined ? {} : { context }), + 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, context); + 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.lengthOf(turns, 2); + assert.include(turns[0]?.text ?? "", "Why [assistant-quote-1]"); + assert.include(turns[1]?.text ?? "", "Context handoff (full_thread_summary):"); + 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 5f9f2f3ae33f6..c64ba6616d81e 100644 --- a/packages/provider-core/src/server/handoffBudget.ts +++ b/packages/provider-core/src/server/handoffBudget.ts @@ -8,6 +8,8 @@ import type { OrchestrationV2ProviderThread, OrchestrationV2TurnItem, } from "@t3tools/contracts"; +import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations"; +import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import * as Config from "effect/Config"; @@ -129,7 +131,15 @@ 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( + projectComposerContextForProvider({ + text: item.text, + records: item.context?.records ?? [], + }), + ); + break; case "assistant_message": text = item.text; break;