Show a reply's steps, and catch up on a reply whose live push was missed
A reply that works through tools now shows its latest tool step under
"is thinking…" (not its closing note, which is the answer). While a reply
is due, the open conversation checks the session's latest messages every
few seconds (api/admin/bots/session-messages) and replays any it missed
as server:bot-message, so the thread, rail and unread all take it and the
thinking row can't stay up until a reload. MCP and custom tool steps read
in words ("Using Exa search (web search exa)").
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -48,10 +48,21 @@ describe("describeJobStep", () => {
|
||||
tool_calls: [
|
||||
{ id: "a", name: "browser_navigate", arguments: "{}" },
|
||||
{ id: "b", name: "browser_click", arguments: "{}" },
|
||||
{ id: "c", name: "some_mcp_tool", arguments: "{}" },
|
||||
{ id: "c", name: "get_weather", arguments: "{}" },
|
||||
],
|
||||
})?.text,
|
||||
).toBe("Using the browser, Using some_mcp_tool");
|
||||
).toBe("Using the browser, Using get weather");
|
||||
});
|
||||
|
||||
test("names an MCP server's tool in words", () => {
|
||||
expect(
|
||||
describeJobStep({
|
||||
index: 0,
|
||||
text: null,
|
||||
candidate,
|
||||
tool_calls: [{ id: "a", name: "exa_search__web_search_exa", arguments: "{}" }],
|
||||
})?.text,
|
||||
).toBe("Using Exa search (web search exa)");
|
||||
});
|
||||
|
||||
test("keeps the first line of a text-only turn, cut to one line", () => {
|
||||
|
||||
@@ -57,5 +57,21 @@ function firstLine({ text }: { text: string | null }) {
|
||||
function toolLabel({ name }: { name: string }) {
|
||||
const prefix = Object.keys(JobStepToolPrefixLabels).find((p) => name.startsWith(p));
|
||||
|
||||
return JobStepToolLabels[name] || (prefix && JobStepToolPrefixLabels[prefix]) || `Using ${name}`;
|
||||
return JobStepToolLabels[name] || (prefix && JobStepToolPrefixLabels[prefix]) || `Using ${readableToolName({ name })}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* A tool name in words: `get_weather` reads "get weather", and an MCP
|
||||
* server's `<slug>__<tool>` reads "Exa search (web search exa)".
|
||||
*/
|
||||
function readableToolName({ name }: { name: string }) {
|
||||
const words = (part: string) => part.replace(/_+/g, " ").trim();
|
||||
const [server, ...rest] = name.split("__");
|
||||
const tool = words(rest.join(" "));
|
||||
|
||||
if (!tool) return words(name);
|
||||
|
||||
const server_words = words(server || "");
|
||||
|
||||
return `${server_words.charAt(0).toUpperCase()}${server_words.slice(1)} (${tool})`;
|
||||
}
|
||||
|
||||
@@ -213,6 +213,16 @@ export const AppData = {
|
||||
* bot changes from the database for its own sockets (`liveRelay`).
|
||||
*/
|
||||
LiveRelayPollMs: 1500,
|
||||
/**
|
||||
* While a reply is due in the open conversation, how often the page
|
||||
* checks for new messages itself, in case a live push was missed (the
|
||||
* socket was down) and the thinking row would otherwise stay up.
|
||||
*/
|
||||
BotReplyCatchUpMs: 5000,
|
||||
/**
|
||||
* How many of a session's latest messages that check reads.
|
||||
*/
|
||||
BotReplyCatchUpMessages: 20,
|
||||
/**
|
||||
* How much of the conversation a bot sees when it replies.
|
||||
*/
|
||||
|
||||
@@ -5,7 +5,7 @@ type Params = {
|
||||
/**
|
||||
* Under `/api/admin/bots/`.
|
||||
*/
|
||||
path: "list-sessions" | "new-session" | "reopen-session";
|
||||
path: "list-sessions" | "new-session" | "reopen-session" | "session-messages";
|
||||
body: Record<string, unknown>;
|
||||
};
|
||||
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import { useEffect, useRef } from "react";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import type { BotMessageWithAttachments, WebSocketDataType } from "@/src/types";
|
||||
import postSessionApi from "../(functions)/post-session-api";
|
||||
|
||||
type Params = {
|
||||
session_id?: number | null;
|
||||
/**
|
||||
* The thread's last message is the user's, so a reply is due.
|
||||
*/
|
||||
awaiting_reply: boolean;
|
||||
/**
|
||||
* The thread's last message: only newer ones are replayed.
|
||||
*/
|
||||
last_message_id?: number | null;
|
||||
};
|
||||
|
||||
/**
|
||||
* While a reply is due, checks the session's latest messages every
|
||||
* `AppData.BotReplyCatchUpMs`. Replies arrive live; this only catches one
|
||||
* whose push was missed (the socket was down at the time), so the
|
||||
* thinking row can't stay up until a reload. A missed message is replayed
|
||||
* as the `server:bot-message` it would have been, so the thread, the rail
|
||||
* and the unread flag all take it the usual way.
|
||||
*/
|
||||
export default function useReplyCatchUp({ session_id, awaiting_reply, last_message_id }: Params) {
|
||||
const last_message = useRef(Number(last_message_id) || 0);
|
||||
|
||||
last_message.current = Number(last_message_id) || 0;
|
||||
|
||||
useEffect(() => {
|
||||
if (!awaiting_reply || !session_id) return;
|
||||
|
||||
let stale = false;
|
||||
|
||||
const timer = setInterval(async () => {
|
||||
const res = await postSessionApi<BotMessageWithAttachments>({
|
||||
path: "session-messages",
|
||||
body: { session_id },
|
||||
}).catch(() => null);
|
||||
|
||||
if (stale || !res?.success || !Array.isArray(res.payload)) return;
|
||||
|
||||
const missed = (res.payload as BotMessageWithAttachments[]).filter(
|
||||
(message) => Number(message.id) > last_message.current,
|
||||
);
|
||||
|
||||
for (const message of missed) {
|
||||
const data: WebSocketDataType = { event: "server:bot-message", data: message };
|
||||
|
||||
window.dispatchEvent(new CustomEvent("wsDataEvent", { detail: { data, message: data } }));
|
||||
}
|
||||
}, AppData.BotReplyCatchUpMs);
|
||||
|
||||
return () => {
|
||||
stale = true;
|
||||
clearInterval(timer);
|
||||
};
|
||||
}, [awaiting_reply, session_id]);
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
import { useRef, useState } from "react";
|
||||
import useServerEvent from "@/src/hooks/use-server-event";
|
||||
import type { BotJobUpdateData } from "@/src/types";
|
||||
|
||||
type Params = {
|
||||
bot_id: number;
|
||||
/**
|
||||
* The open session: steps of replies in other sessions are left out.
|
||||
*/
|
||||
session_id?: number | null;
|
||||
/**
|
||||
* The thread's last message. A step belongs to the reply due after it,
|
||||
* so a new message (the reply itself, or another from the user) hides
|
||||
* it.
|
||||
*/
|
||||
last_message_id?: number | null;
|
||||
};
|
||||
|
||||
type Step = {
|
||||
text: string;
|
||||
after_message_id: number | null;
|
||||
};
|
||||
|
||||
/**
|
||||
* The latest step of the reply running in the open session (a
|
||||
* `bots_queue_updates` row's text, from `server:bot-job-update` without a
|
||||
* task), so a reply that works through tools shows what it's doing. Only
|
||||
* tool steps: the closing note is the answer itself.
|
||||
*/
|
||||
export default function useReplyProgress({ bot_id, session_id, last_message_id }: Params) {
|
||||
const [step, setStep] = useState<Step | null>(null);
|
||||
const last_message = useRef(last_message_id || null);
|
||||
|
||||
last_message.current = last_message_id || null;
|
||||
|
||||
useServerEvent({
|
||||
event: "server:bot-job-update",
|
||||
onData: (update: BotJobUpdateData) => {
|
||||
if (update.bot_id != bot_id || update.task_id || !update.text) return;
|
||||
|
||||
/**
|
||||
* A turn without tools ends the reply: its text is the answer,
|
||||
* about to land as the message.
|
||||
*/
|
||||
if (update.kind == "note") return;
|
||||
if (!session_id || update.session_id != session_id) return;
|
||||
|
||||
setStep({ text: update.text, after_message_id: last_message.current });
|
||||
},
|
||||
});
|
||||
|
||||
return step && step.after_message_id == (last_message_id || null) ? step.text : null;
|
||||
}
|
||||
@@ -16,6 +16,10 @@ type Props = {
|
||||
* The latest step of the bot's active task.
|
||||
*/
|
||||
progress?: string | null;
|
||||
/**
|
||||
* The latest step of the reply running in this conversation.
|
||||
*/
|
||||
reply_progress?: string | null;
|
||||
/**
|
||||
* The thread's last message is the user's, so a reply is due here.
|
||||
*/
|
||||
@@ -29,13 +33,13 @@ type Props = {
|
||||
/**
|
||||
* Under the last message while the bot is working on a reply or a task
|
||||
* (or waiting for a model or its computer), so slow work doesn't look like
|
||||
* none. A task shows its latest step. The bot's face is one per user,
|
||||
* none. A task or a reply shows its latest step. The bot's face is one per user,
|
||||
* shared by all their sessions with it, so a reply shows from the thread
|
||||
* alone: every user message queues one, so while the user has the last
|
||||
* word a reply is due here, even while the bot works elsewhere. Only the
|
||||
* waits (for a model, for its computer) come from the face.
|
||||
*/
|
||||
export default function BotThinkingRow({ bot, progress, awaiting_reply, has_task }: Props) {
|
||||
export default function BotThinkingRow({ bot, progress, reply_progress, awaiting_reply, has_task }: Props) {
|
||||
const name = bot.name || "The bot";
|
||||
const asleep = bot.face == "asleep";
|
||||
const waiting_model = asleep && bot.status_line == BotWaitingStatusLine && Boolean(awaiting_reply || has_task);
|
||||
@@ -53,7 +57,7 @@ export default function BotThinkingRow({ bot, progress, awaiting_reply, has_task
|
||||
: working
|
||||
? `${name} is working…`
|
||||
: `${bot.status_line}…`;
|
||||
const detail = working ? progress : null;
|
||||
const detail = working ? progress : thinking ? reply_progress : null;
|
||||
|
||||
return (
|
||||
<Row role="status" className="w-full gap-3! flex-nowrap items-center">
|
||||
|
||||
@@ -14,6 +14,10 @@ type Props = {
|
||||
* The latest step of the bot's active task, shown while it works.
|
||||
*/
|
||||
progress?: string | null;
|
||||
/**
|
||||
* The latest step of the reply running here, shown while it thinks.
|
||||
*/
|
||||
reply_progress?: string | null;
|
||||
/**
|
||||
* This session has an active task.
|
||||
*/
|
||||
@@ -26,7 +30,7 @@ type Props = {
|
||||
* While the bot works on a reply or a task, a status row sits under the
|
||||
* last one.
|
||||
*/
|
||||
export default function MessageThread({ bot, messages, progress, has_task }: Props) {
|
||||
export default function MessageThread({ bot, messages, progress, reply_progress, has_task }: Props) {
|
||||
const groups = groupMessages({ messages });
|
||||
|
||||
return (
|
||||
@@ -47,6 +51,7 @@ export default function MessageThread({ bot, messages, progress, has_task }: Pro
|
||||
<BotThinkingRow
|
||||
bot={bot}
|
||||
progress={progress}
|
||||
reply_progress={reply_progress}
|
||||
awaiting_reply={messages[messages.length - 1]?.author == "user"}
|
||||
has_task={has_task}
|
||||
/>
|
||||
|
||||
@@ -8,6 +8,8 @@ import useServerEvent from "@/src/hooks/use-server-event";
|
||||
import { BotComputersContext } from "../(data)/bot-computers-context";
|
||||
import useBotComputers from "../(hooks)/use-bot-computers";
|
||||
import useTaskProgress from "../(hooks)/use-task-progress";
|
||||
import useReplyProgress from "../(hooks)/use-reply-progress";
|
||||
import useReplyCatchUp from "../(hooks)/use-reply-catch-up";
|
||||
import ConversationHeader from "../(partials)/conversation-header";
|
||||
import MessageThread from "../(partials)/message-thread";
|
||||
import Composer from "../(partials)/composer";
|
||||
@@ -51,6 +53,20 @@ export default function Conversation() {
|
||||
task: pageProps?.bot_task || null,
|
||||
});
|
||||
|
||||
const reply_progress = useReplyProgress({
|
||||
bot_id: bot?.id || 0,
|
||||
session_id,
|
||||
last_message_id: messages[messages.length - 1]?.id || null,
|
||||
});
|
||||
|
||||
const awaiting_reply = messages[messages.length - 1]?.author == "user";
|
||||
|
||||
useReplyCatchUp({
|
||||
session_id,
|
||||
awaiting_reply,
|
||||
last_message_id: messages[messages.length - 1]?.id || null,
|
||||
});
|
||||
|
||||
/**
|
||||
* A message that starts a newer session (sent here or in another tab)
|
||||
* moves the thread onto it.
|
||||
@@ -107,6 +123,7 @@ export default function Conversation() {
|
||||
bot={live_bot}
|
||||
messages={messages}
|
||||
progress={progress.text}
|
||||
reply_progress={reply_progress}
|
||||
has_task={Boolean(progress.task_id)}
|
||||
/>
|
||||
<Composer
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import userAuth from "@/src/functions/backend/auth/user-auth";
|
||||
import grabUserBotSession from "@/src/functions/backend/db/bot-sessions/grab-user-bot-session";
|
||||
import grabBotMessages from "@/src/functions/backend/db/bots/grab-bot-messages";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import { BotSessionIdRequestSchema } from "@/src/schemas/bot-schema";
|
||||
import type { BotMessageWithAttachments } from "@/src/types";
|
||||
import AppError from "@/src/utils/app-error";
|
||||
import parseWithSchema from "@/src/utils/parse-with-schema";
|
||||
import safeErrorMessage from "@/src/utils/safe-error-message";
|
||||
import type {
|
||||
APIResponseObject,
|
||||
BunextAPIRouteHandler,
|
||||
} from "@moduletrace/bunext/types";
|
||||
|
||||
/**
|
||||
* The latest messages of one of the session user's own sessions, so the
|
||||
* open conversation can catch up on a reply whose live push it missed.
|
||||
*/
|
||||
export const handler: BunextAPIRouteHandler<
|
||||
APIResponseObject<BotMessageWithAttachments>
|
||||
> = async ({ req, body }) => {
|
||||
try {
|
||||
if (req.method != "POST") {
|
||||
throw new AppError(`Method not allowed`);
|
||||
}
|
||||
|
||||
const { user } = await userAuth({ req });
|
||||
|
||||
if (!user?.logged_in_status || !user.id) {
|
||||
return { success: false, msg: `Unauthorized`, logoutUser: true };
|
||||
}
|
||||
|
||||
const { session_id } = parseWithSchema({ schema: BotSessionIdRequestSchema, data: body });
|
||||
|
||||
const session = await grabUserBotSession({ session_id, user_id: user.id });
|
||||
|
||||
if (!session?.id) {
|
||||
throw new AppError(`Session not found.`);
|
||||
}
|
||||
|
||||
return {
|
||||
success: true,
|
||||
payload: await grabBotMessages({
|
||||
session_id: session.id,
|
||||
limit: AppData.BotReplyCatchUpMessages,
|
||||
}),
|
||||
};
|
||||
} catch (error: unknown) {
|
||||
return { success: false, msg: safeErrorMessage({ error }) };
|
||||
}
|
||||
};
|
||||
Reference in New Issue
Block a user