feat(control): publish authenticated workflow state and apply route

crates/contextful-agent/src/http.rs

@@ -15,6 +15,7 @@
 use crate::mcp::{build_identity, Caller, ReadRecord, Tools};
 use contextful_context::read::Face;
 use contextful_core::ports::Clock;
+use contextful_core::grant::{Action, TablePattern};
 use contextful_core::read::{ReadError, Refusal};
 use contextful_core::AuthorityError;
 use contextful_policy::enforce::refuse::payload;
@@ -41,6 +42,10 @@ pub const MCP_PATH: &str = "/mcp";
 /// The readiness route (`read.register.health`).
 pub const HEALTH_PATH: &str = "/health";
 
+/// The store-published workflow state and validated control claim routes.
+pub const WORKFLOWS_PATH: &str = "/control/workflows";
+pub const APPLY_PATH: &str = "/control/apply";
+
 /// Largest request head the face reads; a head past it answers `431`.
 const REQUEST_HEAD_BYTES: usize = 64 * 1024;
 
@@ -193,6 +198,7 @@ pub struct Admitting<'a, C> {
 pub struct HttpFace<'a, C> {
     tools: Tools<'a>,
     admitting: Admitting<'a, C>,
+    control: Option<&'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)>,
     ceiling: usize,
     in_flight: AtomicUsize,
     exchange: Option<&'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)>,
@@ -243,7 +249,7 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
         let ceiling = ceiling(max_in_flight)?;
         audience(Some(admitting.audience))?;
         let tools = Tools::new(face, clock, record)?;
-        Ok(HttpFace { tools, admitting, ceiling, in_flight: AtomicUsize::new(0), exchange: None, exchange_unconfigured: false })
+        Ok(HttpFace { tools, admitting, control: None, ceiling, in_flight: AtomicUsize::new(0), exchange: None, exchange_unconfigured: false })
     }
 
     /// The binary's exchange route mints the reader credential without putting issuer
@@ -254,6 +260,12 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
         self
     }
 
+    /// Attach the store's control API without adding tools to the closed read face.
+    pub fn with_control(mut self, control: &'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)) -> Self {
+        self.control = Some(control);
+        self
+    }
+
     /// Accept connections on `listener` until it fails. Each accepted connection takes a
     /// slot before its thread starts and holds it until its answer is written, so at most
     /// the ceiling's count of connection threads run. One accepted with every slot held
@@ -343,49 +355,79 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
             }
             ("/auth/exchange", "POST") if self.exchange.is_some() => self.exchange.expect("checked above")(request),
             ("/auth/exchange", _) if self.exchange.is_some() => HttpResponse::message(405, "`/auth/exchange` answers POST").with("Allow", "POST"),
+            (WORKFLOWS_PATH, "GET") | (APPLY_PATH, "POST") if self.control.is_some() => self.control(request),
+            (WORKFLOWS_PATH, _) if self.control.is_some() => HttpResponse::message(405, "`/control/workflows` answers GET").with("Allow", "GET"),
+            (APPLY_PATH, _) if self.control.is_some() => HttpResponse::message(405, "`/control/apply` answers POST").with("Allow", "POST"),
             (other, _) => HttpResponse::message(404, format!("no route `{other}`; the protocol endpoint is `{MCP_PATH}`")),
         }
     }
 
     /// Admit the request's credential, then answer its one message.
     fn message(&self, request: &HttpRequest) -> HttpResponse {
+        let (authority, revocation) = match self.admit(request) {
+            Ok(a) => a,
+            Err(response) => return response,
+        };
+        let message: Value = match serde_json::from_slice(&request.body) {
+            Ok(m @ Value::Object(_)) => m,
+            Ok(_) => return HttpResponse::rpc_error(INVALID_REQUEST, "a request body holds one JSON-RPC message object"),
+            Err(e) => return HttpResponse::rpc_error(PARSE_ERROR, format!("the request body is not JSON: {e}")),
+        };
+        let admission = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience);
+        let boundary = |a: &AdmittedAuthority| effect_boundary(a, &admission);
+        match self.tools.handle(Caller { authority: &authority, boundary: &boundary }, &message) {
+            Some(answer) => HttpResponse::json(200, &answer),
+            None => HttpResponse::empty(202),
+        }
+    }
+
+    fn control(&self, request: &HttpRequest) -> HttpResponse {
+        let (authority, _) = match self.admit(request) {
+            Ok(a) => a,
+            Err(response) => return response,
+        };
+        if !authority.grants().iter().any(|grant| grant.actions.contains(&Action::Admin) && grant.tables.contains(&TablePattern::All)) {
+            return HttpResponse::json(403, &json!({ "error": { "identifier": "ControlAdminGrantMissing" } }));
+        }
+        let revocation = match (self.admitting.revocation)() {
+            Ok(r) => r,
+            Err(why) => return HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}")),
+        };
+        let boundary = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience);
+        if let Err(error) = effect_boundary(&authority, &boundary) {
+            return unadmitted(&error);
+        }
+        self.control.expect("the route is installed only with a control handler")(request)
+    }
+
+    fn admit(&self, request: &HttpRequest) -> Result<(AdmittedAuthority, RevocationState), HttpResponse> {
         let Some(credential) = request.credential() else {
             let missing = ReadError::HttpCredentialMissing(
                 "a request carries `Authorization: Bearer <credential>`, or `Authorization: DPoP <credential>` with a `DPoP` proof from the credential's holder key".into(),
             );
-            return HttpResponse::json(401, &payload(&Refusal::Read(missing))).with("WWW-Authenticate", CHALLENGE);
+            return Err(HttpResponse::json(401, &payload(&Refusal::Read(missing))).with("WWW-Authenticate", CHALLENGE));
         };
         let revocation = match (self.admitting.revocation)() {
             Ok(r) => r,
-            Err(why) => return HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}")),
+            Err(why) => return Err(HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}"))),
         };
         let admission = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience);
         let covered = ProofRequest { method: &request.method, target: &request.target, body: &request.body };
         let checkpoint = self.admitting.checkpoint;
         let keys = match checkpoint.keys() {
             Ok(k) => k,
-            Err(e) => return HttpResponse::unavailable(e.to_string()),
+            Err(e) => return Err(HttpResponse::unavailable(e.to_string())),
         };
         let authority = verify_network(credential, &keys, &admission, |jkt| match request.header("DPoP") {
             Some(proof) => checkpoint.verify_request(jkt, proof, &covered),
             None => Err(ProofRefusal::Refused(AuthorityError::PossessionProofInvalid(
                 "the credential binds a holder key, and the request carries no `DPoP` proof".into(),
             ))),
         });
-        let authority = match authority {
-            Ok(a) => a,
-            Err(ProofRefusal::NonceCacheFull) => return HttpResponse::unavailable(ProofRefusal::NonceCacheFull.to_string()),
-            Err(ProofRefusal::Refused(e)) => return unadmitted(&e),
-        };
-        let message: Value = match serde_json::from_slice(&request.body) {
-            Ok(m @ Value::Object(_)) => m,
-            Ok(_) => return HttpResponse::rpc_error(INVALID_REQUEST, "a request body holds one JSON-RPC message object"),
-            Err(e) => return HttpResponse::rpc_error(PARSE_ERROR, format!("the request body is not JSON: {e}")),
-        };
-        let boundary = |a: &AdmittedAuthority| effect_boundary(a, &admission);
-        match self.tools.handle(Caller { authority: &authority, boundary: &boundary }, &message) {
-            Some(answer) => HttpResponse::json(200, &answer),
-            None => HttpResponse::empty(202),
+        match authority {
+            Ok(a) => Ok((a, revocation)),
+            Err(ProofRefusal::NonceCacheFull) => Err(HttpResponse::unavailable(ProofRefusal::NonceCacheFull.to_string())),
+            Err(ProofRefusal::Refused(e)) => Err(unadmitted(&e)),
         }
     }
 }

crates/contextful-agent/tests/integration/http.rs

@@ -101,10 +101,14 @@ fn holder(n: u8) -> SigningKey {
 
 /// A credential reading `tables` for `ttl` seconds, bound to `holder` when one is given.
 fn credential(signer: &SeedSigner, tables: &str, ttl: u64, holder: Option<&SigningKey>) -> String {
+    credential_with_action(signer, Action::Read, tables, ttl, holder)
+}
+
+fn credential_with_action(signer: &SeedSigner, action: Action, tables: &str, ttl: u64, holder: Option<&SigningKey>) -> String {
     let policy = IssuancePolicy::parse(&format!("default_audience = \"{AUD}\"\nmax_lifetime_secs = 86400\n")).unwrap();
     let subject = Subject { on_behalf_of: Some("user://dana@acme.example".into()), zone: Some("on-prem:hq".into()), ..Subject::default() };
     let grant = Grant {
-        actions: vec![Action::Read],
+        actions: vec![action],
         tables: vec![TablePattern::parse(tables).unwrap()],
         tenant: None,
         aggregate: None,
@@ -121,6 +125,33 @@ fn credential(signer: &SeedSigner, tables: &str, ttl: u64, holder: Option<&Signi
     mint(&plan, &MintClaims { confirmation, ..MintClaims::default() }, signer).unwrap()
 }
 
+#[test]
+fn control_apply_rechecks_revocation_before_dispatch() {
+    let f = fixture();
+    let token = credential_with_action(&f.signer, Action::Admin, "*", 900, None);
+    let clock = FixedClock(at(NOW));
+    let checks = std::sync::atomic::AtomicUsize::new(0);
+    let revocation = || {
+        if checks.fetch_add(1, Ordering::SeqCst) == 0 { Ok(RevocationState::default()) }
+        else { Err("revocation source changed".into()) }
+    };
+    let admitting = Admitting { checkpoint: &f.checkpoint, audience: AUD, revocation: &revocation };
+    let dispatched = AtomicBool::new(false);
+    let handler = |_: &HttpRequest| {
+        dispatched.store(true, Ordering::SeqCst);
+        contextful_agent::http::HttpResponse::json(200, &json!({ "applied": 2 }))
+    };
+    let face = HttpFace::new(&f.face, &clock, &f.audit, admitting, Some(2)).unwrap().with_control(&handler);
+    let request = HttpRequest {
+        method: "POST".into(), target: "/control/apply".into(),
+        headers: vec![("Authorization".into(), format!("Bearer {token}"))], body: b"{}".to_vec(),
+    };
+    let answer = face.answer(&request);
+    assert_eq!(answer.status, 503);
+    assert!(!dispatched.load(Ordering::SeqCst));
+    assert_eq!(checks.load(Ordering::SeqCst), 2);
+}
+
 fn call(tool: &str, arguments: Value) -> Value {
     json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": arguments } })
 }

crates/contextful-cli/src/cadence.rs

@@ -17,6 +17,7 @@ use crate::run::{boot_id, wire_at, ProjectArgs};
 use anyhow::{bail, Context, Result};
 use contextful_context::project::Project;
 use contextful_core::pipeline::declare::{collect, dependent_runs, ManifestFile, PipelineSpec};
+use contextful_core::pipeline::transform::TransformOp;
 use contextful_core::grant::Action;
 use contextful_core::ports::Clock;
 use contextful_core::run::derive::task::Tasks;
@@ -36,7 +37,8 @@ use contextful_policy::verify::{effect_boundary, Admission, AdmittedAuthority};
 use contextful_outbound::egress::{system, Outbound, Transport};
 use serde::Serialize;
 use serde_json::json;
-use std::collections::BTreeMap;
+use serde_json::Value;
+use std::collections::{BTreeMap, BTreeSet};
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::io::Read;
@@ -238,6 +240,53 @@ fn applied(snaps: &Source) -> Result<(Option<u64>, BTreeMap<String, PipelineSpec
     Ok((Some(version), specs.into_iter().map(|d| (d.spec.id.clone(), d.spec)).collect()))
 }
 
+/// The applied snapshot and recorded run outcomes the store publishes for Admin.
+pub(crate) fn published(project: &Project, declaration: &Path) -> Result<Value> {
+    let text = std::fs::read_to_string(declaration)?;
+    let control = control_config(&text, project)?;
+    let (version, pipelines) = applied(&control.source)?;
+    let truncated = pipelines.len() > 1000;
+    let pipelines: Vec<PipelineSpec> = pipelines.into_values().take(1000).collect();
+    let selected: BTreeSet<&str> = pipelines.iter().map(|spec| spec.id.as_str()).collect();
+    let store = contextful_context::Store::open(&project.dir, &project.name)?;
+    let states = contextful_sync::run_state::run_states(&store)?;
+    let mut runs: BTreeMap<String, contextful_sync::RunMark> = BTreeMap::new();
+    for state in states.into_values() {
+        for (id, mark) in state.runs {
+            if selected.contains(id.as_str()) && runs.get(&id).is_none_or(|earlier| mark.started_at > earlier.started_at) {
+                runs.insert(id, mark);
+            }
+        }
+    }
+    let wired = wire_at(project, &None)?;
+    for row in wired.engine.catalog.runs(None)? {
+        if selected.contains(row.pipeline_id.as_str()) && runs.get(&row.pipeline_id).is_none_or(|earlier| row.started_at > earlier.started_at) {
+            runs.insert(row.pipeline_id.clone(), contextful_sync::RunMark {
+                run_id: row.run_id,
+                table: row.table,
+                status: row.status,
+                started_at: row.started_at,
+                ended_at: row.ended_at,
+            });
+        }
+    }
+    let pipelines: Vec<Value> = pipelines.into_iter().map(|spec| json!({
+        "id": spec.id,
+        "schedule": spec.schedule,
+        "tables": spec.tables.iter().map(|table| table.name()).collect::<Vec<_>>(),
+        "source": spec.source.name,
+        "after": spec.after,
+        "steps": spec.transforms.iter().map(|step| match step {
+            TransformOp::Select { .. } => "select",
+            TransformOp::Rename { .. } => "rename",
+            TransformOp::Cast { .. } => "cast",
+            TransformOp::Filter { .. } => "filter",
+            TransformOp::Extract { .. } => "extract",
+        }).collect::<Vec<_>>(),
+    })).collect();
+    Ok(json!({ "applied": version, "pipelines": pipelines, "runs": runs, "truncated": truncated, "declined": 0 }))
+}
+
 /// A locally applied descendant keeps precedence over an older bucket head after every
 /// receipt on the local path verifies under the same pinned issuer keys.
 fn local_descends(snaps: &SnapshotDir, current: u64, bucket: &ControlHead, project: &str, trusted: &[SignerKey]) -> Result<bool, SurfaceError> {

crates/contextful-cli/src/lib.rs

@@ -254,7 +254,7 @@ pub fn main_host(host: Host) {
         #[cfg(feature = "read-plane")]
         Cmd::Mcp(c) => mcp::run(c),
         #[cfg(feature = "read-plane")]
-        Cmd::Serve(c) => serve::run(c),
+        Cmd::Serve(c) => serve::run(c, &tasks),
         #[cfg(feature = "data-plane")]
         Cmd::Derive(c) => derive::run(c),
         #[cfg(feature = "read-plane")]

crates/contextful-cli/src/serve.rs

@@ -7,6 +7,8 @@
 //! network transport, which admits each one on its own credential. Every value is
 //! resolved before the listener binds, so a process that cannot serve binds nothing.
 
+#[cfg(feature = "data-plane")]
+use crate::admit::AdmitArgs;
 use crate::admit::{face, revocation_state, LedgerFile, LivePins, AUDIENCE_VAR, PUBKEY_VAR};
 use crate::project::locate;
 use crate::root::root as project_root;
@@ -18,12 +20,19 @@ use contextful_core::ports::{Clock, SigningPort};
 use contextful_policy::exchange::answer as exchange_answer;
 use contextful_policy::issue::{SeedSigner, DEFAULT_SEED_PATH};
 use contextful_policy::possession::ProofChecker;
+#[cfg(feature = "data-plane")]
+use contextful_agent::http::{HttpRequest, HttpResponse, APPLY_PATH, WORKFLOWS_PATH};
+#[cfg(feature = "data-plane")]
+use contextful_core::surface::SurfaceError;
+use contextful_core::run::derive::task::Tasks;
 use contextful_policy::audit::AuditLog;
 use contextful_policy::keyset::{KeyCheckpoint, StaticPins};
 use contextful_policy::revoke::RevocationState;
 use std::net::TcpListener;
 use std::path::PathBuf;
 use std::sync::Arc;
+#[cfg(feature = "data-plane")]
+use serde_json::{json, Value};
 
 /// The refusals of starting the network transport. `Display` begins with the identifier.
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
@@ -78,7 +87,9 @@ fn issuer_pins(flag: Option<&str>) -> Result<StaticPins, ServeError> {
     StaticPins::parse(text).map_err(|e| ServeError::IssuerKeyUnusable(format!("--public-key / {PUBKEY_VAR}: {e}")))
 }
 
-pub fn run(args: ServeArgs) -> Result<()> {
+pub fn run(args: ServeArgs, tasks: &Tasks) -> Result<()> {
+    #[cfg(not(feature = "data-plane"))]
+    let _ = tasks;
     let clock = SystemClock;
     // The declarations are checked before anything opens (`read.register.serve-declaration`).
     let audience = audience(args.audience.as_deref()).map_err(anyhow::Error::msg)?;
@@ -132,9 +143,43 @@ pub fn run(args: ServeArgs) -> Result<()> {
         let result = exchange_answer(exchange.as_ref(), &request.body, request.header("DPoP"), &proofs, issuance, &ctx);
         contextful_agent::http::HttpResponse::json(result.status, &result.body)
     };
+    #[cfg(feature = "data-plane")]
+    let control_admit = AdmitArgs { public_key: args.public_key.clone(), audience: args.audience.clone(), denylist: args.denylist.clone(), keyset: args.keyset.clone(), holder_key: None };
+    #[cfg(feature = "data-plane")]
+    let control = |request: &HttpRequest| -> HttpResponse {
+        let answer = match request.target.split('?').next().unwrap_or_default() {
+            WORKFLOWS_PATH => crate::cadence::published(&located.project, &located.declaration),
+            APPLY_PATH => {
+                let body: Value = match serde_json::from_slice(&request.body) {
+                    Ok(Value::Object(body)) => Value::Object(body),
+                    _ => return HttpResponse::json(400, &json!({ "error": { "identifier": "ControlRequestMalformed" } })),
+                };
+                let Some(fields) = body.as_object() else { unreachable!() };
+                if fields.len() > 1 || fields.keys().any(|field| field != "id") || fields.get("id").is_some_and(|id| !id.is_string()) {
+                    return HttpResponse::json(400, &json!({ "error": { "identifier": "ControlRequestMalformed" } }));
+                }
+                let project = crate::run::ProjectArgs { project: Some(located.project.name.clone()), now: None };
+                crate::cadence::apply(&project, Some(located.declaration.clone()), body["id"].as_str(), tasks, &control_admit, None)
+                    .and_then(|()| crate::cadence::published(&located.project, &located.declaration))
+            }
+            _ => unreachable!(),
+        };
+        match answer {
+            Ok(state) => HttpResponse::json(200, &state),
+            Err(error) => {
+                let surface = error.chain().find_map(|part| part.downcast_ref::<SurfaceError>());
+                let status = surface.map_or(503, SurfaceError::status);
+                let identifier = surface.map(|refusal| refusal.to_string().split(':').next().unwrap_or("ControlUnavailable").to_string())
+                    .unwrap_or_else(|| "ControlUnavailable".into());
+                HttpResponse::json(status, &json!({ "error": { "identifier": identifier } }))
+            }
+        }
+    };
     let http = HttpFace::new(&face, &clock, &audit, admitting, Some(ceiling))
         .map_err(anyhow::Error::msg)?
         .with_exchange(&exchange_route, exchange.is_none());
+    #[cfg(feature = "data-plane")]
+    let http = http.with_control(&control);
     let listener = TcpListener::bind(&args.http)?;
     eprintln!("listening on http://{}/mcp", listener.local_addr()?);
     http.serve(listener)?;