From 506b7799ac4e3872076c9f72863f0ddf8739d99d Mon Sep 17 00:00:00 2001 From: Larry Klosowski Date: Mon, 19 Oct 2026 09:16:31 -0700 Subject: [PATCH 1/3] feat(sidecar,core): approve carries the payment outcome; capsules see "submitted, outcome unknown" (C1-V3-R) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit citrate-core (C1-V3) releases an approved payment whose broadcast had no known outcome through POST /approvals/approve with {id, outcome: "submitted_outcome_unknown", txHash, message}. The sidecar read only `id`, so the decision record said Completed and the capsule was not told what happened. - hitl: UnknownOutcome (validated: txHash 0x + 64 hex, message non-empty, <= 280 bytes, no control characters, names the hash when there is one). ApprovalQueue::approve_by_id_outcome_unknown resolves the head with it; submit_with_resolution returns Resolution::SubmittedOutcomeUnknown, and submit()/submit_with_outcome never read it as approved. - No unattended re-send: the queue remembers the last 256 unknown-outcome call ids. An identical re-proposal skips the auto-grant fast path and its PendingView names the earlier tx. - ApprovalGate::decide (default mirrors request) returns GateDecision; the eth-send host fn reports "ChainSendOutcomeUnknown: " and never dispatches, so nothing is sent twice. - sidecar: the approve route parses the optional fields (absent = today's behaviour; malformed, orphaned, or on reject = 400, before anything is recorded or resolved). The decision record closes as outcome_unknown with "Submitted · outcome unknown (tx …)". /approvals carries priorOutcomeUnknown and a summary note for a re-proposal. Tests: citrate-agent-core + agent-sidecar 879 -> 898 passed, 0 failed. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_0148EsKJD57GeaM8JwAraU3N --- agent-sidecar/src/lib.rs | 128 ++++++- agent-sidecar/src/outcome_unknown_tests.rs | 387 +++++++++++++++++++++ agent/core/src/capsule/dispatcher.rs | 27 ++ agent/core/src/capsule/linker.rs | 15 +- agent/core/src/capsule/mod.rs | 80 +++++ agent/core/src/capsule/prod_impls.rs | 91 ++++- agent/core/src/hitl/mod.rs | 242 ++++++++++++- agent/core/src/hitl/outcome.rs | 175 ++++++++++ 8 files changed, 1096 insertions(+), 49 deletions(-) create mode 100644 agent-sidecar/src/outcome_unknown_tests.rs create mode 100644 agent/core/src/hitl/outcome.rs diff --git a/agent-sidecar/src/lib.rs b/agent-sidecar/src/lib.rs index 9884ea3..6adc98a 100644 --- a/agent-sidecar/src/lib.rs +++ b/agent-sidecar/src/lib.rs @@ -239,6 +239,21 @@ struct ApprovalBody { to: Option, #[serde(skip_serializing_if = "Option::is_none")] data: Option, + /// C1-V3-R: an identical effect was approved earlier and submitted with an unknown outcome. + /// Absent otherwise. + #[serde( + rename = "priorOutcomeUnknown", + skip_serializing_if = "Option::is_none" + )] + prior_outcome_unknown: Option, +} + +/// C1-V3-R: the earlier submission an identical re-proposal must be checked against. +#[derive(Serialize)] +struct PriorOutcomeBody { + #[serde(rename = "txHash")] + tx_hash: Option, + message: String, } /// AR-B-024 — refuse a non-loopback control bind unless explicitly opted in. @@ -570,15 +585,25 @@ async fn approvals( .get("data_hex") .and_then(|v| v.as_str()) .map(str::to_string); + // C1-V3-R: a re-proposal of an effect whose earlier submission has an unknown outcome + // names that transaction, in the summary the member reads and as a field. + let summary = match &p.prior_unknown { + Some(u) => format!("{} · {}", p.description, u.reproposal_note()), + None => p.description, + }; out.push(ApprovalBody { // AR-B-023: expose the stable call-id (not the tool name) so the // ceremony can assert "I am approving *this* call" when it posts // back to /approvals/approve. id: p.id.clone(), kind: p.risk_level.clone(), - summary: p.description, + summary, to, data, + prior_outcome_unknown: p.prior_unknown.map(|u| PriorOutcomeBody { + tx_hash: u.tx_hash().map(str::to_string), + message: u.message().to_string(), + }), }); } Ok(Json(out)) @@ -620,12 +645,50 @@ async fn reject_head( /// Extract the call-id the human reviewed from an approve/reject body. `None` for anything that is /// not a JSON object carrying a non-empty string `id` (PBA-L6b-009). -fn requested_call_id(body: &[u8]) -> Option { - let v: serde_json::Value = serde_json::from_slice(body).ok()?; +fn requested_call_id(v: &serde_json::Value) -> Option { let id = v.get("id")?.as_str()?; (!id.is_empty()).then(|| id.to_string()) } +/// C1-V3-R: the optional payment outcome on an approve body (data source: citrate-core's +/// signature ceremony, `hermes_resolve_outcome_unknown`). `Ok(None)` when the body carries no +/// outcome, which is today's behaviour. `Err` (400) for a malformed outcome, a `txHash` or +/// `message` without an outcome, or any outcome on a reject: an unknown outcome is never a +/// rejection, because the member approved it and the transaction went out. +fn requested_outcome( + v: &serde_json::Value, + approve: bool, +) -> Result, StatusCode> { + let present = |k: &str| v.get(k).is_some_and(|x| !x.is_null()); + if !present("outcome") { + return if present("txHash") || present("message") { + Err(StatusCode::BAD_REQUEST) + } else { + Ok(None) + }; + } + if !approve { + return Err(StatusCode::BAD_REQUEST); + } + if v.get("outcome").and_then(|o| o.as_str()) + != Some(citrate_agent_core::hitl::OUTCOME_SUBMITTED_UNKNOWN) + { + return Err(StatusCode::BAD_REQUEST); + } + let optional_str = |k: &str| -> Result, StatusCode> { + match v.get(k) { + None | Some(serde_json::Value::Null) => Ok(None), + Some(serde_json::Value::String(s)) => Ok(Some(s.as_str())), + Some(_) => Err(StatusCode::BAD_REQUEST), + } + }; + let tx_hash = optional_str("txHash")?; + let message = optional_str("message")?; + citrate_agent_core::hitl::UnknownOutcome::new(tx_hash, message) + .map(Some) + .map_err(|_| StatusCode::BAD_REQUEST) +} + /// Shared approve/reject resolution, always bound to a call-id (AR-B-023 / PBA-L6b-009). fn resolve_body( st: &Arc, @@ -636,7 +699,10 @@ fn resolve_body( if st.estop.is_stopped() { return Err(StatusCode::SERVICE_UNAVAILABLE); } - let id = requested_call_id(body).ok_or(StatusCode::BAD_REQUEST)?; + let v: serde_json::Value = serde_json::from_slice(body).map_err(|_| StatusCode::BAD_REQUEST)?; + let id = requested_call_id(&v).ok_or(StatusCode::BAD_REQUEST)?; + // C1-V3-R: validated before anything is recorded or resolved. + let unknown = requested_outcome(&v, approve)?; // HUP-S2.6: the member's decision on the reviewed head is recorded (write-ahead for an // approval; fail closed when the record cannot be written). A resolve that does not match // the head records nothing. @@ -648,39 +714,63 @@ fn resolve_body( ), _ => None, }; - let res = if approve { - st.queue.approve_by_id(&id) - } else { - st.queue.reject_by_id(&id) + let res = match (&unknown, approve) { + (Some(u), _) => st.queue.approve_by_id_outcome_unknown(&id, u.clone()), + (None, true) => st.queue.approve_by_id(&id), + (None, false) => st.queue.reject_by_id(&id), }; match res { Ok(()) => { if let (Some(log), Some(p)) = (&log, &head) { - match pending_record { - Some(seq) => hic_records::record_outcome( + match (pending_record, &unknown) { + // C1-V3-R: submitted, outcome unknown. Never "Completed". + (Some(seq), Some(u)) => hic_records::record_outcome( + log, + seq, + citrate_agent_records::Outcome::OutcomeUnknown, + &u.record_detail(), + ), + (Some(seq), None) => hic_records::record_outcome( log, seq, citrate_agent_records::Outcome::Completed, "released to the capsule; a chain effect was signed and broadcast in core's ceremony first", ), - None => { + (None, _) => { if let Err(e) = record_ceremony_decision(log, p, false) { eprintln!("citrate-agent-sidecar: {e}"); } } } } - Ok(Json(serde_json::json!({ "ok": true, "resolved": true }))) + let mut out = serde_json::json!({ "ok": true, "resolved": true }); + if unknown.is_some() { + out["outcome"] = + serde_json::json!(citrate_agent_core::hitl::OUTCOME_SUBMITTED_UNKNOWN); + } + Ok(Json(out)) } // The head is not the call the operator reviewed, or its submitter is gone — refuse. Err(_) => { if let (Some(log), Some(seq)) = (&log, pending_record) { - hic_records::record_outcome( - log, - seq, - citrate_agent_records::Outcome::Failed, - "the call was no longer waiting, so nothing was released", - ); + match &unknown { + // The transaction went out from core even though the capsule was not told. + Some(u) => hic_records::record_outcome( + log, + seq, + citrate_agent_records::Outcome::OutcomeUnknown, + &format!( + "{}; the call was no longer waiting, so the capsule was not told", + u.record_detail() + ), + ), + None => hic_records::record_outcome( + log, + seq, + citrate_agent_records::Outcome::Failed, + "the call was no longer waiting, so nothing was released", + ), + } } Err(StatusCode::CONFLICT) } @@ -2388,6 +2478,8 @@ mod mcp_session_tests; #[cfg(test)] mod metering_session_tests; #[cfg(test)] +mod outcome_unknown_tests; +#[cfg(test)] mod retrieval_http_tests; #[cfg(test)] mod session_reattach_tests; diff --git a/agent-sidecar/src/outcome_unknown_tests.rs b/agent-sidecar/src/outcome_unknown_tests.rs new file mode 100644 index 0000000..f5344ec --- /dev/null +++ b/agent-sidecar/src/outcome_unknown_tests.rs @@ -0,0 +1,387 @@ +//! C1-V3-R — the ceremony bridge's approve route carries the payment outcome. +//! +//! citrate-core (C1-V3, `hermes_resolve_outcome_unknown`) releases a payment the member approved +//! whose broadcast had no known outcome through `POST /approvals/approve` with +//! `{id, outcome: "submitted_outcome_unknown", txHash, message}`. The capsule must be told exactly +//! that message (not success), the decision record must say "Submitted · outcome unknown (tx …)", +//! and an identical re-proposal must reference the earlier transaction and never auto-approve. +//! A body with only `{id}` keeps today's behaviour. + +use super::*; +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use citrate_agent_core::hitl::{ApprovalOutcomePublic, ToolCall}; +use citrate_agent_loop::{AssistantTurn, CompletionRequest, LlmClient, LlmError}; +use citrate_agent_records::{DecisionLog, Entry, LogConfig, Outcome}; +use std::time::Duration; +use tower::ServiceExt; + +const BEARER: &str = "test-bearer-token-outcome-0001"; +const HASH: &str = "0x8f3c2a1b9d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8"; + +struct Quiet; +impl LlmClient for Quiet { + fn complete(&self, _req: &CompletionRequest) -> Result { + Ok(AssistantTurn::text("(done)")) + } +} + +struct Fx { + root: tempfile::TempDir, + st: Arc, +} + +impl Fx { + fn new() -> Self { + let root = tempfile::tempdir().unwrap(); + let (log, _) = + DecisionLog::open(&root.path().join("records"), LogConfig::default()).unwrap(); + let log = Arc::new(log); + let client: Arc = Arc::new(Quiet); + let mgr = sessions::SessionManager::new( + Arc::new(move |_ep: &sessions::LlmEndpoint| client.clone()), + Duration::from_secs(5), + ) + .with_records(log.clone()); + let st = Arc::new(AppState { + estop: EmergencyStop::new(), + queue: Arc::new(ApprovalQueue::new()), + skills: vec![], + dispatch: None, + bearer: BEARER.to_string(), + run_slots: Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_SKILLS)), + sessions: Arc::new(mgr), + }); + Fx { root, st } + } + + fn records(&self) -> Vec { + let dir = self.root.path().join("records"); + let mut r = citrate_agent_records::read::page(&dir, None, 10_000).unwrap(); + r.sort_by_key(|x| x.record.seq); + r + } + + /// The (outcome, detail) closing each ceremony decision, oldest first. + fn ceremony_outcomes(&self) -> Vec<(Decision, Option<(Outcome, String)>)> { + let recs = self.records(); + recs.iter() + .filter_map(|r| match &r.record.entry { + Entry::Decision(d) if d.kind == "ceremony.capsule_effect" => { + let seq = r.record.seq; + let out = recs.iter().find_map(|o| match &o.record.entry { + Entry::Outcome(o) if o.decision_seq == seq => { + Some((o.outcome, o.detail.clone())) + } + _ => None, + }); + Some((d.decision, out)) + } + _ => None, + }) + .collect() + } + + async fn post(&self, path: &str, body: serde_json::Value) -> (StatusCode, serde_json::Value) { + let r = app(self.st.clone()) + .oneshot( + Request::builder() + .method("POST") + .uri(path) + .header("content-type", "application/json") + .header("authorization", format!("Bearer {BEARER}")) + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let s = r.status(); + let bytes = axum::body::to_bytes(r.into_body(), 1 << 20).await.unwrap(); + ( + s, + serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null), + ) + } + + async fn get(&self, path: &str) -> serde_json::Value { + let r = app(self.st.clone()) + .oneshot( + Request::builder() + .method("GET") + .uri(path) + .header("authorization", format!("Bearer {BEARER}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(r.into_body(), 1 << 20).await.unwrap(); + serde_json::from_slice(&bytes).unwrap() + } + + async fn wait_depth(&self, want: usize) { + for _ in 0..400 { + if self.st.queue.depth() == want { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + panic!("queue never reached depth {want}"); + } +} + +use citrate_agent_records::Decision; + +fn payment(id: &str) -> ToolCall { + ToolCall { + call_id: id.to_string(), + name: "pay::eth-send".to_string(), + args: serde_json::json!({ + "to": "0x4a86659BDab24dc444C72fbbaD4cd83491820E40", + "data_hex": "0xa9059cbb", + "data_len": 4 + }), + } +} + +fn core_message(hash: &str) -> String { + format!("submitted, outcome unknown, tx {hash}, check before retrying") +} + +fn unknown_body(id: &str) -> serde_json::Value { + serde_json::json!({ + "id": id, + "outcome": "submitted_outcome_unknown", + "txHash": HASH, + "message": core_message(HASH), + }) +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unknown_outcome_reaches_the_capsule_as_that_message_not_success() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_resolution(payment("p1")).await }); + fx.wait_depth(1).await; + + let (s, v) = fx.post("/approvals/approve", unknown_body("p1")).await; + assert_eq!(s, StatusCode::OK, "{v}"); + assert_eq!(v["resolved"], true); + assert_eq!(v["outcome"], "submitted_outcome_unknown"); + + match waiter.await.unwrap() { + citrate_agent_core::hitl::Resolution::SubmittedOutcomeUnknown(u) => { + assert_eq!(u.message(), core_message(HASH), "exactly core's words"); + assert!(u.message().contains(HASH), "the message names the tx"); + assert_eq!(u.tx_hash(), Some(HASH)); + } + other => panic!("an unknown outcome is not a plain decision: {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_plain_submit_never_reads_an_unknown_outcome_as_approved() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_outcome(payment("p2")).await }); + fx.wait_depth(1).await; + let (s, _) = fx.post("/approvals/approve", unknown_body("p2")).await; + assert_eq!(s, StatusCode::OK); + let out = waiter.await.unwrap(); + assert_eq!(out, ApprovalOutcomePublic::SubmittedOutcomeUnknown); + assert_ne!(out, ApprovalOutcomePublic::Approved); +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_decision_record_says_submitted_outcome_unknown_not_completed() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_outcome(payment("p3")).await }); + fx.wait_depth(1).await; + let (s, _) = fx.post("/approvals/approve", unknown_body("p3")).await; + assert_eq!(s, StatusCode::OK); + let _ = waiter.await; + + let rows = fx.ceremony_outcomes(); + assert_eq!(rows.len(), 1, "{rows:?}"); + assert_eq!(rows[0].0, Decision::Approved, "the member approved it"); + let (outcome, detail) = rows[0].1.clone().expect("the decision is closed"); + assert_eq!(outcome, Outcome::OutcomeUnknown, "never Completed"); + assert_eq!(detail, format!("Submitted · outcome unknown (tx {HASH})")); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unknown_outcome_without_a_hash_says_so() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_resolution(payment("p4")).await }); + fx.wait_depth(1).await; + // What core sends when the ceremony had no hash: txHash null, its own words. + let msg = "submitted, outcome unknown, check recent transactions before retrying"; + let (s, v) = fx + .post( + "/approvals/approve", + serde_json::json!({"id": "p4", "outcome": "submitted_outcome_unknown", + "txHash": null, "message": msg}), + ) + .await; + assert_eq!(s, StatusCode::OK, "{v}"); + match waiter.await.unwrap() { + citrate_agent_core::hitl::Resolution::SubmittedOutcomeUnknown(u) => { + assert_eq!(u.message(), msg); + assert_eq!(u.tx_hash(), None); + } + other => panic!("{other:?}"), + } + let rows = fx.ceremony_outcomes(); + let (outcome, detail) = rows[0].1.clone().unwrap(); + assert_eq!(outcome, Outcome::OutcomeUnknown); + assert_eq!(detail, "Submitted · outcome unknown (tx not known)"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_id_only_approve_keeps_todays_behaviour() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_outcome(payment("p5")).await }); + fx.wait_depth(1).await; + let (s, v) = fx + .post("/approvals/approve", serde_json::json!({"id": "p5"})) + .await; + assert_eq!(s, StatusCode::OK); + assert!(v.get("outcome").is_none(), "{v}"); + assert_eq!(waiter.await.unwrap(), ApprovalOutcomePublic::Approved); + let rows = fx.ceremony_outcomes(); + assert_eq!(rows[0].1.as_ref().unwrap().0, Outcome::Completed); +} + +/// Every malformed outcome body is refused with 400 before anything is resolved or recorded. +#[tokio::test(flavor = "multi_thread")] +async fn malformed_outcome_fields_are_refused_and_resolve_nothing() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_outcome(payment("p6")).await }); + fx.wait_depth(1).await; + + let long = format!("{} {}", core_message(HASH), "x".repeat(400)); + let bad: Vec<(&str, serde_json::Value)> = vec![ + ( + "short hash", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":"0x1234","message":core_message("0x1234")}), + ), + ( + "no 0x", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":&HASH[2..],"message":core_message(&HASH[2..])}), + ), + ( + "non-hex", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":HASH.replace('a',"g"),"message":core_message(&HASH.replace('a',"g"))}), + ), + ( + "hash not a string", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":5,"message":"m"}), + ), + ( + "message too long", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":HASH,"message":long}), + ), + ( + "empty message", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":HASH,"message":""}), + ), + ( + "control chars", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":HASH,"message":format!("{}\n\u{7}", core_message(HASH))}), + ), + ( + "message omits the hash", + serde_json::json!({"id":"p6","outcome":"submitted_outcome_unknown","txHash":HASH,"message":"submitted, outcome unknown, check before retrying"}), + ), + ( + "unknown outcome", + serde_json::json!({"id":"p6","outcome":"completed","txHash":HASH,"message":core_message(HASH)}), + ), + ( + "outcome not a string", + serde_json::json!({"id":"p6","outcome":true}), + ), + ( + "hash without outcome", + serde_json::json!({"id":"p6","txHash":HASH}), + ), + ( + "message without outcome", + serde_json::json!({"id":"p6","message":"anything"}), + ), + ]; + for (why, body) in bad { + let (s, _) = fx.post("/approvals/approve", body).await; + assert_eq!(s, StatusCode::BAD_REQUEST, "{why} must be refused"); + assert_eq!(fx.st.queue.depth(), 1, "{why} must not resolve the head"); + } + // An unknown outcome is never a rejection. + let (s, _) = fx.post("/approvals/reject", unknown_body("p6")).await; + assert_eq!(s, StatusCode::BAD_REQUEST, "reject carries no outcome"); + assert_eq!(fx.st.queue.depth(), 1); + assert!(fx.ceremony_outcomes().is_empty(), "nothing was recorded"); + assert!(!waiter.is_finished()); + + let (s, _) = fx.post("/approvals/approve", unknown_body("p6")).await; + assert_eq!(s, StatusCode::OK, "the well-formed body still resolves it"); + let _ = waiter.await; +} + +/// No auto-retry: an identical payment proposed again after an unknown outcome parks for the +/// member (even under an auto-grant) and the approval card names the earlier transaction. +#[tokio::test(flavor = "multi_thread")] +async fn a_reproposal_references_the_earlier_tx_and_never_auto_approves() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let first = tokio::spawn(async move { q.submit_with_outcome(payment("p7")).await }); + fx.wait_depth(1).await; + let (s, _) = fx.post("/approvals/approve", unknown_body("p7")).await; + assert_eq!(s, StatusCode::OK); + let _ = first.await; + + // A different payment (same tool, other call) carries no such reference. + let q = fx.st.queue.clone(); + let other = tokio::spawn(async move { q.submit_with_outcome(payment("p8")).await }); + fx.wait_depth(1).await; + let cards = fx.get("/approvals").await; + assert!( + cards[0].get("priorOutcomeUnknown").is_none(), + "{}", + cards[0] + ); + let (s, _) = fx + .post("/approvals/reject", serde_json::json!({"id": "p8"})) + .await; + assert_eq!(s, StatusCode::OK); + let _ = other.await; + + // Even a live grant for the tool does not let the identical effect through unattended. + fx.st.queue.add_grant("pay::eth-send"); + let q = fx.st.queue.clone(); + let again = tokio::spawn(async move { q.submit_with_outcome(payment("p7")).await }); + fx.wait_depth(1).await; + assert!(!again.is_finished(), "the re-proposal waits for the member"); + + let cards = fx.get("/approvals").await; + let card = &cards[0]; + assert_eq!(card["id"], "p7"); + assert_eq!(card["priorOutcomeUnknown"]["txHash"], HASH, "{card}"); + assert_eq!(card["priorOutcomeUnknown"]["message"], core_message(HASH)); + let summary = card["summary"].as_str().unwrap(); + assert!( + summary.contains(HASH), + "the card names the earlier tx: {summary}" + ); + assert!(summary.contains("outcome unknown"), "{summary}"); + + let (s, _) = fx + .post("/approvals/reject", serde_json::json!({"id": "p7"})) + .await; + assert_eq!(s, StatusCode::OK); + assert_eq!(again.await.unwrap(), ApprovalOutcomePublic::Rejected); +} diff --git a/agent/core/src/capsule/dispatcher.rs b/agent/core/src/capsule/dispatcher.rs index 9a53f78..2514313 100644 --- a/agent/core/src/capsule/dispatcher.rs +++ b/agent/core/src/capsule/dispatcher.rs @@ -114,8 +114,35 @@ pub trait ApprovalGate: Send + Sync { /// when rejected. Blocks the calling thread until the decision /// is made (or a timeout fires, surfaced as `Err`). fn request(&self, req: ApprovalRequest) -> Result<(), String>; + + /// C1-V3-R — the full decision, including an effect the member approved that citrate-core + /// already submitted with an unknown outcome. The eth-send host fn calls this. The default + /// mirrors [`request`](Self::request), so a gate that cannot see outcomes keeps its meaning. + fn decide(&self, req: ApprovalRequest) -> GateDecision { + match self.request(req) { + Ok(()) => GateDecision::Approved, + Err(reason) => GateDecision::Refused(reason), + } + } } +/// C1-V3-R — what an [`ApprovalGate`] decided about one effect. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GateDecision { + /// Approved: the host performs the effect. + Approved, + /// Rejected or timed out, with the reason. + Refused(String), + /// Approved, and citrate-core already submitted the transaction; its outcome is unknown. + /// The host must NOT perform the effect (that would send it twice) and reports the message, + /// exactly as core sent it, as a non-final result. + SubmittedOutcomeUnknown(String), +} + +/// C1-V3-R — the error class an eth-send result carries when the transaction was submitted with +/// an unknown outcome. Not success and not a failure: check the transaction before retrying. +pub const CHAIN_SEND_OUTCOME_UNKNOWN: &str = "ChainSendOutcomeUnknown"; + /// Pluggable dispatcher for the `citrate:chain/eth-send` host /// function. Production impls sign + submit the calldata via /// `citrate-wallet-core::RpcClient::send_raw_transaction` and diff --git a/agent/core/src/capsule/linker.rs b/agent/core/src/capsule/linker.rs index 96d7b91..76a9282 100644 --- a/agent/core/src/capsule/linker.rs +++ b/agent/core/src/capsule/linker.rs @@ -455,8 +455,19 @@ impl LinkerBuilder { tier: store.data().risk_tier(), required_roles: store.data().required_roles().to_vec(), }; - if let Err(reason) = gate.request(req) { - return Ok((Err(format!("ChainSendApprovalRejected: {reason}")),)); + match gate.decide(req) { + crate::capsule::dispatcher::GateDecision::Approved => {} + crate::capsule::dispatcher::GateDecision::Refused(reason) => { + return Ok((Err(format!("ChainSendApprovalRejected: {reason}")),)); + } + // C1-V3-R: approved and already submitted by citrate-core, outcome unknown. + // Report core's words as a non-final result; never dispatch (no double send). + crate::capsule::dispatcher::GateDecision::SubmittedOutcomeUnknown(message) => { + return Ok((Err(format!( + "{}: {message}", + crate::capsule::dispatcher::CHAIN_SEND_OUTCOME_UNKNOWN + )),)); + } } // Approval granted. Resolve the tx hash: diff --git a/agent/core/src/capsule/mod.rs b/agent/core/src/capsule/mod.rs index 4497608..c127a3a 100644 --- a/agent/core/src/capsule/mod.rs +++ b/agent/core/src/capsule/mod.rs @@ -3032,6 +3032,86 @@ tier = "bundled" ); } + /// C1-V3-R — a gate whose effect core already submitted with an unknown outcome. + #[cfg(test)] + struct UnknownOutcomeGate { + message: String, + } + + #[cfg(test)] + impl dispatcher::ApprovalGate for UnknownOutcomeGate { + fn request(&self, _req: dispatcher::ApprovalRequest) -> Result<(), String> { + Err(self.message.clone()) + } + fn decide(&self, _req: dispatcher::ApprovalRequest) -> dispatcher::GateDecision { + dispatcher::GateDecision::SubmittedOutcomeUnknown(self.message.clone()) + } + } + + /// C1-V3-R — the member approved, core broadcast, the outcome is unknown. The capsule gets + /// exactly core's message under the non-final `ChainSendOutcomeUnknown` class (never a tx + /// hash as success), and the dispatcher is never consulted, so nothing is sent twice. + #[test] + fn eth_send_capsule_reports_an_unknown_outcome_and_never_dispatches() { + use std::sync::Arc; + + let message = "submitted, outcome unknown, tx 0x8f3c2a1b9d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8, check before retrying".to_string(); + let gate = Arc::new(UnknownOutcomeGate { + message: message.clone(), + }); + let dispatcher_impl = Arc::new(MockSendDispatcher { + calls: std::sync::Mutex::new(Vec::new()), + tx_hash: [0xeeu8; 32], + }); + let (mut store, instance) = load_eth_sender_capsule_with( + ETH_SENDER_MANIFEST_OK, + Some(gate), + Some(dispatcher_impl.clone()), + ); + let authorized = + hex::decode("4a86659BDab24dc444C72fbbaD4cd83491820E40").unwrap(); + let result = invoke_eth_sender_send(&mut store, instance, authorized, vec![0x01]); + let err = result.expect_err("an unknown outcome is not success"); + assert_eq!( + err, + format!("{}: {message}", dispatcher::CHAIN_SEND_OUTCOME_UNKNOWN) + ); + assert_eq!( + dispatcher_impl.calls.lock().unwrap().len(), + 0, + "core already broadcast it; the host must not send it again" + ); + } + + /// A gate that only implements `request` keeps its meaning through the default `decide`. + #[test] + fn the_default_decide_mirrors_request() { + let allow = MockGate { + decisions: std::sync::Mutex::new(Vec::new()), + allow: true, + reason: String::new(), + }; + let deny = MockGate { + decisions: std::sync::Mutex::new(Vec::new()), + allow: false, + reason: "no".to_string(), + }; + let req = || dispatcher::ApprovalRequest { + capsule_name: "c".to_string(), + method: "eth-send".to_string(), + to: [0u8; 20], + data: vec![], + tier: manifest::RiskTier::Low, + required_roles: vec![], + }; + use dispatcher::ApprovalGate; + assert_eq!(allow.decide(req()), dispatcher::GateDecision::Approved); + assert_eq!( + deny.decide(req()), + dispatcher::GateDecision::Refused("no".to_string()) + ); + } + // ────────────────────── (end CIT-AGENT-9c-write-host) ───────── // ────────────────── CIT-AGENT-9c-write-tools (5+6+7) ────────── diff --git a/agent/core/src/capsule/prod_impls.rs b/agent/core/src/capsule/prod_impls.rs index 14efa92..7e40f54 100644 --- a/agent/core/src/capsule/prod_impls.rs +++ b/agent/core/src/capsule/prod_impls.rs @@ -11,11 +11,11 @@ use crate::audit::recorder::RecorderClient; use crate::capsule::dispatcher::{ - ApprovalGate, ApprovalRequest, EthSendDispatcher, + ApprovalGate, ApprovalRequest, EthSendDispatcher, GateDecision, }; use crate::capsule::wasm::Address; use crate::hitl::quorum::Quorum; -use crate::hitl::{ApprovalOutcomePublic, ApprovalQueue, Signer, ToolCall}; +use crate::hitl::{ApprovalOutcomePublic, ApprovalQueue, Resolution, Signer, ToolCall}; use serde_json::json; use std::sync::Arc; @@ -61,7 +61,18 @@ impl QueuedApprovalGate { } impl ApprovalGate for QueuedApprovalGate { + /// The plain form of [`decide`](ApprovalGate::decide): only an approval is `Ok`. An unknown + /// outcome is `Err(message)`, so a caller of `request` never performs an effect that + /// citrate-core already submitted. fn request(&self, req: ApprovalRequest) -> Result<(), String> { + match self.decide(req) { + GateDecision::Approved => Ok(()), + GateDecision::Refused(reason) => Err(reason), + GateDecision::SubmittedOutcomeUnknown(message) => Err(message), + } + } + + fn decide(&self, req: ApprovalRequest) -> GateDecision { // Build the ToolCall the queue's UI surfaces will display. // call_id is derived from the (capsule, to, data) tuple via // SHA3 so retries with identical args collide on the same @@ -108,11 +119,17 @@ impl ApprovalGate for QueuedApprovalGate { if matches!(quorum, Quorum::AutoApprove) { // Tier-low: the legacy FIFO fast path is acceptable. - let outcome = tokio::task::block_in_place(move || { + // C1-V3-R: the FIFO track can carry an unknown outcome from core's ceremony. + let resolution = tokio::task::block_in_place(move || { let handle = tokio::runtime::Handle::current(); - handle.block_on(async move { queue.submit_with_outcome(call).await }) + handle.block_on(async move { queue.submit_with_resolution(call).await }) }); - return outcome_to_result(outcome); + return match resolution { + Resolution::Decided(o) => outcome_to_decision(o), + Resolution::SubmittedOutcomeUnknown(u) => { + GateDecision::SubmittedOutcomeUnknown(u.message().to_string()) + } + }; } // Tier medium/high/critical: route through the role-aware quorum. @@ -125,7 +142,7 @@ impl ApprovalGate for QueuedApprovalGate { // propose, so refuse now rather than park the effect (and block this // worker) for the 5-minute timeout. let Some(proposer) = self.proposer.clone() else { - return Err(format!( + return GateDecision::Refused(format!( "{} effect from capsule {:?} refused: tier {:?} needs a role quorum and no \ invoking human is bound to this approval gate", req.method, req.capsule_name, req.tier @@ -140,7 +157,7 @@ impl ApprovalGate for QueuedApprovalGate { .await }) }); - outcome_to_result(outcome) + outcome_to_decision(outcome) } } @@ -159,12 +176,18 @@ fn signing_payload(req: &ApprovalRequest) -> Vec { .into_bytes() } -fn outcome_to_result(outcome: ApprovalOutcomePublic) -> Result<(), String> { +fn outcome_to_decision(outcome: ApprovalOutcomePublic) -> GateDecision { match outcome { - ApprovalOutcomePublic::Approved => Ok(()), - ApprovalOutcomePublic::AutoApproved => Ok(()), - ApprovalOutcomePublic::Rejected => Err("rejected by operator".to_string()), - ApprovalOutcomePublic::TimedOut => Err("approval timed out (5min)".to_string()), + ApprovalOutcomePublic::Approved => GateDecision::Approved, + ApprovalOutcomePublic::AutoApproved => GateDecision::Approved, + ApprovalOutcomePublic::Rejected => GateDecision::Refused("rejected by operator".to_string()), + ApprovalOutcomePublic::TimedOut => { + GateDecision::Refused("approval timed out (5min)".to_string()) + } + // Reached only without the details (the role track); fail closed, never approved. + ApprovalOutcomePublic::SubmittedOutcomeUnknown => GateDecision::SubmittedOutcomeUnknown( + "submitted, outcome unknown, check recent transactions before retrying".to_string(), + ), } } @@ -384,6 +407,50 @@ mod tests { assert!(h.await.expect("join").is_err()); } + /// C1-V3-R: a tier-low effect released by core as submitted with an unknown outcome reaches + /// the host as `SubmittedOutcomeUnknown` with core's exact words; `request` is never `Ok`. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn queued_gate_passes_an_unknown_outcome_through() { + let queue = Arc::new(ApprovalQueue::new()); + let gate = Arc::new(QueuedApprovalGate::new(queue.clone())); + let tx = "0x8f3c2a1b9d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8"; + let words = format!("submitted, outcome unknown, tx {tx}, check before retrying"); + let (addr, data) = ([0xa8u8; 20], vec![7u8]); + let call_id = call_id_for("pay", "eth-send", &addr, &data); + let req = || ApprovalRequest { + capsule_name: "pay".to_string(), + method: "eth-send".to_string(), + to: addr, + data: data.clone(), + tier: RiskTier::Low, + required_roles: vec![], + }; + for use_decide in [true, false] { + let g = gate.clone(); + let r = req(); + let h = tokio::spawn(async move { + if use_decide { + Ok(g.decide(r)) + } else { + g.request(r).map(|_| GateDecision::Approved) + } + }); + while queue.depth() < 1 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let u = crate::hitl::UnknownOutcome::new(Some(tx), Some(&words)).unwrap(); + queue + .approve_by_id_outcome_unknown(&call_id, u) + .expect("the head matches"); + let got = h.await.expect("join"); + if use_decide { + assert_eq!(got, Ok(GateDecision::SubmittedOutcomeUnknown(words.clone()))); + } else { + assert_eq!(got, Err(words.clone()), "request() must never read it as approved"); + } + } + } + /// PBA-L6b-012 tripwire: the production gate must never again record a /// non-human (capsule-derived) proposer for a role-track effect. #[test] diff --git a/agent/core/src/hitl/mod.rs b/agent/core/src/hitl/mod.rs index 79d3806..f2fcd7f 100644 --- a/agent/core/src/hitl/mod.rs +++ b/agent/core/src/hitl/mod.rs @@ -21,6 +21,7 @@ //! description}). Until then they stay here as the agreed defaults. pub mod break_glass; +pub mod outcome; pub mod quorum; pub mod roles; pub mod signing; @@ -28,6 +29,10 @@ pub mod signing; pub use break_glass::{ BreakGlassEntry, BreakGlassError, BreakGlassPhase, BreakGlassRegistry, AFFIRMATION_WINDOW, }; +pub use outcome::{ + is_tx_hash, UnknownOutcome, MAX_OUTCOME_MESSAGE, OUTCOME_SUBMITTED_UNKNOWN, + REMEMBERED_UNKNOWN_OUTCOMES, +}; pub use quorum::Quorum; pub use roles::{can_approve, is_conflict}; pub use signing::{ @@ -77,6 +82,9 @@ pub struct PendingView { pub description: String, pub risk_level: String, pub args_pretty: String, + /// C1-V3-R: an identical call was approved earlier and submitted with an unknown outcome. + /// The approval card names that transaction so the member checks it before approving again. + pub prior_unknown: Option, } // ── Approval queue ───────────────────────────────────────────────── @@ -91,10 +99,12 @@ struct PendingEntry { nonce: u64, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] enum ApprovalOutcome { Approved, Rejected, + /// C1-V3-R: approved; citrate-core submitted the transaction and its outcome is unknown. + SubmittedOutcomeUnknown(UnknownOutcome), } /// BFR-INT-12b WP-5 — default Pending deadline. If the operator @@ -116,6 +126,21 @@ pub enum ApprovalOutcomePublic { Rejected, TimedOut, AutoApproved, + /// C1-V3-R: the member approved and the transaction was submitted (by citrate-core's + /// ceremony), but its outcome is unknown. Not success and not final: the submitter must not + /// perform or re-send the effect. [`ApprovalQueue::submit_with_resolution`] carries the hash + /// and the message. + SubmittedOutcomeUnknown, +} + +/// C1-V3-R: how a FIFO submission ended, with the details an unknown outcome carries. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Resolution { + /// Approved, auto-approved, rejected or timed out. + Decided(ApprovalOutcomePublic), + /// Approved and submitted by citrate-core, outcome unknown: the submitter is told exactly + /// [`UnknownOutcome::message`] and does not perform the effect itself. + SubmittedOutcomeUnknown(UnknownOutcome), } // ── CIT-AGENT-4a: Signer / Signature types ───────────────────────── @@ -268,6 +293,10 @@ pub struct ApprovalQueue { // PBA-L6b-010 — set by the emergency stop. While frozen every new // submission is refused at once (never parked) and the queue is empty. frozen: std::sync::atomic::AtomicBool, + // C1-V3-R — call ids approved and submitted with an unknown outcome, oldest first, at most + // `REMEMBERED_UNKNOWN_OUTCOMES`. An identical re-proposal is never auto-granted and its + // pending view names the earlier transaction. + unknown_outcomes: Mutex>, } /// PBA-L6b-009 — removes a FIFO submission's entry when its waiter goes away @@ -328,17 +357,30 @@ impl ApprovalQueue { } /// Same as [`submit`] but returns the rich outcome so callers - /// can write a faithful decision-log row. + /// can write a faithful decision-log row. An unknown outcome is + /// [`ApprovalOutcomePublic::SubmittedOutcomeUnknown`]; use + /// [`submit_with_resolution`](Self::submit_with_resolution) for its hash and message. pub async fn submit_with_outcome(&self, call: ToolCall) -> ApprovalOutcomePublic { + match self.submit_with_resolution(call).await { + Resolution::Decided(o) => o, + Resolution::SubmittedOutcomeUnknown(_) => ApprovalOutcomePublic::SubmittedOutcomeUnknown, + } + } + + /// C1-V3-R — submit a tool call and learn how it ended, including an approved effect that + /// citrate-core submitted with an unknown outcome (with the hash and the message to report). + pub async fn submit_with_resolution(&self, call: ToolCall) -> Resolution { + use ApprovalOutcomePublic as P; // PBA-L6b-010: after an emergency stop nothing new is queued — not // even an auto-granted call. if self.is_frozen() { - return ApprovalOutcomePublic::Rejected; + return Resolution::Decided(P::Rejected); } // BFR-INT-12b WP-4 — auto-grant fast path. Resolved before - // the queue is touched. - if self.is_trusted(&call.name) { - return ApprovalOutcomePublic::AutoApproved; + // the queue is touched. C1-V3-R: never for a call whose earlier identical submission has + // an unknown outcome; a member must look at it again (no unattended re-send). + if self.is_trusted(&call.name) && self.prior_unknown_outcome(&call.call_id).is_none() { + return Resolution::Decided(P::AutoApproved); } let (tx, rx) = oneshot::channel(); // AR-B-023: remember which entry this future owns so a timeout evicts @@ -351,18 +393,18 @@ impl ApprovalQueue { { let mut q = match self.pending.lock() { Ok(q) => q, - Err(_) => return ApprovalOutcomePublic::Rejected, + Err(_) => return Resolution::Decided(P::Rejected), }; // PBA-L6b-010: re-check under the lock so a submission racing // `freeze_and_drain` cannot slip in after the drain. if self.is_frozen() { - return ApprovalOutcomePublic::Rejected; + return Resolution::Decided(P::Rejected); } // PBA-L6b-012: an identical effect (same call_id) is already // waiting for a human. Refuse the duplicate instead of queueing a // second entry an id-bound approve could not tell apart. if q.iter().any(|e| e.call.call_id == call_id) { - return ApprovalOutcomePublic::Rejected; + return Resolution::Decided(P::Rejected); } q.push_back(PendingEntry { call, @@ -380,10 +422,13 @@ impl ApprovalQueue { // BFR-INT-12b WP-5 — race the resolver against the 5-min // timeout. match tokio::time::timeout(PENDING_TIMEOUT, rx).await { - Ok(Ok(ApprovalOutcome::Approved)) => ApprovalOutcomePublic::Approved, - Ok(Ok(ApprovalOutcome::Rejected)) => ApprovalOutcomePublic::Rejected, - Ok(Err(_)) => ApprovalOutcomePublic::Rejected, // sender dropped - Err(_) => ApprovalOutcomePublic::TimedOut, + Ok(Ok(ApprovalOutcome::Approved)) => Resolution::Decided(P::Approved), + Ok(Ok(ApprovalOutcome::Rejected)) => Resolution::Decided(P::Rejected), + Ok(Ok(ApprovalOutcome::SubmittedOutcomeUnknown(u))) => { + Resolution::SubmittedOutcomeUnknown(u) + } + Ok(Err(_)) => Resolution::Decided(P::Rejected), // sender dropped + Err(_) => Resolution::Decided(P::TimedOut), } } @@ -435,6 +480,43 @@ impl ApprovalQueue { self.resolve_head_by_id(call_id, ApprovalOutcome::Approved) } + /// C1-V3-R — approve a specific pending call by id (exactly as [`approve_by_id`]) whose chain + /// effect citrate-core already submitted with an unknown outcome. The submitter receives + /// [`Resolution::SubmittedOutcomeUnknown`] (never plain approval, so it does not perform the + /// effect again), and the call id is remembered so an identical re-proposal is never + /// auto-granted and names the earlier transaction. The transaction went out whatever the + /// queue's state, so the call id is remembered even when the head no longer matches. + pub fn approve_by_id_outcome_unknown( + &self, + call_id: &str, + outcome: UnknownOutcome, + ) -> Result<(), SignatureError> { + self.remember_unknown_outcome(call_id, outcome.clone()); + self.resolve_head_by_id(call_id, ApprovalOutcome::SubmittedOutcomeUnknown(outcome)) + } + + /// C1-V3-R — the unknown outcome of an earlier identical call, if one is remembered. + pub fn prior_unknown_outcome(&self, call_id: &str) -> Option { + let m = self.unknown_outcomes.lock().ok()?; + m.iter().find(|(id, _)| id == call_id).map(|(_, u)| u.clone()) + } + + /// C1-V3-R — how many unknown outcomes are remembered (at most + /// [`REMEMBERED_UNKNOWN_OUTCOMES`]). + pub fn remembered_unknown_outcomes(&self) -> usize { + self.unknown_outcomes.lock().map(|m| m.len()).unwrap_or(0) + } + + fn remember_unknown_outcome(&self, call_id: &str, outcome: UnknownOutcome) { + if let Ok(mut m) = self.unknown_outcomes.lock() { + m.retain(|(id, _)| id != call_id); + m.push_back((call_id.to_string(), outcome)); + while m.len() > REMEMBERED_UNKNOWN_OUTCOMES { + m.pop_front(); + } + } + } + /// AR-B-023 — reject a specific pending call by id (see [`approve_by_id`]). pub fn reject_by_id(&self, call_id: &str) -> Result<(), SignatureError> { self.resolve_head_by_id(call_id, ApprovalOutcome::Rejected) @@ -495,11 +577,15 @@ impl ApprovalQueue { let entry = q.front()?; let args_pretty = serde_json::to_string_pretty(&entry.call.args) .unwrap_or_else(|_| entry.call.args.to_string()); + let id = entry.call.call_id.clone(); + let name = entry.call.name.clone(); + drop(q); Some(PendingView { - id: entry.call.call_id.clone(), - name: entry.call.name.clone(), - description: describe(&entry.call.name), - risk_level: risk_level(&entry.call.name).to_string(), + prior_unknown: self.prior_unknown_outcome(&id), + description: describe(&name), + risk_level: risk_level(&name).to_string(), + id, + name, args_pretty, }) } @@ -593,6 +679,11 @@ impl ApprovalQueue { let outcome = match tokio::time::timeout(PENDING_TIMEOUT, rx).await { Ok(Ok(ApprovalOutcome::Approved)) => ApprovalOutcomePublic::Approved, Ok(Ok(ApprovalOutcome::Rejected)) => ApprovalOutcomePublic::Rejected, + // Only the FIFO track's approve route carries an outcome; the role track never + // receives one. Read it as not approved (fail closed). + Ok(Ok(ApprovalOutcome::SubmittedOutcomeUnknown(_))) => { + ApprovalOutcomePublic::SubmittedOutcomeUnknown + } Ok(Err(_)) => ApprovalOutcomePublic::Rejected, Err(_) => ApprovalOutcomePublic::TimedOut, }; @@ -1518,4 +1609,121 @@ mod tests { assert_eq!(risk_level("revoke_role"), "medium"); assert_eq!(risk_level("provision_user"), "high"); } + + // ── C1-V3-R: approved, submitted, outcome unknown ───────────────── + + const TX: &str = "0x8f3c2a1b9d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8"; + + fn unknown() -> UnknownOutcome { + UnknownOutcome::new( + Some(TX), + Some(&format!("submitted, outcome unknown, tx {TX}, check before retrying")), + ) + .expect("core's words are valid") + } + + #[tokio::test] + async fn an_unknown_outcome_reaches_the_submitter_with_its_message() { + let q = Arc::new(ApprovalQueue::new()); + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit_with_resolution(mkcall("pay")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + q.approve_by_id_outcome_unknown("call_pay", unknown()) + .expect("the head matches"); + assert_eq!(h.await.unwrap(), Resolution::SubmittedOutcomeUnknown(unknown())); + } + + #[tokio::test] + async fn an_unknown_outcome_is_never_a_plain_approval() { + let q = Arc::new(ApprovalQueue::new()); + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit(mkcall("pay_bool")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + q.approve_by_id_outcome_unknown("call_pay_bool", unknown()) + .expect("the head matches"); + assert!(!h.await.unwrap(), "submit() must not report an unknown outcome as approved"); + + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit_with_outcome(mkcall("pay_rich")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + q.approve_by_id_outcome_unknown("call_pay_rich", unknown()) + .expect("the head matches"); + assert_eq!(h.await.unwrap(), ApprovalOutcomePublic::SubmittedOutcomeUnknown); + } + + #[tokio::test] + async fn an_unknown_outcome_on_a_different_head_resolves_nothing() { + let q = Arc::new(ApprovalQueue::new()); + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit_with_outcome(mkcall("head")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + assert_eq!( + q.approve_by_id_outcome_unknown("call_other", unknown()), + Err(SignatureError::UnknownCallId) + ); + assert_eq!(q.depth(), 1); + q.reject_by_id("call_head").expect("cleanup"); + assert_eq!(h.await.unwrap(), ApprovalOutcomePublic::Rejected); + } + + /// No auto-retry: after an unknown outcome the identical call is never auto-granted, and the + /// pending view names the earlier transaction. + #[tokio::test] + async fn a_reproposal_after_an_unknown_outcome_waits_for_a_human() { + let q = Arc::new(ApprovalQueue::new()); + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit_with_outcome(mkcall("again")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + assert!(q.peek().unwrap().prior_unknown.is_none(), "first proposal: no history"); + q.approve_by_id_outcome_unknown("call_again", unknown()) + .expect("the head matches"); + let _ = h.await; + assert_eq!(q.prior_unknown_outcome("call_again"), Some(unknown())); + + q.add_grant("again"); + let qa = q.clone(); + let h = tokio::spawn(async move { qa.submit_with_outcome(mkcall("again")).await }); + while q.depth() < 1 { + tokio::task::yield_now().await; + } + let view = q.peek().expect("the re-proposal is pending, not auto-granted"); + assert_eq!(view.prior_unknown, Some(unknown())); + q.reject_by_id("call_again").expect("cleanup"); + assert_eq!(h.await.unwrap(), ApprovalOutcomePublic::Rejected); + + // An unrelated granted tool still takes the fast path. + q.add_grant("other"); + assert_eq!( + q.submit_with_outcome(mkcall("other")).await, + ApprovalOutcomePublic::AutoApproved + ); + } + + #[test] + fn unknown_outcomes_are_remembered_within_a_bound() { + let q = ApprovalQueue::new(); + for i in 0..(REMEMBERED_UNKNOWN_OUTCOMES + 10) { + // Nothing is pending, so each resolve fails, but the submitted tx is still remembered. + let _ = q.approve_by_id_outcome_unknown(&format!("c{i}"), unknown()); + } + assert!(q.prior_unknown_outcome("c0").is_none(), "the oldest is dropped"); + assert!(q + .prior_unknown_outcome(&format!("c{}", REMEMBERED_UNKNOWN_OUTCOMES + 9)) + .is_some()); + assert_eq!(q.remembered_unknown_outcomes(), REMEMBERED_UNKNOWN_OUTCOMES); + // The same id twice is one entry. + let _ = q.approve_by_id_outcome_unknown("c300", unknown()); + let _ = q.approve_by_id_outcome_unknown("c300", unknown()); + assert_eq!(q.remembered_unknown_outcomes(), REMEMBERED_UNKNOWN_OUTCOMES); + } } diff --git a/agent/core/src/hitl/outcome.rs b/agent/core/src/hitl/outcome.rs new file mode 100644 index 0000000..bf9186d --- /dev/null +++ b/agent/core/src/hitl/outcome.rs @@ -0,0 +1,175 @@ +//! C1-V3-R — an approved chain effect whose transaction was submitted but whose outcome is unknown. +//! +//! Data source (Rule 7): citrate-core's signature ceremony. When the member approved an effect and +//! core broadcast it but could not learn the result (C1-V3, `hermes_resolve_outcome_unknown`), +//! core releases the effect through the sidecar's approve route with +//! `{outcome: "submitted_outcome_unknown", txHash, message}`. The submitter is then told exactly +//! that message, never success, and an identical re-proposal must reference the earlier +//! transaction instead of being sent again unattended. + +/// The wire value of `outcome` on the approve route for a submitted, unknown-outcome effect. +pub const OUTCOME_SUBMITTED_UNKNOWN: &str = "submitted_outcome_unknown"; + +/// The longest message accepted, in bytes. Core's words with a hash are about 100 bytes. +pub const MAX_OUTCOME_MESSAGE: usize = 280; + +/// How many unknown-outcome call ids the queue remembers for re-proposals (oldest dropped first). +pub const REMEMBERED_UNKNOWN_OUTCOMES: usize = 256; + +/// True for `0x` followed by exactly 64 hex digits. +pub fn is_tx_hash(s: &str) -> bool { + s.strip_prefix("0x") + .is_some_and(|h| h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit())) +} + +/// The words used when core sent an outcome without its own message (the same words core uses). +fn default_message(tx_hash: Option<&str>) -> String { + match tx_hash { + Some(h) => format!("submitted, outcome unknown, tx {h}, check before retrying"), + None => "submitted, outcome unknown, check recent transactions before retrying".to_string(), + } +} + +/// A submitted transaction with no known outcome: its hash when core had one, and the message the +/// submitter (a capsule, through its tool result) must be given. Built only through [`Self::new`], +/// so every value is validated. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnknownOutcome { + tx_hash: Option, + message: String, +} + +impl UnknownOutcome { + /// Validate core's fields. `tx_hash` must be `0x` + 64 hex digits when present. `message` + /// (core's words; `None` composes the same words here) must be non-empty, at most + /// [`MAX_OUTCOME_MESSAGE`] bytes, free of control characters, and must name the hash when + /// there is one, so the submitter is always told which transaction to check. + pub fn new(tx_hash: Option<&str>, message: Option<&str>) -> Result { + if let Some(h) = tx_hash { + if !is_tx_hash(h) { + return Err("txHash must be 0x followed by 64 hex digits".to_string()); + } + } + let message = match message { + Some(m) => m.to_string(), + None => default_message(tx_hash), + }; + if message.trim().is_empty() { + return Err("message must not be empty".to_string()); + } + if message.len() > MAX_OUTCOME_MESSAGE { + return Err(format!( + "message must be at most {MAX_OUTCOME_MESSAGE} bytes" + )); + } + if message.chars().any(char::is_control) { + return Err("message must not contain control characters".to_string()); + } + if let Some(h) = tx_hash { + if !message.contains(h) { + return Err("message must name the transaction hash".to_string()); + } + } + Ok(Self { + tx_hash: tx_hash.map(str::to_string), + message, + }) + } + + /// The submitted transaction's hash, when core had one. + pub fn tx_hash(&self) -> Option<&str> { + self.tx_hash.as_deref() + } + + /// The words the submitter must be given, exactly as core sent them. + pub fn message(&self) -> &str { + &self.message + } + + /// The decision record's outcome detail: "Submitted · outcome unknown (tx )". + pub fn record_detail(&self) -> String { + format!( + "Submitted · outcome unknown (tx {})", + self.tx_hash.as_deref().unwrap_or("not known") + ) + } + + /// The note an identical re-proposal's approval card carries. + pub fn reproposal_note(&self) -> String { + format!( + "an earlier identical request was submitted, outcome unknown (tx {}); check it before approving again", + self.tx_hash.as_deref().unwrap_or("not known") + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const HASH: &str = "0x8f3c2a1b9d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8"; + + #[test] + fn core_words_with_a_hash_are_accepted_verbatim() { + let m = format!("submitted, outcome unknown, tx {HASH}, check before retrying"); + let u = UnknownOutcome::new(Some(HASH), Some(&m)).unwrap(); + assert_eq!(u.message(), m); + assert_eq!(u.tx_hash(), Some(HASH)); + assert_eq!( + u.record_detail(), + format!("Submitted · outcome unknown (tx {HASH})") + ); + assert!(u.reproposal_note().contains(HASH)); + } + + #[test] + fn a_missing_message_gets_cores_words() { + let u = UnknownOutcome::new(Some(HASH), None).unwrap(); + assert_eq!( + u.message(), + format!("submitted, outcome unknown, tx {HASH}, check before retrying") + ); + let none = UnknownOutcome::new(None, None).unwrap(); + assert_eq!( + none.message(), + "submitted, outcome unknown, check recent transactions before retrying" + ); + assert_eq!( + none.record_detail(), + "Submitted · outcome unknown (tx not known)" + ); + } + + #[test] + fn tx_hashes_are_0x_and_64_hex_digits() { + assert!(is_tx_hash(HASH)); + assert!(is_tx_hash(&HASH.to_uppercase().replacen("0X", "0x", 1))); + for bad in [ + "", + "0x", + &HASH[2..], + &format!("{HASH}00"), + &HASH.replace('a', "g"), + ] { + assert!(!is_tx_hash(bad), "{bad:?}"); + assert!( + UnknownOutcome::new(Some(bad), Some("m")).is_err(), + "{bad:?}" + ); + } + } + + #[test] + fn messages_are_bounded_and_must_name_the_hash() { + let ok = format!("submitted, outcome unknown, tx {HASH}, check before retrying"); + let long = format!("{ok}{}", "x".repeat(MAX_OUTCOME_MESSAGE)); + assert!(UnknownOutcome::new(Some(HASH), Some(&long)).is_err()); + let exact = format!("{ok}{}", "x".repeat(MAX_OUTCOME_MESSAGE - ok.len())); + assert!(UnknownOutcome::new(Some(HASH), Some(&exact)).is_ok()); + assert!(UnknownOutcome::new(Some(HASH), Some("")).is_err()); + assert!(UnknownOutcome::new(Some(HASH), Some(" ")).is_err()); + assert!(UnknownOutcome::new(Some(HASH), Some(&format!("{ok}\n"))).is_err()); + assert!(UnknownOutcome::new(Some(HASH), Some("submitted, outcome unknown")).is_err()); + assert!(UnknownOutcome::new(None, Some("submitted, outcome unknown")).is_ok()); + } +} From 8f9d459ddaff3dc55a6bd41ef4f827d8894668d7 Mon Sep 17 00:00:00 2001 From: Larry Klosowski Date: Fri, 23 Oct 2026 00:46:14 -0700 Subject: [PATCH 2/3] test(sidecar): red, an unknown outcome survives a failing record and a late callback (C1-V3-R) Four new tests fail on assertions against 506b779: - an_unknown_outcome_is_delivered_even_when_the_record_cannot_be_written (503 today) - a_late_unknown_outcome_is_recorded_after_the_waiter_is_gone (no receipt today) - a_late_unknown_outcome_never_touches_the_current_head (no receipt today) - every_unknown_outcome_callback_keeps_the_transaction (state-matrix tripwire) Control: an_ordinary_approval_is_still_refused_when_the_record_cannot_be_written keeps the write-ahead refusal for plain approvals. Co-Authored-By: Claude Opus 5.5 (1M context) --- agent-sidecar/src/outcome_unknown_tests.rs | 239 ++++++++++++++++++++- 1 file changed, 237 insertions(+), 2 deletions(-) diff --git a/agent-sidecar/src/outcome_unknown_tests.rs b/agent-sidecar/src/outcome_unknown_tests.rs index f5344ec..aef45aa 100644 --- a/agent-sidecar/src/outcome_unknown_tests.rs +++ b/agent-sidecar/src/outcome_unknown_tests.rs @@ -33,9 +33,12 @@ struct Fx { impl Fx { fn new() -> Self { + Self::new_with(LogConfig::default()) + } + + fn new_with(cfg: LogConfig) -> Self { let root = tempfile::tempdir().unwrap(); - let (log, _) = - DecisionLog::open(&root.path().join("records"), LogConfig::default()).unwrap(); + let (log, _) = DecisionLog::open(&root.path().join("records"), cfg).unwrap(); let log = Arc::new(log); let client: Arc = Arc::new(Quiet); let mgr = sessions::SessionManager::new( @@ -385,3 +388,235 @@ async fn a_reproposal_references_the_earlier_tx_and_never_auto_approves() { assert_eq!(s, StatusCode::OK); assert_eq!(again.await.unwrap(), ApprovalOutcomePublic::Rejected); } + +// ── Round 2: the outcome survives a failing record and a late callback ──────────────────────── + +impl Fx { + /// A records log that refuses every write after the first: each record needs a new segment, + /// and the directory is made read-only, so creating it fails. `heal` restores the directory. + fn with_failing_log() -> Self { + let fx = Fx::new_with(LogConfig { + max_segment_bytes: 1, + max_segments: None, + }); + let log = fx.st.sessions.records().expect("a records log"); + hic_records::record_member_decision(&log, "test.seed", "seed", false, "seed", vec![]) + .expect("the first record fits the open segment"); + fx.set_records_writable(false); + fx + } + + fn set_records_writable(&self, writable: bool) { + use std::os::unix::fs::PermissionsExt; + let mode = if writable { 0o700 } else { 0o500 }; + std::fs::set_permissions( + self.root.path().join("records"), + std::fs::Permissions::from_mode(mode), + ) + .unwrap(); + } + + /// The ceremony decisions bound to `call_id` (by their evidence), with their closing outcome. + fn records_for(&self, call_id: &str) -> Vec<(Decision, Option<(Outcome, String)>)> { + let recs = self.records(); + let uri = format!("sidecar:approval/{call_id}"); + recs.iter() + .filter_map(|r| match &r.record.entry { + Entry::Decision(d) + if d.kind == "ceremony.capsule_effect" + && d.evidence.iter().any(|e| e.uri == uri) => + { + let seq = r.record.seq; + let out = recs.iter().find_map(|o| match &o.record.entry { + Entry::Outcome(o) if o.decision_seq == seq => { + Some((o.outcome, o.detail.clone())) + } + _ => None, + }); + Some((d.decision, out)) + } + _ => None, + }) + .collect() + } +} + +/// Codex P2 (round 1): the transaction already went out, so a decision record that cannot be +/// written must not hide the outcome. The capsule is told, the tx is remembered, and the route +/// says the record is missing instead of refusing. +#[tokio::test(flavor = "multi_thread")] +async fn an_unknown_outcome_is_delivered_even_when_the_record_cannot_be_written() { + let fx = Fx::with_failing_log(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_resolution(payment("f1")).await }); + fx.wait_depth(1).await; + + let (s, v) = fx.post("/approvals/approve", unknown_body("f1")).await; + fx.set_records_writable(true); + assert_eq!(s, StatusCode::OK, "the outcome is not refused: {v}"); + assert_eq!(v["outcome"], "submitted_outcome_unknown"); + assert_eq!(v["recorded"], false, "the route says the record is missing"); + match tokio::time::timeout(Duration::from_secs(5), waiter) + .await + .expect("the waiter is told") + .unwrap() + { + citrate_agent_core::hitl::Resolution::SubmittedOutcomeUnknown(u) => { + assert_eq!(u.message(), core_message(HASH)) + } + other => panic!("{other:?}"), + } + assert_eq!( + fx.st + .queue + .prior_unknown_outcome("f1") + .and_then(|u| u.tx_hash().map(str::to_string)), + Some(HASH.to_string()), + "an identical re-proposal still names the tx" + ); +} + +/// The write-ahead refusal stays for an ordinary approval: nothing is released unrecorded. +#[tokio::test(flavor = "multi_thread")] +async fn an_ordinary_approval_is_still_refused_when_the_record_cannot_be_written() { + let fx = Fx::with_failing_log(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_outcome(payment("f2")).await }); + fx.wait_depth(1).await; + let (s, _) = fx + .post("/approvals/approve", serde_json::json!({"id": "f2"})) + .await; + fx.set_records_writable(true); + assert_eq!(s, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(fx.st.queue.depth(), 1, "nothing was released"); + assert!(!waiter.is_finished()); + fx.st.queue.reject_by_id("f2").unwrap(); + let _ = waiter.await; +} + +/// Codex P2 (round 1): a callback that arrives after the waiter went away still writes the +/// transaction into the decision log, bound to its call id. +#[tokio::test(flavor = "multi_thread")] +async fn a_late_unknown_outcome_is_recorded_after_the_waiter_is_gone() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let waiter = tokio::spawn(async move { q.submit_with_resolution(payment("l1")).await }); + fx.wait_depth(1).await; + waiter.abort(); + let _ = waiter.await; + fx.wait_depth(0).await; + + let (s, _) = fx.post("/approvals/approve", unknown_body("l1")).await; + assert_eq!(s, StatusCode::CONFLICT, "the capsule was not told"); + let rows = fx.records_for("l1"); + assert_eq!(rows.len(), 1, "one receipt for the submitted tx: {rows:?}"); + assert_eq!(rows[0].0, Decision::Approved); + let (outcome, detail) = rows[0].1.clone().expect("closed"); + assert_eq!(outcome, Outcome::OutcomeUnknown); + assert!(detail.contains(HASH), "{detail}"); + assert!(detail.contains("not told"), "{detail}"); + assert!(fx.st.queue.prior_unknown_outcome("l1").is_some()); +} + +/// A late callback for a call that is no longer pending, while another call is at the head, is +/// recorded for its own id and resolves nothing else. +#[tokio::test(flavor = "multi_thread")] +async fn a_late_unknown_outcome_never_touches_the_current_head() { + let fx = Fx::new(); + let q = fx.st.queue.clone(); + let other = tokio::spawn(async move { q.submit_with_outcome(payment("h1")).await }); + fx.wait_depth(1).await; + + let (s, _) = fx.post("/approvals/approve", unknown_body("gone")).await; + assert_eq!(s, StatusCode::CONFLICT); + assert_eq!(fx.st.queue.depth(), 1, "the head is untouched"); + assert!(!other.is_finished()); + assert!( + fx.records_for("h1").is_empty(), + "nothing recorded for the head" + ); + let rows = fx.records_for("gone"); + assert_eq!(rows.len(), 1, "{rows:?}"); + assert_eq!( + rows[0].1.as_ref().map(|o| o.0), + Some(Outcome::OutcomeUnknown) + ); + + let (s, _) = fx + .post("/approvals/reject", serde_json::json!({"id": "h1"})) + .await; + assert_eq!(s, StatusCode::OK); + assert_eq!(other.await.unwrap(), ApprovalOutcomePublic::Rejected); +} + +/// Tripwire: for every state the queue and the log can be in when core's well-formed +/// unknown-outcome callback arrives, the sidecar never refuses it with 503, always remembers the +/// tx, never resolves a different call, and writes a receipt naming the tx whenever the log can +/// be written. A new early return on that path fails one of these rows. +#[tokio::test(flavor = "multi_thread")] +async fn every_unknown_outcome_callback_keeps_the_transaction() { + #[derive(Debug, Clone, Copy)] + enum Head { + Waiting, + Gone, + Other, + } + #[derive(Debug, Clone, Copy)] + enum Log { + Writable, + Failing, + } + for head in [Head::Waiting, Head::Gone, Head::Other] { + for log in [Log::Writable, Log::Failing] { + let why = format!("{head:?}/{log:?}"); + let fx = match log { + Log::Writable => Fx::new(), + Log::Failing => Fx::with_failing_log(), + }; + let q = fx.st.queue.clone(); + let id = match head { + Head::Other => "x-other", + _ => "x-pay", + }; + let waiter = tokio::spawn(async move { q.submit_with_resolution(payment(id)).await }); + fx.wait_depth(1).await; + if let Head::Gone = head { + waiter.abort(); + fx.wait_depth(0).await; + } + let (s, v) = fx.post("/approvals/approve", unknown_body("x-pay")).await; + fx.set_records_writable(true); + assert_ne!(s, StatusCode::SERVICE_UNAVAILABLE, "{why}: {v}"); + assert!( + fx.st.queue.prior_unknown_outcome("x-pay").is_some(), + "{why}: the tx is remembered" + ); + match head { + Head::Waiting => { + assert_eq!(s, StatusCode::OK, "{why}"); + assert!(matches!( + waiter.await.unwrap(), + citrate_agent_core::hitl::Resolution::SubmittedOutcomeUnknown(_) + )); + } + Head::Gone => assert_eq!(s, StatusCode::CONFLICT, "{why}"), + Head::Other => { + assert_eq!(s, StatusCode::CONFLICT, "{why}"); + assert_eq!(fx.st.queue.depth(), 1, "{why}: the other call still waits"); + fx.st.queue.reject_by_id("x-other").unwrap(); + let _ = waiter.await; + } + } + let rows = fx.records_for("x-pay"); + match log { + Log::Writable => { + assert_eq!(rows.len(), 1, "{why}: {rows:?}"); + let (o, d) = rows[0].1.clone().expect("closed"); + assert_eq!(o, Outcome::OutcomeUnknown, "{why}"); + assert!(d.contains(HASH), "{why}: {d}"); + } + Log::Failing => assert!(rows.is_empty(), "{why}: nothing could be written"), + } + } + } +} From 5782a7680abd08c128f7427e076c3ade370eca34 Mon Sep 17 00:00:00 2001 From: Larry Klosowski Date: Fri, 23 Oct 2026 01:26:55 -0700 Subject: [PATCH 3/3] fix(sidecar): keep a submitted tx's unknown outcome when the record fails or the waiter is gone (C1-V3-R) An unknown outcome means citrate-core already broadcast the transaction, so the approve route no longer refuses it with 503 when the decision record cannot be written: the waiter is told, the tx is remembered for re-proposals, and the reply carries "recorded": false. Plain approvals keep the write-ahead refusal. A callback whose call is no longer pending (waiter timed out or cancelled, or another call at the head) now writes an approved ceremony decision bound to its call id and closes it OutcomeUnknown with the tx hash; it still answers 409 and resolves nothing else. record_ceremony_decision takes the call id and subject so a receipt can be written without a pending view; hic_records::try_record_outcome reports whether the outcome was written. Co-Authored-By: Claude Opus 5.5 (1M context) --- agent-sidecar/src/hic_records.rs | 8 +++ agent-sidecar/src/lib.rs | 115 ++++++++++++++++++++----------- 2 files changed, 84 insertions(+), 39 deletions(-) diff --git a/agent-sidecar/src/hic_records.rs b/agent-sidecar/src/hic_records.rs index 496e288..9480e4f 100644 --- a/agent-sidecar/src/hic_records.rs +++ b/agent-sidecar/src/hic_records.rs @@ -103,6 +103,12 @@ pub fn record_auto_decision( /// Close an allowing decision with its outcome. A failure is logged (the decision stays open and /// the log closes it as `outcome_unknown` when it is next opened). pub fn record_outcome(log: &DecisionLog, seq: u64, outcome: Outcome, detail: &str) { + let _written = try_record_outcome(log, seq, outcome, detail); +} + +/// [`record_outcome`], saying whether the outcome was written (C1-V3-R: the approve route reports +/// a missing receipt for a submitted transaction instead of hiding it). +pub fn try_record_outcome(log: &DecisionLog, seq: u64, outcome: Outcome, detail: &str) -> bool { if let Err(e) = log.record_outcome( Actor::member("member"), OutcomeEvent { @@ -114,7 +120,9 @@ pub fn record_outcome(log: &DecisionLog, seq: u64, outcome: Outcome, detail: &st eprintln!( "citrate-agent-sidecar: the outcome of decision {seq} could not be recorded: {e}" ); + return false; } + true } /// Control characters removed, then cut at a character boundary to at most `max` bytes (the diff --git a/agent-sidecar/src/lib.rs b/agent-sidecar/src/lib.rs index 6adc98a..d10941a 100644 --- a/agent-sidecar/src/lib.rs +++ b/agent-sidecar/src/lib.rs @@ -705,13 +705,25 @@ fn resolve_body( let unknown = requested_outcome(&v, approve)?; // HUP-S2.6: the member's decision on the reviewed head is recorded (write-ahead for an // approval; fail closed when the record cannot be written). A resolve that does not match - // the head records nothing. + // the head records nothing, except an unknown outcome (below). + // + // C1-V3-R: an unknown outcome is different. Core already broadcast the transaction, so + // refusing here would only hide it: the outcome is delivered and remembered whatever the + // record or the queue says, and a receipt naming the tx is written for its call id even + // when the waiter is gone. Nothing is released by it, so write-ahead does not apply. let head = st.queue.peek().filter(|p| p.id == id); let log = st.sessions.records(); let pending_record = match (&log, &head) { - (Some(log), Some(p)) if approve => Some( - record_ceremony_decision(log, p, true).map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?, - ), + (Some(log), Some(p)) if approve => { + match record_ceremony_decision(log, &p.id, &ceremony_subject(p), true) { + Ok(seq) => Some(seq), + Err(e) if unknown.is_some() => { + eprintln!("citrate-agent-sidecar: {e}"); + None + } + Err(_) => return Err(StatusCode::SERVICE_UNAVAILABLE), + } + } _ => None, }; let res = match (&unknown, approve) { @@ -719,25 +731,55 @@ fn resolve_body( (None, true) => st.queue.approve_by_id(&id), (None, false) => st.queue.reject_by_id(&id), }; + let told = res.is_ok(); + // C1-V3-R: the receipt of a submitted tx, whether or not the capsule was told. + let unknown_recorded = match (&log, &unknown) { + (Some(log), Some(u)) => { + let detail = if told { + u.record_detail() + } else { + format!( + "{}; the call was no longer waiting, so the capsule was not told", + u.record_detail() + ) + }; + let seq = pending_record.or_else(|| { + let subject = head + .as_ref() + .map(ceremony_subject) + .unwrap_or_else(|| format!("approval {id}: no longer pending")); + match record_ceremony_decision(log, &id, &subject, true) { + Ok(seq) => Some(seq), + Err(e) => { + eprintln!("citrate-agent-sidecar: {e}"); + None + } + } + }); + match seq { + Some(seq) => hic_records::try_record_outcome( + log, + seq, + citrate_agent_records::Outcome::OutcomeUnknown, + &detail, + ), + None => false, + } + } + _ => false, + }; match res { Ok(()) => { - if let (Some(log), Some(p)) = (&log, &head) { - match (pending_record, &unknown) { - // C1-V3-R: submitted, outcome unknown. Never "Completed". - (Some(seq), Some(u)) => hic_records::record_outcome( - log, - seq, - citrate_agent_records::Outcome::OutcomeUnknown, - &u.record_detail(), - ), - (Some(seq), None) => hic_records::record_outcome( + if let (Some(log), Some(p), None) = (&log, &head, &unknown) { + match pending_record { + Some(seq) => hic_records::record_outcome( log, seq, citrate_agent_records::Outcome::Completed, "released to the capsule; a chain effect was signed and broadcast in core's ceremony first", ), - (None, _) => { - if let Err(e) = record_ceremony_decision(log, p, false) { + None => { + if let Err(e) = record_ceremony_decision(log, &p.id, &ceremony_subject(p), false) { eprintln!("citrate-agent-sidecar: {e}"); } } @@ -747,46 +789,41 @@ fn resolve_body( if unknown.is_some() { out["outcome"] = serde_json::json!(citrate_agent_core::hitl::OUTCOME_SUBMITTED_UNKNOWN); + out["recorded"] = serde_json::json!(unknown_recorded); } Ok(Json(out)) } // The head is not the call the operator reviewed, or its submitter is gone — refuse. Err(_) => { - if let (Some(log), Some(seq)) = (&log, pending_record) { - match &unknown { - // The transaction went out from core even though the capsule was not told. - Some(u) => hic_records::record_outcome( - log, - seq, - citrate_agent_records::Outcome::OutcomeUnknown, - &format!( - "{}; the call was no longer waiting, so the capsule was not told", - u.record_detail() - ), - ), - None => hic_records::record_outcome( - log, - seq, - citrate_agent_records::Outcome::Failed, - "the call was no longer waiting, so nothing was released", - ), - } + if let (Some(log), Some(seq), None) = (&log, pending_record, &unknown) { + hic_records::record_outcome( + log, + seq, + citrate_agent_records::Outcome::Failed, + "the call was no longer waiting, so nothing was released", + ); } Err(StatusCode::CONFLICT) } } } -/// HUP-S2.6: one ceremony-bridge decision on the reviewed call `p`. +/// HUP-S2.6: what a ceremony decision on the pending call `p` is about. +fn ceremony_subject(p: &citrate_agent_core::hitl::PendingView) -> String { + format!("{}: {}", p.name, p.description) +} + +/// HUP-S2.6: one ceremony-bridge decision on the call `call_id`, bound to it by its evidence. fn record_ceremony_decision( log: &citrate_agent_records::DecisionLog, - p: &citrate_agent_core::hitl::PendingView, + call_id: &str, + subject: &str, allow: bool, ) -> Result { hic_records::record_member_decision( log, "ceremony.capsule_effect", - &format!("{}: {}", p.name, p.description), + subject, allow, if allow { "the member approved this effect in core's ceremony" @@ -795,7 +832,7 @@ fn record_ceremony_decision( }, vec![citrate_agent_records::EvidenceRef { kind: "approval_call".to_string(), - uri: format!("sidecar:approval/{}", p.id), + uri: format!("sidecar:approval/{call_id}"), digest: None, }], )