Invoke pipeline

RiverBeam in beam_core owns one pipeline. Every entry point, whether the invoke_agent worker task, the invoke, stream and resume commands, or a direct invoke call, ends up in it.

Entry points

Method	Returns	Use
invoke(req)	serde_json::Value (an AgentResult as JSON)	Worker tasks and JSON APIs
invoke_result(req)	AgentResult	Rust callers

Both take an InvokeRequest. Each call gets a new run_id. When the request has no interaction_id, Riverbeam creates one.

Phases

1. Bootstrap

• Logs a request_received audit event.
• When a usage guard is configured, checks the user's plan. A denial under the HardBlock policy stops here with BEM-060 (HTTP 403).
• Writes the session (if new), the interaction, and the run through the configured BeamStore.

2. Security before input

Runs the security manager at stage BeforeInput over the text of the request. A blocking verdict fails the call with BEM-061 (HTTP 403). See Security.

3. Save the input turn

A normal turn saves your message to session memory and as a human message in the store. A resume turn (the request carries actions) resolves each pending action instead and saves an action_result message. See Human approval.

4. Build the context

1. Looks up the agent's metadata in the registry. An unknown agent fails with BEM-020.
2. Resolves a model with the model router.
3. Loads memory for the request's scopes, trimmed to the model's max_tokens. By default the scope is the session. Pass metadata.memory_scopes to load more layers.
4. When knowledge is configured, retrieves the top 4 chunks for the request text and prepends them as a system message (see Knowledge).

The result is an AgentContext with the conversation history and the resolved ModelConfig.

5. Dispatch the agent

Calls the agent's stream method, which returns a list of StreamChunk frames ending in a done frame. When the request has stream: true and a request_id, and a stream is attached, every frame is published to ws.channel.{session_id}:{request_id}. The AgentResult comes from the final done frame.

6. Security before output

Runs the security manager at stage BeforeOutput over the result text. A blocking verdict fails the call with BEM-061.

7. Persist the result

Saves the agent reply to session memory and as an ai message. When the result is interrupted and lists action_required entries, saves one pending action per entry.

8. Finalize

• Closes the run as completed, or interrupted when the agent paused for approval.
• Closes the interaction as completed, unless the run was interrupted.
• Updates the session's last activity.
• Records usage (one request, plus token_usage.total_tokens) for the user.
• Logs agent_completed or agent_interrupted.

Failures

When security blocks a request or the agent fails, Riverbeam marks the run failed and the interaction failed, logs agent_failed, and returns the error.

Errors in the other phases are returned without that cleanup: an invalid resume, an unknown agent (BEM-020), no matching model (BEM-041), or a store, memory or knowledge error. The run then stays running, the interaction stays in_progress, and no agent_failed event is logged.

Errors are RiverbaseError values with BEM-* codes; see Error codes.

With the full migration applied, the session table has a foreign key to agent. A request for an agent name that was never registered then fails while writing the session, with BEM-116 (500), before the registry can answer BEM-020 (404).

Configuring the pipeline

RiverBeam::new(BeamConfig::default()) gives a working pipeline with in-memory defaults. Replace the parts you need:

Part	Default	Replace with
Store	InMemoryStore	with_store / set_store (for example PgBeamStore)
Stream	none (frames are not published)	with_stream / set_stream
Knowledge	none (no retrieval)	with_knowledge / set_knowledge
Memory	InMemoryBackend with TokenLimitStrategy	set_memory
Security	DisabledSecurityManager (allows everything)	with_security / set_security
Usage	none (no quota checks)	set_usage(guard, tracker, plan)
Audit	AuditLogger with no sinks	set_audit
Agents	none	register_agent, register_echo_agent

BeamConfig fields and defaults:

Field	Default
worker_namespace	"riverbeam-worker"
sql_schema	"ref_beam"
knowledge_enabled	true
knowledge_embedding_dim	1536
knowledge_max_context_tokens	2000
default_model	None
security_enabled	true
usage_enabled	true

Current limitations

• The model list is fixed. RiverBeam::new installs one model, gpt-4o-mini (provider openai, text only, 4096 tokens), and there is no method to replace the router. model_override and default_model can only name that model; an unknown model_override is ignored. See Model routing.
• Of the BeamConfig fields, only default_model changes behaviour. BeamWorker copies worker_namespace into queue_name(), but NATS subjects use the task prefix instead. The other fields are not read yet: attaching a component enables it, and leaving it out disables it. PgBeamStore always uses the schema ref_beam.
• The current message is saved to session memory before the history is loaded, and RigAgentRunner appends it again, so the model sees the latest user message twice.
• Agents return all their frames at once, so streamed frames are published after the agent finishes, not token by token.