Agent Runner

vega-core uses a provider-neutral AgentRunner to dispatch AI workloads. Codex is one adapter behind that interface. Stages never call Codex directly; they submit an AgentTask to the runner and receive an AgentResult.

Location

src/agents/

`AgentRunner`

class AgentRunner:
    def __init__(self, adapters: tuple[AgentAdapter, ...]) -> None: ...
    def run(self, task: AgentTask, context: RunContext) -> AgentResult: ...
    def select_adapter(self, task: AgentTask, context: RunContext) -> AgentAdapter: ...
    def cancel(self, task_id: str) -> None: ...

AgentRunner owns:

Adapter selection: If context.policy.agent_provider is set, finds the adapter with that exact name. Otherwise, calls can_run(task, context) on all adapters in order and picks the first that returns True.
Timeout enforcement: Runs the adapter in a ThreadPoolExecutor, polling every 50 ms. Checks both the timeout deadline and the cancellation token. Raises AgentTimedOut or AgentCancelled as appropriate.
Timeout resolution order: Task metadata timeout_seconds → per-stage policy stage_timeout_seconds → policy metadata <provider>_timeout_seconds → agent_timeout_seconds → default_agent_timeout_seconds.
Cancellation registration: Registers the active task with cancellation.register_active_agent_call(task_id, cancel_fn) before running; deregisters after.
Output schema validation: If task.output_schema specifies required keys, the runner validates they are present in result.output. Missing keys produce StageStatus.FAILED with code agent_output_schema_mismatch.
Log events: After a task completes, agent_log events are emitted for each line in result.logs; streaming events during execution come from the adapter itself.
Structured result normalization: Raw adapter output is normalized into a validated AgentResult.

`AgentTask` and `AgentResult`

@dataclass(frozen=True)
class AgentTask:
    task_id: str
    purpose: str               # e.g. "plan", "audit.default", "verify.finding", "patch.generate"
    prompt: str
    working_artifacts: Mapping[str, ArtifactRef] = field(default_factory=dict)
    output_schema: Mapping[str, Any] | None = None  # JSON schema for structured output
    metadata: Mapping[str, Any] = field(default_factory=dict)

@dataclass(frozen=True)
class AgentResult:
    status: StageStatus        # "completed" | "cancelled" | "failed"
    output: Mapping[str, Any]
    artifacts: Mapping[str, ArtifactRef] = field(default_factory=dict)
    logs: tuple[str, ...] = ()
    error: StageError | None = None
    metadata: Mapping[str, Any] = field(default_factory=dict)

`AgentAdapter` protocol

Any agent provider is plugged in by implementing:

class AgentAdapter(Protocol):
    name: str

    def can_run(self, task: AgentTask, context: RunContext) -> bool: ...
    def run(self, task: AgentTask, context: RunContext) -> AgentResult: ...
    def cancel(self, task_id: str) -> None: ...

can_run(...) lets adapters self-select based on task purpose, policy, or resource availability. The runner calls select_adapter(...) to pick the first adapter that returns True.

`CodexAdapter`

The production adapter. It invokes the Codex CLI as a subprocess.

@dataclass
class CodexAdapter:
    name: str = "codex"
    codex_bin: str = "codex"
    cwd: str | Path | None = None
    env: Mapping[str, str] = field(default_factory=dict)
    model: str | None = None
    model_reasoning_effort: str | None = None
    collab: bool = True

    def can_run(self, task: AgentTask, context: Any) -> bool: ...
    def run(self, task: AgentTask, context: Any) -> AgentResult: ...
    def cancel(self, task_id: str) -> None: ...

How `CodexAdapter.run` works

Checks for pre-cancellation — if task_id is already in the cancelled set, raises CodexAdapterCancelled immediately.
Builds a strict child environment — only vars in _CODEX_ENV_ALLOWLIST are forwarded to the subprocess. AWS credential env vars are explicitly excluded and AWS_EC2_METADATA_DISABLED=true is injected. API key values are tracked by _Redactor for log sanitization.

Builds the CLI command:

codex exec --json --dangerously-bypass-approvals-and-sandbox --cd <cwd>
           [--enable multi_agent]   # if collab=True and VEGA_CODEX_MULTI_AGENT_ENABLED
           [--model <model>]
           [-c model_verbosity="high"]
           [-c model_reasoning_effort=<effort>]   # if set
           [-c <llm_proxy_config...>]              # if OPENAI_BASE_URL configured
           [--output-schema <temp_file>]           # if task.output_schema is set
           --output-last-message <temp_file>
           -                                       # read prompt from stdin

Writes the output schema to a temp file if task.output_schema is set. The schema is made strict (all properties required, additionalProperties: false) before writing.
Spawns the subprocess in a new session (start_new_session=True) with the filtered child environment; optionally sets user/group via VEGA_CODEX_RUN_AS_UID / VEGA_CODEX_RUN_AS_GID.
Writes the prompt to stdin and closes it.
Streams stdout line-by-line; each non-empty JSON line is emitted as an agent_stream domain event. Error frames ("error" or "turn_failed" keys) terminate the process immediately.
Forwards stderr on a background thread, also emitting agent_stream events.
Reads the final output from --output-last-message temp file and parses it as JSON. If not valid JSON, wraps raw text as {"text": ..., "content": ...}.
Writes a vega.codex_run_log artifact with full redacted execution context (command, stdin prompt, stdout JSONL, stderr, parsed output).
Returns AgentResult with status=COMPLETED, the parsed output, and metadata.provider="codex".

LLM proxy auto-config

When OPENAI_BASE_URL and OPENAI_API_KEY are set, Codex is configured with a vega_llm_proxy model provider via -c config flags:

-c model_provider=vega_llm_proxy
-c model_providers.vega_llm_proxy.name="Vega LLM Proxy"
-c model_providers.vega_llm_proxy.base_url=<OPENAI_BASE_URL>
-c model_providers.vega_llm_proxy.env_key="OPENAI_API_KEY"

This route is bypassed when VEGA_CORE_CODEX_CONFIG_JSON is set, which allows full Codex config to be injected directly as a JSON object (keys flattened to dotted -c pairs).

Environment isolation

CodexAdapter applies a strict allowlist to the subprocess environment. Only the following vars are forwarded: ANTHROPIC_API_KEY, CODEX_HOME, HOME, LANG, LC_ALL, OPENAI_API_KEY, OPENAI_BASE_URL, PATH, TEMP, TMP, TMPDIR, VEGA_CODEX_MULTI_AGENT_ENABLED, VEGA_PROJECT_ID, VEGA_SCAN_ID, XDG_CACHE_HOME, XDG_CONFIG_HOME. AWS credential vars are explicitly excluded even if they appear in the allowlist.

If VEGA_CODEX_SCRATCH_ROOT is set, it overrides TMPDIR/TMP/TEMP for the Codex process, isolating scratch files to a dedicated directory.

Cancellation

CodexAdapter.cancel(task_id) marks the task as cancelled and calls os.killpg(proc.pid, signal.SIGTERM) to terminate the entire process group. After a 5-second wait, os.killpg(proc.pid, signal.SIGKILL) is sent if the process is still alive. Pre-run cancellation is detected before the subprocess is spawned. Post-run cancellation is detected after proc.wait() returns.

`FakeAgentAdapter`

Used in tests:

@dataclass
class FakeAgentAdapter:
    name: str = "fake"
    responses: dict[str, Mapping[str, Any]] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)
    delay_seconds: float = 0

    def can_run(self, task: AgentTask, context: RunContext) -> bool: ...
    def run(self, task: AgentTask, context: RunContext) -> AgentResult: ...
    def cancel(self, task_id: str) -> None: ...

Wire it into configure_core_services(agent_runner=AgentRunner((fake_adapter,))).

Agent errors

Exception	When raised
`AgentRunnerError`	Base error from runner layer
`AgentSelectionError`	No adapter matches `can_run(...)`
`AgentAdapterExecutionError`	Adapter subprocess or API call failed
`AgentCancelled`	Task was cancelled
`AgentTimedOut`	Task exceeded `stage_timeout_seconds`

These are caught by BaseStage.run(...) and mapped to StageError values.

Agent event flow

The runner emits domain events through the context event bus during execution:

Event kind	When
`agent_started`	Adapter begins execution
`agent_log`	Per-line subprocess output (debug)
`agent_stream`	Streamed partial output
`agent_completed`	Task finished successfully
`agent_failed`	Task failed
`agent_cancelled`	Task was cancelled

Configuration

CodexAdapter is built by build_agent_runner(config: RuntimeConfig) from the CodexRuntimeConfig object. See Configuration for environment variables (VEGA_CODEX_BIN, VEGA_CODEX_REASONING_EFFORT, etc.).

Local vs. AWS

Context	Behavior
Local dev	Codex subprocess spawned in the local shell env
AWS ECS	Codex subprocess spawned in the container; `OPENAI_BASE_URL` points to vega-llm-proxy
Tests	`FakeAgentAdapter` used; no real subprocess

No Docker isolation is used for the agent in either environment. The subprocess runs directly in the runner process's working directory.