"""Secure Gateway: der EINE Ausführungspfad des Aria-V3-Kerns. Telegram-Plugin (gate(), Permission-Intercept) -> Bridge -> Gateway.handle(): 1. Modus freigeschaltet? (config.allowed_modes; Default nur "shadow") 2. AUTHENTIFIZIERUNG: sender_id kommt NUR aus dem vom Transport gesetzten Feld (Plugin: ctx.from.id), nie aus dem Text oder selbstdeklarierten Feldern (unbekannte Felder werden abgelehnt). Abgleich mit der Allowlist; privat zusätzlich chat_id == sender_id. 3. Idempotenz je (chat_id, message_id), persistent. 4. Ein Befehlsparser; Modusbefehle nur für authentifizierte Besitzer. 5. Kontext bauen; Datenschutz- und Aufgabenklasse deterministisch. 6. Verfügbarkeit (Circuit-Breaker) und Budget (ohne Reservierung). 7. Eine Routinglogik (core.route); manuelle Wahl überstimmt nie. 8. Memory nur für Adapter mit accepts_memory, Ausschnitt für Ausschnitt gegen die Matrix geprüft; danach finale Klassifikation. 9. Shadow: nur protokollieren, nichts reservieren, nichts ausführen. 10. Ausführen: Outbound-Digest vom Adapter, Budget reservieren, HMAC- Clearance ausstellen, Entscheidung protokollieren (sonst keine Ausführung), adapter.execute(clearance, job) - der Adapter prüft selbst -, Budget abrechnen, Health verbuchen, Ergebnisprüfung, Ergebnis protokollieren. Es gibt keinen automatischen Wechsel auf ein anderes Backend nach einem Laufzeitfehler. Der einzige Fallback ist der optionale, vollständig geprüfte Fallback bei manueller Wahl (routing.manual_fallback = checked_fallback). """ from __future__ import annotations import json import re import time from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from typing import Any, Callable, Mapping, Optional, Protocol, Sequence from .adapter_contract import ( AdapterInputError, AdapterJob, AdapterResult, BackendAdapter, ExecutionPolicy, ResultStatus, ) from .authorization import ( AuthorizationError, ClearanceIssuer, ClearanceKeyring, ClearanceVerifier, KeyUnavailableError, ) from .budget import BudgetLedger, estimate_tokens from .claude_channel import ClaudeChannelAdapter from .classify import PrivacyAssessment, PrivacyClassifier, classify_task from .codex_adapter import CodexAdapter from .commands import CommandKind, parse_command from .config import ConfigurationError, GatewayConfig from .context import Attachment, ContextEnvelope, MemorySnippet, sha256_hex from .core import ( Backend, CHECKED_FALLBACK, JobRequest, MANUAL_CHOICE_NOT_PERMITTED, Mode, PRIVACY_DENIED, PrivacyTier, RoutingDecision, route, ) from .decision_log import DecisionLog from .health import HealthRegistry from .lifecycle import JobLifecycle, JobStatus from .result_check import check_output from .runner import IsolatedRunner, MockRunner, SubprocessRunner from .state import IdempotencyStore, ModeStore, NonceStore, StateError PROTOCOL_VERSION = 1 _USER_ID = re.compile(r"^\d{1,20}$") _CHAT_ID = re.compile(r"^-?\d{1,20}$") CHAT_TYPES = frozenset({"private", "group", "supergroup", "channel"}) BRIDGE_REQUEST_FIELDS = frozenset({ "transport", "sender_id", "sender_username", "chat_id", "chat_type", "message_id", "ts", "text", "attachments", }) BRIDGE_ATTACHMENT_FIELDS = frozenset({"kind", "file_id", "size", "mime", "name", "local_path", "sha256"}) class Handling(str, Enum): DROP = "drop" NOT_INTERCEPTED = "not_intercepted" REPLY = "reply" BLOCKED = "blocked" FAILED = "failed" FORWARD_TO_CLAUDE_CHANNEL = "forward_to_claude_channel" SHADOW_DECISION = "shadow_decision" ERROR = "error" class Principal(str, Enum): OWNER = "owner" GROUP_MEMBER = "group_member" class RequestError(ValueError): def __init__(self, code: str): super().__init__(code) self.code = code @dataclass(frozen=True) class GatewayRequest: """Eingang vom Transport. sender_id/chat_id/chat_type setzt NUR das Plugin.""" sender_id: str chat_id: str chat_type: str message_id: int text: str transport: str = "telegram" sender_username: Optional[str] = None ts: Optional[str] = None attachments: tuple[Attachment, ...] = () def __post_init__(self): if not isinstance(self.sender_id, str) or not _USER_ID.match(self.sender_id): raise RequestError("INVALID_SENDER_ID") if not isinstance(self.chat_id, str) or not _CHAT_ID.match(self.chat_id): raise RequestError("INVALID_CHAT_ID") if self.chat_type not in CHAT_TYPES: raise RequestError("INVALID_CHAT_TYPE") if type(self.message_id) is not int or self.message_id < 0: raise RequestError("INVALID_MESSAGE_ID") if not isinstance(self.text, str) or "\x00" in self.text: raise RequestError("INVALID_TEXT") if not isinstance(self.transport, str): raise RequestError("INVALID_TRANSPORT") for value, code in ((self.sender_username, "INVALID_USERNAME"), (self.ts, "INVALID_TS")): if value is not None and (not isinstance(value, str) or len(value) > 64 or "\x00" in value): raise RequestError(code) if not isinstance(self.attachments, tuple) or any(not isinstance(a, Attachment) for a in self.attachments): raise RequestError("INVALID_ATTACHMENTS") @classmethod def from_bridge(cls, data: Mapping[str, Any], *, max_text_bytes: int, max_attachments: int) -> "GatewayRequest": if not isinstance(data, Mapping): raise RequestError("INVALID_REQUEST") unknown = set(data) - BRIDGE_REQUEST_FIELDS if unknown: # Selbstdeklarierte Felder (user_id, from, privacy, trusted, ...) sind nie zulässig. raise RequestError("UNKNOWN_FIELD") for required in ("transport", "sender_id", "chat_id", "chat_type", "message_id", "text"): if required not in data: raise RequestError("MISSING_FIELD") text = data["text"] if not isinstance(text, str): raise RequestError("INVALID_TEXT") if len(text.encode("utf-8")) > max_text_bytes: raise RequestError("TEXT_TOO_LARGE") raw_attachments = data.get("attachments", []) if not isinstance(raw_attachments, list) or len(raw_attachments) > max_attachments: raise RequestError("INVALID_ATTACHMENTS") attachments = [] for item in raw_attachments: if not isinstance(item, Mapping) or set(item) - BRIDGE_ATTACHMENT_FIELDS or "kind" not in item: raise RequestError("INVALID_ATTACHMENTS") try: attachments.append(Attachment( kind=item["kind"], file_id=item.get("file_id", ""), name=item.get("name"), mime=item.get("mime"), size=item.get("size"), sha256=item.get("sha256"), local_path=item.get("local_path"), )) except ValueError as exc: raise RequestError("INVALID_ATTACHMENTS") from exc return cls( sender_id=data["sender_id"], chat_id=data["chat_id"], chat_type=data["chat_type"], message_id=data["message_id"], text=text, transport=data["transport"], sender_username=data.get("sender_username"), ts=data.get("ts"), attachments=tuple(attachments), ) @dataclass(frozen=True) class GatewayResponse: handling: Handling code: str job_id: Optional[str] = None backend: Optional[str] = None status: Optional[str] = None reply_text: Optional[str] = None forward: Optional[Mapping[str, Any]] = None flags: tuple[str, ...] = () shadow: bool = False planned_handling: Optional[str] = None privacy: Optional[str] = None delivered: bool = False # Der Kern stellt nie selbst an Telegram zu. def to_dict(self) -> dict: return { "protocol": PROTOCOL_VERSION, "handling": self.handling.value, "code": self.code, "job_id": self.job_id, "backend": self.backend, "status": self.status, "reply_text": self.reply_text, "forward": self.forward, "flags": list(self.flags), "shadow": self.shadow, "planned_handling": self.planned_handling, "privacy": self.privacy, "delivered": False, } class MemoryProvider(Protocol): def snippets_for(self, request: GatewayRequest, backend: str) -> Sequence[MemorySnippet]: """Liefert Memory-Ausschnitte für das gewählte Backend (vertrauenswürdige lokale Quelle).""" _MISSING = object() def _iso(now: float) -> str: return datetime.fromtimestamp(now, tz=timezone.utc).isoformat(timespec="milliseconds") class Gateway: def __init__(self, config: GatewayConfig, *, clock: Callable[[], float] = time.time, memory_provider: Optional[MemoryProvider] = None, adapter_factories: Optional[Mapping[str, Any]] = None, forward_sink=None, codex_runner: Optional[IsolatedRunner] = None): if not isinstance(config, GatewayConfig): raise ConfigurationError("GatewayConfig required") self.config = config self._clock = clock state = config.state_dir privacy = config.privacy self._classifier = PrivacyClassifier( public_marker_enabled=privacy.public_marker_enabled, extra_confidential_keywords=privacy.extra_confidential_keywords, extra_internal_keywords=privacy.extra_internal_keywords, ) self._keyring = ClearanceKeyring(config.clearance.key_file) self._nonces = NonceStore(state / "nonces.json") self._issuer = ClearanceIssuer(self._keyring, privacy.matrix, ttl_seconds=config.clearance.ttl_seconds, clock=clock) self._verifier = ClearanceVerifier(self._keyring, privacy.matrix, self._nonces, self._classifier, max_clock_skew_seconds=config.clearance.max_clock_skew_seconds, clock=clock) self._dedupe = IdempotencyStore(state / "dedupe.json", config.dedupe_ttl_seconds) self._modes = ModeStore(state / "modes.json") self._shadow_dedupe = IdempotencyStore(state / "shadow-dedupe.json", config.dedupe_ttl_seconds) self._shadow_modes = ModeStore(state / "shadow-modes.json") self._budget = BudgetLedger(state / "budget.json", config.budget, clock=clock) self._health = HealthRegistry(state / "health.json", config.health.failure_threshold, config.health.open_seconds, clock=clock) self._log = DecisionLog(state / "decisions.jsonl", config.decision_log_retention_days * 86400, clock=clock) self._memory = memory_provider self._adapters = self._build_adapters(adapter_factories or {}, forward_sink, codex_runner) # ------------------------------------------------------------------ setup def _default_adapter(self, name: str, forward_sink, codex_runner) -> BackendAdapter: if name == "claude": return ClaudeChannelAdapter(self._verifier, sink=forward_sink) codex = self.config.codex runner = codex_runner if runner is None: if codex.real_execution: from .isolation.caged_runner import CagedCodexRunner runner = CagedCodexRunner( codex.workspace_root, profile=codex.profile, resource_limits=codex.resource_limits, auth_file=str(codex.auth_file), codex_auth_writable=codex.codex_auth_writable, home_root=str(codex.home_root), bus_env=dict(codex.user_bus) or None, state_dir=str(self.config.state_dir), protect_startup=codex.tools_protect_startup, write_allowlist=codex.tools_write_allowlist, ) else: runner = MockRunner() return CodexAdapter(self._verifier, runner=runner, command_template=codex.command_template, isolation_wrapper=codex.isolation_wrapper, env=dict(codex.env), max_output_bytes=codex.max_output_bytes, allow_sandbox_bypass=(codex.profile == "tools"), require_output_file=codex.real_execution) def _build_adapters(self, factories: Mapping[str, Any], forward_sink, codex_runner) -> dict: adapters = {} for settings in self.config.backends: if not settings.enabled: continue factory = factories.get(settings.name, _MISSING) if factory is None: continue # bewusst ohne Adapter -> Backend nicht verfügbar if factory is _MISSING: adapter = self._default_adapter(settings.name, forward_sink, codex_runner) else: adapter = factory(self._verifier) if not isinstance(adapter, BackendAdapter): raise ConfigurationError("INVALID_ADAPTER") if adapter.name != settings.name: raise ConfigurationError("ADAPTER_MISMATCH") if adapter.verifier is not self._verifier: raise ConfigurationError("ADAPTER_VERIFIER_MISMATCH") adapters[settings.name] = adapter return adapters @property def adapters(self) -> Mapping[str, BackendAdapter]: return dict(self._adapters) @property def decision_log(self) -> DecisionLog: return self._log # ------------------------------------------------------------------ public def handle(self, request: GatewayRequest) -> GatewayResponse: """Einziger Ausführungspfad.""" return self._process(request, execute=True) def shadow(self, request: GatewayRequest) -> GatewayResponse: """Entscheidet und protokolliert nur. Keine Ausführung, kein Budget, keine Clearance.""" return self._process(request, execute=False) def report_delivery(self, *, job_id: str, nonce: str, ok: bool) -> bool: """Rückmeldung des Plugins zur Claude-Zustellung. Nur für echte (verbrauchte) Nonces.""" if not isinstance(nonce, str) or not self._nonces.was_used(nonce): return False if ok: self._health.record_success("claude") else: self._health.record_failure("claude", "DELIVERY_FAILED") self._log.write({"ts": _iso(self._clock()), "phase": "delivery_report", "job_id": job_id, "backend": "claude", "ok": bool(ok)}) return True def log_rejected_request(self, code: str, *, shadow: bool) -> None: """Protokolliert eine schon an der Bridge abgewiesene Anfrage (ohne Inhalt).""" try: self._log.write({"ts": _iso(self._clock()), "phase": "decision", "shadow": shadow, "handling": "error", "code": code, "rejected_at": "bridge"}) except Exception: pass def status(self) -> dict: backends = {} for settings in self.config.backends: health = self._health.state(settings.name) backends[settings.name] = { "enabled": settings.enabled, "adapter": settings.name in self._adapters, "health": health.status, "consecutive_failures": health.failures, "last_error": health.last_error, "budget": self._budget.usage_today(settings.name), } return { "protocol": PROTOCOL_VERSION, "allowed_modes": sorted(self.config.allowed_modes), "key_available": self._keyring.available(), "codex_real_execution": self.config.codex.real_execution, "backends": backends, } # ----------------------------------------------------------------- intern def _load_allowlist(self) -> tuple[frozenset[str], dict]: auth = self.config.auth dm: Optional[frozenset[str]] = None groups: dict = {} if auth.telegram_access_file is not None: try: with open(auth.telegram_access_file, "r", encoding="utf-8") as handle: data = json.load(handle) except (OSError, ValueError) as exc: raise StateError("ALLOWLIST_UNAVAILABLE") from exc if not isinstance(data, dict): raise StateError("ALLOWLIST_UNAVAILABLE") if data.get("dmPolicy") == "disabled": dm, groups = frozenset(), {} else: allow = data.get("allowFrom", []) dm = frozenset(str(x) for x in allow) if isinstance(allow, list) else frozenset() raw_groups = data.get("groups", {}) if isinstance(raw_groups, dict): groups = {str(k): v for k, v in raw_groups.items() if isinstance(v, dict)} if auth.allowed_sender_ids is not None: dm = auth.allowed_sender_ids if dm is None else dm & auth.allowed_sender_ids return dm or frozenset(), groups def _authenticate(self, request: GatewayRequest) -> tuple[Optional[Principal], str]: if request.transport != "telegram": return None, "UNSUPPORTED_TRANSPORT" allowed, groups = self._load_allowlist() if request.chat_type == "private": if request.chat_id != request.sender_id: return None, "CHAT_SENDER_MISMATCH" if request.sender_id not in allowed: return None, "SENDER_NOT_ALLOWED" return Principal.OWNER, "ALLOWLISTED" if request.chat_type in ("group", "supergroup"): policy = groups.get(request.chat_id) if policy is None: return None, "GROUP_NOT_CONFIGURED" allow_from = policy.get("allowFrom") or [] if not isinstance(allow_from, list): return None, "GROUP_POLICY_INVALID" if allow_from and request.sender_id not in {str(x) for x in allow_from}: return None, "SENDER_NOT_ALLOWED_IN_GROUP" return Principal.GROUP_MEMBER, "GROUP_POLICY" return None, "UNSUPPORTED_CHAT_TYPE" def _backend_snapshot(self, principal: Principal, tokens: int) -> tuple[list[Backend], dict]: backends, states = [], {} for settings in self.config.backends: adapter = self._adapters.get(settings.name) health = self._health.state(settings.name) budget_ok = self._budget.available(settings.name, tokens) restricted = principal is Principal.GROUP_MEMBER and settings.name != "claude" available = (settings.enabled and adapter is not None and health.status != "open" and not restricted) states[settings.name] = { "enabled": settings.enabled, "adapter": adapter is not None, "health": health.status, "budget_available": budget_ok, "principal_restricted": restricted, } backends.append(Backend( settings.name, settings.capabilities, settings.tasks, available=available, budget_allowed=budget_ok, quality_score=settings.quality, max_context_bytes=settings.max_context_bytes, task_quality=settings.task_quality, )) return backends, states def _enrich(self, request: GatewayRequest, backend: str, envelope: ContextEnvelope, record: dict) -> ContextEnvelope: adapter = self._adapters[backend] if self._memory is None or not adapter.accepts_memory: record["memory"] = {"offered": 0, "used": 0} return envelope snippets = list(self._memory.snippets_for(request, backend)) settings = self.config.backend(backend) used, rejected = [], 0 for snippet in snippets: if not isinstance(snippet, MemorySnippet): raise TypeError("INVALID_MEMORY_SNIPPET") tier = self._classifier.classify(ContextEnvelope("", memory=(snippet,))).parts[1][1] if not self.config.privacy.matrix.permits(backend, tier): rejected += 1 continue candidate = ContextEnvelope(envelope.prompt, tuple(used + [snippet]), envelope.attachments) if settings.max_context_bytes is not None and candidate.size_bytes() > settings.max_context_bytes: rejected += 1 continue used.append(snippet) record["memory"] = {"offered": len(snippets), "used": len(used), "rejected": rejected} return ContextEnvelope(envelope.prompt, tuple(used), envelope.attachments) @staticmethod def _explain(decision: RoutingDecision, assessment: PrivacyAssessment, mode: Mode, matrix_tiers: Callable[[str], frozenset]) -> str: tier = assessment.tier.value rules = ", ".join(r for r in assessment.rule_ids if not r.startswith("attachment.")) or "keine Regel" if decision.code == MANUAL_CHOICE_NOT_PERMITTED: wanted = mode.value reasons = {r for c in decision.candidates if c.backend == wanted for r in c.reasons} lines = [f"Nicht gesendet: Modus /{wanted} ist gesetzt, aber {wanted} darf diesen Auftrag nicht bekommen."] if "PRIVACY_NOT_PERMITTED" in reasons: allowed = ", ".join(sorted(t.value for t in matrix_tiers(wanted))) or "nichts" lines.append(f"Datenschutz: Einstufung „{tier}“ ({rules}); {wanted} erhält nur: {allowed}.") if "UNAVAILABLE" in reasons: lines.append(f"{wanted} ist derzeit nicht verfügbar.") if "BUDGET_EXHAUSTED" in reasons: lines.append(f"Das Tagesbudget für {wanted} ist erschöpft.") if reasons & {"CAPABILITY_MISSING", "TASK_UNSUPPORTED"}: lines.append(f"{wanted} unterstützt diese Aufgabe nicht.") if "CONTEXT_TOO_LARGE" in reasons: lines.append("Der Kontext ist für dieses Modell zu groß.") if "QUALITY_BELOW_MINIMUM" in reasons: lines.append("Die Qualitätsuntergrenze für diese Aufgabe wird nicht erreicht.") lines.append("Die manuelle Wahl hebt keine Prüfung auf. Mit /auto oder /claude erneut senden.") return "\n".join(lines) if decision.code == PRIVACY_DENIED: return f"Nicht gesendet: Einstufung „{tier}“ ({rules}) ist für kein Modell freigegeben." reasons = sorted({r for c in decision.candidates for r in c.reasons}) return "Nicht gesendet: kein geeignetes Modell verfügbar (" + (", ".join(reasons) or "keine Backends") + ")." def _base_record(self, request: Any, execute: bool) -> dict: record: dict = {"ts": _iso(self._clock()), "phase": "decision", "shadow": not execute} if isinstance(request, GatewayRequest): record.update({ "transport": request.transport, "sender_id": request.sender_id, "chat_id": request.chat_id, "chat_type": request.chat_type, "message_id": request.message_id, "prompt_sha256": sha256_hex(request.text.encode("utf-8")), "prompt_chars": len(request.text), "attachment_count": len(request.attachments), }) return record def _respond(self, record: dict, response: GatewayResponse, started: float, *, must_log: bool) -> GatewayResponse: record["handling"] = response.handling.value record["code"] = response.code record["status"] = response.status record["flags"] = list(response.flags) if response.planned_handling: record["planned_handling"] = response.planned_handling record["duration_ms"] = round((time.monotonic() - started) * 1000, 3) try: self._log.write(record) except Exception: if must_log: return GatewayResponse(Handling.ERROR, "LOG_UNAVAILABLE", job_id=response.job_id, shadow=response.shadow) return GatewayResponse(**{**response.__dict__, "flags": response.flags + ("LOG_WRITE_FAILED",)}) return response def _process(self, request: Any, *, execute: bool) -> GatewayResponse: started = time.monotonic() record = self._base_record(request, execute) try: response, must_log = self._decide(request, execute, record, started) except Exception as exc: # fail-closed: nie ausführen, wenn etwas unerwartet scheitert record["exception"] = type(exc).__name__ code = exc.code if isinstance(exc, StateError) else "INTERNAL_ERROR" response, must_log = GatewayResponse(Handling.ERROR, code, shadow=not execute), False if response is None: # bereits protokolliert (Ausführungspfad) return record.pop("_final") return self._respond(record, response, started, must_log=must_log or not execute) def _shadow_wrap(self, response: GatewayResponse) -> GatewayResponse: return GatewayResponse( Handling.SHADOW_DECISION, response.code, job_id=response.job_id, backend=response.backend, status=None, reply_text=None, forward=None, flags=response.flags, shadow=True, planned_handling=response.handling.value, privacy=response.privacy, ) def _decide(self, request: Any, execute: bool, record: dict, started: float): mode_name = "execute" if execute else "shadow" if mode_name not in self.config.allowed_modes: return GatewayResponse(Handling.ERROR, "MODE_NOT_ENABLED", shadow=not execute), False if not isinstance(request, GatewayRequest): return GatewayResponse(Handling.ERROR, "INVALID_REQUEST", shadow=not execute), False def done(response: GatewayResponse): return (response if execute else self._shadow_wrap(response)), False principal, auth_reason = self._authenticate(request) record["authentication"] = {"principal": principal.value if principal else None, "reason": auth_reason} if principal is None: return done(GatewayResponse(Handling.DROP, "NOT_AUTHORIZED")) job_id = f"tg-{request.chat_id}-{request.message_id}" record["job_id"] = job_id dedupe = self._dedupe if execute else self._shadow_dedupe modes = self._modes if execute else self._shadow_modes claimed, original = dedupe.claim(f"{request.transport}:{request.chat_id}:{request.message_id}", job_id, self._clock()) if not claimed: record["duplicate_of"] = original return done(GatewayResponse(Handling.DROP, "DUPLICATE_MESSAGE", job_id=job_id)) lifecycle = JobLifecycle(job_id) owner = principal is Principal.OWNER parsed = parse_command(request.text) record["command"] = parsed.kind.value if parsed.kind is CommandKind.PLUGIN_OWNED: return done(GatewayResponse(Handling.NOT_INTERCEPTED, "PLUGIN_OWNED_COMMAND", job_id=job_id)) if parsed.kind in (CommandKind.SET_MODE, CommandKind.ONE_SHOT) and not owner: lifecycle.transition(JobStatus.BLOCKED) return done(GatewayResponse(Handling.BLOCKED, "MODE_COMMAND_NOT_ALLOWED", job_id=job_id, status=lifecycle.status.value)) if parsed.kind is CommandKind.SET_MODE: modes.set(request.chat_id, parsed.mode) record["manual_mode"] = parsed.mode.value hint = ("automatische Auswahl" if parsed.mode is Mode.AUTO else "Datenschutz-, Budget- und Verfügbarkeitsregeln gelten weiter") return done(GatewayResponse(Handling.REPLY, "MODE_SET", job_id=job_id, reply_text=f"Modellmodus für diesen Chat: {parsed.mode.value} ({hint}).")) if parsed.kind is CommandKind.ONE_SHOT: manual_mode, prompt = parsed.mode, parsed.text else: manual_mode = modes.get(request.chat_id) if owner else Mode.AUTO prompt = request.text record["manual_mode"] = manual_mode.value if not prompt.strip() and not request.attachments: lifecycle.transition(JobStatus.BLOCKED) return done(GatewayResponse(Handling.BLOCKED, "EMPTY_MESSAGE", job_id=job_id, status=lifecycle.status.value)) if len(prompt.encode("utf-8")) > self.config.limits.max_text_bytes: lifecycle.transition(JobStatus.BLOCKED) return done(GatewayResponse(Handling.BLOCKED, "TEXT_TOO_LARGE", job_id=job_id, status=lifecycle.status.value)) envelope = ContextEnvelope(prompt, (), request.attachments) assessment = self._classifier.classify(envelope) task = classify_task(envelope, self.config.routing.default_task_class) tokens = estimate_tokens(envelope.size_bytes()) record.update({ "privacy": assessment.tier.value, "privacy_rules": list(assessment.rule_ids), "privacy_parts": [{"part": name, "tier": tier.value} for name, tier in assessment.parts], "task_class": task.task_class, "task_rule": task.rule, "capabilities": sorted(task.capabilities), "context_bytes": envelope.size_bytes(), "token_estimate": tokens, }) backends, states = self._backend_snapshot(principal, tokens) record["backend_state"] = states job = JobRequest(job_id, task.task_class, assessment.tier, task.capabilities, manual_mode, context_bytes=envelope.size_bytes()) decision = route(job, backends, self.config.privacy.matrix, min_quality=self.config.routing.min_quality, manual_fallback=self.config.routing.manual_fallback == "checked_fallback") record["routing"] = { "code": decision.code, "selected": decision.backend, "tie_break": decision.tie_break, "manual": decision.manual, "fallback_from": decision.fallback_from, "candidates": [{"backend": c.backend, "eligible": c.eligible, "reasons": list(c.reasons), "quality": c.quality} for c in decision.candidates], } if not decision.allowed or decision.backend is None: lifecycle.transition(JobStatus.BLOCKED) reply = (self._explain(decision, assessment, manual_mode, self.config.privacy.matrix.tiers_for) if owner else None) return done(GatewayResponse(Handling.BLOCKED, decision.code, job_id=job_id, status=lifecycle.status.value, reply_text=reply, privacy=assessment.tier.value)) backend = decision.backend flags = ("CHECKED_FALLBACK",) if decision.code == CHECKED_FALLBACK else () try: final_envelope = self._enrich(request, backend, envelope, record) except Exception: lifecycle.transition(JobStatus.BLOCKED) return done(GatewayResponse(Handling.BLOCKED, "MEMORY_PROVIDER_ERROR", job_id=job_id, backend=backend, status=lifecycle.status.value)) final = self._classifier.classify(final_envelope) record["final_privacy"] = final.tier.value if not self.config.privacy.matrix.permits(backend, final.tier): lifecycle.transition(JobStatus.BLOCKED) return done(GatewayResponse(Handling.BLOCKED, "PRIVACY_CHANGED_BY_ENRICHMENT", job_id=job_id, backend=backend, status=lifecycle.status.value)) planned = (Handling.FORWARD_TO_CLAUDE_CHANNEL if backend == "claude" else Handling.REPLY) if not execute: return GatewayResponse( Handling.SHADOW_DECISION, decision.code, job_id=job_id, backend=backend, flags=flags, shadow=True, planned_handling=planned.value, privacy=final.tier.value, ), True return self._execute(request, record, started, lifecycle, job_id, backend, final_envelope, final, flags, owner, decision) def _execute(self, request, record, started, lifecycle, job_id, backend, envelope, assessment, flags, owner, decision): adapter = self._adapters[backend] policy = ExecutionPolicy(timeout_seconds=self.config.codex.timeout_seconds) if backend == "codex" \ else ExecutionPolicy() job = AdapterJob(job_id, envelope, request.sender_id, request.chat_id, request.message_id, request.sender_username, request.ts, policy) record["final_context_sha256"] = envelope.digest() def blocked(code: str, handling: Handling = Handling.BLOCKED): if lifecycle.status is JobStatus.CREATED: lifecycle.transition(JobStatus.BLOCKED) return GatewayResponse(handling, code, job_id=job_id, backend=backend, status=lifecycle.status.value, privacy=assessment.tier.value), True try: outbound_digest = adapter.outbound_digest(job) except AdapterInputError as exc: return blocked(exc.code) tokens = estimate_tokens(envelope.size_bytes()) if not self._budget.reserve(backend, job_id, tokens): return blocked("BUDGET_EXHAUSTED") try: clearance = self._issuer.issue(job_id=job_id, backend=backend, privacy=assessment.tier, context_digest=outbound_digest, sender_id=request.sender_id, chat_id=request.chat_id) except KeyUnavailableError as exc: self._budget.settle(job_id, consumed=False) return GatewayResponse(Handling.ERROR, exc.code, job_id=job_id, backend=backend), True except AuthorizationError as exc: self._budget.settle(job_id, consumed=False) return blocked(exc.code) except Exception: self._budget.settle(job_id, consumed=False) raise # Die Clearance selbst wird nie protokolliert (Wächter), nur Metadaten. record["authorization"] = {"clearance_issued": True, "expires_at": clearance.expires_at, "outbound_sha256": outbound_digest} lifecycle.transition(JobStatus.ROUTED) record["status"] = lifecycle.status.value try: self._log.write({**record, "handling": "pending", "code": "CLEARANCE_ISSUED"}) except Exception: self._budget.settle(job_id, consumed=False) lifecycle.transition(JobStatus.FAILED) return GatewayResponse(Handling.ERROR, "LOG_UNAVAILABLE", job_id=job_id, backend=backend, status=lifecycle.status.value), True lifecycle.transition(JobStatus.RUNNING) try: result = adapter.execute(clearance, job) except Exception: # execute() fängt selbst; das hier ist nur die letzte Linie result = AdapterResult(job_id, backend, ResultStatus.FAILED, error_code="ADAPTER_EXCEPTION") consumed = result.status is not ResultStatus.BLOCKED self._budget.settle(job_id, consumed=consumed) outcome_record = {k: v for k, v in record.items() if k in ( "ts", "shadow", "job_id", "transport", "sender_id", "chat_id", "chat_type", "message_id")} outcome_record["phase"] = "outcome" outcome_record["backend"] = backend outcome_record["adapter_status"] = result.status.value outcome_record["adapter_code"] = result.error_code if result.status is ResultStatus.SUCCESS: self._health.record_success(backend) checked = check_output(result.output, max_chars=self.config.limits.max_reply_chars) lifecycle.transition(JobStatus.COMPLETED) outcome_record["output_chars"] = len(result.output) outcome_record["output_sha256"] = sha256_hex(result.output.encode("utf-8")) outcome_record["redactions"] = list(checked.redactions) response = GatewayResponse(Handling.REPLY, "COMPLETED", job_id=job_id, backend=backend, status=lifecycle.status.value, reply_text=checked.text, flags=flags + checked.flags, privacy=assessment.tier.value) elif result.status is ResultStatus.FORWARD: lifecycle.transition(JobStatus.COMPLETED) notice = None if decision.code == CHECKED_FALLBACK and owner: notice = (f"Hinweis: {decision.fallback_from} war für diesen Auftrag nicht zulässig; " f"nach vollständiger Prüfung an {backend} weitergeleitet.") response = GatewayResponse(Handling.FORWARD_TO_CLAUDE_CHANNEL, "FORWARD_ISSUED", job_id=job_id, backend=backend, status=lifecycle.status.value, reply_text=notice, forward=result.forward, flags=flags, privacy=assessment.tier.value) elif result.status is ResultStatus.FAILED: counted = self._health.record_failure(backend, result.error_code or "") outcome_record["health_counted"] = counted lifecycle.transition(JobStatus.TIMED_OUT if result.timed_out else JobStatus.FAILED) reply = (f"Auftrag an {backend} fehlgeschlagen ({result.error_code}). " "Es gibt keinen automatischen Wechsel auf ein anderes Modell.") if owner else None response = GatewayResponse(Handling.FAILED, result.error_code or "FAILED", job_id=job_id, backend=backend, status=lifecycle.status.value, reply_text=reply, flags=flags, privacy=assessment.tier.value) else: lifecycle.transition(JobStatus.FAILED) response = GatewayResponse(Handling.BLOCKED, result.error_code or "ADAPTER_BLOCKED", job_id=job_id, backend=backend, status=lifecycle.status.value, flags=flags, privacy=assessment.tier.value) outcome_record["duration_ms"] = round((time.monotonic() - started) * 1000, 3) outcome_record["handling"] = response.handling.value outcome_record["code"] = response.code outcome_record["status"] = response.status outcome_record["flags"] = list(response.flags) try: self._log.write(outcome_record) except Exception: response = GatewayResponse(**{**response.__dict__, "flags": response.flags + ("LOG_WRITE_FAILED",)}) record["_final"] = response return None, True