Cheetah / Cheetah documentation

Response waiting and history

History is the source of complete retained messages. RPC correlation and response notification carry only enough metadata to wake application code; they do not replace the stored record.

Read the full messages for a command

history = app_node.history_for(user_id)
messages = await history.get_by_command_id(resolution.command_id)

terminal = next(
    message for message in messages
    if message.get("kind") in {"result", "error"}
)
payload = terminal.get("payload", {})

Command lookup can return progress, terminal responses, and other retained records associated with the command. The interface does not guarantee their returned order. Sort by sequence, timestamp, or the application-relevant field when order matters.

History is not a complete outbound command ledger. Ordinary App dispatch does not append the command message. Telemetry is intentionally excluded from user history, and some control paths have separate records. Treat history as the configured retained-message store, not as proof that it observed every fact in the system.

Wait for selected response messages

Use wait_for_responses() when progress is itself an application input or when code wants any matching terminal message rather than one RPC resolution:

messages = await scoped.wait_for_responses(
    command_ids=[command_id],
    timeout_ms=10_000,
    consume=False,
    limit=100,
)

Choose exactly one selector form:

• command_ids=[...] selects messages associated with those commands;
• stream_suffixes=[...] selects user-scoped streams.

Passing both or neither raises ValueError. A negative timeout raises ValueError; a non-positive limit returns an empty list. The helper responds to progress, result, and error. Ordinary event messages can be retained, but they do not wake this command-response waiter.

The notifier registration happens before the first history read. If a response arrives during setup, either the query sees the stored message or the notifier wakes the waiter; there is no ordinary check-then-subscribe gap. REST and App construction must therefore receive compatible response notifiers from the same component family.

Consumption is a single-reader convenience

consume defaults to True. The helper reads matching messages and then asks history to delete those explicit records.

This is not an atomic distributed queue claim. Do not have several application tasks consume the same command IDs or streams concurrently. The current convenience method returns the messages but not the underlying HistoryDeleteResult; a partial deletion failure can make a returned message visible again. Use consume=False for observation, diagnostics, multiple readers, or any workflow that still needs retention.

Page through one stream

The scoped App view exposes the common query:

page = await scoped.get_history(
    "business-results",
    cursor=None,
    limit=100,
)

The dedicated scoped history view adds timestamp filtering, append, command lookup, and explicit deletion:

history = app_node.history_for(user_id)

page = await history.query(
    "business-results",
    cursor=None,
    limit=100,
    after_timestamp_ms=None,
)

HistoryPage contains:

Cursors belong to the selected history backend and stream contract. Do not parse, manufacture, or carry them to a different backend. Cursor-based incremental reading is the portable checkpoint model. Current in-memory and Redis implementations differ in the precise basis of after_timestamp_ms: one compares the message timestamp and the other uses Redis stream insertion position.

Append and delete through a scoped view

stored_id = await history.append(
    "business-results",
    {
        "message_id": "result-42",
        "kind": "event",
        "payload": {"value": 42},
    },
)

History append identity is (canonical stream_key, message_id). Repeating the same message ID in one stream retains the first record. The same ID in another stream, including another principal's stream, is independent. This idempotency scope is not the same as REST ingestion deduplication and does not create exactly-once command execution.

Explicit deletion uses full canonical stream keys and message IDs:

from servercheetah.types.history import HistoryMessageRef

result = await history.delete([
    HistoryMessageRef(
        stream_key=f"{history.user_id}/business-results",
        message_id="result-42",
    )
])

The scoped view rejects references outside its bound user. HistoryDeleteResult reports the requested and deleted counts, missing references, and non-fatal cleanup failures. Deleting a missing record is idempotent and reported as missing rather than raised as an error.

Retention and backend qualifications

The in-memory store is bounded process-local state. The Redis store supplies distributed streams and indexes with configured length and lifetime policies. Optional spillover can move large top-level payloads to a separate store while Redis retains the ordered reference.

For managed spillover, newly written references and payloads receive one absolute expiry. The managed store owns logical expiry and physical cleanup; the built-in file store performs throttled cleanup while managed writes continue and exposes an explicit cleanup hook for scheduled or decommission work. A shared file store requires access from every reading node and reasonably synchronized clocks. Do not interpret Redis stream length alone as the total external-storage bound.

Normal response ingestion stores a message before notifying response waiters and resolving a terminal RPC. The deliberate malformed-message path can resolve a recognizable RPC with a synthetic error before that error is stored, so prompt failure is still possible when history storage itself is impaired.

Continue to Presence and payload references for the App helpers that deliberately write and sign history-backed values.