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 {