Tighten message stream boundary checks

eslint.config.mjs

@@ -711,7 +711,7 @@ export default defineConfig([
             {
               group: nonAppServerBannedAppServerProtocolImportPatterns,
               message:
-                "Source modules outside app-server must use domain models and app-server services instead of app-server protocol modules. The turn display/history protocol remains the only feature-side exception.",
+                "Source modules outside app-server must use domain models and app-server services instead of app-server protocol modules. Chat ingestion and message-stream conversion may consume app-server turn protocol at the boundary; feature state and UI must use Panel-owned models.",
             },
             {
               group: generatedAppServerSourceImportPatterns,

src/app-server/protocol/file-change.ts

@@ -0,0 +1,3 @@
+import type { FileUpdateChange as GeneratedFileUpdateChange } from "../../generated/app-server/v2/FileUpdateChange";
+
+export type FileUpdateChange = GeneratedFileUpdateChange;

src/app-server/protocol/turn.ts

@@ -1,4 +1,3 @@
-import type { FileUpdateChange as GeneratedFileUpdateChange } from "../../generated/app-server/v2/FileUpdateChange";
 import type { ThreadItem as GeneratedTurnItem } from "../../generated/app-server/v2/ThreadItem";
 import type { Turn as GeneratedTurnRecord } from "../../generated/app-server/v2/Turn";
 import type { UserInput } from "../../generated/app-server/v2/UserInput";
@@ -9,7 +8,6 @@ import {
   type ThreadTranscriptEntry,
 } from "../../domain/threads/transcript";
 
-export type FileUpdateChange = GeneratedFileUpdateChange;
 export type TurnItem = GeneratedTurnItem;
 export type TurnRecord = GeneratedTurnRecord;
 

src/features/chat/message-stream/from-turn-items.ts

@@ -7,7 +7,8 @@ import type {
   MessageStreamPrimaryTarget,
 } from "./items";
 import type { HistoricalTurn } from "../../../domain/threads/history";
-import type { FileUpdateChange, TurnItem } from "../../../app-server/protocol/turn";
+import type { FileUpdateChange } from "../../../app-server/protocol/file-change";
+import type { TurnItem } from "../../../app-server/protocol/turn";
 import { definedProp } from "../../../utils";
 import { referencedThreadMetadataFromPrompt, type ReferencedThreadMetadata } from "../../../domain/threads/reference";
 import { turnUserItemText } from "../../../app-server/protocol/turn";

src/features/chat/message-stream/selectors.ts

@@ -20,6 +20,14 @@ export function forkCandidatesFromItems(items: readonly MessageStreamItem[]): re
   return [...turnOutcomeItemsByTurn.values()];
 }
 
+export function latestProposedPlanFromItems(items: readonly MessageStreamItem[]): MessageStreamItem | null {
+  return [...timelineItemsFromMessageStreamItems(items)].reverse().find((item) => item.semanticKind === "proposedPlan")?.streamItem ?? null;
+}
+
+export function latestImplementablePlanFromItems(items: readonly MessageStreamItem[]): MessageStreamItem | null {
+  return [...timelineItemsFromMessageStreamItems(items)].reverse().find((item) => item.actions.canImplementPlan)?.streamItem ?? null;
+}
+
 export function isAssistantAuthoredMessage(item: MessageStreamItem): item is AssistantAuthoredMessageStreamItem {
   return item.kind === "message" && (item.messageKind === "assistantResponse" || item.messageKind === "proposedPlan");
 }

src/features/chat/message-stream/streaming-items.ts

@@ -1,4 +1,4 @@
-import type { FileUpdateChange } from "../../../app-server/protocol/turn";
+import type { FileUpdateChange } from "../../../app-server/protocol/file-change";
 import type { MessageStreamItem, MessageStreamItemKind } from "./items";
 import { normalizeFileChanges } from "./from-turn-items";
 

src/features/chat/panel/snapshot.ts

@@ -1,6 +1,7 @@
 import type { OpenCodexPanelSnapshot } from "../../../workspace/open-panel-snapshot";
 import type { ChatState } from "../state/reducer";
 import type { MessageStreamItem } from "../message-stream/items";
+import { latestProposedPlanFromItems } from "../message-stream/selectors";
 import type { RestoredThreadState } from "../lifecycle";
 
 export function openPanelTurnLifecycle(state: ChatState["turn"]["lifecycle"]): OpenCodexPanelSnapshot["turnLifecycle"] {
@@ -10,7 +11,7 @@ export function openPanelTurnLifecycle(state: ChatState["turn"]["lifecycle"]): O
 }
 
 export function latestProposedPlanItem(items: readonly MessageStreamItem[]): MessageStreamItem | null {
-  return [...items].reverse().find((item) => item.kind === "message" && item.messageKind === "proposedPlan") ?? null;
+  return latestProposedPlanFromItems(items);
 }
 
 export function parseRestoredThreadState(state: unknown): RestoredThreadState | null {

src/features/chat/protocol/inbound/notification-plan.ts

@@ -1,12 +1,8 @@
 import { activeThreadSettingsAppliedAction } from "../../state/actions";
 import type { McpServerStartupStatus } from "../../../../domain/server/diagnostics";
 import { threadTokenUsageFromRuntimeUsage } from "../../../../domain/runtime/metrics";
-import {
-  completedConversationSummaryFromTurnRecord,
-  type FileUpdateChange,
-  type TurnItem,
-  type TurnRecord,
-} from "../../../../app-server/protocol/turn";
+import type { FileUpdateChange } from "../../../../app-server/protocol/file-change";
+import { completedConversationSummaryFromTurnRecord, type TurnItem, type TurnRecord } from "../../../../app-server/protocol/turn";
 import type { ServerNotification } from "../../../../app-server/connection/rpc-messages";
 import { normalizeExplicitThreadName } from "../../../../domain/threads/model";
 import type { ThreadConversationSummary } from "../../../../domain/threads/transcript";
@@ -491,13 +487,13 @@ function reconciledCompletedTurnItems(state: ChatState, turn: TurnRecord): reado
   const stateMessageStreamItems = items.map((item) => serverUserMessageForOptimisticItem(item, serverUserMessagesByClientId) ?? item);
   let mergedTurnItems = stateMessageStreamItems
     .filter((item) => item.turnId === turn.id)
-    .filter((item) => !isOptimisticUserMessage(item, serverUserClientIds, serverUserFallbackTexts));
+    .filter((item) => !isReconciledOptimisticUserMessage(item, turn.id, serverUserClientIds, serverUserFallbackTexts));
   for (const item of turnItems) {
     mergedTurnItems = upsertMessageStreamItemById(mergedTurnItems, item);
   }
   const retainedItems = stateMessageStreamItems
     .filter((item) => item.turnId !== turn.id)
-    .filter((item) => !isOptimisticUserMessage(item, serverUserClientIds, serverUserFallbackTexts));
+    .filter((item) => !isReconciledOptimisticUserMessage(item, turn.id, serverUserClientIds, serverUserFallbackTexts));
   return [...retainedItems, ...mergedTurnItems];
 }
 
@@ -535,9 +531,24 @@ function serverUserMessageForOptimisticItem(
   return serverUserMessagesByClientId.get(item.id) ?? null;
 }
 
-function isOptimisticUserMessage(item: MessageStreamItem, serverUserClientIds: Set<string>, serverUserFallbackTexts: Set<string>): boolean {
+function isReconciledOptimisticUserMessage(
+  item: MessageStreamItem,
+  completedTurnId: string,
+  serverUserClientIds: Set<string>,
+  serverUserFallbackTexts: Set<string>,
+): boolean {
   if (!isUserMessage(item) || !isLocalUserMessageId(item.id)) return false;
-  return serverUserClientIds.has(item.id) || serverUserFallbackTexts.has(item.text);
+  return serverUserClientIds.has(item.id) || isFallbackOptimisticUserMessageForTurn(item, completedTurnId, serverUserFallbackTexts);
+}
+
+function isFallbackOptimisticUserMessageForTurn(
+  item: MessageStreamMessageItem & { role: "user" },
+  completedTurnId: string,
+  serverUserFallbackTexts: Set<string>,
+): boolean {
+  if (serverUserFallbackTexts.size === 0) return false;
+  if (item.turnId && item.turnId !== completedTurnId) return false;
+  return serverUserFallbackTexts.has(item.copyText ?? item.text);
 }
 
 function isLocalUserMessageId(id: string): boolean {

src/features/chat/state/selectors.ts

@@ -9,6 +9,7 @@ import type {
 } from "./reducer";
 import type { Thread } from "../../../domain/threads/model";
 import type { MessageStreamItem } from "../message-stream/items";
+import { latestImplementablePlanFromItems } from "../message-stream/selectors";
 import { messageStreamItems, messageStreamIsEmpty } from "./message-stream";
 
 export interface SubmissionStateSnapshot {
@@ -32,7 +33,7 @@ export function listedThreads(state: ChatState): readonly Thread[] {
   return state.threadList.listedThreads;
 }
 
-export function displayItemsEmpty(state: ChatState): boolean {
+export function messageStreamItemsEmpty(state: ChatState): boolean {
   return messageStreamIsEmpty(state.messageStream);
 }
 
@@ -60,8 +61,5 @@ export function implementPlanCandidateFromState(state: {
   if (!state.activeThread.id || chatTurnBusy(state) || state.runtime.selectedCollaborationMode !== "plan") {
     return null;
   }
-  return (
-    [...messageStreamItems(state.messageStream)].reverse().find((item) => item.kind === "message" && item.messageKind === "proposedPlan") ??
-    null
-  );
+  return latestImplementablePlanFromItems(messageStreamItems(state.messageStream));
 }

src/features/chat/threads/resume-controller.ts

@@ -1,6 +1,6 @@
 import type { AppServerClient } from "../../../app-server/connection/client";
 import type { ThreadTokenUsage } from "../../../domain/runtime/metrics";
-import { activeThreadId, canSwitchToThread, displayItemsEmpty, listedThreads } from "../state/selectors";
+import { activeThreadId, canSwitchToThread, listedThreads, messageStreamItemsEmpty } from "../state/selectors";
 import type { ChatStateStore } from "../state/reducer";
 import type { RestorationController } from "./restoration-controller";
 import { resumedThreadActionFromAppServerResponse } from "./resume";
@@ -53,7 +53,7 @@ export class ResumeController {
       if (this.isStale(resume)) return;
       await this.host.syncThreadGoal(response.thread.id);
       if (this.isStale(resume)) return;
-      const renderFallbackMessage = displayItemsEmpty(this.host.stateStore.getState());
+      const renderFallbackMessage = messageStreamItemsEmpty(this.host.stateStore.getState());
       if (renderFallbackMessage) {
         this.host.addSystemMessage(resumedThreadMessage(response.thread.id));
       }

tests/features/chat/conversation/turns/plan-implementation.test.ts

@@ -18,6 +18,15 @@ const planItem = (id: string): MessageStreamItem => ({
   messageState: "completed",
 });
 
+const streamingPlanItem = (id: string): MessageStreamItem => ({
+  id,
+  kind: "message",
+  role: "assistant",
+  text: "Plan",
+  messageKind: "proposedPlan",
+  messageState: "streaming",
+});
+
 function resumeThread(stateStore: ChatStateStore, items: readonly MessageStreamItem[]): void {
   stateStore.dispatch({
     type: "active-thread/resumed",
@@ -71,6 +80,15 @@ describe("createPlanImplementation", () => {
     expect(implementPlanCandidateFromState(stateStore.getState())).toBe(latest);
   });
 
+  it("ignores streaming proposed plans until they are implementable turn outcomes", () => {
+    const stateStore = createChatStateStore(createChatState());
+    const completed = planItem("completed");
+    const streaming = streamingPlanItem("streaming");
+    resumeThread(stateStore, [completed, streaming]);
+
+    expect(implementPlanCandidateFromState(stateStore.getState())).toBe(completed);
+  });
+
   it("switches out of plan mode and submits the implementation prompt", async () => {
     const { controller, ensureConnected, requestDefaultCollaborationModeForNextTurn, sendTurnText, stateStore } = createController();
     const plan = planItem("plan");

tests/features/chat/protocol/inbound/controller.test.ts

@@ -380,7 +380,7 @@ describe("ChatInboundController", () => {
       expect(chatStateMessageStreamItems(state)[0]?.turnId).toBeUndefined();
     });
 
-    it("keeps repeated hook runs with the same run id as separate display items", () => {
+    it("keeps repeated hook runs with the same run id as separate message stream items", () => {
       const state = createChatState();
       state.activeThread.id = "thread-active";
       state.turn.lifecycle = { kind: "running", turnId: "turn-active" };
@@ -1136,7 +1136,7 @@ describe("ChatInboundController", () => {
       expect(chatStateMessageStreamItems(state).some((item) => item.id === "local-user-1")).toBe(false);
     });
 
-    it("reconciles optimistic user echoes by client id before falling back to text only when client ids are absent", () => {
+    it("reconciles optimistic user echoes by client id before falling back to same-turn text only when client ids are absent", () => {
       const state = createChatState();
       state.activeThread.id = "thread-active";
       state.turn.lifecycle = { kind: "running", turnId: "turn-active" };
@@ -1193,6 +1193,14 @@ describe("ChatInboundController", () => {
           text: "fallback text",
           turnId: "turn-active",
         },
+        {
+          id: "local-user-other-turn",
+          kind: "message",
+          messageKind: "user",
+          role: "user",
+          text: "fallback text",
+          turnId: "turn-other",
+        },
       ]);
       const fallbackControllerWithoutClientId = controllerForState(fallbackStateWithoutClientId);
 
@@ -1220,9 +1228,12 @@ describe("ChatInboundController", () => {
         },
       } satisfies Extract<ServerNotification, { method: "turn/completed" }>);
 
-      expect(chatStateMessageStreamItems(fallbackStateWithoutClientId)).toEqual([
-        expect.objectContaining({ id: "server-u1", text: "fallback text" }),
-      ]);
+      expect(chatStateMessageStreamItems(fallbackStateWithoutClientId)).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ id: "server-u1", text: "fallback text" }),
+          expect.objectContaining({ id: "local-user-other-turn", text: "fallback text" }),
+        ]),
+      );
       expect(chatStateMessageStreamItems(fallbackStateWithoutClientId).some((item) => item.id === "local-user-without-client-id")).toBe(
         false,
       );

tests/features/chat/threads/title-context.test.ts

@@ -8,7 +8,7 @@ import {
 } from "../../../../src/features/chat/threads/title-context";
 
 describe("chat thread title context", () => {
-  it("extracts title context from streamed display items when completed turn items are not loaded", () => {
+  it("extracts title context from streamed message stream items when completed turn items are not loaded", () => {
     expect(
       threadTitleContextFromMessageStreamItems("turn", [
         { id: "u1", kind: "message", messageKind: "user", role: "user", text: "自動命名を直したい", turnId: "turn" },
@@ -86,7 +86,7 @@ describe("chat thread title context", () => {
     });
   });
 
-  it("finds the first visible display item title context", () => {
+  it("finds the first visible message stream item title context", () => {
     expect(
       firstThreadTitleContextFromMessageStreamItems([
         { id: "u1", kind: "message", messageKind: "user", role: "user", text: "表示済み履歴から命名したい", turnId: "visible" },

tests/features/chat/view-snapshot.test.ts

@@ -16,7 +16,7 @@ describe("chat view snapshots", () => {
       latestProposedPlanItem([
         { id: "first", kind: "message", messageKind: "proposedPlan", role: "assistant", text: "plan", messageState: "completed" },
         { id: "user", kind: "message", messageKind: "user", role: "user", text: "ok" },
-        { id: "latest", kind: "message", messageKind: "proposedPlan", role: "assistant", text: "plan", messageState: "completed" },
+        { id: "latest", kind: "message", messageKind: "proposedPlan", role: "assistant", text: "plan", messageState: "streaming" },
       ])?.id,
     ).toBe("latest");
   });

tests/scripts/eslint-config.test.ts

@@ -109,8 +109,8 @@ export const convert = threadTokenUsageFromAppServerUsage;
     expect(messages).toContain("no-restricted-imports");
   });
 
-  it("allows turn protocol imports as the feature-side protocol exception", async () => {
-    const messages = await lintSource(
+  it("allows turn protocol imports at chat ingestion and message-stream conversion boundaries", async () => {
+    const conversionMessages = await lintSource(
       "src/features/chat/message-stream/from-turn-items.ts",
       `
 import type { TurnItem } from "../../../app-server/protocol/turn";
@@ -118,8 +118,17 @@ import type { TurnItem } from "../../../app-server/protocol/turn";
 export type Item = TurnItem;
 `,
     );
+    const ingestionMessages = await lintSource(
+      "src/features/chat/protocol/inbound/notification-plan.ts",
+      `
+import type { TurnRecord } from "../../../../app-server/protocol/turn";
 
-    expect(messages).not.toContain("no-restricted-imports");
+export type Turn = TurnRecord;
+`,
+    );
+
+    expect(conversionMessages).not.toContain("no-restricted-imports");
+    expect(ingestionMessages).not.toContain("no-restricted-imports");
   });
 
   it("reports direct ChatState alias mutation", async () => {