Part 12 · Channels — how you talk to the agent

Level: ⭐⭐⭐ · Time: ~4 h · Requires: parts 2 (gateway, trust), 10 (sandbox) Produces: a channel adapter, pairing, allowlists, agent routing, streaming

---

0. A channel is a trust boundary, not an input

The most common mistake: treating a channel as "a way of delivering text". A channel brings a trust level with it — and that is its most important property.

console on your machine      →  You, authenticated by the operating system
HTTP from the loopback       →  a process on this machine — not necessarily you
private message from a messenger  →  ⚠️ someone who knows your number
group                        →  ⚠️⚠️ many people, including strangers
webhook                      →  ⚠️⚠️ anyone who guessed the URL

OpenClaw states this as a design principle: an incoming message is untrusted. Channels with private messages require pairing approval before anything even reaches the agent.

⭐ In part 2 you built the mechanism (poziom_zaufania(zrodlo)). Here you fill it with content.

---

1. Anatomy of an adapter

Every channel does exactly four things. Nothing more — the agent logic is shared.

# $OSA/kanaly/bazowy.py
class Kanal:
    nazwa: str
    zaufanie_domyslne: str = "piaskownica"    # P66 — the narrowest

    def uwierzytelnij(self) -> None:
        """1. Connection and identity (token, QR, Unix socket)."""

    def normalizuj(self, surowe) -> dict:
        """2. Channel format → uniform message."""
        return {"tresc": ..., "nadawca": ..., "watek": ...,
                "zalaczniki": [], "zrodlo": self.nazwa}

    def wolno(self, wiadomosc: dict) -> tuple[bool, str]:
        """3. Access control. ⭐ Security is decided here."""
        return False, "brak reguły"

    def wyslij(self, watek, tresc: str) -> None:
        """4. Outbound formatting and splitting."""

⚠️ zaufanie_domyslne = "piaskownica" in the base class is deliberate. A new channel that you forget about in the configuration gets the narrowest permissions. This is P66 expressed in the type, not in a comment.

---

2. Access control — four gates in sequence

def wolno(self, w: dict) -> tuple[bool, str]:
    # 1. Allowlist — cheapest, first
    if self.k["lista_dozwolonych"] and w["nadawca"] not in self.k["lista_dozwolonych"]:
        if self.k["polityka_nieznanych"] == "parowanie":
            return False, self._rozpocznij_parowanie(w["nadawca"])
        return False, ""                      # ⭐ silently — see P110

    # 2. A group requires a mention
    if w.get("grupa") and self.k["grupy"].get("wymaga_wzmianki", True):
        if f"@{self.k['moja_nazwa']}" not in w["tresc"]:
            return False, ""

    # 3. Rate limit per sender
    if not self._limit_ok(w["nadawca"]):
        return False, "Too many messages. Try again in a minute."

    # 4. Size
    if len(w["tresc"]) > self.k.get("max_znakow", 8000):
        return False, f"Message too long (limit {self.k['max_znakow']} characters)."

    return True, ""

2.1 · Pairing

def _rozpocznij_parowanie(self, nadawca) -> str:
    kod = secrets.token_hex(3).upper()
    self.oczekujace[kod] = {"nadawca": nadawca, "kiedy": TERAZ()}
    print(f"[PAROWANIE] {self.nazwa}: {nadawca} → code {kod}\n"
          f"  approve: sesja-cli.py parowanie zatwierdz {kod}", file=sys.stderr)
    return (f"I don't know you. Your pairing code: {kod}\n"
            f"Ask the owner to approve it.")

⭐ Approval happens outside the channel — on the machine, in a terminal. This matters: if pairing could be approved by a message, the channel would pair itself. Access control cannot run through the channel it controls.

---

3. Three channels for your system

3.1 · The Unix socket — the safest channel you have

# $OSA/kanaly/gniazdo.py
"""Unix socket: the file system does the authentication. Zero configuration."""
import socket, os, json, pathlib

SCIEZKA = pathlib.Path("~/.local/run/osa.sock").expanduser()

class KanalGniazdo(Kanal):
    nazwa, zaufanie_domyslne = "gniazdo", "host"

    def uwierzytelnij(self):
        SCIEZKA.parent.mkdir(parents=True, exist_ok=True)
        SCIEZKA.unlink(missing_ok=True)
        self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.s.bind(str(SCIEZKA))
        os.chmod(SCIEZKA, 0o600)          # ⭐ only you — this is ALL the authentication
        self.s.listen(4)

    def wolno(self, w): return True, ""   # the file permissions have already decided

⭐ chmod 600 on a socket is full authentication in one line. No tokens, passwords or rotation — the kernel enforces it better than any code you would write. For a local channel this is the right answer, and it is a pity it is so rarely used.

3.2 · HTTP — when you need a token

The loopback is not enough as authentication: any process on the machine (including code in the sandbox, if it has network access) can connect.

def wolno(self, w):
    podany = w.get("naglowki", {}).get("X-OSA-Token", "")
    if not secrets.compare_digest(podany, self.token):    # ⚠️ P111
        return False, ""
    return True, ""

⚠️ secrets.compare_digest, not ==. An == comparison stops at the first difference and leaks the length of the common prefix through execution time. With a local attacker this is practically exploitable.

3.3 · Sending — splitting and formatting

def podziel(tresc: str, limit: int) -> list[str]:
    """Splits by paragraphs, then sentences. NEVER in the middle of a code block."""
    if len(tresc) <= limit:
        return [tresc]
    czesci, biezaca, w_kodzie = [], "", False
    for akapit in tresc.split("\n\n"):
        if akapit.count("```") % 2:
            w_kodzie = not w_kodzie
        if len(biezaca) + len(akapit) > limit and biezaca and not w_kodzie:
            czesci.append(biezaca.rstrip()); biezaca = ""
        biezaca += akapit + "\n\n"
    if biezaca.strip():
        czesci.append(biezaca.rstrip())
    return [f"({i+1}/{len(czesci)}) {c}" if len(czesci) > 1 else c
            for i, c in enumerate(czesci)]

⚠️ The not w_kodzie condition is the one everyone forgets. A code block cut in half means two messages with an odd number of markers — the messenger will render the rest of the conversation as code.

---

4. Routing — the channel chooses the agent

OpenClaw lets you map a channel and a conversation partner to a separate agent with its own workspace, model and sandbox. The pattern is good and cheap.

{
  "routing": [
    {"gdy": {"zrodlo": "gniazdo"},                "agent": "archiwista", "zaufanie": "host"},
    {"gdy": {"zrodlo": "http", "sciezka": "/skan"},"agent": "archiwista", "zaufanie": "piaskownica"},
    {"gdy": {"zrodlo": "telegram", "dm": true},    "agent": "gospodarz",  "zaufanie": "piaskownica"},
    {"gdy": {"zrodlo": "telegram", "grupa": true}, "agent": "gospodarz",  "zaufanie": "piaskownica",
     "narzedzia": ["przypomnij", "kalendarz_czytaj"]},
    {"gdy": {},                                    "agent": "archiwista", "zaufanie": "piaskownica"}
  ]
}

⭐ The fourth rule shows a mechanism worth noticing: the same agent, a narrower set of tools. Gospodarz in a group is the same AGENTS.md, the same memory — but two tools instead of ten. You do not need a separate agent to restrict permissions; you need the filter from part 7.

def rozstrzygnij(wiadomosc: dict, reguly: list) -> dict:
    for r in reguly:
        if all(wiadomosc.get(k) == v for k, v in r["gdy"].items()):
            return r
    return {"agent": "archiwista", "zaufanie": "piaskownica"}   # P66

---

5. Streaming

async def strumien_do_komunikatora(kanal, watek, zrodlo, min_odstep=1.0):
    """Editing the message instead of tokens. Less often than once a second — API limits."""
    bufor, id_wiad, ostatnia = "", None, 0
    async for token in zrodlo:
        bufor += token
        if time.time() - ostatnia >= min_odstep:
            id_wiad = (await kanal.edytuj(watek, id_wiad, bufor + " ▋")
                       if id_wiad else await kanal.wyslij(watek, bufor + " ▋"))
            ostatnia = time.time()
    await kanal.edytuj(watek, id_wiad, bufor)      # no cursor at the end

💡 Streaming is a feature of perceived speed, not real speed. With a TTFT under a second (part 8 of course 1) it gives less than it seems. Do it for channels where it is free, and do not fight for it in a messenger.

---

6. EXPERIMENT 12 — two channels, two levels

Hypothesis: the same request through two channels produces different behavior, because the trust level differs — and the difference is enforced, not declared.

mkdir -p $OSA/pomiary/kanaly

⭐ Row 2 is the most important. The correct response over HTTP is not "I'm not allowed" — it is the behavior of an agent that does not know such a tool exists, because the filter from part 7 removed it from the schema. A refusal reveals the existence of a capability; a missing tool reveals nothing.

6.1 · Additional measurement — pairing and rate

python3 testy/kanaly.py --atak parowanie   # 50 messages from an unknown sender
python3 testy/kanaly.py --atak tempo       # 100 messages in 10 s from a known one
python3 testy/kanaly.py --atak token       # 1000 attempts to guess the HTTP token

Expected: the unknown sender gets one pairing code (not 50); the known one hits the rate limit; token attempts are rejected in constant time (check the spread of timings — if it grows with the length of the common prefix, you have P111).

---

7. EXERCISE

1. Implement KanalGniazdo (trust host) and KanalHTTP (piaskownica, token).
2. Hook up routing from configuration; the default rule is the narrowest.
3. Add pairing with approval outside the channel.
4. Implement podziel() and test it on a response with a code block in the middle.
5. Run the table from §6 and record the results.

python3 sesja-cli.py kanaly
# gniazdo  ✓ ~/.local/run/osa.sock (0600)  trust: host          today: 34 msgs
# http     ✓ 127.0.0.1:18789 (token)       trust: piaskownica   today: 8 msgs, 2 rejected
# telegram ✗ disabled
#
# pending pairings: 1
#   A3F9C1  +48***1234  (2026-09-23 14:02)

6. Bonus task: add a third channel — a webhook for the scanner. Think about what trust to give it, given that the content comes from a file that someone put in the directory. (Hint: the same as the Researcher.)

---

8. Pitfalls of this part

---

9. Checkpoint

cd $OSA
stat -c '%a' ~/.local/run/osa.sock | grep -q 600 && echo "✓ socket 0600"
python3 - <<'EOF'
from kanaly.bazowy import Kanal
from routing import rozstrzygnij
assert Kanal.zaufanie_domyslne == "piaskownica", "P66"
assert rozstrzygnij({"zrodlo":"nieznany"}, [])["zaufanie"] == "piaskownica"
from kanaly.podzial import podziel
t = "tekst\n\n```python\nx=1\ny=2\n```\n\nkoniec"
assert all(c.count("```") % 2 == 0 for c in podziel(t, 30)), "P112"
print("✓ default trust, routing, code splitting")
EOF
python3 testy/kanaly.py --atak token --sprawdz-czas && echo "✓ constant-time comparison"
git add -A && git commit -m "part 12: channels, pairing, routing, streaming"

• A channel carries a trust level; the default is the narrowest
• Four access-control gates, in order from the cheapest
• Pairing approved outside the channel
• The Unix socket authenticates with file permissions
• The token is compared in constant time
• Routing maps a channel to an agent and a set of tools
• Splitting never cuts a code block
• ⭐ Six attempts through two channels produce different, enforced behavior

---

Next: 13-MULTI-AGENT-FLEET.md — three agents start cooperating, and you learn when that is worse than one.