interactive_chat_controls.py
Rama main · réplica de lectura del repositorio original.
"""Explicit one-chat controls; no server imports, HTTP proxy, or authority lookup.
The router authenticates the share and supplies its chat identity. The owner
injects native models/callbacks and projects their results for the guest. Native
exceptions are deliberately not converted to safe denials: a callback may have
already committed work. Only ChatControlError is known to precede invocation.
"""
from __future__ import annotations
import json
import re
from types import MappingProxyType
SETTINGS_FIELDS = frozenset({
"title", "backend", "model", "effort", "system_prompt",
"claude_permission_mode", "codex_approval_policy", "codex_sandbox_mode",
"codex_permission_profile", "codex_approvals_reviewer", "cursor_permission_mode",
"provider_jobs_access",
})
GOAL_FIELDS = frozenset({"objective", "status", "token_budget", "time_budget_seconds"})
JOB_FIELDS = frozenset({
"title", "prompt", "schedule_kind", "interval_seconds", "cron_expression",
"rrule", "timezone", "loop", "max_runs", "enabled", "backend",
})
ACTIONS = frozenset({
"state", "turn.stop", "turn.steer", "queue.run_now", "queue.edit", "queue.delete",
"queue.move", "settings.update", "goal.set", "goal.resume", "goal.pause", "goal.delete",
"job.create", "job.update", "job.delete", "job.toggle", "job.run", "approval.respond",
})
class ChatControlError(ValueError):
"""A bounded rejection before any native callback or mutation."""
def __init__(self, code="invalid_request"):
self.code = code
super().__init__("This chat control request is not permitted" if code == "forbidden"
else "Invalid chat control request")
def _identity(value):
if not isinstance(value, str) or re.fullmatch(r"[A-Za-z0-9_.-]{1,128}", value) is None:
raise ChatControlError()
return value
def _fields(payload, allowed=(), required=(), *, nonempty=False):
if set(payload) - set(allowed) or not set(required).issubset(payload) or nonempty and not payload:
raise ChatControlError()
class InteractiveChatControls:
"""Dispatch to fixed native functions with a server-bound first argument.
Callback keys normally equal action names; approval uses approval.codex or
approval.claude. Native models are keyed by their class names. turn.steer is
the owner's enqueue/exact-promotion wrapper; job.run is the owner's atomic
expected-session wrapper, never the global unscoped run_job endpoint.
"""
def __init__(self, callbacks, models):
self.callbacks = MappingProxyType(dict(callbacks))
self.models = MappingProxyType(dict(models))
def _model(self, name, fields):
model = self.models.get(name)
if model is None:
raise ChatControlError("forbidden")
try:
return model.model_validate(fields)
except ValueError:
# Never expose native validation input, paths or object reprs.
raise ChatControlError() from None
async def dispatch(self, session_id, action, payload, *, share_id=None, request_id=None):
_identity(session_id)
if not isinstance(action, str) or action not in ACTIONS:
raise ChatControlError("forbidden")
if not isinstance(payload, dict) or any(not isinstance(key, str) for key in payload):
raise ChatControlError()
try:
encoded = json.dumps(payload, ensure_ascii=False, allow_nan=False).encode("utf-8")
except (TypeError, ValueError, UnicodeError, RecursionError):
raise ChatControlError() from None
if len(encoded) > 64 * 1024:
raise ChatControlError()
payload = json.loads(encoded) # Callback sees an owned immutable-in-transit value.
callback_key, args, kwargs = action, [session_id], {}
if action in {"state", "turn.stop", "goal.delete", "goal.resume", "goal.pause"}:
_fields(payload)
if action in {"goal.resume", "goal.pause"}:
args.append(self._model("CodexGoalRequest", {"status": "active" if action == "goal.resume" else "paused"}))
elif action == "turn.steer":
_fields(payload, {"prompt"}, {"prompt"})
if not isinstance(payload["prompt"], str) or not payload["prompt"].strip():
raise ChatControlError()
if not isinstance(share_id, str) or re.fullmatch(r"interactive_[a-f0-9]{32}", share_id) is None:
raise ChatControlError("forbidden")
if not isinstance(request_id, str) or re.fullmatch(r"[A-Za-z0-9_-]{16,128}", request_id) is None:
raise ChatControlError()
args.append(payload["prompt"])
kwargs = {"share_id": share_id, "request_id": request_id}
elif action in {"queue.delete", "queue.run_now", "job.delete", "job.run"}:
_fields(payload, {"id"}, {"id"})
args.append(_identity(payload["id"]))
if action == "queue.run_now":
args.append(self._model("RunQueuedTurnNowRequest", {"accept_deferred_queue_response": True}))
elif action == "queue.edit":
_fields(payload, {"id", "prompt", "expected_message_revision"}, {"id", "prompt"})
args.append(_identity(payload.pop("id")))
if not isinstance(payload["prompt"], str):
raise ChatControlError()
args.append(self._model("UpdateQueuedTurnRequest", payload))
elif action == "queue.move":
_fields(payload, {"id", "direction", "expected_adjacent_queued_id"}, {"id", "direction"})
args.append(_identity(payload.pop("id")))
if not isinstance(payload["direction"], str) or payload["direction"] not in {"up", "down"}:
raise ChatControlError()
if "expected_adjacent_queued_id" in payload:
_identity(payload["expected_adjacent_queued_id"])
args.append(self._model("MoveQueuedTurnRequest", payload))
elif action == "settings.update":
_fields(payload, SETTINGS_FIELDS, nonempty=True)
args.append(self._model("UpdateSessionRequest", payload))
elif action == "goal.set":
_fields(payload, GOAL_FIELDS, nonempty=True)
args.append(self._model("CodexGoalRequest", payload))
elif action == "job.create":
_fields(payload, JOB_FIELDS | {"first_run_at"}, {"title", "prompt"})
args.append(self._model("CreateScopedJobRequest", {**payload, "context_mode": "chat"}))
elif action in {"job.update", "job.toggle"}:
allowed = {"id", "enabled"} if action == "job.toggle" else JOB_FIELDS | {"id", "next_run_at"}
_fields(payload, allowed, {"id", "enabled"} if action == "job.toggle" else {"id"})
args.append(_identity(payload.pop("id")))
if not payload or action == "job.toggle" and type(payload["enabled"]) is not bool:
raise ChatControlError()
args.append(self._model("UpdateJobRequest", payload))
elif action == "approval.respond":
_fields(payload, {"id", "backend", "response"}, {"id", "backend", "response"})
backend = payload["backend"]
if backend not in ("codex", "claude") or not isinstance(payload["response"], dict):
raise ChatControlError()
args.append(_identity(payload["id"]))
args.append(self._model("CodexInteractionResponseRequest" if backend == "codex" else "ClaudeInteractionResponseRequest",
{"response": payload["response"]}))
callback_key = "approval." + backend
callback = self.callbacks.get(callback_key)
if not callable(callback):
raise ChatControlError("forbidden")
return await callback(*args, **kwargs)