1use std::collections::BTreeMap;
26use std::path::PathBuf;
27use std::sync::Mutex;
28
29use crate::PacksetClient;
30use anyhow::{bail, Context, Result};
31use serde::{Deserialize, Serialize};
32use serde_json::Value;
33
34pub const ORIGIN_USER: &str = "user-declared";
35pub const ORIGIN_AGENT: &str = "agent-derived";
36pub const ORIGIN_TOOL: &str = "tool-echo";
37pub const ORIGIN_PEER: &str = "peer";
38
39pub const LEVEL_L0: &str = "derived";
41pub const LEVEL_EXPLICIT: &str = "explicit";
43
44#[must_use]
49pub fn admission_level(origin: &str) -> &'static str {
50 match origin {
51 ORIGIN_USER => LEVEL_EXPLICIT,
52 _ => LEVEL_L0,
53 }
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct Filed {
58 pub id: String,
59 pub remote: bool,
60}
61
62#[derive(Debug, Clone, Serialize, Deserialize)]
63struct Record {
64 id: String,
65 status: String,
66 remote: bool,
67 #[serde(default, skip_serializing_if = "Option::is_none")]
68 remote_id: Option<String>,
69 text: String,
70 kind: String,
71 #[serde(default, skip_serializing_if = "String::is_empty")]
73 origin: String,
74 atom: Value,
75}
76
77fn gate() -> std::sync::MutexGuard<'static, ()> {
78 static GATE: Mutex<()> = Mutex::new(());
79 GATE.lock().unwrap_or_else(|e| e.into_inner())
80}
81
82fn origin_cache() -> &'static Mutex<BTreeMap<String, bool>> {
83 static KEPT: std::sync::OnceLock<Mutex<BTreeMap<String, bool>>> = std::sync::OnceLock::new();
84 KEPT.get_or_init(|| Mutex::new(BTreeMap::new()))
85}
86
87fn proposal_cache() -> &'static Mutex<BTreeMap<String, bool>> {
88 static MODE: std::sync::OnceLock<Mutex<BTreeMap<String, bool>>> = std::sync::OnceLock::new();
89 MODE.get_or_init(|| Mutex::new(BTreeMap::new()))
90}
91
92fn proposals_path() -> PathBuf {
93 std::env::var_os("XDG_STATE_HOME")
94 .map(PathBuf::from)
95 .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
96 .unwrap_or_else(|| PathBuf::from(".local/state"))
97 .join("ljos")
98 .join("proposals.jsonl")
99}
100
101fn http_timeout() -> std::time::Duration {
102 std::env::var("PACKSET_TIMEOUT_MS")
103 .ok()
104 .and_then(|v| v.trim().parse::<u64>().ok())
105 .filter(|ms| *ms > 0)
106 .map_or(
107 std::time::Duration::from_secs(30),
108 std::time::Duration::from_millis,
109 )
110}
111
112pub fn stamp_origin(atom: &mut Value, origin: &str) {
113 if let Some(map) = atom.as_object_mut() {
114 map.insert("origin".into(), Value::String(origin.to_string()));
115 }
116}
117
118pub fn stamp_person(atom: &mut Value) {
122 stamp_origin(atom, ORIGIN_USER);
123 atom["source"]["basis"] = Value::String(crate::Basis::Person.as_str().into());
124}
125
126fn strip_origin(atom: &mut Value) {
127 if let Some(map) = atom.as_object_mut() {
128 map.remove("origin");
129 }
130}
131
132fn origin_rejected(text: &str) -> bool {
133 let lower = text.to_lowercase();
134 lower.contains("origin")
135 && (lower.contains("unknown")
136 || lower.contains("unexpected")
137 || lower.contains("not a field")
138 || lower.contains("unrecognized"))
139}
140
141pub fn post_kept(client: &PacksetClient, atom: &Value) -> Result<Value> {
144 let base = client.base().to_string();
145 let mut body = atom.clone();
146 let known = origin_cache()
147 .lock()
148 .unwrap_or_else(|e| e.into_inner())
149 .get(&base)
150 .copied();
151 if known == Some(false) {
152 strip_origin(&mut body);
153 }
154 match client.post_atom(&body) {
155 Ok(v) => {
156 if body.get("origin").is_some() {
157 origin_cache()
158 .lock()
159 .unwrap_or_else(|e| e.into_inner())
160 .insert(base, true);
161 }
162 Ok(v)
163 }
164 Err(e) => {
165 let text = e.to_string();
166 if body.get("origin").is_some() && origin_rejected(&text) {
167 origin_cache()
168 .lock()
169 .unwrap_or_else(|err| err.into_inner())
170 .insert(base, false);
171 strip_origin(&mut body);
172 return client.post_atom(&body).map_err(anyhow::Error::from);
173 }
174 Err(anyhow::Error::from(e))
175 }
176 }
177}
178
179fn read_latest() -> BTreeMap<String, Record> {
180 let bytes = crate::read_state_file(&proposals_path()).unwrap_or_default();
181 crate::complete_rows::<Record>(&bytes)
182 .into_iter()
183 .map(|rec| (rec.id.clone(), rec))
184 .collect()
185}
186
187fn append(rec: &Record) -> Result<()> {
188 let path = proposals_path();
189 if let Some(dir) = path.parent() {
190 std::fs::create_dir_all(dir)?;
191 }
192 let mut line = serde_json::to_string(rec)?;
193 line.push('\n');
194 crate::append_locked_row(&path, line.as_bytes(), crate::Durability::Durable)
195 .with_context(|| format!("admit: cannot write {}", path.display()))
196}
197
198fn proposal_is_local(base: &str) -> bool {
199 proposal_cache()
200 .lock()
201 .unwrap_or_else(|e| e.into_inner())
202 .get(base)
203 .copied()
204 == Some(false)
205}
206
207fn remember_proposal_mode(base: &str, remote: bool) {
208 proposal_cache()
209 .lock()
210 .unwrap_or_else(|e| e.into_inner())
211 .insert(base.to_string(), remote);
212}
213
214fn held_proposal_id(text: &str) -> Option<String> {
216 let marker = "held as proposal ";
217 let rest = text.get(text.find(marker)? + marker.len()..)?;
218 let id: String = rest.chars().take_while(|c| c.is_ascii_hexdigit()).collect();
219 (id.len() == 32).then_some(id)
220}
221
222fn filing_is_local(text: &str) -> bool {
224 let lower = text.to_lowercase();
225 origin_rejected(text)
226 || lower.contains(": 404:")
227 || lower.contains("status code 404")
228 || (lower.contains("403") && lower.contains("not allowed"))
229}
230
231fn post_direct(client: &PacksetClient, atom: &Value) -> Result<Option<String>> {
233 let base = client.base();
234 if proposal_is_local(base) {
235 return Ok(None);
236 }
237 match client.post_atom(atom) {
238 Ok(_) => {
239 remember_proposal_mode(base, false);
241 Ok(None)
242 }
243 Err(e) => {
244 let text = e.to_string();
245 if let Some(id) = held_proposal_id(&text) {
246 remember_proposal_mode(base, true);
247 return Ok(Some(id));
248 }
249 if filing_is_local(&text) {
250 remember_proposal_mode(base, false);
251 return Ok(None);
252 }
253 if !text.contains("bad response") {
256 return Ok(None);
257 }
258 bail!("propose: POST /v1/atoms failed: {text}");
259 }
260 }
261}
262
263pub fn write_agent(client: &PacksetClient, atom: Value) -> Result<Value> {
278 let text = atom["text"].as_str().unwrap_or("").trim().to_string();
279 if text.is_empty() {
280 bail!("remember: empty text is not a claim");
281 }
282 if atom["kind"].as_str() == Some("lesson") && past_hard_cap(&text) {
283 let copies = crate::pack_copy::pack_copies(&text);
284 let unchanged = copies.is_empty() || (copies.len() == 1 && copies[0] == text);
285 if !unchanged {
286 let mut last = None;
287 let mut first_err = None;
288 for copy in &copies {
289 let mut one = atom.clone();
290 one["text"] = Value::String(copy.clone());
291 match write_agent_one(client, one) {
292 Ok(posted) => last = Some(posted),
293 Err(e) if crate::note_can_wait(&e) => return Err(e),
294 Err(e) => {
295 if first_err.is_none() {
296 first_err = Some(e);
297 }
298 }
299 }
300 }
301 if let Some(posted) = last {
302 return Ok(posted);
303 }
304 return Err(
305 first_err.unwrap_or_else(|| anyhow::anyhow!("remember: empty text is not a claim"))
306 );
307 }
308 }
309 write_agent_one(client, atom)
310}
311
312fn past_hard_cap(text: &str) -> bool {
315 packset_core::prose::sentences(text).len() > packset_core::prose::MAX_ATOM_SENTENCES
316 || text.chars().count() > packset_core::record::TEXT_SOFT_CAP
317}
318
319pub(crate) fn write_agent_one(client: &PacksetClient, mut atom: Value) -> Result<Value> {
326 let text = atom["text"].as_str().unwrap_or("").trim();
327 if text.is_empty() {
328 bail!("remember: empty text is not a claim");
329 }
330 stamp_origin(&mut atom, ORIGIN_AGENT);
331 atom["source"]["basis"] = Value::String(crate::Basis::Agent.as_str().into());
332 match post_kept(client, &atom) {
333 Ok(posted) => Ok(posted),
334 Err(e) => {
335 let text = format!("{e:#}");
336 let Some(id) = held_proposal_id(&text) else {
337 return Err(e);
338 };
339 let workspace = atom
340 .get("workspace")
341 .and_then(Value::as_str)
342 .filter(|w| !w.is_empty())
343 .map_or_else(|| client.workspace(), str::to_string);
344 let accepted = accept_on_writer(client, &workspace, &id)?;
345 relabel_agent(client, &workspace, accepted)
349 }
350 }
351}
352
353pub fn propose_atom(client: &PacksetClient, atom: Value) -> Result<Filed> {
359 file_proposal(client, atom, true)
360}
361
362pub fn propose_admission(client: &PacksetClient, atom: Value) -> Result<Filed> {
370 file_proposal(client, atom, false)
371}
372
373fn file_proposal(client: &PacksetClient, mut atom: Value, post: bool) -> Result<Filed> {
374 let _g = gate();
375 let text = atom["text"].as_str().unwrap_or("").trim().to_string();
376 let kind = atom["kind"].as_str().unwrap_or("lesson").to_string();
377 if text.is_empty() {
378 bail!("propose: empty text is not a proposal");
379 }
380 if atom
381 .get("workspace")
382 .and_then(Value::as_str)
383 .unwrap_or("")
384 .is_empty()
385 {
386 if let Some(map) = atom.as_object_mut() {
387 map.insert("workspace".into(), Value::String(client.workspace()));
388 }
389 }
390 if post {
391 stamp_origin(&mut atom, ORIGIN_AGENT);
392 } else {
393 let named = atom["origin"].as_str().unwrap_or("");
394 if !packset_core::atom::known_origin(named) {
395 stamp_origin(&mut atom, ORIGIN_AGENT);
396 }
397 let origin = atom["origin"].as_str().unwrap_or(ORIGIN_AGENT).to_string();
398 if let Some(map) = atom.as_object_mut() {
399 map.insert(
400 "level".into(),
401 Value::String(admission_level(&origin).to_string()),
402 );
403 }
404 }
405 let origin = atom["origin"].as_str().unwrap_or(ORIGIN_AGENT).to_string();
406 let id = crate::work_id(&format!("{kind}\n{text}"));
407 if let Some(existing) = read_latest().get(&id) {
408 if matches!(existing.status.as_str(), "open" | "accepted" | "satisfied") {
409 return Ok(Filed {
410 id,
411 remote: existing.remote,
412 });
413 }
414 }
415 let remote_id = if post {
416 post_direct(client, &atom)?
417 } else {
418 None
419 };
420 let rec = Record {
421 id: id.clone(),
422 status: "open".into(),
423 remote: remote_id.is_some(),
424 remote_id,
425 text,
426 kind,
427 origin,
428 atom,
429 };
430 let remote = rec.remote;
431 append(&rec)?;
432 Ok(Filed { id, remote })
433}
434
435fn promote_level(atom: &mut Value) {
437 if let Some(map) = atom.as_object_mut() {
438 map.insert("level".into(), Value::String(LEVEL_EXPLICIT.to_string()));
439 }
440}
441
442pub fn satisfy_text(text: &str) {
445 let _g = gate();
446 let text = text.trim();
447 if text.is_empty() {
448 return;
449 }
450 let open: Vec<Record> = read_latest()
451 .into_values()
452 .filter(|rec| rec.status == "open" && rec.text.trim() == text)
453 .collect();
454 for mut rec in open {
455 rec.status = "satisfied".into();
456 if let Err(e) = append(&rec) {
460 eprintln!(
461 "ljos: a satisfied proposal was not recorded in {}: {e:#}",
462 proposals_path().display()
463 );
464 }
465 close_remote(&rec);
466 }
467}
468
469fn workspace_of(client: &PacksetClient, rec: &Record) -> String {
470 let workspace = rec
471 .atom
472 .get("workspace")
473 .and_then(Value::as_str)
474 .unwrap_or("");
475 if workspace.is_empty() {
476 client.workspace()
477 } else {
478 workspace.to_string()
479 }
480}
481
482fn close_remote(rec: &Record) {
485 if !rec.remote {
486 return;
487 }
488 let Some(remote_id) = rec.remote_id.as_deref() else {
489 return;
490 };
491 let Ok(client) = crate::pack() else {
492 return;
493 };
494 let workspace = workspace_of(&client, rec);
495 let _ = accept_on_writer(&client, &workspace, remote_id);
496}
497
498fn accepted_line(id: &str, kind: &str, posted: &Value) -> String {
499 let origin = posted
500 .get("origin")
501 .and_then(Value::as_str)
502 .filter(|origin| !origin.is_empty())
503 .unwrap_or(ORIGIN_AGENT);
504 format!(
505 "accepted {id} as {kind} origin {origin}{}\n",
506 crate::revision_note(posted)
507 )
508}
509
510fn relabel_agent(client: &PacksetClient, workspace: &str, accepted: Value) -> Result<Value> {
514 let origin = accepted.get("origin").and_then(Value::as_str).unwrap_or("");
515 if origin == ORIGIN_AGENT {
516 return Ok(accepted);
517 }
518 let Some(id) = accepted
519 .get("id")
520 .and_then(Value::as_str)
521 .filter(|id| !id.is_empty())
522 else {
523 bail!(
524 "accept stored the lesson without an id, so its origin could not be put back to agent-derived"
525 );
526 };
527 let body = serde_json::json!({
528 "workspace": workspace,
529 "id": id,
530 "fields": {"origin": ORIGIN_AGENT},
531 });
532 let bytes = serde_json::to_vec(&body).unwrap_or_default();
533 match client.roundtrip(
534 "POST",
535 "/v1/atoms/update",
536 &[],
537 Some(&bytes),
538 true,
539 http_timeout(),
540 ) {
541 Ok(answer) if (200..300).contains(&answer.status) => serde_json::from_str(&answer.body)
542 .context("update: the pack's answer was not json"),
543 Ok(answer) => bail!(
544 "the lesson {id} was stored as {origin}; putting the origin back to agent-derived failed: {}: {}",
545 answer.status,
546 answer.body
547 ),
548 Err(e) => bail!(
549 "the lesson {id} was stored as {origin}; putting the origin back to agent-derived failed: {e}"
550 ),
551 }
552}
553
554fn accept_on_writer(client: &PacksetClient, workspace: &str, id: &str) -> Result<Value> {
555 let body = serde_json::json!({"workspace": workspace, "id": id});
556 let bytes = serde_json::to_vec(&body).unwrap_or_default();
557 match client.roundtrip(
558 "POST",
559 "/v1/proposals/accept",
560 &[],
561 Some(&bytes),
562 true,
563 http_timeout(),
564 ) {
565 Ok(answer) if (200..300).contains(&answer.status) => {
566 serde_json::from_str(&answer.body).context("accept: the pack's answer was not json")
567 }
568 Ok(answer) => {
569 bail!(
570 "accept: POST /v1/proposals/accept failed: {}: {}",
571 answer.status,
572 answer.body
573 )
574 }
575 Err(e) => bail!("accept: POST /v1/proposals/accept failed: {e}"),
576 }
577}
578
579pub fn resolve_id(id: &str) -> Result<String> {
586 let open: Vec<String> = read_latest()
587 .into_values()
588 .filter(|r| r.status == "open")
589 .map(|r| r.id)
590 .collect();
591 crate::resolve_short_id("accept", id, &open)
592}
593
594#[must_use]
597pub fn proposed_by(id: &str) -> Option<String> {
598 let rec = read_latest().get(id.trim()).cloned()?;
599 rec.atom["source"]["session"]
600 .as_str()
601 .map(str::trim)
602 .filter(|s| !s.is_empty())
603 .map(str::to_string)
604}
605
606pub fn accept(id: &str) -> Result<String> {
609 let _g = gate();
610 let id = id.trim();
611 if id.is_empty() {
612 bail!("accept: a proposal id is required");
613 }
614 let Some(mut rec) = read_latest().get(id).cloned() else {
615 bail!("accept: no open proposal {id}");
616 };
617 match rec.status.as_str() {
618 "satisfied" => {
619 close_remote(&rec);
620 return Ok(format!("{id} was already written by remember or prefer\n"));
621 }
622 "accepted" => return Ok(format!("{id} is already in the pack\n")),
623 "open" => {}
624 _ => bail!("accept: no open proposal {id}"),
625 }
626 let client = crate::pack()?;
627 if rec.remote {
628 let remote_id = rec.remote_id.clone().unwrap_or_else(|| rec.id.clone());
629 let workspace = workspace_of(&client, &rec);
630 let posted = accept_on_writer(&client, &workspace, &remote_id)?;
631 rec.status = "accepted".into();
632 append(&rec)?;
633 return Ok(accepted_line(&rec.id, &rec.kind, &posted));
634 }
635 promote_level(&mut rec.atom);
639 match post_kept(&client, &rec.atom) {
640 Ok(posted) => {
641 rec.status = "accepted".into();
642 append(&rec)?;
643 Ok(accepted_line(&rec.id, &rec.kind, &posted))
644 }
645 Err(e) => {
646 let text = e.to_string();
647 let Some(remote_id) = held_proposal_id(&text) else {
648 return Err(e).with_context(|| format!("accept: POST /v1/atoms failed for {id}"));
649 };
650 let workspace = workspace_of(&client, &rec);
653 let posted = accept_on_writer(&client, &workspace, &remote_id)?;
654 rec.remote = true;
655 rec.remote_id = Some(remote_id);
656 rec.status = "accepted".into();
657 append(&rec)?;
658 Ok(accepted_line(&rec.id, &rec.kind, &posted))
659 }
660 }
661}
662
663#[cfg(test)]
664mod tests {
665 use super::*;
666 use std::os::unix::fs::PermissionsExt;
667
668 fn record(id: String) -> Record {
669 Record {
670 id,
671 status: "proposed".into(),
672 remote: false,
673 remote_id: None,
674 text: "use uv for python envs".into(),
675 kind: "lesson".into(),
676 origin: ORIGIN_AGENT.into(),
677 atom: serde_json::json!({"text": "use uv for python envs"}),
678 }
679 }
680
681 #[test]
684 fn concurrent_proposal_rows_each_parse_and_stay_private() {
685 let _g = crate::test_env_lock();
686 let dir = tempfile::tempdir().unwrap();
687 unsafe { std::env::set_var("XDG_STATE_HOME", dir.path()) };
688 const THREADS: usize = 16;
689 const EACH: usize = 8;
690 let handles: Vec<_> = (0..THREADS)
691 .map(|thread| {
692 std::thread::spawn(move || {
693 for n in 0..EACH {
694 append(&record(format!("{thread}-{n}"))).unwrap();
695 }
696 })
697 })
698 .collect();
699 for handle in handles {
700 handle.join().expect("proposal writer");
701 }
702 let path = proposals_path();
703 let text = std::fs::read_to_string(&path).unwrap();
704 for line in text.lines() {
705 serde_json::from_str::<Record>(line)
706 .unwrap_or_else(|err| panic!("proposal row does not parse ({err}): {line}"));
707 }
708 assert_eq!(read_latest().len(), THREADS * EACH, "{text}");
709 let mode = std::fs::symlink_metadata(&path)
710 .unwrap()
711 .permissions()
712 .mode()
713 & 0o777;
714 assert_eq!(mode, 0o600, "proposals mode {mode:o}");
715 }
716
717 #[test]
720 fn a_torn_last_proposal_leaves_the_others() {
721 let _g = crate::test_env_lock();
722 let dir = tempfile::tempdir().unwrap();
723 unsafe { std::env::set_var("XDG_STATE_HOME", dir.path()) };
724 append(&record("a".into())).unwrap();
725 let path = proposals_path();
726 let mut bytes = std::fs::read(&path).unwrap();
727 bytes.extend_from_slice(b"{\"id\":\"b\",\"text\":\"caf\xc3");
728 std::fs::write(&path, bytes).unwrap();
729 let ids: Vec<String> = read_latest().into_keys().collect();
730 assert_eq!(ids, ["a"]);
731 }
732
733 #[test]
735 fn a_symlinked_proposals_file_is_refused() {
736 let _g = crate::test_env_lock();
737 let dir = tempfile::tempdir().unwrap();
738 unsafe { std::env::set_var("XDG_STATE_HOME", dir.path()) };
739 let path = proposals_path();
740 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
741 let outside = dir.path().join("outside");
742 std::fs::write(&outside, "kept\n").unwrap();
743 std::os::unix::fs::symlink(&outside, &path).unwrap();
744 assert!(append(&record("a".into())).is_err());
745 assert_eq!(std::fs::read_to_string(&outside).unwrap(), "kept\n");
746 }
747}