{"record":{"id":"61f8b005b566a016","repo":"astrid-runtime/astrid","slug":"gateway-is-not-wired-to-a-live-event-bus-agent-re","errorCode":null,"errorMessage":"gateway is not wired to a live event bus; agent request stream unavailable","messagePattern":"gateway is not wired to a live event bus; agent request stream unavailable","errorType":"http","errorClass":"GatewayError::Internal","httpStatus":500,"severity":"error","filePath":"crates/astrid-gateway/src/routes/agent.rs","lineNumber":399,"sourceCode":"/// cannot spoof a grant prompt into a user's stream.\n#[utoipa::path(\n    get,\n    path = \"/api/agent/requests\",\n    tag = \"agent\",\n    responses(\n        (status = 200, description = \"Server-Sent Events stream of pending `approval` and `elicit` requests scoped to the authenticated principal. Starts with `event: ready`.\", content_type = \"text/event-stream\"),\n        (status = 401, body = ErrorBody, description = \"Missing / invalid bearer.\"),\n        (status = 500, body = ErrorBody, description = \"Gateway not wired to a live event bus.\"),\n    )\n)]\npub async fn get_requests(\n    State(state): State<Arc<GatewayState>>,\n    req: Request<axum::body::Body>,\n) -> GatewayResult<Sse<impl Stream<Item = Result<Event, Infallible>>>> {\n    let caller = caller_from(&req)?.clone();\n\n    let Some(bus) = state.event_bus.clone() else {\n        return Err(GatewayError::Internal(anyhow::anyhow!(\n            \"gateway is not wired to a live event bus; agent request stream unavailable\"\n        )));\n    };\n\n    let conn_route_uuid = Uuid::new_v4();\n    let subscribe = |topic: &'static str| {\n        bus.subscribe_topic_routed(conn_route_uuid, topic, \"gateway\", \"gateway::agent_requests\")\n    };\n    let mut approval_rx = subscribe(\"astrid.v1.approval\");\n    let mut elicit_rx = subscribe(\"astrid.v1.elicit\");\n    let principal = caller.principal.to_string();\n\n    let stream = async_stream::stream! {\n        yield Ok::<Event, Infallible>(\n            Event::default()\n                .event(\"ready\")\n                .data(serde_json::json!({ \"principal\": principal }).to_string())\n        );","sourceCodeStart":381,"sourceCodeEnd":417,"githubUrl":"https://github.com/astrid-runtime/astrid/blob/affd8760f44190dbdfbec23403f4c4b642c33112/crates/astrid-gateway/src/routes/agent.rs#L381-L417","documentation":"The GET agent requests SSE handler (get_requests) subscribes to agent request events on the event bus and streams them to the client. This error is thrown when GatewayState.event_bus is None, so there is no bus to subscribe to and no stream can be produced.","triggerScenarios":"Opening the agent request SSE stream when the gateway was built without an event bus (event_bus = None in GatewayState).","commonSituations":"Gateway launched in a mode without the agent runtime/event bus; misconfiguration that skips bus construction at startup; wiring regression after a refactor of GatewayState initialization; clients pointed at a deployment that intentionally has no agent support.","solutions":["Inject the live event bus into GatewayState during gateway construction and restart","Enable the event bus subsystem in the deployment configuration","If agent streaming is not offered, remove or gate the SSE endpoint for such deployments","Verify the startup logs/health endpoint that the bus is connected before serving agent routes"],"exampleFix":"// before\nlet Some(bus) = state.event_bus.clone() else {\n    return Err(GatewayError::Internal(anyhow::anyhow!(\n        \"gateway is not wired to a live event bus; agent request stream unavailable\"\n    )));\n};\n// after: fail fast at startup, not per-request\nlet bus = config.event_bus.as_ref()\n    .ok_or_else(|| anyhow!(\"event bus required for agent routes\"))?;\nlet state = GatewayState { event_bus: Some(bus.clone()) };","handlingStrategy":"validation","validationCode":"// Verify bus availability before opening the SSE stream\npub fn agent_stream_available(state: &GatewayState) -> bool {\n    state.event_bus.is_some()\n}","typeGuard":"fn bus_of(state: &GatewayState) -> Option<Arc<EventBus>> {\n    state.event_bus.clone()\n}","tryCatchPattern":"match get_requests(state, req).await {\n    Err(GatewayError::Internal(e)) if e.to_string().contains(\"request stream unavailable\") => {\n        StatusCode::SERVICE_UNAVAILABLE\n    }\n    other => other.map(Into::into),\n}","preventionTips":["Wire the bus before mounting the requests SSE route; gate the route on bus presence","Surface bus status in /healthz so operators notice before clients connect","In bus-less deployments (tests/local), stub the SSE route instead of leaving it live","Add an integration test asserting the requests stream works end-to-end when the bus is configured"],"tags":["event-bus","misconfiguration","agent","sse"],"backgroundTag":"missing-dependency","analyzedSha":"affd8760f44190dbdfbec23403f4c4b642c33112","analyzedAt":"2026-09-09T21:28:12.402Z","contentChangedAt":"2026-09-09T21:28:12.402Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}