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 9884ea3..d10941a 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,26 +699,78 @@ 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. + // 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 = 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), + }; + 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) { + if let (Some(log), Some(p), None) = (&log, &head, &unknown) { match pending_record { Some(seq) => hic_records::record_outcome( log, @@ -664,17 +779,23 @@ fn resolve_body( "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) { + if let Err(e) = record_ceremony_decision(log, &p.id, &ceremony_subject(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); + 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) { + if let (Some(log), Some(seq), None) = (&log, pending_record, &unknown) { hic_records::record_outcome( log, seq, @@ -687,16 +808,22 @@ fn resolve_body( } } -/// 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" @@ -705,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, }], ) @@ -2388,6 +2515,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..aef45aa --- /dev/null +++ b/agent-sidecar/src/outcome_unknown_tests.rs @@ -0,0 +1,622 @@ +//! 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 { + Self::new_with(LogConfig::default()) + } + + fn new_with(cfg: LogConfig) -> Self { + let root = tempfile::tempdir().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( + 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); +} + +// ── 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"), + } + } + } +} 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()); + } +}