use std::collections::BTreeMap;
use std::path::PathBuf;
use runx_contracts::{ClosureDisposition, JsonValue, Receipt};
use thiserror::Error;
use super::harness::{HarnessReplayError, HarnessReplayOutput};
#[cfg(feature = "cli-tool")]
use super::runner::GraphRun;
use super::skill_front::{InlineHarnessReport, SkillRunError};
use crate::effects::RuntimeEffectRegistry;
#[derive(Clone, Debug, PartialEq)]
pub struct SkillRunRequest {
pub skill_path: PathBuf,
pub receipt_dir: Option<PathBuf>,
pub run_id: Option<String>,
pub answers_path: Option<PathBuf>,
pub inputs: BTreeMap<String, JsonValue>,
pub env: BTreeMap<String, String>,
pub cwd: PathBuf,
pub local_credential: Option<LocalCredentialDescriptor>,
}
#[derive(Clone, PartialEq, Eq)]
pub struct LocalCredentialDescriptor {
pub provider: String,
pub auth_mode: String,
pub env_var: String,
pub material_ref: String,
pub scopes: Vec<String>,
pub secret: String,
}
impl std::fmt::Debug for LocalCredentialDescriptor {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("LocalCredentialDescriptor")
.field("provider", &self.provider)
.field("auth_mode", &self.auth_mode)
.field("env_var", &self.env_var)
.field("material_ref", &self.material_ref)
.field("scopes", &self.scopes)
.field("secret", &"[redacted]")
.finish()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GraphRunRequest {
pub graph_path: PathBuf,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct HarnessRunRequest {
pub fixture_path: PathBuf,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InlineHarnessRequest {
pub skill_path: PathBuf,
pub receipt_dir: Option<PathBuf>,
pub env: Option<BTreeMap<String, String>>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RunContinuation {
pub run_id: Option<String>,
pub answers_path: Option<PathBuf>,
}
#[derive(Clone, Debug, PartialEq)]
pub enum RunRequest {
Skill(Box<SkillRunRequest>),
Graph(GraphRunRequest),
Harness(HarnessRunRequest),
}
#[derive(Clone, Debug, PartialEq)]
pub struct RunResult {
pub status: RunStatus,
pub output: JsonValue,
pub receipt_refs: Vec<String>,
pub child_receipt_refs: Vec<String>,
pub pending_requests: Vec<JsonValue>,
pub diagnostics: Vec<String>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RunStatus {
NeedsAgent,
Sealed,
Succeeded,
Failed,
}
#[derive(Debug, Error)]
pub enum OrchestratorError {
#[error(transparent)]
SkillRun(#[from] SkillRunError),
#[error(transparent)]
Runtime(#[from] crate::RuntimeError),
#[error(transparent)]
Harness(#[from] HarnessReplayError),
#[error(
"native graph orchestration is unavailable because runx-runtime was built without the cli-tool feature"
)]
CliToolFeatureDisabled,
}
#[derive(Clone, Debug, Default)]
pub struct LocalOrchestrator {
effects: RuntimeEffectRegistry,
}
impl LocalOrchestrator {
#[must_use]
pub fn with_effects(effects: RuntimeEffectRegistry) -> Self {
Self { effects }
}
pub fn run(&self, request: RunRequest) -> Result<RunResult, OrchestratorError> {
match request {
RunRequest::Skill(request) => self.run_skill(&request),
RunRequest::Graph(request) => self.run_graph(&request),
RunRequest::Harness(request) => self.run_harness(&request),
}
}
pub fn run_skill(&self, request: &SkillRunRequest) -> Result<RunResult, OrchestratorError> {
let output = super::skill_front::execute_skill_run_with_effects(request, &self.effects)?;
Ok(skill_result(output))
}
pub fn run_skill_with_runner(
&self,
request: &SkillRunRequest,
runner: &str,
) -> Result<RunResult, OrchestratorError> {
let overrides = super::skill_front::SkillRunOverrides {
runner: Some(runner.to_owned()),
seeded_answers: None,
};
let output = super::skill_front::execute_skill_run_with_overrides(
request,
&overrides,
&self.effects,
)?;
Ok(skill_result(output))
}
pub fn run_graph(&self, request: &GraphRunRequest) -> Result<RunResult, OrchestratorError> {
#[cfg(feature = "cli-tool")]
{
let mut options = super::runner::RuntimeOptions::from_process_env()?;
options.effects = self.effects.clone();
let runtime =
super::runner::Runtime::new(crate::adapters::cli_tool::CliToolAdapter, options);
graph_result(runtime.run_graph_file(&request.graph_path)?)
}
#[cfg(not(feature = "cli-tool"))]
{
let _ = request;
Err(OrchestratorError::CliToolFeatureDisabled)
}
}
pub fn run_harness(&self, request: &HarnessRunRequest) -> Result<RunResult, OrchestratorError> {
harness_result(self.run_harness_fixture(request)?)
}
pub fn run_inline_harness(
&self,
request: &InlineHarnessRequest,
) -> Result<InlineHarnessReport, OrchestratorError> {
Ok(super::skill_front::run_inline_harness_with_effects(
&request.skill_path,
request.receipt_dir.as_deref(),
request.env.as_ref(),
&self.effects,
)?)
}
#[cfg(feature = "cli-tool")]
pub fn run_harness_fixture(
&self,
request: &HarnessRunRequest,
) -> Result<HarnessReplayOutput, OrchestratorError> {
let mut options = super::runner::RuntimeOptions::from_process_env()?;
options.created_at = crate::time::DEFAULT_CREATED_AT.to_owned();
options.effects = self.effects.clone();
Ok(super::harness::run_harness_fixture_with_adapter(
&request.fixture_path,
super::skill_front::SkillRunGraphAdapter::default(),
options,
)?)
}
#[cfg(not(feature = "cli-tool"))]
pub fn run_harness_fixture(
&self,
request: &HarnessRunRequest,
) -> Result<HarnessReplayOutput, OrchestratorError> {
let _ = self;
let _ = request;
Err(OrchestratorError::CliToolFeatureDisabled)
}
}
fn skill_result(output: JsonValue) -> RunResult {
let status = match object_string(&output, "status") {
Some("needs_agent") => RunStatus::NeedsAgent,
Some("sealed") => RunStatus::Sealed,
_ => RunStatus::Succeeded,
};
let receipt_refs = object_string(&output, "receipt_id")
.map(|receipt_id| vec![receipt_id.to_owned()])
.unwrap_or_default();
let pending_requests = object_array(&output, "requests")
.map(|requests| requests.to_vec())
.unwrap_or_default();
RunResult {
status,
output,
receipt_refs,
child_receipt_refs: Vec::new(),
pending_requests,
diagnostics: Vec::new(),
}
}
#[cfg(feature = "cli-tool")]
fn graph_result(run: GraphRun) -> Result<RunResult, OrchestratorError> {
let status = status_from_receipt(&run.receipt);
let output = receipt_json(&run.receipt)?;
Ok(RunResult {
status,
output,
receipt_refs: vec![run.receipt.id.to_string()],
child_receipt_refs: child_receipt_refs(&run.receipt),
pending_requests: Vec::new(),
diagnostics: Vec::new(),
})
}
fn harness_result(output: HarnessReplayOutput) -> Result<RunResult, OrchestratorError> {
let status = status_from_receipt(&output.receipt);
let value = receipt_json(&output.receipt)?;
Ok(RunResult {
status,
output: value,
receipt_refs: vec![output.receipt.id.to_string()],
child_receipt_refs: child_receipt_refs(&output.receipt),
pending_requests: Vec::new(),
diagnostics: Vec::new(),
})
}
fn status_from_receipt(receipt: &Receipt) -> RunStatus {
match receipt.seal.disposition {
ClosureDisposition::Closed => RunStatus::Sealed,
_ => RunStatus::Failed,
}
}
fn receipt_json(receipt: &Receipt) -> Result<JsonValue, OrchestratorError> {
let value = serde_json::to_value(receipt)
.map_err(|source| crate::RuntimeError::json("serializing orchestrated receipt", source))?;
serde_json::from_value(value)
.map_err(|source| crate::RuntimeError::json("normalizing orchestrated receipt", source))
.map_err(Into::into)
}
fn child_receipt_refs(receipt: &Receipt) -> Vec<String> {
receipt
.lineage
.as_ref()
.map(|lineage| {
lineage
.children
.iter()
.map(|reference| reference.uri.clone().into_string())
.collect()
})
.unwrap_or_default()
}
fn object_string<'a>(value: &'a JsonValue, key: &str) -> Option<&'a str> {
let JsonValue::Object(object) = value else {
return None;
};
let JsonValue::String(value) = object.get(key)? else {
return None;
};
Some(value)
}
fn object_array<'a>(value: &'a JsonValue, key: &str) -> Option<&'a Vec<JsonValue>> {
let JsonValue::Object(object) = value else {
return None;
};
let JsonValue::Array(value) = object.get(key)? else {
return None;
};
Some(value)
}