Give the room a second call to a seat: cut names an activation whose lease it ended

docs/agent.md

@@ -676,14 +676,30 @@ system prompt and the context, and sends the two strings with the model
 id and the hand the activation holds. The seat side resolves the definition
 by name through the runtime's catalog, builds the Pi `Agent`, and reaches
 the room through three calls: `view`, `commit` and `lease`. The room
-reaches a seat through one, `wake`, which carries the line a running
-activation is steered with when a message caused it. Every request and
-response survives a round trip through `JSON.stringify` unchanged
+reaches a seat through two. `wake` carries the line a running activation
+is steered with when a message caused it. `cut` names an activation whose
+lease the room ended, so the seat side stops it now, wherever the seat
+runs. Every request and response survives a round trip through
+`JSON.stringify` unchanged
 ([`wire.ts`](../packages/ambion/src/wire.ts)), so a seat and a room can
 live in two processes. The room answers the three calls from the fold: a
 lease is a row on the log, and the seat side releases it when the
 activation ends.
 
+**A cut is the room's word, and the record is written before it.** The
+room ends every lease the seat holds as `revoked`, and then it cuts. A
+seat that never hears the cut is refused whatever it writes, because its
+lease ended. A seat that hears it aborts the activation and moves on to
+the wake that queued behind it. A model call that ignores the abort
+finishes on its own, past a seat that has moved on. A renewal the room
+refuses cuts the activation the same way, and a renewal that never
+reached the room leaves the lease to expire where it stands: the seat
+cuts the activation at that expiry, when the room expires the lease.
+
+**A cut reaches a seat the run never woke.** A run that resumes over a
+live lease holds no wire to the seat that took it. The room opens one to
+say the cut, so a seat that still runs the activation stops.
+
 **A host owns a `Runtime`.** It holds the clock, the session opener, the
 model call, the catalog, the rooms that are running, the workspace names
 that are taken, and the policy for wakes and retries: how long a lease

packages/ambion/src/seat/seat.ts

@@ -102,6 +102,9 @@ interface Current {
 	activation: Activation;
 	/** The activation ran to its end, and its release is in flight. It takes no steer. */
 	over: boolean;
+	/** Resolves when the room ended the lease: the actor moves on, whatever the run still does. */
+	cut: () => void;
+	cutOff: Promise<void>;
 }
 
 /**
@@ -156,22 +159,44 @@ export class SeatActor implements SeatPort {
 		if (!this.queued.includes(id)) this.queued.push(id);
 	}
 
+	/**
+	 * The room ended this activation's lease. The activation is aborted, and
+	 * the actor moves on at once: a run that ignores the abort is left to
+	 * finish on its own, and every call it still makes is answered stale.
+	 */
+	async cut(activation: string): Promise<void> {
+		if (this.current?.id === activation) this.cutCurrent();
+	}
+
 	/** Cut the activation in flight, whatever its id. The room hears how it ended. */
 	abort(): void {
-		this.current?.activation.abort();
+		this.cutCurrent();
+	}
+
+	private cutCurrent(): void {
+		const current = this.current;
+		if (current === undefined) return;
+		current.activation.abort();
+		current.cut();
 	}
 
 	private async take(id: string): Promise<void> {
 		// Held before the claim, so a steer that lands while the claim is in
 		// flight reaches the activation and not the floor.
 		const activation = new Activation(id, this.context.seat, this.host(id));
-		const current: Current = { id, activation, over: false };
+		let cut = () => {};
+		const cutOff = new Promise<void>((resolve) => {
+			cut = resolve;
+		});
+		const current: Current = { id, activation, over: false, cut, cutOff };
 		this.current = current;
 		const claimed = await this.claim(id);
 		if (claimed !== undefined) {
-			const stopRenewing = this.renewUntil(activation, claimed.expiry);
+			const stopRenewing = this.renewUntil(current, claimed.expiry);
 			try {
-				await activation.run();
+				// The cut ends the wait, and never the run: a run that ignores the
+				// abort finishes on its own, past a seat that took its next wake.
+				await Promise.race([activation.run(), cutOff]);
 			} finally {
 				stopRenewing();
 				// Over, and holding the seat through the release: a wake that lands
@@ -225,31 +250,44 @@ export class SeatActor implements SeatPort {
 		}
 	}
 
-	/** One renewal: the new expiry, or nothing when the room refused it or it never reached the room. */
-	private async renew(activation: Activation): Promise<number | undefined> {
+	/**
+	 * One renewal: the new expiry, `stale` when the room refused it, or
+	 * `lost` when it never reached the room.
+	 */
+	private async renew(activation: Activation): Promise<number | 'stale' | 'lost'> {
 		try {
 			const renewed = await this.room.lease({ activation: activation.id, phase: 'running' });
-			return 'stale' in renewed ? undefined : renewed.ok.expiry;
+			return 'stale' in renewed ? 'stale' : renewed.ok.expiry;
 		} catch {
-			return undefined;
+			return 'lost';
 		}
 	}
 
 	/**
 	 * Renew at half the expiry, for as long as the activation runs and the
-	 * room renews it. The cancel stops the loop for good: a renewal in flight
-	 * when the activation ends arms nothing when it comes back.
+	 * room renews it. A refused renewal cuts the activation now: its lease
+	 * ended, so nothing it writes lands. A renewal that never reached the
+	 * room leaves the lease to expire where it stands, and the actor cuts
+	 * the activation at that expiry, when the room expires the lease. The
+	 * cancel stops the loop for good: a renewal in flight when the
+	 * activation ends arms nothing when it comes back.
 	 */
-	private renewUntil(activation: Activation, firstExpiry: number): () => void {
+	private renewUntil(current: Current, firstExpiry: number): () => void {
 		const clock = this.context.clock;
 		let stopped = false;
 		let cancel = () => {};
+		const cut = () => {
+			if (this.current === current) this.cutCurrent();
+		};
 		const schedule = (expiry: number) => {
-			cancel = clock.alarm(clock.now() + (expiry - clock.now()) / 2, () => void again());
+			cancel = clock.alarm(clock.now() + (expiry - clock.now()) / 2, () => void again(expiry));
 		};
-		const again = async () => {
-			const renewed = await this.renew(activation);
-			if (!stopped && renewed !== undefined) schedule(renewed);
+		const again = async (held: number) => {
+			const renewed = await this.renew(current.activation);
+			if (stopped) return;
+			if (renewed === 'stale') cut();
+			else if (renewed === 'lost') cancel = clock.alarm(held, cut);
+			else schedule(renewed);
 		};
 		schedule(firstExpiry);
 		return () => {

packages/ambion/src/session.ts

@@ -38,7 +38,7 @@ import { activationId, draftId, isExpired, isLive, parseId, seatOf } from './roo
 import type { VisitRuntime } from './room/presence.ts';
 import { type Decision, decide, liveSeats, working } from './room/reconcile.ts';
 import { type RoomFacts, seatsOf, viewOf } from './room/view.ts';
-import { inProcessTransport, SeatActor, wakes } from './seat/seat.ts';
+import { inProcessTransport, wakes } from './seat/seat.ts';
 import {
 	type AgentDefinition,
 	type AgentSeat,
@@ -1280,13 +1280,14 @@ class SessionImpl implements Session, RunningRoom {
 
 	/**
 	 * Cut one seat: every lease it holds ends revoked, every wake pending for
-	 * it and every draft due for it is written off the same way, and the
-	 * activation in flight is aborted where the seat runs in this process.
+	 * it and every draft due for it is written off the same way, and the seat
+	 * side is told to stop, wherever the seat runs. The room writes first, so
+	 * a seat that never hears the cut is refused whatever it writes after it.
 	 */
 	private async cut(seat: string, ids: string[]): Promise<void> {
 		for (const id of ids) await this.end(id, seat, 'revoked');
-		const port = this.ports.get(seat);
-		if (port instanceof SeatActor) port.abort();
+		const port = this.port(seat);
+		for (const id of ids) void port.cut(id).catch(() => {});
 	}
 
 	/** Closes the run: what is live is revoked, what is present is marked gone, and the name comes free. */

packages/ambion/src/wire.ts

@@ -9,8 +9,10 @@
  * The seat reaches the room through three calls: `view` reads what an
  * activation is given, `commit` puts one message on the record, and
  * `lease` claims, renews or releases the activation. The room reaches a
- * seat through one: `wake` names an activation the seat runs, and carries
- * the line a running activation is steered with when a message caused it.
+ * seat through two: `wake` names an activation the seat runs, and carries
+ * the line a running activation is steered with when a message caused it;
+ * `cut` names an activation whose lease the room ended, so the seat side
+ * stops it now.
  */
 import type { Attention, Message, Seq } from './types.ts';
 
@@ -93,6 +95,8 @@ export interface Wake {
 
 export interface SeatPort {
 	wake(wake: Wake): Promise<void>;
+	/** The room ended this activation's lease: stop it, and run what queued behind it. */
+	cut(activation: string): Promise<void>;
 }
 
 // -- a seat reaching its room -------------------------------------------------

packages/ambion/test/consistency.test.ts

@@ -43,7 +43,7 @@ function mulberry32(seed: number): () => number {
 	};
 }
 
-const OPERATIONS: Operation[] = ['wake', 'view', 'commit', 'lease'];
+const OPERATIONS: Operation[] = ['wake', 'cut', 'view', 'commit', 'lease'];
 const RETRY = { attempts: 3, backoff: (attempt: number) => attempt * 30_000 };
 
 /** One room over one storage, and the run that holds it now. */
@@ -60,7 +60,7 @@ class Cluster {
 	session!: Session;
 	private disk: FailMode = false;
 	private failedBefore = 0;
-	/** Room calls the nemesis dropped, counted when taken: every one fails an activation, and that is one error. A dropped wake is sent again and fails nothing. */
+	/** Room calls the nemesis dropped, counted when taken: every one fails an activation, and that is one error. A dropped wake is sent again, and a dropped cut leaves the record's word to stand, so both fail nothing. */
 	private dropped = 0;
 	private droppedBefore = 0;
 	/** Leases live when time jumped past the whole expiry: every one expires, and that is one error. */
@@ -255,7 +255,7 @@ class Cluster {
 		const kind = this.pick(['drop', 'duplicate', 'delay'] as const);
 		const on = this.pick(OPERATIONS);
 		const taken = () => {
-			if (kind === 'drop' && on !== 'wake') this.dropped += 1;
+			if (kind === 'drop' && on !== 'wake' && on !== 'cut') this.dropped += 1;
 			return true;
 		};
 		this.faults.push({ on, kind, match: taken, ...(kind === 'delay' ? { ms: 2_000 } : {}) });

packages/ambion/test/property.test.ts

@@ -108,7 +108,7 @@ const STEPS = [
 	'crash',
 ] as const;
 type Step = (typeof STEPS)[number];
-const OPERATIONS: Operation[] = ['wake', 'view', 'commit', 'lease'];
+const OPERATIONS: Operation[] = ['wake', 'cut', 'view', 'commit', 'lease'];
 
 /**
  * What a step may hear back: the storage refused the write, the visit is

packages/ambion/test/restart.test.ts

@@ -484,6 +484,64 @@ describe.each(storages)('a room resumed on $name', (storage) => {
 		}
 	});
 
+	it('cuts a lease the last run took, over a wire this run has not opened yet', async () => {
+		const { opened, clock } = await world(storage);
+		try {
+			// the activation never answers, so the run the crash leaves behind
+			// writes nothing after the test ends
+			const script = byAgent({
+				alpha: (_c, _n, call) =>
+					call === 1 ? new Promise<never>(() => {}) : Promise.resolve(quiet()),
+			});
+			const name = roomName(`restart-${storage.name}`);
+			const first = createRuntime({ sessions: opened.sessions, clock, agents });
+			const session = startSession({
+				name,
+				assistant,
+				agents: [alpha],
+				runtime: first,
+				streamFn: scripted(script),
+			});
+			const visit = await visitSession(session, priya);
+			await visit.deliver({ text: 'Anyone?' });
+			await tick();
+			expect(session.seats().find((s) => s.name === 'alpha')).toMatchObject({ status: 'active' });
+			crash(first, session);
+
+			// the resumed run inherits the live lease and never wakes alpha, so it
+			// holds no port for that seat when the abort revokes what it inherited
+			const cuts: string[] = [];
+			const inProcess = inProcessTransport();
+			const second = createRuntime({
+				sessions: opened.sessions,
+				clock,
+				agents,
+				transport: {
+					connect: (room, seat, host) => {
+						const port = inProcess.connect(room, seat, host);
+						return {
+							wake: (wake) => port.wake(wake),
+							cut: (activation) => {
+								cuts.push(activation);
+								return port.cut(activation);
+							},
+						};
+					},
+				},
+			});
+			const resumed = await resumeSession(name, { runtime: second, streamFn: scripted(script) });
+			expect(resumed.seats().find((s) => s.name === 'alpha')).toMatchObject({ status: 'active' });
+			resumed.abort();
+			await resumed.settled();
+			await tick();
+			// the seat side hears the cut over the wire, and the room opened it to say so
+			expect(cuts).toEqual(['2:alpha']);
+			await stopSession(resumed);
+		} finally {
+			await opened.dispose();
+		}
+	});
+
 	it('refuses to resume a name whose seats the catalog does not hold, and one with no composition', async () => {
 		const { opened, runtime } = await world(storage);
 		try {