From 14dfa417674358255eb68cc3f312abf6f1e4ec47 Mon Sep 17 00:00:00 2001 From: Renfei Zhou Date: Sat, 14 Nov 2026 02:28:45 +0000 Subject: [PATCH] feat(fork): ack notes, Edited mark and a divider in display modes In Messages only, Messages + Dashboard and Chat + Replies, an agent could send exactly one message per typed message, a rewrite left no trace, and an edited reply did not count as new. - send_to_user takes an optional ack (at most 200 bytes): per user message the agent may send one ack, then one answer. A send after the answer is refused, and replace_send_to_user always leaves an answer, so it can turn the ack into the answer. One stage table per user message (replyStage.ts) decides sends and replaces for both the Messages path and each sub-thread, and a replace waits while the answer it would rewrite is still being delivered. - The history fold records which rewrite changed a message. The web shows a small "Edited" mark under it, and in Chat + Replies an edited reply takes the edit's time, so it shows as new and moves the sub-thread's latest activity. - A thin rule parts two agent messages drawn in a row in a view. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- .../fork/toolParameterDescriptions.test.ts | 10 +- .../fork/userMessageTools/SendToUser.test.ts | 139 +++++++++++- .../src/fork/userMessageTools/SendToUser.ts | 205 ++++++++++++------ .../SendToUserSubThreads.test.ts | 82 ++++++- .../src/fork/userMessageTools/replyStage.ts | 141 ++++++++++++ .../src/provider/agentContextBudget.json | 2 +- .../components/chat/MessagesTimeline.logic.ts | 4 +- .../components/chat/TimelineAssistantRows.tsx | 4 + .../fork/displayMode/AgentMessageDivider.tsx | 4 + apps/web/src/fork/displayMode/EditedMark.tsx | 27 +++ .../fork/displayMode/displayModeTimeline.ts | 84 +++++-- .../fork/subThreads/subThreadTimeline.test.ts | 113 +++++++++- .../src/fork/subThreads/subThreadsState.ts | 4 +- apps/web/src/types.ts | 1 + .../fork/displayModes/displayModes.test.ts | 52 ++++- .../src/fork/displayModes/displayModes.ts | 32 ++- 17 files changed, 789 insertions(+), 117 deletions(-) create mode 100644 apps/server/src/fork/userMessageTools/replyStage.ts create mode 100644 apps/web/src/fork/displayMode/AgentMessageDivider.tsx create mode 100644 apps/web/src/fork/displayMode/EditedMark.tsx diff --git a/README.md b/README.md index d072a6c345d3e..8af4d70e9fc74 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ This is a fork of T3 Code that integrates a few additional features, including: - **Display mode per thread:** Right-click a thread and set Display mode to "Messages only" to hide the agent's work and see just your messages and the replies the agent sends to you with its `send_to_user` tool. The mode applies in the web and desktop apps, not in the mobile app. - "Messages + Dashboard" adds a Dashboard tab to the right panel (+ → Dashboard): six status slots that the agent writes with its `dashboard_set` tool and reads with `dashboard_get`. Each slot shows how long ago it changed, and new text stays highlighted until you click the slot. Select text in a slot to quote it with Cite, as you would in a message. - "Chat + Replies" keeps the main list short: it shows only first messages, and the replies to a message show in a Replies tab in the right panel (+ → Replies, or the reply line under a message). Point at a message and choose Reply to answer under it; the tab has its own reply box. Under each first message you see how many replies it has, how many are new, and a few generated words on what they are about. The agent answers a message with `send_to_user` and that message's id, so each answer lands under the message it answers. - - In all three modes the agent sends one reply per message you type. To say more, it rewrites that reply in place with its `replace_send_to_user` tool. + - In all three modes the agent sends one reply per message you type, and before a long task it may first send a short "on it" note; a thin line parts two agent messages in a row. To say more, it rewrites its reply in place with its `replace_send_to_user` tool. A rewritten reply shows a small "Edited" mark, and in Chat + Replies it counts as new. - Copy buttons on pages in the preview browser now put the copied text into your own clipboard. - **Two preview browser modes:** Settings → Browser → Preview browser picks the browser the server runs for preview tabs. **T3 native** (fast) is the default. It runs chrome-headless-shell, Chrome's small build that runs without a window. **Chrome** (slow) runs full Chrome, for pages that don't work in T3 native. On platforms where T3 has no Chrome for Testing build (Google's Chrome build for test automation), both choices run T3 native. Switching closes the open preview tabs. - **Compact when idle for Claude threads:** a choice in Settings → Providers → Advanced → Compact idle threads. Besides compacting before the prompt cache expires, it compacts a large idle Claude thread (default 400K tokens or more) a set time (default 15 minutes) after the last message or answer from a user or a bot, so the compaction runs while no one waits for a reply. The token counts and the wait are set right under the choice. diff --git a/apps/server/src/fork/toolParameterDescriptions.test.ts b/apps/server/src/fork/toolParameterDescriptions.test.ts index c407b065e751e..1118da3bf5dbb 100644 --- a/apps/server/src/fork/toolParameterDescriptions.test.ts +++ b/apps/server/src/fork/toolParameterDescriptions.test.ts @@ -10,8 +10,16 @@ const parameterDescription = (tool: Tool.Any, parameter: string): unknown => // Agents see a parameter's rules only through its description in the tool's JSON schema. describe("fork tool parameter descriptions", () => { - it("send_to_user describes text", () => { + it("send_to_user describes text and ack", () => { expect(parameterDescription(SendToUserToolkit.tools.send_to_user, "text")).toMatch(/bytes/); + // A nullable parameter keeps its description on the boolean branch. + const schema = Tool.getJsonSchema(SendToUserToolkit.tools.send_to_user) as { + properties: { ack: { anyOf: ReadonlyArray } }; + }; + expect(schema.properties.ack.anyOf).toContainEqual({ + type: "boolean", + description: expect.stringContaining("at most 200 bytes"), + }); }); it("replace_send_to_user describes user_message, old_text and new_text", () => { diff --git a/apps/server/src/fork/userMessageTools/SendToUser.test.ts b/apps/server/src/fork/userMessageTools/SendToUser.test.ts index a6e314af51982..dcab81964c4ad 100644 --- a/apps/server/src/fork/userMessageTools/SendToUser.test.ts +++ b/apps/server/src/fork/userMessageTools/SendToUser.test.ts @@ -94,6 +94,8 @@ const sent = (id: string, text: string, status: OrchestrationV2TurnItem["status" toolName: "mcp__t3-code__send_to_user", input: { text }, }) satisfies OrchestrationV2TurnItem; +const acked = (id: string, text: string, status: OrchestrationV2TurnItem["status"] = "completed") => + ({ ...sent(id, text, status), input: { text, ack: true } }) satisfies OrchestrationV2TurnItem; const rewritten = (id: string, text: string) => ({ ...base(id), @@ -158,6 +160,20 @@ const SHOWN = { ok: true, message: "Shown to the user. If you are done, end your turn with no text.", }; +const ACKED = { + ok: true, + message: "Shown to the user as an ack. Send your answer when it is ready.", +}; +const SECOND_ANSWER = + "Refused: you already answered the user's last message. To change your answer, call replace_send_to_user."; +const SECOND_ACK = + "Refused: you already sent an ack for the user's last message. Send your answer without ack, or make the ack your answer with replace_send_to_user."; +const LATE_ACK = + "Refused: you already answered the user's last message, and an ack goes only before the answer. To change your answer, call replace_send_to_user."; +const DELIVERING = + "Refused: your message for the user's last message is still being delivered. Call replace_send_to_user again."; +/** The first line of a refusal: its reason. */ +const reasonOf = (result: McpSchema.CallToolResult) => refusalText(result).split("\n")[0]; it.effect("shows the message to the user without needing any capability", () => Effect.gen(function* () { @@ -234,6 +250,22 @@ describe("the 1000-byte limit", () => { }).pipe(Effect.provide(layer)), ); + it.effect("shows an ack of 200 bytes and refuses one byte more, in every display mode", () => + Effect.gen(function* () { + history = [typed("u1", "Fix the build.")]; + const ack = (text: string, threadId: ThreadId) => + call("send_to_user", { text, ack: true }, { threadId }); + for (const threadId of [normalThread, messagesThread]) { + expect(refusalText(yield* ack("a".repeat(201), threadId))).toBe( + "an ack is at most 200 bytes. Say only that you got the message and are on it.", + ); + expect((yield* ack("a".repeat(200), threadId)).isError).not.toBe(true); + } + const answer = yield* call("send_to_user", { text: "a".repeat(1_000) }); + expect(answer.structuredContent).toEqual(SHOWN); + }).pipe(Effect.provide(layer)), + ); + it.effect("gives the same answer in every display mode", () => Effect.gen(function* () { history = [typed("u1", "Fix the build.")]; @@ -328,7 +360,7 @@ describe("one send per typed message", () => { const error = refusalText(yield* call("send_to_user", { text: "Done." })); expect(error).toBe( [ - "Refused: you already sent the user a message after their last one, and this thread allows one per message of theirs. To change it, call replace_send_to_user.", + SECOND_ANSWER, "Latest messages between you and the user:", "[you] On it.", "[user] Any news?", @@ -340,6 +372,23 @@ describe("one send per typed message", () => { }).pipe(Effect.provide(layer)), ); + it.effect("takes one ack before the answer, and no send after the answer", () => + Effect.gen(function* () { + const named = (text: string, ack = false) => + call("send_to_user", { text, ack }, { toolUseId: "toolu_own" }); + history = [typed("u1", "Fix the build.")]; + expect((yield* named("On it.", true)).structuredContent).toEqual(ACKED); + history = [typed("u1", "Fix the build."), acked("a1", "On it.")]; + expect(reasonOf(yield* named("Still on it.", true))).toBe(SECOND_ACK); + expect((yield* named("Done.")).structuredContent).toEqual(SHOWN); + for (const answered of [sent("s1", "Done."), rewritten("w1", "Done.")]) { + history = [typed("u1", "Fix the build."), acked("a1", "On it."), answered]; + expect(reasonOf(yield* named("Also: types pass."))).toBe(SECOND_ANSWER); + expect(reasonOf(yield* named("On it.", true))).toBe(LATE_ACK); + } + }).pipe(Effect.provide(layer)), + ); + it.effect("limits Messages + Dashboard too, and not Normal", () => Effect.gen(function* () { history = [typed("u1", "Fix the build."), sent("s1", "On it.")]; @@ -410,6 +459,15 @@ describe("one send per typed message", () => { history = [typed("u1", "Fix the build."), sent("s1", "Done."), typed("u2", "Thanks.")]; const afterTyping = yield* call("send_to_user", { text: "You're welcome." }); expect(afterTyping.structuredContent).toEqual(SHOWN); + history = [typed("u3", "Ship it.")]; + const ack = yield* call("send_to_user", { text: "On it.", ack: true }); + expect(ack.structuredContent).toEqual(ACKED); + expect(reasonOf(yield* call("send_to_user", { text: "On it.", ack: true }))).toBe( + SECOND_ACK, + ); + const answer = yield* call("send_to_user", { text: "Shipped." }); + expect(answer.structuredContent).toEqual(SHOWN); + expect(reasonOf(yield* call("send_to_user", { text: "Tagged too." }))).toBe(SECOND_ANSWER); }).pipe(Effect.provide(layer)), ); @@ -492,16 +550,23 @@ describe("replace_send_to_user", () => { }).pipe(Effect.provide(layer)), ); - it.effect("expects an empty user_message when the user has typed nothing", () => - Effect.gen(function* () { - history = [typed("r", "Continue where you left off.", "server"), sent("s1", "On it.")]; - const error = refusalText(yield* replace("Continue where you left off.", "On it.")); - expect(error.split("\n")[0]).toBe( - "Refused: user_message is 0% like the user's last message; expected an empty string.", - ); - const result = yield* replace("", "On it."); - expect(result.structuredContent).toEqual(SHOWN); - }).pipe(Effect.provide(layer)), + it.effect( + "expects an empty user_message, and refuses an ack, when the user has typed nothing", + () => + Effect.gen(function* () { + history = [typed("r", "Continue where you left off.", "server")]; + expect(reasonOf(yield* call("send_to_user", { text: "On it.", ack: true }))).toBe( + "Refused: an ack answers a message from the user, and they have typed none. Send without ack.", + ); + expect((yield* call("send_to_user", { text: "On it." })).structuredContent).toEqual(SHOWN); + history = [typed("r", "Continue where you left off.", "server"), sent("s1", "On it.")]; + const error = refusalText(yield* replace("Continue where you left off.", "On it.")); + expect(error.split("\n")[0]).toBe( + "Refused: user_message is 0% like the user's last message; expected an empty string.", + ); + const result = yield* replace("", "On it."); + expect(result.structuredContent).toEqual(SHOWN); + }).pipe(Effect.provide(layer)), ); it.effect( @@ -531,6 +596,56 @@ describe("replace_send_to_user", () => { }).pipe(Effect.provide(layer)), ); + it.effect("refuses a replace while the answer after an ack is still being delivered", () => + Effect.gen(function* () { + const replaceAck = (toolUseId?: string) => + call( + "replace_send_to_user", + { user_message: "Fix the build.", old_text: "On it.", new_text: "Done." }, + toolUseId === undefined ? {} : { toolUseId }, + ); + history = [ + typed("u1", "Fix the build."), + acked("a1", "On it."), + sent("s1", "Fixed.", "running"), + ]; + expect(reasonOf(yield* replaceAck("toolu_own"))).toBe(DELIVERING); + history = [typed("u1", "Fix the build."), acked("a1", "On it.")]; + expect((yield* call("send_to_user", { text: "Fixed." })).structuredContent).toEqual(SHOWN); + expect(reasonOf(yield* replaceAck())).toBe(DELIVERING); + history = [typed("u1", "Fix the build."), acked("a1", "On it."), sent("s1", "Fixed.")]; + const stored = yield* call("replace_send_to_user", { + user_message: "Fix the build.", + old_text: "Fixed.", + new_text: "Fixed and pushed.", + }); + expect(stored.structuredContent).toEqual(SHOWN); + }).pipe(Effect.provide(layer)), + ); + + it.effect("makes an ack the answer, and then holds against another answer", () => + Effect.gen(function* () { + history = [typed("u1", "Fix the build."), acked("a1", "On it.")]; + const replaced = yield* replace("Fix the build.", "On it."); + expect(replaced.structuredContent).toEqual(SHOWN); + expect(reasonOf(yield* call("send_to_user", { text: "Done." }))).toBe(SECOND_ANSWER); + }).pipe(Effect.provide(layer)), + ); + + it.effect("holds a named replace of an ack against another answer until it is stored", () => + Effect.gen(function* () { + history = [typed("u1", "Fix the build."), acked("a1", "On it.")]; + const namedReplace = yield* call( + "replace_send_to_user", + { user_message: "Fix the build.", old_text: "On it.", new_text: "Done." }, + { toolUseId: "toolu_replace" }, + ); + expect(namedReplace.structuredContent).toEqual(SHOWN); + const namedSend = yield* call("send_to_user", { text: "Done." }, { toolUseId: "toolu_send" }); + expect(reasonOf(namedSend)).toBe(SECOND_ANSWER); + }).pipe(Effect.provide(layer)), + ); + it.effect("refuses when there is no send since the user's last message", () => Effect.gen(function* () { history = [sent("s1", "On it."), typed("u1", "Fix the build.")]; @@ -546,7 +661,7 @@ describe("replace_send_to_user", () => { history = [typed("u1", "Fix the build."), sent("s1", "On it.")]; const error = refusalText(yield* replace("Fix the build.", "On it.", normalThread)); expect(error.split("\n")[0]).toBe( - "Refused: replace_send_to_user works only in a thread whose display mode allows one message from you per message of the user. This thread's mode does not. Use send_to_user.", + "Refused: replace_send_to_user works only in a thread whose display mode allows one answer from you per message of the user. This thread's mode does not. Use send_to_user.", ); const dashboard = yield* replace("Fix the build.", "On it.", dashboardThread); expect(dashboard.structuredContent).toEqual(SHOWN); diff --git a/apps/server/src/fork/userMessageTools/SendToUser.ts b/apps/server/src/fork/userMessageTools/SendToUser.ts index 6fa90e23a5f7d..d1c296e7fc736 100644 --- a/apps/server/src/fork/userMessageTools/SendToUser.ts +++ b/apps/server/src/fork/userMessageTools/SendToUser.ts @@ -5,6 +5,7 @@ import { } from "@t3tools/contracts"; import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations"; import { + ACK_MAX_BYTES, displayEventFromTurnItem, foldCores, freshFirstPointer, @@ -22,7 +23,6 @@ import { type VisibleCore, WRAPPED_CORE_COUNT, } from "@t3tools/shared/displayModes"; -import { resolveT3McpToolId } from "@t3tools/shared/t3McpToolPresentation"; import { TEXT_MATCH_MIN_RATIO, textSimilarity } from "@t3tools/shared/textMatch"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; @@ -36,6 +36,13 @@ import * as McpToolAccess from "../../mcp/McpToolAccess.ts"; import * as ServerSettings from "../../serverSettings.ts"; import { DisplayModeWrap } from "../displayModeWrap/DisplayModeWrap.ts"; import * as SubThreads from "../subThreads/SubThreads.ts"; +import { + judgeReplace, + judgeSend, + makeReplyHolds, + type ReplyCall, + replyCallOf, +} from "./replyStage.ts"; // The description goes before the checks: one set after a custom check is left out of the JSON schema. const MessageText = Schema.String.annotate({ @@ -45,13 +52,15 @@ const MessageText = Schema.String.annotate({ const jsonEscapedBytes = (text: string) => Buffer.byteLength(JSON.stringify(text), "utf8") - 2; // Each MessageText field goes through this in its handler; a schema failure loses the exact text. -const checkMessageSize = (text: string) => - jsonEscapedBytes(text) <= SEND_TO_USER_MAX_BYTES +const checkMessageSize = (text: string, ack = false) => + jsonEscapedBytes(text) <= (ack ? ACK_MAX_BYTES : SEND_TO_USER_MAX_BYTES) ? Effect.void : Effect.fail( new OrchestratorMcpFailure({ code: "invalid_request", - message: "only say one concise point at a time", + message: ack + ? `an ack is at most ${ACK_MAX_BYTES} bytes. Say only that you got the message and are on it.` + : "only say one concise point at a time", }), ); @@ -74,6 +83,11 @@ const UNNAMED_SEND_HOLD_MS = 10_000; const MIN_PERCENT = Math.round(TEXT_MATCH_MIN_RATIO * 100); const DONE = "If you are done, end your turn with no text."; const SHOWN = { ok: true as const, message: `Shown to the user. ${DONE}` }; +const ACKED = { + ok: true as const, + message: "Shown to the user as an ack. Send your answer when it is ready.", +}; +const LAST_MESSAGE = { what: "the user's last message", replace: REPLACE_SEND_TO_USER_TOOL_NAME }; // An agent that fills every optional field sends null, an empty string or false for "none". const Pointer = (description: string) => @@ -109,6 +123,13 @@ const SendToUserTool = Tool.make(SEND_TO_USER_TOOL_NAME, { }), ), ), + ack: Schema.optional( + Schema.NullOr( + Schema.Boolean.annotate({ + description: `True for a note of at most ${ACK_MAX_BYTES} bytes that you got the message and are on it, only when your answer will take a while. At most one, before the answer.`, + }), + ), + ), }), success: Schema.Struct({ ok: Schema.Literal(true), @@ -125,7 +146,7 @@ const SendToUserTool = Tool.make(SEND_TO_USER_TOOL_NAME, { .annotate(Tool.OpenWorld, false); const ReplaceSendToUserTool = Tool.make(REPLACE_SEND_TO_USER_TOOL_NAME, { - description: `In threads set to Messages only, Messages + Dashboard or Chat + Replies, the user gets one message from you per message of theirs. This replaces that message in place. Pass the user's last message and your current message as the recap shows them (their words only), each at least ${MIN_PERCENT}% alike, and the whole new text. In Chat + Replies, also pass reply_to or sub_thread.`, + description: `In threads set to Messages only, Messages + Dashboard or Chat + Replies, you answer each message of the user once, after at most one ack. This replaces your latest message for it in place; an ack becomes the answer. Pass the user's last message and your current message as the recap shows them (their words only), each at least ${MIN_PERCENT}% alike, and the whole new text. In Chat + Replies, also pass reply_to or sub_thread.`, parameters: Schema.Struct({ reply_to: Pointer( "Chat + Replies only. The id of the user's message whose answer you replace. user_message is then that message.", @@ -153,9 +174,6 @@ const ReplaceSendToUserTool = Tool.make(REPLACE_SEND_TO_USER_TOOL_NAME, { export const SendToUserToolkit = Toolkit.make(SendToUserTool, ReplaceSendToUserTool); -const isSendCall = (item: OrchestrationV2TurnItem) => - item.type === "dynamic_tool" && resolveT3McpToolId(item.toolName) === SEND_TO_USER_TOOL_NAME; - /** The items stored before the calling tool use, when the provider names it; else all of them. */ function itemsBefore(shown: ReadonlyArray, ownItem: string | undefined) { const ownAt = @@ -173,6 +191,19 @@ const coresOf = (items: ReadonlyArray) => }), ); +/** The calls toward the answer to the user message at `at`: those after it that `concerns` keeps. */ +function replyCallsAfter( + items: ReadonlyArray, + at: number, + named: boolean, + concerns: (item: OrchestrationV2TurnItem) => boolean = () => true, +): ReadonlyArray { + return items.slice(at + 1).flatMap((item) => { + const call = concerns(item) ? replyCallOf(item, named) : null; + return call === null ? [] : [call]; + }); +} + /** * What the thread shows since the user's last typed message. `ownItem` is the native id of the * calling tool use when the provider names it; only then do sends still running count, and only @@ -183,20 +214,15 @@ function readTurn(shown: ReadonlyArray, ownItem: string const items = itemsBefore(shown, ownItem); const cores = coresOf(items); const typedAt = cores.findLastIndex((core) => core.from === "user"); - const since = items.slice( - items.findLastIndex((item) => displayEventFromTurnItem(item)?.type === "user") + 1, - ); return { latest: cores.slice(-WRAPPED_CORE_COUNT), typed: typedAt < 0 ? undefined : cores[typedAt], current: cores.slice(typedAt + 1).findLast((core) => core.from === "agent"), - sends: since.filter((item) => { - if (!isSendCall(item)) return false; - if (ownItem === undefined) return displayEventFromTurnItem(item)?.type === "sent"; - return ( - item.nativeItemRef?.nativeId !== ownItem && displayEventFromTurnItem(item)?.type !== "other" - ); - }).length, + calls: replyCallsAfter( + items, + items.findLastIndex((item) => displayEventFromTurnItem(item)?.type === "user"), + ownItem !== undefined, + ), }; } @@ -258,7 +284,7 @@ function unansweredLines(cores: ReadonlyArray): ReadonlyArray answer.replyTo === core.shortId) && + !cores.slice(index + 1).some((answer) => answer.replyTo === core.shortId && !answer.ack) && !cores.slice(index + 1).some((later) => later.from === "user" && later.first === core.first), ); return open.length === 0 @@ -308,6 +334,25 @@ const inputField = (item: OrchestrationV2TurnItem, field: string): unknown => ? Reflect.get(item.input, field) : undefined; +/** The calls toward the answer to the user message `target` that name it by `replyTo`. */ +function answerCalls( + items: ReadonlyArray, + target: string, + replyTo: string, + named: boolean, +) { + // After the message: an earlier message may have the same short id, with an answer of its own. + const targetAt = items.findIndex( + (item) => item.type === "user_message" && item.messageId === target, + ); + return replyCallsAfter( + items, + targetAt, + named, + (item) => messagePointer(inputField(item, "reply_to")) === replyTo, + ); +} + const orchestrationError = (error: { readonly message: string }) => new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }); @@ -347,19 +392,13 @@ export const layerHandlers = McpToolAccess.toLayer( }); // A call that does not name itself cannot tell a send still running from its own, so two such - // sends in one response would both pass. The send it accepted counts for a few seconds instead, - // while the provider has not yet stored it, and only until the user types again. + // sends in one response would both pass. The stage that an accepted call reached counts for a + // few seconds instead, while the provider has not yet stored it, for that user message only. /** The ids given to first messages that may not be in the history yet, each with its hold's end. */ const givenIds = new Map(); - const unnamedSends = new Map< - string, - { readonly typed: string | null; readonly until: number } - >(); - /** The hold under `key`. Holds that ran out leave the map here, so it stays small. */ - const liveHold = (key: string, now: number) => { - for (const [held, { until }] of unnamedSends) if (until <= now) unnamedSends.delete(held); - return unnamedSends.get(key); - }; + const holds = makeReplyHolds(UNNAMED_SEND_HOLD_MS); + const holdKey = (threadId: ThreadId, userMessage: string | undefined) => + `${threadId}\n${userMessage ?? ""}`; const readThreadCores = Effect.fn("SendToUser.readThreadCores")(function* (threadId: ThreadId) { const client = yield* Effect.serviceOption(McpSchema.McpServerClient); @@ -391,6 +430,7 @@ export const layerHandlers = McpToolAccess.toLayer( readonly reply_to?: string | null | undefined; readonly user_message?: string | null | undefined; readonly new_sub_thread?: boolean | null | undefined; + readonly ack?: boolean | null | undefined; }, ) { const { items, cores, named } = yield* readThreadCores(threadId); @@ -405,6 +445,16 @@ export const layerHandlers = McpToolAccess.toLayer( unansweredLines(cores), ); } + if (replyTo === undefined && input.ack === true) { + return yield* subThreadRefusal( + "invalid_request", + [ + "an ack answers a message from the user, so it needs reply_to. Send a new message without ack.", + ], + [], + unansweredLines(cores), + ); + } 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; @@ -457,34 +507,20 @@ export const layerHandlers = McpToolAccess.toLayer( ); } const now = yield* Clock.currentTimeMillis; - const holdKey = `${threadId}\n${replyTo}`; - const held = liveHold(holdKey, now); - // After the message: an earlier message may have the same short id, with an answer of its own. - const targetAt = items.findIndex( - (item) => item.type === "user_message" && item.messageId === target.id, - ); - const answers = items - .slice(targetAt + 1) - .filter( - (item) => - isSendCall(item) && - messagePointer(inputField(item, "reply_to")) === replyTo && - (named - ? displayEventFromTurnItem(item)?.type !== "other" - : displayEventFromTurnItem(item)?.type === "sent"), - ).length; - if (answers > 0 || (!named && held !== undefined && now < held.until)) { - return yield* subThreadRefusal( - "invalid_request", - [ - `you already answered message "${replyTo}". To change your answer, call replace_send_to_user with reply_to "${replyTo}".`, - ], - thread, - [], - ); - } - if (!named) - unnamedSends.set(holdKey, { typed: target.id, until: now + UNNAMED_SEND_HOLD_MS }); + const key = holdKey(threadId, target.id); + const verdict = judgeSend({ + calls: answerCalls(items, target.id, replyTo, named), + held: holds.held(key, named, now), + kind: input.ack === true ? "ack" : "answer", + subject: { + what: `message "${replyTo}"`, + replace: `${REPLACE_SEND_TO_USER_TOOL_NAME} with reply_to "${replyTo}"`, + }, + }); + if (!verdict.ok) + return yield* subThreadRefusal("invalid_request", [verdict.reason], thread, []); + holds.record(key, named, input.ack === true ? "ack" : "answer", verdict.next, now); + if (input.ack === true) return ACKED; yield* subThreadChanged( threadId, [...thread, { id: "", from: "agent", text: input.text, first: target.first, reply: true }], @@ -506,7 +542,7 @@ export const layerHandlers = McpToolAccess.toLayer( readonly new_text: string; }, ) { - const { cores, coresNow } = yield* readThreadCores(threadId); + const { items, cores, coresNow, named } = yield* readThreadCores(threadId); const replyTo = messagePointer(input.reply_to); const subThread = messagePointer(input.sub_thread); if ((replyTo === undefined) === (subThread === undefined)) { @@ -530,6 +566,19 @@ export const layerHandlers = McpToolAccess.toLayer( ); } const thread = subThreadCores(cores, target.first); + const now = yield* Clock.currentTimeMillis; + const key = holdKey(threadId, target.id); + const verdict = judgeReplace({ + calls: answerCalls(items, target.id, replyTo, named), + held: holds.held(key, named, now), + subject: { + what: `message "${replyTo}"`, + replace: `${REPLACE_SEND_TO_USER_TOOL_NAME} with reply_to "${replyTo}"`, + }, + }); + if (!verdict.ok) { + return yield* subThreadRefusal("invalid_request", [verdict.reason], thread, []); + } const answer = thread .slice(thread.indexOf(target) + 1) .findLast((core) => core.replyTo === replyTo); @@ -578,6 +627,7 @@ export const layerHandlers = McpToolAccess.toLayer( CHECK_SUB_THREAD, ]); } + holds.record(key, named, "replace", verdict.next, now); yield* subThreadChanged( threadId, thread.map((core) => (core === answer ? { ...core, text: input.new_text } : core)), @@ -623,30 +673,34 @@ export const layerHandlers = McpToolAccess.toLayer( return { [SEND_TO_USER_TOOL_NAME]: McpToolAccess.readsAsCaller((input) => Effect.gen(function* () { - const { text } = input; - yield* checkMessageSize(text); + const ack = input.ack === true; + yield* checkMessageSize(input.text, ack); const { threadId, mode } = yield* callerThread; if (mode.subThreads) return yield* sendInSubThread(threadId, input).pipe(lockFor(threadId).withPermits(1)); if (!mode.oneSendPerMessage) return SHOWN; return yield* Effect.gen(function* () { const turn = yield* readThreadTurn(threadId); - const now = yield* Clock.currentTimeMillis; - const typed = turn.typed?.id ?? null; - const held = liveHold(threadId, now); - const heldSend = !turn.named && held?.typed === typed && now < held.until; - if (turn.sends > 0 || heldSend) { + if (ack && turn.typed === undefined) { return yield* refusal( "invalid_request", [ - "you already sent the user a message after their last one, and this thread allows one per message of theirs. To change it, call replace_send_to_user.", + "an ack answers a message from the user, and they have typed none. Send without ack.", ], turn, ); } - if (!turn.named) - unnamedSends.set(threadId, { typed, until: now + UNNAMED_SEND_HOLD_MS }); - return SHOWN; + const now = yield* Clock.currentTimeMillis; + const key = holdKey(threadId, turn.typed?.id); + const verdict = judgeSend({ + calls: turn.calls, + held: holds.held(key, turn.named, now), + kind: ack ? "ack" : "answer", + subject: LAST_MESSAGE, + }); + if (!verdict.ok) return yield* refusal("invalid_request", [verdict.reason], turn); + holds.record(key, turn.named, ack ? "ack" : "answer", verdict.next, now); + return ack ? ACKED : SHOWN; }).pipe(lockFor(threadId).withPermits(1)); }), ), @@ -677,11 +731,19 @@ export const layerHandlers = McpToolAccess.toLayer( return yield* refusal( "capability_denied", [ - "replace_send_to_user works only in a thread whose display mode allows one message from you per message of the user. This thread's mode does not. Use send_to_user.", + "replace_send_to_user works only in a thread whose display mode allows one answer from you per message of the user. This thread's mode does not. Use send_to_user.", ], turn, ); } + const now = yield* Clock.currentTimeMillis; + const key = holdKey(threadId, turn.typed?.id); + const verdict = judgeReplace({ + calls: turn.calls, + held: holds.held(key, turn.named, now), + subject: LAST_MESSAGE, + }); + if (!verdict.ok) return yield* refusal("invalid_request", [verdict.reason], turn); if (turn.current === undefined) { return yield* refusal( "invalid_request", @@ -705,6 +767,7 @@ export const layerHandlers = McpToolAccess.toLayer( ]), ]; if (mismatches.length > 0) return yield* refusal("invalid_request", mismatches, turn); + holds.record(key, turn.named, "replace", verdict.next, now); return SHOWN; }).pipe(lockFor(threadId).withPermits(1)); }), diff --git a/apps/server/src/fork/userMessageTools/SendToUserSubThreads.test.ts b/apps/server/src/fork/userMessageTools/SendToUserSubThreads.test.ts index c9ba30ad9bcb9..d0a7764bf2099 100644 --- a/apps/server/src/fork/userMessageTools/SendToUserSubThreads.test.ts +++ b/apps/server/src/fork/userMessageTools/SendToUserSubThreads.test.ts @@ -109,6 +109,8 @@ const sent = (id: string, input: Record) => }) satisfies OrchestrationV2TurnItem; const answer = (id: string, text: string, to: string) => sent(id, { text, reply_to: shortId(to), user_message: "" }); +const ackOf = (id: string, text: string, to: string) => + sent(id, { text, reply_to: shortId(to), user_message: "", ack: true }); const news = (id: string, text: string) => sent(id, { text, new_sub_thread: true }); let history: ReadonlyArray = []; @@ -182,7 +184,8 @@ const twoSubThreads = () => [ it.effect("refuses a send that names neither a message to answer nor a new sub-thread", () => Effect.gen(function* () { - setup(twoSubThreads()); + // An ack is not an answer, so its message stays in the list. + setup([...twoSubThreads(), ackOf("k2", "On it.", "q2")]); for (const args of [ { text: "Lint passes too." }, { text: "Lint passes too.", reply_to: "", new_sub_thread: false }, @@ -349,6 +352,34 @@ it.effect("refuses a second answer to one message, and points to the replace too }).pipe(Effect.provide(layer)), ); +it.effect("takes one ack before the answer to a message, and none for a new message", () => + Effect.gen(function* () { + setup(twoSubThreads()); + const q2 = { reply_to: shortId("q2"), user_message: "And the lint job?" }; + const ack = yield* call("send_to_user", { text: "Checking.", ack: true, ...q2 }); + expect(ack.structuredContent).toEqual({ + ok: true, + message: "Shown to the user as an ack. Send your answer when it is ready.", + }); + expect(changes).toEqual([]); + expect( + refusalLines(yield* call("send_to_user", { text: "Still checking.", ack: true, ...q2 }))[0], + ).toBe( + `Refused: you already sent an ack for message "${shortId("q2")}". Send your answer without ack, or make the ack your answer with replace_send_to_user with reply_to "${shortId("q2")}".`, + ); + const answer = yield* call("send_to_user", { text: "Lint passes too.", ...q2 }); + expect(answer.isError).not.toBe(true); + expect(changes.map((change) => change.lines.at(-1))).toEqual(["[you] Lint passes too."]); + expect( + refusalLines(yield* call("send_to_user", { text: "Done.", ack: true, ...q2 }))[0], + ).toContain("an ack goes only before the answer"); + const fresh = yield* call("send_to_user", { text: "On it.", ack: true, new_sub_thread: true }); + expect(refusalLines(fresh)[0]).toBe( + "Refused: an ack answers a message from the user, so it needs reply_to. Send a new message without ack.", + ); + }).pipe(Effect.provide(layer)), +); + it.effect("holds an accepted answer against a second one that arrives before it is stored", () => Effect.gen(function* () { setup(twoSubThreads()); @@ -368,6 +399,25 @@ it.effect("holds an accepted answer against a second one that arrives before it }).pipe(Effect.provide(layer)), ); +it.effect("holds a replace that makes an ack the answer, before the replace is stored", () => + Effect.gen(function* () { + setup([...twoSubThreads(), ackOf("k2", "Checking.", "q2")]); + const replaced = yield* call("replace_send_to_user", { + reply_to: shortId("q2"), + user_message: "And the lint job?", + old_text: "Checking.", + new_text: "Lint passes too.", + }); + expect(replaced.isError).not.toBe(true); + const again = yield* call("send_to_user", { + text: "Types pass too.", + reply_to: shortId("q2"), + user_message: "And the lint job?", + }); + expect(refusalLines(again)[0]).toContain("you already answered"); + }).pipe(Effect.provide(layer)), +); + it.effect("replaces the answer to the named message only", () => Effect.gen(function* () { setup([...twoSubThreads(), answer("a2", "Lint passes too.", "q2")]); @@ -491,7 +541,7 @@ it.effect("keeps sub-thread pointers out of a thread with no sub-threads", () => { text: "Done.", reply_to: shortId("u1"), user_message: "Fix the build." }, messagesThread, ); - expect(refusalLines(ignored)[0]).toContain("you already sent the user a message"); + expect(refusalLines(ignored)[0]).toContain("you already answered the user's last message"); const refused = refusalLines( yield* call( "replace_send_to_user", @@ -515,6 +565,10 @@ it.effect("gives two user messages with one short id an answer each", () => // Both ids hash to "003d88". const [early, late] = ["message-11914", "message-20610"] as const; expect(shortId(early)).toBe(shortId(late)); + // The answer to the earlier one is not stored yet: its hold is that message's, not the id's. + setup([typed(early, "First?")]); + const first = { text: "One.", reply_to: shortId(early), user_message: "First?" }; + expect((yield* call("send_to_user", first)).isError).not.toBe(true); setup([typed(early, "First?"), answer("a1", "One.", early), typed(late, "Second?")]); const args = { text: "Two.", reply_to: shortId(late), user_message: "Second?" }; expect((yield* call("send_to_user", args)).isError).not.toBe(true); @@ -530,6 +584,30 @@ it.effect("gives two user messages with one short id an answer each", () => }).pipe(Effect.provide(layer)), ); +it.effect("refuses a replace while the answer after an ack is still being delivered", () => + Effect.gen(function* () { + setup([ + ...twoSubThreads(), + ackOf("k2", "Checking.", "q2"), + { ...answer("a2", "Lint passes too.", "q2"), status: "running" as const }, + ]); + const replaced = yield* call( + "replace_send_to_user", + { + reply_to: shortId("q2"), + user_message: "And the lint job?", + old_text: "Checking.", + new_text: "Lint and types pass.", + }, + repliesThread, + "r2", + ); + expect(refusalLines(replaced)[0]).toBe( + `Refused: your message for message "${shortId("q2")}" is still being delivered. Call replace_send_to_user with reply_to "${shortId("q2")}" again.`, + ); + }).pipe(Effect.provide(layer)), +); + it.effect("judges a named call by the history before it, and a replace also by replies since", () => Effect.gen(function* () { const running = (id: string, tool: string, input: Record) => diff --git a/apps/server/src/fork/userMessageTools/replyStage.ts b/apps/server/src/fork/userMessageTools/replyStage.ts new file mode 100644 index 0000000000000..37b0b901d6fdc --- /dev/null +++ b/apps/server/src/fork/userMessageTools/replyStage.ts @@ -0,0 +1,141 @@ +import type { OrchestrationV2TurnItem } from "@t3tools/contracts"; +import { + displayEventFromTurnItem, + isAckInput, + REPLACE_SEND_TO_USER_TOOL_NAME, + SEND_TO_USER_TOOL_NAME, +} from "@t3tools/shared/displayModes"; +import { resolveT3McpToolId } from "@t3tools/shared/t3McpToolPresentation"; + +/** How far the agent's answer to one user message has come. It only moves forward. */ +export type ReplyStage = "open" | "acked" | "answered"; + +/** A call toward that answer. `delivered` is false for a send that still runs. */ +export interface ReplyCall { + readonly kind: "ack" | "answer" | "replace"; + readonly delivered: boolean; +} + +/** Names the user message in refusals, and the call that changes its answer. */ +export interface ReplySubject { + readonly what: string; + readonly replace: string; +} + +export type ReplyVerdict = + | { readonly ok: true; readonly next: ReplyStage } + | { readonly ok: false; readonly reason: string }; + +type Refusal = "second-ack" | "late-ack" | "second-answer"; + +const RANK: Record = { open: 0, acked: 1, answered: 2 }; +const REACHES: Record = { + ack: "acked", + answer: "answered", + replace: "answered", +}; +const SEND_STEPS: Record< + ReplyStage, + Record<"ack" | "answer", { readonly next: ReplyStage } | { readonly refused: Refusal }> +> = { + open: { ack: { next: "acked" }, answer: { next: "answered" } }, + acked: { ack: { refused: "second-ack" }, answer: { next: "answered" } }, + answered: { ack: { refused: "late-ack" }, answer: { refused: "second-answer" } }, +}; +const REFUSALS: Record string> = { + "second-answer": (s) => + `you already answered ${s.what}. To change your answer, call ${s.replace}.`, + "second-ack": (s) => + `you already sent an ack for ${s.what}. Send your answer without ack, or make the ack your answer with ${s.replace}.`, + "late-ack": (s) => + `you already answered ${s.what}, and an ack goes only before the answer. To change your answer, call ${s.replace}.`, +}; +const DELIVERING = (s: ReplySubject) => + `your message for ${s.what} is still being delivered. Call ${s.replace} again.`; + +/** + * The call that `item` makes toward an answer. Null for any other item, and for a call that does + * not count: a refused or failed one, a replace still running, and without `named`, a send still + * running, which the caller's hold covers instead. + */ +export function replyCallOf(item: OrchestrationV2TurnItem, named: boolean): ReplyCall | null { + if (item.type !== "dynamic_tool") return null; + const event = displayEventFromTurnItem(item); + switch (resolveT3McpToolId(item.toolName)) { + case SEND_TO_USER_TOOL_NAME: { + const kind = isAckInput(item.input) ? "ack" : "answer"; + if (event?.type === "sent") return { kind, delivered: true }; + return named && event === null ? { kind, delivered: false } : null; + } + case REPLACE_SEND_TO_USER_TOOL_NAME: + return event?.type === "rewrite" ? { kind: "replace", delivered: true } : null; + default: + return null; + } +} + +const furthest = (stages: Iterable) => { + let stage: ReplyStage = "open"; + for (const reached of stages) if (RANK[reached] > RANK[stage]) stage = reached; + return stage; +}; + +const reachOf = (calls: ReadonlyArray) => calls.map((call) => REACHES[call.kind]); + +/** Whether a send of `kind` may go out now, and the stage it leaves. `held` is the hold's stage. */ +export function judgeSend(input: { + readonly calls: ReadonlyArray; + readonly held: ReplyStage; + readonly kind: "ack" | "answer"; + readonly subject: ReplySubject; +}): ReplyVerdict { + const step = SEND_STEPS[furthest([input.held, ...reachOf(input.calls)])][input.kind]; + return "next" in step + ? { ok: true, next: step.next } + : { ok: false, reason: REFUSALS[step.refused](input.subject) }; +} + +/** + * Whether a replace may go out now. It checks its old text against the delivered messages, so it + * waits while an accepted send is not delivered yet. A replace always leaves an answer. + */ +export function judgeReplace(input: { + readonly calls: ReadonlyArray; + readonly held: ReplyStage; + readonly subject: ReplySubject; +}): ReplyVerdict { + const accepted = furthest([input.held, ...reachOf(input.calls)]); + const delivered = furthest(reachOf(input.calls.filter((call) => call.delivered))); + return RANK[accepted] > RANK[delivered] + ? { ok: false, reason: DELIVERING(input.subject) } + : { ok: true, next: "answered" }; +} + +/** + * The stages that accepted calls reached before the provider stored them, by user message, each + * for `holdMs`. A call that names itself counts from history while it runs, except a replace: a + * running replace may still be refused, so only a replace that the handler accepted is held. + */ +export function makeReplyHolds(holdMs: number) { + const holds = new Map< + string, + { readonly stage: ReplyStage; readonly until: number; readonly replace: boolean } + >(); + return { + held: (key: string, named: boolean, now: number): ReplyStage => { + for (const [held, { until }] of holds) if (until <= now) holds.delete(held); + const hold = holds.get(key); + return hold === undefined || (named && !hold.replace) ? "open" : hold.stage; + }, + record: ( + key: string, + named: boolean, + call: ReplyCall["kind"], + stage: ReplyStage, + now: number, + ) => { + if (named && call !== "replace") return; + holds.set(key, { stage, until: now + holdMs, replace: call === "replace" }); + }, + }; +} diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 2e2f37cb50643..c6a5115d546ee 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5294, "codexAdditionalContextBytes": 7024, - "mcpToolListJsonBytes": 52134 + "mcpToolListJsonBytes": 52373 } diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index c522cb30d193b..3a08177cee30f 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -605,6 +605,7 @@ type MessagesTimelineRowContent = groupedWithPrevious?: boolean; /** Set when the message above is a steer too, so the label shows once. */ continuesSteer?: boolean; + followsAgentMessage?: boolean; } | { kind: "assistant-meta"; @@ -2791,7 +2792,8 @@ function isRowUnchanged(a: MessagesTimelineRow, b: MessagesTimelineRow): boolean a.revertTurnCount === bm.revertTurnCount && a.groupedWithNext === bm.groupedWithNext && a.groupedWithPrevious === bm.groupedWithPrevious && - a.continuesSteer === bm.continuesSteer + a.continuesSteer === bm.continuesSteer && + a.followsAgentMessage === bm.followsAgentMessage ); } } diff --git a/apps/web/src/components/chat/TimelineAssistantRows.tsx b/apps/web/src/components/chat/TimelineAssistantRows.tsx index 71357d8f6f962..49e4e0c19adb2 100644 --- a/apps/web/src/components/chat/TimelineAssistantRows.tsx +++ b/apps/web/src/components/chat/TimelineAssistantRows.tsx @@ -31,6 +31,8 @@ import { useV2ItemSupport } from "../../state/v2ItemSupport"; import { TimelineRowCtx, TimelineRowActivityCtx, type TimelineRow } from "./timelineRowContext"; import { MESSAGE_HEADING_LEVEL, MessageAuthorHeading } from "./TimelineRowParts"; import { SubThreadReplyLine } from "../../fork/subThreads/SubThreadReplyLine"; +import { AgentMessageDivider } from "../../fork/displayMode/AgentMessageDivider"; +import { EditedMark } from "../../fork/displayMode/EditedMark"; export function AssistantTimelineRow({ row }: { row: Extract }) { const ctx = use(TimelineRowCtx); @@ -46,6 +48,7 @@ export function AssistantTimelineRow({ row }: { row: Extract
+ T3 Code
+ {ctx.onOpenTurnDiff ? ( : null; +} diff --git a/apps/web/src/fork/displayMode/EditedMark.tsx b/apps/web/src/fork/displayMode/EditedMark.tsx new file mode 100644 index 0000000000000..c4503f7d9ab31 --- /dev/null +++ b/apps/web/src/fork/displayMode/EditedMark.tsx @@ -0,0 +1,27 @@ +import { PencilIcon } from "lucide-react"; +import { use } from "react"; + +import { TimelineRowCtx } from "../../components/chat/timelineRowContext"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../../components/ui/tooltip"; +import { formatChatTimestampTooltip } from "../../timestampFormat"; + +/** Under a message that a rewrite changed, where other chat apps mark an edited message. */ +export function EditedMark(props: { readonly editedAt: string | undefined }) { + if (props.editedAt === undefined) return null; + const { timestampFormat } = use(TimelineRowCtx); + return ( +
+ + + } + > + + Edited + + {formatChatTimestampTooltip(props.editedAt, timestampFormat)} + +
+ ); +} diff --git a/apps/web/src/fork/displayMode/displayModeTimeline.ts b/apps/web/src/fork/displayMode/displayModeTimeline.ts index 8dd986085eec9..1b11e91a84992 100644 --- a/apps/web/src/fork/displayMode/displayModeTimeline.ts +++ b/apps/web/src/fork/displayMode/displayModeTimeline.ts @@ -8,6 +8,7 @@ import { isReplaceSendCall, type VisibleCore, } from "@t3tools/shared/displayModes"; +import * as DateTime from "effect/DateTime"; import type { MessagesTimelineRow } from "../../components/chat/MessagesTimeline.logic"; import type { TimelineEntry } from "../../session-logic"; @@ -32,7 +33,7 @@ function displayEventFromEntry(entry: TimelineEntry): DisplayEvent | null { } // Entries change on every streamed delta. A reply that keeps its object keeps its row from redrawing, -// until a rewrite changes its text without changing its entry. +// until a rewrite changes its text or edit time without changing its entry. const repliesInRun = new WeakMap>(); const repliesOutsideRun = new WeakMap>(); @@ -40,10 +41,15 @@ const repliesOutsideRun = new WeakMap = { id: entry.id, kind: "message", @@ -57,6 +63,7 @@ function replyEntry(entry: TimelineEntry, text: string, inRun: boolean): Timelin createdBy: "agent", createdAt: entry.createdAt, updatedAt: entry.createdAt, + ...(editedAt === undefined ? {} : { editedAt }), }, }; replies.set(entry, reply); @@ -67,6 +74,32 @@ interface FoldedTimeline { readonly events: ReadonlyArray; readonly shown: ReadonlyArray; readonly cores: ReadonlyArray; + /** By core id, when the rewrite that last changed the core's text completed. Edited cores only. */ + readonly editedAt: ReadonlyMap; +} + +const NOT_EDITED: ReadonlyMap = new Map(); + +function editTimes( + entries: ReadonlyArray, + events: ReadonlyArray, + cores: ReadonlyArray, +): ReadonlyMap { + if (!cores.some((core) => core.editedBy !== undefined)) return NOT_EDITED; + // An entry's time is its call's start; the user saw the new text when the call completed. + const rewriteAt = new Map(); + events.forEach((event, index) => { + if (event?.type !== "rewrite") return; + const entry = entries[index]!; + const completedAt = entry.kind === "work" ? entry.entry.structuredPayload?.completedAt : null; + rewriteAt.set(event.id, completedAt ? DateTime.formatIso(completedAt) : entry.createdAt); + }); + return new Map( + cores.flatMap((core): [string, string][] => { + const at = core.editedBy === undefined ? undefined : rewriteAt.get(core.editedBy); + return at === undefined ? [] : [[core.id, at]]; + }), + ); } // The rows and the sub-thread summaries of one timeline come from the same fold. @@ -77,7 +110,8 @@ function foldTimeline(entries: ReadonlyArray): FoldedTimeline { if (folded === undefined) { const events = entries.map(displayEventFromEntry); const shown = events.filter((event) => event !== null); - folded = { events, shown, cores: foldCores(shown) }; + const cores = foldCores(shown); + folded = { events, shown, cores, editedAt: editTimes(entries, events, cores) }; foldedTimelines.set(entries, folded); } return folded; @@ -91,7 +125,8 @@ const sameSummary = (a: SubThreadSummary, b: SubThreadSummary) => a.lastFrom === b.lastFrom && a.lastText === b.lastText && a.firstText === b.firstText && - a.agentReplyTimes.length === b.agentReplyTimes.length; + a.agentReplyTimes.length === b.agentReplyTimes.length && + a.agentReplyTimes.every((at, index) => at === b.agentReplyTimes[index]); /** * Each sub-thread that the loaded entries show, by the id of its first message, which must be @@ -105,9 +140,11 @@ export function timelineSubThreads( const known = summariesOfTimeline.get(entries); if (known !== undefined) return known; const createdAt = new Map(entries.map((entry) => [entry.id, entry.createdAt])); + const { cores, editedAt } = foldTimeline(entries); const summaries = new Map(); - for (const core of foldTimeline(entries).cores) { - const at = createdAt.get(core.id) ?? ""; + for (const core of cores) { + // Only the latest agent message of a sub-thread can be rewritten, so agentReplyTimes stays in order. + const at = editedAt.get(core.id) ?? createdAt.get(core.id) ?? ""; if (!core.reply) { summaries.set(core.id, { first: core.id, @@ -123,12 +160,12 @@ export function timelineSubThreads( } const summary = core.first === null ? undefined : summaries.get(core.first); if (summary === undefined) continue; + // A reply typed while an edit was delivered comes after the edited message but is older. + const latest = !(Date.parse(at) < Date.parse(summary.lastAt)); summaries.set(summary.first, { ...summary, replies: summary.replies + 1, - lastAt: at, - lastFrom: core.from, - lastText: core.text, + ...(latest && { lastAt: at, lastFrom: core.from, lastText: core.text }), agentReplyTimes: core.from === "agent" ? [...summary.agentReplyTimes, at] : summary.agentReplyTimes, }); @@ -189,7 +226,7 @@ export function applyDisplayMode( entries: ReadonlyArray, subThread: string | null = null, ): ReadonlyArray { - const { events, shown, cores } = foldTimeline(entries); + const { events, shown, cores, editedAt } = foldTimeline(entries); const coreTexts = new Map(cores.map((core) => [core.id, core.text])); const shownIds = mode.subThreads ? new Set( @@ -207,7 +244,12 @@ export function applyDisplayMode( const event = events[index]; return [ event?.type === "sent" - ? replyEntry(entry, coreTexts.get(entry.id) ?? event.text, shownIds === null) + ? replyEntry( + entry, + coreTexts.get(entry.id) ?? event.text, + editedAt.get(entry.id), + shownIds === null, + ) : entry, ]; }); @@ -216,12 +258,22 @@ export function applyDisplayMode( : rows; } -/** A mode that chooses its replies draws no Working or Thinking row. */ +/** + * A mode that chooses its replies draws no Working or Thinking row, and marks an agent message + * drawn right after another one, with no user message between, so a rule can part the two. + */ export function displayModeRows( rows: MessagesTimelineRow[], repliesChosen: boolean, ): MessagesTimelineRow[] { - return repliesChosen - ? rows.filter((row) => row.kind !== "working" && row.kind !== "thinking") - : rows; + if (!repliesChosen) return rows; + let afterAgent = false; + return rows.flatMap((row): MessagesTimelineRow[] => { + if (row.kind === "working" || row.kind === "thinking") return []; + if (row.kind !== "message") return [row]; + const fromAgent = row.message.role === "assistant"; + const follows = fromAgent && afterAgent; + afterAgent = fromAgent; + return [follows ? { ...row, followsAgentMessage: true } : row]; + }); } diff --git a/apps/web/src/fork/subThreads/subThreadTimeline.test.ts b/apps/web/src/fork/subThreads/subThreadTimeline.test.ts index 0bd554517bcba..acd045d74485d 100644 --- a/apps/web/src/fork/subThreads/subThreadTimeline.test.ts +++ b/apps/web/src/fork/subThreads/subThreadTimeline.test.ts @@ -17,10 +17,12 @@ import { import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; +import type { MessagesTimelineRow } from "../../components/chat/MessagesTimeline.logic"; import { type ComposerThreadDraftState, useComposerDraftStore } from "../../composerDraftStore"; -import { deriveTimelineEntriesFromVisibleTurnItems } from "../../session-logic"; +import { deriveTimelineEntriesFromVisibleTurnItems, type TimelineEntry } from "../../session-logic"; import { applyDisplayMode, + displayModeRows, rowShownIn, subThreadOfReply, timelineSubThreads, @@ -90,6 +92,17 @@ const sent = (id: string, second: number, input: Record) => output: { ok: true }, }) satisfies OrchestrationV2TurnItem; +/** A replace call that started at `second` and completed at `done`. */ +const rewrite = (id: string, second: number, done: number, input: Record) => + ({ + ...base(id, second), + completedAt: DateTime.makeUnsafe(time(done)), + type: "dynamic_tool", + toolName: "mcp__t3-code__replace_send_to_user", + input: { user_message: "x", old_text: "x", ...input }, + output: { ok: true }, + }) satisfies OrchestrationV2TurnItem; + const entriesOf = (items: ReadonlyArray) => deriveTimelineEntriesFromVisibleTurnItems({ visibleTurnItems: items.map((item, position): OrchestrationV2ProjectedTurnItem => ({ @@ -114,6 +127,9 @@ const items = [ ]; const entries = entriesOf(items); const ids = (rows: ReadonlyArray<{ readonly id: string }>) => rows.map((row) => row.id); +const editLint = rewrite("w2", 8, 9, { reply_to: shortId("q2"), new_text: "Lint and types pass." }); +const readAt = (at: string | null) => + new Map([["q1", { firstId: "q1", readAt: at, summary: null }]]); describe("Chat + Replies rows", () => { it("draws only the first messages in the main list", () => { @@ -133,6 +149,28 @@ describe("Chat + Replies rows", () => { ]); }); + it("marks a rewritten message with the time its rewrite completed, a first message too", () => { + const editNews = rewrite("w1", 8, 9, { + sub_thread: shortId("The deploy is done."), + new_text: "The deploy is done and checked.", + }); + const edited = entriesOf([...items, editNews, editLint]); + const editedAt = (rows: ReadonlyArray) => + rows.map((row) => (row.kind === "message" ? row.message.editedAt : null)); + expect(editedAt(applyDisplayMode(CHAT_REPLIES_DISPLAY_MODE, edited))).toEqual([ + undefined, + time(9), + ]); + expect(editedAt(applyDisplayMode(CHAT_REPLIES_DISPLAY_MODE, edited, "q1"))).toEqual([ + undefined, + undefined, + undefined, + time(9), + ]); + // A first message has no unread state: its edit is only marked. + expect(timelineSubThreads(edited, null).get("news")?.agentReplyTimes).toEqual([]); + }); + it("stays a flat list in Messages only", () => { expect(ids(applyDisplayMode(MESSAGES_ONLY_DISPLAY_MODE, entries))).toEqual([ "q1", @@ -208,8 +246,6 @@ describe("timelineSubThreads", () => { it("counts as unread the agent's replies after the user's last read", () => { const summary = timelineSubThreads(entries, null).get("q1")!; - const readAt = (at: string | null) => - new Map([["q1", { firstId: "q1", readAt: at, summary: null }]]); expect(unreadReplies(summary, new Map())).toBe(2); expect(unreadReplies(summary, readAt(null))).toBe(2); expect(unreadReplies(summary, readAt(time(3)))).toBe(1); @@ -217,6 +253,77 @@ describe("timelineSubThreads", () => { // Before the server's first answer, the read times are unknown and nothing shows as new. expect(unreadReplies(summary, NO_STATES)).toBe(0); }); + + it("counts a reply edited after the user read it as new, from when the edit completed", () => { + const before = timelineSubThreads(entries, null); + const edited = timelineSubThreads(entriesOf([...items, editLint]), before); + const summary = edited.get("q1")!; + expect(summary).toMatchObject({ + replies: 3, + lastAt: time(9), + lastText: "Lint and types pass.", + agentReplyTimes: [time(3), time(9)], + }); + expect(unreadReplies(summary, readAt(time(6)))).toBe(1); + expect(edited.get("news")).toBe(before.get("news")); + // The user replied while the edit ran: only the reply's time moves, and still makes a new summary. + const followUp = typed("q3", 7, "And the types?", "q1"); + const unedited = timelineSubThreads(entriesOf([...items, followUp]), null); + const late = timelineSubThreads(entriesOf([...items, followUp, editLint]), unedited); + expect(late.get("q1")).not.toBe(unedited.get("q1")); + expect(late.get("q1")).toMatchObject({ + agentReplyTimes: [time(3), time(9)], + lastAt: time(9), + lastFrom: "agent", + lastText: "Lint and types pass.", + }); + }); +}); + +describe("displayModeRows", () => { + const message = (id: string, role: "user" | "assistant") => ({ + id: MessageId.make(id), + role, + text: id, + runId: null, + streaming: false, + createdAt: time(0), + updatedAt: time(0), + }); + const row = (id: string, role: "user" | "assistant"): MessagesTimelineRow => ({ + kind: "message", + id, + createdAt: time(0), + message: message(id, role), + showAssistantMeta: false, + showAssistantCopyButton: false, + assistantCopyStreaming: false, + }); + const meta: MessagesTimelineRow = { + kind: "assistant-meta", + id: "a1:meta", + createdAt: time(0), + message: message("a1", "assistant"), + showAssistantCopyButton: true, + assistantCopyStreaming: false, + }; + + it("marks an agent message drawn right after another, only in modes that choose replies", () => { + const rows = [ + row("u1", "user"), + row("a1", "assistant"), + meta, + row("a2", "assistant"), + row("u2", "user"), + row("a3", "assistant"), + ]; + expect(displayModeRows(rows, false)).toBe(rows); + const marked = displayModeRows(rows, true); + expect( + marked.map((each) => each.kind === "message" && each.followsAgentMessage === true), + ).toEqual([false, false, false, true, false, false]); + expect(marked[1]).toBe(rows[1]); + }); }); describe("withSubThreadRecord", () => { diff --git a/apps/web/src/fork/subThreads/subThreadsState.ts b/apps/web/src/fork/subThreads/subThreadsState.ts index bd91e708e847b..450d800c34331 100644 --- a/apps/web/src/fork/subThreads/subThreadsState.ts +++ b/apps/web/src/fork/subThreads/subThreadsState.ts @@ -36,11 +36,11 @@ export interface SubThreadSummary { readonly firstFrom: "user" | "agent"; readonly firstText: string; readonly replies: number; - /** When its latest message was made. */ + /** The activity time of its latest message: when it was sent, or last edited. */ readonly lastAt: string; readonly lastFrom: "user" | "agent"; readonly lastText: string; - /** When each of the agent's replies was made, oldest first. */ + /** The activity time of each of the agent's replies (send, or latest edit), in message order. */ readonly agentReplyTimes: ReadonlyArray; } diff --git a/apps/web/src/types.ts b/apps/web/src/types.ts index 2f61b38b9b182..614825916e293 100644 --- a/apps/web/src/types.ts +++ b/apps/web/src/types.ts @@ -89,6 +89,7 @@ export interface ChatMessage { readonly createdAt: string; readonly updatedAt: string; readonly inputIntent?: OrchestrationV2UserMessageInputIntent | undefined; + readonly editedAt?: string | undefined; } export interface ProposedPlan { diff --git a/packages/shared/src/fork/displayModes/displayModes.test.ts b/packages/shared/src/fork/displayModes/displayModes.test.ts index 39297de1409b6..9c35557f70677 100644 --- a/packages/shared/src/fork/displayModes/displayModes.test.ts +++ b/packages/shared/src/fork/displayModes/displayModes.test.ts @@ -147,6 +147,33 @@ describe("displayEventFromTurnItem", () => { { type: "dynamic_tool", toolName: "mcp__t3-code__send_to_user", input: { text: "Done" } }, { type: "sent", id: "item-1", text: "Done" }, ], + [ + "a completed ack", + { + type: "dynamic_tool", + toolName: "mcp__t3-code__send_to_user", + input: { text: "On it.", ack: true }, + }, + { type: "sent", id: "item-1", text: "On it.", ack: true }, + ], + [ + "a send with ack false", + { + type: "dynamic_tool", + toolName: "mcp__t3-code__send_to_user", + input: { text: "Done", ack: false }, + }, + { type: "sent", id: "item-1", text: "Done" }, + ], + [ + "a send with ack null", + { + type: "dynamic_tool", + toolName: "mcp__t3-code__send_to_user", + input: { text: "Done", ack: null }, + }, + { type: "sent", id: "item-1", text: "Done" }, + ], [ "a failed send", { @@ -275,11 +302,25 @@ describe("visibleCores", () => { ).toMatchObject([ { id: "sent:A", from: "agent", text: "A" }, { id: "user:B", from: "user", text: "B" }, - { id: "sent:C", from: "agent", text: "C3" }, + { id: "sent:C", from: "agent", text: "C3", editedBy: "rewrite:C3" }, { id: "user:D", from: "user", text: "D" }, ]); }); + it("marks a send edited only by a rewrite that changes its text, and any rewrite ends an ack", () => { + const ack: DisplayEvent = { type: "sent", id: "sent:On it.", text: "On it.", ack: true }; + const core = { id: "sent:On it.", from: "agent", first: "sent:On it.", reply: false }; + const shortIdOf = { shortId: shortId("On it.") }; + expect(foldCores([ack])).toEqual([{ ...core, ...shortIdOf, text: "On it.", ack: true }]); + expect(foldCores([ack, rewrite("On it.")])).toEqual([ + { ...core, ...shortIdOf, text: "On it." }, + ]); + const again: DisplayEvent = { type: "rewrite", id: "again", text: "Done." }; + expect(foldCores([ack, rewrite("Done."), again])).toEqual([ + { ...core, ...shortIdOf, text: "Done.", editedBy: "rewrite:Done." }, + ]); + }); + it("shows nothing for a rewrite with no send before it", () => { expect(MESSAGES_ONLY_DISPLAY_MODE.visibleCores!([rewrite("X"), user("A")])).toMatchObject([ { id: "user:A", from: "user", text: "A" }, @@ -480,6 +521,15 @@ describe("sub-threads", () => { "Because", "A2", ]); + expect(cores.map((core) => core.editedBy)).toEqual([ + undefined, + "r2", + "r1", + undefined, + undefined, + undefined, + undefined, + ]); }); it("names the agent's first message by the text it was sent with, also after a rewrite", () => { diff --git a/packages/shared/src/fork/displayModes/displayModes.ts b/packages/shared/src/fork/displayModes/displayModes.ts index 0aba9f39f7a40..d699fd3643b50 100644 --- a/packages/shared/src/fork/displayModes/displayModes.ts +++ b/packages/shared/src/fork/displayModes/displayModes.ts @@ -17,6 +17,8 @@ export const REPLACE_SEND_TO_USER_TOOL_NAME = "replace_send_to_user"; export const SEND_TO_USER_PAYLOAD_LIKE = `%"toolName":"%${SEND_TO_USER_TOOL_NAME}"%`; /** One concise point per message: a longer send or rewrite is refused. */ export const SEND_TO_USER_MAX_BYTES = 1_000; +/** An ack only says that the agent got the message and is on it. */ +export const ACK_MAX_BYTES = 200; export const DASHBOARD_SET_TOOL_NAME = "dashboard_set"; export const DASHBOARD_GET_TOOL_NAME = "dashboard_get"; @@ -47,6 +49,7 @@ export type DisplayEvent = /** * `replyTo` is the short id of the user message that the send answers. Without it, the send is a * first message, and `pointer` is the short id that the send returned for it, when it returned one. + * `ack` marks a note that the agent got the message, sent before its answer. */ | { readonly type: "sent"; @@ -54,6 +57,7 @@ export type DisplayEvent = readonly text: string; readonly replyTo?: string | undefined; readonly pointer?: string | undefined; + readonly ack?: true; } /** * A completed `replace_send_to_user` call. With `replyTo`, the latest answer to that user message @@ -93,6 +97,10 @@ export interface VisibleCore { readonly shortId?: string | undefined; /** For an agent's reply, the short id of the user message that it answers. */ readonly replyTo?: string | undefined; + /** An ack that no rewrite has made the answer yet. */ + readonly ack?: true; + /** The id of the latest rewrite that changed the core's text. */ + readonly editedBy?: string; } /** Six hex digits of the FNV-1a hash of `text`. The agent names a message by this. */ @@ -129,8 +137,8 @@ export interface DisplayMode { /** Whether the thread has a Dashboard that the user sees and the agent writes. */ readonly dashboard: boolean; /** - * Whether the agent may send one message after each message that the user types. To say more, - * it rewrites that message with `replace_send_to_user`. + * Whether the agent may send one answer after each message that the user types, after at most + * one ack. To say more, it rewrites its latest message with `replace_send_to_user`. */ readonly oneSendPerMessage: boolean; /** @@ -167,8 +175,8 @@ export const messagePointer = (value: unknown): string | undefined => /** * The messages that `events` show, in order. Each `user` and `sent` event adds one; a `rewrite` * sets the text of the `sent` it names, or of the latest `sent` before it, and adds none. A rewrite - * whose send is not among the events, such as one on a history page not loaded yet, shows nothing - * and goes to `onUnmatched`. + * always leaves an answer, so it ends an ack. A rewrite whose send is not among the events, such as + * one on a history page not loaded yet, shows nothing and goes to `onUnmatched`. */ export function foldCores( events: Iterable, @@ -207,6 +215,7 @@ export function foldCores( first: event.id, reply: false, shortId: id, + ...(event.ack && { ack: event.ack }), }) - 1; agentFirstByShortId.set(id, latestSent); } else { @@ -219,6 +228,7 @@ export function foldCores( first: answered === undefined ? null : cores[answered]!.first, reply: true, replyTo: event.replyTo, + ...(event.ack && { ack: event.ack }), }) - 1; answerByReplyTo.set(event.replyTo, latestSent); } @@ -230,8 +240,13 @@ export function foldCores( : event.subThread !== undefined ? agentFirstByShortId.get(event.subThread) : latestSent; - if (target !== undefined) cores[target] = { ...cores[target]!, text: event.text }; - else onUnmatched?.(event); + if (target === undefined) { + onUnmatched?.(event); + break; + } + const { ack: _ack, ...core } = cores[target]!; + cores[target] = + core.text === event.text ? core : { ...core, text: event.text, editedBy: event.id }; break; } case "other": @@ -433,6 +448,10 @@ const MESSAGE_TOOLS: Readonly< const pointerField = (input: object, field: string) => messagePointer(Reflect.get(input, field)); +/** Whether a `send_to_user` call's input marks it as an ack. The only reader of the flag. */ +export const isAckInput = (input: unknown): boolean => + typeof input === "object" && input !== null && Reflect.get(input, "ack") === true; + /** A call that ended without reaching the user is `other`. A call that still runs shows nothing yet. */ function messageToolEvent( item: Extract, @@ -455,6 +474,7 @@ function messageToolEvent( text, replyTo, pointer: messagePointer(compactDynamicToolOutput(item.output)?.messageId), + ...(isAckInput(input) && { ack: true as const }), } : { type: "rewrite", id: item.id, text, replyTo, subThread: pointerField(input, "sub_thread") }; }