feat: eventhooker provider adapter layer and interpreter invariants

src/hook_master/__init__.py

@@ -11,6 +11,8 @@
 
 from __future__ import annotations
 
+from . import providers
+
 __version__ = "0.2.0"
 
-__all__ = ["__version__"]
+__all__ = ["__version__", "providers"]

src/hook_master/__main__.py

@@ -0,0 +1,6 @@
+from __future__ import annotations
+
+from .cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/hook_master/cli.py

@@ -4,6 +4,7 @@
 import json
 import sys
 from pathlib import Path
+from typing import Any
 
 from . import doctor as doctor_module
 from . import materialize
@@ -51,6 +52,17 @@ def parser() -> argparse.ArgumentParser:
     doctor = commands.add_parser("doctor")
     doctor.add_argument("--id", dest="entry_id")
     doctor.add_argument("--timing", action="store_true")
+    doctor.add_argument("--self-test", action="store_true")
+
+    commands.add_parser("providers")
+
+    snippet = commands.add_parser("install-snippet")
+    snippet.add_argument("--provider", default="claude")
+    snippet.add_argument("--id", dest="entry_id")
+    snippet.add_argument("--script")
+    snippet.add_argument("--event")
+    snippet.add_argument("--python", default="python")
+    snippet.add_argument("--out", type=Path, default=None)
 
     consent = commands.add_parser("consent")
     consent.add_argument("id")
@@ -105,9 +117,53 @@ def main(argv: list[str] | None = None) -> int:
             _print(result)
             return 0 if result["ok"] else 1
         elif args.command == "doctor":
+            if getattr(args, "self_test", False):
+                from .providers.invariants import run_self_test
+
+                st = run_self_test()
+                _print(st)
+                if not st["ok"]:
+                    return 2
             result = doctor_module.run(registry, entry_id=args.entry_id, timing=args.timing)
             _print(result)
             return result["exit_code"]
+        elif args.command == "providers":
+            from .providers import PROVIDER_REGISTRY, resolve_provider
+
+            res: dict[str, Any] = {}
+            for name, prov in PROVIDER_REGISTRY.items():
+                res[name] = "verfuegbar" if prov.is_available() else "nicht verfuegbar"
+            res["gewaehlt"] = resolve_provider().name
+            _print(res)
+        elif args.command == "install-snippet":
+            from .providers import PROVIDER_REGISTRY, format_raw_script_snippet
+
+            if args.entry_id:
+                snippet_dict = format_raw_script_snippet(
+                    args.entry_id,
+                    provider_name=args.provider,
+                    script_path=args.script,
+                    python_executable=args.python,
+                    event_override=args.event,
+                )
+            elif args.script:
+                prov = PROVIDER_REGISTRY.get(args.provider)
+                if prov and hasattr(prov, "script_snippet"):
+                    snippet_dict = prov.script_snippet(
+                        args.script,
+                        event=args.event or "PreToolUse",
+                        python_executable=args.python,
+                    )
+                else:
+                    raise ValueError(f"Provider '{args.provider}' unterstuetzt kein script_snippet")
+            else:
+                raise ValueError("install-snippet benoetigt --id oder --script")
+            out_text = json.dumps(snippet_dict, ensure_ascii=False, indent=2)
+            if args.out:
+                args.out.write_text(out_text, encoding="utf-8")
+                print(f"geschrieben nach {args.out}", file=sys.stderr)
+            else:
+                print(out_text)
         elif args.command == "consent":
             record = ConsentStore().grant(args.id, by=args.by, note=args.note)
             _print(record.to_dict())

src/hook_master/doctor.py

@@ -48,21 +48,85 @@ def _check_config_file(path: Path) -> dict[str, Any]:
     except OSError as exc:
         return {"path": str(path), "state": "unreadable", "error": str(exc)}
     suffix = path.suffix.lower()
+    parsed_data = None
     try:
         if suffix == ".json":
-            json.loads(text)
+            parsed_data = json.loads(text)
         elif suffix == ".toml":
             try:
                 import tomllib  # Python 3.11+
             except ModuleNotFoundError:
                 import tomli as tomllib  # type: ignore[no-redef]
-            tomllib.loads(text)
-        # anderes Format (z.B. .sh) -- nur Existenz+Lesbarkeit zaehlen hier.
+            parsed_data = tomllib.loads(text)
+        if parsed_data is not None:
+            invariant_errors = _scan_hooks_for_invariants(parsed_data)
+            if invariant_errors:
+                return {
+                    "path": str(path),
+                    "state": "invalid-hook-invariants",
+                    "error": "; ".join(invariant_errors),
+                }
         return {"path": str(path), "state": "valid"}
     except Exception as exc:  # noqa: BLE001 -- Parser-Fehler jeder Art melden, nicht crashen
         return {"path": str(path), "state": "invalid", "error": str(exc)}
 
 
+def _scan_hooks_for_invariants(data: Any) -> list[str]:
+    from .providers.invariants import (
+        InterpreterAliasError,
+        InvalidTimeoutError,
+        extract_executable_candidate,
+        validate_interpreter,
+        validate_timeout,
+    )
+
+    errors = []
+    SHELL_BUILTINS = {"echo", "exit", "true", "false", "cd", "set", "export", "source"}
+
+    def check_cmd(cmd: Any, event_name: str) -> None:
+        if not cmd or not isinstance(cmd, str):
+            return
+        exe = extract_executable_candidate(cmd)
+        if exe.lower() in SHELL_BUILTINS:
+            return
+        try:
+            validate_interpreter(exe)
+        except InterpreterAliasError as exc:
+            errors.append(f"{event_name}: {exc}")
+        except FileNotFoundError:
+            pass
+
+    def check_timeout(timeout_val: Any, event_name: str) -> None:
+        if timeout_val is not None:
+            try:
+                validate_timeout(timeout_val)
+            except InvalidTimeoutError as exc:
+                errors.append(f"{event_name}: {exc}")
+
+    if isinstance(data, dict):
+        hooks_section = data.get("hooks")
+        if isinstance(hooks_section, dict):
+            for event, groups in hooks_section.items():
+                if isinstance(groups, list):
+                    for group in groups:
+                        if isinstance(group, dict):
+                            items = group.get("hooks", [group]) if "hooks" in group else [group]
+                            for item in items:
+                                if isinstance(item, dict):
+                                    cmd = item.get("command") or item.get("commandWindows")
+                                    check_cmd(cmd, str(event))
+                                    if "timeout" in item:
+                                        check_timeout(item["timeout"], str(event))
+        elif isinstance(hooks_section, list):
+            for item in hooks_section:
+                if isinstance(item, dict):
+                    event_name = str(item.get("event", "hook"))
+                    check_cmd(item.get("command"), event_name)
+                    if "timeout" in item:
+                        check_timeout(item["timeout"], event_name)
+    return errors
+
+
 def _check_executable(path: Path) -> dict[str, Any]:
     if path.suffix.lower() != ".py":
         return {"state": "skipped-not-python"}
@@ -91,7 +155,7 @@ def bump(level: str) -> None:
             continue
         result = _check_config_file(cfg_path)
         result.update({"check": "config", "agent": target["agent"]})
-        if result["state"] in {"missing", "unreadable", "invalid"}:
+        if result["state"] in {"missing", "unreadable", "invalid", "invalid-hook-invariants"}:
             bump("error")
         findings.append(result)
 

src/hook_master/providers/__init__.py

@@ -0,0 +1,82 @@
+"""Gemeinsame Provider-Adapter-Schicht fuer Agent-Hooks (T-20260921-750493182).
+
+Vereinheitlicht die Provider-Registrierungs-Formate (claude/codex/kimi/agy)
+als gemeinsame Schicht in hook-master, ueber die sowohl Domaenenmodule
+(memoryhooker, workflowhooker) als auch kanonische Roh-Skripte (guards.py,
+token_budget_guard.py, notaus_wake_check.py) registriert werden.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .agy import AgyProvider
+from .base import (
+    BaseProvider,
+    InterpreterAliasError,
+    InvalidTimeoutError,
+    ManualProvider,
+    Provider,
+    UnimplementedProvider,
+    validate_interpreter,
+    validate_timeout,
+)
+from .claude import ClaudeProvider
+from .codex import CodexProvider
+from .git import GitProvider
+from .invariants import run_self_test, verify_timeout_kills
+from .kimi import KimiProvider
+from .raw_scripts import (
+    CANONICAL_SCRIPTS,
+    format_raw_script_snippet,
+    resolve_canonical_script,
+)
+
+PROVIDER_REGISTRY: dict[str, Provider] = {
+    "claude": ClaudeProvider(),
+    "codex": CodexProvider(),
+    "git": GitProvider(),
+    "manual": ManualProvider(),
+    "kimi": KimiProvider(),
+    "agy": AgyProvider(),
+}
+
+
+def resolve_provider(
+    order: list[str] | Any = None,
+    registry: dict[str, Provider] | None = None,
+) -> Provider:
+    """Erster verfuegbarer Provider in der Reihenfolge gewinnt (Fallback-Kette)."""
+    reg = registry or PROVIDER_REGISTRY
+    order_list = getattr(order, "order", order)
+    if not order_list:
+        order_list = ["claude", "codex", "kimi", "agy", "git", "manual"]
+    for name in order_list:
+        provider = reg.get(name)
+        if provider is not None and provider.is_available():
+            return provider
+    return reg.get("manual", ManualProvider())
+
+
+__all__ = [
+    "CANONICAL_SCRIPTS",
+    "PROVIDER_REGISTRY",
+    "AgyProvider",
+    "BaseProvider",
+    "ClaudeProvider",
+    "CodexProvider",
+    "GitProvider",
+    "InterpreterAliasError",
+    "InvalidTimeoutError",
+    "KimiProvider",
+    "ManualProvider",
+    "Provider",
+    "UnimplementedProvider",
+    "format_raw_script_snippet",
+    "resolve_canonical_script",
+    "resolve_provider",
+    "run_self_test",
+    "validate_interpreter",
+    "validate_timeout",
+    "verify_timeout_kills",
+]

src/hook_master/providers/agy.py

@@ -0,0 +1,42 @@
+"""Antigravity (agy) Provider: erzeugt Hook-Konfiguration als Dict/JSON."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from .base import BaseProvider
+
+FORBIDDEN_DEFAULT_EVENT = "PreToolUse"
+
+
+class AgyProvider(BaseProvider):
+    name = "agy"
+    events = ("PreInvocation",)
+    forbidden_events = (FORBIDDEN_DEFAULT_EVENT,)
+
+    def is_available(self) -> bool:
+        return True
+
+    def format_hook(self, command: str, matcher: str | None = None) -> dict[str, Any]:
+        if matcher:
+            return {"matcher": matcher, "hooks": [{"type": "command", "command": command}]}
+        return {"type": "command", "command": command}
+
+    def script_snippet(
+        self,
+        script_path: str | Path,
+        event: str = "PreInvocation",
+        python_executable: str = "python",
+        python_args: list[str] | None = None,
+        matcher: str | None = None,
+        timeout: int | float | None = None,
+    ) -> dict[str, Any]:
+        self.validate_command(python_executable, timeout)
+        args_str = f" {' '.join(python_args)}" if python_args else ""
+        norm_path = str(script_path).replace("\\", "/")
+        cmd = f'{python_executable}{args_str} "{norm_path}"'
+        return {"hooks": {event: [self.format_hook(cmd, matcher=matcher)]}}
+
+
+__all__ = ["AgyProvider", "FORBIDDEN_DEFAULT_EVENT"]

src/hook_master/providers/base.py

@@ -0,0 +1,73 @@
+"""Gemeinsames Provider-Protokoll, Basisklassen und Validierung."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+from .invariants import (
+    InterpreterAliasError,
+    InvalidTimeoutError,
+    validate_interpreter,
+    validate_timeout,
+)
+
+
+@runtime_checkable
+class Provider(Protocol):
+    name: str
+
+    def is_available(self) -> bool:
+        ...
+
+
+class BaseProvider:
+    name: str = "base"
+    events: tuple[str, ...] = ()
+    forbidden_events: tuple[str, ...] = ()
+
+    def is_available(self) -> bool:
+        return True
+
+    def validate_command(
+        self,
+        python_executable: str = "python",
+        timeout: int | float | None = None,
+    ) -> Path:
+        resolved = validate_interpreter(python_executable)
+        if timeout is not None:
+            validate_timeout(timeout)
+        return resolved
+
+
+class UnimplementedProvider(BaseProvider):
+    """Dokumentierter Stub fuer einen Provider, dessen Hook-Bindung noch
+    nicht ermittelt ist (README: 'je Anbieter zu ermitteln, nicht zu raten')."""
+
+    name = "unimplemented"
+    reason = "Hook-Bindung fuer diesen Anbieter ist noch nicht ermittelt."
+
+    def is_available(self) -> bool:
+        return False
+
+
+class ManualProvider(BaseProvider):
+    """Kein Hook -- stellt nur eine CLI bereit, die der Agent manuell aufruft.
+    Immer verfuegbar: garantiert, dass die Provider-Kette nie leerlaeuft."""
+
+    name = "manual"
+
+    def is_available(self) -> bool:
+        return True
+
+
+__all__ = [
+    "BaseProvider",
+    "InterpreterAliasError",
+    "InvalidTimeoutError",
+    "ManualProvider",
+    "Provider",
+    "UnimplementedProvider",
+    "validate_interpreter",
+    "validate_timeout",
+]