feat(agent): M7 — headless serve control protocol over stdio

crates/agent-core/ARCHITECTURE.md

@@ -30,9 +30,16 @@ Everything above the wire is testable with mocks because of two trait boundaries
 - **`Tool`** — tests register a mock (`EchoTool`) to exercise dispatch without a real capability.
 - **`ModelTransport`** — tests replay scripted `StreamEvent`s to exercise the loop without a network.
 
+## Observation surface
+
+`Agent::run` exposes only streamed model events (`FnMut(&StreamEvent)`); `Agent::run_events` exposes
+the full [`AgentEvent`] stream — `Stream(StreamEvent)`, `ToolStart`/`ToolEnd` (tool boundaries), and
+`TurnEnd`. The headless `serve` in the `beyond-ai-agent` crate serializes these to its clients.
+
 ## Milestone status
 
-Scaffolded: the type model, the `Tool`/`ToolRegistry` seam, the `ModelTransport` seam, `Session`,
-and errors — all unit-tested. **Not yet built:** the wire dialect adapters (OpenAI + Anthropic), the
-HTTP-to-gateway transport, the agent loop itself, and the `serve` control API. See the repo plan for
-the milestone sequence (M2–M8).
+Built and tested: the type model, the `Tool`/`ToolRegistry` seam, the `ModelTransport` seam +
+`GatewayClient` HTTP transport, `MockTransport`, the OpenAI/Anthropic wire dialects, the agent loop
+(`run`/`run_events`), and `Session`. The coding tools, the `run` CLI, and the headless `serve`
+control protocol live in the `beyond-ai-agent` crate. Remaining: the Beyond platform tools
+(fork/sync/logs) and the full gateway e2e (M8).

crates/agent-core/src/agent.rs

@@ -10,6 +10,7 @@
 use std::sync::Arc;
 
 use futures::StreamExt;
+use serde::Serialize;
 use serde_json::{Value, json};
 
 use crate::error::{Error, Result};
@@ -18,6 +19,31 @@ use crate::session::Session;
 use crate::tool::ToolRegistry;
 use crate::transport::{ModelRequest, ModelTransport};
 
+/// An observable event from a run: a streamed model event, a tool-invocation boundary, or a turn
+/// boundary. The headless server serializes these to its clients; [`Agent::run`] exposes only the
+/// `Stream` events.
+#[derive(Debug, Clone, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum AgentEvent {
+    /// A streamed model event (text/tool deltas, usage, stop).
+    Stream(StreamEvent),
+    /// A tool is about to run, with the arguments the model supplied.
+    ToolStart {
+        id: String,
+        name: String,
+        input: Value,
+    },
+    /// A tool finished (or errored); `result` is what's fed back to the model.
+    ToolEnd {
+        id: String,
+        name: String,
+        result: String,
+        is_error: bool,
+    },
+    /// One model turn completed.
+    TurnEnd { stop_reason: StopReason, step: u32 },
+}
+
 /// Default per-turn output token ceiling.
 const DEFAULT_MAX_TOKENS: u32 = 4096;
 /// Default ceiling on loop iterations before bailing — a runaway-tool-call backstop.
@@ -77,6 +103,21 @@ impl Agent {
     pub async fn run<F>(&self, session: &mut Session, mut on_event: F) -> Result<()>
     where
         F: FnMut(&StreamEvent),
+    {
+        self.run_events(session, move |ev| {
+            if let AgentEvent::Stream(s) = &ev {
+                on_event(s);
+            }
+        })
+        .await
+    }
+
+    /// Drive the loop to completion, emitting an [`AgentEvent`] for every streamed model event, tool
+    /// invocation, and turn boundary — the full observation surface the headless server streams to
+    /// its clients. Returns when the model ends its turn without tools, or [`Error::MaxSteps`].
+    pub async fn run_events<F>(&self, session: &mut Session, mut sink: F) -> Result<()>
+    where
+        F: FnMut(AgentEvent),
     {
         loop {
             if session.steps >= self.max_steps {
@@ -93,10 +134,17 @@ impl Agent {
                 req = req.with_system(system.clone());
             }
 
-            let turn = self.run_turn(req, &mut on_event).await?;
+            let turn = {
+                let mut emit = |ev: StreamEvent| sink(AgentEvent::Stream(ev));
+                self.run_turn(req, &mut emit).await?
+            };
             session.push(Message::assistant(turn.blocks));
             session.record_usage(turn.input_tokens, turn.output_tokens);
             session.steps += 1;
+            sink(AgentEvent::TurnEnd {
+                stop_reason: turn.stop_reason,
+                step: session.steps,
+            });
 
             // Collect the tool calls the assistant just made.
             let calls: Vec<(String, String, Value)> = session
@@ -116,28 +164,36 @@ impl Agent {
             // Run each tool and feed results back as a user turn. A tool's own failure becomes an
             // error `tool_result`, not an aborted run — the model can react to it next turn.
             for (id, name, input) in calls {
+                sink(AgentEvent::ToolStart {
+                    id: id.clone(),
+                    name: name.clone(),
+                    input: input.clone(),
+                });
                 let (content, is_error) = match self.tools.get(&name) {
                     Some(tool) => match tool.run(input).await {
                         Ok(out) => (out, false),
                         Err(e) => (e.to_string(), true),
                     },
                     None => (format!("unknown tool: {name}"), true),
                 };
+                sink(AgentEvent::ToolEnd {
+                    id: id.clone(),
+                    name: name.clone(),
+                    result: content.clone(),
+                    is_error,
+                });
                 session.push(Message::tool_result(id, content, is_error));
             }
         }
     }
 
     /// Stream and assemble a single model turn into content blocks + accounting.
-    async fn run_turn<F>(&self, req: ModelRequest, on_event: &mut F) -> Result<Turn>
-    where
-        F: FnMut(&StreamEvent),
-    {
+    async fn run_turn(&self, req: ModelRequest, emit: &mut dyn FnMut(StreamEvent)) -> Result<Turn> {
         let mut stream = self.transport.stream(req).await?;
         let mut acc = Accumulator::default();
         while let Some(ev) = stream.next().await {
             let ev = ev?;
-            on_event(&ev);
+            emit(ev.clone());
             acc.apply(ev);
         }
         Ok(acc.finish())

crates/agent-core/src/lib.rs

@@ -28,7 +28,7 @@ pub mod session;
 pub mod tool;
 pub mod transport;
 
-pub use agent::Agent;
+pub use agent::{Agent, AgentEvent};
 pub use client::GatewayClient;
 pub use error::{Error, Result, ToolError};
 pub use message::{ContentBlock, Message, Role, StopReason, StreamEvent, ToolDef};

crates/agent/src/main.rs

@@ -1,8 +1,9 @@
 //! Beyond agent harness — CLI.
 //!
-//! `run` drives a one-shot coding task to completion through the gateway. `serve` (M7) will expose
-//! the headless control API. `tools` lists the advertised tool set. Model traffic always flows
-//! through the gateway (`AI_GATEWAY_URL`) authenticated with a `bai_v1` key (`AI_AGENT_KEY`).
+//! `run` drives a one-shot coding task to completion through the gateway. `serve` exposes the
+//! headless control protocol (newline-delimited JSON over stdio). `tools` lists the advertised tool
+//! set. Model traffic always flows through the gateway (`AI_GATEWAY_URL`) authenticated with a
+//! `bai_v1` key (`AI_AGENT_KEY`).
 
 // Unit tests assert preconditions with `.unwrap()`; allow that under `test` (matches the gateway and
 // agent-core crate roots). Production paths stay panic-free per the workspace lints.
@@ -14,6 +15,7 @@ use std::sync::Arc;
 use agent_core::{Agent, GatewayClient, Session, StreamEvent};
 use clap::{Parser, Subcommand};
 
+mod serve;
 mod tools;
 
 /// Default model when neither `--model` nor `AI_AGENT_MODEL` is set.
@@ -51,8 +53,24 @@ enum Command {
         #[arg(long, default_value_t = 24)]
         max_steps: u32,
     },
-    /// Run the headless agent server exposing an attachable control API. (M7.)
-    Serve,
+    /// Run the headless agent server: a newline-delimited JSON control protocol over stdio.
+    Serve {
+        /// Model id (default `claude-opus-4-8`, or `AI_AGENT_MODEL`).
+        #[arg(long, env = "AI_AGENT_MODEL")]
+        model: Option<String>,
+        /// Gateway base URL (default `http://ai.internal`, or `AI_GATEWAY_URL`).
+        #[arg(long, env = "AI_GATEWAY_URL")]
+        gateway_url: Option<String>,
+        /// Virtual key (`bai_v1…`) or BYO provider key. Required; or set `AI_AGENT_KEY`.
+        #[arg(long, env = "AI_AGENT_KEY")]
+        key: Option<String>,
+        /// Persist/restore session state here so a later `serve` reattaches with the transcript.
+        #[arg(long, env = "AI_AGENT_SESSION_FILE")]
+        session_file: Option<String>,
+        /// Max loop iterations per prompt before bailing.
+        #[arg(long, default_value_t = 24)]
+        max_steps: u32,
+    },
     /// List the tools the agent advertises to the model.
     Tools,
 }
@@ -73,8 +91,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         } => {
             run_task(task, model, gateway_url, key, max_steps).await?;
         }
-        Command::Serve => {
-            println!("scaffold: `serve` is not wired yet (headless control API = M7).");
+        Command::Serve {
+            model,
+            gateway_url,
+            key,
+            session_file,
+            max_steps,
+        } => {
+            let key = key
+                .ok_or("no gateway key: pass --key or set AI_AGENT_KEY (a bai_v1… virtual key)")?;
+            serve::serve(serve::ServeConfig {
+                gateway: gateway_url.unwrap_or_else(|| DEFAULT_GATEWAY.to_string()),
+                key,
+                model: model.unwrap_or_else(|| DEFAULT_MODEL.to_string()),
+                max_steps,
+                system: SYSTEM_PROMPT.to_string(),
+                session_file,
+            })
+            .await?;
         }
         Command::Tools => {
             let reg = tools::default_registry();