feat: propagate end-user identity across the serving plane (#14550)

* feat(serving): make session scoping consistent across serving APIs

The serving-plane end-user session scoping (merge an identified user into the
effective session_id; move an anonymous run into a reserved, non-persisting
namespace) previously existed only on the v2 workflow router. Extend it to the
v1 serving entry points so per-user memory isolation is identical no matter
which door a run comes through.

Introduce one shared helper, resolve_serving_scope, in lfx.workflow.
end_user_identity: it reads the serving settings, resolves the trusted
end-user identity, and returns the ScopedSession (effective session_id +
persist), or None when the feature is off (byte-for-byte the pre-feature
behavior, which is the default). The v2 router's _scope_parsed_to_end_user is
refactored onto it (behavior preserved) so the scoping decision lives in
exactly one place, and end_user_required_detail centralizes the HTTP 401 body.

simple_run_flow gains an http_request parameter and applies the scope before
run_graph_internal; it is threaded from every v1 serving caller — /run (sync
and streaming, via _run_flow_internal and run_flow_generator), the webhook
path (simple_run_flow_task), and the OpenAI Responses endpoint. /build (GUI
editor) and the public playground path are intentionally excluded. MCP inbound
calls simple_run_flow directly with no live request, so it runs with
http_request=None for now (scoping skipped, BC-safe) and gets its own header
propagation in a follow-up.

No DB schema change.

Tests:
- resolve_serving_scope unit contract (feature off / identified / anonymous /
  spoofed-trust-off / required-401).
- v2 router tests updated for the shared-helper stub target; behavior preserved.
- HTTP integration over the real /run stack: identified scopes the returned
  session, two end-users on one session id are isolated, an anonymous run lands
  in the reserved namespace, the feature-off default is unchanged, and a
  required-but-absent identity is rejected with 401.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* feat(serving): scope message.user_id to the end user on identified serving runs

P1 made session scoping consistent across serving entry points. This carries the
end-user identity onto the graph so an identified run stamps its chat messages
with the end user, not the shared service account, and both the write and read
paths resolve the owner the same way.

graph.end_user_id is a new in-memory carrier (default None, so the editor plane
and every feature-off run are byte-for-byte unchanged); subgraphs inherit it so
sub-flows and loop iterations scope to the same end user. The P1
resolve_serving_scope helper now returns the raw id on ScopedSession, so both v1
(simple_run_flow sets graph.end_user_id directly) and v2 (ParsedWorkflowRun ->
the sync/stream/background build sites, surviving the worker re-parse) read it
from the one shared seam.

A single resolver, resolve_message_owner_id in lfx.memory.flow_context, resolves
message.user_id as the end-user id when present, else the executing user id. It
is used by both the write path (Component._store_message) and the read path
(_safe_graph_user_id) so the stored owner and the retrieval predicate always
agree. message.user_id is UUID-typed while the header is an opaque string, so a
non-UUID end-user id is mapped to a stable uuid5 (fixed namespace) rather than
collapsing onto the service account; a UUID-shaped id is used directly.

Behavior by case: an identified serving run stores under the end user; an
anonymous serving run is already ephemeral (persist=False) and writes nothing;
an editor / feature-off run stores under the executing user exactly as before.
No DB schema change (reuses the existing nullable message.user_id column).

Tests: resolver + uuid5-derivation contract, ScopedSession carries the id,
subgraph inheritance, background round-trip, and end-to-end proofs (graph run
and the real /run HTTP stack) that an identified run stamps message.user_id
with the end user, not the service account. Live-verified with curl + DB
inspection: two end users on one session id isolated, anonymous persists zero
rows.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* feat(serving): isolate job status/stop/resume to the end user

On the serving plane every end user shares the one service account (SID), so the
existing job status/stop/resume checks (keyed on current_user.id = SID) let any
end user read, stop, or resume another end user's run. Scope the jobs lifecycle
to the end user.

job.user_id is dual-role: besides ownership it is the re-enqueue executing user
(_user_stub(job.user_id) -> resolve_flow_for_execution fetches the SID-owned flow,
before any graph exists to carry the SID). Stamping the end user there would 404
every identified resume/recovery. So the end user is recorded in
job_metadata['end_user_id'] (JSON, no schema change) while job.user_id stays SID,
and isolation is a check layer:

- create_job(+end_user_id) records it at the 3 live serving create sites
  (v2 sync, v2 background submit, v1 simple_run_flow); key omitted when absent.
- _caller_owns_job_end_user gates GET status / POST stop / POST {id}/resume:
  superuser bypass, else derive(request header) == derive(job end_user_id).
  Mismatch -> 404 (no existence leak). SID owner / superuser check and
  ensure_resume_execute_permission on resume are unchanged.
- resolve_serving_end_user_id (new lfx helper) resolves the raw trusted id with
  require_identity forced off, so a status/stop/resume call never raises.
- resume re-applies graph.end_user_id on the checkpoint-rebuilt graph (F5), so
  post-pause messages stamp the end user, not the SID.

BC: feature off / anonymous -> no end_user_id stored, check is a pass-through.
No migration; execution-as-SID invariant intact.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* feat(serving): propagate end-user identity through MCP inbound and outbound

MCP-originated runs bypassed the serving-plane end-user scoping every other entry
point (/run, /workflows) applies, and an MCPTools component calling out had no way
to forward the end user to a sibling project. Wire both, fail-closed.

Inbound (server): simple_run_flow scopes a run only when handed an http_request,
but the MCP SDK dispatch has no live request on the call chain. The streamable
endpoint now stashes request.headers in a contextvar (current_request_headers_ctx,
the same pattern as current_request_variables_ctx); handle_call_tool replays them
through a minimal shim as http_request, so an MCP tool run gets the identical
session prefix + graph.end_user_id + persist + create_job(end_user_id) as /run. A
REQUIRED-but-absent identity raises and is caught by the existing handler as a tool
error. SSE handlers are left untouched (SSE is unsupported in v1 projects).

Outbound (client): update_tools takes the run's end_user_id (from graph.end_user_id)
and, for Streamable_HTTP/SSE targets, appends the serving end-user header ONLY when
the target host is on the operator allowlist LANGFLOW_SERVING_INTERNAL_MCP_HOSTS
(host or host:port; default empty). External hosts never receive it — the id is PII.
Detection is host-only because this code is in lfx and cannot import langflow's URL
helper; the host allowlist is the trust boundary. The connection pool key
(_get_server_key) already hashes url+header values, so per-end-user headers yield
distinct pooled sessions with no cross-user reuse.

BC: feature off / anonymous / empty allowlist -> both paths are no-ops. Existing
by-name header substitution is unaffected.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* feat(serving): attribute telemetry, agent file writes, and monitor queries to the end user

The final serving-plane identity slice: three consumers of graph.end_user_id, each
strict-BC (no-op when the end user is absent — editor / anonymous / feature-off).

Telemetry: initialize_run — the one seam v1/v2/MCP/resume all hit before start_tracers
— fills the separate tracing_user_id from end_user_id when set and not already provided.
The primary trace user_id stays the SID (executing/billing owner); the end user surfaces
as the langflow.tracing_user_id label. An explicit caller label (v1 input_request.user_id)
still wins; editor v2 stays None.

Filesystem: _resolve_user_id prepends graph.end_user_id to the sandbox-namespace
candidates, so an identified end user owns the files a component writes during their
session (read and write share the per-call pinned id). This is the agent's sandbox tool,
not the SID-owned resource lookups (KB / uploaded files / variables), which stay SID.

Monitor: GET /monitor/messages gains an optional end_user_id filter, resolved through
derive_message_owner_uuid (D6) so the predicate matches the UUID the write stamped —
an indexed per-end-user pull instead of a session-prefix scan. The flow-ownership gate
is unchanged.

MemoryBaseSession stamping is intentionally skipped: per-user MB memory is already
segregated by the session-id prefix and the table has no user column, so it cannot be
done without a schema change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* [autofix.ci] apply automated fixes

* fix(serving): gate the pending list, harden job-owner checks, webhook 401, and telemetry PII

Addresses first-round review on the end-user identity PR.

B1 (blocker): GET /workflows/pending enumerated suspended HITL jobs by flow_id filtered
only on the shared SID, leaking another end user's merged session_id + HITL prompt. It now
takes http_request and filters rows to the requesting end user, superuser bypassing. The
ownership rule is extracted into one shared _end_user_matches used by both the single-job
guard and the list, so the two can't drift.

I1 (CI red): create_job read graph.end_user_id directly, but the warm-run path can hand
back a lightweight graph stand-in without the attribute -> 500. Use getattr(..., None),
matching every other end_user_id read.

I2 (CI red): _caller_owns_job_end_user did (job.job_metadata or {}).get(...), which on a
non-dict/legacy/mocked value either silently denied or raised. Coerce to {} unless it is a
real dict, so a security guard decides from a validated shape.

I3: the webhook runs in a fire-and-forget task that never raises, so serving_end_user_required
could never surface its 401 there — a required-but-absent identity got a 202 and was silently
dropped. Resolve identity synchronously in webhook_run_flow before scheduling.

I4: the end-user id is PII (outbound MCP forwarding is allowlist-gated for this reason), but
telemetry sent it to third-party tracing providers unconditionally. Gate it behind a new
serving_trace_end_user setting, default off (fail-closed), documented.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(serving): scope reattach /events to the end user (self-review)

Self-review of the PR found the 5th jobs endpoint, GET /workflows/{job_id}/events, still
gated on SID ownership only (via service.status) — a different end user sharing the service
account, with a known job id, could reattach to another's live event stream. Apply the same
_caller_owns_job_end_user gate before the SSE stream opens, mirroring status/stop/resume.

All five workflow-router jobs endpoints (status, stop, pending, resume, events) are now
uniformly scoped through the one shared _end_user_matches rule.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(serving): suppress superuser bypass on jobs endpoints when serving is on

On the serving plane the shared service account (SID) is itself a superuser, so
the unconditional `if current_user.is_superuser: return True` in the jobs
end-user gate fired on every gateway request — turning the whole end-user
isolation layer into a no-op for the exact deployment it exists for. Any end
user (all riding the one SID) could read, stop, resume, or reattach to another
end user's run, and /pending enumerated every end user's suspended HITL job.

Gate the bypass on the feature being OFF: new serving_end_user_enabled() reports
the deployment-level switch (header configured), independent of whether a given
request carries the header — so an anonymous serving request stays scoped too.
When serving is on, _caller_owns_job_end_user and list_pending_workflows drop the
superuser shortcut and fall through to the end-user match. Feature off (editor
plane) keeps the admin/editor bypass unchanged.

test_superuser_always_authorized encoded the old behavior; it is replaced by
test_superuser_bypass_suppressed_when_serving_on (the SID as "bob" cannot reach
alice's job, can reach its own) and test_superuser_bypass_holds_when_feature_off.

* fix(serving): scope Memory Base cross-session recall to the end user

On the serving plane one service account (SID) owns a single Memory Base store
that holds every end user's chunks, separated only by the session-id metadata
(the <end_user>::<base> prefix). Retrieval with filter_by_session on is scoped by
that prefix, but turning it off ("cross-conversation recall") drops the session
predicate entirely — the only end-user discriminator — so a similarity search
runs unfiltered across all end users' chunks. alice's query returns bob's memory.

Stamp the end-user id as its own vector-store metadata field at ingestion and, on
cross-session recall, filter on it by exact match. No DB migration (vector-store
metadata is schemaless).

The id is recovered from the already-scoped session_id via one shared lfx helper,
end_user_id_from_scoped_session, used at BOTH ingest and retrieve so the stamped
owner and the query predicate are identical by construction. Deriving from the
session id (not threading graph.end_user_id) is what makes it uniform across every
ingestion path — the regenerate/reconcile and manual-trigger paths have no graph,
only the session id, so a threaded value would leave their chunks unstamped and
invisible to their own end user under the off-toggle.

- end_user_id_from_scoped_session: leading segment of <end_user>::<base>; None when
  the feature is off, the session is anonymous (anon::), or unprefixed — so editor /
  feature-off sessions stamp and filter nothing (byte-for-byte unchanged). Gated on
  serving_end_user_enabled() so a feature-off session that happens to contain "::" is
  never parsed.
- build_documents_from_messages / build_preprocessed_document take end_user_id and
  stamp it only when present (absent -> no key, so the exact-match filter never
  falsely excludes off/editor/anonymous chunks). ingest_memory_task derives it once.
- _build_where_clause: filter on -> session predicate (already end-user scoped);
  filter off + end_user_id -> end_user_id predicate; else no predicate (unchanged).
- retrieve_memory fail-closes: serving on + cross-session recall + anonymous caller
  returns empty rather than run an unfiltered search over the shared store.

BC: pre-fix chunks carry no end_user_id key, so the exact-match filter excludes them
under the off-toggle — the safe direction (miss, never leak), and a non-issue on a
fresh serving store. Feature off / editor: no stamp, no filter, no fail-close.

Tests: helper matrix (identified / anonymous / unprefixed / first-separator / feature
off); both doc builders stamp-when-present / no-key-when-absent; where-clause on /
off-with-user / off-without-user; retrieve fail-closed for anonymous cross-session
recall and not-blocked when the feature is off.

* fix(serving): surface required-identity 401 on all doors + stamp HITL card owner

Two QA findings from executing the end-user identity one-pager, both unmet
acceptance criteria of this PR (not regressions).

BUG-01 — required mode returned HTTP 200 instead of 401 on three serving doors
(`/v1/run?stream=true`, `/v1/responses` sync + stream). The gate still BLOCKED
execution (nothing ran, nothing persisted), but reported the rejection as a
success-shaped body: streaming doors return their response before the run
executes, and create_response's blanket `except Exception` converted the
identity 401 into an OpenAIErrorResponse at the route's default 200. An
OpenAI-compatible client never raises on a 200, and status-code alerting/retry
goes blind in required mode. Fix: the same synchronous resolve_serving_scope
pre-check the webhook already uses (I3), added ahead of the stream branch in
_run_flow_internal and ahead of the run in create_response, so a required-but-
absent identity is a real 401 with the shared end_user_required_detail() body.
Idempotent with the scope simple_run_flow applies again; feature off / identity
present -> no-op.

BUG-02 — the Human-input card persisted with user_id = NULL, so the new indexed
per-user pull (`GET /monitor/messages?end_user_id=...`) missed it even for its
own owner (the predicate is an exact owner match, so NULL never returns). The
card is built by hand in persist_human_input_card and calls astore_message
directly, bypassing Component._store_message and the shared owner resolver. Fix:
recover the end user from the scoped session_id (end_user_id_from_scoped_session)
and derive the owner (derive_message_owner_uuid) — the same helpers the component
write path and the monitor read path use — and pass it as astore_message's
user_id so write == read. Off / editor / anonymous -> NULL, unchanged.

No schema change, no new settings; both paths are strict no-ops when the feature
is off. Tests: required-mode 401 on streaming /run and /responses (sync+stream)
plus identity-present still 200; HITL card retrievable by its owner via the
per-user filter and NULL-owner preserved when the feature is off.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* fix(serving): namespace SaveToFile writes by end-user identity

SaveToFile previously namespaced every write under the service-account
id (SID), so on the serving plane all end users shared one folder. Derive
a filesystem-safe segment from graph.end_user_id and use it for both the
local storage scope and the S3 key prefix; None off / editor / anonymous
so the destination falls back to the SID scope exactly as before (BC).

* fix(serving): scope background supersede to the run's session

A new background submit cancelled every SUSPENDED run of the flow for the
same user_id, ignoring session. Under the single service-account model that
crossed end users — alice's rerun cancelled bob's suspended HITL pause. Scope
the supersede to the submitting run's effective session (session_id or flow
id), matching the runner/status normalization, so only the same conversation's
stale pause is replaced. Running jobs and other sessions are untouched.

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>