diff --git a/tests/test_coordinator_end_to_end.py b/tests/test_coordinator_end_to_end.py
index ac1ee34b..6585b40a 100644
--- a/tests/test_coordinator_end_to_end.py
+++ b/tests/test_coordinator_end_to_end.py
@@ -178,12 +178,13 @@ def test_create_list_detail_lifecycle(tmp_path):
     ids = {c["ws_id"] for c in coordinators}
     assert ws_id in ids
 
-    # Coordinator created by a different user is invisible to our caller.
+    # Trusted-team visibility: every ``admin.coordinator`` caller sees
+    # every active coordinator regardless of owner.
     mgr.create(user_id="other-user", name="not-mine")
     resp = client.get("/v1/api/coordinator", headers=_COORD_HEADERS)
     assert resp.status_code == 200
     names = {c["name"] for c in resp.json()["coordinators"]}
-    assert "not-mine" not in names
+    assert "not-mine" in names
 
     # --- Detail ---
     resp = client.get(f"/v1/api/coordinator/{ws_id}", headers=_COORD_HEADERS)
@@ -431,12 +432,14 @@ def test_lazy_rehydration_on_detail_get(tmp_path):
     # The endpoint triggers lazy rehydration — manager now tracks it.
     assert mgr.get("persisted-coord") is not None
 
-    # Non-owner cannot reach the same endpoint (returns 404 — no existence leak).
+    # Trusted-team visibility: any admin.coordinator caller can read
+    # the coordinator's detail, regardless of ``user_id``.
     resp_stranger = client.get(
         "/v1/api/coordinator/persisted-coord",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp_stranger.status_code == 404
+    assert resp_stranger.status_code == 200
+    assert resp_stranger.json()["user_id"] == "user-1"
 
     # A workstream with kind='interactive' is not reachable via the coordinator
     # endpoint even when it exists in storage.
diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py
index c8af9444..99c6c5da 100644
--- a/tests/test_coordinator_endpoints.py
+++ b/tests/test_coordinator_endpoints.py
@@ -260,7 +260,10 @@ def test_create_returns_ws_id_and_records_audit(storage):
     assert "coordinator.create" in actions
 
 
-def test_list_filters_by_caller(storage):
+def test_list_returns_cluster_wide(storage):
+    # Trusted-team visibility: any caller with admin.coordinator sees
+    # every active coordinator regardless of owner.  ``user_id`` stays
+    # on the response as metadata.
     mgr = _build_mgr(storage)
     mgr.create(user_id="user-1", name="mine")
     mgr.create(user_id="user-2", name="theirs")
@@ -269,23 +272,7 @@ def test_list_filters_by_caller(storage):
     assert resp.status_code == 200
     body = resp.json()
     names = {c["name"] for c in body["coordinators"]}
-    assert names == {"mine"}
-
-
-def test_list_admin_sees_all(storage):
-    mgr = _build_mgr(storage)
-    mgr.create(user_id="user-1", name="mine")
-    mgr.create(user_id="user-2", name="theirs")
-    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.get(
-        "/v1/api/coordinator",
-        headers={
-            "X-Test-User": "admin-1",
-            "X-Test-Perms": "admin.coordinator,admin.users",
-        },
-    )
-    assert resp.status_code == 200
-    assert len(resp.json()["coordinators"]) == 2
+    assert names == {"mine", "theirs"}
 
 
 def _seed_closed_coord_with_history(
@@ -329,53 +316,19 @@ def saved_storage(tmp_path):
         reset_storage()
 
 
-def test_saved_filters_by_caller(saved_storage):
-    storage = saved_storage
-    mgr = _build_mgr(storage)
-    mine_id = _seed_closed_coord_with_history(mgr, storage, user_id="user-1", name="mine")
-    _seed_closed_coord_with_history(mgr, storage, user_id="user-2", name="theirs")
-    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.get("/v1/api/coordinator/saved", headers=_COORD_HEADERS)
-    assert resp.status_code == 200
-    body = resp.json()
-    assert {c["ws_id"] for c in body["coordinators"]} == {mine_id}
-
-
-def test_saved_admin_sees_all(saved_storage):
+def test_saved_returns_cluster_wide(saved_storage):
+    # Trusted-team visibility: every ``admin.coordinator`` caller sees
+    # every closed coordinator.
     storage = saved_storage
     mgr = _build_mgr(storage)
     a = _seed_closed_coord_with_history(mgr, storage, user_id="user-1", name="a")
     b = _seed_closed_coord_with_history(mgr, storage, user_id="user-2", name="b")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.get(
-        "/v1/api/coordinator/saved",
-        headers={
-            "X-Test-User": "admin-1",
-            "X-Test-Perms": "admin.coordinator,admin.users",
-        },
-    )
+    resp = client.get("/v1/api/coordinator/saved", headers=_COORD_HEADERS)
     assert resp.status_code == 200
     assert {c["ws_id"] for c in resp.json()["coordinators"]} == {a, b}
 
 
-def test_saved_blank_uid_returns_empty(saved_storage):
-    """Non-admin caller with no sub gets fail-closed empty list.
-
-    Mirrors list_saved_workstreams — empty user_id must not fall through
-    to a cluster-wide query (would leak orphan / migration rows).
-    """
-    storage = saved_storage
-    mgr = _build_mgr(storage)
-    _seed_closed_coord_with_history(mgr, storage, user_id="someone", name="x")
-    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.get(
-        "/v1/api/coordinator/saved",
-        headers={"X-Test-User": "", "X-Test-Perms": "admin.coordinator"},
-    )
-    assert resp.status_code == 200
-    assert resp.json() == {"coordinators": []}
-
-
 def test_saved_excludes_currently_loaded(saved_storage):
     """A coordinator currently in coord_mgr must NOT appear in saved cards.
 
@@ -426,7 +379,10 @@ def test_saved_excludes_active_state_rows(saved_storage):
     assert saved_ids == {closed_id}
 
 
-def test_send_to_someone_elses_coord_returns_404(storage):
+def test_send_any_admin_coordinator_caller_can_send(storage):
+    # Trusted-team model: send is gated on admin.coordinator scope,
+    # not on per-row ownership.  Any caller with the scope can post
+    # to any coordinator.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner", name="theirs")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -435,7 +391,7 @@ def test_send_to_someone_elses_coord_returns_404(storage):
         json={"message": "hi"},
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp.status_code == 404  # not 403 — don't leak existence
+    assert resp.status_code == 200
 
 
 def test_send_requires_message(storage):
@@ -513,7 +469,10 @@ def test_detail_triggers_lazy_rehydration(storage):
     assert mgr.get("persisted-coord") is not None
 
 
-def test_detail_404_when_not_owned(storage):
+def test_detail_any_admin_coordinator_caller_can_open(storage):
+    # Trusted-team model: any ``admin.coordinator`` caller can rehydrate
+    # any persisted coordinator regardless of owner.  ``user_id`` stays
+    # on the response as metadata.
     mgr = _build_mgr(storage)
     storage.register_workstream("coord-x", kind="coordinator", user_id="owner")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -521,7 +480,8 @@ def test_detail_404_when_not_owned(storage):
         "/v1/api/coordinator/coord-x",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
+    assert resp.json()["user_id"] == "owner"
 
 
 def test_detail_404_when_kind_interactive(storage):
@@ -551,15 +511,19 @@ def test_history_returns_messages(storage):
     assert any(m.get("role") == "user" and m.get("content") == "hello" for m in body["messages"])
 
 
-def test_history_404_for_stranger(storage):
+def test_history_any_admin_coordinator_caller_can_read(storage):
+    # Trusted-team visibility: history is readable by any
+    # ``admin.coordinator`` caller.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner")
+    storage.save_message(ws.id, "user", "hello")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
     resp = client.get(
         f"/v1/api/coordinator/{ws.id}/history",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
+    assert resp.json()["ws_id"] == ws.id
 
 
 # ---------------------------------------------------------------------------
@@ -595,7 +559,9 @@ def test_open_returns_already_loaded_when_in_memory(storage):
     assert body.get("already_loaded") is True
 
 
-def test_open_returns_404_on_ownership_mismatch_in_memory(storage):
+def test_open_any_admin_coordinator_caller_succeeds_in_memory(storage):
+    # Trusted-team model: open is gated by admin.coordinator scope
+    # only; any authenticated caller can open any in-memory coordinator.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner", name="theirs")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -603,7 +569,8 @@ def test_open_returns_404_on_ownership_mismatch_in_memory(storage):
         f"/v1/api/coordinator/{ws.id}/open",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp.status_code == 404  # not 403 — don't leak existence
+    assert resp.status_code == 200
+    assert resp.json().get("already_loaded") is True
 
 
 def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
@@ -620,28 +587,9 @@ def test_open_rehydrates_when_not_in_memory(storage, monkeypatch):
     assert body["ws_id"] == "coord-rehy"
     assert body["name"] == "rehydrated"
     assert "already_loaded" not in body
-    mgr.open.assert_called_once_with("coord-rehy", user_id="user-1")
-
-
-def test_open_admin_uses_open_admin(storage, monkeypatch):
-    mgr = _build_mgr(storage)
-    rehydrated = MagicMock()
-    rehydrated.id = "coord-rehy"
-    rehydrated.name = "r"
-    rehydrated.user_id = "someone-else"
-    # SessionManager.open accepts (ws_id, *, user_id, admin) — admin
-    # callers pass admin=True instead of the old ``open_admin`` helper.
-    monkeypatch.setattr(mgr, "open", MagicMock(return_value=rehydrated))
-    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.post(
-        "/v1/api/coordinator/coord-rehy/open",
-        headers={
-            "X-Test-User": "admin-1",
-            "X-Test-Perms": "admin.coordinator,admin.users",
-        },
-    )
-    assert resp.status_code == 200
-    mgr.open.assert_called_once_with("coord-rehy", user_id="", admin=True)
+    # SessionManager.open takes a single positional arg now — no
+    # per-caller ownership / admin plumbing.
+    mgr.open.assert_called_once_with("coord-rehy")
 
 
 def test_open_returns_404_when_unknown_ws_id(storage, monkeypatch):
@@ -721,28 +669,16 @@ def test_children_returns_interactive_children(storage):
     assert kinds == {"interactive"}
 
 
-def test_children_ownership_404(storage):
-    mgr = _build_mgr(storage)
-    ws = mgr.create(user_id="owner")
-    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
-    resp = client.get(
-        f"/v1/api/coordinator/{ws.id}/children",
-        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
-    )
-    assert resp.status_code == 404
-
-
-def test_children_admin_bypass(storage):
+def test_children_any_admin_coordinator_caller_sees_subtree(storage):
+    # Trusted-team visibility: the children subtree is readable by any
+    # ``admin.coordinator`` caller.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner")
     _seed_child(storage, ws.id, "a" * 32)
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
     resp = client.get(
         f"/v1/api/coordinator/{ws.id}/children",
-        headers={
-            "X-Test-User": "admin-1",
-            "X-Test-Perms": "admin.coordinator,admin.users",
-        },
+        headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
     assert resp.status_code == 200
     assert len(resp.json()["items"]) == 1
@@ -804,7 +740,9 @@ def test_tasks_corrupt_envelope_returns_empty(storage):
     assert resp.json() == {"version": 1, "tasks": []}
 
 
-def test_tasks_ownership_404(storage):
+def test_tasks_any_admin_coordinator_caller_can_read(storage):
+    # Trusted-team visibility: any ``admin.coordinator`` caller can
+    # read the tasks envelope.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -812,7 +750,7 @@ def test_tasks_ownership_404(storage):
         f"/v1/api/coordinator/{ws.id}/tasks",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.coordinator"},
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
 
 
 # ---------------------------------------------------------------------------
@@ -844,7 +782,8 @@ def test_cluster_inspect_invalid_ws_id_400(storage):
     assert resp.status_code == 400
 
 
-def test_cluster_inspect_ownership_404(storage):
+def test_cluster_inspect_any_inspect_caller_sees_detail(storage):
+    # Trusted-team visibility: admin.cluster.inspect sees every row.
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="owner")
     client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
@@ -852,7 +791,8 @@ def test_cluster_inspect_ownership_404(storage):
         f"/v1/api/cluster/ws/{ws.id}/detail",
         headers={"X-Test-User": "stranger", "X-Test-Perms": "admin.cluster.inspect"},
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
+    assert resp.json()["persisted"]["ws_id"] == ws.id
 
 
 def test_cluster_inspect_coordinator_self_path(storage):
@@ -1116,41 +1056,10 @@ def test_coordinator_rows_filters_by_caller_identity(storage):
         )
         return request
 
-    alice_rows = _coordinator_rows(
-        _request_for("alice", frozenset({"read"})),
-    )
-    names = {r["name"] for r in alice_rows}
-    assert names == {"alice-coord"}, "non-admin alice must NOT see bob's coordinator"
-
-    bob_rows = _coordinator_rows(_request_for("bob", frozenset({"read"})))
-    assert {r["name"] for r in bob_rows} == {"bob-coord"}
-
-    admin_rows = _coordinator_rows(
-        _request_for("admin-1", frozenset({"read", "admin.users"})),
-    )
-    assert {r["name"] for r in admin_rows} == {"alice-coord", "bob-coord"}
-
-
-def test_coordinator_rows_empty_user_id_returns_empty(storage):
-    """Defense-in-depth: a request with no user_id (shouldn't reach this
-    path through the auth middleware, but defensive) gets zero rows,
-    not a list_all leak."""
-    from unittest.mock import MagicMock
-
-    from turnstone.console.server import _coordinator_rows
-
-    mgr = _build_mgr(storage)
-    mgr.create(user_id="alice", name="alice-coord")
-
-    request = MagicMock()
-    request.app.state.coord_mgr = mgr
-    request.state.auth_result = AuthResult(
-        user_id="",
-        scopes=frozenset({"read"}),
-        token_source="test",
-        permissions=frozenset({"read"}),
-    )
-    assert _coordinator_rows(request) == []
+    # Trusted-team visibility: every caller sees every coordinator.
+    for caller in ("alice", "bob", "admin-1"):
+        rows = _coordinator_rows(_request_for(caller, frozenset({"read"})))
+        assert {r["name"] for r in rows} == {"alice-coord", "bob-coord"}
 
 
 def _persisted_rows_request(storage, mgr, user_id: str, perms: frozenset[str]):
@@ -1232,11 +1141,10 @@ def test_coordinator_rows_dedupes_by_ws_id_in_memory_wins(storage):
     assert rows[0]["state"] == "idle"
 
 
-def test_coordinator_rows_persisted_respects_tenant_filter(storage):
-    """Non-admin callers must not see OTHER tenants' persisted
-    (closed) coordinators either.  Regression lock — the user_id
-    kwarg on list_workstreams is pushed through; the defense-in-depth
-    client-side empty-string check catches the tail."""
+def test_coordinator_rows_persisted_cluster_wide(storage):
+    # Trusted-team visibility: every caller sees every persisted row,
+    # including rows from other identities and orphan (empty-user_id)
+    # rows.  ``user_id`` stays on the response as metadata.
     from turnstone.console.server import _coordinator_rows
     from turnstone.core.workstream import WorkstreamKind
 
@@ -1259,33 +1167,6 @@ def test_coordinator_rows_persisted_respects_tenant_filter(storage):
         kind=WorkstreamKind.COORDINATOR,
         parent_ws_id=None,
     )
-
-    alice_req = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"}))
-    alice_rows = _coordinator_rows(alice_req)
-    assert {r["name"] for r in alice_rows} == {"alice-closed"}
-
-    bob_req = _persisted_rows_request(storage, mgr, "bob", frozenset({"read"}))
-    bob_rows = _coordinator_rows(bob_req)
-    assert {r["name"] for r in bob_rows} == {"bob-closed"}
-
-    admin_req = _persisted_rows_request(storage, mgr, "admin-1", frozenset({"read", "admin.users"}))
-    admin_rows = _coordinator_rows(admin_req)
-    assert {r["name"] for r in admin_rows} == {"alice-closed", "bob-closed"}
-
-
-def test_coordinator_rows_persisted_skips_orphan_rows_for_non_admin(storage):
-    """Defense-in-depth empty-string tenancy check — a persisted row
-    with empty user_id (migration-artifact / system-owned) must NOT
-    be visible to a non-admin caller with empty caller_uid either.
-    The SQL user_id filter above already enforces this, but duplicate
-    the check client-side so orphan rows never leak to any
-    hypothetical empty-sub JWT."""
-    from unittest.mock import MagicMock
-
-    from turnstone.console.server import _coordinator_rows
-    from turnstone.core.workstream import WorkstreamKind
-
-    mgr = _build_mgr(storage)
     storage.register_workstream(
         "c" * 32,
         node_id="console",
@@ -1296,14 +1177,11 @@ def test_coordinator_rows_persisted_skips_orphan_rows_for_non_admin(storage):
         parent_ws_id=None,
     )
 
-    # Non-admin with empty caller_uid must not see the orphan.
-    request = MagicMock()
-    request.app.state.coord_mgr = mgr
-    request.app.state.auth_storage = storage
-    request.state.auth_result = AuthResult(
-        user_id="",
-        scopes=frozenset({"read"}),
-        token_source="test",
-        permissions=frozenset({"read"}),
-    )
-    assert _coordinator_rows(request) == []
+    for caller, perms in (
+        ("alice", frozenset({"read"})),
+        ("bob", frozenset({"read"})),
+        ("admin-1", frozenset({"read", "admin.users"})),
+    ):
+        request = _persisted_rows_request(storage, mgr, caller, perms)
+        rows = _coordinator_rows(request)
+        assert {r["name"] for r in rows} == {"alice-closed", "bob-closed", "orphan-closed"}
diff --git a/tests/test_coordinator_governance.py b/tests/test_coordinator_governance.py
index f0184020..290713d9 100644
--- a/tests/test_coordinator_governance.py
+++ b/tests/test_coordinator_governance.py
@@ -340,7 +340,10 @@ def test_restrict_rejects_non_object_body(storage):
     assert resp.status_code == 400
 
 
-def test_trust_toggle_tenant_404_on_foreign_coord(storage):
+def test_trust_toggle_cluster_wide_access(storage):
+    # Trusted-team model: the trust toggle is gated on the scope
+    # permission, not on row-level ownership.  A caller holding
+    # ``coordinator.trust.send`` may toggle any coord's trust state.
     mgr = _build_mgr(storage)
     coord = mgr.create(user_id="user-owner", name="coord-a")
     coord.session, _ = _make_session_mock()
@@ -353,7 +356,7 @@ def test_trust_toggle_tenant_404_on_foreign_coord(storage):
             "X-Test-Perms": "admin.coordinator,coordinator.trust.send",
         },
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
 
 
 def test_trust_toggle_404_when_session_not_loaded(storage):
diff --git a/tests/test_phase6_endpoints.py b/tests/test_phase6_endpoints.py
index a11230c7..59fb70ce 100644
--- a/tests/test_phase6_endpoints.py
+++ b/tests/test_phase6_endpoints.py
@@ -230,11 +230,11 @@ def test_bulk_live_admin_bypass_returns_live(storage):
     assert body["denied"] == []
 
 
-def test_bulk_live_tenant_filter_marks_foreign_rows_denied(storage):
-    """A non-admin caller whose user_id doesn't match the row's owner
-    gets the ws_id in ``denied`` rather than ``results`` — no
-    existence-oracle leak."""
-    # Seed a foreign-owned interactive workstream.
+def test_bulk_live_cluster_wide_visibility(storage):
+    """Trusted-team visibility: any ``admin.cluster.inspect`` caller
+    sees every row in ``results``.  ``denied`` is reserved for ids
+    that don't correspond to a persisted workstream (no existence
+    oracle for unknown ids)."""
     ws_id = "b" * 32
     _seed_workstream(storage, ws_id=ws_id, node_id="node-a", user_id="stranger")
     client = _make_client(storage, coord_mgr=_build_mgr(storage))
@@ -244,22 +244,18 @@ def test_bulk_live_tenant_filter_marks_foreign_rows_denied(storage):
     )
     assert resp.status_code == 200
     body = resp.json()
-    assert body["denied"] == [ws_id]
-    assert body["results"] == {}
+    assert ws_id in body["results"]
+    assert body["denied"] == []
 
 
-def test_bulk_live_empty_caller_uid_denies_empty_owner_rows(storage):
-    """Regression for #bug-3 / #sec-2: a caller with empty user_id
-    must NOT see rows with empty user_id (orphan / system-owned).
-    Either side empty → denied.  Admin bypass honoured (tested
-    elsewhere)."""
-    ws_id = "c" * 32
-    _seed_workstream(storage, ws_id=ws_id, node_id="node-a", user_id="")
+def test_bulk_live_unknown_ids_route_to_denied(storage):
+    """Unknown ids (not in storage) land in ``denied`` so the endpoint
+    can't be used as an existence oracle."""
+    ws_id = "c" * 32  # not seeded
     client = _make_client(storage, coord_mgr=_build_mgr(storage))
-    # caller_uid="" (empty X-Test-User) + non-admin perm.
     resp = client.get(
         f"/v1/api/cluster/ws/live?ids={ws_id}",
-        headers={"X-Test-User": "", "X-Test-Perms": "admin.cluster.inspect"},
+        headers={"X-Test-User": "user-1", "X-Test-Perms": "admin.cluster.inspect"},
     )
     assert resp.status_code == 200
     body = resp.json()
@@ -315,9 +311,9 @@ def test_metrics_invalid_ws_id_400(storage):
     assert resp.status_code == 400
 
 
-def test_metrics_ownership_404_mask(storage):
-    """A ws_id owned by another tenant returns 404, not 403 — no
-    existence-oracle leak (mirrors coordinator_detail)."""
+def test_metrics_any_admin_coordinator_caller_can_read(storage):
+    """Trusted-team visibility: metrics are readable by any caller
+    with ``admin.coordinator`` regardless of the coordinator owner."""
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="stranger")
     client = _make_client(storage, coord_mgr=mgr)
@@ -325,7 +321,8 @@ def test_metrics_ownership_404_mask(storage):
         f"/v1/api/coordinator/{ws.id}/metrics",
         headers=_METRICS_HEADERS,
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
+    assert resp.json()["ws_id"] == ws.id
 
 
 def test_metrics_empty_coordinator_defaults(storage):
@@ -393,20 +390,13 @@ def test_metrics_spawns_and_state_counts(storage):
     assert body["child_state_counts"] == {"idle": 1, "running": 1, "closed": 1}
 
 
-def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
-    """Defense-in-depth: a non-admin caller's aggregate counts must
-    exclude children whose parent_ws_id matches the coord but whose
-    user_id drifted to another tenant (forged / migration-era rows).
-    The primary defense is the 404-mask on coord ownership; this is
-    the secondary defense inside the aggregate queries (Copilot
-    review finding on PR #381).
-
-    Admin bypass sees the raw aggregate (no tenant filter) — same
-    pattern coordinator_children follows.
+def test_metrics_cluster_wide_aggregates(storage):
+    """Trusted-team model: aggregates are cluster-wide across every
+    caller with ``admin.coordinator``.  Every child under the
+    coordinator counts, regardless of the ``user_id`` on the row.
     """
     mgr = _build_mgr(storage)
     ws = mgr.create(user_id="alice")
-    # Legitimate child owned by alice.
     _seed_workstream(
         storage,
         ws_id="aa" * 16,
@@ -415,7 +405,6 @@ def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
         parent_ws_id=ws.id,
         state="idle",
     )
-    # Forged / drifted child — same parent_ws_id but foreign owner.
     _seed_workstream(
         storage,
         ws_id="bb" * 16,
@@ -426,30 +415,16 @@ def test_metrics_tenant_filter_excludes_forged_cross_tenant_child(storage):
     )
     client = _make_client(storage, coord_mgr=mgr)
 
-    # Alice (non-admin) — counts must exclude bob's forged row.
-    resp = client.get(
-        f"/v1/api/coordinator/{ws.id}/metrics",
-        headers={"X-Test-User": "alice", "X-Test-Perms": "admin.coordinator"},
-    )
-    assert resp.status_code == 200
-    body = resp.json()
-    assert body["spawns_total"] == 1
-    assert body["child_state_counts"] == {"idle": 1}
-    # "running" (bob's forged child) filtered out.
-    assert "running" not in body["child_state_counts"]
-
-    # Admin sees both.
-    resp_admin = client.get(
-        f"/v1/api/coordinator/{ws.id}/metrics",
-        headers={
-            "X-Test-User": "admin-1",
-            "X-Test-Perms": "admin.coordinator,admin.users",
-        },
-    )
-    assert resp_admin.status_code == 200
-    body_admin = resp_admin.json()
-    assert body_admin["spawns_total"] == 2
-    assert body_admin["child_state_counts"] == {"idle": 1, "running": 1}
+    # Every admin.coordinator caller sees both children.
+    for caller in ("alice", "bob", "admin-1"):
+        resp = client.get(
+            f"/v1/api/coordinator/{ws.id}/metrics",
+            headers={"X-Test-User": caller, "X-Test-Perms": "admin.coordinator"},
+        )
+        assert resp.status_code == 200, caller
+        body = resp.json()
+        assert body["spawns_total"] == 2, caller
+        assert body["child_state_counts"] == {"idle": 1, "running": 1}, caller
 
 
 def test_metrics_judge_fallback_rate_substring_match(storage):
diff --git a/tests/test_server_attachments_endpoints.py b/tests/test_server_attachments_endpoints.py
index 20f45ee6..2abbc189 100644
--- a/tests/test_server_attachments_endpoints.py
+++ b/tests/test_server_attachments_endpoints.py
@@ -210,16 +210,17 @@ class TestUploadRejections:
         )
         assert resp.status_code == 404
 
-    def test_foreign_workstream_is_not_found(self, app_client):
+    def test_any_caller_can_attach_to_workstream(self, app_client):
+        # Trusted-team model: attaching to any workstream is gated on
+        # scope auth, not ownership.  The attachment is filed under
+        # the ws's persisted owner so existing storage shape holds.
         client, _ = app_client
-        # userA tries to attach to ws-B (owned by userB) — we mask this as
-        # 404 to avoid leaking workstream existence to non-owners.
         resp = client.post(
             "/v1/api/workstreams/ws-B/attachments",
             files={"file": ("x.md", b"x", "text/markdown")},
             headers=_auth("userA"),
         )
-        assert resp.status_code == 404
+        assert resp.status_code == 200
 
 
 class TestPendingCap:
@@ -298,13 +299,17 @@ class TestListAttachments:
         assert all("content" not in a for a in atts)
         assert {a["filename"] for a in atts} == {"a.md", "b.md"}
 
-    def test_list_isolated_per_user(self, app_client):
+    def test_list_visible_cluster_wide(self, app_client):
+        # Trusted-team visibility: any authenticated caller can list
+        # the attachments on any workstream.  Attachments are filed
+        # under the ws's owner uid so a cross-caller lister still sees
+        # the owner's pending uploads.
         client, _ = app_client
         _upload(client, "ws-A", "userA", "mine.md", b"mine", "text/markdown")
-        # userB can't even GET listing on ws-A (not their workstream);
-        # masked as 404 to avoid existence-leak.
         resp = client.get("/v1/api/workstreams/ws-A/attachments", headers=_auth("userB"))
-        assert resp.status_code == 404
+        assert resp.status_code == 200
+        atts = resp.json()["attachments"]
+        assert {a["filename"] for a in atts} == {"mine.md"}
 
 
 class TestGetContent:
@@ -343,15 +348,19 @@ class TestGetContent:
         assert resp.headers["content-type"].startswith("text/plain")
         assert resp.headers.get("x-content-type-options") == "nosniff"
 
-    def test_get_content_wrong_user_is_not_found(self, app_client):
+    def test_get_content_visible_cluster_wide(self, app_client):
+        # Trusted-team visibility: any authenticated caller can fetch
+        # the content of an attachment on any workstream.  Attachments
+        # are keyed by the ws's persisted owner uid so userB still
+        # resolves userA's blob via _require_ws_access's owner return.
         client, _ = app_client
         aid = _upload(client, "ws-A", "userA", "t.md", b"x", "text/markdown")
         resp = client.get(
             f"/v1/api/workstreams/ws-A/attachments/{aid}/content",
             headers=_auth("userB"),
         )
-        # 404 rather than 403 — caller can't distinguish from "ws doesn't exist".
-        assert resp.status_code == 404
+        assert resp.status_code == 200
+        assert resp.content == b"x"
 
     def test_get_content_cross_workstream_id_404(self, app_client):
         client, _ = app_client
@@ -400,12 +409,14 @@ class TestDelete:
         resp = client.delete(f"/v1/api/workstreams/ws-A/attachments/{aid}", headers=_auth("userA"))
         assert resp.status_code == 404
 
-    def test_delete_wrong_user_is_not_found(self, app_client):
+    def test_delete_cluster_wide(self, app_client):
+        # Trusted-team model: any authenticated caller can delete an
+        # attachment on any workstream.  The filed ``user_id`` stays
+        # for audit even after a cross-caller delete.
         client, _ = app_client
         aid = _upload(client, "ws-A", "userA", "t.md", b"x", "text/markdown")
         resp = client.delete(f"/v1/api/workstreams/ws-A/attachments/{aid}", headers=_auth("userB"))
-        # userB doesn't own ws-A — masked as 404 to avoid existence-leak.
-        assert resp.status_code == 404
+        assert resp.status_code == 200
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py
index 05349464..6dae9925 100644
--- a/tests/test_server_authz.py
+++ b/tests/test_server_authz.py
@@ -272,7 +272,10 @@ def _register_ws(storage: Any, ws_id: str, owner: str) -> None:
 
 
 class TestCrossTenantDelete:
-    def test_non_owner_cannot_delete(self, app_client):
+    def test_any_caller_can_delete(self, app_client):
+        # Trusted-team model: scope auth gates the endpoint, not
+        # row-level ownership.  ``user_id`` stays on audit + storage
+        # metadata.
         from turnstone.core.storage import get_storage
 
         client, _mgr = app_client
@@ -283,9 +286,7 @@ class TestCrossTenantDelete:
             "/v1/api/workstreams/ws-victim/delete",
             headers=_auth("attacker-user"),
         )
-        assert resp.status_code == 404
-        # Victim's workstream still present in storage.
-        assert storage.get_workstream("ws-victim") is not None
+        assert resp.status_code == 200
 
     def test_owner_delete_records_audit(self, app_client):
         from turnstone.core.storage import get_storage
@@ -336,7 +337,11 @@ class TestCrossTenantClose:
 
 
 class TestCrossTenantTitle:
-    def test_non_owner_cannot_refresh_title(self, app_client):
+    def test_refresh_title_requires_live_session(self, app_client):
+        # Trusted-team model: scope-level auth is the gate; any caller
+        # can hit the endpoint.  A not-currently-active workstream
+        # still 404s because the refresh needs the live session, not
+        # because of tenant mismatch.
         from turnstone.core.storage import get_storage
 
         client, _mgr = app_client
@@ -348,8 +353,13 @@ class TestCrossTenantTitle:
             headers=_auth("attacker-user"),
         )
         assert resp.status_code == 404
+        assert "not active" in resp.json().get("error", "") or "not found" in resp.json().get(
+            "error", ""
+        )
 
-    def test_non_owner_cannot_set_title(self, app_client):
+    def test_any_caller_can_set_title(self, app_client):
+        # Trusted-team model: title is editable by any authenticated
+        # caller; ``user_id`` remains metadata.
         from turnstone.core.storage import get_storage
 
         client, _mgr = app_client
@@ -358,14 +368,16 @@ class TestCrossTenantTitle:
         _register_ws(storage, "ws-victim", "victim-user")
         resp = client.post(
             "/v1/api/workstreams/ws-victim/title",
-            json={"title": "phishing title"},
+            json={"title": "updated title"},
             headers=_auth("attacker-user"),
         )
-        assert resp.status_code == 404
+        assert resp.status_code == 200
 
 
 class TestCrossTenantOpen:
-    def test_non_owner_cannot_open_persisted(self, app_client):
+    def test_any_caller_can_open_persisted(self, app_client):
+        # Trusted-team model: open is gated on scope auth, not on row
+        # ownership.  The persisted ``user_id`` stays as metadata.
         from turnstone.core.storage import get_storage
 
         client, _mgr = app_client
@@ -376,7 +388,7 @@ class TestCrossTenantOpen:
             "/v1/api/workstreams/ws-victim/open",
             headers=_auth("attacker-user"),
         )
-        assert resp.status_code == 404
+        assert resp.status_code == 200
 
 
 class TestListWorkstreamsTrustedTeamVisibility:
diff --git a/tests/test_service_auth_boundary.py b/tests/test_service_auth_boundary.py
index ebc0c24c..a7310c48 100644
--- a/tests/test_service_auth_boundary.py
+++ b/tests/test_service_auth_boundary.py
@@ -57,58 +57,13 @@ def _request_with_auth(
 
 
 # ---------------------------------------------------------------------------
-# _effective_user_filter — console edition (admin, service, uid, DENY)
+# _effective_user_filter — the console edition was deleted alongside the
+# row-level ownership gates (trusted-team unification).  Only the server
+# edition survives — it still differentiates service callers (cluster-
+# wide) from scoped users (tenant-pinned aggregates on node endpoints).
 # ---------------------------------------------------------------------------
 
 
-class TestConsoleEffectiveUserFilter:
-    def test_admin_returns_none(self):
-        from turnstone.console.server import _effective_user_filter
-
-        req = _request_with_auth(user_id="alice", permissions=frozenset({"admin.users"}))
-        assert _effective_user_filter(req) is None
-
-    def test_admin_roles_perm_also_bypasses(self):
-        from turnstone.console.server import _effective_user_filter
-
-        req = _request_with_auth(user_id="carol", permissions=frozenset({"admin.roles"}))
-        assert _effective_user_filter(req) is None
-
-    def test_service_scope_returns_none(self):
-        from turnstone.console.server import _effective_user_filter
-
-        req = _request_with_auth(user_id="svc-proxy", scopes=frozenset({"service"}))
-        assert _effective_user_filter(req) is None
-
-    def test_scoped_caller_returns_uid(self):
-        from turnstone.console.server import _effective_user_filter
-
-        req = _request_with_auth(user_id="alice", scopes=frozenset({"read"}))
-        assert _effective_user_filter(req) == "alice"
-
-    def test_blank_sub_non_service_returns_deny_sentinel(self):
-        from turnstone.console.server import DENY_EMPTY_SUB, _effective_user_filter
-
-        req = _request_with_auth(user_id="", scopes=frozenset({"read"}))
-        result = _effective_user_filter(req)
-        assert result is DENY_EMPTY_SUB, (
-            "blank-sub non-service callers must fail closed — "
-            "passing through to storage with user_id=None is a "
-            "service escape and user_id='' matches legacy orphans"
-        )
-
-    def test_deny_sentinel_is_singleton(self):
-        """Callers compare with ``is``; equality against a bare object()
-        must never match the sentinel, and two separate reads of the
-        attribute return the same instance (ruling out a property /
-        factory that would break ``is`` identity)."""
-        from turnstone.console.server import DENY_EMPTY_SUB as FIRST_READ
-        from turnstone.console.server import DENY_EMPTY_SUB as SECOND_READ
-
-        assert FIRST_READ is not object()
-        assert FIRST_READ is SECOND_READ
-
-
 # ---------------------------------------------------------------------------
 # _effective_user_filter — server edition (service, uid, DENY — no admin)
 # ---------------------------------------------------------------------------
@@ -513,15 +468,17 @@ class TestClusterEventsSseGate:
 
 
 class TestDenySentinelSharedIdentity:
-    def test_console_and_server_share_one_sentinel(self):
+    def test_core_and_server_share_one_sentinel(self):
         """The sentinel is compared with ``is``; a future refactor
         that re-introduced per-module duplicates would silently break
-        the identity check.  Lock the cross-module invariant."""
-        from turnstone.console.server import DENY_EMPTY_SUB as CONSOLE_DENY
+        the identity check.  Lock the cross-module invariant.
+
+        Only the server + core surfaces consume the sentinel after the
+        trusted-team unification — the console no longer gates on
+        row ownership, so its ``_effective_user_filter`` was removed."""
         from turnstone.core.auth import DENY_EMPTY_SUB as CORE_DENY
         from turnstone.server import DENY_EMPTY_SUB as SERVER_DENY
 
-        assert CORE_DENY is CONSOLE_DENY
         assert CORE_DENY is SERVER_DENY
 
 
diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py
index 85197b0e..55c29c15 100644
--- a/tests/test_session_manager.py
+++ b/tests/test_session_manager.py
@@ -277,9 +277,12 @@ def test_create_rolls_back_slot_on_session_failure() -> None:
     mgr, _, storage = _make_manager(adapter=adapter)
     with pytest.raises(RuntimeError, match="build_session forced failure"):
         mgr.create(user_id="u1")
-    # Slot freed, row deleted, no dangling capacity consumption.
+    # Slot freed — no dangling capacity consumption.  The storage row
+    # survives construction failure on purpose: the next ``open(ws_id)``
+    # retries build_session rather than forcing the user to create a
+    # brand-new workstream.
     assert mgr.count == 0
-    assert len(storage.rows) == 0
+    assert len(storage.rows) == 1
 
 
 def test_create_rolls_back_slot_on_persist_failure() -> None:
@@ -329,7 +332,7 @@ def test_concurrent_create_does_not_exceed_max_active() -> None:
 
 def test_open_returns_none_for_missing_row() -> None:
     mgr, _, _ = _make_manager()
-    assert mgr.open("missing", user_id="u1") is None
+    assert mgr.open("missing") is None
 
 
 def test_open_blocks_deleted_state() -> None:
@@ -338,7 +341,7 @@ def test_open_blocks_deleted_state() -> None:
     mgr.close(ws.id)
     # Flip the row to the tombstone state — open must refuse it.
     storage.rows[ws.id].state = "deleted"
-    assert mgr.open(ws.id, user_id="u1") is None
+    assert mgr.open(ws.id) is None
 
 
 def test_open_resurrects_closed_state() -> None:
@@ -348,7 +351,7 @@ def test_open_resurrects_closed_state() -> None:
     mgr.close(ws_id)
     assert mgr.get(ws_id) is None
 
-    reopened = mgr.open(ws_id, user_id="u1")
+    reopened = mgr.open(ws_id)
     assert reopened is not None
     assert reopened.id == ws_id
     assert reopened.session is not None
@@ -357,13 +360,18 @@ def test_open_resurrects_closed_state() -> None:
     assert ws_id in [e.ws_id for e in adapter.events_of("created")]
 
 
-def test_open_rejects_non_matching_user_when_not_admin() -> None:
+def test_open_ignores_owner_mismatch() -> None:
+    # Turnstone is a trusted-team tool; row-level ownership is
+    # metadata, not an access boundary.  ``open`` no longer cares
+    # who the caller is — any authenticated caller can rehydrate
+    # any persisted workstream.  Scope-level auth at the HTTP
+    # layer is the only gate.
     mgr, _, _ = _make_manager()
     ws = mgr.create(user_id="u1")
     mgr.close(ws.id)
-    assert mgr.open(ws.id, user_id="u2", admin=False) is None
-    # Admin bypass works.
-    assert mgr.open(ws.id, user_id="u2", admin=True) is not None
+    reopened = mgr.open(ws.id)
+    assert reopened is not None
+    assert reopened.user_id == "u1"  # metadata preserved
 
 
 def test_open_rejects_wrong_kind() -> None:
@@ -372,7 +380,7 @@ def test_open_rejects_wrong_kind() -> None:
     # Storage row claims a different kind than our adapter's.
     storage.rows[ws.id].kind = WorkstreamKind.COORDINATOR.value
     mgr.close(ws.id)
-    assert mgr.open(ws.id, user_id="u1") is None
+    assert mgr.open(ws.id) is None
 
 
 def test_concurrent_open_for_same_ws_id_returns_same_session() -> None:
@@ -387,7 +395,7 @@ def test_concurrent_open_for_same_ws_id_returns_same_session() -> None:
     lock = threading.Lock()
 
     def _open() -> None:
-        r = mgr.open(ws_id, user_id="u1")
+        r = mgr.open(ws_id)
         with lock:
             results.append(r)
 
diff --git a/turnstone/console/server.py b/turnstone/console/server.py
index 129706cb..3583006d 100644
--- a/turnstone/console/server.py
+++ b/turnstone/console/server.py
@@ -46,11 +46,9 @@ from turnstone.console.metrics import ConsoleMetrics
 from turnstone.console.router import ConsoleRouter
 from turnstone.core.audit import record_audit
 from turnstone.core.auth import (
-    DENY_EMPTY_SUB,
     JWT_AUD_CONSOLE,
     JWT_AUD_SERVER,
     AuthMiddleware,
-    _DenyFilter,
     create_jwt,
     jwt_version_slot,
     require_permission,
@@ -450,26 +448,15 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
     In-memory wins on ws_id conflict so live state stays authoritative
     for active sessions.
 
-    Ownership filter: non-admin callers must not see other tenants'
-    coordinator rows.  Admins (``admin.users`` / ``admin.roles``) get
-    the full set.  Unauthenticated callers shouldn't reach this path
-    — the endpoint sits behind global auth — but the filter defaults
-    to empty on a missing ``user_id`` rather than leaking.
+    Trusted-team visibility (post-#400): the cluster dashboard shows
+    every coordinator regardless of caller identity; ``user_id`` is
+    surfaced on each row as display metadata.
     """
     coord_mgr = getattr(request.app.state, "coord_mgr", None)
     if coord_mgr is None:
         return []
-    filt = _effective_user_filter(request)
-    if filt is DENY_EMPTY_SUB:
-        return []
-    # filt is now ``None`` (admin / service) or a non-empty caller uid.
-    # ``list_for_user`` lived on the old CoordinatorManager; SessionManager
-    # only exposes ``list_all`` — inline the owner filter at the call site.
     try:
-        all_wss = coord_mgr.list_all()
-        wss = (
-            all_wss if filt is None else [ws for ws in all_wss if ws.user_id and ws.user_id == filt]
-        )
+        wss = coord_mgr.list_all()
     except Exception:
         log.debug("cluster_workstreams.coord_list_failed", exc_info=True)
         return []
@@ -506,16 +493,15 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
 
     # Second lane — persisted coordinator rows, used to surface
     # closed / error / deleted coordinators the manager has already
-    # evicted from ``self._workstreams``.  ``filt`` carries the same
-    # tenant decision as the in-memory lane above (None = admin /
-    # service; str = scoped caller).
+    # evicted from ``self._workstreams``.  Cluster-wide (trusted-team
+    # visibility).
     storage = getattr(request.app.state, "auth_storage", None)
     if storage is None:
         return rows
     try:
         persisted = storage.list_workstreams(
             kind=WorkstreamKind.COORDINATOR,
-            user_id=filt,
+            user_id=None,
             limit=200,
         )
     except Exception:
@@ -534,12 +520,6 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]:
         if not row_id or row_id in seen:
             continue
         row_owner = m.get("user_id") or ""
-        # Defense-in-depth empty-string tenancy check.  The SQL user_id
-        # filter above already enforces the match when ``filt`` is set;
-        # drop any rows whose owner still lands empty (migration
-        # artifacts) for non-admin callers.
-        if filt is not None and (not row_owner or row_owner != filt):
-            continue
         rows.append(
             {
                 "id": row_id,
@@ -894,15 +874,6 @@ async def cluster_ws_detail(request: Request) -> JSONResponse:
     if row is None:
         return JSONResponse({"error": "workstream not found"}, status_code=404)
 
-    user_id = _auth_user_id(request)
-    err404 = _check_row_owner_or_404(
-        request,
-        row.get("user_id") or "",
-        user_id,
-    )
-    if err404 is not None:
-        return err404
-
     try:
         live = await _fetch_live_block(request, row, ws_id)
     except Exception:
@@ -991,14 +962,6 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
     if not cleaned:
         return JSONResponse({"results": {}, "denied": [], "truncated": False})
 
-    filt = _effective_user_filter(request)
-    if filt is DENY_EMPTY_SUB:
-        # Non-admin, non-service caller with a blank sub — every row
-        # is denied by the empty-string rule.  Skip the batch fetch
-        # entirely and route all ids to ``denied`` (no existence
-        # oracle).
-        return JSONResponse({"results": {}, "denied": cleaned, "truncated": truncated})
-
     try:
         rows = await asyncio.to_thread(storage.get_workstreams_batch, cleaned)
     except Exception:
@@ -1020,14 +983,9 @@ async def cluster_ws_live_bulk(request: Request) -> JSONResponse:
     for wid in cleaned:
         row = rows.get(wid)
         if row is None:
-            denied.append(wid)
-            continue
-        # Tenant check — admin / service (``filt is None``) sees all
-        # rows; scoped callers (``filt`` is a uid) see only matching
-        # rows.  An orphan / migration-artifact row with ``user_id=""``
-        # is denied for scoped callers by the ``not row_owner`` guard.
-        row_owner = row.get("user_id") or ""
-        if filt is not None and (not row_owner or row_owner != filt):
+            # Missing rows route to ``denied`` rather than ``results``
+            # so the endpoint can't be used as an existence oracle for
+            # ids outside the caller's knowledge.
             denied.append(wid)
             continue
         owned_rows.append((wid, row))
@@ -2368,28 +2326,6 @@ def _require_admin_coordinator(
     )
 
 
-def _check_row_owner_or_404(
-    request: Request,
-    row_owner: str,
-    user_id: str,
-    *,
-    error_msg: str = "workstream not found",
-) -> JSONResponse | None:
-    """Ownership gate with the empty-string defense.
-
-    Non-admin callers need BOTH ``user_id`` and ``row_owner`` to be
-    non-empty AND equal.  Either side being empty is a 404 — otherwise
-    an orphan / migration-artifact / system-owned row with
-    ``user_id=""`` would leak to a (hypothetical) caller whose JWT
-    carries an empty ``sub`` claim.  Admin bypass honoured.
-    """
-    if _is_admin(request):
-        return None
-    if not user_id or not row_owner or row_owner != user_id:
-        return JSONResponse({"error": error_msg}, status_code=404)
-    return None
-
-
 def _resolve_coordinator_or_404(
     request: Request,
     coord_mgr: Any,
@@ -2397,21 +2333,22 @@ def _resolve_coordinator_or_404(
     ws_id: str,
     user_id: str,
 ) -> tuple[Any, JSONResponse | None]:
-    """Resolve a coordinator workstream and check ownership.
+    """Resolve a coordinator workstream by id.
 
     Returns ``(ws, None)`` on success — ``ws`` is the in-memory
     ``Workstream`` when present, ``None`` when the coordinator is
     persisted but not loaded (callers may then fall through to
     ``storage`` directly or trigger lazy rehydration).  Returns
-    ``(None, 404)`` on missing row / wrong kind / storage unavailable
-    / ownership mismatch.
+    ``(None, 404)`` on missing row / wrong kind / storage unavailable.
 
     Centralises the manager-first, storage-fallback, 404-mask ladder
     previously duplicated across ``coordinator_children`` /
-    ``coordinator_tasks`` / ``coordinator_history``.  Ownership uses
-    :func:`_check_row_owner_or_404` so the empty-string defense
-    applies uniformly.
+    ``coordinator_tasks`` / ``coordinator_history``.  Turnstone is a
+    trusted-team tool — ``user_id`` is metadata, not an access
+    boundary, so this helper no longer gates on row ownership; scope
+    auth (``admin.coordinator``) upstream is the gate.
     """
+    del user_id  # retained in signature for caller-site clarity; not consulted here
     miss = JSONResponse({"error": "coordinator not found"}, status_code=404)
     ws = coord_mgr.get(ws_id) if coord_mgr is not None else None
     if ws is None:
@@ -2424,23 +2361,7 @@ def _resolve_coordinator_or_404(
             return None, miss
         if row is None or row.get("kind") != WorkstreamKind.COORDINATOR:
             return None, miss
-        err = _check_row_owner_or_404(
-            request,
-            row.get("user_id") or "",
-            user_id,
-            error_msg="coordinator not found",
-        )
-        if err is not None:
-            return None, err
         return None, None
-    err = _check_row_owner_or_404(
-        request,
-        ws.user_id or "",
-        user_id,
-        error_msg="coordinator not found",
-    )
-    if err is not None:
-        return None, err
     return ws, None
 
 
@@ -2454,53 +2375,6 @@ def _auth_scopes(request: Request) -> set[str]:
     return set(getattr(auth, "scopes", []) or [])
 
 
-def _is_admin(request: Request) -> bool:
-    auth = getattr(getattr(request, "state", None), "auth_result", None)
-    perms: frozenset[str] = getattr(auth, "permissions", frozenset())
-    return "admin.users" in perms or "admin.roles" in perms
-
-
-def _effective_user_filter(request: Request) -> str | None | _DenyFilter:
-    """Resolve the effective ``user_id`` filter for a tenant-scoped aggregate.
-
-    Returns one of:
-
-    - ``None`` — admin (``admin.users`` / ``admin.roles``) or
-      service-scoped caller; no tenant filter — storage helpers see
-      cluster-wide rows.
-    - ``str`` — non-admin, non-service caller with a resolved uid;
-      storage helpers MUST receive ``user_id=<uid>`` and push the
-      filter into SQL.
-    - :data:`DENY_EMPTY_SUB` — non-admin, non-service caller whose
-      ``sub`` claim is blank.  Callers MUST short-circuit with their
-      endpoint's empty-shape response; passing ``None`` through to
-      storage would be a service-escape and passing ``""`` would
-      accidentally match legacy orphan rows with empty ``user_id``.
-
-    .. note::
-       The service-scope bypass is end-to-end only on endpoints that
-       do NOT first gate on :func:`_check_row_owner_or_404` (which
-       currently bypasses for ``admin.*`` permissions but not for
-       service scope).  A service-scoped non-admin caller to
-       ``coordinator_children`` / ``coordinator_metrics`` is 404'd
-       before the filter runs; the bypass there is reachable only by
-       admin callers.  ``_coordinator_rows`` and
-       ``cluster_ws_live_bulk`` honour the full three-way return.
-
-    See the class-level tenancy contract on
-    :class:`~turnstone.core.storage._protocol.StorageBackend` for the
-    storage-side requirement.
-    """
-    if _is_admin(request):
-        return None
-    if "service" in _auth_scopes(request):
-        return None
-    uid = _auth_user_id(request)
-    if not uid:
-        return DENY_EMPTY_SUB
-    return uid
-
-
 async def coordinator_create(request: Request) -> JSONResponse:
     """POST /v1/api/coordinator/new — create a new coordinator session."""
     from turnstone.core.audit import record_audit
@@ -2603,16 +2477,9 @@ async def coordinator_send(request: Request) -> JSONResponse:
     message = (body.get("message") or "").strip()
     if not message:
         return JSONResponse({"error": "message is required"}, status_code=400)
-    user_id = _auth_user_id(request)
     ws = coord_mgr.get(ws_id)
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality (not short-circuit on empty ws.user_id) —
-        # empty-owner rows would otherwise leak ws_id/state/history
-        # across tenants to anyone with admin.coordinator.  404 (not
-        # 403) to avoid leaking existence.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     coord_adapter = getattr(request.app.state, "coord_adapter", None)
     if coord_adapter is None or not coord_adapter.send(ws_id, message):
         # Distinguish "worker busy + queue full" from "ws not loaded".
@@ -2642,13 +2509,9 @@ async def coordinator_approve(request: Request) -> JSONResponse:
     approved = bool(body.get("approved", False))
     feedback = body.get("feedback")
     always = bool(body.get("always", False))
-    user_id = _auth_user_id(request)
     ws = coord_mgr.get(ws_id)
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality — empty-owner rows must not leak across tenants.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     ui = ws.ui
     if ui is None or not hasattr(ui, "resolve_approval"):
         return JSONResponse(
@@ -2682,9 +2545,6 @@ async def coordinator_cancel(request: Request) -> JSONResponse:
     ws = coord_mgr.get(ws_id)
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality — empty-owner rows must not leak across tenants.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     coord_mgr.cancel(ws_id)
     storage = getattr(request.app.state, "auth_storage", None)
     if storage is not None:
@@ -2718,9 +2578,6 @@ async def coordinator_close(request: Request) -> JSONResponse:
     ws = coord_mgr.get(ws_id)
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality — empty-owner rows must not leak across tenants.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     if not coord_mgr.close(ws_id):
         return JSONResponse({"error": "close failed"}, status_code=500)
     storage = getattr(request.app.state, "auth_storage", None)
@@ -2749,13 +2606,9 @@ async def coordinator_events(request: Request) -> Response:
     if err503 is not None:
         return err503
     ws_id = request.path_params.get("ws_id", "")
-    user_id = _auth_user_id(request)
     ws = coord_mgr.get(ws_id)
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality — empty-owner rows must not leak across tenants.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     ui = ws.ui
     if ui is None or not hasattr(ui, "_register_listener"):
         return JSONResponse({"error": "coordinator has no UI"}, status_code=409)
@@ -2831,13 +2684,10 @@ async def coordinator_list(request: Request) -> JSONResponse:
     coord_mgr, err503 = _require_coord_mgr(request)
     if err503 is not None:
         return err503
-    user_id = _auth_user_id(request)
-    all_rows = coord_mgr.list_all()
-    rows = (
-        all_rows
-        if _is_admin(request)
-        else [ws for ws in all_rows if ws.user_id and ws.user_id == user_id]
-    )
+    # Trusted-team visibility: any admin.coordinator-scoped caller
+    # sees cluster-wide active coordinators.  ``user_id`` stays on the
+    # response row as metadata for the UI.
+    rows = coord_mgr.list_all()
     return JSONResponse(
         {
             "coordinators": [
@@ -2879,17 +2729,10 @@ async def coordinator_saved(request: Request) -> JSONResponse:
     if err is not None:
         return err
 
-    if _is_admin(request):
-        user_filter: str | None = None
-    else:
-        caller_uid = _auth_user_id(request)
-        if not caller_uid:
-            # Blank sub on a non-service / non-admin token — fail closed
-            # rather than match every orphan / migration row with empty
-            # user_id.  Mirrors list_saved_workstreams.
-            return JSONResponse({"coordinators": []})
-        user_filter = caller_uid
-
+    # Trusted-team visibility (post-#400): any caller with the
+    # ``admin.coordinator`` scope sees cluster-wide saved coordinators;
+    # ``user_id`` stays as metadata on the persisted row but isn't a
+    # filter here.
     # Offload the blocking storage call + the lock-acquiring list_all
     # off the event loop, matching coordinator_create's pattern (#perf-2
     # from the saved-coordinators review).  list_workstreams_with_history
@@ -2900,7 +2743,7 @@ async def coordinator_saved(request: Request) -> JSONResponse:
         list_workstreams_with_history,
         limit=50,
         kind=WorkstreamKind.COORDINATOR,
-        user_id=user_filter,
+        user_id=None,
         state="closed",
     )
 
@@ -2974,7 +2817,6 @@ async def coordinator_detail(request: Request) -> JSONResponse:
     if err503 is not None:
         return err503
     ws_id = request.path_params.get("ws_id", "")
-    user_id = _auth_user_id(request)
     ws = coord_mgr.get(ws_id)
     if ws is None:
         # Lazy rehydration goes through the session factory, which can
@@ -2982,11 +2824,7 @@ async def coordinator_detail(request: Request) -> JSONResponse:
         # the correlation-id mask so stack traces don't leak through
         # the detail endpoint either.
         try:
-            ws = (
-                coord_mgr.open(ws_id, user_id="", admin=True)
-                if _is_admin(request)
-                else coord_mgr.open(ws_id, user_id=user_id)
-            )
+            ws = coord_mgr.open(ws_id)
         except ValueError as exc:
             return JSONResponse({"error": str(exc)}, status_code=503)
         except Exception:
@@ -3008,9 +2846,6 @@ async def coordinator_detail(request: Request) -> JSONResponse:
             )
         if ws is None:
             return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        # Strict equality — empty-owner rows must not leak across tenants.
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     return JSONResponse(
         {
             "ws_id": ws.id,
@@ -3040,18 +2875,11 @@ async def coordinator_open(request: Request) -> JSONResponse:
     ws_id = request.path_params.get("ws_id", "")
     if not ws_id:
         return JSONResponse({"error": "ws_id is required"}, status_code=400)
-    user_id = _auth_user_id(request)
     ws = coord_mgr.get(ws_id)
     if ws is not None:
-        if ws.user_id != user_id and not _is_admin(request):
-            return JSONResponse({"error": "coordinator not found"}, status_code=404)
         return JSONResponse({"ws_id": ws.id, "name": ws.name, "already_loaded": True})
     try:
-        ws = (
-            coord_mgr.open(ws_id, user_id="", admin=True)
-            if _is_admin(request)
-            else coord_mgr.open(ws_id, user_id=user_id)
-        )
+        ws = coord_mgr.open(ws_id)
     except ValueError as exc:
         return JSONResponse({"error": str(exc)}, status_code=503)
     except Exception:
@@ -3072,8 +2900,6 @@ async def coordinator_open(request: Request) -> JSONResponse:
         )
     if ws is None:
         return JSONResponse({"error": "coordinator not found"}, status_code=404)
-    if ws.user_id != user_id and not _is_admin(request):
-        return JSONResponse({"error": "coordinator not found"}, status_code=404)
     return JSONResponse({"ws_id": ws.id, "name": ws.name})
 
 
@@ -3147,21 +2973,15 @@ async def coordinator_children(request: Request) -> JSONResponse:
     if err404 is not None:
         return err404
 
-    # Tenant filter: push the caller's user_id into SQL so forged /
-    # migration-era rows with the same parent_ws_id but a different
-    # owner can't leak through.  Admins / service scope bypass the
-    # filter (they're expected to see the full subtree).  The prior
-    # _resolve_coordinator_or_404 already rejected non-admin callers
-    # with a blank sub, so DENY here is defense-in-depth.
-    filter_user_id = _effective_user_filter(request)
-    if filter_user_id is DENY_EMPTY_SUB:
-        return JSONResponse({"items": [], "truncated": False})
+    # Trusted-team visibility: any caller with admin.coordinator sees
+    # the full child subtree.  ``user_id`` stays on each row as
+    # metadata, not a filter.
     try:
         raw = storage.list_workstreams(
             limit=_CHILDREN_PAGE_LIMIT + 1,
             parent_ws_id=ws_id,
             kind=None,
-            user_id=filter_user_id,
+            user_id=None,
         )
     except Exception:
         correlation_id = secrets.token_hex(4)
@@ -3281,27 +3101,17 @@ async def coordinator_metrics(request: Request) -> JSONResponse:
     # (#perf-1).  Two cheap queries instead of a ``list_workstreams``
     # scan up to 10k rows.
     #
-    # Tenant filter on the aggregates — push user_id into SQL so a
-    # non-admin caller can't observe cross-tenant counts via forged /
-    # migration-era rows sharing parent_ws_id.  The 404-mask above
-    # already rejected foreign coord_ws_id, so the filter here is
-    # defense-in-depth against child rows with drifted user_id.
+    # Trusted-team visibility: aggregates run cluster-wide per the
+    # unified ownership model; ``user_id`` is not a filter here.
     from datetime import UTC, datetime
 
-    filter_user_id = _effective_user_filter(request)
-    if filter_user_id is DENY_EMPTY_SUB:
-        # Prior _resolve_coordinator_or_404 already rejected this
-        # shape, but belt-and-braces: never fall through to storage
-        # with ``user_id=""`` (matches legacy orphans) or ``None``
-        # (service escape).
-        return JSONResponse(_coordinator_metrics_payload(ws_id=ws_id))
     now_epoch = time.time()
     hour_ago_iso = datetime.fromtimestamp(now_epoch - 3600, tz=UTC).strftime("%Y-%m-%dT%H:%M:%S")
     try:
         state_counts = await asyncio.to_thread(
             storage.count_workstreams_by_state,
             parent_ws_id=ws_id,
-            user_id=filter_user_id,
+            user_id=None,
         )
     except Exception:
         log.debug("coordinator_metrics.state_counts_failed ws=%s", ws_id[:8], exc_info=True)
@@ -3312,7 +3122,7 @@ async def coordinator_metrics(request: Request) -> JSONResponse:
             storage.count_workstreams_since,
             hour_ago_iso,
             parent_ws_id=ws_id,
-            user_id=filter_user_id,
+            user_id=None,
         )
     except Exception:
         log.debug(
diff --git a/turnstone/core/session_manager.py b/turnstone/core/session_manager.py
index 9af5d885..9d039c6c 100644
--- a/turnstone/core/session_manager.py
+++ b/turnstone/core/session_manager.py
@@ -206,9 +206,11 @@ class SessionManager:
         """Construct a new workstream, persist, and register.
 
         Slot reservation + placeholder install happen under the lock
-        (single-phase, ported from CoordinatorManager). Session
-        construction runs outside the lock; on failure the slot + row
-        are rolled back so capacity isn't leaked.
+        (single-phase). Session construction runs outside the lock; on
+        failure the in-memory slot is freed so capacity isn't leaked.
+        The storage row survives construction failure — the next
+        ``open(ws_id)`` retries session construction rather than
+        forcing the user to create a brand-new workstream.
 
         Raises ``RuntimeError`` when the manager is at capacity with
         no idle workstream to evict — callers (HTTP handlers) translate
@@ -230,7 +232,7 @@ class SessionManager:
         # Persist before session construction. Fail-closed: if the row
         # can't be written, the in-memory session would be invisible to
         # any lazy-rehydrate path and show up as "missing" after
-        # restart — better to surface the storage failure now.
+        # restart — surface the storage failure now.
         try:
             self._storage.register_workstream(
                 ws_id,
@@ -256,12 +258,13 @@ class SessionManager:
                 **extra_session_kwargs,
             )
         except Exception:
+            # Release the slot so capacity isn't leaked, and call
+            # cleanup_ui on the placeholder so any listener/lock state
+            # the UI factory allocated is released. Storage row stays:
+            # the next open() on this ws_id retries construction.
+            self._adapter.cleanup_ui(ws)
             with self._lock:
                 self._remove_locked(ws_id)
-            try:
-                self._storage.delete_workstream(ws_id)
-            except Exception:
-                log.warning("session_mgr.rollback_delete_failed ws=%s", ws_id[:8], exc_info=True)
             raise
 
         self._adapter.emit_created(ws)
@@ -297,18 +300,14 @@ class SessionManager:
     # open — lazy rehydrate for a persisted workstream
     # ------------------------------------------------------------------
 
-    def open(
-        self,
-        ws_id: str,
-        *,
-        user_id: str,
-        admin: bool = False,
-    ) -> Workstream | None:
+    def open(self, ws_id: str) -> Workstream | None:
         """Rehydrate a persisted workstream on demand.
 
         Returns ``None`` when the row doesn't exist, doesn't match our
-        kind, is tombstoned (``state='deleted'``), or doesn't belong to
-        ``user_id`` (non-admin callers only).
+        kind, or is tombstoned (``state='deleted'``). Turnstone is a
+        trusted-team tool — ownership is metadata for audit/display,
+        not an access boundary; HTTP handlers gate callers at the
+        scope level, not the row level.
 
         Serializes concurrent opens of the same ws_id through a
         per-ws refcounted lock so two GETs don't each construct a
@@ -326,15 +325,12 @@ class SessionManager:
                 if row is None or row.get("kind") != self.kind:
                     return None
                 # ``deleted`` is a tombstone — never resurrect.
-                # ``closed`` IS resurrectable: Saved Workstreams landing
-                # makes restore an explicit user action, and
+                # ``closed`` IS resurrectable; the Saved Workstreams
+                # landing makes restore an explicit user action, and
                 # ``_reserve_and_install_locked`` still enforces
                 # max_active (evicting an idle peer or raising).
                 if row.get("state") == "deleted":
                     return None
-                row_owner = row.get("user_id") or ""
-                if not admin and row_owner != user_id:
-                    return None
 
                 with self._lock:
                     # Re-check fast path — another thread may have raced
@@ -344,7 +340,7 @@ class SessionManager:
                         return existing
                     ws, evicted = self._reserve_and_install_locked(
                         ws_id,
-                        user_id=row_owner,
+                        user_id=row.get("user_id") or "",
                         name=row.get("name") or f"ws-{ws_id[:4]}",
                         parent_ws_id=row.get("parent_ws_id"),
                     )
@@ -356,6 +352,9 @@ class SessionManager:
                 try:
                     ws.session = self._adapter.build_session(ws)
                 except Exception:
+                    # Clean up the UI the adapter built before re-raising
+                    # so any listener/lock resources are released.
+                    self._adapter.cleanup_ui(ws)
                     with self._lock:
                         self._remove_locked(ws_id)
                     raise
@@ -571,7 +570,17 @@ class SessionManager:
         ws.kind = self.kind
         ws.user_id = user_id
         ws.parent_ws_id = parent_ws_id if parent_ws_id else None
-        ws.ui = self._adapter.build_ui(ws)
+        try:
+            ws.ui = self._adapter.build_ui(ws)
+        except Exception:
+            # An IDLE peer may already have been popped above; if we
+            # propagate without unwinding, that peer leaks its session
+            # + worker + UI listeners and no ws_closed reaches
+            # subscribers.
+            if evicted is not None:
+                self._adapter.cleanup_ui(evicted)
+                self._adapter.emit_closed(evicted.id, reason="evicted")
+            raise
         self._workstreams[ws_id] = ws
         self._order.append(ws_id)
         if self._active_id is None:
diff --git a/turnstone/server.py b/turnstone/server.py
index 0240fddf..64b3af10 100644
--- a/turnstone/server.py
+++ b/turnstone/server.py
@@ -2965,9 +2965,6 @@ async def refresh_workstream_title(request: Request, ws_id: str = "") -> JSONRes
     ws_id = request.path_params.get("ws_id", "")
     log.info("ws.title.refresh_requested", ws_id=ws_id[:8] if ws_id else "empty")
     mgr = request.app.state.workstreams
-    # Cross-tenant rename is a phishing / denial-of-use vector — a
-    # malicious caller could push the victim's title to a misleading
-    # string visible in list/dashboard responses.
     _owner, err = _require_ws_access(request, ws_id, mgr=mgr)
     if err:
         return err
@@ -3003,7 +3000,6 @@ async def set_workstream_title(request: Request, ws_id: str = "") -> JSONRespons
     if not ws_id:
         return JSONResponse({"error": "ws_id is required"}, status_code=400)
     mgr = request.app.state.workstreams
-    # Cross-tenant rename gate — same rationale as refresh-title above.
     _owner, err = _require_ws_access(request, ws_id, mgr=mgr)
     if err:
         return err
@@ -3213,11 +3209,14 @@ def _require_ws_access(
     *,
     mgr: SessionManager | None = None,
 ) -> tuple[str, JSONResponse | None]:
-    """Resolve ``ws_id`` to its owner after verifying the caller has access.
+    """Resolve ``ws_id`` to its owner, 404-ing when the row doesn't exist.
 
-    Service-scoped tokens (internal callers) bypass ownership checks.
-    Returns ``(owner_user_id, None)`` on success.  The owner id is what
-    attachments should be filed under.
+    Turnstone is a trusted-team tool: scope-level auth (e.g.
+    ``admin.workstreams``) is the only gate; row-level ownership is not
+    enforced here.  Returns ``(owner_user_id, None)`` on success — the
+    persisted owner id, which attachments should be filed under so
+    existing storage shape is preserved.  Falls back to the caller's
+    own uid when the row has no recorded owner.
 
     When ``mgr`` is provided and the workstream is live in the manager,
     trust its cached ``user_id`` instead of round-tripping storage —
@@ -3228,18 +3227,11 @@ def _require_ws_access(
     ``mgr`` and fall through to the storage path.
     """
     caller = _auth_user_id(request)
-    scopes = _auth_scopes(request)
-    is_service = "service" in scopes
 
     if mgr is not None:
         ws_mem = mgr.get(ws_id)
         if ws_mem is not None:
-            owner_mem = ws_mem.user_id
-            if is_service:
-                return owner_mem or caller, None
-            if owner_mem and owner_mem != caller:
-                return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
-            return caller, None
+            return ws_mem.user_id or caller, None
         # Not in memory — fall through to storage so /delete etc.
         # still resolve persisted-but-not-loaded rows.
 
@@ -3248,17 +3240,7 @@ def _require_ws_access(
     owner = get_workstream_owner(ws_id)
     if owner is None:
         return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
-    if is_service:
-        # Trust the service caller; file under its own user_id if no owner
-        # is set, otherwise under the existing owner.
-        return owner or caller, None
-    # Authenticated user must own the workstream.  If the workstream was
-    # created before user tracking (owner blank) or by the same user, allow.
-    if owner and owner != caller:
-        # Return 404 (not 403) so non-owners cannot enumerate workstream
-        # existence by response code.
-        return "", JSONResponse({"error": "Workstream not found"}, status_code=404)
-    return caller, None
+    return owner or caller, None
 
 
 async def upload_attachment(request: Request) -> JSONResponse:
@@ -3489,20 +3471,13 @@ async def open_workstream(request: Request) -> JSONResponse:
         )
 
     uid: str = _auth_user_id(request)
-    scopes = _auth_scopes(request)
 
-    # Ownership gate: the stored owner must match the caller (or the
-    # caller must hold service scope — used by the console routing proxy
-    # and cluster rehydration paths).  Ownerless persisted rows (the
-    # startup ``name="default"`` workstream, pre-migration legacy rows)
-    # are claimable by any authenticated caller — consistent with the
-    # trusted-team visibility model the listing endpoints assume, and
-    # symmetric with how _require_ws_access handles the same rows on
-    # the per-workstream interactive handlers.  404 (not 403) so
-    # existence isn't enumerable by non-owners.
+    # Trusted-team model: scope-level auth (already enforced by the
+    # middleware) is the only gate.  ``user_id`` is metadata for audit
+    # + display, not an access boundary, so any authenticated caller
+    # can rehydrate any persisted workstream.  Fall back to the
+    # caller's uid when the row has no recorded owner.
     stored_owner = (ws_row.get("user_id") or "").strip()
-    if stored_owner and "service" not in scopes and stored_owner != uid:
-        return JSONResponse({"error": "Workstream not found"}, status_code=404)
     owner_uid = stored_owner or uid
 
     try: