| 1 | """Tests for workflow-state coordination helpers.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from pathlib import Path |
| 6 | |
| 7 | import pytest |
| 8 | |
| 9 | from loader.agent.loop import Agent, AgentConfig |
| 10 | from loader.runtime.conversation import ConversationRuntime |
| 11 | from loader.runtime.dod import DefinitionOfDoneStore |
| 12 | from loader.runtime.events import TurnSummary |
| 13 | from loader.runtime.workflow import WorkflowDecisionKind, WorkflowMode |
| 14 | from loader.runtime.workflow_policy import ModeDecision |
| 15 | from tests.helpers.runtime_harness import ScriptedBackend |
| 16 | |
| 17 | |
| 18 | def non_streaming_config() -> AgentConfig: |
| 19 | """Shared config for direct workflow-state tests.""" |
| 20 | |
| 21 | return AgentConfig(auto_context=False, stream=False, max_iterations=8) |
| 22 | |
| 23 | |
| 24 | @pytest.mark.asyncio |
| 25 | async def test_workflow_state_applies_mode_to_session_dod_and_timeline( |
| 26 | temp_dir: Path, |
| 27 | ) -> None: |
| 28 | backend = ScriptedBackend() |
| 29 | agent = Agent( |
| 30 | backend=backend, |
| 31 | config=non_streaming_config(), |
| 32 | project_root=temp_dir, |
| 33 | ) |
| 34 | agent.prompt_format = "native" |
| 35 | agent.prompt_sections = ["Workflow Context", "Runtime Config"] |
| 36 | runtime = ConversationRuntime(agent) |
| 37 | dod = runtime.dod_store.create_or_resume("Tighten the workflow state controller.") |
| 38 | brief_path = temp_dir / "brief.md" |
| 39 | brief_path.write_text("# Task Brief\n\n## Likely Touchpoints\n- src/loader/runtime\n") |
| 40 | dod.clarify_brief = str(brief_path) |
| 41 | summary = TurnSummary(final_response="") |
| 42 | events = [] |
| 43 | |
| 44 | async def capture(event) -> None: |
| 45 | events.append(event) |
| 46 | |
| 47 | decision = ModeDecision.transition( |
| 48 | WorkflowMode.PLAN, |
| 49 | reason_code="task_is_complex", |
| 50 | reason_summary="task complexity favors planning first", |
| 51 | decision_kind=WorkflowDecisionKind.HANDOFF, |
| 52 | scheduled_next_mode=WorkflowMode.EXECUTE, |
| 53 | ) |
| 54 | |
| 55 | await runtime.workflow_state.set_workflow_mode( |
| 56 | decision, |
| 57 | dod=dod, |
| 58 | emit=capture, |
| 59 | summary=summary, |
| 60 | ) |
| 61 | |
| 62 | reloaded = DefinitionOfDoneStore(temp_dir).load(Path(dod.storage_path)) |
| 63 | |
| 64 | assert agent.workflow_mode == "plan" |
| 65 | assert agent.session.workflow_mode == "plan" |
| 66 | assert summary.workflow_mode == "plan" |
| 67 | assert summary.workflow_reason_code == "task_is_complex" |
| 68 | assert summary.workflow_decision_kind == "handoff" |
| 69 | assert dod.current_mode == "plan" |
| 70 | assert dod.mode_history[-1] == "plan" |
| 71 | assert reloaded.current_mode == "plan" |
| 72 | assert summary.definition_of_done is dod |
| 73 | assert summary.workflow_timeline[-1].mode == "plan" |
| 74 | assert summary.workflow_timeline[-1].kind == "handoff" |
| 75 | assert summary.workflow_timeline[-1].artifact_paths == [str(brief_path)] |
| 76 | assert summary.workflow_timeline[-1].prompt_format == "native" |
| 77 | assert any( |
| 78 | event.type == "workflow_mode" and event.workflow_mode == "plan" |
| 79 | for event in events |
| 80 | ) |
| 81 | |
| 82 | |
| 83 | def test_workflow_state_appends_execute_bridge_once( |
| 84 | temp_dir: Path, |
| 85 | ) -> None: |
| 86 | backend = ScriptedBackend() |
| 87 | agent = Agent( |
| 88 | backend=backend, |
| 89 | config=non_streaming_config(), |
| 90 | project_root=temp_dir, |
| 91 | ) |
| 92 | runtime = ConversationRuntime(agent) |
| 93 | dod = runtime.dod_store.create_or_resume("Use the persisted workflow artifacts.") |
| 94 | |
| 95 | brief_path = temp_dir / "brief.md" |
| 96 | brief_path.write_text("# Task Brief\n\n## In Scope\n- Runtime workflow\n") |
| 97 | plan_path = temp_dir / "plan.md" |
| 98 | plan_path.write_text("# Implementation Plan\n\n- Extract the controller\n") |
| 99 | verify_path = temp_dir / "verify.md" |
| 100 | verify_path.write_text("# Verification Plan\n\n- Run uv run pytest -q\n") |
| 101 | dod.clarify_brief = str(brief_path) |
| 102 | dod.implementation_plan = str(plan_path) |
| 103 | dod.verification_plan = str(verify_path) |
| 104 | |
| 105 | runtime.workflow_state.maybe_append_execute_bridge(dod) |
| 106 | runtime.workflow_state.maybe_append_execute_bridge(dod) |
| 107 | |
| 108 | bridge_messages = [ |
| 109 | message.content |
| 110 | for message in agent.session.messages |
| 111 | if "[WORKFLOW BRIDGE]" in message.content |
| 112 | ] |
| 113 | |
| 114 | assert len(bridge_messages) == 1 |
| 115 | assert "Use the clarify brief below as the requirements source of truth." in bridge_messages[0] |
| 116 | assert "# Implementation Plan" in bridge_messages[0] |
| 117 | assert "# Verification Plan" in bridge_messages[0] |