From a956bbc030560328bf1f446e48c402a196f909f3 Mon Sep 17 00:00:00 2001
From: Sunshine <1621354073@qq.com>
Date: Tue, 1 Sep 2026 15:51:37 +0800
Subject: [PATCH] fix(runs): reject cancel actions on GET stream joins (#5092)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
* fix(runs): reject cancel actions on GET stream joins
stream_existing_run is registered for both GET and POST, and its
?action=interrupt|rollback branch cancels the run. The CSRF middleware
exempts GET, so a session-authenticated browser could be forced
cross-site (img/script/top-level navigation) into
GET /api/threads/{id}/runs/{run_id}/stream?action=interrupt|rollback —
a state-changing GET that bypasses the CSRF protection guarding the
POST variant. Introduced with the dual registration in #1403.
The handler's docstring already documents cancel-then-stream as
POST-only (the LangGraph SDK's joinStream/useStream stop button uses
POST); enforce it: GET with an action answers 405, action-less GET
joins and POST cancel-then-stream are unchanged.
Regression drives the real router: GET+action is 405 with the run left
running, plain GET join still streams, POST+action still cancels.
* fix(runs): scope the 405 detail to the action requirement
"GET is a read-only stream join" overstates the current main: on a
locally-owned run with the default on_disconnect=cancel, a GET join's
disconnect can still trigger cancellation. That observer-disconnect
vector is closed by #5041; the detail here should only claim what this
guard enforces.
* fix(runs): harden GET stream action rejection
* fix(runs): align stream schema with method contract
* test(runs): pin GET stream action 405 through the production stack
Review follow-up (defence-in-depth): the GET-action suite drove bare
FastAPI() apps, so nothing pinned that a session-authenticated
cross-site GET reaches the route gate at all once CSRF exempts the
safe method. test_pat_auth.py already assembles the production
middleware order (AuthMiddleware inner, CSRFMiddleware outer), so its
mirror app now registers the real _reject_get_stream_action
dependency on a GET join route.
The new case pins the end-to-end premise: an authenticated GET
?action=interrupt is answered 405 + Allow: POST by the production
route dependency, while the same unauthenticated GET dies at
AuthMiddleware's 401 before any route logic runs.
Validation: focused suites (test_pat_auth, test_stream_get_action,
test_csrf_middleware) — 62 passed; ruff check + format clean; the new
case errors on the pre-fix baseline (guard absent), confirming the
pin.
---
README.md | 8 ++
backend/app/gateway/AGENTS.md | 2 +-
backend/app/gateway/routers/thread_runs.py | 70 ++++++++---
backend/tests/test_openapi_operation_ids.py | 14 +++
backend/tests/test_pat_auth.py | 41 ++++++-
backend/tests/test_sse_observer_disconnect.py | 2 +-
.../tests/test_stream_get_action_rejected.py | 116 ++++++++++++++++++
7 files changed, 231 insertions(+), 22 deletions(-)
create mode 100644 backend/tests/test_stream_get_action_rejected.py
diff --git a/README.md b/README.md
index 362ebb3fe..cdd9843ae 100644
--- a/README.md
+++ b/README.md
@@ -799,6 +799,14 @@ LangSmith and Langfuse attach as LangChain callbacks, so you can enable both and
For Docker deployments, tracing is disabled by default. Set `LANGSMITH_TRACING=true` and `LANGSMITH_API_KEY` in your `.env` to enable it.
+#### Existing-Run Stream Actions
+
+Existing-run SSE joins are observation-only on `GET`: supplying
+`action=interrupt|rollback` returns `405`. Cancellation on this stream route is
+`POST`-only and requires the `runs:cancel` permission. Accordingly, the OpenAPI
+contract exposes `action` and `wait` only on `POST`; the `GET` operation exposes
+only its path parameters.
+
#### Personal Access Tokens
Non-interactive clients (CI pipelines, scripts, server-to-server integrations)
diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md
index 6929040c8..bb23f8fd7 100644
--- a/backend/app/gateway/AGENTS.md
+++ b/backend/app/gateway/AGENTS.md
@@ -57,7 +57,7 @@ owner-scoped assistant version selection remains enabled.
| **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update explicitly. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). |
| **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |
| **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
-| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, an empty run-event message feed is seeded from an existing checkpoint head so legacy checkpoint-only history receives earlier thread-global sequence numbers and remains visible after the new run; a thread with no checkpoint or an already-populated feed skips this compatibility path. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. |
+| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty event feed from an existing checkpoint so legacy checkpoint-only history keeps earlier thread-global ordering and stays visible; skip without a checkpoint or when the feed is populated. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. |
| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
| **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback |
| **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. |
diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py
index f257820c5..66b5b3d3a 100644
--- a/backend/app/gateway/routers/thread_runs.py
+++ b/backend/app/gateway/routers/thread_runs.py
@@ -18,7 +18,7 @@ from copy import deepcopy
from datetime import UTC, datetime
from typing import Any, Literal
-from fastapi import APIRouter, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import Response, StreamingResponse
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
@@ -1019,26 +1019,34 @@ async def join_run(thread_id: ThreadId, run_id: str, request: Request) -> Stream
)
-# Register GET and POST as separate routes so each method gets a unique OpenAPI
-# operationId. ``api_route(methods=["GET", "POST"])`` shares one route registration
-# across both methods, which makes FastAPI emit the same ``operationId`` twice and
-# warn about a duplicate operation id during OpenAPI generation.
-@router.get("/{thread_id}/runs/{run_id}/stream", response_model=None)
-@router.post("/{thread_id}/runs/{run_id}/stream", response_model=None)
-@require_permission("runs", "read", owner_check=True)
-async def stream_existing_run(
+def _reject_get_stream_action(
+ action: Literal["interrupt", "rollback"] | None = Query(default=None, include_in_schema=False),
+) -> None:
+ """Keep the GET join read-only before thread ownership or run lookup."""
+ if action is not None:
+ # SameSite=Lax still sends the session cookie on a cross-site top-level
+ # safe navigation. Reject the state-changing action before the endpoint
+ # wrapper performs its thread ownership lookup.
+ raise HTTPException(
+ status_code=405,
+ detail="`action` is only supported on POST requests",
+ headers={"Allow": "POST"},
+ )
+
+
+async def _stream_existing_run(
thread_id: ThreadId,
run_id: str,
request: Request,
- action: Literal["interrupt", "rollback"] | None = Query(default=None, description="Cancel action"),
- wait: int = Query(default=0, description="Block until cancelled (1) or return immediately (0)"),
-):
- """Join an existing run's SSE stream (GET), or cancel-then-stream (POST).
+ *,
+ action: Literal["interrupt", "rollback"] | None,
+ wait: int,
+) -> Response:
+ """Join an existing run's SSE stream, optionally cancelling it first.
- The LangGraph SDK's ``joinStream`` and ``useStream`` stop button both use
- ``POST`` to this endpoint. When ``action=interrupt`` or ``action=rollback``
- is present the run is cancelled first; the response then streams any
- remaining buffered events so the client observes a clean shutdown.
+ When ``action=interrupt`` or ``action=rollback`` is present the run is
+ cancelled first; the response then streams any remaining buffered events
+ so the client observes a clean shutdown.
"""
require_cancel_permission_when_action(request, action)
@@ -1099,6 +1107,34 @@ async def stream_existing_run(
)
+# Register POST before GET to preserve the historical route precedence and
+# Allow header, while separate signatures keep cancel-only parameters off the
+# GET schema. The shared route name keeps generated operationIds stable.
+@router.post("/{thread_id}/runs/{run_id}/stream", response_model=None, name="stream_existing_run")
+@require_permission("runs", "read", owner_check=True)
+async def stream_existing_run(
+ thread_id: ThreadId,
+ run_id: str,
+ request: Request,
+ action: Literal["interrupt", "rollback"] | None = Query(default=None, description="Cancel action"),
+ wait: int = Query(default=0, description="Block until cancelled (1) or return immediately (0)"),
+) -> Response:
+ """Join an existing run's SSE stream, optionally cancelling it first."""
+ return await _stream_existing_run(thread_id, run_id, request, action=action, wait=wait)
+
+
+@router.get(
+ "/{thread_id}/runs/{run_id}/stream",
+ response_model=None,
+ dependencies=[Depends(_reject_get_stream_action)],
+ name="stream_existing_run",
+)
+@require_permission("runs", "read", owner_check=True)
+async def join_existing_run_stream(thread_id: ThreadId, run_id: str, request: Request) -> Response:
+ """Join an existing run's observation-only SSE stream."""
+ return await _stream_existing_run(thread_id, run_id, request, action=None, wait=0)
+
+
# ---------------------------------------------------------------------------
# Messages / Events / Token usage endpoints
# ---------------------------------------------------------------------------
diff --git a/backend/tests/test_openapi_operation_ids.py b/backend/tests/test_openapi_operation_ids.py
index d6bb46162..318f06606 100644
--- a/backend/tests/test_openapi_operation_ids.py
+++ b/backend/tests/test_openapi_operation_ids.py
@@ -77,3 +77,17 @@ def test_stream_existing_run_exposes_distinct_get_and_post(openapi_spec: dict) -
post_op_id = path_item["post"].get("operationId")
assert get_op_id and post_op_id, "Both GET and POST must have operationIds"
assert get_op_id != post_op_id, f"GET and POST share operationId {get_op_id!r}, which breaks OpenAPI codegen"
+
+
+def test_stream_existing_run_exposes_method_specific_query_parameters(openapi_spec: dict) -> None:
+ """Only POST advertises the cancel-then-stream query contract."""
+ path = "/api/threads/{thread_id}/runs/{run_id}/stream"
+ path_item = openapi_spec["paths"][path]
+
+ get_parameters = {(parameter["in"], parameter["name"]) for parameter in path_item["get"].get("parameters", [])}
+ post_parameters = {(parameter["in"], parameter["name"]) for parameter in path_item["post"].get("parameters", [])}
+
+ assert ("query", "action") not in get_parameters
+ assert ("query", "wait") not in get_parameters
+ assert ("query", "action") in post_parameters
+ assert ("query", "wait") in post_parameters
diff --git a/backend/tests/test_pat_auth.py b/backend/tests/test_pat_auth.py
index 8d460c15c..eeaa05c4b 100644
--- a/backend/tests/test_pat_auth.py
+++ b/backend/tests/test_pat_auth.py
@@ -1,8 +1,9 @@
"""Integration tests for PAT authentication (#4849).
Covers credential precedence in AuthMiddleware, the CSRF boundary for
-Bearer-authenticated requests, scope intersection, PAT management routes,
-and the self-protection rules (a PAT may not manage PATs or auth state).
+Bearer-authenticated requests and for the safe-method stream join (#5092),
+scope intersection, PAT management routes, and the self-protection rules
+(a PAT may not manage PATs or auth state).
"""
from __future__ import annotations
@@ -12,7 +13,7 @@ from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
import pytest
-from fastapi import FastAPI, Request
+from fastapi import Depends, FastAPI, Request
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from starlette.testclient import TestClient
@@ -115,6 +116,20 @@ def _make_pat_app(with_pat_repo: bool = True):
require_cancel_permission_when_action(request, action)
return {"ok": True}
+ # Mirrors the real GET-only join surface (thread_runs.py
+ # join_existing_run_stream): registers the production route dependency
+ # that rejects cancel actions before thread ownership or run lookup, so
+ # the guard is exercised through the production middleware order above.
+ from app.gateway.routers.thread_runs import _reject_get_stream_action
+
+ @app.get(
+ "/api/threads/{thread_id}/runs/{run_id}/stream",
+ dependencies=[Depends(_reject_get_stream_action)],
+ )
+ @require_permission("runs", "read")
+ async def join_stream(thread_id: str, run_id: str, request: Request):
+ return {"ok": True}
+
# Mirrors the real run-creation entrypoints (thread_runs.py / runs.py):
# runs:create at the decorator, plus the cancel-capability gate that
# start_run applies to mutating multitask strategies. RunCreateRequest is
@@ -311,6 +326,26 @@ def test_auth_endpoint_origin_check_not_bypassed_by_bearer(client):
assert response.json()["detail"] == "Cross-site auth request denied."
+def test_session_get_stream_action_dies_at_route_gate_not_csrf(client):
+ """#5092 defence-in-depth, end-to-end through the production middleware
+ order: SameSite=Lax still attaches the session cookie to a cross-site
+ top-level GET, and CSRF exempts safe methods — so the route gate is the
+ only thing standing between that navigation and a run cancel. An
+ authenticated GET with ?action=interrupt is answered 405 + Allow: POST by
+ the real _reject_get_stream_action dependency, while the same
+ unauthenticated GET dies at AuthMiddleware's 401 before any route logic
+ runs."""
+ _session_cookie(client)
+ denied = client.get("/api/threads/t1/runs/run-1/stream?action=interrupt")
+ assert denied.status_code == 405
+ assert denied.headers["allow"] == "POST"
+ assert denied.json()["detail"] == "`action` is only supported on POST requests"
+
+ client.cookies.clear()
+ unauthed = client.get("/api/threads/t1/runs/run-1/stream?action=interrupt")
+ assert unauthed.status_code == 401
+
+
# ── Management routes + self-protection (#4849 point 6) ───────────────────
diff --git a/backend/tests/test_sse_observer_disconnect.py b/backend/tests/test_sse_observer_disconnect.py
index b45ccbbb1..99e90db97 100644
--- a/backend/tests/test_sse_observer_disconnect.py
+++ b/backend/tests/test_sse_observer_disconnect.py
@@ -104,7 +104,7 @@ def test_join_routes_wire_sse_consumer_as_observers():
from app.gateway.routers import thread_runs
thread_runs_source = inspect.getsource(thread_runs)
- # join_run + stream_existing_run (GET and POST share one handler)
+ # join_run + the shared existing-run stream implementation
assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)") == 2
# stream_run — the creator's create-and-stream endpoint
assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr),") == 1
diff --git a/backend/tests/test_stream_get_action_rejected.py b/backend/tests/test_stream_get_action_rejected.py
new file mode 100644
index 000000000..ac53d0512
--- /dev/null
+++ b/backend/tests/test_stream_get_action_rejected.py
@@ -0,0 +1,116 @@
+"""GET on the join-stream route must not carry cancel actions.
+
+The existing-run stream path supports both GET and POST; POST's ``action``
+branch cancels the run. The CSRF middleware exempts GET, while a SameSite=Lax
+session cookie still accompanies a cross-site top-level safe navigation. An
+attacker-induced navigation to
+``GET .../runs/{run_id}/stream?action=interrupt|rollback`` was therefore a
+state-changing GET that bypassed the CSRF protection guarding the POST
+variant. These tests pin that GET stays a read-only join and POST keeps
+cancelling.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from _router_auth_helpers import make_authed_test_app
+from fastapi.testclient import TestClient
+
+from app.gateway.routers import thread_runs
+from deerflow.runtime import RunManager, RunStatus
+from deerflow.runtime.stream_bridge import MemoryStreamBridge
+
+THREAD_ID = "thread-get-action"
+
+
+def _make_seeded_run_client(run_status: RunStatus = RunStatus.running) -> tuple[TestClient, RunManager, str]:
+ mgr = RunManager()
+
+ async def _seed():
+ record = await mgr.create(THREAD_ID)
+ await mgr.set_status(record.run_id, run_status)
+ return record.run_id
+
+ run_id = asyncio.run(_seed())
+ app = make_authed_test_app()
+ app.include_router(thread_runs.router)
+ app.state.run_manager = mgr
+ app.state.stream_bridge = MemoryStreamBridge()
+ return TestClient(app, raise_server_exceptions=False), mgr, run_id
+
+
+def _get_run_status(mgr: RunManager, run_id: str) -> RunStatus:
+ async def _read_status() -> RunStatus:
+ record = await mgr.get(run_id)
+ assert record is not None
+ return record.status
+
+ return asyncio.run(_read_status())
+
+
+@pytest.mark.parametrize("action", ("interrupt", "rollback"))
+def test_get_with_cancel_action_is_rejected(action: str):
+ """GET + action=interrupt|rollback must answer 405, not cancel."""
+ client, mgr, run_id = _make_seeded_run_client()
+ response = client.get(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action={action}")
+ assert response.status_code == 405
+ assert response.headers["allow"] == "POST"
+ assert "POST" in response.json()["detail"]
+ assert _get_run_status(mgr, run_id) == RunStatus.running
+
+
+def test_get_with_invalid_action_has_one_validation_error():
+ """The dependency must not duplicate the endpoint's query validation."""
+ client, mgr, run_id = _make_seeded_run_client()
+
+ response = client.get(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action=invalid")
+
+ assert response.status_code == 422
+ assert len(response.json()["detail"]) == 1
+ assert _get_run_status(mgr, run_id) == RunStatus.running
+
+
+def test_unsupported_method_preserves_post_allow_header():
+ """Splitting the handlers must not change Starlette's route precedence."""
+ client, _, run_id = _make_seeded_run_client()
+
+ response = client.put(f"/api/threads/{THREAD_ID}/runs/{run_id}/stream")
+
+ assert response.status_code == 405
+ assert response.headers["allow"] == "POST"
+
+
+def test_get_with_action_is_rejected_before_owner_lookup():
+ """The method gate must not reveal whether a thread metadata row exists."""
+ app = make_authed_test_app(owner_check_passes=False)
+ app.include_router(thread_runs.router)
+ client = TestClient(app, raise_server_exceptions=False)
+
+ response = client.get(f"/api/threads/{THREAD_ID}/runs/missing-run/stream?action=interrupt")
+
+ assert response.status_code == 405
+ assert response.headers["allow"] == "POST"
+ app.state.thread_store.check_access.assert_not_awaited()
+
+
+def test_get_without_action_still_joins():
+ """The method guard must not break the plain read-only GET join. The
+ seeded run is terminal so the SSE stream emits `end` and completes."""
+ client, _, run_id = _make_seeded_run_client(run_status=RunStatus.success)
+ with client.stream("GET", f"/api/threads/{THREAD_ID}/runs/{run_id}/stream") as response:
+ assert response.status_code == 200
+ events = [line for line in response.iter_lines() if line.startswith("event:")]
+
+ assert events[-1].strip() == "event: end"
+
+
+@pytest.mark.parametrize("action", ("interrupt", "rollback"))
+def test_post_with_cancel_action_still_cancels(action: str):
+ """The documented POST cancel-then-stream flow is unchanged."""
+ client, mgr, run_id = _make_seeded_run_client()
+ with client.stream("POST", f"/api/threads/{THREAD_ID}/runs/{run_id}/stream?action={action}") as response:
+ assert response.status_code == 200
+
+ assert _get_run_status(mgr, run_id) == RunStatus.interrupted