features/steps/subplan_status_registry_steps.py

@@ -0,0 +1,521 @@
+"""Step implementations for subplan status registry BDD tests."""
+
+import time
+
+from behave import given, then, when
+
+from cleveragents.domain.models.core.subplan_status_registry import (
+    SubplanStatusRegistry,
+)
+
+
+@given('a parent plan with ULID "{parent_plan_id}"')
+def step_parent_plan_with_ulid(context, parent_plan_id: str) -> None:
+    """Create a parent plan context."""
+    context.parent_plan_id = parent_plan_id
+
+
+@given("a subplan status registry for the parent plan")
+def step_create_registry(context) -> None:
+    """Create a subplan status registry."""
+    context.registry = SubplanStatusRegistry(parent_plan_id=context.parent_plan_id)
+    context.subplans = {}  # Track subplans by ID for easy access
+
+
+@when("I register a subplan with:")
+def step_register_subplan(context) -> None:
+    """Register a new subplan."""
+    data = {row["key"]: row["value"] for row in context.table}
+    subplan_id = data["subplan_id"]
+    action_name = data["action_name"]
+    target_resources = data["target_resources"].split(",")
+
+    context.registry.register_subplan(subplan_id, action_name, target_resources)
+    context.subplans[subplan_id] = {
+        "action_name": action_name,
+        "target_resources": target_resources,
+    }
+    context.last_subplan_id = subplan_id
+
+
+@given('a registered subplan with ID "{subplan_id}"')
+def step_registered_subplan(context, subplan_id: str) -> None:
+    """Create a registered subplan."""
+    context.registry.register_subplan(
+        subplan_id, "test.action", ["resource-1", "resource-2"]
+    )
+    context.subplans[subplan_id] = {
+        "action_name": "test.action",
+        "target_resources": ["resource-1", "resource-2"],
+    }
+    context.last_subplan_id = subplan_id
+
+
+@given('a registered subplan with ID "{subplan_id}" with action "{action_name}"')
+def step_registered_subplan_with_action(
+    context, subplan_id: str, action_name: str
+) -> None:
+    """Create a registered subplan with specific action."""
+    context.registry.register_subplan(subplan_id, action_name, ["resource-1"])
+    context.subplans[subplan_id] = {
+        "action_name": action_name,
+        "target_resources": ["resource-1"],
+    }
+    context.last_subplan_id = subplan_id
+
+
+@given('a running subplan with ID "{subplan_id}"')
+def step_running_subplan(context, subplan_id: str) -> None:
+    """Create a running subplan."""
+    context.registry.register_subplan(subplan_id, "test.action", ["resource-1"])
+    context.registry.start_subplan(subplan_id)
+    context.subplans[subplan_id] = {
+        "action_name": "test.action",
+        "target_resources": ["resource-1"],
+    }
+    context.last_subplan_id = subplan_id
+
+
+@given('a completed subplan with ID "{subplan_id}"')
+def step_completed_subplan(context, subplan_id: str) -> None:
+    """Create a completed subplan."""
+    context.registry.register_subplan(subplan_id, "test.action", ["resource-1"])
+    context.registry.start_subplan(subplan_id)
+    context.registry.complete_subplan(
+        subplan_id, changeset_summary="Modified 1 file", files_modified=1
+    )
+    context.subplans[subplan_id] = {
+        "action_name": "test.action",
+        "target_resources": ["resource-1"],
+    }
+    context.last_subplan_id = subplan_id
+
+
+@given('a failed subplan with ID "{subplan_id}"')
+def step_failed_subplan(context, subplan_id: str) -> None:
+    """Create a failed subplan."""
+    context.registry.register_subplan(subplan_id, "test.action", ["resource-1"])
+    context.registry.start_subplan(subplan_id)
+    context.registry.fail_subplan(subplan_id, "Test failure")
+    context.subplans[subplan_id] = {
+        "action_name": "test.action",
+        "target_resources": ["resource-1"],
+    }
+    context.last_subplan_id = subplan_id
+
+
+@when("I start the subplan")
+def step_start_subplan(context) -> None:
+    """Start the last subplan."""
+    context.registry.start_subplan(context.last_subplan_id)
+
+
+@when("I complete the subplan")
+def step_complete_subplan(context) -> None:
+    """Complete the last subplan."""
+    context.registry.complete_subplan(
+        context.last_subplan_id, changeset_summary="Modified 1 file", files_modified=1
+    )
+
+
+@when("I complete the subplan with:")
+def step_complete_subplan_with_data(context) -> None:
+    """Complete the last subplan with specific data."""
+    data = {row["key"]: row["value"] for row in context.table}
+    changeset_summary = data.get("changeset_summary")
+    files_modified = int(data.get("files_modified", 0))
+
+    context.registry.complete_subplan(
+        context.last_subplan_id,
+        changeset_summary=changeset_summary,
+        files_modified=files_modified,
+    )
+
+
+@when('I complete the subplan with changeset "{changeset}"')
+def step_complete_subplan_with_changeset(context, changeset: str) -> None:
+    """Complete the last subplan with changeset."""
+    context.registry.complete_subplan(
+        context.last_subplan_id, changeset_summary=changeset, files_modified=1
+    )
+
+
+@when('I fail the subplan with error "{error}"')
+def step_fail_subplan(context, error: str) -> None:
+    """Fail the last subplan."""
+    context.registry.fail_subplan(context.last_subplan_id, error)
+
+
+@when('I get the status of subplan "{subplan_id}"')
+def step_get_subplan_status(context, subplan_id: str) -> None:
+    """Get the status of a specific subplan."""
+    context.last_status = context.registry.get_subplan_status(subplan_id)
+
+
+@when("I get all subplans")
+def step_get_all_subplans(context) -> None:
+    """Get all subplans."""
+    context.all_subplans = context.registry.get_all_subplans()
+
+
+@when("I get all completed results")
+def step_get_completed_results(context) -> None:
+    """Get all completed results."""
+    context.completed_results = context.registry.get_completed_results()
+
+
+@when("I get all failed subplans")
+def step_get_failed_subplans(context) -> None:
+    """Get all failed subplans."""
+    context.failed_subplans = context.registry.get_failed_subplans()
+
+
+@when("I check if all subplans are completed")
+def step_check_all_completed(context) -> None:
+    """Check if all subplans are completed."""
+    context.all_completed = context.registry.is_all_completed()
+
+
+@when("I get the completion status")
+def step_get_completion_status(context) -> None:
+    """Get the completion status."""
+    context.completion_status = context.registry.get_completion_status()
+
+
+@when("I record the registry updated_at timestamp")
+def step_record_timestamp(context) -> None:
+    """Record the current updated_at timestamp."""
+    context.recorded_timestamp = context.registry.updated_at
+
+
+@when("I wait 100 milliseconds")
+def step_wait_milliseconds(context) -> None:
+    """Wait for 100 milliseconds."""
+    time.sleep(0.1)
+
+
+@when("I try to start the subplan")
+def step_try_start_subplan(context) -> None:
+    """Try to start the last subplan (may fail)."""
+    try:
+        context.registry.start_subplan(context.last_subplan_id)
+        context.error_raised = False
+    except ValueError as e:
+        context.error_raised = True
+        context.error_message = str(e)
+
+
+@when("I try to complete the subplan")
+def step_try_complete_subplan(context) -> None:
+    """Try to complete the last subplan (may fail)."""
+    try:
+        context.registry.complete_subplan(context.last_subplan_id)
+        context.error_raised = False
+    except ValueError as e:
+        context.error_raised = True
+        context.error_message = str(e)
+
+
+@when('I try to fail the subplan with error "{error}"')
+def step_try_fail_subplan(context, error: str) -> None:
+    """Try to fail the last subplan (may fail)."""
+    try:
+        context.registry.fail_subplan(context.last_subplan_id, error)
+        context.error_raised = False
+    except ValueError as e:
+        context.error_raised = True
+        context.error_message = str(e)
+
+
+@when("I register 10 subplans")
+def step_register_multiple_subplans(context) -> None:
+    """Register 10 subplans."""
+    for i in range(10):
+        subplan_id = f"01ARZ3NDEKTSV4RRFFQ69G5FA{chr(65 + i)}"
+        context.registry.register_subplan(
+            subplan_id, f"action.{i}", [f"resource-{i}"]
+        )
+        context.subplans[subplan_id] = {
+            "action_name": f"action.{i}",
+            "target_resources": [f"resource-{i}"],
+        }
+
+
+@when("I start 5 of them")
+def step_start_multiple_subplans(context) -> None:
+    """Start 5 subplans."""
+    subplan_ids = list(context.registry.pending_subplans.keys())[:5]
+    for subplan_id in subplan_ids:
+        context.registry.start_subplan(subplan_id)
+
+
+@when("I complete 3 of them")
+def step_complete_multiple_subplans(context) -> None:
+    """Complete 3 subplans."""
+    subplan_ids = list(context.registry.running_subplans.keys())[:3]
+    for subplan_id in subplan_ids:
+        context.registry.complete_subplan(subplan_id)
+
+
+@when("I fail 1 of them")
+def step_fail_one_subplan(context) -> None:
+    """Fail 1 subplan."""
+    subplan_ids = list(context.registry.running_subplans.keys())
+    if subplan_ids:
+        context.registry.fail_subplan(subplan_ids[0], "Test failure")
+
+
+@then("the subplan should be in pending state")
+def step_subplan_in_pending_state(context) -> None:
+    """Verify subplan is in pending state."""
+    assert context.last_subplan_id in context.registry.pending_subplans
+    assert context.last_subplan_id not in context.registry.running_subplans
+    assert context.last_subplan_id not in context.registry.completed_subplans
+    assert context.last_subplan_id not in context.registry.failed_subplans
+
+
+@then("the subplan should be in running state")
+def step_subplan_in_running_state(context) -> None:
+    """Verify subplan is in running state."""
+    assert context.last_subplan_id not in context.registry.pending_subplans
+    assert context.last_subplan_id in context.registry.running_subplans
+    assert context.last_subplan_id not in context.registry.completed_subplans
+    assert context.last_subplan_id not in context.registry.failed_subplans
+
+
+@then("the subplan should be in completed state")
+def step_subplan_in_completed_state(context) -> None:
+    """Verify subplan is in completed state."""
+    assert context.last_subplan_id not in context.registry.pending_subplans
+    assert context.last_subplan_id not in context.registry.running_subplans
+    assert context.last_subplan_id in context.registry.completed_subplans
+    assert context.last_subplan_id not in context.registry.failed_subplans
+
+
+@then("the subplan should be in failed state")
+def step_subplan_in_failed_state(context) -> None:
+    """Verify subplan is in failed state."""
+    assert context.last_subplan_id not in context.registry.pending_subplans
+    assert context.last_subplan_id not in context.registry.running_subplans
+    assert context.last_subplan_id not in context.registry.completed_subplans
+    assert context.last_subplan_id in context.registry.failed_subplans
+
+
+@then("the registry should contain {count:d} pending subplan")
+def step_check_pending_count(context, count: int) -> None:
+    """Verify pending subplan count."""
+    assert len(context.registry.pending_subplans) == count
+
+
+@then("the registry should contain {count:d} pending subplans")
+def step_check_pending_count_plural(context, count: int) -> None:
+    """Verify pending subplan count (plural)."""
+    assert len(context.registry.pending_subplans) == count
+
+
+@then("the registry should contain {count:d} running subplan")
+def step_check_running_count(context, count: int) -> None:
+    """Verify running subplan count."""
+    assert len(context.registry.running_subplans) == count
+
+
+@then("the registry should contain {count:d} running subplans")
+def step_check_running_count_plural(context, count: int) -> None:
+    """Verify running subplan count (plural)."""
+    assert len(context.registry.running_subplans) == count
+
+
+@then("the registry should contain {count:d} completed subplan")
+def step_check_completed_count(context, count: int) -> None:
+    """Verify completed subplan count."""
+    assert len(context.registry.completed_subplans) == count
+
+
+@then("the registry should contain {count:d} completed subplans")
+def step_check_completed_count_plural(context, count: int) -> None:
+    """Verify completed subplan count (plural)."""
+    assert len(context.registry.completed_subplans) == count
+
+
+@then("the registry should contain {count:d} failed subplan")
+def step_check_failed_count(context, count: int) -> None:
+    """Verify failed subplan count."""
+    assert len(context.registry.failed_subplans) == count
+
+
+@then("the registry should contain {count:d} failed subplans")
+def step_check_failed_count_plural(context, count: int) -> None:
+    """Verify failed subplan count (plural)."""
+    assert len(context.registry.failed_subplans) == count
+
+
+@then("the subplan should have a started_at timestamp")
+def step_check_started_at(context) -> None:
+    """Verify subplan has started_at timestamp."""
+    metadata = context.registry.running_subplans[context.last_subplan_id]
+    assert "started_at" in metadata
+    assert metadata["started_at"] is not None
+
+
+@then("the subplan should have a completed_at timestamp")
+def step_check_completed_at(context) -> None:
+    """Verify subplan has completed_at timestamp."""
+    metadata = context.registry.completed_subplans[context.last_subplan_id]
+    assert "completed_at" in metadata
+    assert metadata["completed_at"] is not None
+
+
+@then("the subplan should have a failed_at timestamp")
+def step_check_failed_at(context) -> None:
+    """Verify subplan has failed_at timestamp."""
+    metadata = context.registry.failed_subplans[context.last_subplan_id]
+    assert "failed_at" in metadata
+    assert metadata["failed_at"] is not None
+
+
+@then("the subplan result should contain the changeset summary")
+def step_check_changeset_summary(context) -> None:
+    """Verify subplan result contains changeset summary."""
+    metadata = context.registry.completed_subplans[context.last_subplan_id]
+    assert "changeset_summary" in metadata
+    assert metadata["changeset_summary"] is not None
+
+
+@then('the subplan error should be "{error}"')
+def step_check_error_message(context, error: str) -> None:
+    """Verify subplan error message."""
+    metadata = context.registry.failed_subplans[context.last_subplan_id]
+    assert metadata["error"] == error
+
+
+@then('the status should be "{status}"')
+def step_check_status(context, status: str) -> None:
+    """Verify subplan status."""
+    assert context.last_status is not None
+    assert context.last_status["status"] == status
+
+
+@then("the metadata should contain the action_name")
+def step_check_metadata_action_name(context) -> None:
+    """Verify metadata contains action_name."""
+    assert "metadata" in context.last_status
+    assert "action_name" in context.last_status["metadata"]
+
+
+@then("the metadata should contain the target_resources")
+def step_check_metadata_target_resources(context) -> None:
+    """Verify metadata contains target_resources."""
+    assert "metadata" in context.last_status
+    assert "target_resources" in context.last_status["metadata"]
+
+
+@then("the result should contain {count:d} pending subplan")
+def step_check_result_pending(context, count: int) -> None:
+    """Verify result contains pending subplans."""
+    assert len(context.all_subplans["pending"]) == count
+
+
+@then("the result should contain {count:d} running subplan")
+def step_check_result_running(context, count: int) -> None:
+    """Verify result contains running subplans."""
+    assert len(context.all_subplans["running"]) == count
+
+
+@then("the result should contain {count:d} completed subplan")
+def step_check_result_completed(context, count: int) -> None:
+    """Verify result contains completed subplans."""
+    assert len(context.all_subplans["completed"]) == count
+
+
+@then("the result should contain {count:d} failed subplans")
+def step_check_result_failed(context, count: int) -> None:
+    """Verify result contains failed subplans."""
+    assert len(context.all_subplans["failed"]) == count
+
+
+@then("the results should contain {count:d} subplans")
+def step_check_results_count(context, count: int) -> None:
+    """Verify results contain expected number of subplans."""
+    assert len(context.completed_results) == count
+
+
+@then("the results should contain {count:d} failed subplans")
+def step_check_failed_results_count(context, count: int) -> None:
+    """Verify failed results contain expected number of subplans."""
+    assert len(context.failed_subplans) == count
+
+
+@then("each result should have a completed_at timestamp")
+def step_check_each_completed_at(context) -> None:
+    """Verify each result has completed_at timestamp."""
+    for _subplan_id, metadata in context.completed_results.items():
+        assert "completed_at" in metadata
+
+
+@then("each result should have changeset_summary")
+def step_check_each_changeset_summary(context) -> None:
+    """Verify each result has changeset_summary."""
+    for _subplan_id, metadata in context.completed_results.items():
+        assert "changeset_summary" in metadata
+
+
+@then("each result should have an error message")
+def step_check_each_error_message(context) -> None:
+    """Verify each result has error message."""
+    for _subplan_id, metadata in context.failed_subplans.items():
+        assert "error" in metadata
+
+
+@then("each result should have a failed_at timestamp")
+def step_check_each_failed_at(context) -> None:
+    """Verify each result has failed_at timestamp."""
+    for _subplan_id, metadata in context.failed_subplans.items():
+        assert "failed_at" in metadata
+
+
+@then("the result should be false")
+def step_check_result_false(context) -> None:
+    """Verify result is false."""
+    assert context.all_completed is False
+
+
+@then("the result should be true")
+def step_check_result_true(context) -> None:
+    """Verify result is true."""
+    assert context.all_completed is True
+
+
+@then("the status should show:")
+def step_check_status_values(context) -> None:
+    """Verify status values."""
+    for row in context.table:
+        key = row["key"]
+        expected_value = row["value"]
+        if expected_value.lower() in ("true", "false"):
+            expected_value = expected_value.lower() == "true"
+        else:
+            expected_value = int(expected_value)
+        assert context.completion_status[key] == expected_value
+
+
+@then('an error should be raised with message containing "{message}"')
+def step_check_error_message_contains(context, message: str) -> None:
+    """Verify error was raised with message containing text."""
+    assert context.error_raised is True
+    assert message in context.error_message
+
+
+@then("the registry updated_at should be newer than the recorded timestamp")
+def step_check_updated_at_newer(context) -> None:
+    """Verify registry updated_at is newer than recorded timestamp."""
+    assert context.registry.updated_at > context.recorded_timestamp
+
+
+@then("the completed subplan metadata should contain:")
+def step_check_completed_metadata(context) -> None:
+    """Verify completed subplan metadata contains expected values."""
+    metadata = context.registry.completed_subplans[context.last_subplan_id]
+    for row in context.table:
+        key = row["key"]
+        expected_value = row["value"]
+        assert metadata[key] == expected_value

features/subplan_status_registry.feature

@@ -0,0 +1,148 @@
+Feature: Subplan Status Registry and Real-Time Tracking
+  As a parent plan executor
+  I want to track the status of all spawned subplans in real-time
+  So that I can monitor execution progress and access subplan results
+
+  Background:
+    Given a parent plan with ULID "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And a subplan status registry for the parent plan
+
+  Scenario: Register a new subplan in pending state
+    When I register a subplan with:
+      | subplan_id      | 01ARZ3NDEKTSV4RRFFQ69G5FAW |
+      | action_name     | deploy.infrastructure      |
+      | target_resources| resource-1,resource-2      |
+    Then the subplan should be in pending state
+    And the registry should contain 1 pending subplan
+    And the registry should contain 0 running subplans
+
+  Scenario: Move subplan from pending to running state
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I start the subplan
+    Then the subplan should be in running state
+    And the registry should contain 0 pending subplans
+    And the registry should contain 1 running subplan
+    And the subplan should have a started_at timestamp
+
+  Scenario: Complete a running subplan successfully
+    Given a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I complete the subplan with:
+      | changeset_summary | Modified 5 files |
+      | files_modified    | 5                |
+    Then the subplan should be in completed state
+    And the registry should contain 0 running subplans
+    And the registry should contain 1 completed subplan
+    And the subplan should have a completed_at timestamp
+    And the subplan result should contain the changeset summary
+
+  Scenario: Fail a running subplan with error
+    Given a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I fail the subplan with error "Connection timeout"
+    Then the subplan should be in failed state
+    And the registry should contain 0 running subplans
+    And the registry should contain 1 failed subplan
+    And the subplan should have a failed_at timestamp
+    And the subplan error should be "Connection timeout"
+
+  Scenario: Fail a pending subplan
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I fail the subplan with error "Invalid configuration"
+    Then the subplan should be in failed state
+    And the registry should contain 0 pending subplans
+    And the registry should contain 1 failed subplan
+
+  Scenario: Get status of a specific subplan
+    Given a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I get the status of subplan "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    Then the status should be "RUNNING"
+    And the metadata should contain the action_name
+    And the metadata should contain the target_resources
+
+  Scenario: Get all subplans organized by status
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    And a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+    And a completed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAY"
+    When I get all subplans
+    Then the result should contain 1 pending subplan
+    And the result should contain 1 running subplan
+    And the result should contain 1 completed subplan
+    And the result should contain 0 failed subplans
+
+  Scenario: Get completed results from registry
+    Given a completed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    And a completed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+    When I get all completed results
+    Then the results should contain 2 subplans
+    And each result should have a completed_at timestamp
+    And each result should have changeset_summary
+
+  Scenario: Get failed subplans from registry
+    Given a failed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    And a failed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+    When I get all failed subplans
+    Then the results should contain 2 failed subplans
+    And each result should have an error message
+    And each result should have a failed_at timestamp
+
+  Scenario: Check if all subplans are completed
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I check if all subplans are completed
+    Then the result should be false
+    When I complete the subplan
+    And I check if all subplans are completed
+    Then the result should be true
+
+  Scenario: Get completion status summary
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    And a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+    And a completed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAY"
+    And a failed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAZ"
+    When I get the completion status
+    Then the status should show:
+      | total_subplans | 4     |
+      | pending        | 1     |
+      | running        | 1     |
+      | completed      | 1     |
+      | failed         | 1     |
+      | all_done       | false |
+
+  Scenario: Registry rejects starting non-pending subplan
+    Given a running subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I try to start the subplan
+    Then an error should be raised with message containing "not found in pending state"
+
+  Scenario: Registry rejects completing non-running subplan
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I try to complete the subplan
+    Then an error should be raised with message containing "not found in running state"
+
+  Scenario: Registry rejects failing non-active subplan
+    Given a completed subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I try to fail the subplan with error "Test error"
+    Then an error should be raised with message containing "not found in pending or running state"
+
+  Scenario: Registry tracks multiple concurrent subplans
+    When I register 10 subplans
+    And I start 5 of them
+    And I complete 3 of them
+    And I fail 1 of them
+    Then the registry should contain 5 pending subplans
+    And the registry should contain 2 running subplans
+    And the registry should contain 3 completed subplans
+    And the registry should contain 1 failed subplan
+
+  Scenario: Registry updates timestamp on each operation
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+    When I record the registry updated_at timestamp
+    And I wait 100 milliseconds
+    And I start the subplan
+    Then the registry updated_at should be newer than the recorded timestamp
+
+  Scenario: Registry persists subplan metadata through lifecycle
+    Given a registered subplan with ID "01ARZ3NDEKTSV4RRFFQ69G5FAW" with action "deploy.app"
+    When I start the subplan
+    And I complete the subplan with changeset "Modified 3 files"
+    Then the completed subplan metadata should contain:
+      | subplan_id       | 01ARZ3NDEKTSV4RRFFQ69G5FAW |
+      | action_name      | deploy.app                 |
+      | changeset_summary| Modified 3 files           |

src/cleveragents/domain/models/core/subplan_status_registry.py

@@ -0,0 +1,255 @@
+"""Subplan Status Registry for real-time tracking of spawned subplans.
+
+This module provides the SubplanStatusRegistry class which manages the lifecycle
+and status tracking of all subplans spawned by a parent plan. It enables real-time
+status updates, result storage, and database persistence.
+"""
+
+from datetime import datetime
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+
+class SubplanStatusUpdate(BaseModel):
+    """Represents a status update for a subplan."""
+
+    subplan_id: str = Field(..., description="The subplan's plan_id (ULID)")
+    status: str = Field(..., description="New processing state")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    error: str | None = Field(default=None, description="Error message if failed")
+    changeset_summary: str | None = Field(
+        default=None, description="Summary of changes made"
+    )
+    files_modified: int = Field(default=0, ge=0, description="Number of files modified")
+
+
+class SubplanStatusRegistry(BaseModel):
+    """Registry for tracking status of all spawned subplans in a parent plan.
+
+    This registry maintains real-time status information for each subplan,
+    enabling the parent plan to:
+    - Track subplan lifecycle (pending -> running -> complete/failed)
+    - Access subplan results after completion
+    - Monitor concurrent subplan execution
+    - Persist status to database for durability
+
+    The registry is stored on the parent plan and updated as subplans progress
+    through their execution lifecycle.
+    """
+
+    registry_id: str = Field(
+        default_factory=lambda: str(uuid4()),
+        description="Unique identifier for this registry instance",
+    )
+    parent_plan_id: str = Field(..., description="The parent plan's ULID")
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow, description="When this registry was created"
+    )
+    updated_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        description="When this registry was last updated",
+    )
+
+    # Tracking collections
+    pending_subplans: dict[str, dict] = Field(
+        default_factory=dict,
+        description="Subplans waiting to start (subplan_id -> metadata)",
+    )
+    running_subplans: dict[str, dict] = Field(
+        default_factory=dict,
+        description="Subplans currently executing (subplan_id -> metadata)",
+    )
+    completed_subplans: dict[str, dict] = Field(
+        default_factory=dict,
+        description="Successfully completed subplans (subplan_id -> results)",
+    )
+    failed_subplans: dict[str, dict] = Field(
+        default_factory=dict,
+        description="Failed subplans (subplan_id -> error info)",
+    )
+
+    def register_subplan(
+        self,
+        subplan_id: str,
+        action_name: str,
+        target_resources: list[str],
+    ) -> None:
+        """Register a new subplan in the pending state.
+
+        Args:
+            subplan_id: The subplan's ULID
+            action_name: Namespaced action name that created the subplan
+            target_resources: Resource IDs this subplan operates on
+        """
+        metadata = {
+            "subplan_id": subplan_id,
+            "action_name": action_name,
+            "target_resources": target_resources,
+            "registered_at": datetime.utcnow().isoformat(),
+        }
+        self.pending_subplans[subplan_id] = metadata
+        self.updated_at = datetime.utcnow()
+
+    def start_subplan(self, subplan_id: str) -> None:
+        """Move a subplan from pending to running state.
+
+        Args:
+            subplan_id: The subplan's ULID
+
+        Raises:
+            ValueError: If subplan is not in pending state
+        """
+        if subplan_id not in self.pending_subplans:
+            raise ValueError(
+                f"Subplan {subplan_id} not found in pending state. "
+                f"Cannot start subplan that was not registered."
+            )
+
+        metadata = self.pending_subplans.pop(subplan_id)
+        metadata["started_at"] = datetime.utcnow().isoformat()
+        self.running_subplans[subplan_id] = metadata
+        self.updated_at = datetime.utcnow()
+
+    def complete_subplan(
+        self,
+        subplan_id: str,
+        changeset_summary: str | None = None,
+        files_modified: int = 0,
+    ) -> None:
+        """Mark a subplan as successfully completed.
+
+        Args:
+            subplan_id: The subplan's ULID
+            changeset_summary: Summary of changes made by the subplan
+            files_modified: Number of files modified
+
+        Raises:
+            ValueError: If subplan is not in running state
+        """
+        if subplan_id not in self.running_subplans:
+            raise ValueError(
+                f"Subplan {subplan_id} not found in running state. "
+                f"Cannot complete subplan that is not running."
+            )
+
+        metadata = self.running_subplans.pop(subplan_id)
+        metadata["completed_at"] = datetime.utcnow().isoformat()
+        metadata["changeset_summary"] = changeset_summary
+        metadata["files_modified"] = files_modified
+        self.completed_subplans[subplan_id] = metadata
+        self.updated_at = datetime.utcnow()
+
+    def fail_subplan(self, subplan_id: str, error: str) -> None:
+        """Mark a subplan as failed.
+
+        Args:
+            subplan_id: The subplan's ULID
+            error: Error message describing the failure
+
+        Raises:
+            ValueError: If subplan is not in pending or running state
+        """
+        if subplan_id in self.pending_subplans:
+            metadata = self.pending_subplans.pop(subplan_id)
+        elif subplan_id in self.running_subplans:
+            metadata = self.running_subplans.pop(subplan_id)
+        else:
+            raise ValueError(
+                f"Subplan {subplan_id} not found in pending or running state. "
+                f"Cannot fail subplan that is not active."
+            )
+
+        metadata["failed_at"] = datetime.utcnow().isoformat()
+        metadata["error"] = error
+        self.failed_subplans[subplan_id] = metadata
+        self.updated_at = datetime.utcnow()
+
+    def get_subplan_status(self, subplan_id: str) -> dict | None:
+        """Get the current status and metadata for a subplan.
+
+        Args:
+            subplan_id: The subplan's ULID
+
+        Returns:
+            Dictionary with status and metadata, or None if not found
+        """
+        if subplan_id in self.pending_subplans:
+            return {
+                "status": "QUEUED",
+                "metadata": self.pending_subplans[subplan_id],
+            }
+        elif subplan_id in self.running_subplans:
+            return {
+                "status": "RUNNING",
+                "metadata": self.running_subplans[subplan_id],
+            }
+        elif subplan_id in self.completed_subplans:
+            return {
+                "status": "COMPLETED",
+                "metadata": self.completed_subplans[subplan_id],
+            }
+        elif subplan_id in self.failed_subplans:
+            return {
+                "status": "ERRORED",
+                "metadata": self.failed_subplans[subplan_id],
+            }
+        return None
+
+    def get_all_subplans(self) -> dict[str, dict]:
+        """Get all subplans organized by status.
+
+        Returns:
+            Dictionary with keys: pending, running, completed, failed
+        """
+        return {
+            "pending": self.pending_subplans.copy(),
+            "running": self.running_subplans.copy(),
+            "completed": self.completed_subplans.copy(),
+            "failed": self.failed_subplans.copy(),
+        }
+
+    def get_completed_results(self) -> dict[str, dict]:
+        """Get results from all completed subplans.
+
+        Returns:
+            Dictionary mapping subplan_id to result metadata
+        """
+        return self.completed_subplans.copy()
+
+    def get_failed_subplans(self) -> dict[str, dict]:
+        """Get information about all failed subplans.
+
+        Returns:
+            Dictionary mapping subplan_id to error information
+        """
+        return self.failed_subplans.copy()
+
+    def is_all_completed(self) -> bool:
+        """Check if all registered subplans have completed or failed.
+
+        Returns:
+            True if no subplans are pending or running
+        """
+        return len(self.pending_subplans) == 0 and len(self.running_subplans) == 0
+
+    def get_completion_status(self) -> dict:
+        """Get overall completion status of the registry.
+
+        Returns:
+            Dictionary with counts and status information
+        """
+        total = (
+            len(self.pending_subplans)
+            + len(self.running_subplans)
+            + len(self.completed_subplans)
+            + len(self.failed_subplans)
+        )
+        return {
+            "total_subplans": total,
+            "pending": len(self.pending_subplans),
+            "running": len(self.running_subplans),
+            "completed": len(self.completed_subplans),
+            "failed": len(self.failed_subplans),
+            "all_done": self.is_all_completed(),
+        }