Let bots take in messages sent while they reply

A reply read the session once at the start, so a message sent while it
ran was never seen, and the follow-up reply queued for it then finished
silently: the message was saved before the reply, so the thread ended
with the bot. Those messages went unanswered.

The text loop now reads new user messages after every round of tools and
before an answer goes out, and the bot decides from what each one means
whether it changes the work (folded in, or the unsent answer redrafted)
or is a separate request, which it acknowledges at once with the new
send_message tool and does after the current work. Messages sent that
way are marked interim, so the thinking row stays up while the bot
works. A reply job records the newest user message it read, and a queued
reply only stays silent when a finished reply already read everything.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/bots-queue/has-unanswered-message.test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, test } from "bun:test";
+import hasUnansweredMessage from "./has-unanswered-message";
+
+const user = (id: number) => ({ id, author: "user" as const });
+const bot = (id: number) => ({ id, author: "bot" as const });
+
+describe("hasUnansweredMessage", () => {
+    test("a message sent while the last reply ran is answered, though the thread ends with the bot", () => {
+        /**
+         * A (10) was being answered when B (11) came in; the reply (12)
+         * was saved after B but its run never read it.
+         */
+        expect(hasUnansweredMessage({ history: [user(10), user(11), bot(12)], answered_id: 10 })).toBe(true);
+    });
+
+    test("a message the last reply took in is not answered twice", () => {
+        expect(hasUnansweredMessage({ history: [user(10), user(11), bot(12)], answered_id: 11 })).toBe(false);
+    });
+
+    test("a burst after the answer is answered", () => {
+        expect(
+            hasUnansweredMessage({ history: [user(10), bot(12), user(13), user(14)], answered_id: 10 }),
+        ).toBe(true);
+    });
+
+    test("without a record, the user having the last word decides", () => {
+        expect(hasUnansweredMessage({ history: [user(1)], answered_id: null })).toBe(true);
+        expect(hasUnansweredMessage({ history: [user(1), bot(2)], answered_id: null })).toBe(false);
+    });
+
+    test("nothing from the user, nothing to answer", () => {
+        expect(hasUnansweredMessage({ history: [], answered_id: null })).toBe(false);
+        expect(hasUnansweredMessage({ history: [bot(4)], answered_id: 2 })).toBe(false);
+    });
+});

src/bots-queue/has-unanswered-message.ts

@@ -0,0 +1,34 @@
+import type { BUN_SQLITE_COBBLE_BOT_MESSAGES } from "@/db/types/db";
+
+type Params = {
+    /**
+     * The session's latest messages, oldest first.
+     */
+    history: Pick<BUN_SQLITE_COBBLE_BOT_MESSAGES, "id" | "author">[];
+    /**
+     * The newest user message a finished reply answered
+     * (`grabAnsweredMessageId`), null when none is on record.
+     */
+    answered_id: number | null;
+};
+
+/**
+ * Whether a reply has anything left to answer. A user message that came in
+ * while the last reply ran is saved before that reply, so the thread ends
+ * with the bot's message even though the bot never read it: what the last
+ * reply read decides, not who wrote last. Without a record (no reply has
+ * finished in the session), the user having the last word does.
+ */
+export default function hasUnansweredMessage({ history, answered_id }: Params) {
+    const user_ids = history
+        .filter((message) => message.author == "user")
+        .map((message) => Number(message.id));
+
+    if (!user_ids.length) return false;
+
+    if (answered_id) {
+        return Math.max(...user_ids) > answered_id;
+    }
+
+    return history[history.length - 1]?.author == "user";
+}

src/bots-queue/run-bot-reply.ts

@@ -28,6 +28,12 @@ import jobSessionId from "@/src/functions/backend/db/bots-queue/job-session-id";
 import loadAttachmentParts from "./load-attachment-parts";
 import grabMessageRefNames from "@/src/functions/backend/db/bots/grab-message-ref-names";
 import withTakeOverRefs from "./with-take-over-refs";
+import hasUnansweredMessage from "./has-unanswered-message";
+import grabAnsweredMessageId from "@/src/functions/backend/db/bots-queue/grab-answered-message-id";
+import markReplyAnswered from "@/src/functions/backend/db/bots-queue/mark-reply-answered";
+import sendMessageTool from "@/src/harness/tools/bot/send-message-tool";
+import type { BotMessageWithAttachments } from "@/src/types";
+import type { HarnessMessage } from "@/src/harness/types";
 import type { BotJobContext, BotJobOutcome } from "./types";
 
 type Params = BotJobContext & {
@@ -47,8 +53,12 @@ type Params = BotJobContext & {
  * instructions, start_task, and for the coordinator delegate_task) plus
  * its MCP servers' and custom tools (`openBotExtensions`, script tools on
  * its current computer).
- * Finishes silently when the last message is already the bot's, i.e. an
- * earlier reply answered everything.
+ * Messages the user sends while it works join the run (`incoming`): one
+ * that changes the work changes the answer before it goes out, and one
+ * that's separate can be acknowledged at once (`send_message`) and done
+ * after the current work. The job records the newest
+ * one it read (`markReplyAnswered`). Finishes silently when an earlier
+ * reply already read everything (`hasUnansweredMessage`).
  */
 export default async function runBotReply({
     bot,
@@ -65,10 +75,48 @@ export default async function runBotReply({
         limit: AppData.BotReplyHistoryMessages,
     });
 
-    if (history[history.length - 1]?.author != "user") {
+    const answered_id = await grabAnsweredMessageId({ session_id });
+
+    if (!hasUnansweredMessage({ history, answered_id })) {
         return { result: "", silent: true };
     }
 
+    /**
+     * The newest message the run has read, and of those the newest user
+     * message, which is what the reply answers.
+     */
+    let seen_id = Math.max(0, ...history.map((message) => Number(message.id)));
+    let answering_id = newestUserMessageId({ messages: history });
+
+    const toTurns = async ({ messages }: { messages: BotMessageWithAttachments[] }) =>
+        botMessagesToHarness({
+            messages,
+            parts: await loadAttachmentParts({ messages }),
+            refs: await grabMessageRefNames({
+                texts: messages.filter((m) => m.author == "user").map((m) => m.text || ""),
+                user_id,
+            }),
+        });
+
+    /**
+     * What the user wrote since the run last looked.
+     */
+    const incoming = async (): Promise<HarnessMessage[]> => {
+        const fresh = await grabBotMessages({
+            session_id,
+            after_id: seen_id,
+            author: "user",
+            limit: AppData.BotReplyHistoryMessages,
+        });
+
+        if (!fresh.length) return [];
+
+        seen_id = Math.max(seen_id, ...fresh.map((message) => Number(message.id)));
+        answering_id = Math.max(answering_id, newestUserMessageId({ messages: fresh }));
+
+        return await toTurns({ messages: fresh });
+    };
+
     const prompt = await grabBotPrompt({ bot_id: bot.id, user_id, workspace_id });
     const computers = await grabBotComputers({ bot_id: bot.id, workspace_id });
 
@@ -165,22 +213,26 @@ export default async function runBotReply({
             ]
                 .filter(Boolean)
                 .join("\n\n"),
-            messages: botMessagesToHarness({
-                messages: history,
-                parts: await loadAttachmentParts({ messages: history }),
-                refs: await grabMessageRefNames({
-                    texts: history.filter((m) => m.author == "user").map((m) => m.text || ""),
-                    user_id,
-                }),
-            }),
-            tools: [...tools, ...file_tools, ...extensions.tools, ...skills.tools],
+            messages: await toTurns({ messages: history }),
+            tools: [
+                ...tools,
+                sendMessageTool({ bot_id: bot.id, session_id: Number(session_id), workspace_id }),
+                ...file_tools,
+                ...extensions.tools,
+                ...skills.tools,
+            ],
             max_steps: await grabReplyMaxToolSteps(),
             bot_id: bot.id,
             workspace_id,
             signal,
             onStep: report,
+            incoming,
         });
 
+        if (job.id && answering_id) {
+            await markReplyAnswered({ job_id: job.id, message_id: answering_id });
+        }
+
         return {
             ...withTakeOverRefs({ text: result.text, refs: take_overs }),
             attachments: attachments.length ? attachments : undefined,
@@ -189,3 +241,10 @@ export default async function runBotReply({
         await extensions.close();
     }
 }
+
+function newestUserMessageId({ messages }: { messages: Pick<BotMessageWithAttachments, "id" | "author">[] }) {
+    return Math.max(
+        0,
+        ...messages.filter((message) => message.author == "user").map((message) => Number(message.id)),
+    );
+}