feat(agents): Atlas broker — POST /agents/atlas SSE stream [Story-1209]

Story 1209Epic None5 files · +1140/−0

PR #305 · pramaan-io/pramaan-functions · merged 2026-06-11

Why this matters

One double-click. Two Bedrock bills. A frozen event loop.

What breaks without a broker between counsel and Atlas

Trust the client once — bill twice, bleed tenants, freeze the loop.

The fix — Fingerprint the request, trust the session, pump the stream — Atlas runs once per intent, pays once, leaks never.

How it flows

sequenceDiagram
  autonumber
  participant client as Counsel (browser)
  participant broker as Atlas broker (FastAPI)
  participant db as Postgres + RLS
  participant bedrock as Bedrock AgentCore
  client->>broker: POST /agents/atlas (form_id, input)
  broker->>db: RLS session → firm_id
  broker->>db: INSERT agent_run_key — ON CONFLICT DO NOTHING
  broker->>broker: stamp firm_id on Atlas payload (server-trusted)
  broker->>bedrock: invoke_agent_runtime (thread executor)
  bedrock-->>broker: StreamingBody chunks → asyncio.Queue
  broker->>client: SSE: AtlasTurn events + X-Atlas-Run-Id

src/app_api/router_agents.py

Method	in → out	what it does
`_payload_hash`	`(input_data) → str`	fingerprints input as form_id + sha256[:16] so the same request always lands on the same run key
`_reserve_run_id`	`(db, firm_id, idempotency_key) → tuple[UUID, bool]`	INSERTs the run key with ON CONFLICT DO NOTHING and returns (run_id, created?) — one Bedrock call per unique request
`_firm_id_from_session`	`(db) → UUID`	reads firm_id from the RLS session, never from the request body — kills tenant spoofing in one line
`_pump`	`() → None`	runs in a thread, drains Bedrock's sync StreamingBody into an asyncio.Queue so the event loop never blocks

What changed, file by file

docs/walkthroughs/story-1209-walkthrough.html+703

Walkthrough HTML for Story nine-eighty-eight — documents the Atlas broker, explicitly referencing the payload hash, the SSE pump, and the AtlasReviewRequest model that land in the router.

new class	signature
`_payload_hash`	`(input_data) → str`
`_pump`	`() → None`
`AtlasReviewRequest`	`(class)`

New fields: queue : asyncio.Queue[bytes | None]

src/app_api/router_agents.py+294

Adds the POST /agents/atlas broker — payload fingerprint, RLS-session firm_id, ON CONFLICT DO NOTHING run reservation, and a thread-pumped SSE streamer that drains Bedrock's sync StreamingBody.

new class	signature
`AtlasReviewRequest`	`(class)`
`_payload_hash`	`(input_data) → str`
`_firm_id_from_session`	`(db) → UUID`
`_reserve_run_id`	`(db, firm_id, idempotency_key) → tuple[UUID, bool]`
`_agentcore_client`	`() → Any`
`_stream_atlas_sse`	`(runtime_arn, runtime_session_id, atlas_payload) → AsyncIterator[str]`
`_pump`	`() → None`
`atlas_intake_review`	`(req, db, current_user, _) → StreamingResponse`

New fields: form_id : strcounsel_message : str | Nonemode : strqueue : asyncio.Queue[bytes | None]

Files changed (5)

File	±
`docs/walkthroughs/story-1209-walkthrough.html`	+703
`openapi/app-api.json` ★	+136
`src/app_api/app.py`	+2
`src/app_api/router_agents.py` ★	+294
`src/shared/audit/coverage.py` ★	+5