diff --git a/README.md b/README.md index d8bca22..2512fb5 100644 --- a/README.md +++ b/README.md @@ -142,9 +142,26 @@ tapesctl sync # backstop: sweep transcripts no live tailer saw ``` `sync` is safe to run repeatedly — the ingest endpoint keys rows on a content -hash, so re-offering an unchanged transcript is a cheap `deduped`. It sweeps +hash, so an unchanged transcript is reported as `already present`. It sweeps `~/.claude/projects` by default (`--projects-root` to point elsewhere), and -`--since-days` bounds how far back it looks. +`--since-days` bounds how far back it looks. The summary distinguishes `new +versions` from files `already present`, then separately reports how many unique +sessions had asynchronous projection queued. That line means queued, not +projected: `sync` does not poll the read API, so reads may lag briefly. Even an +already-present upload requeues projection server-side. + +Pass global `-v` to print every file's harness session id, path, +server-reported record count, and outcome (`new`, `already present`, `failed`, +or `unavailable` when a successful response omits dedup status). Ack fields are +independent, so either the count or outcome can be unavailable while the other +is known. Normal mode omits successful per-file detail. + +A historical transcript can create partial, browsable transcript-derived calls +even when no wire calls were captured. That reconstruction lacks full wire +fidelity: it cannot recover exact provider request/response bytes, and the +transcript may omit some harness-side calls or context. If usable wire capture +arrives later, it replaces the transcript-derived calls rather than duplicating +them; the transcript still supplies causal structure. ### Capturing `pi` diff --git a/crates/tapesctl/src/cli.rs b/crates/tapesctl/src/cli.rs index d6ad715..902fc51 100644 --- a/crates/tapesctl/src/cli.rs +++ b/crates/tapesctl/src/cli.rs @@ -25,7 +25,8 @@ use clap::{Args, Parser, Subcommand}; arg_required_else_help = true )] pub struct Cli { - /// Increase log verbosity (`-v` debug, `-vv` trace). `RUST_LOG` overrides. + /// Increase detail (`-v` adds sync file outcomes and debug logs; `-vv` + /// enables trace logs). `RUST_LOG` overrides only the log level. #[arg(short, long, global = true, action = clap::ArgAction::Count)] pub verbose: u8, @@ -304,8 +305,8 @@ pub enum Command { /// Sweep completed harness transcripts into the tapes ingest server. /// /// The live tailer that runs during `start` is the primary path; this is the - /// backstop for sessions no capture was running for (dedup makes re-push - /// safe). + /// backstop for sessions no capture was running for. Re-push is safe and + /// requeues asynchronous projection; `-v` prints each file's outcome. Sync(SyncArgs), /// Read sessions. diff --git a/crates/tapesctl/src/lib.rs b/crates/tapesctl/src/lib.rs index 6e4b890..050d65f 100644 --- a/crates/tapesctl/src/lib.rs +++ b/crates/tapesctl/src/lib.rs @@ -290,6 +290,7 @@ pub async fn dispatch(invocation: Invocation) -> Result<()> { /// Dispatch a parsed CLI invocation. pub async fn run(cli: Cli) -> Result<()> { + let verbosity = cli.verbose; match cli.command { Command::Version => { println!("{}", banner()); @@ -297,7 +298,7 @@ pub async fn run(cli: Cli) -> Result<()> { } Command::Start(args) => start(args).await, Command::Capture(args) => capture::run(args).await, - Command::Sync(args) => transcript::sync::run(args).await, + Command::Sync(args) => transcript::sync::run_with_verbosity(args, verbosity).await, Command::Sessions(command) => api::sessions(command).await, Command::Traces(command) => api::traces(command).await, Command::Spans(command) => api::spans(command).await, diff --git a/crates/tapesctl/src/transcript/client.rs b/crates/tapesctl/src/transcript/client.rs index 8c265c9..133d1ee 100644 --- a/crates/tapesctl/src/transcript/client.rs +++ b/crates/tapesctl/src/transcript/client.rs @@ -31,7 +31,7 @@ //! else, exactly like the Go reference client. use serde::Deserialize; -use serde_json::value::RawValue; +use serde_json::{Value, value::RawValue}; use snafu::ResultExt; use std::time::Duration; use tapes_harnesses::transcript::{ @@ -57,14 +57,57 @@ pub struct UploadOutcome { pub records: usize, } -/// The server's 202 acknowledgement. Both fields default so a server that grows -/// the response — or trims it — does not turn a successful upload into an error. +/// The server's 202 acknowledgement. Each field is advisory and independently +/// optional: older servers and proxies may return only one of them. #[derive(Debug, Default, Deserialize)] struct TranscriptAck { - #[serde(default)] - deduped: bool, - #[serde(default)] - records: usize, + deduped: Option, + records: Option, +} + +impl TranscriptAck { + fn deduped(&self) -> Option { + self.deduped.as_ref()?.as_bool() + } + + fn records(&self) -> Option { + usize::try_from(self.records.as_ref()?.as_u64()?).ok() + } +} + +/// Successful upload details before unavailable fields are collapsed onto the +/// historical public defaults in [`UploadOutcome`]. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub(crate) struct DetailedUploadOutcome { + deduped: Option, + records: Option, +} + +impl DetailedUploadOutcome { + pub(crate) fn deduped(self) -> Option { + self.deduped + } + + pub(crate) fn records(self) -> Option { + self.records + } + + pub(crate) fn deduped_for_log(self) -> String { + self.deduped + .map_or_else(|| "unavailable".to_owned(), |value| value.to_string()) + } + + pub(crate) fn records_for_log(self) -> String { + self.records + .map_or_else(|| "unavailable".to_owned(), |value| value.to_string()) + } + + fn compatibility_outcome(self) -> UploadOutcome { + UploadOutcome { + deduped: self.deduped.unwrap_or(false), + records: self.records.unwrap_or(0), + } + } } /// A client for one tapes ingest server's transcript lane. @@ -109,6 +152,17 @@ impl TranscriptClient { /// Post one already-assembled payload. pub async fn post_transcript(&self, payload: &TranscriptPayload<'_>) -> Result { + self.post_transcript_detailed(payload) + .await + .map(DetailedUploadOutcome::compatibility_outcome) + } + + /// Post one payload without collapsing missing acknowledgement fields onto + /// their historical public defaults. + pub(crate) async fn post_transcript_detailed( + &self, + payload: &TranscriptPayload<'_>, + ) -> Result { let response = self .http .post(self.endpoint.clone()) @@ -128,11 +182,13 @@ impl TranscriptClient { } // A body that does not parse is not a failed upload: the server already - // answered 2xx, and the ack is only advisory detail for the log. - let ack: TranscriptAck = response.json().await.unwrap_or_default(); - Ok(UploadOutcome { - deduped: ack.deduped, - records: ack.records, + // answered 2xx. Keep each advisory field independently optional for + // truth-sensitive callers; the public path defaults only the field that + // was unavailable. + let ack = response.json::().await.unwrap_or_default(); + Ok(DetailedUploadOutcome { + deduped: ack.deduped(), + records: ack.records(), }) } @@ -147,13 +203,25 @@ impl TranscriptClient { session: &TranscriptSession, file: &TranscriptFile, ) -> Result { + self.upload_file_detailed(session, file) + .await + .map(DetailedUploadOutcome::compatibility_outcome) + } + + /// Upload one file while preserving dedup status and record count as + /// independently optional acknowledgement fields. + pub(crate) async fn upload_file_detailed( + &self, + session: &TranscriptSession, + file: &TranscriptFile, + ) -> Result { let raw = std::fs::read(&file.path).context(error::TranscriptReadSnafu { path: file.path.clone(), })?; let records = RawValue::from_string(jsonl_to_records(&raw)).context(error::TranscriptRecordsSnafu)?; let payload = build_payload(session, file, &records); - self.post_transcript(&payload).await + self.post_transcript_detailed(&payload).await } } @@ -230,6 +298,122 @@ mod tests { assert_eq!(outcome.records, 3); } + #[test] + fn detailed_acknowledgement_log_values_are_readable() { + let partial = DetailedUploadOutcome { + deduped: Some(true), + records: None, + }; + assert_eq!(partial.deduped_for_log(), "true"); + assert_eq!(partial.records_for_log(), "unavailable"); + + let malformed = DetailedUploadOutcome::default(); + assert_eq!(malformed.deduped_for_log(), "unavailable"); + assert_eq!(malformed.records_for_log(), "unavailable"); + } + + #[tokio::test] + async fn a_dedup_only_ack_preserves_the_public_dedup_field() { + let server = + ingest_server(ResponseTemplate::new(202).set_body_string(r#"{"deduped":true}"#)).await; + let dir = tempfile::tempdir().unwrap(); + let file = write_jsonl(dir.path(), "sid-1.jsonl", "{\"a\":1}\n"); + let client = TranscriptClient::new(&Url::parse(&server.uri()).unwrap()).unwrap(); + + let outcome = client.upload_file(&session(), &file).await.unwrap(); + assert_eq!( + outcome, + UploadOutcome { + deduped: true, + records: 0, + } + ); + + let detailed = client + .upload_file_detailed(&session(), &file) + .await + .unwrap(); + assert_eq!(detailed.deduped(), Some(true)); + assert_eq!(detailed.records(), None); + } + + #[tokio::test] + async fn a_records_only_ack_preserves_the_public_record_count() { + let server = + ingest_server(ResponseTemplate::new(202).set_body_string(r#"{"records":3}"#)).await; + let dir = tempfile::tempdir().unwrap(); + let file = write_jsonl(dir.path(), "sid-1.jsonl", "{\"a\":1}\n"); + let client = TranscriptClient::new(&Url::parse(&server.uri()).unwrap()).unwrap(); + + let outcome = client.upload_file(&session(), &file).await.unwrap(); + assert_eq!( + outcome, + UploadOutcome { + deduped: false, + records: 3, + } + ); + + let detailed = client + .upload_file_detailed(&session(), &file) + .await + .unwrap(); + assert_eq!(detailed.deduped(), None); + assert_eq!(detailed.records(), Some(3)); + } + + #[tokio::test] + async fn a_malformed_records_type_does_not_discard_known_dedup_status() { + let server = ingest_server( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":true,"records":"bad"}"#), + ) + .await; + let dir = tempfile::tempdir().unwrap(); + let file = write_jsonl(dir.path(), "sid-1.jsonl", "{\"a\":1}\n"); + let client = TranscriptClient::new(&Url::parse(&server.uri()).unwrap()).unwrap(); + + let detailed = client + .upload_file_detailed(&session(), &file) + .await + .unwrap(); + assert_eq!(detailed.deduped(), Some(true)); + assert_eq!(detailed.records(), None); + + assert_eq!( + client.upload_file(&session(), &file).await.unwrap(), + UploadOutcome { + deduped: true, + records: 0, + }, + ); + } + + #[tokio::test] + async fn a_malformed_dedup_type_does_not_discard_known_record_count() { + let server = ingest_server( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":"bad","records":3}"#), + ) + .await; + let dir = tempfile::tempdir().unwrap(); + let file = write_jsonl(dir.path(), "sid-1.jsonl", "{\"a\":1}\n"); + let client = TranscriptClient::new(&Url::parse(&server.uri()).unwrap()).unwrap(); + + let detailed = client + .upload_file_detailed(&session(), &file) + .await + .unwrap(); + assert_eq!(detailed.deduped(), None); + assert_eq!(detailed.records(), Some(3)); + + assert_eq!( + client.upload_file(&session(), &file).await.unwrap(), + UploadOutcome { + deduped: false, + records: 3, + }, + ); + } + #[tokio::test] async fn the_records_array_reaches_the_server_with_key_order_intact() { // The dedup key is a hash of these exact bytes, so a re-serialization @@ -305,6 +489,17 @@ mod tests { let outcome = client.upload_file(&session(), &file).await.unwrap(); assert!(!outcome.deduped); + assert_eq!( + outcome.records, 0, + "the public compatibility default remains numeric" + ); + + let detailed = client + .upload_file_detailed(&session(), &file) + .await + .unwrap(); + assert_eq!(detailed.deduped(), None); + assert_eq!(detailed.records(), None); } #[tokio::test] diff --git a/crates/tapesctl/src/transcript/codex_anchors.rs b/crates/tapesctl/src/transcript/codex_anchors.rs index 60054ed..772a889 100644 --- a/crates/tapesctl/src/transcript/codex_anchors.rs +++ b/crates/tapesctl/src/transcript/codex_anchors.rs @@ -308,12 +308,13 @@ impl CodexAnchorLane { return false; }; let payload = build_anchor_payload(rollout, anchor, HARNESS_ID_CODEX, &records); - match self.client.post_transcript(&payload).await { + match self.client.post_transcript_detailed(&payload).await { Ok(outcome) => { debug!( thread_id = %anchor.thread_id, call_id = %anchor.call_id, - deduped = outcome.deduped, + deduped = %outcome.deduped_for_log(), + records = %outcome.records_for_log(), "codex spawn anchor pushed", ); true diff --git a/crates/tapesctl/src/transcript/sync.rs b/crates/tapesctl/src/transcript/sync.rs index 5a99b6e..a2cc684 100644 --- a/crates/tapesctl/src/transcript/sync.rs +++ b/crates/tapesctl/src/transcript/sync.rs @@ -25,6 +25,7 @@ //! `--since` bounds the sweep for cost, not correctness: a long-lived transcript //! tree is a lot of pointless dedups at every run. Widening it is always safe. +use std::collections::HashSet; use std::path::PathBuf; use std::time::Duration; @@ -34,7 +35,7 @@ use tapes_harnesses::transcript::{SweepOptions, TranscriptSession, sweep}; use tracing::{info, warn}; use url::Url; -use super::client::TranscriptClient; +use super::client::{DetailedUploadOutcome, TranscriptClient}; use super::tailer::default_projects_root; use crate::cli::SyncArgs; use crate::error::{Error, Result, error}; @@ -60,16 +61,156 @@ pub struct SyncSummary { } impl SyncSummary { - /// A one-line human summary. + /// Successful files whose acknowledgement did not identify their outcome. + /// Derived from the compatibility fields so aliases can never drift. + fn unavailable(&self) -> usize { + self.files.saturating_sub( + self.stored + .saturating_add(self.deduped) + .saturating_add(self.failed), + ) + } + + /// The human summary of upload outcomes. #[must_use] pub fn render(&self) -> String { + let version_label = if self.stored == 1 { + "new version" + } else { + "new versions" + }; + let unavailable = self.unavailable(); + let unavailable_suffix = if unavailable == 0 { + String::new() + } else { + format!(", {unavailable} outcome(s) unavailable") + }; format!( - "tapesctl: swept {} session(s), {} file(s): {} stored, {} deduped, {} failed", - self.sessions, self.files, self.stored, self.deduped, self.failed, + "tapesctl: swept {} session(s), {} file(s): {} {}, {} already present, {} failed{}", + self.sessions, + self.files, + self.stored, + version_label, + self.deduped, + self.failed, + unavailable_suffix, ) } } +/// Server outcome for one offered transcript file. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FileOutcome { + /// The server accepted a new content version. + New, + /// The exact content version was already present. + AlreadyPresent, + /// The server succeeded but did not report whether the content was new. + Unavailable, + /// No successful server response was received. + Failed, +} + +impl FileOutcome { + fn from_acknowledgement( + summary: &mut SyncSummary, + details: DetailedUploadOutcome, + ) -> (Option, Self) { + let outcome = match details.deduped() { + Some(true) => { + summary.deduped += 1; + Self::AlreadyPresent + } + Some(false) => { + summary.stored += 1; + Self::New + } + None => Self::Unavailable, + }; + (details.records(), outcome) + } + + fn as_str(self) -> &'static str { + match self { + Self::New => "new", + Self::AlreadyPresent => "already present", + Self::Unavailable => "unavailable (dedup status unavailable)", + Self::Failed => "failed", + } + } +} + +/// Testable account of one offered file. +#[derive(Debug, Clone, PartialEq, Eq)] +struct FileReport { + /// Harness session id carried in the upload envelope. + pub harness_session_id: String, + /// Transcript path offered to the server. + pub path: PathBuf, + /// Record count from the server, independent of whether it reported dedup status. + pub records: Option, + /// Whether this version was new, already present, or failed. + pub outcome: FileOutcome, +} + +impl FileReport { + /// The detail line printed by `sync -v`. + #[must_use] + pub fn render(&self) -> String { + let records = self + .records + .map_or_else(|| "unavailable".to_owned(), |records| records.to_string()); + format!( + "tapesctl: sync file: session {}, path {}, server records {}, outcome {}", + self.harness_session_id, + self.path.display(), + records, + self.outcome.as_str(), + ) + } +} + +/// Complete, renderable report for one sync sweep. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct SyncReport { + /// Aggregate outcomes. + summary: SyncSummary, + /// Unique harness sessions with at least one successful response. + queued_sessions: usize, + /// One outcome per offered file, in sweep order. + files: Vec, +} + +impl SyncReport { + /// Render user output. Per-file detail is a deliberate `-v` behavior, not + /// an incidental consequence of whichever tracing filter is installed. + #[must_use] + pub fn render(&self, verbosity: u8) -> Vec { + let mut lines = if verbosity > 0 { + self.files.iter().map(FileReport::render).collect() + } else { + Vec::new() + }; + lines.push(self.summary.render()); + lines.push(format!( + "tapesctl: projection queued asynchronously for {} unique session(s)", + self.queued_sessions, + )); + lines + } + + /// Preserve sync's nonzero exit behavior after all report lines print. + pub fn ensure_complete(&self) -> Result<()> { + if self.summary.failed > 0 { + return Err(Error::SyncIncomplete { + failed: self.summary.failed, + files: self.summary.files, + }); + } + Ok(()) + } +} + /// Resolved configuration for one `tapesctl sync`. #[derive(Debug, Clone)] pub struct SyncConfig { @@ -120,8 +261,13 @@ impl SyncConfig { } } -/// Run one sweep. +/// Run one sweep through the compatibility API. pub async fn run(args: SyncArgs) -> Result<()> { + run_with_verbosity(args, 0).await +} + +/// Run one CLI sweep with the global verbosity resolved by dispatch. +pub(crate) async fn run_with_verbosity(args: SyncArgs, verbosity: u8) -> Result<()> { let config = SyncConfig::resolve(args)?; let client = TranscriptClient::new(&config.ingest_url)?; info!( @@ -130,28 +276,29 @@ pub async fn run(args: SyncArgs) -> Result<()> { "sweeping transcripts", ); - let summary = sweep_into(&client, &config).await; - println!("{}", summary.render()); + let report = sweep_report(&client, &config).await; + for line in report.render(verbosity) { + println!("{line}"); + } // A partial failure is still a failure for an explicitly invoked command: // unlike background capture — which must never take the harness down — the // user ran this to move data and deserves a non-zero exit if it did not // all move. Everything that *did* land is already durable. - if summary.failed > 0 { - return Err(Error::SyncIncomplete { - failed: summary.failed, - files: summary.files, - }); - } - Ok(()) + report.ensure_complete() } -/// Sweep and push, collecting tallies. Split from [`run`] so tests can drive it -/// without going through argument resolution or stdout. +/// Sweep and push, collecting compatibility tallies. pub async fn sweep_into(client: &TranscriptClient, config: &SyncConfig) -> SyncSummary { - let mut summary = SyncSummary::default(); + sweep_report(client, config).await.summary +} + +/// Sweep and push while retaining acknowledgement detail for CLI rendering. +async fn sweep_report(client: &TranscriptClient, config: &SyncConfig) -> SyncReport { + let mut report = SyncReport::default(); + let mut queued_sessions = HashSet::new(); let swept = sweep(&config.projects_root, &config.sweep_options()); - summary.sessions = swept.len(); + report.summary.sessions = swept.len(); for session in swept { // The envelope is rebuilt from the transcript's own records — a swept @@ -163,22 +310,32 @@ pub async fn sweep_into(client: &TranscriptClient, config: &SyncConfig) -> SyncS .with_auth_subject(config.auth_subject.clone()); for file in &session.files { - summary.files += 1; - match client.upload_file(&envelope, file).await { - Ok(outcome) if outcome.deduped => summary.deduped += 1, - Ok(_) => summary.stored += 1, + report.summary.files += 1; + let (records, outcome) = match client.upload_file_detailed(&envelope, file).await { + Ok(details) => { + queued_sessions.insert(session.session_id.clone()); + FileOutcome::from_acknowledgement(&mut report.summary, details) + } Err(err) => { warn!( error = %err, file = %file.label(&session.session_id), "transcript push failed", ); - summary.failed += 1; + report.summary.failed += 1; + (None, FileOutcome::Failed) } - } + }; + report.files.push(FileReport { + harness_session_id: session.session_id.clone(), + path: file.path.clone(), + records, + outcome, + }); } } - summary + report.queued_sessions = queued_sessions.len(); + report } #[cfg(test)] @@ -186,7 +343,7 @@ pub async fn sweep_into(client: &TranscriptClient, config: &SyncConfig) -> SyncS mod tests { use super::*; use tapes_harnesses::attribution::claude::fork_parent::encode_cwd; - use wiremock::matchers::{method, path}; + use wiremock::matchers::{body_string_contains, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; fn args() -> SyncArgs { @@ -266,28 +423,129 @@ mod tests { assert!(subject.starts_with("local:"), "got: {subject}"); } + #[test] + fn summary_distinguishes_new_versions_from_already_present_files() { + // This literal is also a source-compatibility regression check for the + // original public fields. + let summary = SyncSummary { + sessions: 2, + files: 3, + stored: 2, + deduped: 1, + failed: 0, + }; + + let rendered = summary.render(); + assert_eq!( + rendered, + "tapesctl: swept 2 session(s), 3 file(s): 2 new versions, 1 already present, 0 failed", + ); + assert!(!rendered.contains("stored"), "got: {rendered}"); + assert!(!rendered.contains("deduped"), "got: {rendered}"); + } + #[tokio::test] - async fn every_transcript_in_the_tree_is_offered_including_subagents() { - let server = - server_replying(ResponseTemplate::new(202).set_body_string(r#"{"records":1}"#)).await; + async fn queued_projection_counts_unique_successful_sessions_not_files() { + let server = server_replying( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":false,"records":1}"#), + ) + .await; let tree = tempfile::tempdir().unwrap(); write_session(tree.path(), "/tmp/one", "sid-1", &["a1"]); write_session(tree.path(), "/tmp/two", "sid-2", &[]); let config = config_for(&server, tree.path().to_path_buf()); let client = TranscriptClient::new(&config.ingest_url).unwrap(); - let summary = sweep_into(&client, &config).await; + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.sessions, 2); + assert_eq!(report.summary.files, 3, "two mains plus one subagent"); + assert_eq!(report.summary.stored, 3); + assert_eq!(report.summary.failed, 0); + assert_eq!(report.queued_sessions, 2); + assert_eq!( + report.render(0)[1], + "tapesctl: projection queued asynchronously for 2 unique session(s)", + ); + } + + #[test] + fn verbose_output_reports_every_file_session_path_records_and_outcome() { + let report = SyncReport { + summary: SyncSummary::default(), + queued_sessions: 2, + files: vec![ + FileReport { + harness_session_id: "sid-new".to_owned(), + path: PathBuf::from("/tmp/new.jsonl"), + records: Some(3), + outcome: FileOutcome::New, + }, + FileReport { + harness_session_id: "sid-present".to_owned(), + path: PathBuf::from("/tmp/present.jsonl"), + records: Some(5), + outcome: FileOutcome::AlreadyPresent, + }, + FileReport { + harness_session_id: "sid-failed".to_owned(), + path: PathBuf::from("/tmp/failed.jsonl"), + records: None, + outcome: FileOutcome::Failed, + }, + ], + }; + + let lines = report.render(1); + assert_eq!(lines.len(), 5, "three files plus two summary lines"); + assert!(lines[0].contains("session sid-new"), "got: {}", lines[0]); + assert!( + lines[0].contains("path /tmp/new.jsonl"), + "got: {}", + lines[0] + ); + assert!(lines[0].contains("server records 3"), "got: {}", lines[0]); + assert!(lines[0].ends_with("outcome new"), "got: {}", lines[0]); + assert!( + lines[1].ends_with("outcome already present"), + "got: {}", + lines[1], + ); + assert!( + lines[2].contains("server records unavailable") && lines[2].ends_with("outcome failed"), + "got: {}", + lines[2], + ); + } + + #[test] + fn normal_output_omits_per_file_success_detail() { + let report = SyncReport { + summary: SyncSummary { + sessions: 1, + files: 1, + stored: 1, + deduped: 0, + failed: 0, + }, + queued_sessions: 1, + files: vec![FileReport { + harness_session_id: "sid-1".to_owned(), + path: PathBuf::from("/tmp/sid-1.jsonl"), + records: Some(4), + outcome: FileOutcome::New, + }], + }; - assert_eq!(summary.sessions, 2); - assert_eq!(summary.files, 3, "two mains plus one subagent"); - assert_eq!(summary.stored, 3); - assert_eq!(summary.failed, 0); + let lines = report.render(0); + assert_eq!(lines.len(), 2, "only summary and projection status"); + assert!(lines.iter().all(|line| !line.contains("/tmp/sid-1.jsonl"))); } #[tokio::test] - async fn a_dedup_is_counted_as_success_not_failure() { + async fn a_dedup_is_success_and_requeues_projection() { // Re-running sync over an already-synced tree is the expected steady - // state, and it must exit zero. + // state, and it must exit zero. Ingest also requeues projection. let server = server_replying( ResponseTemplate::new(202).set_body_string(r#"{"deduped":true,"records":1}"#), ) @@ -297,11 +555,104 @@ mod tests { let config = config_for(&server, tree.path().to_path_buf()); let client = TranscriptClient::new(&config.ingest_url).unwrap(); - let summary = sweep_into(&client, &config).await; + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.deduped, 1); + assert_eq!(report.summary.stored, 0); + assert_eq!(report.summary.failed, 0); + assert_eq!(report.queued_sessions, 1); + assert!(report.ensure_complete().is_ok()); + } + + #[tokio::test] + async fn verbose_output_keeps_a_known_dedup_outcome_when_records_are_missing() { + let server = + server_replying(ResponseTemplate::new(202).set_body_string(r#"{"deduped":true}"#)) + .await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/one", "sid-1", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.deduped, 1); + assert_eq!(report.summary.stored, 0); + let detail = report.files[0].render(); + assert!( + detail.contains("server records unavailable"), + "got: {detail}" + ); + assert!(detail.ends_with("outcome already present"), "got: {detail}"); + } + + #[tokio::test] + async fn verbose_output_keeps_a_known_record_count_when_dedup_status_is_missing() { + let server = + server_replying(ResponseTemplate::new(202).set_body_string(r#"{"records":3}"#)).await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/one", "sid-1", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.deduped, 0); + assert_eq!(report.summary.stored, 0); + assert_eq!(report.summary.unavailable(), 1); + let detail = report.files[0].render(); + assert!(detail.contains("server records 3"), "got: {detail}"); + assert!( + detail.ends_with("outcome unavailable (dedup status unavailable)"), + "got: {detail}" + ); + } - assert_eq!(summary.deduped, 1); - assert_eq!(summary.stored, 0); - assert_eq!(summary.failed, 0); + #[tokio::test] + async fn verbose_output_keeps_dedup_when_the_records_type_is_malformed() { + let server = server_replying( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":true,"records":"bad"}"#), + ) + .await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/one", "sid-1", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.deduped, 1); + assert_eq!(report.summary.stored, 0); + let detail = report.files[0].render(); + assert!( + detail.contains("server records unavailable"), + "got: {detail}" + ); + assert!(detail.ends_with("outcome already present"), "got: {detail}"); + } + + #[tokio::test] + async fn verbose_output_keeps_records_when_the_dedup_type_is_malformed() { + let server = server_replying( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":"bad","records":3}"#), + ) + .await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/one", "sid-1", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.deduped, 0); + assert_eq!(report.summary.stored, 0); + assert_eq!(report.summary.unavailable(), 1); + let detail = report.files[0].render(); + assert!(detail.contains("server records 3"), "got: {detail}"); + assert!( + detail.ends_with("outcome unavailable (dedup status unavailable)"), + "got: {detail}" + ); } #[tokio::test] @@ -331,18 +682,101 @@ mod tests { } #[tokio::test] - async fn a_rejected_file_is_counted_as_failed() { - let server = - server_replying(ResponseTemplate::new(400).set_body_string("bad envelope")).await; + async fn partial_failure_queues_only_successful_sessions_and_still_exits_nonzero() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/ingest/transcript")) + .and(body_string_contains("sid-ok")) + .respond_with( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":false,"records":2}"#), + ) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/v1/ingest/transcript")) + .and(body_string_contains("sid-failed")) + .respond_with(ResponseTemplate::new(400).set_body_string("bad envelope")) + .mount(&server) + .await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/ok", "sid-ok", &[]); + write_session(tree.path(), "/tmp/failed", "sid-failed", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.files, 2); + assert_eq!(report.summary.stored, 1); + assert_eq!(report.summary.failed, 1); + assert_eq!(report.queued_sessions, 1); + assert!(matches!( + report.ensure_complete(), + Err(Error::SyncIncomplete { + failed: 1, + files: 2, + }), + )); + } + + async fn assert_acknowledgement_unavailable(template: ResponseTemplate) { + let server = server_replying(template).await; let tree = tempfile::tempdir().unwrap(); write_session(tree.path(), "/tmp/one", "sid-1", &[]); let config = config_for(&server, tree.path().to_path_buf()); let client = TranscriptClient::new(&config.ingest_url).unwrap(); - let summary = sweep_into(&client, &config).await; + let report = sweep_report(&client, &config).await; + + assert_eq!(report.summary.files, 1); + assert_eq!(report.summary.stored, 0, "unknown is not new"); + assert_eq!(report.summary.deduped, 0, "unknown is not already present"); + assert_eq!(report.summary.failed, 0, "the 2xx remains successful"); + assert_eq!(report.queued_sessions, 1); + assert!(report.summary.render().contains("1 outcome(s) unavailable")); + let detail = report.files[0].render(); + assert!( + detail.contains("server records unavailable"), + "got: {detail}" + ); + assert!( + detail.contains("outcome unavailable (dedup status unavailable)"), + "got: {detail}", + ); + assert!(report.ensure_complete().is_ok()); + } + + #[tokio::test] + async fn an_unparseable_acknowledgement_is_successful_but_not_reported_as_new() { + assert_acknowledgement_unavailable(ResponseTemplate::new(202).set_body_string("not json")) + .await; + } - assert_eq!(summary.failed, 1); - assert_eq!(summary.stored, 0); + #[tokio::test] + async fn a_missing_acknowledgement_is_successful_but_not_reported_as_new() { + assert_acknowledgement_unavailable(ResponseTemplate::new(204)).await; + } + + #[tokio::test] + async fn public_sweep_into_still_returns_the_compatibility_summary() { + let server = server_replying( + ResponseTemplate::new(202).set_body_string(r#"{"deduped":false,"records":1}"#), + ) + .await; + let tree = tempfile::tempdir().unwrap(); + write_session(tree.path(), "/tmp/one", "sid-1", &[]); + + let config = config_for(&server, tree.path().to_path_buf()); + let client = TranscriptClient::new(&config.ingest_url).unwrap(); + let summary: SyncSummary = sweep_into(&client, &config).await; + + assert_eq!(summary.stored, 1); + assert_eq!(summary.deduped, 0); + } + + #[test] + fn public_run_still_accepts_only_sync_args() { + std::mem::drop(run(args())); } #[tokio::test] diff --git a/crates/tapesctl/src/transcript/tailer.rs b/crates/tapesctl/src/transcript/tailer.rs index 9f95738..6d438a0 100644 --- a/crates/tapesctl/src/transcript/tailer.rs +++ b/crates/tapesctl/src/transcript/tailer.rs @@ -362,14 +362,14 @@ impl Tailer { .iter() .find(|(path, _)| path == &file.path) .map(|(path, fp)| (path.clone(), *fp)); - match self.client.upload_file(&envelope, file).await { + match self.client.upload_file_detailed(&envelope, file).await { Ok(outcome) => { debug!( session = %session.session_id, file = %file.label(&session.session_id), reason = reason.as_str(), - deduped = outcome.deduped, - records = outcome.records, + deduped = %outcome.deduped_for_log(), + records = %outcome.records_for_log(), "transcript pushed", ); // The fingerprint taken *before* the read is what gets diff --git a/crates/tapesctl/tests/public_compatibility.rs b/crates/tapesctl/tests/public_compatibility.rs new file mode 100644 index 0000000..2176c11 --- /dev/null +++ b/crates/tapesctl/tests/public_compatibility.rs @@ -0,0 +1,41 @@ +//! Compile-time regression checks for the transcript library surface. + +use std::future::Future; + +use tapesctl::cli::SyncArgs; +use tapesctl::transcript::client::{TranscriptClient, UploadOutcome}; +use tapesctl::transcript::sync::{SyncConfig, SyncSummary}; + +#[test] +fn transcript_outcomes_and_sync_keep_their_original_public_shapes() { + let outcome = UploadOutcome { + deduped: true, + records: 7, + }; + let UploadOutcome { deduped, records } = outcome; + assert!(deduped); + assert_eq!(records, 7); + + let summary = SyncSummary { + sessions: 1, + files: 2, + stored: 1, + deduped: 1, + failed: 0, + }; + assert_eq!(summary.stored + summary.deduped, summary.files); + + #[allow(clippy::result_large_err)] + fn run_compat(args: SyncArgs) -> impl Future> { + tapesctl::transcript::sync::run(args) + } + fn sweep_compat<'a>( + client: &'a TranscriptClient, + config: &'a SyncConfig, + ) -> impl Future + 'a { + tapesctl::transcript::sync::sweep_into(client, config) + } + + let _ = run_compat; + let _ = sweep_compat; +} diff --git a/docs/capture.md b/docs/capture.md index 69ba69c..b0b0c13 100644 --- a/docs/capture.md +++ b/docs/capture.md @@ -79,11 +79,26 @@ tapesctl sync --ingest-url http://localhost:8082 ``` ``` -tapesctl: swept 2 session(s), 2 file(s): 2 stored, 0 deduped, 0 failed +tapesctl: swept 2 session(s), 2 file(s): 2 new versions, 0 already present, 0 failed +tapesctl: projection queued asynchronously for 2 unique session(s) ``` -Unlike `start`, `sync` logs to stderr as usual — only `start` diverts its -diagnostics to a file, and only because a harness owns the terminal. +The second line is deliberately narrower than "projected." Every session with +at least one accepted or already-present file has projection queued by ingest, +but that work runs asynchronously. `sync` does not poll the read API, and a +read immediately afterwards may still show the previous projection. An +already-present file is still a successful upload and requeues projection +server-side. + +Global `-v` adds one line per file with its harness session id, path, the record +count reported by the server, and outcome (`new`, `already present`, `failed`, +or `unavailable`). A failed request has no server count. For a successful +response, the acknowledgement fields are independent: an omitted record count +prints `server records unavailable` without hiding a known outcome, while an +omitted dedup status prints `outcome unavailable` without hiding a known count. +Normal mode omits successful per-file detail. Unlike `start`, `sync` logs to +stderr as usual — only `start` diverts its diagnostics to a file, and only +because a harness owns the terminal. Two things about `sync` are not visible from its help text: @@ -98,13 +113,28 @@ Two things about `sync` are not visible from its help text: the flag name suggests. Re-running `sync` is cheap and safe: the ingest endpoint keys rows on a content -hash, so an unchanged transcript comes back `deduped`. `tapesctl` keeps no -client-side ledger of what it has already sent, by design — a ledger that -disagreed with the server would be worse than no ledger. +hash, so the summary counts unchanged content as `already present`. `tapesctl` +keeps no client-side ledger of what it has already sent, by design — a ledger +that disagreed with the server would be worse than no ledger. + +Syncing an old transcript is **partial historical reconstruction**, not a +replacement for capture. When a session has no usable wire calls, its +transcript can still produce partial, browsable LLM calls and causal structure. +Those transcript-derived calls lack full wire fidelity: exact provider requests +and response bytes are unavailable, and the harness transcript can omit some +side calls or context. + +The fallback is session-wide. If usable wire capture arrives later, projection +switches to the wire call inventory; the transcript continues to reconcile +causal structure, while wire-derived calls replace and prune the partial +transcript projection instead of appearing beside it as duplicates. Any undelivered transcript makes `sync` exit `1`, deliberately, because `sync` -is an explicit request to move data. The summary line still prints first, and -everything that did land is durable. +is an explicit request to move data. Both status lines still print before the +command returns its final error, and everything that did land is durable. The +queued-session count includes only +unique sessions with at least one successful response; a wholly failed session +is not claimed as queued. ## Which harness uses which mechanism diff --git a/docs/commands.md b/docs/commands.md index 6ff1596..9cea5f4 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -20,7 +20,7 @@ subcommand and reach every leaf. | flag | type | default | notes | |---|---|---|---| -| `-v`, `--verbose` | count | `0` | `-v` is `debug`, `-vv` is `trace`. `RUST_LOG` overrides both | +| `-v`, `--verbose` | count | `0` | `-v` adds sync file outcomes and enables `debug`; `-vv` enables `trace`. `RUST_LOG` overrides the log level, not sync detail | | `--api-url ` | string | `http://localhost:8081` | falls back to `TAPES_API_URL`, then `config.toml` | | `-h`, `--help` | flag | — | | | `-V`, `--version` | flag | — | prints one line; see [`version`](#version) before trusting it | @@ -229,16 +229,48 @@ is a cost bound, never a correctness one. so `--projects-root` pointed at another harness's tree will not do what the name suggests. -Prints one line: +Normal mode prints two lines: ``` -tapesctl: swept 2 session(s), 2 file(s): 2 stored, 0 deduped, 0 failed +tapesctl: swept 2 session(s), 2 file(s): 1 new version, 1 already present, 0 failed +tapesctl: projection queued asynchronously for 2 unique session(s) ``` +The first line distinguishes accepted new content versions from files the +server already held. If a successful response omits dedup status, it adds an +`outcome(s) unavailable` count rather than treating that file as new. The second line does **not** mean projection completed: ingest +queued asynchronous projection for each unique session with at least one +successful response, and `sync` does not poll the read API. Reads may lag. +Deduplicated uploads are successes and requeue projection server-side; the +server remains the only source of truth, with no client upload ledger. + +Global `-v` adds one line per offered file: + +``` +tapesctl: sync file: session sid-1, path /home/me/.claude/projects/-work/sid-1.jsonl, server records 42, outcome new +tapesctl: sync file: session sid-2, path /home/me/.claude/projects/-work/sid-2.jsonl, server records 18, outcome already present +``` + +The outcomes are `new`, `already present`, `failed`, and `unavailable`. A +failure has no server-reported record count. Successful acknowledgement fields +are independent: if `records` is absent only the count is `unavailable`; if +`deduped` is absent only the outcome is `unavailable`. Sync preserves whichever +field the server did report rather than guessing or discarding both. Normal +mode omits successful per-file detail. + Any failure then exits `1` with ` of transcript(s) could not be -delivered`. The summary prints first, and everything that landed is durable. -Deduplication is entirely server-side, keyed on a content hash; a dedup counts -as a success. +delivered`. Both aggregate lines print before the final error, and everything +that landed is durable. The queued count excludes a session when all of its +files failed, but includes it once when any file received a successful response. + +A historical sync can create partial, browsable calls directly from transcript +content even when the wire proxy captured none. This is lower fidelity than +wire capture: exact provider requests and response bytes are unavailable, and +some harness-side calls or context may not appear in the transcript. Once a +usable wire call arrives, the whole session switches to wire projection; +wire-derived calls replace the transcript-derived fallback rather than +combining with it, while transcript evidence continues to supply causal +structure. ## sessions diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 8ec1602..4e7a038 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -213,7 +213,17 @@ or `--no-transcripts` was passed, or nothing was tailing when the session ran. tapesctl sync --ingest-url http://localhost:8082 --since-days 0 ``` -`sync` is safe to repeat — the server dedups on a content hash. +`sync` is safe to repeat — the server dedups on a content hash, reports the +file as `already present`, and requeues asynchronous projection. The queued +status does not mean the read model is ready; `sync` does not poll, so retry the +read after a short delay. + +If the wire proxy was not running, `sync` can still create partial, browsable +calls from the transcript. They lack full wire fidelity: exact provider +requests and response bytes are unavailable, and the transcript may omit some +harness-side calls or context. A later usable wire capture replaces this +session-wide transcript fallback rather than duplicating it; transcript +evidence remains available to restore causal structure. ## `sync` says it swept less than expected @@ -227,8 +237,23 @@ name suggests. **`sync` exited 1 but the summary looked fine.** Any undelivered transcript fails the command, deliberately, because `sync` is an explicit request to move -data. The summary prints first and everything that landed is durable — re-run -to retry the rest. +data. The aggregate upload and asynchronous-projection lines print before the +final error and everything that landed is durable — re-run to retry the rest. +The queued-session count includes only unique sessions for which at least one +file got a successful response; a session whose every file failed is excluded. + +**The session is still missing structure immediately after a successful +sync.** Projection is asynchronous. `projection queued asynchronously` means +ingest accepted the work, not that the read model is already updated. `sync` +does not poll the read API. Wait briefly and read again. + +**You need to see which file did what.** Run `tapesctl -v sync ...`. Each file +line includes its harness session id, path, server-reported record count, and +`new`, `already present`, `failed`, or `unavailable` outcome. Ack fields are +independent: a response can report a count without dedup status, or dedup status +without a count, and sync renders the known field instead of discarding both. +Normal mode omits successful per-file detail; failures still contribute to the +nonzero exit. ## `search` returns 503