GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate Planning to keyed Mind authority

Browse source
This commit is contained in:
Meta Vens 2026-08-18 02:22:11 +02:00
commit c9329ed6ec
10 changed files with 1566 additions and 1044 deletions
Download patch file Download diff file Expand all files Collapse all files

306
epiphany-core/src/agent_launch.rs
View file

@ -1066,83 +1066,6 @@ pub fn build_epiphany_role_launch_request(
)
}
pub fn build_epiphany_frontier_plan_mind_launch_request(
thread_id: &str,
expected_revision: Option<u64>,
max_runtime_seconds: Option<u64>,
state: &EpiphanyThreadState,
mind_request_id: String,
) -> Result<EpiphanyJobLaunchRequest, String> {
let linked_subgoal_ids = epiphany_active_subgoal_ids(Some(state));
let linked_graph_node_ids = epiphany_active_graph_node_ids(Some(state));
let authority_scope = "epiphany.procedure.mind_admission_review";
let launch_document = EpiphanyWorkerLaunchDocument::Role(EpiphanyRoleWorkerLaunchDocument {
thread_id: thread_id.to_string(),
role_id: "mindAdmissionReview".to_string(),
state_revision: state.revision,
objective: state.objective.clone(),
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,
active_subgoal_id: state.active_subgoal_id.clone(),
active_subgoals: state
.subgoals
.iter()
.filter(|subgoal| Some(subgoal.id.as_str()) == state.active_subgoal_id.as_deref())
.cloned()
.collect(),
active_graph_node_ids: linked_graph_node_ids.clone(),
investigation_checkpoint: state.investigation_checkpoint.clone(),
scratch: state.scratch.clone(),
invariants: state.invariants.clone(),
graphs: Some(state.graphs.clone()),
recent_evidence: state.recent_evidence.iter().take(8).cloned().collect(),
recent_observations: state.observations.iter().take(8).cloned().collect(),
graph_frontier: state.graph_frontier.clone(),
graph_checkpoint: state.graph_checkpoint.clone(),
planning: Some(state.planning.clone()),
churn: state.churn.clone(),
});
let output_contract_id = launch_document.output_contract_id().to_string();
let organ_launch_contract = default_launch_organ_contract(
authority_scope,
launch_document.document_kind(),
&output_contract_id,
);
Ok(EpiphanyJobLaunchRequest {
expected_revision,
binding_id: EPIPHANY_MIND_ROLE_BINDING_ID.to_string(),
kind: CoreEpiphanyJobKind::Specialist,
scope: "role-scoped frontier plan judgment".to_string(),
owner_role: EPIPHANY_MIND_OWNER_ROLE.to_string(),
authority_scope: authority_scope.to_string(),
linked_subgoal_ids,
linked_graph_node_ids,
instruction: epiphany_worker_prompt(
epiphany_specialist_prompt_config().roles.mind.as_str(),
),
launch_document,
output_contract_id,
organ_launch_contract,
max_runtime_seconds,
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: Some(mind_request_id),
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,
})
}
pub fn build_epiphany_role_launch_request_with_dynamic_context(
thread_id: &str,
role_id: EpiphanyRoleResultRoleId,
@ -1514,7 +1437,6 @@ fn epiphany_active_graph_node_ids(state: Option<&EpiphanyThreadState>) -> Vec<St
#[cfg(test)]
mod tests {
use super::*;
use sha2::Digest;
#[test]
fn admitted_direction_schema_bounds_autonomous_proposal_fanout() {
@ -1595,234 +1517,6 @@ mod tests {
assert!(!prompt.contains("Heartbeat: every lane"));
}
#[test]
fn frontier_plan_mind_launch_is_a_real_prompted_role_with_strict_schema() {
let request = build_epiphany_frontier_plan_mind_launch_request(
"thread-mind",
Some(0),
Some(60),
&EpiphanyThreadState::default(),
"mind-request-1".into(),
)
.expect("Mind launch request");
assert_eq!(request.owner_role, EPIPHANY_MIND_OWNER_ROLE);
assert_eq!(request.binding_id, EPIPHANY_MIND_ROLE_BINDING_ID);
assert!(
request
.instruction
.contains("admission-review procedure serving Epiphany Mind")
);
assert_eq!(
request.frontier_plan_mind_request_id.as_deref(),
Some("mind-request-1")
);
let EpiphanyWorkerLaunchDocument::Role(document) = request.launch_document else {
panic!("Mind must use a role launch document")
};
assert_eq!(document.role_id, "mindAdmissionReview");
let schema = epiphany_frontier_plan_mind_output_schema();
assert_eq!(
schema["properties"]["roleId"]["const"],
"mindAdmissionReview"
);
assert_eq!(
schema["properties"]["frontierPlanMindDecision"]["properties"]["decision"]["enum"],
serde_json::json!(["adopt", "refuse", "hold"])
);
assert!(
schema["properties"]
.get("frontierPlanMindRequestId")
.is_none()
);
for runtime_owned in [
"mindRequestId",
"planningRequestId",
"imaginationResultId",
"candidateId",
"candidateSha256",
"decidedAt",
] {
assert!(
schema["properties"]["frontierPlanMindDecision"]["properties"]
.get(runtime_owned)
.is_none()
);
}
}
#[test]
fn modeling_schema_exposes_semantic_operations_without_runtime_authority() {
let schema = epiphany_role_launch_output_schema(EpiphanyRoleResultRoleId::Modeling);
let required = schema["required"].as_array().expect("required fields");
assert!(required.iter().any(|value| value == "repoModelOperations"));
for runtime_owned in [
"repoModelPatch",
"repositoryBodyObservationBasis",
"repoFrontierModelingRequestId",
"proposalModelingRequestId",
] {
assert!(schema["properties"].get(runtime_owned).is_none());
}
let operations = schema["properties"]["repoModelOperations"]["items"]["anyOf"]
.as_array()
.expect("typed Modeling operations");
let operation_names = operations
.iter()
.map(|operation| {
operation["properties"]["operation"]["const"]
.as_str()
.unwrap()
})
.collect::<Vec<_>>();
assert_eq!(
operation_names,
vec![
"put_node",
"retire_node",
"put_edge",
"retire_edge",
"put_frontier",
"create_surface_offer",
"deprecate_surface_offer",
"withdraw_surface_offer",
"create_dependency_claim",
"retire_dependency_claim"
]
);
let node_kinds = operations[0]["properties"]["node"]["properties"]["kind"]["enum"]
.as_array()
.expect("typed RepoModel node kinds");
assert!(node_kinds.iter().any(|kind| kind == "runtime_contract"));
assert!(!node_kinds.iter().any(|kind| kind == "claim"));
assert!(
operations[0]["properties"]["node"]["properties"]
.get("created_at")
.is_none()
);
assert!(operations[5]["properties"].get("surface_id").is_none());
assert!(operations[8]["properties"].get("claim_id").is_none());
assert_eq!(
operations[5]["required"],
serde_json::json!(["operation", "label", "contract", "source_refs"])
);
assert_eq!(
operations[8]["required"],
serde_json::json!([
"operation",
"label",
"target",
"entanglement_kind",
"failure_semantics",
"impact_scope",
"source_refs"
])
);
assert_eq!(
schema["allOf"][0]["then"]["properties"]["repoModelOperations"]["maxContains"],
1
);
assert_eq!(
schema["allOf"][1]["then"]["properties"]["repoModelOperations"]["maxItems"],
0
);
}
#[test]
fn verdict_bound_modeling_schema_exposes_only_exact_frontier_revision_authority() {
let authority = crate::RepoFrontierVerdictModelingLaunchAuthority {
request: crate::RepoFrontierModelingRequest {
schema_version: crate::REPO_FRONTIER_MODELING_REQUEST_SCHEMA_VERSION.to_string(),
request_id: "modeling-request-exact".into(),
model_projection_digest: format!("sha256:{}", "a".repeat(64)),
model_source_documents: vec![crate::EpiphanyMindDocumentVersion {
store_id: "epiphany-mind".into(),
document_type: "epiphany.mind.repo_model.identity.v1".into(),
document_key: crate::REPO_MODEL_IDENTITY_KEY.into(),
schema_id: Some("EpiphanyRepoModelIdentityDocument".into()),
payload_msgpack: vec![1],
payload_sha256: format!("sha256:{:x}", sha2::Sha256::digest([1])),
}],
route_id: "route-exact".into(),
frontier_item_id: "frontier-exact".into(),
frontier_item_hash: "frontier-hash".into(),
verification_request_id: "verification-exact".into(),
soul_verdict_receipt_id: "soul-exact".into(),
verification_result_id: "verification-result".into(),
verification_job_id: "verification-job".into(),
allowed_disposition: crate::RepoFrontierVerdictDisposition::Resolved,
requested_at: "2026-08-08T00:00:00Z".into(),
contract: crate::REPO_FRONTIER_MODELING_REQUEST_CONTRACT.into(),
},
frontier_item: crate::RepoFrontierItem {
id: "frontier-exact".into(),
migration_body: "runtime".into(),
question: "Did the consequence hold?".into(),
gap: "Awaiting verdict incorporation.".into(),
target_claim_ids: vec!["claim-exact".into()],
source_scope: vec!["epiphany-core".into()],
recommended_next_organ: "Hands".into(),
adopted_plan: Some(crate::RepoFrontierAdoptedPlan {
command: "cargo test".into(),
..Default::default()
}),
dependency_item_ids: Vec::new(),
status: crate::RepoFrontierStatus::Active,
evidence_refs: vec!["prior-evidence".into()],
public_source_refs: Vec::new(),
created_at: Some("2026-08-07T00:00:00Z".into()),
updated_at: Some("2026-08-07T00:00:00Z".into()),
retired_at: None,
superseded_by: None,
},
soul_verdict: crate::SoulVerdictReceipt {
schema_version: crate::SOUL_VERDICT_RECEIPT_SCHEMA_VERSION.into(),
receipt_id: "soul-exact".into(),
source_result_id: "verification-result".into(),
source_job_id: "verification-job".into(),
verdict: "pass".into(),
summary: "verified".into(),
evidence_ids: Vec::new(),
risks: Vec::new(),
emitted_at: "2026-08-08T00:00:00Z".into(),
contract: "test".into(),
verification_request_id: "verification-exact".into(),
frontier_route_id: "route-exact".into(),
},
};
let schema = epiphany_frontier_verdict_modeling_output_schema(&authority);
assert!(
schema["properties"]
.get("repoFrontierModelingRequestId")
.is_none()
);
assert!(schema["properties"].get("repoModelOperations").is_none());
assert_eq!(schema["properties"]["frontierVerdictGap"]["type"], "string");
assert!(
schema["required"]
.as_array()
.unwrap()
.iter()
.any(|value| value == "frontierVerdictGap")
);
assert_eq!(schema["properties"]["evidenceIds"]["minItems"], 1);
}
#[test]
fn modeling_node_schema_matches_mind_nonempty_invariants() {
let schema = repo_model_node_output_schema();
assert_eq!(schema["properties"]["title"]["minLength"], 1);
assert_eq!(schema["properties"]["claim"]["minLength"], 1);
assert_eq!(schema["properties"]["action_implication"]["minLength"], 1);
assert_eq!(
schema["anyOf"],
serde_json::json!([
{"properties": {"question": {"minLength": 1}}},
{"properties": {"tension": {"minLength": 1}}}
])
);
}
#[test]
fn role_self_patch_schema_matches_canonical_numeric_memory_contract() {
let schema = epiphany_role_launch_output_schema(EpiphanyRoleResultRoleId::Verification);

177
epiphany-core/src/bin/epiphany-mvp-coordinator.rs
View file

@ -634,39 +634,26 @@ fn run_coordinator(args: &Args) -> Result<Value> {
"reviewFrontierPlanningFailure" => {
let lifecycle =
epiphany_core::runtime_repo_frontier_planning_lifecycle(&runtime_store)?;
let (result_id, job_id, role_id, binding_id) = match lifecycle.stage {
epiphany_core::RepoFrontierPlanningLifecycleStage::ImaginationFailed => (
lifecycle.imagination_result_id.as_deref(),
lifecycle.imagination_job_id.as_deref(),
"imagination",
epiphany_core::EPIPHANY_IMAGINATION_ROLE_BINDING_ID,
),
epiphany_core::RepoFrontierPlanningLifecycleStage::MindFailed => (
lifecycle.mind_result_id.as_deref(),
lifecycle.mind_job_id.as_deref(),
"mindAdmissionReview",
epiphany_core::EPIPHANY_MIND_ROLE_BINDING_ID,
),
let job_id = match lifecycle.stage {
epiphany_core::RepoFrontierPlanningLifecycleStage::ImaginationFailed => {
lifecycle.imagination_job_id.as_deref()
}
epiphany_core::RepoFrontierPlanningLifecycleStage::MindFailed => {
lifecycle.mind_job_id.as_deref()
}
_ => return Err(anyhow!("frontier planning failure review stage changed")),
};
let result_id = result_id.ok_or_else(|| {
anyhow!("frontier planning failure review requires a typed result")
})?;
let job_id = job_id.ok_or_else(|| {
anyhow!("frontier planning failure review requires a launch job")
})?;
let reviewed = supersede_frontier_planning_failure(
let reviewed = epiphany_core::review_repo_frontier_planning_failure(
&runtime_store,
&thread_id,
revision,
result_id,
job_id,
role_id,
binding_id,
&now(),
)?;
push_event(
&mut step,
json!({"type": "frontierPlanningFailureReview", "review": status_cli::sanitize_for_operator(reviewed)}),
json!({"type": "frontierPlanningFailureReview", "review": status_cli::sanitize_for_operator(serde_json::to_value(reviewed)?)}),
);
final_status = collect_coordinator_status(&runtime_store, &thread_id)?;
}
@ -682,23 +669,13 @@ fn run_coordinator(args: &Args) -> Result<Value> {
final_status = collect_coordinator_status(&runtime_store, &thread_id)?;
}
"launchImagination" => {
let lifecycle =
epiphany_core::runtime_repo_frontier_planning_lifecycle(&runtime_store)?;
let request_id = lifecycle.planning_request_id.as_deref().ok_or_else(|| {
anyhow!("launchImagination requires a current planning request")
})?;
let launch = launch_frontier_imagination(
let worker_job_id = epiphany_core::launch_current_frontier_planning_work(
&runtime_store,
&local_verse_store,
&thread_id,
revision,
args.max_runtime_seconds,
request_id,
&now(),
)?;
let worker_job_id = worker_job_id_from_launch(&launch)?;
push_event(
&mut step,
json!({"type": "frontierImaginationLaunch", "launch": status_cli::sanitize_for_operator(launch), "runtimeJobId": worker_job_id}),
json!({"type": "frontierImaginationLaunch", "runtimeJobId": worker_job_id}),
);
let worker_run = launch_worker_runtime_detached(
&model_runtime_bin,
@ -741,22 +718,13 @@ fn run_coordinator(args: &Args) -> Result<Value> {
final_status = collect_coordinator_status(&runtime_store, &thread_id)?;
}
"launchMindPlanReview" => {
let lifecycle =
epiphany_core::runtime_repo_frontier_planning_lifecycle(&runtime_store)?;
let request_id = lifecycle.mind_request_id.as_deref().ok_or_else(|| {
anyhow!("launchMindPlanReview requires a current Mind request")
})?;
let launch = launch_frontier_mind(
let worker_job_id = epiphany_core::launch_current_frontier_plan_mind_work(
&runtime_store,
&thread_id,
revision,
args.max_runtime_seconds,
request_id,
&now(),
)?;
let worker_job_id = worker_job_id_from_launch(&launch)?;
push_event(
&mut step,
json!({"type": "frontierMindLaunch", "launch": status_cli::sanitize_for_operator(launch), "runtimeJobId": worker_job_id}),
json!({"type": "frontierMindLaunch", "runtimeJobId": worker_job_id}),
);
let worker_run = launch_worker_runtime_detached(
&model_runtime_bin,
@ -1890,61 +1858,6 @@ fn launch_role(
}))
}
fn launch_frontier_imagination(
runtime_store: &Path,
local_verse_store: &Path,
thread_id: &str,
expected_revision: Option<i64>,
max_runtime_seconds: u64,
planning_request_id: &str,
) -> Result<Value> {
let service = epiphany_core::EpiphanyCoordinatorService::new(runtime_store);
let state = service
.state()?
.ok_or_else(|| anyhow!("cannot launch Imagination without native coordinator state"))?;
let role = epiphany_core::EpiphanyRoleResultRoleId::Imagination;
let context = epiphany_core::render_launch_dynamic_prompt_context(
runtime_store,
local_verse_store,
&state,
epiphany_core::role_launch_context_focus(&state, epiphany_core::epiphany_role_label(role)),
)
.map_err(anyhow::Error::msg)?;
let mut request = epiphany_core::build_epiphany_role_launch_request_with_dynamic_context(
thread_id,
role,
expected_revision.and_then(|value| u64::try_from(value).ok()),
Some(max_runtime_seconds),
&state,
Some(context),
)
.map_err(anyhow::Error::msg)?;
request.frontier_planning_request_id = Some(planning_request_id.to_string());
launch_job_value(&service, thread_id, &state, &request)
}
fn launch_frontier_mind(
runtime_store: &Path,
thread_id: &str,
expected_revision: Option<i64>,
max_runtime_seconds: u64,
mind_request_id: &str,
) -> Result<Value> {
let service = epiphany_core::EpiphanyCoordinatorService::new(runtime_store);
let state = service
.state()?
.ok_or_else(|| anyhow!("cannot launch Mind without native coordinator state"))?;
let request = epiphany_core::build_epiphany_frontier_plan_mind_launch_request(
thread_id,
expected_revision.and_then(|value| u64::try_from(value).ok()),
Some(max_runtime_seconds),
&state,
mind_request_id.to_string(),
)
.map_err(anyhow::Error::msg)?;
launch_job_value(&service, thread_id, &state, &request)
}
fn launch_job_value(
service: &epiphany_core::EpiphanyCoordinatorService,
thread_id: &str,
@ -2226,64 +2139,6 @@ fn supersede_role_result(
}))
}
fn supersede_frontier_planning_failure(
runtime_store: &Path,
thread_id: &str,
expected_revision: Option<i64>,
result_id: &str,
job_id: &str,
role_id: &str,
binding_id: &str,
) -> Result<Value> {
let service = epiphany_core::EpiphanyCoordinatorService::new(runtime_store);
let state = service
.state()?
.ok_or_else(|| anyhow!("planning failure review requires coordinator state"))?;
if let Some(existing) = state
.acceptance_receipts
.iter()
.find(|receipt| receipt.result_id == result_id)
{
if existing.job_id == job_id
&& existing.binding_id == binding_id
&& existing.surface == "roleFailureReview"
&& existing.role_id == role_id
&& existing.status == "superseded"
{
return Ok(json!({"revision": state.revision, "receipt": existing, "changed": false}));
}
return Err(anyhow!(
"planning failure already has conflicting review authority"
));
}
let receipt = epiphany_state_model::EpiphanyAcceptanceReceipt {
id: format!("frontier-planning-failure-review-{}", Uuid::new_v4()),
result_id: result_id.to_string(),
job_id: job_id.to_string(),
binding_id: binding_id.to_string(),
surface: "roleFailureReview".to_string(),
role_id: role_id.to_string(),
status: "superseded".to_string(),
accepted_at: now(),
accepted_observation_id: None,
accepted_evidence_id: None,
summary: Some(
"Reviewed non-executable frontier Imagination failure; Self may schedule one next immutable attempt."
.to_string(),
),
};
let applied = service.apply_state_update(
thread_id,
epiphany_core::EpiphanyStateUpdate {
expected_revision: expected_revision.and_then(|value| u64::try_from(value).ok()),
acceptance_receipts: vec![receipt.clone()],
..Default::default()
},
None,
)?;
Ok(json!({"revision": applied.revision, "receipt": receipt, "changed": true}))
}
fn default_binding_id_for_role(role_id: &str) -> String {
match role_id {
"imagination" => epiphany_core::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.to_string(),

380
epiphany-core/src/coordinator_launch.rs
View file

@ -114,37 +114,6 @@ fn commit_coordinator_job_launch_in_cache(
}
}
}
let frontier_planning_launch =
if let Some(request_id) = request.frontier_planning_request_id.as_deref() {
let (planning, identity, attempt_ordinal, superseded_failure_result_id) =
validate_frontier_planning_launch(cache, current_state, request, request_id)?;
let projection = RepoFrontierPlanningContextProjection::from_request(&planning);
match &mut effective_launch_document {
EpiphanyWorkerLaunchDocument::Role(document) => {
if document.proposal_modeling_context.is_some() {
return Err(anyhow!(
"frontier planning context is exclusive of Modeling authority contexts"
));
}
document.frontier_planning_context = Some(projection);
}
EpiphanyWorkerLaunchDocument::Reorient(_) => {
return Err(anyhow!(
"reorient launch cannot carry frontier planning context"
));
}
}
let bytes = rmp_serde::to_vec_named(&effective_launch_document)?;
Some((
planning,
identity,
format!("{:x}", Sha256::digest(bytes)),
attempt_ordinal,
superseded_failure_result_id,
))
} else {
None
};
let imagination_consideration_launch = if let Some(request_id) =
request.imagination_consideration_request_id.as_deref()
{
@ -218,48 +187,6 @@ fn commit_coordinator_job_launch_in_cache(
} else {
None
};
let frontier_plan_mind_launch =
if let Some(request_id) = request.frontier_plan_mind_request_id.as_deref() {
let (
mind_request,
planning,
candidate,
identity,
attempt_ordinal,
superseded_failure_result_id,
) = validate_frontier_plan_mind_launch(cache, current_state, request, request_id)?;
let projection =
RepoFrontierPlanMindContextProjection::new(&mind_request, &planning, &candidate);
match &mut effective_launch_document {
EpiphanyWorkerLaunchDocument::Role(document) => {
if document.proposal_modeling_context.is_some()
|| document.frontier_planning_context.is_some()
|| document.frontier_plan_mind_context.is_some()
{
return Err(anyhow!("Mind frontier decision context is exclusive"));
}
document.frontier_plan_mind_context = Some(projection);
}
EpiphanyWorkerLaunchDocument::Reorient(_) => {
return Err(anyhow!(
"reorient launch cannot carry Mind decision context"
));
}
}
let hash = format!(
"{:x}",
Sha256::digest(rmp_serde::to_vec_named(&effective_launch_document)?)
);
Some((
mind_request,
identity,
hash,
attempt_ordinal,
superseded_failure_result_id,
))
} else {
None
};
let prepared = prepare_runtime_spine_heartbeat_job(
&cache,
RuntimeSpineHeartbeatJobOptions {
@ -316,50 +243,6 @@ fn commit_coordinator_job_launch_in_cache(
})
.ok_or_else(|| anyhow!("prepared launch omitted its runtime identity replacement"))?;
let runtime_identity_replacement = batch.remove(runtime_identity_position);
if let Some((
planning,
identity,
worker_launch_document_sha256,
attempt_ordinal,
superseded_failure_result_id,
)) = frontier_planning_launch
{
let binding_record_id = if attempt_ordinal == 0 {
format!("repo-frontier-planning-launch-{}", planning.request_id)
} else {
format!(
"repo-frontier-planning-launch-{}-attempt-{attempt_ordinal}",
planning.request_id
)
};
let launch_binding = RepoFrontierPlanningLaunchBinding {
schema_version: REPO_FRONTIER_PLANNING_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id,
planning_request_id: planning.request_id,
job_id: plan.backend_job_id.clone(),
binding_id: request.binding_id.clone(),
runtime_id: identity.runtime_id,
thread_id: planning.thread_id,
launched_at: created_at.clone(),
worker_launch_document_sha256,
contract: REPO_FRONTIER_PLANNING_LAUNCH_BINDING_CONTRACT.into(),
attempt_ordinal,
superseded_failure_result_id,
};
if cache
.get::<RepoFrontierPlanningLaunchBinding>(&launch_binding.binding_record_id)?
.is_some()
{
return Err(anyhow!(
"frontier planning request is already bound to a launch"
));
}
batch.push(
cache
.prepare_entry(&launch_binding.binding_record_id, &launch_binding)?
.0,
);
}
if let Some((consideration, identity, worker_launch_document_sha256, attempt_ordinal)) =
imagination_consideration_launch
{
@ -395,44 +278,6 @@ fn commit_coordinator_job_launch_in_cache(
}
batch.push(cache.prepare_entry(&binding.binding_record_id, &binding)?.0);
}
if let Some((
mind_request,
identity,
worker_launch_document_sha256,
attempt_ordinal,
superseded_failure_result_id,
)) = frontier_plan_mind_launch
{
let binding_record_id = if attempt_ordinal == 0 {
format!("repo-frontier-plan-mind-launch-{}", mind_request.request_id)
} else {
format!(
"repo-frontier-plan-mind-launch-{}-attempt-{attempt_ordinal}",
mind_request.request_id
)
};
let binding = RepoFrontierPlanMindLaunchBinding {
schema_version: REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id,
mind_request_id: mind_request.request_id,
job_id: plan.backend_job_id.clone(),
binding_id: request.binding_id.clone(),
runtime_id: identity.runtime_id,
thread_id: mind_request.thread_id,
launched_at: created_at.clone(),
worker_launch_document_sha256,
contract: REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_CONTRACT.into(),
attempt_ordinal,
superseded_failure_result_id,
};
if cache
.get::<RepoFrontierPlanMindLaunchBinding>(&binding.binding_record_id)?
.is_some()
{
return Err(anyhow!("Mind request is already bound to a launch"));
}
batch.push(cache.prepare_entry(&binding.binding_record_id, &binding)?.0);
}
if matches!(
request.binding_id.as_str(),
EPIPHANY_RESEARCH_ROLE_BINDING_ID
@ -531,14 +376,12 @@ pub fn plan_coordinator_job_launch(
state.revision
));
}
if let Some(request_id) = request.frontier_planning_request_id.as_deref() {
let mut cache = runtime_spine_cache(runtime_store)?;
cache.pull_all_backing_stores()?;
validate_frontier_planning_launch(&cache, state, request, request_id)?;
} else if let Some(request_id) = request.frontier_plan_mind_request_id.as_deref() {
let mut cache = runtime_spine_cache(runtime_store)?;
cache.pull_all_backing_stores()?;
validate_frontier_plan_mind_launch(&cache, state, request, request_id)?;
if request.frontier_planning_request_id.is_some()
|| request.frontier_plan_mind_request_id.is_some()
{
return Err(anyhow!(
"frontier Planning and plan Mind launches are owned by keyed current work"
));
} else if request.owner_role == EPIPHANY_MODELING_OWNER_ROLE {
// Ordinary Modeling launches remain valid, but carry no proposal authority.
}
@ -596,217 +439,6 @@ pub fn plan_coordinator_job_launch(
})
}
fn validate_frontier_plan_mind_launch(
cache: &CultCache,
state: &EpiphanyThreadState,
launch: &EpiphanyJobLaunchRequest,
request_id: &str,
) -> Result<(
RepoFrontierPlanMindRequest,
RepoFrontierPlanningRequest,
RepoFrontierPlanCandidate,
EpiphanyRuntimeIdentity,
u64,
Option<String>,
)> {
if launch.owner_role != EPIPHANY_MIND_OWNER_ROLE
|| launch.binding_id != EPIPHANY_MIND_ROLE_BINDING_ID
{
return Err(anyhow!(
"frontier plan decision may only be carried by the Mind role launch"
));
}
let request = cache
.get::<RepoFrontierPlanMindRequest>(request_id)?
.ok_or_else(|| anyhow!("Mind request does not exist"))?;
let (planning, candidate) =
crate::runtime_spine::validate_repo_frontier_plan_mind_request(cache, &request)?;
let identity = cache
.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("Mind launch requires runtime identity"))?;
let persisted = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("Mind launch requires thread state"))?;
if persisted.state()? != *state
|| request.runtime_id != identity.runtime_id
|| request.thread_id.is_empty()
|| launch.launch_document.thread_id() != persisted.thread_id
{
return Err(anyhow!("Mind launch provenance mismatch"));
}
let persisted_state = persisted.state()?;
let mut launches = cache
.get_all::<RepoFrontierPlanMindLaunchBinding>()?
.into_iter()
.filter(|binding| binding.mind_request_id == request_id)
.collect::<Vec<_>>();
launches.sort_by_key(|binding| binding.attempt_ordinal);
for (expected, binding) in launches.iter().enumerate() {
if binding.attempt_ordinal != expected as u64 {
return Err(anyhow!("Mind planning attempts must be contiguous"));
}
}
let superseded_failure_result_id = if let Some(latest) = launches.last() {
let results = cache
.get_all::<EpiphanyRuntimeRoleWorkerResult>()?
.into_iter()
.filter(|result| result.job_id == latest.job_id)
.collect::<Vec<_>>();
if results.len() != 1 {
return Err(anyhow!("Mind retry requires one typed prior result"));
}
let failure = &results[0];
if !failure.role_id.eq_ignore_ascii_case("mindAdmissionReview")
|| failure.frontier_plan_mind_request_id.is_some()
|| failure.frontier_plan_mind_decision_msgpack.is_some()
|| failure.item_error.as_deref().is_none_or(str::is_empty)
{
return Err(anyhow!(
"Mind retry requires a non-executable typed failure"
));
}
let reviewed = persisted_state
.acceptance_receipts
.iter()
.filter(|receipt| {
receipt.result_id == failure.result_id
&& receipt.job_id == failure.job_id
&& receipt.binding_id == EPIPHANY_MIND_ROLE_BINDING_ID
&& receipt.surface == "roleFailureReview"
&& receipt.role_id == "mindAdmissionReview"
&& receipt.status == "superseded"
})
.count();
if reviewed != 1 {
return Err(anyhow!("Mind retry requires one explicit failure review"));
}
Some(failure.result_id.clone())
} else {
None
};
let attempt_ordinal = u64::try_from(launches.len())
.map_err(|_| anyhow!("Mind planning attempt ordinal overflow"))?;
Ok((
request,
planning,
candidate,
identity,
attempt_ordinal,
superseded_failure_result_id,
))
}
fn validate_frontier_planning_launch(
cache: &CultCache,
state: &EpiphanyThreadState,
launch: &EpiphanyJobLaunchRequest,
request_id: &str,
) -> Result<(
RepoFrontierPlanningRequest,
EpiphanyRuntimeIdentity,
u64,
Option<String>,
)> {
if match &launch.launch_document {
EpiphanyWorkerLaunchDocument::Role(document) => {
document.frontier_planning_context.is_some()
}
EpiphanyWorkerLaunchDocument::Reorient(_) => false,
} {
return Err(anyhow!("caller cannot author frontier planning context"));
}
if launch.owner_role != EPIPHANY_IMAGINATION_OWNER_ROLE
|| launch.binding_id != EPIPHANY_IMAGINATION_ROLE_BINDING_ID
{
return Err(anyhow!(
"frontier planning may only be carried by the Imagination role launch"
));
}
let planning = cache
.get::<RepoFrontierPlanningRequest>(request_id)?
.ok_or_else(|| anyhow!("frontier planning request {request_id:?} does not exist"))?;
crate::runtime_spine::validate_actionable_repo_frontier_planning_request(cache, &planning)?;
let identity = cache
.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("frontier planning launch requires runtime identity"))?;
let persisted_state = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("frontier planning launch requires authoritative thread state"))?;
let persisted_state_value = persisted_state.state()?;
if planning.request_id != request_id
|| planning.runtime_id != identity.runtime_id
|| persisted_state_value != *state
|| launch.launch_document.thread_id() != persisted_state.thread_id
{
return Err(anyhow!(
"frontier planning launch provenance binding mismatch"
));
}
let mut launches = cache
.get_all::<RepoFrontierPlanningLaunchBinding>()?
.into_iter()
.filter(|binding| binding.planning_request_id == request_id)
.collect::<Vec<_>>();
launches.sort_by_key(|binding| binding.attempt_ordinal);
for (expected, binding) in launches.iter().enumerate() {
if binding.attempt_ordinal != expected as u64 {
return Err(anyhow!(
"frontier planning attempts must be contiguous and monotonic"
));
}
}
let superseded_failure_result_id = if let Some(latest) = launches.last() {
let results = cache
.get_all::<EpiphanyRuntimeRoleWorkerResult>()?
.into_iter()
.filter(|result| result.job_id == latest.job_id)
.collect::<Vec<_>>();
if results.len() != 1 {
return Err(anyhow!(
"frontier planning retry requires exactly one typed prior attempt result"
));
}
let failure = &results[0];
if !failure.role_id.eq_ignore_ascii_case("imagination")
|| failure.frontier_planning_request_id.is_some()
|| failure.frontier_plan_candidate_msgpack.is_some()
|| failure.item_error.as_deref().is_none_or(str::is_empty)
{
return Err(anyhow!(
"frontier planning retry requires a non-executable typed prior failure"
));
}
let reviewed = persisted_state_value
.acceptance_receipts
.iter()
.filter(|receipt| {
receipt.result_id == failure.result_id
&& receipt.job_id == failure.job_id
&& receipt.binding_id == EPIPHANY_IMAGINATION_ROLE_BINDING_ID
&& receipt.surface == "roleFailureReview"
&& receipt.role_id == "imagination"
&& receipt.status == "superseded"
})
.collect::<Vec<_>>();
if reviewed.len() != 1 {
return Err(anyhow!(
"frontier planning retry requires exactly one explicit failure review"
));
}
Some(failure.result_id.clone())
} else {
None
};
let attempt_ordinal = u64::try_from(launches.len())
.map_err(|_| anyhow!("frontier planning attempt ordinal overflow"))?;
Ok((
planning,
identity,
attempt_ordinal,
superseded_failure_result_id,
))
}
fn validate_imagination_consideration_launch(
cache: &CultCache,
state: &EpiphanyThreadState,

897
epiphany-core/src/current_work.rs
View file

@ -734,6 +734,401 @@ pub fn current_frontier_verification_review_job_id(
.and_then(|work| work.job_id))
}
pub fn launch_current_frontier_planning_work(
store_path: impl AsRef<Path>,
created_at: &str,
) -> Result<String> {
let store_path = store_path.as_ref();
chrono::DateTime::parse_from_rfc3339(created_at)
.map_err(|_| anyhow!("frontier Planning launch time is invalid"))?;
let lifecycle = crate::runtime_repo_frontier_planning_lifecycle(store_path)?;
if lifecycle.stage != RepoFrontierPlanningLifecycleStage::ImaginationLaunchReady {
return Err(anyhow!("Mind has no launchable frontier Planning work"));
}
let request_id = lifecycle
.planning_request_id
.ok_or_else(|| anyhow!("frontier Planning work lost its request"))?;
let mut cache = crate::runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let request = cache
.get::<crate::RepoFrontierPlanningRequest>(&request_id)?
.ok_or_else(|| anyhow!("frontier Planning request disappeared"))?;
crate::runtime_spine::validate_actionable_repo_frontier_planning_request(&cache, &request)?;
let identity = cache
.get::<crate::EpiphanyRuntimeIdentity>(crate::RUNTIME_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("frontier Planning launch requires runtime identity"))?;
let mut prior = cache
.get_all::<crate::RepoFrontierPlanningLaunchBinding>()?
.into_iter()
.filter(|binding| binding.planning_request_id == request.request_id)
.collect::<Vec<_>>();
prior.sort_by_key(|binding| binding.attempt_ordinal);
let attempt_ordinal = prior.len() as u64;
let superseded_failure_result_id = if let Some(latest) = prior.last() {
let result = cache
.get::<crate::EpiphanyRuntimeRoleWorkerResult>(&latest.job_id)?
.ok_or_else(|| anyhow!("frontier Planning retry lost its failure result"))?;
crate::runtime_spine::repo_frontier_planning_failure_review(
&cache,
&request.request_id,
"imagination",
&result,
)?
.map(|review| review.result_id)
.ok_or_else(|| anyhow!("frontier Planning retry is not reviewed"))?
.into()
} else {
None
};
let job_id = format!(
"frontier-planning-{}-attempt-{attempt_ordinal}",
request.request_id
);
let launch_document =
crate::EpiphanyWorkerLaunchDocument::Role(crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: job_id.clone(),
role_id: "imagination".into(),
state_revision: 0,
objective: None,
dynamic_prompt_context: None,
repository_body_observation_basis: None,
proposal_modeling_context: None,
frontier_verdict_modeling_context: None,
frontier_planning_context: Some(
crate::RepoFrontierPlanningContextProjection::from_request(&request),
),
frontier_research_context: None,
frontier_verification_context: None,
frontier_plan_mind_context: None,
imagination_consideration_context: None,
admitted_model_direction_consideration_context: None,
active_subgoal_id: None,
active_subgoals: Vec::new(),
active_graph_node_ids: Vec::new(),
investigation_checkpoint: None,
scratch: None,
invariants: Vec::new(),
graphs: None,
recent_evidence: Vec::new(),
recent_observations: Vec::new(),
graph_frontier: None,
graph_checkpoint: None,
planning: None,
churn: None,
});
let authority_scope = "epiphany.role.imagination".to_string();
let output_contract_id = launch_document.output_contract_id().to_string();
let prepared = crate::prepare_runtime_spine_heartbeat_job(
&cache,
crate::RuntimeSpineHeartbeatJobOptions {
runtime_id: identity.runtime_id.clone(),
display_name: "Epiphany Local".into(),
session_id: crate::EPIPHANY_RUNTIME_ROOT_SESSION_ID.into(),
objective: "Propose one bounded plan for the exact frontier authority".into(),
coordinator_note: "Planning current-work launch transaction opened this session."
.into(),
job_id: job_id.clone(),
role: crate::EPIPHANY_IMAGINATION_OWNER_ROLE.into(),
binding_id: crate::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
authority_scope: authority_scope.clone(),
instruction: "Act as Epiphany Imagination. Propose one bounded candidate for only the exact typed frontier and return the dedicated planning candidate contract.".into(),
launch_document,
output_contract_id: output_contract_id.clone(),
organ_launch_contract: crate::default_launch_organ_contract(
&authority_scope,
"role",
&output_contract_id,
),
proposal_modeling_request_id: None,
frontier_planning_request_id: Some(request.request_id.clone()),
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: created_at.into(),
},
)?;
let launch = prepared
.envelopes
.iter()
.find(|envelope| {
envelope.r#type == crate::EpiphanyRuntimeWorkerLaunchRequest::TYPE
&& envelope.key == job_id
})
.ok_or_else(|| anyhow!("frontier Planning launch preparation lost its worker"))?;
let worker_launch: crate::EpiphanyRuntimeWorkerLaunchRequest =
rmp_serde::from_slice(&launch.payload)?;
let binding_record_id = if attempt_ordinal == 0 {
format!("repo-frontier-planning-launch-{}", request.request_id)
} else {
format!(
"repo-frontier-planning-launch-{}-attempt-{attempt_ordinal}",
request.request_id
)
};
let binding = crate::RepoFrontierPlanningLaunchBinding {
schema_version: crate::REPO_FRONTIER_PLANNING_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id,
planning_request_id: request.request_id.clone(),
job_id: job_id.clone(),
binding_id: crate::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
runtime_id: identity.runtime_id,
thread_id: job_id.clone(),
launched_at: created_at.into(),
worker_launch_document_sha256: format!(
"{:x}",
Sha256::digest(&worker_launch.launch_document_msgpack)
),
contract: crate::REPO_FRONTIER_PLANNING_LAUNCH_BINDING_CONTRACT.into(),
attempt_ordinal,
superseded_failure_result_id,
};
let snapshot = cache.snapshot_envelopes();
let mut expected = vec![
cache
.get_envelope::<crate::RepoFrontierPlanningRequest>(&request.request_id)?
.ok_or_else(|| anyhow!("frontier Planning launch lost its request envelope"))?,
];
for source in request
.frontier_authority_documents
.iter()
.chain(request.claim_obligation_documents.iter())
{
let envelope = snapshot
.iter()
.find(|envelope| {
envelope.r#type == source.document_type && envelope.key == source.document_key
})
.ok_or_else(|| anyhow!("frontier Planning launch lost exact authority"))?;
if EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", envelope)? != *source {
return Err(anyhow!("frontier Planning launch authority changed"));
}
expected.push(envelope.clone());
}
commit_current_work_launch(
store_path,
&cache,
expected,
prepared.envelopes,
vec![cache.prepare_entry(&binding.binding_record_id, &binding)?.0],
"frontier Planning",
)?;
Ok(job_id)
}
pub fn launch_current_frontier_plan_mind_work(
store_path: impl AsRef<Path>,
created_at: &str,
) -> Result<String> {
let store_path = store_path.as_ref();
chrono::DateTime::parse_from_rfc3339(created_at)
.map_err(|_| anyhow!("frontier plan Mind launch time is invalid"))?;
let lifecycle = crate::runtime_repo_frontier_planning_lifecycle(store_path)?;
if lifecycle.stage != RepoFrontierPlanningLifecycleStage::MindLaunchReady {
return Err(anyhow!("Mind has no launchable frontier plan review"));
}
let planning_request_id = lifecycle
.planning_request_id
.ok_or_else(|| anyhow!("frontier plan Mind launch lost its Planning request"))?;
let request_id = lifecycle
.mind_request_id
.ok_or_else(|| anyhow!("frontier plan Mind launch lost its request"))?;
let mut cache = crate::runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let request = cache
.get::<crate::RepoFrontierPlanMindRequest>(&request_id)?
.ok_or_else(|| anyhow!("frontier plan Mind request disappeared"))?;
let (planning, candidate) =
crate::runtime_spine::validate_repo_frontier_plan_mind_request(&cache, &request)?;
if planning.request_id != planning_request_id {
return Err(anyhow!(
"frontier plan Mind request crossed Planning authority"
));
}
let identity = cache
.get::<crate::EpiphanyRuntimeIdentity>(crate::RUNTIME_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("frontier plan Mind launch requires runtime identity"))?;
let mut prior = cache
.get_all::<crate::RepoFrontierPlanMindLaunchBinding>()?
.into_iter()
.filter(|binding| binding.mind_request_id == request.request_id)
.collect::<Vec<_>>();
prior.sort_by_key(|binding| binding.attempt_ordinal);
let attempt_ordinal = prior.len() as u64;
let superseded_failure_result_id = if let Some(latest) = prior.last() {
let result = cache
.get::<crate::EpiphanyRuntimeRoleWorkerResult>(&latest.job_id)?
.ok_or_else(|| anyhow!("frontier plan Mind retry lost its failure result"))?;
crate::runtime_spine::repo_frontier_planning_failure_review(
&cache,
&planning.request_id,
"mind",
&result,
)?
.map(|review| review.result_id)
.ok_or_else(|| anyhow!("frontier plan Mind retry is not reviewed"))?
.into()
} else {
None
};
let job_id = format!(
"frontier-plan-mind-{}-attempt-{attempt_ordinal}",
request.request_id
);
let launch_document =
crate::EpiphanyWorkerLaunchDocument::Role(crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: job_id.clone(),
role_id: "mindAdmissionReview".into(),
state_revision: 0,
objective: None,
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: Some(crate::RepoFrontierPlanMindContextProjection::new(
&request, &planning, &candidate,
)),
imagination_consideration_context: None,
admitted_model_direction_consideration_context: None,
active_subgoal_id: None,
active_subgoals: Vec::new(),
active_graph_node_ids: Vec::new(),
investigation_checkpoint: None,
scratch: None,
invariants: Vec::new(),
graphs: None,
recent_evidence: Vec::new(),
recent_observations: Vec::new(),
graph_frontier: None,
graph_checkpoint: None,
planning: None,
churn: None,
});
let authority_scope = "epiphany.role.mind".to_string();
let output_contract_id = launch_document.output_contract_id().to_string();
let prepared = crate::prepare_runtime_spine_heartbeat_job(
&cache,
crate::RuntimeSpineHeartbeatJobOptions {
runtime_id: identity.runtime_id.clone(),
display_name: "Epiphany Local".into(),
session_id: crate::EPIPHANY_RUNTIME_ROOT_SESSION_ID.into(),
objective: "Adjudicate one exact Imagination plan candidate".into(),
coordinator_note: "Plan Mind current-work launch transaction opened this session."
.into(),
job_id: job_id.clone(),
role: crate::EPIPHANY_MIND_OWNER_ROLE.into(),
binding_id: crate::EPIPHANY_MIND_ROLE_BINDING_ID.into(),
authority_scope: authority_scope.clone(),
instruction: "Act as Epiphany Mind. Adopt, refuse, or hold only the exact typed candidate and return the dedicated plan decision contract.".into(),
launch_document,
output_contract_id: output_contract_id.clone(),
organ_launch_contract: crate::default_launch_organ_contract(
&authority_scope,
"role",
&output_contract_id,
),
proposal_modeling_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: Some(request.request_id.clone()),
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: created_at.into(),
},
)?;
let worker_envelope = prepared
.envelopes
.iter()
.find(|envelope| {
envelope.r#type == crate::EpiphanyRuntimeWorkerLaunchRequest::TYPE
&& envelope.key == job_id
})
.ok_or_else(|| anyhow!("frontier plan Mind preparation lost its worker"))?;
let worker_launch: crate::EpiphanyRuntimeWorkerLaunchRequest =
rmp_serde::from_slice(&worker_envelope.payload)?;
let binding_record_id = if attempt_ordinal == 0 {
format!("repo-frontier-plan-mind-launch-{}", request.request_id)
} else {
format!(
"repo-frontier-plan-mind-launch-{}-attempt-{attempt_ordinal}",
request.request_id
)
};
let binding = crate::RepoFrontierPlanMindLaunchBinding {
schema_version: crate::REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id,
mind_request_id: request.request_id.clone(),
job_id: job_id.clone(),
binding_id: crate::EPIPHANY_MIND_ROLE_BINDING_ID.into(),
runtime_id: identity.runtime_id,
thread_id: job_id.clone(),
launched_at: created_at.into(),
worker_launch_document_sha256: format!(
"{:x}",
Sha256::digest(&worker_launch.launch_document_msgpack)
),
contract: crate::REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_CONTRACT.into(),
attempt_ordinal,
superseded_failure_result_id,
};
let snapshot = cache.snapshot_envelopes();
let mut expected = Vec::new();
for (document_type, document_key) in [
(
crate::RepoFrontierPlanMindRequest::TYPE,
request.request_id.as_str(),
),
(
crate::RepoFrontierPlanningRequest::TYPE,
planning.request_id.as_str(),
),
(
crate::EpiphanyRuntimeRoleWorkerResult::TYPE,
request.imagination_job_id.as_str(),
),
] {
expected.push(
snapshot
.iter()
.find(|envelope| envelope.r#type == document_type && envelope.key == document_key)
.cloned()
.ok_or_else(|| anyhow!("frontier plan Mind launch lost a strong source"))?,
);
}
for source in planning
.frontier_authority_documents
.iter()
.chain(planning.claim_obligation_documents.iter())
{
let envelope = snapshot
.iter()
.find(|envelope| {
envelope.r#type == source.document_type && envelope.key == source.document_key
})
.ok_or_else(|| anyhow!("frontier plan Mind launch lost exact authority"))?;
if EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", envelope)? != *source {
return Err(anyhow!("frontier plan Mind authority changed"));
}
expected.push(envelope.clone());
}
commit_current_work_launch(
store_path,
&cache,
expected,
prepared.envelopes,
vec![cache.prepare_entry(&binding.binding_record_id, &binding)?.0],
"frontier plan Mind",
)?;
Ok(job_id)
}
pub fn launch_current_frontier_verification_work(
store_path: impl AsRef<Path>,
created_at: &str,
@ -3617,6 +4012,508 @@ mod tests {
.any(|node| node.id == "concurrent-unrelated-node")
);
// Planning is one frontier-owned two-pass current-work family. The
// complete RepoModel is audit cargo; exact frontier/dependency and
// per-claim obligation documents own admission conflicts.
let planning_frontier = RepoFrontierItem {
id: "planning-frontier".into(),
migration_body: "epiphany".into(),
question: "How should the keyed Mind cut continue?".into(),
gap: "The exact frontier needs one bounded plan.".into(),
target_claim_ids: vec!["body-node".into()],
source_scope: vec!["epiphany-core/src".into()],
recommended_next_organ: "Imagination".into(),
status: RepoFrontierStatus::Active,
evidence_refs: vec![verdict.receipt_id.clone()],
..Default::default()
};
let planning_proposal = crate::EpiphanyRepoModelMutationProposal::new(
"repo-model-mutation-proposal-planning-frontier",
"planning-frontier-fixture-request",
"planning-frontier-fixture-result",
vec![verdict.receipt_id.clone()],
body.clone(),
vec![crate::EpiphanyRepoModelMutationOperation::PutFrontier {
item: planning_frontier.clone(),
}],
)?;
let planning_plan = crate::plan_repo_model_mutation(&store, &planning_proposal)?;
let mut planning_cache = crate::runtime_spine_cache(&store)?;
planning_cache.pull_all_backing_stores()?;
let planning_provenance = planning_cache
.get_envelope::<crate::SoulVerdictReceipt>(&verdict.receipt_id)?
.expect("typed planning provenance");
assert!(matches!(
crate::commit_typed_organ_mind_mutation(
&store,
"Modeling",
planning_provenance,
"Modeling.planning_obligation_fixture",
planning_plan.strong_reads,
planning_plan.writes,
"2026-08-17T00:00:21.100Z",
)?,
crate::EpiphanyMindCommitOutcome::Committed(_)
));
let planning_request = crate::select_and_commit_repo_frontier_planning_request(
&store,
"2026-08-17T00:00:21.200Z",
)?;
assert_eq!(planning_request.frontier_item_id, planning_frontier.id);
assert_eq!(planning_request.claim_obligation_documents.len(), 1);
assert!(
planning_request.frontier_authority_documents.len()
< planning_request.model_source_documents.len()
);
assert!(
planning_cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.is_none()
);
let challenged_plan_store = temp.path().join("challenged-plan.cc");
std::fs::copy(&store, &challenged_plan_store)?;
let mut challenged_cache = crate::runtime_spine_cache(&challenged_plan_store)?;
challenged_cache.pull_all_backing_stores()?;
let challenged_model = crate::assemble_repo_model_view(&challenged_plan_store)?;
let challenged_claim = challenged_model
.nodes
.iter()
.find(|node| node.id == "body-node")
.expect("planning claim");
let packet = crate::EyesEvidencePacket {
schema_version: crate::EYES_EVIDENCE_PACKET_SCHEMA_VERSION.into(),
packet_id: "planning-challenge-packet".into(),
source_result_id: "planning-challenge-result".into(),
source_job_id: "planning-challenge-job".into(),
source_role_id: "research".into(),
evidence_ids: vec!["planning-challenge-evidence".into()],
observation_ids: vec!["planning-challenge-observation".into()],
source_refs: vec![
"github://GameCult/Epiphany@0000000000000000000000000000000000000000/README.md"
.into(),
],
summary: "External evidence challenges the planning claim.".into(),
uncertainty: "Bounded fixture uncertainty.".into(),
emitted_at: "2026-08-17T00:00:21.300Z".into(),
contract: "epiphany.eyes.evidence_packet.fixture".into(),
source_lookup_receipt_ids: Vec::new(),
research_request_id: "planning-challenge-research-request".into(),
decision_context_id: "planning-challenge-decision-context".into(),
};
crate::runtime_spine::put_eyes_evidence_packet(&challenged_plan_store, &packet)?;
let challenge = crate::RepoModelClaimChallenge {
schema_version: crate::REPO_MODEL_CLAIM_CHALLENGE_SCHEMA_VERSION.into(),
challenge_id: "planning-claim-challenge".into(),
eyes_evidence_packet_id: packet.packet_id.clone(),
eyes_evidence_packet_sha256: format!(
"{:x}",
Sha256::digest(rmp_serde::to_vec_named(&packet)?)
),
source_result_id: packet.source_result_id.clone(),
source_job_id: packet.source_job_id.clone(),
model_projection_digest: planning_request.model_projection_digest.clone(),
model_source_documents: planning_request.model_source_documents.clone(),
target_claim_id: challenged_claim.id.clone(),
target_claim_sha256: format!(
"{:x}",
Sha256::digest(rmp_serde::to_vec_named(challenged_claim)?)
),
disposition: crate::RepoModelClaimChallengeDisposition::EvidenceInsufficient,
finding: "The exact target claim requires external correction.".into(),
uncertainty: "The plan must be regenerated after evidence admission.".into(),
source_refs: packet.source_refs.clone(),
evidence_ids: packet.evidence_ids.clone(),
challenged_at: "2026-08-17T00:00:21.400Z".into(),
contract: crate::REPO_MODEL_CLAIM_CHALLENGE_CONTRACT.into(),
};
crate::commit_repo_model_claim_challenge(&challenged_plan_store, &challenge)?;
challenged_cache.pull_all_backing_stores()?;
let challenged_snapshot = challenged_cache.snapshot_envelopes();
assert!(
crate::runtime_spine::validate_actionable_repo_frontier_planning_request(
&challenged_cache,
&planning_request,
)
.is_err()
);
assert!(
launch_current_frontier_planning_work(
&challenged_plan_store,
"2026-08-17T00:00:21.500Z",
)
.is_err()
);
challenged_cache.pull_all_backing_stores()?;
assert_eq!(challenged_cache.snapshot_envelopes(), challenged_snapshot);
let mut disjoint_planning_node = final_model
.nodes
.iter()
.find(|node| node.id == "body-node")
.cloned()
.expect("seeded Body node");
disjoint_planning_node.id = "planning-concurrent-node".into();
disjoint_planning_node.title = "Planning concurrent node".into();
disjoint_planning_node.claim = "Unrelated keyed work merges with planning".into();
let disjoint_planning_proposal = crate::EpiphanyRepoModelMutationProposal::new(
"repo-model-mutation-proposal-planning-concurrent",
"planning-concurrent-request",
"planning-concurrent-result",
vec![verdict.receipt_id.clone()],
body.clone(),
vec![crate::EpiphanyRepoModelMutationOperation::PutNode {
node: disjoint_planning_node,
}],
)?;
let disjoint_planning_plan =
crate::plan_repo_model_mutation(&store, &disjoint_planning_proposal)?;
planning_cache.pull_all_backing_stores()?;
let disjoint_planning_provenance = planning_cache
.get_envelope::<crate::SoulVerdictReceipt>(&verdict.receipt_id)?
.expect("typed disjoint planning provenance");
assert!(matches!(
crate::commit_typed_organ_mind_mutation(
&store,
"Modeling",
disjoint_planning_provenance,
"Modeling.concurrent_with_planning_fixture",
disjoint_planning_plan.strong_reads,
disjoint_planning_plan.writes,
"2026-08-17T00:00:21.600Z",
)?,
crate::EpiphanyMindCommitOutcome::Committed(_)
));
let mut current_planning_cache = crate::runtime_spine_cache(&store)?;
current_planning_cache.pull_all_backing_stores()?;
crate::runtime_spine::validate_actionable_repo_frontier_planning_request(
&current_planning_cache,
&planning_request,
)?;
let planning_job =
launch_current_frontier_planning_work(&store, "2026-08-17T00:00:21.700Z")?;
let mut planning_cache = crate::runtime_spine_cache(&store)?;
planning_cache.pull_all_backing_stores()?;
let planning_launch = planning_cache
.get::<crate::EpiphanyRuntimeWorkerLaunchRequest>(&planning_job)?
.expect("exact Planning launch");
assert_eq!(
planning_launch.frontier_planning_request_id.as_deref(),
Some(planning_request.request_id.as_str())
);
let planning_basis = crate::worker_reasoning_basis(&store, &planning_launch)?;
crate::put_reasoning_basis(&store, &planning_basis)?;
let mut planning_native = epiphany_model_adapter::EpiphanyModelRequest::new(
"planning-model-request",
"planning-conversation",
"openai-codex",
"gpt-test",
"imagine",
);
planning_native.reasoning_basis_id = Some(planning_basis.basis_id.clone());
planning_native.source_worker_job_id = Some(planning_job.clone());
planning_native.output_contract_id = Some(planning_launch.output_contract_id.clone());
planning_native.output_schema_json = Some(serde_json::to_string(
&crate::epiphany_frontier_planning_output_schema(),
)?);
let planning_provider = epiphany_openai_adapter::request_from_native(&planning_native);
crate::open_runtime_model_execution(
&store,
crate::RuntimeSpineSessionOptions {
session_id: "planning-model-session".into(),
objective: "Propose the exact frontier plan.".into(),
created_at: "2026-08-17T00:00:21.750Z".into(),
coordinator_note: "Bound Planning model pass.".into(),
},
crate::RuntimeSpineJobOptions {
job_id: "planning-model-job".into(),
session_id: "planning-model-session".into(),
role: "openai-model".into(),
created_at: "2026-08-17T00:00:21.750Z".into(),
summary: "Bound Planning inference.".into(),
artifact_refs: Vec::new(),
},
&planning_native,
&planning_provider,
"2026-08-17T00:00:21.750Z",
)?;
let planning_context = crate::EpiphanyDecisionContext::new(
&planning_basis,
planning_native,
planning_provider,
Vec::new(),
)?;
crate::put_decision_context(&store, &planning_context)?;
let mut candidate = crate::RepoFrontierPlanCandidate {
schema_version: crate::REPO_FRONTIER_PLAN_CANDIDATE_SCHEMA_VERSION.into(),
candidate_id: "pending".into(),
planning_request_id: planning_request.request_id.clone(),
model_projection_digest: planning_request.model_projection_digest.clone(),
model_source_documents: planning_request.model_source_documents.clone(),
frontier_item_id: planning_request.frontier_item_id.clone(),
frontier_item_hash: planning_request.frontier_item_hash.clone(),
safe_paths: planning_request.source_scope.clone(),
action: "Replace aggregate planning authority with keyed current work.".into(),
command: "cargo test --manifest-path epiphany-core/Cargo.toml --lib".into(),
checks: vec!["planning lifecycle is thread-free".into()],
stop_conditions: vec!["exact frontier authority changes".into()],
rollback_steps: vec!["revert the keyed Planning commit".into()],
commit_message: "Migrate Planning to keyed Mind authority".into(),
proposed_at: "2026-08-17T00:00:21.800Z".into(),
contract: crate::REPO_FRONTIER_PLANNING_CONTRACT.into(),
};
candidate.candidate_id = crate::canonical_repo_frontier_plan_candidate_id(&candidate)?;
let planning_result = crate::EpiphanyRuntimeRoleWorkerResult {
schema_version: crate::RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION.into(),
result_id: "planning-result".into(),
job_id: planning_job.clone(),
role_id: "imagination".into(),
verdict: "candidate-ready".into(),
summary: "proposed one exact bounded plan".into(),
next_safe_move: "request Mind review".into(),
checkpoint_summary: None,
scratch_summary: None,
files_inspected: Vec::new(),
frontier_node_ids: vec!["body-node".into()],
evidence_ids: vec![verdict.receipt_id.clone()],
artifact_refs: Vec::new(),
open_questions: Vec::new(),
evidence_gaps: Vec::new(),
risks: Vec::new(),
state_patch_msgpack: None,
self_patch_msgpack: None,
item_error: None,
metadata: Default::default(),
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: Some(planning_request.request_id.clone()),
frontier_plan_candidate_msgpack: Some(rmp_serde::to_vec_named(&candidate)?),
frontier_plan_mind_request_id: None,
frontier_plan_mind_decision_msgpack: None,
repository_body_observation_basis: None,
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: planning_context.context_id.clone(),
};
let planning_process = crate::ProcessInstanceIdentity {
process_id: 46,
creation_token: 11,
created_at_rfc3339: Some("2026-08-17T00:00:21.900Z".into()),
executable_path: "planning-worker".into(),
};
let planning_activation = "planning-activation";
crate::claim_runtime_worker_process(
&store,
&planning_job,
&planning_process,
&format!("{:x}", Sha256::digest(planning_activation.as_bytes())),
"2026-08-17T00:00:21.900Z",
)?;
crate::activate_runtime_worker_process(
&store,
&planning_job,
&planning_process,
planning_activation,
"2026-08-17T00:00:21.950Z",
)?;
crate::put_runtime_role_worker_result(&store, &planning_result)?;
crate::complete_runtime_job(
&store,
crate::RuntimeSpineJobResultOptions {
result_id: format!("runtime-result-{planning_job}"),
job_id: planning_job.clone(),
completed_at: "2026-08-17T00:00:22Z".into(),
verdict: planning_result.verdict.clone(),
summary: planning_result.summary.clone(),
next_safe_move: planning_result.next_safe_move.clone(),
evidence_refs: planning_result.evidence_ids.clone(),
artifact_refs: Vec::new(),
decision_context_id: Some(planning_context.context_id.clone()),
},
)?;
let mind_request = crate::commit_repo_frontier_plan_mind_request(
&store,
&planning_result.result_id,
"2026-08-17T00:00:22.100Z",
)?;
let mind_job = launch_current_frontier_plan_mind_work(&store, "2026-08-17T00:00:22.200Z")?;
planning_cache.pull_all_backing_stores()?;
let mind_launch = planning_cache
.get::<crate::EpiphanyRuntimeWorkerLaunchRequest>(&mind_job)?
.expect("exact plan Mind launch");
assert_eq!(
mind_launch.frontier_plan_mind_request_id.as_deref(),
Some(mind_request.request_id.as_str())
);
let mind_basis = crate::worker_reasoning_basis(&store, &mind_launch)?;
crate::put_reasoning_basis(&store, &mind_basis)?;
let mut mind_native = epiphany_model_adapter::EpiphanyModelRequest::new(
"plan-mind-model-request",
"plan-mind-conversation",
"openai-codex",
"gpt-test",
"judge",
);
mind_native.reasoning_basis_id = Some(mind_basis.basis_id.clone());
mind_native.source_worker_job_id = Some(mind_job.clone());
mind_native.output_contract_id = Some(mind_launch.output_contract_id.clone());
mind_native.output_schema_json = Some(serde_json::to_string(
&crate::epiphany_frontier_plan_mind_output_schema(),
)?);
let mind_provider = epiphany_openai_adapter::request_from_native(&mind_native);
crate::open_runtime_model_execution(
&store,
crate::RuntimeSpineSessionOptions {
session_id: "plan-mind-model-session".into(),
objective: "Adjudicate the exact frontier plan.".into(),
created_at: "2026-08-17T00:00:22.250Z".into(),
coordinator_note: "Bound plan Mind model pass.".into(),
},
crate::RuntimeSpineJobOptions {
job_id: "plan-mind-model-job".into(),
session_id: "plan-mind-model-session".into(),
role: "openai-model".into(),
created_at: "2026-08-17T00:00:22.250Z".into(),
summary: "Bound plan Mind inference.".into(),
artifact_refs: Vec::new(),
},
&mind_native,
&mind_provider,
"2026-08-17T00:00:22.250Z",
)?;
let mind_context = crate::EpiphanyDecisionContext::new(
&mind_basis,
mind_native,
mind_provider,
Vec::new(),
)?;
crate::put_decision_context(&store, &mind_context)?;
let mind_decision = crate::RepoFrontierPlanMindDecision {
mind_request_id: mind_request.request_id.clone(),
planning_request_id: planning_request.request_id.clone(),
imagination_result_id: planning_result.result_id.clone(),
candidate_id: candidate.candidate_id.clone(),
candidate_sha256: mind_request.candidate_sha256.clone(),
decision: crate::RepoFrontierPlanDecision::Adopt,
rationale: "The candidate is bounded by the exact frontier and claim guard.".into(),
decided_at: "2026-08-17T00:00:22.300Z".into(),
};
let mind_result = crate::EpiphanyRuntimeRoleWorkerResult {
schema_version: crate::RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION.into(),
result_id: "plan-mind-result".into(),
job_id: mind_job.clone(),
role_id: "mindAdmissionReview".into(),
verdict: "adopt".into(),
summary: "adopted the exact bounded plan".into(),
next_safe_move: "commit plan decision".into(),
checkpoint_summary: None,
scratch_summary: None,
files_inspected: Vec::new(),
frontier_node_ids: vec!["body-node".into()],
evidence_ids: vec![candidate.candidate_id.clone()],
artifact_refs: Vec::new(),
open_questions: Vec::new(),
evidence_gaps: Vec::new(),
risks: Vec::new(),
state_patch_msgpack: None,
self_patch_msgpack: None,
item_error: None,
metadata: Default::default(),
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: Some(mind_request.request_id.clone()),
frontier_plan_mind_decision_msgpack: Some(rmp_serde::to_vec_named(&mind_decision)?),
repository_body_observation_basis: None,
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: mind_context.context_id.clone(),
};
let mind_process = crate::ProcessInstanceIdentity {
process_id: 47,
creation_token: 12,
created_at_rfc3339: Some("2026-08-17T00:00:22.400Z".into()),
executable_path: "plan-mind-worker".into(),
};
let mind_activation = "plan-mind-activation";
crate::claim_runtime_worker_process(
&store,
&mind_job,
&mind_process,
&format!("{:x}", Sha256::digest(mind_activation.as_bytes())),
"2026-08-17T00:00:22.400Z",
)?;
crate::activate_runtime_worker_process(
&store,
&mind_job,
&mind_process,
mind_activation,
"2026-08-17T00:00:22.450Z",
)?;
crate::put_runtime_role_worker_result(&store, &mind_result)?;
crate::complete_runtime_job(
&store,
crate::RuntimeSpineJobResultOptions {
result_id: format!("runtime-result-{mind_job}"),
job_id: mind_job.clone(),
completed_at: "2026-08-17T00:00:22.500Z".into(),
verdict: mind_result.verdict.clone(),
summary: mind_result.summary.clone(),
next_safe_move: mind_result.next_safe_move.clone(),
evidence_refs: mind_result.evidence_ids.clone(),
artifact_refs: Vec::new(),
decision_context_id: Some(mind_context.context_id.clone()),
},
)?;
let plan_decision =
crate::commit_repo_frontier_plan_decision(&store, &mind_result.result_id)?;
assert_eq!(
plan_decision.decision,
crate::RepoFrontierPlanDecision::Adopt
);
assert_eq!(
crate::commit_repo_frontier_plan_decision(&store, &mind_result.result_id)?,
plan_decision
);
let adopted_model = crate::assemble_repo_model_view(&store)?;
let adopted_frontier = adopted_model
.frontier
.iter()
.find(|item| item.id == planning_frontier.id)
.expect("adopted Planning frontier");
assert_eq!(
adopted_frontier
.adopted_plan
.as_ref()
.expect("Mind adoption")
.candidate_id,
candidate.candidate_id
);
assert!(
adopted_model
.nodes
.iter()
.any(|node| node.id == "planning-concurrent-node")
);
assert!(
planning_cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.is_none()
);
// Eyes is its own exact current-work family. It launches from an
// explicit external-evidence obligation, survives unrelated keyed
// Mind commits, and conflicts if that exact frontier changes.

2
epiphany-core/src/lib.rs
View file

@ -103,7 +103,6 @@ pub use agent_launch::EpiphanyRolePromptConfig;
pub use agent_launch::EpiphanySharedPromptConfig;
pub use agent_launch::EpiphanySpecialistPromptConfig;
pub use agent_launch::build_epiphany_admitted_model_direction_consideration_launch_request;
pub use agent_launch::build_epiphany_frontier_plan_mind_launch_request;
pub use agent_launch::build_epiphany_imagination_consideration_launch_request;
pub use agent_launch::build_epiphany_job_launch_request;
pub use agent_launch::build_epiphany_reorient_launch_instruction;
@ -1115,6 +1114,7 @@ pub use runtime_spine::retain_completed_runtime_sessions;
pub use runtime_spine::retain_coordinator_run_receipts;
pub use runtime_spine::retain_failed_runtime_worker_attempts;
pub use runtime_spine::retain_fulfilled_runtime_worker_attempts;
pub use runtime_spine::review_repo_frontier_planning_failure;
pub use runtime_spine::runtime_authenticated_public_source_lookups_for_worker;
pub use runtime_spine::runtime_continuity_recovery_receipt;
pub use runtime_spine::runtime_current_repo_model;

101
epiphany-core/src/repo_model_documents.rs
View file

@ -100,7 +100,7 @@ repo_document!(
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "epiphany.mind.repo_model.claim_obligations.v1",
type = "epiphany.mind.repo_model.claim_obligations.v2",
schema = "EpiphanyRepoModelClaimObligationsDocument"
)]
pub struct EpiphanyRepoModelClaimObligationsDocument {
@ -108,6 +108,8 @@ pub struct EpiphanyRepoModelClaimObligationsDocument {
pub node_id: String,
#[cultcache(key = 1)]
pub unresolved_frontier_ids: Vec<String>,
#[cultcache(key = 2)]
pub active_challenge_ids: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@ -357,7 +359,7 @@ impl EpiphanyRepoModelSeed {
require_semantic_id(value, label)?;
}
let documents = self.documents()?;
let obligations = claim_obligations_for_frontier(&documents.frontier);
let obligations = claim_obligations_for_frontier(&documents.nodes, &documents.frontier);
validate_repo_model_parts(
&EpiphanyRepoModelIdentityDocument {
schema_epoch: REPO_MODEL_SCHEMA_EPOCH.into(),
@ -573,7 +575,7 @@ pub fn initialize_keyed_repo_model(
}
let documents = seed.documents()?;
let obligations = claim_obligations_for_frontier(&documents.frontier);
let obligations = claim_obligations_for_frontier(&documents.nodes, &documents.frontier);
let mut writes = Vec::new();
writes.push(cache.prepare_entry(REPO_MODEL_IDENTITY_KEY, &identity)?.0);
for value in &documents.domains {
@ -646,7 +648,7 @@ pub fn initialize_keyed_repo_model(
summaries: documents.summaries.clone(),
frontier: documents.frontier.clone(),
lifecycle_receipts: documents.lifecycle_receipts.clone(),
claim_obligations: claim_obligations_for_frontier(&documents.frontier),
claim_obligations: claim_obligations_for_frontier(&documents.nodes, &documents.frontier),
surface_offers: Vec::new(),
dependency_claims: Vec::new(),
dependency_verifications: Vec::new(),
@ -692,7 +694,8 @@ fn repo_model_view_matches_seed(
&& view.summaries == documents.summaries
&& view.frontier == documents.frontier
&& view.lifecycle_receipts == documents.lifecycle_receipts
&& view.claim_obligations == claim_obligations_for_frontier(&documents.frontier))
&& view.claim_obligations
== claim_obligations_for_frontier(&documents.nodes, &documents.frontier))
}
pub(crate) fn repo_model_write_key(envelope: &CultCacheEnvelope) -> Result<Option<String>> {
@ -939,10 +942,24 @@ pub fn plan_repo_model_mutation(
insert_strong_envelope(&mut strong, existing)?;
}
nodes.insert(node.id.clone(), node.clone());
let obligation = obligations.entry(node.id.clone()).or_insert_with(|| {
EpiphanyRepoModelClaimObligationsDocument {
node_id: node.id.clone(),
unresolved_frontier_ids: Vec::new(),
active_challenge_ids: Vec::new(),
}
});
if let Some(existing) =
cache.get_envelope::<EpiphanyRepoModelClaimObligationsDocument>(&node.id)?
{
insert_strong_envelope(&mut strong, existing)?;
}
obligation.active_challenge_ids.clear();
let envelope = cache
.prepare_entry(&node.id, &EpiphanyRepoModelNodeDocument::new(node)?)?
.0;
insert_envelope(&mut writes, envelope)?;
insert_envelope(&mut writes, cache.prepare_entry(&node.id, obligation)?.0)?;
}
EpiphanyRepoModelMutationOperation::RetireNode { node_id } => {
require_semantic_id(node_id, "RepoModel node")?;
@ -958,11 +975,14 @@ pub fn plan_repo_model_mutation(
EpiphanyRepoModelClaimObligationsDocument {
node_id: node_id.clone(),
unresolved_frontier_ids: Vec::new(),
active_challenge_ids: Vec::new(),
}
});
if !obligation.unresolved_frontier_ids.is_empty() {
if !obligation.unresolved_frontier_ids.is_empty()
|| !obligation.active_challenge_ids.is_empty()
{
return Err(anyhow!(
"RepoModel node retirement is blocked by unresolved frontier"
"RepoModel node retirement is blocked by unresolved claim obligations"
));
}
if let Some(existing) =
@ -1108,6 +1128,7 @@ pub fn plan_repo_model_mutation(
EpiphanyRepoModelClaimObligationsDocument {
node_id: node_id.clone(),
unresolved_frontier_ids: Vec::new(),
active_challenge_ids: Vec::new(),
}
});
if let Some(existing) =
@ -1510,9 +1531,13 @@ fn frontier_is_unresolved(item: &RepoFrontierItem) -> bool {
}
fn claim_obligations_for_frontier(
nodes: &[EpiphanyMemoryNode],
frontier: &[RepoFrontierItem],
) -> Vec<EpiphanyRepoModelClaimObligationsDocument> {
let mut obligations = BTreeMap::<String, BTreeSet<String>>::new();
let mut obligations = nodes
.iter()
.map(|node| (node.id.clone(), BTreeSet::<String>::new()))
.collect::<BTreeMap<_, _>>();
for item in frontier.iter().filter(|item| frontier_is_unresolved(item)) {
for node_id in &item.target_claim_ids {
obligations
@ -1527,6 +1552,7 @@ fn claim_obligations_for_frontier(
|(node_id, unresolved_frontier_ids)| EpiphanyRepoModelClaimObligationsDocument {
node_id,
unresolved_frontier_ids: unresolved_frontier_ids.into_iter().collect(),
active_challenge_ids: Vec::new(),
},
)
.collect()
@ -1602,6 +1628,37 @@ pub(crate) fn assemble_repo_model_view_from_cache(
lifecycle_receipts.clone(),
claim_obligations.clone(),
)?;
let challenges = cache.get_all::<crate::RepoModelClaimChallenge>()?;
for obligation in &claim_obligations {
for challenge_id in &obligation.active_challenge_ids {
let challenge = challenges
.iter()
.find(|challenge| challenge.challenge_id == *challenge_id)
.ok_or_else(|| anyhow!("claim obligation names a missing challenge"))?;
if challenge.target_claim_id != obligation.node_id {
return Err(anyhow!("claim obligation challenge targets another node"));
}
}
}
for challenge in &challenges {
let Some(node) = nodes
.iter()
.find(|node| node.id == challenge.target_claim_id)
else {
continue;
};
let node_sha256 = format!("{:x}", Sha256::digest(rmp_serde::to_vec_named(node)?));
if node_sha256 == challenge.target_claim_sha256
&& !claim_obligations.iter().any(|obligation| {
obligation.node_id == challenge.target_claim_id
&& obligation
.active_challenge_ids
.contains(&challenge.challenge_id)
})
{
return Err(anyhow!("current claim challenge lost its claim obligation"));
}
}
let mut source_documents = cache
.snapshot_envelopes()
@ -1681,10 +1738,30 @@ fn validate_claim_obligations(
frontier: &[RepoFrontierItem],
obligations: &[EpiphanyRepoModelClaimObligationsDocument],
) -> Result<()> {
if obligations.len() != nodes.len() {
return Err(anyhow!(
"every RepoModel node requires one claim obligation"
));
}
for obligation in obligations {
if !nodes.iter().any(|node| node.id == obligation.node_id) {
return Err(anyhow!("RepoModel claim obligation names a missing node"));
}
if !obligation
.unresolved_frontier_ids
.windows(2)
.all(|pair| pair[0] < pair[1])
|| !obligation
.active_challenge_ids
.windows(2)
.all(|pair| pair[0] < pair[1])
|| obligation
.active_challenge_ids
.iter()
.any(|id| id.trim().is_empty())
{
return Err(anyhow!("RepoModel claim obligation is not canonical"));
}
for frontier_id in &obligation.unresolved_frontier_ids {
let item = frontier
.iter()
@ -1915,6 +1992,14 @@ mod tests {
..Default::default()
})?,
)?;
cache.put(
"node-1",
&EpiphanyRepoModelClaimObligationsDocument {
node_id: "node-1".into(),
unresolved_frontier_ids: Vec::new(),
active_challenge_ids: Vec::new(),
},
)?;
let first = assemble_repo_model_view(&store)?;
let second = assemble_repo_model_view(&store)?;
assert_eq!(first, second);

46
epiphany-core/src/repo_model_gateway.rs
View file

@ -62,7 +62,7 @@ pub const REPO_FRONTIER_MODELING_REQUEST_CONTRACT: &str =
pub const REPO_FRONTIER_WORK_PROPOSAL_SCHEMA_VERSION: &str =
"epiphany.repo_frontier_work_proposal.v0";
pub const REPO_FRONTIER_PLANNING_REQUEST_SCHEMA_VERSION: &str =
"epiphany.self.repo_frontier_planning_request.v2";
"epiphany.self.repo_frontier_planning_request.v4";
pub const REPO_FRONTIER_PLAN_CANDIDATE_SCHEMA_VERSION: &str =
"epiphany.imagination.repo_frontier_plan_candidate.v0";
pub const REPO_FRONTIER_PLAN_DECISION_RECEIPT_SCHEMA_VERSION: &str =
@ -70,7 +70,9 @@ pub const REPO_FRONTIER_PLAN_DECISION_RECEIPT_SCHEMA_VERSION: &str =
pub const LEGACY_REPO_FRONTIER_PLAN_DECISION_RECEIPT_SCHEMA_VERSION: &str =
"epiphany.mind.repo_frontier_plan_decision_receipt.v0";
pub const REPO_FRONTIER_PLAN_DECISION_CONTRACT: &str = "epiphany.repo_frontier_plan_decision.v0";
pub const REPO_FRONTIER_PLANNING_CONTRACT: &str = "epiphany.repo_frontier_planning.v2";
pub const REPO_FRONTIER_PLANNING_CONTRACT: &str = "epiphany.repo_frontier_planning.v4";
pub const REPO_FRONTIER_PLANNING_FAILURE_REVIEW_SCHEMA_VERSION: &str =
"epiphany.self.repo_frontier_planning_failure_review.v0";
pub const REPO_FRONTIER_RESEARCH_REQUEST_SCHEMA_VERSION: &str =
"epiphany.self.repo_frontier_research_request.v3";
pub const REPO_FRONTIER_RESEARCH_REQUEST_CONTRACT: &str =
@ -101,9 +103,9 @@ pub const REPO_FRONTIER_PLANNING_LAUNCH_BINDING_SCHEMA_VERSION: &str =
pub const REPO_FRONTIER_PLANNING_LAUNCH_BINDING_CONTRACT: &str =
"epiphany.repo_frontier_planning_launch_binding.v0";
pub const REPO_FRONTIER_PLAN_MIND_REQUEST_SCHEMA_VERSION: &str =
"epiphany.self.repo_frontier_plan_mind_request.v0";
"epiphany.self.repo_frontier_plan_mind_request.v1";
pub const REPO_FRONTIER_PLAN_MIND_REQUEST_CONTRACT: &str =
"epiphany.repo_frontier_plan_mind_request.v0";
"epiphany.repo_frontier_plan_mind_request.v1";
pub const REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_SCHEMA_VERSION: &str =
"epiphany.coordinator.repo_frontier_plan_mind_launch_binding.v0";
pub const REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_CONTRACT: &str =
@ -433,8 +435,14 @@ pub struct RepoFrontierPlanningRequest {
pub contract: String,
#[cultcache(key = 10)]
pub runtime_id: String,
#[cultcache(key = 11)]
pub thread_id: String,
/// Exact current documents that own whether this planning work remains
/// actionable. The complete model sources above are immutable audit cargo.
#[cultcache(key = 12)]
pub frontier_authority_documents: Vec<crate::EpiphanyMindDocumentVersion>,
/// Per-claim guards close the otherwise invisible race between planning
/// eligibility and a newly admitted external-evidence challenge.
#[cultcache(key = 13)]
pub claim_obligation_documents: Vec<crate::EpiphanyMindDocumentVersion>,
}
#[derive(Debug, Clone, PartialEq, Eq, DatabaseEntry)]
@ -628,8 +636,6 @@ pub struct RepoFrontierPlanMindRequest {
pub candidate_sha256: String,
#[cultcache(key = 7)]
pub runtime_id: String,
#[cultcache(key = 8)]
pub thread_id: String,
#[cultcache(key = 9)]
pub requested_at: String,
#[cultcache(key = 10)]
@ -809,6 +815,30 @@ pub struct RepoFrontierModelingRequest {
pub contract: String,
}
#[derive(Debug, Clone, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "epiphany.self.repo_frontier_planning_failure_review",
schema = "RepoFrontierPlanningFailureReview"
)]
pub struct RepoFrontierPlanningFailureReview {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub review_id: String,
#[cultcache(key = 2)]
pub planning_request_id: String,
#[cultcache(key = 3)]
pub pass_kind: String,
#[cultcache(key = 4)]
pub job_id: String,
#[cultcache(key = 5)]
pub result_id: String,
#[cultcache(key = 6)]
pub disposition: String,
#[cultcache(key = 7)]
pub reviewed_at: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RepoFrontierNextOrgan {

688
epiphany-core/src/runtime_spine.rs
View file

@ -59,7 +59,8 @@ use crate::repo_model_gateway::{
REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_CONTRACT,
REPO_FRONTIER_PLAN_MIND_LAUNCH_BINDING_SCHEMA_VERSION,
REPO_FRONTIER_PLAN_MIND_REQUEST_CONTRACT, REPO_FRONTIER_PLAN_MIND_REQUEST_SCHEMA_VERSION,
REPO_FRONTIER_PLANNING_CONTRACT, REPO_FRONTIER_PLANNING_LAUNCH_BINDING_CONTRACT,
REPO_FRONTIER_PLANNING_CONTRACT, REPO_FRONTIER_PLANNING_FAILURE_REVIEW_SCHEMA_VERSION,
REPO_FRONTIER_PLANNING_LAUNCH_BINDING_CONTRACT,
REPO_FRONTIER_PLANNING_LAUNCH_BINDING_SCHEMA_VERSION,
REPO_FRONTIER_PLANNING_REQUEST_SCHEMA_VERSION,
REPO_FRONTIER_PROPOSAL_MODELING_LAUNCH_BINDING_CONTRACT,
@ -77,8 +78,8 @@ use crate::repo_model_gateway::{
RepoFrontierPlanDecisionReceipt, RepoFrontierPlanMindDecision,
RepoFrontierPlanMindLaunchBinding, RepoFrontierPlanMindRequest,
RepoFrontierPlanningCandidateEligibility, RepoFrontierPlanningEligibility,
RepoFrontierPlanningLaunchBinding, RepoFrontierPlanningLifecycle,
RepoFrontierPlanningLifecycleStage, RepoFrontierPlanningRequest,
RepoFrontierPlanningFailureReview, RepoFrontierPlanningLaunchBinding,
RepoFrontierPlanningLifecycle, RepoFrontierPlanningLifecycleStage, RepoFrontierPlanningRequest,
RepoFrontierProposalModelingLaunchBinding, RepoFrontierProposalModelingRequest,
RepoFrontierRelinquishmentReceipt, RepoFrontierResearchRequest, RepoFrontierRoute,
RepoFrontierVerdictDisposition, RepoFrontierWorkProposal, RepoModelClaimChallenge,
@ -1151,6 +1152,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef<Path>) -> Result<CultCache> {
cache.register_entry_type::<RepoFrontierPlanningRequest>()?;
cache.register_entry_type::<RepoFrontierResearchRequest>()?;
cache.register_entry_type::<RepoFrontierPlanningLaunchBinding>()?;
cache.register_entry_type::<RepoFrontierPlanningFailureReview>()?;
cache.register_entry_type::<crate::ImaginationConsiderationRequest>()?;
cache.register_entry_type::<crate::ImaginationConsiderationLaunchBinding>()?;
cache.register_entry_type::<crate::ImaginationConsiderationCandidate>()?;
@ -4836,7 +4838,7 @@ pub fn put_runtime_role_worker_result(
|| binding.job_id != result.job_id
|| binding.binding_id != EPIPHANY_IMAGINATION_ROLE_BINDING_ID
|| binding.runtime_id != request.runtime_id
|| binding.thread_id != request.thread_id
|| binding.thread_id != launch_document.thread_id()
|| binding.worker_launch_document_sha256 != launch_hash
|| worker_launch.job_id != result.job_id
|| worker_launch.role != EPIPHANY_IMAGINATION_OWNER_ROLE
@ -4847,7 +4849,7 @@ pub fn put_runtime_role_worker_result(
|| projection != Some(&expected_projection)
{
return Err(anyhow!(
"frontier planning result does not exactly bind request, launch, runtime, thread, and candidate"
"frontier planning result does not exactly bind request, launch, runtime, launch provenance, and candidate"
));
}
}
@ -5086,7 +5088,7 @@ pub fn put_runtime_role_worker_result(
|| binding.job_id != result.job_id
|| binding.binding_id != EPIPHANY_MIND_ROLE_BINDING_ID
|| binding.runtime_id != request.runtime_id
|| binding.thread_id != request.thread_id
|| binding.thread_id != document.thread_id()
|| binding.worker_launch_document_sha256 != hash
|| launch.role != EPIPHANY_MIND_OWNER_ROLE
|| launch.binding_id != EPIPHANY_MIND_ROLE_BINDING_ID
@ -5094,7 +5096,7 @@ pub fn put_runtime_role_worker_result(
|| projection != Some(&expected)
{
return Err(anyhow!(
"Mind result does not exactly bind request, launch, runtime, thread, and candidate"
"Mind result does not exactly bind request, launch, runtime, launch provenance, and candidate"
));
}
}
@ -6575,10 +6577,6 @@ pub fn select_and_commit_repo_frontier_planning_request(
let mut cache = runtime_spine_cache(runtime_store)?;
cache.pull_all_backing_stores()?;
let identity = require_identity(&cache)?;
let thread = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("planning request requires authoritative thread state"))?;
thread.state()?;
let backing = SingleFileMessagePackBackingStore::new(runtime_store);
let (view, basis) = current_keyed_repo_model(&cache)?;
let model = view.memory_context_projection();
@ -6586,6 +6584,8 @@ pub fn select_and_commit_repo_frontier_planning_request(
let item = actionable_imagination_frontier_item(&model, &challenges)
.ok_or_else(|| anyhow!("planning requires an actionable Imagination frontier"))?;
let item_hash = repo_frontier_item_hash(item)?;
let frontier_authority_documents = frontier_authority_documents(&cache, item)?;
let claim_obligation_documents = planning_claim_obligation_documents(&cache, item)?;
let request_id =
crate::frontier_planning_request_id(&identity.runtime_id, &item.id, &item_hash);
if cache
@ -6613,10 +6613,15 @@ pub fn select_and_commit_repo_frontier_planning_request(
requested_at: at.into(),
contract: REPO_FRONTIER_PLANNING_CONTRACT.into(),
runtime_id: identity.runtime_id,
thread_id: thread.thread_id,
frontier_authority_documents,
claim_obligation_documents,
};
let (envelope, _) = cache.prepare_entry(&request_id, &request)?;
let expected = keyed_repo_model_basis_envelopes(&cache, &basis)?;
let mut expected = frontier_authority_envelopes(&cache, &request.frontier_authority_documents)?;
expected.extend(planning_claim_obligation_envelopes(
&cache,
&request.claim_obligation_documents,
)?);
let mut writes = expected.clone();
writes.push(envelope);
if !backing.compare_and_swap_batch(&expected, writes)? {
@ -6631,57 +6636,142 @@ pub fn select_and_commit_repo_frontier_planning_request(
Err(anyhow!("planning request CAS collision"))
};
}
return Err(anyhow!("planning request lost exact keyed-model CAS"));
return Err(anyhow!("planning request lost exact frontier CAS"));
}
Ok(request)
}
fn validate_repo_frontier_planning_request(
request: &RepoFrontierPlanningRequest,
) -> Result<crate::RepoFrontierItem> {
chrono::DateTime::parse_from_rfc3339(&request.requested_at)
.map_err(|_| anyhow!("planning request timestamp must be RFC3339"))?;
let basis = crate::EpiphanyRepoModelBasis {
projection_digest: request.model_projection_digest.clone(),
source_documents: request.model_source_documents.clone(),
};
basis.validate()?;
if request.schema_version != REPO_FRONTIER_PLANNING_REQUEST_SCHEMA_VERSION
|| request.contract != REPO_FRONTIER_PLANNING_CONTRACT
|| request.runtime_id.trim().is_empty()
|| request.frontier_item_id.trim().is_empty()
|| request.frontier_item_hash.trim().is_empty()
|| request.selected_organ != "Imagination"
|| request.source_scope.is_empty()
|| !safe_sorted_unique_paths(&request.source_scope)
|| request.frontier_authority_documents.is_empty()
|| request.frontier_authority_documents.iter().any(|document| {
document.validate().is_err()
|| document.store_id != "epiphany-mind"
|| document.document_type != crate::EpiphanyRepoModelFrontierDocument::TYPE
|| !request.model_source_documents.contains(document)
})
|| !request.frontier_authority_documents.windows(2).all(|pair| {
(
pair[0].document_type.as_str(),
pair[0].document_key.as_str(),
) < (
pair[1].document_type.as_str(),
pair[1].document_key.as_str(),
)
})
|| request.claim_obligation_documents.iter().any(|document| {
document.validate().is_err()
|| document.store_id != "epiphany-mind"
|| document.document_type != crate::EpiphanyRepoModelClaimObligationsDocument::TYPE
|| !request.model_source_documents.contains(document)
})
|| !request.claim_obligation_documents.windows(2).all(|pair| {
(
pair[0].document_type.as_str(),
pair[0].document_key.as_str(),
) < (
pair[1].document_type.as_str(),
pair[1].document_key.as_str(),
)
})
{
return Err(anyhow!("invalid frontier planning request"));
}
let frontier_source = request
.frontier_authority_documents
.iter()
.find(|document| document.document_key == request.frontier_item_id)
.ok_or_else(|| anyhow!("planning request lost its owning frontier document"))?;
let item = rmp_serde::from_slice::<crate::EpiphanyRepoModelFrontierDocument>(
&frontier_source.payload_msgpack,
)?
.value()?;
let mut expected_authority_ids = vec![item.id.as_str()];
expected_authority_ids.extend(item.dependency_item_ids.iter().map(String::as_str));
expected_authority_ids.sort_unstable();
expected_authority_ids.dedup();
let expected_request_id = crate::frontier_planning_request_id(
&request.runtime_id,
&item.id,
&repo_frontier_item_hash(&item)?,
);
let mut expected_claim_ids = item
.target_claim_ids
.iter()
.map(String::as_str)
.collect::<Vec<_>>();
expected_claim_ids.sort_unstable();
expected_claim_ids.dedup();
if request
.frontier_authority_documents
.iter()
.map(|document| document.document_key.as_str())
.collect::<Vec<_>>()
!= expected_authority_ids
|| request
.claim_obligation_documents
.iter()
.map(|document| document.document_key.as_str())
.collect::<Vec<_>>()
!= expected_claim_ids
|| request.request_id != expected_request_id
|| request.frontier_item_id != item.id
|| request.frontier_item_hash != repo_frontier_item_hash(&item)?
|| request.source_scope != item.source_scope
{
return Err(anyhow!(
"frontier planning request diverges from its sealed frontier authority"
));
}
Ok(item)
}
pub(crate) fn validate_actionable_repo_frontier_planning_request(
cache: &CultCache,
request: &RepoFrontierPlanningRequest,
) -> Result<()> {
chrono::DateTime::parse_from_rfc3339(&request.requested_at)
.map_err(|_| anyhow!("planning request timestamp must be RFC3339"))?;
let sealed_item = validate_repo_frontier_planning_request(request)?;
let identity = require_identity(cache)?;
let thread = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("planning request requires authoritative thread state"))?;
thread.state()?;
let view = require_keyed_repo_model_basis(
cache,
&request.model_projection_digest,
&request.model_source_documents,
)?;
if request.runtime_id != identity.runtime_id {
return Err(anyhow!("planning request belongs to another runtime"));
}
frontier_authority_envelopes(cache, &request.frontier_authority_documents)?;
planning_claim_obligation_envelopes(cache, &request.claim_obligation_documents)?;
let current_item = cache
.get::<crate::EpiphanyRepoModelFrontierDocument>(&request.frontier_item_id)?
.ok_or_else(|| anyhow!("planning request frontier disappeared"))?
.value()?;
if current_item != sealed_item
|| frontier_authority_documents(cache, &current_item)?
!= request.frontier_authority_documents
|| planning_claim_obligation_documents(cache, &current_item)?
!= request.claim_obligation_documents
{
return Err(anyhow!("planning request frontier authority changed"));
}
let view = crate::repo_model_documents::assemble_repo_model_view_from_cache(cache)?;
let model = view.memory_context_projection();
let basis = view.reasoning_basis();
let challenges = current_repo_model_claim_challenges(cache, &model, &basis)?;
let item = model
.frontier
.iter()
.find(|item| item.id == request.frontier_item_id)
.ok_or_else(|| anyhow!("planning request frontier disappeared"))?;
if !imagination_frontier_item_is_actionable(&model, &challenges, item) {
if !imagination_frontier_item_is_actionable(&model, &challenges, &current_item) {
return Err(anyhow!("planning request frontier is no longer actionable"));
}
let item_hash = repo_frontier_item_hash(item)?;
let expected_request_id =
crate::frontier_planning_request_id(&identity.runtime_id, &item.id, &item_hash);
if request.schema_version != REPO_FRONTIER_PLANNING_REQUEST_SCHEMA_VERSION
|| request.contract != REPO_FRONTIER_PLANNING_CONTRACT
|| request.request_id != expected_request_id
|| request.model_projection_digest != basis.projection_digest
|| request.model_source_documents != basis.source_documents
|| request.frontier_item_id != item.id
|| request.frontier_item_hash != item_hash
|| request.selected_organ != "Imagination"
|| request.source_scope != item.source_scope
|| request.runtime_id != identity.runtime_id
|| request.thread_id.is_empty()
{
return Err(anyhow!(
"planning request does not exactly bind its actionable frontier and runtime"
));
}
Ok(())
}
@ -6729,13 +6819,12 @@ pub fn commit_repo_frontier_plan_mind_request(
let request = RepoFrontierPlanMindRequest {
schema_version: REPO_FRONTIER_PLAN_MIND_REQUEST_SCHEMA_VERSION.into(),
request_id: request_id.clone(),
planning_request_id: planning.request_id,
planning_request_id: planning.request_id.clone(),
imagination_result_id: result.result_id.clone(),
imagination_job_id: result.job_id.clone(),
candidate_id: candidate.candidate_id,
candidate_sha256,
runtime_id: planning.runtime_id,
thread_id: planning.thread_id,
runtime_id: planning.runtime_id.clone(),
requested_at: requested_at.into(),
contract: REPO_FRONTIER_PLAN_MIND_REQUEST_CONTRACT.into(),
};
@ -6750,11 +6839,22 @@ pub fn commit_repo_frontier_plan_mind_request(
}
let (envelope, _) = cache.prepare_entry(&request_id, &request)?;
let backing = SingleFileMessagePackBackingStore::new(runtime_store);
let basis = crate::EpiphanyRepoModelBasis {
projection_digest: planning.model_projection_digest.clone(),
source_documents: planning.model_source_documents.clone(),
};
let expected = keyed_repo_model_basis_envelopes(&cache, &basis)?;
let mut expected =
frontier_authority_envelopes(&cache, &planning.frontier_authority_documents)?;
expected.extend(planning_claim_obligation_envelopes(
&cache,
&planning.claim_obligation_documents,
)?);
expected.push(
cache
.get_envelope::<RepoFrontierPlanningRequest>(&planning.request_id)?
.ok_or_else(|| anyhow!("Mind request lost its Planning request envelope"))?,
);
expected.push(
cache
.get_envelope::<EpiphanyRuntimeRoleWorkerResult>(&result.job_id)?
.ok_or_else(|| anyhow!("Mind request lost its Imagination result envelope"))?,
);
let mut writes = expected.clone();
writes.push(envelope);
if !backing.compare_and_swap_batch(&expected, writes)? {
@ -6803,7 +6903,6 @@ fn validate_repo_frontier_plan_mind_request_identity(
|| candidate.candidate_id != request.candidate_id
|| hash != request.candidate_sha256
|| request.runtime_id != planning.runtime_id
|| request.thread_id != planning.thread_id
{
return Err(anyhow!(
"Mind request substituted immutable Imagination causal identity"
@ -6888,7 +6987,7 @@ fn current_repo_model_claim_challenges(
fn validate_repo_model_claim_challenge_chain(
cache: &CultCache,
model: &crate::EpiphanyMemoryGraphSnapshot,
basis: &crate::EpiphanyRepoModelBasis,
_current_basis: &crate::EpiphanyRepoModelBasis,
challenge: &RepoModelClaimChallenge,
require_current_model: bool,
) -> Result<()> {
@ -6918,14 +7017,11 @@ fn validate_repo_model_claim_challenge_chain(
{
return Err(anyhow!("claim challenge substituted Eyes evidence"));
}
if challenge.model_projection_digest != basis.projection_digest
|| challenge.model_source_documents != basis.source_documents
{
return Err(anyhow!("claim challenge keyed model basis mismatch"));
}
if require_current_model {
basis.validate_against_cache(cache)?;
}
let audit_basis = crate::EpiphanyRepoModelBasis {
projection_digest: challenge.model_projection_digest.clone(),
source_documents: challenge.model_source_documents.clone(),
};
audit_basis.validate()?;
let claim = model
.nodes
.iter()
@ -6936,6 +7032,24 @@ fn validate_repo_model_claim_challenge_chain(
{
return Err(anyhow!("claim challenge target claim identity mismatch"));
}
let claim_envelope = cache
.get_envelope::<crate::EpiphanyRepoModelNodeDocument>(&claim.id)?
.ok_or_else(|| anyhow!("claim challenge target claim envelope is missing"))?;
let claim_version =
crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &claim_envelope)?;
if !audit_basis.source_documents.contains(&claim_version) {
return Err(anyhow!(
"claim challenge audit basis omitted its exact target claim"
));
}
if require_current_model {
let obligation = cache
.get::<crate::EpiphanyRepoModelClaimObligationsDocument>(&claim.id)?
.ok_or_else(|| anyhow!("claim challenge target lost its obligation owner"))?;
if obligation.node_id != claim.id {
return Err(anyhow!("claim challenge obligation owner is corrupt"));
}
}
Ok(())
}
@ -6963,10 +7077,19 @@ pub fn commit_repo_model_claim_challenge(
let model = view.memory_context_projection();
validate_repo_model_claim_challenge_chain(&cache, &model, &basis, challenge, true)?;
if let Some(existing) = cache.get::<RepoModelClaimChallenge>(&challenge.challenge_id)? {
return if existing == *challenge {
if existing != *challenge {
return Err(anyhow!("claim challenge ids are immutable"));
}
let obligation = cache
.get::<crate::EpiphanyRepoModelClaimObligationsDocument>(&challenge.target_claim_id)?
.ok_or_else(|| anyhow!("replayed claim challenge lost its obligation"))?;
return if obligation
.active_challenge_ids
.contains(&challenge.challenge_id)
{
Ok(())
} else {
Err(anyhow!("claim challenge ids are immutable"))
Err(anyhow!("replayed claim challenge lost active authority"))
};
}
let packet_envelope = envelopes
@ -6978,17 +7101,55 @@ pub fn commit_repo_model_claim_challenge(
.cloned()
.ok_or_else(|| anyhow!("claim challenge packet envelope is missing"))?;
let (challenge_envelope, _) = cache.prepare_entry(&challenge.challenge_id, challenge)?;
let mut expected = keyed_repo_model_basis_envelopes(&cache, &basis)?;
expected.push(packet_envelope);
let mut writes = expected.clone();
let claim_envelope = cache
.get_envelope::<crate::EpiphanyRepoModelNodeDocument>(&challenge.target_claim_id)?
.ok_or_else(|| anyhow!("claim challenge target envelope is missing"))?;
let mut obligation = cache
.get::<crate::EpiphanyRepoModelClaimObligationsDocument>(&challenge.target_claim_id)?
.ok_or_else(|| anyhow!("claim challenge target obligation is missing"))?;
let obligation_envelope = cache
.get_envelope::<crate::EpiphanyRepoModelClaimObligationsDocument>(
&challenge.target_claim_id,
)?
.ok_or_else(|| anyhow!("claim challenge target obligation envelope is missing"))?;
obligation
.active_challenge_ids
.push(challenge.challenge_id.clone());
obligation.active_challenge_ids.sort();
obligation.active_challenge_ids.dedup();
let expected = vec![claim_envelope, obligation_envelope.clone(), packet_envelope];
let mut writes = expected
.iter()
.filter(|envelope| {
envelope.r#type != obligation_envelope.r#type || envelope.key != obligation_envelope.key
})
.cloned()
.collect::<Vec<_>>();
writes.push(challenge_envelope);
writes.push(
cache
.prepare_entry(&challenge.target_claim_id, &obligation)?
.0,
);
if !backing.compare_and_swap_batch(&expected, writes)? {
let mut reloaded = runtime_spine_cache(store_path)?;
reloaded.pull_all_backing_stores()?;
return match reloaded.get::<RepoModelClaimChallenge>(&challenge.challenge_id)? {
Some(existing) if existing == *challenge => Ok(()),
Some(_) => Err(anyhow!("claim challenge immutable collision")),
None => Err(anyhow!("claim challenge lost exact model/packet CAS")),
return match (
reloaded.get::<RepoModelClaimChallenge>(&challenge.challenge_id)?,
reloaded.get::<crate::EpiphanyRepoModelClaimObligationsDocument>(
&challenge.target_claim_id,
)?,
) {
(Some(existing), Some(obligation))
if existing == *challenge
&& obligation
.active_challenge_ids
.contains(&challenge.challenge_id) =>
{
Ok(())
}
(Some(_), _) => Err(anyhow!("claim challenge immutable collision")),
_ => Err(anyhow!("claim challenge lost exact claim/packet CAS")),
};
}
Ok(())
@ -7016,18 +7177,9 @@ fn validate_repo_frontier_plan_candidate_against_request(
if candidate.candidate_id != expected_candidate_id {
return Err(anyhow!("frontier planning candidate id is not canonical"));
}
let model = require_keyed_repo_model_basis(
cache,
&request.model_projection_digest,
&request.model_source_documents,
)?
.memory_context_projection();
let item = model
.frontier
.iter()
.find(|item| item.id == request.frontier_item_id)
.ok_or_else(|| anyhow!("frontier planning candidate frontier disappeared"))?;
if format!("{:x}", Sha256::digest(rmp_serde::to_vec_named(item)?)) != request.frontier_item_hash
let item = validate_repo_frontier_planning_request(request)?;
frontier_authority_envelopes(cache, &request.frontier_authority_documents)?;
if repo_frontier_item_hash(&item)? != request.frontier_item_hash
|| item.source_scope != request.source_scope
|| !candidate.safe_paths.iter().all(|path| {
request.source_scope.iter().any(|scope| {
@ -7281,7 +7433,8 @@ fn commit_repo_frontier_plan_decision_inner(
let mind_request = cache
.get::<RepoFrontierPlanMindRequest>(&mind_request_id)?
.ok_or_else(|| anyhow!("frontier plan decision requires its typed Mind request"))?;
let (planning, candidate) = validate_repo_frontier_plan_mind_request(&cache, &mind_request)?;
let (planning, candidate) =
validate_repo_frontier_plan_mind_request_identity(&cache, &mind_request)?;
if let FrontierPlanDecisionSource::Operator(review) = source {
if review.candidate_id != candidate.candidate_id
|| review.candidate_sha256 != mind_request.candidate_sha256
@ -7314,42 +7467,64 @@ fn commit_repo_frontier_plan_decision_inner(
rationale,
decided_at: decided_at.clone(),
contract: REPO_FRONTIER_PLAN_DECISION_CONTRACT.into(),
decision_source: Some(decision_source),
decision_source: Some(decision_source.clone()),
};
if let Some(existing) = cache.get::<RepoFrontierPlanDecisionReceipt>(&decision_id)? {
return if existing == receipt {
Ok(existing)
} else {
Err(anyhow!("frontier plan decision identity collision"))
};
if existing != receipt {
return Err(anyhow!("frontier plan decision identity collision"));
}
if existing.decision == RepoFrontierPlanDecision::Adopt {
let item = cache
.get::<crate::EpiphanyRepoModelFrontierDocument>(&planning.frontier_item_id)?
.ok_or_else(|| anyhow!("replayed plan decision lost its frontier"))?
.value()?;
let adopted = item
.adopted_plan
.as_ref()
.ok_or_else(|| anyhow!("replayed plan decision lost its adoption"))?;
if adopted.planning_request_id != planning.request_id
|| adopted.result_id != mind_request.imagination_result_id
|| adopted.candidate_id != candidate.candidate_id
|| adopted.candidate_sha256 != mind_request.candidate_sha256
{
return Err(anyhow!("replayed plan decision adoption was substituted"));
}
}
return Ok(existing);
}
let basis = crate::EpiphanyRepoModelBasis {
projection_digest: planning.model_projection_digest.clone(),
source_documents: planning.model_source_documents.clone(),
};
let mut strong_reads = keyed_repo_model_basis_envelopes(&cache, &basis)?;
validate_actionable_repo_frontier_planning_request(&cache, &planning)?;
validate_repo_frontier_plan_candidate_against_request(&cache, &candidate, &planning)?;
let mut strong_reads =
frontier_authority_envelopes(&cache, &planning.frontier_authority_documents)?;
strong_reads.extend(planning_claim_obligation_envelopes(
&cache,
&planning.claim_obligation_documents,
)?);
for envelope in [
cache.get_envelope::<RepoFrontierPlanningRequest>(&planning.request_id)?,
cache.get_envelope::<RepoFrontierPlanMindRequest>(&mind_request.request_id)?,
cache.get_envelope::<EpiphanyRuntimeRoleWorkerResult>(&mind_request.imagination_job_id)?,
]
.into_iter()
.flatten()
{
strong_reads.push(envelope);
}
if let crate::RepoFrontierPlanDecisionSource::MindWorker { job_id, .. } = &decision_source {
strong_reads.push(
cache
.get_envelope::<EpiphanyRuntimeRoleWorkerResult>(job_id)?
.ok_or_else(|| anyhow!("frontier plan decision lost its Mind result envelope"))?,
);
}
let mut writes = Vec::new();
if decision == RepoFrontierPlanDecision::Adopt {
let view = require_keyed_repo_model_basis(
&cache,
&planning.model_projection_digest,
&planning.model_source_documents,
)?;
let mut item = view
.frontier
.into_iter()
.find(|item| item.id == planning.frontier_item_id)
let item = cache
.get::<crate::EpiphanyRepoModelFrontierDocument>(&planning.frontier_item_id)?
.ok_or_else(|| anyhow!("frontier plan decision target disappeared"))?;
let mut item = item.value()?;
if repo_frontier_item_hash(&item)? != planning.frontier_item_hash {
return Err(anyhow!("frontier plan decision target changed"));
}
@ -7359,29 +7534,23 @@ fn commit_repo_frontier_plan_decision_inner(
job_id: mind_request.imagination_job_id.clone(),
candidate_id: candidate.candidate_id.clone(),
candidate_sha256: mind_request.candidate_sha256.clone(),
safe_paths: candidate.safe_paths,
action: candidate.action,
command: candidate.command,
checks: candidate.checks,
stop_conditions: candidate.stop_conditions,
rollback_steps: candidate.rollback_steps,
commit_message: candidate.commit_message,
safe_paths: candidate.safe_paths.clone(),
action: candidate.action.clone(),
command: candidate.command.clone(),
checks: candidate.checks.clone(),
stop_conditions: candidate.stop_conditions.clone(),
rollback_steps: candidate.rollback_steps.clone(),
commit_message: candidate.commit_message.clone(),
execution_amendment: None,
});
let proposal = crate::EpiphanyRepoModelMutationProposal::new(
format!("repo-frontier-plan-adoption-{decision_id}"),
mind_request.request_id.clone(),
decision_id.clone(),
vec![
candidate.candidate_id.clone(),
mind_request.imagination_result_id.clone(),
],
crate::load_current_runtime_repository_body_basis(runtime_store)?,
vec![crate::EpiphanyRepoModelMutationOperation::PutFrontier { item }],
)?;
let plan = crate::plan_repo_model_mutation(runtime_store, &proposal)?;
strong_reads.extend(plan.strong_reads);
writes.extend(plan.writes);
writes.push(
cache
.prepare_entry(
&planning.frontier_item_id,
&crate::EpiphanyRepoModelFrontierDocument::new(&item)?,
)?
.0,
);
}
writes.push(cache.prepare_entry(&receipt.decision_id, &receipt)?.0);
strong_reads.sort_by(|left, right| {
@ -7583,7 +7752,7 @@ pub fn runtime_has_actionable_eyes_frontier(runtime_store: impl AsRef<Path>) ->
runtime_has_actionable_frontier_for_organ(runtime_store, "Eyes")
}
fn research_frontier_authority_documents(
fn frontier_authority_documents(
cache: &CultCache,
item: &crate::RepoFrontierItem,
) -> Result<Vec<crate::EpiphanyMindDocumentVersion>> {
@ -7596,7 +7765,7 @@ fn research_frontier_authority_documents(
.map(|id| {
let envelope = cache
.get_envelope::<crate::EpiphanyRepoModelFrontierDocument>(id)?
.ok_or_else(|| anyhow!("Research frontier authority document {id:?} is absent"))?;
.ok_or_else(|| anyhow!("frontier authority document {id:?} is absent"))?;
crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &envelope)
})
.collect::<Result<Vec<_>>>()?;
@ -7608,7 +7777,7 @@ fn research_frontier_authority_documents(
Ok(documents)
}
fn research_authority_envelopes(
fn frontier_authority_envelopes(
cache: &CultCache,
documents: &[crate::EpiphanyMindDocumentVersion],
) -> Result<Vec<CultCacheEnvelope>> {
@ -7620,7 +7789,7 @@ fn research_authority_envelopes(
|| document.document_type != crate::EpiphanyRepoModelFrontierDocument::TYPE
{
return Err(anyhow!(
"Research authority contains a non-frontier Mind document"
"frontier authority contains a non-frontier Mind document"
));
}
let envelope = snapshot
@ -7629,11 +7798,72 @@ fn research_authority_envelopes(
envelope.r#type == document.document_type
&& envelope.key == document.document_key
})
.ok_or_else(|| anyhow!("Research frontier authority document is absent"))?;
.ok_or_else(|| anyhow!("frontier authority document is absent"))?;
if crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", envelope)?
!= *document
{
return Err(anyhow!("Research frontier authority document changed"));
return Err(anyhow!("frontier authority document changed"));
}
Ok(envelope.clone())
})
.collect()
}
fn planning_claim_obligation_documents(
cache: &CultCache,
item: &crate::RepoFrontierItem,
) -> Result<Vec<crate::EpiphanyMindDocumentVersion>> {
let mut claim_ids = item.target_claim_ids.clone();
claim_ids.sort();
claim_ids.dedup();
let mut documents = claim_ids
.into_iter()
.map(|claim_id| {
let envelope = cache
.get_envelope::<crate::EpiphanyRepoModelClaimObligationsDocument>(&claim_id)?
.ok_or_else(|| anyhow!("planning claim obligation {claim_id:?} is absent"))?;
crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &envelope)
})
.collect::<Result<Vec<_>>>()?;
documents.sort_by(|left, right| {
left.document_type
.cmp(&right.document_type)
.then(left.document_key.cmp(&right.document_key))
});
Ok(documents)
}
fn planning_claim_obligation_envelopes(
cache: &CultCache,
documents: &[crate::EpiphanyMindDocumentVersion],
) -> Result<Vec<CultCacheEnvelope>> {
let snapshot = cache.snapshot_envelopes();
documents
.iter()
.map(|document| {
if document.store_id != "epiphany-mind"
|| document.document_type != crate::EpiphanyRepoModelClaimObligationsDocument::TYPE
{
return Err(anyhow!(
"planning claim authority contains a non-obligation Mind document"
));
}
let envelope = snapshot
.iter()
.find(|envelope| {
envelope.r#type == document.document_type
&& envelope.key == document.document_key
})
.ok_or_else(|| anyhow!("planning claim obligation is absent"))?;
if crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", envelope)?
!= *document
{
return Err(anyhow!("planning claim obligation changed"));
}
let obligation: crate::EpiphanyRepoModelClaimObligationsDocument =
rmp_serde::from_slice(&envelope.payload)?;
if !obligation.active_challenge_ids.is_empty() {
return Err(anyhow!("planning claim is challenged"));
}
Ok(envelope.clone())
})
@ -7728,14 +7958,14 @@ fn current_repo_frontier_research_request(
if request.runtime_id != identity.runtime_id {
return Ok(None);
}
if research_authority_envelopes(cache, &request.frontier_authority_documents).is_err() {
if frontier_authority_envelopes(cache, &request.frontier_authority_documents).is_err() {
return Ok(None);
}
let frontier = cache
.get::<crate::EpiphanyRepoModelFrontierDocument>(&request.frontier_item_id)?
.ok_or_else(|| anyhow!("Research request frontier document is absent"))?
.value()?;
let expected_authority = research_frontier_authority_documents(cache, &frontier)?;
let expected_authority = frontier_authority_documents(cache, &frontier)?;
if expected_authority != request.frontier_authority_documents {
return Ok(None);
}
@ -7805,7 +8035,7 @@ pub fn select_and_commit_repo_frontier_research_request(
};
}
let (request_envelope, _) = cache.prepare_entry(&request_id, &request)?;
let expected = research_authority_envelopes(&cache, &request.frontier_authority_documents)?;
let expected = frontier_authority_envelopes(&cache, &request.frontier_authority_documents)?;
let mut writes = expected.clone();
writes.push(request_envelope);
if !backing.compare_and_swap_batch(&expected, writes)? {
@ -7844,7 +8074,7 @@ fn repo_frontier_research_request_for_admitted_item(
));
}
let item_hash = repo_frontier_item_hash(item)?;
let frontier_authority_documents = research_frontier_authority_documents(cache, item)?;
let frontier_authority_documents = frontier_authority_documents(cache, item)?;
let public_source_refs = crate::ImmutableGithubSource::canonicalize_set(
item.public_source_refs.iter().map(String::as_str),
)?;
@ -8274,6 +8504,126 @@ pub fn runtime_repo_frontier_planning_eligibility(
})
}
fn repo_frontier_planning_failure_review_id(result_id: &str) -> String {
format!("repo-frontier-planning-failure-review-{result_id}")
}
pub(crate) fn repo_frontier_planning_failure_review(
cache: &CultCache,
planning_request_id: &str,
pass_kind: &str,
result: &EpiphanyRuntimeRoleWorkerResult,
) -> Result<Option<RepoFrontierPlanningFailureReview>> {
let review_id = repo_frontier_planning_failure_review_id(&result.result_id);
let Some(review) = cache.get::<RepoFrontierPlanningFailureReview>(&review_id)? else {
return Ok(None);
};
if review.schema_version != REPO_FRONTIER_PLANNING_FAILURE_REVIEW_SCHEMA_VERSION
|| review.review_id != review_id
|| review.planning_request_id != planning_request_id
|| review.pass_kind != pass_kind
|| review.job_id != result.job_id
|| review.result_id != result.result_id
|| review.disposition != "superseded"
|| chrono::DateTime::parse_from_rfc3339(&review.reviewed_at).is_err()
{
return Err(anyhow!("frontier planning failure review is corrupt"));
}
Ok(Some(review))
}
pub fn review_repo_frontier_planning_failure(
runtime_store: impl AsRef<Path>,
job_id: &str,
reviewed_at: &str,
) -> Result<RepoFrontierPlanningFailureReview> {
chrono::DateTime::parse_from_rfc3339(reviewed_at)
.map_err(|_| anyhow!("planning failure review time must be RFC3339"))?;
let runtime_store = runtime_store.as_ref();
let lifecycle = runtime_repo_frontier_planning_lifecycle(runtime_store)?;
let (pass_kind, expected_job_id) = match lifecycle.stage {
RepoFrontierPlanningLifecycleStage::ImaginationFailed => {
("imagination", lifecycle.imagination_job_id.as_deref())
}
RepoFrontierPlanningLifecycleStage::MindFailed => {
("mind", lifecycle.mind_job_id.as_deref())
}
_ => return Err(anyhow!("frontier planning has no reviewable failed pass")),
};
if expected_job_id != Some(job_id) {
return Err(anyhow!("planning failure review job is not current"));
}
let planning_request_id = lifecycle
.planning_request_id
.ok_or_else(|| anyhow!("planning failure review lost its request"))?;
let mut cache = runtime_spine_cache(runtime_store)?;
cache.pull_all_backing_stores()?;
let result = cache
.get::<EpiphanyRuntimeRoleWorkerResult>(job_id)?
.ok_or_else(|| anyhow!("planning failure review lost its typed result"))?;
let review = RepoFrontierPlanningFailureReview {
schema_version: REPO_FRONTIER_PLANNING_FAILURE_REVIEW_SCHEMA_VERSION.into(),
review_id: repo_frontier_planning_failure_review_id(&result.result_id),
planning_request_id: planning_request_id.clone(),
pass_kind: pass_kind.into(),
job_id: job_id.into(),
result_id: result.result_id.clone(),
disposition: "superseded".into(),
reviewed_at: reviewed_at.into(),
};
if let Some(existing) =
repo_frontier_planning_failure_review(&cache, &planning_request_id, pass_kind, &result)?
{
return if existing == review {
Ok(existing)
} else {
Err(anyhow!("planning failure review identity collision"))
};
}
let snapshot = cache.snapshot_envelopes();
let mut expected = Vec::new();
for (document_type, document_key) in [
(
RepoFrontierPlanningRequest::TYPE,
planning_request_id.as_str(),
),
(EpiphanyRuntimeWorkerLaunchRequest::TYPE, job_id),
(EpiphanyRuntimeJob::TYPE, job_id),
(EpiphanyRuntimeRoleWorkerResult::TYPE, job_id),
] {
expected.push(
snapshot
.iter()
.find(|envelope| envelope.r#type == document_type && envelope.key == document_key)
.cloned()
.ok_or_else(|| anyhow!("planning failure review lost a strong source"))?,
);
}
if pass_kind == "mind" {
let mind_request_id = lifecycle
.mind_request_id
.ok_or_else(|| anyhow!("Mind failure review lost its request"))?;
expected.push(
cache
.get_envelope::<RepoFrontierPlanMindRequest>(&mind_request_id)?
.ok_or_else(|| anyhow!("Mind failure review request envelope is missing"))?,
);
}
let envelope = cache.prepare_entry(&review.review_id, &review)?.0;
let mut writes = expected.clone();
writes.push(envelope);
let backing = SingleFileMessagePackBackingStore::new(runtime_store);
if backing.compare_and_swap_batch(&expected, writes)? {
return Ok(review);
}
let mut reloaded = runtime_spine_cache(runtime_store)?;
reloaded.pull_all_backing_stores()?;
match reloaded.get::<RepoFrontierPlanningFailureReview>(&review.review_id)? {
Some(existing) if existing == review => Ok(existing),
_ => Err(anyhow!("planning failure review lost exact-envelope CAS")),
}
}
/// Read-only Self projection over the existing typed frontier-planning chain.
/// It never creates a request or decision; each mutating stage remains owned by
/// its established commit primitive.
@ -8410,29 +8760,16 @@ pub fn runtime_repo_frontier_planning_lifecycle(
"Self found malformed typed frontier planning failure"
));
}
let reviewed = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("planning failure review requires thread state"))?
.state()?
.acceptance_receipts
.into_iter()
.filter(|receipt| {
receipt.result_id == imagination_result.result_id
&& receipt.job_id == imagination_result.job_id
&& receipt.binding_id == EPIPHANY_IMAGINATION_ROLE_BINDING_ID
&& receipt.surface == "roleFailureReview"
&& receipt.role_id == "imagination"
&& receipt.status == "superseded"
})
.count();
lifecycle.stage = if reviewed == 1 {
let reviewed = repo_frontier_planning_failure_review(
&cache,
&request.request_id,
"imagination",
imagination_result,
)?;
lifecycle.stage = if reviewed.is_some() {
RepoFrontierPlanningLifecycleStage::ImaginationLaunchReady
} else if reviewed == 0 {
RepoFrontierPlanningLifecycleStage::ImaginationFailed
} else {
return Err(anyhow!(
"Self found conflicting frontier planning failure reviews"
));
RepoFrontierPlanningLifecycleStage::ImaginationFailed
};
return Ok(lifecycle);
}
@ -8496,27 +8833,16 @@ pub fn runtime_repo_frontier_planning_lifecycle(
{
return Err(anyhow!("Self found malformed typed Mind failure"));
}
let reviewed = cache
.get::<crate::EpiphanyThreadStateEntry>(crate::THREAD_STATE_KEY)?
.ok_or_else(|| anyhow!("Mind failure review requires thread state"))?
.state()?
.acceptance_receipts
.into_iter()
.filter(|receipt| {
receipt.result_id == mind_result.result_id
&& receipt.job_id == mind_result.job_id
&& receipt.binding_id == EPIPHANY_MIND_ROLE_BINDING_ID
&& receipt.surface == "roleFailureReview"
&& receipt.role_id == "mindAdmissionReview"
&& receipt.status == "superseded"
})
.count();
lifecycle.stage = if reviewed == 1 {
let reviewed = repo_frontier_planning_failure_review(
&cache,
&request.request_id,
"mind",
mind_result,
)?;
lifecycle.stage = if reviewed.is_some() {
RepoFrontierPlanningLifecycleStage::MindLaunchReady
} else if reviewed == 0 {
RepoFrontierPlanningLifecycleStage::MindFailed
} else {
return Err(anyhow!("Self found conflicting Mind failure reviews"));
RepoFrontierPlanningLifecycleStage::MindFailed
};
return Ok(lifecycle);
}

10
epiphany-core/src/surfaces/worker_launch.rs
View file

@ -31,9 +31,9 @@ pub const REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_SCHEMA_VERSION: &str =
pub const REPO_FRONTIER_PROPOSAL_MODELING_CONTEXT_CONTRACT: &str =
"epiphany.repo_frontier_proposal_modeling_context.v1";
pub const REPO_FRONTIER_PLANNING_CONTEXT_SCHEMA_VERSION: &str =
"epiphany.worker.repo_frontier_planning_context.v1";
"epiphany.worker.repo_frontier_planning_context.v2";
pub const REPO_FRONTIER_PLANNING_CONTEXT_CONTRACT: &str =
"epiphany.repo_frontier_planning_context.v1";
"epiphany.repo_frontier_planning_context.v2";
pub const REPO_FRONTIER_RESEARCH_CONTEXT_SCHEMA_VERSION: &str =
"epiphany.worker.repo_frontier_research_context.v3";
pub const REPO_FRONTIER_RESEARCH_CONTEXT_CONTRACT: &str =
@ -292,7 +292,8 @@ pub struct RepoFrontierPlanningContextProjection {
pub source_scope: Vec<String>,
pub requested_at: String,
pub runtime_id: String,
pub thread_id: String,
pub frontier_authority_documents: Vec<crate::EpiphanyMindDocumentVersion>,
pub claim_obligation_documents: Vec<crate::EpiphanyMindDocumentVersion>,
}
impl RepoFrontierPlanningContextProjection {
@ -309,7 +310,8 @@ impl RepoFrontierPlanningContextProjection {
source_scope: request.source_scope.clone(),
requested_at: request.requested_at.clone(),
runtime_id: request.runtime_id.clone(),
thread_id: request.thread_id.clone(),
frontier_authority_documents: request.frontier_authority_documents.clone(),
claim_obligation_documents: request.claim_obligation_documents.clone(),
}
}
}

7
epiphany-openai-runtime/src/lib.rs
View file

@ -3192,13 +3192,14 @@ mod tests {
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(),
source_scope: vec!["src".into(), "tests".into()],
requested_at: "2026-07-15T09:59:00Z".into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
};
let result = role_worker_result_from_ingress(
&launch,
@ -3521,7 +3522,6 @@ mod tests {
candidate_id: "candidate-1".into(),
candidate_sha256: "candidate-hash-1".into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
requested_at: "2026-07-15T09:59:00Z".into(),
contract: epiphany_core::REPO_FRONTIER_PLAN_MIND_REQUEST_CONTRACT.into(),
},
@ -3530,6 +3530,8 @@ mod tests {
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(),
@ -3537,7 +3539,6 @@ mod tests {
requested_at: "2026-07-15T09:58:00Z".into(),
contract: epiphany_core::REPO_FRONTIER_PLANNING_CONTRACT.into(),
runtime_id: "runtime-1".into(),
thread_id: "thread-1".into(),
},
candidate: epiphany_core::RepoFrontierPlanCandidate {
schema_version: epiphany_core::REPO_FRONTIER_PLAN_CANDIDATE_SCHEMA_VERSION.into(),