features/steps/subplan_spawn_orchestration_steps.py

@@ -0,0 +1,284 @@
+"""Step definitions for subplan spawn → execute → merge orchestration.
+
+These steps exercise ``SubplanService.spawn()`` and verify that the
+end-to-end orchestration creates real child ``Plan`` domain objects,
+wires parent-child identity, inherits project links, and correctly
+updates parent ``SubplanStatus`` entries on child completion/failure.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.subplan_service import (
+    SpawnEntry,
+    SpawnResult,
+    SubplanService,
+)
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+    SubplanConfig,
+    SubplanStatus,
+)
+
+_PARENT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
+_ROOT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
+_DECISION_ID_1: str = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
+_DECISION_ID_2: str = "01HGZ6FE0AQDYTR4BXVQZ6DB00"
+
+
+def _mock_decision_service() -> MagicMock:
+    """Create a mock DecisionService for SubplanService construction."""
+    svc: MagicMock = MagicMock()
+    svc.list_by_type = MagicMock(return_value=[])
+    return svc
+
+
+def _make_parent_plan(
+    execution_mode: ExecutionMode = ExecutionMode.SEQUENTIAL,
+) -> Plan:
+    """Create a parent plan suitable for subplan spawning."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=_PARENT_PLAN_ID,
+            root_plan_id=_ROOT_PLAN_ID,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="parent-plan"),
+        description="Parent plan for subplan spawn orchestration test",
+        action_name="local/parent-action",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        project_links=[
+            ProjectLink(project_name="local/my-project"),
+        ],
+        subplan_config=SubplanConfig(
+            execution_mode=execution_mode,
+        ),
+    )
+
+
+def _make_spawn_entries() -> list[SpawnEntry]:
+    """Create spawn entries for testing."""
+    dec1: Decision = Decision(
+        decision_id=_DECISION_ID_1,
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.SUBPLAN_SPAWN,
+        sequence_number=0,
+        question="Spawn child plan for module A?",
+        chosen_option="local/sub-action-a",
+        context_snapshot=ContextSnapshot(),
+    )
+    dec2: Decision = Decision(
+        decision_id=_DECISION_ID_2,
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.SUBPLAN_SPAWN,
+        sequence_number=1,
+        question="Spawn child plan for module B?",
+        chosen_option="local/sub-action-b",
+        context_snapshot=ContextSnapshot(),
+    )
+    return [
+        SpawnEntry(decision=dec1, action_name="local/sub-action-a"),
+        SpawnEntry(decision=dec2, action_name="local/sub-action-b"),
+    ]
+
+
+# -- Given steps (not duplicating tdd_subplan_spawn_orchestration_steps) ----
+
+
+@given("a parent plan configured for parallel subplan spawning")
+def step_parallel_parent_plan(context: Context) -> None:
+    """Set up a parent plan with parallel subplan configuration."""
+    context.parent_plan = _make_parent_plan(
+        execution_mode=ExecutionMode.PARALLEL,
+    )
+    context.subplan_service = SubplanService(
+        decision_service=_mock_decision_service(),
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.PARALLEL,
+    )
+
+
+# -- When steps ---------------------------------------------------------------
+
+
+@when('a child plan fails with error "{error_msg}"')
+def step_child_plan_fails(context: Context, error_msg: str) -> None:
+    """Simulate a child plan failure by updating the parent's subplan status."""
+    result: SpawnResult = context.spawn_result
+    parent: Plan = context.parent_plan
+
+    if result.child_plans:
+        child: Plan = result.child_plans[0]
+        child_id: str = child.identity.plan_id
+        # Update the matching subplan status on the parent
+        for status in parent.subplan_statuses:
+            if status.subplan_id == child_id:
+                status.status = ProcessingState.ERRORED
+                status.error = error_msg
+                break
+        context.affected_child_id = child_id
+
+
+@when(
+    'a child plan completes successfully with changeset summary "{summary}"',
+)
+def step_child_plan_completes(context: Context, summary: str) -> None:
+    """Simulate a child plan completion by updating the parent's subplan status."""
+    result: SpawnResult = context.spawn_result
+    parent: Plan = context.parent_plan
+
+    if result.child_plans:
+        child: Plan = result.child_plans[0]
+        child_id: str = child.identity.plan_id
+        # Update the matching subplan status on the parent
+        for status in parent.subplan_statuses:
+            if status.subplan_id == child_id:
+                status.status = ProcessingState.COMPLETE
+                status.changeset_summary = summary
+                status.files_changed = 1
+                break
+        context.affected_child_id = child_id
+
+
+# -- Then steps ---------------------------------------------------------------
+
+
+@then("each child Plan root_plan_id should trace to the root plan")
+def step_child_plans_root_id(context: Context) -> None:
+    """Assert child Plan objects reference the root plan."""
+    result: SpawnResult = context.spawn_result
+    assert result.child_plans is not None, "No child_plans on SpawnResult"
+    for child in result.child_plans:
+        assert isinstance(child, Plan)
+        assert child.identity.root_plan_id == _ROOT_PLAN_ID, (
+            f"Child plan root_plan_id={child.identity.root_plan_id!r}, "
+            f"expected {_ROOT_PLAN_ID!r}"
+        )
+
+
+@then("each child plan should inherit the parent project links")
+def step_child_plans_inherit_project_links(context: Context) -> None:
+    """Assert child plans carry the parent's project links."""
+    result: SpawnResult = context.spawn_result
+    parent: Plan = context.parent_plan
+    assert result.child_plans is not None, "No child_plans on SpawnResult"
+    parent_project_names: list[str] = [
+        link.project_name for link in parent.project_links
+    ]
+    for child in result.child_plans:
+        child_project_names: list[str] = [
+            link.project_name for link in child.project_links
+        ]
+        assert child_project_names == parent_project_names, (
+            f"Child project links {child_project_names!r} do not match "
+            f"parent {parent_project_names!r}"
+        )
+
+
+@then("the parent plan should have pending subplan statuses")
+def step_parent_has_pending_statuses(context: Context) -> None:
+    """Assert parent plan has queued subplan statuses after spawn."""
+    parent: Plan = context.parent_plan
+    assert len(parent.subplan_statuses) > 0, "No subplan statuses on parent"
+    for status in parent.subplan_statuses:
+        assert status.status == ProcessingState.QUEUED, (
+            f"Expected QUEUED, got {status.status!r}"
+        )
+
+
+@then("the parent plan subplan_statuses count should equal spawn count")
+def step_parent_status_count_matches(context: Context) -> None:
+    """Assert the number of subplan statuses equals the spawn entries."""
+    parent: Plan = context.parent_plan
+    entries: list[SpawnEntry] = context.spawn_entries
+    assert len(parent.subplan_statuses) == len(entries), (
+        f"Parent has {len(parent.subplan_statuses)} statuses, expected {len(entries)}"
+    )
+
+
+@then("the corresponding parent subplan status should be errored")
+def step_parent_status_errored(context: Context) -> None:
+    """Assert the affected child's parent subplan status is errored."""
+    parent: Plan = context.parent_plan
+    child_id: str = context.affected_child_id
+    matching: list[SubplanStatus] = [
+        s for s in parent.subplan_statuses if s.subplan_id == child_id
+    ]
+    assert len(matching) == 1, f"Expected 1 matching status, got {len(matching)}"
+    assert matching[0].status == ProcessingState.ERRORED
+
+
+@then('the parent subplan status error should contain "{fragment}"')
+def step_parent_status_error_contains(context: Context, fragment: str) -> None:
+    """Assert the error message contains the expected fragment."""
+    parent: Plan = context.parent_plan
+    child_id: str = context.affected_child_id
+    matching: list[SubplanStatus] = [
+        s for s in parent.subplan_statuses if s.subplan_id == child_id
+    ]
+    assert len(matching) == 1
+    assert matching[0].error is not None, "Error is None"
+    assert fragment in matching[0].error, (
+        f"Expected '{fragment}' in error '{matching[0].error}'"
+    )
+
+
+@then("the corresponding parent subplan status should be complete")
+def step_parent_status_complete(context: Context) -> None:
+    """Assert the affected child's parent subplan status is complete."""
+    parent: Plan = context.parent_plan
+    child_id: str = context.affected_child_id
+    matching: list[SubplanStatus] = [
+        s for s in parent.subplan_statuses if s.subplan_id == child_id
+    ]
+    assert len(matching) == 1
+    assert matching[0].status == ProcessingState.COMPLETE
+
+
+@then('the parent subplan status changeset summary should be "{summary}"')
+def step_parent_status_changeset_summary(context: Context, summary: str) -> None:
+    """Assert the changeset summary matches."""
+    parent: Plan = context.parent_plan
+    child_id: str = context.affected_child_id
+    matching: list[SubplanStatus] = [
+        s for s in parent.subplan_statuses if s.subplan_id == child_id
+    ]
+    assert len(matching) == 1
+    assert matching[0].changeset_summary == summary, (
+        f"Expected summary '{summary}', got '{matching[0].changeset_summary}'"
+    )
+
+
+@then("the spawn result execution mode should be parallel")
+def step_result_mode_parallel(context: Context) -> None:
+    """Assert the spawn result execution mode is parallel."""
+    result: SpawnResult = context.spawn_result
+    assert result.execution_mode == ExecutionMode.PARALLEL, (
+        f"Expected PARALLEL, got {result.execution_mode!r}"
+    )
+
+
+@then("the spawn result child plan count should match entries")
+def step_result_child_count_matches(context: Context) -> None:
+    """Assert child plan count matches spawn entries."""
+    result: SpawnResult = context.spawn_result
+    entries: list[SpawnEntry] = context.spawn_entries
+    assert len(result.child_plans) == len(entries), (
+        f"Expected {len(entries)} child plans, got {len(result.child_plans)}"
+    )

features/subplan_spawn_orchestration.feature

@@ -0,0 +1,44 @@
+@mock_only @subplan @orchestration
+Feature: Subplan spawn → execute → merge orchestration lifecycle
+  As a system orchestrating hierarchical plan execution
+  I want SubplanService.spawn() to create real child Plan domain objects
+  and wire them through the full strategize → execute → validate cycle
+  So that subplan execution is end-to-end orchestrated
+
+  Background:
+    Given a parent plan configured for subplan spawning
+    And valid spawn entries for the parent plan
+
+  Scenario: Spawn creates child Plans with correct parent-child identity
+    When I spawn subplans via SubplanService
+    Then the spawn result should contain child Plan domain objects
+    And each child Plan should have parent_plan_id set to the parent
+    And each child Plan root_plan_id should trace to the root plan
+
+  Scenario: Child plans inherit project links from parent
+    When I spawn subplans via SubplanService
+    Then each child plan should inherit the parent project links
+
+  Scenario: Parent plan blocks when child plans are pending
+    When I spawn subplans via SubplanService
+    Then the parent plan should have pending subplan statuses
+    And the parent plan subplan_statuses count should equal spawn count
+
+  Scenario: Child plan failure marks parent subplan status as errored
+    When I spawn subplans via SubplanService
+    And a child plan fails with error "ChildExecError: timeout"
+    Then the corresponding parent subplan status should be errored
+    And the parent subplan status error should contain "ChildExecError"
+
+  Scenario: Child plan completion updates parent subplan status
+    When I spawn subplans via SubplanService
+    And a child plan completes successfully with changeset summary "Added tests"
+    Then the corresponding parent subplan status should be complete
+    And the parent subplan status changeset summary should be "Added tests"
+
+  Scenario: Parallel spawn sets execution mode on result
+    Given a parent plan configured for parallel subplan spawning
+    And valid spawn entries for the parent plan
+    When I spawn subplans via SubplanService
+    Then the spawn result execution mode should be parallel
+    And the spawn result child plan count should match entries

features/tdd_subplan_spawn_orchestration.feature

@@ -1,4 +1,4 @@
-@tdd_expected_fail @tdd_bug @tdd_bug_823 @mock_only
+@tdd_bug @tdd_bug_823 @mock_only
 Feature: TDD Bug #823 — subplan spawn creates metadata but does not orchestrate child plan execution
   As a developer
   I want to verify that SubplanService.spawn() creates real child Plan

robot/helper_subplan_spawn_orchestration.py

@@ -0,0 +1,226 @@
+"""Helper script for subplan_spawn_orchestration.robot integration tests.
+
+Each subcommand exercises SubplanService.spawn() in parallel mode and
+verifies end-to-end orchestration: child Plan creation, identity wiring,
+project link inheritance, parent status attachment, and execution mode.
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Callable
+from pathlib import Path
+from typing import NoReturn
+from unittest.mock import MagicMock
+
+# Ensure local source tree is importable.
+_ROOT: Path = Path(__file__).resolve().parents[1]
+_SRC: str = str(_ROOT / "src")
+_ROBOT: str = str(_ROOT / "robot")
+for _p in (_SRC, _ROBOT):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from cleveragents.application.services.subplan_service import (  # noqa: E402
+    SpawnEntry,
+    SpawnResult,
+    SubplanService,
+)
+from cleveragents.domain.models.core.decision import (  # noqa: E402
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    ExecutionMode,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+    SubplanConfig,
+)
+
+_PARENT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
+_ROOT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
+_DEC_ID1: str = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
+_DEC_ID2: str = "01HGZ6FE0AQDYTR4BXVQZ6DB00"
+_DEC_ID3: str = "01HGZ6FE0AQDYTR4BXVQZ6DC00"
+
+
+def _fail(msg: str) -> NoReturn:
+    """Print error message to stderr and exit with code 1."""
+    print(msg, file=sys.stderr)
+    sys.exit(1)
+
+
+def _mock_decision_service() -> MagicMock:
+    """Create a mock DecisionService."""
+    svc: MagicMock = MagicMock()
+    svc.list_by_type = MagicMock(return_value=[])
+    return svc
+
+
+def _make_parent_plan() -> Plan:
+    """Create a parent plan configured for parallel subplan spawning."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=_PARENT_PLAN_ID,
+            root_plan_id=_ROOT_PLAN_ID,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="parallel-parent"),
+        description="Parent plan for parallel subplan orchestration",
+        action_name="local/parallel-action",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        project_links=[
+            ProjectLink(project_name="local/my-project"),
+            ProjectLink(project_name="local/other-project", read_only=True),
+        ],
+        subplan_config=SubplanConfig(
+            execution_mode=ExecutionMode.PARALLEL,
+            max_parallel=5,
+        ),
+    )
+
+
+def _make_spawn_entries() -> list[SpawnEntry]:
+    """Create three spawn entries for parallel testing."""
+    decisions: list[tuple[str, str, str]] = [
+        (_DEC_ID1, "Spawn child A?", "local/sub-action-a"),
+        (_DEC_ID2, "Spawn child B?", "local/sub-action-b"),
+        (_DEC_ID3, "Spawn child C?", "local/sub-action-c"),
+    ]
+    entries: list[SpawnEntry] = []
+    for i, (dec_id, question, action) in enumerate(decisions):
+        dec: Decision = Decision(
+            decision_id=dec_id,
+            plan_id=_PARENT_PLAN_ID,
+            decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN,
+            sequence_number=i,
+            question=question,
+            chosen_option=action,
+            context_snapshot=ContextSnapshot(),
+        )
+        entries.append(SpawnEntry(decision=dec, action_name=action))
+    return entries
+
+
+def _do_spawn() -> tuple[Plan, SubplanConfig, list[SpawnEntry], SpawnResult]:
+    """Execute spawn and return all artifacts."""
+    svc: SubplanService = SubplanService(
+        decision_service=_mock_decision_service(),
+    )
+    parent: Plan = _make_parent_plan()
+    config: SubplanConfig = SubplanConfig(
+        execution_mode=ExecutionMode.PARALLEL,
+        max_parallel=5,
+    )
+    entries: list[SpawnEntry] = _make_spawn_entries()
+    result: SpawnResult = svc.spawn(
+        parent_plan=parent,
+        config=config,
+        spawn_entries=entries,
+    )
+    return parent, config, entries, result
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def _parallel_spawn_identity() -> None:
+    """Verify child Plans have correct parent/root identity."""
+    _parent, _config, entries, result = _do_spawn()
+    child_plans: list[Plan] | None = getattr(result, "child_plans", None)
+    if child_plans is None:
+        _fail("SpawnResult has no child_plans attribute")
+    if len(child_plans) != len(entries):
+        _fail(f"Expected {len(entries)} child plans, got {len(child_plans)}")
+    for child in child_plans:
+        if not isinstance(child, Plan):
+            _fail(f"Expected Plan, got {type(child).__name__}")
+        if child.identity.parent_plan_id != _PARENT_PLAN_ID:
+            _fail(
+                f"parent_plan_id={child.identity.parent_plan_id!r}, "
+                f"expected {_PARENT_PLAN_ID!r}"
+            )
+        if child.identity.root_plan_id != _ROOT_PLAN_ID:
+            _fail(
+                f"root_plan_id={child.identity.root_plan_id!r}, "
+                f"expected {_ROOT_PLAN_ID!r}"
+            )
+        if child.phase != PlanPhase.STRATEGIZE:
+            _fail(f"Child phase={child.phase!r}, expected STRATEGIZE")
+        if child.processing_state != ProcessingState.QUEUED:
+            _fail(f"Child state={child.processing_state!r}, expected QUEUED")
+    print("parallel-spawn-identity-ok")
+
+
+def _parallel_spawn_project_links() -> None:
+    """Verify child plans inherit project links from parent."""
+    parent, _config, _entries, result = _do_spawn()
+    child_plans: list[Plan] | None = getattr(result, "child_plans", None)
+    if child_plans is None:
+        _fail("SpawnResult has no child_plans attribute")
+    parent_names: list[str] = [lnk.project_name for lnk in parent.project_links]
+    for child in child_plans:
+        child_names: list[str] = [lnk.project_name for lnk in child.project_links]
+        if child_names != parent_names:
+            _fail(f"Child project links {child_names!r} != parent {parent_names!r}")
+    print("parallel-spawn-project-links-ok")
+
+
+def _parallel_spawn_parent_statuses() -> None:
+    """Verify spawn() attaches SubplanStatus entries to the parent."""
+    parent, _config, entries, result = _do_spawn()
+    if len(parent.subplan_statuses) != len(entries):
+        _fail(
+            f"Parent has {len(parent.subplan_statuses)} statuses, "
+            f"expected {len(entries)}"
+        )
+    # Verify each status references a child plan
+    child_ids: set[str] = set()
+    child_plans: list[Plan] | None = getattr(result, "child_plans", None)
+    if child_plans:
+        child_ids = {c.identity.plan_id for c in child_plans}
+    for status in parent.subplan_statuses:
+        if status.subplan_id not in child_ids:
+            _fail(f"SubplanStatus id={status.subplan_id!r} not in child plan IDs")
+        if status.status != ProcessingState.QUEUED:
+            _fail(f"Status state={status.status!r}, expected QUEUED")
+    print("parallel-spawn-parent-statuses-ok")
+
+
+def _parallel_spawn_mode() -> None:
+    """Verify spawn result reports PARALLEL execution mode."""
+    _parent, _config, _entries, result = _do_spawn()
+    if result.execution_mode != ExecutionMode.PARALLEL:
+        _fail(f"execution_mode={result.execution_mode!r}, expected PARALLEL")
+    if result.total_spawned != 3:
+        _fail(f"total_spawned={result.total_spawned}, expected 3")
+    print("parallel-spawn-mode-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Callable[[], None]] = {
+    "parallel-spawn-identity": _parallel_spawn_identity,
+    "parallel-spawn-project-links": _parallel_spawn_project_links,
+    "parallel-spawn-parent-statuses": _parallel_spawn_parent_statuses,
+    "parallel-spawn-mode": _parallel_spawn_mode,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(
+            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    cmd: Callable[[], None] = _COMMANDS[sys.argv[1]]
+    cmd()

robot/subplan_spawn_orchestration.robot

@@ -0,0 +1,50 @@
+*** Settings ***
+Documentation    Integration test — parallel subplan spawn orchestration.
+...              Verifies that SubplanService.spawn() creates real child Plan
+...              domain objects, inherits project links, sets correct execution
+...              mode, and attaches SubplanStatus entries to the parent plan
+...              when configured for parallel execution.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_subplan_spawn_orchestration.py
+
+*** Test Cases ***
+Parallel Spawn Creates Child Plans With Correct Identity
+    [Documentation]    Verify parallel spawn creates child Plans with parent/root IDs.
+    [Tags]    subplan    orchestration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    parallel-spawn-identity    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-spawn-identity-ok
+
+Parallel Spawn Child Plans Inherit Project Links
+    [Documentation]    Verify child plans inherit project links from the parent.
+    [Tags]    subplan    orchestration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    parallel-spawn-project-links    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-spawn-project-links-ok
+
+Parallel Spawn Attaches Statuses To Parent
+    [Documentation]    Verify that spawn() attaches SubplanStatus entries to the
+    ...                parent plan for lifecycle tracking.
+    [Tags]    subplan    orchestration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    parallel-spawn-parent-statuses    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-spawn-parent-statuses-ok
+
+Parallel Spawn Execution Mode Is Parallel
+    [Documentation]    Verify the spawn result reports PARALLEL execution mode.
+    [Tags]    subplan    orchestration
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    parallel-spawn-mode    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    parallel-spawn-mode-ok

robot/tdd_subplan_spawn_orchestration.robot

@@ -16,7 +16,7 @@ ${HELPER}    ${CURDIR}/helper_tdd_subplan_spawn_orchestration.py
 TDD Spawn Result Contains Child Plan Objects
     [Documentation]    Verify that spawn() returns actual child Plan domain objects,
     ...                not just SubplanStatus metadata.
-    [Tags]    tdd_expected_fail    tdd_bug    tdd_bug_823
+    [Tags]    tdd_bug    tdd_bug_823
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    spawn-child-plans    cwd=${WORKSPACE}    timeout=30s
     Log    ${result.stdout}
     Log    ${result.stderr}
@@ -26,7 +26,7 @@ TDD Spawn Result Contains Child Plan Objects
 TDD Child Plans Enter Strategize Phase
     [Documentation]    Verify that spawned child plans enter the strategize phase
     ...                and have their lifecycle triggered.
-    [Tags]    tdd_expected_fail    tdd_bug    tdd_bug_823
+    [Tags]    tdd_bug    tdd_bug_823
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    spawn-lifecycle    cwd=${WORKSPACE}    timeout=30s
     Log    ${result.stdout}
     Log    ${result.stderr}
@@ -36,7 +36,7 @@ TDD Child Plans Enter Strategize Phase
 TDD Parent Plan Tracks Child Status
     [Documentation]    Verify that spawn() attaches SubplanStatus entries to
     ...                the parent plan so lifecycle can be tracked.
-    [Tags]    tdd_expected_fail    tdd_bug    tdd_bug_823
+    [Tags]    tdd_bug    tdd_bug_823
     ${result}=    Run Process    ${PYTHON}    ${HELPER}    spawn-parent-tracking    cwd=${WORKSPACE}    timeout=30s
     Log    ${result.stdout}
     Log    ${result.stderr}

src/cleveragents/application/services/subplan_service.py

@@ -27,11 +27,18 @@ import logging
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+from ulid import ULID
+
 from cleveragents.core.exceptions import ValidationError
 from cleveragents.domain.models.core.decision import Decision, DecisionType
 from cleveragents.domain.models.core.plan import (
     ExecutionMode,
+    NamespacedName,
     Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
     SubplanConfig,
     SubplanStatus,
 )
@@ -105,12 +112,14 @@ class SpawnResult:
         metadata: Spawn metadata for each child plan (keyed by subplan_id).
         total_spawned: Number of child plans spawned.
         execution_mode: The execution mode from the SubplanConfig.
+        child_plans: Actual child Plan domain objects created during spawn.
     """
 
     spawned_statuses: list[SubplanStatus]
     metadata: dict[str, SpawnMetadata] = field(default_factory=dict)
     total_spawned: int = 0
     execution_mode: str = ExecutionMode.SEQUENTIAL
+    child_plans: list[Plan] = field(default_factory=list)
 
 
 # ---------------------------------------------------------------------------
@@ -219,18 +228,18 @@ class SubplanService:
         if not validation.valid:
             raise SpawnValidationError(validation.errors)
 
-        # Build statuses and metadata
+        # Build statuses, metadata, and real child Plan objects
         parent_id: str = parent_plan.identity.plan_id
         root_id: str = parent_plan.identity.root_plan_id or parent_id
         mode: str = config.execution_mode.value
 
         statuses: list[SubplanStatus] = []
         metadata: dict[str, SpawnMetadata] = {}
+        child_plans: list[Plan] = []
 
         for entry in spawn_entries:
-            # Use the decision's downstream_plan_ids if available,
-            # otherwise generate a subplan_id from the decision_id
-            subplan_id: str = entry.decision.decision_id
+            # Generate a unique ULID for each child plan
+            subplan_id: str = str(ULID())
 
             status: SubplanStatus = SubplanStatus(
                 subplan_id=subplan_id,
@@ -247,6 +256,36 @@ class SubplanService:
             )
             metadata[subplan_id] = meta
 
+            # Create a real child Plan domain object
+            child_plan: Plan = Plan(
+                identity=PlanIdentity(
+                    plan_id=subplan_id,
+                    parent_plan_id=parent_id,
+                    root_plan_id=root_id,
+                ),
+                namespaced_name=NamespacedName(
+                    server=parent_plan.namespaced_name.server,
+                    namespace=parent_plan.namespaced_name.namespace,
+                    name=f"subplan-{subplan_id[:8]}",
+                ),
+                description=entry.description or f"Child plan for {entry.action_name}",
+                definition_of_done=parent_plan.definition_of_done,
+                action_name=entry.action_name,
+                phase=PlanPhase.STRATEGIZE,
+                processing_state=ProcessingState.QUEUED,
+                strategy_actor=parent_plan.strategy_actor,
+                execution_actor=parent_plan.execution_actor,
+                project_links=list(parent_plan.project_links),
+                timestamps=PlanTimestamps(),
+                created_by=parent_plan.created_by,
+                reusable=parent_plan.reusable,
+                read_only=parent_plan.read_only,
+            )
+            child_plans.append(child_plan)
+
+        # Attach subplan statuses to the parent plan for lifecycle tracking
+        parent_plan.subplan_statuses = list(parent_plan.subplan_statuses) + statuses
+
         logger.info(
             "spawned_subplans",
             extra={
@@ -262,6 +301,7 @@ class SubplanService:
             metadata=metadata,
             total_spawned=len(statuses),
             execution_mode=mode,
+            child_plans=child_plans,
         )
 
     # ------------------------------------------------------------------