Część 12 · Kanały — jak z agentem rozmawiasz

Poziom: ⭐⭐⭐ · Czas: ~4 h · Wymaga: części 2 (brama, zaufanie), 10 (piaskownica) Produkuje: adapter kanału, parowanie, listy dozwolonych, routing agentów, strumieniowanie

---

0. Kanał to granica zaufania, nie wejście

Najczęstszy błąd: traktowanie kanału jako „sposobu dostarczenia tekstu". Kanał wnosi ze sobą poziom zaufania — i to jest jego najważniejsza właściwość.

konsola na Twojej maszynie   →  Ty, uwierzytelniony systemem operacyjnym
HTTP z pętli zwrotnej        →  proces na tej maszynie — niekoniecznie Ty
prywatna z komunikatora      →  ⚠️ ktoś, kto zna Twój numer
grupa                        →  ⚠️⚠️ wielu ludzi, w tym nieznani
webhook                      →  ⚠️⚠️ dowolny, kto zgadł URL

OpenClaw formułuje to jako zasadę projektową: wiadomość przychodząca jest niezaufana. Kanały z prywatnymi wiadomościami wymagają zatwierdzenia parowania, zanim w ogóle dotrą do agenta.

⭐ W części 2 zbudowałeś mechanizm (poziom_zaufania(zrodlo)). Tu go wypełniasz treścią.

---

1. Anatomia adaptera

Każdy kanał robi dokładnie cztery rzeczy. Nic więcej — logika agenta jest wspólna.

# $OSA/kanaly/bazowy.py
class Kanal:
    nazwa: str
    zaufanie_domyslne: str = "piaskownica"    # P66 — najwęziej

    def uwierzytelnij(self) -> None:
        """1. Połączenie i tożsamość (token, QR, gniazdo uniksowe)."""

    def normalizuj(self, surowe) -> dict:
        """2. Format kanału → jednolita wiadomość."""
        return {"tresc": ..., "nadawca": ..., "watek": ...,
                "zalaczniki": [], "zrodlo": self.nazwa}

    def wolno(self, wiadomosc: dict) -> tuple[bool, str]:
        """3. Kontrola dostępu. ⭐ Tu decyduje się bezpieczeństwo."""
        return False, "brak reguły"

    def wyslij(self, watek, tresc: str) -> None:
        """4. Formatowanie i dzielenie wychodzące."""

⚠️ zaufanie_domyslne = "piaskownica" w klasie bazowej jest celowe. Nowy kanał, o którym zapomnisz w konfiguracji, dostaje najwęższe uprawnienia. To jest P66 wyrażone w typie, a nie w komentarzu.

---

2. Kontrola dostępu — cztery bramki po kolei

def wolno(self, w: dict) -> tuple[bool, str]:
    # 1. Lista dozwolonych — najtańsza, najpierw
    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, ""                      # ⭐ cicho — patrz P110

    # 2. Grupa wymaga wzmianki
    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. Limit tempa per nadawca
    if not self._limit_ok(w["nadawca"]):
        return False, "Za dużo wiadomości. Spróbuj za minutę."

    # 4. Rozmiar
    if len(w["tresc"]) > self.k.get("max_znakow", 8000):
        return False, f"Wiadomość za długa (limit {self.k['max_znakow']} znaków)."

    return True, ""

2.1 · Parowanie

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} → kod {kod}\n"
          f"  zatwierdź: sesja-cli.py parowanie zatwierdz {kod}", file=sys.stderr)
    return (f"Nie znam Cię. Twój kod parowania: {kod}\n"
            f"Poproś właściciela o zatwierdzenie.")

⭐ Zatwierdzenie odbywa się poza kanałem — na maszynie, w terminalu. To jest istotne: gdyby dało się zatwierdzić parowanie wiadomością, kanał parowałby sam siebie. Kontrola dostępu nie może przebiegać kanałem, który kontroluje.

---

3. Trzy kanały dla Twojego systemu

3.1 · Gniazdo uniksowe — najbezpieczniejszy kanał, jaki masz

# $OSA/kanaly/gniazdo.py
"""Gniazdo uniksowe: uwierzytelnianie robi system plików. Zero konfiguracji."""
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)          # ⭐ tylko Ty — to CAŁE uwierzytelnianie
        self.s.listen(4)

    def wolno(self, w): return True, ""   # prawa pliku już zdecydowały

⭐ chmod 600 na gnieździe to pełne uwierzytelnianie za jedną linię. Żadnych tokenów, haseł ani rotacji — jądro egzekwuje to lepiej niż jakikolwiek kod, który byś napisał. Dla kanału lokalnego to jest właściwa odpowiedź i szkoda, że tak rzadko używana.

3.2 · HTTP — kiedy potrzebujesz tokenu

Pętla zwrotna nie wystarcza jako uwierzytelnianie: każdy proces na maszynie (w tym kod w piaskownicy, jeśli ma sieć) może się połączyć.

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, nie ==. Porównanie == kończy się na pierwszej różnicy i wycieka długość wspólnego prefiksu przez czas wykonania. Przy lokalnym atakującym jest to praktycznie wykorzystywalne.

3.3 · Wysyłanie — dzielenie i formatowanie

def podziel(tresc: str, limit: int) -> list[str]:
    """Dzieli po akapitach, potem zdaniach. NIGDY w środku bloku kodu."""
    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)]

⚠️ Warunek not w_kodzie jest tym, o którym wszyscy zapominają. Blok kodu przecięty na pół to dwie wiadomości z nieparzystą liczbą znaczników — komunikator zrenderuje resztę rozmowy jako kod.

---

4. Routing — kanał wybiera agenta

OpenClaw pozwala mapować kanał i rozmówcę na osobnego agenta z własną przestrzenią roboczą, modelem i piaskownicą. Wzorzec jest dobry i tani.

{
  "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"}
  ]
}

⭐ Czwarta reguła pokazuje mechanizm, który warto zauważyć: ten sam agent, węższy zestaw narzędzi. Gospodarz w grupie to ten sam AGENTS.md, ta sama pamięć — ale dwa narzędzia zamiast dziesięciu. Nie potrzebujesz osobnego agenta, żeby ograniczyć uprawnienia; potrzebujesz filtra z części 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. Strumieniowanie

async def strumien_do_komunikatora(kanal, watek, zrodlo, min_odstep=1.0):
    """Edycja wiadomości zamiast tokenów. Rzadziej niż co sekundę — limity API."""
    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)      # bez kursora na koniec

💡 Strumieniowanie to funkcja postrzeganej szybkości, nie realnej. Przy TTFT poniżej sekundy (część 8 kursu 1) daje mniej, niż się wydaje. Zrób to dla kanałów, gdzie jest darmowe, i nie walcz o to w komunikatorze.

---

6. EKSPERYMENT 12 — dwa kanały, dwa poziomy

Hipoteza: ta sama prośba przez dwa kanały daje różne zachowanie, bo różni się poziom zaufania — i różnica jest wyegzekwowana, nie deklarowana.

mkdir -p $OSA/pomiary/kanaly

⭐ Wiersz 2 jest najważniejszy. Poprawna odpowiedź przez HTTP to nie „nie wolno mi" — tylko zachowanie agenta, który nie wie, że takie narzędzie istnieje, bo filtr z części 7 usunął je ze schematu. Odmowa ujawnia istnienie zdolności; brak narzędzia nie ujawnia niczego.

6.1 · Pomiar dodatkowy — parowanie i tempo

python3 testy/kanaly.py --atak parowanie   # 50 wiadomości od nieznanego nadawcy
python3 testy/kanaly.py --atak tempo       # 100 wiadomości w 10 s od znanego
python3 testy/kanaly.py --atak token       # 1000 prób zgadnięcia tokenu HTTP

Oczekiwane: nieznany dostaje jeden kod parowania (nie 50); znany trafia na limit tempa; próby tokenu są odrzucane w stałym czasie (sprawdź rozrzut czasów — jeśli rośnie z długością wspólnego prefiksu, masz P111).

---

7. ĆWICZENIE

1. Zaimplementuj KanalGniazdo (zaufanie host) i KanalHTTP (piaskownica, token).
2. Wepnij routing z konfiguracji; reguła domyślna najwęższa.
3. Dodaj parowanie z zatwierdzaniem poza kanałem.
4. Zaimplementuj podziel() i sprawdź na odpowiedzi z blokiem kodu w środku.
5. Uruchom tabelę z §6 i zapisz wyniki.

python3 sesja-cli.py kanaly
# gniazdo  ✓ ~/.local/run/osa.sock (0600)  zaufanie: host          dziś: 34 wiad.
# http     ✓ 127.0.0.1:18789 (token)       zaufanie: piaskownica   dziś: 8 wiad., 2 odrzucone
# telegram ✗ wyłączony
#
# parowania oczekujące: 1
#   A3F9C1  +48***1234  (2026-09-23 14:02)

6. Zadanie dodatkowe: dodaj trzeci kanał — webhook dla skanera. Zastanów się, jakie zaufanie mu dać, skoro treść pochodzi z pliku, który ktoś położył w katalogu. (Podpowiedź: to samo co Badacz.)

---

8. Pułapki tej części

---

9. Checkpoint

cd $OSA
stat -c '%a' ~/.local/run/osa.sock | grep -q 600 && echo "✓ gniazdo 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("✓ zaufanie domyślne, routing, dzielenie kodu")
EOF
python3 testy/kanaly.py --atak token --sprawdz-czas && echo "✓ stały czas porównania"
git add -A && git commit -m "część 12: kanały, parowanie, routing, strumieniowanie"

• Kanał niesie poziom zaufania; domyślny jest najwęższy
• Cztery bramki kontroli dostępu, w kolejności od najtańszej
• Parowanie zatwierdzane poza kanałem
• Gniazdo uniksowe uwierzytelnia prawami pliku
• Token porównywany w stałym czasie
• Routing mapuje kanał na agenta i zestaw narzędzi
• Dzielenie nigdy nie przecina bloku kodu
• ⭐ Sześć prób przez dwa kanały daje różne, wyegzekwowane zachowanie

---

Dalej: 13-FLOTA-WIELU-AGENTOW.md — trzej agenci zaczynają współpracować, a Ty uczysz się, kiedy to jest gorsze niż jeden.