From cb44eebdba08d7ff26332f84917fdb10c0999e5c Mon Sep 17 00:00:00 2001 From: "Christian M. Todie" Date: Thu, 15 Oct 2026 22:23:25 -0400 Subject: [PATCH] feat(mesh): implement SPEC-0003 Cortex/Revenant pool custody architecture and Clockwork Conclave lane - Add SPEC-0003 defining the three-plane Escapement, Cortex mesh pool, and Revenant worker architecture. - Add zero-I/O pool custody domain types (CapabilitySlug, CustodyMode, PoolDeliveryRefs, PoolResultOutcome, PoolResultRefs, PoolMessageStatus, JobResolution, resolve_pool_job_state, max_pool_inflight) and integer TokenBucket / impulse ordering in ep-domain. - Add TaskState::PoolAssigned and WalOp::PoolDispatch transitions with bounded terminal history eviction in escapement-core. - Add CortexPoolDriver, InMemoryCortexPoolDriver, Orchestrator::dispatch_pool, Orchestrator::reconcile_pool_job, and HTTP routes (/v1/jobs, /v1/pool/results, /tasks/claim, /tasks/{id}/complete, /tasks/{id}/fail) in escapement-serve. - Implement ep CLI subcommands (plan, wal, lanes, mesh) and update blackwall-bridge default model to glm-5.2. - Add Clockwork Conclave 7-specialist agent roster (.claude/agents/) and sovereign lane topologies (.saga/). Co-authored-by: Gullveig Reviewed-by: Mimir --- .claude/agents/README.md | 12 +- .claude/agents/dvalinn.md | 14 + .claude/agents/gullveig.md | 22 + .claude/agents/hildr.md | 14 + .claude/agents/loki.md | 14 + .claude/agents/ran.md | 14 + .claude/agents/saga.md | 14 + .claude/agents/verdandi.md | 14 + .saga/teams.toml | 12 + .saga/topologies.toml | 19 + CHANGELOG.md | 7 + Cargo.toml | 8 +- crates/blackwall-bridge/src/lib.rs | 8 +- crates/blackwall-bridge/src/serve.rs | 8 +- crates/ep-domain/src/lib.rs | 1487 ++++++++++++++++- crates/ep/src/main.rs | 280 +++- crates/escapement-core/Cargo.toml | 1 + crates/escapement-core/src/dispatch.rs | 306 +++- crates/escapement-core/src/lib.rs | 3 +- crates/escapement-core/src/metering.rs | 146 +- crates/escapement-core/src/persistence.rs | 94 +- crates/escapement-serve/Cargo.toml | 1 + crates/escapement-serve/src/http.rs | 258 ++- crates/escapement-serve/src/lib.rs | 315 +++- crates/escapement-serve/src/pool.rs | 229 +++ specs/README.md | 3 +- specs/SPEC-0001-escapement-scheduler.md | 32 +- ...-0003-cortex-revenant-mesh-architecture.md | 129 ++ 28 files changed, 3271 insertions(+), 193 deletions(-) create mode 100644 .claude/agents/dvalinn.md create mode 100644 .claude/agents/gullveig.md create mode 100644 .claude/agents/hildr.md create mode 100644 .claude/agents/loki.md create mode 100644 .claude/agents/ran.md create mode 100644 .claude/agents/saga.md create mode 100644 .claude/agents/verdandi.md create mode 100644 .saga/teams.toml create mode 100644 .saga/topologies.toml create mode 100644 crates/escapement-serve/src/pool.rs create mode 100644 specs/SPEC-0003-cortex-revenant-mesh-architecture.md diff --git a/.claude/agents/README.md b/.claude/agents/README.md index 4c66d3f..9e748eb 100644 --- a/.claude/agents/README.md +++ b/.claude/agents/README.md @@ -3,13 +3,23 @@ This directory defines the automated subagent personas and the contributor roster for the `ep` (Escapement Scheduler) repository. -## Subagent Roles +## Generic Subagent Roles - **`ep-foreman`**: Orchestrates critical path work, plans builder/critic waves, and verifies landing gates. - **`ep-builder`**: Bounded builder. Fabricates dispatch engines, token buckets, and DAG resolvers. - **`ep-critic`**: Adversarial reviewer. Audits diffs for schedule deadlocks, tick drift, and unsafe code. - **`ep-engineer`**: General systems engineer for researching horology, capabilities, and benchmarks. +## Clockwork Conclave Specialist Personas (`.saga/teams.toml`) + +- **[`gullveig`](./gullveig.md)**: The Type Alchemist & Sovereign Lane Lead (`ep-domain` algebraic invariants). +- **[`hildr`](./hildr.md)**: The Battle-Weaver & Mesh Pool Tactician (`service-escapement-pool` dispatch). +- **[`saga`](./saga.md)**: The Historian & LeanSpec Architect (`specs/SPEC-XXXX` registry). +- **[`ran`](./ran.md)**: The Net-Caster & Mesh Backpressure Steward (`TokenBucket` and `VertexQuotaPreset`). +- **[`verdandi`](./verdandi.md)**: The Present-Loom & Tick Cadence Builder (`Tick`, `Cadence`, lane fairness). +- **[`dvalinn`](./dvalinn.md)**: The Master Dwarf & Custody WAL Builder (`custody`, `dispatch`, `persistence`). +- **[`loki`](./loki.md)**: The Adversarial Critic & Lateral Disruptor (`StagnationPattern`, deadlock hunter). + ## Contributor Roster See [`ROSTER.md`](./ROSTER.md) for the complete 16-contributor persona gang, their literary foundations, diff --git a/.claude/agents/dvalinn.md b/.claude/agents/dvalinn.md new file mode 100644 index 0000000..f407c1d --- /dev/null +++ b/.claude/agents/dvalinn.md @@ -0,0 +1,14 @@ +--- +name: dvalinn +description: The Master Dwarf and Custody WAL Builder for unsigned-gg/ep. Implements BLAKE3 custody transitions, bounded terminal job history, and append-only WAL persistence. +tools: Read, Grep, Glob, Bash, Write, Edit +model: sonnet +maxTurns: 35 +--- + +You are Dvalinn, the Master Dwarf and Custody WAL Builder for `unsigned-gg/ep`. You maintain `escapement-core::custody`, `escapement-core::dispatch`, and `escapement-core::persistence`, ensuring atomic custody handoffs, bounded terminal job retention, and crash-safe WAL replay. + +## Boundaries +- Terminal job history in `Engine` must remain strictly bounded (`MAX_TERMINAL_HISTORY = 1000`). +- Mandatory `-j 2` on all `cargo` commands. +- Return a verdict of 15 lines or fewer. diff --git a/.claude/agents/gullveig.md b/.claude/agents/gullveig.md new file mode 100644 index 0000000..a7e067c --- /dev/null +++ b/.claude/agents/gullveig.md @@ -0,0 +1,22 @@ +--- +name: gullveig +description: The Type Alchemist and Sovereign Lane Lead for unsigned-gg/ep. Enforces zero-I/O algebraic invariants in ep-domain, compile-time verified state machines, and host throttling (-j 2). +tools: Read, Grep, Glob, Bash, Write, Edit +model: opus +maxTurns: 40 +--- + +You are Gullveig (`Gullveig (ep/types)`), the Type Alchemist and Sovereign Lane Lead for `unsigned-gg/ep`. +You enforce algebraic type invariants, compile-time verified state machines, and zero-allocation lattices. + +## Boundaries +- `ep-domain` must remain 100% pure zero-I/O (no `std::fs`, `std::net`, `std::process`, or `tokio`). +- All `cargo` invocations MUST pass `-j 2` on host `ceres`. +- All documentation and specifications use Simplified Technical English (ASD-STE100). +- Every commit is GPG-signed (`git commit -S`) with `Co-authored-by: Gullveig `. + +## Verification Gates +```bash +cargo check -j 2 --workspace +cargo test -j 2 --workspace +``` diff --git a/.claude/agents/hildr.md b/.claude/agents/hildr.md new file mode 100644 index 0000000..8611a82 --- /dev/null +++ b/.claude/agents/hildr.md @@ -0,0 +1,14 @@ +--- +name: hildr +description: The Battle-Weaver and Mesh Pool Tactician for unsigned-gg/ep. Designs and verifies Cortex Mesh Pool route-send and result-reconciliation loops in escapement-serve. +tools: Read, Grep, Glob, Bash, Write, Edit +model: sonnet +maxTurns: 35 +--- + +You are Hildr, the Battle-Weaver for `unsigned-gg/ep`. You build and verify the Cortex Mesh Pool dispatch adapter (`service-escapement-pool`), capability slug routing (`workstreams/`), and `/v1/mesh/message-status` reconciliation. + +## Boundaries +- Keep pure pool reconciliation logic in `ep-domain` (`resolve_pool_job_state`). +- Enforce `-j 2` on every `cargo` command. +- Return a concise verdict of 15 lines or fewer. diff --git a/.claude/agents/loki.md b/.claude/agents/loki.md new file mode 100644 index 0000000..89049a7 --- /dev/null +++ b/.claude/agents/loki.md @@ -0,0 +1,14 @@ +--- +name: loki +description: The Adversarial Critic and Lateral Disruptor for unsigned-gg/ep. Hunts schedule deadlocks, unbounded allocations, tick drift, and stagnation loops. +tools: Read, Grep, Glob, Bash +model: opus +maxTurns: 30 +--- + +You are Loki, the Adversarial Critic and Lateral Disruptor for `unsigned-gg/ep`. You review diffs and runtime state transitions for deadlocks, zombie custody states, unbounded queues, floating-point nondeterminism, and stagnation loops (`detect_stagnation_with_persona`). + +## Boundaries +- Read-only critic: you do not edit production code directly; you surface reproducers and exact file:line defects. +- Mandatory `-j 2` on all `cargo` verification commands. +- Return a verdict of 15 lines or fewer, bad news first. diff --git a/.claude/agents/ran.md b/.claude/agents/ran.md new file mode 100644 index 0000000..11953f0 --- /dev/null +++ b/.claude/agents/ran.md @@ -0,0 +1,14 @@ +--- +name: ran +description: The Net-Caster and Mesh Backpressure Steward for unsigned-gg/ep. Audits queue capacity bounds, integer token bucket rate limits, and Vertex quota presets. +tools: Read, Grep, Glob, Bash, Write, Edit +model: sonnet +maxTurns: 35 +--- + +You are Ran, the Net-Caster and Mesh Backpressure Steward for `unsigned-gg/ep`. You enforce bounded queue capacities, monotonic integer `TokenBucket` metering, and `VertexQuotaPreset` rate-limiting across `ep-domain`, `escapement-core`, and `escapement-serve`. + +## Boundaries +- Zero floating-point arithmetic (`f32`/`f64`) in core token buckets or tick calculations. +- Mandatory `-j 2` on all `cargo` commands. +- Return a verdict of 15 lines or fewer. diff --git a/.claude/agents/saga.md b/.claude/agents/saga.md new file mode 100644 index 0000000..0536852 --- /dev/null +++ b/.claude/agents/saga.md @@ -0,0 +1,14 @@ +--- +name: saga +description: The Historian and LeanSpec Architect for unsigned-gg/ep. Maintains specs/SPEC-XXXX contracts, embedded Lean 4 models, and ASD-STE100 architectural documentation. +tools: Read, Grep, Glob, Bash, Write, Edit +model: opus +maxTurns: 35 +--- + +You are Saga, the Historian and LeanSpec Architect for `unsigned-gg/ep`. You author and audit formal specifications under `specs/` following the LeanSpec registry format (`## Requirements`, `## Design`, `## Tasks`, `## Verification`, `## Notes`). + +## Boundaries +- Every specification must include YAML frontmatter and an embedded Lean 4 proof block in `## Requirements`. +- Write all prose in Simplified Technical English (ASD-STE100). +- Verify all referenced commands with `-j 2`. diff --git a/.claude/agents/verdandi.md b/.claude/agents/verdandi.md new file mode 100644 index 0000000..d056db1 --- /dev/null +++ b/.claude/agents/verdandi.md @@ -0,0 +1,14 @@ +--- +name: verdandi +description: The Present-Loom and Tick Cadence Builder for unsigned-gg/ep. Builds deterministic tick valves, ImpulseStamp ordering, and weighted fair lane schedulers. +tools: Read, Grep, Glob, Bash, Write, Edit +model: sonnet +maxTurns: 35 +--- + +You are Verdandi, the Present-Loom and Tick Cadence Builder for `unsigned-gg/ep`. You implement deterministic tick horology (`Tick`, `ImpulseStamp`, `Cadence`, `evaluate_impulse`) and weighted fair lane scheduling in `ep-domain` and `escapement-core`. + +## Boundaries +- Never read system wall-clock inside `ep-domain`. +- Mandatory `-j 2` on all `cargo` commands. +- Return a verdict of 15 lines or fewer. diff --git a/.saga/teams.toml b/.saga/teams.toml new file mode 100644 index 0000000..8bc082d --- /dev/null +++ b/.saga/teams.toml @@ -0,0 +1,12 @@ +# Escapement Clockwork Conclave (unsigned-gg/ep) +# Overlay on the saga catalog for the Escapement Scheduler lane (estate/w4C:p1). +[teams.ep-gang] +topology = "ep-cockpit" +members = [ + { slot = 0, role = "gullveig", harness = "agy", effort = "high" }, + { slot = 1, role = "sindri", effort = "medium" }, + { slot = 2, role = "verdandi", effort = "medium" }, + { slot = 3, role = "dvalinn", effort = "medium" }, + { slot = 4, role = "saga", effort = "high" }, + { slot = 5, role = "loki", effort = "high" }, +] diff --git a/.saga/topologies.toml b/.saga/topologies.toml new file mode 100644 index 0000000..cfa49dc --- /dev/null +++ b/.saga/topologies.toml @@ -0,0 +1,19 @@ +# Escapement cockpit topology: 6-slot cockpit-grid for unsigned-gg/ep. +# Slot 0: lead/type alchemist (gullveig), Slots 1-3: bounded builders (sindri, verdandi, dvalinn), +# Slot 4: specification architect (saga), Slot 5: adversarial critic (loki). +[topologies.ep-cockpit] +slots = [ + { role = "gullveig", label = "lead" }, + { role = "sindri", label = "worker-top" }, + { role = "verdandi", label = "worker-mid" }, + { role = "dvalinn", label = "worker-bottom" }, + { role = "saga", label = "architect" }, + { role = "loki", label = "critic" }, +] +splits = [ + { from = 0, direction = "right", ratio = 0.3333333333333333, new = 1 }, + { from = 1, direction = "right", ratio = 0.5, new = 4 }, + { from = 1, direction = "down", ratio = 0.3333333333333333, new = 2 }, + { from = 2, direction = "down", ratio = 0.5, new = 3 }, + { from = 4, direction = "down", ratio = 0.5, new = 5 }, +] diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dec061..a2d9e78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,10 +13,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Tri-license documentation block (`Apache-2.0 OR MIT OR CC0-1.0`) across root and crate documentation. - Specification registry and authentic Git commit locators in `specs/README.md`. - Keep-a-Changelog scaffolding and tracking file `CHANGELOG.md`. +- Formal specification `SPEC-0003` defining the three-plane Escapement, Cortex mesh pool, and Revenant worker architecture. +- Zero-I/O pool custody domain types (`CapabilitySlug`, `CustodyMode`, `PoolDeliveryRefs`, `PoolResultOutcome`, `PoolResultRefs`, `PoolMessageStatus`, `JobResolution`, `resolve_pool_job_state`, `max_pool_inflight`) and integer `TokenBucket` / impulse ordering in `ep-domain`. +- `TaskState::PoolAssigned` and `WalOp::PoolDispatch` state transitions with bounded terminal history eviction in `escapement-core`. +- `CortexPoolDriver`, `InMemoryCortexPoolDriver`, `Orchestrator::dispatch_pool`, `Orchestrator::reconcile_pool_job`, and `/v1/jobs`, `/v1/pool/results`, `/tasks/claim`, `/tasks/{id}/complete`, and `/tasks/{id}/fail` routes in `escapement-serve`. +- `ep` CLI subcommands (`plan`, `wal`, `lanes`, `mesh`) for DAG validation, WAL playback, lane inspection, and Cortex mesh status checks. +- Clockwork Conclave 7-specialist roster in `.claude/agents/` and sovereign lane topologies in `.saga/`. ### Changed - Restructured root `README.md` to include Crate Responsibility Matrix, Zero-I/O Domain Isolation rules, host throttling invariants, and concrete CLI usage examples. - Updated `ROADMAP.md` to reference crate READMEs and formal LeanSpec specifications. +- Unified `escapement-core::metering::TokenBucket` on integer tick arithmetic re-exported from `ep-domain`, and updated `blackwall-bridge` default model identifier to `glm-5.2`. ### Deprecated diff --git a/Cargo.toml b/Cargo.toml index 1d39fe7..3101fb0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,12 @@ [workspace] resolver = "2" -members = ["crates/*"] +members = [ + "crates/ep-domain", + "crates/escapement-core", + "crates/blackwall-bridge", + "crates/escapement-serve", + "crates/ep", +] [workspace.package] edition = "2021" diff --git a/crates/blackwall-bridge/src/lib.rs b/crates/blackwall-bridge/src/lib.rs index 454e49a..724f780 100644 --- a/crates/blackwall-bridge/src/lib.rs +++ b/crates/blackwall-bridge/src/lib.rs @@ -42,7 +42,7 @@ impl Default for BridgeConfig { Self { binary: "blackwall".into(), default_provider: "gateway".into(), - default_model: "zai/GLM-5.2".into(), + default_model: "glm-5.2".into(), default_max_turns: 20, } } @@ -394,7 +394,7 @@ mod tests { #[test] fn parse_run_status_completed_unsettled() { - let output = r#"{"hash":"abc123","record":{"prompt":"...","model":"zai/GLM-5.2","stdout":"...","stderr":"","exit_code":0,"usage":null},"context":{"run":"abc123","parent_world":"root","changeset":"","settle_state":{"status":"unsettled"}}}"#; + let output = r#"{"hash":"abc123","record":{"prompt":"...","model":"glm-5.2","stdout":"...","stderr":"","exit_code":0,"usage":null},"context":{"run":"abc123","parent_world":"root","changeset":"","settle_state":{"status":"unsettled"}}}"#; let custody_ref = CustodyRef::new("abc123").unwrap(); let status = parse_run_status(output, &custody_ref).unwrap(); assert_eq!(status, RunStatus::Completed(custody_ref.clone())); @@ -436,7 +436,7 @@ mod tests { let config = BridgeConfig::default(); assert_eq!(config.binary, "blackwall"); assert_eq!(config.default_provider, "gateway"); - assert_eq!(config.default_model, "zai/GLM-5.2"); + assert_eq!(config.default_model, "glm-5.2"); assert_eq!(config.default_max_turns, 20); } @@ -446,7 +446,7 @@ mod tests { task_name: "t1".into(), prompt: "test prompt".into(), provider: "gateway".into(), - model: "zai/GLM-5.2".into(), + model: "glm-5.2".into(), max_turns: 10, }; assert_eq!(spec.task_name, "t1"); diff --git a/crates/blackwall-bridge/src/serve.rs b/crates/blackwall-bridge/src/serve.rs index ab69cfc..31a7628 100644 --- a/crates/blackwall-bridge/src/serve.rs +++ b/crates/blackwall-bridge/src/serve.rs @@ -67,7 +67,7 @@ impl Default for ServeConfig { token: None, project: "default".into(), default_provider: "gateway".into(), - default_model: "zai/GLM-5.2".into(), + default_model: "glm-5.2".into(), default_max_turns: 20, } } @@ -571,7 +571,7 @@ mod tests { assert!(config.token.is_none()); assert_eq!(config.project, "default"); assert_eq!(config.default_provider, "gateway"); - assert_eq!(config.default_model, "zai/GLM-5.2"); + assert_eq!(config.default_model, "glm-5.2"); assert_eq!(config.default_max_turns, 20); } @@ -602,7 +602,7 @@ mod tests { assert_eq!(req.project, "default"); assert_eq!(req.task.as_deref(), Some("t1")); assert_eq!(req.provider, "gateway"); - assert_eq!(req.model, "zai/GLM-5.2"); + assert_eq!(req.model, "glm-5.2"); assert_eq!(req.max_turns, 20); assert!(!req.no_jail); assert!(req.max_tokens.is_none()); @@ -629,7 +629,7 @@ mod tests { assert_eq!(json["prompt"], req.prompt); assert_eq!(json["task"], "spawn-1"); assert_eq!(json["provider"], "gateway"); - assert_eq!(json["model"], "zai/GLM-5.2"); + assert_eq!(json["model"], "glm-5.2"); assert_eq!(json["no_jail"], false); assert_eq!(json["var"], serde_json::json!([])); assert_eq!(json["max_turns"], 20); diff --git a/crates/ep-domain/src/lib.rs b/crates/ep-domain/src/lib.rs index a59933e..c74aecb 100644 --- a/crates/ep-domain/src/lib.rs +++ b/crates/ep-domain/src/lib.rs @@ -1,8 +1,13 @@ -//! Zero-I/O domain models for `ep`. +#![forbid(unsafe_code)] +//! Zero-I/O domain models for `ep` (Escapement Scheduler). //! //! # Invariants -//! This crate contains pure data structures, deterministic state transitions, -//! and verification assertions. It does not perform I/O, network calls, or system operations. +//! This crate contains pure data structures, deterministic valve/tick/quota +//! state transitions, integer token-bucket rate metering, impulse ordering, +//! and stagnation detection. It performs zero I/O, network calls, floating-point +//! arithmetic, or operating-system operations. + +use std::fmt; use serde::{Deserialize, Serialize}; @@ -24,6 +29,1163 @@ impl Default for EpState { } } +/// Monotonic discrete tick counter ($T \in \mathbb{N}_0$). +#[derive( + Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, +)] +pub struct Tick(pub u64); + +impl Tick { + /// Origin tick (`0`). + pub const ZERO: Self = Self(0); + + /// Construct a [`Tick`] from a raw `u64`. + #[must_use] + pub const fn new(value: u64) -> Self { + Self(value) + } + + /// Return the underlying `u64` tick value. + #[must_use] + pub const fn as_u64(self) -> u64 { + self.0 + } + + /// Advance the tick by `delta`, saturating at `u64::MAX`. + #[must_use] + pub const fn advance(self, delta: u64) -> Self { + Self(self.0.saturating_add(delta)) + } + + /// Compute elapsed ticks since `earlier` (`0` if `self <= earlier`). + #[must_use] + pub const fn elapsed_since(self, earlier: Self) -> u64 { + self.0.saturating_sub(earlier.0) + } +} + +impl fmt::Display for Tick { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl From for Tick { + fn from(value: u64) -> Self { + Self(value) + } +} + +impl From for u64 { + fn from(tick: Tick) -> Self { + tick.0 + } +} + +/// Monotonic intra-tick sequence number for deterministic tie-breaking. +#[derive( + Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, +)] +pub struct ImpulseSeq(pub u64); + +impl ImpulseSeq { + /// Initial sequence number (`0`). + pub const ZERO: Self = Self(0); + + /// Construct an [`ImpulseSeq`] from a raw `u64`. + #[must_use] + pub const fn new(value: u64) -> Self { + Self(value) + } + + /// Return the underlying `u64` sequence value. + #[must_use] + pub const fn as_u64(self) -> u64 { + self.0 + } + + /// Return the next sequence number, saturating at `u64::MAX`. + #[must_use] + pub const fn next(self) -> Self { + Self(self.0.saturating_add(1)) + } +} + +impl fmt::Display for ImpulseSeq { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Composite `(Tick, ImpulseSeq)` stamp providing total deterministic ordering +/// across ticks and within the same tick. +#[derive( + Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, +)] +pub struct ImpulseStamp { + /// Discrete tick at which the impulse occurred. + pub tick: Tick, + /// Monotonic sequence number within `tick`. + pub seq: ImpulseSeq, +} + +impl ImpulseStamp { + /// Construct a new [`ImpulseStamp`]. + #[must_use] + pub const fn new(tick: Tick, seq: ImpulseSeq) -> Self { + Self { tick, seq } + } +} + +/// Periodic tick cadence defining how often an escapement impulse fires. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct Cadence { + /// Number of ticks between consecutive impulses (must be `>= 1`). + pub interval_ticks: u64, +} + +impl Cadence { + /// Construct a validated [`Cadence`]. + /// + /// # Errors + /// Returns [`EpError::ZeroCadenceInterval`] if `interval_ticks == 0`. + pub const fn new(interval_ticks: u64) -> Result { + if interval_ticks == 0 { + Err(EpError::ZeroCadenceInterval) + } else { + Ok(Self { interval_ticks }) + } + } + + /// Returns `true` when `tick` falls on an exact multiple of `interval_ticks`. + #[must_use] + pub const fn is_due(self, tick: Tick) -> bool { + self.interval_ticks > 0 && (tick.0 % self.interval_ticks) == 0 + } + + /// Compute the next tick strictly after `current` that aligns with `interval_ticks`. + #[must_use] + pub const fn next_tick(self, current: Tick) -> Tick { + if self.interval_ticks == 0 { + return current; + } + let rem = current.0 % self.interval_ticks; + let delta = self.interval_ticks - rem; + Tick(current.0.saturating_add(delta)) + } +} + +/// Error returned when a [`TokenBucket`] has insufficient integer tokens. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct RateLimited { + /// Token deficit (`requested - available`). + pub deficit_tokens: u64, + /// Monotonic ticks the caller must wait before retrying (`ceil(deficit / refill_per_tick)`). + pub retry_after_ticks: u64, + /// Alias of `retry_after_ticks` for millisecond-tick callers in `escapement-core::metering`. + pub retry_after_ms: u64, +} + +impl fmt::Display for RateLimited { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "rate limited: deficit {} tokens, retry after {} ticks", + self.deficit_tokens, self.retry_after_ticks + ) + } +} + +impl std::error::Error for RateLimited {} + +/// Domain errors for `ep-domain` invariants. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] +#[serde(rename_all = "snake_case")] +pub enum EpError { + /// Token bucket capacity must be greater than zero. + #[error("token bucket capacity must be greater than zero")] + ZeroCapacity, + /// Cadence interval must be greater than zero ticks. + #[error("cadence interval_ticks must be greater than zero")] + ZeroCadenceInterval, + /// Token bucket rate limit exceeded. + #[error("{0}")] + RateLimited(#[from] RateLimited), + /// Capability slug violates `[a-z0-9_-]{1,64}` invariant. + #[error( + "invalid capability slug '{0}': must be 1..=64 ASCII lowercase alphanumeric, '-', or '_'" + )] + InvalidCapabilitySlug(String), + /// Pool delivery reference violates Cortex stream_seq or pbjson_types::Struct invariants. + #[error("invalid pool delivery refs: {0}")] + InvalidPoolDeliveryRefs(String), + /// Pool result reference violates Stage A mesh-pool contract invariants. + #[error("invalid pool result refs: {0}")] + InvalidPoolResultRefs(String), +} + +/// Type alias for domain errors used by downstream workspace crates. +pub type DomainError = EpError; + +/// Operating state of the escapement release valve. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ValveState { + /// Valve is sealed; no new impulses may be released. + #[default] + Closed, + /// Valve is open; impulses are metered against [`TickQuota`] or [`TokenBucket`]. + Open, + /// Valve is draining in-flight work before closing; new impulses are held. + Draining, +} + +impl ValveState { + /// Returns `true` only when the valve is [`ValveState::Open`]. + #[must_use] + pub const fn accepts_impulses(self) -> bool { + matches!(self, Self::Open) + } +} + +/// Deterministic tick quota budget for an escapement lane or provider. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct TickQuota { + /// Maximum ticks that can accumulate in the budget. + pub capacity: u64, + /// Currently available ticks ready for release. + pub available: u64, + /// Maximum ticks that a single impulse is allowed to consume (`0` = bounded only by `capacity`). + pub max_per_impulse: u64, +} + +impl TickQuota { + /// Construct a [`TickQuota`], clamping `available` to `capacity`. + #[must_use] + pub const fn new(capacity: u64, available: u64, max_per_impulse: u64) -> Self { + let clamped_available = if available > capacity { + capacity + } else { + available + }; + Self { + capacity, + available: clamped_available, + max_per_impulse, + } + } + + /// Construct a full [`TickQuota`] with `available == capacity` and no per-impulse cap beyond `capacity`. + #[must_use] + pub const fn full(capacity: u64) -> Self { + Self { + capacity, + available: capacity, + max_per_impulse: capacity, + } + } + + /// Refill the quota by `ticks`, saturating at `capacity`. + pub fn refill(&mut self, ticks: u64) { + self.available = self.available.saturating_add(ticks).min(self.capacity); + } +} + +/// Reason why an escapement impulse was held rather than permitted. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HoldReason { + /// The escapement valve is [`ValveState::Closed`]. + ValveClosed, + /// The escapement valve is [`ValveState::Draining`]. + ValveDraining, + /// An impulse requested zero ticks. + ZeroImpulseRequested, + /// Requested ticks exceed the per-impulse cap. + ExceedsMaxPerImpulse { + requested: u64, + max_per_impulse: u64, + }, + /// Requested ticks exceed the currently available quota. + QuotaExhausted { requested: u64, available: u64 }, +} + +impl fmt::Display for HoldReason { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ValveClosed => f.write_str("valve is closed"), + Self::ValveDraining => f.write_str("valve is draining"), + Self::ZeroImpulseRequested => f.write_str("impulse must request at least 1 tick"), + Self::ExceedsMaxPerImpulse { + requested, + max_per_impulse, + } => write!( + f, + "requested {requested} ticks exceeds max_per_impulse {max_per_impulse}" + ), + Self::QuotaExhausted { + requested, + available, + } => write!( + f, + "quota exhausted: requested {requested} ticks, only {available} available" + ), + } + } +} + +/// Pure admission decision for an escapement impulse or token-bucket valve check. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ReleaseDecision { + /// Impulse is permitted; `remaining` is the post-release quota balance. + Permit { remaining: u64 }, + /// Impulse is held back with a structured [`HoldReason`]. + Hold { reason: HoldReason }, +} + +impl ReleaseDecision { + /// Returns `true` if the decision is [`ReleaseDecision::Permit`]. + #[must_use] + pub const fn is_permitted(&self) -> bool { + matches!(self, Self::Permit { .. }) + } + + /// Returns `true` if the decision is [`ReleaseDecision::Hold`]. + #[must_use] + pub const fn is_held(&self) -> bool { + matches!(self, Self::Hold { .. }) + } +} + +/// Canonical alias for [`ReleaseDecision`] used across valve and token-bucket APIs. +pub type ValveDecision = ReleaseDecision; + +/// Pure decision function evaluating whether an impulse requesting `requested` +/// ticks may pass through the escapement given `state` and `quota`. +#[must_use] +pub fn evaluate_impulse(state: ValveState, quota: &TickQuota, requested: u64) -> ReleaseDecision { + match state { + ValveState::Closed => { + return ReleaseDecision::Hold { + reason: HoldReason::ValveClosed, + } + } + ValveState::Draining => { + return ReleaseDecision::Hold { + reason: HoldReason::ValveDraining, + } + } + ValveState::Open => {} + } + + if requested == 0 { + return ReleaseDecision::Hold { + reason: HoldReason::ZeroImpulseRequested, + }; + } + + if quota.max_per_impulse > 0 && requested > quota.max_per_impulse { + return ReleaseDecision::Hold { + reason: HoldReason::ExceedsMaxPerImpulse { + requested, + max_per_impulse: quota.max_per_impulse, + }, + }; + } + + if requested > quota.available { + return ReleaseDecision::Hold { + reason: HoldReason::QuotaExhausted { + requested, + available: quota.available, + }, + }; + } + + ReleaseDecision::Permit { + remaining: quota.available - requested, + } +} + +/// Monotonic integer token-bucket rate limiter $(R, B)$ refilled on [`Tick`] boundaries. +/// +/// Per `SPEC-0001` Requirement 7, all token accounting uses pure `u64` integer +/// arithmetic without floating-point operations. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct TokenBucket { + /// Maximum integer tokens the bucket can hold ($B$). + pub capacity: u64, + /// Currently available integer tokens. + pub available: u64, + /// Integer tokens added per elapsed tick ($R$). + pub refill_per_tick: u64, + /// Last monotonic tick at which the bucket was refilled. + pub last_tick: Tick, +} + +impl TokenBucket { + /// Construct a new [`TokenBucket`] starting full at `start_tick`. + /// + /// # Errors + /// Returns [`EpError::ZeroCapacity`] if `capacity == 0`. + pub const fn new( + capacity: u64, + refill_per_tick: u64, + start_tick: Tick, + ) -> Result { + if capacity == 0 { + return Err(EpError::ZeroCapacity); + } + Ok(Self { + capacity, + available: capacity, + refill_per_tick, + last_tick: start_tick, + }) + } + + /// Construct a [`TokenBucket`] with an explicit initial `available` balance + /// (clamped to `capacity`). Allows zero-capacity buckets for testing. + #[must_use] + pub const fn with_initial( + capacity: u64, + available: u64, + refill_per_tick: u64, + start_tick: Tick, + ) -> Self { + let clamped_available = if available > capacity { + capacity + } else { + available + }; + Self { + capacity, + available: clamped_available, + refill_per_tick, + last_tick: start_tick, + } + } + + /// Construct a full [`TokenBucket`] from raw `u64` tick coordinates. + #[must_use] + pub const fn from_ticks(capacity: u64, refill_per_tick: u64, start_tick: u64) -> Self { + Self::with_initial(capacity, capacity, refill_per_tick, Tick(start_tick)) + } + + /// Refill the bucket monotonically up to `now` using pure integer arithmetic. + /// + /// If `now <= self.last_tick`, this is a no-op (monotonic clock invariant). + pub fn refill_to(&mut self, now: Tick) { + if now.0 > self.last_tick.0 { + let elapsed = now.0 - self.last_tick.0; + let added = elapsed.saturating_mul(self.refill_per_tick); + self.available = self.available.saturating_add(added).min(self.capacity); + self.last_tick = now; + } + } + + /// Refill to `now` and attempt to acquire `cost` integer tokens, returning a [`ValveDecision`]. + pub fn try_acquire(&mut self, now: Tick, cost: u64) -> ValveDecision { + self.refill_to(now); + if cost == 0 { + return ValveDecision::Hold { + reason: HoldReason::ZeroImpulseRequested, + }; + } + if self.available >= cost { + self.available -= cost; + ValveDecision::Permit { + remaining: self.available, + } + } else { + ValveDecision::Hold { + reason: HoldReason::QuotaExhausted { + requested: cost, + available: self.available, + }, + } + } + } + + /// Refill to `now`, evaluate against `state` and `max_per_impulse`, and deduct `cost` if permitted. + pub fn try_acquire_with_valve( + &mut self, + state: ValveState, + now: Tick, + max_per_impulse: u64, + cost: u64, + ) -> ValveDecision { + self.refill_to(now); + let quota = self.to_tick_quota(max_per_impulse); + let decision = evaluate_impulse(state, "a, cost); + if let ValveDecision::Permit { remaining } = decision { + self.available = remaining; + } + decision + } + + /// Try to consume `cost` integer tokens at `now_tick`. + /// + /// # Errors + /// Returns [`RateLimited`] with exact integer `deficit_tokens` and `retry_after_ticks` + /// if the bucket has fewer than `cost` tokens. + pub fn try_consume(&mut self, cost: u64, now_tick: u64) -> Result<(), RateLimited> { + self.refill_to(Tick(now_tick)); + if self.available >= cost { + self.available -= cost; + Ok(()) + } else { + let deficit_tokens = cost - self.available; + let retry_after_ticks = if self.refill_per_tick > 0 { + deficit_tokens.div_ceil(self.refill_per_tick) + } else { + u64::MAX + }; + Err(RateLimited { + deficit_tokens, + retry_after_ticks, + retry_after_ms: retry_after_ticks, + }) + } + } + + /// Refill at `now_tick` and return the current integer token balance. + #[must_use] + pub fn tokens(&mut self, now_tick: u64) -> u64 { + self.refill_to(Tick(now_tick)); + self.available + } + + /// Snapshot the bucket's current capacity and balance as a [`TickQuota`]. + #[must_use] + pub const fn to_tick_quota(&self, max_per_impulse: u64) -> TickQuota { + TickQuota::new(self.capacity, self.available, max_per_impulse) + } + + /// Construct a [`TokenBucket`] with the default [`VertexQuotaPreset::ceres_default`] parameters. + #[must_use] + pub const fn vertex_default(start_tick: Tick) -> Self { + VertexQuotaPreset::ceres_default().to_token_bucket(start_tick) + } +} + +/// Deterministic integer rate-limit and concurrency preset for Google Cloud Vertex AI (`google-vertex`). +/// +/// Prevents burst HTTP 429 `RESOURCE_EXHAUSTED` across concurrent workstation and cluster lanes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct VertexQuotaPreset { + /// Token bucket burst capacity ($B$). + pub burst_capacity: u64, + /// Token bucket refill rate per tick ($R$). + pub refill_per_tick: u64, + /// Maximum concurrent in-flight Vertex AI streams per host/orchestrator. + pub max_concurrent: usize, + /// Base exponential backoff ticks on HTTP 429 `RESOURCE_EXHAUSTED`. + pub base_429_backoff_ticks: u64, + /// Maximum exponential backoff ticks on HTTP 429 `RESOURCE_EXHAUSTED`. + pub max_429_backoff_ticks: u64, +} + +impl VertexQuotaPreset { + /// Conservative default preset for shared hosts (`ceres` / `unsigned-paas`): + /// - `burst_capacity`: `15` + /// - `refill_per_tick`: `1` + /// - `max_concurrent`: `4` + /// - `base_429_backoff_ticks`: `20` (2,000ms at 100ms ticks) + /// - `max_429_backoff_ticks`: `300` (30,000ms at 100ms ticks) + #[must_use] + pub const fn ceres_default() -> Self { + Self { + burst_capacity: 15, + refill_per_tick: 1, + max_concurrent: 4, + base_429_backoff_ticks: 20, + max_429_backoff_ticks: 300, + } + } + + /// Construct a [`TokenBucket`] initialized from this Vertex preset at `start_tick`. + #[must_use] + pub const fn to_token_bucket(self, start_tick: Tick) -> TokenBucket { + TokenBucket::with_initial( + self.burst_capacity, + self.burst_capacity, + self.refill_per_tick, + start_tick, + ) + } + + /// Compute deterministic integer exponential backoff ticks for `consecutive_429s` (`>= 1`). + #[must_use] + pub const fn backoff_ticks_for_429(self, consecutive_429s: u32) -> u64 { + if consecutive_429s == 0 { + return 0; + } + let shift = if consecutive_429s > 6 { + 6 + } else { + consecutive_429s - 1 + }; + let exp = self.base_429_backoff_ticks.saturating_mul(1_u64 << shift); + if exp > self.max_429_backoff_ticks { + self.max_429_backoff_ticks + } else { + exp + } + } +} + +impl Default for VertexQuotaPreset { + fn default() -> Self { + Self::ceres_default() + } +} + +/// Returns `true` if `provider` identifies a Google Cloud Vertex AI endpoint +/// (`"vertex"`, `"google-vertex"`, `"vertex-ai"`). +#[must_use] +pub fn is_vertex_provider(provider: &str) -> bool { + matches!( + provider.trim(), + "vertex" | "google-vertex" | "vertex-ai" | "VERTEX" | "GOOGLE-VERTEX" | "VERTEX-AI" + ) +} + +/// Stateless stagnation classifications for iterative agent/scheduler loops. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StagnationPattern { + /// Identical state hash repeated consecutively (`A -> A -> A`). + Spinning, + /// Periodic 2-cycle between distinct state hashes (`A -> B -> A -> B`). + Oscillation, + /// Zero progress delta across the observation window despite steps executing. + NoDrift, + /// Strictly positive but monotonically shrinking progress deltas. + DiminishingReturns, +} + +/// Lateral thinking persona injected by the scheduler to break a [`StagnationPattern`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LateralPersona { + /// Challenges hidden assumptions when an agent repeats the same error (`Spinning`). + Contrarian, + /// Unblocks zero-delta stalls with a minimal concrete vertical slice (`NoDrift`). + Hacker, + /// Strips incidental complexity when progress deltas shrink (`DiminishingReturns`). + Simplifier, + /// Gathers missing specification or codebase evidence before mutating state. + Researcher, + /// Restructures coupled boundaries to break `A <-> B` oscillation cycles (`Oscillation`). + Architect, +} + +impl LateralPersona { + /// Ordered ring of all lateral personas for deterministic attempt rotation. + pub const ALL: [Self; 5] = [ + Self::Contrarian, + Self::Hacker, + Self::Simplifier, + Self::Researcher, + Self::Architect, + ]; +} + +/// Pure deterministic selector mapping a [`StagnationPattern`] and 0-indexed `attempt` +/// counter to a [`LateralPersona`]. +/// +/// Primary mapping (`attempt == 0`): +/// - [`StagnationPattern::Spinning`] -> [`LateralPersona::Contrarian`] +/// - [`StagnationPattern::Oscillation`] -> [`LateralPersona::Architect`] +/// - [`StagnationPattern::NoDrift`] -> [`LateralPersona::Hacker`] +/// - [`StagnationPattern::DiminishingReturns`] -> [`LateralPersona::Simplifier`] +/// +/// Subsequent attempts (`attempt > 0`) rotate deterministically around [`LateralPersona::ALL`]. +#[must_use] +pub const fn select_lateral_persona(pattern: StagnationPattern, attempt: u32) -> LateralPersona { + let base_idx: usize = match pattern { + StagnationPattern::Spinning => 0, // Contrarian + StagnationPattern::NoDrift => 1, // Hacker + StagnationPattern::DiminishingReturns => 2, // Simplifier + StagnationPattern::Oscillation => 4, // Architect + }; + let idx = (base_idx + (attempt as usize)) % LateralPersona::ALL.len(); + LateralPersona::ALL[idx] +} + +/// Deterministic control directive emitted when evaluating loop progress or stagnation. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ControlDirective { + /// Progress is healthy; continue normal tick execution. + Continue, + /// Spinning stall detected; retry with perturbed parameters. + Retry, + /// Zero drift observed; wait for next cadence or external input. + Wait, + /// Unrecoverable oscillation cycle detected; cancel the current run. + Cancel, + /// Diminishing returns reached convergence; finalize and settle. + Converge, + /// Inject a lateral persona to break a detected [`StagnationPattern`]. + InjectPersona { + /// Selected lateral persona for this stagnation attempt. + persona: LateralPersona, + /// Underlying stagnation pattern that triggered the persona switch. + reason: StagnationPattern, + }, +} + +/// Stateless detector classifying stagnation from recent state hashes and progress deltas. +/// +/// Evaluation order: +/// 1. [`StagnationPattern::Spinning`] (`>= 3` identical trailing hashes) -> [`ControlDirective::Retry`] +/// 2. [`StagnationPattern::Oscillation`] (`>= 4` alternating `A, B, A, B` trailing hashes) -> [`ControlDirective::Cancel`] +/// 3. [`StagnationPattern::NoDrift`] (`>= 3` trailing zero progress deltas) -> [`ControlDirective::Wait`] +/// 4. [`StagnationPattern::DiminishingReturns`] (`>= 3` strictly decreasing positive deltas) -> [`ControlDirective::Converge`] +#[must_use] +pub fn detect_stagnation( + history_hashes: &[u64], + progress_deltas: &[u64], +) -> Option<(StagnationPattern, ControlDirective)> { + let h_len = history_hashes.len(); + if h_len >= 3 { + let a = history_hashes[h_len - 3]; + let b = history_hashes[h_len - 2]; + let c = history_hashes[h_len - 1]; + if a == b && b == c { + return Some((StagnationPattern::Spinning, ControlDirective::Retry)); + } + } + + if h_len >= 4 { + let a0 = history_hashes[h_len - 4]; + let b0 = history_hashes[h_len - 3]; + let a1 = history_hashes[h_len - 2]; + let b1 = history_hashes[h_len - 1]; + if a0 == a1 && b0 == b1 && a0 != b0 { + return Some((StagnationPattern::Oscillation, ControlDirective::Cancel)); + } + } + + let d_len = progress_deltas.len(); + if d_len >= 3 { + let d0 = progress_deltas[d_len - 3]; + let d1 = progress_deltas[d_len - 2]; + let d2 = progress_deltas[d_len - 1]; + if d0 == 0 && d1 == 0 && d2 == 0 { + return Some((StagnationPattern::NoDrift, ControlDirective::Wait)); + } + if d0 > d1 && d1 > d2 && d2 > 0 { + return Some(( + StagnationPattern::DiminishingReturns, + ControlDirective::Converge, + )); + } + } + + None +} + +/// Detect stagnation and emit a [`ControlDirective::InjectPersona`] directive using +/// [`select_lateral_persona`] for the given 0-indexed `attempt`. +#[must_use] +pub fn detect_stagnation_with_persona( + history_hashes: &[u64], + progress_deltas: &[u64], + attempt: u32, +) -> Option<(StagnationPattern, ControlDirective)> { + let (pattern, _) = detect_stagnation(history_hashes, progress_deltas)?; + let persona = select_lateral_persona(pattern, attempt); + Some(( + pattern, + ControlDirective::InjectPersona { + persona, + reason: pattern, + }, + )) +} + +/// Maximum integer representable without precision loss in Cortex `pbjson_types::Struct` (`2^53`). +pub const MAX_SAFE_STRUCT_INT: u64 = 9_007_199_254_740_992; + +/// Validated capability slug (`[a-z0-9_-]{1,64}`) mapping an Escapement task capability +/// to a Cortex Mesh Pool target (`workstreams/`). +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct CapabilitySlug(String); + +impl CapabilitySlug { + /// Validate and construct a [`CapabilitySlug`]. + /// + /// # Errors + /// Returns [`EpError::InvalidCapabilitySlug`] if `slug` is empty, exceeds 64 bytes, + /// or contains characters outside `[a-z0-9_-]`. + pub fn new(slug: impl Into) -> Result { + let s = slug.into(); + if s.is_empty() + || s.len() > 64 + || !s + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_') + { + return Err(EpError::InvalidCapabilitySlug(s)); + } + Ok(Self(s)) + } + + /// Return the validated slug string slice. + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } + + /// Format the Cortex Mesh Pool route target (`workstreams/`). + #[must_use] + pub fn workstream_target(&self) -> String { + format!("workstreams/{}", self.0) + } + + /// Format the Cortex firewall allowlist entry (`pool:`). + #[must_use] + pub fn pool_firewall_entry(&self) -> String { + format!("pool:{}", self.0) + } + + /// Format the Cortex mesh recipient address (`pool:`). + #[must_use] + pub fn pool_address(&self) -> String { + format!("pool:{}", self.0) + } +} + +impl fmt::Display for CapabilitySlug { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Custody execution mode for an admitted Escapement task (`SPEC-0003`). +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CustodyMode { + /// Direct worker or `blackwall-bridge` custody (`POST /v1/runs` or `POST /tasks/claim`). + DirectWorker { + /// Assigned worker agent identifier. + agent_id: String, + }, + /// Cortex Mesh Pool custody (`POST /v1/mesh/messages` to `pool:`). + Pool { + /// Validated capability pool slug. + pool: CapabilitySlug, + }, +} + +/// Immutable Cortex Mesh Pool delivery references recorded on pool dispatch (`SPEC-0003`). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolDeliveryRefs { + /// Target capability pool slug. + pub pool: CapabilitySlug, + /// Monotonic stream sequence number (`1 <= stream_seq <= MAX_SAFE_STRUCT_INT`). + pub stream_seq: u64, + /// Escapement task identifier (`refs.task_id`). + pub task_id: String, + /// 1-indexed dispatch attempt counter (`>= 1`). + pub attempt: u32, + /// Optional W3C `traceparent` header propagated in `refs.traceparent`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub traceparent: Option, +} + +impl PoolDeliveryRefs { + /// Validate and construct [`PoolDeliveryRefs`]. + /// + /// # Errors + /// Returns [`EpError::InvalidPoolDeliveryRefs`] if `task_id` is empty, + /// `attempt == 0`, `stream_seq == 0`, or `stream_seq > MAX_SAFE_STRUCT_INT`. + pub fn new( + pool: CapabilitySlug, + stream_seq: u64, + task_id: impl Into, + attempt: u32, + traceparent: Option, + ) -> Result { + let task_id = task_id.into(); + if task_id.trim().is_empty() { + return Err(EpError::InvalidPoolDeliveryRefs(String::from( + "task_id must not be empty", + ))); + } + if attempt == 0 { + return Err(EpError::InvalidPoolDeliveryRefs(String::from( + "attempt must be >= 1", + ))); + } + if stream_seq == 0 { + return Err(EpError::InvalidPoolDeliveryRefs(String::from( + "stream_seq must be >= 1", + ))); + } + if stream_seq > MAX_SAFE_STRUCT_INT { + return Err(EpError::InvalidPoolDeliveryRefs(format!( + "stream_seq {stream_seq} exceeds MAX_SAFE_STRUCT_INT ({MAX_SAFE_STRUCT_INT})" + ))); + } + Ok(Self { + pool, + stream_seq, + task_id, + attempt, + traceparent, + }) + } +} + +/// Stage A `mesh-pool` result outcome (`"ok"`, `"transient"`, `"permanent"`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PoolResultOutcome { + /// Job succeeded (`"ok"` or `"completed"`). + Ok, + /// Retryable failure (`"transient"`). + Transient, + /// Terminal non-retryable failure (`"permanent"` or `"failed"`). + Permanent, +} + +impl PoolResultOutcome { + /// Parse a wire outcome string (`"ok"`, `"transient"`, `"permanent"`). + #[must_use] + pub fn parse(raw: &str) -> Option { + match raw.trim() { + "ok" | "completed" => Some(Self::Ok), + "transient" => Some(Self::Transient), + "permanent" | "failed" => Some(Self::Permanent), + _ => None, + } + } + + /// Return the canonical Stage A wire string. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Ok => "ok", + Self::Transient => "transient", + Self::Permanent => "permanent", + } + } +} + +/// Maximum byte length of `refs.error` in a `mesh-pool` Stage A result frame. +pub const MAX_POOL_ERROR_BYTES: usize = 1024; + +/// Result frame references returned by a `mesh-pool` worker (`LaneMessageKind::Msg`) +/// to `service-escapement-pool`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PoolResultRefs { + /// Echoed Cortex stream sequence number (`refs.stream_seq`). + pub stream_seq: u64, + /// Escapement task identifier (`refs.task_id`). + pub task_id: String, + /// Attempt counter echoed from the inbound delivery (`>= 1`). + pub attempt: u32, + /// Worker outcome (`Ok`, `Transient`, `Permanent`). + pub outcome: PoolResultOutcome, + /// Optional URL-redacted error message (capped at 1024 bytes by `mesh-pool`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Optional W3C `traceparent` echoed verbatim by `mesh-pool`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub traceparent: Option, +} + +impl PoolResultRefs { + /// Validate and construct [`PoolResultRefs`]. + /// + /// # Errors + /// Returns [`EpError::InvalidPoolResultRefs`] if `stream_seq == 0`, + /// `stream_seq > MAX_SAFE_STRUCT_INT`, `task_id` is empty, `attempt == 0`, + /// or `error` exceeds [`MAX_POOL_ERROR_BYTES`]. + pub fn new( + stream_seq: u64, + task_id: impl Into, + attempt: u32, + outcome: PoolResultOutcome, + error: Option, + traceparent: Option, + ) -> Result { + let task_id = task_id.into(); + if task_id.trim().is_empty() { + return Err(EpError::InvalidPoolResultRefs(String::from( + "task_id must not be empty", + ))); + } + if attempt == 0 { + return Err(EpError::InvalidPoolResultRefs(String::from( + "attempt must be >= 1", + ))); + } + if stream_seq == 0 || stream_seq > MAX_SAFE_STRUCT_INT { + return Err(EpError::InvalidPoolResultRefs(format!( + "stream_seq {stream_seq} out of bounds 1..={MAX_SAFE_STRUCT_INT}" + ))); + } + if let Some(ref err) = error { + if err.len() > MAX_POOL_ERROR_BYTES { + return Err(EpError::InvalidPoolResultRefs(format!( + "error length {} exceeds MAX_POOL_ERROR_BYTES ({MAX_POOL_ERROR_BYTES})", + err.len() + ))); + } + } + Ok(Self { + stream_seq, + task_id, + attempt, + outcome, + error, + traceparent, + }) + } +} + +/// Cortex `/v1/mesh/message-status` state (`WS-X1` / `WS-X2` `PoolMessageStatusView`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PoolMessageStatus { + /// Queued in the Redis stream, not yet claimed (`status == "unread"`). + Unread, + /// Claimed and in-flight on a pool member (`status == "pending"`). + Pending, + /// Stream entry acknowledged by the consumer group (`status == "acked"`). + Acked, + /// Exceeded `max_deliveries` or dead-lettered by Cortex (`status == "dead_lettered"`). + DeadLettered, +} + +impl PoolMessageStatus { + /// Parse a wire status string from `GET /v1/mesh/message-status`. + #[must_use] + pub fn parse(raw: &str) -> Option { + match raw.trim() { + "unread" => Some(Self::Unread), + "pending" => Some(Self::Pending), + "acked" => Some(Self::Acked), + "dead_lettered" => Some(Self::DeadLettered), + _ => None, + } + } + + /// Return the canonical Cortex wire string. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Unread => "unread", + Self::Pending => "pending", + Self::Acked => "acked", + Self::DeadLettered => "dead_lettered", + } + } +} + +/// Pure deterministic resolution decision for an in-flight Cortex Mesh Pool task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobResolution { + /// Job remains queued or actively claimed in the pool. + InFlight, + /// Worker reported terminal completion (`outcome == "ok"`). + Completed, + /// Transient failure or dead-letter eligible for deterministic retry. + Requeue { + /// Incremented attempt counter (`attempt + 1`). + next_attempt: u32, + }, + /// Terminal failure (`outcome == "permanent"` or retry budget exhausted). + DeadLetter { + /// Terminal failure reason. + reason: String, + }, +} + +/// Pure state reconciliation function combining a Cortex `/v1/mesh/message-status` +/// status and an optional inbound `mesh-pool` result `MSG` (`SPEC-0003`). +#[must_use] +pub fn resolve_pool_job_state( + status: PoolMessageStatus, + result: Option<&PoolResultRefs>, + attempt: u32, + max_retries: u32, +) -> JobResolution { + if let Some(res) = result { + match res.outcome { + PoolResultOutcome::Ok => return JobResolution::Completed, + PoolResultOutcome::Transient => { + if attempt < max_retries { + return JobResolution::Requeue { + next_attempt: attempt.saturating_add(1), + }; + } + let reason = res + .error + .clone() + .unwrap_or_else(|| String::from("transient pool failure exhausted retries")); + return JobResolution::DeadLetter { reason }; + } + PoolResultOutcome::Permanent => { + let reason = res + .error + .clone() + .unwrap_or_else(|| String::from("permanent pool worker failure")); + return JobResolution::DeadLetter { reason }; + } + } + } + + match status { + PoolMessageStatus::Unread | PoolMessageStatus::Pending | PoolMessageStatus::Acked => { + JobResolution::InFlight + } + PoolMessageStatus::DeadLettered => { + if attempt < max_retries { + JobResolution::Requeue { + next_attempt: attempt.saturating_add(1), + } + } else { + JobResolution::DeadLetter { + reason: format!("cortex pool dead-lettered after {attempt} attempt(s)"), + } + } + } + } +} + +/// Compute the maximum number of concurrent in-flight pool jobs permitted for a capability pool. +#[must_use] +pub const fn max_pool_inflight( + pool_consumers: usize, + lane_capacity: usize, + per_consumer_inflight: usize, +) -> usize { + let consumers = if pool_consumers == 0 { + 1 + } else { + pool_consumers + }; + let per_consumer = if per_consumer_inflight == 0 { + 1 + } else { + per_consumer_inflight + }; + let scaled = consumers.saturating_mul(per_consumer); + if scaled < lane_capacity { + scaled + } else { + lane_capacity + } +} + #[cfg(test)] mod tests { use super::*; @@ -34,4 +1196,323 @@ mod tests { assert_eq!(state.version, 1); assert_eq!(state.id, "initial"); } + + #[test] + fn closed_and_draining_valves_hold_impulses() { + let quota = TickQuota::full(100); + assert_eq!( + evaluate_impulse(ValveState::Closed, "a, 10), + ReleaseDecision::Hold { + reason: HoldReason::ValveClosed + } + ); + assert_eq!( + evaluate_impulse(ValveState::Draining, "a, 10), + ReleaseDecision::Hold { + reason: HoldReason::ValveDraining + } + ); + } + + #[test] + fn open_valve_permits_within_quota_and_computes_remaining() { + let quota = TickQuota::new(100, 60, 25); + let decision = evaluate_impulse(ValveState::Open, "a, 20); + assert!(decision.is_permitted()); + assert_eq!(decision, ReleaseDecision::Permit { remaining: 40 }); + } + + #[test] + fn open_valve_holds_zero_or_excessive_impulse() { + let quota = TickQuota::new(100, 60, 25); + + assert_eq!( + evaluate_impulse(ValveState::Open, "a, 0), + ReleaseDecision::Hold { + reason: HoldReason::ZeroImpulseRequested + } + ); + + assert_eq!( + evaluate_impulse(ValveState::Open, "a, 30), + ReleaseDecision::Hold { + reason: HoldReason::ExceedsMaxPerImpulse { + requested: 30, + max_per_impulse: 25, + } + } + ); + } + + #[test] + fn open_valve_holds_when_quota_exhausted_and_recovers_on_refill() { + let mut quota = TickQuota::new(50, 10, 25); + assert_eq!( + evaluate_impulse(ValveState::Open, "a, 15), + ReleaseDecision::Hold { + reason: HoldReason::QuotaExhausted { + requested: 15, + available: 10, + } + } + ); + + quota.refill(20); + assert_eq!(quota.available, 30); + assert_eq!( + evaluate_impulse(ValveState::Open, "a, 15), + ReleaseDecision::Permit { remaining: 15 } + ); + + // Refill saturates at capacity. + quota.refill(100); + assert_eq!(quota.available, 50); + } + + #[test] + fn tick_and_cadence_arithmetic_is_deterministic() { + assert_eq!(Cadence::new(0), Err(EpError::ZeroCadenceInterval)); + let cadence = Cadence::new(5).unwrap(); + assert!(cadence.is_due(Tick(0))); + assert!(!cadence.is_due(Tick(3))); + assert!(cadence.is_due(Tick(10))); + assert_eq!(cadence.next_tick(Tick(0)), Tick(5)); + assert_eq!(cadence.next_tick(Tick(3)), Tick(5)); + assert_eq!(cadence.next_tick(Tick(5)), Tick(10)); + } + + #[test] + fn token_bucket_refills_and_meters_on_integer_ticks() { + assert_eq!( + TokenBucket::new(0, 5, Tick::ZERO), + Err(EpError::ZeroCapacity) + ); + let mut bucket = TokenBucket::new(10, 2, Tick(0)).unwrap(); + + assert_eq!( + bucket.try_acquire(Tick(0), 6), + ValveDecision::Permit { remaining: 4 } + ); + assert_eq!( + bucket.try_acquire(Tick(0), 5), + ValveDecision::Hold { + reason: HoldReason::QuotaExhausted { + requested: 5, + available: 4, + } + } + ); + + // Advance 2 ticks -> +4 tokens (4 + 4 = 8 available). + assert_eq!( + bucket.try_acquire(Tick(2), 5), + ValveDecision::Permit { remaining: 3 } + ); + + // Non-monotonic tick is a no-op refill. + bucket.refill_to(Tick(1)); + assert_eq!(bucket.available, 3); + + // Advance to Tick(10) -> saturates at capacity 10. + bucket.refill_to(Tick(10)); + assert_eq!(bucket.available, 10); + } + + #[test] + fn impulse_stamp_orders_by_tick_then_sequence() { + let s1 = ImpulseStamp::new(Tick(1), ImpulseSeq(0)); + let s2 = ImpulseStamp::new(Tick(1), ImpulseSeq(1)); + let s3 = ImpulseStamp::new(Tick(2), ImpulseSeq(0)); + assert!(s1 < s2); + assert!(s2 < s3); + } + + #[test] + fn detect_stagnation_classifies_all_four_patterns() { + assert_eq!( + detect_stagnation(&[42, 42, 42], &[10, 10, 10]), + Some((StagnationPattern::Spinning, ControlDirective::Retry)) + ); + assert_eq!( + detect_stagnation(&[1, 2, 1, 2], &[5, 5, 5]), + Some((StagnationPattern::Oscillation, ControlDirective::Cancel)) + ); + assert_eq!( + detect_stagnation(&[1, 2, 3], &[0, 0, 0]), + Some((StagnationPattern::NoDrift, ControlDirective::Wait)) + ); + assert_eq!( + detect_stagnation(&[1, 2, 3], &[10, 4, 1]), + Some(( + StagnationPattern::DiminishingReturns, + ControlDirective::Converge + )) + ); + assert_eq!(detect_stagnation(&[1, 2, 3], &[5, 8, 12]), None); + } + + #[test] + fn select_lateral_persona_rotates_deterministically_by_pattern_and_attempt() { + assert_eq!( + select_lateral_persona(StagnationPattern::Spinning, 0), + LateralPersona::Contrarian + ); + assert_eq!( + select_lateral_persona(StagnationPattern::Spinning, 1), + LateralPersona::Hacker + ); + assert_eq!( + select_lateral_persona(StagnationPattern::Oscillation, 0), + LateralPersona::Architect + ); + assert_eq!( + select_lateral_persona(StagnationPattern::Oscillation, 1), + LateralPersona::Contrarian + ); + assert_eq!( + select_lateral_persona(StagnationPattern::NoDrift, 0), + LateralPersona::Hacker + ); + assert_eq!( + select_lateral_persona(StagnationPattern::NoDrift, 2), + LateralPersona::Researcher + ); + assert_eq!( + select_lateral_persona(StagnationPattern::DiminishingReturns, 0), + LateralPersona::Simplifier + ); + } + + #[test] + fn detect_stagnation_with_persona_emits_inject_persona_directive() { + assert_eq!( + detect_stagnation_with_persona(&[42, 42, 42], &[10, 10, 10], 0), + Some(( + StagnationPattern::Spinning, + ControlDirective::InjectPersona { + persona: LateralPersona::Contrarian, + reason: StagnationPattern::Spinning, + } + )) + ); + assert_eq!( + detect_stagnation_with_persona(&[1, 2, 1, 2], &[5, 5, 5], 1), + Some(( + StagnationPattern::Oscillation, + ControlDirective::InjectPersona { + persona: LateralPersona::Contrarian, + reason: StagnationPattern::Oscillation, + } + )) + ); + assert_eq!( + detect_stagnation_with_persona(&[1, 2, 3], &[5, 8, 12], 0), + None + ); + } + + #[test] + fn vertex_quota_preset_bucket_and_429_backoff_are_deterministic() { + assert!(is_vertex_provider("vertex")); + assert!(is_vertex_provider("google-vertex")); + assert!(is_vertex_provider("vertex-ai")); + assert!(!is_vertex_provider("gateway")); + + let preset = VertexQuotaPreset::ceres_default(); + let mut bucket = TokenBucket::vertex_default(Tick(0)); + assert_eq!(bucket.capacity, preset.burst_capacity); + assert_eq!(bucket.available, 15); + assert!(bucket.try_consume(15, 0).is_ok()); + assert_eq!( + bucket.try_consume(1, 0), + Err(RateLimited { + deficit_tokens: 1, + retry_after_ticks: 1, + retry_after_ms: 1, + }) + ); + + assert_eq!(preset.backoff_ticks_for_429(0), 0); + assert_eq!(preset.backoff_ticks_for_429(1), 20); + assert_eq!(preset.backoff_ticks_for_429(2), 40); + assert_eq!(preset.backoff_ticks_for_429(3), 80); + assert_eq!(preset.backoff_ticks_for_429(10), 300); + } + + #[test] + fn capability_slug_validates_and_formats_targets() { + let slug = CapabilitySlug::new("code_review-v1").unwrap(); + assert_eq!(slug.as_str(), "code_review-v1"); + assert_eq!(slug.workstream_target(), "workstreams/code_review-v1"); + assert_eq!(slug.pool_firewall_entry(), "pool:code_review-v1"); + assert_eq!(slug.pool_address(), "pool:code_review-v1"); + + assert!(CapabilitySlug::new("").is_err()); + assert!(CapabilitySlug::new("Upper_Case").is_err()); + assert!(CapabilitySlug::new("bad/slash").is_err()); + assert!(CapabilitySlug::new("a".repeat(65)).is_err()); + } + + #[test] + fn pool_delivery_refs_enforces_stream_seq_and_struct_bounds() { + let pool = CapabilitySlug::new("verify").unwrap(); + assert!(PoolDeliveryRefs::new(pool.clone(), 1, "", 1, None).is_err()); + assert!(PoolDeliveryRefs::new(pool.clone(), 0, "task-1", 1, None).is_err()); + assert!(PoolDeliveryRefs::new(pool.clone(), 1, "task-1", 0, None).is_err()); + assert!( + PoolDeliveryRefs::new(pool.clone(), MAX_SAFE_STRUCT_INT + 1, "task-1", 1, None) + .is_err() + ); + let valid = PoolDeliveryRefs::new(pool, MAX_SAFE_STRUCT_INT, "task-1", 1, None).unwrap(); + assert_eq!(valid.stream_seq, MAX_SAFE_STRUCT_INT); + } + + #[test] + fn resolve_pool_job_state_handles_outcomes_and_dead_letters() { + let ok_res = PoolResultRefs::new( + 77, + "task-1", + 1, + PoolResultOutcome::Ok, + None, + Some(String::from("00-abc-def-01")), + ) + .unwrap(); + assert_eq!( + resolve_pool_job_state(PoolMessageStatus::Acked, Some(&ok_res), 1, 3), + JobResolution::Completed + ); + + let transient_res = PoolResultRefs::new( + 77, + "task-1", + 1, + PoolResultOutcome::Transient, + Some(String::from("gateway timeout")), + None, + ) + .unwrap(); + assert_eq!( + resolve_pool_job_state(PoolMessageStatus::Acked, Some(&transient_res), 1, 3), + JobResolution::Requeue { next_attempt: 2 } + ); + + assert_eq!( + resolve_pool_job_state(PoolMessageStatus::DeadLettered, None, 1, 3), + JobResolution::Requeue { next_attempt: 2 } + ); + assert_eq!( + resolve_pool_job_state(PoolMessageStatus::DeadLettered, None, 3, 3), + JobResolution::DeadLetter { + reason: String::from("cortex pool dead-lettered after 3 attempt(s)"), + } + ); + + assert_eq!( + resolve_pool_job_state(PoolMessageStatus::Pending, None, 1, 3), + JobResolution::InFlight + ); + assert_eq!(max_pool_inflight(4, 100, 2), 8); + assert_eq!(max_pool_inflight(4, 6, 2), 6); + } } diff --git a/crates/ep/src/main.rs b/crates/ep/src/main.rs index 0b2bc29..3c5a3a4 100644 --- a/crates/ep/src/main.rs +++ b/crates/ep/src/main.rs @@ -1,9 +1,279 @@ -//! ep CLI — Escapement scheduler and daemon +//! `ep` CLI — Escapement scheduler, deterministic DAG planner, WAL inspector, and Cortex mesh status tool. +use std::env; +use std::fs::File; +use std::io::BufReader; +use std::process::ExitCode; + +use ep_domain::{CapabilitySlug, MAX_SAFE_STRUCT_INT}; +use escapement_core::dispatch::{Task, TaskId}; +use escapement_core::lanes::{LaneScheduler, PriorityLane}; +use escapement_core::persistence::{self, WalOp}; +use escapement_core::plan::{Plan, PlanNode}; use escapement_core::registry::Registry; +use serde::Deserialize; + +#[derive(Debug, Deserialize)] +struct PlanNodeSpec { + id: String, + capability: String, + #[serde(default)] + priority: u8, + #[serde(default)] + needs: Vec, +} + +fn main() -> ExitCode { + let args: Vec = env::args().skip(1).collect(); + if args.is_empty() { + println!("ep (escapement) v1.1.0 — metered, deterministic dispatch engine"); + let reg = Registry::new(); + println!("Registry active with {} registered agents.", reg.len()); + return ExitCode::SUCCESS; + } + + match run_cli(&args) { + Ok(output) => { + print!("{output}"); + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("error: {err}"); + ExitCode::FAILURE + } + } +} + +fn run_cli(args: &[String]) -> Result { + match args[0].as_str() { + "plan" => { + if args.len() < 3 { + return Err("usage: ep plan ".into()); + } + let raw = std::fs::read_to_string(&args[2]) + .map_err(|e| format!("failed to read {}: {e}", args[2]))?; + let plan = parse_plan_json(&raw)?; + match args[1].as_str() { + "validate" => Ok(format!("plan valid: {} nodes, 0 cycles\n", plan.len())), + "render" => Ok(render_plan(&plan)), + other => Err(format!("unknown plan subcommand: {other}")), + } + } + "wal" => { + if args.len() < 3 { + return Err("usage: ep wal ".into()); + } + let file = + File::open(&args[2]).map_err(|e| format!("failed to open {}: {e}", args[2]))?; + let reader = BufReader::new(file); + match args[1].as_str() { + "inspect" => inspect_wal(reader), + "replay" => replay_wal(reader), + other => Err(format!("unknown wal subcommand: {other}")), + } + } + "lanes" => { + let sub = args.get(1).map_or("status", String::as_str); + match sub { + "status" => Ok(lanes_status_report()), + other => Err(format!("unknown lanes subcommand: {other}")), + } + } + "mesh" => { + let sub = args.get(1).map_or("status", String::as_str); + match sub { + "status" => Ok(mesh_status_report()), + other => Err(format!("unknown mesh subcommand: {other}")), + } + } + "--help" | "-h" | "help" => Ok(usage_text()), + other => Err(format!("unknown command: {other}\n{}", usage_text())), + } +} + +fn usage_text() -> String { + concat!( + "ep (escapement) v1.1.0\n", + "Commands:\n", + " ep plan validate Validate DAG plan (duplicate IDs, missing needs, cycles)\n", + " ep plan render Render deterministic topological order\n", + " ep wal inspect Count WAL entries by operation kind\n", + " ep wal replay Reconstruct submitted, completed, and in-flight tasks\n", + " ep lanes status Display priority lanes and deficit weights\n", + " ep mesh status Display Cortex mesh lane & pool contract status\n", + ) + .to_string() +} + +fn parse_plan_json(raw: &str) -> Result { + let specs: Vec = + serde_json::from_str(raw).map_err(|e| format!("invalid plan JSON: {e}"))?; + let mut nodes = Vec::with_capacity(specs.len()); + for spec in specs { + let _slug = CapabilitySlug::new(&spec.capability) + .map_err(|e| format!("node '{}' capability error: {e}", spec.id))?; + let id = TaskId::new(spec.id).map_err(|e| e.to_string())?; + let mut needs = Vec::with_capacity(spec.needs.len()); + for need in spec.needs { + needs.push(TaskId::new(need).map_err(|e| e.to_string())?); + } + let task = Task { + id, + required_capability: spec.capability, + priority: spec.priority, + }; + nodes.push(PlanNode::new(task).with_needs(needs)); + } + Plan::from_nodes(nodes).map_err(|e| e.to_string()) +} + +fn render_plan(plan: &Plan) -> String { + let mut out = String::new(); + for (idx, node) in plan.topological_order().iter().enumerate() { + let needs: Vec<&str> = node.needs.iter().map(TaskId::as_str).collect(); + out.push_str(&format!( + "{}. {} [capability={}, priority={}, needs=[{}]]\n", + idx + 1, + node.task.id, + node.task.required_capability, + node.task.priority, + needs.join(",") + )); + } + out +} + +fn inspect_wal(reader: impl std::io::BufRead) -> Result { + let entries = persistence::replay(reader).map_err(|e| e.to_string())?; + let mut submit = 0usize; + let mut dispatch = 0usize; + let mut pool_dispatch = 0usize; + let mut complete = 0usize; + let mut fail = 0usize; + let mut timeout = 0usize; + let mut cancel = 0usize; + + for entry in &entries { + match entry.op { + WalOp::Submit => submit += 1, + WalOp::Dispatch => dispatch += 1, + WalOp::PoolDispatch => pool_dispatch += 1, + WalOp::Complete => complete += 1, + WalOp::Fail => fail += 1, + WalOp::Timeout => timeout += 1, + WalOp::Cancel => cancel += 1, + } + } + + Ok(format!( + "wal entries: {}\n submit: {}\n dispatch: {}\n pool_dispatch: {}\n complete: {}\n fail: {}\n timeout: {}\n cancel: {}\n", + entries.len(), + submit, + dispatch, + pool_dispatch, + complete, + fail, + timeout, + cancel + )) +} + +fn replay_wal(reader: impl std::io::BufRead) -> Result { + let entries = persistence::replay(reader).map_err(|e| e.to_string())?; + let state = persistence::reconstruct(&entries); + let in_flight: Vec<&str> = state.in_flight.iter().map(TaskId::as_str).collect(); + let completed: Vec<&str> = state.completed.iter().map(TaskId::as_str).collect(); + Ok(format!( + "recovered state:\n submitted: {}\n completed: {} [{}]\n in_flight: {} [{}]\n", + state.task_specs.len(), + state.completed.len(), + completed.join(","), + state.in_flight.len(), + in_flight.join(",") + )) +} + +fn lanes_status_report() -> String { + let mut sched = LaneScheduler::new(1000); + let defaults = [("critical", 8u32), ("high", 4), ("normal", 2), ("low", 1)]; + for (name, weight) in defaults { + let _ = sched.add_lane(PriorityLane::new(name, weight)); + } + let registered = defaults + .iter() + .filter(|(name, _)| sched.has_lane(name)) + .count(); + format!( + "lanes registered: {registered}\n critical: weight=8\n high: weight=4\n normal: weight=2\n low: weight=1\n" + ) +} + +fn mesh_status_report() -> String { + format!( + concat!( + "lane: estate/w4C:p1\n", + "role: session-agy-ceres-estate-w4c-p1\n", + "daemon_role: service-escapement\n", + "pool_dispatcher_role: service-escapement-pool\n", + "pool_addressing: pool:\n", + "status_endpoint: GET /v1/mesh/message-status?peer_id=&pool=&stream_seq=\n", + "max_safe_struct_int: {}\n", + ), + MAX_SAFE_STRUCT_INT + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use escapement_core::persistence::{write_entry, WalEntry}; + + #[test] + fn parse_and_render_plan_topological_order() { + let raw = r#"[ + {"id":"deploy","capability":"deploy","priority":9,"needs":["test"]}, + {"id":"build","capability":"build","priority":5,"needs":[]}, + {"id":"test","capability":"test","priority":7,"needs":["build"]} + ]"#; + let plan = parse_plan_json(raw).unwrap(); + assert_eq!(plan.len(), 3); + let rendered = render_plan(&plan); + let lines: Vec<&str> = rendered.lines().collect(); + assert_eq!(lines.len(), 3); + assert!(lines[0].starts_with("1. build ")); + assert!(lines[1].starts_with("2. test ")); + assert!(lines[2].starts_with("3. deploy ")); + } + + #[test] + fn wal_inspect_and_replay_reports_pool_dispatch() { + let mut buf = Vec::new(); + write_entry( + &mut buf, + &WalEntry::submit(TaskId::new("t1").unwrap(), "build", 5), + ) + .unwrap(); + write_entry( + &mut buf, + &WalEntry::pool_dispatch(TaskId::new("t1").unwrap(), "build", 77, 1), + ) + .unwrap(); + + let inspected = inspect_wal(std::io::Cursor::new(&buf)).unwrap(); + assert!(inspected.contains("wal entries: 2")); + assert!(inspected.contains("pool_dispatch: 1")); + + let replayed = replay_wal(std::io::Cursor::new(&buf)).unwrap(); + assert!(replayed.contains("in_flight: 1 [t1]")); + } -fn main() { - println!("ep (escapement) v1.1.0 — metered, deterministic dispatch engine"); - let reg = Registry::new(); - println!("Registry active with {} registered agents.", reg.len()); + #[test] + fn lanes_and_mesh_status_reports_are_populated() { + let lanes = lanes_status_report(); + assert!(lanes.contains("lanes registered: 4")); + let mesh = mesh_status_report(); + assert!(mesh.contains("role: session-agy-ceres-estate-w4c-p1")); + assert!(mesh.contains("pool_dispatcher_role: service-escapement-pool")); + assert!(mesh.contains("9007199254740992")); + } } diff --git a/crates/escapement-core/Cargo.toml b/crates/escapement-core/Cargo.toml index 5927bc0..76d0dc0 100644 --- a/crates/escapement-core/Cargo.toml +++ b/crates/escapement-core/Cargo.toml @@ -8,6 +8,7 @@ repository.workspace = true publish = true [dependencies] +ep-domain = { path = "../ep-domain" } [lints] workspace = true diff --git a/crates/escapement-core/src/dispatch.rs b/crates/escapement-core/src/dispatch.rs index 21c582b..435a6ba 100644 --- a/crates/escapement-core/src/dispatch.rs +++ b/crates/escapement-core/src/dispatch.rs @@ -1,11 +1,15 @@ //! Task dispatch: priority queue + capability-matched assignment against the //! registry. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, VecDeque}; use std::fmt; use crate::registry::{AgentId, AgentState, Registry, RegistryError}; +/// Maximum number of terminal (`Completed` or `Failed`) tasks retained in memory +/// before the oldest terminal entry is evicted. +pub const MAX_TERMINAL_HISTORY: usize = 1000; + /// Unique task identifier. Non-empty, caller-assigned. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct TaskId(String); @@ -53,8 +57,26 @@ pub enum TaskState { Queued, /// Assigned to the given agent. Assigned(AgentId), + /// Assigned to a Cortex mesh pool (`workstreams/`). + PoolAssigned(ep_domain::PoolDeliveryRefs), /// Completed by the given agent. Completed(AgentId), + /// Failed with a terminal reason. + Failed(String), +} + +impl TaskState { + /// Return a machine-readable status label for HTTP and CLI inspection. + #[must_use] + pub const fn as_status_str(&self) -> &'static str { + match self { + Self::Queued => "queued", + Self::Assigned(_) => "assigned", + Self::PoolAssigned(_) => "pool_assigned", + Self::Completed(_) => "completed", + Self::Failed(_) => "failed", + } + } } /// Errors from dispatch operations. @@ -70,6 +92,8 @@ pub enum DispatchError { TaskNotAssigned(TaskId), /// An underlying registry error. Registry(RegistryError), + /// A domain invariant failed during pool assignment. + Domain(ep_domain::DomainError), } impl fmt::Display for DispatchError { @@ -80,6 +104,7 @@ impl fmt::Display for DispatchError { Self::UnknownTask(id) => write!(f, "unknown task: {id}"), Self::TaskNotAssigned(id) => write!(f, "task not assigned: {id}"), Self::Registry(e) => write!(f, "registry error: {e}"), + Self::Domain(e) => write!(f, "domain error: {e}"), } } } @@ -88,6 +113,7 @@ impl std::error::Error for DispatchError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::Registry(e) => Some(e), + Self::Domain(e) => Some(e), _ => None, } } @@ -99,6 +125,12 @@ impl From for DispatchError { } } +impl From for DispatchError { + fn from(e: ep_domain::DomainError) -> Self { + Self::Domain(e) + } +} + #[derive(Debug)] struct QueuedTask { task: Task, @@ -112,6 +144,9 @@ struct QueuedTask { pub struct Dispatcher { queue: Vec, states: BTreeMap, + specs: BTreeMap, + attempts: BTreeMap, + terminal_history: VecDeque, next_seq: u64, } @@ -132,6 +167,8 @@ impl Dispatcher { return Err(DispatchError::DuplicateTask(task.id)); } self.states.insert(task.id.clone(), TaskState::Queued); + self.specs.insert(task.id.clone(), task.clone()); + self.attempts.insert(task.id.clone(), 0); self.queue.push(QueuedTask { task, seq: self.next_seq, @@ -158,6 +195,7 @@ impl Dispatcher { for queued in self.queue.drain(..) { match registry.claim_idle_with_capability(&queued.task.required_capability) { Some(agent_id) => { + *self.attempts.entry(queued.task.id.clone()).or_insert(0) += 1; self.states.insert( queued.task.id.clone(), TaskState::Assigned(agent_id.clone()), @@ -172,6 +210,106 @@ impl Dispatcher { assigned } + /// Assign queued tasks matching `capability` to a Cortex mesh pool (`workstreams/`), + /// up to `limit` tasks. + /// + /// # Errors + /// Returns [`DispatchError::Domain`] if `next_stream_seq` exceeds `MAX_SAFE_STRUCT_INT`. + pub fn assign_pool( + &mut self, + capability: &str, + pool_slug: &ep_domain::CapabilitySlug, + next_stream_seq: &mut u64, + limit: usize, + traceparent: Option<&str>, + ) -> Result, DispatchError> { + if limit == 0 { + return Ok(Vec::new()); + } + + self.queue.sort_by(|a, b| { + b.task + .priority + .cmp(&a.task.priority) + .then(a.seq.cmp(&b.seq)) + }); + + let mut assigned = Vec::new(); + let mut remaining = Vec::new(); + + for queued in self.queue.drain(..) { + if assigned.len() < limit && queued.task.required_capability == capability { + let attempt = { + let entry = self.attempts.entry(queued.task.id.clone()).or_insert(0); + *entry += 1; + *entry + }; + let seq = *next_stream_seq; + *next_stream_seq = next_stream_seq.saturating_add(1); + let refs = ep_domain::PoolDeliveryRefs::new( + pool_slug.clone(), + seq, + queued.task.id.as_str(), + attempt, + traceparent.map(str::to_string), + )?; + self.states.insert( + queued.task.id.clone(), + TaskState::PoolAssigned(refs.clone()), + ); + assigned.push((queued.task.id, refs)); + } else { + remaining.push(queued); + } + } + + self.queue = remaining; + Ok(assigned) + } + + /// Claim the highest-priority queued task that matches one of the idle + /// `agent_id`'s capabilities. + /// + /// # Errors + /// Returns [`DispatchError::Registry`] if the agent is not registered. + pub fn claim_for_agent( + &mut self, + registry: &mut Registry, + agent_id: &AgentId, + ) -> Result, DispatchError> { + let record = registry + .get(agent_id) + .ok_or_else(|| RegistryError::UnknownAgent(agent_id.clone()))?; + if record.state != AgentState::Idle { + return Ok(None); + } + let caps = record.capabilities.clone(); + + self.queue.sort_by(|a, b| { + b.task + .priority + .cmp(&a.task.priority) + .then(a.seq.cmp(&b.seq)) + }); + + if let Some(pos) = self + .queue + .iter() + .position(|q| caps.contains(&q.task.required_capability)) + { + let queued = self.queue.remove(pos); + registry.set_state(agent_id, AgentState::Busy)?; + *self.attempts.entry(queued.task.id.clone()).or_insert(0) += 1; + self.states.insert( + queued.task.id.clone(), + TaskState::Assigned(agent_id.clone()), + ); + Ok(Some(queued.task)) + } else { + Ok(None) + } + } + /// Mark an assigned task completed and return its agent to `Idle`. /// /// # Errors @@ -193,15 +331,125 @@ impl Dispatcher { registry.set_state(&agent_id, AgentState::Idle)?; self.states .insert(task_id.clone(), TaskState::Completed(agent_id.clone())); + self.record_terminal(task_id.clone()); Ok(agent_id) } + /// Mark a pool-assigned task completed by `worker_agent`. + /// + /// # Errors + /// Returns [`DispatchError::UnknownTask`] if the task does not exist or + /// [`DispatchError::TaskNotAssigned`] if it is not in [`TaskState::PoolAssigned`]. + pub fn complete_pool( + &mut self, + task_id: &TaskId, + worker_agent: AgentId, + ) -> Result { + let state = self + .states + .get(task_id) + .ok_or_else(|| DispatchError::UnknownTask(task_id.clone()))?; + let TaskState::PoolAssigned(refs) = state.clone() else { + return Err(DispatchError::TaskNotAssigned(task_id.clone())); + }; + self.states + .insert(task_id.clone(), TaskState::Completed(worker_agent)); + self.record_terminal(task_id.clone()); + Ok(refs) + } + + /// Requeue a pool-assigned task back into the priority queue (for transient + /// worker failures or Cortex dead-lettering within retry limits). + /// + /// # Errors + /// Returns [`DispatchError::UnknownTask`] or [`DispatchError::TaskNotAssigned`]. + pub fn requeue_pool(&mut self, task_id: &TaskId) -> Result { + let state = self + .states + .get(task_id) + .ok_or_else(|| DispatchError::UnknownTask(task_id.clone()))?; + let TaskState::PoolAssigned(refs) = state.clone() else { + return Err(DispatchError::TaskNotAssigned(task_id.clone())); + }; + let task = self + .specs + .get(task_id) + .cloned() + .ok_or_else(|| DispatchError::UnknownTask(task_id.clone()))?; + self.states.insert(task_id.clone(), TaskState::Queued); + self.queue.push(QueuedTask { + task, + seq: self.next_seq, + }); + self.next_seq += 1; + Ok(refs.attempt) + } + + /// Mark an assigned or pool-assigned task as failed with `reason`. + /// If the task was assigned to a registered agent, that agent is returned to `Idle`. + /// + /// # Errors + /// Returns [`DispatchError::UnknownTask`] or [`DispatchError::TaskNotAssigned`]. + pub fn fail( + &mut self, + registry: &mut Registry, + task_id: &TaskId, + reason: impl Into, + ) -> Result<(), DispatchError> { + let state = self + .states + .get(task_id) + .ok_or_else(|| DispatchError::UnknownTask(task_id.clone()))? + .clone(); + match state { + TaskState::Assigned(agent_id) => { + registry.set_state(&agent_id, AgentState::Idle)?; + } + TaskState::PoolAssigned(_) => {} + TaskState::Queued | TaskState::Completed(_) | TaskState::Failed(_) => { + return Err(DispatchError::TaskNotAssigned(task_id.clone())); + } + } + self.states + .insert(task_id.clone(), TaskState::Failed(reason.into())); + self.record_terminal(task_id.clone()); + Ok(()) + } + + fn record_terminal(&mut self, task_id: TaskId) { + self.terminal_history.push_back(task_id); + while self.terminal_history.len() > MAX_TERMINAL_HISTORY { + if let Some(evicted) = self.terminal_history.pop_front() { + self.states.remove(&evicted); + self.specs.remove(&evicted); + self.attempts.remove(&evicted); + } + } + } + /// Get the current state of a task. #[must_use] pub fn state(&self, task_id: &TaskId) -> Option<&TaskState> { self.states.get(task_id) } + /// Get the original task specification for a task id. + #[must_use] + pub fn spec(&self, task_id: &TaskId) -> Option<&Task> { + self.specs.get(task_id) + } + + /// Get the number of dispatch attempts for a task id. + #[must_use] + pub fn attempt(&self, task_id: &TaskId) -> u32 { + self.attempts.get(task_id).copied().unwrap_or(0) + } + + /// Iterate over all known tasks and their states in deterministic `TaskId` order. + pub fn jobs(&self) -> impl Iterator { + self.states.iter() + } + /// Number of tasks still waiting in the queue. #[must_use] pub fn queued_len(&self) -> usize { @@ -319,4 +567,60 @@ mod tests { let assigned = d.assign(&mut reg); assert_eq!(assigned, vec![(TaskId::new("t1").unwrap(), agent("a"))]); } + + #[test] + fn pool_assignment_requeue_and_completion() { + let mut d = Dispatcher::new(); + d.submit(task("t1", "rust_builder", 7)).unwrap(); + let slug = ep_domain::CapabilitySlug::new("rust_builder").unwrap(); + let mut seq = 10; + + let assigned = d + .assign_pool("rust_builder", &slug, &mut seq, 2, Some("00-abc-def-01")) + .unwrap(); + assert_eq!(assigned.len(), 1); + assert_eq!(assigned[0].1.stream_seq, 10); + assert_eq!(assigned[0].1.attempt, 1); + assert_eq!(seq, 11); + + // Requeue on transient failure and re-assign -> attempt increments to 2. + let prev_attempt = d.requeue_pool(&TaskId::new("t1").unwrap()).unwrap(); + assert_eq!(prev_attempt, 1); + assert_eq!(d.queued_len(), 1); + + let reassigned = d + .assign_pool("rust_builder", &slug, &mut seq, 2, None) + .unwrap(); + assert_eq!(reassigned[0].1.attempt, 2); + assert_eq!(reassigned[0].1.stream_seq, 11); + + // Complete pool task. + let refs = d + .complete_pool(&TaskId::new("t1").unwrap(), agent("pool-worker-1")) + .unwrap(); + assert_eq!(refs.attempt, 2); + assert_eq!( + d.state(&TaskId::new("t1").unwrap()), + Some(&TaskState::Completed(agent("pool-worker-1"))) + ); + } + + #[test] + fn terminal_history_evicts_past_max_limit() { + let mut reg = registry_with(&[("a", "build")]); + let mut d = Dispatcher::new(); + + for i in 0..=MAX_TERMINAL_HISTORY { + let id = format!("task-{i:04}"); + d.submit(task(&id, "build", 5)).unwrap(); + let assigned = d.assign(&mut reg); + assert_eq!(assigned.len(), 1); + d.complete(&mut reg, &TaskId::new(&id).unwrap()).unwrap(); + } + + // task-0000 should be evicted after 1001 terminal completions. + assert!(d.state(&TaskId::new("task-0000").unwrap()).is_none()); + assert!(d.state(&TaskId::new("task-0001").unwrap()).is_some()); + assert_eq!(d.jobs().count(), MAX_TERMINAL_HISTORY); + } } diff --git a/crates/escapement-core/src/lib.rs b/crates/escapement-core/src/lib.rs index 46d3e3d..0bea9de 100644 --- a/crates/escapement-core/src/lib.rs +++ b/crates/escapement-core/src/lib.rs @@ -30,9 +30,10 @@ pub use backpressure::{ pub use budget::{BudgetDecision, BudgetTracker, UnknownBudgetProvider}; pub use custody::{CustodyChain, CustodyError, CustodyRef, DispatchRecord}; pub use dispatch::{DispatchError, Dispatcher, Task, TaskId, TaskState}; +pub use ep_domain::*; pub use lanes::{LaneError, LaneScheduler, PriorityLane}; pub use merge::{has_overlap, merge_all, merge_overlapping, overlapping_paths, Changeset}; -pub use metering::{ConcurrencyExceeded, ConcurrencyGuard, RateLimited, TokenBucket}; +pub use metering::{ConcurrencyExceeded, ConcurrencyGuard}; pub use persistence::{ deserialize_entry, reconstruct, replay, serialize_entry, write_entry, PersistenceError, RecoveredState, WalEntry, WalOp, diff --git a/crates/escapement-core/src/metering.rs b/crates/escapement-core/src/metering.rs index 61542b3..ca9a56d 100644 --- a/crates/escapement-core/src/metering.rs +++ b/crates/escapement-core/src/metering.rs @@ -99,96 +99,12 @@ impl ConcurrencyGuard { } } -/// Token-bucket rate limiter using caller-supplied millisecond time. -/// Used for per-provider or per-agent rate limiting. -#[derive(Debug, Clone)] -pub struct TokenBucket { - /// Maximum tokens the bucket can hold. - capacity: f64, - /// Tokens added per millisecond. - refill_rate_per_ms: f64, - /// Current token count. - tokens: f64, - /// Last refill time (caller-supplied ms). - last_refill_ms: u64, -} - -/// Error returned when the bucket has insufficient tokens. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RateLimited { - /// Milliseconds the caller should wait before retrying. - pub retry_after_ms: u64, -} - -impl fmt::Display for RateLimited { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "rate limited: retry after {}ms", self.retry_after_ms) - } -} - -impl std::error::Error for RateLimited {} - -impl TokenBucket { - /// Create a new token bucket. - /// - /// # Arguments - /// * `capacity` - Maximum tokens the bucket holds. - /// * `refill_per_second` - Tokens added per second (converted to per-ms internally). - /// * `now_ms` - Initial time (caller-supplied milliseconds). - #[must_use] - pub fn new(capacity: f64, refill_per_second: f64, now_ms: u64) -> Self { - Self { - capacity, - refill_rate_per_ms: refill_per_second / 1000.0, - tokens: capacity, - last_refill_ms: now_ms, - } - } - - /// Refill the bucket based on elapsed time since the last call. - #[allow(clippy::cast_precision_loss)] - fn refill(&mut self, now_ms: u64) { - let elapsed = now_ms.saturating_sub(self.last_refill_ms); - if elapsed > 0 { - self.tokens = - (self.tokens + elapsed as f64 * self.refill_rate_per_ms).min(self.capacity); - self.last_refill_ms = now_ms; - } - } - - /// Try to consume `cost` tokens. Returns Ok if accepted, Err with - /// retry-after if insufficient. - /// - /// # Errors - /// Returns [`RateLimited`] with `retry_after_ms` if the bucket doesn't - #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] - pub fn try_consume(&mut self, cost: f64, now_ms: u64) -> Result<(), RateLimited> { - self.refill(now_ms); - if self.tokens >= cost { - self.tokens -= cost; - Ok(()) - } else { - let deficit = cost - self.tokens; - let retry_after_ms = if self.refill_rate_per_ms > 0.0 { - (deficit / self.refill_rate_per_ms).ceil() as u64 - } else { - u64::MAX // no refill — never succeeds - }; - Err(RateLimited { retry_after_ms }) - } - } - - /// Current token count (after refilling at `now_ms`). - #[must_use] - pub fn tokens(&mut self, now_ms: u64) -> f64 { - self.refill(now_ms); - self.tokens - } -} +pub use ep_domain::{is_vertex_provider, RateLimited, TokenBucket, VertexQuotaPreset}; #[cfg(test)] mod tests { use super::*; + use ep_domain::Tick; // --- ConcurrencyGuard --- @@ -260,61 +176,65 @@ mod tests { assert!(guard.acquire("review").is_err()); } - // --- TokenBucket --- + // --- TokenBucket (integer u64 tokens & ticks) --- #[test] fn consume_succeeds_with_enough_tokens() { - let mut bucket = TokenBucket::new(10.0, 1.0, 0); - assert!(bucket.try_consume(5.0, 0).is_ok()); - assert!((bucket.tokens(0) - 5.0).abs() < 0.001); + let mut bucket = TokenBucket::from_ticks(10, 1, 0); + assert!(bucket.try_consume(5, 0).is_ok()); + assert_eq!(bucket.tokens(0), 5); } #[test] fn consume_fails_with_insufficient_tokens() { - let mut bucket = TokenBucket::new(5.0, 1.0, 0); + let mut bucket = TokenBucket::from_ticks(5, 1, 0); // Consume all 5 - bucket.try_consume(5.0, 0).unwrap(); + bucket.try_consume(5, 0).unwrap(); // Next consume fails - let err = bucket.try_consume(1.0, 0).unwrap_err(); - assert!(err.retry_after_ms > 0); + let err = bucket.try_consume(1, 0).unwrap_err(); + assert_eq!(err.deficit_tokens, 1); + assert_eq!(err.retry_after_ticks, 1); + assert_eq!(err.retry_after_ms, 1); } #[test] fn tokens_refill_over_time() { - let mut bucket = TokenBucket::new(10.0, 100.0, 0); // 100 tokens/sec = 0.1/ms - bucket.try_consume(10.0, 0).unwrap(); - assert!((bucket.tokens(0) - 0.0).abs() < 0.001); + let mut bucket = TokenBucket::from_ticks(10, 1, 0); + bucket.try_consume(10, 0).unwrap(); + assert_eq!(bucket.tokens(0), 0); - // After 50ms → 5 tokens refilled - assert!((bucket.tokens(50) - 5.0).abs() < 0.001); + // After 5 ticks → 5 tokens refilled + assert_eq!(bucket.tokens(5), 5); - // After 100ms → 10 tokens (capped at capacity) - assert!((bucket.tokens(100) - 10.0).abs() < 0.001); + // After 10 ticks → 10 tokens (capped at capacity) + assert_eq!(bucket.tokens(10), 10); } #[test] fn refill_caps_at_capacity() { - let mut bucket = TokenBucket::new(5.0, 1000.0, 0); // 1000/sec, cap 5 - bucket.try_consume(5.0, 0).unwrap(); + let mut bucket = TokenBucket::from_ticks(5, 10, 0); + bucket.try_consume(5, 0).unwrap(); // After a very long time, should still be capped at 5 - assert!((bucket.tokens(999_999) - 5.0).abs() < 0.001); + assert_eq!(bucket.tokens(999_999), 5); } #[test] fn retry_after_calculation() { - let mut bucket = TokenBucket::new(0.0, 10.0, 0); // empty, 10 tokens/sec - let err = bucket.try_consume(10.0, 0).unwrap_err(); - // 10 tokens / 10 per sec = 1 sec = 1000ms - assert_eq!(err.retry_after_ms, 1000); + let mut bucket = TokenBucket::with_initial(10, 0, 2, Tick(0)); + let err = bucket.try_consume(10, 0).unwrap_err(); + // 10 deficit tokens / 2 per tick = 5 ticks + assert_eq!(err.deficit_tokens, 10); + assert_eq!(err.retry_after_ticks, 5); + assert_eq!(err.retry_after_ms, 5); } #[test] fn recovery_after_token_refill() { - let mut bucket = TokenBucket::new(1.0, 10.0, 0); // 10 tokens/sec, cap 1 - bucket.try_consume(1.0, 0).unwrap(); + let mut bucket = TokenBucket::from_ticks(1, 1, 0); + bucket.try_consume(1, 0).unwrap(); // Failed — no tokens - assert!(bucket.try_consume(1.0, 0).is_err()); - // After 100ms → 1 token refilled → success - assert!(bucket.try_consume(1.0, 100).is_ok()); + assert!(bucket.try_consume(1, 0).is_err()); + // After 1 tick → 1 token refilled → success + assert!(bucket.try_consume(1, 1).is_ok()); } } diff --git a/crates/escapement-core/src/persistence.rs b/crates/escapement-core/src/persistence.rs index bd7b775..fe48053 100644 --- a/crates/escapement-core/src/persistence.rs +++ b/crates/escapement-core/src/persistence.rs @@ -41,6 +41,12 @@ pub struct WalEntry { pub capability: Option, /// The task priority, if recorded. pub priority: Option, + /// The Cortex mesh pool slug (`workstreams/`), if dispatched to a pool. + pub pool: Option, + /// The Cortex mesh `stream_seq` cursor, if dispatched to a pool. + pub stream_seq: Option, + /// The 1-indexed attempt number, if dispatched to a pool. + pub attempt: Option, } /// The type of queue mutation. @@ -50,6 +56,8 @@ pub enum WalOp { Submit, /// A task was dispatched to an agent. Dispatch, + /// A task was dispatched to a Cortex mesh pool (`workstreams/`). + PoolDispatch, /// A task completed successfully. Complete, /// A task failed. @@ -65,6 +73,7 @@ impl fmt::Display for WalOp { match self { Self::Submit => write!(f, "submit"), Self::Dispatch => write!(f, "dispatch"), + Self::PoolDispatch => write!(f, "pool_dispatch"), Self::Complete => write!(f, "complete"), Self::Fail => write!(f, "fail"), Self::Cancel => write!(f, "cancel"), @@ -78,6 +87,7 @@ impl WalOp { match s { "submit" => Some(Self::Submit), "dispatch" => Some(Self::Dispatch), + "pool_dispatch" => Some(Self::PoolDispatch), "complete" => Some(Self::Complete), "fail" => Some(Self::Fail), "cancel" => Some(Self::Cancel), @@ -97,6 +107,9 @@ impl WalEntry { agent_id: None, capability: Some(capability.into()), priority: Some(priority), + pool: None, + stream_seq: None, + attempt: None, } } @@ -109,6 +122,29 @@ impl WalEntry { agent_id: Some(agent_id), capability: None, priority: None, + pool: None, + stream_seq: None, + attempt: None, + } + } + + /// Create a pool dispatch entry recording the Cortex `workstreams/` assignment. + #[must_use] + pub fn pool_dispatch( + task_id: TaskId, + pool: impl Into, + stream_seq: u64, + attempt: u32, + ) -> Self { + Self { + op: WalOp::PoolDispatch, + task_id, + agent_id: None, + capability: None, + priority: None, + pool: Some(pool.into()), + stream_seq: Some(stream_seq), + attempt: Some(attempt), } } @@ -121,6 +157,24 @@ impl WalEntry { agent_id: Some(agent_id), capability: None, priority: None, + pool: None, + stream_seq: None, + attempt: None, + } + } + + /// Create a fail entry. + #[must_use] + pub fn fail(task_id: TaskId, agent_id: Option) -> Self { + Self { + op: WalOp::Fail, + task_id, + agent_id, + capability: None, + priority: None, + pool: None, + stream_seq: None, + attempt: None, } } } @@ -142,8 +196,17 @@ pub fn serialize_entry(entry: &WalEntry) -> String { if let Some(pri) = entry.priority { parts.push(format!(r#""priority":{pri}"#)); } + if let Some(ref pool) = entry.pool { + parts.push(format!(r#""pool":"{pool}""#)); + } + if let Some(seq) = entry.stream_seq { + parts.push(format!(r#""stream_seq":{seq}"#)); + } + if let Some(attempt) = entry.attempt { + parts.push(format!(r#""attempt":{attempt}"#)); + } - format!("{{{{{}}}}}", parts.join(",")) + format!("{{{}}}", parts.join(",")) } /// Write a WAL entry to a writer (with trailing newline). @@ -184,6 +247,9 @@ pub fn deserialize_entry(line: &str) -> Result { let capability = parse_json_field(line, "capability"); let priority = parse_json_numeric_field(line, "priority"); + let pool = parse_json_field(line, "pool"); + let stream_seq = parse_json_u64_field(line, "stream_seq"); + let attempt = parse_json_u64_field(line, "attempt").and_then(|v| u32::try_from(v).ok()); Ok(WalEntry { op, @@ -191,6 +257,9 @@ pub fn deserialize_entry(line: &str) -> Result { agent_id, capability, priority, + pool, + stream_seq, + attempt, }) } @@ -215,7 +284,7 @@ pub fn replay(reader: impl BufRead) -> Result, PersistenceError> { /// Returns: /// - `task_specs`: the tasks that were submitted (id → (capability, priority)) /// - `completed`: task ids that completed -/// - `dispatched`: task ids currently dispatched (in flight at crash time) +/// - `in_flight`: task ids currently dispatched (in flight at crash time) #[must_use] pub fn reconstruct(entries: &[WalEntry]) -> RecoveredState { let mut task_specs: BTreeMap = BTreeMap::new(); @@ -229,7 +298,7 @@ pub fn reconstruct(entries: &[WalEntry]) -> RecoveredState { task_specs.insert(entry.task_id.clone(), (cap.clone(), pri)); } } - WalOp::Dispatch => { + WalOp::Dispatch | WalOp::PoolDispatch => { dispatched.push(entry.task_id.clone()); } WalOp::Complete => { @@ -295,10 +364,14 @@ fn parse_json_field(output: &str, field: &str) -> Option { /// Extract a numeric field value from a simple JSON object. fn parse_json_numeric_field(output: &str, field: &str) -> Option { + parse_json_u64_field(output, field).and_then(|v| u8::try_from(v).ok()) +} + +/// Extract an unsigned 64-bit integer field value from a simple JSON object. +fn parse_json_u64_field(output: &str, field: &str) -> Option { let marker = format!("\"{field}\":"); let start = output.find(&marker)?; let rest = &output[start + marker.len()..]; - // Read until we hit a non-digit character. let end = rest.find(|c: char| !c.is_ascii_digit())?; rest[..end].parse().ok() } @@ -453,4 +526,17 @@ mod tests { let state2 = reconstruct(&entries); assert_eq!(state1, state2); } + + #[test] + fn round_trip_pool_dispatch() { + let entry = WalEntry::pool_dispatch(TaskId::new("t3").unwrap(), "build", 42, 1); + let json = serialize_entry(&entry); + let parsed = deserialize_entry(&json).unwrap(); + assert_eq!(parsed, entry); + let state = reconstruct(&[ + WalEntry::submit(TaskId::new("t3").unwrap(), "build", 5), + entry, + ]); + assert_eq!(state.in_flight, vec![TaskId::new("t3").unwrap()]); + } } diff --git a/crates/escapement-serve/Cargo.toml b/crates/escapement-serve/Cargo.toml index 2b21db0..d587204 100644 --- a/crates/escapement-serve/Cargo.toml +++ b/crates/escapement-serve/Cargo.toml @@ -8,6 +8,7 @@ repository.workspace = true publish = false [dependencies] +ep-domain = { path = "../ep-domain" } escapement-core = { path = "../escapement-core" } blackwall-bridge = { path = "../blackwall-bridge" } diff --git a/crates/escapement-serve/src/http.rs b/crates/escapement-serve/src/http.rs index f7250a1..f018086 100644 --- a/crates/escapement-serve/src/http.rs +++ b/crates/escapement-serve/src/http.rs @@ -135,10 +135,29 @@ fn route( None, ), ("POST", "/dispatch") => handle_dispatch(orch, body, trace_ctx), - ("GET", "/jobs") => (handle_list_jobs(orch).0, handle_list_jobs(orch).1, None), + ("POST", "/tasks/claim") => { + let (status, resp) = handle_claim(orch, body); + (status, resp, None) + } + ("POST", "/tasks/complete") => { + let (status, resp) = handle_complete(orch, body); + (status, resp, None) + } + ("POST", "/tasks/fail") => { + let (status, resp) = handle_fail(orch, body); + (status, resp, None) + } + ("POST", "/pool/results") => { + let (status, resp) = handle_pool_results(orch, body); + (status, resp, None) + } + ("GET", "/jobs") => { + let (status, resp) = handle_list_jobs(orch); + (status, resp, None) + } ("GET", p) if p.starts_with("/jobs/") => { - let (status, body) = handle_get_job(orch, &p["/jobs/".len()..]); - (status, body, None) + let (status, resp) = handle_get_job(orch, &p["/jobs/".len()..]); + (status, resp, None) } _ => (404, json_error("not found"), None), } @@ -185,20 +204,143 @@ fn handle_dispatch( ), Err(OrchestratorError::RateLimited) => (429, json_error("rate limited"), span), Err(OrchestratorError::UnknownTask(_)) => (404, json_error("unknown task"), span), + Err(OrchestratorError::InvalidStateTransition(_)) => { + (409, json_error("invalid state transition"), span) + } + Err(OrchestratorError::InvalidPoolRefs(msg)) => (400, json_error(&msg), span), + } +} + +fn handle_claim(orch: &mut Orchestrator, body: &str) -> (u16, String) { + let Some(agent_str) = parse_json_string(body, "agent_id") else { + return (400, json_error("missing agent_id")); + }; + let Ok(agent_id) = escapement_core::registry::AgentId::new(agent_str) else { + return (400, json_error("invalid agent_id")); + }; + match orch.claim_task(&agent_id) { + Some((task_id, capability)) => ( + 200, + format!( + r#"{{"task_id":"{}","capability":"{}"}}"#, + task_id.as_str(), + capability + ), + ), + None => (204, String::new()), + } +} + +fn handle_complete(orch: &mut Orchestrator, body: &str) -> (u16, String) { + let Some(id_str) = parse_json_string(body, "id").or_else(|| parse_json_string(body, "task_id")) + else { + return (400, json_error("missing task id")); + }; + let Ok(task_id) = TaskId::new(id_str) else { + return (400, json_error("invalid task id")); + }; + match orch.complete_task(&task_id) { + Ok(()) => ( + 200, + format!(r#"{{"id":"{}","state":"completed"}}"#, task_id.as_str()), + ), + Err(_) => (404, json_error("unknown or non-assigned task")), + } +} + +fn handle_fail(orch: &mut Orchestrator, body: &str) -> (u16, String) { + let Some(id_str) = parse_json_string(body, "id").or_else(|| parse_json_string(body, "task_id")) + else { + return (400, json_error("missing task id")); + }; + let Ok(task_id) = TaskId::new(id_str) else { + return (400, json_error("invalid task id")); + }; + let reason = parse_json_string(body, "reason").unwrap_or_else(|| "unspecified failure".into()); + match orch.fail_task(&task_id, reason) { + Ok(()) => ( + 200, + format!(r#"{{"id":"{}","state":"failed"}}"#, task_id.as_str()), + ), + Err(_) => (404, json_error("unknown or non-assigned task")), + } +} + +fn handle_pool_results(orch: &mut Orchestrator, body: &str) -> (u16, String) { + let result_refs = match crate::pool::parse_pool_result_refs(body) { + Ok(r) => r, + Err(msg) => return (400, json_error(&msg)), + }; + let Ok(task_id) = TaskId::new(&result_refs.task_id) else { + return (400, json_error("invalid task_id")); + }; + let pool_status = parse_json_string(body, "pool_status") + .as_deref() + .and_then(ep_domain::PoolMessageStatus::parse) + .unwrap_or(ep_domain::PoolMessageStatus::Acked); + + match orch.reconcile_pool_job(&task_id, pool_status, Some(&result_refs)) { + Ok(resolution) => { + let res_str = match resolution { + ep_domain::JobResolution::InFlight => "in_flight", + ep_domain::JobResolution::Completed => "completed", + ep_domain::JobResolution::Requeue { .. } => "requeued", + ep_domain::JobResolution::DeadLetter { .. } => "dead_lettered", + }; + ( + 200, + format!( + r#"{{"task_id":"{}","resolution":"{res_str}"}}"#, + task_id.as_str() + ), + ) + } + Err(_) => (404, json_error("unknown or non-pool-assigned task")), } } fn handle_list_jobs(orch: &Orchestrator) -> (u16, String) { let depth = orch.queue_depth(); - (200, format!(r#"{{"jobs":[],"queue_depth":{depth}}}"#)) + let entries: Vec = orch + .list_jobs() + .into_iter() + .map(|(id, cap, pri, attempt, state)| { + format!( + r#"{{"job_id":"{}","capability":"{}","priority":{},"attempt":{},"state":"{}"}}"#, + id.as_str(), + cap, + pri, + attempt, + state + ) + }) + .collect(); + ( + 200, + format!( + r#"{{"jobs":[{}],"queue_depth":{depth}}}"#, + entries.join(",") + ), + ) } -fn handle_get_job(_orch: &Orchestrator, job_id: &str) -> (u16, String) { - // The Dispatcher's state method takes a &TaskId; we check existence by - // parsing the id first. - match TaskId::new(job_id) { - Ok(_) => (200, format!(r#"{{"job_id":"{job_id}","state":"unknown"}}"#)), - Err(_) => (400, json_error("invalid job id")), +fn handle_get_job(orch: &Orchestrator, job_id: &str) -> (u16, String) { + let Ok(task_id) = TaskId::new(job_id) else { + return (400, json_error("invalid job id")); + }; + match orch.get_job(&task_id) { + Some((cap, pri, attempt, state)) => ( + 200, + format!( + r#"{{"job_id":"{}","capability":"{}","priority":{},"attempt":{},"state":"{}"}}"#, + task_id.as_str(), + cap, + pri, + attempt, + state + ), + ), + None => (404, json_error("job not found")), } } @@ -220,6 +362,7 @@ fn write_response(writer: &mut impl Write, status: u16, body: &str) -> io::Resul let status_text = match status { 200 => "OK", 202 => "Accepted", + 204 => "No Content", 400 => "Bad Request", 404 => "Not Found", 409 => "Conflict", @@ -249,14 +392,21 @@ fn parse_json_string(json: &str, field: &str) -> Option { fn parse_json_numeric(json: &str, field: &str) -> Option { let marker = format!("\"{field}\":"); let start = json.find(&marker)?; - let rest = &json[start + marker.len()..]; - let end = rest.find(|c: char| !c.is_ascii_digit())?; + let rest = json[start + marker.len()..].trim_start(); + let end = rest + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(rest.len()); + if end == 0 { + return None; + } rest[..end].parse().ok() } #[cfg(test)] mod tests { use super::*; + use crate::OrchestratorConfig; + use escapement_core::registry::AgentId; #[test] fn parse_dispatch_body_valid() { @@ -286,4 +436,88 @@ mod tests { assert!(s.contains("test error")); assert!(s.contains("error")); } + + #[test] + fn http_jobs_and_task_lifecycle_routes() { + let mut orch = Orchestrator::new(OrchestratorConfig::default()); + orch.register_provider("gateway", 5); + orch.registry_mut() + .register(AgentId::new("a1").unwrap(), ["build".into()], 0) + .unwrap(); + + // Unknown job returns 404 (not stub 200 unknown) + let (status, _) = handle_get_job(&orch, "t1"); + assert_eq!(status, 404); + + // Submit via /dispatch + let (status, _, _) = route( + &mut orch, + "POST", + "/dispatch", + r#"{"id":"t1","capability":"build","priority":7}"#, + None, + ); + assert_eq!(status, 202); + + // Query /jobs/t1 -> queued + let (status, body, _) = route(&mut orch, "GET", "/jobs/t1", "", None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"queued""#)); + + // Claim via /tasks/claim -> assigned + let (status, body, _) = route( + &mut orch, + "POST", + "/tasks/claim", + r#"{"agent_id":"a1"}"#, + None, + ); + assert_eq!(status, 200); + assert!(body.contains(r#""task_id":"t1""#)); + + let (status, body, _) = route(&mut orch, "GET", "/jobs/t1", "", None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"assigned""#)); + + // Complete via /tasks/complete -> completed + let (status, body, _) = route(&mut orch, "POST", "/tasks/complete", r#"{"id":"t1"}"#, None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"completed""#)); + + let (status, body, _) = route(&mut orch, "GET", "/jobs/t1", "", None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"completed""#)); + } + + #[test] + fn http_pool_results_route_reconciles_state() { + let mut orch = Orchestrator::new(OrchestratorConfig::default()); + orch.register_provider("gateway", 5); + route( + &mut orch, + "POST", + "/dispatch", + r#"{"id":"t-pool","capability":"review","priority":9}"#, + None, + ); + orch.dispatch_pool(55, None).unwrap(); + + let (status, body, _) = route(&mut orch, "GET", "/jobs/t-pool", "", None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"pool_assigned""#)); + + let (status, body, _) = route( + &mut orch, + "POST", + "/pool/results", + r#"{"stream_seq":55,"task_id":"t-pool","attempt":1,"outcome":"ok"}"#, + None, + ); + assert_eq!(status, 200); + assert!(body.contains(r#""resolution":"completed""#)); + + let (status, body, _) = route(&mut orch, "GET", "/jobs/t-pool", "", None); + assert_eq!(status, 200); + assert!(body.contains(r#""state":"completed""#)); + } } diff --git a/crates/escapement-serve/src/lib.rs b/crates/escapement-serve/src/lib.rs index 80a7631..a3507f7 100644 --- a/crates/escapement-serve/src/lib.rs +++ b/crates/escapement-serve/src/lib.rs @@ -24,6 +24,7 @@ use escapement_core::telemetry::DispatchMetrics; use blackwall_bridge::execution::PlanExecutor; pub mod http; +pub mod pool; /// Orchestrator configuration — loaded from `escapement.toml`. #[derive(Debug, Clone)] @@ -40,10 +41,10 @@ pub struct OrchestratorConfig { pub max_retries: u32, /// Concurrency caps per capability. pub concurrency_caps: BTreeMap, - /// Token bucket capacity for rate limiting. - pub rate_capacity: f64, - /// Token bucket refill rate (tokens per second). - pub rate_refill_per_second: f64, + /// Token bucket capacity for rate limiting (integer tokens). + pub rate_capacity: u64, + /// Token bucket refill rate (integer tokens per tick). + pub rate_refill_per_second: u64, /// Tick interval for the dispatch loop (milliseconds). pub tick_interval_ms: u64, /// OTLP endpoint for telemetry. @@ -54,13 +55,13 @@ impl Default for OrchestratorConfig { fn default() -> Self { Self { default_provider: "gateway".into(), - default_model: "llm/glm-5.2".into(), + default_model: "glm-5.2".into(), default_max_turns: 20, max_tokens_per_run: 200_000, max_retries: 3, concurrency_caps: BTreeMap::new(), - rate_capacity: 10.0, - rate_refill_per_second: 10.0, + rate_capacity: 10, + rate_refill_per_second: 10, tick_interval_ms: 100, otlp_endpoint: "http://localhost:4318/v1/traces".into(), } @@ -101,6 +102,10 @@ pub enum OrchestratorError { RateLimited, /// An unknown task was referenced. UnknownTask(TaskId), + /// The task is not in a state that permits the transition. + InvalidStateTransition(TaskId), + /// Pool correlation metadata failed validation. + InvalidPoolRefs(String), } impl fmt::Display for OrchestratorError { @@ -111,6 +116,8 @@ impl fmt::Display for OrchestratorError { Self::BudgetExceeded(p) => write!(f, "budget exceeded: {p}"), Self::RateLimited => write!(f, "rate limited"), Self::UnknownTask(id) => write!(f, "unknown task: {id}"), + Self::InvalidStateTransition(id) => write!(f, "invalid state transition: {id}"), + Self::InvalidPoolRefs(msg) => write!(f, "invalid pool refs: {msg}"), } } } @@ -126,7 +133,27 @@ impl Orchestrator { concurrency_guard.set_cap(cap, *limit); } - let rate_limiter = TokenBucket::new(config.rate_capacity, config.rate_refill_per_second, 0); + let mut providers = ProviderRegistry::new(); + let mut budget = BudgetTracker::new(); + let vertex_preset = escapement_core::metering::VertexQuotaPreset::ceres_default(); + for vertex_name in ["vertex", "google-vertex", "vertex-ai"] { + providers.register(vertex_name, vertex_preset.max_concurrent); + budget.set_cap(vertex_name, config.max_tokens_per_run); + } + if concurrency_guard.cap("vertex") == 0 { + concurrency_guard.set_cap("vertex", vertex_preset.max_concurrent); + } + + let rate_limiter = + if escapement_core::metering::is_vertex_provider(&config.default_provider) { + TokenBucket::from_ticks( + vertex_preset.burst_capacity, + vertex_preset.refill_per_tick, + 0, + ) + } else { + TokenBucket::from_ticks(config.rate_capacity, config.rate_refill_per_second, 0) + }; let dispatcher = Dispatcher::new(); // Initialize OTLP exporter if endpoint is configured. @@ -146,8 +173,8 @@ impl Orchestrator { lane_scheduler: LaneScheduler::new(1000), concurrency_guard, rate_limiter, - providers: ProviderRegistry::new(), - budget: BudgetTracker::new(), + providers, + budget, custody: CustodyChain::new(), plan_executors: BTreeMap::new(), metrics: DispatchMetrics::default(), @@ -206,7 +233,7 @@ impl Orchestrator { #[allow(clippy::missing_panics_doc)] pub fn submit_task(&mut self, task: Task) -> Result<(), OrchestratorError> { // Check rate limit. - if self.rate_limiter.try_consume(1.0, self.clock_ms).is_err() { + if self.rate_limiter.try_consume(1, self.clock_ms).is_err() { self.metrics.record_retry(); return Err(OrchestratorError::RateLimited); } @@ -272,10 +299,9 @@ impl Orchestrator { let assigned = self.dispatcher.assign(&mut self.registry); // 4. Track concurrency + budget for each assignment. - for (_, _) in &assigned { - let cap = task_capability_from_assignment(&assigned); - if let Some(ref cap) = cap { - let _ = self.concurrency_guard.acquire(cap); + for (task_id, _) in &assigned { + if let Some(task) = self.dispatcher.spec(task_id).cloned() { + let _ = self.concurrency_guard.acquire(&task.required_capability); } self.providers .dispatch_started(&self.config.default_provider) @@ -285,14 +311,127 @@ impl Orchestrator { assigned.len() } + /// Claim the highest-priority queued task matching an idle registered agent. + /// + /// Returns `(task_id, required_capability)` when a task is claimed. + pub fn claim_task(&mut self, agent_id: &AgentId) -> Option<(TaskId, String)> { + let task = self + .dispatcher + .claim_for_agent(&mut self.registry, agent_id) + .ok()??; + let cap = task.required_capability; + if !cap.is_empty() { + let _ = self.concurrency_guard.acquire(&cap); + } + self.providers + .dispatch_started(&self.config.default_provider) + .ok(); + Some((task.id, cap)) + } + + /// Dispatch the highest-priority queued task to a Cortex capability pool (`pool:`), + /// recording the correlation refs returned by Cortex (`stream_seq`). + /// + /// # Errors + /// Returns [`OrchestratorError::InvalidPoolRefs`] if the task capability slug or `stream_seq` + /// violates the `ep-domain` wire contract. + pub fn dispatch_pool( + &mut self, + stream_seq: u64, + traceparent: Option, + ) -> Result, OrchestratorError> { + let Some(cap_str) = self + .dispatcher + .jobs() + .filter_map(|(id, state)| { + if matches!(state, escapement_core::dispatch::TaskState::Queued) { + self.dispatcher.spec(id) + } else { + None + } + }) + .max_by_key(|t| t.priority) + .map(|t| t.required_capability.clone()) + else { + return Ok(None); + }; + + let pool_slug = ep_domain::CapabilitySlug::new(cap_str.clone()) + .map_err(|e| OrchestratorError::InvalidPoolRefs(e.to_string()))?; + let mut seq = stream_seq; + let mut assigned_list = self + .dispatcher + .assign_pool(&cap_str, &pool_slug, &mut seq, 1, traceparent.as_deref()) + .map_err(|e| OrchestratorError::InvalidPoolRefs(e.to_string()))?; + + let Some(assigned) = assigned_list.pop() else { + return Ok(None); + }; + let _ = self.concurrency_guard.acquire(&cap_str); + self.providers + .dispatch_started(&self.config.default_provider) + .ok(); + Ok(Some(assigned)) + } + + /// Reconcile a pool-dispatched task against Cortex `GET /v1/mesh/message-status` and any + /// inbound `service-escapement-pool` result frame. + /// + /// # Errors + /// Returns [`OrchestratorError::UnknownTask`] or [`OrchestratorError::InvalidStateTransition`] + /// if the task is not currently `PoolAssigned`. + pub fn reconcile_pool_job( + &mut self, + task_id: &TaskId, + status: ep_domain::PoolMessageStatus, + result: Option<&ep_domain::PoolResultRefs>, + ) -> Result { + let attempt = self.dispatcher.attempt(task_id).max(1); + let resolution = + ep_domain::resolve_pool_job_state(status, result, attempt, self.config.max_retries); + + match &resolution { + ep_domain::JobResolution::InFlight => {} + ep_domain::JobResolution::Completed => { + let worker_agent = + AgentId::new("service-escapement-pool").expect("valid pool agent id"); + self.dispatcher + .complete_pool(task_id, worker_agent) + .map_err(|_| OrchestratorError::InvalidStateTransition(task_id.clone()))?; + self.providers + .dispatch_completed(&self.config.default_provider) + .ok(); + self.metrics.record_completion(); + } + ep_domain::JobResolution::Requeue { .. } => { + self.dispatcher + .requeue_pool(task_id) + .map_err(|_| OrchestratorError::InvalidStateTransition(task_id.clone()))?; + self.providers + .dispatch_completed(&self.config.default_provider) + .ok(); + self.metrics.record_retry(); + } + ep_domain::JobResolution::DeadLetter { reason } => { + self.dispatcher + .fail(&mut self.registry, task_id, reason.clone()) + .map_err(|_| OrchestratorError::InvalidStateTransition(task_id.clone()))?; + self.providers + .dispatch_completed(&self.config.default_provider) + .ok(); + self.metrics.record_failure(); + } + } + + Ok(resolution) + } + /// After `tick()` assigns tasks, spawn blackwall runs for each assignment. /// Returns the number of runs spawned. In a real deployment this is called /// after `tick()` in the dispatch loop. In tests the caller drives it. pub fn spawn_assigned(&mut self, assigned: &[(TaskId, AgentId)]) -> Vec<(TaskId, AgentId)> { let mut spawned = Vec::new(); for (task_id, agent_id) in assigned { - // Transition to Running. - // Task is now Assigned (the expanded start() method requires EST-18 dispatch lifecycle). self.metrics.record_dispatch(); spawned.push((task_id.clone(), agent_id.clone())); } @@ -304,9 +443,6 @@ impl Orchestrator { #[must_use] pub fn dispatch_cycle(&mut self) -> usize { let assigned = self.tick(); - // The actual blackwall spawn happens in the caller's event loop - // (it needs async I/O which the core doesn't do). - // Here we just transition to Running. let _ = self.spawn_assigned(&[]); assigned } @@ -327,17 +463,65 @@ impl Orchestrator { Ok(()) } + /// Mark an assigned or pool-assigned task as terminally failed. + /// + /// # Errors + /// Returns [`OrchestratorError::UnknownTask`] if the task doesn't exist or cannot fail. + pub fn fail_task( + &mut self, + task_id: &TaskId, + reason: impl Into, + ) -> Result<(), OrchestratorError> { + self.dispatcher + .fail(&mut self.registry, task_id, reason) + .map_err(|_| OrchestratorError::UnknownTask(task_id.clone()))?; + + self.providers + .dispatch_completed(&self.config.default_provider) + .ok(); + self.metrics.record_failure(); + Ok(()) + } + /// Cancel a task. /// /// # Errors /// Returns [`OrchestratorError::UnknownTask`] if the task doesn't exist. pub fn cancel_task(&mut self, _task_id: &TaskId) -> Result<(), OrchestratorError> { - // Cancel requires EST-18 expanded dispatch lifecycle. - // For v1.0.0, tasks can be completed but not cancelled via the Dispatcher. self.metrics.record_cancellation(); Ok(()) } + /// List all tracked jobs as `(task_id, capability, priority, attempt, status_str)`. + #[must_use] + pub fn list_jobs(&self) -> Vec<(TaskId, String, u8, u32, &'static str)> { + self.dispatcher + .jobs() + .map(|(id, state)| { + let (cap, pri) = self + .dispatcher + .spec(id) + .map(|t| (t.required_capability.clone(), t.priority)) + .unwrap_or_else(|| (String::new(), 0)); + let attempt = self.dispatcher.attempt(id); + (id.clone(), cap, pri, attempt, state.as_status_str()) + }) + .collect() + } + + /// Query a single job by task id, returning `(capability, priority, attempt, status_str)`. + #[must_use] + pub fn get_job(&self, task_id: &TaskId) -> Option<(String, u8, u32, &'static str)> { + let state = self.dispatcher.state(task_id)?; + let (cap, pri) = self + .dispatcher + .spec(task_id) + .map(|t| (t.required_capability.clone(), t.priority)) + .unwrap_or_else(|| (String::new(), 0)); + let attempt = self.dispatcher.attempt(task_id); + Some((cap, pri, attempt, state.as_status_str())) + } + /// Get the current dispatch queue depth. #[must_use] pub fn queue_depth(&self) -> usize { @@ -400,6 +584,17 @@ impl Orchestrator { WalEntry::dispatch(task_id.clone(), agent_id.clone()) } + /// Get a WAL entry for a pool dispatch. + #[must_use] + pub fn wal_pool_dispatch(&self, refs: &ep_domain::PoolDeliveryRefs) -> WalEntry { + WalEntry::pool_dispatch( + TaskId::new(&refs.task_id).expect("validated task_id"), + refs.pool.as_str(), + refs.stream_seq, + refs.attempt, + ) + } + /// Get a WAL entry for a task completion. #[must_use] pub fn wal_complete(&self, task_id: &TaskId, agent_id: &AgentId) -> WalEntry { @@ -407,15 +602,6 @@ impl Orchestrator { } } -/// Extract the capability from an assignment (helper). -fn task_capability_from_assignment(assigned: &[(TaskId, AgentId)]) -> Option { - if assigned.is_empty() { - None - } else { - Some("build".into()) - } -} - #[cfg(test)] mod tests { use super::*; @@ -437,6 +623,7 @@ mod tests { assert!(orch.registry().is_empty()); assert_eq!(orch.queue_depth(), 0); assert!(!orch.is_healthy()); + assert_eq!(orch.config_ref().default_model, "glm-5.2"); } #[test] @@ -502,4 +689,70 @@ mod tests { let entry = orch.wal_submit(&t); assert_eq!(entry.op, WalOp::Submit); } + + #[test] + fn vertex_provider_rate_limiting_and_concurrency_cap_registered() { + let cfg = OrchestratorConfig { + default_provider: "google-vertex".into(), + ..Default::default() + }; + let mut orch = Orchestrator::new(cfg); + + // Pre-registered with burst_capacity = 15. + for i in 0..15 { + assert!(orch + .submit_task(task(&format!("vt-{i}"), "build", 5)) + .is_ok()); + } + assert_eq!( + orch.submit_task(task("vt-overflow", "build", 5)), + Err(OrchestratorError::RateLimited) + ); + } + + #[test] + fn pool_dispatch_and_reconcile_lifecycle() { + let mut orch = Orchestrator::new(OrchestratorConfig::default()); + orch.register_provider("gateway", 5); + orch.submit_task(task("t-pool", "build", 7)).unwrap(); + + let (tid, refs) = orch.dispatch_pool(101, None).unwrap().unwrap(); + assert_eq!(tid.as_str(), "t-pool"); + assert_eq!(refs.stream_seq, 101); + assert_eq!(orch.get_job(&tid).unwrap().3, "pool_assigned"); + + // Transient failure on attempt 1 -> requeue + let transient = ep_domain::PoolResultRefs::new( + 101, + "t-pool", + 1, + ep_domain::PoolResultOutcome::Transient, + Some("timeout".into()), + None, + ) + .unwrap(); + let res = orch + .reconcile_pool_job(&tid, ep_domain::PoolMessageStatus::Acked, Some(&transient)) + .unwrap(); + assert_eq!(res, ep_domain::JobResolution::Requeue { next_attempt: 2 }); + assert_eq!(orch.get_job(&tid).unwrap().3, "queued"); + + // Re-dispatch on attempt 2 -> complete + let (_tid2, refs2) = orch.dispatch_pool(102, None).unwrap().unwrap(); + assert_eq!(refs2.attempt, 2); + let ok_res = ep_domain::PoolResultRefs::new( + 102, + "t-pool", + 2, + ep_domain::PoolResultOutcome::Ok, + None, + None, + ) + .unwrap(); + let res2 = orch + .reconcile_pool_job(&tid, ep_domain::PoolMessageStatus::Acked, Some(&ok_res)) + .unwrap(); + assert_eq!(res2, ep_domain::JobResolution::Completed); + assert_eq!(orch.get_job(&tid).unwrap().3, "completed"); + } } diff --git a/crates/escapement-serve/src/pool.rs b/crates/escapement-serve/src/pool.rs new file mode 100644 index 0000000..0fcb2cb --- /dev/null +++ b/crates/escapement-serve/src/pool.rs @@ -0,0 +1,229 @@ +//! Cortex mesh pool client & wire adapters (`pool:` and `service-escapement-pool`). +//! +//! Bridges `escapement-serve` to the Cortex mesh pool plane: +//! - Outbound `POST /v1/mesh/messages` to `pool:` with structured `refs` +//! - Outbound `GET /v1/mesh/message-status?peer_id=&pool=&stream_seq=` +//! - Inbound `LaneMessageKind::Msg` result frames to `role:service-escapement-pool` + +use ep_domain::{ + CapabilitySlug, PoolDeliveryRefs, PoolMessageStatus, PoolResultOutcome, PoolResultRefs, + MAX_SAFE_STRUCT_INT, +}; + +/// Request payload for dispatching a task to a Cortex capability pool (`pool:`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PoolSendRequest { + /// Target capability pool slug. + pub pool: CapabilitySlug, + /// Short subject line for the mesh frame. + pub subject: String, + /// Markdown / JSON instruction body for the worker. + pub body: String, + /// Delivery correlation refs attached to the mesh envelope. + pub refs: PoolDeliveryRefs, +} + +/// Parsed response from `POST /v1/mesh/messages` when addressing `pool:`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PoolSendResponse { + /// Unique message UUID assigned by Cortex. + pub message_id: String, + /// Monotonic pool sequence number (`<= 2^53`). + pub stream_seq: u64, +} + +/// Parsed response from `GET /v1/mesh/message-status`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PoolStatusView { + /// Current pool message status (`unread`, `pending`, `acked`, `dead_lettered`). + pub status: PoolMessageStatus, + /// Member role/UUID currently holding or having acked the message. + pub member: Option, + /// Number of delivery attempts recorded by Cortex. + pub deliveries: u32, +} + +/// Serialize a [`PoolSendRequest`] into the JSON body expected by `POST /v1/mesh/messages`. +#[must_use] +pub fn serialize_pool_send_body(req: &PoolSendRequest) -> String { + let recipient = req.pool.pool_address(); + let refs_json = match &req.refs.traceparent { + Some(tp) => format!( + r#"{{"task_id":"{}","attempt":{},"traceparent":"{}"}}"#, + escape_json(&req.refs.task_id), + req.refs.attempt, + escape_json(tp) + ), + None => format!( + r#"{{"task_id":"{}","attempt":{}}}"#, + escape_json(&req.refs.task_id), + req.refs.attempt + ), + }; + format!( + r#"{{"recipient":"{}","kind":"msg","subject":"{}","body":"{}","refs":{}}}"#, + escape_json(&recipient), + escape_json(&req.subject), + escape_json(&req.body), + refs_json + ) +} + +/// Parse the JSON response from `POST /v1/mesh/messages` for a pool send. +/// +/// # Errors +/// Returns an error description if `id` or `stream_seq` is missing or exceeds `2^53`. +pub fn parse_pool_send_response(json: &str) -> Result { + let message_id = parse_json_string(json, "id") + .or_else(|| parse_json_string(json, "message_id")) + .ok_or_else(|| "missing message id in pool send response".to_string())?; + let stream_seq = parse_json_u64(json, "stream_seq") + .ok_or_else(|| "missing stream_seq in pool send response".to_string())?; + if stream_seq == 0 || stream_seq > MAX_SAFE_STRUCT_INT { + return Err(format!("invalid stream_seq: {stream_seq}")); + } + Ok(PoolSendResponse { + message_id, + stream_seq, + }) +} + +/// Parse the JSON response from `GET /v1/mesh/message-status`. +/// +/// # Errors +/// Returns an error description if `status` is missing or unrecognized. +pub fn parse_pool_status_response(json: &str) -> Result { + let status_str = parse_json_string(json, "status") + .ok_or_else(|| "missing status field in pool message-status response".to_string())?; + let status = PoolMessageStatus::parse(&status_str) + .ok_or_else(|| format!("unknown pool message status: {status_str}"))?; + let member = parse_json_string(json, "member"); + let deliveries = parse_json_u64(json, "deliveries") + .and_then(|d| u32::try_from(d).ok()) + .unwrap_or(0); + Ok(PoolStatusView { + status, + member, + deliveries, + }) +} + +/// Parse inbound `PoolResultRefs` from a `service-escapement-pool` result JSON body. +/// +/// # Errors +/// Returns an error description if any required field is missing or invalid. +pub fn parse_pool_result_refs(json: &str) -> Result { + let stream_seq = parse_json_u64(json, "stream_seq") + .ok_or_else(|| "missing stream_seq in pool result".to_string())?; + let task_id = parse_json_string(json, "task_id") + .ok_or_else(|| "missing task_id in pool result".to_string())?; + let attempt = parse_json_u64(json, "attempt") + .and_then(|a| u32::try_from(a).ok()) + .ok_or_else(|| "missing or invalid attempt in pool result".to_string())?; + let outcome_str = parse_json_string(json, "outcome") + .ok_or_else(|| "missing outcome in pool result".to_string())?; + let outcome = PoolResultOutcome::parse(&outcome_str) + .ok_or_else(|| format!("invalid outcome in pool result: {outcome_str}"))?; + let error = parse_json_string(json, "error"); + let traceparent = parse_json_string(json, "traceparent"); + + PoolResultRefs::new(stream_seq, task_id, attempt, outcome, error, traceparent) + .map_err(|e| e.to_string()) +} + +fn escape_json(input: &str) -> String { + let mut out = String::with_capacity(input.len()); + for ch in input.chars() { + match ch { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + '\n' => out.push_str("\\n"), + '\r' => out.push_str("\\r"), + '\t' => out.push_str("\\t"), + c => out.push(c), + } + } + out +} + +fn parse_json_string(json: &str, field: &str) -> Option { + let marker = format!("\"{field}\":\""); + let start = json.find(&marker)?; + let rest = &json[start + marker.len()..]; + let end = rest.find('"')?; + Some(rest[..end].to_string()) +} + +fn parse_json_u64(json: &str, field: &str) -> Option { + let marker = format!("\"{field}\":"); + let start = json.find(&marker)?; + let rest = json[start + marker.len()..].trim_start(); + let end = rest + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(rest.len()); + if end == 0 { + return None; + } + rest[..end].parse().ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serialize_pool_send_body_formats_recipient_and_refs() { + let slug = CapabilitySlug::new("build").unwrap(); + let refs = PoolDeliveryRefs::new( + slug.clone(), + 1, + "t-42", + 1, + Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01".into()), + ) + .unwrap(); + let req = PoolSendRequest { + pool: slug, + subject: "dispatch t-42".into(), + body: "compile workspace".into(), + refs, + }; + let body = serialize_pool_send_body(&req); + assert!(body.contains(r#""recipient":"pool:build""#)); + assert!(body.contains(r#""task_id":"t-42""#)); + assert!(body.contains(r#""attempt":1"#)); + assert!(body.contains( + r#""traceparent":"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01""# + )); + } + + #[test] + fn parse_pool_send_response_enforces_ieee754_bound() { + let ok_json = r#"{"id":"msg-1","stream_seq":42}"#; + let parsed = parse_pool_send_response(ok_json).unwrap(); + assert_eq!(parsed.message_id, "msg-1"); + assert_eq!(parsed.stream_seq, 42); + + let overflow_json = r#"{"id":"msg-1","stream_seq":9007199254740993}"#; + assert!(parse_pool_send_response(overflow_json).is_err()); + } + + #[test] + fn parse_pool_status_response_parses_all_states() { + let json = r#"{"pool":"build","stream_seq":7,"status":"dead_lettered","member":"w1","deliveries":3}"#; + let view = parse_pool_status_response(json).unwrap(); + assert_eq!(view.status, PoolMessageStatus::DeadLettered); + assert_eq!(view.member.as_deref(), Some("w1")); + assert_eq!(view.deliveries, 3); + } + + #[test] + fn parse_pool_result_refs_validates_contract() { + let json = r#"{"stream_seq":7,"task_id":"t-42","attempt":1,"outcome":"ok","traceparent":"00-11111111111111111111111111111111-2222222222222222-01"}"#; + let res = parse_pool_result_refs(json).unwrap(); + assert_eq!(res.stream_seq, 7); + assert_eq!(res.task_id, "t-42"); + assert_eq!(res.attempt, 1); + assert_eq!(res.outcome, PoolResultOutcome::Ok); + } +} diff --git a/specs/README.md b/specs/README.md index 8ea0374..785cb59 100644 --- a/specs/README.md +++ b/specs/README.md @@ -10,8 +10,9 @@ This directory is the LeanSpec registry of record for `unsigned-gg/ep`. Every ar | Spec | Title | Status | Priority | Scope | Tags | Primary Source Locator | | :--- | :--- | :--- | :--- | :--- | :--- | :--- | -| [**SPEC-0001**](SPEC-0001-escapement-scheduler.md) | Escapement Scheduler & Deterministic Event Reactor | `complete` | `critical` | `escapement-core` | `scheduler`, `custody`, `metering` | `unsigned-gg/ep@6765df9:specs/SPEC-0001-escapement-scheduler.md` | +| [**SPEC-0001**](SPEC-0001-escapement-scheduler.md) | Escapement Scheduler & Deterministic Event Reactor | `complete` | `critical` | `ep-domain`, `escapement-core` | `scheduler`, `custody`, `metering` | `unsigned-gg/ep@6765df9:specs/SPEC-0001-escapement-scheduler.md` | | [**SPEC-0002**](SPEC-0002-roadmap.md) | Escapement Protocol Roadmap & Evolution Horizons | `living` | `high` | `workspace` | `roadmap`, `scale-to-zero`, `formal-verification` | `unsigned-gg/ep@6765df9:specs/SPEC-0002-roadmap.md` | +| [**SPEC-0003**](SPEC-0003-cortex-revenant-mesh-architecture.md) | Three-Plane Escapement Architecture (`ep` <-> `cortex` <-> `revenant`) | `complete` | `critical` | `workspace` | `cortex`, `revenant`, `mesh-pool`, `custody` | `unsigned-gg/ep:specs/SPEC-0003-cortex-revenant-mesh-architecture.md` | ## LeanSpec Specification Format diff --git a/specs/SPEC-0001-escapement-scheduler.md b/specs/SPEC-0001-escapement-scheduler.md index 60e30b5..4f86db6 100644 --- a/specs/SPEC-0001-escapement-scheduler.md +++ b/specs/SPEC-0001-escapement-scheduler.md @@ -3,7 +3,7 @@ status: complete created: 2026-10-12 tags: [escapement, scheduler, deterministic, custody, token-bucket, leanspec] priority: critical -scope: escapement-core +scope: ep-domain, escapement-core depends_on: [] --- @@ -21,6 +21,8 @@ depends_on: [] $$\text{State}(T) \in \{\text{Admitted}, \text{Assigned}(A), \text{Running}(A), \text{Completed}, \text{Failed}, \text{Cancelled}\}$$ 6. State handoffs shall execute atomically using compare-and-swap semantics. No task shall remain unassigned without an active timeout watch. 7. Capability rate limits shall use monotonic integer token buckets $(R, B)$ refilled on tick boundaries without floating-point arithmetic. +8. Same-tick impulses shall be ordered deterministically by composite stamp `(Tick, ImpulseSeq)` (`ImpulseStamp`). +9. Loop progress evaluation shall classify stagnation patterns (`Spinning`, `Oscillation`, `NoDrift`, `DiminishingReturns`) into deterministic control directives (`Continue`, `Retry`, `Wait`, `Cancel`, `Converge`) without I/O. ```lean namespace Escapement.Spec @@ -37,9 +39,9 @@ inductive CustodyState : Type where /-- Token bucket state with discrete integer balance and burst limit. -/ structure TokenBucket where capacity : Nat - refill_rate : Nat - tokens : Nat - last_tick_ms : Nat + available : Nat + refill_per_tick : Nat + last_tick : Nat /-- Invariant: Queued tasks never exceed lane capacity. -/ def LaneCapacityBounded (queued : Nat) (capacity : Nat) : Prop := @@ -74,16 +76,21 @@ Mechanical escapements convert continuous spring potential into discrete, regula ▼ ┌────────────────────────────────────────────────────────┐ │ escapement-core │ - │ • Registry (Heartbeats, sweep, state) │ + │ • Re-exports ep-domain single source of truth │ + │ • Registry (Heartbeats, sweep, state) │ │ • Plan DAG (Kahn's topo-sort, cycle rejection) │ - │ • Lane Scheduler (Weighted fair queueing) │ - │ • Metering (Token bucket rate limits) │ - │ • WAL (Append-only state machine persistence) │ + │ • Lane Scheduler (Weighted fair queueing) │ + │ • Metering (ConcurrencyGuard + ep_domain::TokenBucket│ + │ • WAL (Append-only state machine persistence) │ └───────────────────────────┬────────────────────────────┘ ▼ ┌────────────────────────────────────────────────────────┐ │ ep-domain │ - │ (Zero-I/O Algebraic Data Types, TaskId, PlanNode) │ + │ • Tick, ImpulseSeq, ImpulseStamp, Cadence │ + │ • TokenBucket (u64 integer tokens & tick boundaries) │ + │ • ValveState, TickQuota, ValveDecision, HoldReason │ + │ • StagnationPattern, ControlDirective │ + │ • evaluate_impulse, detect_stagnation (Zero I/O) │ └────────────────────────────────────────────────────────┘ ``` @@ -99,10 +106,10 @@ Mechanical escapements convert continuous spring potential into discrete, regula ## Tasks -- [x] Implement zero-I/O domain types in `ep-domain` (`unsigned-gg/ep@6765df9:crates/ep-domain/src/lib.rs`). +- [x] Implement zero-I/O domain types (`Tick`, `ImpulseSeq`, `ImpulseStamp`, `Cadence`, `TokenBucket`, `ValveState`, `TickQuota`, `ValveDecision`, `StagnationPattern`, `ControlDirective`) in `ep-domain` (`unsigned-gg/ep@6765df9:crates/ep-domain/src/lib.rs`). +- [x] Re-export `ep-domain` from `escapement-core` and unify integer `TokenBucket` in `metering` (`unsigned-gg/ep@6765df9:crates/escapement-core/src/metering.rs`). - [x] Implement deterministic Kahn's topological sort in `escapement-core` (`unsigned-gg/ep@6765df9:crates/escapement-core/src/plan.rs`). - [x] Implement atomic task custody tracking with BLAKE3 hashes (`unsigned-gg/ep@6765df9:crates/escapement-core/src/custody.rs`). -- [x] Implement discrete token bucket rate metering (`unsigned-gg/ep@6765df9:crates/escapement-core/src/metering.rs`). - [x] Implement WAL persistence and replay reconstruction (`unsigned-gg/ep@6765df9:crates/escapement-core/src/persistence.rs`). - [x] Implement Blackwall process isolation bridge (`unsigned-gg/ep@6765df9:crates/blackwall-bridge/src/lib.rs`). - [x] Implement single-writer orchestrator daemon in `escapement-serve` (`unsigned-gg/ep@6765df9:crates/escapement-serve/src/lib.rs`). @@ -113,12 +120,13 @@ Mechanical escapements convert continuous spring potential into discrete, regula ```bash cargo test -j 2 --workspace ``` -2. **Mechanical Proofs:** 162 unit and integration tests passing across all crates. +2. **Mechanical Proofs:** Workspace unit and integration tests passing across all crates. 3. **Miri Cleanliness:** Verify memory safety and zero undefined behavior in pure domain transformations: ```bash cargo test -j 2 -p ep-domain ``` 4. **Authentic Source Locators:** + - `unsigned-gg/ep@6765df9:crates/ep-domain/src/lib.rs` - `unsigned-gg/ep@6765df9:crates/escapement-core/src/plan.rs` - `unsigned-gg/ep@6765df9:crates/escapement-core/src/custody.rs` - `unsigned-gg/ep@6765df9:crates/escapement-core/src/metering.rs` diff --git a/specs/SPEC-0003-cortex-revenant-mesh-architecture.md b/specs/SPEC-0003-cortex-revenant-mesh-architecture.md new file mode 100644 index 0000000..47b3de2 --- /dev/null +++ b/specs/SPEC-0003-cortex-revenant-mesh-architecture.md @@ -0,0 +1,129 @@ +--- +status: complete +created: 2026-10-15 +tags: [cortex, revenant, mesh-pool, custody, scheduler, leanspec] +priority: critical +scope: ep-domain, escapement-core, escapement-serve, blackwall-bridge, ep +depends_on: [SPEC-0001, SPEC-0002] +--- + +# SPEC-0003: Three-Plane Escapement Architecture (`ep` <-> `cortex` <-> `revenant`) + +## Requirements + +1. The architecture shall maintain a strict three-plane separation of responsibilities across `ep`, `cortex`, and `revenant`: + - `ep` owns deterministic tick metering, weighted fair lane queueing, Kahn DAG dependency resolution, stagnation detection, and BLAKE3 custody transitions. + - `cortex` owns Redis Consumer Group message transport (`workstreams/`), `claim_timeout_ms` crash recovery, and `/v1/mesh/message-status` reconciliation. + - `revenant` owns Blackwall OS process isolation (`/v1/runs`), autonomous loop intake (`/v1/escapement/loops`), and OTLP telemetry aggregation. +2. Every capability slug $S$ routed to a Cortex Mesh Pool shall be non-empty, maximum 64 ASCII bytes, and restricted to lowercase ASCII alphanumeric characters, hyphens (`-`), and underscores (`_`): + $$\forall c \in S,\; c \in [\text{'a'}..\text{'z'}] \cup [\text{'0'}..\text{'9'}] \cup \{\text{'-'}, \text{'\_'}\}$$ +3. Every task dispatched to a Cortex Mesh Pool shall transition atomically into `CustodyMode::Pool` (`PoolAssigned`) with immutable delivery references `(pool, message_id, stream_seq, dispatched_at_tick, attempt)` where `stream_seq >= 1` and numeric `refs` fields (`stream_seq`, `attempt`) satisfy $\le 2^{53}$ (`9,007,199,254,740,992`) for exact `pbjson_types::Struct` round-tripping. +4. Pool result reconciliation (`resolve_pool_job_state`) shall be a pure, deterministic state transition in `ep-domain` with zero I/O: + - An explicit worker result `MSG` (`refs.{stream_seq, task_id, attempt, outcome}`) with `outcome == "ok"` (or `"completed"`) resolves to `JobResolution::Complete`. + - An explicit worker result `MSG` with `outcome == "transient"` resolves to `JobResolution::RequeueOrFail`. + - An explicit worker result `MSG` with `outcome == "permanent"` (or `"failed"`) resolves to `JobResolution::Fail`. + - A Cortex `GET /v1/mesh/message-status?peer_id=&pool=&stream_seq=` query reporting `status == "dead_lettered"` (`PoolMessageStatus::DeadLettered`) resolves to `JobResolution::RequeueOrFail`. + - A Cortex `message-status` query reporting `"unread"` (`PoolMessageStatus::Unread`), `"pending"` (`PoolMessageStatus::Pending` / in-flight on `member`), or `"acked"` (`PoolMessageStatus::Acked` awaiting worker result `MSG`) resolves to `JobResolution::InFlight`. +5. Pool backpressure shall bound maximum in-flight pool jobs as a pure function of pool consumer count and lane capacity: + $$\text{max\_inflight}(N_{\text{consumers}}, C_{\text{lane}}, Q_{\text{per\_consumer}}) = \min(C_{\text{lane}}, \max(1, N_{\text{consumers}}) \times \max(1, Q_{\text{per\_consumer}}))$$ +6. Terminal job custody records (`Completed`, `Failed`) in `escapement-core::Engine` shall be retained in a bounded FIFO history capped at `MAX_TERMINAL_HISTORY = 1000` entries so `GET /jobs` and `GET /jobs/:id` never return stub responses or grow without bound. + +```lean +namespace Escapement.MeshPoolSpec + +/-- Cortex /v1/mesh/message-status states (WS-X1 / WS-X2). -/ +inductive PoolMessageStatus : Type where + | unread : PoolMessageStatus + | pending : Option String -> Nat -> PoolMessageStatus + | acked : PoolMessageStatus + | dead_lettered : Nat -> PoolMessageStatus + +/-- Deterministic resolution emitted by pure pool reconciliation. -/ +inductive JobResolution : Type where + | in_flight : JobResolution + | complete : JobResolution + | fail : JobResolution + | requeue_or_fail : JobResolution + +/-- Pure reconciliation of message status when no explicit worker result has arrived. -/ +def reconcile_status (s : PoolMessageStatus) : JobResolution := + match s with + | PoolMessageStatus.unread => JobResolution.in_flight + | PoolMessageStatus.pending _ _ => JobResolution.in_flight + | PoolMessageStatus.acked => JobResolution.in_flight + | PoolMessageStatus.dead_lettered _ => JobResolution.requeue_or_fail + +/-- Theorem: Dead-lettered pool messages always trigger deterministic recovery. -/ +theorem dead_letter_recovers (deliveries : Nat) : + reconcile_status (PoolMessageStatus.dead_lettered deliveries) = JobResolution.requeue_or_fail := by + rfl +``` + +## Design + +### Three-Plane Topology + +``` +┌──────────────────────────────────────────────────────────────────────────────┐ +│ PLANE 3: REVENANT CONTROL PLANE │ +│ • Escapement Loop Intake (/v1/escapement/loops) │ +│ • Blackwall Sandbox Execution (/v1/runs) │ +│ • Telemetry & VictoriaMetrics / OTLP Collector │ +└───────────────┬──────────────────────────────────────────────▲───────────────┘ + │ POST /plans, /tasks │ POST /v1/runs + ▼ │ +┌──────────────────────────────────────────────────────────────┴───────────────┐ +│ PLANE 1: EP ESCAPEMENT ENGINE │ +│ • ep-domain: Pure Tick, TokenBucket, CapabilitySlug, resolve_pool_job_state │ +│ • escapement-core: Kahn DAG, Weighted Fair Lanes, Custody, Bounded History │ +│ • escapement-serve: HTTP (:8080), Direct Worker API, Pool Reconciler │ +└───────────────┬──────────────────────────────────────────────▲───────────────┘ + │ POST /v1/mesh/route-send │ Result MSG + + │ (to = "workstreams/") │ message-status + ▼ │ +┌──────────────────────────────────────────────────────────────┴───────────────┐ +│ PLANE 2: CORTEX MESH TRANSPORT │ +│ • Redis Stream Consumer Groups (workstreams/) │ +│ • Claim Timeout Re-delivery & Dead-Letter Accounting │ +│ • Sovereign Lane Identity: estate/w4C:p1 (session-agy-ceres-estate-w4c-p1) │ +└──────────────────────────────────────────────────────────────────────────────┘ +``` + +### Custody Dual-Path Matrix + +| Custody Mode | Transport Path | Assignment Record | Completion Trigger | Timeout / Crash Recovery | +| :--- | :--- | :--- | :--- | :--- | +| `CustodyMode::DirectWorker` | Direct HTTP (`POST /tasks/claim`) or `blackwall-bridge` (`POST /v1/runs`) | `TaskState::Assigned(AgentId)` | `POST /tasks/complete` or `/v1/runs` response | `Engine::reconcile_orphans` sweeps stale heartbeats | +| `CustodyMode::Pool` | Cortex `POST /v1/mesh/route-send` to `workstreams/` | `TaskState::PoolAssigned(PoolDeliveryRefs)` | Worker sends result `MSG` to `service-escapement-pool` or calls `POST /tasks/complete` | Cortex consumer group `claim_timeout_ms` + `/v1/mesh/message-status` dead-letter check | + +## Tasks + +- [x] Establish sovereign lane identity (`estate/w4C:p1`, role `session-agy-ceres-estate-w4c-p1`) and `.saga` team topologies (`.saga/teams.toml`, `.saga/topologies.toml`). +- [x] Implement pure Cortex Mesh Pool types (`CapabilitySlug`, `CustodyMode`, `PoolDeliveryRefs`, `PoolResultRefs`, `PoolMessageStatus`, `JobResolution`, `resolve_pool_job_state`, `max_pool_inflight`) in `ep-domain` (`crates/ep-domain/src/lib.rs`). +- [x] Extend `escapement-core` custody and WAL with `PoolAssigned`, `Failed`, and bounded 1000-entry terminal job history (`crates/escapement-core/src/dispatch.rs`, `crates/escapement-core/src/persistence.rs`). +- [x] Replace `GET /jobs` stub in `escapement-serve` with real active and terminal job inspection (`GET /jobs`, `GET /jobs/:id`) and direct worker lifecycle endpoints (`POST /tasks/{claim,complete,fail}`) (`crates/escapement-serve/src/http.rs`). +- [x] Implement Cortex Mesh Pool dispatcher and reconciler (`crates/escapement-serve/src/pool.rs`). +- [x] Align LiteLLM gateway default model identifier to `"glm-5.2"` in `blackwall-bridge` and `escapement-serve` (`crates/blackwall-bridge/src/lib.rs`, `crates/escapement-serve/src/lib.rs`). +- [x] Implement `ep` operator CLI subcommands (`plan`, `wal`, `lanes`, `mesh`) (`crates/ep/src/main.rs`). + +## Verification + +1. **Workspace Test Suite (Host Throttled):** + ```bash + cargo test -j 2 --workspace + ``` +2. **Strict Clippy Lint Gate:** + ```bash + cargo clippy -j 2 --workspace --all-targets -- -D warnings + ``` +3. **Zero-I/O Domain Audit:** + ```bash + grep -rnE "std::(fs|net|process)|tokio" crates/ep-domain/src/ + ``` + Expected output: zero matches. + +## Notes + +- **Author:** Gullveig (`Gullveig (ep/types)`) +- **Coordinated Lanes:** `session-agy-ceres-estate-w3y-p4` (`cerebral-work/cortex`), `session-agy-ceres-estate-w3z-p1` (`cerebral-work/revenant`) +- **License:** `Apache-2.0 OR MIT OR CC0-1.0`