Communication Subsystem

The platform communication subsystem delivers asynchronous output from running apps to connected clients. The default chat UI, app widgets, and custom frontends can consume the same typed envelopes over SSE or Socket.IO. App code can emit progress, deltas, files, tool lifecycle steps, custom events, and completion without owning the browser transport.

App REST operations and app-served MCP endpoints are sibling proc surfaces that sit next to this pipeline, not inside the event envelope itself. They can still emit communicator events during a request, but the underlying HTTP request/response or MCP exchange remains a separate surface.

The same communicator can record selected post-firewall envelopes into scoped buffers and hand batches to event sinks. The firewall runs first: an allowed event may be client-visible only or both visible and recorded; a blocked event is neither delivered nor recorded.

Do not confuse outbound streaming with the two inbound work buses. The Conversation Event Bus orders context for one user/conversation/agent lane. The Data Bus delivers app-owned domain mutations to declared handlers. Payload size and JSON shape do not choose between them. See Platform Architecture for the separation and Object Ecosystem & Ontologic Contracts for provider and surface contracts.

ChatCommunicator — App Producer API

Every app entrypoint receives a ChatCommunicator instance. It wraps Redis Pub/Sub with a typed API for outbound async events. Trusted child tool runtimes can carry a portable communicator specification and return events through the supervisor or side-file handoff; generated code does not receive Redis credentials or become an independent publisher. App-defined chat UIs, widgets, and custom frontends therefore see one stream contract across in-process and governed external execution.

External Events and Named-Service Events

external_events[] lets widgets, integrations, subagents, and services enter a conversation lane without pretending to be plain chat text. The plural transport kind is uniformly external_event; semantic meaning such as follow-up, steer, consent grant, or subagent completion remains nested in payload.event.type. Reactive ingress atomically stores the accepted batch and one bodyless wake. ReAct can fold eligible events into a live turn while it owns the listener; without a live owner they are handled by a later turn. A run-to-completion adapter consumes the batch available at turn start and does not inherit ReAct's mid-turn folding merely by using the shared harness timeline.

Event-source block production is total: every consumed lane event ends either folded or acknowledged as skipped. A zero-block event advances the processed cursor without adding a timeline block or invoking a generic post-block hook. react.block_production.no_timeline is only a visibility decision; required business durability belongs to the producing service or explicit source-owned processing.

When a ReAct action, result, or protocol rejection is emitted to the UI, it travels as a lifecycle envelope on the outbound communicator. This makes the Steps view a trace of proposed calls, validation, execution, results, and errors without making the browser stream the source of truth.

Files and Links Are Delivered Out of Band

During an authenticated chat turn, a result file is emitted as chat.files with an object reference. The browser resolves that reference at click time under the user's session; the model receives a delivery note, not a signed URL that it could re-type incorrectly. Conversation links use chat.citations, including placement="chat" when the link belongs in the visible answer. Turn-less MCP clients keep their transport contract: when no chat lane exists, the original URL-bearing result passes through unchanged.

Why ReAct Uses a Channel Protocol

KDCube ReAct does not depend on a model provider's native tool-calling envelope. It listens to runtime-owned semantic channels while the model streams. That lets one response interleave reasoning, code, actions, and other event blocks, including events the agent did not itself initiate.

Code is emitted as code, not escaped inside a JSON argument string. This matters for programs containing quotes, HTML, or nested JSON and reduces avoidable formatting failures. Because the runtime validates the channel protocol itself, models that were not trained for one provider's tool-calling format can still act as the ReAct decision model.

# Standard streaming lifecycle
await communicator.start()                   # chat_start event
await communicator.step("Searching…")       # chat_step — visible progress
await communicator.delta("answer", chunk)   # chat_delta — streaming text
await communicator.event(
    event_type="chat.compaction",
    data={"status": "completed"},
)                                               # chat_compaction — context compaction status
await communicator.complete()                # chat_complete

# Delta markers: answer · thinking · canvas · timeline_text · subsystem
await communicator.delta("thinking", reasoning_chunk)
await communicator.delta("canvas", json_payload)

# Custom service event — broadcast to all session subscribers
await communicator.event(
    event_type="chat.service",
    data={"key": "status.update", "value": "processing"},
)

# P2P — deliver only to a specific connected client (target_sid)
await communicator.event(
    event_type="chat.service",
    data={"key": "private", "value": result},
    target_sid=sid,                           # omit for broadcast
)

# Error
await communicator.error("Something went wrong")

Scoped Recording and Event Sinks

Recording reuses the same event vocabulary as the outbound filter but at a different boundary. The filter allows or suppresses browser and recorder disclosure; recording decides whether an allowed envelope is copied into a bounded in-memory buffer for later sink delivery.

async with communicator.recording(
    selector,
    scope={"owner": "workflow"},
    sink=event_sink,
    send_on_exit=True,
):
    await run_work()

Multiple scopes are additive. If a workflow scope and a tool scope match the same envelope, the recorded item carries both scopes. Platform child tool runtimes receive portable scopes through COMM_SPEC; child-added records return through comm_recorded_events.json and are sent by the host after merge.

Outbound Event Firewall (per app)

Apps can attach an IEventFilter to the workflow to suppress or gate outbound communicator envelopes before they reach Redis, clients, or recorders. The filter sees the caller's user_type, user_id, and full event metadata. Filters are fail-open: an exception allows the event through.

from kdcube_ai_app.apps.chat.sdk.comm import IEventFilter

class MyAppEventFilter(IEventFilter):
    def allow_event(self, *, user_type, user_id, event, data=None) -> bool:
        # hide internal step events from non-privileged users
        if user_type != "privileged":
            if event.type == "chat.step" and event.broadcast:
                return False
        return True

# Wire at entrypoint — passed to entrypoint factory via event_filter param
workflow = MyWorkflow(..., event_filter=MyAppEventFilter())

See comm-system.md and README-comm.md for full detail.

Streaming vs Request-Scoped Surfaces

The communication pipeline carries asynchronous app events. It does not mean every app interface is itself a chat stream. KDCube currently exposes several adjacent surfaces, each with a different contract.

An app operation or MCP request can still produce live UI updates when the app code explicitly emits communicator events. In that case the streaming delivery reuses the same SSE or Socket.IO session identified by auth state and optional KDC-Stream-ID.

App-served MCP transport contract

Route family, transport, and authority are separate choices. For MCP, responsibilities divide as follows:

• Proc: resolve the app route, apply any declared managed guard, preserve request context, and dispatch into the returned FastMCP or ASGI subapp.
• App: serve the MCP contract and, when auth is app-owned, verify the app-specific token, signature, header, or cookie.
• Managed guard: when configured, resolve the bearer handle to a server-side grant/session record, enforce the matching resource_grants and selected operation, and project the grantor identity before the handler runs.
• Client: call the explicit route and send the credential required by that declared contract.

An external MCP credential is not a browser login token. With Delegated Access, it represents a delegated actor operating through a grantor. The token is only a handle; product code must not decode it as the authority source.

Typical authenticated MCP flow:

1. the app declares either an app-owned or managed credential contract
2. secrets or server-side grant records stay in their owning lifecycle
3. the MCP client calls /api/integrations/bundles/{tenant}/{project}/{bundle_id}/mcp/{alias}
4. the declared guard authenticates the request and resolves runtime identity
5. proc dispatches the request to the MCP endpoint under that context

Typical public MCP flow:

1. the client calls /api/integrations/bundles/{tenant}/{project}/{bundle_id}/public/mcp/{alias}
2. proc forwards the request into the app MCP endpoint
3. the app either accepts it as public or still applies its own auth logic if that is the chosen contract

App-authenticated public API hooks

Public app APIs can use several explicit auth ownership shapes:

• Proc-owned: public_auth="none" or public_auth={"mode":"header_secret", ...}
• App-owned: public_auth="bundle"
• Managed delegated access: a Connection Hub REST guard resolves server-side grants and projects the runtime user before product code runs.

For public_auth="bundle", there are three separate responsibilities:

• Proc: resolve the route, parse the request, and invoke the app method
• App: accept request: Request, read the inbound headers/body, and decide whether the hook is valid
• Client / webhook provider: call the explicit public operations route and send the auth material the app contract requires

Typical app-authenticated public hook flow:

1. app props define a non-secret contract such as the header name
2. app secrets store the verification material such as a shared token
3. the caller sends POST /api/integrations/bundles/{tenant}/{project}/{bundle_id}/public/{alias}
4. proc forwards the request into the app method
5. the app verifies the request and either raises 401 / 403 or returns normally