fix(node): read marker refs via rev-parse so present refs return their oid (#558)

- One post_receive_replication_tail per push: the carve-out spawn and the
  durability-bound spawn are mutually exclusive positions on the same
  proven-landed ref set; inv22 now fails on any unfiltered or ungated spawn.
- Marker reconcile: bounded-read errors (git Err/deadline) stay prepared
  and retry instead of quarantining as 'missing marker ref'; only a
  successfully read marker proves missing or mismatch. Err-injection test
  pins no-quarantine on unproven reads.
- Claimed webhook delivery: the synchronous claim phase returns the
  claimed Webhook rows so the send task never re-lists; a list failure
  after claim can no longer drop claimed deliveries. Source gate pins it.

Cargo.lock

@@ -3085,7 +3085,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -6228,7 +6228,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -7162,7 +7162,7 @@ dependencies = [
  "getrandom 0.4.2",
  "once_cell",
  "rustix",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]

crates/gitlawb-node/src/cert.rs

@@ -11,6 +11,8 @@ use uuid::Uuid;
 use crate::db::RefCertificate;
 use crate::state::AppState;
 
+/// Issue a signed ref-update certificate for a successful push. The
+/// live receive-pack handler calls this on every successful push.
 /// Build the v1 signing payload — the EXACT bytes the node signs and
 /// gl's `verify_signature` reconstructs. This is the single source
 /// of truth shared by `issue_ref_certificate` and the frozen
@@ -81,8 +83,36 @@ pub(crate) fn v2_signing_payload(
 
 /// Issue a signed ref-update certificate for a successful push.
 ///
-/// Builds a canonical JSON payload, signs it with the node's Ed25519 key,
-/// persists the certificate, and returns it.
+/// `cert_id` is the deterministic id derived from `(request_id,
+/// ref_name)` (see [`crate::db::ref_cert_id_for`]). It is required so
+/// the recovery drain and the live handler produce the same primary
+/// key: a live push followed by a recovery pass collapses to a
+/// single cert row, and a re-push to the same `(repo, ref)` updates
+/// the existing row's `old_sha` / `new_sha` / `pusher_did` /
+/// `issued_at` / `signature` to the new transition while preserving
+/// the original `id` (the `insert_ref_certificate` upsert is
+/// keyed on `(repo_id, ref_name)` and only updates fields when the
+/// new `issued_at` is strictly greater).
+///
+/// #26 Split PR 1 P1-B: the live handler routes through this
+/// function (the upsert). After the reviewer-1 round-2 fix, the
+/// recovery drain also routes through this function (P1: refresh a
+/// stale cert), so both paths use the same deterministic `cert_id`
+/// and the same upsert. A re-pass is always safe:
+///
+/// - Live handler → live upsert: re-push updates the row, preserves
+///   the original `id`. The contract pinned by
+///   `insert_ref_certificate_upserts_on_repo_ref` is restored.
+/// - Live handler → recovery: live's `ON CONFLICT (id) DO UPDATE`
+///   preserves the original `id`; the recovery's same upsert is
+///   a no-op for an equal-`issued_at` re-run and a refresh for a
+///   strictly-newer one.
+/// - Recovery → live handler: the recovery wrote a row with the
+///   deterministic `id`; the live upsert (which preserves `id` and
+///   only updates other fields when `issued_at` is strictly newer)
+///   is a no-op for an equal-`issued_at` re-run and a refresh for
+///   a strictly-newer one.
+#[allow(dead_code)] // round-trip test in db/mod.rs pins the upsert contract; the live path and the drain use issue_ref_certificate_with_issued_at
 pub async fn issue_ref_certificate(
     state: &AppState,
     repo_id: &str,
@@ -90,9 +120,92 @@ pub async fn issue_ref_certificate(
     old_sha: &str,
     new_sha: &str,
     pusher_did: &str,
+    cert_id: &str,
+) -> Result<RefCertificate> {
+    issue_ref_certificate_with_issued_at(
+        state, repo_id, ref_name, old_sha, new_sha, pusher_did, cert_id, None,
+    )
+    .await
+}
+
+/// #26 Split PR 1 round 4 — variant that lets the caller stamp the
+/// cert's `issued_at` with a transition-time timestamp instead of
+/// `Utc::now()`. The recovery drain passes the persisted
+/// `row.created_at` so a replay after a later live cert does not
+/// outrank the live cert in the `EXCLUDED.issued_at >
+/// ref_certificates.issued_at` upsert guard.
+///
+/// The live handler uses the default `issue_ref_certificate` (no
+/// override), which keeps `Utc::now()` — the reviewer's invariant
+/// is that `issued_at` reflects the transition time, and for a
+/// live push the transition time and the wall-clock are the same.
+///
+/// `issued_at_override` is honored verbatim; passing a value not in
+/// RFC 3339 form is a logic bug (the upsert will mis-order), so
+/// callers must use the row's persisted `created_at`.
+///
+/// # Clippy allow — too many arguments
+/// This is the explicit "stamp a transition-time `issued_at`"
+/// variant of `issue_ref_certificate`. The drain
+/// (`durable_outbox::derive_one`) is the in-crate caller; the
+/// test `replay_of_stale_row_does_not_overwrite_live_cert_b` pins
+/// the contract that a recovery replay's `issued_at` does NOT
+/// outrank a later live cert. Adding a struct-arg would be a
+/// larger refactor for two callers (live + drain) and obscure the
+/// parallel to `issue_ref_certificate` (which is `#[allow]`'d for
+/// the same reason historically).
+#[allow(clippy::too_many_arguments)]
+pub async fn issue_ref_certificate_with_issued_at(
+    state: &AppState,
+    repo_id: &str,
+    ref_name: &str,
+    old_sha: &str,
+    new_sha: &str,
+    pusher_did: &str,
+    cert_id: &str,
+    issued_at_override: Option<String>,
+) -> Result<RefCertificate> {
+    let cert = build_ref_certificate(
+        state,
+        repo_id,
+        ref_name,
+        old_sha,
+        new_sha,
+        pusher_did,
+        Some(cert_id.to_string()),
+        issued_at_override,
+    )
+    .await?;
+    state.db.insert_ref_certificate(&cert).await
+}
+
+/// Shared cert construction: build the JSON payload, sign it with the
+/// node key, and assemble the `RefCertificate` row. `cert_id_override`
+/// lets the recovery path plug in a deterministic id; the live path
+/// passes `None` and gets a fresh UUID. `issued_at_override` lets
+/// the recovery path stamp the cert with the original transition
+/// time so the upsert's `issued_at > issued_at` guard correctly
+/// orders transitions regardless of write order.
+#[allow(clippy::too_many_arguments)]
+async fn build_ref_certificate(
+    state: &AppState,
+    repo_id: &str,
+    ref_name: &str,
+    old_sha: &str,
+    new_sha: &str,
+    pusher_did: &str,
+    cert_id_override: Option<String>,
+    issued_at_override: Option<String>,
 ) -> Result<RefCertificate> {
     let node_did = state.node_did.to_string();
-    let issued_at = Utc::now().to_rfc3339();
+    // P1 (reviewer-1 round 4): when the caller passes a transition-
+    // time `issued_at` (the recovery drain passes `row.created_at`),
+    // use it verbatim so the upsert's per-column guard
+    // `EXCLUDED.issued_at > ref_certificates.issued_at` correctly
+    // orders transitions regardless of write order. The live handler
+    // passes `None` and gets `Utc::now()` — for a live push the
+    // transition time and the wall-clock are the same.
+    let issued_at = issued_at_override.unwrap_or_else(|| Utc::now().to_rfc3339());
 
     // Build the canonical signing payload via the shared builder so
     // the frozen-vector test below and the live signer cannot drift
@@ -104,8 +217,9 @@ pub async fn issue_ref_certificate(
 
     let signature = state.node_keypair.sign_b64(&payload_bytes);
 
+    let id = cert_id_override.unwrap_or_else(|| Uuid::new_v4().to_string());
     let cert = RefCertificate {
-        id: Uuid::new_v4().to_string(),
+        id,
         repo_id: repo_id.to_string(),
         ref_name: ref_name.to_string(),
         old_sha: old_sha.to_string(),
@@ -376,14 +490,16 @@ mod v1_payload_tests {
         let old = "0".repeat(40);
         let new = "a".repeat(40);
         let pusher = "did:key:z6MkPusher";
+        let cert_id = "cert-issuer-observe";
 
         // Drive the live issuer and read the persisted row back by
         // id, so the test covers the stored `version` column and
         // not just the in-process return value.
-        let cert =
-            crate::cert::issue_ref_certificate(&state, repo_id, ref_name, &old, &new, pusher)
-                .await
-                .expect("issue_ref_certificate must succeed");
+        let cert = crate::cert::issue_ref_certificate(
+            &state, repo_id, ref_name, &old, &new, pusher, cert_id,
+        )
+        .await
+        .expect("issue_ref_certificate must succeed");
 
         // The cert returned by the issuer is the source of truth
         // for both the version claim and the signed bytes.

crates/gitlawb-node/src/config.rs

@@ -774,6 +774,45 @@ pub struct Config {
     )]
     pub pin_repair_sweep_delay_secs: u64,
 
+    /// #26 Split PR 1 step 4 — receive-pack queue retention window.
+    /// Terminal `complete` and `rejected_at_git` rows older than
+    /// this are eligible for the periodic purge. `quarantined`
+    /// rows are never purged on a timer. The v30 partial index
+    /// `idx_receive_pack_requests_completed_at` keeps the scan
+    /// cheap regardless of the value.
+    #[arg(
+        long,
+        env = "GITLAWB_QUEUE_RETENTION_DAYS",
+        default_value_t = 7,
+        value_parser = clap::builder::RangedU64ValueParser::<i64>::new().range(1..=365)
+    )]
+    pub queue_retention_days: i64,
+
+    /// #26 Split PR 1 step 4 — receive-pack queue purge batch size.
+    /// Each periodic purge pass deletes at most this many terminal
+    /// rows per batch. The drain and the purge share the same
+    /// `DRAIN_PER_PASS_LIMIT` budget; see the spawn function in
+    /// `main.rs` for the wiring.
+    #[arg(
+        long,
+        env = "GITLAWB_QUEUE_PURGE_BATCH",
+        default_value_t = 1000,
+        value_parser = clap::builder::RangedU64ValueParser::<i64>::new().range(1..=100_000)
+    )]
+    pub queue_purge_batch: i64,
+
+    /// #26 Split PR 1 step 5 — effects-executor retry bound. The
+    /// drain flips a request to `quarantined` after this many
+    /// `EffectsOutcome::Retry` returns, closing the
+    /// infinite-retry DoS window. Default 8 matches the spec; set
+    /// lower in tests.
+    #[arg(
+        long,
+        env = "GITLAWB_EFFECTS_MAX_ATTEMPTS",
+        default_value_t = 8,
+        value_parser = clap::builder::RangedU64ValueParser::<i32>::new().range(1..=1000)
+    )]
+    pub effects_max_attempts: i32,
     /// Per-client-IP rate limit for the anonymous task read routes
     /// (`GET /api/v1/tasks`, `GET /api/v1/tasks/{id}`), in requests per hour.
     /// Both are publicly reachable (`optional_signature`). `GET /api/v1/tasks`
@@ -1285,6 +1324,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn queue_lifecycle_knobs_default_conservatively() {
+        let c = Config::parse_from(["gitlawb-node"]);
+        // 7-day retention matches the v30 partial index comment
+        // and the spec at .gravirei/plans/state-model-durable-post-receive.md.
+        assert_eq!(c.queue_retention_days, 7);
+        // 1000 rows per pass matches DRAIN_PER_PASS_LIMIT in durable_outbox.
+        assert_eq!(c.queue_purge_batch, 1000);
+
+        assert_eq!(
+            Config::parse_from(["gitlawb-node", "--queue-retention-days", "30"])
+                .queue_retention_days,
+            30
+        );
+        assert!(Config::try_parse_from(["gitlawb-node", "--queue-retention-days", "0"]).is_err());
+        assert!(Config::try_parse_from(["gitlawb-node", "--queue-retention-days", "366"]).is_err());
+
+        assert_eq!(
+            Config::parse_from(["gitlawb-node", "--queue-purge-batch", "500"]).queue_purge_batch,
+            500
+        );
+        assert!(Config::try_parse_from(["gitlawb-node", "--queue-purge-batch", "0"]).is_err());
+    }
+
     #[test]
     fn ipfs_walk_per_source_defaults_and_rejects_out_of_range() {
         assert_eq!(Config::parse_from(["gitlawb-node"]).ipfs_walk_per_source, 4);

crates/gitlawb-node/src/git/smart_http.rs

@@ -220,6 +220,7 @@ pub fn response_served_pack(output: &[u8]) -> bool {
 ///
 /// Accepts a push. The caller MUST verify HTTP Signature auth before
 /// calling this function.
+#[allow(dead_code)] // used by tests; production uses receive_pack_raw
 pub async fn receive_pack(
     git_bin: &str,
     repo_path: &Path,
@@ -244,6 +245,219 @@ pub async fn receive_pack(
         .body(Body::from(output))?)
 }
 
+/// Run `git-receive-pack` and return the raw stdout bytes together with
+/// the process exit status. Unlike [`receive_pack`], this does NOT bail
+/// on a non-zero exit: the caller needs the stdout (which contains the
+/// report-status with per-ref ok/ng results) even when the process
+/// exits non-zero. A timeout still returns `Err`.
+///
+/// `reflog_action` (typically `gitlawb-request:<request_id>`) is
+/// exported as `GIT_REFLOG_ACTION` so the ref transaction writes
+/// request-bound landing evidence into the reflog. Recovery requires
+/// that evidence before auto-promoting a stranded row; without it a
+/// later request recreating the same old/new tuple is
+/// indistinguishable from the stranded one.
+#[allow(dead_code)]
+pub async fn receive_pack_raw(
+    git_bin: &str,
+    repo_path: &Path,
+    request_body: Bytes,
+    timeout: Duration,
+    admission: Option<AdmissionGuard>,
+) -> Result<(Vec<u8>, bool)> {
+    receive_pack_raw_with_reflog(git_bin, repo_path, request_body, timeout, admission, None).await
+}
+
+/// [`receive_pack_raw`] with explicit reflog binding.
+pub async fn receive_pack_raw_with_reflog(
+    git_bin: &str,
+    repo_path: &Path,
+    request_body: Bytes,
+    timeout: Duration,
+    admission: Option<AdmissionGuard>,
+    reflog_action: Option<&str>,
+) -> Result<(Vec<u8>, bool)> {
+    let mut command = Command::new(git_bin);
+    command
+        .arg("receive-pack")
+        .arg("--stateless-rpc")
+        .arg(repo_path);
+    if let Some(action) = reflog_action {
+        command.env("GIT_REFLOG_ACTION", action);
+    }
+    let (out, err, status, _admission) = drive_git_child_raw(
+        command,
+        request_body,
+        timeout,
+        "git-receive-pack",
+        admission,
+    )
+    .await?;
+    // On timeout, drive_git_child_raw returns Err — we never reach here.
+    // On success/non-zero exit, we have the stdout and exit status.
+    if !status.success() {
+        let stderr = String::from_utf8_lossy(&err);
+        tracing::warn!(stderr = %stderr, "git-receive-pack exited non-zero");
+    }
+    Ok((out, status.success()))
+}
+
+/// Parse the git-receive-pack report-status output to determine per-ref
+/// success/failure. Returns `(unpack_ok, per_ref_results)` where
+/// `per_ref_results` is a list of `(ref_name, is_ok)`.
+///
+/// The report-status format (after the sideband framing) is:
+/// ```text
+/// unpack ok\n          (or: unpack fail\n)
+/// ok <refname>\n       (per successful ref)
+/// ng <refname> <reason>\n  (per rejected ref)
+/// \n                   (empty line terminates)
+/// ```
+///
+/// Returns `None` if the output cannot be parsed (e.g. the client did
+/// not request report-status, or the output is truncated). In that
+/// case the caller should treat all refs as uncertain.
+pub fn parse_report_status(output: &[u8]) -> Option<(bool, Vec<(String, bool)>)> {
+    let text = std::str::from_utf8(output).ok()?;
+    // Strip sideband framing: each line starts with a pkt-line length
+    // prefix and a channel byte (1 = stdout, 2 = stderr). The actual
+    // data starts after the first `0000` flush packet or after we
+    // strip sideband bytes.
+    //
+    // Framing completeness is part of authority, at whichever level the
+    // framing lives. A double-framed capture carries the report stream
+    // (itself pkt-line framed, flush-terminated) inside the sideband
+    // envelope — real git output has been observed with NO outer flush,
+    // so the outer level cannot require one; the inner frame's flush is
+    // authoritative there. A single-framed (or unframed) payload has no
+    // inner frame, so its only completion marker is the outer flush:
+    // output ending after a complete `unpack ok` / `ok <ref>` packet but
+    // before the outer `0000` is a truncated report and yields `None`
+    // (uncertain), not a successful report.
+    let (stripped, outer_flush) = strip_sideband(text, false)?;
+    // The report-status vocabulary (`unpack ...`, `ok ...`, `ng ...`)
+    // never opens with four hex digits, so a hex quartet here is inner
+    // pkt-line framing, not report text.
+    let stripped = if stripped.len() >= 4
+        && stripped.as_bytes()[..4]
+            .iter()
+            .all(|b| b.is_ascii_hexdigit())
+    {
+        strip_sideband(&stripped, true)?.0
+    } else if !outer_flush {
+        return None;
+    } else {
+        stripped
+    };
+    // The report is framed TWICE when the client negotiated side-band-64k,
+    // which `git push` over smart HTTP does — so this is the common case, not
+    // an exotic one. The outer frame is the side-band envelope; band 1 carries
+    // the report-status stream, which is ITSELF pkt-line encoded. Real bytes
+    // from git 2.50.1 rejecting one ref of a two-ref push:
+    //
+    //   0057\x01000eunpack ok\n0028ng refs/heads/main non-fast-forward\n...
+    //
+    // After one pass the first line reads `000eunpack ok`, which fails the
+    // `unpack ok` check below and makes this return None — i.e. "no report",
+    // which the caller treats as inconclusive and keeps every declared ref.
+    // The per-ref gate would then be inert for exactly the pushes it exists to
+    // filter. The second pass above already removed the inner pkt-line
+    // framing where present (and required its flush); it is a no-op
+    // on the single-framed shape, because plain report text does not begin
+    // with four hex digits.
+    let lines: Vec<&str> = stripped.lines().collect();
+    if lines.is_empty() {
+        return None;
+    }
+
+    // First non-empty line is "unpack ok" or "unpack fail".
+    let unpack_line = lines.iter().find(|l| !l.is_empty())?;
+    let unpack_ok = if unpack_line.starts_with("unpack ok") {
+        true
+    } else if unpack_line.starts_with("unpack fail") {
+        false
+    } else {
+        return None;
+    };
+
+    let mut results = Vec::new();
+    for line in &lines[1..] {
+        let line = line.trim();
+        if line.is_empty() {
+            break;
+        }
+        if let Some(rest) = line.strip_prefix("ok ") {
+            results.push((rest.to_string(), true));
+        } else if let Some(rest) = line.strip_prefix("ng ") {
+            // "ng <refname> <reason>" — skip the reason
+            let ref_name = rest.split_whitespace().next()?.to_string();
+            results.push((ref_name, false));
+        }
+    }
+
+    Some((unpack_ok, results))
+}
+
+/// Strip git sideband framing from a pkt-line encoded output.
+/// Sideband-encoded lines start with a 4-hex-digit length, then a
+/// channel byte (0x01=stdout, 0x02=stderr), then payload. Returns the
+/// decoded payload plus whether the stream saw its terminating flush
+/// packet, or `None` if the framing is malformed. With `require_flush`,
+/// a stream that never sees its flush is also `None` — the caller uses
+/// this for the level whose completion marker is authoritative (see
+/// [`parse_report_status`]).
+fn strip_sideband(text: &str, require_flush: bool) -> Option<(String, bool)> {
+    let mut output = String::new();
+    let mut pos = 0;
+    let bytes = text.as_bytes();
+    let mut saw_flush = false;
+
+    loop {
+        if pos == bytes.len() {
+            break;
+        }
+        if pos + 4 > bytes.len() {
+            // Truncated framing mid-header: fail closed.
+            return None;
+        }
+        let len_str = std::str::from_utf8(&bytes[pos..pos + 4]).ok()?;
+        let len = usize::from_str_radix(len_str, 16).ok()?;
+        if len == 0 {
+            // Flush packet — end of sideband stream
+            saw_flush = true;
+            break;
+        }
+        if len < 4 || pos + len > bytes.len() {
+            // Truncated pkt-line extends past EOF: incomplete framing.
+            return None;
+        }
+        let pkt_data = std::str::from_utf8(&bytes[pos + 4..pos + len]).ok()?;
+        pos += len;
+
+        // Sideband: first byte is channel (1=stdout, 2=stderr)
+        if let Some(payload) = pkt_data.strip_prefix('\x01') {
+            output.push_str(payload);
+        } else if pkt_data.starts_with('\x02') {
+            // stderr — skip (git error messages)
+        } else {
+            // Not sideband encoded — pass through
+            output.push_str(pkt_data);
+        }
+    }
+
+    if require_flush && !saw_flush {
+        // Truncated at a packet boundary: every packet parsed, but the
+        // stream never completed. Indistinguishable from a report git was
+        // killed mid-write, so no authority.
+        return None;
+    }
+    if output.is_empty() {
+        None
+    } else {
+        Some((output, saw_flush))
+    }
+}
+
 /// Sends SIGTERM to a child's whole process group on drop, unless disarmed first.
 ///
 /// A served `git upload-pack`/`receive-pack` forks helpers such as `pack-objects`.
@@ -578,6 +792,121 @@ async fn drive_git_child(
     Ok((out, admission))
 }
 
+/// Like [`drive_git_child`], but returns stdout and stderr even on a
+/// non-zero exit status. Used by [`receive_pack_raw`] so the caller
+/// can parse the report-status output from a failed `git-receive-pack`.
+async fn drive_git_child_raw(
+    mut command: Command,
+    input: Bytes,
+    timeout: Duration,
+    what: &str,
+    admission: Option<AdmissionGuard>,
+) -> Result<(
+    Vec<u8>,
+    Vec<u8>,
+    std::process::ExitStatus,
+    Option<AdmissionGuard>,
+)> {
+    command
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+
+    #[cfg(unix)]
+    command.process_group(0);
+
+    let mut child = command.spawn()?;
+
+    let mut stdin = child.stdin.take();
+    let mut stdout = child.stdout.take().context("git stdout was not piped")?;
+    let mut stderr = child.stderr.take().context("git stderr was not piped")?;
+
+    #[cfg(unix)]
+    let pgid = child.id().map(|id| id as i32);
+    #[cfg(unix)]
+    let mut group_guard = KillGroupOnDrop {
+        child: Some(child),
+        pgid,
+        admission,
+    };
+
+    let mut out = Vec::new();
+    let mut err = Vec::new();
+
+    let interact = async {
+        let write = async {
+            match stdin.take() {
+                Some(mut s) => s.write_all(&input).await,
+                None => Ok(()),
+            }
+        };
+        #[cfg(unix)]
+        let child_ref = group_guard.child_mut();
+        #[cfg(not(unix))]
+        let child_ref = &mut child;
+        let (write_result, r_out, r_err, status) = tokio::join!(
+            write,
+            stdout.read_to_end(&mut out),
+            stderr.read_to_end(&mut err),
+            child_ref.wait(),
+        );
+        r_out?;
+        r_err?;
+        Ok::<_, anyhow::Error>((write_result, status?))
+    };
+
+    let timed = tokio::time::timeout(timeout, interact).await;
+    let (write_result, status, admission) = match timed {
+        Ok(result) => {
+            #[cfg(unix)]
+            let admission = group_guard.disarm();
+            let (write_result, status) = result?;
+            (write_result, status, admission)
+        }
+        Err(_elapsed) => {
+            #[cfg(unix)]
+            {
+                reap_group_on_timeout(group_guard.child_mut()).await;
+                drop(group_guard.disarm());
+            }
+            #[cfg(not(unix))]
+            {
+                let _ = child.start_kill();
+                let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await;
+                drop(admission);
+            }
+            return Err(GitServiceTimeout.into());
+        }
+    };
+
+    // Prefer the child's collected answer over a stdin-write error: an early
+    // rejection (unpack failure, command-level deny) makes git print a complete
+    // report-status and exit WITHOUT draining stdin, so a body larger than the
+    // pipe buffer EPIPEs the write while the answer is already on stdout. The
+    // child is reaped (the join above waited on it), so `out`/`err`/`status`
+    // are complete and authoritative — returning Err here would discard them:
+    // `receive_pack_raw_with_reflog` would propagate the write error and the
+    // handler would mark every prepared row `uncertain` instead of driving
+    // `cancelled` per the report. That durable fate is the effect: on a
+    // non-zero exit the client still gets `Err(AppError::Git)` (the handler
+    // maps the failed run itself, not this write), while on exit 0 an error
+    // here would answer a push git actually landed with a server error. The
+    // non-raw `drive_git_child` orders the same way (exit status and git's own
+    // failure first, stdin-write error last — pinned by
+    // `run_git_service_surfaces_git_stderr_over_a_stdin_epipe`), but it must
+    // fail on non-zero because its callers do not parse output from a failed
+    // child; the raw driver exists precisely to parse that output.
+    if let Err(e) = &write_result {
+        tracing::warn!(
+            err = %e,
+            what = %what,
+            "git stdin write failed after child exit; surfacing collected output"
+        );
+    }
+
+    Ok((out, err, status, admission))
+}
+
 /// Map a validated service name to its git subcommand. Only the two fixed
 /// subcommands can come out, so an unrecognized service can never reach git's argv.
 fn service_to_command(service: &str) -> Result<&'static str> {
@@ -868,6 +1197,86 @@ mod tests {
     use std::process::Command;
     use tempfile::TempDir;
 
+    /// Byte-for-byte capture from `git receive-pack` 2.50.1 rejecting one ref of
+    /// a two-ref push, with side-band-64k negotiated — what `git push` over
+    /// smart HTTP actually sends. The report is framed twice: the outer
+    /// side-band envelope, then the report-status stream's own pkt-lines.
+    ///
+    /// This fixture is the point of the test. A hand-written single-framed
+    /// string parses fine with only one strip pass, so a fixture invented to
+    /// match the parser hides the exact case the parser is for.
+    const REAL_SIDEBAND_REPORT: &[u8] =
+        b"0057\x01000eunpack ok\n0028ng refs/heads/main non-fast-forward\n0018ok refs/heads/third\n0000";
+
+    #[test]
+    fn a_real_double_framed_sideband_report_is_read_not_treated_as_absent() {
+        let (unpack_ok, refs) =
+            parse_report_status(REAL_SIDEBAND_REPORT).expect("a real git report must parse");
+        assert!(unpack_ok, "unpack line must be read through both frames");
+        assert_eq!(
+            refs,
+            vec![
+                ("refs/heads/main".to_string(), false),
+                ("refs/heads/third".to_string(), true),
+            ],
+            "the rejected ref must be distinguished from the accepted one"
+        );
+    }
+
+    /// The single-framed shape must keep working: the second pass has to be a
+    /// no-op there, not a corruption. Plain report text does not begin with
+    /// four hex digits, which is what makes that safe.
+    #[test]
+    fn a_single_framed_report_still_parses_after_the_second_pass() {
+        // Framed programmatically rather than by a hand-counted hex prefix: a
+        // wrong length makes the parser return None, which would look exactly
+        // like the regression this pair of tests exists to catch.
+        let payload = "\x01unpack ok\nok refs/heads/x\n";
+        let single = format!("{:04x}{payload}0000", payload.len() + 4);
+        let (unpack_ok, refs) =
+            parse_report_status(single.as_bytes()).expect("single-framed must parse");
+        assert!(unpack_ok);
+        assert_eq!(refs, vec![("refs/heads/x".to_string(), true)]);
+    }
+
+    #[test]
+    fn truncated_mid_packet_report_is_indeterminate_not_partial() {
+        // Truncate a valid single-framed report mid-packet (overrun):
+        // must return None so the handler persists the whole request as
+        // indeterminate instead of committing a partial ok set.
+        let payload = "\x01unpack ok\nok refs/heads/x\n";
+        let single = format!("{:04x}{payload}0000", payload.len() + 4);
+        let truncated = &single.as_bytes()[..single.len() - 6];
+        assert!(
+            parse_report_status(truncated).is_none(),
+            "mid-packet truncation must be indeterminate"
+        );
+    }
+
+    /// Truncation exactly at a packet boundary (every packet complete, but
+    /// the outer flush never arrived — e.g. git killed after the last `ok`
+    /// line) must also be indeterminate: the payload is complete but the
+    /// frame is not, and framing completeness is part of authority. The
+    /// mid-packet sibling above cannot catch this; only the flush gate can.
+    /// Removing the `require_flush` enforcement returns `Some` and turns
+    /// this red.
+    #[test]
+    fn truncated_at_packet_boundary_report_is_indeterminate_not_partial() {
+        let payload = "\x01unpack ok\nok refs/heads/x\n";
+        let single = format!("{:04x}{payload}0000", payload.len() + 4);
+        let no_flush = &single.as_bytes()[..single.len() - 4];
+        assert!(
+            parse_report_status(no_flush).is_none(),
+            "a report missing its outer flush must be indeterminate, \
+             even with every packet intact"
+        );
+        // Sanity: the untruncated fixture still parses (flush present).
+        assert!(
+            parse_report_status(single.as_bytes()).is_some(),
+            "the same bytes with the flush must parse"
+        );
+    }
+
     #[test]
     fn service_to_command_maps_allowlist_and_rejects_passthrough() {
         assert_eq!(
@@ -2641,6 +3050,62 @@ mod tests {
         );
     }
 
+    // The raw driver's counterpart: a rejection-shaped early exit where git
+    // prints a COMPLETE report-status and exits without draining stdin, so a
+    // body larger than the pipe buffer (~64 KiB) EPIPEs the write while the
+    // answer is already on stdout. `receive_pack_raw` must surface that report
+    // (Ok + stdout + exit status), not discard it as a stdin-write error — an
+    // Err here loses parseable output: `receive_pack_raw_with_reflog`
+    // propagates it and the handler marks every prepared row `uncertain`
+    // instead of driving `cancelled` from the report (the durable fate is the
+    // effect; the client-visible status on a failed run is `AppError::Git`
+    // either way). Goes RED if the stdin-write check returns Err after the
+    // join. Both exit codes are pinned: receive-pack rejections and successes
+    // both carry reports the handler must parse.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn receive_pack_raw_surfaces_report_status_over_a_stdin_epipe() {
+        for exit_code in [0, 1] {
+            let tmp = tempfile::TempDir::new().unwrap();
+            std::fs::write(tmp.path().join("report.bin"), REAL_SIDEBAND_REPORT).unwrap();
+            // Never reads stdin: the pending body blocks the pipe, then EPIPEs
+            // once the fake exits. `cat` delivers the report first.
+            let git_bin = write_fake_git(
+                tmp.path(),
+                &format!("#!/bin/sh\ncat \"$(dirname \"$0\")/report.bin\"\nexit {exit_code}\n"),
+            );
+            let body = Bytes::from(vec![0u8; 256 * 1024]);
+            let (out, ok) = receive_pack_raw(
+                git_bin.to_str().unwrap(),
+                tmp.path(),
+                body,
+                Duration::from_secs(60),
+                None,
+            )
+            .await
+            .expect(
+                "a report-status printed before exit must not be discarded as a \
+                 stdin-write error",
+            );
+            assert_eq!(
+                ok,
+                exit_code == 0,
+                "the exit status rides the Ok return for both report shapes"
+            );
+            let (unpack_ok, results) = parse_report_status(&out)
+                .expect("git's report survives the EPIPE and stays parseable");
+            assert!(unpack_ok, "unpack ok parses");
+            assert_eq!(
+                results,
+                vec![
+                    ("refs/heads/main".to_string(), false),
+                    ("refs/heads/third".to_string(), true),
+                ],
+                "the per-ref report reaches the handler instead of an HTTP error"
+            );
+        }
+    }
+
     // ── #192 F1: response_served_pack detector (real git output) ────────────
 
     // A completing request (want+flush+done) yields NAK + a real packfile;

crates/gitlawb-node/src/git/visibility_pack.rs

@@ -334,6 +334,111 @@ pub(crate) fn run_bounded_git(
     Ok(out)
 }
 
+/// Fail closed unless every ref ultimately resolves to a commit (a ref pointing
+/// directly at a blob or tree, or an annotated tag — even a nested one — of such
+/// an object is refused). `git rev-list --all` silently *skips* such refs, but
+/// `git upload-pack` (serve) and the whole-repo pin fallback
+/// (`git cat-file --batch-all-objects`) still expose their target object, so a
+/// tolerant walk would under-withhold. Refuse rather than leak.
+///
+/// Each ref is peeled fully with `<ref>^{}` through `git cat-file --batch-check`.
+/// Full peeling is why this is not `for-each-ref %(*objecttype)`, which
+/// dereferences only one tag level and so misclassifies a tag-of-a-tag-of-a-
+/// commit as a non-commit.
+///
+/// Internal node-managed refs are exempt from this check since they point
+/// to blobs by design, not repository content: request markers
+/// (`refs/gitlawb/requests/*`, per-push marker blobs) and issue records
+/// (`refs/gitlawb/issues/*`, JSON blobs). The exemption lists exactly
+/// those two namespaces: any other `refs/gitlawb/*` ref or non-commit
+/// content ref still fails closed. Pushes into `refs/gitlawb/*` are
+/// refused at the push edge, so no new bypass ref can appear; the
+/// visibility logic only cares about content refs, not these two
+/// internal metadata namespaces whose binding and advertisement hiding
+/// are verified on the push path.
+fn assert_all_refs_are_commits(repo_path: &Path, git_bin: &str, deadline: Instant) -> Result<()> {
+    let refs_out = run_bounded_git(
+        git_bin,
+        &["for-each-ref", "--format=%(refname)"],
+        repo_path,
+        b"",
+        deadline,
+    )?;
+    let refs_stdout = String::from_utf8_lossy(&refs_out);
+    let refnames: Vec<&str> = refs_stdout
+        .lines()
+        .map(str::trim)
+        .filter(|l| !l.is_empty())
+        .collect();
+    if refnames.is_empty() {
+        return Ok(());
+    }
+
+    // Filter out only the two node-managed blob namespaces before checking
+    // object types. Scoped to `refs/gitlawb/requests/` and
+    // `refs/gitlawb/issues/` — not all of `refs/gitlawb/*` — so any
+    // other internal or content ref pointing at a non-commit still
+    // fails closed.
+    let content_refs: Vec<&str> = refnames
+        .iter()
+        .filter(|r| {
+            !r.starts_with("refs/gitlawb/requests/") && !r.starts_with("refs/gitlawb/issues/")
+        })
+        .copied()
+        .collect();
+
+    if content_refs.is_empty() {
+        return Ok(());
+    }
+
+    // Peel every content ref in one `git cat-file --batch-check` pass: one `<refname>^{}`
+    // query per line, one output line per input line, in order. cat-file echoes the
+    // full query on a `<query> missing` line, so output scales with refname length;
+    // run_bounded_git drains stdout concurrently with the stdin write, so the pipe
+    // cannot deadlock, and the whole peel is bounded by the shared walk deadline.
+    let queries = content_refs
+        .iter()
+        .map(|r| format!("{r}^{{}}"))
+        .collect::<Vec<_>>()
+        .join("\n");
+    let peel_out = run_bounded_git(
+        git_bin,
+        &["cat-file", "--batch-check=%(objecttype)"],
+        repo_path,
+        queries.as_bytes(),
+        deadline,
+    )?;
+
+    let peel_stdout = String::from_utf8_lossy(&peel_out);
+    let types: Vec<&str> = peel_stdout.lines().map(str::trim).collect();
+    // A short read means at least one ref went unclassified — fail closed.
+    if types.len() != content_refs.len() {
+        anyhow::bail!(
+            "git cat-file returned {} lines for {} refs; \
+             refusing to produce a partial (under-withheld) set",
+            types.len(),
+            content_refs.len()
+        );
+    }
+    for (refname, kind) in content_refs.iter().zip(types.iter()) {
+        // git emits `<query> missing` (not the objecttype) when the peel target
+        // is absent; the status word is the last token.
+        if kind.split_ascii_whitespace().last() == Some("missing") {
+            anyhow::bail!(
+                "ref {refname} does not resolve to an object; \
+                 refusing to produce a partial (under-withheld) set"
+            );
+        }
+        if *kind != "commit" {
+            anyhow::bail!(
+                "ref {refname} resolves to a {kind}, not a commit; \
+                 refusing to produce a partial (under-withheld) set"
+            );
+        }
+    }
+    Ok(())
+}
+
 /// List every (blob_oid, "/repo/relative/path") pair reachable from any commit in
 /// `repo_path` — every ref *and* every historical commit those refs reach, not just
 /// the ref tips. `git upload-pack` (serve) and the whole-repo pin fallback
@@ -1424,84 +1529,17 @@ fn pair_decision(
     }
 }
 
-/// Reachable blob AND withheld-subtree tree OIDs the anonymous (or `caller`)
-/// replication audience must not receive. The single source for
-/// `replication_withheld_set`: a blob-only set handed to [`replicable_objects`]
-/// reintroduces #172 (withheld trees pin to public IPFS).
-///
-/// Strict where [`tree_paths`] is lenient. This feeds a withheld filter, so
-/// absence means replicate; a missed reachable tree would leak. The walk
-/// therefore matches [`blob_paths`]: [`assert_all_refs_are_commits`] first, then
-/// a failing `rev-list --all [HEAD]` is an error, then [`object_paths`] (already
-/// fail-closed on ls-tree / non-UTF-8) plus [`root_tree_pairs`]. [`tree_paths`]
-/// stays the lenient allow-list walk for GET /ipfs/{cid} and the full-scan
-/// allowed-tree set, where absence withholds.
-fn assert_all_refs_are_commits(repo_path: &Path, git_bin: &str, deadline: Instant) -> Result<()> {
-    let refs_out = run_bounded_git(
-        git_bin,
-        &["for-each-ref", "--format=%(refname)"],
-        repo_path,
-        b"",
-        deadline,
-    )?;
-    let refs_stdout = String::from_utf8_lossy(&refs_out);
-    let refnames: Vec<&str> = refs_stdout
-        .lines()
-        .map(str::trim)
-        .filter(|l| !l.is_empty())
-        .collect();
-    if refnames.is_empty() {
-        return Ok(());
-    }
-
-    // Peel every ref in one `git cat-file --batch-check` pass: one `<refname>^{}`
-    // query per line, one output line per input line, in order. cat-file echoes the
-    // full query on a `<query> missing` line, so output scales with refname length;
-    // run_bounded_git drains stdout concurrently with the stdin write, so the pipe
-    // cannot deadlock, and the whole peel is bounded by the shared walk deadline.
-    let queries = refnames
-        .iter()
-        .map(|r| format!("{r}^{{}}"))
-        .collect::<Vec<_>>()
-        .join("\n");
-    let peel_out = run_bounded_git(
-        git_bin,
-        &["cat-file", "--batch-check=%(objecttype)"],
-        repo_path,
-        queries.as_bytes(),
-        deadline,
-    )?;
-
-    let peel_stdout = String::from_utf8_lossy(&peel_out);
-    let types: Vec<&str> = peel_stdout.lines().map(str::trim).collect();
-    // A short read means at least one ref went unclassified — fail closed.
-    if types.len() != refnames.len() {
-        anyhow::bail!(
-            "git cat-file returned {} lines for {} refs; \
-             refusing to produce a partial (under-withheld) set",
-            types.len(),
-            refnames.len()
-        );
-    }
-    for (refname, kind) in refnames.iter().zip(types.iter()) {
-        // git emits `<query> missing` (not the objecttype) when the peel target
-        // is absent; the status word is the last token.
-        if kind.split_ascii_whitespace().last() == Some("missing") {
-            anyhow::bail!(
-                "ref {refname} does not resolve to an object; \
-                 refusing to produce a partial (under-withheld) set"
-            );
-        }
-        if *kind != "commit" {
-            anyhow::bail!(
-                "ref {refname} resolves to a {kind}, not a commit; \
-                 refusing to produce a partial (under-withheld) set"
-            );
-        }
-    }
-    Ok(())
-}
-
+/// Root tree oid of every reachable commit, at "/". `ls-tree` never emits a commit's
+/// own root tree (it lists entries *under* a tree), so it is added explicitly here.
+/// Resolved in ONE bounded `git log --no-walk --format=%T --stdin` pass over the
+/// shared commit set — not a per-commit `rev-parse` — so a tree-set walk costs the
+/// same subprocess order as the blob walk. The commit oids go on STDIN, not argv: a
+/// long history has tens of thousands of reachable commits, and passing them all as
+/// arguments overflows ARG_MAX so `git log` fails to spawn — which the caller treats
+/// as a walk error and fail-closed 404s an authorized reader of a reachable/root
+/// tree (#173 P2). `run_bounded_git` drains stdout concurrently with the stdin
+/// write, so a large history cannot deadlock the pipes. A commit whose root tree git
+/// cannot resolve fails the pass (bail), failing closed.
 fn root_tree_pairs(
     repo_path: &Path,
     git_bin: &str,
@@ -1533,6 +1571,18 @@ fn root_tree_pairs(
     Ok(set)
 }
 
+/// Reachable blob AND withheld-subtree tree OIDs the anonymous (or `caller`)
+/// replication audience must not receive. The single source for
+/// `replication_withheld_set`: a blob-only set handed to [`replicable_objects`]
+/// reintroduces #172 (withheld trees pin to public IPFS).
+///
+/// Strict where [`tree_paths`] is lenient. This feeds a withheld filter, so
+/// absence means replicate; a missed reachable tree would leak. The walk
+/// therefore matches [`blob_paths`]: [`assert_all_refs_are_commits`] first, then
+/// a failing `rev-list --all [HEAD]` is an error, then [`object_paths`] (already
+/// fail-closed on ls-tree / non-UTF-8) plus [`root_tree_pairs`]. [`tree_paths`]
+/// stays the lenient allow-list walk for GET /ipfs/{cid} and the full-scan
+/// allowed-tree set, where absence withholds.
 pub fn withheld_object_oids_bounded(
     repo_path: &Path,
     git_bin: &str,
@@ -5963,6 +6013,77 @@ esac\n";
         );
     }
 
+    /// A planted non-exempt `refs/gitlawb/*` blob ref fails the commit walk
+    /// closed, while the two exempt namespaces (`requests/`, `issues/`) pass.
+    /// The exemption narrowing removed a serving path, so this test pins the
+    /// denial: widening the exemption back to all of `refs/gitlawb/*` turns
+    /// the expect_err red. Real git, scratch bare repo.
+    #[test]
+    fn planted_non_exempt_gitlawb_blob_ref_fails_closed() {
+        use std::time::{Duration, Instant};
+        let td = TempDir::new().unwrap();
+        let bare = td.path().join("repo.git");
+        std::fs::create_dir_all(&bare).unwrap();
+        let run = |args: &[&str]| {
+            assert!(
+                Command::new("git")
+                    .args(args)
+                    .current_dir(&bare)
+                    .status()
+                    .unwrap()
+                    .success(),
+                "git {args:?} failed"
+            );
+        };
+        let run_out = |args: &[&str]| {
+            let out = Command::new("git")
+                .args(args)
+                .current_dir(&bare)
+                .output()
+                .unwrap();
+            assert!(out.status.success(), "git {args:?} failed");
+            String::from_utf8_lossy(&out.stdout).trim().to_string()
+        };
+        run(&["init", "-q", "--bare", "."]);
+        // A blob object to point refs at (node markers and issues are blobs).
+        std::fs::write(bare.join("marker-body"), b"marker\n").unwrap();
+        let blob = run_out(&["hash-object", "-w", "marker-body"]);
+        let check = |want_ok: bool, why: &str| {
+            let result =
+                assert_all_refs_are_commits(&bare, "git", Instant::now() + Duration::from_secs(30));
+            if want_ok {
+                result.expect(why);
+            } else {
+                let err = result.expect_err(why);
+                assert!(
+                    err.to_string().contains("refs/gitlawb/evil"),
+                    "the denial must name the offending ref, got: {err:#}"
+                );
+            }
+        };
+        // The bare `refs/gitlawb` name fails closed too (checked first: git
+        // forbids a bare ref coexisting with the subtree below, D/F conflict).
+        run(&["update-ref", "refs/gitlawb", &blob]);
+        let result =
+            assert_all_refs_are_commits(&bare, "git", Instant::now() + Duration::from_secs(30));
+        let err = result.expect_err("bare refs/gitlawb blob ref must fail closed");
+        assert!(
+            err.to_string().contains("refs/gitlawb"),
+            "the denial must name the offending ref, got: {err:#}"
+        );
+        run(&["update-ref", "-d", "refs/gitlawb", "--no-deref"]);
+        // Exempt namespaces with blob targets pass (node-managed by design).
+        run(&["update-ref", "refs/gitlawb/requests/r1", &blob]);
+        run(&["update-ref", "refs/gitlawb/issues/i1", &blob]);
+        check(true, "requests//issues/ blob refs are exempt and must pass");
+        // A non-exempt blob ref fails closed.
+        run(&["update-ref", "refs/gitlawb/evil", &blob]);
+        check(false, "a non-exempt refs/gitlawb blob ref must fail closed");
+        // Cleanup proves the denial is about that ref, not residue.
+        run(&["update-ref", "-d", "refs/gitlawb/evil", "--no-deref"]);
+        check(true, "removing the evil ref must restore the walk");
+    }
+
     /// #218 review round 9 (guidance #1 — single fail-closed
     /// classification contract): for every non-commit ref shape the
     /// parser can produce, the FOUR consumers of `(oid, path)` —

crates/gitlawb-node/src/main.rs

@@ -7,6 +7,7 @@ mod bootstrap;
 mod cert;
 mod config;
 mod db;
+mod durable_outbox;
 mod encrypted_pin;
 mod error;
 mod git;
@@ -624,6 +625,13 @@ async fn main() -> Result<()> {
     }
 
     let _legacy_cid_sweep = spawn_legacy_cid_sweep(&state, &config);
+    let _queue_lifecycle_sweep = spawn_queue_lifecycle_sweep(&state, &config);
+    // NOTE: the due-request worker is spawned AFTER the one-shot startup
+    // reconcile/drain below, not here: its 5s interval would otherwise tick
+    // during operator/PoS setup and interleave with boot recovery on backlog
+    // rows (state gates prevent data loss, but the interleave costs recovery
+    // latency and widens the reconcile/drain race window).
+    let _marker_cleanup_worker = spawn_marker_cleanup_worker(&state);
 
     let router = server::build_router(state.clone());
     // Re-register the socket bound at startup — same fd, so there was never a
@@ -739,6 +747,63 @@ async fn main() -> Result<()> {
     let grace = std::time::Duration::from_secs(config.shutdown_grace_secs);
     info!(grace_secs = config.shutdown_grace_secs, "axum server ready");
 
+    // #26 Split PR 1: drain any `applied` rows left by a previous
+    // process that crashed after Git applied a ref but before the
+    // bookkeeping landed. Runs once, BEFORE the server accepts new
+    // pushes, so a recovery re-derivation does not race a fresh push.
+    // Non-fatal: a transient drain failure is logged and the rows
+    // remain `applied` for the next startup to pick up.
+    //
+    // P1-A: the reconcile step runs FIRST and promotes any `prepared`
+    // or `uncertain` row whose target SHA actually landed on disk.
+    // This is the path that recovers a ref when the post-receive
+    // `mark_pending_ref_transitions_applied` call errored or was
+    // interrupted after `receive_pack` returned Ok, or when
+    // receive-pack returned Err but some refs may have landed
+    // (the `uncertain` state). Without this step, the drain (gated
+    // on `state = 'applied'`) would never see those rows.
+    //
+    // P2 (reviewer-1/2 round 3): use the multi-pass reconcile so
+    // prepared/uncertain rows beyond the first 1000-row page are
+    // processed in the same startup, rather than waiting for the
+    // next restart (where they might age out of MAX_RECONCILE_AGE).
+    match durable_outbox::reconcile_prepared_from_disk_all(
+        state.clone(),
+        durable_outbox::DRAIN_PER_PASS_LIMIT,
+        durable_outbox::DRAIN_MAX_PASSES,
+    )
+    .await
+    {
+        Ok(0) => {}
+        Ok(n) => info!(
+            n,
+            "reconciled prepared/uncertain -> applied via on-disk ref match"
+        ),
+        Err(e) => warn!(
+            err = %e,
+            "pending ref transition reconcile failed at startup (non-fatal; will retry on next start)"
+        ),
+    }
+    match durable_outbox::drain_receive_pack_requests_all(
+        state.clone(),
+        durable_outbox::DRAIN_PER_PASS_LIMIT,
+        durable_outbox::DRAIN_MAX_PASSES,
+    )
+    .await
+    {
+        Ok(0) => {}
+        Ok(n) => info!(n, "drained pending ref transitions from prior run"),
+        Err(e) => warn!(
+            err = %e,
+            "pending ref transition drain failed at startup (non-fatal; will retry on next start)"
+        ),
+    }
+
+    // Start the due-request worker only now that boot recovery is done (see
+    // the note at the other spawns above): its first tick must not interleave
+    // with the one-shot reconcile/drain on backlog rows.
+    let _due_request_worker = spawn_due_request_worker(&state);
+
     // `into_make_service_with_connect_info` exposes the socket peer address as
     // `ConnectInfo<SocketAddr>` so the push limiter can key on the real client
     // when no trusted proxy header applies (see `rate_limit::client_key`).
@@ -771,6 +836,158 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
+/// U4 (#173): spawn the periodic legacy provider-CID repair sweep. Releases before this
+/// version stored the PROVIDER CID (Kubo dag-pb / Pinata CIDv0) in `pinned_cids.cid`,
+/// and this version's `/ipfs/{cid}` resolver withholds any row whose stored key is not
+/// the raw-content CID. The opportunistic repair on the pin path only fires when a push
+/// re-carries the object, which normal git negotiation makes it not do, so those rows
+/// need a walk. DETACHED, never on the boot path: the caller keeps serving while this
+/// runs, and the sweep's own batch bound plus inter-batch delay keep it off the DB's
+/// critical path. Its cursor is durable, so a restart mid-walk resumes instead of
+/// rewinding. It re-arms after every run, including a failed one, so it never returns
+/// and there is no awaited value to log here; the shutdown watcher below is what ends
+/// it.
+///
+/// A named function rather than an inline block in `main` so the WIRING has a seam a
+/// test can call: that the task is spawned at all, that it reads its batch and delay
+/// from the config knobs rather than some other field, that the caller is not blocked
+/// on it, and that the shutdown watcher actually ends it mid-walk. The sweep's own
+/// behavior is covered elsewhere; this is the boot-path half.
+fn spawn_legacy_cid_sweep(state: &AppState, config: &Config) -> tokio::task::JoinHandle<()> {
+    let db = state.db.clone();
+    let repos_dir = config.repos_dir.clone();
+    let git_bin = state.git_bin.clone();
+    let git_timeout = std::time::Duration::from_secs(config.git_service_timeout_secs);
+    let batch = config.pin_repair_sweep_batch;
+    let delay = std::time::Duration::from_secs(config.pin_repair_sweep_delay_secs);
+    let mut shutdown_rx = state.subscribe_shutdown();
+    tokio::spawn(async move {
+        tokio::select! {
+            _ = ipfs_pin::run_sweep_rearmed(
+                &repos_dir, &git_bin, git_timeout, batch, delay,
+                ipfs_pin::SWEEP_REARM_DELAY, &db,
+            ) => {}
+            // Shutdown mid-walk simply drops the run; the persisted cursor means the
+            // next boot picks up where this one stopped.
+            _ = shutdown_rx.changed() => {}
+        }
+    })
+}
+
+/// #26 Split PR 1 step 4 — periodic queue-lifecycle purge. Runs on
+/// the same detached task pattern as `spawn_legacy_cid_sweep`:
+/// tokio::spawn with a shutdown watcher, never on the boot path. The
+/// interval is fixed at 24 hours (the spec calls for "one per
+/// cluster per day"); the inter-batch delay is implicit in the
+/// batch size plus the wall-clock cost of each pass. The drain
+/// (`drain_receive_pack_requests_all`) and the purge
+/// (`purge_request_queue`) share the same `DRAIN_PER_PASS_LIMIT`
+/// budget so a 1000-row purge pass takes roughly the same time as
+/// a 1000-row drain pass.
+fn spawn_queue_lifecycle_sweep(state: &AppState, config: &Config) -> tokio::task::JoinHandle<()> {
+    let app_state = state.clone();
+    let retention_days = config.queue_retention_days;
+    let batch = config.queue_purge_batch;
+    let git_timeout = std::time::Duration::from_secs(config.git_service_timeout_secs.min(30));
+    let mut shutdown_rx = state.subscribe_shutdown();
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(24 * 3600));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        // Consume the immediate first tick: Tokio intervals fire
+        // immediately on creation, which would run the purge
+        // concurrently with startup reconcile/drain. The child purge
+        // is terminal-parent-gated in SQL, but delaying the first
+        // sweep a full period removes the race entirely.
+        interval.tick().await;
+        loop {
+            tokio::select! {
+                _ = interval.tick() => {
+                    if let Err(e) = durable_outbox::purge_request_queue(
+                        &app_state.db,
+                        retention_days,
+                        batch,
+                    ).await {
+                        tracing::warn!(err = %e, "queue lifecycle purge failed; will retry on next tick");
+                    }
+                    // Drain marker tombstones with the bounded deleter;
+                    // failures retain the tombstone for the next tick.
+                    if let Err(e) = durable_outbox::drain_marker_cleanup_queue(
+                        &app_state,
+                        batch,
+                        git_timeout,
+                    ).await {
+                        tracing::warn!(err = %e, "marker cleanup drain failed");
+                    }
+                }
+                _ = shutdown_rx.changed() => {
+                    break;
+                }
+            }
+        }
+    })
+}
+
+/// Shutdown-aware background due-request loop. The startup drain runs
+/// once; without this, `effects_pending` rows with a future
+/// `next_attempt_at` would wait until the next process restart.
+/// Polls the indexed due query with bounded batches and failure
+/// isolation (one request's error never aborts the batch).
+fn spawn_due_request_worker(state: &AppState) -> tokio::task::JoinHandle<()> {
+    let app_state = state.clone();
+    let mut shutdown_rx = state.subscribe_shutdown();
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        loop {
+            tokio::select! {
+                _ = interval.tick() => {
+                    match durable_outbox::drain_receive_pack_requests(
+                        app_state.clone(),
+                        durable_outbox::DRAIN_PER_PASS_LIMIT,
+                    )
+                    .await
+                    {
+                        Ok((0, _)) => {}
+                        Ok((n, _)) => tracing::info!(n, "due request worker drained"),
+                        Err(e) => tracing::warn!(err = %e, "due request worker drain failed"),
+                    }
+                }
+                _ = shutdown_rx.changed() => break,
+            }
+        }
+    })
+}
+
+/// Dedicated marker-tombstone worker at a shorter cadence than the
+/// daily purge, so external Git refs do not linger a full period
+/// after their SQL owner is gone.
+fn spawn_marker_cleanup_worker(state: &AppState) -> tokio::task::JoinHandle<()> {
+    let app_state = state.clone();
+    let git_timeout =
+        std::time::Duration::from_secs(app_state.config.git_service_timeout_secs.min(30));
+    let mut shutdown_rx = state.subscribe_shutdown();
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        loop {
+            tokio::select! {
+                _ = interval.tick() => {
+                    if let Err(e) = durable_outbox::drain_marker_cleanup_queue(
+                        &app_state,
+                        durable_outbox::DRAIN_PER_PASS_LIMIT,
+                        git_timeout,
+                    )
+                    .await
+                    {
+                        tracing::warn!(err = %e, "marker cleanup worker failed");
+                    }
+                }
+                _ = shutdown_rx.changed() => break,
+            }
+        }
+    })
+}
+
 fn spawn_shutdown_signal(tx: watch::Sender<bool>) {
     tokio::spawn(async move {
         #[cfg(unix)]
@@ -1258,55 +1475,6 @@ mod rate_limiter_sweep_tests {
 }
 
 /// it.
-fn spawn_legacy_cid_sweep(state: &AppState, config: &Config) -> tokio::task::JoinHandle<()> {
-    let db = state.db.clone();
-    let repos_dir = config.repos_dir.clone();
-    let git_bin = state.git_bin.clone();
-    let git_timeout = std::time::Duration::from_secs(config.git_service_timeout_secs);
-    let batch = config.pin_repair_sweep_batch;
-    let delay = std::time::Duration::from_secs(config.pin_repair_sweep_delay_secs);
-    let mut shutdown_rx = state.subscribe_shutdown();
-    tokio::spawn(async move {
-        tokio::select! {
-            _ = ipfs_pin::run_sweep_rearmed(
-                &repos_dir, &git_bin, git_timeout, batch, delay,
-                ipfs_pin::SWEEP_REARM_DELAY, &db,
-            ) => {}
-            _ = shutdown_rx.changed() => {}
-        }
-    })
-}
-
-async fn gossip_ping_round(
-    db: &Db,
-    client: &reqwest::Client,
-    failed_once: &mut HashSet<String>,
-    peers: Vec<db::PeerRecord>,
-) {
-    {
-        let current_dids: HashSet<&str> = peers.iter().map(|peer| peer.did.as_str()).collect();
-        failed_once.retain(|did| current_dids.contains(did.as_str()));
-    }
-    for peer in peers {
-        let ok = ping_peer_readiness(client, &peer.http_url).await;
-        match peer_ping_db_update(failed_once, &peer.did, ok) {
-            Some(reachable) => {
-                if let Err(error) = db.mark_peer_ping(&peer.did, reachable).await {
-                    warn!(did = %peer.did, err = %error, "failed to persist peer readiness");
-                }
-            }
-            None => warn!(
-                did = %peer.did,
-                "peer readiness probe failed once; preserving previous federation status"
-            ),
-        }
-    }
-}
-
-/// Decide whether one readiness sample should update the persisted federation
-/// gate. A success is authoritative immediately. A failure must repeat on the
-/// next gossip tick before it can mark a peer unreachable, so one transient
-/// database hiccup does not hide the peer's repositories for five minutes.
 fn peer_ping_db_update(failed_once: &mut HashSet<String>, did: &str, ready: bool) -> Option<bool> {
     if ready {
         failed_once.remove(did);
@@ -1483,6 +1651,32 @@ fn load_or_create_keypair_at(key_path: &std::path::Path) -> Result<Keypair> {
     Ok(kp)
 }
 
+async fn gossip_ping_round(
+    db: &Db,
+    client: &reqwest::Client,
+    failed_once: &mut HashSet<String>,
+    peers: Vec<db::PeerRecord>,
+) {
+    {
+        let current_dids: HashSet<&str> = peers.iter().map(|peer| peer.did.as_str()).collect();
+        failed_once.retain(|did| current_dids.contains(did.as_str()));
+    }
+    for peer in peers {
+        let ok = ping_peer_readiness(client, &peer.http_url).await;
+        match peer_ping_db_update(failed_once, &peer.did, ok) {
+            Some(reachable) => {
+                if let Err(error) = db.mark_peer_ping(&peer.did, reachable).await {
+                    warn!(did = %peer.did, err = %error, "failed to persist peer readiness");
+                }
+            }
+            None => warn!(
+                did = %peer.did,
+                "peer readiness probe failed once; preserving previous federation status"
+            ),
+        }
+    }
+}
+
 #[cfg(test)]
 mod gossip_ssrf_tests {
     use super::{

crates/gitlawb-node/src/sync.rs

@@ -14,6 +14,7 @@
 
 use std::collections::HashMap;
 use std::path::Path;
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use gitlawb_core::identity::Keypair;
@@ -435,9 +436,10 @@ async fn process_batch(
                 .await;
             }
             Err(e) => {
-                warn!(repo = %item.repo, origin = %origin_url, err = %e, "repo sync failed");
-                let _ = db.mark_sync_failed(&item.id).await;
-                crate::metrics::record_sync_processed("failed");
+                // Prune vs terminal classification lives in
+                // `handle_sync_item_error` (defer = stays pending for the
+                // next tick; anything else = failed).
+                handle_sync_item_error(db, &item, &origin_url, e).await;
             }
         }
     }
@@ -691,6 +693,284 @@ async fn existing_promisor_state(repo: &str) -> PromisorProbe {
     }
 }
 
+/// Byte-prefix check for the two node-managed blob namespaces exempt from
+/// the fail-closed commit walk (`visibility_pack.rs`): request markers
+/// (`refs/gitlawb/requests/`) and issue records (`refs/gitlawb/issues/`).
+/// Bytewise so non-UTF-8 refnames (which git accepts) are classified, not
+/// lossy-decoded into a false exempt match. Extracted so the predicate is
+/// unit-testable; [`prune_non_exempt_gitlawb_refs`] enforces it on import.
+fn gitlawb_ref_is_exempt(raw_refname: &[u8]) -> bool {
+    raw_refname.starts_with(b"refs/gitlawb/requests/")
+        || raw_refname.starts_with(b"refs/gitlawb/issues/")
+}
+
+/// Parse one `git show-ref` output line into its refname bytes: `<hex-sha>`
+/// (40-hex SHA-1 or 64-hex SHA-256, which a sha256 mirror inherits from its
+/// origin), one space, then the refname. Empty lines (trailing newline) yield
+/// `Ok(None)`; anything else-shaped is an anomaly the caller fails on rather
+/// than skips — silently passing an unclassifiable line could leave an evil
+/// ref imported. Extracted so malformed input is unit-testable without git.
+fn show_ref_refname(line: &[u8]) -> anyhow::Result<Option<&[u8]>> {
+    if line.is_empty() {
+        return Ok(None);
+    }
+    let Some(sp) = line.iter().position(|b| *b == b' ') else {
+        return Err(anyhow::anyhow!("git show-ref emitted an unparsable line"));
+    };
+    let (sha, raw) = (&line[..sp], &line[sp + 1..]);
+    let sha_ok = (sha.len() == 40 || sha.len() == 64) && sha.iter().all(|b| b.is_ascii_hexdigit());
+    if !sha_ok || raw.is_empty() {
+        return Err(anyhow::anyhow!("git show-ref emitted an unparsable line"));
+    }
+    Ok(Some(raw))
+}
+
+/// Delete imported `refs/gitlawb/*` refs outside the two exempt subtrees.
+///
+/// Mirrors import whatever the origin advertises (`+refs/*:refs/*`), and
+/// `transfer.hideRefs` is a serve-side knob: a repo's own `git fetch` never
+/// consults local hideRefs, and origins hide only `requests/` anyway. So a
+/// planted `refs/gitlawb/<other>` blob ref — creatable on any pre-gate node —
+/// lands on every mirror and then trips `assert_all_refs_are_commits` closed,
+/// wedging serving on each one. Prune at import instead: after clone/fetch,
+/// list `refs/gitlawb` and delete every ref outside the exempt subtrees
+/// (negative refspecs cannot re-include them, so post-import deletion is the
+/// mechanism that works).
+///
+/// Callers map a transient prune error to [`PruneFailed`] so the batch
+/// defers the row (stays `pending`, retried next tick) instead of failing
+/// it: a failed row would leave a possibly-wedged mirror unretried until
+/// the next peer announcement. Deterministic refusals (identity mismatch,
+/// malformed enumeration) map to [`PruneInvalid`], which fails terminally
+/// since no retry can clear them.
+///
+/// Enumeration is `git show-ref` (no pattern), parsed as raw bytes: refnames
+/// cannot contain newlines, so one line per ref is lossless even for
+/// non-UTF-8 names, which a lossy decode would miss. The `<sha> SP <refname>`
+/// structure (not `for-each-ref --format=%(refname)`, which carries no object
+/// id) is what lets the parser width-check 40-hex SHA-1 vs 64-hex SHA-256
+/// and reject anything else-shaped instead of skipping it.
+///
+/// Fail-closed throughout: the exit code is classified (only `1` with empty
+/// output means "no refs"; anything else non-zero, e.g. 128 on a missing
+/// repo, is fatal, since a silent skip would leave an imported evil ref in
+/// place while the sync reports success), and any line that is not
+/// `<hex-sha> SP <refname>` — 40-hex SHA-1 or 64-hex SHA-256, which a sha256
+/// mirror inherits from its origin — fails the sync rather than being
+/// skipped into an unpruned wedge.
+/// Bounded git spawn for the prune below: `kill_on_drop` plus a hard timeout
+/// so a wedged child cannot stall a mirror import or sweep forever. The
+/// refname delete runs on exact bytes, so args arrive as `OsStr`, not `str`.
+/// Non-PruneInvalid failures (including a timeout) wrap as PruneFailed and
+/// defer for retry, matching the transient classification below.
+async fn prune_git(
+    local_path: &Path,
+    args: &[&std::ffi::OsStr],
+    label: &str,
+) -> anyhow::Result<std::process::Output> {
+    let fut = async {
+        // allow-unbounded-git: bounded by the tokio::time::timeout wrapping
+        // this future below, plus kill_on_drop.
+        let mut cmd = tokio::process::Command::new("git");
+        cmd.kill_on_drop(true);
+        cmd.arg("-C").arg(local_path).args(args);
+        cmd.output().await
+    };
+    tokio::time::timeout(std::time::Duration::from_secs(30), fut)
+        .await
+        .map_err(|_| anyhow::anyhow!("{label} timed out"))?
+        .map_err(|e| anyhow::anyhow!("{label} failed to spawn: {e}"))
+}
+
+async fn prune_non_exempt_gitlawb_refs(local_path: &Path) -> anyhow::Result<()> {
+    let local_str = local_path.to_str().unwrap_or(".");
+    // Prove the path is itself a repo before trusting the enumeration:
+    // `git -C` walks up to the nearest ancestor repo, so inside a non-repo
+    // child (a partially-created mirror) `show-ref` would exit 0 with the
+    // ANCESTOR's refs and `update-ref -d` would then run against that repo.
+    // Require `rev-parse --absolute-git-dir` to resolve to the path itself
+    // (mirrors are bare repos, where the git dir IS the path).
+    let git_dir_out = prune_git(
+        local_path,
+        &[
+            std::ffi::OsStr::new("rev-parse"),
+            std::ffi::OsStr::new("--absolute-git-dir"),
+        ],
+        "git rev-parse",
+    )
+    .await?;
+    if !git_dir_out.status.success() {
+        let stderr = String::from_utf8_lossy(&git_dir_out.stderr);
+        return Err(anyhow::anyhow!(PruneInvalid(format!(
+            "mirror path is not a git repo (rev-parse failed): {stderr}"
+        ))));
+    }
+    let actual_dir = PathBuf::from(
+        String::from_utf8_lossy(&git_dir_out.stdout)
+            .trim()
+            .to_string(),
+    );
+    let canonical_path = local_path
+        .canonicalize()
+        .map_err(|e| anyhow::anyhow!(PruneInvalid(format!("mirror path does not resolve: {e}"))))?;
+    let canonical_git_dir = actual_dir
+        .canonicalize()
+        .map_err(|e| anyhow::anyhow!(PruneInvalid(format!("git dir does not resolve: {e}"))))?;
+    if canonical_git_dir != canonical_path {
+        return Err(anyhow::anyhow!(PruneInvalid(format!(
+            "mirror path {local_str} is not itself a repo (git dir resolves elsewhere); refusing to enumerate"
+        ))));
+    }
+    let out = prune_git(
+        local_path,
+        &[std::ffi::OsStr::new("show-ref")],
+        "git show-ref",
+    )
+    .await?;
+    // Classify on the exit code (the `existing_promisor_state` precedent):
+    // 0 parses, 1 with empty output is an empty repo, anything else is a
+    // fatal enumeration failure that must not read as "nothing to prune".
+    match out.status.code() {
+        Some(0) => {}
+        Some(1) if out.stdout.is_empty() => return Ok(()),
+        _ => {
+            let stderr = String::from_utf8_lossy(&out.stderr);
+            return Err(anyhow::anyhow!(
+                "git show-ref failed with code {:?}: {stderr}",
+                out.status.code()
+            ));
+        }
+    }
+    for line in out.stdout.split(|b| *b == b'\n') {
+        // Malformed lines are deterministic (same git, same output every
+        // tick): fail terminally via PruneInvalid, not transiently.
+        let Some(raw) =
+            show_ref_refname(line).map_err(|e| anyhow::anyhow!(PruneInvalid(e.to_string())))?
+        else {
+            continue;
+        };
+        if !(raw == b"refs/gitlawb" || raw.starts_with(b"refs/gitlawb/"))
+            || gitlawb_ref_is_exempt(raw)
+        {
+            continue;
+        }
+        // Delete by exact bytes (not the lossy display string below): a
+        // non-UTF-8 refname would otherwise survive the prune.
+        let refname_os = os_from_raw_refname(raw);
+        let shown = refname_os.to_string_lossy().into_owned();
+        let delete = prune_git(
+            local_path,
+            &[
+                std::ffi::OsStr::new("update-ref"),
+                std::ffi::OsStr::new("-d"),
+                refname_os.as_os_str(),
+                std::ffi::OsStr::new("--no-deref"),
+            ],
+            "git update-ref -d",
+        )
+        .await?;
+        if !delete.status.success() {
+            let stderr = String::from_utf8_lossy(&delete.stderr);
+            return Err(anyhow::anyhow!(
+                "git update-ref -d {shown} failed (mirror still carries a non-exempt refs/gitlawb ref): {stderr}"
+            ));
+        }
+        tracing::info!(
+            gitlawb_ref = %shown,
+            "mirror: pruned non-exempt refs/gitlawb ref on import"
+        );
+    }
+    Ok(())
+}
+
+/// Lossless bytes → OsString for `update-ref -d`: non-UTF-8 refnames must be
+/// deleted by exact bytes. Unix is the serving platform; elsewhere fall back
+/// to a lossy decode (best effort where exact bytes are unrepresentable).
+#[cfg(unix)]
+fn os_from_raw_refname(raw: &[u8]) -> std::ffi::OsString {
+    use std::os::unix::ffi::OsStringExt;
+    std::ffi::OsString::from_vec(raw.to_vec())
+}
+
+#[cfg(not(unix))]
+fn os_from_raw_refname(raw: &[u8]) -> std::ffi::OsString {
+    String::from_utf8_lossy(raw).into_owned().into()
+}
+
+/// Marker for mirror-prune failures: the clone/fetch succeeded, so the
+/// mirror is on disk, but a non-exempt `refs/gitlawb/*` ref may still be
+/// imported. Local-git conditions are transient (a retry can succeed), while
+/// `mark_sync_failed` is terminal until the next peer announcement — and a
+/// failed row would leave a possibly-wedged mirror unretried in the
+/// meantime. The batch error arm downcasts to this and defers the row
+/// (stays `pending`, retried on the next 30s tick, where fetch+prune are
+/// idempotent) instead of failing it.
+#[derive(Debug)]
+struct PruneFailed(String);
+
+impl std::fmt::Display for PruneFailed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "mirror prune failed: {}", self.0)
+    }
+}
+
+impl std::error::Error for PruneFailed {}
+
+/// Marker for deterministic mirror-prune refusals: the identity check (path
+/// is not itself a repo) and malformed `show-ref` output cannot clear on
+/// retry — re-running fetch+prune every tick would spin forever on a
+/// permanently bad path. Unlike [`PruneFailed`], these fail the row
+/// terminally so the condition surfaces instead of looping silently.
+#[derive(Debug)]
+struct PruneInvalid(String);
+
+/// Preserve a deterministic prune refusal through the clone/fetch call
+/// sites: `prune_non_exempt_gitlawb_refs` already returns `PruneInvalid`
+/// for identity mismatches and malformed enumeration, and re-wrapping
+/// every error as `PruneFailed` would stringify that type away before
+/// `handle_sync_item_error` downcasts it — leaving a permanently invalid
+/// mirror deferred (fetch+prune every tick) instead of failed. Transient
+/// failures keep the `PruneFailed` deferral.
+fn wrap_prune_error(e: anyhow::Error) -> anyhow::Error {
+    match e.downcast::<PruneInvalid>() {
+        Ok(invalid) => anyhow::Error::new(invalid),
+        Err(other) => anyhow::anyhow!(PruneFailed(other.to_string())),
+    }
+}
+
+impl std::fmt::Display for PruneInvalid {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "mirror prune refused: {}", self.0)
+    }
+}
+
+impl std::error::Error for PruneInvalid {}
+
+/// Classify a mirror sync failure for one queue row. Transient prune
+/// failures ([`PruneFailed`]) defer: the mirror is on disk but may still
+/// carry a non-exempt refs/gitlawb ref, a transient local-git condition a
+/// retry can clear — failing the row would leave it unretried until the
+/// next peer announcement, wedging serving in the meantime. Deterministic
+/// refusals ([`PruneInvalid`]) and everything else fail the row terminally:
+/// retrying a permanently bad path every tick would spin forever instead of
+/// surfacing. Extracted so the deferral (not just the prune itself) is
+/// unit-testable.
+async fn handle_sync_item_error(
+    db: &Db,
+    item: &crate::db::SyncQueueItem,
+    origin_url: &str,
+    err: anyhow::Error,
+) {
+    if err.downcast_ref::<PruneFailed>().is_some() {
+        warn!(repo = %item.repo, origin = %origin_url, err = %err, "mirror prune failed; leaving sync row pending for retry");
+        crate::metrics::record_sync_processed("deferred");
+        return;
+    }
+    warn!(repo = %item.repo, origin = %origin_url, err = %err, "repo sync failed");
+    let _ = db.mark_sync_failed(&item.id).await;
+    crate::metrics::record_sync_processed("failed");
+}
+
 /// Mirror-clone a repo from a remote URL into a local bare repo.
 /// `Promisor` mode adds `--filter=blob:limit=10g`, which marks the repo a git
 /// promisor (so a pack with origin-omitted withheld blobs is accepted) while
@@ -715,6 +995,13 @@ async fn clone_repo(remote_url: &str, local_path: &Path, mode: MirrorMode) -> an
         let stderr = String::from_utf8_lossy(&out.stderr);
         return Err(anyhow::anyhow!("git clone --mirror failed: {stderr}"));
     }
+    // A fresh mirror imports everything the origin advertises, including any
+    // planted non-exempt `refs/gitlawb/*` ref; prune before serving from it.
+    // Prune outcome flows through `wrap_prune_error`: transient failures
+    // defer (stays pending), deterministic refusals fail terminally.
+    prune_non_exempt_gitlawb_refs(local_path)
+        .await
+        .map_err(wrap_prune_error)?;
     Ok(())
 }
 
@@ -735,7 +1022,11 @@ async fn fetch_repo(local_path: &Path, remote_url: &str, mode: MirrorMode) -> an
     ])
     .await?;
 
-    match mode {
+    // The fetch imports whatever the origin advertises via the stored
+    // `+refs/*:refs/*` refspec; prune non-exempt `refs/gitlawb/*` refs after
+    // every fetch so one planted origin ref cannot wedge serving on the
+    // mirror (see `prune_non_exempt_gitlawb_refs`).
+    let fetch_result = match mode {
         MirrorMode::Promisor => {
             git_run(&["-C", local_str, "config", "remote.origin.promisor", "true"]).await?;
             git_run(&[
@@ -775,7 +1066,13 @@ async fn fetch_repo(local_path: &Path, remote_url: &str, mode: MirrorMode) -> an
                 git_run(&["-C", local_str, "fetch", "--prune", "origin"]).await
             }
         }
-    }
+    };
+    fetch_result?;
+    // Prune after every fetch (see above); outcome flows through
+    // `wrap_prune_error` like the clone path.
+    prune_non_exempt_gitlawb_refs(local_path)
+        .await
+        .map_err(wrap_prune_error)
 }
 
 #[cfg(test)]
@@ -2108,4 +2405,463 @@ mod tests {
         assert!(object_count(&mirror) > 0, "mirror has no objects");
         assert_eq!(sync_status(&pool, "z6Mkfoo/hello").await, "done");
     }
+
+    /// The import-exemption predicate is bytewise over both node-managed
+    /// namespaces and nothing else. Non-UTF-8 bytes never match (git accepts
+    /// them, so a lossy decode could false-exempt). Narrowing either prefix
+    /// turns these red.
+    #[test]
+    fn gitlawb_import_exemption_covers_both_namespaces_by_bytes() {
+        for exempt in [
+            &b"refs/gitlawb/requests/550e8400-e29b-41d4-a716-446655440000"[..],
+            &b"refs/gitlawb/issues/abc123"[..],
+        ] {
+            assert!(
+                gitlawb_ref_is_exempt(exempt),
+                "import must keep {}",
+                String::from_utf8_lossy(exempt)
+            );
+        }
+        for pruned in [
+            &b"refs/gitlawb/evil"[..],
+            &b"refs/gitlawb"[..],
+            &b"refs/gitlawbfoo"[..],
+            &b"refs/heads/main"[..],
+            &b""[..],
+            // Raw 0xFF byte: never an exempt prefix, always pruned.
+            &b"refs/gitlawb/\xffvil"[..],
+        ] {
+            assert!(
+                !gitlawb_ref_is_exempt(pruned),
+                "import must prune {}",
+                String::from_utf8_lossy(pruned)
+            );
+        }
+    }
+
+    /// End-to-end prune on a scratch bare repo: a planted non-exempt
+    /// `refs/gitlawb/*` blob ref is deleted, exempt `requests/`/`issues/`
+    /// blob refs and content refs survive. Removing the prune call from
+    /// `clone_repo`/`fetch_repo` leaves the evil ref in place, which is what
+    /// wedges serving on mirrors — this test plants exactly that state and
+    /// proves the prune removes it.
+    #[test]
+    fn prune_removes_non_exempt_gitlawb_refs_and_keeps_exempt() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        rt.block_on(async {
+            let td = TempDir::new().unwrap();
+            let work = td.path().join("work");
+            let bare = td.path().join("m.git");
+            std::fs::create_dir_all(&work).unwrap();
+            // A real commit for the content branch (branch refs must point
+            // at commits; the evil/exempt refs below point at blobs).
+            std::fs::write(work.join("f.txt"), b"x\n").unwrap();
+            let run_work = |args: &[&str]| {
+                assert!(
+                    Command::new("git")
+                        .args(args)
+                        .current_dir(&work)
+                        .status()
+                        .unwrap()
+                        .success(),
+                    "git {args:?} failed"
+                );
+            };
+            run_work(&["init", "-q"]);
+            run_work(&["config", "user.email", "t@t"]);
+            run_work(&["config", "user.name", "t"]);
+            run_work(&["add", "f.txt"]);
+            run_work(&["commit", "-qm", "init"]);
+            assert!(
+                Command::new("git")
+                    .args([
+                        "clone",
+                        "-q",
+                        "--bare",
+                        work.to_str().unwrap(),
+                        bare.to_str().unwrap()
+                    ])
+                    .current_dir(td.path())
+                    .status()
+                    .unwrap()
+                    .success(),
+                "git clone --bare failed"
+            );
+            let run = |args: &[&str]| {
+                assert!(
+                    Command::new("git")
+                        .args(args)
+                        .current_dir(&bare)
+                        .status()
+                        .unwrap()
+                        .success(),
+                    "git {args:?} failed"
+                );
+            };
+            let run_out = |args: &[&str]| {
+                let out = Command::new("git")
+                    .args(args)
+                    .current_dir(&bare)
+                    .output()
+                    .unwrap();
+                assert!(out.status.success(), "git {args:?} failed");
+                out.stdout
+            };
+            run(&["init", "-q", "--bare", "."]);
+            std::fs::write(bare.join("blob-body"), b"marker\n").unwrap();
+            let blob = String::from_utf8_lossy(&run_out(&["hash-object", "-w", "blob-body"]))
+                .trim()
+                .to_string();
+            // The clone carries the content branch; plant the evil ref and
+            // both exempt refs (all blob-pointing, like node metadata).
+            run(&["update-ref", "refs/gitlawb/evil", &blob]);
+            run(&["update-ref", "refs/gitlawb/requests/r1", &blob]);
+            run(&["update-ref", "refs/gitlawb/issues/i1", &blob]);
+
+            prune_non_exempt_gitlawb_refs(&bare)
+                .await
+                .expect("prune succeeds");
+
+            let raw_refs = run_out(&["for-each-ref", "--format=%(refname)"]);
+            let refs = String::from_utf8_lossy(&raw_refs);
+            let set: std::collections::HashSet<&str> = refs
+                .lines()
+                .map(str::trim)
+                .filter(|l| !l.is_empty())
+                .collect();
+            assert!(
+                !set.contains("refs/gitlawb/evil"),
+                "non-exempt blob ref must be pruned, still present: {set:?}"
+            );
+            for keep in ["refs/gitlawb/requests/r1", "refs/gitlawb/issues/i1"] {
+                assert!(set.contains(keep), "{keep} must survive the prune: {set:?}");
+            }
+            assert!(
+                set.iter().any(|r| r.starts_with("refs/heads/")),
+                "content branch ref must survive the prune: {set:?}"
+            );
+        });
+    }
+
+    /// `show-ref` line parser: 40-hex SHA-1 and 64-hex SHA-256 lines yield
+    /// the refname, empty lines yield None, and anything else-shaped errors
+    /// (the caller fails the sync rather than skipping into an unpruned
+    /// wedge). Reverting to a 40-only width check turns the sha256 case red.
+    #[test]
+    fn show_ref_line_parser_accepts_both_hash_widths_and_rejects_junk() {
+        // 40-hex SHA-1 line.
+        let sha1_line = format!("{} refs/gitlawb/evil", "a".repeat(40));
+        assert_eq!(
+            show_ref_refname(sha1_line.as_bytes()).unwrap(),
+            Some(b"refs/gitlawb/evil".as_slice())
+        );
+        // 64-hex SHA-256 line (a sha256 mirror inherits the width).
+        let sha256_line = format!("{} refs/gitlawb/evil", "b".repeat(64));
+        assert_eq!(
+            show_ref_refname(sha256_line.as_bytes()).unwrap(),
+            Some(b"refs/gitlawb/evil".as_slice())
+        );
+        // Non-UTF-8 refname passes through as exact bytes.
+        let mut raw = vec![b'c'; 40];
+        raw.extend_from_slice(b" refs/gitlawb/\xffvil");
+        assert_eq!(
+            show_ref_refname(&raw).unwrap(),
+            Some(b"refs/gitlawb/\xffvil".as_slice())
+        );
+        // Trailing newline residue.
+        assert_eq!(show_ref_refname(b"").unwrap(), None);
+        // No space, short sha, non-hex sha, empty refname: all errors.
+        let bads: Vec<Vec<u8>> = vec![
+            b"nospaceshere".to_vec(),
+            format!("{}refs/x", "a".repeat(40)).into_bytes(),
+            [b'z'; 40]
+                .iter()
+                .chain(b" refs/x".iter())
+                .copied()
+                .collect(),
+            format!("{} ", "a".repeat(40)).into_bytes(),
+            format!("{} refs/x", "a".repeat(41)).into_bytes(),
+        ];
+        for bad in &bads {
+            assert!(
+                show_ref_refname(bad).is_err(),
+                "malformed show-ref line must error, not skip"
+            );
+        }
+    }
+
+    /// Pruning a path that is not a repo errors. Two shapes: a missing path
+    /// fails `rev-parse` outright (exit 128), while a plain subdirectory of
+    /// a repo would otherwise enumerate the ancestor's refs — covered by the
+    /// identity test below. Either way the prune refuses instead of pruning
+    /// the wrong repo or silently skipping an evil ref.
+    #[test]
+    fn prune_on_non_repo_path_errors() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        rt.block_on(async {
+            let err =
+                prune_non_exempt_gitlawb_refs(Path::new("/tmp/gitlawb-definitely-not-a-repo-xyz"))
+                    .await
+                    .expect_err("non-repo path must fail the prune, not skip it");
+            assert!(
+                err.to_string().contains("not a git repo")
+                    || err.to_string().contains("does not resolve"),
+                "unexpected error shape: {err:#}"
+            );
+        });
+    }
+
+    /// Identity check: pruning a plain subdirectory of a repo refuses instead
+    /// of enumerating the ancestor. `git -C` on the child exits 0 from
+    /// `show-ref` with the ANCESTOR's refs, so without the
+    /// `rev-parse --absolute-git-dir` identity check the delete loop would
+    /// enumerate and delete the wrong repository's refs. Deleting the check
+    /// makes the prune succeed and removes the ancestor's evil ref — both
+    /// assertions go red.
+    #[test]
+    fn prune_on_non_repo_child_of_repo_refuses_and_touches_nothing() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        rt.block_on(async {
+            let td = TempDir::new().unwrap();
+            let ancestor = td.path().join("ancestor.git");
+            std::fs::create_dir_all(&ancestor).unwrap();
+            let run_ancestor = |args: &[&str]| {
+                assert!(
+                    Command::new("git")
+                        .args(args)
+                        .current_dir(&ancestor)
+                        .status()
+                        .unwrap()
+                        .success(),
+                    "git {args:?} failed"
+                );
+            };
+            run_ancestor(&["init", "-q", "--bare", "."]);
+            // Blob target for the planted ref (mirrors the node-metadata shape).
+            std::fs::write(ancestor.join("blob-body"), b"marker\n").unwrap();
+            let out = Command::new("git")
+                .args(["hash-object", "-w", "blob-body"])
+                .current_dir(&ancestor)
+                .output()
+                .unwrap();
+            assert!(out.status.success(), "git hash-object failed");
+            let blob = String::from_utf8_lossy(&out.stdout).trim().to_string();
+            run_ancestor(&["update-ref", "refs/gitlawb/evil", &blob]);
+
+            // A plain child directory: not itself a repo.
+            let child = ancestor.join("not-a-repo");
+            std::fs::create_dir_all(&child).unwrap();
+
+            let err = prune_non_exempt_gitlawb_refs(&child)
+                .await
+                .expect_err("pruning a non-repo child must refuse, not enumerate the ancestor");
+            assert!(
+                err.to_string().contains("not itself a repo"),
+                "refusal must name the identity mismatch, got: {err:#}"
+            );
+
+            // The ancestor's evil ref is untouched.
+            let refs_out = Command::new("git")
+                .args(["for-each-ref", "--format=%(refname)"])
+                .current_dir(&ancestor)
+                .output()
+                .unwrap();
+            let refs = String::from_utf8_lossy(&refs_out.stdout);
+            assert!(
+                refs.lines().any(|l| l.trim() == "refs/gitlawb/evil"),
+                "ancestor evil ref must survive the refused prune"
+            );
+        });
+    }
+
+    /// sha256 mirror: a 64-hex `show-ref` line parses and the planted ref is
+    /// pruned. A 40-only parser skips every line and leaves the evil ref.
+    #[test]
+    fn prune_parses_sha256_show_ref_and_prunes() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        rt.block_on(async {
+            let td = TempDir::new().unwrap();
+            let bare = td.path().join("m256.git");
+            std::fs::create_dir_all(&bare).unwrap();
+            let run = |args: &[&str]| {
+                assert!(
+                    Command::new("git")
+                        .args(args)
+                        .current_dir(&bare)
+                        .status()
+                        .unwrap()
+                        .success(),
+                    "git {args:?} failed"
+                );
+            };
+            let run_out = |args: &[&str]| {
+                let out = Command::new("git")
+                    .args(args)
+                    .current_dir(&bare)
+                    .output()
+                    .unwrap();
+                assert!(out.status.success(), "git {args:?} failed");
+                out.stdout
+            };
+            run(&["init", "-q", "--bare", "--object-format=sha256", "."]);
+            std::fs::write(bare.join("blob-body"), b"marker\n").unwrap();
+            let blob = String::from_utf8_lossy(&run_out(&["hash-object", "-w", "blob-body"]))
+                .trim()
+                .to_string();
+            assert_eq!(
+                blob.len(),
+                64,
+                "sanity: sha256 repo must mint 64-hex objects"
+            );
+            run(&["update-ref", "refs/gitlawb/evil", &blob]);
+            run(&["update-ref", "refs/gitlawb/issues/i1", &blob]);
+
+            prune_non_exempt_gitlawb_refs(&bare)
+                .await
+                .expect("prune succeeds on a sha256 repo");
+
+            let raw_refs = run_out(&["for-each-ref", "--format=%(refname)"]);
+            let refs = String::from_utf8_lossy(&raw_refs);
+            let set: std::collections::HashSet<&str> = refs
+                .lines()
+                .map(str::trim)
+                .filter(|l| !l.is_empty())
+                .collect();
+            assert!(
+                !set.contains("refs/gitlawb/evil"),
+                "evil ref must be pruned from sha256 mirror: {set:?}"
+            );
+            assert!(
+                set.contains("refs/gitlawb/issues/i1"),
+                "exempt ref must survive: {set:?}"
+            );
+        });
+    }
+
+    /// Prune failures defer the queue row (stays `pending` for the next
+    /// tick); deterministic prune refusals and any other sync error fail it
+    /// terminally. Deleting the `downcast_ref::<PruneFailed>` arm restores
+    /// terminal failure and turns the first assertion red.
+    /// Prune failures defer the queue row (stays `pending` for the next
+    /// tick); deterministic prune refusals and any other sync error fail it
+    /// terminally. Deleting the `downcast_ref::<PruneFailed>` arm restores
+    /// terminal failure and turns the first assertion red.
+    ///
+    /// The wrapper below is what the clone/fetch call sites run through:
+    /// a `PruneInvalid` from the helper must still downcast as
+    /// `PruneInvalid` afterwards (not stringified into `PruneFailed`),
+    /// and anything else must arrive as `PruneFailed`.
+    #[test]
+    fn wrap_prune_error_preserves_deterministic_type() {
+        let invalid = wrap_prune_error(anyhow::anyhow!(PruneInvalid("x".to_string())));
+        assert!(
+            invalid.downcast_ref::<PruneInvalid>().is_some(),
+            "deterministic refusal must survive the call-site wrap"
+        );
+        assert!(
+            invalid.downcast_ref::<PruneFailed>().is_none(),
+            "deterministic refusal must not also read as transient"
+        );
+        let transient = wrap_prune_error(anyhow::anyhow!("boom"));
+        assert!(
+            transient.downcast_ref::<PruneFailed>().is_some(),
+            "generic failure must arrive as transient"
+        );
+        assert!(
+            transient.downcast_ref::<PruneInvalid>().is_none(),
+            "generic failure must not read as deterministic"
+        );
+    }
+
+    #[sqlx::test]
+    async fn prune_failure_defers_row_other_failures_fail_it(pool: PgPool) {
+        let state = crate::test_support::test_state(pool).await;
+        let item = |id: &str| crate::db::SyncQueueItem {
+            id: id.to_string(),
+            repo: "z6Mkfoo/hello".to_string(),
+            node_did: "did:key:z6MkOrigin".to_string(),
+            ref_name: "refs/heads/main".to_string(),
+            new_sha: "0".repeat(40),
+            cid: None,
+            status: "pending".to_string(),
+            enqueued_at: chrono::Utc::now().to_rfc3339(),
+        };
+        async fn status_of(pool: &sqlx::PgPool, id: &str) -> String {
+            let row: (String,) = sqlx::query_as("SELECT status FROM sync_queue WHERE id = $1")
+                .bind(id)
+                .fetch_one(pool)
+                .await
+                .unwrap();
+            row.0
+        }
+        // Seed three pending rows directly (enqueue_sync mints random ids).
+        for id in ["sync-prune-defer", "sync-prune-invalid", "sync-other-fail"] {
+            sqlx::query(
+                "INSERT INTO sync_queue (id, repo, node_did, ref_name, new_sha, cid, status, enqueued_at)
+                 VALUES ($1, $2, $3, $4, $5, NULL, 'pending', $6)",
+            )
+            .bind(id)
+            .bind("z6Mkfoo/hello")
+            .bind("did:key:z6MkOrigin")
+            .bind("refs/heads/main")
+            .bind("0".repeat(40))
+            .bind(chrono::Utc::now().to_rfc3339())
+            .execute(state.db.pool())
+            .await
+            .unwrap();
+        }
+
+        handle_sync_item_error(
+            &state.db,
+            &item("sync-prune-defer"),
+            "http://origin.example",
+            anyhow::anyhow!(PruneFailed("boom".to_string())),
+        )
+        .await;
+        assert_eq!(
+            status_of(state.db.pool(), "sync-prune-defer").await,
+            "pending",
+            "prune failure must leave the row pending for retry"
+        );
+
+        handle_sync_item_error(
+            &state.db,
+            &item("sync-other-fail"),
+            "http://origin.example",
+            anyhow::anyhow!("boom"),
+        )
+        .await;
+        assert_eq!(
+            status_of(state.db.pool(), "sync-other-fail").await,
+            "failed",
+            "non-prune failure must fail the row"
+        );
+
+        // Deterministic refusal: no retry can clear it, so it fails
+        // terminally instead of spinning every tick forever.
+        handle_sync_item_error(
+            &state.db,
+            &item("sync-prune-invalid"),
+            "http://origin.example",
+            anyhow::anyhow!(PruneInvalid("not itself a repo".to_string())),
+        )
+        .await;
+        assert_eq!(
+            status_of(state.db.pool(), "sync-prune-invalid").await,
+            "failed",
+            "deterministic prune refusal must fail the row, not defer forever"
+        );
+    }
 }

crates/gitlawb-node/src/webhooks.rs

@@ -38,26 +38,151 @@ pub fn fire_event(
     event: &str,
     payload: serde_json::Value,
 ) {
-    let repo_id = repo_id.to_string();
-    let event = event.to_string();
-    tokio::spawn(async move {
-        fire_event_async(db, http_client, &repo_id, &event, payload).await;
-    });
+    fire_event_occurrence(db, http_client, repo_id, event, payload, None, None);
 }
 
-async fn fire_event_async(
+/// Occurrence-keyed variant: at most one delivery per
+/// `(request_id, ref_name, hook)` is spawned; concurrent executors
+/// collapse via the `webhook_deliveries` ledger. Best-effort on crash
+/// (ledger is claimed before spawn), which the PR scope allows.
+///
+/// For durable outbox usage, call `claim_webhook_delivery_before_spawn` first
+/// to claim deliveries synchronously, then call
+/// `fire_event_occurrence_with_claimed_hooks` with the claimed hook IDs to
+/// spawn HTTP POST only for claimed hooks.
+pub fn fire_event_occurrence(
     db: Arc<Db>,
     http_client: Arc<reqwest::Client>,
     repo_id: &str,
     event: &str,
     payload: serde_json::Value,
+    request_id: Option<&str>,
+    ref_name: Option<&str>,
 ) {
-    let hooks = match db.list_webhooks_for_event(repo_id, event).await {
-        Ok(h) => h,
-        Err(e) => {
-            tracing::warn!(err = %e, "failed to list webhooks for event {event}");
-            return;
+    let repo_id = repo_id.to_string();
+    let event = event.to_string();
+    let request_id = request_id.map(|s| s.to_string());
+    let ref_name = ref_name.map(|s| s.to_string());
+    tokio::spawn(async move {
+        fire_event_async_occurrence(
+            db,
+            http_client,
+            &repo_id,
+            &event,
+            payload,
+            request_id.as_deref(),
+            ref_name.as_deref(),
+            None,
+        )
+        .await;
+    });
+}
+
+/// Variant for durable outbox: only spawn HTTP POST for previously claimed hooks.
+#[allow(clippy::too_many_arguments)]
+pub fn fire_event_occurrence_with_claimed_hooks(
+    db: Arc<Db>,
+    http_client: Arc<reqwest::Client>,
+    repo_id: &str,
+    event: &str,
+    payload: serde_json::Value,
+    request_id: Option<&str>,
+    ref_name: Option<&str>,
+    claimed_hooks: Vec<crate::db::Webhook>,
+) {
+    let repo_id = repo_id.to_string();
+    let event = event.to_string();
+    let request_id = request_id.map(|s| s.to_string());
+    let ref_name = ref_name.map(|s| s.to_string());
+    tokio::spawn(async move {
+        fire_event_async_occurrence(
+            db,
+            http_client,
+            &repo_id,
+            &event,
+            payload,
+            request_id.as_deref(),
+            ref_name.as_deref(),
+            Some(claimed_hooks),
+        )
+        .await;
+    });
+}
+
+/// Claim webhook delivery before spawning HTTP POST.
+/// Returns `Some(hook ids)` successfully claimed, or `None` when the hook
+/// list itself could not be read (transient DB error, unknown membership).
+/// A `None` caller must fall back to best-effort send rather than dropping
+/// the delivery: children are still deleted downstream, so an empty claim
+/// set on DB error would lose the webhook permanently.
+/// This synchronous claim phase must be awaited before child deletion to prevent
+/// permanent webhook loss on crash between deletion and claim.
+pub async fn claim_webhook_delivery_before_spawn(
+    db: Arc<Db>,
+    repo_id: &str,
+    event: &str,
+    request_id: Option<&str>,
+    ref_name: Option<&str>,
+) -> Option<Vec<crate::db::Webhook>> {
+    let mut claimed_hooks = Vec::new();
+
+    // Only claim when we have occurrence context (request_id + ref_name)
+    if let (Some(req), Some(r)) = (request_id, ref_name) {
+        let hooks = match db.list_webhooks_for_event(repo_id, event).await {
+            Ok(h) => h,
+            Err(e) => {
+                tracing::warn!(err = %e, "failed to list webhooks for event {event}");
+                return None;
+            }
+        };
+
+        for hook in hooks {
+            let delivery_id = crate::db::deterministic_id(&["webhook", req, r, &hook.id]);
+            match db
+                .claim_webhook_delivery(&delivery_id, req, repo_id, event)
+                .await
+            {
+                Ok(true) => claimed_hooks.push(hook),
+                Ok(false) => continue, // Already claimed
+                // Send fallback (legacy best-effort): a transient claim
+                // error must not drop the delivery while children are
+                // still deleted. The in-task claimer this replaced sent
+                // anyway on claim error; keep that here.
+                Err(e) => {
+                    tracing::warn!(err = %e, hook_id = %hook.id, "webhook claim failed; sending anyway");
+                    claimed_hooks.push(hook);
+                }
+            }
         }
+    }
+
+    Some(claimed_hooks)
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn fire_event_async_occurrence(
+    db: Arc<Db>,
+    http_client: Arc<reqwest::Client>,
+    repo_id: &str,
+    event: &str,
+    payload: serde_json::Value,
+    request_id: Option<&str>,
+    ref_name: Option<&str>,
+    claimed_hooks: Option<Vec<crate::db::Webhook>>,
+) {
+    // A caller that already claimed deliveries hands the claimed hook rows
+    // through; the send path must not re-list, because a list failure here
+    // would silently drop deliveries already claimed in the ledger.
+    let already_claimed = claimed_hooks.is_some();
+    let hooks = match claimed_hooks {
+        Some(h) => h,
+        None => match db.list_webhooks_for_event(repo_id, event).await {
+            Ok(h) => h,
+            Err(e) => {
+                tracing::warn!(err = %e, "failed to list webhooks for event {event}");
+                return;
+            }
+        },
     };
 
     if hooks.is_empty() {
@@ -76,8 +201,29 @@ async fn fire_event_async(
         let client = Arc::clone(&http_client);
         let event_name = event.to_string();
         let bytes = payload_bytes.clone();
-        let delivery_id = uuid::Uuid::new_v4().to_string();
+        // Stable occurrence delivery key when request context exists;
+        // random UUID otherwise (legacy callers without occurrence).
+        let delivery_id = match (request_id, ref_name) {
+            (Some(req), Some(r)) => crate::db::deterministic_id(&["webhook", req, r, &hook.id]),
+            _ => uuid::Uuid::new_v4().to_string(),
+        };
+        // Claim before spawn so concurrent executors collapse to one
+        // delivery per occurrence. Failures to record fall back to
+        // sending (legacy best-effort) rather than dropping.
+        // Skip claim if already done by caller (durable_outbox path).
+        if !already_claimed && request_id.is_some() {
+            match db
+                .clone()
+                .claim_webhook_delivery(&delivery_id, request_id.unwrap_or(""), repo_id, event)
+                .await
+            {
+                Ok(true) => {}
+                Ok(false) => continue,
+                Err(_) => {}
+            }
+        }
 
+        let db_for_mark = db.clone();
         tokio::spawn(async move {
             // Policy skip, not a delivery failure: the stored URL predates the
             // create-time gate. The "invalid_url" label is documented on
@@ -92,6 +238,7 @@ async fn fire_event_async(
                 return;
             }
 
+            let delivery_for_mark = delivery_id.clone();
             let signature = hook.secret.as_deref().map(|s| sign_payload(s, &bytes));
 
             let mut req = client
@@ -113,6 +260,10 @@ async fn fire_event_async(
                         "http_error"
                     };
                     crate::metrics::record_webhook_delivery(result_label);
+                    // Any HTTP response proves the delivery fired; mark
+                    // sent so the ledger never claims an unsent delivery.
+                    // Network errors stay pending for stale-reclaim.
+                    let _ = db_for_mark.mark_webhook_sent(&delivery_for_mark).await;
                     tracing::info!(
                         url = %hook.url,
                         event = %event_name,

crates/gitlawb-node/tests/inv22_gates.rs

@@ -416,8 +416,8 @@ fn f3_second_writer_leased_until_reap() {
          NOT RepoWriteGuard (which drops at the disconnect instant, reopening F3).",
     );
     let receive = repos_production
-        .find("smart_http::receive_pack(")
-        .expect("F3 gate stale: git_receive_pack no longer calls smart_http::receive_pack");
+        .find("smart_http::receive_pack_raw")
+        .expect("F3 gate stale: git_receive_pack no longer calls smart_http::receive_pack_raw");
     assert!(
         lease_acquire < with_lease && with_lease < receive,
         "F3 gate bypassed: the write lease must be acquired, then carried by the \
@@ -509,8 +509,8 @@ fn inv22_ipfs_walk_admission_reaches_every_blocking_site() {
 /// too, so an ungated spawn would fire for a push git rejected, pinning and announcing
 /// a half-applied repo. Above `release` the `?` on `receive_result` can no longer be
 /// what gates it, so the success check is explicit and this gate binds it: the spawn
-/// must sit inside `if push_succeeded`, and `release` must consume the same flag so
-/// the two cannot drift apart.
+/// must sit inside `if disposition.spawn_tail`, and `release` must consume the same
+/// disposition (`disposition.release_ok`) so the two cannot drift apart.
 ///
 /// This is an ordering check rather than a cancellation-race test on purpose: it is
 /// the companion to `receive_pack_tail_survives_a_disconnect_during_release`, which
@@ -519,7 +519,9 @@ fn inv22_ipfs_walk_admission_reaches_every_blocking_site() {
 ///
 /// MUTATION (RED): move the `tokio::spawn(post_receive_replication_tail` call below
 /// `guard.release(` and the ordering assertion fails; take it out of the
-/// `if push_succeeded` block and the failed-push assertion fails.
+/// `if disposition.spawn_tail` block and the failed-push assertion fails; gate
+/// `spawn_tail`/`release_ok` on the outcome commit and
+/// `post_git_disposition_replication_carve_out` fails with it.
 #[test]
 fn inv22_replication_tail_spawns_at_the_durability_boundary() {
     let repos = src("api/repos.rs");
@@ -534,41 +536,589 @@ fn inv22_replication_tail_spawns_at_the_durability_boundary() {
         .expect("split always yields a first chunk");
 
     let success_flag = production
-        .find("let push_succeeded = receive_result.is_ok();")
-        .expect(
-            "U5 gate missing: the tail's success gate must be bound from receive_result.is_ok()",
+        .find("post_git_disposition(")
+        .expect("U5 gate missing: the tail's disposition must be bound before the gate");
+    // One tail per push, on the proven-landed ref set. Either spawn
+    // position (carve-out early, or durability-bound at release) must sit
+    // inside a `disposition.spawn_tail` gate and feed `landed_refs` —
+    // never the unfiltered `ref_updates`.
+    let spawn_count = production
+        .matches("tokio::spawn(post_receive_replication_tail(")
+        .count();
+    assert!(
+        spawn_count >= 1,
+        "U5 gate missing: the replication tail must be spawned by git_receive_pack"
+    );
+    for (i, (pos, _)) in production
+        .match_indices("tokio::spawn(post_receive_replication_tail(")
+        .enumerate()
+    {
+        let gate = production[..pos].rfind("if disposition.spawn_tail").expect(
+            "U5 gate missing: every tail spawn must be gated on `if disposition.spawn_tail`",
         );
-    let gate_open = production
-        .find("if push_succeeded {")
-        .expect("U5 gate missing: the tail spawn must be gated on the push having succeeded");
+        // The spawn must sit inside the gate's block: brace depth at the
+        // spawn position must still be positive measured from the `if`.
+        // An ungated spawn after the block closes fails this, as does a
+        // stray `disposition.spawn_tail` mention in a comment.
+        let mut depth = 0i32;
+        for ch in production[gate..pos].chars() {
+            match ch {
+                '{' => depth += 1,
+                '}' => depth -= 1,
+                _ => {}
+            }
+        }
+        assert!(
+            depth >= 1,
+            "U5 gate bypassed: tail #{i} sits outside the `if disposition.spawn_tail` block"
+        );
+        assert!(
+            !production[gate..pos].contains("post_receive_replication_tail"),
+            "U5 gate bypassed: a second tail spawn sits between the gate and its spawn"
+        );
+        let call_end = production[pos..].find(");").unwrap() + pos;
+        let call = &production[pos..call_end];
+        assert!(
+            call.contains("landed_refs"),
+            "U5 gate bypassed: tail #{i} must carry the proven-landed ref set, not unfiltered ref_updates"
+        );
+    }
     let spawn = production
         .find("tokio::spawn(post_receive_replication_tail(")
         .expect("U5 gate missing: the replication tail must be spawned by git_receive_pack");
     let release = production
-        .find(".release(push_succeeded)")
-        .expect("U5 gate stale: release must consume the same success flag as the tail gate");
+        .find(".release(disposition.release_ok)")
+        .expect("U5 gate stale: release must consume the same disposition as the tail gate");
     let touch = production
         .find("state.db.touch_repo(")
         .expect("U5 gate stale: git_receive_pack no longer calls touch_repo");
-    let webhook = production
-        .find("webhooks::fire_event(")
-        .expect("U5 gate stale: git_receive_pack no longer fires push webhooks");
+    // #26 Split PR 1 step 3 — the webhook fan-out moved into
+    // `durable_outbox::apply_request_effects`, which the live
+    // handler calls inline (the recovery drain calls the same
+    // function on the next startup). The gate now pins that the
+    // handler is wired to the executor, not to a per-ref inline
+    // webhook call.
+    let effects_executor = production
+        .find("apply_request_effects(&state, &request_id)")
+        .expect("U5 gate stale: git_receive_pack no longer calls apply_request_effects inline");
 
     assert!(
-        success_flag < gate_open && gate_open < spawn,
-        "U5 gate bypassed: the tail must be spawned inside `if push_succeeded`, or a \
-         rejected push spawns a tail that pins and announces a half-applied repo"
-    );
-    // Still inside that block: no `}` may close it between the gate and the spawn.
-    assert!(
-        !production[gate_open + "if push_succeeded {".len()..spawn].contains('}'),
-        "U5 gate bypassed: the tail spawn left the `if push_succeeded` block, so a \
-         rejected push now spawns a tail"
+        success_flag < spawn,
+        "U5 gate bypassed: the disposition must be bound before any tail spawn"
     );
     assert!(
-        spawn < release && spawn < touch && spawn < webhook,
+        spawn < release && spawn < touch && spawn < effects_executor,
         "U5 gate bypassed: the tail must be spawned BEFORE guard.release, touch_repo \
-         and the webhook fan-out, so a disconnect in any of those windows cannot drop \
+         and the effect executor, so a disconnect in any of those windows cannot drop \
          this push's pins, recovery copy, and announcements"
     );
 }
+
+/// #26 Split PR 1 step 3 — drain + handler share a single effect
+/// executor. The live handler (api/repos.rs) and the recovery
+/// drain (`durable_outbox::drain_receive_pack_requests_with`) both
+/// call `apply_request_effects`, so the per-ref effects fan-out
+/// lives in exactly one place. The v29 per-ref walk
+/// (`derive_one`, `drain_pending_ref_transitions_all`,
+/// `lookup_accepted_ordinal`) is dead code: any caller reintroduced
+/// would be the per-ref walk the step-3 PR removed. This gate
+/// fails if a call site slips back in or the new seam is bypassed.
+#[test]
+fn inv26_step3_live_and_drain_share_apply_request_effects() {
+    let repos = src("api/repos.rs");
+    let outbox = src("durable_outbox.rs");
+
+    // The live handler calls `apply_request_effects`. Split the
+    // file at the test attribute so test code can't satisfy the
+    // gate by itself.
+    let production_repos = repos
+        .split("\nmod tests {")
+        .next()
+        .expect("split always yields a first chunk");
+
+    assert!(
+        production_repos.contains("apply_request_effects(&state, &request_id)"),
+        "live handler must call `apply_request_effects(&state, &request_id)`; \
+         reverting to a per-ref inline fan-out splits live and recovery"
+    );
+
+    // The durable intent insert closes the pre-outbox crash window this
+    // PR exists to fix. Wrapping it in `if false` must turn this gate
+    // red; drain-only tests cannot prove the handler creates the rows.
+    assert!(
+        production_repos.contains("insert_receive_pack_request_with_children("),
+        "live handler must persist durable intent via \
+         `insert_receive_pack_request_with_children`; disabling it must fail this gate"
+    );
+
+    // The drain's per-request seam calls the same executor. Test
+    // code lives below `mod drain_tests`, so split there too.
+    let production_outbox = outbox
+        .split("\nmod drain_tests {")
+        .next()
+        .expect("split always yields a first chunk");
+
+    assert!(
+        production_outbox.contains("drain_receive_pack_requests_with"),
+        "drain seam `drain_receive_pack_requests_with` must exist; \
+         removing it forces a per-ref walk back into the drain"
+    );
+    assert!(
+        production_outbox.contains("apply_request_effects"),
+        "durable_outbox production code must define `apply_request_effects`; \
+         removing it splits the executor between live and recovery"
+    );
+
+    // The drain's per-request walker is wired to the executor.
+    // The closure body of `drain_receive_pack_requests_with` is the
+    // only call site — if a future change calls `derive_one`
+    // instead, this assertion fires.
+    let drain_seam_open = production_outbox
+        .find("pub async fn drain_receive_pack_requests_with<F, Fut>")
+        .expect("drain seam must be defined in the production half");
+    let drain_seam_close = production_outbox[drain_seam_open..]
+        .find("\n}\n")
+        .expect("drain seam body must close");
+    let drain_seam_body = &production_outbox[drain_seam_open..drain_seam_open + drain_seam_close];
+    assert!(
+        drain_seam_body.contains("apply_request_effects"),
+        "drain_receive_pack_requests_with must call apply_request_effects; \
+         wiring it to a per-ref helper reintroduces the v29 walk"
+    );
+
+    // The deleted per-ref drain must have zero call sites in the
+    // production half. A regression that re-adds a caller would
+    // bring back the per-ref fan-out.
+    assert!(
+        !production_outbox.contains("drain_pending_ref_transitions_all("),
+        "deleted `drain_pending_ref_transitions_all` must have zero production call sites; \
+         a per-ref walk is reintroduced"
+    );
+    assert!(
+        !production_outbox.contains("derive_one("),
+        "deleted `derive_one` must have zero production call sites; \
+         the per-ref fan-out is reintroduced"
+    );
+    assert!(
+        !production_outbox.contains("lookup_accepted_ordinal("),
+        "deleted `lookup_accepted_ordinal` must have zero production call sites; \
+         the per-ref ordinal lookup is reintroduced"
+    );
+    assert!(
+        !production_repos.contains("derive_one("),
+        "deleted `derive_one` must have zero live-handler call sites"
+    );
+    assert!(
+        !production_repos.contains("drain_pending_ref_transitions"),
+        "deleted per-ref drain functions must have zero live-handler call sites"
+    );
+}
+
+/// #26 Split PR 1 step 4 — the periodic queue-lifecycle purge is
+/// wired in `main.rs` and the contract is enforced by the `idx_receive_pack_requests_completed_at`
+/// partial index from v41.
+///
+/// Assertions:
+/// 1. `main.rs` calls `purge_request_queue` once at boot (well, on
+///    the spawn-task interval) and the spawn function exists.
+/// 2. The DB helpers `purge_completed_receive_pack_requests` and
+///    `purge_completed_pending_ref_transitions` exist in `db/mod.rs`.
+/// 3. The config knobs `queue_retention_days` and `queue_purge_batch` exist
+///    on `Config`.
+/// 4. `quarantined` is NOT in the purge WHERE clause (step 4 does not
+///    introduce the state, but the invariant holds for the future).
+#[test]
+fn inv26_step4_queue_lifecycle_purge_is_wired() {
+    let main_src = src("main.rs");
+    assert!(
+        main_src.contains("spawn_queue_lifecycle_sweep"),
+        "main.rs must spawn the periodic queue-lifecycle purge"
+    );
+    assert!(
+        main_src.contains("purge_request_queue"),
+        "main.rs must call purge_request_queue on the periodic sweep"
+    );
+
+    let db_src = src("db/mod.rs");
+    assert!(
+        db_src.contains("purge_completed_receive_pack_requests"),
+        "db/mod.rs must expose purge_completed_receive_pack_requests"
+    );
+    assert!(
+        db_src.contains("purge_completed_pending_ref_transitions"),
+        "db/mod.rs must expose purge_completed_pending_ref_transitions"
+    );
+
+    let config_src = src("config.rs");
+    assert!(
+        config_src.contains("queue_retention_days"),
+        "Config must expose queue_retention_days"
+    );
+    assert!(
+        config_src.contains("queue_purge_batch"),
+        "Config must expose queue_purge_batch"
+    );
+}
+
+/// #26 Split PR 1 step 5 — the marker gate and the retry-bound
+/// quarantine are wired end-to-end. A missing or hash-mismatched
+/// marker on disk quarantines the request; a Retry over the bound
+/// does too. This gate pins every load-bearing seam, against the
+/// production half of each file (the test modules name the same
+/// identifiers in their own harnesses).
+///
+/// Assertions:
+/// 1. `reconcile_prepared_page` calls `mark_request_quarantined`
+///    and `mark_children_rejected_for_quarantined_parent` on
+///    marker-gate failure.
+/// 2. The drain's `EffectsOutcome::Retry` arm checks
+///    `effects_max_attempts`.
+/// 3. `git::store::read_ref` exists in `git/store.rs`.
+/// 4. `db::mark_request_quarantined`,
+///    `db::mark_children_rejected_for_quarantined_parent`,
+///    `db::get_receive_pack_requests_by_ids`, and
+///    `db::request_state::QUARANTINED` all exist.
+/// 5. The handler writes the marker ref BEFORE calling
+///    `smart_http::receive_pack_raw` (the durability window).
+#[test]
+fn inv26_step5_marker_quarantine_and_bound_are_wired() {
+    let outbox = src("durable_outbox.rs");
+    let store = src("git/store.rs");
+    let db = src("db/mod.rs");
+    let repos = src("api/repos.rs");
+
+    // Split at the TEST MODULE for the production-only assertions.
+    let production_outbox = outbox
+        .split("\nmod drain_tests {")
+        .next()
+        .expect("split always yields a first chunk");
+    let production_repos = repos
+        .split("\nmod tests {")
+        .next()
+        .expect("split always yields a first chunk");
+
+    // (1) The reconcile's marker gate quarantines via the DB helpers.
+    assert!(
+        production_outbox.contains("mark_request_quarantined"),
+        "reconcile_prepared_page must call mark_request_quarantined on marker-gate failure"
+    );
+    assert!(
+        production_outbox.contains("mark_children_rejected_for_quarantined_parent"),
+        "reconcile_prepared_page must call mark_children_rejected_for_quarantined_parent \
+         so quarantined parents cancel their children"
+    );
+
+    // The gate reads the marker ref and compares against the parent's
+    // `request_bytes_hash` via `git::store::read_ref` /
+    // `git::store::marker_value_for`. Reverting either reintroduces
+    // the DoS window the marker gate exists to close.
+    assert!(
+        production_outbox.contains("git::store::read_ref"),
+        "reconcile_prepared_page must read the marker ref via git::store::read_ref"
+    );
+    assert!(
+        production_outbox.contains("marker_value_for"),
+        "reconcile_prepared_page must compute the expected marker value via \
+         git::store::marker_value_for"
+    );
+
+    // (2) The drain's `EffectsOutcome::Retry` arm checks the bound.
+    assert!(
+        production_outbox.contains("effects_max_attempts"),
+        "drain_receive_pack_requests_with must consult effects_max_attempts on Retry"
+    );
+
+    // (3) `git::store::read_ref` is the read seam the gate depends on.
+    assert!(
+        store.contains("pub fn read_ref("),
+        "git::store::read_ref must exist; the marker gate reads through it"
+    );
+    assert!(
+        store.contains("pub fn marker_value_for("),
+        "git::store::marker_value_for must exist; the marker gate computes the expected \
+         value with it (and the live handler writes the value via the same helper)"
+    );
+
+    // (4) DB-side seams the gate depends on.
+    assert!(
+        db.contains("pub async fn mark_request_quarantined"),
+        "Db::mark_request_quarantined must exist"
+    );
+    assert!(
+        db.contains("pub async fn mark_children_rejected_for_quarantined_parent"),
+        "Db::mark_children_rejected_for_quarantined_parent must exist"
+    );
+    assert!(
+        db.contains("pub async fn get_receive_pack_requests_by_ids"),
+        "Db::get_receive_pack_requests_by_ids must exist (avoids N+1 in the marker gate)"
+    );
+    assert!(
+        db.contains("pub const QUARANTINED: &str = \"quarantined\""),
+        "request_state::QUARANTINED must be defined"
+    );
+
+    // (5) The handler writes the marker ref BEFORE the durability
+    // boundary (smart_http::receive_pack_raw). Severing the
+    // ordering re-opens the marker-gate DoS window for live pushes.
+    let marker_write = production_repos
+        .find("git::store::marker_value_for")
+        .expect("U5 gate stale: the live handler no longer computes the marker value");
+    let receive_raw = production_repos
+        .find("smart_http::receive_pack_raw")
+        .expect("U5 gate stale: git_receive_pack no longer calls smart_http::receive_pack_raw");
+    assert!(
+        marker_write < receive_raw,
+        "U5 gate bypassed: the marker ref must be written BEFORE receive_pack_raw so the \
+         reconcile's gate has evidence of the live push"
+    );
+}
+
+/// Issue #26 split 1 — namespace gate and pre-git refusal lifecycle.
+///
+/// The push gate (repos.rs) must deny:
+/// 1. Any ref-update line that fails UTF-8 decode (the raw body is passed verbatim
+///    to git receive-pack, and git accepts refnames with bytes >= 0x80; the gate
+///    never sees these refs, no durable child row exists, and under requests/issues/
+///    the lossy for-each-ref decode still matches the narrowed exemption).
+/// 2. The bare `refs/gitlawb` name (the `starts_with("refs/gitlawb/")` check misses it).
+///
+/// Every explicit pre-git refusal must call `refuse_receive_pack_before_git` to
+/// terminalize the intent aggregate in one transaction, or a dropped future between
+/// the intent insert and `receive_pack` strands a `received` + `prepared` aggregate
+/// that is never purged, never promotable, and poisons its tuple.
+///
+/// The visibility exemption (visibility_pack.rs) was narrowed to `requests/` and
+/// `issues/` only. The push gate is the premise that narrowing relies on (the comment
+/// at visibility_pack.rs:340), so tests must prove:
+/// - A `refs/gitlawb/<other>` ref pointing at a blob is denied on push.
+/// - A stray `refs/gitlawb/<other>` blob ref planted in the pack path fails closed.
+///
+/// Source gates below pin the wiring; behavioral proofs live next to the code:
+/// `parse_ref_updates_refuses_non_utf8_command_line` and
+/// `internal_namespace_gate_covers_bare_name_and_subtree` (api/repos.rs),
+/// `planted_non_exempt_gitlawb_blob_ref_fails_closed` (git/visibility_pack.rs),
+/// and the mirror import prune
+/// `prune_removes_non_exempt_gitlawb_refs_and_keeps_exempt` (sync.rs).
+#[test]
+fn issue_26_namespace_gate_and_refusal_wiring() {
+    let repos = src("api/repos.rs");
+    let vis = src("git/visibility_pack.rs");
+
+    // P1: fail the push on ref-update lines the handler cannot decode.
+    // The gate at parse_ref_updates must refuse when any command line in the
+    // ref-update section is undecodable UTF-8. The check is an Err arm on
+    // `from_utf8(data)` that returns BadRequest, so removing it allows non-UTF-8
+    // refs to bypass the namespace gate.
+    assert!(
+        repos.contains("ref-update line is not valid UTF-8; push refused"),
+        "P1 gate missing: parse_ref_updates must refuse when any ref-update line is \
+         non-UTF-8 (git accepts bytes >= 0x80; the gate never sees them, no durable \
+         child exists, and the lossy for-each-ref decode still matches the exemption)"
+    );
+
+    // P1 (cont): the bare `refs/gitlawb` name must also be caught. The
+    // predicate is extracted as `ref_is_internal_namespace` so it is
+    // unit-testable; pin the bare-name arm inside it and its use at the
+    // push gate (a `starts_with("refs/gitlawb/")`-only check misses the
+    // bare name, and it lands on repos with no internal refs yet).
+    assert!(
+        repos.contains("ref_name == \"refs/gitlawb\"")
+            && repos.contains(".find(|r| ref_is_internal_namespace(r))"),
+        "P1 gate missing: ref_is_internal_namespace must match the bare \
+         'refs/gitlawb' name and the push gate must consult it \
+         (starts_with('refs/gitlawb/') misses it, and it lands on repos with \
+         no internal refs yet)"
+    );
+
+    // P1 (cont): every explicit pre-git refusal must call refuse_receive_pack_before_git
+    // to terminalize the intent. Exactly 6 occurrences: the five inline call
+    // sites (lease cap, per-source cap, write pool, acquire timeout, acquire
+    // error) plus the drop-guard's own pre-git spawn. The post-git guard phase
+    // calls mark_receive_pack_interrupted (a different name), so deleting any
+    // one inline site drops the count to 5 and turns this red. The parse and
+    // namespace refusals correctly carry no call: they return before the
+    // intent insert, so there is no aggregate to terminalize.
+    assert_eq!(
+        repos.matches("refuse_receive_pack_before_git(").count(),
+        6,
+        "P1 wiring: exactly 6 refuse_receive_pack_before_git calls expected (5 \
+         inline refusal sites + guard pre-git spawn); deleting any inline site \
+         must turn this red"
+    );
+    // Pin the five inline sites by their reason literals, so a site that stops
+    // terminalizing (or is renumbered away) fails by name, not just by count.
+    for reason in [
+        "repo write-lease waiter cap reached",
+        "admission per-source cap reached",
+        "git write pool saturated",
+        "acquire_write timed out",
+        "acquire_write failed",
+        "handler dropped before git started",
+    ] {
+        assert!(
+            repos.contains(reason),
+            "P1 wiring: refusal reason literal missing: {reason}"
+        );
+    }
+
+    // P1 (cont): the intent drop-guard must be phase-aware. It is armed before
+    // the insert resolves (covering the cancellation window), flipped to
+    // git-started immediately before the receive_pack await (a drop from there
+    // is a mid-git disconnect, terminalized via mark_receive_pack_interrupted
+    // because the Err arm never runs on a drop), and disarmed once the outcome
+    // commit resolves. Assert the ordering, not just presence: the phase flip
+    // must precede the receive_pack call.
+    assert!(
+        repos.contains("IntentDropGuard::new") && repos.contains("notify_git_started()"),
+        "P1 gate missing: IntentDropGuard must exist and flip to git-started phase \
+         before receive_pack (a dropped future between intent insert and git start \
+         strands a received+prepared aggregate)"
+    );
+    let flip_pos = repos
+        .find("notify_git_started()")
+        .expect("notify_git_started call must exist");
+    let git_pos = repos
+        .find("receive_pack_raw_with_reflog(")
+        .expect("receive_pack call must exist");
+    assert!(
+        flip_pos < git_pos,
+        "P1 ordering: notify_git_started() must precede the receive_pack call, \
+         or a mid-git drop is misclassified as a pre-git shed"
+    );
+    assert!(
+        repos.contains("_intent_guard.disarm()"),
+        "P1 gate missing: the guard must disarm once the outcome commit resolves, \
+         or every push spawns a wasteful cleanup txn at scope end"
+    );
+    // A sync Drop cannot await: cleanup is spawned via the current runtime
+    // handle with an off-runtime fallback. A bare tokio::spawn panics
+    // mid-unwind with no runtime (off-runtime drop, process teardown).
+    assert!(
+        repos.contains("Handle::try_current()"),
+        "P1 gate missing: guard Drop must spawn via Handle::try_current with an \
+         off-runtime arm, not a bare tokio::spawn"
+    );
+
+    // P2: the post-git drop arm must exist and be wired. The guard's
+    // GIT_STARTED phase spawns `mark_receive_pack_interrupted` (parent to
+    // `rejected_at_git`, children to `uncertain` — the Err arm never runs on
+    // a drop). Deleting that spawn keeps every count above green, so pin the
+    // method name; its behavior is proved by
+    // `post_git_interruption_marks_uncertain_and_terminalizes_parent`.
+    assert!(
+        repos.contains("mark_receive_pack_interrupted("),
+        "P2 wiring: the guard's post-git phase must spawn \
+         mark_receive_pack_interrupted (a mid-git drop otherwise strands \
+         received+prepared with no Err arm to mark them uncertain)"
+    );
+    // The guard must commit computed fates rather than a blanket uncertain
+    // once the report is parsed, and only disarm on a resolved commit: pin
+    // both the stash call and the `outcome_commit_ok` gate. Making disarm()
+    // unconditional strands the commit-failure path and misfiles proven
+    // rejections as uncertain.
+    assert!(
+        repos.contains("set_fates(ComputedFates"),
+        "P2 wiring: computed fates must be stashed on the guard once parsed, \
+         or a post-commit drop can only apply a context-free uncertain"
+    );
+    assert!(
+        repos.contains("if outcome_commit_ok"),
+        "P2 wiring: disarm must be gated on outcome_commit_ok; unconditional \
+         disarm strands the persistent-commit-failure path"
+    );
+    // The live-handler `AwaitingSiblings` arm is unreachable through the
+    // current outcome builder (documented at the arm itself) and therefore
+    // has no behavioral test; pin its presence so removing it is conscious.
+    // The drain consumer IS behaviorally driven (sibling-wait tests).
+    assert_eq!(
+        repos.matches("EffectsOutcome::AwaitingSiblings").count(),
+        1,
+        "live AwaitingSiblings arm must exist (defensive symmetry with the \
+         drain consumer); deleting it must turn this red"
+    );
+
+    // P2: pin the mirror prune call sites. The prune helper is called from
+    // both `clone_repo` and `fetch_repo` (plus its own definition) — deleting
+    // either call leaves the direct-helper test green while mirrors import
+    // evil refs again. Count occurrences of the call with its argument.
+    let sync_src = src("sync.rs");
+    assert_eq!(
+        sync_src
+            .matches("prune_non_exempt_gitlawb_refs(local_path)")
+            .count(),
+        2,
+        "P2 wiring: prune must be called from both clone_repo and fetch_repo; \
+         deleting either call must turn this red"
+    );
+    // Both call sites must route through `wrap_prune_error`, which preserves
+    // deterministic `PruneInvalid` refusals instead of stringifying every
+    // error into transient `PruneFailed` (an every-tick deferral spin for
+    // permanently invalid mirrors). Pin the exact call shape at both sites.
+    assert_eq!(
+        sync_src.matches(".map_err(wrap_prune_error)").count(),
+        2,
+        "P2 wiring: both prune call sites must use wrap_prune_error; inline \
+         rewrap must turn this red"
+    );
+
+    // P2: the visibility exemption narrowing removed a serving path. The exemption
+    // is now scoped to `requests/` and `issues/` only (visibility_pack.rs:371), and
+    // the comment at :340 states the push gate is the premise. Tests must prove the
+    // gate works, which the P1 findings above covered (non-UTF-8 line, bare name,
+    // refusal wiring). This check verifies the exemption narrowing landed.
+    assert!(
+        vis.contains(
+            r#"!r.starts_with("refs/gitlawb/requests/") && !r.starts_with("refs/gitlawb/issues/")"#
+        ),
+        "P2 gate missing: the visibility exemption must be narrowed to requests/ and \
+         issues/ only (the push gate is the premise; any other refs/gitlawb/* ref is \
+         denied on push and must fail closed in the pack path)"
+    );
+}
+
+/// Claimed-webhook delivery must not re-list hooks: the durable path claims
+/// synchronously, hands the claimed hook rows to the sender, and a list
+/// failure in the send task would silently drop already-claimed deliveries.
+#[test]
+fn inv26_claimed_hook_delivery_never_relists() {
+    let webhooks = src("webhooks.rs");
+    let production = webhooks
+        .split("\nmod tests {")
+        .next()
+        .expect("split always yields a first chunk");
+
+    // The claimed entry point exists and passes hook ROWS (not ids) so the
+    // sender never needs the hook table again.
+    let entry = production
+        .find("fire_event_occurrence_with_claimed_hooks(")
+        .expect("claimed-hooks sender missing");
+    let claimed_rows = production
+        .find("claimed_hooks: Vec<crate::db::Webhook>")
+        .expect("claimed path must carry Webhook rows, not ids");
+    let list_call = production
+        .find("db.list_webhooks_for_event(repo_id, event)")
+        .expect("unclaimed path still lists hooks");
+    assert!(
+        entry < claimed_rows && claimed_rows < list_call,
+        "claimed sender must hand rows through before any list fallback"
+    );
+
+    // Inside the async worker, the claimed branch must bypass the list.
+    // Inside the async worker (which starts after the claim helper), the
+    // claimed branch must be selected before the list fallback can run.
+    let worker = production
+        .find("async fn fire_event_async_occurrence(")
+        .expect("occurrence worker missing");
+    let already_claimed = production[worker..]
+        .find("let already_claimed = claimed_hooks.is_some();")
+        .map(|i| worker + i)
+        .expect("claimed path must short-circuit before re-listing");
+    let worker_list = production[worker..]
+        .find("db.list_webhooks_for_event(repo_id, event)")
+        .map(|i| worker + i)
+        .expect("unclaimed path still lists hooks");
+    assert!(
+        already_claimed < worker_list,
+        "the claimed branch must be selected before list_webhooks_for_event can run"
+    );
+}