Skip to main content

lonca_daemon/
service.rs

1//! Request handling: authentication, the single writer, side effects and
2//! the merge queue.
3
4use std::collections::{BTreeMap, BTreeSet, VecDeque};
5use std::path::{Path, PathBuf};
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::sync::{Arc, Condvar, Mutex, MutexGuard};
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use base64::Engine;
11use lonca_core::canonical::{digest_of, random_id};
12use lonca_core::keyed;
13use lonca_core::keys::KeyPair;
14use lonca_core::oplog::{OPLOG_REF, OpLog, SignedOp};
15use lonca_core::ops::{
16    Actor, Attestation, BindChecks, CheckResult, Claim, HumanKey, LeaseItem, MAX_UNPROPOSED_PATHS,
17    ObsKind, Observation, ObservationLevel, Op, Phase,
18};
19use lonca_core::store::CommitSpec;
20use lonca_core::{ChangeId, Error, FileMap, GitId, Result, Store};
21use lonca_policy::chain::Submission;
22use lonca_policy::config::{Profile, UnparsedFiles, Verification};
23use lonca_policy::decide::{evidence_problems, level_for, quota_problems, valid_token};
24use lonca_policy::events::{Event as SessionEvent, for_session, tell};
25use lonca_policy::harness::{Event, render, stop_message};
26use lonca_policy::human::{MAX_AGE_MS, MAX_SKEW_MS, check_fresh, verify_login};
27use lonca_policy::state::{ChangeRecord, ChangeStatus, Joins, State};
28use lonca_policy::{details, refs};
29use lonca_symbols::engine::ENGINE_CURRENT;
30use lonca_symbols::integrate::{facts_keyed, integrate_joining};
31use lonca_verify::{Sandbox, Verifier};
32use serde::de::DeserializeOwned;
33use serde_json::{Value, json};
34
35use crate::home::{Home, TokenEntry, Tokens};
36use crate::hooks;
37use crate::limits::{Bucket, Counters, Head};
38use crate::protocol::*;
39use crate::repo::{HUMAN_HEADER, POLICY_REF, RepoHandle, SESSION_HEADER, TRUNK_REF};
40
41pub type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;
42
43pub fn system_clock() -> Clock {
44    Arc::new(|| {
45        SystemTime::now()
46            .duration_since(UNIX_EPOCH)
47            .map(|d| d.as_millis() as i64)
48            .unwrap_or(0)
49    })
50}
51
52#[derive(Clone, Debug)]
53pub struct Settings {
54    /// The file `hook.bind` inspects instead of the one the harness's profile
55    /// names.
56    pub managed_settings: Option<PathBuf>,
57    /// Require managed settings and the hook program to be root-owned.
58    pub require_root_settings: bool,
59    pub git: String,
60    /// How long one git command may run (fetch and push included).
61    pub git_timeout: std::time::Duration,
62    /// How a freeze or an integrity alarm reaches a human at once.
63    pub notify: crate::security::Notify,
64    /// Where a local-path remote may lie, besides the backups directory.
65    /// `None`: anywhere outside the daemon home (cooperative). A hardened
66    /// daemon starts with an empty list unless told otherwise.
67    pub local_roots: Option<Vec<PathBuf>>,
68    /// Where the developer leaves Git bundles to import or restore (a
69    /// hardened install's `--import-dir`). Only a bundle directly in it is
70    /// read, and only from the daemon's own copy (`import.rs`); nothing is
71    /// ever pushed there.
72    pub import_dir: Option<PathBuf>,
73    /// The longest response line (`MAX_LINE` outside tests): `blobs`
74    /// refuses a page that would not fit.
75    pub line_cap: usize,
76    /// Where the network proxy listens (`host:port`), for `lonca run`.
77    pub proxy: Option<String>,
78    /// A hardened install: a new repository starts hardened, and one that
79    /// stands below that is told so.
80    pub hardened: bool,
81}
82
83impl Settings {
84    /// What `lonca daemon` runs with. Administrator files must be root's
85    /// unless it was told otherwise, for a test.
86    pub fn for_daemon(
87        allow_user_owned: bool,
88        managed_settings: Option<PathBuf>,
89        proxy: Option<String>,
90        hardened: bool,
91    ) -> Self {
92        Self {
93            require_root_settings: !allow_user_owned,
94            managed_settings,
95            proxy,
96            hardened,
97            notify: crate::security::Notify::Desktop,
98            ..Self::default()
99        }
100    }
101
102    /// Where the proxy listens, when it listens at an address.
103    pub fn proxy_port(&self) -> Option<u16> {
104        let addr = self.proxy.as_deref()?.parse::<std::net::SocketAddr>();
105        addr.ok().map(|a| a.port())
106    }
107}
108
109impl Default for Settings {
110    fn default() -> Self {
111        Self {
112            managed_settings: None,
113            require_root_settings: true,
114            git: "git".into(),
115            git_timeout: crate::gitcmd::DEFAULT_TIMEOUT,
116            notify: crate::security::Notify::Off,
117            local_roots: None,
118            import_dir: None,
119            line_cap: MAX_LINE,
120            proxy: None,
121            hardened: false,
122        }
123    }
124}
125
126/// What a daemon does about the sandbox it found. A hardened one does not
127/// start without one unless told to, and then it says so at every start.
128/// `Ok` carries what to log.
129pub fn sandbox_at_start(
130    hardened: bool,
131    sandbox: Sandbox,
132    allowed: bool,
133) -> std::result::Result<Option<&'static str>, &'static str> {
134    match (sandbox, hardened, allowed) {
135        (Sandbox::None, true, false) => Err(
136            "no sandbox found: a hardened daemon does not run verification without one; \
137             install bubblewrap, or start it with --allow-unsandboxed",
138        ),
139        (Sandbox::None, true, true) => Ok(Some(
140            "--allow-unsandboxed: no sandbox found, verification commands run unsandboxed",
141        )),
142        (Sandbox::None, false, _) => Ok(Some(
143            "no sandbox found: verification commands run unsandboxed",
144        )),
145        _ => Ok(None),
146    }
147}
148
149/// Where a daemon's local-path remotes may lie (`Settings::local_roots`).
150/// A hardened one reads its backups and its import directory only: git is
151/// given a `--local-root` path as it is, so a link swapped in after the
152/// check, or a repository whose `alternates` name a store, would reach the
153/// daemon home through it.
154pub fn local_roots_at_start(
155    hardened: bool,
156    roots: Vec<PathBuf>,
157) -> std::result::Result<Option<Vec<PathBuf>>, &'static str> {
158    match (hardened, roots.is_empty()) {
159        (true, false) => Err(
160            "--local-root is not for a hardened daemon: leave bundles in its import directory (--import-dir)",
161        ),
162        (true, true) => Ok(Some(roots)),
163        (false, true) => Ok(None),
164        (false, false) => Ok(Some(roots)),
165    }
166}
167
168/// How many entries a `log` answer holds, for the `limit` it was asked with.
169fn log_page(limit: Option<u64>) -> usize {
170    limit.unwrap_or(LOG_PAGE).min(LOG_PAGE) as usize
171}
172
173/// Bytes a page of blob contents may hold once base64-encoded, under a
174/// line cap: three quarters, leaving room for paths, ids and JSON.
175pub fn page_budget(line_cap: usize) -> u64 {
176    (line_cap as u64 / 4) * 3
177}
178
179/// The base64 size of `bytes` bytes.
180fn encoded_len(bytes: u64) -> u64 {
181    bytes.div_ceil(3) * 4
182}
183
184type Hr = std::result::Result<Value, (String, String)>;
185
186/// What git fetches from: the remote as given, or the daemon's copy of a
187/// bundle from the import directory (removed when this goes).
188enum Source {
189    Given(String),
190    Copied(crate::import::Copy),
191}
192
193impl Source {
194    /// What git is given. Keep the source until git is done: a copy goes
195    /// with it.
196    fn arg(&self) -> String {
197        match self {
198            Source::Given(r) => r.clone(),
199            Source::Copied(c) => c.path().to_string_lossy().into_owned(),
200        }
201    }
202}
203
204/// What [`report_unproposed`] found.
205struct LeftBehind {
206    /// The names fit to show, at most `MAX_UNPROPOSED_PATHS`.
207    paths: Vec<String>,
208    total: u64,
209    /// The rounds so far, this one included.
210    round: u32,
211    /// There are rounds left, so the agent is sent back.
212    told: bool,
213}
214
215/// A name that may be shown to a person and to a model: a workspace can hold
216/// files no repository path could be, and those still count in the total.
217fn showable(path: &str) -> bool {
218    lonca_core::paths::validate_repo_path(path).is_ok() && !path.chars().any(char::is_control)
219}
220
221/// `files`, less what a standing proposal of this workspace's sessions
222/// carries, exactly. A rejected, withdrawn or conflicted change delivered
223/// nothing, so its files are still to do.
224fn unproposed(r: &RepoHandle, session: &str, files: Vec<LocalFile>) -> Result<Vec<String>> {
225    let state = &r.chain.state;
226    let standing = state.changes.values().filter(|c| {
227        matches!(
228            c.status,
229            ChangeStatus::Pending { .. } | ChangeStatus::Queued | ChangeStatus::Integrated { .. }
230        ) && c
231            .actor
232            .session()
233            .is_some_and(|s| state.continues(session, s))
234    });
235    let mut left = files;
236    for change in standing {
237        if left.is_empty() {
238            break;
239        }
240        let tree = r.store.commit_files(&change.commit)?;
241        left.retain(|f| tree.get(&f.path) != f.entry.as_ref());
242    }
243    Ok(left.into_iter().map(|f| f.path).collect())
244}
245
246/// Puts what a session left on the record. The log takes a report only when
247/// it says something new, so an agent that stops for ever can not fill it:
248/// past `limit` an unchanged list is written once.
249fn report_unproposed(
250    r: &mut RepoHandle,
251    session: &str,
252    files: Vec<LocalFile>,
253    limit: u32,
254    now: i64,
255) -> std::result::Result<LeftBehind, (String, String)> {
256    let left = unproposed(r, session, files).or_fail()?;
257    let total = left.len() as u64;
258    let paths: Vec<String> = left
259        .into_iter()
260        .filter(|p| showable(p))
261        .take(MAX_UNPROPOSED_PATHS)
262        .collect();
263    let mark = r
264        .chain
265        .state
266        .sessions
267        .get(session)
268        .and_then(|s| s.unproposed.as_ref());
269    let rounds = mark.map_or(0, |m| m.rounds);
270    let told = total > 0 && rounds < limit;
271    // A clean report is news too, once: it tells "ended clean" from "never
272    // said" (#189).
273    let unchanged = mark.is_some_and(|m| m.total == total && m.paths == paths);
274    if told || !unchanged {
275        let op = Op::Observe {
276            session: session.to_string(),
277            events: vec![Observation {
278                phase: Phase::Post,
279                kind: ObsKind::Unproposed {
280                    paths: paths.clone(),
281                    total,
282                },
283                external: false,
284            }],
285        };
286        r.submit(Submission::new(Actor::Daemon, op), now)
287            .or_fail()?;
288    }
289    Ok(LeftBehind {
290        paths,
291        total,
292        round: rounds + u32::from(told),
293        told,
294    })
295}
296
297/// The runner-facing protocol number, kept with the client types.
298pub use lonca_client::PROTOCOL;
299
300/// Longest a `log` request with `wait_ms` is held.
301pub const LOG_WAIT_MAX_MS: i64 = 30_000;
302/// How often a held `log` request re-checks the log on its own, in case
303/// an appending path forgot to signal. Long on purpose: the signal is
304/// what makes the request return at once.
305pub const LOG_POLL_MS: u64 = 2_000;
306
307/// Most entries one `log` answer holds: a longer log is read in pages.
308pub const LOG_PAGE: u64 = 1000;
309
310/// Longest diff text `change.diff` returns; beyond it `truncated` is set.
311pub const DIFF_CAP_BYTES: usize = 2 * 1024 * 1024;
312
313/// Longest verification log `change.get` returns per check (the tail is
314/// kept). Below what the verifier records (64 KiB), so a long log is
315/// cut here and says so.
316pub const EVIDENCE_LOG_CAP_BYTES: usize = 32 * 1024;
317
318/// Length of a human session when `human.login` names none.
319pub const HUMAN_SESSION_DEFAULT_MS: i64 = 60 * 60 * 1000;
320/// The longest human session a login may ask for.
321pub const HUMAN_SESSION_MAX_MS: i64 = 24 * 60 * 60 * 1000;
322/// How long a login nonce is remembered: as long as a signature can be
323/// fresh (its age plus the clock skew allowed), so a replay inside that
324/// window is refused and nothing older needs remembering.
325pub const LOGIN_NONCE_WINDOW_MS: i64 = MAX_AGE_MS + MAX_SKEW_MS;
326
327/// What a request must carry before its method runs. Deny by default: a
328/// method not in [`METHODS`] does not exist.
329#[derive(Clone, Copy, Debug, PartialEq, Eq)]
330pub enum Cred {
331    /// Nothing.
332    Public,
333    /// A presence signature inside the request (`human.login`).
334    Login,
335    /// An agent session token; the repository is the token's.
336    Agent,
337    /// A human session whose key is enrolled in the request's repository.
338    Human,
339    /// A human session whose key is enrolled in at least one repository.
340    AnyHuman,
341    /// `AnyHuman`, or nothing while the daemon serves no repository yet
342    /// (the first `lonca init` or `lonca restore`): there is no key to
343    /// check against, and the socket is the only guard, as it was.
344    Bootstrap,
345    /// `Agent` (for the token's repository) or `Human`.
346    Read,
347    /// Any live token, agent or human.
348    Any,
349}
350
351impl Cred {
352    /// The name used in `docs/decisions.md`.
353    pub fn name(self) -> &'static str {
354        match self {
355            Cred::Public => "public",
356            Cred::Login => "login",
357            Cred::Agent => "agent",
358            Cred::Human => "human",
359            Cred::AnyHuman => "any human",
360            Cred::Bootstrap => "bootstrap",
361            Cred::Read => "read",
362            Cred::Any => "any",
363        }
364    }
365}
366
367/// How many refs one `git push` of the bridge names.
368const PUSH_REFS_PER_CALL: usize = 500;
369
370/// What the launcher of `lonca run` does with an isolated session's token,
371/// from outside the sandbox. Neither reads or changes the repository.
372const LAUNCHER_METHODS: [&str; 2] = ["session.heartbeat", "session.end"];
373
374/// Every method the daemon answers and what it requires. The table in
375/// `docs/decisions.md` ("Daemon authentication") repeats it; a test keeps
376/// the two equal.
377pub const METHODS: &[(&str, Cred)] = &[
378    ("ping", Cred::Public),
379    ("human.login", Cred::Login),
380    ("human.logout", Cred::Public),
381    ("whoami", Cred::Any),
382    ("repo.list", Cred::AnyHuman),
383    ("repo.prepare_init", Cred::Bootstrap),
384    ("repo.init", Cred::Bootstrap),
385    ("repo.restore", Cred::Bootstrap),
386    ("repo.backup", Cred::Human),
387    ("daemon.prepare_rotate", Cred::AnyHuman),
388    ("status", Cred::Read),
389    ("log", Cred::Read),
390    ("tree", Cred::Read),
391    ("blobs", Cred::Read),
392    ("change.get", Cred::Read),
393    ("change.diff", Cred::Read),
394    ("change.resolve", Cred::Read),
395    ("trunk.bundle", Cred::Read),
396    ("trunk.distance", Cred::Read),
397    ("human.prepare_change", Cred::Human),
398    ("human.prepare_policy", Cred::Human),
399    ("human.prepare_import", Cred::Human),
400    ("human.prepare_push", Cred::Human),
401    ("human.prepare_isolate", Cred::Human),
402    ("human.submit", Cred::Human),
403    ("lease.acquire", Cred::Agent),
404    ("lease.release", Cred::Agent),
405    ("lease.renew", Cred::Agent),
406    ("session.heartbeat", Cred::Agent),
407    ("change.propose", Cred::Agent),
408    ("change.withdraw", Cred::Agent),
409    ("facts.preview", Cred::Agent),
410    ("negotiate", Cred::Agent),
411    ("cap.attenuate", Cred::Agent),
412    ("lease.interest", Cred::Agent),
413    ("events", Cred::Agent),
414    ("inbox", Cred::Agent),
415    ("events.watch", Cred::Human),
416    ("session.end", Cred::Agent),
417    ("session.rotate", Cred::Agent),
418    ("hook.event", Cred::Agent),
419    ("hook.input", Cred::Agent),
420    ("hook.bind", Cred::Agent),
421    ("hook.stop", Cred::Agent),
422    ("mcp.call", Cred::Agent),
423    ("harness.setup", Cred::Agent),
424    ("queue.run", Cred::AnyHuman),
425    ("security.events", Cred::AnyHuman),
426];
427
428/// What may send a line past `limits::SHORT_LINE`: what carries files or
429/// a list the repository decides the length of. All of them take a token.
430pub const LONG_LINES: &[&str] = &[
431    "tree",
432    "blobs",
433    "human.prepare_change",
434    "human.prepare_policy",
435    "human.submit",
436    "lease.acquire",
437    "lease.interest",
438    "change.propose",
439    "facts.preview",
440    "session.end",
441    "hook.event",
442    "hook.input",
443    "hook.stop",
444];
445
446/// Agent methods that write nothing to the log, and `session.end`: a session
447/// over its rate must still be able to stop. `mcp.call` writes once in a
448/// session's life at most.
449const NOT_THROTTLED: [&str; 6] = [
450    "facts.preview",
451    "events",
452    "inbox",
453    "harness.setup",
454    "session.end",
455    "mcp.call",
456];
457
458/// Who sent a request, as far as its token says.
459#[derive(Clone, Debug)]
460enum Caller {
461    Anonymous,
462    Agent {
463        repo: String,
464        actor: Actor,
465    },
466    /// A human session: the name and the public key it was opened with.
467    Human {
468        human: String,
469        key: String,
470    },
471}
472
473/// A request that passed its method's credential check.
474struct Auth {
475    caller: Caller,
476    /// The repository the request acts on, when the method has one.
477    repo: Option<String>,
478}
479
480impl Auth {
481    fn repo(&self) -> std::result::Result<String, (String, String)> {
482        self.repo
483            .clone()
484            .ok_or_else(|| ("invalid".into(), "missing repo".into()))
485    }
486
487    /// A human session may only submit what its own human signed.
488    fn same_human(&self, signer: &str) -> std::result::Result<(), (String, String)> {
489        match &self.caller {
490            Caller::Human { human, .. } if human != signer => fail(
491                "forbidden",
492                format!("a session of {human} can not submit a signature by {signer}"),
493            ),
494            _ => Ok(()),
495        }
496    }
497}
498
499/// A session as the proxy counts it: repository and session id.
500type SessionKey = (String, String);
501
502/// What the proxy has open, by session, and what it has carried. Only in
503/// memory: nothing of it is a decision, and a restart starts it over.
504#[derive(Default)]
505struct ProxyBook {
506    /// Each connection let through, by the client's end: the sessions it
507    /// counts against, with the byte budget of each.
508    open: BTreeMap<std::net::SocketAddr, Vec<(SessionKey, u64)>>,
509    /// Bytes each session moved through the proxy.
510    carried: BTreeMap<SessionKey, u64>,
511    /// Connections no process could be named for, so every session the
512    /// proxy serves answered for them (see `lonca status`).
513    unattributed: u64,
514}
515
516impl ProxyBook {
517    fn open_of(&self, key: &SessionKey) -> usize {
518        let held = self.open.values().flatten();
519        held.filter(|(k, _)| k == key).count()
520    }
521}
522
523/// How a host reads under the policy of one session, for the proxy.
524fn host_verdict(cfg: &lonca_policy::config::Observation, host: &str) -> (bool, bool) {
525    // A host the policy names as the agent's own is no outside content: it
526    // is on the record and it taints nothing. A rule of `allow_hosts` lets
527    // its host through and says whether it taints.
528    if cfg.agent_hosts.iter().any(|h| h == host) {
529        return (true, false);
530    }
531    if let Some(rule) = cfg.host_rule(host) {
532        return (true, rule.taint);
533    }
534    (cfg.network == "taint", true)
535}
536
537#[derive(Clone, Debug, PartialEq, Eq)]
538pub struct ProxyVerdict {
539    pub allow: bool,
540    pub allow_private: bool,
541    pub reason: String,
542}
543
544/// Most ids the answer to an ambiguous prefix names.
545const AMBIGUOUS_SHOWN: usize = 5;
546
547/// The one id among `ids` that starts with `prefix`; `not_found` for none,
548/// `invalid` naming the first few for more than one.
549fn resolve_prefix<'a>(ids: impl IntoIterator<Item = &'a ChangeId>, prefix: &str) -> Hr {
550    let found: Vec<&ChangeId> = ids.into_iter().filter(|c| c.starts_with(prefix)).collect();
551    match found.as_slice() {
552        [] => fail("not_found", format!("no change starts with {prefix}")),
553        [one] => Ok(json!({"change": one})),
554        many => {
555            let shown: Vec<String> = many
556                .iter()
557                .take(AMBIGUOUS_SHOWN)
558                .map(|c| c.to_string())
559                .collect();
560            let more = if many.len() > AMBIGUOUS_SHOWN {
561                ", ..."
562            } else {
563                ""
564            };
565            fail(
566                "invalid",
567                format!(
568                    "{} changes start with {prefix}: {}{more}; give more letters",
569                    many.len(),
570                    shown.join(", ")
571                ),
572            )
573        }
574    }
575}
576
577fn fail<T>(code: &str, msg: impl Into<String>) -> std::result::Result<T, (String, String)> {
578    Err((code.into(), msg.into()))
579}
580
581fn from_err(e: Error) -> (String, String) {
582    let code = match e {
583        Error::Invalid(_) => "invalid",
584        Error::Conflict(_) => "conflict",
585        Error::Integrity(_) => "integrity",
586        Error::Unsupported(_) => "unsupported",
587        Error::Git(_) | Error::Io(_) | Error::Json(_) => "internal",
588    };
589    (code.into(), e.to_string())
590}
591
592trait OrFail<T> {
593    fn or_fail(self) -> std::result::Result<T, (String, String)>;
594}
595
596impl<T> OrFail<T> for Result<T> {
597    fn or_fail(self) -> std::result::Result<T, (String, String)> {
598        self.map_err(from_err)
599    }
600}
601
602fn params<T: DeserializeOwned>(v: &Value) -> std::result::Result<T, (String, String)> {
603    serde_json::from_value(v.clone()).map_err(|e| ("invalid".into(), format!("params: {e}")))
604}
605
606fn logged(op: &SignedOp) -> Value {
607    json!({"seq": op.entry.seq, "decision": op.entry.decision})
608}
609
610/// The daemon's signing keys. During a rotation some repositories are on
611/// the current key and some on the next one.
612struct DaemonKeys {
613    current: KeyPair,
614    next: Option<KeyPair>,
615}
616
617impl DaemonKeys {
618    /// The key new repositories and recoveries use.
619    fn for_new(&self) -> &KeyPair {
620        self.next.as_ref().unwrap_or(&self.current)
621    }
622
623    fn all(&self) -> Vec<&KeyPair> {
624        std::iter::once(self.for_new())
625            .chain(self.next.as_ref().map(|_| &self.current))
626            .collect()
627    }
628}
629
630struct Inner {
631    keys: DaemonKeys,
632    repos: BTreeMap<String, RepoHandle>,
633    /// Repositories whose log is on a daemon key this daemon does not hold.
634    /// They accept nothing but `daemon.recover`.
635    foreign: BTreeMap<String, RepoHandle>,
636    tokens: Tokens,
637    /// What each session and those made from it may still do, by repository
638    /// and the first of them. Lives as long as the daemon.
639    buckets: BTreeMap<(String, String), Bucket>,
640    /// Nonces of recent logins, so a login signature opens one session.
641    login_nonces: BTreeMap<String, i64>,
642    prepared: BTreeMap<String, Op>,
643    /// (repo, session) -> workspace root, as reported at bind time.
644    workspaces: BTreeMap<(String, String), PathBuf>,
645}
646
647/// How many trees the daemon remembers as verified.
648const VERIFIED_TREES: usize = 256;
649
650/// For how long: what a check ran on is more than its tree, a toolchain under
651/// a long-lived daemon for one.
652const VERIFIED_FOR_MS: i64 = 3_600_000;
653
654type VerifiedKey = (String, GitId, String);
655
656/// Checks that passed: evidence, so logged again, marked as reused.
657#[derive(Default)]
658struct Verified {
659    runs: BTreeMap<VerifiedKey, (i64, Vec<Attestation>)>,
660    /// In the order they came, the oldest goes first.
661    order: VecDeque<VerifiedKey>,
662}
663
664impl Verified {
665    fn get(&self, key: &VerifiedKey, now: i64) -> Option<Vec<Attestation>> {
666        let fresh = self
667            .runs
668            .get(key)
669            .filter(|(at, _)| now - at < VERIFIED_FOR_MS);
670        fresh.map(|(_, run)| run.clone())
671    }
672
673    fn insert(&mut self, key: VerifiedKey, run: &[Attestation], now: i64) {
674        let entry = (now, run.to_vec());
675        if self.runs.insert(key.clone(), entry).is_none() {
676            self.order.push_back(key);
677        }
678        if self.order.len() > VERIFIED_TREES
679            && let Some(oldest) = self.order.pop_front()
680        {
681            self.runs.remove(&oldest);
682        }
683    }
684}
685
686/// `a` as logged for a run that did not happen.
687/// The identify lines of a piece of evidence (`env.<n>.command`, `.output`,
688/// `.error`), in their order: what its `env` digest says in words.
689fn environment(details: &BTreeMap<String, String>) -> Vec<Value> {
690    (0..lonca_policy::config::MAX_IDENTIFY)
691        .map_while(|n| {
692            let get = |k: &str| details.get(&format!("env.{n}.{k}"));
693            let command = get("command")?;
694            Some(json!({"command": command, "output": get("output"), "error": get("error")}))
695        })
696        .collect()
697}
698
699fn reused(mut a: Attestation) -> Attestation {
700    a.details.insert("reused".into(), "true".into());
701    a
702}
703
704/// The sessions an `events.watch` of `human` looks at: those it names, each
705/// one this human started, or every live one this human started. Another
706/// human's session and one never started get the same refusal.
707fn watched(
708    state: &State,
709    human: &str,
710    named: Option<&[String]>,
711    now: i64,
712) -> std::result::Result<Vec<String>, (String, String)> {
713    let mine = |id: &str| {
714        state
715            .sessions
716            .get(id)
717            .is_some_and(|s| s.root_human == human)
718    };
719    match named {
720        Some(named) => match named.iter().find(|id| !mine(id)) {
721            Some(id) => fail("forbidden", format!("session {id} is not one you started")),
722            None => Ok(named.to_vec()),
723        },
724        None => {
725            let live = state.sessions.values().filter(|s| s.root_human == human);
726            let live = live.filter(|s| state.session_active(&s.id, now));
727            Ok(live.map(|s| s.id.clone()).collect())
728        }
729    }
730}
731
732/// Of the sessions an `events.watch` names, those that can not act any
733/// more, with why: the reason the first of it and its parents was ended
734/// with, or `expired`.
735fn ended_of(state: &State, named: Option<&[String]>, now: i64) -> Vec<EndedSession> {
736    let reason = |id: &str| {
737        let mut cur = state.sessions.get(id);
738        while let Some(s) = cur {
739            if let Some(r) = &s.ended {
740                return r.clone();
741            }
742            if now >= s.expires_ms {
743                return "expired".into();
744            }
745            cur = s.parent.as_deref().and_then(|p| state.sessions.get(p));
746        }
747        "expired".into()
748    };
749    let named = named.unwrap_or_default().iter();
750    let gone = named.filter(|id| !state.session_active(id, now));
751    gone.map(|id| EndedSession {
752        session: id.clone(),
753        reason: reason(id),
754    })
755    .collect()
756}
757
758/// The log from `from`, and from earlier when an approval or a rejection in
759/// it names an older request: [`for_session`] needs that request to know
760/// whose it was.
761fn ops_for_events(store: &Store, from: u64) -> Result<Vec<SignedOp>> {
762    let log = OpLog::new(store);
763    let ops = log.read_from(from)?;
764    let targets = ops.iter().filter_map(|o| match &o.entry.op {
765        Op::Approve { target_seq, .. } | Op::Reject { target_seq, .. } => Some(*target_seq),
766        _ => None,
767    });
768    match targets.min() {
769        Some(oldest) if oldest < from => log.read_from(oldest),
770        _ => Ok(ops),
771    }
772}
773
774/// What a watcher is told of `event`: its kind, entry, change and sender,
775/// never what was written.
776fn watched_event(session: &str, event: &SessionEvent) -> WatchedEvent {
777    let mut out = WatchedEvent {
778        session: session.to_string(),
779        kind: String::new(),
780        seq: None,
781        change: None,
782        from: None,
783        reservation: None,
784        until_ms: None,
785    };
786    let (kind, seq) = match event {
787        SessionEvent::Integrated { seq, change } => {
788            out.change = Some(*change);
789            ("integrated", Some(*seq))
790        }
791        SessionEvent::Conflicted { seq, change, .. } => {
792            out.change = Some(*change);
793            ("conflicted", Some(*seq))
794        }
795        SessionEvent::Failed { seq, change, .. } => {
796            out.change = Some(*change);
797            ("failed", Some(*seq))
798        }
799        SessionEvent::Approved { seq, .. } => ("approved", Some(*seq)),
800        SessionEvent::Rejected { seq, .. } => ("rejected", Some(*seq)),
801        SessionEvent::Message { seq, from, .. } => {
802            out.from.clone_from(from);
803            ("message", Some(*seq))
804        }
805        SessionEvent::LeaseReceived { seq, from, .. } => {
806            out.from.clone_from(from);
807            ("lease_received", Some(*seq))
808        }
809        SessionEvent::SymbolsChanged { seq, change, .. } => {
810            out.change = Some(*change);
811            ("symbols_changed", Some(*seq))
812        }
813        SessionEvent::Reserved { until_ms, .. } => {
814            out.until_ms = Some(*until_ms);
815            ("reserved", None)
816        }
817    };
818    out.kind = kind.into();
819    out.seq = seq;
820    out
821}
822
823/// A reservation's name for a watcher: the same while the same session
824/// holds the same items, whatever its window says.
825fn reservation_id(session: &str, items: &[LeaseItem]) -> String {
826    let mut id = digest_of(&(session, items));
827    id.truncate(16);
828    id
829}
830
831/// The change is still the proposal `commit` on `base`: withdrawn and
832/// proposed again, it can stand at the same place with other content.
833fn same_proposal(state: &State, change: ChangeId, commit: &GitId, base: &GitId) -> bool {
834    let rec = state.changes.get(&change);
835    rec.is_some_and(|c| c.commit == *commit && c.base == *base)
836}
837
838/// The changes a run is for, while it runs. An agent can not withdraw one
839/// then: the run would end unlogged, and so cost its session nothing.
840struct Running<'a> {
841    set: &'a Mutex<BTreeSet<ChangeId>>,
842    held: Vec<ChangeId>,
843}
844
845impl Running<'_> {
846    fn hold(&mut self, changes: impl IntoIterator<Item = ChangeId>) {
847        let mut set = self.set.lock().unwrap_or_else(|e| e.into_inner());
848        for c in changes {
849            set.insert(c);
850            self.held.push(c);
851        }
852    }
853}
854
855impl Drop for Running<'_> {
856    fn drop(&mut self) {
857        let mut set = self.set.lock().unwrap_or_else(|e| e.into_inner());
858        for c in &self.held {
859            set.remove(c);
860        }
861    }
862}
863
864pub struct Service {
865    pub home: Home,
866    pub counters: Counters,
867    clock: Clock,
868    verifier: Verifier,
869    /// Checks that passed, by repository, tree and configuration. It lives
870    /// as long as the daemon: its environment is part of what passed.
871    verified: Mutex<Verified>,
872    proxy: Mutex<ProxyBook>,
873    security: crate::security::SecurityLog,
874    /// Repositories whose auditor flag was already told as an alarm.
875    flagged: Mutex<BTreeSet<String>>,
876    verification_runs: AtomicU64,
877    /// Changes whose checks are running, see [`Running`].
878    running: Mutex<BTreeSet<ChangeId>>,
879    settings: Settings,
880    /// When this daemon started, by its clock: hook events sent before it
881    /// went to another daemon ([`crate::hooked::START_GRACE_MS`]).
882    started_ms: i64,
883    inner: Mutex<Inner>,
884    wake: (Mutex<u64>, Condvar),
885    /// Bumped whenever a log may have grown; `log` with `wait_ms` sleeps
886    /// on it.
887    appended: (Mutex<()>, Condvar),
888}
889
890impl Service {
891    pub fn open(home: Home, settings: Settings, clock: Clock) -> Result<Arc<Self>> {
892        home.ensure()?;
893        // What a daemon killed during a copy left behind (`import.rs`).
894        let _ = std::fs::remove_dir_all(home.imports());
895        let (current, next) = home.load_keys()?;
896        let keys = DaemonKeys { current, next };
897        let tokens = Tokens::load(&home.tokens_path())?;
898        let mut repos = BTreeMap::new();
899        let mut foreign = BTreeMap::new();
900        for entry in std::fs::read_dir(home.repos())? {
901            let path = entry?.path();
902            let Some(name) = path
903                .file_name()
904                .and_then(|n| n.to_str())
905                .and_then(|n| n.strip_suffix(".git"))
906            else {
907                continue;
908            };
909            let r = RepoHandle::open(name, &path, &keys.all())?;
910            if r.foreign() {
911                eprintln!(
912                    "lonca: repository {name} is on another daemon key; \
913                     recover it with `lonca daemon-key recover --repo {name}`"
914                );
915                foreign.insert(name.to_string(), r);
916            } else {
917                repos.insert(name.to_string(), r);
918            }
919        }
920        let svc = Arc::new(Self {
921            verifier: Verifier::new(home.scratch()),
922            verified: Mutex::default(),
923            verification_runs: AtomicU64::new(0),
924            proxy: Mutex::default(),
925            security: crate::security::SecurityLog::new(&home.root, settings.notify.clone()),
926            flagged: Mutex::default(),
927            running: Mutex::default(),
928            home,
929            counters: Counters::default(),
930            started_ms: clock(),
931            clock,
932            settings,
933            inner: Mutex::new(Inner {
934                keys,
935                repos,
936                foreign,
937                tokens,
938                buckets: BTreeMap::new(),
939                login_nonces: BTreeMap::new(),
940                prepared: BTreeMap::new(),
941                workspaces: BTreeMap::new(),
942            }),
943            wake: (Mutex::new(0), Condvar::new()),
944            appended: (Mutex::new(()), Condvar::new()),
945        });
946        {
947            let mut inner = svc.lock();
948            let now = svc.now();
949            for r in inner.repos.values_mut() {
950                if r.initialized() {
951                    Self::reconcile(r, now)
952                        .map_err(|e| Error::Integrity(format!("{}: {}", e.0, e.1)))?;
953                }
954            }
955            svc.promote_if_done(&mut inner)?;
956            Self::sweep_tokens(&mut inner, now)
957                .map_err(|e| Error::Invalid(format!("{}: {}", e.0, e.1)))?;
958        }
959        Ok(svc)
960    }
961
962    /// How this daemon runs git: a clean environment, its own git home,
963    /// a time limit ([`crate::gitcmd`]).
964    fn git_runner(&self) -> crate::gitcmd::Git {
965        crate::gitcmd::Git {
966            bin: self.settings.git.clone(),
967            home: self.home.root.join("git-home"),
968            timeout: self.settings.git_timeout,
969        }
970    }
971
972    fn git(
973        &self,
974        git_dir: &std::path::Path,
975        args: &[&str],
976    ) -> std::result::Result<String, (String, String)> {
977        use crate::gitcmd::Failure;
978        self.git_runner().run(git_dir, args).map_err(|f| match f {
979            Failure::Start(e) => ("internal".to_string(), format!("git: {e}")),
980            Failure::Timeout(t) => (
981                "git".to_string(),
982                format!(
983                    "git did not finish within {} s and was stopped",
984                    t.as_secs()
985                ),
986            ),
987            Failure::Exit(said) => ("git".to_string(), said),
988        })
989    }
990
991    /// A remote the daemon may fetch from or push to. A local path must
992    /// exist and lie outside the daemon home (its backups excepted): the
993    /// stores of other repositories, the key and the tokens live there, and
994    /// git would read them with the daemon's rights. Where
995    /// `local_roots` is set, it must lie under one of them, or in the
996    /// backups.
997    fn check_remote(&self, r: &str) -> std::result::Result<(), (String, String)> {
998        validate_remote(r)?;
999        if !r.starts_with('/') {
1000            return Ok(());
1001        }
1002        let refuse = |why: &str| fail("forbidden", format!("local remote {r}: {why}"));
1003        let Ok(path) = std::fs::canonicalize(r) else {
1004            return refuse("it does not exist");
1005        };
1006        let canon = |p: &Path| std::fs::canonicalize(p).unwrap_or_else(|_| p.to_path_buf());
1007        let backups = canon(&self.home.backups());
1008        if path.starts_with(&backups) {
1009            return Ok(());
1010        }
1011        if let Some(dir) = &self.settings.import_dir
1012            && path.starts_with(canon(dir))
1013        {
1014            return refuse("the import directory is only read, from a bundle copied first");
1015        }
1016        if path.starts_with(canon(&self.home.root)) {
1017            return refuse("inside the daemon home");
1018        }
1019        match &self.settings.local_roots {
1020            None => Ok(()),
1021            Some(roots) if roots.iter().any(|root| path.starts_with(canon(root))) => Ok(()),
1022            Some(_) => {
1023                refuse("outside the directories this daemon may read (`lonca daemon --local-root`)")
1024            }
1025        }
1026    }
1027
1028    /// A remote to fetch from: `check_remote`, or a bundle directly in the
1029    /// import directory, which git reads from the daemon's copy. Anything
1030    /// else below the import directory is refused.
1031    fn source(&self, r: &str) -> std::result::Result<Source, (String, String)> {
1032        validate_remote(r)?;
1033        let Some(dir) = self.import_dir_given(r) else {
1034            self.check_remote(r)?;
1035            return Ok(Source::Given(r.to_string()));
1036        };
1037        let refuse = |why: &str| fail("forbidden", format!("local remote {r}: {why}"));
1038        let path = Path::new(r);
1039        let canon = |p: &Path| std::fs::canonicalize(p).ok();
1040        let (Some(name), Some(parent)) = (path.file_name(), path.parent()) else {
1041            return refuse("name a bundle directly in the import directory");
1042        };
1043        if canon(parent).as_deref() != Some(dir.as_path()) {
1044            return refuse("name a bundle directly in the import directory");
1045        }
1046        let into = self.home.imports();
1047        match crate::import::copy_bundle(
1048            &dir,
1049            name,
1050            &into,
1051            &random_id("imp"),
1052            crate::import::MAX_BUNDLE,
1053        ) {
1054            Ok(copy) => Ok(Source::Copied(copy)),
1055            Err(why) => refuse(&why),
1056        }
1057    }
1058
1059    /// The import directory, resolved, when `r` names something below it.
1060    fn import_dir_given(&self, r: &str) -> Option<PathBuf> {
1061        let dir = std::fs::canonicalize(self.settings.import_dir.as_ref()?).ok()?;
1062        let path = Path::new(r);
1063        let below = |p: &Path| p.starts_with(&dir);
1064        let parent = path.parent().and_then(|p| std::fs::canonicalize(p).ok());
1065        (below(path) || parent.as_deref().is_some_and(below)).then_some(dir)
1066    }
1067
1068    /// A branch to fetch: Lonca's own test, then git's.
1069    fn check_branch(&self, b: &str) -> std::result::Result<(), (String, String)> {
1070        validate_branch(b)?;
1071        if !crate::gitcmd::branch_ok(&self.git_runner(), b) {
1072            return fail("invalid", "invalid branch name");
1073        }
1074        Ok(())
1075    }
1076
1077    fn lock(&self) -> MutexGuard<'_, Inner> {
1078        self.inner.lock().unwrap_or_else(|e| e.into_inner())
1079    }
1080
1081    pub fn now(&self) -> i64 {
1082        (self.clock)()
1083    }
1084
1085    /// The daemon's current key.
1086    pub fn public_key(&self) -> String {
1087        self.lock().keys.current.public_hex()
1088    }
1089
1090    /// The key of an unfinished rotation.
1091    pub fn next_key(&self) -> Option<String> {
1092        self.lock().keys.next.as_ref().map(KeyPair::public_hex)
1093    }
1094
1095    /// Ends a rotation once no initialized repository uses the current key:
1096    /// the next key becomes the current one and the old secret is deleted.
1097    fn promote_if_done(&self, inner: &mut Inner) -> Result<()> {
1098        let Some(next) = inner.keys.next.clone() else {
1099            return Ok(());
1100        };
1101        let current = inner.keys.current.public_hex();
1102        if inner
1103            .repos
1104            .values()
1105            .any(|r| r.initialized() && r.chain.public_key() == current)
1106        {
1107            return Ok(());
1108        }
1109        self.home.promote_next_key()?;
1110        inner.keys = DaemonKeys {
1111            current: next.clone(),
1112            next: None,
1113        };
1114        // A prepared genesis named the old key: it must be prepared again.
1115        for r in inner.repos.values_mut().filter(|r| !r.initialized()) {
1116            r.chain.set_key(next.clone());
1117            inner.prepared.remove(&r.name);
1118        }
1119        Ok(())
1120    }
1121
1122    fn notify_queue(&self) {
1123        let (m, c) = &self.wake;
1124        *m.lock().unwrap_or_else(|e| e.into_inner()) += 1;
1125        c.notify_all();
1126    }
1127
1128    /// Wakes `log` requests waiting for an entry. Called after anything
1129    /// that may have appended; a missed call costs a waiter at most one
1130    /// poll slice ([`LOG_POLL_MS`]), never the entry. The waiter re-reads
1131    /// the log's state under the lock, so the signal carries no value.
1132    fn notify_appended(&self) {
1133        let (m, c) = &self.appended;
1134        let _held = m.lock().unwrap_or_else(|e| e.into_inner());
1135        c.notify_all();
1136    }
1137
1138    /// A request from nothing known to be in a sandbox.
1139    pub fn handle(&self, req: Request) -> Response {
1140        self.handle_from(req, &|| false)
1141    }
1142
1143    /// A request; `in_sandbox` asks the operating system about whoever sent
1144    /// it, which is only done when it matters.
1145    pub fn handle_from(&self, req: Request, in_sandbox: &dyn Fn() -> bool) -> Response {
1146        let r = match self.dispatch(&req, in_sandbox) {
1147            Ok(v) => Response::ok(req.id, v),
1148            Err((code, msg)) => {
1149                self.refusal(&req, &code, &msg);
1150                Response::err(req.id, &code, msg)
1151            }
1152        };
1153        self.take_alarms();
1154        if !matches!(
1155            req.method.as_str(),
1156            "ping"
1157                | "status"
1158                | "log"
1159                | "tree"
1160                | "change.get"
1161                | "whoami"
1162                | "repo.list"
1163                | "events.watch"
1164        ) {
1165            self.notify_appended();
1166        }
1167        r
1168    }
1169
1170    /// A refusal that is a security event: who was refused what, never
1171    /// with the token or the request.
1172    fn refusal(&self, req: &Request, code: &str, message: &str) {
1173        use crate::security::kind;
1174        let k = match code {
1175            "unauthenticated" | "forbidden" => kind::AUTH_FAILURE,
1176            "rate_limited" => kind::RATE_LIMITED,
1177            "not_in_sandbox" => kind::OUTSIDE_SANDBOX,
1178            _ => return,
1179        };
1180        let mut e = crate::security::Event::new(self.now(), k)
1181            .method(&req.method)
1182            .code(code)
1183            .detail(message);
1184        if let Some(repo) = &req.repo {
1185            e = e.repo(repo);
1186        }
1187        self.security.record(e);
1188    }
1189
1190    /// Records a security event.
1191    pub fn security_event(&self, event: crate::security::Event) {
1192        self.security.record(event);
1193    }
1194
1195    /// Moves the events of entries appended since into the security log.
1196    fn take_alarms(&self) {
1197        let taken: Vec<crate::security::Event> = {
1198            let mut guard = self.lock();
1199            let inner = &mut *guard;
1200            let repos = inner.repos.values_mut().chain(inner.foreign.values_mut());
1201            repos.flat_map(|r| std::mem::take(&mut r.alarms)).collect()
1202        };
1203        for e in taken {
1204            self.security.record(e);
1205        }
1206    }
1207
1208    /// The security events of the last `since_ms` (default a day): counts by
1209    /// kind, and the latest 100, of the repositories the human is enrolled
1210    /// in and of none.
1211    fn security_events(&self, auth: &Auth, p: SecurityEvents) -> Hr {
1212        let Caller::Human { human, key } = &auth.caller else {
1213            return fail("forbidden", "a human session is required");
1214        };
1215        let mine = {
1216            let inner = self.lock();
1217            Self::enrolled_repos(&inner, human, key)
1218        };
1219        let since = p.since_ms.unwrap_or_else(|| self.now() - 86_400_000);
1220        let events: Vec<crate::security::Event> = self
1221            .security
1222            .read_since(since)
1223            .into_iter()
1224            .filter(|e| e.repo.as_ref().is_none_or(|r| mine.contains(r)))
1225            .collect();
1226        let counts = crate::security::summary(&events);
1227        let latest = &events[events.len().saturating_sub(100)..];
1228        Ok(json!({"since_ms": since, "counts": counts, "events": latest}))
1229    }
1230
1231    /// An isolated session's token works in a sandbox only, but for what
1232    /// `lonca run` itself does for it from outside. Refused elsewhere, and
1233    /// the session pays: the level no longer says what happened.
1234    fn hold_to_sandbox(
1235        &self,
1236        req: &Request,
1237        in_sandbox: &dyn Fn() -> bool,
1238    ) -> std::result::Result<(), (String, String)> {
1239        if LAUNCHER_METHODS.contains(&req.method.as_str()) {
1240            return Ok(());
1241        }
1242        let mut guard = self.lock();
1243        let now = self.now();
1244        let (repo, actor) = Self::agent(&guard, req, now)?;
1245        let session = actor.session().expect("agent").to_string();
1246        let r = Self::repo(&mut guard, &repo)?;
1247        let held = r.chain.state.sessions.get(&session);
1248        let Some(s) = held.filter(|s| s.level == ObservationLevel::Isolated) else {
1249            return Ok(());
1250        };
1251        let clean = s.taint.is_none();
1252        if in_sandbox() {
1253            return Ok(());
1254        }
1255        if clean {
1256            let t = Op::Taint {
1257                session,
1258                reason: "its token was used outside its sandbox".into(),
1259            };
1260            r.submit(Submission::new(Actor::Daemon, t), now).or_fail()?;
1261        }
1262        fail(
1263            "not_in_sandbox",
1264            "this session runs in a sandbox, and this request came from outside it",
1265        )
1266    }
1267
1268    /// A session past `[sessions] ops_per_minute` goes no further, and the
1269    /// log does not grow. This is no decision: nothing of it is logged, so
1270    /// `decide()` and a replay never see it. Sessions made from one another
1271    /// spend from one bucket, or a child would be a way round it.
1272    fn throttle(&self, req: &Request) -> std::result::Result<(), (String, String)> {
1273        let mut guard = self.lock();
1274        let inner = &mut *guard;
1275        let now = self.now();
1276        let (repo, actor) = Self::agent(inner, req, now)?;
1277        let session = actor.session().expect("agent");
1278        let r = Self::repo(inner, &repo)?;
1279        let rate = r.policy().or_fail()?.config.sessions.ops_per_minute as u64;
1280        let root = r.chain.state.root_session(session).to_string();
1281        let bucket = inner.buckets.entry((repo, root));
1282        let bucket = bucket.or_insert_with(|| Bucket::full(rate, now));
1283        bucket.take(rate, now).map_err(|wait_ms| {
1284            Counters::add(&self.counters.rate_limited);
1285            let msg = format!("this session is over its rate; try again in {wait_ms} ms");
1286            ("rate_limited".to_string(), msg)
1287        })
1288    }
1289
1290    /// Whether a line may go on past `limits::SHORT_LINE`, from what its
1291    /// first bytes say.
1292    pub fn long_line(&self, head: &Head) -> std::result::Result<(), (String, String)> {
1293        let method = head.method.as_deref().unwrap_or_default();
1294        let found = METHODS.iter().find(|(m, _)| *m == method);
1295        let Some(&(_, cred)) = found.filter(|_| LONG_LINES.contains(&method)) else {
1296            return fail("too_large", "this request line is too long");
1297        };
1298        if head.token.is_none() {
1299            return fail("too_large", "a long line names its method and token first");
1300        }
1301        let req = Request {
1302            id: head.id,
1303            method: method.into(),
1304            repo: head.repo.clone(),
1305            token: head.token.clone(),
1306            params: Value::Null,
1307        };
1308        self.authorize(&req, cred).map(|_| ())
1309    }
1310
1311    fn dispatch(&self, req: &Request, in_sandbox: &dyn Fn() -> bool) -> Hr {
1312        let p = &req.params;
1313        let Some(&(_, cred)) = METHODS.iter().find(|(m, _)| *m == req.method) else {
1314            return fail("unknown_method", format!("unknown method {}", req.method));
1315        };
1316        // The human operations that need no session: those signed with the
1317        // recovery key, as the key that would open one is the key that was
1318        // lost. The signature is checked by decide(), like any signature.
1319        let by_recovery_key = p["op"]["op"] == "human.recover"
1320            || (p["op"]["op"] == "daemon.recover" && !p["op"]["recover"].is_null());
1321        let cred = if req.method == "human.submit" && by_recovery_key {
1322            Cred::Public
1323        } else {
1324            cred
1325        };
1326        let auth = self.authorize(req, cred)?;
1327        if cred == Cred::Agent {
1328            self.hold_to_sandbox(req, in_sandbox)?;
1329            if !NOT_THROTTLED.contains(&req.method.as_str()) {
1330                self.throttle(req)?;
1331            }
1332        }
1333        match req.method.as_str() {
1334            "ping" => Ok(json!({
1335                "version": env!("CARGO_PKG_VERSION"),
1336                "protocol": PROTOCOL,
1337                "daemon_key": self.public_key(),
1338                "next_key": self.next_key(),
1339                "proxy": self.settings.proxy,
1340                "hardened": self.settings.hardened,
1341                "refused": self.counters.to_json(),
1342            })),
1343            "human.login" => self.login(req, params(p)?),
1344            "human.logout" => self.logout(req),
1345            "daemon.prepare_rotate" => self.prepare_rotate(),
1346            "repo.list" => self.repo_list(&auth),
1347            "security.events" => self.security_events(&auth, params(p)?),
1348            "repo.prepare_init" => self.prepare_init(params(p)?),
1349            "repo.init" => self.init(&auth, params(p)?),
1350            "repo.backup" => self.backup(&auth),
1351            "repo.restore" => self.restore(params(p)?),
1352            "status" => self.status(&auth),
1353            "log" => self.log(&auth, params(p)?),
1354            "tree" => self.tree(&auth, params(p)?),
1355            "blobs" => self.blobs(&auth, params(p)?),
1356            "change.get" => self.change_get(&auth, params(p)?),
1357            "change.diff" => self.change_diff(&auth, params(p)?),
1358            "change.resolve" => self.change_resolve(&auth, params(p)?),
1359            "trunk.bundle" => self.trunk_bundle(&auth, params(p)?),
1360            "trunk.distance" => self.trunk_distance(&auth, params(p)?),
1361            "whoami" => self.whoami(&auth),
1362            "human.prepare_change" => self.prepare_change(&auth, params(p)?),
1363            "human.prepare_policy" => self.prepare_policy(&auth, params(p)?),
1364            "human.prepare_import" => self.prepare_import(&auth, params(p)?),
1365            "human.prepare_push" => self.prepare_push(&auth, params(p)?),
1366            "human.prepare_isolate" => self.prepare_isolate(&auth, params(p)?),
1367            "human.submit" => self.human_submit(&auth, params(p)?),
1368            "lease.acquire" => self.lease_acquire(req, params(p)?),
1369            "lease.release" => {
1370                let s: SetRef = params(p)?;
1371                self.agent_op(req, |_| Ok(Op::LeaseRelease { set: s.set.clone() }))
1372            }
1373            "lease.renew" => {
1374                let s: SetRef = params(p)?;
1375                let ttl = s.ttl_ms.unwrap_or(10 * 60_000);
1376                self.agent_op(req, |_| {
1377                    Ok(Op::LeaseRenew {
1378                        set: s.set.clone(),
1379                        ttl_ms: ttl,
1380                    })
1381                })
1382            }
1383            "change.propose" => self.propose(req, params(p)?),
1384            "change.withdraw" => {
1385                let c: ChangeRef = params(p)?;
1386                // Asked under the lock that the queue takes the change under.
1387                self.agent_op(req, |_| {
1388                    if self.running_lock().contains(&c.change) {
1389                        return Err((
1390                            "verifying".into(),
1391                            "its checks are running; withdraw it once they end".into(),
1392                        ));
1393                    }
1394                    Ok(Op::ChangeWithdraw { change: c.change })
1395                })
1396            }
1397            "facts.preview" => self.facts_preview(req, params(p)?),
1398            "negotiate" => {
1399                let n: Negotiate = params(p)?;
1400                self.agent_op(req, |_| {
1401                    Ok(Op::Negotiate {
1402                        thread: n.thread.clone(),
1403                        to_session: n.to_session.clone(),
1404                        message: n.message.clone(),
1405                    })
1406                })
1407            }
1408            "cap.attenuate" => self.attenuate(req, params(p)?),
1409            "lease.interest" => self.lease_interest(req, params(p)?),
1410            "events" => self.events(req, params(p)?),
1411            "inbox" => self.inbox(req),
1412            "events.watch" => self.watch(&auth, params(p)?),
1413            "session.end" => self.agent_end(req, params(p)?),
1414            "session.heartbeat" => self.heartbeat(req),
1415            "session.rotate" => self.rotate(req),
1416            "hook.event" => self.hook_event(req, params(p)?),
1417            "hook.input" => self.hook_input(req, params(p)?),
1418            "hook.bind" => self.hook_bind(req, params(p)?),
1419            "hook.stop" => self.hook_stop(req, params(p)?),
1420            "mcp.call" => self.mcp_call(req, params(p)?),
1421            "harness.setup" => self.harness_setup(req, params(p)?),
1422            "queue.run" => {
1423                self.notify_queue();
1424                Ok(json!({"integrated": 0, "woken": true}))
1425            }
1426            m => fail("unknown_method", format!("unknown method {m}")),
1427        }
1428    }
1429
1430    fn repo_name(req: &Request) -> std::result::Result<String, (String, String)> {
1431        req.repo
1432            .clone()
1433            .ok_or_else(|| ("invalid".into(), "missing repo".into()))
1434    }
1435
1436    fn repo<'a>(
1437        inner: &'a mut Inner,
1438        name: &str,
1439    ) -> std::result::Result<&'a mut RepoHandle, (String, String)> {
1440        if let Some(r) = inner.foreign.get(name) {
1441            return fail(
1442                "foreign_key",
1443                format!(
1444                    "repository {name} is on daemon key {}, which this daemon does not hold; \
1445                     recover it with `lonca daemon-key recover --repo {name}`, or, if your \
1446                     key was lost too, `lonca human recover --repo {name} --recovery-key <file>`",
1447                    r.chain.state.daemon_key.get(..16).unwrap_or_default()
1448                ),
1449            );
1450        }
1451        match inner.repos.get_mut(name) {
1452            Some(r) if r.initialized() => Ok(r),
1453            _ => fail("not_found", format!("no repository {name}")),
1454        }
1455    }
1456
1457    /// Resolves the request's token. A token that is unknown, ended or
1458    /// expired is refused; no token is `Anonymous`, and the method's
1459    /// credential decides whether that is enough.
1460    fn caller(
1461        inner: &Inner,
1462        req: &Request,
1463        now: i64,
1464    ) -> std::result::Result<Caller, (String, String)> {
1465        let Some(token) = req.token.as_deref() else {
1466            return Ok(Caller::Anonymous);
1467        };
1468        match inner.tokens.lookup(token).cloned() {
1469            None if token.starts_with("lonca_h_") => fail(
1470                "unauthenticated",
1471                "unknown human session token: it expired or was ended; log in again (`lonca human login`)",
1472            ),
1473            None => fail(
1474                "unauthenticated",
1475                "unknown session token: its session ended or expired, or it has a new token; \
1476                 a human continues the workspace with `lonca session start --continue`",
1477            ),
1478            Some(TokenEntry::Agent {
1479                repo,
1480                session,
1481                expires_ms,
1482            }) => {
1483                let past_end = expires_ms.is_some_and(|end| end <= now);
1484                let Some(r) = inner.repos.get(&repo) else {
1485                    return fail("unauthenticated", "token for unknown repository");
1486                };
1487                let Some(s) = r.chain.state.sessions.get(&session) else {
1488                    return fail("unauthenticated", "unknown session");
1489                };
1490                if s.ended.is_some() {
1491                    return fail("unauthenticated", "session ended");
1492                }
1493                // Expired, or under a parent that is over: nothing works any
1494                // more but what refuses such a session on its own terms
1495                // (`session.heartbeat`) or tidies up after it (`session.end`).
1496                if (past_end || !r.chain.state.session_active(&session, now))
1497                    && !LAUNCHER_METHODS.contains(&req.method.as_str())
1498                {
1499                    return fail(
1500                        "unauthenticated",
1501                        "session expired; a human continues the workspace with \
1502                         `lonca session start --continue`",
1503                    );
1504                }
1505                Ok(Caller::Agent {
1506                    repo,
1507                    actor: Actor::Agent {
1508                        name: s.agent.clone(),
1509                        session,
1510                    },
1511                })
1512            }
1513            Some(TokenEntry::Human {
1514                human,
1515                key,
1516                expires_ms,
1517            }) => {
1518                if expires_ms <= now {
1519                    return fail("unauthenticated", "human session expired");
1520                }
1521                Ok(Caller::Human { human, key })
1522            }
1523        }
1524    }
1525
1526    /// Resolves an agent token to (repo, actor).
1527    fn agent(
1528        inner: &Inner,
1529        req: &Request,
1530        now: i64,
1531    ) -> std::result::Result<(String, Actor), (String, String)> {
1532        match Self::caller(inner, req, now)? {
1533            Caller::Agent { repo, actor } => Ok((repo, actor)),
1534            Caller::Human { .. } => fail("forbidden", "an agent session token is required"),
1535            Caller::Anonymous => fail("unauthenticated", "missing session token"),
1536        }
1537    }
1538
1539    /// Is this human, with this key, enrolled in the repository? Foreign
1540    /// repositories count: `daemon.recover` is a human operation there.
1541    fn enrolled(inner: &Inner, human: &str, key: &str, repo: &str) -> bool {
1542        inner
1543            .repos
1544            .get(repo)
1545            .or_else(|| inner.foreign.get(repo))
1546            .and_then(|r| r.chain.state.humans.get(human))
1547            .is_some_and(|k| k.public == key)
1548    }
1549
1550    /// The repositories a human session is enrolled in.
1551    fn enrolled_repos(inner: &Inner, human: &str, key: &str) -> Vec<String> {
1552        inner
1553            .repos
1554            .keys()
1555            .chain(inner.foreign.keys())
1556            .filter(|r| Self::enrolled(inner, human, key, r))
1557            .cloned()
1558            .collect()
1559    }
1560
1561    /// No repository yet: nothing to check a human against.
1562    fn bootstrap(inner: &Inner) -> bool {
1563        inner.repos.values().all(|r| !r.initialized()) && inner.foreign.is_empty()
1564    }
1565
1566    /// Checks the request's token against what the method requires and
1567    /// settles which repository the request acts on.
1568    fn authorize(&self, req: &Request, cred: Cred) -> std::result::Result<Auth, (String, String)> {
1569        let inner = self.lock();
1570        let caller = Self::caller(&inner, req, self.now())?;
1571        const AGENT: &str = "an agent session token is required";
1572        const HUMAN: &str = "a human session is required (lonca human login)";
1573        const EITHER: &str =
1574            "a session token is required: an agent's, or a human session (lonca human login)";
1575        let repo = match (cred, &caller) {
1576            (Cred::Public | Cred::Login, _) => req.repo.clone(),
1577            (Cred::Any, Caller::Anonymous) => return fail("unauthenticated", EITHER),
1578            (Cred::Any, Caller::Human { .. }) => req.repo.clone(),
1579            (Cred::Agent | Cred::Read | Cred::Any, Caller::Agent { repo, .. }) => {
1580                if req.repo.as_ref().is_some_and(|r| r != repo) {
1581                    return fail("forbidden", "this session token is for another repository");
1582                }
1583                Some(repo.clone())
1584            }
1585            (Cred::Agent, Caller::Human { .. }) => return fail("forbidden", AGENT),
1586            (Cred::Agent, Caller::Anonymous) => return fail("unauthenticated", AGENT),
1587            (Cred::Human | Cred::Read, Caller::Human { human, key }) => {
1588                let name = Self::repo_name(req)?;
1589                if !Self::enrolled(&inner, human, key, &name) {
1590                    return fail(
1591                        "forbidden",
1592                        format!("this human session is not enrolled in repository {name}"),
1593                    );
1594                }
1595                Some(name)
1596            }
1597            (Cred::Read, Caller::Anonymous) => return fail("unauthenticated", EITHER),
1598            (Cred::Human | Cred::AnyHuman | Cred::Bootstrap, Caller::Agent { .. }) => {
1599                return fail("forbidden", HUMAN);
1600            }
1601            (Cred::AnyHuman | Cred::Bootstrap, Caller::Human { human, key }) => {
1602                if Self::enrolled_repos(&inner, human, key).is_empty() {
1603                    return fail(
1604                        "forbidden",
1605                        "this human session is not enrolled in any repository",
1606                    );
1607                }
1608                req.repo.clone()
1609            }
1610            (Cred::Bootstrap, Caller::Anonymous) if Self::bootstrap(&inner) => req.repo.clone(),
1611            (Cred::Human | Cred::AnyHuman | Cred::Bootstrap, Caller::Anonymous) => {
1612                return fail("unauthenticated", HUMAN);
1613            }
1614        };
1615        Ok(Auth { caller, repo })
1616    }
1617
1618    // ---------------------------------------------------------------- human sessions
1619
1620    /// Opens a human session: the signature must verify with the key the
1621    /// named repository (or, without one, any repository) holds for the
1622    /// human. The token is returned once and stored as a hash.
1623    fn login(&self, req: &Request, p: HumanLogin) -> Hr {
1624        let ttl = p.ttl_ms.unwrap_or(HUMAN_SESSION_DEFAULT_MS);
1625        if ttl <= 0 || ttl > HUMAN_SESSION_MAX_MS {
1626            return fail(
1627                "invalid",
1628                format!("ttl_ms must be between 1 and {HUMAN_SESSION_MAX_MS}"),
1629            );
1630        }
1631        let now = self.now();
1632        check_fresh(&p.sig, now).map_err(|e| ("unauthenticated".to_string(), e.to_string()))?;
1633        let mut guard = self.lock();
1634        let inner = &mut *guard;
1635        inner
1636            .login_nonces
1637            .retain(|_, issued| *issued >= now - LOGIN_NONCE_WINDOW_MS);
1638        if inner.login_nonces.contains_key(&p.sig.nonce) {
1639            return fail("unauthenticated", "login signature already used");
1640        }
1641        let human = &p.sig.human;
1642        let repo = req.repo.as_deref();
1643        let mut keys: Vec<HumanKey> = Vec::new();
1644        if valid_token(human, 64) {
1645            let handles = inner.repos.values().chain(inner.foreign.values());
1646            for r in handles.filter(|r| repo.is_none_or(|n| n == r.name)) {
1647                if let Some(k) = r.chain.state.humans.get(human)
1648                    && !keys.contains(k)
1649                {
1650                    keys.push(k.clone());
1651                }
1652            }
1653        }
1654        let Some(key) = keys
1655            .iter()
1656            .find(|k| verify_login(k, repo, ttl, &p.sig, now).is_ok())
1657        else {
1658            return fail(
1659                "unauthenticated",
1660                "login failed: no enrolled key of that human verifies this signature",
1661            );
1662        };
1663        inner
1664            .login_nonces
1665            .insert(p.sig.nonce.clone(), p.sig.issued_ms);
1666        Self::sweep_tokens(inner, now)?;
1667        let expires_ms = now + ttl;
1668        let token = inner
1669            .tokens
1670            .issue_human(human, &key.public, expires_ms)
1671            .or_fail()?;
1672        Ok(json!({"token": token, "human": human, "expires_ms": expires_ms}))
1673    }
1674
1675    /// Ends the human session whose token the request carries. Agents end
1676    /// theirs with `session.end`, which is logged.
1677    fn logout(&self, req: &Request) -> Hr {
1678        let Some(token) = req.token.as_deref() else {
1679            return fail("invalid", "missing token");
1680        };
1681        let mut inner = self.lock();
1682        if matches!(inner.tokens.lookup(token), Some(TokenEntry::Agent { .. })) {
1683            return fail("forbidden", "agents end their session with session.end");
1684        }
1685        inner.tokens.revoke(token).or_fail()?;
1686        Ok(json!({"ended": true}))
1687    }
1688
1689    /// The repositories the human session is enrolled in and this daemon
1690    /// signs for. Foreign ones wait for `daemon.recover` and are not
1691    /// listed, as before.
1692    fn repo_list(&self, auth: &Auth) -> Hr {
1693        let Caller::Human { human, key } = &auth.caller else {
1694            return fail("forbidden", "a human session is required");
1695        };
1696        let inner = self.lock();
1697        let repos: Vec<&String> = inner
1698            .repos
1699            .keys()
1700            .filter(|r| Self::enrolled(&inner, human, key, r))
1701            .collect();
1702        Ok(json!(repos))
1703    }
1704
1705    // ---------------------------------------------------------------- init
1706
1707    fn prepare_init(&self, p: PrepareInit) -> Hr {
1708        if !lonca_policy::decide::valid_token(&p.repo, 64)
1709            || !lonca_policy::decide::valid_token(&p.human, 64)
1710        {
1711            return fail(
1712                "invalid",
1713                "repository and human names: letters, digits, '.', '_', '-'",
1714            );
1715        }
1716        let (profile, presence) = self.starting_profile(&p)?;
1717        // A copy from the import directory is made before the lock: it may
1718        // take a while, and nobody waits for it.
1719        let src = p
1720            .import_from
1721            .as_deref()
1722            .map(|s| self.source(s))
1723            .transpose()?;
1724        let mut guard = self.lock();
1725        let inner = &mut *guard;
1726        if inner.repos.get(&p.repo).is_some_and(|r| r.initialized())
1727            || inner.foreign.contains_key(&p.repo)
1728        {
1729            return fail("conflict", format!("repository {} exists", p.repo));
1730        }
1731        let now = self.now();
1732        if !inner.repos.contains_key(&p.repo) {
1733            let path = self.home.repo_path(&p.repo);
1734            let r = if path.exists() {
1735                RepoHandle::open(&p.repo, &path, &inner.keys.all())
1736            } else {
1737                RepoHandle::create(&p.repo, &path, inner.keys.for_new().clone())
1738            }
1739            .or_fail()?;
1740            inner.repos.insert(p.repo.clone(), r);
1741        }
1742        let r = inner.repos.get_mut(&p.repo).expect("inserted");
1743        r.chain.set_key(inner.keys.for_new().clone());
1744        let trunk = match &src {
1745            Some(src) => {
1746                let branch = p.branch.clone().unwrap_or_else(|| "main".into());
1747                self.check_branch(&branch)?;
1748                self.git(
1749                    r.store.path(),
1750                    &[
1751                        "fetch",
1752                        "--no-tags",
1753                        "--",
1754                        &src.arg(),
1755                        &format!("+{branch}:refs/lonca/import/initial"),
1756                    ],
1757                )?;
1758                r.store
1759                    .ref_get("refs/lonca/import/initial")
1760                    .or_fail()?
1761                    .ok_or(("invalid".into(), "import fetched nothing".into()))?
1762            }
1763            None => {
1764                let tree = r.store.empty_tree().or_fail()?;
1765                r.store
1766                    .write_commit(&CommitSpec {
1767                        tree,
1768                        parents: vec![],
1769                        author_name: p.human.clone(),
1770                        author_email: format!("{}@humans.lonca.invalid", p.human),
1771                        time_ms: now,
1772                        message: "lonca: initial commit\n".into(),
1773                        headers: vec![],
1774                    })
1775                    .or_fail()?
1776            }
1777        };
1778        let (config, rules) = profile.files(presence);
1779        let policy = r.write_policy(&config, &rules, None, now).or_fail()?;
1780        // What a workspace of this tree will meet: files above the change
1781        // cap can never be proposed back, and a tree above one page is
1782        // fetched in pages. Said now, not at the first session.
1783        let max_file = lonca_policy::config::Config::default()
1784            .changes
1785            .max_file_bytes;
1786        let budget = page_budget(self.settings.line_cap);
1787        let mut large_files = Vec::new();
1788        let mut tree_bytes = 0u64;
1789        for (path, e) in r.store.commit_files(&trunk).or_fail()? {
1790            if e.mode == lonca_core::FileMode::Submodule {
1791                continue;
1792            }
1793            let size = r.store.blob_size(&e.blob).or_fail()?;
1794            tree_bytes += size;
1795            if size > max_file {
1796                large_files.push(json!({"path": path, "size": size}));
1797            }
1798        }
1799        let mut warnings: Vec<String> = large_files
1800            .iter()
1801            .map(|f| {
1802                format!(
1803                    "{} is {} bytes, above the change cap of {max_file}: a workspace gets it, a proposal can not change it",
1804                    f["path"].as_str().unwrap_or_default(),
1805                    f["size"]
1806                )
1807            })
1808            .collect();
1809        if encoded_len(tree_bytes) > budget {
1810            warnings.push(format!(
1811                "the tree is {tree_bytes} bytes, more than one page ({budget} encoded): workspaces fetch it in pages"
1812            ));
1813        }
1814        let op = Op::RepoInit {
1815            repo: p.repo.clone(),
1816            daemon_key: r.chain.public_key(),
1817            human: p.human,
1818            human_key: p.human_key,
1819            policy,
1820            trunk,
1821        };
1822        inner.prepared.insert(p.repo, op.clone());
1823        Ok(json!({
1824            "op": op, "warnings": warnings, "large_files": large_files, "tree_bytes": tree_bytes,
1825            "profile": profile.name(),
1826        }))
1827    }
1828
1829    /// The profile of a new repository, and whether it asks for presence.
1830    /// A key that can not prove presence would be locked out of the
1831    /// repository by its own first entry, so that is said before it is
1832    /// made; leaving presence off is a word the human has to say.
1833    fn starting_profile(
1834        &self,
1835        p: &PrepareInit,
1836    ) -> std::result::Result<(Profile, bool), (String, String)> {
1837        let own = if self.settings.hardened {
1838            Profile::Hardened
1839        } else {
1840            Profile::Cooperative
1841        };
1842        let profile = p.profile.unwrap_or(own);
1843        if profile == Profile::Cooperative {
1844            return match p.without_presence {
1845                true => fail(
1846                    "invalid",
1847                    "--without-presence goes with the hardened profile",
1848                ),
1849                false => Ok((profile, false)),
1850            };
1851        }
1852        if !p.without_presence && !lonca_policy::human::proves_presence(&p.human_key.alg) {
1853            return fail(
1854                "invalid",
1855                "a hardened repository requires presence and this key can not prove it: \
1856                 use `lonca human keygen --secure-enclave` (Mac) or `--security-key` (a \
1857                 FIDO security key), or with neither pass --without-presence",
1858            );
1859        }
1860        Ok((profile, !p.without_presence))
1861    }
1862
1863    fn init(&self, auth: &Auth, p: HumanSubmit) -> Hr {
1864        let Op::RepoInit {
1865            repo,
1866            human,
1867            trunk,
1868            policy,
1869            ..
1870        } = &p.op
1871        else {
1872            return fail("invalid", "expected repo.init");
1873        };
1874        auth.same_human(&p.sig.human)?;
1875        let mut inner = self.lock();
1876        if inner.prepared.get(repo) != Some(&p.op) {
1877            return fail(
1878                "invalid",
1879                "this init was not prepared by this daemon; call repo.prepare_init",
1880            );
1881        }
1882        let now = self.now();
1883        let r = inner.repos.get_mut(repo).expect("prepared repos exist");
1884        let mut sub = Submission::new(
1885            Actor::Human {
1886                name: human.clone(),
1887            },
1888            p.op.clone(),
1889        );
1890        sub.human = Some(p.sig);
1891        let dry = r.dry_run(sub.clone(), now).or_fail()?;
1892        if !dry.entry.decision.is_allow() {
1893            return fail("denied", dry.entry.decision.reasons.join(", "));
1894        }
1895        let op = r.submit(sub, now).or_fail()?;
1896        r.store.ref_cas(TRUNK_REF, trunk, None).or_fail()?;
1897        r.store.ref_cas(POLICY_REF, policy, None).or_fail()?;
1898        inner.prepared.remove(repo);
1899        Ok(logged(&op))
1900    }
1901
1902    // ---------------------------------------------------------------- reads
1903
1904    fn status(&self, auth: &Auth) -> Hr {
1905        let name = auth.repo()?;
1906        let mut inner = self.lock();
1907        let now = self.now();
1908        let r = Self::repo(&mut inner, &name)?;
1909        let policy = r.policy().ok();
1910        let unparsed = policy.as_ref().map(|p| p.config.leases.unparsed_files);
1911        // Where a new file's key is reserved: an agent has no lonca.toml.
1912        let keyed: Option<serde_json::Map<String, Value>> = policy.as_ref().map(|p| {
1913            let dirs = p.config.paths.keyed.keys();
1914            dirs.map(|d| (d.clone(), json!(lonca_core::keyed::container(d))))
1915                .collect()
1916        });
1917        // Names only: a command can hold paths and environment details.
1918        let policy_checks: Option<Vec<&String>> = policy
1919            .as_ref()
1920            .map(|p| p.config.verification.checks.keys().collect());
1921        let below_hardened = match &policy {
1922            Some(p) if self.settings.hardened => p.below_hardened(),
1923            _ => vec![],
1924        };
1925        let s = &r.chain.state;
1926        let runs = |c: &ChangeRecord| policy.as_ref().map(|p| runs_for(&p.config.verification, c));
1927        let sessions: Vec<Value> = s
1928            .sessions
1929            .values()
1930            .map(|x| {
1931                json!({
1932                    "id": x.id, "agent": x.agent, "task": x.task, "parent": x.parent, "level": x.level,
1933                    "harness": x.harness,
1934                    "taint": x.taint, "active": s.session_active(&x.id, now), "expires_ms": x.expires_ms,
1935                    "last_seen_ms": x.heartbeat.map(|h| h.seen_ms),
1936                    "verification_ms": s.session_spent(&x.id),
1937                    "continued_from": x.continued_from, "continued_by": x.continued_by,
1938                    "unproposed": x.unproposed,
1939                    "bound_ms": x.bound_ms, "unhooked": x.unhooked,
1940                    "last_hook_ms": r.hooked.get(&x.id).and_then(|h| h.last_ms),
1941                })
1942            })
1943            .collect();
1944        let leases: Vec<Value> = s
1945            .active_leases(now)
1946            .map(|a| {
1947                let l = a.lease;
1948                let pinned: Vec<Value> = l
1949                    .pins
1950                    .iter()
1951                    .map(|(change, items)| json!({"change": change, "items": items}))
1952                    .collect();
1953                json!({"set": l.set, "session": a.holder, "items": a.items, "intent": l.intent, "expires_ms": l.expires_ms, "pinned": pinned})
1954            })
1955            .collect();
1956        let changes: Vec<Value> = s
1957            .changes
1958            .values()
1959            .map(|c| {
1960                json!({
1961                    "change": c.change, "actor": c.actor.label(), "status": c.status, "provenance": c.provenance,
1962                    "commit": c.commit, "base": c.base, "intent": c.claim.intent, "created_ms": c.created_ms,
1963                    "version": c.version, "paths": touched_paths(&c.facts),
1964                    "checks": c.checks, "runs": runs(c),
1965                })
1966            })
1967            .collect();
1968        // One call fills a review list: what the operation is, whose, on
1969        // which change, since when, touching what.
1970        let pending: Vec<Value> = s
1971            .pending
1972            .values()
1973            .map(|p| {
1974                let session = p.actor.session().and_then(|id| s.sessions.get(id));
1975                let (change, base, intent) = match &p.op {
1976                    Op::ChangePropose {
1977                        change, base, claim, ..
1978                    } => (Some(*change), Some(base.clone()), Some(claim.intent.clone())),
1979                    _ => (None, None, None),
1980                };
1981                let record = change.and_then(|c| s.changes.get(&c));
1982                json!({
1983                    "seq": p.seq, "created_ms": p.time_ms, "actor": p.actor.label(), "op": p.op.kind().as_str(),
1984                    "reasons": p.reasons, "approvals": p.approvals, "required": p.required,
1985                    "change": change, "base": base, "intent": intent,
1986                    "agent": session.map(|x| &x.agent), "task": session.map(|x| &x.task),
1987                    "paths": p.facts.as_ref().map(touched_paths).unwrap_or_default(),
1988                    "checks": p.checks, "runs": record.and_then(runs),
1989                })
1990            })
1991            .collect();
1992        // Public keys only: which key each human signs with, and the
1993        // recovery key if one is set.
1994        let humans: serde_json::Map<String, Value> = s
1995            .humans
1996            .iter()
1997            .map(|(n, k)| {
1998                (
1999                    n.clone(),
2000                    json!({"alg": k.alg, "public": k.public, "recovery": s.recovery.get(n).map(|r| &r.public)}),
2001                )
2002            })
2003            .collect();
2004        let interests: Vec<Value> = s
2005            .interests
2006            .iter()
2007            .map(|(seq, i)| json!({"seq": seq, "session": i.session, "items": i.items, "intent": i.intent, "blocks_until_ms": i.blocks_until_ms}))
2008            .collect();
2009        Ok(json!({
2010            "interests": interests, "reservations": s.reservations_at(now),
2011            "repo": s.repo, "trunk": s.trunk, "policy": s.policy, "policy_checks": policy_checks,
2012            "frozen": s.frozen, "queue": s.queue,
2013            "verification_day_ms": s.day_spent(now),
2014            "unparsed_files": unparsed, "keyed": keyed, "below_hardened": below_hardened,
2015            "humans": humans, "sessions": sessions, "leases": leases, "changes": changes, "pending": pending,
2016            "next_seq": s.next_seq, "head_hash": s.head_hash, "now_ms": now, "daemon_key": s.daemon_key,
2017            "proxy_unattributed": self.proxy_unattributed(),
2018        }))
2019    }
2020
2021    /// Ends what the verifications of the daemon before this one left
2022    /// running, and removes their directories; returns how many there were.
2023    /// Only once this daemon holds the socket, and so knows it is the only
2024    /// one on this home: see [`Verifier::clear_left_over`].
2025    pub fn clear_left_over_verifications(&self) -> usize {
2026        self.verifier.clear_left_over()
2027    }
2028
2029    /// What to say at the start about `sandbox`: the repositories whose
2030    /// verification can not pass without one.
2031    pub fn sandbox_warnings(&self, sandbox: Sandbox) -> Vec<String> {
2032        if sandbox != Sandbox::None {
2033            return vec![];
2034        }
2035        let mut inner = self.lock();
2036        let mut out = Vec::new();
2037        for (name, r) in &mut inner.repos {
2038            if r.policy()
2039                .is_ok_and(|p| p.config.verification.require_sandbox)
2040            {
2041                out.push(format!(
2042                    "{name} requires a sandbox and there is none: \
2043                     its changes will end in unsandboxed_verification"
2044                ));
2045            }
2046        }
2047        out
2048    }
2049
2050    /// The log from `from`. With `wait_ms` and nothing past `from`, the
2051    /// request is held (not the lock) until an entry arrives or the time
2052    /// runs out, so a client follows the log with one open request.
2053    fn log(&self, auth: &Auth, q: LogQuery) -> Hr {
2054        let name = auth.repo()?;
2055        let wait = q.wait_ms.unwrap_or(0).clamp(0, LOG_WAIT_MAX_MS);
2056        let deadline = std::time::Instant::now() + Duration::from_millis(wait as u64);
2057        let store = loop {
2058            let mut inner = self.lock();
2059            let r = Self::repo(&mut inner, &name)?;
2060            if r.chain.state.next_seq > q.from {
2061                break r.store.clone();
2062            }
2063            drop(inner);
2064            let left = deadline.saturating_duration_since(std::time::Instant::now());
2065            if left.is_zero() {
2066                return Ok(json!([]));
2067            }
2068            let (m, c) = &self.appended;
2069            let held = m.lock().unwrap_or_else(|e| e.into_inner());
2070            let _ = c
2071                .wait_timeout(held, left.min(Duration::from_millis(LOG_POLL_MS)))
2072                .unwrap_or_else(|e| e.into_inner());
2073        };
2074        let ops = OpLog::new(&store).read_from(q.from).or_fail()?;
2075        Ok(json!(ops[..log_page(q.limit).min(ops.len())]))
2076    }
2077
2078    fn tree(&self, auth: &Auth, q: TreeQuery) -> Hr {
2079        let name = auth.repo()?;
2080        let mut inner = self.lock();
2081        let r = Self::repo(&mut inner, &name)?;
2082        let store = r.store.clone();
2083        let redacted = r.chain.state.redacted.clone();
2084        drop(inner);
2085        let files = store.commit_files(&q.commit).or_fail()?;
2086        let mut out = Vec::new();
2087        for (path, e) in files {
2088            if let Some(only) = &q.paths
2089                && !only.contains(&path)
2090            {
2091                continue;
2092            }
2093            let size = if e.mode == lonca_core::FileMode::Submodule {
2094                0
2095            } else {
2096                store.blob_size(&e.blob).or_fail()?
2097            };
2098            let content_b64 = if q.with_content
2099                && e.mode != lonca_core::FileMode::Submodule
2100                && !redacted.contains(&e.blob)
2101            {
2102                Some(
2103                    base64::engine::general_purpose::STANDARD
2104                        .encode(store.read_blob(&e.blob).or_fail()?),
2105                )
2106            } else {
2107                None
2108            };
2109            out.push(TreeFile {
2110                path,
2111                mode: e.mode,
2112                blob: e.blob,
2113                size,
2114                content_b64,
2115            });
2116        }
2117        Ok(json!(out))
2118    }
2119
2120    /// One page of file contents of a commit. Every path must be in the
2121    /// commit's tree (a page never mixes commits), and the page must fit
2122    /// the line cap once encoded; a single file that does not fit is
2123    /// named, so the client can leave it out or ask for a bigger cap.
2124    fn blobs(&self, auth: &Auth, q: BlobsQuery) -> Hr {
2125        let name = auth.repo()?;
2126        let mut inner = self.lock();
2127        let r = Self::repo(&mut inner, &name)?;
2128        let store = r.store.clone();
2129        let redacted = r.chain.state.redacted.clone();
2130        drop(inner);
2131        if q.paths.is_empty() {
2132            return fail("invalid", "blobs: no paths");
2133        }
2134        let files = store.commit_files(&q.commit).or_fail()?;
2135        let budget = page_budget(self.settings.line_cap);
2136        let mut total = 0u64;
2137        let mut wanted = Vec::new();
2138        for path in &q.paths {
2139            let Some(e) = files.get(path) else {
2140                return fail("not_found", format!("blobs: {path} is not in {}", q.commit));
2141            };
2142            let size = if e.mode == lonca_core::FileMode::Submodule {
2143                0
2144            } else {
2145                store.blob_size(&e.blob).or_fail()?
2146            };
2147            let encoded = encoded_len(size);
2148            if encoded > budget {
2149                return fail(
2150                    "too_large",
2151                    format!(
2152                        "blobs: {path} is {size} bytes, more than one page holds ({budget} encoded)"
2153                    ),
2154                );
2155            }
2156            total += encoded;
2157            if total > budget {
2158                return fail(
2159                    "too_large",
2160                    format!(
2161                        "blobs: this page would exceed {budget} encoded bytes; ask for fewer files"
2162                    ),
2163                );
2164            }
2165            wanted.push((path.clone(), e.clone(), size));
2166        }
2167        let mut out = Vec::with_capacity(wanted.len());
2168        for (path, e, size) in wanted {
2169            let content_b64 =
2170                if e.mode != lonca_core::FileMode::Submodule && !redacted.contains(&e.blob) {
2171                    Some(
2172                        base64::engine::general_purpose::STANDARD
2173                            .encode(store.read_blob(&e.blob).or_fail()?),
2174                    )
2175                } else {
2176                    None
2177                };
2178            out.push(TreeFile {
2179                path,
2180                mode: e.mode,
2181                blob: e.blob,
2182                size,
2183                content_b64,
2184            });
2185        }
2186        Ok(json!(out))
2187    }
2188
2189    /// The change record, plus `evidence`: the checks of its last
2190    /// integration attempt with their results and log tails, read from
2191    /// the log and the store (never the reviewer's own git).
2192    fn change_get(&self, auth: &Auth, c: ChangeRef) -> Hr {
2193        let name = auth.repo()?;
2194        let mut inner = self.lock();
2195        let r = Self::repo(&mut inner, &name)?;
2196        let Some(rec) = r.chain.state.changes.get(&c.change).cloned() else {
2197            return fail("not_found", "unknown change");
2198        };
2199        let store = r.store.clone();
2200        let redacted = r.chain.state.redacted.clone();
2201        let runs = r
2202            .policy()
2203            .ok()
2204            .map(|p| runs_for(&p.config.verification, &rec));
2205        drop(inner);
2206        let mut out = json!(rec);
2207        out["runs"] = json!(runs);
2208        let ops = OpLog::new(&store).read_all().or_fail()?;
2209        let names_it = |o: &&SignedOp| match &o.entry.op {
2210            Op::ChangeIntegrate { change, .. } => *change == c.change,
2211            Op::ChangeIntegrateBatch { merged, .. } => merged.iter().any(|m| m.0 == c.change),
2212            _ => false,
2213        };
2214        let last = ops.iter().rev().find(names_it);
2215        let evidence: Vec<Value> = match last {
2216            Some(o) => o
2217                .entry
2218                .evidence
2219                .iter()
2220                .map(|a| {
2221                    let (log, log_truncated) = match &a.log {
2222                        Some(id) if !redacted.contains(id) => match store.read_blob(id) {
2223                            Ok(bytes) => {
2224                                let start = bytes.len().saturating_sub(EVIDENCE_LOG_CAP_BYTES);
2225                                (
2226                                    Value::String(
2227                                        String::from_utf8_lossy(&bytes[start..]).into_owned(),
2228                                    ),
2229                                    start > 0,
2230                                )
2231                            }
2232                            Err(_) => (Value::Null, false),
2233                        },
2234                        _ => (Value::Null, false),
2235                    };
2236                    let d = &a.details;
2237                    json!({
2238                        "check": a.check, "result": a.result,
2239                        "exit": d.get("exit").and_then(|x| x.parse::<i64>().ok()),
2240                        "duration_ms": d.get("duration_ms").and_then(|x| x.parse::<i64>().ok()),
2241                        "attempts": d.get("attempts").and_then(|x| x.parse::<u32>().ok()),
2242                        "sandbox": d.get("sandbox"), "error": d.get("error"),
2243                        "env": a.env, "environment": environment(d),
2244                        "details": d, "log": log, "log_truncated": log_truncated,
2245                    })
2246                })
2247                .collect(),
2248            None => vec![],
2249        };
2250        out["evidence"] = json!(evidence);
2251        out["integrate_seq"] = json!(last.map(|o| o.entry.seq));
2252        out["integrate_decision"] = json!(last.map(|o| &o.entry.decision));
2253        Ok(out)
2254    }
2255
2256    /// A Git bundle of trunk (`refs/heads/main`) for a git checkout, thin
2257    /// against the commits the checkout says it has (unknown ones are
2258    /// ignored). Refused when a redacted blob is among the objects it
2259    /// would carry: git needs whole objects, so a redacted history can be
2260    /// worked on from a workspace only. One line: the bundle must fit a
2261    /// page.
2262    fn trunk_bundle(&self, auth: &Auth, p: TrunkBundle) -> Hr {
2263        let name = auth.repo()?;
2264        let mut inner = self.lock();
2265        let r = Self::repo(&mut inner, &name)?;
2266        let store = r.store.clone();
2267        let trunk = r.chain.state.trunk.clone().expect("initialized");
2268        let redacted = r.chain.state.redacted.clone();
2269        let have: Vec<String> = p
2270            .have
2271            .iter()
2272            .filter(|h| store.read_commit(h).is_ok())
2273            .map(|h| format!("^{h}"))
2274            .collect();
2275        let mut args: Vec<&str> = vec!["rev-list", "--objects"];
2276        args.extend(have.iter().map(String::as_str));
2277        args.push("refs/heads/main");
2278        let objects = self.git(store.path(), &args)?;
2279        let mut hidden: Vec<&str> = Vec::new();
2280        for line in objects.lines() {
2281            let (id, path) = line.split_once(' ').unwrap_or((line, ""));
2282            if GitId::parse(id).is_ok_and(|g| redacted.contains(&g)) {
2283                hidden.push(path);
2284            }
2285        }
2286        if !hidden.is_empty() {
2287            return fail(
2288                "redacted",
2289                format!(
2290                    "trunk history carries redacted content ({}); git needs whole objects, work from a workspace",
2291                    hidden.join(", ")
2292                ),
2293            );
2294        }
2295        // Nothing past what the checkout has: no bundle, just the id.
2296        if objects.trim().is_empty() {
2297            return Ok(json!({
2298                "trunk": trunk, "bundle_b64": "", "bytes": 0, "thin": true, "up_to_date": true,
2299            }));
2300        }
2301        let dir = self.home.scratch();
2302        std::fs::create_dir_all(&dir).map_err(|e| from_err(e.into()))?;
2303        let tmp = dir.join(format!("{name}-{}.bundle", random_id("b")));
2304        let tmp_s = tmp.to_string_lossy().into_owned();
2305        let mut args: Vec<&str> = vec!["bundle", "create", "-q", &tmp_s];
2306        args.extend(have.iter().map(String::as_str));
2307        args.push("refs/heads/main");
2308        let made = self.git(store.path(), &args);
2309        let bytes = made.and_then(|_| std::fs::read(&tmp).map_err(|e| from_err(e.into())));
2310        let _ = std::fs::remove_file(&tmp);
2311        let bytes = bytes?;
2312        let budget = page_budget(self.settings.line_cap);
2313        if encoded_len(bytes.len() as u64) > budget {
2314            return fail(
2315                "too_large",
2316                format!(
2317                    "the bundle is {} bytes, more than one page holds; fetch from a closer point (have) or restore a backup",
2318                    bytes.len()
2319                ),
2320            );
2321        }
2322        Ok(json!({
2323            "trunk": trunk,
2324            "bundle_b64": base64::engine::general_purpose::STANDARD.encode(&bytes),
2325            "bytes": bytes.len(),
2326            "thin": !have.is_empty(),
2327            "up_to_date": false,
2328        }))
2329    }
2330
2331    /// How many integrations lie between a commit that once was trunk
2332    /// and trunk now: null when it never was trunk.
2333    fn trunk_distance(&self, auth: &Auth, p: TrunkDistance) -> Hr {
2334        let name = auth.repo()?;
2335        let mut inner = self.lock();
2336        let r = Self::repo(&mut inner, &name)?;
2337        let s = &r.chain.state;
2338        Ok(json!({"trunk": s.trunk, "base": p.base, "behind": s.behind(&p.base)}))
2339    }
2340
2341    /// The unified diff of a change between its base and its commit,
2342    /// computed by git in the store. Files whose blob on either side is
2343    /// redacted are left out and named. Capped at [`DIFF_CAP_BYTES`].
2344    /// The one change whose id starts with what a person typed (`pending`
2345    /// and `log` show 12 letters).
2346    fn change_resolve(&self, auth: &Auth, p: ChangeResolve) -> Hr {
2347        if !lonca_core::changeid::is_prefix(&p.prefix) {
2348            return fail(
2349                "invalid",
2350                format!(
2351                    "a change id prefix is {} to 32 letters k-z",
2352                    lonca_core::changeid::MIN_PREFIX
2353                ),
2354            );
2355        }
2356        let name = auth.repo()?;
2357        let mut inner = self.lock();
2358        let r = Self::repo(&mut inner, &name)?;
2359        resolve_prefix(r.chain.state.changes.keys(), &p.prefix)
2360    }
2361
2362    fn change_diff(&self, auth: &Auth, p: ChangeDiff) -> Hr {
2363        if let Some(f) = &p.format
2364            && f != "unified"
2365        {
2366            return fail("invalid", "format: only unified");
2367        }
2368        let name = auth.repo()?;
2369        let mut inner = self.lock();
2370        let r = Self::repo(&mut inner, &name)?;
2371        let Some(rec) = r.chain.state.changes.get(&p.change).cloned() else {
2372            return fail("not_found", "unknown change");
2373        };
2374        let store = r.store.clone();
2375        let redacted = r.chain.state.redacted.clone();
2376        drop(inner);
2377        let before = store.commit_files(&rec.base).or_fail()?;
2378        let after = store.commit_files(&rec.commit).or_fail()?;
2379        let mut paths: Vec<String> = Vec::new();
2380        let mut redacted_paths: Vec<String> = Vec::new();
2381        let mut status: BTreeMap<String, &str> = BTreeMap::new();
2382        let mut seen = std::collections::BTreeSet::new();
2383        for path in before.keys().chain(after.keys()) {
2384            if !seen.insert(path.clone()) {
2385                continue;
2386            }
2387            let (b, a) = (before.get(path), after.get(path));
2388            let st = match (b, a) {
2389                (Some(x), Some(y)) if x == y => continue,
2390                (Some(_), Some(_)) => "modified",
2391                (None, Some(_)) => "added",
2392                (Some(_), None) => "deleted",
2393                (None, None) => continue,
2394            };
2395            if [b, a]
2396                .into_iter()
2397                .flatten()
2398                .any(|e| redacted.contains(&e.blob))
2399            {
2400                redacted_paths.push(path.clone());
2401                continue;
2402            }
2403            status.insert(path.clone(), st);
2404            paths.push(path.clone());
2405        }
2406        // Symbol facts by file, so a reviewer sees what each file changes.
2407        let by_file = |items: &[String]| -> BTreeMap<String, Vec<String>> {
2408            let mut m: BTreeMap<String, Vec<String>> = BTreeMap::new();
2409            for s in items {
2410                if let Some((p, _)) = s.split_once('#') {
2411                    m.entry(p.to_string()).or_default().push(s.clone());
2412                }
2413            }
2414            m
2415        };
2416        let edits = by_file(&rec.facts.edits);
2417        let adds = by_file(&rec.facts.adds);
2418        let deletes = by_file(&rec.facts.deletes);
2419        let sigs = by_file(&rec.facts.signature_changes);
2420        type Counts = BTreeMap<String, (Option<u64>, Option<u64>)>;
2421        let (mut diff, mut counts): (String, Counts) = (String::new(), BTreeMap::new());
2422        if !paths.is_empty() {
2423            let (base, commit) = (rec.base.to_string(), rec.commit.to_string());
2424            let mut args: Vec<&str> = vec![
2425                "diff",
2426                "--no-color",
2427                "--no-ext-diff",
2428                "--numstat",
2429                &base,
2430                &commit,
2431                "--",
2432            ];
2433            args.extend(paths.iter().map(String::as_str));
2434            for line in self.git(store.path(), &args)?.lines() {
2435                let mut it = line.split('\t');
2436                let (a, d, path) = (it.next(), it.next(), it.next());
2437                if let (Some(a), Some(d), Some(path)) = (a, d, path) {
2438                    counts.insert(path.to_string(), (a.parse().ok(), d.parse().ok()));
2439                }
2440            }
2441            args[3] = "--patch";
2442            diff = self.git(store.path(), &args)?;
2443        }
2444        let (diff, truncated) = truncate_at_line(diff, DIFF_CAP_BYTES);
2445        let files: Vec<Value> = paths
2446            .iter()
2447            .map(|p| {
2448                let (added, removed) = counts.get(p).copied().unwrap_or((None, None));
2449                json!({
2450                    "path": p, "status": status[p], "added": added, "removed": removed,
2451                    "symbols": {
2452                        "edits": edits.get(p).cloned().unwrap_or_default(),
2453                        "adds": adds.get(p).cloned().unwrap_or_default(),
2454                        "deletes": deletes.get(p).cloned().unwrap_or_default(),
2455                        "signature_changes": sigs.get(p).cloned().unwrap_or_default(),
2456                    },
2457                })
2458            })
2459            .collect();
2460        Ok(json!({
2461            "change": rec.change, "base": rec.base, "commit": rec.commit, "format": "unified",
2462            "files": files, "redacted": redacted_paths, "diff": diff, "truncated": truncated,
2463        }))
2464    }
2465
2466    fn whoami(&self, auth: &Auth) -> Hr {
2467        let inner = self.lock();
2468        match &auth.caller {
2469            Caller::Agent { repo, actor } => {
2470                let r = inner.repos.get(repo).expect("checked");
2471                let s = r
2472                    .chain
2473                    .state
2474                    .sessions
2475                    .get(actor.session().unwrap_or_default());
2476                Ok(json!({"repo": repo, "actor": actor.label(), "session": s}))
2477            }
2478            Caller::Human { human, key } => Ok(json!({
2479                "actor": Actor::Human { name: human.clone() }.label(),
2480                "human": human,
2481                "key": key,
2482                "repos": Self::enrolled_repos(&inner, human, key),
2483            })),
2484            Caller::Anonymous => fail("unauthenticated", "missing session token"),
2485        }
2486    }
2487
2488    // ---------------------------------------------------------------- humans
2489
2490    fn prepare_change(&self, auth: &Auth, p: PrepareChange) -> Hr {
2491        let name = auth.repo()?;
2492        let mut inner = self.lock();
2493        let r = Self::repo(&mut inner, &name)?;
2494        let now = self.now();
2495        let change = p.change.unwrap_or_else(ChangeId::random);
2496        let max = r.policy().or_fail()?.config.changes.max_file_bytes;
2497        let commit = r
2498            .build_change_commit(
2499                &p.base,
2500                &p.files,
2501                change,
2502                &p.author,
2503                &format!("{}@humans.lonca.invalid", p.author),
2504                (HUMAN_HEADER, &p.author),
2505                &p.intent,
2506                now,
2507                max,
2508            )
2509            .or_fail()?;
2510        let keyed = &r.policy().or_fail()?.keyed;
2511        let a = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
2512        let f = &a.facts;
2513        let claim = Claim {
2514            intent: p.intent,
2515            edits: f.edits.clone(),
2516            adds: f.adds.clone(),
2517            deletes: f.deletes.clone(),
2518            deleted_files: f.deleted_files.clone(),
2519            signature_changes: f.signature_changes.clone(),
2520            tests: f.tests.clone(),
2521            revert_of: p.revert_of,
2522        };
2523        let op = Op::ChangePropose {
2524            change,
2525            commit,
2526            base: p.base,
2527            claim,
2528        };
2529        Ok(json!({"op": op, "facts": a.facts, "syntax_errors": a.syntax_errors}))
2530    }
2531
2532    fn prepare_policy(&self, auth: &Auth, p: PreparePolicy) -> Hr {
2533        let name = auth.repo()?;
2534        let mut inner = self.lock();
2535        let r = Self::repo(&mut inner, &name)?;
2536        let parent = r.chain.state.policy.clone();
2537        let commit = r
2538            .write_policy(&p.config, &p.policy, parent.as_ref(), self.now())
2539            .or_fail()?;
2540        Ok(json!({"op": Op::PolicyUpdate { policy: commit }}))
2541    }
2542
2543    fn prepare_import(&self, auth: &Auth, p: PrepareRemote) -> Hr {
2544        let name = auth.repo()?;
2545        let src = self.source(&p.remote)?;
2546        let branch = p.branch.unwrap_or_else(|| "main".into());
2547        self.check_branch(&branch)?;
2548        let store = {
2549            let mut inner = self.lock();
2550            Self::repo(&mut inner, &name)?.store.clone()
2551        };
2552        self.git(
2553            store.path(),
2554            &[
2555                "fetch",
2556                "--no-tags",
2557                "--",
2558                &src.arg(),
2559                &format!("+{branch}:refs/lonca/incoming"),
2560            ],
2561        )?;
2562        let mut inner = self.lock();
2563        let r = Self::repo(&mut inner, &name)?;
2564        let commit = r
2565            .store
2566            .ref_get("refs/lonca/incoming")
2567            .or_fail()?
2568            .ok_or(("invalid".into(), "nothing fetched".into()))?;
2569        let trunk = r.chain.state.trunk.clone().expect("initialized");
2570        if commit == trunk {
2571            return fail("up_to_date", "trunk already contains the remote branch");
2572        }
2573        if !is_ancestor(&r.store, &trunk, &commit).or_fail()? {
2574            return fail(
2575                "not_fast_forward",
2576                "the remote branch does not contain trunk; diverged history is not imported",
2577            );
2578        }
2579        Ok(json!({"op": Op::ImportExternal { remote: p.remote, commit, trunk_before: trunk }}))
2580    }
2581
2582    fn prepare_push(&self, auth: &Auth, p: PrepareRemote) -> Hr {
2583        let name = auth.repo()?;
2584        self.check_remote(&p.remote)?;
2585        let mut inner = self.lock();
2586        let r = Self::repo(&mut inner, &name)?;
2587        let s = &r.chain.state;
2588        Ok(
2589            json!({"op": Op::BridgePush { remote: p.remote, oplog_seq: s.next_seq - 1, oplog_hash: s.head_hash.clone() }}),
2590        )
2591    }
2592
2593    /// What a sandbox for `session` has to be built from: the roots from
2594    /// the log, never from a file in the workspace, which an agent could
2595    /// have edited to widen its sandbox.
2596    fn prepare_isolate(&self, auth: &Auth, p: PrepareIsolate) -> Hr {
2597        let name = auth.repo()?;
2598        let mut inner = self.lock();
2599        let r = Self::repo(&mut inner, &name)?;
2600        let Some(s) = r.chain.state.sessions.get(&p.session) else {
2601            return fail("not_found", "no such session");
2602        };
2603        let base_known = r.chain.state.trunk_history.contains(&p.base);
2604        let scratch = s.scope.scratch.clone();
2605        let harness = s.harness.clone();
2606        let policy = r.policy().or_fail()?;
2607        let profile = harness.as_deref().and_then(|h| policy.config.harness(h));
2608        Ok(json!({
2609            "scratch": scratch,
2610            "proxy": self.settings.proxy,
2611            "home": self.home.root,
2612            "base_known": base_known,
2613            "harness": harness,
2614            "setup": profile.map(|p| &p.setup),
2615        }))
2616    }
2617
2618    // ---------------------------------------------------------------- backup, restore
2619
2620    /// Writes a Git bundle of the repository, every ref included (trunk,
2621    /// the operation log, policy, proposals), into the backups directory.
2622    /// Under the lock, so the bundle is the store at one point of the log.
2623    /// Not logged: a backup is a read, like `log` or `tree`.
2624    fn backup(&self, auth: &Auth) -> Hr {
2625        let name = auth.repo()?;
2626        let dir = self.home.backups();
2627        std::fs::create_dir_all(&dir).map_err(|e| from_err(e.into()))?;
2628        set_mode(&dir, 0o750)?;
2629        let mut inner = self.lock();
2630        let r = Self::repo(&mut inner, &name)?;
2631        let seq = r.chain.state.next_seq - 1;
2632        let file = dir.join(format!("{name}-{seq:06}.bundle"));
2633        let tmp = dir.join(format!("{name}-{seq:06}.bundle.tmp"));
2634        self.git(
2635            r.store.path(),
2636            &["bundle", "create", "-q", &tmp.to_string_lossy(), "--all"],
2637        )?;
2638        set_mode(&tmp, 0o640)?;
2639        std::fs::rename(&tmp, &file).map_err(|e| from_err(e.into()))?;
2640        Ok(json!({
2641            "repo": name,
2642            "path": file,
2643            "seq": seq,
2644            "head_hash": r.chain.state.head_hash,
2645            "daemon_key": r.chain.state.daemon_key,
2646        }))
2647    }
2648
2649    /// Creates a repository from a backup: a bundle, a bare store or a
2650    /// remote Lonca pushed to. Trunk, the operation log, policy and
2651    /// proposals are fetched; the log must verify and replay, else nothing
2652    /// is kept. A store whose log is on another daemon key comes back
2653    /// foreign, to be continued with `daemon.recover`. An existing
2654    /// repository is never replaced: the log is append-only, and a restore
2655    /// over it would be the rewrite the auditor exists to catch.
2656    fn restore(&self, p: Restore) -> Hr {
2657        if !lonca_policy::decide::valid_token(&p.repo, 64) {
2658            return fail(
2659                "invalid",
2660                "repository names: letters, digits, '.', '_', '-'",
2661            );
2662        }
2663        let src = self.source(&p.from)?;
2664        let mut guard = self.lock();
2665        let inner = &mut *guard;
2666        let path = self.home.repo_path(&p.repo);
2667        if inner.repos.contains_key(&p.repo) || inner.foreign.contains_key(&p.repo) || path.exists()
2668        {
2669            return fail(
2670                "conflict",
2671                format!(
2672                    "repository {} exists; a restore never replaces a repository. \
2673                     Move its store away (as the daemon's user) or restore under another name",
2674                    p.repo
2675                ),
2676            );
2677        }
2678        let fetched = Store::init_bare(&path).or_fail().and_then(|store| {
2679            self.git(
2680                store.path(),
2681                &[
2682                    "fetch",
2683                    "-q",
2684                    "--no-tags",
2685                    "--",
2686                    &src.arg(),
2687                    "+refs/heads/*:refs/heads/*",
2688                    "+refs/lonca/*:refs/lonca/*",
2689                ],
2690            )?;
2691            let r = RepoHandle::open(&p.repo, &path, &inner.keys.all()).or_fail()?;
2692            if !r.initialized() {
2693                return fail("invalid", "the source has no operation log");
2694            }
2695            // Human signatures bind the name the log was created under.
2696            if r.chain.state.repo != p.repo {
2697                return fail(
2698                    "invalid",
2699                    format!(
2700                        "the source is a backup of repository {}; restore it under that name",
2701                        r.chain.state.repo
2702                    ),
2703                );
2704            }
2705            Ok(r)
2706        });
2707        let mut r = match fetched {
2708            Ok(r) => r,
2709            Err(e) => {
2710                let _ = std::fs::remove_dir_all(&path);
2711                return Err(e);
2712            }
2713        };
2714        let foreign = r.foreign();
2715        let out = json!({
2716            "repo": p.repo,
2717            "seq": r.chain.state.next_seq - 1,
2718            "head_hash": r.chain.state.head_hash,
2719            "daemon_key": r.chain.state.daemon_key,
2720            "foreign": foreign,
2721            "path": path,
2722        });
2723        if foreign {
2724            inner.foreign.insert(p.repo, r);
2725        } else {
2726            // A backup taken between appending an entry and applying it has
2727            // a trunk behind its log; the log says what trunk is.
2728            Self::reconcile(&mut r, self.now())?;
2729            inner.repos.insert(p.repo, r);
2730        }
2731        Ok(out)
2732    }
2733
2734    // ---------------------------------------------------------------- daemon keys
2735
2736    /// Starts (or resumes) a rotation: creates the next key and lists the
2737    /// repositories still on the current one. Each of them moves with a
2738    /// human-signed `daemon.rotate` naming the next key.
2739    fn prepare_rotate(&self) -> Hr {
2740        let mut guard = self.lock();
2741        let inner = &mut *guard;
2742        if inner.keys.next.is_none() {
2743            inner.keys.next = Some(self.home.load_or_create_next_key().or_fail()?);
2744        }
2745        let next = inner.keys.for_new().clone();
2746        let current = inner.keys.current.public_hex();
2747        let repos: Vec<&String> = inner
2748            .repos
2749            .iter()
2750            .filter(|(_, r)| r.initialized() && r.chain.public_key() == current)
2751            .map(|(n, _)| n)
2752            .collect();
2753        Ok(json!({"key": next.public_hex(), "repos": repos}))
2754    }
2755
2756    /// Re-anchors a foreign repository on this daemon's key. Nothing is
2757    /// logged unless the recovery is allowed: the entry is signed by the
2758    /// new key, which the log only accepts for an allowed recovery.
2759    fn daemon_recover(&self, name: &str, p: HumanSubmit) -> Hr {
2760        let mut guard = self.lock();
2761        let inner = &mut *guard;
2762        let now = self.now();
2763        let Some(r) = inner.foreign.get_mut(name) else {
2764            if inner.repos.get(name).is_some_and(|r| r.initialized()) {
2765                return fail(
2766                    "invalid",
2767                    format!(
2768                        "this daemon holds the key of repository {name}; \
2769                         use `lonca daemon-key rotate`"
2770                    ),
2771                );
2772            }
2773            return fail("not_found", format!("no repository {name}"));
2774        };
2775        let key = inner.keys.for_new().clone();
2776        if p.op.daemon_key_switch() != Some(key.public_hex().as_str()) {
2777            return fail(
2778                "invalid",
2779                format!(
2780                    "a recovery here must name this daemon's key {}",
2781                    key.public_hex()
2782                ),
2783            );
2784        }
2785        r.chain.set_key(key);
2786        let mut sub = Submission::new(
2787            Actor::Human {
2788                name: p.sig.human.clone(),
2789            },
2790            p.op,
2791        );
2792        sub.human = Some(p.sig);
2793        let dry = r.dry_run(sub.clone(), now).or_fail()?;
2794        if !dry.entry.decision.is_allow() {
2795            return fail("denied", dry.entry.decision.reasons.join(", "));
2796        }
2797        let op = r.submit(sub, now).or_fail()?;
2798        let mut r = inner.foreign.remove(name).expect("present");
2799        Self::reconcile(&mut r, now)?;
2800        inner.repos.insert(name.to_string(), r);
2801        Self::revoke_ended_tokens(inner, name)?;
2802        Ok(logged(&op))
2803    }
2804
2805    fn human_submit(&self, auth: &Auth, p: HumanSubmit) -> Hr {
2806        let name = auth.repo()?;
2807        auth.same_human(&p.sig.human)?;
2808        if matches!(p.op, Op::DaemonRecover { .. }) {
2809            return self.daemon_recover(&name, p);
2810        }
2811        let mut guard = self.lock();
2812        let inner = &mut *guard;
2813        let now = self.now();
2814        let next = inner.keys.next.clone();
2815        let r = Self::repo(inner, &name)?;
2816        Self::reconcile(r, now)?;
2817        let mut sub = Submission::new(
2818            Actor::Human {
2819                name: p.sig.human.clone(),
2820            },
2821            p.op.clone(),
2822        );
2823        sub.human = Some(p.sig);
2824        let workspace = p.workspace.filter(|w| Path::new(w).is_absolute());
2825        // Store, remote, and the proposal refs the log has: only those go.
2826        let mut push: Option<(PathBuf, String, Vec<String>)> = None;
2827        match &p.op {
2828            Op::RepoInit { .. } => return fail("invalid", "use repo.init"),
2829            Op::ChangePropose {
2830                change,
2831                commit,
2832                base,
2833                ..
2834            } => {
2835                r.check_proposal(commit, base, *change).or_fail()?;
2836                let keyed = &r.policy().or_fail()?.keyed;
2837                sub.facts = Some(
2838                    facts_keyed(&r.store, ENGINE_CURRENT, base, commit, keyed)
2839                        .or_fail()?
2840                        .facts,
2841                );
2842                sub.symbols_engine = Some(ENGINE_CURRENT);
2843            }
2844            Op::PolicyUpdate { policy } => {
2845                RepoHandle::load_policy(&r.store, policy).or_fail()?;
2846            }
2847            // Only to a key this daemon holds: anything else would end the
2848            // log, since nobody could sign the next entry.
2849            Op::DaemonRotate { key }
2850                if next.as_ref().map(KeyPair::public_hex).as_ref() != Some(key) =>
2851            {
2852                return fail("invalid", "rotate to the key from daemon.prepare_rotate");
2853            }
2854            Op::ImportExternal {
2855                commit,
2856                trunk_before,
2857                ..
2858            } if !is_ancestor(&r.store, trunk_before, commit).or_fail()? => {
2859                return fail("not_fast_forward", "import must contain trunk");
2860            }
2861            _ => {}
2862        }
2863        let old_policy = r.chain.state.policy.clone();
2864        let seq = r.chain.state.next_seq;
2865        if let Op::ChangePropose { change, commit, .. } = &p.op {
2866            r.store
2867                .ref_cas(&format!("refs/lonca/changes/{change}/{seq}"), commit, None)
2868                .or_fail()?;
2869        }
2870        let op = r.submit(sub, now).or_fail()?;
2871        let mut out = logged(&op);
2872        if !op.entry.decision.is_allow() {
2873            if op.entry.decision.outcome == lonca_core::ops::Outcome::Escalate {
2874                self.notify_queue();
2875            }
2876            return Ok(out);
2877        }
2878        match &op.entry.op {
2879            Op::SessionStart {
2880                session, continues, ..
2881            } => {
2882                let token = Self::issue_token(inner, &name, session)?;
2883                out["token"] = json!(token);
2884                if let Some(c) = continues {
2885                    Self::revoke_tokens(inner, &name, &c.session)?;
2886                }
2887            }
2888            Op::PolicyUpdate { policy } => {
2889                let r = Self::repo(inner, &name)?;
2890                r.store
2891                    .ref_cas(POLICY_REF, policy, old_policy.as_ref())
2892                    .or_fail()?;
2893            }
2894            Op::ImportExternal {
2895                commit,
2896                trunk_before,
2897                ..
2898            } => {
2899                let r = Self::repo(inner, &name)?;
2900                r.store
2901                    .ref_cas(TRUNK_REF, commit, Some(trunk_before))
2902                    .or_fail()?;
2903            }
2904            // Pushed below, once the lock is let go: a slow remote must
2905            // not hold up every other request.
2906            Op::BridgePush { remote, .. } => {
2907                let r = Self::repo(inner, &name)?;
2908                let changes = r.chain.state.proposed.iter();
2909                let changes = changes.map(|(seq, (c, _))| refs::change_ref_name(c, *seq));
2910                push = Some((
2911                    r.store.path().to_path_buf(),
2912                    remote.clone(),
2913                    changes.collect(),
2914                ));
2915            }
2916            // The old token was seen outside the sandbox, so it stops here:
2917            // from now on the level belongs to whoever holds the new one.
2918            Op::SessionIsolate { session, .. } => {
2919                let old = std::slice::from_ref(session);
2920                inner.tokens.revoke_sessions(&name, old).or_fail()?;
2921                out["token"] = json!(Self::issue_token(inner, &name, session)?);
2922                if let Some(ws) = workspace {
2923                    let key = (name.clone(), session.clone());
2924                    inner.workspaces.insert(key, PathBuf::from(ws));
2925                }
2926            }
2927            Op::CapRevoke { session, .. } | Op::SessionEnd { session, .. } => {
2928                Self::revoke_tokens(inner, &name, session)?;
2929            }
2930            // Key operations end every session of the human; an approval
2931            // may release an operation that ends sessions.
2932            Op::HumanRotate { .. } | Op::HumanRevoke { .. } | Op::HumanRecover { .. } => {
2933                Self::revoke_ended_tokens(inner, &name)?;
2934            }
2935            Op::Approve { .. } => {
2936                Self::revoke_ended_tokens(inner, &name)?;
2937                self.notify_queue();
2938            }
2939            Op::DaemonRotate { .. } => {
2940                let r = Self::repo(inner, &name)?;
2941                r.chain.set_key(next.expect("checked above"));
2942                self.promote_if_done(inner).or_fail()?;
2943            }
2944            Op::ChangePropose { .. } => self.notify_queue(),
2945            _ => {}
2946        }
2947        drop(guard);
2948        if let Some((dir, remote, changes)) = push {
2949            let mut pushed = String::new();
2950            let mine = ["refs/heads/main", "refs/lonca/oplog", "refs/lonca/policy"];
2951            let all: Vec<String> = mine.iter().map(|r| r.to_string()).chain(changes).collect();
2952            // In pieces, so that no command line grows with the log.
2953            for part in all.chunks(PUSH_REFS_PER_CALL) {
2954                let specs: Vec<String> = part.iter().map(|r| format!("{r}:{r}")).collect();
2955                let mut args = vec!["push", "--porcelain", "--", remote.as_str()];
2956                args.extend(specs.iter().map(String::as_str));
2957                pushed.push_str(&self.git(&dir, &args)?);
2958            }
2959            out["push"] = json!(pushed);
2960        }
2961        Ok(out)
2962    }
2963
2964    /// A token for `session`, which ends with the session.
2965    fn issue_token(
2966        inner: &mut Inner,
2967        repo: &str,
2968        session: &str,
2969    ) -> std::result::Result<String, (String, String)> {
2970        let r = Self::repo(inner, repo)?;
2971        let Some(s) = r.chain.state.sessions.get(session) else {
2972            return fail("internal", "a token for a session the log does not have");
2973        };
2974        let expires_ms = s.expires_ms;
2975        inner.tokens.issue(repo, session, expires_ms).or_fail()
2976    }
2977
2978    /// Drops the tokens that no longer work: past their end, or of a
2979    /// session the log says is over (ended, or under a parent that is).
2980    fn sweep_tokens(inner: &mut Inner, now: i64) -> std::result::Result<(), (String, String)> {
2981        let Inner { tokens, repos, .. } = inner;
2982        let live = |repo: &str, session: &str| {
2983            repos
2984                .get(repo)
2985                .is_some_and(|r| r.chain.state.session_active(session, now))
2986        };
2987        tokens.sweep_expired(now, live).or_fail()
2988    }
2989
2990    fn revoke_tokens(
2991        inner: &mut Inner,
2992        repo: &str,
2993        root: &str,
2994    ) -> std::result::Result<(), (String, String)> {
2995        let r = Self::repo(inner, repo)?;
2996        let ended: Vec<String> = r
2997            .chain
2998            .state
2999            .sessions
3000            .values()
3001            .filter(|s| s.ended.is_some() && r.chain.state.descends_from(&s.id, root))
3002            .map(|s| s.id.clone())
3003            .collect();
3004        inner.tokens.revoke_sessions(repo, &ended).or_fail()
3005    }
3006
3007    /// Drops the token of every ended session of a repository.
3008    fn revoke_ended_tokens(
3009        inner: &mut Inner,
3010        repo: &str,
3011    ) -> std::result::Result<(), (String, String)> {
3012        let r = Self::repo(inner, repo)?;
3013        let ended: Vec<String> = r
3014            .chain
3015            .state
3016            .sessions
3017            .values()
3018            .filter(|s| s.ended.is_some())
3019            .map(|s| s.id.clone())
3020            .collect();
3021        inner.tokens.revoke_sessions(repo, &ended).or_fail()
3022    }
3023
3024    // ---------------------------------------------------------------- agents
3025
3026    /// Runs one agent operation built by `make`.
3027    fn agent_op(
3028        &self,
3029        req: &Request,
3030        make: impl FnOnce(&Actor) -> std::result::Result<Op, (String, String)>,
3031    ) -> Hr {
3032        let mut guard = self.lock();
3033        let now = self.now();
3034        let (repo, actor) = Self::agent(&guard, req, now)?;
3035        let op = make(&actor)?;
3036        let r = Self::repo(&mut guard, &repo)?;
3037        Self::reconcile(r, now)?;
3038        let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3039        Ok(logged(&signed))
3040    }
3041
3042    fn lease_acquire(&self, req: &Request, p: LeaseAcquire) -> Hr {
3043        let set = p.set.clone().unwrap_or_else(|| random_id("lse"));
3044        let mut guard = self.lock();
3045        let now = self.now();
3046        let (repo, actor) = Self::agent(&guard, req, now)?;
3047        let session = actor.session().expect("agent").to_string();
3048        let r = Self::repo(&mut guard, &repo)?;
3049        Self::reconcile(r, now)?;
3050        let mut items = p.items;
3051        let keys = Self::next_keys(r, &mut items, now)?;
3052        let op = Op::LeaseAcquire {
3053            set: set.clone(),
3054            items: items.clone(),
3055            intent: p.intent,
3056            ttl_ms: p.ttl_ms,
3057        };
3058        let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3059        let mut out = logged(&signed);
3060        out["set"] = json!(set);
3061        if !keys.is_empty() {
3062            out["keys"] = json!(keys);
3063        }
3064        let details = Self::lease_details(r, &session, &items, now);
3065        out["advice"] = json!(details::advice(&details));
3066        out["details"] = json!(details);
3067        Ok(out)
3068    }
3069
3070    /// `next` among the names of an insert on a keyed directory's
3071    /// `dir#<dir>`: the number after the highest key a file of trunk has
3072    /// there or a lease holds, each `next` one more. Resolved here, so the
3073    /// log records the number and `decide` stays pure. Returns what each
3074    /// container got.
3075    fn next_keys(
3076        r: &mut RepoHandle,
3077        items: &mut [LeaseItem],
3078        now: i64,
3079    ) -> std::result::Result<BTreeMap<String, Vec<String>>, (String, String)> {
3080        let wants = |i: &LeaseItem| i.names.iter().any(|n| n == keyed::NEXT);
3081        if !items.iter().any(wants) {
3082            return Ok(BTreeMap::new());
3083        }
3084        let policy = r.policy().or_fail()?;
3085        let trunk = match &r.chain.state.trunk {
3086            Some(t) => r.store.commit_files(t).or_fail()?,
3087            None => FileMap::new(),
3088        };
3089        let state = &r.chain.state;
3090        let mut out = BTreeMap::new();
3091        for item in items.iter_mut().filter(|i| wants(i)) {
3092            let Some(dir) = item
3093                .symbol
3094                .strip_suffix(keyed::DIR)
3095                .and_then(|d| d.strip_suffix('#'))
3096            else {
3097                return fail(
3098                    "invalid",
3099                    format!("`{}` is a key: only for dir#<dir>", keyed::NEXT),
3100                );
3101            };
3102            let held: Vec<String> = state
3103                .active_leases(now)
3104                .flat_map(|l| l.items)
3105                .filter(|i| i.symbol == item.symbol)
3106                .flat_map(|i| i.names.iter().cloned())
3107                .collect();
3108            let mut taken: Vec<String> = held;
3109            taken.extend(item.names.iter().filter(|n| *n != keyed::NEXT).cloned());
3110            let mut names = Vec::new();
3111            for n in &item.names {
3112                if n != keyed::NEXT {
3113                    names.push(n.clone());
3114                    continue;
3115                }
3116                let tree = trunk.keys().map(String::as_str);
3117                let reserved = taken.iter().map(String::as_str);
3118                let Some(key) = policy.keyed.next_key(dir, tree, reserved) else {
3119                    return fail(
3120                        "invalid",
3121                        format!(
3122                            "`{}`: {dir} is not a keyed directory whose keys are numbers",
3123                            keyed::NEXT
3124                        ),
3125                    );
3126                };
3127                taken.push(key.clone());
3128                names.push(key);
3129            }
3130            let got: Vec<String> = names
3131                .iter()
3132                .filter(|n| !item.names.contains(n))
3133                .cloned()
3134                .collect();
3135            out.entry(item.symbol.clone())
3136                .or_insert_with(Vec::new)
3137                .extend(got);
3138            item.names = names;
3139        }
3140        Ok(out)
3141    }
3142
3143    fn lease_details(
3144        r: &RepoHandle,
3145        session: &str,
3146        items: &[LeaseItem],
3147        now: i64,
3148    ) -> Vec<details::Detail> {
3149        let state = &r.chain.state;
3150        let mut out = details::lease_held(state, session, items, now);
3151        out.extend(details::lease_reserved(state, session, items, now));
3152        out.extend(details::lease_wanted(state, session, items, now));
3153        out
3154    }
3155
3156    fn lease_interest(&self, req: &Request, p: LeaseInterest) -> Hr {
3157        let mut guard = self.lock();
3158        let now = self.now();
3159        let (repo, actor) = Self::agent(&guard, req, now)?;
3160        let session = actor.session().expect("agent").to_string();
3161        let r = Self::repo(&mut guard, &repo)?;
3162        Self::reconcile(r, now)?;
3163        let op = Op::LeaseInterest {
3164            items: p.items.clone(),
3165            intent: p.intent,
3166        };
3167        let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3168        let mut out = logged(&signed);
3169        out["details"] = json!(Self::lease_details(r, &session, &p.items, now));
3170        Ok(out)
3171    }
3172
3173    /// Derived from the log for the caller; nothing is written.
3174    fn events(&self, req: &Request, p: Events) -> Hr {
3175        let mut guard = self.lock();
3176        let now = self.now();
3177        let (repo, actor) = Self::agent(&guard, req, now)?;
3178        let session = actor.session().expect("agent").to_string();
3179        let r = Self::repo(&mut guard, &repo)?;
3180        let ops = OpLog::new(&r.store).read_all().or_fail()?;
3181        let state = &r.chain.state;
3182        let events = for_session(state, &session, &ops, p.from, now);
3183        let next = state.next_seq;
3184        r.told
3185            .entry(session)
3186            .or_default()
3187            .saw(p.from, &events, next);
3188        Ok(json!({"events": events, "next": next}))
3189    }
3190
3191    /// What `session` has not been told yet, and it has been told now.
3192    fn unread(
3193        r: &mut RepoHandle,
3194        session: &str,
3195        now: i64,
3196    ) -> std::result::Result<Vec<SessionEvent>, (String, String)> {
3197        let told = r.told.entry(session.to_string()).or_default();
3198        let from = told.from();
3199        let ops = OpLog::new(&r.store).read_from(from).or_fail()?;
3200        let events = for_session(&r.chain.state, session, &ops, from, now);
3201        Ok(told.unread(events, r.chain.state.next_seq))
3202    }
3203
3204    /// News for the caller since it was last told. `lonca mcp` adds this to
3205    /// the answer of each tool, so it reaches an agent on any harness.
3206    fn inbox(&self, req: &Request) -> Hr {
3207        let mut guard = self.lock();
3208        let now = self.now();
3209        let (repo, actor) = Self::agent(&guard, req, now)?;
3210        let session = actor.session().expect("agent").to_string();
3211        let r = Self::repo(&mut guard, &repo)?;
3212        let events = Self::unread(r, &session, now)?;
3213        let text = tell(&events, now, "");
3214        Ok(json!({"events": events, "text": text}))
3215    }
3216
3217    /// What happened to the sessions a human started, for a runner that
3218    /// wakes agents: never told (an agent's `inbox` keeps it), never a
3219    /// message's words. With `wait_ms` and nothing new, held (not the lock)
3220    /// like `log`; the clock is looked at every [`LOG_POLL_MS`], since a
3221    /// reservation or an expiry needs no entry.
3222    fn watch(&self, auth: &Auth, p: Watch) -> Hr {
3223        let Caller::Human { human, .. } = &auth.caller else {
3224            return fail(
3225                "forbidden",
3226                "a human session is required (lonca human login)",
3227            );
3228        };
3229        let name = auth.repo()?;
3230        let wait = p.wait_ms.unwrap_or(0).clamp(0, LOG_WAIT_MAX_MS);
3231        let deadline = std::time::Instant::now() + Duration::from_millis(wait as u64);
3232        let known: BTreeSet<&String> = p.reserved.iter().collect();
3233        let mut from = p.from;
3234        loop {
3235            let mut inner = self.lock();
3236            let now = self.now();
3237            let r = Self::repo(&mut inner, &name)?;
3238            let state = &r.chain.state;
3239            let next = state.next_seq;
3240            let start = *from.get_or_insert(next);
3241            let sessions = watched(state, human, p.sessions.as_deref(), now)?;
3242            let ended = ended_of(state, p.sessions.as_deref(), now);
3243            let mut events = Vec::new();
3244            if next > start && !sessions.is_empty() {
3245                let ops = ops_for_events(&r.store, start).or_fail()?;
3246                for s in &sessions {
3247                    let all = for_session(state, s, &ops, start, now);
3248                    let logged = all
3249                        .iter()
3250                        .filter(|e| !matches!(e, SessionEvent::Reserved { .. }));
3251                    events.extend(logged.map(|e| watched_event(s, e)));
3252                }
3253            }
3254            let mut reserved = Vec::new();
3255            for res in state.reservations_at(now) {
3256                if !sessions.contains(&res.session) {
3257                    continue;
3258                }
3259                let id = reservation_id(&res.session, &res.items);
3260                if !known.contains(&id) {
3261                    events.push(WatchedEvent {
3262                        session: res.session.clone(),
3263                        kind: "reserved".into(),
3264                        seq: None,
3265                        change: None,
3266                        from: None,
3267                        reservation: Some(id.clone()),
3268                        until_ms: Some(res.until_ms),
3269                    });
3270                }
3271                reserved.push(id);
3272            }
3273            let left = deadline.saturating_duration_since(std::time::Instant::now());
3274            if !events.is_empty() || !ended.is_empty() || left.is_zero() {
3275                let out = Watched {
3276                    events,
3277                    next,
3278                    reserved,
3279                    ended,
3280                };
3281                return Ok(json!(out));
3282            }
3283            // Read up to here: what it held was nobody's this caller watches.
3284            from = Some(next);
3285            drop(inner);
3286            let (m, c) = &self.appended;
3287            let held = m.lock().unwrap_or_else(|e| e.into_inner());
3288            let _ = c
3289                .wait_timeout(held, left.min(Duration::from_millis(LOG_POLL_MS)))
3290                .unwrap_or_else(|e| e.into_inner());
3291        }
3292    }
3293
3294    /// The same for a hook after a tool call: an agent that is editing calls
3295    /// no tool of Lonca's for minutes, and the sender is waiting.
3296    fn news_for_hook(
3297        &self,
3298        req: &Request,
3299        tool_prefix: &str,
3300    ) -> std::result::Result<Option<String>, (String, String)> {
3301        let mut guard = self.lock();
3302        let now = self.now();
3303        let (repo, actor) = Self::agent(&guard, req, now)?;
3304        let session = actor.session().expect("agent").to_string();
3305        let r = Self::repo(&mut guard, &repo)?;
3306        let events = Self::unread(r, &session, now)?;
3307        Ok(tell(&events, now, tool_prefix))
3308    }
3309
3310    fn propose(&self, req: &Request, p: ProposeChange) -> Hr {
3311        let mut guard = self.lock();
3312        let now = self.now();
3313        let (repo, actor) = Self::agent(&guard, req, now)?;
3314        let r = Self::repo(&mut guard, &repo)?;
3315        Self::reconcile(r, now)?;
3316        let session = actor.session().expect("agent").to_string();
3317        let change = p.change.unwrap_or_else(ChangeId::random);
3318        let max = r.policy().or_fail()?.config.changes.max_file_bytes;
3319        let commit = r
3320            .build_change_commit(
3321                &p.base,
3322                &p.files,
3323                change,
3324                &actor.label(),
3325                &format!("{session}@agents.lonca.invalid"),
3326                (SESSION_HEADER, &session),
3327                &p.claim.intent,
3328                now,
3329                max,
3330            )
3331            .or_fail()?;
3332        let keyed = &r.policy().or_fail()?.keyed;
3333        let analysis = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
3334        let seq = r.chain.state.next_seq;
3335        r.store
3336            .ref_cas(&format!("refs/lonca/changes/{change}/{seq}"), &commit, None)
3337            .or_fail()?;
3338        let mut sub = Submission::new(
3339            actor,
3340            Op::ChangePropose {
3341                change,
3342                commit: commit.clone(),
3343                base: p.base,
3344                claim: p.claim,
3345            },
3346        );
3347        sub.facts = Some(analysis.facts.clone());
3348        sub.symbols_engine = Some(ENGINE_CURRENT);
3349        let op = r.submit(sub, now).or_fail()?;
3350        let unobserved = op
3351            .entry
3352            .decision
3353            .reasons
3354            .iter()
3355            .any(|x| x.starts_with("unobserved_write:"));
3356        if unobserved
3357            && r.chain
3358                .state
3359                .sessions
3360                .get(&session)
3361                .is_some_and(|s| s.taint.is_none())
3362        {
3363            let t = Op::Taint {
3364                session: session.clone(),
3365                reason: "changed files that no observed tool wrote".into(),
3366            };
3367            r.submit(Submission::new(Actor::Daemon, t), now).or_fail()?;
3368        }
3369        if op.entry.decision.is_allow() {
3370            self.notify_queue();
3371        }
3372        let mut out = logged(&op);
3373        out["change"] = json!(change);
3374        out["commit"] = json!(commit);
3375        out["facts"] = json!(analysis.facts);
3376        out["syntax_errors"] = json!(analysis.syntax_errors);
3377        let policy = r.policy().or_fail()?;
3378        let unparsed = policy.config.leases.unparsed_files;
3379        let state = &r.chain.state;
3380        let mut missing = details::lease_missing(state, &session, &analysis.facts, unparsed, now);
3381        let keys = details::key_report(state, &session, &analysis.facts, &policy.keyed, now);
3382        let keys = keys.into_iter().filter(|d| d.code != details::HELD_BY_YOU);
3383        missing.extend(keys.map(|d| details::Detail {
3384            code: if d.code == details::FREE {
3385                details::LEASE_MISSING
3386            } else {
3387                d.code
3388            },
3389            ..d
3390        }));
3391        out["advice"] = json!(details::advice(&missing));
3392        out["details"] = json!(missing);
3393        Ok(out)
3394    }
3395
3396    fn facts_preview(&self, req: &Request, p: FactsPreview) -> Hr {
3397        let mut guard = self.lock();
3398        let now = self.now();
3399        let (repo, actor) = Self::agent(&guard, req, now)?;
3400        let r = Self::repo(&mut guard, &repo)?;
3401        let max = r.policy().or_fail()?.config.changes.max_file_bytes;
3402        let change = ChangeId::random();
3403        let commit = r
3404            .build_change_commit(
3405                &p.base,
3406                &p.files,
3407                change,
3408                &actor.label(),
3409                "preview@lonca.invalid",
3410                ("lonca-preview", "1"),
3411                "",
3412                now,
3413                max,
3414            )
3415            .or_fail()?;
3416        let keyed = &r.policy().or_fail()?.keyed;
3417        let a = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
3418        let session = actor.session().expect("agent");
3419        let unparsed = r.policy().or_fail()?.config.leases.unparsed_files;
3420        let mut leases = details::lease_report(&r.chain.state, session, &a.facts, unparsed, now);
3421        leases.extend(details::key_report(
3422            &r.chain.state,
3423            session,
3424            &a.facts,
3425            keyed,
3426            now,
3427        ));
3428        let no_lease = match unparsed {
3429            UnparsedFiles::Optimistic => a.facts.unparsed_files.clone(),
3430            UnparsedFiles::File => Vec::new(),
3431        };
3432        Ok(json!({
3433            "facts": a.facts, "syntax_errors": a.syntax_errors,
3434            "acquire": details::to_acquire(&leases), "advice": details::advice(&leases),
3435            "leases": leases, "no_lease": no_lease,
3436        }))
3437    }
3438
3439    fn attenuate(&self, req: &Request, p: Attenuate) -> Hr {
3440        let child = random_id("ses");
3441        let mut guard = self.lock();
3442        let now = self.now();
3443        let (repo, actor) = Self::agent(&guard, req, now)?;
3444        let parent = actor.session().expect("agent").to_string();
3445        let inner = &mut *guard;
3446        let r = Self::repo(inner, &repo)?;
3447        let op = Op::CapAttenuate {
3448            parent,
3449            child: child.clone(),
3450            agent: p.agent,
3451            scope: p.scope,
3452            expires_ms: p.expires_ms,
3453            leases: p.leases,
3454        };
3455        let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3456        let mut out = logged(&signed);
3457        if signed.entry.decision.is_allow() {
3458            out["session"] = json!(child);
3459            out["token"] = json!(Self::issue_token(inner, &repo, &child)?);
3460        }
3461        Ok(out)
3462    }
3463
3464    /// A new token for the caller's session, and the one it came with
3465    /// stops working at once. Not logged: the log never holds tokens, and a
3466    /// token says nothing the session's entries do not.
3467    fn rotate(&self, req: &Request) -> Hr {
3468        let mut guard = self.lock();
3469        let now = self.now();
3470        let (repo, actor) = Self::agent(&guard, req, now)?;
3471        let session = actor.session().unwrap_or_default();
3472        let r = Self::repo(&mut guard, &repo)?;
3473        let level = r.chain.state.sessions.get(session).map(|s| s.level);
3474        if level == Some(ObservationLevel::Isolated) {
3475            return fail(
3476                "forbidden",
3477                "an isolated session's token is its launcher's; `lonca run` replaces it",
3478            );
3479        }
3480        let old = req.token.as_deref().unwrap_or_default();
3481        match guard.tokens.rotate(old).or_fail()? {
3482            Some(token) => {
3483                drop(guard);
3484                use crate::security::{Event, kind};
3485                let e = Event::new(now, kind::TOKEN_ROTATED)
3486                    .repo(&repo)
3487                    .session(session);
3488                self.security.record(e);
3489                Ok(json!({"token": token}))
3490            }
3491            None => fail("unauthenticated", "unknown session token"),
3492        }
3493    }
3494
3495    fn agent_end(&self, req: &Request, end: SessionEnd) -> Hr {
3496        let mut guard = self.lock();
3497        let now = self.now();
3498        let (repo, actor) = Self::agent(&guard, req, now)?;
3499        let session = actor.session().expect("agent").to_string();
3500        let inner = &mut *guard;
3501        let r = Self::repo(inner, &repo)?;
3502        // Nobody is told anything here: it is on the record for `status`.
3503        let left = match end.files {
3504            Some(files) => Some(report_unproposed(r, &session, files, 0, now)?),
3505            None => None,
3506        };
3507        let op = Op::SessionEnd {
3508            session: session.clone(),
3509            reason: "ended by agent".into(),
3510        };
3511        let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3512        if signed.entry.decision.is_allow() {
3513            Self::revoke_tokens(inner, &repo, &session)?;
3514        } else {
3515            // An expired session's end is denied, and its token goes anyway.
3516            Self::sweep_tokens(inner, now)?;
3517        }
3518        let mut out = logged(&signed);
3519        if let Some(left) = left {
3520            out["unproposed"] = json!(left.paths);
3521            out["unproposed_total"] = json!(left.total);
3522        }
3523        Ok(out)
3524    }
3525
3526    // ---------------------------------------------------------------- hooks
3527
3528    /// The caller is alive: logged, and answered with what it needs to go
3529    /// on: the timeout, and its leases to renew against the daemon's clock.
3530    fn heartbeat(&self, req: &Request) -> Hr {
3531        let mut guard = self.lock();
3532        let now = self.now();
3533        let (repo, actor) = Self::agent(&guard, req, now)?;
3534        let session = actor.session().expect("agent").to_string();
3535        let r = Self::repo(&mut guard, &repo)?;
3536        // Refused here and not in the log: a client that does not stop
3537        // beating must not fill it.
3538        if !r.chain.state.session_active(&session, now) {
3539            return fail("session_inactive", "this session has ended or expired");
3540        }
3541        let limits = r.policy().or_fail()?.config.leases.clone();
3542        let beat = Observation {
3543            phase: Phase::Pre,
3544            kind: ObsKind::Heartbeat,
3545            external: false,
3546        };
3547        let op = Op::Observe {
3548            session: session.clone(),
3549            events: vec![beat],
3550        };
3551        let signed = r
3552            .submit(Submission::new(Actor::Daemon, op), now)
3553            .or_fail()?;
3554        let leases: Vec<Value> = r
3555            .chain
3556            .state
3557            .active_leases(now)
3558            .filter(|a| a.holder == session)
3559            .map(|a| json!({"set": a.lease.set, "expires_ms": a.lease.expires_ms}))
3560            .collect();
3561        let mut out = logged(&signed);
3562        out["now_ms"] = json!(now);
3563        out["timeout_ms"] = json!(limits.heartbeat_timeout_ms);
3564        out["max_ttl_ms"] = json!(limits.max_ttl_minutes * 60_000);
3565        out["leases"] = json!(leases);
3566        Ok(out)
3567    }
3568
3569    fn hook_event(&self, req: &Request, ev: HookEvent) -> Hr {
3570        let mut guard = self.lock();
3571        let now = self.now();
3572        let (repo, actor) = Self::agent(&guard, req, now)?;
3573        let session = actor.session().expect("agent").to_string();
3574        let r = Self::repo(&mut guard, &repo)?;
3575        if !std::path::Path::new(&ev.workspace).is_absolute() {
3576            return fail("invalid", "workspace must be an absolute path");
3577        }
3578        let policy = r.policy().or_fail()?;
3579        let scratch = r
3580            .chain
3581            .state
3582            .sessions
3583            .get(&session)
3584            .map(|s| s.scope.scratch.clone())
3585            .unwrap_or_default();
3586        let (events, verdict) = hooks::translate(&ev, &policy, &scratch);
3587        // Heard, and for a tool of `lonca mcp` kept for its call (#190). A
3588        // refused call never comes, so only an allowed one waits.
3589        let heard = Self::heard(r, &session, now);
3590        let harness = ev.harness.as_deref().unwrap_or("claude-code");
3591        if ev.hook_event_name == "PreToolUse"
3592            && verdict.decision != "deny"
3593            && let Some(profile) = policy.config.harness(harness)
3594            && let Some(tool) = ev
3595                .tool_name
3596                .as_deref()
3597                .and_then(|t| t.strip_prefix(&profile.tool_prefix()))
3598        {
3599            heard.before(tool, now);
3600        }
3601        if !events.is_empty() {
3602            r.submit(
3603                Submission::new(
3604                    Actor::Daemon,
3605                    Op::Observe {
3606                        session: session.clone(),
3607                        events,
3608                    },
3609                ),
3610                now,
3611            )
3612            .or_fail()?;
3613        }
3614        if verdict.decision == "deny" {
3615            drop(guard);
3616            use crate::security::{Event, kind};
3617            let tool = ev.tool_name.as_deref().unwrap_or("?");
3618            let e = Event::new(now, kind::HOOK_DENIED)
3619                .repo(&repo)
3620                .session(&session)
3621                .method("hook.event")
3622                .detail(&format!("{tool}: {}", verdict.reason));
3623            self.security.record(e);
3624        }
3625        Ok(json!(verdict))
3626    }
3627
3628    /// Reads a hook's raw input with the harness's profile. A session start
3629    /// is answered with what the hook needs for `hook.bind`; a tool call goes
3630    /// the way of `hook.event`, and a refusal comes in the harness's shape.
3631    fn hook_input(&self, req: &Request, h: HookInput) -> Hr {
3632        let profile = {
3633            let mut guard = self.lock();
3634            let (repo, _) = Self::agent(&guard, req, self.now())?;
3635            let policy = Self::repo(&mut guard, &repo)?.policy().or_fail()?;
3636            policy.config.harness(&h.harness).cloned()
3637        };
3638        let Some(profile) = profile else {
3639            return fail("invalid", format!("unknown harness {}", h.harness));
3640        };
3641        let field = |name: &str| h.input.get(name).and_then(Value::as_str);
3642        let name = field(&profile.fields.event).unwrap_or("");
3643        let phase = match profile.event_of(name) {
3644            Some(Event::SessionStart) => {
3645                return Ok(json!({
3646                    "phase": "session_start",
3647                    "session": field(&profile.fields.session).unwrap_or(""),
3648                    "context": profile.context,
3649                    "tool_prefix": profile.tool_prefix(),
3650                }));
3651            }
3652            // The hook comes back with `hook.stop`: only it can see the
3653            // workspace, and no event but this one needs that.
3654            Some(Event::Stop) => return Ok(json!({"phase": "stop"})),
3655            Some(Event::BeforeTool) => "PreToolUse",
3656            Some(Event::AfterTool) => "PostToolUse",
3657            None => {
3658                let message = format!("{name:?} is not a {} hook event", h.harness);
3659                return fail("invalid", message);
3660            }
3661        };
3662        let ev = HookEvent {
3663            hook_event_name: phase.to_string(),
3664            tool_name: field(&profile.fields.tool).map(str::to_string),
3665            tool_input: h
3666                .input
3667                .get(&profile.fields.input)
3668                .cloned()
3669                .unwrap_or(Value::Null),
3670            cwd: field(&profile.fields.cwd).map(str::to_string),
3671            workspace: h.workspace.clone(),
3672            harness: Some(h.harness.clone()),
3673        };
3674        let mut out = self.hook_event(req, ev)?;
3675        if out["decision"] == "deny" {
3676            let reason = out["reason"].as_str().unwrap_or("").to_string();
3677            out["output"] = render(&profile.deny, "reason", &reason);
3678        } else if let Some(answer) = (phase == "PostToolUse")
3679            .then_some(profile.tool_context.as_ref())
3680            .flatten()
3681            && let Some(news) = self.news_for_hook(req, &profile.tool_prefix())?
3682        {
3683            out["output"] = render(answer, "context", &news);
3684        }
3685        Ok(out)
3686    }
3687
3688    /// A session's hooks were heard from now. The first time, the sessions
3689    /// no longer active are forgotten, so the map lives as long as they do.
3690    fn heard<'a>(r: &'a mut RepoHandle, session: &str, now: i64) -> &'a mut crate::hooked::Hooked {
3691        if !r.hooked.contains_key(session) {
3692            let state = &r.chain.state;
3693            r.hooked.retain(|s, _| state.session_active(s, now));
3694        }
3695        let h = r.hooked.entry(session.to_string()).or_default();
3696        h.heard(now);
3697        h
3698    }
3699
3700    /// `lonca mcp` is about to run `tool`. On a harness whose before-tool
3701    /// event covers MCP tools that event came first, unless the hooks are
3702    /// not running: then the session is marked `unhooked` in the log, which
3703    /// taints it (#190). Nothing is refused here; what the session does next
3704    /// is decided as a tainted session's.
3705    fn mcp_call(&self, req: &Request, call: McpCall) -> Hr {
3706        use crate::hooked::{START_GRACE_MS, valid_tool};
3707        if !valid_tool(&call.tool) {
3708            return fail("invalid", "tool must be a tool name of lonca mcp");
3709        }
3710        let mut guard = self.lock();
3711        let now = self.now();
3712        let (repo, actor) = Self::agent(&guard, req, now)?;
3713        let session = actor.session().expect("agent").to_string();
3714        let r = Self::repo(&mut guard, &repo)?;
3715        let policy = r.policy().or_fail()?;
3716        let Some(s) = r.chain.state.sessions.get(&session) else {
3717            return fail("unauthenticated", "unknown session");
3718        };
3719        // A sandbox observes what hooks would; a session with no harness
3720        // named, or one whose event skips MCP tools, has nothing to prove.
3721        let checked = s.level != ObservationLevel::Isolated
3722            && s.harness
3723                .as_deref()
3724                .and_then(|h| policy.config.harness(h))
3725                .is_some_and(|p| p.events.before_tool_mcp);
3726        if !checked {
3727            return Ok(json!({"checked": false, "hooked": null}));
3728        }
3729        let marked = s.unhooked.is_some();
3730        // Only a session older than this daemon can have sent its event to
3731        // the one before.
3732        let fresh =
3733            s.created_ms < self.started_ms && now.saturating_sub(self.started_ms) < START_GRACE_MS;
3734        let took = r
3735            .hooked
3736            .get_mut(&session)
3737            .is_some_and(|h| h.take(&call.tool, now));
3738        let hooked = took || fresh;
3739        if !hooked && !marked {
3740            let op = Op::Observe {
3741                session: session.clone(),
3742                events: vec![Observation {
3743                    phase: Phase::Pre,
3744                    kind: ObsKind::Unhooked {
3745                        tool: call.tool.clone(),
3746                    },
3747                    external: false,
3748                }],
3749            };
3750            r.submit(Submission::new(Actor::Daemon, op), now)
3751                .or_fail()?;
3752        }
3753        Ok(json!({"checked": true, "hooked": hooked}))
3754    }
3755
3756    /// The agent is about to finish its turn: what it left is recorded, and
3757    /// while rounds remain it is told to go on, in the harness's shape.
3758    fn hook_stop(&self, req: &Request, h: HookStop) -> Hr {
3759        let mut guard = self.lock();
3760        let now = self.now();
3761        let (repo, actor) = Self::agent(&guard, req, now)?;
3762        let session = actor.session().expect("agent").to_string();
3763        let r = Self::repo(&mut guard, &repo)?;
3764        let policy = r.policy().or_fail()?;
3765        let limit = policy.config.sessions.stop_rounds;
3766        let Some(profile) = policy.config.harness(&h.harness).cloned() else {
3767            return fail("invalid", format!("unknown harness {}", h.harness));
3768        };
3769        let Some(answer) = &profile.stop_deny else {
3770            return fail("invalid", format!("{} has no stop event", h.harness));
3771        };
3772        Self::heard(r, &session, now);
3773        let left = report_unproposed(r, &session, h.files, limit, now)?;
3774        let mut out = json!({
3775            "unproposed": left.paths, "total": left.total, "round": left.round, "limit": limit,
3776        });
3777        if left.told {
3778            let reason = stop_message(&profile.tool_prefix(), &left.paths, left.total);
3779            out["output"] = render(answer, "reason", &reason);
3780        }
3781        Ok(out)
3782    }
3783
3784    /// The setup part of a harness's profile, from the signed policy. A
3785    /// harness with no profile is an answer, not an error: it still gets
3786    /// the rules and the MCP server.
3787    fn harness_setup(&self, req: &Request, h: HarnessSetup) -> Hr {
3788        let mut guard = self.lock();
3789        let (repo, _) = Self::agent(&guard, req, self.now())?;
3790        let policy = Self::repo(&mut guard, &repo)?.policy().or_fail()?;
3791        Ok(json!({"setup": policy.config.harness(&h.harness).map(|p| &p.setup)}))
3792    }
3793
3794    fn hook_bind(&self, req: &Request, b: HookBind) -> Hr {
3795        let mut guard = self.lock();
3796        let now = self.now();
3797        let (repo, actor) = Self::agent(&guard, req, now)?;
3798        let session = actor.session().expect("agent").to_string();
3799        let r = Self::repo(&mut guard, &repo)?;
3800        Self::heard(r, &session, now);
3801        // A hook of a harness that runs in a sandbox: it has nothing to add,
3802        // and what the hook can see of the machine is the sandbox's.
3803        let s = r.chain.state.sessions.get(&session);
3804        if s.is_some_and(|s| s.level == ObservationLevel::Isolated) {
3805            let scratch = s.map(|s| s.scope.scratch.clone()).unwrap_or_default();
3806            return Ok(json!({
3807                "level": ObservationLevel::Isolated,
3808                "notes": ["the session runs in a Lonca sandbox"],
3809                "scratch": scratch,
3810            }));
3811        }
3812        // The request says what the hook was told it is. Which table counts
3813        // is the human's word, from the session's start.
3814        let named = s.and_then(|s| s.harness.clone());
3815        let policy = r.policy().or_fail()?;
3816        let locked = named
3817            .as_deref()
3818            .filter(|n| *n == b.harness)
3819            .and_then(|n| policy.config.locked(n));
3820        let file = locked.and_then(|l| {
3821            let named = || l.file_here().map(PathBuf::from);
3822            self.settings.managed_settings.clone().or_else(named)
3823        });
3824        let ins = match (locked, file) {
3825            (Some(l), Some(file)) => hooks::inspect(
3826                l,
3827                &file,
3828                self.settings.require_root_settings,
3829                self.settings.proxy_port(),
3830            ),
3831            _ => hooks::Inspection {
3832                notes: vec![match named {
3833                    None => "the session was started with no harness named".to_string(),
3834                    Some(n) if n != b.harness => {
3835                        format!("the session was started for {n}, not {}", b.harness)
3836                    }
3837                    Some(n) => format!("no locked profile for harness {n}"),
3838                }],
3839                ..Default::default()
3840            },
3841        };
3842        let checks = BindChecks {
3843            harness: b.harness,
3844            harness_session: b.harness_session,
3845            managed_hooks: ins.managed_hooks,
3846            locked_sandbox: ins.locked_sandbox,
3847            sudo_without_password: b.sudo_without_password,
3848            config_digest: ins.config_digest,
3849            hook_binary_protected: ins.hook_binary_protected,
3850        };
3851        let level = level_for(&checks);
3852        if checks.sudo_without_password {
3853            use crate::security::{Event, kind};
3854            let e = Event::new(now, kind::SUDO_WITHOUT_PASSWORD)
3855                .repo(&repo)
3856                .session(&session)
3857                .detail("the hook reports that sudo runs without a password");
3858            self.security.record(e);
3859        }
3860        let signed = r
3861            .submit(
3862                Submission::new(
3863                    Actor::Daemon,
3864                    Op::SessionBind {
3865                        session: session.clone(),
3866                        checks,
3867                        level,
3868                    },
3869                ),
3870                now,
3871            )
3872            .or_fail()?;
3873        let mut out = logged(&signed);
3874        // Where the session stands, which a denial leaves as it was.
3875        let held = r.chain.state.sessions.get(&session).map(|s| s.level);
3876        out["level"] = json!(held.unwrap_or(ObservationLevel::Unobserved));
3877        out["notes"] = json!(ins.notes);
3878        // The roots the hook program names at SessionStart: from the log,
3879        // never from a file the agent could edit.
3880        out["scratch"] = json!(
3881            r.chain
3882                .state
3883                .sessions
3884                .get(&session)
3885                .map(|s| s.scope.scratch.clone())
3886                .unwrap_or_default()
3887        );
3888        if let Some(ws) = b
3889            .workspace
3890            .filter(|w| std::path::Path::new(w).is_absolute())
3891        {
3892            let session = signed.entry.op.clone();
3893            if let Op::SessionBind { session, .. } = session {
3894                guard.workspaces.insert((repo, session), PathBuf::from(ws));
3895            }
3896        }
3897        Ok(out)
3898    }
3899
3900    // ---------------------------------------------------------------- network
3901
3902    /// Records an outbound connection seen by the Lonca proxy and decides
3903    /// whether to let it through. `cwd` is the working directory of the
3904    /// connecting process when it could be determined; the connection is
3905    /// attributed to the sessions whose workspace contains it, otherwise to
3906    /// every session routed through the proxy (fail closed).
3907    /// The proxy's question: may this connection go through? Every session
3908    /// it is attributed to has its policy asked (port, host, connections
3909    /// and bytes), and the attempt is on each one's record before the
3910    /// answer, refused or not.
3911    pub fn proxy_connect(
3912        &self,
3913        cwd: Option<&std::path::Path>,
3914        host: &str,
3915        port: u16,
3916        peer: std::net::SocketAddr,
3917    ) -> ProxyVerdict {
3918        let now = self.now();
3919        let mut guard = self.lock();
3920        let inner = &mut *guard;
3921        let mut matched: Vec<(String, String)> = match cwd {
3922            Some(c) => inner
3923                .workspaces
3924                .iter()
3925                .filter(|(_, root)| c.starts_with(root))
3926                .map(|(k, _)| k.clone())
3927                .collect(),
3928            None => vec![],
3929        };
3930        // A session that is over answers for nothing: a command could stand
3931        // in its directory to keep its own session clean.
3932        matched.retain(|(repo, session)| {
3933            let r = inner.repos.get(repo);
3934            r.is_some_and(|r| r.chain.state.session_active(session, now))
3935        });
3936        let attributed = !matched.is_empty();
3937        if !attributed {
3938            for (name, r) in inner.repos.iter().filter(|(_, r)| r.initialized()) {
3939                let s = &r.chain.state;
3940                for x in s.sessions.values() {
3941                    if x.level >= ObservationLevel::HooksSandbox && s.session_active(&x.id, now) {
3942                        matched.push((name.clone(), x.id.clone()));
3943                    }
3944                }
3945            }
3946        }
3947        if matched.is_empty() {
3948            return ProxyVerdict {
3949                allow: false,
3950                allow_private: false,
3951                reason: "no Lonca session uses this proxy".into(),
3952            };
3953        }
3954        let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
3955        if !attributed {
3956            book.unattributed += 1;
3957        }
3958        let shown = match port {
3959            80 | 443 => host.to_string(),
3960            p => format!("{host}:{p}"),
3961        };
3962        let mut allow = true;
3963        let mut allow_private = true;
3964        let mut reason = None;
3965        let mut budgets = Vec::new();
3966        for key in &matched {
3967            let Some(r) = inner.repos.get_mut(&key.0) else {
3968                continue;
3969            };
3970            let Ok(policy) = r.policy() else {
3971                allow = false;
3972                continue;
3973            };
3974            let cfg = &policy.config.observation;
3975            let (passes, external) = host_verdict(cfg, host);
3976            let refused = if !cfg.allow_ports.contains(&port) {
3977                Some(format!("port {port} is not in observation.allow_ports"))
3978            } else if !passes {
3979                Some("network access is denied by repository policy".to_string())
3980            } else if book.open_of(key) >= cfg.max_connections as usize {
3981                Some(format!(
3982                    "this session has {} connections open through the proxy (observation.max_connections)",
3983                    cfg.max_connections
3984                ))
3985            } else if book.carried.get(key).copied().unwrap_or(0) >= cfg.max_bytes {
3986                Some("this session has moved observation.max_bytes through the proxy".to_string())
3987            } else {
3988                None
3989            };
3990            if let Some(why) = refused {
3991                allow = false;
3992                reason.get_or_insert(why);
3993            }
3994            allow_private &= cfg.allow_private_network;
3995            budgets.push((key.clone(), cfg.max_bytes));
3996            let ev = Observation {
3997                phase: Phase::Pre,
3998                kind: ObsKind::Fetch {
3999                    host: shown.clone(),
4000                },
4001                external,
4002            };
4003            let op = Op::Observe {
4004                session: key.1.clone(),
4005                events: vec![ev],
4006            };
4007            if r.submit(Submission::new(Actor::Daemon, op), now).is_err() {
4008                allow = false;
4009                reason.get_or_insert("the attempt could not be recorded".to_string());
4010            }
4011        }
4012        if allow {
4013            book.open.insert(peer, budgets);
4014        }
4015        drop(book);
4016        drop(guard);
4017        self.notify_appended();
4018        ProxyVerdict {
4019            allow,
4020            allow_private,
4021            reason: reason.unwrap_or_default(),
4022        }
4023    }
4024
4025    /// A connection the proxy let through has ended.
4026    pub fn proxy_finished(&self, peer: std::net::SocketAddr) {
4027        let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
4028        book.open.remove(&peer);
4029    }
4030
4031    /// Counts `bytes` against every session the connection of `peer` counts
4032    /// against; `false` once one of them is past its budget, which cuts it.
4033    pub fn proxy_carried(&self, peer: std::net::SocketAddr, bytes: u64) -> bool {
4034        let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
4035        let Some(budgets) = book.open.get(&peer).cloned() else {
4036            return false;
4037        };
4038        let mut within = true;
4039        for (key, max) in budgets {
4040            let used = book.carried.entry(key).or_insert(0);
4041            *used = used.saturating_add(bytes);
4042            within &= *used <= max;
4043        }
4044        within
4045    }
4046
4047    /// Connections the proxy could not name a process for, since the
4048    /// daemon started.
4049    pub fn proxy_unattributed(&self) -> u64 {
4050        self.proxy.lock().map(|b| b.unattributed).unwrap_or(0)
4051    }
4052
4053    // ---------------------------------------------------------------- integrity
4054
4055    /// Compares refs with the log. A ref the daemon itself left behind
4056    /// (crash between log append and ref update) is moved forward silently;
4057    /// anything else is recorded as `import.unmediated` and undone.
4058    fn reconcile(r: &mut RepoHandle, now: i64) -> std::result::Result<(), (String, String)> {
4059        let state = &r.chain.state;
4060        let (Some(trunk), Some(policy)) = (state.trunk.clone(), state.policy.clone()) else {
4061            return Ok(());
4062        };
4063        let on = r.policy().or_fail()?.config.integrity.on_unmediated.clone();
4064        for (refname, expected, history) in [(TRUNK_REF, trunk, true), (POLICY_REF, policy, false)]
4065        {
4066            let found = r.store.ref_get(refname).or_fail()?;
4067            if found.as_ref() == Some(&expected) {
4068                continue;
4069            }
4070            let recovering = history
4071                && found
4072                    .as_ref()
4073                    .is_some_and(|f| r.chain.state.trunk_history.contains(f));
4074            if !recovering {
4075                let op = Op::ImportUnmediated {
4076                    refname: refname.into(),
4077                    found: found.clone(),
4078                    expected: Some(expected.clone()),
4079                };
4080                r.submit(Submission::new(Actor::Daemon, op), now)
4081                    .or_fail()?;
4082                if on == "freeze" || refname == POLICY_REF {
4083                    let reason = format!("{refname} changed outside Lonca");
4084                    if r.chain.state.frozen.is_none() {
4085                        r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4086                            .or_fail()?;
4087                    }
4088                }
4089            }
4090            r.store.ref_force(refname, &expected).or_fail()?;
4091        }
4092        Self::reconcile_oplog(r, now)?;
4093        Self::reconcile_others(r, now, on == "freeze")
4094    }
4095
4096    /// The log's own ref must name the head this daemon appends to. Moved,
4097    /// it is put back first (an append checks it), then the move is logged
4098    /// and the repository frozen: a log rewritten under the daemon is the
4099    /// gravest write there is.
4100    fn reconcile_oplog(r: &mut RepoHandle, now: i64) -> std::result::Result<(), (String, String)> {
4101        let Some(head) = r.head.clone() else {
4102            return Ok(());
4103        };
4104        let found = r.store.ref_get(OPLOG_REF).or_fail()?;
4105        if found.as_ref() == Some(&head) {
4106            return Ok(());
4107        }
4108        r.store.ref_force(OPLOG_REF, &head).or_fail()?;
4109        let op = Op::ImportUnmediated {
4110            refname: OPLOG_REF.into(),
4111            found,
4112            expected: Some(head),
4113        };
4114        r.submit(Submission::new(Actor::Daemon, op), now)
4115            .or_fail()?;
4116        Self::freeze_for(r, OPLOG_REF, now)
4117    }
4118
4119    /// Every other ref: what the log has must name what it says, a
4120    /// proposal's ref must be there, and nothing else may be (see
4121    /// `lonca_policy::refs`). A write is logged as `import.unmediated` and
4122    /// undone; the policy's `integrity.on_unmediated` says whether the
4123    /// repository freezes as well. The ref of a proposal never logged (the
4124    /// daemon stopped between the two) goes without a word.
4125    fn reconcile_others(
4126        r: &mut RepoHandle,
4127        now: i64,
4128        freeze: bool,
4129    ) -> std::result::Result<(), (String, String)> {
4130        use lonca_policy::refs::{RefCheck, check, missing_changes};
4131        let refs = r.store.refs_with_prefix("refs/").or_fail()?;
4132        let present: BTreeSet<String> = refs.iter().map(|(n, _)| n.clone()).collect();
4133        let mut undone = Vec::new();
4134        for (name, found) in refs {
4135            if [TRUNK_REF, POLICY_REF, OPLOG_REF].contains(&name.as_str()) {
4136                continue;
4137            }
4138            match check(&r.chain.state, &name, &found) {
4139                RefCheck::Fine => {}
4140                RefCheck::NotLogged => r.store.ref_delete(&name, &found).or_fail()?,
4141                RefCheck::Differs(expected) => {
4142                    match &expected {
4143                        Some(e) => r.store.ref_force(&name, e).or_fail()?,
4144                        None => r.store.ref_delete(&name, &found).or_fail()?,
4145                    }
4146                    undone.push((name, Some(found), expected));
4147                }
4148                RefCheck::Unknown => {
4149                    r.store.ref_delete(&name, &found).or_fail()?;
4150                    undone.push((name, Some(found), None));
4151                }
4152            }
4153        }
4154        let missing: Vec<(String, GitId)> = missing_changes(&r.chain.state, &present)
4155            .map(|(n, c)| (n, c.clone()))
4156            .collect();
4157        for (name, commit) in missing {
4158            if r.store.has_object(&commit) {
4159                r.store.ref_force(&name, &commit).or_fail()?;
4160            }
4161            undone.push((name, None, Some(commit)));
4162        }
4163        for (refname, found, expected) in undone {
4164            let op = Op::ImportUnmediated {
4165                refname: refname.clone(),
4166                found,
4167                expected,
4168            };
4169            r.submit(Submission::new(Actor::Daemon, op), now)
4170                .or_fail()?;
4171            if freeze {
4172                Self::freeze_for(r, &refname, now)?;
4173            }
4174        }
4175        Ok(())
4176    }
4177
4178    fn freeze_for(
4179        r: &mut RepoHandle,
4180        refname: &str,
4181        now: i64,
4182    ) -> std::result::Result<(), (String, String)> {
4183        if r.chain.state.frozen.is_none() {
4184            let reason = format!("{refname} changed outside Lonca");
4185            r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4186                .or_fail()?;
4187        }
4188        Ok(())
4189    }
4190
4191    /// Checks refs of every repository (run periodically).
4192    pub fn reconcile_all(&self) {
4193        let now = self.now();
4194        let mut inner = self.lock();
4195        let before: Vec<u64> = inner
4196            .repos
4197            .values()
4198            .map(|r| r.chain.state.next_seq)
4199            .collect();
4200        for r in inner.repos.values_mut().filter(|r| r.initialized()) {
4201            if let Err((code, message)) = Self::reconcile(r, now) {
4202                use crate::security::{Event, kind};
4203                let e = Event::new(now, kind::RECONCILE_ERROR)
4204                    .repo(&r.name)
4205                    .code(&code)
4206                    .detail(&message);
4207                self.security.record(e);
4208            }
4209            // The auditor's only write: a flag file asking for a freeze.
4210            let flag = self.home.freeze_flag(&r.name);
4211            if let Ok(text) = std::fs::read_to_string(&flag) {
4212                let reason: String = text.chars().filter(|c| !c.is_control()).take(200).collect();
4213                let reason = if reason.is_empty() {
4214                    "auditor finding".to_string()
4215                } else {
4216                    reason
4217                };
4218                if !self
4219                    .flagged
4220                    .lock()
4221                    .unwrap_or_else(|e| e.into_inner())
4222                    .contains(&r.name)
4223                {
4224                    use crate::security::{Event, kind};
4225                    let e = Event::new(now, kind::INTEGRITY_ALARM)
4226                        .repo(&r.name)
4227                        .detail(&reason);
4228                    self.security.record(e);
4229                }
4230                self.flagged
4231                    .lock()
4232                    .unwrap_or_else(|e| e.into_inner())
4233                    .insert(r.name.clone());
4234                if r.chain.state.frozen.is_none()
4235                    && r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4236                        .is_err()
4237                {
4238                    continue;
4239                }
4240                let _ = std::fs::remove_file(&flag);
4241                self.flagged
4242                    .lock()
4243                    .unwrap_or_else(|e| e.into_inner())
4244                    .remove(&r.name);
4245            }
4246        }
4247        let after: Vec<u64> = inner
4248            .repos
4249            .values()
4250            .map(|r| r.chain.state.next_seq)
4251            .collect();
4252        drop(inner);
4253        if before != after {
4254            self.notify_appended();
4255        }
4256    }
4257
4258    // ---------------------------------------------------------------- queue
4259
4260    /// Integrates the head of one repository's queue. Merging and
4261    /// verification run without the lock; the result is only logged if the
4262    /// queue head and trunk are unchanged.
4263    pub fn process_one(&self, repo: &str) -> Result<Option<SignedOp>> {
4264        match self.process_batch(repo)? {
4265            Some(done) => Ok(Some(done)),
4266            None => self.process_head(repo),
4267        }
4268    }
4269
4270    /// Syntax over what changed, and the configured checks of the tree.
4271    fn verify(
4272        &self,
4273        repo: &str,
4274        store: &Store,
4275        vcfg: &Verification,
4276        (files, touched): (&FileMap, &[String]),
4277        tip: &GitId,
4278        reuse: bool,
4279    ) -> Result<Vec<Attestation>> {
4280        let tree = store.read_commit(tip)?.tree;
4281        let mut evidence = vec![self.verifier.syntax(store, files, touched, &tree)?];
4282        let key = (repo.to_string(), tree.clone(), digest_of(vcfg));
4283        let now = self.now();
4284        let known = self.verified_lock().get(&key, now).filter(|_| reuse);
4285        if let Some(known) = known {
4286            evidence.extend(known.into_iter().map(reused));
4287            return Ok(evidence);
4288        }
4289        self.verification_runs.fetch_add(1, Ordering::Relaxed);
4290        let run = self.verifier.run(store, files, &tree, vcfg)?;
4291        if reuse && !run.is_empty() && run.iter().all(|a| a.result == CheckResult::Pass) {
4292            self.verified_lock().insert(key, &run, now);
4293        }
4294        evidence.extend(run);
4295        Ok(evidence)
4296    }
4297
4298    /// The changes at the head of the queue as one, if the policy says so:
4299    /// merged in order off the lock, and a prefix whose tip passes is
4300    /// integrated by one entry. Halving finds it, and the head alone is
4301    /// left to [`Self::process_head`], which logs a failure as it always did.
4302    /// The tree of a prefix seen before is not verified again. What can not
4303    /// be merged or verified here ends the stack in front of it: it is not
4304    /// the changes ahead of it that wait for that.
4305    fn process_batch(&self, repo: &str) -> Result<Option<SignedOp>> {
4306        let mut running = self.running();
4307        let (store, jobs, reaches, trunk, vcfg, joins, (keyed, digest)) = {
4308            let mut inner = self.lock();
4309            let Some(r) = inner.repos.get_mut(repo).filter(|r| r.initialized()) else {
4310                return Ok(None);
4311            };
4312            if r.chain.state.frozen.is_some() {
4313                return Ok(None);
4314            }
4315            let policy = r.policy()?;
4316            let vcfg = policy.config.verification.clone();
4317            let size = policy.config.queue.batch_size as usize;
4318            let t = r.chain.entry_time(self.now());
4319            let state = &r.chain.state;
4320            let quota = &policy.config.quota;
4321            let free = |c: &&ChangeId| {
4322                quota_problems(state, &state.changes[*c].actor, t, quota).is_empty()
4323            };
4324            let job = |c: &ChangeId| {
4325                (
4326                    *c,
4327                    state.changes[c].commit.clone(),
4328                    state.changes[c].base.clone(),
4329                )
4330            };
4331            let reach = |c: &ChangeId| {
4332                let rec = &state.changes[c];
4333                (rec.facts.reaches_others(), rec.owed_checks(&vcfg))
4334            };
4335            let held: Vec<ChangeId> = state
4336                .queue
4337                .iter()
4338                .take(size)
4339                .take_while(free)
4340                .copied()
4341                .collect();
4342            // Each member is merged onto the ones in front of it.
4343            let joins: Vec<_> = (0..held.len())
4344                .map(|k| state.joins(held[k], &held[..k], &policy.config))
4345                .collect();
4346            let jobs: Vec<_> = held.iter().map(job).collect();
4347            let reaches: Vec<_> = state.queue.iter().take(size).map(reach).collect();
4348            let trunk = state.trunk.clone().expect("initialized");
4349            if jobs.len() >= 2 {
4350                running.hold(jobs.iter().map(|j| j.0));
4351            }
4352            let keyed = (policy.keyed.clone(), policy.digest.clone());
4353            (r.store.clone(), jobs, reaches, trunk, vcfg, joins, keyed)
4354        };
4355        // At most the policy's size, so never two under the default.
4356        if jobs.len() < 2 {
4357            return Ok(None);
4358        }
4359        // Merged in order, up to the first that conflicts.
4360        let mut stack: Vec<(ChangeId, GitId, FileMap, Vec<String>)> = Vec::new();
4361        let mut tip = trunk.clone();
4362        for ((change, commit, base), joins) in jobs.iter().zip(&joins) {
4363            let may_join = |s: &str| joins.allows(s);
4364            let i = integrate_joining(
4365                &store,
4366                ENGINE_CURRENT,
4367                base,
4368                &tip,
4369                commit,
4370                &may_join,
4371                &keyed,
4372            );
4373            let Some((merged, i)) = i.ok().and_then(|i| Some((i.merged.clone()?, i))) else {
4374                break;
4375            };
4376            tip = merged.clone();
4377            stack.push((*change, merged, i.files, i.touched));
4378        }
4379        let mut length = stack.len();
4380        while length >= 2 {
4381            let prefix = &stack[..length];
4382            let (_, tip, files, _) = &prefix[length - 1];
4383            let touched: BTreeSet<String> =
4384                prefix.iter().flat_map(|p| p.3.iter().cloned()).collect();
4385            let touched: Vec<String> = touched.into_iter().collect();
4386            let reach = reaches[..length].iter().any(|r| r.0);
4387            let named = reaches[..length].iter().flat_map(|r| r.1.iter().cloned());
4388            let owed = vcfg.for_change(reach, &named.collect());
4389            let evidence = self.verify(repo, &store, &owed, (files, &touched), tip, true);
4390            if let Some(evidence) = evidence
4391                .ok()
4392                .filter(|e| evidence_problems(&owed, e).is_empty())
4393            {
4394                let merged = prefix.iter().map(|p| (p.0, p.1.clone())).collect();
4395                let joins = &joins[..length];
4396                return self.log_batch(repo, trunk, merged, (&jobs, joins), &digest, evidence);
4397            }
4398            length /= 2;
4399        }
4400        Ok(None)
4401    }
4402
4403    fn log_batch(
4404        &self,
4405        repo: &str,
4406        trunk: GitId,
4407        merged: Vec<(ChangeId, GitId)>,
4408        (jobs, joins): (&[(ChangeId, GitId, GitId)], &[Joins]),
4409        digest: &str,
4410        evidence: Vec<Attestation>,
4411    ) -> Result<Option<SignedOp>> {
4412        let mut inner = self.lock();
4413        let r = inner.repos.get_mut(repo).expect("exists");
4414        // The merge read the keyed directories of the policy it started
4415        // under; the auditor reads those of the policy at the entry.
4416        if r.policy()?.digest != digest {
4417            return Ok(None);
4418        }
4419        let config = r.policy()?.config.clone();
4420        let state = &r.chain.state;
4421        let still = jobs.iter().take(merged.len()).zip(&state.queue);
4422        let same = still.filter(|(job, queued)| job.0 == **queued);
4423        let same = same.filter(|(job, _)| same_proposal(state, job.0, &job.1, &job.2));
4424        if same.count() != merged.len() || state.trunk.as_ref() != Some(&trunk) {
4425            return Ok(None);
4426        }
4427        // As for one change: what may be joined is read again, and the head
4428        // goes alone if it is not what the batch merged with.
4429        let members: Vec<ChangeId> = merged.iter().map(|m| m.0).collect();
4430        let now = (0..members.len()).map(|k| state.joins(members[k], &members[..k], &config));
4431        if !now.eq(joins.iter().cloned()) {
4432            return Ok(None);
4433        }
4434        let tip = merged.last().map(|(_, tip)| tip.clone()).expect("a batch");
4435        let op = Op::ChangeIntegrateBatch {
4436            trunk_before: trunk.clone(),
4437            merged,
4438        };
4439        let mut sub = Submission::new(Actor::Daemon, op);
4440        sub.evidence = evidence;
4441        sub.symbols_engine = Some(ENGINE_CURRENT);
4442        let signed = r.submit(sub, self.now())?;
4443        if signed.entry.decision.is_allow() {
4444            r.store.ref_cas(TRUNK_REF, &tip, Some(&trunk))?;
4445        }
4446        drop(inner);
4447        self.notify_appended();
4448        Ok(Some(signed))
4449    }
4450
4451    /// The change at the head of the queue, alone.
4452    fn process_head(&self, repo: &str) -> Result<Option<SignedOp>> {
4453        let mut running = self.running();
4454        let (store, change, commit, base, trunk, vcfg, reuse, past, joins, (keyed, digest)) = {
4455            let mut inner = self.lock();
4456            let Some(r) = inner.repos.get_mut(repo).filter(|r| r.initialized()) else {
4457                return Ok(None);
4458            };
4459            if r.chain.state.frozen.is_some() {
4460                return Ok(None);
4461            }
4462            let Some(change) = r.chain.state.queue.first().copied() else {
4463                return Ok(None);
4464            };
4465            let rec = &r.chain.state.changes[&change];
4466            let (commit, base) = (rec.commit.clone(), rec.base.clone());
4467            let reaches = rec.facts.reaches_others();
4468            let trunk = r.chain.state.trunk.clone().expect("initialized");
4469            let policy = r.policy()?;
4470            let named = r.chain.state.changes[&change].owed_checks(&policy.config.verification);
4471            let vcfg = policy.config.verification.for_change(reaches, &named);
4472            let reuse = policy.config.queue.batch_size > 1;
4473            let t = r.chain.entry_time(self.now());
4474            let state = &r.chain.state;
4475            let actor = &state.changes[&change].actor;
4476            let past = !quota_problems(state, actor, t, &policy.config.quota).is_empty();
4477            let joins = state.joins(change, &[], &policy.config);
4478            running.hold([change]);
4479            (
4480                r.store.clone(),
4481                change,
4482                commit,
4483                base,
4484                trunk,
4485                vcfg,
4486                reuse,
4487                past,
4488                joins,
4489                (policy.keyed.clone(), policy.digest.clone()),
4490            )
4491        };
4492        let may_join = |s: &str| joins.allows(s);
4493        let integ = integrate_joining(
4494            &store,
4495            ENGINE_CURRENT,
4496            &base,
4497            &trunk,
4498            &commit,
4499            &may_join,
4500            &keyed,
4501        )?;
4502        let mut evidence = Vec::new();
4503        if let Some(m) = integ.merged.as_ref().filter(|_| !past) {
4504            let changed = (&integ.files, integ.touched.as_slice());
4505            evidence = self.verify(repo, &store, &vcfg, changed, m, reuse)?;
4506        }
4507        let mut inner = self.lock();
4508        let r = inner.repos.get_mut(repo).expect("exists");
4509        if r.chain.state.queue.first() != Some(&change)
4510            || !same_proposal(&r.chain.state, change, &commit, &base)
4511            || r.chain.state.trunk.as_ref() != Some(&trunk)
4512        {
4513            return Ok(None);
4514        }
4515        // The auditor reads what may be joined from the policy in force at
4516        // the entry: one that changed under the run merges again.
4517        let config = &r.policy()?.config;
4518        if r.chain.state.joins(change, &[], config) != joins || r.policy()?.digest != digest {
4519            drop(inner);
4520            return self.process_head(repo);
4521        }
4522        // Not verified, so it can only stand as what `decide` refuses: past
4523        // midnight, or under a wider policy, it is verified after all.
4524        let now = self.now();
4525        let t = r.chain.entry_time(now);
4526        let quota = r.policy()?.config.quota.clone();
4527        let actor = &r.chain.state.changes[&change].actor;
4528        if past && quota_problems(&r.chain.state, actor, t, &quota).is_empty() {
4529            drop(inner);
4530            return self.process_head(repo);
4531        }
4532        let op = Op::ChangeIntegrate {
4533            change,
4534            trunk_before: trunk.clone(),
4535            merged: integ.merged.clone(),
4536            conflicts: integ.conflicts,
4537        };
4538        let mut sub = Submission::new(Actor::Daemon, op);
4539        sub.evidence = evidence;
4540        sub.symbols_engine = Some(ENGINE_CURRENT);
4541        let signed = r.submit(sub, now)?;
4542        if signed.entry.decision.is_allow()
4543            && let Some(m) = &integ.merged
4544        {
4545            r.store.ref_cas(TRUNK_REF, m, Some(&trunk))?;
4546        }
4547        drop(inner);
4548        self.notify_appended();
4549        Ok(Some(signed))
4550    }
4551
4552    /// Drains every queue. Returns the number of integration attempts.
4553    /// How often the configured checks of a tree ran.
4554    pub fn verification_runs(&self) -> u64 {
4555        self.verification_runs.load(Ordering::Relaxed)
4556    }
4557
4558    fn running_lock(&self) -> std::sync::MutexGuard<'_, BTreeSet<ChangeId>> {
4559        self.running.lock().unwrap_or_else(|e| e.into_inner())
4560    }
4561
4562    fn running(&self) -> Running<'_> {
4563        Running {
4564            set: &self.running,
4565            held: Vec::new(),
4566        }
4567    }
4568
4569    fn verified_lock(&self) -> std::sync::MutexGuard<'_, Verified> {
4570        self.verified.lock().unwrap_or_else(|e| e.into_inner())
4571    }
4572
4573    pub fn process_all(&self) -> usize {
4574        let names: Vec<String> = self.lock().repos.keys().cloned().collect();
4575        let mut n = 0;
4576        let mut work: VecDeque<String> = names.into();
4577        while let Some(name) = work.pop_front() {
4578            match self.process_one(&name) {
4579                Ok(Some(done)) => {
4580                    n += match &done.entry.op {
4581                        Op::ChangeIntegrateBatch { merged, .. } => merged.len(),
4582                        _ => 1,
4583                    };
4584                    work.push_back(name);
4585                }
4586                Ok(None) => {}
4587                Err(e) => eprintln!("lonca: queue {name}: {e}"),
4588            }
4589        }
4590        n
4591    }
4592
4593    /// Drops tokens that no longer work, so `tokens.json` holds only live
4594    /// ones. A write fails only with the disk; the next round tries again.
4595    fn sweep_all(&self) {
4596        let mut inner = self.lock();
4597        let now = self.now();
4598        if let Err((code, message)) = Self::sweep_tokens(&mut inner, now) {
4599            eprintln!("lonca: sweeping tokens: {code}: {message}");
4600        }
4601    }
4602
4603    /// Background worker: drains queues when woken, and checks refs.
4604    pub fn spawn_worker(self: &Arc<Self>) -> std::thread::JoinHandle<()> {
4605        let svc = self.clone();
4606        std::thread::spawn(move || {
4607            let mut seen = 0;
4608            loop {
4609                svc.process_all();
4610                svc.reconcile_all();
4611                svc.sweep_all();
4612                svc.take_alarms();
4613                let (m, c) = &svc.wake;
4614                let guard = m.lock().unwrap_or_else(|e| e.into_inner());
4615                if *guard == seen {
4616                    let (g, _) = c
4617                        .wait_timeout(guard, Duration::from_secs(2))
4618                        .unwrap_or_else(|e| e.into_inner());
4619                    seen = *g;
4620                } else {
4621                    seen = *guard;
4622                }
4623            }
4624        })
4625    }
4626}
4627
4628/// Cuts `text` to at most `cap` bytes, on a line boundary (the last
4629/// complete line that fits), never inside a UTF-8 character. The flag
4630/// says whether anything was cut.
4631fn truncate_at_line(mut text: String, cap: usize) -> (String, bool) {
4632    if text.len() <= cap {
4633        return (text, false);
4634    }
4635    let mut cut = cap;
4636    while !text.is_char_boundary(cut) {
4637        cut -= 1;
4638    }
4639    text.truncate(cut);
4640    match text.rfind('\n') {
4641        Some(nl) => text.truncate(nl + 1),
4642        None => text.clear(),
4643    }
4644    (text, true)
4645}
4646
4647/// The files a change touches, as its facts say: added, modified and
4648/// deleted, sorted, each once.
4649/// What the merge queue will run for `c` under `vc`, by the words its
4650/// evidence and denials use: `build`, `tests`, `extended`, then the named
4651/// checks it owes. Pending ones too: the human sees it before the yes.
4652fn runs_for(vc: &Verification, c: &ChangeRecord) -> Vec<String> {
4653    let owed = vc.for_change(c.facts.reaches_others(), &c.owed_checks(vc));
4654    let builtin = [
4655        ("build", &owed.build),
4656        ("tests", &owed.test),
4657        ("extended", &owed.on_signature_change),
4658    ];
4659    let builtin = builtin.into_iter().filter(|(_, c)| c.is_some());
4660    let builtin = builtin.map(|(name, _)| name.to_string());
4661    builtin.chain(owed.checks.into_keys()).collect()
4662}
4663
4664fn touched_paths(f: &lonca_core::ops::Facts) -> Vec<String> {
4665    let mut paths: Vec<String> = f
4666        .added_files
4667        .iter()
4668        .chain(&f.modified_files)
4669        .chain(&f.deleted_files)
4670        .cloned()
4671        .collect();
4672    paths.sort();
4673    paths.dedup();
4674    paths
4675}
4676
4677/// Remotes: absolute paths, https://, ssh:// or scp-like `user@host:path`.
4678/// `ext::` and other transports that run commands are refused, as is
4679/// anything that could be read as an option.
4680fn validate_remote(r: &str) -> std::result::Result<(), (String, String)> {
4681    let ok_scheme = r.starts_with("https://") || r.starts_with("ssh://") || r.starts_with('/');
4682    let scp_like = !r.contains("://") && r.contains('@') && r.contains(':') && !r.starts_with('/');
4683    if r.starts_with('-')
4684        || r.contains("::")
4685        || r.chars().any(char::is_control)
4686        || !(ok_scheme || scp_like)
4687    {
4688        return fail(
4689            "invalid",
4690            "remote must be an absolute path, https://, ssh:// or user@host:path",
4691        );
4692    }
4693    Ok(())
4694}
4695
4696fn validate_branch(b: &str) -> std::result::Result<(), (String, String)> {
4697    let ok = !b.is_empty()
4698        && !b.starts_with('-')
4699        && !b.contains("..")
4700        && b.chars()
4701            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '/' | '_' | '-' | '.'));
4702    if ok {
4703        Ok(())
4704    } else {
4705        fail("invalid", "invalid branch name")
4706    }
4707}
4708
4709#[cfg(unix)]
4710fn set_mode(p: &std::path::Path, mode: u32) -> std::result::Result<(), (String, String)> {
4711    use std::os::unix::fs::PermissionsExt;
4712    std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode))
4713        .map_err(|e| from_err(e.into()))
4714}
4715
4716#[cfg(not(unix))]
4717fn set_mode(_p: &std::path::Path, _mode: u32) -> std::result::Result<(), (String, String)> {
4718    Ok(())
4719}
4720
4721/// Is `ancestor` reachable from `commit` through parents?
4722pub fn is_ancestor(store: &Store, ancestor: &GitId, commit: &GitId) -> Result<bool> {
4723    let mut seen = std::collections::BTreeSet::new();
4724    let mut todo = vec![commit.clone()];
4725    while let Some(c) = todo.pop() {
4726        if &c == ancestor {
4727            return Ok(true);
4728        }
4729        if !seen.insert(c.clone()) || seen.len() > 1_000_000 {
4730            continue;
4731        }
4732        todo.extend(store.read_commit(&c)?.parents);
4733    }
4734    Ok(false)
4735}
4736
4737#[cfg(test)]
4738mod tests {
4739    /// Every kind of event reaches a watcher as kind, entry, change and
4740    /// sender; what was written stays out.
4741    #[test]
4742    fn a_watcher_gets_the_kind_and_never_the_words() {
4743        use super::{ChangeId, SessionEvent as E, watched_event};
4744        use lonca_core::ops::{LeaseItem, NegotiationMsg};
4745        let c = ChangeId::random();
4746        let words = "WORDS-OF-ANOTHER-AGENT";
4747        let cases = [
4748            (
4749                E::Integrated { seq: 1, change: c },
4750                "integrated",
4751                Some(c),
4752                None,
4753            ),
4754            (
4755                E::Conflicted {
4756                    seq: 1,
4757                    change: c,
4758                    conflicts: vec![words.into()],
4759                },
4760                "conflicted",
4761                Some(c),
4762                None,
4763            ),
4764            (
4765                E::Failed {
4766                    seq: 1,
4767                    change: c,
4768                    reasons: vec![words.into()],
4769                },
4770                "failed",
4771                Some(c),
4772                None,
4773            ),
4774            (
4775                E::Approved {
4776                    seq: 1,
4777                    target_seq: 0,
4778                },
4779                "approved",
4780                None,
4781                None,
4782            ),
4783            (
4784                E::Rejected {
4785                    seq: 1,
4786                    target_seq: 0,
4787                    reason: words.into(),
4788                },
4789                "rejected",
4790                None,
4791                None,
4792            ),
4793            (
4794                E::Message {
4795                    seq: 1,
4796                    thread: words.into(),
4797                    from: Some("s9".into()),
4798                    message: NegotiationMsg::Counter { note: words.into() },
4799                },
4800                "message",
4801                None,
4802                Some("s9"),
4803            ),
4804            (
4805                E::LeaseReceived {
4806                    seq: 1,
4807                    set: words.into(),
4808                    from: Some("s9".into()),
4809                },
4810                "lease_received",
4811                None,
4812                Some("s9"),
4813            ),
4814            (
4815                E::SymbolsChanged {
4816                    seq: 1,
4817                    change: c,
4818                    from: words.into(),
4819                    sets: vec![words.into()],
4820                    symbols: vec![words.into()],
4821                },
4822                "symbols_changed",
4823                Some(c),
4824                None,
4825            ),
4826        ];
4827        for (event, kind, change, from) in cases {
4828            let w = watched_event("s1", &event);
4829            assert_eq!(w.session, "s1");
4830            assert_eq!(w.kind, kind);
4831            assert_eq!(w.seq, Some(1), "{kind}");
4832            assert_eq!(w.change, change, "{kind}");
4833            assert_eq!(w.from.as_deref(), from, "{kind}");
4834            let said = serde_json::to_string(&w).unwrap();
4835            assert!(!said.contains(words), "{said}");
4836        }
4837        let r = E::Reserved {
4838            items: vec![LeaseItem {
4839                symbol: words.into(),
4840                kind: lonca_core::ops::LeaseKind::Edit,
4841                names: vec![],
4842            }],
4843            until_ms: 7,
4844        };
4845        let w = watched_event("s1", &r);
4846        assert_eq!(
4847            (w.kind.as_str(), w.seq, w.until_ms),
4848            ("reserved", None, Some(7))
4849        );
4850        assert!(!serde_json::to_string(&w).unwrap().contains(words));
4851    }
4852
4853    #[test]
4854    fn a_reservation_s_name_follows_its_session_and_items() {
4855        use super::reservation_id;
4856        use lonca_core::ops::{LeaseItem, LeaseKind};
4857        let item = |s: &str| LeaseItem {
4858            symbol: s.into(),
4859            kind: LeaseKind::Edit,
4860            names: vec![],
4861        };
4862        let a = reservation_id("s1", &[item("x")]);
4863        assert_eq!(a.len(), 16);
4864        assert_eq!(a, reservation_id("s1", &[item("x")]));
4865        assert_ne!(a, reservation_id("s2", &[item("x")]));
4866        assert_ne!(a, reservation_id("s1", &[item("y")]));
4867    }
4868
4869    #[test]
4870    fn a_prefix_names_one_change_or_says_why_not() {
4871        use super::{ChangeId, resolve_prefix};
4872        let id = |s: &str| ChangeId::from_reverse_hex(s).unwrap();
4873        let a = id("zzzzkkkkzzzzkkkkzzzzkkkkzzzzkkkk");
4874        let b = id("zzzzykkkzzzzkkkkzzzzkkkkzzzzkkkk");
4875        let ids = [a, b];
4876        assert_eq!(
4877            resolve_prefix(&ids, "zzzzk").unwrap(),
4878            serde_json::json!({"change": a.to_string()})
4879        );
4880        assert_eq!(
4881            resolve_prefix(&ids, &a.to_string()).unwrap()["change"],
4882            a.to_string()
4883        );
4884        let (code, msg) = resolve_prefix(&ids, "zzzz").unwrap_err();
4885        assert_eq!(code, "invalid");
4886        assert_eq!(
4887            msg,
4888            format!("2 changes start with zzzz: {a}, {b}; give more letters")
4889        );
4890        let (code, msg) = resolve_prefix(&ids, "yyyy").unwrap_err();
4891        assert_eq!(
4892            (code.as_str(), msg.as_str()),
4893            ("not_found", "no change starts with yyyy")
4894        );
4895        // Many: the first five are named, then "...".
4896        let many: Vec<ChangeId> = ["k", "l", "m", "n", "o", "p"]
4897            .iter()
4898            .map(|c| id(&format!("kkkk{}", c.repeat(28))))
4899            .collect();
4900        let (_, msg) = resolve_prefix(&many, "kkkk").unwrap_err();
4901        assert!(msg.starts_with("6 changes start with kkkk: "), "{msg}");
4902        assert!(msg.ends_with(", ...; give more letters"), "{msg}");
4903        assert!(
4904            msg.contains("kkkkoooo") && !msg.contains("kkkkpppp"),
4905            "{msg}"
4906        );
4907        let (_, msg) = resolve_prefix(&many[..5], "kkkk").unwrap_err();
4908        assert!(
4909            msg.ends_with("kkkkoooooooooooooooooooooooooooo; give more letters"),
4910            "{msg}"
4911        );
4912    }
4913
4914    use super::*;
4915
4916    fn verified_key(i: usize) -> VerifiedKey {
4917        let tree = GitId::of_blob(i.to_string().as_bytes());
4918        ("demo".to_string(), tree, "checks".to_string())
4919    }
4920
4921    /// A result is gone at the hour, not a tick after it.
4922    #[test]
4923    fn the_daemon_trusts_only_root_s_files_unless_told_otherwise() {
4924        let file = Some(PathBuf::from("/m"));
4925        let strict = Settings::for_daemon(false, file.clone(), Some("p".into()), true);
4926        assert!(strict.require_root_settings);
4927        assert_eq!(strict.managed_settings, file);
4928        assert_eq!(strict.proxy.as_deref(), Some("p"));
4929        assert!(strict.hardened);
4930        assert!(!Settings::for_daemon(true, None, None, false).require_root_settings);
4931        assert!(!Settings::for_daemon(true, None, None, false).hardened);
4932    }
4933
4934    #[test]
4935    fn the_proxy_s_port_is_read_from_its_address() {
4936        let port = |proxy: Option<&str>| {
4937            Settings::for_daemon(false, None, proxy.map(String::from), false).proxy_port()
4938        };
4939        assert_eq!(port(Some("127.0.0.1:8471")), Some(8471));
4940        assert_eq!(port(Some("[::1]:8472")), Some(8472));
4941        assert_eq!(port(Some("p")), None);
4942        assert_eq!(port(None), None);
4943    }
4944
4945    #[test]
4946    fn a_verified_tree_is_known_until_the_hour() {
4947        let mut verified = Verified::default();
4948        verified.insert(verified_key(0), &[], 1_000);
4949        let at = |ms| verified.get(&verified_key(0), 1_000 + ms).is_some();
4950        assert!(at(VERIFIED_FOR_MS - 1));
4951        assert!(!at(VERIFIED_FOR_MS));
4952    }
4953
4954    /// As many as it says it keeps, then the oldest goes first.
4955    #[test]
4956    fn the_oldest_verified_tree_goes_first() {
4957        let mut verified = Verified::default();
4958        for i in 0..VERIFIED_TREES {
4959            verified.insert(verified_key(i), &[], 0);
4960        }
4961        let known = |v: &Verified, i| v.get(&verified_key(i), 0).is_some();
4962        assert!((0..VERIFIED_TREES).all(|i| known(&verified, i)));
4963        verified.insert(verified_key(VERIFIED_TREES), &[], 0);
4964        assert!(!known(&verified, 0));
4965        assert!((1..=VERIFIED_TREES).all(|i| known(&verified, i)));
4966    }
4967
4968    #[test]
4969    fn queue_run_wakes_the_worker() {
4970        let dir = tempfile::tempdir().unwrap();
4971        let home = Home::new(dir.path().join("h"));
4972        let svc = Service::open(home, Settings::default(), system_clock()).unwrap();
4973        let before = *svc.wake.0.lock().unwrap();
4974        svc.notify_queue();
4975        assert_eq!(*svc.wake.0.lock().unwrap(), before + 1);
4976    }
4977
4978    #[test]
4979    fn a_log_page_is_what_was_asked_up_to_the_cap() {
4980        assert_eq!(log_page(Some(3)), 3);
4981        assert_eq!(log_page(None), LOG_PAGE as usize);
4982        assert_eq!(log_page(Some(u64::MAX)), LOG_PAGE as usize);
4983    }
4984
4985    /// The rows of the method table in docs/decisions.md.
4986    fn documented_methods() -> BTreeMap<String, String> {
4987        let doc = include_str!("../../../docs/decisions.md");
4988        let section = doc
4989            .split("\n## ")
4990            .find(|s| s.starts_with("Daemon authentication"))
4991            .expect("a Daemon authentication section");
4992        section
4993            .lines()
4994            .filter_map(|l| l.strip_prefix("| `"))
4995            .map(|l| {
4996                let (method, rest) = l.split_once("` | ").expect("| `method` | cred |");
4997                let cred = rest.split(" |").next().unwrap_or_default().trim();
4998                (method.to_string(), cred.to_string())
4999            })
5000            .collect()
5001    }
5002
5003    /// `docs/protocol.md` is the runner's contract: every method has a
5004    /// `### \`method\`` section, nothing else does, and the number it
5005    /// states is the one `ping` returns. The number is pinned here so a
5006    /// change to it is a conscious one, made with a note in the document.
5007    #[test]
5008    fn the_protocol_doc_lists_every_method_and_states_the_number() {
5009        let doc = include_str!("../../../docs/protocol.md");
5010        let sections: Vec<&str> = doc
5011            .lines()
5012            .filter_map(|l| l.strip_prefix("### `"))
5013            .map(|l| l.trim_end_matches('`'))
5014            .collect();
5015        for (m, _) in METHODS {
5016            assert!(
5017                sections.contains(m),
5018                "{m} has no section in docs/protocol.md"
5019            );
5020        }
5021        assert_eq!(sections.len(), METHODS.len(), "{sections:?}");
5022        assert_eq!(PROTOCOL, 1);
5023        assert!(
5024            doc.contains(&format!("**Protocol number: {PROTOCOL}.**")),
5025            "docs/protocol.md must state the current protocol number"
5026        );
5027        assert!(doc.contains(&format!("\n- **{PROTOCOL}** (")));
5028        assert!(doc.contains("capped at 30 000") && LOG_WAIT_MAX_MS == 30_000);
5029    }
5030
5031    #[test]
5032    fn the_decisions_doc_lists_every_method_with_its_credential() {
5033        let documented = documented_methods();
5034        for (m, c) in METHODS {
5035            assert_eq!(
5036                documented.get(*m).map(String::as_str),
5037                Some(c.name()),
5038                "{m} in docs/decisions.md"
5039            );
5040        }
5041        assert_eq!(documented.len(), METHODS.len(), "{documented:?}");
5042    }
5043
5044    /// Every method in the table has a handler, and nothing outside the
5045    /// table does. Without a token the handler is reached only by public
5046    /// and bootstrap methods; the rest answer `unauthenticated`.
5047    #[test]
5048    fn every_listed_method_is_dispatched_and_nothing_else() {
5049        let dir = tempfile::tempdir().unwrap();
5050        let svc = Service::open(
5051            Home::new(dir.path().join("h")),
5052            Settings {
5053                managed_settings: Some(dir.path().join("m.json")),
5054                require_root_settings: false,
5055                git: "git".into(),
5056                line_cap: MAX_LINE,
5057                hardened: false,
5058                proxy: None,
5059                ..Settings::default()
5060            },
5061            Arc::new(|| 1_760_000_000_000),
5062        )
5063        .unwrap();
5064        let call = |method: &str, token: Option<&str>| {
5065            let r = svc.handle(Request {
5066                id: 1,
5067                method: method.into(),
5068                repo: Some("demo".into()),
5069                token: token.map(String::from),
5070                params: json!({}),
5071            });
5072            r.error.map(|e| e.code).unwrap_or_default()
5073        };
5074        for (m, cred) in METHODS {
5075            let code = call(m, None);
5076            assert_ne!(code, "unknown_method", "{m}");
5077            match cred {
5078                Cred::Public | Cred::Login | Cred::Bootstrap => {
5079                    assert_ne!(code, "unauthenticated", "{m}: {code}")
5080                }
5081                _ => assert_eq!(code, "unauthenticated", "{m}: {code}"),
5082            }
5083            // A token nobody issued is refused before any handler runs.
5084            assert_eq!(call(m, Some("lonca_nobody")), "unauthenticated", "{m}");
5085        }
5086        assert_eq!(call("admin.everything", None), "unknown_method");
5087        assert_eq!(call("", None), "unknown_method");
5088    }
5089
5090    /// The session constants are what the document says; a wrong operator
5091    /// in one would change every session's length.
5092    #[test]
5093    fn a_diff_is_cut_on_a_line_inside_the_cap_never_inside_a_character() {
5094        assert_eq!(DIFF_CAP_BYTES, 2_097_152);
5095        assert_eq!(EVIDENCE_LOG_CAP_BYTES, 32_768);
5096        let short = "a\nb\n".to_string();
5097        assert_eq!(truncate_at_line(short.clone(), 4), (short.clone(), false));
5098        assert_eq!(truncate_at_line(short, 3), ("a\n".to_string(), true));
5099        // Lines of four bytes, "xé\n"; byte 202 falls inside the "é" of
5100        // line 51, so the cut steps back and the last whole line ends at
5101        // 200. A cut that jumped (say halved) would end far earlier.
5102        let text: String = "xé\n".repeat(100);
5103        let (cut, truncated) = truncate_at_line(text, 202);
5104        assert!(truncated);
5105        assert_eq!(cut.len(), 200, "{}", cut.len());
5106        assert!(cut.ends_with("xé\n") && cut.is_char_boundary(cut.len()));
5107        // No whole line fits: nothing is returned, never a partial character.
5108        assert_eq!(
5109            truncate_at_line("xé\n".to_string(), 2),
5110            (String::new(), true)
5111        );
5112    }
5113
5114    #[test]
5115    fn session_constants_are_what_the_document_says() {
5116        assert_eq!(HUMAN_SESSION_DEFAULT_MS, 3_600_000);
5117        assert_eq!(HUMAN_SESSION_MAX_MS, 86_400_000);
5118        assert_eq!(LOGIN_NONCE_WINDOW_MS, 660_000);
5119        assert_eq!(LOGIN_NONCE_WINDOW_MS, MAX_AGE_MS + MAX_SKEW_MS);
5120    }
5121
5122    #[test]
5123    fn every_error_keeps_its_message_under_a_stable_code() {
5124        let cases = [
5125            (Error::Invalid("a".into()), "invalid", "invalid: a"),
5126            (Error::Conflict("b".into()), "conflict", "conflict: b"),
5127            (Error::Integrity("c".into()), "integrity", "integrity: c"),
5128            (
5129                Error::Unsupported("op 3: log format 9".into()),
5130                "unsupported",
5131                "unsupported: op 3: log format 9",
5132            ),
5133            (Error::Git("d".into()), "internal", "git: d"),
5134            (Error::Io(std::io::Error::other("e")), "internal", "io: e"),
5135        ];
5136        for (err, code, msg) in cases {
5137            assert_eq!(from_err(err), (code.to_string(), msg.to_string()));
5138        }
5139        let json = serde_json::from_str::<Value>("{").unwrap_err();
5140        assert_eq!(from_err(Error::Json(json)).0, "internal");
5141    }
5142}