0001 — asyncai architecture

Status: accepted

Date: 2026-11-02

Context

asyncai is a Rust/tokio port of the core event loop in sshai/src/sshai/agi.py. It keeps the stream-driven runtime: typed actions and events, a dynamic task multiplexer, an agent runtime reactor, and a top-level orchestrator with waiting rules. The policy engine, Caddy, tmux, and SSE MCP transport are intentionally out of scope.

Claude Code session persistence and harness abstraction are covered by 0002.

Components

flowchart LR
    Input["stdin / atirc sub"] --> Orch["Orchestrator"]
    Orch -->|actions| Runtime["AgentRuntime"]
    Runtime --> Backend["AgentBackend"]
    Backend --> Model["OpenAI-compatible endpoint"]
    Backend -->|tool calls| Tools["ToolExecutor"]
    Tools --> MCP["stdio MCP servers"]
    Runtime -->|events| Orch
    Orch --> Out["NDJSON stdout"]
    Orch --> Session["SessionLogger"]
    Backend -->|ToolObserver| Session
    Session --> Harness["Session harnesses"]

Component	Responsibility
Orchestrator	Multiplexes input, waiting rules and agent events; owns routing state and NDJSON output
AgentRuntime	Owns agent/thread state, run lifecycle, message history and queued runs
AgentBackend	Model protocol seam; streams messages and a final outcome
ToolExecutor	Lists and calls tools; stdio MCP implementation uses rmcp
SessionLogger	Fans transcript events out to one or more session harnesses
WorkSet	Dynamic first-completed task multiplexer over tokio::task::JoinSet
WaitingEngine	Reactive rules linking input to actions once prerequisite events are seen
Notice stream	Background inputs (MCP watch notices, ADR 0005; the prompt's On Start section) handled like input; coalesce with active runs

Concurrency model

Single tokio runtime. Two nested reactors share WorkSet:

flowchart TD
    M["Orchestrator reactor"] --> A["input next"]
    M --> N["notice next"]
    M --> B["waiting rule next"]
    M --> C["agent event next"]
    M --> D["AgentRuntime task"]

    R["AgentRuntime reactor"] --> E["action next"]
    R --> F["run completion"]
    R --> G["run message next"]

WorkSet wraps a JoinSet. Callers consume completed futures one at a time and may spawn new futures between iterations. Dropping the set aborts remaining tasks, so reactors are structured and leave no orphan tasks.

Async iterators are pumped with one outstanding __anext__-style pull. A stop sentinel converts normal stream exhaustion into a value so it does not become a task error.

Orchestrator flow

flowchart TD
    In["input line"] --> Rules["WaitingEngine"]
    Rules -->|NEW_THREAD / ADD_MESSAGE / RUN_THREAD| Actions["action channel"]
    Actions --> Runtime["AgentRuntime"]
    Runtime --> Backend["AgentBackend"]
    Backend --> Events["AGIEvent stream"]
    Events --> State["agent/thread state"]
    State --> Rules
    Events --> NDJSON["NDJSON output"]

Waiting rules:

• agent created (NewAgentCreated or ExistingAgentRetrieved) → NewThread
• thread created → AddMessage with the user text
• thread message added → RunThread

Notices (Notice { source, text }, main.system.notice_stream) take the same rules as input. A notice that arrives during a run is appended and sets queued_run, so N notices during a run cause one follow-up run.

A prompt's On Start section (a heading of any level named On Start, up to the next heading of the same or higher level) is removed from the system prompt and sent as the first notice (prompt:on-start), with any watch notices already queued at startup (the --watch-state restore notice) folded in, so startup is one turn. Sessions record it as a monitor event, never a user record. Replies follow the notice rules: atirc posts to the first channel; NO_REPLY posts nothing.

WaitingEngine records event types it has seen. Rules fire when their EventSpec matches. ThreadMessageAdded is forgotten on each new input so the next turn waits for the next message. new_thread_requested prevents buffered inputs from creating duplicate threads.

Thread lifecycle

AgentRuntime is the source of truth for conversation history. The orchestrator stores only routing metadata: agent, active thread, most recent run ID/status.

flowchart TD
    New["NewThread"] --> Empty["thread.messages = []"]
    Empty --> Active["AddMessage appends user message"]
    Active --> Running["RunThread snapshots messages"]
    Running -->|queued_run while active| Running
    Running -->|complete| Active

Run sequence:

1. RunThread snapshots thread.messages, records message_count, emits NewThreadRunCreated, and spawns the backend run plus a message receiver.
2. Backend events become NewThreadMessage events.
3. Finalization waits for both the backend outcome and end of the message stream.
4. Messages appended while the run was active are preserved. The assistant result is inserted before those messages so the newest user message remains last: prefix + assistant tail + appended user messages.
5. ThreadRunComplete / ThreadRunFailed is emitted. If queued_run is set, the next run starts with the full history.

CheckThread reports the active run or the most recent status.

Backend and tool loop

OpenAiCompatibleBackend speaks /v1/chat/completions. Tool calling uses the OpenAI tools / tool_calls protocol with a bounded loop.

sequenceDiagram
    participant R as AgentRuntime
    participant B as Backend
    participant M as Model endpoint
    participant T as ToolExecutor / MCP

    R->>B: start_run(messages)
    B->>M: chat(tools)
    M-->>B: tool_calls
    B->>T: list_tools / call_tool
    T-->>B: tool result
    B->>M: chat(tool results)
    M-->>B: assistant text
    B-->>R: Message + Completed

• With --mcp-stdio, the executor is CodeModeToolExecutor (ADR 0004). The model sees one tool, run_code, and calls MCP tools from Python executed by Monty. One run_code round can fan out many MCP calls and return a filtered result.
• ToolExecutor::run_scope is called once per run, so code-mode REPL state lives for one turn.
• Tool definitions are listed from the executor at run start.
• Tool calls/results are appended as role: tool messages.
• Maximum eight tool rounds. The ninth request withdraws the tools and adds a "tool budget exhausted" note as a user message (chat templates such as llama.cpp's reject a system message after the first position), so the model answers with what it has. A tool call in that response is a backend error.
• Tool failures become error text returned to the model.
• ToolObserver notifies the session layer about each call and result.
• MockBackend provides deterministic replies, delays and injected failures.

Session fanout

flowchart LR
    Events["TranscriptEvent"] --> Logger["SessionLogger"]
    Logger --> Claude["claude-code-cli harness"]
    Logger --> Future["future harnesses"]

SessionLogger owns a single writer task; every configured harness receives the same ordered event stream. Each harness allocates its own UUID and owns its own file. Monitor commands feed MonitorEvent / MonitorConsumed into the same stream; the Claude harness expands them to queue-operation:enqueue, queued_command attachment, and queue-operation:remove. See 0002 for the format and no-overwrite rules.

atirc bridge

flowchart LR
    Lib["atirc Rust client"] -->|connected event| Name["agent name / identities"]
    Lib -->|message event| Mode["run_atirc"]
    Mode -->|messages| Orch["Orchestrator input"]
    Orch --> Runtime["AgentRuntime"]
    Runtime -->|assistant event| Pub["atirc client publish"]
    Pub --> Channel["channel"]

• asyncai atirc uses the atirc crate directly. No atirc sub stdin and no atirc pub subprocess.
• Flags mirror atirc sub: -t/--token, repeatable -c/--channel, --cursor.
• Token: --token, ATIRC_TOKEN, or auto-mint from ATP_USERNAME + ATP_PASSWORD with --agent-name as nick.
• PDS: ATIRC_PDS or derived from ATP_USERNAME.
• Connected event supplies agent name and self identities; --agent-name overrides.
• Default handles every non-self message; --mention-only narrows to nick mentions. Own identity is filtered: server/library can echo published messages back to the subscriber.
• Message text publishes without a nick prefix. Replies are split into IRC-sized posts: every non-empty line publishes separately, and lines over 400 bytes wrap at word boundaries. Blank lines are dropped.
• --max-turns bounds reply loops.
• atirc messages are serialized as monitor/task-notification records with a synthetic atirc sub ... description. --no-stdin gives interactive monitor-only mode with no user records.

Package tree

src/
  actions.rs       actions and payloads
  events.rs        events, EventType, payloads
  message.rs       roles, content, messages, run status
  state.rs         orchestrator agent/thread records
  ids.rs           deterministic IDs
  kv.rs            KVStore trait + memory store
  concurrent.rs    WorkSet multiplexer
  waiting.rs       EventSpec + WaitingEngine
  backend.rs       AgentBackend trait, mock backend
  runtime.rs       AgentRuntime
  orchestrator.rs  Orchestrator
  atirc.rs         atirc protocol and publish bridge
  tools.rs         ToolExecutor / ToolObserver traits
  resources.rs     MCP resources + watch event extraction (ADR 0005)
  code_mode/       run_code tool (ADR 0004)
    mod.rs         executor, host loop, timeouts, audit, mutating warns
    catalog.rs     MCP schemas -> Python stubs, Returns: sketches
    dispatch.rs    shared kwarg / budget validation
    protocol.rs    Job / HostEvent protocol both sandboxes speak
    engine.rs      in-process sandbox (engine thread)
    pool.rs        monty-pool sandbox (worker subprocess, heap cap)
    convert.rs     JSON <-> Monty values
    helpers.py     items / names / dig / count_by / to_json
    scope.rs       module-level names: reserved helpers, variables note
    builtins.rs    list_resources / read_resource / watch_* (ADR 0005)
    watch.rs       WatchManager: subscriptions, filters, dedupe, debounce, --watch-state
  log.rs           NDJSON stderr logging (all stderr output)
  mcp/             rmcp client, simple HTTP GET server, kcp servers
  session/         SessionLogger + harness tree
  cli.rs           clap CLI
  main.rs          entrypoint
prompts/           editable system prompts
docs/adrs/         architecture and session decisions
tests/             unit and integration coverage

Explicitly not ported

• policy engine workflow execution and GitHub Actions
• Caddy admin/reverse-proxy management
• tmux attachment, Unix-socket input/output, SSH daemon
• OpenAI Agents SDK integration; backend trait is the seam
• keyring persistence; MemoryKvStore implements the trait