From f86f7ac83976a38996e9cfbcaf7856792bda929b Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Wed, 23 Sep 2026 00:02:11 -0400 Subject: [PATCH 1/7] fix(ingest): accept transcript archive replay locators --- src/ai_transcript_ingest/legacy_receipt.rs | 46 ++++++++++-- src/ai_transcript_replay_tests.rs | 84 +++++++++++++++++++++- 2 files changed, 123 insertions(+), 7 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index 142a6c9ee..994b95a33 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -15,12 +15,43 @@ impl std::fmt::Display for IdempotencyConflict { impl std::error::Error for IdempotencyConflict {} fn envelope_fingerprint(envelope: &EvidenceEnvelope) -> anyhow::Result { - // Supplemental display metadata can change without a transcript revision. + // Display metadata and the file's location can change without changing + // the transcript record (for example, when Codex archives a session). let mut evidence = envelope.clone(); evidence.source.title = None; evidence.source.title_provenance = None; + evidence.source.locator.clear(); let encoded = serde_json::to_vec(&evidence)?; - Ok(format!("evidence-v2:sha256:{:x}", Sha256::digest(encoded))) + Ok(format!("evidence-v3:sha256:{:x}", Sha256::digest(encoded))) +} + +fn v2_fingerprint_matches( + tx: &rusqlite::Transaction<'_>, + key: &str, + envelope: &EvidenceEnvelope, + previous: &str, +) -> anyhow::Result { + let stored_locator: Option = tx + .query_row( + "SELECT l.ai_transcript_path FROM ai_transcript_forward_receipts r + JOIN logs l ON l.id = r.log_id WHERE r.source_record_id = ?1", + [key], + |row| row.get(0), + ) + .optional()? + .flatten(); + let Some(stored_locator) = stored_locator else { + return Ok(false); + }; + let mut original = envelope.clone(); + original.source.title = None; + original.source.title_provenance = None; + original.source.locator = stored_locator; + Ok(previous + == format!( + "evidence-v2:sha256:{:x}", + Sha256::digest(serde_json::to_vec(&original)?) + )) } fn receipt_key(forwarder_identity: &str, source_record_id: &str, shared_bearer: bool) -> String { @@ -65,6 +96,7 @@ fn old_fingerprint_matches( let mut original = envelope.clone(); original.source.title = source.title; original.source.title_provenance = source.title_provenance; + original.source.locator = source.locator; Ok(previous == format!( "sha256:{:x}", @@ -86,7 +118,7 @@ fn legacy_receipt_matches( "SELECT r.envelope_version, r.provider, r.source_identity, r.source_epoch, r.source_revision, l.timestamp, l.message, l.ai_project, l.ai_session_id, - l.ai_transcript_path, l.metadata_json + l.metadata_json FROM ai_transcript_forward_receipts r JOIN logs l ON l.id = r.log_id WHERE r.source_record_id = ?1", @@ -103,7 +135,6 @@ fn legacy_receipt_matches( row.get::<_, Option>(7)?, row.get::<_, Option>(8)?, row.get::<_, Option>(9)?, - row.get::<_, Option>(10)?, )) }, ) @@ -118,7 +149,6 @@ fn legacy_receipt_matches( message, ai_project, ai_session_id, - locator, metadata_json, )) = stored else { @@ -136,9 +166,11 @@ fn legacy_receipt_matches( let mut incoming_source = envelope.source.clone(); incoming_source.title = None; incoming_source.title_provenance = None; + incoming_source.locator.clear(); if let Some(source) = &mut stored_source { source.title = None; source.title_provenance = None; + source.locator.clear(); } let stored_capabilities = metadata .get("capabilities") @@ -171,7 +203,6 @@ fn legacy_receipt_matches( && message == envelope.message && ai_project == envelope.ai_project && ai_session_id == envelope.ai_session_id - && locator.as_deref() == Some(envelope.source.locator.as_str()) // A legacy canonical row does not record whether its timestamp came // from the source envelope or the receiver clock. Requiring the replay // to supply the stored value avoids silently binding an ambiguous @@ -257,6 +288,9 @@ fn insert_envelopes_with_identity( // immutable evidence against the canonical row, never merely // because the caller reused an existing source-record ID. let matches = match previous_fingerprint.as_deref() { + Some(previous) if previous.starts_with("evidence-v2:sha256:") => { + v2_fingerprint_matches(&tx, &stored_receipt_key, &envelope, previous)? + } Some(previous) if previous.starts_with("sha256:") => { old_fingerprint_matches(&tx, &stored_receipt_key, &envelope, previous)? } diff --git a/src/ai_transcript_replay_tests.rs b/src/ai_transcript_replay_tests.rs index a37462f83..3b5905870 100644 --- a/src/ai_transcript_replay_tests.rs +++ b/src/ai_transcript_replay_tests.rs @@ -1,5 +1,86 @@ use super::*; +#[tokio::test] +async fn archived_transcript_replay_accepts_a_changed_locator() { + use sha2::{Digest, Sha256}; + + let (app, dir) = test_app(Some("secret")); + let original = sample_record(); + let envelope: EvidenceEnvelope = serde_json::from_value(original["envelope"].clone()).unwrap(); + let mut v2_evidence = scrub_envelope(envelope).unwrap(); + v2_evidence.source.title = None; + v2_evidence.source.title_provenance = None; + let v2_fingerprint = format!( + "evidence-v2:sha256:{:x}", + Sha256::digest(serde_json::to_vec(&v2_evidence).unwrap()) + ); + + let first = app + .clone() + .oneshot(transcript_request( + json!({"records": [original]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::OK); + + let conn = rusqlite::Connection::open(dir.path().join("ai-transcript-ingest-test.db")).unwrap(); + conn.execute( + "UPDATE ai_transcript_forward_receipts SET request_fingerprint = ?1", + [&v2_fingerprint], + ) + .unwrap(); + + let mut archived = sample_record(); + archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); + let replay = app + .clone() + .oneshot(transcript_request( + json!({"records": [archived.clone()]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(replay.status(), StatusCode::OK); + let body = axum::body::to_bytes(replay.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["receipts"][0]["disposition"], "duplicate"); + + archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); + let second_replay = app + .clone() + .oneshot(transcript_request( + json!({"records": [archived]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(second_replay.status(), StatusCode::OK); + let fingerprint: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(fingerprint.starts_with("evidence-v3:sha256:")); + + let mut changed_evidence = sample_record(); + changed_evidence["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); + changed_evidence["envelope"]["message"] = json!("different evidence"); + let conflict = app + .oneshot(transcript_request( + json!({"records": [changed_evidence]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(conflict.status(), StatusCode::CONFLICT); + let log_count: i64 = conn + .query_row("SELECT COUNT(*) FROM logs", [], |row| row.get(0)) + .unwrap(); + assert_eq!(log_count, 1); +} + #[tokio::test] async fn title_changes_and_missing_metadata_are_duplicate_replays() { let (app, dir) = test_app(Some("secret")); @@ -30,7 +111,7 @@ async fn title_changes_and_missing_metadata_are_duplicate_replays() { } #[tokio::test] -async fn old_full_envelope_receipt_accepts_title_only_changes_but_not_evidence_changes() { +async fn old_full_envelope_receipt_accepts_title_and_locator_changes_but_not_evidence_changes() { check_old_receipt_replay(false).await; check_old_receipt_replay(true).await; } @@ -93,6 +174,7 @@ async fn check_old_receipt_replay(timestamp_absent: bool) { replay["envelope"]["timestamp"] = serde_json::Value::Null; } replay["envelope"]["source"]["title"] = json!("Renamed"); + replay["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); let renamed = app .clone() .oneshot(transcript_request( From 19f6530511e2fbffb464dae56ea9228bd89fb138 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:11:26 -0400 Subject: [PATCH 2/7] fix: preserve rollback-safe transcript receipt fingerprints --- src/ai_transcript_ingest/legacy_receipt.rs | 154 ++++++++++++---- src/ai_transcript_replay_tests.rs | 202 +++++++++++++++++---- 2 files changed, 281 insertions(+), 75 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index 994b95a33..901eb253e 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -14,24 +14,44 @@ impl std::fmt::Display for IdempotencyConflict { impl std::error::Error for IdempotencyConflict {} +fn v2_fingerprint_with_locator( + envelope: &EvidenceEnvelope, + locator: &str, +) -> anyhow::Result { + // Supplemental display metadata can change without a transcript revision. + // Keep the persisted v2 format rollback-compatible with older Cortex + // releases; locator mutability is handled during replay verification. + let mut evidence = envelope.clone(); + evidence.source.title = None; + evidence.source.title_provenance = None; + evidence.source.locator = locator.to_string(); + let encoded = serde_json::to_vec(&evidence)?; + Ok(format!("evidence-v2:sha256:{:x}", Sha256::digest(encoded))) +} + fn envelope_fingerprint(envelope: &EvidenceEnvelope) -> anyhow::Result { - // Display metadata and the file's location can change without changing - // the transcript record (for example, when Codex archives a session). + v2_fingerprint_with_locator(envelope, &envelope.source.locator) +} + +fn transient_v3_fingerprint(envelope: &EvidenceEnvelope) -> anyhow::Result { + // A pre-review patched deployment briefly wrote v3 receipts whose only + // semantic difference was excluding the movable locator. Continue to read + // those receipts and lazily rebind them to the durable v2 format. let mut evidence = envelope.clone(); evidence.source.title = None; evidence.source.title_provenance = None; evidence.source.locator.clear(); - let encoded = serde_json::to_vec(&evidence)?; - Ok(format!("evidence-v3:sha256:{:x}", Sha256::digest(encoded))) + Ok(format!( + "evidence-v3:sha256:{:x}", + Sha256::digest(serde_json::to_vec(&evidence)?) + )) } -fn v2_fingerprint_matches( +fn stored_locator( tx: &rusqlite::Transaction<'_>, key: &str, - envelope: &EvidenceEnvelope, - previous: &str, -) -> anyhow::Result { - let stored_locator: Option = tx +) -> anyhow::Result> { + Ok(tx .query_row( "SELECT l.ai_transcript_path FROM ai_transcript_forward_receipts r JOIN logs l ON l.id = r.log_id WHERE r.source_record_id = ?1", @@ -39,19 +59,34 @@ fn v2_fingerprint_matches( |row| row.get(0), ) .optional()? - .flatten(); - let Some(stored_locator) = stored_locator else { - return Ok(false); + .flatten()) +} + +fn canonical_v2_fingerprint( + tx: &rusqlite::Transaction<'_>, + key: &str, + envelope: &EvidenceEnvelope, +) -> anyhow::Result> { + let Some(locator) = stored_locator(tx, key)? else { + return Ok(None); }; - let mut original = envelope.clone(); - original.source.title = None; - original.source.title_provenance = None; - original.source.locator = stored_locator; - Ok(previous - == format!( - "evidence-v2:sha256:{:x}", - Sha256::digest(serde_json::to_vec(&original)?) - )) + Ok(Some(v2_fingerprint_with_locator(envelope, &locator)?)) +} + +fn v2_fingerprint_matches( + tx: &rusqlite::Transaction<'_>, + key: &str, + envelope: &EvidenceEnvelope, + previous: &str, +) -> anyhow::Result { + Ok(canonical_v2_fingerprint(tx, key, envelope)?.as_deref() == Some(previous)) +} + +fn transient_v3_fingerprint_matches( + envelope: &EvidenceEnvelope, + previous: &str, +) -> anyhow::Result { + Ok(previous == transient_v3_fingerprint(envelope)?) } fn receipt_key(forwarder_identity: &str, source_record_id: &str, shared_bearer: bool) -> String { @@ -284,28 +319,67 @@ fn insert_envelopes_with_identity( .optional()?; if let Some(previous_fingerprint) = already_accepted { if previous_fingerprint.as_deref() != Some(request_fingerprint.as_str()) { - // Old hashes included titles. Rebind only after comparing all - // immutable evidence against the canonical row, never merely - // because the caller reused an existing source-record ID. - let matches = match previous_fingerprint.as_deref() { - Some(previous) if previous.starts_with("evidence-v2:sha256:") => { - v2_fingerprint_matches(&tx, &stored_receipt_key, &envelope, previous)? - } - Some(previous) if previous.starts_with("sha256:") => { - old_fingerprint_matches(&tx, &stored_receipt_key, &envelope, previous)? - } - Some(_) => false, - None => legacy_receipt_matches(&tx, &stored_receipt_key, &envelope)?, - }; + // Compatibility fingerprints are validated against the + // canonical row before any rebinding. Persist v2 so a rollback + // to the previous Cortex release can still read receipts + // created by this version. + let (matches, replacement_fingerprint) = + match previous_fingerprint.as_deref() { + Some(previous) if previous.starts_with("evidence-v3:sha256:") => { + let matches = transient_v3_fingerprint_matches(&envelope, previous)?; + let replacement = if matches { + canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + } else { + None + }; + (matches, replacement) + } + Some(previous) if previous.starts_with("evidence-v2:sha256:") => ( + v2_fingerprint_matches( + &tx, + &stored_receipt_key, + &envelope, + previous, + )?, + None, + ), + Some(previous) if previous.starts_with("sha256:") => { + let matches = old_fingerprint_matches( + &tx, + &stored_receipt_key, + &envelope, + previous, + )?; + let replacement = if matches { + canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + } else { + None + }; + (matches, replacement) + } + Some(_) => (false, None), + None => { + let matches = + legacy_receipt_matches(&tx, &stored_receipt_key, &envelope)?; + let replacement = if matches { + canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + } else { + None + }; + (matches, replacement) + } + }; if !matches { return Err(IdempotencyConflict.into()); } - tx.execute( - "UPDATE ai_transcript_forward_receipts - SET request_fingerprint = ?2 - WHERE source_record_id = ?1", - rusqlite::params![stored_receipt_key, request_fingerprint], - )?; + if let Some(replacement_fingerprint) = replacement_fingerprint { + tx.execute( + "UPDATE ai_transcript_forward_receipts + SET request_fingerprint = ?2 + WHERE source_record_id = ?1", + rusqlite::params![stored_receipt_key, replacement_fingerprint], + )?; + } } receipts.push(AiTranscriptReceipt { source_record_id: envelope.source_record_id, diff --git a/src/ai_transcript_replay_tests.rs b/src/ai_transcript_replay_tests.rs index 3b5905870..fc2f616c7 100644 --- a/src/ai_transcript_replay_tests.rs +++ b/src/ai_transcript_replay_tests.rs @@ -1,19 +1,116 @@ use super::*; -#[tokio::test] -async fn archived_transcript_replay_accepts_a_changed_locator() { +fn receipt_v2_fingerprint(record: &serde_json::Value) -> String { + use sha2::{Digest, Sha256}; + + let envelope: EvidenceEnvelope = + serde_json::from_value(record["envelope"].clone()).unwrap(); + let mut evidence = scrub_envelope(envelope).unwrap(); + evidence.source.title = None; + evidence.source.title_provenance = None; + format!( + "evidence-v2:sha256:{:x}", + Sha256::digest(serde_json::to_vec(&evidence).unwrap()) + ) +} + +fn transient_receipt_v3_fingerprint(record: &serde_json::Value) -> String { use sha2::{Digest, Sha256}; + let envelope: EvidenceEnvelope = + serde_json::from_value(record["envelope"].clone()).unwrap(); + let mut evidence = scrub_envelope(envelope).unwrap(); + evidence.source.title = None; + evidence.source.title_provenance = None; + evidence.source.locator.clear(); + format!( + "evidence-v3:sha256:{:x}", + Sha256::digest(serde_json::to_vec(&evidence).unwrap()) + ) +} + +#[tokio::test] +async fn archived_transcript_replay_accepts_changed_locator_and_preserves_v2_receipt() { let (app, dir) = test_app(Some("secret")); let original = sample_record(); - let envelope: EvidenceEnvelope = serde_json::from_value(original["envelope"].clone()).unwrap(); - let mut v2_evidence = scrub_envelope(envelope).unwrap(); - v2_evidence.source.title = None; - v2_evidence.source.title_provenance = None; - let v2_fingerprint = format!( - "evidence-v2:sha256:{:x}", - Sha256::digest(serde_json::to_vec(&v2_evidence).unwrap()) - ); + let original_v2 = receipt_v2_fingerprint(&original); + + let first = app + .clone() + .oneshot(transcript_request( + json!({"records": [original]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::OK); + + let conn = rusqlite::Connection::open(dir.path().join("ai-transcript-ingest-test.db")).unwrap(); + let fingerprint: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(fingerprint, original_v2); + + for locator in ["f", "1"] { + let mut archived = sample_record(); + archived["envelope"]["source"]["locator"] = + json!(format!("sha256:{}", locator.repeat(64))); + let replay = app + .clone() + .oneshot(transcript_request( + json!({"records": [archived]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(replay.status(), StatusCode::OK); + let body = axum::body::to_bytes(replay.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["receipts"][0]["disposition"], "duplicate"); + } + + let fingerprint: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(fingerprint, original_v2); + + let mut changed_evidence = sample_record(); + changed_evidence["envelope"]["source"]["locator"] = + json!(format!("sha256:{}", "1".repeat(64))); + changed_evidence["envelope"]["message"] = json!("different evidence"); + let conflict = app + .oneshot(transcript_request( + json!({"records": [changed_evidence]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(conflict.status(), StatusCode::CONFLICT); + let (log_count, fingerprint): (i64, String) = conn + .query_row( + "SELECT (SELECT COUNT(*) FROM logs), request_fingerprint + FROM ai_transcript_forward_receipts", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(log_count, 1); + assert_eq!(fingerprint, original_v2); +} + +#[tokio::test] +async fn transient_v3_receipt_rebinds_to_rollback_safe_v2() { + let (app, dir) = test_app(Some("secret")); + let original = sample_record(); + let original_v2 = receipt_v2_fingerprint(&original); + let transient_v3 = transient_receipt_v3_fingerprint(&original); let first = app .clone() @@ -27,35 +124,77 @@ async fn archived_transcript_replay_accepts_a_changed_locator() { let conn = rusqlite::Connection::open(dir.path().join("ai-transcript-ingest-test.db")).unwrap(); conn.execute( "UPDATE ai_transcript_forward_receipts SET request_fingerprint = ?1", - [&v2_fingerprint], + [&transient_v3], ) .unwrap(); let mut archived = sample_record(); - archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); + archived["envelope"]["source"]["locator"] = + json!(format!("sha256:{}", "f".repeat(64))); let replay = app .clone() .oneshot(transcript_request( - json!({"records": [archived.clone()]}).to_string(), + json!({"records": [archived]}).to_string(), )) .await .unwrap(); assert_eq!(replay.status(), StatusCode::OK); - let body = axum::body::to_bytes(replay.into_body(), usize::MAX) + + let fingerprint: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(fingerprint, original_v2); + + let mut moved_again = sample_record(); + moved_again["envelope"]["source"]["locator"] = + json!(format!("sha256:{}", "1".repeat(64))); + let replay = app + .oneshot(transcript_request( + json!({"records": [moved_again]}).to_string(), + )) .await .unwrap(); - let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); - assert_eq!(body["receipts"][0]["disposition"], "duplicate"); + assert_eq!(replay.status(), StatusCode::OK); +} + +#[tokio::test] +async fn legacy_null_fingerprint_receipt_accepts_locator_move_and_rebinds_v2() { + let (app, dir) = test_app(Some("secret")); + let original = sample_record(); + let original_v2 = receipt_v2_fingerprint(&original); - archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); - let second_replay = app + let first = app + .clone() + .oneshot(transcript_request( + json!({"records": [original]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::OK); + + let conn = rusqlite::Connection::open(dir.path().join("ai-transcript-ingest-test.db")).unwrap(); + conn.execute( + "UPDATE ai_transcript_forward_receipts SET request_fingerprint = NULL", + [], + ) + .unwrap(); + + let mut archived = sample_record(); + archived["envelope"]["source"]["locator"] = + json!(format!("sha256:{}", "f".repeat(64))); + let replay = app .clone() .oneshot(transcript_request( json!({"records": [archived]}).to_string(), )) .await .unwrap(); - assert_eq!(second_replay.status(), StatusCode::OK); + assert_eq!(replay.status(), StatusCode::OK); + let fingerprint: String = conn .query_row( "SELECT request_fingerprint FROM ai_transcript_forward_receipts", @@ -63,22 +202,7 @@ async fn archived_transcript_replay_accepts_a_changed_locator() { |row| row.get(0), ) .unwrap(); - assert!(fingerprint.starts_with("evidence-v3:sha256:")); - - let mut changed_evidence = sample_record(); - changed_evidence["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); - changed_evidence["envelope"]["message"] = json!("different evidence"); - let conflict = app - .oneshot(transcript_request( - json!({"records": [changed_evidence]}).to_string(), - )) - .await - .unwrap(); - assert_eq!(conflict.status(), StatusCode::CONFLICT); - let log_count: i64 = conn - .query_row("SELECT COUNT(*) FROM logs", [], |row| row.get(0)) - .unwrap(); - assert_eq!(log_count, 1); + assert_eq!(fingerprint, original_v2); } #[tokio::test] @@ -183,6 +307,14 @@ async fn check_old_receipt_replay(timestamp_absent: bool) { .await .unwrap(); assert_eq!(renamed.status(), StatusCode::OK); + let rebound: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(rebound, receipt_v2_fingerprint(&sample_record())); replay["envelope"]["message"] = json!("different evidence"); let changed = app .oneshot(transcript_request(json!({"records": [replay]}).to_string())) From 120fe6a2ea8f76a438c8892481e99795e5ad2b99 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:13:02 -0400 Subject: [PATCH 3/7] fix: harden transcript replay review findings --- src/ai_transcript_ingest/legacy_receipt.rs | 50 +++++++++------------- src/ai_transcript_replay_tests.rs | 3 +- 2 files changed, 22 insertions(+), 31 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index 901eb253e..b80e99615 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -47,39 +47,22 @@ fn transient_v3_fingerprint(envelope: &EvidenceEnvelope) -> anyhow::Result, - key: &str, -) -> anyhow::Result> { - Ok(tx - .query_row( - "SELECT l.ai_transcript_path FROM ai_transcript_forward_receipts r - JOIN logs l ON l.id = r.log_id WHERE r.source_record_id = ?1", - [key], - |row| row.get(0), - ) - .optional()? - .flatten()) -} - fn canonical_v2_fingerprint( - tx: &rusqlite::Transaction<'_>, - key: &str, envelope: &EvidenceEnvelope, + stored_locator: Option<&str>, ) -> anyhow::Result> { - let Some(locator) = stored_locator(tx, key)? else { + let Some(locator) = stored_locator else { return Ok(None); }; - Ok(Some(v2_fingerprint_with_locator(envelope, &locator)?)) + Ok(Some(v2_fingerprint_with_locator(envelope, locator)?)) } fn v2_fingerprint_matches( - tx: &rusqlite::Transaction<'_>, - key: &str, envelope: &EvidenceEnvelope, + stored_locator: Option<&str>, previous: &str, ) -> anyhow::Result { - Ok(canonical_v2_fingerprint(tx, key, envelope)?.as_deref() == Some(previous)) + Ok(canonical_v2_fingerprint(envelope, stored_locator)?.as_deref() == Some(previous)) } fn transient_v3_fingerprint_matches( @@ -100,8 +83,8 @@ fn receipt_key(forwarder_identity: &str, source_record_id: &str, shared_bearer: } } -/// Reconstruct only mutable titles from the stored row, then check the old -/// full-envelope hash. This also preserves timestamp-less exact replays: the +/// Reconstruct mutable display/location metadata from the stored row, then +/// check the old full-envelope hash. This also preserves timestamp-less exact replays: the /// hash, unlike a canonical log timestamp, retains their original `None`. fn old_fingerprint_matches( tx: &rusqlite::Transaction<'_>, @@ -312,12 +295,20 @@ fn insert_envelopes_with_identity( )?; let already_accepted = tx .query_row( - "SELECT request_fingerprint FROM ai_transcript_forward_receipts WHERE source_record_id = ?1", + "SELECT r.request_fingerprint, l.ai_transcript_path + FROM ai_transcript_forward_receipts r + JOIN logs l ON l.id = r.log_id + WHERE r.source_record_id = ?1", [&stored_receipt_key], - |row| row.get::<_, Option>(0), + |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, Option>(1)?, + )) + }, ) .optional()?; - if let Some(previous_fingerprint) = already_accepted { + if let Some((previous_fingerprint, stored_locator)) = already_accepted { if previous_fingerprint.as_deref() != Some(request_fingerprint.as_str()) { // Compatibility fingerprints are validated against the // canonical row before any rebinding. Persist v2 so a rollback @@ -328,7 +319,7 @@ fn insert_envelopes_with_identity( Some(previous) if previous.starts_with("evidence-v3:sha256:") => { let matches = transient_v3_fingerprint_matches(&envelope, previous)?; let replacement = if matches { - canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? } else { None }; @@ -336,9 +327,8 @@ fn insert_envelopes_with_identity( } Some(previous) if previous.starts_with("evidence-v2:sha256:") => ( v2_fingerprint_matches( - &tx, - &stored_receipt_key, &envelope, + stored_locator.as_deref(), previous, )?, None, diff --git a/src/ai_transcript_replay_tests.rs b/src/ai_transcript_replay_tests.rs index fc2f616c7..1072a3812 100644 --- a/src/ai_transcript_replay_tests.rs +++ b/src/ai_transcript_replay_tests.rs @@ -247,6 +247,7 @@ async fn check_old_receipt_replay(timestamp_absent: bool) { if timestamp_absent { original["envelope"]["timestamp"] = serde_json::Value::Null; } + let expected_v2 = receipt_v2_fingerprint(&original); let envelope: EvidenceEnvelope = serde_json::from_value(original["envelope"].clone()).unwrap(); let scrubbed = scrub_envelope(envelope).unwrap(); let old_hash = format!( @@ -314,7 +315,7 @@ async fn check_old_receipt_replay(timestamp_absent: bool) { |row| row.get(0), ) .unwrap(); - assert_eq!(rebound, receipt_v2_fingerprint(&sample_record())); + assert_eq!(rebound, expected_v2); replay["envelope"]["message"] = json!("different evidence"); let changed = app .oneshot(transcript_request(json!({"records": [replay]}).to_string())) From 988ad1a511a316f102980e9309008aae09912268 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:13:29 -0400 Subject: [PATCH 4/7] fix: complete locator lookup refactor --- src/ai_transcript_ingest/legacy_receipt.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index b80e99615..84d14b909 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -341,7 +341,7 @@ fn insert_envelopes_with_identity( previous, )?; let replacement = if matches { - canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? } else { None }; @@ -352,7 +352,7 @@ fn insert_envelopes_with_identity( let matches = legacy_receipt_matches(&tx, &stored_receipt_key, &envelope)?; let replacement = if matches { - canonical_v2_fingerprint(&tx, &stored_receipt_key, &envelope)? + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? } else { None }; From 284817690d4f3b3765c679fd4cffb8c4fb696af5 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:14:46 -0400 Subject: [PATCH 5/7] test: qualify transient v3 conflict handling --- src/ai_transcript_replay_tests.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/ai_transcript_replay_tests.rs b/src/ai_transcript_replay_tests.rs index 1072a3812..456f3dd4f 100644 --- a/src/ai_transcript_replay_tests.rs +++ b/src/ai_transcript_replay_tests.rs @@ -128,6 +128,25 @@ async fn transient_v3_receipt_rebinds_to_rollback_safe_v2() { ) .unwrap(); + let mut changed = sample_record(); + changed["envelope"]["message"] = json!("changed transient v3 evidence"); + let conflict = app + .clone() + .oneshot(transcript_request( + json!({"records": [changed]}).to_string(), + )) + .await + .unwrap(); + assert_eq!(conflict.status(), StatusCode::CONFLICT); + let fingerprint: String = conn + .query_row( + "SELECT request_fingerprint FROM ai_transcript_forward_receipts", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(fingerprint, transient_v3); + let mut archived = sample_record(); archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); From 92be26b3eaab5be1ba1d39b16d2a191b2c8fb0f5 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Sat, 26 Sep 2026 05:56:45 -0400 Subject: [PATCH 6/7] style: apply rustfmt after Lavra review --- src/ai_transcript_ingest/legacy_receipt.rs | 80 ++++++++++------------ src/ai_transcript_replay_tests.rs | 21 ++---- 2 files changed, 42 insertions(+), 59 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index 84d14b909..a77604892 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -314,51 +314,41 @@ fn insert_envelopes_with_identity( // canonical row before any rebinding. Persist v2 so a rollback // to the previous Cortex release can still read receipts // created by this version. - let (matches, replacement_fingerprint) = - match previous_fingerprint.as_deref() { - Some(previous) if previous.starts_with("evidence-v3:sha256:") => { - let matches = transient_v3_fingerprint_matches(&envelope, previous)?; - let replacement = if matches { - canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? - } else { - None - }; - (matches, replacement) - } - Some(previous) if previous.starts_with("evidence-v2:sha256:") => ( - v2_fingerprint_matches( - &envelope, - stored_locator.as_deref(), - previous, - )?, - None, - ), - Some(previous) if previous.starts_with("sha256:") => { - let matches = old_fingerprint_matches( - &tx, - &stored_receipt_key, - &envelope, - previous, - )?; - let replacement = if matches { - canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? - } else { - None - }; - (matches, replacement) - } - Some(_) => (false, None), - None => { - let matches = - legacy_receipt_matches(&tx, &stored_receipt_key, &envelope)?; - let replacement = if matches { - canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? - } else { - None - }; - (matches, replacement) - } - }; + let (matches, replacement_fingerprint) = match previous_fingerprint.as_deref() { + Some(previous) if previous.starts_with("evidence-v3:sha256:") => { + let matches = transient_v3_fingerprint_matches(&envelope, previous)?; + let replacement = if matches { + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? + } else { + None + }; + (matches, replacement) + } + Some(previous) if previous.starts_with("evidence-v2:sha256:") => ( + v2_fingerprint_matches(&envelope, stored_locator.as_deref(), previous)?, + None, + ), + Some(previous) if previous.starts_with("sha256:") => { + let matches = + old_fingerprint_matches(&tx, &stored_receipt_key, &envelope, previous)?; + let replacement = if matches { + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? + } else { + None + }; + (matches, replacement) + } + Some(_) => (false, None), + None => { + let matches = legacy_receipt_matches(&tx, &stored_receipt_key, &envelope)?; + let replacement = if matches { + canonical_v2_fingerprint(&envelope, stored_locator.as_deref())? + } else { + None + }; + (matches, replacement) + } + }; if !matches { return Err(IdempotencyConflict.into()); } diff --git a/src/ai_transcript_replay_tests.rs b/src/ai_transcript_replay_tests.rs index 456f3dd4f..11230a1c7 100644 --- a/src/ai_transcript_replay_tests.rs +++ b/src/ai_transcript_replay_tests.rs @@ -3,8 +3,7 @@ use super::*; fn receipt_v2_fingerprint(record: &serde_json::Value) -> String { use sha2::{Digest, Sha256}; - let envelope: EvidenceEnvelope = - serde_json::from_value(record["envelope"].clone()).unwrap(); + let envelope: EvidenceEnvelope = serde_json::from_value(record["envelope"].clone()).unwrap(); let mut evidence = scrub_envelope(envelope).unwrap(); evidence.source.title = None; evidence.source.title_provenance = None; @@ -17,8 +16,7 @@ fn receipt_v2_fingerprint(record: &serde_json::Value) -> String { fn transient_receipt_v3_fingerprint(record: &serde_json::Value) -> String { use sha2::{Digest, Sha256}; - let envelope: EvidenceEnvelope = - serde_json::from_value(record["envelope"].clone()).unwrap(); + let envelope: EvidenceEnvelope = serde_json::from_value(record["envelope"].clone()).unwrap(); let mut evidence = scrub_envelope(envelope).unwrap(); evidence.source.title = None; evidence.source.title_provenance = None; @@ -56,8 +54,7 @@ async fn archived_transcript_replay_accepts_changed_locator_and_preserves_v2_rec for locator in ["f", "1"] { let mut archived = sample_record(); - archived["envelope"]["source"]["locator"] = - json!(format!("sha256:{}", locator.repeat(64))); + archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", locator.repeat(64))); let replay = app .clone() .oneshot(transcript_request( @@ -83,8 +80,7 @@ async fn archived_transcript_replay_accepts_changed_locator_and_preserves_v2_rec assert_eq!(fingerprint, original_v2); let mut changed_evidence = sample_record(); - changed_evidence["envelope"]["source"]["locator"] = - json!(format!("sha256:{}", "1".repeat(64))); + changed_evidence["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); changed_evidence["envelope"]["message"] = json!("different evidence"); let conflict = app .oneshot(transcript_request( @@ -148,8 +144,7 @@ async fn transient_v3_receipt_rebinds_to_rollback_safe_v2() { assert_eq!(fingerprint, transient_v3); let mut archived = sample_record(); - archived["envelope"]["source"]["locator"] = - json!(format!("sha256:{}", "f".repeat(64))); + archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); let replay = app .clone() .oneshot(transcript_request( @@ -169,8 +164,7 @@ async fn transient_v3_receipt_rebinds_to_rollback_safe_v2() { assert_eq!(fingerprint, original_v2); let mut moved_again = sample_record(); - moved_again["envelope"]["source"]["locator"] = - json!(format!("sha256:{}", "1".repeat(64))); + moved_again["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "1".repeat(64))); let replay = app .oneshot(transcript_request( json!({"records": [moved_again]}).to_string(), @@ -203,8 +197,7 @@ async fn legacy_null_fingerprint_receipt_accepts_locator_move_and_rebinds_v2() { .unwrap(); let mut archived = sample_record(); - archived["envelope"]["source"]["locator"] = - json!(format!("sha256:{}", "f".repeat(64))); + archived["envelope"]["source"]["locator"] = json!(format!("sha256:{}", "f".repeat(64))); let replay = app .clone() .oneshot(transcript_request( From f816aefaf1492720f005ed4112041aacbebe0a31 Mon Sep 17 00:00:00 2001 From: jmagar <38927646+jmagar@users.noreply.github.com> Date: Sat, 26 Sep 2026 06:00:13 -0400 Subject: [PATCH 7/7] docs: clarify transcript replay compatibility invariants --- src/ai_transcript_ingest/legacy_receipt.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/ai_transcript_ingest/legacy_receipt.rs b/src/ai_transcript_ingest/legacy_receipt.rs index a77604892..c646a573c 100644 --- a/src/ai_transcript_ingest/legacy_receipt.rs +++ b/src/ai_transcript_ingest/legacy_receipt.rs @@ -84,8 +84,9 @@ fn receipt_key(forwarder_identity: &str, source_record_id: &str, shared_bearer: } /// Reconstruct mutable display/location metadata from the stored row, then -/// check the old full-envelope hash. This also preserves timestamp-less exact replays: the -/// hash, unlike a canonical log timestamp, retains their original `None`. +/// check the old full-envelope hash. This also preserves timestamp-less exact +/// replays: the hash, unlike a canonical log timestamp, retains their original +/// `None`. fn old_fingerprint_matches( tx: &rusqlite::Transaction<'_>, key: &str, @@ -93,7 +94,8 @@ fn old_fingerprint_matches( previous: &str, ) -> anyhow::Result { // An exact old hash is sufficient even if bounded log metadata omitted - // its source fields. Canonical metadata is needed only for a title change. + // its source fields. Canonical metadata is needed only when mutable title + // or locator metadata changed. if previous == format!("sha256:{:x}", Sha256::digest(serde_json::to_vec(envelope)?)) { return Ok(true); } @@ -312,8 +314,8 @@ fn insert_envelopes_with_identity( if previous_fingerprint.as_deref() != Some(request_fingerprint.as_str()) { // Compatibility fingerprints are validated against the // canonical row before any rebinding. Persist v2 so a rollback - // to the previous Cortex release can still read receipts - // created by this version. + // still recognizes the durable receipt format; locator-move + // tolerance itself is provided by this version's verifier. let (matches, replacement_fingerprint) = match previous_fingerprint.as_deref() { Some(previous) if previous.starts_with("evidence-v3:sha256:") => { let matches = transient_v3_fingerprint_matches(&envelope, previous)?;