--- a/crates/airouter-agent-client/src/session_service.rs +++ b/crates/airouter-agent-client/src/session_service.rs @@ -22,6 +22,16 @@ /// The daemon copies and hashes bytes before acknowledging admission. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub attachments: Vec, + /// Saved resource bindings, not a client-owned temporary configuration. + /// The daemon supplies the registered login, admitted prompt and model. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub grok: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct GrokConfig { + pub mcp_servers: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] --- a/crates/airouter-agent-client/src/native/launch/codex.rs +++ b/crates/airouter-agent-client/src/native/launch/codex.rs @@ -16,7 +16,8 @@ } /// A host-selected stdio server. Environment order is retained in argv. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct McpServer { pub name: String, pub command: String, --- a/crates/airouter-agent-client/src/native/grok.rs +++ b/crates/airouter-agent-client/src/native/grok.rs @@ -2,7 +2,7 @@ //! and contained process runner; never infers zero cost from absent usage. use super::{ToolResultHooks, events::ProviderStreamEvent}; use crate::{ - process::{JsonlProcessInput, ProcessRunError, run_jsonl_process_with_checked_line_handler}, + process::{JsonlProcessInput, ProcessRunError, run_jsonl_process_cancellable}, protocol::session::Outcome, session_stream::{Frame, Tracker}, }; @@ -196,24 +196,54 @@ } pub async fn run( + input: JsonlProcessInput, + hooks: ToolResultHooks, + emit: F, +) -> Result +where + F: FnMut(Frame) -> Result<(), String>, +{ + run_cancellable(input, hooks, emit, crate::CancellationToken::new()).await +} + +pub async fn run_cancellable( mut input: JsonlProcessInput, hooks: ToolResultHooks, mut emit: F, + cancellation: crate::CancellationToken, ) -> Result where F: FnMut(Frame) -> Result<(), String>, { input.stdout_idle_retries = 0; let mut parser = Parser::new(hooks); - let output = run_jsonl_process_with_checked_line_handler(input, |line| { - let value: Value = serde_json::from_str(line) - .map_err(|_| ProcessRunError::Executor("invalid Grok session frame".into()))?; - for frame in parser.ingest(&value).map_err(ProcessRunError::Executor)? { - emit(frame).map_err(ProcessRunError::Executor)?; - } - Ok(()) - }) - .await - .map_err(|e| e.to_string())?; - parser.finish(output.exit_code) -} + let output = run_jsonl_process_cancellable( + input, + |line| { + let value: Value = serde_json::from_str(line) + .map_err(|_| ProcessRunError::Executor("invalid Grok session frame".into()))?; + for frame in parser.ingest(&value).map_err(ProcessRunError::Executor)? { + emit(frame).map_err(ProcessRunError::Executor)?; + } + Ok(()) + }, + cancellation, + ) + .await; + match output { + Ok(Some(output)) => parser.finish(output.exit_code), + Ok(None) => { + parser.error = Some("Grok session cancelled after verified local cleanup".into()); + let mut outcome = parser.finish(None)?; + outcome.final_response_id = None; + Ok(outcome) + } + Err(ProcessRunError::CleanupConfirmed(error)) => { + parser.error = Some(error.to_string().chars().take(3000).collect()); + let mut outcome = parser.finish(None)?; + outcome.final_response_id = None; + Ok(outcome) + } + Err(error) => Err(error.to_string()), + } +} --- a/crates/airouter-agent-client/src/native/session.rs +++ b/crates/airouter-agent-client/src/native/session.rs @@ -233,11 +233,7 @@ F: FnMut(Frame) -> Result<(), String>, { if executor == Executor::NativeGrok { - return tokio::select! { - biased; - _ = cancellation.cancelled() => Err("Grok session cancelled; reconcile original receipt".into()), - result = super::grok::run(input, hooks, emit) => result, - }; + return super::grok::run_cancellable(input, hooks, emit, cancellation).await; } let mut parser = SessionParser::new(&executor, hooks)?; // An admitted full session has one reservation; silence cannot authorize a --- a/crates/airouter-agentd/src/session_service/native.rs +++ b/crates/airouter-agentd/src/session_service/native.rs @@ -21,6 +21,9 @@ pub struct Registration { pub program: PathBuf, pub executor: Executor, + /// Existing operator-managed Grok login; never accepted from a submission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub credential_home: Option, } pub struct NativeBackend { @@ -39,10 +42,21 @@ if !registration.program.is_absolute() || !matches!( registration.executor, - Executor::NativeCodex | Executor::NativeClaude + Executor::NativeCodex | Executor::NativeClaude | Executor::NativeGrok ) { - return Err("register an absolute Codex/Claude program".into()); + return Err("register an absolute Codex/Claude/Grok program".into()); + } + if registration.executor == Executor::NativeGrok { + if !registration + .credential_home + .as_ref() + .is_some_and(|p| p.is_absolute() && p.is_dir()) + { + return Err("Grok requires an existing registered credential home".into()); + } + } else if registration.credential_home.is_some() { + return Err("credential_home is only supported for Grok".into()); } } Ok(Self { @@ -55,6 +69,8 @@ pub struct NativeHost { input: JsonlProcessInput, executor: Executor, + // Configuration must outlive every contained descendant, including cleanup. + _grok: Option, } fn result_hooks() -> ToolResultHooks { @@ -90,6 +106,54 @@ return Err("native execution placement and retries belong to the service".into()); } let mut input = request.input.clone(); + let grok = if registration.executor == Executor::NativeGrok { + let config = request + .grok + .as_ref() + .ok_or("Grok resource bindings missing")?; + if input.plan.args.len() != 1 || !input.plan.env.is_empty() || !input.stdin.is_empty() { + return Err("Grok launch configuration belongs to the service".into()); + } + let text = native::conversation::render( + run.request.messages.iter().map(|message| { + ( + match message.role { + airouter_agent_protocol::InputRole::User => "user", + airouter_agent_protocol::InputRole::Assistant => "assistant", + }, + message.content.as_str(), + ) + }), + None, + true, + ); + let prepared = native::launch::grok::prepare(native::launch::grok::Settings { + cli_path: registration + .program + .to_str() + .ok_or("invalid Grok executable path")?, + model: &session.profile.model, + cwd: &input.plan.cwd, + credential_home: registration + .credential_home + .as_ref() + .and_then(|p| p.to_str()) + .ok_or("invalid Grok credential home")?, + system_prompt: &run.request.instructions, + user_text: &text, + mcp_servers: &config.mcp_servers, + max_turns: 25, + })?; + let timeout = input.stdout_idle_timeout; + input = prepared.input.clone(); + input.stdout_idle_timeout = timeout; + Some(prepared) + } else { + if request.grok.is_some() { + return Err("Grok configuration requires a registered Grok executor".into()); + } + None + }; input.executor = Some(self.executor.clone()); input.watchdog = Some(ProcessWatchdogRegistration::new( run.id.to_string(), @@ -99,6 +163,7 @@ Ok(NativeHost { input, executor: registration.executor.clone(), + _grok: grok, }) } } @@ -149,15 +214,26 @@ (json!({"content":"example: {\"isError\":true}"}), 0), ] { let mut parser = CodexStreamParser::new(result_hooks()); - let events = parser.ingest_line(&json!({"type":"item.completed","item":{ - "type":"mcp_tool_call","id":"fixture","name":"resource", - "status":"completed","result":body, - }}).to_string()); - let [ProviderStreamEvent::ToolResult { exit_code, output, .. }] = events.as_slice() else { + let events = parser.ingest_line( + &json!({"type":"item.completed","item":{ + "type":"mcp_tool_call","id":"fixture","name":"resource", + "status":"completed","result":body, + }}) + .to_string(), + ); + let [ + ProviderStreamEvent::ToolResult { + exit_code, output, .. + }, + ] = events.as_slice() + else { panic!("expected one MCP tool result"); }; assert_eq!(*exit_code, expected); - assert_eq!(serde_json::from_str::(output).unwrap(), body); - } - } -} + assert_eq!( + serde_json::from_str::(output).unwrap(), + body + ); + } + } +} --- a/crates/airouter-agentd/tests/session_service/mod.rs +++ b/crates/airouter-agentd/tests/session_service/mod.rs @@ -100,6 +100,7 @@ version: wire::VERSION.into(), binding: prepared.binding.clone(), attachments: vec![], + grok: None, input: JsonlProcessInput::new( ProcessLaunchPlan::new( vec!["/fixture/registered-program".into()], @@ -349,6 +350,7 @@ Registration { program: "/fixture/registered-program".into(), executor: session::Executor::NativeCodex, + credential_home: None, }, )] .into(), @@ -402,6 +404,43 @@ } #[tokio::test] +async fn grok_service_requires_registered_credentials_and_saved_resource_bindings() { + use airouter_agent_client::process::external::ExternalExecutor; + use airouter_agentd::session_service::native::{NativeBackend, Registration}; + let f = Fixture::new().await; + let mut prepared = f.managed_session().await; + prepared.run.session.as_mut().unwrap().profile.executor = session::Executor::NativeGrok; + let dir = tempfile::tempdir().unwrap(); + let mut request = submit(&prepared, dir.path()); + request.input.stdin.clear(); + request.grok = Some(wire::GrokConfig { mcp_servers: vec![] }); + let registration = Registration { + program: "/fixture/registered-program".into(), + executor: session::Executor::NativeGrok, + credential_home: Some(dir.path().into()), + }; + let backend = |registration| NativeBackend::new( + [(prepared.run.request.profile.clone(), registration)].into(), + ExternalExecutor { executable: "/fixture/executor".into(), cgroup_root: "/fixture/cgroup".into() }, + ); + let mut missing = registration.clone(); + missing.credential_home = None; + assert!(backend(missing).is_err()); + let host = backend(registration).unwrap(); + assert!(host.prepare(&request, &prepared.run).is_ok()); + let saved = serde_json::to_vec(&request).unwrap(); + assert_eq!(serde_json::from_slice::(&saved).unwrap(), request); + request.input.plan.env.insert("GROK_AUTH_PATH".into(), "/unregistered/login".into()); + assert!(host.prepare(&request, &prepared.run).is_err()); + request.input.plan.env.clear(); + request.input.stdin = "different prompt".into(); + assert!(host.prepare(&request, &prepared.run).is_err()); + request.input.stdin.clear(); + request.grok = None; + assert!(host.prepare(&request, &prepared.run).is_err()); +} + +#[tokio::test] async fn recovered_cancel_before_attempt_has_complete_empty_activity() { let f = Fixture::new().await; let prepared = f.managed_session().await; --- a/crates/airouter-agentd/tests/controller_fencing.rs +++ b/crates/airouter-agentd/tests/controller_fencing.rs @@ -104,6 +104,7 @@ let mut app=Application::new("workbench".into(),"controller-fixture-only-credential-123456789",vec!["fixture".into()]); app.task_queues=vec!["lifecycle".into(),"control".into()];app.managed_hosts=vec!["workbench".into()];app.require_controller=true; let profiles=[("fixture-codex",session::Executor::NativeCodex,"openai","exact-native",vec!["text","image"]), + ("fixture-grok",session::Executor::NativeGrok,"grok","grok-4.6",vec!["text"]), ("fixture-claude",session::Executor::NativeClaude,"anthropic","exact-claude",vec!["text"]), ("fixture-http",session::Executor::HttpAgent,"openai-compatible","exact-local",vec!["text"])]; let sessions=profiles.into_iter().map(|(name,executor,provider,model,modalities)| { --- a/crates/airouter-agentd/SESSION_SERVICE.md +++ b/crates/airouter-agentd/SESSION_SERVICE.md @@ -6,11 +6,13 @@ modes 0700 and 0600. There is no TCP listener or shared-group admission. The framework application credential stays in the daemon's protected token file. -This draft supports registered Codex sessions with text/images and text Claude -adapters. The registered +This service supports registered Codex sessions with text/images and text Claude +and Grok adapters. The registered executable, native executor and cgroup placement come from daemon configuration. Prompt/argument/environment assembly remains a trusted application adapter -input during migration. This is not a model-facing arbitrary command API. +input for Codex/Claude during migration. Grok's launch configuration is assembled +in the daemon from the saved resource bindings and admitted run. This is not a +model-facing arbitrary command API. ## Ownership and persistence @@ -54,6 +56,17 @@ controller `binding`, and shared `JsonlProcessInput` as `input`. Execution placement is never accepted from that wire input. New requests are limited to 8 MiB and the configured active-job capacity (1–64). + +Grok submissions add `grok = { mcp_servers: [...] }`, using the shared native +`McpServer` fields. Their `input` contains only the registered executable as its +single argument, the workspace and optional idle timeout; environment and stdin +must be empty. The daemon renders the admitted messages, uses the admitted +model/instructions, and creates a private per-execution MCP configuration. The +profile's required `credential_home` selects the existing login; it cannot be +overridden by a submission. The configuration stays alive through descendant +cleanup and is then removed. Exact retries and accepted-but-unattempted recovery +need only the saved submission, not temporary application files. The existing +25-turn setting and restricted Grok features remain unchanged. For admitted images, also supply `attachments = [{ reference, path }]` in the same order as the framework session input. Each source is an app-authorized @@ -110,17 +123,20 @@ does not kill the AIRouter session. Run this service as the app account, never root. Keep the existing AIRouter Codex UID/home registration and login. -## Remaining production integration +## Consumer requirements -Before Workbench cutover, move its permission/tool bridge into an -independently available application resource service, and replace its direct -native driver with submit/observe/project calls. Transfer heartbeat and explicit -cancellation together. Do not leave observer-drop cancellation enabled there. +Applications keep their permission/tool bridge in an independently available +resource service and use submit/observe/project calls for native execution. +Transfer heartbeat and explicit cancellation together. Observer loss must not +cancel the execution. Workbench has deployed this separation for Codex/Claude; +Grok adoption requires the additional registration and compatible consumer. The app adapter must now supply the admitted image reference/path mapping and retain the original submission for exact retries after its temporary files end. -Grok and HTTP adapters are not registered by this draft. Their existing paths -continue until separately migrated. Prompt/persona/model/context policy still -follows the main AI ownership completion sequence. +Consumers must register Grok's exact profile/program and existing credential home +before selecting the service. HTTP adapters still require their separate resource +callback migration. Prompt/persona/model/context policy follows the main AI +ownership completion sequence; installing a daemon does not establish consumer +adoption. Focused qualification uses the existing private Unix framework fixture for observer loss/renewal, duplicate binding checks, cancellation/cleanup, saved --- a/config/session-service.example.toml +++ b/config/session-service.example.toml @@ -16,3 +16,9 @@ [profiles.example_native_codex] program = "/usr/libexec/airouter/airouter-codex" executor = "native_codex" + +# Optional: enable only with the exact admitted Grok profile and existing login. +# [profiles.native-grok-grok-4-6-v1] +# program = "/usr/bin/grok" +# executor = "native_grok" +# credential_home = "/var/lib/ai/grok"