diff --git a/docs/platform-v2.md b/docs/platform-v2.md index 3d4b15a8..c08f3dc5 100644 --- a/docs/platform-v2.md +++ b/docs/platform-v2.md @@ -37,6 +37,14 @@ interfaces; platform code does not redefine them. Room payloads are never trusted configuration. +The shared reference-adapter runtime can host either the pinned +OpenAI-compatible participant model or the Codex participant backend. Discord, +Matrix, and Telegram continue to own and authenticate their native ingress and +output. Generic JSONL ingress requires an exact HMAC-authenticated source and +payload envelope; raw channel-shaped prompt text is not an event. Both +backends use the same observation, attention, ACK, scheduling, +participant-host, authorization, transport, and receipt owners. + The shared runtime always owns the attention prompt, attention model selection, judgment schema, policy, scheduling, and wake facts. For Nunchi-owned participants it also owns the normal-turn prompt and action schema. A platform diff --git a/integrations/codex/README.md b/integrations/codex/README.md index 7057527b..dad22f57 100644 --- a/integrations/codex/README.md +++ b/integrations/codex/README.md @@ -1,6 +1,8 @@ # Codex V2 room presence -`nunchi-codex-room-runner` is the only executable Codex integration. +`nunchi-codex-room-runner` is the dedicated shared-Discord Codex host. The +reference Discord, Matrix, Telegram, and generic channel adapters can also +host the same Codex participant backend. It consumes `notifications/nunchi/v2/discord-event`, retains exact self and peer events, runs the participant-bound attention model once, and wakes a @@ -10,28 +12,72 @@ or remains silent. The runner owns the only Discord output seam. The Codex subprocess receives no Nunchi, provider, or room transport credentials and is instructed not to call -Discord tools. The host fixes an isolated working directory, read-only -sandbox, output schema, bounded environment, and disabled shell, browser, -plugin, app, skill, and MCP capabilities; configuration cannot add arbitrary -arguments, binaries, or working directories. Its action is validated and -dispatched with a short-lived exact HMAC through the shared MCP transport. -There is no prompt hook, pre-tool hook, configuration app, or send-time social -judgment. - -Persistent mode binds the saved Codex task ID to the pinned profile, exact -participant, native actor, room, continuity scope, model, sandbox, and disabled -feature set. Corrupt or mismatched state fails safely. Fresh mode starts an -isolated task per opportunity. Host-mediated context expansion may resume the -same task within that opportunity, but capability handles and cursors never -enter the Codex packet. +Discord tools. The current supported `capability_mode` is explicitly +`reduced`: the host fixes an isolated working directory, read-only sandbox, +output schema, bounded environment, and disables effect-bearing shell, +browser, plugin, app, skill, and MCP capabilities. A configured-capability +mode is rejected until those native effects have a version-checked final-effect +bridge. This is an open product limitation tracked by issue #60, not normal +Codex parity. The validated participant action is dispatched with a short-lived +exact HMAC through the shared MCP transport. The runner uses no second social +prompt hook, pre-tool hook, configuration app, or send-time social judgment. + +Persistent mode requires a pinned `runtime_identity` and binds the saved Codex +task ID to the exact executable path and digest, reported Codex version, +`CODEX_HOME`, provider, account binding, credential scope, authentication mode, +continuity generation, profile, participant, native actor, room, model, sandbox, +and reduced feature set. The subprocess is launched under that exact +`CODEX_HOME`; `codex login status` must agree with the pinned authentication +mode. For ChatGPT, API-key, or access-token modes, persistent continuity also +requires readable file-backed authentication: the stored account ID is checked +when available, otherwise a non-secret credential digest is bound. A keyring +whose account identity Codex cannot expose must use fresh mode rather than make +a false continuity claim. The runtime identity is re-attested immediately +before execution. The task ID is staged after a successful, valid participant +result and is made durable only after the shared host accepts participant +silence or reaches the output commit point. Process failure, malformed output, +expiry, cancellation, or host rejection cannot advance resumable state. Because +non-interactive Codex resume is not transactional, the runner consumes the old +pin before resuming it; an unaccepted attempt leaves a recoverable diagnostic +marker and the next opportunity starts a new task instead of reusing possibly +mutated history. An accepted attempt commits the task and clears that marker. +Corrupt or mismatched state fails safely; the probe reports the incompatibility +and a new-task repair path. Fresh mode starts an isolated task per opportunity and +never reports persistent continuity. Host-mediated context expansion may +resume the same staged task within that opportunity, but capability handles and +cursors never enter the Codex packet. + +For portable hosting, `nunchi-discord`, `nunchi-matrix`, `nunchi-telegram`, +and `nunchi-channel` accept a `codex` participant block in place of +`participant_model`. They retain their native normalizer, authenticated self +check, continuity facts, transport, scheduler, participant host, and receipt +owners; only the isolated participant invocation changes to Codex. Repeated +native delivery/event IDs converge on the same durable observation +deduplication boundary. + +The generic `nunchi-channel` surface additionally requires +`ingress_auth: {"source_id":"...","hmac_key_env":"..."}`. Each input line is +an envelope containing `payload` and `authorization`. Authorization version 1 +binds the trusted source ID and SHA-256 of canonical compact JSON payload +bytes. Its HMAC-SHA256 material is +`nunchi.channel.ingress.v1 + NUL + source_id + NUL + payload_sha256`. The key +comes from the named environment variable, is at least 32 bytes, and is never +passed into Codex. Native adapters use provider-authenticated ingress and +reject this generic authentication block. The runner authenticates its exact participant/room MCP session before notifications or tools are available and verifies the gateway-attested Discord self actor and target participant on every closed notification before retaining facts. -The bundled plugin marker intentionally contains no tools and no hooks. It is -not required to run room presence. +The bundled `UserPromptSubmit` hook does not treat raw `` markup as a +native event. It blocks that prompt with a usable native/HMAC-adapter +alternative because current prompt-hook input cannot attest native identity, +mentions, replies, reactions, or continuity. Routing an already interactive +third-party channel session into the same Codex task therefore remains +unsupported until that channel exposes an authenticated native event seam; +the adapter host is the usable autonomous alternative. These are source +interfaces only until each installed adapter has attributable live proof. ```sh nunchi-codex-room-runner --probe @@ -42,4 +88,28 @@ nunchi-codex-room-runner \ The config contains exactly: `schema_version`, `binding`, pinned `profile`, `attention`, `limits`, `state_directory`, shared MCP `transport`, and `codex`. -See `docs/platform-v2.md` for the shared ownership and conformance boundary. +`codex` accepts `model`, `timeout_seconds`, `session_mode`, `capability_mode`, +and `runtime_identity`. Persistent mode requires this closed identity shape: + +```json +{ + "provider": "openai", + "account_id": "workspace-or-account-opaque-id", + "credential_scope": "chatgpt:workspace-opaque-id", + "auth_mode": "chatgpt", + "codex_home": "/absolute/private/codex-home", + "continuity_generation": 1 +} +``` + +`auth_mode` is one of `chatgpt`, `api-key`, or `access-token` and is checked +against `codex login status`. Do not put a token or secret in `account_id` or +`credential_scope`. The pinned +top-level config SHA protects these non-secret identity claims. `--probe` +reports the configured session mode, committed task or incompatibility, runtime +digest/version, hashed account binding, disabled capabilities, and the current +reduced-mode limitation. See `docs/platform-v2.md` for the shared ownership and +conformance boundary. + +Persistent mode also requires an explicit non-empty `model`; relying on a +changing runtime default cannot establish resumable identity. diff --git a/integrations/codex/nunchi-codex/.codex-plugin/plugin.json b/integrations/codex/nunchi-codex/.codex-plugin/plugin.json index 011cb607..9055afbb 100644 --- a/integrations/codex/nunchi-codex/.codex-plugin/plugin.json +++ b/integrations/codex/nunchi-codex/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "nunchi-codex", "version": "2.0.0", - "description": "Documentation marker for host-owned Nunchi V2 Codex room presence.", + "description": "Nunchi V2 Codex room presence and authenticated-ingress guard.", "author": { "name": "Nunchi contributors", "url": "https://github.com/mentatzoe/nunchi" @@ -17,8 +17,8 @@ ], "interface": { "displayName": "Nunchi Codex V2", - "shortDescription": "Host-owned Codex room presence; no direct tools or hooks.", - "longDescription": "The installed nunchi-codex-room-runner owns observation, participant-shaped attention, Codex wake, and the single outbound commit point. This marker intentionally contributes no direct Discord tools and no lifecycle hooks.", + "shortDescription": "Host-owned Codex room presence with safe channel ingress.", + "longDescription": "Nunchi native adapters own authenticated observation, participant-shaped attention, Codex wake, and the single outbound commit point. The bundled prompt hook refuses unauthenticated raw channel markup and directs the operator to a native or HMAC-authenticated adapter; it does not derive platform facts from prompt text.", "developerName": "Nunchi contributors", "category": "Developer Tools", "capabilities": [ diff --git a/integrations/codex/nunchi-codex/hooks/hooks.json b/integrations/codex/nunchi-codex/hooks/hooks.json index deffac97..47512abd 100644 --- a/integrations/codex/nunchi-codex/hooks/hooks.json +++ b/integrations/codex/nunchi-codex/hooks/hooks.json @@ -1,3 +1,15 @@ { - "hooks": {} + "hooks": { + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "nunchi-codex-ingress-hook", + "statusMessage": "Checking Nunchi channel ingress" + } + ] + } + ] + } } diff --git a/pyproject.toml b/pyproject.toml index 04469b8f..031be19d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,7 @@ nunchi-telegram = "nunchi.adapters.telegram:main" nunchi-discord = "nunchi.adapters.discord:main" nunchi-mcp-discord = "nunchi.mcp_discord.server:main" nunchi-codex-room-runner = "nunchi.integrations.codex_v2:main" +nunchi-codex-ingress-hook = "nunchi.integrations.codex_ingress_hook:main" nunchi-hermes-dashboard = "nunchi.integrations.hermes_dashboard_install:main" nunchi-claude-code-room-runner = "nunchi.integrations.claude_code_v2:main" nunchi-service-worker = "nunchi.service_worker:main" diff --git a/src/nunchi/adapters/runtime.py b/src/nunchi/adapters/runtime.py index a0d5f71e..0fc50798 100644 --- a/src/nunchi/adapters/runtime.py +++ b/src/nunchi/adapters/runtime.py @@ -5,6 +5,7 @@ from collections.abc import Mapping from copy import deepcopy import hashlib +import hmac import json import os from pathlib import Path @@ -120,6 +121,74 @@ def _policy(raw: Any) -> AttentionPolicy: raise ValidationError(f"adapter attention policy is invalid: {exc}") from exc +def _canonical_json(value: Any) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + + +class ChannelIngressAuthenticator: + """Authenticate one generic native delivery before normalization.""" + + def __init__(self, raw: Any) -> None: + if not isinstance(raw, Mapping) or set(raw) != {"source_id", "hmac_key_env"}: + raise ValidationError("generic channel ingress_auth has an invalid closed shape") + source_id = raw["source_id"] + key_env = raw["hmac_key_env"] + if not isinstance(source_id, str) or not source_id: + raise ValidationError("generic channel ingress source_id must be non-empty") + if not isinstance(key_env, str) or not key_env: + raise ValidationError("generic channel ingress hmac_key_env must be non-empty") + key = os.environ.get(key_env) + if key is None or len(key.encode("utf-8")) < 32: + raise ValidationError( + f"generic channel ingress key is absent or shorter than 32 bytes in {key_env}" + ) + self.source_id = source_id + self._key = key.encode("utf-8") + + def unwrap(self, envelope: Any) -> Mapping[str, Any]: + if not isinstance(envelope, Mapping) or set(envelope) != { + "payload", + "authorization", + }: + raise ValidationError( + "generic channel ingress requires one authenticated payload envelope" + ) + payload = envelope["payload"] + authorization = envelope["authorization"] + if not isinstance(payload, Mapping): + raise ValidationError("generic channel ingress payload must be an object") + if not isinstance(authorization, Mapping) or set(authorization) != { + "schema_version", + "source_id", + "payload_sha256", + "mac", + }: + raise ValidationError("generic channel ingress authorization is malformed") + if authorization["schema_version"] != 1: + raise ValidationError("generic channel ingress authorization version is unsupported") + if authorization["source_id"] != self.source_id: + raise ValidationError("generic channel ingress source differs from trusted binding") + payload_sha256 = hashlib.sha256(_canonical_json(payload)).hexdigest() + if authorization["payload_sha256"] != payload_sha256: + raise ValidationError("generic channel ingress payload digest differs") + material = ( + b"nunchi.channel.ingress.v1\0" + + self.source_id.encode("utf-8") + + b"\0" + + payload_sha256.encode("ascii") + ) + expected = hmac.new(self._key, material, hashlib.sha256).hexdigest() + supplied = authorization["mac"] + if not isinstance(supplied, str) or not hmac.compare_digest(supplied, expected): + raise ValidationError("generic channel ingress authentication failed") + return payload + + class JsonLineTransport: """Host-attested generic outbound seam used by ``nunchi-channel``.""" @@ -174,7 +243,6 @@ def __init__( "binding", "profile", "attention", - "participant_model", "limits", "state_directory", } @@ -183,8 +251,14 @@ def __init__( "transport", "participant_timeout_seconds", "ack", + "ingress_auth", } - if set(config) - (required | optional) or required - set(config): + participant_backends = set(config) & {"participant_model", "codex"} + if ( + set(config) - (required | optional | {"participant_model", "codex"}) + or required - set(config) + or len(participant_backends) != 1 + ): raise ValidationError("adapter config has a missing or unexpected field") if config["schema_version"] != 2: raise ValidationError("adapter config schema_version must be 2") @@ -219,6 +293,18 @@ def __init__( surface == "channel" and self.binding.platform ): raise ValidationError("adapter surface and trusted platform binding differ") + ingress_auth_raw = config.get("ingress_auth") + if surface == "channel": + if ingress_auth_raw is None: + raise ValidationError( + "generic channel requires ingress_auth; use a native adapter when " + "the upstream source cannot sign canonical deliveries" + ) + self.ingress_auth = ChannelIngressAuthenticator(ingress_auth_raw) + else: + if ingress_auth_raw is not None: + raise ValidationError("native adapters do not accept generic ingress_auth") + self.ingress_auth = None profile_raw = config["profile"] if not isinstance(profile_raw, Mapping) or set(profile_raw) != {"path", "sha256"}: @@ -242,14 +328,6 @@ def __init__( if policy.preattention_enabled else None ) - participant_raw = config["participant_model"] - if not isinstance(participant_raw, Mapping): - raise ValidationError("adapter participant_model must be an object") - participant = OpenAICompatibleParticipant.from_trusted_config( - profile=profile, - config=participant_raw, - environment=os.environ, - ) try: limits = ObservationLimits(**config["limits"]) except (TypeError, ValueError) as exc: @@ -259,7 +337,39 @@ def __init__( stem = hashlib.sha256( f"{surface}\0{self.binding.participant_id}\0{self.binding.continuity_scope_id}".encode() ).hexdigest()[:24] - receipts = ReceiptJournal(state_directory / f"{stem}.receipts.jsonl") + if "participant_model" in participant_backends: + participant_raw = config["participant_model"] + if not isinstance(participant_raw, Mapping): + raise ValidationError("adapter participant_model must be an object") + participant = OpenAICompatibleParticipant.from_trusted_config( + profile=profile, + config=participant_raw, + environment=os.environ, + ) + receipts = ReceiptJournal(state_directory / f"{stem}.receipts.jsonl") + self.participant_backend = "openai-compatible" + else: + codex_raw = config["codex"] + if not isinstance(codex_raw, Mapping): + raise ValidationError("adapter codex config must be an object") + # This local import avoids making the shared adapter layer depend on + # a platform participant at module-import time. + from ..integrations.codex_v2 import ( # noqa: PLC0415 + CodexParticipant, + CodexTaskReceiptJournal, + ) + + participant = CodexParticipant( + profile=profile, + config=codex_raw, + binding=self.binding, + state_directory=state_directory / f"{stem}.codex", + ) + receipts = CodexTaskReceiptJournal( + state_directory / f"{stem}.receipts.jsonl", + participant=participant, + ) + self.participant_backend = "codex" observation = ObservationProvider( self.binding, limits=limits, @@ -318,6 +428,7 @@ def __init__( ) self.surface = surface self.transport = transport + self.participant = participant self.pipeline = NunchiV2Pipeline( observation=observation, attention=attention, @@ -332,7 +443,7 @@ def process( *, live: bool = True, ) -> DeliveryOutcome: - delivery = NORMALIZERS[self.surface](payload, self.binding) + delivery = self._normalize(payload) if not live: observed = self.pipeline.observation.observe( delivery_id=delivery.delivery_id, @@ -355,7 +466,7 @@ def submit( live: bool = True, ) -> DeliveryOutcome: """Retain native ingress promptly and schedule work off the callback.""" - delivery = NORMALIZERS[self.surface](payload, self.binding) + delivery = self._normalize(payload) if not live: observed = self.pipeline.observation.observe( delivery_id=delivery.delivery_id, @@ -374,8 +485,13 @@ def submit( def drain(self, timeout: float | None = None) -> bool: return self.lane.drain(timeout) + def _normalize(self, payload: Mapping[str, Any]): + if self.ingress_auth is not None: + payload = self.ingress_auth.unwrap(payload) + return NORMALIZERS[self.surface](payload, self.binding) + def probe(self) -> dict[str, Any]: - return { + result = { "product": "nunchi", "product_version": __version__, "generation": 2, @@ -385,6 +501,15 @@ def probe(self) -> dict[str, Any]: "actor_id": self.binding.actor_id, "room_id": self.binding.room_id, "continuity_scope_id": self.binding.continuity_scope_id, + "participant_backend": self.participant_backend, + "ingress_authentication": ( + { + "mode": "hmac-sha256", + "source_id": self.ingress_auth.source_id, + } + if self.ingress_auth is not None + else {"mode": "native-adapter"} + ), "capabilities": deepcopy(CAPABILITIES[self.surface]), "interfaces": { "I-010A": 1, @@ -401,6 +526,15 @@ def probe(self) -> dict[str, Any]: "operator_schema_version": 1, "v1_fallback": False, } + if self.participant_backend == "codex": + result["codex"] = { + "session_mode": self.participant.session_mode, + "persistent_session": self.participant.session_mode == "persistent", + "task_state": self.participant.session_status(), + "runtime_identity": self.participant.runtime_status(), + "capability_mode": self.participant.capability_mode, + } + return result def restart(self) -> None: self.lane.restart() diff --git a/src/nunchi/integrations/codex_ingress_hook.py b/src/nunchi/integrations/codex_ingress_hook.py new file mode 100644 index 00000000..c168e692 --- /dev/null +++ b/src/nunchi/integrations/codex_ingress_hook.py @@ -0,0 +1,66 @@ +"""Codex hook guard for unauthenticated channel-shaped prompt text.""" + +from __future__ import annotations + +from collections.abc import Mapping +import json +import re +import sys +from typing import Any, TextIO + +from ..errors import ValidationError + + +_RAW_CHANNEL_MARKUP = re.compile(r")", re.IGNORECASE) +_ALTERNATIVE = ( + "Use the configured nunchi-discord, nunchi-matrix, or nunchi-telegram " + "native adapter; a custom text source must send a canonical delivery " + "through HMAC-authenticated nunchi-channel ingress." +) + + +def evaluate(document: Any) -> dict[str, str] | None: + """Return a Codex hook decision without deriving facts from prompt markup.""" + + if not isinstance(document, Mapping): + raise ValidationError("Codex hook input must be an object") + if document.get("hook_event_name") != "UserPromptSubmit": + raise ValidationError("Codex ingress guard requires UserPromptSubmit") + prompt = document.get("prompt") + if not isinstance(prompt, str): + raise ValidationError("Codex UserPromptSubmit prompt must be a string") + if not _RAW_CHANNEL_MARKUP.search(prompt): + return None + return { + "decision": "block", + "reason": ( + "Nunchi cannot authenticate native identity, mention, reply, reaction, " + f"or continuity facts from raw prompt markup. {_ALTERNATIVE}" + ), + } + + +def main( + argv: list[str] | None = None, + *, + stdin: TextIO | None = None, + stdout: TextIO | None = None, + stderr: TextIO | None = None, +) -> int: + del argv + stdin = stdin or sys.stdin + stdout = stdout or sys.stdout + stderr = stderr or sys.stderr + try: + document = json.load(stdin) + decision = evaluate(document) + except (json.JSONDecodeError, ValidationError) as exc: + print(f"Nunchi Codex ingress guard could not validate hook input: {exc}", file=stderr) + return 2 + if decision is not None: + print(json.dumps(decision, sort_keys=True, separators=(",", ":")), file=stdout) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/nunchi/integrations/codex_v2.py b/src/nunchi/integrations/codex_v2.py index a5c056ea..350aee56 100644 --- a/src/nunchi/integrations/codex_v2.py +++ b/src/nunchi/integrations/codex_v2.py @@ -10,7 +10,9 @@ import os from pathlib import Path import re +import secrets import shutil +import stat import subprocess import sys import threading @@ -72,6 +74,184 @@ "unified_exec", "workspace_dependencies", ) +_MAX_PENDING_TASKS = 8 +_MAX_AUTH_DOCUMENT_BYTES = 1_048_576 +_RUNTIME_IDENTITY_FIELDS = { + "provider", + "account_id", + "credential_scope", + "auth_mode", + "codex_home", + "continuity_generation", +} + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def _atomic_write(path: Path, payload: bytes, *, mode: int = 0o600) -> None: + """Durably replace one private state file without following symlinks.""" + + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + fd = os.open( + temporary, + os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, + mode, + ) + try: + try: + if os.write(fd, payload) != len(payload): + raise OSError(f"short write to {path}") + os.fsync(fd) + finally: + os.close(fd) + os.replace(temporary, path) + directory_fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + except BaseException: + try: + os.unlink(temporary) + except OSError: + pass + raise + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _identity_environment(codex_home: Path) -> dict[str, str]: + environment = { + key: os.environ[key] + for key in ( + "HOME", + "LANG", + "LC_ALL", + "LOGNAME", + "PATH", + "TMPDIR", + "USER", + ) + if key in os.environ + } + environment["CODEX_HOME"] = str(codex_home) + return environment + + +def _codex_version(binary: str, environment: Mapping[str, str]) -> str: + try: + completed = subprocess.run( + [binary, "--version"], + env=dict(environment), + capture_output=True, + text=True, + timeout=20, + check=False, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise ValidationError(f"Codex runtime version is unavailable: {exc}") from exc + version = (completed.stdout or completed.stderr).strip() + if completed.returncode != 0 or not version: + raise ValidationError("Codex runtime version is unavailable") + return version + + +def _codex_auth_mode(binary: str, environment: Mapping[str, str]) -> str: + try: + completed = subprocess.run( + [binary, "login", "status"], + env=dict(environment), + capture_output=True, + text=True, + timeout=20, + check=False, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise ValidationError(f"Codex authentication status is unavailable: {exc}") from exc + reported = f"{completed.stdout}\n{completed.stderr}".lower() + if completed.returncode != 0: + return "absent" + if "chatgpt" in reported: + return "chatgpt" + if "api key" in reported or "api-key" in reported: + return "api-key" + if "access token" in reported or "access-token" in reported: + return "access-token" + return "unknown" + + +def _credential_binding( + codex_home: Path, + *, + auth_mode: str, + expected_account_id: str, + provider: str, + credential_scope: str, +) -> str: + """Return a non-secret binding for the exact credential used by Codex.""" + + auth_path = codex_home / "auth.json" + try: + fd = os.open(auth_path, os.O_RDONLY | os.O_NOFOLLOW) + try: + metadata = os.fstat(fd) + if not stat.S_ISREG(metadata.st_mode): + raise OSError("auth.json is not a regular file") + with os.fdopen(fd, encoding="utf-8") as source: + fd = -1 + raw = source.read(_MAX_AUTH_DOCUMENT_BYTES + 1) + if len(raw.encode("utf-8")) > _MAX_AUTH_DOCUMENT_BYTES: + raise OSError("auth.json exceeds the identity read limit") + document = json.loads(raw) + finally: + if fd >= 0: + os.close(fd) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise ValidationError( + "persistent Codex mode requires readable file-backed credential identity: " + f"{exc}" + ) from exc + if not isinstance(document, Mapping): + raise ValidationError("Codex auth.json identity has an invalid shape") + tokens = document.get("tokens") + tokens = tokens if isinstance(tokens, Mapping) else {} + stored_account_id = tokens.get("account_id") + if isinstance(stored_account_id, str) and stored_account_id: + if stored_account_id != expected_account_id: + raise ValidationError( + "Codex credential account differs from pinned runtime_identity" + ) + material = ( + f"{provider}\0{auth_mode}\0{stored_account_id}\0{credential_scope}" + ) + return hashlib.sha256(material.encode()).hexdigest() + + secret = None + for candidate in ( + document.get("OPENAI_API_KEY"), + document.get("api_key"), + tokens.get("access_token"), + ): + if isinstance(candidate, str) and candidate: + secret = candidate + break + if secret is None: + raise ValidationError( + "Codex credential identity is unavailable for persistent continuity; " + "use fresh mode or file-backed authentication" + ) + secret_digest = hashlib.sha256(secret.encode()).hexdigest() + material = f"{provider}\0{auth_mode}\0{secret_digest}\0{credential_scope}" + return hashlib.sha256(material.encode()).hexdigest() def _strip_json_fence(text: str) -> str: @@ -139,7 +319,13 @@ def __init__( binding: ParticipantBinding, state_directory: str | Path, ) -> None: - allowed = {"model", "timeout_seconds", "session_mode"} + allowed = { + "model", + "timeout_seconds", + "session_mode", + "runtime_identity", + "capability_mode", + } if set(config) - allowed: raise ValidationError("Codex participant config has unexpected fields") self.profile = profile @@ -159,12 +345,105 @@ def __init__( self.timeout_seconds = float(config.get("timeout_seconds", 300)) if not math.isfinite(self.timeout_seconds) or self.timeout_seconds <= 0: raise ValidationError("Codex timeout must be positive and finite") - self.session_mode = str(config.get("session_mode", "persistent")) + self.session_mode = str(config.get("session_mode", "fresh")) if self.session_mode not in ("persistent", "fresh"): raise ValidationError("Codex session_mode must be persistent or fresh") + if self.session_mode == "persistent" and self.model is None: + raise ValidationError("persistent Codex mode requires an exact model") + self.capability_mode = str(config.get("capability_mode", "reduced")) + if self.capability_mode != "reduced": + raise ValidationError( + "Codex configured capabilities are not yet secured by a final-effect " + "bridge; use explicit capability_mode='reduced'" + ) self.session_path = state_root / "codex-v2-session.json" + self.inflight_session_path = state_root / "codex-v2-session.inflight.json" self.output_schema_path = state_root / "codex-v2-action.schema.json" self._write_output_schema() + + identity_raw = config.get("runtime_identity") + if self.session_mode == "persistent" and identity_raw is None: + raise ValidationError( + "persistent Codex mode requires a pinned runtime_identity" + ) + if identity_raw is not None: + if not isinstance(identity_raw, Mapping) or set(identity_raw) != _RUNTIME_IDENTITY_FIELDS: + raise ValidationError("Codex runtime_identity has an invalid closed shape") + strings = { + key: identity_raw[key] + for key in ( + "provider", + "account_id", + "credential_scope", + "auth_mode", + "codex_home", + ) + } + if any(not isinstance(value, str) or not value for value in strings.values()): + raise ValidationError("Codex runtime_identity strings must be non-empty") + generation = identity_raw["continuity_generation"] + if isinstance(generation, bool) or not isinstance(generation, int) or generation < 1: + raise ValidationError( + "Codex continuity_generation must be a positive integer" + ) + if strings["auth_mode"] not in { + "chatgpt", + "api-key", + "access-token", + }: + raise ValidationError("Codex runtime_identity auth_mode is unsupported") + configured_home = Path(strings["codex_home"]) + if not configured_home.is_absolute(): + raise ValidationError("Codex runtime_identity codex_home must be absolute") + try: + self.codex_home = configured_home.resolve(strict=True) + except OSError as exc: + raise ValidationError( + f"Codex runtime_identity codex_home is unavailable: {exc}" + ) from exc + if not self.codex_home.is_dir(): + raise ValidationError("Codex runtime_identity codex_home must be a directory") + try: + binary_path = Path(self.binary).resolve(strict=True) + binary_sha256 = _sha256_file(binary_path) + except OSError as exc: + raise ValidationError( + f"Codex executable identity is unavailable: {exc}" + ) from exc + self.binary = str(binary_path) + identity_environment = _identity_environment(self.codex_home) + version = _codex_version(self.binary, identity_environment) + expected_auth_mode = strings["auth_mode"] + observed_auth_mode = _codex_auth_mode(self.binary, identity_environment) + if observed_auth_mode != expected_auth_mode: + raise ValidationError( + "Codex authenticated runtime differs from pinned runtime_identity" + ) + credential_binding_sha256 = _credential_binding( + self.codex_home, + auth_mode=expected_auth_mode, + expected_account_id=strings["account_id"], + provider=strings["provider"], + credential_scope=strings["credential_scope"], + ) + self.runtime_identity: dict[str, Any] = { + **dict(identity_raw), + "codex_home": str(self.codex_home), + "binary_path": str(binary_path), + "binary_sha256": binary_sha256, + "codex_version": version, + "credential_binding_sha256": credential_binding_sha256, + } + else: + ambient_home = os.environ.get("CODEX_HOME") + self.codex_home = Path( + ambient_home if ambient_home else Path.home() / ".codex" + ).resolve() + self.runtime_identity = { + "bound": False, + "codex_home": str(self.codex_home), + "binary_path": self.binary, + } behavior = { "profile_sha256": self.profile.sha256, "participant_id": self.binding.participant_id, @@ -172,17 +451,17 @@ def __init__( "room_id": self.binding.room_id, "continuity_scope_id": self.binding.continuity_scope_id, "model": self.model, + "runtime_identity": self.runtime_identity, + "capability_mode": self.capability_mode, "disabled_features": list(_DISABLED_CODEX_FEATURES), "sandbox": "read-only", } self.behavior_sha256 = hashlib.sha256( - json.dumps( - behavior, - sort_keys=True, - separators=(",", ":"), - ).encode() + _canonical_json(behavior).encode() ).hexdigest() - self._lock = threading.Lock() + self._lock = threading.RLock() + self._pending_lock = threading.Lock() + self._pending_tasks: dict[str, str] = {} def _write_output_schema(self) -> None: schema = { @@ -200,26 +479,14 @@ def _write_output_schema(self) -> None: }, "required": ["action_json"], } - payload = json.dumps(schema, sort_keys=True, separators=(",", ":")).encode() - temporary = self.output_schema_path.with_suffix(".tmp") - fd = os.open(temporary, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600) - try: - if os.write(fd, payload) != len(payload): - raise OSError("short Codex output-schema write") - os.fsync(fd) - finally: - os.close(fd) - os.replace(temporary, self.output_schema_path) - directory_fd = os.open(self.output_schema_path.parent, os.O_RDONLY) - try: - os.fsync(directory_fd) - finally: - os.close(directory_fd) + _atomic_write(self.output_schema_path, _canonical_json(schema).encode()) def _load_session(self) -> str | None: if self.session_mode == "fresh" or not self.session_path.exists(): return None try: + if self.session_path.is_symlink(): + raise OSError("session state is a symlink") state = json.loads(self.session_path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise RuntimeError(f"Codex session state is not trustworthy: {exc}") from exc @@ -232,6 +499,9 @@ def _load_session(self) -> str | None: "continuity_scope_id", "profile_sha256", "behavior_sha256", + "model", + "capability_mode", + "runtime_identity", } if not isinstance(state, dict) or set(state) != expected: raise RuntimeError("Codex session state has an invalid closed shape") @@ -243,16 +513,53 @@ def _load_session(self) -> str | None: or state["continuity_scope_id"] != self.binding.continuity_scope_id or state["profile_sha256"] != self.profile.sha256 or state["behavior_sha256"] != self.behavior_sha256 + or state["model"] != self.model + or state["capability_mode"] != self.capability_mode + or state["runtime_identity"] != self.runtime_identity or not isinstance(state["thread_id"], str) or not _THREAD_ID.fullmatch(state["thread_id"]) ): raise RuntimeError("Codex session state binding is invalid") return state["thread_id"] + def stage_task(self, request_id: str, thread_id: str) -> None: + """Hold a task ID until the core-owned host accepts this turn.""" + + if ( + self.session_mode != "persistent" + or not isinstance(request_id, str) + or not _THREAD_ID.fullmatch(thread_id) + ): + return + with self._pending_lock: + self._pending_tasks.pop(request_id, None) + self._pending_tasks[request_id] = thread_id + while len(self._pending_tasks) > _MAX_PENDING_TASKS: + self._pending_tasks.pop(next(iter(self._pending_tasks))) + + def commit_task(self, request_id: str) -> None: + """Persist one staged task after host acceptance is durably recorded.""" + + with self._pending_lock: + thread_id = self._pending_tasks.pop(request_id, None) + if thread_id is not None: + with self._lock: + self._save_session(thread_id) + self._clear_inflight_session() + + def discard_task(self, request_id: str | None) -> None: + if not isinstance(request_id, str): + return + with self._pending_lock: + self._pending_tasks.pop(request_id, None) + + @property + def pending_task_count(self) -> int: + with self._pending_lock: + return len(self._pending_tasks) + def _save_session(self, thread_id: str) -> None: - self.session_path.parent.mkdir(parents=True, exist_ok=True) - temporary = self.session_path.with_suffix(".tmp") - payload = json.dumps( + payload = _canonical_json( { "schema_version": 2, "thread_id": thread_id, @@ -262,24 +569,148 @@ def _save_session(self, thread_id: str) -> None: "continuity_scope_id": self.binding.continuity_scope_id, "profile_sha256": self.profile.sha256, "behavior_sha256": self.behavior_sha256, + "model": self.model, + "capability_mode": self.capability_mode, + "runtime_identity": self.runtime_identity, }, - sort_keys=True, - separators=(",", ":"), ).encode() - fd = os.open(temporary, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600) + _atomic_write(self.session_path, payload) + + def _consume_committed_session(self) -> None: + """Make a resumed task non-authoritative before Codex can mutate it. + + Codex has no non-interactive transactional resume seam. Moving the pin + first means a crash, malformed result, cancellation, expiry, or host + rejection resets safely instead of resuming a task whose failed turn + may already have changed its internal history. Host acceptance writes a + fresh committed pin and then clears this recoverable diagnostic marker. + """ + + if not self.session_path.exists(): + return + os.replace(self.session_path, self.inflight_session_path) + directory_fd = os.open(self.session_path.parent, os.O_RDONLY) try: - if os.write(fd, payload) != len(payload): - raise OSError("short Codex session-state write") - os.fsync(fd) + os.fsync(directory_fd) finally: - os.close(fd) - os.replace(temporary, self.session_path) - directory_fd = os.open(self.session_path.parent, os.O_RDONLY) + os.close(directory_fd) + + def _clear_inflight_session(self) -> None: + try: + os.unlink(self.inflight_session_path) + except FileNotFoundError: + return + directory_fd = os.open(self.inflight_session_path.parent, os.O_RDONLY) try: os.fsync(directory_fd) finally: os.close(directory_fd) + def session_status(self) -> dict[str, Any]: + with self._lock: + if self.session_mode == "fresh": + return { + "mode": "fresh", + "status": "new-task", + "compatible": True, + "committed_task_id": None, + "reset_reason": "fresh-mode", + } + try: + thread_id = self._load_session() + except RuntimeError as exc: + return { + "mode": "persistent", + "status": "incompatible", + "compatible": False, + "committed_task_id": None, + "reset_reason": str(exc), + "repair": ( + "quarantine codex-v2-session.json after inspection, then " + "restart to create a new task under the pinned runtime identity" + ), + } + if thread_id is None and self.inflight_session_path.exists(): + return { + "mode": "persistent", + "status": "reset-required", + "compatible": True, + "committed_task_id": None, + "reset_reason": ( + "the prior committed task was consumed before a turn that " + "did not reach host acceptance" + ), + "repair": ( + "the next accepted opportunity will create and commit a new task" + ), + } + return { + "mode": "persistent", + "status": "committed" if thread_id is not None else "new-task", + "compatible": True, + "committed_task_id": thread_id, + "reset_reason": None if thread_id is not None else "not-created", + } + + def runtime_status(self) -> dict[str, Any]: + identity = self.runtime_identity + if identity.get("bound") is False: + return { + "bound": False, + "codex_home": identity["codex_home"], + "binary_path": identity["binary_path"], + } + account_binding = hashlib.sha256( + ( + f"{identity['provider']}\0{identity['account_id']}\0" + f"{identity['credential_scope']}" + ).encode() + ).hexdigest() + return { + "bound": True, + "provider": identity["provider"], + "auth_mode": identity["auth_mode"], + "credential_scope": identity["credential_scope"], + "account_binding_sha256": account_binding, + "codex_home": identity["codex_home"], + "binary_path": identity["binary_path"], + "binary_sha256": identity["binary_sha256"], + "codex_version": identity["codex_version"], + "credential_binding_sha256": identity["credential_binding_sha256"], + "continuity_generation": identity["continuity_generation"], + } + + def _environment(self) -> dict[str, str]: + return _identity_environment(self.codex_home) + + def _verify_runtime_identity(self) -> None: + """Re-attest persistent identity immediately before native execution.""" + + identity = self.runtime_identity + if identity.get("bound") is False: + return + try: + binary_path = Path(self.binary).resolve(strict=True) + binary_sha256 = _sha256_file(binary_path) + except OSError as exc: + raise RuntimeError(f"Codex executable identity is unavailable: {exc}") from exc + environment = self._environment() + observed = { + "binary_path": str(binary_path), + "binary_sha256": binary_sha256, + "codex_version": _codex_version(self.binary, environment), + "auth_mode": _codex_auth_mode(self.binary, environment), + "credential_binding_sha256": _credential_binding( + self.codex_home, + auth_mode=identity["auth_mode"], + expected_account_id=identity["account_id"], + provider=identity["provider"], + credential_scope=identity["credential_scope"], + ), + } + if any(identity[key] != value for key, value in observed.items()): + raise RuntimeError("Codex runtime identity changed before execution") + def _prompt(self, protocol: ParticipantTurnProtocol) -> str: """Compatibility accessor; the prompt bytes are owned by core.""" @@ -292,7 +723,10 @@ def run_protocol(self, *, wake, opportunity, expand, cancel): opportunity=opportunity, ) with self._lock: + self._verify_runtime_identity() active_thread = self._load_session() + if active_thread is not None: + self._consume_committed_session() extra = [ "--ignore-user-config", "--ignore-rules", @@ -336,26 +770,14 @@ def run_protocol(self, *, wake, opportunity, expand, cancel): process = subprocess.Popen( command, cwd=self.working_directory, - env={ - key: os.environ[key] - for key in ( - "CODEX_HOME", - "HOME", - "LANG", - "LC_ALL", - "LOGNAME", - "PATH", - "TMPDIR", - "USER", - ) - if key in os.environ - }, + env=self._environment(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) while process.poll() is None: if cancel.is_set() or time.monotonic() >= deadline: + self.discard_task(protocol.request_id) process.terminate() try: process.wait(timeout=2) @@ -371,12 +793,6 @@ def run_protocol(self, *, wake, opportunity, expand, cancel): f"expected {active_thread}" ) active_thread = thread_id or active_thread - if self.session_mode == "persistent": - if active_thread is None: - raise RuntimeError( - "Codex did not report a persistent task ID" - ) - self._save_session(active_thread) if process.returncode != 0: raise RuntimeError( (stderr or f"Codex exited {process.returncode}")[-500:] @@ -387,6 +803,12 @@ def run_protocol(self, *, wake, opportunity, expand, cancel): ) done, action = protocol.consume(raw_action, expand=expand) if done: + if self.session_mode == "persistent": + if active_thread is None: + raise RuntimeError( + "Codex did not report a persistent task ID" + ) + self.stage_task(protocol.request_id, active_thread) return action if active_thread is None: raise RuntimeError( @@ -411,6 +833,26 @@ def __call__(self, *, wake, expand, cancel): ) +class CodexTaskReceiptJournal(ReceiptJournal): + """Commit Codex task continuity only after the host accepts the turn.""" + + def __init__(self, path, *, participant: CodexParticipant | None = None, **kwargs) -> None: + super().__init__(path, **kwargs) + self.participant = participant + + def append(self, record, *, writer): + appended = super().append(record, writer=writer) + participant = self.participant + if participant is None: + return appended + if appended["stage"] == "transport": + participant.commit_task(appended["request_id"]) + elif appended["stage"] == "participant-host": + if appended["body"].get("outcome") == "silent": + participant.commit_task(appended["request_id"]) + return appended + + class CodexRoomRuntime: def __init__(self, config: Mapping[str, Any], client: StreamableMCPClient) -> None: required = { @@ -464,7 +906,16 @@ def __init__(self, config: Mapping[str, Any], client: StreamableMCPClient) -> No limits = ObservationLimits(**config["limits"]) state = Path(config["state_directory"]) state.mkdir(parents=True, exist_ok=True) - receipts = ReceiptJournal(state / "codex-v2-receipts.jsonl") + participant = CodexParticipant( + profile=profile, + config=config["codex"], + binding=self.binding, + state_directory=state, + ) + receipts = CodexTaskReceiptJournal( + state / "codex-v2-receipts.jsonl", + participant=participant, + ) observation = ObservationProvider( self.binding, limits=limits, @@ -479,12 +930,6 @@ def __init__(self, config: Mapping[str, Any], client: StreamableMCPClient) -> No scheduler = ConversationOpportunityScheduler( f"{self.binding.participant_id}:{self.binding.continuity_scope_id}" ) - participant = CodexParticipant( - profile=profile, - config=config["codex"], - binding=self.binding, - state_directory=state, - ) try: ack_policy = AckPolicy(**dict(config.get("ack", {}))) except (TypeError, ValueError) as exc: @@ -522,6 +967,7 @@ def __init__(self, config: Mapping[str, Any], client: StreamableMCPClient) -> No ) self.lane = AsyncDeliveryLane(self.pipeline) self.client = client + self.participant = participant self.output_secret = self._output_secret(config["transport"]) @staticmethod @@ -625,6 +1071,7 @@ def transport_interrupted(self) -> None: ) def probe(self): + session = self.participant.session_status() return { "product": "nunchi", "product_version": __version__, @@ -633,7 +1080,16 @@ def probe(self): "participant_id": self.binding.participant_id, "actor_id": self.binding.actor_id, "room_id": self.binding.room_id, - "persistent_session": True, + "session_mode": self.participant.session_mode, + "persistent_session": self.participant.session_mode == "persistent", + "task_state": session, + "runtime_identity": self.participant.runtime_status(), + "capability_mode": self.participant.capability_mode, + "disabled_capabilities": list(_DISABLED_CODEX_FEATURES), + "capability_limitation": ( + "normal Codex effect-bearing capabilities remain unavailable until " + "they have a version-checked final-effect bridge" + ), "shared_discord_transport": True, "send_time_social_judgment": False, "v1_fallback": False, diff --git a/tests/v2/test_runtime_hardening.py b/tests/v2/test_runtime_hardening.py index 782760d8..4b8dea0f 100644 --- a/tests/v2/test_runtime_hardening.py +++ b/tests/v2/test_runtime_hardening.py @@ -1027,7 +1027,14 @@ def test_packaging_and_codex_bundle_have_no_v1_execution_paths(self): / ".mcp.json" ).read_text() ) - self.assertEqual({"hooks": {}}, hooks) + self.assertEqual({"UserPromptSubmit"}, set(hooks["hooks"])) + prompt_handlers = hooks["hooks"]["UserPromptSubmit"][0]["hooks"] + self.assertEqual(1, len(prompt_handlers)) + self.assertEqual("command", prompt_handlers[0]["type"]) + self.assertEqual( + "nunchi-codex-ingress-hook", + prompt_handlers[0]["command"], + ) self.assertEqual({"mcpServers": {}}, tools) diff --git a/tests/v2/test_surfaces.py b/tests/v2/test_surfaces.py index 0d06cfb9..1bc90f62 100644 --- a/tests/v2/test_surfaces.py +++ b/tests/v2/test_surfaces.py @@ -2,6 +2,7 @@ from copy import deepcopy import hashlib +import hmac import io import json import os @@ -18,6 +19,7 @@ normalize_matrix_event, normalize_telegram_update, ) +from nunchi.adapters.runtime import ReferenceAdapterRuntime from nunchi.adapters.matrix import MatrixTransport from nunchi.adapters.telegram import TelegramTransport from nunchi.adapters.discord import DiscordPyTransport, DurableGatewaySequence @@ -26,7 +28,10 @@ from nunchi.integrations.codex_v2 import ( CodexParticipant, CodexRoomRuntime, + CodexTaskReceiptJournal, + _atomic_write, MCPDiscordTransport, + _credential_binding, _parse_codex_output, ) from nunchi.integrations.mcp_client import StreamableMCPClient @@ -886,6 +891,29 @@ def _codex_identity(): ), ) + @staticmethod + def _persistent_runtime(directory, **overrides): + root = Path(directory) + codex_home = root / "codex-home" + codex_home.mkdir(exist_ok=True) + binary = root / "codex" + binary.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + binary.chmod(0o700) + identity = { + "provider": "openai", + "account_id": "workspace:test-account", + "credential_scope": "chatgpt:test-workspace", + "auth_mode": "chatgpt", + "codex_home": str(codex_home), + "continuity_generation": 1, + } + (codex_home / "auth.json").write_text( + json.dumps({"tokens": {"account_id": identity["account_id"]}}), + encoding="utf-8", + ) + identity.update(overrides) + return binary, identity + @staticmethod def _wake(): return { @@ -1203,23 +1231,707 @@ def kill(self): def test_codex_persistent_task_is_bound_to_profile_actor_room_and_behavior(self): profile, binding = self._codex_identity() thread_id = "019f9432-9300-7dd1-8225-d7f10f921968" + with tempfile.TemporaryDirectory() as directory: + binary, identity = self._persistent_runtime(directory) + with ( + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_version", + return_value="codex-cli test", + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_auth_mode", + return_value="chatgpt", + ), + ): + first = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + first._save_session(thread_id) + self.assertEqual(thread_id, first._load_session()) + first._consume_committed_session() + self.assertFalse(first.session_path.exists()) + self.assertTrue(first.inflight_session_path.exists()) + first.stage_task("accepted-request", thread_id) + first.commit_task("accepted-request") + self.assertEqual(thread_id, first._load_session()) + self.assertFalse(first.inflight_session_path.exists()) + changed = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-b", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + with self.assertRaises(RuntimeError): + changed._load_session() + with self.assertRaisesRegex(ValidationError, "account differs"): + CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": { + **identity, + "account_id": "workspace:different-account", + "continuity_generation": 2, + }, + }, + binding=binding, + state_directory=directory, + ) + + def test_codex_persistent_mode_requires_an_exact_runtime_identity(self): + profile, binding = self._codex_identity() with tempfile.TemporaryDirectory() as directory, self._installed_codex(): - first = CodexParticipant( - profile=profile, - config={"session_mode": "persistent", "model": "model-a"}, - binding=binding, - state_directory=directory, + with self.assertRaisesRegex( + ValidationError, + "requires a pinned runtime_identity", + ): + CodexParticipant( + profile=profile, + config={"session_mode": "persistent", "model": "model-a"}, + binding=binding, + state_directory=directory, + ) + + def test_codex_credential_binding_checks_the_file_backed_account(self): + with tempfile.TemporaryDirectory() as directory: + codex_home = Path(directory) + (codex_home / "auth.json").write_text( + json.dumps( + { + "auth_mode": "chatgpt", + "tokens": { + "account_id": "account-1", + "access_token": "secret-token", + }, + } + ), + encoding="utf-8", + ) + digest = _credential_binding( + codex_home, + auth_mode="chatgpt", + expected_account_id="account-1", + provider="openai", + credential_scope="chatgpt:workspace-1", + ) + self.assertEqual(64, len(digest)) + self.assertNotIn("secret-token", digest) + with self.assertRaisesRegex(ValidationError, "account differs"): + _credential_binding( + codex_home, + auth_mode="chatgpt", + expected_account_id="account-2", + provider="openai", + credential_scope="chatgpt:workspace-1", + ) + + def test_codex_atomic_state_failure_preserves_prior_committed_bytes(self): + with tempfile.TemporaryDirectory() as directory: + state = Path(directory) / "session.json" + state.write_bytes(b"prior") + with ( + mock.patch( + "nunchi.integrations.codex_v2.os.write", + side_effect=OSError("write failed"), + ), + self.assertRaisesRegex(OSError, "write failed"), + ): + _atomic_write(state, b"replacement") + self.assertEqual(b"prior", state.read_bytes()) + + def test_codex_runtime_identity_is_rechecked_before_execution(self): + profile, binding = self._codex_identity() + with tempfile.TemporaryDirectory() as directory: + binary, identity = self._persistent_runtime(directory) + with ( + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_version", + return_value="codex-cli test", + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_auth_mode", + return_value="chatgpt", + ), + ): + participant = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + binary.write_text("#!/bin/sh\nexit 7\n", encoding="utf-8") + with self.assertRaisesRegex(RuntimeError, "identity changed"): + participant._verify_runtime_identity() + + def test_codex_uses_shared_authenticated_reference_adapters(self): + class Transport: + def ordinary_action_capabilities(self): + return ("message", "reply", "reaction") + + def reaction_capability(self): + return None + + def dispatch(self, *, action, wake): + return TransportResult("sent", "test") + + cases = { + "discord": ( + "discord:actor:9", + "42", + { + "t": "MESSAGE_CREATE", + "s": 1, + "delivery_epoch": "gateway-a", + "d": { + "id": "100", + "channel_id": "42", + "author": {"id": "7", "username": "Zoe", "bot": False}, + "content": "hello", + "mentions": [], + "mention_everyone": False, + }, + }, + ), + "matrix": ( + "matrix:actor:@vigil:example", + "!room:example", + { + "room_id": "!room:example", + "event": { + "event_id": "$event", + "type": "m.room.message", + "sender": "@zoe:example", + "content": {"msgtype": "m.text", "body": "hello"}, + }, + }, + ), + "telegram": ( + "telegram:actor:9", + "-42", + { + "update_id": 1, + "message": { + "message_id": 100, + "chat": {"id": -42}, + "from": {"id": 7, "first_name": "Zoe", "is_bot": False}, + "text": "hello", + }, + }, + ), + "channel": ( + "custom:actor:9", + "room-42", + { + "delivery_id": "custom:delivery:1", + "room_id": "room-42", + "event": { + "id": "custom:message:100", + "type": "message", + "author_id": "custom:actor:7", + "text": "hello", + "mentioned_actor_ids": [], + "mentions_room": False, + }, + "actors": {"custom:actor:7": {"kind": "human"}}, + }, + ), + } + ingress_key = "i" * 32 + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + binary = root / "codex" + binary.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + binary.chmod(0o700) + for surface, (actor_id, room_id, payload) in cases.items(): + with self.subTest(surface=surface): + case_root = root / surface + case_root.mkdir() + profile_path = case_root / "profile.json" + profile_bytes = json.dumps( + { + "profile_id": "vigil", + "participant_id": "vigil", + "actor_id": actor_id, + "instructions": "Contribute carefully.", + "provenance": "trusted:test", + }, + sort_keys=True, + separators=(",", ":"), + ).encode() + profile_path.write_bytes(profile_bytes) + binding_platform = "custom" if surface == "channel" else surface + config = { + "schema_version": 2, + "binding": { + "participant_id": "vigil", + "actor_id": actor_id, + "platform": binding_platform, + "room_id": room_id, + "continuity_scope_id": f"{binding_platform}:{room_id}", + }, + "profile": { + "path": str(profile_path), + "sha256": hashlib.sha256(profile_bytes).hexdigest(), + }, + "attention": { + "policy": { + "preattention_enabled": False, + "suppression_enabled": False, + }, + "model": {}, + }, + "codex": {"session_mode": "fresh"}, + "limits": {}, + "state_directory": str(case_root / "state"), + **( + { + "ingress_auth": { + "source_id": "trusted-channel-plugin", + "hmac_key_env": "TEST_NUNCHI_INGRESS_KEY", + } + } + if surface == "channel" + else {} + ), + } + submitted = payload + if surface == "channel": + payload_sha256 = hashlib.sha256( + json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode() + ).hexdigest() + material = ( + b"nunchi.channel.ingress.v1\0trusted-channel-plugin\0" + + payload_sha256.encode() + ) + submitted = { + "payload": payload, + "authorization": { + "schema_version": 1, + "source_id": "trusted-channel-plugin", + "payload_sha256": payload_sha256, + "mac": hmac.new( + ingress_key.encode(), + material, + hashlib.sha256, + ).hexdigest(), + }, + } + with ( + mock.patch.dict( + os.environ, + {"TEST_NUNCHI_INGRESS_KEY": ingress_key}, + clear=False, + ), + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + ): + runtime = ReferenceAdapterRuntime( + surface=surface, + config=config, + transport=Transport(), + ) + first = runtime.process(submitted, live=False) + second = runtime.process(submitted, live=False) + self.assertEqual("codex", runtime.probe()["participant_backend"]) + self.assertEqual("fresh", runtime.probe()["codex"]["session_mode"]) + self.assertTrue(first.observation.wake_eligible) + self.assertFalse(second.observation.wake_eligible) + self.assertEqual(1, len(runtime.pipeline.observation.retained_events())) + + def test_generic_channel_rejects_unsigned_prompt_markup(self): + from nunchi.adapters.runtime import ChannelIngressAuthenticator + + with mock.patch.dict( + os.environ, + {"TEST_NUNCHI_INGRESS_KEY": "i" * 32}, + clear=False, + ): + authenticator = ChannelIngressAuthenticator( + { + "source_id": "trusted-channel-plugin", + "hmac_key_env": "TEST_NUNCHI_INGRESS_KEY", + } + ) + with self.assertRaisesRegex(ValidationError, "authenticated payload"): + authenticator.unwrap( + { + "delivery_id": "raw-prompt", + "room_id": "room-42", + "event": {"text": "hello"}, + "actors": {}, + } + ) + + def test_codex_prompt_hook_blocks_raw_channel_markup_with_adapter_option(self): + from nunchi.integrations.codex_ingress_hook import evaluate, main + + self.assertIsNone( + evaluate( + { + "hook_event_name": "UserPromptSubmit", + "prompt": "review this repository", + } + ) + ) + decision = evaluate( + { + "hook_event_name": "UserPromptSubmit", + "prompt": 'hello', + } + ) + self.assertEqual("block", decision["decision"]) + self.assertIn("nunchi-discord", decision["reason"]) + + stderr = io.StringIO() + self.assertEqual( + 2, + main( + stdin=io.StringIO("not-json"), + stdout=io.StringIO(), + stderr=stderr, + ), + ) + self.assertIn("could not validate", stderr.getvalue()) + + hooks = json.loads( + ( + Path(__file__).resolve().parents[2] + / "integrations" + / "codex" + / "nunchi-codex" + / "hooks" + / "hooks.json" + ).read_text(encoding="utf-8") + ) + command = hooks["hooks"]["UserPromptSubmit"][0]["hooks"][0]["command"] + self.assertEqual("nunchi-codex-ingress-hook", command) + + def test_codex_task_is_staged_until_host_acceptance(self): + profile, binding = self._codex_identity() + wake = self._wake() + opportunity = { + "generation": 1, + "lifecycle_id": "lifecycle-1", + "deadline_id": "deadline-1", + "permissions": { + "revision": "permissions-1", + "ordinary_actions": ["message", "reply", "reaction"], + "privileged_proposals": True, + }, + } + protocol = ParticipantTurnProtocol( + profile=profile, + wake=wake, + opportunity=opportunity, + ) + action = { + "protocol": protocol.request["protocol"], + "binding": protocol.request["binding"], + "action": {"kind": "silence"}, + } + task_id = "019f9432-9300-7dd1-8225-d7f10f921968" + stdout = "\n".join( + ( + json.dumps({"type": "thread.started", "thread_id": task_id}), + json.dumps( + { + "type": "item.completed", + "item": { + "type": "agent_message", + "text": json.dumps( + {"action_json": json.dumps(action)} + ), + }, + } + ), ) - first._save_session(thread_id) - self.assertEqual(thread_id, first._load_session()) - changed = CodexParticipant( + ) + + class Process: + returncode = 0 + + def poll(self): + return 0 + + def communicate(self): + return stdout, "" + + with tempfile.TemporaryDirectory() as directory: + binary, identity = self._persistent_runtime(directory) + with ( + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_version", + return_value="codex-cli test", + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_auth_mode", + return_value="chatgpt", + ), + ): + participant = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + with mock.patch( + "nunchi.integrations.codex_v2.subprocess.Popen", + return_value=Process(), + ), mock.patch.object(participant, "_verify_runtime_identity"): + self.assertIsNone( + participant.run_protocol( + wake=wake, + opportunity=opportunity, + expand=lambda **_: {}, + cancel=threading.Event(), + ) + ) + self.assertEqual(1, participant.pending_task_count) + self.assertFalse(participant.session_path.exists()) + + journal = object.__new__(CodexTaskReceiptJournal) + journal.participant = participant + accepted = { + "request_id": "r", + "stage": "participant-host", + "writer": "participant-host", + "body": {"outcome": "silent"}, + } + with mock.patch( + "nunchi.integrations.codex_v2.ReceiptJournal.append", + return_value=accepted, + ): + CodexTaskReceiptJournal.append( + journal, + accepted, + writer="participant-host", + ) + self.assertEqual(0, participant.pending_task_count) + self.assertEqual(task_id, participant._load_session()) + + def test_failed_or_malformed_codex_turn_never_stages_task_state(self): + profile, binding = self._codex_identity() + wake = self._wake() + opportunity = { + "generation": 1, + "lifecycle_id": "lifecycle-1", + "deadline_id": "deadline-1", + "permissions": { + "revision": "permissions-1", + "ordinary_actions": ["message", "reply", "reaction"], + "privileged_proposals": True, + }, + } + protocol = ParticipantTurnProtocol( + profile=profile, + wake=wake, + opportunity=opportunity, + ) + action = { + "protocol": protocol.request["protocol"], + "binding": protocol.request["binding"], + "action": {"kind": "silence"}, + } + task_id = "019f9432-9300-7dd1-8225-d7f10f921968" + + class Process: + def __init__(self, returncode, final_text): + self.returncode = returncode + self.final_text = final_text + + def poll(self): + return self.returncode + + def communicate(self): + return ( + "\n".join( + ( + json.dumps( + {"type": "thread.started", "thread_id": task_id} + ), + json.dumps( + { + "type": "item.completed", + "item": { + "type": "agent_message", + "text": self.final_text, + }, + } + ), + ) + ), + "failed", + ) + + cases = ( + Process(7, json.dumps({"action_json": json.dumps(action)})), + Process(0, json.dumps({"action_json": "not-json"})), + ) + with tempfile.TemporaryDirectory() as directory: + binary, identity = self._persistent_runtime(directory) + with ( + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_version", + return_value="codex-cli test", + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_auth_mode", + return_value="chatgpt", + ), + ): + participant = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + for process in cases: + with ( + self.subTest(returncode=process.returncode), + mock.patch( + "nunchi.integrations.codex_v2.subprocess.Popen", + return_value=process, + ), + mock.patch.object(participant, "_verify_runtime_identity"), + self.assertRaises(RuntimeError), + ): + participant.run_protocol( + wake=wake, + opportunity=opportunity, + expand=lambda **_: {}, + cancel=threading.Event(), + ) + self.assertEqual(0, participant.pending_task_count) + self.assertFalse(participant.session_path.exists()) + + participant._save_session(task_id) + malformed_resume = Process( + 0, + json.dumps({"action_json": "not-json"}), + ) + with ( + mock.patch( + "nunchi.integrations.codex_v2.subprocess.Popen", + return_value=malformed_resume, + ), + mock.patch.object(participant, "_verify_runtime_identity"), + self.assertRaises(RuntimeError), + ): + participant.run_protocol( + wake=wake, + opportunity=opportunity, + expand=lambda **_: {}, + cancel=threading.Event(), + ) + self.assertFalse(participant.session_path.exists()) + self.assertTrue(participant.inflight_session_path.exists()) + self.assertEqual("reset-required", participant.session_status()["status"]) + + def test_fresh_mode_and_reduced_capability_are_reported_truthfully(self): + profile, binding = self._codex_identity() + with tempfile.TemporaryDirectory() as directory, self._installed_codex(): + participant = CodexParticipant( profile=profile, - config={"session_mode": "persistent", "model": "model-b"}, + config={"session_mode": "fresh", "capability_mode": "reduced"}, binding=binding, state_directory=directory, ) - with self.assertRaises(RuntimeError): - changed._load_session() + self.assertEqual("fresh", participant.session_status()["mode"]) + self.assertFalse(participant.session_path.exists()) + with self.assertRaisesRegex( + ValidationError, + "final-effect bridge", + ): + CodexParticipant( + profile=profile, + config={ + "session_mode": "fresh", + "capability_mode": "configured", + }, + binding=binding, + state_directory=directory, + ) + + def test_corrupt_task_state_is_reported_as_incompatible(self): + profile, binding = self._codex_identity() + with tempfile.TemporaryDirectory() as directory: + binary, identity = self._persistent_runtime(directory) + with ( + mock.patch( + "nunchi.integrations.codex_v2.shutil.which", + return_value=str(binary), + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_version", + return_value="codex-cli test", + ), + mock.patch( + "nunchi.integrations.codex_v2._codex_auth_mode", + return_value="chatgpt", + ), + ): + participant = CodexParticipant( + profile=profile, + config={ + "session_mode": "persistent", + "model": "model-a", + "runtime_identity": identity, + }, + binding=binding, + state_directory=directory, + ) + participant.session_path.write_text("not-json", encoding="utf-8") + status = participant.session_status() + self.assertEqual("incompatible", status["status"]) + self.assertFalse(status["compatible"]) + self.assertIn("quarantine", status["repair"]) def test_codex_registers_and_rejects_wrong_target_before_observation(self): secret = "s" * 32 @@ -1302,6 +2014,11 @@ def call_tool(self, name, arguments): self._installed_codex(), ): runtime = CodexRoomRuntime(config, Client()) + probe = runtime.probe() + self.assertEqual("fresh", probe["session_mode"]) + self.assertFalse(probe["persistent_session"]) + self.assertEqual("reduced", probe["capability_mode"]) + self.assertTrue(probe["disabled_capabilities"]) runtime.register_transport() before = runtime.pipeline.observation.retained_events() with self.assertRaises(ValidationError):