diff --git a/crates/orchestrator-cli/src/services/runtime/runtime_daemon/daemon_task_dispatch.rs b/crates/orchestrator-cli/src/services/runtime/runtime_daemon/daemon_task_dispatch.rs index 827bc804..dfafced6 100644 --- a/crates/orchestrator-cli/src/services/runtime/runtime_daemon/daemon_task_dispatch.rs +++ b/crates/orchestrator-cli/src/services/runtime/runtime_daemon/daemon_task_dispatch.rs @@ -23,9 +23,13 @@ pub async fn dispatch_queued_entries_via_runner( let project_root_path = std::path::Path::new(root); let exclude_subjects: Vec = active_subject_ids.iter().cloned().map(QueueSubjectId::new).collect(); + // TASK-1072: mint the durable ids BEFORE the atomic lease. The queue stores + // these exact ids and the runner creates/loads the journal rows with them, + // eliminating the previous queue-UUID / runner-UUID split. + let workflow_ids: Vec = (0..limit).map(|_| uuid::Uuid::new_v4().to_string()).collect(); let lease_req = QueueLeaseRequest { max: limit, - workflow_ids: None, + workflow_ids: Some(workflow_ids), exclude_subjects: if exclude_subjects.is_empty() { None } else { Some(exclude_subjects) }, }; match plugin_clients::call_queue_lease(project_root_path, &lease_req).await { @@ -47,6 +51,15 @@ pub async fn dispatch_queued_entries_via_runner( continue; } }; + let Some(workflow_id) = entry.workflow_id.clone() else { + warn!( + actor = protocol::ACTOR_DAEMON, + entry_id = %entry.entry_id, + "queue/lease returned an assigned entry without workflow_id; closing it as failed" + ); + undecodable_entry_ids.push(entry.entry_id.clone()); + continue; + }; // Within-batch dedupe: the queue's `exclude_subjects` filter // honors the snapshot we sent at lease time, but multiple // pending entries for the same subject (different @@ -74,8 +87,11 @@ pub async fn dispatch_queued_entries_via_runner( plugin_owned_subject_keys.insert(subject_key); } leased_entry_ids.push(entry.entry_id.clone()); - planned_starts - .push(PlannedDispatchStart { dispatch, selection_source: DispatchSelectionSource::DispatchQueue }); + planned_starts.push(PlannedDispatchStart { + dispatch, + workflow_id: Some(workflow_id), + selection_source: DispatchSelectionSource::DispatchQueue, + }); } } Ok(None) => { diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs b/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs index 8d885a14..4db34a74 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/build_runner_command_from_dispatch.rs @@ -13,7 +13,19 @@ pub fn build_runner_command( phase_routing: Option<&PhaseRoutingConfig>, mcp_config: Option<&McpRuntimeConfig>, ) -> std::process::Command { - build_runner_command_with_resume(dispatch, project_root, phase_routing, mcp_config, None) + build_runner_command_with_target(dispatch, project_root, phase_routing, mcp_config, None, None) +} + +/// Build a fresh runner command whose workflow row is created idempotently +/// with the kernel-selected `workflow_id`. +pub fn build_runner_command_with_id( + dispatch: &SubjectDispatch, + project_root: &str, + phase_routing: Option<&PhaseRoutingConfig>, + mcp_config: Option<&McpRuntimeConfig>, + workflow_id: &str, +) -> std::process::Command { + build_runner_command_with_target(dispatch, project_root, phase_routing, mcp_config, Some(workflow_id), None) } /// Build the `execute` command for the workflow runner. @@ -31,6 +43,18 @@ pub fn build_runner_command_with_resume( mcp_config: Option<&McpRuntimeConfig>, resume_workflow_id: Option<&str>, ) -> std::process::Command { + build_runner_command_with_target(dispatch, project_root, phase_routing, mcp_config, None, resume_workflow_id) +} + +fn build_runner_command_with_target( + dispatch: &SubjectDispatch, + project_root: &str, + phase_routing: Option<&PhaseRoutingConfig>, + mcp_config: Option<&McpRuntimeConfig>, + new_workflow_id: Option<&str>, + resume_workflow_id: Option<&str>, +) -> std::process::Command { + debug_assert!(new_workflow_id.is_none() || resume_workflow_id.is_none()); let mut cmd = std::process::Command::new(resolve_workflow_runner_binary_for(Some(project_root))); cmd.arg("execute"); @@ -72,7 +96,9 @@ pub fn build_runner_command_with_resume( } } - if let Some(workflow_id) = resume_workflow_id { + if let Some(workflow_id) = new_workflow_id { + cmd.arg("--new-workflow-id").arg(workflow_id); + } else if let Some(workflow_id) = resume_workflow_id { cmd.arg("--workflow-id").arg(workflow_id); } @@ -478,7 +504,16 @@ mod tests { use protocol::{SubjectDispatch, SubjectDispatchExt}; use serde_json::json; - use super::build_runner_command_from_dispatch; + use super::{build_runner_command_from_dispatch, build_runner_command_with_id}; + + #[test] + fn queue_dispatch_passes_kernel_selected_workflow_id_as_fresh_bootstrap() { + let dispatch = SubjectDispatch::for_task("TASK-ONE-ID", "coding"); + let command = build_runner_command_with_id(&dispatch, "/tmp/project", None, None, "wf-authoritative"); + let args = command.get_args().map(|arg| arg.to_string_lossy().into_owned()).collect::>(); + assert!(args.windows(2).any(|pair| pair == ["--new-workflow-id", "wf-authoritative"])); + assert!(!args.iter().any(|arg| arg == "--workflow-id")); + } #[test] fn runner_command_uses_subject_workflow_ref_and_input_from_dispatch() { diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/dispatch_execution.rs b/crates/orchestrator-daemon-runtime/src/dispatch/dispatch_execution.rs index 1a88bf7c..2da9f671 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/dispatch_execution.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/dispatch_execution.rs @@ -24,7 +24,13 @@ where break; } - match process_manager.spawn_workflow_runner(&planned_start.dispatch, project_root) { + let spawn = match planned_start.workflow_id.as_deref() { + Some(workflow_id) => { + process_manager.spawn_workflow_runner_with_id(&planned_start.dispatch, project_root, workflow_id) + } + None => process_manager.spawn_workflow_runner(&planned_start.dispatch, project_root), + }; + match spawn { Ok(()) => { notice_sink.notice(DispatchNotice::Started { dispatch: planned_start.dispatch.clone(), @@ -32,7 +38,7 @@ where }); started_workflows.push(DispatchWorkflowStart { dispatch: planned_start.dispatch.clone(), - workflow_id: None, + workflow_id: planned_start.workflow_id.clone(), selection_source: planned_start.selection_source, }); } @@ -79,6 +85,7 @@ mod tests { let mut manager = ProcessManager::new().with_workflow_concurrency_max(Some(0)); let starts = vec![PlannedDispatchStart { dispatch: SubjectDispatch::for_task("TASK-DEFER", "standard"), + workflow_id: Some("wf-defer".to_string()), selection_source: DispatchSelectionSource::DispatchQueue, }]; let mut sink = RecordingSink { notices: Vec::new() }; @@ -114,6 +121,7 @@ mod tests { let mut manager = ProcessManager::new().with_workflow_concurrency_max(None); let starts = vec![PlannedDispatchStart { dispatch: SubjectDispatch::for_task("TASK-FAIL", "standard"), + workflow_id: Some("wf-fail".to_string()), selection_source: DispatchSelectionSource::DispatchQueue, }]; let mut sink = RecordingSink { notices: Vec::new() }; diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/mod.rs b/crates/orchestrator-daemon-runtime/src/dispatch/mod.rs index e4017074..3bd8948b 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/mod.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/mod.rs @@ -29,7 +29,8 @@ mod ready_dispatch_plan; pub mod reattach; pub use build_runner_command_from_dispatch::{ - build_runner_command, build_runner_command_from_dispatch, build_runner_command_with_resume, + build_runner_command, build_runner_command_from_dispatch, build_runner_command_with_id, + build_runner_command_with_resume, }; pub use completed_process::CompletedProcess; pub use completion_reconciliation_plan::{ diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs b/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs index 003037b8..98b677a3 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs @@ -12,6 +12,7 @@ use protocol::SubjectDispatch; use tokio::process::{Child, Command}; use tokio::task::JoinHandle; +use super::{build_runner_command_with_id, build_runner_command_with_resume}; #[cfg(unix)] use crate::control::WorkflowEventBroadcaster; use crate::dispatch::environment_broker::{ @@ -20,7 +21,7 @@ use crate::dispatch::environment_broker::{ }; #[cfg(unix)] use crate::dispatch::event_pipe::SubprocessEventPipe; -use crate::{build_runner_command_with_resume, CompletedProcess, RunnerEvent}; +use crate::{CompletedProcess, RunnerEvent}; #[cfg(unix)] #[allow(unsafe_code)] @@ -168,13 +169,11 @@ struct WorkflowProcess { event_pipe: Option, agent_session_id: Option, project_root: Option, - /// BU-4 journal-resume: the EXISTING workflow id this runner was spawned to - /// continue (via `execute --workflow-id`), if any. Used as the - /// `workflow_id` fallback in completion reconciliation so an early runner - /// exit (before any `workflow_events` is emitted) still fails/cancels the - /// TARGETED run rather than leaving it Running — which would otherwise let - /// the resume sweep re-dispatch it every tick in an endless loop. - resume_workflow_id: Option, + /// Kernel-selected workflow id for either idempotent fresh creation or + /// journal resume. Used as the completion fallback when the runner exits + /// before emitting an event, keeping queue and journal reconciliation on + /// the same identity. + target_workflow_id: Option, /// REQ-048 cross-phase environment broker: the broker `run_id` this spawn /// was bound to (the workflow run id), set ONLY when the dispatch routes to /// a non-local environment and a broker is wired. The daemon tears the run's @@ -302,7 +301,18 @@ impl ProcessManager { } pub fn spawn_workflow_runner(&mut self, dispatch: &SubjectDispatch, project_root: &str) -> Result<()> { - self.spawn_workflow_runner_inner(dispatch, project_root, None) + self.spawn_workflow_runner_inner(dispatch, project_root, None, None) + } + + /// Start a fresh queue-backed workflow using the kernel-selected durable id. + /// The runner creates the journal row idempotently with this exact value. + pub fn spawn_workflow_runner_with_id( + &mut self, + dispatch: &SubjectDispatch, + project_root: &str, + workflow_id: &str, + ) -> Result<()> { + self.spawn_workflow_runner_inner(dispatch, project_root, Some(workflow_id), None) } /// BU-4 journal-resume re-dispatch: spawn a runner that CONTINUES the @@ -316,13 +326,14 @@ impl ProcessManager { project_root: &str, resume_workflow_id: &str, ) -> Result<()> { - self.spawn_workflow_runner_inner(dispatch, project_root, Some(resume_workflow_id)) + self.spawn_workflow_runner_inner(dispatch, project_root, None, Some(resume_workflow_id)) } fn spawn_workflow_runner_inner( &mut self, dispatch: &SubjectDispatch, project_root: &str, + new_workflow_id: Option<&str>, resume_workflow_id: Option<&str>, ) -> Result<()> { if let Some(cap) = self.workflow_concurrency_max { @@ -331,13 +342,24 @@ impl ProcessManager { } } - let std_cmd = build_runner_command_with_resume( - dispatch, - project_root, - self.phase_routing.as_ref(), - self.mcp_config.as_ref(), - resume_workflow_id, - ); + debug_assert!(new_workflow_id.is_none() || resume_workflow_id.is_none()); + let target_workflow_id = new_workflow_id.or(resume_workflow_id); + let std_cmd = match new_workflow_id { + Some(workflow_id) => build_runner_command_with_id( + dispatch, + project_root, + self.phase_routing.as_ref(), + self.mcp_config.as_ref(), + workflow_id, + ), + None => build_runner_command_with_resume( + dispatch, + project_root, + self.phase_routing.as_ref(), + self.mcp_config.as_ref(), + resume_workflow_id, + ), + }; let command_line: Vec = std::iter::once(std_cmd.get_program().to_string_lossy().into_owned()) .chain(std_cmd.get_args().map(|a| a.to_string_lossy().into_owned())) .collect(); @@ -447,7 +469,7 @@ impl ProcessManager { // the run's SHARED node from the daemon instead of preparing its own. // Returns the broker run_id so the completion path can tear it down. let environment_run_id = - self.configure_environment_broker(dispatch, project_root, resume_workflow_id, &mut command); + self.configure_environment_broker(dispatch, project_root, target_workflow_id, &mut command); // Bind the subprocess workflow_events back-channel before fork so the // env var we set on the child points to a listener that's already @@ -531,7 +553,7 @@ impl ProcessManager { event_pipe, agent_session_id, project_root: Some(project_root_path), - resume_workflow_id: resume_workflow_id.map(String::from), + target_workflow_id: target_workflow_id.map(String::from), environment_run_id, }); @@ -544,19 +566,14 @@ impl ProcessManager { /// broker `run_id` (the workflow run id) when brokered, else `None` (the /// runner then uses its legacy owned-environment path). /// - /// The run_id must be STABLE across every phase of a workflow run so all - /// phases acquire the SAME node. It is keyed on the SUBJECT: every phase is a - /// separate dispatch for the same subject, so `subject_id()` is identical - /// across phases and known before phase 1. Keying on a workflow id does NOT - /// work — the runner mints its own id and `--workflow-id` means "resume an - /// EXISTING workflow", so a daemon-minted id cannot be handed down for a - /// fresh phase. A subjectless adhoc run falls back to the resume id / a fresh - /// mint (single-phase only — no cross-phase sharing without a subject). + /// The run_id must be STABLE across every phase. Queue-backed dispatch now + /// supplies the authoritative workflow id before spawn, so use it first; + /// legacy ad-hoc callers fall back to the subject or a fresh local id. fn configure_environment_broker( &self, dispatch: &SubjectDispatch, project_root: &str, - resume_workflow_id: Option<&str>, + target_workflow_id: Option<&str>, command: &mut Command, ) -> Option { let broker = self.environment_broker.as_ref()?; @@ -584,13 +601,12 @@ impl ProcessManager { if is_local_environment(&environment_id) { return None; } - let run_id = match dispatch.subject_id() { - Some(subject) if !subject.is_empty() => format!("run-{subject}"), - _ => match resume_workflow_id { - Some(id) => id.to_string(), - None => format!("wf-{}", uuid::Uuid::new_v4().simple()), - }, - }; + let run_id = target_workflow_id + .map(str::to_string) + .or_else(|| { + dispatch.subject_id().filter(|subject| !subject.is_empty()).map(|subject| format!("run-{subject}")) + }) + .unwrap_or_else(|| format!("wf-{}", uuid::Uuid::new_v4().simple())); broker.register_run(&run_id, project_root, &environment_id); command.env(ANIMUS_ENVIRONMENT_BROKER_SOCKET_ENV, broker.socket_path()); command.env(ANIMUS_ENVIRONMENT_BROKER_TOKEN_ENV, broker.token()); @@ -696,7 +712,7 @@ impl ProcessManager { subject_id: process.subject_key, subject_kind: Some(process.subject_kind), task_id: process.task_id, - workflow_id: process.resume_workflow_id.take(), + workflow_id: process.target_workflow_id.take(), workflow_ref: Some(process.workflow_ref), workflow_status: Some(WorkflowStatus::Failed), schedule_id: process.schedule_id, @@ -719,7 +735,7 @@ impl ProcessManager { subject_id: process.subject_key, subject_kind: Some(process.subject_kind), task_id: process.task_id, - workflow_id: process.resume_workflow_id.take(), + workflow_id: process.target_workflow_id.take(), workflow_ref: Some(process.workflow_ref), workflow_status: None, schedule_id: process.schedule_id, @@ -748,17 +764,30 @@ impl ProcessManager { cleanup_agent_record(&process); let exit_code = status.code(); let events = parse_runner_events(&process.stderr_lines); - // Prefer the runner-reported id; fall back to the resume - // target so an early exit before any event still reconciles - // the known run (BU-4: prevents endless re-dispatch). - let resume_target = process.resume_workflow_id.take(); - let workflow_id = latest_runner_workflow_id(&events).or_else(|| resume_target.clone()); + let workflow_target = process.target_workflow_id.take(); + let runner_workflow_id = latest_runner_workflow_id(&events); + let (workflow_id, identity_mismatch) = + reconcile_workflow_identity(workflow_target.as_deref(), runner_workflow_id.as_deref()); let mut workflow_status = latest_runner_workflow_status(&events); - let (success, failure_reason) = if status.success() { + let (mut success, mut failure_reason) = if status.success() { (true, None) } else { (false, Some(format!("workflow runner exited unsuccessfully with status {:?}", exit_code))) }; + if let Some(mismatch) = identity_mismatch { + tracing::error!( + target: "animus.runtime.workflow_identity", + kernel_workflow_id = workflow_target.as_deref().unwrap_or("-"), + runner_workflow_id = runner_workflow_id.as_deref().unwrap_or("-"), + "runner reported a workflow id that diverges from the authoritative dispatch id" + ); + success = false; + workflow_status = Some(WorkflowStatus::Failed); + failure_reason = Some(match failure_reason { + Some(existing) => format!("{existing}; {mismatch}"), + None => mismatch, + }); + } // BU-4 (codex P2): a RESUME-spawned runner that exits // non-zero WITHOUT reporting a workflow status (e.g. a bad // `--workflow-id`, a plugin/startup failure, an arg parse @@ -766,9 +795,9 @@ impl ProcessManager { // the TARGETED run as Failed. Otherwise the projector only // blocks the task, the workflow stays Running, and the // journal-resume sweep re-dispatches it every tick - // (livelock). Scoped to resume spawns so normal dispatch - // behavior is byte-identical. - if workflow_status.is_none() && !status.success() && resume_target.is_some() { + // (livelock). This is safe for both resume and preallocated + // fresh spawns because each has an authoritative journal id. + if workflow_status.is_none() && !status.success() && workflow_target.is_some() { workflow_status = Some(WorkflowStatus::Failed); } @@ -817,7 +846,7 @@ impl ProcessManager { subject_id: process.subject_key, subject_kind: Some(process.subject_kind), task_id: process.task_id, - workflow_id: process.resume_workflow_id.take(), + workflow_id: process.target_workflow_id.take(), workflow_ref: Some(process.workflow_ref), workflow_status: None, schedule_id: process.schedule_id, @@ -957,6 +986,25 @@ fn latest_runner_workflow_id(events: &[RunnerEvent]) -> Option { events.iter().rev().find_map(|event| event.workflow_id.clone()) } +/// Select the identity used for completion reconciliation. A queue/kernel +/// target is authoritative: runner output may confirm it but can never replace +/// it. Divergence fails closed and is returned as a diagnostic containing both +/// values. The runner id remains authoritative only for legacy ad-hoc spawns. +fn reconcile_workflow_identity( + workflow_target: Option<&str>, + runner_workflow_id: Option<&str>, +) -> (Option, Option) { + match (workflow_target, runner_workflow_id) { + (Some(target), Some(reported)) if target != reported => ( + Some(target.to_string()), + Some(format!("workflow identity mismatch: authoritative_id={target} runner_reported_id={reported}")), + ), + (Some(target), _) => (Some(target.to_string()), None), + (None, Some(reported)) => (Some(reported.to_string()), None), + (None, None) => (None, None), + } +} + fn latest_runner_workflow_status(events: &[RunnerEvent]) -> Option { events.iter().rev().find_map(|event| event.workflow_status) } @@ -1087,6 +1135,19 @@ mod tests { assert!(clean.contains("code=Some(0)")); } + #[test] + fn kernel_workflow_id_wins_and_divergence_fails_closed_with_both_ids() { + let (workflow_id, mismatch) = reconcile_workflow_identity(Some("queue-id"), Some("runner-id")); + assert_eq!(workflow_id.as_deref(), Some("queue-id")); + let mismatch = mismatch.expect("divergence must fail closed"); + assert!(mismatch.contains("authoritative_id=queue-id")); + assert!(mismatch.contains("runner_reported_id=runner-id")); + + let (legacy_id, mismatch) = reconcile_workflow_identity(None, Some("runner-only-id")); + assert_eq!(legacy_id.as_deref(), Some("runner-only-id")); + assert!(mismatch.is_none()); + } + #[test] fn capped_escaped_tail_bounds_memory_and_escapes_newlines() { // A pathological megabyte-long single stderr line must NOT be copied diff --git a/crates/orchestrator-daemon-runtime/src/dispatch/ready_dispatch_plan.rs b/crates/orchestrator-daemon-runtime/src/dispatch/ready_dispatch_plan.rs index c6fdc03f..b7360f43 100644 --- a/crates/orchestrator-daemon-runtime/src/dispatch/ready_dispatch_plan.rs +++ b/crates/orchestrator-daemon-runtime/src/dispatch/ready_dispatch_plan.rs @@ -5,6 +5,11 @@ use crate::DispatchSelectionSource; #[derive(Debug, Clone, PartialEq)] pub struct PlannedDispatchStart { pub dispatch: SubjectDispatch, + /// Kernel-selected durable workflow id. Queue-backed dispatches populate + /// this before leasing so the queue row, workflow journal, runner, and + /// environment broker all use one identity. `None` preserves the legacy + /// ad-hoc fresh-dispatch path. + pub workflow_id: Option, pub selection_source: DispatchSelectionSource, }