fix(chat): keep an injection made during the turn that consumed the lane

Consuming the instructions lane marked the blocks read but left them in it, so
an injection made in that turn's onTurnComplete queued behind them and the next
turn's clear destroyed both. Turn 1 carried its instruction and every turn after
it silently carried none, which is worse than the per-read draining it replaced.

The consumed blocks now move to turn-scoped state, so a second options build in
the same turn still sees them while the lane holds only what is pending. Also
guards the stash lookup: outside a turn both sides of the turn comparison are
undefined, so the optional-chained check matched and dereferenced nothing.

Author: ericallam

packages/trigger-sdk/src/v3/ai.ts

@@ -2712,10 +2712,18 @@ const chatBackgroundQueueKey = locals.create<ModelMessage[]>("chat.backgroundQue
 const chatInjectedInstructionsKey = locals.create<SystemModelMessage[]>(
   "chat.injectedInstructions"
 );
-/** The turn that consumed the instructions lane, so a second read in the same turn still sees it. */
-const chatInstructionsConsumedTurnKey = locals.create<number | undefined>(
-  "chat.injectedInstructionsConsumedTurn"
-);
+/**
+ * What a turn already consumed from the instructions lane, so a second
+ * `toStreamTextOptions()` call in the same turn sees the same blocks.
+ *
+ * Consumed blocks are moved here rather than left in the pending lane: leaving
+ * them there means an injection made during the consumed turn sits behind them,
+ * and clearing the lane on the next turn destroys both.
+ */
+const chatInstructionsConsumedKey = locals.create<{
+  turn: number;
+  blocks: SystemModelMessage[];
+}>("chat.injectedInstructionsConsumed");
 
 /**
  * Run-scoped pipe counter. Stored in locals so concurrent runs in the
@@ -4734,28 +4742,34 @@ function toStreamTextOptions(options?: ToStreamTextOptionsOptions): Record<strin
    * caching, and the addition reads as a later amendment. A changed prefix does
    * cost the first call its cache hit, on turns that actually injected.
    */
+  /**
+   * Consumed once per turn, not once per read, and moved out of the lane rather
+   * than marked read in place.
+   *
+   * Per turn, because a `run()` that builds options twice (a cheap classifier
+   * pass and then the answer) has to see the injection in both, and draining on
+   * read hands it to whichever call ran first. Moved out, because blocks left in
+   * the lane sit in front of anything injected during the same turn, and
+   * clearing the lane on the next turn then destroys both. Outside a turn there
+   * is no turn to scope the stash to, so the lane drains on read there.
+   */
   const injectedInstructions = locals.get(chatInjectedInstructionsKey);
-  if (injectedInstructions && injectedInstructions.length > 0) {
-    /**
-     * Consumed once per turn, not once per read. A `run()` that builds options
-     * twice, a cheap classifier pass and then the answer, has to see the
-     * injection in both: draining on read hands it to whichever call ran first
-     * and drops it from the rest without saying so. Outside a turn there is no
-     * turn to scope that to, so the lane drains on read there instead.
-     */
-    const currentTurn = locals.get(chatTurnContextKey)?.turn;
-    const consumedTurn = locals.get(chatInstructionsConsumedTurnKey);
-
-    let blocks: SystemModelMessage[];
-    if (currentTurn === undefined) {
-      blocks = injectedInstructions.splice(0);
-    } else if (consumedTurn !== undefined && consumedTurn !== currentTurn) {
-      injectedInstructions.length = 0;
-      blocks = [];
-    } else {
-      locals.set(chatInstructionsConsumedTurnKey, currentTurn);
-      blocks = injectedInstructions;
+  const currentTurn = locals.get(chatTurnContextKey)?.turn;
+  const consumedThisTurn =
+    currentTurn === undefined ? undefined : locals.get(chatInstructionsConsumedKey);
+
+  let injectedBlocks: SystemModelMessage[] = [];
+  if (consumedThisTurn && consumedThisTurn.turn === currentTurn) {
+    injectedBlocks = consumedThisTurn.blocks;
+  } else if (injectedInstructions && injectedInstructions.length > 0) {
+    injectedBlocks = injectedInstructions.splice(0);
+    if (currentTurn !== undefined) {
+      locals.set(chatInstructionsConsumedKey, { turn: currentTurn, blocks: injectedBlocks });
     }
+  }
+
+  if (injectedBlocks.length > 0) {
+    const blocks = injectedBlocks;
 
     const injectedText = blocks
       .map((block) => (typeof block.content === "string" ? block.content : ""))

packages/trigger-sdk/test/inject-system-instructions.test.ts

@@ -289,4 +289,58 @@ describe("chat.inject with a system role", () => {
       await harness.close();
     }
   });
+
+  it("gives each turn only its own injection, over consecutive turns", async () => {
+    /**
+     * Consuming the lane has to move the blocks out of it, not mark them read in
+     * place. Left in place, an injection made during the consumed turn queues
+     * behind them and the next turn's clear destroys both: turn 1 gets its
+     * instruction and every turn after it silently gets none.
+     */
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: textStream("ok") }),
+    });
+
+    let n = 0;
+
+    const agent = chat.agent({
+      id: "inject-system-consecutive",
+      onTurnComplete: async () => {
+        n++;
+        chat.inject([{ role: "system", content: `INJECT-${n}` }]);
+      },
+      run: async ({ messages, signal }) =>
+        streamText({
+          ...chat.toStreamTextOptions(),
+          model,
+          messages,
+          abortSignal: signal,
+        }),
+    });
+
+    const harness = mockChatAgent(agent, { chatId: "inject-system-consecutive" });
+
+    try {
+      for (const id of ["u1", "u2", "u3", "u4"]) {
+        await harness.sendMessage({ id, role: "user", parts: [{ type: "text", text: id }] });
+        await new Promise((r) => setTimeout(r, 40));
+      }
+
+      const injectionsSeenOn = (turn: number) => {
+        const system = JSON.stringify(
+          model.doStreamCalls[turn]!.prompt.filter((m) => m.role === "system")
+        );
+        return ["INJECT-1", "INJECT-2", "INJECT-3"].filter((key) => system.includes(key));
+      };
+
+      // Turn 0 predates any injection; after that each turn carries exactly the
+      // one injected at the end of the turn before it.
+      expect(injectionsSeenOn(0)).toEqual([]);
+      expect(injectionsSeenOn(1)).toEqual(["INJECT-1"]);
+      expect(injectionsSeenOn(2)).toEqual(["INJECT-2"]);
+      expect(injectionsSeenOn(3)).toEqual(["INJECT-3"]);
+    } finally {
+      await harness.close();
+    }
+  });
 });