trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns14g rEventsFlows

/talk takes a message while the agent is working: queue it next, or steer the current turn #8893

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/8819-talk-takes-a-message-while-the-agent-is → main 3 commits fork 2 threads · 2 unresolved

  • Merged. scout/8819-talk-takes-a-message-while-the-agent-is was merged into main as 65654ce.

Review

Sign in to review this change.
15 files changed +670 −42
docs/design/areas/talk.md +1
@@ -31,3 +31,4 @@
3131 - `.talk-attach` — the "Attach" file-picker label in the composer bar, a `.btn` that stays hidden until the script runs (the picker needs it).
3232
3333 Under a reply, `.talk-cost` is one quiet line with the turn's dollars; above the log, `.talk-context` says how full the context is and, from 80 %, suggests a new conversation. Both are `small muted` text, no new colour.
34 - `.talk-queued` — a message waiting for the running turn: the person bubble, dimmed with a dashed border, its "Queued" line and Cancel under it.
docs/features/talk.md +9 −1
@@ -41,6 +41,14 @@ A turn can be ended mid-reply (`stopTalk` in `src/actions/talk.ts`, `stopTurn` i
4141
4242 It checks identity, same origin and token scope. Viewers of `/talk` see the `Stopped.` turn arrive over the stream.
4343
44 ## Queue and Steer while an answer runs (#8819)
45
46 The composer stays open while the agent works. `say()` in `src/crew/conversation.ts` takes `during`: `inject` (the old behaviour, still the default for the API without `during`, MCP and plain `ssh talk`), `queue` (kept in memory for after the turn; up to 5; each runs as its own turn, in order) or `interrupt` (stops the turn, keeps its words and the `Stopped.` line, and sends the message first). Stop clears the queue; a forge restart drops it. A file attached to a queued message is held: it cannot be removed (409) or sent with another message (`file_taken`), and one gone by the time it runs is left out.
47
48 - **Web:** while an answer runs Send reads Queue and Steer and Stop appear. A queued message shows dimmed, "Queued", with Cancel (a post form to `POST /talk/queue/:id/cancel`, so it works without the script); every open tab sees it over the stream (`queued` frame). Cancel puts the text back in the box.
49 - **API:** `POST /api/v1/conversation/messages` with `during: "queue" | "steer"` (anything else is 422 `invalid_during`; a sixth queued message is 429 `queue_full`); `DELETE /api/v1/conversation/queue/:id` (404 for an id that is not yours or already went); `GET /api/v1/conversation/turns` lists `queued`.
50 - **Terminal:** `ssh <forge> talk /queue <words>`, `talk /steer <words>`, `talk /queued` (ids and words), `talk /unqueue <id>`; they work with `--attach` too.
51
4452 ## Chat features: have, adding, not adding (#8752)
4553
4654 Prior art read: Vercel AI Elements (Apache-2.0 React on shadcn; anatomy and behaviour ported, no code copied), AI SDK UI, ChatGPT and Claude.ai. No React: the page stays server-drawn with a small script.
@@ -58,7 +66,7 @@ Prior art read: Vercel AI Elements (Apache-2.0 React on shadcn; anatomy and beha
5866 | Thinking as one closed line; a steps line under a reply ("read 3 files, opened #12"); links to each issue, pull, file or app the turn opened or changed | this pull | AI Elements `Reasoning`, `ChainOfThought` |
5967 | Many conversations: new, list, search, rename, archive, URL each | next pull | needs a migration: today one row per person |
6068 | Edit and resend, regenerate, Retry (see below) | have (#8817) | replaced turns are kept and marked, not deleted |
61 | Queue | next pull | needs a queue on `crew/conversation.ts` |
69 | Queue and Steer while an answer runs | #8819 | `Queue` / `PromptInput` |
6270 | Cost under each reply and how full the context is (header, and in the terminal) | this pull | the run's own usage record, never an estimate |
6371 | Attachments, tables, plan line | later | need storage / token data / markdown tables |
6472 | Code highlighting and tables in replies | this pull (#8816) | a reply is stored whole and drawn whole, so no half-open fence is ever tokenised |
src/actions/talk.ts +12 −3
@@ -9,7 +9,7 @@ import type { App } from "../app.ts";
99 import type { ConversationFile } from "../core/conversation_files.ts";
1010 import { RateLimiter } from "../core/ratelimit.ts";
1111 import type { User } from "../core/users.ts";
12 import { conversationOptions, redo, say, stopTurn, type RedoHow, type RedoRefusal, type SayRefusal, type Said } from "../crew/conversation.ts";
12 import { cancelQueued, queuedOf, conversationOptions, redo, say, stopTurn, type During, type RedoHow, type RedoRefusal, type SayRefusal, type Said } from "../crew/conversation.ts";
1313 import { err, ok, type Result } from "../kernel/result.ts";
1414
1515 /** Messages a person may send in a minute. */
@@ -29,6 +29,8 @@ export const TALK_WORDS: Readonly<Record<TalkRefusal, readonly [number, string]>
2929 off: [503, "the forge's agent is not running here"],
3030 running: [409, "the forge is still answering; wait for it to finish"],
3131 nothing: [409, "there is nothing here to do that to"],
32 file_taken: [409, "that file already waits in another message"],
33 queue_full: [429, "five messages are already waiting; cancel one or wait for the answer"],
3234 rate_limited: [429, `at most ${TALK_PER_MINUTE} messages a minute`],
3335 };
3436
@@ -48,12 +50,12 @@ const watch = (app: App, said: Said): void => {
4850 said.done?.catch((error: unknown) => app.log.error("conversation turn crashed", { error: String(error) }));
4951 };
5052
51 export function talk(app: App, user: User | undefined, text: string, files: readonly ConversationFile[] = []): Result<Said, TalkRefusal> {
53 export function talk(app: App, user: User | undefined, text: string, files: readonly ConversationFile[] = [], during: During = "inject"): Result<Said, TalkRefusal> {
5254 if (user === undefined) return err("unauthorized");
5355 const options = conversationOptions(app);
5456 if (options === undefined) return err("off");
5557 if (!take(app, user)) return err("rate_limited");
56 const said = say(app, options, user.id, text, files);
58 const said = say(app, options, user.id, text, files, during);
5759 if (said.ok) watch(app, said.value);
5860 return said;
5961 }
@@ -75,3 +77,10 @@ export function stopTalk(app: App, user: User | undefined): Result<true, "unauth
7577 if (conversationOptions(app) === undefined) return err("off");
7678 return stopTurn(app, user.id) ? ok(true) : err("idle");
7779 }
80
81 /** Take back a message waiting for the running turn; false when it already went. */
82 export const cancelTalk = (app: App, user: User | undefined, queuedId: string): boolean => user !== undefined && cancelQueued(app, user.id, queuedId);
83
84 /** The person's waiting messages, oldest first. */
85 export const queuedTalk = (app: App, user: User | undefined): readonly { id: string; text: string }[] =>
86 user === undefined ? [] : queuedOf(app, user.id).map((q) => ({ id: q.id, text: q.text }));
src/contract/openapi/routes/deleteApiV1ConversationQueueId.ts added +13
@@ -0,0 +1,13 @@
1 import type { Route } from "../shared.ts";
2
3 export const route: Route = {
4 method: "delete",
5 path: "/api/v1/conversation/queue/{id}",
6 summary: "Take back a message queued for after the running turn; one that already went stays sent",
7 auth: true,
8 reply: { ok: "boolean" },
9 refusals: {
10 not_found: "no such waiting message of yours (404)",
11 forbidden: "the request is cross-origin, or the token may only read",
12 },
13 };
src/contract/openapi/routes/index.ts +1
@@ -5,6 +5,7 @@ export { route as deleteApiV1AdminInvitesId } from "./deleteApiV1AdminInvitesId.
55 export { route as deleteApiV1AgentsUsername } from "./deleteApiV1AgentsUsername.ts";
66 export { route as deleteApiV1BoardsSlug } from "./deleteApiV1BoardsSlug.ts";
77 export { route as deleteApiV1ConversationFilesId } from "./deleteApiV1ConversationFilesId.ts";
8 export { route as deleteApiV1ConversationQueueId } from "./deleteApiV1ConversationQueueId.ts";
89 export { route as deleteApiV1OrgsOrgTeamsTeam } from "./deleteApiV1OrgsOrgTeamsTeam.ts";
910 export { route as deleteApiV1OrgsOrgTeamsTeamMembersUsername } from "./deleteApiV1OrgsOrgTeamsTeamMembersUsername.ts";
1011 export { route as deleteApiV1ReposOwnerName } from "./deleteApiV1ReposOwnerName.ts";
src/contract/openapi/routes/postApiV1ConversationMessages.ts +4 −1
@@ -3,13 +3,16 @@ import type { Route } from "../shared.ts";
33 export const route: Route = {
44 method: "post",
55 path: "/api/v1/conversation/messages",
6 summary: "Say something to the forge's own agent, as the signed-in person: it answers through the forge's MCP with that person's permissions, resuming their conversation; a message sent while it answers is steered into the turn; answers 202 at once, read the answer from the turns; at most 12 a minute per person",
6 summary: "Say something to the forge's own agent, as the signed-in person: it answers through the forge's MCP with that person's permissions, resuming their conversation; a message sent while it answers is handed to the turn, or with `during` (optional) set to `queue` waits for it (at most 5), or set to `steer` stops it and goes at once; answers 202 at once, read the answer from the turns; at most 12 a minute per person",
77 auth: true,
88 status: 202,
99 body: { text: "string" },
1010 reply: { how: "string" },
1111 refusals: {
1212 invalid_text: "text is not a string of 1 to 20000 characters",
13 invalid_during: "during is not queue or steer",
14 file_taken: "an attached file already waits in a queued message (409)",
15 queue_full: "five messages already wait (429)",
1316 too_large: "the body is over 128000 bytes (413)",
1417 no_account: "no Claude account is granted to the person (403)",
1518 resting: "the account that serves the person is resting or out of budget (429)",
src/crew/conversation.ts +103 −6
@@ -20,7 +20,7 @@ import { join } from "node:path";
2020 import type { App } from "../app.ts";
2121 import { dailyLeft } from "../core/account_ceiling.ts";
2222 import { appendTurn, lastExchange, listTurns, openConversation, removeTurn, restoreTurns, setAsideFrom, setClaudeSession, type Conversation } from "../core/conversations.ts";
23 import { attach, deliver, NOTHING, release, type ConversationFile, type Delivery } from "../core/conversation_files.ts";
23 import { attach, deliver, fileOwnedBy, NOTHING, release, type ConversationFile, type Delivery } from "../core/conversation_files.ts";
2424 import { carryWrites, forgetWrites, unended } from "../core/conversation_writes.ts";
2525 import { cancelHeld, WORDS_MAX } from "../core/fence.ts";
2626 import { thinkingOf } from "../core/talk_work.ts";
@@ -113,6 +113,8 @@ export function stopTurn(app: App, userId: string): boolean {
113113 const conversation = openConversation(app.db, userId);
114114 const stop = stoppers.get(conversation.id)?.stop;
115115 if (stop === undefined) return false;
116 // Stop is the person calling it off: what waited behind the turn goes too (interrupting keeps it, `say`).
117 queues.delete(conversation.id);
116118 // Cancelled first: no Confirm may land while the process is going down.
117119 cancelHeld(app.db, `conversation:${conversation.id}`);
118120 stop();
@@ -192,22 +194,114 @@ export function mcpConfig(localUrl: string): string {
192194 });
193195 }
194196
195 export type SayRefusal = "unauthorized" | "empty" | "too_long" | "no_account" | "resting" | "busy" | "answering";
197 export type SayRefusal = "unauthorized" | "empty" | "too_long" | "no_account" | "resting" | "busy" | "answering" | "queue_full" | "file_taken";
196198
197199 export interface Said {
198 /** `started` a new turn, or `steered` one already running. */
199 readonly how: "started" | "steered";
200 /** `started` a new turn, `steered` words into the one running, `queued` for after it, or `interrupted` it to send these at once. */
201 readonly how: "started" | "steered" | "queued" | "interrupted";
200202 /** Settles when the turn does; only a started turn has one. */
201203 readonly done?: Promise<void>;
204 /** The queue id of a `queued` message, to cancel it by. */
205 readonly queued?: string;
202206 }
203207
204 /** Store a person's message and answer it: steer the turn that runs, or start one. */
205 export function say(app: App, options: ConversationOptions, userId: string, text: string, files: readonly ConversationFile[] = []): Result<Said, SayRefusal> {
208 /**
209 * What a message does when a turn is running (#8819): `inject` hands the words
210 * to the turn (the default, for callers that never asked otherwise), `queue`
211 * keeps them for after it, `interrupt` stops the turn and sends them at once.
212 * With no turn running, all three start one.
213 */
214 export type During = "inject" | "queue" | "interrupt";
215
216 /** A message waiting for the running turn to end. Held in memory: a restart drops it. */
217 export interface Queued {
218 readonly id: string;
219 readonly text: string;
220 readonly files: readonly ConversationFile[];
221 }
222
223 /** Most messages that may wait behind one turn. */
224 export const QUEUE_MAX = 5;
225
226 const queues = new Map<string, Queued[]>();
227
228 /** The person's messages waiting for their running turn, oldest first. */
229 export const queuedOf = (app: App, userId: string): readonly Queued[] => queues.get(openConversation(app.db, userId).id) ?? [];
230
231 /** Whether a message waiting in a queue holds this file: it cannot be removed or sent twice meanwhile. */
232 export const fileQueued = (fileId: string): boolean => [...queues.values()].some((l) => l.some((q) => q.files.some((f) => f.id === fileId)));
233
234 /** Take back a waiting message; false when it is gone (already sent, or never theirs). */
235 export function cancelQueued(app: App, userId: string, queuedId: string): boolean {
236 const key = openConversation(app.db, userId).id;
237 const list = queues.get(key) ?? [];
238 const at = list.findIndex((q) => q.id === queuedId);
239 if (at < 0) return false;
240 list.splice(at, 1);
241 if (list.length === 0) queues.delete(key);
242 return true;
243 }
244
245 /**
246 * The turn ended: send the next waiting message as a turn of its own. A
247 * message that cannot start ends the wait with one failed line, for the rest
248 * would fail the same way.
249 */
250 function drain(app: App, options: ConversationOptions, conversation: Conversation): void {
251 const list = queues.get(conversation.id);
252 const next = list?.shift();
253 if (list?.length === 0) queues.delete(conversation.id);
254 if (next === undefined) return;
255 const userId = conversation.userId;
256 const current = openConversation(app.db, userId);
257 if (current.id !== conversation.id) return void queues.delete(conversation.id);
258 const stored = appendTurn(app.db, userId, current.id, { role: "person", text: next.text });
259 if (!stored.ok) return drain(app, options, current);
260 // A file taken back or sent some other way while this waited is left out, not sent twice.
261 const files = next.files.filter((f) => {
262 const now = fileOwnedBy(app.db, userId, f.id);
263 return now !== undefined && now.turnId === undefined;
264 });
265 attach(app.db, files, stored.value.id);
266 const started = begin(app, options, current, next.text, 0, current.claudeSessionId || undefined, deliver(app.config.dataDir, userId, files));
267 if (started.ok) {
268 started.value.catch((error: unknown) => app.log.error("conversation turn crashed", { error: String(error) }));
269 return;
270 }
271 release(app.db, stored.value.id);
272 queues.delete(current.id);
273 appendTurn(app.db, userId, current.id, { role: "agent", text: failureLine("error"), failed: true });
274 }
275
276 /** Store a person's message and answer it: steer, queue or interrupt the turn that runs (`During`), or start one. */
277 export function say(
278 app: App,
279 options: ConversationOptions,
280 userId: string,
281 text: string,
282 files: readonly ConversationFile[] = [],
283 during: During = "inject",
284 ): Result<Said, SayRefusal> {
206285 const user = getUserById(app.db, userId);
207286 if (user === undefined) return err("unauthorized");
208287 if (text.trim() === "") return err("empty");
209288 if (text.length > WORDS_MAX) return err("too_long");
210289 const conversation = openConversation(app.db, userId);
290 if (during !== "inject" && live.has(conversation.id)) {
291 const list = queues.get(conversation.id) ?? [];
292 const waiting: Queued = { id: id(), text, files };
293 if (files.some((f) => fileQueued(f.id))) return err("file_taken");
294 if (during === "queue") {
295 if (list.length >= QUEUE_MAX) return err("queue_full");
296 queues.set(conversation.id, [...list, waiting]);
297 return ok({ how: "queued", queued: waiting.id });
298 }
299 // Ahead of the others: it is what the person wants now. The stopped turn's end sends it.
300 queues.set(conversation.id, [waiting, ...list]);
301 stoppers.get(conversation.id)?.stop();
302 cancelHeld(app.db, `conversation:${conversation.id}`);
303 return ok({ how: "interrupted" });
304 }
211305 const stored = appendTurn(app.db, userId, conversation.id, { role: "person", text });
212306 if (!stored.ok) return err("too_long");
213307 attach(app.db, files, stored.value.id);
@@ -494,6 +588,9 @@ function begin(
494588 } finally {
495589 if (live.get(conversation.id) === runId) live.delete(conversation.id);
496590 if (stoppers.get(conversation.id)?.runId === runId) stoppers.delete(conversation.id);
591 // A restart leaves the next message unsent (its turn would be cut again); anything else sends it.
592 if (options.signal.aborted) queues.delete(conversation.id);
593 else drain(app, options, conversation);
497594 }
498595 })();
499596 return ok(turn);
src/http/api/conversation_talk.ts +18 −5
@@ -9,14 +9,14 @@
99
1010 import type { App } from "../../app.ts";
1111 import { RateLimiter } from "../../core/ratelimit.ts";
12 import { stopTalk, TALK_WORDS, talk } from "../../actions/talk.ts";
12 import { cancelTalk, stopTalk, TALK_WORDS, talk } from "../../actions/talk.ts";
1313 import { FILE_WORDS, fileBytes, fileOwnedBy, FILES_PER_MESSAGE, IMAGE_MAX, pendingAmong, removePending, storeFile } from "../../core/conversation_files.ts";
1414 import { DEFAULT_PAGE, listTurns, MAX_PAGE, openConversation } from "../../core/conversations.ts";
1515 import { WORDS_MAX } from "../../core/fence.ts";
1616 import { notePhase, takeSessionTime } from "../../core/route_timing.ts";
1717 import { scopeAllows } from "../../core/tokens.ts";
1818 import { worksOf } from "../../core/talk_work.ts";
19 import { turnRunning } from "../../crew/conversation.ts";
19 import { fileQueued, queuedOf, turnRunning } from "../../crew/conversation.ts";
2020 import { crossOrigin, type Handler } from "../context.ts";
2121 import { readBounded } from "../limits.ts";
2222 import { apiError, boundedBody, forbidden, json, unauthorized } from "../respond.ts";
@@ -62,12 +62,24 @@ export function conversationTalkRoutes(router: Router<Handler>): void {
6262 return apiError("invalid_files", 422, `files is a list of at most ${FILES_PER_MESSAGE} attachment ids`);
6363 const files = pendingAmong(ctx.app.db, user.id, ids as string[]);
6464 if (!files.ok) return apiError("not_found", 404, FILE_WORDS.not_found[1]);
65 const said = talk(ctx.app, user, text, files.value);
65 // What to do when a turn runs: absent hands the words to it; `queue` waits for it; `steer` stops it and sends these at once.
66 const during = input["during"];
67 if (during !== undefined && during !== "queue" && during !== "steer") return apiError("invalid_during", 422, "during is queue or steer");
68 const said = talk(ctx.app, user, text, files.value, during === "queue" ? "queue" : during === "steer" ? "interrupt" : "inject");
6669 if (!said.ok) {
6770 const [status, words] = TALK_WORDS[said.error];
6871 return json({ error: said.error, message: words }, status, said.error === "rate_limited" || said.error === "busy" ? { "retry-after": "60" } : {});
6972 }
70 return json({ how: said.value.how }, 202);
73 return json(said.value.queued === undefined ? { how: said.value.how } : { how: said.value.how, queued: said.value.queued }, 202);
74 });
75
76 router.add("DELETE", "/api/v1/conversation/queue/:id", (ctx) => {
77 const user = ctx.auth.user;
78 if (user === undefined) return unauthorized();
79 if (crossOrigin(ctx.request, ctx.url, ctx.app.log)) return forbidden();
80 if (ctx.auth.via === "token" && !scopeAllows(ctx.auth.scope ?? "ro", "rw")) return forbidden();
81 if (!cancelTalk(ctx.app, user, ctx.params["id"] ?? "")) return apiError("not_found", 404, "that message already went, or was never waiting");
82 return json({ ok: true });
7183 });
7284
7385 router.add("POST", "/api/v1/conversation/stop", (ctx) => {
@@ -125,6 +137,7 @@ export function conversationTalkRoutes(router: Router<Handler>): void {
125137 if (user === undefined) return unauthorized();
126138 if (crossOrigin(ctx.request, ctx.url, ctx.app.log)) return forbidden();
127139 if (ctx.auth.via === "token" && !scopeAllows(ctx.auth.scope ?? "ro", "rw")) return forbidden();
140 if (fileQueued(ctx.params["id"] ?? "")) return apiError("file_taken", 409, "that file waits in a queued message; cancel the message first");
128141 if (!removePending(ctx.app.db, ctx.app.config.dataDir, user.id, ctx.params["id"] ?? "")) return apiError("not_found", 404, FILE_WORDS.not_found[1]);
129142 return json({ ok: true });
130143 });
@@ -142,6 +155,6 @@ export function conversationTalkRoutes(router: Router<Handler>): void {
142155 const all = turns.ok ? turns.value : [];
143156 const works = worksOf(ctx.app.db, user.id, conversation.id, all);
144157 const shown = all.map((t) => (t.role === "agent" ? { ...t, work: works.get(t.id) ?? null } : t));
145 return json({ running: turnRunning(ctx.app, user.id), turns: shown });
158 return json({ running: turnRunning(ctx.app, user.id), queued: queuedOf(ctx.app, user.id).map((q) => ({ id: q.id, text: q.text })), turns: shown });
146159 });
147160 }
src/http/server.ts +2 −1
@@ -124,7 +124,7 @@ import {
124124 verifyPage,
125125 themeSubmit,
126126 } from "../web/pages/site.ts";
127 import { talkAgainSend, talkConfirm, talkSend, talkStream, talkView } from "../web/pages/talk.ts";
127 import { talkAgainSend, talkConfirm, talkQueueCancel, talkSend, talkStream, talkView } from "../web/pages/talk.ts";
128128 import {
129129 recoverFinish,
130130 recoverOptions,
@@ -842,6 +842,7 @@ export function buildRouter(): Router<Handler> {
842842 router.add("POST", "/talk", talkSend);
843843 router.add("GET", "/talk/stream", talkStream);
844844 router.add("POST", "/talk/again/:how", talkAgainSend);
845 router.add("POST", "/talk/queue/:id/cancel", talkQueueCancel);
845846 router.add("POST", "/talk/confirm/:id", talkConfirm);
846847 router.add("POST", "/signin/passkey/options", signinOptions);
847848 router.add("POST", "/signin/passkey", signinPasskey);
src/tui/ssh-talk.ts +28 −4
@@ -8,6 +8,10 @@
88 * `--stop`) ends the running answer, as Stop does on /talk: what the agent
99 * had said stays, its processes end, a held write is cancelled.
1010 *
11 * While an answer runs, `talk /queue <words>` keeps the words for after it
12 * (`talk /queued` lists what waits, `talk /unqueue <id>` takes one back) and
13 * `talk /steer <words>` stops the answer and sends them at once (#8819).
14 *
1115 * `talk-again edit <words>`, `talk-again regenerate` and `talk-again retry`
1216 * are the same three actions the page has (#8817), through `talkAgain`.
1317 *
@@ -18,7 +22,7 @@
1822 * the page's attachments are (`core/conversation_files.ts`).
1923 */
2024
21 import { TALK_WORDS, stopTalk, talk, talkAgain } from "../actions/talk.ts";
25 import { cancelTalk, queuedTalk, TALK_WORDS, stopTalk, talk, talkAgain } from "../actions/talk.ts";
2226 import { FILE_WORDS, IMAGE_MAX, pendingAmong, removePending, storeFile } from "../core/conversation_files.ts";
2327 import { contextOf, contextText, dollars, turnCosts } from "../core/conversation_cost.ts";
2428 import { openConversation } from "../core/conversations.ts";
@@ -42,6 +46,12 @@ export function isStop(words: string): boolean {
4246 return words.trim() === "/stop" || words.trim() === "--stop";
4347 }
4448
49 /** `/queue <words>` or `/steer <words>`: what to do with the words while an answer runs; undefined for plain words. */
50 export function parseDuring(words: string): { during: "queue" | "interrupt"; words: string } | undefined {
51 const m = /^\/(queue|steer)(?:[ \t]+([\s\S]*))?$/.exec(words.trim());
52 return m === null ? undefined : { during: m[1] === "queue" ? "queue" : "interrupt", words: m[2] ?? "" };
53 }
54
4555 /** `--attach <name> <words>`: the name given to the file and the words after it; undefined without the flag. */
4656 export function parseAttach(words: string): { name: string; words: string } | undefined {
4757 const m = /^--attach[ \t]+(\S+)(?:[ \t]+([\s\S]*))?$/.exec(words.trim());
@@ -76,6 +86,20 @@ export function serveTalk(app: App, session: SshSession, user: User | undefined,
7686 session.exit(1);
7787 };
7888 if (/[\x00-\x1f\x7f-\x9f\u2028\u2029]/.test(words)) return refuse(CONTROL_WORDS);
89 // `/queued` lists the messages waiting for the running answer; `/unqueue <id>` takes one back.
90 if (words.trim() === "/queued") {
91 const waiting = queuedTalk(app, user);
92 session.write(waiting.length === 0 ? `nothing is waiting${CRLF}` : waiting.map((q) => `${q.id} ${q.text.slice(0, 80)}${CRLF}`).join(""));
93 return session.exit(0);
94 }
95 const unqueue = /^\/unqueue[ \t]+(\S+)$/.exec(words.trim());
96 if (unqueue !== null) {
97 if (!cancelTalk(app, user, unqueue[1]!)) return refuse("that message already went, or was never waiting");
98 session.write(`cancelled${CRLF}`);
99 return session.exit(0);
100 }
101 const during = parseDuring(words);
102 if (during !== undefined) words = during.words;
79103 if (isStop(words)) {
80104 const stopped = stopTalk(app, user);
81105 if (!stopped.ok)
@@ -88,7 +112,7 @@ export function serveTalk(app: App, session: SshSession, user: User | undefined,
88112 const finish = (said: ReturnType<typeof talk>): void => {
89113 if (!said.ok) return refuse(TALK_WORDS[said.error][1]);
90114 session.write(
91 `${said.value.how === "steered" ? "said to the forge's agent, mid-answer" : "the forge's agent is answering"}${CRLF}`,
115 `${said.value.how === "steered" ? "said to the forge's agent, mid-answer" : said.value.how === "queued" ? `queued, it sends after this answer (cancel: talk /unqueue ${said.value.queued})` : said.value.how === "interrupted" ? "stopped the answer; sending now" : "the forge's agent is answering"}${CRLF}`,
92116 );
93117 if (user !== undefined) {
94118 // The figures of the turn before this one, from its own usage record.
@@ -101,14 +125,14 @@ export function serveTalk(app: App, session: SshSession, user: User | undefined,
101125 session.exit(0);
102126 };
103127 const attach = parseAttach(words);
104 if (attach === undefined) return finish(talk(app, user, words));
128 if (attach === undefined) return finish(talk(app, user, words, [], during?.during ?? "inject"));
105129 if (user === undefined) return refuse(TALK_WORDS.unauthorized[1]);
106130 void readStdin(session, IMAGE_MAX).then((bytes) => {
107131 if (bytes === "too_large") return refuse(FILE_WORDS.too_large[1]);
108132 const stored = storeFile(app.db, app.config.dataDir, user.id, attach.name, bytes);
109133 if (!stored.ok) return refuse(FILE_WORDS[stored.error][1]);
110134 const files = pendingAmong(app.db, user.id, [stored.value.id]);
111 const said = talk(app, user, attach.words, files.ok ? files.value : []);
135 const said = talk(app, user, attach.words, files.ok ? files.value : [], during?.during ?? "inject");
112136 // A message that was not taken leaves no file waiting behind it.
113137 if (!said.ok) removePending(app.db, app.config.dataDir, user.id, stored.value.id);
114138 finish(said);
src/web/pages/talk.ts +31 −3
@@ -15,7 +15,7 @@
1515 */
1616
1717 import type { Database } from "bun:sqlite";
18 import { TALK_WORDS, talk, talkAgain } from "../../actions/talk.ts";
18 import { cancelTalk, queuedTalk, TALK_WORDS, talk, talkAgain } from "../../actions/talk.ts";
1919 import { REFUSED, sayNo, sayYes, waitingConfirms } from "../../actions/conversation_confirms.ts";
2020 import { filesOfTurns, type ConversationFile } from "../../core/conversation_files.ts";
2121 import { DEFAULT_PAGE, lastExchange, listTurns, MAX_PAGE, openConversation, turnsAfter, type Turn } from "../../core/conversations.ts";
@@ -108,6 +108,11 @@ export function exchangeActs(db: Database, userId: string, conversationId: strin
108108 return acts;
109109 }
110110
111 /** A message waiting for the running turn; Cancel is a form that works without the script. */
112 export const queuedHtml = (q: { id: string; text: string }): Html =>
113 html`<li class="talk-turn talk-person talk-queued" data-queued="${q.id}"><div class="talk-body talk-plain">${q.text}</div>
114 <form class="talk-meta small muted" method="post" action="${HERE}/queue/${q.id}/cancel" data-unqueue>Queued, sends after this answer ${btn("Cancel", { label: "Cancel this queued message" })}</form></li>`;
sentinelcrew outdated

Cancel is a bare type=button that only works with JavaScript; on a reload without JS it is dead. Use a post form (same-origin checked) so it works with JS off, per DESIGN.md.
115
111116 /** How full the context is, one quiet line above the log; always drawn so a finished run can fill it. */
112117 const contextLine = (c: Context | undefined): Html =>
113118 html`<p class="talk-context small muted" data-context ${c === undefined ? html`hidden` : null}>${c === undefined ? "" : contextText(c)}</p>`;
@@ -193,6 +198,7 @@ function talkPage(ctx: Ctx, before: string | undefined, prefill: string, edit: b
193198 ? html`<li class="muted talk-turn">Ask the forge anything, or tell it what to change. It acts as you, with your permissions, and asks before anything hard to undo.
194199 <span class="talk-suggest">${SUGGESTIONS.map((q) => html`<a class="btn" href="${HERE}?q=${encodeURIComponent(q)}" data-suggest="${q}">${q}</a>`)}</span></li>`
195200 : turns.map((t) => turnHtml(t, waiting, costs.get(t.id), acts.get(t.id), works.get(t.id), sent.get(t.id)))}
201 ${before === undefined && running ? queuedTalk(ctx.app, user).map(queuedHtml) : null}
196202 </ol>
197203 <div id="end"></div>
198204 </div>
@@ -205,6 +211,8 @@ function talkPage(ctx: Ctx, before: string | undefined, prefill: string, edit: b
205211 <label class="btn talk-attach" data-attach hidden>Attach<input type="file" multiple hidden data-pick /></label>
206212 <span class="spacer"></span>
207213 ${editing === undefined ? null : html`<a class="btn" href="${HERE}">Keep as it was</a>`}
214 <button type="button" class="btn" data-stop hidden>Stop</button>
215 <button type="button" class="btn" data-steer hidden title="Stop this answer and send now">Steer</button>
208216 ${btn(editing === undefined ? "Send" : "Resend", { variant: "primary" })}
209217 </div>
210218 </form>
@@ -230,7 +238,7 @@ export async function talkSend(ctx: Ctx): Promise<Response> {
230238 const text = form.get("text");
231239 if (typeof text !== "string" || text.trim() === "" || text.length > WORDS_MAX)
232240 return flashTo(ctx, `${HERE}#end`, { tone: "danger", message: `Say something, up to ${WORDS_MAX} characters.` });
233 const said = talk(ctx.app, user, text);
241 const said = talk(ctx.app, user, text, [], "queue");
234242 if (!said.ok) return flashTo(ctx, `${HERE}#end`, { tone: "danger", message: TALK_WORDS[said.error][1] });
235243 return flashTo(ctx, `${HERE}#end`, { tone: "success", message: "Sent. Reload to see the answer." });
236244 }
@@ -258,6 +266,18 @@ export async function talkAgainSend(ctx: Ctx): Promise<Response> {
258266 return asJson ? json({ ok: false, message }, status) : flashTo(ctx, `${HERE}#end`, { tone: "danger", message });
259267 }
260268
269 // POST /talk/queue/:id/cancel — take back a message waiting for the running turn.
270 export async function talkQueueCancel(ctx: Ctx): Promise<Response> {
271 const user = ctx.auth.user;
272 if (user === undefined) return signInFirst(ctx);
273 if (ctx.auth.via !== "session") return forbiddenPage(ctx, "Do this from your own signed-in browser.");
274 const form = await readForm(ctx, "rw");
275 if (form instanceof Response) return form;
276 const gone = !cancelTalk(ctx.app, user, ctx.params["id"] ?? "");
277 if ((ctx.request.headers.get("accept") ?? "").includes("application/json")) return gone ? json({ ok: false, message: "That message already went." }, 404) : json({ ok: true });
278 return flashTo(ctx, `${HERE}#end`, gone ? { tone: "danger", message: "That message already went." } : { tone: "success", message: "Cancelled." });
279 }
280
261281 // POST /talk/confirm/:id — the person's yes or no, from their own session.
262282 export async function talkConfirm(ctx: Ctx): Promise<Response> {
263283 const user = ctx.auth.user;
@@ -300,6 +320,7 @@ export function talkStream(ctx: Ctx): Response {
300320 return eventStream(app, ctx.request, (emit, close) => {
301321 const started = Date.now();
302322 let told = false;
323 let shown = "";
303324 const tick = (): void => {
304325 // Asked before the read: a turn that ends after this look is read whole next time.
305326 const running = clock.running(app, user.id);
@@ -309,10 +330,17 @@ export function talkStream(ctx: Ctx): Response {
309330 const works = worksOf(app.db, user.id, conversation.id, fresh);
310331 const sent = filesOfTurns(app.db, user.id, fresh.filter((t) => t.role === "person").map((t) => t.id));
311332 for (const turn of fresh) {
312 emit("turn", { id: turn.id, html: turnHtml(turn, waiting, undefined, undefined, works.get(turn.id), sent.get(turn.id)).text }, turn.id);
333 emit("turn", { id: turn.id, role: turn.role, html: turnHtml(turn, waiting, undefined, undefined, works.get(turn.id), sent.get(turn.id)).text }, turn.id);
313334 cursor = turn.id;
314335 }
315336 }
337 // Every tab and viewer sees what waits: sent whenever the list changes.
338 const waits = queuedTalk(app, user);
339 const key = waits.map((q) => q.id).join();
340 if (key !== shown && (running || shown !== "")) {
341 shown = key;
342 emit("queued", { items: waits.map((q) => ({ id: q.id, html: queuedHtml(q).text })) });
343 }
316344 if (fresh.length >= MAX_PAGE) return;
317345 if (!running) {
318346 // The run stored its figures before it stopped running: send them with the end.
src/web/pages/talk_client.ts +40 −8
@@ -50,6 +50,7 @@ export const talkCss = /* css */ `
5050 .talk-replaced { opacity: 0.55; }
5151 .talk-failed form { margin-top: 6px; }
5252 .talk-failed .btn, .talk-act { min-height: 2.75rem; }
53 .talk-queued { opacity: 0.7; border-style: dashed; }
5354 .talk-acts { display: flex; gap: 6px; justify-content: flex-end; margin-top: 4px; }
5455 .talk-agent .talk-acts { justify-content: flex-start; }
5556 `;
@@ -144,14 +145,16 @@ if (root) {
144145 return false;
145146 }
146147 };
147 // While a reply streams, Send is Stop; Escape stops too.
148 const action = form.querySelector(".bar button.btn");
149 const sendLabel = action.textContent;
148 // While a reply streams the composer stays open: Send queues the message, Steer stops the
149 // answer and sends it at once, Stop only stops. Escape stops too.
150 const action = form.querySelector(".bar button.btn:not([data-stop]):not([data-steer])");
151 const stopBtn = form.querySelector("[data-stop]");
152 const steerBtn = form.querySelector("[data-steer]");
150153 let answering = false;
151154 const mode = (on) => {
152155 answering = on;
153 action.textContent = on ? "Stop" : sendLabel;
154 action.dataset.stop = on ? "1" : "";
156 action.textContent = on ? "Queue" : "Send";
157 stopBtn.hidden = steerBtn.hidden = !on;
155158 };
156159 const stop = async () => {
157160 say("Stopping…");
@@ -162,8 +165,25 @@ if (root) {
162165 say("That was not sent: no connection.");
163166 }
164167 };
165 action.addEventListener("click", (e) => { if (answering) { e.preventDefault(); stop(); } });
168 stopBtn.addEventListener("click", stop);
169 steerBtn.addEventListener("click", () => form.requestSubmit(steerBtn));
166170 document.addEventListener("keydown", (e) => { if (e.key === "Escape" && answering) { e.preventDefault(); stop(); } });
171 // Queued messages are drawn by the server (they are the same on every tab); Cancel is its form, done in place.
172 log.addEventListener("submit", async (e) => {
173 const f = e.target.closest("form[data-unqueue]");
174 if (!f) return;
175 e.preventDefault();
176 const li = f.closest("[data-queued]");
177 try {
178 const res = await fetch(f.action, { method: "POST", headers: { accept: "application/json" }, body: new URLSearchParams() });
179 if (!res.ok && res.status !== 404) { say("That was not cancelled."); return; }
180 if (res.ok) { const t = li.querySelector(".talk-body").textContent; box.value = box.value ? box.value + "\\n" + t : t; grow(); say("Cancelled. The message is back in the box."); }
181 else say("It had already gone.");
182 li.remove();
183 } catch (_) {
184 say("That was not cancelled: no connection.");
185 }
186 });
167187 const follow = () => {
168188 if (paged) return;
169189 if (source) source.close();
@@ -172,10 +192,18 @@ if (root) {
172192 source.addEventListener("turn", (e) => {
173193 const turn = JSON.parse(e.data);
174194 last = turn.id;
175 if (!document.getElementById("t-" + turn.id)) { log.insertAdjacentHTML("beforeend", turn.html); decorate(log); }
195 if (!document.getElementById("t-" + turn.id)) {
196 const q = log.querySelector("[data-queued]");
197 if (q) q.insertAdjacentHTML("beforebegin", turn.html); else log.insertAdjacentHTML("beforeend", turn.html);
198 decorate(log);
199 }
176200 retry = 0;
177201 toEnd();
178202 });
203 source.addEventListener("queued", (e) => {
204 log.querySelectorAll("[data-queued]").forEach((q) => q.remove());
205 for (const q of JSON.parse(e.data).items) log.insertAdjacentHTML("beforeend", q.html);
206 });
179207 source.addEventListener("running", () => { dropActs(); mode(true); say("The forge is answering…"); });
180208 source.addEventListener("done", (e) => {
181209 mode(false); dropActs(); say(""); source.close(); source = null;
@@ -249,12 +277,13 @@ if (root) {
249277 if (await again("edit", new URLSearchParams({ text }))) { try { localStorage.removeItem(draftKey); } catch (_) {} }
250278 return;
251279 }
280 const steer = e.submitter === steerBtn;
252281 say("Sending…");
253282 try {
254283 const res = await fetch("/api/v1/conversation/messages", {
255284 method: "POST",
256285 headers: { "content-type": "application/json" },
257 body: JSON.stringify({ text, files: attached.map((f) => f.id) }),
286 body: JSON.stringify({ text, files: attached.map((f) => f.id), during: steer && answering ? "steer" : "queue" }),
258287 });
259288 if (!res.ok) {
260289 const body = await res.json().catch(() => ({}));
@@ -267,6 +296,9 @@ if (root) {
267296 grow();
268297 try { localStorage.removeItem(draftKey); } catch (_) {}
269298 stuck = true;
299 const out = await res.json().catch(() => ({}));
300 if (out.how === "queued") { say("Queued: it sends after this answer."); if (!source) follow(); return; }
301 if (out.how === "interrupted") { say("Stopping, then sending…"); return; }
270302 if (paged) { location.href = "/talk"; return; }
271303 follow();
272304 } catch (_) {
test/conversation-queue-api.test.ts added +222
@@ -0,0 +1,222 @@
1 /**
2 * The doors to queue and steer (#8819): the API's `during`, its queue limit,
3 * Cancel by DELETE and by the page's form, the `queued` list, the held file,
4 * the stream's `queued` frame, and `ssh talk /queue|/steer|/queued|/unqueue`.
5 */
6
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { chmodSync, writeFileSync } from "node:fs";
9 import { mkdtemp, rm } from "node:fs/promises";
10 import { tmpdir } from "node:os";
11 import { join } from "node:path";
12 import { createApp, type App } from "../src/app.ts";
13 import { configureConversation, conversationOptions, QUEUE_MAX, queuedOf, say, stopTurn, turnRunning } from "../src/crew/conversation.ts";
14 import { storeFile } from "../src/core/conversation_files.ts";
15 import { addKey, fingerprint, parseAuthorizedKey } from "../src/core/keys.ts";
16 import { createSession } from "../src/core/sessions.ts";
17 import { createToken } from "../src/core/tokens.ts";
18 import { createUser, type User } from "../src/core/users.ts";
19 import { serve, type Servers } from "../src/http/server.ts";
20 import { forgeSshSession } from "../src/tui/ssh.ts";
21 import { parseDuring } from "../src/tui/ssh-talk.ts";
22 import { quietLogs } from "./helpers/globals.ts";
23
24 quietLogs();
25
26 const KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIB2S7bF0R4NmuTUvVJXPPjdI1AF8XHXeYSNfr8DJ5eDo alice";
27
28 let dir: string;
29 let app: App;
30 let servers: Servers;
31 let alice: User;
32 let rw: string;
33 let ro: string;
34 let bob: string;
35
36 const headers = (t: string, extra: Record<string, string> = {}) => ({ "content-type": "application/json", authorization: `Bearer ${t}`, ...extra });
37 const send = (body: object, t = rw, extra: Record<string, string> = {}) =>
38 fetch(`${servers.url}/api/v1/conversation/messages`, { method: "POST", headers: headers(t, extra), body: JSON.stringify(body) });
39 const until = async (ok: () => boolean) => {
40 for (let i = 0; i < 100 && !ok(); i++) await Bun.sleep(10);
41 };
42 /** Starts a turn that holds until stopped, without counting against the minute's messages. */
43 const start = async () => {
44 say(app, conversationOptions(app)!, alice.id, "go", [], "queue");
45 await until(() => turnRunning(app, alice.id));
46 };
47 /** Ends whatever runs, and what waits, so the next test starts clean. */
48 const clear = async () => {
49 stopTurn(app, alice.id);
50 await until(() => !turnRunning(app, alice.id));
51 };
52 const exec = async (command: string) => {
53 const parsed = parseAuthorizedKey(KEY);
54 if (!parsed.ok) throw new Error("key");
55 let out = "";
56 let err = "";
57 let code: number | undefined;
58 await new Promise<void>((resolve) => {
59 forgeSshSession(app)({
60 key: { fingerprint: fingerprint(parsed.value.blob), line: KEY },
61 username: "whoever",
62 remoteAddress: "203.0.113.9",
63 kind: "exec",
64 command,
65 term: undefined,
66 env: {},
67 cols: 0,
68 rows: 0,
69 widthPx: 0,
70 heightPx: 0,
71 ended: false,
72 pending: 0,
73 write: (d: Uint8Array | string) => void (out += typeof d === "string" ? d : Buffer.from(d).toString()),
74 writeStderr: (d: string) => void (err += d),
75 onData() {},
76 onResize() {},
77 onEof() {},
78 onClose() {},
79 exit(c: number) {
80 code ??= c;
81 resolve();
82 },
83 drained: async () => {},
84 } as never);
85 });
86 return { out, err, code };
87 };
88
89 beforeAll(async () => {
90 dir = await mkdtemp(join(tmpdir(), "forge-conv-queue-api-"));
91 app = createApp({ dataDir: dir, port: 0, host: "forge.test" });
92 servers = await serve(app);
93 const made = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
94 if (!made.ok) throw new Error(made.error);
95 alice = made.value;
96 const other = await createUser(app.db, { origin: "seed", trusted: true, username: "bob", email: "[email protected]", password: "a-good-passphrase" });
97 if (!other.ok) throw new Error(other.error);
98 const parsed = parseAuthorizedKey(KEY);
99 if (!parsed.ok) throw new Error("key");
100 addKey(app.db, alice.id, "laptop", parsed.value);
101 rw = createToken(app.db, alice.id, "personal", "rw").token;
102 ro = createToken(app.db, alice.id, "personal", "ro").token;
103 bob = createToken(app.db, other.value.id, "personal", "rw").token;
104 // Holds each run until stopped: the turn is "running" for as long as a test needs.
105 const bin = join(dir, "claude");
106 writeFileSync(bin, `#!/bin/sh\nif [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi\nIFS= read -r first\necho '{"type":"system","subtype":"init","session_id":"s1","model":"fake"}'\nsleep 30\n`);
107 chmodSync(bin, 0o755);
108 configureConversation(app, { binary: bin, localUrl: servers.url, budgetUsd: 1, timeoutMs: 60_000, signal: new AbortController().signal });
109 });
110 afterAll(async () => {
111 await clear();
112 await servers.stop();
113 await rm(dir, { recursive: true, force: true });
114 });
115
116 describe("queue and steer through the doors", () => {
117 test("`during` is queue or steer; a queued message is listed, a sixth is refused, and DELETE takes one back", async () => {
118 await start();
119 expect(((await (await send({ text: "x", during: "later" })).json()) as { error: string }).error).toBe("invalid_during");
120 expect((await send({ text: "x", during: 5 })).status).toBe(422);
121 const ids: string[] = [];
122 for (let i = 0; i < QUEUE_MAX; i++) {
123 const res = await send({ text: `wait ${i}`, during: "queue" });
124 expect(res.status).toBe(202);
125 const body = (await res.json()) as { how: string; queued: string };
126 expect(body.how).toBe("queued");
127 ids.push(body.queued);
128 }
129 const full = await send({ text: "one too many", during: "queue" });
130 expect(full.status).toBe(429);
131 expect(((await full.json()) as { error: string }).error).toBe("queue_full");
132 const listed = (await (await fetch(`${servers.url}/api/v1/conversation/turns`, { headers: headers(rw) })).json()) as { queued: { id: string; text: string }[] };
133 expect(listed.queued.map((q) => q.text)).toEqual(["wait 0", "wait 1", "wait 2", "wait 3", "wait 4"]);
134 const del = (id: string, t: string | undefined, extra: Record<string, string> = {}) =>
135 fetch(`${servers.url}/api/v1/conversation/queue/${id}`, { method: "DELETE", headers: t === undefined ? extra : headers(t, extra) });
136 expect((await del(ids[0]!, undefined)).status).toBe(401);
137 expect((await del(ids[0]!, rw, { origin: "https://elsewhere.example" })).status).toBe(403);
138 expect((await del(ids[0]!, ro)).status).toBe(403);
139 expect((await del(ids[0]!, bob)).status).toBe(404);
140 expect(queuedOf(app, alice.id)).toHaveLength(QUEUE_MAX);
141 expect((await del(ids[0]!, rw)).status).toBe(200);
142 expect((await del(ids[0]!, rw)).status).toBe(404);
143 expect(queuedOf(app, alice.id)).toHaveLength(QUEUE_MAX - 1);
144 await clear();
145 });
146
147 test("steer by POST stops the turn and sends the message at once", async () => {
148 await start();
149 say(app, conversationOptions(app)!, alice.id, "after", [], "queue");
150 const res = await send({ text: "now", during: "steer" });
151 expect(((await res.json()) as { how: string }).how).toBe("interrupted");
152 // The queue holds the steering message ahead of the one that waited, until the stopped turn ends.
153 await until(() => queuedOf(app, alice.id).length === 1);
154 expect(queuedOf(app, alice.id).map((q) => q.text)).toEqual(["after"]);
155 await clear();
156 });
157
158 test("a file in a queued message is held: not removable, not sent twice", async () => {
159 await start();
160 const file = storeFile(app.db, dir, alice.id, "n.txt", new TextEncoder().encode("notes"));
161 if (!file.ok) throw new Error(file.error);
162 expect((await send({ text: "with file", during: "queue", files: [file.value.id] })).status).toBe(202);
163 expect((await send({ text: "same file", during: "queue", files: [file.value.id] })).status).toBe(409);
164 const del = await fetch(`${servers.url}/api/v1/conversation/files/${file.value.id}`, { method: "DELETE", headers: headers(rw) });
165 expect(del.status).toBe(409);
166 await clear();
167 });
168
169 test("the page's Cancel is a post form that works without the script, for the signed-in owner only", async () => {
170 await start();
171 const queued = say(app, conversationOptions(app)!, alice.id, "hold on", [], "queue");
172 const id = queued.ok ? queued.value.queued : "";
173 const cookie = `forge_session=${createSession(app.db, alice.id).token}`;
174 const page = await (await fetch(`${servers.url}/talk`, { headers: { cookie } })).text();
175 expect(page).toContain(`action="/talk/queue/${id}/cancel"`);
176 expect(page).toContain("hold on");
177 const cancel = (extra: Record<string, string>) =>
178 fetch(`${servers.url}/talk/queue/${id}/cancel`, { method: "POST", redirect: "manual", headers: { cookie, ...extra }, body: new URLSearchParams() });
179 expect((await cancel({ origin: "https://elsewhere.example" })).status).toBe(403);
180 expect(queuedOf(app, alice.id)).toHaveLength(1);
181 expect((await cancel({ origin: servers.url })).status).toBe(303);
182 expect(queuedOf(app, alice.id)).toHaveLength(0);
183 expect((await cancel({ origin: servers.url, accept: "application/json" })).status).toBe(404);
184 await clear();
185 });
186
187 test("the stream tells every viewer what waits", async () => {
188 await start();
189 say(app, conversationOptions(app)!, alice.id, "seen on another tab", [], "queue");
190 const cookie = `forge_session=${createSession(app.db, alice.id).token}`;
191 const abort = new AbortController();
192 const res = await fetch(`${servers.url}/talk/stream?after=`, { headers: { cookie }, signal: abort.signal });
193 const reader = res.body!.getReader();
194 let seen = "";
195 while (!seen.includes("seen on another tab")) {
196 const chunk = await reader.read();
197 if (chunk.done) break;
198 seen += new TextDecoder().decode(chunk.value);
199 }
200 abort.abort();
201 expect(seen).toContain("event: queued");
202 expect(seen).toContain("seen on another tab");
203 await clear();
204 });
205
206 test("ssh talk /queue, /queued, /unqueue and /steer", async () => {
207 expect(parseDuring("/queue hello there")).toEqual({ during: "queue", words: "hello there" });
208 expect(parseDuring("/steer now")).toEqual({ during: "interrupt", words: "now" });
209 expect(parseDuring("plain words")).toBeUndefined();
210 await start();
211 const queued = await exec("talk /queue later please");
212 expect(queued.code).toBe(0);
213 expect(queued.out).toContain("queued");
214 const id = /talk \/unqueue (\S+)\)/.exec(queued.out)![1]!;
215 expect((await exec("talk /queued")).out).toContain(`${id} later please`);
216 expect((await exec(`talk /unqueue ${id}`)).out).toContain("cancelled");
217 expect((await exec(`talk /unqueue ${id}`)).code).toBe(1);
218 expect((await exec("talk /queued")).out).toContain("nothing is waiting");
219 expect((await exec("talk /steer instead")).out).toContain("stopped the answer");
220 await clear();
221 });
222 });
test/conversation-queue.test.ts added +129
@@ -0,0 +1,129 @@
1 /**
2 * A message sent while a turn runs (#8819): Send queues it for after the turn,
3 * in order, with Cancel; Steer stops the turn, keeps its partial answer, and
4 * sends the message at once; Stop forgets what waited.
5 */
6
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { chmodSync, existsSync, readFileSync, writeFileSync } from "node:fs";
9 import { mkdtemp, rm } from "node:fs/promises";
10 import { tmpdir } from "node:os";
11 import { join } from "node:path";
12 import { createApp, type App } from "../src/app.ts";
13 import { listTurns, openConversation } from "../src/core/conversations.ts";
14 import { createUser, type User } from "../src/core/users.ts";
15 import { cancelQueued, queuedOf, say, STOPPED_LINE, stopTurn, turnRunning, type ConversationOptions } from "../src/crew/conversation.ts";
16
17 let dir: string;
18 let app: App;
19 let alice: User;
20
21 const options = (binary: string): ConversationOptions => ({ binary, localUrl: "http://127.0.0.1:9", budgetUsd: 1, timeoutMs: 30_000, signal: new AbortController().signal });
22 const turns = (u: User) => {
23 const r = listTurns(app.db, u.id, openConversation(app.db, u.id).id);
24 return r.ok ? r.value.map((t) => `${t.role}:${t.text}`) : [];
25 };
26
27 /** A stand-in that logs each message it is given and holds until `<name>.gate.<n>` exists, n being its place among the runs. */
28 function chain(name: string): string {
29 const path = join(dir, name);
30 writeFileSync(
31 path,
32 `#!/bin/sh
33 if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi
34 IFS= read -r first
35 printf '%s\\n' "$first" | sed 's/.*"text":"\\([^"]*\\)".*/\\1/' >> "${dir}/${name}.log"
36 n=$(wc -l < "${dir}/${name}.log" | tr -d ' ')
37 echo '{"type":"system","subtype":"init","session_id":"sess-chain","model":"fake"}'
38 echo '{"type":"assistant","session_id":"sess-chain","message":{"id":"m'$n'","content":[{"type":"text","text":"partial '$n'"}],"usage":{"input_tokens":1,"output_tokens":1}}}'
39 while [ ! -f "${dir}/${name}.gate.$n" ]; do sleep 0.05; done
40 echo '{"type":"result","subtype":"success","is_error":false,"result":"partial '$n'","total_cost_usd":0.1,"num_turns":1,"session_id":"sess-chain","terminal_reason":"completed","usage":{"input_tokens":1,"output_tokens":1}}'
41 `,
42 );
43 chmodSync(path, 0o755);
44 return path;
45 }
46 const logged = (name: string): string[] => (existsSync(join(dir, `${name}.log`)) ? readFileSync(join(dir, `${name}.log`), "utf8").trim().split("\n") : []);
47 const until = async (ok: () => boolean): Promise<void> => {
48 for (let i = 0; i < 200 && !ok(); i++) await Bun.sleep(20);
49 };
50
51
52 beforeAll(async () => {
53 dir = await mkdtemp(join(tmpdir(), "forge-conv-queue-"));
54 app = createApp({ dataDir: dir, port: 0 });
55 const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
56 if (!a.ok) throw new Error("user");
57 alice = a.value;
58 });
59 afterAll(async () => {
60 app.db.close();
61 await rm(dir, { recursive: true, force: true });
62 });
63
64 describe("a message sent during a turn", () => {
65 test("two messages sent during a turn queue, can be cancelled, and run in order after it (#8819)", async () => {
66 const bin = chain("queue");
67 const base = turns(alice).length;
68 const o = options(bin);
69 const first = say(app, o, alice.id, "first", [], "queue");
70 expect(first.ok && first.value.how).toBe("started");
71 await until(() => logged("queue").length === 1);
72 const two = say(app, o, alice.id, "second", [], "queue");
73 const three = say(app, o, alice.id, "third", [], "queue");
74 const gone = say(app, o, alice.id, "never", [], "queue");
75 expect(two.ok && two.value.how).toBe("queued");
76 expect(three.ok && three.value.how).toBe("queued");
77 // Waiting messages are not turns yet; the running turn is not disturbed.
78 expect(queuedOf(app, alice.id).map((q) => q.text)).toEqual(["second", "third", "never"]);
79 expect(turns(alice).some((t) => t === "person:second")).toBe(false);
80 expect(gone.ok && gone.value.queued && cancelQueued(app, alice.id, gone.value.queued)).toBe(true);
81 expect(cancelQueued(app, alice.id, "nope")).toBe(false);
82 writeFileSync(join(dir, "queue.gate.1"), "");
83 await until(() => logged("queue").length === 2);
84 expect(logged("queue")).toEqual(["first", "second"]);
85 expect(queuedOf(app, alice.id).map((q) => q.text)).toEqual(["third"]);
86 writeFileSync(join(dir, "queue.gate.2"), "");
87 await until(() => logged("queue").length === 3);
88 writeFileSync(join(dir, "queue.gate.3"), "");
89 await until(() => !turnRunning(app, alice.id));
90 expect(logged("queue")).toEqual(["first", "second", "third"]);
91 const mine = turns(alice).slice(base).filter((t) => /^(person:(first|second|third|never)|agent:partial)/.test(t));
92 expect(mine).toEqual(["person:first", "agent:partial 1", "person:second", "agent:partial 2", "person:third", "agent:partial 3"]);
93 });
94
95 test("steer stops the turn, keeps its partial answer, and starts a new turn with the message at once (#8819)", async () => {
96 const bin = chain("steer-now");
97 const base = turns(alice).length;
98 const o = options(bin);
99 const first = say(app, o, alice.id, "long job", [], "queue");
100 expect(first.ok && first.value.how).toBe("started");
101 await until(() => logged("steer-now").length === 1);
102 const queued = say(app, o, alice.id, "later", [], "queue");
103 expect(queued.ok && queued.value.how).toBe("queued");
104 const steered = say(app, o, alice.id, "do this instead", [], "interrupt");
105 expect(steered.ok && steered.value.how).toBe("interrupted");
106 // The stopped turn ends; the steering message runs ahead of the one already waiting.
107 await until(() => logged("steer-now").length === 2);
108 expect(logged("steer-now")).toEqual(["long job", "do this instead"]);
109 writeFileSync(join(dir, "steer-now.gate.2"), "");
110 await until(() => logged("steer-now").length === 3);
111 writeFileSync(join(dir, "steer-now.gate.3"), "");
112 await until(() => !turnRunning(app, alice.id));
113 expect(logged("steer-now")).toEqual(["long job", "do this instead", "later"]);
114 const mine = turns(alice).slice(base).filter((t) => /^(person:(long job|do this instead|later)|agent:(partial|Stopped))/.test(t));
115 expect(mine).toEqual(["person:long job", "agent:partial 1", `agent:${STOPPED_LINE}`, "person:do this instead", "agent:partial 2", "person:later", "agent:partial 3"]);
116 });
117
118 test("Stop forgets what waited behind the turn", async () => {
119 const bin = chain("stop-queue");
120 const o = options(bin);
121 say(app, o, alice.id, "running", [], "queue");
122 await until(() => logged("stop-queue").length === 1);
123 say(app, o, alice.id, "waiting", [], "queue");
124 expect(stopTurn(app, alice.id)).toBe(true);
125 await until(() => !turnRunning(app, alice.id));
126 expect(queuedOf(app, alice.id)).toEqual([]);
127 expect(logged("stop-queue")).toEqual(["running"]);
128 });
129 });
test/talk-page.test.ts +57 −10
@@ -192,7 +192,7 @@ describe("the page", () => {
192192 test("the page's own bytes are held: stylesheet and script stay small", async () => {
193193 const { talkCss, talkJs } = await import("../src/web/pages/talk_client.ts");
194194 expect(Buffer.byteLength(talkCss)).toBeLessThan(3_600);
195 expect(Buffer.byteLength(talkJs)).toBeLessThan(11_600);
195 expect(Buffer.byteLength(talkJs)).toBeLessThan(13_400);
196196 say(alice, "agent", "```ts\nlet x = 1;\n```");
197197 const page = await get("/talk", alice);
198198 const size = (await page.arrayBuffer()).byteLength;
@@ -200,21 +200,65 @@ describe("the page", () => {
200200 });
201201
202202 describe("the script, driven against a DOM", () => {
203 test("while a reply streams Send is Stop, Escape stops too, and it is Send again when the turn ends", async () => {
203 test("while a reply streams the composer stays open: Send queues, Steer and Stop appear, Escape stops, and it is Send again when the turn ends", async () => {
204204 const t = await boot();
205 const button = t.doc.querySelector(".bar .btn");
206 expect(button.textContent).toBe("Send");
205 const send = t.doc.querySelector('.bar button[type="submit"]');
206 const stop = t.doc.querySelector("[data-stop]");
207 expect(send.textContent).toBe("Send");
208 expect(stop.hidden).toBe(true);
207209 t.sources.at(-1)!.on["running"]!({ data: "{}" });
208 expect(button.textContent).toBe("Stop");
209 button.dispatchEvent(new t.window.MouseEvent("click", { bubbles: true, cancelable: true }));
210 expect(send.textContent).toBe("Queue");
211 expect(stop.hidden).toBe(false);
212 stop.dispatchEvent(new t.window.MouseEvent("click", { bubbles: true, cancelable: true }));
210213 t.doc.dispatchEvent(new t.window.KeyboardEvent("keydown", { key: "Escape", bubbles: true, cancelable: true }));
211214 expect(t.urls).toEqual(["/api/v1/conversation/stop", "/api/v1/conversation/stop"]);
212215 t.sources.at(-1)!.on["done"]!({ data: "{}" });
213 expect(button.textContent).toBe("Send");
216 expect(send.textContent).toBe("Send");
217 expect(stop.hidden).toBe(true);
214218 t.doc.dispatchEvent(new t.window.KeyboardEvent("keydown", { key: "Escape", bubbles: true, cancelable: true }));
215219 expect(t.urls).toHaveLength(2);
216220 });
217221
222 const queuedFrame = (id: string, text: string) => ({ id, html: `<li class="talk-turn talk-queued" data-queued="${id}"><div class="talk-body">${text}</div><form data-unqueue method="post" action="/talk/queue/${id}/cancel"><button>Cancel</button></form></li>` });
223
224 test("a message sent mid-answer is queued; the stream's queued frame draws it; Cancel puts it back without overwriting what is typed (#8819)", async () => {
225 const t = await boot();
226 t.how.value = "queued";
227 t.sources.at(-1)!.on["running"]!({ data: "{}" });
228 t.box.value = "and then this";
229 t.key({});
230 await tick();
231 expect(t.bodies.at(-1)).toMatchObject({ text: "and then this", during: "queue" });
232 expect(t.status.textContent).toContain("Queued");
233 t.sources.at(-1)!.on["queued"]!({ data: JSON.stringify({ items: [queuedFrame("q1", "and then this")] }) });
234 t.sources.at(-1)!.on["queued"]!({ data: JSON.stringify({ items: [queuedFrame("q1", "and then this")] }) });
235 expect(t.doc.querySelectorAll("[data-queued]")).toHaveLength(1);
236 t.box.value = "something new";
237 t.doc.querySelector("form[data-unqueue]").dispatchEvent(new t.window.Event("submit", { bubbles: true, cancelable: true }));
238 await tick();
239 expect(t.urls.at(-1)).toEndWith("/talk/queue/q1/cancel");
240 expect(t.doc.querySelector("[data-queued]")).toBeNull();
241 expect(t.box.value).toBe("something new\nand then this");
242 });
243
244 test("Cancel keeps the message when the server fails and drops it only on a 404; Steer sends a steer", async () => {
245 const t = await boot();
246 t.sources.at(-1)!.on["running"]!({ data: "{}" });
247 t.sources.at(-1)!.on["queued"]!({ data: JSON.stringify({ items: [queuedFrame("q1", "keep me")] }) });
248 t.cancel.status = 500;
249 t.doc.querySelector("form[data-unqueue]").dispatchEvent(new t.window.Event("submit", { bubbles: true, cancelable: true }));
250 await tick();
251 expect(t.doc.querySelector("[data-queued]")).not.toBeNull();
252 t.cancel.status = 404;
253 t.doc.querySelector("form[data-unqueue]").dispatchEvent(new t.window.Event("submit", { bubbles: true, cancelable: true }));
254 await tick();
255 expect(t.doc.querySelector("[data-queued]")).toBeNull();
256 t.box.value = "do this instead";
257 t.doc.querySelector("[data-steer]").dispatchEvent(new t.window.MouseEvent("click", { bubbles: true, cancelable: true }));
258 await tick();
259 expect(t.bodies.at(-1)).toMatchObject({ text: "do this instead", during: "steer" });
260 });
261
218262 const boot = async (user = "u1", store: Record<string, string> = {}) => {
219263 const sources: Array<{ on: Record<string, (e: { data: string }) => void> }> = [];
220264 const { Window } = await import("happy-dom");
@@ -223,7 +267,9 @@ describe("the page", () => {
223267 const doc = window.document as any;
224268 doc.body.innerHTML = `<div class="talk" data-talk data-user="${user}" data-paged="0" data-running="1"><div>
225269 <ol class="talk-log" role="log" data-last=""><li class="talk-turn talk-agent" id="t-1"><div class="talk-body"><pre><code>let a = 1;</code></pre></div><p class="talk-meta">Forge · now</p></li></ol><div id="end"></div></div>
226 <form class="talk-compose"><p class="talk-status" role="status"></p><ul data-chips></ul><textarea name="text"></textarea><div class="bar"><span data-hint></span><label data-attach hidden><input type="file" data-pick /></label><button class="btn" type="submit">Send</button></div></form></div>`;
270 <form class="talk-compose"><p class="talk-status" role="status"></p><ul data-chips></ul><textarea name="text"></textarea><div class="bar"><span data-hint></span><label data-attach hidden><input type="file" data-pick /></label><button class="btn" type="button" data-stop hidden>Stop</button><button class="btn" type="button" data-steer hidden>Steer</button><button class="btn" type="submit">Send</button></div></form></div>`;
271 const how = { value: "started" };
272 const cancel = { status: 200 };
227273 const sent: string[] = [];
228274 const urls: string[] = [];
229275 const copied: string[] = [];
@@ -232,10 +278,11 @@ describe("the page", () => {
232278 urls.push(u);
233279 if (u.endsWith("/files")) return { ok: true, json: async () => ({ id: "f1", name: "shot.png" }) };
234280 if (init.method === "DELETE") return { ok: true, json: async () => ({}) };
281 if (u.includes("/queue/")) return { ok: cancel.status === 200, status: cancel.status, json: async () => ({}) };
235282 const body = JSON.parse(init.body);
236283 sent.push(body.text);
237284 bodies.push(body);
238 return { ok: true, json: async () => ({}) };
285 return { ok: true, json: async () => ({ how: how.value, queued: "q1" }) };
239286 };
240287 const local = { getItem: (k: string) => store[k] ?? null, setItem: (k: string, v: string) => { store[k] = v; }, removeItem: (k: string) => { delete store[k]; } };
241288 const nav = { clipboard: { writeText: async (t: string) => { copied.push(t); } } };
@@ -245,7 +292,7 @@ describe("the page", () => {
245292 const box = doc.querySelector("textarea");
246293 const key = (init: object) => box.dispatchEvent(new window.KeyboardEvent("keydown", { key: "Enter", bubbles: true, cancelable: true, ...init }));
247294 const emit = (id: string, html: string) => sources.at(-1)!.on["turn"]!({ data: JSON.stringify({ id, html }) });
248 return { emit, bodies, sources, urls, window, doc, box, sent, copied, store, key, status: doc.querySelector(".talk-status") };
295 return { how, cancel, emit, bodies, sources, urls, window, doc, box, sent, copied, store, key, status: doc.querySelector(".talk-status") };
249296 };
250297 const tick = () => new Promise((r) => setTimeout(r, 5));
251298

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs