diff --git a/Cargo.lock b/Cargo.lock index dc701c4..f417372 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2708,7 +2708,7 @@ dependencies = [ [[package]] name = "ptask-bot" -version = "3.52.3" +version = "3.52.8" dependencies = [ "anyhow", "jiff", @@ -2724,7 +2724,7 @@ dependencies = [ [[package]] name = "ptask-cli" -version = "3.52.3" +version = "3.52.8" dependencies = [ "anyhow", "axum", @@ -2752,7 +2752,7 @@ dependencies = [ [[package]] name = "ptask-core" -version = "3.52.3" +version = "3.52.8" dependencies = [ "hex", "hmac", @@ -2774,7 +2774,7 @@ dependencies = [ [[package]] name = "ptask-distill" -version = "3.52.3" +version = "3.52.8" dependencies = [ "anyhow", "candle-core", @@ -2798,7 +2798,7 @@ dependencies = [ [[package]] name = "ptask-notify" -version = "3.52.3" +version = "3.52.8" dependencies = [ "lettre", "ptask-core", @@ -2811,7 +2811,7 @@ dependencies = [ [[package]] name = "ptask-server" -version = "3.52.3" +version = "3.52.8" dependencies = [ "anyhow", "axum", @@ -2843,7 +2843,7 @@ dependencies = [ [[package]] name = "ptask-tui" -version = "3.52.3" +version = "3.52.8" dependencies = [ "anyhow", "crossterm", diff --git a/Cargo.toml b/Cargo.toml index b35d305..977f9b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ members = [ resolver = "2" [workspace.package] -version = "3.52.3" +version = "3.52.8" edition = "2024" license-file = "LICENSE" authors = ["PureTensor, Inc. "] @@ -52,12 +52,12 @@ rmcp = { version = "3.3.0", features = ["server", "macros", "schemars", "transpo futures-util = { version = "0.3", default-features = false } # Internal crates -ptask-core = { path = "crates/ptask-core", version = "3.52.3" } -ptask-tui = { path = "crates/ptask-tui", version = "3.52.3" } -ptask-server = { path = "crates/ptask-server", version = "3.52.3" } -ptask-bot = { path = "crates/ptask-bot", version = "3.52.3" } -ptask-distill = { path = "crates/ptask-distill", version = "3.52.3" } -ptask-notify = { path = "crates/ptask-notify", version = "3.52.3" } +ptask-core = { path = "crates/ptask-core", version = "3.52.8" } +ptask-tui = { path = "crates/ptask-tui", version = "3.52.8" } +ptask-server = { path = "crates/ptask-server", version = "3.52.8" } +ptask-bot = { path = "crates/ptask-bot", version = "3.52.8" } +ptask-distill = { path = "crates/ptask-distill", version = "3.52.8" } +ptask-notify = { path = "crates/ptask-notify", version = "3.52.8" } # HMAC for outbound webhook signing (v0.3.5). sha2 = "0.10" diff --git a/crates/ptask-core/src/approval_journal.rs b/crates/ptask-core/src/approval_journal.rs index d57d4d9..9021992 100644 --- a/crates/ptask-core/src/approval_journal.rs +++ b/crates/ptask-core/src/approval_journal.rs @@ -48,7 +48,8 @@ pub struct Journal { key_file: PathBuf, } -/// One journal line. `ev` is request, approve, reject, withdraw or consume. +/// One journal line. `ev` is request, approve, reject, withdraw, consume, +/// or consume_void (a consume whose database transaction did not commit). #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct Record { pub n: u64, @@ -251,23 +252,97 @@ impl Journal { /// consume record under the same lock so a replay is refused across /// processes. Shadow mode logs a refusal and still records the consume /// when the chain is intact. - pub fn check_and_record_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result<()> { + /// + /// `Ok(true)` means a consume line was appended. `Ok(false)` is shadow + /// mode softening a failure that wrote nothing. Callers that void a + /// consume when their own database transaction rolls back must void + /// only after `Ok(true)`: a refusal leaves any earlier committed + /// consume in place. A shadow refusal appends a consume only when none + /// is open; an already-open consume returns `Ok(false)` and writes + /// nothing, so a rolled-back retry cannot void it. + pub fn check_and_record_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result { let res = self.locked(|key, file, chain| { let verdict = authorise(&chain, claim); match verdict { Ok(()) => { - Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts)) + Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts))?; + Ok(true) } Err(e) if self.mode == Mode::Shadow => { tracing::warn!( approval = claim.uuid, "approval journal (shadow) would refuse consume: {e}" ); - Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts)) + // An open consume already committed. Another line and + // Ok(true) would arm ConsumeVoidGuard; a rolled-back + // retry would then void it and reopen the approval. + if consume_is_open(&chain, claim.uuid) { + Ok(false) + } else { + Self::append_in( + key, + file, + &chain, + Self::record("consume", claim, actor, ts), + )?; + Ok(true) + } } Err(e) => Err(e), } }); + match res { + Ok(appended) => Ok(appended), + Err(e) if self.mode == Mode::Shadow => { + tracing::warn!("approval journal (shadow) error ignored: {e}"); + Ok(false) + } + Err(e) => Err(e), + } + } + + /// Whether the journal currently authorises `claim` (an approve is + /// present, unconsumed, and matches). Does not append. Shadow mode + /// logs a refusal and returns `Ok`. Used by `payload` and `verify`, + /// which release or attest before `consume` latches. + pub fn authorises(&self, claim: &Claim) -> Result<()> { + let res = self.locked(|_key, _file, chain| authorise(&chain, claim)); + match res { + Err(e) if self.mode == Mode::Shadow => { + tracing::warn!( + approval = claim.uuid, + "approval journal (shadow) would refuse release: {e}" + ); + Ok(()) + } + other => other, + } + } + + /// Append `consume_void` when `uuid` has an unvoided consume. + /// + /// `consume` writes the journal line before the SQLite commit, because + /// the journal is the authority and must not lag a committed latch. + /// If that transaction then rolls back, the consume line would brick + /// the still-approved row forever (`authorise` treats consume as + /// sticky). Only a holder of the journal key can append the void, and + /// the void does not look at the database: clearing `consumed_at` in + /// SQLite must not undo a consume that did commit. + /// + /// No-op when this uuid has no open consume (the check failed before + /// any line was written, or a void is already in place). + pub fn void_uncommitted_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result<()> { + let res = self.locked(|key, file, chain| { + if !consume_is_open(&chain, claim.uuid) { + return Ok(()); + } + Self::append_in( + key, + file, + &chain, + Self::record("consume_void", claim, actor, ts), + ) + }); self.soften(res) } @@ -301,6 +376,10 @@ fn authorise(chain: &[Record], claim: &Claim) -> Result<()> { approve.insert("a", r); } "consume" => consumed = true, + // A consume whose database transaction rolled back. Only the + // key holder can write this; it does not clear a consume that + // has no matching void. + "consume_void" => consumed = false, "reject" | "withdraw" => { approve.remove("a"); } @@ -332,3 +411,16 @@ fn authorise(chain: &[Record], claim: &Claim) -> Result<()> { } Ok(()) } + +/// True when the latest consume for `uuid` has not been voided. +fn consume_is_open(chain: &[Record], uuid: &str) -> bool { + let mut consumed = false; + for r in chain.iter().filter(|r| r.uuid == uuid) { + match r.ev.as_str() { + "consume" => consumed = true, + "consume_void" => consumed = false, + _ => {} + } + } + consumed +} diff --git a/crates/ptask-core/src/approvals.rs b/crates/ptask-core/src/approvals.rs index 6669684..73bc63c 100644 --- a/crates/ptask-core/src/approvals.rs +++ b/crates/ptask-core/src/approvals.rs @@ -910,6 +910,13 @@ pub fn payload_bytes(db: &Db, id: &str) -> Result> { if ap.consumed_at.is_some() { return Err(Error::Approval(ApprovalError::AlreadyConsumed)); } + // Enforce mode: the executor pattern is payload, then act, then + // consume. A row written straight into SQLite passes the database + // gate and would otherwise release its bytes before consume refused + // it. The journal check does not latch. + if let Some(j) = Journal::from_env() { + j.authorises(&claim_of(&ap))?; + } stored_payload(ap) } @@ -1363,6 +1370,11 @@ pub fn verify(db: &Db, id: &str, offered: &PayloadSource) -> Result { let ap = get(db, id)?; let digest = offered_digest(offered)?; gate_status_and_digest(&ap, &digest, &dates::now_in_operator_tz()?)?; + // Same journal authority as consume, without the latch. A forged + // approved row must not verify in enforce mode. + if let Some(j) = Journal::from_env() { + j.authorises(&claim_of(&ap))?; + } Ok(ap) } @@ -1390,9 +1402,32 @@ pub fn consume(db: &Db, id: &str, offered: &PayloadSource, ctx: &EventCtx) -> Re // The journal, not the row, is the authority in enforce mode: a row // written straight into SQLite has no journaled approval and is refused // here, inside the transaction, so the latch rolls back. - if let Some(j) = Journal::from_env() { - j.check_and_record_consume(&claim_of(¤t), actor, &now)?; - } + // + // The consume line is fsynced before commit so a committed latch can + // never race ahead of the journal. The guard is declared after `tx` + // and dropped first, still holding the write lock: if `record_event` + // or `commit` fails, it appends `consume_void` before the rollback is + // visible. Without that, the journal consume sticks, the row stays + // approved and unconsumed, and every retry is AlreadyConsumed. + let mut void_guard = ConsumeVoidGuard { + journal: Journal::from_env(), + uuid: current.uuid.clone(), + digest: current.digest.clone(), + kind: current.kind.clone(), + expires_at: current.expires_at.clone(), + actor: actor.to_string(), + ts: now.clone(), + armed: false, + }; + // Arm only when this call appended a consume. A refusal (forged row, + // or a committed consume whose database latch was cleared) must not + // void an earlier consume that did commit. + let appended = if let Some(j) = void_guard.journal.as_ref() { + j.check_and_record_consume(&claim_of(¤t), actor, &now)? + } else { + false + }; + void_guard.armed = appended; record_event( &tx, ctx, @@ -1401,9 +1436,49 @@ pub fn consume(db: &Db, id: &str, offered: &PayloadSource, ctx: &EventCtx) -> Re serde_json::json!({"approval_id": current.ap_id()}), )?; tx.commit()?; + void_guard.armed = false; get(db, ¤t.uuid) } +/// Voids a journal consume when dropped while still armed. +/// +/// Declared after the SQLite transaction so Drop runs while that +/// transaction, and its write lock, is still held. +struct ConsumeVoidGuard { + journal: Option, + uuid: String, + digest: String, + kind: String, + expires_at: Option, + actor: String, + ts: String, + armed: bool, +} + +impl Drop for ConsumeVoidGuard { + fn drop(&mut self) { + if !self.armed { + return; + } + let Some(journal) = self.journal.as_ref() else { + return; + }; + let claim = Claim { + uuid: &self.uuid, + digest: &self.digest, + kind: &self.kind, + expires_at: self.expires_at.as_deref(), + decided_by: None, + }; + if let Err(e) = journal.void_uncommitted_consume(&claim, &self.actor, &self.ts) { + tracing::error!( + approval = %self.uuid, + "approval journal consume survived a rolled-back transaction and could not be voided: {e}" + ); + } + } +} + /// Mark pending rows whose `expires_at` is in the past as expired. /// Idempotent: already-expired rows are left alone. Returns how many /// newly expired. Runs on every request, so it reads only the three diff --git a/crates/ptask-core/src/dupes.rs b/crates/ptask-core/src/dupes.rs index 8250ac7..eb93ff1 100644 --- a/crates/ptask-core/src/dupes.rs +++ b/crates/ptask-core/src/dupes.rs @@ -1005,7 +1005,29 @@ pub(crate) fn unmerge_if_needed( )?; } if payload.get("recurrence_copied").and_then(|v| v.as_bool()) == Some(true) { - tx.execute("DELETE FROM pt_recurrence WHERE task_uuid=?1", [&canon])?; + // Take the rule back only when it is still the copied row. + // A later deadline edit (or an advanced occurrence) rewrites + // next_occurrence and anchor on the canonical; deleting the + // row then throws that schedule away. The duplicate still has + // the original rule either way. + let unchanged: bool = tx.query_row( + "SELECT EXISTS( + SELECT 1 + FROM pt_recurrence c + JOIN pt_recurrence d ON d.task_uuid = ?2 + WHERE c.task_uuid = ?1 + AND c.rrule = d.rrule + AND c.mode = d.mode + AND c.original_input = d.original_input + AND c.next_occurrence = d.next_occurrence + AND c.anchor IS d.anchor + )", + params![canon, duplicate_uuid], + |r| r.get(0), + )?; + if unchanged { + tx.execute("DELETE FROM pt_recurrence WHERE task_uuid=?1", [&canon])?; + } } if let Some(g) = payload.get("goal_copied").and_then(|v| v.as_str()) { tx.execute( @@ -1041,8 +1063,11 @@ pub(crate) fn unmerge_if_needed( params![child, duplicate_uuid, now], )?; } - // The carried criteria go back with the duplicate (which kept its - // own throughout), checked or not: they were the duplicate's. + // An unchecked carried criterion goes back with the duplicate, + // which kept its own copy. A check recorded on the canonical is + // evidence: the duplicate's copy was never checked, and the reopen + // after this unmerge clears checks on the duplicate, so removing + // the canonical row would drop the evidence from every checklist. let carried: Vec = payload .get("criteria_carried") .and_then(|v| v.as_array()) @@ -1051,7 +1076,7 @@ pub(crate) fn unmerge_if_needed( if !carried.is_empty() { for c in crate::criteria::list_in_conn(tx, &canon)? .into_iter() - .filter(|c| carried.contains(&c.n)) + .filter(|c| carried.contains(&c.n) && !c.done) { crate::criteria::remove_in_conn( tx, @@ -1309,4 +1334,122 @@ mod tests { .contains("dismissed") ); } + + fn has_recurrence(db: &Db, id: &str) -> bool { + db.with_conn(|c| { + Ok(c.query_row( + "SELECT EXISTS(SELECT 1 FROM pt_recurrence WHERE task_uuid=?1)", + [id], + |r| r.get(0), + )?) + }) + .unwrap() + } + + fn recurring(db: &Db, title: &str) -> crate::Task { + let rec = crate::recurrence::parse("every monday at 9am").unwrap(); + let mut new = NewTask::minimal(title); + new.deadline = Some("2026-05-18T09:00:00+01:00".into()); + tasks::create_with_extensions( + db, + new, + tasks::Extensions { + recurrence: Some(rec), + ..Default::default() + }, + &EventCtx::test(), + ) + .unwrap() + } + + /// Undo of a merge deletes the copied rule. A deadline edit after the + /// merge rewrites that rule's next occurrence and anchor; the delete + /// must not throw the edited schedule away. An untouched copy still goes. + #[test] + fn undo_merge_keeps_a_recurrence_edited_after_the_copy() { + let ctx = EventCtx::test(); + + let (_d, db) = fresh(); + let canon = add(&db, "Canonical export"); + let dup = recurring(&db, "Weekly export check"); + merge(&db, &dup.id, &canon.id, None, &ctx).unwrap(); + assert!(has_recurrence(&db, &canon.id)); + tasks::undo_last(&db, &ctx).unwrap(); + assert!( + !has_recurrence(&db, &canon.id), + "an untouched copied rule is part of the merge and must go back" + ); + assert!(has_recurrence(&db, &dup.id)); + assert_eq!(status(&db, &dup.id), "todo"); + + let (_d, db) = fresh(); + let canon = add(&db, "Canonical export"); + let dup = recurring(&db, "Weekly export check"); + merge(&db, &dup.id, &canon.id, None, &ctx).unwrap(); + tasks::update_deadline(&db, &canon.id, Some("2026-12-01"), &ctx).unwrap(); + tasks::undo_last(&db, &ctx).unwrap(); + assert!( + has_recurrence(&db, &canon.id), + "a schedule edited after the merge must survive undo" + ); + assert!( + has_recurrence(&db, &dup.id), + "the duplicate keeps the original rule" + ); + assert_eq!(status(&db, &dup.id), "todo"); + } + + /// Checking a carried criterion records the evidence on the canonical. + /// Undo used to remove that row. The duplicate's own copy was never + /// checked, and reopen clears its checks, so the evidence vanished. + #[test] + fn undo_merge_keeps_a_checked_carried_criterion() { + let ctx = EventCtx::test(); + let (_d, db) = fresh(); + let canon = add(&db, "Ship the backup drill"); + let dup = tasks::create_with_extensions( + &db, + NewTask::minimal("Backup drill"), + tasks::Extensions { + acceptance: vec!["offsite copy verified".into(), "tape labelled".into()], + ..Default::default() + }, + &ctx, + ) + .unwrap(); + merge(&db, &dup.id, &canon.id, None, &ctx).unwrap(); + let carried = crate::criteria::list(&db, &canon.id).unwrap(); + let offsite = carried + .iter() + .find(|c| c.text == "offsite copy verified") + .expect("the merge carries the duplicate's criteria"); + crate::criteria::check(&db, &canon.id, offsite.n, Some("sha256 matches"), &ctx).unwrap(); + + tasks::undo_last(&db, &ctx).unwrap(); + assert_eq!(status(&db, &dup.id), "todo"); + + let on_canon = crate::criteria::list(&db, &canon.id).unwrap(); + let kept = on_canon + .iter() + .find(|c| c.text == "offsite copy verified") + .expect("the checked criterion and its evidence must stay on the canonical task"); + assert!(kept.done, "{kept:?}"); + assert_eq!(kept.evidence.as_deref(), Some("sha256 matches")); + assert!( + !on_canon.iter().any(|c| c.text == "tape labelled"), + "an unchecked carried criterion still goes back with the duplicate: {on_canon:?}" + ); + + let on_dup = crate::criteria::list(&db, &dup.id).unwrap(); + assert!( + on_dup + .iter() + .any(|c| c.text == "offsite copy verified" && !c.done), + "the duplicate keeps its own unchecked copy: {on_dup:?}" + ); + assert!( + on_dup.iter().any(|c| c.text == "tape labelled" && !c.done), + "{on_dup:?}" + ); + } } diff --git a/crates/ptask-core/tests/approval_journal.rs b/crates/ptask-core/tests/approval_journal.rs index 7f2cbc8..7086dbf 100644 --- a/crates/ptask-core/tests/approval_journal.rs +++ b/crates/ptask-core/tests/approval_journal.rs @@ -99,6 +99,17 @@ fn forge_approved_row(s: &Setup) -> String { "AP-9001".to_string() } +fn journal_evs(path: &std::path::Path) -> Vec { + std::fs::read_to_string(path) + .unwrap() + .lines() + .filter_map(|line| { + let rest = line.split("\"ev\":\"").nth(1)?; + rest.split('"').next().map(str::to_string) + }) + .collect() +} + #[test] fn forged_row_consume_fails() { let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); @@ -226,6 +237,319 @@ fn explicit_off_values_stay_off() { } } +/// The executor reads bytes with `payload`, then acts, then `consume`. +/// Enforce mode must refuse that read for a row that was never journaled, +/// or the action happens before consume can refuse it. +#[test] +fn forged_row_does_not_release_payload_or_verify() { + let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); + let s = setup(Some("enforce")); + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute( + "INSERT INTO approvals (id, seq, kind, title, payload, payload_kind, digest, requester, + status, created_at, decided_by, decided_via, decided_at) + VALUES ('forged-bytes', 9002, 'email', 'forged letter', ?1, 'file', ?2, 'requester', + 'approved', '2026-01-01T00:00:00+00:00', 'operator', 'api', + '2026-01-01T00:00:01+00:00')", + rusqlite::params![b"send the money".as_slice(), digest()], + ) + .unwrap(); + drop(conn); + + let released = approvals::payload_bytes(&s.db, "AP-9002"); + assert!( + matches!(released, Err(Error::Approval(ApprovalError::Unverified(_)))), + "a forged row must not release its bytes: {released:?}" + ); + let verified = approvals::verify(&s.db, "AP-9002", &PayloadSource::Digest(digest())); + assert!( + matches!(verified, Err(Error::Approval(ApprovalError::Unverified(_)))), + "a forged row must not verify: {verified:?}" + ); + + let ap = approvals::request( + &s.db, + RequestInput { + kind: "email".into(), + title: "Real letter".into(), + request_note: None, + payload: PayloadSource::Json(br#"{"to":"a@b.c"}"#.to_vec()), + task_pt_id: None, + expires_in: Some("1h".into()), + }, + &EventCtx::local("requester"), + ) + .unwrap() + .approval; + approve(&s.db, &ap.ap_id()); + let bytes = approvals::payload_bytes(&s.db, &ap.ap_id()).expect("journaled approval releases"); + assert_eq!(bytes, br#"{"to":"a@b.c"}"#); + approvals::verify( + &s.db, + &ap.ap_id(), + &PayloadSource::Json(br#"{"to":"a@b.c"}"#.to_vec()), + ) + .expect("journaled approval verifies"); +} + +/// `record_event` runs after the journal consume is fsynced. A colliding +/// idempotency key makes that insert fail and the SQLite transaction roll +/// back. The journal consume must be voided, or the still-approved row can +/// never be consumed. +#[test] +fn rolled_back_consume_is_voided_and_can_be_retried() { + let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); + let s = setup(Some("enforce")); + let ap = request(&s.db, Some("1h")); + approve(&s.db, &ap.ap_id()); + + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute( + "INSERT INTO pt_event_log (uuid, task_uuid, event_type, payload, ts) + VALUES ('collide', NULL, 'probe', '{}', '2026-01-01T00:00:00+00:00')", + [], + ) + .unwrap(); + drop(conn); + + let ctx = EventCtx::local("executor").with_uuid("collide"); + let first = approvals::consume(&s.db, &ap.ap_id(), &PayloadSource::Digest(digest()), &ctx); + assert!( + first.is_err(), + "colliding event uuid must fail the consume: {first:?}" + ); + let row = approvals::get(&s.db, &ap.ap_id()).unwrap(); + assert_eq!(row.status, "approved"); + assert!( + row.consumed_at.is_none(), + "the database latch must roll back with the event insert" + ); + + let journal = std::fs::read_to_string(&s.journal).unwrap(); + assert!( + journal.contains("\"ev\":\"consume_void\""), + "the uncommitted consume must be voided in the journal: {journal}" + ); + + consume(&s.db, &ap.ap_id(), &digest()).expect("retry after the rolled-back consume"); + let consumed = approvals::get(&s.db, &ap.ap_id()).unwrap(); + assert!(consumed.consumed_at.is_some(), "the retry latches"); + assert!( + consume(&s.db, &ap.ap_id(), &digest()).is_err(), + "a committed consume is still one-shot" + ); +} + +/// A committed consume must stay consumed even if a SQLite writer clears +/// the latch. The void path must not treat "row unconsumed" as authority. +#[test] +fn clearing_the_db_latch_does_not_reopen_a_committed_consume() { + let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); + let s = setup(Some("enforce")); + let ap = request(&s.db, Some("1h")); + approve(&s.db, &ap.ap_id()); + consume(&s.db, &ap.ap_id(), &digest()).unwrap(); + + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute_batch("DROP TRIGGER IF EXISTS approvals_lock_after_decision;") + .unwrap(); + conn.execute( + "UPDATE approvals SET consumed_at = NULL, consumed_by = NULL", + [], + ) + .unwrap(); + drop(conn); + + let r = consume(&s.db, &ap.ap_id(), &digest()); + assert!( + matches!(r, Err(Error::Approval(ApprovalError::AlreadyConsumed))), + "clearing consumed_at must not reopen a journaled consume: {r:?}" + ); + // The refused attempt must not void the committed consume either. + let again = consume(&s.db, &ap.ap_id(), &digest()); + assert!( + matches!(again, Err(Error::Approval(ApprovalError::AlreadyConsumed))), + "a refused retry must leave the journal consume in place: {again:?}" + ); + let journal = std::fs::read_to_string(&s.journal).unwrap(); + assert!( + !journal.contains("\"ev\":\"consume_void\""), + "clearing the latch must not write a void: {journal}" + ); +} + +/// A first shadow consume whose database transaction rolls back is voided, +/// whether `authorise` allowed it or shadow softened a refusal while nothing +/// was open. The row stays unconsumed and a retry can proceed. +#[test] +fn shadow_rolled_back_first_consume_is_voided() { + let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); + let s = setup(Some("shadow")); + let ap = request(&s.db, Some("1h")); + approve(&s.db, &ap.ap_id()); + + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute( + "INSERT INTO pt_event_log (uuid, task_uuid, event_type, payload, ts) + VALUES ('collide-shadow-first', NULL, 'probe', '{}', '2026-01-01T00:00:00+00:00')", + [], + ) + .unwrap(); + drop(conn); + + let ctx = EventCtx::local("executor").with_uuid("collide-shadow-first"); + let first = approvals::consume(&s.db, &ap.ap_id(), &PayloadSource::Digest(digest()), &ctx); + assert!( + first.is_err(), + "colliding event uuid must fail the consume: {first:?}" + ); + let row = approvals::get(&s.db, &ap.ap_id()).unwrap(); + assert_eq!(row.status, "approved"); + assert!( + row.consumed_at.is_none(), + "the database latch must roll back with the event insert" + ); + let evs = journal_evs(&s.journal); + assert!( + evs.iter().any(|e| e == "consume_void"), + "a rolled-back first shadow consume must be voided: {evs:?}" + ); + + consume(&s.db, &ap.ap_id(), &digest()).expect("retry after the rolled-back consume"); + let consumed = approvals::get(&s.db, &ap.ap_id()).unwrap(); + assert!(consumed.consumed_at.is_some(), "the retry latches"); + + // Nothing is open and authorise refuses (forged row, shadow Err arm). + // The arm must still append, and the rollback must void that line. + let forged = forge_approved_row(&s); + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute( + "INSERT INTO pt_event_log (uuid, task_uuid, event_type, payload, ts) + VALUES ('collide-shadow-forged', NULL, 'probe', '{}', '2026-01-01T00:00:00+00:00')", + [], + ) + .unwrap(); + drop(conn); + let forged_ctx = EventCtx::local("executor").with_uuid("collide-shadow-forged"); + let forged_try = approvals::consume( + &s.db, + &forged, + &PayloadSource::Digest(digest()), + &forged_ctx, + ); + assert!( + forged_try.is_err(), + "colliding event uuid must fail the forged consume: {forged_try:?}" + ); + let evs = journal_evs(&s.journal); + let forged_consume = evs.iter().filter(|e| *e == "consume").count(); + let forged_void = evs.iter().filter(|e| *e == "consume_void").count(); + // The journaled approval above left consume, consume_void, consume. + assert_eq!( + forged_consume, 3, + "the softened refusal must append its own consume: {evs:?}" + ); + assert_eq!( + forged_void, 2, + "a rolled-back shadow refusal with nothing open must be voided: {evs:?}" + ); +} + +/// Shadow mode must not arm `ConsumeVoidGuard` for a consume that is already +/// open. A committed consume, a cleared `consumed_at`, and a colliding event +/// uuid used to append a second consume and then `consume_void`. That void +/// reopens the approval, so flipping to enforce later allows another consume. +#[test] +fn shadow_rollback_must_not_void_a_committed_consume() { + let _g = ENV.lock().unwrap_or_else(|e| e.into_inner()); + let s = setup(Some("shadow")); + let ap = request(&s.db, Some("1h")); + approve(&s.db, &ap.ap_id()); + consume(&s.db, &ap.ap_id(), &digest()).expect("shadow consume commits"); + + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute_batch("DROP TRIGGER IF EXISTS approvals_lock_after_decision;") + .unwrap(); + conn.execute( + "UPDATE approvals SET consumed_at = NULL, consumed_by = NULL", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO pt_event_log (uuid, task_uuid, event_type, payload, ts) + VALUES ('collide-shadow-open', NULL, 'probe', '{}', '2026-01-01T00:00:00+00:00')", + [], + ) + .unwrap(); + drop(conn); + + let ctx = EventCtx::local("executor").with_uuid("collide-shadow-open"); + let rolled = approvals::consume(&s.db, &ap.ap_id(), &PayloadSource::Digest(digest()), &ctx); + assert!( + rolled.is_err(), + "colliding event uuid must roll the retry back: {rolled:?}" + ); + let row = approvals::get(&s.db, &ap.ap_id()).unwrap(); + assert!( + row.consumed_at.is_none(), + "the retry's database latch must roll back" + ); + let evs = journal_evs(&s.journal); + assert_eq!( + evs.iter().filter(|e| *e == "consume").count(), + 1, + "a rolled-back retry must not append another consume: {evs:?}" + ); + assert!( + !evs.iter().any(|e| e == "consume_void"), + "a rolled-back retry of an open consume must not arm the void guard: {evs:?}" + ); + + // The latch is still clear, so a later call reaches the journal again. + // Shadow allows the row update and must still leave the open consume alone. + consume(&s.db, &ap.ap_id(), &digest()).expect("shadow still allows the row latch"); + let evs = journal_evs(&s.journal); + assert_eq!( + evs.iter().filter(|e| *e == "consume").count(), + 1, + "the committed consume stays the only one: {evs:?}" + ); + assert!( + !evs.iter().any(|e| e == "consume_void"), + "the open consume must stay open: {evs:?}" + ); + + let conn = rusqlite::Connection::open(s.db_path()).unwrap(); + conn.execute( + "UPDATE approvals SET consumed_at = NULL, consumed_by = NULL", + [], + ) + .unwrap(); + drop(conn); + // SAFETY: this test holds ENV for the whole run. + unsafe { + std::env::set_var("PTASK_APPROVAL_JOURNAL", "enforce"); + } + let enforced = consume(&s.db, &ap.ap_id(), &digest()); + assert!( + matches!( + enforced, + Err(Error::Approval(ApprovalError::AlreadyConsumed)) + ), + "flipping to enforce must not allow a second consume: {enforced:?}" + ); + let evs = journal_evs(&s.journal); + assert_eq!( + evs.iter().filter(|e| *e == "consume").count(), + 1, + "enforce must not append over the open consume: {evs:?}" + ); + assert!( + !evs.iter().any(|e| e == "consume_void"), + "the enforce refusal must not void the committed consume: {evs:?}" + ); +} + #[test] fn verify_of_missing_journal_is_an_error_not_zero_records() { use ptask_core::approval_journal::{Journal, Mode}; diff --git a/docs/approvals.md b/docs/approvals.md index 8d072a2..78e0656 100644 --- a/docs/approvals.md +++ b/docs/approvals.md @@ -106,7 +106,11 @@ pt approval consume AP-12 --payload-file /tmp/letter.html approved, not past `expires_at`, not yet consumed. Otherwise it writes nothing to stdout and exits 3 (pending), 4 (rejected / withdrawn / expired) or 6 (already consumed). A digest-only request has no bytes to release -(exit 1). +(exit 1). With `PTASK_APPROVAL_JOURNAL=enforce`, `payload` and `verify` +also require the same unconsumed journaled `approve` as `consume` (exit +7 when it is missing). They do not latch. `preview` and `in_force` stay +database reads: `preview` is what the operator decides on, and a poller +that then calls `payload` still hits the journal. Every approval JSON object (`show`/`ls --json`, HTTP, MCP) carries `"in_force": true|false`, the same verdict computed at read time. Pollers @@ -314,13 +318,20 @@ whoever can write the SQLite file can insert one. An optional off-database journal closes that. It is an append-only JSONL file with an HMAC-SHA256 hash chain (each record's MAC covers the previous MAC and the record's fields). `request`, `approve`, `reject`, `withdraw` and `consume` each append -a record inside the same transaction as the row change. +a record before the row change commits. The journal is not inside SQLite, so +a `consume` line is written before `commit`. If that transaction then rolls +back (the event insert hits a used idempotency key, or the commit fails), +the broker appends `consume_void` while it still holds the write lock. The +row stays approved and unconsumed, and a retry can consume it. Clearing +`consumed_at` in SQLite does not void a consume that did commit: only the +key holder can append `consume_void`, and only for a consume this process +has not committed. | `PTASK_APPROVAL_JOURNAL` | Behaviour | |---|---| | unset, `off`, `0`, `false` or `no` (default) | Unchanged. Nothing is read or written. | | `shadow` | Records are appended; a consume the journal would refuse is logged and allowed. Journal errors are logged only. | -| `enforce` | `consume` must find, in a verified chain, an unconsumed `approve` record whose digest (the action binding), kind, approver and `expires_at` equal the row's. Otherwise it exits **7** and the latch rolls back. A broken, truncated or unreadable journal also refuses. Any other value (a typo such as `enforcd`, or `true`, `on`, `disabled`) is treated as `enforce` and logs a warning, so a mistyped flag never turns the gate off. | +| `enforce` | `consume`, `payload` and `verify` must find, in a verified chain, an unconsumed `approve` record whose digest (the action binding), kind, approver and `expires_at` equal the row's. Otherwise they exit **7**. `consume` rolls its latch back; `payload` and `verify` do not write one. A broken, truncated or unreadable journal also refuses. Any other value (a typo such as `enforcd`, or `true`, `on`, `disabled`) is treated as `enforce` and logs a warning, so a mistyped flag never turns the gate off. | Configure with `PTASK_APPROVAL_JOURNAL_PATH` and `PTASK_APPROVAL_JOURNAL_KEY_FILE` (at least 16 bytes; never logged). Put both diff --git a/docs/gen/pt.1 b/docs/gen/pt.1 index 4e19823..e6c2cda 100644 --- a/docs/gen/pt.1 +++ b/docs/gen/pt.1 @@ -1,6 +1,6 @@ .ie \n(.g .ds Aq \(aq .el .ds Aq ' -.TH pt 1 "pt 3.52.3" +.TH pt 1 "pt 3.52.8" .SH NAME pt \- Sovereign task manager for PureTensor .SH SYNOPSIS @@ -208,4 +208,4 @@ Generate shell completions (bash/zsh/fish) to stdout pt\-help(1) Print this message or the help of the given subcommand(s) .SH VERSION -v3.52.3 +v3.52.8