Skip to main content

choir_node/
platform.rs

1//! The platform API: signed op submission and view queries over HTTP.
2//!
3//! This is the production composition the sequencer's `policy.rs` test
4//! proved: signature verification (choir-identity) + cached-view CAS
5//! (choir-view) running inside the single-writer thread. The daemon fronts
6//! it with signed single and batch submission, materialized-view and log
7//! reads, workspace provisioning, and review-queue endpoints. The two core
8//! operation paths are:
9//!
10//! - `POST /api/submit` — body `{"channel", "payload_hex",
11//!   "key_id", "signature_hex"}`; `workspace` remains accepted as the
12//!   legacy v1 alias for `channel`. Payload bytes are a serialized
13//!   [`ViewOp`]. Admitted ops answer `{"seq", "hash"}`; rejections are
14//!   HTTP 400 with the policy's reason.
15//! - `GET /api/view` — the current materialized state plus runtime-only
16//!   projections for D24 T3 concentration and complete-view growth. They
17//!   derive from the same coherent snapshot and never enter persisted ops
18//!   or hash input.
19//!
20//! Hex (not JSON-embedding) carries the payload because the signature
21//! covers the exact bytes the author serialized; re-encoding through a
22//! JSON tree could legally reorder/respace them and break verification.
23
24use std::collections::{BTreeMap, BTreeSet, VecDeque};
25use std::io::Write;
26use std::sync::{Arc, Mutex};
27use std::time::{Duration, Instant};
28
29use choir_identity::{ActorKey, Registry};
30use choir_oplog::{ContentHash, OpEntry, OpLog, Witness};
31use choir_sequencer::journal;
32use choir_sequencer::journal::Journal as _;
33use choir_sequencer::lag::LagMeter;
34use choir_sequencer::{Sequencer, SequencerHandle, Submission, SubmitPolicy};
35use choir_view::{
36    reviewer_operator, ArchiveAuthorization, Authorization, Basis, ChangeState,
37    CreateAuthorization, OpKind, Provenance, ReviewStatus, Verdict, View, ViewOp,
38};
39
40use crate::reject::{Code, Rejection};
41
42/// Default maximum operations accepted in one `/api/submit-batch` body.
43pub const DEFAULT_BATCH_OPS: usize = 256;
44
45/// Operator-selected bound for live review detail retained in memory.
46///
47/// Complete reviews older than `max_live` are archived immediately. An
48/// incomplete review is never archived unless `lapse_after` is explicitly
49/// set, because choosing when an unanswered review is abandoned is policy,
50/// not a harmless memory optimization.
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub struct ReviewRetention {
53    max_live: usize,
54    lapse_after: Option<Duration>,
55}
56
57impl ReviewRetention {
58    /// Retains at most `max_live` live reviews when enough reviews are
59    /// complete and therefore safe to archive. Incomplete reviews never
60    /// lapse under this configuration.
61    #[must_use]
62    pub const fn keep(max_live: usize) -> Self {
63        Self {
64            max_live,
65            lapse_after: None,
66        }
67    }
68
69    /// Allows an over-limit incomplete review to lapse after `age` since
70    /// this node observed its request.
71    ///
72    /// The op log has no timestamp, so reviews replayed at startup begin a
73    /// fresh grace period. Restarting can delay a lapse, never make one
74    /// happen early. The emitted `ArchiveReview { lapsed: true }` op is
75    /// persisted, so replicas replay the decision rather than their clocks.
76    #[must_use]
77    pub const fn lapse_incomplete_after(mut self, age: Duration) -> Self {
78        self.lapse_after = Some(age);
79        self
80    }
81}
82
83/// One live review in request-sequence order. `observed_at` is consulted
84/// only for the explicitly configured incomplete-review lapse policy.
85struct TrackedReview {
86    id: String,
87    observed_at: Option<Instant>,
88}
89
90/// Runtime state that exists only when retention was explicitly enabled.
91/// Archived ids are removed, so the tracker grows with live review detail,
92/// not with the repository's lifetime.
93struct ReviewRetentionState {
94    config: ReviewRetention,
95    live: VecDeque<TrackedReview>,
96    /// Set when an observed op may have changed what is prunable, cleared
97    /// by a pass. Without it the submit path rescans the whole tracker on
98    /// every write for as long as the node holds more live reviews than
99    /// the bound and none of them is archivable — which is the *default*
100    /// shape, because incomplete reviews never lapse unless an age is
101    /// configured. Measured at 4.4x on the submit path before this
102    /// existed (1,000 unanswered reviews, `--review-retention 10`).
103    prunable_changed: bool,
104}
105
106impl ReviewRetentionState {
107    fn new(config: ReviewRetention) -> Self {
108        Self {
109            config,
110            live: VecDeque::new(),
111            prunable_changed: false,
112        }
113    }
114
115    fn observe(&mut self, op: &ViewOp, observed_at: Option<Instant>) {
116        match &op.kind {
117            OpKind::RequestReview { id, .. } => {
118                self.live.push_back(TrackedReview {
119                    id: id.clone(),
120                    observed_at,
121                });
122                self.prunable_changed = true;
123            }
124            OpKind::ArchiveReview { id, .. } => {
125                self.live.retain(|review| review.id != *id);
126                self.prunable_changed = true;
127            }
128            // The ops that cannot move a review's count or completeness,
129            // named as an exclusion rather than listing the review ops
130            // positively: a review op added later then defaults to arming
131            // the flag — a wasted scan, never silently stopped pruning.
132            OpKind::SetWorkspaceHead { .. }
133            | OpKind::SetRef { .. }
134            | OpKind::DeleteRef { .. }
135            | OpKind::DeleteWorkspace { .. }
136            | OpKind::RecordProvenance { .. }
137            | OpKind::CreateChange { .. }
138            | OpKind::CheckpointChange { .. }
139            | OpKind::ArchiveChange { .. } => {}
140            _ => self.prunable_changed = true,
141        }
142    }
143
144    /// Whether a pass could possibly find work. Deliberately conservative:
145    /// it may say yes when the answer turns out to be no, never no when
146    /// the answer is yes.
147    fn worth_a_pass(&self) -> bool {
148        // Tracked ids are dropped on archive, so this is an upper bound on
149        // the live count: at or under the bound, no pass can find work.
150        if self.live.len() <= self.config.max_live {
151            return false;
152        }
153        if self.prunable_changed {
154            return true;
155        }
156        // Under a lapse policy the clock alone can make the oldest review
157        // eligible with no op arriving. The deque is in request order, so
158        // the front carries the earliest deadline and one comparison
159        // settles it. Count-only retention still reads no clock.
160        match (self.config.lapse_after, self.live.front()) {
161            (Some(age), Some(oldest)) => oldest
162                .observed_at
163                .is_some_and(|at| Instant::now().saturating_duration_since(at) >= age),
164            _ => false,
165        }
166    }
167}
168
169/// Result of maintenance emitted after an already-successful user request.
170/// A retention failure cannot roll that request back, so it is reported as
171/// an additive response field rather than changing the request's status.
172#[derive(Default)]
173struct ReviewPruneOutcome {
174    archived: Vec<String>,
175    errors: Vec<serde_json::Value>,
176}
177
178/// Sliding window over recent admitted entries: `base` is the seq of
179/// the first held entry, so `/api/log?from=` keeps absolute semantics
180/// after old entries are dropped (full history lives in the op log).
181pub struct LogWindow {
182    base: u64,
183    /// A `VecDeque`, not a `Vec`: eviction pops from the front, which is
184    /// O(1) here and an O(n) memmove of up to `cap` entries there. At the
185    /// 100k cap that cost was paid on every push once full.
186    entries: std::collections::VecDeque<OpEntry>,
187    /// Entries kept before the oldest is dropped. A field rather than a
188    /// constant so a test can drive eviction without writing 100k ops.
189    cap: usize,
190    /// `signing_hash(channel, payload)` → the `(seq, hash)` it landed
191    /// as, for the entries currently in the window.
192    ///
193    /// Lets a resubmission be told "this already landed, here is where"
194    /// instead of the CAS failure it would otherwise see — the two are
195    /// indistinguishable today, and an agent that cannot tell them apart
196    /// either retries a completed write or abandons a successful one.
197    ///
198    /// **Bounded by the window, deliberately.** An index over the whole
199    /// log would grow with the repository's lifetime, which is the exact
200    /// growth profile `FileLog`'s offset index was just built to remove.
201    /// A window is enough because duplicate submissions come from client
202    /// retries — seconds apart, not months — so anything old enough to
203    /// have fallen out of the window is not a retry.
204    ///
205    /// A `HashMap`, and that is safe here specifically because this is
206    /// runtime state that is never serialized or hashed. Invariant 3 —
207    /// "maps in hashed structs are `BTreeMap` so serialization stays
208    /// canonical" — applies to persisted structures; an in-memory index
209    /// has no canonical form to break. Do not "fix" this to a `BTreeMap`
210    /// for consistency: `ContentHash` is `Hash` but not `Ord`.
211    /// Stores the **seq only**. The entry hash is recoverable from the
212    /// window when the rare already-applied path needs it, and computing
213    /// it here would re-serialize every entry on the write path — the
214    /// allocation budget caught exactly that.
215    by_signing: std::collections::HashMap<ContentHash, u64>,
216    /// Entry hash → seq, for the heads a scoped op may name.
217    ///
218    /// Same bound, same reason, and it costs no hashing at all: the
219    /// sequencer hands `push` the hash it already computed, and eviction
220    /// reads the dropped entry's hash out of the next entry's `parent`
221    /// rather than re-deriving it.
222    by_hash: std::collections::HashMap<ContentHash, u64>,
223}
224
225/// Entries retained in memory for `/api/log`; older reads are served from
226/// the persisted op log when the node has one.
227const LOG_WINDOW_CAP: usize = 100_000;
228
229/// Entries per `/api/log` page, whichever source served them.
230const LOG_PAGE: usize = 500;
231
232/// Approval weight required to move a protected ref. The assignment draw
233/// targets the same number of independent operators, but a thin pool may
234/// return fewer; under-assignment stays visible and cannot lower this
235/// landing threshold (D24 layer 5).
236const REQUIRED_APPROVAL_WEIGHT: usize = 2;
237
238/// D24 T3 fires above these bounds. Shares use exact integer arithmetic;
239/// basis points are presentation only and never drive the decision.
240const T3_MAX_AGENT_KEYS_PER_OPERATOR: usize = 100;
241const T3_MAX_SHARE_PERCENT: usize = 1;
242
243/// D24 T2's declared bounds, recorded so the projection can state exactly
244/// which question it is *not* answering. Nothing is ever compared against
245/// them: choir persists no validity classification, so the numerator of a
246/// slop rate does not exist. See [`new_actor_review_outcomes_json`].
247const T2_MAX_INVALID_OR_SLOP_PERCENT: usize = 20;
248const T2_MIN_VALID_PERCENT: usize = 5;
249
250/// Version of the append-only newcomer audit and operator adjudication rows.
251/// These files are not part of the signed op log, but they are persisted
252/// measurement inputs and therefore carry the same explicit-version discipline.
253const NEWCOMER_AUDIT_FORMAT_VERSION: u64 = 1;
254
255#[derive(Debug, Clone)]
256struct NewcomerAttempt {
257    started_at_unix_ms: u64,
258    first_outcome: &'static str,
259    first_rejection_code: Option<String>,
260    first_accepted_at_unix_ms: Option<u64>,
261}
262
263/// Sparse, durable T4 evidence. Incumbents are the keys present when the
264/// operator enables the audit; only keys first admitted after that boundary are
265/// newcomers. At most two outcome records are written per actor (first attempt,
266/// then first acceptance after a rejection), so instrumentation cost scales with
267/// newcomers rather than submissions.
268struct NewcomerAudit {
269    file: std::fs::File,
270    adjudications_path: std::path::PathBuf,
271    activated: bool,
272    incumbents: BTreeSet<String>,
273    attempts: BTreeMap<u64, NewcomerAttempt>,
274    by_actor: BTreeMap<String, u64>,
275    appeals: BTreeSet<u64>,
276    next_attempt_id: u64,
277    available: bool,
278}
279
280impl NewcomerAudit {
281    fn open(
282        audit_path: &std::path::Path,
283        adjudications_path: std::path::PathBuf,
284        incumbents: BTreeSet<String>,
285    ) -> Result<Self, String> {
286        if let Some(parent) = audit_path.parent() {
287            std::fs::create_dir_all(parent)
288                .map_err(|e| format!("create newcomer audit directory: {e}"))?;
289        }
290        let existing = match std::fs::read_to_string(audit_path) {
291            Ok(existing) => existing,
292            Err(error) if error.kind() == std::io::ErrorKind::NotFound => String::new(),
293            Err(error) => return Err(format!("read newcomer audit: {error}")),
294        };
295        let mut options = std::fs::OpenOptions::new();
296        options.create(true).append(true);
297        #[cfg(unix)]
298        {
299            use std::os::unix::fs::OpenOptionsExt;
300            options.mode(0o600);
301        }
302        let file = options
303            .open(audit_path)
304            .map_err(|e| format!("open newcomer audit: {e}"))?;
305        #[cfg(unix)]
306        {
307            use std::os::unix::fs::PermissionsExt;
308            std::fs::set_permissions(audit_path, std::fs::Permissions::from_mode(0o600))
309                .map_err(|e| format!("chmod newcomer audit: {e}"))?;
310        }
311        let mut audit = Self {
312            file,
313            adjudications_path,
314            activated: false,
315            incumbents: BTreeSet::new(),
316            attempts: BTreeMap::new(),
317            by_actor: BTreeMap::new(),
318            appeals: BTreeSet::new(),
319            next_attempt_id: 0,
320            available: true,
321        };
322        for (index, line) in existing.lines().enumerate() {
323            if line.trim().is_empty() {
324                continue;
325            }
326            let value: serde_json::Value = serde_json::from_str(line)
327                .map_err(|e| format!("newcomer audit line {}: {e}", index + 1))?;
328            audit
329                .replay(&value)
330                .map_err(|e| format!("newcomer audit line {}: {e}", index + 1))?;
331        }
332        if audit.activated {
333            return Ok(audit);
334        }
335        if existing.lines().any(|line| !line.trim().is_empty()) {
336            return Err("newcomer audit has rows before its activation boundary".to_string());
337        }
338        let record = serde_json::json!({
339            "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
340            "kind": "activation",
341            "observed_at_unix_ms": unix_ms(),
342            "incumbent_actor_keys": incumbents,
343        });
344        audit.append(&record)?;
345        audit.replay(&record)?;
346        Ok(audit)
347    }
348
349    fn replay(&mut self, value: &serde_json::Value) -> Result<(), String> {
350        if value["format_version"].as_u64() != Some(NEWCOMER_AUDIT_FORMAT_VERSION) {
351            return Err("unsupported format_version".to_string());
352        }
353        let kind = value["kind"].as_str().ok_or("missing kind")?;
354        match kind {
355            "activation" => {
356                if self.activated || !self.attempts.is_empty() || !self.appeals.is_empty() {
357                    return Err("duplicate or late activation boundary".to_string());
358                }
359                value["observed_at_unix_ms"]
360                    .as_u64()
361                    .ok_or("activation needs observed_at_unix_ms")?;
362                let incumbent_actor_keys = value["incumbent_actor_keys"]
363                    .as_array()
364                    .ok_or("activation needs incumbent_actor_keys")?;
365                for actor_key in incumbent_actor_keys {
366                    let actor_key = actor_key
367                        .as_str()
368                        .filter(|actor_key| !actor_key.is_empty())
369                        .ok_or("incumbent actor keys must be non-empty strings")?;
370                    if !self.incumbents.insert(actor_key.to_string()) {
371                        return Err("activation has a duplicate incumbent actor key".to_string());
372                    }
373                }
374                self.activated = true;
375            }
376            "first_attempt" => {
377                if !self.activated {
378                    return Err("first_attempt precedes activation".to_string());
379                }
380                let attempt_id = value["attempt_id"].as_u64().ok_or("missing attempt_id")?;
381                if self.attempts.contains_key(&attempt_id) {
382                    return Err("duplicate first_attempt".to_string());
383                }
384                let actor_key = value["actor_key"]
385                    .as_str()
386                    .filter(|value| !value.is_empty())
387                    .ok_or("missing actor_key")?
388                    .to_string();
389                if self.incumbents.contains(&actor_key) {
390                    return Err("first_attempt belongs to an incumbent actor key".to_string());
391                }
392                if self.by_actor.contains_key(&actor_key) {
393                    return Err("actor has more than one first_attempt".to_string());
394                }
395                let started_at_unix_ms = value["started_at_unix_ms"]
396                    .as_u64()
397                    .ok_or("missing started_at_unix_ms")?;
398                let first_outcome = match value["outcome"].as_str() {
399                    Some("accepted") => "accepted",
400                    Some("rejected") => "rejected",
401                    _ => return Err("outcome must be accepted or rejected".to_string()),
402                };
403                let first_rejection_code = value["rejection_code"].as_str().map(str::to_string);
404                if (first_outcome == "rejected") != first_rejection_code.is_some() {
405                    return Err("rejected first attempts need one rejection_code".to_string());
406                }
407                let first_accepted_at_unix_ms = (first_outcome == "accepted").then_some(
408                    value["completed_at_unix_ms"]
409                        .as_u64()
410                        .ok_or("accepted attempt needs completed_at_unix_ms")?,
411                );
412                self.by_actor.insert(actor_key, attempt_id);
413                self.attempts.insert(
414                    attempt_id,
415                    NewcomerAttempt {
416                        started_at_unix_ms,
417                        first_outcome,
418                        first_rejection_code,
419                        first_accepted_at_unix_ms,
420                    },
421                );
422                self.next_attempt_id = self.next_attempt_id.max(attempt_id.saturating_add(1));
423            }
424            "first_accept" => {
425                let attempt_id = value["attempt_id"].as_u64().ok_or("missing attempt_id")?;
426                let completed = value["completed_at_unix_ms"]
427                    .as_u64()
428                    .ok_or("first_accept needs completed_at_unix_ms")?;
429                let attempt = self
430                    .attempts
431                    .get_mut(&attempt_id)
432                    .ok_or("first_accept precedes first_attempt")?;
433                if attempt.first_outcome != "rejected" {
434                    return Err("first_accept follows an accepted first attempt".to_string());
435                }
436                if attempt
437                    .first_accepted_at_unix_ms
438                    .replace(completed)
439                    .is_some()
440                {
441                    return Err("duplicate first_accept".to_string());
442                }
443            }
444            "appeal" => {
445                let attempt_id = value["attempt_id"].as_u64().ok_or("missing attempt_id")?;
446                if !self.attempts.contains_key(&attempt_id) {
447                    return Err("appeal references an unknown attempt".to_string());
448                }
449                if self.attempts[&attempt_id].first_outcome != "rejected" {
450                    return Err("appeal references an accepted first attempt".to_string());
451                }
452                if !self.appeals.insert(attempt_id) {
453                    return Err("duplicate appeal".to_string());
454                }
455            }
456            _ => return Err("unknown kind".to_string()),
457        }
458        Ok(())
459    }
460
461    fn append(&mut self, value: &serde_json::Value) -> Result<(), String> {
462        serde_json::to_writer(&mut self.file, value)
463            .map_err(|e| format!("write newcomer audit: {e}"))?;
464        self.file
465            .write_all(b"\n")
466            .and_then(|_| self.file.sync_data())
467            .map_err(|e| format!("sync newcomer audit: {e}"))
468    }
469
470    fn observe(
471        &mut self,
472        actor_key: &str,
473        started_at_unix_ms: u64,
474        accepted: bool,
475        rejection_code: Option<&str>,
476    ) -> Result<Option<u64>, String> {
477        // `actor_key` is the *claimed* key id, straight off an
478        // unverified signature, so nothing may be recorded under it
479        // until a signature check has vouched for it. Both signature
480        // failures have to be listed: this read `== Some("unknown_key")`
481        // while that code covered every verification failure, and
482        // splitting `bad_signature` out of it would otherwise have let
483        // anyone who knows a trusted key id append audit rows in that
484        // actor's name by sending deliberate garbage.
485        if self.incumbents.contains(actor_key)
486            || matches!(rejection_code, Some("unknown_key" | "bad_signature"))
487        {
488            return Ok(None);
489        }
490        let completed_at_unix_ms = unix_ms();
491        if let Some(attempt_id) = self.by_actor.get(actor_key).copied() {
492            let needs_accept = accepted
493                && self.attempts[&attempt_id]
494                    .first_accepted_at_unix_ms
495                    .is_none();
496            if needs_accept {
497                let record = serde_json::json!({
498                    "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
499                    "kind": "first_accept",
500                    "attempt_id": attempt_id,
501                    "completed_at_unix_ms": completed_at_unix_ms,
502                });
503                if let Err(error) = self.append(&record) {
504                    self.available = false;
505                    return Err(error);
506                }
507                self.attempts
508                    .get_mut(&attempt_id)
509                    .expect("attempt exists")
510                    .first_accepted_at_unix_ms = Some(completed_at_unix_ms);
511            }
512            return Ok(Some(attempt_id));
513        }
514
515        let attempt_id = self.next_attempt_id;
516        let outcome = if accepted { "accepted" } else { "rejected" };
517        let mut record = serde_json::json!({
518            "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
519            "kind": "first_attempt",
520            "attempt_id": attempt_id,
521            "actor_key": actor_key,
522            "started_at_unix_ms": started_at_unix_ms,
523            "completed_at_unix_ms": completed_at_unix_ms,
524            "outcome": outcome,
525        });
526        if let Some(code) = rejection_code {
527            record["rejection_code"] = serde_json::json!(code);
528        }
529        if let Err(error) = self.append(&record) {
530            self.available = false;
531            return Err(error);
532        }
533        self.replay(&record)?;
534        Ok(Some(attempt_id))
535    }
536
537    fn appeal(&mut self, attempt_id: u64) -> Result<(), String> {
538        let Some(attempt) = self.attempts.get(&attempt_id) else {
539            return Err("no such newcomer attempt".to_string());
540        };
541        if attempt.first_outcome != "rejected" {
542            return Err("only a rejected first attempt can be appealed".to_string());
543        }
544        if self.appeals.contains(&attempt_id) {
545            return Ok(());
546        }
547        let record = serde_json::json!({
548            "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
549            "kind": "appeal",
550            "attempt_id": attempt_id,
551            "observed_at_unix_ms": unix_ms(),
552        });
553        if let Err(error) = self.append(&record) {
554            self.available = false;
555            return Err(error);
556        }
557        self.replay(&record)
558    }
559}
560
561fn unix_ms() -> u64 {
562    std::time::SystemTime::now()
563        .duration_since(std::time::UNIX_EPOCH)
564        .map_or(0, |duration| {
565            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
566        })
567}
568
569fn median_u64(values: &mut [u64]) -> Option<u64> {
570    if values.is_empty() {
571        return None;
572    }
573    values.sort_unstable();
574    let middle = values.len() / 2;
575    if values.len() % 2 == 1 {
576        Some(values[middle])
577    } else {
578        Some(((u128::from(values[middle - 1]) + u128::from(values[middle])) / 2) as u64)
579    }
580}
581
582fn read_newcomer_adjudications(
583    path: &std::path::Path,
584    attempts: &BTreeMap<u64, NewcomerAttempt>,
585) -> Result<(BTreeMap<u64, bool>, String), String> {
586    let bytes = std::fs::read(path).map_err(|e| format!("read adjudications: {e}"))?;
587    let snapshot_hash = ContentHash::blake3(&bytes).to_hex();
588    let text = String::from_utf8(bytes).map_err(|_| "adjudications are not UTF-8".to_string())?;
589    let mut rows = BTreeMap::new();
590    for (index, line) in text.lines().enumerate() {
591        if line.trim().is_empty() {
592            continue;
593        }
594        let value: serde_json::Value = serde_json::from_str(line)
595            .map_err(|e| format!("adjudication line {}: {e}", index + 1))?;
596        if value["format_version"].as_u64() != Some(NEWCOMER_AUDIT_FORMAT_VERSION) {
597            return Err(format!(
598                "adjudication line {} has unsupported format_version",
599                index + 1
600            ));
601        }
602        let attempt_id = value["attempt_id"]
603            .as_u64()
604            .ok_or_else(|| format!("adjudication line {} needs attempt_id", index + 1))?;
605        let legitimate = value["legitimate"]
606            .as_bool()
607            .ok_or_else(|| format!("adjudication line {} needs legitimate", index + 1))?;
608        if !attempts.contains_key(&attempt_id) {
609            return Err(format!(
610                "adjudication line {} references an unknown attempt",
611                index + 1
612            ));
613        }
614        if rows.insert(attempt_id, legitimate).is_some() {
615            return Err(format!(
616                "adjudication line {} duplicates an attempt",
617                index + 1
618            ));
619        }
620    }
621    Ok((rows, snapshot_hash))
622}
623
624fn newcomer_harm_json(audit: Option<&Arc<Mutex<NewcomerAudit>>>) -> serde_json::Value {
625    let Some(audit) = audit else {
626        return serde_json::json!({
627            "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
628            "configured": false,
629            "available": false,
630            "tripwire_status": "indeterminate",
631            "evaluation_complete": false,
632        });
633    };
634    let audit = audit.lock().expect("newcomer audit lock");
635    let adjudications = read_newcomer_adjudications(&audit.adjudications_path, &audit.attempts);
636    let (rows, snapshot_hash, adjudications_available, adjudications_error) = match adjudications {
637        Ok((rows, hash)) => (rows, Some(hash), true, None),
638        Err(error) => (BTreeMap::new(), None, false, Some(error)),
639    };
640
641    let mut first_accepted = 0usize;
642    let mut first_rejected = 0usize;
643    let mut rejection_codes: BTreeMap<String, usize> = BTreeMap::new();
644    let mut legitimate = 0usize;
645    let mut legitimate_first_rejected = 0usize;
646    let mut legitimate_pending_acceptance = 0usize;
647    let mut accepted_latencies = Vec::new();
648    for (attempt_id, attempt) in &audit.attempts {
649        if attempt.first_outcome == "accepted" {
650            first_accepted += 1;
651        } else {
652            first_rejected += 1;
653            *rejection_codes
654                .entry(
655                    attempt
656                        .first_rejection_code
657                        .clone()
658                        .expect("rejected attempts carry a code"),
659                )
660                .or_default() += 1;
661        }
662        if rows.get(attempt_id) != Some(&true) {
663            continue;
664        }
665        legitimate += 1;
666        legitimate_first_rejected += usize::from(attempt.first_outcome == "rejected");
667        match attempt.first_accepted_at_unix_ms {
668            Some(accepted) => {
669                accepted_latencies.push(accepted.saturating_sub(attempt.started_at_unix_ms))
670            }
671            None => legitimate_pending_acceptance += 1,
672        }
673    }
674    let median_time_to_first_accepted_ms = median_u64(&mut accepted_latencies);
675    let adjudicated = rows.len();
676    let total = audit.attempts.len();
677    let unresolved_appeals = audit
678        .appeals
679        .iter()
680        .filter(|attempt_id| !rows.contains_key(attempt_id))
681        .count();
682    let false_reject_rate_basis_points =
683        (legitimate != 0).then(|| share_basis_points(legitimate_first_rejected, legitimate));
684    let measurement_complete = audit.available
685        && adjudications_available
686        && legitimate != 0
687        && adjudicated == total
688        && legitimate_pending_acceptance == 0;
689    serde_json::json!({
690        "format_version": NEWCOMER_AUDIT_FORMAT_VERSION,
691        "configured": true,
692        "available": audit.available && adjudications_available,
693        "scope": "post-activation cryptographically verified signed-API actor keys",
694        "thresholds": {
695            "false_reject_rate_basis_points": null,
696            "median_time_to_first_accepted_ms": null,
697            "status": "unset_pending_first_measurement",
698        },
699        "audit": {
700            "available": audit.available,
701            "incumbent_actor_keys_excluded": audit.incumbents.len(),
702            "first_attempts": total,
703            "first_attempts_accepted": first_accepted,
704            "first_attempts_rejected": first_rejected,
705            "first_rejections_by_code": rejection_codes,
706        },
707        "appeals": {
708            "submitted": audit.appeals.len(),
709            "unresolved": unresolved_appeals,
710        },
711        "adjudications": {
712            "available": adjudications_available,
713            "snapshot_hash": snapshot_hash,
714            "error": adjudications_error,
715            "attempts": adjudicated,
716            "coverage_basis_points": share_basis_points(adjudicated, total),
717            "legitimate_attempts": legitimate,
718        },
719        "measurements": {
720            "legitimate_first_attempts_rejected": legitimate_first_rejected,
721            "false_reject_rate_basis_points": false_reject_rate_basis_points,
722            "accepted_legitimate_newcomers": accepted_latencies.len(),
723            "legitimate_newcomers_pending_acceptance": legitimate_pending_acceptance,
724            "median_time_to_first_accepted_ms": median_time_to_first_accepted_ms,
725        },
726        "measurement_complete": measurement_complete,
727        "tripwire_status": "indeterminate",
728        "evaluation_complete": false,
729        "semantics": {
730            "false_reject": "an operator-adjudicated legitimate actor whose first verified signed-API attempt was rejected",
731            "time_to_first_accepted": "elapsed wall time from that actor's first verified signed-API attempt to its first accepted signed operation",
732            "excluded": "unverified unknown-key claims, incumbent keys, Git/Basic-auth pushes, and HTTP-only workspace provisioning",
733        },
734    })
735}
736
737/// One log author's signed identity claim. Resolution is delayed until a
738/// report is read so a hot-reloaded binding file changes the whole current
739/// projection consistently, including entries replayed before the reload.
740#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
741struct ActorEvidence {
742    key_id: Option<String>,
743    channel: String,
744}
745
746impl ActorEvidence {
747    fn from_entry(entry: &OpEntry) -> Self {
748        Self {
749            key_id: entry.author_sig.as_ref().map(|sig| sig.key_id.clone()),
750            channel: entry.channel.clone(),
751        }
752    }
753}
754
755/// Evidence available for one ref move. A directly bound signer wins. A
756/// node-signed Git move may instead be attributed to the unique bound
757/// operator whose approved review named the exact `(ref, target)` pair.
758#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
759struct RefAttribution {
760    direct: ActorEvidence,
761    approved_requesters: Vec<ActorEvidence>,
762}
763
764/// Which source a binding snapshot was built from.
765///
766/// D24 T3 attribution and channel admission read *different* sources on
767/// purpose, so the snapshot has to say which one it is rather than leaving
768/// a reader to infer it from the call site.
769#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
770enum BindingSource {
771    /// The operator-written trusted-keys file. Mutable, unsequenced, and
772    /// therefore usable for admission but not as attribution evidence.
773    #[default]
774    KeysFile,
775    /// [`View::bindings`]: sequenced `BindKey` ops, replayable from the log.
776    DurableLog,
777}
778
779impl BindingSource {
780    fn as_str(self) -> &'static str {
781        match self {
782            Self::KeysFile => "keys_file",
783            Self::DurableLog => "durable_log",
784        }
785    }
786}
787
788/// Current binding snapshot. The effective map preserves admission's
789/// existing one-name-per-actor behaviour; the deterministic records retain
790/// enough information to report duplicate cross-operator bindings as
791/// ambiguous rather than choosing whichever row happened to come last.
792#[derive(Default)]
793struct KeyBindings {
794    effective: std::collections::HashMap<ContentHash, String>,
795    operators_by_actor: BTreeMap<String, BTreeSet<String>>,
796    names_by_actor: BTreeMap<String, BTreeSet<String>>,
797    unbound_actors: BTreeSet<String>,
798    configured: bool,
799    available: bool,
800    source: BindingSource,
801}
802
803impl KeyBindings {
804    fn unavailable(configured: bool) -> Self {
805        Self {
806            configured,
807            ..Self::default()
808        }
809    }
810
811    /// Builds the attribution snapshot from the durable record instead of
812    /// the keys file.
813    ///
814    /// `population` supplies *who is trusted*, which the log cannot answer:
815    /// a `BindKey` names a key, but only the keys file says which keys the
816    /// node accepts at all. So the two compose rather than compete — the
817    /// file decides the denominator, the log decides attribution, and a
818    /// trusted key with no sequenced binding lands in `unbound_actors` and
819    /// holds `evaluation_complete` at false.
820    ///
821    /// Without a readable population there is no denominator, so the
822    /// snapshot is unavailable rather than reporting completeness over
823    /// whatever subset happens to be bound.
824    fn from_view(view: &View, population: &Self) -> Self {
825        if !population.available {
826            return Self {
827                source: BindingSource::DurableLog,
828                ..Self::unavailable(population.configured)
829            };
830        }
831        let trusted: BTreeSet<&String> = population
832            .names_by_actor
833            .keys()
834            .chain(population.unbound_actors.iter())
835            .collect();
836
837        let mut operators_by_actor: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
838        let mut names_by_actor: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
839        for (key_id, binding) in &view.bindings {
840            if !trusted.contains(key_id) {
841                continue;
842            }
843            // Revoked keys keep their operator. Attribution must survive
844            // withdrawal, or an operator could shed a concentration count
845            // by revoking the key that earned it.
846            operators_by_actor
847                .entry(key_id.clone())
848                .or_default()
849                .insert(binding.operator.clone());
850            // Only a bound channel attributes activity, mirroring the keys
851            // file, where a key with no name attributes nothing. The
852            // operator is known; which channel it speaks as is not.
853            if let Some(channel) = &binding.channel {
854                names_by_actor
855                    .entry(key_id.clone())
856                    .or_default()
857                    .insert(channel.clone());
858            }
859        }
860        let unbound_actors = trusted
861            .into_iter()
862            .filter(|actor| !names_by_actor.contains_key(*actor))
863            .cloned()
864            .collect();
865        Self {
866            // Admission is not served from this snapshot; leaving the map
867            // empty keeps it structurally unable to answer `bound_name`.
868            effective: std::collections::HashMap::new(),
869            operators_by_actor,
870            names_by_actor,
871            unbound_actors,
872            configured: population.configured,
873            available: true,
874            source: BindingSource::DurableLog,
875        }
876    }
877
878    fn from_signers(signers: &[crate::TrustedKey]) -> Self {
879        let mut effective = std::collections::HashMap::new();
880        let mut names_by_actor: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
881        let mut all_actors = BTreeSet::new();
882        for signer in signers {
883            all_actors.insert(signer.actor_id.clone());
884            if let Some(name) = &signer.name {
885                effective.insert(ContentHash::blake3(&signer.key), name.clone());
886                names_by_actor
887                    .entry(signer.actor_id.clone())
888                    .or_default()
889                    .insert(name.clone());
890            }
891        }
892        let operators_by_actor = names_by_actor
893            .iter()
894            .map(|(actor, names)| {
895                (
896                    actor.clone(),
897                    names
898                        .iter()
899                        .map(|name| reviewer_operator(name).to_string())
900                        .collect(),
901                )
902            })
903            .collect();
904        let unbound_actors = all_actors
905            .iter()
906            .filter(|actor| !names_by_actor.contains_key(*actor))
907            .cloned()
908            .collect();
909        Self {
910            effective,
911            operators_by_actor,
912            names_by_actor,
913            unbound_actors,
914            configured: true,
915            available: true,
916            source: BindingSource::KeysFile,
917        }
918    }
919
920    fn bound_name(&self, actor_id: &ContentHash) -> Option<&str> {
921        self.effective.get(actor_id).map(String::as_str)
922    }
923
924    fn resolve_evidence(&self, evidence: &ActorEvidence) -> AttributionResolution {
925        let Some(key_id) = evidence.key_id.as_ref() else {
926            return AttributionResolution::Unknown;
927        };
928        let Some(names) = self.names_by_actor.get(key_id) else {
929            return AttributionResolution::Unknown;
930        };
931        if !names.contains(&evidence.channel) {
932            return AttributionResolution::Unknown;
933        }
934        let Some(operators) = self.operators_by_actor.get(key_id) else {
935            return AttributionResolution::Unknown;
936        };
937        match operators.len() {
938            0 => AttributionResolution::Unknown,
939            1 => AttributionResolution::Operator(operators.first().expect("one operator").clone()),
940            _ => AttributionResolution::Ambiguous,
941        }
942    }
943
944    fn snapshot_hash(&self) -> Option<String> {
945        self.available.then(|| {
946            let mut records = self.names_by_actor.clone();
947            for actor in &self.unbound_actors {
948                records.entry(actor.clone()).or_default();
949            }
950            let bytes =
951                serde_json::to_vec(&records).expect("binding snapshot is always serializable");
952            ContentHash::blake3(&bytes).to_hex()
953        })
954    }
955}
956
957/// D24 T2 load evidence: objective counts folded straight from the log.
958///
959/// These answer "how much reviewer work did this cohort create, and how much
960/// of it landed" without any human calling anything slop. That matters
961/// because a classifier-based slop rate goes blind exactly when it is needed:
962/// a flood collapses adjudication coverage, and incomplete coverage must
963/// report `indeterminate`. These counts keep working during the flood, and
964/// no contributor can suppress them by declining to request a review.
965#[derive(Default)]
966struct NewActorLoadState {
967    /// Accepted entries per signing key. The unit is a *contribution
968    /// offered*, not a review: `RequestReview` is authored by the
969    /// contributor, so a review-based denominator lets an actor choose
970    /// whether to be measured.
971    submissions: BTreeMap<String, usize>,
972    /// `PostVerdict` ops per review id: reviewer rounds actually consumed.
973    /// Re-review after changes is the cost signal, so it is counted rather
974    /// than collapsed into a final verdict.
975    verdict_rounds: BTreeMap<String, usize>,
976    /// Reviews whose exact `(target_ref, target)` was observed live in
977    /// [`View::refs`]. Landing is *observed*; approval is not landing.
978    landed: BTreeSet<String>,
979}
980
981impl NewActorLoadState {
982    fn observe(&mut self, entry: &OpEntry, op: &ViewOp, view: &View) {
983        if let Some(signature) = &entry.author_sig {
984            // Only clone the key the first time it is seen. `entry()` would
985            // allocate on every accepted op, and a repeat submitter is the
986            // common case: the allocation budget caught exactly that.
987            if let Some(count) = self.submissions.get_mut(&signature.key_id) {
988                *count += 1;
989            } else {
990                self.submissions.insert(signature.key_id.clone(), 1);
991            }
992        }
993        match &op.kind {
994            OpKind::PostVerdict { id, .. } => {
995                *self.verdict_rounds.entry(id.clone()).or_default() += 1;
996            }
997            OpKind::SetRef { name, commit, .. } => {
998                for (id, review) in &view.reviews {
999                    if review.target_ref.as_deref() == Some(name)
1000                        && review.target.as_ref() == Some(commit)
1001                    {
1002                        self.landed.insert(id.clone());
1003                    }
1004                }
1005            }
1006            _ => {}
1007        }
1008    }
1009}
1010
1011#[derive(Default)]
1012struct ConcentrationState {
1013    review_requesters: BTreeMap<String, ActorEvidence>,
1014    active_branches: BTreeMap<String, RefAttribution>,
1015    ref_updates: BTreeMap<String, BTreeMap<RefAttribution, usize>>,
1016    /// T2's load evidence, folded here rather than behind a second mutex:
1017    /// it needs the same `(entry, op, view)` triple at the same point of the
1018    /// same single-writer fold, and a parallel tracker would add plumbing
1019    /// plus a second chance for the two to disagree about `as_of_seq`.
1020    new_actor_load: NewActorLoadState,
1021    as_of_seq: Option<u64>,
1022}
1023
1024impl ConcentrationState {
1025    fn observe(&mut self, entry: &OpEntry, op: &ViewOp, view: &View) {
1026        self.as_of_seq = Some(entry.seq);
1027        self.new_actor_load.observe(entry, op, view);
1028        match &op.kind {
1029            OpKind::RequestReview { id, .. } => {
1030                self.review_requesters
1031                    .insert(id.clone(), ActorEvidence::from_entry(entry));
1032            }
1033            OpKind::SetRef { name, commit, prev } => {
1034                let mut approved_requesters: Vec<_> = view
1035                    .reviews
1036                    .iter()
1037                    .filter(|(_, review)| {
1038                        review.target_ref.as_deref() == Some(name)
1039                            && review.target.as_ref() == Some(commit)
1040                            && review.approved()
1041                    })
1042                    .filter_map(|(id, _)| self.review_requesters.get(id).cloned())
1043                    .collect();
1044                approved_requesters.sort();
1045                approved_requesters.dedup();
1046                let attribution = RefAttribution {
1047                    direct: ActorEvidence::from_entry(entry),
1048                    approved_requesters,
1049                };
1050                if is_branch_ref(name) {
1051                    self.active_branches
1052                        .insert(name.clone(), attribution.clone());
1053                }
1054                if prev.is_some() {
1055                    *self
1056                        .ref_updates
1057                        .entry(name.clone())
1058                        .or_default()
1059                        .entry(attribution)
1060                        .or_default() += 1;
1061                }
1062            }
1063            OpKind::DeleteRef { name, .. } => {
1064                self.active_branches.remove(name);
1065            }
1066            _ => {}
1067        }
1068    }
1069}
1070
1071fn is_branch_ref(name: &str) -> bool {
1072    name.strip_prefix("refs/heads/")
1073        .or_else(|| name.split_once(":refs/heads/").map(|(_, branch)| branch))
1074        .is_some_and(|branch| !branch.is_empty())
1075}
1076
1077fn share_basis_points(part: usize, total: usize) -> usize {
1078    if total == 0 {
1079        return 0;
1080    }
1081    ((part as u128 * 10_000) / total as u128) as usize
1082}
1083
1084fn share_tripped(part: usize, total: usize) -> bool {
1085    total != 0 && part as u128 * 100 > total as u128 * T3_MAX_SHARE_PERCENT as u128
1086}
1087
1088enum AttributionResolution {
1089    Operator(String),
1090    Unknown,
1091    Ambiguous,
1092}
1093
1094fn resolve_attribution(
1095    attribution: &RefAttribution,
1096    bindings: &KeyBindings,
1097) -> AttributionResolution {
1098    match bindings.resolve_evidence(&attribution.direct) {
1099        AttributionResolution::Operator(operator) => {
1100            return AttributionResolution::Operator(operator);
1101        }
1102        AttributionResolution::Ambiguous => return AttributionResolution::Ambiguous,
1103        AttributionResolution::Unknown => {}
1104    }
1105    let mut operators = BTreeSet::new();
1106    let mut ambiguous = false;
1107    let mut unknown = false;
1108    for requester in &attribution.approved_requesters {
1109        match bindings.resolve_evidence(requester) {
1110            AttributionResolution::Operator(operator) => {
1111                operators.insert(operator);
1112            }
1113            AttributionResolution::Ambiguous => ambiguous = true,
1114            AttributionResolution::Unknown => unknown = true,
1115        }
1116    }
1117    if ambiguous || (unknown && !operators.is_empty()) {
1118        return AttributionResolution::Ambiguous;
1119    }
1120    if unknown {
1121        return AttributionResolution::Unknown;
1122    }
1123    match operators.len() {
1124        0 => AttributionResolution::Unknown,
1125        1 => AttributionResolution::Operator(
1126            operators.first().expect("one requester operator").clone(),
1127        ),
1128        _ => AttributionResolution::Ambiguous,
1129    }
1130}
1131
1132#[derive(Default)]
1133struct OperatorConcentration {
1134    agent_keys: usize,
1135    active_branches: usize,
1136    protected_updates: usize,
1137}
1138
1139struct ProtectedPolicySnapshot {
1140    configured: bool,
1141    available: bool,
1142    hash: Option<String>,
1143    patterns: Vec<String>,
1144}
1145
1146impl ProtectedPolicySnapshot {
1147    fn read(path: Option<&std::path::Path>) -> Self {
1148        let Some(path) = path else {
1149            return Self {
1150                configured: false,
1151                available: false,
1152                hash: None,
1153                patterns: Vec::new(),
1154            };
1155        };
1156        match std::fs::read_to_string(path) {
1157            Ok(text) => Self {
1158                configured: true,
1159                available: true,
1160                hash: Some(ContentHash::blake3(text.as_bytes()).to_hex()),
1161                patterns: text
1162                    .lines()
1163                    .map(str::trim)
1164                    .filter(|line| !line.is_empty() && !line.starts_with('#'))
1165                    .map(str::to_string)
1166                    .collect(),
1167            },
1168            Err(_) => Self {
1169                configured: true,
1170                available: false,
1171                hash: None,
1172                patterns: Vec::new(),
1173            },
1174        }
1175    }
1176
1177    fn matches(&self, name: &str) -> bool {
1178        self.patterns.iter().any(|pattern| {
1179            pattern
1180                .strip_suffix('*')
1181                .map_or(name == pattern, |prefix| name.starts_with(prefix))
1182        })
1183    }
1184}
1185
1186fn concentration_json(
1187    state: &ConcentrationState,
1188    bindings: &KeyBindings,
1189    protected: &ProtectedPolicySnapshot,
1190) -> serde_json::Value {
1191    let mut operators: BTreeMap<String, OperatorConcentration> = BTreeMap::new();
1192    let mut ambiguous_agent_keys = 0usize;
1193    // Only `KeyBindings::from_view` feeds this function, and the fold lets
1194    // one key name exactly one operator for its lifetime, so today every
1195    // set here has exactly one member and `ambiguous_agent_keys` is always
1196    // zero. The other arms are kept deliberately, not by oversight: the
1197    // keys-file shape they answer is still constructible by
1198    // `from_signers`, and if a snapshot from that source is ever routed
1199    // here, refusing to pick a winner is the behaviour that belongs. Note
1200    // this is only the *per-key* ambiguity; ambiguity across several
1201    // requester keys is live and counted in `ambiguous_active_branches`.
1202    for operator_set in bindings.operators_by_actor.values() {
1203        match operator_set.len() {
1204            0 => {}
1205            1 => {
1206                operators
1207                    .entry(operator_set.first().expect("one operator").clone())
1208                    .or_default()
1209                    .agent_keys += 1;
1210            }
1211            _ => ambiguous_agent_keys += 1,
1212        }
1213    }
1214
1215    let active_total = state.active_branches.len();
1216    let mut unknown_active = 0usize;
1217    let mut ambiguous_active = 0usize;
1218    for attribution in state.active_branches.values() {
1219        match resolve_attribution(attribution, bindings) {
1220            AttributionResolution::Operator(operator) => {
1221                operators.entry(operator).or_default().active_branches += 1;
1222            }
1223            AttributionResolution::Unknown => unknown_active += 1,
1224            AttributionResolution::Ambiguous => ambiguous_active += 1,
1225        }
1226    }
1227
1228    let mut protected_total = 0usize;
1229    let mut unknown_protected = 0usize;
1230    let mut ambiguous_protected = 0usize;
1231    if protected.available {
1232        for (name, by_attribution) in &state.ref_updates {
1233            if !protected.matches(name) {
1234                continue;
1235            }
1236            for (attribution, count) in by_attribution {
1237                protected_total += count;
1238                match resolve_attribution(attribution, bindings) {
1239                    AttributionResolution::Operator(operator) => {
1240                        operators.entry(operator).or_default().protected_updates += count;
1241                    }
1242                    AttributionResolution::Unknown => unknown_protected += count,
1243                    AttributionResolution::Ambiguous => ambiguous_protected += count,
1244                }
1245            }
1246        }
1247    }
1248
1249    let mut any_tripwire = false;
1250    let operator_rows: BTreeMap<String, serde_json::Value> = operators
1251        .into_iter()
1252        .map(|(operator, row)| {
1253            let agent_keys_tripped = row.agent_keys > T3_MAX_AGENT_KEYS_PER_OPERATOR;
1254            let active_tripped = share_tripped(row.active_branches, active_total);
1255            let protected_tripped = protected
1256                .available
1257                .then(|| share_tripped(row.protected_updates, protected_total));
1258            any_tripwire |=
1259                agent_keys_tripped || active_tripped || protected_tripped.unwrap_or(false);
1260            let protected_updates = protected.available.then_some(row.protected_updates);
1261            let protected_share = protected
1262                .available
1263                .then(|| share_basis_points(row.protected_updates, protected_total));
1264            (
1265                operator,
1266                serde_json::json!({
1267                    "agent_keys": row.agent_keys,
1268                    "active_branches": row.active_branches,
1269                    "active_branch_share_basis_points":
1270                        share_basis_points(row.active_branches, active_total),
1271                    "protected_updates": protected_updates,
1272                    "protected_update_share_basis_points": protected_share,
1273                    "tripwires": {
1274                        "agent_keys": agent_keys_tripped,
1275                        "active_branch_share": active_tripped,
1276                        "protected_update_share": protected_tripped,
1277                    },
1278                }),
1279            )
1280        })
1281        .collect();
1282
1283    let attributed_active = active_total - unknown_active - ambiguous_active;
1284    let attributed_protected = protected_total - unknown_protected - ambiguous_protected;
1285    let evaluation_complete = bindings.available
1286        && bindings.unbound_actors.is_empty()
1287        && ambiguous_agent_keys == 0
1288        && unknown_active == 0
1289        && ambiguous_active == 0
1290        && protected.available
1291        && unknown_protected == 0
1292        && ambiguous_protected == 0;
1293    let tripwire_status = if any_tripwire {
1294        "observed"
1295    } else if evaluation_complete {
1296        "not_observed"
1297    } else {
1298        "indeterminate"
1299    };
1300    serde_json::json!({
1301        "format_version": 1,
1302        "as_of_seq": state.as_of_seq,
1303        "thresholds": {
1304            "max_agent_keys_per_operator": T3_MAX_AGENT_KEYS_PER_OPERATOR,
1305            "max_share_basis_points": T3_MAX_SHARE_PERCENT * 100,
1306            "comparison": "strictly_greater_than",
1307        },
1308        "bindings": {
1309            "configured": bindings.configured,
1310            "available": bindings.available,
1311            "snapshot_hash": bindings.snapshot_hash(),
1312            // Two sources feed this block and one name for both would read
1313            // as more precise than it is. Attribution is what has to be
1314            // replayable: `durable_log` means every operator name below
1315            // came from a sequenced `BindKey`, not from whatever the keys
1316            // file happened to say at read time. The population — which
1317            // keys the node trusts at all — is not in the log and stays
1318            // with the file, which is why coverage can be incomplete even
1319            // when attribution is sound.
1320            "attribution_source": bindings.source.as_str(),
1321            "population_source": "keys_file",
1322        },
1323        "protected_policy": {
1324            "configured": protected.configured,
1325            "available": protected.available,
1326            "snapshot_hash": protected.hash.as_deref(),
1327            "classification": "current_policy",
1328        },
1329        "totals": {
1330            "bound_agent_keys": bindings.available.then_some(bindings.names_by_actor.len()),
1331            "unbound_agent_keys": bindings.available.then_some(bindings.unbound_actors.len()),
1332            "ambiguous_agent_keys": bindings.available.then_some(ambiguous_agent_keys),
1333            "active_branches": active_total,
1334            "attributed_active_branches": attributed_active,
1335            "unattributed_active_branches": unknown_active + ambiguous_active,
1336            "unknown_active_branches": unknown_active,
1337            "ambiguous_active_branches": ambiguous_active,
1338            "active_branch_attribution_coverage_basis_points":
1339                share_basis_points(attributed_active, active_total),
1340            "protected_updates": protected.available.then_some(protected_total),
1341            "attributed_protected_updates":
1342                protected.available.then_some(attributed_protected),
1343            "unattributed_protected_updates": protected
1344                .available
1345                .then_some(unknown_protected + ambiguous_protected),
1346            "unknown_protected_updates": protected.available.then_some(unknown_protected),
1347            "ambiguous_protected_updates":
1348                protected.available.then_some(ambiguous_protected),
1349            "protected_update_attribution_coverage_basis_points": protected
1350                .available
1351                .then(|| share_basis_points(attributed_protected, protected_total)),
1352        },
1353        "operators": operator_rows,
1354        "tripwire_observed": any_tripwire,
1355        "tripwire_status": tripwire_status,
1356        "evaluation_complete": evaluation_complete,
1357        "semantics": {
1358            "active_branches": "current branch refs grouped by last attributable mover, not ownership",
1359            "protected_updates": "admitted non-creation ref updates matching the current protected policy, not proof of Git publication or merge commits",
1360        },
1361    })
1362}
1363
1364/// Version of the operator-written review adjudication rows. Like the
1365/// newcomer adjudications, this file is not part of the signed op log but is
1366/// a persisted measurement input, so it carries the same explicit version.
1367const REVIEW_ADJUDICATION_FORMAT_VERSION: u64 = 1;
1368
1369/// One operator judgement about a cohort contribution.
1370///
1371/// `Invalid` and `Slop` are deliberately distinct. Invalid is good-faith and
1372/// wrong, which is what newcomers do constantly and must not by itself trip a
1373/// Sybil wire. Slop is unresponsive work that burns reviewer time, which is
1374/// what the cited curl bands are actually about. Collapsing them is how a
1375/// competent newcomer having a bad week reads as an attack.
1376#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1377enum Classification {
1378    Valid,
1379    Invalid,
1380    Slop,
1381    Unclear,
1382}
1383
1384impl Classification {
1385    fn parse(raw: &str) -> Option<Self> {
1386        match raw {
1387            "valid" => Some(Self::Valid),
1388            "invalid" => Some(Self::Invalid),
1389            "slop" => Some(Self::Slop),
1390            "unclear" => Some(Self::Unclear),
1391            _ => None,
1392        }
1393    }
1394}
1395
1396/// Reads the operator's review classifications. Keyed by review id, so a
1397/// classification survives archiving even though the verdict bulk does not.
1398///
1399/// A row naming an unknown review is an error rather than a skipped line: a
1400/// typo that silently vanished would quietly shrink measured coverage.
1401fn read_review_adjudications(
1402    path: &std::path::Path,
1403    reviews: &BTreeMap<String, choir_view::ReviewState>,
1404) -> Result<(BTreeMap<String, Classification>, String), String> {
1405    let bytes = std::fs::read(path).map_err(|e| format!("read review adjudications: {e}"))?;
1406    let snapshot_hash = ContentHash::blake3(&bytes).to_hex();
1407    let text =
1408        String::from_utf8(bytes).map_err(|_| "review adjudications are not UTF-8".to_string())?;
1409    let mut rows = BTreeMap::new();
1410    for (index, line) in text.lines().enumerate() {
1411        if line.trim().is_empty() {
1412            continue;
1413        }
1414        let value: serde_json::Value = serde_json::from_str(line)
1415            .map_err(|e| format!("review adjudication line {}: {e}", index + 1))?;
1416        if value["format_version"].as_u64() != Some(REVIEW_ADJUDICATION_FORMAT_VERSION) {
1417            return Err(format!(
1418                "review adjudication line {} has unsupported format_version",
1419                index + 1
1420            ));
1421        }
1422        let review_id = value["review_id"]
1423            .as_str()
1424            .ok_or_else(|| format!("review adjudication line {} needs review_id", index + 1))?;
1425        let classification = value["classification"]
1426            .as_str()
1427            .and_then(Classification::parse)
1428            .ok_or_else(|| {
1429                format!(
1430                    "review adjudication line {} needs classification valid|invalid|slop|unclear",
1431                    index + 1
1432                )
1433            })?;
1434        if !reviews.contains_key(review_id) {
1435            return Err(format!(
1436                "review adjudication line {} references an unknown review",
1437                index + 1
1438            ));
1439        }
1440        if rows.insert(review_id.to_string(), classification).is_some() {
1441            return Err(format!(
1442                "review adjudication line {} duplicates a review",
1443                index + 1
1444            ));
1445        }
1446    }
1447    Ok((rows, snapshot_hash))
1448}
1449
1450/// The T2 cohort: durable, post-activation, non-incumbent actor keys, read
1451/// from the same T4 audit that defines "newcomer" everywhere else.
1452///
1453/// `None` means no cohort is defined — the audit is off, unreadable, or was
1454/// never activated — which is deliberately different from an empty cohort.
1455/// An empty cohort is the honest statement "no newcomers yet"; `None` is
1456/// "this node cannot tell you who is new".
1457fn new_actor_cohort(audit: Option<&Arc<Mutex<NewcomerAudit>>>) -> Option<BTreeSet<String>> {
1458    let audit = audit?.lock().expect("newcomer audit lock");
1459    (audit.available && audit.activated).then(|| audit.by_actor.keys().cloned().collect())
1460}
1461
1462/// D24 T2 evidence for the new-actor cohort, and an explicit statement that
1463/// the tripwire itself is **not evaluable here**.
1464///
1465/// T2 asks for an invalid/slop rate. Choir persists no such judgement.
1466/// [`choir_view::Verdict::Approve`] means "may land" and
1467/// [`choir_view::Verdict::RequestChanges`] means "needs work", the latter is
1468/// overwritable by the former, and neither proves a contribution valid,
1469/// invalid, or slop. Relabelling one as "slop" would manufacture evidence the
1470/// model does not contain.
1471///
1472/// The unit is a **contribution offered**, not a completed review. A review is
1473/// opened by its own contributor, so a review-shaped denominator lets an actor
1474/// choose whether to be measured: submit a hundred slop changes, request
1475/// review on the three good ones, and a review-based rate reads zero. Reviews
1476/// remain a reported sub-metric.
1477///
1478/// Three counting rules exist to close laundering paths rather than to be
1479/// tidy. An archived review with no classification can never enter the
1480/// numerator or the denominator, because `--review-lapse-after-secs` turns an
1481/// unanswered review into `Archived { approved: false }` on a wall clock and
1482/// that must not become evidence. Pending reviews are reported, never counted.
1483/// And `load` needs no adjudication at all, so it keeps measuring during the
1484/// flood in which a classifier's coverage would collapse to `indeterminate`.
1485///
1486/// Requester attribution is joined from the `RequestReview` log entry's author
1487/// signature, never from the rendered review: archiving discards reviewer and
1488/// verdict detail, so a rendered row cannot supply historical evidence.
1489/// A `RequestReview` author is also not proven to have authored the commit
1490/// under review; that limitation ships in the response.
1491fn new_actor_review_outcomes_json(
1492    state: &ConcentrationState,
1493    view: &View,
1494    cohort: Option<&BTreeSet<String>>,
1495    adjudications_path: Option<&std::path::Path>,
1496) -> serde_json::Value {
1497    let adjudications =
1498        adjudications_path.map(|path| read_review_adjudications(path, &view.reviews));
1499    let (classifications, snapshot_hash, adjudications_available, adjudications_error) =
1500        match &adjudications {
1501            None => (BTreeMap::new(), None, false, None),
1502            Some(Ok((rows, hash))) => (rows.clone(), Some(hash.clone()), true, None),
1503            Some(Err(error)) => (BTreeMap::new(), None, false, Some(error.clone())),
1504        };
1505
1506    let mut unknown_requester = 0usize;
1507    let mut unsigned_author = 0usize;
1508    let mut signed = 0usize;
1509    let mut attributed = 0usize;
1510    let mut approved = 0usize;
1511    let mut request_changes = 0usize;
1512    let mut pending = 0usize;
1513    let mut archived_classified = 0usize;
1514    let mut archived_evidence_lost = 0usize;
1515    let mut slashed = 0usize;
1516    let mut eligible = 0usize;
1517    let mut classified: BTreeMap<&'static str, usize> = BTreeMap::new();
1518    let mut review_rounds = 0usize;
1519    let mut review_rounds_unlanded = 0usize;
1520    let mut landed = 0usize;
1521    for (id, review) in &view.reviews {
1522        let Some(evidence) = state.review_requesters.get(id) else {
1523            unknown_requester += 1;
1524            continue;
1525        };
1526        let Some(key_id) = evidence.key_id.as_deref() else {
1527            unsigned_author += 1;
1528            continue;
1529        };
1530        signed += 1;
1531        let Some(cohort) = cohort else { continue };
1532        if !cohort.contains(key_id) {
1533            continue;
1534        }
1535        attributed += 1;
1536        if review.re_review_required() {
1537            slashed += 1;
1538        }
1539        let rounds = state
1540            .new_actor_load
1541            .verdict_rounds
1542            .get(id)
1543            .copied()
1544            .unwrap_or_default();
1545        review_rounds += rounds;
1546        if state.new_actor_load.landed.contains(id) {
1547            landed += 1;
1548        } else {
1549            review_rounds_unlanded += rounds;
1550        }
1551        let classification = classifications.get(id).copied();
1552        let counts_toward_rate = if matches!(review.status, ReviewStatus::Archived { .. }) {
1553            // Archiving destroys the verdicts an adjudicator needs, and a
1554            // lapse archives a review nobody ever answered. Without a
1555            // standing classification such a row is evidence of nothing.
1556            if classification.is_some() {
1557                archived_classified += 1;
1558                true
1559            } else {
1560                archived_evidence_lost += 1;
1561                false
1562            }
1563        } else if !review.complete() {
1564            pending += 1;
1565            false
1566        } else if review
1567            .verdicts
1568            .values()
1569            .any(|answer| answer.verdict == Verdict::RequestChanges)
1570        {
1571            request_changes += 1;
1572            true
1573        } else {
1574            approved += 1;
1575            true
1576        };
1577        if counts_toward_rate {
1578            eligible += 1;
1579            if let Some(classification) = classification {
1580                *classified
1581                    .entry(match classification {
1582                        Classification::Valid => "valid",
1583                        Classification::Invalid => "invalid",
1584                        Classification::Slop => "slop",
1585                        Classification::Unclear => "unclear",
1586                    })
1587                    .or_default() += 1;
1588            }
1589        }
1590    }
1591
1592    let available = cohort.is_some();
1593    let submissions: usize = cohort
1594        .map(|cohort| {
1595            cohort
1596                .iter()
1597                .filter_map(|key| state.new_actor_load.submissions.get(key))
1598                .sum()
1599        })
1600        .unwrap_or_default();
1601    let adjudicated: usize = classified.values().sum();
1602    let classified_rows: BTreeMap<String, usize> = ["valid", "invalid", "slop", "unclear"]
1603        .into_iter()
1604        .map(|name| {
1605            (
1606                name.to_string(),
1607                classified.get(name).copied().unwrap_or_default(),
1608            )
1609        })
1610        .collect();
1611    serde_json::json!({
1612        "format_version": 2,
1613        "as_of_seq": state.as_of_seq,
1614        "unit": "contribution_offered",
1615        "cohort": {
1616            "definition": "post_activation_non_incumbent_actor_keys",
1617            "source": "newcomer_harm audit",
1618            "available": available,
1619            "actor_keys": cohort.map(BTreeSet::len),
1620        },
1621        "policy": {
1622            "graduation": null,
1623            "trailing_window": null,
1624            "tripwire_subject": null,
1625            "sampling": "census",
1626            "status": "unset_pending_operator_decision",
1627        },
1628        "load": available.then(|| serde_json::json!({
1629            "accepted_operations": submissions,
1630            "reviews_requested": attributed,
1631            "review_rounds": review_rounds,
1632            "review_rounds_on_unlanded": review_rounds_unlanded,
1633            "reviews_landed": landed,
1634            "reviews_never_landed": attributed - landed,
1635        })),
1636        "declared_tripwire": {
1637            "max_invalid_or_slop_percent": T2_MAX_INVALID_OR_SLOP_PERCENT,
1638            "min_valid_percent": T2_MIN_VALID_PERCENT,
1639            "evaluable": false,
1640            "blocked_on": "cohort exit, observation window and tripwire subject are unset, and census adjudication cannot survive a flood",
1641        },
1642        "totals": {
1643            "reviews": view.reviews.len(),
1644            "attributed_to_cohort": available.then_some(attributed),
1645            "excluded_outside_cohort": available.then(|| signed - attributed),
1646            "excluded_unsigned_author": unsigned_author,
1647            "excluded_unknown_requester": unknown_requester,
1648        },
1649        "review_outcomes": available.then(|| serde_json::json!({
1650            "approved": approved,
1651            "request_changes": request_changes,
1652            "pending": pending,
1653            "archived_classified": archived_classified,
1654            "archived_evidence_lost": archived_evidence_lost,
1655            "re_review_required": slashed,
1656        })),
1657        "adjudication": {
1658            "configured": adjudications_path.is_some(),
1659            "available": adjudications_available,
1660            "snapshot_hash": snapshot_hash,
1661            "error": adjudications_error,
1662            "eligible": available.then_some(eligible),
1663            "adjudicated": available.then_some(adjudicated),
1664            "coverage_basis_points": available.then(|| share_basis_points(adjudicated, eligible)),
1665            "classified": available.then_some(classified_rows),
1666            "independent": false,
1667        },
1668        "tripwire_observed": null,
1669        "tripwire_status": "indeterminate",
1670        "evaluation_complete": false,
1671        "semantics": {
1672            "unit": "a contribution offered by a cohort key; reviews are a sub-metric because the contributor opens their own review and can decline to",
1673            "accepted_operations": "every accepted signed operation authored by a cohort key, review participation included; counted because work that never reaches review is invisible to a review-shaped denominator",
1674            "load": "objective and adjudication-free, so it keeps measuring when a classifier's coverage collapses; a rate alone cannot see a flood",
1675            "approved": "live, every listed reviewer answered, and no verdict is RequestChanges — not a validity judgement",
1676            "request_changes": "live, every listed reviewer answered, and at least one standing verdict is RequestChanges — 'needs work', overwritable, not slop",
1677            "pending": "live and unassigned or not yet answered by every listed reviewer; reported, never counted",
1678            "archived_evidence_lost": "archived with no standing classification; a lapse archives an unanswered review on a wall clock, so it is excluded from both numerator and denominator",
1679            "re_review_required": "overlaps the buckets above; counts cohort reviews carrying a retroactive approval slash",
1680            "attribution": "the RequestReview log entry's signing key; that author is not proven to have authored the commit under review",
1681            "independent": "a single-operator node classifies contributions its own reviewers approved; this is self-adjudication, not an independent judgement",
1682            "indeterminate": "structural: the cohort has no exit, the window and tripwire subject are unset, and census adjudication goes blind in the flood it should detect",
1683        },
1684    })
1685}
1686
1687/// Deterministic read-time measurement of the complete authoritative view.
1688///
1689/// The serialized total deliberately contains only the four sections owned by
1690/// `View`. Runtime projections are excluded so adding this report cannot make
1691/// its own byte count grow recursively, and adding another projection later
1692/// cannot rewrite the historical meaning of the measurement.
1693fn view_growth_json(
1694    counts: serde_json::Value,
1695    workspaces: &serde_json::Value,
1696    refs: &serde_json::Value,
1697    reviews: &serde_json::Value,
1698    provenance: &serde_json::Value,
1699    measured: &[(&str, &serde_json::Value)],
1700    as_of_seq: Option<u64>,
1701) -> serde_json::Value {
1702    let serialized_bytes = |value: &serde_json::Value| {
1703        serde_json::to_vec(value)
1704            .expect("materialized view JSON is always serializable")
1705            .len()
1706    };
1707    // `measured` is deliberately absent from the total below. The four
1708    // sections passed by name are the authoritative view and
1709    // `total_authoritative_view` is a tracked series; folding a fifth
1710    // section in would move every past reading and destroy comparability.
1711    // They are still *measured* — see the sibling byte counts — because a
1712    // map nobody counts is how a view grows without anyone noticing.
1713    //
1714    // A slice rather than one argument each, so the next section that
1715    // has to be watched without being totalled costs a call site and not
1716    // a signature. The two shapes are not interchangeable: an argument
1717    // added to the authoritative four changes what a historical number
1718    // means, and this one cannot.
1719    let authoritative = serde_json::json!({
1720        "workspaces": workspaces,
1721        "refs": refs,
1722        "reviews": reviews,
1723        "provenance": provenance,
1724    });
1725    let mut bytes = serde_json::json!({
1726        "workspaces": serialized_bytes(workspaces),
1727        "refs": serialized_bytes(refs),
1728        "reviews": serialized_bytes(reviews),
1729        "provenance": serialized_bytes(provenance),
1730        "total_authoritative_view": serialized_bytes(&authoritative),
1731    });
1732    for (name, value) in measured {
1733        bytes[*name] = serde_json::json!(serialized_bytes(value));
1734    }
1735    serde_json::json!({
1736        "format_version": 1,
1737        "as_of_seq": as_of_seq,
1738        "counts": counts,
1739        "serialized_bytes": bytes,
1740    })
1741}
1742
1743fn view_growth_counts(view: &View) -> serde_json::Value {
1744    let live_reviews = view
1745        .reviews
1746        .values()
1747        .filter(|review| matches!(review.status, ReviewStatus::Live))
1748        .count();
1749    let provenance_records = view.provenance.values().map(BTreeMap::len).sum::<usize>();
1750    // Revoked bindings are counted, never subtracted: the row survives
1751    // revocation, so a `bindings` count that dropped on revoke would
1752    // understate the map it is meant to size.
1753    let revoked_bindings = view
1754        .bindings
1755        .values()
1756        .filter(|binding| binding.is_revoked())
1757        .count();
1758    // Edges, not subjects: the outer map is what paging bounds, and the
1759    // inner one is where the graph actually grows. Counting rows would
1760    // report one number for an operator with a single vouch and for one
1761    // every operator on the node stands behind.
1762    let vouch_edges = view.vouches.values().map(BTreeMap::len).sum::<usize>();
1763    // One row per witness and no more, which is the claim worth
1764    // measuring: this section is bounded by the witness population, not
1765    // by how many snapshots the node has taken.
1766    let witnesses = view.witnessed.len();
1767    let witnesses_current = view.latest_snapshot.as_ref().map_or(0, |latest| {
1768        let id = latest.id();
1769        view.witnessed
1770            .values()
1771            .filter(|state| state.snapshot == id)
1772            .count()
1773    });
1774    serde_json::json!({
1775        "workspaces": view.workspaces.len(),
1776        "refs": view.refs.len(),
1777        "reviews": view.reviews.len(),
1778        "live_reviews": live_reviews,
1779        "archived_reviews": view.reviews.len() - live_reviews,
1780        "provenance_subjects": view.provenance.len(),
1781        "provenance_records": provenance_records,
1782        "bindings": view.bindings.len(),
1783        "revoked_bindings": revoked_bindings,
1784        "vouch_subjects": view.vouches.len(),
1785        "vouch_edges": vouch_edges,
1786        "witnesses": witnesses,
1787        "witnesses_current": witnesses_current,
1788    })
1789}
1790
1791/// Nanosecond clock reading, as a nonzero xorshift seed.
1792fn seed_from_clock() -> u64 {
1793    std::time::SystemTime::now()
1794        .duration_since(std::time::UNIX_EPOCH)
1795        .map_or(0x9e37_79b9_7f4a_7c15, |d| d.as_nanos() as u64)
1796        | 1
1797}
1798
1799/// FNV-1a, so two reviews drawn in the same nanosecond still diverge.
1800fn fnv1a(bytes: &[u8]) -> u64 {
1801    bytes.iter().fold(0xcbf2_9ce4_8422_2325, |h, b| {
1802        (h ^ u64::from(*b)).wrapping_mul(0x100_0000_01b3)
1803    })
1804}
1805
1806impl LogWindow {
1807    fn push(&mut self, entry: OpEntry, hash: ContentHash) {
1808        // `signing_hash` covers only (channel, payload); `content_hash`
1809        // would serialize the whole entry, and this runs per admitted op
1810        // -- which is why `hash` is passed in by the sequencer that
1811        // already computed it rather than derived here.
1812        self.by_signing.insert(entry.signing_hash(), entry.seq);
1813        self.by_hash.insert(hash, entry.seq);
1814        self.entries.push_back(entry);
1815        self.trim();
1816    }
1817
1818    /// Whether `head` is an entry still inside the window, which is what
1819    /// makes an op signed against it admissible.
1820    fn holds_head(&self, head: &ContentHash) -> bool {
1821        self.by_hash.contains_key(head)
1822    }
1823
1824    /// The head a client should sign its next scope against.
1825    ///
1826    /// Derived rather than stored: keeping it would mean cloning a hash
1827    /// on every admitted op to serve a value only read by `/api/view`
1828    /// and by the ops the daemon signs itself. `None` for an empty
1829    /// window, which is also a window that would admit no head.
1830    fn head_hash(&self) -> Option<ContentHash> {
1831        self.entries.back().map(OpEntry::content_hash)
1832    }
1833
1834    /// The seq an identical submission already landed as, if any. The
1835    /// caller has usually just computed the signing hash for the
1836    /// signature check, so it is taken rather than recomputed.
1837    fn seq_for_signing(&self, signing: &ContentHash) -> Option<u64> {
1838        self.by_signing.get(signing).copied()
1839    }
1840
1841    /// The `(seq, hash)` an identical submission already landed as.
1842    ///
1843    /// Keyed on the **signing** hash, not the entry hash: an entry's hash
1844    /// covers `seq` and `parent`, which the sequencer assigns after the
1845    /// author signs, so a resubmitted identical op hashes differently
1846    /// every time and could never match. `signing_hash(channel,
1847    /// payload)` is position-independent by construction, which is
1848    /// exactly the identity "the same submission" needs.
1849    fn already_applied(&self, channel: &str, payload: &[u8]) -> Option<(u64, ContentHash)> {
1850        let seq = *self
1851            .by_signing
1852            .get(&choir_oplog::signing_hash(channel, payload))?;
1853        // Hash the entry only on a hit, which is a client retry rather
1854        // than the common path.
1855        let entry = self.entries.get(seq.checked_sub(self.base)? as usize)?;
1856        Some((seq, entry.content_hash()))
1857    }
1858
1859    /// Drops the oldest entries until the window fits its cap, advancing
1860    /// `base` so `/api/log?from=` keeps absolute sequence semantics.
1861    fn trim(&mut self) {
1862        while self.entries.len() > self.cap {
1863            if let Some(dropped) = self.entries.pop_front() {
1864                // Evict from the index with the entry, or the map becomes
1865                // the unbounded thing the window exists to avoid.
1866                self.by_signing.remove(&dropped.signing_hash());
1867                // The dropped entry's own hash is the new front's
1868                // `parent`, so its eviction costs a lookup instead of a
1869                // re-serialization. An emptied window holds no heads at
1870                // all, which is the only case with no successor to ask.
1871                match self.entries.front().and_then(|e| e.parent.as_ref()) {
1872                    Some(parent) => {
1873                        self.by_hash.remove(parent);
1874                    }
1875                    None => self.by_hash.clear(),
1876                }
1877            }
1878            self.base += 1;
1879        }
1880    }
1881}
1882
1883/// Replays the platform's runtime projections in one pass. Concentration
1884/// attribution needs the entry author as well as the typed payload, while
1885/// review retention additionally needs request order. Neither belongs in
1886/// the persisted `ViewOp` format, and neither justifies a second log scan.
1887fn materialize_platform_state(
1888    log: &dyn OpLog,
1889    retention_config: Option<ReviewRetention>,
1890) -> Result<
1891    (
1892        View,
1893        Option<ReviewRetentionState>,
1894        ConcentrationState,
1895        crate::quota::WorkspaceTally,
1896    ),
1897    choir_view::ViewError,
1898> {
1899    let mut view = View::default();
1900    let mut retention = retention_config.map(ReviewRetentionState::new);
1901    let mut concentration = ConcentrationState::default();
1902    // D37. Folded in this loop rather than in a pass of its own, which is
1903    // the whole reason a per-user workspace ceiling needs no persisted
1904    // file: the log a restart already replays is the tally's storage.
1905    let mut workspace_tally = crate::quota::WorkspaceTally::default();
1906    // Stored entries have no timestamp. Giving every pre-existing live
1907    // review `now` starts a fresh grace period after restart, which can
1908    // delay an incomplete-review lapse but can never trigger one early.
1909    let observed_at =
1910        retention_config.and_then(|config| config.lapse_after.map(|_| Instant::now()));
1911    for seq in 0..log.len() {
1912        let entry = log.get(seq).expect("seq < len");
1913        let op = ViewOp::from_payload(&entry.payload)?;
1914        view.apply(&op)?;
1915        concentration.observe(&entry, &op, &view);
1916        workspace_tally.observe(&entry, &op);
1917        if let Some(retention) = &mut retention {
1918            retention.observe(&op, observed_at);
1919        }
1920    }
1921    Ok((view, retention, concentration, workspace_tally))
1922}
1923
1924/// Verify author signature, then CAS against the shared view. Runs on
1925/// the sequencer's writer thread; API readers share the view mutex.
1926/// The op variant's name, from its own externally-tagged serialization
1927/// rather than a hand-written match: a variant added later journals
1928/// correctly without anyone remembering to extend a table here.
1929fn op_type_name(op: &ViewOp) -> Option<String> {
1930    serde_json::to_value(&op.kind)
1931        .ok()
1932        .and_then(|v| v.as_object().and_then(|o| o.keys().next().cloned()))
1933}
1934
1935/// A journal slot the builder fills after the writer thread is running.
1936///
1937/// [`Sequencer::spawn_with_journal`] takes its journal by value at
1938/// construction, but every `with_*` builder runs afterwards, so the
1939/// value handed to the writer has to be a slot rather than a journal.
1940/// Same shape as the hot-reloadable config beside it.
1941///
1942/// `on` is separate from the lock on purpose: [`journal::Journal::enabled`]
1943/// is consulted once per op on the writer thread, and a node with no
1944/// journal should pay one relaxed atomic load, not a mutex acquisition.
1945#[derive(Clone, Default)]
1946struct SharedJournal {
1947    inner: Arc<Mutex<Option<Box<dyn journal::Journal>>>>,
1948    on: Arc<std::sync::atomic::AtomicBool>,
1949}
1950
1951impl SharedJournal {
1952    /// Installs `journal`, and switches recording on.
1953    fn install(&self, journal: Box<dyn journal::Journal>) {
1954        if let Ok(mut slot) = self.inner.lock() {
1955            *slot = Some(journal);
1956            self.on.store(true, std::sync::atomic::Ordering::Relaxed);
1957        }
1958    }
1959}
1960
1961impl journal::Journal for SharedJournal {
1962    fn record(&self, event: journal::Event) {
1963        if !self.enabled() {
1964            return;
1965        }
1966        if let Ok(slot) = self.inner.lock() {
1967            if let Some(journal) = slot.as_ref() {
1968                journal.record(event);
1969            }
1970        }
1971    }
1972
1973    fn enabled(&self) -> bool {
1974        self.on.load(std::sync::atomic::Ordering::Relaxed)
1975    }
1976}
1977
1978struct ChoirPolicy {
1979    /// Where this policy reports CAS failures. The sequencer cannot:
1980    /// a refusal reaches it as an opaque string, so contention and
1981    /// nonsense look identical from there.
1982    journal: SharedJournal,
1983    /// What the last `check` identified, handed to the journal through
1984    /// [`SubmitPolicy::subject`] rather than derived a second time.
1985    subject: (Option<String>, Option<String>),
1986    registry: Registry,
1987    view: Arc<Mutex<View>>,
1988    entries: Arc<Mutex<LogWindow>>,
1989    /// When set, the trusted-keys file is checked before every signature;
1990    /// registering or removing a key takes effect on its next submission.
1991    keys_file: Option<std::path::PathBuf>,
1992    keys_mtime: Option<std::time::SystemTime>,
1993    node_pub: Vec<u8>,
1994    /// The node key's actor id: the only author allowed to assign reviewers.
1995    node_id: ContentHash,
1996    /// When set, a `RequestReview` may not name its own reviewers —
1997    /// every review must go through the node's draw (D24 layer 5).
1998    /// Shared with [`Platform`] so the switch is one value, not two.
1999    require_assignment: Arc<std::sync::atomic::AtomicBool>,
2000    /// Operator's protected-ref list: the same switch, but conditioned on
2001    /// where the review proposes to land. Shared with [`Platform`].
2002    protected_refs: Arc<Mutex<Option<std::path::PathBuf>>>,
2003    /// The operator's ACL file, read here for [`crate::acl::Level::Own`]
2004    /// grants alone (D42).
2005    ///
2006    /// Deliberately the *file*, not the merged table the HTTP layer
2007    /// enforces. Self-service (D36) contributes grants to that merge, and
2008    /// ownership authorizes a landing on a protected ref, so reading the
2009    /// merge would make ownership reachable by whatever self-service can
2010    /// issue. Reading the file makes "only the operator grants ownership"
2011    /// true by construction rather than by auditing the issuer.
2012    acl_file: Arc<Mutex<Option<std::path::PathBuf>>>,
2013    /// When set, a protected ref only moves to a commit some approved
2014    /// review already named — the landing half of the gate.
2015    require_review: Arc<std::sync::atomic::AtomicBool>,
2016    /// When set, every submission must carry a [`choir_view::OpScope`]
2017    /// naming this node and a head still in the window. Shared with
2018    /// [`Platform`], like the other gates.
2019    require_scope: Arc<std::sync::atomic::AtomicBool>,
2020    /// Actor id → the channel name that key is bound to,
2021    /// for keys whose trusted-keys line carries a name. Keys absent from
2022    /// this map are unconstrained, which is what every key was before the
2023    /// name column existed.
2024    ///
2025    /// Shared with [`Platform`], because a *tightening* must not wait for
2026    /// an unrelated event: the accept loop refreshes it on mtime change,
2027    /// while submission admission refreshes the signature registry too.
2028    key_names: Arc<Mutex<KeyBindings>>,
2029    /// Entry-author-aware runtime projection for D24 T3. Updated on the
2030    /// writer thread immediately after the ordinary view fold.
2031    concentration: Arc<Mutex<ConcentrationState>>,
2032    /// Present only when the operator enabled review retention. Updated
2033    /// after the view fold on the same writer thread, so its FIFO order
2034    /// matches the sequencer order exactly.
2035    review_retention: Option<Arc<Mutex<ReviewRetentionState>>>,
2036    /// The credential store, when the node has one (D39). Shared with
2037    /// [`Platform`] as a slot rather than taken at construction because
2038    /// the two are enabled independently and in either order; a node
2039    /// without accounts simply has no passkey authors.
2040    passkeys: Arc<Mutex<Option<Arc<crate::accounts::Accounts>>>>,
2041    /// Webhook delivery handle (D32), shared with [`Platform`] so the
2042    /// operator's `--hooks-file` can be attached after the writer thread
2043    /// is already running. Offering an event to it never blocks: see
2044    /// [`crate::hooks`] for why invariant 5 forbids anything else here.
2045    hooks: Arc<Mutex<Option<crate::hooks::Hooks>>>,
2046    /// Which channel holds which workspace (D37). Folded here for the
2047    /// reason `concentration` is: it needs the entry's channel as well as
2048    /// the typed payload, and it must advance in the same single-writer
2049    /// step as the view or the two can disagree about what exists.
2050    workspace_tally: Arc<Mutex<crate::quota::WorkspaceTally>>,
2051}
2052
2053impl ChoirPolicy {
2054    /// Rebuilds the registry from the keys file iff its mtime changed
2055    /// since the last (re)load. Returns whether a reload happened.
2056    fn reload_keys(&mut self) -> bool {
2057        let Some(path) = &self.keys_file else {
2058            return false;
2059        };
2060        let mtime = std::fs::metadata(path).and_then(|m| m.modified()).ok();
2061        if mtime.is_none() || mtime == self.keys_mtime {
2062            return false;
2063        }
2064        // Same parser startup uses, so the two cannot drift — and a
2065        // malformed file keeps the previous registry *and* the previous
2066        // name bindings rather than half-applying either.
2067        let Ok(signers) = crate::parse_keys_file(path) else {
2068            self.keys_mtime = mtime;
2069            return false;
2070        };
2071        let mut registry = Registry::new();
2072        if let Ok(node_pub) = <[u8; 32]>::try_from(self.node_pub.as_slice()) {
2073            registry.register(&node_pub).ok();
2074        }
2075        for signer in &signers {
2076            registry.register(&signer.key).ok();
2077        }
2078        self.registry = registry;
2079        *self.key_names.lock().expect("key names lock") = KeyBindings::from_signers(&signers);
2080        self.keys_mtime = mtime;
2081        true
2082    }
2083
2084    /// Enforces the name a key is bound to, for ops whose submission
2085    /// channel is an identity claim rather than a workspace name.
2086    ///
2087    /// Scope is deliberately narrow. `sub.channel` is the
2088    /// signature-covered attribution/identity channel, not a workspace id.
2089    /// The v1 wire field was named `workspace`; the daemon also uses
2090    /// synthetic push-attribution channels
2091    /// (`git/<user>`, `key/<principal>`). Binding every channel would
2092    /// break workspace provisioning and every git-derived op.
2093    ///
2094    /// A key with no bound name is unconstrained — exactly its behaviour
2095    /// before the name column existed — so this cannot break a running
2096    /// node, and the operator opts in one line at a time.
2097    fn channel_is_owned(&self, actor_id: &ContentHash, channel: &str) -> Result<(), String> {
2098        let names = self.key_names.lock().expect("key names lock");
2099        match names.bound_name(actor_id) {
2100            Some(bound) if bound != channel => Err(Rejection::new(
2101                Code::ChannelNotOwned,
2102                "this key is bound to a different channel",
2103                "submit on the channel your key is bound to, or ask the operator to bind a \
2104                 key to the channel you want",
2105            )
2106            .with_states(Some(bound.to_string()), Some(channel.to_string()))
2107            .encode()),
2108            _ => Ok(()),
2109        }
2110    }
2111
2112    /// Whether `name` matches the operator's protected-ref list. One
2113    /// pattern per line, `#` comments allowed, exact match or a single
2114    /// trailing `*` prefix glob (`repo.git:refs/heads/release/*`).
2115    ///
2116    /// Read per call rather than cached, so editing the list takes effect
2117    /// with no restart. A `RequestReview` always reaches here; a
2118    /// `SetRef`/`DeleteRef` only under `--require-review`, which is the
2119    /// mode that also puts a file read on the git-push path. If push
2120    /// throughput ever notices, an mtime-stat cache is the fix — measure
2121    /// before adding one.
2122    ///
2123    /// Fails **closed**: an unreadable list refuses the op instead of
2124    /// quietly demoting a gate to an advisory.
2125    fn ref_is_protected(&self, name: &str) -> Result<bool, String> {
2126        let guard = self.protected_refs.lock().expect("protected refs lock");
2127        let Some(path) = guard.as_ref() else {
2128            return Ok(false);
2129        };
2130        let text = std::fs::read_to_string(path).map_err(|e| {
2131            Rejection::new(
2132                Code::PolicyUnavailable,
2133                format!("protected-ref list unreadable: {e}"),
2134                "this is an operator problem, not a client one: the gate fails closed \
2135                     rather than guessing. Retry after the operator restores the file",
2136            )
2137            .encode()
2138        })?;
2139        Ok(text.lines().map(str::trim).any(|p| {
2140            if p.is_empty() || p.starts_with('#') {
2141                return false;
2142            }
2143            match p.strip_suffix('*') {
2144                Some(prefix) => name.starts_with(prefix),
2145                None => name == p,
2146            }
2147        }))
2148    }
2149
2150    /// The greatest capped approval weight of any approved review that
2151    /// named exactly this `(ref, commit)` pair as where it wanted to land.
2152    ///
2153    /// Both halves matter. Matching only the commit would let an approval
2154    /// for a scratch branch land the same commit on `main`; matching only
2155    /// the ref would let any approved review authorize any later commit.
2156    fn approval_weight_for(&self, name: &str, commit: &ContentHash) -> usize {
2157        self.view
2158            .lock()
2159            .expect("view lock")
2160            .reviews
2161            .values()
2162            .filter(|r| {
2163                r.target_ref.as_deref() == Some(name)
2164                    && r.target.as_ref() == Some(commit)
2165                    && r.approved()
2166            })
2167            .map(choir_view::ReviewState::approval_weight)
2168            .max()
2169            .unwrap_or(0)
2170    }
2171
2172    /// The operator's ACL, parsed, or `None` when no file is configured.
2173    ///
2174    /// Read per call rather than cached, so granting ownership takes
2175    /// effect with no restart — the same trade [`Self::ref_is_protected`]
2176    /// makes, and paid on the same narrow path, since both are reached
2177    /// only for a landing that is already gated.
2178    ///
2179    /// Fails **closed**. An unreadable or malformed file refuses the
2180    /// landing rather than concluding there are no owners, because "no
2181    /// owners" is precisely the branch that falls back to the weaker
2182    /// rule: a gate that loses its policy file must not quietly demote
2183    /// itself to the policy it was configured to replace.
2184    fn acl_now(&self) -> Result<Option<crate::acl::Effective>, String> {
2185        let guard = self.acl_file.lock().expect("acl file lock");
2186        let Some(path) = guard.as_ref() else {
2187            return Ok(None);
2188        };
2189        let text = std::fs::read_to_string(path).map_err(|e| {
2190            Rejection::new(
2191                Code::PolicyUnavailable,
2192                format!("acl file unreadable: {e}"),
2193                "this is an operator problem, not a client one: the ownership gate fails \
2194                 closed rather than guessing. Retry once the operator restores the file",
2195            )
2196            .encode()
2197        })?;
2198        crate::acl::Acl::parse(&text)
2199            .map(|table| Some(table.at(crate::accounts::now_secs())))
2200            .map_err(|e| {
2201                Rejection::new(
2202                    Code::PolicyUnavailable,
2203                    format!("acl file unparseable: {e}"),
2204                    "ask the operator to repair the ACL file; the ownership gate refuses rather \
2205                 than enforcing a table it only partly understands",
2206                )
2207                .encode()
2208            })
2209    }
2210
2211    /// The ACL user this submission acts as, when one can be established
2212    /// from evidence rather than from a claim (D42).
2213    ///
2214    /// Three populations reach admission and only two of them resolve:
2215    ///
2216    /// - **A transport push.** `provenance` is a label only the node may
2217    ///   apply — refused for any other signer earlier in `check` — so a
2218    ///   [`Provenance::PushTransport`] op carries a channel the node
2219    ///   synthesized from the authenticated HTTP user. `git/<user>` is
2220    ///   therefore evidence about who pushed, not a claim by them.
2221    /// - **An author-signed op.** The identity is the verified signing
2222    ///   key, and the operator's trusted-keys name column is what binds a
2223    ///   key to a name. An **unbound key resolves to nobody**: unbound
2224    ///   keys are deliberately unconstrained in their choice of channel,
2225    ///   so honouring the channel there would let any trusted key call
2226    ///   itself an owner.
2227    /// - **A certified push.** The channel is `key/<signer>`, naming the
2228    ///   push certificate's key rather than an ACL user, and the
2229    ///   authenticated user is not carried into the op. Resolves to
2230    ///   nobody, so a `git push --signed` cannot assert ownership; see
2231    ///   the rejection in [`Self::landing_is_authorized`], which says so
2232    ///   rather than reporting a generic refusal.
2233    fn acting_user(&self, sub: &Submission, op: &ViewOp, actor_id: &ContentHash) -> Option<String> {
2234        match op.provenance {
2235            Some(Provenance::PushTransport) => {
2236                sub.channel.strip_prefix("git/").map(ToString::to_string)
2237            }
2238            Some(Provenance::PushCertified) => None,
2239            None => {
2240                let names = self.key_names.lock().expect("key names lock");
2241                names.bound_name(actor_id).map(ToString::to_string)
2242            }
2243        }
2244    }
2245
2246    /// The landing half of the protected-ref gate: may this submission
2247    /// move this protected ref to this commit?
2248    ///
2249    /// Two rules, and which one applies is a property of the *repository*
2250    /// rather than of the actor (D42):
2251    ///
2252    /// - **Somebody owns the repository.** An owner's assent is necessary
2253    ///   and sufficient. No quantity of non-owner approval substitutes,
2254    ///   and no second opinion is required alongside it.
2255    /// - **Nobody owns it.** The pre-D42 rule stands untouched: approval
2256    ///   weight `REQUIRED_APPROVAL_WEIGHT`, from that many distinct
2257    ///   operators.
2258    ///
2259    /// So a node with no `--acl-file`, or one whose ACL grants nobody
2260    /// `own`, behaves exactly as every node did before D42. Ownership is
2261    /// opt-in per repository, and opting one repository in leaves every
2262    /// other one alone.
2263    ///
2264    /// **Returns the authorization rather than a bare yes** (D43). The
2265    /// gate is evaluated at apply time against files that are not in the
2266    /// log, so this function is the only place that knows why a landing
2267    /// was allowed. A [`OpKind::Submit`] records that answer, and the
2268    /// brief's constraint is that the record must come out of the
2269    /// evaluation that admitted the op and not a second pass that
2270    /// recomputes it — two computations can disagree, and a field that
2271    /// can disagree with the decision it describes is decoration. So
2272    /// there is one function, the `SetRef` path discards its value, and
2273    /// the `Submit` path writes it down.
2274    ///
2275    /// `review` narrows the weight rule to a single named review; `None`
2276    /// keeps the pre-D43 behaviour of taking the best of every review
2277    /// naming this `(ref, commit)`.
2278    fn authorization_for(
2279        &self,
2280        name: &str,
2281        commit: &ContentHash,
2282        review: Option<&str>,
2283        sub: &Submission,
2284        op: &ViewOp,
2285        actor_id: &ContentHash,
2286    ) -> Result<Authorization, String> {
2287        let acl = self.acl_now()?;
2288        let repo = crate::acl::ref_repo(name);
2289        if let (Some(acl), Some(repo)) = (acl.as_ref(), repo.as_ref()) {
2290            if acl.has_owner(repo) {
2291                return self.owner_assented(acl, repo, name, commit, review, sub, op, actor_id);
2292            }
2293        }
2294        let (approval_weight, approvers) = self.weight_and_approvers(name, commit, review)?;
2295        if approval_weight < REQUIRED_APPROVAL_WEIGHT {
2296            return Err(Rejection::new(
2297                Code::ReviewRequired,
2298                format!(
2299                    "{name} is protected and this commit has approval weight \
2300                     {approval_weight}, below the required {REQUIRED_APPROVAL_WEIGHT}"
2301                ),
2302                "open a review naming this ref and commit (`choir review ... --ref \
2303                 <repo:ref>`), obtain approvals from two distinct operators, then push again",
2304            )
2305            .with_states(
2306                Some(format!(
2307                    "approval weight {REQUIRED_APPROVAL_WEIGHT} for {}",
2308                    commit.to_hex()
2309                )),
2310                Some(format!("approval weight {approval_weight}")),
2311            )
2312            .encode());
2313        }
2314        Ok(Authorization::new(
2315            Basis::ApprovalWeight {
2316                required: u32::try_from(REQUIRED_APPROVAL_WEIGHT).unwrap_or(u32::MAX),
2317                met: u32::try_from(approval_weight).unwrap_or(u32::MAX),
2318            },
2319            approvers,
2320        ))
2321    }
2322
2323    /// The approval weight backing a landing, and the actor ids of the
2324    /// approvals it counted (D43).
2325    ///
2326    /// The two come from one call because they must describe the same
2327    /// review: `approval_weight_for` takes the maximum over every review
2328    /// naming `(ref, commit)`, so a weight and a separately-derived
2329    /// approver list could easily belong to different rows.
2330    ///
2331    /// Approver ids are resolved only when a `Submit` asked for them
2332    /// (`review` is `Some`). A plain push does not need them and must not
2333    /// be refused for an unbound reviewer key, which would change the
2334    /// pre-D43 push gate.
2335    fn weight_and_approvers(
2336        &self,
2337        name: &str,
2338        commit: &ContentHash,
2339        review: Option<&str>,
2340    ) -> Result<(usize, Vec<ContentHash>), String> {
2341        let Some(review) = review else {
2342            return Ok((self.approval_weight_for(name, commit), Vec::new()));
2343        };
2344        let view = self.view.lock().expect("view lock");
2345        let Some(state) = view.reviews.get(review) else {
2346            return Ok((0, Vec::new()));
2347        };
2348        if state.target_ref.as_deref() != Some(name) || state.target.as_ref() != Some(commit) {
2349            return Ok((0, Vec::new()));
2350        }
2351        if !state.approved() {
2352            return Ok((0, Vec::new()));
2353        }
2354        let approvers = state
2355            .counted_approvers()
2356            .into_iter()
2357            .map(|(channel, at)| view.bound_actor_at(channel, at))
2358            .collect::<Result<Vec<_>, String>>()
2359            .map_err(|e| Self::unbound_approver(&e))?;
2360        Ok((state.approval_weight(), approvers))
2361    }
2362
2363    /// The rejection for an approval this node cannot name an actor id
2364    /// for (D43).
2365    ///
2366    /// A landing whose approvers cannot be identified is refused rather
2367    /// than recorded with a gap. That is a liveness cost on a node with
2368    /// no `BindKey` records, and it buys the property the whole record
2369    /// exists for: `channel_is_owned` constrains *bound* keys only, so
2370    /// an approval from an unbound channel is one nobody can be held to.
2371    fn unbound_approver(reason: &str) -> String {
2372        Rejection::new(
2373            Code::ReviewRequired,
2374            format!("this landing cannot name its approvers: {reason}"),
2375            "a merge records who approved it as actor ids, which it reads from the log's own \
2376             key bindings. Ask the operator to bind that reviewer's key (`choir bind-key`), \
2377             then merge again",
2378        )
2379        .encode()
2380    }
2381
2382    /// Whether an owner of `repo` assented to this landing, in either of
2383    /// the two ways that count (D42).
2384    ///
2385    /// One question with two answers, not a rule plus an exception:
2386    /// performing the landing is assent, and so is having approved a
2387    /// review that names this exact `(ref, commit)`. The first is what
2388    /// lets an owner land their own work without reviewing themselves —
2389    /// which they could not do anyway, since the reviewer draw excludes
2390    /// the requester's own operator.
2391    ///
2392    /// Returns which of the two answers applied, so a [`OpKind::Submit`]
2393    /// can record it (D43). The order is load-bearing and is the order
2394    /// the two answers are checked in: an owner who both approved and
2395    /// landed is recorded as having landed, because that is the assent
2396    /// the gate actually rested on.
2397    ///
2398    /// **The two answers match the ACL's user column against different
2399    /// namespaces, and an operator writing that file has to know which.**
2400    /// The landing answer asks about [`Self::acting_user`] — for a push
2401    /// that is the transport channel minus its `git/` prefix, which is
2402    /// the `--auth-file` username; for a signed op it is the *bound name*
2403    /// of the signing key. The approval answer asks about a reviewer
2404    /// channel, the key of [`choir_view::ReviewState::verdicts`], spelled
2405    /// the way the reviewer pool spells it (`someone/reviewer`).
2406    ///
2407    /// So `alice choir/choir.git own` grants the landing answer and never
2408    /// the approval one, and `alice/reviewer choir/choir.git own` grants
2409    /// the reverse. Neither is wrong and nothing warns which was meant.
2410    /// Granting the wrong spelling still flips [`crate::acl::Effective::has_owner`],
2411    /// which switches the repository out of the approval-weight rule —
2412    /// so a mismatched grant does not fall back, it narrows the gate to a
2413    /// rule the intended actor cannot satisfy.
2414    ///
2415    /// For an owner landing their own work the landing answer is the only
2416    /// reachable one regardless, because the reviewer draw excludes the
2417    /// requester's own operator.
2418    #[allow(clippy::too_many_arguments)]
2419    fn owner_assented(
2420        &self,
2421        acl: &crate::acl::Effective,
2422        repo: &str,
2423        name: &str,
2424        commit: &ContentHash,
2425        review: Option<&str>,
2426        sub: &Submission,
2427        op: &ViewOp,
2428        actor_id: &ContentHash,
2429    ) -> Result<Authorization, String> {
2430        let acting = self.acting_user(sub, op, actor_id);
2431        if let Some(user) = acting.as_deref() {
2432            if acl.allows_repo(user, repo, crate::acl::Level::Own) {
2433                return Ok(Authorization::new(
2434                    Basis::OwnerLanded {
2435                        owner: user.to_string(),
2436                    },
2437                    Vec::new(),
2438                ));
2439            }
2440        }
2441        if let Some((owner, approved_at)) = self.owner_approved(acl, repo, name, commit, review) {
2442            // Only a `Submit` needs the id, and only a `Submit` may be
2443            // refused for the want of one. Resolving it on the push path
2444            // too would make an unbound reviewer key break a landing that
2445            // D42 admits today, which is a gate change wearing the shape
2446            // of a record change.
2447            let approvers = match review {
2448                Some(_) => vec![self
2449                    .view
2450                    .lock()
2451                    .expect("view lock")
2452                    .bound_actor_at(&owner, approved_at)
2453                    .map_err(|e| Self::unbound_approver(&e))?],
2454                None => Vec::new(),
2455            };
2456            return Ok(Authorization::new(
2457                Basis::OwnerApproved { owner },
2458                approvers,
2459            ));
2460        }
2461        // A certified push fails here for a reason the generic message
2462        // would misdescribe: not "you are not an owner" but "the node
2463        // cannot tell who you are", which has a different repair.
2464        let next = if matches!(op.provenance, Some(Provenance::PushCertified)) {
2465            "a signed push cannot assert repository ownership: its channel names the push \
2466             certificate's key rather than an ACL user. Push without `--signed`, or have an \
2467             owner approve a review naming this ref and commit"
2468        } else {
2469            "have an owner of this repository approve a review naming this ref and commit, \
2470             or land it as an owner yourself. An owner submitting directly must have their \
2471             key bound to their name in the trusted-keys file, or the node cannot tell the \
2472             key is theirs"
2473        };
2474        Err(Rejection::new(
2475            Code::ReviewRequired,
2476            format!(
2477                "{name} is protected and {repo} is owned; no owner has assented to this commit"
2478            ),
2479            next,
2480        )
2481        .with_states(
2482            Some(format!("owner assent for {}", commit.to_hex())),
2483            Some(match acting {
2484                Some(user) => format!("a landing by {user}, who does not own {repo}"),
2485                None => "a landing by an identity the node cannot resolve to an ACL user".into(),
2486            }),
2487        )
2488        .encode())
2489    }
2490
2491    /// Whether an owner of `repo` approved a review naming exactly this
2492    /// `(ref, commit)` pair (D42).
2493    ///
2494    /// Deliberately **not** routed through [`choir_view::ReviewState::approved`],
2495    /// which returns false until every drawn reviewer has answered. Under
2496    /// D42 an owner's approval is sufficient on its own, so consulting
2497    /// `approved()` would let one reviewer who never replies veto a
2498    /// landing the owner had already assented to — a liveness bug wearing
2499    /// the shape of a safety check.
2500    ///
2501    /// A slashed approval does not count. A retroactively invalidated
2502    /// verdict is invalidated for an owner exactly as for anybody else,
2503    /// which is the whole point of `SlashApproval` existing.
2504    ///
2505    /// **The slash test is per operator, not per channel** — the same
2506    /// predicate the weight rule applies, via
2507    /// [`choir_view::ReviewState::approval_stands`]. D42 shipped this
2508    /// check spelled `slashes.contains_key(reviewer)`, which let a
2509    /// slashed operator's *other* channel keep authorizing while its
2510    /// weight contribution was already gone. Two answers to "is this
2511    /// approval still good" is one more than a gate may have.
2512    ///
2513    /// Returns the approving owner's channel, which a
2514    /// [`OpKind::Submit`] records (D43). `review` narrows the search to
2515    /// one named review; `None` searches every review naming this
2516    /// `(ref, commit)`, which is the pre-D43 push behaviour.
2517    fn owner_approved(
2518        &self,
2519        acl: &crate::acl::Effective,
2520        repo: &str,
2521        name: &str,
2522        commit: &ContentHash,
2523        review: Option<&str>,
2524    ) -> Option<(String, u64)> {
2525        let view = self.view.lock().expect("view lock");
2526        view.reviews
2527            .iter()
2528            .filter(|(id, state)| {
2529                review.is_none_or(|wanted| wanted == id.as_str())
2530                    && state.target_ref.as_deref() == Some(name)
2531                    && state.target.as_ref() == Some(commit)
2532            })
2533            .find_map(|(_, state)| {
2534                state.verdicts.keys().find_map(|reviewer| {
2535                    // The position comes back with the name because the
2536                    // approval has to be resolved to the key that was
2537                    // live when it was cast, not the one holding the
2538                    // channel now (D44).
2539                    let at = state.standing_approval_at(reviewer)?;
2540                    acl.allows_repo(reviewer, repo, crate::acl::Level::Own)
2541                        .then(|| (reviewer.clone(), at))
2542                })
2543            })
2544    }
2545
2546    /// Admits a [`OpKind::Submit`] only if the authorization it carries
2547    /// is the one this node's own gate produces (D43).
2548    ///
2549    /// The op is author-signed, so `authorization` arrives as a claim.
2550    /// It is never trusted and never patched: the gate runs, and the
2551    /// claim must equal its result exactly. A client that writes its own
2552    /// basis gets a rejection naming both sides, not a landing.
2553    ///
2554    /// Refused outright when no rule would examine the landing — the
2555    /// operator is not running the review gate, or the ref is not
2556    /// protected. `Submit`'s entire value is the record, and a record of
2557    /// a decision nothing made is worse than no record: it reads, to
2558    /// every later auditor, exactly like one that was checked.
2559    #[allow(clippy::too_many_arguments)]
2560    fn submit_is_authorized(
2561        &self,
2562        gating: bool,
2563        review: &str,
2564        name: &str,
2565        commit: &ContentHash,
2566        claimed: &Authorization,
2567        sub: &Submission,
2568        op: &ViewOp,
2569        actor_id: &ContentHash,
2570    ) -> Result<(), String> {
2571        if !gating || !self.ref_is_protected(name)? {
2572            return Err(Rejection::new(
2573                Code::ReviewRequired,
2574                format!(
2575                    "{name} is not gated on this node, so a landing record for it would \
2576                     assert a review that nothing performed"
2577                ),
2578                "move the ref with an ordinary `SetRef`, or ask the operator to protect this \
2579                 ref and enable `--require-review`",
2580            )
2581            .encode());
2582        }
2583        let computed = self.authorization_for(name, commit, Some(review), sub, op, actor_id)?;
2584        if computed == *claimed {
2585            return Ok(());
2586        }
2587        // `expected` carries the gate's answer as its own canonical JSON,
2588        // not a summary of it. That is what makes the mismatch a usable
2589        // two-step protocol rather than a dead end: a client submits,
2590        // reads the authorization it should have signed, and submits
2591        // again. The alternative is a second implementation of this rule
2592        // on the read path so a client could ask in advance -- and a
2593        // record that can disagree with the decision it describes is the
2594        // one thing this field must never be.
2595        Err(Rejection::new(
2596            Code::ReviewRequired,
2597            format!(
2598                "the authorization on this submit is not the one {name}'s gate produced: \
2599                 it admits this landing as {}, and the submit claims {}",
2600                Self::basis_summary(&computed),
2601                Self::basis_summary(claimed)
2602            ),
2603            "sign and resubmit with the authorization in `expected`, verbatim. It is the \
2604             gate's own answer and a client cannot assert it",
2605        )
2606        .with_states(
2607            serde_json::to_string(&computed).ok(),
2608            serde_json::to_string(claimed).ok(),
2609        )
2610        .encode())
2611    }
2612
2613    /// One-line rendering of an authorization, for the prose half of a
2614    /// mismatch rejection. Approvers are counted rather than listed: the
2615    /// hex ids would bury the part that differs.
2616    fn basis_summary(authorization: &Authorization) -> String {
2617        let basis = match &authorization.basis {
2618            Basis::OwnerLanded { owner } => format!("landed by owner {owner}"),
2619            Basis::OwnerApproved { owner } => format!("approved by owner {owner}"),
2620            Basis::ApprovalWeight { required, met } => {
2621                format!("approval weight {met} against a required {required}")
2622            }
2623        };
2624        format!("{basis} with {} approver(s)", authorization.approvers.len())
2625    }
2626
2627    /// Refuses a submission that has already been admitted, and one whose
2628    /// author never bound it to this log at all.
2629    ///
2630    /// Together these are the replay defence, and they compose into
2631    /// at-most-once permanently even though both indexes are bounded by
2632    /// the window. The argument is short enough to check: a scoped op is
2633    /// admissible only while the head it names is still in the window;
2634    /// its own signing hash entered the window at a *later* sequence
2635    /// than that head, so whenever the head is still there the signing
2636    /// hash is too and the duplicate check refuses it. Once the head is
2637    /// gone the scope check refuses it. There is no sequence at which
2638    /// neither fires.
2639    ///
2640    /// Without a scope only the duplicate half applies, which bounds a
2641    /// replay to the window instead of refusing it outright — the reason
2642    /// `--require-scope` exists.
2643    fn admit_once(&self, signing: &ContentHash, op: &ViewOp) -> Result<(), String> {
2644        // Nothing is cloned out of the window here. Every admitted op runs
2645        // this, while only a refused one needs the head to explain itself,
2646        // so the head is read again on the rejection path instead of
2647        // copied on the hot one -- one allocation per op, which the
2648        // allocation budget notices.
2649        let (already, holds_head, nothing_evicted) = {
2650            let window = self.entries.lock().expect("entries lock");
2651            (
2652                window.seq_for_signing(signing),
2653                op.scope
2654                    .as_ref()
2655                    .and_then(|s| s.head.as_ref())
2656                    .is_some_and(|h| window.holds_head(h)),
2657                window.base == 0,
2658            )
2659        };
2660        let window_head = || {
2661            self.entries
2662                .lock()
2663                .expect("entries lock")
2664                .head_hash()
2665                .as_ref()
2666                .map(ContentHash::to_hex)
2667        };
2668        // A signature is admissible once. `prev` cannot enforce that: it
2669        // compares state, and state recurs — land a commit, revert it,
2670        // and the reverted-away op's CAS matches again. The HTTP layer
2671        // answers any rejection whose submission already landed as 200
2672        // `already_applied` with the original seq, so a lost-response
2673        // retry still reads as success while a replay becomes a no-op.
2674        if let Some(seq) = already {
2675            return Err(Rejection::new(
2676                Code::DuplicateSubmission,
2677                format!("these exact signed bytes already landed at seq {seq}"),
2678                "if you are retrying, read `seq` from this response — it names the op you \
2679                 already have. If you meant a second, distinct change, sign a new op: two \
2680                 otherwise byte-identical ops are told apart by their scope.",
2681            )
2682            .with_states(Some(seq.to_string()), None)
2683            .encode());
2684        }
2685        let Some(scope) = &op.scope else {
2686            if self
2687                .require_scope
2688                .load(std::sync::atomic::Ordering::Relaxed)
2689            {
2690                return Err(Rejection::new(
2691                    Code::ScopeRequired,
2692                    "this node admits only ops signed for its own log and a recent head",
2693                    "read `log.node` and `log.head` from GET /api/view, put them in the \
2694                     op's `scope`, and sign that; `choir submit` does it for you",
2695                )
2696                .with_states(None, window_head())
2697                .encode());
2698            }
2699            return Ok(());
2700        };
2701        if scope.node != self.node_id {
2702            return Err(Rejection::new(
2703                Code::ForeignScope,
2704                "this op was signed for another node's log",
2705                "sign a scope naming this node; its id is in `actual` and in `log.node` \
2706                 of GET /api/view",
2707            )
2708            .with_states(Some(scope.node.to_hex()), Some(self.node_id.to_hex()))
2709            .encode());
2710        }
2711        match &scope.head {
2712            Some(_) if holds_head => Ok(()),
2713            // The op names no head, which is what a client signs when it
2714            // read an empty log — including every op of a batch it signed
2715            // in one go, since only the first of those lands against a
2716            // log that is still empty.
2717            //
2718            // Admissible while this window has evicted nothing, because
2719            // that is exactly the span the duplicate index above covers
2720            // in full: an unevicted window holds every entry, so a replay
2721            // of a headless op cannot slip past it. The moment the first
2722            // entry is evicted the guarantee would thin out, and the op
2723            // stops being admissible instead.
2724            None if nothing_evicted => Ok(()),
2725            None => Err(Rejection::new(
2726                Code::StaleScope,
2727                "the op names no head, and this log has evicted entries since",
2728                "re-read `log.head` from GET /api/view and sign a fresh op against it",
2729            )
2730            .with_states(
2731                Some("a log with nothing evicted".to_string()),
2732                window_head(),
2733            )
2734            .encode()),
2735            Some(head) => Err(Rejection::new(
2736                Code::StaleScope,
2737                "the head this op was signed against is no longer in the window",
2738                "re-read `log.head` from GET /api/view and sign a fresh op against it; a \
2739                 signature stays admissible only as long as the head it names does",
2740            )
2741            .with_states(Some(head.to_hex()), window_head())
2742            .encode()),
2743        }
2744    }
2745}
2746
2747impl ChoirPolicy {
2748    /// Verifies a passkey submission against the credential the *channel*
2749    /// enrolled (D39).
2750    ///
2751    /// The channel is the account name, and that is the whole binding:
2752    /// `Accounts::passkey_spki` is keyed by `(account, credential_id)`, so
2753    /// an assertion by alice's authenticator submitted on bob's channel
2754    /// finds no key and is refused. There is no separate table mapping
2755    /// credentials to channels, because the store already is one — and a
2756    /// second table would be a second thing to keep in agreement.
2757    ///
2758    /// The returned actor id is `blake3` of the credential's public key,
2759    /// the same rule [`choir_identity::ActorKey::actor_id`] uses for an
2760    /// ed25519 key. It is an in-process authorization handle only: an
2761    /// `OpEntry` records the channel and the signature, never an actor
2762    /// id, so this derivation is not frozen into the log and changing it
2763    /// later would not be a migration.
2764    fn verify_passkey(
2765        &self,
2766        signing: &ContentHash,
2767        sig: &Witness,
2768        channel: &str,
2769    ) -> Result<ContentHash, choir_identity::IdentityError> {
2770        let store = self.passkeys.lock().expect("passkey store lock").clone();
2771        let Some(store) = store else {
2772            // No store at all: the node has no self-service, so it has no
2773            // enrolled credentials and this key id is unknown to it.
2774            return Err(choir_identity::IdentityError::UnknownKey(
2775                sig.key_id.clone(),
2776            ));
2777        };
2778        let spki = store
2779            .passkey_spki(channel, &sig.key_id)
2780            .ok_or_else(|| choir_identity::IdentityError::UnknownKey(sig.key_id.clone()))?;
2781        choir_identity::verify_webauthn_assertion(&spki, signing, sig)?;
2782        Ok(ContentHash::blake3(&spki))
2783    }
2784}
2785
2786impl SubmitPolicy for ChoirPolicy {
2787    fn check(&mut self, sub: &Submission) -> Result<(), String> {
2788        let sig = sub.author_sig.as_ref().ok_or("unsigned submission")?;
2789        // Refresh before verification so removing a trusted key takes
2790        // effect on that key's very next request. A failed verification
2791        // cannot trigger this tightening: a removed key is still present
2792        // in the stale registry and would verify successfully.
2793        self.reload_keys();
2794        let signing = choir_oplog::signing_hash(&sub.channel, &sub.payload);
2795        let mut verified_actor = if sig.scheme_id() == choir_oplog::scheme::WEBAUTHN_ES256 {
2796            self.verify_passkey(&signing, sig, &sub.channel)
2797        } else {
2798            self.registry.verify_signing_hash(&signing, sig)
2799        };
2800        // Retry a failed signature in case the file changed between the
2801        // pre-verification metadata check and this verification. Only the
2802        // ed25519 path has a file behind it; a passkey lives in the store
2803        // and is read fresh on every call already.
2804        if verified_actor.is_err()
2805            && sig.scheme_id() != choir_oplog::scheme::WEBAUTHN_ES256
2806            && self.reload_keys()
2807        {
2808            verified_actor = self.registry.verify_signing_hash(&signing, sig);
2809        }
2810        self.subject = (None, None);
2811        let actor_id = verified_actor.map_err(|e| {
2812            // Two failures with opposite repairs, and one of them is an
2813            // attack signal, so they cannot share a code. A key id the
2814            // node has no record of is a trust gap the operator closes.
2815            // A signature that does not verify under a key the node
2816            // already trusts is either corruption or a lifted signature
2817            // replayed onto other bytes -- and answering that with "ask
2818            // the operator to register your public key" hands the party
2819            // being impersonated the one repair that helps the attacker.
2820            // Anything else is reported as the bad signature too: of the
2821            // two directions to be wrong in, refusing is the safe one.
2822            let detail = format!("signature check failed: {e:?}");
2823            match &e {
2824                choir_identity::IdentityError::UnknownKey(_) => Rejection::new(
2825                    Code::UnknownKey,
2826                    detail,
2827                    "ask the operator to add your public key to the node's trusted-keys file                  (`choir key <file> <you>` prints the line); it takes effect on the next request",
2828                ),
2829                _ => Rejection::new(
2830                    Code::BadSignature,
2831                    detail,
2832                    "re-sign the exact bytes you are submitting; a signature covers one \
2833                     (channel, payload) pair and does not carry to another. Registering a key \
2834                     does not help here, the key this names is already trusted -- if you did not \
2835                     send this, a signature of yours was replayed onto bytes you never signed",
2836                ),
2837            }
2838            .encode()
2839        })?;
2840        // Derived here, while the signature is already verified, and
2841        // only when something will read it: naming the op means
2842        // serializing its kind, which is not free per op.
2843        let journalling = self.journal.enabled();
2844        if journalling {
2845            self.subject.0 = Some(actor_id.to_hex());
2846        }
2847        let op = ViewOp::from_payload(&sub.payload).map_err(|e| {
2848            Rejection::new(
2849                Code::MalformedOp,
2850                format!("payload did not decode as a ViewOp: {e:?}"),
2851                "sign the bytes of a serialized ViewOp; `choir submit` does this correctly",
2852            )
2853            .encode()
2854        })?;
2855        if journalling {
2856            self.subject.1 = op_type_name(&op);
2857        }
2858        // Before any policy that asks what the op *does*: has this
2859        // signature already been spent, and was it ever meant for this
2860        // log at all.
2861        self.admit_once(&signing, &op)?;
2862        // A verdict's claimed reviewer must be the signature-covered
2863        // submission channel: the log's author attribution and the
2864        // view's verdict attribution can never diverge.
2865        if let OpKind::PostVerdict { reviewer, .. } = &op.kind {
2866            if *reviewer != sub.channel {
2867                return Err(Rejection::new(
2868                    Code::ReviewerMismatch,
2869                    "a verdict's reviewer must be the channel it was signed on",
2870                    "resubmit on your own channel: `choir verdict` signs on the reviewer name                      by construction",
2871                )
2872                .with_states(Some(sub.channel.clone()), Some(reviewer.clone()))
2873                .encode());
2874            }
2875        }
2876        // A comment's claimed author is bound the same way and for a
2877        // sharper reason: a verdict in the wrong name is a wrong
2878        // authorization, a comment in the wrong name is words somebody
2879        // never said. The view stores `author`, so the payload claim is
2880        // what a reader sees, and it must be the channel that signed.
2881        if let OpKind::PostComment { author, .. } = &op.kind {
2882            if *author != sub.channel {
2883                return Err(Rejection::new(
2884                    Code::ReviewerMismatch,
2885                    "a comment's author must be the channel it was signed on",
2886                    "resubmit on your own channel: `choir comment` signs on the author name by \
2887                     construction",
2888                )
2889                .with_states(Some(sub.channel.clone()), Some(author.clone()))
2890                .encode());
2891            }
2892        }
2893        // A receipt's claimed viewer is bound the same way: the receipt
2894        // exists to attribute attention, and attention recorded in
2895        // somebody else's name is exactly the false "it was looked at"
2896        // signal the op exists to remove.
2897        if let OpKind::ViewedReview { viewer, .. } = &op.kind {
2898            if *viewer != sub.channel {
2899                return Err(Rejection::new(
2900                    Code::ReviewerMismatch,
2901                    "a receipt's viewer must be the channel it was signed on",
2902                    "resubmit on your own channel: `choir viewed` signs on the viewer name by \
2903                     construction",
2904                )
2905                .with_states(Some(sub.channel.clone()), Some(viewer.clone()))
2906                .encode());
2907            }
2908        }
2909        // A vouch's claimed voucher is bound the same way, one level up:
2910        // the payload names an *operator* and the submission is signed on
2911        // a *channel*, so the comparison is against the channel's
2912        // operator prefix rather than the channel itself. `ops/agent` and
2913        // `ops` are the same operator and either may sign; `rival` may
2914        // not, and an unchecked `voucher` is precisely a Sybil writing
2915        // somebody else's endorsements (D65).
2916        if let OpKind::CountersignSnapshot { witness, .. } = &op.kind {
2917            let operator = reviewer_operator(&sub.channel);
2918            if witness != operator {
2919                return Err(Rejection::new(
2920                    Code::ReviewerMismatch,
2921                    "a countersignature's witness must be the operator of the channel it was \
2922                     signed on",
2923                    "resubmit on a channel belonging to that operator: `choir witness` derives \
2924                     the witness from the channel by construction",
2925                )
2926                .with_states(Some(operator.to_string()), Some(witness.clone()))
2927                .encode());
2928            }
2929        }
2930        if let OpKind::Vouch { voucher, .. } | OpKind::WithdrawVouch { voucher, .. } = &op.kind {
2931            let operator = reviewer_operator(&sub.channel);
2932            if voucher != operator {
2933                return Err(Rejection::new(
2934                    Code::ReviewerMismatch,
2935                    "a vouch's voucher must be the operator of the channel it was signed on",
2936                    "resubmit on a channel belonging to that operator: `choir vouch` derives \
2937                     the voucher from the channel by construction",
2938                )
2939                .with_states(Some(operator.to_string()), Some(voucher.clone()))
2940                .encode());
2941            }
2942        }
2943        // ...and the channel itself must belong to the signing key, or
2944        // the check above only proves a claim is self-consistent, not
2945        // that it is true. Review ops only: see `channel_is_owned`.
2946        if matches!(
2947            op.kind,
2948            OpKind::PostVerdict { .. }
2949                | OpKind::RequestReview { .. }
2950                | OpKind::PostComment { .. }
2951                | OpKind::ViewedReview { .. }
2952                | OpKind::Vouch { .. }
2953                | OpKind::WithdrawVouch { .. }
2954                | OpKind::CountersignSnapshot { .. }
2955        ) {
2956            self.channel_is_owned(&actor_id, &sub.channel)?;
2957        }
2958        // D41: a provenance label claims "the node signed this on behalf
2959        // of a git pusher". Unenforced, the label would be exactly the
2960        // laundering it exists to prevent — any author could dress an op
2961        // as push-derived, or downstream code could trust the label
2962        // without checking the signer. Enforced here, an accepted labeled
2963        // op always carries the node's own signature.
2964        if op.provenance.is_some() && actor_id != self.node_id {
2965            return Err(Rejection::new(
2966                Code::NodeOnly,
2967                "only the node may label an op with a push provenance",
2968                "submit without `provenance`: author-signed ops are the default class and need no label",
2969            )
2970            .encode());
2971        }
2972        // D24 layer 5: the requester does not choose who reviews them.
2973        // Only the daemon's own key may fill in a reviewer list; every
2974        // other author gets a rejection, so an accepted assignment in
2975        // the log always came from the node's pool draw.
2976        if matches!(op.kind, OpKind::AssignReviewers { .. }) && actor_id != self.node_id {
2977            return Err(Rejection::new(
2978                Code::NodeOnly,
2979                "only the node may assign reviewers",
2980                "request a review with an empty reviewer list and the node will draw them",
2981            )
2982            .encode());
2983        }
2984        // Archiving drops a review's verdicts, so an unguarded one is a
2985        // way to erase a RequestChanges you did not like. Node key only,
2986        // same reasoning as assignment: it is retention, not review.
2987        if matches!(op.kind, OpKind::ArchiveReview { .. }) && actor_id != self.node_id {
2988            return Err(Rejection::new(
2989                Code::NodeOnly,
2990                "only the node may archive reviews",
2991                "nothing to do: archiving is retention, performed by the node",
2992            )
2993            .encode());
2994        }
2995        // Stable change creation is coupled to physical workspace
2996        // provisioning. Only the node may record it, after the exact base
2997        // was verified and the checkout was materialized.
2998        if matches!(
2999            op.kind,
3000            OpKind::CreateChange { .. } | OpKind::ArchiveChange { .. }
3001        ) && actor_id != self.node_id
3002        {
3003            return Err(Rejection::new(
3004                Code::NodeOnly,
3005                "only the node may author physical workspace lifecycle operations",
3006                "call POST /api/workspace to create, or POST /api/workspace/archive with an owner-signed authorization to archive",
3007            )
3008            .encode());
3009        }
3010        if let OpKind::CreateChange {
3011            id,
3012            owner,
3013            workspace,
3014            base_revision,
3015            idempotency_key,
3016            owner_sig: Some(owner_sig),
3017            cone,
3018        } = &op.kind
3019        {
3020            // The cone is rederived into the authorization rather than
3021            // trusted from the op, which is what makes it a *declaration
3022            // by the owner*: a node that attached a scope its owner did
3023            // not sign produces different bytes here and fails the check
3024            // below (D50).
3025            let authorization = CreateAuthorization::new(
3026                id.clone(),
3027                owner.clone(),
3028                workspace.clone(),
3029                base_revision.clone(),
3030                idempotency_key.clone(),
3031            )
3032            .with_cone(cone.clone())
3033            .to_payload();
3034            let mut verified_owner =
3035                self.registry
3036                    .verify_submission(owner, &authorization, owner_sig);
3037            if verified_owner.is_err() && self.reload_keys() {
3038                verified_owner = self
3039                    .registry
3040                    .verify_submission(owner, &authorization, owner_sig);
3041            }
3042            let owner_actor = verified_owner.map_err(|error| {
3043                Rejection::new(
3044                    Code::UnknownKey,
3045                    format!("create owner signature check failed: {error:?}"),
3046                    "register the owner key, then retry the same signed create authorization",
3047                )
3048                .encode()
3049            })?;
3050            self.channel_is_owned(&owner_actor, owner)?;
3051        }
3052        if let OpKind::CheckpointChange { id, .. } = &op.kind {
3053            let owner = self
3054                .view
3055                .lock()
3056                .expect("view lock")
3057                .changes
3058                .get(id)
3059                .map(|change| change.owner.clone());
3060            if let Some(owner) = owner {
3061                if owner != sub.channel {
3062                    return Err(Rejection::new(
3063                        Code::ChannelNotOwned,
3064                        format!("change {id} is owned by a different channel"),
3065                        "sign the change operation on the owner channel reported in GET /api/view, or \
3066                         create a separate change",
3067                    )
3068                    .with_states(Some(owner), Some(sub.channel.clone()))
3069                    .encode());
3070                }
3071                self.channel_is_owned(&actor_id, &sub.channel)?;
3072            }
3073        }
3074        if let OpKind::ArchiveChange {
3075            id,
3076            workspace,
3077            prev_revision,
3078            owner,
3079            owner_sig,
3080        } = &op.kind
3081        {
3082            let authorization =
3083                ArchiveAuthorization::new(id.clone(), workspace.clone(), prev_revision.clone())
3084                    .to_payload();
3085            let mut verified_owner =
3086                self.registry
3087                    .verify_submission(owner, &authorization, owner_sig);
3088            if verified_owner.is_err() && self.reload_keys() {
3089                verified_owner = self
3090                    .registry
3091                    .verify_submission(owner, &authorization, owner_sig);
3092            }
3093            let owner_actor = verified_owner.map_err(|error| {
3094                Rejection::new(
3095                    Code::UnknownKey,
3096                    format!("archive owner signature check failed: {error:?}"),
3097                    "register the owner key, then retry the same signed archive authorization",
3098                )
3099                .encode()
3100            })?;
3101            self.channel_is_owned(&owner_actor, owner)?;
3102        }
3103        // Legacy workspace moves and deletes remain compatible for
3104        // legacy workspaces. Once a workspace is enrolled in a stable
3105        // change, non-node callers must use CheckpointChange and the
3106        // recoverable archive endpoint so identity cannot be bypassed.
3107        let bound_workspace = match &op.kind {
3108            OpKind::SetWorkspaceHead { workspace, .. } | OpKind::DeleteWorkspace { workspace } => {
3109                self.view
3110                    .lock()
3111                    .expect("view lock")
3112                    .changes
3113                    .values()
3114                    .any(|change| change.active_workspace.as_deref() == Some(workspace))
3115            }
3116            _ => false,
3117        };
3118        if bound_workspace && actor_id != self.node_id {
3119            return Err(Rejection::new(
3120                Code::WorkspaceState,
3121                "a change-bound workspace cannot be moved or removed through a legacy op",
3122                "use choir checkpoint to publish a revision, or choir workspace-archive to \
3123                 archive the bound workspace",
3124            )
3125            .encode());
3126        }
3127        // A slash can withdraw authorization from a review whose detail
3128        // has already been compacted. Only the node key may make that
3129        // durable attestation; otherwise any trusted author could erase
3130        // another operator's approval weight.
3131        if matches!(op.kind, OpKind::SlashApproval { .. }) && actor_id != self.node_id {
3132            return Err(Rejection::new(
3133                Code::NodeOnly,
3134                "only the node may slash approvals",
3135                "ask the operator to run `choir slash` with the node key",
3136            )
3137            .encode());
3138        }
3139        // A ref snapshot is the node's own attestation of its complete
3140        // ref-state (D25): the unit a witness will cosign and the thing
3141        // two readers compare to detect equivocation. The fold already
3142        // refuses an untruthful one; this guard is about authorship —
3143        // signed by anyone else it attests nothing about the node while
3144        // reading as though it did.
3145        if matches!(op.kind, OpKind::RecordRefSnapshot { .. }) && actor_id != self.node_id {
3146            return Err(Rejection::new(
3147                Code::NodeOnly,
3148                "only the node may record ref snapshots",
3149                "read the latest snapshot from the view; the node attests its own ref-state",
3150            )
3151            .encode());
3152        }
3153        if matches!(op.kind, OpKind::CountersignSnapshot { .. }) && actor_id == self.node_id {
3154            return Err(Rejection::new(
3155                Code::NodeOnly,
3156                "the node cannot witness its own ref-state attestation",
3157                "a witness is worth counting only because it is not the node that made the \
3158                 claim; have an independent operator cosign it",
3159            )
3160            .encode());
3161        }
3162        // Key bindings are the durable operator record that T3 attribution
3163        // and T1's ordering primitive read, so a binding any trusted key
3164        // could author is evidence forgeable by the actors it is meant to
3165        // weigh -- worse than no record, because it reads as sequenced
3166        // proof. `check` is a series of per-variant guards with a
3167        // fall-through to `validate`, i.e. admit-by-default, so these
3168        // variants have to name themselves here to be refused.
3169        //
3170        // This also supplies the second condition on re-binding: the fold
3171        // lets a binding correct its channel (keeping `bound_at` pinned),
3172        // and that correction is only safe while the node is the one
3173        // making it.
3174        if matches!(op.kind, OpKind::BindKey { .. } | OpKind::RevokeKey { .. })
3175            && actor_id != self.node_id
3176        {
3177            return Err(Rejection::new(
3178                Code::NodeOnly,
3179                "only the node may bind or revoke operator keys",
3180                "ask the operator to record this binding with the node key",
3181            )
3182            .encode());
3183        }
3184        // Required-assignment closes the other half of the same loop:
3185        // naming your own reviewers is refused, so the node's draw is the
3186        // only way a review gets reviewers. Two ways to switch it on —
3187        // node-wide, or per-ref once the review says where it wants to
3188        // land. Node-wide wins because it is the stricter of the two.
3189        if let OpKind::RequestReview {
3190            reviewers,
3191            target_ref,
3192            ..
3193        } = &op.kind
3194        {
3195            if !reviewers.is_empty() {
3196                if self
3197                    .require_assignment
3198                    .load(std::sync::atomic::Ordering::Relaxed)
3199                {
3200                    return Err(Rejection::new(
3201                        Code::AssignmentRequired,
3202                        "this node assigns reviewers",
3203                        "resubmit the same review with an empty reviewer list; the node draws \
3204                         them and returns the names in the response",
3205                    )
3206                    .encode());
3207                }
3208                if let Some(name) = target_ref {
3209                    if self.ref_is_protected(name)? {
3210                        return Err(Rejection::new(
3211                            Code::ProtectedRef,
3212                            format!("{name} is a protected ref"),
3213                            "resubmit with an empty reviewer list; on a protected ref only a \
3214                             node-drawn reviewer list is accepted",
3215                        )
3216                        .encode());
3217                    }
3218                }
3219            }
3220        }
3221        // The landing half of the gate: a protected ref only moves to a
3222        // commit that some independently approved review already named
3223        // as its destination. This is what turns `--protected-refs` from an
3224        // advisory into enforcement — without it a requester escapes by
3225        // simply omitting `target_ref`.
3226        //
3227        // No exemption for the node's own key. Every git push arrives
3228        // here as a node-signed `SetRef`, so exempting the node would
3229        // exempt every push, which is the whole population being gated.
3230        let gating = self
3231            .require_review
3232            .load(std::sync::atomic::Ordering::Relaxed);
3233        if gating {
3234            match &op.kind {
3235                OpKind::SetRef { name, commit, prev } if self.ref_is_protected(name)? => {
3236                    // Creating a protected ref is allowed: there is no
3237                    // history to hijack yet, and deletion is refused
3238                    // below, so "delete then re-create" is not a way in.
3239                    if prev.is_some() {
3240                        self.authorization_for(name, commit, None, sub, &op, &actor_id)?;
3241                    }
3242                }
3243                OpKind::DeleteRef { name, .. } if self.ref_is_protected(name)? => {
3244                    return Err(Rejection::new(
3245                        Code::RefUndeletable,
3246                        format!("{name} is protected and cannot be deleted"),
3247                        "delete a different ref, or ask the operator to remove this one from \
3248                         the protected-ref list",
3249                    )
3250                    .encode());
3251                }
3252                _ => {}
3253            }
3254        }
3255        // Deliberately outside the `gating` block above, and deliberately
3256        // not guarded on `prev`. A `Submit` exists to carry an
3257        // authorization record; letting one through unexamined -- because
3258        // the operator turned the gate off, or because the ref is not
3259        // protected, or because it creates the ref rather than moving it
3260        // -- would mint a signed claim that a rule admitted a landing
3261        // when no rule looked at it. The claim is the asset here, so it
3262        // is the thing that must never be issued unbacked.
3263        if let OpKind::Submit {
3264            review,
3265            name,
3266            commit,
3267            authorization,
3268            ..
3269        } = &op.kind
3270        {
3271            self.submit_is_authorized(
3272                gating,
3273                review,
3274                name,
3275                commit,
3276                authorization,
3277                sub,
3278                &op,
3279                &actor_id,
3280            )?;
3281        }
3282        // Admission is a read. Every precondition `View::apply` enforces
3283        // is a CAS comparison or a key lookup, so this asks the shared
3284        // view directly instead of deep-cloning it -- four nested
3285        // BTreeMaps per submission, O(total state), which grew with the
3286        // repo's lifetime rather than with the size of the op.
3287        //
3288        // `View::apply` calls the same `validate`, so admission and
3289        // application cannot disagree; that shared path is what keeps
3290        // `accepted`'s "checked in check()" honest.
3291        self.view
3292            .lock()
3293            .expect("view lock")
3294            .validate(&op)
3295            .map_err(|e| {
3296                // A lost CAS is recorded as contention in its own right,
3297                // carrying both sides. A rejection count alone cannot
3298                // separate "two writers raced this ref" from "a client
3299                // sent nonsense", and only the first is a fact about
3300                // load.
3301                if let choir_view::ViewError::StaleHead {
3302                    expected, actual, ..
3303                } = &e
3304                {
3305                    self.journal.record(journal::Event::CasFailure {
3306                        workspace: sub.channel.clone(),
3307                        expected: expected.as_ref().map(ContentHash::to_hex),
3308                        actual: actual.as_ref().map(ContentHash::to_hex),
3309                    });
3310                }
3311                crate::reject::from_view_error(&e).encode()
3312            })
3313    }
3314
3315    fn subject(&self) -> (Option<String>, Option<String>) {
3316        self.subject.clone()
3317    }
3318
3319    fn accepted(&mut self, entry: &OpEntry, hash: &ContentHash) {
3320        let op = ViewOp::from_payload(&entry.payload).expect("checked in check()");
3321        {
3322            let mut view = self.view.lock().expect("view lock");
3323            view.apply(&op).expect("checked in check()");
3324            self.concentration
3325                .lock()
3326                .expect("concentration lock")
3327                .observe(entry, &op, &view);
3328            // D37, inside the same view-lock scope as the fold above:
3329            // a reader that took the tally between the two would see a
3330            // workspace the view already has and the tally does not.
3331            self.workspace_tally
3332                .lock()
3333                .expect("workspace tally lock")
3334                .observe(entry, &op);
3335        }
3336        if let Some(retention) = &self.review_retention {
3337            let mut retention = retention.lock().expect("review retention lock");
3338            let observed_at = retention.config.lapse_after.map(|_| Instant::now());
3339            retention.observe(&op, observed_at);
3340        }
3341        self.entries
3342            .lock()
3343            .expect("entries lock")
3344            .push(entry.clone(), hash.clone());
3345        self.offer_hook(entry, hash, &op);
3346    }
3347}
3348
3349/// A ref value as a receiver wants to read it: the git oid when the
3350/// value is one, and the self-describing content hash otherwise (the
3351/// platform API can point a ref at something that is not a git object).
3352fn oid_text(hash: &ContentHash) -> String {
3353    hash.git_oid().unwrap_or_else(|| hash.to_hex())
3354}
3355
3356impl ChoirPolicy {
3357    /// Hands a landed ref to the webhook delivery thread (D32).
3358    ///
3359    /// This runs on the writer thread, which is why it does nothing but
3360    /// build a small struct and `try_send` it. Matching, config reload,
3361    /// address vetting and `curl` all happen on the delivery thread; a
3362    /// full queue drops the event and counts it. Invariant 5 is the
3363    /// reason, and it is not a stylistic one: the target address belongs
3364    /// to somebody else, so a receiver that stops answering would
3365    /// otherwise stall op admission for everyone.
3366    fn offer_hook(&self, entry: &OpEntry, hash: &ContentHash, op: &ViewOp) {
3367        let guard = self.hooks.lock().expect("hooks lock");
3368        let Some(hooks) = guard.as_ref() else {
3369            return;
3370        };
3371        let (name, old, new) = match &op.kind {
3372            OpKind::SetRef { name, commit, prev } => {
3373                (name, prev.as_ref().map(oid_text), Some(oid_text(commit)))
3374            }
3375            OpKind::DeleteRef { name, prev } => (name, prev.as_ref().map(oid_text), None),
3376            _ => return,
3377        };
3378        hooks.offer(crate::hooks::RefEvent {
3379            key: name.clone(),
3380            old,
3381            new,
3382            seq: entry.seq,
3383            entry: hash.to_hex(),
3384            actor: entry.channel.clone(),
3385            key_id: entry.author_sig.as_ref().map(|sig| sig.key_id.clone()),
3386        });
3387    }
3388}
3389
3390/// A running platform: the sequencer plus the shared view it maintains.
3391pub struct Platform {
3392    /// The slot `with_journal` fills; shared with the writer thread and
3393    /// the admission policy, which both hold clones.
3394    journal: SharedJournal,
3395    handle: SequencerHandle,
3396    view: Arc<Mutex<View>>,
3397    /// The daemon's own key: signs ops it derives from authenticated git
3398    /// pushes. Attribution: a verified push certificate names the
3399    /// pusher's key (`key/<principal>`); otherwise the basic-auth user
3400    /// (`git/<user>`).
3401    node_key: Arc<ActorKey>,
3402    entries: Arc<Mutex<LogWindow>>,
3403    /// Operator-curated file of eligible reviewer names, one per line.
3404    /// Read fresh on every draw, so editing it takes effect at once.
3405    /// `None` = no pool, and unassigned reviews stay unassigned.
3406    reviewer_pool: Option<std::path::PathBuf>,
3407    /// Optional operator conflict graph plus the maximum graph distance
3408    /// excluded from a review draw. The file is read fresh on every draw,
3409    /// like the reviewer pool. Edges are undirected operator pairs.
3410    reviewer_conflict_graph: Option<(std::path::PathBuf, usize)>,
3411    /// The persisted op log, for readers that have fallen behind the
3412    /// in-memory window. `None` (an in-memory log) means such a reader
3413    /// gets a loud gap error instead of a resync.
3414    log_path: Option<std::path::PathBuf>,
3415    /// Absolute operation-count ceiling for one signed batch.
3416    batch_limit: usize,
3417    /// Shared with the policy: when set, self-named reviewers are
3418    /// refused and every review goes through the node's draw.
3419    require_assignment: Arc<std::sync::atomic::AtomicBool>,
3420    /// Shared with the policy: the same refusal, but only for reviews
3421    /// that propose to land on a ref the operator marked protected.
3422    protected_refs: Arc<Mutex<Option<std::path::PathBuf>>>,
3423    /// Shared with the policy: when set, a protected ref only moves to a
3424    /// commit an approved review already named.
3425    require_review: Arc<std::sync::atomic::AtomicBool>,
3426    /// Shared with the policy: the operator's ACL file, consulted for
3427    /// [`crate::acl::Level::Own`] grants when a landing is gated (D42).
3428    acl_file: Arc<Mutex<Option<std::path::PathBuf>>>,
3429    /// Shared with the policy: the credential store passkey submissions
3430    /// are verified against (D39). Filled by [`Platform::attach_accounts`].
3431    passkeys: Arc<Mutex<Option<Arc<crate::accounts::Accounts>>>>,
3432    /// Shared with the policy: when set, only scoped ops are admitted.
3433    require_scope: Arc<std::sync::atomic::AtomicBool>,
3434    /// Shared with the policy: actor id → bound channel name.
3435    key_names: Arc<Mutex<KeyBindings>>,
3436    /// D24 T3 runtime projection, replayed from the signed log at startup.
3437    concentration: Arc<Mutex<ConcentrationState>>,
3438    /// D37 per-user workspace tally, replayed from the same log in the
3439    /// same pass. Shared with the policy, which advances it.
3440    workspace_tally: Arc<Mutex<crate::quota::WorkspaceTally>>,
3441    /// Sequence-ordered live reviews, allocated only under an explicit
3442    /// retention configuration.
3443    review_retention: Option<Arc<Mutex<ReviewRetentionState>>>,
3444    /// Opt-in D24 T4 audit. Sparse and separate from the signed op log:
3445    /// rejected requests never enter that log, while this evidence must.
3446    newcomer_audit: Option<Arc<Mutex<NewcomerAudit>>>,
3447    /// Opt-in D24 T2 operator classifications, re-read on every report so a
3448    /// fresh judgement lands without a restart. Absent means uncounted, not
3449    /// unclassified: coverage simply stays zero.
3450    review_adjudications: Option<std::path::PathBuf>,
3451    /// Serializes concurrent maintenance passes. User submissions still
3452    /// race normally through the sequencer; only duplicate pruning scans
3453    /// and archive batches are coalesced.
3454    review_prune_lock: Mutex<()>,
3455    /// Shared with the policy: the webhook delivery handle (D32), or
3456    /// `None` when the operator passed no `--hooks-file`.
3457    hooks: Arc<Mutex<Option<crate::hooks::Hooks>>>,
3458    /// The writer's own latency record for the traffic this node is
3459    /// serving, so the Phase-0 decision-latency gate is checked in
3460    /// production and not only by the test suite.
3461    lag: Arc<LagMeter>,
3462    /// Where drained gate breaches are appended, one JSON object per
3463    /// line. `None` keeps them in memory only, where the ring eventually
3464    /// drops the oldest (reported, never silent).
3465    lag_log: Option<std::path::PathBuf>,
3466    /// Last failure to write the lag log and how many writes have failed,
3467    /// surfaced in the report. A breach record that could not be written
3468    /// is itself an operational fact; swallowing it would make the lag log
3469    /// a check that cannot fail. The count does not reset on a later
3470    /// success, so a transient failure is still visible afterwards.
3471    lag_log_error: Mutex<(Option<String>, u64)>,
3472    // Kept alive for the daemon's lifetime; the writer thread exits with
3473    // the process.
3474    _sequencer: Sequencer,
3475}
3476
3477/// Exact owner-authorized binding the node records after materializing a
3478/// physical workspace.
3479pub struct AuthorizedChangeCreate<'a> {
3480    /// Stable logical contribution id.
3481    pub id: &'a str,
3482    /// Bound signing channel.
3483    pub owner: &'a str,
3484    /// Namespaced physical workspace id.
3485    pub workspace: &'a str,
3486    /// Full Git object id selected as the immutable base.
3487    pub base_hex: &'a str,
3488    /// Owner-scoped retry identity.
3489    pub idempotency_key: &'a str,
3490    /// Owner proof over the matching [`CreateAuthorization`].
3491    pub owner_sig: Witness,
3492    /// Directory prefixes the owner declared this change works within,
3493    /// covered by `owner_sig`. Empty means the whole tree.
3494    pub cone: Vec<String>,
3495}
3496
3497impl Platform {
3498    /// Replays `log` into a view and starts the admission sequencer over
3499    /// it with `registry` as the trusted key set plus the daemon's own
3500    /// `node_key` (registered automatically, for git-derived ops).
3501    ///
3502    /// # Errors
3503    ///
3504    /// Returns a description of any replay failure (a log written
3505    /// through this platform always replays cleanly).
3506    pub fn start(
3507        registry: Registry,
3508        log: Box<dyn OpLog>,
3509        node_key: ActorKey,
3510    ) -> Result<Self, String> {
3511        Self::start_inner(registry, log, node_key, None, None)
3512    }
3513
3514    /// [`Platform::start`] with explicit live-review retention.
3515    ///
3516    /// Retention is a startup choice because recovering FIFO request order
3517    /// belongs in the same replay pass that builds the view. The ordinary
3518    /// constructor allocates no tracker and performs no retention checks.
3519    ///
3520    /// # Errors
3521    ///
3522    /// Same as [`Platform::start`].
3523    pub fn start_with_review_retention(
3524        registry: Registry,
3525        log: Box<dyn OpLog>,
3526        node_key: ActorKey,
3527        retention: ReviewRetention,
3528    ) -> Result<Self, String> {
3529        Self::start_inner(registry, log, node_key, None, Some(retention))
3530    }
3531
3532    /// [`Platform::start`] with a trusted-keys file that is hot-reloaded
3533    /// (on mtime change) before signature verification: registering a key
3534    /// is appending a line, no restart, and removing one refuses its next
3535    /// submission. The file's contents replace the whole registry.
3536    ///
3537    /// # Errors
3538    ///
3539    /// Same as [`Platform::start`].
3540    pub fn start_reloading(
3541        registry: Registry,
3542        log: Box<dyn OpLog>,
3543        node_key: ActorKey,
3544        keys_file: Option<std::path::PathBuf>,
3545    ) -> Result<Self, String> {
3546        Self::start_inner(registry, log, node_key, keys_file, None)
3547    }
3548
3549    /// [`Platform::start_reloading`] with explicit live-review retention.
3550    ///
3551    /// # Errors
3552    ///
3553    /// Same as [`Platform::start`].
3554    pub fn start_reloading_with_review_retention(
3555        registry: Registry,
3556        log: Box<dyn OpLog>,
3557        node_key: ActorKey,
3558        keys_file: Option<std::path::PathBuf>,
3559        retention: ReviewRetention,
3560    ) -> Result<Self, String> {
3561        Self::start_inner(registry, log, node_key, keys_file, Some(retention))
3562    }
3563
3564    fn start_inner(
3565        mut registry: Registry,
3566        log: Box<dyn OpLog>,
3567        node_key: ActorKey,
3568        keys_file: Option<std::path::PathBuf>,
3569        retention: Option<ReviewRetention>,
3570    ) -> Result<Self, String> {
3571        registry
3572            .register(&node_key.public_key_bytes())
3573            .map_err(|e| format!("register node key: {e:?}"))?;
3574        let (view, review_retention, concentration, workspace_tally) =
3575            materialize_platform_state(log.as_ref(), retention)
3576                .map_err(|e| format!("replay: {e:?}"))?;
3577        let review_retention = review_retention.map(|state| Arc::new(Mutex::new(state)));
3578        let view = Arc::new(Mutex::new(view));
3579        let concentration = Arc::new(Mutex::new(concentration));
3580        let workspace_tally = Arc::new(Mutex::new(workspace_tally));
3581        // Fill the window from the tail only. Materialising the whole log
3582        // into a Vec and pushing each entry through the window cloned
3583        // every entry twice at startup and held a second full copy of the
3584        // log in memory alongside the log itself -- O(total ops) for a
3585        // window that keeps at most `cap`.
3586        let len = log.len();
3587        let start = len.saturating_sub(LOG_WINDOW_CAP as u64);
3588        let mut window = LogWindow {
3589            base: start,
3590            entries: std::collections::VecDeque::with_capacity(
3591                (len - start).min(LOG_WINDOW_CAP as u64) as usize,
3592            ),
3593            cap: LOG_WINDOW_CAP,
3594            by_signing: std::collections::HashMap::new(),
3595            by_hash: std::collections::HashMap::new(),
3596        };
3597        // Seeding needs each entry's hash, and the log stores it already:
3598        // every entry's `parent` is its predecessor's hash, and the last
3599        // one's is the log head. So a restart re-indexes the window
3600        // without hashing anything, and a scope signed just before the
3601        // restart is still admissible just after it.
3602        let mut pending: Option<OpEntry> = None;
3603        for i in start..len {
3604            let current = log.get(i);
3605            if let (Some(previous), Some(current)) = (pending.take(), current.as_ref()) {
3606                let hash = current
3607                    .parent
3608                    .clone()
3609                    .unwrap_or_else(|| previous.content_hash());
3610                window.push(previous, hash);
3611            }
3612            pending = current;
3613        }
3614        if let Some(last) = pending {
3615            let hash = log.head().unwrap_or_else(|| last.content_hash());
3616            window.push(last, hash);
3617        }
3618        let entries = Arc::new(Mutex::new(window));
3619        let keys_mtime = keys_file
3620            .as_ref()
3621            .and_then(|p| std::fs::metadata(p).and_then(|m| m.modified()).ok());
3622        // Name bindings are read from the same file at startup; a
3623        // malformed file here is not fatal because `start_reloading`
3624        // already accepted the caller's registry.
3625        let key_names = Arc::new(Mutex::new(match keys_file.as_ref() {
3626            Some(path) => crate::parse_keys_file(path).map_or_else(
3627                |_| KeyBindings::unavailable(true),
3628                |signers| KeyBindings::from_signers(&signers),
3629            ),
3630            None => KeyBindings::unavailable(false),
3631        }));
3632        let require_assignment = Arc::new(std::sync::atomic::AtomicBool::new(false));
3633        let protected_refs = Arc::new(Mutex::new(None));
3634        let acl_file = Arc::new(Mutex::new(None));
3635        let require_review = Arc::new(std::sync::atomic::AtomicBool::new(false));
3636        let require_scope = Arc::new(std::sync::atomic::AtomicBool::new(false));
3637        let hooks = Arc::new(Mutex::new(None));
3638        let passkeys = Arc::new(Mutex::new(None));
3639        let shared_journal = SharedJournal::default();
3640        let sequencer = Sequencer::spawn_with_journal(
3641            log,
3642            Box::new(ChoirPolicy {
3643                journal: shared_journal.clone(),
3644                subject: (None, None),
3645                require_assignment: require_assignment.clone(),
3646                protected_refs: protected_refs.clone(),
3647                acl_file: acl_file.clone(),
3648                require_review: require_review.clone(),
3649                require_scope: require_scope.clone(),
3650                registry,
3651                view: view.clone(),
3652                entries: entries.clone(),
3653                keys_file,
3654                keys_mtime,
3655                node_pub: node_key.public_key_bytes().to_vec(),
3656                node_id: node_key.actor_id(),
3657                key_names: key_names.clone(),
3658                concentration: concentration.clone(),
3659                review_retention: review_retention.clone(),
3660                hooks: hooks.clone(),
3661                workspace_tally: workspace_tally.clone(),
3662                passkeys: passkeys.clone(),
3663            }),
3664            Box::new(shared_journal.clone()),
3665        );
3666        let platform = Self {
3667            journal: shared_journal,
3668            handle: sequencer.handle(),
3669            lag: sequencer.lag(),
3670            lag_log: None,
3671            lag_log_error: Mutex::new((None, 0)),
3672            view,
3673            node_key: Arc::new(node_key),
3674            entries,
3675            reviewer_pool: None,
3676            reviewer_conflict_graph: None,
3677            log_path: None,
3678            batch_limit: DEFAULT_BATCH_OPS,
3679            require_assignment,
3680            protected_refs,
3681            acl_file,
3682            require_review,
3683            require_scope,
3684            passkeys,
3685            key_names,
3686            concentration,
3687            workspace_tally,
3688            review_retention,
3689            newcomer_audit: None,
3690            review_adjudications: None,
3691            review_prune_lock: Mutex::new(()),
3692            hooks,
3693            _sequencer: sequencer,
3694        };
3695        // Enabling a bound applies it at startup, not only after some
3696        // unrelated client happens to write. There are no external handles
3697        // yet, so any refusal here is a real maintenance/startup failure.
3698        if platform.review_retention.is_some() {
3699            let outcome = platform.prune_reviews();
3700            if !outcome.errors.is_empty() {
3701                return Err(format!(
3702                    "review retention failed during startup: {}",
3703                    serde_json::Value::Array(outcome.errors)
3704                ));
3705            }
3706        }
3707        Ok(platform)
3708    }
3709
3710    /// Enables sparse, durable D24 T4 newcomer measurement.
3711    ///
3712    /// `incumbent_actor_keys` is the trusted-key snapshot at activation;
3713    /// those actors are excluded because the audit cannot reconstruct their
3714    /// first attempt or time-to-first-acceptance. The audit records only a
3715    /// later actor's first verified signed-API attempt, its first eventual
3716    /// acceptance, and an optional appeal. Operator adjudications are JSONL
3717    /// rows in the separate file and are re-read for every report.
3718    ///
3719    /// Both files are created mode 0600. Neither changes the signed op log or
3720    /// any hash input.
3721    ///
3722    /// # Errors
3723    ///
3724    /// Unusable paths or an invalid existing audit file.
3725    pub fn with_newcomer_audit(
3726        mut self,
3727        audit_path: std::path::PathBuf,
3728        adjudications_path: std::path::PathBuf,
3729        incumbent_actor_keys: Vec<String>,
3730    ) -> Result<Self, String> {
3731        if let Some(parent) = adjudications_path.parent() {
3732            std::fs::create_dir_all(parent)
3733                .map_err(|e| format!("create adjudications directory: {e}"))?;
3734        }
3735        let mut options = std::fs::OpenOptions::new();
3736        options.create(true).append(true);
3737        #[cfg(unix)]
3738        {
3739            use std::os::unix::fs::OpenOptionsExt;
3740            options.mode(0o600);
3741        }
3742        options
3743            .open(&adjudications_path)
3744            .map_err(|e| format!("open adjudications: {e}"))?;
3745        #[cfg(unix)]
3746        {
3747            use std::os::unix::fs::PermissionsExt;
3748            std::fs::set_permissions(&adjudications_path, std::fs::Permissions::from_mode(0o600))
3749                .map_err(|e| format!("chmod adjudications: {e}"))?;
3750        }
3751        let audit = NewcomerAudit::open(
3752            &audit_path,
3753            adjudications_path,
3754            incumbent_actor_keys.into_iter().collect(),
3755        )?;
3756        self.newcomer_audit = Some(Arc::new(Mutex::new(audit)));
3757        Ok(self)
3758    }
3759
3760    /// Enables the D24 T2 operator classification file: versioned 0600 JSONL
3761    /// rows of `{"format_version":1,"review_id":…,"classification":…}` where
3762    /// classification is `valid`, `invalid`, `slop` or `unclear`.
3763    ///
3764    /// Separate from the signed op log on purpose, exactly like the T4
3765    /// adjudications: a judgement about a contribution is the operator's
3766    /// opinion, not a sequenced claim any actor can make. Keying it by review
3767    /// id rather than by verdict is what lets a classification outlive
3768    /// archiving, which discards the verdict bulk.
3769    ///
3770    /// Enabling it does not make T2 evaluable. The cohort still has no exit
3771    /// rule, the observation window and tripwire subject are unset, and census
3772    /// adjudication cannot survive the flood it would need to detect.
3773    ///
3774    /// # Errors
3775    ///
3776    /// The file or its directory cannot be created.
3777    pub fn with_review_adjudications(mut self, path: std::path::PathBuf) -> Result<Self, String> {
3778        if let Some(parent) = path.parent() {
3779            std::fs::create_dir_all(parent)
3780                .map_err(|e| format!("create review adjudications directory: {e}"))?;
3781        }
3782        let mut options = std::fs::OpenOptions::new();
3783        options.create(true).append(true);
3784        #[cfg(unix)]
3785        {
3786            use std::os::unix::fs::OpenOptionsExt;
3787            options.mode(0o600);
3788        }
3789        options
3790            .open(&path)
3791            .map_err(|e| format!("open review adjudications: {e}"))?;
3792        #[cfg(unix)]
3793        {
3794            use std::os::unix::fs::PermissionsExt;
3795            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
3796                .map_err(|e| format!("chmod review adjudications: {e}"))?;
3797        }
3798        self.review_adjudications = Some(path);
3799        Ok(self)
3800    }
3801
3802    /// Refuses any `RequestReview` that names its own reviewers, so the
3803    /// node's draw is the only path to a reviewer list (D24 layer 5,
3804    /// "the requester does not choose who reviews them" — enforced
3805    /// rather than merely offered).
3806    ///
3807    /// Requires a reviewer pool: without one no review can ever be
3808    /// assigned, so every request would stall unassigned.
3809    ///
3810    /// Scope: node-wide, the blunt instrument. For "only reviews landing
3811    /// somewhere that matters", see [`Platform::with_protected_refs`],
3812    /// which conditions on `RequestReview`'s `target_ref`.
3813    #[must_use]
3814    pub fn with_required_assignment(self) -> Self {
3815        self.require_assignment
3816            .store(true, std::sync::atomic::Ordering::Relaxed);
3817        self
3818    }
3819
3820    /// Points the platform at an operator-curated list of protected refs
3821    /// (one `<repo>:<refname>` pattern per line, `#` comments allowed, a
3822    /// single trailing `*` acting as a prefix glob). A `RequestReview`
3823    /// whose `target_ref` matches must go through the node's draw; one
3824    /// naming an unprotected ref, or naming no ref at all, may still pick
3825    /// its own reviewers.
3826    ///
3827    /// This is the per-review half of D24 layer 5 — "privilege-bearing"
3828    /// finally has a definition the code can read, rather than the
3829    /// node-wide approximation of [`Platform::with_required_assignment`].
3830    ///
3831    /// Requires a reviewer pool, for the same reason.
3832    ///
3833    /// On its own this only binds *reviews*: a requester who omits
3834    /// `target_ref` still escapes it. [`Platform::with_required_review`]
3835    /// is the other half, and closes that.
3836    #[must_use]
3837    pub fn with_protected_refs(self, path: std::path::PathBuf) -> Self {
3838        *self.protected_refs.lock().expect("protected refs lock") = Some(path);
3839        self
3840    }
3841
3842    /// Points the admission policy at the operator's ACL file, so a
3843    /// landing on a protected ref can ask who owns the repository (D42).
3844    ///
3845    /// Without this the ownership rule is simply absent and every
3846    /// protected ref keeps the approval-weight gate, which is the
3847    /// behaviour of every node built before D42. Ownership is opt-in per
3848    /// repository even once the file is attached: a repository nobody
3849    /// holds `own` over is unaffected.
3850    ///
3851    /// The path, not a parsed table, because the file is hot-reloadable
3852    /// and admission must see an ownership change without a restart —
3853    /// the same reason [`Platform::with_protected_refs`] takes a path.
3854    #[must_use]
3855    pub fn with_acl_file(self, path: std::path::PathBuf) -> Self {
3856        *self.acl_file.lock().expect("acl file lock") = Some(path);
3857        self
3858    }
3859
3860    /// Gives the admission policy the credential store, so a submission
3861    /// signed by an enrolled passkey can be verified (D39).
3862    ///
3863    /// A setter rather than a constructor argument because accounts and
3864    /// the platform are enabled independently, in either order, by
3865    /// separate flags. Until this is called a passkey submission is
3866    /// refused for want of a store, which is the same answer a node
3867    /// without self-service gives permanently.
3868    pub fn attach_accounts(&self, store: Arc<crate::accounts::Accounts>) {
3869        *self.passkeys.lock().expect("passkey store lock") = Some(store);
3870    }
3871
3872    /// Writes the credential's public key into a passkey submission, so
3873    /// the entry it becomes can be checked by someone holding nothing
3874    /// but the log (D45).
3875    ///
3876    /// The node supplies this rather than the client because the client
3877    /// *cannot*: `getPublicKey()` exists on a WebAuthn registration
3878    /// response only, so a browser holding an assertion has the
3879    /// credential id and no key. The value written is the one the
3880    /// admission check is about to verify against, read from the same
3881    /// store by the same `(channel, credential id)` pair.
3882    ///
3883    /// **The request cannot influence this field.** `decode_submission`
3884    /// never reads a `credential_key` from the body, so there is no
3885    /// mismatch to refuse and no path by which a caller asserts key
3886    /// material for a credential it does not hold. "Refuse a bad value"
3887    /// and "make a bad value unrepresentable" answer the same question;
3888    /// the second needs no test to stay true.
3889    ///
3890    /// **Absent signature, absent store and unknown credential are all
3891    /// left alone.** Each is a reason [`ChoirPolicy::verify_passkey`]
3892    /// refuses this submission a moment later, with a code and a repair.
3893    /// Refusing here as well would be a second refusal for one cause,
3894    /// raised from the shape of the request instead of by the check that
3895    /// owns the question.
3896    fn stamp_credential_key(&self, sub: &mut DecodedSubmission) {
3897        let Some(sig) = sub.author_sig.as_mut() else {
3898            return;
3899        };
3900        if sig.scheme_id() != choir_oplog::scheme::WEBAUTHN_ES256 {
3901            return;
3902        }
3903        let store = self.passkeys.lock().expect("passkey store lock").clone();
3904        if let Some(spki) = store.and_then(|s| s.passkey_spki(&sub.channel, &sig.key_id)) {
3905            sig.credential_key = Some(spki);
3906        }
3907    }
3908
3909    /// A protected ref only moves to a commit that an **approved** review
3910    /// with weight from two distinct operators already named as its
3911    /// destination, and can never be deleted.
3912    /// This is the landing half of the gate: with it, omitting
3913    /// `target_ref` stops being an escape and becomes a refusal, because
3914    /// the push itself is what gets checked.
3915    ///
3916    /// Requires [`Platform::with_protected_refs`] — with no list nothing
3917    /// is protected and the flag would do nothing.
3918    ///
3919    /// **No exemption for the node's own key.** Every git push reaches the
3920    /// sequencer as a node-signed `SetRef`, so exempting the node would
3921    /// exempt every push. The consequence is deliberate and operational:
3922    /// switching this on means this daemon's *own* repository can only be
3923    /// advanced through a review, `choirctl sync` included.
3924    ///
3925    /// Creating a protected ref is allowed (`prev == None`): there is no
3926    /// history to hijack yet, and since deletion is refused, "delete then
3927    /// re-create" is not a way back in.
3928    ///
3929    /// Not covered, and not silently implied: force-pushes and
3930    /// non-fast-forward updates are only constrained by the CAS `prev`
3931    /// git itself supplies. A sufficiently weighted approval of commit X
3932    /// authorizes landing X, whether or not X is a descendant of the
3933    /// current tip.
3934    #[must_use]
3935    pub fn with_required_review(self) -> Self {
3936        self.require_review
3937            .store(true, std::sync::atomic::Ordering::Relaxed);
3938        self
3939    }
3940
3941    /// Admits only ops whose author bound them to this log and a head
3942    /// still in the window — the replay defence, turned on.
3943    ///
3944    /// Off by default because it is a wire-compatibility break, not
3945    /// because unscoped is safe: an unscoped signature is admissible on
3946    /// any node that trusts the key, and admissible again on the node it
3947    /// came from as soon as CAS state returns to what it expected. Every
3948    /// client in this repository always sends a scope, so turning this
3949    /// on costs them nothing; a client that predates scopes stops
3950    /// working, which is the whole reason for the flag.
3951    #[must_use]
3952    pub fn with_required_scope(self) -> Self {
3953        self.require_scope
3954            .store(true, std::sync::atomic::Ordering::Relaxed);
3955        self
3956    }
3957
3958    /// The log identity a client signs a scope against: this node's
3959    /// actor id and the head it should name.
3960    #[must_use]
3961    pub fn scope_now(&self) -> (ContentHash, Option<ContentHash>) {
3962        let head = self.entries.lock().expect("entries lock").head_hash();
3963        (self.node_key.actor_id(), head)
3964    }
3965
3966    /// The round of proposals aimed at `branch` of `repo` (D68).
3967    ///
3968    /// `None` when the branch does not exist. A round with no proposals
3969    /// is `Some` with an empty list: "nobody has proposed anything" and
3970    /// "there is nothing to propose to" are different answers, and only
3971    /// the second is a misconfiguration.
3972    #[must_use]
3973    pub fn proposal_round(&self, repo: &str, branch: &str) -> Option<crate::queue::ProposalRound> {
3974        let view = self.view.lock().expect("view lock");
3975        crate::queue::ProposalRound::from_view(&view, repo, branch)
3976    }
3977
3978    /// A landing that moves `refname` from `base`, signed by this node.
3979    ///
3980    /// The scope is read per landing rather than captured here, because
3981    /// a round appends as it goes; see [`crate::queue::ScopeSource`].
3982    #[must_use]
3983    pub fn ref_landing(&self, refname: String, base: &str) -> crate::queue::RefLanding {
3984        let entries = Arc::clone(&self.entries);
3985        let node = self.node_key.actor_id();
3986        crate::queue::RefLanding::new(
3987            Arc::clone(&self.node_key),
3988            refname,
3989            base,
3990            Box::new(move || {
3991                (
3992                    node.clone(),
3993                    entries.lock().expect("entries lock").head_hash(),
3994                )
3995            }),
3996        )
3997    }
3998
3999    /// A check reporter that this node signs and scopes (D49, D68).
4000    ///
4001    /// The reason `set_check_reporter` was off by default and composed
4002    /// into nothing: the identity a check is reported under belongs to
4003    /// whoever runs the queue, and a bridge has no node-side identity to
4004    /// use. A node does -- its own key -- and its landings are in a log
4005    /// that is kept rather than thrown away, so a report against them is
4006    /// answerable later.
4007    #[must_use]
4008    pub fn check_reporter(
4009        &self,
4010        name: String,
4011        target_ref: Option<String>,
4012    ) -> choir_queue::CheckReporter {
4013        let key = Arc::clone(&self.node_key);
4014        let entries = Arc::clone(&self.entries);
4015        let node = self.node_key.actor_id();
4016        choir_queue::CheckReporter {
4017            channel: crate::queue::QUEUE_CHANNEL.to_string(),
4018            name,
4019            target_ref,
4020            seal: Some(Arc::new(move |channel: &str, op: &ViewOp| {
4021                let head = entries.lock().expect("entries lock").head_hash();
4022                let payload = op.clone().in_scope(node.clone(), head).to_payload();
4023                let sig = key.sign_submission(channel, &payload);
4024                (payload, Some(sig))
4025            })),
4026        }
4027    }
4028
4029    /// Runs one speculative round over this node's own log (D5, D68).
4030    ///
4031    /// `workdir` must be a **worktree of the repository this node
4032    /// serves**, made with `git worktree add --detach` against the bare
4033    /// repo, and owned outright by the queue: every speculative merge
4034    /// detaches it and forces it to a candidate state, so a tree
4035    /// anybody else is working in is the wrong argument.
4036    ///
4037    /// A worktree and not an independent clone, and this is not a
4038    /// preference. A landing names the merge commit the speculator
4039    /// built. An independent clone writes that commit into its own
4040    /// object store, where the served repository cannot see it, so the
4041    /// log would name a commit git does not have --- which
4042    /// [`Platform::reconcile_git_refs`] correctly reads as the view
4043    /// being wrong and compensates back to git's value, silently
4044    /// undoing every landing in the round. A worktree shares the object
4045    /// store, so the commit is already there when the op is submitted.
4046    ///
4047    /// `None` when the branch does not exist. Otherwise the report says
4048    /// what landed, in order, and every landing in it is an op in this
4049    /// node's log rather than a number the round kept to itself.
4050    pub fn run_proposal_queue(
4051        &self,
4052        repo: &str,
4053        branch: &str,
4054        workdir: &std::path::Path,
4055        ci: &mut dyn choir_queue::executor::CiExecutor,
4056        template: choir_queue::JobTemplate,
4057    ) -> Option<choir_queue::QueueReport> {
4058        let round = self.proposal_round(repo, branch)?;
4059        let mut queue = choir_queue::MergeQueue::with_speculator(
4060            &round.base,
4061            Box::new(choir_queue::git::GitSpeculator::new(workdir.to_path_buf())),
4062        );
4063        queue.set_job_template(template);
4064        queue.set_landing(Box::new(self.ref_landing(round.target_ref(), &round.base)));
4065        // Every verdict the round reaches is recorded (D49). The subject
4066        // is the speculative commit CI actually ran against, so a check
4067        // is answerable about a tree rather than about a round number.
4068        queue.set_check_reporter(
4069            self.check_reporter("ci/queue".to_string(), Some(round.target_ref())),
4070        );
4071        for change in round.changes() {
4072            queue.submit(change);
4073        }
4074        Some(self.drain_queue(&mut queue, ci))
4075    }
4076
4077    /// Drains a caller-built queue through this node's sequencer.
4078    ///
4079    /// The escape hatch under [`Platform::run_proposal_queue`], for a
4080    /// caller whose round is not simply "every proposal on this
4081    /// branch". The sequencer is the point: a landing recorded through
4082    /// anything else is not in the log this node serves.
4083    pub fn drain_queue(
4084        &self,
4085        queue: &mut choir_queue::MergeQueue,
4086        ci: &mut dyn choir_queue::executor::CiExecutor,
4087    ) -> choir_queue::QueueReport {
4088        queue.drain(ci, &self._sequencer)
4089    }
4090
4091    /// Points the platform at the JSON-lines file its op log persists to,
4092    /// so `/api/log?from=` can serve entries that have already been
4093    /// evicted from the in-memory window. Without it, a reader that has
4094    /// fallen further behind than the window is told so and cannot
4095    /// resync.
4096    ///
4097    /// The file is append-only, so reading a prefix while the writer
4098    /// thread appends is safe: already-written lines never change.
4099    #[must_use]
4100    pub fn with_log_path(mut self, path: std::path::PathBuf) -> Self {
4101        self.log_path = Some(path);
4102        self
4103    }
4104
4105    /// Sets the absolute operation-count ceiling for one batch request.
4106    #[must_use]
4107    pub fn with_batch_limit(mut self, max_ops: usize) -> Self {
4108        self.batch_limit = max_ops.max(1);
4109        self
4110    }
4111
4112    /// Replaces the actor-id → bound-name map from a freshly parsed
4113    /// trusted-keys file.
4114    ///
4115    /// Called from the daemon's accept loop when the file's mtime moves,
4116    /// so binding a name to an already-trusted key takes effect on the
4117    /// next request rather than waiting for some later signature failure.
4118    /// That matters because a binding is a *tightening*: a gate that
4119    /// applies at an unpredictable future moment is not a gate.
4120    pub fn set_key_names(&self, signers: &[crate::TrustedKey]) {
4121        *self.key_names.lock().expect("key names lock") = KeyBindings::from_signers(signers);
4122    }
4123
4124    /// Shrinks the in-memory `/api/log` window. Exists so tests can
4125    /// exercise eviction and the resync path without writing 100k ops.
4126    #[must_use]
4127    pub fn with_log_window_cap(self, cap: usize) -> Self {
4128        let mut window = self.entries.lock().expect("entries lock");
4129        window.cap = cap.max(1);
4130        // Trim to the new cap now rather than waiting for the next push.
4131        // Without this a platform started over an existing log stays
4132        // over-full until something is submitted, so `/api/log` would
4133        // serve entries from below `base` and a reader could not tell the
4134        // window had shrunk.
4135        window.trim();
4136        drop(window);
4137        self
4138    }
4139
4140    /// Points the platform at an operator-curated pool of eligible
4141    /// reviewer names (one per line, `#` comments allowed). With a pool
4142    /// set, a `RequestReview` carrying an empty reviewer list is
4143    /// answered by a node-signed [`OpKind::AssignReviewers`] drawn from
4144    /// the pool, excluding the requester — D24 layer 5, so a requester
4145    /// cannot pick a friendly reviewer.
4146    #[must_use]
4147    pub fn with_reviewer_pool(mut self, path: std::path::PathBuf) -> Self {
4148        self.reviewer_pool = Some(path);
4149        self
4150    }
4151
4152    /// Excludes reviewer operators whose shortest path from the requester
4153    /// in `path` is at most `max_distance`. Each non-comment line is one
4154    /// undirected `<operator> <operator>` edge. Operator names, not full
4155    /// `operator/agent` channels, belong in the graph.
4156    ///
4157    /// The graph is operator-supplied runtime policy rather than op-log
4158    /// state: changing it affects future draws without changing persisted
4159    /// operations or replay. It is read for every draw and fails closed;
4160    /// an unreadable or malformed graph leaves the review unassigned.
4161    /// Distance zero retains the existing same-operator exclusion.
4162    #[must_use]
4163    pub fn with_reviewer_conflict_graph(
4164        mut self,
4165        path: std::path::PathBuf,
4166        max_distance: usize,
4167    ) -> Self {
4168        self.reviewer_conflict_graph = Some((path, max_distance));
4169        self
4170    }
4171}
4172
4173/// The channel a push-derived op is attributed to, and how it was
4174/// established.
4175///
4176/// Extracted so the ref op and the review a magic push opens beside it
4177/// cannot disagree about who pushed. They are two ops about one act, and
4178/// a review drawn against a different channel than the ref it belongs to
4179/// would exclude the wrong operator from its own reviewer draw.
4180fn push_attribution(user: &str, cert: Option<(&str, &str)>) -> (String, Provenance) {
4181    // Verified push certificate ("G" = good signature) attributes the op
4182    // to the pusher's own key; otherwise the transport user. Either way
4183    // the node signs, and the payload says so (D41): the channel prefix
4184    // alone carried this class only by convention.
4185    match cert {
4186        Some(("G", signer)) if !signer.is_empty() => {
4187            (format!("key/{signer}"), Provenance::PushCertified)
4188        }
4189        _ => (crate::quota::channel_for(user), Provenance::PushTransport),
4190    }
4191}
4192
4193/// A proposal pushed to the magic refspec, as parsed from its refname
4194/// (D53).
4195///
4196/// Gerrit's `refs/for/<branch>` and AGit's after it are the spelling
4197/// every reviewer of a git-hosted project already knows, and the point
4198/// of borrowing it is that the client is `git` and nothing else: no
4199/// binary to install, no key to mint, one push.
4200///
4201/// Two departures from Gerrit, both deliberate:
4202///
4203/// 1. **The ref is really created.** Gerrit intercepts `refs/for/*` in a
4204///    server it owns end to end and, in its own documentation's words,
4205///    lies to the client about the result. Doing that over git's wire
4206///    protocol needs a `proc-receive` hook, which is a second hook type
4207///    and a second protocol; and a node that reported a ref it did not
4208///    write would be a node whose push receipts cannot be trusted. Here
4209///    the ref exists, holds the objects, and is sequenced like any other.
4210/// 2. **A topic is required.** `refs/for/main` alone is one ref shared by
4211///    everyone proposing onto `main`, so the second contributor's push
4212///    would be a non-fast-forward against the first one's proposal
4213///    rather than a proposal of their own. The topic is what makes the
4214///    ref theirs.
4215#[derive(Debug)]
4216struct MagicRef {
4217    /// Branch the proposal asks to land on, as a short name.
4218    onto: String,
4219    /// Review id, derived so that re-pushing the same topic reaches the
4220    /// same review rather than opening a second one.
4221    review_id: String,
4222}
4223
4224impl MagicRef {
4225    /// Parses `refs/for/<branch>/<topic>`, or `None` for any other ref.
4226    ///
4227    /// `Err` is reserved for a ref that *is* under `refs/for/` and
4228    /// cannot be used, because that is a pusher who meant to propose and
4229    /// needs telling why it did not work -- the one case where silence
4230    /// would look like success.
4231    fn parse(refname: &str) -> Option<Result<Self, String>> {
4232        let rest = refname.strip_prefix("refs/for/")?;
4233        let mut segments = rest.split('/').filter(|s| !s.is_empty());
4234        let (Some(onto), Some(first_topic)) = (segments.next(), segments.next()) else {
4235            return Some(Err(format!(
4236                "push to refs/for/<branch>/<topic>, not {refname}: a topic is what makes this \
4237                 proposal yours rather than one ref shared by everyone proposing onto that branch"
4238            )));
4239        };
4240        let topic: Vec<&str> = std::iter::once(first_topic).chain(segments).collect();
4241        let topic = topic.join("-");
4242        if !crate::provision::safe_segment(onto) {
4243            return Some(Err(format!(
4244                "`{onto}` is not a branch name this node can land on"
4245            )));
4246        }
4247        // The id reaches a URL (`/r/<repo>/review/<id>`), which admits
4248        // one path segment. The branch is folded in so that the same
4249        // topic proposed onto two branches is two reviews.
4250        let review_id = format!("for-{onto}-{topic}");
4251        if !crate::provision::safe_segment(&review_id) {
4252            return Some(Err(format!(
4253                "`{topic}` holds characters a review id cannot: use letters, digits, `-`, `_` \
4254                 or `.`"
4255            )));
4256        }
4257        Some(Ok(Self {
4258            onto: onto.to_string(),
4259            review_id,
4260        }))
4261    }
4262}
4263
4264impl Platform {
4265    /// Opens a review for a push to the magic refspec, if the ref was one.
4266    ///
4267    /// Runs after the ref op is already durable, and returns `Ok` when
4268    /// the ref was not a magic one -- so an ordinary push pays nothing
4269    /// and cannot fail here.
4270    ///
4271    /// A second push to the same topic does **not** re-request: a review
4272    /// is one long-lived object per proposal, re-posting a verdict is the
4273    /// re-review flow, and asking again would be refused. What advances
4274    /// is the ref, which is where a reviewer reads the current commit
4275    /// from.
4276    fn open_magic_review(
4277        &self,
4278        repo: &str,
4279        magic: &MagicRef,
4280        new_hex: &str,
4281        user: &str,
4282        cert: Option<(&str, &str)>,
4283    ) -> Result<(), String> {
4284        if self
4285            .view
4286            .lock()
4287            .expect("view lock")
4288            .reviews
4289            .contains_key(&magic.review_id)
4290        {
4291            return Ok(());
4292        }
4293        let target = ContentHash::from_git_oid(new_hex).ok_or("bad new oid")?;
4294        self.submit_ref_op(
4295            OpKind::RequestReview {
4296                id: magic.review_id.clone(),
4297                target,
4298                // Empty on purpose: the node draws them, and a pusher
4299                // cannot name their own reviewers here any more than
4300                // they can through the CLI.
4301                reviewers: Vec::new(),
4302                target_ref: Some(format!("{repo}:refs/heads/{}", magic.onto)),
4303            },
4304            user,
4305            cert,
4306        )?;
4307        // The draw is a second, separate submission -- the same shape
4308        // `/api/submit` uses, and for the same reason: it names
4309        // reviewers, so it cannot be folded into the request that has
4310        // not been admitted yet.
4311        //
4312        // A failed draw does not fail the push. The request stands and
4313        // is visibly unassigned, which is a state no verdict can
4314        // complete; refusing the push instead would throw away objects
4315        // the node has already accepted over an empty reviewer pool.
4316        let (channel, _) = push_attribution(user, cert);
4317        if let Err(reason) = self.assign_reviewers(&magic.review_id, &channel) {
4318            eprintln!(
4319                "choir: review {} opened unassigned: {reason}",
4320                magic.review_id
4321            );
4322        }
4323        Ok(())
4324    }
4325}
4326
4327/// Refuses a ref update that the pusher's grant does not reach (D60).
4328///
4329/// The other half of [`crate::acl::Level::Propose`], and the half that
4330/// has a refname to look at. [`crate::acl::git_requirement`] admits the
4331/// push at `propose` because git sends the ref list only after the
4332/// server has agreed to receive the pack, so there is nothing to check
4333/// at that boundary. The refname first exists here, when the
4334/// `pre-receive` hook reports it.
4335///
4336/// Checked against the **merged** table rather than
4337/// [`Platform::acl_now`]. That reader exists so `own` cannot be
4338/// self-issued (D42); `write` carries no such rule, and a grant issued
4339/// by self-service (D36) is as real as one the operator typed. Reading
4340/// the file alone here would refuse a legitimate pusher whose grant came
4341/// from an invite.
4342///
4343/// Refuses before anything is submitted, which is the rule
4344/// [`Platform::git_update`] already follows for a proposal ref it cannot
4345/// parse: git applies no ref until the hook exits zero, so a refusal at
4346/// this point leaves no op in the log and never enters the compensating
4347/// retraction pass that `Node::create_repo` documents.
4348///
4349/// `None` whenever the question does not arise: a body this does not
4350/// understand, or a pusher who holds `write` and is therefore not
4351/// limited to proposals.
4352pub(crate) fn proposal_denial(
4353    acl: &crate::acl::Effective,
4354    body: &[u8],
4355) -> Option<crate::acl::Denial> {
4356    let json: serde_json::Value = serde_json::from_slice(body).ok()?;
4357    let field = |key: &str| {
4358        json.get(key)
4359            .and_then(serde_json::Value::as_str)
4360            .unwrap_or_default()
4361    };
4362    let (repo, refname, user) = (field("repo"), field("refname"), field("user"));
4363    if repo.is_empty() || refname.is_empty() {
4364        return None;
4365    }
4366    if acl.allows_repo(user, repo, crate::acl::Level::Write) {
4367        return None;
4368    }
4369    let refuse = |reason: String| {
4370        Some(crate::acl::Denial {
4371            status: 403,
4372            reason,
4373        })
4374    };
4375    let Some(rest) = refname.strip_prefix("refs/for/") else {
4376        return refuse(format!(
4377            "`{user}` may propose to {repo} but not write {refname}; \
4378             push to refs/for/<branch>/{user}/<topic> to open a review instead"
4379        ));
4380    };
4381    // Under their own name, so two proposers cannot reach one ref. A
4382    // `propose` grant is the level given to somebody the repository does
4383    // not trust, and several of them hold it at once: without this,
4384    // whoever pushes second silently takes over the first one's proposal,
4385    // or deletes it.
4386    //
4387    // Read off the raw segments rather than `MagicRef`'s topic, because
4388    // that topic is joined with dashes and therefore lossy -- `a/b` and
4389    // `a-b` reach the same review id, and a rule about who owns a ref
4390    // must not be decided by a form that has already merged two names.
4391    //
4392    // A ref under `refs/for/` that this refuses is answered here rather
4393    // than by `git_update`'s own "a topic is what makes this proposal
4394    // yours": for this pusher, naming themselves is the missing part, and
4395    // the message that says so is the more useful of the two. A `write`
4396    // holder never reaches here and still gets the other one.
4397    let mut segments = rest.split('/').filter(|s| !s.is_empty());
4398    let (branch, owner) = (segments.next(), segments.next());
4399    if owner != Some(user) {
4400        let branch = branch.unwrap_or("<branch>");
4401        return refuse(format!(
4402            "`{user}` may propose to {repo} only under their own name; \
4403             push to refs/for/{branch}/{user}/<topic>"
4404        ));
4405    }
4406    None
4407}
4408
4409impl Platform {
4410    /// Routes one git ref update (from a repo's `update` hook) through
4411    /// the sequencer: CAS against the view, node-signed, totally ordered
4412    /// with API ops. Refs are namespaced `<repo>:<refname>`; git oids
4413    /// enter the envelope with their own codec ([`ContentHash::from_git_oid`]).
4414    ///
4415    /// A push to `refs/for/<branch>/<topic>` additionally opens a review
4416    /// targeting that branch, which is the whole magic-refspec path: one
4417    /// `git push`, no client but git, and reviewers drawn by the node.
4418    ///
4419    /// # Errors
4420    ///
4421    /// The policy's rejection reason (stale CAS = concurrent update git
4422    /// itself would also have refused), or the reason a `refs/for/` push
4423    /// could not be read as a proposal.
4424    pub fn git_update(
4425        &self,
4426        repo: &str,
4427        refname: &str,
4428        old_hex: &str,
4429        new_hex: &str,
4430        user: &str,
4431        cert: Option<(&str, &str)>,
4432    ) -> Result<(), String> {
4433        // Read before anything is submitted: a `refs/for/` push that
4434        // cannot be read as a proposal must be refused whole, not left
4435        // as a created ref with no review beside it.
4436        let magic = match MagicRef::parse(refname) {
4437            Some(Ok(magic)) => Some(magic),
4438            Some(Err(reason)) => return Err(reason),
4439            None => None,
4440        };
4441        let name = format!("{repo}:{refname}");
4442        let prev = if is_zero_oid(old_hex) {
4443            None
4444        } else {
4445            Some(ContentHash::from_git_oid(old_hex).ok_or("bad old oid")?)
4446        };
4447        let deleting = is_zero_oid(new_hex);
4448        let kind = if deleting {
4449            OpKind::DeleteRef { name, prev }
4450        } else {
4451            OpKind::SetRef {
4452                name,
4453                commit: ContentHash::from_git_oid(new_hex).ok_or("bad new oid")?,
4454                prev,
4455            }
4456        };
4457        self.submit_ref_op(kind, user, cert)?;
4458        // Only on a ref that now points somewhere. Deleting a proposal
4459        // ref withdraws the objects; it does not open a review on the
4460        // zero oid, and it deliberately does not close the review
4461        // either -- a review is append-only history, and abandoning one
4462        // is its own signed act.
4463        match magic {
4464            Some(magic) if !deleting => self.open_magic_review(repo, &magic, new_hex, user, cert),
4465            _ => Ok(()),
4466        }
4467    }
4468
4469    /// Retracts a ref op this push already had accepted, because the push
4470    /// as a whole is being refused and git will apply none of it.
4471    ///
4472    /// `pre-receive` submits one op per ref but git applies no ref until
4473    /// the hook exits zero, so a push whose third ref is refused has
4474    /// already put two ops in the durable log. Without this the view keeps
4475    /// refs git never created — and they cannot be pushed afterwards
4476    /// either, because the pusher's `old` is git's (absent) value while
4477    /// the view holds the stranded one, so every retry loses the CAS. The
4478    /// ref becomes permanently unpushable.
4479    ///
4480    /// The log is append-only, so the repair is a compensating op, not an
4481    /// erasure: the abort is part of the history rather than hidden from
4482    /// it. `old`/`new` are the same values the accepted op carried, so the
4483    /// inverse restores exactly what git still has.
4484    ///
4485    /// # Errors
4486    ///
4487    /// The policy's rejection reason — most likely a lost CAS, meaning
4488    /// something else moved the ref between the accept and this retraction
4489    /// and the stranded value is no longer what would be undone.
4490    pub fn git_abort(
4491        &self,
4492        repo: &str,
4493        refname: &str,
4494        old_hex: &str,
4495        new_hex: &str,
4496        user: &str,
4497        cert: Option<(&str, &str)>,
4498    ) -> Result<(), String> {
4499        let name = format!("{repo}:{refname}");
4500        // CAS on what the accepted op set, so a retraction that races a
4501        // real update loses instead of clobbering it.
4502        let prev = Some(ContentHash::from_git_oid(new_hex).ok_or("bad new oid")?);
4503        let kind = if is_zero_oid(old_hex) {
4504            // The push was creating the ref, so undoing it removes it.
4505            OpKind::DeleteRef { name, prev }
4506        } else {
4507            // It existed before: put it back where git still has it.
4508            OpKind::SetRef {
4509                name,
4510                commit: ContentHash::from_git_oid(old_hex).ok_or("bad old oid")?,
4511                prev,
4512            }
4513        };
4514        self.submit_ref_op(kind, user, cert)
4515    }
4516
4517    /// Brings every bare repo under `root` back into agreement with the
4518    /// view, which is the source of truth. Run at startup, before the
4519    /// node serves anything, so nothing races the repair.
4520    ///
4521    /// The hook's retraction path handles a push that is refused while
4522    /// the daemon is alive. This handles the rest, and it does so without
4523    /// needing to know which of them happened: power loss between the
4524    /// hook's 200 and git writing the ref, a per-ref failure *after*
4525    /// `pre-receive` passed (`receive.deny*`, an `update` hook, a write
4526    /// error), or a retraction that could not be delivered. All of them
4527    /// leave the same state, and it is the state this reads.
4528    ///
4529    /// Two repairs, chosen by whether git can honour the view:
4530    ///
4531    /// - the commit exists in the repo, so git is simply behind: the ref
4532    ///   is written. Git is the follower (D21 single-canonical), so
4533    ///   moving it is the defined direction.
4534    /// - the commit is absent — a refused push has its objects discarded
4535    ///   from the quarantine — so the view names something git can never
4536    ///   have: a compensating op puts the view back to git's value.
4537    ///
4538    /// A ref git holds and the view does not is **reported, never
4539    /// adopted**. Appending an op for it would launder an out-of-band
4540    /// `update-ref` into the signed log as though it had been submitted.
4541    pub fn reconcile_git_refs(&self, root: &std::path::Path) -> RefReconciliation {
4542        let mut report = RefReconciliation::default();
4543        for finding in self.survey_git_refs(root) {
4544            let full = finding.name();
4545            match finding.state {
4546                // Git is behind on a commit it already has, so move it.
4547                RefState::GitBehind => {
4548                    let Some(want) = &finding.log_oid else {
4549                        continue;
4550                    };
4551                    let path = root.join(&finding.repo);
4552                    match write_git_ref(&path, &finding.refname, want, finding.git_oid.as_deref()) {
4553                        Ok(()) => report.applied.push(full),
4554                        Err(e) => report.unreconciled.push(format!("{full}: {e}")),
4555                    }
4556                }
4557                // Git can never hold this value, so the log gives way.
4558                // `git_abort` builds exactly this inverse: back to git's
4559                // value, or gone if git has none.
4560                RefState::LogUnbackable => {
4561                    let Some(want) = &finding.log_oid else {
4562                        continue;
4563                    };
4564                    let old = finding
4565                        .git_oid
4566                        .clone()
4567                        .unwrap_or_else(|| "0".repeat(want.len()));
4568                    match self.git_abort(&finding.repo, &finding.refname, &old, want, "node", None)
4569                    {
4570                        Ok(()) => report.retracted.push(full),
4571                        Err(e) => report.unreconciled.push(format!("{full}: {e}")),
4572                    }
4573                }
4574                RefState::GitOnly | RefState::Unreadable => report
4575                    .unreconciled
4576                    .push(format!("{full}: {}", finding.reason)),
4577            }
4578        }
4579        report
4580    }
4581
4582    /// Every way the repos under `root` and the view currently disagree,
4583    /// with nothing written and no op appended.
4584    ///
4585    /// This is the half of [`Platform::reconcile_git_refs`] that can be
4586    /// run against a live node. The repair deliberately cannot: it writes
4587    /// refs, so it belongs before the first request, where nothing races
4588    /// it. Reading is safe at any time, and without it a divergence that
4589    /// appears while the daemon is up is invisible until the next
4590    /// restart — which is the difference between a monitor and an
4591    /// autopsy.
4592    ///
4593    /// Served by `GET /api/ref-agreement`, which is deliberately its own
4594    /// endpoint rather than a field on `/api/view`: this shells out to
4595    /// git once per repo, and `/api/view` is on the hot path.
4596    ///
4597    /// Scope, stated because [`RefState::GitOnly`] reads like a stronger
4598    /// claim than it is: only repos the log already names are compared.
4599    /// A repo with refs and no log entry at all is not surveyed, so this
4600    /// finds an out-of-band ref beside logged ones, not an entire
4601    /// smuggled repo.
4602    pub fn survey_git_refs(&self, root: &std::path::Path) -> Vec<RefFinding> {
4603        let mut findings = Vec::new();
4604        let view_refs: Vec<(String, ContentHash)> = self
4605            .view
4606            .lock()
4607            .expect("view lock")
4608            .refs
4609            .iter()
4610            .map(|(name, hash)| (name.clone(), hash.clone()))
4611            .collect();
4612
4613        // One `for-each-ref` per repo rather than one `rev-parse` per
4614        // ref: a view with thousands of refs would otherwise start the
4615        // daemon with thousands of subprocesses.
4616        let mut repos: BTreeMap<String, Vec<(String, ContentHash)>> = BTreeMap::new();
4617        for (name, hash) in view_refs {
4618            match name.split_once(':') {
4619                Some((repo, refname)) => repos
4620                    .entry(repo.to_string())
4621                    .or_default()
4622                    .push((refname.to_string(), hash)),
4623                // Not a git-derived ref (the API can set any name), so
4624                // there is no repo to compare it against.
4625                None => continue,
4626            }
4627        }
4628
4629        for (repo, refs) in repos {
4630            // Ref names come out of the log, which anyone admitted can
4631            // write to, so they get the same traversal guard as a repo
4632            // name off the wire.
4633            if repo.split('/').any(|c| c == ".." || c.is_empty()) || repo.starts_with('/') {
4634                findings.push(RefFinding::unreadable(&repo, "", "refused as a repo path"));
4635                continue;
4636            }
4637            let path = root.join(&repo);
4638            let Some(in_git) = read_git_refs(&path) else {
4639                findings.push(RefFinding::unreadable(
4640                    &repo,
4641                    "",
4642                    &format!(
4643                        "the log holds {} ref(s) for a repo that cannot be read here",
4644                        refs.len()
4645                    ),
4646                ));
4647                continue;
4648            };
4649            let wanted: BTreeSet<String> = refs.iter().map(|(name, _)| name.clone()).collect();
4650            for (refname, want) in refs {
4651                let Some(want_oid) = want.git_oid() else {
4652                    findings.push(RefFinding::unreadable(
4653                        &repo,
4654                        &refname,
4655                        "log value is not a git oid",
4656                    ));
4657                    continue;
4658                };
4659                let have = in_git.get(&refname).cloned();
4660                if have.as_deref() == Some(want_oid.as_str()) {
4661                    continue;
4662                }
4663                // Whether git *can* be moved to the log is the whole
4664                // difference between the two repairs, so it is decided
4665                // here, while reading, and not again while writing.
4666                let state = if object_exists(&path, &want_oid) {
4667                    RefState::GitBehind
4668                } else {
4669                    RefState::LogUnbackable
4670                };
4671                findings.push(RefFinding {
4672                    repo: repo.clone(),
4673                    refname,
4674                    log_oid: Some(want_oid),
4675                    git_oid: have,
4676                    reason: match state {
4677                        RefState::GitBehind => "git is behind a commit it already has".into(),
4678                        _ => "the log names a commit this repo does not have".into(),
4679                    },
4680                    state,
4681                });
4682            }
4683            for (refname, oid) in &in_git {
4684                if !wanted.contains(refname) {
4685                    // A remote-tracking ref is this repo's own record of
4686                    // what it pushed elsewhere — the on-box follower feed
4687                    // writes refs/remotes/<follower>/* on every push — not
4688                    // canonical state, so its absence from the log is not a
4689                    // divergence. Skipped only on this side: a log that
4690                    // *does* name one is still compared above, and still
4691                    // retracted if git cannot back it.
4692                    if refname.starts_with("refs/remotes/") {
4693                        continue;
4694                    }
4695                    findings.push(RefFinding {
4696                        repo: repo.clone(),
4697                        refname: refname.clone(),
4698                        log_oid: None,
4699                        git_oid: Some(oid.clone()),
4700                        state: RefState::GitOnly,
4701                        reason: "in git, not in the log".into(),
4702                    });
4703                }
4704            }
4705        }
4706        findings
4707    }
4708
4709    /// Signs a git-derived ref op as the node and submits it, attributing
4710    /// it to the pusher's own key when the push certificate verified.
4711    fn submit_ref_op(
4712        &self,
4713        kind: OpKind,
4714        user: &str,
4715        cert: Option<(&str, &str)>,
4716    ) -> Result<(), String> {
4717        let (node, head) = self.scope_now();
4718        // Verified push certificate ("G" = good signature) attributes
4719        // the op to the pusher's own key; otherwise the transport user.
4720        // Either way the node signs, and the payload says so (D41): the
4721        // channel prefix alone carried this class only by convention.
4722        let (workspace, provenance) = push_attribution(user, cert);
4723        let payload = ViewOp::new(kind)
4724            .in_scope(node, head)
4725            .with_provenance(provenance)
4726            .to_payload();
4727        let sig = self.node_key.sign_submission(&workspace, &payload);
4728        self.handle
4729            .try_submit(&workspace, payload, Some(sig))
4730            .map(|_| ())?;
4731        // Every ref movement the node authors ends in an attestation of
4732        // where the refs now stand. Per accepted ref rather than per
4733        // push, because the pre-receive protocol has no end-of-push
4734        // signal to hang a single emission on.
4735        self.record_snapshot();
4736        Ok(())
4737    }
4738
4739    /// Attests the current ref-state (D25): submits a node-signed
4740    /// [`OpKind::RecordRefSnapshot`] of the view as it stands, and on
4741    /// admission projects the snapshot's canonical bytes to
4742    /// `refs.snapshot` beside the op log — the detached copy a backup
4743    /// pulls with the log, byte-identical to the payload in it.
4744    ///
4745    /// Best-effort on both legs. A lost submission race means another
4746    /// writer moved the view between read and submit, and that writer's
4747    /// own ref op ends in another attestation, so the chain catches up
4748    /// without retries here. The file write is tmp-plus-rename with a
4749    /// per-call tmp name: concurrent admissions cannot tear the file,
4750    /// and if their renames land out of order it briefly holds the older
4751    /// of two valid snapshots until the next attestation replaces it.
4752    fn record_snapshot(&self) {
4753        let snapshot = self.view.lock().expect("view lock").snapshot();
4754        let bytes = snapshot.canonical_bytes();
4755        let (node, head) = self.scope_now();
4756        let payload = ViewOp::new(OpKind::RecordRefSnapshot { snapshot })
4757            .in_scope(node, head)
4758            .to_payload();
4759        let channel = "node/snapshot";
4760        let sig = self.node_key.sign_submission(channel, &payload);
4761        if self.handle.try_submit(channel, payload, Some(sig)).is_err() {
4762            return;
4763        }
4764        let Some(log) = &self.log_path else { return };
4765        static TMP: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
4766        let n = TMP.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
4767        let tmp = log.with_file_name(format!("refs.snapshot.{n}.tmp"));
4768        if std::fs::write(&tmp, &bytes)
4769            .and_then(|()| std::fs::rename(&tmp, log.with_file_name("refs.snapshot")))
4770            .is_err()
4771        {
4772            // The attestation is in the log either way; a missing
4773            // detached copy fails the next backup pull loudly, which is
4774            // the reader that cares.
4775            std::fs::remove_file(&tmp).ok();
4776        }
4777    }
4778
4779    /// Points `workspace` at git oid `head_hex` with a node-signed op,
4780    /// using the view's current head as the CAS `prev` (a lost race is
4781    /// a sequencer rejection, not a clobber). `attribution` is the
4782    /// submission channel (e.g. `git/<user>`), as for git-derived ops.
4783    ///
4784    /// # Errors
4785    ///
4786    /// Bad oid, or the policy's rejection reason.
4787    pub fn set_workspace_head(
4788        &self,
4789        workspace: &str,
4790        head_hex: &str,
4791        attribution: &str,
4792    ) -> Result<(), String> {
4793        let commit = ContentHash::from_git_oid(head_hex).ok_or("bad head oid")?;
4794        let prev = self
4795            .view
4796            .lock()
4797            .expect("view lock")
4798            .workspaces
4799            .get(workspace)
4800            .cloned();
4801        let (node, head) = self.scope_now();
4802        let payload = ViewOp::new(OpKind::SetWorkspaceHead {
4803            workspace: workspace.to_string(),
4804            commit,
4805            prev,
4806        })
4807        .in_scope(node, head)
4808        .to_payload();
4809        let sig = self.node_key.sign_submission(attribution, &payload);
4810        self.handle
4811            .try_submit(attribution, payload, Some(sig))
4812            .map(|_| ())
4813    }
4814
4815    /// Atomically creates a stable change and registers its workspace at
4816    /// an exact Git revision with a node-signed operation.
4817    pub fn create_change(
4818        &self,
4819        request: AuthorizedChangeCreate<'_>,
4820        attribution: &str,
4821    ) -> Result<choir_sequencer::Accepted, String> {
4822        let AuthorizedChangeCreate {
4823            id,
4824            owner,
4825            workspace,
4826            base_hex,
4827            idempotency_key,
4828            owner_sig,
4829            cone,
4830        } = request;
4831        let base_revision = ContentHash::from_git_oid(base_hex).ok_or("bad base revision")?;
4832        let payload = ViewOp::new(OpKind::CreateChange {
4833            id: id.to_string(),
4834            owner: owner.to_string(),
4835            workspace: workspace.to_string(),
4836            base_revision,
4837            idempotency_key: idempotency_key.to_string(),
4838            owner_sig: Some(owner_sig),
4839            cone,
4840        })
4841        .to_payload();
4842        let sig = self.node_key.sign_submission(attribution, &payload);
4843        match self
4844            .handle
4845            .try_submit(attribution, payload.clone(), Some(sig))
4846        {
4847            Ok(accepted) => Ok(accepted),
4848            Err(reason) => match self
4849                .entries
4850                .lock()
4851                .expect("entries lock")
4852                .already_applied(attribution, &payload)
4853            {
4854                Some((seq, hash)) => Ok(choir_sequencer::Accepted {
4855                    seq,
4856                    hash,
4857                    decision_latency: Duration::ZERO,
4858                }),
4859                None => Err(reason),
4860            },
4861        }
4862    }
4863
4864    /// Decodes the exact owner-signed create binding before any physical
4865    /// workspace is copied. Sequencer admission verifies the embedded
4866    /// signature before recording the node-authored change operation.
4867    pub fn decode_create_change_request(
4868        &self,
4869        request: &serde_json::Value,
4870        expected_id: &str,
4871        expected_owner: &str,
4872        expected_workspace: &str,
4873        expected_revision: &ContentHash,
4874        expected_idempotency_key: &str,
4875    ) -> Result<(Witness, Vec<String>), String> {
4876        let (sub, cone) = decode_create_submission(
4877            request,
4878            expected_id,
4879            expected_owner,
4880            expected_workspace,
4881            expected_revision,
4882            expected_idempotency_key,
4883        )?;
4884        // The cone comes back out of the *signed* authorization and
4885        // never out of the request body around it. That is the whole
4886        // guarantee: the node reports a scope its owner signed, or it
4887        // reports none (D50).
4888        Ok((
4889            sub.author_sig
4890                .expect("decode_submission always returns a signature"),
4891            cone,
4892        ))
4893    }
4894
4895    /// Submits a node-authored [`OpKind::ArchiveChange`] carrying the
4896    /// owner's signed authorization, after verifying that the signed
4897    /// payload names exactly the resource the endpoint already moved.
4898    pub fn submit_archive_change(
4899        &self,
4900        request: &serde_json::Value,
4901        expected_id: &str,
4902        expected_workspace: &str,
4903        expected_revision: &ContentHash,
4904        attribution: &str,
4905    ) -> Result<choir_sequencer::Accepted, String> {
4906        let sub =
4907            decode_archive_submission(request, expected_id, expected_workspace, expected_revision)?;
4908        let payload = ViewOp::new(OpKind::ArchiveChange {
4909            id: expected_id.to_string(),
4910            workspace: expected_workspace.to_string(),
4911            prev_revision: expected_revision.clone(),
4912            owner: sub.channel,
4913            owner_sig: sub
4914                .author_sig
4915                .expect("decode_submission always returns a signature"),
4916        })
4917        .to_payload();
4918        let signature = self.node_key.sign_submission(attribution, &payload);
4919        self.handle
4920            .try_submit(attribution, payload, Some(signature))
4921    }
4922
4923    /// Checks the archive request's signed payload shape before the
4924    /// filesystem is renamed. Signature and current-state admission still
4925    /// happen on the sequencer after the rename, with rollback on refusal.
4926    pub fn validate_archive_change_request(
4927        &self,
4928        request: &serde_json::Value,
4929        expected_id: &str,
4930        expected_workspace: &str,
4931        expected_revision: &ContentHash,
4932    ) -> Result<(), String> {
4933        decode_archive_submission(request, expected_id, expected_workspace, expected_revision)
4934            .map(|_| ())
4935    }
4936
4937    /// Original operation identity for an identical completed archive
4938    /// request still present in the durable log window.
4939    #[must_use]
4940    pub fn archive_change_receipt(
4941        &self,
4942        request: &serde_json::Value,
4943        expected_id: &str,
4944        expected_workspace: &str,
4945        expected_revision: &ContentHash,
4946        attribution: &str,
4947    ) -> Option<(u64, ContentHash)> {
4948        let sub =
4949            decode_archive_submission(request, expected_id, expected_workspace, expected_revision)
4950                .ok()?;
4951        let payload = ViewOp::new(OpKind::ArchiveChange {
4952            id: expected_id.to_string(),
4953            workspace: expected_workspace.to_string(),
4954            prev_revision: expected_revision.clone(),
4955            owner: sub.channel,
4956            owner_sig: sub.author_sig?,
4957        })
4958        .to_payload();
4959        self.entries
4960            .lock()
4961            .expect("entries lock")
4962            .already_applied(attribution, &payload)
4963    }
4964
4965    /// Current materialized state for one stable change.
4966    #[must_use]
4967    pub fn change_state(&self, id: &str) -> Option<ChangeState> {
4968        self.view
4969            .lock()
4970            .expect("view lock")
4971            .changes
4972            .get(id)
4973            .cloned()
4974    }
4975
4976    /// Finds the change created by one owner-scoped idempotency key.
4977    #[must_use]
4978    pub fn change_for_idempotency(&self, owner: &str, key: &str) -> Option<(String, ChangeState)> {
4979        self.view
4980            .lock()
4981            .expect("view lock")
4982            .changes
4983            .iter()
4984            .find(|(_, change)| change.owner == owner && change.idempotency_key == key)
4985            .map(|(id, change)| (id.clone(), change.clone()))
4986    }
4987
4988    /// Inactive change generations that previously used `workspace`.
4989    /// This supports migration of the original unversioned archive path
4990    /// when a later generation reuses a deterministic workspace name.
4991    #[must_use]
4992    pub fn archived_change_ids_for_workspace(&self, workspace: &str) -> Vec<String> {
4993        self.view
4994            .lock()
4995            .expect("view lock")
4996            .changes
4997            .iter()
4998            .filter(|(_, change)| {
4999                change.workspace_id == workspace && change.active_workspace.is_none()
5000            })
5001            .map(|(id, _)| id.clone())
5002            .collect()
5003    }
5004
5005    /// Current exact head of an active workspace.
5006    #[must_use]
5007    pub fn workspace_head(&self, workspace: &str) -> Option<ContentHash> {
5008        self.view
5009            .lock()
5010            .expect("view lock")
5011            .workspaces
5012            .get(workspace)
5013            .cloned()
5014    }
5015
5016    /// How many workspaces `channel` currently holds (D37).
5017    ///
5018    /// Read from the projection replayed out of the op log, so a node
5019    /// that has just restarted answers the same number it answered
5020    /// before — the property a per-user ceiling is worthless without.
5021    #[must_use]
5022    pub fn workspaces_held_by(&self, channel: &str) -> usize {
5023        self.workspace_tally
5024            .lock()
5025            .expect("workspace tally lock")
5026            .held_by(channel)
5027    }
5028
5029    /// Every workspace the D37 tally is tracking, sorted.
5030    ///
5031    /// Exposed for the test that pins the tally against
5032    /// [`View::workspaces`]: the two are folded from the same operations
5033    /// and a divergence between them is the tripwire on the D37 register
5034    /// row.
5035    #[must_use]
5036    pub fn tallied_workspaces(&self) -> Vec<String> {
5037        self.workspace_tally
5038            .lock()
5039            .expect("workspace tally lock")
5040            .workspaces()
5041            .map(ToString::to_string)
5042            .collect()
5043    }
5044
5045    /// Draws reviewers for unassigned review `id` and records them with
5046    /// a node-signed op.
5047    ///
5048    /// Candidates are the pool minus everyone sharing the requester's
5049    /// **operator** and, when configured, every operator within the chosen
5050    /// conflict-graph distance. The draw takes at most one reviewer per
5051    /// operator so `REQUIRED_APPROVAL_WEIGHT` reviewers means that many
5052    /// *independent* ones under the configured policy.
5053    ///
5054    /// Excluding only the requester's own name was the original rule and
5055    /// it does not survive the multi-operator case, which is the normal
5056    /// one: an operator running three agents in the pool satisfies
5057    /// two-person integrity by themselves, and can manufacture more
5058    /// agreement by registering more agents. That is the Sybil move D24
5059    /// says must be blocked at the operator level, so the exclusion has
5060    /// to be at that level too.
5061    ///
5062    /// A pool that cannot supply `REQUIRED_APPROVAL_WEIGHT` distinct operators
5063    /// draws fewer rather than doubling up — a visibly under-assigned
5064    /// review beats one that looks independent and is not.
5065    ///
5066    /// # Errors
5067    ///
5068    /// No pool configured, an unreadable pool or conflict graph, malformed
5069    /// graph data, no candidate outside the conflict distance, or the
5070    /// sequencer's rejection reason (e.g. the review was assigned by a
5071    /// concurrent request).
5072    pub fn assign_reviewers(&self, id: &str, requester: &str) -> Result<Vec<String>, String> {
5073        let path = self
5074            .reviewer_pool
5075            .as_ref()
5076            .ok_or("no reviewer pool configured")?;
5077        let text = std::fs::read_to_string(path).map_err(|e| format!("read reviewer pool: {e}"))?;
5078        let mine = reviewer_operator(requester);
5079        let excluded_operators = match &self.reviewer_conflict_graph {
5080            Some((path, max_distance)) => operators_within_distance(path, mine, *max_distance)?,
5081            None => BTreeSet::from([mine.to_string()]),
5082        };
5083        let mut pool: Vec<String> = text
5084            .lines()
5085            .map(str::trim)
5086            .filter(|l| {
5087                !l.is_empty()
5088                    && !l.starts_with('#')
5089                    && !excluded_operators.contains(reviewer_operator(l))
5090            })
5091            .map(String::from)
5092            .collect();
5093        if pool.is_empty() {
5094            return Err(format!(
5095                "reviewer pool has nobody outside the configured conflict distance from {mine:?}"
5096            ));
5097        }
5098        // Partial Fisher-Yates with a hand-rolled xorshift (no rand
5099        // dep). The draw is node-side and recorded in the log, so
5100        // replay reproduces it from the op, not from this seed.
5101        let mut state = seed_from_clock() ^ fnv1a(id.as_bytes());
5102        let mut drawn: Vec<String> = Vec::new();
5103        // Owned, not borrowed: the shuffle mutates `pool` under it.
5104        let mut seen_operators: BTreeSet<String> = BTreeSet::new();
5105        seen_operators.insert(mine.to_string());
5106        for i in 0..pool.len() {
5107            if drawn.len() == REQUIRED_APPROVAL_WEIGHT {
5108                break;
5109            }
5110            state ^= state << 13;
5111            state ^= state >> 7;
5112            state ^= state << 17;
5113            let j = i + (state as usize) % (pool.len() - i);
5114            pool.swap(i, j);
5115            // One seat per operator: a second agent from an operator
5116            // already drawn would add a name, not a second opinion.
5117            if seen_operators.insert(reviewer_operator(&pool[i]).to_string()) {
5118                drawn.push(pool[i].clone());
5119            }
5120        }
5121        let mut pool = drawn;
5122        pool.sort();
5123
5124        let (node, head) = self.scope_now();
5125        let payload = ViewOp::new(OpKind::AssignReviewers {
5126            id: id.to_string(),
5127            reviewers: pool.clone(),
5128        })
5129        .in_scope(node, head)
5130        .to_payload();
5131        let channel = "node/assign";
5132        let sig = self.node_key.sign_submission(channel, &payload);
5133        self.handle.try_submit(channel, payload, Some(sig))?;
5134        Ok(pool)
5135    }
5136
5137    /// Emits enough FIFO `ArchiveReview` ops to bring live review detail
5138    /// back to the configured count when eligible reviews exist.
5139    ///
5140    /// Complete reviews are eligible without a clock. Incomplete reviews
5141    /// are eligible only after the operator-selected lapse age. Multiple
5142    /// archives are offered together so they share durability barriers.
5143    fn prune_reviews(&self) -> ReviewPruneOutcome {
5144        let Some(retention) = &self.review_retention else {
5145            return ReviewPruneOutcome::default();
5146        };
5147        let _pass = self.review_prune_lock.lock().expect("review prune lock");
5148
5149        let candidates: Vec<(String, bool)> = {
5150            // Same lock order as `ChoirPolicy::accepted`: view first,
5151            // tracker second. Holding both makes the count and eligibility
5152            // one writer-consistent snapshot; they are released before any
5153            // submission is sent back to the sequencer.
5154            let view = self.view.lock().expect("view lock");
5155            let mut retention = retention.lock().expect("review retention lock");
5156            if !retention.worth_a_pass() {
5157                return ReviewPruneOutcome::default();
5158            }
5159            // Cleared while the lock is still held, so an op observed from
5160            // here on re-arms the flag instead of being swallowed by the
5161            // pass that did not see it.
5162            retention.prunable_changed = false;
5163            let live_count = retention
5164                .live
5165                .iter()
5166                .filter(|tracked| {
5167                    view.reviews.get(&tracked.id).is_some_and(|review| {
5168                        matches!(review.status, choir_view::ReviewStatus::Live)
5169                    })
5170                })
5171                .count();
5172            let mut needed = live_count.saturating_sub(retention.config.max_live);
5173            // Count-only retention never reads a clock. A timestamp is
5174            // created and consulted only under the explicit lapse policy.
5175            let now = retention.config.lapse_after.map(|_| Instant::now());
5176            let mut candidates = Vec::with_capacity(needed);
5177            for tracked in &retention.live {
5178                if needed == 0 {
5179                    break;
5180                }
5181                let Some(review) = view.reviews.get(&tracked.id) else {
5182                    continue;
5183                };
5184                if !matches!(review.status, choir_view::ReviewStatus::Live) {
5185                    continue;
5186                }
5187                let lapsed = if review.complete() {
5188                    false
5189                } else if retention.config.lapse_after.is_some_and(|age| {
5190                    now.zip(tracked.observed_at)
5191                        .is_some_and(|(now, observed_at)| {
5192                            now.saturating_duration_since(observed_at) >= age
5193                        })
5194                }) {
5195                    true
5196                } else {
5197                    continue;
5198                };
5199                candidates.push((tracked.id.clone(), lapsed));
5200                needed -= 1;
5201            }
5202            candidates
5203        };
5204
5205        if candidates.is_empty() {
5206            return ReviewPruneOutcome::default();
5207        }
5208
5209        let channel = "node/archive";
5210        let (node, head) = self.scope_now();
5211        let submissions: Vec<Submission> = candidates
5212            .iter()
5213            .map(|(id, lapsed)| {
5214                let payload = ViewOp::new(OpKind::ArchiveReview {
5215                    id: id.clone(),
5216                    lapsed: *lapsed,
5217                })
5218                .in_scope(node.clone(), head.clone())
5219                .to_payload();
5220                Submission {
5221                    channel: channel.to_string(),
5222                    author_sig: Some(self.node_key.sign_submission(channel, &payload)),
5223                    payload,
5224                }
5225            })
5226            .collect();
5227        let results = self.handle.try_submit_many(submissions);
5228        let mut outcome = ReviewPruneOutcome::default();
5229        for ((id, _), result) in candidates.into_iter().zip(results) {
5230            match result {
5231                Ok(_) => outcome.archived.push(id),
5232                Err(reason) => {
5233                    let mut error = Rejection::decode(&reason).to_json();
5234                    error["review"] = serde_json::json!(id);
5235                    outcome.errors.push(error);
5236                }
5237            }
5238        }
5239        outcome
5240    }
5241
5242    fn add_retention_outcome(&self, response: &mut serde_json::Value) {
5243        let outcome = self.prune_reviews();
5244        if !outcome.archived.is_empty() {
5245            response["archived_reviews"] = serde_json::json!(outcome.archived);
5246        }
5247        if !outcome.errors.is_empty() {
5248            response["retention_errors"] = serde_json::json!(outcome.errors);
5249        }
5250    }
5251
5252    fn add_newcomer_outcome(
5253        &self,
5254        response: &mut serde_json::Value,
5255        sub: &DecodedSubmission,
5256        started_at_unix_ms: u64,
5257        accepted: bool,
5258        rejection_code: Option<&str>,
5259    ) {
5260        let Some(audit) = &self.newcomer_audit else {
5261            return;
5262        };
5263        let Some(actor_key) = sub
5264            .author_sig
5265            .as_ref()
5266            .map(|signature| signature.key_id.as_str())
5267        else {
5268            return;
5269        };
5270        match audit.lock().expect("newcomer audit lock").observe(
5271            actor_key,
5272            started_at_unix_ms,
5273            accepted,
5274            rejection_code,
5275        ) {
5276            Ok(Some(attempt_id)) => {
5277                response["newcomer_attempt_id"] = serde_json::json!(attempt_id);
5278            }
5279            Ok(None) => {}
5280            Err(error) => {
5281                response["newcomer_audit_error"] = serde_json::json!(error);
5282            }
5283        }
5284    }
5285
5286    /// Whether the sequencer has failed a durability barrier and stopped
5287    /// accepting.
5288    ///
5289    /// The daemon's accept loop polls this so a node that can no longer
5290    /// persist exits rather than staying up refusing everything. Process
5291    /// supervision only restarts a process that *exits*, so without this
5292    /// a transient fsync error is permanent downtime that looks like
5293    /// uptime.
5294    #[must_use]
5295    pub fn durability_failed(&self) -> bool {
5296        self.handle.durability_failed()
5297    }
5298
5299    /// Records every admission decision to `path` as JSONL.
5300    ///
5301    /// Derived data (see [`journal`]): the file is appended to from its
5302    /// own thread, never read back, and its loss or truncation changes
5303    /// no decision this node makes. Safe to rotate by moving it aside;
5304    /// the daemon keeps writing to the open handle until restarted.
5305    ///
5306    /// # Errors
5307    ///
5308    /// If `path` cannot be opened for append.
5309    pub fn with_journal(self, path: &std::path::Path) -> std::io::Result<Self> {
5310        self.journal
5311            .install(Box::new(journal::FileJournal::create(path)?));
5312        Ok(self)
5313    }
5314
5315    /// Appends gate breaches to `path`, one JSON object per line.
5316    ///
5317    /// Separate from the op log on purpose: a breach is an observation
5318    /// about this node's storage and load, not a fact about the ordered
5319    /// history, and it must not change a hash anyone else replays.
5320    #[must_use]
5321    pub fn with_lag_log(mut self, path: std::path::PathBuf) -> Self {
5322        self.lag_log = Some(path);
5323        self
5324    }
5325
5326    /// Enables outbound ref-landed webhooks (D32) from the subscription
5327    /// file `config`, recording every delivery attempt in `log`.
5328    ///
5329    /// Starts one delivery thread. The sequencer's writer thread never
5330    /// waits on it: it offers events to a bounded queue and drops
5331    /// (counted, and written to `log`) when that queue is full, because
5332    /// a receiver this node does not control must not be able to delay
5333    /// op admission. See [`crate::hooks`].
5334    ///
5335    /// # Errors
5336    ///
5337    /// Returns a message when the subscription file cannot be read or
5338    /// does not parse, or when the delivery thread cannot be started.
5339    pub fn with_hooks(
5340        self,
5341        config: std::path::PathBuf,
5342        log: std::path::PathBuf,
5343    ) -> Result<Self, String> {
5344        let hooks = crate::hooks::Hooks::start(config, log)?;
5345        *self.hooks.lock().expect("hooks lock") = Some(hooks);
5346        Ok(self)
5347    }
5348
5349    /// Events the webhook queue dropped because it was full. Zero when
5350    /// no `--hooks-file` is configured.
5351    #[must_use]
5352    pub fn hook_drops(&self) -> u64 {
5353        self.hooks
5354            .lock()
5355            .expect("hooks lock")
5356            .as_ref()
5357            .map_or(0, crate::hooks::Hooks::dropped)
5358    }
5359
5360    /// The live latency record, for a caller that wants to tighten the
5361    /// gate or read it without going through the API.
5362    #[must_use]
5363    pub fn lag(&self) -> Arc<LagMeter> {
5364        self.lag.clone()
5365    }
5366
5367    /// Writes any breaches recorded since the last drain to the lag log.
5368    ///
5369    /// Called by the daemon's accept loop, which is the same place it
5370    /// polls for a failed durability barrier: both are things the writer
5371    /// thread can only report, never act on. No traffic means no drain,
5372    /// which is harmless because no traffic also means no breaches.
5373    pub fn drain_lag_log(&self) {
5374        let Some(path) = self.lag_log.as_ref() else {
5375            return;
5376        };
5377        let (breaches, dropped) = self.lag.drain();
5378        if breaches.is_empty() && dropped == 0 {
5379            return;
5380        }
5381        let gate_us = u64::try_from(self.lag.gate().as_micros()).unwrap_or(u64::MAX);
5382        let mut lines = String::new();
5383        if dropped > 0 {
5384            // The gap is written into the log rather than only counted,
5385            // so a reader of the file alone can see that it is not the
5386            // whole story.
5387            lines.push_str(
5388                &serde_json::json!({
5389                    "format_version": 1,
5390                    "event": "breaches_dropped",
5391                    "count": dropped,
5392                })
5393                .to_string(),
5394            );
5395            lines.push('\n');
5396        }
5397        for breach in breaches {
5398            lines.push_str(
5399                &serde_json::json!({
5400                    "format_version": 1,
5401                    "event": "gate_breach",
5402                    "seq": breach.seq,
5403                    "at_unix_ms": breach.at_unix_ms,
5404                    "gate_us": gate_us,
5405                    "decision_us": breach.decision_us,
5406                    "durable_us": breach.durable_us,
5407                    "batch": breach.batch,
5408                })
5409                .to_string(),
5410            );
5411            lines.push('\n');
5412        }
5413        let write = std::fs::OpenOptions::new()
5414            .create(true)
5415            .append(true)
5416            .open(path)
5417            .and_then(|mut file| std::io::Write::write_all(&mut file, lines.as_bytes()));
5418        if let Err(e) = write {
5419            // The reason, not the path: this string is served over the
5420            // API, and the path names the operator's home directory.
5421            let mut last = self.lag_log_error.lock().expect("lag log error lock");
5422            last.0 = Some(e.to_string());
5423            last.1 += 1;
5424        }
5425    }
5426
5427    /// The latency report as served under `/api/view`.
5428    fn lag_json(&self) -> serde_json::Value {
5429        let report = self.lag.report();
5430        let last_error = self
5431            .lag_log_error
5432            .lock()
5433            .expect("lag log error lock")
5434            .clone();
5435        serde_json::json!({
5436            "format_version": 1,
5437            "gate_us": report.gate_us,
5438            "gate_basis": "dequeue to acknowledgement, durability barrier included",
5439            "observed_ops": report.observed_ops,
5440            "decision": {
5441                "basis": "dequeue to append; the Phase-0 gate as written",
5442                "p50_us": report.decision_p50_us,
5443                "p99_us": report.decision_p99_us,
5444                "max_us": report.decision_max_us,
5445                "breaches": report.decision_breaches,
5446            },
5447            "durable": {
5448                "basis": "dequeue to acknowledgement; what a submitter waits out",
5449                "p50_us": report.durable_p50_us,
5450                "p99_us": report.durable_p99_us,
5451                "max_us": report.durable_max_us,
5452                "breaches": report.durable_breaches,
5453            },
5454            "percentile_basis": "power-of-two bucket upper bound capped at the observed maximum: over-estimates by at most 2x, never under-estimates",
5455            "since": "process start; not replayed from the log",
5456            // Whether, not where. An API client learning the daemon's
5457            // filesystem layout (which carries the operator's home
5458            // directory) buys nothing it can act on; the operator already
5459            // knows the path from the runbook.
5460            "log_configured": self.lag_log.is_some(),
5461            "log_error": last_error.0,
5462            "log_write_failures": last_error.1,
5463            "pending_breaches": report.pending_breaches,
5464        })
5465    }
5466
5467    /// The sequence the next admitted op will occupy, which is the
5468    /// cheapest complete description of "what state is this node in".
5469    ///
5470    /// Exposed for the browser surface's cache: it asks this before
5471    /// deciding whether to rebuild a page, so an unchanged node costs
5472    /// one `u64` read rather than a full view serialization.
5473    pub fn view_seq(&self) -> u64 {
5474        self.view.lock().expect("view lock").next_seq
5475    }
5476
5477    /// The store's generation and the handle-to-name map to render
5478    /// channels through (D46).
5479    ///
5480    /// Both together because a caller that resolves names has to cache
5481    /// on the generation the map was read at: revoking an account
5482    /// deletes a name and appends no op, so the view sequence does not
5483    /// move and a page keyed on it alone would keep showing the deleted
5484    /// name.
5485    ///
5486    /// `(0, empty)` on a node with no accounts store attached, which
5487    /// resolves nothing and renders every channel as itself — the same
5488    /// answer a store gives for an account issued before D46 or issued
5489    /// with an explicit `user`.
5490    #[must_use]
5491    pub fn roster(&self) -> (u64, std::collections::BTreeMap<String, String>) {
5492        match &*self.passkeys.lock().expect("accounts lock") {
5493            Some(accounts) => (accounts.generation(), accounts.roster()),
5494            None => (0, std::collections::BTreeMap::new()),
5495        }
5496    }
5497
5498    /// Repository the review `id` proposes to land on, in the canonical
5499    /// D29 spelling, or `None` when the review is unknown or unbound.
5500    ///
5501    /// Exposed for per-repository authorization: it is what lets posting
5502    /// a verdict require write on the repository under review, instead
5503    /// of the node-wide grant every review op would otherwise need.
5504    pub fn review_repo(&self, id: &str) -> Option<String> {
5505        self.view
5506            .lock()
5507            .expect("view lock")
5508            .reviews
5509            .get(id)?
5510            .target_ref
5511            .as_deref()
5512            .and_then(|target| target.split_once(':'))
5513            .map(|(repo, _)| crate::acl::normalize_repo(repo))
5514    }
5515
5516    /// One review as the API renders it, or `None` when no such review
5517    /// exists.
5518    ///
5519    /// Exposed for the D34 review page, which needs one review rather
5520    /// than the whole view. Same `review_json` the API uses, so the page
5521    /// and the API cannot describe a review differently.
5522    pub fn review_json(&self, id: &str) -> Option<serde_json::Value> {
5523        let view = self.view.lock().expect("view lock");
5524        view.reviews.get(id).map(review_json)
5525    }
5526
5527    /// Every review whose target ref names `repo`, newest id last.
5528    ///
5529    /// A review with no target ref belongs to no repository and is
5530    /// omitted: it cannot be shown under one without asserting a
5531    /// relationship the requester never signed.
5532    pub fn reviews_for_repo(&self, repo: &str) -> Vec<(String, serde_json::Value)> {
5533        let wanted = crate::acl::normalize_repo(repo);
5534        let view = self.view.lock().expect("view lock");
5535        view.reviews
5536            .iter()
5537            .filter(|(_, r)| {
5538                r.target_ref
5539                    .as_deref()
5540                    .and_then(|target| target.split_once(':'))
5541                    .is_some_and(|(named, _)| crate::acl::normalize_repo(named) == wanted)
5542            })
5543            .map(|(id, r)| (id.clone(), review_json(r)))
5544            .collect()
5545    }
5546
5547    /// Every live review that `channel` was drawn for and has not
5548    /// answered, paired with the repository its target ref names.
5549    ///
5550    /// The reviewer's own queue, which the view has always held and no
5551    /// page has ever shown. Three filters and no others:
5552    ///
5553    /// - **drawn**: `channel` is on the review's reviewer list. Being
5554    ///   able to read a repository is not being asked about it.
5555    /// - **unanswered**: no verdict of theirs stands. A review they have
5556    ///   already answered is not owed, whatever anybody else has said.
5557    /// - **live**: not archived. An archived review has dropped its
5558    ///   verdicts, so an answer to it would land nowhere.
5559    ///
5560    /// A review with no target ref is omitted for the same reason
5561    /// [`Self::reviews_for_repo`] omits one: it belongs to no repository,
5562    /// and this page's rows are grouped by one. It cannot become
5563    /// invisible that way — nothing can be drawn on a ref that is not
5564    /// named.
5565    ///
5566    /// **This does no authorization.** The caller filters by what the
5567    /// reader may read, because the ACL lives there and a second copy of
5568    /// that decision here is a second copy that can disagree.
5569    pub fn reviews_awaiting(&self, channel: &str) -> Vec<(String, String, serde_json::Value)> {
5570        let view = self.view.lock().expect("view lock");
5571        view.reviews
5572            .iter()
5573            .filter(|(_, r)| !matches!(r.status, choir_view::ReviewStatus::Archived { .. }))
5574            .filter(|(_, r)| r.reviewers.iter().any(|who| who == channel))
5575            .filter(|(_, r)| !r.verdicts.contains_key(channel))
5576            .filter_map(|(id, r)| {
5577                let repo = r
5578                    .target_ref
5579                    .as_deref()
5580                    .and_then(|target| target.split_once(':'))
5581                    .map(|(named, _)| crate::acl::normalize_repo(named))?;
5582                Some((id.clone(), repo, review_json(r)))
5583            })
5584            .collect()
5585    }
5586
5587    /// Handles one `/api/...` request, returning `(status, json_body)`.
5588    pub fn handle_api(&self, method: &str, path: &str, body: &[u8]) -> (u16, String) {
5589        match (method, path) {
5590            // Query-tolerant: `bound` reads `?limit=`/`?offset=` off the
5591            // same URL after this returns, and an exact match here would
5592            // have sent every paged request to the catch-all instead.
5593            ("GET", path) if path == "/api/view" || path.starts_with("/api/view?") => {
5594                let protected_path = self
5595                    .protected_refs
5596                    .lock()
5597                    .expect("protected refs lock")
5598                    .clone();
5599                let protected = ProtectedPolicySnapshot::read(protected_path.as_deref());
5600                // Read before the view guard: the audit mutex is taken
5601                // under the view lock nowhere else, and this keeps it
5602                // that way.
5603                let cohort = new_actor_cohort(self.newcomer_audit.as_ref());
5604                // Writer order is view -> concentration. Holding both
5605                // through snapshot construction prevents a response whose
5606                // heads include op N while `as_of_seq` and attribution stop
5607                // at N-1. The guards are dropped before byte measurement
5608                // and final response serialization.
5609                let view = self.view.lock().expect("view lock");
5610                let concentration_state = self.concentration.lock().expect("concentration lock");
5611                let key_names = self.key_names.lock().expect("key names lock");
5612                let ws: BTreeMap<_, _> = view
5613                    .workspaces
5614                    .iter()
5615                    .map(|(k, v)| (k.clone(), v.to_hex()))
5616                    .collect();
5617                let refs: BTreeMap<_, _> = view
5618                    .refs
5619                    .iter()
5620                    .map(|(k, v)| (k.clone(), v.to_hex()))
5621                    .collect();
5622                let reviews: BTreeMap<_, _> = view
5623                    .reviews
5624                    .iter()
5625                    .map(|(id, r)| (id.clone(), review_json(r)))
5626                    .collect();
5627                let changes: std::collections::BTreeMap<_, _> = view
5628                    .changes
5629                    .iter()
5630                    .map(|(id, change)| {
5631                        (
5632                            id.clone(),
5633                            serde_json::json!({
5634                                "owner": change.owner,
5635                                "workspace_id": change.workspace_id,
5636                                "active_workspace": change.active_workspace,
5637                                "base_revision": change.base_revision.to_hex(),
5638                                "revision_id": change.revision_id.to_hex(),
5639                                // Omitted when empty, so an unscoped
5640                                // change reads as it always did rather
5641                                // than gaining an empty list (D50).
5642                                "cone": (!change.cone.is_empty())
5643                                    .then(|| change.cone.clone()),
5644                            }),
5645                        )
5646                    })
5647                    .collect();
5648                let bindings: BTreeMap<_, _> = view
5649                    .bindings
5650                    .iter()
5651                    .map(|(key_id, binding)| (key_id.clone(), binding_json(binding)))
5652                    .collect();
5653                // The latest admitted ref-state attestation (D25), as a
5654                // summary: `refs` above already carries the full map, so
5655                // repeating it here would double the hot-path response
5656                // for no reader.
5657                let snapshot = view.latest_snapshot.as_ref().map(|s| {
5658                    serde_json::json!({
5659                        "id": s.id().to_hex(),
5660                        "at_seq": s.at_seq,
5661                        "prev_snapshot": s.prev_snapshot.as_ref().map(ContentHash::to_hex),
5662                    })
5663                });
5664                let ws = serde_json::json!(ws);
5665                let refs = serde_json::json!(refs);
5666                let reviews = serde_json::json!(reviews);
5667                let provenance = serde_json::json!(&view.provenance);
5668                // Not folded into `view_growth`'s authoritative total:
5669                // that byte count is a tracked series, and a fifth
5670                // section would move every past reading. Measured beside
5671                // `bindings` instead, for the same reason.
5672                let checks = serde_json::json!(&view.checks);
5673                let bindings = serde_json::json!(bindings);
5674                // D65. Serialized straight from the fold's own maps:
5675                // there is no projection to write, and a hand-built one
5676                // would be a second place for the shape to drift.
5677                let vouches = serde_json::json!(&view.vouches);
5678                // D67, on the same terms: the fold's map, not a
5679                // projection of it.
5680                let witnessed = serde_json::json!(&view.witnessed);
5681                let counts = view_growth_counts(&view);
5682                let as_of_seq = concentration_state.as_of_seq;
5683                // T3 attribution reads the durable record, not the keys
5684                // file: a tripwire whose evidence the operator can edit in
5685                // place measures the operator's honesty, not concentration.
5686                // `key_names` still supplies the trusted population, which
5687                // no op in the log can answer.
5688                let durable_bindings = KeyBindings::from_view(&view, &key_names);
5689                let concentration =
5690                    concentration_json(&concentration_state, &durable_bindings, &protected);
5691                let new_actor_review_outcomes = new_actor_review_outcomes_json(
5692                    &concentration_state,
5693                    &view,
5694                    cohort.as_ref(),
5695                    self.review_adjudications.as_deref(),
5696                );
5697                drop(key_names);
5698                drop(concentration_state);
5699                // Read under the guard, reported below it. The guard is
5700                // released before `scope_now`, so the position has to be
5701                // taken while the view still holds still.
5702                let next_seq = view.next_seq;
5703                drop(view);
5704                let view_growth = view_growth_json(
5705                    counts,
5706                    &ws,
5707                    &refs,
5708                    &reviews,
5709                    &provenance,
5710                    &[
5711                        ("bindings", &bindings),
5712                        ("vouches", &vouches),
5713                        ("witnessed", &witnessed),
5714                    ],
5715                    as_of_seq,
5716                );
5717                let newcomer_harm = newcomer_harm_json(self.newcomer_audit.as_ref());
5718                // What a client needs to bind its next signature to this
5719                // log: which node, which head, and whether that head is
5720                // being enforced. Read here rather than under the view
5721                // guard above — an op that lands in between only makes
5722                // `head` one entry stale, and a scope naming any head
5723                // still in the window is admissible.
5724                let (node, head) = self.scope_now();
5725                let log = serde_json::json!({
5726                    "node": node.to_hex(),
5727                    "head": head.as_ref().map(ContentHash::to_hex),
5728                    // The position this view describes. It belongs to the
5729                    // log rather than beside it, which is also what keeps
5730                    // it readable: `log` is `Disclosure::Public`, and a
5731                    // reader granted one repository still needs to know
5732                    // where the view they were handed sits. It says no
5733                    // more about a repository than `head` already does.
5734                    //
5735                    // Without it the view could not say its own position,
5736                    // and a caller wanting the age of anything had
5737                    // nothing to measure against -- which is exactly how
5738                    // `ops_since_binding` came to be zero for every key
5739                    // on every real node while its doctest passed.
5740                    "next_seq": next_seq,
5741                    "scope_required": self
5742                        .require_scope
5743                        .load(std::sync::atomic::Ordering::Relaxed),
5744                });
5745                let body = serde_json::json!({
5746                    "log": log,
5747                    "snapshot": snapshot,
5748                    "workspaces": ws,
5749                    "changes": changes,
5750                    "refs": refs,
5751                    "reviews": reviews,
5752                    "provenance": provenance,
5753                    "checks": checks,
5754                    "bindings": bindings,
5755                    "vouches": vouches,
5756                    "witnessed": witnessed,
5757                    "concentration": concentration,
5758                    "view_growth": view_growth,
5759                    "newcomer_harm": newcomer_harm,
5760                    "new_actor_review_outcomes": new_actor_review_outcomes,
5761                    "sequencer_lag": self.lag_json(),
5762                    "build": crate::build_json(),
5763                });
5764                (200, body.to_string())
5765            }
5766            // Pending queue for one reviewer: reviews that fanned out to
5767            // them and are still unanswered by them.
5768            ("GET", path) if path.starts_with("/api/reviews") => {
5769                let reviewer = decode_query_value(
5770                    path.split_once("reviewer=")
5771                        .map(|(_, v)| v.split('&').next().unwrap_or(v))
5772                        .unwrap_or(""),
5773                );
5774                let reviewer = reviewer.as_str();
5775                let view = self.view.lock().expect("view lock");
5776                let pending: std::collections::BTreeMap<_, _> = view
5777                    .reviews
5778                    .iter()
5779                    .filter(|(_, r)| {
5780                        r.reviewers.iter().any(|x| x == reviewer)
5781                            && !r.verdicts.contains_key(reviewer)
5782                    })
5783                    .map(|(id, r)| (id.clone(), review_json(r)))
5784                    .collect();
5785                (200, serde_json::json!({ "pending": pending }).to_string())
5786            }
5787            ("POST", "/api/appeal") => {
5788                let req: serde_json::Value = match serde_json::from_slice(body) {
5789                    Ok(value) => value,
5790                    Err(error) => {
5791                        return (
5792                            400,
5793                            Rejection::new(
5794                                Code::MalformedRequest,
5795                                format!("request body is not valid JSON: {error}"),
5796                                "send {\"attempt_id\": N} using the newcomer_attempt_id from \
5797                                 the rejected response",
5798                            )
5799                            .body(),
5800                        );
5801                    }
5802                };
5803                let Some(attempt_id) = req.get("attempt_id").and_then(serde_json::Value::as_u64)
5804                else {
5805                    return (
5806                        400,
5807                        Rejection::new(
5808                            Code::MalformedRequest,
5809                            "appeal needs an integer attempt_id",
5810                            "send the newcomer_attempt_id from the rejected response",
5811                        )
5812                        .body(),
5813                    );
5814                };
5815                let Some(audit) = &self.newcomer_audit else {
5816                    return (
5817                        503,
5818                        Rejection::new(
5819                            Code::PolicyUnavailable,
5820                            "newcomer audit is not enabled",
5821                            "ask the operator to enable the newcomer audit before filing appeals",
5822                        )
5823                        .body(),
5824                    );
5825                };
5826                match audit
5827                    .lock()
5828                    .expect("newcomer audit lock")
5829                    .appeal(attempt_id)
5830                {
5831                    Ok(()) => (
5832                        200,
5833                        serde_json::json!({ "appealed": attempt_id }).to_string(),
5834                    ),
5835                    Err(error) => (
5836                        400,
5837                        Rejection::new(
5838                            Code::MalformedRequest,
5839                            error,
5840                            "use the attempt id from a rejected first-attempt response",
5841                        )
5842                        .body(),
5843                    ),
5844                }
5845            }
5846            ("POST", "/api/submit") => self.submit(body),
5847            ("POST", "/api/submit-batch") => {
5848                let req: serde_json::Value = match serde_json::from_slice(body) {
5849                    Ok(v) => v,
5850                    Err(e) => return (400, Rejection::new(
5851                            Code::MalformedRequest,
5852                            format!("request body is not valid JSON: {e}"),
5853                            "send a JSON object; GET /llms.txt lists the fields each endpoint wants",
5854                        )
5855                        .body()),
5856                };
5857                let Some(ops) = req.get("ops").and_then(|v| v.as_array()) else {
5858                    return (400, r#"{"error":"need ops array"}"#.to_string());
5859                };
5860                if ops.len() > self.batch_limit {
5861                    return (
5862                        413,
5863                        serde_json::json!({
5864                            "error": "batch has too many operations",
5865                            "limit_ops": self.batch_limit,
5866                            "actual_ops": ops.len(),
5867                        })
5868                        .to_string(),
5869                    );
5870                }
5871                // Ops are admitted in array order; each result is
5872                // independent (a rejection does not abort the batch).
5873                //
5874                // Every op is decoded from the already-parsed request,
5875                // offered to the sequencer, and only then waited on. The
5876                // previous shape called the single-op path in a loop,
5877                // which re-serialised each op to a string, re-parsed it,
5878                // blocked for its reply, and re-parsed the reply. Blocking
5879                // per op was the expensive part: it left the writer's
5880                // queue empty every time it looked, so a 500-op body paid
5881                // 500 durability barriers instead of ceil(500/MAX_BATCH).
5882                let mut decoded = Vec::with_capacity(ops.len());
5883                for op in ops {
5884                    let mut one = decode_submission(op);
5885                    if let Ok(sub) = one.as_mut() {
5886                        self.stamp_credential_key(sub);
5887                    }
5888                    decoded.push(one);
5889                }
5890                let started_at_unix_ms: Vec<u64> = decoded.iter().map(|_| unix_ms()).collect();
5891                // Malformed ops never reach the sequencer. Well-formed
5892                // ones are pushed in request order, so pulling one
5893                // outcome per `Ok` below keeps results aligned with the
5894                // request array without any index bookkeeping.
5895                let subs: Vec<Submission> = decoded
5896                    .iter()
5897                    .filter_map(|d| d.as_ref().ok())
5898                    .map(|sub| Submission {
5899                        channel: sub.channel.clone(),
5900                        payload: sub.payload.clone(),
5901                        author_sig: sub.author_sig.clone(),
5902                    })
5903                    .collect();
5904                let mut outcomes = self.handle.try_submit_many(subs).into_iter();
5905
5906                let mut accepted = 0u64;
5907                let mut rejected = 0u64;
5908                let mut results: Vec<serde_json::Value> = Vec::with_capacity(decoded.len());
5909                for (index, d) in decoded.iter().enumerate() {
5910                    let value = match d {
5911                        Err(reason) => {
5912                            rejected += 1;
5913                            serde_json::json!({ "error": reason })
5914                        }
5915                        Ok(sub) => match outcomes.next().expect("one outcome per submitted op") {
5916                            Ok(acc) => {
5917                                accepted += 1;
5918                                let mut value = self.batch_result(acc, sub);
5919                                self.add_newcomer_outcome(
5920                                    &mut value,
5921                                    sub,
5922                                    started_at_unix_ms[index],
5923                                    true,
5924                                    None,
5925                                );
5926                                value
5927                            }
5928                            Err(reason) => {
5929                                rejected += 1;
5930                                let rejection = Rejection::decode(&reason).to_json();
5931                                let code = rejection["code"].as_str().map(str::to_string);
5932                                let mut value = serde_json::json!({ "error": reason });
5933                                self.add_newcomer_outcome(
5934                                    &mut value,
5935                                    sub,
5936                                    started_at_unix_ms[index],
5937                                    false,
5938                                    code.as_deref(),
5939                                );
5940                                value
5941                            }
5942                        },
5943                    };
5944                    results.push(value);
5945                }
5946                let mut response = serde_json::json!({
5947                    "accepted": accepted,
5948                    "rejected": rejected,
5949                    "results": results,
5950                });
5951                if accepted != 0 {
5952                    // Once per request, after every assignment response
5953                    // has been produced. Calling this from `batch_result`
5954                    // would rescan and emit once per element, undoing the
5955                    // durability-barrier win of the batch endpoint.
5956                    self.add_retention_outcome(&mut response);
5957                }
5958                (200, response.to_string())
5959            }
5960            ("POST", "/api/git-update") => {
5961                let req: serde_json::Value = match serde_json::from_slice(body) {
5962                    Ok(v) => v,
5963                    Err(e) => return (400, Rejection::new(
5964                            Code::MalformedRequest,
5965                            format!("request body is not valid JSON: {e}"),
5966                            "send a JSON object; GET /llms.txt lists the fields each endpoint wants",
5967                        )
5968                        .body()),
5969                };
5970                let f = |k: &str| {
5971                    req.get(k)
5972                        .and_then(|v| v.as_str())
5973                        .unwrap_or("")
5974                        .to_string()
5975                };
5976                let (status, signer) = (f("cert_status"), f("signer"));
5977                match self.git_update(
5978                    &f("repo"),
5979                    &f("refname"),
5980                    &f("old"),
5981                    &f("new"),
5982                    &f("user"),
5983                    Some((status.as_str(), signer.as_str())),
5984                ) {
5985                    Ok(()) => (200, r#"{"ok":true}"#.to_string()),
5986                    Err(reason) => (400, Rejection::decode(&reason).body()),
5987                }
5988            }
5989            // The other half of the same hook: the push is being refused,
5990            // so every ref already accepted for it has to be put back.
5991            ("POST", "/api/git-abort") => {
5992                let req: serde_json::Value = match serde_json::from_slice(body) {
5993                    Ok(v) => v,
5994                    Err(e) => return (400, Rejection::new(
5995                            Code::MalformedRequest,
5996                            format!("request body is not valid JSON: {e}"),
5997                            "send a JSON object; GET /llms.txt lists the fields each endpoint wants",
5998                        )
5999                        .body()),
6000                };
6001                let f = |k: &str| {
6002                    req.get(k)
6003                        .and_then(|v| v.as_str())
6004                        .unwrap_or("")
6005                        .to_string()
6006                };
6007                let (status, signer) = (f("cert_status"), f("signer"));
6008                match self.git_abort(
6009                    &f("repo"),
6010                    &f("refname"),
6011                    &f("old"),
6012                    &f("new"),
6013                    &f("user"),
6014                    Some((status.as_str(), signer.as_str())),
6015                ) {
6016                    Ok(()) => (200, r#"{"ok":true}"#.to_string()),
6017                    Err(reason) => (400, Rejection::decode(&reason).body()),
6018                }
6019            }
6020            ("GET", path) if path.starts_with("/api/log") => {
6021                let from: usize = path
6022                    .split_once("from=")
6023                    .and_then(|(_, v)| v.split('&').next()?.parse().ok())
6024                    .unwrap_or(0);
6025                let window = self.entries.lock().expect("entries lock");
6026                let base = window.base as usize;
6027                // Behind the window: the entries the reader still needs
6028                // are no longer in memory. Serving from `base` here
6029                // would hand back a normal-looking page with a silent
6030                // hole in it, so take the persisted log instead — and
6031                // if there is none, say so loudly rather than lie.
6032                if from < base {
6033                    drop(window);
6034                    let Some(path) = &self.log_path else {
6035                        // The brief's reference example: a refusal that
6036                        // already refused correctly (a page with a hole is
6037                        // worse than an error), now carrying the same
6038                        // reason code and named action as every other.
6039                        let mut body = Rejection::new(
6040                            Code::LogEvicted,
6041                            "requested entries have been evicted and no persisted log is configured",
6042                            format!(
6043                                "resync from seq {base} instead; entries before it are gone from \
6044                                 this node"
6045                            ),
6046                        )
6047                        .with_states(Some(from.to_string()), Some(format!("oldest available {base}")))
6048                        .to_json();
6049                        body["window_base"] = serde_json::json!(base);
6050                        return (409, body.to_string());
6051                    };
6052                    return match replay_from_disk(path, from, LOG_PAGE) {
6053                        Ok(rows) => (
6054                            200,
6055                            serde_json::json!({
6056                                "entries": rows,
6057                                "window_base": base,
6058                                "source": "log",
6059                            })
6060                            .to_string(),
6061                        ),
6062                        Err(e) => (
6063                            500,
6064                            Rejection::new(
6065                                Code::Unclassified,
6066                                format!("resync from the persisted log failed: {e}"),
6067                                "retry; this is a node-side read failure, not something the \
6068                                 submission can fix",
6069                            )
6070                            .body(),
6071                        ),
6072                    };
6073                }
6074                let rows: Vec<serde_json::Value> = window
6075                    .entries
6076                    .iter()
6077                    .skip(from - base)
6078                    .take(LOG_PAGE)
6079                    .map(entry_json)
6080                    .collect();
6081                (
6082                    200,
6083                    serde_json::json!({
6084                        "entries": rows,
6085                        "window_base": window.base,
6086                        "source": "window",
6087                    })
6088                    .to_string(),
6089                )
6090            }
6091            _ => (404, r#"{"error":"no such endpoint"}"#.to_string()),
6092        }
6093    }
6094
6095    fn submit(&self, body: &[u8]) -> (u16, String) {
6096        let req: serde_json::Value =
6097            match serde_json::from_slice(body) {
6098                Ok(v) => v,
6099                Err(e) => return (
6100                    400,
6101                    Rejection::new(
6102                        Code::MalformedRequest,
6103                        format!("request body is not valid JSON: {e}"),
6104                        "send a JSON object; GET /llms.txt lists the fields each endpoint wants",
6105                    )
6106                    .body(),
6107                ),
6108            };
6109        let mut sub = match decode_submission(&req) {
6110            Ok(sub) => sub,
6111            Err(reason) => return (400, Rejection::decode(&reason).body()),
6112        };
6113        self.stamp_credential_key(&mut sub);
6114        let started_at_unix_ms = unix_ms();
6115        match self
6116            .handle
6117            .try_submit(&sub.channel, sub.payload.clone(), sub.author_sig.clone())
6118        {
6119            Ok(acc) => {
6120                let mut response = self.batch_result(acc, &sub);
6121                self.add_retention_outcome(&mut response);
6122                self.add_newcomer_outcome(&mut response, &sub, started_at_unix_ms, true, None);
6123                (200, response.to_string())
6124            }
6125            Err(reason) => {
6126                // A retry of an operation that already landed fails CAS
6127                // in exactly the same way as a genuine conflict. They
6128                // call for opposite actions -- read back and proceed,
6129                // versus re-read and rebase -- so an agent that cannot
6130                // tell them apart either retries a completed write or
6131                // abandons a successful one.
6132                //
6133                // Only the policy's own duplicate refusal is converted,
6134                // and that is load-bearing. This lookup used to run for
6135                // *any* rejection, which meant a submission carrying a
6136                // corrupted signature over an already-applied payload was
6137                // answered 200 `already_applied`: the signature check had
6138                // failed, and the response said success. It changed no
6139                // state, but a check whose failure is reported as a
6140                // success is not a check. `duplicate_submission` is
6141                // raised only after the signature verifies.
6142                if Rejection::decode(&reason).code == Code::DuplicateSubmission.as_str() {
6143                    if let Some((seq, hash)) = self
6144                        .entries
6145                        .lock()
6146                        .expect("entries lock")
6147                        .already_applied(&sub.channel, &sub.payload)
6148                    {
6149                        let mut response = serde_json::json!({
6150                            "seq": seq,
6151                            "hash": hash.to_hex(),
6152                            "already_applied": true,
6153                        });
6154                        self.add_newcomer_outcome(
6155                            &mut response,
6156                            &sub,
6157                            started_at_unix_ms,
6158                            true,
6159                            None,
6160                        );
6161                        return (200, response.to_string());
6162                    }
6163                }
6164                let mut response = Rejection::decode(&reason).to_json();
6165                let code = response["code"].as_str().map(str::to_string);
6166                self.add_newcomer_outcome(
6167                    &mut response,
6168                    &sub,
6169                    started_at_unix_ms,
6170                    false,
6171                    code.as_deref(),
6172                );
6173                (400, response.to_string())
6174            }
6175        }
6176    }
6177
6178    /// The success body for one admitted op, shared by `/api/submit` and
6179    /// `/api/submit-batch` so the two cannot answer differently.
6180    ///
6181    /// An unassigned review request is answered with a node-signed
6182    /// assignment draw once the request itself is admitted. That draw is a
6183    /// *further* submission, so it deliberately happens here, after the
6184    /// batch's own barrier, rather than being folded into it.
6185    fn batch_result(
6186        &self,
6187        acc: choir_sequencer::Accepted,
6188        sub: &DecodedSubmission,
6189    ) -> serde_json::Value {
6190        let mut resp = serde_json::json!({ "seq": acc.seq, "hash": acc.hash.to_hex() });
6191        if let Some(id) = &sub.unassigned_review {
6192            match self.assign_reviewers(id, &sub.channel) {
6193                Ok(reviewers) => resp["reviewers"] = serde_json::json!(reviewers),
6194                // The request stands; it is visibly unassigned, which is
6195                // a state no verdict can complete.
6196                Err(e) => resp["assignment_error"] = serde_json::json!(e),
6197            }
6198        }
6199        resp
6200    }
6201}
6202
6203/// How one ref disagrees between the log and a bare repo.
6204#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6205pub enum RefState {
6206    /// The log names a commit the repo has but is not pointing at. Git is
6207    /// the follower, so this is repaired by moving git.
6208    GitBehind,
6209    /// The log names a commit the repo does not have and cannot get — a
6210    /// refused push's objects go out with the quarantine. Repaired by the
6211    /// log giving way, through a compensating op.
6212    LogUnbackable,
6213    /// Git holds a ref the log has never seen. Reported only: adopting it
6214    /// would launder an out-of-band `update-ref` into the signed log.
6215    GitOnly,
6216    /// The comparison could not be made at all.
6217    Unreadable,
6218}
6219
6220impl RefState {
6221    /// Stable wire name, so a monitor can match on it.
6222    pub fn as_str(self) -> &'static str {
6223        match self {
6224            RefState::GitBehind => "git_behind",
6225            RefState::LogUnbackable => "log_unbackable",
6226            RefState::GitOnly => "git_only",
6227            RefState::Unreadable => "unreadable",
6228        }
6229    }
6230}
6231
6232/// One disagreement found by [`Platform::survey_git_refs`].
6233#[derive(Debug, Clone)]
6234pub struct RefFinding {
6235    /// Repo the ref lives in, as it appears in the namespaced log name.
6236    pub repo: String,
6237    /// Ref name inside that repo; empty when the whole repo is the
6238    /// problem.
6239    pub refname: String,
6240    /// What the log says, as a git oid.
6241    pub log_oid: Option<String>,
6242    /// What the repo says.
6243    pub git_oid: Option<String>,
6244    /// Which disagreement this is.
6245    pub state: RefState,
6246    /// Human-readable cause, for the startup log and the endpoint.
6247    pub reason: String,
6248}
6249
6250impl RefFinding {
6251    fn unreadable(repo: &str, refname: &str, reason: &str) -> Self {
6252        Self {
6253            repo: repo.to_string(),
6254            refname: refname.to_string(),
6255            log_oid: None,
6256            git_oid: None,
6257            state: RefState::Unreadable,
6258            reason: reason.to_string(),
6259        }
6260    }
6261
6262    /// The namespaced `<repo>:<refname>` name, as the log spells it.
6263    pub fn name(&self) -> String {
6264        if self.refname.is_empty() {
6265            self.repo.clone()
6266        } else {
6267            format!("{}:{}", self.repo, self.refname)
6268        }
6269    }
6270
6271    /// The finding as it goes over the wire.
6272    pub fn to_json(&self) -> serde_json::Value {
6273        serde_json::json!({
6274            "ref": self.name(),
6275            "state": self.state.as_str(),
6276            "log_oid": self.log_oid,
6277            "git_oid": self.git_oid,
6278            "reason": self.reason,
6279        })
6280    }
6281}
6282
6283/// What [`Platform::reconcile_git_refs`] did, so a caller can report it.
6284/// An all-empty report is the normal case and the only silent one.
6285#[derive(Debug, Default)]
6286pub struct RefReconciliation {
6287    /// Refs written into git because the view held a commit git already
6288    /// had but had not pointed at.
6289    pub applied: Vec<String>,
6290    /// Refs the view gave up, because git can never hold the commit they
6291    /// named. Each one is a compensating op in the log.
6292    pub retracted: Vec<String>,
6293    /// Disagreements left standing, each with its reason. These need an
6294    /// operator: repairing them automatically would either lose history
6295    /// or launder an out-of-band ref into the signed log.
6296    pub unreconciled: Vec<String>,
6297}
6298
6299impl RefReconciliation {
6300    /// Whether anything at all was out of agreement.
6301    pub fn is_empty(&self) -> bool {
6302        self.applied.is_empty() && self.retracted.is_empty() && self.unreconciled.is_empty()
6303    }
6304}
6305
6306/// Every ref in the bare repo at `path`, or `None` if it cannot be read
6307/// (no such repo, or not a repo).
6308fn read_git_refs(path: &std::path::Path) -> Option<BTreeMap<String, String>> {
6309    let out = std::process::Command::new("git")
6310        .args(["for-each-ref", "--format=%(refname) %(objectname)"])
6311        .current_dir(path)
6312        .output()
6313        .ok()?;
6314    if !out.status.success() {
6315        return None;
6316    }
6317    Some(
6318        String::from_utf8_lossy(&out.stdout)
6319            .lines()
6320            .filter_map(|line| {
6321                let (refname, oid) = line.split_once(' ')?;
6322                Some((refname.to_string(), oid.to_string()))
6323            })
6324            .collect(),
6325    )
6326}
6327
6328/// Whether `oid` names an object the repo actually has. A refused push
6329/// leaves its objects in a discarded quarantine, so the view can name a
6330/// commit that was never admitted to the repo.
6331fn object_exists(path: &std::path::Path, oid: &str) -> bool {
6332    std::process::Command::new("git")
6333        .args(["cat-file", "-e", &format!("{oid}^{{object}}")])
6334        .current_dir(path)
6335        .output()
6336        .is_ok_and(|out| out.status.success())
6337}
6338
6339/// Points `refname` at `oid`, CAS'd on `have` so a concurrent writer
6340/// loses rather than gets clobbered.
6341fn write_git_ref(
6342    path: &std::path::Path,
6343    refname: &str,
6344    oid: &str,
6345    have: Option<&str>,
6346) -> Result<(), String> {
6347    let old = have.map_or_else(|| "0".repeat(oid.len()), str::to_string);
6348    let out = std::process::Command::new("git")
6349        .args(["update-ref", refname, oid, &old])
6350        .current_dir(path)
6351        .output()
6352        .map_err(|e| e.to_string())?;
6353    if out.status.success() {
6354        return Ok(());
6355    }
6356    Err(String::from_utf8_lossy(&out.stderr).trim().to_string())
6357}
6358
6359/// Git's "this ref does not exist" oid: all zeros, at whatever width the
6360/// repo's hash function uses.
6361fn is_zero_oid(hex: &str) -> bool {
6362    !hex.is_empty() && hex.chars().all(|c| c == '0')
6363}
6364
6365/// Parses an undirected operator graph and returns every operator within
6366/// `max_distance` of `start`, including `start` itself at distance zero.
6367fn operators_within_distance(
6368    path: &std::path::Path,
6369    start: &str,
6370    max_distance: usize,
6371) -> Result<BTreeSet<String>, String> {
6372    let text =
6373        std::fs::read_to_string(path).map_err(|e| format!("read reviewer conflict graph: {e}"))?;
6374    let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
6375    for (index, raw) in text.lines().enumerate() {
6376        let line = raw.trim();
6377        if line.is_empty() || line.starts_with('#') {
6378            continue;
6379        }
6380        let mut fields = line.split_whitespace();
6381        let left = fields.next();
6382        let right = fields.next();
6383        if left.is_none() || right.is_none() || fields.next().is_some() {
6384            return Err(format!(
6385                "reviewer conflict graph line {} must be '<operator> <operator>'",
6386                index + 1
6387            ));
6388        }
6389        let left = left.expect("checked above");
6390        let right = right.expect("checked above");
6391        if left.contains('/') || right.contains('/') {
6392            return Err(format!(
6393                "reviewer conflict graph line {} must name operators, not operator/agent channels",
6394                index + 1
6395            ));
6396        }
6397        graph
6398            .entry(left.to_string())
6399            .or_default()
6400            .insert(right.to_string());
6401        graph
6402            .entry(right.to_string())
6403            .or_default()
6404            .insert(left.to_string());
6405    }
6406
6407    let mut seen = BTreeSet::from([start.to_string()]);
6408    let mut queue = VecDeque::from([(start.to_string(), 0usize)]);
6409    while let Some((operator, distance)) = queue.pop_front() {
6410        if distance == max_distance {
6411            continue;
6412        }
6413        let Some(neighbors) = graph.get(&operator) else {
6414            continue;
6415        };
6416        for neighbor in neighbors {
6417            if seen.insert(neighbor.clone()) {
6418                queue.push_back((neighbor.clone(), distance + 1));
6419            }
6420        }
6421    }
6422    Ok(seen)
6423}
6424
6425/// One decoded `/api/submit` body: the wire fields turned into the bytes
6426/// the sequencer wants, plus whatever the response will need afterwards.
6427struct DecodedSubmission {
6428    channel: String,
6429    payload: Vec<u8>,
6430    author_sig: Option<Witness>,
6431    /// Review id when this op opens a review naming no reviewers, so the
6432    /// node knows to draw for it once the op is admitted.
6433    unassigned_review: Option<String>,
6434}
6435
6436fn decode_create_submission(
6437    request: &serde_json::Value,
6438    expected_id: &str,
6439    expected_owner: &str,
6440    expected_workspace: &str,
6441    expected_revision: &ContentHash,
6442    expected_idempotency_key: &str,
6443) -> Result<(DecodedSubmission, Vec<String>), String> {
6444    let sub = decode_submission(request).map_err(|reason| {
6445        Rejection::new(
6446            Code::MalformedRequest,
6447            reason,
6448            "sign the exact CreateAuthorization payload with the requested owner key and include channel, payload_hex, key_id and signature_hex",
6449        )
6450        .encode()
6451    })?;
6452    let authorization = CreateAuthorization::from_payload(&sub.payload)
6453        .map_err(|error| crate::reject::from_view_error(&error).encode())?;
6454    match authorization {
6455        CreateAuthorization {
6456            id,
6457            owner,
6458            workspace,
6459            base_revision,
6460            idempotency_key,
6461            cone,
6462            ..
6463        } if id == expected_id
6464            && owner == expected_owner
6465            && sub.channel == expected_owner
6466            && workspace == expected_workspace
6467            && &base_revision == expected_revision
6468            && idempotency_key == expected_idempotency_key => Ok((sub, cone)),
6469        _ => Err(Rejection::new(
6470            Code::WorkspaceState,
6471            "signed create payload does not match the requested owner, change, workspace, base and idempotency key",
6472            "rebuild CreateAuthorization from the exact request binding, sign it on the owner channel and retry",
6473        )
6474        .encode()),
6475    }
6476}
6477
6478fn decode_archive_submission(
6479    request: &serde_json::Value,
6480    expected_id: &str,
6481    expected_workspace: &str,
6482    expected_revision: &ContentHash,
6483) -> Result<DecodedSubmission, String> {
6484    let sub = decode_submission(request).map_err(|reason| {
6485        Rejection::new(
6486            Code::MalformedRequest,
6487            reason,
6488            "sign the exact ArchiveAuthorization payload with the bound owner key and include channel, payload_hex, key_id and signature_hex",
6489        )
6490        .encode()
6491    })?;
6492    let authorization = ArchiveAuthorization::from_payload(&sub.payload)
6493        .map_err(|error| crate::reject::from_view_error(&error).encode())?;
6494    match authorization {
6495        ArchiveAuthorization {
6496            id,
6497            workspace,
6498            prev_revision,
6499            ..
6500        } if id == expected_id
6501            && workspace == expected_workspace
6502            && &prev_revision == expected_revision => Ok(sub),
6503        _ => Err(Rejection::new(
6504            Code::WorkspaceState,
6505            "signed archive payload does not match the requested change, workspace and revision",
6506            "re-read GET /api/view, rebuild ArchiveAuthorization from that exact state, sign it on the owner channel and retry",
6507        )
6508        .encode()),
6509    }
6510}
6511
6512/// Turns one already-parsed request object into a submission.
6513///
6514/// Split out so `/api/submit-batch` can decode straight from the parsed
6515/// request array. The previous batch path re-serialised each element back
6516/// to a string and re-parsed it through the single-op entry point, which
6517/// is two extra JSON round-trips per op on the endpoint that exists to
6518/// avoid per-op overhead.
6519fn decode_submission(req: &serde_json::Value) -> Result<DecodedSubmission, String> {
6520    let field = |name: &str| req.get(name).and_then(|v| v.as_str());
6521    let channel = match (field("channel"), field("workspace")) {
6522        (Some(channel), None) | (None, Some(channel)) => channel,
6523        (Some(channel), Some(legacy)) if channel == legacy => channel,
6524        (Some(_), Some(_)) => {
6525            return Err("channel and legacy workspace fields disagree".to_string())
6526        }
6527        (None, None) => return Err("need channel (or legacy workspace)".to_string()),
6528    };
6529    let (Some(payload_hex), Some(key_id), Some(signature_hex)) = (
6530        field("payload_hex"),
6531        field("key_id"),
6532        field("signature_hex"),
6533    ) else {
6534        return Err("need payload_hex, key_id, signature_hex".to_string());
6535    };
6536    let (Some(payload), Some(signature)) = (hex_decode(payload_hex), hex_decode(signature_hex))
6537    else {
6538        return Err("bad hex".to_string());
6539    };
6540    let unassigned_review = match ViewOp::from_payload(&payload) {
6541        Ok(op) => match op.kind {
6542            OpKind::RequestReview { id, reviewers, .. } if reviewers.is_empty() => Some(id),
6543            _ => None,
6544        },
6545        Err(_) => None,
6546    };
6547    // A submission that names no scheme is ed25519, the only one that
6548    // existed before D39 -- the same rule `Witness::scheme_id` applies to
6549    // stored entries, so the wire and the log agree about what silence
6550    // means.
6551    let author_sig = match req.get("scheme").and_then(serde_json::Value::as_u64) {
6552        None => Witness::ed25519(key_id.to_string(), signature),
6553        Some(tag) if tag == u64::from(choir_oplog::scheme::WEBAUTHN_ES256) => {
6554            let (Some(auth_hex), Some(client_hex)) = (
6555                field("authenticator_data_hex"),
6556                field("client_data_json_hex"),
6557            ) else {
6558                return Err(
6559                    "a webauthn signature needs authenticator_data_hex and client_data_json_hex"
6560                        .to_string(),
6561                );
6562            };
6563            let (Some(authenticator_data), Some(client_data_json)) =
6564                (hex_decode(auth_hex), hex_decode(client_hex))
6565            else {
6566                return Err("bad hex in the webauthn fields".to_string());
6567            };
6568            Witness::webauthn_es256(
6569                key_id.to_string(),
6570                signature,
6571                authenticator_data,
6572                client_data_json,
6573            )
6574        }
6575        // Named rather than silently reinterpreted. A scheme this binary
6576        // cannot check is refused here, where the caller learns which tag
6577        // was rejected, instead of being handed to ed25519 and failing as
6578        // a bad signature.
6579        Some(tag) => return Err(format!("unknown signature scheme {tag}")),
6580    };
6581    Ok(DecodedSubmission {
6582        channel: channel.to_string(),
6583        payload,
6584        author_sig: Some(author_sig),
6585        unassigned_review,
6586    })
6587}
6588
6589/// JSON shape of one log entry, shared by the in-memory window and the
6590/// on-disk resync path so a catching-up reader cannot tell them apart.
6591///
6592/// Every field of the hashed form is here, which is what makes the
6593/// chain checkable by someone who does not trust this node: the client
6594/// rebuilds the canonical bytes, hashes them, and compares against
6595/// `hash`. `SYNC.md` writes that recipe out, and
6596/// `tests/sync_contract.rs` executes it — if a field is added here and
6597/// not there, that test fails rather than a third-party client
6598/// silently losing the ability to verify.
6599fn entry_json(e: &OpEntry) -> serde_json::Value {
6600    let mut value = serde_json::json!({
6601        "seq": e.seq,
6602        "workspace": e.channel,
6603        "payload_hex": hex_encode(&e.payload),
6604        "author_key": e.author_sig.as_ref().map(|w| w.key_id.clone()),
6605        // Chain position. `parent` alone lets a client join two pages
6606        // (page N+1's first parent is page N's last hash); `hash` is
6607        // the node's claim about this entry, which the client is meant
6608        // to recompute from the fields below rather than believe.
6609        "hash": e.content_hash().to_hex(),
6610        "parent": e.parent.as_ref().map(ContentHash::to_hex),
6611        "format_version": e.format_version,
6612        // Empty until Phase 2 (D16), and sent anyway: witnesses are
6613        // inside the hashed form, so a client that left them out of its
6614        // recomputation would verify fine today and break on the first
6615        // cosigned entry.
6616        "witnesses": &e.witnesses,
6617        // The signature itself, not just whose it is. Without the bytes
6618        // a client can only take the node's word for authorship.
6619        "author_sig_hex": e.author_sig.as_ref().map(|w| hex_encode(&w.signature)),
6620    });
6621    // D39 put three more fields inside `Witness` and D45 a fourth, and
6622    // therefore inside the hashed form. Omitting them here made a
6623    // passkey-signed entry unreproducible: a client rebuilding from the
6624    // served fields computes a different hash and cannot tell a
6625    // legitimate entry from a lying node. That is the exact failure the
6626    // witnesses comment above warns about, arriving through a different
6627    // field — the fields were added to the format and not to this shape.
6628    //
6629    // So this block grows with `Witness`, every time, and forgetting it
6630    // is silent. `sync_contract.rs` is where that is caught.
6631    //
6632    // Emitted only when present, so an ed25519 entry's served JSON is
6633    // byte-identical to what it always was and no existing client sees a
6634    // new key. That mirrors how they are serialized in the hashed form.
6635    if let Some(sig) = e.author_sig.as_ref() {
6636        let object = value.as_object_mut().expect("entry_json builds an object");
6637        if let Some(scheme) = sig.scheme {
6638            object.insert("author_scheme".into(), serde_json::json!(scheme));
6639        }
6640        if let Some(data) = sig.authenticator_data.as_ref() {
6641            object.insert(
6642                "authenticator_data_hex".into(),
6643                serde_json::json!(hex_encode(data)),
6644            );
6645        }
6646        if let Some(data) = sig.client_data_json.as_ref() {
6647            object.insert(
6648                "client_data_json_hex".into(),
6649                serde_json::json!(hex_encode(data)),
6650            );
6651        }
6652        if let Some(key) = sig.credential_key.as_ref() {
6653            object.insert(
6654                "credential_key_hex".into(),
6655                serde_json::json!(hex_encode(key)),
6656            );
6657        }
6658    }
6659    value
6660}
6661
6662/// Reads up to `take` entries starting at `from` straight out of the
6663/// persisted JSON-lines log, for readers behind the in-memory window.
6664///
6665/// Line number is sequence number, so this skips rather than parses the
6666/// prefix — O(bytes before `from`) per call, which is the price of a
6667/// resync and is paid only by readers that fell behind.
6668fn replay_from_disk(
6669    path: &std::path::Path,
6670    from: usize,
6671    take: usize,
6672) -> Result<Vec<serde_json::Value>, String> {
6673    use std::io::BufRead;
6674    let file = std::fs::File::open(path).map_err(|e| format!("open log: {e}"))?;
6675    let mut reader = std::io::BufReader::new(file);
6676
6677    // Skip by scanning for newlines rather than `.lines().skip(from)`,
6678    // which allocates and UTF-8 validates a String for every line thrown
6679    // away. Still O(bytes before `from`), but it no longer allocates
6680    // per skipped entry, and a resync from a long log is exactly the case
6681    // where "one allocation per line you do not want" is worst.
6682    //
6683    // `read_line` into a reused buffer would be the obvious fix; this
6684    // uses `read_until` on the raw bytes so the skipped prefix is never
6685    // UTF-8 checked either. The entries actually returned are still
6686    // decoded through `serde_json`, which validates them properly.
6687    let mut scratch = Vec::new();
6688    for _ in 0..from {
6689        scratch.clear();
6690        let n = reader
6691            .read_until(b'\n', &mut scratch)
6692            .map_err(|e| format!("read log: {e}"))?;
6693        if n == 0 {
6694            // `from` is past the end of the log: an empty page, not an
6695            // error. The caller already answered 409 for the case where
6696            // the reader is behind the window with no log to fall back on.
6697            return Ok(Vec::new());
6698        }
6699    }
6700
6701    let mut rows = Vec::with_capacity(take.min(LOG_PAGE));
6702    for _ in 0..take {
6703        scratch.clear();
6704        let n = reader
6705            .read_until(b'\n', &mut scratch)
6706            .map_err(|e| format!("read log: {e}"))?;
6707        if n == 0 {
6708            break;
6709        }
6710        let entry: OpEntry = serde_json::from_slice(trim_newline(&scratch))
6711            .map_err(|e| format!("decode log line: {e}"))?;
6712        rows.push(entry_json(&entry));
6713    }
6714    Ok(rows)
6715}
6716
6717/// Percent-decodes one query-string value.
6718///
6719/// `/api/reviews?reviewer=` compares its value against a channel name,
6720/// and every channel here is `operator/agent` — a name with a slash in
6721/// it. A client that escapes the slash, which the `choir` CLI does, was
6722/// answered with an empty queue rather than with its reviews: `choir
6723/// reviews` reported nothing to do to the very reviewer `choir state`
6724/// named as blocking a change. The endpoint's own tests missed it by
6725/// using single-word reviewer names, which no real channel is.
6726///
6727/// Undecodable input is returned unchanged rather than dropped: a name
6728/// that was never encoded is still a name, and the comparison it fails
6729/// is the right outcome for one that is genuinely unknown.
6730fn decode_query_value(raw: &str) -> String {
6731    let bytes = raw.as_bytes();
6732    let mut out = Vec::with_capacity(bytes.len());
6733    let mut i = 0;
6734    while i < bytes.len() {
6735        match bytes[i] {
6736            b'%' => {
6737                let Some(byte) = raw
6738                    .get(i + 1..i + 3)
6739                    .and_then(|hex| u8::from_str_radix(hex, 16).ok())
6740                else {
6741                    return raw.to_string();
6742                };
6743                out.push(byte);
6744                i += 3;
6745            }
6746            b'+' => {
6747                out.push(b' ');
6748                i += 1;
6749            }
6750            byte => {
6751                out.push(byte);
6752                i += 1;
6753            }
6754        }
6755    }
6756    String::from_utf8(out).unwrap_or_else(|_| raw.to_string())
6757}
6758
6759/// Drops a trailing `\n` and an optional preceding `\r`, so a line read
6760/// with `read_until` decodes the same as one produced by `.lines()`.
6761fn trim_newline(line: &[u8]) -> &[u8] {
6762    let line = line.strip_suffix(b"\n").unwrap_or(line);
6763    line.strip_suffix(b"\r").unwrap_or(line)
6764}
6765
6766/// JSON shape of one review's state (shared by /api/view and
6767/// /api/reviews).
6768/// One durable key binding, as `/api/view` reports it.
6769///
6770/// Keyed by actor id, so a client joins this against `author_sig.key_id`
6771/// and the `key_id` a witness carries without deriving anything.
6772///
6773/// `bound_at` is the seq of the op that *first* bound the key and never
6774/// moves, which is what makes it orderable; a later re-bind changes only
6775/// `channel`. `revoked` is present and non-null once withdrawn, and the
6776/// row survives revocation on purpose — attribution for past work must
6777/// not disappear at the moment revocation makes it interesting.
6778fn binding_json(binding: &choir_view::KeyBinding) -> serde_json::Value {
6779    serde_json::json!({
6780        "operator": binding.operator,
6781        "channel": binding.channel,
6782        "bound_at": binding.bound_at,
6783        "revoked": binding.revoked.as_ref().map(|revocation| {
6784            serde_json::json!({ "at": revocation.at, "reason": revocation.reason })
6785        }),
6786    })
6787}
6788
6789fn review_json(r: &choir_view::ReviewState) -> serde_json::Value {
6790    let verdicts: std::collections::BTreeMap<_, _> = r
6791        .verdicts
6792        .iter()
6793        .map(|(who, answer)| {
6794            (
6795                who.clone(),
6796                // `at` is served for the same reason `comments[].at` is,
6797                // plus one specific to D44: it is what decides which key
6798                // gets credited for this approval, so a client checking
6799                // the `expected` authorization a rejection hands back
6800                // cannot rederive it without this number.
6801                serde_json::json!({
6802                    "verdict": format!("{:?}", answer.verdict),
6803                    "note": answer.note,
6804                    "at": answer.at,
6805                }),
6806            )
6807        })
6808        .collect();
6809    // An array, not an object: the thread's order is the order the
6810    // sequencer admitted it, and a JSON object keyed by comment id would
6811    // invite a reader to sort by something else (D38).
6812    let comments: Vec<_> = r
6813        .comments
6814        .iter()
6815        .map(|c| {
6816            serde_json::json!({
6817                "id": c.id,
6818                "author": c.author,
6819                "body": c.body,
6820                "at": c.at,
6821            })
6822        })
6823        .collect();
6824    serde_json::json!({
6825        "target": r.target.as_ref().map(choir_oplog::ContentHash::to_hex),
6826        "target_ref": r.target_ref,
6827        "reviewers": r.reviewers,
6828        "comments": comments,
6829        "verdicts": verdicts,
6830        // viewer → fold position of that viewer's first read. What lets
6831        // an author tell "reviewed and ignored" from "nobody looked".
6832        "viewed": r.viewed,
6833        "slashes": r.slashes,
6834        "complete": r.complete(),
6835        "approved": r.approved(),
6836        "approval_weight": r.approval_weight(),
6837        "re_review_required": r.re_review_required(),
6838        // Empty reviewers on a live review means unassigned; on an
6839        // archived one it means emptied. A reader must be able to tell.
6840        "archived": matches!(r.status, choir_view::ReviewStatus::Archived { .. }),
6841    })
6842}
6843
6844/// Decodes lowercase/uppercase hex; `None` on any bad input.
6845pub fn hex_decode(s: &str) -> Option<Vec<u8>> {
6846    if !s.len().is_multiple_of(2) {
6847        return None;
6848    }
6849    (0..s.len())
6850        .step_by(2)
6851        .map(|i| u8::from_str_radix(s.get(i..i + 2)?, 16).ok())
6852        .collect()
6853}
6854
6855/// Encodes bytes as lowercase hex (client-side convenience, used by
6856/// tests and the demo).
6857pub fn hex_encode(bytes: &[u8]) -> String {
6858    bytes.iter().map(|b| format!("{b:02x}")).collect()
6859}
6860
6861#[cfg(test)]
6862mod magic_ref_tests {
6863    use super::MagicRef;
6864
6865    fn ok(refname: &str) -> MagicRef {
6866        MagicRef::parse(refname)
6867            .unwrap_or_else(|| panic!("{refname} is not recognised as a magic ref"))
6868            .unwrap_or_else(|e| panic!("{refname} refused: {e}"))
6869    }
6870
6871    fn refused(refname: &str) -> String {
6872        MagicRef::parse(refname)
6873            .unwrap_or_else(|| panic!("{refname} was not read as a magic ref at all"))
6874            .expect_err("expected a refusal")
6875    }
6876
6877    #[test]
6878    fn an_ordinary_ref_is_not_magic_at_all() {
6879        // The cost of the feature on every normal push is this `None`.
6880        assert!(MagicRef::parse("refs/heads/main").is_none());
6881        assert!(MagicRef::parse("refs/tags/v1").is_none());
6882        // Adjacent but not under the prefix.
6883        assert!(MagicRef::parse("refs/format/main/x").is_none());
6884    }
6885
6886    #[test]
6887    fn a_branch_and_topic_become_a_destination_and_a_review_id() {
6888        let magic = ok("refs/for/main/fix-parser");
6889        assert_eq!(magic.onto, "main");
6890        assert_eq!(magic.review_id, "for-main-fix-parser");
6891    }
6892
6893    #[test]
6894    fn re_pushing_the_same_topic_reaches_the_same_review() {
6895        assert_eq!(
6896            ok("refs/for/main/fix-parser").review_id,
6897            ok("refs/for/main/fix-parser").review_id
6898        );
6899    }
6900
6901    #[test]
6902    fn the_same_topic_onto_two_branches_is_two_reviews() {
6903        // Without the branch in the id, retargeting would silently
6904        // collide with somebody's proposal onto another branch.
6905        assert_ne!(
6906            ok("refs/for/main/fix").review_id,
6907            ok("refs/for/release/fix").review_id
6908        );
6909    }
6910
6911    #[test]
6912    fn a_deeper_topic_still_yields_one_path_segment() {
6913        // Review ids reach a URL, which admits one segment.
6914        let magic = ok("refs/for/main/team/fix-parser");
6915        assert!(!magic.review_id.contains('/'));
6916        assert_eq!(magic.review_id, "for-main-team-fix-parser");
6917    }
6918
6919    #[test]
6920    fn a_missing_topic_is_refused_with_the_reason() {
6921        // The case that would otherwise make every proposal onto `main`
6922        // fight over one ref.
6923        let reason = refused("refs/for/main");
6924        assert!(reason.contains("<topic>"), "{reason}");
6925        assert!(reason.contains("shared"), "{reason}");
6926    }
6927
6928    #[test]
6929    fn a_topic_that_cannot_be_a_review_id_is_refused_not_mangled() {
6930        // Silently sanitising would map two topics onto one review.
6931        let reason = refused("refs/for/main/fix parser");
6932        assert!(reason.contains("review id"), "{reason}");
6933    }
6934
6935    #[test]
6936    fn an_unsafe_branch_is_refused() {
6937        assert!(refused("refs/for/../fix").contains("branch name"));
6938    }
6939}