plugins/agent/native/src/owner.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.
 */

use crate::{
    Configuration, IMPLEMENTATION, NativeContext, configuration_schema, contract::*, workspace,
};
use futures_util::future::BoxFuture;
use maka_agent_engine::{Engine, RunError, RunInput, RunWork};

mod assistant;
mod attachments;
mod auxiliary;
mod compaction;
mod context;
mod execution;
mod files;
mod fork;
mod goal;
mod graph;
mod imports;
mod interactions;
mod lifecycle;
mod notifications;
mod plan;
mod preferences;
mod projects;
pub(crate) use notifications::ReturnRoute;
mod git;
mod global_instructions;
mod patches;
mod queue;
mod recall;
mod recap;
mod runs;
mod scheduler;
mod skills;
mod title;
mod todo;
mod work;
mod worktrees;
use maka_agent_facts::composition::SourceKind;
use maka_agent_facts::composition::SourceRevision;
use maka_agent_facts::configuration::policy::RuntimePolicySnapshot;
use maka_agent_facts::event::Invocation;
use maka_agent_facts::execution::*;
use maka_agent_facts::input::DeliveredMessage;
use maka_agent_facts::input::MessageInput;
use maka_agent_facts::language::{Language, LanguagePreference};
use maka_agent_facts::message::*;
use maka_config::ConfigurationStore;
use maka_event_log::{EventLog, sessions::SessionRecord};
use maka_model::ModelExecutor;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::{
    collections::{BTreeMap, BTreeSet},
    sync::Arc,
    time::Duration,
};
use tokio::sync::Mutex;
use tokio_util::{sync::CancellationToken, task::TaskTracker};
use uuid::Uuid;

use maka_agent_api::session::SessionConfiguration;
/// Frozen accepted message prerequisites, interpreted only by this Agent.
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct MessageBasis {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    compaction_prompt: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    graph: Option<maka_graph::Epoch>,
    #[serde(default, skip_serializing_if = "is_agent_collaboration")]
    collaboration_mode: CollaborationMode,
    #[serde(default, skip_serializing_if = "BackgroundKind::is_schedule")]
    background_kind: BackgroundKind,
    #[serde(default)]
    tool_discovery: bool,
    #[serde(default)]
    allowed_tools: Option<Vec<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    background: Option<ScheduledTarget>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    internal_tools: Vec<maka_agent_facts::tools::ToolDefinition>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    internal_bindings: BTreeMap<String, Option<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    behavior: Option<SystemPrompt>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    model: Option<crate::model::AcceptedModel>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    executor: Option<crate::executor::Accepted>,
    workspace: Workspace,
    fingerprint: String,
    tools: Vec<crate::tools::SelectedTool>,
    #[serde(default, skip_serializing_if = "Language::is_english")]
    language: Language,
}
fn is_agent_collaboration(mode: &CollaborationMode) -> bool {
    *mode == CollaborationMode::Agent
}
#[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum BackgroundKind {
    #[default]
    Schedule,
    Goal(Uuid),
    Plan(String),
}
impl BackgroundKind {
    fn is_schedule(&self) -> bool {
        *self == Self::Schedule
    }
}
enum Selection {
    Model(ModelSelection, Option<ThinkingLevel>),
    Executor(maka_agent_api::executor::Selection),
}
struct Submission {
    id: String,
    text: String,
    attachments: Vec<String>,
    selection: Selection,
}
enum PreparedBackend {
    Model(Box<(crate::model::Source, maka_agent_engine::PreparedModel)>),
    Executor(Box<crate::executor::Backend>),
}
struct Control {
    compacting: bool,
    invocation: Invocation,
    basis: MessageBasis,
    cancellation: CancellationToken,
    failure: Option<&'static str>,
}
#[derive(Default)]
struct Runs {
    closing: bool,
    controls: BTreeMap<String, Control>,
    /// Sessions whose original execution has not confirmed physical cleanup.
    quarantined: BTreeSet<String>,
}
pub(crate) struct Owner {
    graphs: Option<Arc<crate::workhub::tasks::graph::Graphs>>,
    pub store: Arc<EventLog>,
    pub recall: Arc<recall::Recall>,
    scheduler: Arc<scheduler::Scheduler>,
    todo: Arc<todo::Todo>,
    goals: Arc<goal::Goals>,
    plans: Arc<plan::Plans>,
    recap_storage: Arc<maka_host_runtime::storage::Storage>,
    auxiliary: auxiliary::Auxiliary,
    pub settings: Configuration,
    pub prefix: String,
    pub data_source: DataSource,
    pub(crate) configuration: Arc<ConfigurationStore>,
    pub(crate) catalog: Arc<maka_providers::catalog::Catalog>,
    credentials: crate::model::Credentials,
    revision: SourceRevision,
    behavior: Option<BuiltinBehavior>,
    configured_locale: LanguagePreference,
    native_language: Language,
    pub workspace_roots: Vec<workspace::Root>,
    instruction_home: Option<Arc<cap_std::fs::Dir>>,
    pub import_roots: Vec<workspace::Root>,
    internal_tools: Vec<maka_agent_facts::tools::ToolDefinition>,
    internal_bindings: BTreeMap<String, Option<String>>,
    managed_worktrees: std::path::PathBuf,
    engine: Engine,
    tools: maka_tools::ToolCatalog,
    tool_basis: Vec<crate::tools::SelectedTool>,
    editing_tools: Vec<(String, maka_agent_facts::execution::EditingTools)>,
    planning_tools: Vec<String>,
    skills: Option<skills::Library>,
    executors: crate::executor::Executors,
    runs: Arc<runs::Registry>,
    workers: TaskTracker,
    producers: TaskTracker,
    producer_gate: Arc<Mutex<()>>,
    #[cfg(test)]
    title_write_gate: std::sync::Mutex<Option<Arc<title::WriteGate>>>,
    #[cfg(all(test, target_os = "macos"))]
    file_native_hints: std::sync::atomic::AtomicBool,
    stop: CancellationToken,
}
#[derive(Clone, Copy)]
pub(crate) enum BuiltinBehavior {
    WorkHub,
}

impl Owner {
    pub(crate) fn resolve_language(&self, snapshot: &RuntimePolicySnapshot) -> Language {
        snapshot
            .policy
            .language
            .unwrap_or(self.configured_locale)
            .resolve(self.native_language)
    }

    pub(crate) async fn executor_setup(
        &self,
        input: maka_agent_api::executor::setup::Input,
        context: maka_host_runtime::application::Context,
    ) -> Result<Value, String> {
        self.executors.setup(input, context).await
    }

    pub async fn open(
        context: NativeContext,
        input: Value,
        catalog: Arc<maka_providers::catalog::Catalog>,
    ) -> Result<Arc<Self>, String> {
        Self::open_behavior(context, input, None, vec![], BTreeMap::new(), None, catalog).await
    }
    pub(crate) async fn open_behavior(
        context: NativeContext,
        input: Value,
        behavior: Option<BuiltinBehavior>,
        mut additional: Vec<maka_agent_facts::tools::ToolRegistration>,
        internal_bindings: BTreeMap<String, Option<String>>,
        graphs: Option<Arc<crate::workhub::tasks::graph::Graphs>>,
        catalog: Arc<maka_providers::catalog::Catalog>,
    ) -> Result<Arc<Self>, String> {
        let implementation = match behavior {
            Some(BuiltinBehavior::WorkHub) => maka_agent_api::workhub::IMPLEMENTATION,
            None => IMPLEMENTATION,
        };
        if !jsonschema_valid(&input) {
            return Err("agent_configuration_invalid".into());
        }
        let settings: Configuration =
            serde_json::from_value(input).map_err(|_| "agent_configuration_invalid")?;
        let persistence = context.persistence().ok_or("agent_requires_persistence")?;
        let root = persistence
            .store()
            .owned_root()
            .map_err(|_| "agent_root_unavailable")?;
        let data_source = DataSource {
            state_domain: persistence.source().state_domain.clone(),
            deployment: persistence.source().deployment,
            namespace: context.activation().instance().clone(),
        };
        let prefix = maka_agent_authority::session_namespace(
            &persistence.source().state_domain,
            persistence.source().deployment,
            context.activation().instance(),
            implementation != IMPLEMENTATION,
        );
        let managed_worktrees = root.canonical_path().join("subagent-worktrees");
        let workspace_roots =
            workspace::open(settings.workspace_roots.clone(), &managed_worktrees)?;
        let import_roots = if settings.import_roots.is_empty() {
            Vec::new()
        } else {
            workspace::open(settings.import_roots.clone(), root.canonical_path())?
        };
        let cells = maka_tools::CodeExecutor::new(maka_code_runtime::CellLimits::default())
            .map_err(|_| "agent_executor_unavailable")?;
        let workers = TaskTracker::new();
        let stop = CancellationToken::new();
        let revision = SourceRevision {
            kind: SourceKind::ModelProvider,
            name: implementation.into(),
            package_id: context.package().id.clone(),
            entry_id: implementation.into(),
            activation: context.activation().epoch().to_string(),
            revision: context.package().revision.clone(),
        };
        // Match Codex's five-minute stream idle allowance for slow model steps.
        let models = ModelExecutor::for_source(Duration::from_secs(300), revision.clone())
            .map_err(|_| "agent_executor_unavailable")?;
        let auxiliary = auxiliary::Auxiliary::new(revision.clone())?;
        let configured_locale = LanguagePreference::from_locale(&settings.locale);
        let native_language = maka_config::language::native_language();
        let skills = skills::Library::select(&context, &settings, revision.clone())?;
        let instruction_home = assistant::open_home(settings.assistant.as_ref())?;
        let executors = crate::executor::Executors::capture(&context, &settings.executors)?;
        let configuration = Arc::new(
            ConfigurationStore::for_root(root)
                .await
                .map_err(|error| {
                    tracing::error!(stage = "configuration_store_open", cause = crate::diagnostics::config_cause(&error),
                        activation = %context.activation().epoch(), instance = ?context.activation().instance(),
                        "Agent configuration store failed to open");
                    "agent_configuration_unavailable"
                })?,
        );
        let recall = Arc::new(recall::Recall {
            store: persistence.store().clone(),
            configuration: configuration.clone(),
            prefix: prefix.clone(),
            data_source: data_source.clone(),
            enabled: settings.recall,
            stop: stop.clone(),
            workers: workers.clone(),
            services: context.services(),
        });
        if settings.recall {
            additional.extend(recall.tools());
        }
        let scheduler = Arc::new(scheduler::Scheduler::new(
            persistence.store().clone(),
            prefix.clone(),
        ));
        if settings.scheduled_tasks {
            additional.extend(scheduler.tools());
        }
        let plans = Arc::new(plan::Plans::new(
            persistence.store().clone(),
            prefix.clone(),
        ));
        if settings.plans {
            additional.extend(plans.tools());
        }
        let goals = Arc::new(goal::Goals::new(
            persistence.store().clone(),
            prefix.clone(),
        ));
        if settings.goals {
            additional.extend(goals.tools());
        }
        let todo = Arc::new(todo::Todo::new(persistence.store().clone(), prefix.clone()));
        let recap_storage = Arc::new(maka_host_runtime::storage::Storage::new(
            persistence.store().clone(),
            prefix.clone(),
            maka_capabilities::storage::Namespace::new(
                "maka.session-recap",
                maka_plugin_core::scope::Scope::Profile,
            )
            .expect("recap namespace"),
        ));
        if settings.todo {
            additional.extend(todo.tools());
        }
        let internal_tools = additional
            .iter()
            .map(|tool| tool.definition.clone())
            .collect();
        let prepared_tools = crate::tools::catalog(
            context.services(),
            persistence.store().clone(),
            configuration.clone(),
            &settings.tools,
            additional,
            workers.clone(),
        )
        .and_then(|mut catalog| {
            if settings.plans {
                catalog.planning.push("SubmitPlan".into());
                catalog.tools = catalog
                    .tools
                    .with_always_visible(&["SubmitPlan", "update_plan", "cancel_plan"])
                    .and_then(|tools| {
                        tools.with_request_binding(
                            ["SubmitPlan", "update_plan", "cancel_plan"]
                                .into_iter()
                                .map(str::to_owned)
                                .collect(),
                            plans.clone(),
                        )
                    })
                    .map_err(|error| error.to_string())?;
            }
            if graphs.as_ref().is_some_and(|graphs| graphs.enabled()) {
                catalog.tools = catalog
                    .tools
                    .with_always_visible(crate::workhub::tasks::graph::NAMES)
                    .map_err(|error| error.to_string())?;
            }
            if settings.goals {
                catalog.tools = catalog
                    .tools
                    .with_always_visible(&["GoalStatus"])
                    .map_err(|error| error.to_string())?;
            }
            if settings.recall {
                catalog
                    .planning
                    .extend(["Recall", "RecallMore", "RecallMaterial"].map(str::to_owned));
                catalog.tools = catalog
                    .tools
                    .with_request_binding(
                        maka_agent_facts::tools::ToolPreparer::names(recall.as_ref())
                            .into_iter()
                            .collect(),
                        recall.clone(),
                    )
                    .map_err(|error| error.to_string())?;
            }
            Ok(catalog)
        });
        let crate::tools::Catalog {
            tools,
            basis: tool_basis,
            editing: editing_tools,
            planning: planning_tools,
        } = match prepared_tools {
            Ok(tools) => tools,
            Err(error) => {
                configuration
                    .shutdown()
                    .await
                    .map_err(|error| {
                        tracing::error!(stage = "configuration_store_shutdown", cause = crate::diagnostics::config_cause(&error),
                            activation = %context.activation().epoch(), instance = ?context.activation().instance(),
                            "Agent configuration store cleanup failed");
                        "agent_configuration_cleanup_unconfirmed"
                    })?;
                return Err(error);
            }
        };
        let credentials = crate::model::Credentials {
            executor: models.clone(),
            refresh: Arc::new(crate::model::oauth::Authority::new(
                workers.clone(),
                stop.clone(),
            )),
        };
        let engine = Engine::with_pricing(
            persistence.store().clone(),
            models,
            cells,
            Arc::new(crate::model::Prices(configuration.clone())),
        );
        Ok(Arc::new(Self {
            graphs,
            plans,
            goals,
            auxiliary,
            todo,
            recap_storage,
            scheduler,
            recall,
            store: persistence.store().clone(),
            prefix,
            data_source,
            workspace_roots,
            import_roots,
            managed_worktrees,
            revision,
            behavior,
            configured_locale,
            native_language,
            instruction_home,
            internal_tools,
            internal_bindings,
            settings,
            configuration,
            catalog,
            credentials,
            engine,
            tools,
            tool_basis,
            editing_tools,
            planning_tools,
            skills,
            executors,
            runs: Arc::default(),
            workers,
            producers: TaskTracker::new(),
            producer_gate: Arc::default(),
            #[cfg(test)]
            title_write_gate: std::sync::Mutex::default(),
            #[cfg(all(test, target_os = "macos"))]
            file_native_hints: std::sync::atomic::AtomicBool::new(true),
            stop,
        }))
    }
    pub fn operation_id(&self, operation: &maka_plugin_core::operation::RecoveryRef) -> String {
        // Use the actual complete original reference. UUIDs alone do not name
        // operations across provider namespaces or deployments. The Host journal
        // still rejects conflicting reuse inside its original operation namespace.
        let bytes = serde_json::to_vec(operation).expect("admitted operation reference");
        let digest = hex::encode(Sha256::digest(bytes));
        format!("{}{}", self.prefix, &digest[..32])
    }
    pub fn owns(&self, id: &str) -> bool {
        owns_session(&self.prefix, id)
    }
    pub async fn require_session(&self, id: &str) -> Result<(), String> {
        if !self.owns(id) {
            return Err("agent_session_unavailable".into());
        }
        if self
            .store
            .session_is_available(id)
            .await
            .map_err(|_| "agent_store_unavailable".to_owned())?
        {
            Ok(())
        } else {
            Err("agent_session_unavailable".into())
        }
    }
    pub async fn session(&self, id: &str) -> Result<SessionRecord<SessionConfiguration>, String> {
        if !self.owns(id) {
            return Err("agent_session_unavailable".into());
        }
        self.store
            .get_session(id)
            .await
            .map_err(|_| "agent_store_unavailable".to_owned())?
            .ok_or("agent_session_unavailable".into())
    }
    pub async fn create(
        self: &Arc<Self>,
        input: Create,
        operation: &maka_plugin_core::operation::RecoveryRef,
        scope: &maka_capabilities::call::Scope,
    ) -> Result<Created, String> {
        let operation = operation.clone();
        self.resource_work(scope, move |owner, scope| {
            Box::pin(async move { owner.create_inner(input, &operation, &scope).await })
        })
        .await
    }
    async fn create_inner(
        &self,
        input: Create,
        operation: &maka_plugin_core::operation::RecoveryRef,
        scope: &maka_capabilities::call::Scope,
    ) -> Result<Created, String> {
        if let Some(reason) = metadata_rejection(&input.title, &input.labels) {
            return Ok(Created::Rejected { reason });
        }
        let runs = self.runs.lock().await;
        if runs.closing {
            return Ok(Created::Rejected {
                reason: Rejection::Closed,
            });
        }
        let mut tools = input.tools;
        if let Some(names) = &mut tools {
            names.sort();
            if names.windows(2).any(|pair| pair[0] == pair[1])
                || names.iter().any(|name| !self.tools.contains(name))
            {
                return Ok(Created::Rejected {
                    reason: Rejection::ToolsUnavailable,
                });
            }
        }
        let selected = match input.workspace {
            WorkspaceSelection::IsolatedGit { selection } => {
                let Some(workspace) = self.isolated_workspace(selection, operation, scope).await?
                else {
                    return Ok(Created::Rejected {
                        reason: Rejection::WorkspaceUnavailable,
                    });
                };
                workspace
            }
            WorkspaceSelection::Project { project, selection } => {
                let Some(workspace) = self.select_project(project, selection).await? else {
                    return Ok(Created::Rejected {
                        reason: Rejection::WorkspaceUnavailable,
                    });
                };
                workspace
            }
            WorkspaceSelection::Directory { selection } => {
                let Some(workspace) = self.create_project_workspace(selection).await? else {
                    return Ok(Created::Rejected {
                        reason: Rejection::WorkspaceUnavailable,
                    });
                };
                workspace
            }
            WorkspaceSelection::Session { session_id } => {
                if !self.owns(&session_id) {
                    return Ok(Created::Rejected {
                        reason: Rejection::SessionUnavailable,
                    });
                }
                let Some(session) = self
                    .store
                    .get_session::<SessionConfiguration>(&session_id)
                    .await
                    .map_err(|_| "agent_store_unavailable")?
                else {
                    return Ok(Created::Rejected {
                        reason: Rejection::SessionUnavailable,
                    });
                };
                if session.archived
                    || !self
                        .project_accepts_session(&session.configuration.workspace)
                        .await?
                {
                    return Ok(Created::Rejected {
                        reason: Rejection::SessionUnavailable,
                    });
                }
                if workspace::validate(
                    &self.workspace_roots,
                    &session.configuration.workspace,
                    &self.managed_worktrees,
                )
                .await
                .is_err()
                {
                    return Ok(Created::Rejected {
                        reason: Rejection::WorkspaceUnavailable,
                    });
                }
                if let Some(parent) = session.configuration.tools {
                    tools = Some(match tools {
                        None => parent,
                        Some(names) => names
                            .into_iter()
                            .filter(|name| parent.contains(name))
                            .collect(),
                    });
                }
                session.configuration.workspace
            }
        };
        let id = self.operation_id(operation);
        let record = self
            .store
            .create_session(
                &id,
                &operation.fingerprint,
                &SessionConfiguration {
                    backend: None,
                    collaboration_mode: CollaborationMode::Agent,
                    tools,
                    title: input.title,
                    title_is_manual: !input.automatic_title,
                    labels: input.labels,
                    workspace: selected,
                    permissions: self.settings.permissions.clone(),
                    boundary_revision: 0,
                },
                now(),
            )
            .await
            .map_err(|_| "agent_creation_failed")?;
        Ok(Created::Created {
            session: Box::new(crate::service::summary(self, record)),
        })
    }
    pub async fn mark_read(&self, input: MarkRead) -> Result<MarkReadResult, String> {
        let runs = self.runs.lock().await;
        if runs.closing {
            return Ok(MarkReadResult::Rejected {
                reason: Rejection::Closed,
            });
        }
        if !self.owns(&input.session_id) {
            return Ok(MarkReadResult::Rejected {
                reason: Rejection::SessionUnavailable,
            });
        }
        let session = self
            .store
            .get_session::<SessionConfiguration>(&input.session_id)
            .await
            .map_err(|_| "agent_store_unavailable")?;
        if session.is_none_or(|session| session.archived) {
            return Ok(MarkReadResult::Rejected {
                reason: Rejection::SessionUnavailable,
            });
        }
        // The original store compares the exposed message with the durable tail
        // inside its transaction. A stale reader cannot clear a newer message.
        let record = self
            .store
            .set_session_read_marker::<SessionConfiguration>(&input.session_id, &input.message_id)
            .await
            .map_err(|_| "agent_read_marker_failed")?;
        Ok(MarkReadResult::Observed {
            session: Box::new(crate::service::summary(self, record)),
        })
    }
    pub async fn metadata(&self, input: Metadata) -> Result<MetadataResult, String> {
        if let Some(reason) = metadata_rejection(&input.title, &input.labels) {
            return Ok(MetadataResult::Rejected { reason });
        }
        let runs = self.runs.lock().await;
        if runs.closing {
            return Ok(MetadataResult::Rejected {
                reason: Rejection::Closed,
            });
        }
        if !self.owns(&input.session_id) {
            return Ok(MetadataResult::Rejected {
                reason: Rejection::SessionUnavailable,
            });
        }
        let Some(session) = self
            .store
            .get_session::<SessionConfiguration>(&input.session_id)
            .await
            .map_err(|_| "agent_store_unavailable")?
        else {
            return Ok(MetadataResult::Rejected {
                reason: Rejection::SessionUnavailable,
            });
        };
        if session.archived {
            return Ok(MetadataResult::Rejected {
                reason: Rejection::SessionUnavailable,
            });
        }
        use maka_event_log::sessions::SessionMutation;
        let mode_changed = input
            .collaboration_mode
            .is_some_and(|mode| mode != session.configuration.collaboration_mode);
        if mode_changed {
            if self.graph_selected(&input.session_id).await? {
                return Ok(MetadataResult::Rejected {
                    reason: Rejection::Busy,
                });
            }
            if input.collaboration_mode == Some(CollaborationMode::Plan) && !self.settings.plans {
                return Ok(MetadataResult::Rejected {
                    reason: Rejection::ToolsUnavailable,
                });
            }
            if runs.controls.contains_key(&input.session_id)
                || !self.store.message_queue(&input.session_id).await.map_err(|_| "agent_queue_unavailable")?.entries.is_empty()
                || self.store.latest_turn_boundary(&input.session_id).await.map_err(|_| "agent_execution_unavailable")?
                    .is_some_and(|boundary| matches!(boundary.state, maka_event_log::turns::InvocationState::Ended { outcome: maka_agent_facts::event::InvocationOutcome::HandoffPaused { .. }, .. })) {
                return Ok(MetadataResult::Rejected { reason: Rejection::Busy });
            }
        }
        let permissions = input.permissions.is_some() || mode_changed;
        let update = move |configuration: &mut SessionConfiguration| {
            // Policy changes do not turn an automatic title into a manual title.
            if (input.collaboration_mode.is_none() && input.permissions.is_none())
                || configuration.title != input.title
            {
                configuration.title_is_manual = true;
            }
            configuration.title = input.title;
            configuration.labels = input.labels;
            if mode_changed
                || input.permissions.as_ref().is_some_and(|change| {
                    change.settings != configuration.permissions || change.revoke_grants
                })
            {
                configuration.boundary_revision = configuration
                    .boundary_revision
                    .checked_add(1)
                    .filter(|revision| *revision <= maka_agent_facts::interaction::MAX_SAFE_INTEGER)
                    .ok_or_else(|| {
                        maka_event_log::StoreError::InvalidTransition(
                            "permission boundary revision exhausted".into(),
                        )
                    })?;
            }
            if let Some(change) = input.permissions {
                configuration.permissions = change.settings;
            }
            if let Some(mode) = input.collaboration_mode {
                configuration.collaboration_mode = mode;
            }
            Ok(())
        };
        let changed = if permissions {
            self.store
                .update_session_permissions(
                    &input.session_id,
                    input.expected_revision,
                    now(),
                    update,
                )
                .await
        } else {
            self.store
                .update_session_metadata(&input.session_id, input.expected_revision, update)
                .await
        }
        .map_err(|_| "agent_metadata_failed")?;
        match changed {
            SessionMutation::Committed(record) => Ok(MetadataResult::Committed {
                session: Box::new(crate::service::summary(self, record)),
            }),
            SessionMutation::RevisionConflict { expected, actual } => {
                Ok(MetadataResult::RevisionConflict { expected, actual })
            }
        }
    }
    pub(super) fn tool_names(&self) -> crate::contract::ToolNames {
        crate::contract::ToolNames {
            names: self.tools.names(),
        }
    }
    fn selected_tools(
        &self,
        basis: &MessageBasis,
        discovery: bool,
        editing: Option<maka_agent_facts::execution::EditingTools>,
    ) -> maka_tools::ToolCatalog {
        let selected = plan::select(&self.tools, basis, &self.planning_tools);
        let selected = if basis.graph.is_some() {
            selected
        } else {
            selected.excluding(crate::workhub::tasks::graph::NAMES)
        };
        let selected = match basis.allowed_tools.as_deref() {
            None => selected.clone(),
            Some(allowed) => {
                let names = selected.names();
                let excluded: Vec<_> = names
                    .iter()
                    .filter(|name| !allowed.contains(name))
                    .map(String::as_str)
                    .collect();
                selected.excluding(&excluded)
            }
        };
        let selected = match editing {
            None => selected,
            Some(editing) => {
                let excluded: Vec<_> = self
                    .editing_tools
                    .iter()
                    .filter(|(_, style)| *style != editing)
                    .map(|(name, _)| name.as_str())
                    .collect();
                selected.excluding(&excluded)
            }
        };
        if discovery {
            selected.with_discovery()
        } else {
            selected
        }
    }
    pub async fn send(
        self: &Arc<Self>,
        input: Send,
        operation: &maka_plugin_core::operation::RecoveryRef,
        parent: &maka_capabilities::call::Scope,
    ) -> Result<crate::contract::SendResult, String> {
        let Send {
            session_id: id,
            message: text,
            model: selection,
            thinking_level,
            attachments,
        } = input;
        self.submit(
            Submission {
                id,
                text,
                attachments,
                selection: Selection::Model(selection, thinking_level),
            },
            operation,
            parent,
        )
        .await
    }
    pub async fn execute(
        self: &Arc<Self>,
        input: Execute,
        operation: &maka_plugin_core::operation::RecoveryRef,
        parent: &maka_capabilities::call::Scope,
    ) -> Result<SendResult, String> {
        self.submit(
            Submission {
                id: input.session_id,
                text: input.message,
                attachments: input.attachments,
                selection: Selection::Executor(input.executor),
            },
            operation,
            parent,
        )
        .await
    }
    async fn submit(
        self: &Arc<Self>,
        input: Submission,
        operation: &maka_plugin_core::operation::RecoveryRef,
        parent: &maka_capabilities::call::Scope,
    ) -> Result<SendResult, String> {
        let Submission {
            id,
            text,
            attachments,
            selection,
        } = input;
        if text.trim().is_empty() && attachments.is_empty() {
            return Err("agent_message_invalid".into());
        }
        let admission = self.runs.session(&id).await;
        let queued = {
            let runs = admission.state().await;
            if runs.closing {
                return Err("agent_closed".into());
            }
            if runs.quarantined.contains(&id) {
                return Err("agent_cleanup_unconfirmed".into());
            }
            let current = runs.controls.get(&id);
            if let Some(reason) = current.and_then(|control| control.failure) {
                return Err(reason.into());
            }
            current.map(|control| control.invocation.clone())
        };
        let session = self.session(&id).await?;
        if session.archived {
            return Err("agent_session_configuration_changed".into());
        }
        let backend = match selection {
            Selection::Model(selection, thinking_level) => PreparedBackend::Model(Box::new(
                crate::model::Source::select(
                    self.configuration.clone(),
                    self.catalog.clone(),
                    self.credentials.clone(),
                    selection,
                    thinking_level,
                    id.clone(),
                    self.revision.clone(),
                )
                .await
                .map_err(public_error)?,
            )),
            Selection::Executor(selection) => {
                if session.configuration.tools.is_some() {
                    return Err("agent_executor_cannot_enforce_tool_selection".into());
                }
                PreparedBackend::Executor(Box::new(self.executors.select(selection)?))
            }
        };
        let (model, executor) = match &backend {
            PreparedBackend::Model(value) => (Some(value.0.accepted.clone()), None),
            PreparedBackend::Executor(value) => (None, Some(value.accepted.clone())),
        };
        workspace::validate(
            &self.workspace_roots,
            &session.configuration.workspace,
            &self.managed_worktrees,
        )
        .await?;
        let external = executor.is_some();
        if session.configuration.collaboration_mode == CollaborationMode::Plan
            && (!self.settings.plans || external)
        {
            return Err("agent_planning_unavailable".into());
        }
        let snapshot = self
            .configuration
            .runtime_policy()
            .await
            .map_err(|_| "agent_preferences_unavailable")?;
        let language = self.resolve_language(&snapshot);
        let mut basis = MessageBasis {
            language,
            compaction_prompt: Some(self.effective_compaction_prompt(&snapshot, language)),
            graph: None,
            collaboration_mode: session.configuration.collaboration_mode,
            background_kind: BackgroundKind::Schedule,
            tool_discovery: !external && self.settings.tool_discovery,
            allowed_tools: session.configuration.tools.clone(),
            background: None,
            internal_tools: if external {
                vec![]
            } else {
                self.internal_tools.clone()
            },
            internal_bindings: if external {
                BTreeMap::new()
            } else {
                self.internal_bindings.clone()
            },
            behavior: self
                .capture_behavior(&session.configuration.workspace, &snapshot, language)
                .await?,
            model,
            executor,
            workspace: session.configuration.workspace.clone(),
            fingerprint: operation.fingerprint.clone(),
            tools: if external {
                vec![]
            } else {
                self.tool_basis.clone()
            },
        };
        self.prepare_plan_behavior(&id, &mut basis).await?;
        self.prepare_graph_behavior(&id, &mut basis, queued.is_none())
            .await?;
        let identity = self.operation_id(operation);
        let invocation = queued.clone().unwrap_or_else(|| Invocation {
            session_id: id.clone(),
            turn_id: format!("t_{identity}"),
            run_id: format!("r_{identity}"),
            invocation_id: format!("i_{identity}"),
        });
        let mut message: MessageInput = text.into();
        message.attachments = self.attachment_refs(&id, attachments).await?;
        let unprepared = message;
        let (message, required_tools) = match self
            .prepare_skills(
                unprepared.clone(),
                language,
                Some(session.configuration.workspace.clone()),
                parent,
                if basis.executor.is_some() {
                    vec![]
                } else {
                    self.selected_tools(
                        &basis,
                        false,
                        basis.model.as_ref().and_then(|model| model.editing_tools),
                    )
                    .names()
                },
            )
            .await?
        {
            skills::Input::Ready {
                message,
                required_tools,
            } => (message, required_tools),
            skills::Input::Rejected { reason, skills } => {
                return Ok(SendResult::Rejected {
                    session_id: id,
                    reason,
                    skills,
                });
            }
        };
        let delivered = RootSourceMessage {
            message: DeliveredMessage {
                origin: None,
                message_id: format!("m_{identity}"),
                submitted_content_digest: unprepared
                    .content_digest()
                    .map_err(|_| "agent_message_invalid")?,
                content: message.clone(),
            },
            unprepared_content: unprepared,
            submitted_placement: if queued.is_some() {
                Placement::NextTurn
            } else {
                Placement::CurrentTurn
            },
            disposition: if queued.is_some() {
                MessageDisposition::Followup
            } else {
                MessageDisposition::TurnStarted
            },
            submitted_intent: None,
        };
        let pending = maka_event_log::message_admissions::PendingMessageAdmission {
            execution: maka_event_log::message_admissions::PendingExecution {
                owner: self.prefix.clone(),
                basis: serde_json::to_value(&basis).expect("accepted message basis"),
            },
            invocation: invocation.clone(),
            steering_invocation: None,
            source: delivered.clone(),
            required_tools,
            admitted_at: now(),
        };
        if queued.is_some() {
            loop {
                let revision = self
                    .store
                    .message_queue_revision(&id)
                    .await
                    .map_err(|_| "agent_queue_unavailable")?;
                match self
                    .store
                    .admit_queued_message(&self.revision.activation, revision, pending.clone())
                    .await
                {
                    Ok(receipt) => {
                        return Ok(SendResult::Queued {
                            session_id: id,
                            message_id: delivered.message.message_id,
                            revision: receipt.queue_revision,
                        });
                    }
                    Err(maka_event_log::StoreError::RevisionConflict { .. }) => continue,
                    Err(_) => return Err("agent_admission_failed".into()),
                }
            }
        }
        let input = self
            .input(invocation.clone(), backend, basis.clone(), delivered)
            .await?;
        self.store
            .admit_message(pending)
            .await
            .map_err(|_| "agent_admission_failed")?;
        let cancellation = self.stop.child_token();
        let running = match self.start(input, cancellation.clone()).await {
            Ok(running) => running,
            Err(error) => {
                if error.is_cleanup_unconfirmed() {
                    admission.state().await.quarantined.insert(id.clone());
                }
                return Err(public_error(error));
            }
        };
        self.publish(&mut *admission.state().await, &running, basis);
        let owner = self.clone();
        self.workers
            .spawn(async move { owner.drive(running).await });
        Ok(crate::contract::SendResult::Started {
            session_id: invocation.session_id.clone(),
            invocation_id: invocation.invocation_id,
            message_id: format!("m_{identity}"),
        })
    }
}
fn metadata_rejection(title: &str, labels: &[String]) -> Option<Rejection> {
    if title.trim().is_empty() || title.chars().any(char::is_control) {
        Some(Rejection::TitleInvalid)
    } else if labels
        .iter()
        .any(|label| label.trim().is_empty() || label.chars().any(char::is_control))
    {
        Some(Rejection::LabelsInvalid)
    } else {
        None
    }
}

impl Owner {
    pub(crate) fn prepare_retirement(
        &self,
    ) -> Option<Box<dyn maka_host_runtime::application::NativeRetirement>> {
        let producer = self.producer_gate.clone().try_lock_owned().ok()?;
        let runs = self.runs.clone().try_lock_owned().ok()?;
        if !runs.controls.is_empty()
            || !runs.quarantined.is_empty()
            || !self.workers.is_empty()
            || !self.auxiliary.resources.is_idle()
            || self.goals.keeps_host_awake()
            || self.plans.keeps_host_awake()
            || self
                .graphs
                .as_ref()
                .is_some_and(|graphs| graphs.keeps_host_awake())
        {
            return None;
        }
        Some(Box::new(Idle {
            runs,
            _producer: producer,
            stop: self.stop.clone(),
        }))
    }
    pub fn close(self: &Arc<Self>) -> BoxFuture<'static, Result<(), String>> {
        let owner = self.clone();
        Box::pin(async move {
            {
                let mut runs = owner.runs.lock().await;
                runs.closing = true;
                owner.stop.cancel();
            }
            owner.producers.close();
            owner.producers.wait().await;
            owner.workers.close();
            owner.workers.wait().await;
            owner.engine.drain().await;
            let scheduler = owner.scheduler.finish().await;
            let todo = owner.todo.finish().await;
            let goals = owner.goals.finish().await;
            let plans = owner.plans.finish().await;
            let graph = match &owner.graphs {
                Some(graphs) => graphs.finish().await,
                None => Ok(()),
            };
            let recap = owner.recap_storage.finish().await;
            let auxiliary = owner.auxiliary.resources.finish().await;
            let configuration = owner.configuration.shutdown().await;
            if let Err(error) = &configuration {
                tracing::error!(stage = "configuration_store_shutdown", cause = crate::diagnostics::config_cause(error),
                    activation = %owner.revision.activation, instance = ?owner.data_source.namespace,
                    "Agent configuration store cleanup failed");
            }
            let runs = owner.runs.lock().await;
            if !runs.quarantined.is_empty() {
                return Err("agent_cleanup_unconfirmed".into());
            }
            scheduler?;
            todo?;
            goals?;
            plans?;
            graph?;
            recap?;
            auxiliary.map_err(|error| error.to_string())?;
            configuration.map_err(|_| "agent_configuration_cleanup_unconfirmed".into())
        })
    }
}
struct Idle {
    runs: runs::Exclusive,
    _producer: tokio::sync::OwnedMutexGuard<()>,
    stop: CancellationToken,
}
impl maka_host_runtime::application::NativeRetirement for Idle {
    fn commit(mut self: Box<Self>) {
        self.runs.closing = true;
        self.stop.cancel();
    }
}
use maka_agent_authority::now;
fn jsonschema_valid(input: &Value) -> bool {
    jsonschema::validator_for(&configuration_schema())
        .expect("configuration schema")
        .is_valid(input)
}

fn public_error(error: RunError) -> String {
    match error {
        RunError::Busy => "agent_session_busy",
        RunError::Cancelled => "agent_cancelled",
        RunError::CleanupUnconfirmed => "agent_cleanup_unconfirmed",
        _ => "agent_execution_unavailable",
    }
    .into()
}

#[cfg(test)]
#[path = "owner/tests.rs"]
mod tests;

pub(crate) fn owns_session(prefix: &str, id: &str) -> bool {
    id.strip_prefix(prefix).is_some_and(|tail| {
        tail.len() == 32
            && tail
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    })
}