From 53b9eec10f2d85a97ca7f71f00c9a1cab9de1d16 Mon Sep 17 00:00:00 2001 From: Matt Yeazel Date: Tue, 25 Aug 2026 13:08:52 -0700 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(sync):=20report=20queued=20tra?= =?UTF-8?q?nscript=20projection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Treat successful transcript acknowledgements as accepted immutable versions and distinguish new, deduplicated, and unknown outcomes. Report projection as asynchronously queued per session rather than completed, and preserve successful acknowledgements when another upload fails. Keep the public sync entry points compatible while making detailed per-file outcomes available to verbose callers. --- README.md | 21 +- crates/tapesctl/src/cli.rs | 7 +- crates/tapesctl/src/lib.rs | 3 +- crates/tapesctl/src/transcript/client.rs | 221 +++++++- .../tapesctl/src/transcript/codex_anchors.rs | 5 +- crates/tapesctl/src/transcript/sync.rs | 526 ++++++++++++++++-- crates/tapesctl/src/transcript/tailer.rs | 6 +- crates/tapesctl/tests/public_compatibility.rs | 41 ++ docs/capture.md | 46 +- docs/commands.md | 44 +- docs/troubleshooting.md | 31 +- 11 files changed, 864 insertions(+), 87 deletions(-) create mode 100644 crates/tapesctl/tests/public_compatibility.rs 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