Use one authoritative workflow ID across dispatch

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<QueueSubjectId> = 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<String> = (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) => {

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::<Vec<_>>();
+        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() {

crates/orchestrator-daemon-runtime/src/dispatch/dispatch_execution.rs

@@ -24,15 +24,21 @@ 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(),
                     selection_source: planned_start.selection_source,
                 });
                 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() };

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::{