From 8cb067edca3612aafb0ac673a45761fd5ce5e209 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Wed, 10 Jun 2026 12:25:28 -0400 Subject: [PATCH 1/8] Route directed evolution worker results --- .../src/directed_evolution.rs | 236 +++++++++++++++--- .../src/directed_evolution/evidence.rs | 47 ++-- .../src/directed_evolution/tests.rs | 138 ++++++++++ crates/paw-codex-worker/src/worker_types.rs | 2 + 4 files changed, 366 insertions(+), 57 deletions(-) diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index 217c93fc3..bc4d22dbd 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -22,7 +22,7 @@ async fn handle_queued_directed_evolution_work_item( { eliminate_stale_directed_evolution_stage_result(client, config, &work_item, &reason) .await?; - post_directed_evolution_action( + post_paw_orchestration_action( client, config, "WorkItems", @@ -39,8 +39,8 @@ async fn handle_queued_directed_evolution_work_item( return Ok(()); } - let brain_run_id = create_entity(client, config, "BrainRuns", json!({})).await?; - post_directed_evolution_action( + let worker_run_id = create_entity(client, config, "WorkerRuns", json!({})).await?; + post_paw_orchestration_action( client, config, "WorkItems", @@ -52,34 +52,32 @@ async fn handle_queued_directed_evolution_work_item( }), ) .await?; - post_directed_evolution_action( + post_paw_orchestration_action( client, config, - "BrainRuns", - &brain_run_id, - "StartBrainRun", - json!({ - "Role": work_item.role, - "WorkItemId": work_item.id, - "AgentKind": directed_evolution_agent_kind_for_role(&work_item.role), - "Model": directed_evolution_model_for_role(&work_item.role), - "ParentSessionId": env::var("CODEX_SESSION_ID").unwrap_or_default(), - "CorrelationJson": work_item.correlation_json, - }), + "WorkerRuns", + &worker_run_id, + "StartWorkerRun", + directed_evolution_start_worker_run_body( + &work_item, + &config.worker_id, + &worker_run_id, + &env::var("CODEX_SESSION_ID").unwrap_or_default(), + ), ) .await?; - post_directed_evolution_action( + post_paw_orchestration_action( client, config, "WorkItems", &work_item.id, "StartWorkItem", - json!({ "BrainRunId": brain_run_id }), + directed_evolution_start_work_item_body(&worker_run_id), ) .await?; info!( work_item_id = %work_item.id, - brain_run_id = %brain_run_id, + worker_run_id = %worker_run_id, role = %work_item.role, target_entity_type = %work_item.target_entity_type, target_entity_id = %work_item.target_entity_id, @@ -89,22 +87,22 @@ async fn handle_queued_directed_evolution_work_item( match run_directed_evolution_codex_role(client, config, &work_item).await { Ok(output_json) => { let summary = directed_evolution_summary(&work_item, &output_json); - let evidence_artifact_id = record_directed_evolution_brain_evidence( + let evidence_artifact_id = record_directed_evolution_worker_evidence( client, config, &work_item, - &brain_run_id, - "codex_brain_run", + &worker_run_id, + "codex_worker_run", &output_json, &summary, ) .await?; - post_directed_evolution_action( + post_paw_orchestration_action( client, config, - "BrainRuns", - &brain_run_id, - "SucceedBrainRun", + "WorkerRuns", + &worker_run_id, + "SucceedWorkerRun", json!({ "OutputJson": output_json, "EvidenceArtifactId": evidence_artifact_id, @@ -112,7 +110,17 @@ async fn handle_queued_directed_evolution_work_item( }), ) .await?; - post_directed_evolution_action( + let receipt_id = route_directed_evolution_success_receipt( + client, + config, + &work_item, + &worker_run_id, + &output_json, + &evidence_artifact_id, + &summary, + ) + .await?; + post_paw_orchestration_action( client, config, "WorkItems", @@ -127,21 +135,22 @@ async fn handle_queued_directed_evolution_work_item( .await?; info!( work_item_id = %work_item.id, - brain_run_id = %brain_run_id, + worker_run_id = %worker_run_id, role = %work_item.role, evidence_artifact_id = %evidence_artifact_id, - "completed Directed Evolution Codex brain run" + receipt_id = %receipt_id, + "completed Directed Evolution Codex worker run" ); Ok(()) } Err(error) => { let failure_reason = format!("Directed Evolution Codex role failed: {error}"); - let evidence_artifact_id = match record_directed_evolution_brain_evidence( + let evidence_artifact_id = match record_directed_evolution_worker_evidence( client, config, &work_item, - &brain_run_id, - "codex_brain_run_failure", + &worker_run_id, + "codex_worker_run_failure", &serde_json::to_string(&json!({ "status": "failed", "failure_reason": failure_reason, @@ -152,16 +161,16 @@ async fn handle_queued_directed_evolution_work_item( { Ok(id) => id, Err(report_error) => { - warn!(%report_error, work_item_id, brain_run_id, "failed to record Directed Evolution failure evidence"); + warn!(%report_error, work_item_id, worker_run_id, "failed to record Directed Evolution failure evidence"); String::new() } }; - if let Err(report_error) = post_directed_evolution_action( + if let Err(report_error) = post_paw_orchestration_action( client, config, - "BrainRuns", - &brain_run_id, - "FailBrainRun", + "WorkerRuns", + &worker_run_id, + "FailWorkerRun", json!({ "FailureReason": failure_reason, "EvidenceArtifactId": evidence_artifact_id, @@ -169,9 +178,21 @@ async fn handle_queued_directed_evolution_work_item( ) .await { - warn!(%report_error, work_item_id, brain_run_id, "failed to report BrainRun failure"); + warn!(%report_error, work_item_id, worker_run_id, "failed to report WorkerRun failure"); + } + if let Err(report_error) = route_directed_evolution_failure_receipt( + client, + config, + &work_item, + &worker_run_id, + &failure_reason, + &evidence_artifact_id, + ) + .await + { + warn!(%report_error, work_item_id, worker_run_id, "failed to route Directed Evolution failure receipt"); } - post_directed_evolution_action( + post_paw_orchestration_action( client, config, "WorkItems", @@ -185,10 +206,10 @@ async fn handle_queued_directed_evolution_work_item( .await?; warn!( work_item_id = %work_item.id, - brain_run_id = %brain_run_id, + worker_run_id = %worker_run_id, role = %work_item.role, evidence_artifact_id = %evidence_artifact_id, - "failed Directed Evolution Codex brain run" + "failed Directed Evolution Codex worker run" ); Ok(()) } @@ -314,6 +335,121 @@ include!("directed_evolution/human_episode_defaults.rs"); include!("directed_evolution/human_episode_plan.rs"); include!("directed_evolution/human_episode.rs"); +fn directed_evolution_start_worker_run_body( + work_item: &DirectedEvolutionWorkItemState, + worker_id: &str, + worker_run_id: &str, + parent_session_id: &str, +) -> Value { + json!({ + "Role": work_item.role, + "WorkItemId": work_item.id, + "WorkerId": worker_id, + "ProviderId": DIRECTED_EVOLUTION_WORKER_PROVIDER_ID, + "AgentKind": directed_evolution_agent_kind_for_role(&work_item.role), + "Model": directed_evolution_model_for_role(&work_item.role), + "SessionId": worker_run_id, + "ParentSessionId": parent_session_id, + "CorrelationJson": work_item.correlation_json, + }) +} + +fn directed_evolution_start_work_item_body(worker_run_id: &str) -> Value { + json!({ "WorkerRunId": worker_run_id }) +} + +fn directed_evolution_success_receipt_body( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + result_json: &str, + evidence_artifact_id: &str, + summary: &str, +) -> Value { + json!({ + "WorkItemId": work_item.id, + "Role": work_item.role, + "TargetEntityType": work_item.target_entity_type, + "TargetEntityId": work_item.target_entity_id, + "WorkerRunId": worker_run_id, + "ResultJson": result_json, + "EvidenceArtifactId": evidence_artifact_id, + "Summary": summary, + "CorrelationJson": work_item.correlation_json, + }) +} + +fn directed_evolution_failure_receipt_body( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + failure_reason: &str, + evidence_artifact_id: &str, +) -> Value { + json!({ + "WorkItemId": work_item.id, + "Role": work_item.role, + "TargetEntityType": work_item.target_entity_type, + "TargetEntityId": work_item.target_entity_id, + "WorkerRunId": worker_run_id, + "FailureReason": failure_reason, + "EvidenceArtifactId": evidence_artifact_id, + "CorrelationJson": work_item.correlation_json, + }) +} + +async fn route_directed_evolution_success_receipt( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + result_json: &str, + evidence_artifact_id: &str, + summary: &str, +) -> Result { + let receipt_id = create_entity(client, config, "WorkItemReceipts", json!({})).await?; + post_directed_evolution_action( + client, + config, + "WorkItemReceipts", + &receipt_id, + "RouteSucceededWorkItem", + directed_evolution_success_receipt_body( + work_item, + worker_run_id, + result_json, + evidence_artifact_id, + summary, + ), + ) + .await?; + Ok(receipt_id) +} + +async fn route_directed_evolution_failure_receipt( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + failure_reason: &str, + evidence_artifact_id: &str, +) -> Result { + let receipt_id = create_entity(client, config, "WorkItemReceipts", json!({})).await?; + post_directed_evolution_action( + client, + config, + "WorkItemReceipts", + &receipt_id, + "RouteFailedWorkItem", + directed_evolution_failure_receipt_body( + work_item, + worker_run_id, + failure_reason, + evidence_artifact_id, + ), + ) + .await?; + Ok(receipt_id) +} + async fn post_directed_evolution_action( client: &reqwest::Client, config: &Config, @@ -334,6 +470,26 @@ async fn post_directed_evolution_action( .await } +async fn post_paw_orchestration_action( + client: &reqwest::Client, + config: &Config, + entity_set: &str, + entity_id: &str, + action: &str, + body: Value, +) -> Result<()> { + post_entity_action_with_namespace( + client, + config, + entity_set, + entity_id, + PAW_ORCHESTRATION_NAMESPACE, + action, + body, + ) + .await +} + async fn run_directed_evolution_codex_role( client: &reqwest::Client, config: &Config, diff --git a/crates/paw-codex-worker/src/directed_evolution/evidence.rs b/crates/paw-codex-worker/src/directed_evolution/evidence.rs index da8282682..3ee989c81 100644 --- a/crates/paw-codex-worker/src/directed_evolution/evidence.rs +++ b/crates/paw-codex-worker/src/directed_evolution/evidence.rs @@ -1,8 +1,8 @@ -async fn record_directed_evolution_brain_evidence( +async fn record_directed_evolution_worker_evidence( client: &reqwest::Client, config: &Config, work_item: &DirectedEvolutionWorkItemState, - brain_run_id: &str, + worker_run_id: &str, artifact_kind: &str, output_json: &str, summary: &str, @@ -14,17 +14,8 @@ async fn record_directed_evolution_brain_evidence( }); let evidence_id = create_entity(client, config, "EvidenceArtifacts", json!({})).await?; let uri = directed_evolution_evidence_uri(work_item, &output_value); - let correlation = json!({ - "work_item_id": work_item.id, - "brain_run_id": brain_run_id, - "role": work_item.role, - "target_entity_type": work_item.target_entity_type, - "target_entity_id": work_item.target_entity_id, - "context_ref": work_item.context_ref, - "output_schema_ref": work_item.output_schema_ref, - "datadog": directed_evolution_datadog_context(work_item), - "output": output_value, - }); + let correlation = + directed_evolution_evidence_correlation(work_item, worker_run_id, output_value.clone()); let evidence_summary = directed_evolution_first_evidence_scope_summary(&output_value); post_directed_evolution_action( client, @@ -53,15 +44,37 @@ async fn record_directed_evolution_brain_evidence( "EvidenceArtifacts", &evidence_id, "LinkEvidenceArtifact", - json!({ - "TargetEntityType": "BrainRun", - "TargetEntityId": brain_run_id, - }), + directed_evolution_evidence_link_body(worker_run_id), ) .await?; Ok(evidence_id) } +fn directed_evolution_evidence_correlation( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + output_value: Value, +) -> Value { + json!({ + "work_item_id": work_item.id, + "worker_run_id": worker_run_id, + "role": work_item.role, + "target_entity_type": work_item.target_entity_type, + "target_entity_id": work_item.target_entity_id, + "context_ref": work_item.context_ref, + "output_schema_ref": work_item.output_schema_ref, + "datadog": directed_evolution_datadog_context(work_item), + "output": output_value, + }) +} + +fn directed_evolution_evidence_link_body(worker_run_id: &str) -> Value { + json!({ + "TargetEntityType": "WorkerRun", + "TargetEntityId": worker_run_id, + }) +} + #[derive(Default)] struct DirectedEvolutionEvidenceScopeSummary { query: String, diff --git a/crates/paw-codex-worker/src/directed_evolution/tests.rs b/crates/paw-codex-worker/src/directed_evolution/tests.rs index 4c5754250..2622e5892 100644 --- a/crates/paw-codex-worker/src/directed_evolution/tests.rs +++ b/crates/paw-codex-worker/src/directed_evolution/tests.rs @@ -335,6 +335,144 @@ mod directed_evolution_tests { assert!(!directed_evolution_mechanical_evaluator_role("simulated_user")); } + #[test] + fn directed_evolution_worker_run_start_body_uses_worker_contract() { + let work_item = DirectedEvolutionWorkItemState { + id: "wi-observer".to_string(), + status: "Queued".to_string(), + role: "observer".to_string(), + target_entity_type: "Organism".to_string(), + target_entity_id: "organism-agent-answers".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: "{\"batch_id\":\"batch-1\"}".to_string(), + }; + + let body = directed_evolution_start_worker_run_body( + &work_item, + "genesis-local-sim-worker", + "wr-observer", + "parent-session-1", + ); + + assert_eq!(PAW_ORCHESTRATION_NAMESPACE, "Temper.PawOrchestration"); + assert_eq!( + DIRECTED_EVOLUTION_WORKER_PROVIDER_ID, + "local_codex" + ); + assert_eq!(body["Role"], "observer"); + assert_eq!(body["WorkItemId"], "wi-observer"); + assert_eq!(body["WorkerId"], "genesis-local-sim-worker"); + assert_eq!(body["ProviderId"], "local_codex"); + assert_eq!(body["AgentKind"], "codex"); + assert_eq!(body["Model"], "codex-cli"); + assert_eq!(body["SessionId"], "wr-observer"); + assert_eq!(body["ParentSessionId"], "parent-session-1"); + assert_eq!(body["CorrelationJson"], "{\"batch_id\":\"batch-1\"}"); + } + + #[test] + fn directed_evolution_work_item_start_body_uses_worker_run_id() { + let body = directed_evolution_start_work_item_body("wr-1"); + + assert_eq!(body, json!({ "WorkerRunId": "wr-1" })); + assert!(body.get("BrainRunId").is_none()); + } + + #[test] + fn directed_evolution_evidence_correlation_links_worker_run() { + let work_item = DirectedEvolutionWorkItemState { + id: "wi-observer".to_string(), + status: "Running".to_string(), + role: "observer".to_string(), + target_entity_type: "Organism".to_string(), + target_entity_id: "organism-agent-answers".to_string(), + prompt_ref: String::new(), + context_ref: "organism:agent-answers".to_string(), + output_schema_ref: "schema:observer".to_string(), + correlation_json: "{}".to_string(), + }; + let output = json!({ + "summary": "Inventory found enough runtime state and telemetry to suggest one pressure." + }); + + let correlation = + directed_evolution_evidence_correlation(&work_item, "wr-observer", output.clone()); + let link = directed_evolution_evidence_link_body("wr-observer"); + + assert_eq!(correlation["worker_run_id"], "wr-observer"); + assert!(correlation.get("brain_run_id").is_none()); + assert_eq!(correlation["output"], output); + assert_eq!(link, json!({ + "TargetEntityType": "WorkerRun", + "TargetEntityId": "wr-observer", + })); + } + + #[test] + fn directed_evolution_success_receipt_routes_worker_output() { + let work_item = DirectedEvolutionWorkItemState { + id: "wi-observer".to_string(), + status: "Running".to_string(), + role: "observer".to_string(), + target_entity_type: "Organism".to_string(), + target_entity_id: "organism-agent-answers".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: "{\"phase\":\"seed-observation\"}".to_string(), + }; + + let body = directed_evolution_success_receipt_body( + &work_item, + "wr-observer", + "{\"actionable\":true}", + "evidence-1", + "Observer found one direction.", + ); + + assert_eq!(body["WorkItemId"], "wi-observer"); + assert_eq!(body["Role"], "observer"); + assert_eq!(body["TargetEntityType"], "Organism"); + assert_eq!(body["TargetEntityId"], "organism-agent-answers"); + assert_eq!(body["WorkerRunId"], "wr-observer"); + assert_eq!(body["ResultJson"], "{\"actionable\":true}"); + assert_eq!(body["EvidenceArtifactId"], "evidence-1"); + assert_eq!(body["Summary"], "Observer found one direction."); + assert_eq!(body["CorrelationJson"], "{\"phase\":\"seed-observation\"}"); + } + + #[test] + fn directed_evolution_failure_receipt_routes_worker_failure() { + let work_item = DirectedEvolutionWorkItemState { + id: "wi-observer".to_string(), + status: "Running".to_string(), + role: "observer".to_string(), + target_entity_type: "Organism".to_string(), + target_entity_id: "organism-agent-answers".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: "{\"phase\":\"seed-observation\"}".to_string(), + }; + + let body = directed_evolution_failure_receipt_body( + &work_item, + "wr-observer", + "observer failed", + "evidence-failure", + ); + + assert_eq!(body["WorkItemId"], "wi-observer"); + assert_eq!(body["Role"], "observer"); + assert_eq!(body["WorkerRunId"], "wr-observer"); + assert_eq!(body["FailureReason"], "observer failed"); + assert_eq!(body["EvidenceArtifactId"], "evidence-failure"); + assert_eq!(body["CorrelationJson"], "{\"phase\":\"seed-observation\"}"); + assert!(body.get("ResultJson").is_none()); + } + #[test] fn directed_evolution_repo_mapping_accepts_app_ref_prefix() { let previous = env::var_os("DIRECTED_EVOLUTION_ORGANISM_REPOS_JSON"); diff --git a/crates/paw-codex-worker/src/worker_types.rs b/crates/paw-codex-worker/src/worker_types.rs index fd5570b98..6c56d5cb6 100644 --- a/crates/paw-codex-worker/src/worker_types.rs +++ b/crates/paw-codex-worker/src/worker_types.rs @@ -7,6 +7,8 @@ const EVALUATION_START_LABEL: &str = "EvaluationRun.Start"; const EVALUATION_PASS_LABEL: &str = "EvaluationRun.Pass"; const EVALUATION_FAIL_LABEL: &str = "EvaluationRun.Fail"; const DIRECTED_EVOLUTION_NAMESPACE: &str = "Temper.DirectedEvolution"; +const PAW_ORCHESTRATION_NAMESPACE: &str = "Temper.PawOrchestration"; +const DIRECTED_EVOLUTION_WORKER_PROVIDER_ID: &str = "local_codex"; #[derive(Clone, Debug)] struct Config { From 42541f874620da13be9c6d07d8973ee988cbee8d Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 11:28:47 -0400 Subject: [PATCH 2/8] fix(de): best-effort success receipt routing + Running WorkItem boot recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A success-path receipt routing failure previously propagated between SucceedWorkerRun and SucceedWorkItem, stranding the WorkItem in Running (boot backlog only re-claims Queued). Routing is now best-effort on the success path, mirroring the failure path — ResultJson still lands on the WorkItem via SucceedWorkItem. Boot recovery now also fails Running WorkItems claimed by this worker (ClaimedBy-scoped OData filter), failing the attached WorkerRun and routing a failure receipt so the control plane can re-dispatch. Recovery runs only in the pre-loop boot block: in the reconnect loop it would fail work items this process is still executing. Co-Authored-By: Claude Fable 5 --- crates/paw-codex-worker/src/boot_watch.rs | 18 +++- .../src/directed_evolution.rs | 101 +++++++++++++++++- .../src/directed_evolution/tests.rs | 58 ++++++++++ crates/paw-codex-worker/src/main.rs | 4 + crates/paw-codex-worker/src/temper_api.rs | 6 ++ crates/paw-codex-worker/src/worker_types.rs | 2 + 6 files changed, 183 insertions(+), 6 deletions(-) diff --git a/crates/paw-codex-worker/src/boot_watch.rs b/crates/paw-codex-worker/src/boot_watch.rs index a81a1b105..2ee31e5ca 100644 --- a/crates/paw-codex-worker/src/boot_watch.rs +++ b/crates/paw-codex-worker/src/boot_watch.rs @@ -88,19 +88,29 @@ async fn query_boot_entity_ids( config: &Config, entity_set: &str, status: &str, +) -> Result> { + query_boot_entity_ids_filtered(client, config, entity_set, &format!("Status eq '{status}'")) + .await +} + +async fn query_boot_entity_ids_filtered( + client: &reqwest::Client, + config: &Config, + entity_set: &str, + filter: &str, ) -> Result> { let url = format!( - "{}/tdata/{}?$filter=Status eq '{}'&$orderby=Id desc&$top=50", - config.temper_url, entity_set, status + "{}/tdata/{}?$filter={}&$orderby=Id desc&$top=50", + config.temper_url, entity_set, filter ); let response = client .get(url) .headers(headers(config)?) .send() .await - .with_context(|| format!("query {entity_set} with status {status} on boot"))?; + .with_context(|| format!("query {entity_set} with filter {filter} on boot"))?; if !response.status().is_success() { - warn!(entity_set, status, http_status = %response.status(), "boot query failed"); + warn!(entity_set, filter, http_status = %response.status(), "boot query failed"); return Ok(Vec::new()); } diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index bc4d22dbd..1244c2e78 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -110,7 +110,11 @@ async fn handle_queued_directed_evolution_work_item( }), ) .await?; - let receipt_id = route_directed_evolution_success_receipt( + // Receipt routing is best-effort, mirroring the failure path: a + // routing error must not leave the WorkItem stranded in Running + // with a succeeded WorkerRun. ResultJson still lands on the + // WorkItem via SucceedWorkItem below. + let receipt_id = match route_directed_evolution_success_receipt( client, config, &work_item, @@ -119,7 +123,14 @@ async fn handle_queued_directed_evolution_work_item( &evidence_artifact_id, &summary, ) - .await?; + .await + { + Ok(receipt_id) => receipt_id, + Err(report_error) => { + warn!(%report_error, work_item_id, worker_run_id, "failed to route Directed Evolution success receipt"); + String::new() + } + }; post_paw_orchestration_action( client, config, @@ -450,6 +461,92 @@ async fn route_directed_evolution_failure_receipt( Ok(receipt_id) } +fn directed_evolution_running_recovery_filter(worker_id: &str) -> String { + // OData escapes a single quote inside a string literal by doubling it. + let escaped = worker_id.replace('\'', "''"); + format!("Status eq 'Running' and ClaimedBy eq '{escaped}'") +} + +fn directed_evolution_restart_failure_reason(worker_id: &str) -> String { + format!( + "worker {worker_id} restarted while this work item was running; failed for control-plane re-dispatch" + ) +} + +async fn recover_boot_running_directed_evolution_work_items( + client: &reqwest::Client, + config: &Config, +) -> Result<()> { + let filter = directed_evolution_running_recovery_filter(&config.worker_id); + let ids = query_boot_entity_ids_filtered(client, config, "WorkItems", &filter).await?; + for work_item_id in ids { + if let Err(error) = + fail_recovered_directed_evolution_work_item(client, config, &work_item_id).await + { + warn!(%error, work_item_id, "failed to recover Running Directed Evolution WorkItem"); + } + } + Ok(()) +} + +async fn fail_recovered_directed_evolution_work_item( + client: &reqwest::Client, + config: &Config, + work_item_id: &str, +) -> Result<()> { + let work_item = fetch_directed_evolution_work_item(client, config, work_item_id).await?; + if work_item.status != "Running" { + return Ok(()); + } + let failure_reason = directed_evolution_restart_failure_reason(&config.worker_id); + if !work_item.worker_run_id.trim().is_empty() { + if let Err(report_error) = post_paw_orchestration_action( + client, + config, + "WorkerRuns", + &work_item.worker_run_id, + "FailWorkerRun", + json!({ + "FailureReason": failure_reason, + "EvidenceArtifactId": "", + }), + ) + .await + { + warn!(%report_error, work_item_id, worker_run_id = %work_item.worker_run_id, "failed to fail recovered Directed Evolution WorkerRun"); + } + } + if let Err(report_error) = route_directed_evolution_failure_receipt( + client, + config, + &work_item, + &work_item.worker_run_id, + &failure_reason, + "", + ) + .await + { + warn!(%report_error, work_item_id, "failed to route recovered Directed Evolution failure receipt"); + } + post_paw_orchestration_action( + client, + config, + "WorkItems", + &work_item.id, + "FailWorkItem", + json!({ + "FailureReason": failure_reason, + "EvidenceArtifactId": "", + }), + ) + .await?; + warn!( + work_item_id, + "failed Running Directed Evolution WorkItem after worker restart" + ); + Ok(()) +} + async fn post_directed_evolution_action( client: &reqwest::Client, config: &Config, diff --git a/crates/paw-codex-worker/src/directed_evolution/tests.rs b/crates/paw-codex-worker/src/directed_evolution/tests.rs index 2622e5892..e6a771d8b 100644 --- a/crates/paw-codex-worker/src/directed_evolution/tests.rs +++ b/crates/paw-codex-worker/src/directed_evolution/tests.rs @@ -32,6 +32,44 @@ mod directed_evolution_tests { assert_eq!(work_item.correlation_json, "{\"episode_id\":\"episode-1\"}"); } + #[test] + fn directed_evolution_work_item_parses_worker_run_id() { + let value = json!({ + "entity_id": "wi-7", + "status": "Running", + "fields": { + "Role": "simulated_user", + "WorkerRunId": "run-9" + } + }); + + let work_item = + directed_evolution_work_item_from_odata_value(value).expect("WorkItem should parse"); + + assert_eq!(work_item.status, "Running"); + assert_eq!(work_item.worker_run_id, "run-9"); + } + + #[test] + fn running_recovery_filter_scopes_to_this_worker_and_escapes_quotes() { + assert_eq!( + directed_evolution_running_recovery_filter("mac-mini-codex-prod"), + "Status eq 'Running' and ClaimedBy eq 'mac-mini-codex-prod'" + ); + assert_eq!( + directed_evolution_running_recovery_filter("o'brien"), + "Status eq 'Running' and ClaimedBy eq 'o''brien'" + ); + } + + #[test] + fn restart_recovery_failure_reason_names_worker_restart() { + let reason = directed_evolution_restart_failure_reason("mac-mini-codex-prod"); + + assert!(reason.contains("mac-mini-codex-prod")); + assert!(reason.contains("restart")); + } + #[test] fn directed_evolution_prompt_uses_literal_prompt_ref() { let work_item = DirectedEvolutionWorkItemState { @@ -44,6 +82,7 @@ mod directed_evolution_tests { context_ref: "ctx-1".to_string(), output_schema_ref: "schema-1".to_string(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let prompt = directed_evolution_prompt(&work_item); @@ -66,6 +105,7 @@ mod directed_evolution_tests { context_ref: "signal:sig-1".to_string(), output_schema_ref: "schema-1".to_string(), correlation_json: "{\"source\":\"datadog\"}".to_string(), + worker_run_id: String::new(), }; let prompt = directed_evolution_prompt(&work_item); @@ -90,6 +130,7 @@ mod directed_evolution_tests { context_ref: "stage-result:sr-1".to_string(), output_schema_ref: "schema-1".to_string(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let prompt = directed_evolution_prompt(&work_item); @@ -121,6 +162,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let reason = stale_directed_evolution_stage_work_reason( @@ -145,6 +187,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let reason = stale_directed_evolution_stage_work_reason( @@ -169,6 +212,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let reason = stale_directed_evolution_stage_work_reason( @@ -208,6 +252,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; assert!(stale_stage_work_targets_stage_result(&work_item)); @@ -230,6 +275,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let output = directed_evolution_state_verifier_output( @@ -263,6 +309,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let output = directed_evolution_state_verifier_output( @@ -295,6 +342,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let output = directed_evolution_state_verifier_output( @@ -347,6 +395,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{\"batch_id\":\"batch-1\"}".to_string(), + worker_run_id: String::new(), }; let body = directed_evolution_start_worker_run_body( @@ -392,6 +441,7 @@ mod directed_evolution_tests { context_ref: "organism:agent-answers".to_string(), output_schema_ref: "schema:observer".to_string(), correlation_json: "{}".to_string(), + worker_run_id: String::new(), }; let output = json!({ "summary": "Inventory found enough runtime state and telemetry to suggest one pressure." @@ -422,6 +472,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{\"phase\":\"seed-observation\"}".to_string(), + worker_run_id: String::new(), }; let body = directed_evolution_success_receipt_body( @@ -455,6 +506,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: "{\"phase\":\"seed-observation\"}".to_string(), + worker_run_id: String::new(), }; let body = directed_evolution_failure_receipt_body( @@ -523,6 +575,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let tenant = directed_evolution_variant_tenant("de-variant", &work_item); @@ -566,6 +619,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let mut payload = json!({ "summary": "Adds answer evidence confidence.", @@ -661,6 +715,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let uri = directed_evolution_evidence_uri( @@ -685,6 +740,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let uri = directed_evolution_evidence_uri( @@ -728,6 +784,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let summary = directed_evolution_summary( @@ -761,6 +818,7 @@ mod directed_evolution_tests { context_ref: String::new(), output_schema_ref: String::new(), correlation_json: String::new(), + worker_run_id: String::new(), }; let context = directed_evolution_datadog_context(&work_item); diff --git a/crates/paw-codex-worker/src/main.rs b/crates/paw-codex-worker/src/main.rs index fb9e36735..4f9388b18 100644 --- a/crates/paw-codex-worker/src/main.rs +++ b/crates/paw-codex-worker/src/main.rs @@ -70,6 +70,10 @@ async fn main() -> Result<()> { if config.poll_on_start { recover_boot_running_runs(&client, &config).await?; + // Fail-based recovery must run only here, before any work starts: + // in the reconnect loop below it would fail work items this + // process is still executing. + recover_boot_running_directed_evolution_work_items(&client, &config).await?; claim_boot_queued_runs(&client, &config).await?; claim_boot_requested_review_runs(&client, &config).await?; claim_boot_queued_evaluation_runs(&client, &config).await?; diff --git a/crates/paw-codex-worker/src/temper_api.rs b/crates/paw-codex-worker/src/temper_api.rs index 81d12e40a..96fd828b2 100644 --- a/crates/paw-codex-worker/src/temper_api.rs +++ b/crates/paw-codex-worker/src/temper_api.rs @@ -563,6 +563,12 @@ fn directed_evolution_work_item_from_odata_value( &["correlation_json", "CorrelationJson"], &["correlation_json", "CorrelationJson"], ), + worker_run_id: first_string( + &value, + &fields, + &["worker_run_id", "WorkerRunId"], + &["worker_run_id", "WorkerRunId"], + ), }) } diff --git a/crates/paw-codex-worker/src/worker_types.rs b/crates/paw-codex-worker/src/worker_types.rs index 6c56d5cb6..158c09a16 100644 --- a/crates/paw-codex-worker/src/worker_types.rs +++ b/crates/paw-codex-worker/src/worker_types.rs @@ -277,6 +277,8 @@ struct DirectedEvolutionWorkItemState { output_schema_ref: String, #[serde(default, rename = "CorrelationJson")] correlation_json: String, + #[serde(default, rename = "WorkerRunId")] + worker_run_id: String, } #[derive(Clone, Debug, PartialEq, Eq)] From b56a13748c3f94ffff6cf9aa409c7eeaabb22833 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 16:31:47 -0400 Subject: [PATCH 3/8] feat(de): send ADR-0041 observe metadata on worker calls, probes, and Codex children The temper kernel (ADR-0041, nerdsane/temper#300) parses the X-Temper-Observe-Metadata header into temper.observation.* span attributes, but the worker sent nothing. Resolve the Directed Evolution join fields from a WorkItem's CorrelationJson once (DirectedEvolutionJoinFields) and reuse them to: - attach the header (producer.work_item_id, producer.worker_run_id, de.role, de.* join fields; kernel limits: 32 keys, 96-byte keys, 1024-byte values) to every worker->Temper action and entity create made on behalf of a DE work item - attach the header to observer runtime OData probes - inject TEMPER_OBSERVE_METADATA and DD_TAGS (de.* datadog tags, appended to any inherited DD_TAGS) into DE Codex child processes mechanically, not just via prompt text The unused create_entity wrapper is removed; all DE creates now go through create_entity_with_observe_metadata. Co-Authored-By: Claude Fable 5 --- .../paw-codex-worker/fixtures/fake-codex.sh | 4 + crates/paw-codex-worker/src/codex_plan.rs | 1 + .../src/directed_evolution.rs | 108 +++- .../src/directed_evolution/evidence.rs | 13 +- .../directed_evolution/observe_metadata.rs | 181 +++++++ .../observe_metadata_tests.rs | 471 ++++++++++++++++++ .../directed_evolution/observer_sources.rs | 14 +- crates/paw-codex-worker/src/execution.rs | 24 +- crates/paw-codex-worker/src/http_headers.rs | 14 + crates/paw-codex-worker/src/temper_api.rs | 24 +- 10 files changed, 834 insertions(+), 20 deletions(-) create mode 100644 crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs create mode 100644 crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs diff --git a/crates/paw-codex-worker/fixtures/fake-codex.sh b/crates/paw-codex-worker/fixtures/fake-codex.sh index ff44ef894..8e5f32d53 100755 --- a/crates/paw-codex-worker/fixtures/fake-codex.sh +++ b/crates/paw-codex-worker/fixtures/fake-codex.sh @@ -70,6 +70,10 @@ case "$prompt" in "PAW_CODEX_DOCTOR_EXEC_SMOKE:"*) echo "PAW_CODEX_DOCTOR_EXEC_OK" ;; + "PAW_FAKE_CODEX_PRINT_OBSERVE_ENV:"*) + printf 'TEMPER_OBSERVE_METADATA=%s\n' "${TEMPER_OBSERVE_METADATA:-}" + printf 'DD_TAGS=%s\n' "${DD_TAGS:-}" + ;; "You are the independent reviewer"* | "You are the independent repo-health Patrol scan reviewer"*) echo "SUMMARY: Fake reviewer approved the agent-led worker E2E output." echo "LIVE_E2E: Confirmed the fake implementer marker exists in the assigned worktree." diff --git a/crates/paw-codex-worker/src/codex_plan.rs b/crates/paw-codex-worker/src/codex_plan.rs index c92eb7d71..f9d4088f2 100644 --- a/crates/paw-codex-worker/src/codex_plan.rs +++ b/crates/paw-codex-worker/src/codex_plan.rs @@ -46,6 +46,7 @@ async fn run_codex_plan_mode( workdir, codex_plan_args(workdir, &prompt), "run local codex plan mode", + &[], ) .await?; let stdout = String::from_utf8_lossy(&output.stdout); diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index 1244c2e78..82c134b15 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -17,11 +17,20 @@ async fn handle_queued_directed_evolution_work_item( debug!(work_item_id, "Directed Evolution WorkItem has no brain role yet"); return Ok(()); } + let join_fields = directed_evolution_join_fields(&work_item.correlation_json); + let observe_metadata_pre_run = + directed_evolution_work_item_observe_metadata(&work_item, "", &join_fields); if let Some(reason) = stale_directed_evolution_work_item_reason(client, config, &work_item).await? { - eliminate_stale_directed_evolution_stage_result(client, config, &work_item, &reason) - .await?; + eliminate_stale_directed_evolution_stage_result( + client, + config, + &work_item, + &reason, + Some(&observe_metadata_pre_run), + ) + .await?; post_paw_orchestration_action( client, config, @@ -29,6 +38,7 @@ async fn handle_queued_directed_evolution_work_item( &work_item.id, "CancelWorkItem", json!({ "Reason": reason }), + Some(&observe_metadata_pre_run), ) .await?; info!( @@ -39,7 +49,16 @@ async fn handle_queued_directed_evolution_work_item( return Ok(()); } - let worker_run_id = create_entity(client, config, "WorkerRuns", json!({})).await?; + let worker_run_id = create_entity_with_observe_metadata( + client, + config, + "WorkerRuns", + json!({}), + Some(&observe_metadata_pre_run), + ) + .await?; + let observe_metadata = + directed_evolution_work_item_observe_metadata(&work_item, &worker_run_id, &join_fields); post_paw_orchestration_action( client, config, @@ -50,6 +69,7 @@ async fn handle_queued_directed_evolution_work_item( "WorkerId": config.worker_id, "ClaimedBy": config.worker_id, }), + Some(&observe_metadata), ) .await?; post_paw_orchestration_action( @@ -64,6 +84,7 @@ async fn handle_queued_directed_evolution_work_item( &worker_run_id, &env::var("CODEX_SESSION_ID").unwrap_or_default(), ), + Some(&observe_metadata), ) .await?; post_paw_orchestration_action( @@ -73,6 +94,7 @@ async fn handle_queued_directed_evolution_work_item( &work_item.id, "StartWorkItem", directed_evolution_start_work_item_body(&worker_run_id), + Some(&observe_metadata), ) .await?; info!( @@ -84,7 +106,7 @@ async fn handle_queued_directed_evolution_work_item( "started Directed Evolution worker run" ); - match run_directed_evolution_codex_role(client, config, &work_item).await { + match run_directed_evolution_codex_role(client, config, &work_item, &worker_run_id).await { Ok(output_json) => { let summary = directed_evolution_summary(&work_item, &output_json); let evidence_artifact_id = record_directed_evolution_worker_evidence( @@ -95,6 +117,7 @@ async fn handle_queued_directed_evolution_work_item( "codex_worker_run", &output_json, &summary, + Some(&observe_metadata), ) .await?; post_paw_orchestration_action( @@ -108,6 +131,7 @@ async fn handle_queued_directed_evolution_work_item( "EvidenceArtifactId": evidence_artifact_id, "Summary": summary, }), + Some(&observe_metadata), ) .await?; // Receipt routing is best-effort, mirroring the failure path: a @@ -122,6 +146,7 @@ async fn handle_queued_directed_evolution_work_item( &output_json, &evidence_artifact_id, &summary, + Some(&observe_metadata), ) .await { @@ -142,6 +167,7 @@ async fn handle_queued_directed_evolution_work_item( "EvidenceArtifactId": evidence_artifact_id, "Summary": summary, }), + Some(&observe_metadata), ) .await?; info!( @@ -167,6 +193,7 @@ async fn handle_queued_directed_evolution_work_item( "failure_reason": failure_reason, }))?, &failure_reason, + Some(&observe_metadata), ) .await { @@ -186,6 +213,7 @@ async fn handle_queued_directed_evolution_work_item( "FailureReason": failure_reason, "EvidenceArtifactId": evidence_artifact_id, }), + Some(&observe_metadata), ) .await { @@ -198,6 +226,7 @@ async fn handle_queued_directed_evolution_work_item( &worker_run_id, &failure_reason, &evidence_artifact_id, + Some(&observe_metadata), ) .await { @@ -213,6 +242,7 @@ async fn handle_queued_directed_evolution_work_item( "FailureReason": failure_reason, "EvidenceArtifactId": evidence_artifact_id, }), + Some(&observe_metadata), ) .await?; warn!( @@ -261,6 +291,7 @@ async fn eliminate_stale_directed_evolution_stage_result( config: &Config, work_item: &DirectedEvolutionWorkItemState, reason: &str, + observe_metadata: Option<&str>, ) -> Result<()> { if !stale_stage_work_targets_stage_result(work_item) { return Ok(()); @@ -284,6 +315,7 @@ async fn eliminate_stale_directed_evolution_stage_result( "EvidenceArtifactId": value_field_string(&stage_fields, &["EvidenceArtifactId", "evidence_artifact_id"]), "Reason": reason, }), + observe_metadata, ) .await } @@ -340,6 +372,7 @@ fn stale_stage_result_should_eliminate(stage_fields: &Value) -> bool { } +include!("directed_evolution/observe_metadata.rs"); include!("directed_evolution/evidence.rs"); include!("directed_evolution/observer_sources.rs"); include!("directed_evolution/human_episode_defaults.rs"); @@ -407,6 +440,7 @@ fn directed_evolution_failure_receipt_body( }) } +#[allow(clippy::too_many_arguments)] async fn route_directed_evolution_success_receipt( client: &reqwest::Client, config: &Config, @@ -415,8 +449,16 @@ async fn route_directed_evolution_success_receipt( result_json: &str, evidence_artifact_id: &str, summary: &str, + observe_metadata: Option<&str>, ) -> Result { - let receipt_id = create_entity(client, config, "WorkItemReceipts", json!({})).await?; + let receipt_id = create_entity_with_observe_metadata( + client, + config, + "WorkItemReceipts", + json!({}), + observe_metadata, + ) + .await?; post_directed_evolution_action( client, config, @@ -430,6 +472,7 @@ async fn route_directed_evolution_success_receipt( evidence_artifact_id, summary, ), + observe_metadata, ) .await?; Ok(receipt_id) @@ -442,8 +485,16 @@ async fn route_directed_evolution_failure_receipt( worker_run_id: &str, failure_reason: &str, evidence_artifact_id: &str, + observe_metadata: Option<&str>, ) -> Result { - let receipt_id = create_entity(client, config, "WorkItemReceipts", json!({})).await?; + let receipt_id = create_entity_with_observe_metadata( + client, + config, + "WorkItemReceipts", + json!({}), + observe_metadata, + ) + .await?; post_directed_evolution_action( client, config, @@ -456,6 +507,7 @@ async fn route_directed_evolution_failure_receipt( failure_reason, evidence_artifact_id, ), + observe_metadata, ) .await?; Ok(receipt_id) @@ -498,6 +550,12 @@ async fn fail_recovered_directed_evolution_work_item( if work_item.status != "Running" { return Ok(()); } + let join_fields = directed_evolution_join_fields(&work_item.correlation_json); + let observe_metadata = directed_evolution_work_item_observe_metadata( + &work_item, + &work_item.worker_run_id, + &join_fields, + ); let failure_reason = directed_evolution_restart_failure_reason(&config.worker_id); if !work_item.worker_run_id.trim().is_empty() { if let Err(report_error) = post_paw_orchestration_action( @@ -510,6 +568,7 @@ async fn fail_recovered_directed_evolution_work_item( "FailureReason": failure_reason, "EvidenceArtifactId": "", }), + Some(&observe_metadata), ) .await { @@ -523,6 +582,7 @@ async fn fail_recovered_directed_evolution_work_item( &work_item.worker_run_id, &failure_reason, "", + Some(&observe_metadata), ) .await { @@ -538,6 +598,7 @@ async fn fail_recovered_directed_evolution_work_item( "FailureReason": failure_reason, "EvidenceArtifactId": "", }), + Some(&observe_metadata), ) .await?; warn!( @@ -554,8 +615,9 @@ async fn post_directed_evolution_action( entity_id: &str, action: &str, body: Value, + observe_metadata: Option<&str>, ) -> Result<()> { - post_entity_action_with_namespace( + post_entity_action_with_namespace_observed( client, config, entity_set, @@ -563,6 +625,7 @@ async fn post_directed_evolution_action( DIRECTED_EVOLUTION_NAMESPACE, action, body, + observe_metadata, ) .await } @@ -574,8 +637,9 @@ async fn post_paw_orchestration_action( entity_id: &str, action: &str, body: Value, + observe_metadata: Option<&str>, ) -> Result<()> { - post_entity_action_with_namespace( + post_entity_action_with_namespace_observed( client, config, entity_set, @@ -583,6 +647,7 @@ async fn post_paw_orchestration_action( PAW_ORCHESTRATION_NAMESPACE, action, body, + observe_metadata, ) .await } @@ -591,7 +656,9 @@ async fn run_directed_evolution_codex_role( client: &reqwest::Client, config: &Config, work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, ) -> Result { + let join_fields = directed_evolution_join_fields(&work_item.correlation_json); let mut prompt = directed_evolution_prompt(work_item); info!( work_item_id = %work_item.id, @@ -628,9 +695,16 @@ async fn run_directed_evolution_codex_role( let workdir = resolve_directed_evolution_workdir(client, config, work_item).await?; if work_item.role == "observer" { - let inventory = - directed_evolution_observer_source_inventory_prompt(client, config, work_item, &workdir) - .await?; + let observe_metadata = + directed_evolution_work_item_observe_metadata(work_item, worker_run_id, &join_fields); + let inventory = directed_evolution_observer_source_inventory_prompt( + client, + config, + work_item, + &workdir, + &observe_metadata, + ) + .await?; prompt.push_str( "\n\nObserver source inventory:\n\ The following JSON is a source map, not a script. Use it to orient yourself, then inspect any \ @@ -649,11 +723,20 @@ additional available source that could confirm, contradict, or refine the observ } else { directed_evolution_git_status_snapshot(&workdir.path).await? }; - let output = match run_codex_exec_command( + // ADR-0041: the Codex child inherits the correlation context mechanically + // (TEMPER_OBSERVE_METADATA + DD_TAGS), not just via prompt text. + let child_env = directed_evolution_codex_child_env( + work_item, + worker_run_id, + &join_fields, + env::var("DD_TAGS").ok().as_deref(), + ); + let output = match run_codex_exec_command_with_env( config, &workdir.path, prompt, "run Directed Evolution Codex role", + &child_env, ) .await { @@ -784,3 +867,4 @@ include!("directed_evolution/mechanical_evaluator.rs"); include!("directed_evolution/prompt.rs"); include!("directed_evolution/tests.rs"); include!("directed_evolution/prompt_tests.rs"); +include!("directed_evolution/observe_metadata_tests.rs"); diff --git a/crates/paw-codex-worker/src/directed_evolution/evidence.rs b/crates/paw-codex-worker/src/directed_evolution/evidence.rs index 3ee989c81..8cf3d5921 100644 --- a/crates/paw-codex-worker/src/directed_evolution/evidence.rs +++ b/crates/paw-codex-worker/src/directed_evolution/evidence.rs @@ -1,3 +1,4 @@ +#[allow(clippy::too_many_arguments)] async fn record_directed_evolution_worker_evidence( client: &reqwest::Client, config: &Config, @@ -6,13 +7,21 @@ async fn record_directed_evolution_worker_evidence( artifact_kind: &str, output_json: &str, summary: &str, + observe_metadata: Option<&str>, ) -> Result { let output_value = serde_json::from_str::(output_json).unwrap_or_else(|_| { json!({ "raw": output_json, }) }); - let evidence_id = create_entity(client, config, "EvidenceArtifacts", json!({})).await?; + let evidence_id = create_entity_with_observe_metadata( + client, + config, + "EvidenceArtifacts", + json!({}), + observe_metadata, + ) + .await?; let uri = directed_evolution_evidence_uri(work_item, &output_value); let correlation = directed_evolution_evidence_correlation(work_item, worker_run_id, output_value.clone()); @@ -36,6 +45,7 @@ async fn record_directed_evolution_worker_evidence( "ZeroResultMeaning": evidence_summary.zero_result_meaning, "EvidenceProvenance": directed_evolution_evidence_provenance(&work_item.role, &output_value), }), + observe_metadata, ) .await?; post_directed_evolution_action( @@ -45,6 +55,7 @@ async fn record_directed_evolution_worker_evidence( &evidence_id, "LinkEvidenceArtifact", directed_evolution_evidence_link_body(worker_run_id), + observe_metadata, ) .await?; Ok(evidence_id) diff --git a/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs b/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs new file mode 100644 index 000000000..c3ebb6028 --- /dev/null +++ b/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs @@ -0,0 +1,181 @@ +// ADR-0041 observability metadata: the temper kernel parses the +// X-Temper-Observe-Metadata header (a flat JSON object) into +// temper.observation.* span attributes. The worker resolves the Directed +// Evolution join fields from a WorkItem's CorrelationJson exactly once here +// and reuses them for the kernel header, the Datadog context, the prompt +// header block, and the Codex child environment. + +const TEMPER_OBSERVE_METADATA_HEADER: &str = "x-temper-observe-metadata"; +const TEMPER_OBSERVE_METADATA_MAX_KEYS: usize = 32; +const TEMPER_OBSERVE_METADATA_MAX_KEY_BYTES: usize = 96; +const TEMPER_OBSERVE_METADATA_MAX_VALUE_BYTES: usize = 1_024; + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct DirectedEvolutionJoinFields { + episode_id: String, + direction_id: String, + generation_id: String, + variant_id: String, + evaluation_stage_id: String, + stage_result_id: String, + trial_id: String, + simulated_user_plan_id: String, + persona_index: String, + run_index: String, + runtime_ref: String, + app_ref: String, + runtime_tenant: String, +} + +impl DirectedEvolutionJoinFields { + fn entries(&self) -> Vec<(&'static str, &str)> { + [ + ("episode_id", self.episode_id.as_str()), + ("direction_id", self.direction_id.as_str()), + ("generation_id", self.generation_id.as_str()), + ("variant_id", self.variant_id.as_str()), + ("evaluation_stage_id", self.evaluation_stage_id.as_str()), + ("stage_result_id", self.stage_result_id.as_str()), + ("trial_id", self.trial_id.as_str()), + ( + "simulated_user_plan_id", + self.simulated_user_plan_id.as_str(), + ), + ("persona_index", self.persona_index.as_str()), + ("run_index", self.run_index.as_str()), + ("runtime_ref", self.runtime_ref.as_str()), + ("app_ref", self.app_ref.as_str()), + ("runtime_tenant", self.runtime_tenant.as_str()), + ] + .into_iter() + .filter(|(_, value)| !value.trim().is_empty()) + .collect() + } +} + +fn directed_evolution_join_fields(correlation_json: &str) -> DirectedEvolutionJoinFields { + let correlation = + serde_json::from_str::(correlation_json).unwrap_or_else(|_| json!({})); + let field = |key: &str| { + directed_evolution_correlation_string(&correlation, key).unwrap_or_default() + }; + DirectedEvolutionJoinFields { + episode_id: field("episode_id"), + direction_id: field("direction_id"), + generation_id: field("generation_id"), + variant_id: field("variant_id"), + evaluation_stage_id: field("evaluation_stage_id"), + stage_result_id: field("stage_result_id"), + trial_id: field("trial_id"), + simulated_user_plan_id: field("simulated_user_plan_id"), + persona_index: field("persona_index"), + run_index: field("run_index"), + runtime_ref: field("runtime_ref"), + app_ref: field("app_ref"), + runtime_tenant: field("runtime_tenant"), + } +} + +fn directed_evolution_observe_metadata_value( + work_item_id: &str, + worker_run_id: &str, + role: &str, + join: &DirectedEvolutionJoinFields, +) -> String { + let mut pairs: Vec<(String, &str)> = vec![ + ("producer.work_item_id".to_string(), work_item_id), + ("producer.worker_run_id".to_string(), worker_run_id), + ("de.role".to_string(), role), + ]; + for (key, value) in join.entries() { + pairs.push((format!("de.{key}"), value)); + } + + let mut object = serde_json::Map::new(); + for (key, value) in pairs { + if object.len() >= TEMPER_OBSERVE_METADATA_MAX_KEYS { + break; + } + let value = value.trim(); + if value.is_empty() || key.len() > TEMPER_OBSERVE_METADATA_MAX_KEY_BYTES { + continue; + } + object.insert( + key, + json!(truncate_utf8_bytes( + value, + TEMPER_OBSERVE_METADATA_MAX_VALUE_BYTES + )), + ); + } + Value::Object(object).to_string() +} + +fn directed_evolution_work_item_observe_metadata( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + join: &DirectedEvolutionJoinFields, +) -> String { + directed_evolution_observe_metadata_value(&work_item.id, worker_run_id, &work_item.role, join) +} + +fn directed_evolution_dd_tags( + existing: Option<&str>, + role: &str, + join: &DirectedEvolutionJoinFields, +) -> String { + let mut tags: Vec = existing + .map(|value| { + value + .split(',') + .map(str::trim) + .filter(|tag| !tag.is_empty()) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(); + if !role.trim().is_empty() { + tags.push(format!("de.role:{}", sanitize_datadog_tag_value(role))); + } + for (key, value) in join.entries() { + tags.push(format!("de.{key}:{}", sanitize_datadog_tag_value(value))); + } + tags.join(",") +} + +fn sanitize_datadog_tag_value(value: &str) -> String { + value + .trim() + .chars() + .map(|ch| if ch.is_whitespace() || ch == ',' { '_' } else { ch }) + .collect() +} + +fn directed_evolution_codex_child_env( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + join: &DirectedEvolutionJoinFields, + existing_dd_tags: Option<&str>, +) -> Vec<(String, String)> { + vec![ + ( + "TEMPER_OBSERVE_METADATA".to_string(), + directed_evolution_work_item_observe_metadata(work_item, worker_run_id, join), + ), + ( + "DD_TAGS".to_string(), + directed_evolution_dd_tags(existing_dd_tags, &work_item.role, join), + ), + ] +} + +fn truncate_utf8_bytes(value: &str, max_bytes: usize) -> String { + if value.len() <= max_bytes { + return value.to_string(); + } + let mut end = max_bytes; + while end > 0 && !value.is_char_boundary(end) { + end -= 1; + } + value[..end].to_string() +} diff --git a/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs b/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs new file mode 100644 index 000000000..6cce04503 --- /dev/null +++ b/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs @@ -0,0 +1,471 @@ +#[cfg(test)] +mod directed_evolution_observe_metadata_tests { + use super::*; + use std::fs; + use std::time::{SystemTime, UNIX_EPOCH}; + use tokio::io::{AsyncReadExt as ObserveAsyncReadExt, AsyncWriteExt as ObserveAsyncWriteExt}; + + const FULL_CORRELATION: &str = r#"{ + "episode_id": "ep-1", + "direction_id": "dir-1", + "generation_id": "gen-1", + "variant_id": "var-1", + "evaluation_stage_id": "stage-1", + "stage_result_id": "sr-1", + "trial_id": "trial-1", + "simulated_user_plan_id": "plan-1", + "persona_index": 2, + "run_index": 1, + "runtime_ref": "temper://tenant/de-variant-1/app/nerdsane/agent-answers@abc123", + "app_ref": "nerdsane/agent-answers@abc123", + "runtime_tenant": "de-variant-1", + "runtime_base_url": "https://temper.example", + "unrelated_key": "ignored" + }"#; + + fn observe_test_work_item(role: &str, correlation_json: &str) -> DirectedEvolutionWorkItemState { + DirectedEvolutionWorkItemState { + id: "wi-1".to_string(), + status: "Running".to_string(), + role: role.to_string(), + target_entity_type: "Trial".to_string(), + target_entity_id: "trial-1".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: correlation_json.to_string(), + worker_run_id: String::new(), + } + } + + #[test] + fn join_fields_resolve_all_known_correlation_keys_once() { + let join = directed_evolution_join_fields(FULL_CORRELATION); + + assert_eq!(join.episode_id, "ep-1"); + assert_eq!(join.direction_id, "dir-1"); + assert_eq!(join.generation_id, "gen-1"); + assert_eq!(join.variant_id, "var-1"); + assert_eq!(join.evaluation_stage_id, "stage-1"); + assert_eq!(join.stage_result_id, "sr-1"); + assert_eq!(join.trial_id, "trial-1"); + assert_eq!(join.simulated_user_plan_id, "plan-1"); + assert_eq!(join.persona_index, "2"); + assert_eq!(join.run_index, "1"); + assert_eq!( + join.runtime_ref, + "temper://tenant/de-variant-1/app/nerdsane/agent-answers@abc123" + ); + assert_eq!(join.app_ref, "nerdsane/agent-answers@abc123"); + assert_eq!(join.runtime_tenant, "de-variant-1"); + } + + #[test] + fn join_fields_tolerate_invalid_correlation_json() { + let join = directed_evolution_join_fields("not-json"); + + assert_eq!(join, DirectedEvolutionJoinFields::default()); + assert!(join.entries().is_empty()); + } + + #[test] + fn observe_metadata_value_carries_producer_and_de_join_keys() { + let join = directed_evolution_join_fields(FULL_CORRELATION); + + let value = + directed_evolution_observe_metadata_value("wi-1", "wr-1", "simulated_user", &join); + let parsed: Value = serde_json::from_str(&value).expect("header value should be JSON"); + + assert_eq!(parsed["producer.work_item_id"], "wi-1"); + assert_eq!(parsed["producer.worker_run_id"], "wr-1"); + assert_eq!(parsed["de.role"], "simulated_user"); + assert_eq!(parsed["de.episode_id"], "ep-1"); + assert_eq!(parsed["de.direction_id"], "dir-1"); + assert_eq!(parsed["de.generation_id"], "gen-1"); + assert_eq!(parsed["de.variant_id"], "var-1"); + assert_eq!(parsed["de.evaluation_stage_id"], "stage-1"); + assert_eq!(parsed["de.stage_result_id"], "sr-1"); + assert_eq!(parsed["de.trial_id"], "trial-1"); + assert_eq!(parsed["de.simulated_user_plan_id"], "plan-1"); + assert_eq!(parsed["de.persona_index"], "2"); + assert_eq!(parsed["de.run_index"], "1"); + assert_eq!( + parsed["de.runtime_ref"], + "temper://tenant/de-variant-1/app/nerdsane/agent-answers@abc123" + ); + assert_eq!(parsed["de.app_ref"], "nerdsane/agent-answers@abc123"); + assert_eq!(parsed["de.runtime_tenant"], "de-variant-1"); + assert!(parsed.get("de.unrelated_key").is_none()); + assert!(parsed.get("de.runtime_base_url").is_none()); + assert!(!value.contains('\n'), "header values must be single-line"); + } + + #[test] + fn observe_metadata_value_skips_empty_fields() { + let join = directed_evolution_join_fields(r#"{"episode_id":"ep-1"}"#); + + let value = directed_evolution_observe_metadata_value("wi-1", "", "observer", &join); + let parsed: Value = serde_json::from_str(&value).expect("header value should be JSON"); + + assert_eq!(parsed["producer.work_item_id"], "wi-1"); + assert!(parsed.get("producer.worker_run_id").is_none()); + assert_eq!(parsed["de.role"], "observer"); + assert_eq!(parsed["de.episode_id"], "ep-1"); + assert!(parsed.get("de.variant_id").is_none()); + assert!(parsed.get("de.runtime_tenant").is_none()); + } + + #[test] + fn observe_metadata_value_truncates_values_to_kernel_limit() { + let long = "x".repeat(5_000); + let join = directed_evolution_join_fields(&format!(r#"{{"episode_id":"{long}"}}"#)); + + let value = directed_evolution_observe_metadata_value("wi-1", "wr-1", "observer", &join); + let parsed: Value = serde_json::from_str(&value).expect("header value should be JSON"); + + let episode = parsed["de.episode_id"].as_str().expect("episode id string"); + assert_eq!(episode.len(), 1_024); + assert!(episode.chars().all(|ch| ch == 'x')); + } + + #[test] + fn observe_metadata_value_caps_total_keys_at_kernel_limit() { + let join = directed_evolution_join_fields(FULL_CORRELATION); + + let value = + directed_evolution_observe_metadata_value("wi-1", "wr-1", "simulated_user", &join); + let parsed: Value = serde_json::from_str(&value).expect("header value should be JSON"); + + assert!(parsed.as_object().expect("object").len() <= 32); + } + + #[test] + fn dd_tags_join_de_pairs_and_preserve_existing_tags() { + let join = directed_evolution_join_fields(FULL_CORRELATION); + + let tags = directed_evolution_dd_tags( + Some("env:prod,service:temperpaw"), + "simulated_user", + &join, + ); + + assert!(tags.starts_with("env:prod,service:temperpaw,de.role:simulated_user")); + assert!(tags.contains("de.episode_id:ep-1")); + assert!(tags.contains("de.trial_id:trial-1")); + assert!(tags.contains("de.persona_index:2")); + assert!(tags.contains("de.runtime_tenant:de-variant-1")); + } + + #[test] + fn dd_tags_sanitize_values_for_datadog() { + let join = directed_evolution_join_fields(r#"{"episode_id":"ep 1,with comma"}"#); + + let tags = directed_evolution_dd_tags(None, "observer", &join); + + assert_eq!(tags, "de.role:observer,de.episode_id:ep_1_with_comma"); + } + + #[test] + fn codex_child_env_injects_observe_metadata_and_dd_tags() { + let work_item = observe_test_work_item("simulated_user", FULL_CORRELATION); + let join = directed_evolution_join_fields(&work_item.correlation_json); + + let env = directed_evolution_codex_child_env(&work_item, "wr-1", &join, None); + + let observe = env + .iter() + .find(|(key, _)| key == "TEMPER_OBSERVE_METADATA") + .map(|(_, value)| value.clone()) + .expect("child env should carry TEMPER_OBSERVE_METADATA"); + assert_eq!( + observe, + directed_evolution_observe_metadata_value("wi-1", "wr-1", "simulated_user", &join) + ); + let dd_tags = env + .iter() + .find(|(key, _)| key == "DD_TAGS") + .map(|(_, value)| value.clone()) + .expect("child env should carry DD_TAGS"); + assert!(dd_tags.contains("de.role:simulated_user")); + assert!(dd_tags.contains("de.episode_id:ep-1")); + } + + #[test] + fn codex_child_env_appends_to_existing_dd_tags() { + let work_item = observe_test_work_item("simulated_user", FULL_CORRELATION); + let join = directed_evolution_join_fields(&work_item.correlation_json); + + let env = + directed_evolution_codex_child_env(&work_item, "wr-1", &join, Some("env:prod")); + + let dd_tags = env + .iter() + .find(|(key, _)| key == "DD_TAGS") + .map(|(_, value)| value.clone()) + .expect("child env should carry DD_TAGS"); + assert!(dd_tags.starts_with("env:prod,de.role:simulated_user")); + } + + fn observe_test_config(temper_url: String) -> Config { + Config { + temper_url, + tenant: "default".to_string(), + worker_id: "mac-mini-codex-1".to_string(), + worker_token: Some("secret".to_string()), + workspace_root: PathBuf::from("/tmp/worktrees"), + repo_root: PathBuf::from("/tmp/temperpaw"), + codex_bin: "codex".to_string(), + max_concurrent_runs: 1, + enable_execution: false, + poll_on_start: true, + codex_exec_smoke: false, + codex_exec_timeout: Duration::from_secs(30), + } + } + + /// One-shot HTTP server that records the raw request text and replies + /// with a fixed JSON body. + async fn capture_one_request( + response_body: &'static str, + ) -> (String, tokio::task::JoinHandle) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind capture server"); + let addr = listener.local_addr().expect("capture server addr"); + let handle = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept request"); + let mut buffer = Vec::new(); + let mut chunk = [0_u8; 1024]; + let header_end = loop { + let read = ObserveAsyncReadExt::read(&mut stream, &mut chunk) + .await + .expect("read request"); + if read == 0 { + break None; + } + buffer.extend_from_slice(&chunk[..read]); + if let Some(pos) = buffer.windows(4).position(|window| window == b"\r\n\r\n") { + break Some(pos); + } + }; + if let Some(header_end) = header_end { + let header_text = String::from_utf8_lossy(&buffer[..header_end]).to_string(); + let content_length = header_text + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + if name.eq_ignore_ascii_case("content-length") { + value.trim().parse::().ok() + } else { + None + } + }) + .unwrap_or(0); + let body_start = header_end + 4; + while buffer.len() < body_start + content_length { + let read = ObserveAsyncReadExt::read(&mut stream, &mut chunk) + .await + .expect("read request body"); + if read == 0 { + break; + } + buffer.extend_from_slice(&chunk[..read]); + } + } + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + response_body.len(), + response_body + ); + ObserveAsyncWriteExt::write_all(&mut stream, response.as_bytes()) + .await + .expect("write response"); + let _ = ObserveAsyncWriteExt::shutdown(&mut stream).await; + String::from_utf8_lossy(&buffer).to_string() + }); + (format!("http://{addr}"), handle) + } + + fn observe_metadata_header_line(raw_request: &str) -> Option { + raw_request + .lines() + .find(|line| { + line.to_ascii_lowercase() + .starts_with("x-temper-observe-metadata:") + }) + .map(str::to_string) + } + + #[tokio::test] + async fn directed_evolution_action_posts_carry_observe_metadata_header() { + let (url, capture) = capture_one_request("{}").await; + let config = observe_test_config(url); + let client = reqwest::Client::new(); + let join = directed_evolution_join_fields(r#"{"episode_id":"ep-1"}"#); + let metadata = + directed_evolution_observe_metadata_value("wi-1", "wr-1", "observer", &join); + + post_directed_evolution_action( + &client, + &config, + "StageResults", + "sr-1", + "EliminateStageResult", + json!({}), + Some(&metadata), + ) + .await + .expect("post action"); + + let raw = capture.await.expect("captured request"); + let header = observe_metadata_header_line(&raw) + .expect("DE action posts should carry x-temper-observe-metadata"); + assert!(header.contains("producer.work_item_id")); + assert!(header.contains("de.episode_id")); + } + + #[tokio::test] + async fn paw_orchestration_posts_carry_observe_metadata_header() { + let (url, capture) = capture_one_request("{}").await; + let config = observe_test_config(url); + let client = reqwest::Client::new(); + let join = directed_evolution_join_fields(r#"{"trial_id":"trial-1"}"#); + let metadata = + directed_evolution_observe_metadata_value("wi-1", "wr-1", "simulated_user", &join); + + post_paw_orchestration_action( + &client, + &config, + "WorkItems", + "wi-1", + "StartWorkItem", + json!({ "WorkerRunId": "wr-1" }), + Some(&metadata), + ) + .await + .expect("post action"); + + let raw = capture.await.expect("captured request"); + let header = observe_metadata_header_line(&raw) + .expect("orchestration posts should carry x-temper-observe-metadata"); + assert!(header.contains("de.trial_id")); + assert!(header.contains("producer.worker_run_id")); + } + + #[tokio::test] + async fn directed_evolution_entity_creates_carry_observe_metadata_header() { + let (url, capture) = capture_one_request(r#"{"entity_id":"e-1"}"#).await; + let config = observe_test_config(url); + let client = reqwest::Client::new(); + let join = directed_evolution_join_fields(r#"{"episode_id":"ep-1"}"#); + let metadata = + directed_evolution_observe_metadata_value("wi-1", "", "observer", &join); + + let id = create_entity_with_observe_metadata( + &client, + &config, + "WorkerRuns", + json!({}), + Some(&metadata), + ) + .await + .expect("create entity"); + + assert_eq!(id, "e-1"); + let raw = capture.await.expect("captured request"); + let header = observe_metadata_header_line(&raw) + .expect("DE entity creates should carry x-temper-observe-metadata"); + assert!(header.contains("producer.work_item_id")); + } + + #[tokio::test] + async fn observer_runtime_probe_sends_observe_metadata_header() { + let (url, capture) = capture_one_request("{}").await; + let client = reqwest::Client::new(); + let join = directed_evolution_join_fields(r#"{"runtime_tenant":"de-variant-1"}"#); + let metadata = directed_evolution_observe_metadata_value("wi-1", "", "observer", &join); + + let result = observer_runtime_get_text( + &client, + &format!("{url}/tdata/$metadata"), + "de-variant-1", + None, + Some(&metadata), + ) + .await; + + assert_eq!(result["status"], "available"); + let raw = capture.await.expect("captured request"); + let header = observe_metadata_header_line(&raw) + .expect("observer runtime probes should carry x-temper-observe-metadata"); + assert!(header.contains("producer.work_item_id")); + assert!(header.contains("de.role")); + assert!(header.contains("de.runtime_tenant")); + } + + #[tokio::test] + async fn codex_child_process_receives_observe_env_mechanically() { + let fixture = Path::new(env!("CARGO_MANIFEST_DIR")).join("fixtures/fake-codex.sh"); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("time") + .as_nanos(); + let root = env::temp_dir().join(format!( + "paw-codex-worker-observe-env-{}-{nanos}", + std::process::id() + )); + fs::create_dir_all(&root).expect("temp dir"); + let mut config = observe_test_config("http://127.0.0.1:3497".to_string()); + config.codex_bin = fixture.display().to_string(); + let work_item = observe_test_work_item("simulated_user", FULL_CORRELATION); + let join = directed_evolution_join_fields(&work_item.correlation_json); + let child_env = directed_evolution_codex_child_env(&work_item, "wr-1", &join, None); + + let output = run_codex_exec_command_with_env( + &config, + &root, + "PAW_FAKE_CODEX_PRINT_OBSERVE_ENV: report".to_string(), + "observe env fixture", + &child_env, + ) + .await + .expect("run fixture codex"); + + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains(r#""producer.work_item_id":"wi-1""#), + "child should see TEMPER_OBSERVE_METADATA: {stdout}" + ); + assert!( + stdout.contains("de.episode_id:ep-1"), + "child should see DD_TAGS: {stdout}" + ); + fs::remove_dir_all(root).ok(); + } + + #[test] + fn headers_with_observe_metadata_attach_kernel_header() { + let config = observe_test_config("http://127.0.0.1:3497".to_string()); + let join = directed_evolution_join_fields(r#"{"episode_id":"ep-1"}"#); + let value = directed_evolution_observe_metadata_value("wi-1", "wr-1", "observer", &join); + + let with_header = + headers_with_observe_metadata(&config, Some(&value)).expect("headers should build"); + let without_header = + headers_with_observe_metadata(&config, None).expect("headers should build"); + + assert_eq!( + with_header + .get("x-temper-observe-metadata") + .and_then(|header| header.to_str().ok()), + Some(value.as_str()) + ); + assert_eq!( + with_header + .get("x-temper-principal-id") + .and_then(|header| header.to_str().ok()), + Some("mac-mini-codex-1") + ); + assert!(!without_header.contains_key("x-temper-observe-metadata")); + } +} diff --git a/crates/paw-codex-worker/src/directed_evolution/observer_sources.rs b/crates/paw-codex-worker/src/directed_evolution/observer_sources.rs index 5999cbcfb..adc07b482 100644 --- a/crates/paw-codex-worker/src/directed_evolution/observer_sources.rs +++ b/crates/paw-codex-worker/src/directed_evolution/observer_sources.rs @@ -3,6 +3,7 @@ async fn directed_evolution_observer_source_inventory_prompt( config: &Config, work_item: &DirectedEvolutionWorkItemState, workdir: &DirectedEvolutionWorkdir, + observe_metadata: &str, ) -> Result { let correlation = serde_json::from_str::(&work_item.correlation_json) .unwrap_or_else(|_| json!({})); @@ -33,7 +34,7 @@ async fn directed_evolution_observer_source_inventory_prompt( "correlation": correlation, "sources": { "genesis_control_plane": directed_evolution_observer_genesis_sources(client, config, work_item, &correlation).await, - "runtime_odata_state": directed_evolution_observer_runtime_sources(client, &correlation).await, + "runtime_odata_state": directed_evolution_observer_runtime_sources(client, &correlation, observe_metadata).await, "datadog_observability": directed_evolution_observer_datadog_sources(client, work_item, &correlation).await, "app_source_or_description": directed_evolution_observer_app_sources(workdir, &correlation), }, @@ -237,6 +238,7 @@ fn directed_evolution_observer_fields_sample(entity_set: &str, id: &str, fields: async fn directed_evolution_observer_runtime_sources( client: &reqwest::Client, correlation: &Value, + observe_metadata: &str, ) -> Value { let Some(runtime_base) = observer_string(correlation, "runtime_base_url") .or_else(|| observer_string(correlation, "runtimeBaseUrl")) @@ -257,6 +259,7 @@ async fn directed_evolution_observer_runtime_sources( &format!("{runtime_base}/tdata/$metadata"), &tenant, auth_token.as_deref(), + Some(observe_metadata), ) .await; let entity_sets = metadata @@ -274,6 +277,7 @@ async fn directed_evolution_observer_runtime_sources( &tenant, auth_token.as_deref(), &entity_set, + Some(observe_metadata), ) .await, ); @@ -294,6 +298,7 @@ async fn observer_runtime_get_text( url: &str, tenant: &str, auth_token: Option<&str>, + observe_metadata: Option<&str>, ) -> Value { let mut request = client.get(url).header(ACCEPT, "application/json"); if !tenant.trim().is_empty() { @@ -302,6 +307,9 @@ async fn observer_runtime_get_text( if let Some(token) = auth_token.filter(|value| !value.trim().is_empty()) { request = request.header(AUTHORIZATION, format!("Bearer {token}")); } + if let Some(metadata) = observe_metadata.filter(|value| !value.trim().is_empty()) { + request = request.header(TEMPER_OBSERVE_METADATA_HEADER, metadata); + } match request.send().await { Ok(response) => { let status = response.status(); @@ -325,9 +333,11 @@ async fn observer_runtime_collection_sample( tenant: &str, auth_token: Option<&str>, entity_set: &str, + observe_metadata: Option<&str>, ) -> Value { let url = format!("{runtime_base}/tdata/{entity_set}?$top=12"); - let result = observer_runtime_get_text(client, &url, tenant, auth_token).await; + let result = + observer_runtime_get_text(client, &url, tenant, auth_token, observe_metadata).await; let samples = result .get("body_preview") .and_then(Value::as_str) diff --git a/crates/paw-codex-worker/src/execution.rs b/crates/paw-codex-worker/src/execution.rs index 00289cc88..395c3ab8d 100644 --- a/crates/paw-codex-worker/src/execution.rs +++ b/crates/paw-codex-worker/src/execution.rs @@ -374,8 +374,24 @@ async fn run_codex_exec_command( prompt: String, context_label: &str, ) -> Result { - run_codex_exec_command_with_args(config, workdir, codex_exec_args(workdir, &prompt), context_label) - .await + run_codex_exec_command_with_env(config, workdir, prompt, context_label, &[]).await +} + +async fn run_codex_exec_command_with_env( + config: &Config, + workdir: &Path, + prompt: String, + context_label: &str, + extra_env: &[(String, String)], +) -> Result { + run_codex_exec_command_with_args( + config, + workdir, + codex_exec_args(workdir, &prompt), + context_label, + extra_env, + ) + .await } async fn run_codex_exec_command_with_args( @@ -383,6 +399,7 @@ async fn run_codex_exec_command_with_args( workdir: &Path, args: Vec, context_label: &str, + extra_env: &[(String, String)], ) -> Result { let mut command = Command::new(&config.codex_bin); command @@ -391,6 +408,9 @@ async fn run_codex_exec_command_with_args( .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true); + for (key, value) in extra_env { + command.env(key, value); + } configure_process_group(&mut command); let mut child = command diff --git a/crates/paw-codex-worker/src/http_headers.rs b/crates/paw-codex-worker/src/http_headers.rs index d38d4416b..0da919045 100644 --- a/crates/paw-codex-worker/src/http_headers.rs +++ b/crates/paw-codex-worker/src/http_headers.rs @@ -35,6 +35,20 @@ fn headers(config: &Config) -> Result { Ok(headers) } +fn headers_with_observe_metadata( + config: &Config, + observe_metadata: Option<&str>, +) -> Result { + let mut headers = headers(config)?; + if let Some(value) = observe_metadata.filter(|value| !value.trim().is_empty()) { + headers.insert( + TEMPER_OBSERVE_METADATA_HEADER, + HeaderValue::from_str(value).context("invalid X-Temper-Observe-Metadata value")?, + ); + } + Ok(headers) +} + fn event_stream_headers(config: &Config) -> Result { let mut headers = HeaderMap::new(); headers.insert( diff --git a/crates/paw-codex-worker/src/temper_api.rs b/crates/paw-codex-worker/src/temper_api.rs index 96fd828b2..e99be819d 100644 --- a/crates/paw-codex-worker/src/temper_api.rs +++ b/crates/paw-codex-worker/src/temper_api.rs @@ -313,12 +313,29 @@ async fn post_entity_action_with_namespace( namespace: &str, action: &str, body: Value, +) -> Result<()> { + post_entity_action_with_namespace_observed( + client, config, entity_set, entity_id, namespace, action, body, None, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn post_entity_action_with_namespace_observed( + client: &reqwest::Client, + config: &Config, + entity_set: &str, + entity_id: &str, + namespace: &str, + action: &str, + body: Value, + observe_metadata: Option<&str>, ) -> Result<()> { let response = client .post(config.entity_action_url_with_namespace( entity_set, entity_id, namespace, action, )) - .headers(headers(config)?) + .headers(headers_with_observe_metadata(config, observe_metadata)?) .header(CONTENT_TYPE, "application/json") .json(&body) .send() @@ -332,15 +349,16 @@ async fn post_entity_action_with_namespace( Ok(()) } -async fn create_entity( +async fn create_entity_with_observe_metadata( client: &reqwest::Client, config: &Config, entity_set: &str, body: Value, + observe_metadata: Option<&str>, ) -> Result { let response = client .post(format!("{}/tdata/{}", config.temper_url, entity_set)) - .headers(headers(config)?) + .headers(headers_with_observe_metadata(config, observe_metadata)?) .header(CONTENT_TYPE, "application/json") .json(&body) .send() From 0a93201f6d9d4e740a5e61b828a9c8cbd2193497 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 16:35:06 -0400 Subject: [PATCH 4/8] feat(de): resolve first-class join fields into Datadog context and prompt header directed_evolution_datadog_context previously carried only service/env/work_item/role/target/control_tenant; the episode, direction, generation, variant, stage, stage-result, trial, simulated-user-plan, persona/run index, app_ref, runtime_ref, and runtime_tenant joins were buried in raw CorrelationJson. Resolve them through the shared DirectedEvolutionJoinFields parser (one parse, same source as the ADR-0041 header) and surface them in: - the Datadog context recorded on evidence correlation - a ResolvedCorrelation block (de.* lines) in the DE prompt header Co-Authored-By: Claude Fable 5 --- .../src/directed_evolution/evidence.rs | 11 ++- .../observe_metadata_tests.rs | 70 +++++++++++++++++++ .../src/directed_evolution/prompt.rs | 18 +++++ 3 files changed, 97 insertions(+), 2 deletions(-) diff --git a/crates/paw-codex-worker/src/directed_evolution/evidence.rs b/crates/paw-codex-worker/src/directed_evolution/evidence.rs index 8cf3d5921..65a349eee 100644 --- a/crates/paw-codex-worker/src/directed_evolution/evidence.rs +++ b/crates/paw-codex-worker/src/directed_evolution/evidence.rs @@ -239,7 +239,7 @@ fn directed_evolution_datadog_context(work_item: &DirectedEvolutionWorkItemState let env_name = env::var("DD_ENV").unwrap_or_else(|_| "local".to_string()); let site = env::var("DD_SITE").unwrap_or_else(|_| "datadoghq.com".to_string()); let query = format!("service:{service} env:{env_name} @work_item_id:{}", work_item.id); - json!({ + let mut context = json!({ "service": service, "env": env_name, "work_item_id": work_item.id, @@ -252,7 +252,14 @@ fn directed_evolution_datadog_context(work_item: &DirectedEvolutionWorkItemState "https://app.{site}/logs?query={}", encode_url_component(&query) ), - }) + }); + let join = directed_evolution_join_fields(&work_item.correlation_json); + if let Some(object) = context.as_object_mut() { + for (key, value) in join.entries() { + object.insert(key.to_string(), json!(value)); + } + } + context } fn config_tenant_label() -> String { diff --git a/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs b/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs index 6cce04503..96f4b5723 100644 --- a/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs +++ b/crates/paw-codex-worker/src/directed_evolution/observe_metadata_tests.rs @@ -443,6 +443,76 @@ mod directed_evolution_observe_metadata_tests { fs::remove_dir_all(root).ok(); } + #[test] + fn datadog_context_resolves_full_join_field_set() { + let work_item = observe_test_work_item("simulated_user", FULL_CORRELATION); + + let context = directed_evolution_datadog_context(&work_item); + + assert_eq!(context["work_item_id"], "wi-1"); + assert_eq!(context["role"], "simulated_user"); + assert_eq!(context["episode_id"], "ep-1"); + assert_eq!(context["direction_id"], "dir-1"); + assert_eq!(context["generation_id"], "gen-1"); + assert_eq!(context["variant_id"], "var-1"); + assert_eq!(context["evaluation_stage_id"], "stage-1"); + assert_eq!(context["stage_result_id"], "sr-1"); + assert_eq!(context["trial_id"], "trial-1"); + assert_eq!(context["simulated_user_plan_id"], "plan-1"); + assert_eq!(context["persona_index"], "2"); + assert_eq!(context["run_index"], "1"); + assert_eq!( + context["runtime_ref"], + "temper://tenant/de-variant-1/app/nerdsane/agent-answers@abc123" + ); + assert_eq!(context["app_ref"], "nerdsane/agent-answers@abc123"); + assert_eq!(context["runtime_tenant"], "de-variant-1"); + } + + #[test] + fn datadog_context_omits_unresolved_join_fields() { + let work_item = observe_test_work_item("observer", r#"{"episode_id":"ep-1"}"#); + + let context = directed_evolution_datadog_context(&work_item); + + assert_eq!(context["episode_id"], "ep-1"); + assert!(context.get("variant_id").is_none()); + assert!(context.get("trial_id").is_none()); + } + + #[test] + fn directed_evolution_prompt_header_includes_resolved_join_fields() { + let work_item = observe_test_work_item("simulated_user", FULL_CORRELATION); + + let prompt = directed_evolution_prompt(&work_item); + + assert!(prompt.contains("ResolvedCorrelation:")); + assert!(prompt.contains("de.episode_id: ep-1")); + assert!(prompt.contains("de.direction_id: dir-1")); + assert!(prompt.contains("de.generation_id: gen-1")); + assert!(prompt.contains("de.variant_id: var-1")); + assert!(prompt.contains("de.evaluation_stage_id: stage-1")); + assert!(prompt.contains("de.stage_result_id: sr-1")); + assert!(prompt.contains("de.trial_id: trial-1")); + assert!(prompt.contains("de.simulated_user_plan_id: plan-1")); + assert!(prompt.contains("de.persona_index: 2")); + assert!(prompt.contains("de.run_index: 1")); + assert!(prompt.contains( + "de.runtime_ref: temper://tenant/de-variant-1/app/nerdsane/agent-answers@abc123" + )); + assert!(prompt.contains("de.app_ref: nerdsane/agent-answers@abc123")); + assert!(prompt.contains("de.runtime_tenant: de-variant-1")); + } + + #[test] + fn directed_evolution_prompt_marks_empty_resolved_correlation() { + let work_item = observe_test_work_item("observer", "{}"); + + let prompt = directed_evolution_prompt(&work_item); + + assert!(prompt.contains("ResolvedCorrelation:\n(none)")); + } + #[test] fn headers_with_observe_metadata_attach_kernel_header() { let config = observe_test_config("http://127.0.0.1:3497".to_string()); diff --git a/crates/paw-codex-worker/src/directed_evolution/prompt.rs b/crates/paw-codex-worker/src/directed_evolution/prompt.rs index 6572c8902..d6ed79035 100644 --- a/crates/paw-codex-worker/src/directed_evolution/prompt.rs +++ b/crates/paw-codex-worker/src/directed_evolution/prompt.rs @@ -40,6 +40,9 @@ fn directed_evolution_prompt(work_item: &DirectedEvolutionWorkItemState) -> Stri let output_contract = directed_evolution_output_contract(&work_item.role); let prompt_body = directed_evolution_worker_prompt_body(&work_item.role, &literal_prompt_ref(&work_item.prompt_ref)); + let join_block = directed_evolution_prompt_join_block(&directed_evolution_join_fields( + &work_item.correlation_json, + )); format!( r#"You are a Codex WorkerRun executing a Directed Evolution WorkItem. @@ -50,6 +53,8 @@ TargetEntityId: {target_entity_id} ContextRef: {context_ref} OutputSchemaRef: {output_schema_ref} CorrelationJson: {correlation_json} +ResolvedCorrelation: +{join_block} Role contract: {role_contract} @@ -68,12 +73,25 @@ Required output shape for this role: context_ref = work_item.context_ref, output_schema_ref = work_item.output_schema_ref, correlation_json = work_item.correlation_json, + join_block = join_block, role_contract = role_contract, prompt_body = prompt_body, output_contract = output_contract, ) } +fn directed_evolution_prompt_join_block(join: &DirectedEvolutionJoinFields) -> String { + let entries = join.entries(); + if entries.is_empty() { + return "(none)".to_string(); + } + entries + .iter() + .map(|(key, value)| format!("de.{key}: {value}")) + .collect::>() + .join("\n") +} + fn directed_evolution_output_contract(role: &str) -> &'static str { match role { "observer" => { From 167b51432712bb572adc5a77731dab7daf0a2447 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 16:40:01 -0400 Subject: [PATCH 5/8] feat(de): preflight runtime credentials and assert role/target integrity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Simulated-user and evaluator Codex runs previously discovered missing runtime credentials as a confusing 401 inside the Codex child. Before launching a DE Codex role whose correlation carries runtime_base_url, resolve the auth env var names (correlation runtime_auth_env_vars plus the TEMPERPAW_RUNTIME_API_KEY/TEMPER_API_KEY fallbacks, mirroring the observer path) and verify at least one is set. If none is, fail the work item up-front with 'runtime credential missing: none of [names] is set' through the existing FailWorkerRun/receipt/FailWorkItem path. Env var values are never logged. Also assert target types at role dispatch (simulated_user -> Trial, evaluator roles -> StageResult) and fail mismatches with a clear reason; the mechanical evaluator's now-redundant StageResult bail is removed. Worker-side telemetry fail-closed/threshold logic stays out of scope — the genesis router owns it (ADR-0018). Co-Authored-By: Claude Fable 5 --- .../src/directed_evolution.rs | 91 +++++++ .../mechanical_evaluator.rs | 10 +- .../directed_evolution/observe_metadata.rs | 6 + .../src/directed_evolution/tests.rs | 243 ++++++++++++++++++ 4 files changed, 342 insertions(+), 8 deletions(-) diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index 82c134b15..dd55db2bf 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -364,6 +364,85 @@ fn directed_evolution_stage_evaluator_role(role: &str) -> bool { ) } +/// Cheap dispatch assertion: simulated users exercise Trials, evaluator +/// roles judge StageResults. A mismatch is a control-plane routing bug and +/// must fail the work item with a clear reason instead of confusing the +/// downstream brain. +fn directed_evolution_role_target_mismatch( + role: &str, + target_entity_type: &str, +) -> Option { + let expected = if role == "simulated_user" { + "Trial" + } else if directed_evolution_stage_evaluator_role(role) { + "StageResult" + } else { + return None; + }; + if target_entity_type == expected { + return None; + } + Some(format!( + "Directed Evolution role {role} requires a {expected} target, got {target_entity_type}" + )) +} + +/// Codex roles that exercise a variant runtime with bearer credentials. +/// The observer is excluded: it resolves runtime auth best-effort inside +/// its source inventory and degrades to other sources. Mechanical +/// evaluator roles never launch Codex and never authenticate. +fn directed_evolution_runtime_credential_role(role: &str) -> bool { + matches!( + role, + "simulated_user" | "reviewer" | "viability_evaluator" | "telemetry_evaluator" + ) +} + +/// B9 runtime auth preflight: when a work item's correlation names a +/// runtime to exercise, verify at least one of the runtime auth env vars +/// is set before launching Codex. Only env var NAMES are resolved and +/// reported — never values. A missing credential otherwise surfaces as a +/// confusing 401 inside the Codex child. +fn directed_evolution_runtime_credential_failure( + role: &str, + join: &DirectedEvolutionJoinFields, + env_value: impl Fn(&str) -> Option, +) -> Option { + if !directed_evolution_runtime_credential_role(role) { + return None; + } + if join.runtime_base_url.trim().is_empty() { + return None; + } + let names = directed_evolution_runtime_auth_env_var_names(&join.runtime_auth_env_vars); + let any_set = names.iter().any(|name| { + env_value(name) + .map(|value| !value.trim().is_empty()) + .unwrap_or(false) + }); + if any_set { + return None; + } + Some(format!( + "runtime credential missing: none of [{}] is set", + names.join(", ") + )) +} + +fn directed_evolution_runtime_auth_env_var_names(configured: &[String]) -> Vec { + let mut names: Vec = configured + .iter() + .map(|name| name.trim().to_string()) + .filter(|name| !name.is_empty()) + .collect(); + for fallback in ["TEMPERPAW_RUNTIME_API_KEY", "TEMPER_API_KEY"] { + if !names.iter().any(|name| name == fallback) { + names.push(fallback.to_string()); + } + } + names +} + fn stale_stage_result_should_eliminate(stage_fields: &Value) -> bool { matches!( value_field_string(stage_fields, &["Status", "status"]).as_str(), @@ -658,6 +737,11 @@ async fn run_directed_evolution_codex_role( work_item: &DirectedEvolutionWorkItemState, worker_run_id: &str, ) -> Result { + if let Some(reason) = + directed_evolution_role_target_mismatch(&work_item.role, &work_item.target_entity_type) + { + bail!("{reason}"); + } let join_fields = directed_evolution_join_fields(&work_item.correlation_json); let mut prompt = directed_evolution_prompt(work_item); info!( @@ -692,6 +776,13 @@ async fn run_directed_evolution_codex_role( return serde_json::to_string(&directed_evolution_promotion_output(&materialization)) .context("serialize Directed Evolution promoter output"); } + if let Some(reason) = directed_evolution_runtime_credential_failure( + &work_item.role, + &join_fields, + |name| env::var(name).ok(), + ) { + bail!("{reason}"); + } let workdir = resolve_directed_evolution_workdir(client, config, work_item).await?; if work_item.role == "observer" { diff --git a/crates/paw-codex-worker/src/directed_evolution/mechanical_evaluator.rs b/crates/paw-codex-worker/src/directed_evolution/mechanical_evaluator.rs index bb325086d..4ab51b8b6 100644 --- a/crates/paw-codex-worker/src/directed_evolution/mechanical_evaluator.rs +++ b/crates/paw-codex-worker/src/directed_evolution/mechanical_evaluator.rs @@ -7,14 +7,8 @@ async fn run_directed_evolution_mechanical_evaluator( config: &Config, work_item: &DirectedEvolutionWorkItemState, ) -> Result { - if work_item.target_entity_type != "StageResult" { - bail!( - "mechanical Directed Evolution evaluator {} requires a StageResult target, got {}", - work_item.role, - work_item.target_entity_type - ); - } - + // Target-type integrity is asserted once at role dispatch + // (directed_evolution_role_target_mismatch) before this runs. let stage_result = fetch_directed_evolution_entity_fields(client, config, "StageResults", &work_item.target_entity_id) .await?; diff --git a/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs b/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs index c3ebb6028..fd347bab1 100644 --- a/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs +++ b/crates/paw-codex-worker/src/directed_evolution/observe_metadata.rs @@ -25,6 +25,10 @@ struct DirectedEvolutionJoinFields { runtime_ref: String, app_ref: String, runtime_tenant: String, + /// Runtime auth preflight inputs (B9). Resolved from the same single + /// CorrelationJson parse but not part of the de.* join-field entries. + runtime_base_url: String, + runtime_auth_env_vars: Vec, } impl DirectedEvolutionJoinFields { @@ -73,6 +77,8 @@ fn directed_evolution_join_fields(correlation_json: &str) -> DirectedEvolutionJo runtime_ref: field("runtime_ref"), app_ref: field("app_ref"), runtime_tenant: field("runtime_tenant"), + runtime_base_url: field("runtime_base_url"), + runtime_auth_env_vars: observer_string_array(&correlation, "runtime_auth_env_vars"), } } diff --git a/crates/paw-codex-worker/src/directed_evolution/tests.rs b/crates/paw-codex-worker/src/directed_evolution/tests.rs index e6a771d8b..a8661db7b 100644 --- a/crates/paw-codex-worker/src/directed_evolution/tests.rs +++ b/crates/paw-codex-worker/src/directed_evolution/tests.rs @@ -953,6 +953,249 @@ mod directed_evolution_tests { assert!(!plan.evaluator_ref.trim().is_empty()); } + #[test] + fn role_target_mismatch_fails_simulated_user_without_trial_target() { + let reason = directed_evolution_role_target_mismatch("simulated_user", "StageResult") + .expect("simulated_user requires a Trial target"); + + assert!(reason.contains("simulated_user")); + assert!(reason.contains("Trial")); + assert!(reason.contains("StageResult")); + } + + #[test] + fn role_target_mismatch_fails_evaluator_roles_without_stage_result_target() { + for role in [ + "reviewer", + "viability_evaluator", + "state_verifier", + "telemetry_evaluator", + "wasm_evaluator", + ] { + let reason = directed_evolution_role_target_mismatch(role, "Trial") + .unwrap_or_else(|| panic!("{role} requires a StageResult target")); + assert!(reason.contains(role)); + assert!(reason.contains("StageResult")); + } + } + + #[test] + fn role_target_mismatch_allows_expected_and_unconstrained_targets() { + assert!(directed_evolution_role_target_mismatch("simulated_user", "Trial").is_none()); + assert!(directed_evolution_role_target_mismatch("reviewer", "StageResult").is_none()); + assert!(directed_evolution_role_target_mismatch("observer", "Organism").is_none()); + assert!( + directed_evolution_role_target_mismatch("variant_generator", "Generation").is_none() + ); + } + + #[test] + fn runtime_credential_preflight_fails_when_no_named_env_var_is_set() { + let join = directed_evolution_join_fields( + r#"{ + "runtime_base_url": "https://temper.example", + "runtime_auth_env_vars": ["DE_RUNTIME_KEY_A", "DE_RUNTIME_KEY_B"] + }"#, + ); + + let reason = + directed_evolution_runtime_credential_failure("simulated_user", &join, |_| None) + .expect("missing runtime credentials should fail preflight"); + + assert_eq!( + reason, + "runtime credential missing: none of [DE_RUNTIME_KEY_A, DE_RUNTIME_KEY_B, TEMPERPAW_RUNTIME_API_KEY, TEMPER_API_KEY] is set" + ); + } + + #[test] + fn runtime_credential_preflight_passes_when_any_named_env_var_is_set() { + let join = directed_evolution_join_fields( + r#"{ + "runtime_base_url": "https://temper.example", + "runtime_auth_env_vars": ["DE_RUNTIME_KEY_A"] + }"#, + ); + + let configured = directed_evolution_runtime_credential_failure( + "telemetry_evaluator", + &join, + |name| (name == "DE_RUNTIME_KEY_A").then(|| "secret".to_string()), + ); + let fallback = directed_evolution_runtime_credential_failure( + "reviewer", + &directed_evolution_join_fields(r#"{"runtime_base_url":"https://temper.example"}"#), + |name| (name == "TEMPER_API_KEY").then(|| "secret".to_string()), + ); + + assert!(configured.is_none()); + assert!(fallback.is_none()); + } + + #[test] + fn runtime_credential_preflight_ignores_blank_env_values() { + let join = directed_evolution_join_fields( + r#"{"runtime_base_url":"https://temper.example"}"#, + ); + + let reason = directed_evolution_runtime_credential_failure("simulated_user", &join, |_| { + Some(" ".to_string()) + }) + .expect("blank runtime credentials should fail preflight"); + + assert!(reason.starts_with("runtime credential missing: none of [")); + } + + #[test] + fn runtime_credential_preflight_skips_non_runtime_roles_and_items() { + let runtime_join = directed_evolution_join_fields( + r#"{"runtime_base_url":"https://temper.example"}"#, + ); + let no_runtime_join = directed_evolution_join_fields(r#"{"episode_id":"ep-1"}"#); + + assert!( + directed_evolution_runtime_credential_failure("observer", &runtime_join, |_| None) + .is_none(), + "observer resolves runtime auth best-effort in its source inventory" + ); + assert!( + directed_evolution_runtime_credential_failure( + "variant_generator", + &runtime_join, + |_| None + ) + .is_none() + ); + assert!( + directed_evolution_runtime_credential_failure( + "simulated_user", + &no_runtime_join, + |_| None + ) + .is_none(), + "work items without runtime_base_url have no runtime to authenticate against" + ); + } + + #[tokio::test] + async fn codex_role_dispatch_fails_fast_on_role_target_mismatch() { + let config = Config { + temper_url: "http://127.0.0.1:9".to_string(), + tenant: "default".to_string(), + worker_id: "mac-mini-codex-1".to_string(), + worker_token: None, + workspace_root: PathBuf::from("/tmp/worktrees"), + repo_root: PathBuf::from("/tmp/temperpaw"), + codex_bin: "codex".to_string(), + max_concurrent_runs: 1, + enable_execution: false, + poll_on_start: true, + codex_exec_smoke: false, + codex_exec_timeout: Duration::from_secs(30), + }; + let client = reqwest::Client::new(); + let work_item = DirectedEvolutionWorkItemState { + id: "wi-mismatch".to_string(), + status: "Running".to_string(), + role: "simulated_user".to_string(), + target_entity_type: "StageResult".to_string(), + target_entity_id: "sr-1".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: "{}".to_string(), + worker_run_id: String::new(), + }; + + let error = run_directed_evolution_codex_role(&client, &config, &work_item, "wr-1") + .await + .expect_err("role/target mismatch must fail dispatch up-front"); + + assert!(error.to_string().contains("Trial")); + assert!(error.to_string().contains("StageResult")); + } + + #[tokio::test] + async fn codex_role_dispatch_fails_fast_on_missing_runtime_credentials() { + let previous_runtime_key = env::var_os("TEMPERPAW_RUNTIME_API_KEY"); + let previous_api_key = env::var_os("TEMPER_API_KEY"); + unsafe { + env::remove_var("TEMPERPAW_RUNTIME_API_KEY"); + env::remove_var("TEMPER_API_KEY"); + } + let config = Config { + temper_url: "http://127.0.0.1:9".to_string(), + tenant: "default".to_string(), + worker_id: "mac-mini-codex-1".to_string(), + worker_token: None, + workspace_root: PathBuf::from("/tmp/worktrees"), + repo_root: PathBuf::from("/tmp/temperpaw"), + codex_bin: "codex".to_string(), + max_concurrent_runs: 1, + enable_execution: true, + poll_on_start: true, + codex_exec_smoke: false, + codex_exec_timeout: Duration::from_secs(30), + }; + let client = reqwest::Client::new(); + let work_item = DirectedEvolutionWorkItemState { + id: "wi-preflight".to_string(), + status: "Running".to_string(), + role: "simulated_user".to_string(), + target_entity_type: "Trial".to_string(), + target_entity_id: "trial-1".to_string(), + prompt_ref: String::new(), + context_ref: String::new(), + output_schema_ref: String::new(), + correlation_json: r#"{ + "runtime_base_url": "https://temper.example", + "runtime_auth_env_vars": ["PAW_TEST_DE_RUNTIME_KEY_UNSET"] + }"# + .to_string(), + worker_run_id: String::new(), + }; + + let result = run_directed_evolution_codex_role(&client, &config, &work_item, "wr-1").await; + + unsafe { + if let Some(value) = previous_runtime_key { + env::set_var("TEMPERPAW_RUNTIME_API_KEY", value); + } + if let Some(value) = previous_api_key { + env::set_var("TEMPER_API_KEY", value); + } + } + let error = result.expect_err("missing runtime credentials must fail dispatch up-front"); + assert!( + error + .to_string() + .contains("runtime credential missing: none of [PAW_TEST_DE_RUNTIME_KEY_UNSET, TEMPERPAW_RUNTIME_API_KEY, TEMPER_API_KEY] is set"), + "unexpected error: {error}" + ); + } + + #[test] + fn join_fields_resolve_runtime_auth_preflight_inputs() { + let join = directed_evolution_join_fields( + r#"{ + "runtime_base_url": "https://temper.example/", + "runtime_auth_env_vars": ["DE_RUNTIME_KEY_A", " ", "DE_RUNTIME_KEY_B"] + }"#, + ); + + assert_eq!(join.runtime_base_url, "https://temper.example/"); + assert_eq!( + join.runtime_auth_env_vars, + vec!["DE_RUNTIME_KEY_A".to_string(), "DE_RUNTIME_KEY_B".to_string()] + ); + assert!( + join.entries() + .iter() + .all(|(key, _)| *key != "runtime_base_url" && *key != "runtime_auth_env_vars"), + "preflight inputs are not de.* join fields" + ); + } + #[test] fn directed_evolution_start_episode_command_accepts_contract_path() { let command = parse_worker_command([ From ec8a66bf156bf949c041aaaa88fddb1667e9969f Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 16:42:59 -0400 Subject: [PATCH 6/8] refactor(de): collapse nested if in boot recovery per clippy Co-Authored-By: Claude Fable 5 --- crates/paw-codex-worker/src/directed_evolution.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index dd55db2bf..8430e311f 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -636,8 +636,8 @@ async fn fail_recovered_directed_evolution_work_item( &join_fields, ); let failure_reason = directed_evolution_restart_failure_reason(&config.worker_id); - if !work_item.worker_run_id.trim().is_empty() { - if let Err(report_error) = post_paw_orchestration_action( + if !work_item.worker_run_id.trim().is_empty() + && let Err(report_error) = post_paw_orchestration_action( client, config, "WorkerRuns", @@ -650,9 +650,8 @@ async fn fail_recovered_directed_evolution_work_item( Some(&observe_metadata), ) .await - { - warn!(%report_error, work_item_id, worker_run_id = %work_item.worker_run_id, "failed to fail recovered Directed Evolution WorkerRun"); - } + { + warn!(%report_error, work_item_id, worker_run_id = %work_item.worker_run_id, "failed to fail recovered Directed Evolution WorkerRun"); } if let Err(report_error) = route_directed_evolution_failure_receipt( client, From 81a5aa4c18f505b9dc271efd7602ad0e4caf37b4 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Thu, 11 Jun 2026 16:42:59 -0400 Subject: [PATCH 7/8] docs(de): record B7-B9 worker observability plan and progress Co-Authored-By: Claude Fable 5 --- .../001_20260611_134500_de-worker-b7-b9.md | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 .progress/001_20260611_134500_de-worker-b7-b9.md diff --git a/.progress/001_20260611_134500_de-worker-b7-b9.md b/.progress/001_20260611_134500_de-worker-b7-b9.md new file mode 100644 index 000000000..f67528dc0 --- /dev/null +++ b/.progress/001_20260611_134500_de-worker-b7-b9.md @@ -0,0 +1,53 @@ +# DE worker B7-B9 (crates/paw-codex-worker only) + +Branch: codex/de-vision-completion-20260611 (origin = GitHub nerdsane/temperpaw, PR #393 — do not edit PR, do not merge). +Baseline: 97 tests green (`cargo test -p paw-codex-worker`). + +## What we are addressing +The temper kernel (ADR-0041) parses `X-Temper-Observe-Metadata` into `temper.observation.*` +span attributes, but the temperpaw worker sends nothing — DE telemetry cannot be joined to +episodes/variants/trials. Join fields live only in raw CorrelationJson. Simulated-user and +evaluator Codex runs hit confusing in-run 401s when runtime credentials are missing. + +## Expected end state +1. Every worker→Temper call made on behalf of a DE work item carries the observe-metadata + header (producer.work_item_id, producer.worker_run_id, de.role, de.* join fields). +2. Observer runtime probes carry it too. +3. DE Codex children get TEMPER_OBSERVE_METADATA + DD_TAGS env vars mechanically. +4. Datadog context + prompt header expose the full resolved join-field set from one parser. +5. Missing runtime credentials and role/target mismatches fail the work item up-front with a + structured reason through the existing FailWorkerRun/receipt/FailWorkItem path. + +## Plan (red-green per phase, one commit each) +1. B7a: new `directed_evolution/observe_metadata.rs` — `DirectedEvolutionJoinFields` parser + (one place, shared by B7+B8), header-value builder with kernel limits (32 keys, key<=96B, + value<=1024B), DD_TAGS builder, codex child env builder. Tests first (red), then impl. +2. B7b: thread `observe_metadata: Option<&str>` through `post_directed_evolution_action`, + `post_paw_orchestration_action`, DE `create_entity` calls; header inserted via + `headers_with_observe_metadata`. Observer probes (`observer_runtime_get_text`/collection + sample) take the header. Codex child env injection in `run_codex_exec_command` DE path. +3. B8: extend `directed_evolution_datadog_context` with resolved join fields; add resolved + join block to the DE prompt header. Tests assert full field set in both. +4. B9: runtime-auth preflight (correlation runtime_auth_env_vars + TEMPERPAW_RUNTIME_API_KEY/ + TEMPER_API_KEY fallbacks; fail "runtime credential missing: none of [names] is set") for + simulated_user + codex evaluator roles when runtime_base_url present; role/target-type + assertions (simulated_user→Trial, evaluator→StageResult). Remove the now-redundant + StageResult bail in mechanical_evaluator.rs. +5. Full `cargo test -p paw-codex-worker`, push. + +NOT in scope: B10 temper rev bump; genesis-side telemetry fail-closed/threshold logic (ADR-0018). + +## Status +- [x] Baseline green (97) +- [x] Phase 1+2 (B7) — commit b56a1374, 113 tests green (16 new) +- [x] Phase 3 (B8) — commit 0a93201f, 117 tests green (4 new) +- [x] Phase 4 (B9) — commit 167b5143, 127 tests green (10 new) +- [x] Clippy clean (collapsed pre-existing nested if in boot recovery) +- [x] Full suite green (127, baseline 97) + pushed + +## Deviations from plan +- B7a and B7b landed as one commit (b56a1374): committing the pure helpers + alone would have left dead-code warnings in the binary target. +- DirectedEvolutionJoinFields also carries runtime_base_url + + runtime_auth_env_vars (excluded from de.* entries) so B9's preflight + shares the same single CorrelationJson parse. From e9e599f3260d1dd625c18d36f6bed8cd0d8bd540 Mon Sep 17 00:00:00 2001 From: rita-aga Date: Fri, 3 Jul 2026 16:40:16 -0400 Subject: [PATCH 8/8] refactor(de): split directed_evolution.rs under the giant-module budget Main gained a repo-health gate capping paw-codex-worker sources at 900 lines; directed_evolution.rs stood at 960 after the merge. Move-only split into directed_evolution/staleness.rs (stage-evaluator targeting, stale stage-result elimination, runtime credential guards; 198 lines) and directed_evolution/receipts.rs (receipt bodies, receipt routing, boot recovery of Running work items; 231 lines), leaving the parent at 542. Worker tests 127/127; the budget gate passes. Co-Authored-By: Claude Fable 5 --- .../src/directed_evolution.rs | 422 +----------------- .../src/directed_evolution/receipts.rs | 231 ++++++++++ .../src/directed_evolution/staleness.rs | 198 ++++++++ 3 files changed, 431 insertions(+), 420 deletions(-) create mode 100644 crates/paw-codex-worker/src/directed_evolution/receipts.rs create mode 100644 crates/paw-codex-worker/src/directed_evolution/staleness.rs diff --git a/crates/paw-codex-worker/src/directed_evolution.rs b/crates/paw-codex-worker/src/directed_evolution.rs index 8430e311f..97720750c 100644 --- a/crates/paw-codex-worker/src/directed_evolution.rs +++ b/crates/paw-codex-worker/src/directed_evolution.rs @@ -257,199 +257,6 @@ async fn handle_queued_directed_evolution_work_item( } } -async fn stale_directed_evolution_work_item_reason( - client: &reqwest::Client, - config: &Config, - work_item: &DirectedEvolutionWorkItemState, -) -> Result> { - if !directed_evolution_stage_evaluator_role(&work_item.role) - || work_item.target_entity_type != "StageResult" - { - return Ok(None); - } - - let stage_fields = - fetch_directed_evolution_entity_fields(client, config, "StageResults", &work_item.target_entity_id) - .await?; - let variant_fields = { - let variant_id = value_field_string(&stage_fields, &["VariantId", "variant_id"]); - if variant_id.trim().is_empty() { - json!({}) - } else { - fetch_directed_evolution_entity_fields(client, config, "Variants", &variant_id).await? - } - }; - Ok(stale_directed_evolution_stage_work_reason( - work_item, - &stage_fields, - &variant_fields, - )) -} - -async fn eliminate_stale_directed_evolution_stage_result( - client: &reqwest::Client, - config: &Config, - work_item: &DirectedEvolutionWorkItemState, - reason: &str, - observe_metadata: Option<&str>, -) -> Result<()> { - if !stale_stage_work_targets_stage_result(work_item) { - return Ok(()); - } - - let stage_fields = - fetch_directed_evolution_entity_fields(client, config, "StageResults", &work_item.target_entity_id) - .await?; - if !stale_stage_result_should_eliminate(&stage_fields) { - return Ok(()); - } - - post_directed_evolution_action( - client, - config, - "StageResults", - &work_item.target_entity_id, - "EliminateStageResult", - json!({ - "EliminationRuleId": "stale-after-variant-terminal", - "EvidenceArtifactId": value_field_string(&stage_fields, &["EvidenceArtifactId", "evidence_artifact_id"]), - "Reason": reason, - }), - observe_metadata, - ) - .await -} - -fn stale_directed_evolution_stage_work_reason( - work_item: &DirectedEvolutionWorkItemState, - stage_fields: &Value, - variant_fields: &Value, -) -> Option { - if !stale_stage_work_targets_stage_result(work_item) { - return None; - } - let stage_status = value_field_string(stage_fields, &["Status", "status"]); - if !stage_status.trim().is_empty() && stage_status != "Running" { - return Some(format!( - "Target StageResult {} is already {}; skipping stale {} work", - work_item.target_entity_id, stage_status, work_item.role - )); - } - let variant_status = value_field_string(variant_fields, &["Status", "status"]); - if matches!( - variant_status.as_str(), - "Eliminated" | "Promoted" | "Superseded" | "Failed" - ) { - return Some(format!( - "Target variant is already {}; skipping stale {} work for StageResult {}", - variant_status, work_item.role, work_item.target_entity_id - )); - } - None -} - -fn stale_stage_work_targets_stage_result(work_item: &DirectedEvolutionWorkItemState) -> bool { - directed_evolution_stage_evaluator_role(&work_item.role) - && work_item.target_entity_type == "StageResult" -} - -fn directed_evolution_stage_evaluator_role(role: &str) -> bool { - matches!( - role, - "reviewer" - | "viability_evaluator" - | "state_verifier" - | "telemetry_evaluator" - | "wasm_evaluator" - ) -} - -/// Cheap dispatch assertion: simulated users exercise Trials, evaluator -/// roles judge StageResults. A mismatch is a control-plane routing bug and -/// must fail the work item with a clear reason instead of confusing the -/// downstream brain. -fn directed_evolution_role_target_mismatch( - role: &str, - target_entity_type: &str, -) -> Option { - let expected = if role == "simulated_user" { - "Trial" - } else if directed_evolution_stage_evaluator_role(role) { - "StageResult" - } else { - return None; - }; - if target_entity_type == expected { - return None; - } - Some(format!( - "Directed Evolution role {role} requires a {expected} target, got {target_entity_type}" - )) -} - -/// Codex roles that exercise a variant runtime with bearer credentials. -/// The observer is excluded: it resolves runtime auth best-effort inside -/// its source inventory and degrades to other sources. Mechanical -/// evaluator roles never launch Codex and never authenticate. -fn directed_evolution_runtime_credential_role(role: &str) -> bool { - matches!( - role, - "simulated_user" | "reviewer" | "viability_evaluator" | "telemetry_evaluator" - ) -} - -/// B9 runtime auth preflight: when a work item's correlation names a -/// runtime to exercise, verify at least one of the runtime auth env vars -/// is set before launching Codex. Only env var NAMES are resolved and -/// reported — never values. A missing credential otherwise surfaces as a -/// confusing 401 inside the Codex child. -fn directed_evolution_runtime_credential_failure( - role: &str, - join: &DirectedEvolutionJoinFields, - env_value: impl Fn(&str) -> Option, -) -> Option { - if !directed_evolution_runtime_credential_role(role) { - return None; - } - if join.runtime_base_url.trim().is_empty() { - return None; - } - let names = directed_evolution_runtime_auth_env_var_names(&join.runtime_auth_env_vars); - let any_set = names.iter().any(|name| { - env_value(name) - .map(|value| !value.trim().is_empty()) - .unwrap_or(false) - }); - if any_set { - return None; - } - Some(format!( - "runtime credential missing: none of [{}] is set", - names.join(", ") - )) -} - -fn directed_evolution_runtime_auth_env_var_names(configured: &[String]) -> Vec { - let mut names: Vec = configured - .iter() - .map(|name| name.trim().to_string()) - .filter(|name| !name.is_empty()) - .collect(); - for fallback in ["TEMPERPAW_RUNTIME_API_KEY", "TEMPER_API_KEY"] { - if !names.iter().any(|name| name == fallback) { - names.push(fallback.to_string()); - } - } - names -} - -fn stale_stage_result_should_eliminate(stage_fields: &Value) -> bool { - matches!( - value_field_string(stage_fields, &["Status", "status"]).as_str(), - "Running" | "Failed" - ) -} - include!("directed_evolution/observe_metadata.rs"); include!("directed_evolution/evidence.rs"); @@ -458,233 +265,6 @@ include!("directed_evolution/human_episode_defaults.rs"); include!("directed_evolution/human_episode_plan.rs"); include!("directed_evolution/human_episode.rs"); -fn directed_evolution_start_worker_run_body( - work_item: &DirectedEvolutionWorkItemState, - worker_id: &str, - worker_run_id: &str, - parent_session_id: &str, -) -> Value { - json!({ - "Role": work_item.role, - "WorkItemId": work_item.id, - "WorkerId": worker_id, - "ProviderId": DIRECTED_EVOLUTION_WORKER_PROVIDER_ID, - "AgentKind": directed_evolution_agent_kind_for_role(&work_item.role), - "Model": directed_evolution_model_for_role(&work_item.role), - "SessionId": worker_run_id, - "ParentSessionId": parent_session_id, - "CorrelationJson": work_item.correlation_json, - }) -} - -fn directed_evolution_start_work_item_body(worker_run_id: &str) -> Value { - json!({ "WorkerRunId": worker_run_id }) -} - -fn directed_evolution_success_receipt_body( - work_item: &DirectedEvolutionWorkItemState, - worker_run_id: &str, - result_json: &str, - evidence_artifact_id: &str, - summary: &str, -) -> Value { - json!({ - "WorkItemId": work_item.id, - "Role": work_item.role, - "TargetEntityType": work_item.target_entity_type, - "TargetEntityId": work_item.target_entity_id, - "WorkerRunId": worker_run_id, - "ResultJson": result_json, - "EvidenceArtifactId": evidence_artifact_id, - "Summary": summary, - "CorrelationJson": work_item.correlation_json, - }) -} - -fn directed_evolution_failure_receipt_body( - work_item: &DirectedEvolutionWorkItemState, - worker_run_id: &str, - failure_reason: &str, - evidence_artifact_id: &str, -) -> Value { - json!({ - "WorkItemId": work_item.id, - "Role": work_item.role, - "TargetEntityType": work_item.target_entity_type, - "TargetEntityId": work_item.target_entity_id, - "WorkerRunId": worker_run_id, - "FailureReason": failure_reason, - "EvidenceArtifactId": evidence_artifact_id, - "CorrelationJson": work_item.correlation_json, - }) -} - -#[allow(clippy::too_many_arguments)] -async fn route_directed_evolution_success_receipt( - client: &reqwest::Client, - config: &Config, - work_item: &DirectedEvolutionWorkItemState, - worker_run_id: &str, - result_json: &str, - evidence_artifact_id: &str, - summary: &str, - observe_metadata: Option<&str>, -) -> Result { - let receipt_id = create_entity_with_observe_metadata( - client, - config, - "WorkItemReceipts", - json!({}), - observe_metadata, - ) - .await?; - post_directed_evolution_action( - client, - config, - "WorkItemReceipts", - &receipt_id, - "RouteSucceededWorkItem", - directed_evolution_success_receipt_body( - work_item, - worker_run_id, - result_json, - evidence_artifact_id, - summary, - ), - observe_metadata, - ) - .await?; - Ok(receipt_id) -} - -async fn route_directed_evolution_failure_receipt( - client: &reqwest::Client, - config: &Config, - work_item: &DirectedEvolutionWorkItemState, - worker_run_id: &str, - failure_reason: &str, - evidence_artifact_id: &str, - observe_metadata: Option<&str>, -) -> Result { - let receipt_id = create_entity_with_observe_metadata( - client, - config, - "WorkItemReceipts", - json!({}), - observe_metadata, - ) - .await?; - post_directed_evolution_action( - client, - config, - "WorkItemReceipts", - &receipt_id, - "RouteFailedWorkItem", - directed_evolution_failure_receipt_body( - work_item, - worker_run_id, - failure_reason, - evidence_artifact_id, - ), - observe_metadata, - ) - .await?; - Ok(receipt_id) -} - -fn directed_evolution_running_recovery_filter(worker_id: &str) -> String { - // OData escapes a single quote inside a string literal by doubling it. - let escaped = worker_id.replace('\'', "''"); - format!("Status eq 'Running' and ClaimedBy eq '{escaped}'") -} - -fn directed_evolution_restart_failure_reason(worker_id: &str) -> String { - format!( - "worker {worker_id} restarted while this work item was running; failed for control-plane re-dispatch" - ) -} - -async fn recover_boot_running_directed_evolution_work_items( - client: &reqwest::Client, - config: &Config, -) -> Result<()> { - let filter = directed_evolution_running_recovery_filter(&config.worker_id); - let ids = query_boot_entity_ids_filtered(client, config, "WorkItems", &filter).await?; - for work_item_id in ids { - if let Err(error) = - fail_recovered_directed_evolution_work_item(client, config, &work_item_id).await - { - warn!(%error, work_item_id, "failed to recover Running Directed Evolution WorkItem"); - } - } - Ok(()) -} - -async fn fail_recovered_directed_evolution_work_item( - client: &reqwest::Client, - config: &Config, - work_item_id: &str, -) -> Result<()> { - let work_item = fetch_directed_evolution_work_item(client, config, work_item_id).await?; - if work_item.status != "Running" { - return Ok(()); - } - let join_fields = directed_evolution_join_fields(&work_item.correlation_json); - let observe_metadata = directed_evolution_work_item_observe_metadata( - &work_item, - &work_item.worker_run_id, - &join_fields, - ); - let failure_reason = directed_evolution_restart_failure_reason(&config.worker_id); - if !work_item.worker_run_id.trim().is_empty() - && let Err(report_error) = post_paw_orchestration_action( - client, - config, - "WorkerRuns", - &work_item.worker_run_id, - "FailWorkerRun", - json!({ - "FailureReason": failure_reason, - "EvidenceArtifactId": "", - }), - Some(&observe_metadata), - ) - .await - { - warn!(%report_error, work_item_id, worker_run_id = %work_item.worker_run_id, "failed to fail recovered Directed Evolution WorkerRun"); - } - if let Err(report_error) = route_directed_evolution_failure_receipt( - client, - config, - &work_item, - &work_item.worker_run_id, - &failure_reason, - "", - Some(&observe_metadata), - ) - .await - { - warn!(%report_error, work_item_id, "failed to route recovered Directed Evolution failure receipt"); - } - post_paw_orchestration_action( - client, - config, - "WorkItems", - &work_item.id, - "FailWorkItem", - json!({ - "FailureReason": failure_reason, - "EvidenceArtifactId": "", - }), - Some(&observe_metadata), - ) - .await?; - warn!( - work_item_id, - "failed Running Directed Evolution WorkItem after worker restart" - ); - Ok(()) -} async fn post_directed_evolution_action( client: &reqwest::Client, @@ -952,6 +532,8 @@ async fn recover_directed_evolution_variant_output( } +include!("directed_evolution/staleness.rs"); +include!("directed_evolution/receipts.rs"); include!("directed_evolution/workdir.rs"); include!("directed_evolution/mechanical_evaluator.rs"); include!("directed_evolution/prompt.rs"); diff --git a/crates/paw-codex-worker/src/directed_evolution/receipts.rs b/crates/paw-codex-worker/src/directed_evolution/receipts.rs new file mode 100644 index 000000000..d3e79d80f --- /dev/null +++ b/crates/paw-codex-worker/src/directed_evolution/receipts.rs @@ -0,0 +1,231 @@ +// WorkerRun/WorkItem receipt bodies, receipt routing, and boot-time +// recovery of Running Directed Evolution work items after a worker +// restart. Included into main.rs's flat namespace via directed_evolution.rs. + +fn directed_evolution_start_worker_run_body( + work_item: &DirectedEvolutionWorkItemState, + worker_id: &str, + worker_run_id: &str, + parent_session_id: &str, +) -> Value { + json!({ + "Role": work_item.role, + "WorkItemId": work_item.id, + "WorkerId": worker_id, + "ProviderId": DIRECTED_EVOLUTION_WORKER_PROVIDER_ID, + "AgentKind": directed_evolution_agent_kind_for_role(&work_item.role), + "Model": directed_evolution_model_for_role(&work_item.role), + "SessionId": worker_run_id, + "ParentSessionId": parent_session_id, + "CorrelationJson": work_item.correlation_json, + }) +} + +fn directed_evolution_start_work_item_body(worker_run_id: &str) -> Value { + json!({ "WorkerRunId": worker_run_id }) +} + +fn directed_evolution_success_receipt_body( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + result_json: &str, + evidence_artifact_id: &str, + summary: &str, +) -> Value { + json!({ + "WorkItemId": work_item.id, + "Role": work_item.role, + "TargetEntityType": work_item.target_entity_type, + "TargetEntityId": work_item.target_entity_id, + "WorkerRunId": worker_run_id, + "ResultJson": result_json, + "EvidenceArtifactId": evidence_artifact_id, + "Summary": summary, + "CorrelationJson": work_item.correlation_json, + }) +} + +fn directed_evolution_failure_receipt_body( + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + failure_reason: &str, + evidence_artifact_id: &str, +) -> Value { + json!({ + "WorkItemId": work_item.id, + "Role": work_item.role, + "TargetEntityType": work_item.target_entity_type, + "TargetEntityId": work_item.target_entity_id, + "WorkerRunId": worker_run_id, + "FailureReason": failure_reason, + "EvidenceArtifactId": evidence_artifact_id, + "CorrelationJson": work_item.correlation_json, + }) +} + +#[allow(clippy::too_many_arguments)] +async fn route_directed_evolution_success_receipt( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + result_json: &str, + evidence_artifact_id: &str, + summary: &str, + observe_metadata: Option<&str>, +) -> Result { + let receipt_id = create_entity_with_observe_metadata( + client, + config, + "WorkItemReceipts", + json!({}), + observe_metadata, + ) + .await?; + post_directed_evolution_action( + client, + config, + "WorkItemReceipts", + &receipt_id, + "RouteSucceededWorkItem", + directed_evolution_success_receipt_body( + work_item, + worker_run_id, + result_json, + evidence_artifact_id, + summary, + ), + observe_metadata, + ) + .await?; + Ok(receipt_id) +} + +async fn route_directed_evolution_failure_receipt( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, + worker_run_id: &str, + failure_reason: &str, + evidence_artifact_id: &str, + observe_metadata: Option<&str>, +) -> Result { + let receipt_id = create_entity_with_observe_metadata( + client, + config, + "WorkItemReceipts", + json!({}), + observe_metadata, + ) + .await?; + post_directed_evolution_action( + client, + config, + "WorkItemReceipts", + &receipt_id, + "RouteFailedWorkItem", + directed_evolution_failure_receipt_body( + work_item, + worker_run_id, + failure_reason, + evidence_artifact_id, + ), + observe_metadata, + ) + .await?; + Ok(receipt_id) +} + +fn directed_evolution_running_recovery_filter(worker_id: &str) -> String { + // OData escapes a single quote inside a string literal by doubling it. + let escaped = worker_id.replace('\'', "''"); + format!("Status eq 'Running' and ClaimedBy eq '{escaped}'") +} + +fn directed_evolution_restart_failure_reason(worker_id: &str) -> String { + format!( + "worker {worker_id} restarted while this work item was running; failed for control-plane re-dispatch" + ) +} + +async fn recover_boot_running_directed_evolution_work_items( + client: &reqwest::Client, + config: &Config, +) -> Result<()> { + let filter = directed_evolution_running_recovery_filter(&config.worker_id); + let ids = query_boot_entity_ids_filtered(client, config, "WorkItems", &filter).await?; + for work_item_id in ids { + if let Err(error) = + fail_recovered_directed_evolution_work_item(client, config, &work_item_id).await + { + warn!(%error, work_item_id, "failed to recover Running Directed Evolution WorkItem"); + } + } + Ok(()) +} + +async fn fail_recovered_directed_evolution_work_item( + client: &reqwest::Client, + config: &Config, + work_item_id: &str, +) -> Result<()> { + let work_item = fetch_directed_evolution_work_item(client, config, work_item_id).await?; + if work_item.status != "Running" { + return Ok(()); + } + let join_fields = directed_evolution_join_fields(&work_item.correlation_json); + let observe_metadata = directed_evolution_work_item_observe_metadata( + &work_item, + &work_item.worker_run_id, + &join_fields, + ); + let failure_reason = directed_evolution_restart_failure_reason(&config.worker_id); + if !work_item.worker_run_id.trim().is_empty() + && let Err(report_error) = post_paw_orchestration_action( + client, + config, + "WorkerRuns", + &work_item.worker_run_id, + "FailWorkerRun", + json!({ + "FailureReason": failure_reason, + "EvidenceArtifactId": "", + }), + Some(&observe_metadata), + ) + .await + { + warn!(%report_error, work_item_id, worker_run_id = %work_item.worker_run_id, "failed to fail recovered Directed Evolution WorkerRun"); + } + if let Err(report_error) = route_directed_evolution_failure_receipt( + client, + config, + &work_item, + &work_item.worker_run_id, + &failure_reason, + "", + Some(&observe_metadata), + ) + .await + { + warn!(%report_error, work_item_id, "failed to route recovered Directed Evolution failure receipt"); + } + post_paw_orchestration_action( + client, + config, + "WorkItems", + &work_item.id, + "FailWorkItem", + json!({ + "FailureReason": failure_reason, + "EvidenceArtifactId": "", + }), + Some(&observe_metadata), + ) + .await?; + warn!( + work_item_id, + "failed Running Directed Evolution WorkItem after worker restart" + ); + Ok(()) +} diff --git a/crates/paw-codex-worker/src/directed_evolution/staleness.rs b/crates/paw-codex-worker/src/directed_evolution/staleness.rs new file mode 100644 index 000000000..6d5af7544 --- /dev/null +++ b/crates/paw-codex-worker/src/directed_evolution/staleness.rs @@ -0,0 +1,198 @@ +// Staleness and role-guard checks for Directed Evolution work items: +// stage-evaluator targeting, stale stage-result elimination, and runtime +// credential preflight guards. Included into main.rs's flat namespace via +// directed_evolution.rs (see the include! block there). + +async fn stale_directed_evolution_work_item_reason( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, +) -> Result> { + if !directed_evolution_stage_evaluator_role(&work_item.role) + || work_item.target_entity_type != "StageResult" + { + return Ok(None); + } + + let stage_fields = + fetch_directed_evolution_entity_fields(client, config, "StageResults", &work_item.target_entity_id) + .await?; + let variant_fields = { + let variant_id = value_field_string(&stage_fields, &["VariantId", "variant_id"]); + if variant_id.trim().is_empty() { + json!({}) + } else { + fetch_directed_evolution_entity_fields(client, config, "Variants", &variant_id).await? + } + }; + Ok(stale_directed_evolution_stage_work_reason( + work_item, + &stage_fields, + &variant_fields, + )) +} + +async fn eliminate_stale_directed_evolution_stage_result( + client: &reqwest::Client, + config: &Config, + work_item: &DirectedEvolutionWorkItemState, + reason: &str, + observe_metadata: Option<&str>, +) -> Result<()> { + if !stale_stage_work_targets_stage_result(work_item) { + return Ok(()); + } + + let stage_fields = + fetch_directed_evolution_entity_fields(client, config, "StageResults", &work_item.target_entity_id) + .await?; + if !stale_stage_result_should_eliminate(&stage_fields) { + return Ok(()); + } + + post_directed_evolution_action( + client, + config, + "StageResults", + &work_item.target_entity_id, + "EliminateStageResult", + json!({ + "EliminationRuleId": "stale-after-variant-terminal", + "EvidenceArtifactId": value_field_string(&stage_fields, &["EvidenceArtifactId", "evidence_artifact_id"]), + "Reason": reason, + }), + observe_metadata, + ) + .await +} + +fn stale_directed_evolution_stage_work_reason( + work_item: &DirectedEvolutionWorkItemState, + stage_fields: &Value, + variant_fields: &Value, +) -> Option { + if !stale_stage_work_targets_stage_result(work_item) { + return None; + } + let stage_status = value_field_string(stage_fields, &["Status", "status"]); + if !stage_status.trim().is_empty() && stage_status != "Running" { + return Some(format!( + "Target StageResult {} is already {}; skipping stale {} work", + work_item.target_entity_id, stage_status, work_item.role + )); + } + let variant_status = value_field_string(variant_fields, &["Status", "status"]); + if matches!( + variant_status.as_str(), + "Eliminated" | "Promoted" | "Superseded" | "Failed" + ) { + return Some(format!( + "Target variant is already {}; skipping stale {} work for StageResult {}", + variant_status, work_item.role, work_item.target_entity_id + )); + } + None +} + +fn stale_stage_work_targets_stage_result(work_item: &DirectedEvolutionWorkItemState) -> bool { + directed_evolution_stage_evaluator_role(&work_item.role) + && work_item.target_entity_type == "StageResult" +} + +fn directed_evolution_stage_evaluator_role(role: &str) -> bool { + matches!( + role, + "reviewer" + | "viability_evaluator" + | "state_verifier" + | "telemetry_evaluator" + | "wasm_evaluator" + ) +} + +/// Cheap dispatch assertion: simulated users exercise Trials, evaluator +/// roles judge StageResults. A mismatch is a control-plane routing bug and +/// must fail the work item with a clear reason instead of confusing the +/// downstream brain. +fn directed_evolution_role_target_mismatch( + role: &str, + target_entity_type: &str, +) -> Option { + let expected = if role == "simulated_user" { + "Trial" + } else if directed_evolution_stage_evaluator_role(role) { + "StageResult" + } else { + return None; + }; + if target_entity_type == expected { + return None; + } + Some(format!( + "Directed Evolution role {role} requires a {expected} target, got {target_entity_type}" + )) +} + +/// Codex roles that exercise a variant runtime with bearer credentials. +/// The observer is excluded: it resolves runtime auth best-effort inside +/// its source inventory and degrades to other sources. Mechanical +/// evaluator roles never launch Codex and never authenticate. +fn directed_evolution_runtime_credential_role(role: &str) -> bool { + matches!( + role, + "simulated_user" | "reviewer" | "viability_evaluator" | "telemetry_evaluator" + ) +} + +/// B9 runtime auth preflight: when a work item's correlation names a +/// runtime to exercise, verify at least one of the runtime auth env vars +/// is set before launching Codex. Only env var NAMES are resolved and +/// reported — never values. A missing credential otherwise surfaces as a +/// confusing 401 inside the Codex child. +fn directed_evolution_runtime_credential_failure( + role: &str, + join: &DirectedEvolutionJoinFields, + env_value: impl Fn(&str) -> Option, +) -> Option { + if !directed_evolution_runtime_credential_role(role) { + return None; + } + if join.runtime_base_url.trim().is_empty() { + return None; + } + let names = directed_evolution_runtime_auth_env_var_names(&join.runtime_auth_env_vars); + let any_set = names.iter().any(|name| { + env_value(name) + .map(|value| !value.trim().is_empty()) + .unwrap_or(false) + }); + if any_set { + return None; + } + Some(format!( + "runtime credential missing: none of [{}] is set", + names.join(", ") + )) +} + +fn directed_evolution_runtime_auth_env_var_names(configured: &[String]) -> Vec { + let mut names: Vec = configured + .iter() + .map(|name| name.trim().to_string()) + .filter(|name| !name.is_empty()) + .collect(); + for fallback in ["TEMPERPAW_RUNTIME_API_KEY", "TEMPER_API_KEY"] { + if !names.iter().any(|name| name == fallback) { + names.push(fallback.to_string()); + } + } + names +} + +fn stale_stage_result_should_eliminate(stage_fields: &Value) -> bool { + matches!( + value_field_string(stage_fields, &["Status", "status"]).as_str(), + "Running" | "Failed" + ) +} +