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}