diff --git a/coma/clutch.py b/coma/clutch.py
new file mode 100644
index 0000000..3784cd7
--- /dev/null
+++ b/coma/clutch.py
@@ -0,0 +1,172 @@
+"""Clutch-Anbindung fuer rollenspezifische Modellauswahl und Fallback.
+
+E02: Clutch entscheidet Anbieter/Modell-Fallback je Rolle einzeln waehlbar,
+lokal ueberstimmbar, abschaltbar.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+from dataclasses import dataclass
+from typing import Callable, Mapping, Sequence
+
+from .session import Candidate, normalize_provider, PROVIDERS
+
+# Bekannte Gaenge auf CLI-Provider abbilden
+GANG_PREFIX_MAP: tuple[tuple[str, str, Callable[[str], str]], ...] = (
+    ("claude-", "claude", lambda name: "sonnet" if "sonnet" in name else ("opus" if "opus" in name else ("haiku" if "haiku" in name else name.replace("claude-", "")))),
+    ("openai-", "codex", lambda name: name.replace("openai-", "")),
+    ("agy-", "agy", lambda name: name.replace("agy-", "")),
+    ("kimi-", "kimi", lambda name: ""),
+    ("gemini-", "agy", lambda name: name),
+)
+
+
+def is_clutch_available(
+    which: Callable[[str], str | None] = shutil.which,
+    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+) -> bool:
+    """Prueft, ob die clutch-CLI vorhanden und lauffaehig ist."""
+    exe = which("clutch")
+    if not exe:
+        return False
+    try:
+        res = run([exe, "--version"], capture_output=True, text=True, timeout=5.0)
+        return res.returncode == 0
+    except (OSError, subprocess.SubprocessError):
+        return False
+
+
+def get_clutch_models_status(
+    which: Callable[[str], str | None] = shutil.which,
+    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+    timeout: float = 10.0,
+) -> dict[str, dict]:
+    """Liest Modell-Verfuegbarkeit und Quota-Sperren via `clutch models --status --json`."""
+    exe = which("clutch")
+    if not exe:
+        return {}
+    try:
+        res = run(
+            [exe, "models", "--status", "--json"],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+        if res.returncode != 0 or not res.stdout.strip():
+            return {}
+        data = json.loads(res.stdout)
+        if not isinstance(data, list):
+            return {}
+        return {str(item.get("name")): item for item in data if isinstance(item, dict) and "name" in item}
+    except (OSError, subprocess.SubprocessError, json.JSONDecodeError):
+        return {}
+
+
+def gang_to_candidate(
+    gang_name: str,
+    effort: str = "",
+    models_status: Mapping[str, dict] | None = None,
+) -> tuple[Candidate | None, str | None]:
+    """Mappt einen clutch-Gang auf (Candidate, skip_reason)."""
+    raw_gang = str(gang_name).strip()
+    if not raw_gang:
+        return None, "leerer Gangname"
+
+    if models_status:
+        status = models_status.get(raw_gang, {})
+        avail = status.get("availability")
+        if avail in ("blocked", "disabled"):
+            reason = status.get("availability_reason") or f"Modell ist {avail}"
+            return None, f"{raw_gang} — {reason}"
+
+    # 1. Bekannte Praefixe
+    for prefix, provider, model_fn in GANG_PREFIX_MAP:
+        if raw_gang.startswith(prefix):
+            model = model_fn(raw_gang)
+            return Candidate(provider, model, effort, source="clutch"), None
+
+    # 2. Status-Lookup (Runners)
+    if models_status:
+        status = models_status.get(raw_gang, {})
+        runners = status.get("runners") or []
+        for r in runners:
+            if r in PROVIDERS:
+                model_id = str(status.get("model_id") or "")
+                return Candidate(r, model_id, effort, source="clutch"), None
+
+    return None, f"{raw_gang} — kein passender CLI-Provider (Runner wird nicht unterstuetzt)"
+
+
+def resolve_clutch_candidates(
+    prompt: str,
+    *,
+    zweck: str = "",
+    effort: str = "",
+    fallback: bool = True,
+    which: Callable[[str], str | None] = shutil.which,
+    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+    timeout: float = 10.0,
+) -> tuple[tuple[Candidate, ...], tuple[str, ...]]:
+    """Fragt clutch route ab und liefert gerankte Kandidaten."""
+    exe = which("clutch")
+    if not exe:
+        return (), ("clutch CLI nicht gefunden",)
+
+    cmd = [exe, "route", str(prompt or "TASKPLAN worker"), "--json"]
+    if zweck:
+        cmd.extend(["--zweck", str(zweck)])
+    if effort:
+        cmd.extend(["--effort", str(effort)])
+
+    try:
+        res = run(cmd, capture_output=True, text=True, timeout=timeout)
+        if res.returncode != 0 or not res.stdout.strip():
+            err = (res.stderr or res.stdout or f"Exit {res.returncode}").strip()
+            return (), (f"clutch route fehlgeschlagen: {err}",)
+        data = json.loads(res.stdout)
+        if not isinstance(data, dict):
+            return (), ("clutch route lieferte kein gueltiges JSON-Objekt",)
+    except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as exc:
+        return (), (f"clutch route Aufruf fehlgeschlagen: {exc}",)
+
+    models_status = get_clutch_models_status(which=which, run=run, timeout=timeout)
+
+    primary_gang = data.get("gang")
+    if not primary_gang:
+        return (), ("clutch route enthielt keinen primaeren Gang",)
+
+    clutch_effort = data.get("effort") or effort or ""
+    gangs = [primary_gang]
+    if fallback:
+        alternativen = data.get("alternativen") or []
+        if isinstance(alternativen, list):
+            gangs.extend(alternativen)
+
+    candidates: list[Candidate] = []
+    skipped: list[str] = []
+    seen_candidates: set[tuple[str, str, str]] = set()
+
+    for gang in gangs:
+        cand, reason = gang_to_candidate(gang, clutch_effort, models_status)
+        if cand is None:
+            if reason:
+                skipped.append(reason)
+            continue
+        key = (cand.provider, cand.model, cand.effort)
+        if key not in seen_candidates:
+            seen_candidates.add(key)
+            candidates.append(cand)
+
+    return tuple(candidates), tuple(skipped)
+
+
+__all__ = [
+    "GANG_PREFIX_MAP",
+    "gang_to_candidate",
+    "get_clutch_models_status",
+    "is_clutch_available",
+    "resolve_clutch_candidates",
+]
diff --git a/coma/session.py b/coma/session.py
index 039ffbd..d3572d2 100644
--- a/coma/session.py
+++ b/coma/session.py
@@ -76,6 +76,7 @@ class Candidate:
     provider: str
     model: str = ""
     effort: str = ""
+    source: str = "config"
 
 
 @dataclass(frozen=True)
@@ -272,16 +273,16 @@ def ordered_candidates(
     """Deduplizierte Kette: explizite Wahl, Default, Ersatzanbieter."""
     wanted = (primary,) + ((provider_default,) if provider_default else ()) + tuple(fallbacks)
     chain: list[Candidate] = []
-    seen: set[Candidate] = set()
+    seen: set[tuple[str, str, str]] = set()
     for item in wanted:
-        candidate = Candidate(
-            normalize_provider(item.provider),
-            str(item.model).strip(),
-            str(item.effort).strip(),
-        )
-        if candidate not in seen:
-            seen.add(candidate)
-            chain.append(candidate)
+        provider = normalize_provider(item.provider)
+        model = str(item.model).strip()
+        effort = str(item.effort).strip()
+        source = getattr(item, "source", "config")
+        key = (provider, model, effort)
+        if key not in seen:
+            seen.add(key)
+            chain.append(Candidate(provider, model, effort, source=source))
     return tuple(chain)
 
 
@@ -296,10 +297,12 @@ def available_candidates(
     skipped: list[str] = []
     unavailable: set[str] = set()
     for raw in candidates:
+        source = getattr(raw, "source", "config")
         candidate = Candidate(
             normalize_provider(raw.provider),
             str(raw.model).strip(),
             str(raw.effort).strip(),
+            source=source,
         )
         name = candidate.provider
         if name in unavailable:
@@ -429,9 +432,18 @@ def stop() -> None:
     return False, f"Exit {proc.returncode} ohne {PROBE_TOKEN}"
 
 
+from .clutch import (
+    GANG_PREFIX_MAP,
+    gang_to_candidate,
+    get_clutch_models_status,
+    is_clutch_available,
+    resolve_clutch_candidates,
+)
+
 __all__ = [
     "CAPABILITIES",
     "DEFAULT_PROBE_TIMEOUT",
+    "GANG_PREFIX_MAP",
     "PROBE_REQUEST",
     "PROBE_SENTINEL",
     "PROBE_TOKEN",
@@ -443,8 +455,12 @@ def stop() -> None:
     "available_candidates",
     "build_probe_command",
     "build_session_plan",
+    "gang_to_candidate",
+    "get_clutch_models_status",
+    "is_clutch_available",
     "normalize_mode",
     "normalize_provider",
     "ordered_candidates",
     "probe",
+    "resolve_clutch_candidates",
 ]
diff --git a/tests/test_clutch_resolution.py b/tests/test_clutch_resolution.py
new file mode 100644
index 0000000..18fba9d
--- /dev/null
+++ b/tests/test_clutch_resolution.py
@@ -0,0 +1,191 @@
+"""Tests fuer die Clutch-Anbindung in COMA (T-20260906-737455509)."""
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Sequence
+
+import pytest
+
+from coma.clutch import (
+    gang_to_candidate,
+    get_clutch_models_status,
+    is_clutch_available,
+    resolve_clutch_candidates,
+)
+from coma.session import Candidate, ordered_candidates, available_candidates
+
+
+class FakeCompletedProcess:
+    def __init__(self, returncode: int = 0, stdout: str = "", stderr: str = ""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+def test_is_clutch_available():
+    # 1. Nicht im Pfad
+    assert not is_clutch_available(which=lambda _: None)
+
+    # 2. Im Pfad, aber Fehler beim Ausfuehren
+    def run_fail(*args, **kwargs):
+        raise OSError("Permission denied")
+    assert not is_clutch_available(which=lambda _: "/bin/clutch", run=run_fail)
+
+    # 3. Im Pfad, Exit != 0
+    assert not is_clutch_available(
+        which=lambda _: "/bin/clutch",
+        run=lambda *args, **kwargs: FakeCompletedProcess(returncode=1),
+    )
+
+    # 4. Erfolgreich
+    assert is_clutch_available(
+        which=lambda _: "/bin/clutch",
+        run=lambda *args, **kwargs: FakeCompletedProcess(returncode=0, stdout="clutch 0.6.3"),
+    )
+
+
+def test_gang_to_candidate():
+    # Claude
+    cand, skip = gang_to_candidate("claude-sonnet", effort="high")
+    assert skip is None
+    assert cand == Candidate("claude", "sonnet", "high", source="clutch")
+
+    # Claude Opus
+    cand, skip = gang_to_candidate("claude-opus", effort="xhigh")
+    assert skip is None
+    assert cand == Candidate("claude", "opus", "xhigh", source="clutch")
+
+    # Codex (OpenAI)
+    cand, skip = gang_to_candidate("openai-gpt-5.6-terra", effort="high")
+    assert skip is None
+    assert cand == Candidate("codex", "gpt-5.6-terra", "high", source="clutch")
+
+    # AGY (Gemini / Antigravity)
+    cand, skip = gang_to_candidate("agy-gemini-3.5-flash", effort="low")
+    assert skip is None
+    assert cand == Candidate("agy", "gemini-3.5-flash", "low", source="clutch")
+
+    # Kimi
+    cand, skip = gang_to_candidate("kimi-cli", effort="max")
+    assert skip is None
+    assert cand == Candidate("kimi", "", "max", source="clutch")
+
+    # Blockiertes Modell via models_status
+    status_map = {
+        "claude-sonnet": {
+            "name": "claude-sonnet",
+            "availability": "blocked",
+            "availability_reason": "Quota erreicht",
+        }
+    }
+    cand, skip = gang_to_candidate("claude-sonnet", effort="high", models_status=status_map)
+    assert cand is None
+    assert "Quota erreicht" in skip
+
+    # Unbekannter/Nicht unterstuetzter Runner
+    cand, skip = gang_to_candidate("ollama-qwen3", effort="low")
+    assert cand is None
+    assert "nicht unterstuetzt" in skip
+
+
+def test_resolve_clutch_candidates_missing_clutch():
+    candidates, skipped = resolve_clutch_candidates(
+        "Fix auth bug",
+        which=lambda _: None,
+    )
+    assert candidates == ()
+    assert "nicht gefunden" in skipped[0]
+
+
+def test_resolve_clutch_candidates_route_failure():
+    def run_mock(cmd, *args, **kwargs):
+        return FakeCompletedProcess(returncode=1, stderr="Syntax error in route")
+
+    candidates, skipped = resolve_clutch_candidates(
+        "Fix auth bug",
+        which=lambda _: "/bin/clutch",
+        run=run_mock,
+    )
+    assert candidates == ()
+    assert "fehlgeschlagen" in skipped[0]
+
+
+def test_resolve_clutch_candidates_success():
+    route_output = json.dumps({
+        "gang": "claude-sonnet",
+        "provider": "anthropic",
+        "effort": "high",
+        "alternativen": [
+            "openai-gpt-5.6-terra",
+            "agy-gemini-3.5-flash",
+            "ollama-qwen3",
+        ]
+    })
+    status_output = json.dumps([
+        {"name": "claude-sonnet", "availability": "available", "runners": ["claude"]},
+        {"name": "openai-gpt-5.6-terra", "availability": "available", "runners": ["codex"]},
+        {"name": "agy-gemini-3.5-flash", "availability": "available", "runners": ["agy"]},
+        {"name": "ollama-qwen3", "availability": "available", "runners": ["ollama"]},
+    ])
+
+    def run_mock(cmd, *args, **kwargs):
+        if "route" in cmd:
+            return FakeCompletedProcess(returncode=0, stdout=route_output)
+        if "models" in cmd:
+            return FakeCompletedProcess(returncode=0, stdout=status_output)
+        return FakeCompletedProcess(returncode=1)
+
+    candidates, skipped = resolve_clutch_candidates(
+        "Fix auth bug",
+        zweck="coding",
+        effort="high",
+        which=lambda _: "/bin/clutch",
+        run=run_mock,
+    )
+    assert len(candidates) == 3
+    assert candidates[0] == Candidate("claude", "sonnet", "high", source="clutch")
+    assert candidates[1] == Candidate("codex", "gpt-5.6-terra", "high", source="clutch")
+    assert candidates[2] == Candidate("agy", "gemini-3.5-flash", "high", source="clutch")
+    assert any("ollama-qwen3" in s for s in skipped)
+
+
+def test_resolve_clutch_candidates_no_fallback():
+    route_output = json.dumps({
+        "gang": "claude-sonnet",
+        "effort": "high",
+        "alternativen": ["openai-gpt-5.6-terra"],
+    })
+
+    def run_mock(cmd, *args, **kwargs):
+        if "route" in cmd:
+            return FakeCompletedProcess(returncode=0, stdout=route_output)
+        if "models" in cmd:
+            return FakeCompletedProcess(returncode=0, stdout="[]")
+        return FakeCompletedProcess(returncode=1)
+
+    candidates, skipped = resolve_clutch_candidates(
+        "Fix auth bug",
+        fallback=False,
+        which=lambda _: "/bin/clutch",
+        run=run_mock,
+    )
+    assert len(candidates) == 1
+    assert candidates[0] == Candidate("claude", "sonnet", "high", source="clutch")
+
+
+def test_ordered_and_available_candidates_preserves_source():
+    c1 = Candidate("claude", "sonnet", "high", source="clutch")
+    c2 = Candidate("codex", "gpt-5.6-terra", "high", source="config")
+
+    ordered = ordered_candidates(c1, fallbacks=[c2])
+    assert ordered[0].source == "clutch"
+    assert ordered[1].source == "config"
+
+    available, skipped = available_candidates(
+        ordered,
+        which=lambda p: f"/bin/{p}",
+    )
+    assert len(available) == 2
+    assert available[0].source == "clutch"
+    assert available[1].source == "config"