QueueWaves
Research cascade-monitoring application for microservice architectures.
QueueWaves models each service as a phase oscillator driven by its request
queue depth. It emits alerts when configured coherence, regime, or chimera
thresholds are crossed. A threshold crossing is not proof that a cascade is in
progress, and this repository does not establish prospective lead time over
ordinary service telemetry.
Pipeline position
QueueWaves is a complete SPO application — it instantiates the
full pipeline from data collection to alerting:
Prometheus/StatsD (queue metrics)
│
↓
Collector.poll()
│
↓
InformationalExtractor → PhaseState[]
│
↓
CouplingBuilder.build() → K_nm
│
↓
UPDEEngine.step() → phases
│
↓
compute_order_parameter() → R
│
↓
RegimeManager.evaluate() → Regime
│
↓
detect_chimera() → chimera_index
│
↓
Alerter → Slack / PagerDuty / webhook
Architecture
Collector → Pipeline → Detector → Alerter
│ │ │ │
Prometheus phase chimera Slack/
/StatsD extraction + regime PagerDuty
+ UPDE analysis webhook
The collector polls queue metrics, the pipeline extracts phases and
runs UPDE integration, the detector evaluates synchronisation health,
and the alerter fires notifications when thresholds are crossed.
Theory
Microservice queue depths oscillate around steady state. In normal
operation, these oscillations are loosely synchronised (R ≈ 0.4-0.7).
When a cascade develops:
- Upstream services accumulate requests (queue depth rises)
- Downstream services starve (queue depth falls)
- R drops sharply as phase coherence breaks
QueueWaves can surface the R drop for investigation. Any lead-time, false-alarm,
or intervention benefit must be measured on a representative service trace
against ordinary queue, error-rate, and latency baselines before operational
use.
Configuration
config
Validated QueueWaves configuration and BindingSpec compilation.
The module parses YAML configuration into typed service, threshold, coupling,
alert, server, and security records, rejecting malformed URLs, invalid channel
or extractor identifiers, unsafe threshold ordering, and missing services
before runtime use. ConfigCompiler converts a validated deployment config
into an SPO BindingSpec for local analysis only; it does not start scraping,
open sockets, or install alert sinks.
Classes
ServiceDef
dataclass
ServiceDef(
name: str,
promql: str,
layer: str,
channel: str = "P",
extractor_type: str | None = None,
)
A monitored service: name, PromQL query, hierarchy layer, and channel.
ThresholdConfig
dataclass
ThresholdConfig(
r_bad_warn: float = 0.5,
r_bad_critical: float = 0.7,
plv_cascade: float = 0.85,
imprint_chronic: float = 1.5,
cooldown_seconds: float = 300.0,
)
Anomaly detection thresholds for R_bad, PLV cascade, and chronic imprint.
CouplingConfig
dataclass
CouplingConfig(strength: float = 0.5, decay: float = 0.25)
Phase coupling strength and decay parameters.
AlertSink
dataclass
AlertSink(url: str, format: str = 'generic')
Webhook endpoint for anomaly alerts.
ServerConfig
dataclass
ServerConfig(host: str = '127.0.0.1', port: int = 8080)
QueueWaves HTTP server bind address.
SecurityConfig
dataclass
SecurityConfig(
mode: str = "development",
api_key_env: str = "QUEUEWAVES_API_KEY",
rate_limit_per_minute: int = 120,
)
QueueWaves production network security policy.
QueueWavesConfig
dataclass
QueueWavesConfig(
prometheus_url: str,
services: list[ServiceDef],
scrape_interval_s: float = 15.0,
buffer_length: int = 64,
thresholds: ThresholdConfig = ThresholdConfig(),
coupling: CouplingConfig = CouplingConfig(),
alert_sinks: list[AlertSink] = list(),
server: ServerConfig = ServerConfig(),
security: SecurityConfig = SecurityConfig(),
)
Top-level configuration for a QueueWaves deployment.
ConfigCompiler
Converts user-facing QueueWavesConfig into an SPO BindingSpec.
Methods:
compile
compile(cfg: QueueWavesConfig) -> BindingSpec
Translate QueueWavesConfig into an SPO BindingSpec.
Parameters
cfg : QueueWavesConfig
The configuration object.
Returns
BindingSpec
Translate QueueWavesConfig into an SPO BindingSpec.
Raises
ValueError
If the inputs are invalid or inconsistent.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/config.py
| def compile(self, cfg: QueueWavesConfig) -> BindingSpec:
"""Translate QueueWavesConfig into an SPO BindingSpec.
Parameters
----------
cfg : QueueWavesConfig
The configuration object.
Returns
-------
BindingSpec
Translate QueueWavesConfig into an SPO BindingSpec.
Raises
------
ValueError
If the inputs are invalid or inconsistent.
"""
layers_by_name: dict[str, list[ServiceDef]] = {}
for svc in cfg.services:
layers_by_name.setdefault(svc.layer, []).append(svc)
layers: list[HierarchyLayer] = []
osc_families: dict[str, OscillatorFamily] = {}
good_layers: list[int] = []
bad_layers: list[int] = []
for layer_name, svcs in sorted(
layers_by_name.items(), key=lambda kv: _LAYER_ORDER.get(kv[0], 99)
):
idx = _LAYER_ORDER.get(layer_name, len(layers))
osc_ids = [svc.name for svc in svcs]
layers.append(
HierarchyLayer(
name=layer_name,
index=idx,
oscillator_ids=osc_ids,
)
)
for svc in svcs:
ext = svc.extractor_type
if ext is None:
raise ValueError("service.extractor_type was not resolved")
osc_families[svc.name] = OscillatorFamily(
channel=svc.channel, extractor_type=ext, config={}
)
# micro = bad (retry storms sync), macro = good (coordinated throughput)
if layer_name == "micro":
bad_layers.append(idx)
else:
good_layers.append(idx)
boundaries = [
BoundaryDef(
name="r_bad_warn",
variable="R_bad",
lower=None,
upper=cfg.thresholds.r_bad_warn,
severity="soft",
),
BoundaryDef(
name="r_bad_critical",
variable="R_bad",
lower=None,
upper=cfg.thresholds.r_bad_critical,
severity="hard",
),
]
actuators = [
ActuatorMapping(
name="coupling_adj", knob="K", scope="global", limits=(-0.5, 0.5)
),
ActuatorMapping(
name="damping_adj", knob="zeta", scope="global", limits=(0.0, 0.5)
),
]
return BindingSpec(
name="queuewaves",
version="0.1.0",
safety_tier="production",
sample_period_s=cfg.scrape_interval_s,
control_period_s=cfg.scrape_interval_s,
layers=layers,
oscillator_families=osc_families,
coupling=CouplingSpec(
base_strength=cfg.coupling.strength,
decay_alpha=cfg.coupling.decay,
templates={},
),
drivers=DriverSpec(physical={}, informational={}, symbolic={}),
objectives=ObjectivePartition(
good_layers=good_layers, bad_layers=bad_layers
),
boundaries=boundaries,
actuators=actuators,
)
|
Functions:
load_config
load_config(path: Path) -> QueueWavesConfig
Load QueueWavesConfig from a YAML file.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/config.py
| def load_config(path: Path) -> QueueWavesConfig:
"""Load QueueWavesConfig from a YAML file."""
try:
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
except (RecursionError, yaml.YAMLError):
raise ValueError("QueueWaves config YAML parse error") from None
if not isinstance(raw, dict):
msg = f"QueueWaves config must be a YAML mapping, got {type(raw).__name__}"
raise ValueError(msg)
if "prometheus_url" not in raw:
msg = "QueueWaves config missing required key 'prometheus_url'"
raise ValueError(msg)
services = [
ServiceDef(
name=s["name"],
promql=s["promql"],
layer=s.get("layer", "micro"),
channel=s.get("channel", "P"),
extractor_type=s.get("extractor_type"),
)
for s in raw.get("services", [])
]
thresh_raw = raw.get("thresholds", {})
thresholds = ThresholdConfig(
r_bad_warn=thresh_raw.get("r_bad_warn", 0.50),
r_bad_critical=thresh_raw.get("r_bad_critical", 0.70),
plv_cascade=thresh_raw.get("plv_cascade", 0.85),
imprint_chronic=thresh_raw.get("imprint_chronic", 1.5),
cooldown_seconds=thresh_raw.get("cooldown_seconds", 300.0),
)
coup_raw = raw.get("coupling", {})
coupling_cfg = CouplingConfig(
strength=coup_raw.get("strength", 0.50),
decay=coup_raw.get("decay", 0.25),
)
sinks = [
AlertSink(url=s["url"], format=s.get("format", "generic"))
for s in raw.get("alert_sinks", [])
]
srv_raw = raw.get("server", {})
server_cfg = ServerConfig(
host=srv_raw.get("host", "127.0.0.1"),
port=srv_raw.get("port", 8080),
)
sec_raw = raw.get("security", {})
security_cfg = SecurityConfig(
mode=sec_raw.get("mode", "development"),
api_key_env=sec_raw.get("api_key_env", "QUEUEWAVES_API_KEY"),
rate_limit_per_minute=sec_raw.get("rate_limit_per_minute", 120),
)
return QueueWavesConfig(
prometheus_url=raw["prometheus_url"],
services=services,
scrape_interval_s=raw.get("scrape_interval_s", 15.0),
buffer_length=raw.get("buffer_length", 64),
thresholds=thresholds,
coupling=coupling_cfg,
alert_sinks=sinks,
server=server_cfg,
security=security_cfg,
)
|
Pipeline
pipeline
QueueWaves phase-compute pipeline for one-tick diagnostic snapshots.
PhaseComputePipeline compiles QueueWaves configuration into an SPO binding,
extracts phases from ready service buffers, advances the local UPDE state,
updates imprint memory, evaluates boundary/supervisor diagnostics, and returns
a JSON-compatible snapshot. The pipeline mutates only its owned in-memory
simulation state; it does not scrape Prometheus, send alerts, open network
sockets, or apply actions to external systems.
Classes
ServiceSnapshot
dataclass
ServiceSnapshot(
name: str,
layer: str,
phase: float,
omega: float,
amplitude: float,
imprint: float,
)
Per-service phase state at a single pipeline tick.
PipelineSnapshot
dataclass
PipelineSnapshot(
tick: int,
timestamp: float,
r_good: float,
r_bad: float,
regime: str,
services: list[ServiceSnapshot],
plv_matrix: list[list[float]],
layer_states: list[dict[str, Any]],
boundary_violations: list[str],
actions: list[dict[str, Any]],
)
Full pipeline state after one tick: order params, services, anomalies.
Methods:
to_dict
to_dict() -> dict[str, Any]
Serialise snapshot to a JSON-compatible dict.
Returns
dict[str, Any]
Serialise snapshot to a JSON-compatible dict.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/pipeline.py
| def to_dict(self) -> dict[str, Any]:
"""Serialise snapshot to a JSON-compatible dict.
Returns
-------
dict[str, Any]
Serialise snapshot to a JSON-compatible dict.
"""
return {
"tick": self.tick,
"timestamp": self.timestamp,
"r_good": self.r_good,
"r_bad": self.r_bad,
"regime": self.regime,
"services": [
{
"name": s.name,
"layer": s.layer,
"phase": s.phase,
"omega": s.omega,
"amplitude": s.amplitude,
"imprint": s.imprint,
}
for s in self.services
],
"plv_matrix": self.plv_matrix,
"layer_states": self.layer_states,
"boundary_violations": self.boundary_violations,
"actions": self.actions,
}
|
PhaseComputePipeline
PhaseComputePipeline(cfg: QueueWavesConfig)
Wraps SPO core components into a single tick() call for QueueWaves.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/pipeline.py
| def __init__(self, cfg: QueueWavesConfig):
self._cfg = cfg
compiler = ConfigCompiler()
self._spec: BindingSpec = compiler.compile(cfg)
self._n_osc = sum(len(layer.oscillator_ids) for layer in self._spec.layers)
self._service_names = [
oid for layer in self._spec.layers for oid in layer.oscillator_ids
]
builder = CouplingBuilder()
self._coupling: CouplingState = builder.build(
self._n_osc,
self._spec.coupling.base_strength,
self._spec.coupling.decay_alpha,
)
self._engine = UPDEEngine(self._n_osc, dt=self._spec.sample_period_s)
self._boundary_observer = BoundaryObserver(self._spec.boundaries)
self._regime_manager = RegimeManager()
self._supervisor = SupervisorPolicy(self._regime_manager)
self._imprint_model = ImprintModel(decay_rate=0.01, saturation=5.0)
self._imprint_state = ImprintState(m_k=np.zeros(self._n_osc), last_update=0.0)
self._extractor = PhysicalExtractor()
rng = np.random.default_rng(42)
self._phases = rng.uniform(0, TWO_PI, self._n_osc)
self._omegas = np.ones(self._n_osc, dtype=np.float64)
self._amplitudes = np.ones(self._n_osc, dtype=np.float64)
# Rolling history of (phase, amplitude) samples per oscillator for
# phase-amplitude coupling estimation. 32 samples is the minimum
# window size at which modulation_index (18-bin histogram) produces a
# stable estimate; deeper windows trade memory for smoother PAC.
self._pac_window = 32
self._subcritical_threshold = 0.1
self._phase_history: list[FloatArray] = []
self._amplitude_history: list[FloatArray] = []
self._layer_osc_ranges: dict[int, list[int]] = {}
osc_idx = 0
for layer in self._spec.layers:
n_layer = len(layer.oscillator_ids)
self._layer_osc_ranges[layer.index] = list(
range(osc_idx, osc_idx + n_layer)
)
osc_idx += n_layer
self._service_layer_map: dict[str, str] = {}
for svc in cfg.services:
self._service_layer_map[svc.name] = svc.layer
self._tick_count = 0
|
Attributes
tick_count
property
Number of pipeline ticks executed so far.
Returns
int
Number of pipeline ticks executed so far.
regime
property
Current regime label from the regime manager.
Returns
str
Current regime label from the regime manager.
imprint_levels
property
imprint_levels: FloatArray
Per-oscillator imprint memory levels.
Returns
FloatArray
Per-oscillator imprint memory levels.
Methods:
tick
tick(buffers: dict[str, FloatArray]) -> PipelineSnapshot
Run one full pipeline cycle.
Parameters
buffers : dict[str, FloatArray]
mapping service_name -> 1-D signal array (ring buffer contents).
Returns
PipelineSnapshot
The result.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/pipeline.py
| def tick(self, buffers: dict[str, FloatArray]) -> PipelineSnapshot:
"""Run one full pipeline cycle.
Parameters
----------
buffers : dict[str, FloatArray]
mapping service_name -> 1-D signal array (ring buffer contents).
Returns
-------
PipelineSnapshot
The result.
"""
self._tick_count += 1
# 1. Extract phases from raw signals via Hilbert
for i, svc_name in enumerate(self._service_names):
signal = buffers.get(svc_name)
if signal is None or len(signal) < 4:
continue
states = self._extractor.extract(signal, 1.0 / self._cfg.scrape_interval_s)
if states:
self._phases[i] = states[0].theta
self._omegas[i] = max(states[0].omega, 0.01)
self._amplitudes[i] = float(states[0].amplitude)
# 2. Imprint-modulated coupling
eff_knm = self._imprint_model.modulate_coupling(
self._coupling.knm, self._imprint_state
)
# 3. UPDE step
self._phases = self._engine.step(
self._phases, self._omegas, eff_knm, 0.0, 0.0, self._coupling.alpha
)
# 4. Order parameters per layer
layer_states: list[LayerState] = []
for layer in self._spec.layers:
osc_ids = self._layer_osc_ranges[layer.index]
if osc_ids:
r, psi = compute_order_parameter(self._phases[osc_ids])
else:
r, psi = 0.0, 0.0
layer_states.append(LayerState(R=r, psi=psi))
# 5. R_good, R_bad
good_phases = [
self._phases[i]
for idx in self._spec.objectives.good_layers
for i in self._layer_osc_ranges.get(idx, [])
]
bad_phases = [
self._phases[i]
for idx in self._spec.objectives.bad_layers
for i in self._layer_osc_ranges.get(idx, [])
]
r_good = (
compute_order_parameter(np.array(good_phases))[0] if good_phases else 0.0
)
r_bad = compute_order_parameter(np.array(bad_phases))[0] if bad_phases else 0.0
# 6. PLV matrix
n_layers = len(self._spec.layers)
plv_mat = np.zeros((n_layers, n_layers))
for li in range(n_layers):
for lj in range(li + 1, n_layers):
ids_i = self._layer_osc_ranges[self._spec.layers[li].index]
ids_j = self._layer_osc_ranges[self._spec.layers[lj].index]
if ids_i and ids_j:
min_len = min(len(ids_i), len(ids_j))
plv = compute_plv(
self._phases[ids_i[:min_len]],
self._phases[ids_j[:min_len]],
)
plv_mat[li, lj] = plv
plv_mat[lj, li] = plv
# 7. Amplitude-derived metrics
# PhysicalExtractor.amplitude is the Hilbert envelope of the raw
# service signal; even for phase-only UPDE integration we surface
# these to keep the policy engine metric chain unbroken.
mean_amp = float(np.mean(self._amplitudes)) if self._n_osc else 0.0
if self._n_osc:
sub_count = int(np.sum(self._amplitudes < self._subcritical_threshold))
sub_frac = sub_count / self._n_osc
else:
sub_frac = 0.0
self._phase_history.append(self._phases.copy())
self._amplitude_history.append(self._amplitudes.copy())
if len(self._phase_history) > self._pac_window:
self._phase_history.pop(0)
self._amplitude_history.pop(0)
pac_max_val = 0.0
if len(self._phase_history) >= self._pac_window and self._n_osc:
ph_hist = np.asarray(self._phase_history)
am_hist = np.asarray(self._amplitude_history)
pac_vals = [
modulation_index(ph_hist[:, i], am_hist[:, i])
for i in range(self._n_osc)
]
pac_max_val = float(max(pac_vals)) if pac_vals else 0.0
# 8. Boundary observation
mean_r = float(np.mean([ls.R for ls in layer_states])) if layer_states else 0.0
upde_state = UPDEState(
layers=layer_states,
cross_layer_alignment=plv_mat,
stability_proxy=mean_r,
regime_id=self._regime_manager.current_regime.value,
mean_amplitude=mean_amp,
pac_max=pac_max_val,
subcritical_fraction=sub_frac,
)
obs_values = {
"R": mean_r,
"R_bad": r_bad,
"mean_amplitude": mean_amp,
"pac_max": pac_max_val,
"subcritical_fraction": sub_frac,
}
for i, ls in enumerate(layer_states):
obs_values[f"R_{i}"] = ls.R
boundary_state = self._boundary_observer.observe(obs_values)
# 9. Supervisor actions
actions = self._supervisor.decide(upde_state, boundary_state)
# 10. Imprint update
exposure = np.array(
[
layer_states[i].R
for i, layer in enumerate(self._spec.layers)
for _ in layer.oscillator_ids
]
)
self._imprint_state = self._imprint_model.update(
self._imprint_state, exposure, self._spec.sample_period_s
)
# 11. Build snapshot
svc_snapshots = []
for i, svc_name in enumerate(self._service_names):
svc_snapshots.append(
ServiceSnapshot(
name=svc_name,
layer=self._service_layer_map.get(svc_name, "unknown"),
phase=float(self._phases[i]),
omega=float(self._omegas[i]),
amplitude=float(self._amplitudes[i]),
imprint=float(self._imprint_state.m_k[i]),
)
)
return PipelineSnapshot(
tick=self._tick_count,
timestamp=time.time(),
r_good=r_good,
r_bad=r_bad,
regime=self._regime_manager.current_regime.value,
services=svc_snapshots,
plv_matrix=plv_mat.tolist(),
layer_states=[{"R": ls.R, "psi": ls.psi} for ls in layer_states],
boundary_violations=boundary_state.violations,
actions=[
{"knob": a.knob, "scope": a.scope, "value": a.value} for a in actions
],
)
|
Functions:
Detector
detector
Pure anomaly detection over QueueWaves pipeline snapshots.
The detector classifies retry-storm formation, cascade propagation, and chronic
degradation from an immutable snapshot and validated thresholds. It returns
structured anomaly records with severity, service, value, threshold, tick, and
message fields only; dispatch, suppression, operator notification, and runtime
control decisions are handled by separate modules.
Classes
Anomaly
dataclass
Anomaly(
type: str,
severity: str,
service: str,
value: float,
threshold: float,
tick: int,
message: str,
)
A detected anomaly: retry storm, cascade propagation, or chronic degradation.
AnomalyDetector
AnomalyDetector(thresholds: ThresholdConfig)
Detects three anomaly types from pipeline snapshots.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/detector.py
| def __init__(self, thresholds: ThresholdConfig):
self._t = thresholds
|
Methods:
detect
detect(snap: PipelineSnapshot) -> list[Anomaly]
Run all anomaly checks against a pipeline snapshot.
Parameters
snap : PipelineSnapshot
The runtime snapshot.
Returns
list[Anomaly]
All anomaly checks against a pipeline snapshot.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/detector.py
| def detect(self, snap: PipelineSnapshot) -> list[Anomaly]:
"""Run all anomaly checks against a pipeline snapshot.
Parameters
----------
snap : PipelineSnapshot
The runtime snapshot.
Returns
-------
list[Anomaly]
All anomaly checks against a pipeline snapshot.
"""
anomalies: list[Anomaly] = []
anomalies.extend(self._check_retry_storm(snap))
anomalies.extend(self._check_cascade(snap))
anomalies.extend(self._check_chronic(snap))
return anomalies
|
Alerter
alerter
Webhook alert dispatch with cooldown-based anomaly deduplication.
The alerter formats QueueWaves anomalies for generic JSON or Slack-style sinks,
tracks per-anomaly cooldown state, and reports only anomalies that pass
deduplication. HTTP delivery is best-effort and logs configured sink failures
without raising control-plane exceptions; the synchronous path exercises the
same cooldown bookkeeping without network I/O for tests and dry runs.
Classes
WebhookAlerter
WebhookAlerter(
sinks: list[AlertSink], cooldown_seconds: float = 300.0
)
Posts anomaly alerts to configured webhook sinks with deduplication.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/alerter.py
| def __init__(self, sinks: list[AlertSink], cooldown_seconds: float = 300.0):
cooldown = _require_cooldown_seconds(cooldown_seconds)
self._sinks = sinks
self._cooldown = cooldown
self._last_fired: dict[str, float] = {}
self._suppressed_count: dict[str, int] = {}
|
Methods:
send
async
send(anomalies: list[Anomaly]) -> list[Anomaly]
Post anomalies to all sinks. Returns the list of actually sent anomalies.
Parameters
anomalies : list[Anomaly]
Detected anomaly records.
Returns
list[Anomaly]
Post anomalies to all sinks. Returns the list of actually sent anomalies.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/alerter.py
| async def send(self, anomalies: list[Anomaly]) -> list[Anomaly]:
"""Post anomalies to all sinks. Returns the list of actually sent anomalies.
Parameters
----------
anomalies : list[Anomaly]
Detected anomaly records.
Returns
-------
list[Anomaly]
Post anomalies to all sinks. Returns the list of actually sent anomalies.
"""
import httpx
now = time.time()
to_send: list[tuple[Anomaly, int]] = []
for a in anomalies:
key = f"{a.type}:{a.service}"
last = self._last_fired.get(key, 0.0)
if now - last < self._cooldown:
self._suppressed_count[key] = self._suppressed_count.get(key, 0) + 1
continue
suppressed = self._suppressed_count.pop(key, 0)
self._last_fired[key] = now
to_send.append((a, suppressed))
if not to_send:
return []
async with httpx.AsyncClient(timeout=10.0) as client:
for sink in self._sinks:
for anomaly, suppressed in to_send:
if sink.format == "slack":
payload = _format_slack(anomaly, suppressed)
else:
payload = _format_generic(anomaly, suppressed)
try:
resp = await client.post(sink.url, json=payload)
resp.raise_for_status()
except _SEND_ERRORS:
logger.warning("alert POST failed for configured sink")
return [a for a, _ in to_send]
|
send_sync
send_sync(anomalies: list[Anomaly]) -> list[Anomaly]
Run the synchronous dedup-only path for testing (no HTTP).
Parameters
anomalies : list[Anomaly]
Detected anomaly records.
Returns
list[Anomaly]
The synchronous dedup-only path for testing (no HTTP).
Source code in src/scpn_phase_orchestrator/apps/queuewaves/alerter.py
| def send_sync(self, anomalies: list[Anomaly]) -> list[Anomaly]:
"""Run the synchronous dedup-only path for testing (no HTTP).
Parameters
----------
anomalies : list[Anomaly]
Detected anomaly records.
Returns
-------
list[Anomaly]
The synchronous dedup-only path for testing (no HTTP).
"""
now = time.time()
sent: list[Anomaly] = []
for a in anomalies:
key = f"{a.type}:{a.service}"
last = self._last_fired.get(key, 0.0)
if now - last < self._cooldown:
self._suppressed_count[key] = self._suppressed_count.get(key, 0) + 1
continue
self._suppressed_count.pop(key, 0)
self._last_fired[key] = now
sent.append(a)
return sent
|
Collector
collector
Prometheus metric collection into per-service finite ring buffers.
MetricBuffer stores timestamp/value samples for one monitored service, and
PrometheusCollector scrapes configured PromQL instant queries into those
buffers through an optional httpx client. Scrape failures are logged per service
and do not mutate unrelated buffers. The collector only returns ready signal
arrays for downstream phase extraction; it does not infer anomalies or send
alerts.
Classes
MetricBuffer
MetricBuffer(maxlen: int = 64)
Fixed-size ring buffer of (timestamp, value) pairs for one service.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def __init__(self, maxlen: int = 64):
if maxlen < 1:
raise ValueError(f"maxlen must be >= 1, got {maxlen}")
self._maxlen = maxlen
self._buf: deque[tuple[float, float]] = deque(maxlen=maxlen)
|
Attributes
ready
property
True when at least 4 samples are buffered (minimum for phase extraction).
Returns
bool
True when at least 4 samples are buffered (minimum for phase extraction).
full
property
True when the buffer has reached maximum capacity.
Returns
bool
True when the buffer has reached maximum capacity.
Methods:
push
push(timestamp: float, value: float) -> None
Append a (timestamp, value) pair to the ring buffer.
Parameters
timestamp : float
The sample timestamp.
value : float
The value.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def push(self, timestamp: float, value: float) -> None:
"""Append a (timestamp, value) pair to the ring buffer.
Parameters
----------
timestamp : float
The sample timestamp.
value : float
The value.
"""
self._buf.append((timestamp, value))
|
values_array
values_array() -> FloatArray
Return buffered values as a float64 array (timestamps excluded).
Returns
FloatArray
Buffered values as a float64 array (timestamps excluded).
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def values_array(self) -> FloatArray:
"""Return buffered values as a float64 array (timestamps excluded).
Returns
-------
FloatArray
Buffered values as a float64 array (timestamps excluded).
"""
return np.array([v for _, v in self._buf], dtype=np.float64)
|
PrometheusCollector
PrometheusCollector(
prometheus_url: str,
queries: dict[str, str],
buffer_length: int = 64,
)
Scrapes Prometheus instant query API for configured services.
Requires httpx (installed via pip install scpn-phase-orchestrator[queuewaves]).
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def __init__(
self,
prometheus_url: str,
queries: dict[str, str],
buffer_length: int = 64,
):
self._base_url = prometheus_url.rstrip("/")
self._queries = queries
self._buffers: dict[str, MetricBuffer] = {
name: MetricBuffer(maxlen=buffer_length) for name in queries
}
self._client: Any = None
|
Attributes
buffers
property
buffers: dict[str, MetricBuffer]
Per-service metric ring buffers keyed by service name.
Returns
dict[str, MetricBuffer]
Per-service metric ring buffers keyed by service name.
Methods:
close
async
Close the underlying HTTP client.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| async def close(self) -> None:
"""Close the underlying HTTP client."""
if self._client is not None:
await self._client.aclose()
self._client = None
|
scrape
async
scrape() -> dict[str, MetricBuffer]
Fire one PromQL instant query per service, push results into buffers.
Returns
dict[str, MetricBuffer]
Fire one PromQL instant query per service, push results into buffers.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| async def scrape(self) -> dict[str, MetricBuffer]:
"""Fire one PromQL instant query per service, push results into buffers.
Returns
-------
dict[str, MetricBuffer]
Fire one PromQL instant query per service, push results into buffers.
"""
client = await self._get_client()
for name, promql in self._queries.items():
try:
resp = await client.get(
f"{self._base_url}/api/v1/query",
params={"query": promql},
)
resp.raise_for_status()
data = resp.json()
results = data.get("data", {}).get("result", [])
if results:
ts, val = results[0]["value"]
self._buffers[name].push(float(ts), float(val))
except _SCRAPE_ERRORS:
logger.warning("scrape failed for %s", name, exc_info=True)
except (KeyError, IndexError, TypeError, ValueError):
# A 2xx response whose JSON lacks ``data.result[0].value`` or
# carries a non-numeric sample must not abort the scrape of the
# remaining services; log and skip this one, honouring the
# per-service isolation contract stated in the module docstring.
logger.warning(
"malformed Prometheus response for %s", name, exc_info=True
)
return self._buffers
|
scrape_sync
scrape_sync(
values: dict[str, tuple[float, float]],
) -> dict[str, MetricBuffer]
Push values synchronously for testing: {name: (timestamp, value)}.
Parameters
values : dict[str, tuple[float, float]]
The scalar values.
Returns
dict[str, MetricBuffer]
Push values synchronously for testing: {name: (timestamp, value)}.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def scrape_sync(
self,
values: dict[str, tuple[float, float]],
) -> dict[str, MetricBuffer]:
"""Push values synchronously for testing: {name: (timestamp, value)}.
Parameters
----------
values : dict[str, tuple[float, float]]
The scalar values.
Returns
-------
dict[str, MetricBuffer]
Push values synchronously for testing: {name: (timestamp, value)}.
"""
for name, (ts, val) in values.items():
if name in self._buffers:
self._buffers[name].push(ts, val)
return self._buffers
|
get_signal_arrays
get_signal_arrays() -> dict[str, FloatArray]
Return value arrays for all ready buffers.
Returns
dict[str, FloatArray]
Value arrays for all ready buffers.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/collector.py
| def get_signal_arrays(self) -> dict[str, FloatArray]:
"""Return value arrays for all ready buffers.
Returns
-------
dict[str, FloatArray]
Value arrays for all ready buffers.
"""
return {
name: buf.values_array() for name, buf in self._buffers.items() if buf.ready
}
|
Server
server
FastAPI assembly for QueueWaves read-only monitoring surfaces.
create_app wires configured collection, phase computation, anomaly
detection, alert dispatch, REST endpoints, and a read-only WebSocket stream into
one ASGI app. Production mode requires an API key and rate limits requests
before exposing state. Incoming WebSocket messages must be explicit keepalives;
the server does not accept remote actuation commands.
Classes
Functions:
create_app
create_app(cfg: QueueWavesConfig) -> object
Build a FastAPI application wired to the given config.
Parameters
cfg : QueueWavesConfig
The configuration object.
Returns
object
A FastAPI application wired to the given config.
Raises
ValueError
If the inputs are invalid or inconsistent.
RuntimeError
If the operation fails.
HTTPException
If the request is invalid.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/server.py
| def create_app(cfg: QueueWavesConfig) -> object:
"""Build a FastAPI application wired to the given config.
Parameters
----------
cfg : QueueWavesConfig
The configuration object.
Returns
-------
object
A FastAPI application wired to the given config.
Raises
------
ValueError
If the inputs are invalid or inconsistent.
RuntimeError
If the operation fails.
HTTPException
If the request is invalid.
"""
from fastapi import (
Depends,
FastAPI,
Header,
HTTPException,
Request,
WebSocket,
WebSocketDisconnect,
)
from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
queries = {svc.name: svc.promql for svc in cfg.services}
collector = PrometheusCollector(cfg.prometheus_url, queries, cfg.buffer_length)
pipeline = PhaseComputePipeline(cfg)
detector = AnomalyDetector(cfg.thresholds)
alerter = WebhookAlerter(cfg.alert_sinks, cfg.thresholds.cooldown_seconds)
history: deque[PipelineSnapshot] = deque(maxlen=_MAX_HISTORY)
active_anomalies: list[Any] = []
ws_clients: set[WebSocket] = set()
mode = cfg.security.mode.strip().lower()
if mode not in ("development", "production"):
raise ValueError("QueueWaves security.mode must be development or production")
api_key = os.environ.get(cfg.security.api_key_env)
if mode == "production" and not api_key:
raise RuntimeError(
f"{cfg.security.api_key_env} is required when QueueWaves runs in production"
)
rate_limit = cfg.security.rate_limit_per_minute if mode == "production" else 0
if rate_limit < 0:
raise ValueError("QueueWaves rate_limit_per_minute must be non-negative")
limiter = FixedWindowRateLimiter(rate_limit) if rate_limit > 0 else None
async def _require_network_access(
request: Request,
x_api_key: str | None = Header(None),
) -> None:
"""Assert network access is permitted, else raise."""
if api_key is None:
identity = request.client.host if request.client is not None else "local"
elif x_api_key is None or not hmac.compare_digest(x_api_key, api_key):
raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")
else:
identity = x_api_key
if limiter is not None and not limiter.allow(identity):
raise HTTPException(status_code=429, detail="Rate limit exceeded")
async def _broadcast(msg: dict[str, Any]) -> None: # pragma: no cover
"""Broadcast a message to all connected websocket clients."""
payload = json.dumps(msg)
dead: list[WebSocket] = []
for ws in ws_clients:
try:
await ws.send_text(payload)
except _IO_ERRORS:
dead.append(ws)
for ws in dead:
ws_clients.discard(ws)
async def _pipeline_loop() -> None: # pragma: no cover
"""Run the detection pipeline loop until cancelled."""
nonlocal active_anomalies
while True:
try:
await collector.scrape()
except _IO_ERRORS:
logger.warning("scrape cycle failed", exc_info=True)
signals = collector.get_signal_arrays()
if not signals:
await asyncio.sleep(cfg.scrape_interval_s)
continue
snap = pipeline.tick(signals)
history.append(snap)
anomalies = detector.detect(snap)
active_anomalies = anomalies
if anomalies:
try:
await alerter.send(anomalies)
except _IO_ERRORS:
logger.warning("alert send failed", exc_info=True)
tick_msg = {"type": "tick", "data": snap.to_dict()}
await _broadcast(tick_msg)
for a in anomalies:
await _broadcast(
{
"type": "anomaly",
"data": {
"type": a.type,
"severity": a.severity,
"service": a.service,
"message": a.message,
"value": a.value,
"tick": a.tick,
},
}
)
await asyncio.sleep(cfg.scrape_interval_s)
@asynccontextmanager
async def _lifespan( # pragma: no cover
_app: FastAPI,
) -> AsyncIterator[None]:
"""Manage the server application lifespan (startup and shutdown)."""
task = asyncio.create_task(_pipeline_loop())
yield
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
await collector.close()
app = FastAPI(title="QueueWaves", version="0.1.0", lifespan=_lifespan)
static_dir = Path(__file__).parent / "static"
if static_dir.is_dir():
app.mount(
"/static",
StaticFiles(directory=str(static_dir), html=True),
name="static",
)
# --- REST endpoints ---
@app.get("/api/v1/health", dependencies=[Depends(_require_network_access)])
async def health() -> dict[str, Any]:
"""Handle GET /api/v1/health — liveness check with tick counter."""
return {"status": "ok", "tick": pipeline.tick_count}
@app.get("/api/v1/state", dependencies=[Depends(_require_network_access)])
async def state() -> Any:
"""Handle GET /api/v1/state — return latest pipeline snapshot."""
if not history:
return JSONResponse({"error": "no data yet"}, status_code=503)
return history[-1].to_dict() # pragma: no cover — data branch, ASGI thread
@app.get("/api/v1/state/history", dependencies=[Depends(_require_network_access)])
async def state_history(n: int = 100) -> list[dict[str, Any]]:
"""Handle GET /api/v1/state/history — return last n snapshots."""
return [s.to_dict() for s in _tail_page(list(history), n)]
@app.get("/api/v1/anomalies", dependencies=[Depends(_require_network_access)])
async def anomalies() -> list[dict[str, Any]]:
"""Handle GET /api/v1/anomalies — return active anomaly list."""
return [
{
"type": a.type,
"severity": a.severity,
"service": a.service,
"value": a.value,
"message": a.message,
"tick": a.tick,
}
for a in active_anomalies
]
@app.get("/api/v1/services", dependencies=[Depends(_require_network_access)])
async def services() -> list[dict[str, Any]]:
"""Handle GET /api/v1/services — return per-service phase state."""
if not history:
return []
snap = history[-1] # pragma: no cover — data branch, ASGI thread
return [ # pragma: no cover
{
"name": s.name,
"layer": s.layer,
"phase": s.phase,
"omega": s.omega,
"imprint": s.imprint,
}
for s in snap.services
]
@app.get("/api/v1/plv", dependencies=[Depends(_require_network_access)])
async def plv() -> dict[str, Any]:
"""Handle GET /api/v1/plv — return cross-layer PLV matrix."""
if not history:
return {"matrix": []}
return {"matrix": history[-1].plv_matrix} # pragma: no cover
@app.get(
"/api/v1/metrics/prometheus", dependencies=[Depends(_require_network_access)]
)
async def prom_metrics() -> PlainTextResponse:
"""Handle GET /api/v1/metrics/prometheus — export Prometheus text metrics."""
lines: list[str] = []
if history: # pragma: no cover — data branch, ASGI thread
snap = history[-1]
lines.append(f"queuewaves_r_good {snap.r_good:.6f}")
lines.append(f"queuewaves_r_bad {snap.r_bad:.6f}")
lines.append(f'queuewaves_regime{{name="{snap.regime}"}} 1')
lines.append(f"queuewaves_tick {snap.tick}")
for svc in snap.services:
lines.append(
f'queuewaves_phase{{service="{svc.name}"}} {svc.phase:.6f}'
)
lines.append(
f'queuewaves_imprint{{service="{svc.name}"}} {svc.imprint:.6f}'
)
return PlainTextResponse("\n".join(lines) + "\n", media_type="text/plain")
@app.post("/api/v1/check", dependencies=[Depends(_require_network_access)])
async def check() -> Any:
"""One-shot: scrape, analyze, return result."""
await collector.scrape()
signals = collector.get_signal_arrays()
if not signals:
return JSONResponse({"error": "not enough data"}, status_code=503)
snap = pipeline.tick(signals)
anoms = detector.detect(snap)
return {
"r_good": snap.r_good,
"r_bad": snap.r_bad,
"regime": snap.regime,
"anomalies": [
{"type": a.type, "severity": a.severity, "message": a.message}
for a in anoms
],
}
@app.get("/")
async def root() -> Any:
"""Handle GET / — serve dashboard index or fallback text."""
index = static_dir / "index.html"
if index.exists():
return FileResponse(str(index))
msg = "QueueWaves is running. No dashboard found." # pragma: no cover
return PlainTextResponse(msg) # pragma: no cover
# --- WebSocket ---
@app.websocket("/ws/stream")
async def ws_stream(websocket: WebSocket) -> None:
"""Read-only observer stream with keepalive-only inbound messages."""
ws_key = websocket.headers.get("x-api-key")
if api_key is not None and (
ws_key is None or not hmac.compare_digest(ws_key, api_key)
):
await websocket.close(code=1008, reason="Invalid or missing X-API-Key")
return
if limiter is not None:
identity = websocket.headers.get("x-api-key") or "websocket"
if not limiter.allow(identity):
await websocket.close(code=1013, reason="Rate limit exceeded")
return
await websocket.accept()
ws_clients.add(websocket)
try:
while True:
msg = await websocket.receive_text()
if _websocket_message_exceeds_limit(msg):
await websocket.close(code=1009, reason="Message too large")
break
if not _is_keepalive_message(msg):
await websocket.close(code=1003, reason="Unsupported message")
break
except WebSocketDisconnect:
pass
finally:
ws_clients.discard(websocket)
return app
|
run_server
run_server(
config_path: str,
host: str = "127.0.0.1",
port: int = 8080,
) -> None
Entry point for CLI: load config, create app, run uvicorn.
Source code in src/scpn_phase_orchestrator/apps/queuewaves/server.py
| def run_server(config_path: str, host: str = "127.0.0.1", port: int = 8080) -> None:
"""Entry point for CLI: load config, create app, run uvicorn."""
import uvicorn
cfg = load_config(Path(config_path))
app = create_app(cfg)
# type ignore: uvicorn's app parameter typing is narrower than FastAPI instances.
uvicorn.run(app, host=host, port=port, log_level="info") # type: ignore[arg-type]
|
Operational interpretation
QueueWaves is a reference application of the same reusable SPO pipeline, so this
page should be read as a production playbook example, not a side experiment.
The full cascade-response loop depends on three invariants:
- deterministic extraction,
- bounded synchronization metrics,
- auditable alert generation.
That pattern allows teams in service operations to test "is this model
behaving like a control instrument" before adding new collectors, targets, or
alert channels.
In this setup, the alert layer is the last mile. The earlier layers are
responsible for making sure phase and coupling evidence is structurally valid
before escalation.
Business interpretation
QueueWaves demonstrates how the SPO control contract translates into service
operations workflows:
- telemetry normalisation,
- phase extraction,
- synchrony risk scoring,
- deterministic escalation handoff.
The goal is not perfect prediction of every incident. It is early, explainable
detection with bounded false-merge behaviour and auditable alert evidence.
For operations teams, this page is useful as a reference implementation pattern:
same engine, different domain surface, predictable failure boundaries.