#s16: Team Protocols — Teammates Need Agreements

s01 → ... → s14 → s15 → s16 → s17 → s18 → s19 → s20

"Teammates need agreements" — request-response pattern drives all negotiation.

Harness Layer: Protocols — Structured handshakes between agents.

#The Problem

s15's teammates can work, but coordination is loose: Lead sends a message, teammate replies, no structured protocol. Two scenarios expose the gap:

Shutdown: Lead wants Alice to shut down. Killing the thread outright leaves half-written files on disk. A handshake is needed: Lead sends a request, Alice confirms after wrapping up.

Plan approval: Bob wants to refactor the auth module, a high-risk operation. Lead should review Bob's plan first, approve before Bob proceeds.

Both scenarios share the same structure: one side sends a request, the other replies, both linked by the same ID. A state machine tracks: pending → approved / rejected.

#The Solution

)

Teaching code continues the agent capability arc from earlier chapters and adds structured protocols on top of S15's team communication. To stay focused on the protocol mechanism, it omits full error recovery, memory, and skill systems. Added: ProtocolState (request state tracking), dispatch_message (routes incoming messages by type to handlers), match_response (correlates response to request via request_id, with type validation).

Two protocols, one mechanism:

Teaching version demonstrates the request-response message flow for plan approval, but does not implement execution gating (intercepting bash/write_file when not approved). Real CC has a permission gating mechanism for teammates.

#How It Works

#ProtocolState: Request State

Each protocol request creates a state record tracking who sent it, to whom, current status, and payload:

@dataclass
class ProtocolState:
    request_id: str      # Unique ID, e.g. "req_004281"
    type: str            # "shutdown" | "plan_approval"
    sender: str          # Sender
    target: str          # Recipient
    status: str          # pending | approved | rejected
    payload: str         # Plan text or shutdown reason
    created_at: float    # Timestamp

pending_requests: dict[str, ProtocolState] = {}

A record is created when sending a request, found via request_id when receiving a response, and its status updated.

#Four-Step Protocol Flow

Using shutdown as an example, the full chain:

1. Lead sends request
   req_id = new_request_id()           # "req_004281"
   pending_requests[req_id] = ProtocolState(type="shutdown", status="pending", ...)
   BUS.send("lead", "alice", "shutdown_request", metadata={"request_id": req_id})

2. Teammate receives → dispatch
   inbox = BUS.read_inbox("alice")
   msg_type = msg["type"]              # "shutdown_request"
   → routed to handle_shutdown_request()

3. Teammate replies
   BUS.send("alice", "lead", "shutdown_response",
            metadata={"request_id": req_id, "approve": True})

4. Lead receives response → match
   match_response("shutdown_response", req_id, approve=True)
   pending_requests[req_id].status = "approved"

request_id is the correlation key across the entire chain: the request carries it out, the response carries it back.

#dispatch_message: Route by Type

A teammate's inbox receives both plain messages and protocol messages. handle_inbox_message dispatches by message type:

def handle_inbox_message(name, msg, messages):
    msg_type = msg.get("type", "message")
    req_id = msg.get("metadata", {}).get("request_id", "")

    if msg_type == "shutdown_request":
        BUS.send(name, "lead", "Shutting down.", "shutdown_response",
                 {"request_id": req_id, "approve": True})
        return True   # Stop the loop

    if msg_type == "plan_approval_response":
        approve = msg["metadata"].get("approve", False)
        messages.append({"role": "user",
            "content": "[Plan approved]" if approve else "[Plan rejected]"})
    return False       # Continue

Adding a new protocol type means adding a new if branch.

#match_response: Type Validation

match_response doesn't just find state by request_id, it also validates that the response type matches the request type:

def match_response(response_type, request_id, approve):
    state = pending_requests.get(request_id)
    if not state:
        return
    if state.type == "shutdown" and response_type != "shutdown_response":
        return  # type mismatch, skip
    if state.type == "plan_approval" and response_type != "plan_approval_response":
        return
    if state.status != "pending":
        return  # already resolved, skip duplicate
    state.status = "approved" if approve else "rejected"

A shutdown_response cannot accidentally approve a plan_approval request.

#Unified Inbox Consumer: consume_lead_inbox

Both the check_inbox tool and the main loop call the same consume_lead_inbox() function, routing protocol messages before returning remaining content. This prevents messages from being consumed without protocol state updates:

def consume_lead_inbox(route_protocol=True) -> list[dict]:
    msgs = BUS.read_inbox("lead")
    if route_protocol:
        for msg in msgs:
            meta = msg.get("metadata", {})
            req_id = meta.get("request_id", "")
            msg_type = msg.get("type", "")
            if req_id and msg_type.endswith("_response"):
                match_response(msg_type, req_id, meta.get("approve", False))
    return msgs

The main loop also injects inbox messages into history so the LLM can see and react to them.

#Teammate Idle Loop: Wait Instead of Exit

s15's teammates exit after 10 rounds. s16's teammates enter idle waiting after the LLM returns a non-tool_use response: poll inbox, respond to shutdown_request and exit, or continue working on new messages.

LLM returns non-tool_use
  → idle: poll inbox every second
  → receives shutdown_request → reply shutdown_response → exit
  → receives new message → inject into messages → continue LLM turn

Teaching version omits idle_notification to Lead. Real CC sends idle_notification when idle, so Lead knows the teammate is free for new tasks.

#Putting It Together

1. Lead: "Have Alice create a file, then shut her down"
2. Lead → spawn_teammate("alice", "backend", "Create config.py")
3. alice thread starts → write_file("config.py", "...") → done → idle
4. Lead → request_shutdown("alice")
   → BUS.send("shutdown_request", {request_id: "req_000142"})
5. alice idle poll receives → handle_shutdown_request
   → BUS.send("shutdown_response", {request_id: "req_000142", approve: True})
6. Lead consume_lead_inbox → match_response("req_000142", approve=True)
   → pending_requests["req_000142"].status = "approved"
   → inbox message injected into history, LLM sees shutdown result

Shutdown handshake complete: request → confirm → shutdown. Every step tracked by request_id.

#Changes from s15

#Try It

cd learn-claude-code
python s16_team_protocols/code.py

Try these prompts:

1. Spawn alice as a backend dev. Ask her to create a file. Then request her shutdown.
2. Spawn bob with a refactoring task. Have him submit a plan first. Then review and approve it.

What to observe: Is the shutdown handshake complete (request → confirm → shutdown)? Does pending_requests state transition correctly? Is request_id consistent between request and response? Can the idle teammate receive shutdown_request?

#What's Next

In s15-s16, Lead must assign tasks to each teammate. "Alice does this, Bob does that." With 10 unclaimed tasks on the board, Lead has to manually assign each one.

What if teammates could check the board and claim tasks themselves? Lead only needs to create tasks; teammates discover, claim, and complete them on their own.

s17 Autonomous Agents → Self-organizing teammates, no leader assignment needed.