diff --git a/npm/core/src/generated/ClientMessage.ts b/npm/core/src/generated/ClientMessage.ts new file mode 100644 index 0000000..30974a8 --- /dev/null +++ b/npm/core/src/generated/ClientMessage.ts @@ -0,0 +1,34 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { CommandEnvelope } from "./CommandEnvelope"; +import type { KindRevision } from "./KindRevision"; +import type { ObserverSpec } from "./ObserverSpec"; +import type { StateLayer } from "./StateLayer"; + +/** + * Client → substrate frames. One union; the transport is fully + * self-describing and there is no out-of-band routing. + */ +export type ClientMessage = { "type": "subscribe", +/** + * Widget kinds to receive. Empty = none (explicit opt-in, + * consistent with [`ObserverSpec`]). + */ +kinds: Array, +/** + * Cadence layers to receive. Empty = none. + */ +layers: Array, +/** + * Revisions already held; substrate MAY skip matching + * snapshots. Empty = client holds nothing, send everything. + */ +last_seen: Array, } | { "type": "command" } & CommandEnvelope | { "type": "observe", +/** + * The observer's identity, budget, and perception scope. + */ +spec: ObserverSpec, +/** + * Revisions this observer already perceived; same skip rule + * as `Subscribe::last_seen`. + */ +last_seen: Array, }; diff --git a/npm/core/src/generated/KindRevision.ts b/npm/core/src/generated/KindRevision.ts new file mode 100644 index 0000000..21a0ee0 --- /dev/null +++ b/npm/core/src/generated/KindRevision.ts @@ -0,0 +1,17 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * A `(kind, revision)` pair the client already holds — sent on + * (re)subscribe so the substrate may skip redundant snapshots. + */ +export type KindRevision = { +/** + * Widget kind (matches [`StateEnvelope::kind`]). + */ +kind: string, +/** + * Last revision of that kind the client rendered. + * + * TS `number`, same rationale as [`StateEnvelope::revision`]. + */ +revision: number, }; diff --git a/npm/core/src/generated/ServerMessage.ts b/npm/core/src/generated/ServerMessage.ts new file mode 100644 index 0000000..2720855 --- /dev/null +++ b/npm/core/src/generated/ServerMessage.ts @@ -0,0 +1,7 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { StateEnvelope } from "./StateEnvelope"; + +/** + * Substrate → client frames. + */ +export type ServerMessage = { "type": "state" } & StateEnvelope; diff --git a/npm/core/src/index.ts b/npm/core/src/index.ts index 5976eaf..03cf984 100644 --- a/npm/core/src/index.ts +++ b/npm/core/src/index.ts @@ -15,3 +15,6 @@ export type { StateEnvelope } from "./generated/StateEnvelope"; export type { CommandSource } from "./generated/CommandSource"; export type { CommandEnvelope } from "./generated/CommandEnvelope"; export type { ObserverSpec } from "./generated/ObserverSpec"; +export type { KindRevision } from "./generated/KindRevision"; +export type { ClientMessage } from "./generated/ClientMessage"; +export type { ServerMessage } from "./generated/ServerMessage"; diff --git a/positron-core/src/lib.rs b/positron-core/src/lib.rs index 3eaeb9d..625329b 100644 --- a/positron-core/src/lib.rs +++ b/positron-core/src/lib.rs @@ -42,6 +42,7 @@ #![warn(missing_docs)] #![warn(rust_2018_idioms)] +pub mod session; pub mod wire; use std::fmt::Debug; diff --git a/positron-core/src/session.rs b/positron-core/src/session.rs new file mode 100644 index 0000000..81cf982 --- /dev/null +++ b/positron-core/src/session.rs @@ -0,0 +1,251 @@ +//! The transport-session protocol — the typed frames a host exchanges +//! with a substrate over any transport (WebSocket, UDS, airc). +//! +//! [`crate::wire`] defines the envelopes; this module defines the +//! *conversation*: a self-describing union per direction, so a +//! transport carries exactly one client frame type and one server +//! frame type and nothing is ever stringly-routed. +//! +//! ## Snapshot-then-live — the resync contract +//! +//! On [`ClientMessage::Subscribe`] — first connect and every +//! reconnect alike — the substrate MUST: +//! +//! 1. Immediately emit the **current** [`crate::wire::StateEnvelope`] +//! for each subscribed kind (revision-tagged), and +//! 2. then stream live updates from that moment forward. +//! +//! The substrate never replays history: a reconnect can therefore +//! never flood (no transcript replay) and never gap (the snapshot +//! covers the outage; live covers the rest). Renderers reconcile by +//! revision diff and re-render purely from the newest state. +//! +//! **Skip rule (exact equality).** The substrate MAY skip a kind's +//! snapshot **only when the client's `last_seen` revision EXACTLY +//! equals the substrate's current revision** for that kind. Never a +//! `>=` comparison: a substrate restart may reset its revision +//! counter, and under `>=` a client holding `last_seen: 500` against +//! a freshly-restarted substrate at revision 3 would keep stale state +//! forever. Exact equality makes counter resets safe by construction +//! — any mismatch, in either direction, sends the snapshot. (The skip +//! is purely an optimization; when in doubt, send.) +//! +//! **Subscribe is declarative (replace, not merge).** A `Subscribe` +//! frame declares the connection's complete interest set; it REPLACES +//! any previous subscription on the connection. Clients always send +//! their whole world — no incremental add/remove bookkeeping exists +//! to drift. Snapshot-then-live applies to every kind in the new set +//! (subject to the skip rule); kinds absent from the new set stop +//! flowing. Identical re-subscribes are therefore idempotent as a +//! special case of replacement. +//! +//! **Observers resync identically.** [`ClientMessage::Observe`] is +//! declarative per `observer_id` (re-observe REPLACES that observer's +//! registration) and triggers the same snapshot-then-live with the +//! same exact-equality skip on its `last_seen`. A reconnecting AI +//! observer rebuilds its perceived world exactly like a renderer +//! does — there is one resync contract, not a human one and an AI +//! one. +//! +//! **Ordering.** Within one kind, the substrate MUST emit `State` +//! frames in non-decreasing revision order over an ordered transport. +//! Consumers MUST drop an envelope whose revision is lower than one +//! already rendered for that kind (when both carry revisions) — this +//! makes accidental reordering harmless rather than corrupting. +//! **The drop watermark is scoped to the current subscription**: it +//! RESETS on every `Subscribe`/`Observe` the consumer sends, and +//! knowledge from before a reconnect travels ONLY in `last_seen`. +//! Without this scoping, a client that rendered revision 500, then +//! reconnected to a counter-reset substrate, would drop the +//! snapshot@3 it just asked for (3 < 500) and every frame after it — +//! the stale-forever gap reappearing one layer up. Watermark resets; +//! `last_seen` remembers; the two never share a counter. +//! +//! This contract is what makes reconnect tolerance *structural* for +//! consumers: a widget with no local source-of-truth cache cannot be +//! corrupted by a dropped transport, because the next subscribe +//! rebuilds its entire world from one snapshot. +//! +//! ## Deliberate v0 omissions +//! +//! There is no ack/error frame in [`ServerMessage`], so +//! `CommandEnvelope::correlation_id` has nothing protocol-level to +//! correlate with yet — command *results* currently surface as new +//! state (the substrate acts, state changes, the change streams +//! down). A `Result`/`Error` frame keyed by `correlation_id` is the +//! expected v0.x addition once a consumer actually needs +//! request-shaped feedback; adding a `ServerMessage` variant is +//! additive and non-breaking for tagged unions. +//! +//! The exact-equality skip has one residual ABA case: a client whose +//! `last_seen` revision N came from a pre-restart substrate may meet +//! a restarted substrate whose counter has independently reached the +//! same N with different content — the skip then wrongly omits one +//! snapshot. The staleness is transient (any next mutation of that +//! kind bumps the revision and streams down), and closing it fully +//! requires an epoch/generation id on revisions — a v0.x candidate +//! if real substrates restart often enough to care. Substrates that +//! persist their revision counters never enter this case. + +use serde::{Deserialize, Serialize}; +use ts_rs::TS; + +use crate::wire::{CommandEnvelope, ObserverSpec, StateEnvelope, StateLayer}; + +/// A `(kind, revision)` pair the client already holds — sent on +/// (re)subscribe so the substrate may skip redundant snapshots. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)] +#[ts(export)] +pub struct KindRevision { + /// Widget kind (matches [`StateEnvelope::kind`]). + pub kind: String, + /// Last revision of that kind the client rendered. + /// + /// TS `number`, same rationale as [`StateEnvelope::revision`]. + #[ts(type = "number")] + pub revision: u64, +} + +/// Client → substrate frames. One union; the transport is fully +/// self-describing and there is no out-of-band routing. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)] +#[serde(rename_all = "snake_case", tag = "type")] +#[ts(export)] +pub enum ClientMessage { + /// Declare (or re-declare, after a reconnect) what this client + /// renders. Triggers snapshot-then-live per the module contract. + /// Always idempotent: subscribing twice is a resync, never an + /// error and never a duplicate live stream. + Subscribe { + /// Widget kinds to receive. Empty = none (explicit opt-in, + /// consistent with [`ObserverSpec`]). + kinds: Vec, + /// Cadence layers to receive. Empty = none. + layers: Vec, + /// Revisions already held; substrate MAY skip matching + /// snapshots. Empty = client holds nothing, send everything. + #[serde(default)] + last_seen: Vec, + }, + /// A user (or observer) action — see [`CommandEnvelope`] for the + /// provenance contract. + Command(CommandEnvelope), + /// Register (or re-register, after a reconnect) an AI observer. + /// Declarative per `spec.observer_id` — replaces that observer's + /// previous registration — and triggers snapshot-then-live with + /// the exact-equality skip, identically to `Subscribe`. One + /// resync contract for humans and AIs. Perception budget enforced + /// substrate-side. + Observe { + /// The observer's identity, budget, and perception scope. + spec: ObserverSpec, + /// Revisions this observer already perceived; same skip rule + /// as `Subscribe::last_seen`. + #[serde(default)] + last_seen: Vec, + }, +} + +/// Substrate → client frames. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)] +#[serde(rename_all = "snake_case", tag = "type")] +#[ts(export)] +pub enum ServerMessage { + /// A state update — either a snapshot (immediately after + /// `Subscribe`) or a live change. Deliberately the SAME frame for + /// both: consumers reconcile by `kind` + `revision`, never by + /// "which phase am I in" bookkeeping. If a client must + /// distinguish, the revision diff already tells it. + State(StateEnvelope), +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::wire::CommandSource; + use uuid::Uuid; + + /// The session frames round-trip and the tag layout is pinned — + /// these JSON shapes are what ContinuumHost (Rust) and + /// positron-lit (TS) both build against. + #[test] + fn client_frames_round_trip_with_pinned_tags() { + let sub = ClientMessage::Subscribe { + kinds: vec!["chat".into()], + layers: vec![StateLayer::Session], + last_seen: vec![KindRevision { + kind: "chat".into(), + revision: 41, + }], + }; + let json = serde_json::to_string(&sub).unwrap(); + assert!(json.starts_with(r#"{"type":"subscribe""#), "{json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), sub); + + let cmd = ClientMessage::Command(CommandEnvelope { + kind: "chat".into(), + command: "chat/send".into(), + params: serde_json::json!({"text": "hi"}), + correlation_id: Uuid::from_u128(7), + source: CommandSource::Human, + }); + let json = serde_json::to_string(&cmd).unwrap(); + assert!(json.starts_with(r#"{"type":"command""#), "{json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), cmd); + } + + #[test] + fn server_state_frame_round_trips() { + let state = ServerMessage::State(StateEnvelope { + kind: "chat".into(), + revision: Some(42), + layer: StateLayer::Session, + payload: serde_json::json!({"messages": ["hi"]}), + }); + let json = serde_json::to_string(&state).unwrap(); + assert!(json.starts_with(r#"{"type":"state""#), "{json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), state); + } + + /// `last_seen` is optional on the wire — a minimal subscribe from + /// a fresh client (or an older build) decodes with "send + /// everything" semantics, never an error. + #[test] + fn bare_subscribe_decodes_with_empty_last_seen() { + let bare = r#"{"type":"subscribe","kinds":["chat"],"layers":["session"]}"#; + let msg: ClientMessage = serde_json::from_str(bare).unwrap(); + match msg { + ClientMessage::Subscribe { last_seen, .. } => assert!(last_seen.is_empty()), + other => panic!("expected Subscribe, got {other:?}"), + } + } + + /// The `observe` frame's tag layout is pinned (it was the one + /// variant without committed-test coverage), and a bare observe + /// without `last_seen` decodes as perceive-from-scratch — the + /// observer resync contract mirrors the subscriber one exactly. + #[test] + fn observe_round_trips_with_pinned_tag_and_default_last_seen() { + let obs = ClientMessage::Observe { + spec: ObserverSpec { + observer_id: "maya".into(), + budget_hz: 4, + kinds: vec!["chat".into()], + layers: vec![StateLayer::Session, StateLayer::Semantic], + }, + last_seen: vec![KindRevision { + kind: "chat".into(), + revision: 12, + }], + }; + let json = serde_json::to_string(&obs).unwrap(); + assert!(json.starts_with(r#"{"type":"observe""#), "{json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), obs); + + let bare = r#"{"type":"observe","spec":{"observer_id":"maya","budget_hz":4,"kinds":["chat"],"layers":["session"]}}"#; + match serde_json::from_str::(bare).unwrap() { + ClientMessage::Observe { last_seen, .. } => assert!(last_seen.is_empty()), + other => panic!("expected Observe, got {other:?}"), + } + } +}