Where the wait goes: prefill, decode and a phone channel

Draft chapter. The textbook guide covers setup and the supplied code. Practice in Exercise Book 9, then check Solutions 9 once you have tried it.

Every reply a deployed agent sends has a wait inside it, and a lab expects you to know where that wait goes. Part A measures it: prefill, decode and the prefix cache, derived and then timed on two small local models. Part B builds a phone channel that survives a lost reply without running the model twice. It keeps its own queue, cursor and outbox with a supplied reference implementation; Chapter 8 builds the same separation of durable work from scratch.

Lucy has left the shop to collect packaging. She wants to ask for the opening brief without returning to the development terminal. The agent already knows how to inspect stock, create drafts and use a tested opening procedure. A phone interface should deliver Lucy's request into that same program and return its result. It should not create a second agent with a separate memory and a different interpretation of her preferences.

The interesting failure is easy to overlook. A messaging service delivers a request, the local process stores it, and the connection drops before either side is certain what happened. When the request arrives again, should the model run twice? On the return path, the service might accept a report and lose its reply. Sending the report again would make the interface look unreliable even though the underlying stock work completed correctly.

Part A first measures where the wait for a reply goes inside the model. Part B then builds one Telegram adapter around durable intake, stable session identity and an explicit outbound delivery state. You will keep the local bot interface available for offline exercises. The same loop and active opening skill from Chapter 7 will perform the work. Consequential purchasing remains outside this chapter's dispatcher.

Learning objectives

Part A measures latency. After it you should be able to:

• separate a reply's time into prefill and decode, and explain why they are limited by different things;
• fit prefill time as $b\,n + c\,n^2$, and explain the quadratic term;
• explain what streaming changes for the person waiting;
• derive the prompt tokens a growing conversation rereads, with and without a prefix cache.

Part B builds the phone channel. After it you should be able to:

• implement a narrow HTTP bot adapter;
• authenticate an operator by an explicit numeric allowlist;
• bind private requests to stable sessions;
• commit accepted work and the inbound cursor together;
• serialize conflicting session work;
• distinguish a successful report delivery from an unknown outcome.

The deliverable is a phone-capable agent with a deterministic local checkpoint. The local checkpoint proves admission, duplicate handling, context routing and delivery-state behavior without credentials. The optional Telegram run adds a real service interaction and requires you to inspect the reply on your phone. These are different observations. A fake bot returning a message identifier cannot prove that a handset received anything.

Part A: where does the wait for a reply go?

When Lucy messages the agent from her phone, she waits. Part of the wait is the network and the queue, which later chapters measure. The rest happens inside the model, in two phases with different costs. This part derives both, measures them on two local models, and shows what the prefix cache does to a conversation that grows with every message. The timings below were recorded on an Apple Silicon Mac (the receipt records macOS on arm64); on Colab's GPU the numbers will differ, and the shapes of the curves are what carry over.

The functions live in the chapter's learner file.

import json
import runpy

lat = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch09_latency_learner.py")
measured = json.loads(open("docs/evidence/book-ch09/ch09-latency-receipt-v1.json").read())

Two phases: prefill and decode

A reply is generated in two phases:

• Prefill reads the whole prompt: the system instructions, the conversation so far and Lucy's new message. All of its tokens are known in advance, so the model processes them together in large matrix multiplications. Its time grows with the number of prompt tokens $n$.
• Decode then writes the reply one token at a time, because each token depends on the one before. Each step is limited by reading the model's weights from memory, so its time per token is roughly fixed; Chapter 19 works that limit out in full.

Prefill has two parts. Every prompt token passes through the model's layers, about two floating-point operations per parameter, which costs a fixed $b$ seconds per token. And in attention, every token is compared with every token before it, which adds a cost that grows with the square of the prompt length. With a decode cost of $d$ seconds per generated token, the time to the first token and the time to a whole reply of $m$ tokens are

$ T_{\text{first}} = b,n + c,n^2 + d, \qquad T_{\text{reply}} = T_{\text{first}} + (m - 1),d. $

Prefill processes all the prompt's tokens together in large matrix multiplications, so it is limited by arithmetic. Decode is limited by memory. Prefill therefore handles tokens far faster than decode writes them.

Measured: prefill and decode

The experiment sent prompts of one to eight copies of Lucy's forty notes to qwen2.5:0.5b and qwen2.5:1.5b, three times each. Each prompt began with a fresh random marker so that nothing cached could be reused. Ollama reports how many prompt tokens it evaluated and how long prefill took. Dividing prefill time by $n$ gives $b + c\,n$, so a least-squares line through the time per token against $n$ gives $b$ and $c$.

uv run python book/textbook/experiments/profrod_sovereign_agent_textbook_ch09_latency_v1.py \
    --out ch09-latency-receipt.json

for model, m in measured["models"].items():
    p = m["prefill"]
    print(
        f"{model}: b = {p['seconds_per_token']:.2e} s, c = {p['seconds_per_token_squared']:.2e} s; "
        f"prefill {p['tokens_per_second_shortest']} to {p['tokens_per_second_longest']} tokens/s; "
        f"decode {round(1 / m['decode']['seconds_per_output_token'])} tokens/s"
    )

qwen2.5:0.5b: b = 1.62e-04 s, c = 2.61e-08 s; prefill 5607 to 3543 tokens/s; decode 213 tokens/s
qwen2.5:1.5b: b = 5.39e-04 s, c = 5.81e-08 s; prefill 1742 to 1302 tokens/s; decode 129 tokens/s

For both models, the time per prompt token rises with the prompt's length, as the $n^2$ term predicts. qwen2.5:0.5b read about 5,600 tokens per second from a 600-token prompt, but only 3,500 per second from a 4,500-token one. At that length attention took about 40% of the prefill time: $c\,n^2 = 0.52$ seconds of 1.25.

The larger model's per-token cost $b$ is 3.3 times the smaller's, close to its roughly three times as many parameters, as work limited by arithmetic should be.

Decode was far slower: 213 and 129 tokens per second, 26 and 13 times slower than prefill on a short prompt. Between the two models decode differs by a factor of only 1.65, not the parameter ratio, because it is limited by memory traffic and fixed overheads rather than arithmetic.

Streaming changes what Lucy feels

A person reading a reply starts reading at the first token. Streaming sends each piece as it is generated, so what Lucy waits for is $T_{\text{first}}$, not $T_{\text{reply}}$. The experiment streamed five answers of up to 150 tokens about the shop notes.

for model, m in measured["models"].items():
    first = [r["first_piece_seconds"] for r in m["streaming"]]
    whole = [r["whole_reply_seconds"] for r in m["streaming"]]
    print(f"{model}: first piece {sorted(first)}, whole reply {sorted(whole)}")

qwen2.5:0.5b: first piece [0.102, 0.102, 0.103, 0.104, 0.105], whole reply [0.294, 0.311, 0.395, 0.447, 0.468]
qwen2.5:1.5b: first piece [0.124, 0.126, 0.128, 0.129, 0.134], whole reply [0.603, 0.665, 0.74, 1.339, 1.34]

The first piece arrived after about 0.10 seconds for the small model and 0.13 for the larger one, whatever the length of the reply. The whole reply took up to 0.47 and 1.34 seconds. Streaming does not make the model faster: it lets Lucy start reading as soon as the first words exist, and the longer the reply, the more it saves.

A growing conversation and the prefix cache

Every message Lucy sends is appended to the conversation, and the whole conversation is sent again. If each turn adds $m$ tokens, turn $i$ has about $i\,m$ tokens of history, and a conversation of $k$ turns makes the model read

$ \sum_{i=1}^{k} i,m = m,\frac{k(k+1)}{2} $

prompt tokens in total. That is quadratic in the length of the conversation. A prefix cache keeps the model's internal state for the start of the prompt (its key–value cache, from Chapter 19) and reuses it when the next prompt begins the same way. Then each turn reads only what is new, $k\,m$ in total.

without = lat["conversation_prefill"](8, 100, cached=False)
with_cache = lat["conversation_prefill"](8, 100, cached=True)
print(without, with_cache)

3600 800

The experiment held an eight-turn conversation about the shop with qwen2.5:0.5b, answering Chapter 5's questions from the notes. It ran twice: once with the history unchanged, so the prefix could be reused, and once with a marker at the very start that changed every turn, so nothing could be reused.

for mode, rows in measured["conversation"].items():
    print(mode, [r["prompt_tokens_evaluated"] for r in rows])
    print(mode, [r["prefill_seconds"] for r in rows])

cached [583, 616, 646, 676, 708, 739, 769, 802]
cached [0.0067, 0.0187, 0.0216, 0.0218, 0.0284, 0.0301, 0.0312, 0.0261]
uncached [615, 649, 691, 721, 753, 787, 815, 849]
uncached [0.128, 0.1214, 0.1279, 0.134, 0.1384, 0.1461, 0.1523, 0.1596]

Ollama reported nearly the full prompt length in both runs, so its token count cannot show the cache; the prefill time can. With the history unchanged, turns 2 to 8 took 0.019 to 0.031 seconds of prefill. With the changing marker they took 0.121 to 0.160 seconds, five to seven times as long, growing with every turn. The whole reply took about 0.19 seconds per turn with the cache and 0.31 without.

The first turn of the cached run took only 0.007 seconds of prefill. An earlier run of this same experiment had sent the identical first prompt shortly before, and the server still held its state. A cache outlives the request that filled it.

The decision

• Stream replies to Lucy's phone when the channel allows it, or send a short acknowledgment first. The first token arrives long before the whole reply.
• Keep the start of the prompt stable. Anything that changes at the start (a timestamp, a random identifier, a reordered tool list) defeats the prefix cache, and every turn then rereads the whole conversation.
• Bound the history. Even with a cache, a long conversation costs context memory and prefill after every change near its start. Summaries and retrieval (Chapter 5) keep it bounded.

Part B: talk to the agent from your phone

Part A measured what Lucy waits for inside the model. This part builds the channel between her phone and the agent, and makes each message and report durable along the way.

Give each identity one job

A bot account identifies the application's messaging endpoint. A human user identifier identifies the operator. An update identifier identifies one incoming delivery, and a local work identifier identifies the durable task created from it. A session groups turns that may use the same conversation context and preferences. Keeping these identities separate prevents a reconnect from accidentally becoming a new conversation or a duplicate purchase request.

In the real adapter the account namespace comes from the non-secret numeric prefix of the bot credential. The credential itself is never a session identifier and never enters a prompt. A second bot with the same update number must still create a different origin identity. Conversely, restarting the same bot must preserve the namespace so its duplicate delivery still finds the existing work record.

Our human-authentication policy is deliberately narrow: a private chat, a non-bot sender on the numeric allowlist, and a chat identifier equal to that sender's identifier. Group messages, usernames, forwarded identity claims and text such as “I am Lucy” do not qualify. The HTTPS Bot API response supplies the sender metadata; our code decides which identities are allowed. This policy is separate from later approval to spend money.

View full-size diagram

flowchart LR
    P[Lucy's private message] --> B[Bot API adapter]
    B --> A[Operator and shape checks]
    A --> I[Durable work intake]
    I --> S[Session context and active skills]
    S --> L[Existing agent loop]
    L --> O[Durable report state]
    O --> B
    B --> R[Lucy's phone]

Prepare a dedicated bot and operator identity

Telegram's official tutorial describes obtaining a bot credential from BotFather using /newbot. Keep that credential in an operator-owned environment file outside the repository. It authenticates the bot, while your numeric user identifier determines who our application admits. Use a dedicated teaching bot and a private chat that you initiate yourself. Follow the official setup tutorial for the account steps, checked on 7 September 2026.

Set SOVEREIGN_AGENT_TELEGRAM_TOKEN and SOVEREIGN_AGENT_OPERATORS before a live run: in the local environment, or on Google Colab as two entries in the notebook's Secrets panel with notebook access switched on. The latter is a comma-separated list of positive numeric user identifiers. Obtain and verify your own identifier locally; do not replace the allowlist with “accept the first person who sends a message.” That would turn a race to contact a new bot into an authentication mechanism.

Use restrictive file permissions and avoid commands that echo environment values. The runtime reads the token from the environment, passes it to the HTTP transport over the child's standard input, and emits generic errors rather than URLs containing credentials. The file should never be pasted into a bug report. Tests use recognizable fake credentials so they can verify that error paths do not include them.

Build the narrow HTTP adapter

We use two Bot API operations: getUpdates for long polling and sendMessage for plain-text reports. The official API states that a higher polling offset confirms earlier updates, that polling and webhooks are mutually exclusive, and that upstream updates are retained for at most 24 hours. Our local persistence begins after intake; it cannot recover messages that expired before reaching the host. These protocol facts come from the Bot API reference, checked on 7 September 2026.

The adapter exposes a small call interface so a local object can exercise the rest of the system. Its requests go through your Chapter 3 transport, the same one the model provider uses: it limits the whole request deadline and response size, refuses redirects and keeps credentials out of process arguments. Reusing this transport does not delegate agent behavior to a framework. The model loop, intake and delivery rules remain ordinary Python that we construct here.

import json
import os
import re
import runpy
import tempfile
import time
import uuid
from pathlib import Path
from typing import Any, Protocol

LEARNER = Path("book/textbook/learner").resolve()
SKILLS = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch07_skills_learner.py")
QUEUE = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch08_work_queue_learner.py")
TRANSPORT = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch03_http_transport_learner.py")
MEMORY, LOOP = SKILLS["MEMORY"], SKILLS["LOOP"]
TOOLS, Assignment = LOOP["shop_tools"], QUEUE["Assignment"]


class Bot(Protocol):
    account: str

    def call(self, method: str, data: dict[str, Any]) -> Any: ...


class Telegram:
    """The two Bot API operations this agent uses, over Chapter 3's killable transport."""

    def __init__(self, token: str) -> None:
        if not re.fullmatch(r"[0-9]+:[a-zA-Z0-9_-]+", token):
            raise ValueError("invalid Telegram credential format")
        self._token = token
        self.account = token.split(":", 1)[0]

    def call(self, method: str, data: dict[str, Any]) -> Any:
        if method not in {"getUpdates", "sendMessage"}:
            raise ValueError("unsupported Telegram operation")
        try:
            response = TRANSPORT["request"](
                "https://api.telegram.org/bot" + self._token + "/" + method,
                data=json.dumps(data).encode(),
                headers={"Content-Type": "application/json"},
                timeout=35,
            )
            if response.status != 200:
                raise OSError("Telegram declined operation")
            result = json.loads(response.body)
            if not isinstance(result, dict) or result.get("ok") is not True:
                raise ValueError("Telegram declined operation")
            return result["result"]
        except (OSError, ValueError, KeyError, TypeError):
            # API URLs contain the token. Never expose exception URLs or bodies.
            raise OSError("Telegram request failed; inspect connectivity and credentials") from None


try:
    Telegram("not-a-bot-credential")
except ValueError as error:
    print(error)

invalid Telegram credential format

The method allowlist prevents callers from turning this object into a general bot administration client. The response must be an object, must report success and must contain its result. A malformed envelope or refused operation becomes a generic OSError. We deliberately do not put the server response body into the exception because responses and URLs can contain data that does not belong in logs.

A syntactically valid credential is not proof that Telegram accepts it. That requires a real authenticated request. Similarly, a successful HTTP status does not replace validation of the JSON envelope. The adapter performs both checks and leaves the next layer to validate whether a returned update batch or delivery receipt has the expected shape.

class LocalBot:
    account = "teaching"

    def __init__(self, updates):
        self.updates = updates
        self.offsets = []
        self.sent = []
        self.lose_next_reply = False

    def call(self, method, data):
        if method == "getUpdates":
            self.offsets.append(data["offset"])
            return self.updates
        self.sent.append(data)
        if self.lose_next_reply:
            self.lose_next_reply = False
            raise TimeoutError("accepted remotely but reply lost")
        return {"message_id": 900 + len(self.sent)}


def update(
    identifier,
    actor=123,
    text="Prepare replenishment drafts from current stock. State USD amounts.",
):
    return {
        "update_id": identifier,
        "message": {
            "from": {"id": actor, "is_bot": False},
            "chat": {"id": actor, "type": "private"},
            "text": text,
        },
    }


bot = LocalBot([update(101)])
print(bot.call("getUpdates", {"offset": 0})[0]["update_id"])
print(bot.offsets)

101
[0]

This local bot intentionally returns repeated updates even when the supplied offset has advanced. That is an adversarial exercise for our own duplicate handling, not a claim that the real service normally ignores offsets. Its outbound list represents accepted messages in the fixture. Losing a reply after adding to that list lets us test uncertainty without relying on an unreliable network accident.

Build durable admission before advancing the cursor

A phone message becomes work in the queue you built in Chapter 8: a work row with a unique source, a session and the request text, and later a report. The channel adds three small tables of its own: the cursor and poller lease for each bot account, and a route recording which channel and recipient each piece of work came from, so its reports go back there and nowhere else. They are one more line of migrations, owned by channel, on the same Chapter 4 store as your memory and skills.

Chapter 8's queue compares rows as tuples, and Chapter 5's memory reads them by column name. Record is a row that does both. An accepted source must not be reused for different content; otherwise a duplicate identifier could silently replace the request that an operator authorized.

Chapter 8's admit opens its own transaction, and the intake below must admit a whole batch and move its cursor in one. So admit_message applies Chapter 8's rules inside the caller's transaction. A request that arrives when the queue is full is not dropped: it is recorded as finished work whose report tells the sender to try again later.

CHANNEL_MIGRATIONS = {
    1: (
        # One inbound cursor and one live poller per bot account.
        "CREATE TABLE channel_cursor (channel TEXT PRIMARY KEY, offset INTEGER NOT NULL)",
        "CREATE TABLE channel_leases ("
        " channel TEXT PRIMARY KEY, owner TEXT NOT NULL, expires REAL NOT NULL)",
        # Where each piece of work came from, so its reports go back there and nowhere else.
        "CREATE TABLE channel_routes ("
        " work_id INTEGER PRIMARY KEY, channel TEXT NOT NULL, recipient TEXT NOT NULL)",
        # A report whose recipient left the allowlist before it was sent. It is never sent.
        "CREATE TABLE report_denials ("
        " report_id TEXT PRIMARY KEY, recipient TEXT NOT NULL, created REAL NOT NULL)",
    ),
}


class Record(tuple):
    """A row that compares as a tuple, as Chapter 8's queue expects, and reads by column name,
    as Chapter 5's memory expects, including dict(row)."""

    names: tuple[str, ...] = ()

    def __getitem__(self, key):
        return super().__getitem__(self.names.index(key) if isinstance(key, str) else key)

    def keys(self):
        return self.names


def named_rows(cursor, values):
    record = Record(values)
    record.names = tuple(column[0] for column in cursor.description)
    return record


def open_channel(path: str | Path, *, capacity: int = 10):
    """Chapter 7's store of memory and skills, Chapter 8's queue, and the channel's own tables."""
    db = SKILLS["open_skills"](path)
    queue = QUEUE["WorkQueue"](db, capacity=capacity)
    db.migrate("channel", CHANNEL_MIGRATIONS)
    db.connection.row_factory = named_rows
    return db, queue


def admit_message(connection, queue, source_id: str, session_id: str, text: str, route):
    """Chapter 8's admission, inside the caller's transaction, with the message's route.

    Returns (work_id, new). A known source with the same content is a duplicate; with other
    content it is refused. Over capacity, the request is recorded as finished work whose report
    tells the sender, rather than silently dropped."""
    row = connection.execute(
        "SELECT work_id, session_id, text FROM work WHERE source_id = ?", (source_id,)
    ).fetchone()
    if row is not None:
        if (row[1], row[2]) != (session_id, text):
            raise ValueError("intake identity reused for different content")
        return row[0], False
    (open_work,) = connection.execute(
        "SELECT COUNT(*) FROM work WHERE state IN ('pending', 'running')"
    ).fetchone()
    full = open_work >= queue.capacity
    work_id = connection.execute(
        "INSERT INTO work (source_id, session_id, text, state) VALUES (?, ?, ?, ?)",
        (source_id, session_id, text, "finished" if full else "pending"),
    ).lastrowid
    connection.execute("INSERT INTO channel_routes VALUES (?, ?, ?)", (work_id, *route))
    if full:
        connection.execute(
            "INSERT INTO reports (report_id, work_id, generation, body, delivery)"
            " VALUES (?, ?, 1, ?, 'pending')",
            (f"r{work_id}.1", work_id, "The queue is full; nothing was started. Try again later."),
        )
    return work_id, True


scratch = tempfile.TemporaryDirectory(prefix="lucy-phone-chapter-")
root = Path(scratch.name)
db, queue = open_channel(root / "agent.sqlite")
with db.immediate() as connection:
    first, _ = admit_message(connection, queue, "local:probe", "local-session", "brief", ("local", ""))
with db.immediate() as connection:
    again, _ = admit_message(connection, queue, "local:probe", "local-session", "brief", ("local", ""))
print("same durable identity:", first == again)
try:
    with db.immediate() as connection:
        admit_message(
            connection, queue, "local:probe", "local-session", "changed request", ("local", "")
        )
except ValueError as error:
    print(error)

same durable identity: True
intake identity reused for different content

A duplicate lookup occurs before the capacity check, so a repeated delivery does not take a second place in the queue. When the queue is full, the request is still persisted, with its original identity and payload, as finished work and a report that says why nothing was started. It is not executed later while the program pretends the queue had room.

This chapter's dispatcher exposes stock and draft tools only; a command-looking message cannot create purchasing authority. Chapter 11 will connect explicit approval to exact proposals.

Commit the batch and cursor together

The inbound cursor belongs to a bot account. Read it, request updates from that offset, validate the returned batch, then enter one immediate transaction. For each allowed message, create or recover the corresponding work record. Advance the cursor only after every batch member has been considered. Committing the cursor separately would create a window in which the remote service can forget a request that the runtime never stored.

A single local poller owns a short lease for that bot account. It acquires the lease before the network call and rechecks ownership inside the intake transaction. A process that outlives its lease may have a response in memory, but that response does not entitle it to commit over a replacement poller. The lease is released only if its owner token still matches.

def check_operators(operators: frozenset[int]) -> None:
    if not operators or any(type(actor) is not int or actor <= 0 for actor in operators):
        raise ValueError("explicit numeric operator allowlist required")


def allowed_text(message: Any, operators: frozenset[int]) -> tuple[int, str] | None:
    """The sender and text of a private message from an allowlisted person, or None to ignore.
    A malformed message raises instead: the whole batch is then refused, cursor and all."""
    if not isinstance(message, dict):
        raise ValueError("invalid Telegram message object")
    sender, chat = message.get("from", {}), message.get("chat", {})
    if not isinstance(sender, dict) or not isinstance(chat, dict):
        raise ValueError("invalid Telegram sender or chat object")
    actor, text = sender.get("id"), message.get("text")
    if (
        type(actor) is int
        and actor in operators
        and chat.get("type") == "private"
        and type(chat.get("id")) is int
        and chat.get("id") == actor
        and not sender.get("is_bot")
        and isinstance(text, str)
        and text.strip()
        and len(text.encode()) <= 16_384
    ):
        return actor, text
    return None


def poll(db, queue, bot: Bot, operators: frozenset[int]) -> list[int]:
    """One bounded poll. Returns the newly admitted work ids; the queue, not this list, is
    the durable record."""
    check_operators(operators)
    channel = "telegram:" + bot.account
    owner = uuid.uuid4().hex
    with db.immediate() as connection:
        lease = connection.execute(
            "SELECT expires FROM channel_leases WHERE channel=?", (channel,)
        ).fetchone()
        if lease and lease[0] > time.time():
            raise PermissionError("another poller owns this bot account")
        connection.execute(
            "INSERT INTO channel_leases(channel,owner,expires) VALUES (?,?,?) "
            "ON CONFLICT(channel) DO UPDATE SET owner=excluded.owner,expires=excluded.expires",
            (channel, owner, time.time() + 40),
        )
    try:
        return _poll_owned(db, queue, bot, operators, channel, owner)
    finally:
        with db.immediate() as connection:
            connection.execute(
                "DELETE FROM channel_leases WHERE channel=? AND owner=?", (channel, owner)
            )


def _poll_owned(db, queue, bot, operators, channel: str, owner: str) -> list[int]:
    cursor = db.connection.execute(
        "SELECT offset FROM channel_cursor WHERE channel=?", (channel,)
    ).fetchone()
    offset = cursor[0] if cursor else 0
    updates = bot.call(
        "getUpdates",
        {"offset": offset, "limit": 100, "timeout": 20, "allowed_updates": ["message"]},
    )
    if not isinstance(updates, list) or len(updates) > 100:
        raise ValueError("invalid Telegram update batch")
    admitted = []
    with db.immediate() as connection:
        current = connection.execute(
            "SELECT 1 FROM channel_leases WHERE channel=? AND owner=? AND expires>?",
            (channel, owner, time.time()),
        ).fetchone()
        if not current:
            raise PermissionError("poller claim expired")
        highest = offset - 1
        for update in updates:
            if not isinstance(update, dict):
                raise ValueError("invalid Telegram update object")
            update_id = update.get("update_id")
            if type(update_id) is not int or update_id < 0:
                raise ValueError("invalid update identity")
            # Do not skip an earlier member of an unordered batch. The cursor is published only
            # after every member has been considered, in the same transaction as the work.
            highest = max(highest, update_id)
            allowed = allowed_text(update.get("message", {}), operators)
            if allowed is None:
                continue
            actor, text = allowed
            session = f"{channel}:{actor}"
            work_id, new = admit_message(
                connection, queue, f"{channel}:{update_id}", session, text, (channel, str(actor))
            )
            if new:
                admitted.append(work_id)
        connection.execute(
            "INSERT INTO channel_cursor(channel,offset) VALUES (?,?) "
            "ON CONFLICT(channel) DO UPDATE SET offset=excluded.offset",
            (channel, max(offset, highest + 1)),
        )
    return admitted


bot = LocalBot([update(103, actor=999), update(101), update(102)])
operators = frozenset({123})
identifiers = poll(db, queue, bot, operators)
print("accepted private requests:", len(identifiers))
print(
    "cursor:",
    db.connection.execute(
        "SELECT offset FROM channel_cursor WHERE channel='telegram:teaching'"
    ).fetchone()[0],
)
print("new work from replay:", len(poll(db, queue, bot, operators)))

accepted private requests: 2
cursor: 104
new work from replay: 0

The batch deliberately places update 103 before updates 101 and 102. Tracking the highest identifier does not justify skipping smaller members of the same returned batch; both allowed requests must become durable. The cursor moves past the unauthorized message as well, because the adapter has made and committed its decision to ignore that payload. Re-requesting an unauthorized message forever would not make it authorized.

An absent text field is a supported reason to ignore an update: this adapter does not turn photos or voice recordings into prompts. A malformed object is different. It raises a controlled validation error, rolling back the entire batch and its cursor. We do not silently advance past a response shape that the program cannot interpret. The supervised service can report a generic retry state without logging the malformed body.

View full-size diagram

sequenceDiagram
    participant P as Poller
    participant T as Bot API
    participant D as SQLite
    P->>D: Acquire account lease and read cursor
    P->>T: getUpdates at committed offset
    T-->>P: Bounded batch
    P->>D: Begin immediate transaction
    P->>D: Recheck lease and persist accepted work
    P->>D: Store highest considered ID plus one
    P->>D: Commit work and cursor together
    P->>T: Next poll confirms earlier updates

bad = LocalBot([update(104), {"update_id": 105, "message": None}])
try:
    poll(db, queue, bad, operators)
except ValueError as error:
    print(error)
print(
    "phone work rows:",
    db.connection.execute(
        "SELECT count(*) FROM channel_routes WHERE channel='telegram:teaching'"
    ).fetchone()[0],
)
print(
    "cursor unchanged:",
    db.connection.execute(
        "SELECT offset FROM channel_cursor WHERE channel='telegram:teaching'"
    ).fetchone()[0],
)
print(
    "poller leases:",
    db.connection.execute("SELECT count(*) FROM channel_leases").fetchone()[0],
)

invalid Telegram message object
phone work rows: 2
cursor unchanged: 104
poller leases: 0

The first request in that bad batch looked valid. Its absence afterward proves the transaction rolled it back along with the cursor. This is stronger evidence than checking only that the final invalid object raised an exception. The next valid batch can retry from the unchanged cursor, and the released lease allows a new poller to proceed.

Route durable work into the existing loop

The result of poll is a convenient list of newly admitted identifiers; it is not the durable queue. After a process restart, that Python list is gone. A worker must query the stored work records, claim one eligible assignment, assemble context using the assignment's session, and finish the record with its result. The real service already follows that pattern, and the chapter checkpoint does too.

View full-size diagram

sequenceDiagram
    participant A as First process
    participant D as SQLite
    participant B as Replacement worker
    participant C as Competing worker
    A->>D: Commit two requests for one session
    Note over A: Process exits and its Python list is lost
    B->>D: Query durable ready work
    D-->>B: Both requests remain
    B->>D: Claim first request
    C->>D: Try second request in same session
    D-->>C: No eligible claim yet
    B->>D: Finish first request
    C->>D: Claim second request

Two requests in the same private session should not concurrently assemble context from the same unfinished conversation. claim below is Chapter 8's claim with one more rule: an item waits while another item of its session is running. Different sessions can be handled separately. A holder that crashes leaves its session waiting until Chapter 13 adds leases and replacement; here we can already demonstrate the visible claim behavior using two database connections.

def activate_opening_skill(db) -> None:
    """Stage the repository's opening procedure and activate it through Chapter 7's evaluation."""
    if SKILLS["skill_snapshot"](db)[1]:
        return
    source = LEARNER.parent / "skills" / "profrod_sovereign_agent_textbook_opening_check_v1.toml"
    candidate = SKILLS["stage_skill"](db, source)
    SKILLS["activate_skill"](
        db,
        candidate.name,
        candidate.version,
        evaluate=lambda skill: SKILLS["evaluate_opening"](SKILLS["OfflineShopModel"], skill),
        required_cases=frozenset(case.name for case in SKILLS["OPENING_CASES"]),
    )


def claim(db, worker_id: str, *, work_id: int | None = None) -> Any:
    """Chapter 8's claim, with one more rule: a session with a running turn waits.

    Two turns of one conversation must not assemble context from the same unfinished history.
    Chapter 13 adds leases, so a crashed holder does not block its session forever."""
    with db.immediate() as connection:
        row = connection.execute(
            "SELECT w.work_id, w.session_id, w.text FROM work w"
            " WHERE w.state = 'pending' AND (? IS NULL OR w.work_id = ?)"
            " AND NOT EXISTS (SELECT 1 FROM work other"
            "   WHERE other.session_id = w.session_id AND other.state = 'running')"
            " ORDER BY w.work_id LIMIT 1",
            (work_id, work_id),
        ).fetchone()
        if row is None:
            return None
        connection.execute(
            "UPDATE work SET state = 'running', worker_id = ? WHERE work_id = ?",
            (worker_id, row[0]),
        )
    return Assignment(row[0], row[1], row[2], worker_id)


activate_opening_skill(db)
MEMORY["remember"](db, "telegram:teaching:123", "format", "three bullets", "lucy/explicit-message")
db.close()
db, queue = open_channel(root / "agent.sqlite")
queued = [
    row[0]
    for row in db.connection.execute(
        "SELECT work_id FROM channel_routes WHERE channel='telegram:teaching' ORDER BY work_id"
    )
]
owner = claim(db, "phone-worker", work_id=queued[0])
other, _ = open_channel(root / "agent.sqlite")
print("second session claim:", claim(other, "other-worker", work_id=queued[1]))
other.close()
allowed = TOOLS["build_tools"](TOOLS["SHOP"]).allowed
selected = SKILLS["context"](db, owner.session_id, owner.text, allowed=allowed)
print("right preference:", "three bullets" in selected[0]["content"])
print("active procedure:", "skill_guidance" in selected[0]["content"])

second session claim: None
right preference: True
active procedure: True

Notice that the preference belongs to the phone session, not the earlier local session named lucy. Automatically combining sessions merely because two messages mention the same person's name would defeat the identity boundary. This chapter deliberately seeds the authenticated session's preference as an operator action. A later user interface can offer explicit, authenticated preference editing without allowing the model to relabel an arbitrary session.

run_claim passes the selected context into your Chapter 3 loop over the three shop tools and finishes the work through Chapter 8's queue, which checks that this worker still holds it and creates its report. The result also enters Chapter 5's history, under the memory revision the turn started with, so a later turn in the same session can see it and a forget can exclude it. It counts as draft evidence only if the draft observations match the independently authored quantities. A durable per-account model allowance arrives in Chapter 18; here each turn is bounded by the loop's own limits.

def run_claim(db, queue, assignment, model) -> tuple[bool, Any]:
    """One turn: the session's context, Chapter 3's loop over the shop tools, and a report.

    The finished result also enters Chapter 5's history, under the memory revision the turn
    started with, so a later turn in the same session can see it and a forget can exclude it."""
    revision = MEMORY["memory_revision"](db, assignment.session_id)
    dispatcher = TOOLS["build_tools"](TOOLS["SHOP"])
    messages = SKILLS["context"](
        db, assignment.session_id, assignment.text, allowed=dispatcher.allowed
    )
    result = LOOP["run_loop"](model, dispatcher, messages)
    passed = result.status == "COMPLETED" and LOOP["draft_evidence"](result)
    answer = result.answer or "The agent stopped: " + result.status
    queue.finish(assignment, answer)
    MEMORY["record_result"](db, assignment.session_id, assignment.text, answer, revision)
    return passed, result


passed, result = run_claim(db, queue, owner, SKILLS["OfflineShopModel"]())
print("first draft evidence:", passed)
owner = claim(db, "phone-worker", work_id=queued[1])
passed, result = run_claim(db, queue, owner, SKILLS["OfflineShopModel"]())
print("second draft evidence:", passed)
print(
    "phone results:",
    db.connection.execute(
        "SELECT count(*) FROM reports JOIN channel_routes USING (work_id)"
        " WHERE channel='telegram:teaching'"
    ).fetchone()[0],
)

first draft evidence: True
second draft evidence: True
phone results: 2

The offline model does not interpret the formatting preference; it supplies authored responses to test the data flow. We checked that the selected context contained the preference, while the transcript checker verifies the draft operations. Only a live model run and a reading of its answer can assess how well it follows the requested format. Avoid collapsing those observations into a single claim that the agent “understands Lucy.”

Give each report its own identity

A single request may produce an approval report and, later, a confirmed-order report. They need separate delivery records. If we keep one delivery flag on the work row, sending the first report sets it to confirmed; replacing the text with the final result can then suppress that result forever. Resetting the flag is also unsafe: an acknowledgement still in flight for the old text could mark the new text sent.

You built that outbox in Chapter 8. Each report has its own identity, r plus the work id and a generation, an immutable body, which a trigger protects, and its own delivery state and receipt. finish creates the report in the same transaction that finishes the work. A full queue's refusal creates a report too. The channel adds only the destination, through the work's route. Historical uncertain reports remain visible even after a later report succeeds.

Treat report delivery as an external effect

There are two state machines in this chapter: execution of the work and delivery of its result. A work record can be finished while its report is still pending. A report can become unknown even though the stock calculation completed correctly. Keeping those states separate helps the operator distinguish “the agent did not finish” from “the agent finished, but I may not have received its answer.”

The first four are Chapter 8's delivery states; its schema allows only those, so a denial is recorded in its own table beside the report. The sender checks the current operator allowlist immediately before admitting a send. It then commits SENDING before contacting Telegram. If the response is lost, the local program cannot know whether the message reached the service. This chapter chooses a visible uncertain state instead of automatically sending another copy. That is a delivery policy, not a claim of exactly-once messaging.

def deliver_one(db, bot: Bot, operators: frozenset[int]) -> str | None:
    """Chapter 8's send_one, with a destination and an allowlist.

    Sends this bot's oldest pending report to the recipient its work came from, once. The
    recipient must still be allowed; if not, the report is denied and never sent. "sending" is
    committed before the call. A valid message id confirms it; anything else leaves it "unknown",
    which this function never picks up again. Returns "confirmed", "unknown", "denied" or None."""
    check_operators(operators)
    channel = "telegram:" + bot.account
    with db.immediate() as connection:
        row = connection.execute(
            "SELECT r.report_id, r.body, c.recipient FROM reports r"
            " JOIN channel_routes c ON c.work_id = r.work_id"
            " WHERE c.channel = ? AND r.delivery = 'pending'"
            " AND r.report_id NOT IN (SELECT report_id FROM report_denials)"
            " ORDER BY r.work_id, r.generation LIMIT 1",
            (channel,),
        ).fetchone()
        if row is None:
            return None
        report_id, body, recipient = row
        if not recipient.isdigit() or int(recipient) not in operators:
            connection.execute(
                "INSERT INTO report_denials VALUES (?, ?, ?)", (report_id, recipient, time.time())
            )
            return "denied"
        # A crash after this commit is ambiguous, even if no HTTP call happened.
        connection.execute(
            "UPDATE reports SET delivery = 'sending' WHERE report_id = ?", (report_id,)
        )
    try:
        text = body if len(body) <= 3900 else body[:3800] + "\n[Report truncated.]"
        result = bot.call("sendMessage", {"chat_id": int(recipient), "text": text})
        if (
            not isinstance(result, dict)
            or type(result.get("message_id")) is not int
            or result["message_id"] <= 0
        ):
            raise ValueError("missing delivery receipt")
        outcome, receipt = "confirmed", json.dumps({"message_id": result["message_id"]})
    except (OSError, ValueError):
        outcome, receipt = "unknown", None
    with db.immediate() as connection:
        updated = connection.execute(
            "UPDATE reports SET delivery = ?, receipt = ? WHERE report_id = ?"
            " AND delivery = 'sending'",
            (outcome, receipt, report_id),
        )
        if not updated.rowcount:
            return "unknown"
    return outcome


bot.lose_next_reply = True
print("first delivery:", deliver_one(db, bot, operators))
print("second delivery:", deliver_one(db, bot, operators))
print("accepted by local bot:", len(bot.sent))
print("further automatic delivery:", deliver_one(db, bot, operators))
receipt = db.connection.execute(
    "SELECT receipt FROM reports WHERE delivery='confirmed'"
).fetchone()[0]
print("recorded successful message:", json.loads(receipt)["message_id"])

first delivery: unknown
second delivery: confirmed
accepted by local bot: 2
further automatic delivery: None
recorded successful message: 902

The fixture accepted the first report before losing the reply. We know that because we control the local bot's list. The runtime only sees the timeout, so it correctly records unknown. The second report belongs to the second request and receives a valid receipt. It is not a retry of the first report. This distinction is why a global count of sent messages cannot substitute for per-report delivery state.

The state and its receipt change in one statement, and only if the report is still sending. If restore or another recovery action changed the delivery state before this completion arrived, the conditional update affects no row and no stale success is recorded. An old network response does not get to overwrite a new local decision.

The receipt keeps only the message identifier. It does not copy the raw server response or bot credential; the report's route already names the channel and recipient. The message identifier is a delivery receipt from the service; it is not proof that Lucy read the report. A handset observation remains necessary for the user-facing result, and natural-language correctness still needs its own evaluation.

View full-size diagram

stateDiagram-v2
    [*] --> pending
    pending --> denied: Recipient no longer allowed
    pending --> sending: Commit send admission
    sending --> confirmed: Commit valid receipt
    sending --> unknown: Reply lost
    confirmed --> [*]
    unknown --> [*]: No blind retry
    denied --> [*]

Plain-text reports avoid an additional formatting language. Telegram documents a 4096-character text limit for sendMessage; the adapter leaves headroom and adds a visible truncation notice for long answers. The full local result remains available in the work record. Inspect a long or non-ASCII report on the actual channel rather than assuming that a character-count check guarantees an attractive render. Bot API text contract

late = poll(db, queue, LocalBot([update(106)]), operators)
owner = claim(db, "phone-worker", work_id=late[0])
queue.finish(owner, "An already completed report.")
print("delivery after allowlist change:", deliver_one(db, bot, frozenset({456})))
print("accepted messages unchanged:", len(bot.sent))
db.close()
db, queue = open_channel(root / "agent.sqlite")
print(
    "uncertain reports retained:",
    db.connection.execute("SELECT count(*) FROM reports WHERE delivery='unknown'").fetchone()[0],
)
db.close()
scratch.cleanup()

delivery after allowlist change: denied
accepted messages unchanged: 2
uncertain reports retained: 1

Revocation here governs a send that has not yet been admitted. It cannot retract a report already accepted by Telegram, and a local transaction cannot cancel a network request already in flight. The same distinction will matter more when the external effect is a supplier order. Chapter 12 will add provider-supported operation identity and reconciliation for that separate problem.

Serve from Google Colab

Most readers run this book on Google Colab, whose current runtime is Python 3.12. Three things change there, and the channel handles each one in code rather than in instructions.

Credentials come from Colab's Secrets panel, not from environment variables. secret reads the environment first and, only inside a Colab runtime, the panel. A secret that is missing, or that this notebook has not been given access to, comes back empty and is never printed. Colab answers those requests only inside the notebook's own kernel, so read secrets there, not in a child process.

A notebook cell is the natural place to serve, and a cell has to end. serve runs a bounded number of rounds of poll, answer and deliver. Each poll waits up to twenty seconds for a message, so ninety rounds keep a cell answering for about half an hour.

Cells get stopped, and runtimes get reset. A stopped cell raises KeyboardInterrupt in the middle of a turn; a lost runtime runs no handler at all. Either way the work is left running, and Chapter 8's states let a running item only finish. Left alone, it would hold Lucy's session forever. With one serving notebook per database, the serving worker's own label identifies its orphans. release_interrupted finishes each one with a report asking Lucy to send the request again, both when a cell is stopped and when the next serve starts. Chapter 13 replaces this single-worker rule with leases that work for any number of workers.

def secret(name: str) -> str:
    """A credential from the environment, or else from Colab's Secrets panel on Colab.

    Returned, never printed. Empty when neither has it, or when the Colab notebook has not been
    given access to it; the caller decides what a missing credential means."""
    value = os.environ.get(name, "")
    if value:
        return value
    try:
        from google.colab import userdata  # only importable inside a Colab runtime
    except ImportError:
        return ""
    try:
        return userdata.get(name) or ""
    except (userdata.SecretNotFoundError, userdata.NotebookAccessError, userdata.TimeoutException):
        return ""


def release_interrupted(db, queue, worker_id: str) -> list[int]:
    """Finish the work this worker label left running when its process stopped.

    A stopped Colab cell or a lost runtime cannot finish its turn, and Chapter 8's states allow a
    running item only to finish. Left running, it would hold its session forever: Chapter 13's
    leases solve this for many workers. With one serving notebook per database, the serving
    worker's own label identifies its orphans, and each gets a report asking Lucy to ask again."""
    orphans = [
        row[0]
        for row in db.connection.execute(
            "SELECT work_id FROM work WHERE state = 'running' AND worker_id = ? ORDER BY work_id",
            (worker_id,),
        )
    ]
    for work_id in orphans:
        session_id, text = db.connection.execute(
            "SELECT session_id, text FROM work WHERE work_id = ?", (work_id,)
        ).fetchone()
        queue.finish(
            Assignment(work_id, session_id, text, worker_id),
            "The agent stopped before answering this request. Please send it again.",
        )
    return orphans


def serve(
    db,
    queue,
    bot: Bot,
    operators: frozenset[int],
    model_factory,
    *,
    rounds: int = 1,
    worker_id: str = "phone-worker",
) -> dict[str, Any]:
    """Poll, answer and deliver, `rounds` times: the whole channel in one bounded call.

    Each poll waits up to 20 seconds for a message, so serve(..., rounds=90) keeps a Colab cell
    answering for about half an hour. Stopping the cell raises KeyboardInterrupt: the turn in
    progress is finished with a report saying so, and the interrupt is raised again. Admitted
    work and pending reports stay in the database for the next call."""
    if type(rounds) is not int or not 1 <= rounds <= 10_000:
        raise ValueError("a bounded number of rounds is required")
    channel = "telegram:" + bot.account
    summary: dict[str, Any] = {
        "recovered": release_interrupted(db, queue, worker_id),
        "admitted": 0,
        "ran": [],
        "deliveries": [],
    }
    for _ in range(rounds):
        summary["admitted"] += len(poll(db, queue, bot, operators))
        pending = [
            row[0]
            for row in db.connection.execute(
                "SELECT w.work_id FROM work w JOIN channel_routes c ON c.work_id = w.work_id"
                " WHERE c.channel = ? AND w.state = 'pending' ORDER BY w.work_id",
                (channel,),
            )
        ]
        for work_id in pending:
            held = claim(db, worker_id, work_id=work_id)
            if held is None:
                continue
            try:
                passed, _ = run_claim(db, queue, held, model_factory())
            except KeyboardInterrupt:
                release_interrupted(db, queue, worker_id)
                raise
            summary["ran"].append({"work": work_id, "draft_evidence": passed})
        while (outcome := deliver_one(db, bot, operators)) is not None:
            summary["deliveries"].append(outcome)
    return summary


colab_scratch = tempfile.TemporaryDirectory(prefix="lucy-colab-chapter-")
db, queue = open_channel(Path(colab_scratch.name) / "agent.sqlite")
activate_opening_skill(db)
served = serve(
    db, queue, LocalBot([update(301)]), frozenset({123}), SKILLS["OfflineShopModel"], rounds=2
)
print("admitted:", served["admitted"], "answered:", len(served["ran"]))
print("deliveries:", served["deliveries"])
print("secret with no environment and no Colab:", repr(secret("LUCY_EXAMPLE_SECRET")))
db.close()
colab_scratch.cleanup()

admitted: 1 answered: 1
deliveries: ['confirmed']
secret with no environment and no Colab: ''

The second round saw the same update again, because the local bot replays it, and admitted nothing new. The checkpoint goes further: it stands in for Colab's userdata module to show that shared, unshared and missing secrets behave as described, stops a cell in the middle of a turn, and abandons a turn the way a lost runtime would. Each leaves a report and a free session behind.

The course's Colab notebook, Talk to your agent from your phone, on Google Colab, puts these pieces in order: it clones the course, runs the offline checkpoint, helps you find your user ID, serves your bot and saves the database to Google Drive between sessions. The agent keeps its database in /content, which a runtime reset erases, and copies it to Drive with SQLite's own backup rather than running it directly on Drive.

Compare a single adapter with a shared gateway

OpenClaw's architecture document at commit 354538083db0a8728e16238cbd0b7a304416ff24 describes a long-lived gateway owning messaging connections, with control clients connecting over WebSocket. It documents typed requests, events, device pairing and idempotency keys for side-effecting methods. These are published architectural choices in the pinned gateway document.

The trade-off I infer is that a shared gateway can coordinate several clients and channels through one control interface, while Lucy's adapter has fewer moving parts for one private chat. Our design still separates the transport from work execution; it does not introduce a second WebSocket protocol. That comparison does not establish which whole project has better reliability or security.

An experiment that could change our choice is adding a second simultaneous operator interface. Measure how much authentication, routing, cancellation and state-notification logic gets duplicated. If maintaining those shared behaviors becomes the dominant cost, a gateway may earn its additional protocol and process boundaries. One extra channel alone is not proof that every interface must become a new service.

In a coding agent

Claude Code applies this chapter's prefix-cache measurement to every turn. Prompt caching is on by default. The API matches the start of each request against content it recently processed, and "the match is exact, so a change anywhere in the prefix recomputes everything after it." Claude Code therefore orders each request so that content that rarely changes comes first: the system prompt, then project context, then the conversation.

The documentation lists what breaks the prefix: switching models, changing effort, connecting or removing an MCP server, enabling a plugin, denying an entire tool, compacting. Each costs one slower, more expensive turn. Plan mode and skill loading are added as conversation messages for this reason, so the cached prefix survives them. You measured the same effect here: a stable prompt start cut prefill five- to sevenfold.

Try it in Claude Code: note the cache reads in /usage, switch models with /model, send one more message, and compare.

Run the cumulative checkpoint and the phone experiment

Run the offline checkpoint first; it runs only your code, on Python 3.12 as on Colab. It first checks the intake's refusals: forged chats, groups, bots, empty text, unknown senders, malformed batches, a changed replay and a second poller, and a full queue that answers the sender. Then it checks the Colab behavior: secrets from the panel, a serving cell, a stopped cell and a lost runtime. Then it admits an unordered batch containing two allowed messages and one unauthorized message, reopens the database, suppresses repeated intake, observes a competing session claim, runs the two draft tasks and injects an accepted delivery whose reply is lost. No network credentials or real purchases are involved.

import runpy
import sys
from unittest.mock import patch

checkpoint = runpy.run_path(
    "book/textbook/checkpoints/profrod_sovereign_agent_ch09_telegram_messaging_checkpoint.py"
)
with patch.object(sys, "argv", ["ch09.py"]):
    assert checkpoint["main"]() == 0

ok   prefill b n + c n^2, reply time, and quadratic against linear conversation cost
ok   prefill fits recompute; cached turns always prefilled faster than uncached ones
Refused: forged chats, groups, bots, empty text, unknown senders and malformed batches
Changed replay refused; a full queue answers the sender instead of dropping the request
Colab secrets: read from its panel; missing or unshared ones are empty, never shown
Serving cell: answers, then a stopped cell and a lost runtime each leave a report
Accepted private requests: 2
Duplicate intake after restart: 0
Conflicting session claim: None
Completed drafts: 2
First delivery: unknown
Second delivery: confirmed
Automatic resend: None
Recorded send receipts: 1
Delivery after the allowlist changed: denied
Purchase tool: refused by the dispatcher

For account setup, the companion telegram_identity_v1.py generates a one-time challenge on your local terminal. Send that exact text from the intended private account to your dedicated bot. It reports the matching sender's numeric identifier for your review, without changing an allowlist or creating work. It makes up to three bounded polls and does not advance the confirmation offset. Use a new teaching bot without another active poller or an unrelated backlog.

uv run python book/textbook/appendices/profrod_sovereign_agent_textbook_telegram_identity_v1.py

The helper is a setup instrument, not automatic enrollment. Its challenge demonstrates control of the private account that sent it to this bot, assuming the challenge remains under your control. A display name is insufficient. If multiple identities return the same challenge, setup refuses and requires a new challenge. After reviewing the identifier, place it in the operator environment setting and keep that setting operator-owned.

Now send the stock request from your phone: “Prepare replenishment drafts from current stock. State USD amounts.” Run one bounded channel pass using a persistent, dedicated test directory:

uv run python book/textbook/checkpoints/profrod_sovereign_agent_ch09_telegram_messaging_checkpoint.py --telegram --root /tmp/lucy-phone-test

This command serves one round with the authored offline model while contacting the real bot service. That isolates channel behavior from model variation. Add --rounds 15 to keep it answering for about five minutes. Add --live --model qwen3 to use a local HTTP model, or --claude claude-haiku-4-5-20251001 to use Claude with an ANTHROPIC_API_KEY, capped at one US dollar. On Colab, which has no local model server, use the notebook above, where Claude is the live option. The bot credential never appears in the output.

The checkpoint queries durable pending work, including requests admitted by a prior process, and drains a bounded set of pending reports. It attributes each result's delivery state by its work identifier instead of assuming the next outbound report belongs to the just-completed turn. If no work or delivery occurred, it exits without claiming a new success. An unknown delivery remains uncertain across subsequent invocations.

A setup message may itself arrive as ordinary text during the first run; the teaching opening skill may respond with a stock brief. No setup message grants authority. Keep the bot separate from any personal or production automation so those initial messages and the controlled failure experiments do not affect another workflow.

Expected observations

The local checkpoint should report two admitted requests, no new work from replay, a refused concurrent claim for the same session, two completed draft results, one unknown delivery and one confirmed delivery. The uncertain attempt is not automatically repeated after reopening. The confirmed report has one durable message-identifier receipt, a report for a removed recipient is denied without a network call, and the dispatcher refuses a purchase.

For the live channel, inspect the actual private reply and its content on your phone. Check that the reply belongs to the requested bot and the correct conversation. Stop the program after intake and run it again to observe durable processing. Keep a returned API receipt, a locally completed work record and a human reading of the answer as separate evidence rather than using any one as a substitute for all three.

Learner verification

Inspect the stored origin, session, channel and recipient for an allowed request, then compare them with an unauthorized update that did not create work. Replay the same update with changed text and confirm refusal without replacing the original payload. Remove the recipient from the allowlist before sending a pending report and observe denied without another network call.

Read the successful delivery event and identify the exact work record it supports. Inject a timeout after the local bot appends its accepted message, then confirm that a restart leaves the corresponding result uncertain. A program that simply repeats every pending-looking send would fail this lesson even if Lucy eventually receives an answer.

Practice

Exercise 1: Two bots, one update number

Create two local bot objects with different account namespaces and the same update identifier. Both allowed requests should become durable, with distinct origins and sessions. Then replay the first bot's update unchanged and with a modified payload. Explain why bot-account identity belongs in the deduplication key.

Exercise 2: A malformed second member

Start a batch with a valid allowed request, then add a message whose sender is a list instead of an object. Verify that work, cursor and poller ownership return to the expected state after refusal. Compare this with a well-formed message containing a photo and no text: the latter is unsupported content that the adapter may ignore, not a malformed object it must reinterpret.

Exercise 3: A slow model and a waiting message

Hold the first work claim while a second request arrives in the same private session. Confirm that intake can remain durable while conflicting execution waits. Release or finish the first assignment and demonstrate the second becoming eligible. Measure the delay without changing the rule that the two turns must not concurrently assemble stale session context.

Exercise 4: Delivery after local state changes

Change an admitted report's delivery state to unknown before a late successful API response returns. Confirm that the old completion does not record a receipt or overwrite the changed state. Explain what the fixture knows about remote acceptance, what the runtime can record safely, and why the operator may still need to inspect the actual conversation.

Exercise 5: Break the cache on purpose

Put the current time at the start of the system prompt, as many agents do, and rerun the conversation measurement. Then move the time to the end of the latest message instead. Report prefill time per turn for both. Which other common prompt contents would defeat the cache in the same way?

Exercise 6: A latency budget for Lucy's phone

Suppose Lucy should see the first words of a reply within one second, and the whole reply within five. Using the measured $b$, $c$ and $d$, find the longest prompt and the longest reply each model can serve within that budget, with and without streaming. Check one of your answers against a real call.

Operating limits of the teaching channel

Use one bot account with one active installation database. The poller lease coordinates processes sharing that database; it cannot coordinate two independent installations using the same token. Stop the other installation if Telegram reports a competing poller or webhook conflict. Never try to solve that conflict by deleting the local cursor.

A malformed batch is refused atomically and retried with bounded backoff. If the same malformed payload persists, intake stays blocked until the provider or adapter is repaired. This teaching implementation deliberately does not skip a whole batch after a retry count: that would silently discard valid messages alongside the malformed member. Stop polling, preserve the cursor and diagnostic context without the token, correct the parser against a retained synthetic reproduction, and resume at the same cursor. Local status, approval and receipt commands remain available during this channel outage.

An outbound unknown or interrupted sending report is not automatically replayed, and this version has no operator resend command. Inspect the actual conversation and the report rows locally for the current state. A later changed business result gets a new report identity; it does not erase the earlier uncertainty or pretend that Lucy read it. This conservative policy favors avoiding duplicate unsolicited messages over guaranteed delivery.

Active recall

Without rereading: why does prefill read tokens faster than decode writes them? Where does the $n^2$ term come from? What does streaming change, and what does it not? How many prompt tokens does an eight-turn conversation reread without a prefix cache, and with one? Which identity deduplicates a request, and which identity selects memory? Why must accepted work and the inbound cursor share a transaction? What does the poller lease prevent that a numeric operator allowlist does not? Why is a DONE work record compatible with an UNKNOWN report delivery? What does a message identifier establish, and what must still be observed on the phone?

Vocabulary

Prefill reads the prompt; decode writes the reply one token at a time. Attention compares each token with those before it. Streaming sends a reply piece by piece. A prefix cache keeps the model's state for a prompt's start and reuses it. A channel adapter translates between one messaging protocol and runtime work. An inbound cursor identifies the next remote update position to request. A durable origin binds one external delivery to its immutable local request. A poller lease authorizes one local consumer to commit a bot account's intake. An outbound receipt records the service's successful message identifier, while an unknown delivery preserves uncertainty after an admitted send lacks a committed completion.

Summary

Prefill read prompt tokens 13 to 26 times faster than decode wrote reply tokens, and its cost per token grew with the prompt's length because of attention. Streaming put the first words in front of Lucy in about a tenth of a second. Keeping the start of the prompt stable let the prefix cache cut each turn's prefill five- to sevenfold, which a token counter alone did not show.

You built a narrow bot API client, admitted only explicitly allowed private senders, bound accepted requests to stable sessions, and committed intake with its cursor. The existing skill and model loop now handle work originating from the phone. Results have a separate delivery state and a durable successful receipt, so losing a network response does not silently trigger a duplicate report. The next chapter creates work from schedules and stock events while the operator is away.

Keep building with Prof Rod

Found this material through a colleague, classroom or shared download? Get the complete book at profrod.ai/book and join the Prof Rod learner community. Bring one result, one question or one failure you learned from. Share this resource with another learner and keep its source links with it so they can find the full course and future updates.