Skip to content

Move quiescence out of the room: Exchanges owns the whole span #44

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andreisavu wants to merge 1 commit into
base: main
Choose a base branch
from
+321 −70
Open

Move quiescence out of the room: Exchanges owns the whole span #44

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions docs/exchange.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,29 @@ mid-question: the record keeps what was said, and nobody is mid-question
after a restart. A person whose question the room was working on asks
again, and that question opens a new exchange.

**`Exchanges` owns the whole span.** It holds the open exchange, whether an
activation that counts as work began since the seats last settled, and the
two promises of §6. The room holds none of that. What is running is a fact
about the seats, so the room reads it off them and passes two booleans in
at every transition: whether a seat that speaks for itself is working, and
whether nothing at all is. Every transition is synchronous and takes no
model, so [`exchange.test.ts`](../packages/ambion/test/exchange.test.ts)
proves the lifecycle with no room around it.

The transitions, in the order the room calls them:

| Call | When | What it decides |
| --------------------------- | ----------------------------------------- | -------------------------------------------------------------------------- |
| `note(message, fromPerson)` | A message landed | Whether it opened an exchange |
| `stir()` | An activation that counts as work began | The next settle is the seats stopping |
| `settle(working, through)` | An activation ended, or a question routed | Nothing while working; else who waited, what closed, whether a seat worked |
| `quiesce(idle)` | After every settle | Whether the room went quiet, once |
| `drain()` | The run stops | Nobody waits on a room that never goes quiet |

The assistant drafting stirs nothing, so a settle at a draft's end reports
that no seat worked. That is the one fact a failed draft waits on: the next
settle where a seat did ([`assistant.md`](assistant.md) §5).

A closed exchange is an owner and a range, so it is derivable from the
record. Nothing derives it today; a host that wants a history of exchanges
records the `exchange_closed` events as they arrive.
Expand Down Expand Up @@ -250,6 +273,12 @@ The exchange is proved beside the assistant that first reads one, in

All in-process, in vitest, on a scripted stream.

The lifecycle alone is proved in
[`exchange.test.ts`](../packages/ambion/test/exchange.test.ts), with no
room and no stream: what opens one, what a settle closes and reports, when
each promise resolves, and what the assistant's scheduler makes of a settle
where a seat worked and one where none did (§5).

The live run is
[`demos/2026-08-31-one-exchange-one-message.html`](../demos/2026-08-31-one-exchange-one-message.html):
four questions opened four exchanges, and each one closed into one message.
84 changes: 77 additions & 7 deletions packages/ambion/src/exchange.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,12 @@
* - **What lands while it is open steers it and changes nothing.** Not the
* owner, not the range, not who the answer belongs to.
*
* This module also holds the two ends the room reports, `settled` and
* `quiet`, because they are the two edges of this span. What is running is
* a fact about the seats, so the room reads it off them and passes it in.
* Every transition here is synchronous and takes no model, so the whole
* lifecycle is provable with no room around it.
*
* The design contract is `docs/exchange.md`; `docs/assistant.md` says what an
* assistant makes of one.
*/
Expand All @@ -43,13 +49,38 @@ export interface ClosedExchange extends Exchange {
readonly through: Seq;
}

/** The seats settled: what that closed, and whether a seat worked since they last settled. */
export interface Settled {
/** The exchange the room was working on, or nothing when it worked on its own account. */
readonly closed: ClosedExchange | undefined;
/**
* Whether an activation that counts as work began since the last settle.
* A second settle at one quiescence — a question that woke nobody, an
* aborted activation ending after the exchange closed — is not the seats
* stopping again, and the assistant reads the difference.
*/
readonly worked: boolean;
}

/**
* The open exchange, if there is one. Run state: an exchange belongs to a
* running room, and a restart begins with none — the record keeps what was
* said, and nobody is mid-question after a restart.
* The room's exchanges: the open one, and the two ends the room reports.
*
* Run state: an exchange belongs to a running room, and a restart begins with
* none — the record keeps what was said, and nobody is mid-question after a
* restart.
*
* Two facts arrive from the room at every transition, and this holds neither:
* whether a seat that speaks for itself is taking an activation (*working*),
* and whether nothing at all is (*idle*). The room draws that one distinction
* about its assistant, and the seats hold the activations, so there is no
* count here to keep in step.
*/
export class Exchanges {
private open: Exchange | undefined;
/** Whether an activation that counts as work began since the seats last settled. */
private stirred = false;
private readonly settledWaiters: (() => void)[] = [];
private readonly quietWaiters: (() => void)[] = [];

/** What the room is working on, or nothing when nobody has asked. */
current(): Exchange | undefined {
Expand All @@ -72,13 +103,52 @@ export class Exchanges {
return this.open;
}

/** An activation that counts as work began: the next settle is the seats stopping. */
stir(): void {
this.stirred = true;
}

/**
* The room went quiet. Closes whatever was open and returns it with the
* range it held, or nothing when the room was working on its own account.
* Something stopped. While a seat is working nothing settles, and the
* result says so. Otherwise the seats have settled: whoever waited hears
* it first, then the open exchange closes with the range it reached, and
* the result says whether a seat worked since the last settle.
*/
close(through: Seq): ClosedExchange | undefined {
settle(working: boolean, through: Seq): Settled | undefined {
if (working) return undefined;
for (const resolve of this.settledWaiters.splice(0)) resolve();
const worked = this.stirred;
this.stirred = false;
const open = this.open;
this.open = undefined;
return open === undefined ? undefined : { ...open, through };
return { closed: open === undefined ? undefined : { ...open, through }, worked };
}

/**
* Nothing at all is taking an activation, so the room is quiet: whoever
* waited hears it, and the result says the room should say so. An active
* room is not quiet, and the result says nothing.
*/
quiesce(idle: boolean): boolean {
if (!idle) return false;
for (const resolve of this.quietWaiters.splice(0)) resolve();
return true;
}

/** Resolves when no seat that speaks for itself is taking an activation. */
settled(working: boolean): Promise<void> {
if (!working) return Promise.resolve();
return new Promise((resolve) => this.settledWaiters.push(resolve));
}

/** Resolves when nothing at all is taking an activation. */
quiet(idle: boolean): Promise<void> {
if (idle) return Promise.resolve();
return new Promise((resolve) => this.quietWaiters.push(resolve));
}

/** A stopped room never goes quiet on its own, so nobody waits on it. */
drain(): void {
for (const resolve of this.quietWaiters.splice(0)) resolve();
}
}
87 changes: 33 additions & 54 deletions packages/ambion/src/session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@
* - **Route.** Who hears a message, and who wakes for it.
* - **Give an activation what only the room knows.** The model, the prompt, the
* hands, and the room as it stands at that moment.
* - **Say when it has stopped.** An exchange closed, and nothing running.
*
* When the room has stopped is the exchange's own fact, and `exchange.ts`
* holds it: the room reads what is running off the seats and hands it over.
*/
import type {
AgentTool,
Expand All @@ -37,7 +39,7 @@ import {
summariseTool,
} from './assistant.ts';
import { seated } from './define.ts';
import { type ClosedExchange, type Exchange, Exchanges } from './exchange.ts';
import { type Exchange, Exchanges, type Settled } from './exchange.ts';
import { Attendance, type VisitRuntime } from './presence.ts';
import { openOrCreate, persistTurns, RecordStore } from './record.ts';
import {
Expand Down Expand Up @@ -256,19 +258,10 @@ class SessionImpl implements Session {
private readonly assistant: Assistant;
private readonly here = new Attendance(() => this.record);
private readonly listeners = new Set<(event: SessionEvent) => void>();
private readonly settledWaiters: (() => void)[] = [];
private readonly quietWaiters: (() => void)[] = [];
private readonly streamFn: StreamFn;
private readonly customStream: boolean;
private stopped = false;
/**
* Whether a seat has worked since the room last settled. A failed draft
* waits for the seats to stop again, and a second settle at one quiescence
* — an aborted activation ending after an unseat closed the exchange, a
* question that woke nobody — is not the seats stopping again.
*/
private stirred = false;
/** The room's exchanges: what a question opened, and what quiescence closes. */
/** The room's exchanges: what a question opened, what quiescence closes, and both ends. */
private readonly exchanges = new Exchanges();

constructor(options: StartSessionOptions) {
Expand Down Expand Up @@ -393,8 +386,7 @@ class SessionImpl implements Session {
}

settled(): Promise<void> {
if (!this.working()) return Promise.resolve();
return new Promise((resolve) => this.settledWaiters.push(resolve));
return this.exchanges.settled(this.working());
}

/**
Expand Down Expand Up @@ -425,8 +417,7 @@ class SessionImpl implements Session {
// The same condition the `quiet` event reports. A summary a race left
// owed is not work in flight: it waits for the next quiet room, and the
// room is quiet in the meantime.
if (this.idle()) return Promise.resolve();
return new Promise((resolve) => this.quietWaiters.push(resolve));
return this.exchanges.quiet(this.idle());
}

abort(): void {
Expand Down Expand Up @@ -514,10 +505,7 @@ class SessionImpl implements Session {
// A question that wakes no seat has no seat to stop, so the exchange it
// opened would never close. The same check an ending activation runs,
// and the same last word: the room was quiet, and it says so.
if (this.exchanges.current() !== undefined && !this.working()) {
this.settle();
if (this.idle()) this.markQuiet();
}
if (this.exchanges.current() !== undefined) this.rest(false);
}

/**
Expand Down Expand Up @@ -592,7 +580,7 @@ class SessionImpl implements Session {
// not leave a room that can never be started again.
if (running.get(this.name) === this) running.delete(this.name);
// A stopped room never goes quiet on its own, so nobody waits on it.
for (const resolve of this.quietWaiters.splice(0)) resolve();
this.exchanges.drain();
}
}

Expand Down Expand Up @@ -693,7 +681,7 @@ class SessionImpl implements Session {
// The seat holding it is what makes the room busy: there is no count to
// keep in step, and so none to drift.
seat.activation = activation;
if (this.closingOf(seat) === undefined) this.stirred = true;
if (this.closingOf(seat) === undefined) this.exchanges.stir();
this.emit({ type: 'activation_start', agent: seat.def.name });
// The assistant's activations are one pass: a summary answers a room that
// moved with a redraft inside its own tool, and a composition decides on
Expand All @@ -708,28 +696,26 @@ class SessionImpl implements Session {
private ended(seat: SeatRuntime, activation: Activation): void {
seat.activation = undefined;
const assistant = this.assistant.is(seat.def.name);
const drafted = assistant && this.assistant.composing() === undefined;
if (assistant) {
this.assistant.activationEnded({ wrote: activation.spoke, failed: activation.failed });
}
this.emit({ type: 'activation_end', agent: seat.def.name, spoke: activation.spoke });
// An exchange ends when the seats stop, and a composing assistant is one
// of them. The assistant writing about an exchange is not the room still
// working on it, so a draft's end closes none — which also keeps a failing
// assistant from retrying for ever. What a draft's end frees is the seat,
// for whoever was owed while it drafted.
if (drafted) this.draftNext(this.assistant.dueAfterDraft(...this.dueArgs()));
else if (!this.working()) this.settle();
else if (assistant) this.draftNext(this.assistant.dueAfterDraft(...this.dueArgs()));
if (this.idle()) this.markQuiet();
this.rest(assistant);
}

/** The seats stopped: whoever waited hears it, and the exchange closes. */
private settle(): void {
for (const resolve of this.settledWaiters.splice(0)) resolve();
const worked = this.stirred;
this.stirred = false;
this.closeExchange(worked);
/**
* Something stopped, so the room may have too. The exchange decides: the
* seats settle when none that speaks for itself is working, and the room
* is quiet when nothing at all is. The assistant writing about an exchange
* is not the room still working on it, so a draft's end closes none — which
* also keeps a failing assistant from retrying for ever. What a draft's end
* frees is the seat, for whoever was owed while it drafted.
*/
private rest(freed: boolean): void {
const settled = this.exchanges.settle(this.working(), this.store.lastSeq);
if (settled) this.closed(settled);
else if (freed) this.draftNext(this.assistant.dueAfterDraft(...this.dueArgs()));
this.markQuiet();
}

/** The range the assistant is closing, when this seat is the assistant and it is closing one. */
Expand Down Expand Up @@ -956,18 +942,9 @@ class SessionImpl implements Session {
// -- the assistant ------------------------------------------------------------

/**
* The room went quiet, so the exchange it was working on is over. The host
* The seats settled, so the exchange they were working on is over. The host
* hears that before anything is written about it: the assistant is the first
* reader of a closed exchange and not the only one.
*/
private closeExchange(worked: boolean): void {
const closing = this.exchanges.close(this.store.lastSeq);
if (closing) this.emit({ type: 'exchange_closed', exchange: closing });
this.summariseClosed(closing, worked);
}

/**
* What the assistant makes of a closed exchange: its owner is owed the one
* reader of a closed exchange and not the only one. Its owner is owed the one
* message that stands for it, and the room activates the assistant for it.
* Nothing else in the room wakes for a close — the assistant is seated `none`,
* and the close is the one thing that reaches it.
Expand All @@ -977,8 +954,11 @@ class SessionImpl implements Session {
* waits for the next time the seats stop rather than retrying on itself —
* and a settle that no seat worked before is not the seats stopping again.
*/
private summariseClosed(closing: ClosedExchange | undefined, worked: boolean): void {
if (closing) this.assistant.owe(closing.owner, closing.from);
private closed({ closed, worked }: Settled): void {
if (closed) {
this.emit({ type: 'exchange_closed', exchange: closed });
this.assistant.owe(closed.owner, closed.from);
}
const due = worked
? this.assistant.dueAtQuiescence(...this.dueArgs())
: this.assistant.dueAfterDraft(...this.dueArgs());
Expand Down Expand Up @@ -1017,9 +997,8 @@ class SessionImpl implements Session {
* has gone quiet.
*/
private markQuiet(): void {
if (this.stopped || !this.idle()) return;
this.emit({ type: 'quiet' });
for (const resolve of this.quietWaiters.splice(0)) resolve();
if (this.stopped) return;
if (this.exchanges.quiesce(this.idle())) this.emit({ type: 'quiet' });
}

// -- what an agent reads -------------------------------------------------
Expand Down
Loading
Loading