core.proxy_adapter module

Proxy PlatformAdapter that publishes to Redis instead of calling SDK APIs.

class core.proxy_adapter.ProxyPlatformAdapter(event_bus, platform, trace_id=None)

Bases: object

Drop-in replacement for PlatformAdapter on worker nodes.

Instead of holding a live Discord/Matrix client, this adapter serializes all send() calls into outbound stream envelopes. The gateway consumes these and dispatches to the real adapter.

Parameters:
__init__(event_bus, platform, trace_id=None)

Initialize the proxy adapter for one platform on a worker node.

Stores the shared RedisEventBus used to publish outbound envelopes, the platform name (e.g. "discord") that selects the destination stream, and an optional trace_id used to derive deterministic per-action idempotency keys. A defaultdict(int) of per-action-type counters backs that key derivation in _maybe_inject_key().

This constructor has no side effects of its own. It is instantiated by the inference worker when it needs an adapter to hand to the message processor / tools (via ToolContext.adapter) so their send calls become Redis envelopes instead of live SDK calls.

Parameters:

  • event_bus (RedisEventBus) – Shared event bus whose publish_outbound writes the outbound Redis Stream and whose _redis backs RPC replies.

  • platform (str) – Platform identifier selecting the outbound stream and gateway consumer group (e.g. "discord", "matrix").

  • trace_id (str | None) – Optional request trace id; when set, send actions get a derived message_key for gateway-side idempotency.

Return type:

None

bind_effect_scope(operation_id)

Bind child outbound IDs to one durable logical tool call.

Return type:

tuple[Token, Token, Token]

Parameters:

operation_id (str)

reset_effect_scope(tokens)

Restore the task’s prior child-effect identity context.

Return type:

None

Parameters:

tokens (tuple[Token, Token, Token])

effect_scope_is_ambiguous()

Return whether a durable child delivery had an unknown outcome.

Return type:

bool

async send(channel_id, text, **kwargs)

Publish a text message envelope for the Gateway to deliver.

Serializes a type: "message" envelope (channel, text, and any non-None extra kwargs such as reply_to) instead of calling a live SDK; the Gateway’s outbound consumer reconstructs and sends it on the real client.

It first calls _maybe_inject_key() to stamp a message_key for idempotency, then awaits event_bus.publish_outbound(platform, ...), which XADDs onto the sg:stream:outbound:{platform} stream. Called broadly across the worker: the message processor and command router (message_processor/processor.py, command_router.py), tools via ctx.adapter.send / platform.send, and task_manager.py all use it to emit chat output.

Parameters:

  • channel_id (str) – Destination channel/instance identifier.

  • text (str) – Message body to send.

  • **kwargs – Extra envelope fields (e.g. reply_to, message_key); None values are dropped.

Return type:

None

async send_file(channel_id, file_data, filename='file', mimetype='application/octet-stream', **kwargs)

Publish a file-attachment envelope for the Gateway to upload.

Base64-encodes the raw file_data so the bytes survive Redis Stream serialization, then emits a type: "file" envelope carrying the channel, encoded payload, filename, and mimetype.

It calls _maybe_inject_key() (action "file") for idempotency. Inside a durable tool-effect scope it waits for the Gateway ledger’s cached upload receipt; otherwise it XADDs an ordinary outbound file envelope. The Gateway decodes and uploads via the live adapter’s send_file. Called by the outbound consumer’s own delegation path and by many media tools through ctx.adapter.send_file (e.g. tools/generate_image.py, tools/elevenlabs_tts.py, tools/render_mermaid.py) and background agents to attach generated artifacts.

Parameters:

  • channel_id (str) – Destination channel/instance identifier.

  • file_data (bytes) – Raw file bytes to send (base64-encoded for transport).

  • filename (str) – Suggested attachment filename. Defaults to "file".

  • mimetype (str) – MIME type of the attachment. Defaults to "application/octet-stream".

  • **kwargs – Extra envelope fields (e.g. caption text, message_key); None values are dropped.

Return type:

str | None

async send_with_buttons(channel_id, text, view=None, **kwargs)

Publish a text-with-buttons envelope, normalizing the button spec.

Accepts either a plain list of button dicts or a Discord-style view object and flattens it into a serializable buttons list (label, custom_id, optional emoji, and a string-ified style) since live UI view objects cannot cross Redis. An explicit buttons kwarg, if present, overrides whatever was derived from view.

After building the list it calls _maybe_inject_key() (action "buttons") and awaits event_bus.publish_outbound to XADD a type: "buttons" envelope onto sg:stream:outbound:{platform}; the Gateway’s outbound consumer rebuilds the interactive view. Called by the message processor’s send path (message_processor/generate_and_send.py, processor.py) and the outbound consumer’s delegation branch.

Parameters:

  • channel_id (str) – Destination channel/instance identifier.

  • text (str) – Message body shown above the buttons.

  • view (Any) – Either a list of button dicts or a view object exposing children with label/custom_id/emoji/style.

  • **kwargs – Extra envelope fields; an explicit buttons key replaces the derived list. None values are dropped.

Return type:

None

async send_private_codex_authorization(channel_id, user_id, text)

Atomically re-prove and deliver a device code on the Gateway.

The worker has no live participant state, so this uses a trace-bound, restart-stable durable RPC. The Gateway’s real adapter performs the privacy proof immediately before the platform send and caches either the delivery receipt or the deterministic pre-send rejection.

Return type:

dict[str, Any]

Parameters:
async deliver_message_durable(channel_id, text, *, rpc_call_id, view=None, buttons=None, **kwargs)

Deliver a final message through the Gateway effect ledger.

Unlike send(), this waits for a terminal, ledger-authenticated Gateway RPC result. Reusing rpc_call_id therefore returns the cached platform message id without sending the message again.

Message semantics plus the stable cache message_key cross the RPC boundary. The Gateway uses that key to re-sync the platform message ID and NCM/CoT aliases exactly as ordinary outbound dispatch does.

Return type:

Any

Parameters:
async deliver_stream_chunk_durable(channel_id, text, *, stream_id, chunk_index, reply_to=None)

Deliver one streaming post with a restart-stable effect identity.

stream_id identifies one logical generated response and chunk_index identifies its ordered post. The derived Gateway RPC ID and message-cache key are deterministic, so retrying the same chunk returns its cached receipt instead of posting a duplicate. Different chunks remain independent durable effects and can be awaited in order.

Return type:

Any

Parameters:

  • channel_id (str)

  • text (str)

  • stream_id (str)

  • chunk_index (int)

  • reply_to (str | int | None)

async edit_message_durable(channel_id, message_id, content, *, stream_id, lane, chunk_index)

Set one stream message’s exact content through the Gateway ledger.

The effect identity includes the content digest. Replaying the same lane/content is cached; a newer draft naturally receives a different identity. Advancing lane provides an explicit repair attempt after an ambiguous edit, which is safe because Discord edits are exact-set mutations rather than appends.

Return type:

dict[str, Any]

Parameters:
async delete_message_durable(channel_id, message_id, *, stream_id, lane, chunk_index)

Idempotently remove one stream message through the Gateway ledger.

Return type:

dict[str, Any]

Parameters:
async bind_stream_messages_durable(channel_id, message_ids, *, stream_id)

Bind a final ordered stream receipt to message-cache aliases.

Return type:

dict[str, Any]

Parameters:
async deliver_file_durable(channel_id, file_data, *, filename='file', mimetype='application/octet-stream', rpc_call_id, message_key)

Upload one file through the Gateway’s durable effect ledger.

Return type:

Any

Parameters:

  • channel_id (str)

  • file_data (bytes)

  • filename (str)

  • mimetype (str)

  • rpc_call_id (str)

  • message_key (str)

property name: str

Return this adapter’s platform identifier.

Exposes the platform passed at construction (e.g. "discord") so the proxy satisfies the same read-only name contract as a live PlatformAdapter. Read-only with no side effects.

Read by the message processor and helpers that index adapters by name, e.g. the {a.name: a for a in adapters} maps built in message_processor/processor.py, generate_and_send.py, and channel_heartbeat.py.

Returns:

The platform name.

Return type:

str

property bot_identity: dict[str, str]

Return the bot’s identity descriptor for this platform.

Provides the static identity the live PlatformAdapter would normally derive from its connected client, which the worker-side proxy cannot reach. The mapping carries this adapter’s platform plus the fixed "Stargazer" user_id and display name. Read-only and built fresh on each access with no side effects.

Satisfies the bot_identity slot of the adapter interface; no internal caller in this repo reads it on the proxy specifically.

Returns:

Keys platform, user_id, and display_name.

Return type:

dict[str, str]

async send_typing(channel_id)

Publish a one-shot typing indicator for the channel.

Emits a type: "typing" envelope (no idempotency key, since typing is transient/fire-and-forget) by awaiting event_bus.publish_outbound onto sg:stream:outbound:{platform}; the Gateway’s outbound consumer triggers a single typing pulse on the live client. Invoked from that same consumer’s delegation branch (core/outbound_consumer.py) as the fallback when continuous typing is not requested.

Parameters:

channel_id (str) – Channel/instance to show the typing indicator in.

Return type:

None

async start_typing(channel_id)

Publish a request to begin a sustained typing indicator.

Emits a type: "start_typing" envelope via event_bus.publish_outbound onto sg:stream:outbound:{platform}; the Gateway’s outbound consumer keeps a continuous typing loop alive on the live client until a matching stop_typing() arrives. Called by the message processor around long operations (message_processor/processor.py) and by the outbound consumer’s own delegation branch (core/outbound_consumer.py).

Parameters:

channel_id (str) – Channel/instance to begin showing typing in.

Return type:

None

async stop_typing(channel_id)

Publish a request to stop a sustained typing indicator.

Emits a type: "stop_typing" envelope via event_bus.publish_outbound onto sg:stream:outbound:{platform}; the Gateway’s outbound consumer tears down the typing loop started by start_typing(). Called by the message processor when long operations finish (message_processor/processor.py).

Parameters:

channel_id (str) – Channel/instance to stop showing typing in.

Return type:

None

async add_reaction(channel_id, message_id, emoji, **kwargs)

Publish a reaction envelope to add an emoji to a message.

Emits a type: "reaction" envelope carrying the target message_id and emoji; the Gateway’s outbound consumer adds the reaction on the live client.

It calls _maybe_inject_key() (action "reaction") for idempotency and awaits event_bus.publish_outbound onto sg:stream:outbound:{platform}. Invoked from the outbound consumer’s delegation branch (core/outbound_consumer.py); higher-level reaction tooling (e.g. tools/discord_react.py) reaches this path through the adapter interface.

Parameters:

  • channel_id (str) – Channel/instance containing the target message.

  • message_id (str) – Identifier of the message to react to.

  • emoji (str) – Emoji (unicode or platform shortcode) to add.

  • **kwargs – Extra envelope fields (e.g. message_key); None values are dropped.

Return type:

None

async set_presence(text, emoji=None)

Publish a presence/status update for the Gateway to apply.

Presence is a global (non-channel) action: the Gateway’s outbound consumer calls adapter.set_presence on the live client. No channel_id or message_key is attached — it is fire-and-forget and must not go through the per-message idempotency/claim path.

Return type:

None

Parameters:

  • text (str)

  • emoji (str | None)

async fetch_history(channel_id, limit=10, **kwargs)

Alias for fetch_channel_history().

Tools call the PlatformAdapter interface method name fetch_history; the proxy implements the RPC under fetch_channel_history. Without this alias those calls raise AttributeError on worker nodes.

Parameters:

  • channel_id (str)

  • limit (int)

async fetch_channel_history(channel_id, limit=10, **kwargs)

Fetch recent channel history from the Gateway over a Redis-Streams RPC.

A worker node has no live client, so it cannot read history locally. This publishes a type: "rpc_request" envelope (method fetch_channel_history) onto the platform’s outbound stream via event_bus.publish_outbound and then blocks on the event bus’s Redis client with blpop (30s) on a unique sg:rpc:reply:{uuid} list key; the Gateway’s core.outbound_consumer.OutboundStreamConsumer._handle_rpc_request() runs the real fetch_history and pushes a JSON result back, which is decoded and returned. On timeout or any error it logs and returns an empty list, and it always best-effort deletes the reply key in a finally so abandoned replies do not accumulate.

Parameters:

  • channel_id (str) – Channel/instance whose recent messages to fetch.

  • limit (int) – Maximum number of messages to request. Defaults to 10.

  • **kwargs – Extra arguments forwarded to the Gateway-side fetch.

Returns:

Decoded history entries, or an empty list on timeout/error.

Return type:

list[HistoricalMessage]

async delegate_to_gateway(method, *, rpc_call_id=None, **kwargs)

Run a privileged Gateway action with one stable effect identity.

Every transport retry republishes the exact same rpc_call_id, payload, fingerprint, and reply key. The Gateway’s durable effect ledger therefore treats retries and stream redelivery as the same logical operation rather than executing the platform mutation again. Replies are accepted only when both their operation id and payload fingerprint match this request.

Exhausting the reply budget is explicitly AMBIGUOUS: the caller must not automatically issue a new operation id because the platform effect may already have happened. This is deliberately at-most-once automatic execution, not an unsafe claim of transactional exactly-once delivery.

Parameters:

  • method (str) – Name of the Gateway-side adapter method to invoke.

  • rpc_call_id (str | None) – Stable logical operation identity supplied by the tool executor. It is mandatory because a locally generated identity would not survive an inference-process restart.

  • **kwargs – Keyword arguments forwarded to that method as rpc_args.

Returns:

The decoded result on a terminal Gateway reply. On reply

exhaustion, a non-retryable AMBIGUOUS error mapping.

Return type:

Any

async should_skip_channel_heartbeat(channel_id)

Ask the Gateway over RPC whether this channel’s heartbeat should be skipped.

Heartbeat suppression depends on live presence/state that only the Gateway’s real client holds, so the worker proxies the decision. It publishes a type: "rpc_request" envelope (method should_skip_channel_heartbeat) via event_bus.publish_outbound and blocks on a unique sg:rpc:reply:{uuid} list with blpop (10s) on the event bus’s Redis client; the Gateway’s _handle_rpc_request calls the real adapter method and pushes back a JSON boolean. On timeout or any error it logs and conservatively returns False (do not skip), and it always best-effort deletes the reply key in a finally.

Called by the channel heartbeat loop (message_processor/channel_heartbeat.py) before emitting a heartbeat, and covered by the RPC migration tests (tests/core/migration/test_proxy_adapter_rpc.py).

Parameters:

channel_id (str) – Channel/instance whose heartbeat eligibility to check.

Returns:

True to skip the heartbeat; False to proceed (also the safe default on timeout/error).

Return type:

bool

async get_channel_name(channel_id)

Return a synthetic display name for a channel on the worker node.

Worker nodes do not hold a live client and cannot resolve real channel names cheaply, so this returns the deterministic placeholder "channel_{channel_id}" rather than performing an RPC. It has no side effects and makes no network calls. No internal callers were found in this repo; it satisfies the PlatformAdapter interface so code expecting get_channel_name does not raise on worker nodes.

Parameters:

channel_id (str) – Channel/instance identifier to label.

Returns:

The placeholder name "channel_{channel_id}".

Return type:

str