68b45d7
@@ -0,0 +1,189 @@
/**
 * The `flow` label's dispatcher, `src/crew/flow_label.ts`, imported and run
 * (#7512). Release 0.7.283 did not boot because of it (#7436), and no test
 * loaded it directly: `actions/issues.ts` loads it only when a label lands.
 * Imported first here, so a failure at import fails this file; then loaded
 * cold in a fresh process the way `src/main.ts` boots, through the label.
 *
 * Labelling a held issue `flow` starts its run, that issue only, paid by
 * whoever last assigned it; an issue that is not held, or has no assignee,
 * starts nothing and throws nothing.
 */
import { dispatchFlowLabelled } from "../src/crew/flow_label.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 { addToCrew, createAgent } from "../src/actions/crew.ts";
import { assignIssue, openIssue, setIssueState } from "../src/actions/issues.ts";
import { createRepo, updateRepoSettings } from "../src/actions/repos.ts";
import { getIssueById, getLabel, setIssueLabels, type Issue } from "../src/core/issues.ts";
import { getRepoById, type Repo } from "../src/core/repos.ts";
import { createUser, getUserByName, type User } from "../src/core/users.ts";
import { flowOnlyHolds } from "../src/crew/flow_only.ts";
import { sweepAssigned } from "../src/crew/dispatch.ts";
const BOOT = join(import.meta.dir, "helpers", "flow-label-boot.ts");
describe("the flow-label dispatcher", () => {
  let dataDir: string;
  let app: App;
  let alice: User;
  let gus: User;
  let bot: User;
  let repo: Repo;
  const fresh = () => getRepoById(app.db, repo.id)!;
  const open = (title: string): Issue => {
    const o = openIssue(app, alice, fresh(), { title, body: "" });
    if (!o.ok) throw new Error(String(o.error));
    return o.value;
  };
  const assign = (issue: Issue, to: User | null = bot): Issue => {
    const a = assignIssue(app, alice, fresh(), issue, to);
    if (!a.ok) throw new Error(String(a.error));
    return a.value;
  };
  const runsOn = (issue: Issue): number =>
    app.db
      .query<{ n: number }, [string]>("SELECT COUNT(*) AS n FROM run WHERE issue_id = ?1 AND kind = 'agent'")
      .get(issue.id)!.n;
  /** Who each run queued on the issue was announced as asked by: the account it is charged to. */
  const payers = (issue: Issue): (string | null)[] =>
    app.db
      .query<{ actor_id: string | null }, [string]>(
        `SELECT actor_id FROM event WHERE kind = 'run.queued' AND json_extract(payload, '$.issueId') = ?1 ORDER BY id`,
      )
      .all(issue.id)
      .map((r) => r.actor_id);
  /** Label `flow` as the store does, then hand the issue to the dispatcher. */
  const label = (issue: Issue): void => {
    setIssueLabels(app.db, issue.id, [getLabel(app.db, repo.id, "flow")!.id]);
    dispatchFlowLabelled(app, fresh(), getIssueById(app.db, issue.id)!);
  };
  /** An assignment recorded as a guest's: the dispatcher reads the assigner from the event. */
  const assignedBy = (issue: Issue, by: User): void => {
    app.bus.append({
      kind: "issue.assigned",
      repoId: repo.id,
      actorId: by.id,
      payload: { number: issue.number, assignee: bot.username },
    });
  };
  beforeAll(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-flow-label-dispatch-"));
    app = createApp({ dataDir, port: 0 });
    const a = await createUser(app.db, {
      origin: "seed",
      trusted: true,
      username: "alice",
      email: "[email protected]",
      password: "a-good-passphrase",
      adminIfFirst: true,
    });
    if (!a.ok) throw new Error(a.error);
    alice = a.value;
    const g = await createUser(app.db, {
      origin: "seed",
      trusted: true,
      username: "gus",
      email: "[email protected]",
      password: "a-good-passphrase",
    });
    if (!g.ok) throw new Error(g.error);
    gus = g.value;
    const made = await createRepo(app, alice, { name: "svc", visibility: "private" });
    if (!made.ok) throw new Error(String(made.error));
    repo = made.value;
    const created = await createAgent(app, alice, { username: "bot" });
    if (!created.ok) throw new Error(String(created.error));
    expect(addToCrew(app, alice, repo, "bot").ok).toBe(true);
    bot = getUserByName(app.db, "bot")!;
    const on = await updateRepoSettings(app, alice, fresh(), { flowOnly: true });
    if (!on.ok) throw new Error(JSON.stringify(on.error));
  });
  afterAll(async () => {
    app.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  test("a held issue labelled `flow` starts exactly one run, that issue only, charged to its assigner", () => {
    const held = assign(open("Held"));
    const beside = assign(open("Beside, also held"));
    sweepAssigned(app);
    expect(flowOnlyHolds(app.db, fresh(), held)).toBe(true);
    expect(runsOn(held)).toBe(0);
    expect(runsOn(beside)).toBe(0);
    label(held);
    expect(runsOn(held)).toBe(1);
    expect(runsOn(beside)).toBe(0);
    expect(payers(held)).toEqual([alice.id]);
    // Labelled again, or swept, it is the same one run.
    label(held);
    sweepAssigned(app);
    expect(runsOn(held)).toBe(1);
    expect(runsOn(beside)).toBe(0);
  });
  test("the last assigner pays: a guest's last word starts nothing, an owner's after it does", () => {
    // Assigned by alice, then last by a guest: a guest's word starts no run.
    const guestLast = assign(open("Guest last"));
    assignedBy(guestLast, gus);
    label(guestLast);
    expect(runsOn(guestLast)).toBe(0);
    expect(payers(guestLast)).toEqual([]);
    // A guest first, then alice: alice is the last assigner, and pays.
    const ownerLast = open("Owner last");
    assignedBy(ownerLast, gus);
    assign(ownerLast);
    label(ownerLast);
    expect(runsOn(ownerLast)).toBe(1);
    expect(payers(ownerLast)).toEqual([alice.id]);
  });
  test("an issue that is not held, or has no assignee, starts nothing and throws nothing", () => {
    const nobody = open("No assignee");
    expect(() => label(nobody)).not.toThrow();
    expect(runsOn(nobody)).toBe(0);
    const person = assign(open("A person's"), alice);
    expect(() => label(person)).not.toThrow();
    expect(runsOn(person)).toBe(0);
    // Assigned while already labelled `flow`: never held, and its one run is the assignment's.
    const flowing = open("Already flow");
    setIssueLabels(app.db, flowing.id, [getLabel(app.db, repo.id, "flow")!.id]);
    assign(flowing);
    sweepAssigned(app);
    expect(flowOnlyHolds(app.db, fresh(), getIssueById(app.db, flowing.id)!)).toBe(false);
    expect(runsOn(flowing)).toBe(1);
    expect(() => label(flowing)).not.toThrow();
    expect(runsOn(flowing)).toBe(1);
    const closed = assign(open("Closed"));
    const shut = setIssueState(app, alice, fresh(), closed, "closed");
    if (!shut.ok) throw new Error(String(shut.error));
    expect(() => label(closed)).not.toThrow();
    expect(runsOn(closed)).toBe(0);
  });
  test("loaded cold as the forge boots, the label on a held issue starts its run", async () => {
    const held = assign(open("Held at boot"));
    expect(runsOn(held)).toBe(0);
    const child = Bun.spawn([process.execPath, BOOT, dataDir, "alice", "svc", String(held.number)], {
      stdout: "pipe",
      stderr: "pipe",
      env: { PATH: process.env["PATH"] ?? "", HOME: process.env["HOME"] ?? "" },
    });
    const [out, err] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text()]);
    expect({ code: await child.exited, err: (await child.exited) === 0 ? "" : err }).toEqual({ code: 0, err: "" });
    expect(out.trim().split("\n").at(-1)).toBe("1");
    expect(runsOn(held)).toBe(1);
  }, 30_000);
});
@@ -0,0 +1,28 @@
/**
 * The `flow` label in a fresh process, loaded as `src/main.ts` loads the
 * forge at boot (`test/flow-label-dispatch.test.ts`): the app and the HTTP
 * server's module graph, then a top-level await, then an owner labels a held
 * issue `flow`. `actions/issues.ts` loads the dispatcher only at that call, so
 * a dispatcher that cannot load cold fails here and nowhere earlier.
 *
 * Printed: the number of agent runs on the issue afterwards.
 */
import { createApp } from "../../src/app.ts";
import "../../src/http/server.ts";
import { setIssueLabels } from "../../src/actions/issues.ts";
import { getIssue } from "../../src/core/issues.ts";
import { getRepo } from "../../src/core/repos.ts";
import { getUserByName } from "../../src/core/users.ts";
const [dataDir, owner, name, number] = process.argv.slice(2);
const app = createApp({ dataDir: dataDir!, port: 0 });
await Bun.sleep(0);
const repo = getRepo(app.db, owner!, name!)!;
const issue = getIssue(app.db, repo.id, Number(number))!;
const labelled = setIssueLabels(app, getUserByName(app.db, owner!)!, repo, issue, ["flow"]);
if (!labelled.ok) throw new Error(String(labelled.error));
const runs = app.db
  .query<{ n: number }, [string]>("SELECT COUNT(*) AS n FROM run WHERE issue_id = ?1 AND kind = 'agent'")
  .get(issue.id)!.n;
app.close();
console.log(runs);
		@@ -0,0 +1,189 @@
	1	/**
	2	* The `flow` label's dispatcher, `src/crew/flow_label.ts`, imported and run
	3	* (#7512). Release 0.7.283 did not boot because of it (#7436), and no test
	4	* loaded it directly: `actions/issues.ts` loads it only when a label lands.
	5	* Imported first here, so a failure at import fails this file; then loaded
	6	* cold in a fresh process the way `src/main.ts` boots, through the label.
	7	*
	8	* Labelling a held issue `flow` starts its run, that issue only, paid by
	9	* whoever last assigned it; an issue that is not held, or has no assignee,
	10	* starts nothing and throws nothing.
	11	*/
	12	import { dispatchFlowLabelled } from "../src/crew/flow_label.ts";
	13	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	14	import { mkdtemp, rm } from "node:fs/promises";
	15	import { tmpdir } from "node:os";
	16	import { join } from "node:path";
	17	import { createApp, type App } from "../src/app.ts";
	18	import { addToCrew, createAgent } from "../src/actions/crew.ts";
	19	import { assignIssue, openIssue, setIssueState } from "../src/actions/issues.ts";
	20	import { createRepo, updateRepoSettings } from "../src/actions/repos.ts";
	21	import { getIssueById, getLabel, setIssueLabels, type Issue } from "../src/core/issues.ts";
	22	import { getRepoById, type Repo } from "../src/core/repos.ts";
	23	import { createUser, getUserByName, type User } from "../src/core/users.ts";
	24	import { flowOnlyHolds } from "../src/crew/flow_only.ts";
	25	import { sweepAssigned } from "../src/crew/dispatch.ts";
	26
	27	const BOOT = join(import.meta.dir, "helpers", "flow-label-boot.ts");
	28
	29	describe("the flow-label dispatcher", () => {
	30	let dataDir: string;
	31	let app: App;
	32	let alice: User;
	33	let gus: User;
	34	let bot: User;
	35	let repo: Repo;
	36
	37	const fresh = () => getRepoById(app.db, repo.id)!;
	38	const open = (title: string): Issue => {
	39	const o = openIssue(app, alice, fresh(), { title, body: "" });
	40	if (!o.ok) throw new Error(String(o.error));
	41	return o.value;
	42	};
	43	const assign = (issue: Issue, to: User \| null = bot): Issue => {
	44	const a = assignIssue(app, alice, fresh(), issue, to);
	45	if (!a.ok) throw new Error(String(a.error));
	46	return a.value;
	47	};
	48	const runsOn = (issue: Issue): number =>
	49	app.db
	50	.query<{ n: number }, [string]>("SELECT COUNT(*) AS n FROM run WHERE issue_id = ?1 AND kind = 'agent'")
	51	.get(issue.id)!.n;
	52	/** Who each run queued on the issue was announced as asked by: the account it is charged to. */
	53	const payers = (issue: Issue): (string \| null)[] =>
	54	app.db
	55	.query<{ actor_id: string \| null }, [string]>(
	56	`SELECT actor_id FROM event WHERE kind = 'run.queued' AND json_extract(payload, '$.issueId') = ?1 ORDER BY id`,
	57	)
	58	.all(issue.id)
	59	.map((r) => r.actor_id);
	60	/** Label `flow` as the store does, then hand the issue to the dispatcher. */
	61	const label = (issue: Issue): void => {
	62	setIssueLabels(app.db, issue.id, [getLabel(app.db, repo.id, "flow")!.id]);
	63	dispatchFlowLabelled(app, fresh(), getIssueById(app.db, issue.id)!);
	64	};
	65	/** An assignment recorded as a guest's: the dispatcher reads the assigner from the event. */
	66	const assignedBy = (issue: Issue, by: User): void => {
	67	app.bus.append({
	68	kind: "issue.assigned",
	69	repoId: repo.id,
	70	actorId: by.id,
	71	payload: { number: issue.number, assignee: bot.username },
	72	});
	73	};
	74
	75	beforeAll(async () => {
	76	dataDir = await mkdtemp(join(tmpdir(), "forge-flow-label-dispatch-"));
	77	app = createApp({ dataDir, port: 0 });
	78	const a = await createUser(app.db, {
	79	origin: "seed",
	80	trusted: true,
	81	username: "alice",
	82	email: "[email protected]",
	83	password: "a-good-passphrase",
	84	adminIfFirst: true,
	85	});
	86	if (!a.ok) throw new Error(a.error);
	87	alice = a.value;
	88	const g = await createUser(app.db, {
	89	origin: "seed",
	90	trusted: true,
	91	username: "gus",
	92	email: "[email protected]",
	93	password: "a-good-passphrase",
	94	});
	95	if (!g.ok) throw new Error(g.error);
	96	gus = g.value;
	97	const made = await createRepo(app, alice, { name: "svc", visibility: "private" });
	98	if (!made.ok) throw new Error(String(made.error));
	99	repo = made.value;
	100	const created = await createAgent(app, alice, { username: "bot" });
	101	if (!created.ok) throw new Error(String(created.error));
	102	expect(addToCrew(app, alice, repo, "bot").ok).toBe(true);
	103	bot = getUserByName(app.db, "bot")!;
	104	const on = await updateRepoSettings(app, alice, fresh(), { flowOnly: true });
	105	if (!on.ok) throw new Error(JSON.stringify(on.error));
	106	});
	107
	108	afterAll(async () => {
	109	app.close();
	110	await rm(dataDir, { recursive: true, force: true });
	111	});
	112
	113	test("a held issue labelled `flow` starts exactly one run, that issue only, charged to its assigner", () => {
	114	const held = assign(open("Held"));
	115	const beside = assign(open("Beside, also held"));
	116	sweepAssigned(app);
	117	expect(flowOnlyHolds(app.db, fresh(), held)).toBe(true);
	118	expect(runsOn(held)).toBe(0);
	119	expect(runsOn(beside)).toBe(0);
	120
	121	label(held);
	122	expect(runsOn(held)).toBe(1);
	123	expect(runsOn(beside)).toBe(0);
	124	expect(payers(held)).toEqual([alice.id]);
	125
	126	// Labelled again, or swept, it is the same one run.
	127	label(held);
	128	sweepAssigned(app);
	129	expect(runsOn(held)).toBe(1);
	130	expect(runsOn(beside)).toBe(0);
	131	});
	132
	133	test("the last assigner pays: a guest's last word starts nothing, an owner's after it does", () => {
	134	// Assigned by alice, then last by a guest: a guest's word starts no run.
	135	const guestLast = assign(open("Guest last"));
	136	assignedBy(guestLast, gus);
	137	label(guestLast);
	138	expect(runsOn(guestLast)).toBe(0);
	139	expect(payers(guestLast)).toEqual([]);
	140
	141	// A guest first, then alice: alice is the last assigner, and pays.
	142	const ownerLast = open("Owner last");
	143	assignedBy(ownerLast, gus);
	144	assign(ownerLast);
	145	label(ownerLast);
	146	expect(runsOn(ownerLast)).toBe(1);
	147	expect(payers(ownerLast)).toEqual([alice.id]);
	148	});
	149
	150	test("an issue that is not held, or has no assignee, starts nothing and throws nothing", () => {
	151	const nobody = open("No assignee");
	152	expect(() => label(nobody)).not.toThrow();
	153	expect(runsOn(nobody)).toBe(0);
	154
	155	const person = assign(open("A person's"), alice);
	156	expect(() => label(person)).not.toThrow();
	157	expect(runsOn(person)).toBe(0);
	158
	159	// Assigned while already labelled `flow`: never held, and its one run is the assignment's.
	160	const flowing = open("Already flow");
	161	setIssueLabels(app.db, flowing.id, [getLabel(app.db, repo.id, "flow")!.id]);
	162	assign(flowing);
	163	sweepAssigned(app);
	164	expect(flowOnlyHolds(app.db, fresh(), getIssueById(app.db, flowing.id)!)).toBe(false);
	165	expect(runsOn(flowing)).toBe(1);
	166	expect(() => label(flowing)).not.toThrow();
	167	expect(runsOn(flowing)).toBe(1);
	168
	169	const closed = assign(open("Closed"));
	170	const shut = setIssueState(app, alice, fresh(), closed, "closed");
	171	if (!shut.ok) throw new Error(String(shut.error));
	172	expect(() => label(closed)).not.toThrow();
	173	expect(runsOn(closed)).toBe(0);
	174	});
	175
	176	test("loaded cold as the forge boots, the label on a held issue starts its run", async () => {
	177	const held = assign(open("Held at boot"));
	178	expect(runsOn(held)).toBe(0);
	179	const child = Bun.spawn([process.execPath, BOOT, dataDir, "alice", "svc", String(held.number)], {
	180	stdout: "pipe",
	181	stderr: "pipe",
	182	env: { PATH: process.env["PATH"] ?? "", HOME: process.env["HOME"] ?? "" },
	183	});
	184	const [out, err] = await Promise.all([new Response(child.stdout).text(), new Response(child.stderr).text()]);
	185	expect({ code: await child.exited, err: (await child.exited) === 0 ? "" : err }).toEqual({ code: 0, err: "" });
	186	expect(out.trim().split("\n").at(-1)).toBe("1");
	187	expect(runsOn(held)).toBe(1);
	188	}, 30_000);
	189	});
		@@ -0,0 +1,28 @@
	1	/**
	2	* The `flow` label in a fresh process, loaded as `src/main.ts` loads the
	3	* forge at boot (`test/flow-label-dispatch.test.ts`): the app and the HTTP
	4	* server's module graph, then a top-level await, then an owner labels a held
	5	* issue `flow`. `actions/issues.ts` loads the dispatcher only at that call, so
	6	* a dispatcher that cannot load cold fails here and nowhere earlier.
	7	*
	8	* Printed: the number of agent runs on the issue afterwards.
	9	*/
	10	import { createApp } from "../../src/app.ts";
	11	import "../../src/http/server.ts";
	12	import { setIssueLabels } from "../../src/actions/issues.ts";
	13	import { getIssue } from "../../src/core/issues.ts";
	14	import { getRepo } from "../../src/core/repos.ts";
	15	import { getUserByName } from "../../src/core/users.ts";
	16
	17	const [dataDir, owner, name, number] = process.argv.slice(2);
	18	const app = createApp({ dataDir: dataDir!, port: 0 });
	19	await Bun.sleep(0);
	20	const repo = getRepo(app.db, owner!, name!)!;
	21	const issue = getIssue(app.db, repo.id, Number(number))!;
	22	const labelled = setIssueLabels(app, getUserByName(app.db, owner!)!, repo, issue, ["flow"]);
	23	if (!labelled.ok) throw new Error(String(labelled.error));
	24	const runs = app.db
	25	.query<{ n: number }, [string]>("SELECT COUNT(*) AS n FROM run WHERE issue_id = ?1 AND kind = 'agent'")
	26	.get(issue.id)!.n;
	27	app.close();
	28	console.log(runs);