Public: anyone can read this; the crew can change it.
A resumed turn of the forge's agent cannot repeat a write that already happened #8410
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/8408-a-resumed-turn-of-the-forge-s-agent-cann → main 3 commits fork 3 threads · 3 unresolved
5 files
changed +149
−5
docs/features/talk-resume-writes.md added +32
| @@ -0,0 +1,32 @@ | ||
| 1 | # A write the restart cut short, in a conversation | |
| 2 | ||
| 3 | When the forge restarts while its agent is answering you on `/talk`, the turn | |
| 4 | is picked up where it stopped (`resumeInterrupted` in | |
| 5 | `src/crew/conversation.ts`). If a write the agent had started never recorded | |
| 6 | how it ended, for example a `close` or a `merge`, you see one extra line | |
| 7 | after "The forge restarted…": | |
| 8 | ||
| 9 | > The restart came while this write was running, and I do not know whether | |
| 10 | > it finished: close. Until this answer ends, it cannot run again unchecked; | |
| 11 | > please look, and if it still must be done, ask me after this answer and I | |
| 12 | > will do it then. | |
| 13 | ||
| 14 | The line names only the writes that started and never ended, not the ones that | |
| 15 | finished. If every write ended, there is no line. | |
| 16 | ||
| 17 | ## Getting the write done again | |
| 18 | ||
| 19 | 1. Check whether it happened: open the issue, pull or page it touched. | |
| 20 | 2. If it did not happen, wait for the resumed answer to finish, then send a | |
| 21 | new message asking for it, e.g. "close issue 6 again". | |
| 22 | ||
| 23 | That message starts a fresh turn, and a fresh turn forgets the cut turn's | |
| 24 | write records (`forgetWrites`, `src/core/conversation_writes.ts`), so the same | |
| 25 | write runs the first time it is sent. During the resumed answer itself, | |
| 26 | sending the identical write again is answered `already_done` once | |
| 27 | (`claimDone`, `src/http/mcp.ts`), so the restart never makes a write run | |
| 28 | twice without anyone checking. | |
| 29 | ||
| 30 | The line is an ordinary agent turn. The page draws it with `renderMarkdown`, | |
| 31 | which escapes its input (`src/web/pages/talk.ts`), and `forge talk` prints it | |
| 32 | through `inert` (`src/cli/commands/talk.ts`). |
src/core/conversation_writes.ts +8
| @@ -84,3 +84,11 @@ export function claimDone( | ||
| 84 | 84 | db.query("DELETE FROM conversation_write WHERE id = ?1").run(row.id); |
| 85 | 85 | return { result: row.result }; |
| 86 | 86 | } |
| 87 | ||
| 88 | /** The tools of writes this turn began and never recorded the end of: what a person is shown on resume. */ | |
| 89 | export function unended(db: Database, via: string): string[] { | |
| 90 | return db | |
| 91 | .query<{ tool: string }, [string]>("SELECT tool FROM conversation_write WHERE via = ?1 AND result IS NULL ORDER BY id") | |
| 92 | .all(via) | |
| 93 | .map((r) => r.tool); | |
| 94 | } |
src/crew/conversation.ts +20 −2
| @@ -20,7 +20,7 @@ import { join } from "node:path"; | ||
| 20 | 20 | import type { App } from "../app.ts"; |
| 21 | 21 | import { dailyLeft } from "../core/account_ceiling.ts"; |
| 22 | 22 | import { appendTurn, openConversation, removeTurn, setClaudeSession, type Conversation } from "../core/conversations.ts"; |
| 23 | import { carryWrites, forgetWrites } from "../core/conversation_writes.ts"; | |
| 23 | import { carryWrites, forgetWrites, unended } from "../core/conversation_writes.ts"; | |
| 24 | 24 | import { WORDS_MAX } from "../core/fence.ts"; |
| 25 | 25 | import { PLATFORM, pickSubscription, type Pick } from "../core/subscriptions.ts"; |
| 26 | 26 | import { getUserById } from "../core/users.ts"; |
| @@ -77,8 +77,23 @@ export const turnRunning = (app: App, userId: string): boolean => live.has(openC | ||
| 77 | 77 | /** The one line the person sees when a restart cut their answer short. */ |
| 78 | 78 | export const RESUMED_LINE = "The forge restarted while I was answering; I am picking it up where I stopped."; |
| 79 | 79 | |
| 80 | /** | |
| 81 | * The line naming writes the restart cut between doing and reporting. It | |
| 82 | * promises what happens: during the resumed answer the cut turn's record | |
outdated
Promises the write can be done again, but claimDone answers already_done for the same call. Make the promise true (and tested) or reword it. The nested singular/plural ternaries are also hard to read; split them into named words. |
||
| 83 | * stops the same write running blind, and the person's next message after | |
| 84 | * that answer starts a fresh turn, which forgets the records | |
| 85 | * (`forgetWrites` in `begin`), so the write asked for then runs. | |
| 86 | */ | |
| 87 | export const unendedLine = (tools: readonly string[]): string => { | |
| 88 | const one = tools.length === 1; | |
| 89 | const subject = one ? "this write was" : "these writes were"; | |
| 90 | const pronoun = one ? "it" : "they"; | |
| 91 | const object = one ? "it" : "them"; | |
| 92 | return `The restart came while ${subject} running, and I do not know whether ${pronoun} finished: ${[...new Set(tools)].join(", ")}. Until this answer ends, ${pronoun} cannot run again unchecked; please look, and if ${pronoun} still must be done, ask me after this answer and I will do ${object} then.`; | |
| 93 | }; | |
| 94 | ||
| 80 | 95 | const RESUME_PROMPT = |
| 81 | "The forge restarted while you were answering and you are resuming this conversation. Before any write, read to see whether it already happened; a held write was answered once and is not to be asked again. Then finish your answer to the person's last message."; | |
| 96 | "The forge restarted while you were answering and you are resuming this conversation. Before any write, read to see whether it already happened; a held write's confirmation was answered once and is not to be asked for again. Then finish your answer to the person's last message."; | |
| 82 | 97 | |
| 83 | 98 | /** Turns with a session listening, by conversation id: the run id a message is steered to. */ |
| 84 | 99 | const live = new Map<string, string>(); |
| @@ -152,12 +167,15 @@ export function resumeInterrupted(app: App, options: ConversationOptions): Promi | ||
| 152 | 167 | for (const row of rows) { |
| 153 | 168 | // The cut turn's token is revoked now, resumed or not. |
| 154 | 169 | app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(`conversation:${row.conversation_id}`); |
| 170 | const unknown = unended(app.db, `conversation:${row.conversation_id}`); | |
| 155 | 171 | carryWrites(app.db, `conversation:${row.conversation_id}`); |
| 156 | 172 | app.db.query("UPDATE conversation_run SET state = 'failed', finished_at = ?2 WHERE id = ?1").run(row.id, new Date().toISOString()); |
| 157 | 173 | if (row.resumes >= 1) continue; |
| 158 | 174 | const conversation = openConversation(app.db, row.user_id); |
| 159 | 175 | if (conversation.id !== row.conversation_id) continue; |
| 160 | 176 | appendTurn(app.db, row.user_id, conversation.id, { role: "agent", text: RESUMED_LINE }); |
| 177 | // A write begun and not reported is shown, not retried: the person decides. | |
| 178 | if (unknown.length > 0) appendTurn(app.db, row.user_id, conversation.id, { role: "agent", text: unendedLine(unknown) }); | |
| 161 | 179 | const session = conversation.claudeSessionId || row.session_id; |
| 162 | 180 | const started = begin(app, options, conversation, row.prompt, row.resumes + 1, session === "" ? undefined : session); |
| 163 | 181 | if (started.ok) done.push(started.value); |
test/conversation-resume-writes.test.ts +28 −1
| @@ -10,7 +10,7 @@ import { tmpdir } from "node:os"; | ||
| 10 | 10 | import { join } from "node:path"; |
| 11 | 11 | import { createApp, type App } from "../src/app.ts"; |
| 12 | 12 | import { createRepo } from "../src/actions/repos.ts"; |
| 13 | import { argsHash, carryWrites, KEEP_MS, recordWrite } from "../src/core/conversation_writes.ts"; | |
| 13 | import { argsHash, carryWrites, completeWrite, KEEP_MS, recordWrite, unended } from "../src/core/conversation_writes.ts"; | |
| 14 | 14 | import { createIssue, getIssue } from "../src/core/issues.ts"; |
| 15 | 15 | import { createUser, type User } from "../src/core/users.ts"; |
| 16 | 16 | import { mintViaToken } from "../src/core/via_token.ts"; |
| @@ -105,6 +105,33 @@ describe("a resumed turn", () => { | ||
| 105 | 105 | expect(state(w)).toBe("open"); |
| 106 | 106 | }); |
| 107 | 107 | |
| 108 | test("a write the cut turn began and never ended is answered once, then the identical call runs", async () => { | |
| 109 | const w = await world(); | |
| 110 | const args = { repo: w.repo, number: w.number }; | |
| 111 | recordWrite(app.db, w.via, "close", args); | |
| 112 | carryWrites(app.db, w.via); | |
| 113 | expect(JSON.parse((await call(w.mint(), "close", args)).text).already_done).toBe(true); | |
| 114 | expect(state(w)).toBe("open"); | |
| 115 | // The record is claimed by the first answer: the same call now goes ahead. | |
| 116 | const repeat = await call(w.mint(), "close", args); | |
| 117 | expect(JSON.parse(repeat.text).already_done).toBeUndefined(); | |
| 118 | expect(state(w)).toBe("closed"); | |
| 119 | }); | |
| 120 | ||
| 121 | test("unended: only rows with no result, for that via, in id order", () => { | |
| 122 | const via = "conversation:unended-a"; | |
| 123 | const insert = (rid: string, v: string, tool: string, result: string | null) => | |
| 124 | app.db.query("INSERT INTO conversation_write (id, via, tool, args_hash, result, created_at) VALUES (?1,?2,?3,'h',?4,?5)").run(rid, v, tool, result, new Date().toISOString()); | |
| 125 | insert("u3", via, "merge", null); | |
| 126 | insert("u1", via, "close", null); | |
| 127 | insert("u2", via, "comment", "done"); | |
| 128 | insert("u4", "conversation:unended-b", "reopen", null); | |
| 129 | expect(unended(app.db, via)).toEqual(["close", "merge"]); | |
| 130 | completeWrite(app.db, "u1", "closed"); | |
| 131 | expect(unended(app.db, via)).toEqual(["merge"]); | |
| 132 | expect(unended(app.db, "conversation:unended-none")).toEqual([]); | |
| 133 | }); | |
| 134 | ||
| 108 | 135 | test("a different call, or a turn that was not cut, is done as asked", async () => { |
| 109 | 136 | const w = await world(); |
| 110 | 137 | const args = { repo: w.repo, number: w.number }; |
test/conversation-turn.test.ts +61 −2
| @@ -12,14 +12,14 @@ import { tmpdir } from "node:os"; | ||
| 12 | 12 | import { join } from "node:path"; |
| 13 | 13 | import { createApp, type App } from "../src/app.ts"; |
| 14 | 14 | import { clearAccountCeiling, daySpent, setAccountCeiling } from "../src/core/account_ceiling.ts"; |
| 15 | import { recordWrite } from "../src/core/conversation_writes.ts"; | |
| 15 | import { claimDone, completeWrite, recordWrite } from "../src/core/conversation_writes.ts"; | |
| 16 | 16 | import { listTurns, openConversation } from "../src/core/conversations.ts"; |
| 17 | 17 | 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 | 21 | import { TOOLS } from "../src/crew/claude.ts"; |
| 22 | import { MCP_SERVER, mcpConfig, RESUMED_LINE, resumeInterrupted, say, TURNS_LIVE_MAX, type ConversationOptions } from "../src/crew/conversation.ts"; | |
| 22 | import { MCP_SERVER, mcpConfig, RESUMED_LINE, resumeInterrupted, say, unendedLine, TURNS_LIVE_MAX, type ConversationOptions } from "../src/crew/conversation.ts"; | |
| 23 | 23 | |
| 24 | 24 | let dir: string; |
| 25 | 25 | let app: App; |
| @@ -170,6 +170,8 @@ describe("a conversation turn", () => { | ||
| 170 | 170 | const args = lines("resumed"); |
| 171 | 171 | expect(args[args.indexOf("--resume") + 1]).toBe("sess-cut"); |
| 172 | 172 | expect(turns(alice).filter((t) => t === `agent:${RESUMED_LINE}`)).toHaveLength(1); |
| 173 | // The write with no outcome is shown to the person, once. | |
| 174 | expect(turns(alice).filter((t) => t === `agent:${unendedLine(["close"])}`)).toHaveLength(1); | |
| 173 | 175 | // Once: a second boot finds nothing to resume. |
| 174 | 176 | expect(resumeInterrupted(app, options(stub("never", "sess-cut", "answer")))).toHaveLength(0); |
| 175 | 177 | expect(existsSync(join(dir, "never.argv"))).toBe(false); |
| @@ -194,6 +196,63 @@ describe("a conversation turn", () => { | ||
| 194 | 196 | expect(carried(via)).toBe(false); |
| 195 | 197 | }); |
| 196 | 198 | |
| 199 | /** Cut a turn after `record` ran, resume it with a turn that sends nothing, and return the turns the resume added. */ | |
| 200 | async function cutAndResume(name: string, record: (via: string) => void): Promise<{ via: string; added: string[] }> { | |
| 201 | const stop = new AbortController(); | |
| 202 | const cut = say(app, options(stub(`${name}-cut`, `sess-${name}`, "hang"), stop.signal), alice.id, `do ${name}`); | |
| 203 | while (!existsSync(join(dir, `${name}-cut.first`))) await Bun.sleep(20); | |
| 204 | const via = `conversation:${openConversation(app.db, alice.id).id}`; | |
| 205 | record(via); | |
| 206 | stop.abort(); | |
| 207 | if (cut.ok) await cut.value.done; | |
| 208 | const before = turns(alice).length; | |
| 209 | await Promise.all(resumeInterrupted(app, options(stub(`${name}-resumed`, `sess-${name}`, "answer")))); | |
| 210 | return { via, added: turns(alice).slice(before) }; | |
| 211 | } | |
| 212 | const resumedTurns = (unknown?: string) => [ | |
| 213 | `agent:${RESUMED_LINE}`, | |
| 214 | ...(unknown === undefined ? [] : [`agent:${unknown}`]), | |
| 215 | expect.stringMatching(/^tool:mcp__forge__now/) as unknown as string, | |
| 216 | "agent:All quiet.", | |
| 217 | ]; | |
| 218 | ||
| 219 | test("a resume where every write ended appends no unended line", async () => { | |
| 220 | const { added } = await cutAndResume("ended", (via) => completeWrite(app.db, recordWrite(app.db, via, "close", { number: 3 }), "closed")); | |
| 221 | expect(added).toEqual(resumedTurns()); | |
| 222 | }); | |
| 223 | ||
| 224 | test("one write ended and one did not: the line names only the one that did not", async () => { | |
| 225 | const { added } = await cutAndResume("mixed", (via) => { | |
| 226 | completeWrite(app.db, recordWrite(app.db, via, "close", { number: 4 }), "closed"); | |
| 227 | recordWrite(app.db, via, "merge", { number: 5 }); | |
| 228 | }); | |
| 229 | expect(added).toEqual(resumedTurns(unendedLine(["merge"]))); | |
| 230 | }); | |
| 231 | ||
| 232 | test("the resumed turn sends nothing; the person asks after it, and the same write runs at once (#8410)", async () => { | |
| 233 | const args = { repo: "alice/x", number: 6 }; | |
| 234 | const { via, added } = await cutAndResume("asked", (v) => recordWrite(app.db, v, "close", args)); | |
| 235 | expect(added).toEqual(resumedTurns(unendedLine(["close"]))); | |
| 236 | // The record still stands after the resumed answer: an identical call now would be answered already_done. | |
| 237 | expect(app.db.query("SELECT 1 FROM conversation_write WHERE via = ?1 AND resumed = 1").get(via)).not.toBeNull(); | |
| 238 | const asked = say(app, options(stub("asked-again", "sess-asked", "answer")), alice.id, "close issue 6 again"); | |
| 239 | expect(asked.ok && asked.value.how).toBe("started"); | |
| 240 | if (asked.ok) await asked.value.done; | |
| 241 | // The person's turn is fresh: the door (`http/mcp.ts`, `claimDone`) finds no record, so the first identical call runs. | |
| 242 | expect(claimDone(app.db, via, "close", args)).toBeUndefined(); | |
| 243 | }); | |
| 244 | ||
| 245 | test("the unended line is worded for one write and for several", () => { | |
| 246 | const one = unendedLine(["close"]); | |
| 247 | expect(one).toContain("this write was running"); | |
| 248 | expect(one).toContain("whether it finished: close."); | |
| 249 | expect(one).toContain("ask me after this answer and I will do it then."); | |
| 250 | const many = unendedLine(["close", "merge", "close"]); | |
| 251 | expect(many).toContain("these writes were running"); | |
| 252 | expect(many).toContain("whether they finished: close, merge."); | |
| 253 | expect(many).toContain("ask me after this answer and I will do them then."); | |
| 254 | }); | |
| 255 | ||
| 197 | 256 | test("a person with no account and no admin is refused, and nothing is stored as a run", async () => { |
| 198 | 257 | const b = await createUser(app.db, { origin: "seed", trusted: true, username: "bob", email: "[email protected]", password: "a-good-passphrase" }); |
| 199 | 258 | if (!b.ok) throw new Error("user"); |
new above
offline
No direct test of unended(): add one for NULL-only, per-via and id order.