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

feat(maestro): server-side prompt queue for structured sessions

Browse Source 
Prompts sent while a turn is active are queued on the daemon instead of
in the browser's React state. The queue is persisted in the session
sidecar and drained as the next turn when the current one completes.

Fixes the bug where queued messages were delivered to the wrong session
when switching focus in the UI: the client-side queue captured the new
sessionId in its drain closure.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Miguel Palhas
2026-09-12 14:10:55 +01:00
co-authored by Claude Opus 4.6
parent cd5bbcfcb1
commit ebe25a1a55
17 changed files with 266 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/components/structured/StructuredPane.test.tsx
+5 -19
View File
@@ -308,7 +308,7 @@ describe("prompt and cancel", () => {
);
});
it("queues a second message while a turn is in flight, and sends it when the turn ends", async () => {
it("always POSTs a prompt even while a turn is active (server queues it)", async () => {
const { socket } = mount();
snapshot(socket, [], busy);
fireEvent.change(document.querySelector<HTMLTextAreaElement>(".structured-input")!, {
@@ -316,31 +316,17 @@ describe("prompt and cancel", () => {
});
fireEvent.submit(document.querySelector("form.structured-composer")!);
expect(sendStructuredPrompt).not.toHaveBeenCalled();
expect(document.querySelector(".st-row-queued")?.textContent).toBe("next up");
socket.emit({ type: "reduced_state", state: idle });
await waitFor(() => expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "next up"));
});
it("concatenates multiple queued messages into a single prompt", async () => {
it("renders server-side queued prompts from state", () => {
const { socket } = mount();
snapshot(socket, [], busy);
const input = document.querySelector<HTMLTextAreaElement>(".structured-input")!;
const form = document.querySelector("form.structured-composer")!;
snapshot(socket, [], { ...busy, queued_prompts: ["first", "second"] });
fireEvent.change(input, { target: { value: "first" } });
fireEvent.submit(form);
fireEvent.change(input, { target: { value: "second" } });
fireEvent.submit(form);
expect(sendStructuredPrompt).not.toHaveBeenCalled();
const queued = document.querySelectorAll(".st-row-queued");
expect(queued).toHaveLength(2);
socket.emit({ type: "reduced_state", state: idle });
await waitFor(() => expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "first\n\nsecond"));
expect(sendStructuredPrompt).toHaveBeenCalledTimes(1);
expect(queued[0].textContent).toBe("first");
expect(queued[1].textContent).toBe("second");
});
it("hands a failed send's draft back to the operator", async () => {
web/src/components/structured/StructuredPane.tsx
+7 -32
View File
@@ -48,11 +48,6 @@ import "./structured.css";
* count as following. */
const FOLLOW_BOTTOM_THRESHOLD_PX = 64;
interface QueuedPrompt {
text: string;
attachments: File[];
}
export interface StructuredPaneProps {
sessionId: Uuid;
/** Grabs the composer once mounted. A sidebar preview passes `false`. */
@@ -85,7 +80,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
const [expandedGroups, setExpandedGroups] = useState<ReadonlySet<string>>(new Set());
const [draft, setDraft] = useState("");
const [attachments, setAttachments] = useState<File[]>([]);
const [queue, setQueue] = useState<QueuedPrompt[]>([]);
const [uploading, setUploading] = useState(false);
const [configBusy, setConfigBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
@@ -100,7 +94,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
const submission = useRef(0);
const accepted = useRef<{ generation: number; socketGeneration: number } | null>(null);
const optimisticStart = useRef<number | null>(null);
const promptPending = useRef(false);
const [awaitingBegin, setAwaitingBegin] = useState<number | null>(null);
useImperativeHandle(
@@ -166,7 +159,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
const el = scrollRef.current;
if (!el || !followingBottom.current) return;
el.scrollTop = Math.max(0, el.scrollHeight - el.clientHeight);
}, [items, active, queue]);
}, [items, active, state.queued_prompts]);
useEffect(() => {
if (!active) return;
@@ -207,7 +200,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
setAwaitingBegin(generation);
optimisticStart.current = Date.now();
accepted.current = null;
promptPending.current = true;
setUploading(files.length > 0);
setDraft("");
setError(null);
@@ -224,7 +217,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
void send().then(
() => {
promptPending.current = false;
if (generation !== submission.current) return;
setUploading(false);
setAttachments((current) => (current === files ? [] : current));
@@ -235,7 +227,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
if (socketGeneration > submissionSocket) session.reconnect();
},
(e: unknown) => {
promptPending.current = false;
if (generation !== submission.current) return;
setAwaitingBegin((current) => (current === generation ? null : current));
optimisticStart.current = null;
@@ -251,26 +242,10 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
[sessionId, socketGeneration, session],
);
// Queued messages are concatenated into a single prompt so the agent sees
// one user turn instead of N back-to-back turns. The daemon rejects a
// second prompt while a turn is in flight (`409 turn_in_flight`), so the
// queue is client-local (docs/STRUCTURED-VIEW.md).
useEffect(() => {
if (active || promptPending.current || queue.length === 0) return;
const text = queue.map((q) => q.text).join("\n\n");
const files = queue.flatMap((q) => q.attachments);
setQueue([]);
dispatchPrompt(text, files);
}, [active, queue, dispatchPrompt]);
// The server accepts prompts at any time: if a turn is active the prompt is
// queued server-side and drained when the turn completes. The client always
// POSTs immediately — no local queue.
const submit = (text: string, files: File[]) => {
if (active || queue.length > 0) {
setQueue((current) => [...current, { text, attachments: files }]);
setAttachments([]);
setDraft("");
setError(null);
return;
}
dispatchPrompt(text, files);
};
@@ -365,10 +340,10 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
</ToolGroup>
),
)}
{queue.map((queued, index) => (
{(state.queued_prompts ?? []).map((text, index) => (
<div key={`queued-${index}`} className="st-row st-row-queued">
<div className="st-body">
<div className="st-user-text">{queued.text}</div>
<div className="st-user-text">{text}</div>
</div>
</div>
))}