rho-coding-agent 2.25.0

//! Shared workflow plan/run policy used by the CLI and the model tool.

use std::{
    collections::BTreeMap,
    path::{Path, PathBuf},
};

use crate::{
    app::workflow_runtime::RecoveryDecision,
    workflow::{
        normalize_workflow, validate_runtime_budgets, validate_workflow, verify_directory_identity,
        verify_executable_identity, CollectedSources, Digest, FrozenSchedulerSettings,
        FrozenWorkflow, InputName, PlanConsent, PlanId, PlanInventoryItem, PlannerIdentity,
        PlanningLimits, ResolvedNode, RunId, RunInventoryItem, RunLifecycle, SourceCollector,
        StoredPlan, StoredRun, WorkflowService, WorkflowStore, WorkflowValue,
        FROZEN_WORKFLOW_SCHEMA_VERSION,
    },
};

use super::{
    cancel::CancellationOutcome,
    plan_host::{resolve_nodes_with_host, DiscoveringPlanHost, PlanHost},
    planner_worker::{self, run_supervised_planner, PlannerWorkerPlan},
    STARLARK_VERSION,
};

// Receipt: freeze-time workflow kind ceilings. Independent of
// behavior.agent_concurrency, which is the live delegated-agent slot cap.
const DEFAULT_PARALLEL_NODES: u32 = 4;
const DEFAULT_PARALLEL_AGENTS: u32 = 4;
const DEFAULT_PARALLEL_COMMANDS: u32 = 4;
const PLANNER_FORMAT_VERSION: u32 = 1;

pub(crate) struct PreparedPlan {
    pub(crate) sources: CollectedSources,
    pub(crate) workflow: FrozenWorkflow,
}

pub(crate) use crate::workflow::RunRecord;

/// Owns validate | plan | run | status | cancel | resume policy for both adapters.
pub(crate) struct WorkflowOps {
    service: WorkflowService,
    workspace: PathBuf,
    config_path: Option<PathBuf>,
}

impl WorkflowOps {
    pub(crate) fn open(workspace: PathBuf, config_path: Option<PathBuf>) -> anyhow::Result<Self> {
        let workspace = workspace.canonicalize()?;
        Ok(Self {
            service: WorkflowService::new(WorkflowStore::new(&crate::paths::rho_dir()?)?),
            workspace,
            config_path,
        })
    }

    pub(crate) async fn collect_local_sources(
        &self,
        entry: &Path,
        limits: &PlanningLimits,
    ) -> anyhow::Result<CollectedSources> {
        Ok(SourceCollector::new(&self.workspace, limits)?
            .collect(entry)
            .await?)
    }

    pub(crate) async fn prepare_from_sources(
        &self,
        sources: CollectedSources,
        inputs: BTreeMap<InputName, WorkflowValue>,
        host: &dyn PlanHost,
        limits: &PlanningLimits,
    ) -> anyhow::Result<PreparedPlan> {
        let planned = run_supervised_planner(&sources, inputs, limits).await?;
        let resolved_nodes = resolve_nodes_with_host(&planned.program, host)?;
        freeze_planned_workflow(sources, planned, resolved_nodes, limits)
    }

    pub(crate) async fn prepare_local(
        &self,
        entry: &Path,
        inputs: BTreeMap<InputName, WorkflowValue>,
        config: &crate::config::Config,
        available_tools: &crate::agent::AgentCapabilities,
        limits: &PlanningLimits,
    ) -> anyhow::Result<PreparedPlan> {
        let sources = self.collect_local_sources(entry, limits).await?;
        let host = DiscoveringPlanHost::new(&self.workspace, config, available_tools, entry)?;
        self.prepare_from_sources(sources, inputs, &host, limits)
            .await
    }

    /// [`Self::prepare_local`] with JSON input values, as `/workflow:<name>`
    /// and the hub supply them, under the default planning limits.
    pub(crate) async fn prepare_local_json(
        &self,
        entry: &Path,
        inputs: BTreeMap<String, serde_json::Value>,
        config: &crate::config::Config,
        available_tools: &crate::agent::AgentCapabilities,
    ) -> anyhow::Result<PreparedPlan> {
        let inputs = inputs
            .into_iter()
            .map(|(name, value)| Ok((InputName::new(name)?, WorkflowValue::from_json(value)?)))
            .collect::<anyhow::Result<BTreeMap<_, _>>>()?;
        let limits = super::planning_limits()?;
        self.prepare_local(entry, inputs, config, available_tools, &limits)
            .await
    }

    /// Single freeze pipeline output stored once. No second normalize/validate build.
    pub(crate) fn store_plan(&self, prepared: &PreparedPlan) -> anyhow::Result<StoredPlan> {
        Ok(self.service.store_frozen(
            &prepared.workflow,
            workspace_identity(&self.workspace)?,
            &prepared.sources.sources,
        )?)
    }

    pub(crate) fn load_plan_prefix(&self, prefix: &str) -> anyhow::Result<StoredPlan> {
        let plan_id = self.service.store().resolve_plan(prefix)?;
        Ok(self.service.store().load_plan(plan_id)?)
    }

    pub(crate) fn load_plan_id(&self, plan_id: PlanId) -> anyhow::Result<StoredPlan> {
        Ok(self.service.store().load_plan(plan_id)?)
    }

    pub(crate) fn load_run_prefix(&self, prefix: &str) -> anyhow::Result<StoredRun> {
        let run_id = self.service.store().resolve_run(prefix)?;
        Ok(self.service.store().load_run(run_id)?)
    }

    /// Loads a run for display, including read-only runs from older releases.
    pub(crate) fn load_run_record(&self, prefix: &str) -> anyhow::Result<RunRecord> {
        let store = self.service.store();
        let run_id = store.resolve_run(prefix)?;
        self.load_run_record_id(run_id)
    }

    pub(crate) fn load_run_record_id(&self, run_id: RunId) -> anyhow::Result<RunRecord> {
        Ok(self.service.store().load_run_record(run_id)?)
    }

    pub(crate) fn load_run_id(&self, run_id: RunId) -> anyhow::Result<StoredRun> {
        Ok(self.service.store().load_run(run_id)?)
    }

    pub(crate) fn read_run_inventory(&self, run_id: RunId) -> anyhow::Result<RunInventoryItem> {
        Ok(self.service.store().read_run_inventory(run_id)?)
    }

    pub(crate) fn list_workspace_plans(&self) -> anyhow::Result<Vec<PlanInventoryItem>> {
        let identity = workspace_identity(&self.workspace)?;
        Ok(self
            .service
            .store()
            .list_plan_inventory()?
            .into_iter()
            .filter(|plan| plan.workspace_identity == identity)
            .collect())
    }

    pub(crate) fn list_workspace_runs(&self) -> anyhow::Result<Vec<RunInventoryItem>> {
        let identity = workspace_identity(&self.workspace)?;
        Ok(self
            .service
            .store()
            .list_run_inventory()?
            .into_iter()
            .filter(|run| run.workspace_identity == identity)
            .collect())
    }

    pub(crate) fn delete_workspace_plan(&self, plan_id: PlanId) -> anyhow::Result<()> {
        let manifest = self.service.store().read_plan_inventory(plan_id)?;
        let identity = workspace_identity(&self.workspace)?;
        if manifest.workspace_identity != identity {
            anyhow::bail!("plan belongs to another workspace");
        }
        Ok(self.service.store().delete_plan(plan_id)?)
    }

    pub(crate) fn delete_workspace_run(&self, run_id: RunId) -> anyhow::Result<()> {
        let run = self.service.store().read_run_inventory(run_id)?;
        let identity = workspace_identity(&self.workspace)?;
        if run.workspace_identity != identity {
            anyhow::bail!("run belongs to another workspace");
        }
        Ok(self.service.store().delete_run(run_id)?)
    }

    pub(crate) fn recheck_plan(&self, plan: &StoredPlan) -> anyhow::Result<()> {
        let current_workspace = workspace_identity(&self.workspace)?;
        if current_workspace != plan.manifest.workspace_identity {
            anyhow::bail!(
                "workflow plan workspace is '{}', but the current workspace is '{}'",
                plan.manifest.workspace_identity,
                current_workspace
            );
        }
        recheck_frozen_graph(&plan.graph, self.config_path.clone())
    }

    pub(crate) fn recheck_run(&self, run: &StoredRun) -> anyhow::Result<()> {
        let current_workspace = workspace_identity(&self.workspace)?;
        if current_workspace != run.manifest.workspace_identity {
            anyhow::bail!(
                "workflow run workspace is '{}', but the current workspace is '{}'",
                run.manifest.workspace_identity,
                current_workspace
            );
        }
        if crate::workflow::program_digest(&run.graph)? != run.manifest.program_digest {
            anyhow::bail!("workflow run digest does not match its copied frozen graph");
        }
        if !run.manifest.consent.confirmed
            || run.manifest.consent.program_digest != run.manifest.program_digest
        {
            anyhow::bail!("workflow run consent does not match its copied frozen graph");
        }
        recheck_frozen_graph(&run.graph, self.config_path.clone())
    }

    pub(crate) fn create_confirmed_run(&self, plan: &StoredPlan) -> anyhow::Result<StoredRun> {
        Ok(self.service.create_run(
            plan,
            PlanConsent {
                program_digest: plan.manifest.program_digest.clone(),
                confirmed: true,
            },
        )?)
    }

    /// Freezes `prepared` and creates a confirmed run from it, for starts
    /// that go straight from source to a run (hub Start, `/workflow:<name>`,
    /// the tool's `start`). The launch-only plan is removed once the run owns
    /// its frozen graph; failing to remove it is logged, not fatal, because
    /// the run is already valid. A failed run creation keeps the plan.
    pub(crate) fn create_launch_run(&self, prepared: &PreparedPlan) -> anyhow::Result<StoredRun> {
        use anyhow::Context as _;

        let plan = self.store_plan(prepared).context("could not save plan")?;
        let run = self
            .recheck_plan(&plan)
            .and_then(|()| self.create_confirmed_run(&plan))
            .context("could not create run")?;
        if let Err(error) = self.delete_workspace_plan(plan.manifest.plan_id) {
            tracing::warn!(
                plan_id = %plan.manifest.plan_id,
                error = %error,
                "could not remove launch-only workflow plan"
            );
        }
        Ok(run)
    }

    pub(crate) fn prepare_run(&self, plan_prefix_or_id: &str) -> anyhow::Result<StoredPlan> {
        let plan = self.load_plan_prefix(plan_prefix_or_id)?;
        self.recheck_plan(&plan)?;
        Ok(plan)
    }

    pub(crate) fn prepare_run_id(&self, plan_id: PlanId) -> anyhow::Result<StoredPlan> {
        let plan = self.load_plan_id(plan_id)?;
        self.recheck_plan(&plan)?;
        Ok(plan)
    }

    pub(crate) fn prepare_resume(
        &self,
        run: &StoredRun,
        recover_uncertain: bool,
    ) -> anyhow::Result<RecoveryDecision> {
        self.recheck_run(run)?;
        if run.state.state.lifecycle == RunLifecycle::NeedsRecovery && !recover_uncertain {
            anyhow::bail!(
                "workflow run {} has uncertain attempts; confirm that no prior process remains, then use --recover-uncertain",
                run.manifest.run_id
            );
        }
        Ok(if recover_uncertain {
            RecoveryDecision::ConfirmNoProcess
        } else {
            RecoveryDecision::NormalResume
        })
    }

    pub(crate) async fn cancel(
        &self,
        run_id: RunId,
        lifecycle: RunLifecycle,
    ) -> anyhow::Result<CancellationOutcome> {
        let rho_home = crate::paths::rho_dir()?;
        Ok(super::request_cancellation(&rho_home, run_id, lifecycle).await?)
    }
}

pub(crate) fn freeze_planned_workflow(
    sources: CollectedSources,
    planned: PlannerWorkerPlan,
    resolved_nodes: BTreeMap<crate::workflow::NodeId, ResolvedNode>,
    limits: &PlanningLimits,
) -> anyhow::Result<PreparedPlan> {
    // Authoritative freeze: normalize + validate exactly once before store/validate display.
    let workflow = normalize_workflow(FrozenWorkflow {
        schema_version: FROZEN_WORKFLOW_SCHEMA_VERSION,
        planner: PlannerIdentity {
            name: "rho".to_owned(),
            format_version: PLANNER_FORMAT_VERSION,
            starlark_version: STARLARK_VERSION.to_owned(),
        },
        program_digest: Digest(String::new()),
        sources: sources.manifest.clone(),
        inputs: planned.inputs,
        program: planned.program,
        resolved_nodes,
        scheduler: FrozenSchedulerSettings {
            max_parallel_nodes: DEFAULT_PARALLEL_NODES,
            max_parallel_agents: DEFAULT_PARALLEL_AGENTS,
            max_parallel_commands: DEFAULT_PARALLEL_COMMANDS,
        },
        runtime_limits: limits.frozen_runtime_limits(),
    })?;
    validate_workflow(&workflow)?;
    validate_runtime_budgets(&workflow, limits)?;
    Ok(PreparedPlan { sources, workflow })
}

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

pub(crate) fn workspace_identity(path: &Path) -> anyhow::Result<String> {
    Ok(crate::paths::display(&path.canonicalize()?))
}

fn recheck_frozen_graph(
    graph: &FrozenWorkflow,
    config_path: Option<PathBuf>,
) -> anyhow::Result<()> {
    validate_workflow(graph)?;
    validate_runtime_budgets(graph, &planner_worker::planning_limits()?)?;
    let config = crate::app::config_repository::ConfigRepository::new(config_path).load()?;
    for resolved in graph.resolved_nodes.values() {
        let agent = match resolved {
            ResolvedNode::Command(command) => {
                verify_executable_identity(&command.executable_identity)?;
                verify_directory_identity(&command.cwd_identity)?;
                continue;
            }
            ResolvedNode::Agent(agent) => agent,
        };
        match (&agent.executable, &agent.executable_identity) {
            (Some(_), Some(identity)) => {
                verify_executable_identity(identity)?;
            }
            (Some(path), None) => anyhow::bail!(
                "frozen agent '{}' records executable '{}' without a frozen identity",
                agent.agent_id,
                path
            ),
            (None, _) => {}
        }
        if agent.trust_required && !crate::workspace::ProjectTrust::from_agents_env().is_trusted() {
            anyhow::bail!(
                "workflow plan requires trusted project agent '{}'; trust is not active",
                agent.agent_id
            );
        }
        if agent.plugin_trust_required
            && !crate::workspace::ProjectTrust::from_plugins_env().is_trusted()
        {
            anyhow::bail!(
                "workflow plan requires trusted project plugin agent '{}'; plugin trust is not active",
                agent.agent_id
            );
        }
        let current_rank = permission_rank(config.permission_mode.as_str()).ok_or_else(|| {
            anyhow::anyhow!(
                "current permission mode '{}' is unsupported",
                config.permission_mode
            )
        })?;
        let ceiling_rank = permission_rank(&agent.permission_ceiling).ok_or_else(|| {
            anyhow::anyhow!(
                "frozen permission ceiling '{}' for agent '{}' is unsupported",
                agent.permission_ceiling,
                agent.agent_id
            )
        })?;
        if current_rank > ceiling_rank {
            anyhow::bail!(
                "current permission mode '{}' would widen frozen authority '{}' for agent '{}'",
                config.permission_mode,
                agent.permission_ceiling,
                agent.agent_id
            );
        }
    }
    Ok(())
}

fn permission_rank(value: &str) -> Option<u8> {
    value
        .parse::<crate::permission::PermissionMode>()
        .ok()
        .map(crate::permission::PermissionMode::restrictiveness_rank)
}