Implement AgentSessionService gRPC handler (closes #1119)

packages/agent/Cargo.toml

@@ -34,12 +34,26 @@ tempfile = "3"
 libc = "0.2"
 
 [features]
-# Exposes test-only hooks (e.g. `LocalAgentService::set_system_prompt` and
-# `AgentSession::system_prompt_override`) so integration tests in `tests/`
-# can inject prebuilt prompts without a live database. The `tests/` binaries
-# link against the library in non-test mode, so `#[cfg(test)]` alone is not
-# enough — this feature gate is what keeps the hook out of production builds
-# while making it visible to integration tests run with `--features testing`.
+# Exposes test-only hooks (e.g. `LocalAgentService::set_system_prompt`,
+# `AgentSession::system_prompt_override`, `PtySession::launch_for_test`) so
+# integration tests in `tests/` can inject prebuilt prompts and spawn shell
+# utilities without a real agent binary on PATH. The `tests/` binaries link
+# against the library in non-test mode, so `#[cfg(test)]` alone is not
+# enough — this feature gate is what keeps the hooks out of production
+# builds while making them visible to integration tests run with
+# `--features testing`.
+#
+# Footgun: Cargo unifies features across dep/dev-dep within a single
+# downstream package. If a sibling crate (e.g. nodespace-daemon) depends on
+# nodespace-agent as a regular dependency *and* as a dev-dependency with
+# `features = ["testing"]`, then during `cargo test -p that-crate` the
+# downstream's *production* code links against this crate with the
+# `testing` feature enabled. That is currently safe — every gate behind
+# `testing` only widens visibility (pub(crate) → pub) without changing
+# runtime behavior. Keep it that way: do not put `#[cfg(feature =
+# "testing")]` on code that mutates state, swaps implementations, or skips
+# safety checks. Test-only stubs belong in test-only modules in the
+# downstream crate, not behind this feature flag.
 testing = []
 
 [dev-dependencies]

packages/agent/src/pty/manager.rs

@@ -237,8 +237,7 @@ mod tests {
         for _ in 0..10 {
             // `true` exits with status 0 essentially immediately. The watcher
             // very plausibly fires before the manager's auto-prune subscribes.
-            let session =
-                PtySession::launch_for_test("true", vec![]).expect("launch true session");
+            let session = PtySession::launch_for_test("true", vec![]).expect("launch true session");
             manager.insert(session).await;
         }
         wait_for_len(&manager, 0, Duration::from_secs(3)).await;

packages/agent/src/pty/session.rs

@@ -358,12 +358,16 @@ fn spawn_reader_task(mut reader: Box<dyn Read + Send>, output_tx: broadcast::Sen
     });
 }
 
-#[cfg(test)]
+#[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.
-    pub(crate) fn launch_for_test(binary: &str, args: Vec<String>) -> anyhow::Result<Self> {
+    ///
+    /// Gated by the `testing` feature so it is reachable from integration
+    /// tests in sibling crates (e.g. `nodespace-daemon`) without being part
+    /// of the production surface.
+    pub fn launch_for_test(binary: &str, args: Vec<String>) -> anyhow::Result<Self> {
         let session_dir = tempfile::tempdir()?;
         let binary_path = which::which(binary)
             .map_err(|e| anyhow::anyhow!("test binary '{}' not on PATH: {}", binary, e))?;
@@ -415,7 +419,7 @@ impl PtySession {
     }
 
     /// Test-only accessor: where this session's temp dir lives.
-    pub(crate) fn session_dir_path(&self) -> &std::path::Path {
+    pub fn session_dir_path(&self) -> &std::path::Path {
         self._session_dir.path()
     }
 }

packages/cli/src/commands/node.rs

@@ -102,6 +102,8 @@ async fn create(
             properties: String::new(),
             collection: String::new(),
             lifecycle_status: String::new(),
+            id: String::new(),
+            insert_after_node_id: String::new(),
         })
         .await
         .context("CreateNode RPC failed")?

packages/cli/tests/cli_integration.rs

@@ -108,6 +108,8 @@ async fn create_get_update_children_delete_round_trip() {
             properties: String::new(),
             collection: String::new(),
             lifecycle_status: String::new(),
+            id: String::new(),
+            insert_after_node_id: String::new(),
         })
         .await
         .expect("seed parent")

packages/daemon/Cargo.toml

@@ -23,6 +23,8 @@ serde_json = { workspace = true }
 async-trait = { workspace = true }
 chrono = { workspace = true }
 nodespace-core = { workspace = true }
+nodespace-agent = { workspace = true }
+uuid = { workspace = true }
 
 # System tray (ADR-031): the daemon owns the tray icon, not the Tauri app.
 # tray-icon ships an event loop via `tao`; we run it on the main thread on
@@ -39,3 +41,10 @@ protoc-bin-vendored = "3"
 
 [dev-dependencies]
 tempfile = "3.0"
+# Integration tests need PtySession::launch_for_test to spawn shell utilities
+# (echo, cat) without a real agent binary. The `testing` feature exposes that
+# helper as part of the agent crate's public surface.
+nodespace-agent = { workspace = true, features = ["testing"] }
+tokio = { workspace = true, features = ["full", "test-util"] }
+tokio-stream = "0.1"
+tonic = "0.12"

packages/daemon/proto/agent_session_service.proto

@@ -4,9 +4,10 @@ package nodespace;
 
 // AgentSessionService manages PTY/terminal-style agent sessions.
 //
-// A session represents a running agent process with a pseudo-terminal attached.
-// Clients can launch sessions, stream output, write input, resize the terminal,
-// and terminate sessions.
+// A session represents an external agent CLI (Claude Code, Codex, Gemini CLI,
+// Pi, OpenCode) running inside a pseudo-terminal. Clients can launch sessions,
+// stream output, write input, resize the terminal, list active sessions, and
+// terminate sessions.
 service AgentSessionService {
   rpc LaunchSession(LaunchSessionRequest) returns (LaunchSessionResponse);
   rpc StreamOutput(StreamOutputRequest) returns (stream OutputChunk);
@@ -20,29 +21,28 @@ service AgentSessionService {
 // LaunchSession
 // ---------------------------------------------------------------------------
 
-// EnvVar is a single environment variable key-value pair.
-message EnvVar {
-  string key = 1;
-  string value = 2;
-}
-
 message LaunchSessionRequest {
-  // Command to execute, e.g. "/usr/bin/python3" or "node"
-  string command = 1;
-  // Arguments passed to the command
-  repeated string args = 2;
-  // Working directory for the process; empty = server default
-  string cwd = 3;
-  // Additional environment variables merged with the server's environment
-  repeated EnvVar env = 4;
-  // Initial terminal dimensions
-  uint32 cols = 5;
-  uint32 rows = 6;
+  // Catalogued agent identifier (kebab-case): "claude-code" | "codex" |
+  // "gemini-cli" | "pi" | "open-code". Matches the AgentType enum's serde
+  // form in packages/agent/src/agent_types.rs.
+  string agent_type = 1;
+  // Initial prompt passed as a CLI argument when the agent supports it
+  // (e.g. `claude "<prompt>"`). Omit to launch the agent in its default
+  // interactive mode.
+  optional string prompt = 2;
+  // Initial terminal dimensions. Valid range: [1, 65535]. Zero on either
+  // axis falls back to the engine default (80x24); callers are expected to
+  // call ResizeTerminal as soon as they know the real terminal geometry.
+  // (ResizeTerminal, by contrast, rejects zero — its semantics are "apply
+  // exactly this size", not "use the default".)
+  uint32 cols = 3;
+  uint32 rows = 4;
 }
 
 message LaunchSessionResponse {
+  // UUID assigned to the new session.
   string session_id = 1;
-  // Unix timestamp (seconds) when the session was created
+  // Unix timestamp (seconds) when the session was created.
   int64 created_at = 2;
 }
 
@@ -54,19 +54,17 @@ message StreamOutputRequest {
   string session_id = 1;
 }
 
-// StreamKind distinguishes the output channel.
-enum StreamKind {
-  STREAM_KIND_UNSPECIFIED = 0;
-  STREAM_KIND_STDOUT = 1;
-  STREAM_KIND_STDERR = 2;
-}
-
+// OutputChunk is one slice of raw bytes read from the PTY master.
+//
+// PTYs merge stdout and stderr into a single byte stream — there is no way to
+// distinguish them at this layer. The bytes may contain partial UTF-8 sequences
+// and terminal escape codes; consumers are expected to render them the way a
+// terminal would.
 message OutputChunk {
-  // Raw bytes from the PTY (may contain ANSI escape sequences)
+  // Raw bytes read from the PTY (may include ANSI escape sequences).
   bytes data = 1;
-  StreamKind stream_kind = 2;
-  // Unix timestamp in milliseconds when this chunk was produced
-  int64 timestamp_ms = 3;
+  // Unix timestamp in milliseconds when this chunk was produced.
+  int64 timestamp_ms = 2;
 }
 
 // ---------------------------------------------------------------------------
@@ -75,13 +73,18 @@ message OutputChunk {
 
 message WriteInputRequest {
   string session_id = 1;
-  // Raw bytes to write (keystrokes, commands, etc.)
+  // Raw bytes to write (keystrokes, commands, etc.). May include control
+  // characters (Ctrl-C as 0x03, etc.) — they pass through to the PTY verbatim.
   bytes data = 2;
 }
 
 message WriteInputResponse {
-  // Number of bytes actually written
-  int32 bytes_written = 1;
+  // Number of bytes accepted by the PTY writer. The PTY engine writes the
+  // entire buffer atomically and flushes, so this always equals
+  // `data.len()` on success. `int64` because `data` is unbounded and a
+  // `usize → int32` cast could wrap on a 2 GiB+ payload (unlikely for a
+  // PTY, but the contract is the contract).
+  int64 bytes_written = 1;
 }
 
 // ---------------------------------------------------------------------------
@@ -90,66 +93,45 @@ message WriteInputResponse {
 
 message ResizeRequest {
   string session_id = 1;
+  // Terminal dimensions. Valid range: [1, 65535] (the underlying PTY uses
+  // u16). Zero is rejected with InvalidArgument — unlike LaunchSession,
+  // ResizeTerminal has no "default" semantic.
   uint32 cols = 2;
   uint32 rows = 3;
 }
 
-message ResizeResponse {
-  // Echoed dimensions after resize
-  uint32 cols = 1;
-  uint32 rows = 2;
-}
+message ResizeResponse {}
 
 // ---------------------------------------------------------------------------
 // TerminateSession
 // ---------------------------------------------------------------------------
 
 message TerminateSessionRequest {
   string session_id = 1;
-  // Signal number to send (e.g. 15 = SIGTERM, 9 = SIGKILL).
-  // Defaults to SIGTERM (15) when 0.
-  int32 signal = 2;
 }
 
 message TerminateSessionResponse {
   string session_id = 1;
+  // False when the session was not in the manager (already cleaned up by the
+  // natural-exit watcher); true when the handler actively terminated it.
   bool was_running = 2;
 }
 
 // ---------------------------------------------------------------------------
 // ListSessions
 // ---------------------------------------------------------------------------
 
-message ListSessionsRequest {
-  // Filter by status. Empty string = all sessions.
-  // Valid values: "running", "exited", "all"
-  string status_filter = 1;
-}
-
-// SessionStatus represents the current state of an agent session.
-enum SessionStatus {
-  SESSION_STATUS_UNSPECIFIED = 0;
-  SESSION_STATUS_RUNNING = 1;
-  SESSION_STATUS_EXITED = 2;
-}
+message ListSessionsRequest {}
 
 message SessionInfo {
   string session_id = 1;
-  string command = 2;
-  repeated string args = 3;
-  string cwd = 4;
-  SessionStatus status = 5;
-  // exit_code is unset while the session is still running; set to the process
-  // exit code once status = SESSION_STATUS_EXITED. optional int32 lets callers
-  // distinguish "running, no exit code yet" from "exited with code 0".
-  optional int32 exit_code = 6;
-  int64 created_at = 7; // Unix timestamp (seconds)
-  int64 exited_at = 8;  // Unix timestamp (seconds); 0 if still running
-  uint32 cols = 9;
-  uint32 rows = 10;
+  // Catalogued agent identifier (kebab-case), e.g. "claude-code".
+  string agent_type = 2;
+  // Unix timestamp (seconds) when the session was started.
+  int64 started_at = 3;
 }
 
 message ListSessionsResponse {
   repeated SessionInfo sessions = 1;
-  int32 count = 2;
+  uint32 count = 2;
 }

packages/daemon/src/lib.rs

@@ -30,4 +30,4 @@ pub use nodespace::node_service_client::NodeServiceClient;
 pub use nodespace::node_service_server::NodeServiceServer;
 pub use nodespace::{NodeData, SessionInfo};
 
-pub use services::NodeServiceImpl;
+pub use services::{AgentSessionHandler, NodeServiceImpl};

packages/daemon/src/main.rs

@@ -16,9 +16,13 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
+use nodespace_agent::acp::context_assembly::GraphContextAssembler;
+use nodespace_agent::pty::PtySessionManager;
 use nodespace_core::{NodeService as CoreNodeService, SurrealStore};
 use nodespace_daemon::tray::layer::TrayMetricsLayer;
-use nodespace_daemon::{tray, NodeServiceImpl, NodeServiceServer};
+use nodespace_daemon::{
+    tray, AgentSessionHandler, AgentSessionServiceServer, NodeServiceImpl, NodeServiceServer,
+};
 use tonic::transport::Server;
 
 /// Default address the daemon binds to. ADR-031 standardizes on
@@ -106,11 +110,12 @@ async fn serve_headless() -> Result<()> {
     tracing::info!(db_path = %db_path.display(), %addr, "Starting nodespaced (headless)");
 
     let shutdown = install_shutdown_handler().context("Failed to install signal handlers")?;
-    let service = build_node_service(&db_path).await?;
+    let services = build_services(&db_path).await?;
 
     tracing::info!(%addr, "gRPC server listening");
     Server::builder()
-        .add_service(NodeServiceServer::new(service))
+        .add_service(NodeServiceServer::new(services.node))
+        .add_service(AgentSessionServiceServer::new(services.agent_session))
         .serve_with_shutdown(addr, shutdown)
         .await
         .context("gRPC server terminated with error")?;
@@ -128,7 +133,7 @@ async fn serve_grpc(controller: tray::TrayController) -> Result<()> {
 
     let signal_shutdown =
         install_shutdown_handler().context("Failed to install signal handlers")?;
-    let service = build_node_service(&db_path).await?;
+    let services = build_services(&db_path).await?;
 
     // `TrayController` is `Clone`; one copy goes to the metrics layer, the
     // other drives the shutdown future.
@@ -143,15 +148,23 @@ async fn serve_grpc(controller: tray::TrayController) -> Result<()> {
     tracing::info!(%addr, "gRPC server listening");
     Server::builder()
         .layer(TrayMetricsLayer::new(controller))
-        .add_service(NodeServiceServer::new(service))
+        .add_service(NodeServiceServer::new(services.node))
+        .add_service(AgentSessionServiceServer::new(services.agent_session))
         .serve_with_shutdown(addr, combined_shutdown)
         .await
         .context("gRPC server terminated with error")?;
     Ok(())
 }
 
-/// Open the database and assemble the gRPC service implementation.
-async fn build_node_service(db_path: &std::path::Path) -> Result<NodeServiceImpl> {
+/// Bundle of gRPC service implementations registered by both server loops.
+struct DaemonServices {
+    node: NodeServiceImpl,
+    agent_session: AgentSessionHandler,
+}
+
+/// Open the database and assemble every gRPC service implementation the
+/// daemon exposes.
+async fn build_services(db_path: &std::path::Path) -> Result<DaemonServices> {
     if let Some(parent) = db_path.parent() {
         tokio::fs::create_dir_all(parent).await.with_context(|| {
             format!("Failed to create database parent dir: {}", parent.display())
@@ -171,8 +184,25 @@ async fn build_node_service(db_path: &std::path::Path) -> Result<NodeServiceImpl
     );
 
     // Embedding service is wired by a follow-up issue. The gRPC `SearchNodes`
-    // handler returns `Unavailable` until it is provided.
-    Ok(NodeServiceImpl::new(node_service, None))
+    // handler returns `Unavailable` until it is provided. The same handle
+    // would feed semantic expansion in [`GraphContextAssembler`]; for now the
+    // assembler runs without semantic neighbours (it logs and skips).
+    let embedding_service = None;
+
+    let node = NodeServiceImpl::new(node_service.clone(), embedding_service.clone());
+
+    // The PTY engine and context assembler are shared across every
+    // `AgentSessionService` RPC call. `PtySessionManager` is `Clone` but
+    // wrapping in `Arc` keeps the manager itself a single instance so all
+    // sessions live in one map.
+    let manager = Arc::new(PtySessionManager::new());
+    let assembler = Arc::new(GraphContextAssembler::new(node_service, embedding_service));
+    let agent_session = AgentSessionHandler::new(manager, assembler);
+
+    Ok(DaemonServices {
+        node,
+        agent_session,
+    })
 }
 
 /// Install the shutdown signal future at boot time so a failure to register