plugins/agent/ports/src/model.rs

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

//! Model protocol implementations captured by their execution owner.
use futures_util::future::BoxFuture;
pub use maka_agent_facts::model::ModelEvent;
pub use maka_agent_facts::model::error::ModelError as Error;
pub use maka_agent_facts::model::prompt::Message;
pub use maka_agent_facts::model::request::Credentials;
pub use maka_agent_facts::model::request::Provider;
pub use maka_agent_facts::model::request::ProviderKind;
pub use maka_agent_facts::model::request::Request;
use maka_capabilities::http;
use serde::{Deserialize, Serialize};
use std::{sync::Arc, time::Duration};
use tokio_util::sync::CancellationToken;

/// Scoped I/O is supplied after Host model admission. Implementations cannot
/// replace routing policy or reconstruct this capability from a Session ID.
pub trait Transport: Send + Sync {
    /// Process-local routing identity, excluding the serialized credential.
    /// A changed routing policy must invalidate cached connections.
    fn identity(&self) -> u64;
    fn request(&self, request: http::Request) -> BoxFuture<'_, Result<http::Response, Error>>;
    fn connect(&self, request: Connect) -> BoxFuture<'_, Result<Arc<dyn Socket>, Error>>;
}
#[derive(Clone, Serialize, Deserialize)]
pub struct Connect {
    pub url: String,
    #[serde(default)]
    pub headers: Vec<(String, String)>,
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum Frame {
    Text(String),
    Binary(Vec<u8>),
}
/// One outstanding reader and writer. Cancelling a pending receive must not
/// consume the next frame; dropping the last handle closes the connection.
pub trait Socket: Send + Sync {
    fn send(&self, frame: Frame) -> BoxFuture<'_, Result<(), Error>>;
    fn receive(&self) -> BoxFuture<'_, Result<Option<Frame>, Error>>;
    fn close(&self) -> BoxFuture<'_, Result<(), Error>>;
}
/// A model event and the in-flight bytes it owns. Consumers release the
/// reservation after journaling; retained response text has a separate lifetime.
#[derive(Debug)]
pub struct Emission {
    pub event: ModelEvent,
    pub reservation: Option<Arc<tokio::sync::OwnedSemaphorePermit>>,
}

impl From<ModelEvent> for Emission {
    fn from(event: ModelEvent) -> Self {
        Self {
            event,
            reservation: None,
        }
    }
}

pub trait Events: Send + Sync {
    /// Observed provider progress (including heartbeats and partial wire items)
    /// resets the idle deadline without inventing a canonical model event.
    fn progress(&self) {}
    /// Backpressure is observed before the next event is produced.
    fn emit(&self, event: Emission) -> BoxFuture<'_, Result<(), Error>>;
}
#[derive(Clone)]
pub struct Context {
    pub transport: Arc<dyn Transport>,
    pub events: Arc<dyn Events>,
    pub cancellation: CancellationToken,
    pub idle_timeout: Duration,
}
/// Disposable per-conversation optimization state, never a history authority.
pub trait Session: Send + Sync {
    fn stream(&self, request: Request, context: Context) -> BoxFuture<'static, Result<(), Error>>;
    /// Select an unused fallback after the caller exhausts replay-safe retries
    /// and drains the failed stream. Returns true only when transport changes;
    /// this never sends a request or authorizes replay itself.
    fn try_switch_fallback_transport(&self) -> bool {
        false
    }
    fn needs_confirmation(&self) -> bool {
        false
    }
    /// Check committed history projected by Host after local tool settlement.
    /// This is not a plugin claim or permission to change canonical history.
    fn confirm(
        &self,
        _prompt: &[Message],
        _settled_tool_call_ids: &[&str],
        _response_id: Option<&str>,
    ) -> bool {
        false
    }
}
#[derive(Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Lifetime {
    Request,
    Conversation,
}
pub trait ProviderAdapter: Send + Sync {
    /// Complete stream items have provider-confirmed boundaries, rather than
    /// being synthesized by an SDK when an incomplete stream is closed.
    fn supports_item_acceptance(&self) -> bool {
        false
    }

    /// Owns idle deadlines for external I/O and returns protocol-aware failures.
    /// Local event backpressure is not provider silence. Other adapters retain
    /// the executor watchdog; cancellation and cleanup always remain bounded.
    fn enforces_idle_timeout(&self) -> bool {
        false
    }

    fn open(
        &self,
        lifetime: Lifetime,
        cancellation: CancellationToken,
    ) -> BoxFuture<'_, Result<Arc<dyn Session>, Error>>;
}
/// One immutable protocol binding, retained through retries and stream cleanup.
#[derive(Clone)]
pub struct Binding(Arc<BoundAdapter>);
struct BoundAdapter {
    provider: Arc<dyn ProviderAdapter>,
    source: maka_agent_facts::composition::SourceRevision,
}
impl Binding {
    pub fn new(
        provider: Arc<dyn ProviderAdapter>,
        source: maka_agent_facts::composition::SourceRevision,
    ) -> Self {
        Self(Arc::new(BoundAdapter { provider, source }))
    }
    pub fn supports_item_acceptance(&self) -> bool {
        self.0.provider.supports_item_acceptance()
    }
    pub fn enforces_idle_timeout(&self) -> bool {
        self.0.provider.enforces_idle_timeout()
    }
    pub async fn open(
        &self,
        lifetime: Lifetime,
        cancellation: CancellationToken,
    ) -> Result<Arc<dyn Session>, Error> {
        self.0.provider.open(lifetime, cancellation).await
    }
    pub fn same_registration(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.0, &other.0)
    }
    pub fn source(&self) -> &maka_agent_facts::composition::SourceRevision {
        &self.0.source
    }
}