feat(api): the session endpoint, and a claim that only one caller can win

A run of a box had core support and no HTTP surface. This adds both halves of it: a person asks for a run and reads it back, and the host agent the box is placed on is handed the work and reports what happened. The access rule is the point. An agent may only see or touch a run whose box is placed on its own hardware, and that is a `where` clause on every one of the three agent endpoints rather than a check next to them — host credentials are long-lived secrets sitting on hardware in somebody's home, so what one leaking can reach has to be decided by the query. "No such run" and "not your run" are the same refusal, so ids cannot be discovered by reporting states at them. `Session.setState` updated on the id alone, which means two agents polling the same work both succeed and both start the same box. There is one host today, which is exactly why that would have been built wrong and stayed wrong. The state a run is moving out of is now part of the `where` clause, so the database picks the winner; the loser gets a conflict rather than a silent no-op. Three cases that look alike are kept apart: re-reporting a state you already reported changes nothing and is not an error, a transition that does not exist is refused with the run left where it was, and another host reporting anything is forbidden. Asking for a run makes no decision about where it happens — a box already names its hardware, so the run inherits it by join. Placement therefore gets an interface at box creation, where the decision actually is, with the single-host case as its implementation and a deliberate refusal when there is more than one candidate and no policy to choose with. Tests cover the wire shape from both sides, the query scoping, the claim, and the timestamp idempotence a run's billing rests on.
2026-09-22 18:55:22 +03:00 · 2026-09-04 18:57:08 +03:00
parent aaa1bbd0f4
commit bbe729e5c7
8 changed files with 1658 additions and 2 deletions
--- a/packages/core/src/session/index.ts
+++ b/packages/core/src/session/index.ts
@@ -1,9 +1,11 @@
-import { and, desc, eq, isNull, sql } from 'drizzle-orm';
+import { and, desc, eq, inArray, isNull, notInArray, sql } from 'drizzle-orm';
 import z from 'zod';

+import { BoxTable, BoxTier } from '../box/box.sql.js';
 import { Database } from '../db/index.js';
 import { Examples } from '../examples.js';
 import { fn } from '../fn.js';
+import { GameTable } from '../game/game.sql.js';
 import { SessionState, SessionTable } from './session.sql.js';

 /**
@@ -188,6 +190,328 @@ export namespace Session {
 		}
 	);

+	/**
+	 * The states an agent is allowed to move a run into.
+	 *
+	 * `requested` is missing on purpose: it is written once, when the row is
+	 * created, and nothing may put a run back there.
+	 */
+	export const ReportableState = z.enum(['starting', 'live', 'ended', 'failed']);
+
+	export type ReportableState = z.infer<typeof ReportableState>;
+
+	/**
+	 * Where a run may go next, and nowhere else.
+	 *
+	 * `requested → live` is missing although it is the tempting shortcut:
+	 * skipping `starting` means nothing ever holds the claim, and the claim is
+	 * the only mutual exclusion in this design. `ended` and `failed` are
+	 * terminal, so their entries are empty rather than absent — a state with no
+	 * exits is a fact worth writing down.
+	 */
+	export const NEXT_STATES: Record<Info['state'], readonly Info['state'][]> = {
+		requested: ['starting'],
+		starting: ['live', 'failed'],
+		live: ['ended', 'failed'],
+		ended: [],
+		failed: []
+	};
+
+	/**
+	 * A unit of work handed to the agent that will carry it out.
+	 *
+	 * There is no queue: a run in state `requested` *is* the work order, and
+	 * the agent that fulfils it moves that same row along. Two sources of truth
+	 * for one piece of work is how a queue and a database come to disagree
+	 * about whether something ran.
+	 *
+	 * `kind` is on the wire while there is only one value, so that a second
+	 * kind is an addition rather than a redesign of the poll.
+	 */
+	export const Job = z
+		.object({
+			kind: z.literal('session.start').meta({
+				description: 'What the agent is being asked to do',
+				example: 'session.start'
+			}),
+			sessionId: z.string().meta({
+				description: 'The run to report progress against',
+				example: Examples.Session.id
+			}),
+			boxId: z.string().meta({
+				description: 'The box to start',
+				example: Examples.Session.boxId
+			}),
+			boxTier: z.enum(BoxTier.enumValues).meta({
+				description: 'The size the box was asked for, which also sets output geometry',
+				example: Examples.Box.tier
+			}),
+			gameId: z.string().meta({
+				description: 'The game to launch',
+				example: Examples.Session.gameId
+			}),
+			steamAppId: z.number().int().meta({
+				description: 'The same game, in the id the store knows it by',
+				example: Examples.Game.steamAppId
+			}),
+			linkedAccountId: z.string().meta({
+				description: 'Which linked account is playing',
+				example: Examples.Session.linkedAccountId
+			})
+		})
+		.meta({
+			ref: 'SessionJob',
+			description: 'One run waiting to be started, as handed to the agent that will start it'
+		});
+
+	export type Job = z.infer<typeof Job>;
+
+	/**
+	 * The work waiting for one host.
+	 *
+	 * The scope is the join and not a filter the caller asks for: a box names
+	 * the hardware it is placed on, a run reaches its hardware through its box,
+	 * and so what one set of long-lived credentials can see is decided by this
+	 * `where` clause rather than by whoever is holding them.
+	 */
+	export const listJobsForMachine = fn(z.string(), async (machineId) => {
+		return Database.use(async (tx) => {
+			return tx
+				.select({ session: SessionTable, box: BoxTable, game: GameTable })
+				.from(SessionTable)
+				.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
+				.innerJoin(GameTable, eq(SessionTable.gameId, GameTable.id))
+				.where(
+					and(
+						eq(BoxTable.machineId, machineId),
+						eq(SessionTable.state, 'requested'),
+						isNull(SessionTable.timeDeleted),
+						isNull(BoxTable.timeDeleted)
+					)
+				)
+				.orderBy(SessionTable.timeCreated)
+				.then((rows) =>
+					rows.map(
+						(row): Job => ({
+							kind: 'session.start',
+							sessionId: row.session.id,
+							boxId: row.box.id,
+							boxTier: row.box.tier as Job['boxTier'],
+							gameId: row.game.id,
+							steamAppId: row.game.steamAppId,
+							linkedAccountId: row.session.linkedAccountId
+						})
+					)
+				);
+		});
+	});
+
+	/** One run, visible only to the host its box is placed on. */
+	export const forMachine = fn(
+		z.object({ id: Info.shape.id, machineId: z.string() }),
+		async (input) => {
+			return Database.use(async (tx) => {
+				return tx
+					.select({ session: SessionTable })
+					.from(SessionTable)
+					.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
+					.where(
+						and(
+							eq(SessionTable.id, input.id),
+							eq(BoxTable.machineId, input.machineId),
+							isNull(SessionTable.timeDeleted),
+							isNull(BoxTable.timeDeleted)
+						)
+					)
+					.then((rows) => {
+						const row = rows.at(0);
+						return row ? serialize(row.session) : null;
+					});
+			});
+		}
+	);
+
+	/** One run, visible only to the person who owns its box. */
+	export const forOwner = fn(z.object({ id: Info.shape.id, userId: z.string() }), async (input) => {
+		return Database.use(async (tx) => {
+			return tx
+				.select({ session: SessionTable })
+				.from(SessionTable)
+				.innerJoin(BoxTable, eq(SessionTable.boxId, BoxTable.id))
+				.where(
+					and(
+						eq(SessionTable.id, input.id),
+						eq(BoxTable.userId, input.userId),
+						isNull(SessionTable.timeDeleted),
+						isNull(BoxTable.timeDeleted)
+					)
+				)
+				.then((rows) => {
+					const row = rows.at(0);
+					return row ? serialize(row.session) : null;
+				});
+		});
+	});
+
+	/** The boxes one host is responsible for, as a subquery to scope a write. */
+	function boxesOn(tx: Parameters<Parameters<typeof Database.use>[0]>[0], machineId: string) {
+		return tx
+			.select({ id: BoxTable.id })
+			.from(BoxTable)
+			.where(and(eq(BoxTable.machineId, machineId), isNull(BoxTable.timeDeleted)));
+	}
+
+	/**
+	 * Move a run from one exact state to another, or do nothing at all.
+	 *
+	 * This is the claim, and it is why `setState` is not enough on its own:
+	 * updating on the id alone means two agents polling the same work both
+	 * succeed and both start the same box. The current state is part of the
+	 * `where` clause, so the database decides the winner and the loser gets
+	 * null rather than a row. There is one host today, which is exactly why
+	 * this would otherwise be built wrong and stay wrong.
+	 *
+	 * The host is in the same `where` clause. The caller checking first is not
+	 * the same thing as the write being scoped, and only one of the two is
+	 * still true when somebody adds a second caller.
+	 */
+	export const compareAndSetState = fn(
+		z.object({
+			id: Info.shape.id,
+			machineId: z.string(),
+			from: z.enum(SessionState.enumValues),
+			to: z.enum(SessionState.enumValues),
+			errorMessage: Info.shape.errorMessage
+		}),
+		async (input) => {
+			const now = sql`now()`;
+			return Database.use(async (tx) => {
+				return tx
+					.update(SessionTable)
+					.set({
+						state: input.to,
+						errorMessage: input.to === 'failed' ? (input.errorMessage ?? null) : null,
+						...(input.to === 'live'
+							? { timeStarted: sql`coalesce(${SessionTable.timeStarted}, ${now})` }
+							: {}),
+						...(input.to === 'ended' || input.to === 'failed'
+							? { timeStopped: sql`coalesce(${SessionTable.timeStopped}, ${now})` }
+							: {})
+					})
+					.where(
+						and(
+							eq(SessionTable.id, input.id),
+							eq(SessionTable.state, input.from),
+							isNull(SessionTable.timeDeleted),
+							inArray(SessionTable.boxId, boxesOn(tx, input.machineId))
+						)
+					)
+					.returning()
+					.then((rows) => {
+						const row = rows.at(0);
+						return row ? serialize(row) : null;
+					});
+			});
+		}
+	);
+
+	/**
+	 * What happened when an agent reported a state.
+	 *
+	 * Four outcomes that look alike from a distance and are not, which is the
+	 * whole reason this is not a boolean:
+	 *
+	 * - `forbidden` — no such run, or it is not on this host. One answer for
+	 *   both, so reporting states at ids cannot be used to discover them.
+	 * - `unchanged` — already in that state. A retry after a lost response is
+	 *   not a broken agent and must not be told it is.
+	 * - `illegal` — not a transition that exists. The row does not move.
+	 * - `lost` — a legal transition that something else got to first.
+	 * - `moved` — it happened.
+	 */
+	export type TransitionOutcome = 'forbidden' | 'unchanged' | 'illegal' | 'lost' | 'moved';
+
+	export interface TransitionResult {
+		outcome: TransitionOutcome;
+		session: Info | null;
+	}
+
+	export const transition = fn(
+		z.object({
+			id: Info.shape.id,
+			machineId: z.string(),
+			state: z.enum(SessionState.enumValues),
+			errorMessage: Info.shape.errorMessage
+		}),
+		async (input): Promise<TransitionResult> => {
+			const current = await forMachine({ id: input.id, machineId: input.machineId });
+			if (!current) return { outcome: 'forbidden', session: null };
+			if (current.state === input.state) return { outcome: 'unchanged', session: current };
+			if (!NEXT_STATES[current.state].includes(input.state)) {
+				return { outcome: 'illegal', session: current };
+			}
+
+			const moved = await compareAndSetState({
+				id: input.id,
+				machineId: input.machineId,
+				from: current.state,
+				to: input.state,
+				errorMessage: input.errorMessage
+			});
+			// The state read above is not the state written below, and the gap
+			// is where two agents race. Nothing moved means somebody else did.
+			if (!moved) return { outcome: 'lost', session: current };
+			return { outcome: 'moved', session: moved };
+		}
+	);
+
+	export interface TicketResult {
+		outcome: 'forbidden' | 'closed' | 'published';
+		session: Info | null;
+	}
+
+	/**
+	 * Publish a ticket for a run, on behalf of the host it is placed on.
+	 *
+	 * A ticket may appear while the state is still `starting` — it is
+	 * republished as addresses are discovered, so the client polls and re-reads
+	 * rather than keeping the first one. A run that has stopped is refused: an
+	 * address for something that is not there can only mislead whoever is
+	 * still polling.
+	 */
+	export const publishTicket = fn(
+		z.object({
+			id: Info.shape.id,
+			machineId: z.string(),
+			ticket: z.string().min(1)
+		}),
+		async (input): Promise<TicketResult> => {
+			const current = await forMachine({ id: input.id, machineId: input.machineId });
+			if (!current) return { outcome: 'forbidden', session: null };
+
+			return Database.use(async (tx) => {
+				return tx
+					.update(SessionTable)
+					.set({ ticket: input.ticket })
+					.where(
+						and(
+							eq(SessionTable.id, input.id),
+							notInArray(SessionTable.state, ['ended', 'failed']),
+							isNull(SessionTable.timeDeleted),
+							inArray(SessionTable.boxId, boxesOn(tx, input.machineId))
+						)
+					)
+					.returning()
+					.then((rows): TicketResult => {
+						const row = rows.at(0);
+						return row
+							? { outcome: 'published', session: serialize(row) }
+							: { outcome: 'closed', session: current };
+					});
+			});
+		}
+	);
+
 	export function serialize(input: typeof SessionTable.$inferSelect): z.infer<typeof Info> {
 		return {
 			id: input.id,
--- a/packages/core/src/session/session.test.ts
+++ b/packages/core/src/session/session.test.ts
@@ -41,7 +41,7 @@ async function scene(label: string, steamAppId: number) {
 		label,
 		tier: 'sm'
 	});
-	return { owner, box, gameId: await newGame(steamAppId) };
+	return { owner, machineId, box, gameId: await newGame(steamAppId) };
 }

 afterAll(async () => {
@@ -173,3 +173,244 @@ describe('Session', () => {
 		expect(await Session.listByBox(box.id)).toHaveLength(0);
 	});
 });
+
+describe('Session jobs', () => {
+	test('a requested session is the job, and it carries its kind', async () => {
+		const { owner, machineId, box, gameId } = await scene('ses-job-kind', 5410);
+		const session = await Session.create({
+			id: Identifier.ascending('session'),
+			boxId: box.id,
+			gameId,
+			linkedAccountId: owner.linkedAccountId
+		});
+
+		const jobs = await Session.listJobsForMachine(machineId);
+		expect(jobs).toHaveLength(1);
+		// The kind is on the wire from the first day there is only one, so the
+		// second kind is an addition rather than a redesign.
+		expect(jobs[0]!.kind).toBe('session.start');
+		expect(jobs[0]!.sessionId).toBe(session.id);
+		expect(jobs[0]!.boxId).toBe(box.id);
+		expect(jobs[0]!.boxTier).toBe('sm');
+		expect(jobs[0]!.gameId).toBe(gameId);
+		expect(jobs[0]!.steamAppId).toBe(5410);
+		expect(jobs[0]!.linkedAccountId).toBe(owner.linkedAccountId);
+	});
+
+	test('a job belongs to the machine its box is placed on and to no other', async () => {
+		const mine = await scene('ses-job-mine', 5411);
+		const theirs = await scene('ses-job-theirs', 5412);
+
+		const session = await Session.create({
+			id: Identifier.ascending('session'),
+			boxId: theirs.box.id,
+			gameId: theirs.gameId,
+			linkedAccountId: theirs.owner.linkedAccountId
+		});
+
+		// The scope is the join, not a filter the caller asks for. A machine
+		// credential is a long-lived secret on hardware in somebody's home, so
+		// what one leaking can reach is decided here.
+		expect(await Session.listJobsForMachine(mine.machineId)).toHaveLength(0);
+		expect((await Session.listJobsForMachine(theirs.machineId)).map((j) => j.sessionId)).toEqual([
+			session.id
+		]);
+	});
+
+	test('only a requested session is work; a claimed one is not offered again', async () => {
+		const { owner, machineId, box, gameId } = await scene('ses-job-claimed', 5413);
+		const session = await Session.create({
+			id: Identifier.ascending('session'),
+			boxId: box.id,
+			gameId,
+			linkedAccountId: owner.linkedAccountId
+		});
+
+		expect(await Session.listJobsForMachine(machineId)).toHaveLength(1);
+		await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'starting',
+			errorMessage: null
+		});
+		expect(await Session.listJobsForMachine(machineId)).toHaveLength(0);
+	});
+});
+
+describe('Session claim', () => {
+	async function requested(label: string, steamAppId: number) {
+		const s = await scene(label, steamAppId);
+		const session = await Session.create({
+			id: Identifier.ascending('session'),
+			boxId: s.box.id,
+			gameId: s.gameId,
+			linkedAccountId: s.owner.linkedAccountId
+		});
+		return { ...s, session };
+	}
+
+	test('the claim is a compare-and-set, so the second attempt finds nothing to move', async () => {
+		const { machineId, session } = await requested('ses-cas', 5420);
+
+		const won = await Session.compareAndSetState({
+			id: session.id,
+			machineId,
+			from: 'requested',
+			to: 'starting',
+			errorMessage: null
+		});
+		expect(won?.state).toBe('starting');
+
+		// The same attempt again. The row is no longer `requested`, so the
+		// update matches nothing — which is what stops two agents from both
+		// starting the same box. Updating on the id alone would succeed twice.
+		const lost = await Session.compareAndSetState({
+			id: session.id,
+			machineId,
+			from: 'requested',
+			to: 'starting',
+			errorMessage: null
+		});
+		expect(lost).toBeNull();
+		expect((await Session.fromID(session.id))?.state).toBe('starting');
+	});
+
+	test('a machine that is not the box’s host cannot move the row', async () => {
+		const { session } = await requested('ses-cas-mine', 5421);
+		const other = await scene('ses-cas-other', 5422);
+
+		const result = await Session.transition({
+			id: session.id,
+			machineId: other.machineId,
+			state: 'starting',
+			errorMessage: null
+		});
+		expect(result.outcome).toBe('forbidden');
+		expect((await Session.fromID(session.id))?.state).toBe('requested');
+
+		// And the compare-and-set is scoped in the same query, not only by the
+		// classification above it.
+		expect(
+			await Session.compareAndSetState({
+				id: session.id,
+				machineId: other.machineId,
+				from: 'requested',
+				to: 'starting',
+				errorMessage: null
+			})
+		).toBeNull();
+	});
+
+	test('a session that does not exist is refused the same way as one that is not yours', async () => {
+		const other = await scene('ses-cas-ghost', 5423);
+		const result = await Session.transition({
+			// A well-formed id for a row that was never written.
+			id: Identifier.ascending('session'),
+			machineId: other.machineId,
+			state: 'starting',
+			errorMessage: null
+		});
+		// Same answer as somebody else's session: an agent must not be able to
+		// learn which ids exist by reporting states at them.
+		expect(result.outcome).toBe('forbidden');
+	});
+
+	test('re-reporting the state you already reported changes nothing', async () => {
+		const { machineId, session } = await requested('ses-repeat', 5424);
+
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		const again = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'starting',
+			errorMessage: null
+		});
+		// A retry after a lost response is not a broken agent.
+		expect(again.outcome).toBe('unchanged');
+		expect(again.session?.state).toBe('starting');
+	});
+
+	test('a transition off the table is refused and the row does not move', async () => {
+		const { machineId, session } = await requested('ses-illegal', 5425);
+
+		// Skipping `starting` means nothing ever holds the claim, and the claim
+		// is the only mutual exclusion here — so it is refused however tempting
+		// the shortcut looks.
+		const skipped = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'live',
+			errorMessage: null
+		});
+		expect(skipped.outcome).toBe('illegal');
+		expect((await Session.fromID(session.id))?.state).toBe('requested');
+
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		await Session.transition({ id: session.id, machineId, state: 'failed', errorMessage: 'no' });
+
+		// Terminal is terminal: a dead session cannot be resurrected.
+		const raised = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'live',
+			errorMessage: null
+		});
+		expect(raised.outcome).toBe('illegal');
+		expect((await Session.fromID(session.id))?.state).toBe('failed');
+	});
+
+	test('the timestamps survive a duplicate report, which is what billing rests on', async () => {
+		const { machineId, session } = await requested('ses-idempotent', 5426);
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		const live = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'live',
+			errorMessage: null
+		});
+		expect(live.session?.timeStarted).not.toBeNull();
+
+		const repeat = await Session.transition({
+			id: session.id,
+			machineId,
+			state: 'live',
+			errorMessage: null
+		});
+		expect(repeat.session?.timeStarted).toBe(live.session!.timeStarted);
+	});
+
+	test('publishing a ticket is scoped to the host too', async () => {
+		const { machineId, session } = await requested('ses-ticket-scope', 5427);
+		const other = await scene('ses-ticket-other', 5428);
+
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+
+		const refused = await Session.publishTicket({
+			id: session.id,
+			machineId: other.machineId,
+			ticket: 'stolen'
+		});
+		expect(refused.outcome).toBe('forbidden');
+		expect((await Session.fromID(session.id))?.ticket).toBeNull();
+
+		// A ticket may appear while the state is still `starting`.
+		const first = await Session.publishTicket({ id: session.id, machineId, ticket: 'one' });
+		expect(first.outcome).toBe('published');
+		expect(first.session?.ticket).toBe('one');
+		expect(first.session?.state).toBe('starting');
+
+		const second = await Session.publishTicket({ id: session.id, machineId, ticket: 'two' });
+		expect(second.session?.ticket).toBe('two');
+	});
+
+	test('a stopped session has no address to publish', async () => {
+		const { machineId, session } = await requested('ses-ticket-dead', 5429);
+		await Session.transition({ id: session.id, machineId, state: 'starting', errorMessage: null });
+		await Session.transition({ id: session.id, machineId, state: 'live', errorMessage: null });
+		await Session.transition({ id: session.id, machineId, state: 'ended', errorMessage: null });
+
+		const result = await Session.publishTicket({ id: session.id, machineId, ticket: 'late' });
+		expect(result.outcome).toBe('closed');
+		expect((await Session.fromID(session.id))?.ticket).toBeNull();
+	});
+});