objectiveai_cli/filesystem/
read.rs

1//! Local read + list of published agents/swarms/functions/profiles.
2//!
3//! Repositories live at `state_dir/<kind>/<owner>/<repository>` (the same
4//! layout [`publish`](super::publish) writes to) and are git repos. Reads
5//! resolve the file either from the working tree (resolving HEAD for the
6//! reported commit) or from a specific commit; listing walks the kind
7//! directory as a concurrent stream, resolving each repo's HEAD and
8//! validating its JSON, skipping anything that fails.
9//!
10//! All directory/file IO uses `tokio::fs`; git (libgit2, blocking) runs on
11//! `spawn_blocking`.
12
13use std::path::{Path, PathBuf};
14use std::pin::Pin;
15
16use futures::{Stream, StreamExt};
17use objectiveai_sdk::RemotePath;
18use serde::de::DeserializeOwned;
19
20use super::publish::Kind;
21use super::{Client, Error};
22
23impl Client {
24    /// `state_dir/<kind>/<owner>/<repository>`.
25    fn repo_path(&self, kind: Kind, owner: &str, repository: &str) -> PathBuf {
26        self.state_dir()
27            .join(kind.as_str())
28            .join(owner)
29            .join(repository)
30    }
31
32    /// Resolves the HEAD commit SHA for a repository.
33    pub fn resolve_head(
34        &self,
35        kind: Kind,
36        owner: &str,
37        repository: &str,
38    ) -> Result<String, Error> {
39        Ok(head_sha(&self.repo_path(kind, owner, repository))?)
40    }
41
42    /// Reads and deserializes a kind's JSON file from a repository.
43    ///
44    /// `commit = Some` reads from that specific git commit; `commit = None`
45    /// reads the working tree and resolves HEAD for the reported commit.
46    /// Returns `Ok(None)` if the repository or file does not exist.
47    pub async fn read_json<T: DeserializeOwned>(
48        &self,
49        kind: Kind,
50        owner: &str,
51        repository: &str,
52        commit: Option<&str>,
53    ) -> Result<Option<(T, String)>, Error> {
54        let repo_path = self.repo_path(kind, owner, repository);
55        let file_name = kind.filename();
56
57        let (content, resolved) = match commit {
58            Some(sha) => match read_file_at_commit(&repo_path, file_name, sha) {
59                Ok(content) => (content, sha.to_string()),
60                Err(e) if is_not_found(&e) => return Ok(None),
61                Err(e) => return Err(e),
62            },
63            None => {
64                let file_path = repo_path.join(file_name);
65                match tokio::fs::read_to_string(&file_path).await {
66                    Ok(content) => {
67                        let resolved = head_sha(&repo_path)
68                            .unwrap_or_else(|_| "HEAD".to_string());
69                        (content, resolved)
70                    }
71                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
72                        return Ok(None);
73                    }
74                    Err(e) => return Err(Error::Read(file_path, e)),
75                }
76            }
77        };
78
79        let value = serde_json::from_str::<T>(&content)
80            .map_err(|e| Error::Parse(repo_path.join(file_name), e))?;
81        Ok(Some((value, resolved)))
82    }
83
84    /// Streams every valid repository of `kind` under `state_dir/<kind>`.
85    ///
86    /// The outer stream lists the `<owner>` directories; each owner is then
87    /// flat-mapped into an inner stream listing its `<repository>`
88    /// directories, and each repo resolves to a `Client` remote path by
89    /// reading HEAD and parsing the kind's JSON. Repos whose HEAD or JSON
90    /// fails are skipped. Nothing is buffered into a `Vec`: directories are
91    /// walked lazily as the stream is polled.
92    pub async fn list(
93        &self,
94        kind: Kind,
95    ) -> Pin<Box<dyn Stream<Item = RemotePath> + Send>> {
96        let kind_dir = self.state_dir().join(kind.as_str());
97        Box::pin(
98            // Outer stream: the `<owner>` directories under `<kind>`.
99            read_dir_stream(kind_dir)
100                .filter_map(|owner| async move { is_dir(&owner).await.then_some(owner) })
101                // Each owner becomes its own inner stream of repositories.
102                .flat_map(move |owner| {
103                    let owner_name = owner.file_name().to_string_lossy().into_owned();
104                    // Inner stream: the `<repository>` directories of this owner.
105                    read_dir_stream(owner.path())
106                        .filter_map(|repo| async move {
107                            is_dir(&repo).await.then_some(repo)
108                        })
109                        .map(move |repo| {
110                            let owner = owner_name.clone();
111                            let repository =
112                                repo.file_name().to_string_lossy().into_owned();
113                            resolve_entry(kind, owner, repository, repo.path())
114                        })
115                        .buffer_unordered(16)
116                        .filter_map(|opt| async move { opt })
117                }),
118        )
119    }
120}
121
122/// Whether a directory entry is itself a directory (false on any IO error).
123async fn is_dir(entry: &tokio::fs::DirEntry) -> bool {
124    entry.file_type().await.map(|t| t.is_dir()).unwrap_or(false)
125}
126
127/// Lazily streams the entries of `dir`. The directory is opened on first
128/// poll; if it can't be opened the stream is simply empty.
129fn read_dir_stream(
130    dir: PathBuf,
131) -> impl Stream<Item = tokio::fs::DirEntry> + Send {
132    enum State {
133        Unopened(PathBuf),
134        Open(tokio::fs::ReadDir),
135    }
136    futures::stream::unfold(State::Unopened(dir), |state| async move {
137        let mut read_dir = match state {
138            State::Unopened(dir) => tokio::fs::read_dir(&dir).await.ok()?,
139            State::Open(read_dir) => read_dir,
140        };
141        match read_dir.next_entry().await {
142            Ok(Some(entry)) => Some((entry, State::Open(read_dir))),
143            _ => None,
144        }
145    })
146}
147
148/// Resolves one candidate repo to a `Client` remote path, or `None` if
149/// its HEAD can't be resolved or its JSON doesn't parse.
150async fn resolve_entry(
151    kind: Kind,
152    owner: String,
153    repository: String,
154    repo_path: PathBuf,
155) -> Option<RemotePath> {
156    // git HEAD resolution is blocking — run it off the async runtime.
157    let head_path = repo_path.clone();
158    let commit = tokio::task::spawn_blocking(move || head_sha(&head_path))
159        .await
160        .ok()?
161        .ok()?;
162    // Validate the JSON parses as the kind's definition.
163    let content =
164        tokio::fs::read_to_string(repo_path.join(kind.filename())).await.ok()?;
165    if !validates(kind, &content) {
166        return None;
167    }
168    Some(RemotePath::Client { owner, repository, commit })
169}
170
171/// Resolves the HEAD commit SHA for a repository directory (blocking git).
172fn head_sha(repo_path: &Path) -> Result<String, git2::Error> {
173    let repo = git2::Repository::open(repo_path)?;
174    let head = repo.head()?;
175    let commit = head.peel_to_commit()?;
176    Ok(commit.id().to_string())
177}
178
179/// Whether `content` parses as the kind's base/full definition.
180fn validates(kind: Kind, content: &str) -> bool {
181    match kind {
182        Kind::Agents => serde_json::from_str::<
183            objectiveai_sdk::agent::RemoteAgentBaseWithFallbacks,
184        >(content)
185        .is_ok(),
186        Kind::Swarms => {
187            serde_json::from_str::<objectiveai_sdk::swarm::RemoteSwarmBase>(
188                content,
189            )
190            .is_ok()
191        }
192        Kind::Functions => serde_json::from_str::<
193            objectiveai_sdk::functions::FullRemoteFunction,
194        >(content)
195        .is_ok(),
196        Kind::Profiles => serde_json::from_str::<
197            objectiveai_sdk::functions::RemoteProfile,
198        >(content)
199        .is_ok(),
200    }
201}
202
203/// Returns true if the git error represents a "not found" condition.
204fn is_not_found(e: &Error) -> bool {
205    match e {
206        Error::Git(e) => {
207            e.code() == git2::ErrorCode::NotFound
208                || e.class() == git2::ErrorClass::Object
209                || e.class() == git2::ErrorClass::Reference
210        }
211        _ => false,
212    }
213}
214
215/// Reads a file from a git repository at a specific commit (blocking git).
216fn read_file_at_commit(
217    repo_path: &Path,
218    file_name: &str,
219    commit_sha: &str,
220) -> Result<String, Error> {
221    let repo = git2::Repository::open(repo_path)?;
222    let oid = git2::Oid::from_str(commit_sha)?;
223    let commit = repo.find_commit(oid)?;
224    let tree = commit.tree()?;
225    let entry = tree.get_name(file_name).ok_or_else(|| {
226        git2::Error::from_str(&format!(
227            "{} not found at commit {}",
228            file_name, commit_sha
229        ))
230    })?;
231    let blob = repo.find_blob(entry.id())?;
232    String::from_utf8(blob.content().to_vec())
233        .map_err(|e| Error::Utf8(repo_path.join(file_name), e))
234}
objectiveai_cli/filesystem/read.rs

objectiveai_cli/filesystem/
read.rs
