fix(hub): show a received message before the coordinator saves it

apps/hub/src/mail/conversation-routes.ts

@@ -25,7 +25,11 @@ import {
   createCoordinatorQuestionProseReader,
   type CoordinatorRepoDirSource,
 } from "./coordinator-question.ts";
-import { isRecord, projectRunTurns } from "./conversation-turns.ts";
+import {
+  isRecord,
+  projectRunTurns,
+  unsavedPersonTurns,
+} from "./conversation-turns.ts";
 import type { Actor } from "../lib/actor.ts";
 
 const log = getLogger(["corbits-builder", "hub", "conversation"]);
@@ -244,7 +248,11 @@ export function createConversationRoutes(
           }
           if (isRecord(payload)) {
             const live = await read(repoStore, current);
-            const turns = [...live.turns, ...(await priorTurns(repoStore))];
+            const turns = [
+              ...live.turns,
+              ...unsavedPersonTurns(events, live.turns),
+              ...(await priorTurns(repoStore)),
+            ];
             turns.sort((left, right) => left.at - right.at);
             if (live.threads !== undefined) payload["threads"] = live.threads;
             payload["turns"] = turns.slice(-maxConversationTurns);

apps/hub/src/mail/conversation-turns.test.ts

@@ -6,7 +6,10 @@ import { $ } from "bun";
 import { declineNote } from "../config/decline.ts";
 import { resumeMessage } from "../config/resume.ts";
 import { specialistEndedMessage } from "../config/specialist-ended.ts";
-import { readDurableConversationTurns } from "./conversation-turns.ts";
+import {
+  readDurableConversationTurns,
+  unsavedPersonTurns,
+} from "./conversation-turns.ts";
 
 const person = "b65b0e2f4c4a444db11b7321dc413964@sunny-harbor-panda.localhost";
 const dirs: string[] = [];
@@ -201,3 +204,48 @@ test("a resumed message is read after the separator, not at a quoted line that s
     { role: "person", text: "Real words" },
   ]);
 });
+
+const received = (seq: number, at: string, from: string, text: string) => ({
+  seq,
+  type: "SignalReceived",
+  body: {
+    type: "SignalReceived",
+    seq,
+    signalName: "input",
+    signalId: `<m${String(seq)}@example.com>`,
+    at,
+    payload: {
+      headers: { from, to: ["coordinator@example.com"] },
+      rawHeaders: {},
+      parts: [{ contentType: "text/plain", ref: "r", text }],
+    },
+  },
+});
+
+test("a person's received message reads as a turn until the coordinator saves it", () => {
+  const savedAt = Date.parse("2026-11-04T10:00:00.000Z");
+  const saved = [{ role: "person" as const, text: "First", at: savedAt }];
+  const events = [
+    received(1, "2026-11-04T09:59:59.900Z", person, "First"),
+    received(
+      2,
+      "2026-11-04T10:00:05.000Z",
+      "run_04ba732b8686649f77a2797393bafbc5@sunny-harbor-panda.localhost",
+      "From a run",
+    ),
+    received(3, "2026-11-04T10:00:06.000Z", person, "Second\r\n"),
+  ];
+  expect(unsavedPersonTurns(events, saved)).toEqual([
+    {
+      role: "person",
+      text: "Second",
+      at: Date.parse("2026-11-04T10:00:06.000Z"),
+    },
+  ]);
+  expect(
+    unsavedPersonTurns(events, [
+      ...saved,
+      { role: "person", text: "Second", at: savedAt + 6001 },
+    ]),
+  ).toEqual([]);
+});

apps/hub/src/mail/conversation-turns.ts

@@ -6,6 +6,7 @@ import { reconstructDurableConversation } from "@intx/workflow-host";
 import { deriveWorkflowRunRepoId } from "@intx/workflow-deploy";
 import { extractAddrSpec } from "@intx/mime";
 import { isRunAddress } from "@intx/types/agent-address";
+import { isMail } from "@intx/types/runtime";
 import { ApprovalDecision } from "@intx/types/signals";
 import { type } from "arktype";
 import {
@@ -88,6 +89,12 @@ function approvalText(
       });
 }
 
+function personText(body: string): string | undefined {
+  const text = withoutResumeContext(deliveredText(body));
+  if (text === undefined || text.trim() === "") return undefined;
+  return clippedTurnText(personTurnText(text));
+}
+
 // A missing or corrupt turn list is empty; it does not throw.
 function mapMirrorConversationTurns(
   turns: unknown,
@@ -118,22 +125,54 @@ function mapMirrorConversationTurns(
         mapped.push({ role: "person", text: clippedTurnText(text), at });
       continue;
     }
-    const text = withoutResumeContext(
-      deliveredText(raw.replace(inboundEnvelope, "")),
-    );
-    if (text === undefined || text.trim() === "") continue;
-    mapped.push({
-      role: "person",
-      text: clippedTurnText(personTurnText(text)),
-      at,
-    });
+    const text = personText(raw.replace(inboundEnvelope, ""));
+    if (text !== undefined) mapped.push({ role: "person", text, at });
   }
   mapped.sort((left, right) => left.at - right.at);
   return mapped.length > maxConversationTurns
     ? mapped.slice(-maxConversationTurns)
     : mapped;
 }
 
+/**
+ * The person's mail the run has received but its coordinator has not saved yet. The coordinator
+ * saves a message only with its reply, and takes a message in after the run commits its receipt,
+ * so a received message is unsaved while no saved person turn is as late as the receipt.
+ */
+export function unsavedPersonTurns(
+  events: unknown,
+  saved: readonly ConversationTextTurn[],
+): ConversationTextTurn[] {
+  if (!Array.isArray(events)) return [];
+  const latestSaved = Math.max(
+    -Infinity,
+    ...saved.flatMap((turn) => (turn.role === "person" ? [turn.at] : [])),
+  );
+  const unsaved: ConversationTextTurn[] = [];
+  for (const event of events) {
+    if (!isRecord(event) || event["type"] !== "SignalReceived") continue;
+    const body = event["body"];
+    if (!isRecord(body) || typeof body["at"] !== "string") continue;
+    const at = Date.parse(body["at"]);
+    const mail = body["payload"];
+    if (!(at > latestSaved) || !isMail(mail)) continue;
+    if (mail.headers.from !== undefined && runSent(mail.headers.from)) continue;
+    const text = personText(
+      mail.parts
+        .flatMap((part) =>
+          part.disposition !== "attachment" &&
+          part.contentType.startsWith("text/plain") &&
+          part.text !== undefined
+            ? [part.text]
+            : [],
+        )
+        .join(""),
+    );
+    if (text !== undefined) unsaved.push({ role: "person", text, at });
+  }
+  return unsaved;
+}
+
 /**
  * Person and assistant text from every agent-state directory in the run repo.
  * A missing or corrupt mirror is an empty list so the events read can still