diff --git a/packages/agent/src/agent_catalog/context_assembly.rs b/packages/agent/src/agent_catalog/context_assembly.rs deleted file mode 100644 index 6f0a8e8a7..000000000 --- a/packages/agent/src/agent_catalog/context_assembly.rs +++ /dev/null @@ -1,973 +0,0 @@ -//! Build the knowledge-graph context file consumed by a PTY-spawned agent (ADR-032). -//! -//! [`GraphContextAssembler`] gathers nodes from the knowledge graph and writes -//! the result to the agent's working directory as `CLAUDE.md` (Claude Code) or -//! `AGENTS.md` (every other supported agent). `write_context_file` also copies -//! `SKILL.md` (the CLI-vocabulary companion doc, embedded at compile time from -//! `packages/skill/SKILL.md`) into the same directory — it, not the knowledge-graph -//! context file, carries all tool/capability guidance for the PTY-spawned agent, -//! since the agent drives NodeSpace exclusively through the `nodespace` CLI. - -use crate::agent_catalog::registry::SystemAgentRegistry; -use crate::agent_types::{AgentType, ContextError, ContextFile}; -use nodespace_core::models::Node; -use nodespace_core::ops::search_ops::{ - normalize_enumerate_query, search_semantic, SearchSemanticInput, -}; -use nodespace_core::services::NodeEmbeddingService; -use nodespace_core::NodeService; -use std::collections::{HashMap, HashSet}; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use tokio::sync::RwLock; - -/// `packages/skill/SKILL.md`, embedded at compile time so the PTY session always -/// gets exactly the companion doc that shipped with this daemon build — no -/// filesystem lookup, no drift between dev and installed builds. -const SKILL_MD: &str = include_str!("../../../skill/SKILL.md"); - -/// `packages/skill/references/cli.md`, embedded for the same reason as -/// [`SKILL_MD`]. -/// -/// SKILL.md links to this by relative path — it is the on-demand tier holding -/// the full CLI reference, kept out of the body so the body stays within the -/// Agent Skills size recommendation. Embedding it too means a PTY session gets -/// the whole skill rather than a body pointing at a file that was never -/// written next to it. -const SKILL_CLI_REFERENCE: &str = include_str!("../../../skill/references/cli.md"); - -/// `packages/skill/references/graph-authored-guidance.md`, embedded for the -/// same reason as [`SKILL_MD`]. -/// -/// SKILL.md links to this by relative path before its first `skill guidance` -/// call — it covers the trust boundary for graph-fetched procedural -/// guidance. Embedding it too means a PTY session gets the whole skill -/// rather than a body pointing at a file that was never written next to it. -const SKILL_GRAPH_AUTHORED_GUIDANCE: &str = - include_str!("../../../skill/references/graph-authored-guidance.md"); - -/// Default token budget when none is specified by the caller. -const DEFAULT_TOKEN_BUDGET: u32 = 50_000; - -/// ~4 chars/token heuristic — verified within 10% of real tokenizers for -/// typical English markdown content. -const CHARS_PER_TOKEN: u32 = 4; - -/// Maximum character length for selected (seed) nodes. -const SEED_NODE_CONTENT_LIMIT: usize = 2000; - -/// Maximum character length for semantic-neighbor nodes. -const NEIGHBOR_CONTENT_LIMIT: usize = 500; - -/// Semantic neighbours fetched per seed node. -const NEIGHBORS_PER_SEED: usize = 5; - -/// Similarity threshold for semantic search. -const SEMANTIC_THRESHOLD: f32 = 0.3; - -/// Header explaining what the context file is and how to read it. -const CONTEXT_HEADER: &str = "\ -# NodeSpace Knowledge Graph Context - -This file is auto-generated by NodeSpace for each agent session. It captures -the nodes you (the agent) have been given and the immediate graph context -around them. Treat the sections below as the authoritative starting point. -"; - -/// Look up the context-file convention for `agent_type` in [`AGENT_CATALOG`]. -/// -/// Falls back to [`ContextFile::AgentsMd`] only if `agent_type` is missing -/// from the catalog — which `AgentType` being a closed enum makes impossible -/// today, but the explicit default keeps `write_context_file` infallible if -/// the catalog and enum ever drift. -fn context_file_for(agent_type: AgentType) -> ContextFile { - SystemAgentRegistry::new() - .get(agent_type) - .map(|d| d.context_file) - .unwrap_or(ContextFile::AgentsMd) -} - -fn estimate_tokens(text: &str) -> u32 { - (text.len() as u32).div_ceil(CHARS_PER_TOKEN) -} - -fn truncate_content(content: &str, max_chars: usize) -> String { - if content.len() <= max_chars { - content.to_string() - } else { - let cut = content.floor_char_boundary(max_chars.saturating_sub(3)); - format!("{}...", &content[..cut]) - } -} - -fn node_title(node: &Node) -> String { - if let Some(ref title) = node.title { - if !title.is_empty() { - return title.clone(); - } - } - let first_line = node.content.lines().next().unwrap_or("(untitled)"); - let stripped = first_line - .trim_start_matches('#') - .trim_start_matches(' ') - .trim(); - if stripped.is_empty() { - "(untitled)".to_string() - } else { - truncate_content(stripped, 80) - } -} - -fn format_properties(properties: &serde_json::Value) -> Option { - let obj = properties.as_object()?; - if obj.is_empty() { - return None; - } - - let mut pairs = Vec::new(); - for (key, value) in obj { - if value.is_null() { - continue; - } - if let Some(inner_obj) = value.as_object() { - for (inner_key, inner_val) in inner_obj { - if !inner_val.is_null() { - pairs.push(format!( - "{}.{}: {}", - key, - inner_key, - format_json_value(inner_val) - )); - } - } - } else { - pairs.push(format!("{}: {}", key, format_json_value(value))); - } - } - - if pairs.is_empty() { - None - } else { - Some(pairs.join(", ")) - } -} - -fn format_json_value(value: &serde_json::Value) -> String { - match value { - serde_json::Value::String(s) => s.clone(), - serde_json::Value::Bool(b) => b.to_string(), - serde_json::Value::Number(n) => n.to_string(), - serde_json::Value::Array(arr) => { - let items: Vec = arr.iter().map(format_json_value).collect(); - format!("[{}]", items.join(", ")) - } - serde_json::Value::Null => "null".to_string(), - other => other.to_string(), - } -} - -#[derive(Debug, Clone)] -struct CollectedRelationship { - from_title: String, - relationship_type: String, - to_title: String, -} - -/// Assembles per-session knowledge-graph context and writes it to disk. -/// -/// Service references are injected so this crate stays decoupled from Tauri's -/// `AppServices`. The desktop-app layer resolves services and constructs the -/// assembler before each agent session starts. -pub struct GraphContextAssembler { - node_service: Arc, - /// Updated by the background embedding-load task once the model is ready. - embedding_service: Arc>>>, - seed_node_ids: Vec, - token_budget: u32, -} - -impl GraphContextAssembler { - pub fn new( - node_service: Arc, - embedding_service: Arc>>>, - ) -> Self { - Self { - node_service, - embedding_service, - seed_node_ids: Vec::new(), - token_budget: DEFAULT_TOKEN_BUDGET, - } - } - - /// Set the seed node IDs to include in the assembled context. - pub fn with_seed_nodes(mut self, node_ids: Vec) -> Self { - self.seed_node_ids = node_ids; - self - } - - /// Override the token budget used to gate context size. - pub fn with_token_budget(mut self, budget: u32) -> Self { - self.token_budget = if budget == 0 { - DEFAULT_TOKEN_BUDGET - } else { - budget - }; - self - } - - /// Build the context markdown without writing to disk. - pub async fn assemble_context(&self) -> Result { - // --- Step 1: fetch seed nodes --- - let mut seed_nodes: Vec = Vec::new(); - for id in &self.seed_node_ids { - match self.fetch_node(id).await { - Ok(node) => seed_nodes.push(node), - Err(ContextError::NodeNotFound(nid)) => { - tracing::warn!("Seed node not found, skipping: {}", nid); - } - Err(e) => return Err(e), - } - } - - // --- Step 2: semantic expansion (best-effort) --- - let seed_ids: HashSet = seed_nodes.iter().map(|n| n.id.clone()).collect(); - let embedding_guard = self.embedding_service.read().await; - let semantic_neighbors = match embedding_guard.as_ref() { - Some(embedding_service) => { - self.find_semantic_neighbors(embedding_service, &seed_nodes, &seed_ids) - .await - } - None => { - tracing::info!("Embedding service unavailable, skipping semantic expansion"); - Vec::new() - } - }; - - // --- Step 3: title map for relationship labels --- - let mut node_titles: HashMap = HashMap::new(); - for node in &seed_nodes { - node_titles.insert(node.id.clone(), node_title(node)); - } - for (node, _) in &semantic_neighbors { - node_titles.insert(node.id.clone(), node_title(node)); - } - - // --- Step 4: 1-hop relationship traversal --- - let relationships = self.collect_relationships(&seed_nodes, &node_titles).await; - - // --- Step 5: render markdown within token budget --- - Ok(build_context_markdown( - &seed_nodes, - &semantic_neighbors, - &relationships, - self.token_budget, - )) - } - - /// Assemble the context and write it to `session_dir` as the appropriate - /// agent context file (`CLAUDE.md` for Claude Code, `AGENTS.md` otherwise), - /// and copy `SKILL.md` into the same directory so the PTY-spawned agent - /// picks it up as a project-local skill the same way it would in any - /// directory a user `cd`s into with `SKILL.md` present. - /// - /// Returns the absolute path of the written context file. - pub async fn write_context_file( - &self, - session_dir: &Path, - agent_type: AgentType, - ) -> Result { - let content = self.assemble_context().await?; - write_skill_md(session_dir).await?; - write_context_to_dir(session_dir, agent_type, &content).await - } - - async fn fetch_node(&self, node_id: &str) -> Result { - self.node_service - .get_node(node_id) - .await - .map_err(|e| ContextError::Other(e.into()))? - .ok_or_else(|| ContextError::NodeNotFound(node_id.to_string())) - } - - async fn find_semantic_neighbors( - &self, - embedding_service: &Arc, - seed_nodes: &[Node], - seed_ids: &HashSet, - ) -> Vec<(Node, f64)> { - let mut seen: HashSet = seed_ids.clone(); - let mut neighbors: Vec<(Node, f64)> = Vec::new(); - - for seed in seed_nodes { - let query = if seed.content.len() > 200 { - &seed.content[..seed.content.floor_char_boundary(200)] - } else { - &seed.content - }; - - // An empty or `*` query asks search to list nodes, not to find - // what is near this seed. - if normalize_enumerate_query(query).is_none() { - continue; - } - - match search_semantic( - &self.node_service, - embedding_service, - neighbor_search_input(query), - ) - .await - { - Ok(output) => { - // `nodes` and `matched_nodes` are the same results in the - // same order; the score is carried only on the former. - debug_assert_eq!(output.nodes.len(), output.matched_nodes.len()); - let scores = output - .nodes - .iter() - .map(|n| n["similarity"].as_f64().unwrap_or(0.0)); - for (node, score) in output.matched_nodes.into_iter().zip(scores) { - if !seen.contains(&node.id) { - seen.insert(node.id.clone()); - neighbors.push((node, score)); - } - } - } - Err(e) => { - tracing::warn!("Semantic search failed for seed node '{}': {}", seed.id, e); - } - } - } - - neighbors.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); - neighbors - } - - async fn collect_relationships( - &self, - nodes: &[Node], - node_titles: &HashMap, - ) -> Vec { - let mut relationships = Vec::new(); - - for node in nodes { - let from_title = node_titles - .get(&node.id) - .cloned() - .unwrap_or_else(|| node_title(node)); - - match self.node_service.get_children(&node.id).await { - Ok(children) => { - for child in &children { - let to_title = node_title(child); - relationships.push(CollectedRelationship { - from_title: from_title.clone(), - relationship_type: "has_child".to_string(), - to_title, - }); - } - } - Err(e) => { - tracing::warn!("Failed to get children for node '{}': {}", node.id, e); - } - } - - for mention_id in &node.mentions { - let to_title = node_titles - .get(mention_id) - .cloned() - .unwrap_or_else(|| mention_id.clone()); - relationships.push(CollectedRelationship { - from_title: from_title.clone(), - relationship_type: "mentions".to_string(), - to_title, - }); - } - } - - relationships - } -} - -/// The search that finds a seed node's semantic neighbours. -/// -/// The neighbours are the user's knowledge near the seed, so the search names -/// the `knowledge` scope: the user's documents, records and user-defined -/// types. Skills, tools and the built-in schemas are embedded in the same -/// index and are close to many seeds, but they are not context for the nodes -/// the session was given, and the session gets its tool guidance from -/// `SKILL.md`. -fn neighbor_search_input(query: &str) -> SearchSemanticInput { - SearchSemanticInput { - query: query.to_string(), - threshold: Some(SEMANTIC_THRESHOLD), - limit: Some(NEIGHBORS_PER_SEED), - collection_id: None, - collection: None, - exclude_collections: None, - include_markdown: Some(0), - include_archived: None, - scope: Some("knowledge".to_string()), - node_types: None, - property_filters: None, - include_edges: None, - graph_boost: None, - include_title_matches: None, - } -} - -/// Write the embedded skill — `SKILL.md` plus its `references/` tier — into -/// `session_dir`. -/// -/// All of it is written because SKILL.md links to each `references/*.md` file -/// by relative path; writing only the body would leave those links dangling -/// and the reference content unreachable from the session. -async fn write_skill_md(session_dir: &Path) -> Result<(), ContextError> { - tokio::fs::write(session_dir.join("SKILL.md"), SKILL_MD).await?; - let references = session_dir.join("references"); - tokio::fs::create_dir_all(&references).await?; - tokio::fs::write(references.join("cli.md"), SKILL_CLI_REFERENCE).await?; - tokio::fs::write( - references.join("graph-authored-guidance.md"), - SKILL_GRAPH_AUTHORED_GUIDANCE, - ) - .await?; - Ok(()) -} - -/// Write `content` to the agent's context file inside `session_dir`. Pure -/// file I/O — extracted from [`GraphContextAssembler::write_context_file`] -/// so the naming + write behaviour can be tested without a `NodeService`. -async fn write_context_to_dir( - session_dir: &Path, - agent_type: AgentType, - content: &str, -) -> Result { - let context_file = context_file_for(agent_type); - let path = session_dir.join(context_file.filename()); - tokio::fs::write(&path, content).await?; - Ok(path) -} - -fn format_node_section(node: &Node, content_limit: usize) -> String { - let title = node_title(node); - let truncated_content = truncate_content(&node.content, content_limit); - let mut section = format!( - "#### {} [{}] (id: {})\n{}\n", - title, node.node_type, node.id, truncated_content - ); - - if let Some(props) = format_properties(&node.properties) { - section.push_str(&format!("\nProperties: {}\n", props)); - } - - section -} - -fn build_context_markdown( - seed_nodes: &[Node], - semantic_neighbors: &[(Node, f64)], - relationships: &[CollectedRelationship], - token_budget: u32, -) -> String { - let mut sections: Vec = Vec::new(); - sections.push(CONTEXT_HEADER.to_string()); - - let header_tokens = estimate_tokens(CONTEXT_HEADER); - let mut remaining_budget = token_budget.saturating_sub(header_tokens); - - if !seed_nodes.is_empty() { - let mut selected_section = String::from("## Selected Nodes\n\n"); - - for node in seed_nodes { - let node_section = format_node_section(node, SEED_NODE_CONTENT_LIMIT); - let section_tokens = estimate_tokens(&node_section); - - if section_tokens > remaining_budget { - let reduced = format_node_section(node, 500); - let reduced_tokens = estimate_tokens(&reduced); - if reduced_tokens <= remaining_budget { - selected_section.push_str(&reduced); - remaining_budget = remaining_budget.saturating_sub(reduced_tokens); - } - } else { - selected_section.push_str(&node_section); - remaining_budget = remaining_budget.saturating_sub(section_tokens); - } - } - - let section_header_tokens = estimate_tokens("## Selected Nodes\n\n"); - remaining_budget = remaining_budget.saturating_sub(section_header_tokens); - sections.push(selected_section); - } - - if !relationships.is_empty() { - let mut rel_section = String::from("## Relationships\n"); - let rel_header_tokens = estimate_tokens("## Relationships\n"); - remaining_budget = remaining_budget.saturating_sub(rel_header_tokens); - - for rel in relationships { - let line = format!( - "- \"{}\" -> {} -> \"{}\"\n", - rel.from_title, rel.relationship_type, rel.to_title - ); - let line_tokens = estimate_tokens(&line); - if line_tokens > remaining_budget { - break; - } - rel_section.push_str(&line); - remaining_budget = remaining_budget.saturating_sub(line_tokens); - } - - sections.push(rel_section); - } - - if !semantic_neighbors.is_empty() { - let mut neighbor_section = String::from("## Related Context (auto-discovered)\n\n"); - let neighbor_header_tokens = estimate_tokens("## Related Context (auto-discovered)\n\n"); - remaining_budget = remaining_budget.saturating_sub(neighbor_header_tokens); - - for (node, _score) in semantic_neighbors { - let node_section = format_node_section(node, NEIGHBOR_CONTENT_LIMIT); - let section_tokens = estimate_tokens(&node_section); - - if section_tokens > remaining_budget { - break; - } - - neighbor_section.push_str(&node_section); - remaining_budget = remaining_budget.saturating_sub(section_tokens); - } - - sections.push(neighbor_section); - } - - sections.join("\n") -} - -#[cfg(test)] -mod tests { - use super::*; - - // ---- helpers ----------------------------------------------------------- - - fn make_test_node(id: &str, node_type: &str, content: &str, title: Option<&str>) -> Node { - Node { - id: id.to_string(), - node_type: node_type.to_string(), - content: content.to_string(), - version: 1, - created_at: chrono::Utc::now(), - modified_at: chrono::Utc::now(), - properties: serde_json::json!({}), - mentions: vec![], - mentioned_in: vec![], - title: title.map(|s| s.to_string()), - lifecycle_status: "active".to_string(), - } - } - - // ---- agent_type -> context_file mapping -------------------------------- - - #[test] - fn claude_code_maps_to_claude_md() { - assert_eq!( - context_file_for(AgentType::ClaudeCode), - ContextFile::ClaudeMd - ); - assert_eq!(ContextFile::ClaudeMd.filename(), "CLAUDE.md"); - } - - #[test] - fn non_claude_agents_map_to_agents_md() { - for agent_type in [ - AgentType::Codex, - AgentType::AntigravityCli, - AgentType::Pi, - AgentType::OpenCode, - ] { - assert_eq!(context_file_for(agent_type), ContextFile::AgentsMd); - } - assert_eq!(ContextFile::AgentsMd.filename(), "AGENTS.md"); - } - - // ---- estimate_tokens / truncate_content -------------------------------- - - #[test] - fn estimate_tokens_rounds_up() { - assert_eq!(estimate_tokens(""), 0); - assert_eq!(estimate_tokens("12345678"), 2); // 8/4 - assert_eq!(estimate_tokens("Hello"), 2); // ceil(5/4) - } - - #[test] - fn truncate_content_short_passthrough() { - assert_eq!(truncate_content("Hello", 10), "Hello"); - } - - #[test] - fn truncate_content_long_truncates() { - let result = truncate_content("Hello World, this is a long string", 15); - assert!(result.ends_with("...")); - } - - // ---- node_title -------------------------------------------------------- - - #[test] - fn node_title_uses_title_field_when_present() { - let node = make_test_node("1", "text", "# heading", Some("My Title")); - assert_eq!(node_title(&node), "My Title"); - } - - #[test] - fn node_title_falls_back_to_first_line() { - let node = make_test_node("1", "text", "# My Heading\nBody", None); - assert_eq!(node_title(&node), "My Heading"); - } - - #[test] - fn node_title_empty_content_is_untitled() { - let node = make_test_node("1", "text", "", None); - assert_eq!(node_title(&node), "(untitled)"); - } - - // ---- build_context_markdown ------------------------------------------- - - #[test] - fn empty_context_still_includes_header() { - let markdown = build_context_markdown(&[], &[], &[], 50_000); - assert!(markdown.contains("# NodeSpace Knowledge Graph Context")); - } - - #[test] - fn context_includes_seed_nodes() { - let nodes = vec![ - make_test_node("node-1", "text", "First node content", Some("First Node")), - make_test_node("node-2", "task", "Second node content", Some("Second Node")), - ]; - let markdown = build_context_markdown(&nodes, &[], &[], 50_000); - - assert!(markdown.contains("## Selected Nodes")); - assert!(markdown.contains("First Node")); - assert!(markdown.contains("[text]")); - assert!(markdown.contains("Second Node")); - assert!(markdown.contains("[task]")); - } - - #[test] - fn context_includes_relationships() { - let nodes = vec![make_test_node("n1", "text", "Content", Some("Node One"))]; - let relationships = vec![CollectedRelationship { - from_title: "Node One".to_string(), - relationship_type: "mentions".to_string(), - to_title: "Node Two".to_string(), - }]; - - let markdown = build_context_markdown(&nodes, &[], &relationships, 50_000); - assert!(markdown.contains("## Relationships")); - assert!(markdown.contains("\"Node One\" -> mentions -> \"Node Two\"")); - } - - #[test] - fn context_includes_semantic_neighbors() { - let seeds = vec![make_test_node("s1", "text", "Seed content", Some("Seed"))]; - let neighbors = vec![( - make_test_node("n1", "text", "Neighbor content", Some("Neighbor")), - 0.85, - )]; - - let markdown = build_context_markdown(&seeds, &neighbors, &[], 50_000); - assert!(markdown.contains("## Related Context (auto-discovered)")); - assert!(markdown.contains("Neighbor")); - } - - #[test] - fn budget_enforcement_drops_neighbors_first() { - let seeds = vec![make_test_node( - "s1", - "text", - &"Seed content. ".repeat(50), - Some("Important Seed"), - )]; - let neighbors: Vec<(Node, f64)> = (0..20) - .map(|i| { - ( - make_test_node( - &format!("n{}", i), - "text", - &"Neighbor content. ".repeat(50), - Some(&format!("Neighbor {}", i)), - ), - 0.8 - (i as f64 * 0.01), - ) - }) - .collect(); - - let markdown = build_context_markdown(&seeds, &neighbors, &[], 5_000); - - // Seed should always be present - assert!(markdown.contains("Important Seed")); - // Budget should not be wildly exceeded (10% tolerance for heuristic). - let actual = estimate_tokens(&markdown); - assert!(actual <= 5_500, "{} tokens exceeds 5000+10% budget", actual); - } - - #[test] - fn very_large_node_body_is_truncated() { - let huge = "X".repeat(100_000); - let node = make_test_node("huge", "text", &huge, Some("Huge")); - let markdown = build_context_markdown(&[node], &[], &[], 50_000); - assert!(markdown.len() < 100_000); - assert!(markdown.contains("Huge")); - } - - #[test] - fn format_node_section_includes_properties() { - let mut node = make_test_node("fmt1", "text", "Body text", Some("Title")); - node.properties = serde_json::json!({"task": {"status": "done"}}); - let section = format_node_section(&node, 2000); - - assert!(section.contains("#### Title [text] (id: fmt1)")); - assert!(section.contains("Body text")); - assert!(section.contains("Properties:")); - assert!(section.contains("task.status: done")); - } - - // ---- write_context_to_dir (the body of write_context_file) ------------ - - #[tokio::test] - async fn write_context_to_dir_writes_claude_md_for_claude_code() { - let temp = tempfile::tempdir().unwrap(); - let path = write_context_to_dir(temp.path(), AgentType::ClaudeCode, "hello") - .await - .unwrap(); - - assert_eq!(path, temp.path().join("CLAUDE.md")); - assert_eq!(tokio::fs::read_to_string(&path).await.unwrap(), "hello"); - } - - #[tokio::test] - async fn write_context_to_dir_writes_agents_md_for_other_agents() { - for agent_type in [ - AgentType::Codex, - AgentType::AntigravityCli, - AgentType::Pi, - AgentType::OpenCode, - ] { - let temp = tempfile::tempdir().unwrap(); - let path = write_context_to_dir(temp.path(), agent_type, "hello") - .await - .unwrap(); - - assert_eq!( - path, - temp.path().join("AGENTS.md"), - "{:?} should write AGENTS.md", - agent_type - ); - assert_eq!(tokio::fs::read_to_string(&path).await.unwrap(), "hello"); - } - } - - #[tokio::test] - async fn write_context_to_dir_returns_err_when_dir_missing() { - let temp = tempfile::tempdir().unwrap(); - let missing = temp.path().join("does-not-exist"); - let err = write_context_to_dir(&missing, AgentType::ClaudeCode, "hello") - .await - .unwrap_err(); - - assert!(matches!(err, ContextError::WriteFailed(_))); - } - - // ---- write_skill_md ----------------------------------------------------- - - #[test] - fn embedded_skill_md_matches_source_file() { - // Sanity check that include_str! is resolving the sibling package's - // SKILL.md and not an empty/stale copy. - assert!(SKILL_MD.contains("NodeSpace")); - } - - #[test] - fn embedded_graph_authored_guidance_matches_source_file() { - assert!(SKILL_GRAPH_AUTHORED_GUIDANCE.contains("graph-fetched")); - } - - #[tokio::test] - async fn write_skill_md_writes_embedded_content() { - let session_dir = tempfile::tempdir().unwrap(); - write_skill_md(session_dir.path()).await.unwrap(); - - let written = tokio::fs::read_to_string(session_dir.path().join("SKILL.md")) - .await - .unwrap(); - assert_eq!(written, SKILL_MD); - } - - /// SKILL.md links to each `references/*.md` file by relative path, so the - /// session dir must carry all of them too — otherwise the body points at - /// a file that was never written and its content is silently - /// unreachable. Scans `packages/skill/references/` directly (rather than - /// hardcoding filenames) so a future reference file added there without - /// a matching embed/write in `write_skill_md` fails this test instead of - /// silently reproducing this exact gap. - #[tokio::test] - async fn write_skill_md_also_writes_the_references_tier() { - let session_dir = tempfile::tempdir().unwrap(); - write_skill_md(session_dir.path()).await.unwrap(); - - let body = tokio::fs::read_to_string(session_dir.path().join("SKILL.md")) - .await - .unwrap(); - - let references_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../skill/references"); - let mut checked = 0; - for entry in std::fs::read_dir(&references_dir).unwrap() { - let path = entry.unwrap().path(); - if path.extension().is_none_or(|ext| ext != "md") { - continue; - } - let name = path.file_name().unwrap().to_string_lossy().to_string(); - assert!( - body.contains(&format!("references/{name}")), - "SKILL.md no longer links to references/{name} — if the reference \ - layout changed, update this test to match" - ); - - let expected = std::fs::read_to_string(&path).unwrap(); - let written = - tokio::fs::read_to_string(session_dir.path().join("references").join(&name)) - .await - .unwrap_or_else(|_| { - panic!("references/{name} was not written next to SKILL.md") - }); - assert_eq!(written, expected, "references/{name} was written stale"); - checked += 1; - } - assert!(checked > 0, "no reference files found to check"); - } - - #[tokio::test] - async fn write_skill_md_returns_err_when_dir_missing() { - let temp = tempfile::tempdir().unwrap(); - let missing = temp.path().join("does-not-exist"); - - let err = write_skill_md(&missing).await.unwrap_err(); - assert!(matches!(err, ContextError::WriteFailed(_))); - } - - /// Neighbour expansion names its scope and no type list. Naming the scope - /// states what the search is for and keeps its filter on: a search that - /// names types and leaves the scope out replaces the scope with the type - /// list. - #[test] - fn neighbor_search_is_scoped_to_knowledge() { - let input = neighbor_search_input("Hall Nine booking"); - - assert_eq!(input.query, "Hall Nine booking"); - assert_eq!(input.scope.as_deref(), Some("knowledge")); - assert!(input.node_types.is_none()); - assert_eq!(input.limit, Some(NEIGHBORS_PER_SEED)); - assert_eq!(input.threshold, Some(SEMANTIC_THRESHOLD)); - assert_eq!(input.include_markdown, Some(0)); - } - - /// The scope the neighbour search names is one search accepts, and it - /// keeps the user's notes and drops the built-in schemas. Run as a - /// listing (`*`), which applies the same scope filter as a query and - /// needs no embedding model; the ranking of a real query is pinned by - /// `live_context_assembly_scope`. - #[tokio::test] - async fn neighbor_search_scope_keeps_notes_and_drops_built_in_schemas() { - use nodespace_core::db::SqliteStore; - use nodespace_core::services::NodeAccessor; - use nodespace_nlp_engine::{EmbeddingConfig, EmbeddingService}; - - let db_tmp = tempfile::TempDir::new().unwrap(); - let mut store = Arc::new( - SqliteStore::new(db_tmp.path().join("scope-test.db")) - .await - .unwrap(), - ); - let node_service = Arc::new(NodeService::new(&mut store).await.unwrap()); - let node_accessor: Arc = node_service.clone(); - let embedding_service = Arc::new(NodeEmbeddingService::new( - Arc::new(EmbeddingService::new(EmbeddingConfig::default()).unwrap()), - store.clone(), - node_accessor, - node_service.behaviors().clone(), - )); - let note = Node::new( - "text".to_string(), - "Hall Nine booking for the offsite".to_string(), - serde_json::json!({}), - ); - node_service.create_node(note.clone()).await.unwrap(); - - let listing = |scope: Option| SearchSemanticInput { - query: "*".to_string(), - limit: Some(1000), - scope, - ..neighbor_search_input("") - }; - let ids = |output: nodespace_core::ops::search_ops::SearchSemanticOutput| { - output - .matched_nodes - .into_iter() - .map(|n| n.id) - .collect::>() - }; - - let everything = ids(search_semantic( - &node_service, - &embedding_service, - listing(Some("everything".to_string())), - ) - .await - .unwrap()); - assert!( - everything.contains("checkbox"), - "the built-in checkbox schema is there to be found" - ); - - let scoped = ids(search_semantic( - &node_service, - &embedding_service, - listing(neighbor_search_input("").scope), - ) - .await - .unwrap()); - assert!(scoped.contains(¬e.id), "the user's note is in scope"); - assert!( - !scoped.contains("checkbox"), - "a built-in schema is outside the neighbour search's scope" - ); - } - - #[tokio::test] - async fn write_context_file_writes_both_context_file_and_skill_md() { - use nodespace_core::db::SqliteStore; - - let db_tmp = tempfile::TempDir::new().unwrap(); - let db_path = db_tmp.path().join("context-assembly-test.db"); - let mut store = Arc::new(SqliteStore::new(db_path).await.unwrap()); - let node_service = Arc::new(NodeService::new(&mut store).await.unwrap()); - let embedding_service: Arc>>> = - Arc::new(RwLock::new(None)); - let assembler = GraphContextAssembler::new(node_service, embedding_service); - - let session_dir = tempfile::tempdir().unwrap(); - let path = assembler - .write_context_file(session_dir.path(), AgentType::ClaudeCode) - .await - .unwrap(); - - assert_eq!(path, session_dir.path().join("CLAUDE.md")); - assert!(session_dir.path().join("CLAUDE.md").exists()); - let skill_md = tokio::fs::read_to_string(session_dir.path().join("SKILL.md")) - .await - .unwrap(); - assert_eq!(skill_md, SKILL_MD); - } -} diff --git a/packages/agent/src/agent_catalog/mod.rs b/packages/agent/src/agent_catalog/mod.rs index ef6ad734b..d10899055 100644 --- a/packages/agent/src/agent_catalog/mod.rs +++ b/packages/agent/src/agent_catalog/mod.rs @@ -1,2 +1 @@ -pub mod context_assembly; pub mod registry; diff --git a/packages/agent/src/agent_catalog/registry.rs b/packages/agent/src/agent_catalog/registry.rs index 458e247c1..a5f40ae87 100644 --- a/packages/agent/src/agent_catalog/registry.rs +++ b/packages/agent/src/agent_catalog/registry.rs @@ -1,12 +1,10 @@ //! PTY agent catalog (ADR-032). //! //! Hardcoded catalog of external agent CLIs that can be spawned in a PTY. -//! Each entry names the binary, the context file the agent expects to find -//! in its working directory (`CLAUDE.md` for Claude Code, `AGENTS.md` for -//! everything else), the flag used to resume a previous session, and where -//! the agent keeps the sessions that flag resumes. +//! Each entry names the binary and the environment variables the agent reads +//! its credentials and configuration from. -use crate::agent_types::{AgentType, ContextFile}; +use crate::agent_types::AgentType; /// Static description of an external agent CLI spawned via PTY. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -15,31 +13,14 @@ pub struct AgentDefinition { pub agent_type: AgentType, /// Human-readable display name. pub name: &'static str, - /// Binary name to spawn (resolved on `PATH`). + /// Binary name to spawn, resolved on the search path + /// [`crate::pty::detection`] builds. pub binary: &'static str, - /// Context file the agent reads on startup. - pub context_file: ContextFile, - /// CLI flag used to resume a prior session, if the agent supports one. - pub resume_flag: Option<&'static str>, - /// Where the agent records its sessions, for the agents whose resume flag - /// takes a session id. `None` when the agent has no id to resume by. - pub session_store: Option, - /// Auth-related environment variables this agent reads. PTY children get - /// a cleared environment plus a fixed base allowlist plus these — never - /// the daemon's full environment (see [`crate::pty::session`]). - pub auth_env_vars: &'static [&'static str], -} - -/// The on-disk layout an agent records its own sessions in, under the user's -/// home directory. [`crate::pty::find_harness_session_id`] reads it to learn -/// the id a finished session can be resumed by. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SessionStore { - /// `~/.claude/projects//.jsonl`. - ClaudeProjects, - /// `~/.codex/sessions///
/rollout-
/`). -const CODEX_ROLLOUT_DEPTH: usize = 3; - -/// Find the id `agent_type`'s harness recorded for the session that ran in -/// `session_dir`, looking in the harness's session store under `home`. -/// -/// Returns `None` for a harness with no session store, and when the store -/// holds no record of the session (the harness exited before starting a -/// conversation, say). When the harness recorded several conversations in the -/// directory (the user cleared or switched conversations mid-session), the -/// one written to last is returned: it is the one the session ended on. -/// -/// `started_at` is when the PTY session launched. Codex files every -/// conversation on the machine in one tree, so a rollout last written before -/// then cannot be this session's and is not opened. Claude Code's directory -/// for the working directory holds this session's conversations only, and -/// needs no such filter. -/// -/// Reads the filesystem synchronously: call it off the async executor. -pub fn find_harness_session_id( - agent_type: AgentType, - home: &Path, - session_dir: &Path, - started_at: DateTime, -) -> Option { - let store = SystemAgentRegistry::new().get(agent_type)?.session_store?; - let working_dirs = working_dir_spellings(session_dir); - match store { - SessionStore::ClaudeProjects => claude_session_id(home, &working_dirs), - SessionStore::CodexRollouts => codex_session_id(home, &working_dirs, started_at.into()), - } -} - -/// The ways a harness may have spelled the session's working directory: as -/// launched, and with symlinks resolved (what a process reads back as its -/// current directory). -fn working_dir_spellings(session_dir: &Path) -> Vec { - let mut spellings = vec![session_dir.to_path_buf()]; - if let Ok(resolved) = session_dir.canonicalize() { - if resolved != session_dir { - spellings.push(resolved); - } - } - spellings -} - -/// Claude Code keeps one directory per working directory, named after it, and -/// in it one `.jsonl` per conversation. -fn claude_session_id(home: &Path, working_dirs: &[PathBuf]) -> Option { - let projects = home.join(".claude").join("projects"); - working_dirs - .iter() - .flat_map(|dir| files_in(&projects.join(claude_project_dir_name(dir)))) - .filter(|path| path.extension().is_some_and(|ext| ext == "jsonl")) - .filter_map(|path| { - // A conversation's file is named by its id, a UUID. Anything else - // in the directory is not a conversation that can be resumed. - let id = path.file_stem()?.to_str()?.to_string(); - Uuid::parse_str(&id).ok()?; - Some((modified(&path)?, id)) - }) - .max_by_key(|(written, _)| *written) - .map(|(_, id)| id) -} - -/// The directory name Claude Code derives from a working directory: the path -/// with every character that is not an ASCII letter or digit replaced by `-`. -fn claude_project_dir_name(working_dir: &Path) -> String { - working_dir - .to_string_lossy() - .chars() - .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }) - .collect() -} - -/// Codex keeps one rollout file per conversation, filed by date. The first -/// line of a rollout records the conversation's id and working directory. -fn codex_session_id( - home: &Path, - working_dirs: &[PathBuf], - started_at: SystemTime, -) -> Option { - let mut rollouts = Vec::new(); - collect_rollouts( - &home.join(".codex").join("sessions"), - CODEX_ROLLOUT_DEPTH, - &mut rollouts, - ); - rollouts - .into_iter() - .filter_map(|path| { - let written = modified(&path)?; - if written < started_at { - return None; - } - let id = codex_rollout_session_id(&path, working_dirs)?; - Some((written, id)) - }) - .max_by_key(|(written, _)| *written) - .map(|(_, id)| id) -} - -/// Gather the `rollout-*.jsonl` files `depth` directories below `dir`. -fn collect_rollouts(dir: &Path, depth: usize, rollouts: &mut Vec) { - for path in files_in(dir) { - if depth > 0 { - if path.is_dir() { - collect_rollouts(&path, depth - 1, rollouts); - } - continue; - } - let is_rollout = path - .file_name() - .and_then(|name| name.to_str()) - .is_some_and(|name| name.starts_with("rollout-") && name.ends_with(".jsonl")); - if is_rollout { - rollouts.push(path); - } - } -} - -/// The session id a rollout file records, if it ran in one of `working_dirs`. -fn codex_rollout_session_id(rollout: &Path, working_dirs: &[PathBuf]) -> Option { - let mut first_line = String::new(); - BufReader::new(fs::File::open(rollout).ok()?) - .read_line(&mut first_line) - .ok()?; - let record: serde_json::Value = serde_json::from_str(&first_line).ok()?; - if record.get("type")?.as_str()? != "session_meta" { - return None; - } - let meta = record.get("payload")?; - let working_dir = Path::new(meta.get("cwd")?.as_str()?); - if !working_dirs.iter().any(|dir| dir == working_dir) { - return None; - } - Some(meta.get("id")?.as_str()?.to_string()) -} - -/// The entries of `dir`, or none when it cannot be read (it does not exist -/// when the harness has never run). -fn files_in(dir: &Path) -> Vec { - fs::read_dir(dir) - .map(|entries| entries.flatten().map(|entry| entry.path()).collect()) - .unwrap_or_default() -} - -fn modified(path: &Path) -> Option { - fs::metadata(path).ok()?.modified().ok() -} - -#[cfg(test)] -mod tests { - use super::*; - use std::time::Duration; - - const SESSION_A: &str = "0c5a8c1e-7d0b-4b5e-9f3a-2f1d6f0f8a01"; - const SESSION_B: &str = "7b1e2a44-3c9d-4f6a-8e21-5a0c9d3e7b02"; - - /// A home directory holding a PTY session's working directory, as the - /// daemon lays it out. - struct Fixture { - home: tempfile::TempDir, - session_dir: PathBuf, - started_at: DateTime, - } - - impl Fixture { - fn new() -> Self { - let home = tempfile::TempDir::new().unwrap(); - // Resolved up front, so the fixture's spelling of the directory is - // the one a harness would read back. - let root = home.path().canonicalize().unwrap(); - let session_dir = root - .join(".nodespace") - .join("agent-sessions") - .join("5f0e0c1a-1111-4222-8333-444455556666"); - fs::create_dir_all(&session_dir).unwrap(); - Self { - home, - session_dir, - started_at: Utc::now() - chrono::Duration::minutes(5), - } - } - - fn home(&self) -> &Path { - self.home.path() - } - - fn find(&self, agent_type: AgentType) -> Option { - find_harness_session_id(agent_type, self.home(), &self.session_dir, self.started_at) - } - - /// A conversation file in Claude Code's store for `working_dir`. - fn claude_conversation(&self, working_dir: &Path, file_name: &str) -> PathBuf { - let dir = self - .home() - .join(".claude") - .join("projects") - .join(claude_project_dir_name(working_dir)); - fs::create_dir_all(&dir).unwrap(); - let path = dir.join(file_name); - fs::write(&path, "{\"type\":\"mode\",\"mode\":\"normal\"}\n").unwrap(); - path - } - - /// A rollout file in Codex's store, recording `working_dir`. - fn codex_rollout(&self, id: &str, working_dir: &Path) -> PathBuf { - let dir = self - .home() - .join(".codex") - .join("sessions") - .join("2026") - .join("10") - .join("02"); - fs::create_dir_all(&dir).unwrap(); - let path = dir.join(format!("rollout-2026-10-02T09-15-00-{id}.jsonl")); - let meta = serde_json::json!({ - "timestamp": "2026-10-02T09:15:00.000Z", - "type": "session_meta", - "payload": { "id": id, "cwd": working_dir, "cli_version": "0.50.0" } - }); - fs::write(&path, format!("{meta}\n{{\"type\":\"response_item\"}}\n")).unwrap(); - path - } - } - - fn set_modified(path: &Path, when: SystemTime) { - fs::File::options() - .write(true) - .open(path) - .unwrap() - .set_modified(when) - .unwrap(); - } - - #[test] - fn claude_code_names_its_project_directory_after_the_working_directory() { - assert_eq!( - claude_project_dir_name(Path::new( - "/Users/sam/.nodespace/agent-sessions/5f0e0c1a-1111-4222-8333-444455556666" - )), - "-Users-sam--nodespace-agent-sessions-5f0e0c1a-1111-4222-8333-444455556666" - ); - } - - #[test] - fn a_claude_code_session_is_found_by_its_working_directory() { - let fixture = Fixture::new(); - fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_A}.jsonl")); - // Another session's conversation, in another working directory. - fixture.claude_conversation( - &fixture.session_dir.with_file_name("another-session"), - &format!("{SESSION_B}.jsonl"), - ); - - assert_eq!( - fixture.find(AgentType::ClaudeCode).as_deref(), - Some(SESSION_A) - ); - } - - #[test] - fn the_claude_code_conversation_written_to_last_is_the_one_returned() { - let fixture = Fixture::new(); - let first = - fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_A}.jsonl")); - let last = fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_B}.jsonl")); - let now = SystemTime::now(); - set_modified(&first, now - Duration::from_secs(120)); - set_modified(&last, now - Duration::from_secs(10)); - - assert_eq!( - fixture.find(AgentType::ClaudeCode).as_deref(), - Some(SESSION_B) - ); - } - - #[test] - fn only_a_claude_code_conversation_file_names_a_session() { - let fixture = Fixture::new(); - fixture.claude_conversation(&fixture.session_dir, "agent-a1b2c3.jsonl"); - fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_A}.txt")); - fs::create_dir( - fixture - .home() - .join(".claude") - .join("projects") - .join(claude_project_dir_name(&fixture.session_dir)) - .join(SESSION_B), - ) - .unwrap(); - - assert_eq!(fixture.find(AgentType::ClaudeCode), None); - } - - #[test] - fn a_codex_session_is_found_by_the_working_directory_its_rollout_records() { - let fixture = Fixture::new(); - fixture.codex_rollout(SESSION_A, &fixture.session_dir); - fixture.codex_rollout( - SESSION_B, - &fixture.session_dir.with_file_name("another-session"), - ); - - assert_eq!(fixture.find(AgentType::Codex).as_deref(), Some(SESSION_A)); - } - - #[test] - fn the_codex_rollout_written_to_last_is_the_one_returned() { - let fixture = Fixture::new(); - let first = fixture.codex_rollout(SESSION_A, &fixture.session_dir); - let last = fixture.codex_rollout(SESSION_B, &fixture.session_dir); - let now = SystemTime::now(); - set_modified(&first, now - Duration::from_secs(120)); - set_modified(&last, now - Duration::from_secs(10)); - - assert_eq!(fixture.find(AgentType::Codex).as_deref(), Some(SESSION_B)); - } - - #[test] - fn a_codex_rollout_last_written_before_the_session_started_is_not_read() { - let fixture = Fixture::new(); - let rollout = fixture.codex_rollout(SESSION_A, &fixture.session_dir); - set_modified( - &rollout, - SystemTime::from(fixture.started_at) - Duration::from_secs(60), - ); - - assert_eq!(fixture.find(AgentType::Codex), None); - } - - #[test] - fn a_codex_file_that_is_not_a_rollout_record_is_ignored() { - let fixture = Fixture::new(); - let rollout = fixture.codex_rollout(SESSION_A, &fixture.session_dir); - // Not a session record on its first line. - fs::write(&rollout, "{\"type\":\"response_item\"}\n").unwrap(); - // Not JSON at all. - fs::write( - rollout.with_file_name(format!("rollout-2026-10-02T09-20-00-{SESSION_B}.jsonl")), - "not json\n", - ) - .unwrap(); - // A session record, but not in a rollout file. - let meta = serde_json::json!({ - "type": "session_meta", - "payload": { "id": SESSION_B, "cwd": fixture.session_dir } - }); - fs::write(rollout.with_file_name("notes.jsonl"), format!("{meta}\n")).unwrap(); - - assert_eq!(fixture.find(AgentType::Codex), None); - } - - #[cfg(unix)] - #[test] - fn a_working_directory_reached_through_a_symlink_is_matched_resolved() { - let fixture = Fixture::new(); - fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_A}.jsonl")); - fixture.codex_rollout(SESSION_B, &fixture.session_dir); - let link = fixture.session_dir.with_file_name("linked-session"); - std::os::unix::fs::symlink(&fixture.session_dir, &link).unwrap(); - - let find = |agent_type| { - find_harness_session_id(agent_type, fixture.home(), &link, fixture.started_at) - }; - assert_eq!(find(AgentType::ClaudeCode).as_deref(), Some(SESSION_A)); - assert_eq!(find(AgentType::Codex).as_deref(), Some(SESSION_B)); - } - - #[test] - fn a_session_the_harness_never_recorded_has_no_id() { - let fixture = Fixture::new(); - assert_eq!(fixture.find(AgentType::ClaudeCode), None); - assert_eq!(fixture.find(AgentType::Codex), None); - } - - #[test] - fn a_harness_without_a_session_store_has_no_id() { - let fixture = Fixture::new(); - fixture.claude_conversation(&fixture.session_dir, &format!("{SESSION_A}.jsonl")); - fixture.codex_rollout(SESSION_B, &fixture.session_dir); - - for agent_type in [ - AgentType::AntigravityCli, - AgentType::Pi, - AgentType::OpenCode, - ] { - assert_eq!(fixture.find(agent_type), None, "{agent_type:?}"); - } - } -} diff --git a/packages/agent/src/pty/manager.rs b/packages/agent/src/pty/manager.rs index d4a94d843..4e4cb1206 100644 --- a/packages/agent/src/pty/manager.rs +++ b/packages/agent/src/pty/manager.rs @@ -8,16 +8,18 @@ //! the underlying agent process exits on its own. use std::collections::HashMap; -use std::sync::Arc; +use std::ffi::OsString; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, OnceLock}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use uuid::Uuid; -use crate::agent_catalog::context_assembly::GraphContextAssembler; use crate::agent_types::AgentType; -use crate::pty::session::PtySession; +use crate::pty::detection::{agent_search_path, detect_all_agents_on, AgentAvailability}; +use crate::pty::session::{PtySession, SessionLaunch}; /// Lightweight snapshot of a session suitable for listing in a UI / RPC. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -44,6 +46,13 @@ impl SessionMetadata { #[derive(Default, Clone)] pub struct PtySessionManager { sessions: Arc>>>, + /// The socket the daemon holding this manager serves on, once it is + /// serving. A launched session is told it, so the `nodespace` commands + /// run inside reach the daemon that launched them. + daemon_socket: Arc>, + /// The search path agent binaries are found on, when it is not the + /// machine's own ([`agent_search_path`]). + search_path: Arc>, } impl PtySessionManager { @@ -51,23 +60,66 @@ impl PtySessionManager { Self::default() } + /// Record the socket the daemon serves on. The first call stands. + pub fn set_daemon_socket(&self, socket: PathBuf) { + let _ = self.daemon_socket.set(socket); + } + + /// The socket the daemon serves on, when it has said. + pub fn daemon_socket(&self) -> Option<&Path> { + self.daemon_socket.get().map(PathBuf::as_path) + } + + /// Use `path` as the search path agent binaries are found on, in place of + /// the machine's own. The first call stands. + pub fn set_agent_search_path(&self, path: OsString) { + let _ = self.search_path.set(path); + } + + /// The search path agent binaries are found on. Detection and launch both + /// read it here, so a binary detection reports is the one launch starts. + /// + /// Building the machine's own spawns a process on macOS: call this off + /// the async executor. + pub fn agent_search_path(&self) -> OsString { + self.search_path + .get() + .cloned() + .unwrap_or_else(agent_search_path) + } + + /// Which agents are ready to launch, on [`Self::agent_search_path`]. Call + /// it off the async executor. + pub fn detect_agents(&self) -> Vec { + detect_all_agents_on(&self.agent_search_path()) + } + + /// [`Self::detect_agents`], with the search path it looked on: the one to + /// launch with. + pub fn detect_agents_with_path(&self) -> (Vec, OsString) { + let path = self.agent_search_path(); + (detect_all_agents_on(&path), path) + } + /// Launch a new agent session and register it. Returns the session id. /// - /// `node_id` is the `ai-chat-pty` node the session is a view onto, when - /// there is one. + /// `search_path` is the path the caller found the agent ready on + /// ([`Self::agent_search_path`]): the binary is resolved on that same + /// value, not on one built again. pub async fn launch( &self, - agent_type: AgentType, - initial_prompt: Option, - node_id: Option, - assembler: &GraphContextAssembler, + launch: SessionLaunch, + search_path: OsString, ) -> anyhow::Result { - let session = PtySession::launch(agent_type, initial_prompt, node_id, assembler).await?; + // Resolving the binary, creating the folder and forking all block. + let session = tokio::task::spawn_blocking(move || PtySession::launch(launch, search_path)) + .await + .map_err(|e| anyhow::anyhow!("launch task panicked: {e}"))??; Ok(self.insert(session).await) } /// Register an already-launched session. Split out so tests can build - /// sessions without a real [`GraphContextAssembler`]. + /// sessions without a real agent binary. pub async fn insert(&self, session: PtySession) -> Uuid { let id = session.id; let mut exit_rx = session.subscribe_exit(); @@ -79,11 +131,9 @@ impl PtySessionManager { } // Auto-cleanup: when the child process exits, remove the session entry - // from the map. The session directory at - // `~/.nodespace/agent-sessions//` is NOT deleted — it persists - // on disk so artifacts survive restarts. `watch` latches the final - // value, so this works whether the child exits before or after the - // receiver was constructed. + // from the map. Its working directory is left as it is. `watch` + // latches the final value, so this works whether the child exits + // before or after the receiver was constructed. let sessions = self.sessions.clone(); tokio::spawn(async move { // Fast path: exit already happened before insert returned. @@ -112,8 +162,8 @@ impl PtySessionManager { /// was not in the map (already cleaned up by the natural-exit watcher). /// /// Removing the session from the map drops the `Arc` when no - /// other caller holds a reference. The session directory on disk is NOT - /// deleted — it persists at `~/.nodespace/agent-sessions//`. + /// other caller holds a reference. The session's working directory is + /// left as it is. pub async fn terminate(&self, id: &Uuid) -> anyhow::Result { let session = { let mut sessions = self.sessions.lock().await; diff --git a/packages/agent/src/pty/mod.rs b/packages/agent/src/pty/mod.rs index a8cc8aecb..52cd4a0ea 100644 --- a/packages/agent/src/pty/mod.rs +++ b/packages/agent/src/pty/mod.rs @@ -6,25 +6,28 @@ //! //! The session lifecycle is: //! -//! 1. Create a persistent session directory at `~/.nodespace/agent-sessions//`. -//! 2. Write the context file ([`GraphContextAssembler::write_context_file`]) so the -//! agent picks up its `CLAUDE.md` / `AGENTS.md` on launch, and copy `SKILL.md` -//! into the same directory so the agent drives NodeSpace via the `nodespace` CLI. -//! 3. Spawn the agent binary inside a freshly opened PTY rooted at the session dir. -//! 4. Stream stdout/stderr bytes through a `broadcast::Sender`. -//! 5. Accept stdin via [`PtySession::write_input`] and resize via [`PtySession::resize`]. -//! 6. On [`PtySession::terminate`] (or when the child exits naturally), the session -//! directory is **not** deleted — artifacts survive across restarts. +//! 1. Take the working directory the launch names (a project's folder on this +//! machine, ADR-093 §8), or create a private one at +//! `~/.nodespace/agent-sessions//`. Nothing is written into it. +//! 2. Spawn the agent binary inside a freshly opened PTY rooted there, with an +//! allowlisted environment that names the database, the daemon socket, the +//! session and what it was launched for. +//! 3. Stream stdout/stderr bytes through a `broadcast::Sender`. +//! 4. Accept stdin via [`PtySession::write_input`] and resize via [`PtySession::resize`]. +//! 5. Take the harness's own session id when its plugin reports it +//! ([`PtySession::report_harness_session_id`]). pub mod capture; pub mod detection; -pub mod harness_session; pub mod manager; pub mod plain_text; pub mod session; pub use capture::SessionCapture; -pub use detection::{detect_all_agents, AgentAvailability}; -pub use harness_session::find_harness_session_id; +pub use detection::{ + agent_search_path, detect_all_agents, detect_all_agents_on, resolve_binary, AgentAvailability, +}; pub use manager::{PtySessionManager, SessionMetadata}; -pub use session::{ExitStatus, OutputChunk, PtySession}; +pub use session::{ + ExitStatus, OutputChunk, PtySession, SessionLaunch, LAUNCHED_FOR_ENV_VAR, SESSION_ENV_VAR, +}; diff --git a/packages/agent/src/pty/session.rs b/packages/agent/src/pty/session.rs index f7a58aa96..692821d13 100644 --- a/packages/agent/src/pty/session.rs +++ b/packages/agent/src/pty/session.rs @@ -2,8 +2,6 @@ //! //! A [`PtySession`] owns: //! -//! * a persistent per-session directory at `~/.nodespace/agent-sessions//` -//! (the agent's working directory — survives session end), //! * the master end of a PTY pair (`portable-pty`), //! * a writer into the PTY's stdin, //! * a [`portable_pty::ChildKiller`] handle for the spawned process, @@ -12,9 +10,14 @@ //! * a watcher task (which owns the actual `Child`) that broadcasts the //! process exit status through `broadcast::Sender`. //! -//! The session directory is **not** deleted on session end. It persists under -//! `~/.nodespace/agent-sessions//` so artifacts, context files, -//! and agent output survive across restarts. +//! The agent runs in the working directory the launch names: a project's +//! folder on this machine (ADR-093 §8). A launch that names none runs in a +//! private folder the launch says where to create (the daemon's +//! `agent-sessions//`), which is not deleted when the session +//! ends. Nothing is written into either at +//! launch: the session gets NodeSpace's context through its harness plugin +//! and the `nodespace` CLI, which the environment set here points at the +//! right database. //! //! ## Concurrency //! @@ -24,6 +27,7 @@ //! [`portable_pty::ChildKiller`], avoiding the obvious deadlock of a kill //! call waiting on a mutex the wait-loop already holds. +use std::ffi::OsString; use std::io::{Read, Write}; use std::path::PathBuf; use std::sync::Arc; @@ -35,10 +39,20 @@ use serde::{Deserialize, Serialize}; use tokio::sync::{broadcast, watch, Mutex}; use uuid::Uuid; -use crate::agent_catalog::context_assembly::GraphContextAssembler; use crate::agent_catalog::registry::SystemAgentRegistry; -use crate::agent_types::{AgentType, ContextError}; +use crate::agent_types::AgentType; use crate::pty::capture::SessionCapture; +use crate::pty::detection::resolve_binary; + +/// Names this session to the `nodespace` commands run inside it: the PTY +/// session's id. A harness plugin reports the harness's own session id +/// against it. +pub const SESSION_ENV_VAR: &str = "NODESPACE_SESSION"; + +/// Names what the session was launched for: the task when the launch named +/// one, otherwise the chat node the session is a view onto. A harness plugin +/// opens with that work's context. +pub const LAUNCHED_FOR_ENV_VAR: &str = "NODESPACE_LAUNCHED_FOR"; /// Number of buffered chunks per output subscriber. Slow consumers that fall /// behind by more than this many chunks will see `RecvError::Lagged`; that is @@ -59,13 +73,13 @@ const DEFAULT_PTY_COLS: u16 = 80; /// Environment variables every PTY child gets regardless of which agent is /// spawned — the minimum a POSIX CLI needs to resolve binaries, find the /// user's home, and render correctly in a terminal. Everything else is -/// dropped via `env_clear()`; auth vars are layered in per agent via -/// [`AgentDefinition::auth_env_vars`]. +/// dropped via `env_clear()`; an agent's own variables are layered in via +/// [`crate::agent_catalog::registry::AgentDefinition::env_vars`]. const BASE_ENV_ALLOWLIST: &[&str] = &["HOME", "PATH", "TERM", "SHELL", "LANG", "USER", "TMPDIR"]; /// Clear the command's inherited environment and repopulate it from /// `std::env::var` using `BASE_ENV_ALLOWLIST` plus `extra_vars` (typically an -/// agent's `auth_env_vars`). Only variables actually present in the daemon's +/// agent's `env_vars`). Only variables actually present in the daemon's /// environment are forwarded — a missing var is silently skipped rather than /// passed through empty. fn apply_env_allowlist(cmd: &mut CommandBuilder, extra_vars: &[&str]) { @@ -77,6 +91,42 @@ fn apply_env_allowlist(cmd: &mut CommandBuilder, extra_vars: &[&str]) { } } +/// What a launch asks for. +#[derive(Debug, Clone)] +pub struct SessionLaunch { + /// Which agent to start. + pub agent_type: AgentType, + /// Passed to the agent as its first argument, when given. + pub initial_prompt: Option, + /// The `ai-chat-pty` node the session is a view onto, when there is one. + pub node_id: Option, + /// The folder the agent runs in: the project's, on this machine. `None` + /// runs it in a private folder created under `session_folders`. + pub working_dir: Option, + /// Where a session with no working directory of its own gets a private + /// folder, named by its id. The caller decides it, so it follows wherever + /// the caller keeps NodeSpace's state. + pub session_folders: Option, + /// What NodeSpace sets in the session's environment on top of the + /// allowlist: the database, the daemon socket and what the session was + /// launched for. [`SESSION_ENV_VAR`] is added by the launch itself. + pub env: Vec<(String, String)>, +} + +impl SessionLaunch { + /// A launch of `agent_type` with nothing else named. + pub fn new(agent_type: AgentType) -> Self { + Self { + agent_type, + initial_prompt: None, + node_id: None, + working_dir: None, + session_folders: None, + env: Vec::new(), + } + } +} + /// One chunk of raw bytes read from the PTY master. /// /// The PTY merges stdout and stderr into a single byte stream — there is no @@ -138,101 +188,130 @@ pub struct PtySession { /// reader task so all chunks land here without extra subscriptions. capture: Arc>, - /// Persistent working directory for this session. Lives at - /// `~/.nodespace/agent-sessions//` and is never deleted on session - /// end — artifacts survive across restarts. - pub session_dir: PathBuf, + /// The harness's own id for the conversation, once its plugin has + /// reported one. The latest report stands: a harness that starts a new + /// conversation mid-session reports again. + harness_session_id: std::sync::Mutex>, + + /// The folder the agent runs in: the project's, or the session's private + /// folder. + pub working_dir: PathBuf, +} + +/// What [`PtySession::spawn_in_pty`] starts: a resolved binary, where it runs +/// and the environment it gets. +struct Spawn<'a> { + id: Uuid, + agent_type: AgentType, + binary_path: PathBuf, + args: Vec, + working_dir: PathBuf, + /// Variables forwarded from the daemon's environment when set there, on + /// top of the base allowlist. + forwarded_vars: &'a [&'a str], + /// The child's `PATH`, when it is not the daemon's own. + search_path: Option, + /// Variables set to the given values. + env: Vec<(String, String)>, } impl PtySession { - /// Spawn the agent binary for `agent_type` in a fresh PTY. + /// Spawn the agent binary a launch names in a fresh PTY. /// /// Steps, in order: /// - /// 1. Generate a session UUID and create a persistent directory at - /// `~/.nodespace/agent-sessions//`. - /// 2. Have `assembler` write the context file (`CLAUDE.md` / `AGENTS.md`) and - /// `SKILL.md` into the session directory. - /// 3. Resolve the agent binary on `PATH` via [`which::which`]. - /// 4. Open a PTY pair and spawn the binary with `cwd` set to the session dir. + /// 1. Generate a session UUID. + /// 2. Take the launch's working directory, or create the session's + /// private folder, `//`. + /// 3. Resolve the agent binary on `search_path`, the one detection used. + /// 4. Open a PTY pair and spawn the binary there, with the allowlisted + /// environment plus what the launch sets, and `search_path` as its + /// `PATH`. /// 5. Start the reader and exit-watcher tasks. - pub async fn launch( - agent_type: AgentType, - initial_prompt: Option, - node_id: Option, - assembler: &GraphContextAssembler, - ) -> anyhow::Result { + pub fn launch(launch: SessionLaunch, search_path: OsString) -> anyhow::Result { + let SessionLaunch { + agent_type, + initial_prompt, + node_id, + working_dir, + session_folders, + mut env, + } = launch; let session_id = Uuid::new_v4(); - let sessions_base = dirs::home_dir() - .context("HOME not set")? - .join(".nodespace") - .join("agent-sessions"); - let session_dir = sessions_base.join(session_id.to_string()); - std::fs::create_dir_all(&session_dir) - .with_context(|| format!("create session dir {}", session_dir.display()))?; - - assembler - .write_context_file(&session_dir, agent_type) - .await - .map_err(|e| match e { - ContextError::Other(err) => err, - other => anyhow::Error::new(other), - })?; let definition = SystemAgentRegistry::new() .get(agent_type) .ok_or_else(|| anyhow::anyhow!("agent {:?} missing from catalog", agent_type))?; - let binary_path = which::which(definition.binary).map_err(|e| { + let binary_path = resolve_binary(definition.binary, &search_path).ok_or_else(|| { anyhow::anyhow!( - "agent binary '{}' not found on PATH: {}", - definition.binary, - e + "agent binary '{}' not found on the search path", + definition.binary ) })?; - let mut session = Self::spawn_in_pty( - session_id, + let working_dir = match working_dir { + Some(dir) => dir, + None => { + let dir = session_folders + .context( + "the launch names neither a working directory nor where to create one", + )? + .join(session_id.to_string()); + std::fs::create_dir_all(&dir) + .with_context(|| format!("create session dir {}", dir.display()))?; + dir + } + }; + + env.push((SESSION_ENV_VAR.to_string(), session_id.to_string())); + + let mut session = Self::spawn_in_pty(Spawn { + id: session_id, agent_type, binary_path, - initial_prompt, - session_dir, - DEFAULT_PTY_ROWS, - DEFAULT_PTY_COLS, - definition.auth_env_vars, - )?; + args: initial_prompt.into_iter().collect(), + working_dir, + forwarded_vars: definition.env_vars, + search_path: Some(search_path), + env, + })?; session.node_id = node_id; Ok(session) } - /// Inner helper that opens the PTY, spawns the process, and wires up - /// reader / exit-watcher tasks. Split out from [`launch`](Self::launch) - /// so tests can construct sessions without going through - /// [`GraphContextAssembler`]. - #[allow(clippy::too_many_arguments)] - fn spawn_in_pty( - id: Uuid, - agent_type: AgentType, - binary_path: std::path::PathBuf, - initial_prompt: Option, - session_dir: PathBuf, - rows: u16, - cols: u16, - auth_env_vars: &[&str], - ) -> anyhow::Result { + /// Opens the PTY, spawns the process, and wires up the reader and + /// exit-watcher tasks. + fn spawn_in_pty(spawn: Spawn<'_>) -> anyhow::Result { + let Spawn { + id, + agent_type, + binary_path, + args, + working_dir, + forwarded_vars, + search_path, + env, + } = spawn; let pty_system = native_pty_system(); let pair = pty_system.openpty(PtySize { - rows, - cols, + rows: DEFAULT_PTY_ROWS, + cols: DEFAULT_PTY_COLS, pixel_width: 0, pixel_height: 0, })?; let mut cmd = CommandBuilder::new(binary_path); - cmd.cwd(&session_dir); - apply_env_allowlist(&mut cmd, auth_env_vars); - if let Some(prompt) = initial_prompt { - cmd.arg(prompt); + cmd.cwd(&working_dir); + apply_env_allowlist(&mut cmd, forwarded_vars); + if let Some(path) = search_path { + cmd.env("PATH", path); + } + for (key, value) in env { + cmd.env(key, value); + } + for arg in args { + cmd.arg(arg); } let child = pair.slave.spawn_command(cmd)?; @@ -268,10 +347,28 @@ impl PtySession { output_tx, exit_tx, capture, - session_dir, + harness_session_id: std::sync::Mutex::new(None), + working_dir, }) } + /// Record the id the harness gave its own conversation, as its plugin + /// reported it. A later report replaces an earlier one. + pub fn report_harness_session_id(&self, id: String) { + *self + .harness_session_id + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(id); + } + + /// The harness's own id for the conversation, when one was reported. + pub fn harness_session_id(&self) -> Option { + self.harness_session_id + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } + /// Subscribe to the PTY's output byte stream. /// /// Every subscriber sees the same stream from the point of subscription @@ -331,9 +428,8 @@ impl PtySession { /// /// Takes `&self` rather than consuming the session, because typical /// callers hold the session behind an `Arc` (the manager hands out - /// `Arc` from `get()`). The session directory at - /// `~/.nodespace/agent-sessions//` is NOT deleted on session end — - /// it persists so artifacts survive across daemon restarts. + /// `Arc` from `get()`). The working directory is left as it + /// is. /// /// Safe to call when the child has already exited: returns immediately /// without erroring. @@ -425,9 +521,9 @@ fn spawn_reader_task( #[cfg(any(test, feature = "testing"))] impl PtySession { - /// Test-only constructor: spawn an arbitrary binary in a PTY without - /// going through [`GraphContextAssembler`]. Lets tests use shell utilities - /// (`cat`, `sh -c '...'`) instead of depending on a real agent binary. + /// Test-only constructor: spawn an arbitrary binary in a PTY, so tests + /// can use shell utilities (`cat`, `sh -c '...'`) instead of depending on + /// a real agent binary. /// /// Creates a temporary directory under `std::env::temp_dir()` for the /// session. The directory persists until the OS cleans up the temp dir. @@ -436,67 +532,53 @@ impl PtySession { /// tests in sibling crates (e.g. `nodespace-daemon`) without being part /// of the production surface. pub fn launch_for_test(binary: &str, args: Vec) -> anyhow::Result { + Self::launch_for_test_as(binary, args, SessionLaunch::new(AgentType::ClaudeCode)) + } + + /// Test-only constructor: spawn an arbitrary binary the way + /// [`launch`](Self::launch) spawns an agent's, with `launch`'s working + /// directory, environment and node. The agent's own variables are + /// forwarded as they are for a real launch. + pub fn launch_for_test_as( + binary: &str, + args: Vec, + launch: SessionLaunch, + ) -> anyhow::Result { let id = Uuid::new_v4(); - let session_dir = std::env::temp_dir() - .join("nodespace-agent-test-sessions") - .join(id.to_string()); - std::fs::create_dir_all(&session_dir) - .with_context(|| format!("create test session dir {}", session_dir.display()))?; + let working_dir = match launch.working_dir { + Some(dir) => dir, + None => { + let dir = std::env::temp_dir() + .join("nodespace-agent-test-sessions") + .join(id.to_string()); + std::fs::create_dir_all(&dir) + .with_context(|| format!("create test session dir {}", dir.display()))?; + dir + } + }; let binary_path = which::which(binary) .map_err(|e| anyhow::anyhow!("test binary '{}' not on PATH: {}", binary, e))?; + let forwarded_vars = SystemAgentRegistry::new() + .get(launch.agent_type) + .map(|definition| definition.env_vars) + .unwrap_or_default(); - let pty_system = native_pty_system(); - let pair = pty_system.openpty(PtySize { - rows: DEFAULT_PTY_ROWS, - cols: DEFAULT_PTY_COLS, - pixel_width: 0, - pixel_height: 0, - })?; - let mut cmd = CommandBuilder::new(binary_path); - cmd.cwd(&session_dir); - apply_env_allowlist(&mut cmd, &[]); - for a in &args { - cmd.arg(a); - } - let child = pair.slave.spawn_command(cmd)?; - drop(pair.slave); - - let reader = pair.master.try_clone_reader()?; - let writer = pair.master.take_writer()?; - let child_killer = child.clone_killer(); - - let (output_tx, _) = broadcast::channel::(OUTPUT_CHANNEL_CAPACITY); - let (exit_tx, _) = watch::channel::>(None); - - let started_at = Utc::now(); - - let master = Arc::new(Mutex::new(pair.master)); - let writer = Arc::new(Mutex::new(writer)); - let child_killer = Arc::new(Mutex::new(child_killer)); - let capture = Arc::new(Mutex::new(SessionCapture::new())); - - spawn_reader_task(reader, output_tx.clone(), capture.clone()); - spawn_exit_watcher_task(child, exit_tx.clone()); + let mut env = launch.env; + env.push((SESSION_ENV_VAR.to_string(), id.to_string())); - Ok(Self { + let mut session = Self::spawn_in_pty(Spawn { id, - agent_type: AgentType::ClaudeCode, - started_at, - node_id: None, - master, - writer, - child_killer, - output_tx, - exit_tx, - capture, - session_dir, - }) - } - - /// Test-only accessor: the session's working directory path. - pub fn session_dir_path(&self) -> &std::path::Path { - self.session_dir.as_path() + agent_type: launch.agent_type, + binary_path, + args, + working_dir, + forwarded_vars, + search_path: None, + env, + })?; + session.node_id = launch.node_id; + Ok(session) } } @@ -601,7 +683,7 @@ mod tests { .expect("launch test session"); assert_eq!(session.agent_type, AgentType::ClaudeCode); - assert!(session.session_dir_path().exists()); + assert!(session.working_dir.exists()); let mut rx = session.subscribe_output(); let mut exit_rx = session.subscribe_exit(); @@ -671,11 +753,11 @@ mod tests { } #[tokio::test] - async fn session_dir_persists_after_process_exit_and_session_drop() { + async fn the_working_directory_outlives_the_process_and_the_session() { let session = PtySession::launch_for_test("sh", vec!["-c".into(), "echo done".into()]) .expect("launch echo session"); - let dir_path = session.session_dir_path().to_path_buf(); + let dir_path = session.working_dir.clone(); assert!( dir_path.exists(), "session dir should exist immediately after launch" @@ -867,4 +949,91 @@ mod tests { "extra var absent from process env should not appear" ); } + + // ---- What a launch sets ------------------------------------------------ + + /// Run `script` under `sh` the way a launch spawns an agent, and return + /// what it printed. + async fn run_as_launched(launch: SessionLaunch, script: &str) -> (PtySession, String) { + let session = + PtySession::launch_for_test_as("sh", vec!["-c".into(), script.into()], launch) + .expect("launch session"); + let mut rx = session.subscribe_output(); + let mut exit_rx = session.subscribe_exit(); + let output = collect_output(&mut rx, Duration::from_secs(2), 64).await; + await_exit(&mut exit_rx, Duration::from_secs(2)).await; + (session, String::from_utf8_lossy(&output).into_owned()) + } + + #[tokio::test] + async fn a_session_runs_in_the_folder_the_launch_names() { + let project = tempfile::TempDir::new().unwrap(); + let folder = project.path().canonicalize().unwrap(); + let launch = SessionLaunch { + working_dir: Some(folder.clone()), + ..SessionLaunch::new(AgentType::ClaudeCode) + }; + + let (session, text) = run_as_launched(launch, "pwd -P").await; + + assert_eq!(session.working_dir, folder); + assert!( + text.contains(&*folder.to_string_lossy()), + "the agent should run in the named folder, got: {text:?}" + ); + // Nothing is written there at launch: no context file, no skill copy. + assert_eq!(std::fs::read_dir(&folder).unwrap().count(), 0); + } + + #[tokio::test] + async fn the_environment_carries_what_the_launch_sets_and_the_session_id() { + let launch = SessionLaunch { + env: vec![ + ("NODESPACE_DATABASE".to_string(), "db-two".to_string()), + (LAUNCHED_FOR_ENV_VAR.to_string(), "task-1".to_string()), + ], + ..SessionLaunch::new(AgentType::ClaudeCode) + }; + + let (session, text) = run_as_launched(launch, "env").await; + + assert!(text.contains("NODESPACE_DATABASE=db-two"), "{text:?}"); + assert!( + text.contains(&format!("{LAUNCHED_FOR_ENV_VAR}=task-1")), + "{text:?}" + ); + assert!( + text.contains(&format!("{SESSION_ENV_VAR}={}", session.id)), + "{text:?}" + ); + } + + /// An agent's own variables reach it when the daemon has them, and no + /// other agent's do: the allowlist stays an allowlist. + #[tokio::test] + async fn an_agents_own_variables_are_forwarded_and_no_others() { + let _config = EnvVarGuard::set("CLAUDE_CONFIG_DIR", "/profiles/work"); + let _other = EnvVarGuard::set("NODESPACE_TEST_NOT_LISTED", "should-not-leak"); + + let (_, claude) = run_as_launched(SessionLaunch::new(AgentType::ClaudeCode), "env").await; + assert!( + claude.contains("CLAUDE_CONFIG_DIR=/profiles/work"), + "{claude:?}" + ); + assert!(!claude.contains("NODESPACE_TEST_NOT_LISTED"), "{claude:?}"); + + let (_, codex) = run_as_launched(SessionLaunch::new(AgentType::Codex), "env").await; + assert!(!codex.contains("CLAUDE_CONFIG_DIR"), "{codex:?}"); + } + + #[tokio::test] + async fn the_latest_reported_harness_session_id_stands() { + let session = PtySession::launch_for_test("true", vec![]).expect("launch session"); + assert_eq!(session.harness_session_id(), None); + + session.report_harness_session_id("first".to_string()); + session.report_harness_session_id("second".to_string()); + + assert_eq!(session.harness_session_id().as_deref(), Some("second")); + } } diff --git a/packages/agent/tests/it/live_context_assembly_scope.rs b/packages/agent/tests/it/live_context_assembly_scope.rs deleted file mode 100644 index 057d5fd43..000000000 --- a/packages/agent/tests/it/live_context_assembly_scope.rs +++ /dev/null @@ -1,164 +0,0 @@ -//! Live, real-embedding-model test: the context file a PTY session is given -//! lists the user's knowledge near its seed nodes, and no system content. -//! -//! Skills, tools and every schema are embedded so retrieval can find them by -//! meaning. Those vectors sit in the index the assembler's neighbour search -//! reads, and a built-in type's name ("Checkbox", "Task") or a skill's -//! description is close to many ordinary notes, so a search with no scope -//! returns them as "related context" for a seed note. -//! -//! Ignored by default — loads a real embedding model from the standard -//! NodeSpace catalog path and skips when the model is not on disk. Run -//! explicitly: -//! -//! ```text -//! .tools/bin/cargo-nextest nextest run -p nodespace-agent --test it \ -//! live_context_assembly_scope:: --run-ignored only --no-capture -//! ``` - -use std::collections::HashSet; -use std::sync::Arc; - -use nodespace_agent::agent_catalog::context_assembly::GraphContextAssembler; -use nodespace_agent::skill_pipeline::{seed_skill_nodes, seed_tool_nodes}; -use nodespace_core::db::SqliteStore; -use nodespace_core::markdown::prepare_nodes_from_template; -use nodespace_core::models::Node; -use nodespace_core::services::{NodeAccessor, NodeEmbeddingService, NodeService, SearchScope}; -use nodespace_nlp_engine::{EmbeddingConfig, EmbeddingService}; -use serde_json::json; -use tempfile::TempDir; -use tokio::sync::RwLock; - -/// A root text note. -async fn note(node_service: &NodeService, content: &str) -> Node { - let node = Node::new("text".to_string(), content.to_string(), json!({})); - node_service - .create_node(node.clone()) - .await - .expect("note must create"); - node -} - -#[tokio::test] -#[ignore = "requires the locked nomic-embed-text-v1.5 GGUF on disk"] -async fn related_context_holds_the_users_knowledge_and_no_system_content() { - let temp_dir = TempDir::new().expect("tempdir"); - let db_path = temp_dir.path().join("test.db"); - let mut store = Arc::new(SqliteStore::new(db_path).await.expect("store must open")); - let node_service = Arc::new( - NodeService::new(&mut store) - .await - .expect("node service must init"), - ); - - let mut nlp = EmbeddingService::new(EmbeddingConfig::default()).expect("config must validate"); - if nlp.initialize().is_err() || !nlp.is_initialized() { - eprintln!( - "SKIP live_context_assembly_scope: nomic-embed-text-v1.5 model not found on disk" - ); - return; - } - let node_accessor: Arc = node_service.clone(); - let embedding_service = Arc::new(NodeEmbeddingService::new( - Arc::new(nlp), - store.clone(), - node_accessor, - node_service.behaviors().clone(), - )); - - let groups: Vec<_> = seed_skill_nodes() - .iter() - .chain(seed_tool_nodes().iter()) - .map(|t| prepare_nodes_from_template(t).expect("template must parse")) - .collect(); - node_service - .seed_nodes_from_templates(groups) - .await - .expect("seeding the built-in skills and tools must succeed"); - - let seed = note( - &node_service, - "Create a task with a checkbox for every item on the packing list", - ) - .await; - let related = note( - &node_service, - "Packing list for the offsite: one checkbox per item, tick it when packed", - ) - .await; - - // Embed what the write paths queued (the built-in schemas, the skills and - // the tools) and the two notes. - for node in [&seed, &related] { - embedding_service - .embed_root_node(&node.id) - .await - .expect("note must embed"); - } - let queued = store - .get_stale_embedding_root_ids(None, 0, 3) - .await - .expect("the queue must read"); - for id in queued { - embedding_service - .embed_root_node(&id) - .await - .unwrap_or_else(|e| panic!("queued root {id} must embed: {e}")); - } - - // What a search with no scope returns for this seed. The test proves - // nothing unless system content really is among the seed's nearest - // neighbours. - let no_user_types = HashSet::new(); - let unscoped = embedding_service - .semantic_search_nodes(&seed.content, 50, 0.3, None, false) - .await - .expect("search must succeed"); - let system: Vec<&Node> = unscoped - .iter() - .map(|(node, _)| node) - .filter(|node| { - !NodeEmbeddingService::matches_scope(node, &SearchScope::Knowledge, &no_user_types) - }) - .collect(); - let nearest_five: Vec<&str> = unscoped - .iter() - .take(5) - .map(|(n, _)| n.id.as_str()) - .collect(); - eprintln!( - "unscoped: {:?}", - unscoped - .iter() - .map(|(n, score)| (n.node_type.as_str(), n.id.as_str(), *score)) - .collect::>() - ); - assert!( - system.iter().any(|n| nearest_five.contains(&n.id.as_str())), - "the seed must have system content among its five nearest neighbours, got {nearest_five:?}" - ); - - let context = GraphContextAssembler::new( - node_service.clone(), - Arc::new(RwLock::new(Some(embedding_service.clone()))), - ) - .with_seed_nodes(vec![seed.id.clone()]) - .assemble_context() - .await - .expect("context must assemble"); - eprintln!("{context}"); - - assert!( - context.contains(&format!("(id: {})", related.id)), - "the related note must be in the context" - ); - for node in system { - assert!( - !context.contains(&format!("(id: {})", node.id)), - "{} `{}` is system content and must not be in the context", - node.node_type, - node.id - ); - } -} diff --git a/packages/agent/tests/it/live_embedding_prefix_measurement.rs b/packages/agent/tests/it/live_embedding_prefix_measurement.rs index 4859a6edb..114ec626f 100644 --- a/packages/agent/tests/it/live_embedding_prefix_measurement.rs +++ b/packages/agent/tests/it/live_embedding_prefix_measurement.rs @@ -51,9 +51,9 @@ use tempfile::TempDir; /// `MAX_SEMANTIC_SCHEMAS`. const SCHEMA_THRESHOLD: f64 = 0.2; const MAX_SEMANTIC_SCHEMAS: usize = 5; -/// `SEMANTIC_THRESHOLD` in `local_agent/tools.rs` and `context_assembly.rs`. +/// `SEMANTIC_THRESHOLD` in `local_agent/tools.rs`. const SEMANTIC_THRESHOLD: f32 = 0.3; -/// `context_assembly::NEIGHBORS_PER_SEED`. +/// How many neighbours of a seed node the measurement reads. const NEIGHBORS_PER_SEED: usize = 5; // --------------------------------------------------------------------------- diff --git a/packages/agent/tests/it/main.rs b/packages/agent/tests/it/main.rs index 5e6a68cff..6124de080 100644 --- a/packages/agent/tests/it/main.rs +++ b/packages/agent/tests/it/main.rs @@ -16,7 +16,6 @@ mod get_related_nodes_both_direction; mod golden_scenario6_handauthored; mod golden_scenario6_sequence; mod link_field_values; -mod live_context_assembly_scope; mod live_delete_resolved_incidents; mod live_embedding_prefix_measurement; mod live_openai_compat_routing; diff --git a/packages/cli/Cargo.toml b/packages/cli/Cargo.toml index 4483362e4..621d11a90 100644 --- a/packages/cli/Cargo.toml +++ b/packages/cli/Cargo.toml @@ -28,6 +28,9 @@ serde_json = { workspace = true } regex = "1" nodespace-daemon = { path = "../daemon" } nodespace-proto = { workspace = true } +# The agent ids `session launch` takes and the variable that names a launched +# session: one definition, shared with the daemon that launches it. +nodespace-agent = { workspace = true } # For `flatten_namespaced_properties` — the single definition of the storage # nesting → flat API shape rule this CLI's output must honour. nodespace-types = { workspace = true } @@ -47,7 +50,6 @@ libc = "0.2" [dev-dependencies] nodespace-core = { workspace = true } -nodespace-agent = { workspace = true } nodespace-nlp-engine = { workspace = true } tempfile = "3.0" tokio = { workspace = true, features = ["full", "test-util"] } diff --git a/packages/cli/src/commands/session.rs b/packages/cli/src/commands/session.rs index 966daa008..c2cbcaf70 100644 --- a/packages/cli/src/commands/session.rs +++ b/packages/cli/src/commands/session.rs @@ -4,9 +4,11 @@ use anyhow::{Context, Result}; use chrono::{Local, TimeZone}; use clap::{Args, Subcommand}; use crossterm::terminal; +use nodespace_agent::agent_types::AgentType; +use nodespace_agent::pty::SESSION_ENV_VAR; use nodespace_daemon::nodespace::{ - LaunchSessionRequest, ListSessionsRequest, StreamOutputRequest, TerminateSessionRequest, - WriteInputRequest, + LaunchSessionRequest, ListSessionsRequest, ReportHarnessSessionRequest, StreamOutputRequest, + TerminateSessionRequest, WriteInputRequest, }; use tokio::io::AsyncReadExt; use tonic::Request; @@ -25,17 +27,44 @@ pub enum SessionAction { List(ListArgs), /// Terminate a running session. Kill(KillArgs), + /// Tell NodeSpace the agent's own id for the conversation running in a + /// launched session. An agent's plugin runs this when the session starts. + #[command(name = "report-harness-session")] + ReportHarnessSession(ReportHarnessSessionArgs), +} + +/// The ids `session launch` takes: the agents the daemon can launch. +fn agent_ids() -> Vec<&'static str> { + AgentType::ALL.map(AgentType::id).to_vec() } #[derive(Args, Debug)] pub struct LaunchArgs { /// Agent to launch: claude-code, codex, antigravity, pi, opencode + #[arg(value_parser = clap::builder::PossibleValuesParser::new(agent_ids()))] pub agent: String, /// Initial prompt passed to the agent at launch time. #[arg(long)] pub prompt: Option, + /// Id of the project to launch the session for. The session runs in that + /// project's folder on this machine. Without it the session runs in a + /// private folder of its own. + #[arg(long)] + pub project: Option, + + /// The project's folder on this machine: the absolute path of its + /// checkout. Needed the first time a session is launched for a project, + /// and remembered on this machine from then on. + #[arg(long, requires = "project")] + pub folder: Option, + + /// Id of the task to launch the session for. The agent's plugin opens + /// with that task's context. + #[arg(long)] + pub task: Option, + /// Terminal width in columns (defaults to current terminal width). #[arg(long)] pub cols: Option, @@ -60,12 +89,26 @@ pub struct KillArgs { pub session_id: String, } -pub async fn run(client: &mut SessionClient, action: SessionAction, _json: bool) -> Result<()> { +#[derive(Args, Debug)] +pub struct ReportHarnessSessionArgs { + /// The agent's own id for the conversation: the one its resume flag takes. + pub harness_session_id: String, + + /// The launched session to report for. A launched session's environment + /// names it in `NODESPACE_SESSION`. + #[arg(long, env = SESSION_ENV_VAR)] + pub session: String, +} + +pub async fn run(client: &mut SessionClient, action: SessionAction, json: bool) -> Result<()> { match action { SessionAction::Launch(args) => launch(client, args).await, SessionAction::Attach(args) => attach(client, args).await, SessionAction::List(_) => list(client).await, SessionAction::Kill(args) => kill(client, args).await, + SessionAction::ReportHarnessSession(args) => { + report_harness_session(client, args, json).await + } } } @@ -81,6 +124,9 @@ async fn launch(client: &mut SessionClient, args: LaunchArgs) -> Result<()> { // CLI-launched sessions are not tied to an ai-chat-pty node, so // nothing is written when the session ends. node_id: None, + project_id: args.project, + project_folder: args.folder, + task_id: args.task, })) .await .context("LaunchSession RPC failed")? @@ -92,6 +138,31 @@ async fn launch(client: &mut SessionClient, args: LaunchArgs) -> Result<()> { stream_bridge(client, resp.session_id).await } +async fn report_harness_session( + client: &mut SessionClient, + args: ReportHarnessSessionArgs, + json: bool, +) -> Result<()> { + let resp = client + .report_harness_session(Request::new(ReportHarnessSessionRequest { + session_id: args.session.clone(), + harness_session_id: args.harness_session_id, + })) + .await + .context("ReportHarnessSession RPC failed")? + .into_inner(); + + if json { + println!( + "{}", + serde_json::json!({ "session_id": args.session, "node_id": resp.node_id }) + ); + } else { + println!("Recorded for session {}.", args.session); + } + Ok(()) +} + async fn attach(client: &mut SessionClient, args: AttachArgs) -> Result<()> { stream_bridge(client, args.session_id).await } @@ -247,6 +318,69 @@ fn format_unix_time(unix_secs: i64) -> String { #[cfg(test)] mod tests { use super::dropped_notice; + use super::{LaunchArgs, ReportHarnessSessionArgs, SESSION_ENV_VAR}; + use clap::{CommandFactory, Parser}; + use nodespace_agent::agent_types::AgentType; + + #[derive(Parser)] + struct Launch { + #[command(flatten)] + args: LaunchArgs, + } + + #[derive(Parser)] + struct Report { + #[command(flatten)] + args: ReportHarnessSessionArgs, + } + + /// `session launch` lists and takes exactly the ids the daemon accepts. + #[test] + fn launch_takes_the_daemons_agent_ids_and_no_others() { + let help = Launch::command().render_long_help().to_string(); + for agent in AgentType::ALL { + assert!(help.contains(agent.id()), "help omits {}", agent.id()); + assert!(Launch::try_parse_from(["launch", agent.id()]).is_ok()); + } + for other in ["open-code", "antigravity-cli", "gemini-cli"] { + assert!(!help.contains(other), "help lists {other}"); + assert!(Launch::try_parse_from(["launch", other]).is_err()); + } + } + + #[test] + fn launch_takes_a_project_a_folder_and_a_task() { + let launch = Launch::try_parse_from([ + "launch", + "claude-code", + "--project", + "p1", + "--folder", + "/work/core", + "--task", + "t1", + ]) + .unwrap() + .args; + assert_eq!(launch.project.as_deref(), Some("p1")); + assert_eq!(launch.folder.as_deref(), Some("/work/core")); + assert_eq!(launch.task.as_deref(), Some("t1")); + + // A folder belongs to a project. + assert!(Launch::try_parse_from(["launch", "codex", "--folder", "/work/core"]).is_err()); + } + + /// The session is named by flag here: the variable a launch sets is + /// process-global, and other tests share the process. + #[test] + fn a_report_names_its_session() { + let report = Report::try_parse_from(["report", "h-1", "--session", "s-1"]) + .unwrap() + .args; + assert_eq!(report.harness_session_id, "h-1"); + assert_eq!(report.session, "s-1"); + assert_eq!(SESSION_ENV_VAR, "NODESPACE_SESSION"); + } #[test] fn dropped_notice_starts_on_a_fresh_line_and_pluralises() { diff --git a/packages/cli/src/lib.rs b/packages/cli/src/lib.rs index 6b3e901c1..4648c3ea8 100644 --- a/packages/cli/src/lib.rs +++ b/packages/cli/src/lib.rs @@ -113,7 +113,7 @@ pub struct Cli { /// Target a specific local database by name or id (ADR-053). /// When omitted, requests route to the daemon's default database. /// Honors the `NODESPACE_DATABASE` environment variable when this flag is absent. - #[arg(long, global = true, env = "NODESPACE_DATABASE")] + #[arg(long, global = true, env = nodespace_proto::DATABASE_ENV_VAR)] pub database: Option, #[command(subcommand)] diff --git a/packages/cli/tests/it/cli_integration.rs b/packages/cli/tests/it/cli_integration.rs index 24f8e7223..711535b4f 100644 --- a/packages/cli/tests/it/cli_integration.rs +++ b/packages/cli/tests/it/cli_integration.rs @@ -165,7 +165,7 @@ pub(crate) async fn spawn_test_daemon_with_seeded_skills( /// A model-less build context — with `has_model = false` no embedding wiring /// runs, so the dropped watch sender is harmless (it is never read). -fn routing_test_context() -> SharedContext { +pub(crate) fn routing_test_context() -> SharedContext { let (_tx, model) = watch::channel::>>(None); SharedContext { pty_manager: Arc::new(PtySessionManager::new()), diff --git a/packages/cli/tests/it/launched_session.rs b/packages/cli/tests/it/launched_session.rs new file mode 100644 index 000000000..2790fbf5b --- /dev/null +++ b/packages/cli/tests/it/launched_session.rs @@ -0,0 +1,525 @@ +//! A session the daemon launches for a project (ADR-093 §8), end to end. +//! +//! The daemon serves two databases and launches an agent in the second. The +//! agent is a shell script standing in for `claude`, found on a search path +//! the test owns, so nothing here depends on an installed agent. The script +//! prints where it runs and what its environment names, then runs the compiled +//! `nodespace` binary the way an agent or its plugin would: with no flags, +//! from the environment the launch gave it. + +#![cfg(unix)] + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use nodespace_cli::{ + connect, connect_database, connect_session, DatabaseIdInterceptor, NodeClient, SessionClient, +}; +use nodespace_daemon::nodespace::{ + CreateDatabaseRequest, CreateNodeRequest, GetNodeRequest, LaunchSessionRequest, + StreamOutputRequest, TerminateSessionRequest, UpdateNodeRequest, WriteInputRequest, +}; +use nodespace_daemon::{ + AgentSessionServiceServer, DatabaseManager, DatabaseServiceImpl, DatabaseServiceServer, + DbManagerLayer, NodeServiceServer, +}; +use tempfile::TempDir; +use tokio::net::UnixListener; +use tokio::sync::oneshot; +use tokio::time::timeout; +use tokio_stream::wrappers::UnixListenerStream; +use tonic::transport::Server; +use tonic::Code; + +use crate::cli_integration::routing_test_context; + +/// What the stand-in agent prints once it has run its commands. +const DONE: &str = "AGENT-DONE"; + +struct Daemon { + sock: PathBuf, + home: TempDir, + /// The id of the second, non-default database. + second_id: String, + _shutdown: oneshot::Sender<()>, +} + +impl Daemon { + async fn node(&self) -> NodeClient { + let interceptor = DatabaseIdInterceptor::for_id(&self.second_id).expect("interceptor"); + connect(&self.sock, interceptor) + .await + .expect("connect node") + } + + async fn session(&self) -> SessionClient { + let interceptor = DatabaseIdInterceptor::for_id(&self.second_id).expect("interceptor"); + connect_session(&self.sock, interceptor) + .await + .expect("connect session") + } + + /// A folder standing in for the project's checkout, resolved the way a + /// process reads its own working directory back. + fn checkout(&self) -> PathBuf { + let dir = self.home.path().join("checkout"); + std::fs::create_dir_all(&dir).expect("create checkout"); + dir.canonicalize().expect("resolve checkout") + } +} + +/// Write the stand-in `claude` into `bin`. It waits for a line of input, so +/// the test is streaming before it prints, then waits for a second before it +/// exits, so the test decides when the session ends. +fn write_agent(bin: &Path, task_id: &str) { + let nodespace = env!("CARGO_BIN_EXE_nodespace"); + let script = format!( + r#"#!/bin/sh +read _go +echo "CWD=$(pwd -P)" +echo "LAUNCHED_FOR=${{NODESPACE_LAUNCHED_FOR-unset}}" +echo "DATABASE=$NODESPACE_DATABASE" +echo "PATH=$PATH" +'{nodespace}' node get {task_id} +echo "GET-EXIT=$?" +'{nodespace}' session report-harness-session "harness-$NODESPACE_SESSION" +echo "REPORT-EXIT=$?" +echo {DONE} +read _end +"# + ); + std::fs::create_dir_all(bin).expect("create bin"); + let path = bin.join("claude"); + std::fs::write(&path, script).expect("write agent"); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).expect("chmod agent"); +} + +/// The multi-database stack with `AgentSessionService` beside it, a second +/// database registered, and agents found only in the test's own folder. +async fn spawn_daemon() -> Daemon { + let home = TempDir::new().expect("tempdir"); + // The daemon's own files (its capture settings among them) are read from + // here, not from the home of whoever runs the test. Each test is a process + // of its own under nextest, so the variable is this test's alone. + std::env::set_var("NODESPACE_HOME", home.path().join("nodespace-home")); + let sock = home.path().join("daemon.sock"); + + let context = routing_test_context(); + context.pty_manager.set_daemon_socket(sock.clone()); + let search_path = std::env::join_paths([ + home.path().join("bin"), + PathBuf::from("/usr/bin"), + PathBuf::from("/bin"), + ]) + .expect("search path"); + context.pty_manager.set_agent_search_path(search_path); + + let manager = Arc::new( + DatabaseManager::load(home.path().join("databases.toml"), context) + .await + .expect("load DatabaseManager"), + ); + let default_id = manager + .ensure_default_registered("Default".into(), home.path().join("default.db")) + .await + .expect("register default"); + let bundle = manager + .get_or_open(&default_id) + .await + .expect("open default"); + let node = bundle.node_service_grpc.clone(); + let agent_session = bundle.agent_session.clone(); + + let listener = UnixListener::bind(&sock).expect("bind socket"); + let (shutdown, shutdown_rx) = oneshot::channel::<()>(); + let served = manager.clone(); + tokio::spawn(async move { + Server::builder() + .layer(DbManagerLayer::new(served.clone())) + .add_service(DatabaseServiceServer::new(DatabaseServiceImpl::new(served))) + .add_service(NodeServiceServer::new(node)) + .add_service(AgentSessionServiceServer::new(agent_session)) + .serve_with_incoming_shutdown(UnixListenerStream::new(listener), async move { + let _ = shutdown_rx.await; + }) + .await + .expect("server crashed"); + }); + + let mut db = None; + for _ in 0..50 { + if let Ok(client) = connect_database(&sock).await { + db = Some(client); + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + let second_id = db + .expect("daemon accepts connections") + .create(CreateDatabaseRequest { + name: "Second".into(), + path: Some(home.path().join("second.db").display().to_string()), + }) + .await + .expect("create second database") + .into_inner() + .id; + + Daemon { + sock, + home, + second_id, + _shutdown: shutdown, + } +} + +async fn create(node: &mut NodeClient, node_type: &str, content: &str, properties: &str) -> String { + node.create_node(CreateNodeRequest { + node_type: node_type.into(), + content: content.into(), + parent_id: None, + properties: properties.into(), + collections: Vec::new(), + collection_ids: Vec::new(), + lifecycle_status: None, + id: None, + position: None, + }) + .await + .unwrap_or_else(|e| panic!("create {node_type}: {e}")) + .into_inner() + .node_id +} + +/// A node's stored properties, as the daemon returns them. +async fn properties(node: &mut NodeClient, id: &str) -> serde_json::Value { + let data = node + .get_node(GetNodeRequest { + node_id: id.to_string(), + }) + .await + .expect("get node") + .into_inner() + .node_data + .expect("node data"); + serde_json::from_str(&data.properties).expect("properties are JSON") +} + +fn launch_for(project_id: &str) -> LaunchSessionRequest { + LaunchSessionRequest { + agent_type: "claude-code".into(), + prompt: None, + cols: 0, + rows: 0, + node_id: None, + project_id: Some(project_id.to_string()), + project_folder: None, + task_id: None, + } +} + +/// Start the stand-in agent's commands and return what it printed up to +/// [`DONE`]. +async fn run_agent(session: &mut SessionClient, session_id: &str) -> String { + let mut stream = session + .stream_output(StreamOutputRequest { + session_id: session_id.to_string(), + }) + .await + .expect("stream output") + .into_inner(); + send_line(session, session_id).await; + + let mut output = String::new(); + let read = timeout(Duration::from_secs(30), async { + while let Ok(Some(chunk)) = stream.message().await { + output.push_str(&String::from_utf8_lossy(&chunk.data)); + // The agent's own line, not the terminal's echo of the script. + if output.contains(&format!("{DONE}\r\n")) { + break; + } + } + }) + .await; + assert!(read.is_ok(), "the agent did not finish, printed: {output}"); + output +} + +async fn send_line(session: &mut SessionClient, session_id: &str) { + session + .write_input(WriteInputRequest { + session_id: session_id.to_string(), + data: b"\n".to_vec(), + }) + .await + .expect("write input"); +} + +/// Wait for the daemon to record the end of the session on `chat_id`, and +/// return the chat's own fields. +async fn ended_chat(node: &mut NodeClient, chat_id: &str) -> serde_json::Value { + for _ in 0..200 { + let chat = properties(node, chat_id).await["ai-chat-pty"].clone(); + if chat["session_status"] == "ended" { + return chat; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the session on {chat_id} was never recorded as ended"); +} + +#[tokio::test] +async fn a_session_launched_for_a_project_runs_in_its_folder_against_its_database() { + let daemon = spawn_daemon().await; + let mut node = daemon.node().await; + let mut session = daemon.session().await; + + let project_id = create(&mut node, "project", "Widgets", "").await; + let task_id = create(&mut node, "task", "Add the gauge", "").await; + let chat_id = create( + &mut node, + "ai-chat-pty", + "Session", + r#"{"agent":"claude-code"}"#, + ) + .await; + write_agent(&daemon.home.path().join("bin"), &task_id); + let checkout = daemon.checkout(); + + // The project has no folder on this machine yet, and the launch names none. + let refused = session + .launch_session(launch_for(&project_id)) + .await + .expect_err("a project with no folder cannot host a session"); + assert_eq!(refused.code(), Code::FailedPrecondition); + assert!(refused.message().contains("Widgets"), "{refused}"); + + // A launch refused for another reason stores no folder: here, a task + // that does not exist. + let unknown_task = session + .launch_session(LaunchSessionRequest { + project_folder: Some(checkout.display().to_string()), + task_id: Some("no-such-task".into()), + ..launch_for(&project_id) + }) + .await + .expect_err("an unknown task is refused"); + assert_eq!(unknown_task.code(), Code::NotFound); + assert!(properties(&mut node, &project_id).await["project"]["checkout_path"].is_null()); + + // A folder that does not exist is refused, and not remembered. + let missing = session + .launch_session(LaunchSessionRequest { + project_folder: Some(checkout.join("nowhere").display().to_string()), + ..launch_for(&project_id) + }) + .await + .expect_err("a folder that does not exist is refused"); + assert_eq!(missing.code(), Code::InvalidArgument); + assert!(properties(&mut node, &project_id).await["project"]["checkout_path"].is_null()); + + let launched = session + .launch_session(LaunchSessionRequest { + node_id: Some(chat_id.clone()), + project_folder: Some(checkout.display().to_string()), + task_id: Some(task_id.clone()), + ..launch_for(&project_id) + }) + .await + .expect("launch for the project") + .into_inner(); + assert_eq!(Path::new(&launched.working_dir), checkout); + + let output = run_agent(&mut session, &launched.session_id).await; + + // It runs in the project's folder, and nothing was written there. + assert!( + output.contains(&format!("CWD={}", checkout.display())), + "{output}" + ); + assert_eq!(std::fs::read_dir(&checkout).unwrap().count(), 0); + // Its environment names the task it was launched for, and the database. + assert!( + output.contains(&format!("LAUNCHED_FOR={task_id}")), + "{output}" + ); + assert!( + output.contains(&format!("DATABASE={}", daemon.second_id)), + "{output}" + ); + // A `nodespace` command with no flags reads that database: the task + // exists in the second database and nowhere else. + assert!(output.contains("GET-EXIT=0"), "{output}"); + assert!(output.contains("Add the gauge"), "{output}"); + assert!(output.contains("REPORT-EXIT=0"), "{output}"); + + // The folder is remembered on the project: the next launch names none. + assert_eq!( + properties(&mut node, &project_id).await["project"]["checkout_path"], + checkout.display().to_string() + ); + // The agent's PATH is the search path it was found on, so it can run what + // detection saw. + let bin = daemon.home.path().join("bin"); + assert!( + output.contains(&format!("PATH={}:", bin.display())), + "{output}" + ); + + let again = session + .launch_session(launch_for(&project_id)) + .await + .expect("the remembered folder is used") + .into_inner(); + session + .terminate_session(TerminateSessionRequest { + session_id: again.session_id.clone(), + }) + .await + .expect("end the second session"); + assert_eq!(Path::new(&again.working_dir), checkout); + + // The session's end records the id the agent reported for itself. + send_line(&mut session, &launched.session_id).await; + let chat = ended_chat(&mut node, &chat_id).await; + assert_eq!( + chat["session_id"], + format!("harness-{}", launched.session_id) + ); +} + +/// Sessions in one project share a working directory, so the directory says +/// nothing about which conversation is whose: each session's own report does. +#[tokio::test] +async fn two_sessions_in_one_project_each_record_their_own_harness_session() { + let daemon = spawn_daemon().await; + let mut node = daemon.node().await; + let mut session = daemon.session().await; + + let project_id = create(&mut node, "project", "Widgets", "").await; + let task_id = create(&mut node, "task", "Add the gauge", "").await; + write_agent(&daemon.home.path().join("bin"), &task_id); + let checkout = daemon.checkout(); + + let mut launched = Vec::new(); + for title in ["First", "Second"] { + let chat_id = create( + &mut node, + "ai-chat-pty", + title, + r#"{"agent":"claude-code"}"#, + ) + .await; + let response = session + .launch_session(LaunchSessionRequest { + node_id: Some(chat_id.clone()), + project_folder: Some(checkout.display().to_string()), + ..launch_for(&project_id) + }) + .await + .expect("launch") + .into_inner(); + assert_eq!(Path::new(&response.working_dir), checkout); + launched.push((chat_id, response.session_id)); + } + + // Both are running in the one folder before either reports or ends. + for (chat_id, session_id) in &launched { + let output = run_agent(&mut session, session_id).await; + // With no task named, the session is launched for its chat node. + assert!( + output.contains(&format!("LAUNCHED_FOR={chat_id}")), + "{output}" + ); + } + for (_, session_id) in &launched { + send_line(&mut session, session_id).await; + } + + for (chat_id, session_id) in &launched { + let chat = ended_chat(&mut node, chat_id).await; + assert_eq!(chat["session_id"], format!("harness-{session_id}")); + } +} + +#[tokio::test] +async fn a_launch_with_no_project_runs_in_a_private_folder_with_the_environment_set() { + let daemon = spawn_daemon().await; + let mut node = daemon.node().await; + let mut session = daemon.session().await; + let task_id = create(&mut node, "task", "Add the gauge", "").await; + write_agent(&daemon.home.path().join("bin"), &task_id); + + // A folder is a project's: naming one without a project is refused. + let refused = session + .launch_session(LaunchSessionRequest { + project_id: None, + project_folder: Some(daemon.checkout().display().to_string()), + ..launch_for("") + }) + .await + .expect_err("a folder needs a project"); + assert_eq!(refused.code(), Code::InvalidArgument); + + let launched = session + .launch_session(LaunchSessionRequest { + project_id: None, + ..launch_for("") + }) + .await + .expect("launch with no project") + .into_inner(); + // Under the daemon's own home, not the home of whoever runs the test. + let working_dir = PathBuf::from(&launched.working_dir); + assert_eq!( + working_dir, + daemon + .home + .path() + .join("nodespace-home") + .join(".nodespace") + .join("agent-sessions") + .join(&launched.session_id) + ); + + let output = run_agent(&mut session, &launched.session_id).await; + assert!( + output.contains(&format!("DATABASE={}", daemon.second_id)), + "{output}" + ); + assert!(output.contains("GET-EXIT=0"), "{output}"); + // Nothing was launched for, so the variable is absent, not empty. + assert!(output.contains("LAUNCHED_FOR=unset\r\n"), "{output}"); + // No context file and no copy of the skill are written. + assert_eq!(std::fs::read_dir(&working_dir).unwrap().count(), 0); + + send_line(&mut session, &launched.session_id).await; +} + +/// The stored folder can be written by any update of the project, so a launch +/// checks it as it checks one it is handed. +#[tokio::test] +async fn a_stored_folder_that_is_not_absolute_refuses_the_launch() { + let daemon = spawn_daemon().await; + let mut node = daemon.node().await; + let mut session = daemon.session().await; + let project_id = create(&mut node, "project", "Widgets", "").await; + write_agent(&daemon.home.path().join("bin"), "unused"); + + node.update_node(UpdateNodeRequest { + node_id: project_id.clone(), + properties: Some(r#"{"checkout_path":"."}"#.into()), + ..Default::default() + }) + .await + .expect("store a relative folder"); + + let refused = session + .launch_session(launch_for(&project_id)) + .await + .expect_err("a relative folder is not where a session runs"); + assert_eq!(refused.code(), Code::FailedPrecondition); + assert!(refused.message().contains("absolute path"), "{refused}"); +} diff --git a/packages/cli/tests/it/main.rs b/packages/cli/tests/it/main.rs index fa2fc7010..d5797e67d 100644 --- a/packages/cli/tests/it/main.rs +++ b/packages/cli/tests/it/main.rs @@ -14,6 +14,7 @@ mod cli_integration; mod cli_reference_examples; mod home_isolation; mod import_dir_multi; +mod launched_session; mod mcp_integration; mod requires_extension; mod skill_md_generation; diff --git a/packages/core/src/models/core_schemas.rs b/packages/core/src/models/core_schemas.rs index efeb7d3f8..f41bcfef1 100644 --- a/packages/core/src/models/core_schemas.rs +++ b/packages/core/src/models/core_schemas.rs @@ -535,6 +535,18 @@ pub fn get_core_schemas() -> Vec { "The project's source repository: its title and URL. A checkout of that \ repository belongs to this project.", ), + // Where that checkout is on this machine (ADR-093 §8): an + // absolute path, so it means nothing on another one. + SchemaField { + local_only: true, + ..text_field( + "checkout_path", + "Checkout folder", + "The folder on this machine that holds the project's checkout. An \ + agent session launched for the project runs there. It is set on \ + each machine and stays on it.", + ) + }, ], // A project has many tasks; the inverse (a task's single project) is // derived from this declaration, so `task` needs no entry of its own. diff --git a/packages/core/tests/it/typed_flat_updates_test.rs b/packages/core/tests/it/typed_flat_updates_test.rs index e33941282..af512ad43 100644 --- a/packages/core/tests/it/typed_flat_updates_test.rs +++ b/packages/core/tests/it/typed_flat_updates_test.rs @@ -473,6 +473,61 @@ fn link(title: &str, url: &str) -> LinkValue { } } +/// `checkout_path` is the project's folder on this machine: set, read back +/// typed, cleared, and declared machine-bound. +#[tokio::test] +async fn a_project_update_writes_and_clears_its_checkout_path() { + let (svc, _tmp) = test_service().await; + let project = create(&svc, "project", "Apollo", json!({})).await; + assert!(typed(&svc, &project.id).await.get("checkoutPath").is_none()); + + let updated = svc + .update_project_node( + &project.id, + project.version, + ProjectNodeUpdate { + checkout_path: Some(Some("/work/apollo".to_string())), + ..Default::default() + }, + ) + .await + .expect("setting the checkout path succeeds"); + assert_eq!( + updated.properties["project"]["checkout_path"], + json!("/work/apollo") + ); + let wire = typed(&svc, &project.id).await; + assert_eq!(wire["checkoutPath"], json!("/work/apollo")); + assert!(wire["properties"].get("checkout_path").is_none()); + + let cleared = svc + .update_project_node( + &project.id, + updated.version, + ProjectNodeUpdate { + checkout_path: Some(None), + ..Default::default() + }, + ) + .await + .expect("clearing the checkout path succeeds"); + assert!(cleared.properties["project"]["checkout_path"].is_null()); + assert!(typed(&svc, &project.id).await.get("checkoutPath").is_none()); + + // An absolute path means nothing on another machine: the field stays here. + let field = nodespace_core::models::core_schemas::get_core_schemas() + .into_iter() + .find(|schema| schema.envelope.id == "project") + .and_then(|schema| { + schema + .fields + .into_iter() + .find(|field| field.name == "checkout_path") + }) + .expect("project declares checkout_path"); + assert!(field.local_only); +} + /// `repository` is a link: set, read back typed, and cleared. #[tokio::test] async fn a_project_update_writes_and_clears_its_repository() { diff --git a/packages/daemon/src/db_routing.rs b/packages/daemon/src/db_routing.rs index 034ce5eaa..ad0dd01ba 100644 --- a/packages/daemon/src/db_routing.rs +++ b/packages/daemon/src/db_routing.rs @@ -88,6 +88,31 @@ pub(crate) async fn routed_database_services( Ok(Some(services)) } +/// The id of the database a routed request targets: the one its +/// `x-ns-database-id` header names, or the default when it names none. +/// +/// `None` on a host without the routing middleware, which serves one database +/// and has no id for it. Call it beside [`route_or_self`], which is what +/// rejects a header such a host cannot honour. +pub(crate) async fn routed_database_id( + request: &tonic::Request, +) -> Result, Status> { + let Some(manager) = request.extensions().get::>() else { + return Ok(None); + }; + let header = request + .metadata() + .get(DATABASE_ID_HEADER) + .map(|v| v.to_str()) + .transpose() + .map_err(|_| Status::invalid_argument("x-ns-database-id must be valid ASCII"))?; + manager + .resolve_database_id(header) + .await + .map(|id| Some(id.as_str().to_string())) + .map_err(|e| Status::not_found(e.to_string())) +} + /// The common `route` adapter: resolve the target database via /// [`routed_database_services`] and return `pick`'s service from it, or a /// clone of `this` when no routing is in play. Services whose routed handle diff --git a/packages/daemon/src/lib.rs b/packages/daemon/src/lib.rs index 6591109e3..2d34b17dc 100644 --- a/packages/daemon/src/lib.rs +++ b/packages/daemon/src/lib.rs @@ -182,8 +182,9 @@ pub use nodespace_proto::{ DatabaseServiceClient, DatabaseServiceServer, EmbeddingsServiceClient, EmbeddingsServiceServer, GetCaptureSettingsRequest, ImportServiceClient, ImportServiceServer, LaunchSessionRequest, LaunchSessionResponse, ListSessionsRequest, ListSessionsResponse, LocalAgentServiceClient, - LocalAgentServiceServer, NodeData, NodeServiceClient, NodeServiceServer, ResizeRequest, - ResizeResponse, SessionInfo, SettingsServiceClient, SettingsServiceServer, StreamOutputRequest, + LocalAgentServiceServer, NodeData, NodeServiceClient, NodeServiceServer, + ReportHarnessSessionRequest, ReportHarnessSessionResponse, ResizeRequest, ResizeResponse, + SessionInfo, SettingsServiceClient, SettingsServiceServer, StreamOutputRequest, TerminateSessionRequest, TerminateSessionResponse, UpdateCaptureSettingsRequest, WriteInputRequest, WriteInputResponse, }; diff --git a/packages/daemon/src/main.rs b/packages/daemon/src/main.rs index 4b75802a8..379a88afe 100644 --- a/packages/daemon/src/main.rs +++ b/packages/daemon/src/main.rs @@ -1676,6 +1676,9 @@ async fn serve_headless() -> Result<()> { let shutdown = install_shutdown_handler().context("Failed to install signal handlers")?; // _model_task: dropping a JoinHandle does not cancel the task in tokio — it detaches. let (shared, _model_task) = build_shared_services(DaemonExtensions::none()).await?; + // A launched agent session is told where this daemon serves, so the + // `nodespace` commands run inside it reach this daemon. + shared.context.pty_manager.set_daemon_socket(sock.clone()); let (manager, bundle) = open_default_database( DatabaseManager::default_registry_path()?, &incompatible_database::marker_path()?, @@ -1756,6 +1759,9 @@ async fn serve_grpc(controller: tray::TrayController) -> Result<()> { install_shutdown_handler().context("Failed to install signal handlers")?; // _model_task: dropping a JoinHandle does not cancel the task in tokio — it detaches. let (shared, _model_task) = build_shared_services(DaemonExtensions::none()).await?; + // A launched agent session is told where this daemon serves, so the + // `nodespace` commands run inside it reach this daemon. + shared.context.pty_manager.set_daemon_socket(sock.clone()); let (manager, bundle) = open_default_database( DatabaseManager::default_registry_path()?, &incompatible_database::marker_path()?, @@ -1894,6 +1900,12 @@ async fn serve_headless() -> Result<()> { let shutdown = install_shutdown_handler().context("Failed to install signal handlers")?; // _model_task: dropping a JoinHandle does not cancel the task in tokio — it detaches. let (shared, _model_task) = build_shared_services(DaemonExtensions::none()).await?; + // A launched agent session is told where this daemon serves, so the + // `nodespace` commands run inside it reach this daemon. + shared + .context + .pty_manager + .set_daemon_socket(std::path::PathBuf::from(&name)); let (manager, bundle) = open_default_database( DatabaseManager::default_registry_path()?, &incompatible_database::marker_path()?, @@ -2006,6 +2018,12 @@ async fn serve_grpc(controller: tray::TrayController) -> Result<()> { install_shutdown_handler().context("Failed to install signal handlers")?; // _model_task: dropping a JoinHandle does not cancel the task in tokio — it detaches. let (shared, _model_task) = build_shared_services(DaemonExtensions::none()).await?; + // A launched agent session is told where this daemon serves, so the + // `nodespace` commands run inside it reach this daemon. + shared + .context + .pty_manager + .set_daemon_socket(std::path::PathBuf::from(&name)); let (manager, bundle) = open_default_database( DatabaseManager::default_registry_path()?, &incompatible_database::marker_path()?, diff --git a/packages/daemon/src/services/agent_session_service.rs b/packages/daemon/src/services/agent_session_service.rs index 76434c276..71fbde51c 100644 --- a/packages/daemon/src/services/agent_session_service.rs +++ b/packages/daemon/src/services/agent_session_service.rs @@ -16,11 +16,9 @@ use std::str::FromStr; use std::sync::Arc; use chrono::Utc; -use nodespace_agent::agent_catalog::context_assembly::GraphContextAssembler; use nodespace_agent::agent_types::AgentType; -use nodespace_agent::pty::{ - detect_all_agents, find_harness_session_id, PtySession, PtySessionManager, -}; +use nodespace_agent::pty::{PtySession, PtySessionManager, SessionLaunch, LAUNCHED_FOR_ENV_VAR}; +use nodespace_core::models::{CoreNodeType, ProjectNodeUpdate}; use nodespace_core::services::NodeService as CoreNodeService; use tokio::sync::broadcast::error::RecvError; use tonic::{Request, Response, Status}; @@ -29,9 +27,9 @@ use uuid::Uuid; use crate::nodespace::{ agent_session_service_server::AgentSessionService, AgentAvailability, CheckAvailabilityRequest, CheckAvailabilityResponse, LaunchSessionRequest, LaunchSessionResponse, ListSessionsRequest, - ListSessionsResponse, OutputChunk, ResizeRequest, ResizeResponse, SessionInfo, - StreamOutputRequest, TerminateSessionRequest, TerminateSessionResponse, WriteInputRequest, - WriteInputResponse, + ListSessionsResponse, OutputChunk, ReportHarnessSessionRequest, ReportHarnessSessionResponse, + ResizeRequest, ResizeResponse, SessionInfo, StreamOutputRequest, TerminateSessionRequest, + TerminateSessionResponse, WriteInputRequest, WriteInputResponse, }; use crate::services::capture_service::{finalize_capture, CompletedSession, SessionSummarizer}; use crate::services::settings_service::{read_capture_settings, CaptureConfig}; @@ -40,7 +38,6 @@ use crate::services::settings_service::{read_capture_settings, CaptureConfig}; #[derive(Clone)] pub struct AgentSessionHandler { manager: Arc, - assembler: Arc, node_service: Arc, config_path: PathBuf, /// Derives a finished session's summary, when capture saves one. @@ -50,14 +47,12 @@ pub struct AgentSessionHandler { impl AgentSessionHandler { pub fn new( manager: Arc, - assembler: Arc, node_service: Arc, config_path: PathBuf, summarizer: Arc, ) -> Self { Self { manager, - assembler, node_service, config_path, summarizer, @@ -75,11 +70,106 @@ impl AgentSessionHandler { /// is process-global (the same `Arc` backs every database), so PTY /// operations (stream/write/resize/terminate/list) act on the shared session /// set regardless of database. Launch is the sole handler that reads the - /// per-database context assembler and node service (for context + capture), + /// per-database node service (for the project's folder and for capture), /// so it follows the targeted database. async fn route(&self, request: &Request) -> Result { crate::db_routing::route_or_self(self, request, |s| &s.agent_session).await } + + /// The folder a session launched for `project_id` runs in: the project's + /// `checkout_path`, its folder on this machine (ADR-093 §8). + /// + /// `given` is the folder the launch names, for a project that has none + /// yet or whose folder is being changed. It is checked and then stored on + /// the project, so the next launch needs no folder named. + async fn project_folder( + &self, + project_id: &str, + given: Option<&str>, + ) -> Result { + let project = self + .node_service + .get_node(project_id) + .await + .map_err(|e| Status::internal(format!("reading project {project_id} failed: {e}")))? + .ok_or_else(|| Status::not_found(format!("project not found: {project_id}")))?; + let is_project = self + .node_service + .type_is_a(&project.node_type, CoreNodeType::Project) + .await + .map_err(|e| Status::internal(format!("reading project {project_id} failed: {e}")))?; + if !is_project { + return Err(Status::invalid_argument(format!( + "{project_id} is a {}, not a project", + project.node_type + ))); + } + + let stored = project + .properties + .get(CoreNodeType::Project.as_str()) + .and_then(|bucket| bucket.get("checkout_path")) + .and_then(|value| value.as_str()) + .filter(|path| !path.is_empty()); + + let Some(given) = given else { + let stored = stored.ok_or_else(|| { + Status::failed_precondition(format!( + "project '{}' has no folder on this machine: name its checkout folder to launch a session for it", + project.content + )) + })?; + // The stored value can be written by any update of the project, + // so it gets the checks a named folder gets. + return checked_folder(stored).map_err(|problem| { + Status::failed_precondition(format!( + "the folder of project '{}' cannot be used: {problem}", + project.content + )) + }); + }; + + let folder = checked_folder(given).map_err(Status::invalid_argument)?; + if stored != Some(given) { + self.node_service + .update_project_node( + project_id, + project.version, + ProjectNodeUpdate { + checkout_path: Some(Some(given.to_string())), + ..Default::default() + }, + ) + .await + .map_err(|e| match e { + // The project changed between the read and the write. + nodespace_core::services::NodeServiceError::VersionConflict { .. } => { + Status::aborted(format!( + "project {project_id} changed while its folder was being stored: launch again" + )) + } + other => Status::internal(format!( + "storing the folder of project {project_id} failed: {other}" + )), + })?; + } + Ok(folder) + } +} + +/// A project's folder as a path a session can run in: absolute, and an +/// existing directory on this machine. The error says which it is not. +fn checked_folder(folder: &str) -> Result { + let path = PathBuf::from(folder); + if !path.is_absolute() { + return Err(format!( + "a project's folder is an absolute path, got: {folder}" + )); + } + if !path.is_dir() { + return Err(format!("not a directory on this machine: {folder}")); + } + Ok(path) } #[tonic::async_trait] @@ -89,6 +179,7 @@ impl AgentSessionService for AgentSessionHandler { request: Request, ) -> Result, Status> { let this = self.route(&request).await?; + let database_id = crate::db_routing::routed_database_id(&request).await?; let req = request.into_inner(); let agent_type = parse_agent_type(&req.agent_type).map_err(Status::invalid_argument)?; @@ -98,13 +189,18 @@ impl AgentSessionService for AgentSessionHandler { // but the RPC guard ensures the daemon never spawns a doomed process // even if the UI check is skipped or stale. // - // detect_all_agents() spawns /usr/libexec/path_helper via + // Building the search path spawns /usr/libexec/path_helper via // std::process::Command on macOS — a blocking call that must not run // on a Tokio executor thread. - let availability = tokio::task::spawn_blocking(move || { - detect_all_agents() - .into_iter() - .find(|a| a.agent_type == agent_type) + let manager = this.manager.clone(); + // The path the agent is found ready on is the one it is launched + // with: built once, used for both. + let (availability, search_path) = tokio::task::spawn_blocking(move || { + let (agents, path) = manager.detect_agents_with_path(); + ( + agents.into_iter().find(|a| a.agent_type == agent_type), + path, + ) }) .await .map_err(|e| Status::internal(format!("agent detection task panicked: {e}")))?; @@ -133,9 +229,47 @@ impl AgentSessionService for AgentSessionHandler { } } + // Everything the launch names is checked before the project's folder + // is stored: a refused launch changes nothing. + if req.project_id.is_none() && req.project_folder.is_some() { + return Err(Status::invalid_argument( + "a project folder was named without a project", + )); + } + if let Some(task_id) = req.task_id.as_deref() { + this.node_service + .get_node(task_id) + .await + .map_err(|e| Status::internal(format!("reading {task_id} failed: {e}")))? + .ok_or_else(|| Status::not_found(format!("node not found: {task_id}")))?; + } + let working_dir = match req.project_id.as_deref() { + Some(project_id) => Some( + this.project_folder(project_id, req.project_folder.as_deref()) + .await?, + ), + None => None, + }; + // Beside the daemon's other state, so it follows NODESPACE_HOME. + let session_folders = crate::nodespace_dir() + .map_err(|e| Status::internal(format!("resolving the NodeSpace home failed: {e}")))? + .join("agent-sessions"); + + let launch = SessionLaunch { + agent_type, + initial_prompt: req.prompt, + session_folders: Some(session_folders), + env: session_environment( + database_id, + this.manager.daemon_socket(), + req.task_id.as_deref().or(req.node_id.as_deref()), + ), + node_id: req.node_id, + working_dir, + }; let id = this .manager - .launch(agent_type, req.prompt, req.node_id, &this.assembler) + .launch(launch, search_path) .await .map_err(|e| Status::internal(format!("launch session failed: {e}")))?; @@ -151,6 +285,10 @@ impl AgentSessionService for AgentSessionHandler { .as_ref() .map(|s| s.started_at.timestamp()) .unwrap_or_else(current_unix_secs); + let working_dir = session + .as_ref() + .map(|s| s.working_dir.to_string_lossy().into_owned()) + .unwrap_or_default(); // Apply requested dimensions when the caller passed non-zero values. // Zero on either axis means "keep the engine default" (80x24); @@ -213,17 +351,12 @@ impl AgentSessionService for AgentSessionHandler { }; let capture = session.snapshot_capture().await; - // Only a session with a node has anywhere to record the id. - let harness_session_id = match session.node_id { - Some(_) => harness_session_id(&session).await, - None => None, - }; let completed = CompletedSession { id: session.id, node_id: session.node_id.clone(), ended_at: Utc::now(), exit_status, - harness_session_id, + harness_session_id: session.harness_session_id(), }; // Deriving the summary can wait a long time for the model; // the finished PTY session is not kept alive for it. @@ -250,6 +383,30 @@ impl AgentSessionService for AgentSessionHandler { Ok(Response::new(LaunchSessionResponse { session_id: id.to_string(), created_at, + working_dir, + })) + } + + async fn report_harness_session( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let id = parse_session_id(&req.session_id).map_err(Status::invalid_argument)?; + let harness_session_id = req.harness_session_id.trim(); + if harness_session_id.is_empty() { + return Err(Status::invalid_argument("harness_session_id is empty")); + } + let session = self + .manager + .get(&id) + .await + .ok_or_else(|| Status::not_found(format!("session not found: {id}")))?; + + session.report_harness_session_id(harness_session_id.to_string()); + + Ok(Response::new(ReportHarnessSessionResponse { + node_id: session.node_id.clone(), })) } @@ -409,10 +566,11 @@ impl AgentSessionService for AgentSessionHandler { &self, _request: Request, ) -> Result, Status> { - // detect_all_agents() spawns /usr/libexec/path_helper via + // Building the search path spawns /usr/libexec/path_helper via // std::process::Command on macOS — a blocking call that must not run // on a Tokio executor thread. - let results = tokio::task::spawn_blocking(detect_all_agents) + let manager = self.manager.clone(); + let results = tokio::task::spawn_blocking(move || manager.detect_agents()) .await .map_err(|e| Status::internal(format!("agent detection task panicked: {e}")))?; let agents: Vec = results @@ -444,48 +602,47 @@ fn parse_session_id(raw: &str) -> Result { Uuid::from_str(raw).map_err(|e| format!("invalid session_id '{raw}': {e}")) } -/// Convert the proto's `agent_type` string into the canonical [`AgentType`]. -/// -/// The only accepted form is the kebab-case serde representation of -/// [`AgentType`] (`"claude-code"`, `"codex"`, `"antigravity-cli"`, `"pi"`, -/// `"open-code"`). Snake-case is rejected — CLAUDE.md is explicit that -/// greenfield code carries no backward-compat aliases. +/// Convert the proto's `agent_type` string into the [`AgentType`] it names. +/// The only accepted form is the agent's id ([`AgentType::id`]). fn parse_agent_type(raw: &str) -> Result { - serde_json::from_value::(serde_json::Value::String(raw.to_string())).map_err(|_| { + AgentType::from_id(raw).ok_or_else(|| { format!( - "unknown agent_type '{raw}'; expected one of: claude-code, codex, antigravity-cli, pi, open-code" + "unknown agent_type '{raw}'; expected one of: {}", + AgentType::ALL.map(AgentType::id).join(", ") ) }) } fn agent_type_to_string(agent_type: AgentType) -> String { - // serde serialization mirrors the kebab-case form parse_agent_type accepts. - // `AgentType` is in-workspace and closed, so the debug-format fallback - // is genuinely unreachable today — it exists as defense-in-depth, not as - // a graceful-degradation path. The right way to add a new variant is to - // extend parse_agent_type / agent_type_to_string together; do NOT rely - // on the fallback to ship a new variant. - serde_json::to_value(agent_type) - .ok() - .and_then(|v| v.as_str().map(str::to_string)) - .unwrap_or_else(|| format!("{agent_type:?}")) + agent_type.id().to_string() } -/// The id the harness that ran in `session` gave its own conversation: the -/// one its resume flag takes. Read from the harness's session store under the -/// user's home directory, which is where the harness wrote it (a PTY child is -/// given `HOME` and none of the variables that would move the store). -async fn harness_session_id(session: &PtySession) -> Option { - let home = dirs::home_dir()?; - let agent_type = session.agent_type; - let session_dir = session.session_dir.clone(); - let started_at = session.started_at; - tokio::task::spawn_blocking(move || { - find_harness_session_id(agent_type, &home, &session_dir, started_at) - }) - .await - .ok() - .flatten() +/// What NodeSpace sets in a launched session's environment, on top of the +/// allowlist the PTY engine forwards: the database the session was launched +/// from and the socket of the daemon that launched it, so a `nodespace` +/// command inside reaches both, and what the session was launched for. +/// +/// A value that is not known is left out, never set empty: the CLI reads an +/// absent variable as "the default". +fn session_environment( + database_id: Option, + daemon_socket: Option<&std::path::Path>, + launched_for: Option<&str>, +) -> Vec<(String, String)> { + let mut env = Vec::new(); + if let Some(database_id) = database_id { + env.push((nodespace_proto::DATABASE_ENV_VAR.to_string(), database_id)); + } + if let Some(socket) = daemon_socket { + env.push(( + nodespace_proto::socket::SOCKET_ENV_VAR.to_string(), + socket.to_string_lossy().into_owned(), + )); + } + if let Some(launched_for) = launched_for { + env.push((LAUNCHED_FOR_ENV_VAR.to_string(), launched_for.to_string())); + } + env } /// Apply a resize to an already-located [`PtySession`]. @@ -524,11 +681,11 @@ mod tests { ); assert_eq!(parse_agent_type("codex").unwrap(), AgentType::Codex); assert_eq!( - parse_agent_type("antigravity-cli").unwrap(), - AgentType::AntigravityCli + parse_agent_type("antigravity").unwrap(), + AgentType::Antigravity ); assert_eq!(parse_agent_type("pi").unwrap(), AgentType::Pi); - assert_eq!(parse_agent_type("open-code").unwrap(), AgentType::OpenCode); + assert_eq!(parse_agent_type("opencode").unwrap(), AgentType::OpenCode); } #[test] @@ -559,7 +716,7 @@ mod tests { for t in [ AgentType::ClaudeCode, AgentType::Codex, - AgentType::AntigravityCli, + AgentType::Antigravity, AgentType::Pi, AgentType::OpenCode, ] { diff --git a/packages/daemon/src/services/assembly.rs b/packages/daemon/src/services/assembly.rs index 414c7e336..e13f8f500 100644 --- a/packages/daemon/src/services/assembly.rs +++ b/packages/daemon/src/services/assembly.rs @@ -10,7 +10,6 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, OnceLock}; use anyhow::{Context, Result}; -use nodespace_agent::agent_catalog::context_assembly::GraphContextAssembler; use nodespace_agent::prompt_assembler::PromptAssembler; use nodespace_agent::pty::PtySessionManager; use nodespace_agent::skill_pipeline::{link_seeded_skills, seed_skill_nodes, seed_tool_nodes}; @@ -487,10 +486,6 @@ pub async fn build_database_services( ) }); - let assembler = Arc::new(GraphContextAssembler::new( - node_service.clone(), - embedding_svc_state.clone(), - )); // Resolved through `nodespace_dir` so it follows NODESPACE_HOME, exactly as // the database and the ADR-053 registry do. Reading it from the real home // instead left an isolated daemon serving a temp database while taking its @@ -498,7 +493,6 @@ pub async fn build_database_services( let capture_config_path = crate::nodespace_dir()?.join("daemon.toml"); let agent_session = AgentSessionHandler::new( shared.pty_manager.clone(), - assembler, node_service.clone(), capture_config_path, Arc::new(LocalModelSummarizer::new(shared.local_agent.clone())), @@ -636,13 +630,8 @@ async fn build_unrouted_services(shared: &SharedContext) -> Result, pub ended_at: DateTime, pub exit_status: ExitStatus, - /// The id the harness recorded for its own conversation, which its resume - /// flag takes. `None` for a harness that records none, or one that exited - /// before starting a conversation. + /// The id the harness gave its own conversation, which its resume flag + /// takes, as its plugin reported it. `None` for a harness with no plugin, + /// or one that exited before its plugin reported. pub harness_session_id: Option, } @@ -712,82 +712,10 @@ mod tests { } } - /// The stored session id is the one the harness recorded in its own - /// session store, for each harness that has one. - #[tokio::test] - async fn the_harnesss_own_session_id_is_stored_for_claude_code_and_codex() { - use nodespace_agent::agent_types::AgentType; - use nodespace_agent::pty::find_harness_session_id; - - const CODEX_SESSION_ID: &str = "7b1e2a44-3c9d-4f6a-8e21-5a0c9d3e7b02"; - - let tmp = tempfile::TempDir::new().unwrap(); - let node_service = test_node_service(&tmp).await; - - // A home directory holding each harness's session store, with one - // conversation recorded for the session's working directory. - let home = tmp.path().canonicalize().unwrap().join("home"); - let session_dir = home.join(".nodespace").join("agent-sessions").join("s1"); - std::fs::create_dir_all(&session_dir).unwrap(); - let claude_project: String = session_dir - .to_string_lossy() - .chars() - .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }) - .collect(); - let claude_dir = home.join(".claude").join("projects").join(claude_project); - std::fs::create_dir_all(&claude_dir).unwrap(); - std::fs::write( - claude_dir.join(format!("{HARNESS_SESSION_ID}.jsonl")), - "{}\n", - ) - .unwrap(); - let codex_dir = home - .join(".codex") - .join("sessions") - .join("2026") - .join("01") - .join("01"); - std::fs::create_dir_all(&codex_dir).unwrap(); - std::fs::write( - codex_dir.join(format!( - "rollout-2026-01-01T11-00-00-{CODEX_SESSION_ID}.jsonl" - )), - format!( - "{}\n", - json!({ - "type": "session_meta", - "payload": { "id": CODEX_SESSION_ID, "cwd": session_dir } - }) - ), - ) - .unwrap(); - let started_at = Utc::now() - chrono::Duration::minutes(5); - - for (agent_type, agent, expected) in [ - (AgentType::ClaudeCode, "claude-code", HARNESS_SESSION_ID), - (AgentType::Codex, "codex", CODEX_SESSION_ID), - ] { - let node_id = create_chat(&node_service, "ai-chat-pty", agent).await; - let mut session = make_session(); - session.node_id = Some(node_id.clone()); - session.harness_session_id = - find_harness_session_id(agent_type, &home, &session_dir, started_at); - - finalize_capture( - &session, - &SessionCapture::new(), - &node_service, - ¬_capturing(), - &RecordingSummarizer::unavailable(), - ) - .await - .unwrap(); - - let node = node_service.get_node(&node_id).await.unwrap().unwrap(); - assert_eq!(node.properties["ai-chat-pty"]["session_id"], expected); - } - - // The field the id is stored in never leaves the machine. + /// The field the harness's session id is stored in never leaves the + /// machine: the id names state on it. + #[test] + fn the_session_id_field_is_machine_bound() { let session_id_field = nodespace_core::models::core_schemas::get_core_schemas() .into_iter() .find(|schema| schema.envelope.id == CoreNodeType::AiChatPty.as_str()) diff --git a/packages/daemon/src/services/node_service.rs b/packages/daemon/src/services/node_service.rs index 0b2127868..1a92fddfc 100644 --- a/packages/daemon/src/services/node_service.rs +++ b/packages/daemon/src/services/node_service.rs @@ -1981,6 +1981,7 @@ impl GrpcNodeService for NodeServiceImpl { .map_err(Status::invalid_argument)?, repository: optional_json_clear(req.repository, "repository") .map_err(Status::invalid_argument)?, + checkout_path: optional_string_clear(req.checkout_path), }; match this @@ -5788,6 +5789,7 @@ mod tests { }), end_date: None, repository: None, + checkout_path: None, })) .await .expect_err("malformed date must be rejected"); @@ -5992,6 +5994,7 @@ mod tests { start_date: None, end_date: None, repository, + checkout_path: None, }) }; let typed = |id: &'static str| { @@ -6353,6 +6356,7 @@ mod tests { start_date: None, end_date: None, repository: None, + checkout_path: None, })) .await .expect_err("stale version must conflict"); diff --git a/packages/daemon/tests/it/agent_session_grpc.rs b/packages/daemon/tests/it/agent_session_grpc.rs index a5ff9b6d8..e5841d057 100644 --- a/packages/daemon/tests/it/agent_session_grpc.rs +++ b/packages/daemon/tests/it/agent_session_grpc.rs @@ -16,7 +16,6 @@ use std::sync::Arc; use std::time::Duration; -use nodespace_agent::agent_catalog::context_assembly::GraphContextAssembler; use nodespace_agent::pty::{PtySession, PtySessionManager}; use nodespace_core::{NodeService as CoreNodeService, SqliteStore}; use nodespace_daemon::nodespace::{ @@ -52,8 +51,8 @@ impl SessionSummarizer for NoSummarizer { async fn spawn_test_daemon() -> (Client, Arc, oneshot::Sender<()>, TempDir) { let tempdir = TempDir::new().expect("tempdir"); - // The assembler holds a NodeService handle even though our tests bypass - // LaunchSession. SqliteStore is the only realistic way to produce one. + // The handler reads projects and writes a session's end through a + // NodeService. SqliteStore is the only realistic way to produce one. let mut store = Arc::new( SqliteStore::new(tempdir.path().join("daemon-db")) .await @@ -62,14 +61,9 @@ async fn spawn_test_daemon() -> (Client, Arc, oneshot::Sender let node_service = Arc::new(CoreNodeService::new(&mut store).await.expect("NodeService")); let manager = Arc::new(PtySessionManager::new()); - let assembler = Arc::new(GraphContextAssembler::new( - node_service.clone(), - Arc::new(tokio::sync::RwLock::new(None)), - )); let capture_config_path = tempdir.path().join("daemon.toml"); let handler = AgentSessionHandler::new( manager.clone(), - assembler, node_service, capture_config_path, Arc::new(NoSummarizer), diff --git a/packages/desktop-app/app-lib/src/commands/agent_session.rs b/packages/desktop-app/app-lib/src/commands/agent_session.rs index 14b11cc4d..3d844153e 100644 --- a/packages/desktop-app/app-lib/src/commands/agent_session.rs +++ b/packages/desktop-app/app-lib/src/commands/agent_session.rs @@ -90,6 +90,8 @@ impl StreamingTaskRegistry { pub struct LaunchSessionResult { pub session_id: String, pub created_at: i64, + /// The directory the session runs in. + pub working_dir: String, } #[derive(Debug, Serialize, Clone)] @@ -155,6 +157,13 @@ pub struct LaunchSessionInput { /// ID of the `ai-chat-pty` node this PTY session is a view onto. Capture /// backfills this node at session end. See ADR-088. pub node_id: Option, + /// ID of the project the session is launched for: it runs in that + /// project's folder on this machine (ADR-093 §8). + pub project_id: Option, + /// The project's folder on this machine, when it is being set or changed. + pub project_folder: Option, + /// ID of the task the session is launched to work on. + pub task_id: Option, } // --------------------------------------------------------------------------- @@ -237,6 +246,9 @@ pub async fn launch_session( cols: input.cols, rows: input.rows, node_id: input.node_id, + project_id: input.project_id, + project_folder: input.project_folder, + task_id: input.task_id, })) .await .map_err(status_to_command_error)?; @@ -244,6 +256,7 @@ pub async fn launch_session( let inner = resp.into_inner(); let session_id = inner.session_id.clone(); let created_at = inner.created_at; + let working_dir = inner.working_dir.clone(); // Obtain the client before spawning (State<'_> has a non-'static lifetime). let stream_client = client.agent_session_client().await; @@ -264,6 +277,7 @@ pub async fn launch_session( Ok(LaunchSessionResult { session_id, created_at, + working_dir, }) } diff --git a/packages/desktop-app/app-lib/src/commands/nodes.rs b/packages/desktop-app/app-lib/src/commands/nodes.rs index d1c3da7a5..581ccc192 100644 --- a/packages/desktop-app/app-lib/src/commands/nodes.rs +++ b/packages/desktop-app/app-lib/src/commands/nodes.rs @@ -1292,6 +1292,7 @@ pub async fn update_project_node( start_date: timestamp_clear(update.start_date), end_date: timestamp_clear(update.end_date), repository: json_clear(update.repository, "repository")?, + checkout_path: string_clear(update.checkout_path), }; let resp = c .update_project_node(Request::new(req)) diff --git a/packages/desktop-app/src/lib/components/viewers/ai-chat-pty-session.svelte b/packages/desktop-app/src/lib/components/viewers/ai-chat-pty-session.svelte index acbe9d7a0..2e17a34aa 100644 --- a/packages/desktop-app/src/lib/components/viewers/ai-chat-pty-session.svelte +++ b/packages/desktop-app/src/lib/components/viewers/ai-chat-pty-session.svelte @@ -4,18 +4,26 @@ named like ChatMessage/ChatInput), so it carries no *Node/*Viewer/*View suffix. A PTY agent session IS an `ai-chat-pty` node (ADR-088). This helper renders a - launch config (harness picker + Launch) when no session is running, the - embedded xterm terminal (via pty-terminal.svelte) while it runs, and a + launch config (harness picker, project, Launch) when no session is running, + the embedded xterm terminal (via pty-terminal.svelte) while it runs, and a read-only summary once it ends. The node already exists; capture backfills it at session end via the node_id passed to launch. + + A session launched for a project runs in that project's folder on this + machine (ADR-093 §8). The folder is asked for here the first time and stored + on the project as its machine-bound `checkout_path`; the daemon checks that + it exists before it stores it. -->