GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
This repository has been archived on 2026-10-29. You can view files and clone it, but you cannot make any changes to its state, such as pushing and creating new issues, pull requests or comments.
Epiphany/epiphany-core/src/runtime_spine.rs
Meta Vens dd5530ecb5 Clarify CultNet receipt authority
2026-05-22 23:15:07 +01:00

2325 lines
89 KiB
Rust
Raw Blame History

use crate::EpiphanyWorkerLaunchDocument;
use crate::agent_memory::AGENT_MEMORY_TYPE;
use crate::heartbeat_state::HEARTBEAT_STATE_SCHEMA_VERSION;
use crate::heartbeat_state::HEARTBEAT_STATE_TYPE;
use crate::memory_graph::MEMORY_GRAPH_SCHEMA_VERSION;
use crate::memory_graph::MEMORY_GRAPH_TYPE;
use crate::state_ledger::STATE_LEDGER_SCHEMA_VERSION;
use crate::state_ledger::STATE_LEDGER_STORE_TYPE;
use crate::thread_state_store::THREAD_STATE_SCHEMA_VERSION;
use crate::thread_state_store::THREAD_STATE_TYPE;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use cultcache_rs::CultCache;
use cultcache_rs::DatabaseEntry;
use cultcache_rs::SingleFileMessagePackBackingStore;
use cultnet_rs::CultNetDocumentMutationContract;
use cultnet_rs::CultNetDocumentOperation;
use cultnet_rs::CultNetMessage;
use cultnet_rs::CultNetMutationAuthority;
use cultnet_rs::CultNetSchemaKind;
use cultnet_rs::CultNetSchemaRegistration;
use cultnet_rs::CultNetSchemaRegistry;
use cultnet_rs::CultNetWireContract;
use cultnet_rs::builtin_schema_registry;
use cultnet_rs::encode_cultnet_message_to_vec;
use cultnet_rs::encode_frame;
use epiphany_openai_adapter::EpiphanyOpenAiAdapterStatus;
use epiphany_openai_adapter::EpiphanyOpenAiModelReceipt;
use epiphany_openai_adapter::EpiphanyOpenAiModelRequest;
use epiphany_openai_adapter::EpiphanyOpenAiStreamEvent;
use epiphany_state_model::EpiphanyJobBinding;
use epiphany_state_model::EpiphanyJobKind;
use epiphany_state_model::EpiphanyRuntimeLink;
use epiphany_state_model::EpiphanyThreadState;
use serde::Deserialize;
use serde::Serialize;
use std::collections::BTreeMap;
use std::fs;
use std::path::Path;
use std::path::PathBuf;
pub const RUNTIME_IDENTITY_TYPE: &str = "epiphany.runtime.identity";
pub const RUNTIME_SESSION_TYPE: &str = "epiphany.runtime.session";
pub const RUNTIME_JOB_TYPE: &str = "epiphany.runtime.job";
pub const RUNTIME_WORKER_LAUNCH_REQUEST_TYPE: &str = "epiphany.runtime.worker_launch_request";
pub const RUNTIME_ROLE_WORKER_RESULT_TYPE: &str = "epiphany.runtime.role_worker_result";
pub const RUNTIME_REORIENT_WORKER_RESULT_TYPE: &str = "epiphany.runtime.reorient_worker_result";
pub const RUNTIME_JOB_RESULT_TYPE: &str = "epiphany.runtime.job_result";
pub const RUNTIME_EVENT_TYPE: &str = "epiphany.runtime.event";
pub const OPENAI_ADAPTER_STATUS_TYPE: &str = "epiphany.openai_adapter_status.v0";
pub const OPENAI_MODEL_REQUEST_TYPE: &str = "epiphany.openai_model_request.v0";
pub const OPENAI_MODEL_STREAM_EVENT_TYPE: &str = "epiphany.openai_model_stream_event.v0";
pub const OPENAI_MODEL_RECEIPT_TYPE: &str = "epiphany.openai_model_receipt.v0";
pub const SURFACE_SCENE_TYPE: &str = "epiphany.surface.scene";
pub const SURFACE_FRESHNESS_TYPE: &str = "epiphany.surface.freshness";
pub const SURFACE_CONTEXT_TYPE: &str = "epiphany.surface.context";
pub const SURFACE_GRAPH_QUERY_TYPE: &str = "epiphany.surface.graph_query";
pub const SURFACE_PRESSURE_TYPE: &str = "epiphany.surface.pressure";
pub const SURFACE_REORIENT_TYPE: &str = "epiphany.surface.reorient";
pub const SURFACE_CRRC_TYPE: &str = "epiphany.surface.crrc";
pub const SURFACE_JOBS_TYPE: &str = "epiphany.surface.jobs";
pub const SURFACE_ROLES_TYPE: &str = "epiphany.surface.roles";
pub const SURFACE_ROLE_RESULT_TYPE: &str = "epiphany.surface.role_result";
pub const SURFACE_REORIENT_RESULT_TYPE: &str = "epiphany.surface.reorient_result";
pub const SURFACE_PLANNING_TYPE: &str = "epiphany.surface.planning";
pub const SURFACE_COORDINATOR_TYPE: &str = "epiphany.surface.coordinator";
pub const SURFACE_FACE_TYPE: &str = "epiphany.surface.face";
pub const SURFACE_VOID_MEMORY_TYPE: &str = "epiphany.surface.void_memory";
pub const SURFACE_REPO_INITIALIZATION_TYPE: &str = "epiphany.surface.repo_initialization";
pub const SURFACE_REPO_BIRTH_RUNNER_TYPE: &str = "epiphany.surface.repo_birth_runner";
pub const SURFACE_RIDER_BRIDGE_TYPE: &str = "epiphany.surface.rider_bridge";
pub const SURFACE_UNITY_BRIDGE_TYPE: &str = "epiphany.surface.unity_bridge";
pub const RUNTIME_IDENTITY_KEY: &str = "self";
pub const RUNTIME_SPINE_SCHEMA_VERSION: &str = "epiphany.runtime_spine.v0";
pub const RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION: &str =
"epiphany.runtime.worker_launch_request.v0";
pub const RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION: &str =
"epiphany.runtime.role_worker_result.v0";
pub const RUNTIME_REORIENT_WORKER_RESULT_SCHEMA_VERSION: &str =
"epiphany.runtime.reorient_worker_result.v0";
pub const OPENAI_ADAPTER_STATUS_SCHEMA_VERSION: &str = "epiphany.openai_adapter_status.v0";
pub const OPENAI_MODEL_REQUEST_SCHEMA_VERSION: &str = "epiphany.openai_model_request.v0";
pub const OPENAI_MODEL_STREAM_EVENT_SCHEMA_VERSION: &str = "epiphany.openai_model_stream_event.v0";
pub const OPENAI_MODEL_RECEIPT_SCHEMA_VERSION: &str = "epiphany.openai_model_receipt.v0";
pub const SCENE_SURFACE_SCHEMA_VERSION: &str = "epiphany.scene_surface.v0";
pub const FRESHNESS_SURFACE_SCHEMA_VERSION: &str = "epiphany.freshness_surface.v0";
pub const CONTEXT_SURFACE_SCHEMA_VERSION: &str = "epiphany.context_surface.v0";
pub const GRAPH_QUERY_SURFACE_SCHEMA_VERSION: &str = "epiphany.graph_query_surface.v0";
pub const PRESSURE_SURFACE_SCHEMA_VERSION: &str = "epiphany.pressure_surface.v0";
pub const REORIENT_SURFACE_SCHEMA_VERSION: &str = "epiphany.reorient_surface.v0";
pub const CRRC_SURFACE_SCHEMA_VERSION: &str = "epiphany.crrc_surface.v0";
pub const JOBS_SURFACE_SCHEMA_VERSION: &str = "epiphany.jobs_surface.v0";
pub const ROLES_SURFACE_SCHEMA_VERSION: &str = "epiphany.roles_surface.v0";
pub const ROLE_RESULT_SURFACE_SCHEMA_VERSION: &str = "epiphany.role_result_surface.v0";
pub const REORIENT_RESULT_SURFACE_SCHEMA_VERSION: &str = "epiphany.reorient_result_surface.v0";
pub const PLANNING_SURFACE_SCHEMA_VERSION: &str = "epiphany.planning_surface.v0";
pub const COORDINATOR_SURFACE_SCHEMA_VERSION: &str = "epiphany.coordinator_surface.v0";
pub const FACE_SURFACE_SCHEMA_VERSION: &str = "epiphany.face_surface.v0";
pub const VOID_MEMORY_SURFACE_SCHEMA_VERSION: &str = "epiphany.void_memory_surface.v0";
pub const REPO_INITIALIZATION_SURFACE_SCHEMA_VERSION: &str =
"epiphany.repo_initialization_surface.v0";
pub const REPO_BIRTH_RUNNER_SURFACE_SCHEMA_VERSION: &str = "epiphany.repo_birth_runner_surface.v0";
pub const RIDER_BRIDGE_SURFACE_SCHEMA_VERSION: &str = "epiphany.rider_bridge_surface.v0";
pub const UNITY_BRIDGE_SURFACE_SCHEMA_VERSION: &str = "epiphany.unity_bridge_surface.v0";
pub const AGENT_MEMORY_PAYLOAD_SCHEMA_VERSION: &str = "epiphany.agent_memory.v0";
pub const CULTNET_SCHEMA_INDEX_RELATIVE: &str = "schemas/cultnet/index.json";
#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
struct EpiphanyCultNetSchemaIndex {
schema_version: String,
schemas: Vec<EpiphanyCultNetSchemaIndexEntry>,
}
#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
struct EpiphanyCultNetSchemaIndexEntry {
schema_id: String,
kind: CultNetSchemaKind,
wire_contracts: Vec<CultNetWireContract>,
#[serde(default)]
schema_version: Option<String>,
#[serde(default)]
document_type: Option<String>,
#[serde(default)]
title: Option<String>,
path: String,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "epiphany.runtime.identity", schema = "EpiphanyRuntimeIdentity")]
pub struct EpiphanyRuntimeIdentity {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub runtime_id: String,
#[cultcache(key = 2)]
pub display_name: String,
#[cultcache(key = 3)]
pub runtime_kind: String,
#[cultcache(key = 4)]
pub created_at: String,
#[cultcache(key = 5)]
pub updated_at: String,
#[cultcache(key = 6)]
pub supported_document_types: Vec<String>,
#[cultcache(key = 7, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "epiphany.runtime.session", schema = "EpiphanyRuntimeSession")]
pub struct EpiphanyRuntimeSession {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub session_id: String,
#[cultcache(key = 2)]
pub objective: String,
#[cultcache(key = 3)]
pub status: EpiphanyRuntimeSessionStatus,
#[cultcache(key = 4)]
pub created_at: String,
#[cultcache(key = 5)]
pub updated_at: String,
#[cultcache(key = 6, default)]
pub coordinator_note: String,
#[cultcache(key = 7, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "epiphany.runtime.job", schema = "EpiphanyRuntimeJob")]
pub struct EpiphanyRuntimeJob {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub job_id: String,
#[cultcache(key = 2)]
pub session_id: String,
#[cultcache(key = 3)]
pub role: String,
#[cultcache(key = 4)]
pub status: EpiphanyRuntimeJobStatus,
#[cultcache(key = 5)]
pub created_at: String,
#[cultcache(key = 6)]
pub updated_at: String,
#[cultcache(key = 7, default)]
pub summary: String,
#[cultcache(key = 8, default)]
pub artifact_refs: Vec<String>,
#[cultcache(key = 9, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "epiphany.runtime.worker_launch_request",
schema = "EpiphanyRuntimeWorkerLaunchRequest"
)]
pub struct EpiphanyRuntimeWorkerLaunchRequest {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub job_id: String,
#[cultcache(key = 2)]
pub binding_id: String,
#[cultcache(key = 3)]
pub role: String,
#[cultcache(key = 4)]
pub authority_scope: String,
#[cultcache(key = 5)]
pub instruction: String,
#[cultcache(key = 6)]
pub output_contract_id: String,
#[cultcache(key = 7)]
pub document_kind: String,
#[cultcache(key = 8)]
pub launch_document_msgpack: Vec<u8>,
#[cultcache(key = 9, default)]
pub metadata: BTreeMap<String, String>,
}
impl EpiphanyRuntimeWorkerLaunchRequest {
pub fn launch_document(&self) -> Result<EpiphanyWorkerLaunchDocument> {
let document: EpiphanyWorkerLaunchDocument =
rmp_serde::from_slice(&self.launch_document_msgpack)
.context("failed to decode worker launch document MessagePack")?;
let actual_kind = worker_launch_document_kind(&document);
if actual_kind != self.document_kind {
return Err(anyhow!(
"worker launch document kind mismatch: indexed {:?}, payload {:?}",
self.document_kind,
actual_kind
));
}
Ok(document)
}
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "epiphany.runtime.role_worker_result",
schema = "EpiphanyRuntimeRoleWorkerResult"
)]
pub struct EpiphanyRuntimeRoleWorkerResult {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub result_id: String,
#[cultcache(key = 2)]
pub job_id: String,
#[cultcache(key = 3)]
pub role_id: String,
#[cultcache(key = 4)]
pub verdict: String,
#[cultcache(key = 5)]
pub summary: String,
#[cultcache(key = 6)]
pub next_safe_move: String,
#[cultcache(key = 7, default)]
pub checkpoint_summary: Option<String>,
#[cultcache(key = 8, default)]
pub scratch_summary: Option<String>,
#[cultcache(key = 9, default)]
pub files_inspected: Vec<String>,
#[cultcache(key = 10, default)]
pub frontier_node_ids: Vec<String>,
#[cultcache(key = 11, default)]
pub evidence_ids: Vec<String>,
#[cultcache(key = 12, default)]
pub artifact_refs: Vec<String>,
#[cultcache(key = 13, default)]
pub open_questions: Vec<String>,
#[cultcache(key = 14, default)]
pub evidence_gaps: Vec<String>,
#[cultcache(key = 15, default)]
pub risks: Vec<String>,
#[cultcache(key = 16, default)]
pub state_patch_msgpack: Option<Vec<u8>>,
#[cultcache(key = 17, default)]
pub self_patch_msgpack: Option<Vec<u8>>,
#[cultcache(key = 18, default)]
pub item_error: Option<String>,
#[cultcache(key = 19, default)]
pub metadata: BTreeMap<String, String>,
}
impl EpiphanyRuntimeRoleWorkerResult {
pub fn state_patch(&self) -> Result<Option<crate::EpiphanyRoleStatePatchDocument>> {
decode_optional_msgpack(
self.state_patch_msgpack.as_deref(),
"role worker statePatch",
)
}
pub fn self_patch(&self) -> Result<Option<crate::AgentSelfPatch>> {
decode_optional_msgpack(self.self_patch_msgpack.as_deref(), "role worker selfPatch")
}
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "epiphany.runtime.reorient_worker_result",
schema = "EpiphanyRuntimeReorientWorkerResult"
)]
pub struct EpiphanyRuntimeReorientWorkerResult {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub result_id: String,
#[cultcache(key = 2)]
pub job_id: String,
#[cultcache(key = 3)]
pub mode: String,
#[cultcache(key = 4)]
pub summary: String,
#[cultcache(key = 5)]
pub next_safe_move: String,
#[cultcache(key = 6, default)]
pub checkpoint_still_valid: Option<bool>,
#[cultcache(key = 7, default)]
pub files_inspected: Vec<String>,
#[cultcache(key = 8, default)]
pub frontier_node_ids: Vec<String>,
#[cultcache(key = 9, default)]
pub evidence_ids: Vec<String>,
#[cultcache(key = 10, default)]
pub artifact_refs: Vec<String>,
#[cultcache(key = 11, default)]
pub open_questions: Vec<String>,
#[cultcache(key = 12, default)]
pub continuity_risks: Vec<String>,
#[cultcache(key = 13, default)]
pub item_error: Option<String>,
#[cultcache(key = 14, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "epiphany.runtime.job_result",
schema = "EpiphanyRuntimeJobResult"
)]
pub struct EpiphanyRuntimeJobResult {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub result_id: String,
#[cultcache(key = 2)]
pub job_id: String,
#[cultcache(key = 3)]
pub session_id: String,
#[cultcache(key = 4)]
pub role: String,
#[cultcache(key = 5)]
pub verdict: String,
#[cultcache(key = 6)]
pub summary: String,
#[cultcache(key = 7)]
pub completed_at: String,
#[cultcache(key = 8, default)]
pub next_safe_move: String,
#[cultcache(key = 9, default)]
pub evidence_refs: Vec<String>,
#[cultcache(key = 10, default)]
pub artifact_refs: Vec<String>,
#[cultcache(key = 11, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "epiphany.runtime.event", schema = "EpiphanyRuntimeEvent")]
pub struct EpiphanyRuntimeEvent {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub event_id: String,
#[cultcache(key = 2)]
pub occurred_at: String,
#[cultcache(key = 3)]
pub event_type: String,
#[cultcache(key = 4)]
pub source: String,
#[cultcache(key = 5, default)]
pub session_id: Option<String>,
#[cultcache(key = 6, default)]
pub job_id: Option<String>,
#[cultcache(key = 7, default)]
pub summary: String,
#[cultcache(key = 8, default)]
pub metadata: BTreeMap<String, String>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum EpiphanyRuntimeSessionStatus {
#[default]
Proposed,
Active,
WaitingForReview,
Sleeping,
Completed,
Archived,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum EpiphanyRuntimeJobStatus {
#[default]
Queued,
Running,
WaitingForReview,
Completed,
Failed,
Cancelled,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct EpiphanyRuntimeSpineStatus {
pub store: String,
pub present: bool,
pub runtime_id: Option<String>,
pub display_name: Option<String>,
pub sessions: usize,
pub active_sessions: usize,
pub jobs: usize,
pub open_jobs: usize,
pub job_results: usize,
pub events: usize,
pub supported_document_types: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeSpineInitOptions {
pub runtime_id: String,
pub display_name: String,
pub created_at: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeSpineSessionOptions {
pub session_id: String,
pub objective: String,
pub created_at: String,
pub coordinator_note: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeSpineEventOptions {
pub event_id: String,
pub occurred_at: String,
pub event_type: String,
pub source: String,
pub session_id: Option<String>,
pub job_id: Option<String>,
pub summary: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeSpineJobOptions {
pub job_id: String,
pub session_id: String,
pub role: String,
pub created_at: String,
pub summary: String,
pub artifact_refs: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RuntimeSpineJobResultOptions {
pub result_id: String,
pub job_id: String,
pub completed_at: String,
pub verdict: String,
pub summary: String,
pub next_safe_move: String,
pub evidence_refs: Vec<String>,
pub artifact_refs: Vec<String>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct RuntimeSpineHeartbeatJobOptions {
pub runtime_id: String,
pub display_name: String,
pub session_id: String,
pub objective: String,
pub coordinator_note: String,
pub job_id: String,
pub role: String,
pub binding_id: String,
pub authority_scope: String,
pub instruction: String,
pub launch_document: EpiphanyWorkerLaunchDocument,
pub output_contract_id: String,
pub created_at: String,
}
#[derive(Clone, Debug, PartialEq)]
pub struct RuntimeSpineHeartbeatLaunchPlanOptions {
pub binding_id: String,
pub kind: EpiphanyJobKind,
pub scope: String,
pub owner_role: String,
pub authority_scope: String,
pub linked_subgoal_ids: Vec<String>,
pub linked_graph_node_ids: Vec<String>,
pub instruction: String,
pub launch_document: EpiphanyWorkerLaunchDocument,
pub output_contract_id: String,
pub max_runtime_seconds: Option<u64>,
pub runtime_job_id: String,
}
#[derive(Clone, Debug, PartialEq)]
pub struct RuntimeSpineHeartbeatLaunchPlan {
pub binding: EpiphanyJobBinding,
pub runtime_link: EpiphanyRuntimeLink,
}
#[derive(Debug, Clone, PartialEq)]
pub struct EpiphanyJobLaunchRequest {
pub expected_revision: Option<u64>,
pub binding_id: String,
pub kind: EpiphanyJobKind,
pub scope: String,
pub owner_role: String,
pub authority_scope: String,
pub linked_subgoal_ids: Vec<String>,
pub linked_graph_node_ids: Vec<String>,
pub instruction: String,
pub launch_document: EpiphanyWorkerLaunchDocument,
pub output_contract_id: String,
pub max_runtime_seconds: Option<u64>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct EpiphanyJobLaunchResult {
pub epiphany_state: EpiphanyThreadState,
pub binding_id: String,
pub launcher_job_id: String,
pub backend_job_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EpiphanyJobInterruptRequest {
pub expected_revision: Option<u64>,
pub binding_id: String,
pub reason: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct EpiphanyJobInterruptResult {
pub epiphany_state: EpiphanyThreadState,
pub binding_id: String,
pub cancel_requested: bool,
pub interrupted_thread_ids: Vec<String>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct EpiphanyRuntimeJobSnapshot {
pub job: EpiphanyRuntimeJob,
pub result: Option<EpiphanyRuntimeJobResult>,
}
pub fn runtime_spine_cache(store_path: impl AsRef<Path>) -> Result<CultCache> {
let store_path = store_path.as_ref();
let mut cache = CultCache::new();
cache.register_entry_type::<EpiphanyRuntimeIdentity>()?;
cache.register_entry_type::<EpiphanyRuntimeSession>()?;
cache.register_entry_type::<EpiphanyRuntimeJob>()?;
cache.register_entry_type::<EpiphanyRuntimeWorkerLaunchRequest>()?;
cache.register_entry_type::<EpiphanyRuntimeRoleWorkerResult>()?;
cache.register_entry_type::<EpiphanyRuntimeReorientWorkerResult>()?;
cache.register_entry_type::<EpiphanyRuntimeJobResult>()?;
cache.register_entry_type::<EpiphanyRuntimeEvent>()?;
cache.register_entry_type::<EpiphanyOpenAiAdapterStatus>()?;
cache.register_entry_type::<EpiphanyOpenAiModelRequest>()?;
cache.register_entry_type::<EpiphanyOpenAiStreamEvent>()?;
cache.register_entry_type::<EpiphanyOpenAiModelReceipt>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(
store_path.to_path_buf(),
));
Ok(cache)
}
pub fn initialize_runtime_spine(
store_path: impl AsRef<Path>,
options: RuntimeSpineInitOptions,
) -> Result<EpiphanyRuntimeIdentity> {
validate_non_empty(&options.runtime_id, "runtime id")?;
validate_non_empty(&options.display_name, "display name")?;
validate_non_empty(&options.created_at, "created at")?;
let store_path = store_path.as_ref();
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let existing = cache.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?;
let created_at = existing
.as_ref()
.map(|identity| identity.created_at.clone())
.unwrap_or_else(|| options.created_at.clone());
let identity = EpiphanyRuntimeIdentity {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
runtime_id: options.runtime_id,
display_name: options.display_name,
runtime_kind: "epiphany.native".to_string(),
created_at,
updated_at: options.created_at,
supported_document_types: supported_runtime_document_types(),
metadata: BTreeMap::from([("codexEvacuationBridge".to_string(), "temporary".to_string())]),
};
cache.put(RUNTIME_IDENTITY_KEY, &identity)?;
Ok(identity)
}
pub fn create_runtime_session(
store_path: impl AsRef<Path>,
options: RuntimeSpineSessionOptions,
) -> Result<EpiphanyRuntimeSession> {
validate_non_empty(&options.session_id, "session id")?;
validate_non_empty(&options.objective, "objective")?;
validate_non_empty(&options.created_at, "created at")?;
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
if cache
.get::<EpiphanyRuntimeSession>(&options.session_id)?
.is_some()
{
return Err(anyhow!(
"runtime session {:?} already exists",
options.session_id
));
}
let session = EpiphanyRuntimeSession {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
session_id: options.session_id.clone(),
objective: options.objective,
status: EpiphanyRuntimeSessionStatus::Active,
created_at: options.created_at.clone(),
updated_at: options.created_at,
coordinator_note: options.coordinator_note,
metadata: BTreeMap::new(),
};
cache.put(&options.session_id, &session)?;
Ok(session)
}
pub fn ensure_runtime_session(
store_path: impl AsRef<Path>,
options: RuntimeSpineSessionOptions,
) -> Result<EpiphanyRuntimeSession> {
validate_non_empty(&options.session_id, "session id")?;
validate_non_empty(&options.objective, "objective")?;
validate_non_empty(&options.created_at, "created at")?;
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
if let Some(existing) = cache.get::<EpiphanyRuntimeSession>(&options.session_id)? {
if matches!(
existing.status,
EpiphanyRuntimeSessionStatus::Completed | EpiphanyRuntimeSessionStatus::Archived
) {
return Err(anyhow!(
"runtime session {:?} is terminal and cannot accept jobs",
options.session_id
));
}
return Ok(existing);
}
let session = EpiphanyRuntimeSession {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
session_id: options.session_id.clone(),
objective: options.objective,
status: EpiphanyRuntimeSessionStatus::Active,
created_at: options.created_at.clone(),
updated_at: options.created_at,
coordinator_note: options.coordinator_note,
metadata: BTreeMap::new(),
};
cache.put(&options.session_id, &session)?;
Ok(session)
}
pub fn create_runtime_job(
store_path: impl AsRef<Path>,
options: RuntimeSpineJobOptions,
) -> Result<EpiphanyRuntimeJob> {
validate_non_empty(&options.job_id, "job id")?;
validate_non_empty(&options.session_id, "session id")?;
validate_non_empty(&options.role, "role")?;
validate_non_empty(&options.created_at, "created at")?;
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
let session = cache
.get::<EpiphanyRuntimeSession>(&options.session_id)?
.ok_or_else(|| anyhow!("runtime session {:?} does not exist", options.session_id))?;
if matches!(
session.status,
EpiphanyRuntimeSessionStatus::Completed | EpiphanyRuntimeSessionStatus::Archived
) {
return Err(anyhow!(
"runtime session {:?} is not open for jobs",
options.session_id
));
}
if cache.get::<EpiphanyRuntimeJob>(&options.job_id)?.is_some() {
return Err(anyhow!("runtime job {:?} already exists", options.job_id));
}
let job = EpiphanyRuntimeJob {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
job_id: options.job_id.clone(),
session_id: options.session_id.clone(),
role: options.role,
status: EpiphanyRuntimeJobStatus::Queued,
created_at: options.created_at.clone(),
updated_at: options.created_at.clone(),
summary: options.summary,
artifact_refs: options.artifact_refs,
metadata: BTreeMap::new(),
};
cache.put(&options.job_id, &job)?;
let event = EpiphanyRuntimeEvent {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
event_id: format!("event-job-opened-{}", options.job_id),
occurred_at: options.created_at,
event_type: "job.opened".to_string(),
source: "runtime-spine".to_string(),
session_id: Some(options.session_id),
job_id: Some(options.job_id),
summary: "Native runtime job opened.".to_string(),
metadata: BTreeMap::new(),
};
cache.put(&event.event_id, &event)?;
Ok(job)
}
pub fn plan_runtime_spine_heartbeat_launch(
state: &EpiphanyThreadState,
options: RuntimeSpineHeartbeatLaunchPlanOptions,
) -> Result<RuntimeSpineHeartbeatLaunchPlan> {
validate_heartbeat_launch_options(state, &options)?;
Ok(RuntimeSpineHeartbeatLaunchPlan {
binding: EpiphanyJobBinding {
id: options.binding_id.clone(),
kind: options.kind,
scope: options.scope.clone(),
owner_role: options.owner_role.clone(),
authority_scope: Some(options.authority_scope.clone()),
linked_subgoal_ids: options.linked_subgoal_ids.clone(),
linked_graph_node_ids: options.linked_graph_node_ids.clone(),
blocking_reason: None,
},
runtime_link: EpiphanyRuntimeLink {
id: format!(
"runtime-link-{}-{}",
options.binding_id, options.runtime_job_id
),
binding_id: options.binding_id,
surface: "jobLaunch".to_string(),
role_id: options.owner_role,
authority_scope: options.authority_scope,
runtime_job_id: options.runtime_job_id,
runtime_result_id: None,
linked_subgoal_ids: options.linked_subgoal_ids,
linked_graph_node_ids: options.linked_graph_node_ids,
},
})
}
pub fn replace_or_append_epiphany_job_binding(
mut bindings: Vec<EpiphanyJobBinding>,
replacement: EpiphanyJobBinding,
) -> Vec<EpiphanyJobBinding> {
if let Some(existing) = bindings
.iter_mut()
.find(|binding| binding.id == replacement.id)
{
*existing = replacement;
return bindings;
}
bindings.push(replacement);
bindings
}
pub fn clear_epiphany_job_binding_backend(
mut bindings: Vec<EpiphanyJobBinding>,
binding_index: usize,
blocking_reason: &str,
) -> Vec<EpiphanyJobBinding> {
let binding = &mut bindings[binding_index];
binding.blocking_reason = Some(blocking_reason.to_string());
bindings
}
pub fn open_runtime_spine_heartbeat_job(
store_path: impl AsRef<Path>,
options: RuntimeSpineHeartbeatJobOptions,
) -> Result<EpiphanyRuntimeJob> {
validate_non_empty(&options.runtime_id, "runtime id")?;
validate_non_empty(&options.display_name, "display name")?;
validate_non_empty(&options.session_id, "session id")?;
validate_non_empty(&options.objective, "objective")?;
validate_non_empty(&options.job_id, "job id")?;
validate_non_empty(&options.role, "role")?;
validate_non_empty(&options.binding_id, "binding id")?;
validate_non_empty(&options.authority_scope, "authority scope")?;
validate_non_empty(&options.instruction, "instruction")?;
validate_non_empty(
options.launch_document.thread_id(),
"worker launch document thread id",
)?;
validate_non_empty(&options.output_contract_id, "output contract id")?;
if options.output_contract_id != options.launch_document.output_contract_id() {
return Err(anyhow!(
"worker launch output_contract_id must match the typed launch document"
));
}
validate_non_empty(&options.created_at, "created at")?;
let store_path = store_path.as_ref();
let job_id = options.job_id.clone();
let binding_id = options.binding_id.clone();
let role = options.role.clone();
let authority_scope = options.authority_scope.clone();
let instruction = options.instruction.clone();
let output_contract_id = options.output_contract_id.clone();
let launch_document = options.launch_document.clone();
initialize_runtime_spine(
store_path,
RuntimeSpineInitOptions {
runtime_id: options.runtime_id,
display_name: options.display_name,
created_at: options.created_at.clone(),
},
)?;
ensure_runtime_session(
store_path,
RuntimeSpineSessionOptions {
session_id: options.session_id.clone(),
objective: options.objective,
created_at: options.created_at.clone(),
coordinator_note: options.coordinator_note,
},
)?;
let job = create_runtime_job(
store_path,
RuntimeSpineJobOptions {
job_id: options.job_id,
session_id: options.session_id,
role: options.role,
created_at: options.created_at,
summary: format!(
"Heartbeat activation queued for binding {} with authority {}.",
options.binding_id, options.authority_scope
),
artifact_refs: Vec::new(),
},
)?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
if cache
.get::<EpiphanyRuntimeWorkerLaunchRequest>(&job_id)?
.is_some()
{
return Err(anyhow!(
"runtime worker launch request {:?} already exists",
job_id
));
}
let request = EpiphanyRuntimeWorkerLaunchRequest {
schema_version: RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION.to_string(),
job_id: job_id.clone(),
binding_id,
role,
authority_scope,
instruction,
output_contract_id,
document_kind: worker_launch_document_kind(&launch_document).to_string(),
launch_document_msgpack: encode_worker_launch_document(&launch_document)?,
metadata: BTreeMap::new(),
};
cache.put(&job_id, &request)?;
Ok(job)
}
pub fn runtime_job_snapshot(
store_path: impl AsRef<Path>,
job_id: &str,
) -> Result<Option<EpiphanyRuntimeJobSnapshot>> {
validate_non_empty(job_id, "job id")?;
let store_path = store_path.as_ref();
if !store_path.exists() {
return Ok(None);
}
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let Some(job) = cache.get::<EpiphanyRuntimeJob>(job_id)? else {
return Ok(None);
};
let result = cache
.get_all::<EpiphanyRuntimeJobResult>()?
.into_iter()
.filter(|result| result.job_id == job_id)
.max_by(|left, right| {
left.completed_at
.cmp(&right.completed_at)
.then_with(|| left.result_id.cmp(&right.result_id))
});
Ok(Some(EpiphanyRuntimeJobSnapshot { job, result }))
}
pub fn put_runtime_role_worker_result(
store_path: impl AsRef<Path>,
result: &EpiphanyRuntimeRoleWorkerResult,
) -> Result<()> {
validate_non_empty(&result.job_id, "role worker result job id")?;
validate_non_empty(&result.result_id, "role worker result id")?;
validate_non_empty(&result.role_id, "role worker result role id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache.put(&result.job_id, result)?;
Ok(())
}
pub fn runtime_role_worker_result(
store_path: impl AsRef<Path>,
job_id: &str,
) -> Result<Option<EpiphanyRuntimeRoleWorkerResult>> {
validate_non_empty(job_id, "role worker result job id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache.get::<EpiphanyRuntimeRoleWorkerResult>(job_id)
}
pub fn put_runtime_reorient_worker_result(
store_path: impl AsRef<Path>,
result: &EpiphanyRuntimeReorientWorkerResult,
) -> Result<()> {
validate_non_empty(&result.job_id, "reorient worker result job id")?;
validate_non_empty(&result.result_id, "reorient worker result id")?;
validate_non_empty(&result.mode, "reorient worker result mode")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache.put(&result.job_id, result)?;
Ok(())
}
pub fn runtime_reorient_worker_result(
store_path: impl AsRef<Path>,
job_id: &str,
) -> Result<Option<EpiphanyRuntimeReorientWorkerResult>> {
validate_non_empty(job_id, "reorient worker result job id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache.get::<EpiphanyRuntimeReorientWorkerResult>(job_id)
}
pub fn complete_runtime_job(
store_path: impl AsRef<Path>,
options: RuntimeSpineJobResultOptions,
) -> Result<EpiphanyRuntimeJobResult> {
validate_non_empty(&options.result_id, "result id")?;
validate_non_empty(&options.job_id, "job id")?;
validate_non_empty(&options.completed_at, "completed at")?;
validate_non_empty(&options.verdict, "verdict")?;
validate_non_empty(&options.summary, "summary")?;
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
let mut job = cache
.get::<EpiphanyRuntimeJob>(&options.job_id)?
.ok_or_else(|| anyhow!("runtime job {:?} does not exist", options.job_id))?;
if matches!(
job.status,
EpiphanyRuntimeJobStatus::Completed
| EpiphanyRuntimeJobStatus::Failed
| EpiphanyRuntimeJobStatus::Cancelled
) {
return Err(anyhow!(
"runtime job {:?} is already terminal",
options.job_id
));
}
if cache
.get::<EpiphanyRuntimeJobResult>(&options.result_id)?
.is_some()
{
return Err(anyhow!(
"runtime job result {:?} already exists",
options.result_id
));
}
let terminal_status = terminal_status_for_verdict(&options.verdict);
job.status = terminal_status;
job.updated_at = options.completed_at.clone();
job.summary = options.summary.clone();
job.artifact_refs = merge_refs(&job.artifact_refs, &options.artifact_refs);
let result = EpiphanyRuntimeJobResult {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
result_id: options.result_id.clone(),
job_id: options.job_id.clone(),
session_id: job.session_id.clone(),
role: job.role.clone(),
verdict: options.verdict,
summary: options.summary,
completed_at: options.completed_at.clone(),
next_safe_move: options.next_safe_move,
evidence_refs: options.evidence_refs,
artifact_refs: options.artifact_refs,
metadata: BTreeMap::new(),
};
cache.put(&job.job_id, &job)?;
cache.put(&result.result_id, &result)?;
let event = EpiphanyRuntimeEvent {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
event_id: format!("event-job-completed-{}", options.job_id),
occurred_at: options.completed_at,
event_type: "job.completed".to_string(),
source: "runtime-spine".to_string(),
session_id: Some(result.session_id.clone()),
job_id: Some(options.job_id),
summary: format!(
"Native runtime job completed with verdict {}.",
result.verdict
),
metadata: BTreeMap::from([("resultId".to_string(), result.result_id.clone())]),
};
cache.put(&event.event_id, &event)?;
Ok(result)
}
pub fn append_runtime_event(
store_path: impl AsRef<Path>,
options: RuntimeSpineEventOptions,
) -> Result<EpiphanyRuntimeEvent> {
validate_non_empty(&options.event_id, "event id")?;
validate_non_empty(&options.occurred_at, "occurred at")?;
validate_non_empty(&options.event_type, "event type")?;
validate_non_empty(&options.source, "source")?;
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
if cache
.get::<EpiphanyRuntimeEvent>(&options.event_id)?
.is_some()
{
return Err(anyhow!(
"runtime event {:?} already exists",
options.event_id
));
}
let event = EpiphanyRuntimeEvent {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
event_id: options.event_id.clone(),
occurred_at: options.occurred_at,
event_type: options.event_type,
source: options.source,
session_id: options.session_id,
job_id: options.job_id,
summary: options.summary,
metadata: BTreeMap::new(),
};
cache.put(&options.event_id, &event)?;
Ok(event)
}
pub fn runtime_spine_status(store_path: impl AsRef<Path>) -> Result<EpiphanyRuntimeSpineStatus> {
let store_path = store_path.as_ref();
if !store_path.exists() {
return Ok(EpiphanyRuntimeSpineStatus {
store: store_path.display().to_string(),
present: false,
runtime_id: None,
display_name: None,
sessions: 0,
active_sessions: 0,
jobs: 0,
open_jobs: 0,
job_results: 0,
events: 0,
supported_document_types: Vec::new(),
});
}
let mut cache = runtime_spine_cache(store_path)?;
cache
.pull_all_backing_stores()
.with_context(|| format!("failed to read runtime spine {}", store_path.display()))?;
let identity = cache.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?;
let sessions = cache.get_all::<EpiphanyRuntimeSession>()?;
let jobs = cache.get_all::<EpiphanyRuntimeJob>()?;
let job_results = cache.get_all::<EpiphanyRuntimeJobResult>()?;
let events = cache.get_all::<EpiphanyRuntimeEvent>()?;
let active_sessions = sessions
.iter()
.filter(|session| {
matches!(
session.status,
EpiphanyRuntimeSessionStatus::Active
| EpiphanyRuntimeSessionStatus::WaitingForReview
)
})
.count();
let open_jobs = jobs
.iter()
.filter(|job| {
matches!(
job.status,
EpiphanyRuntimeJobStatus::Queued
| EpiphanyRuntimeJobStatus::Running
| EpiphanyRuntimeJobStatus::WaitingForReview
)
})
.count();
Ok(EpiphanyRuntimeSpineStatus {
store: store_path.display().to_string(),
present: identity.is_some(),
runtime_id: identity.as_ref().map(|item| item.runtime_id.clone()),
display_name: identity.as_ref().map(|item| item.display_name.clone()),
sessions: sessions.len(),
active_sessions,
jobs: jobs.len(),
open_jobs,
job_results: job_results.len(),
events: events.len(),
supported_document_types: identity
.map(|item| item.supported_document_types)
.unwrap_or_default(),
})
}
pub fn runtime_hello_frame(store_path: impl AsRef<Path>) -> Result<Vec<u8>> {
let mut cache = runtime_spine_cache(store_path.as_ref())?;
cache.pull_all_backing_stores()?;
let identity = require_identity(&cache)?;
let message = CultNetMessage::Hello {
runtime_id: identity.runtime_id,
runtime_kind: identity.runtime_kind,
agent_id: Some("self".to_string()),
role: Some("coordinator".to_string()),
display_name: Some(identity.display_name),
supported_document_types: Some(identity.supported_document_types),
supported_mutation_contracts: Some(epiphany_mutation_contracts()),
supported_message_versions: Some(vec![
"cultnet.hello.v0".to_string(),
"cultnet.document_put.v0".to_string(),
"cultnet.snapshot_request.v0".to_string(),
"cultnet.snapshot_response.v0".to_string(),
"cultnet.schema_catalog_request.v0".to_string(),
"cultnet.schema_catalog_response.v0".to_string(),
]),
supports_schema_catalog: Some(true),
};
let payload = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?;
encode_frame(&payload)
}
pub fn write_runtime_hello_frame(
store_path: impl AsRef<Path>,
output_path: impl AsRef<Path>,
) -> Result<usize> {
let output_path = output_path.as_ref();
if let Some(parent) = output_path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
let frame = runtime_hello_frame(store_path)?;
fs::write(output_path, &frame)
.with_context(|| format!("failed to write {}", output_path.display()))?;
Ok(frame.len())
}
pub fn epiphany_schema_registry() -> Result<CultNetSchemaRegistry> {
let mut registry = builtin_schema_registry()?;
let schema_root = epiphany_schema_root();
let index_path = schema_root.join("index.json");
let raw_index = fs::read_to_string(&index_path)
.with_context(|| format!("failed to read {}", index_path.display()))?;
let index: EpiphanyCultNetSchemaIndex = serde_json::from_str(&raw_index)
.with_context(|| format!("failed to parse {}", index_path.display()))?;
if index.schema_version.trim().is_empty() {
return Err(anyhow!(
"CultNet schema index at {} is missing schemaVersion",
index_path.display()
));
}
for entry in index.schemas {
let schema_path = schema_root.join(&entry.path);
let schema_json = fs::read_to_string(&schema_path)
.with_context(|| format!("failed to read {}", schema_path.display()))?;
registry.register(CultNetSchemaRegistration {
schema_id: entry.schema_id,
kind: entry.kind,
wire_contracts: entry.wire_contracts,
schema_version: entry.schema_version,
document_type: entry.document_type,
title: entry.title,
schema_json: Some(schema_json),
})?;
}
Ok(registry)
}
pub fn runtime_schema_catalog_response(
message_id: impl Into<String>,
include_schema_json: bool,
schema_ids: Option<Vec<String>>,
kinds: Option<Vec<CultNetSchemaKind>>,
) -> Result<CultNetMessage> {
let registry = epiphany_schema_registry()?;
registry.create_catalog_response(&CultNetMessage::SchemaCatalogRequest {
message_id: message_id.into(),
include_schema_json: Some(include_schema_json),
schema_ids,
kinds,
})
}
pub fn write_runtime_schema_catalog_json(
output_path: impl AsRef<Path>,
include_schema_json: bool,
) -> Result<usize> {
let output_path = output_path.as_ref();
if let Some(parent) = output_path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
let response = runtime_schema_catalog_response(
"runtime-spine-schema-catalog".to_string(),
include_schema_json,
None,
None,
)?;
let body = serde_json::to_vec_pretty(&response)?;
fs::write(output_path, &body)
.with_context(|| format!("failed to write {}", output_path.display()))?;
Ok(body.len())
}
fn epiphany_schema_root() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.expect("epiphany-core has no parent repo root")
.join(CULTNET_SCHEMA_INDEX_RELATIVE)
.parent()
.expect("cultnet schema index has no parent directory")
.to_path_buf()
}
fn require_identity(cache: &CultCache) -> Result<EpiphanyRuntimeIdentity> {
cache
.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("runtime spine is missing identity; run init first"))
}
fn supported_runtime_document_types() -> Vec<String> {
let mut document_types = Vec::new();
for contract in epiphany_mutation_contracts() {
if !document_types.contains(&contract.document_type) {
document_types.push(contract.document_type);
}
}
document_types
}
fn mutation_contract(
document_type: impl Into<String>,
payload_schema_version: impl Into<String>,
operations: Vec<CultNetDocumentOperation>,
authority: CultNetMutationAuthority,
intent_document_types: Vec<&str>,
receipt_document_types: Vec<&str>,
notes: Vec<&str>,
) -> CultNetDocumentMutationContract {
CultNetDocumentMutationContract {
document_type: document_type.into(),
payload_schema_version: Some(payload_schema_version.into()),
operations,
authority,
intent_document_types: (!intent_document_types.is_empty()).then(|| {
intent_document_types
.into_iter()
.map(str::to_string)
.collect()
}),
receipt_document_types: (!receipt_document_types.is_empty()).then(|| {
receipt_document_types
.into_iter()
.map(str::to_string)
.collect()
}),
notes: (!notes.is_empty()).then(|| notes.into_iter().map(str::to_string).collect()),
}
}
fn read_only_surface_contract(
document_type: impl Into<String>,
payload_schema_version: impl Into<String>,
notes: Vec<&str>,
) -> CultNetDocumentMutationContract {
mutation_contract(
document_type,
payload_schema_version,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
notes,
)
}
fn coordinator_surface_contract(
document_type: impl Into<String>,
payload_schema_version: impl Into<String>,
intent_document_types: Vec<&str>,
receipt_document_types: Vec<&str>,
notes: Vec<&str>,
) -> CultNetDocumentMutationContract {
mutation_contract(
document_type,
payload_schema_version,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
intent_document_types,
receipt_document_types,
notes,
)
}
fn epiphany_mutation_contracts() -> Vec<CultNetDocumentMutationContract> {
vec![
mutation_contract(
RUNTIME_IDENTITY_TYPE,
RUNTIME_SPINE_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec!["Runtime identity is advertised by the coordinator, not remotely mutated."],
),
mutation_contract(
RUNTIME_SESSION_TYPE,
RUNTIME_SPINE_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec!["epiphany.runtime.session_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec!["Sessions change through coordinator-reviewed typed intents."],
),
mutation_contract(
RUNTIME_JOB_TYPE,
RUNTIME_SPINE_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec!["epiphany.heartbeat_pump_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Heartbeat activation owns agent work; external callers submit intents and watch receipts.",
],
),
mutation_contract(
RUNTIME_WORKER_LAUNCH_REQUEST_TYPE,
RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec![RUNTIME_WORKER_LAUNCH_REQUEST_TYPE],
vec![RUNTIME_JOB_TYPE],
vec![
"Worker launch requests are typed task-intent documents; runtime jobs are lifecycle receipts, not the source of work intent.",
"Core/coordinator policy owns the launch yes/no; transport bridges translate, gather host facts, execute side effects, and return receipts.",
],
),
mutation_contract(
RUNTIME_ROLE_WORKER_RESULT_TYPE,
RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"Role worker results preserve the typed finding payload; generic runtime job results are lifecycle receipts.",
],
),
mutation_contract(
RUNTIME_REORIENT_WORKER_RESULT_TYPE,
RUNTIME_REORIENT_WORKER_RESULT_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"Reorient worker results preserve continuity findings separately from generic runtime lifecycle receipts.",
],
),
mutation_contract(
RUNTIME_JOB_RESULT_TYPE,
RUNTIME_SPINE_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"Job results are evidence records; review and acceptance are separate typed flows.",
],
),
mutation_contract(
RUNTIME_EVENT_TYPE,
RUNTIME_SPINE_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec!["Runtime events are append-only projections for inspection."],
),
mutation_contract(
OPENAI_ADAPTER_STATUS_TYPE,
OPENAI_ADAPTER_STATUS_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"OpenAI adapter status is a typed document; it is not a Codex app-server status blob.",
],
),
mutation_contract(
OPENAI_MODEL_REQUEST_TYPE,
OPENAI_MODEL_REQUEST_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec![OPENAI_MODEL_REQUEST_TYPE],
vec![OPENAI_MODEL_STREAM_EVENT_TYPE, OPENAI_MODEL_RECEIPT_TYPE],
vec![
"Model turns enter through typed Epiphany request documents and return typed stream events/receipts.",
"The Codex spine may authenticate and transport; it must not own Epiphany state or scheduling.",
],
),
mutation_contract(
OPENAI_MODEL_STREAM_EVENT_TYPE,
OPENAI_MODEL_STREAM_EVENT_SCHEMA_VERSION,
vec![CultNetDocumentOperation::ReceiptWatch],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec!["OpenAI stream events are receipts from a typed model request."],
),
mutation_contract(
OPENAI_MODEL_RECEIPT_TYPE,
OPENAI_MODEL_RECEIPT_SCHEMA_VERSION,
vec![CultNetDocumentOperation::ReceiptWatch],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec!["Terminal model receipts carry response id, usage, and transport evidence."],
),
mutation_contract(
AGENT_MEMORY_TYPE,
AGENT_MEMORY_PAYLOAD_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec!["epiphany.agent_memory_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Sub-agents request memory mutations; the coordinator accepts, rejects, or explains the refusal.",
],
),
mutation_contract(
HEARTBEAT_STATE_TYPE,
HEARTBEAT_STATE_SCHEMA_VERSION,
vec![
CultNetDocumentOperation::Snapshot,
CultNetDocumentOperation::IntentSubmit,
CultNetDocumentOperation::ReceiptWatch,
],
CultNetMutationAuthority::Coordinator,
vec![
"epiphany.heartbeat_pump_intent.v0",
"epiphany.heartbeat_heat_intent.v0",
"epiphany.circadian_rhythm_intent.v0",
],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Aquarium controls heartbeat and circadian rhythm through typed intents, not blind state replacement.",
"Initiative heat is heartbeat policy: global, group, role, and agent tempo changes enter through the heartbeat heat intent.",
],
),
mutation_contract(
STATE_LEDGER_STORE_TYPE,
STATE_LEDGER_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"The ledger is inspected as durable memory; writes are mediated by role-specific state flows.",
],
),
mutation_contract(
THREAD_STATE_TYPE,
THREAD_STATE_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"The mirrored thread state is the typed repo/control-plane state source; Codex rollout is a compatibility source, not the network contract.",
],
),
mutation_contract(
MEMORY_GRAPH_TYPE,
MEMORY_GRAPH_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"The unified memory graph is typed durable state; Qdrant embeddings are rebuildable cache, not canonical memory.",
],
),
read_only_surface_contract(
SURFACE_SCENE_TYPE,
SCENE_SURFACE_SCHEMA_VERSION,
vec![
"Operator-safe scene reflection over typed Epiphany state.",
"Aquarium should read this before offering live coordination actions.",
],
),
read_only_surface_contract(
SURFACE_FRESHNESS_TYPE,
FRESHNESS_SURFACE_SCHEMA_VERSION,
vec![
"Freshness reflection over retrieval, watcher, and graph staleness signals.",
"Use this to visualize retrieval and graph staleness without inventing a hidden refresh daemon.",
],
),
read_only_surface_contract(
SURFACE_CONTEXT_TYPE,
CONTEXT_SURFACE_SCHEMA_VERSION,
vec![
"Targeted graph, frontier, checkpoint, observation, and evidence context shard over typed Epiphany state.",
"Aquarium should inspect bounded state shards here instead of scraping state blobs by superstition.",
],
),
read_only_surface_contract(
SURFACE_GRAPH_QUERY_TYPE,
GRAPH_QUERY_SURFACE_SCHEMA_VERSION,
vec![
"Bounded graph traversal over typed architecture/dataflow graph state.",
"Use this for architecture/dataflow inspection and frontier neighborhoods without mutating state.",
],
),
read_only_surface_contract(
SURFACE_PRESSURE_TYPE,
PRESSURE_SURFACE_SCHEMA_VERSION,
vec![
"Current context pressure and compaction posture derived from typed pressure inputs.",
"This is a read-only warning surface, not a backdoor to force state mutation.",
],
),
coordinator_surface_contract(
SURFACE_REORIENT_TYPE,
REORIENT_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.reorient_launch_intent.v0",
"epiphany.reorient_accept_intent.v0",
],
vec![
"epiphany.swarm_control_receipt.v0",
"epiphany.reorient_result_surface.v0",
],
vec![
"Reorientation policy is the typed resume/regather verdict surface.",
"Launch and acceptance stay review-gated through explicit typed intents.",
"The bridge may carry a reorient launch intent and report receipts, but the resume/regather verdict belongs to core.",
],
),
read_only_surface_contract(
SURFACE_CRRC_TYPE,
CRRC_SURFACE_SCHEMA_VERSION,
vec![
"CRRC recommendation surface over continuity, pressure, and reorientation signals.",
"Use this to understand continuity pressure without letting CRRC seize authority.",
],
),
coordinator_surface_contract(
SURFACE_JOBS_TYPE,
JOBS_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.job_launch_intent.v0",
"epiphany.job_interrupt_intent.v0",
],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Job reflection over typed job bindings and runtime-spine lifecycle receipts.",
"Heartbeat/runtime-spine owns activation; callers submit typed intents and watch receipts.",
"Launch receipts must name decisionOwner and bridgeRole so transport never grows an unreadable second opinion.",
],
),
coordinator_surface_contract(
SURFACE_ROLES_TYPE,
ROLES_SURFACE_SCHEMA_VERSION,
vec!["epiphany.role_launch_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Role ownership surface for fixed Epiphany lanes and launch affordances.",
"Treat this as the discoverable lane catalog for Aquarium.",
],
),
coordinator_surface_contract(
SURFACE_ROLE_RESULT_TYPE,
ROLE_RESULT_SURFACE_SCHEMA_VERSION,
vec!["epiphany.role_accept_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Role findings are typed review surfaces accepted through explicit role acceptance intents.",
"Semantic findings remain explicitly review-gated.",
],
),
coordinator_surface_contract(
SURFACE_REORIENT_RESULT_TYPE,
REORIENT_RESULT_SURFACE_SCHEMA_VERSION,
vec!["epiphany.reorient_accept_intent.v0"],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Completed reorientation findings are typed review surfaces accepted through explicit reorientation acceptance intents.",
],
),
coordinator_surface_contract(
SURFACE_PLANNING_TYPE,
PLANNING_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.planning_update_intent.v0",
"epiphany.objective_adoption_intent.v0",
],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Planning projection over captures, backlog, roadmap streams, and Objective Drafts.",
"Backlog, captures, and Objective Drafts are planning state until explicit adoption.",
],
),
read_only_surface_contract(
SURFACE_COORDINATOR_TYPE,
COORDINATOR_SURFACE_SCHEMA_VERSION,
vec![
"Fixed-lane recommendation surface derived from typed role, pressure, reorientation, and result signals.",
"Aquarium should treat this as the primary action oracle, not invent its own scheduler.",
],
),
coordinator_surface_contract(
SURFACE_FACE_TYPE,
FACE_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.face_bubble_intent.v0",
"epiphany.character_turn_intent.v0",
"epiphany.discord_persona_post_intent.v0",
],
vec![
"epiphany.face_bubble.v0",
"epiphany.face_chat.v0",
"epiphany.character_turn_packet.v0",
],
vec![
"Face bubble, draft, and Discord persona affordances are projected from typed Face and character-loop artifacts.",
"Humans talk to Face; sealed inner thoughts stay behind the projection boundary.",
],
),
read_only_surface_contract(
SURFACE_VOID_MEMORY_TYPE,
VOID_MEMORY_SURFACE_SCHEMA_VERSION,
vec![
"Void-derived memory status/search/context availability is projected from the typed Void memory bridge.",
"This is an inspection surface for the memory organs, not a license to bypass typed Epiphany state.",
],
),
coordinator_surface_contract(
SURFACE_REPO_INITIALIZATION_TYPE,
REPO_INITIALIZATION_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.repo_startup_intent.v0",
"epiphany.repo_initialization_accept_intent.v0",
],
vec![
"epiphany.repo_personality_artifacts.v0",
"epiphany.repo_initialization_record.v0",
],
vec![
"Repo birth/startup status is projected from typed repo-personality startup and accept-init receipts.",
"Birth specialists are startup-only and remain outside the heartbeat lane system.",
],
),
coordinator_surface_contract(
SURFACE_REPO_BIRTH_RUNNER_TYPE,
REPO_BIRTH_RUNNER_SURFACE_SCHEMA_VERSION,
vec!["epiphany.repo_birth_run_intent.v0"],
vec![
"epiphany.repo_birth_runner.v0",
"epiphany.repo_initialization_record.v0",
],
vec![
"Startup-only birth runner plan/run affordances are projected from typed birth-runner receipts.",
"Aquarium should review birth artifacts and accept them explicitly instead of growing a hidden wizard.",
],
),
coordinator_surface_contract(
SURFACE_RIDER_BRIDGE_TYPE,
RIDER_BRIDGE_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.rider_status_intent.v0",
"epiphany.rider_context_intent.v0",
"epiphany.rider_open_ref_intent.v0",
],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Rider bridge affordances are projected from typed Rider bridge artifacts.",
"Use this surface for source/IDE inspection instead of scraping local artifacts by convention.",
],
),
coordinator_surface_contract(
SURFACE_UNITY_BRIDGE_TYPE,
UNITY_BRIDGE_SURFACE_SCHEMA_VERSION,
vec![
"epiphany.unity_inspect_intent.v0",
"epiphany.unity_probe_intent.v0",
"epiphany.unity_check_compilation_intent.v0",
"epiphany.unity_run_tests_intent.v0",
],
vec!["epiphany.swarm_control_receipt.v0"],
vec![
"Unity bridge affordances are projected from typed Unity bridge artifacts.",
"Runtime/editor truth must come through the pinned bridge and its receipts, not wishful source reading.",
],
),
]
}
fn validate_non_empty(value: &str, field: &str) -> Result<()> {
if value.trim().is_empty() {
return Err(anyhow!("{field} must be non-empty"));
}
Ok(())
}
fn worker_launch_document_kind(document: &EpiphanyWorkerLaunchDocument) -> &'static str {
match document {
EpiphanyWorkerLaunchDocument::Role(_) => "role",
EpiphanyWorkerLaunchDocument::Reorient(_) => "reorient",
}
}
fn encode_worker_launch_document(document: &EpiphanyWorkerLaunchDocument) -> Result<Vec<u8>> {
rmp_serde::to_vec_named(document).context("failed to encode worker launch document MessagePack")
}
fn decode_optional_msgpack<T>(payload: Option<&[u8]>, label: &str) -> Result<Option<T>>
where
T: for<'de> Deserialize<'de>,
{
payload
.map(|payload| {
rmp_serde::from_slice(payload).with_context(|| format!("failed to decode {label}"))
})
.transpose()
}
fn validate_heartbeat_launch_options(
state: &EpiphanyThreadState,
options: &RuntimeSpineHeartbeatLaunchPlanOptions,
) -> Result<()> {
validate_non_empty(&options.binding_id, "epiphany job launch binding_id")?;
if matches!(
options.binding_id.as_str(),
"retrieval-index" | "graph-remap" | "verification"
) {
return Err(anyhow!(
"epiphany job launch binding_id {:?} is reserved for a derived built-in slot",
options.binding_id
));
}
if options.kind != EpiphanyJobKind::Specialist {
return Err(anyhow!(
"epiphany job launch currently supports only specialist heartbeat turns"
));
}
validate_non_empty(&options.scope, "epiphany job launch scope")?;
validate_non_empty(&options.owner_role, "epiphany job launch owner_role")?;
validate_non_empty(
&options.authority_scope,
"epiphany job launch authority_scope",
)?;
validate_non_empty(&options.instruction, "epiphany job launch instruction")?;
validate_non_empty(
options.launch_document.thread_id(),
"epiphany job launch document thread id",
)?;
validate_non_empty(
&options.output_contract_id,
"epiphany job launch output_contract_id",
)?;
if options.output_contract_id != options.launch_document.output_contract_id() {
return Err(anyhow!(
"epiphany job launch output_contract_id must match the typed launch document"
));
}
if let Some(max_runtime_seconds) = options.max_runtime_seconds
&& max_runtime_seconds == 0
{
return Err(anyhow!(
"epiphany job launch max_runtime_seconds must be >= 1"
));
}
let existing_binding = state
.job_bindings
.iter()
.find(|binding| binding.id == options.binding_id);
let existing_runtime_link = state.runtime_links.iter().find(|link| {
link.binding_id == options.binding_id && !link.runtime_job_id.trim().is_empty()
});
if existing_runtime_link.is_some()
&& existing_binding.is_none_or(|binding| binding.blocking_reason.is_none())
{
return Err(anyhow!(
"epiphany job binding {:?} is already bound to an active heartbeat turn; interrupt it before launching a replacement",
options.binding_id
));
}
Ok(())
}
fn terminal_status_for_verdict(verdict: &str) -> EpiphanyRuntimeJobStatus {
if matches!(
verdict,
"failed" | "fail" | "error" | "blocked" | "cancelled" | "canceled"
) {
EpiphanyRuntimeJobStatus::Failed
} else {
EpiphanyRuntimeJobStatus::Completed
}
}
fn merge_refs(existing: &[String], incoming: &[String]) -> Vec<String> {
let mut merged = existing.to_vec();
for item in incoming {
if !merged.contains(item) {
merged.push(item.clone());
}
}
merged
}
#[cfg(test)]
mod tests {
use super::*;
use cultnet_rs::CultNetWireContract;
use cultnet_rs::decode_cultnet_message_from_slice;
use tempfile::tempdir;
#[test]
fn runtime_spine_initializes_sessions_events_and_status() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "epiphany-test".to_string(),
display_name: "Epiphany Test".to_string(),
created_at: "2026-05-06T00:00:00Z".to_string(),
},
)?;
create_runtime_session(
&store,
RuntimeSpineSessionOptions {
session_id: "session-1".to_string(),
objective: "Build the spine.".to_string(),
created_at: "2026-05-06T00:01:00Z".to_string(),
coordinator_note: "Native first.".to_string(),
},
)?;
append_runtime_event(
&store,
RuntimeSpineEventOptions {
event_id: "event-1".to_string(),
occurred_at: "2026-05-06T00:02:00Z".to_string(),
event_type: "session.started".to_string(),
source: "test".to_string(),
session_id: Some("session-1".to_string()),
job_id: None,
summary: "Session started.".to_string(),
},
)?;
let status = runtime_spine_status(&store)?;
assert!(status.present);
assert_eq!(status.runtime_id.as_deref(), Some("epiphany-test"));
assert_eq!(status.sessions, 1);
assert_eq!(status.active_sessions, 1);
assert_eq!(status.events, 1);
Ok(())
}
#[test]
fn runtime_spine_opens_and_completes_native_jobs() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "epiphany-test".to_string(),
display_name: "Epiphany Test".to_string(),
created_at: "2026-05-06T00:00:00Z".to_string(),
},
)?;
create_runtime_session(
&store,
RuntimeSpineSessionOptions {
session_id: "session-1".to_string(),
objective: "Build the job artery.".to_string(),
created_at: "2026-05-06T00:01:00Z".to_string(),
coordinator_note: "Native jobs.".to_string(),
},
)?;
let job = create_runtime_job(
&store,
RuntimeSpineJobOptions {
job_id: "job-1".to_string(),
session_id: "session-1".to_string(),
role: "modeling".to_string(),
created_at: "2026-05-06T00:02:00Z".to_string(),
summary: "Model the target.".to_string(),
artifact_refs: vec!["artifact:modeling-plan".to_string()],
},
)?;
assert_eq!(job.status, EpiphanyRuntimeJobStatus::Queued);
let result = complete_runtime_job(
&store,
RuntimeSpineJobResultOptions {
result_id: "result-1".to_string(),
job_id: "job-1".to_string(),
completed_at: "2026-05-06T00:03:00Z".to_string(),
verdict: "pass".to_string(),
summary: "Model is ready.".to_string(),
next_safe_move: "Launch verification.".to_string(),
evidence_refs: vec!["evidence:model".to_string()],
artifact_refs: vec!["artifact:model".to_string()],
},
)?;
assert_eq!(result.role, "modeling");
let status = runtime_spine_status(&store)?;
assert_eq!(status.jobs, 1);
assert_eq!(status.open_jobs, 0);
assert_eq!(status.job_results, 1);
assert_eq!(status.events, 2);
let snapshot =
runtime_job_snapshot(&store, "job-1")?.expect("completed job snapshot should exist");
assert_eq!(snapshot.job.status, EpiphanyRuntimeJobStatus::Completed);
assert_eq!(
snapshot
.result
.as_ref()
.map(|result| result.result_id.as_str()),
Some("result-1")
);
Ok(())
}
#[test]
fn runtime_spine_opens_heartbeat_job_from_single_typed_call() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
let job = open_runtime_spine_heartbeat_job(
&store,
RuntimeSpineHeartbeatJobOptions {
runtime_id: "epiphany-test".to_string(),
display_name: "Epiphany Test".to_string(),
session_id: "epiphany-main".to_string(),
objective: "Run heartbeat worker.".to_string(),
coordinator_note: "Test launch.".to_string(),
job_id: "heartbeat-job-1".to_string(),
role: "modeling".to_string(),
binding_id: "modeling-checkpoint-worker".to_string(),
authority_scope: "modeling".to_string(),
instruction: "Inspect the checkpoint and return typed role findings.".to_string(),
launch_document: EpiphanyWorkerLaunchDocument::Role(
crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-1".to_string(),
role_id: "modeling".to_string(),
state_revision: 7,
objective: Some("Run heartbeat worker.".to_string()),
active_subgoal_id: None,
active_subgoals: Vec::new(),
active_graph_node_ids: vec!["node-model".to_string()],
investigation_checkpoint: None,
scratch: None,
invariants: Vec::new(),
graphs: None,
recent_evidence: Vec::new(),
recent_observations: Vec::new(),
graph_frontier: None,
graph_checkpoint: None,
planning: None,
churn: None,
},
),
output_contract_id: crate::ROLE_WORKER_OUTPUT_CONTRACT_ID.to_string(),
created_at: "2026-05-06T00:02:00Z".to_string(),
},
)?;
assert_eq!(job.job_id, "heartbeat-job-1");
assert_eq!(job.session_id, "epiphany-main");
assert_eq!(job.role, "modeling");
let status = runtime_spine_status(&store)?;
assert_eq!(status.sessions, 1);
assert_eq!(status.jobs, 1);
assert_eq!(status.open_jobs, 1);
assert_eq!(status.events, 1);
let mut cache = runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
let launch_request = cache
.get::<EpiphanyRuntimeWorkerLaunchRequest>("heartbeat-job-1")?
.expect("typed worker launch request should be durable");
assert_eq!(launch_request.job_id, "heartbeat-job-1");
assert_eq!(launch_request.binding_id, "modeling-checkpoint-worker");
assert_eq!(launch_request.role, "modeling");
assert_eq!(
launch_request.output_contract_id,
crate::ROLE_WORKER_OUTPUT_CONTRACT_ID
);
assert_eq!(launch_request.document_kind, "role");
assert!(!launch_request.launch_document_msgpack.is_empty());
assert_eq!(launch_request.launch_document()?.thread_id(), "thread-1");
Ok(())
}
#[test]
fn heartbeat_launch_plan_leaves_lifecycle_to_runtime_links() {
let launch_plan = plan_runtime_spine_heartbeat_launch(
&EpiphanyThreadState::default(),
RuntimeSpineHeartbeatLaunchPlanOptions {
binding_id: "modeling-checkpoint-worker".to_string(),
kind: EpiphanyJobKind::Specialist,
scope: "role-scoped modeling/checkpoint maintenance".to_string(),
owner_role: "epiphany-modeler".to_string(),
authority_scope: "epiphany.role.modeling".to_string(),
linked_subgoal_ids: vec!["phase-6".to_string()],
linked_graph_node_ids: vec!["runtime-spine".to_string()],
instruction: "Model the target before implementation.".to_string(),
launch_document: EpiphanyWorkerLaunchDocument::Role(
crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-1".to_string(),
role_id: "modeling".to_string(),
state_revision: 7,
objective: Some("keep state typed".to_string()),
active_subgoal_id: None,
active_subgoals: Vec::new(),
active_graph_node_ids: vec!["runtime-spine".to_string()],
investigation_checkpoint: None,
scratch: None,
invariants: Vec::new(),
graphs: None,
recent_evidence: Vec::new(),
recent_observations: Vec::new(),
graph_frontier: None,
graph_checkpoint: None,
planning: None,
churn: None,
},
),
output_contract_id: "epiphany.worker.role_result.v0".to_string(),
max_runtime_seconds: Some(60),
runtime_job_id: "turn-1".to_string(),
},
)
.expect("launch planning should build binding and runtime link");
assert_eq!(
launch_plan.binding.authority_scope.as_deref(),
Some("epiphany.role.modeling")
);
assert_eq!(
launch_plan.runtime_link.id,
"runtime-link-modeling-checkpoint-worker-turn-1"
);
assert_eq!(launch_plan.runtime_link.runtime_job_id, "turn-1");
assert_eq!(launch_plan.runtime_link.runtime_result_id, None);
assert_eq!(launch_plan.runtime_link.role_id, "epiphany-modeler");
}
#[test]
fn runtime_spine_emits_cultnet_hello_frame() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "epiphany-test".to_string(),
display_name: "Epiphany Test".to_string(),
created_at: "2026-05-06T00:00:00Z".to_string(),
},
)?;
let frame = runtime_hello_frame(&store)?;
let payload_len = u32::from_be_bytes([frame[0], frame[1], frame[2], frame[3]]) as usize;
let message =
decode_cultnet_message_from_slice(&frame[4..], CultNetWireContract::CultNetSchemaV0)?;
assert_eq!(payload_len, frame.len() - 4);
match message {
CultNetMessage::Hello {
runtime_id,
runtime_kind,
supported_document_types,
supported_mutation_contracts,
..
} => {
assert_eq!(runtime_id, "epiphany-test");
assert_eq!(runtime_kind, "epiphany.native");
assert!(
supported_document_types
.unwrap()
.contains(&RUNTIME_JOB_RESULT_TYPE.to_string())
);
let contracts = supported_mutation_contracts.unwrap();
let heartbeat_contract = contracts
.iter()
.find(|contract| contract.document_type == HEARTBEAT_STATE_TYPE)
.expect("heartbeat state should advertise mutation contract");
assert_eq!(
heartbeat_contract.authority,
CultNetMutationAuthority::Coordinator
);
assert!(
heartbeat_contract
.operations
.contains(&CultNetDocumentOperation::IntentSubmit)
);
assert!(
heartbeat_contract
.intent_document_types
.as_ref()
.is_some_and(|items| items
.iter()
.any(|item| item == "epiphany.heartbeat_heat_intent.v0"))
);
let coordinator_contract = contracts
.iter()
.find(|contract| contract.document_type == SURFACE_COORDINATOR_TYPE)
.expect("coordinator surface should advertise a read-only contract");
assert_eq!(
coordinator_contract.authority,
CultNetMutationAuthority::ReadOnly
);
let face_contract = contracts
.iter()
.find(|contract| contract.document_type == SURFACE_FACE_TYPE)
.expect("face surface should advertise an interactive contract");
assert_eq!(
face_contract.authority,
CultNetMutationAuthority::Coordinator
);
assert!(
face_contract
.receipt_document_types
.as_ref()
.is_some_and(|items| items
.iter()
.any(|item| item == "epiphany.face_bubble.v0"))
);
let unity_contract = contracts
.iter()
.find(|contract| contract.document_type == SURFACE_UNITY_BRIDGE_TYPE)
.expect("unity bridge surface should advertise an interactive contract");
assert!(
unity_contract
.intent_document_types
.as_ref()
.is_some_and(|items| items
.iter()
.any(|item| item == "epiphany.unity_probe_intent.v0"))
);
let birth_contract = contracts
.iter()
.find(|contract| contract.document_type == SURFACE_REPO_BIRTH_RUNNER_TYPE)
.expect("repo birth runner surface should advertise startup actions");
assert!(
birth_contract
.receipt_document_types
.as_ref()
.is_some_and(|items| items
.iter()
.any(|item| item == "epiphany.repo_birth_runner.v0"))
);
let openai_contract = contracts
.iter()
.find(|contract| contract.document_type == OPENAI_MODEL_REQUEST_TYPE)
.expect("OpenAI adapter should advertise typed model request contract");
assert_eq!(
openai_contract.authority,
CultNetMutationAuthority::Coordinator
);
assert!(
openai_contract
.receipt_document_types
.as_ref()
.is_some_and(|items| items
.iter()
.any(|item| item == OPENAI_MODEL_RECEIPT_TYPE))
);
let memory_graph_contract = contracts
.iter()
.find(|contract| contract.document_type == MEMORY_GRAPH_TYPE)
.expect("memory graph should advertise a read-only contract");
assert_eq!(
memory_graph_contract.authority,
CultNetMutationAuthority::ReadOnly
);
let thread_state_contract = contracts
.iter()
.find(|contract| contract.document_type == THREAD_STATE_TYPE)
.expect("thread state should advertise a read-only contract");
assert_eq!(
thread_state_contract.authority,
CultNetMutationAuthority::ReadOnly
);
}
other => panic!("expected hello, got {other:?}"),
}
Ok(())
}
#[test]
fn runtime_spine_schema_catalog_includes_surface_and_control_receipts() -> Result<()> {
let registry = epiphany_schema_registry()?;
let schemas = registry.list(&cultnet_rs::CultNetSchemaCatalogOptions {
include_schema_json: false,
schema_ids: None,
kinds: None,
});
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(SURFACE_SCENE_TYPE)
&& schema.schema_version.as_deref() == Some(SCENE_SURFACE_SCHEMA_VERSION)
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(SURFACE_COORDINATOR_TYPE)
&& schema.schema_version.as_deref() == Some(COORDINATOR_SURFACE_SCHEMA_VERSION)
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some("epiphany.role_launch_intent.v0")
&& schema.schema_version.as_deref() == Some("epiphany.role_launch_intent.v0")
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some("epiphany.heartbeat_initiative_heat.v0")
&& schema.schema_version.as_deref() == Some("epiphany.heartbeat_initiative_heat.v0")
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some("epiphany.heartbeat_heat_intent.v0")
&& schema.schema_version.as_deref() == Some("epiphany.heartbeat_heat_intent.v0")
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some("epiphany.swarm_control_receipt.v0")
&& schema.schema_version.as_deref() == Some("epiphany.swarm_control_receipt.v0")
}));
let receipt_schema_path =
epiphany_schema_root().join("epiphany.swarm-control-receipt.schema.json");
let receipt_schema: serde_json::Value =
serde_json::from_str(&fs::read_to_string(receipt_schema_path)?)?;
let required = receipt_schema["required"]
.as_array()
.expect("receipt schema should list required fields");
assert!(required.iter().any(|field| field == "decisionOwner"));
assert!(required.iter().any(|field| field == "bridgeRole"));
assert!(
receipt_schema["properties"]["decisionOwner"]["description"]
.as_str()
.is_some_and(|description| description.contains("epiphany-core"))
);
assert!(
receipt_schema["properties"]["bridgeRole"]["description"]
.as_str()
.is_some_and(|description| description.contains("transport"))
);
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(OPENAI_MODEL_REQUEST_TYPE)
&& schema.schema_version.as_deref() == Some(OPENAI_MODEL_REQUEST_SCHEMA_VERSION)
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(OPENAI_MODEL_RECEIPT_TYPE)
&& schema.schema_version.as_deref() == Some(OPENAI_MODEL_RECEIPT_SCHEMA_VERSION)
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(MEMORY_GRAPH_TYPE)
&& schema.schema_version.as_deref() == Some(MEMORY_GRAPH_SCHEMA_VERSION)
}));
assert!(schemas.iter().any(|schema| {
schema.document_type.as_deref() == Some(THREAD_STATE_TYPE)
&& schema.schema_version.as_deref() == Some(THREAD_STATE_SCHEMA_VERSION)
}));
Ok(())
}
}
Reference in a new issue View git blame Copy permalink