diff --git a/docs/agent.md b/docs/agent.md index f2c3a65..00c4be2 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -244,6 +244,8 @@ type Message = from: string; // the participant whose presence changed — stamped by the runtime identity?: string; // on 'arrived' and 'seated': how the room knew them by?: string; // on 'seated': the assistant, when it did the seating + attention?: Attention; // on 'seated': what wakes the seat; absent means 'broadcast' + preferences?: string; // on 'arrived': how the person reads, when they said so } | { kind: 'summary'; @@ -534,22 +536,34 @@ reads takes the narrower type and cannot start anything by accident. One file per concern, in layers an import points down through, and `session.ts` is the room that composes them ([`toolchain.md`](toolchain.md) §1 names the layers, and Biome holds them): the -log in [`log.ts`](../packages/ambion/src/log/log.ts), who is here in -[`presence.ts`](../packages/ambion/src/presence.ts), a seat, what wakes it +log in [`log.ts`](../packages/ambion/src/log/log.ts), every fact folded +over it in [`fold.ts`](../packages/ambion/src/room/fold.ts), who is here in +[`presence.ts`](../packages/ambion/src/room/presence.ts), a seat, what wakes it and the seat's side of the wire in [`seat.ts`](../packages/ambion/src/seat/seat.ts), one activation in [`activation.ts`](../packages/ambion/src/seat/activation.ts), the hands it holds in [`hands.ts`](../packages/ambion/src/seat/hands.ts), an activation's id in [`lease.ts`](../packages/ambion/src/room/lease.ts), the exchange in -[`exchange.ts`](../packages/ambion/src/exchange.ts), what the assistant +[`exchange.ts`](../packages/ambion/src/room/exchange.ts), what the assistant writes in [`assistant.ts`](../packages/ambion/src/room/assistant.ts), what crosses between a seat and its room in -[`wire.ts`](../packages/ambion/src/wire.ts), what an +[`wire.ts`](../packages/ambion/src/wire.ts), what an activation is given +in [`view.ts`](../packages/ambion/src/room/view.ts), what an agent's tools reach into in [`workspace.ts`](../packages/ambion/src/tools/workspace.ts), what a host owns in [`runtime.ts`](../packages/ambion/src/host/runtime.ts), and what any of them reads in [`render.ts`](../packages/ambion/src/render.ts). +**The log holds what the room does not hold in memory.** The roster, the +reserve, the people, the open exchange and the closes are each a fold over +the log ([`fold.ts`](../packages/ambion/src/room/fold.ts)). Three kinds of +entry hold them, in the room's one Pi session: `ambion/message`, +`ambion/close` and `ambion/composition`. Every entry beside a message +carries `after`, the last message seq when it was written. A stopped room +reads back the roster its run left, and a run starts from the composition +row it writes. The leases the room holds and the summaries it owes stay in +memory. + **A seat is seated for the run. An activation lasts seconds.** What an activation has heard, what landed while it worked, and whether it left a mark belong to the activation and end with it. Rule 5's `readThrough` is an diff --git a/docs/assistant.md b/docs/assistant.md index c3a222b..3177369 100644 --- a/docs/assistant.md +++ b/docs/assistant.md @@ -98,7 +98,7 @@ answers for anyone. The unit is the **exchange**, and it belongs to the core: [`exchange.md`](exchange.md) specifies it, and -[`exchange.ts`](../packages/ambion/src/exchange.ts) is where it lives. A +[`exchange.ts`](../packages/ambion/src/room/exchange.ts) is where it lives. A question, and everything the room does until it goes quiet again. A person's question opens one; quiescence closes it; what lands in between steers the seats already working and changes nothing. @@ -269,8 +269,8 @@ writes one message per exchange, to one person, and the exchange says whom **A person's exchange outlives their visit.** Priya may ask and walk out before the room settles. The exchange is still hers, it still closes, and the assistant still writes its summary — addressed to her, the way she -reads, waiting for her. How she reads is run state the room keeps after she -leaves. +reads, waiting for her. How she reads is on the record, with her arrival, +so the room keeps it after she leaves. **Sam gets no summary for a question he did not ask.** His message into Priya's exchange steers whoever is working and owns nothing. His own next @@ -526,11 +526,9 @@ assistant does not mean it always writes — it means somebody is always there to judge whether writing would help. **A restarted room seats it again with the agents.** The assistant is -composition, like an agent. How each person reads is run state, like an -exchange (§6): a person known from a replayed record has no preferences on -file until they visit in the new run. No question can be asked without a -visit, so the assistant never writes for somebody whose preferences the room -has not seen. +composition, like an agent. How each person reads is on the record, with +their latest arrival, so a person known from a replayed record reads the +way they last said they do. **An agent-only room pays for one idle seat.** A room nobody visits seats the assistant, lists it in every roster, and never activates it. That is one @@ -735,10 +733,11 @@ message stands for work somebody stopped. `stopSession` is the other case, below. **A run that stops mid-exchange writes no summary.** `stopSession` aborts -the activations in flight, so the exchange never closes. It aborts a draft in -flight for the same reason, and a draft that does finish after the stop -commits nothing. The person asked and heard nothing, and the record shows a -question, some work and a shutdown. Accepted. +the activations in flight and writes no close, so the exchange stays open +on the log. It aborts a draft in flight for the same reason, and a draft +that does finish after the stop commits nothing. The next run over the same +log closes the exchange at its start ([`exchange.md`](exchange.md) §5) and +writes what it owes then. Accepted. **A widened range is bounded by a race, and nothing else.** A summary covers one exchange, so the only thing that can make a range large is what diff --git a/docs/exchange.md b/docs/exchange.md index 241e530..a4a90cf 100644 --- a/docs/exchange.md +++ b/docs/exchange.md @@ -2,7 +2,7 @@ This document is the design contract for the exchange: the room's own unit of work. It is shipped. The code lives in -[`exchange.ts`](../packages/ambion/src/exchange.ts), and +[`exchange.ts`](../packages/ambion/src/room/exchange.ts), and [`session.ts`](../packages/ambion/src/session.ts) opens and closes one as the room runs. Read [`agent.md`](agent.md) first: an exchange is made of the activations that document specifies, and it changes none of the eight @@ -40,7 +40,7 @@ in a sentence a model reads. ## 2. The shape A room is a sequence of exchanges, and the exchanges have one shape -([`exchange.ts`](../packages/ambion/src/exchange.ts)): +([`exchange.ts`](../packages/ambion/src/room/exchange.ts)): ```ts interface Exchange { @@ -76,13 +76,18 @@ the exchange itself, for the case where the room is busy and has no owner: somebody arrives, the seat that watches the door wakes, and a question lands on top of work nobody asked for. That question still owns what follows. -**Quiescence closes it.** The room settles when no agent is active, and a +**Quiescence closes it.** The room settles when nothing is live, and a room that settles has finished. A seat that says something wakes its -readers inside its own `say`, before its own activation ends, so the room is -never briefly empty in the middle of a burst. What is running is read off -the seats, because a seat holds the activation it is taking, so there is no -count beside them to keep in step. `through` is the record as it stood at -that moment, so a closed exchange names the range it turned out to hold. +readers inside its own `say`, before its own lease ends, so the room is +never briefly empty in the middle of a burst. What is live is read off the +leases held and the wakes still pending, so there is no count beside them +to keep in step. The room writes a close row, and `through` is the record +as it stood at the moment the room went quiet, so a closed exchange names +the range it turned out to hold. A quiet the room observed on one +exchange closes that exchange alone. A question that lands after that +moment and before the row is written opens the next exchange. The host +hears `exchange_opened` for it once the row is on the log, and an exchange +nobody works on closes at once, the way a question that wakes nobody does. A question that wakes no seat has no seat to stop, so the room runs the same check once the question is routed: nothing is working, so the @@ -129,18 +134,23 @@ into a quiet room, opens his own exchange. --- -## 5. Run state +## 5. A fold over the log -An exchange belongs to a running room. `Exchanges` holds the open one in -memory, and a restart begins with none. That is right for a room -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. +An exchange is a fold over the log. The open exchange is the first +question a person asked after the last close row's `through` +(`openExchange` in [`exchange.ts`](../packages/ambion/src/room/exchange.ts)). A +close is a row on the log beside the messages: `{ owner, from, through, +at }`. It takes no seq; `through` orders it. `messages()` returns the +messages alone, and their seqs stay `1..n`. -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. -[`planning/backlog.md`](../planning/backlog.md) holds the work. +A run that starts over a log with an exchange open closes it first. The +seats that worked on it went with the run that started it, so nothing is +live, and the close row lands before the room takes its first message. +The host of the new run hears `exchange_closed` for it, and the assistant +writes what that exchange owes. + +Every closed exchange is on the log, so a host that wants a history of +exchanges reads the close rows off the room's Pi session. --- @@ -167,6 +177,10 @@ controls): the assistant owes nobody one. That is the moment a host waits for when it wants the one message a person reads. +Both wait for the room to be up first: a call made right after +`startSession` answers after the replay, and after the close of an +exchange the last run left open (§5). + The two differ because the assistant is a seat like any other, and its activation counts. The assistant writing about an exchange is not the room still working on it, so a drafting activation closes no exchange, and the @@ -182,9 +196,15 @@ summary is drafted, and that window is the one place it can. **An aborted exchange still closes.** `abort()` cancels the activations in flight and the room settles, so the exchange closes with the range it -reached. **A run that stops mid-exchange closes nothing.** `stopSession` -aborts the activations in flight and takes the room down, and the exchange -never closes: the next run begins with none. +reached. **A stopped room closes nothing.** `stopSession` aborts the +activations in flight and writes no close row. A release that lands after +the stop must not write into a log the next run has started over. The +exchange stays open on the log, and the next run closes it at its start +(§5). A run that dies without `stop` leaves the exchange open the same way. + +**A close the storage refuses leaves the exchange open.** `settled()` and +`quiet()` still answer, and the room still says `quiet`; the next time the +seats stop, the room writes the close again. --- @@ -245,9 +265,20 @@ The exchange is proved beside the assistant that first reads one, in - the person whose question opened the exchange owns it, and a second person speaking into it owns nothing (§4); - an exchange outlives its owner's visit (§4); +- an exchange closes at the quiet the room observed, and a question that + lands before the row is written opens the next (§3); +- a quiet observed on one exchange never closes the next, and a question + the assistant already woke on composes nothing and closes at once (§3); - an exchange closes before anything is written about it, and the room settles before it goes quiet (§6). +[`restart.test.ts`](../packages/ambion/test/restart.test.ts) proves that a +stopped room writes no close, and that the next run closes the exchange +before `quiet()` answers (§5, §6). +[`presence.test.ts`](../packages/ambion/test/presence.test.ts) proves that +a close the storage refuses leaves the exchange open, and that whoever +waits still hears the room (§6). + All in-process, in vitest, on a scripted stream. The live run is diff --git a/docs/presence.md b/docs/presence.md index c7831f9..6989f42 100644 --- a/docs/presence.md +++ b/docs/presence.md @@ -166,8 +166,10 @@ opened one, and two tabs of one person do not make two people. The host decides when that person is gone; the room takes its word for it. A second `visitSession` with the same name and a different identity is -refused: one name is one identity for the life of the opening. The -alternative is a roster that changes under the agents reading it. +refused while the person is present: one name is one identity for as long +as they are in the room. The alternative is a roster that changes under +the agents reading it. An absent person may return under a new identity, +and their next `arrived` carries it. `deliver` on a visit that left throws, and so does it on a visit whose run was stopped. A handle to a finished visit is a stale handle, and the @@ -210,6 +212,8 @@ interface Presence { from: string; /** How the room knew them, on `arrived` alone. */ identity?: string; + /** How they read, on `arrived`, when they said so. */ + preferences?: string; } export type Message = Spoken | Presence | Summary; @@ -325,11 +329,14 @@ forget.** An agent that reads `andrei (present)` and calls lands the message. So does an agent in a session reopened next week, because replaying the record replays the arrivals. -Presence itself is live: it is a fact about a running room, and it dies -with the process. What survives is the record of how it changed, and that -rebuilds everything — who has ever been here, who was here last, when, and -where each of them stopped reading. Presence is kept in one place, and the -place is the record. +Presence is a fold over the record (`foldPeople` in +[`presence.ts`](../packages/ambion/src/room/presence.ts)): a person is present +from their last `arrived` until their next `left`. The record rebuilds +everything — who has ever been here, who is here now, when, and where each +of them stopped reading. Presence is kept in one place, and the place is +the record. A run that dies writes no `left`, so a person stays present +until the host says they left: a visit on the next run hands back a handle +and commits nothing, and `leave()` on it writes the `left`. --- diff --git a/docs/roster.md b/docs/roster.md index 527c09d..18c3766 100644 --- a/docs/roster.md +++ b/docs/roster.md @@ -196,10 +196,10 @@ open wakes it the same way, and the runtime hands it one tool, `seat`, bound to the reserve. The assistant bookends the exchange: it composes the room at the open and consolidates what the room said at the close. -The order inside `publish` is what makes it parallel. A question lands, the -room opens the exchange and activates the assistant, then it routes the -question and activates the seats. The assistant reads the question while -the seats do. +The order inside the commit is what makes it parallel. A question lands, +the room sees the exchange it opened and wakes the assistant, then it +routes the question and wakes the seats. The assistant reads the question +while the seats do. **What the assistant is handed.** The same context every seat reads, and two things more: the reserve (§2) as a second roster, and the ask at the @@ -299,11 +299,13 @@ Unseating is the direction the room cannot take back, so the assistant holds no tool for it. [`planning/backlog.md`](../planning/backlog.md) holds the argument for giving it one. -**`stop` unseats what the run added.** `stopSession` commits `left` for -every person present ([`presence.md`](presence.md) §8). It commits -`unseated`, in the same way and without routing, for every seat the run -added after it started. The next run begins from the composition -`startSession` was given, and the record says who was seated in between. +**`stop` leaves the roster to the next composition.** `stopSession` +aborts every activation in flight and commits `left` for every person +present ([`presence.md`](presence.md) §8). It writes no `unseated`. The +next `startSession` writes its own composition row, the roster folds from +that row and the seatings after it, and the record says who was seated in +between. A read of the stopped room (`readSession`) folds the roster the +run left. **A seat that leaves keeps its downstream session.** Rule 8 puts every activation's turns in `:`. An agent seated, unseated and @@ -321,9 +323,9 @@ until something unrelated activated and ended. This case exists today, in a room where every seat is `named` and a question is undirected. It is common once a room may start with the -assistant alone and an empty reserve. So `publish` runs the same check the end -of an activation runs: after routing, if nothing is working, the room -settles and the exchange closes. The exchange holds one message, the +assistant alone and an empty reserve. So the room runs the same check the +end of an activation runs once the question is committed: after routing, +if nothing is working, the room settles and the exchange closes. The exchange holds one message, the question, and the assistant writes nothing for it, because an exchange the agents said nothing into writes nothing ([`assistant.md`](assistant.md) §4). The host hears `exchange_opened`, `exchange_closed` and `quiet`, in @@ -343,10 +345,13 @@ Each boundary is stated so a later change has to argue with it. - **The assistant never defines an agent.** It seats from the reserve, and the host decides what is in it by writing `available`. §2. - **A seat never reads the reserve.** §2. -- **A seating is on the record, and the starting composition is not.** The - record holds what happened in the run. What the run started with is the - run's, as `agent.md` §5 says of the roster and `presence.md` says of the - people. +- **A seating is on the record, and so is the composition.** Every + `startSession` writes a composition row beside the messages: the + assistant, the goal, the agents seated and the agents in reserve, each + with its name, its identity and its attention. The roster folds from the + latest row and the seatings and unseatings after it, so a stopped room + reads back. A read from a process that holds no definition reports every + identity off the log. - **The threshold reads the record.** The rule that a summary is written when the agents said more than one thing counts messages from any name that is not a person and not the assistant, so an agent that spoke and @@ -384,9 +389,11 @@ this document makes loudly: - the host seats and unseats by hand, an unseat aborts the activation in flight, and a say directed at the unseated colleague is refused with the departure (§5); -- `stop` unseats what the run added and leaves the starting composition - alone (§5); -- the threshold counts an agent that spoke and was unseated (§7). +- `stop` leaves the roster to the next composition row, and the next run + starts from its own (§5); +- the threshold counts an agent that spoke and was unseated (§7); +- every identity reads off the log, in a process that holds no definition + (§7, in [`restart.test.ts`](../packages/ambion/test/restart.test.ts)). All in-process, in vitest, on a scripted stream. diff --git a/docs/toolchain.md b/docs/toolchain.md index f9fd3e2..7202576 100644 --- a/docs/toolchain.md +++ b/docs/toolchain.md @@ -71,14 +71,11 @@ holds, and a reviewer reads a file knowing what it cannot reach. | `types`, `wire`, `define`, `render` | The vocabulary: the public shapes, the wire, and what a participant reads | Nothing that does anything | | `host/` | What a host owns: the runtime value, a clock, an opener | The vocabulary | | `log/` | The log: one serial queue over a Pi session | The vocabulary | -| `room/` | What the room holds beside the log: an activation's id, the assistant's rules | The vocabulary | +| `room/` | Every fact the log holds about the room, pure over it: the fold, the exchange, presence, the view | The vocabulary, the log's entries | | `tools/` | What an agent's tools reach into: the workspace and its backends | The vocabulary, `host/` | | `seat/` | The seat side of the wire: one activation, the hands it holds, the actor, the in-process transport | The vocabulary, `host/`, `tools/` | | `session.ts` | The room, which composes them all | Everything | -The two files beside `session.ts` at the root of `src`, the exchange and -presence, are in no layer yet, and no override constrains them. - Two rules hold across packages: the core imports no platform module (`node:sqlite`, `cloudflare:*`), and every other package reaches the core through `@ambionframework/ambion`, its published surface. diff --git a/packages/ambion/src/exchange.ts b/packages/ambion/src/exchange.ts deleted file mode 100644 index ed3f7ce..0000000 --- a/packages/ambion/src/exchange.ts +++ /dev/null @@ -1,84 +0,0 @@ -/** - * The exchange: a question, and everything the room does until it goes quiet - * again. The room goes from idle, to active, and back to idle, and one person - * owns what happens in between. - * - * This is the room's own unit of work, not the assistant's. An assistant is the first - * thing that reads it — it writes one message per exchange — and it is not the - * last: a client folds the working under the question it answered, a host - * measures what an exchange cost, and a later compactor stands over a stretch of - * them. So the rule lives here, on its own, and every reader takes it from the - * same place. - * - * The rule, in three sentences: - * - * - **A person's question opens one**, when no exchange is open. Nothing else - * does: an agent speaking into a quiet room opens nothing, and arriving or - * leaving asks nobody anything. - * - **Quiescence closes it.** The room settles when no agent is active, and a - * room that settles has finished — a seat that says something wakes its - * readers inside its own `say`, so the active count never dips to zero in - * the middle of a burst. - * - **What lands while it is open steers it and changes nothing.** Not the - * owner, not the range, not who the answer belongs to. - * - * The design contract is `docs/exchange.md`; `docs/assistant.md` says what an - * assistant makes of one. - */ -import { isSpoken, type Message, type Seq } from './types.ts'; - -/** A question the room is working on. */ -export interface Exchange { - /** The person whose question opened it, and who owns what follows. */ - readonly owner: string; - /** The seq of that question: where the exchange starts. */ - readonly from: Seq; - /** When it opened, ISO. */ - readonly at: string; -} - -/** An exchange the room has finished, and the range it turned out to hold. */ -export interface ClosedExchange extends Exchange { - /** The last seq on the record when the room went quiet. */ - readonly through: Seq; -} - -/** - * 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. - */ -export class Exchanges { - private open: Exchange | undefined; - - /** What the room is working on, or nothing when nobody has asked. */ - current(): Exchange | undefined { - return this.open; - } - - /** - * A message landed. It opens an exchange when a person asked something into - * a room that has none open, and returns the one it opened. - * - * The clause is written on the exchange rather than on the room's status, - * for the case that is busy and has no owner: somebody arrives, the seat - * that watches the door wakes, and a question lands on top of work nobody - * asked for. That question still owns what follows. - */ - note(message: Message, fromPerson: boolean): Exchange | undefined { - if (this.open !== undefined) return undefined; - if (!fromPerson || !isSpoken(message)) return undefined; - this.open = { owner: message.from, from: message.seq, at: message.at }; - return this.open; - } - - /** - * 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. - */ - close(through: Seq): ClosedExchange | undefined { - const open = this.open; - this.open = undefined; - return open === undefined ? undefined : { ...open, through }; - } -} diff --git a/packages/ambion/src/index.ts b/packages/ambion/src/index.ts index d3ba0a9..7e8f71f 100644 --- a/packages/ambion/src/index.ts +++ b/packages/ambion/src/index.ts @@ -32,8 +32,6 @@ export { } from '@earendil-works/pi-agent-core'; export type { DefineAgentOptions, DefineHumanOptions, DefineToolOptions } from './define.ts'; export { attentive, defineAgent, defineHuman, defineTool, passive, seated } from './define.ts'; -// The room's own exchange: what a question opened, and what quiescence closed. -export type { ClosedExchange, Exchange } from './exchange.ts'; export type { CreateRuntimeOptions, RunningRoom, @@ -71,6 +69,8 @@ export type { AmbionTool, Attention, Clock, + ClosedExchange, + Exchange, HumanDefinition, HumanSeatInfo, Message, @@ -95,8 +95,10 @@ export type { export { isPresence, isSpoken, isSummary } from './types.ts'; export type { ActivationView, + CloseRow, Commit, CommitResponse, + CompositionRow, EndReason, Hand, Intent, @@ -104,6 +106,7 @@ export type { LeaseResponse, SeatPort, SeatRoom, + SeatRow, Stale, ViewResponse, Wake, diff --git a/packages/ambion/src/log/log.ts b/packages/ambion/src/log/log.ts index 6272314..7a6adf6 100644 --- a/packages/ambion/src/log/log.ts +++ b/packages/ambion/src/log/log.ts @@ -1,5 +1,5 @@ /** - * The log: every message a room committed, in the order it took a seq. + * The log: every entry a room committed, in the order it landed. * * It is the one thing a live room and a read of a stopped one share, so it * knows nothing about either: it replays a Pi session into memory and @@ -14,12 +14,51 @@ * `readThrough`: the seq its author has read. The queue refuses it when the * record moved past that, and hands back what the author missed — rule 5, * enforced where the write happens. + * + * Beside the messages, the log holds rows about the room: what a run + * started with, and the range an exchange turned out to hold. A row takes + * no seq and carries `after`, the last seq when it landed. It joins the + * same queue, so a row and the messages around it land in the order they + * were asked for. */ import type { Session as PiSession } from '@earendil-works/pi-agent-core'; import type { Message, Seq } from '../types.ts'; +import type { CloseRow, CompositionRow, Without } from '../wire.ts'; + +/** The three kinds of custom entry the room writes to its Pi session. */ +const ENTRY_TYPES = { + message: 'ambion/message', + close: 'ambion/close', + composition: 'ambion/composition', +} as const; + +/** One entry on the log: a message with a seq, or a row about the room around the messages. */ +export type LogEntry = + | { type: 'message'; message: Message } + | { type: 'close'; close: CloseRow } + | { type: 'composition'; composition: CompositionRow }; + +/** A row that is not a message: it takes no seq, and carries `after`, the last seq when it was written. */ +export type Row = Exclude; -/** The record lives as custom entries of this type in a Pi session. */ -const MESSAGE_ENTRY = 'ambion/message'; +/** What a caller passes to `write`: the row without `after`, which the log stamps. */ +export type RowData = { + close: Without; + composition: Without; +}[K]; + +const BY_TYPE: Record = { + [ENTRY_TYPES.message]: 'message', + [ENTRY_TYPES.close]: 'close', + [ENTRY_TYPES.composition]: 'composition', +}; + +/** The entry a custom row folds as, or nothing for a row the room does not read. */ +function toEntry(customType: string, data: unknown): LogEntry | undefined { + const type = BY_TYPE[customType]; + if (type === undefined) return undefined; + return { type, [type]: data } as LogEntry; +} /** What a caller commits: the message minus its seq, and the two checks the queue runs. */ export interface CommitIntent { @@ -34,6 +73,8 @@ export interface CommitIntent { export type Committed = { message: T; repeated?: true } | { missed: Message[] }; export class RoomLog { + /** Every entry, replayed then appended, in the order the writes were confirmed. */ + readonly entries: LogEntry[] = []; /** The replayed record, then every message as its write is confirmed. */ readonly messages: Message[] = []; readonly ready: Promise; @@ -57,18 +98,47 @@ export class RoomLog { // findEntries does not promise append order; Pi's seq does. found.sort((a, b) => a.seq - b.seq); for (const entry of found) { - if (entry.type !== 'custom' || entry.customType !== MESSAGE_ENTRY) continue; - this.cache(entry.data as Message); + if (entry.type !== 'custom') continue; + const known = toEntry(entry.customType, entry.data); + if (known !== undefined) this.cache(known); } return piSession; } - private cache(message: Message): void { + private cache(entry: LogEntry): void { + this.entries.push(entry); + if (entry.type !== 'message') return; + const message = entry.message; this.messages.push(message); this.lastSeq = message.seq; if (message.key !== undefined) this.byKey.set(message.key, message); } + /** + * Put a row beside the messages. It takes no seq and carries `after`, the + * last seq when it landed; it joins the same queue, so a row and the + * messages around it land in the order they were asked. The row is built + * where the write happens, and a builder that returns nothing writes + * nothing: the check it ran found the row no longer needed. + */ + write( + type: K, + row: RowData | (() => RowData | undefined), + ): Promise { + const link = this.tail.then(async () => { + const piSession = await this.ready; + const data = typeof row === 'function' ? row() : row; + if (data === undefined) return false; + const stamped = { ...data, after: this.lastSeq }; + await piSession.appendCustomEntry(ENTRY_TYPES[type], stamped); + const entry = toEntry(ENTRY_TYPES[type], stamped); + if (entry !== undefined) this.cache(entry); + return true; + }); + this.tail = link.catch(() => {}); + return link; + } + /** * Commit one message. The check, the append and the cache update run * inside one link of the queue, and `landed` runs there too, before the @@ -79,14 +149,19 @@ export class RoomLog { intent: CommitIntent, landed?: (message: T) => void, ): Promise> { - const link = this.tail.then(() => this.write(intent, landed)); + const link = this.tail.then(() => this.land(intent, landed)); // One write that fails must not stop the next one. The queue keeps its // order; the caller of the failed write sees its failure. this.tail = link.catch(() => {}); return link; } - private async write( + /** Resolves once every write asked for so far has landed or failed. */ + settled(): Promise { + return this.tail.then(() => {}); + } + + private async land( intent: CommitIntent, landed: ((message: T) => void) | undefined, ): Promise> { @@ -101,8 +176,8 @@ export class RoomLog { seq: this.lastSeq + 1, ...(intent.key === undefined ? {} : { key: intent.key }), } as T; - await piSession.appendCustomEntry(MESSAGE_ENTRY, stamped); - this.cache(stamped); + await piSession.appendCustomEntry(ENTRY_TYPES.message, stamped); + this.cache({ type: 'message', message: stamped }); landed?.(stamped); return { message: stamped }; } diff --git a/packages/ambion/src/presence.ts b/packages/ambion/src/presence.ts deleted file mode 100644 index 5ece0f3..0000000 --- a/packages/ambion/src/presence.ts +++ /dev/null @@ -1,86 +0,0 @@ -/** - * Who is in the room, and where each of them stopped reading. - * - * The record says who arrived and who left; this holds the one fact a replay - * cannot rebuild — who is here *now* — and reads everything else off the - * record. A person is in the room or they are not: one name, one visit. - */ -import type { HumanDefinition, Message, PresenceMessage, PresenceStatus, Seq } from './types.ts'; - -/** One person in the room, for as long as they are in it. */ -export interface VisitRuntime { - human: HumanDefinition; - gone: boolean; -} - -/** - * Who is in the room, and where each of them stopped reading. The record is - * the store. This holds the one fact a replay cannot rebuild: who is here - * now. Everything else it answers, it reads off the record. - */ -export class Attendance { - private readonly inRoom = new Map(); - - constructor(private readonly record: () => readonly Message[]) {} - - enter(human: HumanDefinition): VisitRuntime { - const visit: VisitRuntime = { human, gone: false }; - this.inRoom.set(human.name, visit); - return visit; - } - - leave(name: string): void { - this.inRoom.delete(name); - } - - visitOf(name: string): VisitRuntime | undefined { - return this.inRoom.get(name); - } - - all(): VisitRuntime[] { - return [...this.inRoom.values()]; - } - - presenceOf(name: string): PresenceStatus { - return this.inRoom.has(name) ? 'present' : 'absent'; - } - - /** Every person the room knows: the arrivals on the record, and who is here. */ - known(): Map { - const known = new Map(); - for (const message of this.record()) { - if (message.kind !== 'arrived') continue; - known.set(message.from, message.identity ?? ''); - } - for (const visit of this.inRoom.values()) known.set(visit.human.name, visit.human.identity); - return known; - } - - knows(name: string): boolean { - return this.known().has(name); - } - - /** The seq of this person's last `left`, or undefined before their first. */ - sinceOf(name: string): Seq | undefined { - return this.lastPresence(name)?.seq; - } - - /** When this person's presence last changed, ISO. */ - lastChangeAt(name: string): string | undefined { - const record = this.record(); - for (let i = record.length - 1; i >= 0; i -= 1) { - const message = record[i]; - if (message && message.kind !== 'said' && message.from === name) return message.at; - } - return undefined; - } - - private lastPresence(name: string): PresenceMessage | undefined { - const record = this.record(); - for (let i = record.length - 1; i >= 0; i -= 1) { - const message = record[i]; - if (message?.kind === 'left' && message.from === name) return message; - } - return undefined; - } -} diff --git a/packages/ambion/src/render.ts b/packages/ambion/src/render.ts index dabd939..852d025 100644 --- a/packages/ambion/src/render.ts +++ b/packages/ambion/src/render.ts @@ -236,7 +236,7 @@ export interface RoomView { } /** The exchange the assistant is closing: whose it was, how they read, and its range. */ -export interface Closing { +interface Closing { /** The person whose question opened it, and who reads the message. */ readonly person: string; /** How they read, or nothing when they said nothing about it. */ @@ -252,7 +252,7 @@ interface Reserved { } /** The exchange the assistant is composing the room for: whose question, and who is in reserve. */ -export interface ComposingView { +interface ComposingView { readonly person: string; readonly from: Seq; readonly reserve: readonly Reserved[]; diff --git a/packages/ambion/src/room/assistant.ts b/packages/ambion/src/room/assistant.ts index 693943f..d2056e6 100644 --- a/packages/ambion/src/room/assistant.ts +++ b/packages/ambion/src/room/assistant.ts @@ -21,9 +21,10 @@ * * What is left in this file is what the assistant *is*: what a room refuses * to seat as one, the threshold a summary is written above, and what the - * room holds of the assistant while it runs: how each person reads, who is - * owed, and the one draft or composition in flight. The two tools are hands - * the seat side gives it (`seat/hands.ts`). + * room holds of the assistant while it runs: who is owed, and the one draft + * or composition in flight. How each person reads is on the record, with + * their arrival. The two tools are hands the seat side gives it + * (`seat/hands.ts`). */ import type { AgentDefinition, Message, Seq } from '../types.ts'; import { isAgent, isSpoken } from '../types.ts'; @@ -97,16 +98,14 @@ export interface Composing { } /** - * The assistant in one room: how each person reads, who is owed a message, - * and the one it is drafting now. + * The assistant in one room: who is owed a message, and the one it is + * drafting now. * * A seat knows nothing about any of this. The assistant is a seat like every * other, and what makes it the assistant is held here — so the room asks *the * assistant* whether a name is it, rather than every seat carrying the answer. */ export class Assistant { - /** How each person who visited this run reads. Run state: a restart begins empty. */ - private readonly preferences = new Map(); /** * People owed a message, and the seq their range starts at. A race or a * failed activation leaves one owed; the next quiet room writes it. @@ -147,15 +146,6 @@ export class Assistant { return name === this.name; } - /** A person is in the room: how they read, as their latest visit says it. */ - serve(person: string, preferences: string | undefined): void { - this.preferences.set(person, preferences); - } - - preferencesOf(person: string): string | undefined { - return this.preferences.get(person); - } - /** What the assistant is closing, while it is closing it. */ closing(): Draft | undefined { return this.draft; diff --git a/packages/ambion/src/room/exchange.ts b/packages/ambion/src/room/exchange.ts new file mode 100644 index 0000000..5113673 --- /dev/null +++ b/packages/ambion/src/room/exchange.ts @@ -0,0 +1,46 @@ +/** + * The exchange: a question, and everything the room does until it goes quiet + * again. The room goes from idle, to active, and back to idle, and one person + * owns what happens in between. + * + * This is the room's own unit of work, not the assistant's. An assistant is the first + * thing that reads it — it writes one message per exchange — and it is not the + * last: a client folds the working under the question it answered, a host + * measures what an exchange cost, and a later compactor stands over a stretch of + * them. So the rule lives here, on its own, and every reader takes it from the + * same place. + * + * The rule, in three sentences: + * + * - **A person's question opens one**, when no exchange is open. Nothing else + * does: an agent speaking into a quiet room opens nothing, and arriving or + * leaving asks nobody anything. + * - **Quiescence closes it.** The room reconciles when nothing is live, and + * writes a close row that names the range the exchange turned out to hold. + * - **What lands while it is open steers it and changes nothing.** Not the + * owner, not the range, not who the answer belongs to. + * + * An exchange is a fold over the log: the first person's question after the + * last close is the open one. A room resumed mid-exchange continues it. + * + * The design contract is `docs/exchange.md`; `docs/assistant.md` says what an + * assistant makes of one. + */ +import { type Exchange, isSpoken, type Message } from '../types.ts'; +import type { CloseRow } from '../wire.ts'; + +/** + * The open exchange, or nothing when nobody has asked since the last close: + * the first question a person asked after the last close's `through`. + */ +export function openExchange( + messages: readonly Message[], + closes: readonly CloseRow[], + isPerson: (name: string) => boolean, +): Exchange | undefined { + const closedThrough = closes.at(-1)?.through ?? 0; + const question = messages.find( + (message) => message.seq > closedThrough && isSpoken(message) && isPerson(message.from), + ); + return question && { owner: question.from, from: question.seq, at: question.at }; +} diff --git a/packages/ambion/src/room/fold.ts b/packages/ambion/src/room/fold.ts new file mode 100644 index 0000000..90e5f32 --- /dev/null +++ b/packages/ambion/src/room/fold.ts @@ -0,0 +1,95 @@ +/** + * Every fact the log holds about the room, as a fold over its entries. + * + * The roster, the reserve, the people, the open exchange and the closes are + * each one function over the entries. A room that replays the log folds + * the same state the room that wrote it held, which is what lets a stopped + * room say who was in it, and a run start from what its last run held. + */ + +import type { LogEntry } from '../log/log.ts'; +import type { Attention, Exchange, Message, Seq } from '../types.ts'; +import type { CloseRow, CompositionRow, SeatRow } from '../wire.ts'; +import { openExchange } from './exchange.ts'; +import { foldPeople, type PersonState } from './presence.ts'; + +/** One agent on the roster: its name, how the room knows it, what wakes it, and whether it is the assistant. */ +interface RosterSeat { + name: string; + identity: string; + attention: Attention; + assistant: boolean; +} + +export interface RoomState { + readonly composition: CompositionRow | undefined; + readonly roster: RosterSeat[]; + readonly reserve: SeatRow[]; + readonly people: Map; + readonly exchange: Exchange | undefined; + readonly closes: CloseRow[]; + readonly messages: readonly Message[]; + readonly lastSeq: Seq; +} + +/** The entries, sorted by kind. The latest composition stands. */ +function sorted(entries: readonly LogEntry[]) { + const messages: Message[] = []; + const closes: CloseRow[] = []; + let composition: CompositionRow | undefined; + for (const entry of entries) { + if (entry.type === 'message') messages.push(entry.message); + else if (entry.type === 'close') closes.push(entry.close); + else composition = entry.composition; + } + return { messages, closes, composition }; +} + +export function foldRoom(entries: readonly LogEntry[]): RoomState { + const { messages, closes, composition } = sorted(entries); + const people = foldPeople(messages); + const roster = foldRoster(composition, messages); + const isPerson = (name: string) => people.has(name); + return { + composition, + roster, + reserve: + composition?.available.filter((seat) => !roster.some((s) => s.name === seat.name)) ?? [], + people, + exchange: openExchange(messages, closes, isPerson), + closes, + messages, + lastSeq: messages.at(-1)?.seq ?? 0, + }; +} + +/** The latest composition, then every seating and unseating after it, in order. */ +function foldRoster( + composition: CompositionRow | undefined, + messages: readonly Message[], +): RosterSeat[] { + if (composition === undefined) return []; + const roster: RosterSeat[] = [ + ...composition.agents.map((seat) => ({ ...seat, assistant: false })), + { ...composition.assistant, assistant: true }, + ]; + for (const message of messages) { + if (message.seq > composition.after) reseat(roster, message); + } + return roster; +} + +/** One seating or unseating applied to the roster. Any other message changes nothing. */ +function reseat(roster: RosterSeat[], message: Message): void { + if (message.kind !== 'seated' && message.kind !== 'unseated') return; + const at = roster.findIndex((seat) => seat.name === message.from); + if (at >= 0) roster.splice(at, 1); + if (message.kind === 'seated') { + roster.push({ + name: message.from, + identity: message.identity ?? '', + attention: message.attention ?? 'broadcast', + assistant: false, + }); + } +} diff --git a/packages/ambion/src/room/presence.ts b/packages/ambion/src/room/presence.ts new file mode 100644 index 0000000..ed5db61 --- /dev/null +++ b/packages/ambion/src/room/presence.ts @@ -0,0 +1,58 @@ +/** + * Who is in the room, and where each of them stopped reading. + * + * Presence is a fold over the record: a person is present from their last + * `arrived` until their next `left`. A crash writes no `left`, so the person + * stays present until the host says they left. The one thing the record + * does not hold is the handle a host delivers through, and that stays in + * the running room. + */ +import type { HumanDefinition, Message, PresenceStatus, Seq } from '../types.ts'; + +/** One person in the room, for as long as they are in it. */ +export interface VisitRuntime { + human: HumanDefinition; + gone: boolean; +} + +/** One person the record knows, as the record last saw them. */ +export interface PersonState { + name: string; + identity: string; + presence: PresenceStatus; + /** The seq of their last `left`, or undefined before their first. */ + since: Seq | undefined; + /** When their presence last changed, ISO. */ + changedAt: string | undefined; + /** How they read, as their latest arrival said it. */ + preferences: string | undefined; +} + +/** Every person the record knows, in the order the record met them. */ +export function foldPeople(messages: readonly Message[]): Map { + const people = new Map(); + for (const message of messages) { + if (message.kind === 'arrived') { + const known = people.get(message.from); + people.set(message.from, { + name: message.from, + identity: message.identity ?? known?.identity ?? '', + presence: 'present', + since: known?.since, + changedAt: message.at, + preferences: message.preferences ?? known?.preferences, + }); + } else if (message.kind === 'left') { + const known = people.get(message.from); + if (known) { + people.set(message.from, { + ...known, + presence: 'absent', + since: message.seq, + changedAt: message.at, + }); + } + } + } + return people; +} diff --git a/packages/ambion/src/room/view.ts b/packages/ambion/src/room/view.ts new file mode 100644 index 0000000..1167aec --- /dev/null +++ b/packages/ambion/src/room/view.ts @@ -0,0 +1,150 @@ +/** + * What an activation is given, read off the fold and rendered: the seats, + * the people, the record, the hand the activation holds and what it holds + * it for. Every function is pure over the facts it is handed, so the view a + * seat reads in one process is the view it reads in another. + */ + +import { + type PersonView, + type RoomView, + renderSystemPrompt, + renderTurnContext, + type SeatSpeaking, +} from '../render.ts'; +import type { AgentDefinition, Exchange, SeatInfo, Seq } from '../types.ts'; +import type { ActivationView, Hand } from '../wire.ts'; +import type { RoomState } from './fold.ts'; +import { parseId } from './lease.ts'; + +/** What the view is built from: the fold, and what the room holds beside it. */ +export interface RoomFacts { + readonly name: string; + readonly now: number; + readonly assistant: string; + readonly state: RoomState; + /** The seats live now, by name. */ + readonly live: ReadonlySet; + /** The range the assistant is closing now, when it is closing one. */ + readonly drafting: { person: string; from: Seq; through: Seq } | undefined; + /** How many messages landed after this seq. */ + unseen(since: Seq): number; +} + +/** The roster and the people, as `seats()` reports them, off one folded state and nothing else. */ +export function seatsOf(facts: Pick): SeatInfo[] { + const seats: SeatInfo[] = facts.state.roster.map((seat) => ({ + kind: 'agent' as const, + name: seat.name, + identity: seat.identity, + status: facts.live.has(seat.name) ? ('active' as const) : ('idle' as const), + attention: seat.attention, + sessionId: `${facts.name}:${seat.name}`, + ...(seat.assistant ? { assistant: true as const } : {}), + })); + for (const person of facts.state.people.values()) { + seats.push({ + kind: 'human', + name: person.name, + identity: person.identity, + presence: person.presence, + }); + } + return seats; +} + +/** The view one activation reads: two rendered strings, the model id, and the hand. */ +export function viewOf( + id: string, + seat: string, + def: AgentDefinition, + facts: RoomFacts, +): ActivationView { + const state = facts.state; + const { hand, closing, composing } = handOf(id, seat, facts); + const speaking: SeatSpeaking = { + def, + assistant: seat === facts.assistant, + closing: closing && { ...closing, preferences: state.people.get(closing.person)?.preferences }, + composing: composing && { ...composing, reserve: reserved(facts) }, + }; + const room = roomView(facts); + return { + activation: id, + seat, + model: def.model, + lastSeq: state.lastSeq, + systemPrompt: renderSystemPrompt(speaking, room), + context: renderTurnContext(speaking, room), + hand, + ...(closing ? { closing } : {}), + ...(composing ? { composing } : {}), + }; +} + +type Hands = { + hand: Hand; + closing?: ActivationView['closing']; + composing?: ActivationView['composing']; +}; + +/** + * What an activation is for, read off its id: a draft closes the exchange + * the room is closing, the assistant woken by the question that opened an + * exchange composes the room for it, and every other seat speaks. + */ +function handOf(id: string, seat: string, facts: RoomFacts): Hands { + const state = facts.state; + const parsed = parseId(id); + if (parsed?.kind === 'draft') { + const drafting = facts.drafting; + if (drafting === undefined || drafting.through !== parsed.through) return { hand: 'none' }; + return { hand: 'summarise', closing: { ...drafting } }; + } + if (seat !== facts.assistant) return { hand: 'say' }; + const question = parsed && state.messages.find((m) => m.seq === parsed.seq); + const opened = openedBy(parsed?.seq, state); + if (question === undefined || !opened) return { hand: 'none' }; + return { + hand: 'seat', + composing: { person: question.from, from: question.seq, limit: state.reserve.length }, + }; +} + +/** Whether the message at `seq` opened an exchange, open or closed since. */ +function openedBy(seq: Seq | undefined, state: RoomState): boolean { + if (seq === undefined) return false; + return state.exchange?.from === seq || state.closes.some((close) => close.from === seq); +} + +/** The reserve as the assistant reads it: a name and an identity per agent. */ +function reserved(facts: RoomFacts): { name: string; identity: string }[] { + return facts.state.reserve.map((seat) => ({ name: seat.name, identity: seat.identity })); +} + +/** What the prose is given of this room, built fresh for each activation. */ +function roomView(facts: RoomFacts): RoomView { + const state = facts.state; + const exchange: Exchange | undefined = state.exchange; + return { + name: facts.name, + goal: state.composition?.goal, + now: facts.now, + seats: seatsOf(facts), + people: peopleViews(facts), + record: state.messages, + exchange: exchange && { owner: exchange.owner, from: exchange.from }, + }; +} + +/** One entry per person the room knows, with their gap and what they missed. */ +function peopleViews(facts: RoomFacts): PersonView[] { + return [...facts.state.people.values()].map((person) => ({ + name: person.name, + identity: person.identity, + presence: person.presence, + changedAt: person.changedAt, + since: person.since, + unseen: person.since === undefined ? 0 : facts.unseen(person.since), + })); +} diff --git a/packages/ambion/src/session.ts b/packages/ambion/src/session.ts index 02a29f0..34e5d28 100644 --- a/packages/ambion/src/session.ts +++ b/packages/ambion/src/session.ts @@ -1,16 +1,18 @@ /** - * The room: the one place where a record, the seats around it, the people + * The room: the one place where a log, the seats around it, the people * visiting it and the exchanges they open become behaviour. * - * Everything with a life of its own has left. The log is `log/log.ts`, who - * is here is `presence.ts`, a seat, what wakes it and the seat's side of the - * wire is `seat/seat.ts`, one activation is `seat/activation.ts`, the hands - * it holds are `seat/hands.ts`, an exchange is `exchange.ts`, the assistant - * is `room/assistant.ts`, and every sentence a participant reads is - * `render.ts`. What is left is what only a room can do: + * Everything with a life of its own has left. The log is `log/log.ts`, and + * every fact the log holds about the room is a fold over it in + * `room/fold.ts`: the roster, the reserve, the people, the open exchange + * and the closes. A seat, what wakes it and the seat's side of the wire is + * `seat/seat.ts`, one activation is `seat/activation.ts`, the hands it + * holds are `seat/hands.ts`, the assistant is `room/assistant.ts`, what an + * activation reads is `room/view.ts`, and every sentence a participant + * reads is `render.ts`. What is left is what only a room can do: * - * - **Compose.** Seat the agents and the assistant, hold the reserve, admit the - * people, seat and unseat while it runs, and take it all down again. + * - **Compose.** Write the composition, admit the people, seat and unseat + * while it runs, and take it all down again. * - **Commit.** One queue, one seq at a time, for every author (rule 5), and * one `message` event per message however it was written. * - **Route.** Who hears a message, and who wakes for it. @@ -20,11 +22,10 @@ * side releases it; the room cuts an activation it ends, and the seat side * releases the lease then. The expiry the room hands out is what the seat * side renews against. - * - **Say when it has stopped.** An exchange closed, and nothing live. + * - **Say when it has stopped.** An exchange closed, with a row on the log, + * and nothing live. */ import type { SessionRepo, StreamFn } from '@earendil-works/pi-agent-core'; -import { seated } from './define.ts'; -import { type ClosedExchange, type Exchange, Exchanges } from './exchange.ts'; import { defaultRuntime, type RunningRoom, @@ -34,25 +35,20 @@ import { type Transport, } from './host/runtime.ts'; import { RoomLog } from './log/log.ts'; -import { Attendance, type VisitRuntime } from './presence.ts'; -import { - type Closing, - type ComposingView, - type PersonView, - type RoomView, - renderLine, - renderSystemPrompt, - renderTurnContext, - type SeatSpeaking, -} from './render.ts'; +import { renderLine } from './render.ts'; import { Assistant, assertAssistant, type Draft } from './room/assistant.ts'; +import { foldRoom, type RoomState } from './room/fold.ts'; import { activationId, draftId, parseId, seatOf } from './room/lease.ts'; +import type { VisitRuntime } from './room/presence.ts'; +import { type RoomFacts, seatsOf, viewOf } from './room/view.ts'; import { inProcessTransport, SeatActor, wakes } from './seat/seat.ts'; import { type AgentDefinition, type AgentSeat, type Attention, authorOf, + type ClosedExchange, + type Exchange, type HumanDefinition, isAgent, isSeatedAgent, @@ -69,28 +65,33 @@ import { type SummaryMessage, } from './types.ts'; import type { - ActivationView, Commit, CommitResponse, + CompositionRow, EndReason, - Hand, Lease, LeaseResponse, SeatPort, + SeatRow, ViewResponse, + Without, } from './wire.ts'; /** How long a lease runs between renewals. The seat side renews at half of it. */ const LEASE_EXPIRY = 60_000; -/** An agent on the roster or in the reserve: the definition, and the attention it takes when seated. */ +/** An agent with the attention it takes when seated. */ interface Placed { def: AgentDefinition; attention: Attention; - /** Seated after the room started, so `stop` unseats it and the record says so. */ - added?: true; - /** Seated from the reserve, so an unseat returns it there. */ - reserved?: true; +} + +/** What a run starts with, as values. The row on the log is the same, by name. */ +interface Composition { + assistant: AgentDefinition; + goal: string | undefined; + agents: Placed[]; + available: Placed[]; } /** One lease the room holds: whose it is, and whether the activation left a mark. */ @@ -105,6 +106,15 @@ type Drafted = | Omit | Omit; +/** A presence change before the room stamps when it happened. */ +type PresenceDraft = Omit; + +/** The exchange the room went quiet on, and the record as it stood then. */ +interface Observed { + from: Seq; + through: Seq; +} + export interface StartSessionOptions { /** The session's name: the record belongs to it, across every run. */ name: string; @@ -153,10 +163,7 @@ export interface SessionView { } export interface Session extends SessionView { - /** - * The question the room is working on, or nothing when nobody has asked. - * Run state: a restart begins with none. - */ + /** The question the room is working on, or nothing when nobody has asked. A fold over the log. */ exchange(): Exchange | undefined; /** Resolves when no seat that speaks for itself is live and the assistant is not composing. */ settled(): Promise; @@ -231,16 +238,15 @@ export function readSession(name: string, options: ReadSessionOptions = {}): Ses return new ReadOnlySession(name, options.repo ? sessionsOver(options.repo) : runtime.sessions); } +/** A read needs the log and nothing else: every identity it reports is on the log. */ class ReadOnlySession implements SessionView { private readonly log: RoomLog; - private readonly here: Attendance; constructor( readonly name: string, sessions: SessionOpener, ) { this.log = new RoomLog(sessions.open(name)); - this.here = new Attendance(() => this.log.messages); } async messages(options: { since?: Seq } = {}): Promise { @@ -248,14 +254,9 @@ class ReadOnlySession implements SessionView { return this.log.since(options.since); } - /** A room that is not running has no agents standing up, and nobody in it. */ + /** The roster the log folds, and everybody the record knows. Nothing stands up. */ seats(): SeatInfo[] { - return [...this.here.known()].map(([name, identity]) => ({ - kind: 'human' as const, - name, - identity, - presence: this.here.presenceOf(name), - })); + return seatsOf({ name: this.name, state: foldRoom(this.log.entries), live: new Set() }); } /** Nothing is running, so nothing happens. The listener is never called. */ @@ -276,22 +277,20 @@ class SessionImpl implements Session, RunningRoom { readonly stream: StreamFn; readonly model: ModelResolver; readonly sessions: SessionOpener; - private readonly goal?: string; private readonly runtime: Runtime; /** How this room reaches a seat: what the runtime holds, or every seat as an actor in this process. */ private readonly transport: Transport; private readonly log: RoomLog; - /** - * The record replayed, and the composition checked against it: a name the - * record knows as a person cannot be seated. Every operation waits here. - */ + /** The replay, the composition on the log, and the close of an exchange the last run left open. Every operation waits here. */ private readonly ready: Promise; - private readonly agents = new Map(); - /** The reserve: agents the room may seat later, held with the attention they will take. */ - private readonly reserve = new Map(); - /** The room's assistant: how each person reads, who is owed, what it is drafting or composing. */ + /** Every definition this room can seat, by name. */ + private readonly defs = new Map(); + /** The row this run writes about itself. Before the replay, `seats()` folds this row alone. */ + private readonly starting: Without; + /** The handles the host delivers through. Presence itself is a fold over the log. */ + private readonly visits = new Map(); + /** The room's assistant: who is owed, and what it is drafting or composing. */ 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)[] = []; @@ -304,7 +303,10 @@ class SessionImpl implements Session, RunningRoom { /** How many drafts the assistant took at each seq, so a second draft at one seq takes a new id. */ private readonly drafts = new Map(); private readonly ports = new Map(); - /** The record is replayed and the composition holds: a seat's call is answered on the spot. */ + /** The settle in flight: the close row it writes, and whoever it wakes. A caller that asks waits for it. */ + private settling: Promise = Promise.resolve(); + private fold: { length: number; state: RoomState } | undefined; + /** The record is replayed and the composition is on the log: a seat's call is answered on the spot. */ private replayed = false; private stopped = false; /** @@ -314,42 +316,51 @@ class SessionImpl implements Session, RunningRoom { * 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. */ - private readonly exchanges = new Exchanges(); constructor(options: StartSessionOptions, runtime: Runtime) { + const composition = composeFrom(options); this.name = options.name; - this.goal = options.goal?.trim() || undefined; this.runtime = runtime; this.transport = runtime.transport ?? inProcessTransport(); this.sessions = options.repo ? sessionsOver(options.repo) : runtime.sessions; this.log = new RoomLog(this.sessions.open(this.name)); - for (const seat of options.agents ?? []) this.place(seat); - // Seated at the narrow end: nothing said in the room wakes the assistant; - // the open and the close of an exchange do, and it is here for the whole run. - const assistant = assertAssistant(options.assistant); - this.place(seated(assistant, 'none')); - this.assistant = new Assistant(assistant.name); - for (const seat of options.available ?? []) this.hold(seat); this.stream = options.streamFn ?? runtime.stream; this.model = options.streamFn ? stubModel : runtime.model; + this.assistant = new Assistant(composition.assistant.name); + this.starting = compositionRow(composition, this.now()); + this.know(...composition.agents, ...composition.available, { + def: composition.assistant, + attention: 'none', + }); this.ready = this.compose(); void this.ready.catch(() => {}); } + /** A definition the seat side resolves by name: on this room, and on the runtime's catalog. */ + private know(...placed: Placed[]): void { + for (const { def } of placed) { + this.defs.set(def.name, def); + this.runtime.catalog.set(def.name, def); + } + } + /** - * The composition against the record: `assertFreeName` reads the record, - * so the check the constructor ran saw an empty one. This is the check - * that counts, and the first call that needs the room sees its refusal. + * The composition against the record, then on it. A name the record knows + * as a person cannot be seated, and the first call that needs the room sees + * the refusal. The row is what the roster folds from. An exchange the last + * run left open closes now: the seats that worked on it are gone with it. */ private async compose(): Promise { await this.log.ready; - for (const name of [...this.agents.keys(), ...this.reserve.keys()]) { - if (this.here.knows(name)) { + const people = this.state().people; + for (const name of this.defs.keys()) { + if (people.has(name)) { throw new Error(`Duplicate agent name '${name}': one name names one participant.`); } } + await this.log.write('composition', this.starting); this.replayed = true; + if (this.state().exchange !== undefined) await this.settle(); } /** @@ -361,103 +372,30 @@ class SessionImpl implements Session, RunningRoom { if (!this.replayed) await this.ready; } + // -- what the room holds -------------------------------------------------- + /** The room's clock, as an ISO stamp for the record. */ private now(): string { return new Date(this.runtime.clock.now()).toISOString(); } - private get record(): Message[] { - return this.log.messages; - } - - // -- the roster -------------------------------------------------------------- - - /** Seat one agent, refusing a name the room already knows. */ - private place(seat: AgentSeat): Placed { - const placed = this.unwrap(seat); - this.assertFreeName(placed.def.name); - this.agents.set(placed.def.name, placed); - return placed; - } - - /** Hold one agent in reserve, refusing a name the room already knows. */ - private hold(seat: AgentSeat): void { - const held = this.unwrap(seat); - this.assertFreeName(held.def.name); - this.reserve.set(held.def.name, held); - } - - /** The definition and its attention. The seat side resolves the definition by name, so the catalog knows it. */ - private unwrap(seat: AgentSeat): Placed { - const def = isSeatedAgent(seat) ? seat.agent : seat; - if (!isAgent(def)) { - throw new Error('Agents must come from defineAgent or seated().'); - } - this.runtime.catalog.set(def.name, def); - return { def, attention: isSeatedAgent(seat) ? seat.attention : 'broadcast' }; - } - - /** One name names one participant: seated, in reserve, or a person the room knows. */ - private assertFreeName(name: string): void { - if (this.agents.has(name) || this.reserve.has(name) || this.here.knows(name)) { - throw new Error(`Duplicate agent name '${name}': one name names one participant.`); - } - } - - /** The host puts an agent on the roster. From the reserve when it is there; from anywhere else too. */ - async seat(seat: AgentSeat): Promise { - this.assertRunning(); - await this.ready; - const given = this.unwrap(seat); - const held = this.reserve.get(given.def.name); - if (held) this.reserve.delete(given.def.name); - // A bare definition takes the attention its reserve entry carried. - const attention = isSeatedAgent(seat) ? seat.attention : (held?.attention ?? 'broadcast'); - const placed = this.place(seated(given.def, attention)); - placed.added = true; - if (held) placed.reserved = true; - await this.commitPresence({ - kind: 'seated', - from: given.def.name, - identity: given.def.identity, - }); + /** Every fact the log holds about the room, folded over it as it stands. */ + private state(): RoomState { + const length = this.log.entries.length; + if (this.fold?.length !== length) this.fold = { length, state: foldRoom(this.log.entries) }; + return this.fold.state; } - /** The host takes an agent off the roster. Never the assistant. */ - async unseat(agent: AgentDefinition): Promise { - this.assertRunning(); - await this.ready; - const seat = this.agents.get(agent.name); - if (!seat) throw new Error(`'${agent.name}' is not seated in this session.`); - if (this.assistant.is(agent.name)) { - throw new Error(`'${agent.name}' is the assistant: a room cannot run without one.`); - } - this.retire(seat); - await this.commitPresence({ kind: 'unseated', from: agent.name }); + private get record(): readonly Message[] { + return this.log.messages; } - /** - * Off the roster: what was mid-flight is cut, a wake it was sent and never - * claimed is forgotten, and a reserve agent goes back to the reserve. The - * lease it holds ends when the seat side releases it. - */ - private retire(seat: Placed): void { - const name = seat.def.name; - const port = this.ports.get(name); - if (port instanceof SeatActor) port.abort(); - for (const [id, of] of this.pending) if (of === name) this.pending.delete(id); - this.agents.delete(name); - if (seat.reserved) this.reserve.set(name, { def: seat.def, attention: seat.attention }); + private onRoster(name: string, state = this.state()): boolean { + return state.roster.some((seat) => seat.name === name); } - /** The assistant seats one name from the reserve, where its seating lands. */ - private admit(name: string): void { - const held = this.reserve.get(name); - if (!held) throw new Error(`'${name}' is not in the reserve.`); - this.reserve.delete(name); - const placed = this.place(seated(held.def, held.attention)); - placed.added = true; - placed.reserved = true; + private assertRunning(): void { + if (this.stopped) throw new Error(`Session '${this.name}' is stopped.`); } // -- what the host reads ----------------------------------------------------- @@ -478,22 +416,42 @@ class SessionImpl implements Session, RunningRoom { } exchange(): Exchange | undefined { - return this.exchanges.current(); + return this.state().exchange; } - settled(): Promise { - if (!this.working()) return Promise.resolve(); + /** + * Both answers wait for the room to be up, then for every settle in + * flight: a run over a log with an exchange open closes it as it starts, + * and the close and whoever it wakes are on the record before either + * answers. A release that lands while one settle is awaited chains the + * next, and the answer waits for that one too. + */ + async settled(): Promise { + await this.up(); + await this.stilled(); + if (!this.working()) return; return new Promise((resolve) => this.settledWaiters.push(resolve)); } - quiet(): Promise { + async quiet(): Promise { // 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(); + await this.up(); + await this.stilled(); + if (this.idle()) return; return new Promise((resolve) => this.quietWaiters.push(resolve)); } + /** Resolves once the settle chain stands still: nothing was chained while it was awaited. */ + private async stilled(): Promise { + let awaited: Promise; + do { + awaited = this.settling; + await awaited; + } while (awaited !== this.settling); + } + /** Whether a seat is live: it holds a lease, or a wake was sent to it and not yet claimed. */ private live(name: string): boolean { for (const held of this.leases.values()) if (held.seat === name) return true; @@ -501,18 +459,14 @@ class SessionImpl implements Session, RunningRoom { return false; } - /** - * What is live, on the roster. One fact, kept in one place: a lease is held - * or a wake is pending, so nothing counts activations alongside and - * nothing can drift. - */ - private running(): string[] { - return [...this.agents.keys()].filter((name) => this.live(name)); + /** The seats live now, on the roster. */ + private liveSeats(state = this.state()): Set { + return new Set(state.roster.map((seat) => seat.name).filter((name) => this.live(name))); } /** Nothing at all is live. The assistant writing is something. */ private idle(): boolean { - return this.running().length === 0; + return this.liveSeats().size === 0; } /** @@ -523,7 +477,7 @@ class SessionImpl implements Session, RunningRoom { */ private working(): boolean { const composing = this.assistant.composing() !== undefined; - return this.running().some((name) => composing || !this.assistant.is(name)); + return [...this.liveSeats()].some((name) => composing || !this.assistant.is(name)); } /** Cut every activation in flight. Each seat side releases its lease, and the room hears how it ended. */ @@ -536,23 +490,15 @@ class SessionImpl implements Session, RunningRoom { return this.log.since(options.since); } + /** The roster and the people off the fold. Before the replay, the fold is over the row this run writes. */ seats(): SeatInfo[] { - const seats: SeatInfo[] = []; - for (const seat of this.agents.values()) { - seats.push({ - kind: 'agent', - name: seat.def.name, - identity: seat.def.identity, - status: this.live(seat.def.name) ? 'active' : 'idle', - attention: seat.attention, - sessionId: `${this.name}:${seat.def.name}`, - ...(this.assistant.is(seat.def.name) ? { assistant: true as const } : {}), - }); - } - for (const [name, identity] of this.here.known()) { - seats.push({ kind: 'human', name, identity, presence: this.here.presenceOf(name) }); + if (!this.replayed) { + const composition = { ...this.starting, after: 0 }; + const state = foldRoom([{ type: 'composition', composition }]); + return seatsOf({ name: this.name, state, live: new Set() }); } - return seats; + const state = this.state(); + return seatsOf({ name: this.name, state, live: this.liveSeats(state) }); } // -- people ----------------------------------------------------------------- @@ -561,58 +507,50 @@ class SessionImpl implements Session, RunningRoom { async visit(human: HumanDefinition): Promise { this.assertRunning(); await this.ready; - // Checked after the replay: a name the record knows is only known then. this.assertVisitable(human); - const already = this.here.visitOf(human.name); - if (already) return this.handle(already); - // The room changes before the message does: a seat woken by the arrival - // must read a roster that already agrees with it. - const visit = this.here.enter(human); - // How they read outlives the visit: an exchange they opened is finished - // properly or not at all, and its message is written after they leave. - this.assistant.serve(human.name, human.preferences); - await this.commitPresence({ kind: 'arrived', from: human.name, identity: human.identity }); + const known = this.visits.get(human.name); + if (known) return this.handle(known); + const visit: VisitRuntime = { human, gone: false }; + this.visits.set(human.name, visit); + // A person the log holds as present is here already: the last run wrote + // no `left`, and the host's word is what says otherwise. Nothing commits. + if (this.state().people.get(human.name)?.presence !== 'present') { + await this.commitPresence({ + kind: 'arrived', + from: human.name, + identity: human.identity, + ...(human.preferences === undefined ? {} : { preferences: human.preferences }), + }); + } return this.handle(visit); } + /** One name names one participant, and a present person keeps one identity. */ private assertVisitable(human: HumanDefinition): void { - if (this.agents.has(human.name) || this.reserve.has(human.name)) { + const state = this.state(); + if (this.defs.has(human.name) || this.onRoster(human.name, state)) { throw new Error( `'${human.name}' is an agent in this session: one name names one participant.`, ); } - const known = this.here.known().get(human.name); - if (known !== undefined && known !== human.identity) { + const known = state.people.get(human.name); + if (known?.presence === 'present' && known.identity !== human.identity) { throw new Error( `'${human.name}' is already in this session under a different identity: one name is one person.`, ); } } - private assertRunning(): void { - if (this.stopped) throw new Error(`Session '${this.name}' is stopped.`); - } - - private assertLive(visit: VisitRuntime): void { - if (visit.gone) throw new Error(`${visit.human.name}'s visit has ended.`); - } - - private async endVisit(visit: VisitRuntime): Promise { - if (visit.gone) return; - visit.gone = true; - this.here.leave(visit.human.name); - await this.commitPresence({ kind: 'left', from: visit.human.name }); - } - private handle(visit: VisitRuntime): Visit { const session = this; return { human: visit.human, get since() { - return session.here.sinceOf(visit.human.name); + return session.state().people.get(visit.human.name)?.since; }, async deliver(input) { - session.assertLive(visit); + if (visit.gone) throw new Error(`${visit.human.name}'s visit has ended.`); + session.assertRunning(); await session.deliverFrom(visit.human.name, input); }, leave() { @@ -621,12 +559,20 @@ class SessionImpl implements Session, RunningRoom { }; } + private async endVisit(visit: VisitRuntime): Promise { + if (visit.gone) return; + visit.gone = true; + this.visits.delete(visit.human.name); + await this.commitPresence({ kind: 'left', from: visit.human.name }); + } + private async deliverFrom( from: string, input: { to?: Participant; text: string; key?: string }, ): Promise { const to = input.to?.name; - if (to !== undefined && !this.here.knows(to) && !this.agents.has(to)) { + const state = this.state(); + if (to !== undefined && !state.people.has(to) && !this.onRoster(to, state)) { throw new Error(`Cannot direct a delivery to '${to}': not in this session.`); } await this.commitMessage(input.key ?? crypto.randomUUID(), undefined, { @@ -638,6 +584,53 @@ class SessionImpl implements Session, RunningRoom { }); } + // -- the roster ------------------------------------------------------------- + + /** The host puts an agent on the roster. From the reserve when it is there; from anywhere else too. */ + async seat(seat: AgentSeat): Promise { + this.assertRunning(); + await this.ready; + const given = unwrap(seat); + const state = this.state(); + if (this.onRoster(given.def.name, state) || state.people.has(given.def.name)) { + throw new Error(`Duplicate agent name '${given.def.name}': one name names one participant.`); + } + // A bare definition takes the attention its reserve entry carried. + const held = state.reserve.find((s) => s.name === given.def.name); + const attention = isSeatedAgent(seat) ? seat.attention : (held?.attention ?? 'broadcast'); + this.know({ def: given.def, attention }); + await this.commitPresence({ + kind: 'seated', + from: given.def.name, + identity: given.def.identity, + attention, + }); + } + + /** The host takes an agent off the roster. Never the assistant. */ + async unseat(agent: AgentDefinition): Promise { + this.assertRunning(); + await this.ready; + if (!this.onRoster(agent.name)) + throw new Error(`'${agent.name}' is not seated in this session.`); + if (this.assistant.is(agent.name)) { + throw new Error(`'${agent.name}' is the assistant: a room cannot run without one.`); + } + this.retire(agent.name); + await this.commitPresence({ kind: 'unseated', from: agent.name }); + } + + /** + * Off the roster: what was mid-flight is cut, and a wake it was sent and + * never claimed is forgotten. The lease it holds ends when the seat side + * releases it, and the roster changes when the `unseated` lands. + */ + private retire(name: string): void { + const port = this.ports.get(name); + if (port instanceof SeatActor) port.abort(); + for (const [id, of] of this.pending) if (of === name) this.pending.delete(id); + } + /** Closes the run: what is mid-flight ends, what is present is marked gone. */ async stop(): Promise { if (this.stopped) return; @@ -650,20 +643,14 @@ class SessionImpl implements Session, RunningRoom { // A deliberate shutdown observed everybody leaving, so the record // says so, and the host hears it. It wakes nobody: an activation // started to hear that the room is closing is an activation nobody reads. - for (const visit of this.here.all()) { - visit.gone = true; - this.here.leave(visit.human.name); - await this.commitUnrouted({ kind: 'left', from: visit.human.name }); - } - // What the run added leaves with it, the same way: the next run begins - // from the composition `startSession` was given. - for (const seat of this.agents.values()) { - if (!seat.added) continue; - this.retire(seat); - await this.commitUnrouted({ kind: 'unseated', from: seat.def.name }); + for (const person of this.state().people.values()) { + if (person.presence !== 'present') continue; + const visit = this.visits.get(person.name); + if (visit) visit.gone = true; + await this.commitPresence({ kind: 'left', from: person.name }, false); } } finally { - // The name comes free whatever the repo did. A failed write must + // The name comes free whatever the storage did. A failed write must // not leave a room that can never be started again. if (this.runtime.running.get(this.name) === this) this.runtime.running.delete(this.name); // A stopped room never goes quiet on its own, so nobody waits on it. @@ -673,85 +660,71 @@ class SessionImpl implements Session, RunningRoom { // -- commits ---------------------------------------------------------------- - /** A presence change the room commits under a fresh key, and routes. */ - private async commitPresence(change: Omit): Promise { - await this.commitMessage(crypto.randomUUID(), undefined, { - ...change, - at: this.now(), - }); - } - - /** A presence change the closing room commits and routes to nobody: an activation nobody reads. */ - private async commitUnrouted(change: Omit): Promise { - await this.commitMessage( - crypto.randomUUID(), - undefined, - { ...change, at: this.now() }, - { route: false }, - ); - } - /** - * One operation on the room's commit queue: the write, then what the room - * changes for a fresh message, then what it does with it, inside the same - * link of the queue. A repeated key lands nothing, so the room does - * nothing with it either. + * One operation on the room's commit queue: the write, and then what the + * room does with a fresh message, inside the same link of the queue. A + * repeated key lands nothing, so the room does nothing with it either. */ private commitMessage( key: string, readThrough: Seq | undefined, draft: Omit, - options: { route?: boolean; landed?: () => void } = {}, + route = true, ) { return this.log.commit( { key, ...(readThrough === undefined ? {} : { readThrough }), draft }, - (message) => { - options.landed?.(); - if (options.route ?? true) this.committed(message); - else this.emit({ type: 'message', message }); - }, + (message) => (route ? this.committed(message) : this.emit({ type: 'message', message })), ); } + /** A presence change the room observed, under a fresh key: the room's own word, never a retry. */ + private commitPresence(change: PresenceDraft, route = true) { + const draft = { ...change, at: this.now() }; + return this.commitMessage(crypto.randomUUID(), undefined, draft, route); + } + /** * What happens to every message once its write is confirmed: the host - * hears about it, and the room routes it. One message, one event, one - * order — stated here rather than at each of the commit sites. + * hears about it, then what it opened, and the room routes it. One + * message, one event, one order — stated here rather than at each of the + * commit sites. */ private committed(message: Message): void { // The message lands, then what it opened: an exchange is a fact about a // message the host has already seen. Both come before the routing, so // nothing wakes on a message the host has not heard about. this.emit({ type: 'message', message }); - this.noteExchange(message); + this.noteExchange(message.seq); this.dispatch(message); // 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.state().exchange !== undefined && !this.working()) void this.settle(); } /** - * A person's question opens an exchange, and the room says so. When the + * The question at `seq` opened an exchange, and the room says so. When the * room holds agents in reserve, the assistant composes the room for it: it * reads the question while the seats do, and seats who the question needs. */ - private noteExchange(message: Message): void { - const opened = this.exchanges.note(message, this.here.knows(message.from)); - if (!opened) return; - this.emit({ type: 'exchange_opened', exchange: opened }); - if (this.reserve.size === 0 || this.stopped) return; + private noteExchange(seq: Seq): void { + const state = this.state(); + if (state.exchange?.from !== seq) return; + this.emit({ type: 'exchange_opened', exchange: state.exchange }); + if (state.reserve.length === 0 || this.stopped) return; const assistant = this.assistant.name; + // One seat, one activation per seq: a wake the question itself sent + // to the assistant, live or ended, is the activation for this seq, and + // the roster stands for the exchange. + const id = activationId(seq, assistant); + const taken = this.live(assistant) || this.spent.has(id); const composing = this.assistant.compose( - opened.owner, - opened.from, - this.reserve.size, - this.live(assistant), + state.exchange.owner, + state.exchange.from, + state.reserve.length, + taken, ); - if (composing) this.activate(assistant, activationId(opened.from, assistant)); + if (composing) this.activate(assistant, id); } // -- routing ---------------------------------------------------------------- @@ -761,7 +734,8 @@ class SessionImpl implements Session, RunningRoom { * the same for what a person said, what a person did, and what a colleague * said. Every colleague still at work hears it as a steer (rule 2). What * wakes an idle seat is the attention it was seated at, against the reach - * of the message (rules 1, 4 and 6, in `wakes`). + * of the message (rules 1, 4 and 6, in `wakes`). The roster the routing + * reads folds the message itself, so a seating wakes the seat it seats. */ private dispatch(message: Message): void { // The author is excluded, and the seat a message names is its target. For @@ -770,25 +744,24 @@ class SessionImpl implements Session, RunningRoom { const author = authorOf(message); const target = targetOf(message); const fromAssistant = author !== undefined && this.assistant.is(author); - for (const seat of this.agents.values()) { - if (seat.def.name !== author) this.route(seat, message, target, fromAssistant); + for (const seat of this.state().roster) { + if (seat.name !== author) this.route(seat, message, target, fromAssistant); } } /** One seat hears one message: steered in while it is live, or woken when it is at rest. */ private route( - seat: Placed, + seat: { name: string; attention: Attention }, message: Message, target: string | undefined, fromAssistant: boolean, ): void { - const name = seat.def.name; - const id = activationId(message.seq, name); - if (this.live(name)) { - if (!this.hearsSteers(name)) return; - this.send(id, name, { seq: message.seq, line: renderLine(message) }); - } else if (wakes({ name, attention: seat.attention }, target, message, fromAssistant)) { - this.activate(name, id); + const id = activationId(message.seq, seat.name); + if (this.live(seat.name)) { + if (!this.hearsSteers(seat.name)) return; + this.send(id, seat.name, { seq: message.seq, line: renderLine(message) }); + } else if (wakes(seat, target, message, fromAssistant)) { + this.activate(seat.name, id); } } @@ -830,115 +803,27 @@ class SessionImpl implements Session, RunningRoom { await this.up(); const held = this.leases.get(id); if (held === undefined) return stale('the lease ended'); - const seat = this.agents.get(held.seat); - if (seat === undefined) return stale('the seat left the roster'); - return { view: this.viewOf(id, seat) }; - } - - /** The view one activation reads: two rendered strings, the model id, and the hand. */ - private viewOf(id: string, seat: Placed): ActivationView { - const { hand, closing, composing } = this.handOf(id, seat); - const speaking = this.speaking(seat, closing, composing); - const room = this.roomView(); - return { - activation: id, - seat: seat.def.name, - model: seat.def.model, - lastSeq: this.log.lastSeq, - systemPrompt: renderSystemPrompt(speaking, room), - context: renderTurnContext(speaking, room), - hand, - ...(closing ? { closing } : {}), - ...(composing ? { composing } : {}), - }; - } - - /** - * What an activation is for: a draft closes an exchange, the assistant - * woken by a question composes the room for it, and every other seat - * speaks. The assistant woken for anything else holds nothing. - */ - private handOf( - id: string, - seat: Placed, - ): { hand: Hand; closing?: ActivationView['closing']; composing?: ActivationView['composing'] } { - if (parseId(id)?.kind === 'draft') { - const draft = this.assistant.closing(); - if (draft === undefined) return { hand: 'none' }; - return { hand: 'summarise', closing: { ...draft } }; + const state = this.state(); + const def = this.defs.get(held.seat); + if (def === undefined || !this.onRoster(held.seat, state)) { + return stale('the seat left the roster'); } - if (!this.assistant.is(seat.def.name)) return { hand: 'say' }; - const composing = this.assistant.composing(); - return composing === undefined - ? { hand: 'none' } - : { hand: 'seat', composing: { ...composing } }; + return { view: viewOf(id, held.seat, def, this.facts(state)) }; } - /** - * What the prose is given of the seat taking this activation. The assistant - * holds every fact here; a seat holds none of them. - */ - private speaking( - seat: Placed, - closing: ActivationView['closing'], - composing: ActivationView['composing'], - ): SeatSpeaking { - const closingView: Closing | undefined = closing && { - ...closing, - preferences: this.assistant.preferencesOf(closing.person), - }; - const composingView: ComposingView | undefined = composing && { - person: composing.person, - from: composing.from, - reserve: this.reserved(), - }; - return { - def: seat.def, - assistant: this.assistant.is(seat.def.name), - closing: closingView, - composing: composingView, - }; - } - - /** The reserve as the assistant reads it: a name and an identity per agent. */ - private reserved(): { name: string; identity: string }[] { - return [...this.reserve.values()].map(({ def }) => ({ - name: def.name, - identity: def.identity, - })); - } - - /** What the prose is given of this room, built fresh for each activation. */ - private roomView(): RoomView { - const open = this.exchanges.current(); + /** What a view is built from: the fold, and what the room holds beside it. */ + private facts(state: RoomState): RoomFacts { return { name: this.name, - goal: this.goal, now: this.runtime.clock.now(), - seats: this.seats(), - people: this.peopleViews(), - record: this.record, - exchange: open && { owner: open.owner, from: open.from }, + assistant: this.assistant.name, + state, + live: this.liveSeats(state), + drafting: this.assistant.closing(), + unseen: (since) => this.log.since(since).length, }; } - /** One entry per person the room knows, with their gap and what they missed. */ - private peopleViews(): PersonView[] { - const views: PersonView[] = []; - for (const [name, identity] of this.here.known()) { - const since = this.here.sinceOf(name); - views.push({ - name, - identity, - presence: this.here.presenceOf(name), - changedAt: this.here.lastChangeAt(name), - since, - unseen: since === undefined ? 0 : this.log.since(since).length, - }); - } - return views; - } - /** * Rule 5 for a say and for a summary: commit under `readThrough`, the seq * the author has read. The queue refuses a commit the record moved past, @@ -968,13 +853,7 @@ class SessionImpl implements Session, RunningRoom { if (error instanceof RefusedError) return { refused: error.message }; throw error; } - // A seating changes the roster where the message lands, before it - // routes: every seat the seating reaches reads a roster that already - // agrees with it, and a repeated key changes nothing twice. - const landed = drafted.kind === 'seated' ? () => this.admit(drafted.from) : undefined; - const committed = await this.commitMessage(commit.key, commit.readThrough, drafted, { - landed, - }); + const committed = await this.commitMessage(commit.key, commit.readThrough, drafted); if ('missed' in committed) { this.emit({ type: 'conflict', author: held.seat, missed: committed.missed }); return { missed: committed.missed }; @@ -993,8 +872,9 @@ class SessionImpl implements Session, RunningRoom { private draft(commit: Commit, seat: string): Drafted { const intent = commit.intent; const at = this.now(); + const state = this.state(); if (intent.kind === 'said') { - this.assertAddressable(seat, intent.to); + assertAddressable(seat, intent.to, state); return { kind: 'said', at, @@ -1013,31 +893,24 @@ class SessionImpl implements Session, RunningRoom { covers: intent.covers, }; } - const held = this.reserve.get(intent.name); + const held = state.reserve.find((s) => s.name === intent.name); if (held === undefined) { - const names = [...this.reserve.keys()]; + const names = state.reserve.map((s) => s.name); throw new RefusedError( `'${intent.name}' is not in the reserve. ` + (names.length ? `Seat one of: ${names.join(', ')}.` : 'The reserve is empty.'), ); } - return { kind: 'seated', at, from: held.def.name, identity: held.def.identity, by: seat }; - } - - private assertAddressable(seat: string, to: string | undefined): void { - if (to === undefined) return; - const target = this.agents.get(to); - if (!this.here.knows(to) && !target) { - throw new RefusedError(`Unknown participant '${to}'. Address someone from the roster.`); - } - if (to === seat) throw new RefusedError('You cannot address yourself.'); - // A seat at the narrow end wakes for nothing said, so addressing it - // would leave a message nobody reads. Say it to the room instead. - if (target?.attention === 'none') { - throw new RefusedError( - `'${to}' wakes for nothing said. Say it to the room, or to somebody else.`, - ); - } + // The roster folds the seating where it lands, before it routes: every + // seat the seating reaches reads a roster that already agrees with it. + return { + kind: 'seated', + at, + from: held.name, + identity: held.identity, + by: seat, + attention: held.attention, + }; } /** @@ -1050,7 +923,7 @@ class SessionImpl implements Session, RunningRoom { if (this.stopped) return stale('the room is gone'); await this.up(); const seat = seatOf(lease.activation, this.assistant.name); - if (seat === undefined || !this.agents.has(seat)) return stale('the seat is not on the roster'); + if (seat === undefined || !this.onRoster(seat)) return stale('the seat is not on the roster'); return this.claim(lease.activation, seat); } @@ -1064,25 +937,26 @@ class SessionImpl implements Session, RunningRoom { ok: { expiry: this.runtime.clock.now() + LEASE_EXPIRY, lastSeq: this.log.lastSeq }, }; if (this.leases.has(id)) return ok; - if (this.spent.has(id)) return stale('the lease ended'); + // A wake the seat side answered is pending no longer, however it is answered. this.pending.delete(id); + if (this.spent.has(id)) return stale('the lease ended'); this.leases.set(id, { seat, spoke: false }); if (parseId(id)?.kind !== 'draft') this.stirred = true; this.emit({ type: 'activation_start', agent: seat }); return ok; } - private release(id: string, reason: EndReason): LeaseResponse { + private async release(id: string, reason: EndReason): Promise { const held = this.leases.get(id); if (held === undefined) return stale('the lease ended'); this.leases.delete(id); this.spent.add(id); - this.ended(held, reason); + await this.ended(held, reason); return { ok: { expiry: this.runtime.clock.now(), lastSeq: this.log.lastSeq } }; } /** The seat stopped: what that closes, and what it frees. */ - private ended(held: Held, reason: EndReason): void { + private async ended(held: Held, reason: EndReason): Promise { const seat = held.seat; const assistant = this.assistant.is(seat); const drafted = assistant && this.assistant.composing() === undefined; @@ -1098,30 +972,79 @@ class SessionImpl implements Session, RunningRoom { // 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(); + if (!drafted && !this.working()) { + await this.settle(); + return; + } + if (assistant) this.draftNext(this.assistant.dueAfterDraft(...this.dueArgs())); + this.markQuiet(); } // -- the assistant ------------------------------------------------------------ - /** The seats stopped: whoever waited hears it, and the exchange closes. */ - private settle(): void { - for (const resolve of this.settledWaiters.splice(0)) resolve(); + /** + * The seats stopped: the exchange open now closes at the record as it + * stands now, the room says it is quiet when it is, and whoever waited + * hears it after both. One settle at a time; a seat woken in the meantime + * keeps whoever waited waiting. The exchange and the range are taken + * here, when the quiet is observed: a question that lands before the + * close is written opens the next exchange, and a settle observed for one + * exchange never closes another. + */ + private settle(): Promise { + const state = this.state(); + const observed = state.exchange && { from: state.exchange.from, through: state.lastSeq }; + this.settling = this.settling.then(() => this.settleOnce(observed)).catch(() => {}); + return this.settling; + } + + private async settleOnce(observed: Observed | undefined): Promise { const worked = this.stirred; this.stirred = false; - this.closeExchange(worked); + try { + await this.closeExchange(observed, worked); + } finally { + // A close the storage refused leaves the exchange open, and the next + // settle writes it again. The room still says what holds now, and + // nobody waiting on it waits for a write that failed. + this.markQuiet(); + if (!this.working()) for (const resolve of this.settledWaiters.splice(0)) resolve(); + } } /** - * The room went quiet, so the exchange it was working on is over. The host - * hears that before anything is written about it: the assistant is the first + * The room went quiet on `observed`, so that exchange ends at the record + * as it stood: the close is a row on the log, written where the fold still + * says the same exchange is open. A stopped room writes no close; the next + * run over the log closes the exchange at its start. The host hears the + * close 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.log.lastSeq); + private async closeExchange(observed: Observed | undefined, worked: boolean): Promise { + let closing: ClosedExchange | undefined; + await this.log.write('close', () => { + const exchange = this.state().exchange; + if (observed === undefined || exchange?.from !== observed.from || this.stopped) { + return undefined; + } + closing = { ...exchange, through: observed.through }; + return { + owner: exchange.owner, + from: exchange.from, + through: observed.through, + at: this.now(), + }; + }); if (closing) this.emit({ type: 'exchange_closed', exchange: closing }); + // A question that landed after the quiet was observed opened the next + // exchange the moment this one closed, and the room says so now, before + // the assistant is woken for what the closed one owes. The same check + // a commit runs follows: an exchange nobody works on closes at once. + const next = this.state().exchange; + if (closing && next !== undefined) { + this.noteExchange(next.from); + if (!this.working()) void this.settle(); + } this.summariseClosed(closing, worked); } @@ -1170,7 +1093,7 @@ class SessionImpl implements Session, RunningRoom { * the close still counts. */ private speaksForItself(name: string): boolean { - return !this.here.knows(name) && !this.assistant.is(name); + return !this.state().people.has(name) && !this.assistant.is(name); } /** @@ -1189,8 +1112,64 @@ class SessionImpl implements Session, RunningRoom { } } +/** The composition `startSession` was given, checked for duplicates the way the room refuses them. */ +function composeFrom(options: StartSessionOptions): Composition { + const assistant = assertAssistant(options.assistant); + const names = new Set(); + const take = (placed: Placed): Placed => { + if (names.has(placed.def.name)) { + throw new Error(`Duplicate agent name '${placed.def.name}': one name names one participant.`); + } + names.add(placed.def.name); + return placed; + }; + const agents = (options.agents ?? []).map((seat) => take(unwrap(seat))); + take({ def: assistant, attention: 'none' }); + const available = (options.available ?? []).map((seat) => take(unwrap(seat))); + return { assistant, goal: options.goal?.trim() || undefined, agents, available }; +} + +function unwrap(seat: AgentSeat): Placed { + const def = isSeatedAgent(seat) ? seat.agent : seat; + if (!isAgent(def)) throw new Error('Agents must come from defineAgent or seated().'); + return { def, attention: isSeatedAgent(seat) ? seat.attention : 'broadcast' }; +} + +const seatRow = (placed: Placed): SeatRow => ({ + name: placed.def.name, + identity: placed.def.identity, + attention: placed.attention, +}); + +/** The composition as the row the log holds: every seat by name, identity and attention. */ +function compositionRow(composition: Composition, at: string): Without { + return { + assistant: seatRow({ def: composition.assistant, attention: 'none' }), + ...(composition.goal === undefined ? {} : { goal: composition.goal }), + agents: composition.agents.map(seatRow), + available: composition.available.map(seatRow), + at, + }; +} + /** The seat a message names: a directed say names who it addresses, a seating names who it seats. */ function targetOf(message: Message): string | undefined { if (isSpoken(message)) return message.to; return message.kind === 'seated' ? message.from : undefined; } + +function assertAddressable(seat: string, to: string | undefined, state: RoomState): void { + if (to === undefined) return; + const target = state.roster.find((s) => s.name === to); + if (!state.people.has(to) && target === undefined) { + throw new RefusedError(`Unknown participant '${to}'. Address someone from the roster.`); + } + if (to === seat) throw new RefusedError('You cannot address yourself.'); + // A seat at the narrow end wakes for nothing said, so addressing it + // would leave a message nobody reads. Say it to the room instead. + if (target?.attention === 'none') { + throw new RefusedError( + `'${to}' wakes for nothing said. Say it to the room, or to somebody else.`, + ); + } +} diff --git a/packages/ambion/src/types.ts b/packages/ambion/src/types.ts index ed73955..58f3602 100644 --- a/packages/ambion/src/types.ts +++ b/packages/ambion/src/types.ts @@ -12,11 +12,26 @@ import type { } from '@earendil-works/pi-agent-core'; import type { Api, Model } from '@earendil-works/pi-ai'; import type { Static, TSchema } from 'typebox'; -import type { ClosedExchange, Exchange } from './exchange.ts'; /** A position on the record: monotonic, assigned at commit, never reused. */ export type Seq = number; +/** A question the room is working on. */ +export interface Exchange { + /** The person whose question opened it, and who owns what follows. */ + readonly owner: string; + /** The seq of that question: where the exchange starts. */ + readonly from: Seq; + /** When it opened, ISO. */ + readonly at: string; +} + +/** An exchange the room has finished, and the range it turned out to hold. */ +export interface ClosedExchange extends Exchange { + /** The last seq on the record when the room went quiet. */ + readonly through: Seq; +} + // -- what a host provides ----------------------------------------------------- /** The one clock a room reads, and the one alarm it sets. */ @@ -84,6 +99,10 @@ export interface PresenceMessage { * is the seat it names. */ by?: string; + /** What wakes the seat, on `seated`. Absent means `broadcast`. */ + attention?: Attention; + /** How the person reads, on `arrived`, when they said so. */ + preferences?: string; } /** diff --git a/packages/ambion/src/wire.ts b/packages/ambion/src/wire.ts index 308a4ba..b29d5d0 100644 --- a/packages/ambion/src/wire.ts +++ b/packages/ambion/src/wire.ts @@ -1,5 +1,6 @@ /** - * What crosses between a seat and its room. Every shape here is plain JSON: + * What crosses between a seat and its room, and what the log holds beside + * a message. Every shape here is plain JSON: * an optional key is written only when it is present, and no value is * `undefined`, a `Date`, a `Map`, a `Set`, a class instance or a function. A * request and its response survive a round trip through `JSON.stringify` @@ -11,7 +12,42 @@ * 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. */ -import type { Message, Seq } from './types.ts'; +import type { Attention, Message, Seq } from './types.ts'; + +// -- rows on the log beside the messages -------------------------------------- + +/** `Omit` over each member of a union, so a discriminated row keeps its shape. */ +export type Without = T extends unknown ? Omit : never; + +/** The room went quiet with an exchange open, and closed it. */ +export interface CloseRow { + owner: string; + from: Seq; + through: Seq; + after: Seq; + at: string; +} + +/** One seat in a composition: its name, how the room knows it, and what wakes it. */ +export interface SeatRow { + name: string; + identity: string; + attention: Attention; +} + +/** + * What a run started with. The roster folds from the latest one, and a + * reader without the definitions reads every identity off it. + */ +export interface CompositionRow { + /** The assistant's seat. Its attention is `none`. */ + assistant: SeatRow; + goal?: string; + agents: SeatRow[]; + available: SeatRow[]; + after: Seq; + at: string; +} /** Why a lease ended: the activation ran to its end, it never reached the record, or the record kept moving past its drafts. */ export type EndReason = 'released' | 'failed' | 'refused'; diff --git a/packages/ambion/test/assistant.test.ts b/packages/ambion/test/assistant.test.ts index 6cd1bc3..0f5178d 100644 --- a/packages/ambion/test/assistant.test.ts +++ b/packages/ambion/test/assistant.test.ts @@ -4,12 +4,15 @@ import { Type } from 'typebox'; import { afterEach, describe, expect, it } from 'vitest'; import { attentive, + createRuntime, defineAgent, defineHuman, defineTool, InMemorySessionRepo, isSpoken, type Message, + passive, + type Runtime, type Session, type SessionEvent, type SummaryMessage, @@ -25,9 +28,11 @@ import { quiet, type Script, scripted, + seat, speak, summarise, } from './support/scripted.ts'; +import { gatedOpener, memory } from './support/storage.ts'; /** The ordinary assistant: it writes once, then ends its activation. */ const writes = @@ -93,16 +98,20 @@ const started: Session[] = []; function open(options: { script: Script; agents?: Parameters[0]['agents']; + available?: Parameters[0]['available']; assistant?: Parameters[0]['assistant']; repo?: SessionRepo; + runtime?: Runtime; }): Session { const session = startSession({ name: roomName(), goal: 'Decide the pour date and keep the plan honest.', assistant: options.assistant ?? assistant, agents: options.agents ?? [product], + ...(options.available ? { available: options.available } : {}), streamFn: scripted(options.script), ...(options.repo ? { repo: options.repo } : {}), + ...(options.runtime ? { runtime: options.runtime } : {}), }); started.push(session); return session; @@ -833,6 +842,187 @@ describe('an exchange', () => { expect(events.filter((e) => e.type === 'exchange_closed')).toHaveLength(1); }); + it('closes at the quiet it observed, and a question that lands before the row opens the next', async () => { + const gate = deferred(); + const hold = (_type: string, data: unknown) => + (data as { text?: string }).text === 'second?' ? gate.promise : undefined; + const runtime = createRuntime({ sessions: gatedOpener((await memory.open()).sessions, hold) }); + const working = deferred(); + const session = open({ + runtime, + script: byAgent({ + product: async (_context, _name, call) => { + if (call !== 1) return quiet(); + await working.promise; + return quiet(); + }, + }), + }); + const events = collect(session); + const visit = await visitSession(session, priya); + await visit.deliver({ text: 'first?' }); + // the second question is on the queue and not yet on the record when the seat stops + const second = visit.deliver({ text: 'second?' }); + const stopped = new Promise((resolve) => { + const off = session.subscribe((event) => { + if (event.type !== 'activation_end') return; + off(); + resolve(); + }); + }); + working.resolve(); + await stopped; + gate.resolve(); + await second; + await quiescent(session); + + const record = await session.messages(); + const [first, next] = record.filter(isSpoken); + const opened = events.filter((e) => e.type === 'exchange_opened'); + const closed = events.filter((e) => e.type === 'exchange_closed'); + // the first exchange holds what the room had when it went quiet + expect(closed.map((e) => e.type === 'exchange_closed' && e.exchange)).toEqual([ + { owner: 'priya', from: first?.seq, at: first?.at, through: first?.seq }, + { owner: 'priya', from: next?.seq, at: next?.at, through: record.at(-1)?.seq }, + ]); + // and the second question opened its own, announced once the first closed + expect(opened.map((e) => e.type === 'exchange_opened' && e.exchange.from)).toEqual([ + first?.seq, + next?.seq, + ]); + const order = events.map((e) => e.type); + expect(order.lastIndexOf('exchange_opened')).toBeGreaterThan(order.indexOf('exchange_closed')); + }); + + /** Resolves when the named seat's next activation ends. */ + const activationEnded = (session: Session, agent: string) => + new Promise((resolve) => { + const off = session.subscribe((event) => { + if (event.type !== 'activation_end' || event.agent !== agent) return; + off(); + resolve(); + }); + }); + + /** A room over a storage that holds the writes the test names. */ + const gated = async ( + held: (type: string, data: { text?: string }) => Promise | undefined, + ) => + createRuntime({ + sessions: gatedOpener((await memory.open()).sessions, (type, data) => + held(type, data as { text?: string }), + ), + }); + + const surveyor = defineAgent({ + name: 'surveyor', + identity: 'Quantity surveyor.', + instructions: 'x', + model: 'scripted/surveyor', + }); + + const ranges = (events: SessionEvent[]) => + events.flatMap((e) => + e.type === 'exchange_closed' ? [[e.exchange.from, e.exchange.through]] : [], + ); + const openings = (events: SessionEvent[]) => + events.flatMap((e) => (e.type === 'exchange_opened' ? [e.exchange.from] : [])); + + it('never closes the next exchange on a quiet it observed for the last one', async () => { + const gate = deferred(); + const working = deferred(); + const session = open({ + runtime: await gated((_type, data) => (data.text === 'second?' ? gate.promise : undefined)), + agents: [passive(product)], + available: [surveyor], + script: byAgent({ + product: async () => { + await working.promise; + return quiet(); + }, + // composes twice: nobody for the first question, the surveyor for the second + assistant: (_context, _name, call) => (call === 2 ? seat('surveyor') : quiet()), + surveyor: (_context, _name, call) => (call === 1 ? speak('11.7 tonnes.') : quiet()), + }), + }); + const events = collect(session); + const visit = await visitSession(session, priya); + await visit.deliver({ to: product, text: 'first?' }); + await activationEnded(session, 'assistant'); + // the second question wakes nobody, and it is still on the queue when the product stops + const second = visit.deliver({ text: 'second?' }); + const stopped = activationEnded(session, 'product'); + working.resolve(); + await stopped; + gate.resolve(); + await second; + await quiescent(session); + + const record = await session.messages(); + const [first, next] = record.filter(isSpoken); + const seated = record.find((m) => m.kind === 'seated'); + // the first exchange closed where its quiet was observed; the second held the composing + // activation, the seating and what the newcomer said, and closed when they stopped + expect(ranges(events)).toEqual([ + [first?.seq, first?.seq], + [next?.seq, record.at(-1)?.seq], + ]); + expect(openings(events)).toEqual([first?.seq, next?.seq]); + expect(seated).toMatchObject({ from: 'surveyor', by: 'assistant' }); + expect(seated?.seq).toBeGreaterThan(next?.seq ?? 0); + expect(session.exchange()).toBeUndefined(); + }); + + it('composes nothing for a question the assistant already woke on, and closes it at once', async () => { + const hey = deferred(); + const close = deferred(); + let closes = 0; + const working = deferred(); + const session = open({ + runtime: await gated((type, data) => { + if (data.text === 'hey') return hey.promise; + if (type === 'ambion/close' && ++closes === 1) return close.promise; + return undefined; + }), + agents: [passive(product)], + available: [surveyor], + script: byAgent({ + product: async () => { + await working.promise; + return quiet(); + }, + }), + }); + const events = collect(session); + const visit = await visitSession(session, priya); + await visit.deliver({ to: product, text: 'first?' }); + await activationEnded(session, 'assistant'); + // a word to the assistant lands after the quiet was observed, and before the close is written + const said = visit.deliver({ to: assistant, text: 'hey' }); + const stopped = activationEnded(session, 'product'); + working.resolve(); + await stopped; + hey.resolve(); + await said; + // the assistant woke on that word and ended before the close landed + await activationEnded(session, 'assistant'); + close.resolve(); + await quiescent(session); + + const record = await session.messages(); + const [first, next] = record.filter(isSpoken); + // the word opened its own exchange; the assistant had its one activation for it, so the + // roster stood, nobody worked, and the exchange closed holding the word alone + expect(openings(events)).toEqual([first?.seq, next?.seq]); + expect(ranges(events)).toEqual([ + [first?.seq, first?.seq], + [next?.seq, next?.seq], + ]); + expect(record.some((m) => m.kind === 'seated')).toBe(false); + expect(session.seats().find((s) => s.name === 'assistant')).toMatchObject({ status: 'idle' }); + expect(session.exchange()).toBeUndefined(); + }); + it('closes before the summary that stands for it', async () => { const session = open({ script: byAgent({ product: twoAnswers, assistant: writes('Thursday is out.') }), diff --git a/packages/ambion/test/presence.test.ts b/packages/ambion/test/presence.test.ts index 2d83e81..e4e21a0 100644 --- a/packages/ambion/test/presence.test.ts +++ b/packages/ambion/test/presence.test.ts @@ -279,9 +279,11 @@ describe('presence', () => { const view = readSession(name, { repo }); expect((await view.messages()).filter(isSpoken).map((m) => m.text)).toEqual(['for later']); - // no agents stand up, and everybody the record knows is absent - expect(view.seats()).toEqual([ - { kind: 'human', name: 'andrei', identity: andrei.identity, presence: 'absent' }, + // the roster folds from the record, nothing stands up, and everybody the record knows is absent + expect(view.seats().map((s) => [s.name, s.kind === 'agent' ? s.status : s.presence])).toEqual([ + ['watcher', 'idle'], + ['assistant', 'idle'], + ['andrei', 'absent'], ]); }); @@ -370,6 +372,35 @@ describe('a storage that fails', () => { await stopSession(session); }); + it('answers whoever waits when the close itself cannot be written, and closes at the next settle', async () => { + const { session, fail } = await brittle(); + const events = collect(session); + const visit = await visitSession(session, andrei); + await session.settled(); + await visit.deliver({ text: 'first?' }); + // the seat is woken; the host waits for the room to be quiet + const waiting = session.quiet(); + // the close is the next write, and it fails + fail(true); + await expect(waiting).resolves.toBeUndefined(); + expect(events.map((e) => e.type)).not.toContain('exchange_closed'); + expect(events.map((e) => e.type)).toContain('quiet'); + expect(session.exchange()).toMatchObject({ owner: 'andrei' }); + + // the storage mends, the seats work and stop again, and the close is written then + fail(false); + await visit.deliver({ text: 'still there?' }); + await session.quiet(); + expect(session.exchange()).toBeUndefined(); + const closed = events.filter((e) => e.type === 'exchange_closed'); + expect(closed).toHaveLength(1); + const record = await session.messages(); + expect(closed[0]).toMatchObject({ + exchange: { owner: 'andrei', from: record[1]?.seq, through: record.at(-1)?.seq }, + }); + await stopSession(session); + }); + it('frees the name when the shutdown itself cannot write', async () => { const { session, fail } = await brittle(); await visitSession(session, andrei); diff --git a/packages/ambion/test/restart.test.ts b/packages/ambion/test/restart.test.ts new file mode 100644 index 0000000..758eb51 --- /dev/null +++ b/packages/ambion/test/restart.test.ts @@ -0,0 +1,197 @@ +/** + * What a run wrote about itself outlives it. The composition is a row on + * the log, so a stopped room says who was in it, and the next run over the + * same log starts from its own row. + */ +import { describe, expect, it } from 'vitest'; +import { + createRuntime, + defineAgent, + defineHuman, + type Runtime, + readSession, + type Session, + startSession, + stopSession, + visitSession, +} from '../src/index.ts'; +import { type FakeClock, fakeClock } from './support/clock.ts'; +import { collect, deferred, roomName, rowsOf } from './support/room.ts'; +import { byAgent, quiet, scripted } from './support/scripted.ts'; +import { type OpenedStorage, storages } from './support/storage.ts'; + +const assistant = defineAgent({ + name: 'assistant', + identity: 'Writes the one message a person reads.', + instructions: 'Answer what was asked, once.', + model: 'scripted/assistant', +}); +const alpha = defineAgent({ + name: 'alpha', + identity: 'Alpha.', + instructions: 'x', + model: 'scripted/alpha', +}); +const beta = defineAgent({ + name: 'beta', + identity: 'Beta.', + instructions: 'x', + model: 'scripted/beta', +}); +const priya = defineHuman({ name: 'priya', identity: 'Project manager.' }); + +/** Resolves when the named seat's activation has ended, so its lease is released. */ +function ended(session: Session, agent: string): Promise { + return new Promise((resolve) => { + const off = session.subscribe((event) => { + if (event.type !== 'activation_end' || event.agent !== agent) return; + off(); + resolve(); + }); + }); +} + +interface World { + opened: OpenedStorage; + clock: FakeClock; + /** A runtime over the storage. Each call is a new host over the same log. */ + runtime(): Runtime; +} + +async function world(storage: (typeof storages)[number]): Promise { + const opened = await storage.open(); + const clock = fakeClock(); + return { + opened, + clock, + runtime: () => createRuntime({ sessions: opened.sessions, clock }), + }; +} + +describe.each(storages)('a room on $name', (storage) => { + it('writes one composition per run, and the latest roster wins', async () => { + const { opened, runtime } = await world(storage); + try { + const name = roomName(`restart-${storage.name}`); + const one = startSession({ + name, + assistant, + agents: [alpha], + runtime: runtime(), + streamFn: scripted(byAgent({})), + }); + await one.messages(); + expect(one.seats().map((s) => s.name)).toEqual(['alpha', 'assistant']); + await stopSession(one); + + const two = startSession({ + name, + assistant, + agents: [beta], + runtime: runtime(), + streamFn: scripted(byAgent({})), + }); + await two.messages(); + expect(two.seats().map((s) => s.name)).toEqual(['beta', 'assistant']); + await stopSession(two); + + const view = readSession(name, { runtime: runtime() }); + await view.messages(); + expect(view.seats().map((s) => s.name)).toEqual(['beta', 'assistant']); + } finally { + await opened.dispose(); + } + }); + + it('reads every identity off the log, in a process that holds no definition', async () => { + const { opened, runtime } = await world(storage); + try { + const name = roomName(`restart-identity-${storage.name}`); + const session = startSession({ + name, + assistant, + agents: [alpha], + available: [beta], + runtime: runtime(), + streamFn: scripted(byAgent({})), + }); + // before the replay, the seats fold from the row the run is about to write + expect(session.seats().map((s) => [s.name, s.identity])).toEqual([ + ['alpha', 'Alpha.'], + ['assistant', assistant.identity], + ]); + await session.messages(); + await session.seat(beta); + await session.quiet(); + await stopSession(session); + + // a fresh runtime knows no definition: the composition row and the seating carry them + const view = readSession(name, { runtime: createRuntime({ sessions: opened.sessions }) }); + await view.messages(); + expect(view.seats().map((s) => [s.name, s.identity])).toEqual([ + ['alpha', 'Alpha.'], + ['assistant', assistant.identity], + ['beta', 'Beta.'], + ]); + } finally { + await opened.dispose(); + } + }); + + it('leaves an open exchange to the next run, which closes it before it answers', async () => { + const { opened, runtime } = await world(storage); + try { + const name = roomName(`restart-exchange-${storage.name}`); + const working = deferred(); + const one = startSession({ + name, + assistant, + agents: [alpha], + runtime: runtime(), + streamFn: scripted( + byAgent({ + alpha: async () => { + working.resolve(); + await new Promise(() => {}); + return quiet(); + }, + }), + ), + }); + const heard = collect(one); + const visit = await visitSession(one, priya); + await visit.deliver({ text: 'Is anybody there?' }); + await working.promise; + // a stop mid-exchange: the activation is cut, and its release closes nothing + const released = ended(one, 'alpha'); + await stopSession(one); + await released; + expect(heard.map((e) => e.type)).not.toContain('exchange_closed'); + const before = await rowsOf(opened.sessions, name); + expect(before.map((row) => row.type)).not.toContain('ambion/close'); + + // the next run closes it as it starts, and quiet() waits for that close + const two = startSession({ + name, + assistant, + agents: [alpha], + runtime: runtime(), + streamFn: scripted(byAgent({})), + }); + const events = collect(two); + await two.quiet(); + // the close is on the stream when quiet() answers, before any other call replays the log + expect(events.map((e) => e.type)).toContain('exchange_closed'); + expect(two.exchange()).toBeUndefined(); + const question = (await two.messages()).find((m) => m.kind === 'said'); + const closes = (await rowsOf(opened.sessions, name)).filter( + (row) => row.type === 'ambion/close', + ); + expect(closes).toHaveLength(1); + expect(closes[0]?.data).toMatchObject({ owner: 'priya', from: question?.seq }); + await stopSession(two); + } finally { + await opened.dispose(); + } + }); +}); diff --git a/packages/ambion/test/roster.test.ts b/packages/ambion/test/roster.test.ts index 390dce9..889ed83 100644 --- a/packages/ambion/test/roster.test.ts +++ b/packages/ambion/test/roster.test.ts @@ -574,7 +574,7 @@ describe('the host', () => { ); }); - it('unseats what the run added at stop, and leaves the starting composition alone', async () => { + it('leaves the roster to the next composition at stop, and the next run starts from its own', async () => { const session = open({ script: byAgent({}), agents: [product], available: [surveyor] }); await session.seat(surveyor); await session.quiet(); @@ -582,10 +582,23 @@ describe('the host', () => { await stopSession(session); started.pop(); + // the record says who was seated, and a read of the stopped room folds it const { readSession } = await import('../src/index.ts'); - const record = await readSession(session.name).messages(); - expect(kinds(record)).toEqual(['seated', 'unseated']); - expect(record.at(-1)).toMatchObject({ kind: 'unseated', from: 'surveyor' }); + const stopped = readSession(session.name); + expect(kinds(await stopped.messages())).toEqual(['seated']); + expect(seatNames(stopped as Session)).toEqual(['product', 'assistant', 'surveyor']); + + // the next run writes its own composition, and the roster folds from that + const again = startSession({ + name: session.name, + assistant, + agents: [product], + available: [surveyor], + streamFn: scripted(byAgent({})), + }); + started.push(again); + await again.messages(); + expect(seatNames(again)).toEqual(['product', 'assistant']); }); }); diff --git a/packages/ambion/test/session.test.ts b/packages/ambion/test/session.test.ts index 18c9dca..b405b91 100644 --- a/packages/ambion/test/session.test.ts +++ b/packages/ambion/test/session.test.ts @@ -281,10 +281,13 @@ describe('startSession', () => { expect(new Set(seqs).size).toBe(seqs.length); await stopSession(again); - // you can read a room that is not running + // you can read a room that is not running: the record, and the roster it folds const view = readSession(name); expect(spoken(await view.messages()).map((m) => m.text)).toContain('for the record'); - expect(view.seats().every((seat) => seat.kind === 'human')).toBe(true); + expect(view.seats().map((seat) => seat.name)).toEqual(['scribe', 'assistant', 'andrei']); + expect(view.seats().every((seat) => seat.kind === 'human' || seat.status === 'idle')).toBe( + true, + ); const fresh = startSession({ name: roomName('identity'), diff --git a/packages/ambion/test/support/room.ts b/packages/ambion/test/support/room.ts index edff255..e0b4434 100644 --- a/packages/ambion/test/support/room.ts +++ b/packages/ambion/test/support/room.ts @@ -1,8 +1,10 @@ +import type { Session as PiSession } from '@earendil-works/pi-agent-core'; import { defineAgent, defineHuman, type Session, type SessionEvent, + type SessionOpener, visitSession, } from '../../src/index.ts'; @@ -48,6 +50,19 @@ export function deferred(): { promise: Promise; resolve: () => void } { export const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); +/** Every row the room wrote beside its messages, read off Pi's session directly. */ +export async function rowsOf( + sessions: SessionOpener, + name: string, +): Promise<{ type: string; data: unknown }[]> { + const piSession: PiSession = await sessions.open(name); + const entries = await piSession.findEntries(); + entries.sort((a, b) => a.seq - b.seq); + return entries.flatMap((entry) => + entry.type === 'custom' ? [{ type: entry.customType, data: entry.data }] : [], + ); +} + /** * The assistant's activation is over, whatever it decided. A summary commits * inside the tool call, so the activation runs on for a moment after the diff --git a/packages/ambion/test/support/storage.ts b/packages/ambion/test/support/storage.ts index 159679c..a1a77f2 100644 --- a/packages/ambion/test/support/storage.ts +++ b/packages/ambion/test/support/storage.ts @@ -133,3 +133,29 @@ export function faultyOpener(sessions: SessionOpener): FaultyOpener { }, }; } + +// -- a storage that holds a write --------------------------------------------- + +/** An opener whose sessions hold one write until the test lets it land. */ +export function gatedOpener( + sessions: SessionOpener, + held: (customType: string, data: unknown) => Promise | undefined, +): SessionOpener { + const gated = (piSession: PiSession): PiSession => + new Proxy(piSession, { + get(target, property, receiver) { + if (property === 'appendCustomEntry') { + return async (customType: string, data: unknown) => { + await held(customType, data); + return (Reflect.get(target, property, receiver) as (...a: unknown[]) => unknown).apply( + target, + [customType, data], + ); + }; + } + const value = Reflect.get(target, property, receiver); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + return { open: async (id, parentId) => gated(await sessions.open(id, parentId)) }; +} diff --git a/packages/ambion/test/wire.test.ts b/packages/ambion/test/wire.test.ts index 62694a4..8f3aa60 100644 --- a/packages/ambion/test/wire.test.ts +++ b/packages/ambion/test/wire.test.ts @@ -1,22 +1,41 @@ /** - * Everything that crosses between a seat and its room is plain JSON: it - * survives the wire unchanged. + * Everything that crosses between a seat and its room, and every row on the + * log, is plain JSON: it survives the wire unchanged. */ import { describe, expect, it } from 'vitest'; import { type ActivationView, assertWire, + type CloseRow, type Commit, type CommitResponse, + type CompositionRow, + createRuntime, type Lease, type LeaseResponse, roundTrip, type ViewResponse, type Wake, } from '../src/index.ts'; +import { fakeClock } from './support/clock.ts'; +import { roomName, rowsOf } from './support/room.ts'; +import { oneExchange } from './support/scenarios.ts'; +import { jsonl } from './support/storage.ts'; const at = '2026-01-01T09:00:00.000Z'; +const rows: Record = { + close: { owner: 'priya', from: 2, through: 4, after: 4, at }, + composition: { + assistant: { name: 'assistant', identity: 'Writes the one message.', attention: 'none' }, + goal: 'Decide the pour date.', + agents: [{ name: 'product', identity: 'The product.', attention: 'broadcast' }], + available: [{ name: 'surveyor', identity: 'Holds the tonnage.', attention: 'named' }], + after: 0, + at, + }, +}; + const wake: Wake = { room: 'site', seat: 'product', @@ -80,7 +99,7 @@ const responses: Record = }; describe('the wire', () => { - it.each(Object.entries({ wake, ...requests, ...responses }))( + it.each(Object.entries({ ...rows, wake, ...requests, ...responses }))( 'carries %s unchanged', (_name, value) => { expect(() => assertWire(value)).not.toThrow(); @@ -96,4 +115,22 @@ describe('the wire', () => { expect(() => assertWire({ fire: () => {} })).toThrow(/is a function/); expect(() => assertWire({ error: new Error('boom') })).toThrow(/is a Error/); }); + + it('replays a JSONL log whose every row is plain JSON', async () => { + const opened = await jsonl.open(); + try { + const runtime = createRuntime({ sessions: opened.sessions, clock: fakeClock() }); + const name = roomName('wire-jsonl'); + await oneExchange.run({ runtime, name }); + const written = await rowsOf(opened.sessions, name); + expect(written.map((row) => row.type)).toContain('ambion/close'); + expect(written.map((row) => row.type)).toContain('ambion/composition'); + for (const row of written) { + expect(() => assertWire(row.data)).not.toThrow(); + expect(roundTrip(row.data)).toStrictEqual(row.data); + } + } finally { + await opened.dispose(); + } + }); }); diff --git a/planning/backlog.md b/planning/backlog.md index a4c2737..de752de 100644 --- a/planning/backlog.md +++ b/planning/backlog.md @@ -24,23 +24,22 @@ nothing. `test/runtime.test.ts` proves two runtimes never see each other. ### 2. Nothing bounds the record, and the room rescans it per message -**What.** `Attendance.known()` rebuilds a map from the whole record on -every call. The room calls it on each dispatch, on each `seats()`, and per -person in `peopleViews()`, which adds two more linear scans per person. -Every activation renders the whole record into the prompt through +**What.** `foldRoom` folds the whole log again after every entry: the +people, the roster and the open exchange are each a pass over every +message. Every activation renders the whole record into the prompt through `renderRecord`. **Why.** Cost is O(n) per message and O(n²) over a run. Context grows without limit. `docs/agent.md` §8 says Ambion owns no context window, and `docs/assistant.md` §16 forbids a compactor, so today nothing owns it. -**Where.** `packages/ambion/src/presence.ts`, `known()` and -`lastChangeAt()`; `packages/ambion/src/session.ts`, nine call sites; +**Where.** `packages/ambion/src/room/fold.ts`, `foldRoom`; +`packages/ambion/src/room/presence.ts`, `foldPeople`; `packages/ambion/src/render.ts`, `renderRecord`. -**Fix.** Short term: `Attendance` keeps an incremental index that updates -on append. Long term: a window policy on `RoomView.record`, and a decision -in the contract about which module owns it. +**Fix.** Short term: the fold keeps an index it advances per entry. Long +term: a window policy on `RoomView.record`, and a decision in the +contract about which module owns it. ### 3. `session.ts` holds four jobs @@ -321,13 +320,11 @@ both exchanges. Nothing pins that behaviour; the tests cover the failure and the retry separately. See [`docs/assistant.md`](../docs/assistant.md) §5. -### 19. Exchanges are run state +### 19. Exchanges are run state — closed -`Exchanges` holds the open exchange in memory, so a restart begins with none — -right for a room mid-question, and a limit for anything that wants to work -over past exchanges. A closed exchange is an owner and a range, so it is -derivable from the record; nothing derives it today. See -[`docs/exchange.md`](../docs/exchange.md) §5. +An exchange is a fold over the log: the open one is the first question +after the last close row, and every close is a row beside the messages. +See [`docs/exchange.md`](../docs/exchange.md) §5. ### 20. A second non-seat writer @@ -500,3 +497,23 @@ collaboration patterns people and agents work in. **Where.** `seated` in [`define.ts`](../packages/ambion/src/define.ts), the roster in [`render.ts`](../packages/ambion/src/render.ts). + +### 26. The catalog is keyed by bare name, per runtime + +**What.** The seat side resolves a definition by name through the +runtime's catalog (`seat.ts`). Every room a runtime holds writes its +definitions into that one map, so two rooms in one runtime that define the +same name differently share one entry, and the last room to start wins. +The log side no longer has this gap: the composition row and every seating +carry the identity, so a read reports what the run held. + +**Why deferred.** The room already refuses a duplicate name inside one +roster. Two rooms in one runtime with one name and two definitions is a +host that wants two runtimes. The catalog exists so that a transport can +hand a seat in another process the definition it needs by name. + +**Options.** The runtime refuses a second, different definition under a +name it holds. Or the in-process transport hands the actor the room's own +definitions, and the catalog serves the out-of-process case alone. Either +way, a definition digest on the composition row and the claim lets a seat +tell that it runs the definition the room seated. diff --git a/planning/next.md b/planning/next.md index 80aad68..d1613fb 100644 --- a/planning/next.md +++ b/planning/next.md @@ -28,9 +28,9 @@ same name and never see each other. that runs unattended for days must not scan its whole record per message or render all of it per activation. -**Done when.** `Attendance` updates an index on append and `known()` is -O(1). `RoomView.record` takes a window policy, and the design contract -names the module that owns it. +**Done when.** The fold advances an index per entry, and `foldRoom` is +O(1) per message. `RoomView.record` takes a window policy, and the design +contract names the module that owns it. ## 4. Load the provider registry on demand (backlog 4)