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

Starve job bindings of runtime lifecycle

Browse source
This commit is contained in:
Meta Vens 2026-05-12 03:13:12 +01:00
commit ca2ef00b8e
3 changed files with 115 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

66
vendor/codex/codex-rs/app-server-protocol/src/protocol/common.rs vendored
View file

@ -1866,6 +1866,7 @@ mod tests {
"retrieve",
"distill",
"context",
"planning",
"jobs",
"roles",
"coordinator",
@ -2207,7 +2208,7 @@ mod tests {
"ownerRole": "epiphany-modeler",
"launcherJobId": "epiphany-launch-1",
"authorityScope": "epiphany.role.modeling",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "backend-1",
"status": "pending",
"progressNote": "Explicitly launched through the Epiphany authority surface onto the heartbeat backend.",
@ -2302,7 +2303,7 @@ mod tests {
"ownerRole": "epiphany-verifier",
"launcherJobId": "epiphany-launch-2",
"authorityScope": "epiphany.role.verification",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "backend-2",
"status": "completed",
"itemsProcessed": 1,
@ -2376,6 +2377,7 @@ mod tests {
response: v2::ThreadEpiphanyRoleAcceptResponse {
revision: 5,
changed_fields: vec![
v2::ThreadEpiphanyStateUpdatedField::AcceptanceReceipts,
v2::ThreadEpiphanyStateUpdatedField::GraphFrontier,
v2::ThreadEpiphanyStateUpdatedField::Observations,
v2::ThreadEpiphanyStateUpdatedField::Evidence,
@ -2386,6 +2388,7 @@ mod tests {
},
role_id: v2::ThreadEpiphanyRoleId::Modeling,
binding_id: "modeling-checkpoint-worker".to_string(),
accepted_receipt_id: "accept-modeling-1".to_string(),
accepted_observation_id: "obs-modeling-1".to_string(),
accepted_evidence_id: "ev-modeling-1".to_string(),
applied_patch: patch,
@ -2401,12 +2404,13 @@ mod tests {
"id": 10,
"response": {
"revision": 5,
"changedFields": ["graphFrontier", "observations", "evidence"],
"changedFields": ["acceptanceReceipts", "graphFrontier", "observations", "evidence"],
"epiphanyState": {
"revision": 5
},
"roleId": "modeling",
"bindingId": "modeling-checkpoint-worker",
"acceptedReceiptId": "accept-modeling-1",
"acceptedObservationId": "obs-modeling-1",
"acceptedEvidenceId": "ev-modeling-1",
"appliedPatch": {
@ -3296,7 +3300,7 @@ mod tests {
"ownerRole": "epiphany-reorient",
"launcherJobId": "epiphany-launch-1",
"authorityScope": "epiphany.reorient.regather",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "backend-1",
"status": "pending",
"progressNote": "Explicitly launched through the Epiphany authority surface onto the heartbeat backend.",
@ -3383,7 +3387,7 @@ mod tests {
"ownerRole": "epiphany-reorient",
"launcherJobId": "epiphany-launch-1",
"authorityScope": "epiphany.reorient.resume",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "backend-1",
"status": "completed",
"itemsProcessed": 1,
@ -3423,6 +3427,7 @@ mod tests {
response: v2::ThreadEpiphanyReorientAcceptResponse {
revision: 6,
changed_fields: vec![
v2::ThreadEpiphanyStateUpdatedField::AcceptanceReceipts,
v2::ThreadEpiphanyStateUpdatedField::Observations,
v2::ThreadEpiphanyStateUpdatedField::Evidence,
v2::ThreadEpiphanyStateUpdatedField::Scratch,
@ -3447,6 +3452,7 @@ mod tests {
..Default::default()
},
binding_id: "reorient-worker".to_string(),
accepted_receipt_id: "accept-reorient-1".to_string(),
accepted_observation_id: "obs-reorient-1".to_string(),
accepted_evidence_id: "ev-reorient-1".to_string(),
finding: v2::ThreadEpiphanyReorientFinding {
@ -3476,7 +3482,7 @@ mod tests {
"id": 10,
"response": {
"revision": 6,
"changedFields": ["observations", "evidence", "scratch"],
"changedFields": ["acceptanceReceipts", "observations", "evidence", "scratch"],
"epiphanyState": {
"revision": 6,
"observations": [{
@ -3494,6 +3500,7 @@ mod tests {
}]
},
"bindingId": "reorient-worker",
"acceptedReceiptId": "accept-reorient-1",
"acceptedObservationId": "obs-reorient-1",
"acceptedEvidenceId": "ev-reorient-1",
"finding": {
@ -3795,7 +3802,10 @@ mod tests {
request_id: RequestId::Integer(12),
response: v2::ThreadEpiphanyJobLaunchResponse {
revision: 3,
changed_fields: vec![v2::ThreadEpiphanyStateUpdatedField::JobBindings],
changed_fields: vec![
v2::ThreadEpiphanyStateUpdatedField::JobBindings,
v2::ThreadEpiphanyStateUpdatedField::RuntimeLinks,
],
epiphany_state: codex_protocol::protocol::EpiphanyThreadState {
revision: 3,
job_bindings: vec![codex_protocol::protocol::EpiphanyJobBinding {
@ -3803,17 +3813,26 @@ mod tests {
kind: codex_protocol::protocol::EpiphanyJobKind::Specialist,
scope: "role-scoped specialist work".to_string(),
owner_role: "epiphany-harness".to_string(),
launcher_job_id: Some("launcher-specialist".to_string()),
launcher_job_id: None,
authority_scope: Some("epiphany.specialist".to_string()),
backend_kind: Some(
codex_protocol::protocol::EpiphanyJobBackendKind::Heartbeat,
),
backend_job_id: Some("job-123".to_string()),
backend_kind: None,
backend_job_id: None,
linked_subgoal_ids: Vec::new(),
linked_graph_node_ids: Vec::new(),
progress_note: Some("Launched explicitly.".to_string()),
progress_note: None,
blocking_reason: None,
}],
runtime_links: vec![codex_protocol::protocol::EpiphanyRuntimeLink {
id: "runtime-link-specialist-work-job-123".to_string(),
binding_id: "specialist-work".to_string(),
surface: "jobLaunch".to_string(),
role_id: "epiphany-harness".to_string(),
authority_scope: "epiphany.specialist".to_string(),
runtime_job_id: "job-123".to_string(),
runtime_result_id: None,
linked_subgoal_ids: Vec::new(),
linked_graph_node_ids: Vec::new(),
}],
..Default::default()
},
job: v2::ThreadEpiphanyJob {
@ -3821,7 +3840,7 @@ mod tests {
kind: v2::ThreadEpiphanyJobKind::Specialist,
scope: "role-scoped specialist work".to_string(),
owner_role: "epiphany-harness".to_string(),
launcher_job_id: Some("launcher-specialist".to_string()),
launcher_job_id: None,
authority_scope: Some("epiphany.specialist".to_string()),
backend_kind: Some(v2::ThreadEpiphanyJobBackendKind::Heartbeat),
backend_job_id: Some("job-123".to_string()),
@ -3846,7 +3865,7 @@ mod tests {
"id": 12,
"response": {
"revision": 3,
"changedFields": ["jobBindings"],
"changedFields": ["jobBindings", "runtimeLinks"],
"epiphanyState": {
"revision": 3,
"job_bindings": [{
@ -3854,11 +3873,15 @@ mod tests {
"kind": "specialist",
"scope": "role-scoped specialist work",
"owner_role": "epiphany-harness",
"launcher_job_id": "launcher-specialist",
"authority_scope": "epiphany.specialist"
}],
"runtime_links": [{
"id": "runtime-link-specialist-work-job-123",
"binding_id": "specialist-work",
"surface": "jobLaunch",
"role_id": "epiphany-harness",
"authority_scope": "epiphany.specialist",
"backend_kind": "heartbeat",
"backend_job_id": "job-123",
"progress_note": "Launched explicitly."
"runtime_job_id": "job-123"
}]
},
"job": {
@ -3866,9 +3889,8 @@ mod tests {
"kind": "specialist",
"scope": "role-scoped specialist work",
"ownerRole": "epiphany-harness",
"launcherJobId": "launcher-specialist",
"authorityScope": "epiphany.specialist",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "job-123",
"status": "pending",
"progressNote": "Runtime agent job is pending."
@ -4117,7 +4139,7 @@ mod tests {
"ownerRole": "epiphany-harness",
"launcherJobId": "launcher-specialist",
"authorityScope": "epiphany.specialist",
"backendKind": "agentJobs",
"backendKind": "heartbeat",
"backendJobId": "job-123",
"status": "running",
"itemsProcessed": 1,

72
vendor/codex/codex-rs/app-server/src/codex_message_processor.rs vendored
View file

@ -4803,7 +4803,7 @@ impl CodexMessageProcessor {
}
};
let binding_id = launch_request.binding_id.clone();
let changed_fields = vec![ThreadEpiphanyStateUpdatedField::JobBindings];
let changed_fields = epiphany_job_launch_changed_fields();
let launched = match loaded_thread.epiphany_launch_job(launch_request).await {
Ok(launched) => launched,
Err(CodexErr::InvalidRequest(message)) => {
@ -5843,7 +5843,7 @@ impl CodexMessageProcessor {
checkpoint,
&decision,
);
let changed_fields = vec![ThreadEpiphanyStateUpdatedField::JobBindings];
let changed_fields = epiphany_job_launch_changed_fields();
let launched = match loaded_thread.epiphany_launch_job(launch_request).await {
Ok(launched) => launched,
Err(CodexErr::InvalidRequest(message)) => {
@ -6747,7 +6747,7 @@ impl CodexMessageProcessor {
}
};
let changed_fields = vec![ThreadEpiphanyStateUpdatedField::JobBindings];
let changed_fields = epiphany_job_launch_changed_fields();
let launched = match thread
.epiphany_launch_job(EpiphanyJobLaunchRequest {
expected_revision,
@ -17583,7 +17583,7 @@ pub(crate) async fn maybe_run_epiphany_coordinator_automation_for_turn_boundary(
thread_id: thread_id_text,
source: ThreadEpiphanyStateUpdatedSource::JobLaunch,
revision: epiphany_state.revision,
changed_fields: vec![ThreadEpiphanyStateUpdatedField::JobBindings],
changed_fields: epiphany_job_launch_changed_fields(),
epiphany_state,
},
))
@ -17592,6 +17592,13 @@ pub(crate) async fn maybe_run_epiphany_coordinator_automation_for_turn_boundary(
}
}
fn epiphany_job_launch_changed_fields() -> Vec<ThreadEpiphanyStateUpdatedField> {
vec![
ThreadEpiphanyStateUpdatedField::JobBindings,
ThreadEpiphanyStateUpdatedField::RuntimeLinks,
]
}
pub(crate) async fn maybe_run_epiphany_pre_compaction_checkpoint_intervention_for_token_count(
thread_id: ThreadId,
turn_id: String,
@ -17661,10 +17668,11 @@ fn map_epiphany_jobs(
};
for binding in &state.job_bindings {
let runtime_link = latest_epiphany_runtime_link_for_binding(state, binding.id.as_str());
let replacement = if let Some(existing) = jobs.iter().find(|job| job.id == binding.id) {
overlay_epiphany_job_binding(existing.clone(), binding)
overlay_epiphany_job_binding(existing.clone(), binding, runtime_link)
} else {
map_epiphany_bound_job(binding)
map_epiphany_bound_job(binding, runtime_link)
};
if let Some(existing) = jobs.iter_mut().find(|job| job.id == binding.id) {
@ -17677,7 +17685,10 @@ fn map_epiphany_jobs(
jobs
}
fn map_epiphany_bound_job(binding: &EpiphanyJobBinding) -> ThreadEpiphanyJob {
fn map_epiphany_bound_job(
binding: &EpiphanyJobBinding,
runtime_link: Option<&EpiphanyRuntimeLink>,
) -> ThreadEpiphanyJob {
overlay_epiphany_job_binding(
ThreadEpiphanyJob {
id: binding.id.clone(),
@ -17686,8 +17697,12 @@ fn map_epiphany_bound_job(binding: &EpiphanyJobBinding) -> ThreadEpiphanyJob {
owner_role: binding.owner_role.clone(),
launcher_job_id: binding_launcher_job_id(binding).map(str::to_string),
authority_scope: binding.authority_scope.clone(),
backend_kind: binding_backend_kind(binding).map(map_core_epiphany_job_backend_kind),
backend_job_id: binding_projected_backend_job_id(binding).map(str::to_string),
backend_kind: runtime_link
.map(|_| ThreadEpiphanyJobBackendKind::Heartbeat)
.or_else(|| binding_backend_kind(binding).map(map_core_epiphany_job_backend_kind)),
backend_job_id: runtime_link
.map(|link| link.runtime_job_id.clone())
.or_else(|| binding_projected_backend_job_id(binding).map(str::to_string)),
status: if binding.blocking_reason.is_some() {
ThreadEpiphanyJobStatus::Blocked
} else {
@ -17703,20 +17718,26 @@ fn map_epiphany_bound_job(binding: &EpiphanyJobBinding) -> ThreadEpiphanyJob {
linked_graph_node_ids: binding.linked_graph_node_ids.clone(),
},
binding,
runtime_link,
)
}
fn overlay_epiphany_job_binding(
mut job: ThreadEpiphanyJob,
binding: &EpiphanyJobBinding,
runtime_link: Option<&EpiphanyRuntimeLink>,
) -> ThreadEpiphanyJob {
job.kind = map_core_epiphany_job_kind(binding.kind);
job.scope = binding.scope.clone();
job.owner_role = binding.owner_role.clone();
job.launcher_job_id = binding_launcher_job_id(binding).map(str::to_string);
job.authority_scope = binding.authority_scope.clone();
job.backend_kind = binding_backend_kind(binding).map(map_core_epiphany_job_backend_kind);
job.backend_job_id = binding_projected_backend_job_id(binding).map(str::to_string);
job.backend_kind = runtime_link
.map(|_| ThreadEpiphanyJobBackendKind::Heartbeat)
.or_else(|| binding_backend_kind(binding).map(map_core_epiphany_job_backend_kind));
job.backend_job_id = runtime_link
.map(|link| link.runtime_job_id.clone())
.or_else(|| binding_projected_backend_job_id(binding).map(str::to_string));
if !binding.linked_subgoal_ids.is_empty() {
job.linked_subgoal_ids = binding.linked_subgoal_ids.clone();
}
@ -17729,13 +17750,14 @@ fn overlay_epiphany_job_binding(
if let Some(blocking_reason) = binding.blocking_reason.clone() {
job.blocking_reason = Some(blocking_reason);
}
if binding.blocking_reason.is_some() && binding_backend_job_id(binding).is_none() {
if binding.blocking_reason.is_some() {
job.status = ThreadEpiphanyJobStatus::Blocked;
return job;
}
if binding_backend_kind(binding) == Some(CoreEpiphanyJobBackendKind::Heartbeat)
&& binding_backend_job_id(binding).is_some()
if runtime_link.is_some()
|| (binding_backend_kind(binding) == Some(CoreEpiphanyJobBackendKind::Heartbeat)
&& binding_backend_job_id(binding).is_some())
{
job.status = ThreadEpiphanyJobStatus::Pending;
job.blocking_reason = None;
@ -22810,18 +22832,26 @@ mod tests {
kind: codex_protocol::protocol::EpiphanyJobKind::Specialist,
scope: "role-scoped modeling/checkpoint maintenance".to_string(),
owner_role: "epiphany-modeler".to_string(),
launcher_job_id: Some("epiphany-heartbeat-launch-1".to_string()),
launcher_job_id: None,
authority_scope: Some("epiphany.role.modeling".to_string()),
backend_kind: Some(codex_protocol::protocol::EpiphanyJobBackendKind::Heartbeat),
backend_job_id: Some("heartbeat-turn-1".to_string()),
backend_kind: None,
backend_job_id: None,
linked_subgoal_ids: vec!["phase-6".to_string()],
linked_graph_node_ids: vec!["runtime-spine".to_string()],
progress_note: Some(
"Explicitly queued through the Epiphany authority surface for heartbeat activation."
.to_string(),
),
progress_note: None,
blocking_reason: None,
}],
runtime_links: vec![codex_protocol::protocol::EpiphanyRuntimeLink {
id: "runtime-link-modeling-checkpoint-worker-heartbeat-turn-1".to_string(),
binding_id: "modeling-checkpoint-worker".to_string(),
surface: "roleLaunch".to_string(),
role_id: "epiphany-modeler".to_string(),
authority_scope: "epiphany.role.modeling".to_string(),
runtime_job_id: "heartbeat-turn-1".to_string(),
runtime_result_id: None,
linked_subgoal_ids: vec!["phase-6".to_string()],
linked_graph_node_ids: vec!["runtime-spine".to_string()],
}],
..Default::default()
};

69
vendor/codex/codex-rs/core/src/codex_thread.rs vendored
View file

@ -33,6 +33,7 @@ use codex_protocol::protocol::EpiphanyGraphFrontier;
use codex_protocol::protocol::EpiphanyGraphs;
use codex_protocol::protocol::EpiphanyInvariant;
use codex_protocol::protocol::EpiphanyInvestigationCheckpoint;
#[cfg(test)]
use codex_protocol::protocol::EpiphanyJobBackendKind;
use codex_protocol::protocol::EpiphanyJobBinding;
use codex_protocol::protocol::EpiphanyJobKind;
@ -534,16 +535,12 @@ impl CodexThread {
current_state.job_bindings.clone(),
replacement_binding,
);
let next_runtime_links = replace_or_append_epiphany_runtime_link(
current_state.runtime_links.clone(),
replacement_runtime_link,
);
let validation_errors = epiphany_state_update_validation_errors(
&current_state,
&EpiphanyStateUpdate {
job_bindings: Some(next_job_bindings.clone()),
runtime_links: next_runtime_links.clone(),
runtime_links: vec![replacement_runtime_link.clone()],
..Default::default()
},
);
@ -558,7 +555,7 @@ impl CodexThread {
.epiphany_update_state(EpiphanyStateUpdate {
expected_revision: request.expected_revision,
job_bindings: Some(next_job_bindings),
runtime_links: next_runtime_links,
runtime_links: vec![replacement_runtime_link],
..Default::default()
})
.await?;
@ -821,15 +818,15 @@ fn validate_epiphany_job_launch_target(
state: &EpiphanyThreadState,
request: &EpiphanyJobLaunchRequest,
) -> CodexResult<()> {
let Some(existing_binding) = state
let existing_binding = state
.job_bindings
.iter()
.find(|binding| binding.id == request.binding_id)
else {
return Ok(());
};
if binding_backend_job_id_core(existing_binding).is_some()
&& existing_binding.blocking_reason.is_none()
.find(|binding| binding.id == request.binding_id);
let existing_runtime_link = state.runtime_links.iter().find(|link| {
link.binding_id == request.binding_id && !link.runtime_job_id.trim().is_empty()
});
if existing_runtime_link.is_some()
&& existing_binding.is_none_or(|binding| binding.blocking_reason.is_none())
{
return Err(CodexErr::InvalidRequest(format!(
"epiphany job binding {:?} is already bound to an active heartbeat turn; interrupt it before launching a replacement",
@ -841,24 +838,21 @@ fn validate_epiphany_job_launch_target(
fn build_epiphany_job_launch_binding(
request: &EpiphanyJobLaunchRequest,
launcher_job_id: &str,
backend_job_id: &str,
_launcher_job_id: &str,
_backend_job_id: &str,
) -> EpiphanyJobBinding {
EpiphanyJobBinding {
id: request.binding_id.clone(),
kind: request.kind,
scope: request.scope.clone(),
owner_role: request.owner_role.clone(),
launcher_job_id: Some(launcher_job_id.to_string()),
launcher_job_id: None,
authority_scope: Some(request.authority_scope.clone()),
backend_kind: Some(EpiphanyJobBackendKind::Heartbeat),
backend_job_id: Some(backend_job_id.to_string()),
backend_kind: None,
backend_job_id: None,
linked_subgoal_ids: request.linked_subgoal_ids.clone(),
linked_graph_node_ids: request.linked_graph_node_ids.clone(),
progress_note: Some(
"Explicitly queued through the Epiphany authority surface for heartbeat activation."
.to_string(),
),
progress_note: None,
blocking_reason: None,
}
}
@ -959,18 +953,6 @@ fn replace_or_append_epiphany_job_binding(
bindings
}
fn replace_or_append_epiphany_runtime_link(
mut links: Vec<EpiphanyRuntimeLink>,
replacement: EpiphanyRuntimeLink,
) -> Vec<EpiphanyRuntimeLink> {
if let Some(existing) = links.iter_mut().find(|link| link.id == replacement.id) {
*existing = replacement;
return links;
}
links.push(replacement);
links
}
fn clear_epiphany_job_binding_backend(
mut bindings: Vec<EpiphanyJobBinding>,
binding_index: usize,
@ -985,10 +967,6 @@ fn clear_epiphany_job_binding_backend(
bindings
}
fn binding_backend_job_id_core(binding: &EpiphanyJobBinding) -> Option<&str> {
binding.backend_job_id.as_deref()
}
fn epiphany_state_update_validation_errors(
state: &EpiphanyThreadState,
update: &EpiphanyStateUpdate,
@ -2152,17 +2130,10 @@ mod epiphany_update_tests {
build_epiphany_job_launch_binding(&request, "epiphany-heartbeat-launch-1", "turn-1");
let runtime_link = build_epiphany_runtime_link(&request, "turn-1");
assert_eq!(
binding.backend_kind,
Some(EpiphanyJobBackendKind::Heartbeat)
);
assert_eq!(binding.backend_job_id.as_deref(), Some("turn-1"));
assert!(
binding
.progress_note
.as_deref()
.is_some_and(|note| note.contains("heartbeat activation"))
);
assert_eq!(binding.launcher_job_id, None);
assert_eq!(binding.backend_kind, None);
assert_eq!(binding.backend_job_id, None);
assert_eq!(binding.progress_note, None);
assert_eq!(
runtime_link.id,
"runtime-link-modeling-checkpoint-worker-turn-1"