vissue_tui/
control.rs

1//! Socket-backed board. Unix only; clients never bind the control socket.
2
3use std::path::Path;
4use std::sync::Mutex;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::time::Duration;
7
8use serde_json::Value;
9use vissue_control::client::Client;
10use vissue_control::rpc::{
11    CONFLICT, CYCLE, ClaimParams, CreateParams, DeedParams, Error as RpcError, INVALID_STATE,
12    IdParams, InitializeResult, IssueListParams, IssueListResult, MutResult as WireMut, NOT_FOUND,
13    NoteParams, Notification, RecallParams, RelatedParams, Request, SearchParams, TreeParams,
14    UpdateParams,
15};
16use vissue_control::{InitializeParams, PROTOCOL_VERSION};
17use vissue_core::config::Layout;
18use vissue_core::error::Error;
19use vissue_core::views::{
20    AgendaRow, ClaimRow, Excerpt, IssueDetail, ListQuery, Recall, RelatedHit, SearchHit, TreeNode,
21};
22
23use crate::backend::{BackendKind, BoardBackend, ListPage, MutResult, SinceGate, UpdateReq};
24
25/// JSON-RPC client after a matching `initialize`.
26#[derive(Debug)]
27pub struct ControlBackend {
28    layout: Layout,
29    identity: String,
30    client: Mutex<Client>,
31    generation: AtomicU64,
32    revision: AtomicU64,
33    /// Revision of the last full list/ready page, not the last mut/notify.
34    page_revision: AtomicU64,
35    since: SinceGate,
36    last_query: Mutex<Option<ListQuery>>,
37    last_since: Mutex<Option<Option<u64>>>,
38}
39
40impl ControlBackend {
41    /// Connect, `initialize` with a required agent, and refuse a root/prefix
42    /// mismatch so mutations never hit the wrong vault.
43    ///
44    /// # Errors
45    ///
46    /// Returns an error if the socket cannot be reached, `initialize` fails, or
47    /// the serve root/prefix does not match `layout`.
48    pub fn connect(path: &Path, layout: &Layout, agent: &str) -> Result<Self, ControlAttachError> {
49        Self::connect_as(path, layout, agent, "vissue-tui")
50    }
51
52    /// Same as [`Self::connect`] with an explicit `initialize.client` name.
53    ///
54    /// # Errors
55    ///
56    /// Returns an error if the socket cannot be reached, `initialize` fails, or
57    /// the serve root/prefix does not match `layout`.
58    pub fn connect_as(
59        path: &Path,
60        layout: &Layout,
61        agent: &str,
62        client: &str,
63    ) -> Result<Self, ControlAttachError> {
64        let mut client_conn = Client::connect(path).map_err(ControlAttachError::Rpc)?;
65        let params = InitializeParams {
66            protocol_version: PROTOCOL_VERSION,
67            client: client.into(),
68            agent: agent.to_string(),
69        };
70        let value = client_conn
71            .request_typed(&Request::Initialize(params))
72            .map_err(ControlAttachError::Rpc)?;
73        let init: InitializeResult = serde_json::from_value(value)
74            .map_err(|e| ControlAttachError::Rpc(RpcError::Json(e)))?;
75        if !roots_match(layout, &init.root, &init.prefix) {
76            return Err(ControlAttachError::Mismatch {
77                want_root: layout.root().display().to_string(),
78                want_prefix: layout.prefix().to_string(),
79                got_root: init.root,
80                got_prefix: init.prefix,
81            });
82        }
83        Ok(Self {
84            layout: layout.clone(),
85            identity: init.identity,
86            client: Mutex::new(client_conn),
87            generation: AtomicU64::new(init.generation),
88            revision: AtomicU64::new(init.revision),
89            page_revision: AtomicU64::new(0),
90            since: SinceGate::after_attach(),
91            last_query: Mutex::new(None),
92            last_since: Mutex::new(None),
93        })
94    }
95
96    fn call(&self, req: &Request) -> Result<Value, Error> {
97        let mut client = self.client.lock().expect("control client");
98        client.request_typed(req).map_err(map_rpc)
99    }
100
101    fn list_params(&self, q: ListQuery) -> IssueListParams {
102        // Serve `unchanged` is catalog-wide. Only send since_revision when
103        // this is the same ready/project/query as the last full page.
104        let mut last_query = self.last_query.lock().expect("query");
105        let same = last_query.as_ref() == Some(&q);
106        *last_query = Some(q.clone());
107        drop(last_query);
108        let page = self.page_revision.load(Ordering::SeqCst);
109        let since = if same {
110            self.since.next(page)
111        } else {
112            self.since.invalidate();
113            let _ = self.since.next(page);
114            None
115        };
116        *self.last_since.lock().expect("since") = Some(since);
117        IssueListParams {
118            project: q.project,
119            state: q.state,
120            ready: if q.ready { Some(true) } else { None },
121            query: q.query,
122            limit: q.limit,
123            offset: q.offset,
124            since_revision: since,
125        }
126    }
127
128    fn apply_list(&self, result: IssueListResult) -> ListPage {
129        if !result.unchanged {
130            self.page_revision.store(result.revision, Ordering::SeqCst);
131            self.revision.store(result.revision, Ordering::SeqCst);
132            self.generation.store(result.generation, Ordering::SeqCst);
133        }
134        ListPage {
135            issues: result.issues,
136            total: result.total,
137            matched: result.matched,
138            revision: result.revision,
139            generation: result.generation,
140            unchanged: result.unchanged,
141        }
142    }
143
144    fn apply_mut(&self, wire: WireMut) -> MutResult {
145        self.revision.store(wire.revision, Ordering::SeqCst);
146        self.generation.store(wire.generation, Ordering::SeqCst);
147        MutResult {
148            ok: wire.ok,
149            report: wire.report,
150            issue: wire.issue,
151            revision: wire.revision,
152            generation: wire.generation,
153        }
154    }
155}
156
157fn roots_match(layout: &Layout, root: &str, prefix: &str) -> bool {
158    let want_root = layout.root().display().to_string();
159    (root == want_root || Path::new(root) == layout.root()) && prefix == layout.prefix()
160}
161
162/// Why attach refused the live socket.
163#[derive(Debug)]
164pub enum ControlAttachError {
165    /// Socket, framing, or JSON-RPC failure.
166    Rpc(RpcError),
167    /// Serve answered for a different vault than `layout`.
168    Mismatch {
169        /// Board layout root.
170        want_root: String,
171        /// Board layout prefix.
172        want_prefix: String,
173        /// Root `initialize` returned.
174        got_root: String,
175        /// Prefix `initialize` returned.
176        got_prefix: String,
177    },
178}
179
180impl std::fmt::Display for ControlAttachError {
181    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
182        match self {
183            Self::Rpc(err) => write!(f, "{err}"),
184            Self::Mismatch {
185                want_root,
186                want_prefix,
187                got_root,
188                got_prefix,
189            } => write!(
190                f,
191                "serve root/prefix mismatch: want {want_root} {want_prefix}, got {got_root} {got_prefix}"
192            ),
193        }
194    }
195}
196
197impl std::error::Error for ControlAttachError {}
198
199fn map_rpc(err: RpcError) -> Error {
200    match err {
201        RpcError::Rpc(rpc) => match rpc.code {
202            NOT_FOUND => Error::IssueNotFound {
203                id: rpc
204                    .data
205                    .as_ref()
206                    .and_then(|d| d.get("id"))
207                    .and_then(Value::as_str)
208                    .unwrap_or("")
209                    .to_string(),
210            },
211            CONFLICT
212                if rpc
213                    .data
214                    .as_ref()
215                    .and_then(|d| d.get("code"))
216                    .and_then(Value::as_str)
217                    == Some("duplicate_id") =>
218            {
219                Error::DuplicateId {
220                    id: rpc
221                        .data
222                        .as_ref()
223                        .and_then(|d| d.get("id"))
224                        .and_then(Value::as_str)
225                        .unwrap_or("")
226                        .to_string(),
227                    paths: rpc
228                        .data
229                        .as_ref()
230                        .and_then(|d| d.get("paths"))
231                        .and_then(Value::as_array)
232                        .map(|arr| {
233                            arr.iter()
234                                .filter_map(Value::as_str)
235                                .map(std::path::PathBuf::from)
236                                .collect()
237                        })
238                        .unwrap_or_default(),
239                }
240            }
241            CONFLICT => Error::ClaimConflict {
242                id: rpc
243                    .data
244                    .as_ref()
245                    .and_then(|d| d.get("id"))
246                    .and_then(Value::as_str)
247                    .unwrap_or("")
248                    .to_string(),
249                holder: rpc
250                    .data
251                    .as_ref()
252                    .and_then(|d| d.get("holder"))
253                    .and_then(Value::as_str)
254                    .unwrap_or("")
255                    .to_string(),
256                claimed_at: None,
257            },
258            CYCLE => Error::BlockerCycle {
259                blocker: rpc
260                    .data
261                    .as_ref()
262                    .and_then(|d| d.get("block"))
263                    .and_then(Value::as_str)
264                    .unwrap_or("")
265                    .to_string(),
266                issue: rpc
267                    .data
268                    .as_ref()
269                    .and_then(|d| d.get("id"))
270                    .and_then(Value::as_str)
271                    .unwrap_or("")
272                    .to_string(),
273            },
274            INVALID_STATE => Error::InvalidState {
275                id: rpc
276                    .data
277                    .as_ref()
278                    .and_then(|d| d.get("id"))
279                    .and_then(Value::as_str)
280                    .unwrap_or("")
281                    .to_string(),
282                state: rpc
283                    .data
284                    .as_ref()
285                    .and_then(|d| d.get("state"))
286                    .and_then(Value::as_str)
287                    .unwrap_or("")
288                    .to_string(),
289            },
290            _ => Error::Other(anyhow::anyhow!("{}", rpc.message)),
291        },
292        other => Error::Other(anyhow::anyhow!("{other}")),
293    }
294}
295
296fn decode<T: serde::de::DeserializeOwned>(value: Value) -> Result<T, Error> {
297    serde_json::from_value(value).map_err(|e| Error::Other(e.into()))
298}
299
300impl BoardBackend for ControlBackend {
301    fn layout(&self) -> &Layout {
302        &self.layout
303    }
304
305    fn generation(&self) -> u64 {
306        self.generation.load(Ordering::SeqCst)
307    }
308
309    fn revision(&self) -> u64 {
310        self.revision.load(Ordering::SeqCst)
311    }
312
313    fn live(&self) -> BackendKind {
314        BackendKind::Control
315    }
316
317    fn identity(&self) -> &str {
318        &self.identity
319    }
320
321    fn list(&self, q: ListQuery) -> Result<ListPage, Error> {
322        let params = self.list_params(q);
323        let value = self.call(&Request::IssueList(params))?;
324        Ok(self.apply_list(decode(value)?))
325    }
326
327    fn ready(&self, project: Option<&str>) -> Result<ListPage, Error> {
328        let params = self.list_params(ListQuery {
329            project: project.map(str::to_string),
330            ready: true,
331            ..ListQuery::default()
332        });
333        let value = self.call(&Request::IssueReady(params))?;
334        Ok(self.apply_list(decode(value)?))
335    }
336
337    fn get(&self, id: &str) -> Result<IssueDetail, Error> {
338        let value = self.call(&Request::IssueGet(IdParams { id: id.to_string() }))?;
339        let row: vissue_control::rpc::IssueGetResult = decode(value)?;
340        Ok(row.issue)
341    }
342
343    fn excerpt(&self, id: &str) -> Result<Excerpt, Error> {
344        let value = self.call(&Request::IssueExcerpt(IdParams { id: id.to_string() }))?;
345        decode(value)
346    }
347
348    fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchHit>, Error> {
349        let value = self.call(&Request::IssueSearch(SearchParams {
350            query: query.to_string(),
351            limit: Some(limit),
352        }))?;
353        decode(value)
354    }
355
356    fn claims(&self, holder: Option<&str>, project: Option<&str>) -> Result<Vec<ClaimRow>, Error> {
357        let value = self.call(&Request::IssueClaims(vissue_control::rpc::ClaimsParams {
358            holder: holder.map(str::to_string),
359            project: project.map(str::to_string),
360        }))?;
361        decode(value)
362    }
363
364    fn agenda(&self, days: i64, project: Option<&str>) -> Result<Vec<AgendaRow>, Error> {
365        let value = self.call(&Request::IssueAgenda(vissue_control::rpc::AgendaParams {
366            days: Some(days),
367            project: project.map(str::to_string),
368        }))?;
369        decode(value)
370    }
371
372    fn tree(&self, id: &str) -> Result<TreeNode, Error> {
373        let value = self.call(&Request::IssueTree(TreeParams {
374            id: id.to_string(),
375            format: Some("nodes".into()),
376        }))?;
377        match decode::<vissue_control::rpc::TreeResult>(value)? {
378            vissue_control::rpc::TreeResult::Nodes(node) => Ok(node),
379            vissue_control::rpc::TreeResult::Text { text } => Err(Error::Other(anyhow::anyhow!(
380                "serve returned tree text, not nodes: {text}"
381            ))),
382        }
383    }
384
385    fn related(&self, id: &str, depth: usize, limit: usize) -> Result<Vec<RelatedHit>, Error> {
386        let value = self.call(&Request::IssueRelated(RelatedParams {
387            id: id.to_string(),
388            depth: Some(depth),
389            limit: Some(limit),
390        }))?;
391        decode(value)
392    }
393
394    fn recall(&self, id: &str, depth: usize) -> Result<Recall, Error> {
395        let value = self.call(&Request::IssueRecall(RecallParams {
396            id: id.to_string(),
397            depth: Some(depth),
398            excerpts: false,
399        }))?;
400        decode(value)
401    }
402
403    fn deed(&self, id: &str, add: &[String]) -> Result<MutResult, Error> {
404        let value = self.call(&Request::IssueDeed(DeedParams {
405            id: id.to_string(),
406            add: add.to_vec(),
407            remove: Vec::new(),
408        }))?;
409        Ok(self.apply_mut(decode(value)?))
410    }
411
412    fn projects(&self) -> Result<Vec<String>, Error> {
413        let value = self.call(&Request::ProjectList)?;
414        let row: vissue_control::rpc::ProjectListResult = decode(value)?;
415        Ok(row.projects)
416    }
417
418    fn claim(&self, id: &str, force: bool) -> Result<MutResult, Error> {
419        let value = self.call(&Request::IssueClaim(ClaimParams {
420            id: id.to_string(),
421            force,
422            agent: None,
423        }))?;
424        Ok(self.apply_mut(decode(value)?))
425    }
426
427    fn note(&self, id: &str, text: &str) -> Result<MutResult, Error> {
428        let value = self.call(&Request::IssueNote(NoteParams {
429            id: id.to_string(),
430            text: text.to_string(),
431        }))?;
432        Ok(self.apply_mut(decode(value)?))
433    }
434
435    fn update(&self, req: UpdateReq) -> Result<MutResult, Error> {
436        let value = self.call(&Request::IssueUpdate(UpdateParams {
437            id: req.id,
438            state: req.state,
439            priority: req.priority.map(|c| c.to_string()),
440            block: req.block,
441            unblock: req.unblock,
442            if_state: req.if_state,
443            if_gen: req.if_gen,
444            agent: None,
445        }))?;
446        Ok(self.apply_mut(decode(value)?))
447    }
448
449    fn create(&self, project: &str, title: &str) -> Result<MutResult, Error> {
450        let value = self.call(&Request::IssueCreate(CreateParams {
451            project: project.to_string(),
452            title: title.to_string(),
453            agent: None,
454            priority: None,
455            issue_type: None,
456            deadline: None,
457            scheduled: None,
458            tags: None,
459            parent: None,
460            body: None,
461        }))?;
462        Ok(self.apply_mut(decode(value)?))
463    }
464
465    fn open(&self, id: &str) -> Result<IssueDetail, Error> {
466        let value = self.call(&Request::IssueOpen(IdParams { id: id.to_string() }))?;
467        let row: vissue_control::rpc::IssueGetResult = decode(value)?;
468        Ok(row.issue)
469    }
470
471    /// # Panics
472    ///
473    /// Panics if the control client lock is poisoned.
474    fn wait(&self, last: u64, timeout_ms: u64) -> Result<u64, Error> {
475        let mut client = self.client.lock().expect("control client");
476        match client.wait_notification(Duration::from_millis(timeout_ms.max(1))) {
477            Ok(Notification::VaultChanged(changed)) => {
478                self.revision.store(changed.revision, Ordering::SeqCst);
479                self.generation.store(changed.generation, Ordering::SeqCst);
480                Ok(changed.revision)
481            }
482            Ok(_) => Ok(self.revision.load(Ordering::SeqCst)),
483            Err(_) => Ok(last),
484        }
485    }
486
487    /// # Panics
488    ///
489    /// Panics if the since lock is poisoned.
490    fn last_since_revision(&self) -> Option<Option<u64>> {
491        *self.last_since.lock().expect("since")
492    }
493
494    /// # Panics
495    ///
496    /// Panics if the query lock is poisoned.
497    fn invalidate_since(&self) {
498        self.since.invalidate();
499        *self.last_query.lock().expect("query") = None;
500    }
501}
502
503#[cfg(test)]
504mod tests {
505    use super::*;
506    use crate::backend::{BoardBackend, UpdateReq};
507    use serde_json::json;
508    use std::io::{BufReader, Write};
509    use std::os::unix::net::UnixListener;
510    use std::sync::{Arc, Mutex};
511    use std::thread;
512    use vissue_control::frame::{read_message, write_message};
513    use vissue_control::rpc::JsonRpcRequest;
514    use vissue_core::views::ListQuery;
515
516    #[test]
517    fn after_initialize_the_next_list_omits_since_revision() {
518        let dir = tempfile::tempdir().unwrap();
519        let sock = dir.path().join("control.sock");
520        let layout = Layout::new(dir.path().join("vault"), "Software");
521        let seen = Arc::new(Mutex::new(Vec::new()));
522        let seen_cb = Arc::clone(&seen);
523        let root = layout.root().display().to_string();
524        let listener = UnixListener::bind(&sock).unwrap();
525        thread::spawn(move || {
526            let (stream, _) = listener.accept().unwrap();
527            let mut reader = BufReader::new(stream.try_clone().unwrap());
528            let mut writer = stream;
529            while let Ok((payload, framing)) = read_message(&mut reader) {
530                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
531                let body = if req.method == "initialize" {
532                    json!({
533                        "jsonrpc": "2.0",
534                        "id": req.id,
535                        "result": {
536                            "protocolVersion": 1,
537                            "capabilities": [],
538                            "root": root,
539                            "prefix": "Software",
540                            "generation": 9,
541                            "revision": 41,
542                            "identity": "tui"
543                        }
544                    })
545                } else {
546                    let since = req
547                        .params
548                        .as_ref()
549                        .and_then(|p| p.get("since_revision"))
550                        .cloned();
551                    seen_cb.lock().unwrap().push(since);
552                    json!({
553                        "jsonrpc": "2.0",
554                        "id": req.id,
555                        "result": {
556                            "issues": [],
557                            "total": 0,
558                            "matched": 0,
559                            "revision": 41,
560                            "generation": 9,
561                            "unchanged": false
562                        }
563                    })
564                };
565                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
566                writer.flush().unwrap();
567            }
568        });
569
570        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
571        assert_eq!(backend.revision(), 41);
572        assert_eq!(backend.live(), BackendKind::Control);
573        backend.ready(None).unwrap();
574        assert_eq!(backend.last_since_revision(), Some(None));
575        backend.ready(None).unwrap();
576        assert_eq!(backend.last_since_revision(), Some(Some(41)));
577        backend.list(ListQuery::default()).unwrap();
578        assert_eq!(backend.last_since_revision(), Some(None));
579        let seen = seen.lock().unwrap();
580        assert_eq!(seen.len(), 3);
581        assert_eq!(seen[0], None);
582        assert_eq!(seen[1], Some(json!(41)));
583        assert_eq!(seen[2], None);
584    }
585
586    fn serve_methods(path: &std::path::Path, root: String) {
587        let listener = UnixListener::bind(path).unwrap();
588        thread::spawn(move || {
589            let (stream, _) = listener.accept().unwrap();
590            let mut reader = BufReader::new(stream.try_clone().unwrap());
591            let mut writer = stream;
592            while let Ok((payload, framing)) = read_message(&mut reader) {
593                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
594                let result = match req.method.as_str() {
595                    "initialize" => json!({
596                        "protocolVersion":1,"capabilities":[],"root":root,
597                        "prefix":"Software","generation":2,"revision":3,"identity":"tui"
598                    }),
599                    "issue/get" | "issue/show" | "issue/open" => json!({
600                        "id":"atlas-1a2b","project":"atlas","title":"t","state":"TODO",
601                        "priority":"B","properties":{},"org_tags":[],"tags":[],
602                        "blocked_by":[],"parent":null,"claimed_by":null,"claimed_at":null,
603                        "file":"f","line_start":1,"line_end":2,"revision":3
604                    }),
605                    "issue/excerpt" => json!({
606                        "id":"atlas-1a2b","file":"f","line_start":1,"line_end":2,
607                        "text":"body","suppressed":false
608                    }),
609                    "issue/search" | "issue/claims" | "issue/agenda" | "issue/related" => {
610                        json!([])
611                    }
612                    "issue/tree" => json!({
613                        "id":"atlas-1a2b","state":"TODO","title":"t",
614                        "children":[],"blocked_by":[]
615                    }),
616                    "project/list" => json!({"projects":["atlas"],"revision":3}),
617                    "issue/claim" | "issue/note" | "issue/update" => json!({
618                        "ok":true,"report":"ok","issue":null,"revision":4,"generation":3
619                    }),
620                    "issue/list" | "issue/ready" => json!({
621                        "issues":[],"total":0,"matched":0,"revision":3,
622                        "generation":2,"unchanged":false
623                    }),
624                    other => panic!("unexpected {other}"),
625                };
626                let body = json!({"jsonrpc":"2.0","id":req.id,"result":result});
627                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
628                writer.flush().unwrap();
629            }
630        });
631    }
632
633    #[test]
634    fn control_verbs_roundtrip() {
635        let dir = tempfile::tempdir().unwrap();
636        let sock = dir.path().join("control.sock");
637        let layout = Layout::new(dir.path().join("vault"), "Software");
638        serve_methods(&sock, layout.root().display().to_string());
639        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
640        assert_eq!(backend.get("atlas-1a2b").unwrap().id, "atlas-1a2b");
641        assert_eq!(backend.excerpt("atlas-1a2b").unwrap().text, "body");
642        assert!(backend.search("x", 5).unwrap().is_empty());
643        assert!(backend.claims(None, None).unwrap().is_empty());
644        assert!(backend.agenda(14, None).unwrap().is_empty());
645        assert_eq!(backend.tree("atlas-1a2b").unwrap().id, "atlas-1a2b");
646        assert!(backend.related("atlas-1a2b", 2, 5).unwrap().is_empty());
647        assert_eq!(backend.projects().unwrap(), ["atlas"]);
648        assert!(backend.claim("atlas-1a2b", false).unwrap().ok);
649        assert!(backend.note("atlas-1a2b", "hi").unwrap().ok);
650        assert!(
651            backend
652                .update(UpdateReq {
653                    id: "atlas-1a2b".into(),
654                    state: Some("STARTED".into()),
655                    ..UpdateReq::default()
656                })
657                .unwrap()
658                .ok
659        );
660        assert_eq!(backend.open("atlas-1a2b").unwrap().id, "atlas-1a2b");
661        assert_eq!(backend.wait(3, 5).unwrap(), 3);
662    }
663
664    #[test]
665    fn after_claim_next_list_sends_page_revision_not_head() {
666        let dir = tempfile::tempdir().unwrap();
667        let sock = dir.path().join("control.sock");
668        let layout = Layout::new(dir.path().join("vault"), "Software");
669        let seen = Arc::new(Mutex::new(Vec::new()));
670        let seen_cb = Arc::clone(&seen);
671        let root = layout.root().display().to_string();
672        let listener = UnixListener::bind(&sock).unwrap();
673        thread::spawn(move || {
674            let (stream, _) = listener.accept().unwrap();
675            let mut reader = BufReader::new(stream.try_clone().unwrap());
676            let mut writer = stream;
677            while let Ok((payload, framing)) = read_message(&mut reader) {
678                let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
679                let result = match req.method.as_str() {
680                    "initialize" => json!({
681                        "protocolVersion":1,"capabilities":[],"root":root,
682                        "prefix":"Software","generation":2,"revision":10,"identity":"tui"
683                    }),
684                    "issue/ready" | "issue/list" => {
685                        let since = req
686                            .params
687                            .as_ref()
688                            .and_then(|p| p.get("since_revision"))
689                            .cloned();
690                        seen_cb.lock().unwrap().push(since);
691                        json!({
692                            "issues":[{
693                                "id":"atlas-2c3d","state":"TODO","priority":"B",
694                                "title":"Emit a summary table","project":"atlas",
695                                "blocked_by":[],"claimed_by":null,"claimed_at":null
696                            }],
697                            "total":1,"matched":1,"revision":10,
698                            "generation":2,"unchanged":false
699                        })
700                    }
701                    "issue/claim" => json!({
702                        "ok":true,"report":"claimed","issue":null,
703                        "revision":11,"generation":3
704                    }),
705                    other => panic!("unexpected {other}"),
706                };
707                let body = json!({"jsonrpc":"2.0","id":req.id,"result":result});
708                write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
709                writer.flush().unwrap();
710            }
711        });
712
713        let backend = ControlBackend::connect(&sock, &layout, "tui").unwrap();
714        let page = backend.ready(None).unwrap();
715        assert_eq!(page.issues[0].id, "atlas-2c3d");
716        assert_eq!(backend.last_since_revision(), Some(None));
717        assert!(backend.claim("atlas-2c3d", false).unwrap().ok);
718        assert_eq!(backend.revision(), 11);
719        backend.ready(None).unwrap();
720        assert_eq!(backend.last_since_revision(), Some(Some(10)));
721        let seen = seen.lock().unwrap();
722        assert_eq!(seen[0], None);
723        assert_eq!(seen[1], Some(json!(10)));
724    }
725
726    #[test]
727    fn root_mismatch_refuses_the_socket() {
728        let dir = tempfile::tempdir().unwrap();
729        let sock = dir.path().join("control.sock");
730        let layout = Layout::new(dir.path().join("vault"), "Software");
731        let listener = UnixListener::bind(&sock).unwrap();
732        thread::spawn(move || {
733            let (stream, _) = listener.accept().unwrap();
734            let mut reader = BufReader::new(stream.try_clone().unwrap());
735            let mut writer = stream;
736            let (payload, framing) = read_message(&mut reader).unwrap();
737            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
738            let body = json!({
739                "jsonrpc": "2.0",
740                "id": req.id,
741                "result": {
742                    "protocolVersion": 1,
743                    "capabilities": [],
744                    "root": "/other/vault",
745                    "prefix": "Software",
746                    "generation": 1,
747                    "revision": 1,
748                    "identity": "tui"
749                }
750            });
751            write_message(&mut writer, &serde_json::to_vec(&body).unwrap(), framing).unwrap();
752            writer.flush().unwrap();
753        });
754        let err = match ControlBackend::connect(&sock, &layout, "tui") {
755            Ok(_) => panic!("expected root mismatch"),
756            Err(err) => err,
757        };
758        match err {
759            ControlAttachError::Mismatch { got_root, .. } => {
760                assert_eq!(got_root, "/other/vault");
761            }
762            other => panic!("{other:?}"),
763        }
764    }
765}
vissue_tui/control.rs

vissue_tui/
control.rs
