//! `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)
}