use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::sync::{Mutex as StdMutex, OnceLock};
use std::time::{Duration, Instant};
use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine as _;
use serde::Deserialize;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use uuid::Uuid;
use car_feedback_core::bundle::{
collect, extract_dedup_from_named_logs, valid_manifest_id, BundleItem, CollectInputs,
NamedLogInput, RedactedBundle,
};
use car_feedback_core::spool::{
EnqueueOutcome, IdentityLane, Spool, SpoolEntryId, SpoolEntrySummary, SpoolState,
ThrottleVerdict,
};
use car_parslee::feedback_transport::ServerReportRow;
use crate::handler::JsonRpcMessage;
use crate::session::ServerState;
pub(crate) const FEEDBACK_OUTBOX_DIR: &str = "feedback-outbox";
pub(crate) const DAEMON_STDERR_TEE_FILE: &str = "car-server.stderr.log";
pub(crate) const SCREENSHOT_MAX_BYTES: usize = 5 * 1024 * 1024;
pub(crate) const SCREENSHOT_B64_MAX_LEN: usize = SCREENSHOT_MAX_BYTES * 4 / 3 + 4;
pub(crate) const DESCRIPTION_MAX_CHARS: usize =
car_parslee::feedback_transport::DESCRIPTION_MAX_CHARS;
const TITLE_MAX_CHARS: usize = 80;
pub(crate) const PREVIEW_CACHE_CAP: usize = 5;
pub(crate) const PREVIEW_TTL: Duration = Duration::from_secs(10 * 60);
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct ComposeParams {
description: String,
#[serde(default)]
include_screenshot: bool,
#[serde(default)]
include_diagnostics: bool,
#[serde(default)]
screenshot_b64: Option<String>,
#[serde(default)]
host_version: Option<String>,
#[serde(default)]
macos_version: Option<String>,
}
impl ComposeParams {
fn validate(&self) -> Result<(), String> {
let chars = self.description.trim().chars().count();
if chars > DESCRIPTION_MAX_CHARS {
return Err(format!(
"invalid params: description is {chars} characters after trimming; the \
maximum is {DESCRIPTION_MAX_CHARS} — shorten it before submitting"
));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
enum LaneParam {
Authenticated,
Anonymous,
}
#[derive(Debug, Clone, Deserialize)]
struct SubmitParams {
#[serde(flatten)]
compose: ComposeParams,
lane: LaneParam,
#[serde(default)]
org_id: Option<String>,
#[serde(default)]
preview_id: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Default)]
struct StatusParams {
#[serde(default)]
submission_id: Option<String>,
#[serde(default)]
limit: Option<usize>,
}
#[derive(Debug, Clone, Deserialize, Default)]
struct ListParams {
#[serde(default)]
limit: Option<usize>,
}
const FEEDBACK_LIST_DEFAULT_LIMIT: usize = 100;
const FEEDBACK_LIST_MAX_LIMIT: usize = 200;
pub(crate) async fn handle_compose_preview(
req: &JsonRpcMessage,
state: &ServerState,
session: &crate::session::ClientSession,
) -> Result<Value, String> {
let params: ComposeParams = parse_params(&req.params)?;
params.validate()?;
let home = car_home_dir(state)?;
let daemon_ctx = if params.include_diagnostics {
Some(daemon_runtime_context(state).await)
} else {
None
};
let session_id = session.client_id.clone();
let diagnostics = state.feedback_diagnostics.clone();
let (bundle, preview_id) = run_blocking(move || -> Result<_, String> {
let bundle = compose_bundle(¶ms, &home, &diagnostics, daemon_ctx)?;
let preview_id = store_preview(&session_id, &home, ¶ms, &bundle);
Ok((bundle, preview_id))
})
.await??;
let bundle_value =
serde_json::to_value(&bundle).map_err(|e| format!("serialize bundle: {e}"))?;
let manifest = bundle_value.get("manifest").cloned().unwrap_or(Value::Null);
Ok(json!({
"bundle": bundle_value,
"manifest": manifest,
"preview_id": preview_id,
}))
}
pub(crate) async fn handle_submit(
req: &JsonRpcMessage,
state: &ServerState,
session: &crate::session::ClientSession,
) -> Result<Value, String> {
let params: SubmitParams = parse_params(&req.params)?;
params.compose.validate()?;
let active_org = state
.parslee_session
.get()
.and_then(|session| session.identity.active_organization.as_deref())
.filter(|org| !org.trim().is_empty());
let lane = resolve_lane(params.lane, params.org_id.as_deref(), active_org)?;
let home = car_home_dir(state)?;
let daemon_ctx = if params.preview_id.is_none() && params.compose.include_diagnostics {
Some(daemon_runtime_context(state).await)
} else {
None
};
let session_id = session.client_id.clone();
let diagnostics = state.feedback_diagnostics.clone();
run_blocking(move || -> Result<Value, String> {
let source = match params.preview_id.as_deref() {
Some(id) => match take_preview(id, &session_id, &home, ¶ms.compose) {
Some(entry) => SubmitBundle::Previewed(entry),
None => return Err(preview_expired_error()),
},
None => SubmitBundle::Fresh(compose_bundle(
¶ms.compose,
&home,
&diagnostics,
daemon_ctx,
)?),
};
let title = title_from_description(&source.bundle().description);
let spool = match open_spool(&home) {
Ok(spool) => spool,
Err(e) => {
source.release();
return Err(e);
}
};
let id = match spool.enqueue_if_allowed(source.bundle(), lane, &title) {
Ok(EnqueueOutcome::Enqueued(id)) => id,
Ok(EnqueueOutcome::Throttled(verdict)) => {
source.release();
return Err(throttled_error(verdict));
}
Err(e) => {
source.release();
return Err(format!("feedback spool enqueue failed: {e}"));
}
};
crate::feedback_drain::wake_feedback_drain();
submit_result(&id, spool.list())
})
.await?
}
fn submit_result(
id: &SpoolEntryId,
read_back: std::io::Result<Vec<SpoolEntrySummary>>,
) -> Result<Value, String> {
let summary = match read_back {
Ok(rows) => rows.into_iter().find(|s| &s.id == id),
Err(e) => {
tracing::warn!(
target: "car::feedback",
entry = %id, error = %e,
"feedback spool read-back failed after a durable enqueue; \
reporting the enqueued entry from local facts"
);
None
}
};
let (client_submission_id, state) = match summary {
Some(s) => (Some(s.client_submission_id), s.state),
None => (client_submission_id_from_entry_id(id), SpoolState::Queued),
};
Ok(json!({
"submission_id": id.as_str(),
"client_submission_id": client_submission_id,
"state": state_value(&state)?,
"source": "local",
}))
}
fn client_submission_id_from_entry_id(id: &SpoolEntryId) -> Option<String> {
let (_, suffix) = id.as_str().split_once('-')?;
Uuid::parse_str(suffix).ok()?;
Some(suffix.to_string())
}
pub(crate) async fn handle_status(
req: &JsonRpcMessage,
state: &ServerState,
) -> Result<Value, String> {
let params: StatusParams = parse_params(&req.params)?;
let home = car_home_dir(state)?;
run_blocking(move || -> Result<Value, String> {
let spool = open_spool(&home)?;
let entries = if let Some(wanted) = ¶ms.submission_id {
let rows = spool
.list()
.map_err(|e| format!("feedback spool list failed: {e}"))?;
let row = rows
.into_iter()
.find(|r| r.id.as_str() == wanted)
.ok_or_else(|| format!("unknown submission_id: {wanted}"))?;
let description = match spool.load_bundle(&row.id) {
Ok(bundle) => Some(bundle.description),
Err(error) => {
tracing::warn!(
target: "car::feedback",
entry = %row.id,
error = %error,
"feedback bundle is unreadable; returning its durable status without a description"
);
None
}
};
let mut value = summary_value(&row)?;
if let Some(map) = value.as_object_mut() {
map.insert(
"description".to_string(),
description.map(Value::String).unwrap_or(Value::Null),
);
}
vec![value]
} else {
summaries(&spool)?
};
let (entries, has_more) = if params.submission_id.is_some() {
(entries, false)
} else {
limit_rows(entries, params.limit)
};
Ok(json!({
"entries": entries,
"has_more": has_more,
"staleness": staleness_value(&spool)?,
}))
})
.await?
}
pub(crate) async fn handle_list(
req: &JsonRpcMessage,
state: &ServerState,
) -> Result<Value, String> {
let params: ListParams = parse_params(&req.params)?;
let home = car_home_dir(state)?;
let server_rows = fetch_server_rows(state).await;
run_blocking(move || -> Result<Value, String> {
let spool = open_spool(&home)?;
let mut entries = summaries(&spool)?;
if let Some(rows) = server_rows {
entries = merge_server_rows(entries, rows);
}
let (entries, has_more) = limit_rows(entries, params.limit);
Ok(json!({
"entries": entries,
"has_more": has_more,
"staleness": staleness_value(&spool)?,
}))
})
.await?
}
#[derive(Debug, Clone, Copy)]
enum FeedbackDiagnosticsProbe {
Production,
Isolated,
}
#[derive(Debug, Clone)]
pub(crate) enum FeedbackDiagnostics {
Production,
Isolated {
models_dir: PathBuf,
huggingface_hub_root: PathBuf,
},
}
impl FeedbackDiagnostics {
pub(crate) fn production() -> Self {
Self::Production
}
pub(crate) fn isolated(models_dir: PathBuf, huggingface_hub_root: PathBuf) -> Self {
Self::Isolated {
models_dir,
huggingface_hub_root,
}
}
}
pub(crate) fn compose_bundle(
params: &ComposeParams,
car_home: &Path,
diagnostics: &FeedbackDiagnostics,
daemon_ctx: Option<Value>,
) -> Result<RedactedBundle, String> {
match diagnostics {
FeedbackDiagnostics::Production => {
let models_dir = car_inference::default_models_dir();
compose_bundle_at(
params,
car_home,
&models_dir,
None,
FeedbackDiagnosticsProbe::Production,
daemon_ctx,
)
}
FeedbackDiagnostics::Isolated {
models_dir,
huggingface_hub_root,
} => compose_bundle_at(
params,
car_home,
models_dir,
Some(huggingface_hub_root),
FeedbackDiagnosticsProbe::Isolated,
daemon_ctx,
),
}
}
fn compose_bundle_at(
params: &ComposeParams,
car_home: &Path,
models_dir: &Path,
isolated_huggingface_hub: Option<&Path>,
probe: FeedbackDiagnosticsProbe,
daemon_ctx: Option<Value>,
) -> Result<RedactedBundle, String> {
let mut inputs = CollectInputs {
description: params.description.clone(),
state_root: Some(car_home.to_path_buf()),
..CollectInputs::default()
};
if params.host_version.is_some() || params.macos_version.is_some() {
inputs.host_version_fields = Some(json!({
"host_version": params.host_version,
"macos_version": params.macos_version,
}));
}
let mut post_notes: Vec<BundleItem> = Vec::new();
if params.include_diagnostics {
let (log_tails, manifest_notes) = named_log_set(car_home);
inputs.dedup = extract_dedup_from_named_logs(&log_tails, &car_home.join("logs"));
inputs.log_tails = log_tails;
for note in manifest_notes {
post_notes.push(omitted_item("log:supervised-agents", note));
}
let opts = car_inference::doctor::DoctorOptions::default();
let report = match probe {
FeedbackDiagnosticsProbe::Isolated => car_inference::doctor::diagnose_at_isolated(
car_home,
models_dir,
isolated_huggingface_hub
.expect("isolated feedback diagnostics always carry a Hugging Face root"),
&opts,
),
FeedbackDiagnosticsProbe::Production => {
car_inference::doctor::diagnose_at(car_home, models_dir, &opts)
}
};
match serde_json::to_value(&report) {
Ok(v) => inputs.doctor_report = Some(v),
Err(e) => post_notes.push(omitted_item(
"doctor_report",
format!("unserializable: {e}"),
)),
}
inputs.runtime_context = Some(daemon_ctx.unwrap_or_else(|| {
json!({
"daemon_version": env!("CARGO_PKG_VERSION"),
"protocol_version": car_proto::PROTOCOL_VERSION,
})
}));
} else {
post_notes.push(omitted_item("diagnostics", "excluded by user toggle"));
}
let mut bundle = collect(inputs).map_err(|e| e.to_string())?;
if params.include_screenshot {
match validate_screenshot(params.screenshot_b64.as_deref()) {
Ok((b64, byte_len)) => {
let entry = json!({ "jpeg_b64": b64, "byte_len": byte_len });
match &mut bundle.runtime_context {
Some(Value::Object(map)) => {
map.insert("screenshot".to_string(), entry);
}
other => *other = Some(json!({ "screenshot": entry })),
}
bundle.manifest.items.push(BundleItem {
name: "screenshot".to_string(),
bytes: b64.len() as u64,
included: true,
truncated: false,
moved_to_overflow: false,
note: None,
});
bundle.manifest.total_inline_bytes += b64.len() as u64;
}
Err(note) => post_notes.push(omitted_item("screenshot", note)),
}
}
bundle.manifest.items.extend(post_notes);
Ok(bundle)
}
fn validate_screenshot(b64: Option<&str>) -> Result<(&str, usize), String> {
let Some(b64) = b64 else {
return Err("requested but no screenshot_b64 supplied".to_string());
};
if b64.len() > SCREENSHOT_B64_MAX_LEN {
return Err(format!(
"exceeds the {SCREENSHOT_MAX_BYTES}-byte cap ({} bytes of base64 text, over the \
{SCREENSHOT_B64_MAX_LEN}-byte encoded ceiling; not decoded) — screenshot omitted",
b64.len()
));
}
let decoded = BASE64
.decode(b64.as_bytes())
.map_err(|e| format!("invalid base64 ({e}) — screenshot omitted"))?;
if decoded.len() > SCREENSHOT_MAX_BYTES {
return Err(format!(
"exceeds the {SCREENSHOT_MAX_BYTES}-byte cap ({} bytes) — screenshot omitted",
decoded.len()
));
}
let has_soi = decoded.len() >= 4 && decoded.starts_with(&[0xFF, 0xD8, 0xFF]);
let has_eoi = decoded.len() >= 4 && decoded.ends_with(&[0xFF, 0xD9]);
if !has_soi || !has_eoi {
return Err("not a JPEG (missing SOI/EOI magic) — screenshot omitted".to_string());
}
Ok((b64, decoded.len()))
}
fn resolve_lane(
lane: LaneParam,
claimed_org_id: Option<&str>,
active_org_id: Option<&str>,
) -> Result<IdentityLane, String> {
match lane {
LaneParam::Anonymous => Ok(IdentityLane::Anonymous),
LaneParam::Authenticated => {
let active = active_org_id.ok_or_else(|| {
"lane \"authenticated\" requires a signed-in Parslee session with an active \
organization — submit signed-out reports with lane \"anonymous\""
.to_string()
})?;
if let Some(claimed) = claimed_org_id {
if claimed != active {
return Err(
"invalid params: claimed 'org_id' does not match the signed-in session's \
active organization"
.to_string(),
);
}
}
car_parslee::feedback_transport::validate_org_id(active).map_err(|reason| {
format!(
"signed-in session has an invalid active organization id ({reason}) — \
choose a valid Parslee organization before submitting"
)
})?;
Ok(IdentityLane::Authenticated {
org_id: active.to_string(),
})
}
}
}
fn parse_params<T: for<'de> Deserialize<'de>>(params: &Value) -> Result<T, String> {
let value = if params.is_null() {
json!({})
} else {
params.clone()
};
serde_json::from_value(value).map_err(|e| format!("invalid params: {e}"))
}
pub(crate) fn car_home_dir(state: &ServerState) -> Result<PathBuf, String> {
state
.journal_dir
.parent()
.map(PathBuf::from)
.filter(|p| !p.as_os_str().is_empty())
.or_else(car_home::root)
.ok_or_else(|| "no CAR state root resolved (no journal dir parent, no home)".to_string())
}
fn open_spool(car_home: &Path) -> Result<Spool, String> {
Spool::open(&car_home.join(FEEDBACK_OUTBOX_DIR))
.map_err(|e| format!("feedback spool unavailable: {e}"))
}
fn named_log_set(car_home: &Path) -> (Vec<NamedLogInput>, Vec<String>) {
let logs = car_home.join("logs");
let mut out = Vec::new();
let mut notes = Vec::new();
match car_registry::supervisor::Supervisor::list_from_manifest(&car_home.join("agents.json")) {
Ok(agents) => {
for agent in agents {
let id = agent.spec.id;
if !valid_manifest_id(&id) {
notes.push(
"agents.json entry with an invalid id skipped — its logs are omitted"
.to_string(),
);
continue;
}
out.push(NamedLogInput {
name: format!("{id}.stdout"),
path: logs.join(format!("{id}.stdout.log")),
max_bytes: None,
});
out.push(NamedLogInput {
name: format!("{id}.stderr"),
path: logs.join(format!("{id}.stderr.log")),
max_bytes: None,
});
}
}
Err(e) => notes.push(format!("agents.json unreadable: {e}")),
};
out.push(NamedLogInput {
name: "car-server.stderr".to_string(),
path: logs.join(DAEMON_STDERR_TEE_FILE),
max_bytes: None,
});
(out, notes)
}
pub(crate) const PREVIEW_EXPIRED_TOKEN: &str = "PREVIEW_EXPIRED";
fn preview_expired_error() -> String {
format!(
"{PREVIEW_EXPIRED_TOKEN}: the preview handle is missing, expired, or was composed \
from different inputs — call feedback.compose_preview again and re-approve the \
refreshed bundle"
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ComposeKey {
description: String,
include_screenshot: bool,
include_diagnostics: bool,
screenshot_sha256: Option<String>,
host_version: Option<String>,
macos_version: Option<String>,
}
impl ComposeKey {
fn of(params: &ComposeParams) -> Self {
let screenshot_sha256 = params.screenshot_b64.as_deref().map(|b64| {
let mut hasher = Sha256::new();
let decoded = if b64.len() > SCREENSHOT_B64_MAX_LEN {
None
} else {
BASE64.decode(b64.as_bytes()).ok()
};
match decoded {
Some(bytes) => hasher.update(&bytes),
None => hasher.update(b64.as_bytes()),
}
format!("{:x}", hasher.finalize())
});
ComposeKey {
description: params.description.trim().to_string(),
include_screenshot: params.include_screenshot,
include_diagnostics: params.include_diagnostics,
screenshot_sha256,
host_version: params.host_version.clone(),
macos_version: params.macos_version.clone(),
}
}
}
struct PreviewEntry {
id: String,
session_id: String,
car_home: PathBuf,
key: ComposeKey,
bundle: RedactedBundle,
stored_at: Instant,
}
static PREVIEW_CACHE: OnceLock<StdMutex<VecDeque<PreviewEntry>>> = OnceLock::new();
fn preview_cache() -> &'static StdMutex<VecDeque<PreviewEntry>> {
PREVIEW_CACHE.get_or_init(|| StdMutex::new(VecDeque::new()))
}
fn store_preview(
session_id: &str,
car_home: &Path,
params: &ComposeParams,
bundle: &RedactedBundle,
) -> String {
let id = Uuid::new_v4().to_string();
let mut cache = preview_cache().lock().expect("preview cache poisoned");
insert_preview(
&mut cache,
PreviewEntry {
id: id.clone(),
session_id: session_id.to_string(),
car_home: car_home.to_path_buf(),
key: ComposeKey::of(params),
bundle: bundle.clone(),
stored_at: Instant::now(),
},
);
id
}
fn insert_preview(cache: &mut VecDeque<PreviewEntry>, entry: PreviewEntry) {
let car_home = entry.car_home.clone();
cache.retain(|e| e.stored_at.elapsed() < PREVIEW_TTL);
cache.push_back(entry);
while cache.iter().filter(|e| e.car_home == car_home).count() > PREVIEW_CACHE_CAP {
let oldest = cache
.iter()
.enumerate()
.filter(|(_, e)| e.car_home == car_home)
.min_by_key(|(_, e)| e.stored_at)
.map(|(index, _)| index);
match oldest {
Some(index) => {
cache.remove(index);
}
None => break,
}
}
}
fn take_preview(
id: &str,
session_id: &str,
car_home: &Path,
params: &ComposeParams,
) -> Option<PreviewEntry> {
let key = ComposeKey::of(params);
let mut cache = preview_cache().lock().expect("preview cache poisoned");
cache.retain(|e| e.stored_at.elapsed() < PREVIEW_TTL);
let position = cache.iter().position(|e| {
e.id == id && e.session_id == session_id && e.car_home == car_home && e.key == key
})?;
cache.remove(position)
}
fn restore_preview(entry: PreviewEntry) {
let mut cache = preview_cache().lock().expect("preview cache poisoned");
insert_preview(&mut cache, entry);
}
enum SubmitBundle {
Previewed(PreviewEntry),
Fresh(RedactedBundle),
}
impl SubmitBundle {
fn bundle(&self) -> &RedactedBundle {
match self {
SubmitBundle::Previewed(entry) => &entry.bundle,
SubmitBundle::Fresh(bundle) => bundle,
}
}
fn release(self) {
if let SubmitBundle::Previewed(entry) = self {
restore_preview(entry);
}
}
}
fn throttled_error(verdict: ThrottleVerdict) -> String {
match verdict {
ThrottleVerdict::Throttled {
used_in_window,
limit,
retry_after_secs,
} => format!(
"feedback submission throttled: {used_in_window}/{limit} reports in the \
last hour — retry in {retry_after_secs}s"
),
ThrottleVerdict::Allowed { .. } => {
"feedback spool returned an inconsistent throttle verdict".to_string()
}
}
}
#[cfg(test)]
fn age_preview_for_test(id: &str, by: Duration) {
let mut cache = preview_cache().lock().expect("preview cache poisoned");
for entry in cache.iter_mut() {
if entry.id == id {
if let Some(rewound) = entry.stored_at.checked_sub(by) {
entry.stored_at = rewound;
}
}
}
}
async fn daemon_runtime_context(state: &ServerState) -> Value {
let role = if state.observer_manifest_path().is_some() {
"observer"
} else {
"primary"
};
let sessions: Vec<(String, std::sync::Arc<crate::session::ClientSession>)> = state
.sessions
.lock()
.await
.iter()
.map(|(id, session)| (id.clone(), session.clone()))
.collect();
let mut session_ids: Vec<String> = sessions.iter().map(|(id, _)| id.clone()).collect();
session_ids.sort();
let mut metrics_summary: Vec<Value> = Vec::new();
for (id, session) in &sessions {
let summary = {
let log = session.runtime.log.lock().await;
car_eventlog::summarize_log(&log)
};
if let Ok(value) = serde_json::to_value(&summary) {
metrics_summary.push(json!({ "session_id": id, "summary": value }));
}
}
metrics_summary.sort_by(|a, b| {
a["session_id"]
.as_str()
.unwrap_or_default()
.cmp(b["session_id"].as_str().unwrap_or_default())
});
let agents: Vec<Value> = state
.host
.agents()
.await
.into_iter()
.map(|a| {
json!({
"id": a.id,
"session_id": a.session_id,
"status": a.status,
})
})
.collect();
let pending_approvals = state
.host
.approvals()
.await
.iter()
.filter(|a| a.status == car_proto::HostApprovalStatus::Pending)
.count();
json!({
"daemon_version": env!("CARGO_PKG_VERSION"),
"protocol_version": car_proto::PROTOCOL_VERSION,
"daemon": {
"version": env!("CARGO_PKG_VERSION"),
"pid": std::process::id(),
"role": role,
},
"connection": {
"active_ws_sessions": session_ids.len(),
"parslee_session": state.parslee_session.get().is_some(),
},
"active_sessions": session_ids,
"active_agents": agents,
"pending_approvals": pending_approvals,
"metrics_summary": metrics_summary,
})
}
async fn fetch_server_rows(state: &ServerState) -> Option<Vec<ServerReportRow>> {
let session = state.parslee_session.get()?;
let org = session
.identity
.active_organization
.as_deref()
.filter(|org| !org.is_empty())?
.to_string();
let transport = match car_parslee::feedback_transport::FeedbackTransport::live() {
Ok(t) => t,
Err(e) => {
tracing::debug!(
target: "car::feedback",
error = %e,
"feedback.list server merge unavailable (no transport); local-only"
);
return None;
}
};
match transport.fetch_my_feedback(&org).await {
Ok(rows) => Some(rows),
Err(e) => {
tracing::debug!(
target: "car::feedback",
error = ?e,
"feedback.list mine fetch failed; local-only"
);
None
}
}
}
fn limit_rows(mut rows: Vec<Value>, requested: Option<usize>) -> (Vec<Value>, bool) {
let limit = requested
.unwrap_or(FEEDBACK_LIST_DEFAULT_LIMIT)
.min(FEEDBACK_LIST_MAX_LIMIT);
let has_more = rows.len() > limit;
if has_more {
rows = rows.split_off(rows.len() - limit);
}
(rows, has_more)
}
fn merge_server_rows(mut local: Vec<Value>, server: Vec<ServerReportRow>) -> Vec<Value> {
let mut matched: Vec<String> = Vec::new();
for row in &mut local {
let acked_server_id = row["state"]["server_id"].as_str().map(str::to_string);
let hit = server
.iter()
.find(|s| acked_server_id.as_deref() == Some(s.id.as_str()));
if let (Some(s), Some(map)) = (hit, row.as_object_mut()) {
map.insert("source".to_string(), json!("server"));
map.insert("server_status".to_string(), json!(s.status));
map.insert("server_updated_at".to_string(), json!(s.updated_at));
map.insert(
"note".to_string(),
json!(format!("server status: {}", s.status)),
);
matched.push(s.id.clone());
}
}
for s in &server {
if matched.iter().any(|m| m == &s.id) {
continue;
}
let title = s
.description
.as_deref()
.map(title_from_description)
.unwrap_or_else(|| "(server report)".to_string());
local.push(json!({
"id": s.id,
"title": title,
"source": "server",
"server_status": s.status,
"note": format!("server status: {}", s.status),
"updated_at": s.updated_at,
}));
}
local
}
fn omitted_item(name: &str, note: impl Into<String>) -> BundleItem {
BundleItem {
name: name.to_string(),
bytes: 0,
included: false,
truncated: false,
moved_to_overflow: false,
note: Some(note.into()),
}
}
fn title_from_description(description: &str) -> String {
let first_line = description.lines().next().unwrap_or("").trim();
first_line.chars().take(TITLE_MAX_CHARS).collect()
}
fn state_value(state: &SpoolState) -> Result<Value, String> {
serde_json::to_value(state).map_err(|e| format!("serialize spool state: {e}"))
}
fn summaries(spool: &Spool) -> Result<Vec<Value>, String> {
let rows = spool
.list()
.map_err(|e| format!("feedback spool list failed: {e}"))?;
rows.iter().map(summary_value).collect()
}
fn summary_value(row: &SpoolEntrySummary) -> Result<Value, String> {
let mut value = serde_json::to_value(row).map_err(|e| format!("serialize summary: {e}"))?;
if let Some(map) = value.as_object_mut() {
map.insert("source".to_string(), Value::String("local".to_string()));
}
Ok(value)
}
fn staleness_value(spool: &Spool) -> Result<Value, String> {
let notice = spool
.staleness()
.map_err(|e| format!("feedback spool staleness failed: {e}"))?;
Ok(match notice {
Some(n) => json!({
"pending_count": n.pending_count,
"oldest_age_days": n.oldest_age_days,
}),
None => Value::Null,
})
}
async fn run_blocking<T, F>(work: F) -> Result<T, String>
where
T: Send + 'static,
F: FnOnce() -> T + Send + 'static,
{
tokio::task::spawn_blocking(work)
.await
.map_err(|e| format!("feedback task failed: {e}"))
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use tempfile::TempDir;
fn compose_params(description: &str) -> ComposeParams {
ComposeParams {
description: description.to_string(),
include_screenshot: false,
include_diagnostics: false,
screenshot_b64: None,
host_version: None,
macos_version: None,
}
}
fn compose_bundle(
params: &ComposeParams,
car_home: &Path,
daemon_ctx: Option<Value>,
) -> Result<RedactedBundle, String> {
let diagnostics = FeedbackDiagnostics::isolated(
car_home.join("feedback-test-models"),
car_home.join("feedback-test-huggingface-hub"),
);
super::compose_bundle(params, car_home, &diagnostics, daemon_ctx)
}
fn fake_jpeg() -> Vec<u8> {
let mut v = vec![0xFF, 0xD8, 0xFF, 0xE0];
v.extend_from_slice(b"jfif-pixel-payload");
v.extend_from_slice(&[0xFF, 0xD9]);
v
}
#[test]
fn title_is_first_line_capped_at_80_chars() {
assert_eq!(title_from_description("hello\nworld"), "hello");
let long = "x".repeat(200);
assert_eq!(title_from_description(&long).chars().count(), 80);
assert_eq!(title_from_description(" spaced \nrest"), "spaced");
}
#[test]
fn named_log_set_names_agent_pairs_and_the_stderr_tee_never_globs() {
let tmp = TempDir::new().unwrap();
fs::write(
tmp.path().join("agents.json"),
r#"{"agents":[{"id":"agent-a","name":"A","command":"/bin/true"}]}"#,
)
.unwrap();
fs::create_dir_all(tmp.path().join("logs")).unwrap();
fs::write(tmp.path().join("logs/stray.log"), "not yours").unwrap();
let (set, notes) = named_log_set(tmp.path());
assert!(notes.is_empty());
let names: Vec<&str> = set.iter().map(|l| l.name.as_str()).collect();
assert_eq!(
names,
vec!["agent-a.stdout", "agent-a.stderr", "car-server.stderr"]
);
assert!(set.iter().all(|l| !l.path.ends_with("stray.log")));
}
#[test]
fn named_log_set_missing_manifest_is_just_the_tee() {
let tmp = TempDir::new().unwrap();
let (set, notes) = named_log_set(tmp.path());
assert!(notes.is_empty(), "an absent manifest is an empty agent set");
assert_eq!(set.len(), 1);
assert_eq!(set[0].name, "car-server.stderr");
}
#[test]
fn named_log_set_corrupt_manifest_notes_and_still_includes_the_tee() {
let tmp = TempDir::new().unwrap();
fs::write(tmp.path().join("agents.json"), "{not json").unwrap();
let (set, notes) = named_log_set(tmp.path());
assert_eq!(notes.len(), 1);
assert_eq!(set.len(), 1);
}
#[test]
fn named_log_set_rejects_traversal_and_separator_ids_from_the_manifest() {
let tmp = TempDir::new().unwrap();
fs::write(
tmp.path().join("agents.json"),
r#"{"agents":[
{"id":"../../outside","name":"Evil","command":"/bin/true"},
{"id":"/etc/passwd","name":"Evil2","command":"/bin/true"},
{"id":"good-agent","name":"Good","command":"/bin/true"}
]}"#,
)
.unwrap();
let (set, notes) = named_log_set(tmp.path());
let names: Vec<&str> = set.iter().map(|l| l.name.as_str()).collect();
assert_eq!(
names,
vec![
"good-agent.stdout",
"good-agent.stderr",
"car-server.stderr"
],
"only the valid id and the tee may survive"
);
let logs_root = tmp.path().join("logs");
assert!(
set.iter().all(|l| l.path.starts_with(&logs_root)),
"every named-log path must stay under logs/: {set:?}"
);
assert_eq!(notes.len(), 2, "each skipped id leaves a manifest note");
assert!(notes.iter().all(|n| n.contains("invalid id")));
}
#[test]
fn valid_manifest_id_matches_supervisor_rules() {
for good in ["agent-a", "a_b.c", "A9"] {
assert!(valid_manifest_id(good), "{good} should be valid");
}
for bad in ["", ".", "..", "../x", "a/b", "a\\b", "a b", "/abs"] {
assert!(!valid_manifest_id(bad), "{bad} should be invalid");
}
}
#[test]
fn feedback_bundle_counts_models_from_the_global_cache_when_car_home_is_relocated() {
let state_home = TempDir::new().unwrap();
let global_cache = TempDir::new().unwrap();
let huggingface_hub = TempDir::new().unwrap();
let model = global_cache.path().join("Qwen3-Embedding-0.6B");
fs::create_dir(&model).unwrap();
fs::write(model.join("model.gguf"), b"weights").unwrap();
let mut params = compose_params("model count must follow the runtime cache");
params.include_diagnostics = true;
let bundle = compose_bundle_at(
¶ms,
state_home.path(),
global_cache.path(),
Some(huggingface_hub.path()),
FeedbackDiagnosticsProbe::Isolated,
None,
)
.unwrap();
assert!(
!state_home.path().join("models").exists(),
"the relocated state root must not contain the model weights"
);
assert_eq!(
bundle.doctor_report.as_ref().unwrap()["installed_models"],
json!(1),
"only the model in the test-owned global cache may be counted"
);
assert_eq!(
car_inference::doctor::diagnose_at_isolated(
state_home.path(),
&state_home.path().join("models"),
huggingface_hub.path(),
&car_inference::doctor::DoctorOptions::default(),
)
.installed_models,
0,
"the old state-root-derived weights path must remain empty"
);
}
#[test]
fn compose_with_screenshot_toggle_off_drops_supplied_bytes_entirely() {
let tmp = TempDir::new().unwrap();
let payload = BASE64.encode(fake_jpeg());
let mut params = compose_params("something broke");
params.screenshot_b64 = Some(payload.clone());
params.include_screenshot = false;
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let serialized = serde_json::to_string(&bundle).unwrap();
assert!(
!serialized.contains("screenshot"),
"toggle off must leave no screenshot item anywhere (PREV-3): {serialized}"
);
assert!(!serialized.contains(&payload));
}
#[test]
fn compose_with_screenshot_toggle_on_accounts_it_in_the_manifest() {
let tmp = TempDir::new().unwrap();
let jpeg = fake_jpeg();
let payload = BASE64.encode(&jpeg);
let mut params = compose_params("something broke");
params.screenshot_b64 = Some(payload.clone());
params.include_screenshot = true;
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let context = bundle.runtime_context.as_ref().expect("screenshot context");
assert_eq!(context["screenshot"]["jpeg_b64"], json!(payload));
assert_eq!(context["screenshot"]["byte_len"], json!(jpeg.len()));
let item = bundle
.manifest
.items
.iter()
.find(|i| i.name == "screenshot")
.expect("manifest accounts the screenshot");
assert!(item.included);
assert_eq!(item.bytes, payload.len() as u64);
}
#[test]
fn compose_soft_fails_non_jpeg_oversize_and_bad_base64_screenshots() {
let tmp = TempDir::new().unwrap();
let cases: Vec<(String, &str)> = vec![
(
BASE64.encode(b"just some text pretending to be a screenshot"),
"not a JPEG",
),
(BASE64.encode(vec![0u8; SCREENSHOT_MAX_BYTES + 1]), "cap"),
("!!!not-base64!!!".to_string(), "invalid base64"),
];
for (payload, expected_note) in cases {
let mut params = compose_params("desc");
params.include_screenshot = true;
params.screenshot_b64 = Some(payload.clone());
let bundle = compose_bundle(¶ms, tmp.path(), None)
.expect("a bad screenshot must not abort the report (BND-5)");
assert!(
bundle.runtime_context.is_none()
|| bundle
.runtime_context
.as_ref()
.unwrap()
.get("screenshot")
.is_none(),
"the bad screenshot must not ride the bundle"
);
let item = bundle
.manifest
.items
.iter()
.find(|i| i.name == "screenshot")
.expect("the omission must be visible in the manifest");
assert!(!item.included);
let note = item.note.as_deref().unwrap_or_default();
assert!(
note.contains(expected_note),
"note {note:?} should mention {expected_note:?}"
);
let serialized = serde_json::to_string(&bundle).unwrap();
assert!(
!serialized.contains(&payload),
"payload bytes must be dropped"
);
}
}
#[test]
fn oversize_screenshot_b64_is_refused_by_length_before_any_decode() {
let over = "!".repeat(SCREENSHOT_B64_MAX_LEN + 1);
let note = validate_screenshot(Some(&over)).unwrap_err();
assert!(
note.contains("cap"),
"length gate must name the cap: {note}"
);
assert!(
!note.contains("invalid base64"),
"an over-ceiling payload must never reach the decoder: {note}"
);
let at_ceiling = "!".repeat(SCREENSHOT_B64_MAX_LEN);
let note = validate_screenshot(Some(&at_ceiling)).unwrap_err();
assert!(
note.contains("invalid base64"),
"at the ceiling the decoder still runs (strict > gate): {note}"
);
let mut max_jpeg = vec![0xFF, 0xD8, 0xFF, 0xE0];
max_jpeg.resize(SCREENSHOT_MAX_BYTES, 0u8);
let n = max_jpeg.len();
max_jpeg[n - 2] = 0xFF;
max_jpeg[n - 1] = 0xD9;
let b64 = BASE64.encode(&max_jpeg);
assert!(b64.len() <= SCREENSHOT_B64_MAX_LEN);
assert_eq!(
validate_screenshot(Some(&b64)).unwrap().1,
SCREENSHOT_MAX_BYTES
);
let oversize_valid = BASE64.encode(vec![0u8; SCREENSHOT_MAX_BYTES + 1024]);
assert!(oversize_valid.len() > SCREENSHOT_B64_MAX_LEN);
let mut params = compose_params("desc");
params.screenshot_b64 = Some(oversize_valid.clone());
let mut raw_hash = Sha256::new();
raw_hash.update(oversize_valid.as_bytes());
assert_eq!(
ComposeKey::of(¶ms).screenshot_sha256,
Some(format!("{:x}", raw_hash.finalize())),
"the key must hash the raw text of an oversize payload, never decode it"
);
params.include_screenshot = true;
let tmp = TempDir::new().unwrap();
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let item = bundle
.manifest
.items
.iter()
.find(|i| i.name == "screenshot")
.expect("omission visible in the manifest");
assert!(!item.included);
assert!(item.note.as_deref().unwrap_or_default().contains("cap"));
}
#[test]
fn parse_refuses_a_description_over_the_server_maximum() {
let over: ComposeParams = parse_params(&json!({
"description": "a".repeat(DESCRIPTION_MAX_CHARS + 1),
}))
.unwrap();
let err = over.validate().unwrap_err();
assert!(err.starts_with("invalid params"), "{err}");
assert!(err.contains(&DESCRIPTION_MAX_CHARS.to_string()), "{err}");
let at_max: ComposeParams = parse_params(&json!({
"description": "a".repeat(DESCRIPTION_MAX_CHARS),
}))
.unwrap();
at_max.validate().unwrap();
let padded: ComposeParams = parse_params(&json!({
"description": format!(" {} \n", "a".repeat(DESCRIPTION_MAX_CHARS)),
}))
.unwrap();
padded
.validate()
.expect("whitespace around an at-max description is trimmed, not counted");
let submit: SubmitParams = parse_params(&json!({
"description": "a".repeat(DESCRIPTION_MAX_CHARS + 1),
"lane": "anonymous",
}))
.unwrap();
assert!(submit.compose.validate().is_err());
}
#[test]
fn validate_screenshot_accepts_jpeg_magic_and_reports_decoded_len() {
let jpeg = fake_jpeg();
let b64 = BASE64.encode(&jpeg);
let (ret, len) = validate_screenshot(Some(&b64)).unwrap();
assert_eq!(ret, b64);
assert_eq!(len, jpeg.len());
assert!(validate_screenshot(None)
.unwrap_err()
.contains("no screenshot_b64"));
}
#[test]
fn compose_diagnostics_off_notes_the_exclusion_and_collects_no_logs() {
let tmp = TempDir::new().unwrap();
fs::create_dir_all(tmp.path().join("logs")).unwrap();
fs::write(tmp.path().join("logs").join(DAEMON_STDERR_TEE_FILE), "boom").unwrap();
let bundle = compose_bundle(&compose_params("desc"), tmp.path(), None).unwrap();
assert!(bundle.log_tails.is_empty());
assert!(bundle.doctor_report.is_none());
assert!(bundle.runtime_context.is_none());
assert!(bundle
.manifest
.items
.iter()
.any(|i| i.name == "diagnostics" && !i.included));
}
#[test]
fn compose_with_diagnostics_extracts_16hex_dedup_signature_stable_across_timestamps() {
let tmp = TempDir::new().unwrap();
fs::create_dir_all(tmp.path().join("logs")).unwrap();
let tee = tmp.path().join("logs").join(DAEMON_STDERR_TEE_FILE);
fs::write(
&tee,
"2026-08-31T12:00:01Z starting up\n\
2026-08-31T12:00:02Z thread 'main' panicked at src/executor.rs:412: \
index out of bounds: the len is 3 but the index is 9\n",
)
.unwrap();
let mut params = compose_params("it crashed");
params.include_diagnostics = true;
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let sig = bundle
.dedup_signature
.as_deref()
.expect("a panic in the tee must yield a dedup signature")
.to_string();
assert_eq!(sig.len(), 16, "signature must be the 16-hex recipe prefix");
assert!(sig.chars().all(|c| c.is_ascii_hexdigit()));
let item = bundle
.manifest
.items
.iter()
.find(|i| i.name == "dedup_signature")
.expect("the manifest must account item 8");
assert!(item.included);
fs::write(
&tee,
"2026-09-07T23:11:45Z starting up\n\
2026-09-07T23:11:46Z thread 'main' panicked at src/executor.rs:498: \
index out of bounds: the len is 12 but the index is 44\n",
)
.unwrap();
let recurrence = compose_bundle(¶ms, tmp.path(), None).unwrap();
assert_eq!(
recurrence.dedup_signature.as_deref(),
Some(sig.as_str()),
"timestamp/index noise must not change the signature"
);
}
#[test]
fn compose_with_clean_logs_has_no_dedup_signature() {
let tmp = TempDir::new().unwrap();
fs::create_dir_all(tmp.path().join("logs")).unwrap();
fs::write(
tmp.path().join("logs").join(DAEMON_STDERR_TEE_FILE),
"starting up\nall good\nready\n",
)
.unwrap();
let mut params = compose_params("just a suggestion");
params.include_diagnostics = true;
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
assert!(bundle.dedup_signature.is_none(), "item 8 is optional");
assert!(
!bundle
.manifest
.items
.iter()
.any(|i| i.name == "dedup_signature"),
"no signature ⇒ no manifest item"
);
}
#[test]
fn submitted_spool_entry_durably_carries_the_dedup_signature() {
let tmp = TempDir::new().unwrap();
fs::create_dir_all(tmp.path().join("logs")).unwrap();
fs::write(
tmp.path().join("logs").join(DAEMON_STDERR_TEE_FILE),
"thread 'main' panicked at src/lib.rs:7: boom\n",
)
.unwrap();
let mut params = compose_params("crash report");
params.include_diagnostics = true;
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let expected = bundle
.dedup_signature
.clone()
.expect("panic yields a signature");
let spool = Spool::open(&tmp.path().join(FEEDBACK_OUTBOX_DIR)).unwrap();
let id = spool
.enqueue(&bundle, IdentityLane::Anonymous, "crash report")
.unwrap();
let reopened = Spool::open(&tmp.path().join(FEEDBACK_OUTBOX_DIR)).unwrap();
let persisted = reopened.load_bundle(&id).unwrap();
assert_eq!(
persisted.dedup_signature.as_deref(),
Some(expected.as_str())
);
}
#[test]
fn submit_result_survives_a_failed_read_back_without_an_error() {
let tmp = TempDir::new().unwrap();
let root = tmp.path().join(FEEDBACK_OUTBOX_DIR);
let spool = Spool::open(&root).unwrap();
let bundle = compose_bundle(&compose_params("the deck froze"), tmp.path(), None).unwrap();
let id = spool
.enqueue(&bundle, IdentityLane::Anonymous, "the deck froze")
.unwrap();
let healthy = submit_result(&id, spool.list()).unwrap();
assert_eq!(healthy["submission_id"], id.as_str());
assert_eq!(healthy["state"]["state"], "queued");
assert_eq!(healthy["source"], "local");
assert!(healthy["client_submission_id"].is_string());
let io_failure = submit_result(&id, Err(std::io::Error::other("too many open files")))
.expect("a read-back failure after a durable enqueue is not an RPC error");
assert_eq!(io_failure, healthy, "same wire result from local facts");
let missing = submit_result(&id, Ok(Vec::new()))
.expect("an entry the listing misses is still the enqueued entry");
assert_eq!(missing, healthy);
assert_eq!(spool.list().unwrap().len(), 1);
}
#[test]
fn entry_id_carries_the_client_submission_id() {
let tmp = TempDir::new().unwrap();
let spool = Spool::open(&tmp.path().join(FEEDBACK_OUTBOX_DIR)).unwrap();
let bundle = compose_bundle(&compose_params("desc"), tmp.path(), None).unwrap();
let id = spool
.enqueue(&bundle, IdentityLane::Anonymous, "t")
.unwrap();
let listed = spool
.list()
.unwrap()
.into_iter()
.find(|s| s.id == id)
.unwrap()
.client_submission_id;
assert_eq!(
client_submission_id_from_entry_id(&id).as_deref(),
Some(listed.as_str())
);
let bogus: SpoolEntryId = serde_json::from_value(json!("not-a-uuid-suffix")).unwrap();
assert_eq!(client_submission_id_from_entry_id(&bogus), None);
}
#[test]
fn lane_authenticated_without_org_id_is_a_structured_error() {
assert!(resolve_lane(LaneParam::Authenticated, None, None)
.unwrap_err()
.contains("signed-in Parslee session"));
assert!(
resolve_lane(LaneParam::Authenticated, Some("org_other"), Some("org_x"))
.unwrap_err()
.contains("does not match")
);
assert_eq!(
resolve_lane(LaneParam::Authenticated, None, Some("org_x")).unwrap(),
IdentityLane::Authenticated {
org_id: "org_x".to_string()
}
);
assert_eq!(
resolve_lane(LaneParam::Authenticated, Some("org_x"), Some("org_x")).unwrap(),
IdentityLane::Authenticated {
org_id: "org_x".to_string()
}
);
assert_eq!(
resolve_lane(LaneParam::Anonymous, Some("org_x"), Some("org_real")).unwrap(),
IdentityLane::Anonymous
);
}
#[test]
fn preview_cache_round_trips_only_for_matching_home_and_inputs() {
let tmp = TempDir::new().unwrap();
let other = TempDir::new().unwrap();
let params = compose_params("preview me");
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let id = store_preview("session-a", tmp.path(), ¶ms, &bundle);
assert!(take_preview(&id, "session-b", tmp.path(), ¶ms).is_none());
assert!(take_preview(&id, "session-a", other.path(), ¶ms).is_none());
let mut flipped = params.clone();
flipped.include_diagnostics = true;
assert!(take_preview(&id, "session-a", tmp.path(), &flipped).is_none());
let mut edited = params.clone();
edited.description = "preview me, edited".to_string();
assert!(take_preview(&id, "session-a", tmp.path(), &edited).is_none());
let mut swapped_shot = params.clone();
swapped_shot.screenshot_b64 = Some(BASE64.encode(fake_jpeg()));
assert!(take_preview(&id, "session-a", tmp.path(), &swapped_shot).is_none());
let redeemed =
take_preview(&id, "session-a", tmp.path(), ¶ms).expect("valid handle redeems");
assert_eq!(redeemed.bundle, bundle);
assert!(take_preview(&id, "session-a", tmp.path(), ¶ms).is_none());
assert!(take_preview("no-such-handle", "session-a", tmp.path(), ¶ms).is_none());
}
#[test]
fn concurrent_redemption_allows_exactly_one_submitter() {
let tmp = TempDir::new().unwrap();
let home = tmp.path().to_path_buf();
let params = compose_params("one consent, one redemption");
let bundle = compose_bundle(¶ms, &home, None).unwrap();
let id = store_preview("session-a", &home, ¶ms, &bundle);
let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
let handles: Vec<_> = (0..2)
.map(|_| {
let barrier = barrier.clone();
let id = id.clone();
let home = home.clone();
let params = params.clone();
std::thread::spawn(move || {
barrier.wait();
take_preview(&id, "session-a", &home, ¶ms).is_some()
})
})
.collect();
barrier.wait();
let redeemed = handles
.into_iter()
.map(|handle| handle.join().expect("redemption thread"))
.filter(|won| *won)
.count();
assert_eq!(redeemed, 1, "a handle may be consumed only once");
}
#[test]
fn restored_preview_redeems_again_with_its_original_bundle_then_is_consumed() {
let tmp = TempDir::new().unwrap();
let params = compose_params("throttled once, approved once");
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let id = store_preview("session-a", tmp.path(), ¶ms, &bundle);
let redeemed =
take_preview(&id, "session-a", tmp.path(), ¶ms).expect("first redemption");
assert!(
take_preview(&id, "session-a", tmp.path(), ¶ms).is_none(),
"redemption consumes the handle"
);
SubmitBundle::Previewed(redeemed).release();
let again = take_preview(&id, "session-a", tmp.path(), ¶ms)
.expect("a released handle must redeem again without a re-preview");
assert_eq!(
again.bundle, bundle,
"the restored entry holds the approved bytes"
);
SubmitBundle::Fresh(bundle.clone()).release();
assert!(
take_preview(&id, "session-a", tmp.path(), ¶ms).is_none(),
"the second redemption consumed it for good"
);
}
#[test]
fn restoring_a_handle_at_capacity_evicts_by_consent_time_not_position() {
let tmp = TempDir::new().unwrap();
let mut ids = Vec::new();
for i in 0..PREVIEW_CACHE_CAP {
let params = compose_params(&format!("capacity fill {i}"));
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
ids.push((
store_preview("session-a", tmp.path(), ¶ms, &bundle),
params,
));
std::thread::sleep(std::time::Duration::from_millis(2));
}
let (oldest_id, oldest_params) = ids[0].clone();
let redeemed = take_preview(&oldest_id, "session-a", tmp.path(), &oldest_params)
.expect("oldest redeems");
let newest_params = compose_params("minted during the in-flight submit");
let newest_bundle = compose_bundle(&newest_params, tmp.path(), None).unwrap();
let newest_id = store_preview("session-a", tmp.path(), &newest_params, &newest_bundle);
SubmitBundle::Previewed(redeemed).release();
assert!(
take_preview(&newest_id, "session-a", tmp.path(), &newest_params).is_some(),
"the newest unrelated handle must survive a restore at capacity"
);
assert!(
take_preview(&oldest_id, "session-a", tmp.path(), &oldest_params).is_none(),
"the restored entry is the oldest by consent time, so the cap evicts IT"
);
for (id, params) in &ids[1..] {
assert!(
take_preview(id, "session-a", tmp.path(), params).is_some(),
"the other capacity fills are untouched"
);
}
}
#[test]
fn preview_redeems_after_trim_normalization_but_not_after_expiry() {
let tmp = TempDir::new().unwrap();
let mut previewed = compose_params("described in the sheet");
previewed.description = " described in the sheet \n".to_string();
let bundle = compose_bundle(&previewed, tmp.path(), None).unwrap();
let id = store_preview("session-a", tmp.path(), &previewed, &bundle);
let trimmed = compose_params("described in the sheet");
assert!(
take_preview(&id, "session-a", tmp.path(), &trimmed).is_some(),
"a trim-only difference is the host's wire normalization, not a user edit"
);
let id = store_preview("session-a", tmp.path(), &previewed, &bundle);
age_preview_for_test(&id, PREVIEW_TTL + Duration::from_secs(1));
assert!(
take_preview(&id, "session-a", tmp.path(), &previewed).is_none(),
"an expired handle must not redeem"
);
}
#[test]
fn preview_expired_error_carries_the_pinned_token() {
assert!(preview_expired_error().contains("PREVIEW_EXPIRED"));
assert!(preview_expired_error().contains("compose_preview"));
}
#[test]
fn preview_cache_evicts_oldest_beyond_cap() {
let tmp = TempDir::new().unwrap();
let params = compose_params("cap check");
let bundle = compose_bundle(¶ms, tmp.path(), None).unwrap();
let first = store_preview("session-a", tmp.path(), ¶ms, &bundle);
let mut rest = Vec::new();
for _ in 0..PREVIEW_CACHE_CAP {
rest.push(store_preview("session-a", tmp.path(), ¶ms, &bundle));
}
assert!(
take_preview(&first, "session-a", tmp.path(), ¶ms).is_none(),
"the oldest entry past the cap must be evicted"
);
assert!(take_preview(rest.last().unwrap(), "session-a", tmp.path(), ¶ms).is_some());
}
fn local_row(id: &str, csid: &str, state: Value) -> Value {
json!({
"id": id,
"client_submission_id": csid,
"state": state,
"title": "local report",
"source": "local",
})
}
fn server_row(id: &str, status: &str) -> ServerReportRow {
ServerReportRow {
id: id.to_string(),
client_submission_id: None,
status: status.to_string(),
description: Some("filed from another install".to_string()),
omitted: Vec::new(),
created_at: None,
updated_at: Some("2026-09-01T00:00:00Z".to_string()),
}
}
#[test]
fn merge_marks_acknowledged_row_with_advanced_server_state() {
let local = vec![local_row(
"e1",
"csid-1",
json!({"state": "acknowledged", "server_id": "srv-9"}),
)];
let merged = merge_server_rows(local, vec![server_row("srv-9", "resolved")]);
assert_eq!(merged.len(), 1, "matched rows must not duplicate");
assert_eq!(merged[0]["source"], "server");
assert_eq!(merged[0]["server_status"], "resolved");
assert!(merged[0]["note"].as_str().unwrap().contains("resolved"));
}
#[test]
fn merge_appends_server_only_rows_and_leaves_local_rows_local() {
let local = vec![local_row("e1", "csid-1", json!({"state": "queued"}))];
let merged = merge_server_rows(local, vec![server_row("srv-42", "open")]);
assert_eq!(merged.len(), 2);
assert_eq!(
merged[0]["source"], "local",
"unmatched local row stays local"
);
assert!(merged[0].get("note").is_none() || merged[0]["note"].is_null());
assert_eq!(merged[1]["source"], "server");
assert_eq!(merged[1]["id"], "srv-42");
assert_eq!(merged[1]["server_status"], "open");
assert_eq!(merged[1]["title"], "filed from another install");
}
#[test]
fn server_rows_without_client_ids_append_and_never_collapse_local_rows() {
let local = vec![local_row("e1", "csid-7", json!({"state": "queued"}))];
let merged = merge_server_rows(local, vec![server_row("csid-7", "received")]);
assert_eq!(
merged.len(),
2,
"an id coincidence is not a shared key: the rows must not collapse"
);
assert_eq!(merged[0]["source"], "local", "the local row stays local");
assert_eq!(merged[0]["id"], "e1");
assert_eq!(merged[1]["source"], "server");
assert_eq!(merged[1]["id"], "csid-7");
assert_eq!(merged[1]["server_status"], "received");
}
}