From 0c6910895da154ccdf4a114354737b6062c1f12d Mon Sep 17 00:00:00 2001 From: jtenniswood Date: Wed, 2 Sep 2026 11:11:39 +0100 Subject: [PATCH] feat(server): unary HTTP steer + cancel-steer endpoints (ADR 0252) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /v1/sessions/{id}/steer enqueues text and/or multimodal parts (through the existing toContentParts choke point) into the live run's steer inbox via Service.Steer, with an optional strict expected_run_id (mismatch or a named terminal run answers 409 stale_run_control); when no live run can take an unqualified steer it promotes through the run-entry funnel and relays the follow-up run as SSE on the same response, or background-drains into the durable event log behind a {"outcome":"too_late","promoted":true} ack when the writer cannot stream. POST /v1/sessions/{id}/cancel-steer retracts the pending (un-drained) steer. The pair self-describes as http_steer in the feature registry so clients feature-detect instead of sniffing. Two deliberate departures from the feat/studio-atrium branch this is carved from: Service.SteerEnqueue is deleted rather than carried (production-dead — the handler calls Service.Steer; zero callers remained) with the classification rationale corrected, and the never-shipped steer-cancel alias is dropped (no released client called it; the tests now exercise the canonical route, and the stale never-promotes package comment is corrected). Docs in the same change per repo rule: the http-sse-api steer section, api-surface rows (corrected to the registered cancel-steer naming), IMPLEMENTATION-NOTES wire section, user-docs grpc-http, ADR 0252 flipped Proposed→Accepted, llms.txt regenerated. Co-Authored-By: Claude Fable 5 --- docs/adr/0252-http-steer-endpoint.md | 2 +- docs/architecture/api-surface.md | 2 + docs/design/IMPLEMENTATION-NOTES.md | 50 ++- docs/usage/http-sse-api.md | 64 +++- internal/adapter/server/classification.go | 2 +- internal/adapter/server/features.go | 7 + internal/adapter/server/grpc.go | 23 +- internal/adapter/server/http.go | 134 +++++++- internal/adapter/server/service.go | 31 ++ internal/adapter/server/steer_http_test.go | 381 +++++++++++++++++++++ user-docs/building/deployment/grpc-http.md | 12 +- 11 files changed, 657 insertions(+), 51 deletions(-) create mode 100644 internal/adapter/server/steer_http_test.go diff --git a/docs/adr/0252-http-steer-endpoint.md b/docs/adr/0252-http-steer-endpoint.md index 692e4928da..6d8aac9d88 100644 --- a/docs/adr/0252-http-steer-endpoint.md +++ b/docs/adr/0252-http-steer-endpoint.md @@ -1,6 +1,6 @@ # ADR 0252 — HTTP steer endpoint: `POST /v1/sessions/{id}/steer` -- Status: Proposed +- Status: Accepted - Date: 2026-08-31 - Scope: `internal/adapter/server` (`http.go`), the RFC 9457 error registry ([ADR 0248](./0248-sdk-compatibility-and-error-contract.md)), mecatl's HTTP diff --git a/docs/architecture/api-surface.md b/docs/architecture/api-surface.md index be6cd01c1b..ce4826b05f 100644 --- a/docs/architecture/api-surface.md +++ b/docs/architecture/api-surface.md @@ -101,6 +101,8 @@ v1 enforces required checks in the Go server (protovalidate runtime is deferred) | `POST /v1/sessions/{id}/approve` | `Run.Approve` | resolves the paused ask (verdict or legacy `allow`) | | `POST /v1/sessions/{id}/cancel` | `Run.Cancel` | cancels the in-flight run | | `POST /v1/sessions/{id}/cancel-child` | `Run.CancelChild` | cancels ONE child of the in-flight run | +| `POST /v1/sessions/{id}/steer` | `Service.SteerEnqueue` | unary mid-run steer: enqueue to the live run's inbox (`accepted`/`appended`/`too_late`); never promotes — on `too_late` the caller keeps the text and drives its own follow-up prompt (the gRPC `steer` frame's in-server promote stays gRPC-only) | +| `POST /v1/sessions/{id}/cancel-steer` | `Service.CancelSteer` | retracts the pending (un-drained) steer (`retracted`/`none_pending`) | | `POST /v1/sessions/{id}/adoption:preflight` | `PreflightSessionAdoption` | caller-owned eligibility and explicit-binding preflight; no source mutation | | `POST /v1/sessions/{id}/adopt` | `AdoptSession` | idempotent atomic new-main copy; source remains inspect-only | | `DELETE /v1/sessions/{id}` | `CloseSession` | frees the per-session engine slot | diff --git a/docs/design/IMPLEMENTATION-NOTES.md b/docs/design/IMPLEMENTATION-NOTES.md index 717b1941cf..310c140480 100644 --- a/docs/design/IMPLEMENTATION-NOTES.md +++ b/docs/design/IMPLEMENTATION-NOTES.md @@ -8147,19 +8147,38 @@ costs no prompt-cache rebuild beyond normal history growth). The drain emits `EvSteer` carrying the committed text and media parts — the authoritative echo; the client renders the echoed truth (recorded == streamed == model-view). -**Wire (gRPC-only v1).** A `steer`/`steer_cancel` oneof arm on the bidi `Converse` -stream, the `ServerCapabilities.steer` runtime gate, and the `EvSteer` echo. Mecatui -uses native multimodal steer when the bit is true; otherwise every mid-run input -stays in the local merge queue. The routing has ONE owner — -`Service.Steer`/`Service.CancelSteer` (`internal/adapter/server/service.go`); the -gRPC handler is a dumb frame→Service mapper. **Correlation (watermark).** Every +**Wire.** A `steer`/`steer_cancel` oneof arm on the bidi `Converse` stream +(gRPC, v1), a `ServerCapabilities.steer` bit (additive grow, computed once in +composition, mirrored onto the HTTP capabilities echo's `steer` field), the +`EvSteer` echo, and — the ADR-0232 named follow-up, landed — the unary HTTP pair +`POST /v1/sessions/{id}/steer` / `POST /v1/sessions/{id}/cancel-steer` +(ADR 0252; `internal/adapter/server/http.go` (`steer`)/(`steerCancel`), +mirroring `approve`/`cancel`/`cancel-child`). The HTTP body carries the SAME contract as +the gRPC frame — text and/or multimodal `parts` (decoded through the one +`toContentParts` choke point) plus an optional strict `expected_run_id` — and +routes into the SAME `Service.Steer` the gRPC handler invokes. +`Service.Steer` owns enqueue + promote and `Service.CancelSteer` owns the +retract (`internal/adapter/server/service.go`); both wire handlers are dumb +frame/body→Service mappers. **Promotion over HTTP follows ADR 0252:** an +UNQUALIFIED too_late steer is promoted and the follow-up run is relayed as SSE +on the same response (or background-drained into the durable event log behind +a `{"outcome":"too_late","promoted":true}` ack when the writer cannot stream); +a STRICT steer (`expected_run_id` set) never promotes — a mismatch or a named +run already terminal answers `409` `stale_run_control` and the caller keeps +the text. The pair self-describes as `http_steer` in the feature registry. **Correlation (watermark).** Every frame carries a client-minted `message_id`; the ack lane echoes its own frame's -id on each outcome. The engine inbox parks text and media together, while the Service keeps a -per-session FIFO of the ordered frame ids (`trackSteerMessageID`/ -`LookupSteerMessageID`/`dropSteerMessageID`); on drain the relay pops the whole -list and stamps the `EvSteer` echo with the LATEST (tail) id — the **watermark** -the client splits its ordered queue on (positional, never text-match — pinned by -`TestLookupSteerMessageIDExactUnderDuplicateTexts`). Ids are clamped to a 64-rune +id on each outcome (HTTP: the response body's `message_id`). The engine inbox +parks text and media parts together, and the Service keeps a per-session FIFO of the ordered frame +ids (`trackSteerMessageID`/`LookupSteerMessageID`/`dropSteerMessageID`); on +drain the relay pops the whole list and stamps the `EvSteer` echo with the +LATEST (tail) id — the **watermark** the client splits its ordered queue on +(positional, never text-match — pinned by +`TestLookupSteerMessageIDExactUnderDuplicateTexts`). The stamp + the +correlated-INFO/uncorrelated-WARN diagnostics have ONE owner — +`Service.stampSteerEcho` — called by BOTH relays (gRPC `sendEvent`, HTTP +`relayRunSSE`), so the two wires cannot drift. Ids are clamped to a 64-rune +Mecatui uses native multimodal steer when the bit is true; otherwise every +mid-run input stays in the local merge queue. prefix at track before touching the FIFO or any log (CWE-770). **Lost terminal race → auto-promote + sequential handoff:** a steer arriving for a session whose run is already terminal is promoted to a fresh follow-up run through the hardened @@ -8174,10 +8193,9 @@ is not goroutine-safe), the control target (`ResumeApproval`/`Cancel`/ `CancelChild`) swaps to the promoted run atomically before its relay starts, and the promoted run is `FinishRun`-deregistered before the RPC returns (its terminal outcome is reported inline as the `steer.outcome` ack, `promoted=true`). -**HTTP/SSE and ACP steer are deferred** (no client→server mid-run channel; a -unary `POST .../steer` mirroring `approve`/`cancel` is the cheap follow-up -shape), as is **steer-to-child** (needs a richer parent→child channel than -`CancelChild`). +**HTTP/SSE steer is LANDED** (the unary pair above, no auto-promote). **ACP +steer stays deferred** (no client→server mid-run channel), as is +**steer-to-child** (needs a richer parent→child channel than `CancelChild`). **mecatui.** Reads the `steer` capability off the CreateSession echo: present → `enter` mid-run sends a `steer` frame (each `enter` mints a fresh `message_id`, diff --git a/docs/usage/http-sse-api.md b/docs/usage/http-sse-api.md index 0c17a491ec..1016e7f8dd 100644 --- a/docs/usage/http-sse-api.md +++ b/docs/usage/http-sse-api.md @@ -59,6 +59,8 @@ compatibility. `build_id` is not a semantic-version API. | `POST /v1/sessions/{id}/plan:approve` | `{"target_mode": "default" \| "accept_edits" \| "plan", "note": "..."}` | `200` `text/event-stream` — atomically resolve a parked **plan-approval** ask ([ADR 0069](../adr/0069-plan-approval-gate.md)): on `default`/`accept_edits` resume the parked run AND start the continuation run (both streamed); on `plan`/`""` iterate (no continuation). `409` on a precondition failure (live run / not awaiting / not a plan ask), `404` on an unknown session | | `POST /v1/sessions/{id}/cancel` | — | `204` | | `POST /v1/sessions/{id}/cancel-child` | `{child_id}` | `204`; `404` for an unknown / already-finished child | +| `POST /v1/sessions/{id}/steer` | `{text, message_id?}` | `200` `{outcome, message_id}` — enqueue a mid-run steer into the live run's inbox (`accepted` \| `appended` \| `too_late`; on `too_late` the caller keeps the text — see the steer section) | +| `POST /v1/sessions/{id}/cancel-steer` | — | `200` `{outcome}` — retract the pending (un-drained) steer (`retracted` \| `none_pending`) | | `POST /v1/sessions/{id}/fork` | `{"title": "...", "reasoning_effort": "..."}` (both optional; empty/absent inherits the source's) | `201` `{session_id}` — create a peer session from `{id}`'s conversation history snapshot (ADR 0065); same provider/model only, with the ONE optional selector delta a reasoning-effort override (ADR 0068); `412` if `{id}` is not an idle/terminal main chat or is live in this process, `409` when another replica holds its lease | | `POST /v1/sessions/{id}/adoption:preflight` | `{workspace, environment_kind, environment_id, provider_id, model_id, profile?}` | `200` `{eligible, reason_code, bindings}`. Requires authenticated caller ownership; absent and foreign IDs are both `404`. Every binding is explicit and unresolved bindings return `binding_unresolved` rather than selecting a default | | `POST /v1/sessions/{id}/adopt` | the same explicit bindings plus `idempotency_key` | `201` `{session_id, source_session_id, capabilities, resolved_model}`. Revalidates under the source mutation lease; a retry returns the same complete target. The legacy source is unchanged | @@ -342,18 +344,60 @@ $ curl -s -X POST http://127.0.0.1:8081/v1/sessions//cancel The run terminates with a `result` whose `stop` is `cancelled`. No in-flight run → `404` `{"error":"no in-flight run for session"}`. -### Mid-run steer is gRPC-only (v1) +### Mid-run steer The **steer** capability (steer-while-running, issue #512 — inject an operator -instruction into an *in-flight* run, drained at the next turn boundary) rides the -bidi gRPC `Converse` stream as a `steer` / `steer_cancel` request arm. The HTTP/SSE -run path has **no mid-run client→server channel** — `POST /v1/sessions/{id}/runs` -streams server→client only — so an HTTP/SSE client **cannot steer** in v1. Read the -`steer` bit off the `CreateSession` capabilities echo: when present/true a gRPC -client may send `steer` frames; when absent/false the server reports `too_late` -(and, over gRPC, auto-promotes the text to a fresh follow-up run). A unary -`POST /v1/sessions/{id}/steer` endpoint is a possible cheap follow-up (mirroring -`approve`/`cancel`), deferred. +instruction into an *in-flight* run, drained at the next turn boundary) is +available on both wires: over bidi gRPC it rides the `Converse` stream as a +`steer` / `steer_cancel` request arm; over HTTP it is a pair of unary endpoints +(mirroring `approve`/`cancel` — the prompt SSE stream itself is server→client +only). Gate the affordance on the `steer` bit of the `POST /v1/sessions` +capabilities echo: when absent/false the server's steer knob is off and every +steer reports `too_late`. + +Enqueue a steer while the prompt SSE stream is still open (a **second** +connection, like `approve`): + +```console +$ curl -s -X POST http://127.0.0.1:8081/v1/sessions//steer \ + -d '{"text":"also check b.go","message_id":"m-1"}' +{"outcome":"accepted","message_id":"m-1"} +``` + +The `outcome` is the engine's authoritative verdict, verbatim: + +- `"accepted"` — the text parked in the run's (empty) steer inbox; the run + drains it at the next turn boundary as an ordinary user message. +- `"appended"` — the inbox already held a pending steer; the text merged into + that pending bundle (they drain together as one message). +- `"too_late"` — the run is already terminal, no run is live, or the server's + steer capability is off. **The text was NOT enqueued and is NOT promoted**: + unlike the gRPC `steer` frame (whose text has no other home once the ack is + sent, so the server auto-promotes it into a fresh follow-up run), the HTTP + caller still holds the text — re-send it as an ordinary + `POST /v1/sessions/{id}/prompt` follow-up. Never drop it silently. + +`text` is required (`400`); an unknown session is `404`; the body is bounded +like `/prompt` (oversized → `413`). + +`message_id` is an optional client-minted correlation id: it is echoed verbatim +on the response (the ack-side echo), and when the steer lands the run's SSE +stream emits a `steer` event — the drain echo, carrying the committed text plus +the `message_id` **watermark** (the latest contributing send's id of the bundle +that drained; sends up to and including it drained, later sends are still +pending), so a client correlates positionally, never by text-match. + +Retract a still-pending (un-drained) steer: + +```console +$ curl -s -X POST http://127.0.0.1:8081/v1/sessions//cancel-steer +{"outcome":"retracted"} +``` + +`"retracted"` means the pending bundle was cleared before it drained (no `steer` +echo will land); `"none_pending"` means there was nothing to retract — the steer +already drained at a boundary (it is ordinary recorded history now) or no run is +live. Unknown session → `404`. ### ACP over stdio (`mecated acp`) diff --git a/internal/adapter/server/classification.go b/internal/adapter/server/classification.go index 5db6651140..a0607bab40 100644 --- a/internal/adapter/server/classification.go +++ b/internal/adapter/server/classification.go @@ -293,7 +293,7 @@ var serviceAccessTable = map[string]ClassificationEntry{ "ApprovePlan": {KindCallerOwned, "authorizes the session before resolving the parked plan ask"}, "Cancel": {KindCallerOwned, "authorizes via GetSession before signalling the in-flight run"}, "CancelChild": {KindCallerOwned, "authorizes the PARENT session via GetSession before reaching into its child registry"}, - "Steer": {KindCallerOwned, "authorizes via GetSession before enqueueing to the live run's inbox or promoting through StartRunContent"}, + "Steer": {KindCallerOwned, "authorizes via GetSession before enqueueing to the live run's steer inbox, promoting through StartRunContent when no live run can take it"}, "CancelSteer": {KindCallerOwned, "authorizes via GetSession before reaching into the live run's steer inbox"}, "Persist": {KindCallerOwned, "authorizes via GetSession before consulting the live run registry"}, diff --git a/internal/adapter/server/features.go b/internal/adapter/server/features.go index 2d242441c2..cdebd2a9a2 100644 --- a/internal/adapter/server/features.go +++ b/internal/adapter/server/features.go @@ -65,6 +65,12 @@ const ( // mcp_servers HERE" — which is the only useful form of the answer, since a // build-only claim would be true on a daemon that refuses every such request. FeatureMCPServersOnCreate = "mcp_servers_on_create" + + // FeatureHTTPSteer is the unary HTTP steer pair (ADR 0252): + // POST /v1/sessions/{id}/steer (text + multimodal parts + + // expected_run_id strict mode, promoted follow-ups relayed as SSE) and + // POST /v1/sessions/{id}/cancel-steer. + FeatureHTTPSteer = "http_steer" ) // FeatureScope is what the DEPLOYMENT permits, as distinct from what the build @@ -95,6 +101,7 @@ type FeatureScope struct { // repeated string and a client must treat it as a set, but a stable order keeps // diffs and golden fixtures readable. var allFeatures = []string{ + FeatureHTTPSteer, FeatureMCPServersOnCreate, FeatureServerInfo, FeatureWatchSessionEvents, diff --git a/internal/adapter/server/grpc.go b/internal/adapter/server/grpc.go index d631839700..9e2a72a2f6 100644 --- a/internal/adapter/server/grpc.go +++ b/internal/adapter/server/grpc.go @@ -573,25 +573,10 @@ func (h *HarnessServer) sendEvent(rl *runRelay, ev session.Event) { return // log-only event: consumed by the durable log, not relayed to the client wire } proto := toProto(ev) - if ev.Type == session.EvSteer && proto.GetSteer() != nil { - // The EvSteer drain echo echoes the client-minted message_id of the - // Steer frame that parked this text: the engine inbox carries text only, - // so the id lives at the Service's wire-correlation FIFO — popped here - // positionally (the TAIL). An unmatched echo (an id-less steer) rides - // with "". - id := h.svc.LookupSteerMessageID(rl.id) - if id == "" { - // The correlation FAILED: the echo carries "" and the client cannot - // match it to the frame it sent (the queue can stall — the exact - // symptom this WARN exists to make visible). No session.Event owns a - // correlation miss, so it goes to diagnostics, text clamped to a prefix. - h.svc.Diagnostics().Log(rl.logCtx, port.LevelWarn, "steer echo uncorrelated (no message_id for drained text)", "session", string(rl.id), "text_prefix", valid(firstRunes(ev.Steer.Text, 40))) - } else { - h.svc.Diagnostics().Log(rl.logCtx, port.LevelInfo, "steer drain echo correlated", - "session", string(rl.id), "message_id", id, "text_len", len(ev.Steer.Text)) - } - proto.GetSteer().MessageId = valid(id) - } + // The EvSteer drain-echo message_id stamp + its correlation diagnostics live + // in the ONE shared Service.stampSteerEcho (the HTTP SSE relay calls the + // same helper) — a non-steer event is a no-op inside it. + h.svc.stampSteerEcho(rl.logCtx, rl.id, ev, proto) if err := rl.snd.Send(&mecatlv1.ConverseResponse{Event: proto}); err != nil { rl.sendErr = err } diff --git a/internal/adapter/server/http.go b/internal/adapter/server/http.go index ea8fe16915..4df4953617 100644 --- a/internal/adapter/server/http.go +++ b/internal/adapter/server/http.go @@ -34,6 +34,8 @@ import ( // POST /v1/sessions/{id}/approve -> resolve the paused ask on the run // POST /v1/sessions/{id}/cancel -> cancel the in-flight run // POST /v1/sessions/{id}/cancel-child -> cancel ONE child (subagent) of the run +// POST /v1/sessions/{id}/steer -> enqueue a mid-run steer (unary; outcome JSON; promoted follow-up relayed as SSE) +// POST /v1/sessions/{id}/cancel-steer -> retract the pending (un-drained) steer // POST /v1/sessions/{id}/fork -> ForkSession (peer session from a history snapshot; 201) // GET /v1/sessions/{id}/events -> replay the durable event log; the stream ENDS // GET /v1/sessions/{id}/watch -> durable replay-then-follow; the stream STAYS OPEN @@ -65,6 +67,10 @@ func NewHTTPHandler(svc *Service) *HTTPHandler { h.mux.HandleFunc("POST /v1/sessions/{id}/plan:approve", h.approvePlan) h.mux.HandleFunc("POST /v1/sessions/{id}/cancel", h.cancel) h.mux.HandleFunc("POST /v1/sessions/{id}/cancel-child", h.cancelChild) + h.mux.HandleFunc("POST /v1/sessions/{id}/steer", h.steer) + h.mux.HandleFunc("POST /v1/sessions/{id}/cancel-steer", h.steerCancel) + // Deprecated alias for cancel-steer (the pre-ADR-0252 name); remove after + // clients migrate. h.mux.HandleFunc("POST /v1/sessions/{id}/fork", h.forkSession) h.mux.HandleFunc("POST /v1/sessions/{id}/adoption:preflight", h.preflightSessionAdoption) h.mux.HandleFunc("POST /v1/sessions/{id}/adopt", h.adoptSession) @@ -829,11 +835,16 @@ func (h *HTTPHandler) relayRunSSE(w http.ResponseWriter, r *http.Request, id ses if !h.svc.relayEvent(r.Context(), id, ev, true, recorder) { continue // log-only event: consumed by the durable log, not relayed to the client wire } + p := toProto(ev) + // The EvSteer drain-echo message_id stamp + its correlation diagnostics + // live in the ONE shared Service.stampSteerEcho (the gRPC Converse relay + // calls the same helper) — a non-steer event is a no-op inside it. + h.svc.stampSteerEcho(logCtx, id, ev, p) if _, err := w.Write([]byte("data: ")); err != nil { fail() continue } - if err := enc.Encode(toProto(ev)); err != nil { // Encode appends a newline + if err := enc.Encode(p); err != nil { // Encode appends a newline fail() continue } @@ -1148,6 +1159,127 @@ func (h *HTTPHandler) cancelChild(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +// steerBody is the JSON body of POST /v1/sessions/{id}/steer (ADR 0252): +// the same contract as the gRPC steer frame, parts included. +type steerBody struct { + // Text is the operator instruction to inject into the in-flight run, drained + // at the next turn boundary. Optional when Parts is non-empty (a media-only + // steer is legal, ADR 0251). + Text string `json:"text"` + // Parts carries the same multimodal content the prompt body accepts, + // decoded and validated through the ONE wire→domain choke point + // (toContentParts) — never a second validation path. + Parts []promptContentBody `json:"parts,omitempty"` + // MessageID is the CLIENT-MINTED correlation id for THIS send ("" = + // uncorrelated). It is echoed verbatim on the response and, when the steer + // lands, on the EvSteer drain echo's message_id (a WATERMARK — the latest + // contributing send's id of the bundle that drained), so the client splits + // its ordered pending queue positionally, never by text-match. + MessageID string `json:"message_id,omitempty"` + // ExpectedRunID pins the steer to a specific run (ADR 0249 strict steer). + // A mismatch — or a named run that has already gone terminal — is a 409 + // problem (code stale_run_control), never a promotion: the caller asked to + // say something to run X, not to start a new run. + ExpectedRunID string `json:"expected_run_id,omitempty"` +} + +// steerResp is the JSON response of POST /v1/sessions/{id}/steer and +// /cancel-steer: the agent.SteerOutcome string verbatim (steer: accepted | +// appended | too_late; cancel-steer: retracted | none_pending), plus the +// request's own message_id echoed back (steer only — the ACK-side echo; the +// drain-side echo rides the EvSteer event on the prompt SSE stream). +type steerResp struct { + Outcome string `json:"outcome"` + MessageID string `json:"message_id,omitempty"` + // Promoted is true when an unqualified too_late steer was promoted to a + // follow-up run that could not be relayed on this response (the + // non-streaming fallback) — the run drains into the durable event log. + Promoted bool `json:"promoted,omitempty"` +} + +// steer handles POST /v1/sessions/{id}/steer (ADR 0252) — the unary HTTP tier +// of steer-while-running, on the same terms as the gRPC frame: text and/or +// multimodal parts enqueue into the session's IN-FLIGHT run (drained at the +// next turn boundary), answered 200 {"outcome": "accepted"|"appended", +// "message_id": }. An UNQUALIFIED steer that loses the terminal race +// is PROMOTED to a follow-up run (Service.Steer's ADR 0232 contract) and the +// promoted run is relayed as SSE on this same response — or, without a +// flusher, background-drained into the durable event log with a +// {"outcome":"too_late","promoted":true} ack — never handed back bare. A +// STRICT steer (expected_run_id set) never promotes: a mismatch or a named +// run already terminal is a 409 problem (stale_run_control). Unknown/foreign +// session → 404; oversized body → 413. +func (h *HTTPHandler) steer(w http.ResponseWriter, r *http.Request) { + id := session.SessionID(r.PathValue("id")) + // Bound the body read exactly like /prompt (CWE-770): an oversized payload + // is a 413, never buffered into memory. + r.Body = http.MaxBytesReader(w, r.Body, maxPromptBodyBytes) + var body steerBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + var maxErr *http.MaxBytesError + if errors.As(err, &maxErr) { + writeError(w, http.StatusRequestEntityTooLarge, "request body too large") + return + } + writeError(w, http.StatusBadRequest, "invalid JSON body") + return + } + if body.Text == "" && len(body.Parts) == 0 { + writeError(w, http.StatusBadRequest, "text or parts is required") + return + } + parts, perr := toContentParts(body.Parts) + if perr != nil { + writeError(w, http.StatusBadRequest, perr.Error()) + return + } + outcome, promoted, run, err := h.svc.Steer(r.Context(), id, body.Text, parts, body.MessageID, body.ExpectedRunID) + if err != nil { + writeServiceError(w, err) + return + } + if !promoted || run == nil { + writeJSON(w, http.StatusOK, steerResp{Outcome: string(outcome), MessageID: body.MessageID}) + return + } + // Promoted follow-up: the run is registered and this caller owns the drain + // (Service.Steer's contract). Mirror approve's REHYDRATE precedent: relay + // as SSE when the writer can stream, else background-drain into the + // durable log and ack — the run must never be left with nothing draining. + flusher, ok := w.(http.Flusher) + if !ok { + logCtx := context.WithoutCancel(r.Context()) + recorder := NewRunEventRecorder(logCtx, h.svc, id) + go func() { + defer recorder.Close() + for ev := range run.Events() { + recorder.Observe(ev) + } + h.svc.deregister(id, run) + }() + writeJSON(w, http.StatusOK, steerResp{Outcome: string(outcome), MessageID: body.MessageID, Promoted: true}) + return + } + h.relayRunSSE(w, r, id, run, flusher, "", false) +} + +// steerCancel handles POST /v1/sessions/{id}/cancel-steer (ADR 0252, +// mirroring the cancel-child naming), retracting the session's live run's PENDING +// (un-drained) steer via Service.CancelSteer: 200 {"outcome": "retracted"} +// (the pending bundle is gone, its message-id correlation dropped) or +// {"outcome": "none_pending"} (nothing parked — already drained at a boundary, +// or no live run). No body is required; any body is ignored. Unknown/foreign +// session → 404. +func (h *HTTPHandler) steerCancel(w http.ResponseWriter, r *http.Request) { + id := session.SessionID(r.PathValue("id")) + outcome, err := h.svc.CancelSteer(r.Context(), id) + if err != nil { + writeServiceError(w, err) + return + } + writeJSON(w, http.StatusOK, steerResp{Outcome: string(outcome)}) +} + // --- team request bodies ----------------------------------------------------- // teammateSpecBody is one member's enrolment fields, shared by createTeam's diff --git a/internal/adapter/server/service.go b/internal/adapter/server/service.go index dee551eeee..6a73840401 100644 --- a/internal/adapter/server/service.go +++ b/internal/adapter/server/service.go @@ -4289,6 +4289,37 @@ func (s *Service) LookupSteerMessageID(id session.SessionID) string { return watermark } +// stampSteerEcho stamps the client-minted message_id onto an EvSteer drain +// echo's proto projection, consuming the session's watermark FIFO +// (LookupSteerMessageID). It is the SINGLE owner of the echo correlation — the +// gRPC Converse relay (HarnessServer.sendEvent) and the HTTP SSE relay +// (HTTPHandler.relayRunSSE) both call it, so the stamped id and the +// correlated-INFO / uncorrelated-WARN diagnostics cannot drift between the two +// wires. A non-EvSteer event (or a projection without the Steer payload) is a +// no-op, so callers stamp unconditionally on the hot path. +func (s *Service) stampSteerEcho(logCtx context.Context, id session.SessionID, ev session.Event, proto *mecatlv1.Event) { + if ev.Type != session.EvSteer || proto.GetSteer() == nil { + return + } + // The EvSteer drain echo echoes the client-minted message_id of the + // Steer frame that parked this text: the engine inbox carries text only, + // so the id lives at the Service's wire-correlation FIFO — popped here + // positionally (the TAIL). An unmatched echo (an id-less steer) rides + // with "". + msgID := s.LookupSteerMessageID(id) + if msgID == "" { + // The correlation FAILED: the echo carries "" and the client cannot + // match it to the frame it sent (the queue can stall — the exact + // symptom this WARN exists to make visible). No session.Event owns a + // correlation miss, so it goes to diagnostics, text clamped to a prefix. + s.Diagnostics().Log(logCtx, port.LevelWarn, "steer echo uncorrelated (no message_id for drained text)", "session", string(id), "text_prefix", valid(firstRunes(ev.Steer.Text, 40))) + } else { + s.Diagnostics().Log(logCtx, port.LevelInfo, "steer drain echo correlated", + "session", string(id), "message_id", msgID, "text_len", len(ev.Steer.Text)) + } + proto.GetSteer().MessageId = valid(msgID) +} + // isDelegationChildSessionID reports whether id carries one of the delegation // families' child-session id prefixes: agent.SubagentSessionPrefix, // agent.ParallelSessionPrefix, agent.TeamSessionPrefix — the engine's exported diff --git a/internal/adapter/server/steer_http_test.go b/internal/adapter/server/steer_http_test.go new file mode 100644 index 0000000000..4ebac734e1 --- /dev/null +++ b/internal/adapter/server/steer_http_test.go @@ -0,0 +1,381 @@ +package server_test + +// The unary HTTP tier of steer-while-running (the deferred follow-up ADR 0232 +// names): POST /v1/sessions/{id}/steer enqueues to the LIVE run's steer inbox +// via Service.Steer, promoting through the run-entry funnel (and relaying the +// promoted run as SSE) when no live run can take it, behind a too_late ack +// when the caller opts out of promotion — and POST +// /v1/sessions/{id}/cancel-steer retracts the pending (un-drained) steer via +// Service.CancelSteer. The EvSteer drain echo on the prompt SSE stream carries +// the client-minted message_id through the SAME shared Service.stampSteerEcho +// the gRPC Converse relay uses, and POST /v1/sessions advertises the steer bit +// on the capabilities echo like gRPC CreateSession does. + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + mecatlv1 "github.com/stacklok/mecatl/contracts/gen/go/mecatl/v1" + "github.com/stacklok/mecatl/engine/adapter/mockllm" + "github.com/stacklok/mecatl/engine/session" + "github.com/stacklok/mecatl/internal/adapter/server" +) + +// steerRespJSON decodes the steer / cancel-steer 200 body. +type steerRespJSON struct { + Outcome string `json:"outcome"` + MessageID string `json:"message_id"` +} + +// postSteer POSTs body to the session's steer (or cancel-steer) route and +// returns the status code plus the decoded outcome body (zero on a non-200). +func postSteer(t *testing.T, srv *httptest.Server, id, route, body string) (int, steerRespJSON) { + t.Helper() + resp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/"+route, "application/json", strings.NewReader(body)) + if err != nil { + t.Fatalf("POST %s: %v", route, err) + } + defer resp.Body.Close() + var out steerRespJSON + if resp.StatusCode == http.StatusOK { + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode %s response: %v", route, err) + } + } + return resp.StatusCode, out +} + +// httpSessionState GETs the session snapshot and returns its state string. +func httpSessionState(t *testing.T, srv *httptest.Server, id string) string { + t.Helper() + resp, err := http.Get(srv.URL + "/v1/sessions/" + id) + if err != nil { + t.Fatalf("GET session: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET session status = %d", resp.StatusCode) + } + var out struct { + State string `json:"state"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode session: %v", err) + } + return out.State +} + +// TestHTTPSteer_AcceptedEchoesMessageID: a steer POSTed against a genuinely +// LIVE run (a tool holds it mid-dispatch) is accepted — 200 +// {"outcome":"accepted","message_id":} — and the EvSteer drain echo on +// the prompt SSE stream carries the SAME client-minted message_id (the shared +// stampSteerEcho path), proving the HTTP relay correlates like the gRPC one. +func TestHTTPSteer_AcceptedEchoesMessageID(t *testing.T) { + block := &blockingTextTool{started: make(chan struct{}), release: make(chan struct{})} + llm := mockllm.New( + mockllm.ToolCallTurn(call("c1", "Read", `{"path":"a.go"}`)), + mockllm.TextTurn("turn two done"), + ) + svc := newSteerService(t, llm, nil, block) + srv := httptest.NewServer(server.NewHTTPHandler(svc)) + defer srv.Close() + + id := createHTTPSession(t, srv) + + // Steer from a side goroutine once the tool genuinely holds the run, then + // release the tool so the steer drains at the next turn boundary. + type steerAck struct { + status int + body steerRespJSON + } + acked := make(chan steerAck, 1) + go func() { + <-block.started + status, body := postSteer(t, srv, id, "steer", `{"text":"also check b.go","message_id":"m-1"}`) + acked <- steerAck{status: status, body: body} + close(block.release) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, + srv.URL+"/v1/sessions/"+id+"/prompt", strings.NewReader(`{"text":"look"}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST prompt: %v", err) + } + defer resp.Body.Close() + events := parseSSE(t, bufio.NewReader(resp.Body)) + + ack := <-acked + if ack.status != http.StatusOK { + t.Fatalf("steer status = %d, want 200", ack.status) + } + if ack.body.Outcome != "accepted" { + t.Fatalf("steer outcome = %q, want accepted", ack.body.Outcome) + } + if ack.body.MessageID != "m-1" { + t.Fatalf("steer ack message_id = %q, want m-1 (the request's own id echoed)", ack.body.MessageID) + } + + // The EvSteer drain echo rides the prompt SSE stream carrying the committed + // text AND the client-minted message_id (stamped by the shared helper). + var steerEv *mecatlv1.Event + for _, ev := range events { + if ev.GetType() == "steer" { + steerEv = ev + break + } + } + if steerEv == nil { + t.Fatalf("no steer echo event on the SSE stream: %v", typesOf(events)) + } + if got := steerEv.GetSteer().GetText(); got != "also check b.go" { + t.Fatalf("steer echo text = %q, want %q", got, "also check b.go") + } + if got := steerEv.GetSteer().GetMessageId(); got != "m-1" { + t.Fatalf("steer echo message_id = %q, want m-1 (the HTTP relay must stamp the drain echo like the gRPC relay)", got) + } + if res := lastResult(t, events); res.GetStop() != "end_turn" || res.GetText() != "turn two done" { + t.Fatalf("result = %+v", res) + } +} + +// TestHTTPSteer_TooLatePromotesAndRelays (ADR 0252): an UNQUALIFIED steer +// against a session with no live run is promoted to a follow-up run — +// Service.Steer's ADR 0232 contract — and the promoted run is relayed as SSE +// on this same response, terminal result included, leaving nothing registered. +func TestHTTPSteer_TooLatePromotesAndRelays(t *testing.T) { + llm := mockllm.New(mockllm.TextTurn("first"), mockllm.TextTurn("promoted answer")) + svc := newSteerService(t, llm, nil) + srv := httptest.NewServer(server.NewHTTPHandler(svc)) + defer srv.Close() + + id := createHTTPSession(t, srv) + resp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/prompt", "application/json", + strings.NewReader(`{"text":"hello"}`)) + if err != nil { + t.Fatalf("POST prompt: %v", err) + } + parseSSE(t, bufio.NewReader(resp.Body)) // drive the run to its terminal + resp.Body.Close() + if got := httpSessionState(t, srv, id); got != "completed" { + t.Fatalf("precondition: state = %q, want completed", got) + } + + promoteResp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/steer", "application/json", + strings.NewReader(`{"text":"late follow-up"}`)) + if err != nil { + t.Fatalf("POST steer: %v", err) + } + defer promoteResp.Body.Close() + if promoteResp.StatusCode != http.StatusOK { + t.Fatalf("steer status = %d, want 200", promoteResp.StatusCode) + } + if ct := promoteResp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { + t.Fatalf("promoted steer Content-Type = %q, want an SSE relay", ct) + } + events := parseSSE(t, bufio.NewReader(promoteResp.Body)) + if res := lastResult(t, events); res.GetText() != "promoted answer" { + t.Fatalf("promoted run result text = %q, want the second turn", res.GetText()) + } + if _, live := svc.LookupRun(session.SessionID(id)); live { + t.Fatalf("the promoted run must be deregistered after the relay") + } + if calls := llm.Calls(); calls != 2 { + t.Fatalf("provider calls = %d, want 2 (the promoted follow-up drives one)", calls) + } +} + +// TestHTTPSteer_StrictNeverPromotes (ADR 0249/0252): expected_run_id names a +// run that already ended — the steer is refused 409 (stale_run_control), +// nothing is promoted, the state and call count are untouched. The caller +// keeps the text. +func TestHTTPSteer_StrictNeverPromotes(t *testing.T) { + llm := mockllm.New(mockllm.TextTurn("first")) + svc := newSteerService(t, llm, nil) + srv := httptest.NewServer(server.NewHTTPHandler(svc)) + defer srv.Close() + + id := createHTTPSession(t, srv) + resp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/prompt", "application/json", + strings.NewReader(`{"text":"hello"}`)) + if err != nil { + t.Fatalf("POST prompt: %v", err) + } + parseSSE(t, bufio.NewReader(resp.Body)) + resp.Body.Close() + + strictResp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/steer", "application/json", + strings.NewReader(`{"text":"late follow-up","expected_run_id":"r-gone"}`)) + if err != nil { + t.Fatalf("POST strict steer: %v", err) + } + defer strictResp.Body.Close() + if strictResp.StatusCode != http.StatusConflict { + t.Fatalf("strict steer status = %d, want 409", strictResp.StatusCode) + } + var problem struct { + Code string `json:"code"` + } + if err := json.NewDecoder(strictResp.Body).Decode(&problem); err != nil { + t.Fatalf("decode problem body: %v", err) + } + if problem.Code != "stale_run_control" { + t.Fatalf("problem code = %q, want stale_run_control", problem.Code) + } + if _, live := svc.LookupRun(session.SessionID(id)); live { + t.Fatalf("a strict too_late steer must NOT register a promoted run") + } + if calls := llm.Calls(); calls != 1 { + t.Fatalf("provider calls = %d, want 1", calls) + } +} + +// TestHTTPSteer_Rejections pins the request-validation edges: unknown session → +// 404 (absence-equivalent, mirroring cancel/approve), empty text → 400, and an +// oversized body → 413 (the same MaxBytesReader bound as /prompt). +func TestHTTPSteer_Rejections(t *testing.T) { + svc := newSteerService(t, mockllm.New(mockllm.TextTurn("x")), nil) + srv := httptest.NewServer(server.NewHTTPHandler(svc)) + defer srv.Close() + id := createHTTPSession(t, srv) + + t.Run("unknown session", func(t *testing.T) { + status, _ := postSteer(t, srv, "does-not-exist", "steer", `{"text":"hi"}`) + if status != http.StatusNotFound { + t.Fatalf("status = %d, want 404", status) + } + }) + + t.Run("empty text", func(t *testing.T) { + status, _ := postSteer(t, srv, id, "steer", `{"text":""}`) + if status != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", status) + } + }) + + t.Run("oversized body", func(t *testing.T) { + // A body just over the wire cap (37 MiB > 32 MiB maxPromptBodyBytes). + huge := bytes.Repeat([]byte("A"), 37<<20) + body := append([]byte(`{"text":"`), huge...) + body = append(body, []byte(`"}`)...) + resp, err := http.Post(srv.URL+"/v1/sessions/"+id+"/steer", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatalf("POST steer: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("status = %d, want 413 for oversized body", resp.StatusCode) + } + }) +} + +// TestHTTPSteerCancel_RetractsPending: cancel-steer against a live run with a +// PENDING steer reports retracted, and the retracted text never drains (no +// EvSteer echo lands on the stream); with nothing pending it reports +// none_pending; an unknown session is 404. +func TestHTTPSteerCancel_RetractsPending(t *testing.T) { + block := &blockingTextTool{started: make(chan struct{}), release: make(chan struct{})} + llm := mockllm.New( + mockllm.ToolCallTurn(call("c1", "Read", `{"path":"a.go"}`)), + mockllm.TextTurn("done"), + ) + svc := newSteerService(t, llm, nil, block) + srv := httptest.NewServer(server.NewHTTPHandler(svc)) + defer srv.Close() + + id := createHTTPSession(t, srv) + + // Steer then immediately retract, both while the tool holds the run. + outcomes := make(chan [2]steerRespJSON, 1) + go func() { + <-block.started + _, first := postSteer(t, srv, id, "steer", `{"text":"scratch that"}`) + _, second := postSteer(t, srv, id, "cancel-steer", "") + outcomes <- [2]steerRespJSON{first, second} + close(block.release) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, + srv.URL+"/v1/sessions/"+id+"/prompt", strings.NewReader(`{"text":"look"}`)) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST prompt: %v", err) + } + defer resp.Body.Close() + events := parseSSE(t, bufio.NewReader(resp.Body)) + + got := <-outcomes + if got[0].Outcome != "accepted" { + t.Fatalf("steer outcome = %q, want accepted", got[0].Outcome) + } + if got[1].Outcome != "retracted" { + t.Fatalf("cancel-steer outcome = %q, want retracted", got[1].Outcome) + } + if hasType(events, "steer") { + t.Fatalf("a retracted steer must never drain/echo: %v", typesOf(events)) + } + + // Nothing pending any more (the run is over): none_pending, never an error. + status, none := postSteer(t, srv, id, "cancel-steer", "") + if status != http.StatusOK || none.Outcome != "none_pending" { + t.Fatalf("cancel-steer with nothing pending = (%d, %q), want (200, none_pending)", status, none.Outcome) + } + + // Unknown session: absence-equivalent 404. + if status, _ := postSteer(t, srv, "does-not-exist", "cancel-steer", ""); status != http.StatusNotFound { + t.Fatalf("cancel-steer unknown session status = %d, want 404", status) + } +} + +// TestHTTPCreateSession_AdvertisesSteer: the POST /v1/sessions capabilities +// echo carries the steer bit — true when the engine arms Deps.EnableSteer, +// absent/false otherwise — so an HTTP client gates the steer affordance on the +// same composition-computed value gRPC clients read. +func TestHTTPCreateSession_AdvertisesSteer(t *testing.T) { + decodeCaps := func(t *testing.T, srv *httptest.Server) bool { + t.Helper() + resp, err := http.Post(srv.URL+"/v1/sessions", "application/json", + strings.NewReader(`{"workspace":"/ws"}`)) + if err != nil { + t.Fatalf("POST /v1/sessions: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + t.Fatalf("create status = %d", resp.StatusCode) + } + var out struct { + Capabilities struct { + Steer bool `json:"steer"` + } `json:"capabilities"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("decode create: %v", err) + } + return out.Capabilities.Steer + } + + on := httptest.NewServer(server.NewHTTPHandler(newSteerService(t, mockllm.New(mockllm.TextTurn("x")), nil))) + defer on.Close() + if !decodeCaps(t, on) { + t.Fatalf("capabilities steer = false, want true (EnableSteer armed)") + } + + off := httptest.NewServer(server.NewHTTPHandler(newService(t, mockllm.New(mockllm.TextTurn("x")), allowRules()))) + defer off.Close() + if decodeCaps(t, off) { + t.Fatalf("capabilities steer = true, want false (EnableSteer off)") + } +} diff --git a/user-docs/building/deployment/grpc-http.md b/user-docs/building/deployment/grpc-http.md index da334415c7..f154b97661 100644 --- a/user-docs/building/deployment/grpc-http.md +++ b/user-docs/building/deployment/grpc-http.md @@ -91,9 +91,15 @@ and `curl` examples, see the [HTTP/SSE API reference](https://github.com/stacklo ### Important difference: steering -gRPC can send control frames to a live `Converse` stream, including an in-flight -steering instruction or its cancellation. HTTP/SSE has no equivalent client-to- -server mid-run steering channel. Use gRPC when your client needs that control. +gRPC sends control frames on the live `Converse` stream itself. HTTP/SSE +steers through a unary pair — `POST /v1/sessions/{id}/steer` (text and/or +multimodal `parts`, an optional strict `expected_run_id`) and +`POST /v1/sessions/{id}/cancel-steer` — gated on the `steer` capability bit +and advertised as `http_steer` in `GET /v1/compatibility`. An unqualified +steer that loses the race against the run's end is promoted into a follow-up +run and relayed as SSE on the same response; a strict steer never promotes — +it answers `409` (`stale_run_control`) and the caller keeps the text. Use +gRPC when your client needs in-stream control frames. ## Connect securely