ff4a06d · 76eb5ef · 5dd71af · 0f9bbbe · 4687713 · 8fdbb90
@@ -0,0 +1,86 @@
/**
 * The writes of a conversation's agent, kept so a resumed turn does none
 * twice (#8254). Each is recorded before it runs and completed with its
 * answer. A turn the restart cut leaves its records marked `resumed`; the
 * same write sent again by the resumed turn is answered from its record —
 * once, so a deliberate repeat after that goes ahead. A fresh turn starts
 * with none. Plain data in and out; the door is `http/mcp.ts`.
 */
import type { Database } from "bun:sqlite";
import { sha256 } from "../kernel/hash.ts";
import { id } from "../kernel/id.ts";
/** The longest answer kept to tell a resumed turn what was done. */
export const RESULT_MAX = 4000;
/** The arguments as one string whatever the order of their keys. */
const canonical = (v: unknown): string =>
  Array.isArray(v)
    ? `[${v.map(canonical).join(",")}]`
    : v !== null && typeof v === "object"
      ? `{${Object.entries(v)
          .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
          .map(([k, x]) => `${JSON.stringify(k)}:${canonical(x)}`)
          .join(",")}}`
      : (JSON.stringify(v) ?? "null");
export const argsHash = (args: Record<string, unknown>): string => sha256(canonical(args));
/** How long a record is kept: a resume comes within minutes, so a day is far past it. */
export const KEEP_MS = 24 * 60 * 60 * 1000;
/** A write about to run: its record's id. Records past `KEEP_MS` go with it, so none outlives its conversation for long. */
export function recordWrite(db: Database, via: string, tool: string, args: Record<string, unknown>, now: number = Date.now()): string {
  const rid = id();
  db.query("DELETE FROM conversation_write WHERE created_at < ?1").run(new Date(now - KEEP_MS).toISOString());
  db.query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES (?1,?2,?3,?4,?5)").run(
    rid,
    via,
    tool,
    argsHash(args),
    new Date(now).toISOString(),
  );
  return rid;
}
/** The write ended: what it answered, kept for a resumed turn. */
export function completeWrite(db: Database, recordId: string, result: string): void {
  db.query("UPDATE conversation_write SET result = ?2 WHERE id = ?1").run(recordId, result.slice(0, RESULT_MAX));
}
/** The write failed: nothing was done, so nothing is kept to say it was. */
export function dropWrite(db: Database, recordId: string): void {
  db.query("DELETE FROM conversation_write WHERE id = ?1").run(recordId);
}
/** A fresh turn: the last turn's writes are history. */
export function forgetWrites(db: Database, via: string): void {
  db.query("DELETE FROM conversation_write WHERE via = ?1").run(via);
}
/** The turn was cut: what it recorded is what its resumption must not repeat. */
export function carryWrites(db: Database, via: string): void {
  db.query("UPDATE conversation_write SET resumed = 1 WHERE via = ?1").run(via);
}
/**
 * The record of this write from the turn that was cut, claimed once: its
 * answer, or `null` as the answer when it began and never ended. Undefined
 * when the cut turn did not send it.
 */
export function claimDone(
  db: Database,
  via: string,
  tool: string,
  args: Record<string, unknown>,
): { result: string | null } | undefined {
  const row = db
    .query<{ id: string; result: string | null }, [string, string, string]>(
      "SELECT id, result FROM conversation_write WHERE via = ?1 AND tool = ?2 AND args_hash = ?3 AND resumed = 1 ORDER BY id LIMIT 1",
    )
    .get(via, tool, argsHash(args));
  if (row === null) return undefined;
  db.query("DELETE FROM conversation_write WHERE id = ?1").run(row.id);
  return { result: row.result };
}
@@ -0,0 +1,1 @@
31477f4f2cfeb49850f31d7e2374b08407cd3fef58d6bfb120623c392492093c
@@ -0,0 +1,17 @@
// Each write the forge's agent sends is recorded before it runs (#8254): the
// tool, a hash of its arguments, and later what it answered. A turn the
// restart cut is resumed from a session that may not have heard how a write
// ended; the record lets the forge answer the same write again with what was
// done, instead of doing it twice. Only the hash is kept, never the arguments.
export default `
  CREATE TABLE conversation_write (
    id         TEXT PRIMARY KEY,
    via        TEXT NOT NULL,
    tool       TEXT NOT NULL,
    args_hash  TEXT NOT NULL,
    result     TEXT,
    resumed    INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL
  );
  CREATE INDEX conversation_write_via ON conversation_write (via, tool, args_hash);
  `;
@@ -9,8 +9,10 @@
 * A message sent while a turn runs is steered into it (`steerRun`). A turn
 * the forge's restart cut short is found at boot (`resumeInterrupted`, as
 * `interrupted.ts` finds runs) and resumed once from its session. Resuming
 * repeats no write: a held write is single-use (`core/fence.ts`), and the
 * resumed turn is told to check what is done before doing anything.
 * repeats no write: a held write is single-use (`core/fence.ts`), every write
 * the cut turn sent is recorded and answered from its record if sent again
 * (`core/conversation_writes.ts`), and the resumed turn is told to check
      
        sentinelcrew
        12d ago
        outdated
      
      Rewrap this line to match the comment width around it.
    
 * what is done before doing anything.
 */
import { mkdirSync } from "node:fs";
@@ -18,6 +20,7 @@
import { join } from "node:path";
import type { App } from "../app.ts";
import { dailyLeft } from "../core/account_ceiling.ts";
import { appendTurn, openConversation, removeTurn, setClaudeSession, type Conversation } from "../core/conversations.ts";
import { carryWrites, forgetWrites } from "../core/conversation_writes.ts";
import { WORDS_MAX } from "../core/fence.ts";
import { PLATFORM, pickSubscription } from "../core/subscriptions.ts";
import { getUserById } from "../core/users.ts";
@@ -148,6 +151,7 @@
export function resumeInterrupted(app: App, options: ConversationOptions): Promi
  for (const row of rows) {
    // The cut turn's token is revoked now, resumed or not.
    app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(`conversation:${row.conversation_id}`);
    carryWrites(app.db, `conversation:${row.conversation_id}`);
    app.db.query("UPDATE conversation_run SET state = 'failed', finished_at = ?2 WHERE id = ?1").run(row.id, new Date().toISOString());
    if (row.resumes >= 1) continue;
    const conversation = openConversation(app.db, row.user_id);
@@ -185,6 +189,8 @@
function begin(
  // A new turn's token replaces the last one's.
  const via = `conversation:${conversation.id}`;
  app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(via);
  // A fresh turn starts with no writes; a resumed one keeps what the cut turn did.
  if (resumes === 0) forgetWrites(app.db, via);
  const lifetime = Math.min(options.timeoutMs + 60_000, VIA_TOKEN_MAX_MS - 1000);
  const minted = mintViaToken(app.db, userId, via, prompt.slice(0, WORDS_MAX), new Date(now.getTime() + lifetime).toISOString());
  if (!minted.ok) return err("unauthorized");
@@ -32,7 +32,8 @@
import { fileReport } from "../actions/reports.ts";
import { REPORT_RECEIVED } from "../actions/report_words.ts";
import { BODY_MAX, REQUEST_MAX, TITLE_MAX } from "../core/reports.ts";
import { everydayOf } from "../core/everyday.ts";
import { type Args, visible, plansOnly, str, REPO_ARG, lean, fenceAnswer, heldWrite, askedFirst, readThrough } from "./mcp/shared.ts";
import { claimDone, completeWrite, dropWrite, recordWrite } from "../core/conversation_writes.ts";
import { type Args, alreadyDone, visible, plansOnly, str, REPO_ARG, lean, fenceAnswer, heldWrite, askedFirst, readThrough } from "./mcp/shared.ts";
import { CONVERSATION_BRIEF } from "../crew/conversation_brief.ts";
import { TOOLS } from "./mcp/tools.ts";
@@ -201,9 +202,23 @@
function rpcOne(
          ],
          isError: true,
        });
      // The forge's agent's write is recorded before it runs; one a cut turn
      // already sent is answered with what was done, not done again (#8254).
      const via = ctx.auth.through;
      const writing = via !== undefined && tool.writes && !plansOnly(tool, args);
      if (writing) {
        const done = claimDone(ctx.app.db, via, tool.name, args);
        if (done !== undefined) return reply(alreadyDone(tool.name, done.result));
      }
      const recorded = writing ? recordWrite(ctx.app.db, via, tool.name, args) : undefined;
      readThrough(ctx);
      return Promise.resolve(tool.run(ctx, user, args)).then((ran) => {
        const answer = fenceAnswer(ctx, ran);
        if (recorded !== undefined) {
          // A refused or failed write did nothing, so a resumed turn may send it again.
      
        sentinelcrew
        12d ago
        outdated
      
      Stores error: message for failed writes; a resumed turn then hears already_done for a write that did not happen. Only record ok answers (or drop the record on error), and test it.
    
          if (answer.ok) completeWrite(ctx.app.db, recorded, JSON.stringify(answer.value));
          else dropWrite(ctx.app.db, recorded);
        }
        return reply(
          answer.ok
            ? {
@@ -105,6 +105,20 @@
export function heldWrite(ctx: Ctx, user: User, tool: Tool, args: Args): Record<
  };
}
/**
 * A write the cut turn already sent, as the resumed agent reads it: nothing
 * is done again, and what was answered (or that it never ended) is said.
 */
export function alreadyDone(tool: string, result: string | null): Record<string, unknown> {
  const says =
    result === null
      ? "this write was started before the forge restarted and its end was not recorded, so it was not done again: read to see whether it happened, and send it again only if it did not"
      : "this write was already done before the forge restarted, so it was not done again; its answer then is in `was`";
  return {
    content: [{ type: "text", text: JSON.stringify({ already_done: true, tool, says, ...(result === null ? {} : { was: result }) }) }],
  };
}
/** A call whose arguments a person could not read whole is refused, not kept. */
const tooBig = (tool: string): Record<string, unknown> => ({
  content: [
@@ -0,0 +1,151 @@
/**
 * A resumed turn does not repeat a write that finished unreported (#8254):
 * the write is recorded before it runs; the turn is cut, resumed, and the
 * same call sent again is answered with what was done, once.
 */
import { heard } from "../src/core/fence.ts";
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createRepo } from "../src/actions/repos.ts";
import { argsHash, carryWrites, KEEP_MS, recordWrite } from "../src/core/conversation_writes.ts";
import { createIssue, getIssue } from "../src/core/issues.ts";
import { createUser, type User } from "../src/core/users.ts";
import { mintViaToken } from "../src/core/via_token.ts";
import { serve, type Servers } from "../src/http/server.ts";
import { promote } from "./helpers/members.ts";
let dataDir: string;
let app: App;
let servers: Servers;
let alice: User;
let rpcId = 1;
let worlds = 0;
async function call(token: string, name: string, args: Record<string, unknown>) {
  const res = await fetch(`${servers.url}/mcp`, {
    method: "POST",
    headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
    body: JSON.stringify({ jsonrpc: "2.0", id: rpcId++, method: "tools/call", params: { name, arguments: args } }),
  });
  const body: any = await res.json();
  return { text: body.result?.content?.[0]?.text as string, isError: body.result?.isError === true };
}
async function world() {
  const n = ++worlds;
  const made = await createRepo(app, alice, { name: `r${n}`, visibility: "public" });
  if (!made.ok) throw new Error(String(made.error));
  const issue = createIssue(app.db, { repoId: made.value.id, authorId: alice.id, title: "Ship it", body: "now" });
  if (!issue.ok) throw new Error(issue.error);
  const via = `conversation:r${n}`;
  const mint = () => {
    const m = mintViaToken(app.db, alice.id, via, `close ${issue.value.number} in alice/r${n}`, new Date(Date.now() + 600_000).toISOString());
    if (!m.ok) throw new Error(m.error);
    return m.value.token;
  };
  return { repo: `alice/r${n}`, repoId: made.value.id, number: issue.value.number, via, mint };
}
const state = (w: { repoId: string; number: number }) => {
  const i = getIssue(app.db, w.repoId, w.number);
  return i?.state;
};
const reopen = (w: { repoId: string; number: number }) =>
  app.db.query("UPDATE issue SET state = 'open' WHERE repo_id = ?1 AND number = ?2").run(w.repoId, w.number);
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-conversation-resume-writes-"));
  app = createApp({ dataDir, port: 0 });
  servers = await serve(app);
  const u = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
  if (!u.ok) throw new Error(u.error);
  promote(app.db, u.value);
  alice = u.value;
}, 30_000);
afterAll(async () => {
  servers.stop();
  app.close();
  await rm(dataDir, { recursive: true, force: true });
});
describe("a resumed turn", () => {
  test("a write recorded, the turn resumed, the same call sent: done once", async () => {
    const w = await world();
    const args = { repo: w.repo, number: w.number };
    const first = await call(w.mint(), "close", args);
    expect(first.isError).toBe(false);
    expect(state(w)).toBe("closed");
    // Its answer never reached the session; something reopened it meanwhile.
    reopen(w);
    carryWrites(app.db, w.via);
    const again = await call(w.mint(), "close", args);
    expect(JSON.parse(again.text)).toMatchObject({ already_done: true, tool: "close" });
    expect(JSON.parse(again.text).was).toContain("closed");
    expect(state(w)).toBe("open"); // not done twice
    // Once: a deliberate repeat after that goes ahead.
    const third = await call(w.mint(), "close", args);
    expect(JSON.parse(third.text).already_done).toBeUndefined();
    expect(state(w)).toBe("closed");
  });
  test("a write that began and never ended is not repeated either, and says so", async () => {
    const w = await world();
    const args = { repo: w.repo, number: w.number };
    app.db
      .query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES ('cut1', ?1, 'close', ?2, ?3)")
      .run(w.via, argsHash(args), new Date().toISOString());
    carryWrites(app.db, w.via);
    const again = await call(w.mint(), "close", args);
    expect(JSON.parse(again.text)).toMatchObject({ already_done: true });
    expect(JSON.parse(again.text).was).toBeUndefined();
    expect(state(w)).toBe("open");
  });
  test("a different call, or a turn that was not cut, is done as asked", async () => {
    const w = await world();
    const args = { repo: w.repo, number: w.number };
    await call(w.mint(), "close", args);
    reopen(w);
    // Not resumed: the same write again in an ordinary turn goes ahead.
    const same = await call(w.mint(), "close", args);
    expect(JSON.parse(same.text).already_done).toBeUndefined();
    expect(state(w)).toBe("closed");
    // Resumed, but another argument: not the same write.
    reopen(w);
    carryWrites(app.db, w.via);
    // The first answer was fenced text, so the person must have named the write.
    const token = w.mint();
    heard(app.db, w.via, `close ${w.repo} ${w.number} dropped`);
    const other = await call(token, "close", { ...args, reason: "dropped" });
    expect(JSON.parse(other.text).already_done).toBeUndefined();
    expect(state(w)).toBe("closed");
  });
  test("a write that failed is not answered as done when the cut turn is resumed", async () => {
    const w = await world();
    const args = { repo: w.repo, number: w.number + 99 };
    const failed = await call(w.mint(), "close", args);
    expect(failed.isError).toBe(true);
    carryWrites(app.db, w.via);
    const again = await call(w.mint(), "close", args);
    expect(again.text).not.toContain("already_done");
  });
  test("records older than a day go when the next write is recorded", () => {
    const via = "conversation:old";
    const old = recordWrite(app.db, via, "close", { a: 1 }, Date.now() - KEEP_MS - 1000);
    recordWrite(app.db, via, "close", { a: 2 });
    const left = app.db.query("SELECT id FROM conversation_write WHERE via = ?1").all(via);
    expect(left.length).toBe(1);
    expect(left.some((r: any) => r.id === old)).toBe(false);
  });
  test("argument order does not change the hash", () => {
    expect(argsHash({ a: 1, b: { c: [1, 2], d: "x" } })).toBe(argsHash({ b: { d: "x", c: [1, 2] }, a: 1 }));
    expect(argsHash({ a: 1 })).not.toBe(argsHash({ a: 2 }));
  });
});
@@ -154,12 +154,17 @@
describe("a conversation turn", () => {
    // Mid-turn the token is alice's own.
    const live = verifyToken(app.db, readFileSync(join(dir, "cut.token"), "utf8"));
    expect(live.ok && live.value.user.id).toBe(alice.id);
    // A write the cut turn sent, recorded as it began.
    const via = `conversation:${openConversation(app.db, alice.id).id}`;
    app.db.query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES ('w1', ?1, 'close', 'h', '')").run(via);
    stop.abort(); // the forge stops: the row stays running
    if (cut.ok) await cut.value.done;
    expect(app.db.query("SELECT state FROM conversation_run WHERE state = 'running'").all()).toHaveLength(1);
    const again = resumeInterrupted(app, options(stub("resumed", "sess-cut", "answer")));
    expect(again).toHaveLength(1);
    // The resumed turn keeps the cut turn's writes, marked so a repeat is answered from them.
    expect(app.db.query<{ resumed: number }, []>("SELECT resumed FROM conversation_write WHERE id = 'w1'").get()?.resumed).toBe(1);
    await Promise.all(again);
    const args = lines("resumed");
    expect(args[args.indexOf("--resume") + 1]).toBe("sess-cut");
		@@ -0,0 +1,86 @@
	1	/**
	2	* The writes of a conversation's agent, kept so a resumed turn does none
	3	* twice (#8254). Each is recorded before it runs and completed with its
	4	* answer. A turn the restart cut leaves its records marked `resumed`; the
	5	* same write sent again by the resumed turn is answered from its record —
	6	* once, so a deliberate repeat after that goes ahead. A fresh turn starts
	7	* with none. Plain data in and out; the door is `http/mcp.ts`.
	8	*/
	9
	10	import type { Database } from "bun:sqlite";
	11	import { sha256 } from "../kernel/hash.ts";
	12	import { id } from "../kernel/id.ts";
	13
	14	/** The longest answer kept to tell a resumed turn what was done. */
	15	export const RESULT_MAX = 4000;
	16
	17	/** The arguments as one string whatever the order of their keys. */
	18	const canonical = (v: unknown): string =>
	19	Array.isArray(v)
	20	? `[${v.map(canonical).join(",")}]`
	21	: v !== null && typeof v === "object"
	22	? `{${Object.entries(v)
	23	.sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
	24	.map(([k, x]) => `${JSON.stringify(k)}:${canonical(x)}`)
	25	.join(",")}}`
	26	: (JSON.stringify(v) ?? "null");
	27
	28	export const argsHash = (args: Record<string, unknown>): string => sha256(canonical(args));
	29
	30	/** How long a record is kept: a resume comes within minutes, so a day is far past it. */
	31	export const KEEP_MS = 24 * 60 * 60 * 1000;
	32
	33	/** A write about to run: its record's id. Records past `KEEP_MS` go with it, so none outlives its conversation for long. */
	34	export function recordWrite(db: Database, via: string, tool: string, args: Record<string, unknown>, now: number = Date.now()): string {
	35	const rid = id();
	36	db.query("DELETE FROM conversation_write WHERE created_at < ?1").run(new Date(now - KEEP_MS).toISOString());
	37	db.query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES (?1,?2,?3,?4,?5)").run(
	38	rid,
	39	via,
	40	tool,
	41	argsHash(args),
	42	new Date(now).toISOString(),
	43	);
	44	return rid;
	45	}
	46
	47	/** The write ended: what it answered, kept for a resumed turn. */
	48	export function completeWrite(db: Database, recordId: string, result: string): void {
	49	db.query("UPDATE conversation_write SET result = ?2 WHERE id = ?1").run(recordId, result.slice(0, RESULT_MAX));
	50	}
	51
	52	/** The write failed: nothing was done, so nothing is kept to say it was. */
	53	export function dropWrite(db: Database, recordId: string): void {
	54	db.query("DELETE FROM conversation_write WHERE id = ?1").run(recordId);
	55	}
	56
	57	/** A fresh turn: the last turn's writes are history. */
	58	export function forgetWrites(db: Database, via: string): void {
	59	db.query("DELETE FROM conversation_write WHERE via = ?1").run(via);
	60	}
	61
	62	/** The turn was cut: what it recorded is what its resumption must not repeat. */
	63	export function carryWrites(db: Database, via: string): void {
	64	db.query("UPDATE conversation_write SET resumed = 1 WHERE via = ?1").run(via);
	65	}
	66
	67	/**
	68	* The record of this write from the turn that was cut, claimed once: its
	69	* answer, or `null` as the answer when it began and never ended. Undefined
	70	* when the cut turn did not send it.
	71	*/
	72	export function claimDone(
	73	db: Database,
	74	via: string,
	75	tool: string,
	76	args: Record<string, unknown>,
	77	): { result: string \| null } \| undefined {
	78	const row = db
	79	.query<{ id: string; result: string \| null }, [string, string, string]>(
	80	"SELECT id, result FROM conversation_write WHERE via = ?1 AND tool = ?2 AND args_hash = ?3 AND resumed = 1 ORDER BY id LIMIT 1",
	81	)
	82	.get(via, tool, argsHash(args));
	83	if (row === null) return undefined;
	84	db.query("DELETE FROM conversation_write WHERE id = ?1").run(row.id);
	85	return { result: row.result };
	86	}
		@@ -0,0 +1,1 @@
	1	31477f4f2cfeb49850f31d7e2374b08407cd3fef58d6bfb120623c392492093c
		@@ -0,0 +1,17 @@
	1	// Each write the forge's agent sends is recorded before it runs (#8254): the
	2	// tool, a hash of its arguments, and later what it answered. A turn the
	3	// restart cut is resumed from a session that may not have heard how a write
	4	// ended; the record lets the forge answer the same write again with what was
	5	// done, instead of doing it twice. Only the hash is kept, never the arguments.
	6	export default `
	7	CREATE TABLE conversation_write (
	8	id TEXT PRIMARY KEY,
	9	via TEXT NOT NULL,
	10	tool TEXT NOT NULL,
	11	args_hash TEXT NOT NULL,
	12	result TEXT,
	13	resumed INTEGER NOT NULL DEFAULT 0,
	14	created_at TEXT NOT NULL
	15	);
	16	CREATE INDEX conversation_write_via ON conversation_write (via, tool, args_hash);
	17	`;
		@@ -9,8 +9,10 @@
9	9	* A message sent while a turn runs is steered into it (`steerRun`). A turn
10	10	* the forge's restart cut short is found at boot (`resumeInterrupted`, as
11	11	* `interrupted.ts` finds runs) and resumed once from its session. Resuming
12		* repeats no write: a held write is single-use (`core/fence.ts`), and the
13		* resumed turn is told to check what is done before doing anything.
	12	* repeats no write: a held write is single-use (`core/fence.ts`), every write
	13	* the cut turn sent is recorded and answered from its record if sent again
	14	* (`core/conversation_writes.ts`), and the resumed turn is told to check
		sentinelcrew 12d ago outdated Rewrap this line to match the comment width around it.
	15	* what is done before doing anything.
14	16	*/
15	17
16	18	import { mkdirSync } from "node:fs";
		@@ -18,6 +20,7 @@ import { join } from "node:path";
18	20	import type { App } from "../app.ts";
19	21	import { dailyLeft } from "../core/account_ceiling.ts";
20	22	import { appendTurn, openConversation, removeTurn, setClaudeSession, type Conversation } from "../core/conversations.ts";
	23	import { carryWrites, forgetWrites } from "../core/conversation_writes.ts";
21	24	import { WORDS_MAX } from "../core/fence.ts";
22	25	import { PLATFORM, pickSubscription } from "../core/subscriptions.ts";
23	26	import { getUserById } from "../core/users.ts";
		@@ -148,6 +151,7 @@ export function resumeInterrupted(app: App, options: ConversationOptions): Promi
148	151	for (const row of rows) {
149	152	// The cut turn's token is revoked now, resumed or not.
150	153	app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(`conversation:${row.conversation_id}`);
	154	carryWrites(app.db, `conversation:${row.conversation_id}`);
151	155	app.db.query("UPDATE conversation_run SET state = 'failed', finished_at = ?2 WHERE id = ?1").run(row.id, new Date().toISOString());
152	156	if (row.resumes >= 1) continue;
153	157	const conversation = openConversation(app.db, row.user_id);
		@@ -185,6 +189,8 @@ function begin(
185	189	// A new turn's token replaces the last one's.
186	190	const via = `conversation:${conversation.id}`;
187	191	app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(via);
	192	// A fresh turn starts with no writes; a resumed one keeps what the cut turn did.
	193	if (resumes === 0) forgetWrites(app.db, via);
188	194	const lifetime = Math.min(options.timeoutMs + 60_000, VIA_TOKEN_MAX_MS - 1000);
189	195	const minted = mintViaToken(app.db, userId, via, prompt.slice(0, WORDS_MAX), new Date(now.getTime() + lifetime).toISOString());
190	196	if (!minted.ok) return err("unauthorized");
		@@ -32,7 +32,8 @@ import { fileReport } from "../actions/reports.ts";
32	32	import { REPORT_RECEIVED } from "../actions/report_words.ts";
33	33	import { BODY_MAX, REQUEST_MAX, TITLE_MAX } from "../core/reports.ts";
34	34	import { everydayOf } from "../core/everyday.ts";
35		import { type Args, visible, plansOnly, str, REPO_ARG, lean, fenceAnswer, heldWrite, askedFirst, readThrough } from "./mcp/shared.ts";
	35	import { claimDone, completeWrite, dropWrite, recordWrite } from "../core/conversation_writes.ts";
	36	import { type Args, alreadyDone, visible, plansOnly, str, REPO_ARG, lean, fenceAnswer, heldWrite, askedFirst, readThrough } from "./mcp/shared.ts";
36	37	import { CONVERSATION_BRIEF } from "../crew/conversation_brief.ts";
37	38	import { TOOLS } from "./mcp/tools.ts";
38	39
		@@ -201,9 +202,23 @@ function rpcOne(
201	202	],
202	203	isError: true,
203	204	});
	205	// The forge's agent's write is recorded before it runs; one a cut turn
	206	// already sent is answered with what was done, not done again (#8254).
	207	const via = ctx.auth.through;
	208	const writing = via !== undefined && tool.writes && !plansOnly(tool, args);
	209	if (writing) {
	210	const done = claimDone(ctx.app.db, via, tool.name, args);
	211	if (done !== undefined) return reply(alreadyDone(tool.name, done.result));
	212	}
	213	const recorded = writing ? recordWrite(ctx.app.db, via, tool.name, args) : undefined;
204	214	readThrough(ctx);
205	215	return Promise.resolve(tool.run(ctx, user, args)).then((ran) => {
206	216	const answer = fenceAnswer(ctx, ran);
	217	if (recorded !== undefined) {
	218	// A refused or failed write did nothing, so a resumed turn may send it again.
		sentinelcrew 12d ago outdated Stores `error: message` for failed writes; a resumed turn then hears already_done for a write that did not happen. Only record ok answers (or drop the record on error), and test it.
	219	if (answer.ok) completeWrite(ctx.app.db, recorded, JSON.stringify(answer.value));
	220	else dropWrite(ctx.app.db, recorded);
	221	}
207	222	return reply(
208	223	answer.ok
209	224	? {
		@@ -105,6 +105,20 @@ export function heldWrite(ctx: Ctx, user: User, tool: Tool, args: Args): Record<
105	105	};
106	106	}
107	107
	108	/**
	109	* A write the cut turn already sent, as the resumed agent reads it: nothing
	110	* is done again, and what was answered (or that it never ended) is said.
	111	*/
	112	export function alreadyDone(tool: string, result: string \| null): Record<string, unknown> {
	113	const says =
	114	result === null
	115	? "this write was started before the forge restarted and its end was not recorded, so it was not done again: read to see whether it happened, and send it again only if it did not"
	116	: "this write was already done before the forge restarted, so it was not done again; its answer then is in `was`";
	117	return {
	118	content: [{ type: "text", text: JSON.stringify({ already_done: true, tool, says, ...(result === null ? {} : { was: result }) }) }],
	119	};
	120	}
	121
108	122	/** A call whose arguments a person could not read whole is refused, not kept. */
109	123	const tooBig = (tool: string): Record<string, unknown> => ({
110	124	content: [
		@@ -0,0 +1,151 @@
	1	/**
	2	* A resumed turn does not repeat a write that finished unreported (#8254):
	3	* the write is recorded before it runs; the turn is cut, resumed, and the
	4	* same call sent again is answered with what was done, once.
	5	*/
	6	import { heard } from "../src/core/fence.ts";
	7	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	8	import { mkdtemp, rm } from "node:fs/promises";
	9	import { tmpdir } from "node:os";
	10	import { join } from "node:path";
	11	import { createApp, type App } from "../src/app.ts";
	12	import { createRepo } from "../src/actions/repos.ts";
	13	import { argsHash, carryWrites, KEEP_MS, recordWrite } from "../src/core/conversation_writes.ts";
	14	import { createIssue, getIssue } from "../src/core/issues.ts";
	15	import { createUser, type User } from "../src/core/users.ts";
	16	import { mintViaToken } from "../src/core/via_token.ts";
	17	import { serve, type Servers } from "../src/http/server.ts";
	18	import { promote } from "./helpers/members.ts";
	19
	20	let dataDir: string;
	21	let app: App;
	22	let servers: Servers;
	23	let alice: User;
	24	let rpcId = 1;
	25	let worlds = 0;
	26
	27	async function call(token: string, name: string, args: Record<string, unknown>) {
	28	const res = await fetch(`${servers.url}/mcp`, {
	29	method: "POST",
	30	headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
	31	body: JSON.stringify({ jsonrpc: "2.0", id: rpcId++, method: "tools/call", params: { name, arguments: args } }),
	32	});
	33	const body: any = await res.json();
	34	return { text: body.result?.content?.[0]?.text as string, isError: body.result?.isError === true };
	35	}
	36
	37	async function world() {
	38	const n = ++worlds;
	39	const made = await createRepo(app, alice, { name: `r${n}`, visibility: "public" });
	40	if (!made.ok) throw new Error(String(made.error));
	41	const issue = createIssue(app.db, { repoId: made.value.id, authorId: alice.id, title: "Ship it", body: "now" });
	42	if (!issue.ok) throw new Error(issue.error);
	43	const via = `conversation:r${n}`;
	44	const mint = () => {
	45	const m = mintViaToken(app.db, alice.id, via, `close ${issue.value.number} in alice/r${n}`, new Date(Date.now() + 600_000).toISOString());
	46	if (!m.ok) throw new Error(m.error);
	47	return m.value.token;
	48	};
	49	return { repo: `alice/r${n}`, repoId: made.value.id, number: issue.value.number, via, mint };
	50	}
	51
	52	const state = (w: { repoId: string; number: number }) => {
	53	const i = getIssue(app.db, w.repoId, w.number);
	54	return i?.state;
	55	};
	56	const reopen = (w: { repoId: string; number: number }) =>
	57	app.db.query("UPDATE issue SET state = 'open' WHERE repo_id = ?1 AND number = ?2").run(w.repoId, w.number);
	58
	59	beforeAll(async () => {
	60	dataDir = await mkdtemp(join(tmpdir(), "forge-conversation-resume-writes-"));
	61	app = createApp({ dataDir, port: 0 });
	62	servers = await serve(app);
	63	const u = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
	64	if (!u.ok) throw new Error(u.error);
	65	promote(app.db, u.value);
	66	alice = u.value;
	67	}, 30_000);
	68
	69	afterAll(async () => {
	70	servers.stop();
	71	app.close();
	72	await rm(dataDir, { recursive: true, force: true });
	73	});
	74
	75	describe("a resumed turn", () => {
	76	test("a write recorded, the turn resumed, the same call sent: done once", async () => {
	77	const w = await world();
	78	const args = { repo: w.repo, number: w.number };
	79	const first = await call(w.mint(), "close", args);
	80	expect(first.isError).toBe(false);
	81	expect(state(w)).toBe("closed");
	82	// Its answer never reached the session; something reopened it meanwhile.
	83	reopen(w);
	84	carryWrites(app.db, w.via);
	85	const again = await call(w.mint(), "close", args);
	86	expect(JSON.parse(again.text)).toMatchObject({ already_done: true, tool: "close" });
	87	expect(JSON.parse(again.text).was).toContain("closed");
	88	expect(state(w)).toBe("open"); // not done twice
	89	// Once: a deliberate repeat after that goes ahead.
	90	const third = await call(w.mint(), "close", args);
	91	expect(JSON.parse(third.text).already_done).toBeUndefined();
	92	expect(state(w)).toBe("closed");
	93	});
	94
	95	test("a write that began and never ended is not repeated either, and says so", async () => {
	96	const w = await world();
	97	const args = { repo: w.repo, number: w.number };
	98	app.db
	99	.query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES ('cut1', ?1, 'close', ?2, ?3)")
	100	.run(w.via, argsHash(args), new Date().toISOString());
	101	carryWrites(app.db, w.via);
	102	const again = await call(w.mint(), "close", args);
	103	expect(JSON.parse(again.text)).toMatchObject({ already_done: true });
	104	expect(JSON.parse(again.text).was).toBeUndefined();
	105	expect(state(w)).toBe("open");
	106	});
	107
	108	test("a different call, or a turn that was not cut, is done as asked", async () => {
	109	const w = await world();
	110	const args = { repo: w.repo, number: w.number };
	111	await call(w.mint(), "close", args);
	112	reopen(w);
	113	// Not resumed: the same write again in an ordinary turn goes ahead.
	114	const same = await call(w.mint(), "close", args);
	115	expect(JSON.parse(same.text).already_done).toBeUndefined();
	116	expect(state(w)).toBe("closed");
	117	// Resumed, but another argument: not the same write.
	118	reopen(w);
	119	carryWrites(app.db, w.via);
	120	// The first answer was fenced text, so the person must have named the write.
	121	const token = w.mint();
	122	heard(app.db, w.via, `close ${w.repo} ${w.number} dropped`);
	123	const other = await call(token, "close", { ...args, reason: "dropped" });
	124	expect(JSON.parse(other.text).already_done).toBeUndefined();
	125	expect(state(w)).toBe("closed");
	126	});
	127
	128	test("a write that failed is not answered as done when the cut turn is resumed", async () => {
	129	const w = await world();
	130	const args = { repo: w.repo, number: w.number + 99 };
	131	const failed = await call(w.mint(), "close", args);
	132	expect(failed.isError).toBe(true);
	133	carryWrites(app.db, w.via);
	134	const again = await call(w.mint(), "close", args);
	135	expect(again.text).not.toContain("already_done");
	136	});
	137
	138	test("records older than a day go when the next write is recorded", () => {
	139	const via = "conversation:old";
	140	const old = recordWrite(app.db, via, "close", { a: 1 }, Date.now() - KEEP_MS - 1000);
	141	recordWrite(app.db, via, "close", { a: 2 });
	142	const left = app.db.query("SELECT id FROM conversation_write WHERE via = ?1").all(via);
	143	expect(left.length).toBe(1);
	144	expect(left.some((r: any) => r.id === old)).toBe(false);
	145	});
	146
	147	test("argument order does not change the hash", () => {
	148	expect(argsHash({ a: 1, b: { c: [1, 2], d: "x" } })).toBe(argsHash({ b: { d: "x", c: [1, 2] }, a: 1 }));
	149	expect(argsHash({ a: 1 })).not.toBe(argsHash({ a: 2 }));
	150	});
	151	});
		@@ -154,12 +154,17 @@ describe("a conversation turn", () => {
154	154	// Mid-turn the token is alice's own.
155	155	const live = verifyToken(app.db, readFileSync(join(dir, "cut.token"), "utf8"));
156	156	expect(live.ok && live.value.user.id).toBe(alice.id);
	157	// A write the cut turn sent, recorded as it began.
	158	const via = `conversation:${openConversation(app.db, alice.id).id}`;
	159	app.db.query("INSERT INTO conversation_write (id, via, tool, args_hash, created_at) VALUES ('w1', ?1, 'close', 'h', '')").run(via);
157	160	stop.abort(); // the forge stops: the row stays running
158	161	if (cut.ok) await cut.value.done;
159	162	expect(app.db.query("SELECT state FROM conversation_run WHERE state = 'running'").all()).toHaveLength(1);
160	163
161	164	const again = resumeInterrupted(app, options(stub("resumed", "sess-cut", "answer")));
162	165	expect(again).toHaveLength(1);
	166	// The resumed turn keeps the cut turn's writes, marked so a repeat is answered from them.
	167	expect(app.db.query<{ resumed: number }, []>("SELECT resumed FROM conversation_write WHERE id = 'w1'").get()?.resumed).toBe(1);
163	168	await Promise.all(again);
164	169	const args = lines("resumed");
165	170	expect(args[args.indexOf("--resume") + 1]).toBe("sess-cut");