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-openai-runtime/src/lib.rs
2026-08-24 11:43:37 +02:00

3971 lines
168 KiB
Rust

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use chrono::SecondsFormat;
use epiphany_core::EpiphanyRuntimeReorientWorkerResult;
use epiphany_core::EpiphanyRuntimeRoleWorkerResult;
use epiphany_core::EpiphanyRuntimeWorkerLaunchRequest;
use epiphany_core::EpiphanyWorkerLaunchDocument;
use epiphany_core::RuntimeSpineInitOptions;
use epiphany_core::RuntimeSpineJobOptions;
use epiphany_core::RuntimeSpineJobResultOptions;
use epiphany_core::RuntimeSpineSessionClosureOptions;
use epiphany_core::RuntimeSpineSessionOptions;
use epiphany_core::close_runtime_session;
use epiphany_core::complete_runtime_job;
use epiphany_core::initialize_runtime_spine;
use epiphany_core::open_runtime_model_execution;
use epiphany_core::put_runtime_reorient_worker_result;
use epiphany_core::put_runtime_role_worker_result;
use epiphany_core::put_runtime_tool_execution_intent;
use epiphany_core::runtime_identity;
use epiphany_core::runtime_spine_cache;
use epiphany_model_adapter::EpiphanyModelInputItem;
use epiphany_model_adapter::EpiphanyModelReceipt;
use epiphany_model_adapter::EpiphanyModelRequest;
use epiphany_model_adapter::EpiphanyModelStreamEvent;
use epiphany_model_adapter::EpiphanyModelStreamPayload;
use epiphany_model_adapter::EpiphanyModelToolDefinition;
use epiphany_openai_adapter::EpiphanyOpenAiModelRequest;
use epiphany_openai_adapter::EpiphanyOpenAiStreamEvent;
use epiphany_openai_adapter::EpiphanyOpenAiStreamPayload;
use epiphany_openai_codex_spine::EpiphanyCodexOpenAiTransport;
use epiphany_openai_codex_spine::EpiphanyOpenRouterTransport;
use epiphany_openai_codex_spine::auth_manager;
pub use epiphany_openai_codex_spine::default_codex_home;
use epiphany_tool_adapter::EPIPHANY_TOOL_RUNTIME_ADAPTER_ID;
use epiphany_tool_adapter::EpiphanyToolInvocationIntent;
use epiphany_tool_adapter::EpiphanyToolInvocationReceipt;
use epiphany_tool_adapter::receipt_output_for_model;
use epiphany_tool_adapter::tool_invocation_receipt_key;
use serde::Deserialize;
use serde::de::DeserializeOwned;
use serde::de::MapAccess;
use serde::de::SeqAccess;
use serde::de::Visitor;
use sha2::Digest;
mod persona_executor;
pub use persona_executor::*;
pub const OPENAI_RUNTIME_ROLE: &str = "openai-model-adapter";
pub const DEFAULT_PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(90);
pub const DEFAULT_MODEL_PROVIDER: &str = "openai-codex";
pub const OPENROUTER_MODEL_PROVIDER: &str = "openrouter";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EpiphanyProviderProfile {
OpenAiCodex,
OpenRouter,
}
impl EpiphanyProviderProfile {
fn parse(provider: &str) -> Result<Self> {
match provider {
"openai-codex" | "openai" => Ok(Self::OpenAiCodex),
OPENROUTER_MODEL_PROVIDER => Ok(Self::OpenRouter),
_ => Err(anyhow!(
"unsupported model runtime provider {provider:?}; current providers: openai-codex, openrouter"
)),
}
}
}
pub fn worker_model_session_id(worker_job_id: &str) -> String {
format!("openai-worker-session-{worker_job_id}")
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EpiphanyOpenAiRuntimeOptions {
pub store_path: PathBuf,
pub codex_home: PathBuf,
pub provider_credential_path: Option<PathBuf>,
pub session_id: String,
pub job_id: String,
pub objective: String,
pub coordinator_note: String,
pub default_model: Option<String>,
/// Whole-request transport bound. Worker passes set this to `None` when
/// their outer typed runtime budget is the sole deadline owner.
pub request_timeout: Option<Duration>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EpiphanyOpenAiRuntimeRunSummary {
pub store: String,
pub session_id: String,
pub job_id: String,
pub request_id: String,
pub event_count: usize,
pub verdict: String,
pub summary: String,
pub result_id: String,
pub receipt_id: Option<String>,
pub tool_intent_ids: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EpiphanyWorkerRuntimeOptions {
pub store_path: PathBuf,
pub codex_home: PathBuf,
pub provider_credential_path: Option<PathBuf>,
pub provider: String,
pub job_id: String,
pub model: String,
pub request_timeout: Option<Duration>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EpiphanyWorkerRuntimeRunSummary {
pub store: String,
pub job_id: String,
pub binding_id: String,
pub role: String,
pub request_id: String,
pub openai_result_id: String,
pub worker_result_id: String,
pub verdict: String,
pub summary: String,
pub next_safe_move: String,
pub evidence_refs: Vec<String>,
pub artifact_refs: Vec<String>,
}
async fn run_openai_model_turn_bound(
options: EpiphanyOpenAiRuntimeOptions,
model_request: EpiphanyModelRequest,
request: EpiphanyOpenAiModelRequest,
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
ensure_openai_runtime_ready(&options)?;
let provider = model_request.provider.clone();
let profile = EpiphanyProviderProfile::parse(&provider)?;
open_runtime_model_execution(
&options.store_path,
RuntimeSpineSessionOptions {
session_id: options.session_id.clone(),
objective: options.objective.clone(),
created_at: now(),
coordinator_note: options.coordinator_note.clone(),
},
RuntimeSpineJobOptions {
job_id: options.job_id.clone(),
session_id: options.session_id.clone(),
role: OPENAI_RUNTIME_ROLE.to_string(),
created_at: now(),
},
&model_request,
&now(),
)?;
let events = match profile {
EpiphanyProviderProfile::OpenAiCodex => {
let auth_manager = auth_manager(options.codex_home.clone());
let transport = EpiphanyCodexOpenAiTransport::openai(auth_manager);
collect_transport_events(transport.collect_model_events(request.clone()), &request)
.await
}
EpiphanyProviderProfile::OpenRouter => {
let credential = read_static_provider_credential(
options.provider_credential_path.as_deref(),
OPENROUTER_MODEL_PROVIDER,
)?;
let transport = EpiphanyOpenRouterTransport::new(credential, options.request_timeout)?;
collect_transport_events(transport.collect_model_events(request.clone()), &request)
.await
}
};
record_openai_events(&options.store_path, &options, &request, &provider, &events)
}
async fn collect_transport_events<F>(
future: F,
request: &EpiphanyOpenAiModelRequest,
) -> Vec<EpiphanyOpenAiStreamEvent>
where
F: std::future::Future<Output = Result<Vec<EpiphanyOpenAiStreamEvent>>>,
{
match future.await {
Ok(events) => events,
Err(err) => vec![EpiphanyOpenAiStreamEvent {
request_id: request.request_id.clone(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::Failed {
message: err.to_string(),
},
}],
}
}
fn read_static_provider_credential(path: Option<&Path>, provider: &str) -> Result<String> {
let path = path.ok_or_else(|| {
anyhow!("model provider {provider:?} requires an explicit credential file")
})?;
let raw = std::fs::read_to_string(path).with_context(|| {
format!(
"failed to read {provider} credential file {}",
path.display()
)
})?;
let credential = raw.trim();
if credential.is_empty() || raw.trim_matches(['\r', '\n']) != credential {
return Err(anyhow!(
"model provider {provider:?} credential file is empty or contains surrounding whitespace"
));
}
Ok(credential.to_string())
}
pub async fn run_model_turn(
provider: &str,
options: EpiphanyOpenAiRuntimeOptions,
request: EpiphanyModelRequest,
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
EpiphanyProviderProfile::parse(provider)?;
if !provider_matches_request(provider, &request.provider) {
return Err(anyhow!(
"model request provider {:?} does not match selected provider {:?}",
request.provider,
provider
));
}
let provider_request = openai_request_from_model_request(&request);
run_openai_model_turn_bound(options, request, provider_request).await
}
pub async fn run_tool_followup_model_turn(
provider: &str,
options: EpiphanyOpenAiRuntimeOptions,
original_request_id: &str,
followup_request_id: &str,
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
EpiphanyProviderProfile::parse(provider)?;
let request = build_tool_followup_model_request(
&options.store_path,
original_request_id,
followup_request_id,
)?;
run_model_turn(provider, options, request).await
}
pub async fn run_worker_launch_observed<F>(
options: EpiphanyWorkerRuntimeOptions,
on_model_request: F,
) -> Result<EpiphanyWorkerRuntimeRunSummary>
where
F: FnOnce(&str),
{
let launch_request = load_worker_launch_request(&options.store_path, &options.job_id)?;
if !epiphany_core::runtime_requested_public_source_refs_for_worker(
&options.store_path,
&options.job_id,
)?
.is_empty()
{
return Err(anyhow!(
"worker {:?} has request-owned public sources and must run through the governed automatic tool route",
options.job_id
));
}
let basis = epiphany_core::worker_reasoning_basis(&options.store_path, &launch_request)?;
epiphany_core::put_reasoning_basis(&options.store_path, &basis)?;
let model_request =
build_worker_model_request(&launch_request, &options.provider, &options.model, &basis)?;
let openai_options = EpiphanyOpenAiRuntimeOptions {
store_path: options.store_path.clone(),
codex_home: options.codex_home,
provider_credential_path: options.provider_credential_path,
session_id: worker_model_session_id(&launch_request.job_id),
job_id: format!("openai-worker-{}", launch_request.job_id),
objective: format!(
"Run Epiphany worker {} for {}",
launch_request.job_id, launch_request.binding_id
),
coordinator_note: "Native worker runtime route; Codex is auth/model transport only."
.to_string(),
default_model: Some(options.model),
request_timeout: options.request_timeout,
};
on_model_request(&model_request.request_id);
let openai_summary = run_model_turn(
&options.provider,
openai_options.clone(),
model_request.clone(),
)
.await?;
let assistant_text =
assistant_text_from_model_events(&openai_options.store_path, &model_request.request_id)?;
let worker_result = complete_worker_job_from_assistant_text(
&openai_options.store_path,
&launch_request,
&model_request.request_id,
&openai_summary,
&assistant_text,
)?;
close_runtime_session(
&openai_options.store_path,
RuntimeSpineSessionClosureOptions {
session_id: openai_options.session_id.clone(),
completed_at: now(),
summary: format!(
"Worker model execution {} reached terminal result {}.",
launch_request.job_id, worker_result.result_id
),
},
)?;
Ok(EpiphanyWorkerRuntimeRunSummary {
store: openai_options.store_path.display().to_string(),
job_id: launch_request.job_id,
binding_id: launch_request.binding_id,
role: launch_request.role,
request_id: model_request.request_id,
openai_result_id: openai_summary.result_id,
worker_result_id: worker_result.result_id,
verdict: worker_result.verdict,
summary: worker_result.summary,
next_safe_move: worker_result.next_safe_move,
evidence_refs: worker_result.evidence_refs,
artifact_refs: worker_result.artifact_refs,
})
}
fn record_openai_events(
store_path: impl AsRef<Path>,
options: &EpiphanyOpenAiRuntimeOptions,
request: &EpiphanyOpenAiModelRequest,
provider: &str,
events: &[EpiphanyOpenAiStreamEvent],
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
let store_path = store_path.as_ref();
let events = compact_openai_events_for_storage(events);
let tool_intents = tool_invocation_intents_from_openai_events(provider, &events);
for intent in &tool_intents {
put_runtime_tool_execution_intent(
store_path,
&options.session_id,
&options.job_id,
intent,
&now(),
)?;
}
let mut receipt = None;
let mut failure = None;
{
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
for event in &events {
let model_event = model_event_from_openai_event(provider, event);
cache.put(
model_event_key(&model_event.request_id, model_event.sequence),
&model_event,
)?;
if let EpiphanyModelStreamPayload::Completed { receipt } = &model_event.payload {
cache.put(model_receipt_key(&receipt.request_id), receipt)?;
}
match &event.payload {
EpiphanyOpenAiStreamPayload::Completed { receipt: completed } => {
receipt = Some(completed.clone());
}
EpiphanyOpenAiStreamPayload::Failed { message } => {
failure = Some(message.clone());
}
_ => {}
}
}
}
let verdict = if failure.is_some() || receipt.is_none() {
"failed"
} else {
"pass"
};
let summary = if let Some(message) = failure {
format!(
"OpenAI model request {} failed: {message}",
request.request_id
)
} else if let Some(receipt) = &receipt {
format!(
"OpenAI model request {} completed through {}.",
request.request_id,
receipt
.transport
.clone()
.unwrap_or_else(|| "unknown transport".to_string())
)
} else {
format!(
"OpenAI model request {} ended without a terminal receipt.",
request.request_id
)
};
let result_id = format!("result-openai-{}", options.job_id);
complete_runtime_job(
store_path,
RuntimeSpineJobResultOptions {
result_id: result_id.clone(),
job_id: options.job_id.clone(),
completed_at: now(),
verdict: verdict.to_string(),
summary: summary.clone(),
next_safe_move: "Review typed OpenAI receipt before accepting downstream state."
.to_string(),
evidence_refs: Vec::new(),
artifact_refs: Vec::new(),
decision_context_id: None,
},
)?;
Ok(EpiphanyOpenAiRuntimeRunSummary {
store: store_path.display().to_string(),
session_id: options.session_id.clone(),
job_id: options.job_id.clone(),
request_id: request.request_id.clone(),
event_count: events.len(),
verdict: verdict.to_string(),
summary,
result_id,
receipt_id: receipt.map(|item| item.request_id),
tool_intent_ids: tool_intents
.into_iter()
.map(|intent| intent.intent_id)
.collect(),
})
}
/// Diagnostic event admission is native-request owned. The provider request
/// used for summaries and event binding is always lowered internally, never
/// supplied as an independent caller-authored document.
pub fn record_native_model_events(
store_path: impl AsRef<Path>,
options: &EpiphanyOpenAiRuntimeOptions,
request: &EpiphanyModelRequest,
events: &[EpiphanyOpenAiStreamEvent],
) -> Result<EpiphanyOpenAiRuntimeRunSummary> {
let provider_request = epiphany_openai_adapter::request_from_native(request);
record_openai_events(
store_path,
options,
&provider_request,
&request.provider,
events,
)
}
fn compact_openai_events_for_storage(
events: &[EpiphanyOpenAiStreamEvent],
) -> Vec<EpiphanyOpenAiStreamEvent> {
let mut compacted = Vec::new();
let mut text_buffer = String::new();
let mut reasoning_buffer = String::new();
for event in events {
match &event.payload {
EpiphanyOpenAiStreamPayload::TextDelta { text } => {
flush_reasoning_buffer(&mut compacted, event, &mut reasoning_buffer);
text_buffer.push_str(text);
}
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } => {
flush_text_buffer(&mut compacted, event, &mut text_buffer);
reasoning_buffer.push_str(text);
}
_ => {
flush_text_buffer(&mut compacted, event, &mut text_buffer);
flush_reasoning_buffer(&mut compacted, event, &mut reasoning_buffer);
push_compacted_event(&mut compacted, event, event.payload.clone());
}
}
}
if let Some(last) = events.last() {
flush_text_buffer(&mut compacted, last, &mut text_buffer);
flush_reasoning_buffer(&mut compacted, last, &mut reasoning_buffer);
}
compacted
}
fn flush_text_buffer(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
buffer: &mut String,
) {
if !buffer.is_empty() {
let text = std::mem::take(buffer);
push_compacted_event(
compacted,
source,
EpiphanyOpenAiStreamPayload::TextDelta { text },
);
}
}
fn flush_reasoning_buffer(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
buffer: &mut String,
) {
if !buffer.is_empty() {
let text = std::mem::take(buffer);
push_compacted_event(
compacted,
source,
EpiphanyOpenAiStreamPayload::ReasoningDelta { text },
);
}
}
fn push_compacted_event(
compacted: &mut Vec<EpiphanyOpenAiStreamEvent>,
source: &EpiphanyOpenAiStreamEvent,
payload: EpiphanyOpenAiStreamPayload,
) {
compacted.push(EpiphanyOpenAiStreamEvent {
request_id: source.request_id.clone(),
sequence: compacted.len() as u64,
payload,
});
}
pub fn load_worker_launch_request(
store_path: impl AsRef<Path>,
job_id: &str,
) -> Result<EpiphanyRuntimeWorkerLaunchRequest> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache
.get::<EpiphanyRuntimeWorkerLaunchRequest>(job_id)?
.ok_or_else(|| anyhow!("runtime worker launch request {job_id:?} does not exist"))
}
pub fn build_worker_model_request(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
provider: &str,
model: &str,
reasoning_basis: &epiphany_core::EpiphanyReasoningBasis,
) -> Result<EpiphanyModelRequest> {
reasoning_basis.validate()?;
if reasoning_basis.pass_id != launch_request.job_id {
return Err(anyhow!(
"worker model request reasoning basis belongs to another pass"
));
}
let launch_document = launch_request.launch_document()?;
let reasoning_projection = reasoning_basis.projection()?;
match (&launch_document, &reasoning_projection) {
(
EpiphanyWorkerLaunchDocument::Role(document),
epiphany_core::EpiphanyReasoningProjection::RolePass(projection),
) if projection.authority
== epiphany_core::EpiphanyRolePassAuthorityProjection::from(document.clone()) => {}
(
EpiphanyWorkerLaunchDocument::Reorient(document),
epiphany_core::EpiphanyReasoningProjection::ReorientLaunch(projection),
) if projection.authority == *document => {}
_ => {
return Err(anyhow!(
"worker model request reasoning projection does not match its launch authority"
));
}
}
let output_schema_json = worker_output_schema_json(launch_request, &launch_document)?;
let request_id = format!(
"worker-{}-{}",
sanitize_request_id(&launch_request.job_id),
chrono::Utc::now().timestamp_millis()
);
let reasoning_projection_text = serde_json::to_string_pretty(&reasoning_projection)
.context("failed to render sealed worker reasoning projection for model input")?;
let mut instructions = worker_instructions(launch_request, &launch_document);
if launch_request.binding_id == epiphany_core::EPIPHANY_VERIFICATION_ROLE_BINDING_ID {
instructions.push_str("\n\nTool mandate: before returning `needs-evidence` because source files, artifact directories, command artifacts, commit diffs, Hands receipt bodies, or resident grant lifecycle are not inspectable, call the governed read-only tools available on this request. Use `mcp__epiphany_source__read_file` for cited source/artifact files, `mcp__epiphany_source__directory_inventory` for bounded workspace directory counts and bytes, `mcp__epiphany_source__git_show` for commit diffs, `mcp__epiphany_source__read_hands_receipt` for Hands patch/command/commit receipts, and `mcp__epiphany_state__resident_grant_lifecycle` for exact or bounded recent grant-owned lifecycle state. Directory totals are authoritative only when the tool reports `complete=true`. Grant launchability is authoritative only from the typed state projection, never artifact names or acknowledgement presence. If a tool fails, cite that failed tool result and the exact remaining blocker.");
} else if launch_request.binding_id == epiphany_core::EPIPHANY_RESEARCH_ROLE_BINDING_ID {
instructions.push_str("\n\nEvidence mandate: the runtime obtains every immutable public GitHub source named by the typed Research request before this model turn and supplies the exact tool calls and receipts in the input. Cite each requested sourceRef in filesInspected and its evidenceReceiptId in evidence, preserve its contentSha256 in the finding, and report a source gap if any supplied lookup failed. Use the remaining bounded tools only for additional repository or resident-state inspection appropriate to the claim. Never substitute a branch, tag, arbitrary URL, or model memory for immutable public evidence. Directory totals are authoritative only when the inventory reports `complete=true`; grant launchability is authoritative only from the typed lifecycle projection.");
} else if launch_request.binding_id == epiphany_core::EPIPHANY_MODELING_ROLE_BINDING_ID {
instructions.push_str("\n\nTool mandate: Modeling must inspect current repository sources or typed resident state before proposing repository anatomy. Call the bounded read-only file, directory-inventory, Git, or `mcp__epiphany_state__resident_grant_lifecycle` tool appropriate to the claim; cite exact inspected paths/revisions or grant identities in filesInspected and evidence, and emit regather-needed instead of inventing unobserved structure. Directory totals are authoritative only when the inventory reports `complete=true`; grant launchability is authoritative only from the typed lifecycle projection.");
}
let mut request = EpiphanyModelRequest::new(
request_id,
format!("worker-{}", launch_request.binding_id),
provider,
model.to_string(),
instructions,
);
request.input.push(EpiphanyModelInputItem::UserText {
text: format!(
"Execute this sealed Epiphany reasoning projection.\n\n```json\n{reasoning_projection_text}\n```"
),
});
request.reasoning_effort = Some("low".to_string());
request.reasoning_summary = Some("concise".to_string());
request.output_contract_id = Some(launch_request.output_contract_id.clone());
request.output_schema_json = Some(output_schema_json);
request.source_worker_job_id = Some(launch_request.job_id.clone());
request.reasoning_basis_id = Some(reasoning_basis.basis_id.clone());
if matches!(
launch_request.binding_id.as_str(),
epiphany_core::EPIPHANY_RESEARCH_ROLE_BINDING_ID
| epiphany_core::EPIPHANY_MODELING_ROLE_BINDING_ID
| epiphany_core::EPIPHANY_VERIFICATION_ROLE_BINDING_ID
) {
request.tools = repository_source_tools();
}
Ok(request)
}
pub fn complete_worker_job_from_assistant_text(
store_path: impl AsRef<Path>,
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
openai_request_id: &str,
openai_summary: &EpiphanyOpenAiRuntimeRunSummary,
assistant_text: &str,
) -> Result<epiphany_core::EpiphanyRuntimeJobResult> {
let decision_context =
epiphany_core::seal_model_decision_context(store_path.as_ref(), openai_request_id)?;
let launch_document = launch_request.launch_document()?;
let parsed_result = parse_worker_result_ingress(&launch_document, assistant_text);
let parse_error = parsed_result
.as_ref()
.err()
.map(|error| format!("{error:#}"));
let parsed = parsed_result.ok();
let openai_failed = openai_summary.verdict != "pass";
let contract_failed = !openai_failed && parsed.is_none();
if openai_failed || contract_failed {
let summary = if openai_failed {
format!(
"Worker model request {openai_request_id} failed before producing usable output."
)
} else {
format!(
"Worker model response failed declared output contract {}: {}",
launch_request.output_contract_id,
parse_error
.as_deref()
.unwrap_or("structured output was absent")
)
};
let next_safe_move = if contract_failed {
"Repair the worker prompt/output-schema boundary before relaunching; no typed role result was admitted."
.to_string()
} else {
"Inspect provider/tool transport and the sealed model-pass failure before relaunching."
.to_string()
};
return fail_model_backed_worker_job(
store_path,
&launch_request.job_id,
openai_request_id,
if openai_failed {
"provider_or_transport_failure"
} else {
"output_contract_failure"
},
summary,
next_safe_move,
);
}
let verdict = if openai_failed || contract_failed {
"failed".to_string()
} else {
parsed
.as_ref()
.and_then(WorkerResultIngress::verdict)
.unwrap_or_else(|| "completed".to_string())
};
let summary = if openai_failed {
format!("Worker model request {openai_request_id} failed before producing usable output.")
} else if let Some(error) = parse_error.as_deref() {
format!(
"Worker model response failed declared output contract {}: {error}",
launch_request.output_contract_id
)
} else {
parsed
.as_ref()
.and_then(WorkerResultIngress::summary)
.unwrap_or_else(|| "Worker completed without a structured summary.".to_string())
};
let next_safe_move = if contract_failed {
"Repair the worker prompt/output-schema boundary before relaunching; no typed role result was admitted."
.to_string()
} else {
parsed
.as_ref()
.and_then(WorkerResultIngress::next_safe_move)
.unwrap_or_else(|| {
"Review the typed worker runtime result before accepting state.".to_string()
})
};
let mut evidence_refs = parsed
.as_ref()
.map(WorkerResultIngress::evidence_ids)
.unwrap_or_default();
evidence_refs.push(format!("openai-request:{openai_request_id}"));
let mut artifact_refs = parsed
.as_ref()
.map(WorkerResultIngress::artifact_refs)
.unwrap_or_default();
artifact_refs.push(format!("openai-result:{}", openai_summary.result_id));
let result_id = format!("result-worker-{}", launch_request.job_id);
let completed_at = now();
if let Some(parsed) = parsed.as_ref() {
match (&launch_document, parsed) {
(EpiphanyWorkerLaunchDocument::Role(document), WorkerResultIngress::Role(parsed)) => {
let typed_result = role_worker_result_from_ingress(
launch_request,
&document.role_id,
document.repository_body_observation_basis.as_ref(),
document.proposal_modeling_context.as_ref(),
document.frontier_planning_context.as_ref(),
document.frontier_plan_mind_context.as_ref(),
document.imagination_consideration_context.as_ref(),
document
.admitted_model_direction_consideration_context
.as_ref(),
document.frontier_verdict_modeling_context.as_ref(),
&completed_at,
&result_id,
&decision_context.context_id,
parsed,
evidence_refs.clone(),
artifact_refs.clone(),
);
put_runtime_role_worker_result(store_path.as_ref(), &typed_result)?;
}
(EpiphanyWorkerLaunchDocument::Reorient(_), WorkerResultIngress::Reorient(parsed)) => {
let typed_result = reorient_worker_result_from_ingress(
launch_request,
&result_id,
&decision_context.context_id,
parsed,
artifact_refs.clone(),
);
put_runtime_reorient_worker_result(store_path.as_ref(), &typed_result)?;
}
_ => {
return Err(anyhow!(
"worker launch document and parsed result kind diverged"
));
}
}
}
complete_runtime_job(
store_path,
RuntimeSpineJobResultOptions {
result_id,
job_id: launch_request.job_id.clone(),
completed_at,
verdict,
summary,
next_safe_move,
evidence_refs,
artifact_refs,
decision_context_id: Some(decision_context.context_id),
},
)
}
pub fn fail_worker_job(
store_path: impl AsRef<Path>,
job_id: &str,
summary: String,
next_safe_move: String,
) -> Result<epiphany_core::EpiphanyRuntimeJobResult> {
complete_runtime_job(
store_path,
RuntimeSpineJobResultOptions {
result_id: format!("result-worker-{job_id}"),
job_id: job_id.to_string(),
completed_at: now(),
verdict: "failed".to_string(),
summary,
next_safe_move,
evidence_refs: Vec::new(),
artifact_refs: Vec::new(),
decision_context_id: None,
},
)
}
pub fn fail_model_backed_worker_job(
store_path: impl AsRef<Path>,
job_id: &str,
terminal_request_id: &str,
failure_kind: &str,
summary: String,
next_safe_move: String,
) -> Result<epiphany_core::EpiphanyRuntimeJobResult> {
let context =
epiphany_core::seal_model_decision_context(store_path.as_ref(), terminal_request_id)?;
let failure = epiphany_core::terminalize_model_pass_failure_session(
store_path.as_ref(),
epiphany_core::ModelPassFailureTerminalOptions {
decision_context_id: context.context_id.clone(),
failure_kind: failure_kind.to_string(),
summary: summary.clone(),
failed_at: now(),
},
)?;
let launch = load_worker_launch_request(store_path.as_ref(), job_id)?;
if epiphany_core::runtime_role_worker_result(store_path.as_ref(), job_id)?.is_none()
&& let Some(role_failure) =
failed_frontier_planning_role_result(&launch, &summary, &context.context_id)?
{
epiphany_core::put_runtime_role_worker_result(store_path.as_ref(), &role_failure)?;
}
complete_runtime_job(
store_path,
RuntimeSpineJobResultOptions {
result_id: format!("result-worker-{job_id}"),
job_id: job_id.to_string(),
completed_at: now(),
verdict: "failed".to_string(),
summary,
next_safe_move,
evidence_refs: Vec::new(),
artifact_refs: Vec::new(),
decision_context_id: Some(failure.decision_context_id),
},
)
}
pub fn ensure_openai_runtime_ready(options: &EpiphanyOpenAiRuntimeOptions) -> Result<()> {
if runtime_identity(&options.store_path)?.is_some() {
return Ok(());
}
initialize_runtime_spine(
&options.store_path,
RuntimeSpineInitOptions {
runtime_id: "epiphany-openai-runtime".to_string(),
display_name: "Epiphany OpenAI Runtime".to_string(),
created_at: now(),
},
)?;
Ok(())
}
pub fn assistant_text_from_model_events(
store_path: impl AsRef<Path>,
request_id: &str,
) -> Result<String> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let mut events = cache
.get_all::<EpiphanyModelStreamEvent>()?
.into_iter()
.filter(|event| event.request_id == request_id)
.collect::<Vec<_>>();
events.sort_by_key(|event| event.sequence);
let mut text = String::new();
for event in events {
if let EpiphanyModelStreamPayload::TextDelta { text: delta } = event.payload {
text.push_str(&delta);
}
}
Ok(text)
}
pub fn build_tool_followup_model_request(
store_path: impl AsRef<Path>,
original_request_id: &str,
followup_request_id: &str,
) -> Result<EpiphanyModelRequest> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let original = cache
.get::<EpiphanyModelRequest>(&model_request_key(original_request_id))?
.ok_or_else(|| anyhow!("model request {original_request_id:?} does not exist"))?;
let mut call_order = BTreeMap::new();
let mut events = cache
.get_all::<EpiphanyModelStreamEvent>()?
.into_iter()
.filter(|event| event.request_id == original_request_id)
.collect::<Vec<_>>();
events.sort_by_key(|event| event.sequence);
for event in events {
if let EpiphanyModelStreamPayload::ToolCall { call_id, .. } = event.payload {
if call_order.insert(call_id, event.sequence).is_some() {
return Err(anyhow!("model request has duplicate tool-call identity"));
}
}
}
let mut followup_items = Vec::new();
for intent in cache.get_all::<EpiphanyToolInvocationIntent>()? {
if intent.model_request_id.as_deref() != Some(original_request_id) {
continue;
}
let Some(call_id) = intent.call_id.clone() else {
continue;
};
let sequence = *call_order
.get(&call_id)
.ok_or_else(|| anyhow!("tool intent has no exact model-event call owner"))?;
let Some(receipt) = cache.get::<EpiphanyToolInvocationReceipt>(
&tool_invocation_receipt_key(&intent.intent_id),
)?
else {
continue;
};
followup_items.push((sequence, intent, call_id, receipt));
}
followup_items.sort_by(|left, right| {
left.0
.cmp(&right.0)
.then_with(|| left.1.intent_id.cmp(&right.1.intent_id))
});
if followup_items.is_empty() {
return Err(anyhow!(
"model request {original_request_id:?} has no completed tool receipts with call ids"
));
}
let mut followup = original;
followup.request_id = followup_request_id.to_string();
followup.previous_response_id = None;
let mut input = followup.input.clone();
for (_, intent, call_id, receipt) in followup_items {
input.push(EpiphanyModelInputItem::ToolCall {
call_id: call_id.clone(),
name: format!("mcp__{}__{}", intent.server, intent.tool_name),
arguments: intent.arguments_json.clone(),
});
input.push(EpiphanyModelInputItem::ToolResult {
call_id,
output: receipt_output_for_model(&intent, &receipt),
});
}
followup.input = input;
Ok(followup)
}
pub fn append_requested_public_source_receipts(
store_path: impl AsRef<Path>,
request: &mut EpiphanyModelRequest,
intents: &[EpiphanyToolInvocationIntent],
) -> Result<()> {
let store_path = store_path.as_ref();
let source_worker_job_id = request
.source_worker_job_id
.as_deref()
.ok_or_else(|| anyhow!("requested public source context has no source worker"))?;
let expected_sources = epiphany_core::runtime_requested_public_source_refs_for_worker(
store_path,
source_worker_job_id,
)?
.into_iter()
.collect::<BTreeSet<_>>();
let mut observed_sources = BTreeSet::new();
for intent in intents {
let arguments: serde_json::Value = serde_json::from_str(&intent.arguments_json)
.context("requested public source intent arguments are invalid")?;
let component = |name: &str| -> Result<&str> {
arguments
.get(name)
.and_then(serde_json::Value::as_str)
.ok_or_else(|| anyhow!("requested public source intent omitted {name:?}"))
};
observed_sources.insert(
epiphany_core::ImmutableGithubSource::from_components(
component("owner")?,
component("repository")?,
component("revision")?,
component("path")?,
)?
.to_string(),
);
}
if observed_sources != expected_sources || observed_sources.len() != intents.len() {
return Err(anyhow!(
"requested public source model context does not exactly cover its typed Research request"
));
}
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
for intent in intents {
if intent.model_request_id.is_some() {
return Err(anyhow!(
"requested public source intent {:?} is model-owned",
intent.intent_id
));
}
if intent.caller != "epiphany-runtime-requested-public-source"
|| intent.server != "epiphany_public"
|| intent.tool_name != "github_file"
{
return Err(anyhow!(
"requested public source intent {:?} is not the canonical public-source operation",
intent.intent_id
));
}
let binding =
epiphany_core::require_runtime_tool_execution_binding(store_path, &intent.intent_id)?;
if binding.job_id != source_worker_job_id || binding.model_request_id.is_some() {
return Err(anyhow!(
"requested public source intent {:?} is not owned by worker {:?}",
intent.intent_id,
source_worker_job_id
));
}
let call_id = intent
.call_id
.as_deref()
.ok_or_else(|| anyhow!("requested public source intent has no call id"))?;
let receipt = cache
.get::<EpiphanyToolInvocationReceipt>(&tool_invocation_receipt_key(&intent.intent_id))?
.ok_or_else(|| {
anyhow!(
"requested public source intent {:?} has no terminal receipt",
intent.intent_id
)
})?;
if receipt.intent_id != intent.intent_id
|| receipt.adapter != intent.adapter
|| receipt.server != intent.server
|| receipt.tool_name != intent.tool_name
{
return Err(anyhow!(
"requested public source intent {:?} has a foreign receipt",
intent.intent_id
));
}
request.input.push(EpiphanyModelInputItem::ToolCall {
call_id: call_id.to_string(),
name: format!("mcp__{}__{}", intent.server, intent.tool_name),
arguments: intent.arguments_json.clone(),
});
request.input.push(EpiphanyModelInputItem::ToolResult {
call_id: call_id.to_string(),
output: receipt_output_for_model(intent, &receipt),
});
}
Ok(())
}
pub fn default_options(
store_path: PathBuf,
codex_home: PathBuf,
request: &EpiphanyOpenAiModelRequest,
) -> EpiphanyOpenAiRuntimeOptions {
EpiphanyOpenAiRuntimeOptions {
store_path,
codex_home,
provider_credential_path: None,
session_id: format!("openai-session-{}", request.conversation_id),
job_id: format!("openai-job-{}", request.request_id),
objective: format!("Run typed OpenAI model request {}", request.request_id),
coordinator_note: "Native OpenAI runtime route; Codex is auth/model transport only."
.to_string(),
default_model: Some(request.model.clone()),
request_timeout: Some(DEFAULT_PROVIDER_REQUEST_TIMEOUT),
}
}
pub fn model_request_key(request_id: &str) -> String {
request_id.to_string()
}
pub fn model_event_key(request_id: &str, sequence: u64) -> String {
format!("{request_id}:{sequence:08}")
}
pub fn model_receipt_key(request_id: &str) -> String {
request_id.to_string()
}
pub fn openai_request_from_model_request(
request: &EpiphanyModelRequest,
) -> EpiphanyOpenAiModelRequest {
epiphany_openai_adapter::request_from_native(request)
}
fn repository_source_tools() -> Vec<EpiphanyModelToolDefinition> {
vec![
EpiphanyModelToolDefinition {
name: "mcp__epiphany_source__read_file".to_string(),
description: "Read a bounded UTF-8 text slice from the current workspace for source-grounded Eyes, Modeling, or Soul work. Use only for repository sources and operator-safe artifacts in scope.".to_string(),
parameters_json: serde_json::json!({
"type": "object",
"additionalProperties": false,
"properties": {
"path": {"type": "string"},
"startLine": {"type": "integer", "minimum": 1},
"maxLines": {"type": "integer", "minimum": 1, "maximum": 240}
},
"required": ["path"]
})
.to_string(),
},
EpiphanyModelToolDefinition {
name: "mcp__epiphany_source__directory_inventory".to_string(),
description: "Measure a workspace-confined directory with deterministic bounded counts, regular-file bytes, and path samples. Totals are usable as complete evidence only when complete=true; symlinks are counted but never followed.".to_string(),
parameters_json: serde_json::json!({
"type": "object",
"additionalProperties": false,
"properties": {
"path": {"type": "string"},
"maxDepth": {"type": "integer", "minimum": 0, "maximum": 8},
"maxEntries": {"type": "integer", "minimum": 1, "maximum": 4096},
"maxSamples": {"type": "integer", "minimum": 1, "maximum": 100}
},
"required": ["path"]
})
.to_string(),
},
EpiphanyModelToolDefinition {
name: "mcp__epiphany_source__git_show".to_string(),
description: "Read a bounded git show/diff preview for a commit or revision in the current workspace.".to_string(),
parameters_json: serde_json::json!({
"type": "object",
"additionalProperties": false,
"properties": {
"revision": {"type": "string"},
"paths": {"type": "array", "items": {"type": "string"}},
"maxBytes": {"type": "integer", "minimum": 512, "maximum": 24000}
},
"required": ["revision"]
})
.to_string(),
},
EpiphanyModelToolDefinition {
name: "mcp__epiphany_source__read_hands_receipt".to_string(),
description: "Read a typed Hands patch, command, or commit receipt body from the runtime-spine store for Soul verification.".to_string(),
parameters_json: serde_json::json!({
"type": "object",
"additionalProperties": false,
"properties": {
"receiptId": {"type": "string"},
"kind": {"type": "string", "enum": ["patch", "command", "commit"]}
},
"required": ["receiptId", "kind"]
})
.to_string(),
},
EpiphanyModelToolDefinition {
name: "mcp__epiphany_state__resident_grant_lifecycle".to_string(),
description: "Read grant-owned resident Self lifecycle state from the explicitly launch-bound resident store. Use an exact grantId when known or a bounded recent limit. This is observation only; terminal and launchable fields remain derived from the typed grant/state owner.".to_string(),
parameters_json: serde_json::json!({
"type": "object",
"additionalProperties": false,
"properties": {
"grantId": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 100}
}
})
.to_string(),
},
]
}
pub fn model_event_from_openai_event(
provider: &str,
event: &EpiphanyOpenAiStreamEvent,
) -> EpiphanyModelStreamEvent {
EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: event.request_id.clone(),
provider: provider.to_string(),
sequence: event.sequence,
payload: match &event.payload {
EpiphanyOpenAiStreamPayload::TextDelta { text } => {
EpiphanyModelStreamPayload::TextDelta { text: text.clone() }
}
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } => {
EpiphanyModelStreamPayload::ReasoningDelta { text: text.clone() }
}
EpiphanyOpenAiStreamPayload::ToolCall {
call_id,
name,
arguments,
} => EpiphanyModelStreamPayload::ToolCall {
call_id: call_id.clone(),
name: name.clone(),
arguments: arguments.clone(),
},
EpiphanyOpenAiStreamPayload::Completed { receipt } => {
EpiphanyModelStreamPayload::Completed {
receipt: model_receipt_from_openai_receipt(provider, receipt),
}
}
EpiphanyOpenAiStreamPayload::Failed { message } => EpiphanyModelStreamPayload::Failed {
message: message.clone(),
},
},
}
}
pub fn model_receipt_from_openai_receipt(
provider: &str,
receipt: &epiphany_openai_adapter::EpiphanyOpenAiModelReceipt,
) -> EpiphanyModelReceipt {
EpiphanyModelReceipt {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_RECEIPT_SCHEMA_ID.to_string(),
request_id: receipt.request_id.clone(),
provider: provider.to_string(),
model: receipt.model.clone(),
provider_response_id: receipt.response_id.clone(),
input_tokens: receipt.input_tokens,
output_tokens: receipt.output_tokens,
reasoning_output_tokens: receipt.reasoning_output_tokens,
transport: receipt.transport.clone(),
}
}
pub fn tool_invocation_intents_from_openai_events(
provider: &str,
events: &[EpiphanyOpenAiStreamEvent],
) -> Vec<EpiphanyToolInvocationIntent> {
events
.iter()
.filter_map(|event| {
let model_event = model_event_from_openai_event(provider, event);
tool_invocation_intent_from_model_event(&model_event)
})
.collect()
}
pub fn tool_invocation_intent_from_model_event(
event: &EpiphanyModelStreamEvent,
) -> Option<EpiphanyToolInvocationIntent> {
let EpiphanyModelStreamPayload::ToolCall {
call_id,
name,
arguments,
} = &event.payload
else {
return None;
};
let (server, tool_name) = split_mcp_tool_name(name)?;
if !arguments_are_invocation_ready(arguments) {
return None;
}
Some(
EpiphanyToolInvocationIntent::new(
format!(
"model-{}-{}-{}",
sanitize_request_id(&event.request_id),
event.sequence,
sanitize_request_id(call_id)
),
EPIPHANY_TOOL_RUNTIME_ADAPTER_ID,
server,
tool_name,
arguments.clone(),
format!("model-runtime:{}", event.provider),
format!(
"Model request {} emitted MCP tool call {}.",
event.request_id, call_id
),
now(),
)
.with_model_call(call_id.clone(), event.request_id.clone()),
)
}
fn split_mcp_tool_name(name: &str) -> Option<(String, String)> {
let mut parts = name.split("__");
if parts.next()? != "mcp" {
return None;
}
let server = parts.next()?.trim();
let tool = parts.collect::<Vec<_>>().join("__");
if server.is_empty() || tool.trim().is_empty() {
return None;
}
Some((server.to_string(), tool))
}
fn arguments_are_invocation_ready(arguments: &str) -> bool {
let trimmed = arguments.trim();
if trimmed.is_empty() || trimmed == "null" {
return true;
}
matches!(
serde_json::from_str::<serde_json::Value>(trimmed),
Ok(serde_json::Value::Object(_))
)
}
fn provider_matches_request(selected: &str, requested: &str) -> bool {
selected == requested || (selected == "openai" && requested == DEFAULT_MODEL_PROVIDER)
}
fn worker_instructions(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
launch_document: &EpiphanyWorkerLaunchDocument,
) -> String {
let output_contract = worker_output_contract_text(launch_request, launch_document);
let dynamic_context = launch_document
.dynamic_prompt_context()
.map(|context| format!("\n\n{context}"))
.unwrap_or_default();
format!(
"{}{}\n\nReturn only one JSON object through the declared response format. Emit every object key at most once. No Markdown, no commentary.\n\n{}",
launch_request.instruction, dynamic_context, output_contract
)
}
fn worker_output_schema_json(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
document: &EpiphanyWorkerLaunchDocument,
) -> Result<String> {
let schema = match document {
EpiphanyWorkerLaunchDocument::Role(document) => {
if document.frontier_plan_mind_context.is_some() {
return serde_json::to_string_pretty(
&epiphany_core::epiphany_frontier_plan_mind_output_schema(),
)
.context("failed to render worker output schema");
}
let role_id = role_result_id_for_launch_role(&document.role_id)
.with_context(|| format!("unknown role launch id {:?}", document.role_id))?;
if document.frontier_planning_context.is_some() {
epiphany_core::epiphany_frontier_planning_output_schema()
} else if document
.admitted_model_direction_consideration_context
.is_some()
{
epiphany_core::epiphany_admitted_model_direction_consideration_output_schema()
} else if document.imagination_consideration_context.is_some() {
epiphany_core::epiphany_imagination_consideration_output_schema()
} else if document.proposal_modeling_context.is_some() {
epiphany_core::epiphany_proposal_modeling_output_schema()
} else if role_id == epiphany_core::EpiphanyRoleResultRoleId::Modeling
&& launch_request.repo_frontier_modeling_request_id.is_some()
{
let request_id = launch_request
.repo_frontier_modeling_request_id
.as_deref()
.expect("checked verdict-bound Modeling request identity");
let authority = document
.frontier_verdict_modeling_context
.as_ref()
.ok_or_else(|| {
anyhow!(
"verdict-bound Modeling launch {request_id:?} omitted its typed authority body"
)
})?;
if authority.request.request_id != request_id {
return Err(anyhow!(
"verdict-bound Modeling launch identity mismatch: indexed {:?}, authority {:?}",
request_id,
authority.request.request_id
));
}
epiphany_core::epiphany_frontier_verdict_modeling_output_schema(authority)
} else {
epiphany_core::epiphany_role_launch_output_schema(role_id)
}
}
EpiphanyWorkerLaunchDocument::Reorient(_) => {
epiphany_core::epiphany_reorient_launch_output_schema()
}
};
serde_json::to_string_pretty(&schema).context("failed to render worker output schema")
}
fn role_result_id_for_launch_role(
role_id: &str,
) -> Option<epiphany_core::EpiphanyRoleResultRoleId> {
match role_id {
"imagination" => Some(epiphany_core::EpiphanyRoleResultRoleId::Imagination),
"research" => Some(epiphany_core::EpiphanyRoleResultRoleId::Research),
"modeling" => Some(epiphany_core::EpiphanyRoleResultRoleId::Modeling),
"verification" => Some(epiphany_core::EpiphanyRoleResultRoleId::Verification),
"implementation" => Some(epiphany_core::EpiphanyRoleResultRoleId::Implementation),
"reorientation" => Some(epiphany_core::EpiphanyRoleResultRoleId::Reorientation),
_ => None,
}
}
fn worker_output_contract_text(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
document: &EpiphanyWorkerLaunchDocument,
) -> &'static str {
match document {
EpiphanyWorkerLaunchDocument::Role(_)
if launch_request.repo_frontier_modeling_request_id.is_some() =>
{
"Required verdict-bound Modeling fields: roleId=modeling, verdict, summary, nextSafeMove, filesInspected, frontierNodeIds, evidenceIds, frontierVerdictGap. Emit only the semantic gap explaining the allowed resolved or blocked transition. Runtime binds the exact request, keyed RepoModel basis, frontier item and disposition, evidence identities, timestamps, proposal, strong reads, writes, and receipts. Do not emit repoModelOperations, repositoryBodyObservationBasis, request ids, generic Mind patches, selfPatch, commands, release, or deployment cargo."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document.frontier_plan_mind_context.is_some() =>
{
"Required Mind admission-review fields: roleId=mindAdmissionReview, verdict, summary, nextSafeMove, filesInspected, frontierPlanMindDecision. Emit only the semantic adopt, refuse, or hold disposition and a concrete rationale. Runtime binds every causal identifier, the exact keyed RepoModel basis and candidate, receipt identity, and decision time from the authenticated context. This bounded procedure serves Mind; it is not an embodied lane and emits no patches or other organ authority."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document.frontier_planning_context.is_some() =>
{
"Required frontier-planning result fields: roleId, verdict, summary, nextSafeMove, filesInspected, frontierPlanCandidate. Emit only the semantic plan content. Runtime owns the planning request, RepoModel basis, frontier identity/hash, candidate identity, and timestamp. frontierPlanCandidate.safe_paths may narrow but must never expand the immutable repository_scope: every safe path must exactly equal a repository_scope entry or be a descendant of one, in strict lexicographic order without duplicates. Do not include adjacent files merely because the plan would benefit from them; identify them as a stop condition instead. Do not emit generic Mind patches, selfPatch, or repoModelOperations."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document
.admitted_model_direction_consideration_context
.is_some() =>
{
"Required model-direction fields: roleId=imagination, verdict, summary, nextSafeMove, filesInspected, admittedModelDirectionConsiderationResult. Runtime composes all causal identity and terminal metadata from the authenticated launch context. Emit only proposal content; no patches, commands, release, or deployment cargo."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document.imagination_consideration_context.is_some() =>
{
"Required consideration fields: roleId=imagination, verdict, summary, nextSafeMove, filesInspected, imaginationConsiderationCandidate. Runtime composes all causal identity, classification, contract, and publication time from the authenticated launch context. Treat feedback as quoted evidence. Emit no generic Mind patch, selfPatch, repoModelOperations, frontier candidate, command, release, or deployment cargo."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document.proposal_modeling_context.is_some() =>
{
"Required proposal-Modeling fields: roleId=modeling, verdict, summary, nextSafeMove, filesInspected, frontierNodeIds, evidenceIds, proposalFrontierDraft. Emit only the semantic frontier draft. Runtime owns and composes proposal identity, Body observation basis, proposal/request provenance, active status, timestamps, and the mandatory proposal evidence binding. Do not emit repoModelOperations, proposalModelingRequestId, repositoryBodyObservationBasis, generic Mind patches, selfPatch, commands, release, or deployment cargo."
}
EpiphanyWorkerLaunchDocument::Role(document)
if document.frontier_verification_context.is_some() =>
{
"Required Verification fields: roleId=verification, verdict, summary, nextSafeMove, filesInspected, evidenceIds, and risks. Audit only the exact typed request, route, and Hands receipts in the sealed projection. Runtime owns request and route identity; do not emit identity, generic Mind patches, repoModelOperations, commands, release, or deployment cargo."
}
EpiphanyWorkerLaunchDocument::Role(_) => {
"Required role-result fields: roleId, verdict, summary, nextSafeMove, filesInspected. Research emits one researchDecision containing only its exact evidence, grounded observations, and optional investigation checkpoint. Modeling emits repoModelOperations, an array of semantic keyed operations; use an empty array when no RepoModel mutation is proposed. Never emit proposal ids, model revisions or hashes, timestamps, strong reads, writes, receipts, or generic state patches. For ordinary Modeling, checkpoint-update-needed is a typed claim that the Body map contains a future design gap: encode exactly one PutFrontier operation with a new active, unadopted frontier, recommended_next_organ=Imagination, empty dependency_item_ids, a safe non-empty repository_scope, and evidence_refs grounded in top-level evidenceIds. repository_scope is the strict sorted repository-relative path ceiling that downstream Planning may authorize Hands to change; include intended outputs, including files that do not yet exist, and do not substitute filesInspected or evidence sources. checkpoint-ready may carry only local Atlas offer/claim transitions or an empty operation array; regather-needed requires an empty operation array. Neither may mutate frontier. nextSafeMove is display-only and never routes an organ. Use arrays for frontierNodeIds, evidenceIds, openQuestions, evidenceGaps, risks, and artifactRefs when present."
}
EpiphanyWorkerLaunchDocument::Reorient(_) => {
"Required reorient-result fields: mode, summary, nextSafeMove. Include checkpointStillValid, filesInspected, frontierNodeIds, evidenceIds, openQuestions, and continuityRisks when present."
}
}
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
struct RoleWorkerResultIngress {
role_id: Option<String>,
verdict: Option<String>,
summary: Option<String>,
next_safe_move: Option<String>,
checkpoint_summary: Option<String>,
scratch_summary: Option<String>,
#[serde(deserialize_with = "deserialize_null_default")]
files_inspected: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
frontier_node_ids: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
evidence_ids: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
artifact_refs: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
open_questions: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
evidence_gaps: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
risks: Vec<String>,
research_decision: Option<epiphany_core::EpiphanyResearchDecision>,
#[serde(deserialize_with = "deserialize_null_default")]
repo_model_operations: Vec<epiphany_core::EpiphanyRepoModelMutationOperation>,
frontier_verdict_gap: Option<String>,
frontier_plan_candidate: Option<RepoFrontierPlanCandidateIngress>,
frontier_plan_mind_decision: Option<RepoFrontierPlanMindDecisionIngress>,
imagination_consideration_candidate: Option<ImaginationConsiderationCandidateIngress>,
admitted_model_direction_consideration_result:
Option<AdmittedModelDirectionConsiderationResultIngress>,
proposal_frontier_draft: Option<ProposalFrontierDraftIngress>,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
struct ProposalFrontierDraftIngress {
migration_body: String,
question: String,
gap: String,
target_claim_ids: Vec<String>,
repository_scope: Vec<String>,
recommended_next_organ: String,
dependency_item_ids: Vec<String>,
evidence_refs: Vec<String>,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
struct RepoFrontierPlanMindDecisionIngress {
decision: Option<epiphany_core::RepoFrontierPlanDecision>,
rationale: String,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
struct RepoFrontierPlanCandidateIngress {
safe_paths: Vec<String>,
action: String,
command: String,
checks: Vec<String>,
stop_conditions: Vec<String>,
rollback_steps: Vec<String>,
commit_message: String,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
struct ImaginationConsiderationCandidateIngress {
disposition: Option<epiphany_core::ImaginationConsiderationDisposition>,
title: String,
summary: String,
rationale: String,
option_drafts: Vec<epiphany_core::ImaginationOptionDraft>,
uncertainties: Vec<String>,
recommended_review_route: Option<epiphany_core::ImaginationConsiderationReviewRoute>,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
struct AdmittedModelDirectionConsiderationResultIngress {
disposition: Option<epiphany_core::AdmittedModelDirectionDisposition>,
summary: String,
option_drafts: Vec<epiphany_core::ImaginationOptionDraft>,
uncertainties: Vec<String>,
evidence_refs: Vec<String>,
}
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
struct ReorientWorkerResultIngress {
mode: Option<String>,
summary: Option<String>,
next_safe_move: Option<String>,
checkpoint_still_valid: Option<bool>,
#[serde(deserialize_with = "deserialize_null_default")]
files_inspected: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
frontier_node_ids: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
evidence_ids: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
artifact_refs: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
open_questions: Vec<String>,
#[serde(deserialize_with = "deserialize_null_default")]
continuity_risks: Vec<String>,
}
#[derive(Debug, Clone)]
enum WorkerResultIngress {
Role(RoleWorkerResultIngress),
Reorient(ReorientWorkerResultIngress),
}
impl WorkerResultIngress {
fn verdict(&self) -> Option<String> {
match self {
WorkerResultIngress::Role(result) => clean_optional_string(result.verdict.as_deref()),
WorkerResultIngress::Reorient(result) => clean_optional_string(result.mode.as_deref()),
}
}
fn summary(&self) -> Option<String> {
match self {
WorkerResultIngress::Role(result) => clean_optional_string(result.summary.as_deref()),
WorkerResultIngress::Reorient(result) => {
clean_optional_string(result.summary.as_deref())
}
}
}
fn next_safe_move(&self) -> Option<String> {
match self {
WorkerResultIngress::Role(result) => {
clean_optional_string(result.next_safe_move.as_deref())
}
WorkerResultIngress::Reorient(result) => {
clean_optional_string(result.next_safe_move.as_deref())
}
}
}
fn evidence_ids(&self) -> Vec<String> {
match self {
WorkerResultIngress::Role(result) => clean_string_vec(&result.evidence_ids),
WorkerResultIngress::Reorient(result) => clean_string_vec(&result.evidence_ids),
}
}
fn artifact_refs(&self) -> Vec<String> {
match self {
WorkerResultIngress::Role(result) => clean_string_vec(&result.artifact_refs),
WorkerResultIngress::Reorient(result) => clean_string_vec(&result.artifact_refs),
}
}
}
fn parse_worker_result_ingress(
document: &EpiphanyWorkerLaunchDocument,
assistant_text: &str,
) -> Result<WorkerResultIngress> {
match document {
EpiphanyWorkerLaunchDocument::Role(_) => {
parse_assistant_json::<RoleWorkerResultIngress>(assistant_text)
.map(WorkerResultIngress::Role)
}
EpiphanyWorkerLaunchDocument::Reorient(_) => {
parse_assistant_json::<ReorientWorkerResultIngress>(assistant_text)
.map(WorkerResultIngress::Reorient)
}
}
}
fn role_worker_result_from_ingress(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
role_id: &str,
repository_body_observation_basis: Option<&epiphany_core::RepositoryBodyObservationBasis>,
proposal_modeling_context: Option<
&epiphany_core::RepoFrontierProposalModelingContextProjection,
>,
frontier_planning_context: Option<&epiphany_core::RepoFrontierPlanningContextProjection>,
frontier_plan_mind_context: Option<&epiphany_core::RepoFrontierPlanMindContextProjection>,
imagination_consideration_context: Option<
&epiphany_core::ImaginationConsiderationContextProjection,
>,
admitted_model_direction_consideration_context: Option<
&epiphany_core::AdmittedModelDirectionConsiderationContextProjection,
>,
repo_frontier_verdict_modeling_authority: Option<
&epiphany_core::RepoFrontierVerdictModelingLaunchAuthority,
>,
completed_at: &str,
result_id: &str,
decision_context_id: &str,
result: &RoleWorkerResultIngress,
runtime_evidence_ids: Vec<String>,
artifact_refs: Vec<String>,
) -> EpiphanyRuntimeRoleWorkerResult {
let verification_authority = rmp_serde::from_slice::<EpiphanyWorkerLaunchDocument>(
&launch_request.launch_document_msgpack,
)
.ok()
.and_then(|document| match document {
EpiphanyWorkerLaunchDocument::Role(role) => role.frontier_verification_context,
EpiphanyWorkerLaunchDocument::Reorient(_) => None,
});
let accepted_evidence_ids = {
let mut evidence_ids = clean_string_vec(&result.evidence_ids);
evidence_ids.extend(clean_string_vec(&runtime_evidence_ids));
if let Some(context) = proposal_modeling_context {
evidence_ids.push(context.proposal_id.clone());
}
evidence_ids.sort();
evidence_ids.dedup();
evidence_ids
};
let repo_model_proposal = |operations| {
let body_basis = repository_body_observation_basis.cloned().ok_or_else(|| {
anyhow!("repoModelOperations: Modeling launch has no current Body basis")
})?;
epiphany_core::EpiphanyRepoModelMutationProposal::new(
format!("repo-model-mutation-proposal-{}", launch_request.job_id),
launch_request.job_id.clone(),
result_id.to_string(),
accepted_evidence_ids.clone(),
body_basis,
operations,
)
};
let (research_decision_msgpack, research_decision_error) =
encode_optional_document(&result.research_decision, "researchDecision");
let (repo_model_mutation_proposal_msgpack, repo_model_mutation_proposal_error) = if let Some(
context,
) =
proposal_modeling_context
{
match result.proposal_frontier_draft.as_ref() {
Some(draft) => {
let mut target_claim_ids = clean_string_vec(&draft.target_claim_ids);
target_claim_ids.sort();
target_claim_ids.dedup();
let mut repository_scope = clean_string_vec(&draft.repository_scope);
repository_scope.sort();
repository_scope.dedup();
let mut dependency_item_ids = clean_string_vec(&draft.dependency_item_ids);
dependency_item_ids.sort();
dependency_item_ids.dedup();
let mut evidence_refs = clean_string_vec(&draft.evidence_refs);
evidence_refs.push(context.proposal_id.clone());
evidence_refs.sort();
evidence_refs.dedup();
let frontier_id = format!(
"frontier-proposal-{:x}",
sha2::Sha256::digest(context.request_id.as_bytes())
);
let proposal = repo_model_proposal(vec![
epiphany_core::EpiphanyRepoModelMutationOperation::PutFrontier {
item: epiphany_core::RepoFrontierItem {
id: frontier_id,
migration_body: draft.migration_body.trim().to_string(),
question: draft.question.trim().to_string(),
gap: draft.gap.trim().to_string(),
target_claim_ids,
repository_scope,
recommended_next_organ: draft.recommended_next_organ.trim().to_string(),
adopted_plan: None,
dependency_item_ids,
status: epiphany_core::RepoFrontierStatus::Active,
evidence_refs,
public_source_refs: Vec::new(),
created_at: Some(completed_at.to_string()),
updated_at: Some(completed_at.to_string()),
retired_at: None,
superseded_by: None,
},
},
]);
match proposal {
Ok(proposal) => {
encode_optional_document(&Some(proposal), "proposalFrontierDraft")
}
Err(error) => (None, Some(format!("proposalFrontierDraft: {error}"))),
}
}
None => (
None,
Some("proposalFrontierDraft: missing semantic draft".to_string()),
),
}
} else if let Some(authority) = repo_frontier_verdict_modeling_authority {
match clean_optional_string(result.frontier_verdict_gap.as_deref()) {
Some(gap) => {
let mut item = authority.frontier_item.clone();
item.gap = gap;
item.status = match authority.request.allowed_disposition {
epiphany_core::RepoFrontierVerdictDisposition::Resolved => {
epiphany_core::RepoFrontierStatus::Resolved
}
epiphany_core::RepoFrontierVerdictDisposition::Blocked => {
epiphany_core::RepoFrontierStatus::Blocked
}
};
item.updated_at = Some(completed_at.to_string());
item.evidence_refs
.push(authority.request.verification_request_id.clone());
item.evidence_refs
.push(authority.request.soul_verdict_receipt_id.clone());
item.evidence_refs.sort();
item.evidence_refs.dedup();
match repo_model_proposal(vec![
epiphany_core::EpiphanyRepoModelMutationOperation::PutFrontier { item },
]) {
Ok(proposal) => encode_optional_document(&Some(proposal), "frontierVerdictGap"),
Err(error) => (None, Some(format!("frontierVerdictGap: {error}"))),
}
}
None => (
None,
Some("frontierVerdictGap: missing semantic verdict finding".to_string()),
),
}
} else if role_id.eq_ignore_ascii_case("modeling") && !result.repo_model_operations.is_empty() {
if result
.verdict
.as_deref()
.is_some_and(|verdict| verdict.eq_ignore_ascii_case("checkpoint-ready"))
&& result.repo_model_operations.iter().any(|operation| {
!matches!(
operation,
epiphany_core::EpiphanyRepoModelMutationOperation::CreateSurfaceOffer { .. }
| epiphany_core::EpiphanyRepoModelMutationOperation::DeprecateSurfaceOffer { .. }
| epiphany_core::EpiphanyRepoModelMutationOperation::WithdrawSurfaceOffer { .. }
| epiphany_core::EpiphanyRepoModelMutationOperation::CreateDependencyClaim { .. }
| epiphany_core::EpiphanyRepoModelMutationOperation::RetireDependencyClaim { .. }
)
})
{
(
None,
Some(
"repoModelOperations: checkpoint-ready may carry only local Atlas offer/claim transitions"
.to_string(),
),
)
} else {
let mut operations = result.repo_model_operations.clone();
for operation in &mut operations {
match operation {
epiphany_core::EpiphanyRepoModelMutationOperation::PutNode { node } => {
node.created_at = node
.created_at
.as_ref()
.map(|_| completed_at.to_string())
.or_else(|| Some(completed_at.to_string()));
node.updated_at = Some(completed_at.to_string());
}
epiphany_core::EpiphanyRepoModelMutationOperation::PutFrontier { item } => {
item.created_at = item
.created_at
.as_ref()
.map(|_| completed_at.to_string())
.or_else(|| Some(completed_at.to_string()));
item.updated_at = Some(completed_at.to_string());
item.retired_at = None;
item.superseded_by = None;
}
_ => {}
}
}
match repo_model_proposal(operations) {
Ok(proposal) => encode_optional_document(&Some(proposal), "repoModelOperations"),
Err(error) => (None, Some(format!("repoModelOperations: {error}"))),
}
}
} else if !role_id.eq_ignore_ascii_case("modeling") && !result.repo_model_operations.is_empty()
{
(
None,
Some("repoModelOperations: only Modeling may propose RepoModel mutations".to_string()),
)
} else {
(None, None)
};
let (frontier_plan_candidate_msgpack, frontier_plan_candidate_error) = if let Some(ingress) =
result.frontier_plan_candidate.as_ref()
{
let mut safe_paths = clean_string_vec(&ingress.safe_paths);
safe_paths.sort();
safe_paths.dedup();
let mut candidate = epiphany_core::RepoFrontierPlanCandidate {
candidate_id: String::new(),
planning_request_id: frontier_planning_context
.map(|context| context.request_id.clone())
.unwrap_or_default(),
model_projection_digest: frontier_planning_context
.map(|context| context.model_projection_digest.clone())
.unwrap_or_default(),
model_source_documents: frontier_planning_context
.map(|context| context.model_source_documents.clone())
.unwrap_or_default(),
frontier_item_id: frontier_planning_context
.map(|context| context.frontier_item_id.clone())
.unwrap_or_default(),
frontier_item_hash: frontier_planning_context
.map(|context| context.frontier_item_hash.clone())
.unwrap_or_default(),
safe_paths,
action: ingress.action.trim().to_string(),
command: ingress.command.trim().to_string(),
checks: clean_string_vec(&ingress.checks),
stop_conditions: clean_string_vec(&ingress.stop_conditions),
rollback_steps: clean_string_vec(&ingress.rollback_steps),
commit_message: ingress.commit_message.trim().to_string(),
proposed_at: completed_at.to_string(),
};
match epiphany_core::canonical_repo_frontier_plan_candidate_id(&candidate) {
Ok(candidate_id) => {
candidate.candidate_id = candidate_id;
encode_optional_document(&Some(candidate), "frontierPlanCandidate")
}
Err(error) => (None, Some(format!("frontierPlanCandidate: {error}"))),
}
} else {
(None, None)
};
let (frontier_plan_mind_decision_msgpack, frontier_plan_mind_decision_error) =
if let Some(ingress) = result.frontier_plan_mind_decision.as_ref() {
if let Some(decision) = ingress.decision {
match frontier_plan_mind_context {
Some(context) => encode_optional_document(
&Some(epiphany_core::RepoFrontierPlanMindDecision {
mind_request_id: context.request.request_id.clone(),
planning_request_id: context.planning_request.request_id.clone(),
imagination_result_id: context.request.imagination_result_id.clone(),
candidate_id: context.candidate.candidate_id.clone(),
candidate_sha256: context.request.candidate_sha256.clone(),
decision,
rationale: ingress.rationale.trim().into(),
decided_at: completed_at.into(),
}),
"frontierPlanMindDecision",
),
None => (
None,
Some(
"frontierPlanMindDecision: launch lacks immutable Mind context".into(),
),
),
}
} else {
(
None,
Some("frontierPlanMindDecision: missing decision".into()),
)
}
} else {
(None, None)
};
let (imagination_consideration_candidate_msgpack, imagination_consideration_candidate_error) =
if let Some(ingress) = result.imagination_consideration_candidate.as_ref() {
match (
imagination_consideration_context,
ingress.disposition,
ingress.recommended_review_route,
) {
(Some(context), Some(disposition), Some(route)) => encode_optional_document(
&Some(epiphany_core::ImaginationConsiderationCandidate {
schema_version:
epiphany_core::IMAGINATION_CONSIDERATION_CANDIDATE_SCHEMA_VERSION.into(),
candidate_id:
epiphany_core::imagination_consideration_candidate_id_for_launch(
&context.request.request_id,
&launch_request.job_id,
),
request_id: context.request.request_id.clone(),
feedback_id: context.request.feedback_id.clone(),
feedback_packet_sha256: context.request.feedback_packet_sha256.clone(),
source_room_id: context.request.source_room_id.clone(),
source_visibility: context.request.source_visibility.clone(),
data_classification: context.request.data_classification.clone(),
model_projection_digest: context
.request
.model_projection_digest
.clone(),
model_source_documents: context
.request
.model_source_documents
.clone(),
disposition,
title: ingress.title.trim().into(),
summary: ingress.summary.trim().into(),
rationale: ingress.rationale.trim().into(),
option_drafts: ingress.option_drafts.clone(),
uncertainties: clean_string_vec(&ingress.uncertainties),
evidence_refs: context.request.quoted_evidence.source_discussion_refs.clone(),
recommended_review_route: route,
proposed_at: completed_at.into(),
contract: epiphany_core::IMAGINATION_CONSIDERATION_CANDIDATE_CONTRACT.into(),
}),
"imaginationConsiderationCandidate",
),
(None, _, _) => (
None,
Some("imaginationConsiderationCandidate: launch lacks immutable consideration context".into()),
),
_ => (
None,
Some("imaginationConsiderationCandidate: missing disposition or route".into()),
),
}
} else {
(None, None)
};
let (
admitted_model_direction_consideration_result_msgpack,
admitted_model_direction_consideration_result_error,
) = if let Some(ingress) = result
.admitted_model_direction_consideration_result
.as_ref()
{
if let (Some(context), Some(disposition)) = (
admitted_model_direction_consideration_context,
ingress.disposition,
) {
encode_optional_document(
&Some(epiphany_core::AdmittedModelDirectionConsiderationResult {
schema_version:
epiphany_core::ADMITTED_MODEL_DIRECTION_CONSIDERATION_RESULT_SCHEMA_VERSION
.into(),
result_id:
epiphany_core::admitted_model_direction_consideration_result_id_for_launch(
&context.request.request_id,
&launch_request.job_id,
),
request_id: context.request.request_id.clone(),
runtime_id: context.request.runtime_id.clone(),
thread_id: context.request.thread_id.clone(),
model_projection_digest: context.request.model_projection_digest.clone(),
model_source_documents: context.request.model_source_documents.clone(),
disposition,
summary: ingress.summary.trim().into(),
option_drafts: ingress.option_drafts.clone(),
uncertainties: clean_string_vec(&ingress.uncertainties),
evidence_refs: clean_string_vec(&ingress.evidence_refs),
proposed_at: completed_at.into(),
contract: epiphany_core::ADMITTED_MODEL_DIRECTION_CONSIDERATION_RESULT_CONTRACT
.into(),
proposal_only: true,
terminal: true,
}),
"admittedModelDirectionConsiderationResult",
)
} else if admitted_model_direction_consideration_context.is_none() {
(
None,
Some("admittedModelDirectionConsiderationResult: launch lacks immutable consideration context".into()),
)
} else {
(
None,
Some("admittedModelDirectionConsiderationResult: missing disposition".into()),
)
}
} else {
(None, None)
};
EpiphanyRuntimeRoleWorkerResult {
repository_body_observation_basis: role_id
.eq_ignore_ascii_case("modeling")
.then(|| repository_body_observation_basis.cloned())
.flatten(),
result_id: result_id.to_string(),
job_id: launch_request.job_id.clone(),
role_id: clean_optional_string(result.role_id.as_deref())
.unwrap_or_else(|| role_id.to_string()),
verdict: clean_optional_string(result.verdict.as_deref())
.unwrap_or_else(|| "completed".to_string()),
summary: clean_optional_string(result.summary.as_deref())
.unwrap_or_else(|| "Worker completed without a structured summary.".to_string()),
next_safe_move: clean_optional_string(result.next_safe_move.as_deref()).unwrap_or_else(
|| "Review the typed worker runtime result before accepting state.".to_string(),
),
checkpoint_summary: clean_optional_string(result.checkpoint_summary.as_deref()),
scratch_summary: clean_optional_string(result.scratch_summary.as_deref()),
files_inspected: clean_string_vec(&result.files_inspected),
frontier_node_ids: clean_string_vec(&result.frontier_node_ids),
evidence_ids: accepted_evidence_ids,
artifact_refs,
open_questions: clean_string_vec(&result.open_questions),
evidence_gaps: clean_string_vec(&result.evidence_gaps),
risks: clean_string_vec(&result.risks),
research_decision_msgpack,
item_error: merge_optional_errors(
merge_optional_errors(research_decision_error, repo_model_mutation_proposal_error),
merge_optional_errors(
frontier_plan_candidate_error,
merge_optional_errors(
frontier_plan_mind_decision_error,
merge_optional_errors(
imagination_consideration_candidate_error,
admitted_model_direction_consideration_result_error,
),
),
),
),
metadata: std::collections::BTreeMap::new(),
repo_model_mutation_proposal_msgpack,
verification_request_id: verification_authority
.as_ref()
.map(|authority| authority.request.request_id.clone()),
frontier_route_id: verification_authority
.as_ref()
.map(|authority| authority.route.route_id.clone()),
repo_frontier_modeling_request_id: launch_request.repo_frontier_modeling_request_id.clone(),
proposal_modeling_request_id: launch_request.proposal_modeling_request_id.clone(),
repo_frontier_research_request_id: launch_request.repo_frontier_research_request_id.clone(),
frontier_planning_request_id: launch_request.frontier_planning_request_id.clone(),
frontier_plan_candidate_msgpack,
frontier_plan_mind_request_id: frontier_plan_mind_context
.map(|context| context.request.request_id.clone()),
frontier_plan_mind_decision_msgpack,
imagination_consideration_request_id: imagination_consideration_context
.map(|context| context.request.request_id.clone()),
imagination_consideration_candidate_msgpack,
admitted_model_direction_consideration_request_id:
admitted_model_direction_consideration_context
.map(|context| context.request.request_id.clone()),
admitted_model_direction_consideration_result_msgpack,
decision_context_id: decision_context_id.to_string(),
}
}
/// Projects a model-adapter failure into the typed faculty result consumed by
/// Self's frontier-planning lifecycle. The generic runtime job remains the
/// process/transport receipt; this document states only that the exact launched
/// faculty produced no executable candidate or Mind judgment.
pub fn failed_frontier_planning_role_result(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
failure_summary: &str,
decision_context_id: &str,
) -> Result<Option<EpiphanyRuntimeRoleWorkerResult>> {
if launch_request.frontier_planning_request_id.is_none()
&& launch_request.frontier_plan_mind_request_id.is_none()
{
return Ok(None);
}
if decision_context_id.trim().is_empty() {
return Err(anyhow!(
"model-backed planning failure requires its sealed decision context"
));
}
let document = launch_request.launch_document()?;
let EpiphanyWorkerLaunchDocument::Role(document) = document else {
return Err(anyhow!(
"frontier planning failure projection requires a role launch"
));
};
let summary = failure_summary.trim().to_string();
Ok(Some(EpiphanyRuntimeRoleWorkerResult {
result_id: format!("result-worker-{}", launch_request.job_id),
job_id: launch_request.job_id.clone(),
role_id: document.role_id,
verdict: "runtime-error".to_string(),
summary: summary.clone(),
next_safe_move: "Review the immutable planning failure before authorizing another attempt."
.to_string(),
checkpoint_summary: None,
scratch_summary: None,
files_inspected: Vec::new(),
frontier_node_ids: Vec::new(),
evidence_ids: Vec::new(),
artifact_refs: Vec::new(),
open_questions: Vec::new(),
evidence_gaps: Vec::new(),
risks: Vec::new(),
research_decision_msgpack: None,
item_error: Some(summary.clone()),
metadata: std::collections::BTreeMap::new(),
repo_model_mutation_proposal_msgpack: None,
verification_request_id: None,
frontier_route_id: None,
repo_frontier_modeling_request_id: None,
proposal_modeling_request_id: None,
repo_frontier_research_request_id: None,
frontier_planning_request_id: None,
frontier_plan_candidate_msgpack: None,
frontier_plan_mind_request_id: None,
frontier_plan_mind_decision_msgpack: None,
repository_body_observation_basis: document.repository_body_observation_basis,
imagination_consideration_request_id: None,
imagination_consideration_candidate_msgpack: None,
admitted_model_direction_consideration_request_id: None,
admitted_model_direction_consideration_result_msgpack: None,
decision_context_id: decision_context_id.to_string(),
}))
}
fn reorient_worker_result_from_ingress(
launch_request: &EpiphanyRuntimeWorkerLaunchRequest,
result_id: &str,
decision_context_id: &str,
result: &ReorientWorkerResultIngress,
artifact_refs: Vec<String>,
) -> EpiphanyRuntimeReorientWorkerResult {
EpiphanyRuntimeReorientWorkerResult {
result_id: result_id.to_string(),
job_id: launch_request.job_id.clone(),
mode: clean_optional_string(result.mode.as_deref())
.unwrap_or_else(|| "regather".to_string()),
summary: clean_optional_string(result.summary.as_deref()).unwrap_or_else(|| {
"Reorient worker completed without a structured summary.".to_string()
}),
next_safe_move: clean_optional_string(result.next_safe_move.as_deref()).unwrap_or_else(
|| "Review the typed reorient runtime result before accepting state.".to_string(),
),
checkpoint_still_valid: result.checkpoint_still_valid,
files_inspected: clean_string_vec(&result.files_inspected),
frontier_node_ids: clean_string_vec(&result.frontier_node_ids),
evidence_ids: clean_string_vec(&result.evidence_ids),
artifact_refs,
open_questions: clean_string_vec(&result.open_questions),
continuity_risks: clean_string_vec(&result.continuity_risks),
item_error: None,
metadata: std::collections::BTreeMap::new(),
decision_context_id: decision_context_id.to_string(),
}
}
fn encode_optional_document<T>(value: &Option<T>, key: &str) -> (Option<Vec<u8>>, Option<String>)
where
T: serde::Serialize,
{
let Some(document) = value else {
return (None, None);
};
match rmp_serde::to_vec_named(document) {
Ok(payload) => (Some(payload), None),
Err(err) => (None, Some(format!("failed to encode {key}: {err}"))),
}
}
fn merge_optional_errors(left: Option<String>, right: Option<String>) -> Option<String> {
match (left, right) {
(Some(left), Some(right)) => Some(format!("{left}; {right}")),
(Some(left), None) => Some(left),
(None, Some(right)) => Some(right),
(None, None) => None,
}
}
fn parse_assistant_json<T>(text: &str) -> Result<T>
where
T: DeserializeOwned,
{
let trimmed = text.trim();
let candidate = trimmed
.strip_prefix("```json")
.and_then(|value| value.strip_suffix("```"))
.or_else(|| {
trimmed
.strip_prefix("```")
.and_then(|value| value.strip_suffix("```"))
})
.unwrap_or(trimmed)
.trim();
let value = serde_json::from_str::<UniqueJsonValue>(candidate)
.context("assistant text was not typed worker-result JSON")?
.0;
serde_json::from_value(value).context("assistant text was not typed worker-result JSON")
}
fn deserialize_null_default<'de, D, T>(deserializer: D) -> Result<T, D::Error>
where
D: serde::Deserializer<'de>,
T: Deserialize<'de> + Default,
{
Ok(Option::<T>::deserialize(deserializer)?.unwrap_or_default())
}
struct UniqueJsonValue(serde_json::Value);
impl<'de> Deserialize<'de> for UniqueJsonValue {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
deserializer.deserialize_any(UniqueJsonValueVisitor)
}
}
struct UniqueJsonValueVisitor;
impl<'de> Visitor<'de> for UniqueJsonValueVisitor {
type Value = UniqueJsonValue;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("JSON without duplicate object keys")
}
fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::Bool(value)))
}
fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::Number(value.into())))
}
fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::Number(value.into())))
}
fn visit_f64<E>(self, value: f64) -> Result<Self::Value, E>
where
E: serde::de::Error,
{
serde_json::Number::from_f64(value)
.map(serde_json::Value::Number)
.map(UniqueJsonValue)
.ok_or_else(|| E::custom("JSON number is not finite"))
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::String(
value.to_string(),
)))
}
fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::String(value)))
}
fn visit_none<E>(self) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::Null))
}
fn visit_unit<E>(self) -> Result<Self::Value, E> {
Ok(UniqueJsonValue(serde_json::Value::Null))
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
let mut values = Vec::new();
while let Some(value) = sequence.next_element::<UniqueJsonValue>()? {
values.push(value.0);
}
Ok(UniqueJsonValue(serde_json::Value::Array(values)))
}
fn visit_map<A>(self, mut object: A) -> Result<Self::Value, A::Error>
where
A: MapAccess<'de>,
{
let mut values = serde_json::Map::new();
while let Some((key, value)) = object.next_entry::<String, UniqueJsonValue>()? {
if values.insert(key.clone(), value.0).is_some() {
return Err(serde::de::Error::custom(format!("duplicate field `{key}`")));
}
}
Ok(UniqueJsonValue(serde_json::Value::Object(values)))
}
}
fn clean_optional_string(value: Option<&str>) -> Option<String> {
value
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string)
}
fn clean_string_vec(values: &[String]) -> Vec<String> {
values
.iter()
.map(|value| value.trim())
.filter(|value| !value.is_empty())
.map(ToString::to_string)
.collect()
}
fn sanitize_request_id(value: &str) -> String {
value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
ch
} else {
'-'
}
})
.collect()
}
fn now() -> String {
chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true)
}
#[cfg(test)]
mod tests {
use super::*;
use epiphany_core::EpiphanyWorkerLaunchDocument;
use epiphany_core::RuntimeSpineHeartbeatJobOptions;
use epiphany_core::runtime_job_snapshot;
use epiphany_openai_adapter::EpiphanyOpenAiModelReceipt;
use sha2::{Digest, Sha256};
use tempfile::tempdir;
fn seed_test_runtime_job(store: &Path, options: RuntimeSpineHeartbeatJobOptions) -> Result<()> {
epiphany_core::initialize_runtime_spine(
store,
epiphany_core::RuntimeSpineInitOptions {
runtime_id: options.runtime_id.clone(),
display_name: "Epiphany Test".into(),
created_at: options.created_at.clone(),
},
)?;
let mut cache = epiphany_core::runtime_spine_cache(store)?;
cache.pull_all_backing_stores()?;
let prepared = epiphany_core::prepare_runtime_spine_heartbeat_job(&cache, options)?;
cache.put_prepared_batch(prepared)?;
Ok(())
}
fn run_git(repo: &std::path::Path, args: &[&str]) -> Result<()> {
let output = std::process::Command::new("git")
.arg("-C")
.arg(repo)
.args(args)
.output()?;
if !output.status.success() {
return Err(anyhow!(
"git {:?} failed: {}",
args,
String::from_utf8_lossy(&output.stderr)
));
}
Ok(())
}
fn test_reasoning_basis(
launch: &EpiphanyRuntimeWorkerLaunchRequest,
) -> Result<epiphany_core::EpiphanyReasoningBasis> {
let EpiphanyWorkerLaunchDocument::Role(document) = launch.launch_document()? else {
return Err(anyhow!("test fixture requires a role launch"));
};
let payload_msgpack = rmp_serde::to_vec_named(&epiphany_core::EpiphanyMindIdentity {
schema_epoch: epiphany_core::MIND_SCHEMA_EPOCH.into(),
runtime_id: "test-runtime".into(),
})?;
let source_documents = vec![epiphany_core::EpiphanyMindDocumentVersion {
store_id: "epiphany-mind".into(),
document_type: "epiphany.mind.identity.v1".into(),
document_key: epiphany_core::MIND_SCHEMA_EPOCH.into(),
schema_id: None,
payload_sha256: format!("sha256:{:x}", Sha256::digest(&payload_msgpack)),
payload_msgpack,
}];
let mind = epiphany_core::EpiphanyMindPromptProjection {
schema_epoch: epiphany_core::MIND_SCHEMA_EPOCH.into(),
runtime_id: "test-runtime".into(),
projection_digest: epiphany_core::epiphany_mind_projection_digest(&source_documents)?,
objective: None,
active_subgoal_id: None,
subgoals: Vec::new(),
invariants: Vec::new(),
observations: Vec::new(),
evidence: Vec::new(),
verification_audits: Vec::new(),
reorientation_decisions: Vec::new(),
reorientation_failures: Vec::new(),
agent_pass_admission_refusals: Vec::new(),
investigation_checkpoint: None,
mode: None,
repository_body_observation: None,
repo_model: None,
};
epiphany_core::EpiphanyReasoningBasis::new(
&launch.job_id,
&launch.role,
epiphany_core::WORKER_REASONING_PROJECTION_POLICY,
source_documents,
epiphany_core::EpiphanyReasoningProjection::RolePass(
epiphany_core::EpiphanyRoleReasoningProjection {
authority: document.into(),
mind,
modeling_body: None,
},
),
)
}
#[test]
fn model_authored_body_basis_is_refused_at_ingress() {
let error = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{
"roleId":"modeling","verdict":"checkpoint-ready","summary":"mapped",
"nextSafeMove":"Mind review","repositoryBodyObservationBasis":{
"schemaVersion":"epiphany.repository_body.v2","workspaceId":"workspace-1",
"swarmId":"swarm-1","runtimeId":"runtime-1","scope":"git_worktree",
"bodyBindingSha256":"binding-hash","observationId":"workspace-1:1",
"generation":1,"manifestRootSha256":"manifest-root",
"scanStartedAt":"2026-07-15T00:00:00Z",
"scanFinishedAt":"2026-07-15T00:00:01Z"
}
}"#,
)
.expect_err("Body identity is runtime-owned and cannot be laundered by omission");
assert!(format!("{error:#}").contains("unknown field `repositoryBodyObservationBasis`"));
}
#[test]
fn provider_null_optionals_are_inverse_projected_to_canonical_omission() -> Result<()> {
let parsed = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{
"roleId":"research","verdict":"source-gap","summary":"bounded",
"nextSafeMove":"review","filesInspected":[],"frontierNodeIds":null,
"evidenceIds":[],"artifactRefs":null,"openQuestions":null,
"evidenceGaps":["missing"],"risks":null,"researchDecision":null
}"#,
)?;
assert_eq!(parsed.role_id.as_deref(), Some("research"));
assert!(parsed.frontier_node_ids.is_empty());
assert!(parsed.artifact_refs.is_empty());
assert!(parsed.research_decision.is_none());
assert_eq!(parsed.evidence_gaps, vec!["missing"]);
Ok(())
}
#[test]
fn every_static_worker_contract_projects_to_one_strict_provider_shape() -> Result<()> {
let schemas = vec![
epiphany_core::epiphany_role_launch_output_schema(
epiphany_core::EpiphanyRoleResultRoleId::Imagination,
),
epiphany_core::epiphany_role_launch_output_schema(
epiphany_core::EpiphanyRoleResultRoleId::Research,
),
epiphany_core::epiphany_role_launch_output_schema(
epiphany_core::EpiphanyRoleResultRoleId::Modeling,
),
epiphany_core::epiphany_role_launch_output_schema(
epiphany_core::EpiphanyRoleResultRoleId::Verification,
),
epiphany_core::epiphany_frontier_planning_output_schema(),
epiphany_core::epiphany_frontier_plan_mind_output_schema(),
epiphany_core::epiphany_imagination_consideration_output_schema(),
epiphany_core::epiphany_admitted_model_direction_consideration_output_schema(),
epiphany_core::epiphany_proposal_modeling_output_schema(),
epiphany_core::epiphany_reorient_launch_output_schema(),
];
for (index, schema) in schemas.into_iter().enumerate() {
let mut request = EpiphanyOpenAiModelRequest::new(
format!("strict-worker-schema-{index}"),
"strict-worker-schema",
"gpt-5.4",
"Return the typed result.",
);
request.output_contract_id = Some(format!("worker-schema-{index}"));
request.output_schema_json = Some(serde_json::to_string(&schema)?);
let body = epiphany_openai_codex_spine::responses_body_from_epiphany(request)?;
assert_eq!(body["text"]["format"]["strict"], true);
}
Ok(())
}
#[test]
fn research_contract_projects_typed_decision_to_provider_strict_shape() -> Result<()> {
let mut request = EpiphanyOpenAiModelRequest::new(
"strict-research-schema",
"strict-worker-schema",
"gpt-5.4",
"Return the typed result.",
);
request.output_contract_id =
Some(epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.to_string());
request.output_schema_json = Some(serde_json::to_string(
&epiphany_core::epiphany_role_launch_output_schema(
epiphany_core::EpiphanyRoleResultRoleId::Research,
),
)?);
let body = epiphany_openai_codex_spine::responses_body_from_epiphany(request)?;
let decision = &body["text"]["format"]["schema"]["properties"]["researchDecision"];
assert!(decision.get("anyOf").is_none());
assert_eq!(decision["additionalProperties"], false);
assert_eq!(
decision["required"],
serde_json::json!(["evidence", "investigationCheckpoint", "observations"])
);
assert!(decision["properties"].get("scratch").is_none());
assert_eq!(
decision["properties"]["investigationCheckpoint"]["anyOf"][0]["additionalProperties"],
false
);
Ok(())
}
#[test]
fn role_ingress_rejects_duplicate_top_level_fields() {
let error = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{"roleId":"modeling","verdict":"checkpoint-ready","summary":"mapped","nextSafeMove":"review","checkpointSummary":"first","checkpointSummary":"second"}"#,
)
.expect_err("duplicate typed fields must remain ambiguous and fail closed");
assert!(format!("{error:#}").contains("duplicate field `checkpointSummary`"));
}
#[test]
fn proposal_modeling_ingress_composes_runtime_owned_patch_and_identity() -> Result<()> {
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "proposal-job-1".into(),
binding_id: epiphany_core::EPIPHANY_MODELING_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_MODELING_OWNER_ROLE.into(),
authority_scope: "epiphany.role.modeling".into(),
instruction: "model proposal".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: Vec::new(),
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: Some("proposal-request-1".into()),
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let basis = epiphany_core::RepositoryBodyObservationBasis {
schema_version: "epiphany.repository_body.v2".into(),
workspace_id: "workspace-1".into(),
swarm_id: "swarm-1".into(),
runtime_id: "runtime-1".into(),
scope: "git_worktree".into(),
body_binding_sha256: "binding-1".into(),
observation_id: "observation-1".into(),
generation: 1,
manifest_root_sha256: "manifest-1".into(),
scan_started_at: "2026-08-11T08:00:00Z".into(),
scan_finished_at: "2026-08-11T08:00:01Z".into(),
};
let context = epiphany_core::RepoFrontierProposalModelingContextProjection {
schema_version: epiphany_core::REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_SCHEMA_VERSION
.into(),
contract: epiphany_core::REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_CONTRACT.into(),
request_id: "proposal-request-1".into(),
proposal_id: "proposal-1".into(),
proposal_payload_sha256: "payload-1".into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
repository: "GameCult/Epiphany".into(),
workspace: "/workspace".into(),
title: "Map lifecycle".into(),
body: "Inspect typed state".into(),
constraints: Vec::new(),
evidence_refs: vec!["git:source".into()],
model_projection_digest: "model-projection-41".into(),
model_source_documents: Vec::new(),
prior_admission_refusals: Vec::new(),
};
let ingress = RoleWorkerResultIngress {
role_id: Some("modeling".into()),
verdict: Some("checkpoint-ready".into()),
summary: Some("Typed lifecycle is grant-owned.".into()),
next_safe_move: Some("Mind reviews the bounded frontier.".into()),
files_inspected: vec!["epiphany-core/src/resident_self.rs".into()],
frontier_node_ids: vec!["claim-grant-owned".into()],
evidence_ids: vec!["tool:grant-lifecycle".into()],
proposal_frontier_draft: Some(ProposalFrontierDraftIngress {
migration_body: "Resident grant lifecycle".into(),
question: "Does exact grant state own launchability?".into(),
gap: "Broader autonomous turnover remains open.".into(),
target_claim_ids: vec!["claim-grant-owned".into()],
repository_scope: vec!["epiphany-core/src/resident_self.rs".into()],
recommended_next_organ: "Eyes".into(),
dependency_item_ids: Vec::new(),
evidence_refs: vec!["tool:grant-lifecycle".into()],
}),
..Default::default()
};
let result = role_worker_result_from_ingress(
&launch,
"modeling",
Some(&basis),
Some(&context),
None,
None,
None,
None,
None,
"2026-08-11T08:00:02Z",
"result-proposal-job-1",
"decision-context-test",
&ingress,
vec!["openai-request:proposal".into()],
Vec::new(),
);
assert_eq!(result.repository_body_observation_basis, Some(basis));
assert_eq!(
result.proposal_modeling_request_id.as_deref(),
Some("proposal-request-1")
);
assert!(result.evidence_ids.contains(&"proposal-1".to_string()));
let proposal: epiphany_core::EpiphanyRepoModelMutationProposal = rmp_serde::from_slice(
result
.repo_model_mutation_proposal_msgpack
.as_deref()
.expect("runtime-composed proposal"),
)?;
assert_eq!(
proposal.proposal_id,
"repo-model-mutation-proposal-proposal-job-1"
);
let operations = proposal.operations()?;
let epiphany_core::EpiphanyRepoModelMutationOperation::PutFrontier { item } =
&operations[0]
else {
panic!("proposal draft must compose one frontier upsert");
};
assert_eq!(item.status, epiphany_core::RepoFrontierStatus::Active);
assert!(item.adopted_plan.is_none());
assert!(item.evidence_refs.contains(&"proposal-1".to_string()));
assert!(item.public_source_refs.is_empty());
Ok(())
}
#[test]
fn checkpoint_ready_modeling_carries_only_semantic_atlas_transitions() -> Result<()> {
let body = epiphany_core::RepositoryBodyObservationBasis {
schema_version: "epiphany.repository_body.v2".into(),
workspace_id: "eve".into(),
swarm_id: "gamecult-local".into(),
runtime_id: "atlas-runtime".into(),
scope: ".".into(),
body_binding_sha256: "0".repeat(64),
observation_id: "eve:1".into(),
generation: 1,
manifest_root_sha256: "1".repeat(64),
scan_started_at: "2026-08-14T00:00:00Z".into(),
scan_finished_at: "2026-08-14T00:00:01Z".into(),
};
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "atlas-modeling-job".into(),
binding_id: epiphany_core::EPIPHANY_MODELING_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_MODELING_OWNER_ROLE.into(),
authority_scope: "epiphany.role.modeling".into(),
instruction: "Model local Atlas facts.".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: Vec::new(),
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let atlas = RoleWorkerResultIngress {
role_id: Some("modeling".into()),
verdict: Some("checkpoint-ready".into()),
repo_model_operations: vec![
epiphany_core::EpiphanyRepoModelMutationOperation::CreateSurfaceOffer {
label: "Canonical Eve surface".into(),
contract: epiphany_core::AtlasContractDescriptor::ExactSchema {
contract_id: "eve-surface".into(),
schema_id: "gamecult.eve.surface.v1".into(),
},
source_refs: vec!["packages/eve-contracts/schema.rs".into()],
},
],
..Default::default()
};
let accepted = role_worker_result_from_ingress(
&launch,
"modeling",
Some(&body),
None,
None,
None,
None,
None,
None,
"2026-08-14T00:00:00Z",
"atlas-result",
"atlas-decision-context",
&atlas,
vec!["openai-request:atlas-test".into()],
Vec::new(),
);
assert!(accepted.item_error.is_none());
let proposal = accepted
.repo_model_mutation_proposal()?
.expect("semantic Atlas proposal");
assert_eq!(proposal.causal_request_id, launch.job_id);
assert_eq!(proposal.causal_result_id, "atlas-result");
assert_eq!(proposal.evidence_ids, ["openai-request:atlas-test"]);
assert_eq!(proposal.repository_body_observation_basis, body);
assert!(matches!(
proposal.operations()?.as_slice(),
[epiphany_core::EpiphanyRepoModelMutationOperation::CreateSurfaceOffer { .. }]
));
let hostile = RoleWorkerResultIngress {
role_id: Some("modeling".into()),
verdict: Some("checkpoint-ready".into()),
repo_model_operations: vec![
epiphany_core::EpiphanyRepoModelMutationOperation::PutFrontier {
item: epiphany_core::RepoFrontierItem::default(),
},
],
..Default::default()
};
let refused = role_worker_result_from_ingress(
&launch,
"modeling",
Some(&body),
None,
None,
None,
None,
None,
None,
"2026-08-14T00:00:00Z",
"hostile-result",
"hostile-decision-context",
&hostile,
Vec::new(),
Vec::new(),
);
assert!(refused.repo_model_mutation_proposal_msgpack.is_none());
assert!(
refused
.item_error
.as_deref()
.is_some_and(|error| error.contains("only local Atlas"))
);
Ok(())
}
#[test]
fn proposal_modeling_request_carries_narrow_schema_only_as_response_format() -> Result<()> {
let context = epiphany_core::RepoFrontierProposalModelingContextProjection {
schema_version: epiphany_core::REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_SCHEMA_VERSION
.into(),
contract: epiphany_core::REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_CONTRACT.into(),
request_id: "proposal-request-1".into(),
proposal_id: "proposal-1".into(),
proposal_payload_sha256: "payload-1".into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
repository: "GameCult/Epiphany".into(),
workspace: "/workspace".into(),
title: "Map lifecycle".into(),
body: "Inspect typed state".into(),
constraints: Vec::new(),
evidence_refs: vec!["git:source".into()],
model_projection_digest: "model-projection-41".into(),
model_source_documents: Vec::new(),
prior_admission_refusals: Vec::new(),
};
let document =
EpiphanyWorkerLaunchDocument::Role(epiphany_core::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-1".into(),
role_id: "modeling".into(),
objective: Some("obsolete aggregate objective must not cross inference".into()),
dynamic_prompt_context: None,
repository_body_observation_basis: None,
proposal_modeling_context: Some(context),
frontier_verdict_modeling_context: None,
frontier_planning_context: None,
frontier_research_context: None,
frontier_verification_context: None,
frontier_plan_mind_context: None,
imagination_consideration_context: None,
admitted_model_direction_consideration_context: None,
});
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "proposal-job-1".into(),
binding_id: epiphany_core::EPIPHANY_MODELING_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_MODELING_OWNER_ROLE.into(),
authority_scope: "epiphany.role.modeling".into(),
instruction: "Model the exact proposal.".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: rmp_serde::to_vec_named(&document)?,
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: Some("proposal-request-1".into()),
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let request = build_worker_model_request(
&launch,
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
&test_reasoning_basis(&launch)?,
)?;
let schema: serde_json::Value = serde_json::from_str(
request
.output_schema_json
.as_deref()
.expect("proposal response schema"),
)?;
assert_eq!(schema["additionalProperties"], false);
assert!(schema["properties"].get("proposalFrontierDraft").is_some());
assert!(schema["properties"].get("repoModelPatch").is_none());
assert!(!request.instructions.contains("Output schema JSON"));
assert!(
request
.instructions
.contains("Emit only the semantic frontier draft")
);
let rendered_input = serde_json::to_string(&request.input)?;
assert!(rendered_input.contains("sealed Epiphany reasoning projection"));
assert!(!rendered_input.contains("obsolete aggregate objective"));
assert!(!rendered_input.contains("stateRevision"));
Ok(())
}
#[test]
fn verification_ingress_refuses_model_authored_request_or_route_identity() {
let error = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{"roleId":"verification","verdict":"pass","summary":"verified","nextSafeMove":"admit","verificationRequestId":" verification-request-1 ","frontierRouteId":" frontier-route-1 "}"#,
)
.expect_err("causal identity is runtime-owned and must be refused at ingress");
assert!(format!("{error:#}").contains("unknown field"));
}
#[test]
fn frontier_planning_ingress_derives_typed_candidate_identity() -> Result<()> {
let parsed = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{
"roleId":"imagination",
"verdict":"draft-ready",
"summary":"bounded plan",
"nextSafeMove":"Mind admission",
"frontierPlanCandidate":{
"safe_paths":["tests","src","src"],
"action":"Implement the bounded cut.",
"command":"cargo test --lib",
"checks":["focused test passes"],
"stop_conditions":["scope changes"],
"rollback_steps":["revert commit"],
"commit_message":"Implement bounded cut"
}
}"#,
)?;
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "planning-job-1".into(),
binding_id: epiphany_core::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_IMAGINATION_OWNER_ROLE.into(),
authority_scope: "epiphany.role.imagination".into(),
instruction: "plan".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: Vec::new(),
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: None,
frontier_planning_request_id: Some("planning-request-1".into()),
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let planning_context = epiphany_core::RepoFrontierPlanningContextProjection {
schema_version: epiphany_core::REPO_FRONTIER_PLANNING_CONTEXT_SCHEMA_VERSION.into(),
contract: epiphany_core::REPO_FRONTIER_PLANNING_CONTEXT_CONTRACT.into(),
request_id: "planning-request-1".into(),
model_projection_digest: "model-projection-7".into(),
model_source_documents: Vec::new(),
frontier_authority_documents: Vec::new(),
claim_obligation_documents: Vec::new(),
frontier_item_id: "frontier-1".into(),
frontier_item_hash: "frontier-hash".into(),
selected_organ: "Imagination".into(),
repository_scope: vec!["src".into(), "tests".into()],
requested_at: "2026-07-15T09:59:00Z".into(),
runtime_id: "runtime-1".into(),
};
let result = role_worker_result_from_ingress(
&launch,
"imagination",
None,
None,
Some(&planning_context),
None,
None,
None,
None,
"2026-07-15T10:00:00Z",
"planning-result-1",
"decision-context-test",
&parsed,
Vec::new(),
Vec::new(),
);
assert_eq!(
result.frontier_planning_request_id.as_deref(),
Some("planning-request-1")
);
assert!(result.research_decision_msgpack.is_none());
let candidate = result
.frontier_plan_candidate()?
.expect("typed frontier candidate");
assert_eq!(candidate.planning_request_id, "planning-request-1");
assert_eq!(
candidate.candidate_id,
epiphany_core::canonical_repo_frontier_plan_candidate_id(&candidate)?
);
assert_eq!(candidate.safe_paths, ["src", "tests"]);
Ok(())
}
#[test]
fn consideration_ingress_derives_all_causal_identity_from_launch_context() -> Result<()> {
let hostile = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{
"roleId":"imagination","verdict":"suggest","summary":"bounded option",
"nextSafeMove":"Modeling review only",
"imaginationConsiderationCandidate":{
"request_id":"hostile-model-request","title":"Bounded option"
}
}"#,
)
.expect_err("nested model-authored causal identity must be refused");
assert!(format!("{hostile:#}").contains("unknown field `request_id`"));
let parsed = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{
"roleId":"imagination",
"verdict":"suggest",
"summary":"bounded option",
"nextSafeMove":"Modeling review only",
"filesInspected":[],
"imaginationConsiderationCandidate":{
"disposition":"suggest",
"title":"Bounded option",
"summary":"Keep the proposal review-only.",
"rationale":"The option is reversible.",
"option_drafts":[{"title":"Typed sight","summary":"Expose operator-safe provenance."}],
"uncertainties":[],
"recommended_review_route":"modeling_review"
}
}"#,
)?;
let request = epiphany_core::ImaginationConsiderationRequest {
schema_version:
epiphany_core::IMAGINATION_CONSIDERATION_REQUEST_SCHEMA_VERSION.into(),
request_id: "request-1".into(),
feedback_id: "feedback-1".into(),
feedback_admission_id: "admission-1".into(),
feedback_packet_sha256: "sha256-packet-1".into(),
source_room_id: "room-1".into(),
source_visibility: "public".into(),
data_classification: "public_feedback".into(),
source_provider_identity_id: "provider-1".into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
repository: "GameCult/Epiphany".into(),
persona_id: "epiphany".into(),
model_projection_digest: "model-projection-7".into(),
model_source_documents: Vec::new(),
routing_policy_id: "resident-feedback-consideration-v0".into(),
question: epiphany_core::ImaginationConsiderationQuestion::CompareWithCurrentBodyAndSuggestCoherentOptions,
quoted_evidence: epiphany_core::QuotedPersonaFeedbackEvidence {
feedback_text: "Quoted feedback only.".into(),
source_discussion_refs: vec!["discord://message-1".into()],
source_room_id: "room-1".into(),
source_visibility: "public".into(),
data_classification: "public_feedback".into(),
source_actor_id: "actor-1".into(),
source_provider: "bifrost".into(),
},
requested_at: "2026-07-15T09:59:00Z".into(),
contract: epiphany_core::IMAGINATION_CONSIDERATION_REQUEST_CONTRACT.into(),
private_state_included: false,
};
let context = epiphany_core::ImaginationConsiderationContextProjection {
schema_version: "epiphany.worker.imagination_consideration_context.v0".into(),
contract: "epiphany.imagination_consideration_context.v0".into(),
request: request.clone(),
model: epiphany_core::EpiphanyRepoModelView {
identity: epiphany_core::EpiphanyRepoModelIdentityDocument {
schema_epoch: epiphany_core::REPO_MODEL_SCHEMA_EPOCH.into(),
graph_id: "repo-model".into(),
runtime_id: "runtime-1".into(),
swarm_id: "swarm-1".into(),
workspace_id: "workspace-1".into(),
body_binding_sha256: "body-binding".into(),
},
projection_digest: request.model_projection_digest.clone(),
source_documents: Vec::new(),
domains: Vec::new(),
nodes: Vec::new(),
edges: Vec::new(),
frontier: Vec::new(),
claim_obligations: Vec::new(),
surface_offers: Vec::new(),
dependency_claims: Vec::new(),
dependency_verifications: Vec::new(),
dependency_impacts: Vec::new(),
},
};
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "consideration-job-1".into(),
binding_id: epiphany_core::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_IMAGINATION_OWNER_ROLE.into(),
authority_scope: "epiphany.imagination.consideration.proposal_only".into(),
instruction: "consider".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: Vec::new(),
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: Some(request.request_id.clone()),
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let result = role_worker_result_from_ingress(
&launch,
"imagination",
None,
None,
None,
None,
Some(&context),
None,
None,
"2026-07-15T10:00:00Z",
"consideration-result-1",
"decision-context-test",
&parsed,
Vec::new(),
Vec::new(),
);
assert_eq!(
result.imagination_consideration_request_id.as_deref(),
Some("request-1")
);
let candidate = result
.imagination_consideration_candidate()?
.expect("runtime-composed candidate");
assert_eq!(candidate.request_id, request.request_id);
assert_eq!(candidate.feedback_id, request.feedback_id);
assert_eq!(
candidate.feedback_packet_sha256,
request.feedback_packet_sha256
);
assert_eq!(candidate.source_room_id, request.source_room_id);
assert_eq!(candidate.source_visibility, request.source_visibility);
assert_eq!(candidate.data_classification, request.data_classification);
assert_eq!(
candidate.model_projection_digest,
request.model_projection_digest
);
assert_eq!(
candidate.model_source_documents,
request.model_source_documents
);
assert_eq!(
candidate.evidence_refs,
request.quoted_evidence.source_discussion_refs
);
assert_eq!(candidate.proposed_at, "2026-07-15T10:00:00Z");
assert_eq!(
candidate.contract,
epiphany_core::IMAGINATION_CONSIDERATION_CANDIDATE_CONTRACT
);
Ok(())
}
#[test]
fn frontier_planning_runtime_error_projects_non_executable_typed_failure() -> Result<()> {
let document =
EpiphanyWorkerLaunchDocument::Role(epiphany_core::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-1".into(),
role_id: "imagination".into(),
objective: Some("Plan one frontier.".into()),
dynamic_prompt_context: None,
repository_body_observation_basis: None,
proposal_modeling_context: None,
frontier_verdict_modeling_context: None,
frontier_planning_context: None,
frontier_research_context: None,
frontier_verification_context: None,
frontier_plan_mind_context: None,
imagination_consideration_context: None,
admitted_model_direction_consideration_context: None,
});
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "planning-job-failed".into(),
binding_id: epiphany_core::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_IMAGINATION_OWNER_ROLE.into(),
authority_scope: "epiphany.role.imagination".into(),
instruction: "plan".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: rmp_serde::to_vec_named(&document)?,
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: None,
frontier_planning_request_id: Some("planning-request-1".into()),
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let failed = failed_frontier_planning_role_result(
&launch,
"candidate mismatch",
"decision-context-planning-failure",
)?
.expect("typed planning failure");
assert_eq!(failed.job_id, launch.job_id);
assert_eq!(failed.role_id, "imagination");
assert_eq!(failed.item_error.as_deref(), Some("candidate mismatch"));
assert!(failed.frontier_planning_request_id.is_none());
assert!(failed.frontier_plan_candidate_msgpack.is_none());
assert!(failed.research_decision_msgpack.is_none());
assert!(failed.repo_model_mutation_proposal_msgpack.is_none());
assert_eq!(
failed.decision_context_id,
"decision-context-planning-failure"
);
assert!(failed_frontier_planning_role_result(&launch, "candidate mismatch", "").is_err());
Ok(())
}
#[test]
fn mind_ingress_derives_immutable_identity_from_launch_context() -> Result<()> {
let parsed = parse_assistant_json::<RoleWorkerResultIngress>(
r#"{"roleId":"mindAdmissionReview","verdict":"adopt","summary":"bounded","nextSafeMove":"admit","filesInspected":[],"frontierPlanMindDecision":{"decision":"adopt","rationale":"Exact candidate is bounded and falsifiable."}}"#,
)?;
let launch = EpiphanyRuntimeWorkerLaunchRequest {
job_id: "mind-job-1".into(),
binding_id: epiphany_core::EPIPHANY_MIND_ROLE_BINDING_ID.into(),
role: epiphany_core::EPIPHANY_MIND_OWNER_ROLE.into(),
authority_scope: "epiphany.procedure.mind_admission_review".into(),
instruction: "judge".into(),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: Vec::new(),
metadata: std::collections::BTreeMap::new(),
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: Some("mind-request-1".into()),
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
};
let context = epiphany_core::RepoFrontierPlanMindContextProjection {
schema_version: epiphany_core::REPO_FRONTIER_PLAN_MIND_CONTEXT_SCHEMA_VERSION.into(),
contract: epiphany_core::REPO_FRONTIER_PLAN_MIND_CONTEXT_CONTRACT.into(),
request: epiphany_core::RepoFrontierPlanMindRequest {
request_id: "mind-request-1".into(),
planning_request_id: "planning-request-1".into(),
imagination_result_id: "imagination-result-1".into(),
imagination_job_id: "imagination-job-1".into(),
candidate_id: "candidate-1".into(),
candidate_sha256: "candidate-hash-1".into(),
runtime_id: "runtime-1".into(),
requested_at: "2026-07-15T09:59:00Z".into(),
},
planning_request: epiphany_core::RepoFrontierPlanningRequest {
request_id: "planning-request-1".into(),
model_projection_digest: "model-projection-1".into(),
model_source_documents: Vec::new(),
frontier_authority_documents: Vec::new(),
claim_obligation_documents: Vec::new(),
frontier_item_id: "frontier-1".into(),
frontier_item_hash: "frontier-hash-1".into(),
selected_organ: "Imagination".into(),
repository_scope: vec!["epiphany-openai-runtime".into()],
requested_at: "2026-07-15T09:58:00Z".into(),
runtime_id: "runtime-1".into(),
},
candidate: epiphany_core::RepoFrontierPlanCandidate {
candidate_id: "candidate-1".into(),
planning_request_id: "planning-request-1".into(),
model_projection_digest: "model-projection-1".into(),
model_source_documents: Vec::new(),
frontier_item_id: "frontier-1".into(),
frontier_item_hash: "frontier-hash-1".into(),
safe_paths: vec!["epiphany-openai-runtime".into()],
action: "repair".into(),
command: "cargo test".into(),
checks: vec!["focused test".into()],
stop_conditions: vec!["identity mismatch".into()],
rollback_steps: vec!["revert".into()],
commit_message: "Derive Mind identity".into(),
proposed_at: "2026-07-15T09:59:00Z".into(),
},
};
let result = role_worker_result_from_ingress(
&launch,
"mindAdmissionReview",
None,
None,
None,
Some(&context),
None,
None,
None,
"2026-07-15T10:00:00Z",
"mind-result-1",
"decision-context-test",
&parsed,
Vec::new(),
Vec::new(),
);
assert_eq!(
result.frontier_plan_mind_request_id.as_deref(),
Some("mind-request-1")
);
let decision = result
.frontier_plan_mind_decision()?
.expect("typed Mind decision");
assert_eq!(
decision.decision,
epiphany_core::RepoFrontierPlanDecision::Adopt
);
assert_eq!(decision.candidate_sha256, "candidate-hash-1");
assert_eq!(decision.mind_request_id, "mind-request-1");
assert_eq!(decision.planning_request_id, "planning-request-1");
assert_eq!(decision.imagination_result_id, "imagination-result-1");
assert_eq!(decision.candidate_id, "candidate-1");
assert_eq!(decision.decided_at, "2026-07-15T10:00:00Z");
assert!(result.research_decision_msgpack.is_none());
assert!(result.repo_model_mutation_proposal_msgpack.is_none());
Ok(())
}
fn test_openai_event(
request_id: &str,
sequence: u64,
payload: EpiphanyOpenAiStreamPayload,
) -> EpiphanyOpenAiStreamEvent {
EpiphanyOpenAiStreamEvent {
request_id: request_id.to_string(),
sequence,
payload,
}
}
#[test]
fn compacts_openai_text_and_reasoning_deltas_before_storage() {
let events = vec![
test_openai_event(
"req-1",
0,
EpiphanyOpenAiStreamPayload::ReasoningDelta {
text: "think".to_string(),
},
),
test_openai_event(
"req-1",
1,
EpiphanyOpenAiStreamPayload::ReasoningDelta {
text: " small".to_string(),
},
),
test_openai_event(
"req-1",
2,
EpiphanyOpenAiStreamPayload::TextDelta {
text: "{\"role".to_string(),
},
),
test_openai_event(
"req-1",
3,
EpiphanyOpenAiStreamPayload::TextDelta {
text: "Id\":\"modeling\"}".to_string(),
},
),
test_openai_event(
"req-1",
4,
EpiphanyOpenAiStreamPayload::Completed {
receipt: EpiphanyOpenAiModelReceipt::new("req-1", "gpt-5.4"),
},
),
];
let compacted = compact_openai_events_for_storage(&events);
assert_eq!(compacted.len(), 3);
assert_eq!(compacted[0].sequence, 0);
assert!(matches!(
&compacted[0].payload,
EpiphanyOpenAiStreamPayload::ReasoningDelta { text } if text == "think small"
));
assert!(matches!(
&compacted[1].payload,
EpiphanyOpenAiStreamPayload::TextDelta { text } if text == "{\"roleId\":\"modeling\"}"
));
assert!(matches!(
compacted[2].payload,
EpiphanyOpenAiStreamPayload::Completed { .. }
));
}
#[test]
fn records_one_native_model_stream_in_runtime_store() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
let native_request = EpiphanyModelRequest::new(
"req-1",
"conversation-1",
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
"Answer plainly.",
);
let request = epiphany_openai_adapter::request_from_native(&native_request);
let options = default_options(store.clone(), PathBuf::from(".codex"), &request);
ensure_openai_runtime_ready(&options)?;
open_runtime_model_execution(
&store,
RuntimeSpineSessionOptions {
session_id: options.session_id.clone(),
objective: options.objective.clone(),
created_at: now(),
coordinator_note: options.coordinator_note.clone(),
},
RuntimeSpineJobOptions {
job_id: options.job_id.clone(),
session_id: options.session_id.clone(),
role: OPENAI_RUNTIME_ROLE.to_string(),
created_at: now(),
},
&native_request,
&now(),
)?;
let mut receipt = EpiphanyOpenAiModelReceipt::new("req-1", "gpt-5.4");
receipt.response_id = Some("resp-1".to_string());
receipt.transport = Some("test".to_string());
let events = vec![EpiphanyOpenAiStreamEvent {
request_id: "req-1".to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::Completed { receipt },
}];
let summary =
record_openai_events(&store, &options, &request, DEFAULT_MODEL_PROVIDER, &events)?;
assert_eq!(summary.verdict, "pass");
assert_eq!(assistant_text_from_model_events(&store, "req-1")?, "");
let mut cache = runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
assert!(cache.get::<EpiphanyModelRequest>("req-1")?.is_some());
assert!(
cache
.get::<EpiphanyModelStreamEvent>("req-1:00000000")?
.is_some()
);
assert!(cache.get::<EpiphanyModelReceipt>("req-1")?.is_some());
assert!(cache.get::<EpiphanyOpenAiModelRequest>("req-1")?.is_some());
assert_eq!(
runtime_job_snapshot(&store, &options.job_id)?
.expect("snapshot")
.job
.status,
epiphany_core::EpiphanyRuntimeJobStatus::Completed
);
Ok(())
}
#[test]
fn completes_worker_job_from_model_json_without_codex_worker_runtime() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "epiphany-test".into(),
display_name: "Epiphany Test".into(),
created_at: "2026-08-08T00:00:00Z".into(),
},
)?;
let repo = temp.path().join("workspace");
std::fs::create_dir(&repo)?;
run_git(&repo, &["init"])?;
run_git(&repo, &["config", "user.email", "body@example.invalid"])?;
run_git(&repo, &["config", "user.name", "Body Test"])?;
std::fs::write(repo.join("src.txt"), "authenticated Body")?;
run_git(&repo, &["add", "."])?;
run_git(&repo, &["commit", "-m", "seed"])?;
let body_store = temp.path().join("body.cc");
epiphany_core::bind_runtime_to_swarm(&store, "swarm-test", "2026-08-08T00:00:00Z")?;
epiphany_core::bind_repository_body(&repo, &body_store, &store, "workspace-test")?;
let body_basis = epiphany_core::observe_runtime_repository_body_basis(&store)?;
epiphany_core::initialize_keyed_repo_model(
&store,
&epiphany_core::EpiphanyRepoModelSeed::new(
"body-seed",
"body-graph",
"swarm-test",
"workspace-test",
body_basis.body_binding_sha256.clone(),
epiphany_core::EpiphanyRepoModelSeedDocuments {
domains: Vec::new(),
nodes: Vec::new(),
edges: Vec::new(),
frontier: Vec::new(),
},
)?,
"2026-08-08T00:00:01Z",
)?;
let worker_job_id =
epiphany_core::launch_current_body_modeling_work(&store, "2026-08-08T00:00:02Z")?;
let launch_request = load_worker_launch_request(&store, &worker_job_id)?;
let reasoning_basis = epiphany_core::worker_reasoning_basis(&store, &launch_request)?;
epiphany_core::put_reasoning_basis(&store, &reasoning_basis)?;
let model_request = build_worker_model_request(
&launch_request,
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
&reasoning_basis,
)?;
epiphany_core::open_runtime_model_execution(
&store,
RuntimeSpineSessionOptions {
session_id: "openai-worker-session-modeling-checkpoint-worker".into(),
objective: "Run the exact Modeling pass.".into(),
created_at: "2026-08-14T00:00:00Z".into(),
coordinator_note: "decision context fixture".into(),
},
RuntimeSpineJobOptions {
job_id: "openai-worker-worker-job-1".into(),
session_id: "openai-worker-session-modeling-checkpoint-worker".into(),
role: "openai-model-adapter".into(),
created_at: "2026-08-14T00:00:00Z".into(),
},
&model_request,
"2026-08-14T00:00:00Z",
)?;
assert_eq!(
model_request.output_contract_id.as_deref(),
Some(epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID)
);
let output_schema = model_request
.output_schema_json
.as_deref()
.expect("worker model request should carry role output schema");
let output_schema_value: serde_json::Value = serde_json::from_str(output_schema)?;
assert!(
output_schema_value["properties"]
.get("frontierVerdictGap")
.is_none()
);
assert!(
output_schema_value["properties"]
.get("frontierNodeIds")
.is_some()
);
assert!(
output_schema_value["properties"]
.get("repoModelOperations")
.is_some()
);
assert!(!output_schema.contains("incorporate_frontier_verdict"));
assert!(!model_request.instructions.contains("Output schema JSON"));
assert!(model_request.instructions.contains("repoModelOperations"));
assert!(
model_request
.instructions
.contains("Emit every object key at most once")
);
assert_eq!(model_request.reasoning_effort.as_deref(), Some("low"));
assert_eq!(model_request.reasoning_summary.as_deref(), Some("concise"));
assert!(
!model_request
.instructions
.contains("<epiphany_dynamic_context>")
);
let epiphany_core::EpiphanyReasoningProjection::RolePass(sealed_projection) =
reasoning_basis.projection()?
else {
panic!("Body Modeling must seal a role projection")
};
assert!(
sealed_projection
.authority
.frontier_verdict_modeling_context
.is_none()
);
let sealed_body = sealed_projection
.modeling_body
.as_ref()
.expect("Modeling must receive its authenticated Body projection");
assert_eq!(sealed_body.basis, body_basis);
assert!(
sealed_body
.manifest_entries
.iter()
.any(|entry| entry.path == "src.txt")
);
assert!(
model_request
.tools
.iter()
.any(|tool| tool.name == "mcp__epiphany_source__read_file")
);
assert!(
model_request
.tools
.iter()
.any(|tool| tool.name == "mcp__epiphany_source__directory_inventory")
);
assert!(
model_request
.tools
.iter()
.any(|tool| tool.name == "mcp__epiphany_state__resident_grant_lifecycle")
);
assert!(
!model_request
.tools
.iter()
.any(|tool| tool.name == "mcp__epiphany_public__github_file")
);
assert!(model_request.instructions.contains("Modeling must inspect"));
let provider_failure_store = temp.path().join("provider-failure.cc");
std::fs::copy(&store, &provider_failure_store)?;
let provider_failure_summary = EpiphanyOpenAiRuntimeRunSummary {
store: provider_failure_store.display().to_string(),
session_id: "openai-worker-session-modeling-checkpoint-worker".to_string(),
job_id: "openai-worker-worker-job-1".to_string(),
request_id: model_request.request_id.clone(),
event_count: 1,
verdict: "failed".to_string(),
summary: "Provider refused the model pass.".to_string(),
result_id: "result-openai-worker-worker-job-1".to_string(),
receipt_id: None,
tool_intent_ids: Vec::new(),
};
let provider_failure = complete_worker_job_from_assistant_text(
&provider_failure_store,
&launch_request,
&model_request.request_id,
&provider_failure_summary,
"",
)?;
assert_eq!(provider_failure.verdict, "failed");
assert_eq!(
epiphany_core::model_pass_failure_for_request(
&provider_failure_store,
&model_request.request_id,
)?
.expect("provider failure record")
.failure_kind,
"provider_or_transport_failure"
);
assert!(
epiphany_core::runtime_role_worker_result(&provider_failure_store, &worker_job_id)?
.is_none()
);
let contract_failure_store = temp.path().join("contract-failure.cc");
std::fs::copy(&store, &contract_failure_store)?;
let contract_failure_summary = EpiphanyOpenAiRuntimeRunSummary {
store: contract_failure_store.display().to_string(),
verdict: "pass".to_string(),
receipt_id: Some(model_request.request_id.clone()),
..provider_failure_summary.clone()
};
let contract_failure = complete_worker_job_from_assistant_text(
&contract_failure_store,
&launch_request,
&model_request.request_id,
&contract_failure_summary,
"not the declared structured contract",
)?;
assert_eq!(contract_failure.verdict, "failed");
assert_eq!(
epiphany_core::model_pass_failure_for_request(
&contract_failure_store,
&model_request.request_id,
)?
.expect("contract failure record")
.failure_kind,
"output_contract_failure"
);
assert!(
epiphany_core::runtime_role_worker_result(&contract_failure_store, &worker_job_id)?
.is_none()
);
let openai_summary = EpiphanyOpenAiRuntimeRunSummary {
store: store.display().to_string(),
session_id: "openai-worker-session-modeling-checkpoint-worker".to_string(),
job_id: "openai-worker-worker-job-1".to_string(),
request_id: model_request.request_id.clone(),
event_count: 2,
verdict: "pass".to_string(),
summary: "OpenAI model request completed.".to_string(),
result_id: "result-openai-worker-worker-job-1".to_string(),
receipt_id: Some(model_request.request_id.clone()),
tool_intent_ids: Vec::new(),
};
let assistant_text = serde_json::json!({
"roleId": "modeling",
"verdict": "checkpoint-ready",
"summary": "Mapped.",
"nextSafeMove": "Review the runtime-owned Body model.",
"filesInspected": ["src/lib.rs"],
"frontierNodeIds": [],
"evidenceIds": ["body-evidence"],
"artifactRefs": ["artifact:model"],
"repoModelOperations": []
})
.to_string();
let result = complete_worker_job_from_assistant_text(
&store,
&launch_request,
&model_request.request_id,
&openai_summary,
&assistant_text,
)?;
assert_eq!(result.job_id, worker_job_id);
assert_eq!(result.verdict, "checkpoint-ready");
assert_eq!(result.summary, "Mapped.");
assert_eq!(
result.next_safe_move,
"Review the runtime-owned Body model."
);
let runtime_evidence_id = format!("openai-request:{}", model_request.request_id);
assert!(result.evidence_refs.contains(&runtime_evidence_id));
let typed_result = epiphany_core::runtime_role_worker_result(&store, &worker_job_id)?
.expect("typed role worker result");
assert_eq!(typed_result.verdict, "checkpoint-ready");
assert_eq!(typed_result.files_inspected, vec!["src/lib.rs".to_string()]);
assert_eq!(
typed_result.evidence_ids,
vec!["body-evidence".to_string(), runtime_evidence_id]
);
assert_eq!(typed_result.artifact_refs, result.artifact_refs);
assert!(typed_result.repo_model_mutation_proposal()?.is_none());
assert!(
runtime_job_snapshot(&store, &worker_job_id)?
.expect("snapshot")
.result
.is_some()
);
Ok(())
}
#[test]
fn verification_worker_request_advertises_read_only_source_tools() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
seed_test_runtime_job(
&store,
RuntimeSpineHeartbeatJobOptions {
runtime_id: "epiphany-test".to_string(),
session_id: "epiphany-main".to_string(),
objective: "Verify the machine.".to_string(),
coordinator_note: "test".to_string(),
job_id: "verification-job-1".to_string(),
role: "verification".to_string(),
binding_id: epiphany_core::EPIPHANY_VERIFICATION_ROLE_BINDING_ID.to_string(),
authority_scope: "epiphany.role.verification".to_string(),
instruction: "Return the required verification-result JSON.".to_string(),
launch_document: EpiphanyWorkerLaunchDocument::Role(
epiphany_core::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-1".to_string(),
role_id: "verification".to_string(),
objective: Some("Verify Hands receipts.".to_string()),
dynamic_prompt_context: Some(
"<verification_work_loop_telemetry>hands receipts</verification_work_loop_telemetry>"
.to_string(),
),
repository_body_observation_basis: None,
proposal_modeling_context: None,
frontier_verdict_modeling_context: None,
frontier_planning_context: None,
frontier_research_context: None,
frontier_verification_context: None,
frontier_plan_mind_context: None,
imagination_consideration_context: None,
admitted_model_direction_consideration_context: None,
},
),
output_contract_id: epiphany_core::ROLE_WORKER_OUTPUT_CONTRACT_ID.to_string(),
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_research_request_id: None,
repo_frontier_verification_request_id: None,
created_at: now(),
},
)?;
let launch_request = load_worker_launch_request(&store, "verification-job-1")?;
let model_request = build_worker_model_request(
&launch_request,
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
&test_reasoning_basis(&launch_request)?,
)?;
let tool_names = model_request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert!(tool_names.contains(&"mcp__epiphany_source__read_file"));
assert!(tool_names.contains(&"mcp__epiphany_source__directory_inventory"));
assert!(tool_names.contains(&"mcp__epiphany_source__git_show"));
assert!(tool_names.contains(&"mcp__epiphany_source__read_hands_receipt"));
assert!(tool_names.contains(&"mcp__epiphany_state__resident_grant_lifecycle"));
assert!(!tool_names.contains(&"mcp__epiphany_public__github_file"));
assert!(
model_request
.instructions
.contains("mcp__epiphany_source__read_file")
);
assert!(
model_request
.instructions
.contains("mcp__epiphany_source__directory_inventory")
);
assert!(
model_request
.instructions
.contains("mcp__epiphany_state__resident_grant_lifecycle")
);
let mut research_launch = launch_request;
research_launch.binding_id = epiphany_core::EPIPHANY_RESEARCH_ROLE_BINDING_ID.to_string();
research_launch.role = epiphany_core::EPIPHANY_RESEARCH_OWNER_ROLE.to_string();
research_launch.authority_scope = "epiphany.role.research".to_string();
let research_request = build_worker_model_request(
&research_launch,
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
&test_reasoning_basis(&research_launch)?,
)?;
assert!(
!research_request
.tools
.iter()
.any(|tool| tool.name == "mcp__epiphany_public__github_file")
);
assert!(
research_request
.instructions
.contains("runtime obtains every immutable public GitHub source")
);
Ok(())
}
#[test]
fn mcp_model_tool_call_becomes_typed_invocation_intent() -> Result<()> {
let event = EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: "request-1".to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence: 7,
payload: EpiphanyModelStreamPayload::ToolCall {
call_id: "call/1".to_string(),
name: "mcp__calendar_server__list_events".to_string(),
arguments: r#"{"limit":3}"#.to_string(),
},
};
let intent = tool_invocation_intent_from_model_event(&event)
.expect("MCP-shaped tool call should produce an intent");
assert_eq!(
intent.adapter,
epiphany_tool_adapter::EPIPHANY_TOOL_RUNTIME_ADAPTER_ID
);
assert_eq!(intent.server, "calendar_server");
assert_eq!(intent.tool_name, "list_events");
assert_eq!(intent.arguments_json, r#"{"limit":3}"#);
assert_eq!(intent.intent_id, "model-request-1-7-call-1");
assert_eq!(intent.call_id.as_deref(), Some("call/1"));
assert_eq!(intent.model_request_id.as_deref(), Some("request-1"));
Ok(())
}
#[test]
fn builds_tool_followup_model_request_from_receipts() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
let mut native_request = EpiphanyModelRequest::new(
"req-tools",
"conversation-1",
DEFAULT_MODEL_PROVIDER,
"gpt-5.4",
"Answer after tool output.",
);
native_request.output_schema_json =
Some(r#"{"type":"object","required":["researchDecision"]}"#.to_string());
let request = epiphany_openai_adapter::request_from_native(&native_request);
let options = default_options(store.clone(), PathBuf::from(".codex"), &request);
ensure_openai_runtime_ready(&options)?;
open_runtime_model_execution(
&store,
RuntimeSpineSessionOptions {
session_id: options.session_id.clone(),
objective: options.objective.clone(),
created_at: now(),
coordinator_note: options.coordinator_note.clone(),
},
RuntimeSpineJobOptions {
job_id: options.job_id.clone(),
session_id: options.session_id.clone(),
role: OPENAI_RUNTIME_ROLE.to_string(),
created_at: now(),
},
&native_request,
&now(),
)?;
let mut receipt = EpiphanyOpenAiModelReceipt::new("req-tools", "gpt-5.4");
receipt.response_id = Some("resp-tools".to_string());
receipt.transport = Some("test".to_string());
let events = vec![
EpiphanyOpenAiStreamEvent {
request_id: "req-tools".to_string(),
sequence: 0,
payload: EpiphanyOpenAiStreamPayload::ToolCall {
call_id: "call-original".to_string(),
name: "mcp__smoke_server__smoke_tool".to_string(),
arguments: "{}".to_string(),
},
},
EpiphanyOpenAiStreamEvent {
request_id: "req-tools".to_string(),
sequence: 1,
payload: EpiphanyOpenAiStreamPayload::Completed { receipt },
},
];
let summary =
record_openai_events(&store, &options, &request, DEFAULT_MODEL_PROVIDER, &events)?;
let intent_id = summary
.tool_intent_ids
.first()
.expect("tool intent id")
.clone();
let mut cache = runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
let tool_binding =
epiphany_core::require_runtime_tool_execution_binding(&store, &intent_id)?;
assert_eq!(tool_binding.session_id, options.session_id);
assert_eq!(tool_binding.job_id, options.job_id);
assert_eq!(tool_binding.model_request_id.as_deref(), Some("req-tools"));
let mut tool_receipt = EpiphanyToolInvocationReceipt::new(
"receipt-tool",
intent_id.clone(),
epiphany_tool_adapter::EPIPHANY_TOOL_RUNTIME_ADAPTER_ID,
"smoke_server",
"smoke_tool",
"completed",
now(),
);
tool_receipt.result_json = Some(r#"{"ok":true}"#.to_string());
drop(cache);
epiphany_core::put_runtime_tool_execution_receipt(&store, &tool_receipt)?;
let followup =
build_tool_followup_model_request(&store, "req-tools", "req-tools-followup")?;
assert_eq!(followup.request_id, "req-tools-followup");
assert_eq!(followup.previous_response_id, None);
assert_eq!(
followup.output_schema_json.as_deref(),
Some(r#"{"type":"object","required":["researchDecision"]}"#)
);
assert_eq!(followup.input.len(), 2);
assert_eq!(
followup.input[0],
EpiphanyModelInputItem::ToolCall {
call_id: "call-original".to_string(),
name: "mcp__smoke_server__smoke_tool".to_string(),
arguments: "{}".to_string()
}
);
assert_eq!(
followup.input[1],
EpiphanyModelInputItem::ToolResult {
call_id: "call-original".to_string(),
output: r#"{"ok":true}"#.to_string()
}
);
Ok(())
}
#[test]
fn incomplete_or_non_mcp_tool_calls_do_not_create_invocation_intents() {
let base = EpiphanyModelStreamEvent {
schema_id: epiphany_model_adapter::MODEL_ADAPTER_EVENT_SCHEMA_ID.to_string(),
request_id: "request-1".to_string(),
provider: DEFAULT_MODEL_PROVIDER.to_string(),
sequence: 7,
payload: EpiphanyModelStreamPayload::ToolCall {
call_id: "call".to_string(),
name: "shell".to_string(),
arguments: "{}".to_string(),
},
};
assert!(tool_invocation_intent_from_model_event(&base).is_none());
let mut incomplete = base.clone();
incomplete.payload = EpiphanyModelStreamPayload::ToolCall {
call_id: "call".to_string(),
name: "mcp__server__tool".to_string(),
arguments: "{".to_string(),
};
assert!(tool_invocation_intent_from_model_event(&incomplete).is_none());
}
}