From 5908769ec9bdb4f20b2ccf35cc2de61f883c77ec Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 3 Nov 2026 17:53:39 -0800 Subject: [PATCH 1/2] fix(hub): read a person's message back with the line breaks they typed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mail delivery stores a message's line breaks as CRLF, and the conversation mirror returned them unchanged. The project page matches the message it sent against the mirror's copy by exact text, so a message with line breaks never matched and showed twice, and the extra copy kept "Thinking…" on. The mirror reads a person's text back as mail delivers it, by a shared rule: LF line breaks, no spaces or tabs at the end of a line. The specialist-ended and resume-context markers are written with LF, so the mirror now finds them in delivered text. The turn length limit and its clip move to the shared schema, so the page can clip a sent message the same way. --- apps/hub/src/config/resume.ts | 8 +- apps/hub/src/config/specialist-ended.ts | 10 ++- apps/hub/src/mail/conversation-threads.ts | 9 +-- apps/hub/src/mail/conversation-turns.test.ts | 78 +++++++++++++++++++- apps/hub/src/mail/conversation-turns.ts | 24 +++--- packages/schemas/src/conversation.test.ts | 12 +++ packages/schemas/src/conversation.ts | 30 +++++++- 7 files changed, 145 insertions(+), 26 deletions(-) create mode 100644 packages/schemas/src/conversation.test.ts diff --git a/apps/hub/src/config/resume.ts b/apps/hub/src/config/resume.ts index f11eeee1..e2a36a43 100644 --- a/apps/hub/src/config/resume.ts +++ b/apps/hub/src/config/resume.ts @@ -8,8 +8,6 @@ import { SPECIALIST_ENDED_LEAD } from "./specialist-ended.ts"; const RESUME_CONTEXT_LEAD = "Background: this project's conversation continues from an earlier session. This is where it stood, not a new request."; const RESUME_CONTEXT_END = "\n\nThe person's new message:\n\n"; -// Mail delivery turns every line break into CRLF. -const DELIVERED_CONTEXT_END = /\r?\n\r?\nThe person's new message:\r?\n\r?\n/; // Hubs before the held resume context mailed this opening on the person's behalf. const LEGACY_RESUME_OPENING = "I'm picking this project back up."; @@ -77,6 +75,8 @@ export function approvalDecisionLine({ export function withoutResumeContext(text: string): string | undefined { if (text.startsWith(LEGACY_RESUME_OPENING)) return undefined; if (!text.startsWith(RESUME_CONTEXT_LEAD)) return text; - const end = DELIVERED_CONTEXT_END.exec(text); - return end === null ? text : text.slice(end.index + end[0].length); + const end = text.indexOf(RESUME_CONTEXT_END); + if (end !== -1) return text.slice(end + RESUME_CONTEXT_END.length); + // Delivered text is trimmed, so a blank new message leaves the marker without its blank line. + return text.endsWith(RESUME_CONTEXT_END.trimEnd()) ? "" : text; } diff --git a/apps/hub/src/config/specialist-ended.ts b/apps/hub/src/config/specialist-ended.ts index b37fe0d3..30f9e275 100644 --- a/apps/hub/src/config/specialist-ended.ts +++ b/apps/hub/src/config/specialist-ended.ts @@ -33,6 +33,10 @@ const PERSON_LINES: Record> = { }, }; +const NOTICE_LINES = new Set( + Object.values(PERSON_LINES).flatMap((lines) => Object.values(lines)), +); + const TASKS: Record string> = { writer: (kind) => `The writer writing the ${kind ?? "assigned"} document`, reviewer: (kind) => @@ -59,5 +63,9 @@ export function specialistEndedMessage(ended: { export function personTurnText(text: string): string { if (!text.startsWith(SPECIALIST_ENDED_LEAD)) return text; const end = text.indexOf(COORDINATOR_ONLY); - return end === -1 ? text : text.slice(0, end); + if (end === -1) return text; + // A person may type the same lead; only the hub's own notice lines are cut. + return NOTICE_LINES.has(text.slice(SPECIALIST_ENDED_LEAD.length, end)) + ? text.slice(0, end) + : text; } diff --git a/apps/hub/src/mail/conversation-threads.ts b/apps/hub/src/mail/conversation-threads.ts index c3850d94..66afc5ac 100644 --- a/apps/hub/src/mail/conversation-threads.ts +++ b/apps/hub/src/mail/conversation-threads.ts @@ -8,6 +8,7 @@ import { listMail } from "@intx/storage-isogit/node"; import { isRunAddress, parseRunAddress } from "@intx/types/agent-address"; import { SPECIALIST_RESULT_TYPE } from "@corbits-builder/schemas/native-specialist"; import { + clippedTurnText, maxConversationTurns, type ConversationTextTurn, type ConversationThread, @@ -20,9 +21,6 @@ import { requestErrorMetadata } from "../request-error.ts"; const log = getLogger(["corbits-builder", "hub", "conversation-threads"]); -// ConversationEvents rejects a longer turn than this. -export const maxTurnText = 20000; - type ThreadMail = { ordinal: number; messageId: string; @@ -104,10 +102,7 @@ export async function readCoordinatorMailThreads(input: { ); if (from === undefined || part === undefined || !Number.isFinite(at)) continue; - const text = new TextDecoder() - .decode(part.content) - .trim() - .slice(0, maxTurnText); + const text = clippedTurnText(new TextDecoder().decode(part.content).trim()); const thread = mail.get(entry.threadId) ?? { participants: new Set(), messages: [], diff --git a/apps/hub/src/mail/conversation-turns.test.ts b/apps/hub/src/mail/conversation-turns.test.ts index c8d9ad2a..b82b2029 100644 --- a/apps/hub/src/mail/conversation-turns.test.ts +++ b/apps/hub/src/mail/conversation-turns.test.ts @@ -3,6 +3,8 @@ import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { $ } from "bun"; +import { resumeMessage } from "../config/resume.ts"; +import { specialistEndedMessage } from "../config/specialist-ended.ts"; import { readDurableConversationTurns } from "./conversation-turns.ts"; const person = "b65b0e2f4c4a444db11b7321dc413964@sunny-harbor-panda.localhost"; @@ -14,7 +16,7 @@ afterEach(async () => { ); }); -// A run repo whose committed coordinator mirror holds these inbound messages, as Interchange stores them. +// A run repo whose committed coordinator mirror holds these inbound messages, with the CRLF line breaks mail delivers. async function runRepo(inbound: { from: string; text: string }[]) { const dir = await mkdtemp(join(tmpdir(), "sb-conversation-turns-")); dirs.push(dir); @@ -22,7 +24,12 @@ async function runRepo(inbound: { from: string; text: string }[]) { await mkdir(wal, { recursive: true }); const turns = inbound.map(({ from, text }, at) => ({ role: "user", - content: [{ type: "text", text: `[From: ${from}]\n\n${text}` }], + content: [ + { + type: "text", + text: `[From: ${from}]\n\n${text.replaceAll("\n", "\r\n")}`, + }, + ], timestamp: at + 1, })); const metadata = { @@ -97,3 +104,70 @@ test("resume turns read each approval decision as a plain line", async () => { { role: "person", text: "Approved a request." }, ]); }); + +test("a person's words read back with the line breaks they typed", async () => { + const typed = + 'I dont get this one bit\n\nOn "Constraints" v1:\n\n> Use $5 per month'; + const dir = await runRepo([ + { from: person, text: typed }, + { from: person, text: "Sure" }, + ]); + expect(await personTexts(dir, "note")).toEqual([ + { role: "person", text: typed }, + { role: "person", text: "Sure" }, + ]); +}); + +test("a specialist-ended notice reads back without its coordinator-only part", async () => { + const notice = specialistEndedMessage({ + role: "writer", + status: "failed", + reason: { message: "provider refused the request" }, + }); + const dir = await runRepo([{ from: person, text: notice }]); + expect(await personTexts(dir, "note")).toEqual([ + { + role: "person", + text: "Status update: The writer stopped before finishing.", + }, + ]); +}); + +test("a resumed message reads back without its held context", async () => { + const context = + "Background: this project's conversation continues from an earlier session. This is where it stood, not a new request.\n\nEarlier conversation, oldest first:\n\nPerson: hi"; + const dir = await runRepo([ + { from: person, text: resumeMessage(context, "Line one\n\nLine two") }, + ]); + expect(await personTexts(dir, "note")).toEqual([ + { role: "person", text: "Line one\n\nLine two" }, + ]); +}); + +test("a person's own message with the notice lead reads back whole", async () => { + const typed = "Status update: done\n\nmore details"; + const dir = await runRepo([{ from: person, text: typed }]); + expect(await personTexts(dir, "note")).toEqual([ + { role: "person", text: typed }, + ]); +}); + +test("a resumed message with no words leaves no turn", async () => { + const context = + "Background: this project's conversation continues from an earlier session. This is where it stood, not a new request.\n\nEarlier conversation, oldest first:\n\nPerson: hi"; + const dir = await runRepo([ + { from: person, text: resumeMessage(context, " ") }, + ]); + expect(await personTexts(dir, "note")).toEqual([]); +}); + +test("a resumed message is read after the separator, not at a quoted line that starts like it", async () => { + const context = + "Background: this project's conversation continues from an earlier session. This is where it stood, not a new request.\n\nEarlier conversation, oldest first:\n\nPerson: hi\n\nThe person's new message: quoted earlier"; + const dir = await runRepo([ + { from: person, text: resumeMessage(context, "Real words") }, + ]); + expect(await personTexts(dir, "note")).toEqual([ + { role: "person", text: "Real words" }, + ]); +}); diff --git a/apps/hub/src/mail/conversation-turns.ts b/apps/hub/src/mail/conversation-turns.ts index f0ef44dc..5093447e 100644 --- a/apps/hub/src/mail/conversation-turns.ts +++ b/apps/hub/src/mail/conversation-turns.ts @@ -9,6 +9,8 @@ import { isRunAddress } from "@intx/types/agent-address"; import { ApprovalDecision } from "@intx/types/signals"; import { type } from "arktype"; import { + clippedTurnText, + deliveredText, maxConversationTurns, type ConversationTextTurn, type ConversationThread, @@ -19,10 +21,7 @@ import { withoutResumeContext, } from "../config/resume.ts"; import { personTurnText } from "../config/specialist-ended.ts"; -import { - maxTurnText, - readCoordinatorMailThreads, -} from "./conversation-threads.ts"; +import { readCoordinatorMailThreads } from "./conversation-threads.ts"; import { withCommittedAgentState } from "./committed-agent-state.ts"; import type { CoordinatorRepoDirSource } from "./coordinator-question.ts"; import { getLogger } from "@intx/log"; @@ -53,9 +52,6 @@ const textFromContent = (content: unknown): string | undefined => { return text.trim() === "" ? undefined : text; }; -const clipTurn = (text: string) => - text.length > maxTurnText ? text.slice(0, maxTurnText) : text; - // The durable copy of a person's mail keeps the routing line the runtime // wrapped around their words. That line is not part of what they wrote. const inboundEnvelope = @@ -103,7 +99,7 @@ function mapMirrorConversationTurns( const raw = textFromContent(turn["content"]); if (raw === undefined) continue; if (role === "assistant") { - mapped.push({ role, text: clipTurn(raw), at }); + mapped.push({ role, text: clippedTurnText(raw), at }); continue; } const from = inboundEnvelope.exec(raw)?.[1]; @@ -114,12 +110,18 @@ function mapMirrorConversationTurns( ) { const text = approvalText(raw.replace(inboundEnvelope, ""), decisions); if (text !== undefined && text.trim() !== "") - mapped.push({ role: "person", text: clipTurn(text), at }); + mapped.push({ role: "person", text: clippedTurnText(text), at }); continue; } - const text = withoutResumeContext(raw.replace(inboundEnvelope, "")); + const text = withoutResumeContext( + deliveredText(raw.replace(inboundEnvelope, "")), + ); if (text === undefined || text.trim() === "") continue; - mapped.push({ role: "person", text: clipTurn(personTurnText(text)), at }); + mapped.push({ + role: "person", + text: clippedTurnText(personTurnText(text)), + at, + }); } mapped.sort((left, right) => left.at - right.at); return mapped.length > maxConversationTurns diff --git a/packages/schemas/src/conversation.test.ts b/packages/schemas/src/conversation.test.ts new file mode 100644 index 00000000..8ab514d9 --- /dev/null +++ b/packages/schemas/src/conversation.test.ts @@ -0,0 +1,12 @@ +import { expect, test } from "bun:test"; +import { clippedTurnText, deliveredText, maxTurnText } from "./conversation.ts"; + +test("a 200,000-space run inside a line stays while the line's trailing blanks go", () => { + const text = `a${" ".repeat(200_000)}b`; + expect(deliveredText(`${text}\t \r\n`)).toBe(text); +}); + +test("a turn clipped at its limit never keeps half of an emoji", () => { + const text = `${"a".repeat(maxTurnText - 1)}😀 more`; + expect(clippedTurnText(text)).toBe("a".repeat(maxTurnText - 1)); +}); diff --git a/packages/schemas/src/conversation.ts b/packages/schemas/src/conversation.ts index 1f1de48c..a8be5e30 100644 --- a/packages/schemas/src/conversation.ts +++ b/packages/schemas/src/conversation.ts @@ -8,13 +8,41 @@ export type ConversationMessage = Readonly<{ Readonly<{ mimeType: string; data: string; name?: string }> >; }>; + +/** + * A person's words as mail delivers them and the conversation reads them back: LF line breaks, no + * spaces or tabs at the end of a line, trimmed. + */ +export const deliveredText = (text: string) => + text + .replace(/\r\n?/g, "\n") + .split("\n") + .map(withoutTrailingBlanks) + .join("\n") + .trim(); + +// A scan from the end, since a regex for trailing blanks backtracks quadratically on a long run. +function withoutTrailingBlanks(line: string) { + let end = line.length; + while (end > 0 && (line[end - 1] === " " || line[end - 1] === "\t")) end -= 1; + return line.slice(0, end); +} + +export const maxTurnText = 20000; + +export function clippedTurnText(text: string) { + const clipped = text.slice(0, maxTurnText); + // A cut between the halves of a surrogate pair would leave half a character. + return /[\uD800-\uDBFF]$/.test(clipped) ? clipped.slice(0, -1) : clipped; +} + export const ConversationReceipt = type({ runId: "string", address: "string", messageId: "string", }); export type ConversationReceipt = typeof ConversationReceipt.infer; -const turnText = type("string > 0").atMostLength(20000); +const turnText = type("string > 0").atMostLength(maxTurnText); // A participant turn is mail from a workflow run other than the coordinator; // `sender` names its role or run. const conversationTextTurn = type({ From b81706b9e6ce8442c6e79bd04df7d65c0dc6c594 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 3 Nov 2026 22:01:18 -0800 Subject: [PATCH 2/2] fix(web): recognise a sent message as mail delivers it Mail delivery drops spaces and tabs at the end of each line, so a message with them never matched the conversation's copy and showed twice. The page compares each message it sent, including one with files and the opening idea, with the conversation by the text mail delivers, clipped to a turn's length. Only a coordinator turn later than the latest one the page held when it sent the message can be its copy, so a repeated message shows as new and keeps its own files. --- apps/web/src/pages/workspace/index.tsx | 97 ++++--------- apps/web/src/pages/workspace/live-run.ts | 9 +- .../src/pages/workspace/pending-sends.test.ts | 137 ++++++++++++++++++ apps/web/src/pages/workspace/pending-sends.ts | 76 ++++++++++ .../src/pages/workspace/transcript.test.tsx | 19 +++ test/AGENTS.md | 5 +- test/integration/delivered-text.test.ts | 43 ++++++ 7 files changed, 307 insertions(+), 79 deletions(-) create mode 100644 apps/web/src/pages/workspace/pending-sends.test.ts create mode 100644 apps/web/src/pages/workspace/pending-sends.ts create mode 100644 test/integration/delivered-text.test.ts diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index 2aa2aad8..63372a47 100644 --- a/apps/web/src/pages/workspace/index.tsx +++ b/apps/web/src/pages/workspace/index.tsx @@ -54,7 +54,7 @@ import "./persistent-workspace.css"; import { ProductApiError, type ProjectArtifact } from "@corbits-builder/client"; import type { InFlightRun } from "@corbits-builder/schemas/artifact"; import { toast } from "sonner"; -import { sendMessage, type ConversationTurn } from "../../api/conversation.ts"; +import { sendMessage } from "../../api/conversation.ts"; import { subscribeWorkflowRun } from "../../api/run-stream.ts"; import { acceptedMaterials, @@ -67,15 +67,10 @@ import { restoreProject, startProjectCoordinator, } from "../../api/projects.ts"; -import type { ChatMessage, Wait } from "../../api/types.ts"; +import type { Wait } from "../../api/types.ts"; import { keys } from "../../queries/keys.ts"; import { COMPOSER_BOX_CLASS, CONV_CLASS, PANES_CLASS } from "./pane-classes.ts"; -import { - questionsOpen, - RoomThread, - turnItemId, - type TranscriptTurn, -} from "./transcript.tsx"; +import { questionsOpen, RoomThread, turnItemId } from "./transcript.tsx"; import { cardDocuments } from "./collaboration.ts"; import { useFiledVersions } from "./filed-versions.tsx"; import { threadUpdates } from "./specialist-updates.tsx"; @@ -105,6 +100,12 @@ import { type LiveRun, } from "./live-run.ts"; import { createRoomReloader } from "./room-reload.ts"; +import { + latestTurnAt, + SENDING, + withPendingSends, + type SentMessage, +} from "./pending-sends.ts"; import { decideApproval, openDecisions } from "../../api/decisions.ts"; import type { DeliverableState } from "./deliverable-block.tsx"; import { formatPassageQuote, type PassageQuote } from "./passage-note.tsx"; @@ -177,7 +178,7 @@ export { selectedDeliverable }; function openingDraft( projectId: string, openingMessage: string | undefined, -): ChatMessage[] { +): SentMessage[] { const text = openingMessage?.trim() ?? ""; if (text.length === 0) return []; return [ @@ -186,40 +187,11 @@ function openingDraft( author: "me", body: text, at: new Date(0).toISOString(), + after: Number.NEGATIVE_INFINITY, }, ]; } -// A message's stand-in id until the hub's receipt replaces it. -const SENDING = "sending:"; - -function withPendingSends( - mirror: readonly ConversationTurn[], - sent: readonly ChatMessage[], -): TranscriptTurn[] { - const pending = [...sent]; - const shown: TranscriptTurn[] = []; - for (const turn of mirror) { - const index = pending.findIndex( - (message) => - message.body === turn.text && - (message.author === "agent") === (turn.role === "assistant"), - ); - if (index >= 0) pending.splice(index, 1); - shown.push(turn); - } - for (const message of pending) { - const at = Date.parse(message.at); - shown.push({ - role: message.author === "agent" ? "assistant" : "person", - text: message.body, - at: Number.isNaN(at) ? 0 : at, - ...(message.id.startsWith(SENDING) ? { sending: true } : {}), - }); - } - return shown; -} - // The hand-offs that name the document their run works on, by the role they start. const HANDOFF_ROLE: Record = { deploy_writer: "writer", @@ -389,7 +361,7 @@ export function StageWorkspace({ const [mobilePane, setMobilePane] = useState<"conversation" | "artifacts">( "conversation", ); - const [sent, setSent] = useState(() => + const [sent, setSent] = useState(() => openingDraft(projectId, openingMessage), ); const [value, setValue] = useState(""); @@ -423,8 +395,6 @@ export function StageWorkspace({ ? {} : { maxFiles: limits.maxFiles, maxFileBytes: limits.maxFileBytes }), }); - // The files each sent message carried, by its text, as this session sent them. - const filesByText = useRef(new Map()); const liveRunRef = useRef(emptyLiveRun); const committedAtRef = useRef(null); committedAtRef.current = latestAssistantAt(mirror); @@ -490,7 +460,6 @@ export function StageWorkspace({ runKinds.current.clear(); usedCalls.current.clear(); runCards.current.clear(); - filesByText.current.clear(); }, [projectId, openingMessage]); // The durable mirror is the committed transcript. The stream fills the gap @@ -575,18 +544,7 @@ export function StageWorkspace({ const turns = useMemo( () => withPendingSends(mirror, sent).map((turn) => { - const attached = - turn.role === "person" - ? filesByText.current.get(turn.text) - : undefined; - if (attached !== undefined) - return { - ...turn, - parts: - turn.text.length === 0 - ? attached - : [...attached, { kind: "text" as const, text: turn.text }], - }; + if (turn.parts !== undefined) return turn; const steps = turn.role === "assistant" ? stepsByAt.current.get(turn.at) @@ -701,21 +659,21 @@ export function StageWorkspace({ setSendFault(null); setAwaitingSince(Date.now()); // The person's words show the moment they send; the receipt replaces this stand-in. - const pending: ChatMessage = { + const pending: SentMessage = { id: `${SENDING}${String(Date.now())}`, author: "me", body: text, at: new Date().toISOString(), + after: latestTurnAt(mirror), }; - setSent((previous) => [ - ...previous.filter((item) => item.author !== "me" || item.body !== text), - pending, - ]); + setSent((previous) => [...previous, pending]); if (clearComposer) setValue(""); try { const { message, receipt } = await sendMessage(projectId, text); setSent((previous) => - previous.map((item) => (item === pending ? message : item)), + previous.map((item) => + item === pending ? { ...pending, ...message } : item, + ), ); void receipt; reloadRef.current(); @@ -768,19 +726,17 @@ export function StageWorkspace({ setSendFault(null); setAwaitingSince(Date.now()); setAttachFault(null); - filesByText.current.set( - text, - attached.map((item) => ({ - kind: "file" as const, - name: item.name, - mediaType: item.mediaType, - })), - ); - const pending: ChatMessage = { + const pending: SentMessage = { id: `${SENDING}${String(Date.now())}`, author: "me", body: text, at: new Date().toISOString(), + after: latestTurnAt(mirror), + files: attached.map((item) => ({ + kind: "file" as const, + name: item.name, + mediaType: item.mediaType, + })), }; setSent((previous) => [...previous, pending]); setValue(""); @@ -802,7 +758,6 @@ export function StageWorkspace({ } catch (cause) { setAwaitingSince(null); setSent((previous) => previous.filter((item) => item !== pending)); - filesByText.current.delete(text); setValue((current) => (current === "" ? text : current)); setFiles(attached); setAttachFault(await attachFaultText(cause)); diff --git a/apps/web/src/pages/workspace/live-run.ts b/apps/web/src/pages/workspace/live-run.ts index 2189c51c..c24ef242 100644 --- a/apps/web/src/pages/workspace/live-run.ts +++ b/apps/web/src/pages/workspace/live-run.ts @@ -12,6 +12,7 @@ import type { ConversationTurn } from "../../api/conversation.ts"; import type { Wait } from "../../api/types.ts"; import type { RunStreamEvent } from "../../api/run-stream.ts"; import type { ConversationPart } from "@corbits/react-ui"; +import { clippedTurnText } from "@corbits-builder/schemas/conversation"; import { emptyLiveTurn, foldLiveTurn, @@ -74,9 +75,6 @@ const reloadAfter = new Set([ "reactor.error", ]); -// The mirror caps each turn's text at this length. -const MIRROR_TEXT_LIMIT = 20_000; - function hasContent(segment: LiveTurnState): boolean { return segment.text.length > 0 || segment.parts.length > 0; } @@ -162,9 +160,8 @@ export function foldLiveRun( } function sameText(committed: string, live: string): boolean { - const a = committed.trim(); - const b = live.trim(); - return a === b || (a.length >= MIRROR_TEXT_LIMIT && b.startsWith(a)); + const text = committed.trim(); + return text === live.trim() || text === clippedTurnText(live).trim(); } function committedTurn( diff --git a/apps/web/src/pages/workspace/pending-sends.test.ts b/apps/web/src/pages/workspace/pending-sends.test.ts new file mode 100644 index 00000000..b339c5eb --- /dev/null +++ b/apps/web/src/pages/workspace/pending-sends.test.ts @@ -0,0 +1,137 @@ +import { expect, test } from "bun:test"; +import { maxTurnText } from "@corbits-builder/schemas/conversation"; +import { + latestTurnAt, + SENDING, + withPendingSends, + type SentMessage, +} from "./pending-sends.ts"; + +const typed = "First line \r\nsecond line\t\n\nthird"; +const asDelivered = "First line\nsecond line\n\nthird"; +const notes = { + kind: "file" as const, + name: "notes.txt", + mediaType: "text/plain", +}; + +const sent = ( + id: string, + body = typed, + more: Partial = {}, +): SentMessage => ({ + id, + author: "me", + body, + at: new Date(5).toISOString(), + after: 1, + ...more, +}); + +test("a sent message shows once when the conversation holds it as mail delivered it", () => { + const turns = withPendingSends( + [{ role: "person", text: asDelivered, at: 9 }], + [sent("message-1")], + ); + expect(turns).toEqual([{ role: "person", text: asDelivered, at: 9 }]); +}); + +test("a sent message longer than a turn shows once when the conversation holds it clipped", () => { + const long = "a".repeat(maxTurnText + 10); + const clipped = long.slice(0, maxTurnText); + expect( + withPendingSends( + [{ role: "person", text: clipped, at: 9 }], + [sent("message-1", long)], + ), + ).toEqual([{ role: "person", text: clipped, at: 9 }]); +}); + +test("a sent message shows its files while sending and once the conversation holds it", () => { + const withNotes = sent(`${SENDING}1`, typed, { files: [notes] }); + expect(withPendingSends([], [withNotes])).toEqual([ + { + role: "person", + text: typed, + at: 5, + sending: true, + parts: [notes, { kind: "text", text: typed }], + }, + ]); + expect( + withPendingSends( + [{ role: "person", text: asDelivered, at: 9 }], + [withNotes], + ), + ).toEqual([ + { + role: "person", + text: asDelivered, + at: 9, + parts: [notes, { kind: "text", text: asDelivered }], + }, + ]); +}); + +test("a repeated message shows as new, with its files, beside the earlier turn with its words", () => { + const earlier = { role: "person" as const, text: "yes", at: 3 }; + const reply = { role: "assistant" as const, text: "Noted.", at: 4 }; + const repeated = sent(`${SENDING}2`, "yes", { after: 4, files: [notes] }); + expect(withPendingSends([earlier, reply], [repeated])).toEqual([ + earlier, + reply, + { + role: "person", + text: "yes", + at: 5, + sending: true, + parts: [notes, { kind: "text", text: "yes" }], + }, + ]); + expect( + withPendingSends( + [earlier, reply, { role: "person", text: "yes", at: 6 }], + [sent("message-1", "yes"), repeated], + ), + ).toEqual([ + earlier, + reply, + { + role: "person", + text: "yes", + at: 6, + parts: [notes, { kind: "text", text: "yes" }], + }, + ]); +}); + +test("a participant turn with the same words does not take the person's sent message", () => { + expect( + withPendingSends( + [ + { role: "participant", sender: "writer", text: asDelivered, at: 7 }, + { role: "person", text: asDelivered, at: 9 }, + ], + [sent("message-1", typed, { files: [notes] })], + ), + ).toEqual([ + { role: "participant", sender: "writer", text: asDelivered, at: 7 }, + { + role: "person", + text: asDelivered, + at: 9, + parts: [notes, { kind: "text", text: asDelivered }], + }, + ]); +}); + +test("a message is sent after the latest person or assistant turn, ignoring participants, and before every turn of an empty conversation", () => { + expect( + latestTurnAt([ + { role: "person", text: "hi", at: 3 }, + { role: "assistant", text: "Hello.", at: 4 }, + { role: "participant", sender: "writer", text: "Done.", at: 8 }, + ]), + ).toBe(4); + expect(latestTurnAt([])).toBe(Number.NEGATIVE_INFINITY); +}); diff --git a/apps/web/src/pages/workspace/pending-sends.ts b/apps/web/src/pages/workspace/pending-sends.ts new file mode 100644 index 00000000..d7429b83 --- /dev/null +++ b/apps/web/src/pages/workspace/pending-sends.ts @@ -0,0 +1,76 @@ +import type { ConversationPart } from "@corbits/react-ui"; +import { + clippedTurnText, + deliveredText, +} from "@corbits-builder/schemas/conversation"; +import type { ConversationTurn } from "../../api/conversation.ts"; +import type { ChatMessage } from "../../api/types.ts"; +import type { TranscriptTurn } from "./transcript.tsx"; + +// A message's stand-in id until the hub's receipt replaces it. +export const SENDING = "sending:"; + +/** + * A message this session sent, with the files it carried. `after` is the time of the coordinator's + * latest turn the page held when it was sent, so the conversation's copy of the message is a later + * turn. Both times come from the coordinator's own record: the browser's clock never decides a + * match. + */ +export type SentMessage = ChatMessage & + Readonly<{ after: number; files?: readonly ConversationPart[] }>; + +/** + * The time of the coordinator's latest turn; with none, every turn comes after it. Participant + * turns are left out: their times come from another run's mail. + */ +export const latestTurnAt = (mirror: readonly ConversationTurn[]) => + Math.max( + ...mirror.flatMap((turn) => (turn.role === "participant" ? [] : [turn.at])), + ); + +function withFiles( + turn: TranscriptTurn, + files: readonly ConversationPart[] | undefined, +): TranscriptTurn { + if (files === undefined) return turn; + if (turn.text.length === 0) return { ...turn, parts: files }; + return { ...turn, parts: [...files, { kind: "text", text: turn.text }] }; +} + +/** + * The conversation's turns, then this session's sent messages it does not hold yet. The + * conversation holds a message as mail delivered it, clipped to a turn's length, so a sent + * message's copy is a later turn with that text. + */ +export function withPendingSends( + mirror: readonly ConversationTurn[], + sent: readonly SentMessage[], +): TranscriptTurn[] { + const pending = [...sent]; + const shown: TranscriptTurn[] = []; + for (const turn of mirror) { + const index = pending.findIndex( + (message) => + turn.at > message.after && + clippedTurnText(deliveredText(message.body)) === turn.text && + turn.role === (message.author === "agent" ? "assistant" : "person"), + ); + const message = index >= 0 ? pending.splice(index, 1)[0] : undefined; + shown.push( + message === undefined || turn.role !== "person" + ? turn + : withFiles(turn, message.files), + ); + } + for (const message of pending) { + const at = Date.parse(message.at); + const turn: TranscriptTurn = { + role: message.author === "agent" ? "assistant" : "person", + text: message.body, + at: Number.isNaN(at) ? 0 : at, + ...(message.id.startsWith(SENDING) ? { sending: true } : {}), + }; + shown.push(turn.role === "person" ? withFiles(turn, message.files) : turn); + } + return shown; +} diff --git a/apps/web/src/pages/workspace/transcript.test.tsx b/apps/web/src/pages/workspace/transcript.test.tsx index 217837d7..95122f98 100644 --- a/apps/web/src/pages/workspace/transcript.test.tsx +++ b/apps/web/src/pages/workspace/transcript.test.tsx @@ -1,6 +1,10 @@ import { describe, expect, test } from "bun:test"; import { isValidElement, type ReactNode } from "react"; import { renderToStaticMarkup } from "react-dom/server"; +import { + clippedTurnText, + maxTurnText, +} from "@corbits-builder/schemas/conversation"; import { wording } from "../../config.ts"; import { DocumentDecision } from "./document-decision.tsx"; import { @@ -9,6 +13,7 @@ import { foldLiveRun, liveTurnStreaming, parkedOnApproval, + pendingLiveTurn, waitingOnPerson, } from "./live-run.ts"; import { RoomThread } from "./transcript.tsx"; @@ -190,6 +195,20 @@ describe("conversation transcript", () => { expect(awaitingCommit(done, [])).toBe(true); }); + test("a reply over a turn's length is committed by its clipped copy, even when the clip drops half an emoji", () => { + const reply = `${"a".repeat(maxTurnText - 1)}😀 and more`; + const run = [ + { type: "inference.start" }, + { type: "inference.text.delta", data: { partial: { text: reply } } }, + { type: "inference.done" }, + ].reduce((live, event) => foldLiveRun(live, event, null).run, emptyLiveRun); + expect( + pendingLiveTurn(run, [ + { role: "assistant", text: clippedTurnText(reply), at: 5 }, + ]), + ).toBeNull(); + }); + test("a turn held on the person's approval is no longer being written", () => { const held = [ { type: "message.run.started" }, diff --git a/test/AGENTS.md b/test/AGENTS.md index 44300952..2e9fdbcf 100644 --- a/test/AGENTS.md +++ b/test/AGENTS.md @@ -4,8 +4,9 @@ Unit tests sit next to the code they test. This folder holds cross-package and end-to-end tests: - **`test/integration/`:** architecture and boundary checks (package imports, - vendored Interchange revision, workspace exports). Runs in `bun run test` - with no database. + vendored Interchange revision, workspace exports) and Builder's contracts + with vendored Interchange, such as the text its mail delivers. Runs in + `bun run test` with no database. - **`test/e2e/`:** end-to-end tests against an isolated Postgres, the real hub and real sidecars (`custom-provider.test.ts`, `oauth-model-login.test.ts`, `specialist-end-notices.test.ts` and `pipeline.test.ts`). diff --git a/test/integration/delivered-text.test.ts b/test/integration/delivered-text.test.ts new file mode 100644 index 00000000..2a085b37 --- /dev/null +++ b/test/integration/delivered-text.test.ts @@ -0,0 +1,43 @@ +import { expect, test } from "bun:test"; +import { + assembleSignedContent, + decodePartBytes, + extractPartByPath, + parseMimePart, +} from "@intx/mime"; +import { deliveredText } from "@corbits-builder/schemas/conversation"; + +// Mail must deliver a typed message as `deliveredText` of it, or the page will not recognise its +// copy. +const delivered = (text: string, withFile: boolean) => { + const file = { + name: "notes.txt", + contentType: "text/plain", + data: new TextEncoder().encode("notes"), + }; + const signed = assembleSignedContent({ + kind: "conversation", + text, + ...(withFile ? { attachments: [file] } : {}), + }); + const part = parseMimePart(extractPartByPath(signed, "1")); + return new TextDecoder() + .decode(decodePartBytes(part.body, part.headers)) + .replace(/\r\n?/g, "\n"); +}; + +test("mail delivers a typed message as its deliveredText, with or without a file", () => { + const typed = [ + "First line \nkeep two spaces\t\n\nthird \t ", + "old \rline \r\nbreaks\r\rend", + "no-break\u00a0\nideographic\u3000\nline\u2028separator\u0085next", + "code:\n indented line\n\tTabbed", + "From here on\n.\nlone dot", + "café 😀 \nnaïve", + `${"x".repeat(1200)} \nb`, + ]; + for (const text of typed.map((raw) => raw.trim())) { + expect(delivered(text, false)).toBe(deliveredText(text)); + expect(delivered(text, true)).toBe(deliveredText(text)); + } +});