Own exact worker process continuity

This commit is contained in:
Meta Vens 2026-08-10 13:33:48 +01:00
commit 75b841808b
11 changed files with 1493 additions and 92 deletions

View file

@ -31,11 +31,14 @@ use epiphany_core::{
};
use serde_json::Value;
use serde_json::json;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::thread;
use std::time::Duration;
use std::time::Instant;
use uuid::Uuid;
@ -2030,6 +2033,8 @@ fn launch_worker_runtime_detached(
max_runtime_seconds: u64,
auto_tools: bool,
) -> Result<Value> {
let activation_token = Uuid::new_v4().to_string();
let activation_token_sha256 = format!("{:x}", Sha256::digest(activation_token.as_bytes()));
let stdout_path = artifact_dir.join(format!(
"step-{step_index:02}-{}-worker-runtime.stdout.json",
sanitize_id(role_id)
@ -2051,6 +2056,8 @@ fn launch_worker_runtime_detached(
.arg(mcp_config)
.arg("--job-id")
.arg(job_id)
.arg("--activation-token-sha256")
.arg(&activation_token_sha256)
.arg("--max-runtime-seconds")
.arg(max_runtime_seconds.to_string());
if auto_tools {
@ -2073,13 +2080,47 @@ fn launch_worker_runtime_detached(
.stderr(Stdio::from(stderr_file));
#[cfg(windows)]
command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW);
let child = command
let mut child = command
.spawn()
.with_context(|| format!("failed to spawn {}", model_runtime_bin.display()))?;
let process = capture_process_instance(child.id())?;
let claim_deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
if let Some(claim) = epiphany_core::runtime_worker_process_claim(runtime_store, job_id)? {
if claim.process_id != process.process_id
|| claim.process_creation_token != process.creation_token
|| claim.process_executable_path != process.executable_path.display().to_string()
|| claim.activation_token_sha256 != activation_token_sha256
{
return Err(anyhow!(
"worker process claim does not bind the spawned incarnation"
));
}
epiphany_core::activate_runtime_worker_process(
runtime_store,
job_id,
&process,
&activation_token,
&chrono::Utc::now().to_rfc3339(),
)?;
break;
}
if let Some(status) = child.try_wait()? {
return Err(anyhow!(
"worker exited with {status} before claiming runtime authority"
));
}
if std::time::Instant::now() >= claim_deadline {
return Err(anyhow!(
"worker did not claim runtime authority before activation deadline"
));
}
thread::sleep(Duration::from_millis(25));
}
Ok(json!({
"status": "launched",
"jobId": job_id,
"pid": child.id(),
"pid": process.process_id,
"stdout": stdout_path,
"stderr": stderr_path,
"note": "Worker runtime is detached; coordinator observes completion through runtime-spine role/reorient result polling.",

View file

@ -14,6 +14,7 @@ use epiphany_core::{
load_resident_self_state, observe_process_instance, pending_resident_self_acks,
prepare_resident_self_launch, publish_resident_provider_readiness,
recover_receipt_free_dead_coordinator_session,
recover_dead_resident_typed_worker,
resident_cognitive_runtime_id,
resident_prepared_launch_thread_id, resident_self_child_claim,
resident_self_grant_has_typed_request, resident_self_local_provider_status,
@ -318,7 +319,14 @@ fn settle_receipt_free_dead_coordinator(
&args.policy.runtime_store,
&lease.grant_id,
)? {
return Ok(ResidentSelfOutcome::AwaitingFulfillment);
if !recover_dead_resident_typed_worker(
&args.state_store,
&args.policy.runtime_store,
&lease.grant_id,
now,
)? {
return Ok(ResidentSelfOutcome::AwaitingFulfillment);
}
}
}
}

View file

@ -185,7 +185,7 @@ fn commit_coordinator_job_launch_in_cache(
let imagination_consideration_launch = if let Some(request_id) =
request.imagination_consideration_request_id.as_deref()
{
let (consideration, identity) =
let (consideration, identity, attempt_ordinal) =
validate_imagination_consideration_launch(cache, current_state, request, request_id)?;
let model = cache
.get::<EpiphanyMemoryGraphEntry>(MEMORY_GRAPH_KEY)?
@ -219,7 +219,7 @@ fn commit_coordinator_job_launch_in_cache(
"{:x}",
Sha256::digest(rmp_serde::to_vec_named(&effective_launch_document)?)
);
Some((consideration, identity, hash))
Some((consideration, identity, hash, attempt_ordinal))
} else {
None
};
@ -462,15 +462,20 @@ fn commit_coordinator_job_launch_in_cache(
.0,
);
}
if let Some((consideration, identity, worker_launch_document_sha256)) =
if let Some((consideration, identity, worker_launch_document_sha256, attempt_ordinal)) =
imagination_consideration_launch
{
let binding_record_id = if attempt_ordinal == 0 {
format!("imagination-consideration-launch-{}", consideration.request_id)
} else {
format!(
"imagination-consideration-launch-{}-attempt-{attempt_ordinal}",
consideration.request_id
)
};
let binding = ImaginationConsiderationLaunchBinding {
schema_version: IMAGINATION_CONSIDERATION_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id: format!(
"imagination-consideration-launch-{}",
consideration.request_id
),
binding_record_id,
request_id: consideration.request_id,
job_id: plan.backend_job_id.clone(),
binding_id: request.binding_id.clone(),
@ -913,7 +918,7 @@ fn validate_imagination_consideration_launch(
state: &EpiphanyThreadState,
launch: &EpiphanyJobLaunchRequest,
request_id: &str,
) -> Result<(ImaginationConsiderationRequest, EpiphanyRuntimeIdentity)> {
) -> Result<(ImaginationConsiderationRequest, EpiphanyRuntimeIdentity, u64)> {
if match &launch.launch_document {
EpiphanyWorkerLaunchDocument::Role(document) => {
document.imagination_consideration_context.is_some()
@ -947,14 +952,26 @@ fn validate_imagination_consideration_launch(
{
return Err(anyhow!("consideration launch provenance mismatch"));
}
if cache
let prior_bindings = cache
.get_all::<ImaginationConsiderationLaunchBinding>()?
.iter()
.any(|binding| binding.request_id == request_id)
{
return Err(anyhow!("consideration request already bound"));
.filter(|binding| binding.request_id == request_id)
.cloned()
.collect::<Vec<_>>();
for binding in prior_bindings {
if !worker_process_attempt_is_terminal(cache, &binding.job_id)? {
return Err(anyhow!("consideration request already bound"));
}
}
Ok((request, identity))
let attempt_ordinal = u64::try_from(
cache
.get_all::<ImaginationConsiderationLaunchBinding>()?
.iter()
.filter(|binding| binding.request_id == request_id)
.count(),
)
.map_err(|_| anyhow!("consideration attempt ordinal overflow"))?;
Ok((request, identity, attempt_ordinal))
}
fn validate_admitted_model_direction_consideration_launch(
@ -1000,17 +1017,21 @@ fn validate_admitted_model_direction_consideration_launch(
{
return Err(anyhow!("model direction consideration provenance mismatch"));
}
if cache
let prior_launches = cache
.get_all::<EpiphanyRuntimeWorkerLaunchRequest>()?
.iter()
.any(|worker| {
.filter(|worker| {
worker
.admitted_model_direction_consideration_request_id
.as_deref()
== Some(request_id)
})
{
return Err(anyhow!("model direction consideration already bound"));
.cloned()
.collect::<Vec<_>>();
for worker in prior_launches {
if !worker_process_attempt_is_terminal(cache, &worker.job_id)? {
return Err(anyhow!("model direction consideration already bound"));
}
}
Ok(request)
}
@ -1132,15 +1153,18 @@ fn validate_proposal_modeling_launch(
if proposal.source_kind == crate::RepoFrontierProposalSourceKind::Imagination {
crate::runtime_spine::validate_autonomous_proposal_binding(cache, &proposal)?;
}
if cache
let prior_bindings = cache
.get_all::<RepoFrontierProposalModelingLaunchBinding>()?
.iter()
.filter(|binding| binding.proposal_modeling_request_id == request_id)
.any(|binding| !proposal_modeling_launch_was_superseded(state, binding))
{
return Err(anyhow!(
"proposal Modeling selection request already has an unsuperseded launch"
));
.cloned()
.collect::<Vec<_>>();
for binding in prior_bindings {
if !proposal_modeling_launch_was_superseded(cache, &binding)? {
return Err(anyhow!(
"proposal Modeling selection request already has an unsuperseded launch"
));
}
}
let identity = cache
.get::<EpiphanyRuntimeIdentity>(RUNTIME_IDENTITY_KEY)?
@ -1178,17 +1202,23 @@ fn validate_proposal_modeling_launch(
}
fn proposal_modeling_launch_was_superseded(
state: &EpiphanyThreadState,
cache: &CultCache,
binding: &RepoFrontierProposalModelingLaunchBinding,
) -> bool {
state.acceptance_receipts.iter().any(|receipt| {
receipt.job_id == binding.job_id
&& receipt.binding_id == binding.binding_id
&& receipt.surface == "roleFailureReview"
&& receipt.role_id == "modeling"
&& receipt.status == "superseded"
&& !receipt.result_id.trim().is_empty()
})
) -> Result<bool> {
worker_process_attempt_is_terminal(cache, &binding.job_id)
}
fn worker_process_attempt_is_terminal(cache: &CultCache, job_id: &str) -> Result<bool> {
Ok(cache
.get::<crate::EpiphanyRuntimeWorkerProcessClaim>(&format!(
"runtime-worker-process-{job_id}"
))?
.is_some_and(|claim| {
matches!(
claim.status.as_str(),
"terminal-death" | "terminal-unactivated"
)
}))
}
fn terminal_runtime_link_for_binding(
@ -1206,16 +1236,28 @@ fn terminal_runtime_link_for_binding(
if link.runtime_result_id.is_some() {
return Ok(None);
}
let Some(snapshot) = runtime_job_snapshot(runtime_store, &link.runtime_job_id)? else {
if let Some(snapshot) = runtime_job_snapshot(runtime_store, &link.runtime_job_id)?
&& let Some(result) = snapshot.result
{
let mut terminal = link.clone();
terminal.id = format!("{}-{}", link.id, result.result_id);
terminal.surface = "runtimeResult".to_string();
terminal.runtime_result_id = Some(result.result_id);
return Ok(Some(terminal));
}
let Some(claim) = crate::runtime_worker_process_claim(runtime_store, &link.runtime_job_id)? else {
return Ok(None);
};
let Some(result) = snapshot.result else {
if !matches!(claim.status.as_str(), "terminal-death" | "terminal-unactivated") {
return Ok(None);
};
}
let terminal_authority_id = claim
.terminal_authority_id
.ok_or_else(|| anyhow!("terminal worker process claim lost its authority id"))?;
let mut terminal = link.clone();
terminal.id = format!("{}-{}", link.id, result.result_id);
terminal.surface = "runtimeResult".to_string();
terminal.runtime_result_id = Some(result.result_id);
terminal.id = format!("{}-{}", link.id, terminal_authority_id);
terminal.surface = "runtimeWorkerProcessTerminal".to_string();
terminal.runtime_result_id = Some(terminal_authority_id);
Ok(Some(terminal))
}
@ -1644,6 +1686,67 @@ pub(crate) mod tests {
})
}
fn model_direction_launch_fixture(
root: &Path,
suffix: &str,
) -> Result<(
std::path::PathBuf,
EpiphanyThreadState,
EpiphanyJobLaunchRequest,
AdmittedModelDirectionConsiderationRequest,
)> {
let (store, state, _, planning) = frontier_planning_launch_fixture(root, suffix)?;
let request = AdmittedModelDirectionConsiderationRequest {
schema_version: ADMITTED_MODEL_DIRECTION_CONSIDERATION_REQUEST_SCHEMA_VERSION.into(),
request_id: format!("model-direction-{suffix}"),
runtime_id: planning.runtime_id,
thread_id: planning.thread_id.clone(),
model_revision: planning.model_revision,
model_hash: planning.model_hash,
model_admission_receipt_id: planning.admission_receipt_id,
previous_terminal_result_id: None,
requested_at: "2026-07-18T00:00:00Z".into(),
contract: ADMITTED_MODEL_DIRECTION_CONSIDERATION_REQUEST_CONTRACT.into(),
private_state_included: false,
};
let mut cache = coordinator_acceptance_cache(&store)?;
cache.put(&request.request_id, &request)?;
let mut launch = build_epiphany_role_launch_request(
&request.thread_id,
EpiphanyRoleResultRoleId::Imagination,
Some(state.revision),
Some(60),
&state,
)
.map_err(|error| anyhow!(error))?;
launch.admitted_model_direction_consideration_request_id =
Some(request.request_id.clone());
Ok((store, state, launch, request))
}
fn terminalize_test_worker_process(store: &Path, job_id: &str, suffix: &str) -> Result<()> {
let process = crate::ProcessInstanceIdentity {
process_id: 10_000 + u32::try_from(suffix.len())?,
creation_token: 20_000 + u64::try_from(suffix.len())?,
executable_path: std::path::PathBuf::from(format!("/tmp/epiphany-worker-{suffix}")),
created_at_rfc3339: Some("2026-07-18T00:00:01Z".into()),
};
crate::claim_runtime_worker_process(
store,
job_id,
&process,
&"a".repeat(64),
"2026-07-18T00:00:02Z",
)?;
crate::runtime_spine::terminalize_runtime_worker_process_death(
store,
job_id,
&format!("worker-death-{suffix}"),
"2026-07-18T00:00:03Z",
)?;
Ok(())
}
fn launch_frontier_mind_result(
store: &Path,
imagination_result: &EpiphanyRuntimeRoleWorkerResult,
@ -1998,7 +2101,19 @@ pub(crate) mod tests {
}
#[test]
fn proposal_launch_binding_is_reusable_only_after_exact_failure_supersession() {
fn proposal_launch_binding_is_reusable_only_after_exact_process_terminality() -> Result<()> {
let root = tempfile::tempdir()?;
let store = root.path().join("proposal-supersession.cc");
crate::initialize_runtime_spine(
&store,
crate::RuntimeSpineInitOptions {
runtime_id: "runtime-1".into(),
display_name: "Proposal supersession".into(),
created_at: "2026-07-13T05:00:00Z".into(),
},
)?;
let mut cache = crate::runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
let binding = RepoFrontierProposalModelingLaunchBinding {
schema_version: REPO_FRONTIER_PROPOSAL_MODELING_LAUNCH_BINDING_SCHEMA_VERSION.into(),
binding_record_id: "proposal-launch-1".into(),
@ -2014,7 +2129,7 @@ pub(crate) mod tests {
contract: REPO_FRONTIER_PROPOSAL_MODELING_LAUNCH_BINDING_CONTRACT.into(),
};
let mut state = EpiphanyThreadState::default();
assert!(!proposal_modeling_launch_was_superseded(&state, &binding));
assert!(!proposal_modeling_launch_was_superseded(&cache, &binding)?);
state
.acceptance_receipts
@ -2029,10 +2144,27 @@ pub(crate) mod tests {
accepted_at: "2026-07-13T05:01:00Z".into(),
..Default::default()
});
assert!(proposal_modeling_launch_was_superseded(&state, &binding));
assert!(!proposal_modeling_launch_was_superseded(&cache, &binding)?);
state.acceptance_receipts[0].job_id = "other-job".into();
assert!(!proposal_modeling_launch_was_superseded(&state, &binding));
assert!(!proposal_modeling_launch_was_superseded(&cache, &binding)?);
let terminal_claim = crate::EpiphanyRuntimeWorkerProcessClaim {
schema_version: crate::RUNTIME_WORKER_PROCESS_CLAIM_SCHEMA_VERSION.into(),
claim_id: "runtime-worker-process-job-1".into(),
job_id: "job-1".into(),
process_id: 41,
process_creation_token: 73,
process_executable_path: "/release/epiphany-model-runtime".into(),
activation_token_sha256: "a".repeat(64),
status: "terminal-death".into(),
claimed_at: "2026-07-13T05:00:04Z".into(),
activated_at: Some("2026-07-13T05:00:05Z".into()),
terminal_at: Some("2026-07-13T05:00:06Z".into()),
terminal_authority_id: Some("worker-death-recovery-job-1".into()),
};
cache.put(&terminal_claim.claim_id, &terminal_claim)?;
assert!(proposal_modeling_launch_was_superseded(&cache, &binding)?);
Ok(())
}
#[test]
@ -2266,6 +2398,187 @@ pub(crate) mod tests {
Ok(())
}
#[test]
fn consideration_relaunch_requires_exact_terminal_process_and_preserves_attempt_history(
) -> Result<()> {
let root = tempfile::tempdir()?;
let (store, state, launch, request) = consideration_launch_fixture(root.path(), "retry")?;
let first = plan_coordinator_job_launch(
&state,
&launch,
&store,
"launcher-consideration-first".into(),
"backend-consideration-first".into(),
)?;
commit_coordinator_job_launch(
&store,
&request.thread_id,
&state,
&launch,
&first,
"2026-07-18T00:00:01Z".into(),
)?;
let current_state = coordinator_acceptance_cache(&store)?
.get_required::<EpiphanyThreadStateEntry>(THREAD_STATE_KEY)?
.state()?;
let mut retry_launch = build_epiphany_role_launch_request(
&request.thread_id,
EpiphanyRoleResultRoleId::Imagination,
Some(current_state.revision),
Some(60),
&current_state,
)
.map_err(|error| anyhow!(error))?;
retry_launch.imagination_consideration_request_id = Some(request.request_id.clone());
let before_refusal = std::fs::read(&store)?;
assert!(plan_coordinator_job_launch(
&current_state,
&retry_launch,
&store,
"launcher-consideration-refused".into(),
"backend-consideration-refused".into(),
)
.is_err());
assert_eq!(std::fs::read(&store)?, before_refusal);
terminalize_test_worker_process(
&store,
"backend-consideration-first",
"consideration-first",
)?;
let first_binding = coordinator_acceptance_cache(&store)?
.get::<ImaginationConsiderationLaunchBinding>(&format!(
"imagination-consideration-launch-{}",
request.request_id
))?
.expect("first immutable consideration binding");
let first_claim = crate::runtime_worker_process_claim(
&store,
"backend-consideration-first",
)?
.expect("terminal first process claim");
let second = plan_coordinator_job_launch(
&current_state,
&retry_launch,
&store,
"launcher-consideration-second".into(),
"backend-consideration-second".into(),
)?;
commit_coordinator_job_launch(
&store,
&request.thread_id,
&current_state,
&retry_launch,
&second,
"2026-07-18T00:00:04Z".into(),
)?;
let cache = coordinator_acceptance_cache(&store)?;
let second_binding = cache
.get::<ImaginationConsiderationLaunchBinding>(&format!(
"imagination-consideration-launch-{}-attempt-1",
request.request_id
))?
.expect("second immutable consideration binding");
assert_eq!(first_binding.job_id, "backend-consideration-first");
assert_eq!(second_binding.job_id, "backend-consideration-second");
assert_eq!(
cache.get::<ImaginationConsiderationLaunchBinding>(&first_binding.binding_record_id)?,
Some(first_binding)
);
assert_eq!(
crate::runtime_worker_process_claim(&store, "backend-consideration-first")?,
Some(first_claim)
);
Ok(())
}
#[test]
fn model_direction_relaunch_requires_exact_terminal_process() -> Result<()> {
let root = tempfile::tempdir()?;
let (store, state, launch, request) =
model_direction_launch_fixture(root.path(), "retry")?;
let first = plan_coordinator_job_launch(
&state,
&launch,
&store,
"launcher-model-direction-first".into(),
"backend-model-direction-first".into(),
)?;
commit_coordinator_job_launch(
&store,
&request.thread_id,
&state,
&launch,
&first,
"2026-07-18T00:00:01Z".into(),
)?;
let current_state = coordinator_acceptance_cache(&store)?
.get_required::<EpiphanyThreadStateEntry>(THREAD_STATE_KEY)?
.state()?;
let mut retry_launch = build_epiphany_role_launch_request(
&request.thread_id,
EpiphanyRoleResultRoleId::Imagination,
Some(current_state.revision),
Some(60),
&current_state,
)
.map_err(|error| anyhow!(error))?;
retry_launch.admitted_model_direction_consideration_request_id =
Some(request.request_id.clone());
let before_refusal = std::fs::read(&store)?;
assert!(plan_coordinator_job_launch(
&current_state,
&retry_launch,
&store,
"launcher-model-direction-refused".into(),
"backend-model-direction-refused".into(),
)
.is_err());
assert_eq!(std::fs::read(&store)?, before_refusal);
terminalize_test_worker_process(
&store,
"backend-model-direction-first",
"model-direction-first",
)?;
let first_launch = crate::runtime_worker_launch_request(
&store,
"backend-model-direction-first",
)?
.expect("first immutable model direction launch");
let second = plan_coordinator_job_launch(
&current_state,
&retry_launch,
&store,
"launcher-model-direction-second".into(),
"backend-model-direction-second".into(),
)?;
commit_coordinator_job_launch(
&store,
&request.thread_id,
&current_state,
&retry_launch,
&second,
"2026-07-18T00:00:04Z".into(),
)?;
let second_launch = crate::runtime_worker_launch_request(
&store,
"backend-model-direction-second",
)?
.expect("second immutable model direction launch");
assert_ne!(first_launch.job_id, second_launch.job_id);
assert_eq!(
first_launch.admitted_model_direction_consideration_request_id,
second_launch.admitted_model_direction_consideration_request_id
);
assert_eq!(
crate::runtime_worker_launch_request(&store, "backend-model-direction-first")?,
Some(first_launch)
);
Ok(())
}
#[test]
fn consideration_refuses_caller_context_stale_model_and_substituted_candidate() -> Result<()> {
for mutation in 0..3 {

View file

@ -1005,6 +1005,7 @@ pub use runtime_spine::EpiphanyRuntimeSpineStatus;
pub use runtime_spine::EpiphanyRuntimeSwarmBinding;
pub use runtime_spine::EpiphanyRuntimeToolExecutionBinding;
pub use runtime_spine::EpiphanyRuntimeWorkerLaunchRequest;
pub use runtime_spine::EpiphanyRuntimeWorkerProcessClaim;
pub use runtime_spine::EpiphanyToolInvocationStatus;
pub use runtime_spine::PreparedRuntimeSpineHeartbeatJob;
pub use runtime_spine::RUNTIME_EVENT_TYPE;
@ -1026,6 +1027,8 @@ pub use runtime_spine::RUNTIME_TOOL_EXECUTION_BINDING_SCHEMA_VERSION;
pub use runtime_spine::RUNTIME_TOOL_EXECUTION_BINDING_TYPE;
pub use runtime_spine::RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION;
pub use runtime_spine::RUNTIME_WORKER_LAUNCH_REQUEST_TYPE;
pub use runtime_spine::RUNTIME_WORKER_PROCESS_CLAIM_SCHEMA_VERSION;
pub use runtime_spine::RUNTIME_WORKER_PROCESS_CLAIM_TYPE;
pub use runtime_spine::RepoFrontierVerdictModelingLaunchAuthority;
pub use runtime_spine::RuntimeHandsReceiptChainSummary;
pub use runtime_spine::RuntimeSpineEventOptions;
@ -1041,12 +1044,15 @@ pub use runtime_spine::RuntimeTypedFulfillmentEvidence;
pub use runtime_spine::RuntimeTypedRequestRef;
pub use runtime_spine::amend_repo_frontier_execution;
pub use runtime_spine::append_runtime_event;
pub use runtime_spine::abandon_unactivated_runtime_worker_process;
pub use runtime_spine::activate_runtime_worker_process;
pub use runtime_spine::archive_completed_coordinator_session;
pub use runtime_spine::archive_completed_model_session;
pub use runtime_spine::bind_runtime_repository_domain;
pub use runtime_spine::bind_runtime_to_agent_memory_swarm;
pub use runtime_spine::canonical_repo_frontier_plan_candidate_id;
pub use runtime_spine::clear_epiphany_job_binding_backend;
pub use runtime_spine::claim_runtime_worker_process;
pub use runtime_spine::close_runtime_session;
pub use runtime_spine::commit_operator_repo_frontier_plan_review;
pub use runtime_spine::commit_repo_frontier_modeling_request;
@ -1148,6 +1154,7 @@ pub use runtime_spine::runtime_typed_request_fulfillment;
pub use runtime_spine::runtime_typed_request_attempt_exists;
pub use runtime_spine::runtime_worker_launch_body_basis;
pub use runtime_spine::runtime_worker_launch_request;
pub use runtime_spine::runtime_worker_process_claim;
pub use runtime_spine::select_and_commit_repo_frontier_planning_request;
pub use runtime_spine::select_and_commit_repo_frontier_route;
pub use runtime_spine::select_repo_frontier_work_proposal_for_modeling;

View file

@ -1281,6 +1281,109 @@ pub fn resident_self_typed_attempt_exists(
crate::runtime_typed_request_attempt_exists(runtime_store, request)
}
pub fn recover_dead_resident_typed_worker(
resident_store: &Path,
runtime_store: &Path,
grant_id: &str,
recovered_at_millis: u64,
) -> Result<bool> {
let grant = state_cache(resident_store)?
.get::<ResidentSelfHeartbeatGrant>(grant_id)?
.ok_or_else(|| anyhow!("typed worker death recovery lost its grant"))?;
let Some(request) = resident_self_typed_request_ref(&grant)? else {
return Ok(false);
};
let request_id = match request {
crate::RuntimeTypedRequestRef::ProposalModeling(id)
| crate::RuntimeTypedRequestRef::ImaginationConsideration(id)
| crate::RuntimeTypedRequestRef::AdmittedModelDirection(id) => id,
};
let mut cache = crate::runtime_spine_cache(runtime_store)?;
cache.pull_all_backing_stores()?;
let mut launches = cache
.get_all::<crate::EpiphanyRuntimeWorkerLaunchRequest>()?
.into_iter()
.filter(|launch| match request {
crate::RuntimeTypedRequestRef::ProposalModeling(_) => {
launch.proposal_modeling_request_id.as_deref() == Some(request_id)
}
crate::RuntimeTypedRequestRef::ImaginationConsideration(_) => {
launch.imagination_consideration_request_id.as_deref() == Some(request_id)
}
crate::RuntimeTypedRequestRef::AdmittedModelDirection(_) => launch
.admitted_model_direction_consideration_request_id
.as_deref()
== Some(request_id),
})
.collect::<Vec<_>>();
launches.sort_by(|a, b| a.job_id.cmp(&b.job_id));
let mut retryable = Vec::new();
for launch in &launches {
let claim = cache.get::<crate::EpiphanyRuntimeWorkerProcessClaim>(&format!(
"runtime-worker-process-{}",
launch.job_id
))?;
retryable.push((launch, claim));
}
let live = retryable
.iter()
.filter(|(_, claim)| {
claim.as_ref().is_some_and(|claim| {
matches!(claim.status.as_str(), "claimed" | "active" | "terminal-result")
})
})
.count();
if live > 1 {
return Err(anyhow!("typed request has multiple live worker process claims"));
}
let selected = retryable
.iter()
.find(|(_, claim)| {
claim.as_ref().is_some_and(|claim| {
matches!(claim.status.as_str(), "claimed" | "active" | "terminal-result")
})
})
.or_else(|| retryable.last());
let Some((launch, Some(claim))) = selected else {
return Ok(false);
};
match claim.status.as_str() {
"terminal-death" | "terminal-unactivated" => return Ok(true),
"terminal-result" => return Ok(false),
"claimed" | "active" => {}
_ => return Err(anyhow!("typed worker process claim has invalid status")),
}
let identity = crate::ProcessInstanceIdentity {
process_id: claim.process_id,
creation_token: claim.process_creation_token,
created_at_rfc3339: None,
executable_path: PathBuf::from(&claim.process_executable_path),
};
match crate::observe_process_instance(&identity) {
crate::ProcessInstanceObservation::ExactAlive => Ok(false),
crate::ProcessInstanceObservation::Inaccessible
| crate::ProcessInstanceObservation::Indeterminate { .. } => Err(anyhow!(
"Continuity cannot prove the exact typed worker process dead"
)),
crate::ProcessInstanceObservation::ExactExited { .. }
| crate::ProcessInstanceObservation::Missing
| crate::ProcessInstanceObservation::Replaced { .. } => {
let recovered_at = chrono::DateTime::<chrono::Utc>::from_timestamp_millis(
recovered_at_millis as i64,
)
.ok_or_else(|| anyhow!("typed worker death timestamp is out of range"))?
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
crate::runtime_spine::terminalize_runtime_worker_process_death(
runtime_store,
&launch.job_id,
&format!("worker-death-recovery-{}", launch.job_id),
&recovered_at,
)?;
Ok(true)
}
}
}
pub fn resident_self_grant_has_typed_request(
resident_store: &Path,
grant_id: &str,
@ -1335,13 +1438,50 @@ pub fn settle_resident_self_exited_coordinator(
});
}
match verify_resident_self_grant_fulfillment(resident_store, runtime_store, &lease.grant_id) {
Ok(ResidentSelfGrantFulfillment::Pending) if typed && typed_attempt => {
if !recover_dead_resident_typed_worker(
resident_store,
runtime_store,
&lease.grant_id,
now_millis,
)? {
return Ok(ResidentSelfOutcome::AwaitingFulfillment);
}
cancel_resident_self_turn(
resident_store,
lease,
"process-failed",
"Runtime Continuity proved the exact activated worker process terminal before typed fulfillment",
now_millis,
)?;
Ok(if shutdown_requested {
ResidentSelfOutcome::Braked
} else {
ResidentSelfOutcome::Failed
})
}
Ok(ResidentSelfGrantFulfillment::Pending)
if typed
&& (typed_attempt
|| (successful_receipt
&& !(shutdown_requested || brake_engaged || timed_out))) =>
&& successful_receipt
&& !(shutdown_requested || brake_engaged || timed_out) =>
{
Ok(ResidentSelfOutcome::AwaitingFulfillment)
if recover_dead_resident_typed_worker(
resident_store,
runtime_store,
&lease.grant_id,
now_millis,
)? {
cancel_resident_self_turn(
resident_store,
lease,
"process-failed",
"Runtime Continuity proved the exact worker attempt terminal before activation or fulfillment",
now_millis,
)?;
Ok(ResidentSelfOutcome::Failed)
} else {
Ok(ResidentSelfOutcome::AwaitingFulfillment)
}
}
Ok(ResidentSelfGrantFulfillment::Pending) => {
let status = if shutdown_requested {
@ -4334,6 +4474,293 @@ mod tests {
Ok(())
}
#[test]
fn exact_dead_typed_worker_claim_becomes_retry_authority() -> Result<()> {
let temp = tempfile::tempdir()?;
let resident_store = temp.path().join("resident.cc");
let runtime_store = temp.path().join("runtime.cc");
crate::initialize_runtime_spine(
&runtime_store,
crate::RuntimeSpineInitOptions {
runtime_id: "test-runtime".into(),
display_name: "Typed death runtime".into(),
created_at: "2026-08-10T12:20:00Z".into(),
},
)?;
enqueue_resident_self_pressure(
&resident_store,
&ResidentSelfPressure {
schema_version: RESIDENT_SELF_PRESSURE_SCHEMA_VERSION.into(),
pressure_id: "typed-worker-death-pressure".into(),
kind: "repo-frontier-proposal-modeling".into(),
provenance_ref: "cultcache://repo-frontier-proposal-modeling/death-request".into(),
objective: "Recover exact dead typed worker.".into(),
created_at_millis: 1,
status: "pending".into(),
consumed_by_grant_id: None,
private_state_exposed: false,
},
)?;
let grant = heartbeat_issue_resident_self_grant(
&resident_store,
"typed-death-heartbeat",
"typed-death-action",
2,
)?
.expect("typed death grant");
let document = crate::EpiphanyWorkerLaunchDocument::Role(
crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: "typed-death-thread".into(),
role_id: "modeling".into(),
state_revision: 0,
objective: None,
dynamic_prompt_context: None,
repository_body_observation_basis: None,
proposal_modeling_context: None,
claim_repair_context: None,
frontier_planning_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 launch = crate::EpiphanyRuntimeWorkerLaunchRequest {
schema_version: crate::RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION.into(),
job_id: "typed-death-job".into(),
binding_id: crate::EPIPHANY_MODELING_ROLE_BINDING_ID.into(),
role: crate::EPIPHANY_MODELING_OWNER_ROLE.into(),
authority_scope: "modeling".into(),
instruction: "Typed death fixture.".into(),
output_contract_id: crate::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: rmp_serde::to_vec_named(&document)?,
metadata: BTreeMap::new(),
organ_launch_contract: crate::default_launch_organ_contract(
"modeling",
"role",
crate::ROLE_WORKER_OUTPUT_CONTRACT_ID,
),
proposal_modeling_request_id: Some("death-request".into()),
claim_repair_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: None,
repo_frontier_modeling_request_id: None,
repo_frontier_verdict_modeling_authority_msgpack: None,
};
let mut runtime = crate::runtime_spine_cache(&runtime_store)?;
runtime.put(&launch.job_id, &launch)?;
let missing_process = crate::ProcessInstanceIdentity {
process_id: u32::MAX - 2,
creation_token: 91,
created_at_rfc3339: None,
executable_path: PathBuf::from("/release/epiphany-model-runtime"),
};
crate::claim_runtime_worker_process(
&runtime_store,
&launch.job_id,
&missing_process,
&"b".repeat(64),
"2026-08-10T12:20:01Z",
)?;
assert!(resident_self_typed_attempt_exists(
&resident_store,
&runtime_store,
&grant.grant_id,
)?);
assert!(recover_dead_resident_typed_worker(
&resident_store,
&runtime_store,
&grant.grant_id,
1_786_360_000_000,
)?);
assert!(!resident_self_typed_attempt_exists(
&resident_store,
&runtime_store,
&grant.grant_id,
)?);
assert_eq!(
crate::runtime_worker_process_claim(&runtime_store, &launch.job_id)?
.expect("terminal death claim")
.status,
"terminal-death"
);
Ok(())
}
#[test]
fn exact_dead_typed_worker_settles_active_lease_by_canonical_requeue() -> Result<()> {
let temp = tempfile::tempdir()?;
let resident_store = temp.path().join("resident.cc");
let coordinator = temp.path().join("epiphany-mvp-coordinator");
std::fs::write(&coordinator, b"witnessed executable")?;
let mut policy = policy();
policy.coordinator_bin = coordinator.clone();
policy.runtime_store = temp.path().join("runtime.cc");
seed_model_direction_request(
&policy.runtime_store,
"request-worker-death-settlement",
"cognitive-runtime",
"thread-worker-death-settlement",
)?;
enqueue_resident_self_pressure(
&resident_store,
&ResidentSelfPressure {
schema_version: RESIDENT_SELF_PRESSURE_SCHEMA_VERSION.into(),
pressure_id: "pressure-worker-death-settlement".into(),
kind: "admitted-model-direction-consideration".into(),
provenance_ref: "cultcache://admitted-model-direction-consideration/request-worker-death-settlement".into(),
objective: "Prove exact worker death settles the resident lease.".into(),
created_at_millis: 1,
status: "pending".into(),
consumed_by_grant_id: None,
private_state_exposed: false,
},
)?;
let grant = heartbeat_issue_resident_self_grant(
&resident_store,
"heartbeat-worker-death",
"action-worker-death",
2,
)?
.expect("worker death grant");
let prepared = prepare_resident_self_launch(&resident_store, &policy, 3)?
.expect("worker death preparation");
let coordinator_process = LaunchedCoordinator {
process_id: 44,
process_creation_token: 55,
process_executable_path: coordinator,
};
claim_resident_self_preparation_as_child(
&resident_store,
&prepared.preparation_id,
&coordinator_process,
4,
)?;
let lease = acknowledge_resident_self_launch(
&resident_store,
&prepared.preparation_id,
&coordinator_process,
5,
)?;
let document = crate::EpiphanyWorkerLaunchDocument::Role(
crate::EpiphanyRoleWorkerLaunchDocument {
thread_id: "thread-worker-death-settlement".into(),
role_id: "imagination".into(),
state_revision: 0,
objective: None,
dynamic_prompt_context: None,
repository_body_observation_basis: None,
proposal_modeling_context: None,
claim_repair_context: None,
frontier_planning_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 launch = crate::EpiphanyRuntimeWorkerLaunchRequest {
schema_version: crate::RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION.into(),
job_id: "worker-death-settlement-job".into(),
binding_id: crate::EPIPHANY_IMAGINATION_ROLE_BINDING_ID.into(),
role: crate::EPIPHANY_IMAGINATION_OWNER_ROLE.into(),
authority_scope: "imagination".into(),
instruction: "Typed worker death settlement fixture.".into(),
output_contract_id: crate::ROLE_WORKER_OUTPUT_CONTRACT_ID.into(),
document_kind: "role".into(),
launch_document_msgpack: rmp_serde::to_vec_named(&document)?,
metadata: BTreeMap::new(),
organ_launch_contract: crate::default_launch_organ_contract(
"imagination",
"role",
crate::ROLE_WORKER_OUTPUT_CONTRACT_ID,
),
proposal_modeling_request_id: None,
claim_repair_request_id: None,
frontier_planning_request_id: None,
frontier_plan_mind_request_id: None,
imagination_consideration_request_id: None,
admitted_model_direction_consideration_request_id: Some(
"request-worker-death-settlement".into(),
),
repo_frontier_modeling_request_id: None,
repo_frontier_verdict_modeling_authority_msgpack: None,
};
crate::runtime_spine_cache(&policy.runtime_store)?.put(&launch.job_id, &launch)?;
let missing_worker = crate::ProcessInstanceIdentity {
process_id: u32::MAX - 3,
creation_token: 92,
created_at_rfc3339: None,
executable_path: PathBuf::from("/release/epiphany-imagination-runtime"),
};
crate::claim_runtime_worker_process(
&policy.runtime_store,
&launch.job_id,
&missing_worker,
&"c".repeat(64),
"2026-08-10T12:30:01Z",
)?;
let receipt = coordinator_receipt_for_lease(&lease, &policy.runtime_store);
assert_eq!(
settle_resident_self_exited_coordinator(
&resident_store,
&policy.runtime_store,
&lease,
&receipt,
false,
false,
false,
1_786_360_600_000,
policy.cooldown_seconds,
)?,
ResidentSelfOutcome::Failed
);
assert!(load_resident_self_state(&resident_store)?.active_turn.is_none());
let terminal_grant = state_cache(&resident_store)?
.get::<ResidentSelfHeartbeatGrant>(&grant.grant_id)?
.expect("terminal grant");
assert_eq!(terminal_grant.terminal_status.as_deref(), Some("process-failed"));
let pressure = state_cache(&resident_store)?
.get::<ResidentSelfPressure>("pressure-worker-death-settlement")?
.expect("requeued pressure");
assert_eq!(pressure.status, "pending");
assert_eq!(pressure.consumed_by_grant_id, None);
assert_eq!(
crate::runtime_worker_process_claim(&policy.runtime_store, &launch.job_id)?
.expect("terminal worker death claim")
.status,
"terminal-death"
);
Ok(())
}
#[test]
fn standard_heartbeat_owns_self_grant_selection() -> Result<()> {
let temp = tempfile::tempdir()?;

View file

@ -166,6 +166,8 @@ pub const RUNTIME_MODEL_EXECUTION_BINDING_TYPE: &str = "epiphany.runtime.model_e
pub const RUNTIME_TOOL_EXECUTION_BINDING_TYPE: &str = "epiphany.runtime.tool_execution_binding";
pub const ARCHIVED_RUNTIME_SESSION_TYPE: &str = "epiphany.runtime.archived_session";
pub const RUNTIME_WORKER_LAUNCH_REQUEST_TYPE: &str = "epiphany.runtime.worker_launch_request";
pub const RUNTIME_WORKER_PROCESS_CLAIM_TYPE: &str =
"epiphany.runtime.worker_process_claim.v0";
pub const RUNTIME_ROLE_WORKER_RESULT_TYPE: &str = "epiphany.runtime.role_worker_result";
pub const RUNTIME_REORIENT_WORKER_RESULT_TYPE: &str = "epiphany.runtime.reorient_worker_result";
pub const RUNTIME_JOB_RESULT_TYPE: &str = "epiphany.runtime.job_result";
@ -197,6 +199,8 @@ pub const RUNTIME_TOOL_EXECUTION_BINDING_SCHEMA_VERSION: &str =
pub const ARCHIVED_RUNTIME_SESSION_SCHEMA_VERSION: &str = "epiphany.runtime.archived_session.v0";
pub const RUNTIME_WORKER_LAUNCH_REQUEST_SCHEMA_VERSION: &str =
"epiphany.runtime.worker_launch_request.v1";
pub const RUNTIME_WORKER_PROCESS_CLAIM_SCHEMA_VERSION: &str =
"epiphany.runtime.worker_process_claim.v0";
pub const RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION: &str =
"epiphany.runtime.role_worker_result.v3";
pub const RUNTIME_REORIENT_WORKER_RESULT_SCHEMA_VERSION: &str =
@ -451,6 +455,42 @@ pub struct EpiphanyRuntimeWorkerLaunchRequest {
pub repo_frontier_verdict_modeling_authority_msgpack: Option<Vec<u8>>,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "epiphany.runtime.worker_process_claim.v0",
schema = "EpiphanyRuntimeWorkerProcessClaim"
)]
pub struct EpiphanyRuntimeWorkerProcessClaim {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub claim_id: String,
#[cultcache(key = 2)]
pub job_id: String,
#[cultcache(key = 3)]
pub process_id: u32,
#[cultcache(key = 4)]
pub process_creation_token: u64,
#[cultcache(key = 5)]
pub process_executable_path: String,
#[cultcache(key = 6)]
pub activation_token_sha256: String,
#[cultcache(key = 7)]
pub status: String,
#[cultcache(key = 8)]
pub claimed_at: String,
#[cultcache(key = 9, default)]
pub activated_at: Option<String>,
#[cultcache(key = 10, default)]
pub terminal_at: Option<String>,
#[cultcache(key = 11, default)]
pub terminal_authority_id: Option<String>,
}
fn worker_process_claim_id(job_id: &str) -> String {
format!("runtime-worker-process-{job_id}")
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepoFrontierVerdictModelingLaunchAuthority {
pub request: RepoFrontierModelingRequest,
@ -1044,6 +1084,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef<Path>) -> Result<CultCache> {
cache.register_entry_type::<EpiphanyRuntimeToolExecutionBinding>()?;
cache.register_entry_type::<EpiphanyArchivedRuntimeSession>()?;
cache.register_entry_type::<EpiphanyRuntimeWorkerLaunchRequest>()?;
cache.register_entry_type::<EpiphanyRuntimeWorkerProcessClaim>()?;
cache.register_entry_type::<EpiphanyRuntimeRoleWorkerResult>()?;
cache.register_entry_type::<crate::EpiphanyMemoryGraphEntry>()?;
cache.register_entry_type::<RepoModelAdmissionReview>()?;
@ -3370,6 +3411,221 @@ pub fn runtime_worker_launch_body_basis(
.repository_body_observation_basis()
}
pub fn runtime_worker_process_claim(
store_path: impl AsRef<Path>,
job_id: &str,
) -> Result<Option<EpiphanyRuntimeWorkerProcessClaim>> {
validate_non_empty(job_id, "worker process claim job id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
cache.get::<EpiphanyRuntimeWorkerProcessClaim>(&worker_process_claim_id(job_id))
}
pub fn claim_runtime_worker_process(
store_path: impl AsRef<Path>,
job_id: &str,
process: &crate::ProcessInstanceIdentity,
activation_token_sha256: &str,
claimed_at: &str,
) -> Result<EpiphanyRuntimeWorkerProcessClaim> {
let store_path = store_path.as_ref();
validate_non_empty(job_id, "worker process claim job id")?;
validate_non_empty(activation_token_sha256, "worker activation token digest")?;
if activation_token_sha256.len() != 64
|| !activation_token_sha256
.bytes()
.all(|byte| byte.is_ascii_hexdigit())
{
return Err(anyhow!("worker activation token digest is not SHA-256 hex"));
}
chrono::DateTime::parse_from_rfc3339(claimed_at)
.map_err(|error| anyhow!("worker process claim timestamp is invalid: {error}"))?;
if process.process_id == 0
|| process.creation_token == 0
|| process.executable_path.as_os_str().is_empty()
{
return Err(anyhow!("worker process claim identity is incomplete"));
}
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
if cache
.get::<EpiphanyRuntimeWorkerLaunchRequest>(job_id)?
.is_none()
{
return Err(anyhow!("worker process claim requires its immutable launch"));
}
if cache
.get::<EpiphanyRuntimeRoleWorkerResult>(job_id)?
.is_some()
{
return Err(anyhow!("worker process cannot claim after terminal result"));
}
let claim_id = worker_process_claim_id(job_id);
let claim = EpiphanyRuntimeWorkerProcessClaim {
schema_version: RUNTIME_WORKER_PROCESS_CLAIM_SCHEMA_VERSION.into(),
claim_id: claim_id.clone(),
job_id: job_id.into(),
process_id: process.process_id,
process_creation_token: process.creation_token,
process_executable_path: process.executable_path.display().to_string(),
activation_token_sha256: activation_token_sha256.into(),
status: "claimed".into(),
claimed_at: claimed_at.into(),
activated_at: None,
terminal_at: None,
terminal_authority_id: None,
};
if let Some(existing) = cache.get::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)? {
return if existing == claim {
Ok(existing)
} else {
Err(anyhow!("worker process claim identity is already owned"))
};
}
let envelope = cache.prepare_entry(&claim_id, &claim)?.0;
if SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&[], vec![envelope])?
{
Ok(claim)
} else {
Err(anyhow!("worker process claim lost immutable insertion"))
}
}
pub fn activate_runtime_worker_process(
store_path: impl AsRef<Path>,
job_id: &str,
process: &crate::ProcessInstanceIdentity,
activation_token: &str,
activated_at: &str,
) -> Result<EpiphanyRuntimeWorkerProcessClaim> {
let store_path = store_path.as_ref();
validate_non_empty(activation_token, "worker activation token")?;
chrono::DateTime::parse_from_rfc3339(activated_at)
.map_err(|error| anyhow!("worker activation timestamp is invalid: {error}"))?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let claim_id = worker_process_claim_id(job_id);
let current = cache
.get::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker activation requires its process claim"))?;
let digest = format!("{:x}", Sha256::digest(activation_token.as_bytes()));
if current.job_id != job_id
|| current.process_id != process.process_id
|| current.process_creation_token != process.creation_token
|| current.process_executable_path != process.executable_path.display().to_string()
|| current.activation_token_sha256 != digest
{
return Err(anyhow!("worker activation does not bind the exact process claim"));
}
if current.status == "active" {
return Ok(current);
}
if current.status != "claimed" {
return Err(anyhow!("worker activation found terminal process authority"));
}
let current_envelope = cache
.get_envelope::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker activation lost its claim envelope"))?;
let mut next = current;
next.status = "active".into();
next.activated_at = Some(activated_at.into());
let next_envelope = cache.prepare_entry(&claim_id, &next)?.0;
if SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&[current_envelope], vec![next_envelope])?
{
Ok(next)
} else {
Err(anyhow!("worker activation lost its exact claim snapshot"))
}
}
pub fn abandon_unactivated_runtime_worker_process(
store_path: impl AsRef<Path>,
job_id: &str,
process: &crate::ProcessInstanceIdentity,
terminal_at: &str,
) -> Result<EpiphanyRuntimeWorkerProcessClaim> {
let store_path = store_path.as_ref();
chrono::DateTime::parse_from_rfc3339(terminal_at)
.map_err(|error| anyhow!("worker abandonment timestamp is invalid: {error}"))?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let claim_id = worker_process_claim_id(job_id);
let current = cache
.get::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker abandonment requires its process claim"))?;
if current.process_id != process.process_id
|| current.process_creation_token != process.creation_token
|| current.process_executable_path != process.executable_path.display().to_string()
|| current.status != "claimed"
{
return Err(anyhow!("worker abandonment does not own an unactivated exact claim"));
}
let current_envelope = cache
.get_envelope::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker abandonment lost its claim envelope"))?;
let mut next = current;
next.status = "terminal-unactivated".into();
next.terminal_at = Some(terminal_at.into());
next.terminal_authority_id = Some(format!("worker-unactivated-{job_id}"));
let next_envelope = cache.prepare_entry(&claim_id, &next)?.0;
if SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&[current_envelope], vec![next_envelope])?
{
Ok(next)
} else {
Err(anyhow!("worker abandonment lost its exact claim snapshot"))
}
}
pub(crate) fn terminalize_runtime_worker_process_death(
store_path: impl AsRef<Path>,
job_id: &str,
terminal_authority_id: &str,
terminal_at: &str,
) -> Result<EpiphanyRuntimeWorkerProcessClaim> {
let store_path = store_path.as_ref();
validate_non_empty(terminal_authority_id, "worker death terminal authority")?;
chrono::DateTime::parse_from_rfc3339(terminal_at)
.map_err(|error| anyhow!("worker death timestamp is invalid: {error}"))?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let claim_id = worker_process_claim_id(job_id);
let current = cache
.get::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker death recovery requires its process claim"))?;
if current.status == "terminal-death"
&& current.terminal_authority_id.as_deref() == Some(terminal_authority_id)
{
return Ok(current);
}
if !matches!(current.status.as_str(), "claimed" | "active") {
return Err(anyhow!("worker death recovery found terminal process authority"));
}
if cache
.get::<EpiphanyRuntimeRoleWorkerResult>(job_id)?
.is_some()
{
return Err(anyhow!("worker death recovery races a terminal result"));
}
let current_envelope = cache
.get_envelope::<EpiphanyRuntimeWorkerProcessClaim>(&claim_id)?
.ok_or_else(|| anyhow!("worker death recovery lost its claim envelope"))?;
let mut next = current;
next.status = "terminal-death".into();
next.terminal_at = Some(terminal_at.into());
next.terminal_authority_id = Some(terminal_authority_id.into());
let next_envelope = cache.prepare_entry(&claim_id, &next)?.0;
if SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&[current_envelope], vec![next_envelope])?
{
Ok(next)
} else {
Err(anyhow!("worker death recovery lost its exact claim snapshot"))
}
}
pub fn put_runtime_role_worker_result(
store_path: impl AsRef<Path>,
result: &EpiphanyRuntimeRoleWorkerResult,
@ -3853,12 +4109,22 @@ pub fn put_runtime_role_worker_result(
));
}
}
let process_claim_id = worker_process_claim_id(&result.job_id);
let process_claim = cache.get::<EpiphanyRuntimeWorkerProcessClaim>(&process_claim_id)?;
if let Some(existing) = cache.get::<EpiphanyRuntimeRoleWorkerResult>(&result.job_id)? {
if existing != *result {
return Err(anyhow!(
"role worker result is immutable for its runtime job"
));
}
if let Some(claim) = process_claim.as_ref()
&& (claim.status != "terminal-result"
|| claim.terminal_authority_id.as_deref() != Some(result.result_id.as_str()))
{
return Err(anyhow!(
"role worker result is not terminal authority for its process claim"
));
}
if let Some(companion) = model_direction_companion.as_ref() {
return match cache
.get::<crate::AdmittedModelDirectionConsiderationResult>(&companion.result_id)?
@ -3873,6 +4139,23 @@ pub fn put_runtime_role_worker_result(
}
let (envelope, _) = cache.prepare_entry(&result.job_id, result)?;
let mut writes = vec![envelope];
let mut expected = Vec::new();
if let Some(claim) = process_claim.as_ref() {
if claim.status != "active" {
return Err(anyhow!(
"role worker result requires its active process claim"
));
}
let current_envelope = cache
.get_envelope::<EpiphanyRuntimeWorkerProcessClaim>(&process_claim_id)?
.ok_or_else(|| anyhow!("role worker result lost its process claim envelope"))?;
let mut terminal_claim = claim.clone();
terminal_claim.status = "terminal-result".into();
terminal_claim.terminal_at = Some(chrono::Utc::now().to_rfc3339());
terminal_claim.terminal_authority_id = Some(result.result_id.clone());
expected.push(current_envelope);
writes.push(cache.prepare_entry(&process_claim_id, &terminal_claim)?.0);
}
if let Some(companion) = model_direction_companion.as_ref() {
if cache
.get::<crate::AdmittedModelDirectionConsiderationResult>(&companion.result_id)?
@ -3886,7 +4169,7 @@ pub fn put_runtime_role_worker_result(
writes.push(companion_envelope);
}
let backing = SingleFileMessagePackBackingStore::new(store_path);
if backing.compare_and_swap_batch(&[], writes)? {
if backing.compare_and_swap_batch(&expected, writes)? {
return Ok(());
}
let mut reloaded = runtime_spine_cache(store_path)?;
@ -3900,7 +4183,16 @@ pub fn put_runtime_role_worker_result(
.is_some_and(|existing| existing == *companion),
None => true,
};
if worker_matches && companion_matches {
let claim_matches = match process_claim.as_ref() {
Some(_) => reloaded
.get::<EpiphanyRuntimeWorkerProcessClaim>(&process_claim_id)?
.is_some_and(|claim| {
claim.status == "terminal-result"
&& claim.terminal_authority_id.as_deref() == Some(result.result_id.as_str())
}),
None => true,
};
if worker_matches && companion_matches && claim_matches {
Ok(())
} else {
Err(anyhow!(
@ -10155,9 +10447,19 @@ pub fn runtime_typed_request_attempt_exists(
validate_non_empty(request_id, "typed attempt request id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let claims = cache
.get_all::<EpiphanyRuntimeWorkerProcessClaim>()?
.into_iter()
.map(|claim| (claim.job_id.clone(), claim))
.collect::<BTreeMap<_, _>>();
Ok(cache
.get_all::<EpiphanyRuntimeWorkerLaunchRequest>()?
.iter()
.filter(|launch| {
claims.get(&launch.job_id).is_none_or(|claim| {
!matches!(claim.status.as_str(), "terminal-death" | "terminal-unactivated")
})
})
.any(|launch| match request {
RuntimeTypedRequestRef::ProposalModeling(_) => {
launch.proposal_modeling_request_id.as_deref() == Some(request_id)
@ -11305,6 +11607,18 @@ fn epiphany_mutation_contracts() -> Vec<CultNetDocumentMutationContract> {
"Role worker results preserve the typed finding payload; generic runtime job results are lifecycle receipts.",
],
),
mutation_contract(
RUNTIME_WORKER_PROCESS_CLAIM_TYPE,
RUNTIME_WORKER_PROCESS_CLAIM_SCHEMA_VERSION,
vec![CultNetDocumentOperation::Snapshot],
CultNetMutationAuthority::ReadOnly,
vec![],
vec![],
vec![
"The worker claims its native process before model/tool work; the spawning coordinator alone presents the one-use activation preimage.",
"Terminal result and exact process death replace the same claim; absence and age are not terminal authority.",
],
),
mutation_contract(
RUNTIME_REORIENT_WORKER_RESULT_TYPE,
RUNTIME_REORIENT_WORKER_RESULT_SCHEMA_VERSION,
@ -12591,6 +12905,190 @@ pub(crate) mod tests {
Ok(())
}
#[test]
fn worker_process_claim_gates_work_and_terminal_result_replaces_it_atomically() -> Result<()> {
let root = tempdir()?;
let store = root.path().join("worker-process-claim.cc");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "worker-process-runtime".into(),
display_name: "Worker process runtime".into(),
created_at: "2026-08-10T12:00:00Z".into(),
},
)?;
put_test_non_modeling_worker_launch(&store, "worker-process-job", "research")?;
let process = crate::capture_process_instance(std::process::id())?;
let token = "worker-activation-secret";
let token_sha256 = format!("{:x}", Sha256::digest(token.as_bytes()));
claim_runtime_worker_process(
&store,
"worker-process-job",
&process,
&token_sha256,
"2026-08-10T12:00:01Z",
)?;
let result = EpiphanyRuntimeRoleWorkerResult {
schema_version: RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION.into(),
result_id: "worker-process-result".into(),
job_id: "worker-process-job".into(),
role_id: "research".into(),
verdict: "checkpoint-ready".into(),
summary: "Claimed worker completed.".into(),
next_safe_move: "Review evidence.".into(),
checkpoint_summary: None,
scratch_summary: None,
files_inspected: Vec::new(),
frontier_node_ids: Vec::new(),
evidence_ids: Vec::new(),
artifact_refs: Vec::new(),
open_questions: Vec::new(),
evidence_gaps: Vec::new(),
risks: Vec::new(),
state_patch_msgpack: None,
self_patch_msgpack: None,
item_error: None,
metadata: BTreeMap::new(),
repo_model_patch_msgpack: None,
verification_request_id: None,
frontier_route_id: None,
repo_frontier_modeling_request_id: None,
proposal_modeling_request_id: None,
claim_repair_request_id: None,
frontier_planning_request_id: None,
frontier_plan_candidate_msgpack: None,
frontier_plan_mind_request_id: None,
frontier_plan_mind_decision_msgpack: None,
repository_body_observation_basis: 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,
};
assert!(put_runtime_role_worker_result(&store, &result).is_err());
assert!(
activate_runtime_worker_process(
&store,
"worker-process-job",
&process,
"wrong-token",
"2026-08-10T12:00:02Z",
)
.is_err()
);
activate_runtime_worker_process(
&store,
"worker-process-job",
&process,
token,
"2026-08-10T12:00:02Z",
)?;
put_runtime_role_worker_result(&store, &result)?;
let claim = runtime_worker_process_claim(&store, "worker-process-job")?
.expect("terminal process claim");
assert_eq!(claim.status, "terminal-result");
assert_eq!(
claim.terminal_authority_id.as_deref(),
Some("worker-process-result")
);
assert_eq!(runtime_role_worker_result(&store, "worker-process-job")?, Some(result));
Ok(())
}
#[test]
fn worker_process_death_is_single_use_and_excludes_late_result() -> Result<()> {
let root = tempdir()?;
let store = root.path().join("worker-process-death.cc");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "worker-death-runtime".into(),
display_name: "Worker death runtime".into(),
created_at: "2026-08-10T12:10:00Z".into(),
},
)?;
put_test_non_modeling_worker_launch(&store, "worker-death-job", "research")?;
let process = crate::ProcessInstanceIdentity {
process_id: u32::MAX - 1,
creation_token: 77,
created_at_rfc3339: None,
executable_path: PathBuf::from("/release/epiphany-model-runtime"),
};
claim_runtime_worker_process(
&store,
"worker-death-job",
&process,
&"a".repeat(64),
"2026-08-10T12:10:01Z",
)?;
let terminal = terminalize_runtime_worker_process_death(
&store,
"worker-death-job",
"worker-death-recovery-worker-death-job",
"2026-08-10T12:10:02Z",
)?;
assert_eq!(terminal.status, "terminal-death");
assert_eq!(
terminalize_runtime_worker_process_death(
&store,
"worker-death-job",
"worker-death-recovery-worker-death-job",
"2026-08-10T12:10:02Z",
)?,
terminal
);
assert!(
terminalize_runtime_worker_process_death(
&store,
"worker-death-job",
"substituted-authority",
"2026-08-10T12:10:03Z",
)
.is_err()
);
let mut result = EpiphanyRuntimeRoleWorkerResult {
schema_version: RUNTIME_ROLE_WORKER_RESULT_SCHEMA_VERSION.into(),
result_id: "late-result".into(),
job_id: "worker-death-job".into(),
role_id: "research".into(),
verdict: "checkpoint-ready".into(),
summary: "Late result must lose.".into(),
next_safe_move: "None.".into(),
checkpoint_summary: None,
scratch_summary: None,
files_inspected: Vec::new(),
frontier_node_ids: Vec::new(),
evidence_ids: Vec::new(),
artifact_refs: Vec::new(),
open_questions: Vec::new(),
evidence_gaps: Vec::new(),
risks: Vec::new(),
state_patch_msgpack: None,
self_patch_msgpack: None,
item_error: None,
metadata: BTreeMap::new(),
repo_model_patch_msgpack: None,
verification_request_id: None,
frontier_route_id: None,
repo_frontier_modeling_request_id: None,
proposal_modeling_request_id: None,
claim_repair_request_id: None,
frontier_planning_request_id: None,
frontier_plan_candidate_msgpack: None,
frontier_plan_mind_request_id: None,
frontier_plan_mind_decision_msgpack: None,
repository_body_observation_basis: 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,
};
assert!(put_runtime_role_worker_result(&store, &result).is_err());
result.result_id = "other-late-result".into();
assert!(put_runtime_role_worker_result(&store, &result).is_err());
Ok(())
}
#[test]
fn legacy_repo_model_migration_adds_only_exact_projection_pressure() -> Result<()> {
let root = tempdir()?;

View file

@ -150,6 +150,7 @@ async fn main() -> Result<()> {
"run-worker" => {
let options = parse_run_worker_options(args.collect())?;
require_supported_provider(&options.provider)?;
claim_and_wait_for_worker_activation(&options)?;
let timeout_guard = start_run_worker_timeout_watchdog(&options);
let timeout_seconds = options.max_runtime_seconds;
let timeout_store = options.store_path.clone();
@ -526,6 +527,7 @@ struct RunWorkerCliOptions {
cwd: Option<PathBuf>,
max_tool_rounds: usize,
max_runtime_seconds: Option<u64>,
activation_token_sha256: String,
}
struct ToolFollowupCliOptions {
@ -626,6 +628,7 @@ fn parse_run_worker_options(args: Vec<String>) -> Result<RunWorkerCliOptions> {
let mut cwd = None;
let mut max_tool_rounds = 4usize;
let mut max_runtime_seconds = None;
let mut activation_token_sha256 = None;
let mut iter = args.into_iter();
while let Some(arg) = iter.next() {
match arg.as_str() {
@ -648,6 +651,10 @@ fn parse_run_worker_options(args: Vec<String>) -> Result<RunWorkerCliOptions> {
"--max-runtime-seconds" => {
max_runtime_seconds = Some(next_value(&mut iter, "--max-runtime-seconds")?.parse()?)
}
"--activation-token-sha256" => {
activation_token_sha256 =
Some(next_value(&mut iter, "--activation-token-sha256")?)
}
other => return Err(anyhow!("unknown run-worker argument: {other}")),
}
}
@ -663,9 +670,63 @@ fn parse_run_worker_options(args: Vec<String>) -> Result<RunWorkerCliOptions> {
cwd,
max_tool_rounds,
max_runtime_seconds,
activation_token_sha256: activation_token_sha256
.context("run-worker requires --activation-token-sha256")?,
})
}
fn claim_and_wait_for_worker_activation(options: &RunWorkerCliOptions) -> Result<()> {
let process = epiphany_core::capture_process_instance(std::process::id())?;
epiphany_core::claim_runtime_worker_process(
&options.store_path,
&options.job_id,
&process,
&options.activation_token_sha256,
&chrono::Utc::now().to_rfc3339(),
)?;
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
let claim = epiphany_core::runtime_worker_process_claim(
&options.store_path,
&options.job_id,
)?
.ok_or_else(|| anyhow!("worker activation gate lost its process claim"))?;
if claim.process_id != process.process_id
|| claim.process_creation_token != process.creation_token
|| claim.process_executable_path != process.executable_path.display().to_string()
{
return Err(anyhow!("worker activation gate observed a substituted process claim"));
}
match claim.status.as_str() {
"active" => return Ok(()),
"claimed" if std::time::Instant::now() < deadline => {
thread::sleep(Duration::from_millis(25));
}
"claimed" => {
epiphany_core::abandon_unactivated_runtime_worker_process(
&options.store_path,
&options.job_id,
&process,
&chrono::Utc::now().to_rfc3339(),
)?;
fail_worker_job(
&options.store_path,
&options.job_id,
"Worker activation expired before provider/tool work.".into(),
"Runtime Continuity may supersede only from the terminal unactivated process claim."
.into(),
)?;
return Err(anyhow!("worker activation gate expired before model/tool work"));
}
status => {
return Err(anyhow!(
"worker activation gate found non-runnable status {status:?}"
));
}
}
}
}
fn parse_tool_followup_options(args: Vec<String>) -> Result<ToolFollowupCliOptions> {
let mut store_path = PathBuf::from(DEFAULT_STORE);
let mut request_id = None;
@ -1140,6 +1201,33 @@ mod tests {
use epiphany_tool_adapter::EpiphanyToolInvocationIntent;
use tempfile::tempdir;
#[test]
fn worker_cli_requires_activation_capability_before_runtime_path() -> Result<()> {
assert!(
parse_run_worker_options(vec![
"--job-id".into(),
"job-without-gate".into(),
])
.is_err()
);
let options = parse_run_worker_options(vec![
"--job-id".into(),
"job-with-gate".into(),
"--activation-token-sha256".into(),
"a".repeat(64),
])?;
assert_eq!(options.activation_token_sha256, "a".repeat(64));
let source = include_str!("epiphany-openai-runtime.rs");
let gate = source
.find("claim_and_wait_for_worker_activation(&options)?")
.expect("worker main owns activation gate");
let watchdog = source
.find("start_run_worker_timeout_watchdog(&options)")
.expect("worker runtime starts watchdog");
assert!(gate < watchdog);
Ok(())
}
#[test]
fn model_turn_json_ingress_accepts_published_native_object() -> Result<()> {
let request = parse_model_turn_request_json(
@ -1825,5 +1913,5 @@ fn now() -> String {
}
fn usage() -> &'static str {
"usage: epiphany-model-runtime <model-turn|run-worker|tool-followup|tool-followup-turn|smoke> [--provider openai-codex] [--store path] [--codex-home path] [--request path] [--request-id id] [--followup-request-id id] [--output path] [--session-id id] [--job-id id] [--objective text] [--default-model model] [--output-last-message path] [--auto-tools --tool-adapter-bin path --mcp-config path --cwd path --max-tool-rounds n]"
"usage: epiphany-model-runtime <model-turn|run-worker|tool-followup|tool-followup-turn|smoke> [--provider openai-codex] [--store path] [--codex-home path] [--request path] [--request-id id] [--followup-request-id id] [--output path] [--session-id id] [--job-id id] [--activation-token-sha256 hex] [--objective text] [--default-model model] [--output-last-message path] [--auto-tools --tool-adapter-bin path --mcp-config path --cwd path --max-tool-rounds n]"
}

View file

@ -3164,11 +3164,12 @@ three. Exact live replay remains the package acceptance boundary.
# Orphan worker Continuity authority map — 2026-08-10
Current mechanism: the coordinator persists an immutable runtime worker launch,
spawns `epiphany-openai-runtime run-worker` detached, and records the returned
PID only in its artifact JSON. The runtime store has no durable binding between
that launch/job and the worker process incarnation. Coordinator death after
spawn therefore leaves resident Self unable to prove whether the worker may
still publish terminal output.
spawns `epiphany-openai-runtime run-worker` with an activation-capability digest,
then waits for the child-authored exact process claim before presenting the
preimage. The worker cannot enter watchdog, provider, model, or tool paths before
activation. Runtime result admission atomically replaces the active claim with
`terminal-result`; resident Continuity may replace it with `terminal-death` only
after exact native process observation.
- Owner: the worker runtime owns claiming its exact native process incarnation
before model or tool work. Runtime Continuity owns activation, terminalization,
@ -3178,9 +3179,9 @@ still publish terminal output.
explicit activation bound to that claim, native exact-process observation,
and authenticated terminal worker result.
- Outputs: `claimed` before consequence-capable work, `active` only after one
explicit activation, then exactly one `terminal-result` or `terminal-death`.
A terminal-death attempt may be superseded by a new ordinal without deleting
the prior launch family.
explicit activation, then exactly one `terminal-result`, `terminal-death`, or
`terminal-unactivated`. A terminal-death/unactivated attempt may be superseded
by a distinct job/binding incarnation without deleting the prior launch family.
- Derived state: coordinator artifact PID and detached `Child` handle are
diagnostic only. Request identity remains retry-stable; launch, worker claim,
process, session, and artifact directory are attempt-incarnation state.
@ -3192,13 +3193,17 @@ still publish terminal output.
direction must use the same worker start gate and terminal claim primitive.
The worker itself claims before all provider/tool routes; no coordinator-only
happy path may bypass it.
- Cut line: persist a typed worker process claim from inside `run-worker`, gate
work on exact activation, make terminal result close that claim atomically,
and let Continuity write terminal death only after exact native observation.
Then extend each typed launch family with ordinal/supersession bound to that
terminal-death authority. Do not reuse the current unbound artifact PID.
- Cut line: implemented in `runtime_spine.rs`, `epiphany-openai-runtime`, the
coordinator launcher, and resident Self. `roleFailureReview` no longer owns
proposal retry. All three typed lanes consult exact process terminality, and
dead runtime-link projections are demoted before planning a replacement.
The remaining cut is packaged Linux falsification, not another retry policy.
- Verification layer: kill before claim, after claim/before activation, after
activation/before result, and after valid result/before coordinator receipt;
prove no model/tool work before activation, no dual live attempts, same typed
request with distinct attempt/process/artifact identities, exact replay, and
retention non-resurrection across all three typed lanes.
retention non-resurrection across all three typed lanes. Source evidence now
includes exact process/result race tests, activation CLI ordering, typed retry
for proposal/Imagination/model direction, stale-link demotion, canonical Self
settlement, core 660 passed/1 ignored, swarm 9, coordinator 17, and OpenAI
runtime 10. Packaged Linux kill boundaries remain open.

View file

@ -1,25 +1,39 @@
# Fresh workspace handoff
## Orphan-worker ownership mapped — 2026-08-10
## Orphan-worker Continuity source complete — 2026-08-10
The receipt-free pre-open coordinator boundary already has a focused source
test and fails closed on exact runtime-incarnation absence. The remaining
post-open fault is deeper than resident retry policy. Detached worker launch
persists the immutable runtime job/document, but the spawned worker PID exists
only in coordinator artifact JSON. No typed runtime state binds job, launch,
PID, creation token, executable, activation, and terminal state. After
coordinator death, neither resident Self nor Runtime Continuity can prove that
the worker is dead; supersession would permit two workers to publish for one
typed request.
The runtime now persists one `EpiphanyRuntimeWorkerProcessClaim` per immutable
worker job. `run-worker` claims its own PID, creation token, executable, and
activation-capability digest before watchdog, provider, model, or tool work. It
waits for the coordinator to observe that exact claim and present the matching
capability. Failure to activate terminalizes the claim without cognitive work.
An authenticated worker result and `terminal-result` claim replacement share
one snapshot-fenced transaction.
The authority map is now explicit in the algorithmic map. `run-worker` must
claim its own exact process before provider/tool work and wait for an exact
runtime activation. Authenticated result or exact observed process death then
terminalizes that same claim. Only terminal death can ground a new attempt
ordinal. Proposal Modeling, Imagination consideration, and admitted model
direction must share this primitive. Coordinator artifact PID is sight only;
do not promote it or infer death from time/receipt absence. Actual c006 remains
braked. Yggdrasil never builds.
Resident Continuity resolves the request's exact worker launch and claim. Only
native `Missing`, `Replaced`, or exact exit observation may replace a claimed or
active process with `terminal-death`; alive waits, inaccessible/indeterminate
fails closed, and a result race refuses. Self then uses its existing
`process-failed` terminal vocabulary to clear the exact lease, terminalize the
grant, and requeue pressure once. `roleFailureReview`, coordinator receipt
absence, elapsed time, and legacy claimless launches grant no retry authority.
Proposal Modeling, Imagination consideration, and admitted model direction now
share this terminality predicate. Imagination retries receive distinct binding
IDs; every lane receives a distinct backend job. A dead process also demotes
the stale active runtime-link projection so it cannot veto the new launch.
Attempt history and terminal claims remain immutable. Focused user-story tests
prove refusal before terminality and successful retry afterward for all three
lanes, including canonical resident settlement. Full gates pass: core 660
passed/1 ignored, swarm 9/9, coordinator 17/17, OpenAI runtime 10/10.
This cut is not yet committed, packaged, or accepted on Linux. Next commit and
push it, package exactly once with the persistent Starfire cache, authenticate
the 24-binary release and witness, and run the copied-state kill matrix at
pre-claim, claimed/unactivated, activated/pre-result, and valid-result before
coordinator receipt. Prove same typed request with distinct incarnation,
no dual live worker, no pre-activation model/tool work, exact settlement, and
retention non-resurrection. Actual c006 remains braked. Yggdrasil never builds.
## Heartbeat graceful SIGTERM accepted on Linux — 2026-08-10

Binary file not shown.

View file

@ -36,9 +36,9 @@ invariants:
current_status:
summary: Epiphany is a supervised engineering alpha on Starfire, not deployment-ready. Exact b8138a6b active-child SIGTERM and restart, exact 9f369528 deterministic warm packaging, exact c930f505 cross-container ownership, exact ed15f83a resident lifecycle retention, exact 16c4b19f coordinator receipt and Heartbeat artifact retention, exact cce2b8ac narrow model-generation retention, exact eea15e82 model-turn JSON ingress, exact b1e8c89a semantic unattempted-generation retention, exact 6bc3ec3f semantic physical-epoch retirement, and exact 44230a05 Persona conversation retention are proven. Gamecult-ops 059821f owns Yggdrasil stream retention and exact 6ee08a4 owns stable Starfire Persona-permit firewall ingress. Exact 16c4b19f genuine idle and exact a987d2f7 active cognition completed 900-pulse bounded plateaus. Runtime-owned typed fulfillment is accepted live under exact 6dd9b132. Yggdrasil is the small public crossing body, never the release forge.
latest_phase: Exact pushed bfb92cbb packages Heartbeat graceful shutdown as warning-free 24-binary release sha256-6f1f4eff with witness sha256-fda39e34. On copied c006 state under the mounted cognitive brake, continuous Heartbeat emitted no grant, accepted SIGTERM during a 30-second interval, stopped in 0.358 seconds with exit zero/no OOM, and emitted one final typed shutdownRequested receipt. Resident, runtime, and cognitive local-Verse hashes remained byte-identical. Exact c693aaea actual c006 settlement remains accepted; the actual brake stays engaged and old processes stay stopped.
latest_phase: Orphan-worker Continuity is source-complete and awaiting commit/package acceptance. The worker now claims its exact process before model or tool work and cannot proceed until coordinator activation presents the matching capability. Terminal result atomically terminalizes the claim; resident Continuity alone observes exact process death and records terminal-death before canonical requeue. Exact death, not roleFailureReview prose or receipt absence, authorizes distinct retries for proposal Modeling, Imagination consideration, and admitted model direction. Dead runtime-link projections no longer block a valid retry. Full source gates pass: core 660 passed/1 ignored, swarm 9, coordinator 17, OpenAI runtime 10. Actual c006 remains braked and stopped.
operational_state:
local_worktree: heartbeat_graceful_sigterm_accepted_live_pending_mind_commit
local_worktree: worker_process_continuity_source_complete_pending_commit_and_package
operator_wireguard: restored
starfire_persona_permit_firewall: persistent_exact_gamecult_ops_6ee08a4
yggdrasil: canonical_small_public_crossing_body
@ -69,8 +69,8 @@ current_status:
starfire_linux_modeling_peak: 109887488_bytes_resident_body_plus_modeling_circuit
yggdrasil_epiphany_service_contract: repaired_in_gamecult_ops_812b427_not_installed
resident_physiology: c006_exact_c693aaea_settled_braked_no_active_lease_no_attempt_three_old_processes_stopped
unresolved_authority: Coordinator receipt, Heartbeat pulse-artifact, deployed supervisor/projector log retention, genuine-idle and active resource plateaus, exact-grant terminality, restart-persistent cooldown refusal, schedule/action issuance cardinality, runtime-owned typed fulfillment, narrow model-generation retention, published model-turn JSON ingress, semantic logical/physical retention, Persona conversation retention, stable Starfire firewall ownership, completed model-adapter session archival, and normal coordinator lifecycle opening/error closure/legacy repair/archival are accepted live. Receipt-free hard-death recovery is source- and package-accepted but its live kill matrix remains unproven because the first copied run exposed non-atomic storage publication. A typed worker launch without valid terminal fulfillment deliberately remains active awaiting an owner for exact orphan-worker death and supersession; it is not retryable yet. Historical migrated semantic attempts without recorded physical namespace remain durable and non-retirable. Repository Body observation still performs two authenticated full-tree Git index/manifest passes. Execute-mode workers require a persistent process body. Mutable Starfire bind storage is not accepted as resident authority until the packaged CultCache repair and recovery replay pass. One completed Discord consequence remains blocked by the absent Yggdrasil credential.
next_action: Implement the runtime-owned worker start gate mapped in the algorithmic map: child claims exact process identity before model/tools, explicit activation gates work, terminal result or exact observed death terminalizes that claim, and only terminal death may authorize a new typed attempt ordinal. Apply the same primitive to proposal Modeling, Imagination consideration, and admitted model direction. Then package once and run the copied coordinator-death boundary matrix. Keep actual c006 braked. Yggdrasil never builds.
unresolved_authority: Coordinator receipt, Heartbeat pulse-artifact, deployed supervisor/projector log retention, genuine-idle and active resource plateaus, exact-grant terminality, restart-persistent cooldown refusal, schedule/action issuance cardinality, runtime-owned typed fulfillment, narrow model-generation retention, published model-turn JSON ingress, semantic logical/physical retention, Persona conversation retention, stable Starfire firewall ownership, completed model-adapter session archival, and normal coordinator lifecycle opening/error closure/legacy repair/archival are accepted live. Receipt-free coordinator and worker death recovery are source-grounded, but the exact packaged Linux kill matrix is not yet accepted. Historical migrated semantic attempts without recorded physical namespace remain durable and non-retirable. Repository Body observation still performs two authenticated full-tree Git index/manifest passes. Execute-mode workers require a persistent process body. Mutable Starfire bind storage still needs the packaged randomized interruption/read-loop proof before recovered c005 authority may be trusted. One completed Discord consequence remains blocked by the absent Yggdrasil credential.
next_action: Commit and push the source-complete worker Continuity cut, package that exact commit once using the persistent Starfire cache, authenticate all 24 binaries and witness, then run the copied-state death matrix: pre-claim, claimed/unactivated, activated/pre-result, valid-result/pre-coordinator-receipt, and retry with the same typed request but distinct job/process/artifact incarnation across proposal Modeling, Imagination consideration, and admitted model direction. Prove no pre-activation model/tool work, no dual live attempt, exact replay, canonical resident settlement, and retention non-resurrection. Keep actual c006 braked. Yggdrasil never builds.
active_subgoals:
- Prove Persona, Self, Eyes, Modeling, Imagination, Hands, Soul, Mind, Continuity, and nervous-system behavior individually.
@ -87,7 +87,7 @@ organ_readiness:
hands: proven for bounded consequences; native patch, command, commit, push, and typed receipts bind reviewed path ceilings to exact changed-path subsets.
soul: proven for the current bounded route-chain consequence; exact d3bfbda0 launched Verification through typed plan-decision provenance, result f23fe2b6 was accepted at revision 57, and Self then derived Modeling before any repeated Hands authority.
mind: proven for launch-owned resident frontier adjudication; two historical attempts failed closed, exact bd96305e derived immutable identity from authenticated launch, authorized retry 2d2a3e8b completed, and Self atomically committed terminal adopt decision e7d7f3c7. Broader memory retention and long-duration behavior remain open.
continuity: partial; authenticated 55660d78 terminalized the copied v28 dead worker once and refused duplicate restart, while sealed v28 remained unchanged. Exact 100d8854 closes only all-terminal sessions, emits one deterministic event, and refuses post-close jobs. Exact 42bc665a proves bounded retention and restart sequence recovery. A digest-pinned Linux runtime-spine at b5140288 repeated closure invariants on Yggdrasil. Authenticated c005 proves typed resident state survives hard exit, executable-body replacement, runtime-topology correction, and restart while preserving awaitFrontierProposal. Exact b8138a6b proves active-child and idle SIGTERM closure with exit zero and final braked state. Exact c930f505 proves cross-container resident ownership refusal without state mutation while Heartbeat retains its distinct typed lock. Exact ed15f83a packages resident lifecycle retention. Exact 16c4b19f packages coordinator receipt retention: 21 old receipts retired into revision-1 digest sha256:63587221d745315af81c705824d16313d2f748b6ff9840a2b4152127a6758fbd, the newest plus an old receipt named by resident state/pending acknowledgement survived, 720 unrelated runtime envelopes retained digest sha256:c21a12653ce1bf0e64342c6fcbf98d006aeb8fa707de69527886db20e4c20e1c, resident authority counts did not change, and replay did not advance the head or create work. Broader runtime/log retention and the long-duration plateau remain open.
continuity: partial; authenticated 55660d78 terminalized the copied v28 dead worker once and refused duplicate restart, while sealed v28 remained unchanged. Exact 100d8854 closes only all-terminal sessions, emits one deterministic event, and refuses post-close jobs. Exact 42bc665a proves bounded retention and restart sequence recovery. A digest-pinned Linux runtime-spine at b5140288 repeated closure invariants on Yggdrasil. Authenticated c005 proves typed resident state survives hard exit, executable-body replacement, runtime-topology correction, and restart while preserving awaitFrontierProposal. Exact b8138a6b proves active-child and idle SIGTERM closure with exit zero and final braked state. Exact c930f505 proves cross-container resident ownership refusal without state mutation while Heartbeat retains its distinct typed lock. Exact ed15f83a packages resident lifecycle retention. Exact 16c4b19f packages coordinator receipt retention without resurrection. Current source adds exact worker process claim, activation, atomic result terminality, exact-death recovery, typed retry, dead-link demotion, and canonical resident requeue across all three cognition lanes; 660 core, 9 swarm, 17 coordinator, and 10 OpenAI-runtime tests pass. Packaged Linux death boundaries, broader runtime/log retention, and the long-duration plateau remain open.
nervous_system: partial; detached scheduling, runtime jobs, typed feedback, ordinary Windows coverage-store identity, and operator projection work. Exact c0055772 proves Linux heartbeat acknowledgement/regrant, Self single-consumption launch, authenticated plan coordinator completion, and a persistent resident body. It also proves detached model workers cannot survive a one-shot container PID 1 exit. The accepted Linux Eyes-to-Modeling circuit proves detached role workers, native read-only tool loops, typed admission, and causal next-organ routing at a measured 109887488-byte peak. Modeling job commit spent 160183ms in two cold Repository Body scans. A source cut now reuses one private index across both equality scans; 25 focused observer tests pass and a live-worktree mechanism benchmark reduced the pair from 3832ms to 1537ms. Packaged coordinator timing remains open. Authenticated d910158f isolates Self policy behind one root-graph optional coordinator feature; warm policy invalidation is 6.02s. Root Linux packaging is 12m33s cold and 9.08s warm with persistent caches. Long-duration pressure and resource plateau remain open. Yggdrasil is not a release forge at its current memory budget.
forbidden_actions: