Mailboxes and shared work

A mailbox lets one agent process hand work to another while each keeps its own authority. The producer sends under a send_jobs grant, consumers read, claim, and complete under theirs, and the kernel admits each call with its guard checks, call budget, and signed receipt.

The mailbox is a bounded native tool server on the process host. Its payload is data: a message is never an authority token, and naming an agent inside one proves nothing about who sent it. Every step uses the jobs channel from the report project.

1. Grant each endpoint its own operation

Declare the channel in the host configuration, then grant each holder only the tools its role needs. The four limits solve different capacity problems: pending message count, pending bytes, single payload size, and lifetime message count. Plan all four.

{
  "mailboxes": [
    {
      "id": "jobs",
      "renewable_leases": true,
      "limits": {
        "max_pending_messages": 32,
        "max_pending_bytes": 1048576,
        "max_message_bytes": 65536,
        "max_messages": 256
      }
    }
  ]
}

A sender-attesting host records the kernel-selected process as each message’s sender. Attestation identifies the admitted process; the code that ran inside it is outside that claim. A broad parent grant can still send, so narrow it before delegation.

2. Send a message and read it back

The producer sends under a message key, and the consumer reads a snapshot after a cursor. Both check the decision and the materialized result before advancing application state.

def receive_report(client):
    response = client.invoke(
        "report/receive/v1",
        "chio-ipc",
        "receive_jobs",
        {
            "after_sequence": "0",
            "limit": 1,
        },
    )
    result = completed(response)
    if result["status"] != "received" or len(result["messages"]) != 1:
        raise RuntimeError("Report is not available in this observation")
    message = result["messages"][0]
    if message["sender"] != client.inspect()["process_id"]:
        raise RuntimeError("Unexpected sender in this self-send exercise")
    return message

export async function receiveReport(
  client: ProcessClient,
): Promise<Record<string, Json>> {
  const response = await client.invoke(
    "report/receive/v1",
    "chio-ipc",
    "receive_jobs",
    {
      after_sequence: "0",
      limit: 1,
    },
  );
  const result = completed(response);
  assert.equal(result.status, "received");
  assert.ok(Array.isArray(result.messages) && result.messages.length === 1);
  const message = object(result.messages[0]);
  assert.equal(message.sender, (await client.inspect()).process_id);
  return message;
}

async fn receive_report(client: &ProcessClient) -> Result<Value> {
    let response = client
        .invoke(
            "report/receive/v1",
            "chio-ipc",
            "receive_jobs",
            json!({"after_sequence":"0","limit":1}),
            None,
        )
        .await?;
    let result = completed(&response)?;
    let messages = result["messages"].as_array().ok_or("Missing messages")?;
    if result["status"] != "received" || messages.len() != 1 {
        return Err("Report unavailable".into());
    }
    if messages[0]["sender"] != client.inspect().await?.process_id {
        return Err("Unexpected sender".into());
    }
    Ok(messages[0].clone())
}

You should see status received with one message carrying sequence, payload, and the authenticated sender; the producer’s message_key stays with the producer. These functions come from one executable self-send exercise per language, so run the complete three-language recipe project before adapting them. Choosing a language here selects it in every other code tab across the docs.

A repeated send with the same message key, sender, and payload returns the original sequence, even after acknowledgement; a changed payload or sender conflicts. A repeated poll key returns its original snapshot, so observe later messages under a new logical poll key.

Sender identity establishes origin, and the payload still needs validation: check the expected producer and bound arrays, strings, and numbers before calculating. A host that does not attest senders delivers sender: null: treat it as missing attribution, never as the current process, the channel owner, or the last known producer, and stop or route the message for reconciliation when the application requires a specific origin. A message key binds deduplication to sender and payload, so another producer’s key gives no power to overwrite its message.

def completed(response):
    if response["verdict"] != "allow":
        raise RuntimeError("Decision requires attention")
    if response["terminal_state"]["state"] != "completed":
        raise RuntimeError("Original outcome remains uncertain")
    if not response["output"] or response["output"]["kind"] != "value":
        raise RuntimeError("Expected a materialized response")
    value = response["output"]["value"]
    if value.get("isError"):
        raise RuntimeError("Tool returned a completed error")
    return value

sent = producer.invoke(
    "handoff-1", "chio-ipc", "send_jobs",
    {"message_key": "job-17", "payload": {"document": "report-17"}},
)
if completed(sent)["status"] != "sent":
    raise RuntimeError("Mailbox did not accept the handoff")
received = consumer.invoke(
    "poll-1", "chio-ipc", "receive_jobs",
    {"after_sequence": "0", "limit": 1},
)
value = completed(received)
if value["status"] != "received" or len(value["messages"]) != 1:
    raise RuntimeError("Expected one message in this observation")
message = value["messages"][0]
if message["sender"] != "producer" or message["payload"] != {"document": "report-17"}:
    raise RuntimeError("Unexpected sender or application payload")

# Persist this application's progress without replacing other owners' slots.
cp = consumer.inspect()["checkpoint"]
previous = cp["value"] or {}
progress = {"processed": "report-17"}
if "handoff.v1" in previous and previous["handoff.v1"] != progress:
    raise RuntimeError("Checkpoint belongs to different work")
if "handoff.v1" not in previous:
    consumer.checkpoint(cp["revision"], {**previous, "handoff.v1": progress})
acknowledged = consumer.invoke(
    "ack-1", "chio-ipc", "ack_jobs",
    {"through_sequence": message["sequence"]},
)
if completed(acknowledged)["status"] != "acknowledged":
    raise RuntimeError("Acknowledgement did not complete")

3. Save progress, then acknowledge

Receiving leaves a message pending. Validate the payload and checkpoint application progress in your own slot, as the Python handoff does with handoff.v1, before you acknowledge.

def acknowledge_report(client, sequence):
    # Commit application progress before this step; recovery repeats its key.
    response = client.invoke(
        "report/ack/v1",
        "chio-ipc",
        "ack_jobs",
        {
            "through_sequence": sequence,
        },
    )
    if completed(response)["status"] != "acknowledged":
        raise RuntimeError("Acknowledgement did not complete")
    return response

export async function acknowledgeReport(
  client: ProcessClient,
  sequence: string,
): Promise<ToolResult> {
  // Commit application progress before this step; recovery repeats its key.
  const response = await client.invoke(
    "report/ack/v1",
    "chio-ipc",
    "ack_jobs",
    {
      through_sequence: sequence,
    },
  );
  assert.equal(completed(response).status, "acknowledged");
  return response;
}

async fn acknowledge_report(client: &ProcessClient, sequence: &str) -> Result<()> {
    // Commit application progress before this step; recovery repeats its key.
    let response = client
        .invoke(
            "report/ack/v1",
            "chio-ipc",
            "ack_jobs",
            json!({"through_sequence":sequence}),
            None,
        )
        .await?;
    if completed(&response)?["status"] != "acknowledged" {
        return Err("Ack did not complete".into());
    }
    Ok(())
}

You should see status acknowledged. Acknowledgement advances a channel-wide watermark through a sequence and releases queue capacity, which is different from secure erasure: retained calls and receipts can still refer to messages removed from pending delivery.

Because the watermark covers the whole channel, one consumer’s acknowledgement can discard messages another consumer has claimed or has yet to process. Give ack_jobs to a competing-claim worker only when the application deliberately grants that administrative effect.

4. Poll with a durable cursor

A service that polls while producers run treats each receive as a durable observation. A receive that returns zero messages is still a completed logical operation, and replaying it returns the same empty snapshot.

cp = client.inspect()["checkpoint"]
state = cp["value"] or {}
poll = state.get("jobs.poll.v1", {"ordinal": 0, "after": "0"})
# Persist the selected identity before dispatch if this is the first poll.
if "jobs.poll.v1" not in state:
    client.checkpoint(cp["revision"], {**state, "jobs.poll.v1": poll})
response = client.invoke(
    f"jobs-poll-{poll['ordinal']}",
    "chio-ipc",
    "receive_jobs",
    {
        "after_sequence": poll["after"],
        "limit": 1,
        "wait_ms": 1000,
    },
)
if (
    response["verdict"] != "allow"
    or response["terminal_state"]["state"] != "completed"
):
    raise RuntimeError("Poll did not complete")
value = response["output"]["value"]
if value["status"] != "received":
    raise RuntimeError(f"Mailbox status: {value['status']}")
# For nonempty messages, validate sender/payload and checkpoint work before ack.
if not value["messages"]:
    cp = client.inspect()["checkpoint"]
    client.checkpoint(
        cp["revision"],
        {
            **(cp["value"] or {}),
            "jobs.poll.v1": {
                "ordinal": poll["ordinal"] + 1,
                "after": value["next_sequence"],
            },
        },
    )

You should see the cursor and the next poll ordinal advance together, and only after the result is interpreted and any required application progress is saved. The excerpt is one poll: a scheduler chooses when to run the next one and respects the remaining call budget. Optional wait_ms is bounded at 30,000 ms, and sequence and cursor values stay decimal strings.

The first-process starter receives once, because its dependency guarantees the message already exists. A service whose producers run concurrently needs this cursor scheme.

5. Claim work when consumers compete

claim_jobs leases pending work and returns a claim generation. complete_jobs consumes a message only under its sequence and current claim.

claimed = worker.invoke(
    "claim-17", "chio-ipc", "claim_jobs",
    {"limit": 2, "lease_ms": 30000},
)
for message in completed(claimed)["messages"]:
    # Application-supplied: validate sender/payload and retain idempotent progress.
    # Any resource mutation must also enforce its current assignment.
    handle(message)
    response = worker.invoke(
        "complete-" + message["sequence"], "chio-ipc", "complete_jobs",
        {"sequence": message["sequence"], "claim": message["claim"]},
    )
    if completed(response)["status"] != "completed":
        raise RuntimeError("Completion refused; stop relying on this claim")

You should see status completed for each claimed message. A claim accepts up to 16 messages, with a lease from 1,000 to 300,000 ms. Claiming leases a message and completion consumes it. When a lease lapses while a worker pauses, another consumer claims the same sequence under a new generation and the old holder’s completion is refused.

Stop relying on a claim as soon as its completion or renewal is refused or uncertain. An unknown claim outcome after host death stays uncertain until the lease expires. The resource the work changes enforces its own current assignment, as Shared resource ownership shows.

6. Renew a long claim before it lapses

Set renewable_leases when the host initializes the mailbox, and grant renew_jobs explicitly. A worker renews only its current, unexpired claim.

def renew_report(client, claimed):
    request = {
        "sequence": claimed["sequence"],
        "claim": claimed["claim"],
        "lease_ms": 30000,
    }
    # Persist heartbeat number 0 and request before dispatch. Reuse both after
    # transport uncertainty; a later heartbeat gets a new persisted number.
    response = client.invoke("report/heartbeat/0", "chio-ipc", "renew_jobs", request)
    completed(response)
    return response

export async function renewReport(
  client: ProcessClient,
  claimed: Record<string, Json>,
): Promise<ToolResult> {
  assert.ok(
    typeof claimed.sequence === "string" && typeof claimed.claim === "string",
  );
  const request = {
    sequence: claimed.sequence,
    claim: claimed.claim,
    lease_ms: 30000,
  };
  // Persist heartbeat number 0 and request before dispatch. Reuse both after
  // transport uncertainty; a later heartbeat gets a new persisted number.
  const response = await client.invoke(
    "report/heartbeat/0",
    "chio-ipc",
    "renew_jobs",
    request,
  );
  completed(response);
  return response;
}

async fn renew_report(client: &ProcessClient, claimed: &Value) -> Result<ToolResult> {
    let request = json!({"sequence":claimed["sequence"],"claim":claimed["claim"],"lease_ms":30000});
    // Persist heartbeat number 0 and request before dispatch. Reuse both after
    // transport uncertainty; a later heartbeat gets a new persisted number.
    let response = client
        .invoke(
            "report/heartbeat/0",
            "chio-ipc",
            "renew_jobs",
            request,
            None,
        )
        .await?;
    completed(&response)?;
    Ok(response)
}

You should see a new deadline. Save heartbeat ordinal 0 before its dispatch; after validating the returned deadline, checkpoint ordinal 1 before sending the next heartbeat. A retry of heartbeat 0 returns its original deadline and extends nothing. When the host loses a renewal outcome, treat the claim as uncertain and reconcile before relying on it; a further heartbeat leaves the first outcome unresolved.

claimed = client.invoke(
    "claim-report-42",
    "chio-ipc",
    "claim_jobs",
    {
        "limit": 1,
        "lease_ms": 30000,
    },
)
# After checking verdict, terminal state and native status, persist the returned
# sequence and claim before starting work. message is that retained selection.
renewed = client.invoke(
    "report-42-heartbeat-0",
    "chio-ipc",
    "renew_jobs",
    {
        "sequence": message["sequence"],
        "claim": message["claim"],
        "lease_ms": 30000,
    },
)
completed = client.invoke(
    "complete-report-42",
    "chio-ipc",
    "complete_jobs",
    {
        "sequence": message["sequence"],
        "claim": message["claim"],
    },
)

def claim_report(client):
    response = client.invoke(
        "report/claim/v1",
        "chio-ipc",
        "claim_jobs",
        {
            "limit": 1,
            "lease_ms": 30000,
        },
    )
    messages = completed(response)["messages"]
    if len(messages) != 1:
        raise RuntimeError("No claim in this observation")
    return messages[0]

export async function claimReport(
  client: ProcessClient,
): Promise<Record<string, Json>> {
  const response = await client.invoke(
    "report/claim/v1",
    "chio-ipc",
    "claim_jobs",
    {
      limit: 1,
      lease_ms: 30000,
    },
  );
  const messages = completed(response).messages;
  assert.ok(Array.isArray(messages) && messages.length === 1);
  return object(messages[0]);
}

async fn claim_report(client: &ProcessClient) -> Result<Value> {
    let response = client
        .invoke(
            "report/claim/v1",
            "chio-ipc",
            "claim_jobs",
            json!({"limit":1,"lease_ms":30000}),
            None,
        )
        .await?;
    let messages = completed(&response)?["messages"]
        .as_array()
        .ok_or("Missing messages")?;
    if messages.len() != 1 {
        return Err("No claim in this observation".into());
    }
    Ok(messages[0].clone())
}

7. Verify claims, renewal, and limits with the exercise

Download exercise-state.py and its complete policy, host and client setup, then install the project’s dependencies with uv sync. The script creates a fresh host and keeps its state for inspection.

uv run python exercise-state.py --state mailbox-exercise

{
  "checkpoint_conflict_and_blob_isolation": true,
  "stale_completion": {
    "verdict": "deny",
    "terminal_state": {
      "state": "completed"
    }
  },
  "replacement_completion": {
    "status": "completed",
    "sequence": "1"
  },
  "empty_poll_replayed": true,
  "full_then_sent": true,
  "lifetime_exhaustion": true,
  "expired_cursor": true
}

You should see the stale completion refused with verdict deny while the replacement completes sequence 1. The script checks the original renewal receipt, waits for the lease to expire, and observes the replacement claim before submitting the old completion. It also exercises checkpoint conflicts, polling, bounds, revocation, and cancellation, and confirms that the checkpoint survives sealed export and import.

The stale completion is a signed decision in the exercise host’s receipt store. List the mailbox’s refused completions:

chio receipt list --receipt-db mailbox-exercise/receipts.db --admin-all \
  --tool-name complete_jobs --outcome deny \
  | jq -c '{id, tool: .tool_name, reason: .decision.reason}'

{"id":"rcpt-01d2…6b3a","tool":"complete_jobs","reason":"claim superseded: sequence 1 is held under a newer claim generation"}

To repeat the replacement by hand, use two separately scoped consumers:

1. Claim. Consumer A claims one message; record its sequence, claim, and deadline.
2. Expire. Pause A past its deadline, then let B claim the same sequence.
3. Resume. Submit A’s original completion. The mailbox refuses it, and B’s work stays B’s.
4. Finish. B completes, and the message leaves pending delivery.
5. Renew. Repeat a successful renewal under its original key and compare the deadlines byte for byte, then issue the next checkpointed heartbeat key and observe a new renewal.

Check the mailbox refusal and the resource value separately. A signed mailbox denial covers the mailbox, so test the old worker’s write against the database in its own exercise.