"""Codex-Adapter (Schnittstelle aus Phase D). Ausführung über ein austauschbares IsolatedRunner-Objekt. Default ist der MockRunner; der echte SubprocessRunner wird nur gewählt, wenn die Konfiguration codex.real_execution=true setzt UND ein Isolations-Wrapper konfiguriert ist (config.py). Das Befehls-Template kommt aus der Konfiguration und wird beim Laden, beim Erzeugen des Adapters und unmittelbar vor jedem Lauf geprüft: danger-full-access, --dangerously-bypass-approvals-and-sandbox und --yolo werden hart abgelehnt; erlaubt ist nur der Platzhalter {workdir}; der Prompt geht nie in die Argumente, sondern über stdin. """ from __future__ import annotations import os import re from typing import Mapping, Optional, Sequence from .adapter_contract import ( AdapterInputError, AdapterJob, AdapterResult, BackendAdapter, Outbound, ResultStatus, ) from .context import ContextEnvelope, sha256_hex from .hardened_codex import MAX_OUTPUT_BYTES, decode_events from .runner import ( IsolatedRunner, RunResult, RunSpec, UnsafeCommandError, assert_safe_argv, attachment_file_name, validate_env, ) DEFAULT_COMMAND_TEMPLATE = ( "/usr/bin/codex", "exec", "--sandbox", "read-only", "--skip-git-repo-check", "--ephemeral", "--json", "-C", "{workdir}", "-", ) MAX_PROMPT_BYTES = 128 * 1024 _PLACEHOLDER = re.compile(r"\{[^{}]*\}") ALLOWED_PLACEHOLDERS = frozenset({"{workdir}", "{unit}", "{cfg}"}) def validate_command_template(template: Sequence[str], wrapper: Sequence[str] = (), *, allow_bypass_sandbox: bool = False, require_cage: bool = False) -> tuple[str, ...]: if not isinstance(template, (list, tuple)) or not isinstance(wrapper, (list, tuple)): raise UnsafeCommandError("INVALID_COMMAND") if not template: raise UnsafeCommandError("INVALID_COMMAND") full = tuple(wrapper) + tuple(template) assert_safe_argv(full, allow_bypass_sandbox=allow_bypass_sandbox) if not isinstance(template[0], str) or not os.path.isabs(template[0]): raise UnsafeCommandError("COMMAND_NOT_ABSOLUTE") for arg in full: for placeholder in _PLACEHOLDER.findall(arg): if placeholder not in ALLOWED_PLACEHOLDERS: raise UnsafeCommandError("UNKNOWN_PLACEHOLDER") if require_cage: # tools-Profil: die Bypass-Tokens duerfen nur hinter einem echten Kaefig stehen. from .isolation.caged_runner import wrapper_is_tools_cage if not wrapper_is_tools_cage(full): raise UnsafeCommandError("TOOLS_PROFILE_REQUIRES_CAGE") return full def render_prompt(envelope: ContextEnvelope, staged_names: Sequence[str]) -> str: parts = [envelope.prompt] if envelope.memory: parts.append("\n\n[Aria-Kontext: Memory-Ausschnitte]") for snippet in envelope.memory: parts.append(f"\n- ({snippet.source}) {snippet.text}") if staged_names: parts.append("\n\n[Anhänge im Arbeitsverzeichnis]") for name in staged_names: parts.append(f"\n- attachments/{name}") return "".join(parts) class CodexAdapter(BackendAdapter): name = "codex" accepts_memory = True def __init__(self, verifier, *, runner: IsolatedRunner, command_template: Sequence[str] = DEFAULT_COMMAND_TEMPLATE, isolation_wrapper: Sequence[str] = (), env: Optional[Mapping[str, str]] = None, max_output_bytes: int = MAX_OUTPUT_BYTES, allow_sandbox_bypass: bool = False, require_output_file: bool = False): super().__init__(verifier) if not isinstance(runner, IsolatedRunner): raise TypeError("IsolatedRunner required") self._allow_bypass = bool(allow_sandbox_bypass) self._require_output_file = bool(require_output_file) self._argv = validate_command_template(command_template, isolation_wrapper, allow_bypass_sandbox=self._allow_bypass, require_cage=self._allow_bypass) self._env = validate_env(dict(env or {})) self._runner = runner self._max_output = max_output_bytes @property def argv(self) -> tuple[str, ...]: return self._argv @property def runner(self) -> IsolatedRunner: return self._runner def _prepare(self, job: AdapterJob) -> Outbound: envelope = job.envelope files = [] for index, attachment in enumerate(envelope.attachments, 1): if attachment.data is None: raise AdapterInputError("ATTACHMENT_CONTENT_MISSING") files.append((attachment_file_name(index, attachment.name or attachment.kind), attachment.data)) text = render_prompt(envelope, [name for name, _ in files]) if not text.strip(): raise AdapterInputError("EMPTY_PROMPT") stdin = text.encode("utf-8") if len(stdin) > MAX_PROMPT_BYTES: raise AdapterInputError("PROMPT_LIMIT") transmitted = { "context": envelope.canonical_dict(), "stdin_sha256": sha256_hex(stdin), "files": [{"name": n, "sha256": sha256_hex(d), "size": len(d)} for n, d in files], "argv": list(self._argv), } return Outbound(self.name, transmitted, payload=(stdin, tuple(files))) def _transmit(self, outbound: Outbound, job: AdapterJob, clearance) -> AdapterResult: stdin, files = outbound.payload try: # unmittelbar vor dem Lauf erneut; Bypass-Tokens nur im geprueften tools-Kaefig assert_safe_argv(self._argv, allow_bypass_sandbox=self._allow_bypass) except UnsafeCommandError as exc: return AdapterResult(job.job_id, self.name, ResultStatus.BLOCKED, error_code=exc.code) spec = RunSpec( argv=self._argv, stdin=stdin, env=dict(self._env), timeout_seconds=job.policy.timeout_seconds, max_output_bytes=self._max_output, files=files, ) result = self._runner.run(spec) if not isinstance(result, RunResult): return AdapterResult(job.job_id, self.name, ResultStatus.FAILED, error_code="INVALID_RUNNER_RESULT") if not result.ok: code = result.error_code or "RUN_FAILED" return AdapterResult(job.job_id, self.name, ResultStatus.FAILED, error_code=code, timed_out=code == "TIMEOUT") # M3: Erfolg nur, wenn Exit-Code ok UND JSONL (turn.completed + agent_message) passt ... outcome = decode_events(result.stdout, allow_tools=self._allow_bypass) if not outcome.success: return AdapterResult(job.job_id, self.name, ResultStatus.FAILED, error_code=outcome.error_code) output = outcome.output if self._require_output_file: # ... UND die -o last.txt-Datei vorhanden ist und EXAKT (nach strip) der LETZTEN # agent_message entspricht. Teilstring-Akzeptanz ist entfernt (Review #14): sonst wuerde # last.txt="OK" eine JSONL-Nachricht "NICHT OK" passieren lassen. last = result.artifact.decode("utf-8", "replace").strip() if result.artifact else "" expected = outcome.last_message.strip() if not last: return AdapterResult(job.job_id, self.name, ResultStatus.FAILED, error_code="OUTPUT_FILE_MISSING") if last != expected: return AdapterResult(job.job_id, self.name, ResultStatus.FAILED, error_code="OUTPUT_MISMATCH") return AdapterResult(job.job_id, self.name, ResultStatus.SUCCESS, output=output)