bcode_server 0.0.1-alpha.0

//! Transport-neutral application operations for plugin service plumbing.

use super::{ServerState, plugin_event_metric_labels, plugin_service_metric_labels};
use tokio::sync::mpsc;

/// Derive operation correlation only from the host's authenticated invocation scope.
/// Unscoped/legacy calls produce no correlated observation; no current-turn lookup occurs.
pub fn scoped_optional_operation(
    state: &ServerState,
    scope: &bcode_plugin::PluginInvocationScope,
    feature: bcode_model::judgement::Feature,
) -> Option<super::judgement::ConsumerOperationHandle> {
    state
        .optional_judgement
        .observation_for_scope(scope)
        .map(|turn| {
            turn.begin_operation(
                feature,
                bcode_model::judgement::TriggerDisposition::Triggered,
            )
        })
}

/// Route nested optional inference through a host-owned operation correlation.
/// Plugin request IDs cannot select an observer or certify application/coverage.
/// Dropping this future retains pending inference and prevents complete closure.
pub async fn judge_correlated_optional(
    state: &ServerState,
    config: &bcode_config::BcodeConfig,
    input: bcode_model::judgement::OptionalRequest,
    operation: Option<&super::judgement::ConsumerOperationHandle>,
) -> bcode_model::judgement::OptionalResponse {
    if let Some(operation) = operation {
        state
            .optional_judgement
            .judge_observed_with_config(state, config, input, operation)
            .await
    } else {
        state
            .optional_judgement
            .judge_with_config(state, config, input)
            .await
    }
}

/// Failure while routing input to an active plugin invocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RouteInvocationInputError {
    /// Another connected client controls this invocation.
    NotController,
    /// No matching active invocation exists.
    NotActive,
    /// The producer does not own the selected invocation.
    ProducerMismatch,
    /// The producer identifier is empty.
    InvalidProducer,
    /// The schema identifier or version is invalid.
    InvalidSchema,
    /// The input identifier is empty.
    InvalidInputId,
    /// The encoded input exceeds the operation limit.
    TooLarge,
    /// The bounded invocation input queue is full.
    QueueFull,
    /// The invocation input route has closed.
    RouteClosed,
}

impl RouteInvocationInputError {
    /// Stable public operation error code.
    #[must_use]
    pub const fn code(self) -> &'static str {
        match self {
            Self::NotController => "invocation_input_not_controller",
            Self::NotActive => "plugin_invocation_not_active",
            Self::ProducerMismatch => "plugin_invocation_producer_mismatch",
            Self::InvalidProducer => "invalid_invocation_input_producer",
            Self::InvalidSchema => "invalid_invocation_input_schema",
            Self::InvalidInputId => "invalid_invocation_input_id",
            Self::TooLarge => "invocation_input_too_large",
            Self::QueueFull => "invocation_input_queue_full",
            Self::RouteClosed => "invocation_input_route_closed",
        }
    }

    /// Secret-safe public operation error message.
    #[must_use]
    pub const fn message(self) -> &'static str {
        match self {
            Self::NotController => "invocation input requires the controlling client",
            Self::NotActive => "plugin invocation is not active",
            Self::ProducerMismatch => "invocation input producer does not own the invocation",
            Self::InvalidProducer => "invocation input producer id must not be empty",
            Self::InvalidSchema => "invocation input schema and version must be valid",
            Self::InvalidInputId => "invocation input id must not be empty",
            Self::TooLarge => "invocation input exceeds 64 KiB",
            Self::QueueFull => "plugin invocation input queue is full",
            Self::RouteClosed => "plugin invocation input route is closed",
        }
    }
}

/// Return the current plugin service inventory without transport framing.
pub fn list_services(state: &ServerState) -> Vec<bcode_ipc::PluginServiceSummary> {
    state
        .plugins
        .service_summaries()
        .into_iter()
        .map(|(plugin_id, service)| bcode_ipc::PluginServiceSummary {
            plugin_id,
            interface_id: service.interface_id,
            name: service.name,
            description: service.description,
            workflow_blocks: service.workflow_blocks,
        })
        .collect()
}

/// Return the current renderer-neutral plugin contributions without transport framing.
pub fn list_contributions(state: &ServerState) -> bcode_plugin_sdk::PluginContributions {
    let mut command_contributions = state
        .plugins
        .registered_command_contributions(&bcode_command::CommandSurface::Palette);
    command_contributions.extend(
        state
            .plugins
            .registered_command_contributions(&bcode_command::CommandSurface::Slash),
    );
    command_contributions.sort_by(|left, right| left.id.cmp(&right.id));
    command_contributions.dedup_by(|left, right| left.id == right.id);
    bcode_plugin_sdk::PluginContributions {
        command_contributions,
        commands: state.plugins.command_contributions(),
        config_extensions: state.plugins.config_extensions(),
        composer_contributions: state.plugins.registry().composer_contributions(),
    }
}

/// Normalized plugin-service failure safe for public transport boundaries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublicPluginError {
    /// Stable public error code.
    pub code: &'static str,
    /// Secret-safe normalized message.
    pub message: &'static str,
}

/// Normalize a private plugin-host failure for public callers.
#[must_use]
pub const fn normalize_error(_error: &bcode_plugin::PluginLoadError) -> PublicPluginError {
    PublicPluginError {
        code: "plugin_error",
        message: "plugin operation failed; inspect local daemon diagnostics",
    }
}

/// Route input from a connected client. The first successfully enqueued input claims
/// control until that client disconnects or detaches. Rejected input never claims control.
///
/// # Errors
/// Returns an ownership, validation, or bounded queue error without changing control.
pub fn route_controlled_invocation_input(
    state: &ServerState,
    session_id: bcode_session_models::SessionId,
    client_id: bcode_session_models::ClientId,
    input: bcode_tool::ToolInvocationInput,
) -> Result<(), RouteInvocationInputError> {
    let mut controllers = state
        .invocation_input_controllers
        .lock()
        .map_err(|_| RouteInvocationInputError::NotController)?;
    let active = state
        .active_plugin_invocations
        .lock()
        .map_err(|_| RouteInvocationInputError::NotActive)?;
    controllers.retain(|key, _| active.contains_key(key));
    drop(active);
    let key = (session_id, input.invocation_id.clone());
    if controllers
        .get(&key)
        .is_some_and(|owner| *owner != client_id)
    {
        return Err(RouteInvocationInputError::NotController);
    }
    route_invocation_input(state, session_id, input)?;
    controllers.insert(key, client_id);
    drop(controllers);
    Ok(())
}

/// Route one bounded producer-owned input to an active plugin invocation.
pub fn route_invocation_input(
    state: &ServerState,
    session_id: bcode_session_models::SessionId,
    input: bcode_tool::ToolInvocationInput,
) -> Result<(), RouteInvocationInputError> {
    validate_invocation_input(&input)?;
    let active = {
        let invocations = state
            .active_plugin_invocations
            .lock()
            .map_err(|_| RouteInvocationInputError::NotActive)?;
        invocations
            .get(&(session_id, input.invocation_id.clone()))
            .cloned()
            .ok_or(RouteInvocationInputError::NotActive)?
    };
    if active.producer_plugin_id != input.producer_id {
        return Err(RouteInvocationInputError::ProducerMismatch);
    }
    enqueue_validated_invocation_input(&active, input)
}

const MAX_INVOCATION_INPUT_BYTES: usize = 64 * 1024;

struct InvocationInputBudget {
    remaining: usize,
}

impl std::io::Write for InvocationInputBudget {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.remaining = self
            .remaining
            .checked_sub(bytes.len())
            .ok_or_else(|| std::io::Error::other("invocation input exceeds encoded size limit"))?;
        Ok(bytes.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

fn validate_invocation_input(
    input: &bcode_tool::ToolInvocationInput,
) -> Result<(), RouteInvocationInputError> {
    if input.producer_id.trim().is_empty() {
        return Err(RouteInvocationInputError::InvalidProducer);
    }
    if input.schema.trim().is_empty() || input.schema_version == 0 {
        return Err(RouteInvocationInputError::InvalidSchema);
    }
    if input.input_id.trim().is_empty() {
        return Err(RouteInvocationInputError::InvalidInputId);
    }
    serde_json::to_writer(
        InvocationInputBudget {
            remaining: MAX_INVOCATION_INPUT_BYTES,
        },
        input,
    )
    .map_err(|_| RouteInvocationInputError::TooLarge)
}

/// Validate and enqueue input from an already authorized invocation route.
#[cfg(test)]
pub fn enqueue_invocation_input(
    active: &super::ActivePluginInvocation,
    input: bcode_tool::ToolInvocationInput,
) -> Result<(), RouteInvocationInputError> {
    validate_invocation_input(&input)?;
    enqueue_validated_invocation_input(active, input)
}

fn enqueue_validated_invocation_input(
    active: &super::ActivePluginInvocation,
    input: bcode_tool::ToolInvocationInput,
) -> Result<(), RouteInvocationInputError> {
    active.inputs.try_send(input).map_err(|error| match error {
        mpsc::error::TrySendError::Full(_) => RouteInvocationInputError::QueueFull,
        mpsc::error::TrySendError::Closed(_) => RouteInvocationInputError::RouteClosed,
    })
}

/// Normalized plugin service result returned by application operations.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginServiceOperationResult {
    /// Opaque plugin-owned response payload.
    pub payload: Vec<u8>,
    /// Optional normalized plugin service error.
    pub error: Option<PluginServiceOperationError>,
}

/// Normalized plugin service error without implementation details.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginServiceOperationError {
    /// Stable plugin-owned error code.
    pub code: String,
    /// Secret-safe public error message.
    pub message: String,
}

/// Convert one internal plugin response into the normalized operation result.
#[must_use]
fn project_service_response(
    response: bcode_plugin::ServiceResponse,
) -> PluginServiceOperationResult {
    PluginServiceOperationResult {
        payload: response.payload,
        error: response.error.map(|error| PluginServiceOperationError {
            code: error.code,
            message: error.message,
        }),
    }
}

pub async fn invoke_service(
    state: &ServerState,
    plugin_id: &str,
    interface_id: &str,
    operation: String,
    payload: Vec<u8>,
) -> Result<PluginServiceOperationResult, PublicPluginError> {
    let plugin_id = plugin_id.to_owned();
    let interface_id = interface_id.to_owned();
    let mut payload = payload;
    let model_owner = if interface_id == "bcode.auth-broker/v1"
        && operation == "serve"
        && state.model_host.admits(&plugin_id)
    {
        let configured = state
            .startup_config
            .plugins
            .config
            .get(&plugin_id)
            .and_then(|table| table.get("serve"))
            .ok_or(PublicPluginError {
                code: "broker_config_missing",
                message: "broker serve configuration is missing",
            })?;
        let mut value = serde_json::to_value(configured).map_err(|_| PublicPluginError {
            code: "invalid_request",
            message: "invalid configured service request",
        })?;
        let object = value.as_object_mut().ok_or(PublicPluginError {
            code: "invalid_request",
            message: "invalid service request",
        })?;
        let invocation_id = uuid::Uuid::new_v4().to_string();
        object.insert(
            "invocation_id".into(),
            serde_json::Value::String(invocation_id.clone()),
        );
        payload = serde_json::to_vec(&value).map_err(|_| PublicPluginError {
            code: "invalid_request",
            message: "invalid service request",
        })?;
        Some((plugin_id.clone(), invocation_id))
    } else {
        None
    };
    if interface_id == "bcode.auth-broker/v1" && operation == "serve" && model_owner.is_none() {
        return Err(PublicPluginError {
            code: "model_host_not_admitted",
            message: "model host capability is disabled",
        });
    }
    let labels = plugin_service_metric_labels(Some(&plugin_id), &interface_id, &operation);
    let authorized_session_id =
        super::command_invocation_session(&interface_id, &operation, &payload);
    let (bridge, bridge_requests) = super::server_plugin_bridge();
    let activity_revision =
        if interface_id == bcode_session_models::ACTIVITY_PRESENTATION_INTERFACE_ID {
            Some(validate_activity_request(&operation, &payload)?)
        } else {
            None
        };
    let invocation = async {
        let call = state.plugins.invoke_service_with_bridge_scoped(
            &plugin_id,
            interface_id,
            operation,
            payload,
            bcode_plugin::PluginInvocationScope::Global,
            Some(bridge),
        );
        if model_owner.is_some() {
            tokio::select! {
                result = call => result,
                () = async {
                    while !state.shutdown_requested.load(std::sync::atomic::Ordering::SeqCst) {
                        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    }
                } => Err(bcode_plugin::PluginLoadError::ServiceInvocationTimeout {
                    plugin_id: plugin_id.clone(), timeout_ms: 0,
                }),
            }
        } else {
            call.await
        }
    };
    let owner_for_driver = model_owner.clone();
    let result = Box::pin(state.metrics.time_result_async(
        "plugin.service",
        labels,
        drive_service_bridge_owned(
            state,
            authorized_session_id,
            bridge_requests,
            invocation,
            owner_for_driver,
        ),
    ))
    .await
    .map(project_service_response)
    .map_err(|error| normalize_error(&error))?;
    if let Some(revision) = activity_revision.filter(|_| result.error.is_none()) {
        validate_activity_response(&plugin_id, revision, &result.payload)?;
    }
    Ok(result)
}

fn validate_activity_request(operation: &str, payload: &[u8]) -> Result<u64, PublicPluginError> {
    let invalid = || PublicPluginError {
        code: "invalid_activity_request",
        message: "activity projection request is unsupported or invalid",
    };
    if operation != bcode_session_models::OP_PROJECT_ACTIVITY
        || payload.len() > bcode_session_models::MAX_ACTIVITY_PROJECTION_REQUEST_BYTES
    {
        return Err(invalid());
    }
    let request: bcode_session_models::ActivityProjectionRequest =
        serde_json::from_slice(payload).map_err(|_| invalid())?;
    if request.revision == 0 {
        return Err(invalid());
    }
    Ok(request.revision)
}

fn validate_activity_response(
    producer: &str,
    revision: u64,
    payload: &[u8],
) -> Result<(), PublicPluginError> {
    let invalid = || PublicPluginError {
        code: "invalid_activity_presentation",
        message: "activity producer returned an unsupported or invalid presentation",
    };
    if payload.len() > bcode_session_models::MAX_ACTIVITY_PRESENTATION_BYTES {
        return Err(invalid());
    }
    let presentation: bcode_session_models::ActivityPresentation =
        serde_json::from_slice(payload).map_err(|_| invalid())?;
    if presentation.revision != revision {
        return Err(invalid());
    }
    presentation.validate(producer).map_err(|_| invalid())
}

struct ModelOwnerGuard {
    host: std::sync::Arc<super::model_host_operations::ModelHostOperations>,
    owner: Option<(String, String)>,
}

impl Drop for ModelOwnerGuard {
    fn drop(&mut self) {
        if let Some((caller, id)) = self.owner.take()
            && let Ok(runtime) = tokio::runtime::Handle::try_current()
        {
            let host = self.host.clone();
            runtime.spawn(async move {
                host.cancel_owner(&caller, &id).await;
            });
        }
    }
}

#[cfg(test)]
async fn drive_service_bridge<T>(
    state: &ServerState,
    authorized_session_id: Option<bcode_session_models::SessionId>,
    bridge_requests: mpsc::UnboundedReceiver<super::ServerPluginBridgeCall>,
    invocation: impl std::future::Future<Output = T>,
) -> T {
    drive_service_bridge_owned(
        state,
        authorized_session_id,
        bridge_requests,
        invocation,
        None,
    )
    .await
}

async fn drive_service_bridge_owned<T>(
    state: &ServerState,
    authorized_session_id: Option<bcode_session_models::SessionId>,
    mut bridge_requests: mpsc::UnboundedReceiver<super::ServerPluginBridgeCall>,
    invocation: impl std::future::Future<Output = T>,
    model_owner: Option<(String, String)>,
) -> T {
    let _owner_guard = ModelOwnerGuard {
        host: state.model_host.clone(),
        owner: model_owner.clone(),
    };
    tokio::pin!(invocation);
    loop {
        tokio::select! {
            result = &mut invocation => {
                if let Some((caller, id)) = &model_owner { state.model_host.cancel_owner(caller, id).await; }
                break result;
            },
            bridge_call = bridge_requests.recv() => {
                let Some(bridge_call) = bridge_call else {
                    let result = invocation.await;
                    if let Some((caller, id)) = &model_owner { state.model_host.cancel_owner(caller, id).await; }
                    return result;
                };
                let response = if let bcode_plugin_sdk::ServiceBridgeRequest::InvokeService(request) = &bridge_call.request
                    && request.interface_id == bcode_model::host_operation::INTERFACE_ID
                {
                    let result = match &model_owner {
                        Some((caller, id)) if request.invocation_id == *id && !bridge_call.cancellation.is_cancelled() => {
                            state.model_host.clone().dispatch(caller, request.clone()).await
                        }
                        _ => bcode_tool::ToolInvocationServiceResolution::Failed {
                            code: "model_host_not_admitted".into(), message: "model host invocation ownership is unavailable".into(),
                        },
                    };
                    Ok(bcode_plugin_sdk::ServiceBridgeResponse::Service(result))
                } else if let bcode_plugin_sdk::ServiceBridgeRequest::InvokeService(request) =
                    &bridge_call.request
                    && request.route_id.as_deref()
                        == Some(bcode_model::judgement::WORKFLOW_APPLICATION_INTERFACE_ID)
                    && request.interface_id
                        == bcode_model::judgement::WORKFLOW_APPLICATION_INTERFACE_ID
                    && request.operation == bcode_model::judgement::OP_OPTIONAL_JUDGE
                {
                    let cancel = super::TurnCancelState::default();
                    let result = tokio::select! {
                        biased;
                        () = async {
                            while !bridge_call.cancellation.is_cancelled() {
                                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                            }
                        } => bcode_tool::ToolInvocationServiceResolution::Cancelled,
                        result = super::optional_judgement_payload_resolution(
                            state,
                            &state.startup_config,
                            request.payload.clone(),
                            &cancel,
                            None,
                        ) => result,
                    };
                    Ok(bcode_plugin_sdk::ServiceBridgeResponse::Service(result))
                } else {
                    super::resolve_command_plugin_bridge_request(
                        state,
                        authorized_session_id,
                        bridge_call.request,
                        &bridge_call.cancellation,
                    ).await
                };
                let _sent = bridge_call.response.send(response);
            }
        }
    }
}

/// Call the unique provider of one typed plugin service interface.
pub async fn call_service(
    state: &ServerState,
    interface_id: &str,
    operation: String,
    payload: Vec<u8>,
) -> Result<PluginServiceOperationResult, PublicPluginError> {
    // Activity identity is scoped to an explicitly selected producer. Interface discovery
    // alone must not bypass the producer/revision checks in `invoke_service`.
    if interface_id == bcode_session_models::ACTIVITY_PRESENTATION_INTERFACE_ID {
        return Err(PublicPluginError {
            code: "activity_producer_required",
            message: "activity projection requires an explicitly selected producer",
        });
    }
    let plugin_id = state
        .plugins
        .registry()
        .service_registry()
        .unique_provider(interface_id)
        .map_err(|error| normalize_error(&error))?
        .to_owned();
    invoke_service(state, &plugin_id, interface_id, operation, payload).await
}

/// Publish one plugin event through host routing.
pub async fn publish_event(
    state: &ServerState,
    topic: &str,
    payload: &[u8],
) -> Result<usize, PublicPluginError> {
    state
        .metrics
        .time_result_async(
            "plugin.event_delivery",
            plugin_event_metric_labels(topic),
            state.plugins.publish_event(topic, payload),
        )
        .await
        .map_err(|error| normalize_error(&error))
}

#[cfg(test)]
mod tests {
    #[test]
    fn invocation_input_budget_counts_envelope_escaping_and_utf8() {
        for character in ['a', '\n', 'λ'] {
            let mut input = bcode_tool::ToolInvocationInput {
                invocation_id: "invocation".into(),
                input_id: "input".into(),
                producer_id: "producer".into(),
                schema: "schema".into(),
                schema_version: 1,
                payload: serde_json::json!(""),
            };
            let overhead = serde_json::to_vec(&input).unwrap().len();
            let width = serde_json::to_string(&character.to_string()).unwrap().len() - 2;
            let remaining = super::MAX_INVOCATION_INPUT_BYTES - overhead;
            let mut payload = character.to_string().repeat(remaining / width);
            payload.push_str(&"a".repeat(remaining % width));
            input.payload = serde_json::json!(payload);
            assert_eq!(
                serde_json::to_vec(&input).unwrap().len(),
                super::MAX_INVOCATION_INPUT_BYTES
            );
            assert_eq!(super::validate_invocation_input(&input), Ok(()));
            payload.push('a');
            input.payload = serde_json::json!(payload);
            assert_eq!(
                super::validate_invocation_input(&input),
                Err(super::RouteInvocationInputError::TooLarge)
            );
        }
    }

    #[test]
    fn activity_response_requires_selected_producer_and_supported_envelope() {
        let mut view = bcode_session_models::ActivityPresentation {
            version: bcode_session_models::ACTIVITY_PRESENTATION_VERSION,
            producer: "example.plugin".into(),
            activity_id: "activity:1".into(),
            revision: 1,
            schema: "unknown.schema".into(),
            schema_version: 99,
            fallback: "Readable fallback".into(),
            payload: serde_json::json!({}),
        };
        let bytes = serde_json::to_vec(&view).unwrap();
        assert!(super::validate_activity_response("example.plugin", 1, &bytes).is_ok());
        assert!(super::validate_activity_response("other.plugin", 1, &bytes).is_err());
        assert!(super::validate_activity_response("example.plugin", 2, &bytes).is_err());
        view.version = 99;
        assert!(
            super::validate_activity_response(
                "example.plugin",
                1,
                &serde_json::to_vec(&view).unwrap()
            )
            .is_err()
        );
        assert!(super::validate_activity_response("example.plugin", 1, b"not json").is_err());
        assert!(
            super::validate_activity_response(
                "example.plugin",
                1,
                &vec![b' '; bcode_session_models::MAX_ACTIVITY_PRESENTATION_BYTES + 1]
            )
            .is_err()
        );
    }

    #[tokio::test]
    async fn workflow_activity_selection_is_optional_and_fail_open() {
        let mut state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        state.plugins = bcode_plugin::PluginRuntimeHost::load_defaults_with_static_bundled(
            &bcode_plugin::PluginSelection {
                mode: bcode_plugin::PluginSelectionMode::Explicit,
                enabled: std::collections::BTreeSet::from(["bcode.loop".into()]),
                disabled: std::collections::BTreeSet::new(),
            },
            &[bcode_bundled_plugins::static_loop_plugin()],
        )
        .unwrap();
        let mut configuration =
            bcode_workflow::WorkflowPromptConfiguration::preserve_input("build", "unchanged");
        let input = serde_json::json!({"implementation_prompt":"Implement safely", "stop_condition":"Tests pass", "max_iterations":10, "iteration":2, "condition_met":false, "summary":"", "evidence":[]});
        let prompt =
            crate::workflow_prompt_input_message(&configuration.system_prompt, "", &input).unwrap();
        assert!(
            crate::workflow_prompt_activity(&state, &configuration, &input)
                .await
                .is_none()
        );
        configuration.activity_producer = Some(bcode_workflow::WorkflowActivityProducer {
            plugin: "bcode.loop".into(),
            stage: "implementation".into(),
        });
        let presentation = crate::workflow_prompt_activity(&state, &configuration, &input)
            .await
            .unwrap();
        assert_eq!(presentation.activity_id, "iteration:2");
        assert_eq!(presentation.payload["exact_structured_input"], input);
        assert_eq!(
            crate::workflow_prompt_input_message(&configuration.system_prompt, "", &input).unwrap(),
            prompt
        );
        configuration.activity_producer.as_mut().unwrap().plugin = "missing.plugin".into();
        let missing = crate::workflow_prompt_activity(&state, &configuration, &input).await;
        drop(state);
        assert!(missing.is_none());
    }

    #[tokio::test]
    async fn activity_round_trip_through_bundled_loop_producer() {
        let mut state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        state.plugins = bcode_plugin::PluginRuntimeHost::load_defaults_with_static_bundled(
            &bcode_plugin::PluginSelection {
                mode: bcode_plugin::PluginSelectionMode::Explicit,
                enabled: std::collections::BTreeSet::from(["bcode.loop".into()]),
                disabled: std::collections::BTreeSet::new(),
            },
            &[bcode_bundled_plugins::static_loop_plugin()],
        )
        .unwrap();
        let request = bcode_session_models::ActivityProjectionRequest {
            working_document: None,
            stage: "implementation".into(),
            revision: 7,
            input: serde_json::json!({"implementation_prompt":"Implement safely", "stop_condition":"Tests pass", "max_iterations":10, "iteration":2, "condition_met":false, "summary":"", "evidence":[]}),
        };
        let response = super::invoke_service(
            &state,
            "bcode.loop",
            bcode_session_models::ACTIVITY_PRESENTATION_INTERFACE_ID,
            "project".into(),
            serde_json::to_vec(&request).unwrap(),
        )
        .await
        .unwrap();
        drop(state);
        assert!(response.error.is_none());
        let view: bcode_session_models::ActivityPresentation =
            serde_json::from_slice(&response.payload).unwrap();
        assert_eq!(view.producer, "bcode.loop");
        assert_eq!(view.activity_id, "iteration:2");
        assert_eq!(view.revision, 7);
        assert_eq!(view.payload["exact_structured_input"], request.input);
        assert!(view.fallback.contains("Iteration 2 of 10"));
    }

    #[tokio::test]
    async fn activity_interface_discovery_cannot_bypass_producer_validation() {
        let state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        let result = super::call_service(
            &state,
            bcode_session_models::ACTIVITY_PRESENTATION_INTERFACE_ID,
            "project".into(),
            Vec::new(),
        )
        .await
        .unwrap_err();
        assert_eq!(result.code, "activity_producer_required");
        let unrelated =
            super::call_service(&state, "example.other/v1", "project".into(), Vec::new())
                .await
                .unwrap_err();
        drop(state);
        assert_eq!(unrelated.code, "plugin_error");
    }

    #[tokio::test]
    async fn activity_invocation_validates_before_plugin_lookup() {
        let state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        let interface = bcode_session_models::ACTIVITY_PRESENTATION_INTERFACE_ID;
        let invalid = super::invoke_service(
            &state,
            "missing.plugin",
            interface,
            "project".into(),
            b"invalid".to_vec(),
        )
        .await
        .unwrap_err();
        assert_eq!(invalid.code, "invalid_activity_request");
        let request = bcode_session_models::ActivityProjectionRequest {
            working_document: None,
            stage: "producer_stage".into(),
            revision: 1,
            input: serde_json::json!({}),
        };
        let missing = super::invoke_service(
            &state,
            "missing.plugin",
            interface,
            "project".into(),
            serde_json::to_vec(&request).unwrap(),
        )
        .await
        .unwrap_err();
        assert_eq!(missing.code, "plugin_error");
        let unrelated = super::invoke_service(
            &state,
            "missing.plugin",
            "example.other/v1",
            "project".into(),
            b"invalid".to_vec(),
        )
        .await
        .unwrap_err();
        drop(state);
        assert_eq!(unrelated.code, "plugin_error");
    }

    #[test]
    fn activity_request_is_bounded_and_requires_nonzero_revision() {
        let mut request = bcode_session_models::ActivityProjectionRequest {
            working_document: None,
            stage: "producer_defined".into(),
            revision: 3,
            input: serde_json::json!({}),
        };
        let bytes = serde_json::to_vec(&request).unwrap();
        assert_eq!(
            super::validate_activity_request("project", &bytes).unwrap(),
            3
        );
        assert!(super::validate_activity_request("unknown", &bytes).is_err());
        assert!(super::validate_activity_request("project", b"invalid").is_err());
        request.revision = 0;
        assert!(
            super::validate_activity_request("project", &serde_json::to_vec(&request).unwrap())
                .is_err()
        );
        assert!(
            super::validate_activity_request(
                "project",
                &vec![b' '; bcode_session_models::MAX_ACTIVITY_PROJECTION_REQUEST_BYTES + 1]
            )
            .is_err()
        );
    }

    #[test]
    fn invocation_input_budget_rejects_overflow_without_consuming_budget() {
        use std::io::Write as _;
        let mut budget = super::InvocationInputBudget { remaining: 3 };
        assert!(budget.write_all(b"four").is_err());
        assert_eq!(budget.remaining, 3);
        budget.write_all(b"abc").unwrap();
        assert_eq!(budget.remaining, 0);
        assert!(budget.write_all(b"x").is_err());
    }
    #[tokio::test]
    async fn service_bridge_drains_queued_request_before_closure() {
        let state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        let (sender, requests) = tokio::sync::mpsc::unbounded_channel();
        let (response, received) = std::sync::mpsc::sync_channel(1);
        sender
            .send(crate::ServerPluginBridgeCall {
                request: bcode_plugin_sdk::ServiceBridgeRequest::Exchange(
                    bcode_tool::ToolExchangeRequest {
                        invocation_id: "test-invocation".to_owned(),
                        exchange_id: "test-exchange".to_owned(),
                        producer_id: "test".to_owned(),
                        schema: "test.exchange".to_owned(),
                        schema_version: 1,
                        payload: serde_json::Value::Null,
                        response_policy: bcode_tool::ToolExchangeResponsePolicy::Required,
                    },
                ),
                cancellation: bcode_plugin_sdk::ServiceCancellation::default(),
                response,
            })
            .expect("queue bridge request");
        drop(sender);
        let (complete, result) = tokio::sync::oneshot::channel::<u32>();
        let value = async {
            let driver = super::drive_service_bridge(&state, None, requests, result);
            tokio::pin!(driver);
            let mut context = std::task::Context::from_waker(std::task::Waker::noop());
            assert!(std::future::Future::poll(driver.as_mut(), &mut context).is_pending());
            assert_eq!(
                received.try_recv().expect("queued request answered"),
                Err(
                    "command invocation bridge supports nested application services only"
                        .to_owned()
                )
            );
            complete.send(42).expect("complete invocation");
            driver.await.expect("invocation result")
        }
        .await;
        drop(state);
        assert_eq!(value, 42);
    }

    #[tokio::test]
    async fn public_service_bridge_routes_only_optional_application_requests() {
        use bcode_model::judgement;
        let mut state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        state.startup_config.judgement.mode = bcode_config::judgement::JudgementMode::Off;
        for (operation, cancelled) in [
            (judgement::OP_OPTIONAL_JUDGE, false),
            (judgement::OP_OPTIONAL_JUDGE, true),
            (judgement::OP_WORKFLOW_JUDGE, false),
        ] {
            let (sender, requests) = tokio::sync::mpsc::unbounded_channel();
            let (response, received) = std::sync::mpsc::sync_channel(1);
            let cancellation = bcode_plugin_sdk::ServiceCancellation::default();
            if cancelled {
                cancellation.cancel();
            }
            sender.send(crate::ServerPluginBridgeCall {
                request: bcode_plugin_sdk::ServiceBridgeRequest::InvokeService(
                    bcode_tool::ToolInvocationServiceRequest {
                        invocation_id: "optional-test".into(),
                        request_id: "optional-test-request".into(),
                        route_id: Some(judgement::WORKFLOW_APPLICATION_INTERFACE_ID.into()),
                        interface_id: judgement::WORKFLOW_APPLICATION_INTERFACE_ID.into(),
                        operation: operation.into(),
                        payload: serde_json::json!({"feature":"recommendations",
                            "request":{"model_id":"jev-1.13.0","state":{"kind":"text","value":"bounded"},
                                "questions":{"q":{"kind":"yes_no","instructions":"Relevant?"}}}}),
                    }),
                cancellation,
                response,
            }).unwrap();
            drop(sender);
            let (complete, result) = tokio::sync::oneshot::channel::<()>();
            let driver = super::drive_service_bridge(&state, None, requests, result);
            tokio::pin!(driver);
            let mut context = std::task::Context::from_waker(std::task::Waker::noop());
            assert!(std::future::Future::poll(driver.as_mut(), &mut context).is_pending());
            let answer = received.try_recv().unwrap().unwrap();
            if cancelled {
                assert!(matches!(
                    answer,
                    bcode_plugin_sdk::ServiceBridgeResponse::Service(
                        bcode_tool::ToolInvocationServiceResolution::Cancelled
                    )
                ));
            } else if operation == judgement::OP_WORKFLOW_JUDGE {
                assert!(matches!(
                    answer,
                    bcode_plugin_sdk::ServiceBridgeResponse::Service(
                        bcode_tool::ToolInvocationServiceResolution::Unsupported
                    )
                ));
            } else {
                let bcode_plugin_sdk::ServiceBridgeResponse::Service(
                    bcode_tool::ToolInvocationServiceResolution::Responded { payload },
                ) = answer
                else {
                    panic!("optional response expected");
                };
                let response: judgement::OptionalResponse =
                    serde_json::from_value(payload).unwrap();
                let observation = response.observation.unwrap();
                assert_eq!(
                    observation.outcome,
                    judgement::OptionalOperationOutcome::Disabled
                );
                assert_eq!(observation.inference_invocations, 0);
            }
            complete.send(()).unwrap();
            driver.await.unwrap();
        }
        drop(state);
    }

    #[tokio::test]
    async fn closed_service_bridge_yields_until_invocation_completes() {
        let state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        let (bridge, requests) = crate::server_plugin_bridge();
        drop(bridge);
        let (complete, result) = tokio::sync::oneshot::channel::<u32>();
        let value = async {
            let driver = super::drive_service_bridge(&state, None, requests, result);
            tokio::pin!(driver);
            let waker = std::task::Waker::noop();
            let mut context = std::task::Context::from_waker(waker);
            assert!(std::future::Future::poll(driver.as_mut(), &mut context).is_pending());
            complete.send(42).expect("invocation still awaited");
            driver.await.expect("invocation result")
        }
        .await;
        drop(state);
        assert_eq!(value, 42);
    }

    #[tokio::test]
    async fn closed_service_bridge_preserves_invocation_failure() {
        let state = crate::tests::test_server_state(bcode_session::SessionManager::default());
        let (bridge, requests) = crate::server_plugin_bridge();
        drop(bridge);
        let (complete, result) = tokio::sync::oneshot::channel::<u32>();
        let outcome = async {
            let driver = super::drive_service_bridge(&state, None, requests, result);
            tokio::pin!(driver);
            let mut context = std::task::Context::from_waker(std::task::Waker::noop());
            assert!(std::future::Future::poll(driver.as_mut(), &mut context).is_pending());
            drop(complete);
            driver.await
        }
        .await;
        drop(state);
        assert!(outcome.is_err());
    }
}