Stream protocol v1¶

Status: implemented. The frames and commands are in reflexr.core.protocol, and their JSON Schema is generated into schemas/reflexr.v1.json; one handler, Workspaces.execute, carries out every command, once per command_id. REST and the WebSocket are served by reflexr.fastapi, and MCP by reflexr.mcp (phase 5 of RFC-0001). The structure is settled by ADR-0016, ADR-0005, ADR-0044 and ADR-0047. Its shape deliberately matches artifactr's thread protocol, so one client library can speak both.

Clients publish events into a workspace, read its log with resume, and operate rules and runs. The same commands are available over REST, over a WebSocket, and as MCP tools, and every surface hands them to the same handler, so they behave identically.

Every event type and rule name is qualified by its owner's namespace: ops:service.error, reflexr:rule_fired, ops:error-spike (ADR-0039). A type name is a namespace, a : and a local name, matching ^[a-z][a-z0-9_]*:[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)*$. A name without a namespace, in a publish or a type filter, is validation_failed, and its message names the qualified type with that local name: event type 'rule_fired' has no namespace; did you mean 'reflexr:rule_fired'?.

Commands¶

A command is a JSON object with a type. Over REST and WebSocket it travels in a frame that carries a client-chosen command_id, which an MCP tool takes as an argument; a retried command_id returns the original result without running again (Deduplication).

{"type": "command", "command_id": "cmd_7", "command": {"type": "publish", "event": {"type": "ops:service.error", "service": "auth", "severity": 8, "message": "token check failed"}}}

Run commands answer with run: the run as it now is; replay_rule answers with rule: {rule, progress}; and the stored-rule commands with rule_version: {rule, version, seq, duplicate}, where seq is the change's fact. Each outcome has a type naming which it is.

Stored rules are off unless the application configures them, and then its allow hook is asked about each change after authorize: without either, the change is forbidden, as it is for a code rule's name. provenance is an opaque JSON object of at most 4 KiB, kept on the rule and its fact. A rule that fails Rule.check or check_stored, or provenance over 4 KiB, is validation_failed, with every problem in errors. Updating or archiving a name no stored rule has is not_found. A change that would leave the rule as it is appends nothing and answers duplicate: true, checked before expected_version, so a retried change is no conflict; its seq is the head of the log.

Rejections carry a stable type and a message: not_found, invalid_state, validation_failed (with Pydantic's errors), forbidden, depth_exceeded (a publish beyond the causation limit), and unsupported_protocol.

Deduplication¶

Every surface hands a command with its id to Workspaces.execute, which carries it out the first time the id is seen and remembers the result in the workspaces' CommandResults. A command is known by the tenant and workspace it was sent to, the sender (its actor's participant, so a changed display name does not matter) and its command_id; a repeated id returns the first result, outcome or rejection, whatever command it comes with. By default the workspaces remember the 10,000 most recent results in their process (InMemoryCommandResults), so a retry that reaches another process, or arrives after the result was forgotten, runs again, and so does one that arrives while the first is still being carried out.

Beyond that memory, the log deduplicates what carries an id, whichever process a retry reaches: a publish with an event id appends once, and a stored-rule change that would leave the rule as it is answers duplicate: true.

REST¶

reflexr.fastapi serves REST under a prefix the application chooses (/v1 below).

Rejections map to HTTP status codes: not_found → 404, invalid_state → 409, validation_failed and depth_exceeded → 422, forbidden → 403, unsupported_protocol → 400. A body that does not validate, such as an event whose fields do not match its type, is 422. Authentication is the host's: resolve_actor(request) returns the tenant and actor, or raises Unauthorized (401); an optional authorize(tenant, workspace, actor) refuses a workspace (403, with the forbidden rejection as the body's detail, as every rejection has).

WebSocket¶

GET /v1/workspaces/{workspace_id}/stream, subprotocol reflexr.v1.

Handshake and resume¶

sequenceDiagram
    participant C as Client
    participant S as Server
    C->>S: hello {protocol, resume_after_seq or from_head, types?}
    S-->>C: welcome {workspace_id, head_seq, reset}
    S-->>C: event ... (replay after resume_after_seq)
    S-->>C: replay_complete {up_to_seq}
    Note over C,S: Then: event frames as they are appended, and command results

{"type": "hello", "protocol": "reflexr.v1", "resume_after_seq": 41, "types": ["ops:service.error", "reflexr:rule_fired", "reflexr:run_succeeded"]}

• resume_after_seq is the last seq the client has; the server replays everything after it. A client ahead of the log gets reset: true and a replay from the beginning.
• from_head: true starts at the head of the log instead: nothing is replayed, replay_complete (up_to_seq is head_seq) follows welcome at once, and reset is false. The client's position is then welcome.head_seq, and it reconnects with resume_after_seq from there. A hello with from_head and a nonzero resume_after_seq closes with 4400.
• types filters which events are sent, by qualified name; null sends all. A name without a namespace closes the connection with 4400, as any bad hello does. replay_complete is decided on the unfiltered log, so it always arrives.

Server frames¶

• event: an envelope, {"type": "event", "seq", "id", "ts", "workspace_id", "actor", "causation", "correlation_id", "traceparent", "event": {...}}. traceparent is the W3C trace context of the span that published the event, or null.
• command_result: {"type": "command_result", "command_id", "ok", "outcome" | "rejection"}.
• replay_complete: {"type": "replay_complete", "up_to_seq"}.
• error: {"type": "error", "message"}, for a frame that is not a valid command. The connection stays open.

reflexr's own events, alongside the application's:

actor.kind is user, agent (rule, run_id, name), external_agent (client_id, name?), system (name), source (name) or evaluator (name, version). causation is {firing_id, run_id, depth} for an event a run emitted and for reflexr's facts about a firing or run (one step deeper than what fired), and null otherwise.

Close codes¶

MCP¶

reflexr.mcp serves the same commands and reads to MCP clients, with the host's resolve(ctx) returning the client's tenant and ExternalAgentActor. An optional authorize(tenant, workspace, actor), the same hook as REST's, is asked on every tool call, resource read and resource subscription that names a workspace, and a refusal is a tool error, or a failed read or subscription, carrying the forbidden rejection's message. Commands go through the same handler as REST and the WebSocket; a rejection is a tool error carrying its message, followed, for a validation_failed that lists problems, by its errors as JSON.

Every tool that changes something (publish_event, give_feedback, retry_run, skip_run, cancel_run, replay_rule, install_rule, update_rule and archive_rule) takes an optional command_id, the command frame's idempotency key: a retry with the same id returns the first result (Deduplication). A call without one is carried out every time, and an empty one is refused, as in a command frame. The MCP request id cannot serve, since a retry is a new request.

Versioning and schema¶

• The protocol identifier is reflexr.v1, also the WebSocket subprotocol. Within v1, changes are additive: new optional fields and new event types. Clients ignore unknown fields and keep unknown event types as opaque envelopes, since they still carry a seq.
• Qualified names were the one change to v1 that was not additive, made before its first release: every event type and rule name gained its namespace, and reflexr's own facts moved to reflexr: (ADR-0039).
• JSON Schemas for every frame, and for rules, are generated from the Pydantic models into schemas/ and checked in. CI fails if they drift.