feat(codex): event-stream parser tap for the unified harness surface

examples/tutorials/00_sync/harness_codex/Dockerfile

@@ -0,0 +1,50 @@
+# syntax=docker/dockerfile:1.3
+FROM python:3.12-slim
+COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    htop \
+    vim \
+    curl \
+    tar \
+    python3-dev \
+    postgresql-client \
+    build-essential \
+    libpq-dev \
+    gcc \
+    cmake \
+    netcat-openbsd \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN uv pip install --system --upgrade pip setuptools wheel
+
+ENV UV_HTTP_TIMEOUT=1000
+
+# Copy pyproject.toml and README.md to install dependencies
+COPY 00_sync/harness_codex/pyproject.toml /app/harness_codex/pyproject.toml
+COPY 00_sync/harness_codex/README.md /app/harness_codex/README.md
+
+WORKDIR /app/harness_codex
+
+# Copy the project code
+COPY 00_sync/harness_codex/project /app/harness_codex/project
+
+# Copy the test files
+COPY 00_sync/harness_codex/tests /app/harness_codex/tests
+
+# Copy shared test utilities
+COPY test_utils /app/test_utils
+
+# Install the required Python packages with dev dependencies
+RUN uv pip install --system .[dev]
+
+# Set environment variables
+ENV PYTHONPATH=/app
+
+# Set test environment variables
+ENV AGENT_NAME=s-harness-codex
+
+# Run the agent using uvicorn
+CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"]

examples/tutorials/00_sync/harness_codex/README.md

@@ -0,0 +1,40 @@
+# harness_codex (sync)
+
+Tutorial agent demonstrating the `convert_codex_to_agentex_events` tap,
+`CodexTurn`, and `UnifiedEmitter` for a **sync** (HTTP-yield) ACP agent.
+
+## What this tutorial shows
+
+- Spawning `codex exec --json` as a **local asyncio subprocess** (no Scale sandbox).
+- Wrapping the stdout line stream in a `CodexTurn`.
+- Delivering every canonical `StreamTaskMessage*` event to the HTTP caller via
+  `UnifiedEmitter.yield_turn` (tracing as a side-effect).
+
+> **Production isolation note:** A tutorial agent runs the Codex CLI locally.
+> Production-grade isolation (Scale sandbox, secret injection, MCP configuration)
+> is handled by the golden agent at
+> `teams/sgp/agents/golden_agent/project/harness/providers/codex.py`.
+
+## Live runs
+
+Live runs require:
+1. The `codex` CLI on PATH: `npm install -g @openai/codex`
+2. `OPENAI_API_KEY` set in the environment.
+
+## Running offline unit tests
+
+The offline tests inject a fake subprocess and never invoke the real CLI:
+
+```bash
+cd /path/to/scale-agentex-python
+uv run --all-packages --all-extras pytest examples/tutorials/00_sync/harness_codex/tests/test_agent.py -q
+```
+
+## Running live integration tests
+
+```bash
+export CODEX_LIVE_TESTS=1
+export OPENAI_API_KEY=sk-...
+# Start the agent server first, then:
+pytest tests/test_agent.py -v
+```

examples/tutorials/00_sync/harness_codex/conftest.py

@@ -0,0 +1,12 @@
+"""Add the agent's project root to sys.path so ``import project`` works.
+
+Also sets minimal environment variables so the FastACP and tracing modules
+can be imported without a running agent server.
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+os.environ.setdefault("ACP_URL", "http://localhost:8000")

examples/tutorials/00_sync/harness_codex/manifest.yaml

@@ -0,0 +1,58 @@
+build:
+  context:
+    root: ../../
+    include_paths:
+      - 00_sync/harness_codex
+      - test_utils
+    dockerfile: 00_sync/harness_codex/Dockerfile
+    dockerignore: 00_sync/harness_codex/.dockerignore
+
+local_development:
+  agent:
+    port: 8000
+    host_address: host.docker.internal
+  paths:
+    acp: project/acp.py
+
+agent:
+  acp_type: sync
+  name: s-harness-codex
+  description: Sync tutorial agent driving the unified harness surface via local codex CLI subprocess
+
+  temporal:
+    enabled: false
+
+  credentials:
+    - env_var_name: OPENAI_API_KEY
+      secret_name: openai-api-key
+      secret_key: api-key
+    - env_var_name: REDIS_URL
+      secret_name: redis-url-secret
+      secret_key: url
+    - env_var_name: SGP_API_KEY
+      secret_name: sgp-api-key
+      secret_key: api-key
+    - env_var_name: SGP_ACCOUNT_ID
+      secret_name: sgp-account-id
+      secret_key: account-id
+    - env_var_name: SGP_CLIENT_BASE_URL
+      secret_name: sgp-client-base-url
+      secret_key: url
+
+deployment:
+  image:
+    repository: ""
+    tag: "latest"
+
+  global:
+    agent:
+      name: "s-harness-codex"
+      description: "Sync tutorial agent driving the unified harness surface via local codex CLI subprocess"
+    replicaCount: 1
+    resources:
+      requests:
+        cpu: "500m"
+        memory: "1Gi"
+      limits:
+        cpu: "1000m"
+        memory: "2Gi"

examples/tutorials/00_sync/harness_codex/project/acp.py

@@ -0,0 +1,175 @@
+"""Sync ACP handler for the Codex CLI harness tutorial.
+
+Demonstrates the ``convert_codex_to_agentex_events`` tap + ``CodexTurn`` +
+``UnifiedEmitter`` for a sync (HTTP-yield) ACP agent.
+
+The handler:
+1. Spawns ``codex exec --json`` as a LOCAL asyncio subprocess (no sandbox).
+   This is correct for tutorials and local development; production isolation
+   is handled by the golden agent's Scale sandbox at
+   ``teams/sgp/agents/golden_agent/project/harness/providers/codex.py``.
+2. Wraps the stdout line stream in a ``CodexTurn``.
+3. Delivers every canonical ``StreamTaskMessage*`` event via
+   ``UnifiedEmitter.yield_turn``, which traces + yields each event back to
+   the HTTP caller in one pass.
+
+Live runs require:
+- ``codex`` CLI on PATH  (``npm install -g @openai/codex``)
+- ``OPENAI_API_KEY`` set in the environment
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import codecs
+import asyncio
+from typing import AsyncGenerator
+from collections.abc import AsyncIterator
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+import agentex.lib.adk as adk
+from agentex.lib.adk import CodexTurn
+from agentex.lib.types.acp import SendMessageParams
+from agentex.lib.core.harness import UnifiedEmitter
+from agentex.lib.types.tracing import SGPTracingProcessorConfig
+from agentex.lib.utils.logging import make_logger
+from agentex.lib.sdk.fastacp.fastacp import FastACP
+from agentex.types.task_message_update import TaskMessageUpdate
+from agentex.types.task_message_content import TaskMessageContent
+from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config
+
+logger = make_logger(__name__)
+
+add_tracing_processor_config(
+    SGPTracingProcessorConfig(
+        sgp_api_key=os.environ.get("SGP_API_KEY", ""),
+        sgp_account_id=os.environ.get("SGP_ACCOUNT_ID", ""),
+        sgp_base_url=os.environ.get("SGP_CLIENT_BASE_URL", ""),
+    )
+)
+
+acp = FastACP.create(acp_type="sync")
+
+MODEL = os.environ.get("CODEX_MODEL", "o4-mini")
+
+
+async def _spawn_codex(model: str) -> asyncio.subprocess.Process:
+    """Spawn ``codex exec --json`` locally and return the live process.
+
+    Injection seam: tests replace this function with a fake that returns a
+    mock process whose stdout yields pre-recorded event lines.
+
+    The flags mirror the golden agent (codex.py in the golden agent repo):
+      --json                      machine-readable newline-delimited events
+      --skip-git-repo-check       safe to run outside a git repo
+      --dangerously-bypass-approvals-and-sandbox
+                                  skip interactive approval prompts in a
+                                  non-interactive (server) context
+      --model <model>             which OpenAI model to use
+
+    The caller writes the prompt to stdin after the process starts, then
+    closes stdin so codex knows input is complete.
+    """
+    cmd = [
+        "codex",
+        "exec",
+        "--json",
+        "--skip-git-repo-check",
+        "--dangerously-bypass-approvals-and-sandbox",
+        "--model",
+        model,
+        "-",  # read prompt from stdin
+    ]
+    return await asyncio.create_subprocess_exec(
+        *cmd,
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+        # Discard stderr: codex --json writes events to stdout; its stderr is
+        # progress/debug noise. Capturing it with PIPE but never reading it
+        # would deadlock once codex fills the OS pipe buffer (~64 KB).
+        stderr=asyncio.subprocess.DEVNULL,
+        env={**os.environ},
+    )
+
+
+async def _process_stdout(process: asyncio.subprocess.Process) -> AsyncIterator[str]:
+    """Yield newline-delimited JSON lines from the process stdout.
+
+    Uses an incremental UTF-8 decoder so a multibyte character split across two
+    4 KB reads is decoded correctly instead of being corrupted at the boundary.
+    """
+    assert process.stdout is not None
+    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+    buffer = ""
+    while True:
+        chunk = await process.stdout.read(4096)
+        if not chunk:
+            break
+        buffer += decoder.decode(chunk)
+        while "\n" in buffer:
+            line, buffer = buffer.split("\n", 1)
+            line = line.strip()
+            if line:
+                yield line
+    buffer += decoder.decode(b"", final=True)
+    if buffer.strip():
+        yield buffer.strip()
+
+
+@acp.on_message_send
+async def handle_message_send(
+    params: SendMessageParams,
+) -> TaskMessageContent | list[TaskMessageContent] | AsyncGenerator[TaskMessageUpdate, None]:
+    """Handle each message by running ``codex exec`` locally and streaming events."""
+    task_id = params.task.id
+    user_message = params.content.content
+    logger.info("Processing message for task %s", task_id)
+
+    start_ms = int(time.monotonic() * 1000)
+
+    async with adk.tracing.span(
+        trace_id=task_id,
+        task_id=task_id,
+        name="message",
+        input={"message": user_message},
+        data={"__span_type__": "AGENT_WORKFLOW"},
+    ) as turn_span:
+        process = await _spawn_codex(MODEL)
+
+        # Write prompt to stdin then close it so codex knows input is done.
+        assert process.stdin is not None
+        process.stdin.write(user_message.encode("utf-8"))
+        await process.stdin.drain()
+        process.stdin.close()
+
+        turn = CodexTurn(
+            events=_process_stdout(process),
+            model=MODEL,
+        )
+
+        emitter = UnifiedEmitter(
+            task_id=task_id,
+            trace_id=task_id,
+            parent_span_id=turn_span.id if turn_span else None,
+        )
+
+        async for event in emitter.yield_turn(turn):
+            yield event
+
+        await process.wait()
+
+        # Record the real wall-clock duration AFTER streaming completes; setting
+        # it before the stream ran would capture only subprocess spawn overhead.
+        turn.duration_ms = int(time.monotonic() * 1000) - start_ms
+
+        if turn_span:
+            usage = turn.usage()
+            turn_span.output = {
+                "model": usage.model,
+                "input_tokens": usage.input_tokens,
+                "output_tokens": usage.output_tokens,
+            }

examples/tutorials/00_sync/harness_codex/pyproject.toml

@@ -0,0 +1,38 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "s-harness-codex"
+version = "0.1.0"
+description = "Sync tutorial agent driving the unified harness surface via local codex CLI subprocess"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "agentex-sdk",
+    "scale-gp",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest",
+    "pytest-asyncio",
+    "httpx",
+    "black",
+    "isort",
+    "flake8",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["project"]
+
+[tool.black]
+line-length = 88
+target-version = ['py312']
+
+[tool.isort]
+profile = "black"
+line_length = 88
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"