naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(maestro): guard structured web submits

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 08:09:51 +01:00
parent c348381c5f
commit b86335ddf1
2 changed files with 165 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/structured.test.ts
+125
View File
@@ -147,6 +147,8 @@ describe("structured connection: turn/composer gating", () => {
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "do the thing");
expect(input.value).toBe("");
ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } });
ws.emit({ type: "reduced_state", state: idleState });
vi.mocked(sendStructuredPrompt).mockRejectedValueOnce(new Error("boom"));
input.value = "retry me";
@@ -158,6 +160,81 @@ describe("structured connection: turn/composer gating", () => {
conn.dispose();
});
it("disables immediately and dispatches only once before begin-turn is projected", () => {
vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {}));
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
ws.emit({ type: "reduced_state", state: idleState });
const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement;
const send = conn.el.querySelector(".structured-send") as HTMLButtonElement;
const form = conn.el.querySelector(".structured-composer") as HTMLFormElement;
input.value = "only once";
form.dispatchEvent(new SubmitEvent("submit", { bubbles: true, cancelable: true }));
form.dispatchEvent(new SubmitEvent("submit", { bubbles: true, cancelable: true }));
expect(input.disabled).toBe(true);
expect(send.disabled).toBe(true);
expect(sendStructuredPrompt).toHaveBeenCalledTimes(1);
expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "only once");
conn.dispose();
});
it("does not overwrite a newer operator draft when the matching request fails", async () => {
let reject!: (error: unknown) => void;
vi.mocked(sendStructuredPrompt).mockImplementationOnce(
() => new Promise((_, rejectPromise) => { reject = rejectPromise; }),
);
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
ws.emit({ type: "reduced_state", state: idleState });
const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement;
input.value = "first prompt";
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } });
ws.emit({ type: "reduced_state", state: idleState });
input.value = "newer draft";
reject(new Error("late failure"));
await Promise.resolve().then(() => Promise.resolve());
expect(input.value).toBe("newer draft");
expect(conn.el.textContent).toContain("late failure");
conn.dispose();
});
it("ignores an older failure after a newer submission generation starts", async () => {
let rejectFirst!: (error: unknown) => void;
vi.mocked(sendStructuredPrompt).mockImplementationOnce(
() => new Promise((_, rejectPromise) => { rejectFirst = rejectPromise; }),
);
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
ws.open();
ws.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
ws.emit({ type: "reduced_state", state: idleState });
const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement;
input.value = "first prompt";
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
ws.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } });
ws.emit({ type: "reduced_state", state: idleState });
input.value = "second prompt";
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
rejectFirst(new Error("stale failure"));
await Promise.resolve().then(() => Promise.resolve());
expect(sendStructuredPrompt).toHaveBeenCalledTimes(2);
expect(input.value).toBe("");
expect(input.disabled).toBe(true);
expect(conn.el.textContent).not.toContain("stale failure");
conn.dispose();
});
it("cancels the active turn via the cancel button", () => {
const ws = new FakeSocket();
const conn = mount({ sockets: [ws] });
@@ -327,6 +404,54 @@ describe("structured connection: reconnection/error/loading", () => {
conn.dispose();
});
it("keeps a local submission gated across a reconnect until begin-turn arrives", () => {
vi.mocked(sendStructuredPrompt).mockImplementationOnce(() => new Promise(() => {}));
const first = new FakeSocket();
const second = new FakeSocket();
const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
first.open();
first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
first.emit({ type: "reduced_state", state: idleState });
const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement;
input.value = "survive reconnect";
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
first.drop();
vi.advanceTimersByTime(1000);
second.open();
second.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
second.emit({ type: "reduced_state", state: idleState });
expect(input.disabled).toBe(true);
second.emit({ type: "reduced_state", state: { turn: { responding: true, thinking: false, active_tool_calls: [] } } });
second.emit({ type: "reduced_state", state: idleState });
expect(input.disabled).toBe(false);
conn.dispose();
});
it("releases an accepted submission from an idle snapshot after reconnect", async () => {
vi.mocked(sendStructuredPrompt).mockResolvedValueOnce(undefined);
const first = new FakeSocket();
const second = new FakeSocket();
const conn = mount({ sockets: [first, second], shouldReconnect: () => true });
first.open();
first.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
first.emit({ type: "reduced_state", state: idleState });
const input = conn.el.querySelector(".structured-input") as HTMLTextAreaElement;
input.value = "finishes while disconnected";
input.dispatchEvent(new KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true }));
await Promise.resolve().then(() => Promise.resolve());
first.drop();
vi.advanceTimersByTime(1000);
second.open();
second.emit({ type: "transcript_snapshot", snapshot: { cursor: 0, rows: [] } });
second.emit({ type: "reduced_state", state: idleState });
expect(input.disabled).toBe(false);
conn.dispose();
});
it("reports a daemon rejection reason and does not schedule a retry", () => {
const only = new FakeSocket();
const conn = mount({ sockets: [only], shouldReconnect: () => true });
web/src/structured.ts
+40 -10
View File
@@ -109,6 +109,10 @@ export function createStructuredConnection(
let state: ReducedState = { turn: idleTurn() };
let pendingApprovalRowId: string | null = null;
let latestSession: Session | undefined = store?.snapshot().sessions.get(sessionId);
let submissionGeneration = 0;
let awaitingBeginGeneration: number | null = null;
let acceptedSubmission: { generation: number; socketGeneration: number } | null = null;
let socketGeneration = 0;
function lastRowIdOfKind(kind: TranscriptRowKind): string | null {
for (let i = rows.length - 1; i >= 0; i--) if (rows[i].kind === kind) return rows[i].id;
@@ -217,14 +221,23 @@ export function createStructuredConnection(
}
function renderComposerState(): void {
const active = state.turn.responding || state.turn.thinking;
const active = state.turn.responding || state.turn.thinking || awaitingBeginGeneration !== null;
input.disabled = active;
sendBtn.disabled = active;
cancelBtn.hidden = !active;
cancelBtn.hidden = !state.turn.responding && !state.turn.thinking;
}
function applyState(next: ReducedState): void {
function applyState(next: ReducedState, connectionGeneration: number): void {
state = next;
if (state.turn.responding || state.turn.thinking) {
awaitingBeginGeneration = null;
} else if (
awaitingBeginGeneration !== null
&& acceptedSubmission?.generation === awaitingBeginGeneration
&& connectionGeneration > acceptedSubmission.socketGeneration
) {
awaitingBeginGeneration = null;
}
const newPendingId = state.pending_approval ? lastRowIdOfKind("approval") : null;
if (newPendingId !== pendingApprovalRowId) {
const prev = pendingApprovalRowId;
@@ -268,14 +281,30 @@ export function createStructuredConnection(
e.preventDefault();
const text = input.value;
if (!text.trim() || input.disabled) return;
const generation = ++submissionGeneration;
awaitingBeginGeneration = generation;
acceptedSubmission = null;
input.value = "";
autoResize();
clearError();
sendStructuredPrompt(sessionId, text).catch((e: unknown) => {
input.value = text;
autoResize();
showError(String(e));
});
renderComposerState();
sendStructuredPrompt(sessionId, text).then(
() => {
if (generation === submissionGeneration) {
acceptedSubmission = { generation, socketGeneration };
}
},
(e: unknown) => {
if (generation !== submissionGeneration) return;
if (awaitingBeginGeneration === generation) awaitingBeginGeneration = null;
if (input.value === "") {
input.value = text;
autoResize();
}
showError(String(e));
renderComposerState();
},
);
});
cancelBtn.addEventListener("click", () => {
clearError();
@@ -303,6 +332,7 @@ export function createStructuredConnection(
let disposed = false;
function connect(): void {
const connectionGeneration = ++socketGeneration;
ws = socketFactory(sessionId);
ws.onopen = () => {
reconnectMs = INITIAL_RECONNECT_MS;
@@ -321,11 +351,11 @@ export function createStructuredConnection(
applySnapshot(msg.snapshot);
break;
case "reduced_state":
applyState(msg.state);
applyState(msg.state, connectionGeneration);
break;
case "projection_update":
applyDelta(msg.transcript);
applyState(msg.state);
applyState(msg.state, connectionGeneration);
break;
}
};