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