diff --git a/AGENTS.md b/AGENTS.md index 17689fa..63b42df 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -240,7 +240,7 @@ Key methods: **Runtime** (`src/runtime/`, `designs/agent-runtime.md`): `atem runtime` (hidden; one per config dir: an exclusive `flock` on `/runtime.lock` holding its pid, the socket `$XDG_RUNTIME_DIR/atem/runtime.sock` (else in the config dir) recorded in `/runtime.socket`, output in `/runtime.log`) is the broker: it serves local frontends over the same-user socket (local protocol v1, `runtime/protocol.rs`: `hello`, `launch`, `attach`, `detach`, `input`, `paste`, `resize`, `reclaim`, `list`, `stop`, `reload`, `astation_op`/`answer` in; `welcome`, `launched`, `attached`, `output`, `exited`, `sessions`, `detached`, `stopped`, `session`, `astation`, `notice`, `print`/`ask`/`done`, `error` out), holds the App with the machine's one Astation link, and owns the session host. Clients (`atem agent …`, bare `atem`, `atem pair`/`pair verify`/`cred unlock|projects`) start it detached when none answers; a runtime of another protocol without live sessions restarts for them. It exits after 10 minutes with no sessions and no non-passive clients (`ATEM_RUNTIME_IDLE_SECS` overrides), on SIGTERM/SIGINT, or when replaced. Every session runs in its own `atem worker --state /runtime --session ` (hidden, spawned detached; socket `$XDG_RUNTIME_DIR/atem/workers/.sock`, else `/runtime/workers/`; spec, record (`session_id`, `process_gen`, pids and OS start times, exit) and log in `/runtime/workers/`), which owns the agent's PTY, the canonical screen, Claude's chat adapter and the session log, and outlives the broker: a new broker adopts each surviving worker only when its `hello` names the record's session id and process generation and the pid's OS start time matches (Linux `/proc//stat` field 22, macOS `proc_pidinfo`); otherwise the session is `recovery_required` (or `interrupted`/`exited`) and nothing is restarted or replayed. The broker is the only writer authority: every write and paste carries the input epoch it granted, and the worker drops one whose epoch is older than its current one at the PTY write boundary (`pty_session.rs` fenced writes). Session ids are `ses-` + 12 base32 characters; each process is a new `process_gen` (UUID). -**Session host** (`runtime/host_bridge.rs`, `session_protocol/`, `app/hosted.rs`; contract `designs/protocol/atem-session-v1.md` r7.1, compatibility mode): only `atem runtime` serves sessions remotely. Its broker owns `SessionHost` over a `FileJournal` in `/runtime/journal/` (per-session receipt logs under `receipts/`, 0700/0600, `O_NOFOLLOW`, fsync per transition, admission halted after an fsync error, compaction, a lease-epoch floors file, `journal.lock`, and a `runtime_epoch` counter incremented at each start) and makes the App's authenticated Astation link one attachment — new on every connect or re-authentication (`App::astation_link_gen`), fenced when the link drops or is re-authenticated, so clients `hello` again and an old `attachment_gen` is `stale_target`. A second `hello` on the same authenticated link (Astation reset its client, e.g. after sleep) replaces the attachment (`Effect::Replaced`): the old one is fenced (its leases end, streams and conversations close), the new one gets a new `attachment_gen`/`controller_session` with the link's unchanged authorization deadline; receipts stay; at most one a second (`busy`). runtime.log gets one `HH:MM:SS` UTC line per `hello` (welcome, replaced, refused ``). Only the paired Astation the link captured, verified in `cred_state.json`, gets one (`view`, `control`, `client_kind` `native`); any other link gets none and its session messages are answered `unauthorized`. `welcome` names the verified `data_account` (`account_id`), this device's verified `device_id` (`atem_device_id`) and `pairing.json`'s `host_trust_gen`; `runtime_gen`/`runtime_epoch` change when the broker restarts or the identity changes (a pairing verified after the broker started rebinds the host over the same journal). The attachment is authorized for 24 h (no freshness proof yet), on the monotonic clock and in wall time, whichever ends first: then it is fenced and the link dropped, and the reconnect authenticates again. Every 5 s the broker re-checks the link's pairing capture against `pairing.json` and drops a link whose pairing changed (unpair works even without `reload`). Session-v1 (`hello`, `list` → `sessions`/`session_changed` with a per-attachment `seq`, `acquire`/`renew`/`release` → 15 s leases with explicit `takeover`, `submit` → receipts, `receipt_query`) is the registry and the only way to get control. Receipts per kind: `prompt` is pasted once `ChatSession::submit` accepts it and completes at the chat's turn end (a session without chat — Codex with `[codex] hooks` off, Claude before its SessionStart or after its chat was withdrawn — rejects it with no effect: "chat is not available for this agent; use the terminal"); `interrupt` writes Esc only while the chat status is `working` or `waiting` and completes when that turn ends, completes at once with nothing written in `ready` or `unknown`, and is `rejected` ("no turn state; use the terminal") whenever the session has no turn state, for any agent (Codex with its hooks off, Claude before its SessionStart or after its chat was withdrawn), never a blind Esc; `stop` completes at the exit; `approval` completes when the agent applied the decision (see Approval hooks). A refused-before-queueing write is `rejected`, a writer error after queueing `outcome_unknown`; executing work is `outcome_unknown` after a broker restart (never retried); duplicate submits return the recorded receipt, the same `request_id` with other content is `request_id_reused`. Terminal-v1 (`subscribe`/`unsubscribe`/`ack`, and `input`/`resize`/`stop` with the lease; terminal id = session id) and chat-v1 (`open`/`close`/`submit` with a binding, `submit` with the lease; `accepted`/`complete`/`error{code}` from its receipt; at most 8 conversations per session) are bound families. Effects reach the workers: remote bytes are checked with `admit_write` before they are sent and tagged with the worker's input epoch, which the broker raises (`set_input_epoch`, at least the lease epoch) whenever the host's epoch moves after remote bytes or while a lease was held; remote writes and pastes also carry `not_after`, the lease's deadline on CLOCK_MONOTONIC (`pty_session::monotonic_ns`), and the worker drops one at the PTY write boundary once it has passed (reported like an epoch drop; local writes carry none); each remote `input` gets a `written` ack (`control changed; not written` when dropped, `the approval was handed to the terminal; not written` when it released a Codex approval hook); a slow terminal transport drops frames and ends streams with `busy`. Local frontend input goes through `local_input` (it revokes any remote lease first). Duplicate session messages (same attachment, `id` and content) within 60 s are handled once (at most 4096 remembered). `advance` runs every 100 ms of the session clock. Adopted workers send a `chat_snapshot` before any `chat`, so chat goes on after a broker restart. Bare `atem` opens no Astation connection, so only the runtime ever sees session-family messages. A runtime whose session host could not start (its journal or a damaged `runtime_epoch`, never reset silently) answers them `unsupported_version`, and `welcome.host_error` (shown by `atem agent list` and in `runtime.log`) says why, e.g. "remote sessions disabled: is damaged; move it aside to reset the runtime epoch". §9 features (resume, browser grants, account-wide access, relay confidentiality) are refused. A socket alive is not an authenticated Astation: when Astation sends `auth_required` over an already-authenticated socket (e.g. it alone reconnected to the relay), `app/reauth.rs` shows `Astation ✗`, fences the attachment and ignores every other Astation request, then answers only the pinned Astation non-interactively (`AstationClient::answer_auth_required`) with the connect path's proof (`local_proof` on loopback, else the saved session); success restores online as a new attachment, while denial, expiry, a different `astation_id` or no answer within 15 s leaves a notice and closes the socket for the normal reconnect (never pairing). The shared fixtures run against the real runtime in `tests/runtime_fixtures_binary.rs` (results: `designs/protocol/atem-host-handoff.md`). +**Session host** (`runtime/host_bridge.rs`, `session_protocol/`, `app/hosted.rs`; contract `designs/protocol/atem-session-v1.md` r7.1, compatibility mode): only `atem runtime` serves sessions remotely. Its broker owns `SessionHost` over a `FileJournal` in `/runtime/journal/` (per-session receipt logs under `receipts/`, 0700/0600, `O_NOFOLLOW`, fsync per transition, admission halted after an fsync error, compaction, a lease-epoch floors file, `journal.lock`, and a `runtime_epoch` counter incremented at each start) and makes the App's authenticated Astation link one attachment — new on every connect or re-authentication (`App::astation_link_gen`), fenced when the link drops or is re-authenticated, so clients `hello` again and an old `attachment_gen` is `stale_target`. A second `hello` on the same authenticated link (Astation reset its client, e.g. after sleep) replaces the attachment (`Effect::Replaced`): the old one is fenced (its leases end, streams and conversations close), the new one gets a new `attachment_gen`/`controller_session` with the link's unchanged authorization deadline; receipts stay; at most one a second (`busy`). runtime.log gets one `HH:MM:SS` UTC line per `hello` (welcome, replaced, refused ``). Only the paired Astation the link captured, verified in `cred_state.json`, gets one (`view`, `control`, `client_kind` `native`); any other link gets none and its session messages are answered `unauthorized`. `welcome` names the verified `data_account` (`account_id`), this device's verified `device_id` (`atem_device_id`) and `pairing.json`'s `host_trust_gen`; `runtime_gen`/`runtime_epoch` change when the broker restarts or the identity changes (a pairing verified after the broker started rebinds the host over the same journal). The attachment is authorized for 24 h (no freshness proof yet), on the monotonic clock and in wall time, whichever ends first: then it is fenced and the link dropped, and the reconnect authenticates again. Every 5 s the broker re-checks the link's pairing capture against `pairing.json` and drops a link whose pairing changed (unpair works even without `reload`). Session-v1 (`hello`, `list` → `sessions`/`session_changed` with a per-attachment `seq`, `acquire`/`renew`/`release` → 15 s leases with explicit `takeover`, `submit` → receipts, `receipt_query`) is the registry and the only way to get control. Receipts per kind: `prompt` is pasted once `ChatSession::submit` accepts it and completes at the chat's turn end with `detail` `done` or `interrupted` (a session without chat — Codex with `[codex] hooks` off, Claude before its SessionStart or after its chat was withdrawn — rejects it with no effect: "chat is not available for this agent; use the terminal"); `interrupt` writes Esc only while the chat status is `working` or `waiting` and completes when that turn ends, completes at once with nothing written in `ready` or `unknown`, and is `rejected` ("no turn state; use the terminal") whenever the session has no turn state, for any agent (Codex with its hooks off, Claude before its SessionStart or after its chat was withdrawn), never a blind Esc; `stop` completes at the exit; `approval` completes when the agent applied the decision (see Approval hooks). A refused-before-queueing write is `rejected`, a writer error after queueing `outcome_unknown`; executing work is `outcome_unknown` after a broker restart (never retried); duplicate submits return the recorded receipt, the same `request_id` with other content is `request_id_reused`. Terminal-v1 (`subscribe`/`unsubscribe`/`ack`, and `input`/`resize`/`stop` with the lease; terminal id = session id) and chat-v1 (`open`/`close`/`submit` with a binding, `submit` with the lease; `accepted`/`complete`/`error{code}` from its receipt; at most 8 conversations per session) are bound families. Effects reach the workers: remote bytes are checked with `admit_write` before they are sent and tagged with the worker's input epoch, which the broker raises (`set_input_epoch`, at least the lease epoch) whenever the host's epoch moves after remote bytes or while a lease was held; remote writes and pastes also carry `not_after`, the lease's deadline on CLOCK_MONOTONIC (`pty_session::monotonic_ns`), and the worker drops one at the PTY write boundary once it has passed (reported like an epoch drop; local writes carry none); each remote `input` gets a `written` ack (`control changed; not written` when dropped, `the approval was handed to the terminal; not written` when it released a Codex approval hook); a slow terminal transport drops frames and ends streams with `busy`. Local frontend input goes through `local_input` (it revokes any remote lease first). Duplicate session messages (same attachment, `id` and content) within 60 s are handled once (at most 4096 remembered). `advance` runs every 100 ms of the session clock. Adopted workers send a `chat_snapshot` before any `chat` (the whole retained history, 500 messages / 8 MiB, older messages first in `chat_history_part` lines of ≤ 1 MiB; approval cards included, since the broker copies each card to its worker with `chat_card`; a card still pending becomes `closed`), so chat goes on after a broker restart. A `chat` with `gap > 0` makes the broker reload the worker's history (`request_chat_history` → `chat_history`), then open conversations get `error` "the chat history was incomplete …; reopen the chat" and `close`. chat-v1 `open{before}` on an open conversation pages without `opened` or a reset (the page continues its live sequence) and ends with a sequenced `status` carrying `page: {before, count, start}`; cards are filtered for view-only attachments before the limits; a `before` not retained (or named with another `generation`) is `error{code:"stale_target"}`. The reload also carries the worker's chat state (`ChatSession::catch_up`: a missed turn end completes its receipt); a reload lost with its worker, or over 1000 messages, falls back to the same `error`/`close`. Bare `atem` opens no Astation connection, so only the runtime ever sees session-family messages. A runtime whose session host could not start (its journal or a damaged `runtime_epoch`, never reset silently) answers them `unsupported_version`, and `welcome.host_error` (shown by `atem agent list` and in `runtime.log`) says why, e.g. "remote sessions disabled: is damaged; move it aside to reset the runtime epoch". §9 features (resume, browser grants, account-wide access, relay confidentiality) are refused. A socket alive is not an authenticated Astation: when Astation sends `auth_required` over an already-authenticated socket (e.g. it alone reconnected to the relay), `app/reauth.rs` shows `Astation ✗`, fences the attachment and ignores every other Astation request, then answers only the pinned Astation non-interactively (`AstationClient::answer_auth_required`) with the connect path's proof (`local_proof` on loopback, else the saved session); success restores online as a new attachment, while denial, expiry, a different `astation_id` or no answer within 15 s leaves a notice and closes the socket for the normal reconnect (never pairing). The shared fixtures run against the real runtime in `tests/runtime_fixtures_binary.rs` (results: `designs/protocol/atem-host-handoff.md`). **Astation operations through the runtime** (`runtime/ops.rs`, `runtime/op_client.rs`, `app/op_route.rs`): `atem pair` (first pairing), `atem pair verify`, `atem cred unlock`, `atem cred projects` and the paired-token refresh run inside the broker on its existing Astation link (waiting up to 15 s for it); the CLI is only the user interface (`print`/`ask` out, `answer` in, `done` at the end; answers from `/dev/tty`) and never opens its own connection (a runtime that can't be reached fails the command). One operation at a time (a second is refused); its handshake replies are routed to it only and dropped after it ends; a CLI that goes away ends it as a failure (device verification sends `verifyAbort`); a lost link ends it, never replayed. `atem unpair` and `atem logout` send `reload` to a running runtime (never starting one): it re-reads the config and drops a link whose pairing changed, with its attachments. diff --git a/designs/agent-chat-api.md b/designs/agent-chat-api.md index 4a2d54f..4ddd07f 100644 --- a/designs/agent-chat-api.md +++ b/designs/agent-chat-api.md @@ -108,8 +108,29 @@ Reply, in order: `opened`, then history as `message` events. - History is the latest 50 messages / 256 KiB, sent whole (never as deltas), each with `history: true` and the next `sequence`. Replace anything you cached with the same `messageId`. -- Older pages: `open` again with `before: ""`. This re-opens the - conversation (a new `opened`, `sequence` 0) and sends that older page. +- Older pages: `open` again on the same open conversation with + `before: ""` (usually the oldest `messageId` you hold). No + `opened` comes and nothing is reset: the older page (50 messages / 256 KiB + before that message) arrives as `message{history:true}` at the + conversation's next sequence numbers, so keep applying `last + 1`; live + events before and after it keep their own sequence numbers. Insert paged + messages by `messageId` and history order (they are older than what you + have), never by `sequence`. Every page ends with a sequenced `status` + that also carries `page`: + `{"op":"status",…,"state":"ready","since":…,"page":{"before":"m40","count":50,"start":false}}`. + `count` is how many messages the page held; `start: true` means it + reached the oldest message Atem retains, so stop paging. An empty page is + just that marker (`count: 0, start: true`). Treat it as a normal status + update too. Without `control`, approval cards are left out before the + page limits, so you still get older messages after a run of cards. +- Send `generation` (the one from `opened`) with `before`. A `before` Atem + no longer retains (evicted), or one named with another `generation`, is + answered `error{code:"stale_target", text:"that message is no longer in + the history; reopen without before to get the newest page"}`, never an + empty page: "no more history" and "cursor unknown" are different. On an + open conversation the error is sequenced and the conversation stays open. +- `open` without `before` on an open conversation resynchronizes it: a new + `opened` at `sequence` 0 and the newest page; replace your cache. - `opened.unavailable` lists what this agent's chat can't provide: `streaming` (no partial text), `reasoning`, `error_details` (no model/API error behind a failed turn). Hide the matching UI. Unknown names are @@ -134,7 +155,7 @@ All carry `generation` and `sequence` (apply only `last + 1`): | `delta` | `messageId`, `part`: a later part appended to a message already sent (both agents; not a token stream); a part id already present is ignored | | `status` | `state` (`ready`/`working`/`waiting`/`unknown`), `since`, optional `detail`; re-sent every 15 s with the original `since` | | `accepted` | `requestId`: your prompt reached the agent | -| `complete` | `turnId`, optional `requestId`, `outcome`: `interrupted` when the turn ended interrupted (Esc, a deny at the agent's prompt, Codex's `turn_aborted`), else `done`; a duplicate submit after a broker restart gets `done` (the receipt only says completed) | +| `complete` | `turnId`, optional `requestId`, `outcome`: `interrupted` when the turn ended interrupted (Esc, a deny at the agent's prompt, Codex's `turn_aborted`), else `done`; for your own submit it comes from the receipt, whose `detail` keeps the outcome, so a duplicate after a broker restart says the same | | `error` | `text`, optional `requestId`, and `code` on host-made errors: `rejected`/`outcome_unknown` (your submit's receipt), a session-v1 admission code (`stale_target`, `stale_lease`, `lease_held`, `forbidden`, `session_closed`, `busy`, `paused`, ...), `unsupported_version` (malformed message), `busy` (open without chat). The chat's own notices (agent restarted, conversation switched, chat withdrawn) have no `code` | | `rules` | `approvalRules`: the whole list of active Atem session rules, when it changed (conversations with `control` only) | | `close` | the conversation ended | @@ -174,6 +195,7 @@ marks the cut). Text and summaries pass the credential redactor. | Another session resumed inside the agent (`/resume`, Codex `/new`) | `error` "the conversation was switched; reopen the chat", `close`; same `generation` | open again; the earlier turns are history only | | Agent exits | status `unknown`, errors to open conversations, generation retired | history stays readable | | Chat withdrawn (log format changed) | `error`, `close`; descriptor loses `chat-v1` | fall back to the terminal view | +| Atem dropped chat items before they reached the runtime (a huge burst of agent output) | `error` "the chat history was incomplete (some items were dropped) and was reloaded; reopen the chat", `close`; same `generation` | open again: the history is complete again | A new `open` always restores status and history. Atem keeps the newest 500 messages / 8 MiB per agent process. @@ -264,7 +286,10 @@ A draft typed during a running turn is not detected (v1 limit). Reply `receipts` (<= 64 ids, only those bound to that exact target). Receipts survive broker restarts. A session-v1 `receipt` has `request_id`, `target`, -`kind`, `state`, `detail?`, `updated_at`. Work executing across a broker restart +`kind`, `state`, `detail?`, `updated_at`. A `completed` `prompt` receipt's +`detail` is the turn's outcome: `"done"` or `"interrupted"` (the same as the +chat `complete.outcome`; the journal keeps only that word, never prompt or +reply text). Work executing across a broker restart becomes `outcome_unknown` and is never retried. Replay after a reconnect of the same chat submit is safe because the same `requestId` returns the recorded outcome, but the right first move is `receipt_query`. @@ -545,5 +570,8 @@ Ops: `hello`, `list`, `acquire`, chat `open`/`submit` or session-v1 `submit` session; deny `message` 512 bytes. - No semantic turn-completion signal beyond the agent's own turn end; no reasoning; no Discord adapter in Atem (the bridge above is yours). -- Approvals do not survive a broker restart. Atem session rules do, inactive - until the new host confirms them (section 5.5). +- Approvals do not survive a broker restart. Their cards stay in history; one + still pending becomes `closed` with the note "Atem's runtime restarted; + answer in the terminal". Atem session rules survive, inactive until the new + host confirms them (section 5.5). Chat history (500 messages / 8 MiB) + survives a broker restart whole. diff --git a/designs/agent-chat-host.md b/designs/agent-chat-host.md index 0e65067..4cc9631 100644 --- a/designs/agent-chat-host.md +++ b/designs/agent-chat-host.md @@ -349,7 +349,7 @@ on the existing authenticated Astation connection. | op | Direction | Fields | |---|---|---| -| `open` | Astation → Atem | `conversationId`, `agentId`, optional `before` (messageId) | +| `open` | Astation → Atem | `conversationId`, `agentId`, optional `before` (messageId) and `generation` (the one `before` came from) | | `opened` | Atem → Astation | `generation`, `sequence: 0`, `status {state, since}`, `unavailable` (array of strings, below), `approvalRules` (the session's active Atem session rules `[{rule_id, label, scope}]`; `[]` without `control`), then history as `message` events (latest 50 or 256 KB, older pages with `before`) | | `submit` | Astation → Atem | `generation`, `requestId` (UUID), and one of `text`, `approval` (§6a answer) or `approval_rule_revoke` (`{rule_id}`, r7.2) | | `rules` | Atem → Astation | `approvalRules` (the whole list, when it changed; conversations with `control` only) | @@ -360,19 +360,49 @@ on the existing authenticated Astation connection. | `error` | Atem → Astation | `text` (safe, no secrets), optional `requestId`, and `code` on every `error` the session host makes: `rejected` or `outcome_unknown` (a submit's receipt), a session-v1 code from admission (`stale_target`, `stale_lease`, `lease_held`, `forbidden`, …), `unsupported_version` (a malformed chat message) or `busy` (`open` without chat-v1, below); the chat's own errors (the agent exited or was restarted, the conversation was switched, chat withdrawn) carry no `code`; an `error` sent before a conversation is open (unknown agent, chat unavailable, bad request) carries no `generation`/`sequence` | | `close` | either | `conversationId` | -- `open` with `before` on an open conversation reopens it (`opened`, - sequence 0) and sends that older page. +- History pages (since 2026-10-17). `open` with `before` on a conversation + this attachment already has open pages it: no `opened`, nothing reset. + The older page (up to 50 messages / 256 KiB strictly before `before`) + arrives as `message{history:true}` at that conversation's next sequence + numbers, continuing its live sequence, so a live event never collides + with or overtakes a page and the client keeps applying `last + 1`. The + page holds older messages than ones already sent: place messages by + `messageId` (a whole message replaces an earlier copy) and order them by +`createdAt`/history position, not by `sequence`. Every such page ends + with a sequenced `status` (the current status) carrying + `page: {before, count, start}`: `count` is the number of messages the + page held, `start: true` means it reached the oldest retained message (no + older page to ask for). An empty page is that marker alone + (`count: 0, start: true`), never silence. Approval cards are left out of + a view-only attachment's pages before the 50-message / 256 KiB limits, + so a stretch of cards never yields an empty page while older messages + exist (the same holds for the newest page on `opened`). `open` without + `before` on an open conversation is a resynchronization: a new `opened` + at 0 and the newest page; replace the cache. `open` with `before` on a + conversation that isn't open opens it (`opened` at 0) with that page. +- `open` may carry the optional `generation` its `before` came from. A + `before` that isn't in the retained history (evicted, never sent), or + named with another `generation`, is refused with `error{code:"stale_target"}`, + text "that message is no longer in the history; reopen without before to + get the newest page", never an empty page. On an open conversation the + error is sequenced and the conversation stays open; on a new one it + carries no `generation`/`sequence` and opens nothing. `stale_target` is + the contract's code for a reference to something the host no longer + holds (§7: a closed approval, a revoked rule); no other §7 code fits. - `open` on a running session whose chat-v1 is off (Claude before its - SessionStart or after chat was withdrawn, Codex without Atem's hooks or - before its first hook) is answered `error{code:"busy"}` with the text - "chat is not available for this agent" (`: ` when the adapter - gave one) and opens nothing. An exited agent's history stays readable. + SessionStart or after chat was withdrawn, Codex without Atem's hooks) is + answered `error{code:"busy"}` with the text "chat is not available for + this agent" (`: ` when the adapter gave one) and opens nothing. + Codex with Atem's hooks offers chat-v1 from launch: `open` before its + first hook succeeds with status `unknown` and the detail "type a first + prompt in the terminal: …", and a `submit` is refused with that text + until the first hook arrives. An exited agent's history stays readable. - `complete.outcome` is the adapter's turn end: `interrupted` when the turn ended interrupted (Esc, a deny at a permission prompt, Codex's `turn_aborted`), else `done`. A remote prompt's `complete` follows its - completed receipt and carries the outcome the broker remembered for that - request; a duplicate `submit` after a broker restart (the receipt says - only "completed") gets `done`. + completed receipt, whose `detail` is the outcome (`"done"` or + `"interrupted"`, journaled as that word only), so a duplicate `submit` + after a broker restart gets the real outcome too. - `opened.unavailable` names what this agent's chat can't provide, so a client says so instead of waiting for it: `streaming` (no partial text: messages arrive whole), `reasoning`, `error_details` (no model/API error @@ -431,6 +461,8 @@ on the existing authenticated Astation connection. | Transcript format changed (required fields missing) | chat withdrawn for this agent, with the reason (open conversations get `error` and `close`); terminal unaffected | | No acceptance within 10 s of a paste | `error` "delivery unknown"; never resent | | Astation disconnects / re-authenticates | conversations close; Claude keeps running; a new `open` restores status and history | +| The worker dropped chat items before they reached the broker (its backlog overflowed in one read, `chat.gap > 0`) | the broker asks the worker for its whole retained history (`request_chat_history` → `chat_history_part`s + `chat_history`), replaces its own with it (keeping its newer approval card states) and catches up with the worker's chat state carried in `chat_history` (status, current and last completed turn with its outcome, input mark: a turn end it missed completes then, and so does its prompt's receipt), then open conversations get `error` "the chat history was incomplete (some items were dropped) and was reloaded; reopen the chat" (no `code`, like the chat's other own errors) and `close`; a reopen gets the complete history. Without a worker to ask, when the worker goes away before answering, or when the reload exceeds 1000 messages (all its parts are dropped, the history stays as it was, never with a hole), they get the same `error` and `close` | +| Broker restart or adoption | the worker's `chat_snapshot` carries its whole retained history (500 messages / 8 MiB), older messages first in `chat_history_part` lines of at most 1 MiB each, approval cards included (the broker copies every card to the worker, `chat_card`); a card still `pending` becomes `closed` with the note "Atem's runtime restarted; answer in the terminal" (the new broker released its approval). Prompt acceptance in flight is not carried | | Atem exits | the terminal and Claude end with it (as today); a new Atem has a new generation | ## Testing diff --git a/designs/protocol/atem-host-handoff.md b/designs/protocol/atem-host-handoff.md index 2e15a04..bd273af 100644 --- a/designs/protocol/atem-host-handoff.md +++ b/designs/protocol/atem-host-handoff.md @@ -156,6 +156,30 @@ chat-v1: session receipt. - At most 8 open conversations per session: a ninth `open` closes the oldest (`close`). +- History pages (fix/chat-history, 2026-10-17): `open{before}` on a + conversation the attachment already has open sends no `opened` and + resets nothing; the older page arrives as `message{history:true}` at the + conversation's next sequence numbers (the live sequence continues). Merge + paged messages by `messageId`/history order, not by `sequence`. Every + page ends with a sequenced `status` carrying `page: {before, count, + start}` (`start: true`: the oldest retained message was reached; an empty + page is the marker alone, `count: 0, start: true`). Cards are left out of + a view-only attachment's pages before the limits. `open` may carry the + `generation` of its `before`. A `before` that isn't retained, or named + with another `generation`, is `error{code:"stale_target"}` ("that message is no + longer in the history; reopen without before to get the newest page"): + sequenced on an open conversation (it stays open), without + `generation`/`sequence` on a new one (nothing opened). `open` without + `before` on an open conversation still resynchronizes it (`opened` at 0). +- When the worker dropped chat items before the broker got them, the broker + reloads the worker's whole history and chat state (a turn end it missed + completes then, with its receipt), then open conversations get `error` + "the chat history was incomplete (some items were dropped) and was + reloaded; reopen the chat" and `close`. Reopen: the history is complete. +- After a broker restart or adoption the chat history is whole (500 + messages / 8 MiB) and keeps its approval cards; a card still pending + becomes `closed` with the note "Atem's runtime restarted; answer in the + terminal". - Claude: `capabilities: ["terminal-v1","chat-v1"]`. Codex has chat-v1 only with Atem's Codex hooks on (`[codex] hooks = true`; else `["terminal-v1"]`, and a `prompt` there is `rejected` with "chat is not @@ -174,7 +198,9 @@ Receipts (session-v1 `submit`), per kind: - `prompt`: pasted into the agent once its chat accepts it (status `ready`, no local draft), completed when the chat reports the turn's end - (`turn_id`). On a session without chat (Codex with its hooks off) it is + (`turn_id`), with `detail` `"done"` or `"interrupted"` (the turn's + outcome, as chat `complete.outcome`; journaled, so it survives a broker + restart). On a session without chat (Codex with its hooks off) it is `rejected` with no effect. - `interrupt`: writes Esc only while the chat status says a turn runs (`working`/`waiting`) and completes when it ends; in `ready` or `unknown` diff --git a/designs/protocol/atem-session-v1.md b/designs/protocol/atem-session-v1.md index 07f3601..ee5a6a3 100644 --- a/designs/protocol/atem-session-v1.md +++ b/designs/protocol/atem-session-v1.md @@ -7,6 +7,8 @@ process and answers a later byte-equal request itself (`auto_allowed`, no approval, no push, no lease). Attachments with `control` get the active rules (`approval_rules`) and revoke one with a `kind:"approval_rule_revoke"` submission. New fixture: `approval-rule-revoke-needs-the-lease-and-an-active-rule`. +Clarified the same day (no new field): a `completed` `prompt` receipt's +`detail` is its turn's outcome, `"done"` or `"interrupted"` (§6, §7). r7.1 (2026-10-17) lets a second `hello` on a welcomed attachment replace it instead of refusing it `unauthorized` (§3, compatibility mode): after a client reset (e.g. the Mac slept) the client says `hello` again on @@ -323,7 +325,7 @@ Kinds and payloads (v1): | kind | payload | completes when | |---|---|---| -| `prompt` | `{"text": string}` | the provider confirms the turn ended (chat-v1 `complete`) | +| `prompt` | `{"text": string}` | the provider confirms the turn ended (chat-v1 `complete`); `detail` is the turn's outcome, `"done"` or `"interrupted"` | | `interrupt` | `{}` | the provider reports the turn interrupted, or the turn was already over (`detail`) | | `stop` | `{}` | the agent process exits | | `approval` | `{"approval_id", "decision", "rule_id"?, "message"?}` (§6a) | the agent reports it applied the decision | @@ -648,7 +650,8 @@ Errors are answered in the family of the request (ruling i): A chat `submit` hears about its request only through chat-v1 ops mapped from its receipt, never a session `receipt`: `accepted`/`executing` → `accepted{requestId}`; `completed` → `complete{requestId, turnId?, - outcome:"done"}`; `rejected` → `error{requestId, code:"rejected", text: + outcome}` (`outcome` is the receipt's `detail` for a prompt, `"done"` + or `"interrupted"`; `"done"` for the other kinds); `rejected` → `error{requestId, code:"rejected", text: detail}`; `outcome_unknown` → `error{requestId, code:"outcome_unknown", text: detail}`. Refusals at admission use the §7 code. - Removed forms (§0) and requests that don't decode — of any family — get a diff --git a/src/chat/session.rs b/src/chat/session.rs index b1e7e44..c06aaf7 100644 --- a/src/chat/session.rs +++ b/src/chat/session.rs @@ -12,8 +12,9 @@ use serde::{Deserialize, Serialize}; use std::collections::{HashMap, VecDeque}; use std::time::{Duration, Instant}; -const MAX_HISTORY_MESSAGES: usize = 500; -const MAX_HISTORY_BYTES: usize = 8 * 1024 * 1024; +/// History kept per session (oldest evicted first; the newest always stays). +pub const MAX_HISTORY_MESSAGES: usize = 500; +pub const MAX_HISTORY_BYTES: usize = 8 * 1024 * 1024; const MAX_REQUESTS: usize = 1000; const MAX_MESSAGE_BYTES: usize = 131_072; const MAX_PROMPT_BYTES: usize = 32_768; @@ -157,6 +158,9 @@ pub struct ChatSnapshot { pub input_mark: u64, pub current_turn: Option, pub last_completed: Option, + /// How `last_completed` ended. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_outcome: Option, pub history: Vec, } @@ -177,6 +181,7 @@ pub struct ChatSession { input_mark: u64, current_turn: Option<(String, Option)>, last_completed: Option, + last_outcome: Option, } /// Redact credential values (names are kept). See `memory::secrets::redact_secrets`. @@ -303,6 +308,7 @@ impl ChatSession { input_mark: 0, current_turn: None, last_completed: None, + last_outcome: None, } } @@ -322,6 +328,7 @@ impl ChatSession { input_mark: self.input_mark, current_turn: self.current_turn.as_ref().map(|(t, _)| t.clone()), last_completed: self.last_completed.clone(), + last_outcome: self.last_outcome, history: self.history(None, max_messages, max_bytes), } } @@ -339,6 +346,7 @@ impl ChatSession { session.input_mark = snapshot.input_mark; session.current_turn = snapshot.current_turn.map(|t| (t, None)); session.last_completed = snapshot.last_completed; + session.last_outcome = snapshot.last_outcome; for message in snapshot.history { session.push_message(message); } @@ -349,6 +357,107 @@ impl ChatSession { &self.status } + /// `message_id` is still in the retained history of this generation (a + /// valid `before`); `generation`, when the client names one, must be + /// this session's. + pub fn retains(&self, message_id: &str, generation: Option<&str>) -> bool { + generation.is_none_or(|g| g == self.generation) + && self.history.iter().any(|m| m.message_id == message_id) + } + + /// Catches up with the worker's `snapshot` state (its history replaced + /// this one's after items were dropped on the way): the turn the worker + /// saw end completes here too (with its outcome), and status, the + /// current turn and the input mark follow the worker's. + pub fn catch_up(&mut self, snapshot: &ChatSnapshot) -> Vec { + let mut out = Vec::new(); + if self.exited { + return out; + } + self.started |= snapshot.started; + if let Some(done) = &snapshot.last_completed + && self.last_completed.as_ref() != Some(done) + { + let outcome = snapshot.last_outcome.unwrap_or(Outcome::Done); + self.complete(done.clone(), outcome, &mut out); + } + match &snapshot.current_turn { + Some(t) if self.current_turn.as_ref().is_none_or(|(c, _)| c != t) => { + self.current_turn = Some((t.clone(), None)); + } + _ => {} + } + self.input_mark = snapshot.input_mark; + self.set_status( + snapshot.state, + snapshot.detail.as_deref(), + snapshot.since_ms, + &mut out, + ); + out + } + + /// The whole retained history replaced by `history` (the worker's, after + /// its backlog dropped items); status, turns and requests stay. + pub fn replace_history(&mut self, history: Vec) { + self.history.clear(); + self.history_bytes = 0; + for message in history { + self.push_message(message); + } + } + + /// An approval card message the broker made (`approval_card`), kept in + /// this (the worker's) history so a later broker gets it with the + /// snapshot: replaces the message with its id, or is appended. Anything + /// but one assistant message with exactly one approval part is ignored. + pub fn put_card(&mut self, message: ChatMessage) -> bool { + let card = message.role == Role::Assistant + && message.message_id.starts_with("approval-") + && matches!(message.parts.as_slice(), [Part::Approval(_)]); + if !card { + return false; + } + if let Some(m) = self + .history + .iter_mut() + .find(|m| m.message_id == message.message_id) + { + let old = m.bytes(); + *m = message; + let new = m.bytes(); + self.history_bytes = self.history_bytes - old + new; + return true; + } + self.push_message(message); + true + } + + /// Every card still `pending` becomes `closed` with `note` (a broker + /// restart released its approval): the changed messages. + pub fn close_pending_cards(&mut self, note: &str) -> Vec { + let mut changed = Vec::new(); + for m in self.history.iter_mut() { + let old = m.bytes(); + let mut touched = false; + for part in m.parts.iter_mut() { + if let Part::Approval(card) = part + && card.state == crate::chat::adapter::ApprovalCardState::Pending + { + card.state = crate::chat::adapter::ApprovalCardState::Closed; + card.note = Some(note.to_string()); + touched = true; + } + } + if touched { + let new = m.bytes(); + self.history_bytes = self.history_bytes - old + new; + changed.push(m.clone()); + } + } + changed + } + pub fn heartbeat(&self) -> Out { Out::Status(self.status.clone()) } @@ -500,6 +609,7 @@ impl ChatSession { /// compaction) completes nothing a client asked for: no `Complete`. fn complete(&mut self, turn_id: String, outcome: Outcome, out: &mut Vec) { self.last_completed = Some(turn_id.clone()); + self.last_outcome = Some(outcome); let (prompted, request_id) = match &self.current_turn { Some((t, r)) if *t == turn_id => (true, r.clone()), _ => (false, None), @@ -875,28 +985,40 @@ impl ChatSession { max_messages: usize, max_bytes: usize, ) -> Vec { + self.page(before, max_messages, max_bytes, |_| true).0 + } + + /// `history` over the messages `keep` accepts only (the limits count + /// those), and whether the page reaches the oldest retained one of them + /// (the start: nothing older to ask for). + pub fn page( + &self, + before: Option<&str>, + max_messages: usize, + max_bytes: usize, + keep: impl Fn(&ChatMessage) -> bool, + ) -> (Vec, bool) { let end = match before { None => self.history.len(), Some(id) => match self.history.iter().position(|m| m.message_id == id) { Some(i) => i, - None => return Vec::new(), + None => return (Vec::new(), true), }, }; let mut picked = Vec::new(); let mut bytes = 0usize; - for m in self.history.iter().take(end).rev() { - if picked.len() >= max_messages { - break; - } + let mut start = true; + for m in self.history.iter().take(end).rev().filter(|m| keep(m)) { let size = m.bytes(); - if !picked.is_empty() && bytes + size > max_bytes { + if picked.len() >= max_messages || (!picked.is_empty() && bytes + size > max_bytes) { + start = false; break; } bytes += size; picked.push(m.clone()); } picked.reverse(); - picked + (picked, start) } /// Content that was already in a log the adapter switched to (a resume of @@ -1701,6 +1823,94 @@ mod tests { assert_eq!(ids(s.history(None, 10, 3)), ["u4"]); } + #[test] + fn a_page_leaves_cards_out_before_its_limits_and_says_when_it_starts() { + use crate::chat::adapter::{ApprovalCard, ApprovalCardState}; + let mut s = ready(); + for i in 0..3 { + s.on_log(user(&format!("u{i}"), &format!("t{i}"), "hi"), 2, 0); + } + for i in 0..60 { + s.approval_card( + ApprovalCard { + id: "approval".into(), + approval_id: format!("a{i}"), + tool: "Bash".into(), + summary: "ls".into(), + always_rules: Vec::new(), + state: ApprovalCardState::Pending, + rule_id: None, + note: None, + }, + 3, + ); + } + let ids = |v: Vec| v.into_iter().map(|m| m.message_id).collect::>(); + let no_cards = |m: &ChatMessage| !m.message_id.starts_with("approval-"); + // A view-only page: the newest 50 are all cards, yet it holds the + // older messages, and it is the start. + let (page, start) = s.page(None, 50, 1 << 20, no_cards); + assert_eq!(ids(page), ["u0", "u1", "u2"]); + assert!(start); + let (page, start) = s.page(Some("approval-a30"), 2, 1 << 20, no_cards); + assert_eq!(ids(page), ["u1", "u2"]); + assert!(!start, "u0 is older"); + let (page, start) = s.page(Some("u0"), 50, 1 << 20, no_cards); + assert!(page.is_empty() && start); + // With control the cards count. + let (page, start) = s.page(None, 50, 1 << 20, |_| true); + assert_eq!(page.len(), 50); + assert!(!start); + // A cursor of another generation is not retained. + assert!(s.retains("u1", None) && s.retains("u1", Some("g1"))); + assert!(!s.retains("u1", Some("g2")) && !s.retains("nope", None)); + } + + #[test] + fn catching_up_completes_a_turn_whose_end_was_dropped() { + let hook = |turn: &str, at: i64| AgentEvent::PromptSubmitted { + text: "go".into(), + turn_id: Some(turn.into()), + at_ms: Some(at), + }; + let mut s = ready(); + s.submit("r", "g1", "go", 0, false).unwrap(); + s.mark_pasted("r", Instant::now(), 1000); + s.on_event(hook("t1", 1010), 2, 0); + s.on_log( + LogItem::UserMessage { + id: "u1".into(), + turn_id: "t1".into(), + text: "go".into(), + created_at_ms: 1020, + }, + 3, + 0, + ); + assert_eq!(s.status().state, State::Working); + // The worker saw the turn end interrupted; that never arrived here. + let mut worker = s.snapshot(0, 0); + worker.state = State::Ready; + worker.since_ms = 50; + worker.current_turn = Some("t1".into()); + worker.last_completed = Some("t1".into()); + worker.last_outcome = Some(Outcome::Interrupted); + worker.input_mark = 4; + let o = s.catch_up(&worker); + assert!(o.contains(&Out::Complete { + turn_id: "t1".into(), + request_id: Some("r".into()), + outcome: Outcome::Interrupted + })); + assert!( + o.iter() + .any(|o| matches!(o, Out::Status(st) if st.state == State::Ready)) + ); + assert_eq!(s.status().state, State::Ready); + // Once is enough. + assert!(s.catch_up(&worker).is_empty()); + } + #[test] fn history_is_bounded_to_500_messages() { let mut s = ready(); diff --git a/src/chat/wire.rs b/src/chat/wire.rs index a55a150..9fbc737 100644 --- a/src/chat/wire.rs +++ b/src/chat/wire.rs @@ -23,6 +23,8 @@ pub struct ChatEnvelope { pub enum Inbound { Open { before: Option, + /// The generation the client's `before` came from (optional). + generation: Option, }, Submit { generation: String, @@ -77,7 +79,12 @@ pub fn parse_inbound(env: &ChatEnvelope) -> Result { Some(Value::String(s)) if s.len() <= MAX_ID => Some(s.clone()), Some(_) => return Err("invalid before".into()), }; - Ok(Inbound::Open { before }) + let generation = match env.body.get("generation") { + None | Some(Value::Null) => None, + Some(Value::String(s)) if s.len() <= MAX_ID => Some(s.clone()), + Some(_) => return Err("invalid generation".into()), + }; + Ok(Inbound::Open { before, generation }) } "submit" => { let generation = str_field(env, "generation")?; @@ -581,12 +588,16 @@ mod tests { fn parse_each_op() { assert_eq!( parse_inbound(&env("open", json!({}))).unwrap(), - Inbound::Open { before: None } + Inbound::Open { + before: None, + generation: None + } ); assert_eq!( - parse_inbound(&env("open", json!({"before":"m9"}))).unwrap(), + parse_inbound(&env("open", json!({"before":"m9","generation":"g"}))).unwrap(), Inbound::Open { - before: Some("m9".into()) + before: Some("m9".into()), + generation: Some("g".into()) } ); assert_eq!( diff --git a/src/runtime/adopt.rs b/src/runtime/adopt.rs index ddc0a6a..8a5db7e 100644 --- a/src/runtime/adopt.rs +++ b/src/runtime/adopt.rs @@ -307,6 +307,16 @@ impl WorkerHandle { }) } + /// A copy of an approval card message for the worker's chat history. + pub fn chat_card(&self, message: crate::chat::session::ChatMessage) -> Result<()> { + self.send(&BrokerMsg::ChatCard { message }) + } + + /// Asks for the worker's whole retained chat history (`chat_history`). + pub fn request_chat_history(&self) -> Result<()> { + self.send(&BrokerMsg::RequestChatHistory) + } + /// After adoption: the approvals this broker knows (the rest released). pub fn known_approvals(&self, request_ids: Vec) -> Result<()> { self.send(&BrokerMsg::KnownApprovals { request_ids }) diff --git a/src/runtime/broker.rs b/src/runtime/broker.rs index 61d6a07..d75f38e 100644 --- a/src/runtime/broker.rs +++ b/src/runtime/broker.rs @@ -63,6 +63,13 @@ const UNKNOWN_SIZE: (u16, u16) = (24, 80); pub const CHAT_WITHDRAWN: &str = "chat is not available for this agent"; /// chat-v1 conversations must be reopened: another conversation was resumed. pub const CHAT_SWITCHED: &str = "the conversation was switched; reopen the chat"; +/// chat-v1 conversations must be reopened: the worker dropped chat items +/// before they reached the broker, which reloaded the history from it. +pub const CHAT_INCOMPLETE: &str = + "the chat history was incomplete (some items were dropped) and was reloaded; reopen the chat"; +/// The note on a card still pending when a new broker took the session: it +/// released every approval, so the agent asks in its terminal again. +pub const CARD_RELEASED: &str = "Atem's runtime restarted; answer in the terminal"; /// Who a write is for: local ones are reported in `Pumped`, the session /// host's (remote input, prompts, interrupts) through `take_results`. @@ -183,6 +190,13 @@ pub struct BrokerSession { pub activity_detail: Option, /// The worker's approval reports, not taken by the session host yet. approval_events: Vec, + /// Older history from `chat_history_part`s, until the `chat_snapshot` + /// or `chat_history` they belong to. + history_parts: Vec, + /// `request_chat_history` was sent and not answered yet. + history_requested: bool, + /// The `chat_history_part`s went over their bound and were dropped. + history_overflow: bool, } /// What one pass over a worker's messages changed. @@ -244,6 +258,9 @@ impl BrokerSession { activity: Activity::Unknown, activity_detail: None, approval_events: Vec::new(), + history_parts: Vec::new(), + history_requested: false, + history_overflow: false, } } @@ -414,6 +431,20 @@ impl BrokerSession { std::mem::take(&mut self.chat_out) } + /// An approval card in the chat history (`ChatSession::approval_card`), + /// copied to the worker so it survives this broker. + pub fn approval_card( + &mut self, + card: crate::chat::adapter::ApprovalCard, + created_at_ms: i64, + ) -> Out { + let out = self.chat.approval_card(card, created_at_ms); + if let (Out::Message(message), Some(worker)) = (&out, &self.worker) { + let _ = worker.chat_card(message.clone()); + } + out + } + /// Why open chat-v1 conversations must end, once. pub fn take_chat_reset(&mut self) -> Option<&'static str> { self.chat_reset.take() @@ -662,6 +693,12 @@ impl BrokerSession { Err(TryRecvError::Empty) => break, Err(TryRecvError::Disconnected) => { self.worker = None; + // A reload it never answered: conversations reopen on + // what is here. + if std::mem::take(&mut self.history_requested) { + self.history_parts.clear(); + self.chat_reset = Some(CHAT_INCOMPLETE); + } if self.state == SessionState::Running { eprintln!( "atem runtime: the worker of {} went away without an exit", @@ -714,7 +751,7 @@ impl BrokerSession { chat_v1, reason, unavailable, - gap: _, + gap, } => { // The App's order: events, the log, then turn ends. let now = now_ms(); @@ -746,6 +783,71 @@ impl BrokerSession { self.chat_v1 = chat_v1; self.chat_reason = reason; self.chat_unavailable = unavailable; + if gap > 0 { + // The worker's history has what was dropped here: reload + // it, then conversations reopen (`chat_history`). Without + // a worker to ask they reopen now, on what is here. + eprintln!( + "atem runtime: {}: the worker dropped {gap} chat items; reloading the history", + self.id + ); + let asked = self.history_requested + || self + .worker + .as_ref() + .is_some_and(|w| w.request_chat_history().is_ok()); + self.history_requested = asked; + if !asked { + self.chat_reset = Some(CHAT_INCOMPLETE); + } + } + } + WorkerMsg::ChatHistoryPart { history } => { + // Bounded like the history it rebuilds; beyond that nothing + // of it is used (`history_overflow`), never a silent hole. + if self.history_overflow + || self.history_parts.len() + history.len() + > 2 * crate::chat::session::MAX_HISTORY_MESSAGES + { + self.history_parts.clear(); + self.history_overflow = true; + } else { + self.history_parts.extend(history); + } + } + WorkerMsg::ChatHistory { snapshot } => { + self.history_requested = false; + self.chat_reset = Some(CHAT_INCOMPLETE); + let mut snapshot = snapshot; + let mut all = std::mem::take(&mut self.history_parts); + if std::mem::take(&mut self.history_overflow) { + // Too large to rebuild: the history stays as it is + // (conversations still reopen); the state catches up. + eprintln!( + "atem runtime: {}: the worker's history was too large to reload", + self.id + ); + } else { + all.append(&mut snapshot.history); + // This broker's cards win over the worker's copies (an + // update may have crossed the request). + let cards: Vec<_> = self + .chat + .history(None, usize::MAX, usize::MAX) + .into_iter() + .filter(|m| m.message_id.starts_with("approval-")) + .collect(); + self.chat.replace_history(all); + for card in cards { + if self.chat.retains(&card.message_id, None) { + self.chat.put_card(card); + } + } + } + // A turn end dropped on the way completes here now (and its + // prompt's receipt), and the status follows the worker's. + let outs = self.chat.catch_up(&snapshot); + self.chat_out.extend(outs); } WorkerMsg::Activity { state, detail } => { if self.state == SessionState::Running { @@ -856,7 +958,27 @@ impl BrokerSession { } => { // The worker's chat session goes on here; open conversations // (none for a fresh broker) reload it. + let mut snapshot = snapshot; + let mut all = std::mem::take(&mut self.history_parts); + if std::mem::take(&mut self.history_overflow) { + // Too large to rebuild: only the newest part (older + // history is cut, as if evicted, never holed). + eprintln!( + "atem runtime: {}: the worker's snapshot was too large; kept its newest part", + self.id + ); + all.clear(); + } + all.append(&mut snapshot.history); + snapshot.history = all; self.chat = ChatSession::restore(self.process_gen.clone(), snapshot); + // This broker released every approval the worker held for + // the previous one: their cards say so (here and there). + for message in self.chat.close_pending_cards(CARD_RELEASED) { + if let Some(worker) = &self.worker { + let _ = worker.chat_card(message); + } + } if self.state != SessionState::Running { let outs = self.chat.on_process_exit(now_ms()); self.chat_out.extend(outs); diff --git a/src/runtime/broker_tests.rs b/src/runtime/broker_tests.rs index 1be1188..b337e2c 100644 --- a/src/runtime/broker_tests.rs +++ b/src/runtime/broker_tests.rs @@ -202,3 +202,239 @@ async fn the_first_chat_snapshot_names_the_chat_capabilities() { ); let _ = handle.stop(); } + +fn card( + approval_id: &str, + state: crate::chat::adapter::ApprovalCardState, +) -> crate::chat::adapter::ApprovalCard { + crate::chat::adapter::ApprovalCard { + id: "approval".into(), + approval_id: approval_id.into(), + tool: "Bash".into(), + summary: "npm test".into(), + always_rules: Vec::new(), + state, + rule_id: None, + note: None, + } +} + +/// Pumps until `done` holds (10 s at most). +async fn pump_while(session: &mut BrokerSession, mut done: impl FnMut(&mut BrokerSession) -> bool) { + let reached = tokio::time::timeout(Duration::from_secs(10), async { + loop { + session.pump(); + if done(session) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + assert!(reached.is_ok(), "not reached"); +} + +/// Lets the launched worker's first messages (its own snapshot) arrive. +async fn settle(session: &mut BrokerSession) { + tokio::time::sleep(Duration::from_millis(300)).await; + session.pump(); +} + +/// The worker's whole retained history (500 messages / 8 MiB) reaches a new +/// broker in lines under the bound, approval cards included; one still +/// pending is closed (the new broker released its approval). +#[tokio::test] +async fn the_chat_snapshot_carries_the_whole_history_and_the_cards() { + use crate::chat::adapter::{ApprovalCardState, LogItem}; + use crate::chat::session::{MAX_HISTORY_BYTES, MAX_HISTORY_MESSAGES}; + let mut chat = ChatSession::new("g".into(), 0); + chat.on_event( + AgentEvent::SessionStarted { + session_id: "s".into(), + log_path: PathBuf::from("/x"), + source: "startup".into(), + same_log: false, + }, + 1, + 0, + ); + for i in 0..MAX_HISTORY_MESSAGES - 2 { + chat.on_log( + LogItem::UserMessage { + id: format!("u{i}"), + turn_id: format!("t{i}"), + text: format!("{i:05}{}", "x".repeat(16_000)), + created_at_ms: 5, + }, + 2, + 0, + ); + } + chat.approval_card(card("a-done", ApprovalCardState::Denied), 6); + chat.approval_card(card("a-open", ApprovalCardState::Pending), 7); + let retained = chat.history(None, MAX_HISTORY_MESSAGES, MAX_HISTORY_BYTES); + assert_eq!(retained.len(), MAX_HISTORY_MESSAGES); + let lines = super::super::worker::snapshot_messages(&chat, 0, true, None, Vec::new()); + assert!(lines.len() > 1, "more than one line's worth"); + for line in &lines { + assert!( + super::super::worker_protocol::encode_worker(line).len() + < super::super::worker_protocol::MAX_LINE + ); + } + assert!(matches!(lines.last(), Some(WorkerMsg::ChatSnapshot { .. }))); + + let Setup { mut session, _dir } = launch("exec cat").await; + // The launched worker's own (empty) snapshot first. + settle(&mut session).await; + let mut pumped = Pumped::default(); + for line in lines { + session.apply(line, &mut pumped); + } + let history = session.chat.history(None, usize::MAX, usize::MAX); + assert_eq!(history.len(), MAX_HISTORY_MESSAGES); + assert_eq!(history[0].message_id, "u0"); + let state = |id: &str| { + history + .iter() + .find(|m| m.message_id == id) + .and_then(|m| match m.parts.as_slice() { + [crate::chat::adapter::Part::Approval(c)] => Some((c.state, c.note.clone())), + _ => None, + }) + }; + assert_eq!( + state("approval-a-done"), + Some((ApprovalCardState::Denied, None)) + ); + assert_eq!( + state("approval-a-open"), + Some((ApprovalCardState::Closed, Some(CARD_RELEASED.to_string()))) + ); + session.stop().unwrap(); +} + +/// A `chat` batch with a gap (the worker dropped items before they reached +/// the broker): the broker reloads the whole history from the worker, which +/// brings back what the broker missed; the broker's own card states win +/// over the worker's copies; open conversations must reopen. +#[tokio::test] +async fn a_chat_gap_reloads_the_history_from_the_worker() { + use crate::chat::adapter::{ApprovalCardState, Part}; + let Setup { mut session, _dir } = launch("exec cat").await; + settle(&mut session).await; + let _ = session.take_chat_reset(); + // a1 goes to both; the broker then denies it (the copy to the worker + // is "in flight", i.e. never sent here). + session.approval_card(card("a1", ApprovalCardState::Pending), 5); + session + .chat + .approval_card(card("a1", ApprovalCardState::Denied), 6); + // What the broker missed: in the worker's history only. + let mut missed = ChatSession::new("x".into(), 0); + let Out::Message(missed) = missed.approval_card(card("a0", ApprovalCardState::Expired), 4) + else { + panic!("a message"); + }; + session.worker.as_ref().unwrap().chat_card(missed).unwrap(); + let mut pumped = Pumped::default(); + session.apply( + WorkerMsg::Chat { + events: Vec::new(), + log: Vec::new(), + input_count: 0, + chat_v1: true, + reason: None, + unavailable: Vec::new(), + gap: 3, + }, + &mut pumped, + ); + assert_eq!( + session.take_chat_reset(), + None, + "only once the history is back" + ); + pump_while(&mut session, |s| s.chat_reset.is_some()).await; + assert_eq!(session.take_chat_reset(), Some(CHAT_INCOMPLETE)); + let history = session.chat.history(None, usize::MAX, usize::MAX); + let states: Vec<(String, ApprovalCardState)> = history + .iter() + .map(|m| match m.parts.as_slice() { + [Part::Approval(c)] => (m.message_id.clone(), c.state), + _ => panic!("{m:?}"), + }) + .collect(); + assert_eq!( + states, + [ + ("approval-a1".to_string(), ApprovalCardState::Denied), + ("approval-a0".to_string(), ApprovalCardState::Expired) + ], + "the missed card is back; the broker's a1 state wins" + ); + session.stop().unwrap(); +} + +/// A worker that goes away while a reload is pending: conversations reopen +/// on what the broker has, nothing waits for an answer. +#[tokio::test] +async fn a_worker_lost_during_a_reload_ends_the_conversations() { + let Setup { mut session, _dir } = launch("sleep 1; exit 0").await; + let Out::Message(part) = ChatSession::new("x".into(), 0).approval_card( + card("a9", crate::chat::adapter::ApprovalCardState::Expired), + 1, + ) else { + panic!("a message"); + }; + session.history_requested = true; + session.history_parts.push(part); + // The agent exits, the worker is released and its connection closes. + pump_while(&mut session, |s| s.worker.is_none()).await; + assert!(!session.history_requested && session.history_parts.is_empty()); + assert_eq!(session.take_chat_reset(), Some(CHAT_INCOMPLETE)); +} + +/// Parts beyond their bound are all dropped, and the reload falls back to +/// "incomplete" on what is here (never a history with a hole); the state +/// still catches up. +#[tokio::test] +async fn an_oversized_reload_keeps_the_history_and_says_it_is_incomplete() { + use crate::chat::adapter::ApprovalCardState; + let Setup { mut session, _dir } = launch("exec cat").await; + settle(&mut session).await; + let _ = session.take_chat_reset(); + session.approval_card(card("keep", ApprovalCardState::Denied), 1); + let mut source = ChatSession::new("x".into(), 0); + let mut many = Vec::new(); + for i in 0..(2 * crate::chat::session::MAX_HISTORY_MESSAGES + 1) { + if let Out::Message(m) = + source.approval_card(card(&format!("p{i}"), ApprovalCardState::Expired), 1) + { + many.push(m); + } + } + let mut pumped = Pumped::default(); + session.history_requested = true; + for chunk in many.chunks(100) { + session.apply( + WorkerMsg::ChatHistoryPart { + history: chunk.to_vec(), + }, + &mut pumped, + ); + } + let mut snapshot = source.snapshot(0, 0); + snapshot.history = Vec::new(); + session.apply(WorkerMsg::ChatHistory { snapshot }, &mut pumped); + assert_eq!(session.take_chat_reset(), Some(CHAT_INCOMPLETE)); + assert!(session.history_parts.is_empty() && !session.history_overflow); + let ids: Vec = session + .chat + .history(None, usize::MAX, usize::MAX) + .into_iter() + .map(|m| m.message_id) + .collect(); + assert_eq!(ids, ["approval-keep"]); + session.stop().unwrap(); +} diff --git a/src/runtime/host_bridge.rs b/src/runtime/host_bridge.rs index 1edd78b..a5dd688 100644 --- a/src/runtime/host_bridge.rs +++ b/src/runtime/host_bridge.rs @@ -72,8 +72,8 @@ use crate::chat::wire::{self as chat_wire, ChatEnvelope, Inbound}; use crate::session::{Provider, SessionId}; use crate::session_protocol::file_journal::FileJournal; use crate::session_protocol::host::{ - Attach, ChatReply, DispatchReport, Effect, HostConfig, HostContext, LEASE_EPOCH_BLOCK, Origin, - SessionHost, SessionSpec, random_ids, + Attach, ChatReply, DETAIL_TURN_DONE, DETAIL_TURN_INTERRUPTED, DispatchReport, Effect, + HostConfig, HostContext, LEASE_EPOCH_BLOCK, Origin, SessionHost, SessionSpec, random_ids, }; use crate::session_protocol::wire::{ self, Activity, ApprovalCloseReason, ApprovalDecision, ApprovalRequest, ClientKind, Envelope, @@ -105,6 +105,9 @@ const HISTORY_BYTES: usize = 256 * 1024; const MAX_CONVERSATIONS: usize = 8; const RESTARTED: &str = "the agent was restarted; reopen the chat"; const NOT_AVAILABLE: &str = "chat is not available for this agent"; +/// `open{before}` naming a message no longer retained (`stale_target`). +const UNKNOWN_CURSOR: &str = + "that message is no longer in the history; reopen without before to get the newest page"; const NO_TURN: &str = "no turn was running"; /// An interrupt on a session without chat: nothing knows whether a turn runs. const NO_TURN_STATE: &str = "no turn state; use the terminal"; @@ -128,8 +131,6 @@ const RUNTIME_EPOCH_FILE: &str = "runtime_epoch"; const CAPTURE_CHECK: Duration = Duration::from_secs(5); /// Session messages remembered for duplicate detection, at most. const MAX_SEEN: usize = 4096; -/// Turn outcomes kept for duplicate submits (cleared beyond it). -const MAX_OUTCOMES: usize = 4096; /// The broker's monotonic clock in milliseconds, and wall time (tests move /// either forward). @@ -380,8 +381,6 @@ pub struct HostBridge { /// terminal-v1 messages in order (frames, acks, errors) waiting for the /// bounded terminal queue. terminal_out: VecDeque, - /// The outcome a prompt's turn ended with, until its receipt is reported. - outcomes: HashMap, /// approval_id → open approvals. approvals: HashMap, /// Rule revocations sent to a worker: request id → (session, rule id). @@ -433,7 +432,6 @@ impl HostBridge { interrupts: HashMap::new(), pending: HashMap::new(), terminal_out: VecDeque::new(), - outcomes: HashMap::new(), approvals: HashMap::new(), revokes: HashMap::new(), pushes: Default::default(), @@ -1637,7 +1635,7 @@ impl HostBridge { let Some(session) = session_of(cx, sid) else { return; }; - let out = session.chat.approval_card(card, wall_ms() as i64); + let out = session.approval_card(card, wall_ms() as i64); self.broadcast_to(sid, &out, None, true, cx); } @@ -1933,7 +1931,7 @@ impl HostBridge { let Some(session) = session_of(cx, &sid) else { return; }; - let out = session.chat.approval_card(card, wall_ms() as i64); + let out = session.approval_card(card, wall_ms() as i64); self.broadcast_to(&sid, &out, None, true, cx); } @@ -2062,11 +2060,12 @@ impl HostBridge { }; let conversation = env.conversation_id.clone(); match chat_wire::parse_inbound(&env) { - Ok(Inbound::Open { before }) => self.open_conversation( + Ok(Inbound::Open { before, generation }) => self.open_conversation( attachment, sid, &conversation, before.as_deref(), + generation.as_deref(), cx, ), Ok(Inbound::Close) => { @@ -2102,6 +2101,7 @@ impl HostBridge { sid: &str, conversation: &str, before: Option<&str>, + before_generation: Option<&str>, cx: &mut Cx, ) { let Some(session) = session_of(cx, sid) else { @@ -2121,6 +2121,69 @@ impl HostBridge { // Approval cards and session rules only for an attachment that may // answer them. let control = self.host.attachment_can(attachment, Permission::Control); + // An open conversation of this attachment pages: the older page + // continues its live sequence (no `opened`, nothing reset). + let paging = before.is_some() + && self + .chats + .get(sid) + .and_then(|c| c.conversations.get(conversation)) + .is_some_and(|c| c.attachment == attachment); + // A cursor no longer retained (evicted, another generation, never + // sent) is refused, never answered with an empty page that would + // look like the start of history. + if let Some(id) = before + && !session.chat.retains(id, before_generation) + { + let head = if paging { + self.next_head(sid, conversation, cx) + } else { + None + }; + let env = chat_error( + conversation, + sid, + head, + None, + "stale_target", + UNKNOWN_CURSOR, + ); + self.send_chat(env, cx); + return; + } + // Cards are left out before the limits (a view-only page of cards + // still pages on to older messages). + let keep = |m: &crate::chat::session::ChatMessage| control || !is_card(m); + if paging { + let (page, start) = session + .chat + .page(before, HISTORY_MESSAGES, HISTORY_BYTES, keep); + let status = Out::Status(session.chat.status().clone()); + let count = page.len(); + for message in page { + if let Some((generation, sequence)) = self.next_head(sid, conversation, cx) { + let env = chat_wire::history_message( + conversation, + sid, + &generation, + sequence, + &message, + ); + self.send_chat(env, cx); + } + } + // Every page ends with a sequenced `status` naming it, so an + // empty page (the start) is an answer too, never silence. + if let Some((generation, sequence)) = self.next_head(sid, conversation, cx) { + let mut env = chat_wire::event(conversation, sid, &generation, sequence, &status); + env.body.insert( + "page".into(), + json!({"before": before, "count": count, "start": start}), + ); + self.send_chat(env, cx); + } + return; + } let rules = if control { self.host.approval_rules(sid) } else { @@ -2134,11 +2197,11 @@ impl HostBridge { &session.chat_unavailable, &rules, )]; - let history = session + let (history, _) = session .chat - .history(before, HISTORY_MESSAGES, HISTORY_BYTES); + .page(before, HISTORY_MESSAGES, HISTORY_BYTES, keep); let mut sequence = 1; - for message in history.iter().filter(|m| control || !is_card(m)) { + for message in history.iter() { envelopes.push(chat_wire::history_message( conversation, sid, @@ -2214,15 +2277,15 @@ impl HostBridge { ChatReply::Complete { request_id, turn_id, + interrupted, } => { self.submitters.remove(&request_id); - // A turn ended interrupted says so (a duplicate after a - // restart only knows it completed). - let outcome = self - .outcomes - .get(&request_id) - .copied() - .unwrap_or(crate::chat::adapter::Outcome::Done); + // The receipt's detail keeps the outcome (also after a restart). + let outcome = if interrupted { + crate::chat::adapter::Outcome::Interrupted + } else { + crate::chat::adapter::Outcome::Done + }; complete_envelope(conversation, sid, head, request_id, turn_id, outcome) } ChatReply::Error { @@ -2267,12 +2330,14 @@ impl HostBridge { if let Some(request_id) = request_id { except = self.submitters.get(request_id).map(|(_, c)| c.clone()); if self.prompts.remove(request_id).is_some() { - if self.outcomes.len() >= MAX_OUTCOMES { - self.outcomes.clear(); - } - self.outcomes.insert(request_id.clone(), *outcome); + let detail = match outcome { + crate::chat::adapter::Outcome::Done => DETAIL_TURN_DONE, + crate::chat::adapter::Outcome::Interrupted => { + DETAIL_TURN_INTERRUPTED + } + }; let report = DispatchReport::Completed { - detail: None, + detail: Some(detail.to_string()), turn_id: Some(turn_id.clone()), }; let effects = self.host.dispatch_report(request_id, report); diff --git a/src/runtime/host_bridge_tests.rs b/src/runtime/host_bridge_tests.rs index 2e40272..7ecb327 100644 --- a/src/runtime/host_bridge_tests.rs +++ b/src/runtime/host_bridge_tests.rs @@ -796,6 +796,24 @@ fn chat_submit_completes_through_its_receipt_and_survives_a_broker_restart() { ); let receipts = peer.reply(&id).await; assert_eq!(receipts["receipts"][0]["state"], "completed", "{receipts}"); + assert_eq!(receipts["receipts"][0]["detail"], "done", "{receipts}"); + // The journal keeps no prompt text once the request left `accepted`. + let journal = rt.dir.path().join("state").join("journal"); + let mut files = 0; + let mut stack = vec![journal]; + while let Some(path) = stack.pop() { + if path.is_dir() { + stack.extend(std::fs::read_dir(&path).unwrap().map(|e| e.unwrap().path())); + } else { + files += 1; + let bytes = std::fs::read(&path).unwrap(); + let text = String::from_utf8_lossy(&bytes); + assert!(!text.contains("run the tests"), "{}", path.display()); + assert!(!text.contains("no lease"), "{}", path.display()); + } + } + assert!(files > 0, "no journal files"); + let peer = &mut rt.peer; let old_attachment = peer.attachment.clone(); // The broker restarts; its worker survives and is adopted. @@ -863,6 +881,152 @@ fn chat_submit_completes_through_its_receipt_and_survives_a_broker_restart() { }); } +/// The sequences `conversation` got, in arrival order (`opened` is 0). +fn sequences(seen: &[Value], conversation: &str) -> Vec { + seen.iter() + .filter(|v| v["type"] == "agentChat" && v["data"]["conversationId"] == conversation) + .filter_map(|v| v["data"]["sequence"].as_u64()) + .collect() +} + +#[test] +fn paging_older_history_continues_the_live_sequence_and_unknown_cursors_are_refused() { + local(async { + let mut rt = Runtime::start(true).await; + let mut client = rt.client().await; + let session = rt.launch(&mut client, "claude", &fake_claude(), None).await; + let (sid, pg) = (session.id.clone(), session.generation.clone()); + let peer = &mut rt.peer; + peer.hello().await; + listed(peer, &sid).await; + open_chat(peer, "conv-1", &sid, &pg).await; + let lease = peer.acquire(&sid, &pg).await; + for text in ["first", "second"] { + let rid = uuid::Uuid::new_v4().to_string(); + let events = submit_until_complete(peer, "conv-1", &sid, &pg, lease, &rid, text).await; + assert!(is_chat(events.last().unwrap(), "complete")); + } + peer.drain(Duration::from_millis(300)).await; + let live = sequences(&peer.seen, "conv-1"); + let last = *live.last().unwrap(); + + // An older page of the open conversation: no `opened`, the page at + // the next sequences, nothing reset, then its end marker (a + // `status` naming the page). + let from = peer.seen.len(); + peer.chat("open", "conv-1", &sid, &pg, json!({"before": "u-2"})); + let end = peer + .wait_from(from, |v| { + is_chat(v, "status") && v["data"].get("page").is_some() + }) + .await; + let page: Vec<&Value> = peer.seen[from..=end] + .iter() + .filter(|v| v["type"] == "agentChat" && v["data"]["conversationId"] == "conv-1") + .collect(); + assert!(!page.iter().any(|v| is_chat(v, "opened")), "{page:#?}"); + let ids: Vec<&str> = page + .iter() + .filter(|v| is_chat(v, "message")) + .map(|v| v["data"]["messageId"].as_str().unwrap_or("")) + .collect(); + assert_eq!(ids, ["u-1", "msg-1"], "{page:#?}"); + assert_eq!(page.len(), 3, "the page, then its marker: {page:#?}"); + assert!(page[..2].iter().all(|v| v["data"]["history"] == true)); + assert_eq!(page[0]["data"]["sequence"], last + 1); + assert_eq!(page[1]["data"]["sequence"], last + 2); + let marker = &page[2]["data"]; + assert_eq!(marker["sequence"], last + 3); + assert_eq!(marker["state"], "ready"); + assert_eq!( + marker["page"], + json!({"before": "u-2", "count": 2, "start": true}) + ); + + // Live events go on after the page. + let rid = uuid::Uuid::new_v4().to_string(); + submit_until_complete(peer, "conv-1", &sid, &pg, lease, &rid, "third").await; + peer.drain(Duration::from_millis(300)).await; + let all = sequences(&peer.seen, "conv-1"); + assert_eq!( + all, + (0..all.len() as u64).collect::>(), + "never reset or reused" + ); + + // A cursor that isn't retained is refused, never an empty page: in + // the open conversation (sequenced, it stays open) ... + let from = peer.seen.len(); + peer.chat("open", "conv-1", &sid, &pg, json!({"before": "nope"})); + let at = peer + .wait_from(from, |v| { + is_chat(v, "error") && v["data"]["conversationId"] == "conv-1" + }) + .await; + let error = peer.seen[at]["data"].clone(); + assert_eq!(error["code"], "stale_target", "{error}"); + assert_eq!(error["sequence"], all.len() as u64); + // ... and on a new one (not opened). + let from = peer.seen.len(); + peer.chat("open", "conv-2", &sid, &pg, json!({"before": "nope"})); + let at = peer + .wait_from(from, |v| { + v["type"] == "agentChat" && v["data"]["conversationId"] == "conv-2" + }) + .await; + let error = peer.seen[at]["data"].clone(); + assert_eq!(error["op"], "error", "{error}"); + assert_eq!(error["code"], "stale_target"); + assert!(error.get("sequence").is_none()); + // A cursor named with another generation is refused too. + let from = peer.seen.len(); + peer.chat( + "open", + "conv-1", + &sid, + &pg, + json!({"before": "u-2", "generation": "another-generation"}), + ); + let at = peer + .wait_from(from, |v| { + is_chat(v, "error") && v["data"]["conversationId"] == "conv-1" + }) + .await; + assert_eq!(peer.seen[at]["data"]["code"], "stale_target"); + // The oldest message as the cursor: an empty page, whose marker + // says it is the start. + let from = peer.seen.len(); + peer.chat( + "open", + "conv-1", + &sid, + &pg, + json!({"before": "u-1", "generation": pg}), + ); + let end = peer + .wait_from(from, |v| { + is_chat(v, "status") && v["data"].get("page").is_some() + }) + .await; + let got: Vec<&Value> = peer.seen[from..=end] + .iter() + .filter(|v| v["type"] == "agentChat" && v["data"]["conversationId"] == "conv-1") + .collect(); + assert_eq!(got.len(), 1, "only the marker: {got:#?}"); + assert_eq!( + got[0]["data"]["page"], + json!({"before": "u-1", "count": 0, "start": true}) + ); + let all = sequences(&peer.seen, "conv-1"); + assert_eq!( + all, + (0..all.len() as u64).collect::>(), + "never reset or reused" + ); + rt.shutdown().await; + }); +} + #[test] fn a_link_that_is_not_a_verified_pairing_gets_no_attachment() { local(async { @@ -2494,6 +2658,17 @@ fn codex_chat_first_prompt_local_then_remote_prompts_an_interrupt_and_an_approva .await; let at_done = peer.wait_from(from, completed(r3.clone())).await; assert_eq!(peer.seen[at_done]["data"]["outcome"], "interrupted"); + // The prompt's receipt carries the outcome too (journaled, no text). + let id = peer.request( + "receipt_query", + json!({"binding": peer.binding(&sid, &pg, None), "request_ids": [r3]}), + ); + let receipts = peer.reply(&id).await; + assert_eq!(receipts["receipts"][0]["state"], "completed", "{receipts}"); + assert_eq!( + receipts["receipts"][0]["detail"], "interrupted", + "{receipts}" + ); activity_until(peer, &sid, "ready").await; // An approval in a remote turn: waiting, its card, answered from diff --git a/src/runtime/worker.rs b/src/runtime/worker.rs index 2d640ef..481e724 100644 --- a/src/runtime/worker.rs +++ b/src/runtime/worker.rs @@ -53,9 +53,6 @@ pub const OUTPUT_BACKLOG: usize = 1024 * 1024; const HANDSHAKE: Duration = Duration::from_secs(2); /// The refusal a second broker gets while one is connected. pub const ANOTHER_BROKER: &str = "another broker is connected to this worker"; -/// History a chat snapshot carries at most (well below the line bound). -const SNAPSHOT_MESSAGES: usize = 200; -const SNAPSHOT_BYTES: usize = 1024 * 1024; /// How often the chat adapter is read. const CHAT_POLL: Duration = Duration::from_millis(200); /// Input after the one that released a waiting hook (which is consumed) is @@ -734,14 +731,16 @@ impl Worker { // The snapshot covers what the backlog held (the worker's chat // session read it already), so the backlog is not sent too. self.chat_backlog.take(); - let snapshot = WorkerMsg::ChatSnapshot { - snapshot: self.chat.snapshot(SNAPSHOT_MESSAGES, SNAPSHOT_BYTES), - input_count: self.input.input_count(), - chat_v1: self.capabilities.chat_v1, - reason: self.capabilities.reason.clone(), - unavailable: self.unavailable(), - }; - self.send(&snapshot); + let messages = snapshot_messages( + &self.chat, + self.input.input_count(), + self.capabilities.chat_v1, + self.capabilities.reason.clone(), + self.unavailable(), + ); + for message in &messages { + self.send(message); + } } // The session rules outlive a broker (the worker owns them), but // allow nothing until this broker confirms them. @@ -1119,6 +1118,14 @@ impl Worker { rules: rules.unwrap_or_default(), }); } + BrokerMsg::ChatCard { message } => { + self.chat.put_card(message); + } + BrokerMsg::RequestChatHistory => { + for message in history_messages(&self.chat) { + self.send(&message); + } + } } false } @@ -1410,6 +1417,58 @@ impl Worker { } } +/// The worker's whole retained chat history (500 messages / 8 MiB) cut +/// for the line bound: `chat_history_part`s with the older messages, then +/// `last` with the newest. +fn history_lines( + chat: &ChatSession, + last: impl FnOnce(Vec) -> WorkerMsg, +) -> Vec { + use super::worker_chat::{SNAPSHOT_CHUNK_BYTES, split_history}; + use crate::chat::session::{MAX_HISTORY_BYTES, MAX_HISTORY_MESSAGES}; + let history = chat.history(None, MAX_HISTORY_MESSAGES, MAX_HISTORY_BYTES); + let mut runs = split_history(history, SNAPSHOT_CHUNK_BYTES); + let newest = runs.pop().unwrap_or_default(); + let mut lines: Vec = runs + .into_iter() + .map(|history| WorkerMsg::ChatHistoryPart { history }) + .collect(); + lines.push(last(newest)); + lines +} + +/// The `chat_snapshot` a broker gets on connecting (with the parts ahead +/// of it): the chat's state and its whole retained history, approval +/// cards included. +pub(super) fn snapshot_messages( + chat: &ChatSession, + input_count: u64, + chat_v1: bool, + reason: Option, + unavailable: Vec, +) -> Vec { + let mut snapshot = chat.snapshot(0, 0); + history_lines(chat, move |history| { + snapshot.history = history; + WorkerMsg::ChatSnapshot { + snapshot, + input_count, + chat_v1, + reason, + unavailable, + } + }) +} + +/// The answer to `request_chat_history`. +pub(super) fn history_messages(chat: &ChatSession) -> Vec { + let mut snapshot = chat.snapshot(0, 0); + history_lines(chat, move |history| { + snapshot.history = history; + WorkerMsg::ChatHistory { snapshot } + }) +} + /// A write refused for its stale input epoch. #[derive(Debug)] struct StaleEpoch; diff --git a/src/runtime/worker_chat.rs b/src/runtime/worker_chat.rs index 06444d9..7924988 100644 --- a/src/runtime/worker_chat.rs +++ b/src/runtime/worker_chat.rs @@ -112,6 +112,34 @@ pub fn cap_event(mut event: AgentEvent) -> AgentEvent { event } +/// Bytes of history messages (JSON as sent) in one `chat_history_part` or +/// in the final `chat_snapshot`/`chat_history`: a quarter of the line bound, +/// so the largest message (131 072 bytes of text, escaped) always fits. +pub const SNAPSHOT_CHUNK_BYTES: usize = 1024 * 1024; + +/// `history` (oldest first) cut into runs of at most `max_bytes` of encoded +/// messages each, oldest run first; a single larger message is a run of its +/// own. Never empty: no history is one empty run. +pub fn split_history( + history: Vec, + max_bytes: usize, +) -> Vec> { + let mut runs = vec![Vec::new()]; + let mut bytes = 0; + for message in history { + let n = size(&message); + let run = runs.last_mut().expect("never empty"); + if !run.is_empty() && bytes + n > max_bytes { + runs.push(vec![message]); + bytes = n; + } else { + run.push(message); + bytes += n; + } + } + runs +} + #[derive(Debug)] struct Batch { events: Vec, diff --git a/src/runtime/worker_protocol.rs b/src/runtime/worker_protocol.rs index b8cd14d..a84e05f 100644 --- a/src/runtime/worker_protocol.rs +++ b/src/runtime/worker_protocol.rs @@ -166,6 +166,16 @@ pub enum BrokerMsg { ConfirmApprovalRules { rule_ids: Vec, }, + /// An approval card message the broker put in its chat history (made + /// and updated only there): the worker keeps a copy in its history, so + /// a later broker gets the card with the `chat_snapshot`. + ChatCard { + message: crate::chat::session::ChatMessage, + }, + /// The broker's history is incomplete (a `chat` came with `gap`) → + /// `chat_history_part`s, then `chat_history` with the worker's whole + /// retained history. + RequestChatHistory, } /// The identity and trust generation an Atem session rule was granted @@ -239,6 +249,8 @@ impl std::fmt::Debug for BrokerMsg { BrokerMsg::ConfirmApprovalRules { rule_ids } => { write!(f, "ConfirmApprovalRules({rule_ids:?})") } + BrokerMsg::ChatCard { message } => write!(f, "ChatCard({})", message.message_id), + BrokerMsg::RequestChatHistory => f.write_str("RequestChatHistory"), } } } @@ -338,6 +350,9 @@ pub enum WorkerMsg { /// On every connection, before any `chat`: the worker's chat session /// (authoritative; it read everything held back meanwhile), so a broker /// that adopts this worker goes on from its status and history. + /// `snapshot.history` holds the newest messages only when the whole + /// retained history doesn't fit one line: the older ones come first, in + /// `chat_history_part`s (oldest first) that the broker prepends. ChatSnapshot { snapshot: crate::chat::session::ChatSnapshot, input_count: u64, @@ -346,6 +361,18 @@ pub enum WorkerMsg { #[serde(default)] unavailable: Vec, }, + /// Older history ahead of a `chat_snapshot` or `chat_history`, at most + /// `worker_chat::SNAPSHOT_CHUNK_BYTES` of messages per line. + ChatHistoryPart { + history: Vec, + }, + /// Answers `request_chat_history`: the worker's chat state (status, + /// turns, input mark) with the newest messages of its retained history + /// in `snapshot.history` (older ones in the `chat_history_part`s before + /// it). + ChatHistory { + snapshot: crate::chat::session::ChatSnapshot, + }, Activity { state: Activity, detail: Option, diff --git a/src/session_protocol/approval_tests.rs b/src/session_protocol/approval_tests.rs index cf47faf..132becf 100644 --- a/src/session_protocol/approval_tests.rs +++ b/src/session_protocol/approval_tests.rs @@ -1353,7 +1353,8 @@ fn a_rule_revoked_from_chat_completes_in_chat() { ChatReply::Accepted { request_id: rid(7) }, ChatReply::Complete { request_id: rid(7), - turn_id: None + turn_id: None, + interrupted: false } ] ); @@ -1395,3 +1396,54 @@ fn a_trust_change_or_restart_clears_the_session_rules() { rt.apply(fx); assert!(rt.host.approval_rules(S1).is_empty()); } + +/// The outcome of a chat prompt's turn lives in its receipt's detail, so a +/// duplicate submit after a runtime restart still says it was interrupted. +#[test] +fn a_duplicate_chat_prompt_after_a_restart_keeps_its_interrupted_outcome() { + let mut rt = setup(); + let epoch = list_and_hold(&mut rt, "c1"); + let submit = |rt: &FakeRuntime, epoch: u64| { + json!({"type": "agentChat", "data": {"v": 1, "op": "submit", "conversationId": "conv", + "agentId": S1, "generation": P1, "requestId": rid(5), "text": "run it", + "binding": binding(rt, "c1", Some(epoch))}}) + }; + rt.send("c1", &submit(&rt, epoch)); + let fx = rt.host.dispatch_report( + &rid(5), + DispatchReport::Completed { + detail: Some(DETAIL_TURN_INTERRUPTED.into()), + turn_id: Some("t1".into()), + }, + ); + rt.apply(fx); + assert_eq!( + rt.host.receipt(&rid(5)).unwrap().detail.as_deref(), + Some("interrupted") + ); + let fx = rt + .host + .restart("99999999-9999-4999-8999-999999999999", 8) + .unwrap(); + rt.apply(fx); + rt.attach("c1", C1B).unwrap(); + let epoch = list_and_hold(&mut rt, "c1"); + rt.effects.clear(); + rt.send("c1", &submit(&rt, epoch)); + let chats: Vec = rt + .effects + .iter() + .filter_map(|e| match e { + Effect::Chat { reply, .. } => Some(reply.clone()), + _ => None, + }) + .collect(); + assert_eq!( + chats, + [ChatReply::Complete { + request_id: rid(5), + turn_id: None, + interrupted: true + }] + ); +} diff --git a/src/session_protocol/fake_runtime.rs b/src/session_protocol/fake_runtime.rs index 4c34639..7e5a520 100644 --- a/src/session_protocol/fake_runtime.rs +++ b/src/session_protocol/fake_runtime.rs @@ -144,8 +144,10 @@ fn render_chat(conversation_id: &str, session_id: &str, reply: ChatReply) -> Env ChatReply::Complete { request_id, turn_id, + interrupted, } => { - let mut b = json!({"op": "complete", "requestId": request_id, "outcome": "done"}); + let outcome = if interrupted { "interrupted" } else { "done" }; + let mut b = json!({"op": "complete", "requestId": request_id, "outcome": outcome}); if let Some(t) = turn_id { b["turnId"] = Value::String(t); } diff --git a/src/session_protocol/host.rs b/src/session_protocol/host.rs index dda7f9a..4293bbe 100644 --- a/src/session_protocol/host.rs +++ b/src/session_protocol/host.rs @@ -98,6 +98,10 @@ pub const DETAIL_STOP_FAILED: &str = "stop failed"; pub const DETAIL_STILL_RUNNING: &str = "process still running"; pub const DETAIL_UNRECONCILED: &str = "outcome could not be established"; pub const DETAIL_REPAIR: &str = "host counters exhausted; pair again"; +/// A completed `prompt` receipt's `detail`: its turn's outcome (chat-v1 +/// `complete.outcome`), kept in the journal as one of these two words. +pub const DETAIL_TURN_DONE: &str = "done"; +pub const DETAIL_TURN_INTERRUPTED: &str = "interrupted"; // ── Inputs and outputs ──────────────────────────────────────────────────── @@ -278,10 +282,12 @@ pub enum ChatReply { Accepted { request_id: String, }, - /// The turn ended (`outcome: "done"`); `turn_id` is the provider's turn. + /// The turn ended (`outcome: "done"`, or `"interrupted"` when + /// `interrupted`); `turn_id` is the provider's turn. Complete { request_id: String, turn_id: Option, + interrupted: bool, }, /// `code` is a §7 error code for refusals, or `rejected` / /// `outcome_unknown` for a final receipt; `text` its bounded detail. @@ -2836,6 +2842,8 @@ fn chat_reply(e: &Entry) -> ChatReply { ReceiptState::Completed => ChatReply::Complete { request_id: request_id.clone(), turn_id: e.turn_id.clone(), + interrupted: s.kind == ReceiptKind::Prompt + && s.detail.as_deref() == Some(DETAIL_TURN_INTERRUPTED), }, ReceiptState::Rejected => error("rejected"), ReceiptState::OutcomeUnknown => error("outcome_unknown"),