feat(tilth): convert to workspace and add acp-agent/client

- Convert tilth to a Cargo workspace.
- Add acp-agent and acp-client crates to the workspace.
- Add public search function to mcp.rs for non-MCP callers.

tilth/Cargo.toml

@@ -85,3 +85,7 @@ inherits = "release"
 lto = "thin"
 codegen-units = 8
 strip = false
+
+[workspace]
+members = [".", "crates/acp-agent", "crates/acp-client"]
+resolver = "2"

tilth/crates/acp-agent/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "tilth-acp-agent"
+version = "0.1.0"
+edition = "2021"
+description = "ACP (Agent Client Protocol) agent exposing tilth code intelligence — spawnable by Zed, JetBrains, and Eclipse Theia (aurelio-theia)."
+license = "MIT"
+
+[[bin]]
+name = "tilth-acp-agent"
+path = "src/main.rs"
+
+[dependencies]
+agent-client-protocol = "1.2.0"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "io-std", "time"] }
+serde_json = "1"
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
+tilth = { path = "../.." }

tilth/crates/acp-agent/src/main.rs

@@ -0,0 +1,166 @@
+//! tilth-acp-agent — ACP agent that exposes tilth code intelligence to any
+//! ACP client (Zed, JetBrains, Eclipse Theia / aurelio-theia).
+//!
+//! Wire-compatible with the TypeScript `AcpAgent` in aurelio-backend and the
+//! `shared/acp/golden/agent` corpus: protocol version 1, NDJSON over stdio.
+//!
+//! Phase 3 scope: a real initialize / session/new / session/prompt /
+//! session/cancel loop that routes the user's prompt through `tilth_search`
+//! and streams the result back as an agent message chunk.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use agent_client_protocol::schema::v1::{
+    AgentCapabilities, ContentBlock, ContentChunk, Implementation, InitializeRequest,
+    InitializeResponse, NewSessionRequest, NewSessionResponse, PromptCapabilities, PromptRequest,
+    PromptResponse, SessionNotification, SessionUpdate, StopReason, TextContent,
+};
+use agent_client_protocol::{
+    Agent, Channel, Client, ConnectionTo, Dispatch, RawJsonRpcMessage, Result,
+};
+use futures_util::StreamExt;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+
+static SESSION_SEQ: AtomicU64 = AtomicU64::new(1);
+
+fn text_of(blocks: &[ContentBlock]) -> String {
+    let mut out = String::new();
+    for b in blocks {
+        if let ContentBlock::Text(t) = b {
+            if !out.is_empty() {
+                out.push('\n');
+            }
+            out.push_str(&t.text);
+        }
+    }
+    out
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Transport-agnostic NDJSON pump. We own both ends of stdio so stdin EOF is
+    // an explicit signal: the Rust SDK's server-mode `connect_to(Stdio)` never
+    // resolves on a clean inbound EOF (its `run_until` waits on a
+    // `future::pending()` foreground), unlike the TS AgentSideConnection which
+    // exits on EOF. A `Channel::duplex` + two pumps gives us a clean, race-free
+    // termination path for pipes AND regular-file redirection.
+    let (agent_end, pump_end) = Channel::duplex();
+    let Channel {
+        rx: agent_to_us,
+        tx: us_to_agent,
+    } = pump_end;
+
+    // stdin -> agent (requests). On EOF this task returns; the agent is kept
+    // alive (fire-and-forget) so it can finish flushing the in-flight response
+    // during the grace window below. We do NOT close us_to_agent ourselves —
+    // the agent's connect_to is server-mode and never returns on a clean
+    // inbound EOF (same `run_until`/`future::pending()` footgun as Stdio), so
+    // waiting on it would hang. Instead we terminate the process after a grace.
+    let stdin_pump = tokio::spawn(async move {
+        let stdin = BufReader::new(tokio::io::stdin());
+        let mut lines = stdin.lines();
+        while let Ok(Some(line)) = lines.next_line().await {
+            if line.trim().is_empty() {
+                continue;
+            }
+            let item = match serde_json::from_str::<RawJsonRpcMessage>(&line) {
+                Ok(m) => Ok(m),
+                Err(_) => Err(agent_client_protocol::Error::parse_error()),
+            };
+            if us_to_agent.unbounded_send(item).is_err() {
+                break;
+            }
+        }
+    });
+
+    // agent -> stdout (responses / notifications), fire-and-forget.
+    tokio::spawn(async move {
+        let mut stdout = tokio::io::stdout();
+        let mut rx = agent_to_us;
+        while let Some(item) = rx.next().await {
+            if let Ok(msg) = item {
+                if let Ok(mut s) = serde_json::to_string(&msg) {
+                    s.push('\n');
+                    if stdout.write_all(s.as_bytes()).await.is_err() {
+                        break;
+                    }
+                    let _ = stdout.flush().await;
+                }
+            }
+        }
+    });
+
+    let _agent_run = tokio::spawn(async move { build_agent().connect_to(agent_end).await });
+
+    // Terminate on stdin EOF (works for pipes and regular files): the stdin
+    // pump returns once the input is exhausted; a short grace lets the agent
+    // emit the final response; then we exit — matching TS AgentSideConnection
+    // EOF semantics without ever awaiting the SDK's server-mode future.
+    let _ = stdin_pump.await;
+    tokio::time::sleep(std::time::Duration::from_millis(400)).await;
+    std::process::exit(0);
+}
+
+fn build_agent() -> agent_client_protocol::Builder<
+    Agent,
+    impl agent_client_protocol::HandleDispatchFrom<Client>,
+    impl agent_client_protocol::RunWithConnectionTo<Client>,
+> {
+    Agent
+        .builder()
+        .name("tilth-acp-agent")
+        .on_receive_request(
+            async |req: InitializeRequest, responder, _cx| {
+                // Match shared/acp/golden/agent initialize: loadSession:false,
+                // promptCapabilities{image:false,audio:false,embeddedContext:true}.
+                let caps = AgentCapabilities::new()
+                    .prompt_capabilities(PromptCapabilities::new().embedded_context(true));
+                responder.respond(
+                    InitializeResponse::new(req.protocol_version)
+                        .agent_capabilities(caps)
+                        .agent_info(Implementation::new("tilth-acp-agent", "0.1.0")),
+                )
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        .on_receive_request(
+            async |_req: NewSessionRequest, responder, _cx| {
+                let n = SESSION_SEQ.fetch_add(1, Ordering::Relaxed);
+                responder.respond(NewSessionResponse::new(format!("tilth-{n}")))
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        .on_receive_request(
+            async |req: PromptRequest, responder, cx: ConnectionTo<Client>| {
+                let sid = req.session_id;
+                let query = text_of(&req.prompt);
+                let q = query.clone();
+                let result = tokio::task::spawn_blocking(move || run_search(&q))
+                    .await
+                    .unwrap_or_else(|e| Err(format!("tilth task join: {e}")));
+                let body = match result {
+                    Ok(text) => text,
+                    Err(e) => format!("tilth error: {e}"),
+                };
+                let chunk = SessionUpdate::AgentMessageChunk(ContentChunk::new(
+                    ContentBlock::Text(TextContent::new(body)),
+                ));
+                let _ = cx.send_notification(SessionNotification::new(sid, chunk));
+                responder.respond(PromptResponse::new(StopReason::EndTurn))
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        .on_receive_dispatch(
+            async |msg: Dispatch, cx| {
+                msg.respond_with_error(
+                    agent_client_protocol::util::internal_error("unhandled message"),
+                    cx,
+                )
+            },
+            agent_client_protocol::on_receive_dispatch!(),
+        )
+}
+
+fn run_search(query: &str) -> std::result::Result<String, String> {
+    tilth::mcp::search(query)
+}

tilth/crates/acp-client/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "tilth-acp-client"
+version = "0.1.0"
+edition = "2021"
+description = "ACP (Agent Client Protocol) client — Rust counterpart of aurelio-backend's AcpClient. Drives any ACP agent (tilth-acp-agent, claude-code, gemini, kimi)."
+license = "MIT"
+
+[[bin]]
+name = "tilth-acp-client"
+path = "src/main.rs"
+
+[dependencies]
+agent-client-protocol = "1.2.0"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "io-std", "time", "fs", "process"] }
+tokio-util = { version = "0.7", features = ["compat"] }
+serde_json = "1"
+futures-util = { version = "0.3", default-features = false, features = ["std"] }

tilth/crates/acp-client/src/main.rs

@@ -0,0 +1,276 @@
+//! tilth-acp-client — ACP client (Rust counterpart of aurelio-backend's
+//! `AcpClient`). It is the side an ACP *agent* talks to: it handles the
+//! agent->client requests `fs/read_text_file` and `session/request_permission`,
+//! and absorbs `session/update` notifications.
+//!
+//! Two modes, selected by argv:
+//!
+//! * **Passive (default, no args)** — speak ACP over stdin/stdout as a
+//!   server-of-client-methods, conforming to `shared/acp/golden/client/`.
+//!   Used by `npm run test:tilth-client`.
+//! * **Active (`--drive <agent-cmd> [args...]`)** — spawn an ACP agent over a
+//!   pipe, run `initialize → session/new → session/prompt`, stream each
+//!   `session/update` chunk to stdout as NDJSON, and print the final
+//!   `stopReason`. Used by `npm run test:interop-rust` (Rust client → Rust
+//!   agent) and is the Rust counterpart of `AcpClient::prompt`.
+//!
+//! Wire-compatible with the TypeScript `AcpClient` in
+//! `aurelio-theia/aurelio-backend/src/acp/` against `shared/acp/golden/client/`.
+//!
+//! Termination note (passive mode): identical to `tilth-acp-agent`. The Rust
+//! SDK's server-only `connect_to(transport)` never resolves on a clean inbound
+//! EOF — internally `run_until(background, future::pending())` keeps waiting
+//! after the transport actors finish. So we pump NDJSON through a
+//! `Channel::duplex()` we own and terminate on stdin EOF after a short grace
+//! (see Phase 3 notes in `.aurelio/knowledge/acp.md`). Active mode uses
+//! `connect_with`, which returns when the driver closure finishes — no hang.
+
+use std::process;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use agent_client_protocol::schema::v1::{
+    ContentBlock, InitializeRequest, NewSessionRequest, PromptRequest, ReadTextFileRequest,
+    ReadTextFileResponse, RequestPermissionOutcome, RequestPermissionRequest,
+    RequestPermissionResponse, SelectedPermissionOutcome, SessionNotification, StopReason,
+    TextContent,
+};
+use agent_client_protocol::schema::ProtocolVersion;
+use agent_client_protocol::{Agent, ByteStreams, Channel, Client, RawJsonRpcMessage, Result};
+use futures_util::StreamExt;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::process::Command;
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+
+/// Grace between inbound EOF and process exit, so the SDK can flush the
+/// in-flight response it has already buffered from the now-closed stdin.
+const EOF_FLUSH_GRACE: Duration = Duration::from_millis(300);
+
+/// Optional sink invoked for each `session/update` (active mode only).
+type UpdateSink = Option<Arc<dyn Fn(&SessionNotification) + Send + Sync>>;
+
+fn build_client(
+    on_update: UpdateSink,
+) -> agent_client_protocol::Builder<
+    Client,
+    impl agent_client_protocol::HandleDispatchFrom<Agent>,
+    impl agent_client_protocol::RunWithConnectionTo<Agent>,
+> {
+    Client
+        .builder()
+        // fs/read_text_file: read the requested path from disk (path is whatever
+        // the agent/session scoped; confinement is the agent's responsibility).
+        // Return the raw bytes as UTF-8 (lossy) so the agent always gets content.
+        .on_receive_request(
+            async |req: ReadTextFileRequest, responder, _cx| {
+                let body = match tokio::fs::read_to_string(&req.path).await {
+                    Ok(s) => s,
+                    // Lossy fallback for non-UTF8 or unreadable files: read raw
+                    // bytes and re-encode; if even that fails, surface the error
+                    // as content (never fail the JSON-RPC call itself).
+                    Err(_) => match tokio::fs::read(&req.path).await {
+                        Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
+                        Err(e) => format!("tilth-acp-client: read failed: {e}"),
+                    },
+                };
+                responder.respond(ReadTextFileResponse::new(body))
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        // session/request_permission: deterministic policy. Pick the first
+        // `allow_*` option (prefer `allow-once`); if none allow, cancel.
+        .on_receive_request(
+            async |req: RequestPermissionRequest, responder, _cx| {
+                let chosen = req
+                    .options
+                    .iter()
+                    .find(|o| o.option_id.0.as_ref() == "allow-once")
+                    .or_else(|| {
+                        req.options
+                            .iter()
+                            .find(|o| o.option_id.0.as_ref().starts_with("allow"))
+                    })
+                    .map(|o| o.option_id.clone());
+                let outcome = match chosen {
+                    Some(id) => {
+                        RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(id))
+                    }
+                    None => RequestPermissionOutcome::Cancelled,
+                };
+                responder.respond(RequestPermissionResponse::new(outcome))
+            },
+            agent_client_protocol::on_receive_request!(),
+        )
+        // session/update: client absorbs stream notifications. In active mode a
+        // sink prints each one (NDJSON) so a harness can assert a non-empty
+        // agent_message_chunk; in passive mode this is a no-op.
+        .on_receive_notification(
+            async move |notif: SessionNotification, _cx| {
+                if let Some(cb) = &on_update {
+                    cb(&notif);
+                }
+                Ok(())
+            },
+            agent_client_protocol::on_receive_notification!(),
+        )
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Mode split: `--drive <agent-cmd> [args...]` runs the active driver;
+    // anything else (incl. no args) is the passive conformance server.
+    let mut args = std::env::args().skip(1);
+    if matches!(args.next().as_deref(), Some("--drive")) {
+        let agent: Vec<String> = args.collect();
+        return drive(agent).await;
+    }
+    serve_passive().await
+}
+
+/// Passive mode: speak ACP over stdin/stdout as a server-of-client-methods.
+async fn serve_passive() -> Result<()> {
+    let (client_end, pump_end) = Channel::duplex();
+    let Channel {
+        rx: client_to_us,
+        tx: us_to_client,
+    } = pump_end;
+
+    // stdin -> client (agent->client requests). On EOF this task returns; the
+    // client is kept alive (fire-and-forget) so it can finish flushing the
+    // in-flight response during the grace window below. We do NOT await the
+    // client's connect_to — it is server-mode and never returns on a clean
+    // inbound EOF (same `run_until`/`future::pending()` footgun as Stdio), so
+    // waiting on it would hang. Instead we terminate the process after a grace.
+    let stdin_pump = tokio::spawn(async move {
+        let stdin = BufReader::new(tokio::io::stdin());
+        let mut lines = stdin.lines();
+        while let Ok(Some(line)) = lines.next_line().await {
+            if line.trim().is_empty() {
+                continue;
+            }
+            let item = match serde_json::from_str::<RawJsonRpcMessage>(&line) {
+                Ok(m) => Ok(m),
+                Err(_) => Err(agent_client_protocol::Error::parse_error()),
+            };
+            if us_to_client.unbounded_send(item).is_err() {
+                break;
+            }
+        }
+    });
+
+    // client -> stdout (client->agent responses / notifications), fire-and-forget.
+    tokio::spawn(async move {
+        let mut stdout = tokio::io::stdout();
+        let mut rx = client_to_us;
+        while let Some(item) = rx.next().await {
+            if let Ok(msg) = item {
+                if let Ok(mut s) = serde_json::to_string(&msg) {
+                    s.push('\n');
+                    if stdout.write_all(s.as_bytes()).await.is_err() {
+                        break;
+                    }
+                    let _ = stdout.flush().await;
+                }
+            }
+        }
+    });
+
+    let _client_run = tokio::spawn(async move { build_client(None).connect_to(client_end).await });
+
+    // Terminate on stdin EOF (works for pipes and regular files): the stdin
+    // pump returns once the input is exhausted; a short grace lets the client
+    // emit the final response; then we exit — matching TS ClientSideConnection
+    // EOF semantics without ever awaiting the SDK's server-mode future.
+    let _ = stdin_pump.await;
+    tokio::time::sleep(EOF_FLUSH_GRACE).await;
+    process::exit(0);
+}
+
+/// Active mode: spawn an ACP agent and run one prompt turn to completion.
+///
+/// Prints each `session/update` notification to stdout as NDJSON (so a harness
+/// can assert a non-empty `agent_message_chunk`), and finally a single
+/// `{"stopReason":"end_turn"}` summary line. Exits non-zero on any protocol
+/// error, a non-`EndTurn` stop, or a wall-clock timeout.
+async fn drive(agent: Vec<String>) -> Result<()> {
+    if agent.is_empty() {
+        eprintln!("tilth-acp-client --drive: missing agent command");
+        process::exit(2);
+    }
+    let prompt_text =
+        std::env::var("INTEROP_PROMPT").unwrap_or_else(|_| "tilth_search".to_string());
+    let cwd = std::env::var("INTEROP_CWD")
+        .map(std::path::PathBuf::from)
+        .unwrap_or_else(|_| std::env::current_dir().expect("cwd"));
+
+    let mut cmd = Command::new(&agent[0]);
+    cmd.args(&agent[1..])
+        .stdin(std::process::Stdio::piped())
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::inherit())
+        .kill_on_drop(true);
+    let mut child = cmd.spawn().map_err(|e| {
+        agent_client_protocol::Error::internal_error().data(format!("spawn {}: {e}", agent[0]))
+    })?;
+    let child_in = child.stdin.take().expect("child stdin piped");
+    let child_out = child.stdout.take().expect("child stdout piped");
+    let transport = ByteStreams::new(child_in.compat_write(), child_out.compat());
+
+    // Sink: print each session/update as NDJSON and record that we saw one.
+    let saw_update = Arc::new(AtomicBool::new(false));
+    let sink: Arc<dyn Fn(&SessionNotification) + Send + Sync> = {
+        let saw = Arc::clone(&saw_update);
+        Arc::new(move |notif| {
+            saw.store(true, Ordering::Relaxed);
+            if let Ok(line) = serde_json::to_string(notif) {
+                println!("{line}");
+            }
+        })
+    };
+
+    // Bound the whole turn so a hung agent can't wedge CI.
+    let turn = build_client(Some(sink)).connect_with(transport, async move |cx| {
+        let init = cx
+            .send_request(InitializeRequest::new(ProtocolVersion::V1))
+            .block_task()
+            .await?;
+        if init.protocol_version != ProtocolVersion::V1 {
+            return Err(agent_client_protocol::Error::internal_error().data(format!(
+                "unexpected protocol_version {:?}",
+                init.protocol_version
+            )));
+        }
+        let sess = cx
+            .send_request(NewSessionRequest::new(cwd))
+            .block_task()
+            .await?;
+        let session_id = sess.session_id;
+        let prompt = cx
+            .send_request(PromptRequest::new(
+                session_id,
+                vec![ContentBlock::Text(TextContent::new(prompt_text))],
+            ))
+            .block_task()
+            .await?;
+        Ok(prompt.stop_reason)
+    });
+    let stop = tokio::time::timeout(Duration::from_secs(25), turn)
+        .await
+        .map_err(|_| agent_client_protocol::Error::internal_error().data("turn timed out"))??;
+
+    // stop_reason serializes camelCase; compare structurally.
+    if stop != StopReason::EndTurn {
+        eprintln!("tilth-acp-client --drive: unexpected stopReason: {stop:?}");
+        let _ = child.kill().await;
+        process::exit(1);
+    }
+    let saw_chunk = saw_update.load(Ordering::Relaxed);
+    let line =
+        serde_json::json!({ "stopReason": "end_turn", "sawChunk": saw_chunk }).to_string() + "\n";
+    let mut out = tokio::io::stdout();
+    let _ = out.write_all(line.as_bytes()).await;
+    let _ = out.flush().await;
+    let _ = child.kill().await;
+    Ok(())
+}

tilth/src/mcp.rs

@@ -562,6 +562,18 @@ fn tool_search(
     Ok(result)
 }
 
+/// Library entry point for non-MCP callers (e.g. the ACP agent). Runs
+/// `tilth_search` against the current working directory with a fresh cache /
+/// session / index, returning the same formatted text the MCP tool would.
+pub fn search(query: &str) -> Result<String, String> {
+    let cache = OutlineCache::new();
+    let session = Session::new();
+    let index = Arc::new(SymbolIndex::new());
+    let bloom = Arc::new(BloomFilterCache::new());
+    let args = serde_json::json!({ "query": query });
+    tool_search(&args, &cache, &session, &index, &bloom)
+}
+
 fn tool_files(args: &Value, cache: &OutlineCache) -> Result<String, String> {
     let pattern = args
         .get("pattern")