diff --git a/src/backend/tests/unit/api/v1/test_authz_admin_routes.py b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py
new file mode 100644
index 0000000000..8e2bfcaae5
--- /dev/null
+++ b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py
@@ -0,0 +1,510 @@
+"""Route-level tests for the new authz admin endpoints.
+
+Covers ``/authz/roles``, ``/authz/role-assignments``, ``/authz/teams`` (+ members),
+and ``/authz/me/permissions``. Focuses on the security floor (superuser gate),
+input validation, conflict handling, and cycle detection — paths that protect
+ops invariants and would silently break without coverage.
+
+Pattern follows ``test_authz_share_routes.py`` — a fake async session + a stub
+authz service installed via monkeypatch.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy.exc import IntegrityError
+
+# --- shared fakes ----------------------------------------------------- #
+
+
+class _FakeAsyncSession:
+    """Minimal async-session stand-in that records writes + commits."""
+
+    def __init__(
+        self,
+        get_by_type: dict[tuple[type, UUID], Any] | None = None,
+        *,
+        exec_results: list[Any] | None = None,
+        commit_raises: Exception | None = None,
+    ) -> None:
+        self._get_by_type = get_by_type or {}
+        self._exec_results = exec_results or []
+        self._commit_raises = commit_raises
+        self.added: list[Any] = []
+        self.deleted: list[Any] = []
+        self.committed = 0
+        self.rolled_back = 0
+
+    async def get(self, model: type, key: UUID) -> Any:
+        return self._get_by_type.get((model, key))
+
+    def add(self, obj: Any) -> None:
+        self.added.append(obj)
+
+    async def delete(self, obj: Any) -> None:
+        self.deleted.append(obj)
+
+    async def commit(self) -> None:
+        self.committed += 1
+        if self._commit_raises is not None:
+            raise self._commit_raises
+
+    async def rollback(self) -> None:
+        self.rolled_back += 1
+
+    async def refresh(self, _obj: Any) -> None:
+        return None
+
+    async def exec(self, _stmt: Any):
+        if not self._exec_results:
+            return _ExecResult([])
+        return _ExecResult(self._exec_results.pop(0))
+
+
+class _ExecResult:
+    """Mimics ``await session.exec(stmt)`` then ``.first()`` / ``.all()`` / iter."""
+
+    def __init__(self, rows: list[Any]):
+        self._rows = list(rows)
+
+    def first(self):
+        return self._rows[0] if self._rows else None
+
+    def all(self):
+        return list(self._rows)
+
+    def __iter__(self):
+        return iter(self._rows)
+
+
+class _StubAuthz:
+    def __init__(self, *, allow: bool = True) -> None:
+        self._allow = allow
+        self.invalidate_user_calls: list[UUID] = []
+        self.invalidate_role_calls: list[UUID] = []
+        self.invalidate_all_calls = 0
+        self.effective_perms_payload: dict[UUID, list[str]] | None = None
+
+    async def supports_cross_user_fetch(self) -> bool:
+        return False
+
+    async def is_enabled(self) -> bool:
+        return False
+
+    async def enforce(self, **_kwargs) -> bool:
+        return self._allow
+
+    async def batch_enforce(self, **kwargs) -> list[bool]:
+        return [self._allow] * len(kwargs.get("requests", []))
+
+    async def invalidate_user(self, user_id: UUID) -> None:
+        self.invalidate_user_calls.append(user_id)
+
+    async def invalidate_role(self, role_id: UUID) -> None:
+        self.invalidate_role_calls.append(role_id)
+
+    async def invalidate_all(self) -> None:
+        self.invalidate_all_calls += 1
+
+    async def get_effective_permissions(self, **_kwargs) -> dict[UUID, list[str]]:
+        return self.effective_perms_payload or {}
+
+
+def _make_user(*, is_superuser: bool = False) -> SimpleNamespace:
+    return SimpleNamespace(id=uuid4(), is_superuser=is_superuser, username="u")
+
+
+@pytest.fixture
+def stub_authz(monkeypatch):
+    from langflow.api.v1 import authz_me, authz_role_assignments, authz_roles, authz_teams
+
+    def _apply(*, allow: bool = True) -> _StubAuthz:
+        stub = _StubAuthz(allow=allow)
+        for module in (authz_roles, authz_role_assignments, authz_teams, authz_me):
+            monkeypatch.setattr(module, "get_authorization_service", lambda s=stub: s)
+        return stub
+
+    return _apply
+
+
+# =====================================================================
+# /authz/roles
+# =====================================================================
+
+
+@pytest.mark.asyncio
+async def test_create_role_requires_superuser(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.api.v1.schemas.authz_roles import RoleCreate
+
+    stub_authz()
+    session = _FakeAsyncSession()
+    user = _make_user(is_superuser=False)
+    payload = RoleCreate(name="custom", description=None, permissions=["flow:*:read"])
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.create_role(payload=payload, current_user=user, session=session)
+    assert excinfo.value.status_code == 403
+    assert session.added == []
+    assert session.committed == 0
+
+
+@pytest.mark.asyncio
+async def test_create_role_persists_and_invalidates(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.api.v1.schemas.authz_roles import RoleCreate
+    from langflow.services.database.models.auth import AuthzRole  # noqa: F401 — keeps import path live
+
+    authz = stub_authz()
+    session = _FakeAsyncSession()
+    user = _make_user(is_superuser=True)
+    payload = RoleCreate(name="runner", description="x", permissions=["flow:*:execute"])
+
+    result = await authz_roles.create_role(payload=payload, current_user=user, session=session)
+    assert result.name == "runner"
+    assert result.is_system is False
+    assert len(session.added) == 1
+    assert session.committed == 1
+    assert authz.invalidate_all_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_create_role_409_on_name_conflict(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.api.v1.schemas.authz_roles import RoleCreate
+
+    stub_authz()
+    session = _FakeAsyncSession(commit_raises=IntegrityError("dup", {}, Exception()))
+    user = _make_user(is_superuser=True)
+    payload = RoleCreate(name="viewer", permissions=[])
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.create_role(payload=payload, current_user=user, session=session)
+    assert excinfo.value.status_code == 409
+    assert "already exists" in excinfo.value.detail
+    assert session.rolled_back == 1
+
+
+@pytest.mark.asyncio
+async def test_update_role_blocks_system_role(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.api.v1.schemas.authz_roles import RoleUpdate
+    from langflow.services.database.models.auth import AuthzRole
+
+    stub_authz()
+    role_id = uuid4()
+    system_role = SimpleNamespace(
+        id=role_id,
+        name="viewer",
+        description=None,
+        is_system=True,
+        permissions=[],
+        parent_role_id=None,
+    )
+    session = _FakeAsyncSession({(AuthzRole, role_id): system_role})
+    user = _make_user(is_superuser=True)
+    payload = RoleUpdate(name="hacked")
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.update_role(
+            role_id=role_id,
+            payload=payload,
+            current_user=user,
+            session=session,
+        )
+    assert excinfo.value.status_code == 400
+    assert "System roles" in excinfo.value.detail
+
+
+@pytest.mark.asyncio
+async def test_update_role_rejects_self_parent(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.api.v1.schemas.authz_roles import RoleUpdate
+    from langflow.services.database.models.auth import AuthzRole
+
+    stub_authz()
+    role_id = uuid4()
+    role = SimpleNamespace(
+        id=role_id,
+        name="custom",
+        description=None,
+        is_system=False,
+        permissions=[],
+        parent_role_id=None,
+    )
+    session = _FakeAsyncSession({(AuthzRole, role_id): role})
+    user = _make_user(is_superuser=True)
+    payload = RoleUpdate(parent_role_id=role_id)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.update_role(
+            role_id=role_id,
+            payload=payload,
+            current_user=user,
+            session=session,
+        )
+    assert excinfo.value.status_code == 400
+    assert "cannot be its own parent" in excinfo.value.detail
+
+
+@pytest.mark.asyncio
+async def test_delete_role_409_when_assigned(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.services.database.models.auth import AuthzRole, AuthzRoleAssignment  # noqa: F401
+
+    stub_authz()
+    role_id = uuid4()
+    role = SimpleNamespace(id=role_id, is_system=False)
+    # First exec is the "is anyone assigned?" probe — return a fake assignment.
+    fake_assignment = SimpleNamespace(id=uuid4())
+    session = _FakeAsyncSession(
+        {(AuthzRole, role_id): role},
+        exec_results=[[fake_assignment]],
+    )
+    user = _make_user(is_superuser=True)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.delete_role(role_id=role_id, current_user=user, session=session)
+    assert excinfo.value.status_code == 409
+    assert "active assignments" in excinfo.value.detail
+    assert session.deleted == []
+
+
+@pytest.mark.asyncio
+async def test_delete_role_blocks_system_role(stub_authz):
+    from langflow.api.v1 import authz_roles
+    from langflow.services.database.models.auth import AuthzRole
+
+    stub_authz()
+    role_id = uuid4()
+    role = SimpleNamespace(id=role_id, is_system=True)
+    session = _FakeAsyncSession({(AuthzRole, role_id): role})
+    user = _make_user(is_superuser=True)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_roles.delete_role(role_id=role_id, current_user=user, session=session)
+    assert excinfo.value.status_code == 400
+    assert "System roles" in excinfo.value.detail
+
+
+# =====================================================================
+# /authz/role-assignments
+# =====================================================================
+
+
+@pytest.mark.asyncio
+async def test_list_assignments_non_superuser_blocked_from_other_user(stub_authz):
+    from langflow.api.v1 import authz_role_assignments
+
+    stub_authz()
+    session = _FakeAsyncSession()
+    user = _make_user(is_superuser=False)
+    other_user_id = uuid4()
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_role_assignments.list_assignments(
+            session=session,
+            current_user=user,
+            user_id=other_user_id,
+        )
+    assert excinfo.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_list_assignments_self_allowed_for_non_superuser(stub_authz):
+    from langflow.api.v1 import authz_role_assignments
+
+    stub_authz()
+    session = _FakeAsyncSession(exec_results=[[]])  # empty list
+    user = _make_user(is_superuser=False)
+
+    result = await authz_role_assignments.list_assignments(
+        session=session,
+        current_user=user,
+        user_id=user.id,
+    )
+    assert result == []
+
+
+@pytest.mark.asyncio
+async def test_create_assignment_invalid_user_404(stub_authz):
+    from langflow.api.v1 import authz_role_assignments
+    from langflow.api.v1.schemas.authz_role_assignments import RoleAssignmentCreate
+
+    stub_authz()
+    session = _FakeAsyncSession()  # get() returns None
+    user = _make_user(is_superuser=True)
+    payload = RoleAssignmentCreate(user_id=uuid4(), role_id=uuid4())
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_role_assignments.create_assignment(
+            payload=payload,
+            current_user=user,
+            session=session,
+        )
+    assert excinfo.value.status_code == 404
+    assert "user_id" in excinfo.value.detail
+
+
+@pytest.mark.asyncio
+async def test_create_assignment_invokes_invalidate_user(stub_authz):
+    from langflow.api.v1 import authz_role_assignments
+    from langflow.api.v1.schemas.authz_role_assignments import RoleAssignmentCreate
+    from langflow.services.database.models.auth import AuthzRole
+    from langflow.services.database.models.user.model import User
+
+    authz = stub_authz()
+    target_user = SimpleNamespace(id=uuid4())
+    role = SimpleNamespace(id=uuid4(), name="viewer")
+    session = _FakeAsyncSession(
+        {(User, target_user.id): target_user, (AuthzRole, role.id): role},
+    )
+    actor = _make_user(is_superuser=True)
+    payload = RoleAssignmentCreate(user_id=target_user.id, role_id=role.id)
+
+    await authz_role_assignments.create_assignment(
+        payload=payload,
+        current_user=actor,
+        session=session,
+    )
+    assert len(session.added) == 1
+    assert session.committed == 1
+    assert authz.invalidate_user_calls == [target_user.id]
+
+
+# =====================================================================
+# /authz/teams
+# =====================================================================
+
+
+@pytest.mark.asyncio
+async def test_create_team_requires_superuser(stub_authz):
+    from langflow.api.v1 import authz_teams
+    from langflow.api.v1.schemas.authz_teams import TeamCreate
+
+    stub_authz()
+    session = _FakeAsyncSession()
+    user = _make_user(is_superuser=False)
+    payload = TeamCreate(team_name="Eng", adom_name="eng")
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_teams.create_team(payload=payload, current_user=user, session=session)
+    assert excinfo.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_add_member_invalidates_target_user(stub_authz):
+    from langflow.api.v1 import authz_teams
+    from langflow.api.v1.schemas.authz_teams import TeamMemberCreate
+    from langflow.services.database.models.auth import AuthzTeam
+    from langflow.services.database.models.user.model import User
+
+    authz = stub_authz()
+    team = SimpleNamespace(id=uuid4(), team_name="Eng")
+    target_user = SimpleNamespace(id=uuid4())
+    session = _FakeAsyncSession(
+        {(AuthzTeam, team.id): team, (User, target_user.id): target_user},
+    )
+    actor = _make_user(is_superuser=True)
+    payload = TeamMemberCreate(user_id=target_user.id)
+
+    await authz_teams.add_member(
+        team_id=team.id,
+        payload=payload,
+        current_user=actor,
+        session=session,
+    )
+    assert len(session.added) == 1
+    assert authz.invalidate_user_calls == [target_user.id]
+
+
+@pytest.mark.asyncio
+async def test_add_member_duplicate_returns_409(stub_authz):
+    from langflow.api.v1 import authz_teams
+    from langflow.api.v1.schemas.authz_teams import TeamMemberCreate
+    from langflow.services.database.models.auth import AuthzTeam
+    from langflow.services.database.models.user.model import User
+
+    stub_authz()
+    team = SimpleNamespace(id=uuid4(), team_name="Eng")
+    target_user = SimpleNamespace(id=uuid4())
+    session = _FakeAsyncSession(
+        {(AuthzTeam, team.id): team, (User, target_user.id): target_user},
+        commit_raises=IntegrityError("dup", {}, Exception()),
+    )
+    actor = _make_user(is_superuser=True)
+    payload = TeamMemberCreate(user_id=target_user.id)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_teams.add_member(
+            team_id=team.id,
+            payload=payload,
+            current_user=actor,
+            session=session,
+        )
+    assert excinfo.value.status_code == 409
+    assert "already a member" in excinfo.value.detail
+
+
+# =====================================================================
+# /authz/me/permissions
+# =====================================================================
+
+
+@pytest.mark.asyncio
+async def test_me_permissions_returns_per_resource_actions(stub_authz):
+    from langflow.api.v1 import authz_me
+    from langflow.api.v1.authz_me import EffectivePermissionsRequest
+
+    authz = stub_authz()
+    resource_ids = [uuid4(), uuid4()]
+    authz.effective_perms_payload = {
+        resource_ids[0]: ["read", "execute"],
+        resource_ids[1]: ["read", "write", "execute", "delete"],
+    }
+    user = _make_user()
+
+    body = EffectivePermissionsRequest(
+        resource_type="flow",
+        resource_ids=resource_ids,
+    )
+    result = await authz_me.get_effective_permissions(body=body, current_user=user)
+    assert result.resource_type == "flow"
+    assert set(result.permissions[resource_ids[0]]) == {"read", "execute"}
+    assert "delete" in result.permissions[resource_ids[1]]
+
+
+@pytest.mark.asyncio
+async def test_me_permissions_caps_resource_ids_at_500(stub_authz):
+    from langflow.api.v1 import authz_me
+    from langflow.api.v1.authz_me import EffectivePermissionsRequest
+
+    stub_authz()
+    user = _make_user()
+
+    body = EffectivePermissionsRequest(
+        resource_type="flow",
+        resource_ids=[uuid4() for _ in range(501)],
+    )
+    with pytest.raises(HTTPException) as excinfo:
+        await authz_me.get_effective_permissions(body=body, current_user=user)
+    assert excinfo.value.status_code == 400
+    assert "capped at 500" in excinfo.value.detail
+
+
+@pytest.mark.asyncio
+async def test_me_permissions_empty_request_returns_empty(stub_authz):
+    from langflow.api.v1 import authz_me
+    from langflow.api.v1.authz_me import EffectivePermissionsRequest
+
+    stub_authz()
+    user = _make_user()
+
+    body = EffectivePermissionsRequest(resource_type="flow", resource_ids=[])
+    result = await authz_me.get_effective_permissions(body=body, current_user=user)
+    assert result.permissions == {}
diff --git a/src/backend/tests/unit/api/v1/test_chat_build_flow_authz.py b/src/backend/tests/unit/api/v1/test_chat_build_flow_authz.py
new file mode 100644
index 0000000000..1158cddd0b
--- /dev/null
+++ b/src/backend/tests/unit/api/v1/test_chat_build_flow_authz.py
@@ -0,0 +1,210 @@
+"""Route-level tests for the share-aware load + deny->404 path in ``build_flow``.
+
+The handler was historically owner-OR-public scoped, which silently blocked
+shared-flow execution by non-owners even when the registered authorization
+plugin would have allowed it. These tests pin the new behavior:
+
+* Non-owner request loaded via the share-aware helper (plugin decides)
+* Plugin deny translated to 404 so callers can't enumerate UUIDs via 403 vs 404
+* PUBLIC fallback still works when the share-aware load returns None
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from typing import Any
+from uuid import UUID, uuid4
+
+import pytest
+from fastapi import HTTPException
+
+
+class _FakeSession:
+    """Minimal async session that returns canned exec results."""
+
+    def __init__(self, exec_results: list[Any] | None = None) -> None:
+        self._exec_results = exec_results or []
+
+    async def exec(self, _stmt):
+        rows = self._exec_results.pop(0) if self._exec_results else []
+        return _ExecResult(rows)
+
+
+class _ExecResult:
+    def __init__(self, rows: list[Any]):
+        self._rows = list(rows)
+
+    def first(self):
+        return self._rows[0] if self._rows else None
+
+
+def _make_user(*, is_superuser: bool = False) -> SimpleNamespace:
+    return SimpleNamespace(id=uuid4(), is_superuser=is_superuser, username="u")
+
+
+def _make_flow(*, owner_id: UUID, public: bool = False):
+    """Build a flow stub with the attributes build_flow reads."""
+    from langflow.services.database.models.flow.model import AccessTypeEnum
+
+    return SimpleNamespace(
+        id=uuid4(),
+        user_id=owner_id,
+        workspace_id=None,
+        folder_id=None,
+        data=None,
+        access_type=AccessTypeEnum.PUBLIC if public else AccessTypeEnum.PRIVATE,
+    )
+
+
+@pytest.fixture
+def patch_build_flow(monkeypatch):
+    """Install fakes for session_scope, _read_flow, ensure_flow_permission, start_flow_build."""
+    from langflow.api.v1 import chat as chat_module
+
+    state: dict[str, Any] = {"session_exec": [], "read_flow": None, "ensure_raises": None}
+
+    @asynccontextmanager
+    async def fake_session_scope():
+        yield _FakeSession(state["session_exec"])
+
+    async def fake_read_flow(_session, _flow_id, _user_id):
+        return state["read_flow"]
+
+    async def fake_ensure(*_args, **_kwargs):
+        if state["ensure_raises"] is not None:
+            raise state["ensure_raises"]
+
+    async def fake_start_build(**_kwargs):
+        return "fake-job-id"
+
+    monkeypatch.setattr(chat_module, "session_scope", fake_session_scope)
+
+    # _read_flow is imported lazily inside build_flow; patch the helper module.
+    from langflow.api.v1 import flows_helpers
+
+    monkeypatch.setattr(flows_helpers, "_read_flow", fake_read_flow)
+    monkeypatch.setattr(chat_module, "ensure_flow_permission", fake_ensure)
+    monkeypatch.setattr(chat_module, "start_flow_build", fake_start_build)
+
+    # validate_flow_for_current_settings is called when flow.data exists; we
+    # leave flow.data=None so it's not invoked. queue_service.register_job_owner
+    # is awaited at the end — provide a stub queue service in the call.
+
+    return state
+
+
+def _make_queue_service():
+    """Stub the queue service object passed via Depends in the real route."""
+
+    async def register(_job_id, _user_id):
+        return None
+
+    return SimpleNamespace(register_job_owner=register)
+
+
+# --------------------------------------------------------------------- #
+
+
+@pytest.mark.asyncio
+async def test_build_flow_owner_succeeds(patch_build_flow):
+    """Owner can build their own private flow — the historical happy path."""
+    from langflow.api.v1 import chat as chat_module
+
+    owner = _make_user()
+    flow = _make_flow(owner_id=owner.id, public=False)
+    patch_build_flow["read_flow"] = flow
+
+    result = await chat_module.build_flow(
+        flow_id=flow.id,
+        background_tasks=None,
+        current_user=owner,
+        queue_service=_make_queue_service(),
+    )
+    assert result == {"job_id": "fake-job-id"}
+
+
+@pytest.mark.asyncio
+async def test_build_flow_plugin_deny_returns_404_not_403(patch_build_flow):
+    """ensure_flow_permission raising 403 must surface as 404 (UUID privacy)."""
+    from langflow.api.v1 import chat as chat_module
+
+    user = _make_user()
+    # _read_flow finds the flow (cross-user-fetch enabled in plugin would do this)
+    flow = _make_flow(owner_id=uuid4(), public=False)  # owned by someone else
+    patch_build_flow["read_flow"] = flow
+    patch_build_flow["ensure_raises"] = HTTPException(status_code=403, detail="forbidden")
+
+    with pytest.raises(HTTPException) as excinfo:
+        await chat_module.build_flow(
+            flow_id=flow.id,
+            background_tasks=None,
+            current_user=user,
+            queue_service=_make_queue_service(),
+        )
+    # Must be 404 — not 403 — so callers can't probe for resource existence.
+    assert excinfo.value.status_code == 404
+    assert f"Flow with id {flow.id} not found" in excinfo.value.detail
+
+
+@pytest.mark.asyncio
+async def test_build_flow_unknown_flow_returns_404(patch_build_flow):
+    """Both share-aware and PUBLIC fallback miss → 404 before plugin is consulted."""
+    from langflow.api.v1 import chat as chat_module
+
+    user = _make_user()
+    flow_id = uuid4()
+    patch_build_flow["read_flow"] = None
+    patch_build_flow["session_exec"] = [[]]  # PUBLIC fallback also empty
+
+    with pytest.raises(HTTPException) as excinfo:
+        await chat_module.build_flow(
+            flow_id=flow_id,
+            background_tasks=None,
+            current_user=user,
+            queue_service=_make_queue_service(),
+        )
+    assert excinfo.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_build_flow_public_fallback_when_share_aware_misses(patch_build_flow):
+    """Plugin can't see the flow (returns None), but PUBLIC fallback finds it."""
+    from langflow.api.v1 import chat as chat_module
+
+    user = _make_user()
+    owner_id = uuid4()
+    public_flow = _make_flow(owner_id=owner_id, public=True)
+    patch_build_flow["read_flow"] = None  # share-aware load misses
+    patch_build_flow["session_exec"] = [[public_flow]]  # PUBLIC query hits
+
+    result = await chat_module.build_flow(
+        flow_id=public_flow.id,
+        background_tasks=None,
+        current_user=user,
+        queue_service=_make_queue_service(),
+    )
+    assert result == {"job_id": "fake-job-id"}
+
+
+@pytest.mark.asyncio
+async def test_build_flow_non_403_exception_not_converted(patch_build_flow):
+    """A non-403 exception from ensure_flow_permission (e.g. 500) must pass through."""
+    from langflow.api.v1 import chat as chat_module
+
+    user = _make_user()
+    flow = _make_flow(owner_id=user.id, public=False)
+    patch_build_flow["read_flow"] = flow
+    # 500 from upstream — deny_to_404 only rewrites 403.
+    patch_build_flow["ensure_raises"] = HTTPException(status_code=500, detail="upstream")
+
+    with pytest.raises(HTTPException) as excinfo:
+        await chat_module.build_flow(
+            flow_id=flow.id,
+            background_tasks=None,
+            current_user=user,
+            queue_service=_make_queue_service(),
+        )
+    # deny_to_404 only converts 403; other status codes preserved (though detail
+    # is sanitized to the supplied default).
+    assert excinfo.value.status_code == 500