HMS MediaEngine Agent
362e089be0
Der Nutzer hat recht: Der Ordner war voller Entwicklungs-Muell. Jetzt ist sauber getrennt: ROOT (was der Nutzer sieht und braucht): - run.py = das Programm - hms_app/ = der Anwendungscode - HMS MediaEngine.app = macOS Doppelklick-Starter - HMS-Start.vbs = Windows Doppelklick-Starter - HMS-Install.vbs = Windows Erst-Installation - HMS-Mac-Install.command = macOS Homebrew-Installation - HMS-Portable-Install.command = macOS Portable-Installation (16GB-Fix) - installer_gui.py = grafischer Installer - launcher.pyw + launcher_core.py = interne Start-Logik - LIESMICH.txt = 10-Zeilen-Kurzanleitung - .gitignore _entwicklung/ (alles andere, NICHT benoetigt): - packages/ apps/ native/ plugins/ tools/ schemas/ tests/ docs/ build/ fixture_profiles/ - PLAN.md STATUS.md ERRORS.md TEST_REPORT.md CHANGELOG.md README.md - pyproject.toml uv.lock setup_*.sh/ps1 make_mac_app.py Diese Trennung gilt ab sofort fuer alle Commits. Der Nutzer kann _entwicklung/ loeschen wenn er Platz braucht - die App laeuft ohne. Verifiziert: App startet nach Aufraeumen unveraendert (Health 200).
38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
"""Idempotency-Registry für wiederholbare Commands (§6.2, §23.2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import OrderedDict
|
|
from typing import Any
|
|
|
|
|
|
class IdempotencyRegistry:
|
|
"""Merkt sich command_id → Ergebnis; Wiederholungen liefern dasselbe Ack."""
|
|
|
|
def __init__(self, capacity: int = 4096) -> None:
|
|
if capacity <= 0:
|
|
raise ValueError("capacity must be positive")
|
|
self._capacity = capacity
|
|
self._entries: OrderedDict[str, Any] = OrderedDict()
|
|
|
|
def register(self, command_id: str) -> bool:
|
|
"""False, wenn die command_id bereits bekannt ist (Duplikat)."""
|
|
if command_id in self._entries:
|
|
self._entries.move_to_end(command_id)
|
|
return False
|
|
self._entries[command_id] = None # Ergebnis folgt mit complete()
|
|
if len(self._entries) > self._capacity:
|
|
self._entries.popitem(last=False)
|
|
return True
|
|
|
|
def complete(self, command_id: str, result: Any) -> None:
|
|
if command_id in self._entries:
|
|
self._entries[command_id] = result
|
|
self._entries.move_to_end(command_id)
|
|
|
|
def result(self, command_id: str) -> Any | None:
|
|
return self._entries.get(command_id)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._entries)
|