ljos_cli/
admit.rs

1//! Origin-bound admission.
2//!
3//! An agent lesson is filed by posting the atom. A writer that admits by
4//! origin answers 400 `held as proposal <id>` and keeps that proposal,
5//! including the atom. `POST /v1/proposals/accept` with `{"workspace","id"}`
6//! writes the atom and records the acceptance. The same text held again is
7//! that proposal, so a retry does not leave another one.
8//!
9//! `POST /v1/proposals` is the miner. This path does not call it.
10//!
11//! A 400 that names `origin` as unknown, or a 404, means this writer does
12//! not hold. The proposal stays in `$XDG_STATE_HOME/ljos/proposals.jsonl`
13//! and `ljos accept` posts the atom. A writer that does not answer keeps
14//! the proposal local for this call.
15//!
16//! `origin` on a live atom is `user-declared`, `agent-derived` or `peer`.
17//! A 400 that names `origin` as unknown is retried once without the field.
18
19use std::collections::BTreeMap;
20use std::path::PathBuf;
21use std::sync::Mutex;
22
23use anyhow::{bail, Context, Result};
24use packset_client::PacksetClient;
25use serde::{Deserialize, Serialize};
26use serde_json::Value;
27
28pub const ORIGIN_USER: &str = "user-declared";
29pub const ORIGIN_AGENT: &str = "agent-derived";
30pub const ORIGIN_PEER: &str = "peer";
31
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct Filed {
34    pub id: String,
35    pub remote: bool,
36}
37
38#[derive(Debug, Clone, Serialize, Deserialize)]
39struct Record {
40    id: String,
41    status: String,
42    remote: bool,
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    remote_id: Option<String>,
45    text: String,
46    kind: String,
47    atom: Value,
48}
49
50fn gate() -> std::sync::MutexGuard<'static, ()> {
51    static GATE: Mutex<()> = Mutex::new(());
52    GATE.lock().unwrap_or_else(|e| e.into_inner())
53}
54
55fn origin_cache() -> &'static Mutex<BTreeMap<String, bool>> {
56    static KEPT: std::sync::OnceLock<Mutex<BTreeMap<String, bool>>> = std::sync::OnceLock::new();
57    KEPT.get_or_init(|| Mutex::new(BTreeMap::new()))
58}
59
60fn proposal_cache() -> &'static Mutex<BTreeMap<String, bool>> {
61    static MODE: std::sync::OnceLock<Mutex<BTreeMap<String, bool>>> = std::sync::OnceLock::new();
62    MODE.get_or_init(|| Mutex::new(BTreeMap::new()))
63}
64
65fn proposals_path() -> PathBuf {
66    std::env::var_os("XDG_STATE_HOME")
67        .map(PathBuf::from)
68        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
69        .unwrap_or_else(|| PathBuf::from(".local/state"))
70        .join("ljos")
71        .join("proposals.jsonl")
72}
73
74fn http_timeout() -> std::time::Duration {
75    std::env::var("PACKSET_TIMEOUT_MS")
76        .ok()
77        .and_then(|v| v.trim().parse::<u64>().ok())
78        .filter(|ms| *ms > 0)
79        .map_or(
80            std::time::Duration::from_secs(30),
81            std::time::Duration::from_millis,
82        )
83}
84
85pub fn stamp_origin(atom: &mut Value, origin: &str) {
86    if let Some(map) = atom.as_object_mut() {
87        map.insert("origin".into(), Value::String(origin.to_string()));
88    }
89}
90
91fn strip_origin(atom: &mut Value) {
92    if let Some(map) = atom.as_object_mut() {
93        map.remove("origin");
94    }
95}
96
97fn origin_rejected(text: &str) -> bool {
98    let lower = text.to_lowercase();
99    lower.contains("origin")
100        && (lower.contains("unknown")
101            || lower.contains("unexpected")
102            || lower.contains("not a field")
103            || lower.contains("unrecognized"))
104}
105
106/// POST an atom. When this writer refuses `origin`, the field is dropped
107/// and the post is tried once more. The answer is remembered per base URL.
108pub fn post_kept(client: &PacksetClient, atom: &Value) -> Result<Value> {
109    let base = client.base().to_string();
110    let mut body = atom.clone();
111    let known = origin_cache()
112        .lock()
113        .unwrap_or_else(|e| e.into_inner())
114        .get(&base)
115        .copied();
116    if known == Some(false) {
117        strip_origin(&mut body);
118    }
119    match client.post_atom(&body) {
120        Ok(v) => {
121            if body.get("origin").is_some() {
122                origin_cache()
123                    .lock()
124                    .unwrap_or_else(|e| e.into_inner())
125                    .insert(base, true);
126            }
127            Ok(v)
128        }
129        Err(e) => {
130            let text = e.to_string();
131            if body.get("origin").is_some() && origin_rejected(&text) {
132                origin_cache()
133                    .lock()
134                    .unwrap_or_else(|err| err.into_inner())
135                    .insert(base, false);
136                strip_origin(&mut body);
137                return client.post_atom(&body).map_err(anyhow::Error::from);
138            }
139            Err(anyhow::Error::from(e))
140        }
141    }
142}
143
144fn read_latest() -> BTreeMap<String, Record> {
145    let text = std::fs::read_to_string(proposals_path()).unwrap_or_default();
146    let mut by_id = BTreeMap::new();
147    for line in text.lines() {
148        if let Ok(rec) = serde_json::from_str::<Record>(line) {
149            by_id.insert(rec.id.clone(), rec);
150        }
151    }
152    by_id
153}
154
155fn append(rec: &Record) -> Result<()> {
156    let path = proposals_path();
157    if let Some(dir) = path.parent() {
158        std::fs::create_dir_all(dir)?;
159    }
160    let mut line = serde_json::to_string(rec)?;
161    line.push('\n');
162    use std::io::Write;
163    let mut file = std::fs::OpenOptions::new()
164        .create(true)
165        .append(true)
166        .open(path)?;
167    file.write_all(line.as_bytes())?;
168    Ok(())
169}
170
171fn proposal_is_local(base: &str) -> bool {
172    proposal_cache()
173        .lock()
174        .unwrap_or_else(|e| e.into_inner())
175        .get(base)
176        .copied()
177        == Some(false)
178}
179
180fn remember_proposal_mode(base: &str, remote: bool) {
181    proposal_cache()
182        .lock()
183        .unwrap_or_else(|e| e.into_inner())
184        .insert(base.to_string(), remote);
185}
186
187/// The id in `held as proposal <id>`.
188fn held_proposal_id(text: &str) -> Option<String> {
189    let marker = "held as proposal ";
190    let rest = text.get(text.find(marker)? + marker.len()..)?;
191    let id: String = rest.chars().take_while(|c| c.is_ascii_hexdigit()).collect();
192    (id.len() == 32).then_some(id)
193}
194
195/// A writer that will not hold the atom, and did not store it.
196fn filing_is_local(text: &str) -> bool {
197    let lower = text.to_lowercase();
198    origin_rejected(text)
199        || lower.contains(": 404:")
200        || lower.contains("status code 404")
201        || (lower.contains("403") && lower.contains("not allowed"))
202}
203
204/// `Ok(None)` is a writer that does not hold. `Ok(Some(id))` is its proposal id.
205fn post_direct(client: &PacksetClient, atom: &Value) -> Result<Option<String>> {
206    let base = client.base();
207    if proposal_is_local(base) {
208        return Ok(None);
209    }
210    match client.post_atom(atom) {
211        Ok(_) => {
212            // This writer stored the atom. It does not hold proposals.
213            remember_proposal_mode(base, false);
214            Ok(None)
215        }
216        Err(e) => {
217            let text = e.to_string();
218            if let Some(id) = held_proposal_id(&text) {
219                remember_proposal_mode(base, true);
220                return Ok(Some(id));
221            }
222            if filing_is_local(&text) {
223                remember_proposal_mode(base, false);
224                return Ok(None);
225            }
226            // A writer that did not answer keeps this proposal local. The
227            // next filing tries the writer again.
228            if !text.contains("bad response") {
229                return Ok(None);
230            }
231            bail!("propose: POST /v1/atoms failed: {text}");
232        }
233    }
234}
235
236/// File `atom` as an agent proposal by posting it.
237///
238/// A held answer is the proposal. The id is stable for one kind and one
239/// text, so the same lesson is one open proposal. An open, accepted or
240/// satisfied row is returned as it stands.
241pub fn propose_atom(client: &PacksetClient, mut atom: Value) -> Result<Filed> {
242    let _g = gate();
243    let text = atom["text"].as_str().unwrap_or("").trim().to_string();
244    let kind = atom["kind"].as_str().unwrap_or("lesson").to_string();
245    if text.is_empty() {
246        bail!("propose: empty text is not a proposal");
247    }
248    stamp_origin(&mut atom, ORIGIN_AGENT);
249    let id = crate::work_id(&format!("{kind}\n{text}"));
250    if let Some(existing) = read_latest().get(&id) {
251        if matches!(existing.status.as_str(), "open" | "accepted" | "satisfied") {
252            return Ok(Filed {
253                id,
254                remote: existing.remote,
255            });
256        }
257    }
258    let remote_id = post_direct(client, &atom)?;
259    let rec = Record {
260        id: id.clone(),
261        status: "open".into(),
262        remote: remote_id.is_some(),
263        remote_id,
264        text,
265        kind,
266        atom,
267    };
268    let remote = rec.remote;
269    append(&rec)?;
270    Ok(Filed { id, remote })
271}
272
273/// The person's own text wrote the claim. An open proposal of those words
274/// is satisfied, so a later accept does not post a second atom.
275pub fn satisfy_text(text: &str) {
276    let _g = gate();
277    let text = text.trim();
278    if text.is_empty() {
279        return;
280    }
281    let open: Vec<Record> = read_latest()
282        .into_values()
283        .filter(|rec| rec.status == "open" && rec.text.trim() == text)
284        .collect();
285    for mut rec in open {
286        rec.status = "satisfied".into();
287        let _ = append(&rec);
288        close_remote(&rec);
289    }
290}
291
292fn workspace_of(client: &PacksetClient, rec: &Record) -> String {
293    let workspace = rec
294        .atom
295        .get("workspace")
296        .and_then(Value::as_str)
297        .unwrap_or("");
298    if workspace.is_empty() {
299        client.workspace()
300    } else {
301        workspace.to_string()
302    }
303}
304
305/// Close a held proposal whose text is already a live claim. The writer's
306/// duplicate rule keeps one atom when the text and kind match.
307fn close_remote(rec: &Record) {
308    if !rec.remote {
309        return;
310    }
311    let Some(remote_id) = rec.remote_id.as_deref() else {
312        return;
313    };
314    let Ok(client) = crate::pack() else {
315        return;
316    };
317    let workspace = workspace_of(&client, rec);
318    let _ = accept_on_writer(&client, &workspace, remote_id);
319}
320
321fn accepted_line(id: &str, kind: &str, posted: &Value) -> String {
322    let origin = posted
323        .get("origin")
324        .and_then(Value::as_str)
325        .filter(|origin| !origin.is_empty())
326        .unwrap_or(ORIGIN_AGENT);
327    format!(
328        "accepted {id} as {kind} origin {origin}{}\n",
329        crate::revision_note(posted)
330    )
331}
332
333fn accept_on_writer(client: &PacksetClient, workspace: &str, id: &str) -> Result<Value> {
334    let url = format!("{}/v1/proposals/accept", client.base());
335    let body = serde_json::json!({"workspace": workspace, "id": id});
336    match packset_client::authorize(ureq::post(&url))
337        .timeout(http_timeout())
338        .send_json(body)
339    {
340        Ok(resp) => resp
341            .into_json()
342            .context("accept: the pack's answer was not json"),
343        Err(ureq::Error::Status(code, resp)) => {
344            let text = resp.into_string().unwrap_or_default();
345            bail!("accept: POST /v1/proposals/accept failed: {code}: {text}");
346        }
347        Err(e) => bail!("accept: POST /v1/proposals/accept failed: {e}"),
348    }
349}
350
351/// The full id of an open proposal a short id names; see
352/// [`crate::resolve_short_id`].
353///
354/// # Errors
355///
356/// The short id starts more than one open proposal.
357pub fn resolve_id(id: &str) -> Result<String> {
358    let open: Vec<String> = read_latest()
359        .into_values()
360        .filter(|r| r.status == "open")
361        .map(|r| r.id)
362        .collect();
363    crate::resolve_short_id("accept", id, &open)
364}
365
366/// The conversation that filed an open proposal: the holder its atom's
367/// source names. `None` when the id is not on file or the atom names none.
368#[must_use]
369pub fn proposed_by(id: &str) -> Option<String> {
370    let rec = read_latest().get(id.trim()).cloned()?;
371    rec.atom["source"]["session"]
372        .as_str()
373        .map(str::trim)
374        .filter(|s| !s.is_empty())
375        .map(str::to_string)
376}
377
378/// Write one open proposal into the pack. A satisfied proposal was already
379/// written by `remember` or `prefer`. An accepted one is already there.
380pub fn accept(id: &str) -> Result<String> {
381    let _g = gate();
382    let id = id.trim();
383    if id.is_empty() {
384        bail!("accept: a proposal id is required");
385    }
386    let Some(mut rec) = read_latest().get(id).cloned() else {
387        bail!("accept: no open proposal {id}");
388    };
389    match rec.status.as_str() {
390        "satisfied" => {
391            close_remote(&rec);
392            return Ok(format!("{id} was already written by remember or prefer\n"));
393        }
394        "accepted" => return Ok(format!("{id} is already in the pack\n")),
395        "open" => {}
396        _ => bail!("accept: no open proposal {id}"),
397    }
398    let client = crate::pack()?;
399    if rec.remote {
400        let remote_id = rec.remote_id.clone().unwrap_or_else(|| rec.id.clone());
401        let workspace = workspace_of(&client, &rec);
402        let posted = accept_on_writer(&client, &workspace, &remote_id)?;
403        rec.status = "accepted".into();
404        append(&rec)?;
405        return Ok(accepted_line(&rec.id, &rec.kind, &posted));
406    }
407    match post_kept(&client, &rec.atom) {
408        Ok(posted) => {
409            rec.status = "accepted".into();
410            append(&rec)?;
411            Ok(accepted_line(&rec.id, &rec.kind, &posted))
412        }
413        Err(e) => {
414            let text = e.to_string();
415            let Some(remote_id) = held_proposal_id(&text) else {
416                return Err(e).with_context(|| format!("accept: POST /v1/atoms failed for {id}"));
417            };
418            // Filed while this writer was not holding, or the post is the
419            // filing. Accept that proposal. Do not leave it open.
420            let workspace = workspace_of(&client, &rec);
421            let posted = accept_on_writer(&client, &workspace, &remote_id)?;
422            rec.remote = true;
423            rec.remote_id = Some(remote_id);
424            rec.status = "accepted".into();
425            append(&rec)?;
426            Ok(accepted_line(&rec.id, &rec.kind, &posted))
427        }
428    }
429}
ljos_cli/admit.rs

ljos_cli/
admit.rs
