use crate::application::plugin_registry::PluginRegistry;
use crate::application::projection::ProjectionInput;
use crate::application::projection_registry::{ProjectionRegistry, TargetForm};
use crate::application::spec_registry::SpecRegistry;
use crate::domain::error::{WireError, WireResult};
use crate::domain::graph::Node;
use crate::domain::specification::Specification;
use crate::infrastructure::storage::SqliteStorage;
fn resolve_engine_render(
registry: &PluginRegistry,
hint: Option<&str>,
template: &str,
data: &serde_json::Value,
) -> WireResult<String> {
let id = hint.unwrap_or("handlebars");
let engine = registry
.engine(id)
.ok_or_else(|| WireError::Storage(format!("template engine '{id}' not registered")))?;
engine.render(template, data)
}
fn assert_static_projection_kind(
projection_name: &str,
projection_kind: Option<&str>,
) -> WireResult<()> {
match projection_kind {
None | Some("static") => Ok(()),
Some(other) => Err(WireError::Other(format!(
"projection '{projection_name}' has projection_kind '{other}' — \
non-static kinds require the async path; use wire_prompt_context instead"
))),
}
}
#[allow(clippy::too_many_arguments)]
async fn resolve_projection_render_async(
registry: &PluginRegistry,
template_engine_hint: Option<&str>,
projection_kind_hint: Option<&str>,
template: &str,
target_form: TargetForm,
spec_result: &serde_json::Value,
persona_id: Option<&str>,
config: Option<&serde_json::Value>,
) -> WireResult<String> {
let engine_id = template_engine_hint.unwrap_or("handlebars");
let engine = registry.engine(engine_id).ok_or_else(|| {
WireError::Storage(format!("template engine '{engine_id}' not registered"))
})?;
let kind_id = projection_kind_hint.unwrap_or("static");
let projection = registry
.projection(kind_id)
.ok_or_else(|| WireError::Storage(format!("projection kind '{kind_id}' not registered")))?;
let null = serde_json::Value::Null;
let input = ProjectionInput {
spec_result,
template,
template_engine: engine.as_ref(),
target_form,
persona_id,
config: config.unwrap_or(&null),
};
projection.render(input).await
}
pub struct WireInitInput {
pub persona_id: String,
}
#[derive(Debug)]
pub struct RenderedProjection {
pub name: String,
pub target_form: TargetForm,
pub rendered: String,
}
pub struct WireInitOutput {
pub persona_id: String,
pub projections: Vec<RenderedProjection>,
pub warnings: Vec<String>,
}
pub fn wire_init(
input: WireInitInput,
storage: &SqliteStorage,
registry: &PluginRegistry,
) -> WireResult<WireInitOutput> {
let spec_reg = SpecRegistry::new(storage);
let proj_reg = ProjectionRegistry::new(storage);
let mut projections = Vec::new();
let mut warnings = Vec::new();
for name in proj_reg.list()? {
let Some(proj) = proj_reg.get(&name)? else {
continue;
};
let Some(spec) = spec_reg.get(&proj.spec_ref)? else {
warnings.push(format!(
"projection '{name}': spec_ref '{}' not registered",
proj.spec_ref
));
continue;
};
let matched = collect_matching_nodes(storage, &spec)?;
let names: Vec<&str> = matched.iter().map(|n| n.id.as_str()).collect();
let nodes_json: Vec<serde_json::Value> = matched
.iter()
.map(|n| {
serde_json::json!({
"id": n.id,
"type": n.r#type,
"metadata": n.metadata,
})
})
.collect();
let data = serde_json::json!({
"count": matched.len(),
"names": names.join(", "),
"nodes": nodes_json,
"persona_id": input.persona_id,
});
assert_static_projection_kind(&proj.name, proj.projection_kind.as_deref())?;
let rendered = resolve_engine_render(
registry,
proj.template_engine.as_deref(),
&proj.template,
&data,
)?;
projections.push(RenderedProjection {
name: proj.name,
target_form: proj.target_form,
rendered,
});
}
Ok(WireInitOutput {
persona_id: input.persona_id,
projections,
warnings,
})
}
#[derive(Debug)]
pub struct WirePromptContextInput {
pub persona_id: String,
pub projection_names: Option<Vec<String>>,
}
#[derive(Debug)]
pub struct WirePromptContextOutput {
pub persona_id: String,
pub prompt_context: String,
pub projections: Vec<RenderedProjection>,
pub warnings: Vec<String>,
}
struct CollectedAxis {
axis: String,
source_uri: String,
target_form: TargetForm,
template: String,
template_engine: Option<String>,
projection_kind: Option<String>,
projection_config: Option<serde_json::Value>,
projection_name: String,
}
pub async fn wire_prompt_context(
input: WirePromptContextInput,
storage: std::sync::Arc<std::sync::Mutex<SqliteStorage>>,
registry: &PluginRegistry,
) -> WireResult<WirePromptContextOutput> {
use crate::application::persona_pack_resolver::read_projection_overlays;
let overlays = read_projection_overlays(&input.persona_id)?.unwrap_or_default();
let mut warnings = Vec::new();
let collected: Vec<CollectedAxis> = {
let s = storage.lock().map_err(|_| {
crate::domain::error::WireError::Storage("storage mutex poisoned".to_string())
})?;
let proj_reg = ProjectionRegistry::new(&s);
let axes: Vec<String> = if let Some(names) = input.projection_names.as_ref() {
names.clone()
} else {
let spec = Specification::And(vec![
Specification::TypeIs("outline_node".to_string()),
Specification::MetadataEq {
path: "persona".to_string(),
value: serde_json::json!(input.persona_id),
},
]);
let nodes = collect_matching_nodes(&s, &spec)?;
nodes
.iter()
.filter_map(|n| {
n.metadata
.get("axis")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
})
.collect()
};
let mut out: Vec<CollectedAxis> = Vec::new();
for axis in &axes {
let node_id = format!("{}.{}", input.persona_id, axis);
let Some(node) = s.get_node(&node_id)? else {
continue; };
let Some(source_uri) = node.metadata.get("source_uri").and_then(|v| v.as_str()) else {
warnings.push(format!(
"wiring entry '{node_id}' lacks metadata.source_uri — axis skipped"
));
continue;
};
let projection_name = format!("{}.section.{}", input.persona_id, axis);
let (base_template, base_target, base_engine, base_kind, base_config) =
match proj_reg.get(&projection_name)? {
Some(proj) => (
proj.template,
proj.target_form,
proj.template_engine,
proj.projection_kind,
proj.projection_config,
),
None => {
warnings.push(format!(
"axis '{axis}' has no registered projection \
'{projection_name}' — axis skipped"
));
continue;
}
};
let (final_template, final_target) = if let Some(o) = overlays.get(axis) {
(o.strategy.merge(&base_template, &o.template), o.target_form)
} else {
(base_template, base_target)
};
out.push(CollectedAxis {
axis: axis.clone(),
source_uri: source_uri.to_string(),
target_form: final_target,
template: final_template,
template_engine: base_engine,
projection_kind: base_kind,
projection_config: base_config,
projection_name: projection_name.clone(),
});
}
out
};
let mut projections = Vec::new();
for c in collected {
let fetched = match registry.adapter_for_uri(&c.source_uri) {
Some(adapter) => match adapter.fetch(&c.source_uri).await {
Ok(v) => v,
Err(e) => {
warnings.push(format!(
"adapter fetch failed for axis '{}' (uri={}): {e}",
c.axis, c.source_uri
));
serde_json::Value::Null
}
},
None => {
warnings.push(format!(
"no adapter registered for scheme in uri '{}' — axis '{}' skipped",
c.source_uri, c.axis
));
serde_json::Value::Null
}
};
let entries = vec![serde_json::json!({
"wiring_entry": {
"axis": c.axis,
"source_uri": c.source_uri,
},
"fetched_data": fetched,
})];
let data = serde_json::json!({
"count": 1,
"axis": c.axis,
"entries": entries,
"persona_id": input.persona_id,
});
let rendered = resolve_projection_render_async(
registry,
c.template_engine.as_deref(),
c.projection_kind.as_deref(),
&c.template,
c.target_form,
&data,
Some(input.persona_id.as_str()),
c.projection_config.as_ref(),
)
.await?;
projections.push(RenderedProjection {
name: c.projection_name.clone(),
target_form: c.target_form,
rendered,
});
}
let prompt_context = projections
.iter()
.map(|p| p.rendered.as_str())
.collect::<Vec<_>>()
.join("\n");
Ok(WirePromptContextOutput {
persona_id: input.persona_id,
prompt_context,
projections,
warnings,
})
}
fn collect_matching_nodes(storage: &SqliteStorage, spec: &Specification) -> WireResult<Vec<Node>> {
let mut out = Vec::new();
for t in storage.list_types_by_kind("node")? {
for n in storage.list_nodes_by_type(&t)? {
if spec.is_satisfied_by(&n) {
out.push(n);
}
}
}
Ok(out)
}
pub struct GraphScanSummary {
pub orphan_node_count: usize,
pub total_node_count: usize,
pub total_edge_count: usize,
}
pub fn graph_scan_summary(storage: &SqliteStorage) -> WireResult<GraphScanSummary> {
let mut total_nodes = 0_usize;
let mut total_edges = 0_usize;
let mut orphan = 0_usize;
for t in storage.list_types_by_kind("node")? {
for n in storage.list_nodes_by_type(&t)? {
total_nodes += 1;
let out_edges = storage.list_edges_from(&n.id)?;
let in_edges = storage.list_edges_to(&n.id)?;
total_edges += out_edges.len();
if out_edges.is_empty() && in_edges.is_empty() {
orphan += 1;
}
}
}
Ok(GraphScanSummary {
orphan_node_count: orphan,
total_node_count: total_nodes,
total_edge_count: total_edges,
})
}
pub struct WireCloseInput {
pub persona_id: String,
}
pub struct WireCloseOutput {
pub persona_id: String,
pub orphan_node_count: usize,
pub total_node_count: usize,
pub total_edge_count: usize,
pub report_markdown: String,
}
pub fn wire_close(input: WireCloseInput, storage: &SqliteStorage) -> WireResult<WireCloseOutput> {
let summary = graph_scan_summary(storage)?;
let persona = &input.persona_id;
let report_markdown = format!(
"# wire_close report for `{persona}`\n\n\
- total nodes: {total_nodes}\n\
- total edges: {total_edges}\n\
- orphan nodes (0 in + 0 out): {orphan}\n",
total_nodes = summary.total_node_count,
total_edges = summary.total_edge_count,
orphan = summary.orphan_node_count,
);
Ok(WireCloseOutput {
persona_id: input.persona_id,
orphan_node_count: summary.orphan_node_count,
total_node_count: summary.total_node_count,
total_edge_count: summary.total_edge_count,
report_markdown,
})
}
pub struct WireDoctorOutput {
pub orphan_node_count: usize,
pub total_node_count: usize,
pub total_edge_count: usize,
pub report_markdown: String,
}
pub fn wire_doctor(storage: &SqliteStorage) -> WireResult<WireDoctorOutput> {
let summary = graph_scan_summary(storage)?;
let report_markdown = format!(
"# wire_doctor report\n\n\
- total nodes: {total_nodes}\n\
- total edges: {total_edges}\n\
- orphan nodes (0 in + 0 out): {orphan}\n",
total_nodes = summary.total_node_count,
total_edges = summary.total_edge_count,
orphan = summary.orphan_node_count,
);
Ok(WireDoctorOutput {
orphan_node_count: summary.orphan_node_count,
total_node_count: summary.total_node_count,
total_edge_count: summary.total_edge_count,
report_markdown,
})
}
#[derive(Debug)]
pub struct WireQueryInput {
pub spec: Option<Specification>,
pub spec_ref: Option<String>,
pub limit: Option<usize>,
pub offset: Option<usize>,
}
#[derive(Debug)]
pub struct WireQueryNode {
pub id: String,
pub r#type: String,
pub metadata: serde_json::Value,
}
#[derive(Debug)]
pub struct WireQueryOutput {
pub matched: Vec<WireQueryNode>,
pub total_count: usize,
pub returned_count: usize,
}
pub fn wire_query(input: WireQueryInput, storage: &SqliteStorage) -> WireResult<WireQueryOutput> {
let resolved: Specification = match (input.spec, input.spec_ref.as_deref()) {
(Some(s), None) => s,
(None, Some(name)) => SpecRegistry::new(storage)
.get(name)?
.ok_or_else(|| crate::domain::error::WireError::NotFound(format!("spec: {name}")))?,
(Some(_), Some(_)) => {
return Err(crate::domain::error::WireError::InvalidSpec(
"spec and spec_ref are mutually exclusive".into(),
));
}
(None, None) => {
return Err(crate::domain::error::WireError::InvalidSpec(
"either spec or spec_ref is required".into(),
));
}
};
let all = collect_matching_nodes(storage, &resolved)?;
let total_count = all.len();
let offset = input.offset.unwrap_or(0);
let slice: Vec<Node> = match input.limit {
Some(lim) => all.into_iter().skip(offset).take(lim).collect(),
None => all.into_iter().skip(offset).collect(),
};
let returned_count = slice.len();
let matched = slice
.into_iter()
.map(|n| WireQueryNode {
id: n.id,
r#type: n.r#type,
metadata: n.metadata,
})
.collect();
Ok(WireQueryOutput {
matched,
total_count,
returned_count,
})
}
#[derive(Debug)]
pub struct WireRenderInput {
pub projection_ref: String,
}
#[derive(Debug)]
pub struct WireRenderOutput {
pub name: String,
pub target_form: TargetForm,
pub rendered: String,
}
pub fn wire_render(
input: WireRenderInput,
storage: &SqliteStorage,
registry: &PluginRegistry,
) -> WireResult<WireRenderOutput> {
let proj = ProjectionRegistry::new(storage)
.get(&input.projection_ref)?
.ok_or_else(|| {
crate::domain::error::WireError::NotFound(format!(
"projection: {}",
input.projection_ref
))
})?;
let spec = SpecRegistry::new(storage)
.get(&proj.spec_ref)?
.ok_or_else(|| {
crate::domain::error::WireError::NotFound(format!(
"spec_ref (dangling): {}",
proj.spec_ref
))
})?;
let matched = collect_matching_nodes(storage, &spec)?;
let names: Vec<&str> = matched.iter().map(|n| n.id.as_str()).collect();
let nodes_json: Vec<serde_json::Value> = matched
.iter()
.map(|n| {
serde_json::json!({
"id": n.id,
"type": n.r#type,
"metadata": n.metadata,
})
})
.collect();
let data = serde_json::json!({
"count": matched.len(),
"names": names.join(", "),
"nodes": nodes_json,
});
assert_static_projection_kind(&proj.name, proj.projection_kind.as_deref())?;
let rendered = resolve_engine_render(
registry,
proj.template_engine.as_deref(),
&proj.template,
&data,
)?;
Ok(WireRenderOutput {
name: proj.name,
target_form: proj.target_form,
rendered,
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WireNodeUpdateMode {
Merge,
Replace,
}
impl WireNodeUpdateMode {
pub fn as_str(self) -> &'static str {
match self {
WireNodeUpdateMode::Merge => "merge",
WireNodeUpdateMode::Replace => "replace",
}
}
pub fn parse(s: &str) -> WireResult<Self> {
match s {
"merge" => Ok(WireNodeUpdateMode::Merge),
"replace" => Ok(WireNodeUpdateMode::Replace),
other => Err(WireError::Other(format!(
"unknown wire_node_update mode '{other}' — expected 'merge' or 'replace'"
))),
}
}
}
#[derive(Debug)]
pub struct WireNodeUpdateInput {
pub id: String,
pub metadata_patch: serde_json::Value,
pub mode: WireNodeUpdateMode,
}
#[derive(Debug)]
pub struct WireNodeUpdateOutput {
pub id: String,
pub mode: WireNodeUpdateMode,
pub metadata: serde_json::Value,
}
pub fn wire_node_update(
input: WireNodeUpdateInput,
storage: &SqliteStorage,
) -> WireResult<WireNodeUpdateOutput> {
if !input.metadata_patch.is_object() {
return Err(WireError::Other(format!(
"wire_node_update: metadata_patch must be a JSON object, got {}",
type_name_of(&input.metadata_patch)
)));
}
let Some(existing) = storage.get_node(&input.id)? else {
return Err(WireError::NotFound(format!("node: {}", input.id)));
};
let final_metadata = match input.mode {
WireNodeUpdateMode::Replace => input.metadata_patch.clone(),
WireNodeUpdateMode::Merge => {
let mut base = match existing.metadata {
serde_json::Value::Object(map) => map,
_ => serde_json::Map::new(),
};
if let serde_json::Value::Object(patch_obj) = &input.metadata_patch {
for (k, v) in patch_obj {
if v.is_null() {
base.remove(k);
} else {
base.insert(k.clone(), v.clone());
}
}
}
serde_json::Value::Object(base)
}
};
let updated = storage.update_node_metadata(&input.id, &final_metadata)?;
if !updated {
return Err(WireError::Storage(format!(
"wire_node_update: row '{}' vanished between read and write",
input.id
)));
}
Ok(WireNodeUpdateOutput {
id: input.id,
mode: input.mode,
metadata: final_metadata,
})
}
fn type_name_of(v: &serde_json::Value) -> &'static str {
match v {
serde_json::Value::Null => "null",
serde_json::Value::Bool(_) => "bool",
serde_json::Value::Number(_) => "number",
serde_json::Value::String(_) => "string",
serde_json::Value::Array(_) => "array",
serde_json::Value::Object(_) => "object",
}
}
#[derive(Debug)]
pub struct WireDeleteInput {
pub id_or_name: String,
}
#[derive(Debug)]
pub struct WireDeleteOutput {
pub kind: &'static str,
pub id_or_name: String,
pub deleted: bool,
}
pub fn wire_node_delete(
input: WireDeleteInput,
storage: &SqliteStorage,
) -> WireResult<WireDeleteOutput> {
let deleted = storage.delete_node(&input.id_or_name)?;
Ok(WireDeleteOutput {
kind: "node",
id_or_name: input.id_or_name,
deleted,
})
}
pub fn wire_edge_delete(
input: WireDeleteInput,
storage: &SqliteStorage,
) -> WireResult<WireDeleteOutput> {
let deleted = storage.delete_edge(&input.id_or_name)?;
Ok(WireDeleteOutput {
kind: "edge",
id_or_name: input.id_or_name,
deleted,
})
}
pub fn wire_spec_delete(
input: WireDeleteInput,
storage: &SqliteStorage,
) -> WireResult<WireDeleteOutput> {
let deleted = storage.delete_specification(&input.id_or_name)?;
Ok(WireDeleteOutput {
kind: "spec",
id_or_name: input.id_or_name,
deleted,
})
}
pub fn wire_projection_delete(
input: WireDeleteInput,
storage: &SqliteStorage,
) -> WireResult<WireDeleteOutput> {
let deleted = storage.delete_projection(&input.id_or_name)?;
Ok(WireDeleteOutput {
kind: "projection",
id_or_name: input.id_or_name,
deleted,
})
}
pub struct WireNodesCreateBatchInput {
pub nodes: Vec<Node>,
}
pub struct WireBatchOutput {
pub inserted_count: usize,
pub failed_at: Option<usize>,
pub error_message: Option<String>,
}
pub fn wire_nodes_create_batch(
input: WireNodesCreateBatchInput,
storage: &SqliteStorage,
) -> WireResult<WireBatchOutput> {
for (i, n) in input.nodes.iter().enumerate() {
if let Err(e) = storage.insert_node(n) {
return Ok(WireBatchOutput {
inserted_count: i,
failed_at: Some(i),
error_message: Some(e.to_string()),
});
}
}
Ok(WireBatchOutput {
inserted_count: input.nodes.len(),
failed_at: None,
error_message: None,
})
}
pub struct WireEdgesCreateBatchInput {
pub edges: Vec<crate::domain::graph::Edge>,
}
pub fn wire_edges_create_batch(
input: WireEdgesCreateBatchInput,
storage: &SqliteStorage,
) -> WireResult<WireBatchOutput> {
for (i, e) in input.edges.iter().enumerate() {
if let Err(err) = storage.insert_edge(e) {
return Ok(WireBatchOutput {
inserted_count: i,
failed_at: Some(i),
error_message: Some(err.to_string()),
});
}
}
Ok(WireBatchOutput {
inserted_count: input.edges.len(),
failed_at: None,
error_message: None,
})
}
const WORKFLOW_TYPE: &str = "workflow_def";
const TRIGGER_KINDS_P5A: &[&str] = &["on_demand", "on_event"];
const ACTION_KINDS_P5A: &[&str] = &["no_op", "emit_projection"];
#[derive(Debug)]
pub struct WireWorkflowRegisterInput {
pub id: String,
pub persona_id: Option<String>,
pub trigger: serde_json::Value,
pub action: serde_json::Value,
pub enabled: Option<bool>,
}
#[derive(Debug)]
pub struct WireWorkflowRegisterOutput {
pub id: String,
}
pub fn wire_workflow_register(
input: WireWorkflowRegisterInput,
storage: &SqliteStorage,
) -> WireResult<WireWorkflowRegisterOutput> {
let trigger_kind = read_kind(&input.trigger, "trigger")?;
if !TRIGGER_KINDS_P5A.contains(&trigger_kind.as_str()) {
return Err(crate::domain::error::WireError::InvalidSpec(format!(
"trigger.kind '{trigger_kind}' not supported in P5-a (allowed: {:?})",
TRIGGER_KINDS_P5A
)));
}
if trigger_kind == "on_event" {
require_string_field(&input.trigger, "event", "trigger.event")?;
}
let action_kind = read_kind(&input.action, "action")?;
if !ACTION_KINDS_P5A.contains(&action_kind.as_str()) {
return Err(crate::domain::error::WireError::InvalidSpec(format!(
"action.kind '{action_kind}' not supported in P5-a (allowed: {:?})",
ACTION_KINDS_P5A
)));
}
if action_kind == "emit_projection" {
let names = input
.action
.get("projection_names")
.and_then(|v| v.as_array())
.ok_or_else(|| {
crate::domain::error::WireError::InvalidSpec(
"action.projection_names (array) is required for action.kind \
'emit_projection'"
.to_string(),
)
})?;
if names.is_empty() {
return Err(crate::domain::error::WireError::InvalidSpec(
"action.projection_names must contain at least one axis name".to_string(),
));
}
for n in names {
if !n.is_string() {
return Err(crate::domain::error::WireError::InvalidSpec(
"action.projection_names entries must all be strings".to_string(),
));
}
}
}
let mut metadata = serde_json::Map::new();
if let Some(p) = input.persona_id.as_ref() {
metadata.insert("persona".to_string(), serde_json::json!(p));
}
metadata.insert("trigger".to_string(), input.trigger);
metadata.insert("action".to_string(), input.action);
metadata.insert(
"enabled".to_string(),
serde_json::json!(input.enabled.unwrap_or(true)),
);
let node = Node {
id: input.id.clone(),
r#type: WORKFLOW_TYPE.to_string(),
sot_ref: None,
confidence: None,
applicability: None,
last_verified_at: None,
review_due: None,
version: 1,
prev_id: None,
metadata: serde_json::Value::Object(metadata),
};
storage.insert_node(&node)?;
Ok(WireWorkflowRegisterOutput { id: input.id })
}
fn read_kind(value: &serde_json::Value, label: &str) -> WireResult<String> {
value
.get("kind")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or_else(|| {
crate::domain::error::WireError::InvalidSpec(format!(
"{label}.kind (string) is required"
))
})
}
fn require_string_field(value: &serde_json::Value, field: &str, label: &str) -> WireResult<String> {
value
.get(field)
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or_else(|| {
crate::domain::error::WireError::InvalidSpec(format!("{label} (string) is required"))
})
}
#[derive(Debug)]
pub struct WireWorkflowListInput {
pub persona_id: Option<String>,
pub trigger_kind: Option<String>,
pub enabled_only: Option<bool>,
}
#[derive(Debug)]
pub struct WorkflowSummary {
pub id: String,
pub persona_id: Option<String>,
pub trigger: serde_json::Value,
pub action: serde_json::Value,
pub enabled: bool,
}
#[derive(Debug)]
pub struct WireWorkflowListOutput {
pub workflows: Vec<WorkflowSummary>,
}
pub fn wire_workflow_list(
input: WireWorkflowListInput,
storage: &SqliteStorage,
) -> WireResult<WireWorkflowListOutput> {
let spec = Specification::TypeIs(WORKFLOW_TYPE.to_string());
let nodes = collect_matching_nodes(storage, &spec)?;
let enabled_only = input.enabled_only.unwrap_or(true);
let workflows = nodes
.into_iter()
.filter_map(|n| node_to_summary(n).ok())
.filter(|w| {
if enabled_only && !w.enabled {
return false;
}
if let Some(p) = input.persona_id.as_ref() {
if w.persona_id.as_deref() != Some(p.as_str()) {
return false;
}
}
if let Some(tk) = input.trigger_kind.as_ref() {
if w.trigger.get("kind").and_then(|v| v.as_str()) != Some(tk.as_str()) {
return false;
}
}
true
})
.collect();
Ok(WireWorkflowListOutput { workflows })
}
fn node_to_summary(node: Node) -> WireResult<WorkflowSummary> {
let meta = node.metadata;
let persona_id = meta
.get("persona")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let trigger = meta
.get("trigger")
.cloned()
.unwrap_or(serde_json::Value::Null);
let action = meta
.get("action")
.cloned()
.unwrap_or(serde_json::Value::Null);
let enabled = meta
.get("enabled")
.and_then(|v| v.as_bool())
.unwrap_or(true);
Ok(WorkflowSummary {
id: node.id,
persona_id,
trigger,
action,
enabled,
})
}
#[derive(Debug)]
pub struct WireWorkflowFireInput {
pub id: Option<String>,
pub event: Option<String>,
pub persona_id: Option<String>,
pub dry_run: Option<bool>,
}
#[derive(Debug)]
pub struct ResolvedFire {
pub id: String,
pub persona_id: Option<String>,
pub action_kind: String,
pub action_emit_projection_names: Option<Vec<String>>,
pub dry_run: bool,
}
#[derive(Debug)]
pub struct WireWorkflowFireOutput {
pub fired: Vec<ResolvedFire>,
pub skipped: Vec<(String, String)>, }
pub fn wire_workflow_fire(
input: WireWorkflowFireInput,
storage: &SqliteStorage,
) -> WireResult<WireWorkflowFireOutput> {
if input.id.is_some() == input.event.is_some() {
return Err(crate::domain::error::WireError::InvalidSpec(
"exactly one of `id` or `event` is required".to_string(),
));
}
let dry_run = input.dry_run.unwrap_or(false);
let candidates: Vec<WorkflowSummary> = if let Some(id) = input.id.as_ref() {
let Some(node) = storage.get_node(id)? else {
return Ok(WireWorkflowFireOutput {
fired: vec![],
skipped: vec![(id.clone(), "workflow not found".to_string())],
});
};
if node.r#type != WORKFLOW_TYPE {
return Ok(WireWorkflowFireOutput {
fired: vec![],
skipped: vec![(
id.clone(),
format!("node type is '{}', expected '{WORKFLOW_TYPE}'", node.r#type),
)],
});
}
vec![node_to_summary(node)?]
} else {
let spec = Specification::TypeIs(WORKFLOW_TYPE.to_string());
collect_matching_nodes(storage, &spec)?
.into_iter()
.filter_map(|n| node_to_summary(n).ok())
.collect()
};
let mut fired = Vec::new();
let mut skipped = Vec::new();
let event = input.event.as_deref();
for w in candidates {
if !w.enabled {
skipped.push((w.id.clone(), "enabled=false".to_string()));
continue;
}
if let Some(persona_filter) = input.persona_id.as_ref() {
if w.persona_id.as_deref() != Some(persona_filter.as_str()) {
skipped.push((
w.id.clone(),
format!("persona scope mismatch (want={persona_filter})"),
));
continue;
}
}
let trigger_kind = w.trigger.get("kind").and_then(|v| v.as_str()).unwrap_or("");
if let Some(ev) = event {
if trigger_kind != "on_event" {
skipped.push((
w.id.clone(),
format!("trigger.kind='{trigger_kind}' does not match event fan-out"),
));
continue;
}
let wf_event = w
.trigger
.get("event")
.and_then(|v| v.as_str())
.unwrap_or("");
if wf_event != ev {
skipped.push((
w.id.clone(),
format!("trigger.event='{wf_event}' != '{ev}'"),
));
continue;
}
}
let action_kind = w
.action
.get("kind")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let action_emit_projection_names = if action_kind == "emit_projection" {
w.action
.get("projection_names")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(String::from))
.collect()
})
} else {
None
};
fired.push(ResolvedFire {
id: w.id,
persona_id: w.persona_id,
action_kind,
action_emit_projection_names,
dry_run,
});
}
Ok(WireWorkflowFireOutput { fired, skipped })
}
#[derive(Debug)]
pub struct WireWorkflowCheckInput {
pub persona_id: Option<String>,
pub include_exempt: Option<bool>,
pub include_covered: Option<bool>,
}
#[derive(Debug, Clone)]
pub struct UncoveredNode {
pub node_id: String,
pub r#type: String,
pub persona: Option<String>,
pub axis: Option<String>,
pub reasons: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct UndeclaredNode {
pub node_id: String,
pub r#type: String,
pub persona: Option<String>,
pub axis: Option<String>,
}
#[derive(Debug, Clone)]
pub struct ExemptNode {
pub node_id: String,
pub reason: Option<String>,
}
#[derive(Debug, Clone)]
pub struct CoveredNode {
pub node_id: String,
pub axis: Option<String>,
pub covering_workflow_id: String,
}
#[derive(Debug)]
pub struct WireWorkflowCheckOutput {
pub total_nodes: usize,
pub declared_covered_count: usize,
pub declared_covered: Vec<CoveredNode>, pub declared_uncovered: Vec<UncoveredNode>,
pub undeclared: Vec<UndeclaredNode>,
pub exempt: Vec<ExemptNode>, pub workflows_observed: usize,
}
pub fn wire_workflow_check(
input: WireWorkflowCheckInput,
storage: &SqliteStorage,
) -> WireResult<WireWorkflowCheckOutput> {
let include_exempt = input.include_exempt.unwrap_or(false);
let include_covered = input.include_covered.unwrap_or(false);
let wf_spec = Specification::TypeIs(WORKFLOW_TYPE.to_string());
let workflow_nodes = collect_matching_nodes(storage, &wf_spec)?;
let workflows: Vec<WorkflowSummary> = workflow_nodes
.into_iter()
.filter_map(|n| node_to_summary(n).ok())
.filter(|w| w.enabled)
.collect();
let workflows_observed = workflows.len();
let mut all_nodes: Vec<Node> = Vec::new();
for t in storage.list_types_by_kind("node")? {
if t == WORKFLOW_TYPE {
continue;
}
all_nodes.extend(collect_matching_nodes(storage, &Specification::TypeIs(t))?);
}
let candidate_nodes: Vec<Node> = all_nodes
.into_iter()
.filter(|n| {
if let Some(p) = input.persona_id.as_ref() {
n.metadata.get("persona").and_then(|v| v.as_str()) == Some(p.as_str())
} else {
true
}
})
.collect();
let total_nodes = candidate_nodes.len();
let mut declared_covered: Vec<CoveredNode> = Vec::new();
let mut declared_covered_count: usize = 0;
let mut declared_uncovered: Vec<UncoveredNode> = Vec::new();
let mut undeclared: Vec<UndeclaredNode> = Vec::new();
let mut exempt: Vec<ExemptNode> = Vec::new();
for n in candidate_nodes {
let persona = n
.metadata
.get("persona")
.and_then(|v| v.as_str())
.map(String::from);
let axis = n
.metadata
.get("axis")
.and_then(|v| v.as_str())
.map(String::from);
let node_id = n.id.clone();
let node_type = n.r#type.clone();
if n.metadata
.get("maintenance_exempt")
.and_then(|v| v.as_bool())
.unwrap_or(false)
{
let reason = n
.metadata
.get("maintenance_exempt_reason")
.and_then(|v| v.as_str())
.map(String::from);
exempt.push(ExemptNode { node_id, reason });
continue;
}
let maintained_by = n.metadata.get("maintained_by");
if let Some(mb) = maintained_by {
let declared_event = mb.get("event").and_then(|v| v.as_str()).map(String::from);
let declared_workflow_ref = mb
.get("workflow_ref")
.and_then(|v| v.as_str())
.map(String::from);
let mut covering: Option<&WorkflowSummary> = None;
let mut reasons: Vec<String> = Vec::new();
for w in &workflows {
if let Some(want_ref) = declared_workflow_ref.as_ref() {
if &w.id != want_ref {
continue;
}
} else if let Some(want_event) = declared_event.as_ref() {
let kind = w.trigger.get("kind").and_then(|v| v.as_str()).unwrap_or("");
if kind != "on_event" {
continue;
}
let ev = w
.trigger
.get("event")
.and_then(|v| v.as_str())
.unwrap_or("");
if ev != want_event {
continue;
}
} else {
continue;
}
let action_kind = w.action.get("kind").and_then(|v| v.as_str()).unwrap_or("");
if action_kind != "emit_projection" {
continue;
}
if w.persona_id != persona {
continue;
}
let axis_match = w
.action
.get("projection_names")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str())
.any(|a| Some(a) == axis.as_deref())
})
.unwrap_or(false);
if !axis_match {
continue;
}
covering = Some(w);
break;
}
if let Some(w) = covering {
declared_covered_count += 1;
if include_covered {
declared_covered.push(CoveredNode {
node_id,
axis,
covering_workflow_id: w.id.clone(),
});
}
continue;
}
if let Some(want_ref) = declared_workflow_ref.as_ref() {
if !workflows.iter().any(|w| &w.id == want_ref) {
reasons.push(format!(
"maintained_by.workflow_ref='{want_ref}' not found among enabled workflow_def"
));
} else {
reasons.push(format!(
"workflow '{want_ref}' exists but its action does not cover this Node \
(persona/axis/projection_names mismatch)"
));
}
} else if let Some(want_event) = declared_event.as_ref() {
reasons.push(format!(
"no enabled on_event workflow with trigger.event='{want_event}' covers \
persona='{persona}' axis='{axis}'",
persona = persona.as_deref().unwrap_or("<none>"),
axis = axis.as_deref().unwrap_or("<none>")
));
} else {
reasons.push(
"maintained_by present but neither event nor workflow_ref declared".to_string(),
);
}
declared_uncovered.push(UncoveredNode {
node_id,
r#type: node_type,
persona,
axis,
reasons,
});
continue;
}
undeclared.push(UndeclaredNode {
node_id,
r#type: node_type,
persona,
axis,
});
}
let exempt_out = if include_exempt { exempt } else { Vec::new() };
Ok(WireWorkflowCheckOutput {
total_nodes,
declared_covered_count,
declared_covered,
declared_uncovered,
undeclared,
exempt: exempt_out,
workflows_observed,
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::application::projection_registry::NamedProjection;
use crate::domain::graph::{Edge, Node};
use serde_json::json;
fn setup() -> SqliteStorage {
let s = SqliteStorage::open_in_memory().unwrap();
s.migrate().unwrap();
s.seed_default_types().unwrap();
s
}
fn default_registry() -> PluginRegistry {
PluginRegistry::default_for_wire().unwrap()
}
fn bare_node(id: &str, type_: &str) -> Node {
Node {
id: id.into(),
r#type: type_.into(),
sot_ref: None,
confidence: None,
applicability: None,
last_verified_at: None,
review_due: None,
version: 1,
prev_id: None,
metadata: json!({}),
}
}
#[test]
fn wire_init_with_no_projections_yields_empty() {
let s = setup();
let out = wire_init(
WireInitInput {
persona_id: "alpha".into(),
},
&s,
&default_registry(),
)
.unwrap();
assert_eq!(out.persona_id, "alpha");
assert!(out.projections.is_empty());
assert!(out.warnings.is_empty());
}
#[test]
fn wire_init_renders_registered_projection() {
let s = setup();
s.insert_node(&bare_node("alpha", "persona")).unwrap();
s.insert_node(&bare_node("beta", "persona")).unwrap();
SpecRegistry::new(&s)
.register("active_personas", &Specification::TypeIs("persona".into()))
.unwrap();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "_persona_toc".into(),
spec_ref: "active_personas".into(),
template: "Personas ({{count}}): {{names}}".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: None,
projection_config: None,
})
.unwrap();
let out = wire_init(
WireInitInput {
persona_id: "alpha".into(),
},
&s,
&default_registry(),
)
.unwrap();
assert_eq!(out.projections.len(), 1);
let p = &out.projections[0];
assert_eq!(p.name, "_persona_toc");
assert_eq!(p.target_form, TargetForm::Prompt);
assert!(p.rendered.contains("Personas (2):"));
assert!(p.rendered.contains("beta"));
assert!(p.rendered.contains("alpha"));
assert!(out.warnings.is_empty());
}
#[test]
fn wire_init_warns_on_unknown_spec_ref() {
let s = setup();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "broken".into(),
spec_ref: "no_such_spec".into(),
template: "x".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: None,
projection_config: None,
})
.unwrap();
let out = wire_init(
WireInitInput {
persona_id: "alpha".into(),
},
&s,
&default_registry(),
)
.unwrap();
assert!(out.projections.is_empty());
assert_eq!(out.warnings.len(), 1);
assert!(out.warnings[0].contains("no_such_spec"));
}
#[test]
fn wire_close_reports_orphans_and_totals() {
let s = setup();
for id in ["a", "b", "c"] {
s.insert_node(&bare_node(id, "persona")).unwrap();
}
s.insert_edge(&Edge {
id: "e1".into(),
src_node: "a".into(),
tgt_node: "b".into(),
kind: "routes_to".into(),
severity: None,
metadata: json!({}),
version: 1,
prev_id: None,
})
.unwrap();
let out = wire_close(
WireCloseInput {
persona_id: "alpha".into(),
},
&s,
)
.unwrap();
assert_eq!(out.total_node_count, 3);
assert_eq!(out.total_edge_count, 1);
assert_eq!(out.orphan_node_count, 1);
assert!(out
.report_markdown
.contains("orphan nodes (0 in + 0 out): 1"));
assert!(out.report_markdown.contains("total nodes: 3"));
}
#[test]
fn wire_close_empty_graph_zero_everything() {
let s = setup();
let out = wire_close(
WireCloseInput {
persona_id: "alpha".into(),
},
&s,
)
.unwrap();
assert_eq!(out.total_node_count, 0);
assert_eq!(out.total_edge_count, 0);
assert_eq!(out.orphan_node_count, 0);
}
#[test]
fn wire_node_delete_returns_true_when_row_exists() {
let s = setup();
s.insert_node(&bare_node("a", "persona")).unwrap();
let out = wire_node_delete(
WireDeleteInput {
id_or_name: "a".into(),
},
&s,
)
.unwrap();
assert_eq!(out.kind, "node");
assert_eq!(out.id_or_name, "a");
assert!(out.deleted);
let out2 = wire_node_delete(
WireDeleteInput {
id_or_name: "a".into(),
},
&s,
)
.unwrap();
assert!(!out2.deleted);
}
#[test]
fn wire_node_delete_returns_false_when_row_missing() {
let s = setup();
let out = wire_node_delete(
WireDeleteInput {
id_or_name: "ghost".into(),
},
&s,
)
.unwrap();
assert!(!out.deleted);
}
#[test]
fn wire_edge_delete_returns_true_when_row_exists() {
let s = setup();
s.insert_node(&bare_node("a", "persona")).unwrap();
s.insert_node(&bare_node("b", "persona")).unwrap();
s.insert_edge(&Edge {
id: "e1".into(),
src_node: "a".into(),
tgt_node: "b".into(),
kind: "routes_to".into(),
severity: None,
metadata: json!({}),
version: 1,
prev_id: None,
})
.unwrap();
let out = wire_edge_delete(
WireDeleteInput {
id_or_name: "e1".into(),
},
&s,
)
.unwrap();
assert_eq!(out.kind, "edge");
assert!(out.deleted);
}
#[test]
fn wire_spec_delete_returns_true_when_row_exists() {
let s = setup();
SpecRegistry::new(&s)
.register("active_personas", &Specification::TypeIs("persona".into()))
.unwrap();
let out = wire_spec_delete(
WireDeleteInput {
id_or_name: "active_personas".into(),
},
&s,
)
.unwrap();
assert_eq!(out.kind, "spec");
assert!(out.deleted);
}
#[test]
fn wire_projection_delete_returns_true_when_row_exists() {
let s = setup();
SpecRegistry::new(&s)
.register("p", &Specification::TypeIs("persona".into()))
.unwrap();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "doomed".into(),
spec_ref: "p".into(),
template: "x".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: None,
projection_config: None,
})
.unwrap();
let out = wire_projection_delete(
WireDeleteInput {
id_or_name: "doomed".into(),
},
&s,
)
.unwrap();
assert_eq!(out.kind, "projection");
assert!(out.deleted);
assert!(ProjectionRegistry::new(&s).list().unwrap().is_empty());
}
#[test]
fn workflow_register_round_trips_via_list() {
let s = setup();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "alpha.workflow.review_close".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_event","event":"session_close"}),
action: json!({"kind":"emit_projection","projection_names":["review_pending"]}),
enabled: None,
},
&s,
)
.unwrap();
let out = wire_workflow_list(
WireWorkflowListInput {
persona_id: Some("alpha".into()),
trigger_kind: None,
enabled_only: None,
},
&s,
)
.unwrap();
assert_eq!(out.workflows.len(), 1);
let w = &out.workflows[0];
assert_eq!(w.id, "alpha.workflow.review_close");
assert_eq!(w.persona_id.as_deref(), Some("alpha"));
assert!(w.enabled);
assert_eq!(w.trigger["kind"], "on_event");
assert_eq!(w.action["kind"], "emit_projection");
}
#[test]
fn workflow_register_rejects_unsupported_trigger_kind() {
let s = setup();
let err = wire_workflow_register(
WireWorkflowRegisterInput {
id: "x".into(),
persona_id: None,
trigger: json!({"kind":"cron","cron_spec":"0 9 * * *"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap_err();
assert!(err.to_string().contains("cron"));
}
#[test]
fn workflow_register_rejects_on_event_without_event_field() {
let s = setup();
let err = wire_workflow_register(
WireWorkflowRegisterInput {
id: "x".into(),
persona_id: None,
trigger: json!({"kind":"on_event"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap_err();
assert!(err.to_string().contains("event"));
}
#[test]
fn workflow_register_rejects_emit_projection_without_names() {
let s = setup();
let err = wire_workflow_register(
WireWorkflowRegisterInput {
id: "x".into(),
persona_id: None,
trigger: json!({"kind":"on_demand"}),
action: json!({"kind":"emit_projection"}),
enabled: None,
},
&s,
)
.unwrap_err();
assert!(err.to_string().contains("projection_names"));
}
#[test]
fn workflow_list_filters_by_trigger_kind_and_enabled() {
let s = setup();
for (id, kind, enabled) in [
("w1", "on_demand", true),
("w2", "on_event", true),
("w3", "on_demand", false),
] {
let trig = if kind == "on_event" {
json!({"kind":"on_event","event":"e"})
} else {
json!({"kind":"on_demand"})
};
wire_workflow_register(
WireWorkflowRegisterInput {
id: id.into(),
persona_id: None,
trigger: trig,
action: json!({"kind":"no_op"}),
enabled: Some(enabled),
},
&s,
)
.unwrap();
}
let out = wire_workflow_list(
WireWorkflowListInput {
persona_id: None,
trigger_kind: Some("on_demand".into()),
enabled_only: None,
},
&s,
)
.unwrap();
let ids: Vec<&str> = out.workflows.iter().map(|w| w.id.as_str()).collect();
assert_eq!(ids, vec!["w1"]);
let out2 = wire_workflow_list(
WireWorkflowListInput {
persona_id: None,
trigger_kind: Some("on_demand".into()),
enabled_only: Some(false),
},
&s,
)
.unwrap();
let mut ids2: Vec<&str> = out2.workflows.iter().map(|w| w.id.as_str()).collect();
ids2.sort();
assert_eq!(ids2, vec!["w1", "w3"]);
}
#[test]
fn workflow_fire_by_id_returns_resolved_emit_projection() {
let s = setup();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "w1".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_demand"}),
action: json!({"kind":"emit_projection","projection_names":["axis_a","axis_b"]}),
enabled: None,
},
&s,
)
.unwrap();
let out = wire_workflow_fire(
WireWorkflowFireInput {
id: Some("w1".into()),
event: None,
persona_id: None,
dry_run: None,
},
&s,
)
.unwrap();
assert_eq!(out.fired.len(), 1);
assert!(out.skipped.is_empty());
let f = &out.fired[0];
assert_eq!(f.id, "w1");
assert_eq!(f.action_kind, "emit_projection");
assert_eq!(
f.action_emit_projection_names.as_deref(),
Some(&["axis_a".to_string(), "axis_b".to_string()][..])
);
}
#[test]
fn workflow_fire_by_event_skips_unrelated_and_disabled() {
let s = setup();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "match_open".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_event","event":"session_open"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "match_close".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_event","event":"session_close"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "disabled_close".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_event","event":"session_close"}),
action: json!({"kind":"no_op"}),
enabled: Some(false),
},
&s,
)
.unwrap();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "demand_only".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_demand"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap();
let out = wire_workflow_fire(
WireWorkflowFireInput {
id: None,
event: Some("session_close".into()),
persona_id: Some("alpha".into()),
dry_run: None,
},
&s,
)
.unwrap();
let fired_ids: Vec<&str> = out.fired.iter().map(|f| f.id.as_str()).collect();
assert_eq!(fired_ids, vec!["match_close"]);
assert_eq!(out.skipped.len(), 3);
}
#[test]
fn workflow_fire_requires_exactly_one_of_id_or_event() {
let s = setup();
let err = wire_workflow_fire(
WireWorkflowFireInput {
id: None,
event: None,
persona_id: None,
dry_run: None,
},
&s,
)
.unwrap_err();
assert!(err.to_string().contains("id"));
}
#[test]
fn workflow_fire_by_id_handles_missing() {
let s = setup();
let out = wire_workflow_fire(
WireWorkflowFireInput {
id: Some("ghost".into()),
event: None,
persona_id: None,
dry_run: None,
},
&s,
)
.unwrap();
assert!(out.fired.is_empty());
assert_eq!(out.skipped.len(), 1);
assert_eq!(out.skipped[0].0, "ghost");
}
#[test]
fn workflow_delete_uses_node_delete() {
let s = setup();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "w1".into(),
persona_id: None,
trigger: json!({"kind":"on_demand"}),
action: json!({"kind":"no_op"}),
enabled: None,
},
&s,
)
.unwrap();
let out = wire_node_delete(
WireDeleteInput {
id_or_name: "w1".into(),
},
&s,
)
.unwrap();
assert!(out.deleted);
assert!(wire_workflow_list(
WireWorkflowListInput {
persona_id: None,
trigger_kind: None,
enabled_only: Some(false),
},
&s,
)
.unwrap()
.workflows
.is_empty());
}
fn make_outline_node(id: &str, persona: &str, axis: &str, extra: serde_json::Value) -> Node {
let mut meta = serde_json::json!({
"persona": persona,
"axis": axis,
"source_uri": format!("file:~/test/{id}")
});
if let serde_json::Value::Object(ref mut m) = meta {
if let serde_json::Value::Object(extra_map) = extra {
for (k, v) in extra_map {
m.insert(k, v);
}
}
}
Node {
id: id.into(),
r#type: "outline_node".into(),
sot_ref: None,
confidence: None,
applicability: None,
last_verified_at: None,
review_due: None,
version: 1,
prev_id: None,
metadata: meta,
}
}
fn register_emit_workflow(
s: &SqliteStorage,
id: &str,
persona: &str,
event: &str,
axes: &[&str],
) {
let axes_json =
serde_json::Value::Array(axes.iter().map(|a| serde_json::json!(a)).collect());
wire_workflow_register(
WireWorkflowRegisterInput {
id: id.into(),
persona_id: Some(persona.into()),
trigger: json!({"kind":"on_event","event":event}),
action: json!({"kind":"emit_projection","projection_names": axes_json}),
enabled: None,
},
s,
)
.unwrap();
}
#[test]
fn workflow_check_buckets_a_node_as_declared_covered_when_workflow_matches() {
let s = setup();
register_emit_workflow(
&s,
"alpha.workflow.close",
"alpha",
"session_close",
&["handoff"],
);
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"event": "session_close"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: Some(true),
},
&s,
)
.unwrap();
assert_eq!(out.total_nodes, 1);
assert_eq!(out.declared_covered_count, 1);
assert_eq!(out.declared_covered.len(), 1);
assert_eq!(
out.declared_covered[0].covering_workflow_id,
"alpha.workflow.close"
);
assert!(out.declared_uncovered.is_empty());
assert!(out.undeclared.is_empty());
}
#[test]
fn workflow_check_flags_declared_uncovered_when_event_has_no_matching_workflow() {
let s = setup();
register_emit_workflow(
&s,
"alpha.workflow.open",
"alpha",
"session_open",
&["handoff"],
);
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"event": "session_close"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.declared_uncovered.len(), 1);
assert!(out.declared_uncovered[0].reasons[0].contains("session_close"));
}
#[test]
fn workflow_check_flags_declared_uncovered_when_workflow_ref_missing() {
let s = setup();
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"workflow_ref": "alpha.workflow.ghost"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.declared_uncovered.len(), 1);
assert!(out.declared_uncovered[0].reasons[0].contains("ghost"));
assert!(out.declared_uncovered[0].reasons[0].contains("not found"));
}
#[test]
fn workflow_check_flags_declared_uncovered_when_axis_not_in_projection_names() {
let s = setup();
register_emit_workflow(
&s,
"alpha.workflow.close",
"alpha",
"session_close",
&["active"],
);
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"event": "session_close"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.declared_uncovered.len(), 1);
}
#[test]
fn workflow_check_classifies_node_without_maintained_by_as_undeclared() {
let s = setup();
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.undeclared.len(), 1);
assert_eq!(out.undeclared[0].node_id, "alpha.handoff");
assert!(out.declared_covered.is_empty());
assert!(out.declared_uncovered.is_empty());
}
#[test]
fn workflow_check_classifies_exempt_node_and_returns_only_when_include_exempt() {
let s = setup();
s.insert_node(&make_outline_node(
"alpha.static_sot",
"alpha",
"static_sot",
json!({"maintenance_exempt": true, "maintenance_exempt_reason": "external static SoT"}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert!(out.exempt.is_empty());
assert!(out.undeclared.is_empty());
let out2 = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: Some(true),
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out2.exempt.len(), 1);
assert_eq!(
out2.exempt[0].reason.as_deref(),
Some("external static SoT")
);
}
#[test]
fn workflow_check_skips_disabled_workflows_when_resolving_coverage() {
let s = setup();
wire_workflow_register(
WireWorkflowRegisterInput {
id: "alpha.workflow.close".into(),
persona_id: Some("alpha".into()),
trigger: json!({"kind":"on_event","event":"session_close"}),
action: json!({"kind":"emit_projection","projection_names":["handoff"]}),
enabled: Some(false),
},
&s,
)
.unwrap();
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"event": "session_close"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.workflows_observed, 0);
assert_eq!(out.declared_uncovered.len(), 1);
}
#[test]
fn workflow_check_persona_scope_filters_other_personas() {
let s = setup();
s.insert_node(&make_outline_node("alpha.h", "alpha", "handoff", json!({})))
.unwrap();
s.insert_node(&make_outline_node("beta.h", "beta", "handoff", json!({})))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: Some("alpha".into()),
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.total_nodes, 1);
assert_eq!(out.undeclared.len(), 1);
assert_eq!(out.undeclared[0].node_id, "alpha.h");
}
#[test]
fn workflow_check_excludes_workflow_def_nodes_from_total() {
let s = setup();
register_emit_workflow(
&s,
"alpha.workflow.x",
"alpha",
"session_close",
&["handoff"],
);
s.insert_node(&make_outline_node(
"alpha.handoff",
"alpha",
"handoff",
json!({"maintained_by": {"event": "session_close"}}),
))
.unwrap();
let out = wire_workflow_check(
WireWorkflowCheckInput {
persona_id: None,
include_exempt: None,
include_covered: None,
},
&s,
)
.unwrap();
assert_eq!(out.total_nodes, 1);
assert_eq!(out.workflows_observed, 1);
}
#[test]
fn wire_node_delete_cascades_to_referencing_edges() {
let s = setup();
s.insert_node(&bare_node("a", "persona")).unwrap();
s.insert_node(&bare_node("b", "persona")).unwrap();
s.insert_node(&bare_node("c", "persona")).unwrap();
s.insert_edge(&Edge {
id: "e_ab".into(),
src_node: "a".into(),
tgt_node: "b".into(),
kind: "routes_to".into(),
severity: None,
metadata: json!({}),
version: 1,
prev_id: None,
})
.unwrap();
s.insert_edge(&Edge {
id: "e_ca".into(),
src_node: "c".into(),
tgt_node: "a".into(),
kind: "routes_to".into(),
severity: None,
metadata: json!({}),
version: 1,
prev_id: None,
})
.unwrap();
s.insert_edge(&Edge {
id: "e_bc".into(),
src_node: "b".into(),
tgt_node: "c".into(),
kind: "routes_to".into(),
severity: None,
metadata: json!({}),
version: 1,
prev_id: None,
})
.unwrap();
wire_node_delete(
WireDeleteInput {
id_or_name: "a".into(),
},
&s,
)
.unwrap();
assert!(s.get_edge(&"e_ab".to_string()).unwrap().is_none());
assert!(s.get_edge(&"e_ca".to_string()).unwrap().is_none());
assert!(s.get_edge(&"e_bc".to_string()).unwrap().is_some());
}
#[test]
fn wire_init_rejects_non_static_projection_kind() {
let s = setup();
SpecRegistry::new(&s)
.register("p", &Specification::TypeIs("persona".into()))
.unwrap();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "async_only".into(),
spec_ref: "p".into(),
template: "x".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: Some("llm".into()),
projection_config: None,
})
.unwrap();
let result = wire_init(
WireInitInput {
persona_id: "alpha".into(),
},
&s,
&default_registry(),
);
let err = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected non-static projection_kind to fail"),
};
assert!(err.contains("async_only"), "err: {err}");
assert!(err.contains("llm"), "err: {err}");
assert!(err.contains("wire_prompt_context"), "err: {err}");
}
#[test]
fn wire_render_rejects_non_static_projection_kind() {
let s = setup();
SpecRegistry::new(&s)
.register("p", &Specification::TypeIs("persona".into()))
.unwrap();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "summarized".into(),
spec_ref: "p".into(),
template: "x".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: Some("cache".into()),
projection_config: None,
})
.unwrap();
let result = wire_render(
WireRenderInput {
projection_ref: "summarized".into(),
},
&s,
&default_registry(),
);
let err = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected non-static projection_kind to fail"),
};
assert!(err.contains("summarized"), "err: {err}");
assert!(err.contains("cache"), "err: {err}");
assert!(err.contains("wire_prompt_context"), "err: {err}");
}
#[test]
fn wire_init_accepts_explicit_static_projection_kind() {
let s = setup();
s.insert_node(&bare_node("alpha", "persona")).unwrap();
SpecRegistry::new(&s)
.register("p", &Specification::TypeIs("persona".into()))
.unwrap();
ProjectionRegistry::new(&s)
.register(&NamedProjection {
name: "explicit_static".into(),
spec_ref: "p".into(),
template: "n={{count}}".into(),
target_form: TargetForm::Prompt,
template_engine: None,
projection_kind: Some("static".into()),
projection_config: None,
})
.unwrap();
let out = wire_init(
WireInitInput {
persona_id: "alpha".into(),
},
&s,
&default_registry(),
)
.unwrap();
assert_eq!(out.projections.len(), 1);
assert_eq!(out.projections[0].rendered, "n=1");
}
fn seed_wiring_node(s: &SqliteStorage, id: &str, source_uri: &str) {
s.insert_node(&Node {
id: id.into(),
r#type: "outline_node".into(),
sot_ref: None,
confidence: Some(1.0),
applicability: None,
last_verified_at: None,
review_due: None,
version: 1,
prev_id: None,
metadata: json!({
"persona": "shi",
"axis": "mailbox",
"source_uri": source_uri,
}),
})
.unwrap();
}
#[test]
fn node_update_merge_overwrites_one_key_preserves_others() {
let s = setup();
seed_wiring_node(&s, "shi.mailbox", "mini-app://mailbox?alias=for_shi");
let out = wire_node_update(
WireNodeUpdateInput {
id: "shi.mailbox".into(),
metadata_patch: json!({
"source_uri": "mini-app://mailbox?alias=for_shi&limit=10",
}),
mode: WireNodeUpdateMode::Merge,
},
&s,
)
.unwrap();
assert_eq!(out.id, "shi.mailbox");
assert_eq!(out.mode, WireNodeUpdateMode::Merge);
assert_eq!(
out.metadata["source_uri"].as_str().unwrap(),
"mini-app://mailbox?alias=for_shi&limit=10"
);
assert_eq!(out.metadata["persona"].as_str().unwrap(), "shi");
assert_eq!(out.metadata["axis"].as_str().unwrap(), "mailbox");
let stored = s.get_node(&"shi.mailbox".to_string()).unwrap().unwrap();
assert_eq!(
stored.metadata["source_uri"].as_str().unwrap(),
"mini-app://mailbox?alias=for_shi&limit=10"
);
}
#[test]
fn node_update_merge_null_value_deletes_key() {
let s = setup();
seed_wiring_node(&s, "shi.tmp", "mini-app://x");
let out = wire_node_update(
WireNodeUpdateInput {
id: "shi.tmp".into(),
metadata_patch: json!({"axis": null}),
mode: WireNodeUpdateMode::Merge,
},
&s,
)
.unwrap();
assert!(out.metadata.get("axis").is_none());
assert_eq!(out.metadata["persona"].as_str().unwrap(), "shi");
assert_eq!(out.metadata["source_uri"].as_str().unwrap(), "mini-app://x");
}
#[test]
fn node_update_replace_swaps_metadata_wholesale() {
let s = setup();
seed_wiring_node(&s, "shi.tmp", "mini-app://x");
let out = wire_node_update(
WireNodeUpdateInput {
id: "shi.tmp".into(),
metadata_patch: json!({"only_field": 42}),
mode: WireNodeUpdateMode::Replace,
},
&s,
)
.unwrap();
assert_eq!(out.metadata, json!({"only_field": 42}));
assert!(out.metadata.get("persona").is_none());
}
#[test]
fn node_update_unknown_id_returns_not_found() {
let s = setup();
let result = wire_node_update(
WireNodeUpdateInput {
id: "does.not.exist".into(),
metadata_patch: json!({"x": 1}),
mode: WireNodeUpdateMode::Merge,
},
&s,
);
let err = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected NotFound"),
};
assert!(err.contains("does.not.exist"), "err: {err}");
}
#[test]
fn node_update_rejects_non_object_patch() {
let s = setup();
seed_wiring_node(&s, "shi.tmp", "mini-app://x");
let result = wire_node_update(
WireNodeUpdateInput {
id: "shi.tmp".into(),
metadata_patch: json!("not an object"),
mode: WireNodeUpdateMode::Merge,
},
&s,
);
let err = match result {
Err(e) => e.to_string(),
Ok(_) => panic!("expected non-object patch to fail"),
};
assert!(err.contains("must be a JSON object"), "err: {err}");
}
#[test]
fn node_update_mode_parse_rejects_unknown() {
assert_eq!(
WireNodeUpdateMode::parse("merge").unwrap(),
WireNodeUpdateMode::Merge
);
assert_eq!(
WireNodeUpdateMode::parse("replace").unwrap(),
WireNodeUpdateMode::Replace
);
assert!(WireNodeUpdateMode::parse("upsert").is_err());
}
}