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