use std::collections::BTreeMap;
use std::path::Path;
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::{Map, Value};
use crate::error::{Error, Result};
use crate::event::{Envelope, Labels, Source};
pub const DEFAULT_PROFILE: &str = "planner";
pub const MONITOR_PROFILE: &str = "monitor";
const MATCHER_FIELDS: &str = "`source`, `kind`, `run_id`, `node`, `step`, `member`, `persona`";
pub const LAUNCH_CONFIG_SCHEMA_VERSION: u32 = 1;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LaunchConfig {
pub schema_version: u32,
#[serde(default, skip_serializing_if = "Filters::is_empty")]
pub filters: Filters,
}
impl Default for LaunchConfig {
fn default() -> Self {
Self {
schema_version: LAUNCH_CONFIG_SCHEMA_VERSION,
filters: Filters::default(),
}
}
}
impl LaunchConfig {
pub fn load(path: &Path) -> Result<Self> {
let text = std::fs::read_to_string(path).map_err(|source| Error::Ledger {
path: path.to_path_buf(),
source,
})?;
let named = |why: String| Error::Invalid(format!("{}: {why}", path.display()));
let config: Self =
serde_norway::from_str(&text).map_err(|failure| named(failure.to_string()))?;
if config.schema_version != LAUNCH_CONFIG_SCHEMA_VERSION {
return Err(named(format!(
"launch config schema_version {}, and this build reads \
{LAUNCH_CONFIG_SCHEMA_VERSION} — set `schema_version: \
{LAUNCH_CONFIG_SCHEMA_VERSION}`",
config.schema_version
)));
}
Ok(config)
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Filters {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub agentgraph: Option<EventFilter>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub vcs: Option<EventFilter>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub profiles: BTreeMap<String, EventFilter>,
}
impl Filters {
#[must_use]
pub fn is_empty(&self) -> bool {
self == &Self::default()
}
pub fn profile(&self, name: &str) -> Result<EventFilter> {
if let Some(filter) = self.profiles.get(name) {
return Ok(filter.clone());
}
if let Some(filter) = shipped_profile(name) {
return Ok(filter);
}
let mut names: Vec<&str> = self.profiles.keys().map(String::as_str).collect();
for shipped in [DEFAULT_PROFILE, MONITOR_PROFILE] {
if !names.contains(&shipped) {
names.push(shipped);
}
}
names.sort_unstable();
Err(Error::Invalid(format!(
"'{name}' is not a filter profile of this run; it has {}",
names.join(", ")
)))
}
}
fn shipped_profile(name: &str) -> Option<EventFilter> {
match name {
DEFAULT_PROFILE => Some(EventFilter {
include: vec![Matcher {
source: Some(Source::Pipeline),
..Matcher::default()
}],
exclude: Vec::new(),
}),
MONITOR_PROFILE => Some(EventFilter::default()),
_ => None,
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct EventFilter {
#[serde(skip_serializing_if = "Vec::is_empty")]
pub include: Vec<Matcher>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub exclude: Vec<Matcher>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct Matcher {
#[serde(skip_serializing_if = "Option::is_none")]
pub source: Option<Source>,
#[serde(skip_serializing_if = "Option::is_none")]
pub kind: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub run_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub node: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub step: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub member: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub persona: Option<String>,
}
impl EventFilter {
pub fn parse(spec: &str) -> Result<Self> {
serde_norway::from_str(spec)
.map_err(|failure| Error::Invalid(format!("the event filter is unusable: {failure}")))
}
pub fn read(spec: &str) -> Result<Self> {
if spec.trim_start().starts_with('{') {
return Self::parse(spec);
}
let document = std::fs::read_to_string(Path::new(spec)).map_err(|failure| {
Error::Invalid(format!("cannot read the event filter {spec}: {failure}"))
})?;
Self::parse(&document)
}
#[must_use]
pub fn matches(&self, envelope: &Envelope) -> bool {
self.allows(envelope.source, &envelope.kind.0, &envelope.labels)
}
#[must_use]
pub fn allows(&self, source: Source, kind: &str, labels: &Labels) -> bool {
if self
.exclude
.iter()
.any(|matcher| matcher.matches(source, kind, labels))
{
return false;
}
self.include.is_empty()
|| self
.include
.iter()
.any(|matcher| matcher.matches(source, kind, labels))
}
pub fn validate(&self) -> Result<()> {
for (list, matchers) in [("include", &self.include), ("exclude", &self.exclude)] {
for (at, matcher) in matchers.iter().enumerate() {
matcher.check().map_err(|why| {
Error::Invalid(format!(
"the event filter's {list} matcher {}: {why}",
at + 1
))
})?;
}
}
Ok(())
}
}
impl<'de> Deserialize<'de> for EventFilter {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
let document = Value::deserialize(deserializer)?;
from_document(&document).map_err(serde::de::Error::custom)
}
}
fn from_document(document: &Value) -> std::result::Result<EventFilter, String> {
let object = document.as_object().ok_or_else(|| {
format!(
"an event filter is a mapping of `include` and `exclude`, not {}",
shape(document)
)
})?;
if let Some(stray) = object
.keys()
.find(|key| !matches!(key.as_str(), "include" | "exclude"))
{
return Err(format!(
"an event filter names `include` and `exclude`; {stray:?} is neither"
));
}
let filter = EventFilter {
include: matchers(object.get("include"), "include")?,
exclude: matchers(object.get("exclude"), "exclude")?,
};
filter.validate().map_err(|refusal| refusal.to_string())?;
Ok(filter)
}
fn matchers(value: Option<&Value>, list: &str) -> std::result::Result<Vec<Matcher>, String> {
let Some(value) = value else {
return Ok(Vec::new());
};
let entries = value.as_array().ok_or_else(|| {
format!(
"an event filter's `{list}` is a list of matchers, not {}",
shape(value)
)
})?;
entries
.iter()
.enumerate()
.map(|(index, entry)| matcher(entry, list, index + 1))
.collect()
}
fn matcher(value: &Value, list: &str, position: usize) -> std::result::Result<Matcher, String> {
let named = format!("the event filter's {list} matcher {position}");
let fields = value.as_object().ok_or_else(|| {
format!(
"{named} is a mapping of matcher fields, not {}",
shape(value)
)
})?;
let mut matcher = Matcher::default();
for (field, value) in fields {
match field.as_str() {
"source" => {
matcher.source = Some(
serde_json::from_value(value.clone())
.map_err(|failure| format!("{named} names no source family: {failure}"))?,
);
}
"kind" => matcher.kind = Some(text(value, &named, field)?),
"run_id" => matcher.run_id = Some(text(value, &named, field)?),
"node" => matcher.node = Some(text(value, &named, field)?),
"step" => matcher.step = Some(text(value, &named, field)?),
"member" => matcher.member = Some(text(value, &named, field)?),
"persona" => matcher.persona = Some(text(value, &named, field)?),
unknown => {
return Err(format!(
"{named} names {unknown:?}, which is not a matcher field ({MATCHER_FIELDS})"
))
}
}
}
Ok(matcher)
}
fn text(value: &Value, named: &str, field: &str) -> std::result::Result<String, String> {
value.as_str().map(str::to_owned).ok_or_else(|| {
format!(
"{named} matches {field} against {}, which is not a string",
shape(value)
)
})
}
fn shape(value: &Value) -> &'static str {
match value {
Value::Null => "nothing",
Value::Bool(_) => "a boolean",
Value::Number(_) => "a number",
Value::String(_) => "a string",
Value::Array(_) => "a list",
Value::Object(_) => "a mapping",
}
}
impl Matcher {
fn labels_asked(&self) -> [(&'static str, Option<&str>); 5] {
[
("run_id", self.run_id.as_deref()),
("node", self.node.as_deref()),
("step", self.step.as_deref()),
("member", self.member.as_deref()),
("persona", self.persona.as_deref()),
]
}
fn matches(&self, source: Source, kind: &str, labels: &Labels) -> bool {
if self.source.is_some_and(|named| named != source) {
return false;
}
if self
.kind
.as_deref()
.is_some_and(|pattern| !glob(pattern, kind))
{
return false;
}
let typed = [
labels.run_id.as_deref(),
labels.node.as_deref(),
labels.step.as_deref(),
None,
labels.persona.as_deref(),
];
self.labels_asked()
.iter()
.zip(typed)
.all(|((key, asked), typed)| match asked {
None => true,
Some(asked) => stamped(&labels.extra, key, typed) == Some(*asked),
})
}
fn check(&self) -> std::result::Result<(), String> {
let mut named = usize::from(self.source.is_some());
for (field, asked) in
std::iter::once(("kind", self.kind.as_deref())).chain(self.labels_asked())
{
let Some(asked) = asked else { continue };
named += 1;
if asked.trim().is_empty() {
return Err(format!(
"`{field}` is empty, and nothing on the stream carries an empty {field} — \
omit the field to leave it unasked"
));
}
}
if named == 0 {
return Err(format!(
"a matcher naming no field matches every event — name at least one of \
{MATCHER_FIELDS}"
));
}
Ok(())
}
}
fn stamped<'a>(
extra: &'a Map<String, Value>,
key: &str,
typed: Option<&'a str>,
) -> Option<&'a str> {
typed.or_else(|| extra.get(key).and_then(Value::as_str))
}
fn glob(pattern: &str, text: &str) -> bool {
let pattern: Vec<char> = pattern.chars().collect();
let text: Vec<char> = text.chars().collect();
let (mut p, mut t) = (0, 0);
let (mut star, mut resume) = (None, 0);
while t < text.len() {
if pattern.get(p) == Some(&'*') {
star = Some(p);
resume = t;
p += 1;
} else if pattern.get(p) == Some(&text[t]) {
p += 1;
t += 1;
} else if let Some(at) = star {
p = at + 1;
resume += 1;
t = resume;
} else {
return false;
}
}
pattern[p..].iter().all(|character| *character == '*')
}
#[cfg(test)]
mod tests {
use super::*;
const GOLDEN: &str = include_str!("../tests/golden/launch-config-v1.json");
fn golden() -> LaunchConfig {
let kind = |glob: &str| Matcher {
kind: Some(glob.to_string()),
..Matcher::default()
};
LaunchConfig {
schema_version: LAUNCH_CONFIG_SCHEMA_VERSION,
filters: Filters {
agentgraph: Some(EventFilter {
include: Vec::new(),
exclude: vec![kind("turn-activity")],
}),
vcs: Some(EventFilter {
include: vec![kind("gate-*"), kind("session-closed")],
exclude: Vec::new(),
}),
profiles: [
(
DEFAULT_PROFILE.to_string(),
shipped_profile(DEFAULT_PROFILE).expect("planner ships"),
),
(
MONITOR_PROFILE.to_string(),
shipped_profile(MONITOR_PROFILE).expect("monitor ships"),
),
]
.into_iter()
.collect(),
},
}
}
#[test]
fn a_schema_1_launch_config_is_the_shape_the_golden_pins() {
let rendered = serde_json::to_string_pretty(&golden()).expect("it serialises");
assert_eq!(
rendered.trim(),
GOLDEN.trim(),
"the launch config changed shape. If that was deliberate, bump \
LAUNCH_CONFIG_SCHEMA_VERSION and update tests/golden/launch-config-v1.json \
together"
);
}
#[test]
fn the_schema_version_and_the_golden_name_the_same_number() {
let parsed: LaunchConfig = serde_json::from_str(GOLDEN).expect("the golden parses");
assert_eq!(parsed.schema_version, LAUNCH_CONFIG_SCHEMA_VERSION);
assert_eq!(parsed, golden(), "the golden is not the document it pins");
}
#[test]
fn a_launch_config_declaring_no_events_omits_the_block_and_round_trips() {
let bare = LaunchConfig::default();
let rendered = serde_json::to_string(&bare).expect("it serialises");
assert_eq!(rendered, r#"{"schema_version":1}"#);
assert_eq!(
serde_json::from_str::<LaunchConfig>(&rendered).expect("it re-parses"),
bare
);
let minimal: LaunchConfig =
serde_norway::from_str("schema_version: 1\n").expect("a bare config parses");
assert_eq!(minimal, bare);
assert!(minimal.filters.is_empty());
}
#[test]
fn a_launch_config_round_trips_without_losing_or_inventing_a_field() {
let full = golden();
let text = serde_norway::to_string(&full).expect("it serialises as YAML too");
assert_eq!(
serde_norway::from_str::<LaunchConfig>(&text).expect("it re-parses"),
full
);
let value: Value = serde_json::from_str(GOLDEN).expect("the golden is JSON");
assert_eq!(
value["filters"]["profiles"]["monitor"],
serde_json::json!({})
);
assert!(
value["filters"]["agentgraph"].get("include").is_none(),
"an empty include was written out: {value}"
);
}
#[test]
fn a_launch_config_this_build_cannot_read_is_refused_by_name() {
let root = std::env::temp_dir().join(format!("onepipeline-config-{}", std::process::id()));
std::fs::create_dir_all(&root).expect("a scratch directory");
let written = |name: &str, body: &str| {
let path = root.join(name);
std::fs::write(&path, body).expect("the config is written");
path
};
let later = LaunchConfig::load(&written("later.yaml", "schema_version: 2\n"))
.expect_err("a version this build does not read is refused");
let said = later.to_string();
assert!(said.contains("schema_version 2"), "{said}");
assert!(said.contains("schema_version: 1"), "{said}");
let stray = LaunchConfig::load(&written(
"stray.yaml",
"schema_version: 1\nfilterz:\n vcs: {}\n",
))
.expect_err("a key this schema does not declare is refused");
assert!(stray.to_string().contains("filterz"), "{stray}");
let unusable = LaunchConfig::load(&written(
"unusable.yaml",
"schema_version: 1\nfilters:\n vcs:\n include:\n - role: agent\n",
))
.expect_err("a matcher field the grammar does not have is refused");
assert!(unusable.to_string().contains("role"), "{unusable}");
let missing = LaunchConfig::load(&root.join("nothing-here.yaml"))
.expect_err("a file that is not there is refused");
assert!(missing.to_string().contains("nothing-here"), "{missing}");
let _ = std::fs::remove_dir_all(&root);
}
}