Pub-Sub & Semantic Subscriptions

Polling state endpoints does not scale for reactive systems. MuBit provides event streams so workers can wake on state changes.

Decision model

Stream type	Method	Best use
Run-scoped control events	`control.subscribe`	Workflow coordination inside one `run_id`
Semantic event subscriptions	`core.subscribe_events`	Topic-triggered automation across broader changes

Minimal implementation example

Python
Node / TS
Rust

05-event-driven-worker.py

stream = client.control.subscribe({"run_id": run_id, "event_types": ["SetVariable", "SubmitAction"]})

client.control.set_variable({
    "run_id": run_id,
    "name": "ticket_priority",
    "value_json": '"critical"',
    "source": "retrieval",
})

semantic_sub = client.core.subscribe_events({
    "filter_type": "semantic",
    "query_text": "billing escalation",
    "threshold": 0.82,
})

print(semantic_sub)
for event in stream:
    if event:
        print(event)
        break

pubsub_and_events.ts

const stream = await client.control.subscribe({
  run_id: runId,
  event_types: ["SetVariable", "SubmitAction"],
});

await client.control.setVariable({
  run_id: runId,
  name: "ticket_priority",
  value_json: '"critical"',
  source: "retrieval",
});

const semanticSub = await client.core.subscribeEvents({
  filter_type: "semantic",
  query_text: "billing escalation",
  threshold: 0.82,
});

console.log(semanticSub);
const reader = stream.body?.getReader();
const first = reader ? await reader.read() : null;
if (first?.value) {
  console.log(new TextDecoder().decode(first.value));
}

pubsub_and_events.rs

use serde_json::json;

let stream = client.control.subscribe(json!({
    "run_id": run_id,
    "event_types": ["SetVariable", "SubmitAction"]
})).await?;

client.control.set_variable(json!({
    "run_id": run_id,
    "name": "ticket_priority",
    "value_json": "\"critical\"",
    "source": "retrieval"
})).await?;

let semantic_sub = client.core.subscribe_events(json!({
    "filter_type": "semantic",
    "query_text": "billing escalation",
    "threshold": 0.82
})).await?;

println!("{:?}", semantic_sub);
println!("{:?}", stream);

Failure modes and troubleshooting

Symptom	Root cause	Fix
Too many events	Broad filters	Narrow event types and semantic threshold
Missing events	Subscriber on wrong run scope	Verify producer and subscriber `run_id`
Core subscribe denied	Direct-core policy boundary	Use control stream default or enable policy explicitly

Next steps

Add planner/specialist coordination at Sessions and branching.
Review state APIs at State management endpoints.

Overview

Learn the basics

Build with MuBit

Go deeper

Platform

Help

Pub-Sub & Semantic Subscriptions

Decision model

Minimal implementation example

Failure modes and troubleshooting

Next steps

Overview

Learn the basics

Build with MuBit

Go deeper

Platform

Help

​Decision model

​Minimal implementation example

​Failure modes and troubleshooting

​Next steps

Decision model

Minimal implementation example

Failure modes and troubleshooting

Next steps