fix(runs): reject cancel actions on GET stream joins (#5092)

* 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

@@ -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)

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 (`<think>...</think>`, 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 `<think>` 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. |

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
 # ---------------------------------------------------------------------------

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

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) ───────────────────
 
 

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

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