feat(acp-runner): add set_session_model handler to RpcServer

Subscribe to {prefix}.*.agent.session.set_model and persist the
model override to SessionState in KV, matching the existing
set_session_mode pattern.

Author: jramirezhdez02

rsworkspace/crates/acp-nats/src/agent/fork_session.rs

@@ -1,6 +1,6 @@
 use super::Bridge;
 use crate::error::map_nats_error;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, RequestClient, agent};
 use crate::session_id::AcpSessionId;
 use agent_client_protocol::{Error, ErrorCode, ForkSessionRequest, ForkSessionResponse, Result};
 use tracing::{Span, info, instrument};
@@ -11,7 +11,7 @@ use trogon_std::time::GetElapsed;
     skip(bridge, args),
     fields(session_id = %args.session_id, new_session_id = tracing::field::Empty)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: RequestClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: ForkSessionRequest,
 ) -> Result<ForkSessionResponse> {
@@ -44,7 +44,6 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
         Span::current().record("new_session_id", response.session_id.to_string().as_str());
         info!(new_session_id = %response.session_id, "Session forked");
 
-        bridge.schedule_session_ready(response.session_id.clone());
     }
 
     bridge.metrics.record_request(

rsworkspace/crates/acp-nats/src/agent/resume_session.rs

@@ -1,6 +1,6 @@
 use super::Bridge;
 use crate::error::map_nats_error;
-use crate::nats::{self, FlushClient, PublishClient, RequestClient, agent};
+use crate::nats::{self, RequestClient, agent};
 use crate::session_id::AcpSessionId;
 use agent_client_protocol::{
     Error, ErrorCode, Result, ResumeSessionRequest, ResumeSessionResponse,
@@ -13,7 +13,7 @@ use trogon_std::time::GetElapsed;
     skip(bridge, args),
     fields(session_id = %args.session_id)
 )]
-pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapsed>(
+pub async fn handle<N: RequestClient, C: GetElapsed>(
     bridge: &Bridge<N, C>,
     args: ResumeSessionRequest,
 ) -> Result<ResumeSessionResponse> {
@@ -42,10 +42,6 @@ pub async fn handle<N: RequestClient + PublishClient + FlushClient, C: GetElapse
     .await
     .map_err(map_nats_error);
 
-    if result.is_ok() {
-        bridge.schedule_session_ready(args.session_id.clone());
-    }
-
     bridge.metrics.record_request(
         "resume_session",
         bridge.clock.elapsed(start).as_secs_f64(),

rsworkspace/crates/acp-nats/src/nats/subjects.rs

@@ -47,6 +47,10 @@ pub mod agent {
         format!("{}.{}.agent.session.resume", prefix, session_id)
     }
 
+    pub fn session_close(prefix: &str, session_id: &str) -> String {
+        format!("{}.{}.agent.session.close", prefix, session_id)
+    }
+
     pub fn ext_session_ready(prefix: &str, session_id: &str) -> String {
         format!("{}.{}.agent.ext.session.ready", prefix, session_id)
     }
@@ -70,30 +74,6 @@ pub mod agent {
         )
     }
 
-    pub fn session_list(prefix: &str) -> String {
-        format!("{}.agent.session.list", prefix)
-    }
-
-    pub fn session_set_config_option(prefix: &str, session_id: &str) -> String {
-        format!("{}.{}.agent.session.set_config_option", prefix, session_id)
-    }
-
-    pub fn session_set_model(prefix: &str, session_id: &str) -> String {
-        format!("{}.{}.agent.session.set_model", prefix, session_id)
-    }
-
-    pub fn session_fork(prefix: &str, session_id: &str) -> String {
-        format!("{}.{}.agent.session.fork", prefix, session_id)
-    }
-
-    pub fn session_resume(prefix: &str, session_id: &str) -> String {
-        format!("{}.{}.agent.session.resume", prefix, session_id)
-    }
-
-    pub fn session_close(prefix: &str, session_id: &str) -> String {
-        format!("{}.{}.agent.session.close", prefix, session_id)
-    }
-
     /// Alias for `session_prompt` — used by the runner crate.
     pub fn prompt(prefix: &str, session_id: &str) -> String {
         session_prompt(prefix, session_id)

rsworkspace/crates/trogon-acp-runner/src/rpc_server.rs

@@ -4,7 +4,7 @@
 //! NATS request and deserialises the reply.  This module is the other side of
 //! those requests, implementing the actual agent logic for:
 //!   initialize · authenticate · new_session · load_session
-//!   set_session_mode · set_session_config_option
+//!   set_session_mode · set_session_model · set_session_config_option
 //!   list_sessions · fork_session · resume_session
 //!
 //! `prompt` / `cancel` are handled by `runner.rs` via the streaming pub/sub
@@ -19,7 +19,7 @@ use agent_client_protocol::{
     ResumeSessionRequest, ResumeSessionResponse, SessionCapabilities, SessionForkCapabilities,
     SessionInfo, SessionListCapabilities, SessionResumeCapabilities,
     SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, SetSessionModeRequest,
-    SetSessionModeResponse,
+    SetSessionModeResponse, SetSessionModelRequest, SetSessionModelResponse,
 };
 use acp_nats::nats::{ExtSessionReady, agent as subjects};
 use futures_util::StreamExt;
@@ -108,6 +108,10 @@ impl RpcServer {
             .nats
             .subscribe(format!("{}.*.agent.session.set_mode", prefix))
             .await?;
+        let mut set_model_sub = self
+            .nats
+            .subscribe(format!("{}.*.agent.session.set_model", prefix))
+            .await?;
         let mut set_config_sub = self
             .nats
             .subscribe(format!("{}.*.agent.session.set_config_option", prefix))
@@ -146,6 +150,10 @@ impl RpcServer {
                     let Some(msg) = msg else { break; };
                     self.handle_set_session_mode(msg).await;
                 }
+                msg = set_model_sub.next() => {
+                    let Some(msg) = msg else { break; };
+                    self.handle_set_session_model(msg).await;
+                }
                 msg = set_config_sub.next() => {
                     let Some(msg) = msg else { break; };
                     self.handle_set_session_config_option(msg).await;
@@ -271,6 +279,32 @@ impl RpcServer {
         self.reply(&msg, &SetSessionModeResponse::new()).await;
     }
 
+    async fn handle_set_session_model(&self, msg: async_nats::Message) {
+        let request: SetSessionModelRequest = match serde_json::from_slice(&msg.payload) {
+            Ok(r) => r,
+            Err(e) => {
+                warn!(error = %e, "rpc: bad set_session_model payload");
+                return;
+            }
+        };
+
+        let session_id = request.session_id.to_string();
+        match self.store.load(&session_id).await {
+            Ok(mut state) => {
+                state.model = Some(request.model_id.to_string());
+                state.updated_at = now_iso8601();
+                if let Err(e) = self.store.save(&session_id, &state).await {
+                    warn!(session_id = %session_id, error = %e, "rpc: failed to persist model update");
+                }
+            }
+            Err(e) => {
+                warn!(session_id = %session_id, error = %e, "rpc: failed to load session for model update");
+            }
+        }
+
+        self.reply(&msg, &SetSessionModelResponse::new()).await;
+    }
+
     async fn handle_set_session_config_option(&self, msg: async_nats::Message) {
         let _request: SetSessionConfigOptionRequest = match serde_json::from_slice(&msg.payload) {
             Ok(r) => r,