fix(fork): one agent message in a row in the Chat + Replies main list

apps/server/src/fork/displayModeWrap/DisplayModeWrap.test.ts

@@ -54,6 +54,7 @@ const newId = ThreadId.make("thread:display-wrap:new");
 const rewrittenId = ThreadId.make("thread:display-wrap:rewritten");
 const latestId = ThreadId.make("thread:display-wrap:latest");
 const ownCallId = ThreadId.make("thread:display-wrap:own-call");
+const mainFirstId = ThreadId.make("thread:display-wrap:main-first");
 const switchId = ThreadId.make("thread:display-wrap:switch");
 const repliesId = ThreadId.make("thread:display-wrap:replies");
 const providerThreadId = ProviderThreadId.make("provider-thread:display-wrap");
@@ -658,6 +659,41 @@ it.effect("counts back from the caller's own item when it names one", () =>
   }).pipe(Effect.provide(Layer.merge(store, wrapper))),
 );
 
+it.effect("reads back to the latest message of the main list when asked, past many replies", () =>
+  Effect.gen(function* () {
+    const now = yield* DateTime.now;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const wrap = yield* DisplayModeWrap.DisplayModeWrap;
+    const news = itemUpdated(now, mainFirstId, runId, 2, { type: "send_to_user", text: "News." });
+    const replies = Array.from({ length: 125 }, (_, index) =>
+      itemUpdated(now, mainFirstId, runId, index + 3, {
+        type: "user_message",
+        messageId: MessageId.make(`message:display-wrap:main-first:${index + 3}`),
+        text: `Reply ${index + 3}`,
+        inputIntent: "steer",
+        under: `item:${mainFirstId}:2`,
+      }),
+    );
+    const events: ReadonlyArray<OrchestrationV2DomainEvent> = [
+      threadCreated(now, mainFirstId, null),
+      runCreated(now, mainFirstId, runId, "running"),
+      itemUpdated(now, mainFirstId, runId, 1, {
+        type: "user_message",
+        messageId: MessageId.make("message:display-wrap:main-first:1"),
+        text: "Ship it.",
+        inputIntent: "turn_start",
+      }),
+      news,
+      ...replies,
+    ];
+    yield* Effect.forEach(events, (event) => projections.apply(event), { discard: true });
+    const first = (items: ReadonlyArray<OrchestrationV2TurnItem>) => items[0]?.ordinal;
+    assert.strictEqual(first(yield* wrap.latestShownItems(mainFirstId, undefined, 120)), 8);
+    // The agent's news at 2 is the main list's latest message; the user's first message is not needed.
+    assert.strictEqual(first(yield* wrap.latestShownItems(mainFirstId, undefined, 120, true)), 2);
+  }).pipe(Effect.provide(Layer.merge(store, wrapper))),
+);
+
 it.effect("looks back past many of the user's messages to find the agent's earlier work", () =>
   Effect.gen(function* () {
     const now = yield* DateTime.now;

apps/server/src/fork/displayModeWrap/DisplayModeWrap.ts

@@ -10,6 +10,7 @@ import {
 import {
   type DisplayMode,
   displayEventFromTurnItem,
+  isMainListEvent,
   isWebTypedMessage,
   type MessageOrigin,
   NORMAL_DISPLAY_MODE,
@@ -90,12 +91,14 @@ export interface DisplayModeWrapShape {
   /**
    * The user messages and send calls that the screen shows, oldest first, in any state. Reads back
    * past the user's last typed message and at least `cores` messages (default WRAPPED_CORE_COUNT),
-   * counted from the item with `ownNativeId` back when given.
+   * counted from the item with `ownNativeId` back when given. With `toMainListMessage`, also past
+   * the latest message of the main list of a thread with sub-threads.
    */
   readonly latestShownItems: (
     threadId: ThreadId,
     ownNativeId?: string,
     cores?: number,
+    toMainListMessage?: boolean,
   ) => Effect.Effect<
     ReadonlyArray<OrchestrationV2TurnItem>,
     SqlError.SqlError | Schema.SchemaError | ProjectionStore.ProjectionStoreV2Error
@@ -198,13 +201,14 @@ export const layer: Layer.Layer<
 
     // Reads newest first from before `before`, or from the end, until `coreCount` messages
     // (not rewrites) and, without `before`, a typed message, counted from `ownNativeId` back when
-    // given. Visibility gets no items: they only decide interrupt results, which never enter the
-    // agent's transcript.
+    // given, and with `toMainListMessage`, a message of the main list. Visibility gets no items:
+    // they only decide interrupt results, which never enter the agent's transcript.
     const shownItems = Effect.fn("DisplayModeWrap.shownItems")(function* (
       threadId: ThreadId,
       before: MessageId | null,
       ownNativeId?: string,
       coreCount: number = WRAPPED_CORE_COUNT,
+      toMainListMessage = false,
     ) {
       const pageRows = Math.max(PAGE_ROWS, Math.min(coreCount, 64));
       let start = { ordinal: Number.MAX_SAFE_INTEGER, itemId: "" };
@@ -256,6 +260,7 @@ export const layer: Layer.Layer<
       let cores = 0;
       let passedTyped = before !== null;
       let passedOwn = ownNativeId === undefined;
+      let passedMainList = !toMainListMessage;
       for (const segment of segments) {
         let { ordinal, itemId } = segment.before;
         while (true) {
@@ -281,7 +286,8 @@ export const layer: Layer.Layer<
             const event = displayEventFromTurnItem(item);
             if (event?.type === "user") passedTyped = true;
             if (event?.type === "user" || event?.type === "sent") cores++;
-            if (passedTyped && cores >= coreCount) return found.reverse();
+            if (event !== null && isMainListEvent(event)) passedMainList = true;
+            if (passedTyped && passedMainList && cores >= coreCount) return found.reverse();
           }
           const last = rows.at(-1);
           if (last === undefined || rows.length < pageRows) break;
@@ -482,7 +488,9 @@ export const layer: Layer.Layer<
 
     const latestShownItems: DisplayModeWrapShape["latestShownItems"] = Effect.fn(
       "DisplayModeWrap.latestShownItems",
-    )((threadId, ownNativeId, cores) => shownItems(threadId, null, ownNativeId, cores));
+    )((threadId, ownNativeId, cores, toMainListMessage) =>
+      shownItems(threadId, null, ownNativeId, cores, toMainListMessage),
+    );
 
     return DisplayModeWrap.of({
       wrapUserText,

apps/server/src/fork/userMessageTools/SendToUser.ts

@@ -9,6 +9,7 @@ import {
   displayEventFromTurnItem,
   foldCores,
   freshFirstPointer,
+  isMainListEvent,
   messagePointer,
   recapLine,
   recapText,
@@ -353,6 +354,63 @@ function answerCalls(
   );
 }
 
+/** Whether a send still running would be shown as a new message: the handler would accept it. */
+function takenAsNewMessage(item: OrchestrationV2TurnItem) {
+  const text = inputField(item, "text");
+  return (
+    inputField(item, "new_sub_thread") === true &&
+    inputField(item, "ack") !== true &&
+    messagePointer(inputField(item, "reply_to")) === undefined &&
+    typeof text === "string" &&
+    text !== "" &&
+    jsonEscapedBytes(text) <= SEND_TO_USER_MAX_BYTES
+  );
+}
+
+/**
+ * The agent's messages in the main list since the user's latest message there: delivered ones, and
+ * with `named`, sends still running that the handler would take as a new message.
+ */
+function mainListSends(items: ReadonlyArray<OrchestrationV2TurnItem>, named: boolean) {
+  const at = items.findLastIndex((item) => {
+    const event = displayEventFromTurnItem(item);
+    return event?.type === "user" && isMainListEvent(event);
+  });
+  return {
+    anchor: items[at]?.id,
+    sends: replyCallsAfter(items, at, named, (item) => {
+      const event = displayEventFromTurnItem(item);
+      return event === null
+        ? takenAsNewMessage(item)
+        : event.type === "sent" && isMainListEvent(event);
+    }),
+  };
+}
+
+/** After a refused new message: how to change the agent's message in the main list, and what to answer. */
+function mainListTakenLines(cores: ReadonlyArray<VisibleCore>): ReadonlyArray<string> {
+  const since = cores
+    .slice(cores.findLastIndex((core) => core.from === "user" && !core.reply) + 1)
+    .filter((core) => core.from === "agent" && !core.reply);
+  const open = since.filter(
+    (core) => !cores.some((reply) => reply.from === "user" && reply.first === core.id),
+  );
+  return [
+    ...(since.length === 0
+      ? [
+          "That message is still being delivered. To change it once it shows, call replace_send_to_user with sub_thread set to the id that its send returned.",
+        ]
+      : open.length === 0
+        ? ["The user has replied under it, so it stays as it is."]
+        : [
+            "To change it, call replace_send_to_user with sub_thread set to its id:",
+            ...open.map(listedLine),
+          ]),
+    ...unansweredLines(cores.slice(-SUB_THREAD_WINDOW_CORES)),
+    DONE,
+  ];
+}
+
 const orchestrationError = (error: { readonly message: string }) =>
   new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message });
 
@@ -400,12 +458,15 @@ export const layerHandlers = McpToolAccess.toLayer(
     const holdKey = (threadId: ThreadId, userMessage: string | undefined) =>
       `${threadId}\n${userMessage ?? ""}`;
 
-    const readThreadCores = Effect.fn("SendToUser.readThreadCores")(function* (threadId: ThreadId) {
+    const readThreadCores = Effect.fn("SendToUser.readThreadCores")(function* (
+      threadId: ThreadId,
+      toMainListMessage: boolean,
+    ) {
       const client = yield* Effect.serviceOption(McpSchema.McpServerClient);
       const named = Option.getOrUndefined(client)?.requestMetadata?.["claudecode/toolUseId"];
       const toolUseId = typeof named === "string" ? named : undefined;
       const shown = yield* (yield* DisplayModeWrap)
-        .latestShownItems(threadId, toolUseId, SUB_THREAD_WINDOW_CORES)
+        .latestShownItems(threadId, toolUseId, SUB_THREAD_WINDOW_CORES, toMainListMessage)
         .pipe(Effect.mapError(orchestrationError));
       const items = itemsBefore(shown, toolUseId);
       // `cores` is what the agent saw when it made the call. `coresNow` has what the user typed since.
@@ -433,8 +494,11 @@ export const layerHandlers = McpToolAccess.toLayer(
         readonly ack?: boolean | null | undefined;
       },
     ) {
-      const { items, cores, named } = yield* readThreadCores(threadId);
       const replyTo = messagePointer(input.reply_to);
+      const { items, cores, named } = yield* readThreadCores(
+        threadId,
+        replyTo === undefined && input.new_sub_thread === true && input.ack !== true,
+      );
       if ((replyTo !== undefined) === (input.new_sub_thread === true)) {
         return yield* subThreadRefusal(
           "invalid_request",
@@ -456,8 +520,22 @@ export const layerHandlers = McpToolAccess.toLayer(
         );
       }
       if (replyTo === undefined) {
-        // Two first messages sent together get two ids: the first id is not in the history yet.
         const now = yield* Clock.currentTimeMillis;
+        const main = mainListSends(items, named);
+        const mainKey = holdKey(threadId, `main\n${main.anchor ?? ""}`);
+        if (main.sends.length > 0 || holds.held(mainKey, named, now) !== "open") {
+          return yield* subThreadRefusal(
+            "invalid_request",
+            [
+              "you already sent a new message to the main list, and the user has not written there since. You may not send two in a row there.",
+            ],
+            [],
+            mainListTakenLines(cores),
+          );
+        }
+        holds.record(mainKey, named, "answer", "answered", now);
+        // A user message to the main list between two first messages can come before the first is
+        // in the history, so the ids given for a while count as taken.
         for (const [id, until] of givenIds) if (until <= now) givenIds.delete(id);
         const id = freshFirstPointer(
           new Set([
@@ -542,7 +620,7 @@ export const layerHandlers = McpToolAccess.toLayer(
         readonly new_text: string;
       },
     ) {
-      const { items, cores, coresNow, named } = yield* readThreadCores(threadId);
+      const { items, cores, coresNow, named } = yield* readThreadCores(threadId, false);
       const replyTo = messagePointer(input.reply_to);
       const subThread = messagePointer(input.sub_thread);
       if ((replyTo === undefined) === (subThread === undefined)) {