use super::errors::OrchestratorError;
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration as StdDuration;
use harn_vm::event_log::{AnyEventLog, EventLog, LogEvent, Topic};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use serde_json::json;
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;
use crate::cli::OrchestratorLocalArgs;
use crate::package::{self, CollectedManifestTrigger};
pub(crate) use super::harness::{absolutize_from_cwd, load_manifest};
use super::role::OrchestratorRole;
use super::supervisor_state::apply_supervisor_state;
pub(crate) const STATE_SNAPSHOT_FILE: &str = "orchestrator-state.json";
pub(crate) use harn_vm::{
TRIGGERS_LIFECYCLE_TOPIC, TRIGGER_ATTEMPTS_TOPIC, TRIGGER_DLQ_TOPIC,
TRIGGER_INBOX_CLAIMS_TOPIC, TRIGGER_INBOX_ENVELOPES_TOPIC, TRIGGER_INBOX_LEGACY_TOPIC,
TRIGGER_INBOX_OBSERVABILITY_TOPIC, TRIGGER_OUTBOX_TOPIC,
};
pub(crate) struct LoadedOrchestratorContext {
pub vm: harn_vm::Vm,
pub event_log: Arc<AnyEventLog>,
pub collected_triggers: Vec<CollectedManifestTrigger>,
pub bindings: Vec<harn_vm::TriggerBindingSpec>,
pub snapshot: Option<PersistedStateSnapshot>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct PersistedStateSnapshot {
pub status: String,
pub bind: String,
#[serde(default)]
pub triggers: Vec<PersistedTriggerStateSnapshot>,
#[serde(default)]
pub connectors: Vec<String>,
#[serde(default)]
pub activations: Vec<ConnectorActivationSnapshot>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub(crate) struct PersistedTriggerStateSnapshot {
pub id: String,
pub provider: String,
pub kind: String,
pub handler: String,
pub version: Option<u32>,
pub state: Option<String>,
#[serde(default)]
pub received: u64,
#[serde(default)]
pub dispatched: u64,
#[serde(default)]
pub failed: u64,
#[serde(default)]
pub in_flight: u64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct ConnectorActivationSnapshot {
pub provider: String,
pub binding_count: usize,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DispatchHandleRecord {
pub event_id: String,
pub binding_id: String,
pub binding_version: u32,
pub status: String,
pub replay_of_event_id: Option<String>,
pub dlq_entry_id: Option<String>,
pub error: Option<String>,
pub result: Option<serde_json::Value>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DlqAttemptRecord {
pub attempt: u32,
pub at: String,
pub status: String,
pub error: Option<String>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct DlqEntryRecord {
pub id: String,
pub event_id: String,
pub binding_id: String,
pub binding_version: u32,
pub provider: String,
pub kind: String,
pub state: String,
pub error: String,
#[serde(default = "default_dlq_error_class")]
pub error_class: String,
pub event: harn_vm::TriggerEvent,
pub retry_history: Vec<DlqAttemptRecord>,
}
fn default_dlq_error_class() -> String {
"unknown".to_string()
}
#[derive(Clone, Debug, Serialize)]
pub(crate) struct StrandedEnvelopeRecord {
pub inbox_offset: u64,
pub event_id: String,
pub trigger_id: Option<String>,
pub binding_version: Option<u32>,
pub provider: String,
pub kind: String,
pub received_at: OffsetDateTime,
pub age: StdDuration,
}
pub(crate) async fn load_local_runtime(
args: &OrchestratorLocalArgs,
) -> Result<LoadedOrchestratorContext, OrchestratorError> {
harn_vm::reset_thread_local_state();
let config_path = absolutize_from_cwd(&args.config)?;
let (_manifest, manifest_dir) = load_manifest(&config_path)?;
let state_dir = absolutize_from_cwd(&args.state_dir)?;
std::fs::create_dir_all(&state_dir).map_err(|error| {
format!(
"failed to create state dir {}: {error}",
state_dir.display()
)
})?;
let mut vm =
OrchestratorRole::SingleTenant.build_vm(&manifest_dir, &manifest_dir, &state_dir)?;
let extensions = package::load_runtime_extensions(&config_path);
package::install_manifest_handoff_routes(&extensions);
package::install_orchestrator_budget(&extensions);
let collected_triggers = package::collect_manifest_triggers(&mut vm, &extensions)
.await
.map_err(|error| format!("failed to collect manifest triggers: {error}"))?;
let bindings =
package::install_collected_manifest_triggers(&collected_triggers, &extensions).await?;
apply_supervisor_state(&state_dir).await?;
let event_log = harn_vm::event_log::active_event_log()
.ok_or_else(|| "event log was not installed during VM initialization".to_string())?;
let snapshot = read_state_snapshot(&state_dir.join(STATE_SNAPSHOT_FILE))?;
Ok(LoadedOrchestratorContext {
vm,
event_log,
collected_triggers,
bindings,
snapshot,
})
}
pub(crate) async fn trigger_list(
_ctx: &mut LoadedOrchestratorContext,
) -> Result<Vec<harn_vm::TriggerBindingSnapshot>, OrchestratorError> {
Ok(harn_vm::current_trigger_binding_snapshots())
}
pub(crate) async fn trigger_replay(
ctx: &mut LoadedOrchestratorContext,
event_id: &str,
) -> Result<DispatchHandleRecord, OrchestratorError> {
let event_id = serde_json::to_string(event_id).map_err(|error| error.to_string())?;
eval_json(&mut ctx.vm, &format!("trigger_replay({event_id})")).await
}
pub(crate) async fn trigger_inspect_dlq(
ctx: &mut LoadedOrchestratorContext,
) -> Result<Vec<DlqEntryRecord>, OrchestratorError> {
eval_json(&mut ctx.vm, "trigger_inspect_dlq()").await
}
pub(crate) async fn trigger_fire(
ctx: &mut LoadedOrchestratorContext,
binding_id: &str,
event: serde_json::Value,
) -> Result<DispatchHandleRecord, OrchestratorError> {
let binding_id = serde_json::to_string(binding_id).map_err(|error| error.to_string())?;
let event_json = serde_json::to_string(&event).map_err(|error| error.to_string())?;
let event_literal = serde_json::to_string(&event_json).map_err(|error| error.to_string())?;
eval_json(
&mut ctx.vm,
&format!("trigger_fire({binding_id}, json_parse({event_literal}))"),
)
.await
}
pub(crate) fn synthetic_event_for_binding(
ctx: &LoadedOrchestratorContext,
binding_id: &str,
) -> Result<serde_json::Value, OrchestratorError> {
let trigger = ctx
.bindings
.iter()
.find(|trigger| trigger.id == binding_id)
.ok_or_else(|| format!("unknown manifest binding '{binding_id}'"))?;
let kind =
trigger
.match_events
.first()
.cloned()
.unwrap_or_else(|| match trigger.kind.as_str() {
"cron" => "cron.tick".to_string(),
"a2a-push" => "a2a.task.received".to_string(),
kind => kind.to_string(),
});
Ok(json!({
"provider": trigger.provider.as_str(),
"kind": kind,
"headers": {
"x-harn-binding-id": binding_id,
"x-harn-synthetic": "true",
}
}))
}
pub(crate) async fn read_topic(
log: &Arc<AnyEventLog>,
topic_name: &str,
) -> Result<Vec<(u64, LogEvent)>, OrchestratorError> {
let topic = Topic::new(topic_name).map_err(|error| error.to_string())?;
log.read_range(&topic, None, usize::MAX)
.await
.map_err(|error| OrchestratorError::Common(error.to_string()))
}
pub(crate) async fn stranded_envelopes(
log: &Arc<AnyEventLog>,
min_age: StdDuration,
) -> Result<Vec<StrandedEnvelopeRecord>, OrchestratorError> {
let envelopes = read_topic(log, TRIGGER_INBOX_ENVELOPES_TOPIC).await?;
let legacy_inbox = read_topic(log, TRIGGER_INBOX_LEGACY_TOPIC).await?;
let outbox = read_topic(log, TRIGGER_OUTBOX_TOPIC).await?;
let mut any_outbox_event_ids = BTreeSet::new();
let mut outbox_by_event_id = BTreeMap::<String, BTreeSet<String>>::new();
for (_, event) in outbox {
let Some(event_id) = event.headers.get("event_id").cloned() else {
continue;
};
any_outbox_event_ids.insert(event_id.clone());
if let Some(trigger_id) = event.headers.get("trigger_id").cloned() {
outbox_by_event_id
.entry(event_id)
.or_default()
.insert(trigger_id);
}
}
let mut stranded = Vec::new();
for (offset, event) in envelopes.into_iter().chain(legacy_inbox) {
if event.kind != "event_ingested" {
continue;
}
let envelope: harn_vm::triggers::dispatcher::InboxEnvelope =
serde_json::from_value(event.payload)
.map_err(|error| format!("failed to decode trigger inbox envelope: {error}"))?;
let age = age_since(envelope.event.received_at);
if age < min_age {
continue;
}
let event_id = envelope.event.id.0.clone();
let matched_outbox = if let Some(trigger_id) = envelope.trigger_id.as_ref() {
outbox_by_event_id
.get(&event_id)
.is_some_and(|trigger_ids| trigger_ids.contains(trigger_id))
} else {
any_outbox_event_ids.contains(&event_id)
};
if matched_outbox {
continue;
}
stranded.push(StrandedEnvelopeRecord {
inbox_offset: offset,
event_id,
trigger_id: envelope.trigger_id,
binding_version: envelope.binding_version,
provider: envelope.event.provider.as_str().to_string(),
kind: envelope.event.kind,
received_at: envelope.event.received_at,
age,
});
}
stranded.sort_by(|left, right| {
left.received_at
.cmp(&right.received_at)
.then(left.event_id.cmp(&right.event_id))
});
Ok(stranded)
}
pub(crate) async fn append_dlq_entry(
log: &Arc<AnyEventLog>,
entry: &DlqEntryRecord,
) -> Result<(), OrchestratorError> {
let topic = Topic::new(TRIGGER_DLQ_TOPIC).map_err(|error| error.to_string())?;
let payload = serde_json::to_value(entry).map_err(|error| error.to_string())?;
log.append(&topic, LogEvent::new("dlq_entry", payload))
.await
.map(|_| ())
.map_err(|error| OrchestratorError::Common(error.to_string()))
}
pub(crate) fn discard_dlq_entry(
entry: &DlqEntryRecord,
) -> Result<DlqEntryRecord, OrchestratorError> {
let mut next = entry.clone();
next.state = "discarded".to_string();
next.retry_history.push(DlqAttemptRecord {
attempt: (next.retry_history.len() + 1) as u32,
at: OffsetDateTime::now_utc()
.format(&Rfc3339)
.map_err(|error| error.to_string())?,
status: "discarded".to_string(),
error: None,
});
Ok(next)
}
pub(crate) fn print_json<T>(value: &T) -> Result<(), OrchestratorError>
where
T: Serialize,
{
let encoded = serde_json::to_string_pretty(value).map_err(|error| error.to_string())?;
println!("{encoded}");
Ok(())
}
fn read_state_snapshot(path: &Path) -> Result<Option<PersistedStateSnapshot>, OrchestratorError> {
if !path.is_file() {
return Ok(None);
}
let content = std::fs::read_to_string(path)
.map_err(|error| format!("failed to read {}: {error}", path.display()))?;
let snapshot = serde_json::from_str(&content)
.map_err(|error| format!("failed to parse {}: {error}", path.display()))?;
Ok(Some(snapshot))
}
async fn eval_json<T>(vm: &mut harn_vm::Vm, expr: &str) -> Result<T, OrchestratorError>
where
T: DeserializeOwned,
{
let source = format!("pipeline default(harness: Harness) {{\n return {expr}\n}}\n");
let program = harn_parser::check_source_strict(&source).map_err(|error| error.to_string())?;
let chunk = harn_vm::Compiler::new_runtime_owned_source()
.compile(&program)
.map_err(|error| error.to_string())?;
let value = vm
.execute(&chunk)
.await
.map_err(|error| error.to_string())?;
serde_json::from_value(harn_vm::llm::vm_value_to_json(&value)).map_err(|error| {
OrchestratorError::Common(format!("failed to decode builtin result: {error}"))
})
}
pub(super) use crate::format::{
format_duration_coarse as format_duration, format_timestamp_rfc3339 as format_timestamp,
};
fn age_since(then: OffsetDateTime) -> StdDuration {
let now = OffsetDateTime::now_utc();
if now <= then {
return StdDuration::ZERO;
}
let delta = now - then;
StdDuration::new(
delta.whole_seconds() as u64,
delta.subsec_nanoseconds().try_into().unwrap_or_default(),
)
}