Public: anyone can read this; the crew can change it.
A /talk reply in progress survives a reload and shows on another device mid-stream #8833
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/8815-a-talk-reply-in-progress-survives-a-relo → main 1 commit fork
1 file
changed +62
−1
test/conversation-turn.test.ts +62 −1
| @@ -18,6 +18,8 @@ import { monthSpent } from "../src/core/model_grant.ts"; | ||
| 18 | 18 | import { addSubscription, getSubscription, PLATFORM, setGrants, wakeSubscription } from "../src/core/subscriptions.ts"; |
| 19 | 19 | import { verifyToken } from "../src/core/tokens.ts"; |
| 20 | 20 | import { createUser, type User } from "../src/core/users.ts"; |
| 21 | import { createSession } from "../src/core/sessions.ts"; | |
| 22 | import { serve } from "../src/http/server.ts"; | |
| 21 | 23 | import { TOOLS } from "../src/crew/claude.ts"; |
| 22 | 24 | import { MCP_SERVER, mcpConfig, RESUMED_LINE, resumeInterrupted, say, unendedLine, TURNS_LIVE_MAX, type ConversationOptions } from "../src/crew/conversation.ts"; |
| 23 | 25 | |
| @@ -34,7 +36,7 @@ const options = (binary: string, signal = new AbortController().signal): Convers | ||
| 34 | 36 | }); |
| 35 | 37 | |
| 36 | 38 | /** A stand-in `claude`: records what it was given, answers, and holds while told to. */ |
| 37 | function stub(name: string, sid: string, mode: "answer" | "hang" | "steer" | "fail", gone = false): string { | |
| 39 | function stub(name: string, sid: string, mode: "answer" | "hang" | "steer" | "fail" | "gate", gone = false): string { | |
| 38 | 40 | const path = join(dir, name); |
| 39 | 41 | writeFileSync( |
| 40 | 42 | path, |
| @@ -50,6 +52,7 @@ echo '{"type":"system","subtype":"init","session_id":"${sid}","model":"fake"}' | ||
| 50 | 52 | ${mode === "fail" ? "exit 1" : ""} |
| 51 | 53 | echo '{"type":"assistant","session_id":"${sid}","message":{"id":"m1","content":[{"type":"tool_use","name":"mcp__forge__now","input":{}}],"usage":{"input_tokens":10,"output_tokens":5}}}' |
| 52 | 54 | ${mode === "hang" ? "sleep 30" : ""} |
| 55 | ${mode === "gate" ? `while [ ! -f "${dir}/${name}.gate" ]; do sleep 0.05; done` : ""} | |
| 53 | 56 | ${mode === "steer" ? `IFS= read -r second; printf '%s' "$second" > "${dir}/${name}.second"` : ""} |
| 54 | 57 | echo '{"type":"assistant","session_id":"${sid}","message":{"id":"m2","content":[{"type":"text","text":"All quiet."}],"usage":{"input_tokens":20,"output_tokens":9}}}' |
| 55 | 58 | RESULT='{"type":"result","subtype":"success","is_error":false,"result":"All quiet.","total_cost_usd":0.5,"num_turns":2,"session_id":"${sid}","terminal_reason":"completed","usage":{"input_tokens":20,"output_tokens":9}}' |
| @@ -176,6 +179,64 @@ echo '{"type":"result","subtype":"success","is_error":false,"result":"Back.","to | ||
| 176 | 179 | expect(JSON.parse(readFileSync(join(dir, "steered.second"), "utf8")).message.content[0].text).toBe("two"); |
| 177 | 180 | }); |
| 178 | 181 | |
| 182 | test("a reply in progress is read from the stream, dropped halfway, and resumed from the last event id (#8815)", async () => { | |
| 183 | const bin = stub("gated", "sess-g", "gate"); | |
| 184 | const servers = await serve(app); | |
| 185 | const cookie = `forge_session=${createSession(app.db, alice.id).token}`; | |
| 186 | const conversation = openConversation(app.db, alice.id); | |
| 187 | const base = listTurns(app.db, alice.id, conversation.id, { limit: 1 }); | |
| 188 | const from = base.ok && base.value.length > 0 ? base.value.at(-1)!.id : ""; | |
| 189 | const parse = (text: string) => | |
| 190 | text.split("\n\n").flatMap((f) => { | |
| 191 | const event = /^event: (.*)$/m.exec(f)?.[1]; | |
| 192 | return event === undefined ? [] : [{ event, id: /^id: (.*)$/m.exec(f)?.[1] }]; | |
| 193 | }); | |
| 194 | const turnIds = (text: string) => parse(text).filter((f) => f.event === "turn").map((f) => f.id); | |
| 195 | let started: Promise<void> | undefined; | |
| 196 | try { | |
| 197 | const a = say(app, options(bin), alice.id, "stream me"); | |
| 198 | if (!a.ok || a.value.how !== "started") throw new Error("not started"); | |
| 199 | started = a.value.done; | |
| 200 | // First connection: read until the tool call has streamed (the person's message, then the call), then drop. | |
| 201 | const drop = new AbortController(); | |
| 202 | const first = await fetch(`${servers.url}/talk/stream?after=${from}`, { headers: { cookie }, signal: drop.signal }); | |
| 203 | const reader = first.body!.getReader(); | |
| 204 | let seen = ""; | |
| 205 | while (turnIds(seen).length < 2) { | |
| 206 | const chunk = await reader.read(); | |
| 207 | if (chunk.done) break; | |
| 208 | seen += new TextDecoder().decode(chunk.value); | |
| 209 | } | |
| 210 | drop.abort(); | |
| 211 | const before = turnIds(seen); | |
| 212 | expect(before.length).toBe(2); | |
| 213 | // Another device opening mid-turn sees the same so far. | |
| 214 | const second = new AbortController(); | |
| 215 | const other = await fetch(`${servers.url}/talk/stream`, { headers: { cookie, "last-event-id": from }, signal: second.signal }); | |
| 216 | const otherReader = other.body!.getReader(); | |
| 217 | let otherSeen = ""; | |
| 218 | while (turnIds(otherSeen).length < 2) { | |
| 219 | const chunk = await otherReader.read(); | |
| 220 | if (chunk.done) break; | |
| 221 | otherSeen += new TextDecoder().decode(chunk.value); | |
| 222 | } | |
| 223 | second.abort(); | |
| 224 | expect(turnIds(otherSeen)).toEqual(before); | |
| 225 | writeFileSync(join(dir, "gated.gate"), ""); | |
| 226 | await started; | |
| 227 | const rest = parse(await (await fetch(`${servers.url}/talk/stream`, { headers: { cookie, "last-event-id": before[1]! } })).text()); | |
| 228 | const all = turnIds(await (await fetch(`${servers.url}/talk/stream?after=${from}`, { headers: { cookie } })).text()); | |
| 229 | expect(all.slice(0, 2)).toEqual(before); | |
| 230 | expect(rest.filter((f) => f.event === "turn").map((f) => f.id)).toEqual(all.slice(2)); | |
| 231 | expect(all.length).toBe(3); | |
| 232 | expect(rest.at(-1)!.event).toBe("done"); | |
| 233 | } finally { | |
| 234 | writeFileSync(join(dir, "gated.gate"), ""); | |
| 235 | await started; | |
| 236 | servers.stop(); | |
| 237 | } | |
| 238 | }); | |
| 239 | ||
| 179 | 240 | test("a restart mid-turn: found at boot and resumed once from the same session", async () => { |
| 180 | 241 | const stop = new AbortController(); |
| 181 | 242 | const hang = stub("cut", "sess-cut", "hang"); |
new above
offline