Serve governed durable claim writes for console learning

crates/contextful-agent/src/http.rs

@@ -203,6 +203,7 @@ pub struct HttpFace<'a, C> {
     in_flight: AtomicUsize,
     exchange: Option<&'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)>,
     exchange_unconfigured: bool,
+    claim_write: Option<&'a (dyn Fn(&HttpRequest, &AdmittedAuthority, &dyn Fn() -> Result<(), AuthorityError>) -> HttpResponse + Sync)>,
 }
 
 /// A request slot held while one request is in flight.
@@ -249,7 +250,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, control: None, 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, claim_write: None })
     }
 
     /// The binary's exchange route mints the reader credential without putting issuer
@@ -266,6 +267,12 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
         self
     }
 
+    /// The binary owns the claim writer; this route never joins the read MCP tool set.
+    pub fn with_claim_write(mut self, write: &'a (dyn Fn(&HttpRequest, &AdmittedAuthority, &dyn Fn() -> Result<(), AuthorityError>) -> HttpResponse + Sync)) -> Self {
+        self.claim_write = Some(write);
+        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
@@ -353,6 +360,8 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
             (MCP_PATH, _) => {
                 HttpResponse::message(405, "`/mcp` answers POST; the face holds no session and opens no server stream").with("Allow", "POST")
             }
+            ("/memory/claims", "POST") if self.claim_write.is_some() => self.claim_write_request(request),
+            ("/memory/claims", _) if self.claim_write.is_some() => HttpResponse::message(405, "`/memory/claims` answers POST").with("Allow", "POST"),
             ("/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),
@@ -400,6 +409,22 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> {
         self.control.expect("the route is installed only with a control handler")(request)
     }
 
+    fn claim_write_request(&self, request: &HttpRequest) -> HttpResponse {
+        if request.header("Origin").is_some() {
+            return HttpResponse::json(403, &json!({ "error": { "http": 403, "identifier": "MemoryClaimBrowserRefused" } }));
+        }
+        let (authority, _) = match self.admit(request) {
+            Ok(a) => a,
+            Err(response) => return response,
+        };
+        let boundary = || {
+            let revocation = (self.admitting.revocation)().map_err(AuthorityError::AuthorityRevoked)?;
+            let admission = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience);
+            effect_boundary(&authority, &admission)
+        };
+        self.claim_write.expect("route exists")(request, &authority, &boundary)
+    }
+
     fn admit(&self, request: &HttpRequest) -> Result<(AdmittedAuthority, RevocationState), HttpResponse> {
         let Some(credential) = request.credential() else {
             let missing = ReadError::HttpCredentialMissing(

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

@@ -253,6 +253,37 @@ fn stdio_answer(f: &Fixture, token: &str, message: &Value) -> Value {
     server.handle(&message.to_string()).unwrap()
 }
 
+#[test]
+fn server_held_claim_route_admits_each_request_and_keeps_browser_and_mcp_writes_out() {
+    let f: &'static Fixture = Box::leak(Box::new(fixture()));
+    let clock: &'static FixedClock = Box::leak(Box::new(FixedClock(at(NOW))));
+    let admitting = Admitting { checkpoint: &f.checkpoint, audience: AUD, revocation: &no_revocation };
+    let write = |_: &HttpRequest, authority: &AdmittedAuthority, boundary: &dyn Fn() -> Result<(), AuthorityError>| {
+        boundary().unwrap();
+        HttpResponse::json(200, &json!({ "actor": authority.subject().on_behalf_of() }))
+    };
+    let write: &'static _ = Box::leak(Box::new(write));
+    let face = Box::leak(Box::new(HttpFace::new(&f.face, clock, &f.audit, admitting, Some(4)).unwrap().with_claim_write(write)));
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap();
+    std::thread::spawn(move || face.serve(listener));
+    let token = credential(&f.signer, "research/*", 900, None);
+    let mut request = HttpRequest { method: "POST".into(), target: "/memory/claims".into(),
+        headers: vec![("Authorization".into(), format!("Bearer {token}"))], body: b"{}".to_vec() };
+    let (status, _, response) = send(addr, &request);
+    assert_eq!(status, 200);
+    assert_eq!(body(&response)["actor"], "user://dana@acme.example");
+    request.headers.push(("Origin".into(), "https://console.example".into()));
+    let (status, _, _) = send(addr, &request);
+    assert_eq!(status, 403);
+    request.headers.clear();
+    let (status, _, _) = send(addr, &request);
+    assert_eq!(status, 401);
+    let (status, _, response) = send(addr, &unproven(&call("memory.write", json!({})), "Bearer", &token));
+    assert_eq!(status, 200);
+    assert!(body(&response).to_string().contains("no tool"));
+}
+
 /// `contextful serve --http <addr> --audience <aud> --max-in-flight <n>` answers MCP Streamable HTTP at `POST /mcp`: one JSON-RPC message per request, answered as `application/json` by the tool server the stdio transport runs.
 // spec: read.register.network-transport@38ec986b
 #[test]

crates/contextful-cli/src/serve.rs

@@ -17,6 +17,12 @@ use anyhow::Result;
 use contextful_agent::http::{audience, ceiling, Admitting, HttpFace};
 use contextful_core::issue::{IssuancePolicy, MintContext, NodeRole};
 use contextful_core::ports::{Clock, SigningPort};
+#[cfg(feature = "data-plane")]
+use contextful_core::memory::synthesize::CandidateClaim;
+#[cfg(feature = "data-plane")]
+use contextful_memory::write::{write_observed, Observation};
+#[cfg(feature = "data-plane")]
+use contextful_memory::MemoryFault;
 use contextful_policy::exchange::answer as exchange_answer;
 use contextful_policy::issue::{SeedSigner, DEFAULT_SEED_PATH};
 use contextful_policy::possession::ProofChecker;
@@ -34,6 +40,31 @@ use std::sync::Arc;
 #[cfg(feature = "data-plane")]
 use serde_json::{json, Value};
 
+#[cfg(feature = "data-plane")]
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ClaimWriteRequest {
+    into: String,
+    actor: String,
+    session: String,
+    dedup_key: String,
+    claim: CandidateClaim,
+}
+
+#[cfg(feature = "data-plane")]
+fn claim_write_error(error: MemoryFault) -> contextful_agent::http::HttpResponse {
+    let (status, identifier): (u16, String) = match &error {
+        MemoryFault::Denied(_) => (403, "GrantWriteNotCovered".into()),
+        MemoryFault::Authority(why) => (401, why.to_string().split(':').next().unwrap_or("AuthorityRevoked").to_owned()),
+        MemoryFault::Invalid(_) => (400, "MemoryClaimMalformed".into()),
+        MemoryFault::Memory(why) => (400, why.identifier().into()),
+        _ => (503, "MemoryClaimWriteRefused".into()),
+    };
+    contextful_agent::http::HttpResponse::json(status, &serde_json::json!({
+        "error": { "http": status, "identifier": identifier, "message": error.to_string() }
+    }))
+}
+
 /// The refusals of starting the network transport. `Display` begins with the identifier.
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
 pub enum ServeError {
@@ -134,6 +165,42 @@ pub fn run(args: ServeArgs, tasks: &Tasks) -> Result<()> {
         None
     };
     let proofs = ProofChecker::new(SystemClock);
+    #[cfg(feature = "data-plane")]
+    let (node, _) = contextful_context::node::resolve(face.store(), |k| std::env::var(k).ok())?;
+    #[cfg(feature = "data-plane")]
+    let claim_write = |request: &contextful_agent::http::HttpRequest,
+                       authority: &contextful_policy::verify::AdmittedAuthority,
+                       boundary: &dyn Fn() -> Result<(), contextful_core::AuthorityError>| {
+        let body: ClaimWriteRequest = match serde_json::from_slice(&request.body) {
+            Ok(body) => body,
+            Err(_) => return contextful_agent::http::HttpResponse::json(400, &serde_json::json!({
+                "error": { "http": 400, "identifier": "MemoryClaimMalformed" }
+            })),
+        };
+        if body.actor.is_empty() || body.session.is_empty() || body.session.contains(':') ||
+            authority.subject().on_behalf_of() != Some(body.actor.as_str()) ||
+            authority.subject().task() != Some(body.session.as_str()) {
+            return contextful_agent::http::HttpResponse::json(403, &serde_json::json!({
+                "error": { "http": 403, "identifier": "MemoryClaimScopeRefused" }
+            }));
+        }
+        let mut claim = body.claim;
+        if claim.scope.is_some() || body.dedup_key.trim().is_empty() {
+            return contextful_agent::http::HttpResponse::json(400, &serde_json::json!({
+                "error": { "http": 400, "identifier": "MemoryClaimMalformed" }
+            }));
+        }
+        claim.scope = Some(format!("console:{}:{}", body.actor, body.session));
+        let scope = claim.scope.clone().expect("assigned");
+        match write_observed(&face, authority, &body.into, claim,
+            &Observation { observed_at: None, dedup_key: Some(body.dedup_key) }, &node, clock.now(), boundary) {
+            Ok(written) => contextful_agent::http::HttpResponse::json(200, &serde_json::json!({
+                "landed": written.claim.is_some(), "scope": scope,
+                "claim_id": written.claim.map(|claim| claim.claim_id)
+            })),
+            Err(error) => claim_write_error(error),
+        }
+    };
     let exchange_route = |request: &contextful_agent::http::HttpRequest| {
         let Some((issuance, signer)) = mint_material.as_ref() else {
             return contextful_agent::http::HttpResponse::json(404,
@@ -180,6 +247,8 @@ pub fn run(args: ServeArgs, tasks: &Tasks) -> Result<()> {
         .with_exchange(&exchange_route, exchange.is_none());
     #[cfg(feature = "data-plane")]
     let http = http.with_control(&control);
+    #[cfg(feature = "data-plane")]
+    let http = http.with_claim_write(&claim_write);
     let listener = TcpListener::bind(&args.http)?;
     eprintln!("listening on http://{}/mcp", listener.local_addr()?);
     http.serve(listener)?;

crates/contextful-cli/tests/integration/serve.rs

@@ -205,6 +205,84 @@ fn query() -> Value {
     json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "context.query", "arguments": { "sql": "SELECT note_id FROM \"research/notes\"" } } })
 }
 
+fn post_claim(addr: &str, token: &str, body: &Value, browser_origin: bool) -> (u16, Value) {
+    let body = body.to_string();
+    let origin = if browser_origin { "Origin: https://console.example\r\n" } else { "" };
+    let mut stream = TcpStream::connect(addr).unwrap();
+    write!(stream, "POST /memory/claims HTTP/1.1\r\nAuthorization: Bearer {token}\r\n{origin}Content-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len()).unwrap();
+    let mut raw = String::new();
+    stream.read_to_string(&mut raw).unwrap();
+    let (head, data) = raw.split_once("\r\n\r\n").unwrap();
+    (head.split(' ').nth(1).unwrap().parse().unwrap(), serde_json::from_str(data).unwrap_or(Value::Null))
+}
+
+#[test]
+fn served_claim_write_is_durable_actor_bound_and_outside_read_mcp() {
+    let memory = "\n[[table]]\nname = \"memory/facts\"\nshape = \"memory_facts\"\ncolumns = [\"claim_id\", \"subject\", \"predicate\", \"object\", \"scope\", \"tier\", \"confidence\", \"valid_from\", \"valid_to\", \"evidence\", \"superseded_by\", \"grant_id\", \"agent\"]\n";
+    let (dir, public) = project_declaring(memory);
+    let p = dir.path();
+    let mint = |who: &str, task: &str, actions: &[&str]| {
+        let mut args = vec!["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", who,
+            "--task", task, "--zone", "on-prem:hq", "--table", "memory/facts", "--table", "research/notes", "--ttl", "900"];
+        for action in actions { args.extend(["--action", action]); }
+        stdout(&run(p, &args))
+    };
+    let alice = mint("user://alice", "session-1", &["read", "write"]);
+    let bob = mint("user://bob", "session-2", &["read", "write"]);
+    let reader = mint("user://alice", "session-1", &["read"]);
+    let wrong_table = stdout(&run(p, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://alice",
+        "--task", "session-1", "--zone", "on-prem:hq", "--action", "write", "--table", "research/notes", "--ttl", "900"]));
+    let expired = stdout(&run(p, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://alice",
+        "--task", "session-1", "--zone", "on-prem:hq", "--action", "read", "--action", "write",
+        "--table", "memory/facts", "--table", "research/notes", "--ttl", "900", "--now", "2020-01-01T00:00:00Z"]));
+    let claim = json!({ "into": "memory/facts", "actor": "user://alice", "session": "session-1", "dedup_key": "turn-1",
+        "claim": { "subject": "Northwind", "predicate": "filings", "object": "arrived", "confidence": 0.9,
+            "evidence": [{ "table": "research/notes", "run": "run-0001", "seq": 0 }] } });
+    let (listener, addr) = serve(p, &["serve", "--http", "127.0.0.1:0", "--audience", AUD,
+        "--max-in-flight", "2", "--project", "research", "--public-key", &public]);
+    assert_eq!(post_claim(&addr, &reader, &claim, false).0, 403, "read grant cannot write");
+    assert_eq!(post_claim(&addr, &wrong_table, &claim, false).0, 403, "a write grant on another table cannot land memory");
+    assert_eq!(post_claim(&addr, &expired, &claim, false).0, 401, "expired authority cannot write");
+    let mut client_scope = claim.clone();
+    client_scope["claim"]["scope"] = json!("shared");
+    assert_eq!(post_claim(&addr, &alice, &client_scope, false).0, 400, "the caller cannot set claim scope");
+    assert_eq!(post_claim(&addr, &alice, &claim, true).0, 403, "a browser origin cannot reach the write");
+    let (status, first) = post_claim(&addr, &alice, &claim, false);
+    assert_eq!(status, 200, "{first}");
+    assert_eq!(first["landed"], true);
+    assert!(first["scope"].as_str().is_some_and(|scope| scope.contains("alice") && scope.contains("session-1")), "{first}");
+    let (status, duplicate) = post_claim(&addr, &alice, &claim, false);
+    assert_eq!(status, 200, "{duplicate}");
+    assert_eq!(duplicate["landed"], false);
+    let mut borrowed = claim.clone();
+    borrowed["actor"] = json!("user://bob");
+    assert_eq!(post_claim(&addr, &alice, &borrowed, false).0, 403, "a writer cannot author for another operator");
+    borrowed["actor"] = json!("user://alice");
+    borrowed["session"] = json!("session-2");
+    assert_eq!(post_claim(&addr, &alice, &borrowed, false).0, 403, "a writer cannot pick another session");
+    assert_eq!(post_claim(&addr, &bob, &claim, false).0, 403, "the second operator cannot author Alice's claim");
+    let colon = mint("user://alice", "session:1", &["read", "write"]);
+    let mut colon_claim = claim.clone();
+    colon_claim["session"] = json!("session:1");
+    let (status, refused) = post_claim(&addr, &colon, &colon_claim, false);
+    assert_eq!((status, refused["error"]["identifier"].clone()), (403, json!("MemoryClaimScopeRefused")),
+        "a session holding the scope separator cannot alias another actor's scope");
+    let write_tool = json!({ "jsonrpc": "2.0", "id": 7, "method": "tools/call",
+        "params": { "name": "memory.write", "arguments": claim } });
+    let (_, mcp) = post(&addr, &write_tool, &alice, None);
+    assert!(mcp.to_string().contains("no tool"), "{mcp}");
+    drop(listener);
+    let (_restarted, addr) = serve(p, &["serve", "--http", "127.0.0.1:0", "--audience", AUD,
+        "--max-in-flight", "2", "--project", "research", "--public-key", &public]);
+    let recall = json!({ "jsonrpc": "2.0", "id": 8, "method": "tools/call",
+        "params": { "name": "memory.recall", "arguments": { "table": "memory/facts", "subject": "Northwind" } } });
+    let (status, memory) = post(&addr, &recall, &reader, None);
+    assert_eq!(status, 200, "{memory}");
+    assert_eq!(memory["result"]["structuredContent"]["rows"].as_array().map(Vec::len), Some(1), "{memory}");
+    stdout(&run(p, &["token", "revoke", "--principal-class", "delegated"]));
+    assert_eq!(post_claim(&addr, &alice, &claim, false).0, 401, "revoked authority commits nothing");
+}
+
 #[test]
 fn served_exchange_without_policy_returns_the_unconfigured_refusal() {
     let (dir, public) = project();

spec/21-memory.md

@@ -119,7 +119,18 @@ Supersession within one validity line, confidence decay, the direct write and it
   *A-read*
 - `retention-default` — Claims retain their recorded validity until explicit expiry or erasure; ranking decay is opt-in per shape with a declared half-life, and never deletes a claim or changes its validity interval.
   *A-read*
-
+- `served-write` — A writable served face answers `POST /memory/claims` under a per-request network credential, lands a declared claim through the direct write, and keeps the read MCP tool set closed.
+  *A-read*
+- `served-scope` — A claim write whose `actor` or `session` differs from the admitted credential's `on_behalf_of` or `task`, or whose `session` holds a colon, raises `MemoryClaimScopeRefused`; accepted claims take scope from those members, never from the payload.
+  *A-read*
+- `browser-request` — An `Origin` header on the served claim-write route raises `MemoryClaimBrowserRefused` before credential admission and commits nothing.
+  *A-read*
+- `malformed-request` — Malformed JSON, unknown fields, a client-selected claim scope, or an empty dedup key on the served claim-write route raises `MemoryClaimMalformed` and commits nothing.
+  *A-read*
+- `served-dedup` — A served claim write passes its `dedup_key` to {{read.revise.dedup-key}}, returns `landed: false` on restatement, and returns the bound scope on both outcomes.
+  *A-read*
+- `served-fault` — A served claim write whose store effect refuses raises `MemoryClaimWriteRefused` and lands nothing.
+  *A-read*
 
 ## recall