fix(server): cited assistant quotes reach providers expanded again

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<OrchestrationV2ConversationMessage, "text" | "context"> | 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(

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>()(
+  "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<string, ProviderInputTooLongError> => {
+  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);
+};

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<ReadonlyArray<string>>([]);
+    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, '<context kind="terminal" id="ctx_quote">');
+    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);
+  }),
+);

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,

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" },
+      },
+    ]);
+  }),
+);