refactor(plan-and-task): migrate to workflow DSL with sentinel gate pattern

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

Author: MoveCloudROY

examples/e2e/plan_and_task/controller.py

@@ -14,8 +14,8 @@
 )
 from examples.e2e.plan_and_task.state_models import RuntimeState, ReviewVerdict
 from examples.e2e.plan_and_task.state_machine import (
-    VALID_TRANSITIONS,
     WorkflowStateMachine,
+    _COMPILED_WORKFLOW,
 )
 
 logger = get_logger(__name__)
@@ -539,7 +539,7 @@ def _append_task_event(
         )
 
     def _allowed_transitions(self, state: RuntimeState) -> set[str]:
-        return VALID_TRANSITIONS.get(state.phase, set())
+        return {t.target_state_id for t in _COMPILED_WORKFLOW.transitions_by_state.get(state.phase, ())}
 
     def _utcnow_isoformat(self) -> str:
         return datetime.datetime.now(datetime.UTC).isoformat()

examples/e2e/plan_and_task/main.py

@@ -15,11 +15,11 @@
 from ecs_agent.components import (
     ConversationComponent,
     LLMComponent,
-    RenderedSystemPromptComponent,
     SubagentRegistryComponent,
     SubagentSessionTableComponent,
     ToolRegistryComponent,
     UserPromptConfigComponent,
+    WorkflowRuntimeComponent,
 )
 from ecs_agent.components.definitions import ScriptHandler
 from ecs_agent.core import Runner, World
@@ -32,14 +32,15 @@
 from ecs_agent.providers import Model
 from ecs_agent.providers.config import ApiFormat
 from ecs_agent.providers.protocol import LLMModel
+from ecs_agent.systems import WorkflowStateSystem
 from ecs_agent.systems.error_handling import ErrorHandlingSystem
 from ecs_agent.tools import BuiltinToolsSkill
 from ecs_agent.skills.manager import SkillManager
 from ecs_agent.skills.discovery import discover_skills
 from ecs_agent.systems.memory import MemorySystem
 from ecs_agent.systems.reasoning import ReasoningSystem
 from ecs_agent.systems.subagent import SubagentSystem
-from ecs_agent.systems.system_prompt_render_system import SystemPromptRenderSystem, render_prompt_template
+from ecs_agent.systems.system_prompt_render_system import SystemPromptRenderSystem
 from ecs_agent.systems.tool_execution import ToolExecutionSystem
 from ecs_agent.systems.user_prompt_normalization_system import (
     UserPromptNormalizationSystem,
@@ -53,6 +54,7 @@
 )
 
 from ecs_agent.accounting import AccountingSubscriber
+from ecs_agent.workflows import install_workflow
 from examples.e2e.plan_and_task.billing import BillingSubscriber
 from examples.e2e.plan_and_task.scratchbook_adapter import (
     PlanTaskScratchbookAdapter as ArtifactAdapter,
@@ -61,10 +63,8 @@
 from examples.e2e.plan_and_task.controller import PlanController, ResumeAction
 from examples.e2e.plan_and_task.prompts import (
     ADVISOR_SYSTEM_PROMPT,
-    PLAN_MAIN_AGENT_SYSTEM_PROMPT,
     PLAN_QA_REVIEW_SYSTEM_PROMPT,
     QA_SYSTEM_PROMPT,
-    TASK_MAIN_AGENT_SYSTEM_PROMPT,
     WRITE_PLAN_SYSTEM_PROMPT,
     build_write_plan_prompt,
 )
@@ -73,6 +73,7 @@
     derive_workflow_id_from_llm,
 )
 from examples.e2e.plan_and_task.state_models import RuntimeState
+from examples.e2e.plan_and_task.workflow_spec import PLAN_TASK_WORKFLOW_SPEC
 
 logger = get_logger(__name__)
 
@@ -129,7 +130,7 @@ def build_plan_task_world(
     world.add_component(
         agent_id,
         SystemPromptConfigSpec(
-            template_source=PromptTemplateSource(inline=PLAN_MAIN_AGENT_SYSTEM_PROMPT)
+            template_source=PromptTemplateSource(inline="${_workflow_state_prompt}")
         ),
     )
     world.add_component(agent_id, ToolRegistryComponent(tools={}, handlers={}))
@@ -201,39 +202,19 @@ def build_plan_task_world(
     runtime_state: list[RuntimeState | None] = [None]
     _base_dir = base_dir or _WORKFLOW_BASE_DIR
 
-    def _swap_to_task_prompt(w: World, eid: EntityId, trigger_text: str) -> None:
-        spec = w.get_component(eid, SystemPromptConfigSpec)
-        if spec is None:
-            return
-        if (spec.template_source.inline or "") == TASK_MAIN_AGENT_SYSTEM_PROMPT:
-            return
-        new_spec = SystemPromptConfigSpec(
-            template_source=PromptTemplateSource(inline=TASK_MAIN_AGENT_SYSTEM_PROMPT)
-        )
-        w.add_component(eid, new_spec)
-        rendered_text, snapshot = render_prompt_template(
-            template=TASK_MAIN_AGENT_SYSTEM_PROMPT, world=w, entity=eid
-        )
-        w.add_component(
-            eid,
-            RenderedSystemPromptComponent(
-                text=rendered_text,
-                placeholder_snapshot=snapshot,
-            ),
-        )
-        llm = w.get_component(eid, LLMComponent)
-        if llm is not None:
-            llm.system_prompt = rendered_text
+    def _sync_workflow_state(w: World, eid: EntityId, phase: str) -> None:
+        runtime = w.get_component(eid, WorkflowRuntimeComponent)
+        if runtime is not None:
+            runtime.current_state_id = phase
+
+    def _activate_task_phase(
+        w: World, eid: EntityId, phase: str, trigger_text: str
+    ) -> None:
+        _sync_workflow_state(w, eid, phase)
         conv = w.get_component(eid, ConversationComponent)
         if conv is not None:
             conv.messages.clear()
             conv.messages.append(Message(role="user", content=trigger_text))
-        logger.info(
-            "plan_task_system_prompt_switched",
-            entity_id=int(eid),
-            from_prompt="PLAN_MAIN_AGENT",
-            to_prompt="TASK_MAIN_AGENT",
-        )
 
 
     def _load_workflow(w: World, eid: EntityId, workflow_id: str) -> RuntimeState:
@@ -243,6 +224,7 @@ def _load_workflow(w: World, eid: EntityId, workflow_id: str) -> RuntimeState:
         adapter_ref[0] = new_adapter
         runtime_state[0] = state
         w.add_component(eid, build_scratchbook_prompt_config(workflow_id))
+        _sync_workflow_state(w, eid, state.phase)
         return state
 
     async def _on_delegation_completed(event: DelegationCompletedEvent) -> None:
@@ -275,11 +257,13 @@ async def _on_delegation_completed(event: DelegationCompletedEvent) -> None:
                 runtime_state[0] = controller.handle_advisor_review(
                     current, adapter, verdict_str, notes=event.result[:500]
                 )
+                _sync_workflow_state(world, agent_id, _require_state(runtime_state[0]).phase)
             elif event.subagent_name == "qa":
                 new_state = controller.handle_qa_review(
                     current, adapter, verdict_str, notes=event.result[:500]
                 )
                 runtime_state[0] = new_state
+                _sync_workflow_state(world, agent_id, new_state.phase)
                 if new_state.phase == "WRITE_PLAN":
                     conv = world.get_component(agent_id, ConversationComponent)
                     if conv is not None:
@@ -298,10 +282,12 @@ async def _on_delegation_completed(event: DelegationCompletedEvent) -> None:
                 runtime_state[0] = controller.handle_plan_qa_review(
                     current, adapter, verdict_str, notes=event.result[:500]
                 )
+                _sync_workflow_state(world, agent_id, _require_state(runtime_state[0]).phase)
             elif event.subagent_name == "plan_writer":
                 runtime_state[0] = controller.handle_write_plan_completed(
                     current, adapter
                 )
+                _sync_workflow_state(world, agent_id, _require_state(runtime_state[0]).phase)
         except ValueError as exc:
             logger.error(
                 "plan_task_verdict_recording_failed",
@@ -335,6 +321,7 @@ async def _handle_plan_start(
                 _entity_id, build_scratchbook_prompt_config(derived_id)
             )
             runtime_state[0] = controller.handle_plan_start(adapter_ref[0], description)
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             status = controller.get_plan_status(_require_state(runtime_state[0]))
             logger.info(
                 "plan_task_command_plan_start",
@@ -365,6 +352,7 @@ async def _handle_plan_finalize(
             runtime_state[0] = controller.handle_plan_finalize(
                 _require_state(runtime_state[0]), _require_adapter(adapter_ref[0])
             )
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             logger.info(
                 "plan_task_command_plan_finalize",
                 workflow_id=_require_state(runtime_state[0]).workflow_id,
@@ -380,12 +368,15 @@ async def _handle_task_start(
         try:
             from examples.e2e.plan_and_task.task_exec import TaskExec
 
-            # Guard against re-triggering: after _swap_to_task_prompt the /task:start
-            # message stays as the last role="user" entry (tool results use role="tool"),
-            # so the trigger would fire on every subsequent tick. Skip re-initialization
-            # when we are already running in TASK mode.
-            spec = _world.get_component(_entity_id, SystemPromptConfigSpec)
-            if spec is not None and (spec.template_source.inline or "") == TASK_MAIN_AGENT_SYSTEM_PROMPT:
+            # Guard against re-triggering: the /task:start message stays as the last
+            # role="user" entry (tool results use role="tool"), so the trigger would
+            # fire on every subsequent tick. Skip re-initialization once task execution
+            # is already active.
+            workflow_runtime = _world.get_component(_entity_id, WorkflowRuntimeComponent)
+            if (
+                workflow_runtime is not None
+                and workflow_runtime.current_state_id == "TASK_RUNNING"
+            ):
                 return None
 
             if runtime_state[0] is None:
@@ -409,7 +400,12 @@ async def _handle_task_start(
             runtime_state[0] = task_exec.initialize_task_queue(
                 current, _require_adapter(adapter_ref[0])
             )
-            _swap_to_task_prompt(_world, _entity_id, _user_text)
+            _activate_task_phase(
+                _world,
+                _entity_id,
+                _require_state(runtime_state[0]).phase,
+                _user_text,
+            )
             s = _require_state(runtime_state[0])
             logger.info(
                 "plan_task_command_task_start",
@@ -458,10 +454,17 @@ async def _handle_task_resume(
         _world: World, _entity_id: EntityId, _user_text: str
     ) -> str | None:
         try:
+            if runtime_state[0] is not None and runtime_state[0].phase == "TASK_RUNNING":
+                return None
             runtime_state[0] = controller.handle_task_resume(
                 _require_state(runtime_state[0]), _require_adapter(adapter_ref[0])
             )
-            _swap_to_task_prompt(_world, _entity_id, _user_text)
+            _activate_task_phase(
+                _world,
+                _entity_id,
+                _require_state(runtime_state[0]).phase,
+                _user_text,
+            )
             logger.info(
                 "plan_task_command_task_resume",
                 workflow_id=_require_state(runtime_state[0]).workflow_id,
@@ -484,6 +487,7 @@ async def _handle_task_replan(
                 _require_adapter(adapter_ref[0]),
                 reason,
             )
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             s = _require_state(runtime_state[0])
             logger.info(
                 "plan_task_command_task_replan",
@@ -504,6 +508,7 @@ async def _handle_task_abort(
                 _require_adapter(adapter_ref[0]),
                 reason="user abort",
             )
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             s = _require_state(runtime_state[0])
             logger.info(
                 "plan_task_command_task_abort",
@@ -562,6 +567,7 @@ async def _handle_plan_write(
             runtime_state[0] = controller.handle_write_plan(
                 _require_state(runtime_state[0]), adapter
             )
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             s = _require_state(runtime_state[0])
             logger.info("plan_task_command_plan_write", workflow_id=s.workflow_id)
             draft_path = str(
@@ -587,6 +593,7 @@ async def _handle_plan_qa_review(
                 verdict,
                 notes=notes,
             )
+            _sync_workflow_state(_world, _entity_id, _require_state(runtime_state[0]).phase)
             s = _require_state(runtime_state[0])
             logger.info(
                 "plan_task_command_plan_qa_review",
@@ -684,6 +691,9 @@ async def _handle_plan_qa_review(
         UserPromptConfigComponent(triggers=triggers, script_handlers=script_handlers),
     )
 
+    install_workflow(world, agent_id, PLAN_TASK_WORKFLOW_SPEC, agent_key="main")
+
+    world.register_system(WorkflowStateSystem(priority=-25), priority=-25)
     world.register_system(SystemPromptRenderSystem(priority=-20), priority=-20)
     world.register_system(UserPromptNormalizationSystem(priority=-10), priority=-10)
     subagent_system = SubagentSystem(priority=-1)

examples/e2e/plan_and_task/state_machine.py

@@ -5,29 +5,17 @@
 import datetime
 
 from ecs_agent.logging import get_logger
+from ecs_agent.workflows.compiler import compile_workflow
 
 from examples.e2e.plan_and_task.scratchbook_adapter import (
     PlanTaskScratchbookAdapter as ArtifactAdapter,
 )
 from examples.e2e.plan_and_task.state_models import RuntimeState
+from examples.e2e.plan_and_task.workflow_spec import PLAN_TASK_WORKFLOW_SPEC
 
 logger = get_logger(__name__)
 
-VALID_TRANSITIONS: dict[str, set[str]] = {
-    "IDLE": {"DRAFT_INTERVIEW"},
-    "DRAFT_INTERVIEW": {"DRAFT_ADVISOR_REVIEW", "DRAFT_QA_REVIEW"},
-    "DRAFT_ADVISOR_REVIEW": {"DRAFT_QA_REVIEW", "DRAFT_INTERVIEW"},
-    "DRAFT_QA_REVIEW": {"WRITE_PLAN", "DRAFT_INTERVIEW"},
-    "WRITE_PLAN": {"PLAN_QA_REVIEW"},
-    "PLAN_QA_REVIEW": {"PLAN_FINALIZED", "WRITE_PLAN"},
-    "PLAN_FINALIZED": {"TASK_READY"},
-    "TASK_READY": {"TASK_RUNNING"},
-    "TASK_RUNNING": {"TASK_COMPLETED", "TASK_BLOCKED", "TASK_REPLAN", "TASK_ABORTED"},
-    "TASK_BLOCKED": {"TASK_RUNNING", "TASK_REPLAN", "TASK_ABORTED"},
-    "TASK_REPLAN": {"DRAFT_INTERVIEW", "DRAFT_ADVISOR_REVIEW", "TASK_RUNNING"},
-    "TASK_COMPLETED": set(),
-    "TASK_ABORTED": set(),
-}
+_COMPILED_WORKFLOW = compile_workflow(PLAN_TASK_WORKFLOW_SPEC)
 
 _TERMINAL_PHASES: frozenset[str] = frozenset({"TASK_COMPLETED", "TASK_ABORTED"})
 
@@ -48,7 +36,8 @@ def transition(self, state: RuntimeState, to_phase: str) -> RuntimeState:
         Raises:
             ValueError: If the transition is invalid.
         """
-        allowed = VALID_TRANSITIONS.get(state.phase, set())
+        transitions = _COMPILED_WORKFLOW.transitions_by_state.get(state.phase, ())
+        allowed = {transition.target_state_id for transition in transitions}
         if to_phase not in allowed:
             raise ValueError(f"Invalid transition: {state.phase} → {to_phase}")
         state.phase = to_phase
@@ -113,7 +102,7 @@ def _force_phase(self, state: RuntimeState, phase: str) -> None:
 
         This is an administrative-only bypass for exceptional recovery scenarios, such as
         marking in-flight tasks as blocked after a restart. Normal phase transitions must
-        use the transition() method, which validates against VALID_TRANSITIONS.
+        use the transition() method, which validates against the compiled workflow transition graph.
 
         Args:
             state: Current runtime state to modify.

examples/e2e/plan_and_task/task_exec.py

@@ -27,8 +27,8 @@
     TaskRecord,
 )
 from examples.e2e.plan_and_task.state_machine import (
-    VALID_TRANSITIONS,
     WorkflowStateMachine,
+    _COMPILED_WORKFLOW,
 )
 
 logger = get_logger(__name__)
@@ -383,7 +383,7 @@ def _transition_to_running(self, state: RuntimeState) -> RuntimeState:
         return state
 
     def _allowed_transitions(self, state: RuntimeState) -> set[str]:
-        return VALID_TRANSITIONS.get(state.phase, set())
+        return {t.target_state_id for t in _COMPILED_WORKFLOW.transitions_by_state.get(state.phase, ())}
 
     def _utcnow_isoformat(self) -> str:
         return datetime.datetime.now(datetime.UTC).isoformat()