From f6ba20736e31bb0b611c7a558b5849f379c6df28 Mon Sep 17 00:00:00 2001 From: shanzhao Date: Fri, 21 Aug 2026 22:08:16 +0800 Subject: [PATCH 1/3] feat(telemetry): archive model-visible tool evidence --- Cargo.lock | 1 + Cargo.toml | 1 + DEVELOPMENT.md | 1 + core/Cargo.toml | 1 + core/src/telemetry/evidence.rs | 642 +++++++++++++++++++++++++++++++++ core/src/telemetry/mod.rs | 183 ++++++++++ core/src/telemetry/trace.rs | 44 ++- docs/model-visible-evidence.md | 171 +++++++++ site/api/evidence.js | 290 +++++++++++++++ site/vercel.json | 7 + 10 files changed, 1340 insertions(+), 1 deletion(-) create mode 100644 core/src/telemetry/evidence.rs create mode 100644 docs/model-visible-evidence.md create mode 100644 site/api/evidence.js diff --git a/Cargo.lock b/Cargo.lock index ffea8eef..a0d7971f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5089,6 +5089,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "sha2", "symphonia", "tempfile", "thiserror 2.0.18", diff --git a/Cargo.toml b/Cargo.toml index e9e7f516..51ed4b8a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,7 @@ async-tungstenite = { version = "0.27", default-features = false, features = ["t serde = { version = "1", features = ["derive"] } serde_json = "1" base64 = "0.22" +sha2 = "0.10" reqwest = { version = "0.12", features = ["json", "stream"] } diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index eb689671..6371af37 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -139,6 +139,7 @@ in [Website deployment](docs/website-deployment.md). | [Context window management](docs/context-window-management.md) | Agent turns, tool-result bounds, sawtooth compaction, prompt caching, and artifact evidence retention. | | [CLI telemetry schema](docs/telemetry-schema.md) | Telemetry schema, privacy, and configuration contract for the CLI daemon. | | [Telemetry runbook](docs/development/telemetry-runbook.md) | Maintainer runbook for operating CLI telemetry. | +| [Model-visible evidence archive](docs/model-visible-evidence.md) | Lossless ToolResult archive, Axiom proxy setup, privacy gates, and recovery. | | [Release flow](docs/release-flow.md) | GitHub Release workflow, platform build graph, assets, and installer smoke tests. | | [Website deployment](docs/website-deployment.md) | Vercel deployment runbook for `socai.io`. | | [Website launch QA](docs/website-launch-qa.md) | Launch checklist used for the `socai.io` rollout. | diff --git a/core/Cargo.toml b/core/Cargo.toml index 2e98df2a..100e0090 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -16,6 +16,7 @@ anyhow.workspace = true tracing.workspace = true reqwest.workspace = true base64.workspace = true +sha2.workspace = true dirs = "5" libc.workspace = true uuid = { version = "1", features = ["v4"] } diff --git a/core/src/telemetry/evidence.rs b/core/src/telemetry/evidence.rs new file mode 100644 index 00000000..db8a207a --- /dev/null +++ b/core/src/telemetry/evidence.rs @@ -0,0 +1,642 @@ +//! Lossless archive of the tool results that were present on provider request +//! wires. The source is the existing `llm/NNN.request.json` run artifact, not +//! raw tool output and not the size-bounded OTLP chat representation. + +use std::collections::{HashMap, HashSet}; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +use chrono::Utc; +use serde_json::{json, Map, Value}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use super::trace::redact_secrets_in_value; +use crate::agent::llm::LLMResponse; + +pub(crate) const SCHEMA_VERSION: &str = "socai.model-visible-evidence.v1"; +const ARCHIVE_RELATIVE_PATH: &str = "evidence/model-visible-v1.json"; +const CHUNK_MAX_BYTES: usize = 24 * 1024; + +#[derive(Debug)] +struct TraceIdentity { + trace_id: String, + root_span_id: String, +} + +#[derive(Debug)] +struct RunMetadata { + run_id: String, + provider: String, + model: String, + steps: u64, +} + +#[derive(Debug)] +struct ExtractedToolResult { + tool_call_id: String, + tool_name: Option, + message_index: usize, + result_index: usize, + wire_format: &'static str, + content: Value, +} + +/// Build a model-visible evidence archive, keep a shareable copy in the run, +/// patch the trace with a small local summary, and durably stage the upload. +pub(crate) fn stage_run_archive( + run_dir: &Path, + pending_dir: &Path, + content_enabled: bool, +) -> io::Result { + let archive = build_archive(run_dir, content_enabled)?; + let evaluation_id = archive + .get("evaluation_id") + .and_then(Value::as_str) + .ok_or_else(|| io::Error::other("evidence archive has no evaluation_id"))?; + + let local_path = run_dir.join(ARCHIVE_RELATIVE_PATH); + write_json_atomic(&local_path, &archive)?; + patch_trace_summary(run_dir, &archive)?; + + fs::create_dir_all(pending_dir)?; + let destination = pending_dir.join(format!("{}.json", evaluation_id.replace(':', "-"))); + write_json_atomic(&destination, &archive)?; + Ok(destination) +} + +fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { + let trace = read_json(&run_dir.join("trace.json"))?; + let identity = trace_identity(&trace) + .ok_or_else(|| io::Error::other("trace.json has no root trace/span id"))?; + let run = read_json(&run_dir.join("run.json"))?; + let metadata = run_metadata(&run); + let evaluation_id = format!("{}:{}", identity.trace_id, identity.root_span_id); + let created_at = Utc::now().to_rfc3339(); + + let mut records = Vec::new(); + let mut index_entries = Vec::new(); + let mut seen_evidence = HashSet::new(); + let mut accepted_request_count = 0usize; + let mut evidence_object_count = 0usize; + let mut evidence_chunk_count = 0usize; + let mut total_content_bytes = 0usize; + let mut partial = false; + + if content_enabled { + let requests = request_files(run_dir)?; + if requests.is_empty() && metadata.steps > 0 { + partial = true; + } + for (step, request_path) in requests { + let response_path = run_dir.join("llm").join(format!("{step:03}.response.json")); + let request_status = response_status(&response_path); + let mut evidence_ids = Vec::new(); + let mut extraction_error = None; + + if request_status == "accepted" { + accepted_request_count += 1; + match read_json(&request_path).and_then(|payload| { + extract_model_visible_tool_results(&metadata.provider, &payload) + }) { + Ok(results) => { + for result in results { + let mut content = result.content; + let original = content.clone(); + redact_secrets_in_value(&mut content); + let redaction_count = changed_leaf_count(&original, &content); + let content_bytes = + serde_json::to_vec(&content).map_err(io::Error::other)?; + let content_sha256 = sha256_hex(&content_bytes); + let evidence_id = + evidence_id(&evaluation_id, &result.tool_call_id, &content_sha256); + evidence_ids.push(evidence_id.clone()); + + if !seen_evidence.insert(evidence_id.clone()) { + continue; + } + + let content_text = + String::from_utf8(content_bytes).map_err(io::Error::other)?; + let chunks = split_utf8(&content_text, CHUNK_MAX_BYTES); + let chunk_count = chunks.len(); + for (chunk_index, chunk_text) in chunks.into_iter().enumerate() { + let chunk_bytes = chunk_text.as_bytes(); + records.push(common_record( + &identity, + &metadata, + &evaluation_id, + &created_at, + json!({ + "record_type": "evidence_chunk", + "evidence_id": evidence_id, + "tool_call_id": result.tool_call_id, + "tool_name": result.tool_name, + "wire_format": result.wire_format, + "first_observed_step": step, + "message_index": result.message_index, + "result_index": result.result_index, + "content_encoding": "canonical-json-utf8", + "content_sha256": content_sha256, + "content_bytes": content_text.len(), + "chunk_index": chunk_index, + "chunk_count": chunk_count, + "chunk_sha256": sha256_hex(chunk_bytes), + "chunk_bytes": chunk_bytes.len(), + "chunk_text": chunk_text, + "redaction_version": "socai-secret-redactor-v1", + "redaction_count": redaction_count, + "semantic_redaction": false, + }), + )); + } + evidence_object_count += 1; + evidence_chunk_count += chunk_count; + total_content_bytes += content_text.len(); + index_entries.push(json!({ + "evidence_id": evidence_id, + "content_sha256": content_sha256, + "chunk_count": chunk_count, + })); + } + } + Err(error) => { + partial = true; + extraction_error = Some(error.to_string()); + } + } + } else if request_status == "unknown" { + partial = true; + } + + dedupe_preserving_order(&mut evidence_ids); + let manifest_body = json!({ + "step": step, + "request_status": if extraction_error.is_some() { "unsupported" } else { request_status }, + "evidence_ids": evidence_ids, + }); + let manifest_sha256 = sha256_json(&manifest_body)?; + index_entries.push(json!({ + "step": step, + "manifest_sha256": manifest_sha256, + })); + records.push(common_record( + &identity, + &metadata, + &evaluation_id, + &created_at, + json!({ + "record_type": "request_manifest", + "step": step, + "request_status": if extraction_error.is_some() { "unsupported" } else { request_status }, + "evidence_ids": manifest_body["evidence_ids"].clone(), + "evidence_count": manifest_body["evidence_ids"].as_array().map_or(0, Vec::len), + "manifest_sha256": manifest_sha256, + "error": extraction_error, + }), + )); + } + } + + let archive_status = if !content_enabled { + "disabled" + } else if partial { + "partial" + } else { + "complete" + }; + let evidence_index_sha256 = sha256_json(&Value::Array(index_entries))?; + records.push(common_record( + &identity, + &metadata, + &evaluation_id, + &created_at, + json!({ + "record_type": "turn_commit", + "archive_status": archive_status, + "accepted_request_count": accepted_request_count, + "evidence_object_count": evidence_object_count, + "evidence_chunk_count": evidence_chunk_count, + "total_content_bytes": total_content_bytes, + "evidence_index_sha256": evidence_index_sha256, + "telemetry_policy": if content_enabled { "chat_and_evidence_enabled" } else { "evidence_disabled" }, + "committed_at": created_at, + }), + )); + + Ok(json!({ + "schema_version": SCHEMA_VERSION, + "evaluation_id": evaluation_id, + "trace_id": identity.trace_id, + "root_span_id": identity.root_span_id, + "run_id": metadata.run_id, + "archive_status": archive_status, + "accepted_request_count": accepted_request_count, + "evidence_object_count": evidence_object_count, + "evidence_chunk_count": evidence_chunk_count, + "total_content_bytes": total_content_bytes, + "records": records, + })) +} + +fn extract_model_visible_tool_results( + provider: &str, + payload: &Value, +) -> io::Result> { + if payload.get("input").and_then(Value::as_array).is_some() { + return extract_openai_responses(payload); + } + if provider.eq_ignore_ascii_case("anthropic") || has_anthropic_tool_result(payload) { + return extract_anthropic_messages(payload); + } + if payload.get("messages").and_then(Value::as_array).is_some() { + return extract_openai_chat(payload); + } + Err(io::Error::other(format!( + "unsupported provider request shape: {provider}" + ))) +} + +fn extract_openai_chat(payload: &Value) -> io::Result> { + let messages = payload + .get("messages") + .and_then(Value::as_array) + .ok_or_else(|| io::Error::other("OpenAI chat request has no messages"))?; + let mut names = HashMap::new(); + for message in messages { + let Some(calls) = message.get("tool_calls").and_then(Value::as_array) else { + continue; + }; + for call in calls { + if let (Some(id), Some(name)) = ( + call.get("id").and_then(Value::as_str), + call.pointer("/function/name").and_then(Value::as_str), + ) { + names.insert(id.to_string(), name.to_string()); + } + } + } + + let mut results = Vec::new(); + for (message_index, message) in messages.iter().enumerate() { + if message.get("role").and_then(Value::as_str) != Some("tool") { + continue; + } + let tool_call_id = message + .get("tool_call_id") + .and_then(Value::as_str) + .ok_or_else(|| io::Error::other("OpenAI tool message has no tool_call_id"))?; + let content = message + .get("content") + .ok_or_else(|| io::Error::other("OpenAI tool message has no content"))?; + results.push(ExtractedToolResult { + tool_call_id: tool_call_id.to_string(), + tool_name: names.get(tool_call_id).cloned(), + message_index, + result_index: results.len(), + wire_format: "openai_chat_tool_message", + content: content.clone(), + }); + } + Ok(results) +} + +fn extract_openai_responses(payload: &Value) -> io::Result> { + let input = payload + .get("input") + .and_then(Value::as_array) + .ok_or_else(|| io::Error::other("OpenAI Responses request has no input"))?; + let mut names = HashMap::new(); + for item in input { + if item.get("type").and_then(Value::as_str) != Some("function_call") { + continue; + } + if let (Some(id), Some(name)) = ( + item.get("call_id").and_then(Value::as_str), + item.get("name").and_then(Value::as_str), + ) { + names.insert(id.to_string(), name.to_string()); + } + } + + let mut results = Vec::new(); + for (message_index, item) in input.iter().enumerate() { + if item.get("type").and_then(Value::as_str) != Some("function_call_output") { + continue; + } + let tool_call_id = item + .get("call_id") + .and_then(Value::as_str) + .ok_or_else(|| io::Error::other("function_call_output has no call_id"))?; + let content = item + .get("output") + .ok_or_else(|| io::Error::other("function_call_output has no output"))?; + results.push(ExtractedToolResult { + tool_call_id: tool_call_id.to_string(), + tool_name: names.get(tool_call_id).cloned(), + message_index, + result_index: results.len(), + wire_format: "openai_responses_function_call_output", + content: content.clone(), + }); + } + Ok(results) +} + +fn extract_anthropic_messages(payload: &Value) -> io::Result> { + let messages = payload + .get("messages") + .and_then(Value::as_array) + .ok_or_else(|| io::Error::other("Anthropic request has no messages"))?; + let mut names = HashMap::new(); + for message in messages { + let Some(blocks) = message.get("content").and_then(Value::as_array) else { + continue; + }; + for block in blocks { + if block.get("type").and_then(Value::as_str) != Some("tool_use") { + continue; + } + if let (Some(id), Some(name)) = ( + block.get("id").and_then(Value::as_str), + block.get("name").and_then(Value::as_str), + ) { + names.insert(id.to_string(), name.to_string()); + } + } + } + + let mut results = Vec::new(); + for (message_index, message) in messages.iter().enumerate() { + let Some(blocks) = message.get("content").and_then(Value::as_array) else { + continue; + }; + for (block_index, block) in blocks.iter().enumerate() { + if block.get("type").and_then(Value::as_str) != Some("tool_result") { + continue; + } + let tool_call_id = block + .get("tool_use_id") + .and_then(Value::as_str) + .ok_or_else(|| io::Error::other("Anthropic tool_result has no tool_use_id"))?; + let content = block + .get("content") + .ok_or_else(|| io::Error::other("Anthropic tool_result has no content"))?; + results.push(ExtractedToolResult { + tool_call_id: tool_call_id.to_string(), + tool_name: names.get(tool_call_id).cloned(), + message_index, + result_index: block_index, + wire_format: "anthropic_tool_result_block", + content: content.clone(), + }); + } + } + Ok(results) +} + +fn has_anthropic_tool_result(payload: &Value) -> bool { + payload + .get("messages") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|message| message.get("content").and_then(Value::as_array)) + .flatten() + .any(|block| block.get("type").and_then(Value::as_str) == Some("tool_result")) +} + +fn request_files(run_dir: &Path) -> io::Result> { + let llm_dir = run_dir.join("llm"); + let mut files = Vec::new(); + let Ok(entries) = fs::read_dir(llm_dir) else { + return Ok(files); + }; + for entry in entries { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + let Some(step) = name + .strip_suffix(".request.json") + .and_then(|value| value.parse::().ok()) + else { + continue; + }; + files.push((step, entry.path())); + } + files.sort_by_key(|(step, _)| *step); + Ok(files) +} + +fn response_status(path: &Path) -> &'static str { + let Ok(response) = read_json(path) else { + return "unknown"; + }; + if response.get("error").is_some_and(|error| !error.is_null()) { + "failed" + } else if serde_json::from_value::(response).is_ok() { + "accepted" + } else { + "unknown" + } +} + +fn trace_identity(trace: &Value) -> Option { + let root = trace + .pointer("/resourceSpans/0/scopeSpans/0/spans") + .and_then(Value::as_array)? + .iter() + .find(|span| span.get("parentSpanId").is_none())?; + Some(TraceIdentity { + trace_id: root.get("traceId")?.as_str()?.to_string(), + root_span_id: root.get("spanId")?.as_str()?.to_string(), + }) +} + +fn run_metadata(run: &Value) -> RunMetadata { + RunMetadata { + run_id: value_string(run, "id"), + provider: value_string(run, "provider"), + model: value_string(run, "model"), + steps: run.get("steps").and_then(Value::as_u64).unwrap_or_default(), + } +} + +fn value_string(value: &Value, key: &str) -> String { + value + .get(key) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() +} + +fn common_record( + identity: &TraceIdentity, + metadata: &RunMetadata, + evaluation_id: &str, + created_at: &str, + fields: Value, +) -> Value { + let mut record = match fields { + Value::Object(map) => map, + _ => Map::new(), + }; + record.insert("schema_version".into(), json!(SCHEMA_VERSION)); + record.insert("evaluation_id".into(), json!(evaluation_id)); + record.insert("trace_id".into(), json!(identity.trace_id)); + record.insert("root_span_id".into(), json!(identity.root_span_id)); + record.insert("run_id".into(), json!(metadata.run_id)); + record.insert("provider".into(), json!(metadata.provider)); + record.insert("model".into(), json!(metadata.model)); + record.insert("created_at".into(), json!(created_at)); + Value::Object(record) +} + +fn patch_trace_summary(run_dir: &Path, archive: &Value) -> io::Result<()> { + let trace_path = run_dir.join("trace.json"); + let mut trace = read_json(&trace_path)?; + let root = trace + .pointer_mut("/resourceSpans/0/scopeSpans/0/spans") + .and_then(Value::as_array_mut) + .and_then(|spans| { + spans + .iter_mut() + .find(|span| span.get("parentSpanId").is_none()) + }) + .ok_or_else(|| io::Error::other("trace.json has no root span"))?; + let attributes = root + .get_mut("attributes") + .and_then(Value::as_array_mut) + .ok_or_else(|| io::Error::other("trace root has no attributes"))?; + const KEYS: [&str; 6] = [ + "socai.evidence.schema_version", + "socai.evidence.local_status", + "socai.evidence.upload_status_at_trace_build", + "socai.evidence.accepted_request_count", + "socai.evidence.object_count", + "socai.evidence.total_bytes", + ]; + attributes.retain(|attribute| { + let key = attribute + .get("key") + .and_then(Value::as_str) + .unwrap_or_default(); + !KEYS.contains(&key) + }); + for (key, value) in [ + ("socai.evidence.schema_version", json!(SCHEMA_VERSION)), + ( + "socai.evidence.local_status", + archive + .get("archive_status") + .cloned() + .unwrap_or(Value::Null), + ), + ( + "socai.evidence.upload_status_at_trace_build", + json!("queued"), + ), + ] { + attributes.push( + json!({ "key": key, "value": { "stringValue": value.as_str().unwrap_or_default() } }), + ); + } + for (key, field) in [ + ( + "socai.evidence.accepted_request_count", + "accepted_request_count", + ), + ("socai.evidence.object_count", "evidence_object_count"), + ("socai.evidence.total_bytes", "total_content_bytes"), + ] { + let value = archive + .get(field) + .and_then(Value::as_u64) + .unwrap_or_default(); + attributes.push(json!({ "key": key, "value": { "intValue": value.to_string() } })); + } + write_json_atomic(&trace_path, &trace) +} + +fn read_json(path: &Path) -> io::Result { + let bytes = fs::read(path)?; + serde_json::from_slice(&bytes).map_err(io::Error::other) +} + +fn write_json_atomic(path: &Path, value: &Value) -> io::Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let temporary = path.with_extension(format!("{}.tmp", Uuid::new_v4())); + let bytes = serde_json::to_vec(value).map_err(io::Error::other)?; + fs::write(&temporary, bytes)?; + if path.exists() { + let _ = fs::remove_file(path); + } + fs::rename(temporary, path) +} + +fn changed_leaf_count(before: &Value, after: &Value) -> usize { + match (before, after) { + (Value::Array(left), Value::Array(right)) => left + .iter() + .zip(right) + .map(|(a, b)| changed_leaf_count(a, b)) + .sum(), + (Value::Object(left), Value::Object(right)) => left + .iter() + .map(|(key, value)| { + right + .get(key) + .map_or(1, |other| changed_leaf_count(value, other)) + }) + .sum(), + _ => usize::from(before != after), + } +} + +fn split_utf8(text: &str, max_bytes: usize) -> Vec { + if text.is_empty() { + return vec![String::new()]; + } + let mut chunks = Vec::new(); + let mut start = 0; + while start < text.len() { + let mut end = (start + max_bytes).min(text.len()); + while end > start && !text.is_char_boundary(end) { + end -= 1; + } + if end == start { + end = text.len(); + } + chunks.push(text[start..end].to_string()); + start = end; + } + chunks +} + +fn dedupe_preserving_order(values: &mut Vec) { + let mut seen = HashSet::new(); + values.retain(|value| seen.insert(value.clone())); +} + +fn evidence_id(evaluation_id: &str, tool_call_id: &str, content_sha256: &str) -> String { + let input = format!("{evaluation_id}\0{tool_call_id}\0{content_sha256}"); + format!("ev_{}", sha256_hex(input.as_bytes())) +} + +fn sha256_json(value: &Value) -> io::Result { + let bytes = serde_json::to_vec(value).map_err(io::Error::other)?; + Ok(sha256_hex(&bytes)) +} + +fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let mut output = String::with_capacity(digest.len() * 2); + for byte in digest { + use std::fmt::Write; + let _ = write!(output, "{byte:02x}"); + } + output +} diff --git a/core/src/telemetry/mod.rs b/core/src/telemetry/mod.rs index 3a7ea217..12043535 100644 --- a/core/src/telemetry/mod.rs +++ b/core/src/telemetry/mod.rs @@ -7,6 +7,7 @@ use tokio::io::AsyncWriteExt; use tokio::sync::mpsc; use tokio::time::MissedTickBehavior; +mod evidence; pub mod tool_call; pub mod trace; @@ -15,6 +16,7 @@ pub use trace::redact_secrets; const EVENT_SCHEMA_VERSION: u32 = 1; const TELEMETRY_ENDPOINT: &str = "https://socai.io/v1/events"; const TRACES_ENDPOINT: &str = "https://socai.io/v1/traces"; +const EVIDENCE_ENDPOINT: &str = "https://socai.io/v1/evidence"; const CHANNEL_CAPACITY: usize = 512; const REMOTE_BATCH_SIZE: usize = 25; const REMOTE_FLUSH_INTERVAL: Duration = Duration::from_secs(5); @@ -23,6 +25,7 @@ const TRACE_UPLOAD_TIMEOUT: Duration = Duration::from_secs(30); const TRACE_RETRY_BATCH_SIZE: usize = 5; const TRACE_FILE_READY_RETRIES: usize = 20; const TRACE_FILE_READY_DELAY: Duration = Duration::from_millis(50); +const EVIDENCE_BATCH_MAX_BYTES: usize = 512 * 1024; /// Which socai surface is emitting telemetry. Carried verbatim into the `source` /// field of every event and used to decide which device context is meaningful @@ -52,6 +55,7 @@ impl TelemetrySource { pub struct Telemetry { sender: mpsc::Sender, pending_trace_dir: PathBuf, + pending_evidence_dir: PathBuf, } #[derive(Debug)] @@ -61,6 +65,10 @@ enum QueuedItem { TraceFile(PathBuf), /// Durable, identity-free copy under telemetry/pending-traces. PendingTrace(PathBuf), + /// Run directory whose finalized provider request artifacts are not ready. + EvidenceRun(PathBuf), + /// Durable, identity-free archive under telemetry/pending-evidence. + PendingEvidence(PathBuf), } #[derive(Debug)] @@ -76,6 +84,7 @@ struct WorkerConfig { source: TelemetrySource, local_path: PathBuf, pending_trace_dir: PathBuf, + pending_evidence_dir: PathBuf, } #[derive(Debug, Clone)] @@ -95,6 +104,7 @@ impl Telemetry { let session_id = new_session_id(); let local_path = home.join("telemetry/events.jsonl"); let pending_trace_dir = home.join("telemetry/pending-traces"); + let pending_evidence_dir = home.join("telemetry/pending-evidence"); let (sender, receiver) = mpsc::channel(CHANNEL_CAPACITY); let config = WorkerConfig { @@ -103,12 +113,14 @@ impl Telemetry { source, local_path, pending_trace_dir: pending_trace_dir.clone(), + pending_evidence_dir: pending_evidence_dir.clone(), }; spawn_worker(receiver, config); Self { sender, pending_trace_dir, + pending_evidence_dir, } } @@ -126,6 +138,16 @@ impl Telemetry { /// cancellation can call this just before the trace drop guard finishes; /// that source path is retried briefly by the worker. pub fn upload_run_trace(&self, run_dir: &Path) -> bool { + let (evidence_item, _) = match evidence::stage_run_archive( + run_dir, + &self.pending_evidence_dir, + evidence_text_enabled(), + ) { + Ok(path) => (QueuedItem::PendingEvidence(path), true), + Err(_) => (QueuedItem::EvidenceRun(run_dir.to_path_buf()), false), + }; + let _ = self.sender.try_send(evidence_item); + let source = run_dir.join("trace.json"); let (item, staged) = match stage_trace_file(&source, &self.pending_trace_dir) { Ok(path) => (QueuedItem::PendingTrace(path), true), @@ -202,6 +224,14 @@ async fn worker_loop(mut receiver: mpsc::Receiver, config: WorkerCon QueuedItem::PendingTrace(path) => { upload_pending_trace(&client, &config, &path).await; } + QueuedItem::EvidenceRun(run_dir) => { + if let Some(path) = stage_evidence_when_ready(&run_dir, &config.pending_evidence_dir).await { + upload_pending_evidence(&client, &config, &path).await; + } + } + QueuedItem::PendingEvidence(path) => { + upload_pending_evidence(&client, &config, &path).await; + } } } _ = flush_tick.tick() => { @@ -209,6 +239,7 @@ async fn worker_loop(mut receiver: mpsc::Receiver, config: WorkerCon } _ = trace_retry_tick.tick() => { retry_pending_traces(&client, &config).await; + retry_pending_evidence(&client, &config).await; } } } @@ -259,6 +290,19 @@ async fn stage_trace_file_when_ready(source: &Path, pending_dir: &Path) -> Optio None } +async fn stage_evidence_when_ready(run_dir: &Path, pending_dir: &Path) -> Option { + for attempt in 0..TRACE_FILE_READY_RETRIES { + match evidence::stage_run_archive(run_dir, pending_dir, evidence_text_enabled()) { + Ok(path) => return Some(path), + Err(_) if attempt + 1 < TRACE_FILE_READY_RETRIES => { + tokio::time::sleep(TRACE_FILE_READY_DELAY).await; + } + Err(_) => return None, + } + } + None +} + async fn retry_pending_traces(client: &reqwest::Client, config: &WorkerConfig) { let Ok(mut entries) = tokio::fs::read_dir(&config.pending_trace_dir).await else { return; @@ -278,6 +322,25 @@ async fn retry_pending_traces(client: &reqwest::Client, config: &WorkerConfig) { } } +async fn retry_pending_evidence(client: &reqwest::Client, config: &WorkerConfig) { + let Ok(mut entries) = tokio::fs::read_dir(&config.pending_evidence_dir).await else { + return; + }; + let mut paths = Vec::new(); + while paths.len() < TRACE_RETRY_BATCH_SIZE { + let Ok(Some(entry)) = entries.next_entry().await else { + break; + }; + let path = entry.path(); + if path.extension().and_then(|ext| ext.to_str()) == Some("json") { + paths.push(path); + } + } + for path in paths { + upload_pending_evidence(client, config, &path).await; + } +} + /// Append identity resource attributes and POST one durable pending trace. /// The spool file is removed only after the proxy acknowledges the handoff. async fn upload_pending_trace(client: &reqwest::Client, config: &WorkerConfig, path: &Path) { @@ -296,6 +359,107 @@ async fn upload_pending_trace(client: &reqwest::Client, config: &WorkerConfig, p } } +/// Upload content and request manifests first, then the terminal commit. The +/// spool file is retained unless every batch receives a successful proxy ack. +async fn upload_pending_evidence(client: &reqwest::Client, config: &WorkerConfig, path: &Path) { + let Ok(bytes) = tokio::fs::read(path).await else { + return; + }; + let Ok(payload) = serde_json::from_slice::(&bytes) else { + return; + }; + let Some(evaluation_id) = payload.get("evaluation_id").and_then(Value::as_str) else { + return; + }; + let Some(records) = payload.get("records").and_then(Value::as_array) else { + return; + }; + + let mut body_records = Vec::new(); + let mut commits = Vec::new(); + for record in records { + let mut record = record.clone(); + enrich_evidence_record(&mut record, config); + if record.get("record_type").and_then(Value::as_str) == Some("turn_commit") { + commits.push(record); + } else { + body_records.push(record); + } + } + if commits.len() != 1 { + return; + } + if !upload_evidence_batches(client, evaluation_id, body_records).await { + return; + } + if !upload_evidence_batches(client, evaluation_id, commits).await { + return; + } + let _ = tokio::fs::remove_file(path).await; +} + +async fn upload_evidence_batches( + client: &reqwest::Client, + evaluation_id: &str, + records: Vec, +) -> bool { + let mut batch = Vec::new(); + for record in records { + let mut candidate = batch.clone(); + candidate.push(record.clone()); + let candidate_body = json!({ + "schema_version": evidence::SCHEMA_VERSION, + "evaluation_id": evaluation_id, + "records": candidate, + }); + let candidate_len = + serde_json::to_vec(&candidate_body).map_or(usize::MAX, |body| body.len()); + let should_flush = candidate_len > EVIDENCE_BATCH_MAX_BYTES && !batch.is_empty(); + if should_flush + && !post_evidence_batch(client, evaluation_id, std::mem::take(&mut batch)).await + { + return false; + } + batch.push(record); + } + batch.is_empty() || post_evidence_batch(client, evaluation_id, batch).await +} + +async fn post_evidence_batch( + client: &reqwest::Client, + evaluation_id: &str, + records: Vec, +) -> bool { + let expected = records.len() as u64; + let body = json!({ + "schema_version": evidence::SCHEMA_VERSION, + "evaluation_id": evaluation_id, + "records": records, + }); + let Ok(response) = client.post(evidence_endpoint()).json(&body).send().await else { + return false; + }; + if !response.status().is_success() { + return false; + } + let Ok(ack) = response.json::().await else { + return false; + }; + ack.get("ok").and_then(Value::as_bool) == Some(true) + && ack.get("accepted").and_then(Value::as_u64) == Some(expected) +} + +fn enrich_evidence_record(record: &mut Value, config: &WorkerConfig) { + let Some(object) = record.as_object_mut() else { + return; + }; + object.insert("source".into(), json!(config.source.as_str())); + object.insert("install_id".into(), json!(config.install_id)); + object.insert("app_session_id".into(), json!(config.session_id)); + object.insert("app_version".into(), json!(env!("CARGO_PKG_VERSION"))); + object.insert("platform".into(), json!(std::env::consts::OS)); +} + /// The on-disk trace stays identity-free so run dirs can be shared; the /// uploaded copy carries the same source/install identity as events. fn enrich_trace_resource(payload: &mut Value, config: &WorkerConfig) { @@ -354,6 +518,15 @@ fn traces_endpoint() -> String { .unwrap_or_else(|| TRACES_ENDPOINT.to_string()) } +/// `SOCAI_EVIDENCE_ENDPOINT` overrides the production evidence proxy. +fn evidence_endpoint() -> String { + std::env::var("SOCAI_EVIDENCE_ENDPOINT") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| EVIDENCE_ENDPOINT.to_string()) +} + fn enrich_properties(properties: Value, config: &WorkerConfig, timestamp_ms: u64) -> Value { let mut map = match properties { Value::Object(map) => map, @@ -486,6 +659,16 @@ pub fn chat_text_enabled() -> bool { ) } +/// Evidence content follows the chat-text privacy gate and has a narrower +/// opt-out for operators who still want bounded traces without full tool data. +pub fn evidence_text_enabled() -> bool { + chat_text_enabled() + && !env_value_is( + "SOCAI_TELEMETRY_EVIDENCE", + &["0", "false", "off", "disabled", "no"], + ) +} + fn device_info() -> &'static DeviceInfo { static DEVICE_INFO: OnceLock = OnceLock::new(); DEVICE_INFO.get_or_init(|| { diff --git a/core/src/telemetry/trace.rs b/core/src/telemetry/trace.rs index fac3028a..9f65aef4 100644 --- a/core/src/telemetry/trace.rs +++ b/core/src/telemetry/trace.rs @@ -832,7 +832,8 @@ const SECRET_JSON_FIELDS: [&str; 11] = [ /// - `Bearer ` header values pub fn redact_secrets(text: &str) -> String { let text = redact_json_secret_fields(text); - redact_token_runs(&text) + let text = redact_token_runs(&text); + redact_secret_query_values(&text) } /// Recursively scrub every string inside a JSON value — used for structured @@ -965,6 +966,47 @@ fn redact_token_runs(text: &str) -> String { out } +/// Scrub credentials embedded in URLs. These commonly appear in model-visible +/// browser tool results as query parameters rather than structured JSON keys. +fn redact_secret_query_values(text: &str) -> String { + const PARAMETERS: [&str; 3] = ["xsec_token=", "access_token=", "api_key="]; + let lower = text.to_ascii_lowercase(); + let mut out = String::with_capacity(text.len()); + let mut cursor = 0; + while cursor < text.len() { + let match_at = PARAMETERS + .iter() + .filter_map(|parameter| { + lower[cursor..] + .find(parameter) + .map(|offset| (cursor + offset, *parameter)) + }) + .min_by_key(|(index, _)| *index); + let Some((start, parameter)) = match_at else { + out.push_str(&text[cursor..]); + break; + }; + out.push_str(&text[cursor..start + parameter.len()]); + let value_start = start + parameter.len(); + let value_len = text.as_bytes()[value_start..] + .iter() + .take_while(|byte| { + !matches!( + **byte, + b'&' | b'#' | b' ' | b'\t' | b'\r' | b'\n' | b'"' | b'\'' + ) + }) + .count(); + if value_len == 0 { + cursor = value_start; + continue; + } + out.push_str("[redacted]"); + cursor = value_start + value_len; + } + out +} + fn token_run_len(bytes: &[u8], accept: impl Fn(u8) -> bool) -> usize { bytes.iter().take_while(|&&b| accept(b)).count() } diff --git a/docs/model-visible-evidence.md b/docs/model-visible-evidence.md new file mode 100644 index 00000000..857c61dd --- /dev/null +++ b/docs/model-visible-evidence.md @@ -0,0 +1,171 @@ +# Model-visible evidence archive + +socai archives the ToolResult values that were present on successful provider +request wires. This is a separate path from the bounded OTLP chat transcript: +the archive preserves what the agent could observe after normal context +management, without uploading raw tool data the agent never received. + +The archive does not add an LLM call, change the provider request, or increase +input tokens. It post-processes the exact request artifacts already written to +`/llm/NNN.request.json` after a run terminates. + +## Data flow + +```text +tool output + -> normal bound_content_for_history() + -> provider request + llm/NNN.request.json + -> provider-specific ToolResult extraction + -> secret redaction + -> SHA-256 content addressing and UTF-8-safe chunks + -> telemetry/pending-evidence + -> https://socai.io/v1/evidence + -> socai-evidence-prod +``` + +Raw `tools/*/output.json` files remain local and are not evidence archive +sources. If context management omitted comments before a provider request, the +archive contains the same omission marker rather than restoring those comments +from raw output. + +## Local files + +Each terminal run gets a shareable, identity-free archive: + +```text +/evidence/model-visible-v1.json +``` + +Uploads are staged before the telemetry worker returns: + +```text +/telemetry/pending-evidence/-.json +``` + +Pending files survive proxy or Axiom outages. The worker retries up to five +files every 30 seconds and removes a file only after all content/manifests and +the final commit receive successful proxy acknowledgements. + +## Schema + +Schema version: `socai.model-visible-evidence.v1`. + +The dataset contains three record types: + +| Record | Purpose | +| --- | --- | +| `evidence_chunk` | One UTF-8-safe piece of a canonical JSON ToolResult value. | +| `request_manifest` | Evidence IDs present in one logical provider request. | +| `turn_commit` | Terminal object/chunk counts and archive integrity hash. | + +Content is addressed by: + +```text +content_sha256 = SHA256(canonical_json(redacted_tool_result_value)) +evidence_id = SHA256(evaluation_id + tool_call_id + content_sha256) +``` + +Repeated history across provider requests references the same evidence object. +If later context management changes a ToolResult value, its content hash and +evidence ID change, preserving the model-visible version for each request. + +The client uploads all `evidence_chunk` and `request_manifest` records before +the single `turn_commit`. Duplicate records can occur after a lost HTTP +acknowledgement; readers must deduplicate by `(evidence_id, chunk_index)` and +reject conflicting values for the same stable key. + +## Provider request formats + +The extractor recognizes explicit wire shapes rather than recursively looking +for generic `content` fields: + +| Protocol | Tool result shape | +| --- | --- | +| OpenAI-compatible Chat Completions | `messages[]` with `role=tool`. | +| OpenAI Responses | `input[]` with `type=function_call_output`. | +| Anthropic Messages | `messages[].content[]` with `type=tool_result`. | + +An accepted request with an unsupported shape produces an `unsupported` +manifest and a `partial` turn commit. It never falls back to raw tool output. + +## Privacy controls + +The master `SOCAI_TELEMETRY=off` switch disables the telemetry object entirely. +Evidence content additionally follows both gates: + +```bash +SOCAI_TELEMETRY_CHAT_TEXT=off +SOCAI_TELEMETRY_EVIDENCE=off +``` + +If either content gate is off, socai uploads a manifest-only `disabled` commit +and no ToolResult body. The local exact provider request artifact remains part +of the user's run record. + +Before staging, the existing trace secret scrubber removes sensitive JSON +fields, API keys, JWTs, Bearer values, and URL query credentials including +`xsec_token`, `access_token`, and `api_key`. The proxy validates chunk hashes +and forwards content without trimming or rewriting it. + +## Proxy deployment + +The Vercel function is `site/api/evidence.js`, exposed as `/v1/evidence`. +Configure: + +```text +AXIOM_EVIDENCE_TOKEN= +AXIOM_EVIDENCE_DATASET=socai-evidence-prod +AXIOM_EVIDENCE_URL=https://api.axiom.co +AXIOM_ORG_ID= +``` + +Create `socai-evidence-dev` and `socai-evidence-prod` before enabling uploads. +Use an API token limited to ingesting the evidence dataset. Axiom's ingest +endpoint does not accept personal access tokens; see the +[official ingest documentation](https://axiom.co/docs/restapi/ingest). + +The proxy forwards validated arrays to: + +```text +POST /v1/datasets//ingest +``` + +It limits request bodies to 1 MiB, batches to 32 records, individual chunks to +32 KiB, and each install to 240 requests/32 MiB per minute. Proxy errors return +non-2xx so clients retain pending files. + +## Local proxy overrides + +For local integration work: + +```bash +SOCAI_EVIDENCE_ENDPOINT=http://localhost:3000/v1/evidence +SOCAI_TRACES_ENDPOINT=http://localhost:3000/v1/traces +``` + +The client sends records in batches below 512 KiB and sends the terminal commit +only after every preceding batch succeeds. + +## Trace summary + +The root trace span carries no evidence body. It only reports local build state: + +```text +socai.evidence.schema_version +socai.evidence.local_status +socai.evidence.upload_status_at_trace_build +socai.evidence.accepted_request_count +socai.evidence.object_count +socai.evidence.total_bytes +``` + +`upload_status_at_trace_build=queued` does not prove Axiom completeness. A +reader must find and validate the evidence dataset's `turn_commit`, all request +manifests, every referenced chunk, and all SHA-256 values. + +## Current integration boundary + +This implementation produces and reliably uploads the socai-side archive. It +does not change evaluator. Until evaluator gains an evidence repository and +integrity verifier, existing evaluations continue to use the bounded trace +packet. diff --git a/site/api/evidence.js b/site/api/evidence.js new file mode 100644 index 00000000..2ca1f833 --- /dev/null +++ b/site/api/evidence.js @@ -0,0 +1,290 @@ +import { createHash } from 'node:crypto'; + +// Lossless model-visible ToolResult proxy. The client has already applied its +// chat privacy gate and secret redactor; this endpoint validates and forwards +// records without trimming or reshaping evidence content. +const SCHEMA_VERSION = 'socai.model-visible-evidence.v1'; +const DEFAULT_AXIOM_URL = 'https://api.axiom.co'; +const DEFAULT_DATASET = 'socai-evidence-prod'; +const MAX_BODY_BYTES = 1024 * 1024; +const MAX_RECORDS = 32; +const MAX_CHUNK_BYTES = 32 * 1024; +const RATE_LIMIT_WINDOW_MS = 60_000; +const RATE_LIMIT_MAX_REQUESTS = 240; +const RATE_LIMIT_MAX_BYTES = 32 * 1024 * 1024; +const COMMON_FIELDS = [ + 'schema_version', 'record_type', 'evaluation_id', 'trace_id', 'root_span_id', + 'run_id', 'provider', 'model', 'created_at', 'source', 'install_id', + 'app_session_id', 'app_version', 'platform', +]; +const RECORD_FIELDS = { + evidence_chunk: [ + 'evidence_id', 'tool_call_id', 'tool_name', 'wire_format', 'first_observed_step', + 'message_index', 'result_index', 'content_encoding', 'content_sha256', + 'content_bytes', 'chunk_index', 'chunk_count', 'chunk_sha256', 'chunk_bytes', + 'chunk_text', 'redaction_version', 'redaction_count', 'semantic_redaction', + ], + request_manifest: [ + 'step', 'request_status', 'evidence_ids', 'evidence_count', 'manifest_sha256', 'error', + ], + turn_commit: [ + 'archive_status', 'accepted_request_count', 'evidence_object_count', + 'evidence_chunk_count', 'total_content_bytes', 'evidence_index_sha256', + 'telemetry_policy', 'committed_at', + ], +}; + +const rateLimits = new Map(); + +export default async function handler(req, res) { + setSecurityHeaders(res); + + if (req.method === 'OPTIONS') { + res.status(204).end(); + return; + } + if (req.method !== 'POST') { + res.setHeader('Allow', 'POST, OPTIONS'); + res.status(405).json({ ok: false, error: 'method_not_allowed' }); + return; + } + + let input; + let bodyBytes; + try { + ({ input, bodyBytes } = await readJsonBody(req)); + } catch (error) { + res.status(error.statusCode || 400).json({ ok: false, error: error.code || 'invalid_json' }); + return; + } + + const error = validateEnvelope(input); + if (error) { + res.status(400).json({ ok: false, error }); + return; + } + const records = input.records; + if (!consumeRateLimit(rateLimitKey(req, records), bodyBytes)) { + res.status(429).json({ ok: false, error: 'rate_limited' }); + return; + } + + try { + await forwardToAxiom(records); + res.status(202).json({ + ok: true, + accepted: records.length, + batch_sha256: sha256(JSON.stringify(records)), + }); + } catch (error) { + console.error('evidence forward failed', error instanceof Error ? error.message : 'unknown'); + res.status(502).json({ ok: false, error: 'evidence_forward_failed' }); + } +} + +async function readJsonBody(req) { + if (req.body !== undefined && req.body !== null) { + const text = typeof req.body === 'string' ? req.body : JSON.stringify(req.body); + const bodyBytes = Buffer.byteLength(text, 'utf8'); + if (bodyBytes > MAX_BODY_BYTES) { + throw requestError(413, 'body_too_large'); + } + return { input: parseJson(text), bodyBytes }; + } + + let bodyBytes = 0; + const chunks = []; + for await (const chunk of req) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + bodyBytes += buffer.byteLength; + if (bodyBytes > MAX_BODY_BYTES) { + throw requestError(413, 'body_too_large'); + } + chunks.push(buffer); + } + return { input: parseJson(Buffer.concat(chunks).toString('utf8')), bodyBytes }; +} + +function parseJson(text) { + try { + return JSON.parse(text || 'null'); + } catch { + throw requestError(400, 'invalid_json'); + } +} + +function requestError(statusCode, code) { + const error = new Error(code); + error.statusCode = statusCode; + error.code = code; + return error; +} + +function validateEnvelope(input) { + if (!input || typeof input !== 'object' || Array.isArray(input)) return 'invalid_envelope'; + if (input.schema_version !== SCHEMA_VERSION) return 'unsupported_schema'; + if (!validEvaluationId(input.evaluation_id)) return 'invalid_evaluation_id'; + if (!Array.isArray(input.records) || input.records.length === 0) return 'no_records'; + if (input.records.length > MAX_RECORDS) return 'too_many_records'; + + const installIds = new Set(input.records.map((record) => record?.install_id)); + const sources = new Set(input.records.map((record) => record?.source)); + if (installIds.size !== 1) return 'mixed_install_ids'; + if (sources.size !== 1) return 'mixed_sources'; + + for (const record of input.records) { + const error = validateRecord(record, input.evaluation_id); + if (error) return error; + } + return null; +} + +function validateRecord(record, evaluationId) { + if (!record || typeof record !== 'object' || Array.isArray(record)) return 'invalid_record'; + if (record.schema_version !== SCHEMA_VERSION) return 'record_schema_mismatch'; + if (record.evaluation_id !== evaluationId) return 'record_evaluation_mismatch'; + if (`${record.trace_id}:${record.root_span_id}` !== evaluationId) return 'record_identity_mismatch'; + if (!validScalar(record.run_id, 200)) return 'invalid_run_id'; + if (!validScalar(record.install_id, 200)) return 'invalid_install_id'; + if (!validScalar(record.source, 80)) return 'invalid_source'; + if (!validScalar(record.provider, 100)) return 'invalid_provider'; + if (!validScalar(record.model, 200)) return 'invalid_model'; + + const specificFields = RECORD_FIELDS[record.record_type]; + if (!specificFields) return 'unknown_record_type'; + const allowedFields = new Set([...COMMON_FIELDS, ...specificFields]); + if (Object.keys(record).some((field) => !allowedFields.has(field))) return 'unknown_record_field'; + + if (record.record_type === 'evidence_chunk') return validateChunk(record); + if (record.record_type === 'request_manifest') return validateManifest(record); + if (record.record_type === 'turn_commit') return validateCommit(record); + return null; +} + +function validateChunk(record) { + if (!/^ev_[a-f0-9]{64}$/.test(record.evidence_id || '')) return 'invalid_evidence_id'; + if (!validScalar(record.tool_call_id, 300)) return 'invalid_tool_call_id'; + if (record.tool_name !== null && record.tool_name !== undefined && !validScalar(record.tool_name, 200)) { + return 'invalid_tool_name'; + } + if (!/^[a-f0-9]{64}$/.test(record.content_sha256 || '')) return 'invalid_content_hash'; + if (!/^[a-f0-9]{64}$/.test(record.chunk_sha256 || '')) return 'invalid_chunk_hash'; + if (!Number.isInteger(record.chunk_index) || record.chunk_index < 0) return 'invalid_chunk_index'; + if (!Number.isInteger(record.chunk_count) || record.chunk_count < 1) return 'invalid_chunk_count'; + if (record.chunk_count > 4096) return 'chunk_count_too_large'; + if (record.chunk_index >= record.chunk_count) return 'chunk_index_out_of_range'; + if (!Number.isInteger(record.content_bytes) || record.content_bytes < 0) return 'invalid_content_bytes'; + if (!Number.isInteger(record.first_observed_step) || record.first_observed_step < 1) { + return 'invalid_first_observed_step'; + } + if (!validScalar(record.wire_format, 100)) return 'invalid_wire_format'; + if (typeof record.chunk_text !== 'string') return 'invalid_chunk_text'; + const actualBytes = Buffer.byteLength(record.chunk_text, 'utf8'); + if (actualBytes > MAX_CHUNK_BYTES) return 'chunk_too_large'; + if (record.chunk_bytes !== actualBytes) return 'chunk_byte_mismatch'; + if (sha256(record.chunk_text) !== record.chunk_sha256) return 'chunk_hash_mismatch'; + return null; +} + +function validateManifest(record) { + if (!Number.isInteger(record.step) || record.step < 1) return 'invalid_request_step'; + if (!['accepted', 'failed', 'unknown', 'unsupported'].includes(record.request_status)) { + return 'invalid_request_status'; + } + if (!Array.isArray(record.evidence_ids) || record.evidence_ids.length > 500) { + return 'invalid_manifest_evidence_ids'; + } + if (!record.evidence_ids.every((value) => /^ev_[a-f0-9]{64}$/.test(value))) { + return 'invalid_manifest_evidence_id'; + } + if (record.evidence_count !== record.evidence_ids.length) return 'manifest_count_mismatch'; + if (!/^[a-f0-9]{64}$/.test(record.manifest_sha256 || '')) return 'invalid_manifest_hash'; + return null; +} + +function validateCommit(record) { + if (!['complete', 'partial', 'disabled'].includes(record.archive_status)) { + return 'invalid_archive_status'; + } + for (const field of [ + 'accepted_request_count', + 'evidence_object_count', + 'evidence_chunk_count', + 'total_content_bytes', + ]) { + if (!Number.isInteger(record[field]) || record[field] < 0) return `invalid_${field}`; + } + if (!/^[a-f0-9]{64}$/.test(record.evidence_index_sha256 || '')) return 'invalid_index_hash'; + return null; +} + +function validEvaluationId(value) { + return /^[a-f0-9]{32}:[a-f0-9]{16}$/.test(value || ''); +} + +function validScalar(value, maxLength) { + return typeof value === 'string' && value.length > 0 && value.length <= maxLength; +} + +function sha256(text) { + return createHash('sha256').update(text, 'utf8').digest('hex'); +} + +function rateLimitKey(req, records) { + const installId = records.find((record) => validScalar(record.install_id, 200))?.install_id; + if (installId) return `install:${installId}`; + const forwardedFor = String(req.headers['x-forwarded-for'] || '').split(',')[0].trim(); + return `ip:${forwardedFor || req.socket?.remoteAddress || 'unknown'}`; +} + +function consumeRateLimit(key, bytes) { + const now = Date.now(); + for (const [existingKey, bucket] of rateLimits) { + if (now - bucket.startedAt > RATE_LIMIT_WINDOW_MS * 2) rateLimits.delete(existingKey); + } + let bucket = rateLimits.get(key); + if (!bucket || now - bucket.startedAt > RATE_LIMIT_WINDOW_MS) { + bucket = { startedAt: now, requests: 0, bytes: 0 }; + rateLimits.set(key, bucket); + } + bucket.requests += 1; + bucket.bytes += bytes; + return bucket.requests <= RATE_LIMIT_MAX_REQUESTS && bucket.bytes <= RATE_LIMIT_MAX_BYTES; +} + +async function forwardToAxiom(records) { + const token = process.env.AXIOM_EVIDENCE_TOKEN; + if (!token) throw new Error('AXIOM_EVIDENCE_TOKEN is not configured'); + + const dataset = process.env.AXIOM_EVIDENCE_DATASET || DEFAULT_DATASET; + const baseUrl = ( + process.env.AXIOM_EVIDENCE_URL || process.env.AXIOM_URL || DEFAULT_AXIOM_URL + ).replace(/\/+$/, ''); + const response = await fetch(`${baseUrl}/v1/datasets/${encodeURIComponent(dataset)}/ingest`, { + method: 'POST', + headers: axiomHeaders(token), + body: JSON.stringify(records), + }); + if (!response.ok) throw new Error(`Axiom evidence ingest failed: ${response.status}`); + + const result = await response.json().catch(() => null); + if (result && (result.failed > 0 || (Number.isInteger(result.ingested) && result.ingested !== records.length))) { + throw new Error('Axiom evidence ingest was partial'); + } +} + +function axiomHeaders(token) { + const headers = { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }; + if (process.env.AXIOM_ORG_ID) headers['X-Axiom-Org-ID'] = process.env.AXIOM_ORG_ID; + return headers; +} + +function setSecurityHeaders(res) { + res.setHeader('Access-Control-Allow-Origin', '*'); + res.setHeader('Access-Control-Allow-Methods', 'POST, OPTIONS'); + res.setHeader('Access-Control-Allow-Headers', 'Content-Type'); + res.setHeader('Cache-Control', 'no-store'); +} diff --git a/site/vercel.json b/site/vercel.json index d5005cc7..35200ddf 100644 --- a/site/vercel.json +++ b/site/vercel.json @@ -6,6 +6,9 @@ }, "api/traces.js": { "maxDuration": 10 + }, + "api/evidence.js": { + "maxDuration": 10 } }, "rewrites": [ @@ -16,6 +19,10 @@ { "source": "/v1/traces", "destination": "/api/traces" + }, + { + "source": "/v1/evidence", + "destination": "/api/evidence" } ], "redirects": [ From ed73dbbd881b5e3a3ef4416184cb59d2f8079f43 Mon Sep 17 00:00:00 2001 From: shanzhao Date: Tue, 25 Aug 2026 10:30:41 +0800 Subject: [PATCH 2/3] fix(telemetry): harden evidence delivery --- app/src-tauri/src/commands.rs | 30 +- app/src-tauri/src/telemetry.rs | 37 +- core/src/agent/run_logging.rs | 68 +++- core/src/telemetry/evidence.rs | 96 ++++-- core/src/telemetry/mod.rs | 600 ++++++++++++++++++++++++++++++--- core/src/telemetry/trace.rs | 70 +++- docs/data-model.md | 6 + docs/model-visible-evidence.md | 44 ++- site/api/evidence.js | 50 ++- site/api/evidence.test.js | 118 +++++++ site/package.json | 1 + 11 files changed, 993 insertions(+), 127 deletions(-) create mode 100644 site/api/evidence.test.js diff --git a/app/src-tauri/src/commands.rs b/app/src-tauri/src/commands.rs index 55f0202a..d70954d8 100644 --- a/app/src-tauri/src/commands.rs +++ b/app/src-tauri/src/commands.rs @@ -1134,10 +1134,14 @@ pub async fn agent_task_delete( // durable staging before their source run directory disappears. let telemetry = telemetry.inner().clone(); tauri::async_runtime::spawn(async move { - if matches!(terminal_status.as_str(), "cancelled" | "interrupted") { - if let Some(run_dir) = trace_run_dir.as_deref() { - wait_for_trace_staging(run_dir).await; - upload_terminal_run_trace(run_dir, &terminal_status, &telemetry).await; + if let Some(run_dir) = trace_run_dir.as_deref() { + wait_for_observability_staging(run_dir).await; + if !upload_terminal_run_trace(run_dir, &terminal_status, &telemetry).await { + eprintln!( + "preserving deleted task artifacts at {} because observability staging failed", + run_dir.display() + ); + return; } } let _ = tokio::task::spawn_blocking(move || { @@ -1331,7 +1335,7 @@ async fn run_agent_task_background( if let Some(run_dir) = snapshot.run_dir.as_deref() { let _ = mark_agent_run_status(run_dir, "failed", Some(&error)); let _ = mark_run_trace_status(run_dir, "failed"); - telemetry.upload_run_trace(run_dir); + telemetry.upload_run_trace(run_dir).await; } record_desktop_session(&snapshot, &format!("[task failed: {error}]"), "failed"); telemetry.capture( @@ -1367,7 +1371,7 @@ pub(crate) fn record_interrupted_run(snapshot: &AgentTaskSnapshot, message: &str } /// Wait for an aborted agent future's trace drop guard, patch the precise -/// terminal state, then stage the trace durably before the task can be deleted. +/// terminal state, then stage trace and evidence durably before deletion. /// AbortHandle::abort() only schedules cancellation; without this wait the /// telemetry worker can race `trace.json` creation and silently miss the run. pub(crate) async fn upload_terminal_run_trace( @@ -1379,17 +1383,21 @@ pub(crate) async fn upload_terminal_run_trace( const READY_DELAY: std::time::Duration = std::time::Duration::from_millis(25); let run_dir = run_dir.as_ref(); - let staged_marker = run_dir.join(".trace-staged"); + let staged_marker = run_dir.join(".observability-staged"); if staged_marker.is_file() { return true; } for attempt in 0..READY_RETRIES { match mark_run_trace_status(run_dir, status) { Ok(()) => { - if telemetry.upload_run_trace(run_dir) { + if telemetry.upload_run_trace(run_dir).await { let _ = std::fs::write(staged_marker, b""); return true; } + if attempt + 1 < READY_RETRIES { + tokio::time::sleep(READY_DELAY).await; + continue; + } return false; } Err(error) @@ -1403,11 +1411,11 @@ pub(crate) async fn upload_terminal_run_trace( false } -async fn wait_for_trace_staging(run_dir: &std::path::Path) { +async fn wait_for_observability_staging(run_dir: &std::path::Path) { const WAIT_RETRIES: usize = 50; const WAIT_DELAY: std::time::Duration = std::time::Duration::from_millis(25); - let marker = run_dir.join(".trace-staged"); + let marker = run_dir.join(".observability-staged"); for attempt in 0..WAIT_RETRIES { if marker.is_file() { return; @@ -1561,7 +1569,7 @@ async fn run_agent_task_on_shared_page( // and uploads the trace — instead of reporting a completed task. anyhow::bail!(error); } - telemetry.upload_run_trace(&outcome.run_dir); + telemetry.upload_run_trace(&outcome.run_dir).await; let usage = outcome.usage; let estimated_cost = usage.cost.as_ref().map(|cost| cost.total); diff --git a/app/src-tauri/src/telemetry.rs b/app/src-tauri/src/telemetry.rs index fd5b5eb7..f8b3f493 100644 --- a/app/src-tauri/src/telemetry.rs +++ b/app/src-tauri/src/telemetry.rs @@ -35,11 +35,21 @@ impl DesktopTelemetry { /// Upload a run's `trace.json` to the traces proxy. No-op when telemetry /// is off or the file is missing. - pub(crate) fn upload_run_trace(&self, run_dir: impl AsRef) -> bool { - if let Some(telemetry) = &self.0 { - return telemetry.upload_run_trace(run_dir.as_ref()); + pub(crate) async fn upload_run_trace(&self, run_dir: impl AsRef) -> bool { + let run_dir = run_dir.as_ref().to_path_buf(); + let staged = match self.0.clone() { + Some(telemetry) => tokio::task::spawn_blocking({ + let run_dir = run_dir.clone(); + move || telemetry.upload_run_trace(&run_dir) + }) + .await + .unwrap_or(false), + None => true, + }; + if staged { + let _ = std::fs::write(run_dir.join(".observability-staged"), b""); } - false + staged } } @@ -63,3 +73,22 @@ pub(crate) fn duration_ms(started_at: Option, finished_at: Option) -> _ => None, } } + +#[cfg(test)] +mod observability_staging_tests { + use super::DesktopTelemetry; + + #[tokio::test] + async fn telemetry_opt_out_still_allows_task_artifact_deletion() { + let run_dir = std::env::temp_dir().join(format!( + "socai-observability-opt-out-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&run_dir); + std::fs::create_dir_all(&run_dir).expect("create run directory"); + let telemetry = DesktopTelemetry(None); + assert!(telemetry.upload_run_trace(&run_dir).await); + assert!(run_dir.join(".observability-staged").is_file()); + std::fs::remove_dir_all(run_dir).expect("remove run directory"); + } +} diff --git a/core/src/agent/run_logging.rs b/core/src/agent/run_logging.rs index 7142346e..8c9e87b1 100644 --- a/core/src/agent/run_logging.rs +++ b/core/src/agent/run_logging.rs @@ -47,11 +47,15 @@ fn safe_component(value: &str, fallback: &str) -> String { } } -fn write_json_atomic(path: &Path, value: &Value) -> std::io::Result<()> { +pub(crate) fn write_json_atomic(path: &Path, value: &Value) -> std::io::Result<()> { + let bytes = serde_json::to_vec_pretty(value).map_err(std::io::Error::other)?; + write_bytes_atomic(path, &bytes) +} + +pub(crate) fn write_bytes_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } - let bytes = serde_json::to_vec_pretty(value).map_err(std::io::Error::other)?; let name = path .file_name() .and_then(|value| value.to_str()) @@ -65,6 +69,16 @@ fn write_json_atomic(path: &Path, value: &Value) -> std::io::Result<()> { Ok(()) } +fn request_wire_format(provider: &str, payload: &Value) -> &'static str { + if payload.get("input").and_then(Value::as_array).is_some() { + "openai_responses" + } else if provider.eq_ignore_ascii_case("anthropic") { + "anthropic_messages" + } else { + "openai_chat" + } +} + fn env_runs_root(value: Option) -> Option { let value = value?; let value = value.to_string_lossy().trim().to_string(); @@ -130,6 +144,7 @@ pub struct AgentRunRecorder { run_dir: PathBuf, manifest_path: PathBuf, manifest: Mutex, + provider: String, started: Instant, finalized: AtomicBool, } @@ -162,6 +177,7 @@ impl AgentRunRecorder { run_dir, manifest_path, manifest: Mutex::new(manifest), + provider: provider.to_string(), started: Instant::now(), finalized: AtomicBool::new(false), }) @@ -171,6 +187,16 @@ impl AgentRunRecorder { write_json_atomic( &self.run_dir.join(format!("llm/{step:03}.request.json")), payload, + )?; + write_json_atomic( + &self + .run_dir + .join(format!("llm/{step:03}.request.meta.json")), + &json!({ + "schema_version": 1, + "wire_format": request_wire_format(&self.provider, payload), + "status": "prepared", + }), ) } @@ -187,6 +213,7 @@ impl AgentRunRecorder { &self.run_dir.join(format!("llm/{step:03}.response.json")), &value, )?; + self.update_llm_request_status(step, "accepted")?; // Keep running usage/step totals in run.json so the desktop app can // show them live mid-run; `finish` overwrites with the final figures. { @@ -214,7 +241,20 @@ impl AgentRunRecorder { "completed_at": timestamp(), "error": error, }), - ) + )?; + self.update_llm_request_status(step, "failed") + } + + fn update_llm_request_status(&self, step: u32, status: &str) -> std::io::Result<()> { + let path = self + .run_dir + .join(format!("llm/{step:03}.request.meta.json")); + let mut metadata = std::fs::read(&path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .unwrap_or_else(|| json!({ "schema_version": 1 })); + metadata["status"] = json!(status); + write_json_atomic(&path, &metadata) } pub fn start_tool_call( @@ -389,6 +429,28 @@ impl Drop for ToolCallRecorder { } } +#[cfg(test)] +mod request_metadata_tests { + use super::request_wire_format; + use serde_json::json; + + #[test] + fn records_provider_wire_format_next_to_the_exact_request() { + assert_eq!( + request_wire_format("openai", &json!({ "input": [] })), + "openai_responses" + ); + assert_eq!( + request_wire_format("anthropic", &json!({ "messages": [] })), + "anthropic_messages" + ); + assert_eq!( + request_wire_format("deepseek", &json!({ "messages": [] })), + "openai_chat" + ); + } +} + /// Best-effort terminal update when an entrypoint cancels the agent future. pub fn mark_agent_run_status( run_dir: impl AsRef, diff --git a/core/src/telemetry/evidence.rs b/core/src/telemetry/evidence.rs index db8a207a..debd31a4 100644 --- a/core/src/telemetry/evidence.rs +++ b/core/src/telemetry/evidence.rs @@ -10,10 +10,10 @@ use std::path::{Path, PathBuf}; use chrono::Utc; use serde_json::{json, Map, Value}; use sha2::{Digest, Sha256}; -use uuid::Uuid; use super::trace::redact_secrets_in_value; use crate::agent::llm::LLMResponse; +use crate::agent::run_logging::write_json_atomic; pub(crate) const SCHEMA_VERSION: &str = "socai.model-visible-evidence.v1"; const ARCHIVE_RELATIVE_PATH: &str = "evidence/model-visible-v1.json"; @@ -78,6 +78,7 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { let mut records = Vec::new(); let mut index_entries = Vec::new(); let mut seen_evidence = HashSet::new(); + let mut model_view_cache: HashMap = HashMap::new(); let mut accepted_request_count = 0usize; let mut evidence_object_count = 0usize; let mut evidence_chunk_count = 0usize; @@ -91,18 +92,30 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { } for (step, request_path) in requests { let response_path = run_dir.join("llm").join(format!("{step:03}.response.json")); - let request_status = response_status(&response_path); + let (request_status, wire_format) = request_observation(run_dir, step, &response_path); let mut evidence_ids = Vec::new(); let mut extraction_error = None; if request_status == "accepted" { accepted_request_count += 1; match read_json(&request_path).and_then(|payload| { - extract_model_visible_tool_results(&metadata.provider, &payload) + extract_model_visible_tool_results( + &metadata.provider, + wire_format.as_deref(), + &payload, + ) }) { Ok(results) => { for result in results { let mut content = result.content; + if let Some((cached_content, cached_id)) = + model_view_cache.get(&result.tool_call_id) + { + if cached_content == &content { + evidence_ids.push(cached_id.clone()); + continue; + } + } let original = content.clone(); redact_secrets_in_value(&mut content); let redaction_count = changed_leaf_count(&original, &content); @@ -112,6 +125,10 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { let evidence_id = evidence_id(&evaluation_id, &result.tool_call_id, &content_sha256); evidence_ids.push(evidence_id.clone()); + model_view_cache.insert( + result.tool_call_id.clone(), + (original, evidence_id.clone()), + ); if !seen_evidence.insert(evidence_id.clone()) { continue; @@ -171,10 +188,15 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { } dedupe_preserving_order(&mut evidence_ids); + let manifest_status = if extraction_error.is_some() { + "unsupported" + } else { + request_status.as_str() + }; let manifest_body = json!({ "step": step, - "request_status": if extraction_error.is_some() { "unsupported" } else { request_status }, - "evidence_ids": evidence_ids, + "request_status": manifest_status, + "evidence_ids": evidence_ids.clone(), }); let manifest_sha256 = sha256_json(&manifest_body)?; index_entries.push(json!({ @@ -189,9 +211,9 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { json!({ "record_type": "request_manifest", "step": step, - "request_status": if extraction_error.is_some() { "unsupported" } else { request_status }, - "evidence_ids": manifest_body["evidence_ids"].clone(), - "evidence_count": manifest_body["evidence_ids"].as_array().map_or(0, Vec::len), + "request_status": manifest_status, + "evidence_count": evidence_ids.len(), + "evidence_ids": evidence_ids, "manifest_sha256": manifest_sha256, "error": extraction_error, }), @@ -242,8 +264,20 @@ fn build_archive(run_dir: &Path, content_enabled: bool) -> io::Result { fn extract_model_visible_tool_results( provider: &str, + wire_format: Option<&str>, payload: &Value, ) -> io::Result> { + match wire_format { + Some("openai_responses") => return extract_openai_responses(payload), + Some("anthropic_messages") => return extract_anthropic_messages(payload), + Some("openai_chat") => return extract_openai_chat(payload), + Some(other) => { + return Err(io::Error::other(format!( + "unsupported recorded provider wire format: {other}" + ))) + } + None => {} + } if payload.get("input").and_then(Value::as_array).is_some() { return extract_openai_responses(payload); } @@ -442,6 +476,29 @@ fn response_status(path: &Path) -> &'static str { } } +fn request_observation( + run_dir: &Path, + step: u32, + response_path: &Path, +) -> (String, Option) { + let metadata_path = run_dir + .join("llm") + .join(format!("{step:03}.request.meta.json")); + if let Ok(metadata) = read_json(&metadata_path) { + let status = match metadata.get("status").and_then(Value::as_str) { + Some("accepted") => "accepted", + Some("failed") => "failed", + _ => "unknown", + }; + let wire_format = metadata + .get("wire_format") + .and_then(Value::as_str) + .map(str::to_string); + return (status.to_string(), wire_format); + } + (response_status(response_path).to_string(), None) +} + fn trace_identity(trace: &Value) -> Option { let root = trace .pointer("/resourceSpans/0/scopeSpans/0/spans") @@ -490,6 +547,8 @@ fn common_record( record.insert("provider".into(), json!(metadata.provider)); record.insert("model".into(), json!(metadata.model)); record.insert("created_at".into(), json!(created_at)); + record.insert("client_version".into(), json!(env!("CARGO_PKG_VERSION"))); + record.insert("platform".into(), json!(std::env::consts::OS)); Value::Object(record) } @@ -564,19 +623,6 @@ fn read_json(path: &Path) -> io::Result { serde_json::from_slice(&bytes).map_err(io::Error::other) } -fn write_json_atomic(path: &Path, value: &Value) -> io::Result<()> { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; - } - let temporary = path.with_extension(format!("{}.tmp", Uuid::new_v4())); - let bytes = serde_json::to_vec(value).map_err(io::Error::other)?; - fs::write(&temporary, bytes)?; - if path.exists() { - let _ = fs::remove_file(path); - } - fs::rename(temporary, path) -} - fn changed_leaf_count(before: &Value, after: &Value) -> usize { match (before, after) { (Value::Array(left), Value::Array(right)) => left @@ -597,6 +643,10 @@ fn changed_leaf_count(before: &Value, after: &Value) -> usize { } fn split_utf8(text: &str, max_bytes: usize) -> Vec { + assert!( + max_bytes >= 4, + "UTF-8 chunks require a budget of at least 4 bytes" + ); if text.is_empty() { return vec![String::new()]; } @@ -607,9 +657,7 @@ fn split_utf8(text: &str, max_bytes: usize) -> Vec { while end > start && !text.is_char_boundary(end) { end -= 1; } - if end == start { - end = text.len(); - } + debug_assert!(end > start); chunks.push(text[start..end].to_string()); start = end; } diff --git a/core/src/telemetry/mod.rs b/core/src/telemetry/mod.rs index 12043535..7b9adceb 100644 --- a/core/src/telemetry/mod.rs +++ b/core/src/telemetry/mod.rs @@ -3,6 +3,7 @@ use std::sync::OnceLock; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use serde_json::{json, Map, Value}; +use sha2::{Digest, Sha256}; use tokio::io::AsyncWriteExt; use tokio::sync::mpsc; use tokio::time::MissedTickBehavior; @@ -13,6 +14,8 @@ pub mod trace; pub use trace::redact_secrets; +use crate::agent::run_logging::{write_bytes_atomic, write_json_atomic}; + const EVENT_SCHEMA_VERSION: u32 = 1; const TELEMETRY_ENDPOINT: &str = "https://socai.io/v1/events"; const TRACES_ENDPOINT: &str = "https://socai.io/v1/traces"; @@ -26,6 +29,10 @@ const TRACE_RETRY_BATCH_SIZE: usize = 5; const TRACE_FILE_READY_RETRIES: usize = 20; const TRACE_FILE_READY_DELAY: Duration = Duration::from_millis(50); const EVIDENCE_BATCH_MAX_BYTES: usize = 512 * 1024; +const EVIDENCE_BATCH_MAX_RECORDS: usize = 32; +const EVIDENCE_UPLOAD_STATE_VERSION: u32 = 1; +const EVIDENCE_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60); +const EVIDENCE_CONFIG_RETRY_MAX_DELAY: Duration = Duration::from_secs(6 * 60 * 60); /// Which socai surface is emitting telemetry. Carried verbatim into the `source` /// field of every event and used to decide which device context is meaningful @@ -138,7 +145,7 @@ impl Telemetry { /// cancellation can call this just before the trace drop guard finishes; /// that source path is retried briefly by the worker. pub fn upload_run_trace(&self, run_dir: &Path) -> bool { - let (evidence_item, _) = match evidence::stage_run_archive( + let (evidence_item, evidence_staged) = match evidence::stage_run_archive( run_dir, &self.pending_evidence_dir, evidence_text_enabled(), @@ -154,7 +161,7 @@ impl Telemetry { Err(_) => (QueuedItem::TraceFile(source), false), }; let _ = self.sender.try_send(item); - staged + staged && evidence_staged } } @@ -254,12 +261,7 @@ fn stage_trace_file(source: &Path, pending_dir: &Path) -> std::io::Result = Vec::new(); + loop { let Ok(Some(entry)) = entries.next_entry().await else { break; }; let path = entry.path(); if path.extension().and_then(|ext| ext.to_str()) == Some("json") { - paths.push(path); + let next_attempt = upload_state_next_attempt(&path).unwrap_or_default(); + if next_attempt > now { + continue; + } + let modified = entry + .metadata() + .await + .ok() + .and_then(|metadata| metadata.modified().ok()) + .and_then(|time| time.duration_since(UNIX_EPOCH).ok()) + .map_or(0, |duration| duration.as_millis() as u64); + paths.push((next_attempt, modified, path)); } } - for path in paths { + paths.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1))); + for (_, _, path) in paths.into_iter().take(TRACE_RETRY_BATCH_SIZE) { upload_pending_evidence(client, config, &path).await; } } @@ -359,19 +374,38 @@ async fn upload_pending_trace(client: &reqwest::Client, config: &WorkerConfig, p } } -/// Upload content and request manifests first, then the terminal commit. The -/// spool file is retained unless every batch receives a successful proxy ack. +#[derive(Debug)] +enum EvidencePostResult { + Accepted, + Retryable { + status: Option, + error_code: String, + retry_after: Option, + }, + Permanent { + status: Option, + error_code: String, + }, +} + +/// Upload content and request manifests first, then the terminal commit. Batch +/// progress is checkpointed after every ack; permanent payload errors move the +/// archive to `dead/`, while transient failures retain it with a retry time. async fn upload_pending_evidence(client: &reqwest::Client, config: &WorkerConfig, path: &Path) { let Ok(bytes) = tokio::fs::read(path).await else { return; }; + let archive_sha256 = sha256_hex(&bytes); let Ok(payload) = serde_json::from_slice::(&bytes) else { + quarantine_pending_evidence(path, None, "invalid_local_json", None); return; }; let Some(evaluation_id) = payload.get("evaluation_id").and_then(Value::as_str) else { + quarantine_pending_evidence(path, None, "missing_evaluation_id", None); return; }; let Some(records) = payload.get("records").and_then(Value::as_array) else { + quarantine_pending_evidence(path, Some(evaluation_id), "missing_records", None); return; }; @@ -387,66 +421,206 @@ async fn upload_pending_evidence(client: &reqwest::Client, config: &WorkerConfig } } if commits.len() != 1 { + quarantine_pending_evidence(path, Some(evaluation_id), "invalid_commit_count", None); return; } - if !upload_evidence_batches(client, evaluation_id, body_records).await { + let Ok(mut batches) = build_evidence_batches(evaluation_id, body_records) else { + quarantine_pending_evidence(path, Some(evaluation_id), "unbatchable_archive", None); + return; + }; + let Ok(commit_batches) = build_evidence_batches(evaluation_id, commits) else { + quarantine_pending_evidence(path, Some(evaluation_id), "unbatchable_commit", None); + return; + }; + if commit_batches.len() != 1 { + quarantine_pending_evidence(path, Some(evaluation_id), "invalid_commit_batch", None); return; } - if !upload_evidence_batches(client, evaluation_id, commits).await { + batches.extend(commit_batches); + + let mut state = load_evidence_upload_state(path, &archive_sha256); + let mut next_batch = state + .get("next_batch_index") + .and_then(Value::as_u64) + .unwrap_or_default() as usize; + if next_batch > batches.len() { + quarantine_pending_evidence(path, Some(evaluation_id), "invalid_batch_checkpoint", None); return; } + + while next_batch < batches.len() { + let batch = batches[next_batch].clone(); + match post_evidence_batch(client, evaluation_id, next_batch, batches.len(), batch).await { + EvidencePostResult::Accepted => { + next_batch += 1; + state["next_batch_index"] = json!(next_batch); + state["attempt_count"] = json!(0); + state["next_attempt_at_ms"] = json!(0); + state["last_http_status"] = Value::Null; + state["last_error_code"] = Value::Null; + state["last_attempt_at_ms"] = json!(now_ms()); + if write_json_atomic(&evidence_state_path(path), &state).is_err() { + return; + } + } + EvidencePostResult::Retryable { + status, + error_code, + retry_after, + } => { + let attempts = state + .get("attempt_count") + .and_then(Value::as_u64) + .unwrap_or_default() + .saturating_add(1); + let delay = retry_after.unwrap_or_else(|| evidence_retry_delay(attempts, status)); + state["attempt_count"] = json!(attempts); + state["next_attempt_at_ms"] = + json!(now_ms().saturating_add(delay.as_millis() as u64)); + state["last_http_status"] = status.map_or(Value::Null, |value| json!(value)); + state["last_error_code"] = json!(error_code); + state["last_attempt_at_ms"] = json!(now_ms()); + let _ = write_json_atomic(&evidence_state_path(path), &state); + return; + } + EvidencePostResult::Permanent { status, error_code } => { + quarantine_pending_evidence(path, Some(evaluation_id), &error_code, status); + report_evidence_quarantined(client, config, evaluation_id, &error_code, status) + .await; + return; + } + } + } let _ = tokio::fs::remove_file(path).await; + let _ = tokio::fs::remove_file(evidence_state_path(path)).await; } -async fn upload_evidence_batches( - client: &reqwest::Client, +fn build_evidence_batches( evaluation_id: &str, records: Vec, -) -> bool { - let mut batch = Vec::new(); +) -> std::io::Result>> { + let mut batches = Vec::new(); + let mut batch: Vec = Vec::new(); + let envelope_base = evidence_envelope_len(evaluation_id, &[])?; + let mut batch_bytes = envelope_base; for record in records { - let mut candidate = batch.clone(); - candidate.push(record.clone()); - let candidate_body = json!({ - "schema_version": evidence::SCHEMA_VERSION, - "evaluation_id": evaluation_id, - "records": candidate, - }); - let candidate_len = - serde_json::to_vec(&candidate_body).map_or(usize::MAX, |body| body.len()); - let should_flush = candidate_len > EVIDENCE_BATCH_MAX_BYTES && !batch.is_empty(); - if should_flush - && !post_evidence_batch(client, evaluation_id, std::mem::take(&mut batch)).await - { - return false; + let record_bytes = serde_json::to_vec(&record) + .map_err(std::io::Error::other)? + .len(); + let separator_bytes = usize::from(!batch.is_empty()); + let should_flush = batch.len() >= EVIDENCE_BATCH_MAX_RECORDS + || batch_bytes + separator_bytes + record_bytes > EVIDENCE_BATCH_MAX_BYTES; + if should_flush && !batch.is_empty() { + batches.push(std::mem::take(&mut batch)); + batch_bytes = envelope_base; + } + if batch_bytes + record_bytes > EVIDENCE_BATCH_MAX_BYTES { + return Err(std::io::Error::other( + "one evidence record exceeds the client batch byte limit", + )); + } + if !batch.is_empty() { + batch_bytes += 1; } + batch_bytes += record_bytes; batch.push(record); } - batch.is_empty() || post_evidence_batch(client, evaluation_id, batch).await + if !batch.is_empty() { + batches.push(batch); + } + Ok(batches) +} + +fn evidence_envelope_len(evaluation_id: &str, records: &[Value]) -> std::io::Result { + serde_json::to_vec(&json!({ + "schema_version": evidence::SCHEMA_VERSION, + "evaluation_id": evaluation_id, + "records": records, + })) + .map(|body| body.len()) + .map_err(std::io::Error::other) } async fn post_evidence_batch( client: &reqwest::Client, evaluation_id: &str, + batch_index: usize, + batch_count: usize, records: Vec, -) -> bool { +) -> EvidencePostResult { let expected = records.len() as u64; + let batch_sha256 = match serde_json::to_vec(&records) { + Ok(bytes) => sha256_hex(&bytes), + Err(_) => { + return EvidencePostResult::Permanent { + status: None, + error_code: "batch_serialization_failed".to_string(), + } + } + }; let body = json!({ "schema_version": evidence::SCHEMA_VERSION, "evaluation_id": evaluation_id, + "batch_index": batch_index, + "batch_count": batch_count, + "batch_sha256": batch_sha256, "records": records, }); let Ok(response) = client.post(evidence_endpoint()).json(&body).send().await else { - return false; + return EvidencePostResult::Retryable { + status: None, + error_code: "network_error".to_string(), + retry_after: None, + }; }; - if !response.status().is_success() { - return false; + let status = response.status(); + let retry_after = response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .map(Duration::from_secs); + if !status.is_success() { + let error_body = response.json::().await.unwrap_or(Value::Null); + let error_code = error_body + .get("error") + .and_then(Value::as_str) + .unwrap_or("http_error") + .chars() + .take(120) + .collect::(); + let status_code = status.as_u16(); + if matches!(status_code, 400 | 405 | 413 | 422) { + return EvidencePostResult::Permanent { + status: Some(status_code), + error_code, + }; + } + return EvidencePostResult::Retryable { + status: Some(status_code), + error_code, + retry_after, + }; } let Ok(ack) = response.json::().await else { - return false; + return EvidencePostResult::Retryable { + status: Some(status.as_u16()), + error_code: "invalid_proxy_ack".to_string(), + retry_after: None, + }; }; - ack.get("ok").and_then(Value::as_bool) == Some(true) + let accepted = ack.get("ok").and_then(Value::as_bool) == Some(true) && ack.get("accepted").and_then(Value::as_u64) == Some(expected) + && ack.get("batch_sha256").and_then(Value::as_str) == Some(batch_sha256.as_str()); + if accepted { + EvidencePostResult::Accepted + } else { + EvidencePostResult::Retryable { + status: Some(status.as_u16()), + error_code: "proxy_ack_mismatch".to_string(), + retry_after: None, + } + } } fn enrich_evidence_record(record: &mut Value, config: &WorkerConfig) { @@ -455,9 +629,148 @@ fn enrich_evidence_record(record: &mut Value, config: &WorkerConfig) { }; object.insert("source".into(), json!(config.source.as_str())); object.insert("install_id".into(), json!(config.install_id)); - object.insert("app_session_id".into(), json!(config.session_id)); - object.insert("app_version".into(), json!(env!("CARGO_PKG_VERSION"))); - object.insert("platform".into(), json!(std::env::consts::OS)); +} + +fn evidence_state_path(path: &Path) -> PathBuf { + path.with_extension("state") +} + +fn default_evidence_upload_state(archive_sha256: &str) -> Value { + json!({ + "schema_version": EVIDENCE_UPLOAD_STATE_VERSION, + "archive_sha256": archive_sha256, + "next_batch_index": 0, + "attempt_count": 0, + "next_attempt_at_ms": 0, + "last_http_status": null, + "last_error_code": null, + "last_attempt_at_ms": null, + }) +} + +fn load_evidence_upload_state(path: &Path, archive_sha256: &str) -> Value { + let state_path = evidence_state_path(path); + let Ok(bytes) = std::fs::read(state_path) else { + return default_evidence_upload_state(archive_sha256); + }; + let Ok(state) = serde_json::from_slice::(&bytes) else { + return default_evidence_upload_state(archive_sha256); + }; + if state.get("schema_version").and_then(Value::as_u64) + != Some(EVIDENCE_UPLOAD_STATE_VERSION as u64) + || state.get("archive_sha256").and_then(Value::as_str) != Some(archive_sha256) + { + return default_evidence_upload_state(archive_sha256); + } + state +} + +fn upload_state_next_attempt(path: &Path) -> Option { + let bytes = std::fs::read(evidence_state_path(path)).ok()?; + let state = serde_json::from_slice::(&bytes).ok()?; + state.get("next_attempt_at_ms").and_then(Value::as_u64) +} + +fn evidence_retry_delay(attempts: u64, status: Option) -> Duration { + let configuration_error = matches!(status, Some(401 | 403 | 404)); + let base_secs: u64 = if configuration_error { 5 * 60 } else { 30 }; + let max_delay = if configuration_error { + EVIDENCE_CONFIG_RETRY_MAX_DELAY + } else { + EVIDENCE_RETRY_MAX_DELAY + }; + let exponent = attempts.saturating_sub(1).min(10) as u32; + let seconds = base_secs.saturating_mul(1u64 << exponent); + let jitter_ms = (uuid::Uuid::new_v4().as_u128() % 1000) as u64; + Duration::from_secs(seconds.min(max_delay.as_secs())) + Duration::from_millis(jitter_ms) +} + +fn quarantine_pending_evidence( + path: &Path, + evaluation_id: Option<&str>, + error_code: &str, + status: Option, +) { + let Some(parent) = path.parent() else { + return; + }; + let dead_dir = parent.join("dead"); + if std::fs::create_dir_all(&dead_dir).is_err() { + return; + } + let name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("evidence.json"); + let mut destination = dead_dir.join(name); + if destination.exists() { + destination = dead_dir.join(format!( + "{}.{}.json", + name.trim_end_matches(".json"), + now_ms() + )); + } + if std::fs::rename(path, &destination).is_err() { + return; + } + let old_state = evidence_state_path(path); + let _ = std::fs::remove_file(&old_state); + let reason_path = destination.with_extension("reason.json"); + let _ = write_json_atomic( + &reason_path, + &json!({ + "evaluation_id": evaluation_id, + "error_code": error_code.chars().take(120).collect::(), + "http_status": status, + "quarantined_at_ms": now_ms(), + }), + ); + eprintln!( + "quarantined evidence spool {} after permanent error {}", + destination.display(), + error_code + ); +} + +async fn report_evidence_quarantined( + client: &reqwest::Client, + config: &WorkerConfig, + evaluation_id: &str, + error_code: &str, + status: Option, +) { + if cfg!(test) { + return; + } + let properties = enrich_properties( + json!({ + "evaluation_id": evaluation_id, + "error_code": error_code.chars().take(120).collect::(), + "http_status": status, + }), + config, + now_ms(), + ); + let event = remote_event( + "socai_evidence_upload_quarantined", + &config.install_id, + &properties, + ); + let _ = client + .post(TELEMETRY_ENDPOINT) + .json(&json!({ "events": [event] })) + .send() + .await; +} + +fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let mut output = String::with_capacity(digest.len() * 2); + for byte in digest { + use std::fmt::Write; + let _ = write!(output, "{byte:02x}"); + } + output } /// The on-disk trace stays identity-free so run dirs can be shared; the @@ -754,8 +1067,10 @@ fn env_value_is(name: &str, values: &[&str]) -> bool { #[cfg(test)] mod tests { use super::*; + use std::collections::VecDeque; use std::ffi::OsString; use std::sync::{Mutex, OnceLock}; + use tokio::io::AsyncReadExt; fn env_lock() -> &'static Mutex<()> { static LOCK: OnceLock> = OnceLock::new(); @@ -869,4 +1184,201 @@ mod tests { assert_eq!(object.get("install_id"), Some(&json!("install-1"))); assert!(!object.contains_key("created_at_ms")); } + + #[test] + fn evidence_batches_honor_record_and_byte_limits() { + for (count, expected_batches) in [(31, 1), (32, 1), (33, 2), (64, 2), (65, 3)] { + let records = (0..count) + .map(|index| json!({ "record_type": "request_manifest", "step": index + 1 })) + .collect(); + let batches = build_evidence_batches("trace:span", records) + .expect("small records should be batchable"); + assert_eq!(batches.len(), expected_batches, "record count {count}"); + assert!(batches + .iter() + .all(|batch| batch.len() <= EVIDENCE_BATCH_MAX_RECORDS)); + assert!(batches.iter().all(|batch| { + evidence_envelope_len("trace:span", batch) + .is_ok_and(|size| size <= EVIDENCE_BATCH_MAX_BYTES) + })); + } + } + + #[test] + fn evidence_batches_flush_before_the_byte_limit() { + let text = "x".repeat(300_000); + let batches = build_evidence_batches( + "trace:span", + vec![json!({"chunk_text": text}), json!({"chunk_text": text})], + ) + .expect("each individual record is below the byte limit"); + assert_eq!(batches.len(), 2); + assert!(batches.iter().all(|batch| batch.len() == 1)); + } + + #[test] + fn evidence_batch_hash_matches_javascript_json_stringify_order() { + let records = vec![json!({"b": 2, "a": 1})]; + let bytes = serde_json::to_vec(&records).expect("serialize batch fixture"); + assert_eq!( + String::from_utf8(bytes.clone()).unwrap(), + r#"[{"a":1,"b":2}]"# + ); + assert_eq!( + sha256_hex(&bytes), + "44c7deead2ed8313d29655e45c0d1469419213c93d9f44d66da7c7afe46e74e3" + ); + } + + async fn mock_evidence_server( + statuses: Vec, + ) -> (String, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock evidence server"); + let address = listener.local_addr().expect("mock server address"); + let handle = tokio::spawn(async move { + let mut statuses = VecDeque::from(statuses); + let mut requests = Vec::new(); + while let Some(status) = statuses.pop_front() { + let (mut socket, _) = listener.accept().await.expect("accept request"); + let mut bytes = Vec::new(); + let header_end = loop { + let mut chunk = [0u8; 4096]; + let read = socket.read(&mut chunk).await.expect("read request"); + assert!(read > 0, "request ended before headers"); + bytes.extend_from_slice(&chunk[..read]); + if let Some(index) = bytes.windows(4).position(|value| value == b"\r\n\r\n") { + break index + 4; + } + }; + let headers = String::from_utf8_lossy(&bytes[..header_end]).to_ascii_lowercase(); + let content_length = headers + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .and_then(|value| value.trim().parse::().ok()) + .expect("request content length"); + while bytes.len() < header_end + content_length { + let mut chunk = [0u8; 4096]; + let read = socket.read(&mut chunk).await.expect("read request body"); + assert!(read > 0, "request ended before body"); + bytes.extend_from_slice(&chunk[..read]); + } + let body: Value = + serde_json::from_slice(&bytes[header_end..header_end + content_length]) + .expect("parse request body"); + let accepted = body + .get("records") + .and_then(Value::as_array) + .map_or(0, Vec::len); + let batch_hash = body + .get("batch_sha256") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + requests.push(body); + let response_body = if status == 202 { + json!({ "ok": true, "accepted": accepted, "batch_sha256": batch_hash }) + } else { + json!({ "ok": false, "error": "mock_failure" }) + } + .to_string(); + let reason = if status == 202 { "Accepted" } else { "Error" }; + let response = format!( + "HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response_body}", + response_body.len() + ); + socket + .write_all(response.as_bytes()) + .await + .expect("write response"); + } + requests + }); + (format!("http://{address}/v1/evidence"), handle) + } + + fn evidence_test_config(root: &Path) -> WorkerConfig { + WorkerConfig { + install_id: "install-test".to_string(), + session_id: "session-test".to_string(), + source: TelemetrySource::Desktop, + local_path: root.join("events.jsonl"), + pending_trace_dir: root.join("pending-traces"), + pending_evidence_dir: root.join("pending-evidence"), + } + } + + fn evidence_test_archive(path: &Path, record_count: usize) { + let mut records: Vec = (0..record_count) + .map(|step| json!({ "record_type": "request_manifest", "step": step + 1 })) + .collect(); + records.push(json!({ "record_type": "turn_commit" })); + write_json_atomic( + path, + &json!({ + "evaluation_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:bbbbbbbbbbbbbbbb", + "records": records, + }), + ) + .expect("write pending evidence fixture"); + } + + #[tokio::test(flavor = "current_thread")] + async fn evidence_upload_resumes_after_a_mid_file_failure_and_quarantines_400() { + let _lock = env_lock().lock().expect("env lock is not poisoned"); + let temp = tempfile::tempdir().expect("temporary evidence directory"); + let config = evidence_test_config(temp.path()); + std::fs::create_dir_all(&config.pending_evidence_dir).expect("pending directory"); + let path = config.pending_evidence_dir.join("resume.json"); + evidence_test_archive(&path, 65); + let client = reqwest::Client::new(); + + let (endpoint, first_server) = mock_evidence_server(vec![202, 502]).await; + let _guard = EnvGuard::set("SOCAI_EVIDENCE_ENDPOINT", Some(&endpoint)); + upload_pending_evidence(&client, &config, &path).await; + let first_requests = first_server.await.expect("first mock server"); + assert_eq!( + first_requests + .iter() + .map(|body| body["batch_index"].as_u64().unwrap_or_default()) + .collect::>(), + vec![0, 1] + ); + let state: Value = serde_json::from_slice( + &std::fs::read(evidence_state_path(&path)).expect("upload state"), + ) + .expect("parse upload state"); + assert_eq!(state["next_batch_index"], 1); + + let (endpoint, second_server) = mock_evidence_server(vec![202, 202, 202]).await; + std::env::set_var("SOCAI_EVIDENCE_ENDPOINT", endpoint); + upload_pending_evidence(&client, &config, &path).await; + let second_requests = second_server.await.expect("second mock server"); + assert_eq!( + second_requests + .iter() + .map(|body| body["batch_index"].as_u64().unwrap_or_default()) + .collect::>(), + vec![1, 2, 3] + ); + assert!(!path.exists()); + assert!(!evidence_state_path(&path).exists()); + + let poison = config.pending_evidence_dir.join("poison.json"); + evidence_test_archive(&poison, 1); + let (endpoint, poison_server) = mock_evidence_server(vec![400]).await; + std::env::set_var("SOCAI_EVIDENCE_ENDPOINT", endpoint); + upload_pending_evidence(&client, &config, &poison).await; + let _ = poison_server.await.expect("poison mock server"); + assert!(!poison.exists()); + assert!(config + .pending_evidence_dir + .join("dead/poison.json") + .exists()); + assert!(config + .pending_evidence_dir + .join("dead/poison.reason.json") + .exists()); + } } diff --git a/core/src/telemetry/trace.rs b/core/src/telemetry/trace.rs index 9f65aef4..01b5f7b2 100644 --- a/core/src/telemetry/trace.rs +++ b/core/src/telemetry/trace.rs @@ -43,6 +43,7 @@ use crate::agent::llm::{ Block, LLMResponse, Message, MessageContent, MessageRole, TokenUsage, ToolResultContent, }; use crate::agent::r#loop::THINKING_TEXT_PREFIX; +use crate::agent::run_logging::write_bytes_atomic; use crate::agent::signature::md5_hex; /// Safety net for pathological runs; a default run (30 steps) stays far below. @@ -424,7 +425,7 @@ impl RunTraceBuilder { fn write_trace_file(&self, payload: &Value) { if let Ok(bytes) = serde_json::to_vec(payload) { - let _ = std::fs::write(self.run_dir.join("trace.json"), bytes); + let _ = write_bytes_atomic(&self.run_dir.join("trace.json"), &bytes); } } } @@ -471,10 +472,8 @@ pub fn mark_run_trace_status(run_dir: impl AsRef, status: &str) -> std::io attribute["value"] = json!({ "stringValue": status }); } } - std::fs::write( - &path, - serde_json::to_vec(&payload).map_err(std::io::Error::other)?, - ) + let bytes = serde_json::to_vec(&payload).map_err(std::io::Error::other)?; + write_bytes_atomic(&path, &bytes) } /// Flatten a summarizer map (`summarize_tool_args` / `summarize_tool_result`) @@ -969,17 +968,19 @@ fn redact_token_runs(text: &str) -> String { /// Scrub credentials embedded in URLs. These commonly appear in model-visible /// browser tool results as query parameters rather than structured JSON keys. fn redact_secret_query_values(text: &str) -> String { - const PARAMETERS: [&str; 3] = ["xsec_token=", "access_token=", "api_key="]; let lower = text.to_ascii_lowercase(); let mut out = String::with_capacity(text.len()); let mut cursor = 0; while cursor < text.len() { - let match_at = PARAMETERS + let match_at = SECRET_JSON_FIELDS .iter() - .filter_map(|parameter| { + .filter_map(|field| { + let parameter = format!("{field}="); lower[cursor..] - .find(parameter) - .map(|offset| (cursor + offset, *parameter)) + .match_indices(¶meter) + .map(|(offset, _)| cursor + offset) + .find(|&start| query_parameter_boundary(text.as_bytes(), start)) + .map(|start| (start, parameter)) }) .min_by_key(|(index, _)| *index); let Some((start, parameter)) = match_at else { @@ -990,12 +991,7 @@ fn redact_secret_query_values(text: &str) -> String { let value_start = start + parameter.len(); let value_len = text.as_bytes()[value_start..] .iter() - .take_while(|byte| { - !matches!( - **byte, - b'&' | b'#' | b' ' | b'\t' | b'\r' | b'\n' | b'"' | b'\'' - ) - }) + .take_while(|&&byte| secret_query_value_byte(byte)) .count(); if value_len == 0 { cursor = value_start; @@ -1007,6 +1003,22 @@ fn redact_secret_query_values(text: &str) -> String { out } +fn query_parameter_boundary(bytes: &[u8], start: usize) -> bool { + start == 0 + || matches!( + bytes[start - 1], + b'?' | b'&' | b';' | b',' | b' ' | b'\t' | b'\r' | b'\n' | b'"' | b'\'' | b'\\' + ) +} + +fn secret_query_value_byte(byte: u8) -> bool { + byte.is_ascii_alphanumeric() + || matches!( + byte, + b'.' | b'_' | b'~' | b'+' | b'/' | b'-' | b'=' | b'%' | b':' | b'@' + ) +} + fn token_run_len(bytes: &[u8], accept: impl Fn(u8) -> bool) -> usize { bytes.iter().take_while(|&&b| accept(b)).count() } @@ -1146,3 +1158,29 @@ fn truncate_chars(text: &str, max_chars: usize) -> String { let truncated: String = text.chars().take(max_chars).collect(); format!("{truncated}…") } + +#[cfg(test)] +mod evidence_redaction_tests { + use super::redact_secrets; + + #[test] + fn query_secret_redaction_preserves_adjacent_fields() { + assert_eq!( + redact_secrets("access_token=abc,expires_in=3600,scope=read"), + "access_token=[redacted],expires_in=3600,scope=read" + ); + assert_eq!( + redact_secrets(r#"xsec_token=abc\",\"next\":1"#), + r#"xsec_token=[redacted]\",\"next\":1"# + ); + } + + #[test] + fn query_secret_redaction_uses_the_structured_secret_field_set() { + let input = "client_secret=one&refresh_token=two&password=three&safe=four"; + assert_eq!( + redact_secrets(input), + "client_secret=[redacted]&refresh_token=[redacted]&password=[redacted]&safe=four" + ); + } +} diff --git a/docs/data-model.md b/docs/data-model.md index 74898b2f..784bd114 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -38,6 +38,7 @@ The session root is `SOCAI_SESSIONS_DIR`, then `SOCAI_HOME/sessions`, then ├── report.md ├── llm/ │ ├── 001.request.json +│ ├── 001.request.meta.json │ ├── 001.response.json │ └── ... └── tools/ @@ -62,6 +63,11 @@ preparation and provider-specific translation, excluding authentication headers. Its shape therefore follows the active API (Anthropic Messages, OpenAI-compatible Chat Completions, or Responses). +The adjacent `llm/NNN.request.meta.json` contains no prompt or tool content. It +records the request artifact schema, provider wire format, and explicit +`prepared` / `accepted` / `failed` outcome so post-run consumers do not have to +infer those load-bearing facts from a future request/response shape. + The matching response contains text, exposed reasoning, tool calls, stop reason, normalized `usage`, the provider's original `usage` object, request duration, and completion time. Normalized usage separates ordinary input, diff --git a/docs/model-visible-evidence.md b/docs/model-visible-evidence.md index 857c61dd..f364bb92 100644 --- a/docs/model-visible-evidence.md +++ b/docs/model-visible-evidence.md @@ -40,11 +40,17 @@ Uploads are staged before the telemetry worker returns: ```text /telemetry/pending-evidence/-.json +/telemetry/pending-evidence/-.state +/telemetry/pending-evidence/dead/ ``` -Pending files survive proxy or Axiom outages. The worker retries up to five -files every 30 seconds and removes a file only after all content/manifests and -the final commit receive successful proxy acknowledgements. +Pending files survive proxy or Axiom outages. The `.state` sidecar checkpoints +the next batch and retry time. The worker retries eligible oldest files with +exponential backoff, honors numeric `Retry-After`, and removes an archive only +after all content/manifests and the final commit receive verified proxy +acknowledgements. Deterministically invalid local payloads and proxy 400/413/422 +responses move to `dead/` with a content-free reason file instead of retrying +forever or being deleted. ## Schema @@ -70,7 +76,9 @@ If later context management changes a ToolResult value, its content hash and evidence ID change, preserving the model-visible version for each request. The client uploads all `evidence_chunk` and `request_manifest` records before -the single `turn_commit`. Duplicate records can occur after a lost HTTP +the single `turn_commit`, batching at no more than 32 records and 512 KiB. Each +batch has a stable index and SHA-256; the client verifies the proxy echo and +checkpoints every accepted batch. Duplicate records can still occur after a lost HTTP acknowledgement; readers must deduplicate by `(evidence_id, chunk_index)` and reject conflicting values for the same stable key. @@ -85,8 +93,11 @@ for generic `content` fields: | OpenAI Responses | `input[]` with `type=function_call_output`. | | Anthropic Messages | `messages[].content[]` with `type=tool_result`. | -An accepted request with an unsupported shape produces an `unsupported` -manifest and a `partial` turn commit. It never falls back to raw tool output. +New runs record wire format and request outcome explicitly in +`llm/NNN.request.meta.json`. Legacy runs without the sidecar use the shape +detector and response parser as a compatibility fallback. An accepted request +with an unsupported shape produces an `unsupported` manifest and a `partial` +turn commit. It never falls back to raw tool output. ## Privacy controls @@ -104,8 +115,11 @@ of the user's run record. Before staging, the existing trace secret scrubber removes sensitive JSON fields, API keys, JWTs, Bearer values, and URL query credentials including -`xsec_token`, `access_token`, and `api_key`. The proxy validates chunk hashes -and forwards content without trimming or rewriting it. +all sensitive JSON field names, including `xsec_token`, `access_token`, +`refresh_token`, `client_secret`, `password`, and `api_key`. Query-value +scanning accepts token-safe characters only, so commas, JSON escapes, brackets, +and neighboring fields remain intact. The proxy validates chunk and batch +hashes and forwards content without trimming or rewriting it. ## Proxy deployment @@ -130,9 +144,12 @@ The proxy forwards validated arrays to: POST /v1/datasets//ingest ``` -It limits request bodies to 1 MiB, batches to 32 records, individual chunks to -32 KiB, and each install to 240 requests/32 MiB per minute. Proxy errors return -non-2xx so clients retain pending files. +It limits request bodies to 1 MiB, batches to 32 records, and individual chunks +to 32 KiB. It checks wire `Content-Length` before expensive validation and uses +both IP and install buckets. Because Vercel's in-memory buckets are +instance-local and install IDs are client-asserted, production must also enable +Vercel Firewall/WAF rate limits. Proxy errors return non-2xx so clients retain, +back off, or quarantine pending files according to the status class. ## Local proxy overrides @@ -143,8 +160,9 @@ SOCAI_EVIDENCE_ENDPOINT=http://localhost:3000/v1/evidence SOCAI_TRACES_ENDPOINT=http://localhost:3000/v1/traces ``` -The client sends records in batches below 512 KiB and sends the terminal commit -only after every preceding batch succeeds. +The client sends at most 32 records per batch below 512 KiB and sends the +terminal commit only after every preceding batch succeeds. A restart resumes +from the checkpointed batch rather than batch zero. ## Trace summary diff --git a/site/api/evidence.js b/site/api/evidence.js index 2ca1f833..8aad8b0f 100644 --- a/site/api/evidence.js +++ b/site/api/evidence.js @@ -15,7 +15,7 @@ const RATE_LIMIT_MAX_BYTES = 32 * 1024 * 1024; const COMMON_FIELDS = [ 'schema_version', 'record_type', 'evaluation_id', 'trace_id', 'root_span_id', 'run_id', 'provider', 'model', 'created_at', 'source', 'install_id', - 'app_session_id', 'app_version', 'platform', + 'app_session_id', 'app_version', 'client_version', 'platform', ]; const RECORD_FIELDS = { evidence_chunk: [ @@ -58,13 +58,20 @@ export default async function handler(req, res) { return; } + if (!consumeRateLimit(`ip:${clientIp(req)}`, bodyBytes)) { + res.setHeader('Retry-After', '60'); + res.status(429).json({ ok: false, error: 'rate_limited' }); + return; + } + const error = validateEnvelope(input); if (error) { res.status(400).json({ ok: false, error }); return; } const records = input.records; - if (!consumeRateLimit(rateLimitKey(req, records), bodyBytes)) { + if (!consumeRateLimit(`install:${records[0].install_id}`, bodyBytes)) { + res.setHeader('Retry-After', '60'); res.status(429).json({ ok: false, error: 'rate_limited' }); return; } @@ -74,7 +81,7 @@ export default async function handler(req, res) { res.status(202).json({ ok: true, accepted: records.length, - batch_sha256: sha256(JSON.stringify(records)), + batch_sha256: input.batch_sha256, }); } catch (error) { console.error('evidence forward failed', error instanceof Error ? error.message : 'unknown'); @@ -84,12 +91,14 @@ export default async function handler(req, res) { async function readJsonBody(req) { if (req.body !== undefined && req.body !== null) { - const text = typeof req.body === 'string' ? req.body : JSON.stringify(req.body); - const bodyBytes = Buffer.byteLength(text, 'utf8'); + const contentLength = contentLengthBytes(req); + const text = typeof req.body === 'string' ? req.body : null; + const input = text === null ? req.body : parseJson(text); + const bodyBytes = contentLength ?? Buffer.byteLength(text ?? JSON.stringify(input), 'utf8'); if (bodyBytes > MAX_BODY_BYTES) { throw requestError(413, 'body_too_large'); } - return { input: parseJson(text), bodyBytes }; + return { input, bodyBytes }; } let bodyBytes = 0; @@ -126,6 +135,11 @@ function validateEnvelope(input) { if (!validEvaluationId(input.evaluation_id)) return 'invalid_evaluation_id'; if (!Array.isArray(input.records) || input.records.length === 0) return 'no_records'; if (input.records.length > MAX_RECORDS) return 'too_many_records'; + if (!Number.isInteger(input.batch_index) || input.batch_index < 0) return 'invalid_batch_index'; + if (!Number.isInteger(input.batch_count) || input.batch_count < 1) return 'invalid_batch_count'; + if (input.batch_index >= input.batch_count) return 'batch_index_out_of_range'; + if (!/^[a-f0-9]{64}$/.test(input.batch_sha256 || '')) return 'invalid_batch_hash'; + if (sha256(JSON.stringify(input.records)) !== input.batch_sha256) return 'batch_hash_mismatch'; const installIds = new Set(input.records.map((record) => record?.install_id)); const sources = new Set(input.records.map((record) => record?.source)); @@ -191,7 +205,7 @@ function validateManifest(record) { if (!['accepted', 'failed', 'unknown', 'unsupported'].includes(record.request_status)) { return 'invalid_request_status'; } - if (!Array.isArray(record.evidence_ids) || record.evidence_ids.length > 500) { + if (!Array.isArray(record.evidence_ids)) { return 'invalid_manifest_evidence_ids'; } if (!record.evidence_ids.every((value) => /^ev_[a-f0-9]{64}$/.test(value))) { @@ -230,11 +244,16 @@ function sha256(text) { return createHash('sha256').update(text, 'utf8').digest('hex'); } -function rateLimitKey(req, records) { - const installId = records.find((record) => validScalar(record.install_id, 200))?.install_id; - if (installId) return `install:${installId}`; +function clientIp(req) { const forwardedFor = String(req.headers['x-forwarded-for'] || '').split(',')[0].trim(); - return `ip:${forwardedFor || req.socket?.remoteAddress || 'unknown'}`; + return forwardedFor || req.socket?.remoteAddress || 'unknown'; +} + +function contentLengthBytes(req) { + const raw = req.headers?.['content-length']; + if (raw === undefined) return null; + const value = Number.parseInt(String(raw), 10); + return Number.isSafeInteger(value) && value >= 0 ? value : null; } function consumeRateLimit(key, bytes) { @@ -268,7 +287,14 @@ async function forwardToAxiom(records) { if (!response.ok) throw new Error(`Axiom evidence ingest failed: ${response.status}`); const result = await response.json().catch(() => null); - if (result && (result.failed > 0 || (Number.isInteger(result.ingested) && result.ingested !== records.length))) { + const verified = result + && Number.isInteger(result.ingested) + && result.ingested === records.length + && Number.isInteger(result.failed) + && result.failed === 0 + && (result.failures === undefined || result.failures === null + || (Array.isArray(result.failures) && result.failures.length === 0)); + if (!verified) { throw new Error('Axiom evidence ingest was partial'); } } diff --git a/site/api/evidence.test.js b/site/api/evidence.test.js new file mode 100644 index 00000000..7546a5b0 --- /dev/null +++ b/site/api/evidence.test.js @@ -0,0 +1,118 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import test from 'node:test'; + +import handler from './evidence.js'; + +const SCHEMA_VERSION = 'socai.model-visible-evidence.v1'; +let sequence = 0; + +function sha256(value) { + return createHash('sha256').update(value, 'utf8').digest('hex'); +} + +function manifestRecord(step, installId) { + return { + schema_version: SCHEMA_VERSION, + record_type: 'request_manifest', + evaluation_id: `${'a'.repeat(32)}:${'b'.repeat(16)}`, + trace_id: 'a'.repeat(32), + root_span_id: 'b'.repeat(16), + run_id: 'run-test', + provider: 'deepseek', + model: 'deepseek-v4-pro', + created_at: '2026-08-25T00:00:00Z', + source: 'desktop', + install_id: installId, + client_version: '0.5.4', + platform: 'macos', + step, + request_status: 'accepted', + evidence_ids: [], + evidence_count: 0, + manifest_sha256: 'c'.repeat(64), + error: null, + }; +} + +function envelope(count) { + sequence += 1; + const installId = `evidence-test-${sequence}`; + const records = Array.from({ length: count }, (_, index) => manifestRecord(index + 1, installId)); + return { + schema_version: SCHEMA_VERSION, + evaluation_id: records[0].evaluation_id, + batch_index: 0, + batch_count: 1, + batch_sha256: sha256(JSON.stringify(records)), + records, + }; +} + +function response() { + return { + statusCode: 0, + body: null, + headers: {}, + setHeader(key, value) { this.headers[key] = value; }, + status(code) { this.statusCode = code; return this; }, + json(body) { this.body = body; return this; }, + end() { return this; }, + }; +} + +async function invoke(body, headers = {}) { + const res = response(); + await handler({ + method: 'POST', + body, + headers: { 'x-forwarded-for': `127.0.0.${sequence + 1}`, ...headers }, + socket: {}, + }, res); + return res; +} + +test('accepts exactly 32 records and echoes the verified batch hash', { concurrency: false }, async () => { + const body = envelope(32); + globalThis.fetch = async (_url, options) => { + const records = JSON.parse(options.body); + return { ok: true, json: async () => ({ ingested: records.length, failed: 0, failures: [] }) }; + }; + process.env.AXIOM_EVIDENCE_TOKEN = 'test-only'; + const res = await invoke(body); + assert.equal(res.statusCode, 202); + assert.equal(res.body.accepted, 32); + assert.equal(res.body.batch_sha256, body.batch_sha256); +}); + +test('rejects 33 records even when the serialized body is small', { concurrency: false }, async () => { + const body = envelope(33); + const res = await invoke(body); + assert.equal(res.statusCode, 400); + assert.equal(res.body.error, 'too_many_records'); +}); + +test('rejects a mismatched client batch hash', { concurrency: false }, async () => { + const body = envelope(1); + body.batch_sha256 = 'd'.repeat(64); + const res = await invoke(body); + assert.equal(res.statusCode, 400); + assert.equal(res.body.error, 'batch_hash_mismatch'); +}); + +test('does not acknowledge an unverifiable Axiom 2xx response', { concurrency: false }, async () => { + const body = envelope(1); + globalThis.fetch = async () => ({ ok: true, json: async () => { throw new Error('truncated'); } }); + const originalError = console.error; + console.error = () => {}; + const res = await invoke(body).finally(() => { console.error = originalError; }); + assert.equal(res.statusCode, 502); + assert.equal(res.body.error, 'evidence_forward_failed'); +}); + +test('uses wire Content-Length when Vercel pre-parses the body', { concurrency: false }, async () => { + const body = envelope(1); + const res = await invoke(body, { 'content-length': String(1024 * 1024 + 1) }); + assert.equal(res.statusCode, 413); + assert.equal(res.body.error, 'body_too_large'); +}); diff --git a/site/package.json b/site/package.json index d7847bed..52e0480b 100644 --- a/site/package.json +++ b/site/package.json @@ -7,6 +7,7 @@ "scripts": { "dev": "astro dev", "build": "astro build", + "test:evidence": "node --test api/evidence.test.js", "preview": "astro preview" }, "devDependencies": { From a4ee536db700e6b290ee11dc8dad452cbee35b4f Mon Sep 17 00:00:00 2001 From: shanzhao Date: Tue, 25 Aug 2026 10:40:42 +0800 Subject: [PATCH 3/3] fix(telemetry): avoid repeated terminal staging --- app/src-tauri/src/commands.rs | 4 ---- site/api/evidence.js | 4 +++- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/app/src-tauri/src/commands.rs b/app/src-tauri/src/commands.rs index d70954d8..6efc0ab1 100644 --- a/app/src-tauri/src/commands.rs +++ b/app/src-tauri/src/commands.rs @@ -1394,10 +1394,6 @@ pub(crate) async fn upload_terminal_run_trace( let _ = std::fs::write(staged_marker, b""); return true; } - if attempt + 1 < READY_RETRIES { - tokio::time::sleep(READY_DELAY).await; - continue; - } return false; } Err(error) diff --git a/site/api/evidence.js b/site/api/evidence.js index 8aad8b0f..233fc5d3 100644 --- a/site/api/evidence.js +++ b/site/api/evidence.js @@ -252,7 +252,9 @@ function clientIp(req) { function contentLengthBytes(req) { const raw = req.headers?.['content-length']; if (raw === undefined) return null; - const value = Number.parseInt(String(raw), 10); + const text = String(raw).trim(); + if (!/^\d+$/.test(text)) return null; + const value = Number.parseInt(text, 10); return Number.isSafeInteger(value) && value >= 0 ? value : null; }