From 80dab3de0f976bd01e0c7549a3c49d2957b4095a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 14:40:48 -0800 Subject: [PATCH] fix(hub): show a received message before the coordinator saves it --- apps/hub/src/mail/conversation-routes.ts | 12 ++++- apps/hub/src/mail/conversation-turns.test.ts | 50 ++++++++++++++++- apps/hub/src/mail/conversation-turns.ts | 57 ++++++++++++++++---- 3 files changed, 107 insertions(+), 12 deletions(-) diff --git a/apps/hub/src/mail/conversation-routes.ts b/apps/hub/src/mail/conversation-routes.ts index 6e86bae6..1691b733 100644 --- a/apps/hub/src/mail/conversation-routes.ts +++ b/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); diff --git a/apps/hub/src/mail/conversation-turns.test.ts b/apps/hub/src/mail/conversation-turns.test.ts index 1f8cf3eb..9a38c8c1 100644 --- a/apps/hub/src/mail/conversation-turns.test.ts +++ b/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: ``, + 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([]); +}); diff --git a/apps/hub/src/mail/conversation-turns.ts b/apps/hub/src/mail/conversation-turns.ts index fccc9d38..2518f27f 100644 --- a/apps/hub/src/mail/conversation-turns.ts +++ b/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,15 +125,8 @@ 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 @@ -134,6 +134,45 @@ function mapMirrorConversationTurns( : 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