feat(node): hand model grants to plugin-owned adapters

apps/ora-controller/src/cloud/agents/mapping.rs

@@ -36,6 +36,11 @@ pub(in crate::cloud) fn start(
                 checkout_execution_id: ExecutionId::new(spec.checkout_execution_id.clone()),
                 model_binding_id: (!spec.model_binding_id.is_empty())
                     .then(|| ModelBindingId::new(spec.model_binding_id.clone())),
+                agent_configuration: spec
+                    .agent_configuration
+                    .as_ref()
+                    .map(configuration)
+                    .transpose()?,
                 git_identity: GitIdentity {
                     name: identity.name.clone(),
                     email: identity.email.clone(),
@@ -50,6 +55,42 @@ pub(in crate::cloud) fn start(
     Ok(command)
 }
 
+/// Carries Cloud's frozen adapter input verbatim; an unknown protocol or a missing model cannot be
+/// guessed, so the input is refused rather than started on another route.
+fn configuration(
+    input: &proto::AgentModelConfiguration,
+) -> Result<FrozenAgentConfiguration, Error> {
+    let model = input.model.as_ref().ok_or(Error::Conflict)?;
+    Ok(FrozenAgentConfiguration {
+        plugin_id: PluginId::new(input.plugin_id.clone()),
+        plugin_version: PluginVersion::new(input.plugin_version.clone()),
+        target: input.target.clone(),
+        package_sha256: Sha256Digest::new(input.package_sha256.clone()),
+        schema_version: input.schema_version,
+        declaration_fingerprint: Sha256Digest::new(input.declaration_fingerprint.clone()),
+        adapter_version: input.adapter_version,
+        configuration_revision: input.configuration_revision,
+        settings_json: input.settings_json.clone(),
+        protocol: ModelProtocol::parse(&input.protocol).ok_or(Error::Conflict)?,
+        model: BoundModel {
+            id: model.id.clone(),
+            name: model.name.clone(),
+            context_window: model.context_window,
+            max_tokens: model.max_tokens,
+        },
+    })
+}
+
+/// Whether a session input carries a frozen adapter configuration, and which.
+pub(in crate::cloud) fn frozen_configuration(
+    input: Option<&proto::ExecutionInput>,
+) -> Option<&proto::AgentModelConfiguration> {
+    match input.and_then(|v| v.spec.as_ref()) {
+        Some(proto::execution_input::Spec::AgentSession(spec)) => spec.agent_configuration.as_ref(),
+        Some(_) | None => None,
+    }
+}
+
 /// A session restores its prior Revision from the Git Server location Cloud fixed, so a prior
 /// Revision whose repository or delivery ref the Node protocol refuses cannot be restored and the
 /// input is refused rather than started as a fresh session.
@@ -198,6 +239,7 @@ mod tests {
             input: Some(proto::ExecutionInput {
                 spec: Some(proto::execution_input::Spec::AgentSession(
                     proto::AgentSessionSpec {
+                        agent_configuration: None,
                         model_binding_id: String::new(),
                         agent_plugin_id: "official/ora-space.echo".into(),
                         agent_plugin_version: "1.2.3".into(),
@@ -274,6 +316,7 @@ mod tests {
                         agent_plugin_version: PluginVersion::new("1.2.3"),
                         checkout_execution_id: ExecutionId::new("clone"),
                         model_binding_id: None,
+                        agent_configuration: None,
                         git_identity: GitIdentity {
                             name: "User".into(),
                             email: "user@example.com".into(),

apps/ora-controller/src/cloud/agents/previews/tests/server.rs

@@ -114,6 +114,7 @@ impl ExecutionService for Server {
                 input: Some(proto::ExecutionInput {
                     spec: Some(proto::execution_input::Spec::AgentSession(
                         proto::AgentSessionSpec {
+                            agent_configuration: None,
                             agent_plugin_id: "official/ora-space.echo".into(),
                             agent_plugin_version: "1.2.3".into(),
                             checkout_execution_id: "checkout".into(),

apps/ora-controller/src/cloud/claim.rs

@@ -90,6 +90,8 @@ pub(super) async fn batch(
                     && (!matches!(item.input.as_ref().and_then(|input| input.spec.as_ref()),
                     Some(proto::execution_input::Spec::AgentSession(spec)) if !spec.model_binding_id.is_empty())
                         || sandbox.advertises(NodeCapability::ModelProxy))
+                    && agents::mapping::frozen_configuration(item.input.as_ref())
+                        .is_none_or(|configuration| sandbox.admits(configuration))
             }
             Targeted::Delivery => sandbox.advertises(NodeCapability::RevisionDelivery),
         };

apps/ora-controller/src/cloud/fleet.rs

@@ -118,6 +118,8 @@ pub(super) struct Sandbox {
     pub(super) gate: sync::Mutex<Gate>,
     unresolved: Mutex<HashMap<ExecutionId, Instant>>,
     capabilities: Mutex<Vec<NodeCapability>>,
+    /// The adapter facts of the same handshake; only an exact match runs a configured session.
+    model_adapter: Mutex<Option<ModelAdapterCapability>>,
     /// Cloud's Node records of this sandbox, from the latest snapshot.
     known: Mutex<Vec<proto::NodeRecord>>,
 }
@@ -148,6 +150,27 @@ impl Sandbox {
             .contains(&NodeCapability::ModelProxy)
     }
 
+    /// Registration reports the adapter facts of the same handshake as `model_proxy`.
+    pub(super) fn advertised_model_adapter(&self) -> Option<ModelAdapterCapability> {
+        self.model_adapter
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .clone()
+    }
+
+    /// Whether the live session can run a session frozen with `configuration`. Cloud fences
+    /// the same match; checking here keeps an incompatible item queued instead of refused.
+    pub(super) fn admits(&self, configuration: &proto::AgentModelConfiguration) -> bool {
+        let Some(adapter) = self.advertised_model_adapter() else {
+            return false;
+        };
+        self.connected()
+            && adapter.contract_version == configuration.adapter_version
+            && adapter.target == configuration.target
+            && ModelProtocol::parse(&configuration.protocol)
+                .is_some_and(|protocol| adapter.protocols.contains(&protocol))
+    }
+
     /// Whether a session is established right now.
     pub(super) fn connected(&self) -> bool {
         self.identity.borrow().is_some()
@@ -196,11 +219,19 @@ impl Sandbox {
 }
 
 impl SessionObserver for Sandbox {
-    fn capabilities(&self, capabilities: &[NodeCapability]) {
+    fn capabilities(
+        &self,
+        capabilities: &[NodeCapability],
+        model_adapter: Option<&ModelAdapterCapability>,
+    ) {
         *self
             .capabilities
             .lock()
             .unwrap_or_else(PoisonError::into_inner) = capabilities.to_vec();
+        *self
+            .model_adapter
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner) = model_adapter.cloned();
     }
     fn established(&self, node: &NodeRuntimeIdentity) {
         self.identity.send_replace(Some(node.clone()));
@@ -299,6 +330,7 @@ impl Fleet {
             gate: sync::Mutex::new(Gate::Open),
             unresolved: Mutex::default(),
             capabilities: Mutex::default(),
+            model_adapter: Mutex::default(),
             known: Mutex::new(records),
         });
         let (stop, stopping) = watch::channel(false);

apps/ora-controller/src/cloud/reports.rs

@@ -99,6 +99,19 @@ async fn register(
     let epoch = store.epoch()?;
     let binding = &sandbox.binding;
     let model_proxy = sandbox.advertised_model_proxy();
+    let model_adapter =
+        sandbox
+            .advertised_model_adapter()
+            .map(|adapter| proto::ModelAdapterCapabilities {
+                contract_version: adapter.contract_version,
+                target: adapter.target,
+                protocols: adapter
+                    .protocols
+                    .into_iter()
+                    .map(|protocol| protocol.as_str().to_string())
+                    .collect(),
+            });
+    let model_adapter = &model_adapter;
     let response = fault::write(|submission_id| async move {
         let request = store.request(proto::RegisterNodeRequest {
             submission_id,
@@ -111,6 +124,7 @@ async fn register(
             }),
             protocol_version: u32::from(CURRENT_PROTOCOL_VERSION.value()),
             model_proxy,
+            model_adapter: model_adapter.clone(),
         });
         store.nodes().register_node(request).await
     })

apps/ora-controller/src/session.rs

@@ -118,7 +118,12 @@ pub(crate) trait SessionObserver: Send + Sync {
     /// The handshake completed with this Node incarnation; the session is live from here on.
     fn established(&self, node: &NodeRuntimeIdentity);
     /// Records the negotiated capabilities before publishing the connected identity.
-    fn capabilities(&self, _capabilities: &[NodeCapability]) {}
+    fn capabilities(
+        &self,
+        _capabilities: &[NodeCapability],
+        _model_adapter: Option<&ModelAdapterCapability>,
+    ) {
+    }
     /// The Node still answers `Unknown` for an execution after this connection retransmitted its
     /// original command once. It may be a retained uncertain attempt or a command the Node has not
     /// admitted yet; the owner decides how long that may last.
@@ -330,8 +335,12 @@ async fn coordinate<S: CoordinationStore, R: FrameReceiver, W: FrameSender, O: S
         .payload
         .capabilities
         .contains(&NodeCapability::RevisionDelivery);
-    observer.capabilities(&hello.payload.capabilities);
+    observer.capabilities(
+        &hello.payload.capabilities,
+        hello.payload.model_adapter.as_ref(),
+    );
     let capabilities = hello.payload.capabilities;
+    let model_adapter = hello.payload.model_adapter;
     let identity = hello.payload.node;
     observer.established(&identity);
     let mut tick = interval(Duration::from_millis(config.query_interval_ms));
@@ -350,6 +359,7 @@ async fn coordinate<S: CoordinationStore, R: FrameReceiver, W: FrameSender, O: S
         identity.clone(),
         Duration::from_millis(config.query_interval_ms),
         capabilities,
+        model_adapter,
         commands.input.clone(),
         grants.input.clone(),
     );

apps/ora-controller/src/session/control.rs

@@ -22,6 +22,7 @@ impl Control {
         identity: NodeRuntimeIdentity,
         period: Duration,
         capabilities: Vec<NodeCapability>,
+        model_adapter: Option<ModelAdapterCapability>,
         commands: mpsc::Sender<commands::Reply>,
         grants: mpsc::Sender<grants::Request>,
     ) -> Self {
@@ -42,6 +43,7 @@ impl Control {
             store,
             identity,
             capabilities,
+            model_adapter,
             receive,
             Workers { commands, grants },
             output,
@@ -144,6 +146,7 @@ async fn reconcile<S: CoordinationStore>(
     store: S,
     identity: NodeRuntimeIdentity,
     capabilities: Vec<NodeCapability>,
+    model_adapter: Option<ModelAdapterCapability>,
     mut input: mpsc::Receiver<NodeToControllerMessage>,
     workers: Workers,
     output: mpsc::Sender<Action>,
@@ -216,9 +219,17 @@ async fn reconcile<S: CoordinationStore>(
                     .original_agent_dispatch(&identity, &status.operation_id, &status.execution_id)
                     .await?
                 {
+                    // A configured session retransmits only to an incarnation that still admits it.
                     if agent_capable
                         && (agent.payload.spec.model_binding_id.is_none()
                             || capabilities.contains(&NodeCapability::ModelProxy))
+                        && agent.payload.spec.agent_configuration.as_ref().is_none_or(
+                            |configuration| {
+                                model_adapter
+                                    .as_ref()
+                                    .is_some_and(|adapter| adapter.admits(configuration))
+                            },
+                        )
                     {
                         store.dispatch_agent(agent).await?
                     } else {

apps/ora-controller/tests/cloud.rs

@@ -110,7 +110,7 @@ impl Node {
                             self.hellos.fetch_add(1, Ordering::SeqCst);
                             Some(NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage {
                                 protocol_version: CURRENT_PROTOCOL_VERSION,
-                                payload: HelloAccepted { thread_previews: false,
+                                payload: HelloAccepted { model_adapter: None, thread_previews: false,
                                     selected_version: CURRENT_PROTOCOL_VERSION,
                                     node: identity.clone(),
                                     capabilities: vec![NodeCapability::RepositoryClone, NodeCapability::RuntimeControl],

apps/ora-controller/tests/session.rs

@@ -56,6 +56,7 @@ fn mismatched_node_or_missing_clone_capability_rejects_before_dispatch() {
                             &NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage {
                                 protocol_version: CURRENT_PROTOCOL_VERSION,
                                 payload: HelloAccepted {
+                                    model_adapter: None,
                                     thread_previews: false,
                                     selected_version: CURRENT_PROTOCOL_VERSION,
                                     node: NodeRuntimeIdentity {
@@ -119,7 +120,7 @@ fn uncertain_execution_retransmits_at_most_once_per_connection() {
                 let (mut stream, _) = listener.accept().await.unwrap();
                 assert!(matches!(read_controller_message(&mut stream).await.unwrap(), Some(ControllerToNodeMessage::Hello(_))));
                 let identity = NodeRuntimeIdentity { node_id: endpoint.node_id.clone(), incarnation_id: NodeIncarnationId::new("current") };
-                write_node_message(&mut stream, &NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage { protocol_version: CURRENT_PROTOCOL_VERSION, payload: HelloAccepted { thread_previews: false, selected_version: CURRENT_PROTOCOL_VERSION, node: identity.clone(), capabilities: vec![NodeCapability::RepositoryClone] } })).await.unwrap();
+                write_node_message(&mut stream, &NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage { protocol_version: CURRENT_PROTOCOL_VERSION, payload: HelloAccepted { model_adapter: None, thread_previews: false, selected_version: CURRENT_PROTOCOL_VERSION, node: identity.clone(), capabilities: vec![NodeCapability::RepositoryClone] } })).await.unwrap();
                 let mut retries = 0;
                 let mut queries = 0;
                 while queries < 4 {
@@ -167,7 +168,7 @@ fn idle_session_sends_one_heartbeat_per_query_tick() {
                 let (mut stream, _) = listener.accept().await.unwrap();
                 assert!(matches!(read_controller_message(&mut stream).await.unwrap(), Some(ControllerToNodeMessage::Hello(_))));
                 let identity = NodeRuntimeIdentity { node_id: endpoint.node_id.clone(), incarnation_id: NodeIncarnationId::new("current") };
-                write_node_message(&mut stream, &NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage { protocol_version: CURRENT_PROTOCOL_VERSION, payload: HelloAccepted { thread_previews: false, selected_version: CURRENT_PROTOCOL_VERSION, node: identity.clone(), capabilities: vec![NodeCapability::RepositoryClone] } })).await.unwrap();
+                write_node_message(&mut stream, &NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage { protocol_version: CURRENT_PROTOCOL_VERSION, payload: HelloAccepted { model_adapter: None, thread_previews: false, selected_version: CURRENT_PROTOCOL_VERSION, node: identity.clone(), capabilities: vec![NodeCapability::RepositoryClone] } })).await.unwrap();
                 let expected = ControllerToNodeMessage::Heartbeat(ControllerHeartbeatMessage { protocol_version: CURRENT_PROTOCOL_VERSION, payload: ControllerHeartbeat { controller_id: ControllerId::new("owner") } });
                 for _ in 0..5 {
                     let message = timeout(Duration::from_millis(/*millis*/ 500), read_controller_message(&mut stream)).await.unwrap().unwrap().unwrap();
@@ -307,6 +308,7 @@ fn controller_closes_with_the_reason_code() {
                             NodeToControllerMessage::HelloAccepted(HelloAcceptedMessage {
                                 protocol_version: CURRENT_PROTOCOL_VERSION,
                                 payload: HelloAccepted {
+                                    model_adapter: None,
                                     thread_previews: false,
                                     selected_version: CURRENT_PROTOCOL_VERSION,
                                     node: NodeRuntimeIdentity {

apps/ora-controller/tests/support/workspace_cloud.rs

@@ -1037,6 +1037,7 @@ impl NodeReportService for WorkspaceCloud {
                     version: 1,
                     idle_admission_epoch: None,
                     model_proxy: message.model_proxy,
+                    model_adapter: message.model_adapter,
                 };
                 state.nodes.push(record.clone());
                 record

apps/ora-controller/tests/support/workspace_cloud/agents.rs

@@ -46,6 +46,27 @@ impl WorkspaceCloud {
         run: &str,
         prior: Option<proto::PriorRevision>,
         binding: &str,
+    ) {
+        self.queue_session_input(run, prior, binding, /*configuration*/ None);
+    }
+
+    /// Freezes a model binding together with the adapter configuration Cloud admitted.
+    pub fn queue_configured_agent(
+        &self,
+        run: &str,
+        binding: &str,
+        configuration: proto::AgentModelConfiguration,
+    ) {
+        self.queue_session_input(run, /*prior*/ None, binding, Some(configuration));
+    }
+
+    /// Queues one session work item and wakes the Controller.
+    fn queue_session_input(
+        &self,
+        run: &str,
+        prior: Option<proto::PriorRevision>,
+        binding: &str,
+        configuration: Option<proto::AgentModelConfiguration>,
     ) {
         let mut state = self.lock();
         let sandbox = state.sandboxes.last().unwrap();
@@ -60,6 +81,7 @@ impl WorkspaceCloud {
             input: Some(proto::ExecutionInput {
                 spec: Some(proto::execution_input::Spec::AgentSession(
                     proto::AgentSessionSpec {
+                        agent_configuration: configuration,
                         agent_plugin_id: "official/ora-space.echo".into(),
                         agent_plugin_version: "1.2.3".into(),
                         checkout_execution_id: state.clones[0].execution_id.clone(),
@@ -115,6 +137,15 @@ impl WorkspaceCloud {
             },
         ]
     }
+    /// The adapter facts each registered Node record carries, in registration order.
+    pub fn registered_model_adapters(&self) -> Vec<Option<proto::ModelAdapterCapabilities>> {
+        self.lock()
+            .nodes
+            .iter()
+            .map(|node| node.model_adapter.clone())
+            .collect()
+    }
+
     /// Reads registered session facts for assertions and subsequent command input.
     pub fn agent_records(&self) -> Vec<proto::ExecutionRecord> {
         self.lock()

apps/ora-controller/tests/workspaces.rs

@@ -247,6 +247,7 @@ impl Node {
                                 selected_version: CURRENT_PROTOCOL_VERSION,
                                 node: identity.clone(),
                                 capabilities: plugins::capabilities(self),
+                                model_adapter: self.agents.model_adapter.lock().unwrap().clone(),
                             },
                         })) },
                         ControllerToNodeMessage::BindRuntime(binding) => Some(NodeToControllerMessage::RuntimeControlState(RuntimeControlState { binding, unfinished_execution_ids:vec![] })),