//! Autonomous maintenance runner, standing responsibilities, and model attempts.
use std::{sync::Arc, time::Duration};
use anyhow::{Context, Result};
use klbr_core::{
models::{LlmClient, Message, Role},
orientation::{
AdoptReceipt, AdoptionActor, ApplicationBoundary, CarrierMode, OrientationCandidate,
OrientationError,
},
runtime_turn::{run_turn_loop_with_purpose, TurnEvent},
};
use klbr_runtime::{
behavior::Release,
protocol::{
FrontierSelectReq, HostFuture, OrientationAdoptReq, OrientationCurrentReq, OrientationHost,
OrientationStageReq, Purpose, SessionId,
},
session::SessionRuntime,
skills::SkillRoots,
store::Store,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
pub const STANDING_RESPONSIBILITY_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/standing-responsibility.md");
pub const MAINTENANCE_ATTEMPT_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/maintenance-attempt.md");
pub const EVALUATION_REVIEW_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/evaluation-review.md");
pub const STEERING_FORM_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/steering-form.md");
pub const STEERING_SITUATE_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/steering-situate.md");
pub const STEERING_ELABORATE_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/steering-elaborate.md");
pub const STEERING_REPAIR_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/steering-repair.md");
pub const SELF_STEERING_PROMPT: &str =
include_str!("../../python/klbr-runtime/src/klbr/templates/self-steering.md");
/// The skill a maintenance turn loads by default.
const MAINTENANCE_SKILL_ID: &str = "maintenance";
/// Reads a prompt template from the published release pinned by the session policy,
/// falling back to the package templates directory or compiled default.
pub fn load_prompt_from_release(
release: Option<&Release>,
template_name: &str,
default_content: &'static str,
) -> String {
if let Some(rel) = release {
for candidate in &[
rel.path.join(template_name),
rel.path.join(format!("templates/{template_name}")),
rel.path.join(format!("klbr/templates/{template_name}")),
] {
if candidate.is_file() {
if let Ok(content) = std::fs::read_to_string(candidate) {
return content;
}
}
}
}
default_content.to_string()
}
/// Appends standing responsibility guidance to ordinary session prompts.
pub fn install_standing_responsibility(soul: &str) -> String {
install_standing_responsibility_with_release(soul, None)
}
/// Appends standing responsibility guidance respecting the active published release.
pub fn install_standing_responsibility_with_release(
soul: &str,
release: Option<&Release>,
) -> String {
let prompt = load_prompt_from_release(
release,
"standing-responsibility.md",
STANDING_RESPONSIBILITY_PROMPT,
);
if soul.contains("Standing responsibility") {
soul.to_string()
} else {
format!("{soul}\n\n{prompt}")
}
}
async fn persist_orientation_candidate(
store: &Store,
session_id: &SessionId,
candidate: OrientationCandidate,
) -> Result<Value> {
let semantic = candidate.semantic_payload();
let candidate_json = serde_json::to_value(&candidate)?;
store
.stage_orientation_candidate(
session_id,
&candidate.candidate_id,
&candidate.scope,
semantic,
candidate_json,
)
.await
}
async fn adopt_staged_orientation(
store: &Store,
session_id: &SessionId,
candidate_id: &str,
expected_parent: Option<&str>,
adoption_actor: AdoptionActor,
operation_id: &str,
apply_at: Option<ApplicationBoundary>,
) -> Result<AdoptReceipt> {
let candidate: OrientationCandidate = serde_json::from_value(
store
.orientation_candidate(session_id, candidate_id)
.await?
.context("orientation_candidate_not_found")?,
)?;
let apply_at = apply_at.unwrap_or_default();
if !candidate.scope.is_global() && apply_at != ApplicationBoundary::StoredOnly {
anyhow::bail!(
"unsupported_non_global_orientation_delivery: use apply_at='stored_only' to persist without delivery"
);
}
let scope = &candidate.scope;
let current = store
.current_orientation_revision(session_id, scope)
.await?
.map(serde_json::from_value)
.transpose()?;
let (revision, receipt) = candidate
.prepare_adoption(
current.as_ref(),
expected_parent,
adoption_actor,
operation_id,
apply_at,
1,
)
.map_err(|error| anyhow::anyhow!(error))?;
let canonical_request = json!({
"candidate": candidate.semantic_payload(),
"expected_parent": expected_parent,
"adoption_actor": revision.adoption_actor.clone(),
"apply_at": apply_at,
});
let event = store
.commit_orientation_adoption(
session_id,
scope,
candidate_id,
expected_parent,
operation_id,
canonical_request,
serde_json::to_value(revision)?,
serde_json::to_value(receipt)?,
)
.await?;
serde_json::from_value(
event
.payload
.get("receipt")
.cloned()
.context("orientation_adoption_event_missing_receipt")?,
)
.context("invalid_orientation_adoption_receipt")
}
/// Stages a candidate without selecting it.
pub async fn stage_orientation(
store: &Store,
session_id: &SessionId,
candidate: &OrientationCandidate,
) -> std::result::Result<String, OrientationError> {
persist_orientation_candidate(store, session_id, candidate.clone())
.await
.map_err(|error| OrientationError::WriteFailed(error.to_string()))?;
Ok(candidate.candidate_id.clone())
}
/// Stages then adopts a candidate for legacy internal callers.
pub async fn adopt_orientation(
store: &Store,
session_id: &SessionId,
candidate: &OrientationCandidate,
expected_parent: Option<&str>,
adoption_actor: AdoptionActor,
operation_id: &str,
apply_at: Option<&str>,
) -> std::result::Result<AdoptReceipt, OrientationError> {
stage_orientation(store, session_id, candidate).await?;
let apply_at = apply_at
.map(ApplicationBoundary::try_from)
.transpose()
.map_err(OrientationError::UnsupportedApplicationBoundary)?;
match adopt_staged_orientation(
store,
session_id,
&candidate.candidate_id,
expected_parent,
adoption_actor,
operation_id,
apply_at,
)
.await
{
Ok(receipt) => Ok(receipt),
Err(error)
if error
.to_string()
.contains("orientation_stale_parent_conflict") =>
{
let actual = store
.current_orientation_revision(session_id, &candidate.scope)
.await
.map_err(|error| OrientationError::WriteFailed(error.to_string()))?
.and_then(|revision| {
revision
.get("revision_id")
.and_then(Value::as_str)
.map(str::to_owned)
});
Err(OrientationError::StaleParentConflict {
expected: expected_parent.map(str::to_owned),
actual,
})
}
Err(error) => Err(OrientationError::WriteFailed(error.to_string())),
}
}
/// Typed verdict for a completed maintenance attempt.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "verdict", rename_all = "snake_case")]
pub enum MaintenanceVerdict {
Activated { revision: String, epoch: i64 },
NoChange { reason: String },
NeedsEvidence { missing: Vec<String> },
CandidateFailed { error: String },
}
/// Bounded context for an autonomous maintenance attempt.
#[derive(Debug, Clone)]
pub struct MaintenanceAttemptContext {
pub item_id: String,
pub purpose: String,
pub trigger_cause: String,
pub evidence_refs: Vec<String>,
pub max_steps: usize,
pub cell_budget: Duration,
pub prompt_override: Option<String>,
pub steering_carrier: Option<String>,
/// The session's skill roots. A maintenance turn loads `maintenance` by
/// default, because it is the one skill whose subject is the attempt itself;
/// every other id stays one load away in the index.
pub skills: Option<SkillRoots>,
}
impl MaintenanceAttemptContext {
pub fn new(
item_id: impl Into<String>,
purpose: impl Into<String>,
trigger_cause: impl Into<String>,
) -> Self {
Self {
item_id: item_id.into(),
purpose: purpose.into(),
trigger_cause: trigger_cause.into(),
evidence_refs: Vec::new(),
max_steps: 10,
cell_budget: Duration::from_secs(30),
prompt_override: None,
steering_carrier: None,
skills: None,
}
}
pub fn with_skills(mut self, skills: SkillRoots) -> Self {
self.skills = Some(skills);
self
}
pub fn with_evidence_refs(mut self, refs: Vec<String>) -> Self {
self.evidence_refs = refs;
self
}
pub fn with_prompt_override(mut self, prompt: impl Into<String>) -> Self {
self.prompt_override = Some(prompt.into());
self
}
pub fn with_release(mut self, release: &Release) -> Self {
self.prompt_override = Some(load_prompt_from_release(
Some(release),
"maintenance-attempt.md",
MAINTENANCE_ATTEMPT_PROMPT,
));
self
}
/// Explicitly binds an optional steering template at its declared role/placement (S22).
/// Composes alongside the maintenance attempt prompt rather than displacing it.
pub fn with_steering_mode(mut self, mode: CarrierMode, release: Option<&Release>) -> Self {
let (template_bytes, _hash) =
klbr_core::orientation::load_steering_template(mode, release.map(|r| r.path.as_path()));
self.steering_carrier = Some(template_bytes);
self
}
/// Assembles prompt messages for the maintenance attempt, binding the
/// maintenance template, cause metadata, and exact evidence references.
pub fn assemble_messages(&self, existing_history: &[Message]) -> Vec<Message> {
let mut messages = Vec::new();
// 1. Maintenance prompt template as system message (from release or default)
let system_prompt = self
.prompt_override
.as_deref()
.unwrap_or(MAINTENANCE_ATTEMPT_PROMPT);
messages.push(Message::system(system_prompt));
let contract = match self.skills.as_ref() {
Some(skills) => klbr_core::runtime_turn::contract_message(skills),
None => klbr_core::runtime_turn::RUNTIME_CONTRACT.to_string(),
};
messages.push(Message::system(contract));
if let Some(skills) = self.skills.as_ref() {
if let Some(skill) =
klbr_core::runtime_turn::skill_message(skills, MAINTENANCE_SKILL_ID)
{
messages.push(Message::system(skill));
}
}
// Composed steering carrier at declared role/placement (S22)
if let Some(ref carrier) = self.steering_carrier {
messages.push(Message::system(carrier));
}
// 2. Non-system history messages
for msg in existing_history {
if msg.role != Role::System {
messages.push(msg.clone());
}
}
// 3. User message framing the assigned maintenance work with explicit evidence tags
let mut user_content = format!(
"<maintenance_assignment id=\"{0}\" purpose=\"{1}\" cause=\"{2}\">\n",
self.item_id, self.purpose, self.trigger_cause
);
if !self.evidence_refs.is_empty() {
user_content.push_str("<evidence_references>\n");
for r in &self.evidence_refs {
user_content.push_str(&format!(" <ref id=\"{r}\"/>\n"));
}
user_content.push_str("</evidence_references>\n");
}
user_content.push_str(
"Diagnose the observed discrepancy, author any validated candidate in the designated candidate workspace, run checks, and activate if verified.\n</maintenance_assignment>"
);
messages.push(Message::user(user_content));
messages
}
}
pub struct MaintenanceRunner;
impl MaintenanceRunner {
/// Executes a maintenance model attempt through the existing provider adapter.
/// Workbench executions during this attempt run with Purpose::Maintenance.
pub async fn run_attempt(
llm: &LlmClient,
session: &Arc<SessionRuntime>,
ctx: &MaintenanceAttemptContext,
history: &[Message],
) -> Result<MaintenanceVerdict> {
let policy = session.policy().await;
let mut resolved_ctx = ctx.clone();
if resolved_ctx.prompt_override.is_none() {
resolved_ctx = resolved_ctx.with_release(&policy.release);
}
let messages = resolved_ctx.assemble_messages(history);
let initial_epoch = session.hook_epoch().await;
let turn_stream = run_turn_loop_with_purpose(
llm,
session,
messages,
ctx.max_steps,
ctx.cell_budget,
Purpose::Maintenance,
);
tokio::pin!(turn_stream);
use futures::StreamExt;
let mut last_error = None;
while let Some(event) = turn_stream.next().await {
match event {
TurnEvent::Failed(err) => {
last_error = Some(err);
}
TurnEvent::CellOutput(out)
if out.stream == klbr_core::runtime_turn::CellOutputStream::Stderr =>
{
last_error = Some(out.text);
}
_ => {}
}
}
let final_epoch = session.hook_epoch().await;
let policy = session.policy().await;
let final_revision = policy.release.id.clone();
let mut missing_refs = Vec::new();
for r in &ctx.evidence_refs {
if let Some(rest) = r.strip_prefix("event:") {
if let Ok(seq) = rest.parse::<i64>() {
if session
.store
.get_event(&session.id, seq)
.await
.ok()
.flatten()
.is_none()
{
missing_refs.push(r.clone());
}
}
} else if let Some(case_id) = r.strip_prefix("case:") {
if session
.store
.get_observer_case(case_id)
.await
.ok()
.flatten()
.is_none()
{
missing_refs.push(r.clone());
}
}
}
let verdict = if !missing_refs.is_empty() {
MaintenanceVerdict::NeedsEvidence {
missing: missing_refs,
}
} else if final_epoch > initial_epoch {
MaintenanceVerdict::Activated {
revision: final_revision,
epoch: final_epoch,
}
} else if let Some(err) = last_error {
MaintenanceVerdict::CandidateFailed { error: err }
} else {
MaintenanceVerdict::NoChange {
reason: "maintenance attempt finished without candidate activation".to_string(),
}
};
// Durably record the maintenance verdict event in the session store
session
.store
.append(
&session.id,
"maintenance.attempt_completed",
json!({
"type": "maintenance.attempt_completed",
"item_id": ctx.item_id,
"purpose": ctx.purpose,
"verdict": verdict,
}),
)
.await
.context("failed to record maintenance attempt completion")?;
Ok(verdict)
}
}
fn candidate_from_stage(req: OrientationStageReq) -> OrientationCandidate {
let mut candidate = OrientationCandidate::new(
req.candidate_id,
req.text,
req.scope,
req.mode,
req.target,
req.target_selector,
req.carrier_author,
req.parent_revision,
req.source_refs,
);
candidate.carrier_placement = req.carrier_placement.unwrap_or_default();
candidate
}
/// Production implementation of OrientationHost injecting real klbr-core and klbr-daemon
/// capabilities into klbr-runtime without a cyclic crate dependency.
pub struct DaemonOrientationHost;
impl OrientationHost for DaemonOrientationHost {
fn stage_orientation<'a>(
&'a self,
store: &'a Store,
session_id: &'a SessionId,
req: OrientationStageReq,
) -> HostFuture<'a> {
Box::pin(async move {
persist_orientation_candidate(store, session_id, candidate_from_stage(req)).await
})
}
fn current_orientation<'a>(
&'a self,
store: &'a Store,
session_id: &'a SessionId,
req: OrientationCurrentReq,
) -> HostFuture<'a> {
Box::pin(async move {
Ok(store
.current_orientation_revision(session_id, &req.scope)
.await?
.unwrap_or(Value::Null))
})
}
fn adopt_orientation<'a>(
&'a self,
store: &'a Store,
session_id: &'a SessionId,
req: OrientationAdoptReq,
) -> HostFuture<'a> {
Box::pin(async move {
let receipt = adopt_staged_orientation(
store,
session_id,
&req.candidate_id,
req.expected_parent.as_deref(),
req.adoption_actor,
&req.operation_id,
req.apply_at,
)
.await?;
Ok(serde_json::to_value(receipt)?)
})
}
fn select_frontier<'a>(
&'a self,
store: &'a Store,
session_id: &'a SessionId,
req: FrontierSelectReq,
) -> HostFuture<'a> {
Box::pin(async move {
match req.action.as_str() {
"select" => {
let now = crate::disposition::session_head_event_id(store, session_id).await;
let work_id = format!("work-{}", req.frontier_id);
let work_item = klbr_runtime::work::WorkItem::new(
&work_id,
&req.frontier_id,
klbr_runtime::work::RunCause::MaintenanceReady {
item_id: work_id.clone(),
purpose: req.frontier_id.clone(),
},
klbr_runtime::work::ConsequenceMetadata::new(klbr_runtime::work::Impact::High),
klbr_runtime::work::Provenance::agent("frontier.select", now),
now,
)
.with_session(session_id.to_string())
.with_evidence_refs(req.source_refs.clone());
store.insert_work_item(&work_item).await?;
let mut adoption_receipt = None;
if let Some(ref carrier_text) = req.candidate_carrier {
let candidate = klbr_core::orientation::OrientationCandidate::new(
req.frontier_id.clone(),
carrier_text.clone(),
klbr_core::orientation::OrientationScope::Global,
CarrierMode::Form,
req.frontier_id.clone(),
klbr_core::orientation::TargetSelector::Agent,
klbr_core::orientation::CarrierAuthor::Actor { actor_id: "agent".into() },
None,
req.source_refs.clone(),
);
let op_id = format!("op-frontier-{}", req.frontier_id);
let receipt = adopt_orientation(
store,
session_id,
&candidate,
None,
klbr_core::orientation::AdoptionActor::Agent,
&op_id,
Some("next_provider_boundary"),
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
adoption_receipt = Some(receipt);
}
let selection_event = serde_json::json!({
"type": "frontier.selected",
"frontier_id": req.frontier_id,
"action": "select",
"adoption_actor": req.adoption_actor,
"work_item_id": work_id,
"candidate_carrier": req.candidate_carrier,
});
store.append(session_id, "frontier.selected", selection_event).await?;
Ok(serde_json::json!({
"ok": true,
"frontier_id": req.frontier_id,
"work_item_id": work_id,
"adoption_receipt": adoption_receipt,
}))
}
"defer" => {
let now = crate::disposition::session_head_event_id(store, session_id).await;
let work_id = format!("work-{}", req.frontier_id);
let mut work_item = klbr_runtime::work::WorkItem::new(
&work_id,
&req.frontier_id,
klbr_runtime::work::RunCause::MaintenanceReady {
item_id: work_id.clone(),
purpose: req.frontier_id.clone(),
},
klbr_runtime::work::ConsequenceMetadata::new(klbr_runtime::work::Impact::Low),
klbr_runtime::work::Provenance::agent("frontier.defer", now),
now,
)
.with_session(session_id.to_string());
let reason = req.rationale.unwrap_or_else(|| "deferred awaiting condition".into());
work_item.status = klbr_runtime::work::WorkStatus::Deferred { reason: reason.clone() };
work_item.consequence.readiness = klbr_runtime::work::Readiness::Paused { reason: reason.clone() };
store.insert_work_item(&work_item).await?;
let defer_event = serde_json::json!({
"type": "frontier.deferred",
"frontier_id": req.frontier_id,
"action": "defer",
"actor": req.adoption_actor,
"deadline_ms": req.deadline_ms,
"reason": reason,
});
store.append(session_id, "frontier.deferred", defer_event).await?;
Ok(serde_json::json!({
"ok": true,
"frontier_id": req.frontier_id,
"action": "deferred",
}))
}
"abandon" => {
let abandon_event = serde_json::json!({
"type": "frontier.abandoned",
"frontier_id": req.frontier_id,
"action": "abandon",
"actor": req.adoption_actor,
"rationale": req.rationale,
});
store.append(session_id, "frontier.abandoned", abandon_event).await?;
Ok(serde_json::json!({
"ok": true,
"frontier_id": req.frontier_id,
"action": "abandoned",
}))
}
"quiet" => {
let quiet_event = serde_json::json!({
"type": "frontier.quiet",
"action": "quiet",
"rationale": req.rationale,
});
store.append(session_id, "frontier.quiet", quiet_event).await?;
Ok(serde_json::json!({
"ok": true,
"action": "quiet",
}))
}
other => anyhow::bail!("unknown_frontier_action: '{other}' is not an admissible action (must be select | defer | abandon | quiet)"),
}
})
}
}