use serde_json::Value;
use vta_sdk::protocol::matching::{Protocol, ServiceCapabilities};
use vti_common::error::{AppError, bad_gateway_error};
use crate::operations::room_issuance::{SigningContext, VtaKeySigner};
pub const OUTBOUND_SUPPORTED: [Protocol; 1] = [Protocol::Rest];
fn pick_transport(caps: &ServiceCapabilities, host: &str) -> Result<(Protocol, String), AppError> {
for protocol in Protocol::PREFERENCE_ORDER {
if !OUTBOUND_SUPPORTED.contains(&protocol) {
continue;
}
if let Some(endpoint) = caps.endpoint(protocol) {
return Ok((protocol, endpoint.to_string()));
}
}
let advertised: Vec<&str> = Protocol::PREFERENCE_ORDER
.iter()
.filter(|p| caps.endpoint(**p).is_some())
.map(|p| p.as_str())
.collect();
let ours: Vec<&str> = OUTBOUND_SUPPORTED.iter().map(|p| p.as_str()).collect();
Err(AppError::Validation(format!(
"no transport in common with room host `{host}`: it advertises [{}] and this agent can \
initiate [{}]. This is not a host that cannot be reached — it is one this agent cannot \
yet start a conversation with, which is a gap in the agent rather than in the host.",
if advertised.is_empty() {
"nothing".to_string()
} else {
advertised.join(", ")
},
ours.join(", "),
)))
}
pub fn build_room_task(task: &str, host: &str, issuer: &str, payload: Value) -> Value {
serde_json::json!({
"id": format!("urn:uuid:{}", uuid::Uuid::new_v4()),
"type": task,
"recipient": host,
"issuer": issuer,
"issuedAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
"payload": payload,
})
}
async fn verify_host_reply(
reply: &Value,
host: &str,
resolver: &affinidi_did_resolver_cache_sdk::DIDCacheClient,
) -> Result<(), AppError> {
let doc_type = reply
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
if doc_type.starts_with("https://trusttasks.org/spec/trust-task-error/") {
return Ok(());
}
let doc: trust_tasks_rs::TrustTask<Value> =
serde_json::from_value(reply.clone()).map_err(|e| {
bad_gateway_error(format!(
"room host `{host}` sent a reply this agent cannot read as a Trust-Task \
document: {e}"
))
})?;
let vm_resolver = vti_common::auth::TrustTaskVmResolver::from_optional(Some(resolver.clone()));
let signer = vti_common::auth::verify_trust_task_proof_with(&doc, &vm_resolver)
.await
.map_err(|e| {
AppError::Forbidden(format!(
"the reply from room host `{host}` is unsigned or its proof does not verify \
({e}), so nothing in it can be believed — an unsigned answer is bytes, not \
evidence"
))
})?;
if signer != host {
return Err(AppError::Forbidden(format!(
"the reply claiming to come from room host `{host}` is signed by `{signer}`. The \
proof verifies, which means somebody really signed it — just not the party this \
agent asked"
)));
}
Ok(())
}
pub fn read_reply(doc: &Value, expected: &str, host: &str) -> Result<Value, AppError> {
let doc_type = doc.get("type").and_then(Value::as_str).unwrap_or_default();
if doc_type == expected {
return Ok(doc.get("payload").cloned().unwrap_or(Value::Null));
}
if doc_type.starts_with("https://trusttasks.org/spec/trust-task-error/") {
let payload = doc.get("payload").cloned().unwrap_or(Value::Null);
let code = payload
.get("code")
.and_then(Value::as_str)
.unwrap_or("unspecified");
let reason = payload
.get("reason")
.or_else(|| payload.get("message"))
.and_then(Value::as_str)
.unwrap_or("no reason given");
return Err(AppError::Forbidden(format!(
"room host `{host}` refused: {code}: {reason}"
)));
}
Err(AppError::Internal(format!(
"room host `{host}` answered `{doc_type}` where `{expected}` was expected; a reply that \
threads to our request but answers a different task is a contract break rather than a \
task failure"
)))
}
const MUTATING_ROOM_TASKS: &[&str] = &[
vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE,
vti_rooms::wire::ROOMS_RECORDS_CURATE_TYPE,
vti_rooms::wire::ROOMS_EPOCH_MINT_TYPE,
vti_rooms::wire::ROOMS_OWNER_TRANSFER_TYPE,
];
pub async fn refuse_if_caught(
groups: &vti_common::store::KeyspaceHandle,
room_id: &str,
task: &str,
) -> Result<(), AppError> {
if !MUTATING_ROOM_TASKS.contains(&task) {
return Ok(());
}
let history = match crate::operations::room_groups::root_history(groups, room_id).await {
Ok(h) => h,
Err(e) => {
tracing::error!(
room = %room_id,
error = %e,
"could not read this agent's root history; allowing the write rather than \
turning a local fault into an accusation"
);
return Ok(());
}
};
if let Some(caught) = history.caught_at {
return Err(AppError::Forbidden(format!(
"this agent has been given two different record sets for `{room_id}`, both \
claimed at version {caught}. One of them is wrong, and a write cannot explain \
the difference — a write moves the version. Reading this room still works, and \
is how the evidence is gathered; writing to a host that has contradicted itself \
is refused."
)));
}
Ok(())
}
pub async fn send_room_task(
ctx: SigningContext<'_>,
groups: &vti_common::store::KeyspaceHandle,
room_id: &str,
resolver: &affinidi_did_resolver_cache_sdk::DIDCacheClient,
host: &str,
signing_key_id: &str,
issuer: &str,
verification_method: &str,
task: &str,
response_task: &str,
payload: Value,
) -> Result<Value, AppError> {
refuse_if_caught(groups, room_id, task).await?;
let resolved = resolver.resolve(host).await.map_err(|e| {
AppError::Validation(format!(
"the room host `{host}` does not resolve, so there is nothing to send to: {e}"
))
})?;
let doc_value = serde_json::to_value(&resolved.doc)
.map_err(|e| AppError::Internal(format!("serialise the host's DID document: {e}")))?;
let caps = ServiceCapabilities::from_did_document(&doc_value);
let (protocol, endpoint) = pick_transport(&caps, host)?;
let mut document = build_room_task(task, host, issuer, payload);
let signer = VtaKeySigner::new(ctx, signing_key_id, verification_method);
let proof = affinidi_data_integrity::DataIntegrityProof::sign(
&document,
&signer,
affinidi_data_integrity::SignOptions::new(),
)
.await
.map_err(|e| {
let mut cause = String::new();
let mut src: Option<&(dyn std::error::Error + 'static)> = std::error::Error::source(&e);
while let Some(inner) = src {
cause = format!("{cause}: {inner}");
src = inner.source();
}
AppError::Internal(format!(
"sign the request to room host `{host}`: {e}{cause}"
))
})?;
document["proof"] = serde_json::to_value(proof)
.map_err(|e| AppError::Internal(format!("serialise the proof: {e}")))?;
match protocol {
Protocol::Rest => {
let url = format!("{}/trust-tasks", endpoint.trim_end_matches('/'));
let response = vta_sdk::http::rest_client()
.post(&url)
.header("content-type", "application/json")
.json(&document)
.send()
.await
.map_err(|e| {
bad_gateway_error(format!("room host `{host}` at {url} did not answer: {e}"))
})?;
let body = response.text().await.map_err(|e| {
bad_gateway_error(format!("room host `{host}` sent an unreadable body: {e}"))
})?;
let reply: Value = serde_json::from_str(&body).map_err(|e| {
bad_gateway_error(format!(
"room host `{host}` sent a body that is not a Trust-Task document: {e}: {body}"
))
})?;
verify_host_reply(&reply, host, resolver).await?;
read_reply(&reply, response_task, host)
}
Protocol::Tsp | Protocol::Didcomm => Err(AppError::Internal(format!(
"{} is named in OUTBOUND_SUPPORTED but has no send path here",
protocol.as_str()
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn caps_from(services: Value) -> ServiceCapabilities {
ServiceCapabilities::from_did_document(&serde_json::json!({ "service": services }))
}
#[test]
fn a_host_serving_rest_is_reachable() {
let caps = caps_from(serde_json::json!([{
"id": "#rest", "type": "VTARest", "serviceEndpoint": "https://host.example"
}]));
let (protocol, endpoint) = pick_transport(&caps, "did:example:host").expect("reachable");
assert_eq!(protocol, Protocol::Rest);
assert_eq!(endpoint, "https://host.example");
}
#[test]
fn a_tsp_only_host_is_refused_naming_both_sides() {
let caps = caps_from(serde_json::json!([{
"id": "#tsp", "type": "TSPTransport", "serviceEndpoint": "did:example:mediator"
}]));
let err = pick_transport(&caps, "did:example:host").expect_err("no common transport");
let msg = err.to_string();
assert!(
msg.contains("tsp"),
"must name what the host advertises: {msg}"
);
assert!(
msg.contains("rest"),
"must name what this agent can do: {msg}"
);
assert!(
msg.contains("gap in the agent"),
"must say whose limitation it is: {msg}"
);
}
#[test]
fn a_host_advertising_nothing_says_so() {
let err = pick_transport(&caps_from(serde_json::json!([])), "did:example:host")
.expect_err("nothing advertised");
assert!(err.to_string().contains("nothing"));
}
#[test]
fn the_document_is_addressed_and_attributed() {
let doc = build_room_task(
"https://trusttasks.org/spec/rooms/epoch/chain/0.1",
"did:example:host",
"did:example:agent",
serde_json::json!({ "roomId": "did:example:room" }),
);
assert_eq!(doc["recipient"], "did:example:host");
assert_eq!(doc["issuer"], "did:example:agent");
assert_eq!(doc["payload"]["roomId"], "did:example:room");
assert!(doc["id"].as_str().unwrap().starts_with("urn:uuid:"));
}
#[test]
fn a_response_yields_its_payload() {
let reply = serde_json::json!({
"type": "https://trusttasks.org/spec/rooms/epoch/chain/0.1#response",
"payload": { "links": [] }
});
let got = read_reply(
&reply,
"https://trusttasks.org/spec/rooms/epoch/chain/0.1#response",
"did:example:host",
)
.expect("a response");
assert_eq!(got["links"], serde_json::json!([]));
}
#[test]
fn a_refusal_carries_the_hosts_own_code_and_reason() {
let reply = serde_json::json!({
"type": "https://trusttasks.org/spec/trust-task-error/0.5",
"payload": { "code": "private-tier-not-enabled", "reason": "this community has not enabled private rooms" }
});
let err = read_reply(&reply, "irrelevant", "did:example:host").expect_err("a refusal");
let msg = err.to_string();
assert!(msg.contains("private-tier-not-enabled"), "{msg}");
assert!(msg.contains("has not enabled private rooms"), "{msg}");
assert!(
matches!(err, AppError::Forbidden(_)),
"a refusal, not a 502"
);
}
#[tokio::test]
async fn a_refusal_is_read_without_a_proof() {
let reply = serde_json::json!({
"type": "https://trusttasks.org/spec/trust-task-error/0.5",
"payload": { "code": "notAMember", "reason": "no" }
});
verify_host_reply(&reply, "did:example:host", &test_resolver().await)
.await
.expect("a refusal needs no proof");
}
#[tokio::test]
async fn an_unsigned_success_reply_is_refused() {
let reply = serde_json::json!({
"id": "urn:uuid:00000000-0000-4000-8000-000000000001",
"type": "https://trusttasks.org/spec/rooms/epoch/chain/0.1#response",
"issuer": "did:example:host",
"recipient": "did:example:agent",
"issuedAt": "2026-01-01T00:00:00Z",
"payload": { "links": [] }
});
let err = verify_host_reply(&reply, "did:example:host", &test_resolver().await)
.await
.expect_err("an unsigned success reply must not be believed");
let msg = err.to_string();
assert!(
msg.contains("bytes, not") || msg.contains("unsigned"),
"the refusal must say why an unsigned answer is worthless: {msg}"
);
}
async fn test_resolver() -> affinidi_did_resolver_cache_sdk::DIDCacheClient {
affinidi_did_resolver_cache_sdk::DIDCacheClient::new(
affinidi_did_resolver_cache_sdk::config::DIDCacheConfigBuilder::default().build(),
)
.await
.expect("a resolver for tests")
}
#[test]
fn a_reply_answering_a_different_task_is_a_contract_break() {
let reply =
serde_json::json!({ "type": "https://trusttasks.org/spec/rooms/create/0.1#response" });
let err = read_reply(
&reply,
"https://trusttasks.org/spec/rooms/epoch/chain/0.1#response",
"did:example:host",
)
.expect_err("wrong task");
assert!(err.to_string().contains("contract break"));
}
}
#[cfg(test)]
mod write_gate_tests {
use super::*;
use crate::operations::room_groups::observe_head;
use vti_common::config::StoreConfig;
use vti_common::store::Store;
const ROOM: &str = "did:webvh:example.com:rooms:northwind";
const A: &str = "zQmbWqxBEKC3P8tqsKc98xmWNzrzDtRLMiMPL8wBuTGsMnR";
const B: &str = "zQmXo1sV5aJ7bT2kQdF9wRnPzYcH4uMgLtEjV6NrBqWsDpK";
async fn open() -> (tempfile::TempDir, vti_common::store::KeyspaceHandle) {
let dir = tempfile::tempdir().unwrap();
let store = Store::open(&StoreConfig {
data_dir: dir.path().to_path_buf(),
})
.unwrap();
let ks = store.keyspace(crate::keyspaces::ROOM_GROUPS).unwrap();
(dir, ks)
}
#[tokio::test]
async fn a_caught_host_is_refused_writes_and_still_serves_reads() {
let (_d, ks) = open().await;
observe_head(&ks, ROOM, 412, A).await.unwrap();
observe_head(&ks, ROOM, 412, B).await.unwrap();
let refused = refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE)
.await
.expect_err("a write to a caught host must be refused");
let said = refused.to_string();
assert!(
said.contains("412") && said.contains("Reading this room still works"),
"the refusal must name what was observed and what still works: {said}"
);
refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_GET_TYPE)
.await
.expect("reads are served");
refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_LIST_TYPE)
.await
.expect("listings are served");
}
#[tokio::test]
async fn a_host_that_has_not_been_caught_is_not_refused() {
let (_d, ks) = open().await;
observe_head(&ks, ROOM, 412, A).await.unwrap();
observe_head(&ks, ROOM, 413, B).await.unwrap();
refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE)
.await
.expect("two moments are not a contradiction");
}
#[tokio::test]
async fn being_caught_survives_the_history_being_pruned() {
let (_d, ks) = open().await;
observe_head(&ks, ROOM, 1, A).await.unwrap();
observe_head(&ks, ROOM, 1, B).await.unwrap();
for v in 2..=64u64 {
observe_head(&ks, ROOM, v, A).await.unwrap();
}
assert!(
refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE)
.await
.is_err(),
"reading past a conflict must not clear it"
);
}
#[tokio::test]
async fn a_room_with_no_history_at_all_is_written_to() {
let (_d, ks) = open().await;
refuse_if_caught(&ks, ROOM, vti_rooms::wire::ROOMS_RECORDS_PUT_TYPE)
.await
.expect("nothing observed is not something caught");
}
}