ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-19 17:11:42 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ui): send to the selected session; delete exact-run steer machinery (#125862)

Browse Source 
browser no longer chooses or persists run targets; steer sends ride the generic durable outbox with queueMode steer and no expectedRunId/expectedLeafEntryId; deletes steer-lifecycle.ts, steered-chip.ts, kind:"steered", steerTargetRunId, sendState:"steering", target-bound retry, uniqueness failure rows; persisted outbox rows normalize at load; net -625 production LOC.
This commit is contained in:
Peter Steinberger
2026-08-18 08:38:23 -07:00
committed by GitHub
parent 8dd0434f86
commit d84a910fc8
38 changed files with 500 additions and 3182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
-1
View File
@@ -4178,7 +4178,6 @@ ui/src/pages/chat/chat-progress.ts 2
ui/src/pages/chat/chat-queue.ts 1
ui/src/pages/chat/chat-realtime.ts 1
ui/src/pages/chat/chat-send-ack.ts 2
ui/src/pages/chat/chat-send-actions.ts 2
ui/src/pages/chat/chat-send-composer.ts 1
ui/src/pages/chat/chat-send-request.ts 1
ui/src/pages/chat/chat-send-timing.ts 6
ui/src/e2e/chat-flow.active-run-follow-ups.e2e.test.ts
+5 -425
View File
@@ -1,4 +1,3 @@
import type { Page } from "playwright";
import { expect, it } from "vitest";
import {
chatSessionListResponse,
@@ -6,315 +5,12 @@ import {
expectRequestCountStable,
installMockGateway,
requireRecord,
requireString,
waitForRequests,
} from "./chat-flow.test-support.ts";
const suite = createChatFlowE2eSuite();
async function expectChatBubbleAbove(page: Page, upperText: string, lowerText: string) {
const thread = page.locator(".chat-thread-inner");
await expect
.poll(() =>
thread.evaluate(
(element, texts) => {
const bubbles = Array.from(element.querySelectorAll<HTMLElement>(".chat-bubble"));
const matches = texts.map((text) =>
bubbles.filter((bubble) => bubble.textContent?.includes(text)),
);
const counts = matches.map((matchingBubbles) => matchingBubbles.length);
const upperBubble = matches[0]?.[0];
const lowerBubble = matches[1]?.[0];
if (counts.some((count) => count !== 1) || !upperBubble || !lowerBubble) {
return { counts, lowerTop: null, ordered: false, upperTop: null };
}
const upperTop = upperBubble.getBoundingClientRect().top;
const lowerTop = lowerBubble.getBoundingClientRect().top;
return { counts, lowerTop, ordered: upperTop < lowerTop, upperTop };
},
[upperText, lowerText],
),
)
.toEqual({
counts: [1, 1],
lowerTop: expect.any(Number),
ordered: true,
upperTop: expect.any(Number),
});
}
suite.define(() => {
it("keeps cumulative assistant output ordered across a consumed steer", async () => {
const context = await suite.newBrowserContext({
locale: "en-US",
serviceWorkers: "block",
viewport: { height: 900, width: 1280 },
});
const page = await context.newPage();
const runtimeConfig = {
messages: { queue: { byChannel: { webchat: "steer" }, mode: "followup" } },
};
const gateway = await installMockGateway(page, {
methodResponses: {
"config.get": {
config: runtimeConfig,
hash: "queue-steer-config",
issues: [],
raw: JSON.stringify(runtimeConfig),
runtimeConfig,
valid: true,
},
},
});
try {
await page.goto(`${suite.server.baseUrl}chat`);
const originalPrompt = "keep this run active";
await page.locator(".agent-chat__composer-combobox textarea").fill(originalPrompt);
await page.getByRole("button", { name: "Send message" }).click();
const initialSend = await gateway.waitForRequest("chat.send");
const activeRunId = requireString(
requireRecord(initialSend.params).idempotencyKey,
"active chat run id",
);
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [activeRunId],
clientRunId: activeRunId,
hasActiveRun: true,
message: {
__openclaw: {
id: "persisted-original-user",
idempotencyKey: `${activeRunId}:user`,
seq: 1,
},
content: [{ text: originalPrompt, type: "text" }],
role: "user",
timestamp: Date.now(),
},
messageId: "persisted-original-user",
messageSeq: 1,
session: {
activeRunIds: [activeRunId],
hasActiveRun: true,
key: "main",
kind: "direct",
status: "running",
updatedAt: Date.now(),
},
sessionKey: "main",
});
await page.getByRole("button", { name: "Stop generating" }).waitFor({ timeout: 10_000 });
const preSteerReply = "Assistant output before the steer.";
await gateway.emitGatewayEvent("chat", {
deltaText: preSteerReply,
message: {
content: [{ text: preSteerReply, type: "text" }],
role: "assistant",
timestamp: Date.now(),
},
runId: activeRunId,
sessionKey: "main",
state: "delta",
});
await page.getByText(preSteerReply, { exact: true }).waitFor({ timeout: 10_000 });
const queuedFollowUp = "queued follow-up before steer";
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [activeRunId],
clientRunId: "queued-run",
hasActiveRun: true,
message: {
__openclaw: {
id: "persisted-queued-user",
idempotencyKey: "queued-run:user",
seq: 3,
},
content: [{ text: queuedFollowUp, type: "text" }],
role: "user",
timestamp: Date.now(),
},
messageId: "persisted-queued-user",
messageSeq: 3,
session: {
activeRunIds: [activeRunId],
hasActiveRun: true,
key: "main",
kind: "direct",
status: "running",
updatedAt: Date.now(),
},
sessionKey: "main",
});
await page.getByText(queuedFollowUp, { exact: true }).waitFor({ timeout: 10_000 });
await expectChatBubbleAbove(page, preSteerReply, queuedFollowUp);
const followUp = "tighten the active plan";
await page.locator(".agent-chat__composer-combobox textarea").fill(followUp);
await page.getByRole("button", { name: "Steer into the active run" }).click();
const sends = await waitForRequests(gateway, "chat.send", 2);
const steerParams = requireRecord(sends[1]?.params);
expect(steerParams).toMatchObject({
deliver: false,
message: followUp,
sessionKey: "main",
});
const steerRunId = requireString(steerParams.idempotencyKey, "steer run id");
const queue = page.locator(".chat-queue");
await queue.locator(".chat-queue__badge--steered", { hasText: "Steering" }).waitFor({
timeout: 10_000,
});
await queue.getByText(followUp).waitFor({ timeout: 10_000 });
await gateway.emitGatewayEvent("chat", {
runId: steerRunId,
sessionKey: "main",
state: "final",
});
await queue.getByText(followUp).waitFor({ state: "detached", timeout: 10_000 });
await expect
.poll(() => page.locator(".chat-thread .chat-group.user", { hasText: followUp }).count())
.toBe(1);
await expectChatBubbleAbove(page, originalPrompt, preSteerReply);
await expectChatBubbleAbove(page, preSteerReply, queuedFollowUp);
await expectChatBubbleAbove(page, queuedFollowUp, followUp);
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [activeRunId],
clientRunId: activeRunId,
hasActiveRun: true,
message: {
__openclaw: {
id: "persisted-steer-user",
idempotencyKey: `${steerRunId}:user`,
seq: 4,
steerTargetRunId: activeRunId,
},
content: [{ text: followUp, type: "text" }],
role: "user",
timestamp: Date.now(),
},
messageId: "persisted-steer-user",
messageSeq: 4,
session: {
activeRunIds: [activeRunId],
hasActiveRun: true,
key: "main",
kind: "direct",
status: "running",
updatedAt: Date.now(),
},
sessionKey: "main",
});
await expect
.poll(() => page.locator(".chat-thread .chat-group.user", { hasText: followUp }).count())
.toBe(1);
const postSteerReply = "Assistant output after the steer.";
const cumulativeReply = `${preSteerReply} ${postSteerReply}`;
const terminalPostSteerReply = `${postSteerReply} Final unseen suffix.`;
const terminalReply = `${preSteerReply} ${terminalPostSteerReply}`;
await gateway.emitGatewayEvent("chat", {
deltaText: ` ${postSteerReply}`,
message: {
content: [{ text: cumulativeReply, type: "text" }],
role: "assistant",
timestamp: Date.now(),
},
runId: activeRunId,
sessionKey: "main",
state: "delta",
});
await page.locator(".chat-bubble", { hasText: postSteerReply }).waitFor({ timeout: 10_000 });
await expectChatBubbleAbove(page, originalPrompt, preSteerReply);
await expectChatBubbleAbove(page, preSteerReply, queuedFollowUp);
await expectChatBubbleAbove(page, queuedFollowUp, followUp);
await expectChatBubbleAbove(page, followUp, postSteerReply);
const authoritativeMessages = [
{
__openclaw: {
id: "persisted-original-user",
idempotencyKey: `${activeRunId}:user`,
seq: 1,
},
content: [{ text: originalPrompt, type: "text" }],
role: "user",
timestamp: 100,
},
{
__openclaw: { id: "persisted-pre-steer", idempotencyKey: activeRunId, seq: 2 },
content: [{ text: preSteerReply, type: "text" }],
role: "assistant",
timestamp: 200,
},
{
__openclaw: {
id: "persisted-queued-user",
idempotencyKey: "queued-run:user",
seq: 3,
},
content: [{ text: queuedFollowUp, type: "text" }],
role: "user",
timestamp: 250,
},
{
__openclaw: {
id: "persisted-steer-user",
idempotencyKey: `${steerRunId}:user`,
seq: 4,
steerTargetRunId: activeRunId,
},
content: [{ text: followUp, type: "text" }],
role: "user",
timestamp: 50,
},
{
__openclaw: { id: "persisted-post-steer", idempotencyKey: activeRunId, seq: 5 },
content: [{ text: terminalPostSteerReply, type: "text" }],
role: "assistant",
timestamp: 300,
},
];
const terminalHistory = {
messages: authoritativeMessages,
sessionId: "control-ui-e2e-session",
sessionInfo: {
activeRunIds: [],
hasActiveRun: false,
key: "main",
kind: "direct",
status: "done",
updatedAt: Date.now(),
},
thinkingLevel: null,
};
await gateway.setMethodResponse("chat.history", terminalHistory);
await gateway.setMethodResponse("chat.startup", terminalHistory);
await gateway.emitChatFinal({ runId: activeRunId, text: terminalReply });
await page
.getByRole("button", { name: "Stop generating" })
.waitFor({ state: "detached", timeout: 10_000 });
await page
.locator(".chat-bubble", { hasText: terminalPostSteerReply })
.waitFor({ timeout: 10_000 });
await expectChatBubbleAbove(page, preSteerReply, queuedFollowUp);
await expectChatBubbleAbove(page, queuedFollowUp, followUp);
await expectChatBubbleAbove(page, followUp, postSteerReply);
await page.reload();
await page
.locator(".chat-bubble", { hasText: terminalPostSteerReply })
.waitFor({ timeout: 10_000 });
await expectChatBubbleAbove(page, originalPrompt, preSteerReply);
await expectChatBubbleAbove(page, preSteerReply, queuedFollowUp);
await expectChatBubbleAbove(page, queuedFollowUp, followUp);
await expectChatBubbleAbove(page, followUp, postSteerReply);
} finally {
await suite.closeBrowserContext(context);
}
});
it("preserves a non-steer server default for active-run follow-ups", async () => {
const context = await suite.newBrowserContext({
locale: "en-US",
@@ -419,8 +115,7 @@ suite.define(() => {
const composer = page.locator(".agent-chat__composer-combobox textarea");
await composer.fill("keep the first shortcut run active");
await page.getByRole("button", { name: "Send message" }).click();
const firstSend = requireRecord((await gateway.waitForRequest("chat.send")).params);
const firstRunId = requireString(firstSend.idempotencyKey, "first active run id");
await gateway.waitForRequest("chat.send");
await page.getByRole("button", { name: "Stop generating" }).waitFor({ timeout: 10_000 });
const steerText = "steer this keyboard follow-up now";
@@ -431,13 +126,12 @@ suite.define(() => {
const steerParams = requireRecord(firstRunSends[1]?.params);
expect(steerParams).toMatchObject({
deliver: false,
expectedRunId: firstRunId,
message: steerText,
queueMode: "steer",
sessionKey: "main",
});
const steeredRow = page.locator(".chat-queue__item--steered", { hasText: steerText });
await steeredRow.waitFor({ timeout: 10_000 });
expect(steerParams).not.toHaveProperty("expectedRunId");
expect(steerParams).not.toHaveProperty("expectedLeafEntryId");
} finally {
await suite.closeBrowserContext(context);
}
@@ -542,112 +236,6 @@ suite.define(() => {
}
});
it("dismisses an informational steer notice when the steer request lands", async () => {
const context = await suite.newBrowserContext({
locale: "en-US",
serviceWorkers: "block",
viewport: { height: 900, width: 1280 },
});
const page = await context.newPage();
const gateway = await installMockGateway(page);
try {
await page.goto(`${suite.server.baseUrl}settings/appearance`);
await page.locator("[data-settings-follow-up-mode]").selectOption("queue");
await page.goto(`${suite.server.baseUrl}chat?session=main`);
const originalPrompt = "keep this run active";
await page.locator(".agent-chat__composer-combobox textarea").fill(originalPrompt);
await page.getByRole("button", { name: "Send message" }).click();
const activeRequest = await gateway.waitForRequest("chat.send");
const activeRunId = requireString(
requireRecord(activeRequest.params).idempotencyKey,
"active run idempotency key",
);
await gateway.emitGatewayEvent("session.message", {
activeRunIds: [activeRunId],
clientRunId: activeRunId,
hasActiveRun: true,
message: {
__openclaw: {
id: "persisted-notice-original-user",
idempotencyKey: `${activeRunId}:user`,
seq: 1,
},
content: [{ text: originalPrompt, type: "text" }],
role: "user",
timestamp: Date.now(),
},
messageId: "persisted-notice-original-user",
messageSeq: 1,
session: {
activeRunIds: [activeRunId],
hasActiveRun: true,
key: "main",
kind: "direct",
status: "running",
updatedAt: Date.now(),
},
sessionKey: "main",
});
await page.getByRole("button", { name: "Stop generating" }).waitFor({ timeout: 10_000 });
const steerText = "route this into the active run";
await page.locator(".agent-chat__composer-combobox textarea").fill(steerText);
await page.getByRole("button", { name: "Queue message" }).click();
const queue = page.locator(".chat-queue");
await queue.getByText(steerText).waitFor({ timeout: 10_000 });
await queue.getByRole("button", { name: "Steer" }).click();
const sends = await waitForRequests(gateway, "chat.send", 2);
const steerParams = requireRecord(sends[1]?.params);
expect(steerParams).toMatchObject({
expectedRunId: activeRunId,
message: steerText,
queueMode: "steer",
});
const steerRunId = requireString(steerParams.idempotencyKey, "steer idempotency key");
const row = queue.locator(".chat-queue__item--steered", { hasText: steerText });
await row.waitFor({ timeout: 10_000 });
const pendingPresentation = await row.evaluate((element) => {
const badge = element.querySelector<HTMLElement>(".chat-queue__badge--steered");
const icon = element.querySelector(".chat-queue__icon");
const probe = document.createElement("span");
probe.style.color = "var(--info)";
probe.style.background = "var(--info-subtle)";
document.body.append(probe);
const probeStyle = getComputedStyle(probe);
const infoColor = probeStyle.color;
const infoSubtle = probeStyle.backgroundColor;
probe.remove();
return {
badgeColor: badge ? getComputedStyle(badge).color : "",
backgroundColor: getComputedStyle(element).backgroundColor,
iconPoints: icon?.querySelector("polyline")?.getAttribute("points") ?? "",
infoColor,
infoSubtle,
};
});
await gateway.emitGatewayEvent("chat", {
runId: steerRunId,
sessionKey: "main",
state: "final",
});
await row.waitFor({ state: "detached", timeout: 10_000 });
await page.getByText(steerText, { exact: true }).waitFor({ timeout: 10_000 });
await expectChatBubbleAbove(page, "keep this run active", steerText);
expect(pendingPresentation).toMatchObject({
badgeColor: pendingPresentation.infoColor,
backgroundColor: pendingPresentation.infoSubtle,
iconPoints: "15 10 20 15 15 20",
});
await page.getByRole("button", { name: "Stop generating" }).waitFor({ timeout: 10_000 });
} finally {
await suite.closeBrowserContext(context);
}
});
it("steers a restored queued message when only the session row reports the active run", async () => {
const context = await suite.newBrowserContext({
locale: "en-US",
@@ -707,20 +295,12 @@ suite.define(() => {
const steerParams = requireRecord(steerRequest.params);
expect(steerParams).toMatchObject({
deliver: false,
expectedLeafEntryId: "leaf-active",
expectedRunId: "active-run",
message: queuedPrompt,
queueMode: "steer",
sessionKey: "main",
});
await queue.locator(".chat-queue__badge--steered", { hasText: "Steering" }).waitFor({
timeout: 10_000,
});
await gateway.emitChatFinal({
runId: requireString(steerParams.idempotencyKey, "restored steer idempotency key"),
text: "Restored steer completed.",
});
await queue.getByText(queuedPrompt).waitFor({ state: "detached", timeout: 10_000 });
expect(steerParams).not.toHaveProperty("expectedRunId");
expect(steerParams).not.toHaveProperty("expectedLeafEntryId");
} finally {
await suite.closeBrowserContext(context);
}
ui/src/e2e/chat-flow.messaging.e2e.test.ts
+4 -3
View File
@@ -888,7 +888,7 @@ suite.define(() => {
});
});
it("steers the exact run with the current leaf reported by the session row", async () => {
it("sends /steer for the selected session without resolving a run or leaf", async () => {
await withChatPage(async (page) => {
const sessionKey = "main";
const gateway = await installMockGateway(page, {
@@ -929,8 +929,9 @@ suite.define(() => {
expect(params.sessionKey).toBe(sessionKey);
expect(params.message).toBe("use the smaller fix");
expect(params.deliver).toBe(false);
expect(params.expectedRunId).toBe("active-run");
expect(params.expectedLeafEntryId).toBe("leaf-before-steer");
expect(params.queueMode).toBe("steer");
expect(params).not.toHaveProperty("expectedRunId");
expect(params).not.toHaveProperty("expectedLeafEntryId");
await page.getByText("Steered.", { exact: true }).waitFor({ timeout: 10_000 });
expect(await page.getByText("No active run").count()).toBe(0);
ui/src/i18n/locales/en.ts
-4
View File
@@ -4896,8 +4896,6 @@ export const en: TranslationMap = {
},
sendErrors: {
activeLeafChanged: "The session switched branches — review and resend.",
steerRunNoLongerActive:
"This steer still targets the previous run, but that run is no longer active.",
},
waitingForApproval: "Waiting for approval…",
startupStatus: {
@@ -5002,7 +5000,6 @@ export const en: TranslationMap = {
timeout: "The active run ended before the steer message was accepted.",
failed: "Steer failed before it reached the run; try again.",
usage: "Usage: `/steer <message>`",
noActiveRun: "No active run. Use the chat input or `/redirect` instead.",
succeeded: "Steered.",
requestFailed: "Failed to steer: {error}",
},
@@ -5386,7 +5383,6 @@ export const en: TranslationMap = {
editing: "Editing a queued message",
cancelEdit: "Cancel editing and keep the queued message",
states: {
steering: "Steering",
applyingSettings: "Applying chat settings",
waitingForRun: "Waiting for current run",
runningCommand: "Running command",
ui/src/lib/chat/chat-queue-order.test.ts
+1 -1
View File
@@ -133,7 +133,7 @@ describe("chat queue order", () => {
},
{ label: "failed send", item: queued("a", 1, { sendState: "failed" }), movable: true },
{ label: "in-flight send", item: queued("a", 1, { sendState: "sending" }), movable: false },
{ label: "steer chip", item: queued("a", 1, { kind: "steered" }), movable: false },
{ label: "pending run row", item: queued("a", 1, { pendingRunId: "run-1" }), movable: false },
{ label: "joined a run", item: queued("a", 1, { pendingRunId: "run-1" }), movable: false },
{
label: "running a local command",
ui/src/lib/chat/chat-queue-order.ts
+3 -4
View File
@@ -11,8 +11,8 @@ export function chatQueueOrderKey(item: ChatQueuePosition): number {
}
/**
* The one queue comparator. Display projection, drain head selection, steer
* rebuild, and alias merge all sort through it, so what the operator sees is
* The one queue comparator. Display projection, drain head selection, and
* alias merge all sort through it, so what the operator sees is
* what the Gateway receives. Equal positions keep their existing relative order
* through sort stability, which is how same-millisecond arrivals stayed FIFO.
*/
@@ -22,13 +22,12 @@ export function compareChatQueueOrder(left: ChatQueuePosition, right: ChatQueueP
/**
* A row may move while it is still waiting for its turn. Rows already attached
* to a run — sending, steering, running a command, or awaiting settings — keep
* to a run — sending, running a command, or awaiting settings — keep
* their place, so a move can never jump ahead of work already handed over.
*/
export function isMovableChatQueueItem(item: ChatQueueItem): boolean {
return (
!item.pendingRunId &&
item.kind !== "steered" &&
(item.sendState === undefined ||
item.sendState === "waiting-idle" ||
item.sendState === "waiting-reconnect" ||
ui/src/lib/chat/chat-types.ts
+3 -4
View File
@@ -3,6 +3,7 @@
*/
import type { MediaKind } from "@openclaw/media-core/constants";
import type { QueueMode } from "../../../../packages/gateway-protocol/src/schema/logs-chat.js";
import type { toolIcons } from "../../components/icons-tools.ts";
import type { SenderIdentity } from "./sender-label.ts";
@@ -63,7 +64,6 @@ export type ChatQueueItem = {
createdAt: number;
/** Operator-owned queue position; absent means "wherever arrival put it". */
orderKey?: number;
kind?: "queued" | "steered";
attachments?: ChatAttachment[];
refreshSessions?: boolean;
/** Transcript id of the replied-to message; Gateway hydrates reply context. */
@@ -74,13 +74,12 @@ export type ChatQueueItem = {
sendAttempts?: number;
sendError?: string;
sendRunId?: string;
/** Immutable active run selected when this row first became a steer. */
steerTargetRunId?: string;
/** One-send override retained with the durable row for reconnect and retry. */
queueMode?: QueueMode;
sendState?:
| "waiting-model"
| "waiting-idle"
| "executing-command"
| "steering"
| "sending"
| "waiting-reconnect"
| "unconfirmed"
ui/src/lib/chat/outbox-store-codec.ts
+13 -7
View File
@@ -1,5 +1,6 @@
import { isRecord } from "@openclaw/normalization-core/record-coerce";
import { readNonBlankString as normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { normalizeQueueMode } from "../../../../src/auto-reply/reply/queue/normalize.js";
import { normalizeAgentId } from "../sessions/session-key.ts";
import type { ChatAttachment, ChatQueueItem } from "./chat-types.ts";
import { normalizeSenderIdentity } from "./sender-label.ts";
@@ -79,8 +80,15 @@ export function normalizeStoredQueueItem(value: unknown): ChatQueueItem | null {
if (sender) {
item.sender = sender;
}
if (entry.kind === "queued" || entry.kind === "steered") {
item.kind = entry.kind;
const legacySteer =
entry.kind === "steered" ||
normalizeOptionalString(entry.steerTargetRunId) !== undefined ||
entry.sendState === "steering";
const queueMode = legacySteer
? "steer"
: normalizeQueueMode(typeof entry.queueMode === "string" ? entry.queueMode : undefined);
if (queueMode) {
item.queueMode = queueMode;
}
if (attachments.length) {
item.attachments = attachments;
@@ -93,7 +101,9 @@ export function normalizeStoredQueueItem(value: unknown): ChatQueueItem | null {
if (replyToId) {
item.replyToId = replyToId;
}
if (
if (entry.sendState === "steering") {
item.sendState = "unconfirmed";
} else if (
entry.sendState === "failed" ||
entry.sendState === "unconfirmed" ||
entry.sendState === "waiting-idle" ||
@@ -112,10 +122,6 @@ export function normalizeStoredQueueItem(value: unknown): ChatQueueItem | null {
if (sendRunId) {
item.sendRunId = sendRunId;
}
const steerTargetRunId = normalizeOptionalString(entry.steerTargetRunId);
if (steerTargetRunId) {
item.steerTargetRunId = steerTargetRunId;
}
if (typeof entry.sendAttempts === "number" && Number.isFinite(entry.sendAttempts)) {
item.sendAttempts = entry.sendAttempts;
}
ui/src/lib/chat/outbox-store.test.ts
+2 -3
View File
@@ -373,7 +373,6 @@ describe("stored outbox summaries", () => {
undefined,
"waiting-idle",
"executing-command",
"steering",
"sending",
"waiting-reconnect",
] as const;
@@ -427,8 +426,8 @@ describe("stored outbox summaries", () => {
const threadA = storedChatOutboxScopeKey({ sessionKey: "thread-a" });
const threadB = storedChatOutboxScopeKey({ sessionKey: "thread-b" });
expect(summary.total).toBe(9);
expect(summary.countsByScope.get(threadA)).toBe(8);
expect(summary.total).toBe(8);
expect(summary.countsByScope.get(threadA)).toBe(7);
expect(summary.countsByScope.get(threadB)).toBe(1);
expect(summary.attentionCountsByScope.get(threadA)).toBe(2);
expect(summary.attentionCountsByScope.get(threadB)).toBe(1);
ui/src/pages/chat/chat-command-executor.test.ts
+9 -255
View File
@@ -1570,13 +1570,10 @@ describe("executeSlashCommand directives", () => {
});
describe("executeSlashCommand /steer (soft inject)", () => {
it("injects into the current session via chat.send with deliver: false", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("agent:main:main", { status: "running" })] };
}
it("sends the selected session without resolving a run or leaf", async () => {
const request = vi.fn(async (method: string) => {
if (method === "chat.send") {
return { status: "started", runId: "run-1", messageSeq: 2 };
return { status: "started", runId: "run-1" };
}
throw new Error(`unexpected method: ${method}`);
});
@@ -1588,81 +1585,19 @@ describe("executeSlashCommand /steer (soft inject)", () => {
"try a different approach",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
expect(result.pendingCurrentRun).toBe(true);
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
expect(chatSend.payload.message).toBe("try a different approach");
expect(chatSend.payload.deliver).toBe(false);
expect(chatSend.payload.queueMode).toBe("steer");
});
it("uses a unique run id when a real session row omits active leaf context", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return {
sessions: [
row("agent:main:main", {
hasActiveRun: true,
activeRunIds: ["active-run"],
activeLeafEntryId: undefined,
}),
],
};
}
if (method === "chat.send") {
return { status: "started", runId: "run-active-flag", messageSeq: 2 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"continue with the smaller fix",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
expect(result.pendingCurrentRun).toBe(true);
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload).toMatchObject({
sessionKey: "agent:main:main",
message: "continue with the smaller fix",
message: "try a different approach",
deliver: false,
expectedRunId: "active-run",
queueMode: "steer",
idempotencyKey: expect.any(String),
});
expect(chatSend.payload).not.toHaveProperty("expectedRunId");
expect(chatSend.payload).not.toHaveProperty("expectedLeafEntryId");
});
it.each([
["zero", []],
["multiple", ["run-a", "run-b"]],
] as const)("refuses %s authoritative active run ids", async (_label, activeRunIds) => {
const request = vi.fn(async (method: string) => {
if (method === "sessions.list") {
return {
sessions: [
row("agent:main:main", {
hasActiveRun: true,
activeRunIds: [...activeRunIds],
activeLeafEntryId: undefined,
}),
],
};
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"continue safely",
);
expect(result.content).toBe(t("chat.commandResults.steer.noActiveRun"));
expectNoRequestCall(request, "chat.send");
expectNoRequestCall(request, "sessions.list");
});
it("does not mark the current run pending when chat.send returns terminal ok", async () => {
@@ -1696,9 +1631,6 @@ describe("executeSlashCommand /steer (soft inject)", () => {
"reports terminal %s ACK without marking the current run pending",
async (status, expectedKey) => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("agent:main:main", { status: "running" })] };
}
if (method === "chat.send") {
return { status, runId: `run-${status}`, summary: "aborted" };
}
@@ -1722,9 +1654,6 @@ describe("executeSlashCommand /steer (soft inject)", () => {
it("passes selected-agent scope when steering the selected global session", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("global", { status: "running" })] };
}
if (method === "chat.send") {
return { status: "started", runId: "run-global", messageSeq: 2 };
}
@@ -1740,7 +1669,6 @@ describe("executeSlashCommand /steer (soft inject)", () => {
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
expect(request).toHaveBeenCalledWith("sessions.list", { agentId: "work" });
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload).toMatchObject({
sessionKey: "global",
@@ -1750,177 +1678,6 @@ describe("executeSlashCommand /steer (soft inject)", () => {
});
});
it("passes selected-agent scope when steering a selected-global alias", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("global", { status: "running" })] };
}
if (method === "chat.send") {
return { status: "started", runId: "run-global", messageSeq: 2 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:work:main",
"steer",
"try the alias",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
expect(request).toHaveBeenCalledWith("sessions.list", { agentId: "work" });
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload).toMatchObject({
sessionKey: "agent:work:main",
agentId: "work",
message: "try the alias",
deliver: false,
});
});
it("uses cached sessions to avoid an extra sessions.list round trip", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "chat.send") {
return { status: "started", runId: "run-2", messageSeq: 1 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"researcher try a different approach",
{
sessionsResult: {
sessions: [
row("agent:main:main", { status: "running" }),
row("agent:main:subagent:researcher", {
spawnedBy: "agent:main:main",
status: "running",
}),
],
} as SessionsListResult,
},
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
expect(request).toHaveBeenCalledTimes(1);
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
expect(chatSend.payload.message).toBe("researcher try a different approach");
expect(chatSend.payload.deliver).toBe(false);
});
it("does not treat 'all' as a subagent wildcard", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("agent:main:main", { status: "running" })] };
}
if (method === "chat.send") {
return { status: "started", runId: "run-3", messageSeq: 1 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"all good now",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
expect(chatSend.payload.message).toBe("all good now");
expect(chatSend.payload.deliver).toBe(false);
});
it("does not match agent id as target — treats 'main' as message text", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return {
sessions: [
row("agent:main:main", { status: "running" }),
row("agent:main:subagent:researcher", { spawnedBy: "agent:main:main" }),
],
};
}
if (method === "chat.send") {
return { status: "started", runId: "run-4", messageSeq: 1 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"main refine the plan",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
expect(chatSend.payload.message).toBe("main refine the plan");
expect(chatSend.payload.deliver).toBe(false);
});
it("treats subagent-looking prefixes as current-session message text", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return {
sessions: [
row("agent:main:main", { status: "running" }),
row("agent:main:subagent:researcher", {
spawnedBy: "agent:main:main",
endedAt: Date.now() - 60_000,
}),
],
};
}
if (method === "chat.send") {
return { status: "started", runId: "run-5", messageSeq: 1 };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"researcher try again",
);
expect(result.content).toBe(t("chat.commandResults.steer.succeeded"));
const chatSend = requireRequestCall(request, "chat.send");
expect(chatSend.payload.sessionKey).toBe("agent:main:main");
expect(chatSend.payload.message).toBe("researcher try again");
expect(chatSend.payload.deliver).toBe(false);
});
it("returns a no-op summary when the current session has no active run", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("agent:main:main", { status: "done", endedAt: Date.now() })] };
}
throw new Error(`unexpected method: ${method}`);
});
const result = await executeSlashCommand(
createTestGatewayClient(request),
"agent:main:main",
"steer",
"try again",
);
expect(result.content).toBe(t("chat.commandResults.steer.noActiveRun"));
expect(request).toHaveBeenCalledWith("sessions.list", {});
expectNoRequestCall(request, "chat.send");
});
it("returns steer usage when no message is provided", async () => {
const request = vi.fn();
@@ -1936,10 +1693,7 @@ describe("executeSlashCommand /steer (soft inject)", () => {
});
it("returns steer error message on RPC failure", async () => {
const request = vi.fn(async (method: string, _payload?: unknown) => {
if (method === "sessions.list") {
return { sessions: [row("agent:main:main", { status: "running" })] };
}
const request = vi.fn(async () => {
throw new Error("connection lost");
});
ui/src/pages/chat/chat-command-executor.ts
+4 -24
View File
@@ -38,7 +38,6 @@ import {
import { formatUiError, formatUiExternalText } from "../../lib/format-error.ts";
import { formatCompactTokenCount } from "../../lib/format.ts";
import { readSessionMethodAccess } from "../../lib/session-method-access.ts";
import { isSessionRunActive } from "../../lib/session-run-state.ts";
import type { SessionCapability } from "../../lib/sessions/index.ts";
import {
DEFAULT_AGENT_ID,
@@ -811,10 +810,10 @@ async function loadModelCatalog(
}
}
async function resolveSteerTarget(
function resolveCommandMessage(
sessionKey: string,
args: string,
): Promise<{ key: string; message: string } | { error: string }> {
): { key: string; message: string } | { error: string } {
const trimmed = args.trim();
if (!trimmed) {
return { error: "empty" };
@@ -825,12 +824,6 @@ async function resolveSteerTarget(
};
}
function isActiveSteerSession(
session: GatewaySessionRow | undefined,
): session is GatewaySessionRow & { activeRunIds: [string] } {
return Boolean(session && isSessionRunActive(session) && session.activeRunIds?.length === 1);
}
type SteerChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error";
function normalizeSteerChatSendAckStatus(payload: unknown): SteerChatSendAckStatus {
@@ -871,21 +864,12 @@ async function executeSteer(
context: SlashCommandContext,
): Promise<SlashCommandResult> {
try {
const resolved = await resolveSteerTarget(sessionKey, args);
const resolved = resolveCommandMessage(sessionKey, args);
if ("error" in resolved) {
return {
content: resolved.error === "empty" ? t("chat.commandResults.steer.usage") : resolved.error,
};
}
const sessions =
context.sessionsResult ??
(await listSessions(context, selectedGlobalScope(sessionKey, context)));
const targetSession = resolveCurrentSession(sessions, resolved.key);
if (!isActiveSteerSession(targetSession)) {
return {
content: t("chat.commandResults.steer.noActiveRun"),
};
}
assertCurrentSlashCommand(context);
const ackStatus = normalizeSteerChatSendAckStatus(
await client.request("chat.send", {
@@ -894,10 +878,6 @@ async function executeSteer(
message: resolved.message,
deliver: false,
queueMode: "steer",
expectedRunId: targetSession.activeRunIds[0],
...(targetSession.activeLeafEntryId !== undefined
? { expectedLeafEntryId: targetSession.activeLeafEntryId }
: {}),
idempotencyKey: generateUUID(),
}),
);
@@ -926,7 +906,7 @@ async function executeRedirect(
context: SlashCommandContext,
): Promise<SlashCommandResult> {
try {
const resolved = await resolveSteerTarget(sessionKey, args);
const resolved = resolveCommandMessage(sessionKey, args);
if ("error" in resolved) {
return {
content:
ui/src/pages/chat/chat-composer-actions.test.ts
+2 -2
View File
@@ -165,7 +165,7 @@ describe("renderChatComposer controls", () => {
onQueueSteer,
queue: [
{ id: "queued-1", text: "tighten the plan", createdAt: 1 },
{ id: "steered-1", text: "already sent", createdAt: 2, kind: "steered" },
{ id: "pending-1", text: "already sent", createdAt: 2, pendingRunId: "run-1" },
{ id: "local-1", text: "/status", createdAt: 3, localCommandName: "status" },
{
id: "waiting-idle-1",
@@ -175,7 +175,7 @@ describe("renderChatComposer controls", () => {
},
],
});
const steer = [...container.querySelectorAll<HTMLButtonElement>(".chat-queue__steer")];
const steer = [...container.querySelectorAll<HTMLButtonElement>(".chat-queue__action")];
expect(steer).toHaveLength(2);
steer[0]?.click();
steer[1]?.click();
ui/src/pages/chat/chat-composer-queue.test.ts
+13 -18
View File
@@ -12,10 +12,7 @@ afterEach(async () => {
});
describe("chat composer steering queue", () => {
it.each([
{ sendState: "steering" as const, sendRunId: "send-1" },
{ pendingRunId: "run-1", sendRunId: "send-1" },
])("renders one Steering badge for an in-flight or acknowledged steer", (steerState) => {
it("renders the durable steer mode without a run-bound state", () => {
const container = document.createElement("div");
document.body.append(container);
render(
@@ -25,8 +22,8 @@ describe("chat composer steering queue", () => {
id: "steer-1",
text: "change course",
createdAt: 1,
kind: "steered",
...steerState,
queueMode: "steer",
sendState: "waiting-idle",
},
],
onQueueRemove: vi.fn(),
@@ -35,11 +32,9 @@ describe("chat composer steering queue", () => {
);
const badges = container.querySelectorAll(".chat-queue__badge");
expect(badges).toHaveLength(1);
expect(badges[0]?.textContent?.trim()).toBe(t("chat.queue.states.steering"));
const icon = container.querySelector(".chat-queue__icon");
expect(icon?.querySelector('polyline[points="15 10 20 15 15 20"]')).not.toBeNull();
expect(icon?.querySelector("circle")).toBeNull();
expect(badges).toHaveLength(2);
expect(badges[0]?.textContent?.trim()).toBe(t("chat.queue.steer"));
expect(badges[1]?.textContent?.trim()).toBe(t("chat.queue.states.waitingForRun"));
});
it("keeps a failed steer visually classified as an error", () => {
@@ -52,7 +47,7 @@ describe("chat composer steering queue", () => {
id: "failed-steer",
text: "change course",
createdAt: 1,
kind: "steered",
queueMode: "steer",
sendState: "failed",
sendError: "steer rejected",
},
@@ -64,11 +59,11 @@ describe("chat composer steering queue", () => {
const row = container.querySelector(".chat-queue__item");
expect(row?.classList.contains("chat-queue__item--failed")).toBe(true);
expect(row?.classList.contains("chat-queue__item--steered")).toBe(false);
const icon = row?.querySelector(".chat-queue__icon");
expect(icon?.querySelector('path[d^="m21.73 18"]')).not.toBeNull();
expect(icon?.querySelector('polyline[points="15 10 20 15 15 20"]')).toBeNull();
expect(container.querySelector(".chat-queue__badge--steered")).toBeNull();
expect(container.querySelector(".chat-queue__badge")?.textContent?.trim()).toBe(
t("chat.queue.steer"),
);
});
});
@@ -155,7 +150,7 @@ describe("chat composer queue reordering", () => {
it("reserves the handle column on every row so the pills never shift", () => {
const container = renderQueue({
queue: [
{ id: "steer", text: "steer", createdAt: 1, kind: "steered", pendingRunId: "run-1" },
{ id: "pending", text: "pending", createdAt: 1, pendingRunId: "run-1" },
waiting("b", 2),
waiting("c", 3),
],
@@ -224,7 +219,7 @@ describe("chat composer queue reordering", () => {
expect(rows[1]?.querySelector(".chat-queue__edit-input")).not.toBeNull();
expect(rows[1]?.querySelector(".chat-queue__edit-submit")).not.toBeNull();
expect(rows[1]?.querySelector(".chat-queue__edit-cancel")).not.toBeNull();
expect(rows.map((row) => row.querySelector(".chat-queue__steer") !== null)).toEqual([
expect(rows.map((row) => row.querySelector(".chat-queue__action") !== null)).toEqual([
true,
false,
true,
@@ -271,7 +266,7 @@ describe("chat composer queue reordering", () => {
it("keeps a row that already joined a run out of the reorder set", () => {
const container = renderQueue({
queue: [
{ id: "steer", text: "steer", createdAt: 1, kind: "steered", pendingRunId: "run-1" },
{ id: "pending", text: "pending", createdAt: 1, pendingRunId: "run-1" },
waiting("b", 2),
waiting("c", 3),
],
ui/src/pages/chat/chat-gateway.test.ts
-135
View File
@@ -855,129 +855,6 @@ describe("handleChatGatewayEvent", () => {
expect(state.chatStreamSegments).toEqual([]);
});
it("does not replay persisted keyed commentary after retiring a same-run steer", () => {
const originalUser = createTextChatMessage("user", "Ask", undefined, 1);
const persistedCommentary = {
role: "assistant",
content: [{ type: "text", text: "Looking into it." }],
timestamp: 2,
openclawStreamFallback: {
itemId: "preamble-1",
replacementText: "Looking into it.",
source: "segment",
},
};
const state = createState({
sessionKey: "main",
chatRunId: "run-1",
chatMessages: [originalUser, persistedCommentary],
chatQueue: [
{
id: "steer-1",
text: "Focus on the deployment too",
createdAt: 3,
kind: "steered",
pendingRunId: "run-1",
sendRunId: "steer-send-1",
sessionKey: "main",
},
],
chatStream: null,
chatStreamStartedAt: null,
}) as ChatState & {
chatStreamSegments: Array<{ text: string; ts: number; itemId: string }>;
};
state.chatStreamSegments = [{ text: "Looking into it.", ts: 2, itemId: "preamble-1" }];
expect(
handleChatGatewayEvent(state, {
runId: "run-1",
sessionKey: "main",
state: "final",
message: createTextChatMessage("assistant", "Final answer.", undefined, 5),
}),
).toBe("final");
expect(state.chatQueue).toEqual([]);
expect(state.chatMessages).toHaveLength(4);
expectTextChatMessage(state.chatMessages[0], "user", "Ask");
expectTextChatMessage(state.chatMessages[1], "assistant", "Looking into it.");
expectTextChatMessage(state.chatMessages[2], "user", "Focus on the deployment too");
expectTextChatMessage(state.chatMessages[3], "assistant", "Final answer.");
});
it("retires a reply steer chip after an exact-target terminal rejection", () => {
const state = createState({
sessionKey: "main",
chatRunId: "reply-steer-request",
chatQueue: [
{
id: "reply-steer-chip",
text: "Reply with deployment context",
createdAt: 3,
kind: "steered",
pendingRunId: "reply-steer-request",
sendRunId: "reply-steer-request",
sessionKey: "main",
},
],
});
expect(
handleChatGatewayEvent(state, {
runId: "reply-steer-request",
sessionKey: "main",
state: "error",
errorMessage: "active run changed; review and retry",
}),
).toBe("error");
expect(state.chatQueue).toEqual([]);
expect(state.chatRunId).toBeNull();
expect(state.chatRunError).toEqual({
summary: "Error: active run changed; review and retry",
});
});
it("keeps a pending steer chip when an unrelated request run finishes", () => {
const chip = {
id: "pending-steer-chip",
text: "Keep waiting for this steer",
createdAt: 3,
kind: "steered" as const,
pendingRunId: "active-run",
sendRunId: "steer-request-run",
sessionKey: "main",
};
const state = createState({
sessionKey: "main",
chatRunId: "active-run",
chatQueue: [
chip,
{
id: "unrelated-pending-row",
text: "Keep unrelated pending work",
createdAt: 4,
pendingRunId: "unrelated-run",
sessionKey: "main",
},
],
});
handleChatGatewayEvent(state, {
runId: "unrelated-run",
sessionKey: "main",
state: "final",
});
expect(state.chatQueue).toEqual([
chip,
expect.objectContaining({ id: "unrelated-pending-row" }),
]);
expect(state.chatRunId).toBe("active-run");
expect(state.chatMessages).toEqual([]);
});
it("preserves an already-recorded stream boundary for a persisted steer", () => {
const state = createState({
sessionKey: "main",
@@ -1001,17 +878,6 @@ describe("handleChatGatewayEvent", () => {
3,
),
],
chatQueue: [
{
id: "steer-1",
text: "Focus on deployment",
createdAt: 3,
kind: "steered",
pendingRunId: "run-1",
sendRunId: "steer-send-1",
sessionKey: "main",
},
],
}) as ChatState & {
chatStreamSegments: Array<{
text: string;
@@ -1036,7 +902,6 @@ describe("handleChatGatewayEvent", () => {
message: createTextChatMessage("assistant", "Final answer.", undefined, 5),
});
expect(state.chatQueue).toEqual([]);
expect(state.chatMessages).toHaveLength(4);
expectTextChatMessage(state.chatMessages[0], "user", "Ask");
expectTextChatMessage(state.chatMessages[1], "assistant", "Looking into it.");
ui/src/pages/chat/chat-gateway.ts
-35
View File
@@ -25,10 +25,6 @@ import {
} from "./history-merge.ts";
import { reconcileChatRunLifecycle } from "./run-lifecycle.ts";
import { appendChatMessageToCache } from "./session-message-cache.ts";
import {
retireSteeredChipsForRequestRun,
retireSteeredChipsForTerminalRun,
} from "./steer-lifecycle.ts";
import {
latestStreamBoundaryRunId,
reconcileTerminalStreamBoundary,
@@ -66,17 +62,6 @@ function isPendingLocalChatRun(state: ChatState, runId: string): boolean {
return state.chatQueue.some((item) => item.sendRunId === runId && item.sendState === "sending");
}
function isTerminalChatState(value: unknown): boolean {
return value === "final" || value === "aborted" || value === "error";
}
function isEventForDifferentActiveRun(
payload: ChatEventPayload | undefined,
activeRunId: string | null,
): boolean {
return Boolean(activeRunId && payload && payload.runId !== activeRunId);
}
function resolveDeltaChatStreamText(
currentStream: string | null,
payload: ChatEventPayload,
@@ -530,25 +515,5 @@ function handleChatEvent(state: ChatState, payload?: ChatEventPayload) {
}
export function handleChatGatewayEvent(state: ChatState, payload?: ChatEventPayload) {
const activeRunIdBeforeEvent = state.chatRunId;
const terminalEventMatchesChat =
isTerminalChatState(payload?.state) &&
payload !== undefined &&
// Unkeyed events must also carry a real run id: with no active run,
// `undefined === undefined` would let sessionless internal-run terminals
// (e.g. companion answers) materialize into the open main thread.
(chatEventSessionMatches(state, payload) ||
(typeof payload.runId === "string" && payload.runId === activeRunIdBeforeEvent));
const terminalOwnsActiveRun =
terminalEventMatchesChat && !isEventForDifferentActiveRun(payload, activeRunIdBeforeEvent);
// An accepted steer terminal is keyed by the steer request while the chip
// also tracks the active target run. Reconcile either identity before the
// generic different-run path ignores the terminal and leaves stale status.
if (terminalOwnsActiveRun) {
retireSteeredChipsForTerminalRun(state, payload?.runId);
}
if (terminalEventMatchesChat) {
retireSteeredChipsForRequestRun(state, payload?.runId);
}
return handleChatEvent(state, payload);
}
ui/src/pages/chat/chat-history.ts
-3
View File
@@ -67,7 +67,6 @@ import {
readChatSessionSnapshot,
type ChatSessionSnapshot,
} from "./session-message-cache.ts";
import { retirePersistedSteeredChips } from "./steer-lifecycle.ts";
import {
latestPersistedSteerBoundary,
markChatStreamAfterBoundary,
@@ -1559,7 +1558,6 @@ async function loadChatHistoryUncached(
state.chatThinkingLevel = response.sessionInfo.thinkingLevel ?? null;
state.chatQueueModeOverride = response.sessionInfo.queueMode;
state.chatEffectiveQueueMode = response.sessionInfo.effectiveQueueMode;
retirePersistedSteeredChips(state);
replaceCachedChatMessages(state, sessionKey, requestAgentId, response.deltaCursor);
recordChatHistoryTiming(state, "applied", startedAtMs, {
requestSessionKey: sessionKey,
@@ -1650,7 +1648,6 @@ async function loadChatHistoryUncached(
if (Object.hasOwn(res.sessionInfo ?? {}, "activeLeafEntryId")) {
state.chatDisplayedLeafEntryId = nextDisplayedLeafEntryId;
}
retirePersistedSteeredChips(state);
state.chatHistoryPagination = reconciledHistory?.pagination ?? nextPagination;
state.currentSessionId = nextSessionId;
replaceCachedChatMessages(state, sessionKey, requestAgentId, res.deltaCursor);
ui/src/pages/chat/chat-outbox-drain.ts
+24 -11
View File
@@ -27,6 +27,12 @@ import {
updateQueuedMessageForSession,
} from "./chat-queue.ts";
import type { ChatHost } from "./chat-send-contract.ts";
import {
chatMessagesContainQueuedSend,
OFFLINE_QUEUE_STORAGE_ERROR,
preserveQueuedUserTurn,
surfaceChatDeliveryFailure,
} from "./chat-send-support.ts";
import {
listStoredChatOutboxes,
storedChatOutboxScopeKey,
@@ -36,16 +42,12 @@ import {
import { formatConnectError } from "./connect-error.ts";
import { isQueuedMessageBeingEdited } from "./queued-message-edit.ts";
import { isChatBusy } from "./run-lifecycle.ts";
import {
chatMessagesContainQueuedSend,
OFFLINE_QUEUE_STORAGE_ERROR,
preserveQueuedUserTurn,
surfaceChatDeliveryFailure,
} from "./steer-lifecycle.ts";
export type QueuedChatSendResult = "sent" | "pending" | "failed";
export type QueuedChatStorageMode = "durable" | "memory";
export type QueuedChatSendOptions = {
/** Fresh selected-session sends may let the Gateway resolve its effective active-run mode. */
allowActiveRunSend?: boolean;
/** Exact submit-time leaf; restored drains omit it so intervening advances park the draft. */
expectedLeafEntryId?: string | null;
pendingSettings?: Promise<boolean>;
@@ -281,6 +283,9 @@ async function reconcileStoredChatOutboxHead(
// bubbles even mid-run, and a missing row falls through conservatively.
const neverAttempted =
(item.sendAttempts ?? 0) === 0 && item.sendRequestStartedAtMs === undefined;
if (neverAttempted && item.queueMode) {
return "send";
}
if (neverAttempted) {
const row =
!isUiGlobalSessionKey(outbox.sessionKey) || host.sessions.state.agentId === outbox.agentId
@@ -372,12 +377,20 @@ async function drainStoredChatOutbox(
if (!outbox) {
return "empty";
}
const storedItem = outbox.queue.find(
(entry) =>
lane.freshAdmissions.has(entry.id) ||
entry.sendState !== "failed" ||
entry.localCommandName,
// A fresh active-run send is an explicit operator action, not work queued
// behind the run. Let it bypass older FIFO rows; ordinary fresh admissions
// still preserve their existing order.
const freshActiveRunItem = outbox.queue.find(
(entry) => lane.freshAdmissions.has(entry.id) && Boolean(entry.queueMode),
);
const storedItem =
freshActiveRunItem ??
outbox.queue.find(
(entry) =>
lane.freshAdmissions.has(entry.id) ||
entry.sendState !== "failed" ||
entry.localCommandName,
);
const freshItem = storedItem && lane.freshAdmissions.has(storedItem.id);
const item = freshItem
? (readQueuedMessageById(host, storedItem.id) ?? storedItem)
ui/src/pages/chat/chat-queue.ts
+3 -53
View File
@@ -89,32 +89,6 @@ export function syncVisibleChatQueueProjection(
chatOutboxOwner(host).syncHost(host, options);
}
export function setTransientQueuedMessageProjection(
host: ChatQueueScopedSessionHost,
sessionKey: string,
item: ChatQueueItem,
agentId?: string,
): boolean {
const scope = resolveStoredChatOutboxScope(host, sessionKey, agentId);
const owner = chatOutboxOwner(host);
const outbox = owner.durable(host, item.id);
if (!outbox?.queue.some((entry) => entry.id === item.id)) {
return false;
}
owner.projectLive(host, scope, item.id, item);
return true;
}
export function clearTransientQueuedMessageProjection(
host: ChatQueueScopedSessionHost,
sessionKey: string,
id: string,
agentId?: string,
) {
const scope = resolveStoredChatOutboxScope(host, sessionKey, agentId);
chatOutboxOwner(host).projectLive(host, scope, id);
}
export function subscribeChatOutboxProjection(host: ChatQueueScopedSessionHost): () => void {
return chatOutboxOwner(host).subscribe(host);
}
@@ -160,13 +134,12 @@ export function enqueuePendingRunMessage(
if (!trimmed && !hasAttachments) {
return;
}
// Local commands join an existing run without a wire chat.send, so this is
// intentionally a non-SteeredChip pending row with no fake sendRunId.
// Local commands join an existing run without a wire chat.send, so this
// pending row intentionally has no fake send identity.
const item: ChatQueueItem = {
id: generateUUID(),
text: trimmed,
createdAt: Date.now(),
kind: "steered",
attachments: hasAttachments ? cloneChatAttachmentsMetadata(attachments ?? []) : undefined,
pendingRunId,
...(sender ? { sender } : {}),
@@ -183,30 +156,7 @@ export function readChatQueueForScope(
return chatOutboxOwner(host).snapshot(host, scope);
}
export function replacePendingQueuedMessageProjection(
host: ChatQueueScopedSessionHost,
sessionKey: string,
id: string,
pendingRunId: string,
replacement: ChatQueueItem,
agentId?: string,
): boolean {
const queue = readChatQueueForScope(host, sessionKey, agentId);
if (!queue.some((item) => item.id === id && item.pendingRunId === pendingRunId)) {
return false;
}
writeChatQueueForScope(
host,
sessionKey,
queue.map((item) =>
item.id === id && item.pendingRunId === pendingRunId ? replacement : item,
),
agentId,
);
return true;
}
export function writeChatQueueForScope(
function writeChatQueueForScope(
host: ChatQueueScopedSessionHost,
sessionKey: string,
queue: ChatQueueItem[],
ui/src/pages/chat/chat-reset-delivery.ts
+1 -1
View File
@@ -13,7 +13,7 @@ import {
reconnectSafeQueuedSendState,
setChatError,
} from "./chat-send-queue-state.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./steer-lifecycle.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./chat-send-support.ts";
type DeliverChatQueueItem = (
host: ChatHost,
ui/src/pages/chat/chat-send-ack.ts
+1 -1
View File
@@ -1,5 +1,5 @@
// Leaf contract for chat.send acknowledgment shapes and timing records.
// Kept import-free of chat-page modules so lifecycle/steer/history layers
// Kept import-free of chat-page modules so lifecycle and history layers
// can consume ack types without forming import cycles.
import { asNonNegativeFiniteNumber as normalizeAckTimingValue } from "@openclaw/normalization-core/number-coercion";
import type { ChatQueueItem } from "../../lib/chat/chat-types.ts";
ui/src/pages/chat/chat-send-actions.ts
+34 -68
View File
@@ -1,3 +1,4 @@
import type { QueueMode } from "../../../../packages/gateway-protocol/src/schema/logs-chat.js";
import { GatewayRequestError } from "../../api/gateway.ts";
import { t } from "../../i18n/index.ts";
import {
@@ -34,6 +35,7 @@ import {
requestChatSend,
resolveDisplayedLeafEntryId,
} from "./chat-send-request.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./chat-send-support.ts";
import { listStoredChatOutboxes, storedChatOutboxScopeKey } from "./composer-persistence.ts";
import { formatConnectError } from "./connect-error.ts";
import {
@@ -43,14 +45,8 @@ import {
isQueuedMessageReorderBlocked,
QUEUED_MESSAGE_RETRY_CONFLICT_ERROR,
QUEUED_MESSAGE_REORDER_CONFLICT_ERROR,
QUEUED_MESSAGE_STEER_CONFLICT_ERROR,
} from "./queued-message-edit.ts";
import { hasDirectSessionRun } from "./run-lifecycle.ts";
import {
OFFLINE_QUEUE_STORAGE_ERROR,
steerQueuedChatMessage as steerQueuedChatMessageLifecycle,
type SteerSendDependencies,
} from "./steer-lifecycle.ts";
import { isInflightSteer } from "./steered-chip.ts";
function applyChatSendError(state: ChatState, err: unknown, canApplyError: () => boolean): string {
const error = isActiveLeafChangedError(err)
@@ -69,7 +65,13 @@ export async function sendChatMessageWithGeneratedRunId(
state: ChatState,
message: string,
attachments?: ChatAttachment[],
options: Partial<Parameters<SteerSendDependencies["sendChatMessage"]>[3]> = {},
options: {
canApplyError?: () => boolean;
expectedLeafEntryId?: string | null;
queueMode?: QueueMode;
replyToId?: string;
runId?: string;
} = {},
) {
const msg = message.trim();
if (!state.client || !state.connected || (!msg && !attachments?.length)) {
@@ -87,12 +89,11 @@ export async function sendChatMessageWithGeneratedRunId(
message: msg,
attachments,
runId,
...(options.expectedLeafEntryId !== undefined
...(options.queueMode !== "steer" && options.expectedLeafEntryId !== undefined
? { expectedLeafEntryId: options.expectedLeafEntryId }
: expectedLeafEntryId !== undefined
: options.queueMode !== "steer" && expectedLeafEntryId !== undefined
? { expectedLeafEntryId }
: {}),
...(options.expectedRunId ? { expectedRunId: options.expectedRunId } : {}),
...(options.queueMode ? { queueMode: options.queueMode } : {}),
...(options.replyToId ? { replyToId: options.replyToId } : {}),
});
@@ -114,28 +115,23 @@ const resetRetryState = (
sendAttempts: 0,
sendError: undefined,
sendRequestStartedAtMs: undefined,
sendRunId: entry.sendState === "failed" ? generateUUID() : entry.sendRunId,
sendRunId:
entry.sendState === "failed" && entry.queueMode !== "steer" ? generateUUID() : entry.sendRunId,
sendState,
});
export const steerSendDependencies: SteerSendDependencies = {
loadChatHistory: (host) => void loadChatHistory(host),
resumeRestoredOutbox: (host, itemId) => {
const restoredOutbox = findStoredOutbox(host as ChatHost, itemId);
if (!host.chatRunId && restoredOutbox) {
void scheduleStoredChatOutboxDrain(
host as ChatHost,
restoredOutbox,
chatOutboxDrainDependencies,
);
}
},
sendChatMessage: (host, message, attachments, options) =>
sendChatMessageWithGeneratedRunId(host, message, attachments, options),
};
export const steerQueuedChatMessage = (host: ChatHost, id: string) =>
steerQueuedChatMessageLifecycle(host, id, steerSendDependencies);
export async function steerQueuedChatMessage(host: ChatHost, id: string): Promise<void> {
if (isQueuedMessageBeingEdited(host, id)) {
setChatError(host, QUEUED_MESSAGE_STEER_CONFLICT_ERROR);
return;
}
const item = updateQueuedMessage(host, id, (entry) => ({ ...entry, queueMode: "steer" }));
if (!item) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
await retryQueuedChatMessage(host, id);
}
export const resumeStoredChatOutboxes = (host: ChatHost) =>
resumeStoredChatOutboxesDrain(host, chatOutboxDrainDependencies);
@@ -220,7 +216,7 @@ export function moveQueuedChatMessage(
}
export async function retryQueuedChatMessage(host: ChatHost, id: string) {
let item = host.chatQueue.find((entry) => entry.id === id);
const item = host.chatQueue.find((entry) => entry.id === id);
if (isQueuedMessageRetryBlocked(host, id)) {
setChatError(host, QUEUED_MESSAGE_RETRY_CONFLICT_ERROR);
return;
@@ -229,47 +225,11 @@ export async function retryQueuedChatMessage(host: ChatHost, id: string) {
!item ||
item.pendingRunId ||
item.sendState === "executing-command" ||
isInflightSteer(item) ||
item.sendState === "sending" ||
item.sendState === "waiting-model"
) {
return;
}
if (item.kind === "steered") {
if (!host.connected || !host.client) {
setChatError(host, t("chat.sendErrors.steerRunNoLongerActive"));
return;
}
if (hasDirectSessionRun(host)) {
const retry = updateQueuedMessage(host, id, (entry) => ({
...entry,
sendAttempts: 0,
sendError: undefined,
sendRequestStartedAtMs: undefined,
sendState: "waiting-idle",
}));
if (!retry) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
await steerQueuedChatMessageLifecycle(host, id, steerSendDependencies);
return;
}
const converted = updateQueuedMessage(host, id, (entry) => {
const {
kind: _kind,
pendingRunId: _pendingRunId,
steerTargetRunId: _steerTargetRunId,
...queued
} = entry;
return resetRetryState(queued, reconnectSafeQueuedSendState(host));
});
if (!converted) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
item = converted;
}
let outbox = findStoredOutbox(host, item.id);
if (!outbox) {
const wasVolatile = isVolatileQueuedMessage(host, item.id);
@@ -310,7 +270,13 @@ export async function retryQueuedChatMessage(host: ChatHost, id: string) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
const drain = scheduleStoredChatOutboxDrain(host, outbox, chatOutboxDrainDependencies);
const drain = scheduleStoredChatOutboxDrain(
host,
outbox,
chatOutboxDrainDependencies,
retry.queueMode ? retry.id : undefined,
retry.queueMode ? { routingSessionKey: host.sessionKey } : undefined,
);
if (host.chatSending && host.chatSendingScopeKey === storedChatOutboxScopeKey(outbox)) {
void drain;
return;
ui/src/pages/chat/chat-send-delivery.ts
+9 -6
View File
@@ -45,6 +45,11 @@ import {
requestChatSend,
requestSkillWorkshopRevisionChatSend,
} from "./chat-send-request.ts";
import {
formatTerminalChatSendAckError,
OFFLINE_QUEUE_STORAGE_ERROR,
surfaceChatDeliveryFailure,
} from "./chat-send-support.ts";
import {
chatSendAckServerTimingEventFields,
recordChatSendTiming,
@@ -63,11 +68,6 @@ import { resetChatInputHistoryNavigation } from "./input-history.ts";
import { controlUiNowMs, roundedControlUiDurationMs } from "./performance.ts";
import { hasDirectSessionRun, isChatBusy, reconcileChatRunLifecycle } from "./run-lifecycle.ts";
import { resetChatScroll, scheduleChatScroll } from "./scroll.ts";
import {
formatTerminalChatSendAckError,
OFFLINE_QUEUE_STORAGE_ERROR,
surfaceChatDeliveryFailure,
} from "./steer-lifecycle.ts";
import { resetToolStream } from "./tool-stream.ts";
import { buildUserChatMessageContentBlocks } from "./user-message-content.ts";
@@ -306,7 +306,8 @@ async function sendQueuedChatMessage(
runId,
sessionKey,
agentId: prepared.agentId,
...(options?.expectedLeafEntryId !== undefined
...(prepared.queueMode ? { queueMode: prepared.queueMode } : {}),
...(prepared.queueMode !== "steer" && options?.expectedLeafEntryId !== undefined
? { expectedLeafEntryId: options.expectedLeafEntryId }
: {}),
...(prepared.replyToId ? { replyToId: prepared.replyToId } : {}),
@@ -614,6 +615,8 @@ export async function deliverChatQueueItem(
if (
drainResult === undefined &&
routeVisible &&
!admittedItem.queueMode &&
!sendOptions.allowActiveRunSend &&
(isChatBusy(host) || hasDirectSessionRun(host))
) {
const parked = finishChatDeliveryAdmission(
ui/src/pages/chat/chat-send-queue-state.ts
+9 -2
View File
@@ -15,13 +15,13 @@ import {
updateVolatileQueuedMessage,
} from "./chat-queue.ts";
import type { ChatHost } from "./chat-send-contract.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR, surfaceChatDeliveryFailure } from "./chat-send-support.ts";
import { recordChatSendTiming, schedulePendingSendPaintTiming } from "./chat-send-timing.ts";
import { getPendingChatPickerPatch } from "./chat-session.ts";
import { storedChatOutboxScopeKey, type StoredChatOutboxScope } from "./composer-persistence.ts";
import { controlUiNowMs } from "./performance.ts";
import { hasDirectSessionRun, isChatBusy } from "./run-lifecycle.ts";
import { scheduleChatScroll } from "./scroll.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR, surfaceChatDeliveryFailure } from "./steer-lifecycle.ts";
const SKILL_WORKSHOP_CONNECTION_CHANGED_ERROR =
"Skill Workshop revision request cancelled because the Gateway connection changed.";
@@ -45,6 +45,7 @@ export function enqueuePendingSendMessage(
skillWorkshopRevision?: ChatQueueItem["skillWorkshopRevision"],
replyToId?: string,
resumedOrderKey?: number,
queueMode?: ChatQueueItem["queueMode"],
): ChatQueueItem | null {
const trimmed = text.trim();
const hasAttachments = Boolean(attachments && attachments.length > 0);
@@ -64,6 +65,7 @@ export function enqueuePendingSendMessage(
sendAttempts: 0,
sendRunId: generateUUID(),
sendState,
...(queueMode ? { queueMode } : {}),
sendSubmittedAtMs: submittedAtMs,
sessionKey: host.sessionKey,
agentId: scopedAgentIdForSession(host, host.sessionKey),
@@ -192,7 +194,12 @@ export function finishChatDeliveryAdmission(
}
return "pending";
}
if (routeVisible(current.agentId) && (isChatBusy(host) || hasDirectSessionRun(host))) {
const sendsDuringActiveRun = Boolean(current.queueMode || options?.allowActiveRunSend);
if (
!sendsDuringActiveRun &&
routeVisible(current.agentId) &&
(isChatBusy(host) || hasDirectSessionRun(host))
) {
const parked = setState(host.connected && host.client ? "waiting-idle" : "waiting-reconnect");
if (!parked) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
ui/src/pages/chat/chat-send-request.ts
-2
View File
@@ -22,7 +22,6 @@ export async function requestChatSend(
queueMode?: QueueMode;
replyToId?: string;
expectedLeafEntryId?: string | null;
expectedRunId?: string;
},
): Promise<ChatSendAck> {
const routing = resolveChatSendRouting(state, params);
@@ -43,7 +42,6 @@ export async function requestChatSend(
...(params.expectedLeafEntryId !== undefined
? { expectedLeafEntryId: params.expectedLeafEntryId }
: {}),
...(params.expectedRunId ? { expectedRunId: params.expectedRunId } : {}),
idempotencyKey: params.runId,
attachments: buildChatApiAttachments(params.attachments),
});
ui/src/pages/chat/chat-send-submit.ts
+19 -24
View File
@@ -26,7 +26,7 @@ import {
readQueuedMessageById,
} from "./chat-queue.ts";
import { isTerminalFailureChatSendAck } from "./chat-send-ack.ts";
import { sendChatMessageWithGeneratedRunId, steerSendDependencies } from "./chat-send-actions.ts";
import { sendChatMessageWithGeneratedRunId } from "./chat-send-actions.ts";
import {
captureChatCommandComposerRecovery,
cancelChatDelivery,
@@ -48,6 +48,10 @@ import {
waitForPendingChatSettings,
} from "./chat-send-queue-state.ts";
import { resolveDisplayedLeafEntryId } from "./chat-send-request.ts";
import {
formatTerminalChatSendAckError,
OFFLINE_QUEUE_STORAGE_ERROR,
} from "./chat-send-support.ts";
import { recordChatSendTiming } from "./chat-send-timing.ts";
import { getPendingChatPickerPatch } from "./chat-session.ts";
import { withChatSubmitGuard } from "./chat-submit-guard.ts";
@@ -65,11 +69,6 @@ import {
isChatBusy,
isChatStopCommand,
} from "./run-lifecycle.ts";
import {
formatTerminalChatSendAckError,
OFFLINE_QUEUE_STORAGE_ERROR,
sendQueuedChatMessageWithQueueMode as sendQueuedChatMessageWithQueueModeLifecycle,
} from "./steer-lifecycle.ts";
type ChatSendSubmitOptions = {
attachmentsOverride?: readonly ChatAttachment[];
@@ -520,6 +519,16 @@ export async function handleSendChat(
const pendingSettings = getPendingChatPickerPatch(host, submittedSessionKey);
const waitingForSettings = Boolean(pendingSettings);
const directRunActive = hasDirectSessionRun(host);
// Only an explicit browser override replaces inherited Gateway policy.
const followUpMode =
opts?.followUpMode ??
host.chatFollowUpMode ??
normalizeChatFollowUpModeOverride(host.settings?.chatFollowUpMode);
const activeRunQueueMode =
!skillWorkshopRevision && directRunActive && followUpMode !== "queue"
? followUpMode
: undefined;
// The edited row hands its place to the replacement and is retired by the same
// store write, so a rejected write leaves the original queued and editable.
const resumedEdit =
@@ -534,6 +543,7 @@ export async function handleSendChat(
skillWorkshopRevision,
replyToId,
resumedEdit?.orderKey,
activeRunQueueMode,
);
if (!queued) {
return;
@@ -572,6 +582,9 @@ export async function handleSendChat(
const sendResult = await deliverChatQueueItem(host, queued, {
previousDraft: cleared.previousDraft,
previousAttachments: cleared.previousAttachments,
...(!skillWorkshopRevision && directRunActive && followUpMode !== "queue"
? { allowActiveRunSend: true }
: {}),
...(expectedLeafEntryId !== undefined ? { expectedLeafEntryId } : {}),
...(pendingSettings ? { pendingSettings } : {}),
restoreAttachments: Boolean(messageOverride && opts?.restoreDraft),
@@ -588,24 +601,6 @@ export async function handleSendChat(
(isChatBusy(host) || hasDirectSessionRun(host));
if (pendingBusySend) {
recordChatSendTiming(host, pending, "queued-busy", submittedAtMs);
// Only an explicit browser override replaces inherited Gateway policy.
const followUpMode =
opts?.followUpMode ??
host.chatFollowUpMode ??
normalizeChatFollowUpModeOverride(host.settings?.chatFollowUpMode);
if (
!skillWorkshopRevision &&
followUpMode !== "queue" &&
host.connected &&
hasDirectSessionRun(host)
) {
void sendQueuedChatMessageWithQueueModeLifecycle(
host,
pending.id,
followUpMode,
steerSendDependencies,
);
}
}
if (
sendResult !== "failed" &&
ui/src/pages/chat/chat-send-support.ts
+151
View File
@@ -0,0 +1,151 @@
import { asOptionalRecord, isRecord } from "@openclaw/normalization-core/record-coerce";
import type { SessionsListResult } from "../../api/types.ts";
import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts";
import { formatUiError } from "../../lib/format-error.ts";
import { resolveSessionDisplayName } from "../../lib/session-display.ts";
import { visibleSessionMatches } from "../../lib/sessions/index.ts";
import {
areUiSessionKeysEquivalent,
isUiGlobalSessionKey,
normalizeAgentId,
} from "../../lib/sessions/session-key.ts";
import { showToast } from "../../lib/toast.ts";
import { getChatAttachmentDataUrl } from "./attachment-payload-store.ts";
import type { TerminalFailureChatSendAck } from "./chat-send-ack.ts";
import type { ChatState } from "./chat-state-contract.ts";
import { readChatSessionProjectionScope, reduceChatSessionProjection } from "./history-merge.ts";
import { appendChatMessageToCache, readChatMessagesFromCache } from "./session-message-cache.ts";
import { buildUserChatMessageContentBlocks } from "./user-message-content.ts";
type ChatSendSupportHost = ChatState & {
sessionsResult?: SessionsListResult | null;
};
export const OFFLINE_QUEUE_STORAGE_ERROR =
"Could not store this message for reconnect. Free browser storage or reconnect before sending.";
export function formatTerminalChatSendAckError(
ack: TerminalFailureChatSendAck,
context: "chat" | "detached",
): string {
return ack.status === "error"
? "Chat failed before the run started; try again."
: context === "detached"
? "The active run ended before the detached message was accepted."
: "The run ended before the message was accepted.";
}
export function chatMessagesContainQueuedSend(
messages: unknown,
item: ChatQueueItem,
userRoleOnly = false,
): boolean {
return findQueuedSendMessageIndex(messages, item, userRoleOnly) >= 0;
}
function findQueuedSendMessageIndex(
messages: unknown,
item: ChatQueueItem,
userRoleOnly = false,
): number {
if (!item.sendRunId) {
return -1;
}
return (Array.isArray(messages) ? messages : []).findIndex((message) => {
if (!isRecord(message)) {
return false;
}
// Render retirement requires a user-role entry: an assistant entry can
// carry the same run key without proving the queued turn is visible.
if (userRoleOnly && message.role !== "user") {
return false;
}
const markerIdempotencyKey = asOptionalRecord(message["__openclaw"])?.idempotencyKey;
const idempotencyKey = markerIdempotencyKey ?? message.idempotencyKey;
return idempotencyKey === item.sendRunId || idempotencyKey === `${item.sendRunId}:user`;
});
}
function durableDeliveredAttachments(
attachments: readonly ChatAttachment[] | undefined,
): ChatAttachment[] | undefined {
return attachments?.flatMap((attachment) => {
// Composer uploads keep their bytes in the payload store; queue rows carry
// metadata only. Resolve through the store before queue ownership ends.
const dataUrl = getChatAttachmentDataUrl(attachment);
return dataUrl ? [{ ...attachment, dataUrl, previewUrl: dataUrl }] : [];
});
}
export function preserveQueuedUserTurn(state: ChatSendSupportHost, item: ChatQueueItem): void {
const runId = item.sendRunId;
const sessionKey = item.sessionKey ?? state.sessionKey;
if (!runId) {
return;
}
const content = buildUserChatMessageContentBlocks(
item.text,
durableDeliveredAttachments(item.attachments),
);
if (!content.length) {
return;
}
const userMessage = {
role: "user",
content,
timestamp: item.createdAt,
__openclaw: { idempotencyKey: `${runId}:user` },
};
if (visibleSessionMatches(state, sessionKey, item.agentId)) {
if (!chatMessagesContainQueuedSend(state.chatMessages, item, true)) {
const scope = readChatSessionProjectionScope(state, {
sessionKey,
agentId: item.agentId,
});
reduceChatSessionProjection(
state,
{ type: "sendPending", runId, message: userMessage },
{ scope },
);
}
return;
}
if (!state.chatMessagesBySession) {
return;
}
const target = { sessionKey, agentId: item.agentId };
const cached = readChatMessagesFromCache(state.chatMessagesBySession, state, target);
if (!chatMessagesContainQueuedSend(cached, item, true)) {
appendChatMessageToCache(state.chatMessagesBySession, state, target, userMessage);
}
}
type ChatDeliveryFailureHost = Parameters<typeof visibleSessionMatches>[0] & {
lastError?: string | null;
chatError?: string | null;
sessionsResult?: SessionsListResult | null;
};
/** Surface a terminal delivery failure in the owning pane or a named toast. */
export function surfaceChatDeliveryFailure(
host: ChatDeliveryFailureHost,
sessionKey: string,
agentId: string | undefined,
error: string,
): void {
const message = formatUiError(error);
if (visibleSessionMatches(host, sessionKey, agentId)) {
host.lastError = message;
host.chatError = message;
return;
}
const scopedAgentId = agentId ? normalizeAgentId(agentId) : undefined;
const row = host.sessionsResult?.sessions.find(
(session) =>
areUiSessionKeysEquivalent(session.key, sessionKey) &&
(!isUiGlobalSessionKey(sessionKey) ||
!scopedAgentId ||
(session.agentId !== undefined && normalizeAgentId(session.agentId) === scopedAgentId)),
);
showToast({ message: `${resolveSessionDisplayName(sessionKey, row)}: ${message}` });
}
ui/src/pages/chat/chat-send.test.ts
+119 -1338
View File
File diff suppressed because it is too large Load Diff
ui/src/pages/chat/chat-state-events.ts
+7 -16
View File
@@ -27,10 +27,12 @@ import {
shouldHideAssistantChatMessage,
} from "./chat-history.ts";
import {
clearPendingQueueItemsForRun,
readDeliveredQueuedChatSendForRun,
removeDeliveredQueuedChatSendForRun,
} from "./chat-queue.ts";
import { flushChatQueueForEvent, resumeStoredChatOutboxes } from "./chat-send-actions.ts";
import { preserveQueuedUserTurn } from "./chat-send-support.ts";
import { recordChatSendServerTiming } from "./chat-send-timing.ts";
import { refreshCurrentChatSessionList } from "./chat-session.ts";
import type { ChatPageHost } from "./chat-state-host.ts";
@@ -45,12 +47,6 @@ import {
reconcileStaleChatRunAfterSessionStatePublication,
} from "./run-lifecycle.ts";
import { applySessionMessagePayload } from "./session-message-apply.ts";
import {
preserveQueuedUserTurn,
retirePersistedSteeredChips,
retireSteeredChipsForTerminalRun,
} from "./steer-lifecycle.ts";
import { isAckedSteeredChip } from "./steered-chip.ts";
import { rememberAuthoritativeTerminal } from "./terminal-message-identity.ts";
import { handleAgentEvent, handleSessionOperationEvent } from "./tool-stream.ts";
@@ -108,7 +104,7 @@ function finishSessionMessageRunReconcile(
if (!cleared) {
return false;
}
retireSteeredChipsForTerminalRun(state, runId ?? undefined);
clearPendingQueueItemsForRun(state, runId ?? undefined);
void loadChatHistory(state)
.finally(() => {
if (!areUiSessionKeysEquivalent(state.sessionKey, sessionKey)) {
@@ -135,7 +131,6 @@ function handleSessionMessageEvent(state: ChatPageHost, payload: unknown) {
kind: "live",
activeRunId: state.chatRunId,
});
retirePersistedSteeredChips(state);
}
if (matchesChat && event.archived !== null) {
state.selectedChatSessionArchived = event.archived;
@@ -426,6 +421,9 @@ export function handlePageGatewayEvent(state: ChatPageHost, event: GatewayEventF
preserveQueuedUserTurn(state, delivered);
}
const result = handleChatGatewayEvent(state, payload);
if (terminal) {
clearPendingQueueItemsForRun(state, payload?.runId);
}
if (shouldCelebrateFirstReply && result === "final") {
fireFirstReplyConfetti();
}
@@ -514,9 +512,6 @@ function rememberDeliveredQueuedUserTurn(
turns = new Map();
deliveredQueueTurnsByClient.set(owner, turns);
}
const pending = state.chatQueue.find(
(item) => isAckedSteeredChip(item) && item.pendingRunId === runId,
);
const stored = readDeliveredQueuedChatSendForRun(state, runId)?.item;
if (stored) {
turns.delete(runId);
@@ -529,9 +524,5 @@ function rememberDeliveredQueuedUserTurn(
turns.delete(oldestRunId);
}
}
// Original-turn copies first: a run can own both its queued turn (stored, or
// its remembered fallback in `turns`) and a steered follow-up chip; the chip
// is preserved separately by retireSteeredChipsForTerminalRun and must not
// mask the original copy here.
return stored ?? turns.get(runId) ?? pending ?? null;
return stored ?? turns.get(runId) ?? null;
}
ui/src/pages/chat/chat-state-page.ts
+1 -1
View File
@@ -27,6 +27,7 @@ import {
} from "./chat-send-actions.ts";
import { setChatError } from "./chat-send-queue-state.ts";
import { handleSendChat } from "./chat-send-submit.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./chat-send-support.ts";
import { retireChatModelSelectionOwnership } from "./chat-session.ts";
import type { ChatPageHost } from "./chat-state-host.ts";
import {
@@ -59,7 +60,6 @@ import {
normalizeSidebarLayout,
openSlot,
} from "./sidebar-layout.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./steer-lifecycle.ts";
import { resetToolStream } from "./tool-stream.ts";
type ChatPageElement = {
ui/src/pages/chat/chat-state.test.ts
+1 -66
View File
@@ -1,4 +1,3 @@
import { readSessionMessageIdentity } from "@openclaw/gateway-client/browser";
import type { ReactiveController, ReactiveControllerHost } from "lit";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { GatewayBrowserClient } from "../../api/gateway.ts";
@@ -150,18 +149,6 @@ describe("canonical session message recovery", () => {
const { state } = createSessionEventState({
connected: false,
chatMessages: [originalPrompt],
chatQueue: [
{
id: "landed-steer",
text: "Steer prompt",
createdAt: 50,
kind: "steered",
pendingRunId: steerRunId,
sendRunId: steerRunId,
steerTargetRunId: activeRunId,
sessionKey: "agent:main:main",
},
],
chatRunId: activeRunId,
chatStream: null,
chatStreamSegments: [],
@@ -181,23 +168,12 @@ describe("canonical session message recovery", () => {
},
},
});
handlePageGatewayEvent(state, {
type: "event",
event: "chat",
payload: {
sessionKey: state.sessionKey,
runId: steerRunId,
state: "final",
},
});
expect(renderedTranscript(state)).toEqual([
{ role: "user", text: "Original prompt" },
{ role: "assistant", text: "Before steer." },
{ role: "user", text: "Steer prompt" },
]);
expect(state.chatRunId).toBe(activeRunId);
expect(state.chatQueue).toEqual([]);
const segmentsAfterRequestBoundary = state.chatStreamSegments;
const steerEvent = {
type: "event",
@@ -222,6 +198,7 @@ describe("canonical session message recovery", () => {
},
} satisfies Parameters<typeof handlePageGatewayEvent>[1];
handlePageGatewayEvent(state, steerEvent);
const segmentsAfterRequestBoundary = state.chatStreamSegments;
expect(state.chatStreamSegments).toBe(segmentsAfterRequestBoundary);
expect(
state.chatMessages.filter((message) => extractText(message) === "Steer prompt"),
@@ -556,48 +533,6 @@ describe("canonical session message recovery", () => {
expect(state.chatStream).toBe("Current partial reply");
});
it("orders an active queued turn before its landed steer", () => {
const activePrompt = {
id: "active-prompt",
text: "Keep this run active",
createdAt: 1,
sendRunId: "active-run",
sendState: "waiting-model" as const,
sessionKey: "main",
};
const { state } = createSessionEventState({
chatRunId: "active-run",
chatQueue: [
activePrompt,
{
id: "landed-steer-chip",
text: "Use the deployment plan",
createdAt: 2,
kind: "steered",
pendingRunId: "steer-request-run",
sendRunId: "steer-request-run",
steerTargetRunId: "active-run",
sessionKey: "main",
},
],
});
handlePageGatewayEvent(state, {
type: "event",
event: "chat",
payload: {
runId: "steer-request-run",
sessionKey: state.sessionKey,
state: "final",
},
});
expect(state.chatQueue).toEqual([activePrompt]);
expect(
state.chatMessages.map((message) => readSessionMessageIdentity(message)?.idempotencyKey),
).toEqual(["active-run:user", "steer-request-run:user"]);
});
it("renders distinct live peers immediately and coalesces their stale history", async () => {
let resolveHistory!: (result: {
messages: unknown[];
ui/src/pages/chat/chat-thread-build.ts
+1 -1
View File
@@ -27,6 +27,7 @@ import {
resolveWorkingProgress,
shouldRenderQueuedSendInThread,
} from "./chat-progress.ts";
import { chatMessagesContainQueuedSend } from "./chat-send-support.ts";
import {
coalesceToolActivityMessages,
groupMessages,
@@ -56,7 +57,6 @@ import {
type TurnInsertionBounds,
} from "./chat-thread-items.ts";
import { safeNormalizeMessage } from "./chat-turn-boundary.ts";
import { chatMessagesContainQueuedSend } from "./steer-lifecycle.ts";
import { resolveSystemNoticeKind } from "./system-notice-kinds.ts";
import { isLiveTerminalForRun } from "./terminal-message-identity.ts";
import {
ui/src/pages/chat/components/chat-composer-queue.ts
+9 -16
View File
@@ -9,7 +9,6 @@ import {
} from "../../../lib/chat/chat-queue-order.ts";
import type { ChatQueueItem } from "../../../lib/chat/chat-types.ts";
import { isSteerableQueuedMessage } from "../chat-queue.ts";
import { isInflightSteer, isSteeredQueueItem } from "../steered-chip.ts";
import { renderChatAuthorAvatar } from "./chat-author-avatar.ts";
type ChatQueueProps = {
@@ -102,9 +101,9 @@ function renderChatQueueItem(
) {
const stateLabel = sendStateLabel(item);
const failed = item.sendState === "failed" || item.sendState === "unconfirmed";
const steered = isSteeredQueueItem(item) && !failed;
const steerMode = item.queueMode === "steer";
const reconnecting = item.sendState === "waiting-reconnect";
const busy = item.sendState === "executing-command" || isInflightSteer(item);
const busy = item.sendState === "executing-command";
const editing = props.editingId === item.id;
const canSteer =
Boolean(props.canAbort && props.onQueueSteer) && isSteerableQueuedMessage(item) && !editing;
@@ -126,11 +125,9 @@ function renderChatQueueItem(
(item.attachments?.length
? t("chat.queue.imageCount", { count: String(item.attachments.length) })
: "");
const itemClass = `chat-queue__item${steered ? " chat-queue__item--steered" : ""}${
failed ? " chat-queue__item--failed" : ""
}${reconnecting ? " chat-queue__item--reconnect" : ""}${
editing ? " chat-queue__item--editing" : ""
}`;
const itemClass = `chat-queue__item${failed ? " chat-queue__item--failed" : ""}${
reconnecting ? " chat-queue__item--reconnect" : ""
}${editing ? " chat-queue__item--editing" : ""}`;
// Row order keeps the actions on the first flex line; the error wraps below
// them via flex-basis so failed rows grow by one line instead of a card.
return html`
@@ -200,14 +197,10 @@ function renderChatQueueItem(
${reconnecting
? html`<span class="chat-queue__dot" aria-hidden="true"></span>`
: html`<span class="chat-queue__icon" aria-hidden="true">
${failed ? icons.alertTriangle : steered ? icons.cornerDownRight : icons.outbox}
${failed ? icons.alertTriangle : icons.outbox}
</span>`}
${renderChatAuthorAvatar(item.sender)}
${steered
? html`<span class="chat-queue__badge chat-queue__badge--steered"
>${t("chat.queue.states.steering")}</span
>`
: nothing}
${steerMode ? html`<span class="chat-queue__badge">${t("chat.queue.steer")}</span>` : nothing}
${editing
? html`<span class="chat-queue__badge">${t("chat.queue.states.editing")}</span>`
: stateLabel
@@ -245,7 +238,7 @@ function renderChatQueueItem(
${failed && !editing && props.onQueueRetry
? html`
<button
class="chat-queue__retry"
class="chat-queue__action chat-queue__retry"
type="button"
aria-label=${t("chat.queue.retryQueuedMessage")}
@click=${() => props.onQueueRetry?.(item.id)}
@@ -258,7 +251,7 @@ function renderChatQueueItem(
${canSteer
? html`
<button
class="chat-queue__steer"
class="chat-queue__action"
type="button"
aria-label=${t("chat.queue.steerQueuedMessage")}
@click=${() => props.onQueueSteer?.(item.id)}
ui/src/pages/chat/composer-persistence.test.ts
+47 -12
View File
@@ -66,23 +66,58 @@ afterEach(() => {
});
describe("chat composer persistence", () => {
it("round-trips only the immutable steer run identity", () => {
it("loads legacy steer rows as generic mode-bearing sends and never rewrites old fields", () => {
const state = createState();
const steer: ChatQueueItem = {
const gatewayUrl = state.settings?.gatewayUrl;
const storageKey = storageKeyForGateway(gatewayUrl);
sessionStorage.setItem(
storageKey,
JSON.stringify({
version: 2,
gatewayOwner: gatewayUrl,
sessions: {
[`${state.sessionKey}\u0000agent:lily`]: {
queue: [
{
id: "steer-reload",
text: "keep the target",
createdAt: 1,
kind: "steered",
sendRunId: "steer-request",
sendState: "steering",
steerTargetRunId: "active-run",
},
],
updatedAt: 1,
},
},
}),
);
const restored = loadChatComposerSnapshot(state, state.sessionKey)?.queue[0];
expect(restored).toMatchObject({
id: "steer-reload",
text: "keep the target",
createdAt: 1,
kind: "steered",
queueMode: "steer",
sendRunId: "steer-request",
sendState: "unconfirmed",
steerTargetRunId: "active-run",
};
expect(admitStoredChatComposerQueueItem(state, state.sessionKey, steer)).toBe(true);
expect(loadChatComposerSnapshot(state, state.sessionKey)?.queue[0]).toMatchObject({
steerTargetRunId: "active-run",
});
expect(restored).not.toHaveProperty("kind");
expect(restored).not.toHaveProperty("steerTargetRunId");
expect(
updateStoredChatComposerQueueItem(
state,
state.sessionKey,
restored!,
{ ...restored!, text: "updated" },
restored?.agentId,
),
).toBe(true);
const written = sessionStorage.getItem(storageKey) ?? "";
expect(written).toContain('"queueMode":"steer"');
expect(written).not.toContain('"kind":"steered"');
expect(written).not.toContain("steerTargetRunId");
expect(written).not.toContain('"sendState":"steering"');
});
it("notifies stored outbox subscribers on draft presence transitions and queue writes", () => {
ui/src/pages/chat/composer-persistence.ts
+1 -2
View File
@@ -46,7 +46,6 @@ import {
durableComposerScopeIdentity,
type DurableChatComposerSnapshot,
} from "./durable-composer-persistence.ts";
import { isInflightSteer } from "./steered-chip.ts";
const CHAT_COMPOSER_DRAFT_PERSIST_DELAY_MS = 200;
export const CHAT_COMPOSER_DRAFT_STORAGE_ERROR =
@@ -140,7 +139,7 @@ function serializeQueueItem(item: ChatQueueItem): ChatQueueItem | null {
const sendState =
item.sendState === "sending"
? "waiting-reconnect"
: item.sendState === "executing-command" || isInflightSteer(item)
: item.sendState === "executing-command"
? "unconfirmed"
: item.sendState === "waiting-model"
? "failed"
ui/src/pages/chat/queued-message-edit.test.ts
+1 -2
View File
@@ -19,6 +19,7 @@ import {
steerQueuedChatMessage,
} from "./chat-send-actions.ts";
import { handleSendChat } from "./chat-send-submit.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./chat-send-support.ts";
import { listStoredChatOutboxes } from "./composer-persistence.ts";
import {
beginQueuedMessageEdit,
@@ -30,7 +31,6 @@ import {
QUEUED_MESSAGE_REORDER_CONFLICT_ERROR,
updateQueuedMessageEdit,
} from "./queued-message-edit.ts";
import { OFFLINE_QUEUE_STORAGE_ERROR } from "./steer-lifecycle.ts";
const SESSION_KEY = "agent:main";
@@ -471,7 +471,6 @@ describe("queued message edit round-trip", () => {
});
it.each([
{ label: "a steer chip", overrides: { kind: "steered" as const } },
{ label: "a local command", overrides: { localCommandName: "compact" } },
{ label: "a delivery-uncertain row", overrides: { sendState: "unconfirmed" as const } },
])("refuses to edit $label", ({ overrides }) => {
ui/src/pages/chat/steer-lifecycle.ts
-569
View File
@@ -1,569 +0,0 @@
import { asOptionalRecord, isRecord } from "@openclaw/normalization-core/record-coerce";
import type { QueueMode } from "../../../../packages/gateway-protocol/src/schema/logs-chat.js";
import type { SessionsListResult } from "../../api/types.ts";
import { setLastActiveSessionKey } from "../../app/settings.ts";
import { compareChatQueueOrder } from "../../lib/chat/chat-queue-order.ts";
import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts";
import { formatUiError } from "../../lib/format-error.ts";
import { resolveSessionDisplayName } from "../../lib/session-display.ts";
import { visibleSessionMatches } from "../../lib/sessions/index.ts";
import {
areUiSessionKeysEquivalent,
isUiGlobalSessionKey,
normalizeAgentId,
uiSessionRowMatchesSelectedChat,
} from "../../lib/sessions/session-key.ts";
import { showToast } from "../../lib/toast.ts";
import { generateUUID } from "../../lib/uuid.ts";
import {
getChatAttachmentDataUrl,
releaseChatAttachmentPayloads,
} from "./attachment-payload-store.ts";
import {
clearPendingQueueItemsForRun,
clearTransientQueuedMessageProjection,
excludeComposerAttachments,
removeQueuedMessageWithoutReleasing,
replacePendingQueuedMessageProjection,
setTransientQueuedMessageProjection,
type ChatQueueScopedSessionHost,
updateQueuedMessage,
writeChatQueueForScope,
} from "./chat-queue.ts";
import {
isTerminalFailureChatSendAck,
type ChatSendAck,
type TerminalFailureChatSendAck,
} from "./chat-send-ack.ts";
import type { ChatState } from "./chat-state-contract.ts";
import { readChatSessionProjectionScope, reduceChatSessionProjection } from "./history-merge.ts";
import {
isQueuedMessageBeingEdited,
QUEUED_MESSAGE_STEER_CONFLICT_ERROR,
} from "./queued-message-edit.ts";
import { hasDirectSessionRun } from "./run-lifecycle.ts";
import { scheduleChatScroll, type ChatScrollHost } from "./scroll.ts";
import { appendChatMessageToCache, readChatMessagesFromCache } from "./session-message-cache.ts";
import {
ackSteeredChip,
buildInflightSteerChip,
isAckedSteeredChip,
isSteeredQueueItem,
} from "./steered-chip.ts";
import { rolloverChatStream } from "./stream-causal-boundary.ts";
import { buildUserChatMessageContentBlocks } from "./user-message-content.ts";
type SteerLifecycleHost = ChatState &
ChatQueueScopedSessionHost & {
sessionsResult?: SessionsListResult | null;
};
type SteerSendHost = SteerLifecycleHost &
ChatScrollHost &
Parameters<typeof setLastActiveSessionKey>[0];
export type SteerSendDependencies = {
loadChatHistory: (host: SteerSendHost) => void;
resumeRestoredOutbox: (host: SteerSendHost, itemId: string) => void;
sendChatMessage: (
host: SteerSendHost,
message: string,
attachments: ChatAttachment[] | undefined,
options: {
canApplyError: () => boolean;
queueMode?: QueueMode;
runId: string;
expectedRunId?: string;
expectedLeafEntryId?: string | null;
replyToId?: string;
},
) => Promise<SteerChatSendResult>;
};
type SteerTarget = { runId: string; leafEntryId?: string | null };
function resolveSteerTarget(host: SteerLifecycleHost, item: ChatQueueItem): SteerTarget | null {
const matchingRows =
host.sessionsResult?.sessions.filter((row) =>
uiSessionRowMatchesSelectedChat(host, row.key, item.sessionKey ?? host.sessionKey),
) ?? [];
const serverRunIds = new Set(
matchingRows.flatMap((row) => (row.hasActiveRun ? (row.activeRunIds ?? []) : [])),
);
const durableRunId = item.kind === "steered" ? item.steerTargetRunId?.trim() : undefined;
if (item.kind === "steered" && !durableRunId) {
return null;
}
const runId =
durableRunId ||
host.chatRunId?.trim() ||
(serverRunIds.size === 1 ? [...serverRunIds][0] : undefined);
if (!runId) {
return null;
}
const activeRow = matchingRows.find((row) => row.activeRunIds?.includes(runId));
const displayedLeaf =
host.chatRunId?.trim() === runId ? host.chatDisplayedLeafEntryId : undefined;
const leafEntryId =
displayedLeaf === null ? null : displayedLeaf?.trim() || activeRow?.activeLeafEntryId;
return {
runId,
...(leafEntryId === null
? { leafEntryId: null }
: typeof leafEntryId === "string" && leafEntryId.trim()
? { leafEntryId: leafEntryId.trim() }
: {}),
};
}
type RejectedSteerChatSend = { kind: "rejected"; error: string };
type SteerChatSendResult = ChatSendAck | RejectedSteerChatSend | null;
function isRejectedSteerChatSend(result: SteerChatSendResult): result is RejectedSteerChatSend {
return result !== null && "kind" in result && result.kind === "rejected";
}
export const OFFLINE_QUEUE_STORAGE_ERROR =
"Could not store this message for reconnect. Free browser storage or reconnect before sending.";
const UNCONFIRMED_STEER_ERROR =
"Steer delivery could not be confirmed. Check the active run before retrying.";
const UNCONFIRMED_FOLLOW_UP_ERROR =
"Follow-up delivery could not be confirmed. Check the conversation before retrying.";
export function formatTerminalChatSendAckError(
ack: TerminalFailureChatSendAck,
context: "chat" | "detached" | "steer",
): string {
return ack.status === "error"
? context === "steer"
? "Steer failed before it reached the run; try again."
: "Chat failed before the run started; try again."
: context === "detached"
? "The active run ended before the detached message was accepted."
: context === "steer"
? "The active run ended before the steer message was accepted."
: "The run ended before the message was accepted.";
}
export function chatMessagesContainQueuedSend(
messages: unknown,
item: ChatQueueItem,
userRoleOnly = false,
): boolean {
return findQueuedSendMessageIndex(messages, item, userRoleOnly) >= 0;
}
function findQueuedSendMessageIndex(
messages: unknown,
item: ChatQueueItem,
userRoleOnly = false,
): number {
if (!item.sendRunId) {
return -1;
}
return (Array.isArray(messages) ? messages : []).findIndex((message) => {
if (!isRecord(message)) {
return false;
}
// Render retirement requires a user-role entry: an assistant entry can
// carry the same run key without proving the queued turn is visible.
const record = message;
if (userRoleOnly && record.role !== "user") {
return false;
}
const markerIdempotencyKey = asOptionalRecord(record["__openclaw"])?.idempotencyKey;
const idempotencyKey = markerIdempotencyKey ?? record.idempotencyKey;
return idempotencyKey === item.sendRunId || idempotencyKey === `${item.sendRunId}:user`;
});
}
function durableDeliveredAttachments(
attachments: readonly ChatAttachment[] | undefined,
): ChatAttachment[] | undefined {
return attachments?.flatMap((attachment) => {
// Composer uploads keep their bytes in the payload store; queue rows carry
// metadata only. Resolve through the store or attachment-only turns
// materialize empty and vanish at chip retirement.
const dataUrl = getChatAttachmentDataUrl(attachment);
if (!dataUrl) {
return [];
}
// Terminal retirement releases the queue-owned live blob. Pin synthetic
// transcript content to durable bytes before that ownership ends.
return [{ ...attachment, dataUrl, previewUrl: dataUrl }];
});
}
export function preserveQueuedUserTurn(state: SteerLifecycleHost, item: ChatQueueItem): void {
const runId = item.sendRunId;
const sessionKey = item.sessionKey ?? state.sessionKey;
if (!runId) {
return;
}
if (item.kind === "steered") {
// A started target may exist only as an optimistic queue row. Preserve it
// before the landed steer or stable history can invert the user turns.
const targetRunId = item.steerTargetRunId?.trim() || item.pendingRunId;
const target = state.chatQueue.find(
(candidate) => candidate.kind !== "steered" && candidate.sendRunId === targetRunId,
);
if (target) {
preserveQueuedUserTurn(state, target);
}
}
const content = buildUserChatMessageContentBlocks(
item.text,
durableDeliveredAttachments(item.attachments),
);
if (!content.length) {
return;
}
const userMessage = {
role: "user",
content,
timestamp: item.createdAt,
__openclaw: { idempotencyKey: `${runId}:user` },
};
if (visibleSessionMatches(state, sessionKey, item.agentId)) {
if (!chatMessagesContainQueuedSend(state.chatMessages, item, true)) {
const previousMessageCount = state.chatMessages.length;
const scope = readChatSessionProjectionScope(state, {
sessionKey,
agentId: item.agentId,
});
// Steer retirement and history recovery must retain the same pending
// entry; rendering a separate row loses it during a concurrent snapshot.
reduceChatSessionProjection(
state,
{ type: "sendPending", runId, message: userMessage },
{ scope },
);
if (
state.chatMessages.length > previousMessageCount &&
isSteeredQueueItem(item) &&
state.chatRunId
) {
rolloverChatStream(state, { runId: state.chatRunId, boundaryRunId: runId });
}
}
return;
}
if (!state.chatMessagesBySession) {
return;
}
const target = { sessionKey, agentId: item.agentId };
const cached = readChatMessagesFromCache(state.chatMessagesBySession, state, target);
if (!chatMessagesContainQueuedSend(cached, item, true)) {
appendChatMessageToCache(state.chatMessagesBySession, state, target, userMessage);
}
}
export function retireSteeredChipsForTerminalRun(
state: SteerLifecycleHost,
runId: string | undefined,
): void {
if (!runId) {
return;
}
for (const item of state.chatQueue) {
if (isAckedSteeredChip(item) && item.pendingRunId === runId) {
preserveQueuedUserTurn(state, item);
}
}
clearPendingQueueItemsForRun(state, runId);
}
export function retireSteeredChipsForRequestRun(
state: SteerLifecycleHost,
runId: string | undefined,
): void {
if (!runId) {
return;
}
const landed = state.chatQueue.filter(
(item) => isAckedSteeredChip(item) && item.sendRunId === runId,
);
for (const item of landed) {
preserveQueuedUserTurn(state, item);
}
if (landed.length > 0) {
const landedIds = new Set(landed.map((item) => item.id));
writeChatQueueForScope(
state,
state.sessionKey,
state.chatQueue.filter((item) => !landedIds.has(item.id)),
);
for (const item of landed) {
releaseChatAttachmentPayloads(excludeComposerAttachments(state, item.attachments));
}
}
}
export function retirePersistedSteeredChips(state: SteerLifecycleHost): void {
const retired = state.chatQueue.filter(
(item) =>
isAckedSteeredChip(item) && chatMessagesContainQueuedSend(state.chatMessages, item, true),
);
if (retired.length === 0) {
return;
}
const retiredIds = new Set(retired.map((item) => item.id));
writeChatQueueForScope(
state,
state.sessionKey,
state.chatQueue.filter((item) => !retiredIds.has(item.id)),
);
for (const item of retired) {
releaseChatAttachmentPayloads(excludeComposerAttachments(state, item.attachments));
}
}
function setChatError(host: SteerLifecycleHost, error: string | null): void {
const message = error === null ? null : formatUiError(error);
host.lastError = message;
host.chatError = message;
}
type ChatDeliveryFailureHost = Parameters<typeof visibleSessionMatches>[0] & {
lastError?: string | null;
chatError?: string | null;
sessionsResult?: SessionsListResult | null;
};
/**
* Terminal delivery failures must always end in a visible outcome. The pane
* showing this session keeps the inline chat error; after reconnect or alias
* drift the owning pane may no longer be on screen, so anything else surfaces
* a global toast naming the session instead of recording the error only on
* the queued row where nobody sees it.
*/
export function surfaceChatDeliveryFailure(
host: ChatDeliveryFailureHost,
sessionKey: string,
agentId: string | undefined,
error: string,
): void {
const message = formatUiError(error);
if (visibleSessionMatches(host, sessionKey, agentId)) {
host.lastError = message;
host.chatError = message;
return;
}
// Global rows are agent-scoped while sharing one "global" key, so an
// agent-less equivalence match could borrow another agent's label.
const scopedAgentId = agentId ? normalizeAgentId(agentId) : undefined;
const row = host.sessionsResult?.sessions.find(
(session) =>
areUiSessionKeysEquivalent(session.key, sessionKey) &&
(!isUiGlobalSessionKey(sessionKey) ||
!scopedAgentId ||
(session.agentId !== undefined && normalizeAgentId(session.agentId) === scopedAgentId)),
);
showToast({ message: `${resolveSessionDisplayName(sessionKey, row)}: ${message}` });
}
export async function sendQueuedChatMessageWithQueueMode(
host: SteerSendHost,
id: string,
queueMode: QueueMode | undefined,
dependencies: SteerSendDependencies,
): Promise<void> {
if (!host.connected || !hasDirectSessionRun(host)) {
return;
}
const isSteer = queueMode === "steer";
if (isSteer && isQueuedMessageBeingEdited(host, id)) {
setChatError(host, QUEUED_MESSAGE_STEER_CONFLICT_ERROR);
return;
}
const unconfirmedError = isSteer ? UNCONFIRMED_STEER_ERROR : UNCONFIRMED_FOLLOW_UP_ERROR;
const item = host.chatQueue.find(
(entry) =>
entry.id === id &&
!entry.pendingRunId &&
!entry.localCommandName &&
(entry.sendState === undefined || entry.sendState === "waiting-idle"),
);
if (!item) {
return;
}
const steerTarget = isSteer ? resolveSteerTarget(host, item) : null;
if (isSteer && !steerTarget) {
const error =
item.kind === "steered"
? "This restored steer has no original run target and cannot be retried safely."
: "The active run could not be identified uniquely. Review and retry.";
updateQueuedMessage(host, id, (entry) => ({ ...entry, sendError: error, sendState: "failed" }));
setChatError(host, error);
return;
}
const activeRunId = steerTarget?.runId ?? host.chatRunId;
const itemSessionKey = item.sessionKey ?? host.sessionKey;
const message = item.text.trim();
const attachments = item.attachments ?? [];
if (!message && attachments.length === 0) {
return;
}
// Claim the durable row before transport so a crash or ambiguous ACK cannot
// replay the original queued turn after active-run admission may have succeeded.
const claimed = updateQueuedMessage(host, id, (entry) => ({
...entry,
...(isSteer ? { kind: "steered" as const } : {}),
...(steerTarget
? {
steerTargetRunId: steerTarget.runId,
}
: {}),
sendError: unconfirmedError,
sendRunId: entry.sendRunId ?? generateUUID(),
sendState: "unconfirmed",
}));
if (!claimed?.sendRunId) {
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
const pendingItem: ChatQueueItem = {
id: item.id,
text: item.text,
createdAt: item.createdAt,
attachments: item.attachments,
replyToId: item.replyToId,
sendRunId: claimed.sendRunId,
sessionKey: claimed.sessionKey,
agentId: claimed.agentId,
...(claimed.steerTargetRunId ? { steerTargetRunId: claimed.steerTargetRunId } : {}),
};
const steeringChip = buildInflightSteerChip(pendingItem, claimed.sendRunId, activeRunId);
const pendingIndicator = isSteer
? steeringChip
: ({
...pendingItem,
sendState: "sending",
} satisfies ChatQueueItem);
const transientProjection = isSteer
? buildInflightSteerChip({ ...claimed, sendError: undefined }, claimed.sendRunId)
: { ...claimed, sendError: undefined, sendState: "sending" as const };
if (
!setTransientQueuedMessageProjection(host, itemSessionKey, transientProjection, item.agentId)
) {
const restored = updateQueuedMessage(host, id, () => item);
if (!restored) {
host.chatQueue = host.chatQueue.map((entry) => (entry.id === id ? item : entry));
}
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
return;
}
host.chatQueue = host.chatQueue.map((entry) => (entry.id === id ? pendingIndicator : entry));
const result = await dependencies.sendChatMessage(
host,
message,
attachments.length ? attachments : undefined,
{
canApplyError: () => visibleSessionMatches(host, itemSessionKey, item.agentId),
...(queueMode ? { queueMode } : {}),
...(claimed.replyToId ? { replyToId: claimed.replyToId } : {}),
...(steerTarget
? {
expectedRunId: steerTarget.runId,
...(steerTarget.leafEntryId !== undefined
? { expectedLeafEntryId: steerTarget.leafEntryId }
: {}),
}
: {}),
runId: claimed.sendRunId,
},
);
if (isSteer && activeRunId) {
replacePendingQueuedMessageProjection(
host,
itemSessionKey,
id,
activeRunId,
claimed,
item.agentId,
);
}
clearTransientQueuedMessageProjection(host, itemSessionKey, id, item.agentId);
const itemStillVisible = visibleSessionMatches(host, itemSessionKey, item.agentId);
if (!result) {
// A transport failure does not prove active-run admission was rejected. Keep the
// durable row parked so reconnect cannot replay it as a separate turn.
surfaceChatDeliveryFailure(host, itemSessionKey, item.agentId, unconfirmedError);
return;
}
if (isRejectedSteerChatSend(result)) {
const failed = updateQueuedMessage(host, id, (entry) => ({
...entry,
sendError: result.error,
sendState: "failed",
}));
surfaceChatDeliveryFailure(
host,
itemSessionKey,
item.agentId,
failed ? result.error : OFFLINE_QUEUE_STORAGE_ERROR,
);
return;
}
const ack = result;
if (isTerminalFailureChatSendAck(ack)) {
const restored = updateQueuedMessage(host, id, (entry) => ({
...item,
...(entry.attachments?.length ? { attachments: entry.attachments } : {}),
}));
if (!restored) {
surfaceChatDeliveryFailure(host, itemSessionKey, item.agentId, unconfirmedError);
} else {
surfaceChatDeliveryFailure(
host,
itemSessionKey,
item.agentId,
formatTerminalChatSendAckError(ack, isSteer ? "steer" : "chat"),
);
dependencies.resumeRestoredOutbox(host, id);
}
return;
}
const removed = removeQueuedMessageWithoutReleasing(host, id, itemSessionKey, item.agentId);
if (!removed) {
surfaceChatDeliveryFailure(host, itemSessionKey, item.agentId, unconfirmedError);
return;
}
const userTurnAlreadyVisible = chatMessagesContainQueuedSend(host.chatMessages, claimed, true);
if (isSteer && ack.status === "ok") {
preserveQueuedUserTurn(host, claimed);
if (itemStillVisible) {
dependencies.loadChatHistory(host);
}
}
if (isSteer && ack.status !== "ok" && itemStillVisible && !userTurnAlreadyVisible) {
// Key the chip to the run that will emit its terminal cleanup: the active
// run when it still owns the tab, else the steer's own gateway lifecycle
// (session-row-only runs, or the captured run ended mid-request).
const chipRunId = activeRunId && host.chatRunId === activeRunId ? activeRunId : ack.runId;
const steeredIndicator = ackSteeredChip(steeringChip, chipRunId);
writeChatQueueForScope(
host,
itemSessionKey,
[...host.chatQueue.filter((entry) => entry.id !== id), steeredIndicator].toSorted(
compareChatQueueOrder,
),
item.agentId,
);
} else {
releaseChatAttachmentPayloads(attachments);
}
if (itemStillVisible) {
setLastActiveSessionKey(host, itemSessionKey);
scheduleChatScroll(host);
}
}
export function steerQueuedChatMessage(
host: SteerSendHost,
id: string,
dependencies: SteerSendDependencies,
): Promise<void> {
return sendQueuedChatMessageWithQueueMode(host, id, "steer", dependencies);
}
ui/src/pages/chat/steered-chip.ts
-48
View File
@@ -1,48 +0,0 @@
import { hasNonEmptyString as hasString } from "@openclaw/normalization-core/string-coerce";
import type { ChatQueueItem } from "../../lib/chat/chat-types.ts";
type SteerState = { sendState: "steering" } | { sendState?: undefined; pendingRunId: string };
type SteeredQueueItem = ChatQueueItem & { kind: "steered" };
type SteeredChip = ChatQueueItem & { kind: "steered"; sendRunId: string } & SteerState;
type InflightSteerChip = SteeredChip & { sendState: "steering" };
type AckedSteeredChip = SteeredChip & { sendState?: undefined; pendingRunId: string };
export function isSteeredQueueItem(item: ChatQueueItem): item is SteeredQueueItem {
return item.kind === "steered";
}
export function isInflightSteer(item: ChatQueueItem): item is InflightSteerChip {
return isSteeredQueueItem(item) && hasString(item.sendRunId) && item.sendState === "steering";
}
export function isAckedSteeredChip(item: ChatQueueItem): item is AckedSteeredChip {
// An in-flight steer must never materialize: a rejected chat.send would
// otherwise leave a phantom user turn that a later retry duplicates.
return (
isSteeredQueueItem(item) &&
hasString(item.sendRunId) &&
item.sendState === undefined &&
hasString(item.pendingRunId)
);
}
export function buildInflightSteerChip(
item: ChatQueueItem,
sendRunId: string,
pendingRunId?: string | null,
): InflightSteerChip {
// The explicit marker keeps terminal and history retirement from treating
// the chip as acknowledged while chat.send is still unresolved.
return {
...item,
kind: "steered",
sendRunId,
...(pendingRunId ? { pendingRunId } : {}),
sendState: "steering",
};
}
export function ackSteeredChip(chip: InflightSteerChip, runKey: string): AckedSteeredChip {
const { sendState: _sendState, ...acked } = chip;
return { ...acked, pendingRunId: runKey };
}
ui/src/styles/components.css
+3 -19
View File
@@ -2949,10 +2949,6 @@ td.data-table-key-col {
color: var(--muted);
}
.chat-queue__item--steered {
background: var(--info-subtle);
}
/* The edited row keeps its slot so the operator can see where the corrected
message lands; the outline says the composer owns it right now.
Editing is an active state, not a failure, so it uses the informational
@@ -3035,10 +3031,6 @@ td.data-table-key-col {
color: var(--danger);
}
.chat-queue__item--steered .chat-queue__icon {
color: var(--info);
}
/* Reconnect rows mirror the connection pill's amber status dot so the queue
reads as quiet auto-retry status, not a failure needing action. */
.chat-queue__dot {
@@ -3069,11 +3061,6 @@ td.data-table-key-col {
white-space: nowrap;
}
.chat-queue__badge--steered {
background: color-mix(in srgb, var(--info) 16%, transparent);
color: var(--info);
}
.chat-queue__item--failed .chat-queue__badge {
background: color-mix(in srgb, var(--danger) 12%, transparent);
color: var(--danger);
@@ -3139,8 +3126,7 @@ td.data-table-key-col {
margin-left: auto;
}
.chat-queue__steer,
.chat-queue__retry {
.chat-queue__action {
display: inline-flex;
align-items: center;
gap: 4px;
@@ -3155,8 +3141,7 @@ td.data-table-key-col {
cursor: var(--cursor-action);
}
.chat-queue__steer svg,
.chat-queue__retry svg {
.chat-queue__action svg {
width: 12px;
height: 12px;
fill: none;
@@ -3166,8 +3151,7 @@ td.data-table-key-col {
stroke-linejoin: round;
}
.chat-queue__steer:hover:not(:disabled),
.chat-queue__retry:hover:not(:disabled) {
.chat-queue__action:hover:not(:disabled) {
background: color-mix(in srgb, var(--accent) 12%, transparent);
}