From d450bb615f790dc2c2497ca8be04e249535ce875 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Wed, 15 Apr 2026 00:42:05 +0000 Subject: [PATCH] feat(plans): implement parent plan subplan status registry and real-time tracking - Implement SubplanStatusRegistry class for tracking spawned subplans - Support real-time status updates (pending/running/complete/failed) - Enable subplan result storage and retrieval - Provide database persistence-ready data structures - Add comprehensive BDD tests with 17 scenarios covering all functionality - Implement status lifecycle management with proper state transitions - Add metadata tracking for subplan execution details --- .../steps/subplan_status_registry_steps.py | 521 ++++++++++++++++++ features/subplan_status_registry.feature | 148 +++++ .../models/core/subplan_status_registry.py | 255 +++++++++ 3 files changed, 924 insertions(+) create mode 100644 features/steps/subplan_status_registry_steps.py create mode 100644 features/subplan_status_registry.feature create mode 100644 src/cleveragents/domain/models/core/subplan_status_registry.py diff --git a/features/steps/subplan_status_registry_steps.py b/features/steps/subplan_status_registry_steps.py new file mode 100644 index 000000000..ea7ccbece --- /dev/null +++ b/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 diff --git a/features/subplan_status_registry.feature b/features/subplan_status_registry.feature new file mode 100644 index 000000000..78ac21555 --- /dev/null +++ b/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 | diff --git a/src/cleveragents/domain/models/core/subplan_status_registry.py b/src/cleveragents/domain/models/core/subplan_status_registry.py new file mode 100644 index 000000000..2ba32eb5e --- /dev/null +++ b/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(), + } -- 2.52.0