diff --git a/crates/tracedecay-sdk/src/remote_client.rs b/crates/tracedecay-sdk/src/remote_client.rs index e5729aeac..47ee746a3 100644 --- a/crates/tracedecay-sdk/src/remote_client.rs +++ b/crates/tracedecay-sdk/src/remote_client.rs @@ -92,7 +92,7 @@ impl EnrolledRemoteClient { credential: impl AsRef<[u8]>, timeout: Duration, ) -> Result { - Self::build(endpoint, credential, timeout, None) + Self::build(endpoint, credential, timeout, None, false) } /// Builds a client with one explicit additional HTTPS trust root. @@ -107,27 +107,49 @@ impl EnrolledRemoteClient { credential, timeout, Some(root_certificate_pem.as_ref()), + false, ) } + /// Targets the local daemon's application listener, which nests the same + /// Remote Brain router at `/remote` as the external TLS listener. Same + /// operations, envelopes, credential header, and response validation; + /// plaintext HTTP is admitted for loopback hosts only. + pub fn new_local_daemon( + endpoint: impl AsRef, + credential: impl AsRef<[u8]>, + timeout: Duration, + ) -> Result { + Self::build(endpoint, credential, timeout, None, true) + } + fn build( endpoint: impl AsRef, credential: impl AsRef<[u8]>, timeout: Duration, root_certificate_pem: Option<&[u8]>, + allow_loopback_http: bool, ) -> Result { let endpoint = reqwest::Url::parse(endpoint.as_ref()) .map_err(|error| RemoteClientError::Configuration(error.to_string()))?; - if endpoint.scheme() != "https" + let scheme_admitted = match endpoint.scheme() { + "https" => true, + "http" => allow_loopback_http && host_is_loopback(&endpoint), + _ => false, + }; + if !scheme_admitted || endpoint.host_str().is_none() || endpoint.query().is_some() || endpoint.fragment().is_some() || endpoint.username() != "" || endpoint.password().is_some() { - return Err(RemoteClientError::Configuration( - "Remote Brain endpoint must be a credential-free HTTPS URL".to_owned(), - )); + return Err(RemoteClientError::Configuration(if allow_loopback_http { + "local daemon Remote Brain endpoint must be a credential-free loopback HTTP or HTTPS URL" + .to_owned() + } else { + "Remote Brain endpoint must be a credential-free HTTPS URL".to_owned() + })); } let credential = credential.as_ref(); if validate_remote_secret_length(credential).is_err() { @@ -139,6 +161,13 @@ impl EnrolledRemoteClient { HeaderValue::from_bytes([b"Bearer ".as_slice(), credential].concat().as_slice()) .map_err(|error| RemoteClientError::Configuration(error.to_string()))?; let mut builder = HttpClient::builder().timeout(timeout); + if endpoint.scheme() == "http" { + // The loopback-only plaintext admission above is void if a system + // proxy (`HTTP_PROXY`/`ALL_PROXY`) re-routes the request: the + // Bearer enrollment credential would leave the machine + // unencrypted. The loopback target never needs a proxy. + builder = builder.no_proxy(); + } if let Some(pem) = root_certificate_pem { let mut certificates = reqwest::Certificate::from_pem_bundle(pem) .map_err(|error| RemoteClientError::Configuration(error.to_string()))?; @@ -356,6 +385,22 @@ impl EnrolledRemoteClient { } } +/// Whether the endpoint host is a loopback address; any other host requires +/// HTTPS. +fn host_is_loopback(endpoint: &reqwest::Url) -> bool { + let Some(host) = endpoint.host_str() else { + return false; + }; + let address = host + .strip_prefix('[') + .and_then(|host| host.strip_suffix(']')) + .unwrap_or(host); + if let Ok(address) = address.parse::() { + return address.is_loopback(); + } + host.eq_ignore_ascii_case("localhost") +} + fn credential_header(credential: &[u8]) -> Result { if validate_remote_secret_length(credential).is_err() { return Err(RemoteClientError::Configuration( @@ -547,6 +592,47 @@ mod tests { assert!(matches!(error, RemoteClientError::Configuration(_))); } + #[test] + fn local_daemon_target_admits_loopback_http_only() { + let credential = "0123456789abcdef0123456789abcdef"; + for endpoint in [ + "http://127.0.0.1:39181/remote/", + "http://[::1]:39181/remote/", + "http://localhost:39181/remote/", + "https://remote.example/remote/", + ] { + EnrolledRemoteClient::new_local_daemon(endpoint, credential, Duration::from_secs(1)) + .unwrap_or_else(|error| { + panic!("local daemon target must admit {endpoint}: {error}") + }); + } + + for endpoint in [ + "http://remote.example/remote/", + "http://10.0.0.7:39181/remote/", + "ftp://127.0.0.1/remote/", + ] { + let error = EnrolledRemoteClient::new_local_daemon( + endpoint, + credential, + Duration::from_secs(1), + ) + .expect_err("plaintext beyond loopback must fail closed"); + assert!(matches!(error, RemoteClientError::Configuration(_))); + } + } + + #[test] + fn enrolled_remote_target_still_refuses_loopback_http() { + let error = EnrolledRemoteClient::new( + "http://127.0.0.1:39181/remote/", + "0123456789abcdef0123456789abcdef", + Duration::from_secs(1), + ) + .expect_err("the enrolled remote target must stay HTTPS-only"); + assert!(matches!(error, RemoteClientError::Configuration(_))); + } + #[test] fn enrolled_remote_client_rejects_url_credentials() { let error = EnrolledRemoteClient::new( diff --git a/crates/tracedecay-sdk/tests/remote_client_proxy.rs b/crates/tracedecay-sdk/tests/remote_client_proxy.rs new file mode 100644 index 000000000..e9a2f43a7 --- /dev/null +++ b/crates/tracedecay-sdk/tests/remote_client_proxy.rs @@ -0,0 +1,105 @@ +//! The plaintext loopback Remote Brain target must never traverse a system +//! proxy: `HTTP_PROXY`/`ALL_PROXY` would carry the Bearer enrollment +//! credential off the machine unencrypted, defeating the loopback-only +//! plaintext admission. This test lives in its own integration binary because +//! it mutates the process-wide proxy environment. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::thread; +use std::time::Duration; + +use tracedecay_application::RequestId; +use tracedecay_application::remote::protocol::{EnrollmentRequestV1, RemoteProtocolRequestV1}; +use tracedecay_domain::{ + BrainId, BrainNodeId, EntityId, ProjectId, RefId, RemoteCapabilityV1, RemoteRepositoryScopeV1, + RepositoryId, RepositoryStateSnapshotId, UtcMicros, WorktreeId, +}; +use tracedecay_sdk::remote_client::{EnrolledRemoteClient, RemoteClientError}; + +#[test] +fn loopback_http_request_never_routes_through_a_system_proxy() { + let proxy = TcpListener::bind("127.0.0.1:0").expect("bind proxy listener"); + let proxy_address = proxy.local_addr().expect("proxy address"); + let target = TcpListener::bind("127.0.0.1:0").expect("bind target listener"); + let target_port = target.local_addr().expect("target address").port(); + let server = thread::spawn(move || { + let (mut stream, _) = target.accept().expect("direct loopback connection"); + let mut head = [0u8; 2048]; + let _ = stream.read(&mut head).expect("read request head"); + stream + .write_all( + b"HTTP/1.1 200 OK\r\n\ +content-type: application/json\r\n\ +content-length: 2\r\n\ +connection: close\r\n\r\n{}", + ) + .expect("write canned response"); + }); + + // SAFETY: this integration binary holds only this test, so no other + // thread reads or writes the process environment concurrently. + unsafe { + std::env::set_var("HTTP_PROXY", format!("http://{proxy_address}")); + std::env::set_var("ALL_PROXY", format!("http://{proxy_address}")); + } + let client = EnrolledRemoteClient::new_local_daemon( + format!("http://127.0.0.1:{target_port}/remote/"), + "0123456789abcdef0123456789abcdef", + Duration::from_secs(5), + ); + // SAFETY: as above. + unsafe { + std::env::remove_var("HTTP_PROXY"); + std::env::remove_var("ALL_PROXY"); + } + let client = client.expect("loopback client must build under a proxy environment"); + + let outcome = client.enroll(&enrollment_request(), *b"fedcba9876543210fedcba9876543210"); + // A protocol error proves the request reached the direct loopback target + // and got the canned garbage back; a proxied request would have died in + // transport against the never-accepting proxy listener instead. + match outcome { + Err(RemoteClientError::Protocol(_)) => {} + Err(error) => panic!("expected a protocol error from the direct target, got {error:?}"), + Ok(_) => panic!("the canned non-canonical response must fail as protocol"), + } + server.join().expect("target server thread"); + + proxy + .set_nonblocking(true) + .expect("nonblocking proxy accept"); + match proxy.accept() { + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + Ok(_) => panic!("the loopback request must never reach the system proxy"), + Err(error) => panic!("proxy accept failed: {error}"), + } +} + +fn enrollment_request() -> RemoteProtocolRequestV1 { + let brain_id = BrainId::new("brain.remote-proxy-test").unwrap(); + let node_id = BrainNodeId::new("node.remote-proxy-test").unwrap(); + RemoteProtocolRequestV1::new_initial_enrollment( + RequestId::new("request.remote-proxy-test").unwrap(), + brain_id.clone(), + node_id.clone(), + UtcMicros(1_000_000), + EnrollmentRequestV1 { + grant_id: EntityId::new("grant.remote-proxy-test").unwrap(), + grant_revision: 1, + enrollment_id: EntityId::new("enrollment.remote-proxy-test").unwrap(), + brain_id, + node_id, + expires_at: UtcMicros(600_000_000), + capabilities: [RemoteCapabilityV1::Query].into_iter().collect(), + scope: RemoteRepositoryScopeV1 { + project_id: ProjectId::new("project.remote-proxy-test").unwrap(), + repository_id: RepositoryId::new("repository.remote-proxy-test").unwrap(), + worktree_id: WorktreeId::new("worktree.remote-proxy-test").unwrap(), + reference: Some(RefId::new("refs/heads/remote-proxy-test").unwrap()), + snapshot_id: RepositoryStateSnapshotId::new("snapshot.remote-proxy-test").unwrap(), + }, + }, + ) + .unwrap() +} diff --git a/docs/plans/tracedecay-v2/NEXT.md b/docs/plans/tracedecay-v2/NEXT.md index 20912a94a..e54af8938 100644 --- a/docs/plans/tracedecay-v2/NEXT.md +++ b/docs/plans/tracedecay-v2/NEXT.md @@ -472,21 +472,21 @@ the commit as attribution evidence. CLI, and both SDKs. Prove the partial-effect committed receipt and reset-only legal action survive each boundary and physical daemon restart; unit mappers and synthetic SDK envelopes are necessary but not final journey evidence. - PARTIAL, updated 2026-08-15: both CLI legs are proven green in a clean - worktree at `d7c4a4c43` — PartialEffect (`e56fadeff` lineage) AND the - ResetRequired leg, which was un-`#[ignore]`d by `c962cd627` after the - project-open settling gap was fixed; the reset-only legal action survives - a physical restart via CLI. The HTTP/MCP/SDK legs are in flight in - `tests/typed_terminal_restart_acceptance/transport_boundaries.rs` - (checkpoint `29a591519`; module made resolvable by `2d63021e6`). First - full run of those WIP legs, 2026-08-15: both fail with actionable - reasons — `partial_effect_survives_http_mcp_and_rust_sdk_across_restart` - settles without ever reaching the durable commit boundary, and - `reset_required_survives_http_mcp_and_rust_sdk_across_restart` gets a raw - JSON-RPC `-32603` whose `data` carries - `kind:"reset_required"`/reason/retryable instead of the typed problem - envelope the assertion requires (the MCP surface is not wrapping the - reset terminal as a problem envelope). + DONE 2026-08-19: all four legs are green. The CLI legs were proven at + `d7c4a4c43` (PartialEffect via the `e56fadeff` lineage; ResetRequired + un-`#[ignore]`d by `c962cd627`). The HTTP/MCP/Rust-SDK legs in + `tests/typed_terminal_restart_acceptance/transport_boundaries.rs` pass + 2/2 after two production fixes: a `tools/call` refused at project open + with `ResetRequired` now answers on the MCP tool surface with the + canonical problem envelope under the operation's own MCP result contract + (`mcp_project_open_reset_refusal`; previously a raw JSON-RPC `-32603`), + and the socket `DaemonInvocationClient` keeps reading an authoritative + effect over `DAEMON_TOOL_RESPONSE_GRACE` after cancel delivery instead of + fabricating `ResetRequired` at the two-second shutdown bound — a + post-cancel transport failure stays the typed indeterminate settlement, + mirroring `settle_in_process_invocation`. Both journeys prove the + partial-effect committed receipt and reset-only legal action across a + physical daemon restart. - DONE 2026-08-10 (verified 2026-08-13): the ten `schema_unavailable` application bindings were repaired in `d2b094ca7` — the primitive-surface read operations gained typed schemas in diff --git a/src/application_surface.rs b/src/application_surface.rs index 6a2daaae4..e4f3b2480 100644 --- a/src/application_surface.rs +++ b/src/application_surface.rs @@ -4162,6 +4162,33 @@ fn http_adapter_problem( .map(|problem| problem.with_owning_layer(ProblemOwningLayer::Adapter)) } +/// The canonical typed terminal for an MCP `tools/call` whose project open +/// was refused because the store requires an explicit reset. +/// +/// The refusal settles before any project server exists, so the MCP boundary +/// cannot route the call to its handler; the truthful answer for the named +/// operation is the reset-required terminal under its own mounted MCP result +/// contract. Returns `None` for tools without a mounted application binding. +pub(crate) fn mcp_project_open_reset_refusal( + tool_name: &str, + request_id: RequestId, + authority: &str, + reason: &str, +) -> Option { + let operation = ApplicationSurfaceOperation::from_tool_name(tool_name)?; + let catalog = application_surface_catalog_ref().ok()?; + let resolver = CatalogBindingResolver::new(catalog); + let binding = resolve_application_binding(&resolver, BindingSurface::Mcp, operation)?; + let contract = ResultContractRef::from_schema(&binding.result_schema); + let problem = ApplicationProblem::reset_required(SafeDiagnostic { + code: "application.surface.reset_required".to_owned(), + message: format!("The {authority} requires an explicit reset: {reason}"), + }); + ApplicationProblemEnvelope::new(contract, request_id, problem) + .ok() + .map(|envelope| envelope.with_owning_layer(ProblemOwningLayer::Runtime)) +} + fn current_micros() -> Result { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) diff --git a/src/cli.rs b/src/cli.rs index ac15a3513..677606362 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -810,6 +810,22 @@ pub enum RemoteAction { #[arg(long, value_name = "FILE")] enrollment_credential_file: PathBuf, }, + /// Capture one observation into a node's offline spool while the + /// authority is unreachable + Capture { + #[command(flatten)] + authority: RemoteAuthorityArgs, + }, + /// Query one exact observation with honest local/remote coverage + Query { + #[command(flatten)] + authority: RemoteAuthorityArgs, + }, + /// Transfer one encrypted offline-capture frame to an enrolled node's spool + TransferFrame { + #[command(flatten)] + authority: RemoteAuthorityArgs, + }, /// Replay pending offline-capture frames through the current authority Replay { #[command(flatten)] @@ -843,6 +859,15 @@ impl From for tracedecay::remote_command::RemoteCommand { args: authority.into(), enrollment_credential_file, }, + RemoteAction::Capture { authority } => Self::Capture { + args: authority.into(), + }, + RemoteAction::Query { authority } => Self::Query { + args: authority.into(), + }, + RemoteAction::TransferFrame { authority } => Self::TransferFrame { + args: authority.into(), + }, RemoteAction::Replay { authority } => Self::Replay { args: authority.into(), }, diff --git a/src/cli/help.rs b/src/cli/help.rs index 73eea4e47..f49b444dd 100644 --- a/src/cli/help.rs +++ b/src/cli/help.rs @@ -637,12 +637,16 @@ status (one project's statistics)."; pub(crate) const REMOTE_LONG_ABOUT: &str = "\ Operates the Remote Brain production journey from the shell: live mounted \ -status from the running daemon, plus enrolled enroll/replay/backup/restore/\ -failover against an authenticated authority endpoint. `status` never probes \ -local stores; it reads the daemon's in-memory remote mount. Protocol actions \ -take a typed `RemoteProtocolRequestV1` JSON body from `--request-file` \ -(`-` reads stdin) and send credentials only from files. `--json` emits one \ -canonical JSON line."; +status from the running daemon, plus enrolled enroll/capture/query/\ +transfer-frame/replay/backup/restore/failover against an authenticated \ +authority endpoint. `status` never probes local stores; it reads the \ +daemon's in-memory remote mount. `capture` spools one observation on a node \ +whose authority is unreachable, `transfer-frame` moves one encrypted spool \ +frame to an enrolled node, and `query` reads one exact observation and \ +renders the honest remote shard coverage next to the caller's own pending \ +local spool. Protocol actions take a typed `RemoteProtocolRequestV1` JSON \ +body from `--request-file` (`-` reads stdin) and send credentials only from \ +files. `--json` emits one canonical JSON line."; pub(crate) const REMOTE_AFTER_HELP: &str = "\ Examples: @@ -651,6 +655,12 @@ Examples: tracedecay remote enroll --endpoint https://brain.example/remote/ \\ --credential-file grant.bin --enrollment-credential-file enroll.bin \\ --request-file enroll.json --json + tracedecay remote capture --endpoint https://node.example/remote/ \\ + --credential-file cred.bin --request-file capture.json --json + tracedecay remote query --endpoint https://brain.example/remote/ \\ + --credential-file cred.bin --request-file query.json + tracedecay remote transfer-frame --endpoint https://peer.example/remote/ \\ + --credential-file cred.bin --request-file frame.json --json tracedecay remote replay --endpoint https://brain.example/remote/ \\ --credential-file cred.bin --request-file replay.json tracedecay remote backup --endpoint https://brain.example/remote/ \\ diff --git a/src/cli/parse_tests.rs b/src/cli/parse_tests.rs index ea39b2f87..d7430e4da 100644 --- a/src/cli/parse_tests.rs +++ b/src/cli/parse_tests.rs @@ -2128,7 +2128,16 @@ fn remote_status_parses_json_flag() { #[test] fn remote_protocol_actions_require_endpoint_credential_and_request_file() { - for action in ["enroll", "replay", "backup", "restore", "failover"] { + for action in [ + "enroll", + "capture", + "query", + "transfer-frame", + "replay", + "backup", + "restore", + "failover", + ] { let error = match Cli::try_parse_from(["tracedecay", "remote", action]) { Ok(_) => panic!("{action} must require authority flags"), Err(error) => error, @@ -2197,6 +2206,43 @@ fn remote_replay_parses_request_file_and_optional_trust_root() { assert!(authority.json); } +#[test] +fn remote_capture_query_and_transfer_frame_parse_authority_flags() { + for (action, expected) in [ + ("capture", "capture"), + ("query", "query"), + ("transfer-frame", "transfer_frame"), + ] { + let cli = Cli::try_parse_from([ + "tracedecay", + "remote", + action, + "--endpoint", + "https://node.example/remote/", + "--credential-file", + "cred.bin", + "--request-file", + "request.json", + "--json", + ]) + .unwrap_or_else(|error| panic!("remote {action} should parse: {error}")); + + let Some(Commands::Remote { action: parsed }) = cli.command else { + panic!("unexpected remote {action} command"); + }; + let authority = match (&parsed, expected) { + (RemoteAction::Capture { authority }, "capture") + | (RemoteAction::Query { authority }, "query") + | (RemoteAction::TransferFrame { authority }, "transfer_frame") => authority, + _ => panic!("remote {action} parsed into the wrong action"), + }; + assert_eq!(authority.endpoint, "https://node.example/remote/"); + assert_eq!(authority.credential_file, std::path::Path::new("cred.bin")); + assert_eq!(authority.request_file, std::path::Path::new("request.json")); + assert!(authority.json); + } +} + #[test] fn remote_status_rejects_protocol_request_file_flags() { let error = match Cli::try_parse_from([ diff --git a/src/daemon/connection_serving.rs b/src/daemon/connection_serving.rs index bcb800d53..562194563 100644 --- a/src/daemon/connection_serving.rs +++ b/src/daemon/connection_serving.rs @@ -1010,7 +1010,13 @@ async fn serve_broker_socket_client( } Err(error) => { drop(setup_activity); - write_project_open_error(&mut transport, &first_request_line, &error).await?; + write_project_open_error( + &mut transport, + &first_request_line, + &handshake.client_instance_id, + &error, + ) + .await?; return Ok(()); } } @@ -1617,7 +1623,13 @@ pub(super) async fn serve_windows_broker_client_with_class_and_invocation( } Err(error) => { drop(setup_activity); - write_project_open_error(&mut transport, &first_request_line, &error).await?; + write_project_open_error( + &mut transport, + &first_request_line, + &handshake.client_instance_id, + &error, + ) + .await?; return Ok(()); } }; diff --git a/src/daemon/project_open_handshake.rs b/src/daemon/project_open_handshake.rs index e70cc9f5b..467c11a01 100644 --- a/src/daemon/project_open_handshake.rs +++ b/src/daemon/project_open_handshake.rs @@ -161,16 +161,55 @@ fn is_readonly_database_error(err: &TraceDecayError) -> bool { pub(super) async fn write_project_open_error( transport: &mut impl McpTransport, request_line: &str, + connection_scope: &str, error: &TraceDecayError, ) -> Result<()> { - let id = serde_json::from_str::(request_line) - .ok() - .and_then(|request| request.id) - .unwrap_or(serde_json::Value::Null); - let response = project_open_error_response(id, error); + let request = serde_json::from_str::(request_line).ok(); + let response = request + .as_ref() + .and_then(|request| tool_call_open_refusal_response(request, connection_scope, error)) + .unwrap_or_else(|| { + let id = request + .and_then(|request| request.id) + .unwrap_or(serde_json::Value::Null); + project_open_error_response(id, error) + }); write_json_rpc_response(transport, &response).await } +/// A `tools/call` refused at project open answers on the MCP tool surface +/// when the refusal is the reset-required terminal, matching the canonical +/// problem envelope CLI and HTTP callers receive for the same operation. +/// Non-application tools and every other open failure keep the raw shape. +fn tool_call_open_refusal_response( + request: &JsonRpcRequest, + connection_scope: &str, + error: &TraceDecayError, +) -> Option { + if !matches!(classify_mcp_method(&request.method), McpMethod::ToolsCall) { + return None; + } + let TraceDecayError::ResetRequired { authority, reason } = error else { + return None; + }; + let id = request.id.clone()?; + let tool_name = request.params.as_ref()?.get("name")?.as_str()?; + let request_id = crate::request_identity::mcp_connection_request_id(&id, connection_scope)?; + let envelope = crate::application_surface::mcp_project_open_reset_refusal( + tool_name, request_id, authority, reason, + )?; + let text = serde_json::to_string(&envelope).ok()?; + let problem = serde_json::to_value(envelope.problem.as_ref()).ok()?; + Some(JsonRpcResponse::success( + id, + json!({ + "content": [{ "type": "text", "text": text }], + "isError": true, + "problem": problem, + }), + )) +} + pub(super) fn project_open_error_response( id: serde_json::Value, error: &TraceDecayError, @@ -237,6 +276,89 @@ pub(super) fn project_open_error_response( mod tests { use super::*; + #[test] + fn reset_required_tools_call_answers_with_the_canonical_problem_envelope() { + let request: JsonRpcRequest = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { "name": "tracedecay_storage_status", "arguments": {} }, + })) + .expect("canonical tools/call request"); + let error = TraceDecayError::reset_required("project store", "schema v26 is incompatible"); + + let response = tool_call_open_refusal_response(&request, "connection.test", &error) + .expect("an application tools/call refusal must answer on the tool surface"); + + assert!(response.error.is_none(), "the refusal is a tool result"); + let result = response.result.expect("tool result payload"); + assert_eq!(result["isError"], serde_json::json!(true)); + assert_eq!(result["problem"]["kind"], "reset_required"); + assert_eq!( + result["problem"]["legal_actions"], + serde_json::json!(["reset"]) + ); + let text = result["content"][0]["text"] + .as_str() + .expect("rendered envelope text"); + let envelope: serde_json::Value = + serde_json::from_str(text).expect("machine-readable envelope"); + assert_eq!(envelope["problem"]["kind"], "reset_required"); + assert_eq!( + envelope["problem"]["legal_actions"], + serde_json::json!(["reset"]) + ); + assert!( + envelope["problem"]["diagnostic"]["message"] + .as_str() + .is_some_and(|message| message.contains("schema v26 is incompatible")), + "the refusal must carry the store's own reason: {envelope}" + ); + } + + #[test] + fn non_application_tools_and_other_failures_keep_the_raw_refusal_shape() { + let reset = TraceDecayError::reset_required("project store", "incompatible shape"); + let unknown_tool: JsonRpcRequest = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { "name": "tracedecay_not_an_application_tool", "arguments": {} }, + })) + .expect("tools/call request"); + assert!( + tool_call_open_refusal_response(&unknown_tool, "connection.test", &reset).is_none(), + "a tool without a mounted application binding keeps the raw refusal" + ); + + let initialize: JsonRpcRequest = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {}, + })) + .expect("initialize request"); + assert!( + tool_call_open_refusal_response(&initialize, "connection.test", &reset).is_none(), + "protocol bootstrap requests keep the raw refusal" + ); + + let storage_status: JsonRpcRequest = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { "name": "tracedecay_storage_status", "arguments": {} }, + })) + .expect("tools/call request"); + let config = TraceDecayError::Config { + message: "unrelated open failure".to_owned(), + }; + assert!( + tool_call_open_refusal_response(&storage_status, "connection.test", &config).is_none(), + "non-terminal open failures keep the raw refusal" + ); + } + #[test] fn reset_required_project_open_is_serialized_as_a_non_retryable_typed_failure() { let response = project_open_error_response( diff --git a/src/daemon_client/controlled_invocation.rs b/src/daemon_client/controlled_invocation.rs index a3de593a7..259c334fe 100644 --- a/src/daemon_client/controlled_invocation.rs +++ b/src/daemon_client/controlled_invocation.rs @@ -55,14 +55,23 @@ impl DaemonInvocationClient { Err(DaemonInvocationError::Cancelled { stage }) } InvocationCancellationPolicy::AuthoritativeEffect => { + // An authoritative effect settles itself; keep reading + // over the same response grace the daemon's own + // clients use so its real terminal (e.g. a + // `PartialEffect` with a committed receipt) is the + // one reported, exactly as `settle_in_process_invocation` + // does. A transport failure after the cancel attempt + // is the same indeterminate state as an unanswered + // grace: the effect may have committed, so a + // retry-inviting `Unavailable` would be untruthful. match tokio::time::timeout( - crate::daemon::DAEMON_TASK_ABORT_DEADLINE, + crate::daemon::DAEMON_TOOL_RESPONSE_GRACE, &mut invocation, ) .await { - Ok(result) => result.map_err(|_| DaemonInvocationError::Unavailable), - Err(_) => Ok( + Ok(Ok(response)) => Ok(response), + Ok(Err(_)) | Err(_) => Ok( crate::daemon_contract::DaemonInvocationResponse::problem( target_request_id, crate::daemon_contract::DaemonInvocationProblem::ResetRequired, diff --git a/src/daemon_client/controlled_invocation_tests.rs b/src/daemon_client/controlled_invocation_tests.rs index 7a00154e8..03389c11c 100644 --- a/src/daemon_client/controlled_invocation_tests.rs +++ b/src/daemon_client/controlled_invocation_tests.rs @@ -213,14 +213,16 @@ async fn reset_then_reconnect_client( .expect("read cancellation request") .expect("cancellation request"); - let second_stream = listener.accept().await.expect("accept second invocation"); - let (second_reader, mut second_writer) = second_stream.into_split(); - let mut second_lines = BufReader::new(second_reader).lines(); - second_lines - .next_line() - .await - .expect("read second handshake") - .expect("second handshake"); + // The response-grace read polls liveness with handshake-less probe + // connections; skip them like the real daemon's accept loop does. + let (mut second_lines, mut second_writer) = loop { + let second_stream = listener.accept().await.expect("accept second invocation"); + let (second_reader, second_writer) = second_stream.into_split(); + let mut second_lines = BufReader::new(second_reader).lines(); + if let Ok(Some(_handshake)) = second_lines.next_line().await { + break (second_lines, second_writer); + } + }; let second_line = second_lines .next_line() .await @@ -447,8 +449,10 @@ async fn remote_effect_without_authoritative_settlement_returns_reset_required() }); let deadline = deadline_after(Duration::from_secs(10)); + // The unsettled server never answers, so the join bound must outlive the + // full authoritative response grace. let response = tokio::time::timeout( - crate::daemon::DAEMON_TASK_ABORT_DEADLINE + Duration::from_secs(1), + crate::daemon::DAEMON_TOOL_RESPONSE_GRACE + Duration::from_secs(1), client.invoke_controlled( invocation_request(REQUEST_ID, deadline.clone()), deadline, @@ -480,7 +484,7 @@ async fn remote_effect_cancel_delivery_failure_returns_reset_required() { let deadline = deadline_after(Duration::from_secs(10)); let response = tokio::time::timeout( - crate::daemon::DAEMON_TASK_ABORT_DEADLINE + Duration::from_secs(1), + crate::daemon::DAEMON_TOOL_RESPONSE_GRACE + Duration::from_secs(1), client.invoke_controlled( invocation_request(REQUEST_ID, deadline.clone()), deadline, diff --git a/src/remote_command.rs b/src/remote_command.rs index e6b0235af..1e8d57cf1 100644 --- a/src/remote_command.rs +++ b/src/remote_command.rs @@ -1,18 +1,24 @@ //! CLI presentation for the Remote Brain operator plane. +use std::fmt::Write as _; use std::io::Read; use std::path::{Path, PathBuf}; use std::time::Duration; use serde::Serialize; use serde::de::DeserializeOwned; -use tracedecay_application::RemoteListenerReadV1; +use tracedecay_application::remote::composition::{ + PendingLocalEvidenceV1, PendingLocalUnavailableReasonV1, ShardCoverageStateV1, +}; use tracedecay_application::remote::protocol::{ EnrollmentRequestV1, RemoteProtocolRequestV1, RemoteProtocolResponseV1, }; +use tracedecay_application::remote::query::{RemoteExactObservationResultV1, RemoteQueryResultV1}; use tracedecay_application::remote::status::{ RemoteOperationalReadinessV1, RemoteOperationalStatusReadV1, RemoteOperationalStatusV1, + RemoteSpoolOperationalStatusV1, }; +use tracedecay_application::{ApplicationOutcome, RemoteListenerReadV1}; use tracedecay_domain::CurrentRemoteAuthorityStateV1; use tracedecay_sdk::remote_client::{EnrolledRemoteClient, RemoteClientError}; @@ -39,6 +45,15 @@ pub enum RemoteCommand { args: RemoteProtocolArgs, enrollment_credential_file: PathBuf, }, + Capture { + args: RemoteProtocolArgs, + }, + Query { + args: RemoteProtocolArgs, + }, + TransferFrame { + args: RemoteProtocolArgs, + }, Replay { args: RemoteProtocolArgs, }, @@ -60,6 +75,36 @@ pub fn run(command: RemoteCommand) -> Result<()> { args, enrollment_credential_file, } => run_enroll(args, enrollment_credential_file), + RemoteCommand::Capture { args } => { + let request = read_protocol_request(&args.request_file)?; + let client = build_client(&args)?; + emit_protocol_response( + &client.capture(&request).map_err(map_remote_client_error)?, + args.json, + ) + } + RemoteCommand::Query { args } => { + let request = read_protocol_request(&args.request_file)?; + let client = build_client(&args)?; + let response = client.query(&request).map_err(map_remote_client_error)?; + // `--json` emits exactly the canonical wire response. + let local_spool = if args.json { + None + } else { + caller_local_spool_evidence(&response) + }; + emit_query_response(&response, local_spool.as_ref(), args.json) + } + RemoteCommand::TransferFrame { args } => { + let request = read_protocol_request(&args.request_file)?; + let client = build_client(&args)?; + emit_protocol_response( + &client + .transfer_frame(&request) + .map_err(map_remote_client_error)?, + args.json, + ) + } RemoteCommand::Replay { args } => { let request = read_protocol_request(&args.request_file)?; let client = build_client(&args)?; @@ -128,12 +173,24 @@ fn build_client(args: &RemoteProtocolArgs) -> Result { message: "Remote Brain --timeout-secs must be greater than zero".to_owned(), }); } + // A plaintext endpoint selects the local daemon's nested `/remote` mount; + // the SDK client fails closed on any non-loopback plaintext host. + let local_daemon_target = args.endpoint.starts_with("http://"); + if local_daemon_target && args.trust_root_file.is_some() { + return Err(TraceDecayError::Config { + message: "Remote Brain --trust-root-file applies only to HTTPS endpoints".to_owned(), + }); + } let credential = std::fs::read(&args.credential_file).map_err(|error| TraceDecayError::File { message: format!("failed to read Remote Brain credential file: {error}"), path: args.credential_file.display().to_string(), })?; let timeout = Duration::from_secs(args.timeout_secs); + if local_daemon_target { + return EnrolledRemoteClient::new_local_daemon(&args.endpoint, credential, timeout) + .map_err(map_remote_client_error); + } match &args.trust_root_file { Some(path) => { let pem = std::fs::read(path).map_err(|error| TraceDecayError::File { @@ -180,6 +237,29 @@ fn emit_protocol_response( } else { print!("{}", render_protocol_response(response)); } + protocol_exit_status(response) +} + +/// Emits a query response with its coverage evidence, so a found/not-found +/// answer is never read as complete while local captures have not replayed +/// or the shard disclosed a degraded state. +fn emit_query_response( + response: &RemoteProtocolResponseV1, + local_spool: Option<&RemoteSpoolOperationalStatusV1>, + json: bool, +) -> Result<()> { + if json { + print!("{}", canonical_json_line(response)?); + } else { + print!("{}", render_protocol_response(response)); + if let Some(result) = query_payload(response) { + print!("{}", render_query_coverage(result, local_spool)); + } + } + protocol_exit_status(response) +} + +fn protocol_exit_status(response: &RemoteProtocolResponseV1) -> Result<()> { match &response.result { Ok(_) => Ok(()), Err(problem) => Err(TraceDecayError::Config { @@ -191,6 +271,127 @@ fn emit_protocol_response( } } +/// The caller's own spool evidence for a query whose serving node answered +/// `RequestingNodeSpoolNotSupplied`, read from that spool's owning authority: +/// the local daemon's canonical operational status. Every other pending-local +/// answer keeps the serving node's own evidence, and a local daemon that +/// cannot answer leaves the wire's typed absence in place. +fn caller_local_spool_evidence( + response: &RemoteProtocolResponseV1, +) -> Option { + let payload = query_payload(response)?; + if !matches!( + payload.composition.pending_local, + PendingLocalEvidenceV1::Unavailable { + reason: PendingLocalUnavailableReasonV1::RequestingNodeSpoolNotSupplied, + } + ) { + return None; + } + match crate::daemon::live_remote_operational_status() { + Ok(RemoteOperationalStatusReadV1::Observed { status, .. }) => Some(status.spool), + Ok( + RemoteOperationalStatusReadV1::Unconfigured + | RemoteOperationalStatusReadV1::Unavailable, + ) + | Err(_) => None, + } +} + +fn query_payload( + response: &RemoteProtocolResponseV1, +) -> Option<&RemoteQueryResultV1> { + match &response.result { + Ok(envelope) => match &envelope.outcome { + ApplicationOutcome::Evidence(packet) => packet.payload.as_ref(), + ApplicationOutcome::Preview(_) | ApplicationOutcome::Effect(_) => None, + }, + Err(_) => None, + } +} + +fn render_query_coverage( + result: &RemoteQueryResultV1, + local_spool: Option<&RemoteSpoolOperationalStatusV1>, +) -> String { + let mut rendered = format!( + "Coverage: {}\n", + coverage_label(result.composition.coverage) + ); + for contribution in &result.composition.contributions { + let _ = write!( + rendered, + "Remote shard {}@{}: {}", + contribution.manifest.shard_id, + contribution.manifest.generation_id, + coverage_label(contribution.coverage), + ); + if let Some(reason) = &contribution.reason_code { + let _ = write!(rendered, " ({reason})"); + } + rendered.push('\n'); + } + match (&result.composition.pending_local, local_spool) { + (PendingLocalEvidenceV1::Available { evidence }, _) => { + let _ = write!( + rendered, + "Local pending captures: {}\nLocal sequence gap: {}\nLocal quarantined captures: {}\n", + evidence.count, + yes_no(evidence.has_sequence_gap), + yes_no(evidence.has_quarantined), + ); + } + ( + PendingLocalEvidenceV1::Unavailable { + reason: PendingLocalUnavailableReasonV1::RequestingNodeSpoolNotSupplied, + }, + Some(spool), + ) => { + let _ = write!( + rendered, + "Local pending captures (local daemon spool): {}\n\ +Local sequence gap (local daemon spool): {}\n\ +Local quarantined captures (local daemon spool): {}\n", + spool.pending_count, + yes_no(spool.has_sequence_gap), + spool.quarantined_count, + ); + } + (PendingLocalEvidenceV1::Unavailable { reason }, _) => { + let _ = writeln!( + rendered, + "Local pending captures: unavailable ({})", + pending_local_unavailable_label(*reason) + ); + } + } + let observation = match &result.observation { + RemoteExactObservationResultV1::Found(_) => "found", + RemoteExactObservationResultV1::NotFound => "not_found", + }; + let _ = writeln!(rendered, "Observation: {observation}"); + rendered +} + +fn coverage_label(coverage: ShardCoverageStateV1) -> &'static str { + match coverage { + ShardCoverageStateV1::Complete => "complete", + ShardCoverageStateV1::Stale => "stale", + ShardCoverageStateV1::Partial => "partial", + ShardCoverageStateV1::Unknown => "unknown", + ShardCoverageStateV1::Unavailable => "unavailable", + } +} + +fn pending_local_unavailable_label(reason: PendingLocalUnavailableReasonV1) -> &'static str { + match reason { + PendingLocalUnavailableReasonV1::RequestingNodeSpoolNotSupplied => { + "requesting node spool not supplied" + } + PendingLocalUnavailableReasonV1::AuthorityUnavailable => "authority unavailable", + } +} + fn map_remote_client_error(error: RemoteClientError) -> TraceDecayError { TraceDecayError::Config { message: error.to_string(), @@ -441,6 +642,161 @@ mod tests { } } + #[test] + fn query_human_render_surfaces_remote_and_local_coverage_honestly() { + let result: tracedecay_application::remote::query::RemoteQueryResultV1 = + serde_json::from_value(serde_json::json!({ + "composition": { + "contributions": [{ + "manifest": { + "brain_id": "brain.query", + "shard_id": "shard.project", + "generation_id": "generation.7", + "schema_digest": vec![1u8; 32], + "watermark_sequence": 9, + "placement_revision": 3, + "authority_epoch": 4, + "cache_age_millis": 10, + "cache_lag_commits": 0 + }, + "integrity": "verified", + "authenticity": "authenticated", + "freshness": "current", + "completeness": "complete", + "authorization": "authorized", + "coverage": "partial", + "authority_receipt": null, + "value": null, + "reason_code": "authorization_receipt_unavailable" + }], + "pending_local": { + "availability": "available", + "evidence": { + "count": 2, + "oldest_age_millis": 50, + "has_sequence_gap": true, + "has_quarantined": false + } + }, + "coverage": "partial" + }, + "observation": { "state": "not_found" } + })) + .expect("query result fixture"); + + let rendered = super::render_query_coverage(&result, None); + assert!(rendered.contains("Coverage: partial")); + assert!(rendered.contains( + "Remote shard shard.project@generation.7: partial (authorization_receipt_unavailable)" + )); + assert!(rendered.contains("Local pending captures: 2")); + assert!(rendered.contains("Local sequence gap: yes")); + assert!(rendered.contains("Local quarantined captures: no")); + assert!(rendered.contains("Observation: not_found")); + } + + #[test] + fn query_human_render_names_an_unavailable_local_spool() { + let result: tracedecay_application::remote::query::RemoteQueryResultV1 = + serde_json::from_value(serde_json::json!({ + "composition": { + "contributions": [], + "pending_local": { + "availability": "unavailable", + "reason": "requesting_node_spool_not_supplied" + }, + "coverage": "unknown" + }, + "observation": { "state": "not_found" } + })) + .expect("query result fixture"); + + let rendered = super::render_query_coverage(&result, None); + assert!( + rendered.contains( + "Local pending captures: unavailable (requesting node spool not supplied)" + ) + ); + } + + #[test] + fn query_human_render_composes_the_callers_own_spool_when_not_supplied() { + let result: tracedecay_application::remote::query::RemoteQueryResultV1 = + serde_json::from_value(serde_json::json!({ + "composition": { + "contributions": [], + "pending_local": { + "availability": "unavailable", + "reason": "requesting_node_spool_not_supplied" + }, + "coverage": "unknown" + }, + "observation": { "state": "not_found" } + })) + .expect("query result fixture"); + let spool = tracedecay_application::remote::status::RemoteSpoolOperationalStatusV1 { + pending_count: 4, + quarantined_count: 1, + has_sequence_gap: true, + }; + + let rendered = super::render_query_coverage(&result, Some(&spool)); + assert!(rendered.contains("Local pending captures (local daemon spool): 4")); + assert!(rendered.contains("Local sequence gap (local daemon spool): yes")); + assert!(rendered.contains("Local quarantined captures (local daemon spool): 1")); + assert!( + !rendered.contains("unavailable (requesting node spool not supplied)"), + "composed caller evidence must replace the not-supplied absence: {rendered}" + ); + } + + #[test] + fn query_human_render_keeps_other_absences_typed_even_with_local_evidence() { + let result: tracedecay_application::remote::query::RemoteQueryResultV1 = + serde_json::from_value(serde_json::json!({ + "composition": { + "contributions": [], + "pending_local": { + "availability": "unavailable", + "reason": "authority_unavailable" + }, + "coverage": "unknown" + }, + "observation": { "state": "not_found" } + })) + .expect("query result fixture"); + let spool = tracedecay_application::remote::status::RemoteSpoolOperationalStatusV1 { + pending_count: 4, + quarantined_count: 0, + has_sequence_gap: false, + }; + + let rendered = super::render_query_coverage(&result, Some(&spool)); + assert!( + rendered.contains("Local pending captures: unavailable (authority unavailable)"), + "a serving node's own authority absence is not the caller's to hydrate: {rendered}" + ); + } + + #[test] + fn build_client_rejects_a_trust_root_for_a_local_daemon_endpoint() { + let error = build_client(&RemoteProtocolArgs { + endpoint: "http://127.0.0.1:39181/remote/".to_owned(), + credential_file: PathBuf::from("/this/file/must-not-be-read.bin"), + trust_root_file: Some(PathBuf::from("/this/file/must-not-be-read.pem")), + timeout_secs: 30, + request_file: PathBuf::from("request.json"), + json: false, + }) + .expect_err("a trust root with a plaintext loopback endpoint must fail closed"); + match error { + TraceDecayError::Config { message } => { + assert!(message.contains("--trust-root-file")); + } + other => panic!("expected config error, got {other:?}"), + } + } + #[test] fn build_client_rejects_zero_timeout_before_reading_files() { let error = build_client(&RemoteProtocolArgs {