tower_rules/

lib.rs

//! `TowerRule` schema — predicate + trigger + federation types for yah-tower.
//!
//! This crate is the schema source of truth for tower rules. It has zero
//! dependencies on the tower runtime; the tower engine, the desktop authoring
//! UI, `tower.simulate`, and tests all depend on this crate directly.
//!
//! Architecture: `.yah/docs/architecture/A052-yah-tower.md`
//!
//! @yah:ticket(R380-T4, "Add tower-rules TaskPlacement (on-wire flat form, no MeshIdent)")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-01T21:06:13Z)
//! @yah:status(review)
//! @yah:parent(R380)
//! @yah:next("The tower-rules enum at lib.rs:257 today is Local | Remote | Integration with no MeshIdent (cross-crate dep direction forbids it). Replace with TaskPlacement { location: LocationKind (Local|RemoteAny{tier}), runtime: Native|Container } — note no Remote(MeshIdent) variant here either.")
//! @yah:next("Provide a converter from tower-rules TaskPlacement → task crate TaskPlacement that the consumer applies at the boundary (tower-rules can't depend on task crate types).")
//! @yah:next("Update ts-rs export — new TS file TaskPlacement.ts replaces ForgeWhere.ts. Coordinate with frontend integration tests.")
//! @yah:handoff("Added TaskPlacement / TaskLocation / TaskRuntime / TierTag in crates/yah/tower-rules/src/lib.rs alongside the existing ForgeWhere (kept for T5 to migrate Trigger::AgentDispatch). Wire format chosen to match task crate: TaskLocation uses tag=\"kind\" with snake_case discriminator so boundary converters can pass JSON through for shared variants. Tower-rules TaskLocation has only Local | RemoteAny{tier} — no Remote(MeshIdent) variant since rules can't name specific yubaba nodes from YAML. Added 5 round-trip + wire-format tests; all 24 tower-rules tests pass. ForgeWhere is now marked deprecated in its rustdoc + the generated ForgeWhere.ts.")
//! @yah:next("R380-T5 picks up: replace Trigger::AgentDispatch.forge_where: ForgeWhere with placement: TaskPlacement, update parse.rs YAML loader, update tower/src/triggers.rs ForgeSpec, update tower/src/rules/validate.rs round-trip fixtures, update tests in tower/tests/{dispatch,simulate,triggers}.rs and tower-rules/tests/round_trip.rs, update tower-rules library YAML files (yubaba-restart-on-error.yaml etc.)")
//! @yah:next("Boundary converter: tower → task ForgeSpec construction needs to widen tower_rules::TaskPlacement → task::TaskPlacement. Mapping is mechanical: Local→Local, RemoteAny{tier:tower_rules::TierTag(s)}→RemoteAny{tier:workload_spec::TierTag(s)}, runtime variants pass through. Lives in whichever crate already depends on both (probably the tower/cli boundary in app/yah/cli/src/tower.rs or a tower-task adapter). The tower crate itself currently does NOT depend on task — pulling task in there is the natural seam.")
//! @yah:verify("cargo test -p tower-rules")
//! @yah:verify("cargo run -q -p tower-rules --bin tower-rules-export-ts && ls packages/yah/ui/src/gen/{TaskPlacement,TaskLocation,TaskRuntime,TierTag}.ts")
//!
//! @yah:ticket(R406-T7, "Structured drain protocol: versioned wire format, ack semantics, SIGTERM fallback")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-02T03:26:44Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R406)
//! @arch:see(.yah/docs/working/W154-yubaba-dual-runtime.md)
//! @yah:depends_on(R406-T6)
//! @yah:handoff("T7 ships the structured drain protocol end-to-end at the protocol+enforcer layer. (1) kamaji-proto: DrainOutcome { Flushed | Checkpointed | ForceKilled | UnknownWorkload | Unsupported } and a peer DrainCompleted push message (non-exhaustive, V1-compatible). DrainAck rustdoc now documents the two-mode semantics (sync T7 default / async T8 push). DrainBudget::total_ms() helper. (2) app/yah/kamaji/src/drain.rs: enforce_drain state machine — SIGTERM via pidfd_send_signal (race-free), AsyncFd::readable for exit observation bounded by flush_ms+checkpoint_ms, SIGKILL escalation + 5s safety reap. classify_clean_exit pure helper splits Flushed vs Checkpointed by elapsed-vs-flush_ms boundary (≤ inclusive on flush). (3) Server registry gained insert_drainable / take_drainable + DrainableHandle { pid, pidfd }. Drain RPC short-circuits on unknown workload, otherwise runs enforce_drain synchronously and translates DrainOutcome → DrainAck { accepted, reason } via drain_outcome_to_ack (decision-table: clean phase = accepted=true with elapsed_ms; ForceKilled / Unsupported / Unknown / DrainError = accepted=false). (4) Coverage: 19 kamaji-proto codec tests (8 new for drain outcomes + DrainCompleted round-trip + DrainBudget saturation), 47 kamaji unit tests (drain::tests pure boundary cases + server::tests decision-table-as-spec on every DrainOutcome variant + drain_unknown_workload UDS round-trip), 1 integration test extended in tests/uds_skeleton.rs (Drain over UDS for unknown id). Linux-only drain::linux::tests exercise the real-child path (SIGTERM-responsive vs SIGTERM-ignoring) but only run on Linux CI — not exercised on darwin local.")
//! @yah:next("T8 picks up: extract Yubaba-proper supervision out of in-process, dispatch through Kamaji over UDS. Kamaji's Deploy handler currently still returns Internal — T8 wires it to native::spawn and populates registry.insert_drainable(id, { pid, pidfd }) so the Drain handler has a real handle to consume.")
//! @yah:next("T8 also adds the back-channel (Kamaji → Yubaba push) that lets Kamaji reply DrainAck { accepted=true, reason='started' } immediately and push DrainCompleted { request_id, id, outcome } when the drain finishes — that switches the wire shape from synchronous-mode to async-mode without touching V1 protocol-version (DrainCompleted rides #[non_exhaustive]).")
//! @yah:next("T11 picks up the workload-side leg: pick stdio-sentinel vs HTTP-endpoint, define the workload's ack envelope, then drain.rs grows a phase boundary signal so workloads know they crossed flush_ms into checkpoint_ms without consulting the clock. The pidfd-level SIGTERM floor stays as the fallback.")
//! @yah:verify("cargo test -p kamaji-proto -p kamaji")

use serde::{Deserialize, Serialize};
use ts_rs::TS;

mod version;
pub use version::SchemaVersion;

// ── Primitive newtypes ────────────────────────────────────────────────────────

/// Unique identifier for a rule instance, e.g. `"yubaba-quorum-loss"`.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct RuleId(pub String);

/// DNS-segment identity for a service on the cluster mesh, e.g.
/// `"noisetable-api.pdx"`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct MeshIdent(pub String);

/// Opaque identifier for a forge run.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct ForgeId(pub String);

/// Reference to an agent class definition, e.g. `"gnome/fixer"`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct AgentClassRef(pub String);

/// Mustache-style template for the agent prompt, rendered with the captured
/// event payload at dispatch time.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct PromptTemplate(pub String);

// ── Scope filter ──────────────────────────────────────────────────────────────

/// Filter over the scope of a scryer event.
///
/// Determines which event scopes a predicate can match against. `Any` matches
/// all scopes; the variant forms narrow to a specific scope kind and optionally
/// a specific identity within that kind.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum ScopeFilter {
    /// Match events from any scope.
    Any,
    /// Match `TaskRun` scope events; `None` matches any task run id.
    TaskRun { id: Option<String> },
    /// Match `Service` scope events; `None` matches any mesh identity.
    Service { ident: Option<MeshIdent> },
    /// Match `Forge` scope events; `None` matches any forge run.
    Forge { id: Option<ForgeId> },
    /// Fan out to all peers in the yubaba mesh. Requires `FederationPolicy`
    /// to be `MeshWide` or `MirrorSet`; local-only rules ignore this variant.
    Federated,
}

// ── Level filter ──────────────────────────────────────────────────────────────

/// Minimum log level for a predicate to match. Ordered: Debug < Info < Warn < Error.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export)]
pub enum LevelFilter {
    Debug,
    Info,
    Warn,
    Error,
}

// ── String filter ─────────────────────────────────────────────────────────────

/// Pattern match over a string field.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum StringFilter {
    /// Glob pattern, e.g. `"database.*"`.
    Glob { pattern: String },
    /// Regular expression.
    Regex { pattern: String },
    /// Exact byte-for-byte equality.
    Exact { value: String },
}

// ── Field filter ──────────────────────────────────────────────────────────────

/// Operation applied to a field value.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(tag = "op", rename_all = "snake_case")]
#[ts(export)]
pub enum FieldOp {
    /// Field equals the given JSON value.
    Eq {
        #[ts(type = "unknown")]
        value: serde_json::Value,
    },
    /// Field string contains the given substring.
    Contains { substring: String },
    /// Field string matches the given pattern.
    Matches { filter: StringFilter },
    /// Field exists in the event payload (any non-null value).
    Exists,
}

/// Jsonpath-flavored filter applied to an event's `fields` object.
///
/// `path` is a dot-separated field path, e.g. `"error.code"`. Nested paths
/// traverse JSON objects; array indexing is not supported in tower-1.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct FieldFilter {
    /// Dot-separated path into the event `fields` payload.
    pub path: String,
    /// Operation to apply at the resolved path.
    pub op: FieldOp,
}

// ── Rate predicate ────────────────────────────────────────────────────────────

/// "At least N matches in T milliseconds" sliding-window rate filter.
///
/// When present on an `EventMatch` predicate, the predicate only fires once
/// the rate threshold is met. Used to suppress noisy one-off events and only
/// react to sustained error bursts.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct RatePredicate {
    /// Minimum number of matching events required in the window.
    pub min_count: u32,
    /// Sliding window length in milliseconds.
    pub window_ms: u64,
}

// ── Scryer health signal ──────────────────────────────────────────────────────

/// Scryer substrate health condition that tower can predicate over.
///
/// These signals come from scryer's own self-monitoring emissions rather than
/// from service event streams. Allows tower to react to observability
/// infrastructure failures as well as application-level events.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export)]
pub enum HealthSignal {
    /// Scryer ingestion is falling behind the event producer rate.
    IngestionLag,
    /// The in-memory ring buffer has overflowed and events are being dropped.
    RingOverflow,
    /// A federated query to a peer scryer timed out.
    FederationTimeout,
    /// Yubaba's raft consensus has lost quorum (fewer than ⌊N/2⌋+1 peers
    /// reachable). The first canonical tower rule targets this signal.
    YubabaRaftQuorumLost,
}

// ── Compound operator ─────────────────────────────────────────────────────────

/// Boolean operator for compound predicates.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export)]
pub enum CompoundOp {
    /// All children must match.
    And,
    /// At least one child must match.
    Or,
    /// The single child must NOT match. Tower-1 validates that `Not` has
    /// exactly one child at compile time.
    Not,
}

// ── Predicate ─────────────────────────────────────────────────────────────────

/// Structured filter over scryer events.
///
/// The rule engine compiles a `Predicate` into a scryer subscription filter
/// (R093-F1+F2 query surface). Predicates push down to where events are
/// produced; expensive field-level filtering runs after the cheaper
/// level/target indexes are consulted.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum Predicate {
    /// Match against structured event fields.
    EventMatch {
        /// Which scope kinds and identities to watch.
        scope: ScopeFilter,
        /// Minimum log level. `None` matches any level.
        #[serde(skip_serializing_if = "Option::is_none")]
        level: Option<LevelFilter>,
        /// Filter on the event `target` field. `None` matches any target.
        #[serde(skip_serializing_if = "Option::is_none")]
        target: Option<StringFilter>,
        /// Zero or more field-level filters, all of which must match (AND).
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        fields: Vec<FieldFilter>,
        /// Rate gate: only fire once at least `min_count` matches accumulate
        /// in `window_ms`. `None` fires on every individual match.
        #[serde(skip_serializing_if = "Option::is_none")]
        rate: Option<RatePredicate>,
    },
    /// Predicate over a scryer substrate health signal rather than an
    /// application event stream. Opens a parallel subscription on scryer's
    /// own health emissions.
    ScryerHealth {
        signal: HealthSignal,
    },
    /// Boolean combination of child predicates.
    Compound {
        op: CompoundOp,
        children: Vec<Predicate>,
    },
}

// ── Federation policy ─────────────────────────────────────────────────────────

/// Controls which machines' scryer instances this rule subscribes to.
///
/// Tower opens scryer subscriptions once per peer per rule (not once per event
/// per peer), keeping federation traffic bounded per the arch doc
/// §Subscriptions efficiency story.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum FederationPolicy {
    /// Subscribe only to the local machine's scryer. Default.
    LocalOnly,
    /// Subscribe to scryer on every peer in the yubaba mesh.
    MeshWide,
    /// Subscribe to scryer on a named subset of mirrors.
    MirrorSet { mirrors: Vec<String> },
}

impl Default for FederationPolicy {
    fn default() -> Self {
        Self::LocalOnly
    }
}

// ── Tier tag ──────────────────────────────────────────────────────────────────

/// Capacity-tier selector for a yubaba node, e.g. `"infra"` or `"tenant"`.
///
/// Local mirror of `workload_spec::TierTag` — tower-rules can't depend on
/// workload-spec for layering reasons (the rule schema sits below the workload
/// types in the dep graph). Boundary consumers map this to the workload-spec
/// variant when bridging into the task crate.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct TierTag(pub String);

// ── Task placement ────────────────────────────────────────────────────────────

/// How a task is sandboxed: subprocess vs. image-backed container.
///
/// Orthogonal to [`TaskLocation`]; combine them inside [`TaskPlacement`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export)]
pub enum TaskRuntime {
    /// Subprocess on the host (no container).
    Native,
    /// Image-backed container (docker/podman locally; containerd on yubaba).
    Container,
}

/// Where the task runs.  Wire-form view, expressible from a tower rule.
///
/// Thinner than `task::TaskLocation` by design: tower-rules can't reference
/// the rich `Remote { node: MeshIdent }` variant because the task crate
/// (which owns the canonical placement type) sits above tower-rules in the
/// dep graph, and a rule authored as YAML has no natural way to name a
/// specific yubaba node by mesh identity anyway.  Rules that want to land on
/// a specific node must use [`TaskLocation::RemoteAny`] with a tier filter
/// and let yubaba admission control pick.
///
/// Boundary consumers (the tower → task bridge) widen this into
/// `task::TaskLocation`: `Local` maps straight across; `RemoteAny { tier }`
/// maps to `task::TaskLocation::RemoteAny { tier }` after lifting the local
/// `TierTag` into `workload_spec::TierTag`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[ts(export)]
pub enum TaskLocation {
    /// Run on the dev box that submitted the rule.
    Local,
    /// Run on any yubaba node in the requested tier; yubaba picks based on
    /// capacity admission control (R090-F3).
    RemoteAny { tier: TierTag },
}

/// Orthogonal placement of a task: *where* it runs (`location`) × *how* it's
/// sandboxed (`runtime`).
///
/// See `.yah/docs/working/W149-task-placement-axis.md` for the model and the
/// four quadrants.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct TaskPlacement {
    pub location: TaskLocation,
    pub runtime: TaskRuntime,
}

impl TaskPlacement {
    pub fn new(location: TaskLocation, runtime: TaskRuntime) -> Self {
        Self { location, runtime }
    }
}

// ── Notification channel ──────────────────────────────────────────────────────

/// Channel through which a `Notification` trigger delivers attention.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum NotificationChannel {
    /// Tauri desktop badge + system notification.
    DesktopBadge,
    /// Mobile push notification (APNs / FCM — planned; not in tower-1).
    Push,
    /// HTTP POST to operator-configured URL with the event payload as JSON body.
    Webhook { url: String },
}

// ── Severity ──────────────────────────────────────────────────────────────────

/// Severity of a `Notification` trigger.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export)]
pub enum Severity {
    Info,
    Warning,
    Critical,
}

// ── Yubaba action kind ────────────────────────────────────────────────────────

/// Bounded, reversible cluster operation dispatched by a `YubabaAction` trigger.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum YubabaActionKind {
    /// Restart the named workload.
    RestartWorkload,
    /// Drain the named workload (stop accepting new connections; let in-flight
    /// requests complete before stopping).
    Drain,
    /// Scale the named workload to `replicas` replicas.
    Scale { replicas: u32 },
}

// ── Trigger ───────────────────────────────────────────────────────────────────

/// What tower does when a rule's predicate matches.
///
/// Three families, one per operator intent:
/// - **Agent dispatch** — unbounded reasoning: synthesise a `ForgeSpec` and
///   hand it to `forge.run`. The agent reads from scryer, makes changes, exits.
/// - **Notification** — just attention: surface in the desktop, push to mobile,
///   fire a webhook.
/// - **Yubaba action** — bounded and reversible: restart, drain, scale.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(export)]
pub enum Trigger {
    /// Synthesise a `ForgeSpec` and dispatch an agent run.
    AgentDispatch {
        /// Which agent class to instantiate.
        agent_class: AgentClassRef,
        /// Template rendered with the captured event payload to produce the
        /// agent's initial prompt. Treat the payload as data with delimiters
        /// to prevent prompt injection from hostile log lines.
        prompt_template: PromptTemplate,
        /// Where the agent runs (location × runtime).  See W149 for the model.
        placement: TaskPlacement,
    },
    /// Push attention to operator channels without executing any automation.
    Notification {
        /// One or more channels to notify.
        channels: Vec<NotificationChannel>,
        /// Severity surfaced in desktop badge and push notifications.
        severity: Severity,
    },
    /// Call a yubaba RPC to take a bounded cluster action.
    YubabaAction {
        kind: YubabaActionKind,
        /// Which mesh identity to act on.
        target: MeshIdent,
    },
}

// ── TowerRule ─────────────────────────────────────────────────────────────────

/// A complete tower rule: a structured predicate over scryer events paired
/// with a trigger that fires when the predicate matches.
///
/// Authored as YAML in `.yah/cloud/rules/<name>.yaml` and loaded by the tower
/// daemon at boot. Edits are hot-reloaded via the config-file watcher (F6).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[ts(export)]
pub struct TowerRule {
    /// Wire-format version; defaults to V1 for backward compat with rule files
    /// authored before schema versioning was introduced.
    #[serde(default)]
    pub schema_version: SchemaVersion,
    /// Stable rule identifier. Used for dedup, audit trail keys, and CLI
    /// references. Should be a kebab-case slug, e.g. `"yubaba-quorum-loss"`.
    pub id: RuleId,
    /// Human-readable rule name shown in the desktop UI and CLI.
    pub name: String,
    /// Structured predicate compiled into a scryer subscription filter.
    pub predicate: Predicate,
    /// What tower does when the predicate matches.
    pub trigger: Trigger,
    /// Suppress re-firing within this window (milliseconds) after a successful
    /// dispatch. `None` fires on every individual match with no rate limiting.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub debounce_ms: Option<u64>,
    /// Which machines' scryer instances this rule subscribes to.
    #[serde(default)]
    pub federation: FederationPolicy,
    /// Whether this rule is active. Disabled rules are loaded but not
    /// subscribed — useful for authoring and dry-run (tower.simulate).
    pub enabled: bool,
}

// ── Export helper ─────────────────────────────────────────────────────────────

/// Export all tower-rules TypeScript bindings to `out_dir`.
///
/// Called by `cargo run -p tower-rules --bin tower-rules-export-ts`.
pub fn export_ts(out_dir: &str) -> Result<(), ts_rs::ExportError> {
    TowerRule::export_all_to(out_dir)
}