feat: add StateMachineTemplate with conditional branching and fallback states

Extends template system with state machine support:
- TemplateState: action + verify_text + next/fallback transitions
- StateMachineTemplate: non-linear execution with error recovery
- 10 new tests covering state transitions, fallback, JSON roundtrip

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

services/kvm_agent/template_store.py

@@ -77,6 +77,90 @@ class OperationTemplate:
         )
 
 
+@dataclass
+class TemplateState:
+    """A state in a state machine template with conditional transitions."""
+
+    action_type: str = ""
+    action_data: dict = field(default_factory=dict)
+    verify_text: str = ""         # OCR text to verify after action
+    next_state: str = ""          # Transition on verification success
+    fallback_state: str = ""      # Transition on verification failure (empty = abandon)
+    wait_after_ms: int = 1000
+
+
+@dataclass
+class StateMachineTemplate:
+    """A state-machine-based template with conditional branching.
+
+    Unlike linear OperationTemplate, this supports:
+    - Conditional transitions based on OCR verification
+    - Fallback states for error recovery
+    - Non-linear execution flows (loops, retries)
+    """
+
+    task_pattern: str = ""
+    states: dict[str, TemplateState] = field(default_factory=dict)
+    start_state: str = "s0"
+    success_count: int = 0
+    failure_count: int = 0
+
+    @property
+    def reliability(self) -> float:
+        total = self.success_count + self.failure_count
+        return self.success_count / total if total > 0 else 0.0
+
+    def get_start(self) -> Optional[TemplateState]:
+        """Get the start state."""
+        return self.states.get(self.start_state)
+
+    def advance(self, current_state_name: str, verified: bool) -> Optional[tuple[str, TemplateState]]:
+        """Advance to the next state based on verification result.
+
+        Returns (state_name, state) or None if the template should be abandoned.
+        """
+        current = self.states.get(current_state_name)
+        if not current:
+            return None
+
+        next_name = current.next_state if verified else current.fallback_state
+        if not next_name:
+            return None  # No transition = abandon template
+
+        next_state = self.states.get(next_name)
+        if not next_state:
+            return None
+
+        return (next_name, next_state)
+
+    def to_json(self) -> str:
+        data = {
+            "task_pattern": self.task_pattern,
+            "start_state": self.start_state,
+            "success_count": self.success_count,
+            "failure_count": self.failure_count,
+            "states": {
+                name: asdict(state)
+                for name, state in self.states.items()
+            },
+        }
+        return json.dumps(data, ensure_ascii=False)
+
+    @classmethod
+    def from_json(cls, text: str) -> "StateMachineTemplate":
+        data = json.loads(text)
+        states = {}
+        for name, sdata in data.get("states", {}).items():
+            states[name] = TemplateState(**sdata)
+        return cls(
+            task_pattern=data.get("task_pattern", ""),
+            states=states,
+            start_state=data.get("start_state", "s0"),
+            success_count=data.get("success_count", 0),
+            failure_count=data.get("failure_count", 0),
+        )
+
+
 class TemplateStore:
     """CRUD + fuzzy matching for operation templates via mem-bridge."""
 

services/kvm_agent/tests/test_template_store.py

@@ -6,6 +6,8 @@ import pytest
 from kvm_agent.template_store import (
     Condition,
     OperationTemplate,
+    StateMachineTemplate,
+    TemplateState,
     TemplateStep,
     TemplateStore,
 )
@@ -115,3 +117,118 @@ class TestTemplateStore:
         store = TemplateStore(mem)
         found = await store.find_template("anything")
         assert found is None
+
+
+class TestStateMachineTemplate:
+    """Tests for state machine templates with conditional branching."""
+
+    def _make_template(self):
+        """Create a simple 3-state template: click → verify → done."""
+        return StateMachineTemplate(
+            task_pattern="open notepad",
+            states={
+                "s0": TemplateState(
+                    action_type="click",
+                    action_data={"x": 0.5, "y": 0.99, "reason": "search"},
+                    verify_text="Type here to search",
+                    next_state="s1",
+                    fallback_state="",  # abandon if search bar not found
+                ),
+                "s1": TemplateState(
+                    action_type="type",
+                    action_data={"text": "notepad"},
+                    verify_text="Notepad",
+                    next_state="s2",
+                    fallback_state="s0",  # retry from start
+                ),
+                "s2": TemplateState(
+                    action_type="click",
+                    action_data={"x": 0.5, "y": 0.3, "reason": "Notepad"},
+                    verify_text="Untitled",
+                    next_state="",  # done
+                ),
+            },
+            start_state="s0",
+            success_count=5,
+            failure_count=1,
+        )
+
+    def test_get_start(self):
+        tmpl = self._make_template()
+        start = tmpl.get_start()
+        assert start is not None
+        assert start.action_type == "click"
+        assert start.verify_text == "Type here to search"
+
+    def test_advance_success(self):
+        tmpl = self._make_template()
+        result = tmpl.advance("s0", verified=True)
+        assert result is not None
+        name, state = result
+        assert name == "s1"
+        assert state.action_type == "type"
+
+    def test_advance_fallback(self):
+        tmpl = self._make_template()
+        result = tmpl.advance("s1", verified=False)
+        assert result is not None
+        name, state = result
+        assert name == "s0"  # Falls back to start
+
+    def test_advance_abandon_on_no_fallback(self):
+        tmpl = self._make_template()
+        result = tmpl.advance("s0", verified=False)
+        assert result is None  # s0 has no fallback → abandon
+
+    def test_advance_done_state(self):
+        tmpl = self._make_template()
+        result = tmpl.advance("s2", verified=True)
+        assert result is None  # s2.next_state is "" → done
+
+    def test_advance_invalid_state(self):
+        tmpl = self._make_template()
+        result = tmpl.advance("nonexistent", verified=True)
+        assert result is None
+
+    def test_reliability(self):
+        tmpl = self._make_template()
+        assert tmpl.reliability == pytest.approx(5 / 6)
+
+    def test_reliability_zero(self):
+        tmpl = StateMachineTemplate()
+        assert tmpl.reliability == 0.0
+
+    def test_json_roundtrip(self):
+        tmpl = self._make_template()
+        json_str = tmpl.to_json()
+        restored = StateMachineTemplate.from_json(json_str)
+        assert restored.task_pattern == "open notepad"
+        assert len(restored.states) == 3
+        assert restored.start_state == "s0"
+        assert restored.success_count == 5
+        assert restored.states["s1"].fallback_state == "s0"
+
+    def test_full_flow_simulation(self):
+        """Simulate a complete template execution: s0→s1→s2→done."""
+        tmpl = self._make_template()
+
+        # Start
+        current_name = tmpl.start_state
+        current = tmpl.get_start()
+        assert current is not None
+
+        # s0 → s1 (verified)
+        result = tmpl.advance(current_name, verified=True)
+        assert result is not None
+        current_name, current = result
+        assert current_name == "s1"
+
+        # s1 → s2 (verified)
+        result = tmpl.advance(current_name, verified=True)
+        assert result is not None
+        current_name, current = result
+        assert current_name == "s2"
+
+        # s2 → done (verified)
+        result = tmpl.advance(current_name, verified=True)
+        assert result is None  # Template complete