feat(persistence/authz): PR6 T6.4 — check_access takes workspace_id

`ThreadMetaStore.check_access(thread_id, user_id, workspace_id, *,
require_existing)` is now a three-positional method. Cross-workspace is
denied unconditionally — even when the row exists and `user_id` matches
— so the decorator layer can convert the False into a **404** and never
leak the existence of a thread across tenants. Inside the workspace, the
existing legacy semantics still hold (NULL `row.user_id` stays
"shared in workspace", `require_existing` still gates the missing-row
path against ghost-row re-targeting).

`@require_permission(owner_check=True)` in `app/gateway/authz.py` now
reads the active workspace from `get_effective_workspace_id()` (set by
PR4 AuthMiddleware; falls back to "default" in no-auth dev) and passes
it through. The existing 404-not-403 mapping is unchanged.

Existing positional callers in `test_thread_meta_repo.py` and the
permissive mock in `test_threads_router.py` were updated for the new
arity. 58 thread_meta / router / memory tests stay green; 90 auth /
uploads / suggestions tests stay green.

backend/tests/test_thread_meta_repo.py

@@ -64,21 +64,21 @@ class TestThreadMetaRepository:
     @pytest.mark.anyio
     async def test_check_access_no_record_allows(self, tmp_path):
         repo = await _make_repo(tmp_path)
-        assert await repo.check_access("unknown", "user1") is True
+        assert await repo.check_access("unknown", "user1", "test-workspace-autouse") is True
         await _cleanup()
 
     @pytest.mark.anyio
     async def test_check_access_owner_matches(self, tmp_path):
         repo = await _make_repo(tmp_path)
         await repo.create("t1", user_id="user1")
-        assert await repo.check_access("t1", "user1") is True
+        assert await repo.check_access("t1", "user1", "test-workspace-autouse") is True
         await _cleanup()
 
     @pytest.mark.anyio
     async def test_check_access_owner_mismatch(self, tmp_path):
         repo = await _make_repo(tmp_path)
         await repo.create("t1", user_id="user1")
-        assert await repo.check_access("t1", "user2") is False
+        assert await repo.check_access("t1", "user2", "test-workspace-autouse") is False
         await _cleanup()
 
     @pytest.mark.anyio
@@ -87,7 +87,7 @@ class TestThreadMetaRepository:
         # Explicit user_id=None to bypass the new AUTO default that
         # would otherwise pick up the test user from the autouse fixture.
         await repo.create("t1", user_id=None)
-        assert await repo.check_access("t1", "anyone") is True
+        assert await repo.check_access("t1", "anyone", "test-workspace-autouse") is True
         await _cleanup()
 
     @pytest.mark.anyio
@@ -99,21 +99,21 @@ class TestThreadMetaRepository:
         caller "claim" it as untracked. The strict mode demands a row.
         """
         repo = await _make_repo(tmp_path)
-        assert await repo.check_access("never-existed", "user1", require_existing=True) is False
+        assert await repo.check_access("never-existed", "user1", "test-workspace-autouse", require_existing=True) is False
         await _cleanup()
 
     @pytest.mark.anyio
     async def test_check_access_strict_owner_match_allowed(self, tmp_path):
         repo = await _make_repo(tmp_path)
         await repo.create("t1", user_id="user1")
-        assert await repo.check_access("t1", "user1", require_existing=True) is True
+        assert await repo.check_access("t1", "user1", "test-workspace-autouse", require_existing=True) is True
         await _cleanup()
 
     @pytest.mark.anyio
     async def test_check_access_strict_owner_mismatch_denied(self, tmp_path):
         repo = await _make_repo(tmp_path)
         await repo.create("t1", user_id="user1")
-        assert await repo.check_access("t1", "user2", require_existing=True) is False
+        assert await repo.check_access("t1", "user2", "test-workspace-autouse", require_existing=True) is False
         await _cleanup()
 
     @pytest.mark.anyio
@@ -126,7 +126,7 @@ class TestThreadMetaRepository:
         """
         repo = await _make_repo(tmp_path)
         await repo.create("t1", user_id=None)
-        assert await repo.check_access("t1", "anyone", require_existing=True) is True
+        assert await repo.check_access("t1", "anyone", "test-workspace-autouse", require_existing=True) is True
         await _cleanup()
 
     @pytest.mark.anyio

backend/tests/test_thread_meta_workspace_filter.py

@@ -237,6 +237,36 @@ class TestSearchUpdateDeleteWorkspace:
             await _cleanup()
         assert row["metadata"] == {"k": "alpha"}
 
+    @pytest.mark.anyio
+    async def test_check_access_cross_workspace_false(self, tmp_path):
+        """`check_access` returns False for cross-workspace, even with matching user_id."""
+        repo = await _make_repo(tmp_path, workspaces=("ws-alpha", "ws-beta"))
+        token = _use_workspace("ws-alpha")
+        try:
+            await repo.create("t1", user_id="alice")
+        finally:
+            reset_current_workspace(token)
+        try:
+            assert await repo.check_access("t1", "alice", "ws-beta") is False
+            assert await repo.check_access("t1", "alice", "ws-alpha") is True
+        finally:
+            await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_check_access_strict_cross_workspace_false(self, tmp_path):
+        """require_existing=True path also denies cross-workspace."""
+        repo = await _make_repo(tmp_path, workspaces=("ws-alpha", "ws-beta"))
+        token = _use_workspace("ws-alpha")
+        try:
+            await repo.create("t1", user_id="alice")
+        finally:
+            reset_current_workspace(token)
+        try:
+            assert await repo.check_access("t1", "alice", "ws-beta", require_existing=True) is False
+            assert await repo.check_access("t1", "alice", "ws-alpha", require_existing=True) is True
+        finally:
+            await _cleanup()
+
     @pytest.mark.anyio
     async def test_delete_blocked_across_workspace(self, tmp_path):
         repo = await _make_repo(tmp_path, workspaces=("ws-alpha", "ws-beta"))

backend/tests/test_threads_router.py

@@ -27,21 +27,21 @@ class _PermissiveThreadMetaStore(MemoryThreadMetaStore):
     timestamp wire format.
     """
 
-    async def _get_owned_record(self, thread_id, user_id, method_name):  # type: ignore[override]
+    async def _get_owned_record(self, thread_id, user_id, workspace_id, method_name):  # type: ignore[override]
         item = await self._store.aget(THREADS_NS, thread_id)
         return dict(item.value) if item is not None else None
 
-    async def check_access(self, thread_id, user_id, *, require_existing=False):  # type: ignore[override]
+    async def check_access(self, thread_id, user_id, workspace_id, *, require_existing=False):  # type: ignore[override]
         item = await self._store.aget(THREADS_NS, thread_id)
         if item is None:
             return not require_existing
         return True
 
-    async def create(self, thread_id, *, assistant_id=None, user_id=None, display_name=None, metadata=None):  # type: ignore[override]
-        return await super().create(thread_id, assistant_id=assistant_id, user_id=None, display_name=display_name, metadata=metadata)
+    async def create(self, thread_id, *, assistant_id=None, user_id=None, workspace_id=None, display_name=None, metadata=None):  # type: ignore[override]
+        return await super().create(thread_id, assistant_id=assistant_id, user_id=None, workspace_id=None, display_name=display_name, metadata=metadata)
 
-    async def search(self, *, metadata=None, status=None, limit=100, offset=0, user_id=None):  # type: ignore[override]
-        return await super().search(metadata=metadata, status=status, limit=limit, offset=offset, user_id=None)
+    async def search(self, *, metadata=None, status=None, limit=100, offset=0, user_id=None, workspace_id=None):  # type: ignore[override]
+        return await super().search(metadata=metadata, status=status, limit=limit, offset=offset, user_id=None, workspace_id=None)
 
 
 def _build_thread_app() -> tuple[FastAPI, InMemoryStore, InMemorySaver]: