From 553d5581d17ab21d226c69e45c5320c77a61ba13 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 10:42:14 +0700 Subject: [PATCH 1/6] feat(coordinator): route declared BReg action refusals Signed-off-by: Jeremi Joslin --- crates/registry-coordinator/src/authoring.rs | 17 +- .../registry-coordinator/src/breg_action.rs | 104 ++++++- crates/registry-coordinator/src/contracts.rs | 108 ++++++- crates/registry-coordinator/src/definition.rs | 269 ++++++++++++----- .../registry-coordinator/src/mapping_check.rs | 52 +++- crates/registry-coordinator/src/operations.rs | 20 +- crates/registry-coordinator/src/protocol.rs | 68 +++++ crates/registry-coordinator/src/scenarios.rs | 147 +++++++-- crates/registry-coordinator/src/store.rs | 6 +- .../src/store/recovery.rs | 2 +- crates/registry-coordinator/src/worker.rs | 90 +++++- .../tests/authored_project.rs | 147 +++++++++ .../tests/authored_scenarios.rs | 59 ++++ .../registry-coordinator/tests/authoring.rs | 25 ++ .../tests/breg_action_adapter.rs | 48 +++ .../tests/breg_action_native.rs | 32 +- .../tests/casework_adapter.rs | 1 + .../tests/catalog_scenarios.rs | 71 ++++- .../tests/fixtures/legacy-definition-v1.json | 1 + .../tests/postgres_state.rs | 281 +++++++++++++++++- .../tests/scenario_contracts.rs | 22 ++ .../tests/support/breg_service.rs | 15 +- products/coordinator/CHANGELOG.md | 4 + products/coordinator/PRODUCT-CONTRACT.md | 29 +- products/coordinator/README.md | 28 ++ .../governed-action-follow-up/README.md | 27 +- .../governed-action-follow-up/scenarios.yaml | 37 +++ .../governed-action-follow-up/workflow.yaml | 14 + .../generated/operations/README.md | 36 ++- ...e-breg-action.declared-refusal.schema.json | 18 ++ .../generated/project/project.schema.json | 51 +++- .../generated/scenarios/scenarios.schema.json | 25 ++ 32 files changed, 1691 insertions(+), 163 deletions(-) create mode 100644 crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json create mode 100644 products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json diff --git a/crates/registry-coordinator/src/authoring.rs b/crates/registry-coordinator/src/authoring.rs index 2e235cfef..a7005f59b 100644 --- a/crates/registry-coordinator/src/authoring.rs +++ b/crates/registry-coordinator/src/authoring.rs @@ -386,7 +386,7 @@ enum AuthoredStep { on_timeout: Option, }, /// Calls one operation through a connection, then continues with its - /// output available to later steps. + /// successful output available to later steps. Call { /// The connection the call goes through, declared under `connections`. connection: LocalId, @@ -397,6 +397,11 @@ enum AuthoredStep { input: AuthoredInput, /// The step that follows a completed call. next: LocalId, + /// The step that follows a product-declared, no-effect business + /// refusal. Only operations whose reference declares this capability + /// accept the member; other refusals still stop the run. + #[serde(default)] + on_declared_refusal: Option, }, /// Selects a label and continues with the step its case names. Choose { @@ -506,6 +511,7 @@ impl TryFrom for Step { operation, input, next, + on_declared_refusal, } => Self::Call { call: Call { operation: Operation::parse(&operation.0) @@ -514,6 +520,7 @@ impl TryFrom for Step { }, input: input.into(), next: next.into_string(), + on_declared_refusal: on_declared_refusal.map(LocalId::into_string), }, AuthoredStep::Choose { choose, cases } => Self::Choose { choose: choose.into(), @@ -791,13 +798,17 @@ fn operation_input_members() -> serde_json::Value { if let Some(required) = input.get("required") { fields["required"] = required.clone(); } - serde_json::json!({ + let mut rule = serde_json::json!({ "if": { "properties": {"operation": {"const": descriptor.id}}, "required": ["operation"] }, "then": {"properties": {"input": {"properties": {"fields": fields}}}} - }) + }); + if !descriptor.declared_refusal { + rule["then"]["not"] = serde_json::json!({"required": ["onDeclaredRefusal"]}); + } + rule }) .collect() } diff --git a/crates/registry-coordinator/src/breg_action.rs b/crates/registry-coordinator/src/breg_action.rs index 76a7390b1..88d459cdb 100644 --- a/crates/registry-coordinator/src/breg_action.rs +++ b/crates/registry-coordinator/src/breg_action.rs @@ -15,7 +15,7 @@ use serde_json::{Map, Value}; use crate::{ adapters::{refused, retryable, token_failure}, - protocol::{CallOutcome, CallRequest, Operation}, + protocol::{CallOutcome, CallRequest, DeclaredRefusal, Operation}, }; #[derive(Deserialize)] @@ -40,6 +40,22 @@ fn input(request: &CallRequest) -> Result<(ActionInput, BRegIdempotencyKey), Cal Ok((input, key)) } +fn declared_refusal( + mutation: bool, + status: Option, + problem: Option, + refusal_code: Option<&str>, + outcome_unknown: bool, +) -> Option { + (mutation + && !outcome_unknown + && status == Some(422) + && problem == Some(BRegProblemCode::ActionRefused)) + .then_some(refusal_code) + .flatten() + .and_then(DeclaredRefusal::new) +} + fn failure(error: BaseRegistryClientError, mutation: bool) -> CallOutcome { if mutation && error.problem_code() == Some(BRegProblemCode::IdempotencyExpired) { return CallOutcome::ReceiptExpired; @@ -49,6 +65,15 @@ fn failure(error: BaseRegistryClientError, mutation: bool) -> CallOutcome { code: "transport-uncertain".into(), }; } + if let Some(refusal) = declared_refusal( + mutation, + error.status(), + error.problem_code(), + error.refusal_code().map(|code| code.as_str()), + false, + ) { + return CallOutcome::DeclaredRefusal(refusal); + } match error { BaseRegistryClientError::Token(error) => token_failure(&error), BaseRegistryClientError::Transport { .. } => retryable("transport-unavailable"), @@ -165,3 +190,80 @@ pub async fn execute( Err(error) => failure(error, true), } } + +#[cfg(test)] +mod tests { + use super::declared_refusal; + use registry_breg_client::BRegProblemCode; + + #[test] + fn declared_refusal_classification_requires_every_typed_execution_guard() { + let classify = |mutation, status, problem, code, unknown| { + declared_refusal(mutation, status, problem, code, unknown) + .map(|refusal| refusal.refusal_code().to_owned()) + }; + assert_eq!( + classify( + true, + Some(422), + Some(BRegProblemCode::ActionRefused), + Some("not-eligible"), + false, + ), + Some("not-eligible".to_owned()) + ); + for refusal in [ + classify( + false, + Some(422), + Some(BRegProblemCode::ActionRefused), + Some("not-eligible"), + false, + ), + classify( + true, + Some(409), + Some(BRegProblemCode::ActionRefused), + Some("not-eligible"), + false, + ), + classify( + true, + Some(422), + Some(BRegProblemCode::PreconditionFailed), + Some("not-eligible"), + false, + ), + classify( + true, + Some(422), + Some(BRegProblemCode::ActionRefused), + None, + false, + ), + classify( + true, + Some(422), + Some(BRegProblemCode::ActionRefused), + Some("line\nbreak"), + false, + ), + classify( + true, + Some(422), + Some(BRegProblemCode::ActionRefused), + Some(&"x".repeat(129)), + false, + ), + classify( + true, + Some(422), + Some(BRegProblemCode::ActionRefused), + Some("not-eligible"), + true, + ), + ] { + assert_eq!(refusal, None); + } + } +} diff --git a/crates/registry-coordinator/src/contracts.rs b/crates/registry-coordinator/src/contracts.rs index a7ae896e4..cc8c100aa 100644 --- a/crates/registry-coordinator/src/contracts.rs +++ b/crates/registry-coordinator/src/contracts.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: Apache-2.0 //! Operation contracts: the JSON Schema of the input each catalog operation's -//! adapter decodes and of the output it returns. The committed generated files -//! are embedded, so the default build derives no schemas. +//! adapter decodes, the success output it returns, and any declared-refusal +//! output it exposes to an authored branch. The committed generated files are +//! embedded, so the default build derives no schemas. use std::{collections::BTreeMap, sync::OnceLock}; @@ -13,12 +14,15 @@ use serde_json::Value; use crate::protocol::Operation; -/// One operation's input and output schemas, as committed. +/// One operation's input and route-specific output schemas, as committed. pub struct OperationContract { pub input: Value, pub output: Value, + /// Output made available only on an authored `onDeclaredRefusal` route. + pub declared_refusal: Option, input_validator: JSONSchema, output_validator: JSONSchema, + declared_refusal_validator: Option, } /// Where a value leaves a contract. The path is an RFC 6901 pointer into the @@ -56,6 +60,17 @@ impl OperationContract { pub fn check_output(&self, value: &Value) -> Result<(), Violation> { first_violation(&self.output_validator, value) } + + /// Checks a product-declared refusal before it becomes a step output. + pub fn check_declared_refusal(&self, value: &Value) -> Result<(), Violation> { + let Some(validator) = &self.declared_refusal_validator else { + return Err(Violation { + path: String::new(), + expected: "an operation that declares a routable business refusal".into(), + }); + }; + first_violation(validator, value) + } } pub(crate) fn first_violation(validator: &JSONSchema, value: &Value) -> Result<(), Violation> { @@ -165,6 +180,13 @@ static EMBEDDED: &[(&str, &str, &str)] = embedded!( "submit-review-request", ); +static DECLARED_REFUSALS: &[(&str, &str)] = &[( + "invoke-breg-action", + include_str!( + "../../../products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json" + ), +)]; + /// The committed contract of a wait event: the value a wait-for step that /// continues with `next` holds as its output. pub struct EventContract { @@ -222,6 +244,11 @@ pub fn contract(operation: Operation) -> &'static OperationContract { }; let (input, input_validator) = parse(input); let (output, output_validator) = parse(output); + let (declared_refusal, declared_refusal_validator) = DECLARED_REFUSALS + .iter() + .find(|(operation, _)| operation == id) + .map(|(_, schema)| parse(schema)) + .unzip(); ( *id, OperationContract { @@ -229,6 +256,8 @@ pub fn contract(operation: Operation) -> &'static OperationContract { output, input_validator, output_validator, + declared_refusal, + declared_refusal_validator, }, ) }) @@ -342,6 +371,14 @@ pub fn generate() -> Result, serde_json::Error> { let mut text = serde_json::to_string_pretty(&event)?; text.push('\n'); files.insert("review-result.event.schema.json".into(), text); + let mut refusal = derived::output::(); + derived::finish(&mut refusal, "invoke-breg-action declared-refusal output"); + let mut text = serde_json::to_string_pretty(&refusal)?; + text.push('\n'); + files.insert( + "invoke-breg-action.declared-refusal.schema.json".into(), + text, + ); files.insert("README.md".into(), derived::reference(&files)?); Ok(files) } @@ -416,16 +453,18 @@ mod derived { let mut text = String::from( "\n\ # Coordinator operation reference\n\n\ - Each catalog operation has an input contract, the type its adapter decodes, and an \ - output contract, the type its adapter returns as the step output. A mapping that \ + Each catalog operation has an input contract, the type its adapter decodes, and a \ + success output contract, the type its adapter returns on `next`. A mapping that \ feeds a call must produce a value its input contract accepts; a reference into a \ - call step's output must name a path its output contract declares.\n\n\ - | Operation | Version | Product | Effect | Key | Preparation | Recovery | Receipt |\n\ - |---|---|---|---|---|---|---|---|\n", + call step's output must name a path the contract for that route declares. Only the \ + operations marked below accept `onDeclaredRefusal`; that route receives the closed \ + declared-refusal output and no product detail or call input.\n\n\ + | Operation | Version | Product | Effect | Key | Preparation | Recovery | Receipt | Declared refusal |\n\ + |---|---|---|---|---|---|---|---|---|\n", ); for descriptor in crate::operations::descriptors() { text.push_str(&format!( - "| [`{id}`](#{id}) | {version} | `{product}` | {effect} | {key} | {preparation} | {recovery} | {receipt} |\n", + "| [`{id}`](#{id}) | {version} | `{product}` | {effect} | {key} | {preparation} | {recovery} | {receipt} | {declared_refusal} |\n", id = descriptor.id, version = descriptor.version, product = descriptor.product, @@ -434,6 +473,11 @@ mod derived { preparation = if descriptor.requires_preparation { "required" } else { "none" }, recovery = word(&descriptor.recovery)?, receipt = if descriptor.read_receipt { "read" } else { "none" }, + declared_refusal = if descriptor.declared_refusal { + "[`onDeclaredRefusal`](invoke-breg-action.declared-refusal.schema.json)" + } else { + "none" + }, )); } for descriptor in crate::operations::descriptors() { @@ -447,6 +491,14 @@ mod derived { )); text.push_str(&members(&schema)); } + if descriptor.declared_refusal { + let file = format!("{}.declared-refusal.schema.json", descriptor.id); + let schema: Value = serde_json::from_str(&files[&file])?; + text.push_str(&format!( + "\nDeclared-refusal output ([`{file}`]({file})):\n\n" + )); + text.push_str(&members(&schema)); + } } text.push_str( "\n## Wait events\n\n\ @@ -559,7 +611,10 @@ mod tests { for descriptor in crate::operations::descriptors() { let operation = Operation::parse(descriptor.id).unwrap(); let contract = contract(operation); - for schema in [&contract.input, &contract.output] { + for schema in [&contract.input, &contract.output] + .into_iter() + .chain(contract.declared_refusal.iter()) + { assert!(schema.get("$id").is_none()); crate::definition::compile_schema(schema).unwrap_or_else(|error| { panic!("{} contract does not compile: {error}", descriptor.id) @@ -631,6 +686,33 @@ mod tests { ); } + #[test] + fn only_the_breg_action_contract_exposes_a_closed_declared_refusal() { + for descriptor in crate::operations::descriptors() { + let operation = Operation::parse(descriptor.id).unwrap(); + let contract = contract(operation); + assert_eq!( + contract.declared_refusal.is_some(), + descriptor.declared_refusal, + "{}", + descriptor.id + ); + } + let action = contract(Operation::InvokeBregAction); + assert_eq!( + action.check_declared_refusal(&json!({"refusalCode":"not-eligible"})), + Ok(()) + ); + assert!(action + .check_declared_refusal(&json!({ + "refusalCode":"not-eligible", "detail":"private-canary" + })) + .is_err()); + assert!(contract(Operation::ReadRecord) + .check_declared_refusal(&json!({"refusalCode":"not-eligible"})) + .is_err()); + } + #[test] fn the_declared_path_leaves_out_an_unexpected_member_name() { let read = contract(Operation::ReadRecord); @@ -666,5 +748,11 @@ mod tests { for (id, event) in EVENTS { assert_eq!(generated[&format!("{id}.event.schema.json")], *event); } + for (id, refusal) in DECLARED_REFUSALS { + assert_eq!( + generated[&format!("{id}.declared-refusal.schema.json")], + *refusal + ); + } } } diff --git a/crates/registry-coordinator/src/definition.rs b/crates/registry-coordinator/src/definition.rs index e56b5bcc6..49a122f44 100644 --- a/crates/registry-coordinator/src/definition.rs +++ b/crates/registry-coordinator/src/definition.rs @@ -25,7 +25,8 @@ use crate::{ /// The compiled snapshot format, carried by its workflow. pub(crate) const API_VERSION: &str = "id.registrystack.org/formats/coordinator/definition-snapshot/v1alpha1"; -const ADAPTER_ABI: &str = "coordinator/product-operations/v1"; +const ADAPTER_ABI_V1: &str = "coordinator/product-operations/v1"; +const ADAPTER_ABI_V2: &str = "coordinator/product-operations/v2"; const SCHEMA_ABI: &str = "coordinator/jsonschema-0.18/draft202012/formats-asserted+uuid/v1"; const MAX_DOCUMENT_BYTES: usize = 1_048_576; /// Maximum canonical snapshot size, shared by authoring, restore and packaging. @@ -189,6 +190,12 @@ pub enum Step { call: Call, input: Mapping, next: String, + #[serde( + rename = "onDeclaredRefusal", + default, + skip_serializing_if = "Option::is_none" + )] + on_declared_refusal: Option, }, Choose { choose: Mapping, @@ -205,7 +212,18 @@ impl Step { /// Each transition with the authored member that holds it. pub(crate) fn edges(&self) -> Vec<(&'static str, &str)> { match self { - Self::WaitUntil { next, .. } | Self::Call { next, .. } => vec![("next", next)], + Self::WaitUntil { next, .. } => vec![("next", next)], + Self::Call { + next, + on_declared_refusal, + .. + } => std::iter::once(("next", next.as_str())) + .chain( + on_declared_refusal + .iter() + .map(|step| ("onDeclaredRefusal", step.as_str())), + ) + .collect(), Self::WaitFor { next, on_timeout, .. } => std::iter::once(("next", next.as_str())) @@ -267,6 +285,25 @@ fn operation_identities(workflow: &Workflow) -> Vec { identities } +/// Workflows that do not opt into declared-refusal routing retain the exact v1 +/// adapter ABI. The v2 ABI pins both the classification and the branch output +/// shape for workflows that use `onDeclaredRefusal`. +fn adapter_abi(workflow: &Workflow) -> &'static str { + if workflow.steps.values().any(|step| { + matches!( + step, + Step::Call { + on_declared_refusal: Some(_), + .. + } + ) + }) { + ADAPTER_ABI_V2 + } else { + ADAPTER_ABI_V1 + } +} + /// Compiled definitions keyed by the digest of their snapshot text, so one /// worker serving several flows compiles each pinned or in-flight snapshot once. /// @@ -354,11 +391,12 @@ impl Definition { let exit = error.exit_code; crate::authoring::positioned(&document, error.at(&path, "functions")).with_exit(exit) })?; + let adapter_abi = adapter_abi(&workflow).into(); let definition = Self::restore_checked(Snapshot { operation_identities: operation_identities(&workflow), workflow, source, - adapter_abi: ADAPTER_ABI.into(), + adapter_abi, interpreter_abi: INTERPRETER_ABI.into(), schema_abi: SCHEMA_ABI.into(), }) @@ -400,7 +438,7 @@ impl Definition { .iter() .all(OperationIdentity::matches_registered); if !operations_match - || frozen.adapter_abi != ADAPTER_ABI + || frozen.adapter_abi != adapter_abi(&frozen.workflow) || frozen.interpreter_abi != INTERPRETER_ABI || frozen.schema_abi != SCHEMA_ABI { @@ -502,13 +540,20 @@ impl Definition { } } match step { - Step::Call { call, .. } => { + Step::Call { + call, + on_declared_refusal, + .. + } => { item["connection"] = serde_json::json!(call.connection); item["operation"] = serde_json::json!(call.operation); item["product"] = serde_json::json!(call.operation.product()); item["mutating"] = serde_json::json!(call.operation.is_mutating()); item["effect"] = serde_json::json!(call.operation.descriptor().effect); item["recovery"] = serde_json::json!(call.operation.recovery()); + if let Some(target) = on_declared_refusal { + item["onDeclaredRefusal"] = serde_json::json!(target); + } } Step::WaitFor { wait_for, @@ -940,36 +985,53 @@ fn workflow_findings(flow: &Workflow) -> Vec { timeout_seconds, .. } => findings.extend(wait_findings(flow, name, wait_for, *timeout_seconds)), - Step::Call { call, .. } => match flow.connections.get(&call.connection) { - None => findings.push(authored( - "definition.connection", - &format!("steps.{name}.call.connection"), - &format!( - "connection {} is not declared; the declared connections are {connections}", - call.connection - ), - &format!( - "Name one of the declared connections, or declare {} under connections.", - call.connection - ), - )), - Some(kind) if kind != call.operation.product() => findings.push(authored( - "definition.operation", - &format!("steps.{name}.call.operation"), - &format!( - "operation {} belongs to {}, but connection {} is a {kind} connection, which offers {}", - call.operation.as_str(), - call.operation.product(), - call.connection, - offered(kind) - ), - &format!( - "Name an operation {kind} offers, or call a {} connection.", - call.operation.product() - ), - )), - Some(_) => {} - }, + Step::Call { + call, + on_declared_refusal, + .. + } => { + match flow.connections.get(&call.connection) { + None => findings.push(authored( + "definition.connection", + &format!("steps.{name}.call.connection"), + &format!( + "connection {} is not declared; the declared connections are {connections}", + call.connection + ), + &format!( + "Name one of the declared connections, or declare {} under connections.", + call.connection + ), + )), + Some(kind) if kind != call.operation.product() => findings.push(authored( + "definition.operation", + &format!("steps.{name}.call.operation"), + &format!( + "operation {} belongs to {}, but connection {} is a {kind} connection, which offers {}", + call.operation.as_str(), + call.operation.product(), + call.connection, + offered(kind) + ), + &format!( + "Name an operation {kind} offers, or call a {} connection.", + call.operation.product() + ), + )), + Some(_) => {} + } + if on_declared_refusal.is_some() && !call.operation.descriptor().declared_refusal { + findings.push(authored( + "definition.operation", + &format!("steps.{name}.onDeclaredRefusal"), + &format!( + "operation {} does not declare a routable business refusal", + call.operation.as_str() + ), + "Remove onDeclaredRefusal, or use it on an invoke-breg-action call.", + )); + } + } Step::Choose { cases, .. } if cases.is_empty() || cases.len() > 32 || !cases.keys().all(|k| valid_name(k)) => { @@ -1137,8 +1199,16 @@ fn graph_order(flow: &Workflow) -> Vec<&str> { order } -fn graph_findings<'a>(flow: &'a Workflow, incoming: &Incoming<'a>) -> Vec { - let mut findings = Vec::new(); +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum OutputRoute { + CallSuccess, + DeclaredRefusal, + WaitEvent, +} + +pub(crate) type AvailableOutputs = BTreeMap; + +fn reachable_steps(flow: &Workflow) -> BTreeSet<&str> { let mut reachable = BTreeSet::from([flow.start.as_str()]); let mut frontier = vec![flow.start.as_str()]; while let Some(name) = frontier.pop() { @@ -1148,25 +1218,23 @@ fn graph_findings<'a>(flow: &'a Workflow, incoming: &Incoming<'a>) -> Vec Option { + match (step, member) { + (Step::Call { .. }, "next") => Some(OutputRoute::CallSuccess), + (Step::Call { .. }, "onDeclaredRefusal") => Some(OutputRoute::DeclaredRefusal), + (Step::WaitFor { .. }, "next") => Some(OutputRoute::WaitEvent), + _ => None, } - // Topological traversal proves finiteness. The outputs available at a - // step are those every incoming transition carries: what was available at - // its source, plus the source's own output unless the transition is a - // wait's timeout, which leaves the wait without one. A reference therefore - // exists on every route. Only reachable predecessors take part: no route - // arrives from the others. +} + +fn propagated_outputs<'a>( + flow: &'a Workflow, + incoming: &Incoming<'a>, + reachable: &BTreeSet<&'a str>, +) -> BTreeMap<&'a str, AvailableOutputs> { let predecessors = |name: &str| { incoming[name] .iter() @@ -1183,26 +1251,88 @@ fn graph_findings<'a>(flow: &'a Workflow, incoming: &Incoming<'a>) -> Vec> = BTreeMap::new(); + let mut available: BTreeMap<&str, AvailableOutputs> = BTreeMap::new(); while let Some(name) = ready.pop() { let outputs = predecessors(name) .into_iter() .map(|(source, member)| { let mut carried = available[source].clone(); - if member != "onTimeout" { - carried.insert(source); + if let Some(route) = edge_output(&flow.steps[source], member) { + carried.insert(source.to_owned(), route); } carried }) - .reduce(|left, right| left.intersection(&right).copied().collect()) + .reduce(|mut common, incoming| { + // A producer stays available only when every incoming route + // carries the same output shape. Authors consume or normalize + // success and declared-refusal values before their paths join. + common.retain(|producer, route| incoming.get(producer) == Some(route)); + common + }) .unwrap_or_default(); - if let Some(mapping) = flow.steps[name].mapping() { - if let Err(error) = validate_mapping(flow, name, mapping, &outputs) { + available.insert(name, outputs); + for next in flow.steps[name].targets() { + if let Some(count) = pending.get_mut(next) { + *count -= 1; + if *count == 0 { + ready.push(next); + } + } + } + } + available +} + +/// The output shape each producer has on every route to each consumer. The +/// workflow has already passed structural graph checks when the mapping checker +/// calls this helper. +pub(crate) fn output_routes(flow: &Workflow) -> BTreeMap { + let mut incoming: Incoming = flow + .steps + .keys() + .map(|name| (name.as_str(), vec![])) + .collect(); + for (name, step) in &flow.steps { + for (member, target) in step.edges() { + if let Some(sources) = incoming.get_mut(target) { + sources.push((name, member)); + } + } + } + let reachable = reachable_steps(flow); + propagated_outputs(flow, &incoming, &reachable) + .into_iter() + .map(|(name, outputs)| (name.to_owned(), outputs)) + .collect() +} + +fn graph_findings<'a>(flow: &'a Workflow, incoming: &Incoming<'a>) -> Vec { + let mut findings = Vec::new(); + let reachable = reachable_steps(flow); + for name in flow.steps.keys() { + if !reachable.contains(name.as_str()) { + findings.push(authored( + "definition.unreachable", + &format!("steps.{name}"), + &format!( + "step {name} is not reachable from start step {}", + flow.start + ), + &format!("Point a transition at {name}, or remove the step if no route needs it."), + )); + } + } + let available = propagated_outputs(flow, incoming, &reachable); + for (name, outputs) in &available { + if let Some(mapping) = flow.steps[*name].mapping() { + if let Err(error) = validate_mapping(flow, name, mapping, outputs) { findings.push(error); } } - if let Step::WaitFor { wait_for, .. } = &flow.steps[name] { - if correlates(flow, wait_for) && !outputs.contains(wait_for.request()) { + if let Step::WaitFor { wait_for, .. } = &flow.steps[*name] { + if correlates(flow, wait_for) + && outputs.get(wait_for.request()) != Some(&OutputRoute::CallSuccess) + { findings.push(authored( "definition.wait", &format!("steps.{name}.waitFor.request"), @@ -1214,15 +1344,6 @@ fn graph_findings<'a>(flow: &'a Workflow, incoming: &Incoming<'a>) -> Vec, + available: &AvailableOutputs, ) -> Result<()> { let step = &flow.steps[name]; let base = format!("steps.{name}.{}", step.role()); @@ -1332,7 +1453,7 @@ fn validate_mapping( )); } if let Some(prior) = step { - if !available.contains(prior.as_str()) + if !available.contains_key(prior.as_str()) || !matches!( flow.steps.get(prior), Some(Step::Call { .. } | Step::WaitFor { .. }) diff --git a/crates/registry-coordinator/src/mapping_check.rs b/crates/registry-coordinator/src/mapping_check.rs index 573e1a42f..30415f1af 100644 --- a/crates/registry-coordinator/src/mapping_check.rs +++ b/crates/registry-coordinator/src/mapping_check.rs @@ -14,7 +14,10 @@ use serde_json::{Map, Value}; use crate::{ contracts::{contract, event_contract, first_violation, Violation}, - definition::{authored, compile_schema, parse_timestamp, Mapping, Reference, Step, Workflow}, + definition::{ + authored, compile_schema, output_routes, parse_timestamp, AvailableOutputs, Mapping, + OutputRoute, Reference, Step, Workflow, + }, CoordinatorError, Result, }; @@ -29,21 +32,37 @@ fn budget() -> usize { /// Check every step's mapping, in step-name order, and report the first /// mismatch in each step. pub(crate) fn check(workflow: &Workflow) -> Vec { + let routes = output_routes(workflow); workflow .steps .iter() - .filter_map(|(name, step)| check_step(workflow, name, step).err()) + .filter_map(|(name, step)| { + check_step( + workflow, + name, + step, + routes + .get(name) + .expect("a checked workflow reaches every step"), + ) + .err() + }) .collect() } -fn check_step(workflow: &Workflow, name: &str, step: &Step) -> Result<()> { +fn check_step( + workflow: &Workflow, + name: &str, + step: &Step, + available: &AvailableOutputs, +) -> Result<()> { let Some(mapping) = step.mapping() else { return Ok(()); }; let base = format!("steps.{name}.{}", step.role()); let mut read = BTreeMap::new(); for (member, reference) in mapping.references() { - if let Some(schema) = source(workflow, reference, &format!("{base}{member}"))? { + if let Some(schema) = source(workflow, available, reference, &format!("{base}{member}"))? { read.insert(member, schema); } } @@ -83,6 +102,7 @@ struct Read<'a> { /// Resolve an input or step reference's path in its producer's schema. fn source<'a>( workflow: &'a Workflow, + available: &AvailableOutputs, reference: &Reference, field: &str, ) -> Result>> { @@ -93,11 +113,25 @@ fn source<'a>( path, ), Reference::Step { step, path } => match workflow.steps.get(step) { - Some(Step::Call { call, .. }) => ( - &contract(call.operation).output, - format!("the {} output of step {step}", call.operation.as_str()), - path, - ), + Some(Step::Call { call, .. }) => match available.get(step) { + Some(OutputRoute::CallSuccess) => ( + &contract(call.operation).output, + format!("the {} output of step {step}", call.operation.as_str()), + path, + ), + Some(OutputRoute::DeclaredRefusal) => ( + contract(call.operation) + .declared_refusal + .as_ref() + .expect("a declared-refusal route has a committed contract"), + format!( + "the {} declared-refusal output of step {step}", + call.operation.as_str() + ), + path, + ), + _ => return Ok(None), + }, Some(Step::WaitFor { wait_for, .. }) => ( &event_contract(wait_for).output, format!("the {} event of step {step}", wait_for.id()), diff --git a/crates/registry-coordinator/src/operations.rs b/crates/registry-coordinator/src/operations.rs index 6568f9fa9..1da2f067e 100644 --- a/crates/registry-coordinator/src/operations.rs +++ b/crates/registry-coordinator/src/operations.rs @@ -21,8 +21,10 @@ pub enum KeyRequirement { Required, } -/// Binding behavior is versioned independently of the workflow. Changing a -/// descriptor changes its identity and must not silently alter an admitted run. +/// Binding behavior is versioned independently of the workflow. The fields +/// copied into [`OperationIdentity`] must not silently change for an admitted +/// run; authoring capabilities outside that identity are pinned by their +/// workflow ABI. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct OperationDescriptor { pub id: &'static str, @@ -33,6 +35,12 @@ pub struct OperationDescriptor { pub requires_preparation: bool, pub recovery: RecoverySemantics, pub read_receipt: bool, + /// Whether an adapter ABI v2 workflow may route a product-declared, + /// no-effect business refusal through `onDeclaredRefusal`. + /// + /// This capability is pinned by the workflow's adapter ABI rather than by + /// [`OperationIdentity`], so adding it does not invalidate v1 snapshots. + pub declared_refusal: bool, } /// Exact nonsecret binding semantics retained in the immutable snapshot. @@ -80,6 +88,7 @@ const fn read(id: &'static str, product: &'static str) -> OperationDescriptor { requires_preparation: false, recovery: RecoverySemantics::ReadAgain, read_receipt: false, + declared_refusal: false, } } @@ -102,6 +111,7 @@ const fn mutation( RecoverySemantics::SameCommand }, read_receipt, + declared_refusal: false, } } @@ -111,7 +121,10 @@ static CATALOG: &[OperationDescriptor] = &[ read("read-scheduling", "scheduling"), read("read-availability", "scheduling"), mutation("create-appointment", "scheduling", true, false), - mutation("invoke-breg-action", "breg", false, true), + OperationDescriptor { + declared_refusal: true, + ..mutation("invoke-breg-action", "breg", false, true) + }, read("external-get", "external-http"), OperationDescriptor { id: "evaluate-decision", @@ -122,6 +135,7 @@ static CATALOG: &[OperationDescriptor] = &[ requires_preparation: true, recovery: RecoverySemantics::HoldAfterDispatch, read_receipt: false, + declared_refusal: false, }, mutation("submit-review-request", "casework", false, false), mutation("cancel-review-request", "casework", false, false), diff --git a/crates/registry-coordinator/src/protocol.rs b/crates/registry-coordinator/src/protocol.rs index 1d1708ed0..b5443abb6 100644 --- a/crates/registry-coordinator/src/protocol.rs +++ b/crates/registry-coordinator/src/protocol.rs @@ -142,6 +142,50 @@ pub struct CallRequest { pub idempotency_key: Option, } +/// One package-declared business refusal from a validated product response. +/// +/// The code is deliberately absent from `Debug`; callers must opt in to reading +/// or serializing it for an authored refusal branch. +#[derive(Clone, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "schema", schemars(deny_unknown_fields))] +pub struct DeclaredRefusal { + #[cfg_attr( + feature = "schema", + schemars( + length(min = 1, max = 128), + regex(pattern = r"^[^\x00-\x1F\x7F-\x9F]+$") + ) + )] + refusal_code: String, +} + +impl DeclaredRefusal { + /// Retain only the same bounded code shape the maintained product client + /// accepts from its wire contract. + #[must_use] + pub fn new(refusal_code: &str) -> Option { + (!refusal_code.is_empty() + && refusal_code.chars().count() <= 128 + && !refusal_code.chars().any(char::is_control)) + .then(|| Self { + refusal_code: refusal_code.to_owned(), + }) + } + + #[must_use] + pub fn refusal_code(&self) -> &str { + self.refusal_code.as_str() + } +} + +impl std::fmt::Debug for DeclaredRefusal { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("DeclaredRefusal()") + } +} + /// Bounded categories. Never carry a remote error body or credential here. pub enum CallOutcome { Success(Value), @@ -157,12 +201,36 @@ pub enum CallOutcome { Refused { code: String, }, + /// A validated business refusal explicitly declared by the called product. + /// This distinct class is eligible for an authored branch at execution time. + DeclaredRefusal(DeclaredRefusal), Uncertain { code: String, }, ReceiptExpired, } +#[cfg(test)] +mod declared_refusal_tests { + use super::DeclaredRefusal; + use serde_json::json; + + #[test] + fn declared_refusal_is_bounded_serializable_and_redacted() { + let refusal = DeclaredRefusal::new("eligibility-not-met").unwrap(); + assert_eq!(refusal.refusal_code(), "eligibility-not-met"); + assert_eq!( + serde_json::to_value(&refusal).unwrap(), + json!({"refusalCode":"eligibility-not-met"}) + ); + assert_eq!(format!("{refusal:?}"), "DeclaredRefusal()"); + assert!(DeclaredRefusal::new("").is_none()); + assert!(DeclaredRefusal::new("line\nbreak").is_none()); + assert!(DeclaredRefusal::new(&"x".repeat(129)).is_none()); + assert!(DeclaredRefusal::new(&"é".repeat(128)).is_some()); + } +} + /// One read of a review result for a waiting step. A read never changes the /// review, so no answer is uncertain: a failure either repeats or ends the /// wait under its bounded code. diff --git a/crates/registry-coordinator/src/scenarios.rs b/crates/registry-coordinator/src/scenarios.rs index c6d366b75..4e5a614cd 100644 --- a/crates/registry-coordinator/src/scenarios.rs +++ b/crates/registry-coordinator/src/scenarios.rs @@ -5,14 +5,14 @@ use crate::{ authoring, contracts::{contract, event_contract, Violation}, definition::{class, parse_timestamp, Definition, Step, WaitEvent}, - protocol::CallRequest, + protocol::{CallRequest, DeclaredRefusal}, CoordinatorError, Result, }; use chrono::{DateTime, Utc}; use registry_platform_config::AuthoredExpressions; use registry_platform_yaml::{ ApiVersion, BoundedU32, Diagnostic, Document, EnvelopeRule, Expect, ForeignValue, FormatSpec, - Identified, LocalId, Reader, Severity, UniqueIdList, UniqueList, + Identified, Invalid, LocalId, Reader, Severity, UniqueIdList, UniqueList, }; use registry_review_protocol::{ReviewRequestAccepted, ReviewResult}; use serde::{Deserialize, Serialize}; @@ -118,6 +118,13 @@ enum AuthoredReply { /// The code the reply carries, such as product-not-found. code: LocalId, }, + /// The product returned a package-declared, no-effect business refusal. + /// A supported call takes its `onDeclaredRefusal` edge when it has one; + /// otherwise the run stops as refused. + DeclaredRefusal { + /// The validated refusal code the package declares. + refusal_code: RefusalCode, + }, /// The call's effect is unknown; the run stops as uncertain unless a /// recovery consumes another reply. Uncertain { @@ -131,6 +138,47 @@ enum AuthoredReply { } registry_platform_yaml::tagged_union!(AuthoredReply); +/// One package-declared refusal code, kept in the same bounded shape as the +/// protocol value without admitting an arbitrary product problem document. +struct RefusalCode(String); + +impl std::fmt::Debug for RefusalCode { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("RefusalCode()") + } +} + +impl<'de> Deserialize<'de> for RefusalCode { + fn deserialize>( + deserializer: D, + ) -> std::result::Result { + let text = String::deserialize(deserializer)?; + DeclaredRefusal::new(&text).map(|_| Self(text)).ok_or_else(|| { + Invalid::expected( + "a non-empty declared refusal code of at most 128 characters without control characters", + "Write the package-declared refusal code exactly, without control characters.", + ) + .into_error() + }) + } +} + +#[cfg(feature = "schema")] +impl schemars::JsonSchema for RefusalCode { + fn schema_name() -> std::borrow::Cow<'static, str> { + "DeclaredRefusalCode".into() + } + + fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema { + schemars::json_schema!({ + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[^\\u0000-\\u001F\\u007F-\\u009F]+$", + }) + } +} + #[derive(Deserialize)] #[serde( remote = "Self", @@ -277,6 +325,10 @@ pub enum Reply { Refused { refused: String, }, + DeclaredRefusal { + #[serde(rename = "declaredRefusal")] + declared_refusal: String, + }, Uncertain { uncertain: String, }, @@ -471,6 +523,11 @@ impl CoordinatorScenarios { AuthoredReply::Refused { code } => Reply::Refused { refused: code.into_string(), }, + AuthoredReply::DeclaredRefusal { refusal_code } => { + Reply::DeclaredRefusal { + declared_refusal: refusal_code.0, + } + } AuthoredReply::Uncertain { code } => Reply::Uncertain { uncertain: code.into_string(), }, @@ -618,18 +675,39 @@ fn execute(definition: &Definition, scenario: &Scenario) -> Result { + contract(call.operation) + .check_output(success) + .map_err(|violation| { + contract_error( + "scenario.reply-output", + format!("/replies/{step}/{index}/value{}", violation.path), + &violation, + &format!("in a success reply to step {step}, as the {operation} output declares; the synthetic value does not match"), + &format!("Correct the synthetic value so it satisfies the {operation} output."), + ) + })?; + } + Reply::DeclaredRefusal { declared_refusal } => { + if !call.operation.descriptor().declared_refusal { + return Err(error( + "scenario.reply", + format!("/replies/{step}/{index}"), + format!("step {step} calls {operation}, which cannot return a package-declared business refusal"), + "Use declared-refusal only for invoke-breg-action; use refused for an operational or authority refusal.", + )); + } + if DeclaredRefusal::new(declared_refusal).is_none() { + return Err(error( + "scenario.reply", + format!("/replies/{step}/{index}/refusalCode"), + format!("the declared refusal code for step {step} is outside its contract"), + "Write a non-empty package-declared refusal code of at most 128 characters without control characters.", + )); + } + } + _ => {} } } } @@ -791,7 +869,12 @@ fn execute(definition: &Definition, scenario: &Scenario) -> Result { + Step::Call { + call, + next, + on_declared_refusal, + .. + } => { let value = definition.evaluate_synthetic(&step, &scenario.input, &outputs)?; let operation = call.operation.as_str(); contract(call.operation) @@ -847,7 +930,7 @@ fn execute(definition: &Definition, scenario: &Scenario) -> Result Result { - success = Some(value.clone()); + completion = Some((next.clone(), value.clone())); + break; + } + Reply::DeclaredRefusal { declared_refusal } => { + if prior_unknown { + report.state = "uncertain".into(); + } else if let Some(target) = on_declared_refusal { + let refusal = DeclaredRefusal::new(declared_refusal).ok_or_else(|| { + error( + "scenario.reply", + format!("{at}/refusalCode"), + format!("the declared refusal code for step {step} is outside its contract"), + "Write a non-empty package-declared refusal code of at most 128 characters without control characters.", + ) + })?; + let output = serde_json::to_value(refusal).map_err(|_| { + error( + "scenario.reply", + format!("{at}/refusalCode"), + format!("the declared refusal code for step {step} could not be retained"), + "Report this Coordinator defect with the project.", + ) + })?; + completion = Some((target.clone(), output)); + } else { + report.state = "refused".into(); + } break; } Reply::Uncertain { .. } => { @@ -953,10 +1062,10 @@ fn execute(definition: &Definition, scenario: &Scenario) -> Result, pub uncertain: bool, pub receipt_expired: bool, + /// A delivered mutating call whose typed result proves no product effect + /// occurred. This matters only when cancellation raced with completion. + pub completed_without_effect: bool, /// Seconds until a waiting step reads its result again; set only while /// the wait goes on. pub recheck_seconds: Option, @@ -296,6 +299,7 @@ impl Detail { failure_path: None, uncertain, receipt_expired: false, + completed_without_effect: false, recheck_seconds: None, } } @@ -1376,7 +1380,7 @@ impl Store { } else { detail.next.as_deref().unwrap_or(step) }; - let state = if cancelled && dispatch_risk { + let state = if cancelled && dispatch_risk && !detail.completed_without_effect { "cancelled-after-effect" } else if cancelled { "cancelled" diff --git a/crates/registry-coordinator/src/store/recovery.rs b/crates/registry-coordinator/src/store/recovery.rs index 0151f19bb..740930aad 100644 --- a/crates/registry-coordinator/src/store/recovery.rs +++ b/crates/registry-coordinator/src/store/recovery.rs @@ -340,7 +340,7 @@ impl Store { let input_row=tx.query_one(&format!("SELECT input FROM {}.runs WHERE run_id=$1",self.namespace),&[&run]).await.map_err(unavailable)?; let input:Value=self.open_value(run,"input","",&input_row.get::<_,Value>(0))?; let payload=Payload{snapshot,input,outputs,command:Some(command)}; - let detail=Detail{next:Some(next.clone()),output:Some(value),outcome:None,failure_code:None,failure_path:None,uncertain:false,receipt_expired:false,recheck_seconds:None}; + let detail=Detail{next:Some(next.clone()),output:Some(value),outcome:None,failure_code:None,failure_path:None,uncertain:false,receipt_expired:false,completed_without_effect:false,recheck_seconds:None}; let written=self.apply_success(&tx,run,&step,payload,&detail,current.get(4)).await.map_err(unavailable)?; let mapping_failed=written=="failed"; tx.execute(&format!("UPDATE {}.jobs SET state='delivered',attempt=GREATEST(attempt,1),uncertain=false,next_attempt_at=NULL,attempt_started_at=NULL,lease_expires_at=NULL,lease_token=NULL,delivered_at=transaction_timestamp(),dead_lettered_at=NULL,expired_at=NULL,failure_code=NULL,failure_path=NULL WHERE run_id=$1 AND step=$2",self.namespace),&[&run,&step]).await.map_err(unavailable)?; diff --git a/crates/registry-coordinator/src/worker.rs b/crates/registry-coordinator/src/worker.rs index ae574d091..a73e4b473 100644 --- a/crates/registry-coordinator/src/worker.rs +++ b/crates/registry-coordinator/src/worker.rs @@ -87,11 +87,17 @@ fn advance(next: String, output: Option) -> Detail { failure_path: None, uncertain: false, receipt_expired: false, + completed_without_effect: false, recheck_seconds: None, } } -fn call_result(result: CallOutcome, next: &str, uncertain: bool) -> Sent { +fn call_result( + result: CallOutcome, + next: &str, + on_declared_refusal: Option<&str>, + uncertain: bool, +) -> Sent { match result { CallOutcome::Success(value) => accepted(advance(next.to_owned(), Some(value))), CallOutcome::Retryable { code, retry_after } if !uncertain => Sent { @@ -100,6 +106,18 @@ fn call_result(result: CallOutcome, next: &str, uncertain: bool) -> Sent }, CallOutcome::Retryable { code, .. } => failed(safe_code(&code, "remote-retryable"), true), CallOutcome::Refused { code } => failed(safe_code(&code, "remote-refused"), uncertain), + CallOutcome::DeclaredRefusal(refusal) if !uncertain => { + let Some(step) = on_declared_refusal else { + return failed("product-refused", false); + }; + let Ok(output) = serde_json::to_value(refusal) else { + return failed("invalid-response", false); + }; + let mut detail = advance(step.to_owned(), Some(output)); + detail.completed_without_effect = true; + accepted(detail) + } + CallOutcome::DeclaredRefusal(_) => failed("product-refused", true), CallOutcome::Uncertain { code } => failed(safe_code(&code, "remote-uncertain"), true), CallOutcome::ReceiptExpired => { let mut sent = failed("receipt-expired", true); @@ -155,9 +173,9 @@ fn wait_result( Some(ReviewRead::Failed( CallOutcome::Refused { code } | CallOutcome::Uncertain { code }, )) => failed(safe_code(&code, "remote-refused"), false), - Some(ReviewRead::Failed(CallOutcome::Success(_) | CallOutcome::ReceiptExpired)) => { - failed("invalid-response", false) - } + Some(ReviewRead::Failed( + CallOutcome::Success(_) | CallOutcome::DeclaredRefusal(_) | CallOutcome::ReceiptExpired, + )) => failed("invalid-response", false), _ if timed_out => match on_timeout { Some(step) => accepted(advance(step.to_owned(), None)), None => failed("wait-timed-out", false), @@ -170,6 +188,7 @@ fn wait_result( failure_path: None, uncertain: false, receipt_expired: false, + completed_without_effect: false, recheck_seconds: Some(recheck_seconds), }), Some(ReviewRead::Failed(CallOutcome::Retryable { code, retry_after })) => { @@ -287,10 +306,16 @@ impl Worker { failure_path: None, uncertain: false, receipt_expired: false, + completed_without_effect: false, recheck_seconds: None, })) } - Step::Call { call, next, .. } => { + Step::Call { + call, + next, + on_declared_refusal, + .. + } => { // A recovery race or stale queue state cannot turn unknown // inference completion into a second provider evaluation. if job.job.uncertain && !call.operation.can_retry_after_unknown() { @@ -349,7 +374,7 @@ impl Worker { return Ok(failed("preparation-invalid", job.job.uncertain)); } Ok(Err(outcome)) => { - return Ok(call_result(outcome, next, job.job.uncertain)) + return Ok(call_result(outcome, next, None, job.job.uncertain)) } Err(_) => { return Ok(call_result( @@ -358,6 +383,7 @@ impl Worker { retry_after: None, }, next, + None, job.job.uncertain, )) } @@ -411,7 +437,12 @@ impl Worker { retry_after: None, }, }; - Ok(call_result(result, next, job.job.uncertain)) + Ok(call_result( + result, + next, + on_declared_refusal.as_deref(), + job.job.uncertain, + )) } } } @@ -457,7 +488,7 @@ mod tests { use crate::{ definition::WaitEvent, metrics::StepAttempt, - protocol::{CallOutcome, ReviewRead}, + protocol::{CallOutcome, DeclaredRefusal, ReviewRead}, }; use registry_platform_dispatch::SendOutcome; use serde_json::{json, Value}; @@ -667,21 +698,58 @@ mod tests { retry_after, }; let hint = Some(Duration::from_secs(30)); - let sent = call_result(rate_limited(hint), "next", false); + let sent = call_result(rate_limited(hint), "next", None, false); assert_eq!(sent.outcome, SendOutcome::Transient { retry_after: hint }); assert_eq!(sent.detail.failure_code, Some("rate-limited")); assert_eq!( - call_result(rate_limited(None), "next", false).outcome, + call_result(rate_limited(None), "next", None, false).outcome, SendOutcome::Transient { retry_after: None } ); // An attempt after an unknown outcome stays held: a hint never turns // it into an automatic resend. assert_eq!( - call_result(rate_limited(hint), "next", true).outcome, + call_result(rate_limited(hint), "next", None, true).outcome, SendOutcome::MaybeSent ); } + #[test] + fn declared_refusal_routes_only_with_opt_in_and_no_prior_uncertainty() { + let refusal = + || CallOutcome::DeclaredRefusal(DeclaredRefusal::new("label-not-eligible").unwrap()); + let routed = call_result(refusal(), "success", Some("not-eligible"), false); + assert!(matches!(routed.outcome, SendOutcome::Accepted { .. })); + assert_eq!(routed.detail.next.as_deref(), Some("not-eligible")); + assert_eq!( + routed.detail.output, + Some(json!({"refusalCode":"label-not-eligible"})) + ); + assert!(routed.detail.completed_without_effect); + + for prior_uncertainty in [false, true] { + let refused = call_result(refusal(), "success", None, prior_uncertainty); + assert_eq!(refused.detail.failure_code, Some("product-refused")); + assert_eq!(refused.detail.next, None); + assert_eq!(refused.detail.output, None); + assert_eq!( + refused.outcome, + if prior_uncertainty { + SendOutcome::MaybeSent + } else { + SendOutcome::Permanent { + code: registry_platform_dispatch::FailureCode::new("product-refused") + .unwrap(), + } + } + ); + } + let held = call_result(refusal(), "success", Some("not-eligible"), true); + assert_eq!(held.outcome, SendOutcome::MaybeSent); + assert_eq!(held.detail.next, None); + assert_eq!(held.detail.failure_code, Some("product-refused")); + assert!(!held.detail.completed_without_effect); + } + #[test] fn operator_codes_preserve_reviewed_categories_and_scrub_runtime_text() { assert_eq!( diff --git a/crates/registry-coordinator/tests/authored_project.rs b/crates/registry-coordinator/tests/authored_project.rs index 7238c8e50..11aa9180e 100644 --- a/crates/registry-coordinator/tests/authored_project.rs +++ b/crates/registry-coordinator/tests/authored_project.rs @@ -15,6 +15,30 @@ fn project() -> Value { }) } +fn declared_refusal_project() -> Value { + json!({ + "apiVersion": authoring::API_VERSION, "kind": authoring::KIND, + "project": {"id":"declared-refusal", "version":"1"}, + "input": {}, "connections": {"registry":"breg"}, + "functionsFile":"functions.rhai", "deadlineSeconds":60, + "start":"act", + "steps": { + "act": {"type":"call", "connection":"registry", "operation":"invoke-breg-action", + "input":{"type":"function", "function":"command", "arguments":[{"type":"input"}]}, + "next":"accepted", "onDeclaredRefusal":"refused"}, + "accepted": {"type":"finish", "outcome":"accepted"}, + "refused": {"type":"finish", "outcome":"refused", "output":{ + "type":"fields", "fields":{"refusalCode":{"type":"step", "step":"act", "path":"/refusalCode"}} + }} + }, + "outcomes": { + "accepted":{"type":"null"}, + "refused":{"type":"object", "required":["refusalCode"], + "properties":{"refusalCode":{"type":"string"}}, "additionalProperties":false} + } + }) +} + fn write_project(root: &Path, project: &Value, source: &str) { fs::write( root.join("workflow.yaml"), @@ -54,6 +78,117 @@ fn authored_project_converts_to_the_versioned_snapshot_abi() { ); } +#[test] +fn declared_refusal_opt_in_uses_v2_and_restores_only_with_its_shape() { + let root = tempfile::tempdir().unwrap(); + let project = declared_refusal_project(); + write_project(root.path(), &project, "fn command(input) { input }"); + let definition = Definition::load(root.path()).unwrap(); + let snapshot = definition.snapshot().unwrap(); + let frozen: Value = serde_json::from_str(&snapshot).unwrap(); + assert_eq!(frozen["adapterAbi"], "coordinator/product-operations/v2"); + assert_eq!( + frozen["workflow"]["steps"]["act"]["onDeclaredRefusal"], + "refused" + ); + let explained = definition.explain().unwrap(); + let act = explained["steps"] + .as_array() + .unwrap() + .iter() + .find(|step| step["id"] == "act") + .unwrap(); + assert_eq!(act["onDeclaredRefusal"], "refused"); + assert_eq!( + Definition::from_snapshot(&snapshot) + .unwrap() + .snapshot() + .unwrap(), + snapshot + ); + + for changed in [ + { + let mut changed = frozen.clone(); + changed["adapterAbi"] = json!("coordinator/product-operations/v1"); + changed + }, + { + let mut changed = frozen.clone(); + changed["workflow"]["steps"]["act"] + .as_object_mut() + .unwrap() + .remove("onDeclaredRefusal"); + changed + }, + ] { + let error = Definition::from_snapshot(&changed.to_string()) + .err() + .expect("ABI and feature shape must agree"); + assert_eq!(error.code, "definition.abi"); + } +} + +#[test] +fn declared_refusal_is_available_only_on_its_edge_and_not_after_a_mixed_join() { + let root = tempfile::tempdir().unwrap(); + let source = "fn command(input) { input } fn identity(value) { value }"; + let project = declared_refusal_project(); + write_project(root.path(), &project, source); + Definition::load(root.path()).expect("the refusal edge reads its declared output"); + + let mut success_reads_refusal = project.clone(); + success_reads_refusal["steps"]["accepted"]["output"] = json!({ + "type":"fields", "fields":{"refusalCode":{ + "type":"step", "step":"act", "path":"/refusalCode" + }} + }); + success_reads_refusal["outcomes"]["accepted"] = + success_reads_refusal["outcomes"]["refused"].clone(); + write_project(root.path(), &success_reads_refusal, source); + assert_eq!( + Definition::load(root.path()).err().unwrap().code, + "coordinator.mapping.path" + ); + + let mut refusal_reads_success = project.clone(); + refusal_reads_success["steps"]["refused"]["output"]["fields"]["refusalCode"]["path"] = + json!("/applicationId"); + write_project(root.path(), &refusal_reads_success, source); + assert_eq!( + Definition::load(root.path()).err().unwrap().code, + "coordinator.mapping.path" + ); + + let mut joined = project; + joined["steps"]["act"]["next"] = json!("joined"); + joined["steps"]["act"]["onDeclaredRefusal"] = json!("joined"); + joined["steps"].as_object_mut().unwrap().remove("accepted"); + joined["steps"].as_object_mut().unwrap().remove("refused"); + joined["steps"]["joined"] = json!({ + "type":"finish", "outcome":"joined", "output":{ + "type":"function", "function":"identity", "arguments":[{"type":"step", "step":"act"}] + } + }); + joined["outcomes"] = json!({"joined":{}}); + write_project(root.path(), &joined, source); + assert_eq!( + Definition::load(root.path()).err().unwrap().code, + "coordinator.mapping.dominance" + ); +} + +#[test] +fn operations_without_a_declared_refusal_contract_reject_the_edge() { + let root = tempfile::tempdir().unwrap(); + let mut project = declared_refusal_project(); + project["steps"]["act"]["operation"] = json!("read-record"); + write_project(root.path(), &project, "fn command(input) { input }"); + let error = Definition::load(root.path()).err().unwrap(); + assert_eq!(error.code, "coordinator.definition.operation"); + assert_eq!(error.field.as_deref(), Some("/steps/act/onDeclaredRefusal")); +} + #[test] fn function_compile_diagnostics_keep_the_function_file_and_position() { let root = tempfile::tempdir().unwrap(); @@ -423,6 +558,18 @@ fn project_schema_holds_each_fields_mapping_to_its_operation_input_members() { .map(|descriptor| descriptor.id) .collect(); assert_eq!(constrained, catalog); + + let function = + json!({"type": "function", "function": "identity", "arguments": [{"type": "input"}]}); + let mut supported = call_project("invoke-breg-action", function.clone()); + supported["steps"]["call"]["onDeclaredRefusal"] = json!("done"); + assert!(validator.is_valid(&supported)); + let mut unsupported = call_project("read-record", function); + unsupported["steps"]["call"]["onDeclaredRefusal"] = json!("done"); + assert!(!validator.is_valid(&unsupported)); + assert!(call["properties"]["onDeclaredRefusal"] + .get("default") + .is_none()); } #[cfg(feature = "schema")] diff --git a/crates/registry-coordinator/tests/authored_scenarios.rs b/crates/registry-coordinator/tests/authored_scenarios.rs index fe5c3bbab..ccbd46682 100644 --- a/crates/registry-coordinator/tests/authored_scenarios.rs +++ b/crates/registry-coordinator/tests/authored_scenarios.rs @@ -30,6 +30,7 @@ fn authored_scenarios_decode_tagged_replies_and_preserve_foreign_nulls() { for reply in [ "{type: retryable, code: transport-unavailable}", "{type: refused, code: credential-refused}", + "{type: declared-refusal, refusalCode: eligibility-not-met}", "{type: uncertain, code: transport-uncertain}", "{type: receipt-expired}", ] { @@ -39,6 +40,52 @@ fn authored_scenarios_decode_tagged_replies_and_preserve_foreign_nulls() { ) .unwrap(); } + + let document = scenarios::decode( + &source( + "{type: declared-refusal, refusalCode: eligibility-not-met}", + "", + ), + "scenarios.yaml", + ) + .unwrap(); + assert!(matches!( + &document.cases[0].replies["call-one"][0], + Reply::DeclaredRefusal { declared_refusal } + if declared_refusal == "eligibility-not-met" + )); +} + +#[test] +fn declared_refusal_codes_are_nonempty_bounded_and_free_of_controls() { + for (reply, code) in [ + ( + "{type: declared-refusal, refusalCode: ''}".to_owned(), + "config.invalid-value", + ), + ( + format!( + "{{type: declared-refusal, refusalCode: {}}}", + "x".repeat(129) + ), + "config.invalid-value", + ), + ( + r#"{type: declared-refusal, refusalCode: "bad\u007fcode"}"#.to_owned(), + "yaml.control-character", + ), + ] { + let error = scenarios::decode(&source(&reply, ""), "scenarios.yaml") + .err() + .expect("must refuse an invalid declared refusal code"); + let finding = error + .diagnostics + .iter() + .find(|finding| finding.code == code) + .unwrap_or_else(|| panic!("{error}")); + assert_eq!(finding.path, "/cases/0/replies/call-one/0/refusalCode"); + assert!(!error.to_string().contains(&reply)); + } } #[test] @@ -204,6 +251,18 @@ fn scenario_schema_keeps_foreign_payloads_and_closed_tagged_forms() { schema["$defs"]["AuthoredWait"]["oneOf"][0]["properties"]["value"]["x-registry-foreign"], "json" ); + assert_eq!(schema["$defs"]["DeclaredRefusalCode"]["minLength"], 1); + assert_eq!(schema["$defs"]["DeclaredRefusalCode"]["maxLength"], 128); + assert_eq!( + schema["$defs"]["DeclaredRefusalCode"]["pattern"], + "^[^\\u0000-\\u001F\\u007F-\\u009F]+$" + ); + let refusal_code = jsonschema::JSONSchema::options() + .with_draft(jsonschema::Draft::Draft202012) + .compile(&schema["$defs"]["DeclaredRefusalCode"]) + .unwrap(); + assert!(refusal_code.is_valid(&json!("eligibility-not-met"))); + assert!(!refusal_code.is_valid(&json!("bad\u{85}code"))); for tagged in ["AuthoredReply", "AuthoredWait"] { assert!(schema["$defs"][tagged]["oneOf"] .as_array() diff --git a/crates/registry-coordinator/tests/authoring.rs b/crates/registry-coordinator/tests/authoring.rs index c680b055f..13b90c6b6 100644 --- a/crates/registry-coordinator/tests/authoring.rs +++ b/crates/registry-coordinator/tests/authoring.rs @@ -188,6 +188,31 @@ fn snapshot_restores_exact_functions_after_project_changes_or_removal() { ); } +#[test] +fn legacy_v1_snapshot_restores_with_exact_bytes_digest_and_operation_identities() { + const SNAPSHOT: &str = include_str!("fixtures/legacy-definition-v1.json"); + let definition = Definition::from_snapshot(SNAPSHOT).unwrap(); + assert_eq!(definition.snapshot().unwrap(), SNAPSHOT); + assert_eq!( + definition.digest, + "sha256:7298d1fa649b7e0630cd9131564ef0e9ee0c24598a7670a5c8dcd55c5946fdcc" + ); + let frozen: Value = serde_json::from_str(SNAPSHOT).unwrap(); + assert_eq!(frozen["adapterAbi"], "coordinator/product-operations/v1"); + assert!(SNAPSHOT.contains("\"operationIdentities\"")); + assert!(!SNAPSHOT.contains("onDeclaredRefusal")); + assert!(frozen["operationIdentities"] + .as_array() + .unwrap() + .iter() + .all(|identity| identity.get("declaredRefusal").is_none())); + assert!(definition.explain().unwrap()["steps"] + .as_array() + .unwrap() + .iter() + .all(|step| step.get("onDeclaredRefusal").is_none())); +} + #[test] fn definition_cache_compiles_each_snapshot_once_within_its_bound() { use registry_coordinator::definition::DefinitionCache; diff --git a/crates/registry-coordinator/tests/breg_action_adapter.rs b/crates/registry-coordinator/tests/breg_action_adapter.rs index 5338c4820..f9c822b1b 100644 --- a/crates/registry-coordinator/tests/breg_action_adapter.rs +++ b/crates/registry-coordinator/tests/breg_action_adapter.rs @@ -319,6 +319,54 @@ async fn invalid_action_inputs_refuse_before_fetching_conditions() { assert_eq!(server.received_requests().await.unwrap().len(), 1); } +#[tokio::test] +async fn typed_action_refused_problem_exposes_only_its_bounded_code() { + let server = MockServer::start().await; + authority(&server).await; + Mock::given(method("POST")) + .and(path("/tenant/base/v1/actions/update-item")) + .respond_with( + ResponseTemplate::new(422) + .insert_header("traceparent", TRACE) + .insert_header("cache-control", "no-store") + .set_body_raw( + json!({ + "type":"https://id.registrystack.org/problems/registry-breg/action/refused", + "title":"Unprocessable Entity", + "status":422, + "code":"action.refused", + "detail":"Synthetic private label that must not cross the adapter.", + "traceId":"4bf92f3577b34da6a3ce929d0e0e4736", + "fieldPath":"/input/label", + "refusalCode":"label-not-eligible" + }) + .to_string(), + "application/problem+json", + ), + ) + .expect(1) + .mount(&server) + .await; + let client = client(&server, Arc::new(CountingToken::default())); + let command = request(); + let prepared = prepare(&client, "writer", &command) + .await + .unwrap_or_else(|_| panic!("valid preparation")); + let refusal = match execute(&client, "writer", &command, &prepared).await { + CallOutcome::DeclaredRefusal(refusal) => refusal, + _ => panic!("validated action refusal must keep its distinct class"), + }; + assert_eq!(refusal.refusal_code(), "label-not-eligible"); + assert_eq!( + serde_json::to_value(&refusal).unwrap(), + json!({"refusalCode":"label-not-eligible"}) + ); + let debug = format!("{refusal:?}"); + assert!(!debug.contains("label-not-eligible")); + assert!(!debug.contains("private label")); + assert!(!debug.contains("/input/label")); +} + #[tokio::test] async fn expired_receipt_and_known_rollbacks_remain_distinct_from_unknown_effects() { for code in [ diff --git a/crates/registry-coordinator/tests/breg_action_native.rs b/crates/registry-coordinator/tests/breg_action_native.rs index 4a5548e3e..94fd12904 100644 --- a/crates/registry-coordinator/tests/breg_action_native.rs +++ b/crates/registry-coordinator/tests/breg_action_native.rs @@ -10,6 +10,7 @@ use registry_breg_client::{ }; use registry_coordinator::{ adapters::HttpAdapters, + breg_action, protocol::{AdapterSet, CallOutcome, CallRequest, Operation, ReconciliationOutcome}, }; use serde_json::{json, Value}; @@ -103,6 +104,34 @@ async fn real_breg_action_recovers_exactly_once_and_rechecks_current_authority() .await .unwrap(); let record = seed.value.data.record_identifier; + let refused_request = CallRequest { + connection: "applications".into(), + operation: Operation::InvokeBregAction, + input: json!({"action":"update-item", "input":{"targetId":record, "label":"Refuse"}}), + idempotency_key: Some("coordinator-native-refusal".into()), + }; + let refused_prepared = breg_action::prepare(&client, "writer", &refused_request) + .await + .unwrap_or_else(|_| panic!("native refusal preparation")); + let refusal = + match breg_action::execute(&client, "writer", &refused_request, &refused_prepared).await { + CallOutcome::DeclaredRefusal(refusal) => refusal, + _ => panic!("native declared business refusal"), + }; + assert_eq!(refusal.refusal_code(), "label-not-eligible"); + assert_eq!( + serde_json::to_value(&refusal).unwrap(), + json!({"refusalCode":"label-not-eligible"}) + ); + let read_options = BRegRecordOptions::default() + .access_profile("writer") + .unwrap(); + let unchanged = client + .get_record("items", &record, &read_options) + .await + .unwrap(); + assert_eq!(unchanged.value.data.domain_data["label"], "Before"); + assert_eq!(unchanged.value.data.revision_identifier, "1"); let mut proxy = LostResponse::start(&service.base_url); let root = tempfile::tempdir_in(std::env::temp_dir().canonicalize().unwrap()).unwrap(); let mut runtime = support::config(root.path(), &issuer, &proxy.base_url, "http://127.0.0.1:1"); @@ -148,9 +177,6 @@ async fn real_breg_action_recovers_exactly_once_and_rechecks_current_authority() CallOutcome::Uncertain { code } => assert_eq!(code, "transport-uncertain"), _ => panic!("a dropped committed response stays uncertain"), } - let read_options = BRegRecordOptions::default() - .access_profile("writer") - .unwrap(); let committed = client .get_record("items", &record, &read_options) .await diff --git a/crates/registry-coordinator/tests/casework_adapter.rs b/crates/registry-coordinator/tests/casework_adapter.rs index edfd05c18..15ddc55e1 100644 --- a/crates/registry-coordinator/tests/casework_adapter.rs +++ b/crates/registry-coordinator/tests/casework_adapter.rs @@ -100,6 +100,7 @@ fn describe(outcome: &CallOutcome) -> String { CallOutcome::Success(_) => "success".into(), CallOutcome::Retryable { code, .. } => format!("retryable {code}"), CallOutcome::Refused { code } => format!("refused {code}"), + CallOutcome::DeclaredRefusal(_) => "declared-refusal".into(), CallOutcome::Uncertain { code } => format!("uncertain {code}"), CallOutcome::ReceiptExpired => "receipt-expired".into(), } diff --git a/crates/registry-coordinator/tests/catalog_scenarios.rs b/crates/registry-coordinator/tests/catalog_scenarios.rs index 8daa34c16..fe2d69ba6 100644 --- a/crates/registry-coordinator/tests/catalog_scenarios.rs +++ b/crates/registry-coordinator/tests/catalog_scenarios.rs @@ -77,7 +77,7 @@ fn fixture(name: &str) -> (Definition, ScenarioDocument) { fn governed_action_example_covers_refusal_unknown_and_exact_command_recovery() { let (definition, document) = fixture("governed-action-follow-up"); let reports = scenarios::check(&definition, &document).unwrap(); - assert_eq!(reports.len(), 6); + assert_eq!(reports.len(), 8); assert!(matches!( &definition.workflow.steps["request-information"], registry_coordinator::definition::Step::Call {call, ..} @@ -98,6 +98,27 @@ fn governed_action_example_covers_refusal_unknown_and_exact_command_recovery() { assert_eq!(stopped.call_attempts["request-information"], 1); assert_eq!(stopped.call_attempts["notify"], 1); assert_eq!(stopped.frozen_commands, 2); + let declared = reports + .iter() + .find(|report| report.name == "action-declared-refusal") + .unwrap(); + assert_eq!(declared.state, "completed"); + assert_eq!( + declared.outcome.as_deref(), + Some("information-request-refused") + ); + assert_eq!( + declared.output, + Some(json!({"refusalCode": "already-requested"})) + ); + assert!(!declared.call_attempts.contains_key("notify")); + let still_unknown = reports + .iter() + .find(|report| report.name == "action-response-loss-then-declared-refusal-stays-unknown") + .unwrap(); + assert_eq!(still_unknown.state, "uncertain"); + assert_eq!(still_unknown.call_attempts["request-information"], 2); + assert!(!still_unknown.call_attempts.contains_key("notify")); let output = json!({"data":{"domainData":{"missingField":"proof of address"}}}); let arguments = std::collections::BTreeMap::from([("read-application".into(), output)]); @@ -113,6 +134,54 @@ fn governed_action_example_covers_refusal_unknown_and_exact_command_recovery() { assert!(request.get("idempotencyKey").is_none()); } +#[test] +fn declared_refusal_without_an_authored_edge_preserves_refused_behavior() { + let source = example("governed-action-follow-up"); + let root = tempfile::tempdir().unwrap(); + let workflow = std::fs::read_to_string(source.join("workflow.yaml")).unwrap(); + let workflow = workflow.replace(" onDeclaredRefusal: action-refused\n", ""); + let workflow = workflow.replace( + " action-refused:\n type: finish\n outcome: information-request-refused\n output:\n type: fields\n fields:\n refusalCode: {type: step, step: request-information, path: /refusalCode}\n", + "", + ); + let workflow = workflow.replace( + " information-request-refused:\n type: object\n additionalProperties: false\n required: [refusalCode]\n properties:\n refusalCode: {type: string, minLength: 1, maxLength: 128}\n", + "", + ); + assert!(!workflow.contains("onDeclaredRefusal")); + assert!(!workflow.contains("information-request-refused")); + std::fs::write(root.path().join("workflow.yaml"), workflow).unwrap(); + std::fs::copy( + source.join("functions.rhai"), + root.path().join("functions.rhai"), + ) + .unwrap(); + let definition = Definition::load(root.path()).unwrap(); + assert_eq!( + definition.explain().unwrap()["adapterAbi"], + "coordinator/product-operations/v1" + ); + let mut document = scenarios::load(&source.join("scenarios.yaml")).unwrap(); + let scenario = document + .cases + .iter_mut() + .find(|scenario| scenario.name == "action-declared-refusal") + .unwrap(); + scenario.expect.path = vec![ + "read-application".into(), + "needs-information".into(), + "request-information".into(), + ]; + scenario.expect.state = "refused".into(); + scenario.expect.outcome = None; + scenario.expect.skip_outcome_check = false; + scenario.expect.output = None; + + let report = scenarios::run(&definition, scenario).unwrap(); + assert_eq!(report.state, "refused"); + assert_eq!(report.outcome, None); +} + #[test] fn bounded_poll_example_rereads_a_fixed_number_of_times_and_returns_control() { use registry_coordinator::definition::Step; diff --git a/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json b/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json new file mode 100644 index 000000000..73c68d6dc --- /dev/null +++ b/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json @@ -0,0 +1 @@ +{"adapterAbi":"coordinator/product-operations/v1","interpreterAbi":"coordinator/pure-rhai-1.26.1/v1;raw-arithmetic-logic-string-array-map-iterator-instant;ops=100000;calls=32;expr=64;string=16384;array=256;map=256;source=65536;json=131072;depth=32","operationIdentities":[{"effect":"read","id":"read-record","keyRequirement":"none","product":"breg","readReceipt":false,"recovery":"read-again","requiresPreparation":false,"version":1},{"effect":"mutation","id":"submit-message","keyRequirement":"required","product":"messaging","readReceipt":true,"recovery":"same-command-and-receipt","requiresPreparation":false,"version":1}],"schemaAbi":"coordinator/jsonschema-0.18/draft202012/formats-asserted+uuid/v1","source":"// Pure mappings for the synthetic follow-up fixture. The source owns current\n// notice permission and contact data; the Coordinator reads both after waiting.\nfn notice_route(record) {\n if record.data.domainData.noticeAllowed { \"send\" } else { \"skip\" }\n}\n\nfn notice_request(input, record) {\n #{\n senderProfile: \"transactional\",\n to: #{ email: record.data.domainData.email },\n template: #{ id: \"application-follow-up\", version: \"1\" },\n locale: \"en\",\n data: #{ applicationReference: input.applicationId }\n }\n}\n","workflow":{"apiVersion":"id.registrystack.org/formats/coordinator/definition-snapshot/v1alpha1","connections":{"applications":"breg","notices":"messaging"},"deadline":"2592000s","functions":"functions.rhai","id":"delayed-follow-up","input":{"additionalProperties":false,"properties":{"applicationId":{"format":"uuid","type":"string"},"sendAfter":{"format":"date-time","type":"string"}},"required":["applicationId","sendAfter"],"type":"object"},"kind":"Workflow","outcomes":{"message-accepted":{"additionalProperties":false,"properties":{"messageId":{"format":"uuid","type":"string"}},"required":["messageId"],"type":"object"},"no-current-permission":{"type":"null"}},"start":"due","steps":{"accepted":{"finish":"message-accepted","output":{"fields":{"messageId":{"path":"/id","step":"notify","type":"step"}}}},"due":{"next":"read-application","waitUntil":{"path":"/sendAfter","type":"input"}},"notify":{"call":{"connection":"notices","operation":"submit-message"},"input":{"arguments":[{"type":"input"},{"step":"read-application","type":"step"}],"function":"notice_request"},"next":"accepted"},"permission":{"cases":{"send":"notify","skip":"skipped"},"choose":{"arguments":[{"step":"read-application","type":"step"}],"function":"notice_route"}},"read-application":{"call":{"connection":"applications","operation":"read-record"},"input":{"fields":{"collection":{"type":"literal","value":"applications"},"recordId":{"path":"/applicationId","type":"input"}}},"next":"permission"},"skipped":{"finish":"no-current-permission"}},"version":"1"}} \ No newline at end of file diff --git a/crates/registry-coordinator/tests/postgres_state.rs b/crates/registry-coordinator/tests/postgres_state.rs index efd26183b..28b74b127 100644 --- a/crates/registry-coordinator/tests/postgres_state.rs +++ b/crates/registry-coordinator/tests/postgres_state.rs @@ -5,7 +5,7 @@ use async_trait::async_trait; use chrono::{Duration, Utc}; use registry_coordinator::{ definition::Definition, - protocol::{AdapterSet, CallOutcome, CallRequest, Operation}, + protocol::{AdapterSet, CallOutcome, CallRequest, DeclaredRefusal, Operation}, store::{RetryBlockReason, Store}, worker::Worker, }; @@ -39,6 +39,70 @@ fn permission(record) { if record.allowed { "yes" } else { "no" } } fn message(record) { #{ senderProfile: "reminders", to: #{ email: record.contact }, content: #{ text: "Synthetic appointment reminder" } } } fn receipt(message) { #{ messageId: message.messageId } } "#; + +const DECLARED_REFUSAL_WORKFLOW: &str = r#" +apiVersion: id.registrystack.org/formats/coordinator/project/v1alpha1 +kind: CoordinatorProject +project: {id: synthetic-declared-refusal, version: v1} +input: {type: object} +connections: {source: breg} +functionsFile: functions.rhai +deadlineSeconds: 3600 +start: action +steps: + action: + type: call + connection: source + operation: invoke-breg-action + input: {type: function, function: action, arguments: [{type: input}]} + next: applied + onDeclaredRefusal: refused + applied: {type: finish, outcome: applied} + refused: + type: finish + outcome: refused + output: {type: function, function: keep, arguments: [{type: step, step: action}]} +outcomes: + applied: {type: 'null'} + refused: + type: object + additionalProperties: false + required: [refusalCode] + properties: {refusalCode: {type: string, minLength: 1, maxLength: 128}} +"#; + +const DECLARED_REFUSAL_FUNCTIONS: &str = r#" +fn action(input) { #{action: "synthetic-action", input: #{subject: input.id}} } +fn keep(value) { value } +"#; + +const DECLARED_REFUSAL_THEN_CALL_WORKFLOW: &str = r#" +apiVersion: id.registrystack.org/formats/coordinator/project/v1alpha1 +kind: CoordinatorProject +project: {id: synthetic-declared-refusal-deadline, version: v1} +input: {type: object} +connections: {source: breg} +functionsFile: functions.rhai +deadlineSeconds: 3600 +start: action +steps: + action: + type: call + connection: source + operation: invoke-breg-action + input: {type: function, function: action, arguments: [{type: input}]} + next: applied + onDeclaredRefusal: later + later: + type: call + connection: source + operation: invoke-breg-action + input: {type: function, function: action, arguments: [{type: input}]} + next: applied + applied: {type: finish, outcome: applied} +outcomes: + applied: {type: 'null'} +"#; const WORKFLOW: &str = r#" apiVersion: id.registrystack.org/formats/coordinator/project/v1alpha1 kind: CoordinatorProject @@ -115,6 +179,22 @@ async fn harness() -> Harness { project, } } + +async fn declared_refusal_harness() -> Harness { + declared_refusal_harness_with(DECLARED_REFUSAL_WORKFLOW).await +} + +async fn declared_refusal_harness_with(workflow: &str) -> Harness { + let mut h = harness().await; + std::fs::write(h.project.path().join("workflow.yaml"), workflow).unwrap(); + std::fs::write( + h.project.path().join("functions.rhai"), + DECLARED_REFUSAL_FUNCTIONS, + ) + .unwrap(); + h.definition = Definition::load(h.project.path()).expect("valid declared-refusal workflow"); + h +} fn input(at: chrono::DateTime) -> Value { json!({"at":at.to_rfc3339(), "id":"synthetic-contact-canary"}) } @@ -153,6 +233,26 @@ impl Harness { }); database_clock::retention_cutoff(&client).await } + + async fn durable_rows(&self, run: Uuid) -> String { + let (client, connection) = tokio_postgres::connect(&self.url, tokio_postgres::NoTls) + .await + .unwrap(); + tokio::spawn(async move { + connection.await.unwrap(); + }); + let row = client + .query_one( + &format!( + "SELECT row_to_json(r)::text,row_to_json(j)::text FROM {}.runs r JOIN {}.jobs j ON j.run_id=r.run_id WHERE r.run_id=$1 AND j.step=r.step", + self.namespace, self.namespace + ), + &[&run], + ) + .await + .unwrap(); + format!("{}{}", row.get::<_, String>(0), row.get::<_, String>(1)) + } } impl Harness { async fn replace_protected(&self, run: Uuid, purpose: &str, value: &T) { @@ -255,6 +355,12 @@ fn uncertain() -> CallOutcome { } } +fn declared_refusal() -> CallOutcome { + CallOutcome::DeclaredRefusal( + DeclaredRefusal::new("synthetic-not-eligible").expect("bounded synthetic refusal"), + ) +} + struct PreparedFake { preparations: std::sync::atomic::AtomicUsize, calls: Mutex)>>, @@ -306,6 +412,179 @@ impl AdapterSet for PreparedFake { } } +struct DeclaredRefusalFake { + outcomes: Mutex>, + calls: Mutex>, + entered: Notify, + release: Option, +} + +impl DeclaredRefusalFake { + fn new(outcomes: Vec) -> Arc { + Arc::new(Self { + outcomes: Mutex::new(outcomes.into()), + calls: Mutex::new(Vec::new()), + entered: Notify::new(), + release: None, + }) + } + + fn blocked(outcome: CallOutcome) -> Arc { + Arc::new(Self { + outcomes: Mutex::new(VecDeque::from([outcome])), + calls: Mutex::new(Vec::new()), + entered: Notify::new(), + release: Some(Semaphore::new(0)), + }) + } +} + +#[async_trait] +impl AdapterSet for DeclaredRefusalFake { + fn binding_digest(&self) -> &str { + "binding-a" + } + + async fn prepare(&self, _: &CallRequest) -> Result>, CallOutcome> { + Ok(Some(b"synthetic-prepared-action".to_vec())) + } + + async fn call(&self, _: &CallRequest) -> CallOutcome { + panic!("the BReg action must execute with its frozen preparation") + } + + async fn call_prepared(&self, request: &CallRequest, _: Option<&[u8]>) -> CallOutcome { + self.calls.lock().unwrap().push(request.clone()); + if let Some(release) = &self.release { + self.entered.notify_one(); + let permit = release.acquire().await.unwrap(); + permit.forget(); + } + self.outcomes.lock().unwrap().pop_front().unwrap() + } +} + +#[tokio::test] +async fn declared_refusal_is_checkpointed_and_prior_uncertainty_cannot_take_its_branch() { + let h = declared_refusal_harness().await; + let run = h.admit().await; + let fake = DeclaredRefusalFake::new(vec![declared_refusal()]); + let worker = Worker::new(h.store.clone(), fake.clone()); + assert!( + worker.tick().await.unwrap(), + "the action refusal is handled" + ); + let routed = h.store.status(run).await.unwrap(); + assert_eq!(routed.state, "running"); + assert_eq!(routed.step, "refused"); + assert!(!routed.uncertain); + assert_eq!( + h.output_values(run).await["action"], + json!({"refusalCode":"synthetic-not-eligible"}) + ); + assert!( + !h.durable_rows(run).await.contains("synthetic-not-eligible"), + "the branch code is sealed at rest" + ); + h.sql("UPDATE {schema}.runs SET deadline_at=clock_timestamp()-interval '1 second'") + .await; + drop(worker); + let reopened = Arc::new(Store::connect(&h.url, &h.namespace).await.unwrap()); + let restarted = Worker::new(reopened.clone(), fake.clone()); + assert!( + restarted.tick().await.unwrap(), + "the refusal outcome finishes after reconstruction" + ); + let finished = reopened.status(run).await.unwrap(); + assert_eq!(finished.state, "finished"); + assert_eq!(finished.outcome.as_deref(), Some("refused")); + assert_eq!( + finished.output, + Some(json!({"refusalCode":"synthetic-not-eligible"})) + ); + assert_eq!(fake.calls.lock().unwrap().len(), 1); + + let held = declared_refusal_harness().await; + let held_run = held.admit().await; + let held_fake = DeclaredRefusalFake::new(vec![uncertain(), declared_refusal()]); + let held_worker = Worker::new(held.store.clone(), held_fake.clone()); + assert!( + held_worker.tick().await.unwrap(), + "the first response is unknown" + ); + assert_eq!( + held.store.status(held_run).await.unwrap().state, + "attention" + ); + held.store.retry_same(held_run, "binding-a").await.unwrap(); + assert!( + held_worker.tick().await.unwrap(), + "a later business refusal settles no prior uncertainty" + ); + let status = held.store.status(held_run).await.unwrap(); + assert_eq!(status.state, "attention"); + assert!(status.uncertain); + assert_eq!(status.step, "action"); + assert_eq!(status.failure_code.as_deref(), Some("product-refused")); + assert!(held.output_values(held_run).await.is_empty()); + assert_eq!(held_fake.calls.lock().unwrap().len(), 2); +} + +#[tokio::test] +async fn declared_refusal_checkpoint_cannot_start_a_later_call_after_deadline() { + let h = declared_refusal_harness_with(DECLARED_REFUSAL_THEN_CALL_WORKFLOW).await; + let run = h.admit().await; + let fake = DeclaredRefusalFake::new(vec![declared_refusal()]); + let worker = Worker::new(h.store.clone(), fake.clone()); + assert!(worker.tick().await.unwrap(), "the refusal is checkpointed"); + assert_eq!(h.store.status(run).await.unwrap().step, "later"); + h.sql("UPDATE {schema}.runs SET deadline_at=clock_timestamp()-interval '1 second'") + .await; + worker.tick().await.expect("the due call expires"); + let status = h.store.status(run).await.unwrap(); + assert_eq!(status.state, "expired"); + assert_eq!(status.failure_code.as_deref(), Some("deadline-reached")); + assert_eq!(fake.calls.lock().unwrap().len(), 1); +} + +#[tokio::test] +async fn cancellation_racing_a_declared_refusal_records_no_external_effect() { + let h = declared_refusal_harness().await; + let run = h.admit().await; + let fake = DeclaredRefusalFake::blocked(declared_refusal()); + let worker = Worker::new(h.store.clone(), fake.clone()); + let completion = tokio::spawn(async move { worker.tick().await }); + tokio::time::timeout(std::time::Duration::from_secs(3), fake.entered.notified()) + .await + .expect("the action attempt reaches its adapter"); + h.store + .cancel_owned( + run, + &actor("operator", true), + "withdraw-during-declared-refusal", + ) + .await + .unwrap(); + fake.release.as_ref().unwrap().add_permits(1); + assert!(completion.await.unwrap().unwrap()); + let status = h.store.status(run).await.unwrap(); + assert_eq!(status.state, "cancelled"); + assert!(!status.uncertain); + assert_eq!(status.step, "action"); + assert_eq!( + h.output_values(run).await["action"], + json!({"refusalCode":"synthetic-not-eligible"}) + ); + assert!( + !Worker::new(h.store.clone(), fake.clone()) + .tick() + .await + .unwrap(), + "cancellation queues no refusal successor" + ); + assert_eq!(fake.calls.lock().unwrap().len(), 1); +} + impl Harness { /// Seconds from the database clock until the message step is next due. async fn message_due_in(&self, run: Uuid) -> f64 { diff --git a/crates/registry-coordinator/tests/scenario_contracts.rs b/crates/registry-coordinator/tests/scenario_contracts.rs index 0087b0d03..b8dce50f4 100644 --- a/crates/registry-coordinator/tests/scenario_contracts.rs +++ b/crates/registry-coordinator/tests/scenario_contracts.rs @@ -52,6 +52,15 @@ fn case(id: &str, application: Value, reply: Value) -> Value { }) } +fn declared_refusal_case(id: &str) -> Value { + json!({ + "id": id, "admittedAt": "2026-10-09T09:00:00Z", + "input": {"applicationId": "a-1"}, + "replies": {"read": [{"type": "declared-refusal", "refusalCode": "not-eligible"}]}, + "expect": {"path": ["read"], "state": "refused", "outcome": {"type": "none"}} + }) +} + fn scenarios(cases: Vec) -> String { // A YAML document per line keeps each case at a stable source line. let mut text = format!( @@ -175,6 +184,19 @@ fn every_synthetic_success_is_checked_even_when_the_run_stops_first() { assert_eq!(error.code, "scenario.reply-output"); } +#[test] +fn a_declared_refusal_reply_is_rejected_for_an_operation_without_that_contract() { + let error = tested(vec![declared_refusal_case("unsupported")]) + .err() + .unwrap(); + assert_eq!(error.code, "scenario.reply"); + assert_eq!(error.field.as_deref(), Some("/cases/0/replies/read/0")); + assert_eq!( + error.message, + "step read calls read-record, which cannot return a package-declared business refusal" + ); +} + #[test] fn a_missing_member_is_reported_at_the_object_that_lacks_it() { let mut reply = record(json!("v-1")); diff --git a/crates/registry-coordinator/tests/support/breg_service.rs b/crates/registry-coordinator/tests/support/breg_service.rs index 8be475630..3a6536238 100644 --- a/crates/registry-coordinator/tests/support/breg_service.rs +++ b/crates/registry-coordinator/tests/support/breg_service.rs @@ -128,6 +128,17 @@ impl Breg { root_connection.abort(); let project = service.root.join("project"); std::fs::create_dir_all(project.join("tests")).unwrap(); + std::fs::create_dir_all(project.join("scripts")).unwrap(); + crate::support::private_file( + &project.join("scripts/update-item.rhai"), + br#"fn handle(ctx) { + if ctx.inputs.label == "Refuse" { + return #{refusal: #{code: "label-not-eligible", field: "label"}}; + } + #{effects: [#{id: "item", set: #{label: ctx.inputs.label}}]} +} +"#, + ); document( &project.join("registry.yaml"), json!({ @@ -140,7 +151,9 @@ impl Breg { "actions":[{"id":"update-item", "inputs":[ {"id":"target", "apiName":"targetId", "type":"reference", "target":"item", "required":true, "classification":"internal"}, {"id":"label", "type":"string", "required":true, "minimumLength":1, "maximumLength":80, "classification":"internal"}], - "effects":[{"id":"item", "target":{"fromField":"target"}, "operation":"patch", "set":{"label":{"fromField":"label"}}}]}], + "handler":{"type":"rhai", "script":"scripts/update-item.rhai", "abi":"registry.action-handler/v1", + "refusals":[{"code":"label-not-eligible", "label":"The label is not eligible."}], + "writes":[{"id":"item", "target":{"fromField":"target"}, "operation":"patch", "fields":["label"]}]}}], "accessProfiles":[ {"id":"writer", "default":true, "principalClaim":"sub", "requiredScopes":["applications:read"], "actorKind":"service", "requesterClients":["application-reader"], "permissions":{"entities":[{"entity":"item", "operations":["create","get","list"], "readableFields":["owner","label"], "writableFields":["owner","label"], diff --git a/products/coordinator/CHANGELOG.md b/products/coordinator/CHANGELOG.md index 05193d8bd..6a6234116 100644 --- a/products/coordinator/CHANGELOG.md +++ b/products/coordinator/CHANGELOG.md @@ -24,5 +24,9 @@ First release of Registry Coordinator. - `submit-review-request` and `cancel-review-request` hand work to Casework human review as a review producer, and a `wait-for` step on a `review-result` event waits for the reviewer's result. +- `invoke-breg-action` calls may route a package-declared, no-effect BReg + action refusal through `onDeclaredRefusal`. The branch receives only its + validated `refusalCode`; operational failures and prior uncertain attempts + retain their existing failure and recovery behavior. - Add Linux amd64 `coordinator` and `coordinatorctl` binaries and a Docker image that carries both, from v0.41.0. diff --git a/products/coordinator/PRODUCT-CONTRACT.md b/products/coordinator/PRODUCT-CONTRACT.md index f60378ee1..d56c52992 100644 --- a/products/coordinator/PRODUCT-CONTRACT.md +++ b/products/coordinator/PRODUCT-CONTRACT.md @@ -78,6 +78,29 @@ ETags. Changed contracts, stale conditions and revoked authority do not silently produce a replacement command. An ambiguous response remains uncertain; a record read or operator assertion cannot establish action success. +An `invoke-breg-action` call may declare `onDeclaredRefusal`. BReg returns the +typed `422 action.refused` response only after validating the code against the +active package's declared refusal catalogue and guaranteeing that the refusal +had no action effect. The maintained adapter accepts the branch outcome only +for that exact problem class with a bounded refusal code. The branch output is +exactly `{refusalCode}`. It carries no problem detail, field path, +submitted value or raw response. The call's successful receipt remains the +output on `next`. A mapping may read the receipt only on routes reached through +`next`, and the refusal object only on routes reached through +`onDeclaredRefusal`; after those routes join, the call output is unavailable +because its shape is ambiguous. Authentication, authorization, transport, +protocol and preparation failures remain call failures. Without the optional +edge, a declared refusal also stops the run as refused. + +A declared refusal is checkpointed atomically with its branch target. If a +prior attempt has an unknown effect, a later refusal cannot settle that +attempt, so the original command remains uncertain and the refusal branch is +not taken. The original command and workflow deadline remain fixed. A +checkpointed result retains the existing rule that following pure `choose` and +`finish` steps may complete, while a later call or wait still observes expiry. +If cancellation wins before the refusal checkpoint, the worker records the +completion as no-effect and leaves the run cancelled. + `external-get` is a configured read boundary: operators bind a base URL, exact paths, allowed query parameter names, optional service authorization and a local response schema through `runtime.externalHttpConnections`. Calls can supply only @@ -240,7 +263,11 @@ resolve secret values. An explicitly supplied runtime is checked once. Scenario files use `kind: CoordinatorScenarios` and the API version `id.registrystack.org/formats/coordinator/scenarios/v1alpha1`. Each case has a unique local `id`. Replies name their `type`: `success` carries a foreign JSON `value`; `retryable`, `refused` and `uncertain` carry a `code`; -`receipt-expired` has no additional members. `waits` states what each +`declared-refusal` carries a non-empty `refusalCode` of at most 128 characters +without control characters; `receipt-expired` has no additional members. A +declared-refusal reply is accepted only for `invoke-breg-action`. It takes the +authored refusal edge, stops as refused when that edge is absent, and leaves a +prior unknown attempt uncertain. `waits` states what each `wait-for` step the run reaches receives: `{type: event, value}` delivers the event when the wait starts, and `{type: timeout}` moves the virtual clock by the step's `timeoutSeconds`. A run without `onTimeout` stops as `timed-out`, diff --git a/products/coordinator/README.md b/products/coordinator/README.md index 6440cfb4e..4b810a092 100644 --- a/products/coordinator/README.md +++ b/products/coordinator/README.md @@ -53,6 +53,14 @@ its form with `type`. A call `input` or a finish `output` is either `type: function`, a pure function with up to 16 reference `arguments`. `waitUntil` and `choose` take one reference or `type: function`. +Every call uses `next` for its successful operation output. An +`invoke-breg-action` call may also name `onDeclaredRefusal`. That edge receives +only `{refusalCode}` from a validated, package-declared, no-effect action +refusal. A step reached only through `next` can read the action receipt, and a +step reached only through `onDeclaredRefusal` can read the refusal code. At a +join reachable through both edges, the call has no single output shape, so its +output is unavailable to mappings at or beyond the join. + ```yaml read-application: type: call @@ -208,6 +216,26 @@ key are saved before dispatch. Later attempts check current metadata and acquire fresh credentials without refreshing the saved conditions or key. A changed contract, denied authority or stale condition stops the action safely. +A workflow may route a BReg action refusal only from BReg's typed +`422 action.refused` response. BReg validates the code against the active +package's declared refusal catalogue and guarantees that the refusal had no +action effect. The maintained adapter validates the exact problem class and +bounded code. The refusal edge receives that code only, not an HTTP problem, +detail, field path or submitted value. Authentication, +authorization, transport, protocol and command-preparation failures remain +ordinary call failures. Omitting `onDeclaredRefusal` preserves that behavior +for every refusal. + +The worker checkpoints a declared refusal's output and target together, as it +does a successful call result. A refusal received after an earlier uncertain +attempt cannot settle the original command, so the run remains uncertain and +does not branch. The saved command keeps its original deadline; this route does +not renew it. Once a call result is checkpointed, the existing pure-step rules +allow its following `choose` or `finish` to complete, while a later call or wait +still observes expiry. If cancellation wins before the refusal completion is +checkpointed, the no-effect completion is acknowledged and the run remains +cancelled. + `external-get` binds through `runtime.externalHttpConnections`, separately from product connections. Operators configure the base URL, exact permitted paths, query parameter names, optional service authorization and response schema. The diff --git a/products/coordinator/examples/governed-action-follow-up/README.md b/products/coordinator/examples/governed-action-follow-up/README.md index d557c9737..275836fdc 100644 --- a/products/coordinator/examples/governed-action-follow-up/README.md +++ b/products/coordinator/examples/governed-action-follow-up/README.md @@ -21,7 +21,9 @@ The run starts at once and has a `deadlineSeconds` of 300: `no-information-request-needed` and no output; otherwise it continues. 3. `request-information` calls `invoke-breg-action` with action `request-information` and two inputs: `applicationId`, and a `reason` built - from the record's `domainData.missingField`. + from the record's `domainData.missingField`. A package-declared business + refusal takes `onDeclaredRefusal` to `action-refused`, which finishes with + `information-request-refused` and the validated `refusalCode`. 4. `notify` submits one message with sender profile `transactional`, template `information-requested` version `1` and locale `en` to the record's `domainData.email`, carrying the application identifier the action returned. @@ -32,17 +34,23 @@ BReg decides whether the action runs. It checks its action metadata, the `workflow-actions` profile, the action's conditions and current authority; the action name in `functions.rhai` grants nothing. `invoke-breg-action` uses the connection's standing service identity and accepts no Casework grant, so this -connection has no `taskAuthority`. +connection has no `taskAuthority`. Only a package-declared, no-effect action +refusal takes the authored refusal edge. Authentication, authorization, +transport, protocol and preparation refusals stop the run and expose no raw +problem document to the workflow. After a lost action response the run is `uncertain`. `retry-same` resends the -saved prepared command with its original key. Reconciliation is refused for -this step: BReg has no read-only action receipt, and a record read does not -prove that the action succeeded. When Messaging refuses the notice after BReg -accepted the action, the run ends `refused` and the action stays accepted; -cancellation does not roll it back. +saved prepared command with its original key. A later declared refusal cannot +settle whether the original command ran, so that run remains `uncertain` and +does not take the refusal edge. Reconciliation is refused for this step: BReg +has no read-only action receipt, and a record read does not prove that the +action succeeded. When Messaging refuses the notice after BReg accepted the +action, the run ends `refused` and the action stays accepted; cancellation does +not roll it back. The scenarios cover an accepted action and notice, a complete application that -needs no action, a refused action, a lost action response left unresolved, a +needs no action, an authority refusal, a routed declared refusal, a declared +refusal after an uncertain attempt, a lost action response left unresolved, a lost action response recovered with `retry-same`, and a notice refused after an accepted action. They use synthetic replies only, so they do not prove BReg action policy, target preconditions, remote acceptance or provider delivery. @@ -52,7 +60,8 @@ For a deployment, the institution must supply these prerequisites: - BReg exposes a reviewed `workflow-actions` profile that reads the `applications` collection, including `complete`, `missingField` and `email` in `domainData`, and permits the immediate `request-information` action with - `applicationId` and `reason` inputs. + `applicationId` and `reason` inputs. Its package declares the business + refusal codes the workflow may receive. - The token endpoint registers the `application-agent` client for resource `urn:example:applications` with scopes `applications:read` and `applications:act`, and the `case-system` client for resource diff --git a/products/coordinator/examples/governed-action-follow-up/scenarios.yaml b/products/coordinator/examples/governed-action-follow-up/scenarios.yaml index 69804686e..cbbb205e2 100644 --- a/products/coordinator/examples/governed-action-follow-up/scenarios.yaml +++ b/products/coordinator/examples/governed-action-follow-up/scenarios.yaml @@ -61,6 +61,43 @@ cases: path: [read-application, needs-information, request-information] state: refused outcome: {type: none} +- id: action-declared-refusal + admittedAt: '2026-10-10T09:00:00Z' + input: {applicationId: 00000000-0000-4000-8000-000000000001} + replies: + read-application: + - type: success + value: + meta: {registryIdentifier: applications-registry, datasetIdentifier: applications, entityTypeIdentifier: application} + data: {recordIdentifier: 00000000-0000-4000-8000-000000000001, revisionIdentifier: revision-1, domainData: {complete: false, missingField: proof of address}} + request-information: + - type: declared-refusal + refusalCode: already-requested + expect: + path: [read-application, needs-information, request-information, action-refused] + state: completed + outcome: {type: named, value: information-request-refused} + output: {refusalCode: already-requested} +- id: action-response-loss-then-declared-refusal-stays-unknown + admittedAt: '2026-10-10T09:00:00Z' + input: {applicationId: 00000000-0000-4000-8000-000000000001} + replies: + read-application: + - type: success + value: + meta: {registryIdentifier: applications-registry, datasetIdentifier: applications, entityTypeIdentifier: application} + data: {recordIdentifier: 00000000-0000-4000-8000-000000000001, revisionIdentifier: revision-1, domainData: {complete: false, missingField: proof of address}} + request-information: + - type: uncertain + code: transport-uncertain + - type: declared-refusal + refusalCode: already-requested + recovery: + request-information: retry-same + expect: + path: [read-application, needs-information, request-information] + state: uncertain + outcome: {type: none} - id: action-response-loss-keeps-unknown admittedAt: '2026-10-10T09:00:00Z' input: {applicationId: 00000000-0000-4000-8000-000000000001} diff --git a/products/coordinator/examples/governed-action-follow-up/workflow.yaml b/products/coordinator/examples/governed-action-follow-up/workflow.yaml index 79ce1fedc..41a0929ef 100644 --- a/products/coordinator/examples/governed-action-follow-up/workflow.yaml +++ b/products/coordinator/examples/governed-action-follow-up/workflow.yaml @@ -37,6 +37,7 @@ steps: operation: invoke-breg-action input: {type: function, function: action_request, arguments: [{type: input}, {type: step, step: read-application}]} next: notify + onDeclaredRefusal: action-refused notify: type: call connection: notices @@ -54,6 +55,13 @@ steps: skipped: type: finish outcome: no-information-request-needed + action-refused: + type: finish + outcome: information-request-refused + output: + type: fields + fields: + refusalCode: {type: step, step: request-information, path: /refusalCode} outcomes: action-and-notice-accepted: type: object @@ -63,3 +71,9 @@ outcomes: actionApplicationId: {type: string, format: uuid} messageId: {type: string, format: uuid} no-information-request-needed: {type: "null"} + information-request-refused: + type: object + additionalProperties: false + required: [refusalCode] + properties: + refusalCode: {type: string, minLength: 1, maxLength: 128} diff --git a/products/coordinator/generated/operations/README.md b/products/coordinator/generated/operations/README.md index 9d8f6450f..ff408094f 100644 --- a/products/coordinator/generated/operations/README.md +++ b/products/coordinator/generated/operations/README.md @@ -1,20 +1,20 @@ # Coordinator operation reference -Each catalog operation has an input contract, the type its adapter decodes, and an output contract, the type its adapter returns as the step output. A mapping that feeds a call must produce a value its input contract accepts; a reference into a call step's output must name a path its output contract declares. - -| Operation | Version | Product | Effect | Key | Preparation | Recovery | Receipt | -|---|---|---|---|---|---|---|---| -| [`read-record`](#read-record) | 1 | `breg` | read | none | none | read-again | none | -| [`submit-message`](#submit-message) | 1 | `messaging` | mutation | required | none | same-command-and-receipt | read | -| [`read-scheduling`](#read-scheduling) | 1 | `scheduling` | read | none | none | read-again | none | -| [`read-availability`](#read-availability) | 1 | `scheduling` | read | none | none | read-again | none | -| [`create-appointment`](#create-appointment) | 1 | `scheduling` | mutation | required | none | same-command-and-receipt | read | -| [`invoke-breg-action`](#invoke-breg-action) | 1 | `breg` | mutation | required | required | same-command | none | -| [`external-get`](#external-get) | 1 | `external-http` | read | none | none | read-again | none | -| [`evaluate-decision`](#evaluate-decision) | 1 | `decision` | evaluation | none | required | hold-after-dispatch | none | -| [`submit-review-request`](#submit-review-request) | 1 | `casework` | mutation | required | none | same-command | none | -| [`cancel-review-request`](#cancel-review-request) | 1 | `casework` | mutation | required | none | same-command | none | +Each catalog operation has an input contract, the type its adapter decodes, and a success output contract, the type its adapter returns on `next`. A mapping that feeds a call must produce a value its input contract accepts; a reference into a call step's output must name a path the contract for that route declares. Only the operations marked below accept `onDeclaredRefusal`; that route receives the closed declared-refusal output and no product detail or call input. + +| Operation | Version | Product | Effect | Key | Preparation | Recovery | Receipt | Declared refusal | +|---|---|---|---|---|---|---|---|---| +| [`read-record`](#read-record) | 1 | `breg` | read | none | none | read-again | none | none | +| [`submit-message`](#submit-message) | 1 | `messaging` | mutation | required | none | same-command-and-receipt | read | none | +| [`read-scheduling`](#read-scheduling) | 1 | `scheduling` | read | none | none | read-again | none | none | +| [`read-availability`](#read-availability) | 1 | `scheduling` | read | none | none | read-again | none | none | +| [`create-appointment`](#create-appointment) | 1 | `scheduling` | mutation | required | none | same-command-and-receipt | read | none | +| [`invoke-breg-action`](#invoke-breg-action) | 1 | `breg` | mutation | required | required | same-command | none | [`onDeclaredRefusal`](invoke-breg-action.declared-refusal.schema.json) | +| [`external-get`](#external-get) | 1 | `external-http` | read | none | none | read-again | none | none | +| [`evaluate-decision`](#evaluate-decision) | 1 | `decision` | evaluation | none | required | hold-after-dispatch | none | none | +| [`submit-review-request`](#submit-review-request) | 1 | `casework` | mutation | required | none | same-command | none | none | +| [`cancel-review-request`](#cancel-review-request) | 1 | `casework` | mutation | required | none | same-command | none | none | ## read-record @@ -161,6 +161,14 @@ Output ([`invoke-breg-action.output.schema.json`](invoke-breg-action.output.sche Other members are tolerated. +Declared-refusal output ([`invoke-breg-action.declared-refusal.schema.json`](invoke-breg-action.declared-refusal.schema.json)): + +| Member | Type | Required | +|---|---|---| +| `refusalCode` | string | yes | + +No other members are accepted. + ## external-get Input ([`external-get.input.schema.json`](external-get.input.schema.json)): diff --git a/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json b/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json new file mode 100644 index 000000000..e827e54c3 --- /dev/null +++ b/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json @@ -0,0 +1,18 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "description": "One package-declared business refusal from a validated product response.\n\nThe code is deliberately absent from `Debug`; callers must opt in to reading\nor serializing it for an authored refusal branch.", + "properties": { + "refusalCode": { + "maxLength": 128, + "minLength": 1, + "pattern": "^[^\\x00-\\x1F\\x7F-\\x9F]+$", + "type": "string" + } + }, + "required": [ + "refusalCode" + ], + "title": "invoke-breg-action declared-refusal output", + "type": "object" +} diff --git a/products/coordinator/generated/project/project.schema.json b/products/coordinator/generated/project/project.schema.json index 81bcc9973..a21c6b8e9 100644 --- a/products/coordinator/generated/project/project.schema.json +++ b/products/coordinator/generated/project/project.schema.json @@ -247,6 +247,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -279,6 +284,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -317,6 +327,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -343,6 +358,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -379,6 +399,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -441,6 +466,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -471,6 +501,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -502,6 +537,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -539,6 +579,11 @@ ] }, "then": { + "not": { + "required": [ + "onDeclaredRefusal" + ] + }, "properties": { "input": { "properties": { @@ -559,7 +604,7 @@ } } ], - "description": "Calls one operation through a connection, then continues with its\noutput available to later steps.", + "description": "Calls one operation through a connection, then continues with its\nsuccessful output available to later steps.", "properties": { "connection": { "$ref": "#/$defs/LocalId", @@ -573,6 +618,10 @@ "$ref": "#/$defs/LocalId", "description": "The step that follows a completed call." }, + "onDeclaredRefusal": { + "$ref": "#/$defs/LocalId", + "description": "The step that follows a product-declared, no-effect business\nrefusal. Only operations whose reference declares this capability\naccept the member; other refusals still stop the run." + }, "operation": { "$ref": "#/$defs/Operation", "description": "The operation to call; it must belong to the connection's product." diff --git a/products/coordinator/generated/scenarios/scenarios.schema.json b/products/coordinator/generated/scenarios/scenarios.schema.json index ff73e6088..2d86d3be3 100644 --- a/products/coordinator/generated/scenarios/scenarios.schema.json +++ b/products/coordinator/generated/scenarios/scenarios.schema.json @@ -100,6 +100,25 @@ ], "type": "object" }, + { + "additionalProperties": false, + "description": "The product returned a package-declared, no-effect business refusal.\nA supported call takes its `onDeclaredRefusal` edge when it has one;\notherwise the run stops as refused.", + "properties": { + "refusalCode": { + "$ref": "#/$defs/DeclaredRefusalCode", + "description": "The validated refusal code the package declares." + }, + "type": { + "const": "declared-refusal", + "type": "string" + } + }, + "required": [ + "type", + "refusalCode" + ], + "type": "object" + }, { "additionalProperties": false, "description": "The call's effect is unknown; the run stops as uncertain unless a\nrecovery consumes another reply.", @@ -241,6 +260,12 @@ } ] }, + "DeclaredRefusalCode": { + "maxLength": 128, + "minLength": 1, + "pattern": "^[^\\u0000-\\u001F\\u007F-\\u009F]+$", + "type": "string" + }, "ExpectedOutcome": { "oneOf": [ { From cc9d727375344b709195ff6c56cdfb63c33aac99 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 10:50:37 +0700 Subject: [PATCH 2/6] fix(client): export Coordinator run-list query types Signed-off-by: Jeremi Joslin --- crates/registry-stack-client/src/lib.rs | 3 ++- crates/registry-stack-client/tests/coordinator_facade.rs | 6 +++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/registry-stack-client/src/lib.rs b/crates/registry-stack-client/src/lib.rs index 29e73c33e..b21cc2ab1 100644 --- a/crates/registry-stack-client/src/lib.rs +++ b/crates/registry-stack-client/src/lib.rs @@ -75,7 +75,8 @@ pub mod coordinator { BearerToken, CoordinatorClient, CoordinatorClientConfig, CoordinatorClientError, CoordinatorComplete, CoordinatorProtocolFailure, EffectKind, KeyRequirement, OperationIdentity, RecoverySemantics, RecoveryStatus, RetryBlockReason, RetryScheduled, - RunInspection, RunList, RunStatus, StartRunRequest, StepStatus, TransportKind, Uuid, + RunInspection, RunList, RunListQuery, RunListState, RunStatus, StartRunRequest, StepStatus, + TransportKind, Uuid, }; } diff --git a/crates/registry-stack-client/tests/coordinator_facade.rs b/crates/registry-stack-client/tests/coordinator_facade.rs index 14b87a650..44db21027 100644 --- a/crates/registry-stack-client/tests/coordinator_facade.rs +++ b/crates/registry-stack-client/tests/coordinator_facade.rs @@ -13,7 +13,8 @@ use std::collections::BTreeSet; use registry_stack_client::coordinator::{ BearerToken, CoordinatorClient, CoordinatorClientConfig, CoordinatorClientError, - CoordinatorComplete, RetryScheduled, RunInspection, RunList, RunStatus, StartRunRequest, Uuid, + CoordinatorComplete, RetryScheduled, RunInspection, RunList, RunListQuery, RunListState, + RunStatus, StartRunRequest, Uuid, }; /// The client's public surface, read from the crate that publishes it. @@ -30,6 +31,7 @@ async fn every_coordinator_method_names_its_types( idempotency_key: &str, run: Uuid, request: &StartRunRequest, + list_query: &RunListQuery, ) -> Result<(), CoordinatorClientError> { let client: CoordinatorClient = CoordinatorClient::new(config)?; let _: CoordinatorComplete = client.start(token, idempotency_key, request).await?; @@ -39,6 +41,8 @@ async fn every_coordinator_method_names_its_types( .reconcile(token, run, "operator reviewed the original receipt") .await?; let _: CoordinatorComplete = client.list(token, 20).await?; + let _: CoordinatorComplete = client.list_page(token, list_query).await?; + let _: Option = list_query.state; let _: CoordinatorComplete = client .retry_same(token, run, "operator reviewed the inspection") .await?; From ec3f99d35da192448aeb4a175508b546fe16e65a Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:21:56 +0700 Subject: [PATCH 3/6] refactor(coordinator): keep one adapter ABI for declared refusals Registry Coordinator has never been released, so a second adapter ABI selected by the presence of onDeclaredRefusal guards no deployed snapshot. The edge is an optional member of the frozen call step: a snapshot without it keeps its bytes, digest and behavior, and the closed step decoder refuses a member it does not know instead of dropping it. Restore still refuses the edge on an operation that does not declare the capability, now held by its own snapshot test. Signed-off-by: Jeremi Joslin --- crates/registry-coordinator/src/definition.rs | 27 +------ crates/registry-coordinator/src/operations.rs | 12 +-- .../tests/authored_project.rs | 75 +++++++++++++------ .../registry-coordinator/tests/authoring.rs | 25 ------- .../tests/catalog_scenarios.rs | 4 - .../tests/fixtures/legacy-definition-v1.json | 1 - 6 files changed, 63 insertions(+), 81 deletions(-) delete mode 100644 crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json diff --git a/crates/registry-coordinator/src/definition.rs b/crates/registry-coordinator/src/definition.rs index b8b73557d..4efbfb06c 100644 --- a/crates/registry-coordinator/src/definition.rs +++ b/crates/registry-coordinator/src/definition.rs @@ -25,8 +25,7 @@ use crate::{ /// The compiled snapshot format, carried by its workflow. pub(crate) const API_VERSION: &str = "id.registrystack.org/formats/coordinator/definition-snapshot/v1alpha1"; -const ADAPTER_ABI_V1: &str = "coordinator/product-operations/v1"; -const ADAPTER_ABI_V2: &str = "coordinator/product-operations/v2"; +const ADAPTER_ABI: &str = "coordinator/product-operations/v1"; const SCHEMA_ABI: &str = "coordinator/jsonschema-0.18/draft202012/formats-asserted+uuid/v1"; const MAX_DOCUMENT_BYTES: usize = 1_048_576; /// Maximum canonical snapshot size, shared by authoring, restore and packaging. @@ -291,25 +290,6 @@ fn operation_identities(workflow: &Workflow) -> Vec { identities } -/// Workflows that do not opt into declared-refusal routing retain the exact v1 -/// adapter ABI. The v2 ABI pins both the classification and the branch output -/// shape for workflows that use `onDeclaredRefusal`. -fn adapter_abi(workflow: &Workflow) -> &'static str { - if workflow.steps.values().any(|step| { - matches!( - step, - Step::Call { - on_declared_refusal: Some(_), - .. - } - ) - }) { - ADAPTER_ABI_V2 - } else { - ADAPTER_ABI_V1 - } -} - /// Compiled definitions keyed by the digest of their snapshot text, so one /// worker serving several flows compiles each pinned or in-flight snapshot once. /// @@ -397,12 +377,11 @@ impl Definition { let exit = error.exit_code; crate::authoring::positioned(&document, error.at(&path, "functions")).with_exit(exit) })?; - let adapter_abi = adapter_abi(&workflow).into(); let definition = Self::restore_checked(Snapshot { operation_identities: operation_identities(&workflow), workflow, source, - adapter_abi, + adapter_abi: ADAPTER_ABI.into(), interpreter_abi: INTERPRETER_ABI.into(), schema_abi: SCHEMA_ABI.into(), }) @@ -444,7 +423,7 @@ impl Definition { .iter() .all(OperationIdentity::matches_registered); if !operations_match - || frozen.adapter_abi != adapter_abi(&frozen.workflow) + || frozen.adapter_abi != ADAPTER_ABI || frozen.interpreter_abi != INTERPRETER_ABI || frozen.schema_abi != SCHEMA_ABI { diff --git a/crates/registry-coordinator/src/operations.rs b/crates/registry-coordinator/src/operations.rs index 1da2f067e..302ae4df5 100644 --- a/crates/registry-coordinator/src/operations.rs +++ b/crates/registry-coordinator/src/operations.rs @@ -23,8 +23,7 @@ pub enum KeyRequirement { /// Binding behavior is versioned independently of the workflow. The fields /// copied into [`OperationIdentity`] must not silently change for an admitted -/// run; authoring capabilities outside that identity are pinned by their -/// workflow ABI. +/// run. `declared_refusal` is an authoring capability outside that identity. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct OperationDescriptor { pub id: &'static str, @@ -35,11 +34,12 @@ pub struct OperationDescriptor { pub requires_preparation: bool, pub recovery: RecoverySemantics, pub read_receipt: bool, - /// Whether an adapter ABI v2 workflow may route a product-declared, - /// no-effect business refusal through `onDeclaredRefusal`. + /// Whether a call may route a product-declared, no-effect business + /// refusal through `onDeclaredRefusal`. /// - /// This capability is pinned by the workflow's adapter ABI rather than by - /// [`OperationIdentity`], so adding it does not invalidate v1 snapshots. + /// The capability is outside [`OperationIdentity`]. A snapshot carries the + /// authored edge itself, and restoring it refuses an edge on an operation + /// that does not declare the capability. pub declared_refusal: bool, } diff --git a/crates/registry-coordinator/tests/authored_project.rs b/crates/registry-coordinator/tests/authored_project.rs index 11aa9180e..62de50f8e 100644 --- a/crates/registry-coordinator/tests/authored_project.rs +++ b/crates/registry-coordinator/tests/authored_project.rs @@ -79,14 +79,14 @@ fn authored_project_converts_to_the_versioned_snapshot_abi() { } #[test] -fn declared_refusal_opt_in_uses_v2_and_restores_only_with_its_shape() { +fn declared_refusal_edge_is_frozen_in_the_snapshot_and_restores_exactly() { let root = tempfile::tempdir().unwrap(); let project = declared_refusal_project(); write_project(root.path(), &project, "fn command(input) { input }"); let definition = Definition::load(root.path()).unwrap(); let snapshot = definition.snapshot().unwrap(); let frozen: Value = serde_json::from_str(&snapshot).unwrap(); - assert_eq!(frozen["adapterAbi"], "coordinator/product-operations/v2"); + assert_eq!(frozen["adapterAbi"], "coordinator/product-operations/v1"); assert_eq!( frozen["workflow"]["steps"]["act"]["onDeclaredRefusal"], "refused" @@ -107,26 +107,31 @@ fn declared_refusal_opt_in_uses_v2_and_restores_only_with_its_shape() { snapshot ); - for changed in [ - { - let mut changed = frozen.clone(); - changed["adapterAbi"] = json!("coordinator/product-operations/v1"); - changed - }, - { - let mut changed = frozen.clone(); - changed["workflow"]["steps"]["act"] - .as_object_mut() - .unwrap() - .remove("onDeclaredRefusal"); - changed - }, - ] { - let error = Definition::from_snapshot(&changed.to_string()) + // The step decoder is closed: an edge it does not know is refused, never + // dropped, so a snapshot cannot lose its refusal route on restore. + let mut unknown_edge = frozen.clone(); + let act = unknown_edge["workflow"]["steps"]["act"] + .as_object_mut() + .unwrap(); + let target = act.remove("onDeclaredRefusal").unwrap(); + act.insert("onUndeclaredEdge".into(), target); + assert_eq!( + Definition::from_snapshot(&unknown_edge.to_string()) .err() - .expect("ABI and feature shape must agree"); - assert_eq!(error.code, "definition.abi"); - } + .expect("an unknown step member must refuse") + .code, + "definition.snapshot" + ); + + let mut other_abi = frozen; + other_abi["adapterAbi"] = json!("coordinator/product-operations/v2"); + assert_eq!( + Definition::from_snapshot(&other_abi.to_string()) + .err() + .expect("another adapter ABI must refuse") + .code, + "definition.abi" + ); } #[test] @@ -189,6 +194,34 @@ fn operations_without_a_declared_refusal_contract_reject_the_edge() { assert_eq!(error.field.as_deref(), Some("/steps/act/onDeclaredRefusal")); } +#[test] +fn a_snapshot_cannot_carry_the_refusal_edge_on_an_operation_without_it() { + let root = tempfile::tempdir().unwrap(); + let mut project = declared_refusal_project(); + project["steps"]["act"]["operation"] = json!("read-record"); + let steps = project["steps"].as_object_mut().unwrap(); + steps["act"] + .as_object_mut() + .unwrap() + .remove("onDeclaredRefusal"); + steps.remove("refused"); + project["outcomes"] + .as_object_mut() + .unwrap() + .remove("refused"); + write_project(root.path(), &project, "fn command(input) { input }"); + let snapshot = Definition::load(root.path()).unwrap().snapshot().unwrap(); + let mut frozen: Value = serde_json::from_str(&snapshot).unwrap(); + frozen["workflow"]["steps"]["act"]["onDeclaredRefusal"] = json!("accepted"); + assert_eq!( + Definition::from_snapshot(&frozen.to_string()) + .err() + .expect("the restored edge must refuse") + .code, + "definition.operation" + ); +} + #[test] fn function_compile_diagnostics_keep_the_function_file_and_position() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/registry-coordinator/tests/authoring.rs b/crates/registry-coordinator/tests/authoring.rs index 13b90c6b6..c680b055f 100644 --- a/crates/registry-coordinator/tests/authoring.rs +++ b/crates/registry-coordinator/tests/authoring.rs @@ -188,31 +188,6 @@ fn snapshot_restores_exact_functions_after_project_changes_or_removal() { ); } -#[test] -fn legacy_v1_snapshot_restores_with_exact_bytes_digest_and_operation_identities() { - const SNAPSHOT: &str = include_str!("fixtures/legacy-definition-v1.json"); - let definition = Definition::from_snapshot(SNAPSHOT).unwrap(); - assert_eq!(definition.snapshot().unwrap(), SNAPSHOT); - assert_eq!( - definition.digest, - "sha256:7298d1fa649b7e0630cd9131564ef0e9ee0c24598a7670a5c8dcd55c5946fdcc" - ); - let frozen: Value = serde_json::from_str(SNAPSHOT).unwrap(); - assert_eq!(frozen["adapterAbi"], "coordinator/product-operations/v1"); - assert!(SNAPSHOT.contains("\"operationIdentities\"")); - assert!(!SNAPSHOT.contains("onDeclaredRefusal")); - assert!(frozen["operationIdentities"] - .as_array() - .unwrap() - .iter() - .all(|identity| identity.get("declaredRefusal").is_none())); - assert!(definition.explain().unwrap()["steps"] - .as_array() - .unwrap() - .iter() - .all(|step| step.get("onDeclaredRefusal").is_none())); -} - #[test] fn definition_cache_compiles_each_snapshot_once_within_its_bound() { use registry_coordinator::definition::DefinitionCache; diff --git a/crates/registry-coordinator/tests/catalog_scenarios.rs b/crates/registry-coordinator/tests/catalog_scenarios.rs index fb8aa6d4c..050e99139 100644 --- a/crates/registry-coordinator/tests/catalog_scenarios.rs +++ b/crates/registry-coordinator/tests/catalog_scenarios.rs @@ -157,10 +157,6 @@ fn declared_refusal_without_an_authored_edge_preserves_refused_behavior() { ) .unwrap(); let definition = Definition::load(root.path()).unwrap(); - assert_eq!( - definition.explain().unwrap()["adapterAbi"], - "coordinator/product-operations/v1" - ); let mut document = scenarios::load(&source.join("scenarios.yaml")).unwrap(); let scenario = document .cases diff --git a/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json b/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json deleted file mode 100644 index 73c68d6dc..000000000 --- a/crates/registry-coordinator/tests/fixtures/legacy-definition-v1.json +++ /dev/null @@ -1 +0,0 @@ -{"adapterAbi":"coordinator/product-operations/v1","interpreterAbi":"coordinator/pure-rhai-1.26.1/v1;raw-arithmetic-logic-string-array-map-iterator-instant;ops=100000;calls=32;expr=64;string=16384;array=256;map=256;source=65536;json=131072;depth=32","operationIdentities":[{"effect":"read","id":"read-record","keyRequirement":"none","product":"breg","readReceipt":false,"recovery":"read-again","requiresPreparation":false,"version":1},{"effect":"mutation","id":"submit-message","keyRequirement":"required","product":"messaging","readReceipt":true,"recovery":"same-command-and-receipt","requiresPreparation":false,"version":1}],"schemaAbi":"coordinator/jsonschema-0.18/draft202012/formats-asserted+uuid/v1","source":"// Pure mappings for the synthetic follow-up fixture. The source owns current\n// notice permission and contact data; the Coordinator reads both after waiting.\nfn notice_route(record) {\n if record.data.domainData.noticeAllowed { \"send\" } else { \"skip\" }\n}\n\nfn notice_request(input, record) {\n #{\n senderProfile: \"transactional\",\n to: #{ email: record.data.domainData.email },\n template: #{ id: \"application-follow-up\", version: \"1\" },\n locale: \"en\",\n data: #{ applicationReference: input.applicationId }\n }\n}\n","workflow":{"apiVersion":"id.registrystack.org/formats/coordinator/definition-snapshot/v1alpha1","connections":{"applications":"breg","notices":"messaging"},"deadline":"2592000s","functions":"functions.rhai","id":"delayed-follow-up","input":{"additionalProperties":false,"properties":{"applicationId":{"format":"uuid","type":"string"},"sendAfter":{"format":"date-time","type":"string"}},"required":["applicationId","sendAfter"],"type":"object"},"kind":"Workflow","outcomes":{"message-accepted":{"additionalProperties":false,"properties":{"messageId":{"format":"uuid","type":"string"}},"required":["messageId"],"type":"object"},"no-current-permission":{"type":"null"}},"start":"due","steps":{"accepted":{"finish":"message-accepted","output":{"fields":{"messageId":{"path":"/id","step":"notify","type":"step"}}}},"due":{"next":"read-application","waitUntil":{"path":"/sendAfter","type":"input"}},"notify":{"call":{"connection":"notices","operation":"submit-message"},"input":{"arguments":[{"type":"input"},{"step":"read-application","type":"step"}],"function":"notice_request"},"next":"accepted"},"permission":{"cases":{"send":"notify","skip":"skipped"},"choose":{"arguments":[{"step":"read-application","type":"step"}],"function":"notice_route"}},"read-application":{"call":{"connection":"applications","operation":"read-record"},"input":{"fields":{"collection":{"type":"literal","value":"applications"},"recordId":{"path":"/applicationId","type":"input"}}},"next":"permission"},"skipped":{"finish":"no-current-permission"}},"version":"1"}} \ No newline at end of file From aec6872165a669373d2190441ad7310853bf45bf Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:21:57 +0700 Subject: [PATCH 4/6] docs(coordinator): state what omitting onDeclaredRefusal means An optional member with no default says in its description what leaving it out does. Without the edge a declared refusal fails the run with product-refused, like every other refusal. Signed-off-by: Jeremi Joslin --- crates/registry-coordinator/src/authoring.rs | 3 ++- products/coordinator/generated/project/project.schema.json | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/registry-coordinator/src/authoring.rs b/crates/registry-coordinator/src/authoring.rs index fea973953..53bcb0b60 100644 --- a/crates/registry-coordinator/src/authoring.rs +++ b/crates/registry-coordinator/src/authoring.rs @@ -405,7 +405,8 @@ enum AuthoredStep { next: LocalId, /// The step that follows a product-declared, no-effect business /// refusal. Only operations whose reference declares this capability - /// accept the member; other refusals still stop the run. + /// accept the member; other refusals still stop the run. Omit it to + /// fail the run with `product-refused` on a declared refusal too. #[serde(default)] on_declared_refusal: Option, }, diff --git a/products/coordinator/generated/project/project.schema.json b/products/coordinator/generated/project/project.schema.json index 0e7a5da67..fbc921884 100644 --- a/products/coordinator/generated/project/project.schema.json +++ b/products/coordinator/generated/project/project.schema.json @@ -624,7 +624,7 @@ }, "onDeclaredRefusal": { "$ref": "#/$defs/LocalId", - "description": "The step that follows a product-declared, no-effect business\nrefusal. Only operations whose reference declares this capability\naccept the member; other refusals still stop the run." + "description": "The step that follows a product-declared, no-effect business\nrefusal. Only operations whose reference declares this capability\naccept the member; other refusals still stop the run. Omit it to\nfail the run with `product-refused` on a declared refusal too." }, "operation": { "$ref": "#/$defs/Operation", From cf972ba4ebfbddb2c41b8267a02106d924d2a708 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:21:57 +0700 Subject: [PATCH 5/6] test(coordinator): hold malformed action refusals uncertain A server error that carries an action.refused body, and a 422 action.refused without a refusal code, are outside the typed contract the refusal route trusts. Both leave the dispatched command uncertain and never reach the authored branch. Signed-off-by: Jeremi Joslin --- .../tests/breg_action_adapter.rs | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/crates/registry-coordinator/tests/breg_action_adapter.rs b/crates/registry-coordinator/tests/breg_action_adapter.rs index f9c822b1b..462f1d422 100644 --- a/crates/registry-coordinator/tests/breg_action_adapter.rs +++ b/crates/registry-coordinator/tests/breg_action_adapter.rs @@ -367,6 +367,51 @@ async fn typed_action_refused_problem_exposes_only_its_bounded_code() { assert!(!debug.contains("/input/label")); } +#[tokio::test] +async fn action_refused_shaped_problems_outside_the_typed_contract_stay_uncertain() { + let server_error = json!({ + "type":"https://id.registrystack.org/problems/registry-breg/action/refused", + "title":"Internal Server Error", + "status":500, + "code":"action.refused", + "detail":"Synthetic private label that must not cross the adapter.", + "traceId":"4bf92f3577b34da6a3ce929d0e0e4736", + "refusalCode":"label-not-eligible" + }); + let missing_refusal_code = json!({ + "type":"https://id.registrystack.org/problems/registry-breg/action/refused", + "title":"Unprocessable Entity", + "status":422, + "code":"action.refused", + "detail":"Synthetic private label that must not cross the adapter.", + "traceId":"4bf92f3577b34da6a3ce929d0e0e4736" + }); + for (status, body) in [(500, server_error), (422, missing_refusal_code)] { + let server = MockServer::start().await; + authority(&server).await; + Mock::given(method("POST")) + .and(path("/tenant/base/v1/actions/update-item")) + .respond_with( + ResponseTemplate::new(status) + .insert_header("traceparent", TRACE) + .insert_header("cache-control", "no-store") + .set_body_raw(body.to_string(), "application/problem+json"), + ) + .expect(1) + .mount(&server) + .await; + let client = client(&server, Arc::new(CountingToken::default())); + let command = request(); + let prepared = prepare(&client, "writer", &command) + .await + .unwrap_or_else(|_| panic!("valid preparation")); + match execute(&client, "writer", &command, &prepared).await { + CallOutcome::Uncertain { code } => assert_eq!(code, "transport-uncertain"), + _ => panic!("a refusal the typed contract does not cover must stay uncertain"), + } + } +} + #[tokio::test] async fn expired_receipt_and_known_rollbacks_remain_distinct_from_unknown_effects() { for code in [ From 782e788b59ad728672634758ef8270e3b1dfbdb4 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Tue, 13 Oct 2026 16:21:57 +0700 Subject: [PATCH 6/6] refactor(coordinator): trim unused declared-refusal plumbing The adapter classifies a refusal only after an unknown outcome has already returned, so the classifier needs no flag for it. No runtime path validates a value against the declared-refusal contract, so the contract keeps the schema and drops the compiled validator and its check. The generated reference links the refusal schema by operation identifier, and the refusal code pattern spells its control-character bound the way the other Coordinator schemas do. Signed-off-by: Jeremi Joslin --- .../registry-coordinator/src/breg_action.rs | 74 ++++--------------- crates/registry-coordinator/src/contracts.rs | 49 ++++++------ crates/registry-coordinator/src/protocol.rs | 2 +- ...e-breg-action.declared-refusal.schema.json | 2 +- 4 files changed, 36 insertions(+), 91 deletions(-) diff --git a/crates/registry-coordinator/src/breg_action.rs b/crates/registry-coordinator/src/breg_action.rs index 88d459cdb..5be8aef5c 100644 --- a/crates/registry-coordinator/src/breg_action.rs +++ b/crates/registry-coordinator/src/breg_action.rs @@ -45,15 +45,11 @@ fn declared_refusal( status: Option, problem: Option, refusal_code: Option<&str>, - outcome_unknown: bool, ) -> Option { - (mutation - && !outcome_unknown - && status == Some(422) - && problem == Some(BRegProblemCode::ActionRefused)) - .then_some(refusal_code) - .flatten() - .and_then(DeclaredRefusal::new) + (mutation && status == Some(422) && problem == Some(BRegProblemCode::ActionRefused)) + .then_some(refusal_code) + .flatten() + .and_then(DeclaredRefusal::new) } fn failure(error: BaseRegistryClientError, mutation: bool) -> CallOutcome { @@ -70,7 +66,6 @@ fn failure(error: BaseRegistryClientError, mutation: bool) -> CallOutcome { error.status(), error.problem_code(), error.refusal_code().map(|code| code.as_str()), - false, ) { return CallOutcome::DeclaredRefusal(refusal); } @@ -198,70 +193,27 @@ mod tests { #[test] fn declared_refusal_classification_requires_every_typed_execution_guard() { - let classify = |mutation, status, problem, code, unknown| { - declared_refusal(mutation, status, problem, code, unknown) + let classify = |mutation, status, problem, code| { + declared_refusal(mutation, status, problem, code) .map(|refusal| refusal.refusal_code().to_owned()) }; + let refused = Some(BRegProblemCode::ActionRefused); assert_eq!( - classify( - true, - Some(422), - Some(BRegProblemCode::ActionRefused), - Some("not-eligible"), - false, - ), + classify(true, Some(422), refused, Some("not-eligible")), Some("not-eligible".to_owned()) ); for refusal in [ - classify( - false, - Some(422), - Some(BRegProblemCode::ActionRefused), - Some("not-eligible"), - false, - ), - classify( - true, - Some(409), - Some(BRegProblemCode::ActionRefused), - Some("not-eligible"), - false, - ), + classify(false, Some(422), refused, Some("not-eligible")), + classify(true, Some(409), refused, Some("not-eligible")), classify( true, Some(422), Some(BRegProblemCode::PreconditionFailed), Some("not-eligible"), - false, - ), - classify( - true, - Some(422), - Some(BRegProblemCode::ActionRefused), - None, - false, - ), - classify( - true, - Some(422), - Some(BRegProblemCode::ActionRefused), - Some("line\nbreak"), - false, - ), - classify( - true, - Some(422), - Some(BRegProblemCode::ActionRefused), - Some(&"x".repeat(129)), - false, - ), - classify( - true, - Some(422), - Some(BRegProblemCode::ActionRefused), - Some("not-eligible"), - true, ), + classify(true, Some(422), refused, None), + classify(true, Some(422), refused, Some("line\nbreak")), + classify(true, Some(422), refused, Some(&"x".repeat(129))), ] { assert_eq!(refusal, None); } diff --git a/crates/registry-coordinator/src/contracts.rs b/crates/registry-coordinator/src/contracts.rs index 755d28477..d67250491 100644 --- a/crates/registry-coordinator/src/contracts.rs +++ b/crates/registry-coordinator/src/contracts.rs @@ -22,7 +22,6 @@ pub struct OperationContract { pub declared_refusal: Option, input_validator: JSONSchema, output_validator: JSONSchema, - declared_refusal_validator: Option, } /// Where a value leaves a contract. The path is an RFC 6901 pointer into the @@ -60,17 +59,6 @@ impl OperationContract { pub fn check_output(&self, value: &Value) -> Result<(), Violation> { first_violation(&self.output_validator, value) } - - /// Checks a product-declared refusal before it becomes a step output. - pub fn check_declared_refusal(&self, value: &Value) -> Result<(), Violation> { - let Some(validator) = &self.declared_refusal_validator else { - return Err(Violation { - path: String::new(), - expected: "an operation that declares a routable business refusal".into(), - }); - }; - first_violation(validator, value) - } } pub(crate) fn first_violation(validator: &JSONSchema, value: &Value) -> Result<(), Violation> { @@ -244,11 +232,12 @@ pub fn contract(operation: Operation) -> &'static OperationContract { }; let (input, input_validator) = parse(input); let (output, output_validator) = parse(output); - let (declared_refusal, declared_refusal_validator) = DECLARED_REFUSALS + let declared_refusal = DECLARED_REFUSALS .iter() .find(|(operation, _)| operation == id) - .map(|(_, schema)| parse(schema)) - .unzip(); + .map(|(_, schema)| { + serde_json::from_str(schema).expect("committed operation contract is JSON") + }); ( *id, OperationContract { @@ -257,7 +246,6 @@ pub fn contract(operation: Operation) -> &'static OperationContract { input_validator, output_validator, declared_refusal, - declared_refusal_validator, }, ) }) @@ -577,9 +565,12 @@ mod derived { recovery = word(&descriptor.recovery)?, receipt = if descriptor.read_receipt { "read" } else { "none" }, declared_refusal = if descriptor.declared_refusal { - "[`onDeclaredRefusal`](invoke-breg-action.declared-refusal.schema.json)" + format!( + "[`onDeclaredRefusal`]({}.declared-refusal.schema.json)", + descriptor.id + ) } else { - "none" + "none".into() }, )); } @@ -855,19 +846,21 @@ mod tests { descriptor.id ); } - let action = contract(Operation::InvokeBregAction); + let refusal = contract(Operation::InvokeBregAction) + .declared_refusal + .as_ref() + .unwrap(); + let validator = crate::definition::compile_schema(refusal) + .unwrap_or_else(|error| panic!("declared-refusal contract does not compile: {error}")); assert_eq!( - action.check_declared_refusal(&json!({"refusalCode":"not-eligible"})), + first_violation(&validator, &json!({"refusalCode":"not-eligible"})), Ok(()) ); - assert!(action - .check_declared_refusal(&json!({ - "refusalCode":"not-eligible", "detail":"private-canary" - })) - .is_err()); - assert!(contract(Operation::ReadRecord) - .check_declared_refusal(&json!({"refusalCode":"not-eligible"})) - .is_err()); + assert!(first_violation( + &validator, + &json!({"refusalCode":"not-eligible", "detail":"private-canary"}) + ) + .is_err()); } #[test] diff --git a/crates/registry-coordinator/src/protocol.rs b/crates/registry-coordinator/src/protocol.rs index b5443abb6..fe2a2395f 100644 --- a/crates/registry-coordinator/src/protocol.rs +++ b/crates/registry-coordinator/src/protocol.rs @@ -155,7 +155,7 @@ pub struct DeclaredRefusal { feature = "schema", schemars( length(min = 1, max = 128), - regex(pattern = r"^[^\x00-\x1F\x7F-\x9F]+$") + regex(pattern = r"^[^\u0000-\u001F\u007F-\u009F]+$") ) )] refusal_code: String, diff --git a/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json b/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json index e827e54c3..b63ba8ec1 100644 --- a/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json +++ b/products/coordinator/generated/operations/invoke-breg-action.declared-refusal.schema.json @@ -6,7 +6,7 @@ "refusalCode": { "maxLength": 128, "minLength": 1, - "pattern": "^[^\\x00-\\x1F\\x7F-\\x9F]+$", + "pattern": "^[^\\u0000-\\u001F\\u007F-\\u009F]+$", "type": "string" } },