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/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/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({ 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)); + } +});