From 549f87523a2e4667f0dd9b08b21a08b29cbfcdbd Mon Sep 17 00:00:00 2001 From: Otto Date: Thu, 18 Jun 2026 12:16:54 +0200 Subject: [PATCH] Bootstrap SillyHome Future v2 core --- .gitignore | 8 +++ README.md | 32 ++++++++++ app/__init__.py | 2 + app/api/__init__.py | 2 + app/core/__init__.py | 2 + app/core/decision.py | 66 +++++++++++++++++++ app/core/event_core.py | 89 ++++++++++++++++++++++++++ app/core/handoff.py | 32 ++++++++++ app/core/models.py | 130 ++++++++++++++++++++++++++++++++++++++ app/core/stores.py | 74 ++++++++++++++++++++++ app/main.py | 113 +++++++++++++++++++++++++++++++++ docs/V2_ARCHITECTURE.md | 29 +++++++++ pyproject.toml | 35 ++++++++++ tests/test_future_api.py | 22 +++++++ tests/test_future_core.py | 93 +++++++++++++++++++++++++++ 15 files changed, 729 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 app/__init__.py create mode 100644 app/api/__init__.py create mode 100644 app/core/__init__.py create mode 100644 app/core/decision.py create mode 100644 app/core/event_core.py create mode 100644 app/core/handoff.py create mode 100644 app/core/models.py create mode 100644 app/core/stores.py create mode 100644 app/main.py create mode 100644 docs/V2_ARCHITECTURE.md create mode 100644 pyproject.toml create mode 100644 tests/test_future_api.py create mode 100644 tests/test_future_core.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..321a9cf --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.venv/ +.future_store/ +__pycache__/ +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ +*.egg-info/ + diff --git a/README.md b/README.md new file mode 100644 index 0000000..b1baf59 --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# SillyHome Future + +SillyHome Future is the v2 side project for a local-first Smart Home controller. +It is intentionally separate from `sillyhome-next` so v1 can stay stable while +v2 experiments with a cleaner production architecture. + +## v2 Shape + +- Event Core first: Home Assistant events drive decisions, APIs mostly read state. +- Split stores: runtime, learning and control data are persisted separately. +- Decision Engine v2: safety gates, causal trigger support and audit-first output. +- Handoff Matrix: explicit states for HA automation takeover and rollback. +- Rooms and scenes: groups are first-class planning objects. +- No LLM in the control path. + +## Run + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -e ".[dev]" +uvicorn app.main:app --reload +``` + +## Test + +```bash +pytest +ruff check . +mypy app tests +``` + diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..d3c2b66 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,2 @@ +"""SillyHome Future package.""" + diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e95e744 --- /dev/null +++ b/app/api/__init__.py @@ -0,0 +1,2 @@ +"""API modules.""" + diff --git a/app/core/__init__.py b/app/core/__init__.py new file mode 100644 index 0000000..8377f01 --- /dev/null +++ b/app/core/__init__.py @@ -0,0 +1,2 @@ +"""Core v2 modules.""" + diff --git a/app/core/decision.py b/app/core/decision.py new file mode 100644 index 0000000..6955f4e --- /dev/null +++ b/app/core/decision.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from app.core.models import ControlProfile, Decision, LearningProfile, RuntimeState, SafetyStage + +SAFE_DOMAINS = frozenset({"light", "switch", "fan", "cover", "humidifier"}) + + +class DecisionEngineV2: + def decide( + self, + *, + actuator_entity_id: str, + trigger_entity_id: str, + runtime: RuntimeState, + learning: LearningProfile, + control: ControlProfile, + ) -> Decision: + blockers: list[str] = [] + actuator = runtime.entities.get(actuator_entity_id) + trigger = runtime.entities.get(trigger_entity_id) + if actuator is None: + blockers.append("Aktor ist im Runtime-State nicht bekannt.") + if trigger is None: + blockers.append("Trigger ist im Runtime-State nicht bekannt.") + domain = actuator_entity_id.split(".", 1)[0] + if domain not in SAFE_DOMAINS: + blockers.append(f"Domain {domain} ist nicht fuer Active-Control freigegeben.") + if control.manual_block: + blockers.append("Manuelle Sicherheitssperre ist aktiv.") + best = None + for pattern in learning.patterns: + if pattern.trigger_entity_id != trigger_entity_id: + continue + if trigger is not None and pattern.trigger_state != trigger.state: + continue + if best is None or pattern.confidence > best.confidence: + best = pattern + if best is None: + blockers.append("Kein passendes kausales Muster gefunden.") + return Decision( + actuator_entity_id=actuator_entity_id, + allowed=False, + reason="Keine Aktion faellig.", + blockers=blockers, + trigger_entity_id=trigger_entity_id, + ) + if best.confidence < control.min_confidence: + blockers.append( + f"Confidence {best.confidence:.0%} liegt unter {control.min_confidence:.0%}." + ) + if actuator is not None and actuator.state == best.target_state: + blockers.append("Zielzustand ist bereits erreicht.") + dry_run = control.stage is SafetyStage.DRY_RUN + allowed = not blockers and control.stage in {SafetyStage.ACTIVE, SafetyStage.DRY_RUN} + return Decision( + actuator_entity_id=actuator_entity_id, + target_state=best.target_state, + confidence=best.confidence, + allowed=allowed, + executed=False, + dry_run=dry_run, + reason="Ausfuehrung freigegeben." if allowed and not dry_run else "Dry-run oder blockiert.", + blockers=blockers, + trigger_entity_id=trigger_entity_id, + ) + diff --git a/app/core/event_core.py b/app/core/event_core.py new file mode 100644 index 0000000..14e6801 --- /dev/null +++ b/app/core/event_core.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from app.core.decision import DecisionEngineV2 +from app.core.handoff import HandoffMatrix +from app.core.models import ( + AuditEvent, + ControlProfile, + EntityState, + LearningProfile, + StateEvent, +) +from app.core.stores import FutureStores + + +class RoutingIndex: + def affected_actuators(self, trigger_entity_id: str, learning: dict[str, LearningProfile]) -> list[str]: + result = [ + actuator_id + for actuator_id, profile in learning.items() + if any(pattern.trigger_entity_id == trigger_entity_id for pattern in profile.patterns) + ] + return sorted(set(result)) + + +class EventCore: + def __init__(self, stores: FutureStores) -> None: + self._stores = stores + self._router = RoutingIndex() + self._decision = DecisionEngineV2() + self._handoff = HandoffMatrix() + + def process_state_event(self, event: StateEvent) -> list[AuditEvent]: + runtime = self._stores.runtime() + learning = self._stores.learning() + control = self._stores.control() + runtime.entities[event.entity_id] = EntityState( + entity_id=event.entity_id, + domain=event.entity_id.split(".", 1)[0], + state=event.new_state, + changed_at=event.changed_at, + area_name=event.attributes.get("area_name"), + device_id=event.attributes.get("device_id"), + friendly_name=event.attributes.get("friendly_name"), + ) + audit: list[AuditEvent] = [ + AuditEvent( + event_id=_event_id("state"), + kind="state", + entity_id=event.entity_id, + message=f"{event.entity_id} -> {event.new_state}", + ) + ] + for actuator_id in self._router.affected_actuators(event.entity_id, learning.profiles): + learning_profile = learning.profiles[actuator_id] + control_profile = control.profiles.get( + actuator_id, + ControlProfile(actuator_entity_id=actuator_id), + ) + control_profile = control_profile.model_copy( + update={"handoff_mode": self._handoff.classify(control_profile)} + ) + control.profiles[actuator_id] = control_profile + decision = self._decision.decide( + actuator_entity_id=actuator_id, + trigger_entity_id=event.entity_id, + runtime=runtime, + learning=learning_profile, + control=control_profile, + ) + audit.append( + AuditEvent( + event_id=_event_id("decision"), + kind="decision", + entity_id=actuator_id, + message=decision.reason, + decision=decision, + ) + ) + runtime.audit = [*runtime.audit, *audit][-200:] + self._stores.save_runtime(runtime) + self._stores.save_control(control) + return audit + + +def _event_id(prefix: str) -> str: + return f"{prefix}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S%f')}" + diff --git a/app/core/handoff.py b/app/core/handoff.py new file mode 100644 index 0000000..7ab1ca3 --- /dev/null +++ b/app/core/handoff.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from app.core.models import ControlProfile, HandoffMode + + +class HandoffMatrix: + def classify(self, profile: ControlProfile) -> HandoffMode: + active_related = set(profile.related_automation_ids) - set(profile.paused_automation_ids) + if profile.handoff_mode is HandoffMode.CONTROLLED and active_related: + return HandoffMode.CONFLICT + if profile.paused_automation_ids: + return HandoffMode.CONTROLLED + if profile.related_automation_ids: + return HandoffMode.CANDIDATE + return profile.handoff_mode + + def assume_control(self, profile: ControlProfile) -> ControlProfile: + return profile.model_copy( + update={ + "handoff_mode": HandoffMode.CONTROLLED, + "paused_automation_ids": list(profile.related_automation_ids), + } + ) + + def rollback(self, profile: ControlProfile) -> ControlProfile: + return profile.model_copy( + update={ + "handoff_mode": HandoffMode.ROLLBACK, + "paused_automation_ids": [], + } + ) + diff --git a/app/core/models.py b/app/core/models.py new file mode 100644 index 0000000..433316a --- /dev/null +++ b/app/core/models.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from enum import StrEnum + +from pydantic import BaseModel, Field + + +class HandoffMode(StrEnum): + SHADOW = "shadow" + CANDIDATE = "candidate" + CONTROLLED = "controlled" + CONFLICT = "conflict" + ROLLBACK = "rollback" + + +class SafetyStage(StrEnum): + OBSERVE = "observe" + DRY_RUN = "dry_run" + ACTIVE = "active" + BLOCKED = "blocked" + + +class EntityState(BaseModel): + entity_id: str = Field(pattern=r"^[a-z0-9_]+\.[a-z0-9_]+$") + domain: str + state: str | None = None + changed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + area_name: str | None = None + device_id: str | None = None + friendly_name: str | None = None + + +class StateEvent(BaseModel): + entity_id: str = Field(pattern=r"^[a-z0-9_]+\.[a-z0-9_]+$") + new_state: str | None = None + changed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + attributes: dict[str, str] = Field(default_factory=dict) + + +class BehaviorPatternV2(BaseModel): + actuator_entity_id: str + target_state: str + trigger_entity_id: str | None = None + trigger_state: str | None = None + context: dict[str, str] = Field(default_factory=dict) + support: int = Field(default=1, ge=1) + confidence: float = Field(default=0.0, ge=0.0, le=1.0) + source: str = Field(default="observed", max_length=40) + updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + + +class LearningProfile(BaseModel): + actuator_entity_id: str + patterns: list[BehaviorPatternV2] = Field(default_factory=list) + feedback_positive: int = Field(default=0, ge=0) + feedback_negative: int = Field(default=0, ge=0) + model_version: str = "empty" + + +class ControlProfile(BaseModel): + actuator_entity_id: str + stage: SafetyStage = SafetyStage.OBSERVE + min_confidence: float = Field(default=0.82, ge=0.0, le=1.0) + manual_block: bool = False + cooldown_seconds: int = Field(default=900, ge=0) + handoff_mode: HandoffMode = HandoffMode.SHADOW + related_automation_ids: list[str] = Field(default_factory=list) + paused_automation_ids: list[str] = Field(default_factory=list) + + +class RoomProfile(BaseModel): + room_id: str = Field(pattern=r"^[a-z0-9_-]{1,64}$") + name: str + actuator_entity_ids: list[str] = Field(default_factory=list) + context_entity_ids: list[str] = Field(default_factory=list) + + +class SceneProfile(BaseModel): + scene_id: str = Field(pattern=r"^[a-z0-9_-]{1,64}$") + name: str + actuator_entity_ids: list[str] = Field(default_factory=list) + trigger_entity_id: str | None = None + confidence: float = Field(default=0.0, ge=0.0, le=1.0) + + +class Decision(BaseModel): + actuator_entity_id: str + target_state: str | None = None + confidence: float = Field(default=0.0, ge=0.0, le=1.0) + allowed: bool = False + executed: bool = False + dry_run: bool = False + reason: str + blockers: list[str] = Field(default_factory=list) + trigger_entity_id: str | None = None + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + + +class AuditEvent(BaseModel): + event_id: str + kind: str = Field(max_length=40) + entity_id: str | None = None + message: str = Field(max_length=700) + decision: Decision | None = None + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + + +class RuntimeState(BaseModel): + entities: dict[str, EntityState] = Field(default_factory=dict) + audit: list[AuditEvent] = Field(default_factory=list) + websocket_status: str = "unavailable" + + +class LearningState(BaseModel): + profiles: dict[str, LearningProfile] = Field(default_factory=dict) + rooms: dict[str, RoomProfile] = Field(default_factory=dict) + scenes: dict[str, SceneProfile] = Field(default_factory=dict) + + +class ControlState(BaseModel): + profiles: dict[str, ControlProfile] = Field(default_factory=dict) + + +class BackupBundle(BaseModel): + exported_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + runtime: RuntimeState + learning: LearningState + control: ControlState + diff --git a/app/core/stores.py b/app/core/stores.py new file mode 100644 index 0000000..3958489 --- /dev/null +++ b/app/core/stores.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from threading import RLock +from typing import TypeVar + +from pydantic import BaseModel + +from app.core.models import BackupBundle, ControlState, LearningState, RuntimeState + +T = TypeVar("T", bound=BaseModel) + + +class JsonDocumentStore: + def __init__(self, root: str | Path) -> None: + self.root = Path(root).resolve() + self.root.mkdir(parents=True, exist_ok=True) + self._lock = RLock() + + def load(self, name: str, model: type[T], default: T) -> T: + path = self.root / name + with self._lock: + if not path.exists(): + return default + return model.model_validate_json(path.read_text(encoding="utf-8")) + + def save(self, name: str, value: T) -> T: + path = self.root / name + temporary = path.with_suffix(path.suffix + ".tmp") + with self._lock: + temporary.write_text( + json.dumps(value.model_dump(mode="json"), ensure_ascii=True, sort_keys=True) + "\n", + encoding="utf-8", + ) + os.replace(temporary, path) + return value + + +class FutureStores: + def __init__(self, root: str | Path = ".future_store") -> None: + self._docs = JsonDocumentStore(root) + + def runtime(self) -> RuntimeState: + return self._docs.load("runtime.json", RuntimeState, RuntimeState()) + + def save_runtime(self, state: RuntimeState) -> RuntimeState: + return self._docs.save("runtime.json", state) + + def learning(self) -> LearningState: + return self._docs.load("learning.json", LearningState, LearningState()) + + def save_learning(self, state: LearningState) -> LearningState: + return self._docs.save("learning.json", state) + + def control(self) -> ControlState: + return self._docs.load("control.json", ControlState, ControlState()) + + def save_control(self, state: ControlState) -> ControlState: + return self._docs.save("control.json", state) + + def export_backup(self) -> BackupBundle: + return BackupBundle( + runtime=self.runtime(), + learning=self.learning(), + control=self.control(), + ) + + def restore_backup(self, bundle: BackupBundle) -> None: + self.save_runtime(bundle.runtime) + self.save_learning(bundle.learning) + self.save_control(bundle.control) + diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..0f020e4 --- /dev/null +++ b/app/main.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import os + +from fastapi import FastAPI + +from app.core.event_core import EventCore +from app.core.handoff import HandoffMatrix +from app.core.models import ( + AuditEvent, + BackupBundle, + ControlProfile, + ControlState, + HandoffMode, + LearningState, + RuntimeState, + StateEvent, +) +from app.core.stores import FutureStores + +app = FastAPI( + title="SillyHome Future API", + description="SillyHome v2 event-core side project.", + version="2.0.0-alpha.1", +) +stores = FutureStores(os.getenv("SILLYHOME_FUTURE_STORE", ".future_store")) +event_core = EventCore(stores) +handoff = HandoffMatrix() + + +@app.get("/health") +def health() -> dict[str, str]: + return {"status": "ok", "version": app.version} + + +@app.post("/v2/events/state", response_model=list[AuditEvent]) +def ingest_state_event(event: StateEvent) -> list[AuditEvent]: + return event_core.process_state_event(event) + + +@app.get("/v2/runtime", response_model=RuntimeState) +def get_runtime() -> RuntimeState: + return stores.runtime() + + +@app.get("/v2/learning", response_model=LearningState) +def get_learning() -> LearningState: + return stores.learning() + + +@app.put("/v2/learning", response_model=LearningState) +def put_learning(state: LearningState) -> LearningState: + return stores.save_learning(state) + + +@app.get("/v2/control", response_model=ControlState) +def get_control() -> ControlState: + return stores.control() + + +@app.put("/v2/control/{actuator_entity_id}", response_model=ControlProfile) +def put_control(actuator_entity_id: str, profile: ControlProfile) -> ControlProfile: + state = stores.control() + state.profiles[actuator_entity_id] = profile + stores.save_control(state) + return profile + + +@app.post("/v2/handoff/{actuator_entity_id}/assume", response_model=ControlProfile) +def assume_control(actuator_entity_id: str) -> ControlProfile: + state = stores.control() + profile = state.profiles.get( + actuator_entity_id, + ControlProfile(actuator_entity_id=actuator_entity_id), + ) + updated = handoff.assume_control(profile) + state.profiles[actuator_entity_id] = updated + stores.save_control(state) + return updated + + +@app.post("/v2/handoff/{actuator_entity_id}/rollback", response_model=ControlProfile) +def rollback_control(actuator_entity_id: str) -> ControlProfile: + state = stores.control() + profile = state.profiles.get( + actuator_entity_id, + ControlProfile(actuator_entity_id=actuator_entity_id), + ) + updated = handoff.rollback(profile) + state.profiles[actuator_entity_id] = updated + stores.save_control(state) + return updated + + +@app.get("/v2/handoff/{actuator_entity_id}", response_model=HandoffMode) +def classify_handoff(actuator_entity_id: str) -> HandoffMode: + profile = stores.control().profiles.get( + actuator_entity_id, + ControlProfile(actuator_entity_id=actuator_entity_id), + ) + return handoff.classify(profile) + + +@app.get("/v2/backup/export", response_model=BackupBundle) +def export_backup() -> BackupBundle: + return stores.export_backup() + + +@app.post("/v2/backup/restore") +def restore_backup(bundle: BackupBundle) -> dict[str, str]: + stores.restore_backup(bundle) + return {"status": "restored"} + diff --git a/docs/V2_ARCHITECTURE.md b/docs/V2_ARCHITECTURE.md new file mode 100644 index 0000000..0fc460e --- /dev/null +++ b/docs/V2_ARCHITECTURE.md @@ -0,0 +1,29 @@ +# SillyHome Future v2 Architecture + +## Core Principle + +v2 is event-first. Home Assistant state changes enter the Event Core, which +updates runtime state, routes affected actuators, computes a decision, applies +safety and handoff gates, then writes an audit event. + +## Store Split + +- `runtime.json`: current states, audit trail and connection health +- `learning.json`: patterns, room profiles and scene profiles +- `control.json`: safety stage, handoff mode and paused automations + +## Control Path + +LLMs are not part of the control path. Local deterministic insights may explain +state, but switching decisions must remain inspectable and testable. + +## v2 Milestones + +1. Event Core and routing index +2. Store migration and backup/restore +3. Decision Engine v2 +4. Handoff Matrix +5. Rooms and scenes +6. Dashboard v2 +7. Add-on hardening + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d744ba8 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "sillyhome-future" +version = "2.0.0-alpha.1" +description = "SillyHome v2 event-core prototype" +requires-python = ">=3.11" +dependencies = [ + "fastapi>=0.115", + "pydantic>=2.8", + "uvicorn[standard]>=0.30", +] + +[project.optional-dependencies] +dev = [ + "httpx>=0.27", + "mypy>=1.10", + "pytest>=8.2", + "ruff>=0.6", +] + +[tool.setuptools.packages.find] +include = ["app*"] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.mypy] +python_version = "3.11" +strict = true +plugins = [] + diff --git a/tests/test_future_api.py b/tests/test_future_api.py new file mode 100644 index 0000000..e183094 --- /dev/null +++ b/tests/test_future_api.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from fastapi.testclient import TestClient + +from app.main import app, stores + + +def test_health_and_backup_roundtrip(tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("SILLYHOME_FUTURE_STORE", str(tmp_path)) + client = TestClient(app) + + health = client.get("/health") + backup = client.get("/v2/backup/export") + restore = client.post("/v2/backup/restore", json=backup.json()) + + assert stores is not None + assert health.status_code == 200 + assert health.json()["version"] == "2.0.0-alpha.1" + assert backup.status_code == 200 + assert restore.status_code == 200 + assert restore.json() == {"status": "restored"} + diff --git a/tests/test_future_core.py b/tests/test_future_core.py new file mode 100644 index 0000000..afdffc0 --- /dev/null +++ b/tests/test_future_core.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from pathlib import Path + +from app.core.event_core import EventCore +from app.core.handoff import HandoffMatrix +from app.core.models import ( + BehaviorPatternV2, + ControlProfile, + EntityState, + HandoffMode, + LearningProfile, + LearningState, + RuntimeState, + SafetyStage, + StateEvent, +) +from app.core.stores import FutureStores + + +def test_event_core_routes_state_event_to_decision(tmp_path: Path) -> None: + stores = FutureStores(tmp_path) + stores.save_runtime( + RuntimeState( + entities={ + "light.storage": EntityState( + entity_id="light.storage", + domain="light", + state="off", + ) + } + ) + ) + stores.save_learning( + LearningState( + profiles={ + "light.storage": LearningProfile( + actuator_entity_id="light.storage", + patterns=[ + BehaviorPatternV2( + actuator_entity_id="light.storage", + target_state="on", + trigger_entity_id="binary_sensor.storage_door", + trigger_state="on", + support=3, + confidence=0.95, + ) + ], + ) + } + ) + ) + stores.save_control( + stores.control().model_copy( + update={ + "profiles": { + "light.storage": ControlProfile( + actuator_entity_id="light.storage", + stage=SafetyStage.DRY_RUN, + min_confidence=0.8, + ) + } + } + ) + ) + core = EventCore(stores) + + audit = core.process_state_event( + StateEvent(entity_id="binary_sensor.storage_door", new_state="on") + ) + + decisions = [item for item in audit if item.kind == "decision"] + assert decisions + assert decisions[0].decision is not None + assert decisions[0].decision.target_state == "on" + assert decisions[0].decision.allowed is True + assert decisions[0].decision.dry_run is True + + +def test_handoff_matrix_detects_conflict_and_rollback() -> None: + matrix = HandoffMatrix() + profile = ControlProfile( + actuator_entity_id="light.storage", + handoff_mode=HandoffMode.CONTROLLED, + related_automation_ids=["automation.storage"], + ) + + assert matrix.classify(profile) is HandoffMode.CONFLICT + controlled = matrix.assume_control(profile) + assert controlled.paused_automation_ids == ["automation.storage"] + rolled_back = matrix.rollback(controlled) + assert rolled_back.handoff_mode is HandoffMode.ROLLBACK + assert rolled_back.paused_automation_ids == []