Child start turn: a fresh parent turn when a child start commits (#966)

docs/contracts/child-start-turn.md

@@ -0,0 +1,66 @@
+# Child start turn
+
+A workflow can start a child, get its committed identity in a fresh parent
+turn while the child still runs, do other work, and later receive the child's
+result (Server #966, the runtime side of sdk-php#185). No activity, timer,
+signal or query is needed to get that turn. This is a native capability;
+published PHP Server 2.x has no such mode. Workflow Core parity is
+workflow#989.
+
+## Command
+
+`start_child_workflow` takes an optional `await`:
+
+- `result` (the default, and the behavior when the field is absent): the
+  parent's next turn comes when the child resolves, or from other work. The
+  completion creates only the child's task.
+- `started`: the completion that commits `ChildWorkflowScheduled` and
+  `ChildRunStarted` also creates one new parent workflow task, with a fresh
+  task id, lease and claim. Its poll carries both committed events, so the
+  child call id, the child instance id and the initial child run id come from
+  canonical history. The child keeps running, and its resolution gives the
+  parent its usual later turn.
+
+Any other value, or `await` on another command type, is refused with 422 and
+the validation message `await must be result or started.` (or the field-scope
+message) before anything is written.
+
+## Turn rules
+
+- One turn per completion, however many children in it ask for one.
+- No turn when the same completion closes the parent (complete, fail or
+  continue as new), and none when the parent already has a ready or leased
+  workflow task, which delivers the same history.
+- The task payload names the wait `workflow_wait_kind: "child_started"`, with
+  `child_call_id`, `child_workflow_run_id`, `resume_source_kind:
+  "child_workflow_run"`, `resume_source_id`, `workflow_sequence` and
+  `workflow_event_type: "ChildRunStarted"` of the first such child.
+- The completion stays under its immutable task receipt: a duplicate
+  completion, redelivery or retry of the original task starts no second child
+  and creates no second turn, and the original claim stays closed.
+- A cancellation-scope checkpoint keeps the parent's claim, so it refuses a
+  child start with `await: "started"` (`child_start_turn_not_at_checkpoint`)
+  before any child starts.
+- Parent-close policy (Abandon included), cancellation propagation, deadlines,
+  retries, child continue-as-new (`child_call_id` stays; the initial run id is
+  frozen in `ChildRunStarted`) and namespace isolation are unchanged.
+
+## Advertisement
+
+Native advertises `child_workflow_start_turn: true` in cluster info
+`capabilities` and in worker `server_capabilities`. A runtime that does not
+advertise it (native 3.0.x, PHP 2.x) ignores unknown command fields, so an SDK
+must check the capability and refuse `await: "started"` itself before
+submitting the child.
+
+Tests (`tests/execution/child_start_turn.rs`,
+`tests/execution/scope_checkpoints.rs`) cover:
+
+- the flagged start, with its fresh claim and committed identity, the
+  duplicate completion, and child continue-as-new resolving the original call;
+- the default and explicit `result` with an unchanged task count;
+- several children in one completion, and a closing completion;
+- refusal of an unknown value and of the checkpoint case;
+- the advertisement.
+
+They run on SQLite, PostgreSQL and MySQL.

rust/src/runtime/children.rs

@@ -16,6 +16,19 @@ use chrono::{DateTime, Utc};
 use serde_json::{Value, json};
 use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
 
+/// `start_child_workflow.await` (Server #966): the default waits for the
+/// child's result; `started` also gives the parent a turn when the start
+/// commits.
+pub(super) const AWAIT_RESULT: &str = "result";
+pub(super) const AWAIT_STARTED: &str = "started";
+
+/// A started child: its first workflow task, and the start turn payload a
+/// parent gets when it asked for one.
+pub(super) struct StartedChild {
+    pub(super) task: String,
+    pub(super) turn: Value,
+}
+
 #[derive(Clone, Copy)]
 pub(super) enum ChildResolution<'a> {
     Completed(&'a str),
@@ -43,6 +56,7 @@ pub(super) fn validate_child(command: &Value) -> Result<()> {
             "retry_policy",
             "execution_timeout_seconds",
             "run_timeout_seconds",
+            "await",
         ],
     )?;
     if text(command, "workflow_type")?.len() > 191
@@ -297,7 +311,7 @@ where
         task_id: &str,
         command: &Value,
         sequence: i64,
-    ) -> Result<String> {
+    ) -> Result<StartedChild> {
         let parent_run = DB::string(parent, "id")?;
         let parent_instance = DB::string(parent, "workflow_instance_id")?;
         // Published PHP starts a child whatever else the parent has open:
@@ -382,7 +396,45 @@ where
             .bind(&parent_run)
             .execute(&mut **tx)
             .await?;
-        Ok(child_task)
+        // What a start turn names: this call, frozen at its first run.
+        let turn = json!({"workflow_wait_kind":"child_started","child_call_id":call,"child_workflow_run_id":child_run,
+            "resume_source_kind":"child_workflow_run","resume_source_id":child_run,"workflow_sequence":sequence,
+            "workflow_event_type":"ChildRunStarted"});
+        Ok(StartedChild { task: child_task, turn })
+    }
+
+    /// The parent's start turn (Server #966): one fresh workflow task, created
+    /// with the completion that committed an `await: "started"` child start,
+    /// unless that completion closed the parent. As for a child's resolution
+    /// (`child_parent_resume`), only a ready workflow task that replays the
+    /// parent can carry the committed `ChildWorkflowScheduled` and
+    /// `ChildRunStarted` history instead: an update task does not replay the
+    /// workflow, and a leased task may have read history before the start.
+    pub(super) async fn child_start_turn(
+        tx: &mut Transaction<'_, DB>,
+        run_id: &str,
+        task_id: &str,
+        queue: &str,
+        turn: Value,
+    ) -> Result<Option<String>> {
+        let open = Self::scalar("SELECT COUNT(*) FROM workflow_runs WHERE id=$1 AND status IN ('pending','running','waiting')")
+            .bind(run_id)
+            .fetch_one(&mut **tx)
+            .await?;
+        if open == 0 {
+            return Ok(None);
+        }
+        let ready = Self::query("SELECT payload FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status='ready' AND id<>$2 ORDER BY id LIMIT 1001")
+            .bind(run_id)
+            .bind(task_id)
+            .fetch_all(&mut **tx)
+            .await?;
+        for task in &ready {
+            if DB::document_row(task, "payload")?["workflow_update_id"].as_str().is_none() {
+                return Ok(None);
+            }
+        }
+        Self::create_task(tx, run_id, queue, "workflow", turn).await.map(Some)
     }
 
     pub(super) async fn complete_child(

rust/src/runtime/cluster_info.rs

@@ -61,6 +61,7 @@ pub(super) fn capabilities() -> Value {
         "activity_retry_policy",
         "child_workflow_retry_policy",
         "child_workflow_timeouts",
+        "child_workflow_start_turn",
         "parent_close_policy",
         "non_retryable_failures",
         "worker_sessions",
@@ -152,7 +153,9 @@ mod tests {
             "workflow_update_runtime_contract", "waterline_v1_projection", "embedded_v2_import"] {
             assert!(caps.get(name).is_some(), "{name}");
         }
-        assert_eq!(caps.as_object().unwrap().len(), 56);
+        // PHP's names plus native's child start turn (Server #966).
+        assert_eq!(caps.as_object().unwrap().len(), 57);
+        assert_eq!(caps["child_workflow_start_turn"], json!(true));
         assert_eq!(caps["response_compression"], json!([]));
         assert_eq!(caps["timer_runtime_contract"], json!(false));
         assert_eq!(caps["principal_attribution_contract"], json!(true));

rust/src/runtime/completion_rules.rs

@@ -494,6 +494,14 @@ fn controller_scopes(commands: &[Value]) -> Vec<(String, Vec<String>)> {
                 add(at(field), format!("{field} is only supported for schedule_activity commands."));
             }
         }
+        // Native's child start turn (Server #966): `result`, the default, or
+        // `started`.
+        if kind == "start_child_workflow"
+            && has_value(command, "await")
+            && !matches!(command["await"].as_str(), Some(super::children::AWAIT_RESULT | super::children::AWAIT_STARTED))
+        {
+            add(at("await"), "await must be result or started.".into());
+        }
         if has_value(command, "worker_session") && kind != "schedule_activity" {
             add(at("worker_session"), "worker_session is only supported for schedule_activity commands.".into());
         }
@@ -576,6 +584,8 @@ const FIELD_SCOPES: &[(&str, &[&str], &str)] = &[
         "failed_step_sequence identifies an uncaught activity failure and only applies to a fail_workflow command."),
     ("failed_activity_execution_id", &["fail_workflow"],
         "failed_activity_execution_id identifies an uncaught activity failure and only applies to a fail_workflow command."),
+    ("await", &["start_child_workflow"],
+        "await chooses whether a child start gives the parent its next turn when the start commits (started) or when the child resolves (result) and only applies to a start_child_workflow command."),
     ("parent_close_policy", &["start_child_workflow"],
         "parent_close_policy declares how a child workflow reacts when its parent closes and only applies to a start_child_workflow command."),
     ("cancellation_policy", &["start_child_workflow", "schedule_activity"],
@@ -760,7 +770,7 @@ const READS: &[(&str, &[&str])] = &[
         "cancellation_policy", "worker_session"]),
     ("start_timer", &["delay_seconds", "cancellation_cleanup"]),
     ("start_child_workflow", &["workflow_type", "arguments", "payload_codec", "connection", "queue", "retry_policy",
-        "execution_timeout_seconds", "run_timeout_seconds", "parent_close_policy", "cancellation_policy"]),
+        "execution_timeout_seconds", "run_timeout_seconds", "parent_close_policy", "cancellation_policy", "await"]),
     ("continue_as_new", &["arguments", "payload_codec", "workflow_type", "queue"]),
     ("complete_update", &["update_id", "result", "payload_codec"]),
     ("fail_update", &["update_id", "message", "exception_class", "exception_type", "non_retryable"]),
@@ -785,7 +795,7 @@ const SCOPED_FIELDS: &[&str] = &["retry_policy", "start_to_close_timeout", "sche
     "non_retryable", "exception", "failed_step_sequence", "failed_activity_execution_id", "timeout_kind",
     "parent_close_policy", "cancellation_scope_id", "cancellation_cleanup", "delay_seconds", "timeout_seconds",
     "payload_codec", "entries", "principal_subject", "principal_method", "principal_roles", "principal_tenant",
-    "principal_claims"];
+    "principal_claims", "await"];
 
 /// `optionalRetryPolicy` reads these keys and ignores the rest.
 const RETRY_POLICY_FIELDS: &[&str] = &["max_attempts", "backoff_seconds", "non_retryable_error_types"];

rust/src/runtime/scope_checkpoint.rs

@@ -276,6 +276,13 @@ store_methods! {
             }
             Err(error) => return Err(error),
         }
+        // A checkpoint keeps the parent's claim, so a child start turn has
+        // nothing to resume; it belongs to the task's completion (Server #966).
+        if commands.iter().any(|command| command["type"] == "start_child_workflow" && command["await"] == super::children::AWAIT_STARTED) {
+            tx.rollback().await?;
+            return refusal("child_start_turn_not_at_checkpoint");
+        }
+        let start_turn = std::sync::Mutex::new(None);
         let apply = super::store::Apply {
             run: &run,
             task: &task,
@@ -286,6 +293,7 @@ store_methods! {
             external,
             local,
             exceptions,
+            start_turn: &start_turn,
         };
         let mut created = Vec::new();
         for (index, command) in commands.iter().enumerate() {

rust/src/runtime/store.rs

@@ -3188,6 +3188,7 @@ where
         Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false, commands, signal).await?;
         // Published parseCommands applies every other command first, whatever
         // the terminal command's position in the batch.
+        let start_turn = std::sync::Mutex::new(None);
         let apply = Apply {
             run: &run,
             task: &task,
@@ -3198,6 +3199,7 @@ where
             external: &external,
             local: &local,
             exceptions: &exceptions,
+            start_turn: &start_turn,
         };
         let order = (0..commands.len())
             .filter(|index| !is_terminal(&commands[*index]))
@@ -3295,6 +3297,12 @@ where
         if let Some(next) = Self::next_message_task(&mut tx, &run_id, &DB::string(&run, "queue")?, attempted, false).await? {
             created.push(next);
         }
+        let start_turn = start_turn.into_inner().unwrap_or_else(std::sync::PoisonError::into_inner);
+        if let Some(turn) = start_turn
+            && let Some(next) = Self::child_start_turn(&mut tx, &run_id, task_id, &DB::string(&run, "queue")?, turn).await?
+        {
+            created.push(next);
+        }
         Self::record_completion(&mut tx, task_id, &json!({COMPLETION_RECEIPT: {"request": body, "created_task_ids": created}})).await?;
         // PHP's executor leaves every run a completed task did not close
         // waiting, even when the task recorded only markers or side effects.
@@ -3337,7 +3345,7 @@ where
         sequence_ref: &mut i64,
         created: &mut Vec<String>,
     ) -> Result<()> {
-        let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions } = *apply;
+        let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions, start_turn } = *apply;
         let mut sequence = *sequence_ref;
         if matches!(
             command["type"].as_str(),
@@ -3352,7 +3360,12 @@ where
         } else if command["type"] == "start_child_workflow" {
             Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Children, &limits, task_id, attempt).await?;
             sequence += 1;
-            created.push(Self::start_child(tx, &run, task_id, command, sequence).await?);
+            let started = Self::start_child(tx, &run, task_id, command, sequence).await?;
+            created.push(started.task);
+            if command["await"] == super::children::AWAIT_STARTED {
+                let mut turn = start_turn.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+                turn.get_or_insert(started.turn);
+            }
         } else if command["type"] == super::local_activities::COMMAND {
             // `applyRecordLocalActivity`: one authored sequence.
             sequence += 1;
@@ -5071,6 +5084,9 @@ pub(super) struct Apply<'a, R> {
     pub(super) external: &'a super::external_resolution::PreparedCommands,
     pub(super) local: &'a std::collections::BTreeMap<usize, super::local_activities::Record>,
     pub(super) exceptions: &'a std::collections::BTreeMap<usize, String>,
+    /// The first child this batch started with `await: "started"`: the
+    /// parent's start turn (Server #966).
+    pub(super) start_turn: &'a std::sync::Mutex<Option<Value>>,
 }
 
 impl<R> Clone for Apply<'_, R> {
@@ -5195,7 +5211,8 @@ pub(crate) fn capabilities() -> Value {
         "sticky_execution": false, "local_activities": super::local_activities::manifest(), "message_streams": super::message_streams::worker_capability(),
         "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(),
         "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
-        "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true});
+        "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,
+        "child_workflow_start_turn": true});
     value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest();
     value["supported_workflow_task_commands"]
         .as_array_mut()

rust/tests/execution.rs

@@ -370,6 +370,8 @@ mod key_order;
 mod activity_result_repeats;
 #[path = "execution/task_fairness.rs"]
 mod task_fairness;
+#[path = "execution/child_start_turn.rs"]
+mod child_start_turn;
 
 enum TestDatabase {
     Sqlite(tempfile::TempDir),