aion_core/

cluster_event.rs

//! Cluster topology and ownership events for the ops console real-time channel (WS3).
//!
//! This module defines the *typed contract* for the second class of real-time push the ops console
//! consumes: cluster/topology/ownership deltas that today exist only as `tracing` logs and
//! Prometheus gauges. The wire shapes live in `aion-core` (not `aion-server`) for one reason: only
//! this leaf crate depends on `ts-rs`, so this is the single place a Rust type can cross the
//! Rust -> TypeScript boundary into the ops console's generated bindings.
//!
//! # Scope (WS3 FOUNDATION)
//!
//! This file is *types only*. It deliberately does **not** define the broadcast publisher, the
//! emit call-sites in the supervisor / registry, the subscription endpoint, or the namespace gate.
//! Those are wired in later WS3 increments. Defining the contract first lets the ops console and the
//! server agree on shapes before any behaviour ships.
//!
//! # Honesty corrections applied to the original design
//!
//! The design table named several variants whose payloads cannot be sourced from the subsystems
//! that would emit them today. Rather than ship a type whose fields are guaranteed to be faked,
//! those variants are **descoped from Phase 1** and documented here so the omission is explicit and
//! not silently rediscovered at wiring time:
//!
//! - **`NodeMetricsSampled` — DEFERRED.** `aion-server`'s metrics (`observability/metrics.rs`) have
//!   no `node` dimension (`connected_workers` is an `IntGaugeVec` keyed by *namespace*) and there
//!   is no `workflows_running` gauge at all. A `{ node, connected_workers, workflows_running }`
//!   payload cannot be produced by "piggybacking the gauge setters". Sourcing it honestly requires
//!   a real metrics change (node-labelled gauges + a running gauge). Until then the ops console
//!   derives `connected_workers` client-side from [`ClusterEvent::WorkerConnected`] /
//!   [`ClusterEvent::WorkerDisconnected`] deltas against the [`ClusterSnapshot`] baseline. Adding
//!   a timer that scrapes Prometheus to fill this variant is the exact polling-as-push regression
//!   WS3 exists to remove, so the variant is omitted rather than tempting that shortcut.
//! - **`ShardOwnerChanged` / `FencedCasRejected` — DEFERRED (fast-follow).** The haematite seam
//!   (`HaematiteStore::publish_ln(shard)`) carries only `shard`; the fenced-CAS reject is a
//!   `haematite::DatabaseError::Fenced { .. }` destructured with `..` and mapped to
//!   `StoreError::NotOwner { shard }`, carrying only `shard` — never the owner name or the
//!   attempted/current epochs. The proposed payloads require enriching haematite's `Fenced` error
//!   to surface owner identity and epoch, which is a cross-crate (likely cross-repo) change. The
//!   cluster map is still *honest* without these two: shard adoption is fully observable from the
//!   supervisor's `tick()` ([`ClusterEvent::ShardAdopted`] and friends). Only the store-side
//!   CAS-reject detail is delayed.
//!
//! # u64 precision across the TS boundary
//!
//! The ts-rs config exports every `u64` as TS `number` (`with_large_int("number")` in
//! `generated_types.rs`), which truncates above `2^53`. [`ClusterEventMeta::cluster_seq`] and the
//! epoch fields below are `u64`. This is the *same* accepted ceiling that already applies to
//! [`crate::EventEnvelope::seq`]; cluster sequencing follows the established project convention
//! rather than introducing a divergent string encoding. The ceiling is documented on each field so
//! the gap-detection math on the client is aware of the bound. A long-lived deployment must keep
//! `cluster_seq` below `2^53`; in practice this is never reached for a topology-event counter.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use crate::{ActivityId, WorkflowId};

/// Metadata stamped on every [`ClusterEvent`], mirroring [`crate::EventEnvelope`] for the cluster
/// channel.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterEventMeta {
    /// Monotonic sequence number assigned by the cluster publisher (a single deployment-global
    /// `AtomicU64`). The client uses this for gap detection and `after_seq` reconnect dedup, the
    /// cluster analog of [`crate::EventEnvelope::seq`].
    ///
    /// Exported to TypeScript as `number`; see the module docs for the accepted `2^53` ceiling.
    pub cluster_seq: u64,
    /// UTC wall-clock instant at which the originating subsystem observed the state change.
    pub observed_at: DateTime<Utc>,
}

/// Transport a connected worker is delivered to over.
///
/// Mirrors the live `aion_server::worker::registry::WorkerDelivery` discriminants without carrying
/// the (non-serializable) delivery channels, so it can cross the wire and the ts-rs boundary.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(tag = "transport")]
pub enum WorkerTransport {
    /// gRPC bidirectional-stream delivery (the default transport).
    Grpc,
    /// Liminal server-push delivery (feature-gated on the server).
    Liminal,
}

impl WorkerTransport {
    /// The word an operator sees for this transport in a log line.
    ///
    /// Lives beside the discriminants rather than in the server, so the name a
    /// liveness withdrawal prints and the name a cluster event carries cannot
    /// disagree about which link an operator should go and look at.
    #[must_use]
    pub const fn name(self) -> &'static str {
        match self {
            Self::Grpc => "grpc",
            Self::Liminal => "liminal",
        }
    }
}

/// Why a worker left the connected set.
///
/// NOTE (wiring honesty): the single registry deregistration site
/// (`ConnectedWorkerRegistry::deregister` / `remove_worker`) does not today distinguish a transport
/// disconnect from an idle timeout from an explicit deregister. The emit increment MUST derive the
/// reason at the call site from real signal (e.g. a closed delivery channel vs an explicit
/// deregister RPC vs a liveness-timeout sweep) or collapse to the variant it can actually prove.
/// This enum defines the *contract*; it must not be populated with a fabricated distinction.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(tag = "reason")]
pub enum WorkerDeathReason {
    /// The worker's delivery transport dropped (stream/connection closed).
    Disconnect,
    /// The worker was removed by a liveness/idle-timeout sweep.
    Timeout,
    /// The worker explicitly deregistered.
    Deregistered,
}

/// Operator-requested lifecycle state for a durable worker deployment.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum DesiredState {
    /// The deployment should run.
    Running,
    /// The deployment should remain stopped.
    Stopped,
}

impl DesiredState {
    /// Stable human-readable state token.
    #[must_use]
    pub const fn token(self) -> &'static str {
        match self {
            Self::Running => "running",
            Self::Stopped => "stopped",
        }
    }
}

/// Create-or-replace result exposed on the cluster and HTTP wires.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum PutOutcome {
    /// No record previously existed for the key.
    Created,
    /// A record already existed for the key and was replaced.
    Replaced,
}

/// Deployment-record lookup state captured when a worker registers.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum DeploymentAssociation {
    /// The named deployment record existed.
    Known,
    /// The deployment store was checked and the named record was absent.
    Absent,
    /// No deployment store was attached, so existence was not checked.
    Unchecked,
}

/// A cluster/topology/ownership delta pushed over the ops console real-time channel.
///
/// Tagged union on `type` to match the existing [`crate::Event`] wire shape. Every variant carries
/// [`ClusterEventMeta`] as `meta`. Cluster events are deployment-scoped, not namespace-stamped at
/// the envelope level; the `Worker*` variants carry their own `namespaces` so the server-side gate
/// can intersect them against the caller's grants (see the deferred `cluster_filter`).
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
#[serde(tag = "type")]
pub enum ClusterEvent {
    /// A peer was added to this node's watch set (cluster-membership grew mid-session).
    ///
    /// Distinct from [`Self::PeerConnected`]: this is a *topology* change (a new peer to watch),
    /// not a *liveness* transition. Without it, a peer that joins after the priming
    /// [`ClusterSnapshot`] would silently never appear on the map until the next snapshot.
    PeerAdded {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The newly-watched peer's node name.
        peer_name: String,
        /// The peer's forwarding address, when known.
        forward_addr: Option<String>,
    },
    /// A watched peer transitioned from down/unknown to connected (a liveness recovery).
    ///
    /// EMIT NOTE: the supervisor `tick()` resets `consecutive_down`/`adopted` on the same tick it
    /// observes `connected`, so the recovery condition (`was_down = consecutive_down > 0 ||
    /// adopted`) MUST be captured *before* that reset and the emit driven by the captured value.
    /// Emitting after the reset produces no recovery events (every tick looks freshly connected).
    PeerConnected {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The recovered peer's node name.
        peer_name: String,
        /// The peer's forwarding address, when known.
        forward_addr: Option<String>,
    },
    /// A watched peer was observed down on a supervisor tick.
    ///
    /// Emitted on every tick the peer is observed down; `confirmed` flips to `true` once
    /// `consecutive_down >= confirmations` (the debounce threshold that authorizes adoption).
    PeerDisconnected {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The down peer's node name.
        peer_name: String,
        /// Consecutive ticks this peer has been observed down.
        consecutive_down: u32,
        /// Whether the debounce threshold has been crossed (adoption-eligible).
        confirmed: bool,
    },
    /// This node adopted shards previously owned by a failed peer.
    ShardAdopted {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Shard indices adopted in this transition.
        shards: Vec<usize>,
        /// The peer the shards were adopted from.
        from_peer: String,
        /// This node's name (the new owner).
        adopted_by: String,
    },
    /// An adoption attempt failed and will be retried on a subsequent tick.
    ///
    /// HONESTY: the original design hardcoded `will_retry: true`. That is a silent lie — a peer
    /// later observed connected resets `adopted` and stops retrying, and the
    /// "handled elsewhere" path terminally stops. The retry decision is not expressible as a
    /// constant, so the field is omitted; the ops console infers retry by observing whether a
    /// subsequent [`Self::ShardAdopted`] / [`Self::ShardAdoptionSkipped`] for the same
    /// `(from_peer, shards)` arrives, rather than trusting a promise that may never be kept
    /// (ADR-016 no-silent-failure).
    ShardAdoptionFailed {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Shard indices the failed attempt targeted.
        shards: Vec<usize>,
        /// The peer the shards would have been adopted from.
        from_peer: String,
        /// Human-readable adoption error.
        error: String,
    },
    /// Adoption was skipped because the shards are already held by a live third-party owner.
    ShardAdoptionSkipped {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Shard indices that were skipped.
        shards: Vec<usize>,
        /// The peer the shards would have been adopted from.
        from_peer: String,
        /// The live node currently holding the shards.
        held_by: String,
    },
    /// A worker joined the connected set.
    WorkerConnected {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Stable worker identifier.
        worker_id: String,
        /// Namespaces this worker serves.
        namespaces: Vec<String>,
        /// Task-queue pool this worker serves within its namespaces.
        task_queue: String,
        /// Delivery transport for this worker.
        transport: WorkerTransport,
        /// Locality/node label, when the worker reported one.
        node: Option<String>,
        /// Deployment named by the worker's instance identity, when supplied.
        deployment: Option<String>,
        /// Lookup state for the named deployment. `None` means the worker
        /// supplied no instance identity.
        deployment_association: Option<DeploymentAssociation>,
    },
    /// A worker left the connected set.
    WorkerDisconnected {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Stable worker identifier.
        worker_id: String,
        /// Namespaces this worker served.
        namespaces: Vec<String>,
        /// Why the worker left (see [`WorkerDeathReason`] wiring note).
        reason: WorkerDeathReason,
    },
    /// A dispatch parked on an unserved queue with **no availability
    /// deadline** — the wait that can hold a run forever.
    ///
    /// Every other selection outcome has a terminal echo the operator can see:
    /// a served dispatch completes, and a deadline-bounded park expires into a
    /// typed `WorkerUnavailable` refusal that reaches the workflow's own
    /// history. The unbounded park had only a WARN line and the queryable
    /// unserved state — nothing pushed. This variant is that push (#266 T4):
    /// the #266 boot-ordering defect stranded recovered runs in exactly this
    /// state, and the only sign was a log nobody was reading. The status-truth
    /// front (#264) consumes it.
    ///
    /// Emitted under the same transition rule as the WARN it accompanies: once
    /// when a park episode is first classified, and again only when the
    /// classified reason changes. Re-observations of the same reason are
    /// silent, so a queue nobody serves does not flood the channel.
    ///
    /// `reason` and `policy` carry the queue-service taxonomy's canonical
    /// spellings (e.g. `NO_LIVE_POLLERS`, `durable_pending`) as strings rather
    /// than server-crate enums, the same leaf-crate discipline
    /// [`Self::NamespaceCreated`] applies to `origin`.
    DispatchParked {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Namespace the dispatch belongs to.
        namespace: String,
        /// Task queue the dispatch is addressed to.
        task_queue: String,
        /// Activity type the dispatch needs served.
        activity_type: String,
        /// Node label the dispatch is pinned to, if any.
        node: Option<String>,
        /// Canonical queue-service reason spelling (`NO_QUEUE_DECLARATION` /
        /// `NO_LIVE_POLLERS` / `POLLERS_INCOMPATIBLE` / `SATURATED`).
        reason: String,
        /// Canonical queue-service policy spelling (`strict` /
        /// `durable_pending`).
        policy: String,
        /// Workflow whose activity is parked.
        workflow_id: WorkflowId,
        /// Parked activity's history ordinal.
        activity_id: ActivityId,
        /// How long the dispatch had been waiting when this was emitted.
        ///
        /// Exported to TypeScript as `number`; see the module docs for the
        /// accepted `2^53` ceiling.
        waited_ms: u64,
        /// Workers connected for `(namespace, task_queue)`, whatever they
        /// serve, at the census behind this verdict.
        workers_in_pool: usize,
        /// Of those, workers advertising this activity type.
        workers_serving_activity: usize,
        /// Of those, workers that also satisfy the dispatch's node pin.
        compatible_workers: usize,
        /// How long ago a compatible worker was last in service, or `None`
        /// when none has ever been seen for this address in this server's
        /// life.
        ///
        /// Exported to TypeScript as `number`; see the module docs for the
        /// accepted `2^53` ceiling.
        last_compatible_poller_age_ms: Option<u64>,
    },
    /// The cluster supervisor started on this node (lifecycle).
    ///
    /// Lets the calm-state view (ADR-019) distinguish "supervisor running, all peers healthy" from
    /// "supervisor not running".
    SupervisorStarted {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// This node's name.
        node: String,
    },
    /// The cluster supervisor stopped on this node (clean drain / shutdown).
    SupervisorStopped {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// This node's name.
        node: String,
    },
    /// A brand-new namespace was minted in the durable registry (Control-Plane
    /// Phase 1, S8).
    ///
    /// Emitted exactly once per genuinely-new namespace at the registry's single
    /// `MintOutcome::Created` choke-point — so the worker-register seam (S5), the
    /// workflow-start safety net (S6), and the explicit `POST /namespaces` path
    /// (S7) all surface the same delta. An idempotent re-reference of an existing
    /// namespace (`MintOutcome::AlreadyExisted`) never emits this, so the ops
    /// console appends each namespace exactly once with no refresh.
    ///
    /// `origin` is the stable `snake_case` label (`worker_mint` / `start_mint` /
    /// `explicit` / `inferred_from_state`) rather than the `aion-store`
    /// `NamespaceOrigin` enum, because this leaf crate must not depend on the
    /// store crate; the label is the same string the audit `tracing` event logs.
    /// `created_at` is the durable record's first-mint instant, carried so the
    /// console's created column matches the registry without a follow-up fetch.
    NamespaceCreated {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The minted namespace's name (registry primary key).
        name: String,
        /// The durable record's creation instant (its `created_at`).
        created_at: DateTime<Utc>,
        /// How the namespace came to exist, as the stable `snake_case` label.
        origin: String,
    },
    /// A namespace's durable placement directive was changed (Control-Plane
    /// Phase 2, P2-P2 — `PUT /namespaces/{name}/placement`).
    ///
    /// Emitted on the SAME deploy-scoped channel as [`Self::NamespaceCreated`]
    /// after the placement is durably set, so the ops console's namespace panel
    /// reflects an operator's placement change live with no refresh. `placement`
    /// is the stable wire projection ([`NamespacePlacementWire`]) of the durable
    /// `NamespacePlacement`, carried as a `kind` label + label set rather than the
    /// `aion-store` enum, because this leaf crate must not depend on the store
    /// crate (the same discipline [`Self::NamespaceCreated`] applies to `origin`).
    NamespacePlacementChanged {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The namespace whose placement changed (registry primary key).
        name: String,
        /// The new placement directive, as the stable wire projection.
        placement: NamespacePlacementWire,
    },
    /// A periodic snapshot of a namespace's concurrency-quota state (Control-Plane
    /// Phase 2, P2-Q3), pushed on the SAME deploy-scoped channel as
    /// [`Self::NamespaceCreated`] so the ops console renders a live "in-flight /
    /// ceiling" badge that ticks as work flows.
    ///
    /// This is the ONE edge-triggered exception's honest counterweight: unlike the
    /// other variants (each emitted at a real subsystem mutation), quota state
    /// changes on every claim/settle, so per-row emission would be a firehose.
    /// Instead a throttled snapshot task samples each active namespace's REAL
    /// durable state once per cadence and emits this — a periodic snapshot per
    /// active namespace, never a client poll. The console folds the latest snapshot
    /// per namespace, superseding the prior value.
    ///
    /// `in_flight` is the durable **Claimed** outbox-row count for the namespace
    /// (`count_claimed_outbox_rows`) — the SAME notion the keyed backpressure caps,
    /// never the dead `inflight_activities` gauge (P2-Q0). `ceiling` is the tenant's
    /// **cluster-wide** contract: its explicit `max_in_flight_activities` override
    /// or the platform default — the number the operator set, not the per-node
    /// proportional slice (exposing per-node math is the leaky-abstraction footgun
    /// §3.6 rejects). On a single node the durable `in_flight` this node sees is the
    /// whole cluster's; multi-node exact aggregation is the reserved §3.6 follow-up.
    NamespaceQuotaState {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// The namespace this quota snapshot describes (registry primary key).
        namespace: String,
        /// Durable count of currently-**Claimed** outbox rows for the namespace —
        /// the in-flight activities the ceiling caps.
        ///
        /// Exported to TypeScript as `number`; see the module docs for the accepted
        /// `2^53` ceiling (an in-flight count never approaches it).
        in_flight: u64,
        /// The tenant's cluster-wide concurrency ceiling: its explicit
        /// `max_in_flight_activities` override, or the platform default when unset.
        ceiling: u32,
    },
    /// A worker deployment was durably created or replaced by an operator.
    WorkerDeploymentPut {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Durable deployment name.
        name: String,
        /// Typed create-or-replace outcome.
        outcome: PutOutcome,
        /// Typed desired state.
        desired_state: DesiredState,
        /// Version captured from the deployed server binary.
        binary_version: String,
        /// SHA-256 captured from the deployed server executable bytes.
        binary_content_hash: String,
    },
    /// An operator durably changed a deployment's desired state.
    WorkerDeploymentDesiredStateChanged {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Durable deployment name.
        name: String,
        /// Typed desired state.
        desired_state: DesiredState,
    },
    /// An operator durably deleted a deployment record.
    WorkerDeploymentDeleted {
        /// Cluster-event metadata.
        meta: ClusterEventMeta,
        /// Deleted deployment name.
        name: String,
    },
}

/// Stable wire projection of a namespace's durable placement directive
/// (`NamespacePlacement` in `aion-store`), for the ops-console real-time channel
/// (Control-Plane Phase 2).
///
/// Lives in this leaf crate (not `aion-server` / `aion-store`) for the same reason
/// the rest of [`ClusterEvent`] does: only this crate crosses the Rust ->
/// TypeScript boundary via `ts-rs`. `kind` is the stable `snake_case` variant tag
/// (`unplaced` / `prefer` / `pinned`) and `nodes` is the (possibly empty)
/// node-label set; `Unplaced` carries an empty `nodes`. Modelling it as a flat
/// `{kind, nodes}` shape rather than re-encoding the store's tagged form keeps the
/// generated TS binding a single simple type the console can switch on.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct NamespacePlacementWire {
    /// Stable `snake_case` placement-kind tag: `unplaced` / `prefer` / `pinned`.
    pub kind: String,
    /// The placement's node-label set, deterministically ordered. Empty for
    /// `unplaced`; the preferred/required labels for `prefer`/`pinned`.
    pub nodes: Vec<String>,
}

/// A peer entry in the priming [`ClusterSnapshot`].
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterPeer {
    /// The peer's node name.
    pub peer_name: String,
    /// The peer's forwarding address, when known.
    pub forward_addr: Option<String>,
    /// Whether the peer is currently observed connected.
    pub connected: bool,
    /// Consecutive ticks observed down (0 when connected).
    pub consecutive_down: u32,
}

/// A shard ownership entry in the priming [`ClusterSnapshot`].
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterShard {
    /// Shard index.
    pub shard: usize,
    /// The node that currently owns the shard.
    pub owner: String,
    /// The epoch fence value at which the owner holds the shard.
    ///
    /// Exported to TypeScript as `number`; see the module docs for the accepted `2^53` ceiling.
    pub epoch: u64,
}

/// A connected-worker entry in the priming [`ClusterSnapshot`].
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterWorker {
    /// Stable worker identifier.
    pub worker_id: String,
    /// Namespaces this worker serves (already intersected with the caller's grants by the gate).
    pub namespaces: Vec<String>,
    /// Task-queue pool this worker serves.
    pub task_queue: String,
    /// Delivery transport for this worker.
    pub transport: WorkerTransport,
    /// Locality/node label, when reported.
    pub node: Option<String>,
    /// Deployment named by the worker's instance identity, when supplied.
    pub deployment: Option<String>,
    /// Lookup state for that deployment. `None` for workers without instance
    /// identity.
    pub deployment_association: Option<DeploymentAssociation>,
}

/// A durable worker-deployment entry in the priming cluster snapshot.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterDeployment {
    /// Durable deployment name.
    pub name: String,
    /// Typed desired state.
    pub desired_state: DesiredState,
    /// Version captured at deploy time.
    pub binary_version: String,
    /// SHA-256 captured at deploy time.
    pub binary_content_hash: String,
}

/// A calm-state baseline of the whole cluster, sent as the priming reply before the live delta
/// stream so the ops console can render an at-a-glance "all clear" before any [`ClusterEvent`]
/// arrives (ADR-019). On `cluster_lagged` the client re-requests this rather than replaying a
/// (non-durable) delta history.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct ClusterSnapshot {
    /// The reading node's own name (self-identity; baselines the `Peer*` deltas which describe
    /// *other* nodes).
    pub node: String,
    /// The `cluster_seq` this snapshot is consistent as-of; the client applies only deltas with a
    /// strictly greater `cluster_seq`.
    ///
    /// Exported to TypeScript as `number`; see the module docs for the accepted `2^53` ceiling.
    pub as_of_seq: u64,
    /// Watched peers and their current liveness.
    pub peers: Vec<ClusterPeer>,
    /// Shards owned by (or visible to) the reading node.
    pub shards: Vec<ClusterShard>,
    /// Connected workers, already gated to the caller's namespaces.
    pub workers: Vec<ClusterWorker>,
    /// Durable worker deployments visible under the deploy grant.
    pub deployments: Vec<ClusterDeployment>,
}

/// A command the ops console can issue against the cluster channel (ADR-020 command seam).
///
/// Defined now for contract coherence. **Phase 1 ships only [`Self::RequestClusterSnapshot`]** (a
/// read). The mutating variants compile so the contract exists, but their handlers reject with an
/// `unimplemented` wire error — and, per ADR-020, MUST still run the full auth gate
/// (`caller.deploy_granted()`) *before* rejecting, so the seam's authorization contract is
/// exercised now and an `unimplemented` stub is never an auth-bypass-shaped hole.
///
/// Tagged on `command` (distinct from [`ClusterEvent`]'s `type`) because commands and events are
/// different directions on the wire; the ops console's protocol parser keys command frames on
/// `command` and event frames on `type`.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
#[serde(tag = "command")]
pub enum ClusterCommand {
    /// Read the current cluster baseline (peers + shards + workers). Read-only; needs only
    /// namespace-or-deploy scope. **Phase 1.**
    RequestClusterSnapshot {},
    /// Cancel a running workflow. Aspirational — handler returns `unimplemented`.
    CancelWorkflow {
        /// Owning namespace.
        namespace: String,
        /// Target workflow.
        workflow_id: String,
    },
    /// Reopen a failed/closed workflow. Aspirational — handler returns `unimplemented`.
    ReopenWorkflow {
        /// Owning namespace.
        namespace: String,
        /// Target workflow.
        workflow_id: String,
    },
    /// Redrive a single outbox row. Aspirational — handler returns `unimplemented`.
    RedriveOutboxRow {
        /// Owning namespace.
        namespace: String,
        /// Target workflow.
        workflow_id: String,
        /// Outbox row ordinal.
        ordinal: u64,
    },
    /// Drain a node (stop-new-work + finish-in-flight + safe shutdown). Aspirational.
    DrainNode {
        /// Target node name.
        node: String,
    },
    /// Planned epoch-fenced shard handoff to a target node. Aspirational.
    PlannedHandoff {
        /// Shard to move.
        shard: usize,
        /// Destination node.
        target_node: String,
    },
    /// Test-only chaos kill of a node. Aspirational (gated).
    ChaosKillNode {
        /// Target node name.
        node: String,
    },
}

/// A typed terminal error on the cluster channel.
///
/// Mirrors the workflow path's lagged contract: when a subscriber falls behind the bounded cluster
/// broadcast buffer the server sends exactly one of these and closes, carrying the skipped count so
/// the client can decide snapshot-vs-resume (it always re-requests a [`ClusterSnapshot`], since
/// there is no durable cluster history). Surfaced to the UI as a typed error, never silently
/// dropped (ADR-016).
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
#[serde(tag = "kind")]
pub enum ClusterStreamError {
    /// The subscriber lagged past the bounded broadcast buffer; `skipped` deltas were dropped.
    ClusterLagged {
        /// Number of cluster events dropped because the subscriber fell behind.
        ///
        /// Exported to TypeScript as `number`; see the module docs for the accepted `2^53` ceiling.
        skipped: u64,
    },
}