JSONL Subprocess Runner Framework

All Untether engine runners (Claude, Codex, OpenCode, Pi; deprecated Gemini and AMP) extend JsonlSubprocessRunner, which manages subprocess lifecycle, JSONL parsing, session locking, and error handling.

Key files

Class hierarchy

Runner (Protocol)
  BaseRunner (SessionLockMixin)
    JsonlSubprocessRunner
      CodexRunner
      OpenCodeRunner (wraps run_impl: refuses OpenCode 2.x before spawning, #970)
      PiRunner
      GeminiRunner, AmpRunner (deprecated, unsupported)
      ClaudeRunner (overrides run_impl for PTY support)
  (each concrete runner also mixes in ResumeTokenMixin)

Template methods to override

When creating a new runner, override these methods on JsonlSubprocessRunner:

class MyRunner(JsonlSubprocessRunner):
    engine: EngineId = "myengine"

    def command(self) -> str:
        """CLI binary name (e.g. 'codex', 'opencode', 'pi')."""

    def build_args(self, prompt, resume, *, state) -> list[str]:
        """Build CLI arguments. Called once per run."""

    def translate(self, data, *, state, resume, found_session) -> list[UntetherEvent]:
        """Translate a decoded JSON dict to Untether events. Core logic."""

    def new_state(self, prompt, resume) -> Any:
        """Create per-run state (e.g. EventFactory, accumulators)."""

    # Optional overrides:
    def stdin_payload(self, prompt, resume, *, state) -> bytes | None:
        """Data to send on stdin. Default: prompt.encode()."""

    def env(self, *, state) -> dict[str, str] | None:
        """Extra environment variables for the subprocess."""

    def decode_jsonl(self, *, line: bytes) -> Any | None:
        """Custom JSON decoder. Default: json.loads."""

Event model (3 events)

Every run emits exactly this sequence:

StartedEvent       — emitted once when session ID is known
ActionEvent(s)     — zero or more, with phase: started → completed
CompletedEvent     — emitted exactly once, always last

StartedEvent

StartedEvent(engine="codex", resume=ResumeToken(engine="codex", value="thread_123"))

ActionEvent

ActionEvent(
    engine="codex",
    action=Action(id="item_5", kind="command", title="ls -la", detail={...}),
    phase="started",  # or "completed"
    ok=True,          # only on completed phase
)

CompletedEvent

CompletedEvent(engine="codex", ok=True, answer="Done.", resume=token, usage={...})

ActionKind values

Session locking

class SessionLockMixin:
    session_locks: WeakValueDictionary[str, anyio.Semaphore]

    def lock_for(self, token: ResumeToken) -> anyio.Semaphore:
        # Key: "engine:session_id"
        # WeakValueDictionary auto-cleans when semaphore has no references

Locking rules:

• Resume runs: acquire lock immediately before spawning subprocess
• New runs: don't know session_id until StartedEvent; acquire lock when first system.init / thread.started / step_start arrives, before yielding the event
• Lock held until run completes (released in finally)
• Serialises concurrent runs on the same session

Process lifecycle (run_impl)

1. new_state(prompt, resume)           → create per-run state
2. build_args(prompt, resume, state)   → construct CLI command
3. stdin_payload(prompt, resume, state) → optional stdin data
   _check_prespawn_ram_guard(...)      → RAM / max_concurrent_engine_runs gate before ANY spawn
                                          (#350/#589; every spawn site incl. ClaudeRunner, pinned by #838)
4. manage_subprocess(cmd, ...)         → spawn with PIPE for stdin/stdout/stderr
   publish_run_stream(stream, pid)     → create JsonlStreamState, publish stream + PID
                                          together to the per-run RunStreamHandle (#510)
5. _send_payload(proc, payload)        → send stdin, close stdin
6. drain_stderr(proc.stderr)           → log stderr concurrently (task group)
7. _iter_jsonl_events(proc.stdout)     → parse JSONL, call translate()
8. proc.wait()                         → wait for exit code
9. stream_end_events() or              → emit CompletedEvent if not already emitted
   process_error_events()

JSONL stream state

@dataclass
class JsonlStreamState:
    expected_session: ResumeToken | None    # from resume arg
    found_session: ResumeToken | None       # from stream
    did_emit_completed: bool                # guard: exactly one CompletedEvent
    ignored_after_completed: bool           # drop lines after CompletedEvent
    jsonl_seq: int                          # line counter for logging
    followup_turns: bool                    # #776: Claude live sessions read past CompletedEvent
    # ... plus activity / stall-diagnostic fields (last_event_type, saw_result, proc_returncode, …)

Key invariants:

• Exactly one CompletedEvent per run — after emitting, all subsequent lines are dropped, unless the runner sets followup_turns = True (Claude live sessions, #776), where later turns become TurnEvent segments
• Per-run stream binding (#510) — runner instances are shared across chats, so runner.current_stream / runner.last_pid are diagnostics only ("latest spawn in any chat"). The bridge binds a RunStreamHandle via ContextVar in run_runner_with_cancel; runners publish into it with publish_run_stream(). Bridge code reads the handle, never the runner attributes
• Session verification — if expected_session is set and stream yields a different session_id, raise RuntimeError
• Duplicate StartedEvent suppression — a repeat StartedEvent without meta is dropped; one carrying meta passes through as a supplementary event (#225, e.g. Pi's late model) and ProgressTracker.note_event merges it

Error handling

Resume token mixin

class ResumeTokenMixin:
    engine: EngineId
    resume_re: re.Pattern[str]  # engine-specific regex

    def extract_resume(text) -> ResumeToken | None  # parse last match
    def format_resume(token) -> str                  # canonical resume line
    def is_resume_line(line) -> bool                 # for stripping from prompts

Resume line formats per engine:

• Claude: `claude --resume <session_id>`
• Codex: `codex resume <thread_id>`
• OpenCode: `opencode --session <ses_XXX>`
• Pi: `pi --session <token>`

Engine backend registration

Each runner module exports a BACKEND:

# src/untether/runners/myengine.py
BACKEND = EngineBackend(
    id="myengine",
    build_runner=_build_runner,
    cli_cmd="myengine",
    install_cmd="pip install myengine",
)

Registered in pyproject.toml:

[project.entry-points."untether.engine_backends"]
codex = "untether.runners.codex:BACKEND"
claude = "untether.runners.claude:BACKEND"
opencode = "untether.runners.opencode:BACKEND"
pi = "untether.runners.pi:BACKEND"
gemini = "untether.runners.gemini:BACKEND"  # deprecated
amp = "untether.runners.amp:BACKEND"        # deprecated

Discovery: importlib.metadata.entry_points(group="untether.engine_backends")

EventFactory

Helper for consistent event creation:

factory = EventFactory(engine="codex")
factory.started(token, title="Codex")
factory.action_started(action_id="item_1", kind="command", title="ls")
factory.action_completed(action_id="item_1", kind="command", title="ls", ok=True)
factory.completed_ok(answer="Done.", resume=token, usage={...})
factory.completed_error(error="timeout", resume=token)

ClaudeRunner PTY exception

ClaudeRunner overrides run_impl entirely because it uses a PTY for the bidirectional control channel instead of standard PIPE stdin. It manages:

• pty.openpty() for stdin (prevents deadlock with persistent stdin)
• Session registries (_SESSION_STDIN, _REQUEST_TO_SESSION) for concurrent sessions
• Control request/response draining after every JSONL line

See .claude/skills/claude-stream-json/SKILL.md for Claude-specific details.