stream

Overview ¶

Package stream — additions: the `agents.*` HTTP handler. Like `tools.*` and `memory.*`, the eight Agents-page methods are one-shot request/response — POST JSON in, JSON out — and the handler lives in the stream package because its identity + scope-claim gating reuses the same `resolveIdentity` / `auth.HasScope` machinery the subscription surface uses.

Route shape:

POST /v1/agents/{method}

where {method} is the canonical method's verb suffix:

list | get | tools | memory | governance | skills |
permissions | metrics

The handler reads identity from r.Context() (auth.Middleware) or the X-Harbor-* carrier headers (fallback), decodes the JSON body into the method-specific wire request, dispatches into the agentsprotocol.Service, and encodes the response. On failure, a JSON error body with the canonical Protocol Code, identical in shape to the REST control transport's error body.

All eight `agents.*` methods are READ-ONLY projections of the Agent Registry. The five agent-control verbs the Console exposes (Pause / Drain / Restart / Force-Stop / Deregister) are NOT served here — they are the EXISTING shipped `registry.*` control verbs and Harbor mints no control method (CLAUDE.md §13).

Identity is mandatory: a missing / incomplete (tenant, user, session) triple fails closed with CodeIdentityRequired (401). `agent_id` is a registration identity, not an isolation filter — the runtime scopes by the tuple, never by `agent_id` (CLAUDE.md §6).

Package stream — additions: the `auth.*` HTTP handler. Like `tools.*` and `tasks.*`, the single `auth.*` method (`auth.rotate_token`) is one-shot request/response — POST JSON in, JSON out — and the handler lives in the stream package because its identity + scope-claim gating reuses the same `resolveIdentity` / `auth.HasScope` machinery the subscription surface uses.

Route shape:

POST /v1/auth/{method}

where {method} is the canonical method's verb suffix:

rotate_token

`auth.rotate_token` is the ONE net-new Protocol method Harbor ships — the Console Settings page is otherwise a pure consumer of the 72f / 72g posture surfaces. It is an ADMIN method: it rotates the operator's current Protocol-auth token and requires the verified `auth.ScopeAdmin` claim (closed two-scope set — there is NO `auth.admin` scope). A request without the claim is rejected 403 with the canonical CodeIdentityScopeRequired Code. Every successful rotation emits a redacted `audit.admin_scope_used` event through the shipped audit.Redactor (the emit lives in auth.RotateSurface — the handler never bypasses the audit redactor; CLAUDE.md §7 rule 6, §13).

Package stream — additions: the Console Flows-page HTTP handler. Like `events.aggregate` and `pause.list`, these are one-shot request/response endpoints — POST JSON in, JSON out. They live in the stream package because their identity + cross-tenant scope-claim gating is the same one Subscribe / Aggregate / PauseList use, and they share the route prefix family the Console surface registers.

Route shapes (all POST):

POST /v1/flows/list            — paginated flow catalog
POST /v1/flows/describe        — a flow's engine-graph description
POST /v1/flows/runs/list       — a flow's paginated run history
POST /v1/flows/runs/describe   — a single run's per-node timeline
POST /v1/flows/run             — invoke a one-shot run (mutating)
POST /v1/flows/metrics         — a flow's sparkline metrics

Five routes are read-only; `flows/run` mutates and is gated on the verified `auth.ScopeAdmin` claim (closed two-scope set — NO new scope is minted). Every route is identity-mandatory.

The handler emits a per-page audit event on every dispatch: `flows.page_viewed` for the five reads, `flows.run_invoked` for the mutating run. The events flow through the canonical EventBus so the Console activity feed and the audit log see Flows-page traffic.

FlowsHandler is a concurrency-safe compiled artifact — surface / bus / logger are set once at construction; ServeHTTP holds no per-request state.

Package stream — additions: the `events.aggregate` HTTP handler. Unlike `events.subscribe` (long-lived SSE), this is a one-shot request/response — POST JSON in, JSON out. It lives in the stream package because it shares the same events- subsystem dependencies and route prefix (`/v1/events*`), and its scope-claim gating is the same one Subscribe uses.

Route shape:

POST /v1/events/aggregate

The handler reads identity from r.Context() (auth.Middleware) and gates cross-tenant filters on auth.HasScope(ScopeAdmin) OR auth.HasScope(ScopeConsoleFleet). The response body is the wire EventAggregateResponse JSON; on failure, a JSON error body with the canonical Protocol Code.

Package stream — additions: the three `memory.*` read handlers. Like `events.aggregate` and `pause.list`, these are one-shot request/response — POST JSON in, JSON out — and they live in the stream package because their identity + cross-tenant scope-claim gating is the same one Subscribe / Aggregate / PauseList use, and they share the route prefix family the Console subscription surface registers.

Route shapes:

POST /v1/memory/list
POST /v1/memory/get
POST /v1/memory/health

The handler reads identity from r.Context() (auth.Middleware) or the X-Harbor-* carrier headers (fallback), decodes the JSON body, gates cross-tenant filters on auth.HasScope(ScopeAdmin) OR auth.HasScope(ScopeConsoleFleet) — the admin closed two-scope set, NO new memory scope (audit B1) — projects the answer from the memory subsystem, and encodes the response. The response body is the wire types.Memory* JSON; on failure, a JSON error body with the canonical Protocol Code.

All three methods are READ-ONLY. The memory mutation surface (`memory.put` / `memory.delete`) is deferred to a later phase / post-V1 (page-memory.md §10); this handler ships no mutation path.

Package stream — additions: the `pause.list` HTTP handler. Like `events.aggregate`, this is a one-shot request/response — POST JSON in, JSON out — and it lives in the stream package because its identity + cross-tenant scope-claim gating is the same one Subscribe / Aggregate use, and it shares the route prefix family the Console subscription surface registers.

Route shape:

POST /v1/pause/list

The handler reads identity from r.Context() (auth.Middleware) or the X-Harbor-* carrier headers (fallback), decodes the JSON body into a types.PauseListRequest, gates cross-tenant filters on auth.HasScope(ScopeAdmin), projects the snapshot from the unified pause/resume Coordinator, applies the heavy-content bypass on each row, and encodes the response. The response body is the wire types.PauseListResponse JSON; on failure, a JSON error body with the canonical Protocol Code.

pause.list is READ-ONLY against the Coordinator — it does NOT call Resume, does NOT clear checkpoints. Resume actions continue through the `resume` / `approve` / `reject` control methods. The unified pause/resume primitive is never bypassed: pause.list reads the shipped Coordinator state, it does not reinvent pause coordination (CLAUDE.md §7 rule 4, §13).

Package stream — addition: the Console Playground-page `runs.set_overrides` HTTP handler. Like the Sessions / Tasks / Flows page handlers, this is a one-shot request/response endpoint — POST JSON in, JSON out. It lives in the stream package because its identity resolution + defence-in-depth body-identity check are the same `resolveIdentity` machinery the other Console-page handlers use.

Route shape (POST):

POST /v1/runs/set_overrides — record the next-message override

The route is identity-mandatory. The override's target session (`overrides.session_id`) MUST equal the caller's verified session; the runs/protocol.Service rejects a cross-session target with CodeScopeMismatch (403). `runs.set_overrides` is NOT an admin method — it records an override for the operator's OWN session, so no cross-tenant scope claim is consulted.

RunsHandler is a concurrency-safe compiled artifact — service / logger are set once at construction; ServeHTTP holds no per-request state.

Package stream — additions: the Console Sessions-page HTTP handler. Like `events.aggregate`, `pause.list`, and the Flows-page handler, these are one-shot request/response endpoints — POST JSON in, JSON out. They live in the stream package because their identity + cross-tenant scope-claim gating is the same `resolveIdentity` / `auth.HasScope` machinery the subscription surface uses.

Route shapes (both POST):

POST /v1/sessions/list     — paginated, filtered session catalog
POST /v1/sessions/inspect  — a single session's full snapshot

Both routes are read-only and identity-mandatory. A cross-tenant `sessions.list` filter (a `tenant_ids` entry outside the caller's verified tenant) is gated on the verified `auth.ScopeAdmin` claim (closed two-scope set — NO new scope is minted); the sessions/protocol.Service emits an `audit.admin_scope_used` event on every successful admin-scope query.

SessionsHandler is a concurrency-safe compiled artifact — service / logger are set once at construction; ServeHTTP holds no per-request state.

Package stream is the Harbor Protocol SSE event transport — the server→client half of the wire binding RFC §5.4 resolves to (SSE for events + REST/JSON for control). It is a thin adapter over the events.EventBus: a Handler opens a triple-scoped events.Subscription and frames each events.Event as an SSE block (frame.go). The SSE transport adds the wire framing and the connection lifecycle; it adds NO second event channel — the "one bus, no parallel observability channel" is load-bearing.

The route shape ¶

GET /v1/events

The identity triple is carried in request headers (X-Harbor-Tenant / X-Harbor-User / X-Harbor-Session) — a header carrier, not a query string, so the triple is not logged in access logs by default and the JWT validation slots in at the same choke point. An optional X-Harbor-Event-Type header (repeatable) narrows the subscription's event-type selector.

Identity at the edge (RFC §5.5, CLAUDE.md §6) ¶

The handler resolves the triple from the headers and rejects a request with any missing component closed — HTTP 401, before any subscription is opened. The SSE stream is ALWAYS triple-scoped: events.Filter.Admin (cross-tenant fan-in) is NOT exposed on the wire in a later phase — it needs the cryptographic scope claim Harbor adds. resolveIdentity is the single choke point slots JWT validation into.

Reconnect cursor ¶

SSE's native reconnect mechanism is the Last-Event-ID header: a client that drops echoes back the `id:` of the last frame it saw. The handler maps that onto an events.Cursor and, when the bus driver implements events.Replayer, replays the events strictly newer than the cursor before live-tailing — so a reconnecting client does not miss the gap. When the driver does not support replay (or replay is configured off), the handler live-tails from the reconnect point and emits a single explicit `stream.replay_unavailable` comment frame so the gap is SURFACED, never silently masked (CLAUDE.md §5).

Concurrent reuse ¶

Handler is a compiled artifact: the bus, the logger, the keepalive interval and the clock are set once at construction and never mutated. ServeHTTP holds no per-request state on the Handler — each request gets its own events.Subscription, its own keepalive ticker, and its own goroutine, all torn down before ServeHTTP returns. One Handler serves N concurrent streams safely; the goroutine for a stream is cancelled by that request's ctx, never by a shared handler-level ctx (CLAUDE.md §5 "Concurrency"). internal/protocol/transports/ concurrent_test.go pins it under -race.

Package stream — additions: the `tasks.*` HTTP handler. Like `tools.*` and `pause.list`, the two Tasks-page read methods are one-shot request/response — POST JSON in, JSON out — and the handler lives in the stream package because its identity + cross-tenant scope-claim gating reuses the same `resolveIdentity` / `auth.HasScope` machinery the subscription surface uses.

Route shape:

POST /v1/tasks/{method}

where {method} is the canonical method's verb suffix:

list | get

The handler reads identity from r.Context() (auth.Middleware) or the X-Harbor-* carrier headers (fallback), decodes the JSON body into the method-specific wire request, computes the verified `auth.ScopeAdmin` claim (consulted by `tasks.list` only for a cross-tenant fan-in), dispatches into the tasksprotocol.Service, and encodes the response. On failure, a JSON error body with the canonical Protocol Code, identical in shape to the REST control transport's error body.

The handler is READ-ONLY: both methods are pure reads. The Console Tasks page consumes the EXISTING task-control verbs (`cancel` / `pause` / `resume` / `prioritize` / `approve` / `reject`) for mutation through the control transport — there is NO `tasks.*` mutating method (CLAUDE.md §13 "no parallel implementations").

Package stream — additions: the `tools.*` HTTP handler. Like `pause.list` and `events.aggregate`, the seven Tools- page methods are one-shot request/response — POST JSON in, JSON out — and the handler lives in the stream package because its identity + cross-tenant scope-claim gating reuses the same `resolveIdentity` / `auth.HasScope` machinery the subscription surface uses.

Route shape:

POST /v1/tools/{method}

where {method} is the canonical method's verb suffix:

list | get | describe | metrics | content_stats |
set_approval_policy | revoke_oauth

The handler reads identity from r.Context() (auth.Middleware) or the X-Harbor-* carrier headers (fallback), decodes the JSON body into the method-specific wire request, gates the two admin methods on the verified `auth.ScopeAdmin` claim (there is NO `tools.admin` scope), dispatches into the toolsprotocol.Service, and encodes the response. On failure, a JSON error body with the canonical Protocol Code, identical in shape to the REST control transport's error body.

The handler is READ-MOSTLY: five of the seven methods are pure reads. The two admin methods mutate runtime tool state and emit an `audit.admin_scope_used` event through the shipped audit.Redactor (the emit lives in toolsprotocol.Service — the handler never bypasses the audit redactor; CLAUDE.md §7 rule 6, §13).