`2f03a4d`

Add validated turn state machine and workflow metadata

Authored by

espadonne 1 month ago

SHA: 2f03a4d0c7747003aeaad1cde144e844376117cf
Parents: 203d232
Tree: 8e95248

12 changed files

Status	File	+	-
M	`src/loader/runtime/conversation.py`	71	15
M	`src/loader/runtime/events.py`	7	0
M	`src/loader/runtime/finalization.py`	22	6
M	`src/loader/runtime/inspection.py`	35	0
M	`src/loader/runtime/phases.py`	141	7
M	`src/loader/runtime/session.py`	145	1
M	`src/loader/runtime/workflow.py`	96	9
M	`tests/test_inspection.py`	30	0
M	`tests/test_runtime_phases.py`	21	0
M	`tests/test_session_state.py`	22	0
A	`tests/test_turn_state_machine.py`	66	0
M	`tests/test_workflow_runtime.py`	14	1

src/loader/runtime/conversation.pymodified

  from .executor import ToolExecutor
  from .finalization import TurnFinalizer, merge_usage
  from .hooks import build_default_tool_hooks
 -from .phases import TurnPhase, TurnPhaseTracker
 +from .phases import TurnPhase, TurnPhaseTracker, TurnTransitionKind
  from .repair import ResponseRepairer
  from .tool_batches import ToolBatchRunner
  from .tracing import RuntimeTracer
  from .workflow import (
      VERIFICATION_SEPARATOR,
      ClarifyBrief,
 +    ModeDecision,
      ModeRouter,
      PlanningArtifacts,
      WorkflowArtifactStore,
 +    WorkflowDecisionKind,
      WorkflowMode,
      build_execute_bridge,
      sync_todos_to_definition_of_done,
              TurnPhase.PREPARE,
              emit,
              detail="Preparing runtime state",
 +            reason_code="prepare_runtime",
+         )
          await self._prepare_runtime_capabilities()
                  TurnPhase.ASSISTANT,
                  emit,
                  detail="Requesting assistant response",
 +                reason_code="request_assistant_response",
+             )
              await emit(AgentEvent(type="thinking"))
              assistant_turn = await self.turn_requester.request_turn(
                      TurnPhase.REPAIR,
                      emit,
                      detail="Repairing empty assistant response",
 +                    reason_code="repair_empty_response",
 +                    kind=TurnTransitionKind.RETRY,
+                 )
                  empty_retry_count += 1
                  empty_decision = self.repairer.handle_empty_response(
                      TurnPhase.REPAIR,
                      emit,
                      detail="Repairing raw-text tool fallback",
 +                    reason_code="repair_raw_text_tool_fallback",
 +                    kind=TurnTransitionKind.REROUTE,
+                 )
                  await emit(AgentEvent(type="clear_stream"))
                      TurnPhase.TOOLS,
                      emit,
                      detail="Executing tool batch",
 +                    reason_code="execute_tool_batch",
+                 )
                  assistant_message = Message(
                      role=Role.ASSISTANT,
                  actions_taken.extend(batch_result.actions_taken)
                  consecutive_errors = batch_result.consecutive_errors
                  if batch_result.halted:
 -                    return await self._finalize_turn(summary, emit)
 +                    return await self._finalize_turn(
 +                        summary,
 +                        emit,
 +                        reason_code="tool_batch_halted",
 +                        reason_summary="Finalizing after halted tool batch",
 +                    )
                  continue
                      TurnPhase.REPAIR,
                      emit,
                      detail="Repairing fake tool narration",
 +                    reason_code="repair_fake_tool_narration",
 +                    kind=TurnTransitionKind.REROUTE,
+                 )
                  self.agent.session.append(Message(role=Role.ASSISTANT, content=response_content))
                  self.agent.session.append(Message(role=Role.USER, content=repair_message))
                      TurnPhase.REPAIR,
                      emit,
                      detail="Repairing execution deflection",
 +                    reason_code="repair_execution_deflection",
 +                    kind=TurnTransitionKind.REROUTE,
+                 )
                  self.agent.session.append(Message(role=Role.ASSISTANT, content=response_content))
                  self.agent.session.append(
                      TurnPhase.CRITIQUE,
                      emit,
                      detail="Evaluating self-critique",
 +                    reason_code="evaluate_self_critique",
+                 )
                  critique_decision = await self.completion_policy.maybe_self_critique(
                      content=content,
                  TurnPhase.COMPLETION,
                  emit,
                  detail="Checking completion policy",
 +                reason_code="completion_gate",
+             )
              text_loop_decision = await self.completion_policy.maybe_stop_for_text_loop(
                  content=content,
                  summary=summary,
+             )
              if text_loop_decision.should_stop:
 -                return await self._finalize_turn(summary, emit)
 +                return await self._finalize_turn(
 +                    summary,
 +                    emit,
 +                    reason_code="text_loop_bailout",
 +                    reason_summary="Finalizing after text-loop bailout",
 +                )
              self.agent.safeguards.record_response(content)
              effective_task = original_task or task
              await emit(AgentEvent(type="response", content=final_response))
              break
 -        return await self._finalize_turn(summary, emit)
 +        return await self._finalize_turn(
 +            summary,
 +            emit,
 +            reason_code="turn_complete",
 +            reason_summary="Finalizing completed turn",
 +        )
      async def _finalize_turn(
          self,
          summary: TurnSummary,
          emit: EventSink,
 +        *,
 +        reason_code: str,
 +        reason_summary: str,
      ) -> TurnSummary:
          await self.phase_tracker.enter(
              TurnPhase.FINALIZE,
              emit,
 -            detail="Finalizing turn summary",
 +            detail=reason_summary,
 +            reason_code=reason_code,
 +            kind=TurnTransitionKind.TERMINAL,
+         )
          final_summary = self.finalizer.finalize_summary(summary)
          self.phase_tracker.clear()
              and self._artifact_exists(dod.verification_plan),
+         )
          await self._set_workflow_mode(
 -            decision.mode,
 +            decision,
              dod=dod,
              emit=emit,
              summary=summary,
 -            reason=decision.reason,
+         )
          if decision.mode == WorkflowMode.CLARIFY:
                  allow_clarify=False,
+             )
              await self._set_workflow_mode(
 -                decision.mode,
 +                decision.with_context(
 +                    reason_code=f"post_clarify_{decision.reason_code}",
 +                    reason_summary=f"clarify handoff: {decision.reason_summary}",
 +                    decision_kind=WorkflowDecisionKind.HANDOFF,
 +                ),
                  dod=dod,
                  emit=emit,
                  summary=summary,
 -                reason=f"clarify handoff: {decision.reason}",
+             )
          if decision.mode == WorkflowMode.PLAN:
                  on_user_question=on_user_question,
+             )
              await self._set_workflow_mode(
 -                WorkflowMode.EXECUTE,
 +                ModeDecision.transition(
 +                    WorkflowMode.EXECUTE,
 +                    reason_code="plan_artifacts_created",
 +                    reason_summary="plan artifacts created; switching to execute",
 +                    decision_kind=WorkflowDecisionKind.HANDOFF,
 +                ),
                  dod=dod,
                  emit=emit,
                  summary=summary,
 -                reason="plan artifacts created; switching to execute",
+             )
          bridge = build_execute_bridge(
      async def _set_workflow_mode(
          self,
 -        mode: WorkflowMode,
 +        decision: ModeDecision,
          *,
          dod: DefinitionOfDone,
          emit: EventSink,
          summary: TurnSummary,
 -        reason: str,
      ) -> None:
 +        mode = decision.mode
          self.agent.set_workflow_mode(mode.value)
 -        self.agent.session.update_runtime_state(workflow_mode=mode.value)
 +        self.agent.session.update_runtime_state(
 +            workflow_mode=mode.value,
 +            workflow_reason_code=decision.reason_code,
 +            workflow_reason_summary=decision.reason_summary,
 +            workflow_decision_kind=decision.decision_kind.value,
 +            workflow_ambiguity_score=decision.ambiguity_score,
 +            workflow_complexity_score=decision.complexity_score,
 +            workflow_scheduled_next_mode=(
 +                decision.scheduled_next_mode.value
 +                if decision.scheduled_next_mode is not None
 +                else None
 +            ),
 +        )
          dod.current_mode = mode.value
          if not dod.mode_history or dod.mode_history[-1] != mode.value:
              dod.mode_history.append(mode.value)
          summary.workflow_mode = mode.value
 +        summary.workflow_reason_code = decision.reason_code
 +        summary.workflow_reason_summary = decision.reason_summary
 +        summary.workflow_decision_kind = decision.decision_kind.value
          summary.definition_of_done = dod
          self.dod_store.save(dod)
          await emit(
              AgentEvent(
                  type="workflow_mode",
 -                content=f"Workflow: {mode.value} ({reason})",
 +                content=f"Workflow: {mode.value} ({decision.reason_summary})",
                  workflow_mode=mode.value,
                  definition_of_done=dod,
+             )

src/loader/runtime/events.pymodified

      last_verification_result: str | None = None
      workflow_mode: str | None = None
      turn_phase: str | None = None
 +    transition_kind: str | None = None
 +    transition_summary: str | None = None
 +    transition_reason_code: str | None = None
      artifact_kind: str | None = None
      artifact_path: str | None = None
      trace: list[RuntimeTraceEvent] = field(default_factory=list)
      definition_of_done: DefinitionOfDone | None = None
      workflow_mode: str | None = None
 +    workflow_reason_code: str | None = None
 +    workflow_reason_summary: str | None = None
 +    workflow_decision_kind: str | None = None
 +    last_turn_transition_summary: str | None = None
      session_id: str | None = None

src/loader/runtime/finalization.pymodified

  from .memory import MemoryStore
  from .session import normalize_usage
  from .tracing import RuntimeTracer
 -from .workflow import WorkflowMode, extract_verification_commands_from_markdown
 +from .workflow import (
 +    ModeDecision,
 +    WorkflowDecisionKind,
 +    WorkflowMode,
 +    extract_verification_commands_from_markdown,
 +)
  EventSink = Callable[[AgentEvent], Awaitable[None]]
  WorkflowSetter = Callable[
 -    [WorkflowMode, DefinitionOfDone, EventSink, TurnSummary, str],
 +    [ModeDecision, DefinitionOfDone, EventSink, TurnSummary],
      Awaitable[None],
+ ]
+             )
          await self.set_workflow_mode(
 -            WorkflowMode.VERIFY,
 +            ModeDecision.transition(
 +                WorkflowMode.VERIFY,
 +                reason_code="definition_of_done_requires_verification",
 +                reason_summary="definition-of-done gate requires verification",
 +                decision_kind=WorkflowDecisionKind.HANDOFF,
 +            ),
              dod=dod,
              emit=emit,
              summary=summary,
 -            reason="definition-of-done gate requires verification",
+         )
          verification_passed = await self.verify_definition_of_done(
              dod=dod,
          self.dod_store.save(dod)
          await self.emit_dod_status(emit, dod)
          await self.set_workflow_mode(
 -            WorkflowMode.EXECUTE,
 +            ModeDecision.transition(
 +                WorkflowMode.EXECUTE,
 +                reason_code="verification_failed_reentry",
 +                reason_summary="verification failed; returning to execute for fixes",
 +                decision_kind=WorkflowDecisionKind.REENTRY,
 +            ),
              dod=dod,
              emit=emit,
              summary=summary,
 -            reason="verification failed; returning to execute for fixes",
+         )
          failure_prompt = (
              "[DEFINITION OF DONE CHECK FAILED]\n"
              iterations=summary.iterations,
+         )
          summary.session_id = self.agent.session.session_id
 +        summary.last_turn_transition_summary = (
 +            self.agent.session.last_turn_transition_summary
 +        )
          if summary.definition_of_done and summary.definition_of_done.status == "done":
              MemoryStore(self.agent.project_root).capture_definition_of_done(
                  build_verification_summary(summary.definition_of_done.evidence)

src/loader/runtime/inspection.pymodified

      capability_profile: CapabilityProfile
      active_session_id: str | None
      workflow_mode: str
 +    workflow_reason_code: str | None
 +    workflow_reason_summary: str | None
 +    workflow_decision_kind: str | None
 +    workflow_ambiguity_score: float | None
 +    workflow_complexity_score: float | None
 +    workflow_scheduled_next_mode: str | None
      active_turn_phase: str | None
 +    last_turn_transition_summary: str | None
 +    last_turn_transition_kind: str | None
 +    last_turn_transition_reason_code: str | None
      permission_mode: str
      permission_prompting_enabled: bool
      permission_rule_counts: dict[str, int]
      updated_at: str
      message_count: int
      workflow_mode: str
 +    workflow_reason_code: str | None
 +    workflow_reason_summary: str | None
 +    workflow_decision_kind: str | None
      permission_mode: str
      permission_prompting_enabled: bool
      permission_rule_counts: dict[str, int]
      permission_rules_source: str | None
      prompt_format: str | None
      active_turn_phase: str | None
 +    last_turn_transition_summary: str | None
      current_task: str | None
      active_dod_path: str | None
      dod_status: str | None
              capability_profile=capability_profile,
              active_session_id=None,
              workflow_mode="execute",
 +            workflow_reason_code=None,
 +            workflow_reason_summary=None,
 +            workflow_decision_kind=None,
 +            workflow_ambiguity_score=None,
 +            workflow_complexity_score=None,
 +            workflow_scheduled_next_mode=None,
              active_turn_phase=None,
 +            last_turn_transition_summary=None,
 +            last_turn_transition_kind=None,
 +            last_turn_transition_reason_code=None,
              permission_mode=default_permission_mode,
              permission_prompting_enabled=(
                  _coerce_permission_mode(permission_mode) == PermissionMode.PROMPT
          capability_profile=capability_profile,
          active_session_id=snapshot.session_id,
          workflow_mode=snapshot.workflow_mode,
 +        workflow_reason_code=snapshot.workflow_reason_code,
 +        workflow_reason_summary=snapshot.workflow_reason_summary,
 +        workflow_decision_kind=snapshot.workflow_decision_kind,
 +        workflow_ambiguity_score=snapshot.workflow_ambiguity_score,
 +        workflow_complexity_score=snapshot.workflow_complexity_score,
 +        workflow_scheduled_next_mode=snapshot.workflow_scheduled_next_mode,
          active_turn_phase=snapshot.active_turn_phase,
 +        last_turn_transition_summary=snapshot.last_turn_transition_summary,
 +        last_turn_transition_kind=snapshot.last_turn_transition_kind,
 +        last_turn_transition_reason_code=snapshot.last_turn_transition_reason_code,
          permission_mode=snapshot.permission_mode or default_permission_mode,
          permission_prompting_enabled=permission_prompting_enabled,
          permission_rule_counts=permission_rule_counts,
                  updated_at=snapshot.updated_at,
                  message_count=len(snapshot.messages),
                  workflow_mode=snapshot.workflow_mode,
 +                workflow_reason_code=snapshot.workflow_reason_code,
 +                workflow_reason_summary=snapshot.workflow_reason_summary,
 +                workflow_decision_kind=snapshot.workflow_decision_kind,
                  permission_mode=snapshot.permission_mode,
                  permission_prompting_enabled=(
                      snapshot.permission_prompting_enabled
                  permission_rules_source=snapshot.permission_rules_source,
                  prompt_format=snapshot.prompt_format,
                  active_turn_phase=snapshot.active_turn_phase,
 +                last_turn_transition_summary=snapshot.last_turn_transition_summary,
                  current_task=snapshot.current_task,
                  active_dod_path=snapshot.active_dod_path,
                  dod_status=dod.status if dod else None,

src/loader/runtime/phases.pymodified

  from __future__ import annotations
  from collections.abc import Awaitable, Callable
 +from dataclasses import dataclass
  from enum import StrEnum
  from .events import AgentEvent
      FINALIZE = "finalize"
 +class TurnTransitionKind(StrEnum):
 +    """Classification for why one turn-state transition occurred."""
++
 +    NORMAL = "normal"
 +    RETRY = "retry"
 +    REROUTE = "reroute"
 +    RECOVERY = "recovery"
 +    TERMINAL = "terminal"
++
++
 +@dataclass(slots=True)
 +class TurnTransition:
 +    """One validated turn-state transition."""
++
 +    from_phase: str | None
 +    to_phase: str
 +    reason_code: str
 +    reason_summary: str
 +    kind: TurnTransitionKind
++
 +    @property
 +    def summary(self) -> str:
 +        source = self.from_phase or "start"
 +        return (
 +            f"{source} -> {self.to_phase} "
 +            f"[{self.kind.value}] {self.reason_summary}"
 +        )
++
++
 +class TurnStateMachine:
 +    """Validate allowed turn-state transitions."""
++
 +    _ALLOWED_TRANSITIONS: dict[str | None, set[str]] = {
 +        None: {TurnPhase.PREPARE.value},
 +        TurnPhase.PREPARE.value: {
 +            TurnPhase.ASSISTANT.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.ASSISTANT.value: {
 +            TurnPhase.REPAIR.value,
 +            TurnPhase.TOOLS.value,
 +            TurnPhase.CRITIQUE.value,
 +            TurnPhase.COMPLETION.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.REPAIR.value: {
 +            TurnPhase.ASSISTANT.value,
 +            TurnPhase.TOOLS.value,
 +            TurnPhase.COMPLETION.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.TOOLS.value: {
 +            TurnPhase.ASSISTANT.value,
 +            TurnPhase.CRITIQUE.value,
 +            TurnPhase.COMPLETION.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.CRITIQUE.value: {
 +            TurnPhase.ASSISTANT.value,
 +            TurnPhase.COMPLETION.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.COMPLETION.value: {
 +            TurnPhase.ASSISTANT.value,
 +            TurnPhase.FINALIZE.value,
 +        },
 +        TurnPhase.FINALIZE.value: set(),
 +    }
++
 +    def __init__(self) -> None:
 +        self.current_phase: str | None = None
 +        self.last_transition: TurnTransition | None = None
++
 +    def transition(
 +        self,
 +        phase: TurnPhase,
 +        *,
 +        reason_code: str,
 +        reason_summary: str,
 +        kind: TurnTransitionKind = TurnTransitionKind.NORMAL,
 +    ) -> TurnTransition | None:
 +        """Validate and record a transition to the target phase."""
++
 +        if phase.value == self.current_phase:
 +            return None
++
 +        allowed = self._ALLOWED_TRANSITIONS.get(self.current_phase, set())
 +        if phase.value not in allowed:
 +            raise ValueError(
 +                "Invalid turn-state transition: "
 +                f"{self.current_phase or 'start'} -> {phase.value}"
 +            )
++
 +        transition = TurnTransition(
 +            from_phase=self.current_phase,
 +            to_phase=phase.value,
 +            reason_code=reason_code,
 +            reason_summary=reason_summary,
 +            kind=kind,
 +        )
 +        self.current_phase = phase.value
 +        self.last_transition = transition
 +        return transition
++
 +    def clear(self) -> None:
 +        """Reset the active phase after a turn completes."""
++
 +        self.current_phase = None
++
++
  class TurnPhaseTracker:
      """Persist and emit turn-phase transitions."""
      def __init__(self, agent, tracer: RuntimeTracer) -> None:
          self.agent = agent
          self.tracer = tracer
 -        self.current_phase: str | None = None
 +        self.state_machine = TurnStateMachine()
      async def enter(
          self,
          emit: EventSink,
          *,
          detail: str | None = None,
 +        reason_code: str | None = None,
 +        kind: TurnTransitionKind = TurnTransitionKind.NORMAL,
      ) -> None:
          """Move the runtime into a named phase and emit the transition."""
 -        if phase.value == self.current_phase:
 +        summary = detail or f"Phase: {phase.value}"
 +        transition = self.state_machine.transition(
 +            phase,
 +            reason_code=reason_code or phase.value,
 +            reason_summary=summary,
 +            kind=kind,
 +        )
 +        if transition is None:
              return
 -        self.current_phase = phase.value
 -        self.agent.session.update_runtime_state(active_turn_phase=phase.value)
 -        self.tracer.record("turn.phase_changed", phase=phase.value, detail=detail)
 +        self.agent.session.update_runtime_state(
 +            active_turn_phase=phase.value,
 +            last_turn_transition_summary=transition.summary,
 +            last_turn_transition_kind=transition.kind.value,
 +            last_turn_transition_reason_code=transition.reason_code,
 +        )
 +        self.tracer.record(
 +            "turn.phase_changed",
 +            phase=phase.value,
 +            detail=summary,
 +            from_phase=transition.from_phase,
 +            transition_kind=transition.kind.value,
 +            reason_code=transition.reason_code,
 +        )
          await emit(
              AgentEvent(
                  type="turn_phase",
 -                content=detail or f"Phase: {phase.value}",
 +                content=transition.summary,
                  turn_phase=phase.value,
 +                transition_kind=transition.kind.value,
 +                transition_summary=transition.summary,
 +                transition_reason_code=transition.reason_code,
+             )
+         )
      def clear(self) -> None:
          """Clear the persisted active phase when the turn finishes."""
 -        self.current_phase = None
 +        self.state_machine.clear()
          self.agent.session.update_runtime_state(active_turn_phase=None)

src/loader/runtime/session.pymodified

      estimate_message_tokens,
+ )
 -SESSION_VERSION = 3
 +SESSION_VERSION = 4
  DEFAULT_ROTATE_AFTER_BYTES = 256 * 1024
  MAX_ROTATED_FILES = 3
  _UNSET = object()
      return [str(item) for item in value if str(item).strip()]
 +def normalize_optional_text(value: Any) -> str | None:
 +    """Coerce persisted optional text fields."""
++
 +    if value is None:
 +        return None
 +    text = str(value).strip()
 +    return text or None
++
++
 +def normalize_optional_float(value: Any) -> float | None:
 +    """Coerce persisted numeric workflow scores."""
++
 +    if value is None:
 +        return None
 +    return float(value)
++
++
  @dataclass(slots=True)
  class SessionCompaction:
      """Metadata describing the latest transcript compaction."""
      prompt_format: str | None = None
      prompt_sections: list[str] = field(default_factory=list)
      active_turn_phase: str | None = None
 +    workflow_reason_code: str | None = None
 +    workflow_reason_summary: str | None = None
 +    workflow_decision_kind: str | None = None
 +    workflow_ambiguity_score: float | None = None
 +    workflow_complexity_score: float | None = None
 +    workflow_scheduled_next_mode: str | None = None
 +    last_turn_transition_summary: str | None = None
 +    last_turn_transition_kind: str | None = None
 +    last_turn_transition_reason_code: str | None = None
      compaction: SessionCompaction | None = None
      version: int = SESSION_VERSION
              "prompt_format": self.prompt_format,
              "prompt_sections": list(self.prompt_sections),
              "active_turn_phase": self.active_turn_phase,
 +            "workflow_reason_code": self.workflow_reason_code,
 +            "workflow_reason_summary": self.workflow_reason_summary,
 +            "workflow_decision_kind": self.workflow_decision_kind,
 +            "workflow_ambiguity_score": self.workflow_ambiguity_score,
 +            "workflow_complexity_score": self.workflow_complexity_score,
 +            "workflow_scheduled_next_mode": self.workflow_scheduled_next_mode,
 +            "last_turn_transition_summary": self.last_turn_transition_summary,
 +            "last_turn_transition_kind": self.last_turn_transition_kind,
 +            "last_turn_transition_reason_code": self.last_turn_transition_reason_code,
              "compaction": self.compaction.to_dict() if self.compaction else None,
+         }
              prompt_format=data.get("prompt_format"),
              prompt_sections=normalize_prompt_sections(data.get("prompt_sections")),
              active_turn_phase=data.get("active_turn_phase"),
 +            workflow_reason_code=normalize_optional_text(
 +                data.get("workflow_reason_code")
 +            ),
 +            workflow_reason_summary=normalize_optional_text(
 +                data.get("workflow_reason_summary")
 +            ),
 +            workflow_decision_kind=normalize_optional_text(
 +                data.get("workflow_decision_kind")
 +            ),
 +            workflow_ambiguity_score=normalize_optional_float(
 +                data.get("workflow_ambiguity_score")
 +            ),
 +            workflow_complexity_score=normalize_optional_float(
 +                data.get("workflow_complexity_score")
 +            ),
 +            workflow_scheduled_next_mode=normalize_optional_text(
 +                data.get("workflow_scheduled_next_mode")
 +            ),
 +            last_turn_transition_summary=normalize_optional_text(
 +                data.get("last_turn_transition_summary")
 +            ),
 +            last_turn_transition_kind=normalize_optional_text(
 +                data.get("last_turn_transition_kind")
 +            ),
 +            last_turn_transition_reason_code=normalize_optional_text(
 +                data.get("last_turn_transition_reason_code")
 +            ),
              compaction=(
                  SessionCompaction.from_dict(data["compaction"])
                  if data.get("compaction")
      prompt_format: str | None = None
      prompt_sections: list[str] = field(default_factory=list)
      active_turn_phase: str | None = None
 +    workflow_reason_code: str | None = None
 +    workflow_reason_summary: str | None = None
 +    workflow_decision_kind: str | None = None
 +    workflow_ambiguity_score: float | None = None
 +    workflow_complexity_score: float | None = None
 +    workflow_scheduled_next_mode: str | None = None
 +    last_turn_transition_summary: str | None = None
 +    last_turn_transition_kind: str | None = None
 +    last_turn_transition_reason_code: str | None = None
      compaction: SessionCompaction | None = None
      rotate_after_bytes: int = DEFAULT_ROTATE_AFTER_BYTES
      max_rotated_files: int = MAX_ROTATED_FILES
          self.active_dod_path = None
          self.current_task = None
          self.workflow_mode = "execute"
 +        self.workflow_reason_code = None
 +        self.workflow_reason_summary = None
 +        self.workflow_decision_kind = None
 +        self.workflow_ambiguity_score = None
 +        self.workflow_complexity_score = None
 +        self.workflow_scheduled_next_mode = None
 +        self.active_turn_phase = None
 +        self.last_turn_transition_summary = None
 +        self.last_turn_transition_kind = None
 +        self.last_turn_transition_reason_code = None
          self.compaction = None
          self.usage_totals = {}
          self.touch()
          prompt_format: str | None = None,
          prompt_sections: list[str] | None = None,
          active_turn_phase: str | None | object = _UNSET,
 +        workflow_reason_code: str | None | object = _UNSET,
 +        workflow_reason_summary: str | None | object = _UNSET,
 +        workflow_decision_kind: str | None | object = _UNSET,
 +        workflow_ambiguity_score: float | None | object = _UNSET,
 +        workflow_complexity_score: float | None | object = _UNSET,
 +        workflow_scheduled_next_mode: str | None | object = _UNSET,
 +        last_turn_transition_summary: str | None | object = _UNSET,
 +        last_turn_transition_kind: str | None | object = _UNSET,
 +        last_turn_transition_reason_code: str | None | object = _UNSET,
      ) -> None:
          """Update persisted runtime state that lives beside the messages."""
              self.prompt_sections = normalize_prompt_sections(prompt_sections)
          if active_turn_phase is not _UNSET:
              self.active_turn_phase = active_turn_phase
 +        if workflow_reason_code is not _UNSET:
 +            self.workflow_reason_code = normalize_optional_text(workflow_reason_code)
 +        if workflow_reason_summary is not _UNSET:
 +            self.workflow_reason_summary = normalize_optional_text(
 +                workflow_reason_summary
 +            )
 +        if workflow_decision_kind is not _UNSET:
 +            self.workflow_decision_kind = normalize_optional_text(
 +                workflow_decision_kind
 +            )
 +        if workflow_ambiguity_score is not _UNSET:
 +            self.workflow_ambiguity_score = normalize_optional_float(
 +                workflow_ambiguity_score
 +            )
 +        if workflow_complexity_score is not _UNSET:
 +            self.workflow_complexity_score = normalize_optional_float(
 +                workflow_complexity_score
 +            )
 +        if workflow_scheduled_next_mode is not _UNSET:
 +            self.workflow_scheduled_next_mode = normalize_optional_text(
 +                workflow_scheduled_next_mode
 +            )
 +        if last_turn_transition_summary is not _UNSET:
 +            self.last_turn_transition_summary = normalize_optional_text(
 +                last_turn_transition_summary
 +            )
 +        if last_turn_transition_kind is not _UNSET:
 +            self.last_turn_transition_kind = normalize_optional_text(
 +                last_turn_transition_kind
 +            )
 +        if last_turn_transition_reason_code is not _UNSET:
 +            self.last_turn_transition_reason_code = normalize_optional_text(
 +                last_turn_transition_reason_code
 +            )
          self.touch()
          self.persist()
              prompt_format=self.prompt_format,
              prompt_sections=list(self.prompt_sections),
              active_turn_phase=self.active_turn_phase,
 +            workflow_reason_code=self.workflow_reason_code,
 +            workflow_reason_summary=self.workflow_reason_summary,
 +            workflow_decision_kind=self.workflow_decision_kind,
 +            workflow_ambiguity_score=self.workflow_ambiguity_score,
 +            workflow_complexity_score=self.workflow_complexity_score,
 +            workflow_scheduled_next_mode=self.workflow_scheduled_next_mode,
 +            last_turn_transition_summary=self.last_turn_transition_summary,
 +            last_turn_transition_kind=self.last_turn_transition_kind,
 +            last_turn_transition_reason_code=self.last_turn_transition_reason_code,
              compaction=self.compaction,
+         )
          return self.store.save(snapshot)
          instance.prompt_format = snapshot.prompt_format
          instance.prompt_sections = list(snapshot.prompt_sections)
          instance.active_turn_phase = snapshot.active_turn_phase
 +        instance.workflow_reason_code = snapshot.workflow_reason_code
 +        instance.workflow_reason_summary = snapshot.workflow_reason_summary
 +        instance.workflow_decision_kind = snapshot.workflow_decision_kind
 +        instance.workflow_ambiguity_score = snapshot.workflow_ambiguity_score
 +        instance.workflow_complexity_score = snapshot.workflow_complexity_score
 +        instance.workflow_scheduled_next_mode = snapshot.workflow_scheduled_next_mode
 +        instance.last_turn_transition_summary = snapshot.last_turn_transition_summary
 +        instance.last_turn_transition_kind = snapshot.last_turn_transition_kind
 +        instance.last_turn_transition_reason_code = (
 +            snapshot.last_turn_transition_reason_code
 +        )
          instance.compaction = snapshot.compaction
          instance.rotate_after_bytes = rotate_after_bytes
          instance.max_rotated_files = max_rotated_files

src/loader/runtime/workflow.pymodified

          raise ValueError(f"Unknown workflow mode: {value}")
 +class WorkflowDecisionKind(StrEnum):
 +    """Classification for why a workflow mode was selected."""
++
 +    INITIAL_ROUTE = "initial_route"
 +    REQUESTED = "requested"
 +    ARTIFACT_REUSE = "artifact_reuse"
 +    HANDOFF = "handoff"
 +    REENTRY = "reentry"
 +    FORCED = "forced"
++
++
  @dataclass(slots=True)
  class ModeDecision:
      """Router output for the entry point of a task turn."""
      mode: WorkflowMode
 -    reason: str
 +    reason_code: str
 +    reason_summary: str
 +    decision_kind: WorkflowDecisionKind = WorkflowDecisionKind.INITIAL_ROUTE
      ambiguity_score: float = 0.0
      complexity_score: float = 0.0
 +    scheduled_next_mode: WorkflowMode | None = None
++
 +    @property
 +    def reason(self) -> str:
 +        return self.reason_summary
++
 +    @classmethod
 +    def transition(
 +        cls,
 +        mode: WorkflowMode,
 +        *,
 +        reason_code: str,
 +        reason_summary: str,
 +        decision_kind: WorkflowDecisionKind = WorkflowDecisionKind.HANDOFF,
 +        ambiguity_score: float = 0.0,
 +        complexity_score: float = 0.0,
 +        scheduled_next_mode: WorkflowMode | None = None,
 +    ) -> ModeDecision:
 +        """Build a non-router workflow decision for handoffs and reentry."""
++
 +        return cls(
 +            mode=mode,
 +            reason_code=reason_code,
 +            reason_summary=reason_summary,
 +            decision_kind=decision_kind,
 +            ambiguity_score=ambiguity_score,
 +            complexity_score=complexity_score,
 +            scheduled_next_mode=scheduled_next_mode,
 +        )
++
 +    def with_context(
 +        self,
 +        *,
 +        reason_code: str | None = None,
 +        reason_summary: str | None = None,
 +        decision_kind: WorkflowDecisionKind | None = None,
 +        scheduled_next_mode: WorkflowMode | None = None,
 +    ) -> ModeDecision:
 +        """Return a copy with updated contextual routing metadata."""
++
 +        return ModeDecision(
 +            mode=self.mode,
 +            reason_code=reason_code or self.reason_code,
 +            reason_summary=reason_summary or self.reason_summary,
 +            decision_kind=decision_kind or self.decision_kind,
 +            ambiguity_score=self.ambiguity_score,
 +            complexity_score=self.complexity_score,
 +            scheduled_next_mode=scheduled_next_mode,
 +        )
  @dataclass(slots=True)
          if not self.likely_touchpoints:
              self.likely_touchpoints = ["Identify exact files during planning or execution."]
          if not self.assumptions:
 -            self.assumptions = ["Unspecified details stay unchanged unless evidence says otherwise."]
 +            self.assumptions = [
 +                "Unspecified details stay unchanged unless evidence says otherwise.",
 +            ]
          if not self.acceptance_criteria:
              self.acceptance_criteria = list(
                  dict.fromkeys(self.desired_outcome + self.in_scope[:2])
          if requested_mode is not None:
              return ModeDecision(
                  mode=requested_mode,
 -                reason=f"explicit {requested_mode.value} request",
 +                reason_code="explicit_request",
 +                reason_summary=f"explicit {requested_mode.value} request",
 +                decision_kind=WorkflowDecisionKind.REQUESTED,
+             )
          if has_plan:
              return ModeDecision(
                  mode=WorkflowMode.EXECUTE,
 -                reason="reusing existing plan artifacts",
 +                reason_code="existing_plan_artifacts",
 +                reason_summary="reusing existing plan artifacts",
 +                decision_kind=WorkflowDecisionKind.ARTIFACT_REUSE,
+             )
          ambiguity = self._ambiguity_score(task)
          if allow_clarify and not has_brief and ambiguity >= self.clarify_threshold:
              return ModeDecision(
                  mode=WorkflowMode.CLARIFY,
 -                reason="prompt is broad or missing boundaries",
 +                reason_code="task_is_ambiguous",
 +                reason_summary="prompt is broad or missing boundaries",
                  ambiguity_score=ambiguity,
                  complexity_score=complexity,
 +                scheduled_next_mode=WorkflowMode.EXECUTE,
+             )
          if complexity >= self.plan_threshold:
              return ModeDecision(
                  mode=WorkflowMode.PLAN,
 -                reason="task looks complex enough to benefit from a persisted plan",
 +                reason_code="task_is_complex",
 +                reason_summary=(
 +                    "task looks complex enough to benefit from a persisted plan"
 +                ),
                  ambiguity_score=ambiguity,
                  complexity_score=complexity,
 +                scheduled_next_mode=WorkflowMode.EXECUTE,
+             )
          return ModeDecision(
              mode=WorkflowMode.EXECUTE,
 -            reason="task appears concrete enough for direct execution",
 +            reason_code="task_is_concrete",
 +            reason_summary="task appears concrete enough for direct execution",
              ambiguity_score=ambiguity,
              complexity_score=complexity,
+         )
      """Reflect todo state into DoD pending/completed items."""
      special_pending = [
 -        item for item in dod.pending_items if item in {"Complete the requested work", "Collect verification evidence"}
 +        item
 +        for item in dod.pending_items
 +        if item
 +        in {
 +            "Complete the requested work",
 +            "Collect verification evidence",
 +        }
+     ]
      special_completed = [
 -        item for item in dod.completed_items if item in {"Complete the requested work", "Collect verification evidence"}
 +        item
 +        for item in dod.completed_items
 +        if item
 +        in {
 +            "Complete the requested work",
 +            "Collect verification evidence",
 +        }
+     ]
      pending: list[str] = []

tests/test_inspection.pymodified

          permission_rules_source=str(temp_dir / ".loader" / "permission-rules.json"),
          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context", "Mode Guidance"],
 +        workflow_reason_code="verification_failed_reentry",
 +        workflow_reason_summary="verification failed; returning to execute for fixes",
 +        workflow_decision_kind="reentry",
 +        workflow_ambiguity_score=0.1,
 +        workflow_complexity_score=0.7,
 +        workflow_scheduled_next_mode="verify",
 +        active_turn_phase="completion",
 +        last_turn_transition_summary="completion -> finalize [terminal] Finalizing completed turn",
 +        last_turn_transition_kind="terminal",
 +        last_turn_transition_reason_code="turn_complete",
+     )
      SessionStore(temp_dir).save(snapshot)
      return snapshot.session_id, str(dod_path)
          "Workflow Context",
          "Mode Guidance",
+     ]
 +    assert snapshot.workflow_reason_code == "verification_failed_reentry"
 +    assert snapshot.workflow_reason_summary == (
 +        "verification failed; returning to execute for fixes"
 +    )
 +    assert snapshot.workflow_decision_kind == "reentry"
 +    assert snapshot.workflow_scheduled_next_mode == "verify"
 +    assert snapshot.active_turn_phase == "completion"
 +    assert snapshot.last_turn_transition_summary == (
 +        "completion -> finalize [terminal] Finalizing completed turn"
 +    )
      assert len(sessions) == 1
      assert sessions[0].session_id == session_id
          temp_dir / ".loader" / "permission-rules.json"
+     )
      assert sessions[0].prompt_format == "native"
 +    assert sessions[0].workflow_reason_code == "verification_failed_reentry"
 +    assert sessions[0].workflow_reason_summary == (
 +        "verification failed; returning to execute for fixes"
 +    )
 +    assert sessions[0].workflow_decision_kind == "reentry"
 +    assert sessions[0].last_turn_transition_summary == (
 +        "completion -> finalize [terminal] Finalizing completed turn"
 +    )
      assert detail.snapshot.session_id == session_id
      assert detail.is_current is True
      assert detail.snapshot.permission_rules_source == str(
          temp_dir / ".loader" / "permission-rules.json"
+     )
 +    assert detail.snapshot.workflow_reason_code == "verification_failed_reentry"
 +    assert detail.snapshot.last_turn_transition_reason_code == "turn_complete"
  def test_status_and_session_commands_render_persisted_state(

tests/test_runtime_phases.pymodified

+     ]
 +def _turn_phase_events(run) -> list:
 +    return [event for event in run.events if event.type == "turn_phase"]
++
++
  @pytest.mark.asyncio
  async def test_empty_output_enters_repair_phase(temp_dir: Path) -> None:
      backend = ScriptedBackend(
+     )
      phases = _turn_phases(run)
 +    repair_event = next(
 +        event
 +        for event in _turn_phase_events(run)
 +        if event.turn_phase == "repair"
 +    )
      assert "repair" in phases
      assert phases[:3] == ["prepare", "assistant", "repair"]
      assert phases[-2:] == ["completion", "finalize"]
 +    assert repair_event.transition_kind == "retry"
 +    assert repair_event.transition_reason_code == "repair_empty_response"
 +    assert run.agent.last_turn_summary is not None
 +    assert run.agent.last_turn_summary.last_turn_transition_summary == (
 +        "completion -> finalize [terminal] Finalizing completed turn"
 +    )
      assert run.agent.session.active_turn_phase is None
 +    assert run.agent.session.last_turn_transition_reason_code == "turn_complete"
  @pytest.mark.asyncio
      assert "tools" in phases
      assert phases[0] == "prepare"
      assert phases[-1] == "finalize"
 +    assert run.agent.last_turn_summary is not None
 +    assert run.agent.last_turn_summary.last_turn_transition_summary == (
 +        "completion -> finalize [terminal] Finalizing completed turn"
 +    )
 +    assert run.agent.session.last_turn_transition_reason_code == "turn_complete"
      assert any(event.type == "completion_check" for event in run.events)

tests/test_session_state.pymodified

          permission_rules_source=str(temp_dir / ".loader" / "permission-rules.json"),
          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context", "Project Context"],
 +        workflow_reason_code="task_is_complex",
 +        workflow_reason_summary="task looks complex enough to benefit from a persisted plan",
 +        workflow_decision_kind="initial_route",
 +        workflow_ambiguity_score=0.2,
 +        workflow_complexity_score=0.6,
 +        workflow_scheduled_next_mode="execute",
 +        last_turn_transition_summary="completion -> finalize [terminal] Finalizing completed turn",
 +        last_turn_transition_kind="terminal",
 +        last_turn_transition_reason_code="turn_complete",
+     )
      reloaded = ConversationSession.load(
          "Workflow Context",
          "Project Context",
+     ]
 +    assert reloaded.workflow_reason_code == "task_is_complex"
 +    assert reloaded.workflow_reason_summary == (
 +        "task looks complex enough to benefit from a persisted plan"
 +    )
 +    assert reloaded.workflow_decision_kind == "initial_route"
 +    assert reloaded.workflow_ambiguity_score == pytest.approx(0.2)
 +    assert reloaded.workflow_complexity_score == pytest.approx(0.6)
 +    assert reloaded.workflow_scheduled_next_mode == "execute"
 +    assert reloaded.last_turn_transition_summary == (
 +        "completion -> finalize [terminal] Finalizing completed turn"
 +    )
 +    assert reloaded.last_turn_transition_kind == "terminal"
 +    assert reloaded.last_turn_transition_reason_code == "turn_complete"
  @pytest.mark.asyncio

tests/test_turn_state_machine.pyadded

 +"""Tests for the validated runtime turn state machine."""
++
 +from __future__ import annotations
++
 +import pytest
++
 +from loader.runtime.phases import (
 +    TurnPhase,
 +    TurnStateMachine,
 +    TurnTransitionKind,
 +)
++
++
 +def test_turn_state_machine_accepts_valid_transitions() -> None:
 +    machine = TurnStateMachine()
++
 +    prepare = machine.transition(
 +        TurnPhase.PREPARE,
 +        reason_code="prepare_runtime",
 +        reason_summary="Preparing runtime state",
 +    )
 +    assistant = machine.transition(
 +        TurnPhase.ASSISTANT,
 +        reason_code="request_assistant_response",
 +        reason_summary="Requesting assistant response",
 +    )
 +    tools = machine.transition(
 +        TurnPhase.TOOLS,
 +        reason_code="execute_tool_batch",
 +        reason_summary="Executing tool batch",
 +    )
 +    finalize = machine.transition(
 +        TurnPhase.FINALIZE,
 +        reason_code="turn_complete",
 +        reason_summary="Finalizing completed turn",
 +        kind=TurnTransitionKind.TERMINAL,
 +    )
++
 +    assert prepare is not None
 +    assert prepare.from_phase is None
 +    assert prepare.to_phase == "prepare"
 +    assert assistant is not None
 +    assert assistant.from_phase == "prepare"
 +    assert tools is not None
 +    assert tools.from_phase == "assistant"
 +    assert finalize is not None
 +    assert finalize.kind is TurnTransitionKind.TERMINAL
 +    assert machine.current_phase == "finalize"
 +    assert machine.last_transition == finalize
 +    assert finalize.summary == "tools -> finalize [terminal] Finalizing completed turn"
++
++
 +def test_turn_state_machine_rejects_invalid_transitions() -> None:
 +    machine = TurnStateMachine()
 +    machine.transition(
 +        TurnPhase.PREPARE,
 +        reason_code="prepare_runtime",
 +        reason_summary="Preparing runtime state",
 +    )
++
 +    with pytest.raises(ValueError, match="prepare -> tools"):
 +        machine.transition(
 +            TurnPhase.TOOLS,
 +            reason_code="execute_tool_batch",
 +            reason_summary="Executing tool batch",
 +        )

tests/test_workflow_runtime.pymodified

                          id="ask-1",
                          name="AskUserQuestion",
                          arguments={
 -                            "question": "What should stay out of scope for this Loader improvement?",
 +                            "question": (
 +                                "What should stay out of scope for this Loader "
 +                                "improvement?"
 +                            ),
                          },
+                     )
                  ],
      assert Path(dod.clarify_brief).exists()
      assert "runtime behavior" in dod.acceptance_criteria[0].lower()
      assert "## Clarify Mode" in backend.invocations[0].messages[0].content
 +    assert run.agent.last_turn_summary is not None
 +    assert run.agent.last_turn_summary.workflow_mode == "execute"
 +    assert run.agent.last_turn_summary.workflow_reason_code == "post_clarify_task_is_concrete"
 +    assert run.agent.last_turn_summary.workflow_decision_kind == "handoff"
  @pytest.mark.asyncio
      assert Path(dod.verification_plan).exists()
      assert dod.verification_commands == [f"test -f {target}"]
      assert "## Plan Mode" in backend.invocations[0].messages[0].content
 +    assert run.agent.last_turn_summary is not None
 +    assert run.agent.last_turn_summary.workflow_mode == "verify"
 +    assert run.agent.last_turn_summary.workflow_reason_code == (
 +        "definition_of_done_requires_verification"
 +    )
 +    assert run.agent.last_turn_summary.workflow_decision_kind == "handoff"
      verify_calls = [
          event
          for event in run.events