Output Channels

The non-interactive output contract for induscode: a frozen JSON-RPC 2.0 / NDJSON wire surface plus two drivers — oneshot (the print/-p/--json path) and the long-lived bidirectional link server/client — that emit a Conductor's results per CLI mode. Reached as the oneshot and link boot runners, or directly via from induscode.channels import ....

The channels subpackage is how induscode talks to a SessionConductor from outside the interactive console. It defines one declarative contract — a JSON-RPC 2.0 envelope, NDJSON framing, a declarative operation registry, ask/tell dialog primitives, and a flat wire snapshot — and two channels written against it. The oneshot channel runs prompts to settlement and writes either clean final text or a streamed NDJSON event log; the link channel is a long-lived JSON-RPC server plus a generated client that drives a child process over NDJSON. The protocol is a single op registry consumed by both the server (data-driven dispatch) and the driver (a generated client whose method set is exactly the registry keys), with transport, framer, conductor, and dialog all injected so tests run over in-memory pipes with no real stdio.

Table of Contents

• Reaching the Channels
• The Oneshot Channel (print mode)
• The Link Channel (NDJSON wire)
• The Wire Contract
• The Operation Registry
• Session Ops and the Snapshot
• NDJSON Framing
• The Dialog Bridge
• Public API
• Design Notes
• Source Files

Reaching the Channels

The two channels back two of the launcher's runner modes. Boot resolves a RunnerMode from your flags and dispatches to the matching runner:

The oneshot runner builds a ChannelContext over a stdout WritableLine + ndjson_framer + inert_dialog and calls run_oneshot. The output shape is NDJSON when the invocation carries a json flag (-p --json), otherwise clean text. The link runner serves SESSION_OPS over a stdin/stdout LinkServerIo via create_link_server and awaits server.done.

channels owns no session or agent logic of its own — every operation delegates to the conductor on the context, and the two channels just emit a conductor's results. See Conductor for the session loop and Sessions for persistence. Signals and usage ultimately originate from the underlying framework agent loop, but a channel only ever switches on the conductor's own typed signal vocabulary, never a raw framework event.

The Oneshot Channel (print mode)

run_oneshot(ctx, OneshotRequest) runs one non-interactive request to settlement and returns a process exit code (0 clean / 1 faulted). The output shape is not a branch through the runner body — it is a strategy object selected by lookup. Each OneshotStrategy supplies an optional on_start, an optional on_signal, and a mandatory finish(state, ctx) -> exit_code. The runner is shape-agnostic: it subscribes the strategy to conductor.subscribe(...), runs on_start, submits each prompt in order via conductor.submit(...), hands the final ConductorState to finish, and always tears the subscription down.

The text shape emits the answer once, clean — so a caller capturing stdout gets the final answer rather than a token-by-token dribble. Thinking deltas, tool frames, and bookkeeping signals are ignored. The ndjson shape streams every conductor signal as one framed line; its end frame omits fault when None, matching how JSON.stringify dropped undefined fields.

# clean final text (text shape)
pindus -p "summarize this repo"

# streamed NDJSON event log (ndjson shape: the json flag selects it)
pindus -p --json "summarize this repo"
# emits {"type":"signal","name":"start","body":{}}
#       {"type":"signal","name":"text","body":{"kind":"text","delta":"…"}}
#       … {"type":"signal","name":"end","body":{"phase":"…","usage":{…}}}

Driven directly from Python:

from induscode.channels import (
    ChannelContext, OneshotRequest, ndjson_framer, inert_dialog, run_oneshot,
)

ctx = ChannelContext(
    conductor=my_conductor,
    out=my_writable_line,
    framer=ndjson_framer,
    dialog=inert_dialog,
)
exit_code = await run_oneshot(ctx, OneshotRequest(shape="text", prompts=("fix the failing test",)))
# 0 on a clean settlement, 1 if state.phase == "faulted"

inert_dialog is the shared DialogBridge for headless runs: every ask() resolves immediately with the caller's fallback and every tell() is dropped, so the agent never wedges on a prompt with no driver attached. An OneshotRequest carries shape, an ordered tuple of prompts, and optional images (framework indusagi.ai.ImageContent values) riding the first prompt.

The Link Channel (NDJSON wire)

The link channel is a long-lived, bidirectional JSON-RPC 2.0 conversation. The link_runner serves SESSION_OPS over stdin/stdout; a driving parent process writes framed requests and reads framed replies:

-> {"jsonrpc":"2.0","id":"lnk-…","method":"submit","params":{"input":"hi"}}
<- {"jsonrpc":"2.0","id":"lnk-…","result":{"model":"…","streaming":false,"sessionId":"…","usage":{…}}}

Server

create_link_server(registry, conductor, io) starts a data-driven server. It owns a single async for reader over io.in_ (via decode_lines) and classifies each frame:

• an answer frame goes to dialog.deliver(...), settling a suspended ask;
• a request frame (with a method) is dispatched as a concurrent asyncio task; a request carrying an id gets exactly one correlated reply (result or error), a notification (no id) is dispatched for effect and never replied to;
• any other frame is a forward-compatible no-op.

Requests are dispatched as concurrent tasks (tracked in a set, gathered at stream end) rather than awaited in line. This is critical: a handler may itself suspend on a dialog ask whose answer arrives as a later frame on this same reader stream — awaiting in line would deadlock the reader. The returned LinkServer exposes done (the pump task, settles at stream end) and the LinkedDialogBridge it feeds. LinkServerIo is the injectable transport pair: in_ (framed inbound, stdin in prod) and out (framed outbound sink, stdout).

Driver

create_link_driver(io, ...) is the mirror — a generated client. LinkClient.__getattr__ yields a request thunk (params=None) -> awaitable result for any attribute, so the client mirrors whatever ops the server dispatches with a single body (no hand-written per-op methods; underscore names raise AttributeError). Each thunk mints an lnk--prefixed id, frames {jsonrpc,id,method,params?}, and awaits a future settled by the reply reader. The reader routes inbound frames three ways: replies settle pending requests, signals fan to on_signal, asks are handed to on_ask (which posts an answer back). Must run on a live event loop.

from induscode.channels.link import create_link_driver, LinkDriverIo

handle = create_link_driver(
    LinkDriverIo(out=child_stdin, in_=child_stdout),
    on_signal=lambda s: print(s.name),
)
snap = await handle.client.submit({"input": "hello"})   # any attribute -> a request thunk
models = await handle.client.listModels()               # mirrors SESSION_OPS keys for free
handle.close()                                          # rejects every still-pending request

LinkDriverHandle carries client, a done task (resolves at stream end), and close(reason) (rejects every pending request and stops accepting replies). A failed reply is raised as a LinkRequestError carrying the wire code and structured data. When on_ask is omitted the driver dismisses every ask (posts None), so a non-interactive driver never wedges the server.

The Wire Contract

contract.py is the frozen type surface — only shapes and inert pure helpers, no I/O. The envelope is JSON-RPC 2.0: a request is {jsonrpc, id?, method, params?}, a reply is {jsonrpc, id, result} or {jsonrpc, id, error}. Decoded frames are plain dicts; is_reply_ok(reply) discriminates the arms (True iff result is present). Every envelope is stamped with PROTOCOL_VERSION = "2.0" and the driver stamps outgoing ids with REQUEST_ID_PREFIX = "lnk-".

OP_ERROR pins the closed JSON-RPC error code set:

Typed frozen frames cross Python API boundaries: Ask (type="ask", blocking agent→driver request), AskAnswer (type="answer", correlated reply), Tell (type="tell", one-way notice), and Signal (type="signal", uncorrelated server→driver event). An OpError(code, message, data?) frames as {code, message, data?}. mint_wire_id(prefix, now_ms, seq) produces a process-unique <prefix><now36>-<seq36> correlation id. ChannelTimings carries millisecond timeouts — startupMs=1500, shutdownMs=1200, requestMs=45000, dialogMs=90000 — and resolve_timings(overrides) merges a partial override onto the frozen DEFAULT_CHANNEL_TIMINGS (asyncio call sites divide by 1000).

WritableLine (a Protocol with one write(chunk: str) method) and ReadableChunks (AsyncIterable[str | bytes]) are the minimal injectable transport surfaces. ChannelContext bundles conductor, out, framer, and dialog, and is handed to every Op.handle and shared by both channels — fully injected so tests use a fake conductor and a list-backed sink.

The Operation Registry

The protocol is one declarative operation registry, not a hand-written dispatch ladder mirrored by hand-written client methods. An Op binds a wire method name to a typed async handle(params, ctx) -> result. An OpRegistry is Mapping[str, Op] whose keys equal each op's method, so server dispatch is one lookup and the driver's method set is exactly the key set.

• define_ops(record) freezes a record into a read-only OpRegistry (MappingProxyType) — the only sanctioned way to mint a registry.
• build_ops(record) returns an OpRegistryHandle: the frozen registry, its sorted methods tuple, and a bound dispatch closure. The server-side dispatch and the client-side method set can never drift.
• dispatch(registry, method, params, ctx) is the lone server dispatch path — a map lookup then await op.handle(...); raises UnknownOpError (carrying -32601 and the offending method) when unregistered, and propagates handler exceptions.
• has_op(registry, method) and methods_of(registry) round out the helpers.

Adding a row to a registry adds both a server-handled method and a callable driver method for free:

from induscode.channels import Op, build_ops, create_link_server
from induscode.channels.link import LinkServerIo

async def ping(params, ctx):
    return {"pong": True}

registry = build_ops({"ping": Op(method="ping", handle=ping)}).registry
server = create_link_server(registry, my_conductor, LinkServerIo(in_=stdin_chunks, out=stdout))
await server.done   # resolves when the inbound stream ends

Session Ops and the Snapshot

SESSION_OPS is the concrete frozen registry the link serves, each op delegating to ctx.conductor:

Method names and snapshot field casing are kept verbatim (submit, listModels, cycleModel, sessionId, autoCondense, …) for cross-host compatibility — a foreign driver must speak to a Python server and vice versa. listModels currently returns only the single active model entry, and cycleModel probes for the conductor's optional cycle_model method; both are minimal projections of what the conductor exposes.

project_snapshot(state, ...) reshapes a ConductorState into the flat wire LinkSnapshot dict whose keys are exactly LINK_SNAPSHOT_FIELDS:

("model", "thinking", "streaming", "condensing", "faulted",
 "sessionId", "sessionFile", "autoCondense", "messageCount", "queuedCount", "usage")

Busy flags are derived from the coarse phase (streaming is phase == "streaming" or "tooling", condensing and faulted from their phases). thinking defaults to DEFAULT_THINKING = "off" when the snapshot carries no reasoning effort. usage rides through the single app-wide codec. sessionFile is omitted (not nulled) when absent, exactly as JSON.stringify dropped the undefined field. The snapshot is the link's own vocabulary — a remote driver mirrors session state without holding a live conductor.

NDJSON Framing

Framing is one JSON value per line. framer.py implements the NdjsonFramer shape — a frozen (encode_line, decode_lines) pair bundled as the canonical ndjson_framer.

encode_line(value) serializes with json.dumps (compact (",", ":") separators, ensure_ascii=False), escapes U+2028 (LINE SEPARATOR) and U+2029 (PARAGRAPH SEPARATOR) to their \u… forms, and appends exactly one \n. Those two code points are legal inside a JSON string but break a naive line splitter, and json.dumps leaves them raw — so the explicit escape guarantees any value round-trips through the decoder intact.

decode_lines(stream) is a pull-model async generator: it splits strictly on \n, decodes byte chunks through a fresh per-call incremental UTF-8 decoder (so a rune split across chunk boundaries is reassembled), skips empty/whitespace lines, and emits a final unterminated frame at end-of-stream if it carries content. The server, the driver, and the oneshot NDJSON shape all share this one implementation; an embedder may inject an alternate framer via the driver/server options.

The Dialog Bridge

The whole dialog surface the agent needs — multiple-choice, confirm, free-text, editor sessions, notifications, status, titles — reduces to two generic primitives on the DialogBridge protocol: ask(kind, payload, fallback) (a round-trip that suspends for a correlated answer or resolves the fallback after dialogMs) and tell(kind, payload) (fire-and-forget). Concrete methods are rows in DIALOG_TABLE, not bespoke bodies:

make_dialog_methods(bridge) closes the table over one bridge into named convenience thunks (generated from the table, no per-method choreography). On the link, create_dialog_bridge(deps) builds a LinkedDialogBridge whose pending answers are fed externally — the server owns the single reader and routes inbound answers to deliver(). ask() emits an ask--prefixed frame and arms a dialogMs fallback timer; deliver() settles the matching future (a no-op for late/duplicate answers so a misbehaving driver cannot throw into the read loop); tell() fires and returns; drain() settles all pending with None on shutdown. This is the seam that drives the interactive Dialogs when a console is attached.

Public API

All names are importable from induscode.channels; the link half is also under induscode.channels.link.

Design Notes

• Single declarative registry. SESSION_OPS drives both server dispatch (a map lookup, no switch) and the driver's method set (LinkClient is generated via __getattr__, never hand-written). An unregistered name simply rejects with unknownOp (-32601).
• Concurrency. The server dispatches requests as concurrent asyncio tasks, not awaited in line, because a handler can suspend on a dialog ask whose answer arrives as a later frame on the same reader stream; awaiting in line would deadlock. Answer frames are always delivered immediately.
• Framer correctness. U+2028/U+2029 are escaped because json.dumps leaves them raw yet they break line splitters; each decode_lines call uses a fresh incremental UTF-8 decoder so a rune split across byte chunks is reassembled.
• Resilience. inert_dialog and the driver's missing on_ask both dismiss every ask (return fallback / post None), so a headless run or disconnected driver never wedges the agent. LinkedDialogBridge.drain() and the driver's reject-all settle/reject every pending entry on shutdown; deliver() is a no-op for late/duplicate answers.
• One serializer. signal_to_wire is the single Python-specific divergence: it re-attaches the kind ClassVar discriminant that the app-wide message_to_dict codec does not emit. No second serializer is introduced; all field payloads still route through the one codec.
• Verbatim casing & dropped fields. Wire field names/casing are pinned for cross-host compatibility; project_snapshot omits sessionFile when absent and the NDJSON end frame omits fault when None, matching how JSON.stringify dropped undefined fields.

Source Files

The runners that wire these into CLI modes live in boot/runners/oneshot_runner.py and boot/runners/link_runner.py — see Boot. For the full export map see Package Exports.