johnandersen777.bsky.social/asyncai

Switch repository

Loading…

# 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](0002-claude-code-session-files.md).

## Components

```mermaid
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`:

```mermaid
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

```mermaid
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.

```mermaid
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.

```mermaid
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](0004-code-mode.md)). 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

```mermaid
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](0002-claude-code-session-files.md) for the format and no-overwrite
rules.

## atirc bridge

```mermaid
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

```text
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