feat(workflows): M4 — nested sub-workflows (start_child, parent-park, depth ceiling)

A `workflow`-kind step now starts a nested sub-workflow run instead of a
function. The orchestrator's step processing is restructured into prepare →
dispatch (`Prep`): after the shared context/`when`/input resolution, a
function step runs through the executor as before, while a workflow step:

- checks the nesting depth ceiling (`PICLOUD_WORKFLOW_MAX_DEPTH`; the child
  would run at parent depth + 1) — over-deep → the step fails, not infinite;
- resolves the child workflow by name in the run's app scope;
- `start_child_and_park`: in one token-gated tx, seeds a child run (depth + 1,
  `parent_run_id`/`parent_step_id` linkage, correlated under the same
  `root_execution_id`) and PARKS the parent step (`running`, claim_token
  cleared, `child_run_id` set). A parked step is never re-claimed (only `ready`
  steps are) nor reclaimed (only *leased* running steps). The token-gated park
  runs before the child insert, so a stale claim writes nothing (no orphan).

Each tick first runs `resume_finished_children`: a parent step parked on a
now-terminal child is resolved (child output → parent step output, or child
error → parent step failed) and the parent run advanced — idempotent via a
`status='running'` gate. The child's own steps are claimed by the same global
scan, so nesting is just more runs. A failed sub-workflow honors the parent
step's `on_error`.

Shared plumbing extracted for reuse: `advance_run_tx` (out of
`complete_step_and_advance`) and `seed_run_tx` (out of `start_run`).

Apply-time soft-warn (`workflow_nesting_warnings`, wired into `plan_warnings`):
a workflow that nests into itself — directly or via a mutual cycle within the
bundle — is flagged at plan (bounded by the depth ceiling, but almost always a
bug). Not an error.

Tests: nested output-flows-to-parent + depth-ceiling-fails-the-run (DB-gated,
end-to-end), self/mutual-cycle warning (pure unit). fmt + clippy -D warnings
clean, 449 manager-core lib tests, 13 orchestrator DB tests, 26 CLI journeys
(workflows/apply/plan/prune) green, binary boots.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

crates/manager-core/tests/workflow_orchestrator.rs

@@ -616,12 +616,13 @@ async fn seed_scripts(pool: &PgPool, app: AppId, names: &[&str]) {
     }
 }
 
-/// Drive a real orchestrator to terminal using `executor`.
-async fn drive_to_terminal(
+/// Drive a real orchestrator to terminal using `executor` + `config`.
+async fn drive_with(
     pool: &PgPool,
     executor: Arc<dyn ExecutorClient>,
     app: AppId,
     run: picloud_shared::WorkflowRunId,
+    config: WorkflowOrchestratorConfig,
 ) {
     let orch = WorkflowOrchestrator {
         pool: pool.clone(),
@@ -629,9 +630,9 @@ async fn drive_to_terminal(
         executor,
         gate: Arc::new(ExecutionGate::new(8)),
         log_sink: Arc::new(NoopLogSink),
-        config: WorkflowOrchestratorConfig::default(),
+        config,
     };
-    for _ in 0..50 {
+    for _ in 0..80 {
         orch.tick_once().await;
         if get_run(pool, app, run)
             .await
@@ -645,6 +646,76 @@ async fn drive_to_terminal(
     }
 }
 
+/// Drive to terminal with the default config.
+async fn drive_to_terminal(
+    pool: &PgPool,
+    executor: Arc<dyn ExecutorClient>,
+    app: AppId,
+    run: picloud_shared::WorkflowRunId,
+) {
+    drive_with(
+        pool,
+        executor,
+        app,
+        run,
+        WorkflowOrchestratorConfig::default(),
+    )
+    .await;
+}
+
+/// A `workflow`-kind (nested sub-workflow) step.
+fn wf_step(name: &str, workflow: &str, deps: &[&str]) -> WorkflowStepDef {
+    WorkflowStepDef {
+        name: name.into(),
+        function: None,
+        workflow: Some(workflow.into()),
+        input: Value::Null,
+        depends_on: deps.iter().map(|s| (*s).to_string()).collect(),
+        when: None,
+        retry: None,
+        on_error: OnError::Fail,
+    }
+}
+
+/// Create just a group + app; return the app id.
+async fn seed_app(pool: &PgPool) -> AppId {
+    let sfx = Uuid::new_v4().simple().to_string();
+    let grp: (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("wf-grp-{sfx}"))
+            .fetch_one(pool)
+            .await
+            .expect("insert group");
+    let app: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("wf-app-{sfx}"))
+            .bind(grp.0)
+            .fetch_one(pool)
+            .await
+            .expect("insert app");
+    app.0.into()
+}
+
+/// Insert a workflow with a specific name into an existing app.
+async fn seed_named_workflow(
+    pool: &PgPool,
+    app: AppId,
+    name: &str,
+    def: &WorkflowDefinition,
+) -> WorkflowId {
+    let def_json = serde_json::to_value(def).unwrap();
+    let wf: (Uuid,) = sqlx::query_as(
+        "INSERT INTO workflows (app_id, name, definition) VALUES ($1, $2, $3) RETURNING id",
+    )
+    .bind(app.into_inner())
+    .bind(name)
+    .bind(def_json)
+    .fetch_one(pool)
+    .await
+    .expect("insert named workflow");
+    wf.0.into()
+}
+
 fn run_with_input(app: AppId, wf: WorkflowId, input: Value) -> NewWorkflowRun {
     NewWorkflowRun {
         workflow_id: wf,
@@ -762,3 +833,97 @@ async fn missing_input_ref_fails_the_step() {
     assert_eq!(bstep.status, StepStatus::Failed);
     assert!(bstep.error.as_deref().unwrap_or("").contains("input"));
 }
+
+// ===========================================================================
+// M4 — nested sub-workflows.
+// ===========================================================================
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn nested_sub_workflow_runs_and_output_flows_to_parent() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let app = seed_app(&pool).await;
+
+    // Child workflow: one function step producing { answer: 42 }.
+    let child_def = WorkflowDefinition {
+        steps: vec![step("leaf", "fn_leaf", &[])],
+    };
+    seed_named_workflow(&pool, app, "child", &child_def).await;
+
+    // Parent: a `workflow`-kind step "call" → child, then "after" consuming its
+    // output. The child run's output is { leaf: <leaf output> }, so
+    // steps.call.output.leaf.answer reaches through the nesting.
+    let mut after = step("after", "fn_after", &["call"]);
+    after.input = json!({ "got": "{{ steps.call.output.leaf.answer }}" });
+    let parent_def = WorkflowDefinition {
+        steps: vec![wf_step("call", "child", &[]), after],
+    };
+    let parent_wf = seed_named_workflow(&pool, app, "parent", &parent_def).await;
+    seed_scripts(&pool, app, &["fn_leaf", "fn_after"]).await;
+
+    let run = start_run(&pool, new_run(app, parent_wf), &parent_def)
+        .await
+        .expect("start");
+
+    let executor = Arc::new(ScriptedExecutor::new(&[(
+        "fn_leaf",
+        json!({ "answer": 42 }),
+    )]));
+    drive_with(
+        &pool,
+        executor.clone(),
+        app,
+        run,
+        WorkflowOrchestratorConfig::default(),
+    )
+    .await;
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+    // The child's output flowed into the parent step, then into `after`'s input.
+    assert_eq!(executor.body_for("fn_after"), Some(json!({ "got": 42 })));
+    // The parent step "call" carries the child run's output.
+    let steps = list_run_steps(&pool, run).await.unwrap();
+    let call = steps.iter().find(|s| s.step_name == "call").unwrap();
+    assert_eq!(call.status, StepStatus::Succeeded);
+    assert_eq!(call.output, Some(json!({ "leaf": { "answer": 42 } })));
+    assert!(
+        call.child_run_id.is_some(),
+        "parent step links its child run"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn nesting_depth_ceiling_fails_the_run() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let app = seed_app(&pool).await;
+
+    // A workflow that nests into ITSELF — bounded only by the depth ceiling.
+    let def = WorkflowDefinition {
+        steps: vec![wf_step("recurse", "loopwf", &[])],
+    };
+    let wf = seed_named_workflow(&pool, app, "loopwf", &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    // max_depth = 2 → depth 0 and 1 spawn children; depth 2's spawn is refused.
+    let config = WorkflowOrchestratorConfig {
+        max_depth: 2,
+        ..WorkflowOrchestratorConfig::default()
+    };
+    let executor = Arc::new(ScriptedExecutor::new(&[]));
+    drive_with(&pool, executor, app, run, config).await;
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(
+        r.status,
+        RunStatus::Failed,
+        "self-nesting run fails at the depth ceiling, not infinitely"
+    );
+}