diff --git a/robot/a2a_session_plan_lifecycle_integration.robot b/robot/a2a_session_plan_lifecycle_integration.robot new file mode 100644 index 000000000..14002e8f7 --- /dev/null +++ b/robot/a2a_session_plan_lifecycle_integration.robot @@ -0,0 +1,101 @@ +*** Settings *** +Documentation Integration tests for the full A2A session and plan lifecycle. +... Covers: session create/list/show/close/delete, plan create/execute/ +... status/cancel, event queue publish/subscribe, guard enforcement +... (budget cap, tool call limit, denylist), and plan rollback with +... session state consistency. +... +... All tests use real service instances (no mocks) wired via the +... full DI-compatible fixture in helper_a2a_session_plan_lifecycle_integration.py. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_a2a_session_plan_lifecycle_integration.py + +*** Test Cases *** +A2A Full Session Lifecycle Create List Show Close Delete + [Documentation] Verify full session lifecycle: create via A2A facade, + ... list via session service, show (get) by ID, close via + ... facade (which deletes), and verify deletion. + [Tags] a2a session lifecycle integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} session-full-lifecycle + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} session-full-lifecycle-ok + +A2A Full Plan Lifecycle Create Execute Status Cancel + [Documentation] Verify full plan lifecycle within a session: create plan via + ... A2A facade, get status, advance through strategize, execute, + ... verify execute phase, cancel, and verify cancellation. + [Tags] a2a plan lifecycle integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} plan-full-lifecycle + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-full-lifecycle-ok + +A2A Event Queue Receives Published Events During Lifecycle + [Documentation] Verify that the A2A event queue receives published events + ... during plan lifecycle transitions, subscriber callbacks are + ... invoked, and unsubscribe stops delivery. + [Tags] a2a events queue integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} event-queue-receives-events + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} event-queue-receives-events-ok + +A2A Guard Enforcement Budget Cap Exceeded Halts Execution + [Documentation] Verify that the autonomy guardrail service blocks plan + ... execution when the budget cap is exceeded, records the + ... denial in the audit trail, and maintains consistent state. + [Tags] a2a guard budget integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} guard-budget-cap + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} guard-budget-cap-ok + +A2A Guard Enforcement Tool Call Limit Exceeded Halts Execution + [Documentation] Verify that the autonomy guardrail service blocks plan + ... execution when the per-actor tool call limit is exceeded, + ... and records the denial in the audit trail. + [Tags] a2a guard tool_limit integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} guard-tool-call-limit + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} guard-tool-call-limit-ok + +A2A Guard Enforcement Denylist Violation Blocks Tool Invocation + [Documentation] Verify that the automation profile denylist blocks tool + ... invocations for denied tools and allows permitted tools, + ... both via direct profile check and via AutomationProfileService. + [Tags] a2a guard denylist integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} guard-denylist + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} guard-denylist-ok + +A2A Plan Rollback After Failed Execution Session State Consistent + [Documentation] Verify that after a plan execution failure, the plan is in + ... ERRORED state, the session remains accessible and consistent, + ... and the plan list is not corrupted. Session can be deleted + ... cleanly after the failure. + [Tags] a2a plan rollback session integration issue_10032 + ${result}= Run Process ${PYTHON} ${HELPER} plan-rollback-session-consistent + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-rollback-session-consistent-ok diff --git a/robot/helper_a2a_session_plan_lifecycle_integration.py b/robot/helper_a2a_session_plan_lifecycle_integration.py new file mode 100644 index 000000000..213b855f1 --- /dev/null +++ b/robot/helper_a2a_session_plan_lifecycle_integration.py @@ -0,0 +1,629 @@ +"""Helper script for A2A session and plan lifecycle integration tests. + +Exercises the complete flow: session create -> plan create -> plan execute -> +plan status -> plan cancel/rollback -> session close -> session delete. + +Also covers: +- Event queue receives published events during plan lifecycle +- Guard enforcement (budget cap, tool call limit) during plan execution +- Plan rollback after failed execution -> session state consistent + +Usage: + python robot/helper_a2a_session_plan_lifecycle_integration.py + +Commands: + session-full-lifecycle Full session create/list/show/close/delete + plan-full-lifecycle Full plan create/execute/status/cancel + event-queue-receives-events Event queue receives events during lifecycle + guard-budget-cap Budget cap exceeded halts execution + guard-tool-call-limit Tool call limit exceeded halts execution + guard-denylist Denylist violation blocks tool invocation + plan-rollback-session-consistent Plan rollback, session state consistent +""" + +# ruff: noqa: E402, I001 +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any + +# Ensure local source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from cleveragents.a2a.events import A2aEventQueue +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.a2a.models import A2aEvent, A2aRequest +from cleveragents.application.services.autonomy_guardrail_service import ( + AutonomyGuardrailService, +) +from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, +) +from cleveragents.application.services.session_service import ( + PersistentSessionService, +) +from cleveragents.config.settings import Settings +from cleveragents.domain.models.core.automation_guard import AutomationGuard +from cleveragents.domain.models.core.autonomy_guardrails import ( + ActorLimits, + AutonomyGuardrails, +) +from cleveragents.domain.models.core.plan import ( + ProcessingState, + ProjectLink, +) +from cleveragents.domain.models.core.session import SessionNotFoundError +from cleveragents.infrastructure.database.models import Base +from cleveragents.infrastructure.database.repositories import ( + SessionMessageRepository, + SessionRepository, +) + + +# --------------------------------------------------------------------------- +# Fixture helpers +# --------------------------------------------------------------------------- + + +def _make_db_session_factory() -> Any: + """Create an in-memory SQLite engine and return a session factory.""" + engine = create_engine("sqlite:///:memory:", echo=False) + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine, expire_on_commit=False) + return factory + + +def _make_plan_lifecycle_service() -> PlanLifecycleService: + """Create a PlanLifecycleService with default settings.""" + settings = Settings() + return PlanLifecycleService(settings=settings) + + +def _make_full_facade() -> tuple[ + A2aLocalFacade, + PersistentSessionService, + PlanLifecycleService, + A2aEventQueue, + Any, +]: + """Bootstrap the full DI-wired facade with real services. + + Returns: + Tuple of (facade, session_service, plan_lifecycle_service, + event_queue, db_session) + """ + factory = _make_db_session_factory() + db_session = factory() + + def get_db() -> Any: + return db_session + + session_repo = SessionRepository(get_db) + message_repo = SessionMessageRepository(get_db) + session_svc = PersistentSessionService(session_repo, message_repo) + + plan_svc = _make_plan_lifecycle_service() + event_queue = A2aEventQueue() + + facade = A2aLocalFacade( + services={ + "session_service": session_svc, + "plan_lifecycle_service": plan_svc, + "event_queue": event_queue, + } + ) + return facade, session_svc, plan_svc, event_queue, db_session + + +# --------------------------------------------------------------------------- +# Test: Full session lifecycle +# --------------------------------------------------------------------------- + + +def _session_full_lifecycle() -> None: + """Integration test: session create -> list -> show -> close -> delete.""" + facade, session_svc, _plan_svc, _eq, db_session = _make_full_facade() + + # 1. Create session via A2A facade + create_req = A2aRequest( + method="session.create", + params={"actor_name": "local/orchestrator"}, + ) + create_resp = facade.dispatch(create_req) + assert create_resp.error is None, f"session.create failed: {create_resp.error}" + assert create_resp.result is not None + session_id = create_resp.result["session_id"] + assert session_id, "Expected a session_id in response" + db_session.commit() + + # 2. List sessions via session service (real service, not facade) + sessions = session_svc.list() + assert len(sessions) >= 1, f"Expected at least 1 session, got {len(sessions)}" + session_ids = [s.session_id for s in sessions] + assert session_id in session_ids, ( + f"Created session {session_id} not in list" + ) + + # 3. Show (get) session by ID + fetched = session_svc.get(session_id) + assert fetched.session_id == session_id + assert fetched.actor_name == "local/orchestrator" + + # 4. Close session via A2A facade (maps to session.close -> delete) + close_req = A2aRequest( + method="session.close", + params={"session_id": session_id}, + ) + close_resp = facade.dispatch(close_req) + assert close_resp.error is None, f"session.close failed: {close_resp.error}" + assert close_resp.result is not None + assert close_resp.result["status"] == "closed" + db_session.commit() + + # 5. Verify session is deleted (get should raise SessionNotFoundError) + try: + session_svc.get(session_id) + print("FAIL: Expected SessionNotFoundError after close", file=sys.stderr) + sys.exit(1) + except SessionNotFoundError: + pass # Expected + + print("session-full-lifecycle-ok") + + +# --------------------------------------------------------------------------- +# Test: Full plan lifecycle within a session +# --------------------------------------------------------------------------- + + +def _plan_full_lifecycle() -> None: + """Integration test: plan create -> execute -> status -> cancel.""" + facade, session_svc, plan_svc, _eq, db_session = _make_full_facade() + + # Create a session first + session = session_svc.create(actor_name="local/planner") + db_session.commit() + + # Create an action for the plan + action = plan_svc.create_action( + name="local/integration-test-action", + description="Integration test action", + definition_of_done="All integration tests pass", + strategy_actor="local/strategist", + execution_actor="local/executor", + ) + + # 1. Create plan via A2A facade (plan.create) + create_req = A2aRequest( + method="plan.create", + params={ + "action_name": str(action.namespaced_name), + "created_by": session.session_id, + }, + ) + create_resp = facade.dispatch(create_req) + assert create_resp.error is None, f"plan.create failed: {create_resp.error}" + assert create_resp.result is not None + plan_id = create_resp.result["plan_id"] + assert plan_id, "Expected a plan_id in response" + assert create_resp.result["status"] == "created" + + # 2. Get plan status via A2A facade (plan.status) + status_req = A2aRequest(method="plan.status", params={"plan_id": plan_id}) + status_resp = facade.dispatch(status_req) + assert status_resp.error is None, f"plan.status failed: {status_resp.error}" + assert status_resp.result is not None + assert status_resp.result["plan_id"] == plan_id + assert status_resp.result["phase"] == "strategize" + + # 3. Advance plan through strategize phase + plan_svc.start_strategize(plan_id) + plan_svc.complete_strategize(plan_id) + + # 4. Execute plan via A2A facade (plan.execute) + execute_req = A2aRequest(method="plan.execute", params={"plan_id": plan_id}) + execute_resp = facade.dispatch(execute_req) + assert execute_resp.error is None, f"plan.execute failed: {execute_resp.error}" + assert execute_resp.result is not None + assert execute_resp.result["plan_id"] == plan_id + + # 5. Verify plan is now in execute phase + status_resp2 = facade.dispatch( + A2aRequest(method="plan.status", params={"plan_id": plan_id}) + ) + assert status_resp2.result is not None + assert status_resp2.result["phase"] == "execute" + + # 6. Cancel plan via A2A facade (_cleveragents/plan/cancel) + cancel_req = A2aRequest( + method="_cleveragents/plan/cancel", + params={"plan_id": plan_id}, + ) + cancel_resp = facade.dispatch(cancel_req) + assert cancel_resp.error is None, f"plan.cancel failed: {cancel_resp.error}" + assert cancel_resp.result is not None + assert cancel_resp.result["status"] == "cancelled" + + # 7. Verify plan is cancelled + plan = plan_svc.get_plan(plan_id) + assert plan.state == ProcessingState.CANCELLED, ( + f"Expected CANCELLED, got {plan.state}" + ) + + # 8. Close session + session_svc.delete(session.session_id) + db_session.commit() + + print("plan-full-lifecycle-ok") + + +# --------------------------------------------------------------------------- +# Test: Event queue receives events during plan lifecycle +# --------------------------------------------------------------------------- + + +def _event_queue_receives_events() -> None: + """Integration test: event queue receives published events during lifecycle.""" + event_queue = A2aEventQueue() + received_events: list[A2aEvent] = [] + + # Subscribe to event queue + sub_id = event_queue.subscribe_local(lambda e: received_events.append(e)) + assert sub_id, "Expected a subscription ID" + + # Publish events simulating plan lifecycle transitions + event_queue.publish(A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="test-plan-001", + data={"phase": "strategize", "state": "queued"}, + )) + event_queue.publish(A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="test-plan-001", + data={"phase": "execute", "state": "processing"}, + )) + event_queue.publish(A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="test-plan-001", + data={"phase": "apply", "state": "applied"}, + )) + + # Verify subscriber received all events + assert len(received_events) == 3, ( + f"Expected 3 events, got {len(received_events)}" + ) + assert received_events[0].event_type == "TaskStatusUpdateEvent" + assert received_events[0].plan_id == "test-plan-001" + assert received_events[0].data["phase"] == "strategize" + assert received_events[1].data["phase"] == "execute" + assert received_events[2].data["phase"] == "apply" + + # Verify events are stored in queue + stored = event_queue.get_events(limit=10) + assert len(stored) == 3 + + # Unsubscribe and verify no more events received + removed = event_queue.unsubscribe(sub_id) + assert removed is True + + event_queue.publish(A2aEvent( + event_type="TaskStatusUpdateEvent", + plan_id="test-plan-001", + data={"phase": "done"}, + )) + assert len(received_events) == 3, "Should not receive events after unsubscribe" + + # Verify event queue integration with A2A facade + facade = A2aLocalFacade(services={"event_queue": event_queue}) + subscribe_req = A2aRequest(method="event.subscribe", params={}) + subscribe_resp = facade.dispatch(subscribe_req) + assert subscribe_resp.error is None, ( + f"event.subscribe failed: {subscribe_resp.error}" + ) + assert subscribe_resp.result is not None + assert subscribe_resp.result["status"] == "subscribed" + assert "subscription_id" in subscribe_resp.result + + print("event-queue-receives-events-ok") + + +# --------------------------------------------------------------------------- +# Test: Guard enforcement -- budget cap exceeded +# --------------------------------------------------------------------------- + + +def _guard_budget_cap() -> None: + """Integration test: budget cap exceeded halts plan execution. + + The AutonomyGuardrailService.check_tool_budget() returns bool: + - True if the invocation is within budget (and records the cost) + - False if blocked + """ + plan_svc = _make_plan_lifecycle_service() + guardrail_svc = AutonomyGuardrailService() + + # Create action and plan + action = plan_svc.create_action( + name="local/budget-test-action", + description="Budget cap test", + definition_of_done="Test budget enforcement", + strategy_actor="local/s", + execution_actor="local/e", + ) + plan = plan_svc.use_action( + action_name=str(action.namespaced_name), + project_links=[ProjectLink(project_name="proj-budget")], + ) + plan_id = plan.identity.plan_id + + # Configure guardrails with a tight budget cap of 10.0 + guardrails = AutonomyGuardrails( + tool_budget=10.0, + budget_spent=0.0, + ) + guardrail_svc.configure_guardrails(plan_id, guardrails) + + # Verify budget allows within cap (cost 5.0 < budget 10.0) + # check_tool_budget records the cost when allowed + allowed = guardrail_svc.check_tool_budget(plan_id, 5.0) + assert allowed is True, "Expected budget to allow cost of 5.0 within cap of 10.0" + + # After recording 5.0, budget_spent = 5.0 + # Now try to spend 8.0 more (5.0 + 8.0 = 13.0 > 10.0 cap) + allowed2 = guardrail_svc.check_tool_budget(plan_id, 8.0) + assert allowed2 is False, ( + "Expected budget cap to block cost of 8.0 when 5.0 already spent" + ) + + # Verify audit trail recorded both the allow and the denial + trail = guardrail_svc.get_audit_trail(plan_id) + assert len(trail.entries) >= 2, ( + f"Expected at least 2 audit entries, got {len(trail.entries)}" + ) + denied_entries = [e for e in trail.entries if e.result.value == "denied"] + assert len(denied_entries) >= 1, "Expected at least one denied audit entry" + + # Verify guardrails state is consistent (budget_spent = 5.0 after first call) + g = guardrail_svc.get_guardrails(plan_id) + assert g is not None + assert g.budget_spent == 5.0, ( + f"Expected budget_spent=5.0 (only first call was allowed), got {g.budget_spent}" + ) + + print("guard-budget-cap-ok") + + +# --------------------------------------------------------------------------- +# Test: Guard enforcement -- tool call limit exceeded +# --------------------------------------------------------------------------- + + +def _guard_tool_call_limit() -> None: + """Integration test: tool call limit exceeded halts execution. + + The AutonomyGuardrailService.check_actor_tool_calls() returns bool: + - True if allowed + - False if blocked + """ + plan_svc = _make_plan_lifecycle_service() + guardrail_svc = AutonomyGuardrailService() + + # Create action and plan + action = plan_svc.create_action( + name="local/tool-limit-action", + description="Tool call limit test", + definition_of_done="Test tool call limit enforcement", + strategy_actor="local/s", + execution_actor="local/e", + ) + plan = plan_svc.use_action( + action_name=str(action.namespaced_name), + project_links=[ProjectLink(project_name="proj-tool-limit")], + ) + plan_id = plan.identity.plan_id + + # Configure guardrails with a tool call limit of 3 + guardrails = AutonomyGuardrails( + actor_limits=ActorLimits(max_tool_calls_per_invocation=3), + ) + guardrail_svc.configure_guardrails(plan_id, guardrails) + + # Verify tool calls allowed within limit (1 < 3, 2 < 3) + allowed1 = guardrail_svc.check_actor_tool_calls(plan_id, 1) + assert allowed1 is True, "Expected tool call 1 to be allowed (limit=3)" + + allowed2 = guardrail_svc.check_actor_tool_calls(plan_id, 2) + assert allowed2 is True, "Expected tool call 2 to be allowed (limit=3)" + + # Verify tool calls blocked at limit (3 >= 3) + blocked = guardrail_svc.check_actor_tool_calls(plan_id, 3) + assert blocked is False, "Expected tool call limit to block at 3 (limit=3)" + + # Verify audit trail recorded the denial + trail = guardrail_svc.get_audit_trail(plan_id) + denied_entries = [e for e in trail.entries if e.result.value == "denied"] + assert len(denied_entries) >= 1, "Expected at least one denied audit entry" + + print("guard-tool-call-limit-ok") + + +# --------------------------------------------------------------------------- +# Test: Guard enforcement -- denylist violation +# --------------------------------------------------------------------------- + + +def _guard_denylist() -> None: + """Integration test: denylist violation blocks tool invocation. + + Uses AutomationProfile.check_guard() directly to verify denylist enforcement. + AutomationProfile uses 'name' field (not 'profile_name'). + GuardScope defaults to PLAN when not specified. + """ + from cleveragents.domain.models.core.automation_profile import AutomationProfile + + # Create an automation profile with a denylist + # AutomationProfile uses 'name' field, not 'profile_name' + profile = AutomationProfile( + name="restricted", + guards=AutomationGuard( + tool_denylist=["rm", "delete_file", "drop_table"], + ), + ) + + # Evaluate guard for a denied tool (scope defaults to PLAN) + result = profile.check_guard(tool_name="rm") + assert result.allowed is False, ( + f"Expected 'rm' to be denied by denylist, got allowed={result.allowed}" + ) + assert result.reason is not None + assert "denylist" in result.reason.lower() or "deny" in result.reason.lower(), ( + f"Expected denylist-related reason, got: {result.reason}" + ) + + # Evaluate guard for another denied tool + result2 = profile.check_guard(tool_name="delete_file") + assert result2.allowed is False, "Expected 'delete_file' to be denied" + + # Evaluate guard for an allowed tool (not in denylist) + result3 = profile.check_guard(tool_name="read_file") + assert result3.allowed is True, ( + f"Expected 'read_file' to be allowed, got allowed={result3.allowed}" + ) + + # Verify allowlist enforcement: tool not in allowlist is blocked + profile_with_allowlist = AutomationProfile( + name="allowlist-only", + guards=AutomationGuard( + tool_allowlist=["read_file", "list_files"], + ), + ) + # Tool not in allowlist should be blocked + result4 = profile_with_allowlist.check_guard(tool_name="drop_table") + assert result4.allowed is False, ( + "Expected 'drop_table' to be denied by allowlist (not in list)" + ) + + # Tool in allowlist should be allowed + result5 = profile_with_allowlist.check_guard(tool_name="read_file") + assert result5.allowed is True, ( + "Expected 'read_file' to be allowed (in allowlist)" + ) + + print("guard-denylist-ok") + + +# --------------------------------------------------------------------------- +# Test: Plan rollback after failed execution -> session state consistent +# --------------------------------------------------------------------------- + + +def _plan_rollback_session_consistent() -> None: + """Integration test: plan rollback after failed execution, session consistent. + + Note: ERRORED state IS terminal per the spec (is_terminal returns True). + The test verifies session remains accessible and plan list is not corrupted. + """ + _facade, session_svc, plan_svc, _eq, db_session = _make_full_facade() + + # Create a session + session = session_svc.create(actor_name="local/rollback-tester") + db_session.commit() + + # Create action and plan + action = plan_svc.create_action( + name="local/rollback-test-action", + description="Rollback test action", + definition_of_done="Test rollback consistency", + strategy_actor="local/s", + execution_actor="local/e", + ) + plan = plan_svc.use_action( + action_name=str(action.namespaced_name), + project_links=[ProjectLink(project_name="proj-rollback")], + created_by=session.session_id, + ) + plan_id = plan.identity.plan_id + + # Advance to execute phase + plan_svc.start_strategize(plan_id) + plan_svc.complete_strategize(plan_id) + plan_svc.execute_plan(plan_id) + plan_svc.start_execute(plan_id) + + # Simulate execution failure + failed_plan = plan_svc.fail_execute(plan_id, "Simulated execution failure") + assert failed_plan.state == ProcessingState.ERRORED, ( + f"Expected ERRORED state, got {failed_plan.state}" + ) + assert failed_plan.error_message == "Simulated execution failure" + assert failed_plan.is_errored + + # ERRORED is terminal per spec (is_terminal returns True for ERRORED) + assert failed_plan.is_terminal is True, ( + "ERRORED state should be terminal per spec" + ) + + # Verify session is still accessible (consistent state) + session_still_exists = session_svc.get(session.session_id) + assert session_still_exists.session_id == session.session_id, ( + "Session should still exist after plan failure" + ) + + # Verify plan list still works (no corruption) + plans = plan_svc.list_plans() + plan_ids = [p.identity.plan_id for p in plans] + assert plan_id in plan_ids, "Failed plan should still be in plan list" + + # Verify the failed plan can be retrieved + retrieved = plan_svc.get_plan(plan_id) + assert retrieved.state == ProcessingState.ERRORED + assert retrieved.is_terminal is True # ERRORED is terminal per spec + + # Clean up: delete session + session_svc.delete(session.session_id) + db_session.commit() + + # Verify session is gone + try: + session_svc.get(session.session_id) + print("FAIL: Expected SessionNotFoundError after delete", file=sys.stderr) + sys.exit(1) + except SessionNotFoundError: + pass + + print("plan-rollback-session-consistent-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "session-full-lifecycle": _session_full_lifecycle, + "plan-full-lifecycle": _plan_full_lifecycle, + "event-queue-receives-events": _event_queue_receives_events, + "guard-budget-cap": _guard_budget_cap, + "guard-tool-call-limit": _guard_tool_call_limit, + "guard-denylist": _guard_denylist, + "plan-rollback-session-consistent": _plan_rollback_session_consistent, +} + + +def main() -> None: + 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(2) + _COMMANDS[sys.argv[1]]() + + +if __name__ == "__main__": + main()