Source code for acodex.exceptions
from __future__ import annotations
[docs]
class CodexError(Exception):
"""Base exception for all errors raised by acodex."""
[docs]
class CodexCancelledError(CodexError):
"""Raised when a turn is canceled via TurnOptions.signal."""
[docs]
class CodexExecError(CodexError):
"""Raised when the codex executable fails to start or exits unsuccessfully.
Captured process output is available on ``stdout`` and ``stderr`` when present.
"""
def __init__(
self,
message: str,
*,
stdout: str | None = None,
stderr: str | None = None,
) -> None:
super().__init__(_format_exec_error_message(message, stdout=stdout, stderr=stderr))
self.stdout = stdout
self.stderr = stderr
[docs]
class CodexExecutableNotFoundError(CodexExecError):
"""Raised when the codex executable cannot be discovered on PATH."""
def __init__(
self,
message: str,
*,
executable_name: str = "codex",
) -> None:
super().__init__(message)
self.executable_name = executable_name
[docs]
class CodexThreadRunError(CodexError):
"""Raised when a thread run fails while streaming or parsing events."""
[docs]
class CodexThreadStreamNotConsumedError(CodexThreadRunError):
"""Raised when streamed.result is accessed before streamed.events is exhausted."""
[docs]
class CodexStructuredResponseError(CodexThreadRunError):
"""Raised when structured output cannot be created or accessed."""
[docs]
class CodexConfigError(CodexError):
"""Raised when Codex config overrides are invalid or cannot be serialized."""
[docs]
class CodexOutputSchemaError(CodexError):
"""Raised when a TurnOptions.output_schema payload is invalid."""
[docs]
class CodexInternalError(CodexError):
"""Raised when an internal invariant is violated, indicating a bug in acodex."""
def _format_exec_error_message(
message: str,
*,
stdout: str | None,
stderr: str | None,
) -> str:
sections = [message]
normalized_stdout = _normalize_exec_output(stdout)
if normalized_stdout is not None:
sections.extend(("", "STDOUT:", normalized_stdout))
normalized_stderr = _normalize_exec_output(stderr)
if normalized_stderr is not None:
sections.extend(("", "STDERR:", normalized_stderr))
return "\n".join(sections)
def _normalize_exec_output(output: str | None) -> str | None:
if output is None:
return None
normalized_output = output.rstrip("\r\n")
if not normalized_output:
return None
return normalized_output