refactor(workflow): stage runtime input readiness advancement

Add a workflow-service scheduler orchestrator path that advances dependent runtime inference tasks from AwaitingInputs to WaitingDependencyReadiness only after connected upstream task results materialize.

Keep runtime dispatch disabled in this slice: no Ready detour, no handoff synthesis, no graph paths, and no ModelRefV2 or ModelDependencyRequest adaptation. Tests cover materialized and blocked runtime inputs, and the plan records the dedicated session/runtime runner follow-up.

Author: MrScripty

crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs

@@ -423,6 +423,111 @@ impl WorkflowSchedulerTaskOrchestrator {
         }
     }
 
+    #[allow(dead_code)]
+    pub(crate) fn advance_awaiting_runtime_task_inputs(
+        &self,
+        store: &mut WorkflowExecutionSessionStore,
+        session_id: &str,
+        workflow_run_id: &str,
+        task_id: &str,
+    ) -> Result<Option<SchedulerTaskStateRecord>, WorkflowSchedulerTaskOrchestratorError> {
+        let (task_graph, records) = store
+            .active_run_scheduler_task_state(session_id, workflow_run_id)
+            .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)?
+            .ok_or_else(|| {
+                WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+                    WorkflowServiceError::InvalidRequest(format!(
+                        "active workflow run '{}' has no scheduler task graph",
+                        workflow_run_id
+                    )),
+                )
+            })?;
+        let task = task_graph
+            .tasks
+            .iter()
+            .find(|task| task.task_id.as_str() == task_id)
+            .ok_or_else(|| {
+                WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+                    WorkflowServiceError::InvalidRequest(format!(
+                        "scheduler task '{}' is not in active workflow run '{}'",
+                        task_id, workflow_run_id
+                    )),
+                )
+            })?;
+        if task.execution_class != WorkflowSchedulerTaskExecutionClass::RuntimeInference {
+            return Err(WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+                WorkflowServiceError::InvalidRequest(format!(
+                    "scheduler task '{}' is not a runtime inference task",
+                    task_id
+                )),
+            ));
+        }
+        let record = records
+            .iter()
+            .find(|record| record.task_id.as_str() == task_id)
+            .ok_or_else(|| {
+                WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+                    WorkflowServiceError::InvalidRequest(format!(
+                        "scheduler task '{}' has no active task-state record",
+                        task_id
+                    )),
+                )
+            })?;
+        if record.state.kind() != SchedulerTaskStateKind::AwaitingInputs {
+            return Err(WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+                WorkflowServiceError::InvalidRequest(format!(
+                    "scheduler task '{}' must be awaiting inputs before runtime readiness advancement",
+                    task_id
+                )),
+            ));
+        }
+
+        let results = store
+            .active_run_scheduler_task_results(session_id, workflow_run_id)
+            .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)?;
+        match runtime_input_readiness(task, &results) {
+            RuntimeInputReadiness::Blocked => Ok(None),
+            RuntimeInputReadiness::Ready => {
+                let transition =
+                    waiting_dependency_readiness_transition_from_awaiting_inputs(task, record)?;
+                store
+                    .apply_active_run_scheduler_task_transition(
+                        session_id,
+                        workflow_run_id,
+                        transition,
+                    )
+                    .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)
+                    .and_then(applied_task_state_record)
+                    .map(Some)
+            }
+            RuntimeInputReadiness::InputUnavailable(diagnostic) => {
+                let transition =
+                    input_unavailable_transition_from_awaiting_inputs(record, diagnostic)?;
+                store
+                    .apply_active_run_scheduler_task_transition(
+                        session_id,
+                        workflow_run_id,
+                        transition,
+                    )
+                    .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)
+                    .and_then(applied_task_state_record)
+                    .map(Some)
+            }
+            RuntimeInputReadiness::Invalid(diagnostic) => {
+                let transition = invalid_transition_from_awaiting_inputs(record, diagnostic)?;
+                store
+                    .apply_active_run_scheduler_task_transition(
+                        session_id,
+                        workflow_run_id,
+                        transition,
+                    )
+                    .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)
+                    .and_then(applied_task_state_record)
+                    .map(Some)
+            }
+        }
+    }
+
     #[allow(dead_code)]
     pub(crate) fn apply_runtime_dependency_readiness_admission(
         &self,
@@ -941,6 +1046,21 @@ fn ready_transition_from_awaiting_inputs(
     )
 }
 
+#[allow(dead_code)]
+fn waiting_dependency_readiness_transition_from_awaiting_inputs(
+    task: &WorkflowSchedulerTask,
+    record: &SchedulerTaskStateRecord,
+) -> Result<SchedulerTaskStateTransition, WorkflowSchedulerTaskOrchestratorError> {
+    task_state_transition(
+        record,
+        "runtime-inputs-ready",
+        SchedulerTaskStateKind::AwaitingInputs,
+        SchedulerTaskState::WaitingDependencyReadiness {
+            execution_intent: runtime_execution_intent(task)?,
+        },
+    )
+}
+
 fn input_unavailable_transition_from_awaiting_inputs(
     record: &SchedulerTaskStateRecord,
     diagnostic: SchedulerTaskStateDiagnostic,
@@ -1163,6 +1283,40 @@ enum NonRuntimeInputReadiness {
     Invalid(SchedulerTaskStateDiagnostic),
 }
 
+#[allow(dead_code)]
+enum RuntimeInputReadiness {
+    Ready,
+    Blocked,
+    InputUnavailable(SchedulerTaskStateDiagnostic),
+    Invalid(SchedulerTaskStateDiagnostic),
+}
+
+#[allow(dead_code)]
+fn runtime_input_readiness(
+    task: &WorkflowSchedulerTask,
+    results: &[WorkflowSchedulerTaskResult],
+) -> RuntimeInputReadiness {
+    if task.schedulable_intent.is_none() {
+        return RuntimeInputReadiness::Invalid(scheduler_input_diagnostic(
+            SchedulerTaskStateDiagnosticCode::InvalidTask,
+            "runtime scheduler task is missing a typed runtime execution intent",
+        ));
+    }
+    for binding in &task.input_bindings {
+        match materialized_bound_output(task, results, binding) {
+            MaterializedBindingValue::Ready(_) => {}
+            MaterializedBindingValue::Blocked => return RuntimeInputReadiness::Blocked,
+            MaterializedBindingValue::Unavailable(diagnostic) => {
+                return RuntimeInputReadiness::InputUnavailable(diagnostic);
+            }
+            MaterializedBindingValue::Invalid(diagnostic) => {
+                return RuntimeInputReadiness::Invalid(diagnostic);
+            }
+        }
+    }
+    RuntimeInputReadiness::Ready
+}
+
 fn non_runtime_input_readiness(
     task: &WorkflowSchedulerTask,
     results: &[WorkflowSchedulerTaskResult],
@@ -1198,6 +1352,21 @@ fn non_runtime_input_readiness(
     }
 }
 
+#[allow(dead_code)]
+fn runtime_execution_intent(
+    task: &WorkflowSchedulerTask,
+) -> Result<SchedulerTaskExecutionIntent, WorkflowSchedulerTaskOrchestratorError> {
+    let Some(task_intent) = task.schedulable_intent.clone() else {
+        return Err(WorkflowSchedulerTaskOrchestratorError::WorkflowService(
+            WorkflowServiceError::InvalidRequest(format!(
+                "runtime scheduler task '{}' is missing a typed runtime execution intent",
+                task.task_id.as_str()
+            )),
+        ));
+    };
+    Ok(SchedulerTaskExecutionIntent::Runtime { task_intent })
+}
+
 enum MaterializedBindingValue<'a> {
     Ready(&'a WorkflowSchedulerTaskResultValue),
     Blocked,

crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs

@@ -399,6 +399,84 @@ fn orchestrator_initializes_dependent_runtime_task_as_awaiting_inputs() {
     );
 }
 
+#[test]
+fn orchestrator_advances_dependent_runtime_task_when_inputs_materialize() {
+    let orchestrator = orchestrator_without_runtime_host_response();
+    let task_intent = runtime_host_request_fixture().handoff.task_intent;
+    let task_id = task_intent.task_id.as_str().to_string();
+    let mut task = task_from_intent(task_intent.clone());
+    task.dependency_task_ids = vec![SchedulerTaskId::parse("prompt").expect("task id")];
+    task.input_bindings = vec![text_binding("prompt", task.task_id.as_str())];
+    let source = text_input_task_for_runtime_intent(&task_intent, "prompt");
+    let task_graph = task_graph(vec![source, task]);
+    let workflow_run_id = task_graph.workflow_run_id.as_str().to_string();
+    let mut store = WorkflowExecutionSessionStore::new(1, 1);
+    let session_id = begin_active_run_for_task_graph(&mut store, &task_graph);
+    orchestrator
+        .initialize_active_run_task_state(
+            &mut store,
+            &session_id,
+            &workflow_run_id,
+            task_graph.clone(),
+        )
+        .expect("initialize active run task state");
+    store
+        .record_active_run_scheduler_task_result(
+            &session_id,
+            &workflow_run_id,
+            text_result_for_runtime_intent(
+                &task_intent,
+                "prompt",
+                WorkflowSchedulerTaskResultStatus::Completed,
+            ),
+        )
+        .expect("record prompt result");
+
+    let advanced = orchestrator
+        .advance_awaiting_runtime_task_inputs(&mut store, &session_id, &workflow_run_id, &task_id)
+        .expect("advance runtime task")
+        .expect("runtime task should advance");
+
+    assert_eq!(advanced.state_version, 2);
+    let SchedulerTaskState::WaitingDependencyReadiness { execution_intent } = advanced.state else {
+        panic!("expected waiting dependency readiness");
+    };
+    assert!(execution_intent.runtime_task_intent().is_some());
+}
+
+#[test]
+fn orchestrator_leaves_dependent_runtime_task_blocked_without_materialized_input() {
+    let orchestrator = orchestrator_without_runtime_host_response();
+    let task_intent = runtime_host_request_fixture().handoff.task_intent;
+    let task_id = task_intent.task_id.as_str().to_string();
+    let mut task = task_from_intent(task_intent.clone());
+    task.dependency_task_ids = vec![SchedulerTaskId::parse("prompt").expect("task id")];
+    task.input_bindings = vec![text_binding("prompt", task.task_id.as_str())];
+    let source = text_input_task_for_runtime_intent(&task_intent, "prompt");
+    let task_graph = task_graph(vec![source, task]);
+    let workflow_run_id = task_graph.workflow_run_id.as_str().to_string();
+    let mut store = WorkflowExecutionSessionStore::new(1, 1);
+    let session_id = begin_active_run_for_task_graph(&mut store, &task_graph);
+    orchestrator
+        .initialize_active_run_task_state(&mut store, &session_id, &workflow_run_id, task_graph)
+        .expect("initialize active run task state");
+
+    let advanced = orchestrator
+        .advance_awaiting_runtime_task_inputs(&mut store, &session_id, &workflow_run_id, &task_id)
+        .expect("advance runtime task");
+
+    assert!(advanced.is_none());
+    let (_stored_graph, records) = store
+        .active_run_scheduler_task_state(&session_id, &workflow_run_id)
+        .expect("active run task state")
+        .expect("stored task state");
+    let record = records
+        .iter()
+        .find(|record| record.task_id.as_str() == task_id)
+        .expect("runtime task record");
+    assert_eq!(record.state.kind(), SchedulerTaskStateKind::AwaitingInputs);
+}
+
 #[test]
 fn orchestrator_initializes_awaiting_inputs_for_pre_intent_task() {
     let orchestrator = orchestrator_without_runtime_host_response();
@@ -1328,6 +1406,16 @@ fn text_input_task(task_id: &str, _value: &str) -> WorkflowSchedulerTask {
     }
 }
 
+fn text_input_task_for_runtime_intent(
+    task_intent: &SchedulableTaskIntent,
+    task_id: &str,
+) -> WorkflowSchedulerTask {
+    let mut task = text_input_task(task_id, "paint a red cube");
+    task.workflow_id = task_intent.workflow_id.clone();
+    task.workflow_run_id = task_intent.workflow_run_id.clone();
+    task
+}
+
 fn text_output_task() -> WorkflowSchedulerTask {
     WorkflowSchedulerTask {
         workflow_id: scheduler_workflow_id(),
@@ -1392,6 +1480,17 @@ fn text_result(
     )
 }
 
+fn text_result_for_runtime_intent(
+    task_intent: &SchedulableTaskIntent,
+    task_id: &str,
+    status: WorkflowSchedulerTaskResultStatus,
+) -> WorkflowSchedulerTaskResult {
+    let mut result = text_result(task_id, status);
+    result.workflow_id = task_intent.workflow_id.as_str().to_string();
+    result.workflow_run_id = task_intent.workflow_run_id.as_str().to_string();
+    result
+}
+
 fn bool_result(
     task_id: &str,
     status: WorkflowSchedulerTaskResultStatus,

docs/plans/current-image-generation-graphs/05-execution-management.md

@@ -16406,6 +16406,35 @@ Worker rules:
     fixture value.
   - Remaining follow-up: add workflow-service runtime input advancement using
     this scheduler-owned transition contract.
+- 2026-05-29 Milestone 5b workflow-service runtime input advancement slice
+  completed:
+  - Slice scope: workflow-service scheduler task orchestrator/tests plus plan
+    records.
+  - Implementation: added an orchestrator path that advances dependent runtime
+    inference tasks from `AwaitingInputs` to `WaitingDependencyReadiness` only
+    when connected upstream scheduler task results have materialized. Missing
+    inputs leave the task blocked in `AwaitingInputs`; unavailable or invalid
+    materialized inputs produce typed scheduler diagnostics.
+  - No-fallback/no-legacy gate: this does not dispatch runtime work, does not
+    route runtime inference through `Ready`, does not synthesize handoff, does
+    not read graph paths or executable load targets, and does not adapt
+    readiness into `ModelRefV2` or `ModelDependencyRequest`.
+  - Verification passed: `cargo fmt -p pantograph-workflow-service`; `cargo
+    test -p pantograph-workflow-service task_orchestrator --lib --
+    --nocapture`; `cargo check -p pantograph-workflow-service`; `cargo fmt -p
+    pantograph-workflow-service -- --check`; `cargo check -p
+    pantograph-workflow-service --all-features`; `cargo check -p
+    pantograph-workflow-service --no-default-features`; targeted retired
+    path/model-ref source search over touched workflow-service scheduler files
+    and the session API; and `git diff --check`.
+  - Verification caveat: `cargo check -p pantograph-workflow-service` still
+    emits the known unused `set_active_run_execution_plan` warning.
+  - Deviation recorded: session-runner wiring was attempted and reverted
+    because it widened existing session behavior before the dedicated runtime
+    runner boundary was planned and tested.
+  - Remaining follow-up: implement the dedicated session/runtime runner slice
+    so upstream result recording invokes this advancement path without
+    reviving old planned-inference launch behavior.
 
 ### Traceability Links
 

docs/plans/current-image-generation-graphs/milestones/05b-runtime-host-handoff-legacy-removal.md

@@ -82,6 +82,13 @@ this transition only in workflow-service; the legal lifecycle belongs in the
   must not be used to launch inference or build handoff. Runtime requests must
   include the canonical dependency readiness proof and workflow-service-owned
   materialized runtime inputs derived from validated upstream task results.
+- [ ] Wire the session/runtime runner to call workflow-service runtime input
+  advancement after upstream task results are recorded. This must be a
+  dedicated runner slice, because direct wiring through the existing
+  fail-closed runtime session branch changes legacy/session expectations and
+  risks reintroducing broad compatibility behavior. The runner must keep graph
+  editing, validation, dependency readiness, runtime input materialization, and
+  runtime-host dispatch as separate boundaries.
 - [ ] Retire node-engine planned-inference launch ownership for runtime
   inference nodes. Affected nodes must submit or reference scheduler task
   intent and consume scheduler task state/results; missing scheduler task state
@@ -363,3 +370,31 @@ this transition only in workflow-service; the legal lifecycle belongs in the
   path-shaped-field rejection test and the existing typed Pumas fixture value.
   Remaining follow-up: retry workflow-service runtime input advancement on top
   of the scheduler-owned transition contract.
+- 2026-05-29 workflow-service runtime input advancement slice completed.
+  Smallest useful vertical slice: add a workflow-service orchestrator method
+  that advances a dependent runtime inference task from `AwaitingInputs` to
+  `WaitingDependencyReadiness` only after all connected upstream scheduler task
+  results have materialized. Allowed write set:
+  `crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs`,
+  `crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs`,
+  this milestone file, and execution notes. No-fallback confirmation: the
+  slice does not dispatch runtime work, does not route runtime tasks through
+  `Ready`, does not synthesize handoff, does not read graph paths or executable
+  load targets, and does not adapt runtime readiness into `ModelRefV2` or
+  `ModelDependencyRequest`. Blocked inputs return no transition; unavailable
+  or invalid materialized inputs move through typed scheduler diagnostics.
+  Verification passed: `cargo fmt -p pantograph-workflow-service`; `cargo
+  test -p pantograph-workflow-service task_orchestrator --lib --
+  --nocapture`; `cargo check -p pantograph-workflow-service`; `cargo fmt -p
+  pantograph-workflow-service -- --check`; `cargo check -p
+  pantograph-workflow-service --all-features`; `cargo check -p
+  pantograph-workflow-service --no-default-features`; targeted retired
+  path/model-ref source search over touched workflow-service scheduler files
+  and the session API; and `git diff --check`. Verification caveat: `cargo
+  check -p pantograph-workflow-service` still emits the known unused
+  `set_active_run_execution_plan` warning. Deviation recorded: session-runner
+  wiring was attempted and reverted because it widened existing session
+  behavior before the dedicated runtime runner boundary was planned and tested.
+  Remaining follow-up: implement the dedicated session/runtime runner slice so
+  upstream result recording invokes this advancement path without reviving old
+  planned-inference launch behavior.