toki
0be1a3dcce
Epic 3 준비 전에 caller별 IOP direct preflight와 attempt recovery의 검증된 완료 상태를 원격 checkpoint로 보존한다.
508 lines
22 KiB
Python
508 lines
22 KiB
Python
"""Explicit, secret-safe live IOP boundary for benchmark callers.
|
|
|
|
Only the six ``IOP_BENCH_<CALLER>_{BASE_URL,SECRET_ENV}`` values select this
|
|
boundary. The referenced secret is retained in a private runtime object and
|
|
is never included in connectivity evidence, command output, or run metadata.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from collections.abc import Callable, Mapping
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from scripts.agent_benchmark.agy_iop import (
|
|
AGY_CALLER,
|
|
AgyEventParser,
|
|
AgyRuntimeInputs,
|
|
AgyRuntimeObservation,
|
|
build_agy_invocation,
|
|
inspect_agy_iop_capability,
|
|
preflight_agy_iop,
|
|
run_agy_invocation,
|
|
_runtime_identity as _agy_runtime_identity,
|
|
)
|
|
from scripts.agent_benchmark.attempts import Attempt, ExecutionAdapter, PreflightObservation
|
|
from scripts.agent_benchmark.claude_iop import ClaudeIopAdapter, ClaudeIopRuntime, claude_capability
|
|
from scripts.agent_benchmark.codex_iop import (
|
|
BASE_URL_ENV_KEY,
|
|
SECRET_ENV_KEY,
|
|
build_codex_invocation,
|
|
codex_capability,
|
|
run_codex_invocation,
|
|
runtime_from_environment,
|
|
)
|
|
from scripts.agent_benchmark.connectivity import (
|
|
ISSUE_CODE_ORDER,
|
|
ISSUE_RESUME_CODES,
|
|
CallerCapability,
|
|
ConnectivityIssue,
|
|
EffectiveBinding,
|
|
RequestedEffectiveBinding,
|
|
make_result,
|
|
)
|
|
from scripts.agent_benchmark.lifecycle import InvocationResult, run_invocation, spec_digest
|
|
from scripts.agent_benchmark.manifest import CALLER_ENUM, MatrixCell, TOKEN_RE, Timeout
|
|
from scripts.agent_benchmark.workspace import PreparedWorkspace
|
|
|
|
|
|
_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")
|
|
_CALLERS = ("claude", "agy", "codex")
|
|
_TIMEOUT_SECONDS = 10
|
|
|
|
|
|
class LiveIopError(Exception):
|
|
"""Private boundary failure converted into one closed public issue."""
|
|
|
|
def __init__(self, issue_code: str) -> None:
|
|
if issue_code not in ISSUE_RESUME_CODES:
|
|
raise ValueError("invalid live IOP issue")
|
|
self.issue_code = issue_code
|
|
super().__init__(issue_code)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Runtime:
|
|
caller: str
|
|
base_url: str
|
|
secret: str
|
|
endpoint_identity: str
|
|
config: "_ConfigObservation"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _RouteObservation:
|
|
route_kind: str
|
|
route_id: str
|
|
model: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _ConfigObservation:
|
|
routes: tuple[_RouteObservation, ...]
|
|
identity: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Observation:
|
|
catalog_models: tuple[str, ...]
|
|
config_identity: str
|
|
caller_ready: bool
|
|
agy_version: str = ""
|
|
agy_help: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _RuntimeResolution:
|
|
runtime: _Runtime | None
|
|
issue_code: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _InvokerSeams:
|
|
"""Test seams below registry construction; production uses real callers."""
|
|
|
|
claude: Callable[..., InvocationResult]
|
|
agy: Callable[..., InvocationResult]
|
|
codex: Callable[..., Any]
|
|
|
|
|
|
def _identity(label: str, value: str) -> str:
|
|
return "sha256:" + hashlib.sha256(
|
|
b"iop-benchmark-live-v1\0" + label.encode("ascii") + b"\0" + value.encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def _base_url(value: str) -> str:
|
|
if not isinstance(value, str) or not value:
|
|
raise LiveIopError("endpoint_incompatible")
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme not in ("http", "https") or not parsed.netloc or parsed.query or parsed.fragment:
|
|
raise LiveIopError("endpoint_incompatible")
|
|
return value.rstrip("/")
|
|
|
|
|
|
def _models_url(base_url: str) -> str:
|
|
parsed = urlsplit(base_url)
|
|
path = parsed.path.rstrip("/")
|
|
if path.endswith("/v1"):
|
|
path += "/models"
|
|
else:
|
|
path += "/v1/models"
|
|
return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
|
|
|
|
|
|
def _command(argv: tuple[str, ...]) -> str:
|
|
try:
|
|
completed = subprocess.run(
|
|
argv, check=False, capture_output=True, text=True, timeout=_TIMEOUT_SECONDS,
|
|
env={"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"},
|
|
)
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
raise LiveIopError("stream_incompatible") from exc
|
|
if completed.returncode != 0:
|
|
raise LiveIopError("stream_incompatible")
|
|
return completed.stdout + completed.stderr
|
|
|
|
|
|
def _caller_binary(name: str) -> str:
|
|
path = shutil.which(name)
|
|
if not path or not os.path.isfile(path) or not os.access(path, os.X_OK):
|
|
raise LiveIopError("stream_incompatible")
|
|
return os.path.realpath(path)
|
|
|
|
|
|
def _catalog(runtime: _Runtime) -> tuple[tuple[str, ...], str]:
|
|
request = Request(_models_url(runtime.base_url), headers={"Authorization": f"Bearer {runtime.secret}"})
|
|
try:
|
|
with urlopen(request, timeout=_TIMEOUT_SECONDS) as response:
|
|
if response.status in (401, 403):
|
|
raise LiveIopError("auth_incompatible")
|
|
if response.status != 200:
|
|
raise LiveIopError("endpoint_incompatible")
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
except HTTPError as exc:
|
|
if exc.code in (401, 403):
|
|
raise LiveIopError("auth_incompatible") from exc
|
|
raise LiveIopError("endpoint_incompatible") from exc
|
|
except LiveIopError:
|
|
raise
|
|
except OSError as exc:
|
|
# Connection refusal, DNS failure, and timeout all mean that this
|
|
# boundary cannot reach a compatible endpoint. They are distinct
|
|
# from a reachable endpoint with an invalid response schema.
|
|
raise LiveIopError("endpoint_incompatible") from exc
|
|
except ValueError as exc:
|
|
raise LiveIopError("protocol_incompatible") from exc
|
|
records = payload.get("data") if isinstance(payload, dict) else None
|
|
if not isinstance(records, list):
|
|
raise LiveIopError("protocol_incompatible")
|
|
model_ids: list[str] = []
|
|
for item in records:
|
|
model_id = item.get("id") if isinstance(item, dict) else None
|
|
if not isinstance(model_id, str) or not model_id.strip():
|
|
raise LiveIopError("protocol_incompatible")
|
|
model_ids.append(model_id)
|
|
models = tuple(sorted(model_ids))
|
|
if not models or len(set(models)) != len(models):
|
|
raise LiveIopError("protocol_incompatible")
|
|
return models, _identity("catalog", "\n".join(models))
|
|
|
|
|
|
def _observe(runtime: _Runtime) -> _Observation:
|
|
models, config_identity = _catalog(runtime)
|
|
if runtime.caller == "claude":
|
|
_command(("claude", "--version"))
|
|
_command(("claude", "--help"))
|
|
return _Observation(models, config_identity, True)
|
|
if runtime.caller == AGY_CALLER:
|
|
version = _command(("agy", "--version"))
|
|
help_output = _command(("agy", "--help"))
|
|
return _Observation(models, config_identity, True, version, help_output)
|
|
if runtime.caller == "codex":
|
|
_command(("codex", "--version"))
|
|
_command(("codex", "exec", "--help"))
|
|
return _Observation(models, config_identity, True)
|
|
raise LiveIopError("protocol_incompatible")
|
|
|
|
|
|
def _config_from_environment(environment: Mapping[str, str]) -> _ConfigObservation:
|
|
reference = environment.get("IOP_BENCH_CONFIG_OBSERVATION_ENV")
|
|
if not isinstance(reference, str) or not _ENV_NAME.fullmatch(reference):
|
|
raise LiveIopError("route_missing")
|
|
raw = environment.get(reference)
|
|
if not isinstance(raw, str) or not raw:
|
|
raise LiveIopError("route_missing")
|
|
try:
|
|
value = json.loads(raw)
|
|
except (TypeError, json.JSONDecodeError) as exc:
|
|
raise LiveIopError("protocol_incompatible") from exc
|
|
if not isinstance(value, dict) or set(value) != {"schema_version", "routes"} or value.get("schema_version") != "1":
|
|
raise LiveIopError("protocol_incompatible")
|
|
raw_routes = value.get("routes")
|
|
if not isinstance(raw_routes, list) or not raw_routes:
|
|
raise LiveIopError("route_missing")
|
|
routes: list[_RouteObservation] = []
|
|
for item in raw_routes:
|
|
if not isinstance(item, dict) or set(item) != {"route_kind", "route_id", "model"}:
|
|
raise LiveIopError("protocol_incompatible")
|
|
route_kind, route_id, model = (item.get(name) for name in ("route_kind", "route_id", "model"))
|
|
if route_kind != "direct" or not all(isinstance(field, str) and TOKEN_RE.fullmatch(field) for field in (route_id, model)):
|
|
raise LiveIopError("protocol_incompatible")
|
|
routes.append(_RouteObservation(route_kind, route_id, model))
|
|
if len({(item.route_kind, item.route_id) for item in routes}) != len(routes):
|
|
raise LiveIopError("protocol_incompatible")
|
|
routes.sort(key=lambda item: (item.route_kind, item.route_id, item.model))
|
|
canonical = json.dumps({"schema_version": "1", "routes": [item.__dict__ for item in routes]}, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
return _ConfigObservation(tuple(routes), _identity("config", canonical))
|
|
|
|
|
|
def _runtime_from_environment(caller: str, environment: Mapping[str, str]) -> _RuntimeResolution:
|
|
prefix = f"IOP_BENCH_{caller.upper()}_"
|
|
base_key = prefix + "BASE_URL"
|
|
secret_ref_key = prefix + "SECRET_ENV"
|
|
base_url = environment.get(base_key)
|
|
secret_ref = environment.get(secret_ref_key)
|
|
try:
|
|
normalized = _base_url(base_url)
|
|
except LiveIopError as exc:
|
|
return _RuntimeResolution(None, exc.issue_code)
|
|
if not isinstance(secret_ref, str) or not _ENV_NAME.fullmatch(secret_ref):
|
|
return _RuntimeResolution(None, "credential_missing")
|
|
secret = environment.get(secret_ref)
|
|
if not isinstance(secret, str) or not secret:
|
|
return _RuntimeResolution(None, "credential_missing")
|
|
try:
|
|
config = _config_from_environment(environment)
|
|
except LiveIopError as exc:
|
|
return _RuntimeResolution(None, exc.issue_code)
|
|
return _RuntimeResolution(_Runtime(caller, normalized, secret, _identity("endpoint", normalized), config), None)
|
|
|
|
|
|
def _requested(cell: MatrixCell) -> RequestedEffectiveBinding:
|
|
return RequestedEffectiveBinding(
|
|
cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
|
|
cell.iop.request_model, cell.iop.requested_effort,
|
|
)
|
|
|
|
|
|
def _issues(*codes: str) -> tuple[ConnectivityIssue, ...]:
|
|
selected = set(codes)
|
|
return tuple(ConnectivityIssue(code, ISSUE_RESUME_CODES[code]) for code in ISSUE_CODE_ORDER if code in selected)
|
|
|
|
|
|
def _binding_from_config(cell: MatrixCell, capability: CallerCapability, config: _ConfigObservation) -> tuple[RequestedEffectiveBinding, tuple[ConnectivityIssue, ...]]:
|
|
"""Admit a direct cell from config ownership and caller capability only."""
|
|
requested = _requested(cell)
|
|
if cell.iop.route_kind not in capability.route_kinds:
|
|
return requested, _issues("protocol_incompatible")
|
|
if cell.iop.requested_effort not in capability.efforts:
|
|
return requested, _issues("effort_unsupported")
|
|
route = next((item for item in config.routes if (item.route_kind, item.route_id) == (cell.iop.route_kind, cell.iop.route_id)), None)
|
|
if route is None:
|
|
return requested, _issues("route_missing")
|
|
if route.model != cell.iop.request_model:
|
|
return requested, _issues("model_missing")
|
|
return RequestedEffectiveBinding(
|
|
cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
|
|
cell.iop.request_model, cell.iop.requested_effort,
|
|
route.route_kind, route.route_id, route.model, cell.iop.requested_effort,
|
|
(EffectiveBinding("request", route.model, cell.iop.requested_effort),),
|
|
), ()
|
|
|
|
|
|
def _default_claude_invoker(adapter: ClaudeIopAdapter, spec: Any, **kwargs: Any) -> InvocationResult:
|
|
return run_invocation(spec, **kwargs)
|
|
|
|
|
|
def _default_agy_invoker(spec: Any, parser: AgyEventParser, preflight: Any, on_started: Callable[..., None]) -> InvocationResult:
|
|
return run_agy_invocation(spec, parser, preflight, on_started)
|
|
|
|
|
|
def _default_codex_invoker(invocation: Any, on_started: Callable[..., None]) -> Any:
|
|
return run_codex_invocation(invocation, on_started)
|
|
|
|
|
|
_DEFAULT_INVOKERS = _InvokerSeams(_default_claude_invoker, _default_agy_invoker, _default_codex_invoker)
|
|
|
|
|
|
def _bind_live_spec(
|
|
cell: MatrixCell,
|
|
prepared: PreparedWorkspace,
|
|
attempt: Attempt,
|
|
control_dir: str,
|
|
spec: Any,
|
|
) -> Any:
|
|
"""Immutably bind one caller spec to the controller-owned attempt paths."""
|
|
if (
|
|
not isinstance(prepared, PreparedWorkspace)
|
|
or not isinstance(attempt, Attempt)
|
|
or cell.id != attempt.identity.cell_id
|
|
or prepared.identity != attempt.identity
|
|
or not isinstance(control_dir, str)
|
|
or not control_dir
|
|
):
|
|
raise LiveIopError("stream_incompatible")
|
|
try:
|
|
attempt_root = Path(attempt.root).resolve(strict=True)
|
|
prepared_root = Path(prepared.attempt_root).resolve(strict=True)
|
|
workspace_root = Path(prepared.workspace_dir).resolve(strict=True)
|
|
session_root = Path(prepared.session_dir).resolve(strict=True)
|
|
evidence_root = Path(spec.evidence_dir).resolve(strict=True)
|
|
control_path = Path(control_dir)
|
|
alias = control_path.parent
|
|
if (
|
|
not control_path.is_absolute()
|
|
or control_path.name != "control"
|
|
or control_path.exists()
|
|
or control_path.is_symlink()
|
|
or not alias.is_symlink()
|
|
or alias.resolve(strict=True) != attempt_root
|
|
or control_path.resolve(strict=False) != attempt_root / "control"
|
|
):
|
|
raise LiveIopError("stream_incompatible")
|
|
except (OSError, RuntimeError, TypeError, ValueError) as exc:
|
|
raise LiveIopError("stream_incompatible") from exc
|
|
if (
|
|
prepared_root != attempt_root
|
|
or evidence_root != attempt_root
|
|
or workspace_root != attempt_root / "workspace"
|
|
or session_root != attempt_root / "session"
|
|
):
|
|
raise LiveIopError("stream_incompatible")
|
|
return replace(spec, control_dir=control_dir)
|
|
|
|
|
|
class _LiveAdapter:
|
|
"""One caller's live observation and invocation boundary."""
|
|
|
|
def __init__(
|
|
self, caller: str, capability: CallerCapability, runtime: _RuntimeResolution,
|
|
observer: Callable[[_Runtime], _Observation] = _observe,
|
|
binary_resolver: Callable[[str], str] = _caller_binary,
|
|
invokers: _InvokerSeams = _DEFAULT_INVOKERS,
|
|
) -> None:
|
|
self.caller = caller
|
|
self.capability = capability
|
|
self._runtime_resolution = runtime
|
|
self._observer = observer
|
|
self._binary_resolver = binary_resolver
|
|
self._invokers = invokers
|
|
self._agy_preflight: Any = None
|
|
self._admitted_bindings: dict[str, RequestedEffectiveBinding] = {}
|
|
|
|
def preflight(self, cell: MatrixCell) -> PreflightObservation:
|
|
if cell.caller != self.caller:
|
|
raise LiveIopError("protocol_incompatible")
|
|
resolution = self._runtime_resolution
|
|
runtime = resolution.runtime
|
|
if runtime is None:
|
|
result = make_result(cell, self.capability, _requested(cell), _issues(resolution.issue_code or "credential_missing"))
|
|
return PreflightObservation(result, _identity("missing", self.caller), _identity("missing-config", self.caller))
|
|
try:
|
|
observed = self._observer(runtime)
|
|
except LiveIopError as exc:
|
|
result = make_result(cell, self.capability, _requested(cell), _issues(exc.issue_code))
|
|
return PreflightObservation(result, runtime.endpoint_identity, _identity("unobserved-config", self.caller))
|
|
if not observed.caller_ready:
|
|
result = make_result(cell, self.capability, _requested(cell), _issues("stream_incompatible"))
|
|
return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
|
|
if self.caller == AGY_CALLER:
|
|
capability = inspect_agy_iop_capability(observed.agy_version, observed.agy_help)
|
|
agy_observation = AgyRuntimeObservation(
|
|
cell.id, cell.iop.route_kind, cell.iop.route_id,
|
|
_agy_runtime_identity("endpoint", runtime.base_url),
|
|
_agy_runtime_identity("credential", runtime.secret), runtime.config.identity,
|
|
)
|
|
try:
|
|
self._agy_preflight = preflight_agy_iop(
|
|
cell, capability,
|
|
AgyRuntimeInputs(self._binary_resolver("agy"), runtime.base_url, runtime.secret),
|
|
agy_observation,
|
|
)
|
|
except Exception:
|
|
result = make_result(cell, self.capability, _requested(cell), _issues("stream_incompatible"))
|
|
return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
|
|
if self._agy_preflight.issues:
|
|
result = make_result(cell, self.capability, _requested(cell), self._agy_preflight.issues)
|
|
return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
|
|
binding, issues = _binding_from_config(cell, self.capability, runtime.config)
|
|
if not issues and binding.effective_model not in observed.catalog_models:
|
|
binding, issues = _requested(cell), _issues("model_missing")
|
|
result = make_result(cell, self.capability, binding, issues)
|
|
if result.status == "ready":
|
|
self._admitted_bindings[cell.id] = result.binding
|
|
return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
|
|
|
|
def invoke(
|
|
self,
|
|
cell: MatrixCell,
|
|
prepared: PreparedWorkspace,
|
|
attempt: Attempt,
|
|
control_dir: str,
|
|
task_payload: bytes,
|
|
timeout: Timeout,
|
|
on_started: Callable[..., None],
|
|
) -> InvocationResult:
|
|
if cell.caller != self.caller:
|
|
raise LiveIopError("protocol_incompatible")
|
|
runtime = self._runtime_resolution.runtime
|
|
if runtime is None:
|
|
raise LiveIopError(self._runtime_resolution.issue_code or "protocol_incompatible")
|
|
admitted = self._admitted_bindings.get(cell.id)
|
|
if admitted is None:
|
|
raise LiveIopError("stream_incompatible")
|
|
if self.caller == "claude":
|
|
adapter = ClaudeIopAdapter(cell, prepared, ClaudeIopRuntime("claude", runtime.base_url, runtime.secret))
|
|
spec = _bind_live_spec(
|
|
cell,
|
|
prepared,
|
|
attempt,
|
|
control_dir,
|
|
adapter.invocation(task_payload.decode("utf-8"), attempt.root, timeout),
|
|
)
|
|
return self._invokers.claude(adapter, spec, parse_event=adapter.parser(), redact=adapter.redactor(task_payload.decode("utf-8")), on_started=lambda locator: on_started(locator, spec_digest(spec)))
|
|
if self.caller == AGY_CALLER:
|
|
if self._agy_preflight is None:
|
|
raise LiveIopError("stream_incompatible")
|
|
spec = _bind_live_spec(
|
|
cell,
|
|
prepared,
|
|
attempt,
|
|
control_dir,
|
|
build_agy_invocation(
|
|
cell, prepared, task_payload, timeout, self._agy_preflight
|
|
),
|
|
)
|
|
parser = AgyEventParser(cell)
|
|
result = self._invokers.agy(spec, parser, self._agy_preflight, lambda locator: on_started(locator, spec_digest(spec)))
|
|
observed = parser.observed_result(self._agy_preflight.capability, result)
|
|
if observed.status != "ready" or observed.binding != admitted:
|
|
raise LiveIopError("stream_incompatible")
|
|
return result
|
|
if self.caller == "codex":
|
|
invocation = build_codex_invocation(cell, prepared, runtime_from_environment({BASE_URL_ENV_KEY: runtime.base_url, SECRET_ENV_KEY: runtime.secret, "PATH": os.environ.get("PATH", "/usr/bin:/bin")}), task_payload, timeout)
|
|
invocation = replace(
|
|
invocation,
|
|
spec=_bind_live_spec(
|
|
cell,
|
|
prepared,
|
|
attempt,
|
|
control_dir,
|
|
invocation.spec,
|
|
),
|
|
)
|
|
result = self._invokers.codex(invocation, lambda locator: on_started(locator, spec_digest(invocation.spec)))
|
|
expected = (admitted.effective_route_kind, admitted.effective_route_id, admitted.effective_model, admitted.effective_effort)
|
|
if result.effective_binding != expected:
|
|
raise LiveIopError("stream_incompatible")
|
|
return result.lifecycle
|
|
raise LiveIopError("protocol_incompatible")
|
|
|
|
|
|
def build_live_adapter_registry(
|
|
environment: Mapping[str, str], *, observer: Callable[[_Runtime], _Observation] = _observe,
|
|
binary_resolver: Callable[[str], str] = _caller_binary,
|
|
invokers: _InvokerSeams = _DEFAULT_INVOKERS,
|
|
) -> dict[str, ExecutionAdapter]:
|
|
"""Build the fixed caller registry without reading ambient caller settings."""
|
|
if not isinstance(environment, Mapping):
|
|
raise LiveIopError("protocol_incompatible")
|
|
registry: dict[str, ExecutionAdapter] = {
|
|
"claude": _LiveAdapter("claude", claude_capability(), _runtime_from_environment("claude", environment), observer, binary_resolver, invokers),
|
|
"agy": _LiveAdapter(AGY_CALLER, CallerCapability(AGY_CALLER, ("direct", "execution_preset"), ("high", "low", "medium")), _runtime_from_environment("agy", environment), observer, binary_resolver, invokers),
|
|
"codex": _LiveAdapter("codex", codex_capability(), _runtime_from_environment("codex", environment), observer, binary_resolver, invokers),
|
|
}
|
|
if tuple(registry) != CALLER_ENUM:
|
|
raise LiveIopError("protocol_incompatible")
|
|
return registry
|