Untether untether-architecture

install

source · Clone the upstream repo

git clone https://github.com/littlebearapps/untether

Claude Code · Install into ~/.claude/skills/

T=$(mktemp -d) && git clone --depth=1 https://github.com/littlebearapps/untether "$T" && mkdir -p ~/.claude/skills && cp -r "$T/.claude/skills/untether-architecture" ~/.claude/skills/littlebearapps-untether-untether-architecture && rm -rf "$T"

manifest: .claude/skills/untether-architecture/SKILL.md

source content

Untether Architecture

Telegram bridge for agent CLIs (Claude Code, Codex, OpenCode, Pi). Control coding agents from anywhere.

Data flow

Telegram Bot API
    |
    v
TelegramClient (httpx, long polling)
    |
    v
telegram/loop.py (parse updates, dispatch commands/callbacks)
    |
    v
handle_message() in runner_bridge.py
    |
    v
Runner.run(prompt, resume) -> AsyncIterator[UntetherEvent]
    |                              |
    v                              v
ProgressEdits <-- on_event() -- StartedEvent / ActionEvent / CompletedEvent
    |
    v
TelegramPresenter.render_progress() / render_final()
    |
    v
TelegramOutbox (coalesced edits, rate-limited sends)
    |
    v
Telegram Bot API

Core abstractions

Runner (Protocol)

class Runner(Protocol):
    engine: str
    def run(self, prompt: str, resume: ResumeToken | None) -> AsyncIterator[UntetherEvent]
    def is_resume_line(self, line: str) -> bool
    def format_resume(self, token: ResumeToken) -> str
    def extract_resume(self, text: str | None) -> ResumeToken | None

UntetherEvent (discriminated union)

type UntetherEvent = StartedEvent | ActionEvent | CompletedEvent

Every run emits:

StartedEvent

(once) ->

ActionEvent

s (zero+) ->

CompletedEvent

(once, always last).

RunnerBridge (

runner_bridge.py

)

Connects runners to the transport layer:

```
handle_message()
```
— entry point for incoming Telegram messages
Creates
```
ProgressTracker
```
and
```
ProgressEdits
```
Spawns runner in a task group with cancel support
Manages progress message lifecycle (create -> edit -> replace with final)
Handles errors, cancellation, ephemeral cleanup

ProgressTracker (

progress.py

)

Aggregates events into renderable state:

tracker = ProgressTracker(engine="claude")
tracker.note_event(evt)           # returns True if state changed
state = tracker.snapshot(
    resume_formatter=runner.format_resume,
    context_line="myproject@main",
)

Snapshot includes: resume line, action list, action count, context line.

ProgressEdits

Live-updates the Telegram progress message:

Signal-based: only renders when new events arrive
Detects approval button transitions for push notifications
Manages
```
_approval_notified
```
flag and
```
_approval_notify_ref
```
```
delete_ephemeral()
```
cleans up notification messages on run completion

TelegramPresenter (

telegram/bridge.py

)

Renders progress and final messages:

render_progress(state, elapsed_s, label)

RenderedMessage

render_final(state, elapsed_s, status, answer)

RenderedMessage

Inline keyboard buttons in
```
extra["reply_markup"]
```

RenderedMessage

@dataclass
class RenderedMessage:
    text: str
    extra: dict[str, Any]  # reply_markup, parse_mode, followups, etc.

Config system

untether.toml

# ~/.untether/untether.toml
default_engine = "claude"
default_project = "untether"

[transports.telegram]
bot_token = "..."
chat_id = -1001234567890
voice_transcription = true
session_mode = "chat"
topics.enabled = true

[claude]
model = "sonnet"
permission_mode = "plan"

[codex]
profile = "Codex"

[projects.untether]
path = "/home/nathan/untether"

Settings hierarchy

UntetherSettings (pydantic-settings, TOML source)
  ├── TransportsSettings
  │     └── TelegramTransportSettings
  │           ├── TelegramTopicsSettings
  │           └── TelegramFilesSettings
  ├── PluginsSettings
  ├── ProjectSettings (per project)
  └── engine_config(engine_id) -> dict  # [claude], [codex], etc.

Config loaded from
```
~/.untether/untether.toml
```
Engine configs in
```
[engine_id]
```
sections (flat) or
```
[engines.engine_id]
```
(nested)
Environment overrides:
```
UNTETHER__
```
prefix with
```
__
```
nesting

ChatPrefsStore

Per-chat persistent preferences (engine, model, reasoning, permission_mode):

class EngineOverrides:
    engine: str | None
    model: str | None
    reasoning: str | None
    permission_mode: str | None

Stored in
```
telegram_chat_prefs_state.json
```
Set via
```
/agent
```
,
```
/model
```
,
```
/reasoning
```
,
```
/planmode
```
commands
Applied at run time to override global config

Engine backend registration

Entry points (

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"

EngineBackend

@dataclass(frozen=True, slots=True)
class EngineBackend:
    id: str
    build_runner: Callable[[EngineConfig, Path], Runner]
    cli_cmd: str | None = None
    install_cmd: str | None = None

Discovery:

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

Command system

Command handlers (

telegram/commands/

)

File	Commands
`dispatch.py`	Callback dispatch, early answering, ephemeral registration
`claude_control.py`	Approve/Deny/Discuss handlers, cooldown wiring
`planmode.py`	`/planmode` toggle
`usage.py`	`/usage` — Claude Code API usage
`model.py`	`/model` override
`reasoning.py`	`/reasoning` override
`trigger.py`	`/trigger` — mentions-only mode
`agent.py`	`/agent` — engine selection

CommandResult

@dataclass
class CommandResult:
    text: str
    parse_mode: str | None = None  # "HTML" for bold formatting

Commands return

CommandResult

; dispatch sends it as a Telegram message.

Callback dispatch

Callback data format:

<prefix>:<action>:<id>

(max 64 bytes).

```
ctrl:approve:<request_id>
```
— approve control request
```
ctrl:deny:<request_id>
```
— deny control request
```
ctrl:discuss:<request_id>
```
— pause & outline plan

Running tasks

RunningTasks = dict[MessageRef, RunningTask]

@dataclass
class RunningTask:
    resume: ResumeToken | None
    resume_ready: anyio.Event
    cancel_requested: anyio.Event
    done: anyio.Event
    context: RunContext | None

Keyed by progress message ref
```
/cancel
```
sets
```
cancel_requested
```
event
```
done
```
event signals run completion for cleanup

Project system

Projects bind a directory + optional branch to a Telegram context:

[projects.untether]
path = "/home/nathan/untether"
default_engine = "claude"
chat_id = -1001234567890   # optional per-project chat

```
/topic <project> @branch
```
creates bound topics
```
/ctx set <project>
```
binds a chat context
Project alias used as directive prefix:
```
/untether fix the bug
```

Trigger system

Triggers let external events or schedules start agent runs automatically. Opt-in via

[triggers] enabled = true

Cron

run_cron_scheduler()

ticks every minute, checking each

[[triggers.crons]]

entry against the current time via

cron_matches()

(5-field standard syntax). Per-cron

timezone

or global

default_timezone

converts UTC to local wall-clock time via

_resolve_now()

zoneinfo.ZoneInfo

. DST transitions handled automatically.

last_fired

dict prevents double-firing within the same minute.

Webhooks

run_webhook_server()

runs an aiohttp server. Each

[[triggers.webhooks]]

maps a URL path to auth (bearer/HMAC-SHA256/SHA1) + prompt template with

{{field.path}}

substitutions. Rate-limited per-webhook and globally.

Dispatch

Both crons and webhooks feed into

TriggerDispatcher.dispatch_cron()

dispatch_webhook()

→ sends a notification message to Telegram (

⏰

⚡

) → calls

run_job()

with the prompt, threading under the notification.

Key files

```
triggers/cron.py
```
— cron parser, timezone-aware scheduler

triggers/settings.py

—

CronConfig

WebhookConfig

TriggersSettings

(pydantic)

```
triggers/dispatcher.py
```
— notification +
```
run_job()
```
bridge
```
triggers/server.py
```
— aiohttp webhook server
```
triggers/auth.py
```
— bearer/HMAC verification
```
triggers/templating.py
```
—
```
{{field.path}}
```
prompt substitution

Key conventions

Python 3.12+, anyio for async, msgspec for JSONL parsing, structlog for logging
pydantic + pydantic-settings for config validation
Ruff for linting, pytest with coverage for tests
Runner backends registered via entry points
All Telegram writes go through the outbox
Exactly one CompletedEvent per run (enforced by JsonlStreamState)

Untether untether-architecture

Untether Architecture

Data flow

Core abstractions

Runner (Protocol)

UntetherEvent (discriminated union)

RunnerBridge (
`runner_bridge.py`
)

ProgressTracker (
`progress.py`
)

ProgressEdits

TelegramPresenter (
`telegram/bridge.py`
)

RenderedMessage

Config system

untether.toml

Settings hierarchy

ChatPrefsStore

Engine backend registration

Entry points (
`pyproject.toml`
)

EngineBackend

Command system

Command handlers (
`telegram/commands/`
)

CommandResult

Callback dispatch

Running tasks

Project system

Trigger system

Cron

Webhooks

Dispatch

Key files

Key conventions

Untether untether-architecture

Untether Architecture

Data flow

Core abstractions

Runner (Protocol)

UntetherEvent (discriminated union)

RunnerBridge (runner_bridge.py)

ProgressTracker (progress.py)

ProgressEdits

TelegramPresenter (telegram/bridge.py)

RenderedMessage

Config system

untether.toml

Settings hierarchy

ChatPrefsStore

Engine backend registration

Entry points (pyproject.toml)

EngineBackend

Command system

Command handlers (telegram/commands/)

CommandResult

Callback dispatch

Running tasks

Project system

Trigger system

Cron

Webhooks

Dispatch

Key files

Key conventions

RunnerBridge (
`runner_bridge.py`
)

ProgressTracker (
`progress.py`
)

TelegramPresenter (
`telegram/bridge.py`
)

Entry points (
`pyproject.toml`
)

Command handlers (
`telegram/commands/`
)