/** * The AG-UI event emitter, the one part of the event plane that TALKS: it reads a durable source * forward from the WAL's cursor, maps records to events, packs them into frames that provably fit, * and appends them to the principal's event channel under an optimistic-concurrency expectation * with a frozen dedup id. The vocabulary it speaks is `agui.ts`, which does no I/O and does not * import this module. * * The emitter owns no policy about WHAT an event means. That belongs to the per-connector mapping, * and it arrives here as an injected function. What it owns is the property none of the pieces hold * alone: that a frame is either on the wire and folded into the frontier, or not on the wire and * not folded, and that no third state is ever reported as success. */ import { type Part } from "@cotal-ai/core"; import { AguiVocabularyError, type AguiEvent, type AguiFrame, type CotalMeta } from "./agui.js"; import type { DurableSource, SourceGap } from "./durable-source.js"; import type { EventWal } from "./event-wal.js"; import type { SubjectFrontier } from "./subject-frontier.js"; /** * The endpoint surface the emitter needs, declared STRUCTURALLY rather than as `CotalEndpoint`. * * Not for testability as an end in itself — for a specific one. A cell that needs a live broker to * exercise the duplicate-ack halt cannot be written at all before a broker exists, and a cell that * re-implements `encodedSize` is measuring a copy. This interface is the exact set of methods the * emitter calls, so a cell substitutes an instrument and the production path substitutes the real * endpoint, and neither one is a re-implementation of the other. */ export interface EmitterEndpoint { readonly principal: { owner: string; actor: string; }; readonly actorIsEphemeral: boolean; /** The broker's live `max_payload`. Throws when not connected — never guesses a default. */ readonly maxPayload: number; /** The single-replica preflight. The emitter calls this at startup, before anything can publish. */ assertExpectationSemantics(): Promise; encodedSize(o: { channel: string; parts: Part[]; id: string; expectedLastSubjectSeq: number; }): number; multicastExpecting(o: { channel: string; parts: Part[]; id: string; expectedLastSubjectSeq: number; }): Promise<{ ack: { seq: number; duplicate: boolean; }; }>; } /** * One source record's worth of events, with the cursor that resumes AFTER that record. * * One durable emit unit is one frame: the emitter splits only at source-observation boundaries that * are independently reconstructable from the durable source. A frame therefore ends where a record * ends, and carries that record's cursor, so folding it means exactly "every record here is * consumed". */ export interface EmitUnit { /** The run these events belong to. A frame's envelope names ONE run, so units are never * mixed across runs in one frame. */ runId: string; events: AguiEvent[]; cursor: string; /** The turns the source began at or just before this unit, and by the dropped records whose * cursor folded onto it after, in source order: the run each one opened, or `null` for a turn * that opened none. Committed as the last turn begun when the frame carrying the unit is frozen, * see {@link AguiEmitter.queueTerminal}: every turn a frame's cursor covers is committed with * that frame, trailing ones included, or a failure after the freeze would carry a stale last * turn to the next emitter while the log had advanced past them (Sol, round 12 of 2026-10-13, * through the built MCP entry point: a notification's terminal closed the run before it). */ turns?: (string | null)[]; } /** A turn terminal the connector reported at a boundary the record stream cannot see, kept until * the run it belongs to can take it: `error` closes that run with `RUN_ERROR`, its absence with * `RUN_FINISHED`. */ export interface HeldTerminal { timestamp: number; error?: { message: string; code?: string; }; } /** A terminal with the position the source had reached when the connector reported it: the cursor * its {@link DurableSource.end} answered then. Which turn the terminal ends is read off that * position, see {@link AguiEmitter.queueTerminal}. */ export interface ReportedTerminal { terminal: HeldTerminal; at: Promise; } /** * A consent's §13.19.2 capture frontier: the position `P` the connector sampled with the source's * end operation after verifying the consent, and the life generation the consent held then. * * `generation` is re-read, not captured: §13.19.2 requires the generation be compared again after * every asynchronous step, and placement is asynchronous. A revoke or replacement advances it, and * an in-flight activation for a superseded generation must commit nothing — so the emitter asks for * it at the moment of commit rather than trusting the number it was handed when the consent arrived. */ export interface ConsentFrontier { consentId: string; at: Promise; generation: () => number; } /** * What the mapper returns for one source record: its run and its events, or `null` for a record * this plane deliberately drops, and it drops many. `null` is NOT an error: a deliberate drop and * a failed map are kept apart, because conflating them turns a parser bug into skipped history. * * `{ turn: true }` is a record that BEGAN A TURN AND OPENED NO RUN: a prompt the mapper knows and * deliberately does not publish. It is still a turn the harness ends with a terminal, so the * emitter counts it as a turn when it places a terminal reported after it (see * {@link AguiEmitter.queueTerminal}); dropped as `null`, that terminal would land on the run begun * before it. A record that opens a run begins its turn by its `RUN_STARTED`. */ export type RecordMapper = (record: T) => { runId: string; events: AguiEvent[]; } | { turn: true; } | null; /** * A bracket violation that is OURS, not the writer's: the machine that tracks open runs and messages * was lost across a process restart. * * **This exists because two halts that both say "unbalanced" prove nothing about which produced * one.** The WAL persists `epoch`, `frontier` and the pending frame, and NOT the set of open * runs and messages, so a process that dies mid-run restarts with an empty {@link AguiBrackets}, * resumes from `sourceCursor` at events whose `RUN_STARTED` was already published, and refuses the * first of them. Without this class the operator sees "nothing may be emitted outside an open run" * and files a bug against a writer that did nothing wrong. * * It is deliberately a SUBCLASS: every existing catch of {@link AguiVocabularyError} still catches * it, and only code that wants to tell the two apart has to know it exists. */ export declare class AguiBracketStateLost extends AguiVocabularyError { readonly cause: Error; constructor(message: string, cause: Error); } /** * The emitter has stopped and will not publish again without operator action. * * Halting is a SUCCESS of this design, not a failure of it: every halt below is a case where the * alternative is to report success for a message that was not stored, or to fold an ack for a body * we did not write. A halt is loud, bounded and recoverable by a human; the alternative is silent * and permanent. */ export declare class AguiEmitterHalted extends Error { readonly reason: "duplicate-ack" | "cas-loss" | "egress-policy" | "egress-unreadable" | "egress-extra-property" | "egress-run-error"; constructor(reason: "duplicate-ack" | "cas-loss" | "egress-policy" | "egress-unreadable" | "egress-extra-property" | "egress-run-error", message: string); } /** * Pack units into frames, splitting ONLY at unit boundaries and never inside one. * * Deliberately NOT `splitFrames`, and the difference is the durable plane's one-unit-one-frame * rule. `splitFrames` splits at EVENT boundaries, which is the right answer for a frame considered * on its own, but a frame that ends mid-record has no cursor it can honestly store: the only value * available says the whole record was consumed, and folding that after a crash skips the rest of the * record's events with no `seq` gap for a consumer to notice. * * **So the event-boundary split and the one-unit-one-frame rule are in tension, and this resolves it * in the direction the durable plane requires: a single unit that does not fit FAILS LOUD rather * than being truncated at a frame boundary.** That leaves `splitFrames`'s truncation path with no * caller on the durable plane, which is reported as a design conflict rather than decided here. * * @throws {AguiVocabularyError} when one unit cannot fit in a frame alone. */ export declare function packUnits(opts: { threadId: string; epoch: string; firstSeq: number; units: readonly EmitUnit[]; measure: (frame: AguiFrame) => number; limit: number; }): { frame: AguiFrame; cursor: string; unitFrom: number; }[]; /** * The event emitter: one per principal, one thread at a time. * * **BRACKET STATE SURVIVES A RESTART, AND THIS PARAGRAPH USED TO SAY THE OPPOSITE.** It described a * declared gap — an emitter coming back with an empty machine, resuming at events whose * `RUN_STARTED` had already been published, and refusing the first of them — long after the WAL * started persisting the machine. The words were true when they were written and stayed on the page * through the change that falsified them, which is the failure mode a class header is worst at * showing: it is the first thing a cutover author reads about recovery, and it was telling them to * expect a halt the code no longer produces. * * What actually happens: {@link AguiBrackets} is a property of the WRITER'S STREAM across frames, so * the WAL freezes the machine's state WITH each pending frame and promotes it on fold. A restart * therefore reopens knowing exactly which run, messages and tool calls were open at the last FOLDED * position, and {@link AguiBrackets.restore} continues from there rather than from empty. * * **The lost-state path still exists, and it is now the narrow case it should always have been:** a * document that CANNOT SAY what was open. That is a WAL migrated from v1, which recorded no bracket * state at all, and it loads as `null` rather than as an empty machine precisely so the difference * stays visible. Only there does the emitter start empty, resume into an already-open run, and * refuse the first event with {@link AguiBracketStateLost} — a halt rather than a loss, which is the * safe direction, and diagnosed by name rather than surfacing as an anonymous protocol violation. */ export declare class AguiEmitter { private readonly ep; private readonly wal; private readonly source; private readonly map; /** Derived from the endpoint's OWN principal, never from a config name or the launch env. */ readonly channel: string; readonly threadId: string; private readonly onUnreadable; /** * The bracket machine AT THE FOLDED POSITION — deliberately not "wherever validation got to". * * It advances one frame at a time, immediately before that frame's `beginSend`, so the state * frozen with a pending frame is the state that belongs to it. A machine advanced by the whole * batch up front would freeze a state describing events that had not been sent. */ private brackets; private halted; /** True once THIS process has fed an event through the bracket machine. It is the half of the * restart diagnosis that keeps a genuine mid-stream violation from being blamed on a restart. */ private fedAnyEvent; /** True once a pump has read the source: the last turn can be seeded only before that. */ private pumped; /** Terminals owed to runs, by run: a terminal placed on a run the wire has open or a pump is * about to open, published in place of the finish the source synthesizes for that run, or taken * back by {@link takeOwed} for a close the holder makes. Settled at the freeze of the frame that * closes the run. */ private owed; /** Terminals reported at a position no pump has read up to yet, in the order they arrived. */ private queued; /** Frontiers no read has placed yet. An unresolved frontier grants NO eligibility (§13.19.2). */ private frontiers; /** Consents whose frontier a previous read placed: every record from this read on is after `P`. */ private settled; /** Consents placed inside the CURRENT read, at the index from which their records are after `P`. */ private placing; /** * The turn the frozen frames most recently began: the run it opened, `null` for one that opened * none, `undefined` while the log shows no turn. A terminal reported before any record a pump * reads is this turn's. Starts as the run the log has open, which is the last turn the log can * name; a holder that drops an emitter seeds the next one with what that emitter knew. */ private lastTurn; private constructor(); /** * Start an emitter: resolve the channel, run the single-replica preflight, and settle any pending * frame. * * **THIS IS THAT PREFLIGHT'S PRODUCTION CALL SITE, AND UNTIL THIS FUNCTION EXISTED THERE WAS * NONE.** `CotalEndpoint.assertExpectationSemantics()` had zero production callers: it was a * check that shipped, was covered by its own suite, and never ran outside one. That is why it is * called HERE, before recovery and therefore before any publish — a serialized append on an * unverified stream is the exact case it exists to prevent, and doing it after recovery would * leave the one publish that matters most, the re-publish of a frozen frame, outside the guard. */ static start(opts: { endpoint: EmitterEndpoint; /** Already open, so the caller owns `space`, the WAL path, and the `subjectMayExist` judgement * — none of which the emitter can make honestly on the caller's behalf. */ wal: EventWal; /** * The PRINCIPAL-scoped subject frontier. **Required, and not optional with a zero default.** * * The subject is shared by every thread of one principal, so the expectation a publish carries * is a fact about the principal and not about the thread. An optional parameter here would let * a new connector omit it and reintroduce, silently, the defect where an agent's second session * expects an empty subject its own first session filled. There is one thing to pass and there * is no legal way to not pass it. */ subjectFrontier: SubjectFrontier; source: DurableSource; map: RecordMapper; /** * Where a {@link SourceGap} the source stepped over is reported. Optional, and the absence is * not a quiet default: a pump that meets a gap with no reporter refuses, naming the gap, so a * connector that never decided where gaps go keeps the loud answer rather than a silent hole. */ onUnreadable?: (gap: SourceGap) => void; }): Promise>; /** True once the emitter has stopped for good. */ get stopped(): boolean; /** The run {@link closeRun} would close now, or `undefined` at a stopping point. */ get openRunId(): string | undefined; /** * Boot recovery, branching on the WAL's tag. * * `acked` NEVER republishes: the frame landed and we know it, so the only remaining work is to * fold. `sent_unacked` is the genuinely uncertain case and republishes with the SAME frozen `id` * and `E` — never the current tip, because re-deriving either is what turns an uncertain publish * into a second, different message. */ private recover; /** * Read forward, map, pack, and publish. Returns what it did, so a caller can distinguish "nothing * to do" from "did work" without inspecting the WAL. */ pump(): Promise<{ frames: number; events: number; }>; /** * Place every queued terminal the read has reached on the turn the source had begun when the * terminal was reported: the last of the turns begun by the records before its position, or * {@link lastTurn} when none of them is. A terminal reported past this read stays queued for the * next one. Placed on a run, the terminal becomes that run's debt; on a turn that opened no run, * it is consumed; on a run that already has a terminal, or one the wire has closed, it is * dropped as the connector repeating itself. */ private place; /** * How many of the read's records the source held when the terminal was reported, or `undefined` * when the terminal was reported past the read. The position is compared through the source * itself, never by reading the cursor: the first record after the position is the one a read * from it answers, and the cursor it carries is the same string this read gave that record. A * read from the position that answers no record and ends where this read ends covers only lines * the source could not read, so the terminal is after every record this read holds. A * position whose read could not be taken, or that this emitter had already read past (a terminal * is reported before anything after its position is read, so that position is never behind a * read), drops the terminal and fails this pump, once: the plane reports it and the next pump * goes on without it. Waiting for a pump that reads up to such a position would wait forever. */ private boundary; /** * The commit of frozen units and of records advanced past without a frame: the last turn they * began is now the turn a terminal reported before any later record ends, and a run they closed * owes nothing, its terminal being in the frozen frame. */ private committed; /** * Close the run this stream currently has open, at a boundary the RECORD STREAM CANNOT SEE. * * **This exists because the two halves of the mapping were specified against different inputs.** * The plan sources `RUN_FINISHED` from a harness lifecycle hook, and the durable plane reads a * FILE: a hook fires in another process and writes no record, so a hook-sourced terminal has no * vehicle into a record-sourced stream. Deriving the terminal from records instead is possible but * lies about time in two ways that matter to a live view: the finish lands only when the NEXT turn * starts, so a finished agent renders as still running, and the last run of a session never closes * at all, because there is no later record to close it on. This is that vehicle. * * It is a FRAME LIKE ANY OTHER: same epoch, same `seq` line, same write-ahead discipline, same * halt rules. The single thing that differs is the cursor, which is republished UNCHANGED, because * this frame consumes no source record. A frame that advanced the cursor here would mark records * consumed that were never mapped. * * Idempotent by construction rather than by a flag: the bracket machine is the only state it * reads, so once the run is closed there is nothing open to close and it answers `null`. That also * makes it safe on a stream whose run was opened by a PREVIOUS process, since the machine is * restored from the WAL. * * **AN `error` CLOSES THE SAME RUN WITH `RUN_ERROR` INSTEAD, and it is one method rather than two * ON PURPOSE.** `RUN_ERROR` closes a run on its own, so a run that emitted one must never also * emit a `RUN_FINISHED`. With a second method that invariant would be a rule someone has to * remember; with one method and one branch it is a property of the shape: exactly one terminal is * built, and the bracket machine has closed the run by the time anything could ask for another, so * a following close answers `null` like any other close on a settled stream. Which harness signals * mean a turn FAILED is a connector's decision and is stated at each connector's own mapping site; * this file only carries the answer to the wire. * * **`error.message` AND `error.code` NEVER REACH THE WIRE.** Both are upstream values, so the close * publishes {@link RUN_ERROR_EGRESS_MESSAGE} with no code, the same shape the write path's * `egressRunError` in `agui.ts` produces (#1431). `packUnits` bounds the closing frame like any * other, so a close whose envelope cannot fit fails loud. * * @returns the run that was closed, or `null` when the stream was already at a stopping point. */ closeRun(o: { timestamp: number; cotal?: CotalMeta; /** Close with `RUN_ERROR` carrying these instead of `RUN_FINISHED`. */ error?: { message: string; code?: string; }; }): Promise; /** * Queue a terminal the connector reported at a boundary the record stream cannot see, for the * turn the source had begun when it was reported: a terminal that arrived while the plane was * down, or while a re-establish was in flight. The connector reports a turn's terminal once the * turn's records are in the source and before the next turn's are, so the position the source * had reached then names the turn: the last one begun before that position. The next pump reads * the records up to that position and places the terminal, see {@link place}. Placed on a run, * an error is published in place of the first finish the mapper produces for that run, which is * the finish a later prompt synthesizes when it finds the run still open, so a failed turn whose * next prompt was already written when the plane came back is still closed as a failure, after * its own records and before the next run opens, in a frame of its own run; a finish is that * synthesized finish. A terminal the pump places on the run it leaves open is handed back by * {@link takeOwed} for the close the holder makes. A turn that opened no run takes its terminal * and does nothing with it, as a close with no run open does live; a second terminal for one * turn is the connector repeating itself, and replaces nothing. */ queueTerminal(entry: ReportedTerminal): void; /** * Queue a consent's §13.19.2 capture frontier, for the records that lie after it. * * `P` is a position the caller sampled with the source's end operation after verifying the * consent, and the emitter's job is only to say which records fall after it. It says so through * the SAME comparison a reported terminal is placed with ({@link boundary}): source-only * position comparison, never a cursor-string comparison. Physical tool execution time decides * nothing — a slow call whose START lies before `P` is ineligible, and a fast one after it is * eligible, which is the whole point of placing a position rather than reading a clock. * * The frontier grants nothing until a read places it, so a consent in flight is a consent that * can read nothing. It is committed at placement, and the life generation is re-read there: a * revoke or replacement that landed while the position was resolving has advanced the generation, * and the activation then commits nothing, as §13.19.2 requires. A position that cannot be placed * drops the frontier — the caller refuses that consent as `failed-precondition`; it does not fall * back to eligible-for-everything, which is the one failure mode worth more than the surface. */ queueFrontier(frontier: ConsentFrontier): void; /** * Place every queued frontier the read has reached, and retire the ones a previous read placed. * * Retiring first is what makes {@link eligibleAt} cheap and correct: a frontier placed in an * EARLIER read is behind every record of this one, so it needs no index, and only frontiers * placed inside this read carry a boundary to compare against. */ private placeFrontiers; /** * The consents a START at index `i` of the current read is eligible for (§13.19.2/.3). * * A record is eligible for a consent when its durable position lies AFTER that consent's `P`: * either the frontier was placed in an earlier read, or it was placed in this one at a boundary * this record is at or past. The answer is a fresh set so a caller cannot widen the emitter's * own state by holding onto it — an audience that grows is the §13.19.3 leak. */ eligibleAt(i: number): string[]; /** How many frontiers no pump has placed yet. One that is still waiting grants no eligibility. */ get waitingFrontiers(): number; /** How many terminals no pump has placed yet. A pump places every one the read reaches. */ get waitingTerminals(): number; /** Take the terminal owed to `runId` back, for a close the caller makes itself. */ takeOwed(runId: string): ReportedTerminal | undefined; /** * Every terminal this emitter still holds, for a holder that is dropping it: the terminals owed * to runs the wire has not closed, then the ones no pump has placed, each with the position it * was reported at, so the next emitter places them again from the log it starts from. The last * turn the frozen frames began rides along, so the holder can seed the next emitter with it. */ unsettled(): { terminals: ReportedTerminal[]; lastTurn: string | null | undefined; }; /** Seed the last turn the wire began, from the emitter this one replaces: a terminal reported * before any record the first pump reads is that turn's. Allowed before anything is pumped. */ seedLastTurn(turn: string | null | undefined): void; /** * Take the first finish of an owed run out of a record's events and, for a debt that is an * error, answer it with a unit of that run carrying the owed error terminal instead. The error * carries the fixed egress message, as a close does. `before` is the cursor the owed unit * stores: the position ahead of the record the finish came from. A debt that is a finish is * answered by the finish itself and changes nothing. The debt is settled when the frame carrying * the terminal is frozen, not here: a frame whose write fails before the freeze is read again, * and the terminal must still be owed then (Sol, round 11 of 2026-10-13, through the built MCP * entry point: a log write refused once while the owed error was being written closed the failed * run as a finish). */ private takeOwedTerminal; /** Measure a candidate frame EXACTLY as the wire will, at an upper bound over id and expectation. */ private measure; /** Transition 1 then the first network attempt. `frozen` runs once the frame is frozen on disk, * before the wire is tried: the commit of what the frame carries, for state the frame settles. */ private publish; /** * One publish attempt — first or retry — with the FROZEN id and the FROZEN `E`. Never the tip. * * The three outcomes are not symmetric and the asymmetry is the design: * - `!duplicate` → transition 2 then 3. Success becomes durable before the frontier moves. * - `duplicate` → HALT. On a first attempt it means a body WE DID NOT WRITE holds our id, and * folding its `ackSeq` would advance the frontier and the source cursor past events that were * never published. On a retry it cannot happen on a single-replica stream at all, because such a * stream evaluates the expectation before the dedup cache, so observing it proves the stream is * not single-replica. Both are * fail-loud, and neither is a case where guessing is better than stopping. * - CAS loss → HALT. Someone else moved the tip on a subject only this principal may write, or * the subject was purged. Uncertainty plus a moved tip is exactly what must not be guessed at. * * A NETWORK error is deliberately none of these: it leaves `pending` as `sent_unacked`, which is * the state that means "we do not know", and the next boot retries the same frozen frame. */ private attempt; /** * Decide whether a bracket refusal is the WRITER's fault or OURS, and say which. * * Ours iff ALL THREE hold, and each is load-bearing: * - this process has fed NO event through the machine yet, so the machine cannot have been put * into a bad state by anything we did in this run; and * - the frontier is non-virgin, so frames — and therefore possibly an open `RUN_STARTED` — were * published by a PREVIOUS process; and * - the WAL cannot say what was open. Since v2 the machine is PERSISTED, so an ordinary restart * restores it and never reaches here at all; `null` means the document was migrated from v1 and * genuinely never recorded the state. Without this condition the diagnosis would survive as a * permanent excuse for a case the migration fixed. * * Drop the first condition and a genuine mid-stream violation by the writer gets blamed on a * restart that happened an hour ago. Drop the second and a violation on a virgin thread, where * nothing was ever published and nothing could have been lost, gets blamed on a restart that never * happened. Each condition alone produces a confident, wrong diagnosis — which is worse than the * undiagnosed error it replaced, because a named cause stops the search. */ private diagnoseBracket; private halt; } //# sourceMappingURL=agui-emitter.d.ts.map