feat: teams

tests/api/test_agent_team_sharing.py

@@ -0,0 +1,279 @@
+"""Endpoint tests for agent team-sharing paths.
+
+Covers the GET /api/get_agent read/fallback + name-resolution paths and the
+PUT /api/update_agent team-editor save-gate.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from contextlib import contextmanager
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+@pytest.fixture
+def client():
+    from application.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app.test_client()
+
+
+@contextmanager
+def _cm(value):
+    yield value
+
+
+def _patches(sub, repo, team_access, *, prompt_name="Resolved Prompt", source_details=None):
+    if source_details is None:
+        source_details = []
+    return [
+        patch("application.app.handle_auth", return_value={"sub": sub}),
+        patch("application.app.resolve_roles", return_value=["user"]),
+        patch("application.api.user.agents.routes.db_readonly", lambda: _cm(Mock())),
+        patch("application.api.user.agents.routes.AgentsRepository", return_value=repo),
+        patch("application.api.user.agents.routes.team_access_for", return_value=team_access),
+        # Resolve names by id (owner-agnostic) — patched so the test never
+        # touches the DB; the route is what we're asserting wires them in.
+        patch(
+            "application.api.user.agents.routes.resolve_prompt_name",
+            return_value=prompt_name,
+        ),
+        patch(
+            "application.api.user.agents.routes.resolve_source_details",
+            return_value=source_details,
+        ),
+    ]
+
+
+def _run(patches, client, agent_id):
+    for p in patches:
+        p.start()
+    try:
+        return client.get(f"/api/get_agent?id={agent_id}")
+    finally:
+        for p in reversed(patches):
+            p.stop()
+
+
+@pytest.mark.unit
+class TestGetAgentTeamFallback:
+    def test_owner_sees_own_agent(self, client):
+        aid = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = {"id": aid, "name": "Mine", "status": "published"}
+        resp = _run(_patches("alice", repo, None), client, aid)
+        assert resp.status_code == 200
+        data = json.loads(resp.data)
+        assert data["ownership"] == "user"
+        assert data["team_access"] is None
+        repo.get_by_id.assert_not_called()
+
+    def test_team_member_sees_shared_agent(self, client):
+        aid = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = None  # not the owner
+        repo.get_by_id.return_value = {
+            "id": aid,
+            "name": "Shared",
+            "status": "published",
+            # Owner secrets that must NOT leak to a team grantee.
+            "shared_token": "secret-public-token",
+            "key": "agentkey1234567890",
+        }
+        resp = _run(_patches("bob", repo, "viewer"), client, aid)
+        assert resp.status_code == 200
+        data = json.loads(resp.data)
+        assert data["ownership"] == "team"
+        assert data["team_access"] == "viewer"
+        # Owner's public share token / API key are blanked for grantees.
+        assert data["shared_token"] == ""
+        assert data.get("key", "") == ""
+        # get_by_id only reached AFTER the team grant check.
+        repo.get_by_id.assert_called_once_with(aid)
+
+    def test_no_access_returns_404(self, client):
+        aid = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = None
+        resp = _run(_patches("stranger", repo, None), client, aid)
+        assert resp.status_code == 404
+        repo.get_by_id.assert_not_called()
+
+
+@pytest.mark.unit
+class TestGetAgentResolvesNames:
+    """GET /api/get_agent embeds owner-agnostic prompt/source names so a team
+    grantee sees the owner's prompt name + source names rather than a blank
+    prompt / "External KB" (the client can't resolve the owner's resources)."""
+
+    def test_team_member_payload_has_resolved_names(self, client):
+        aid = str(uuid.uuid4())
+        src = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = None  # not the owner
+        repo.get_by_id.return_value = {
+            "id": aid,
+            "name": "Shared",
+            "status": "published",
+            "prompt_id": str(uuid.uuid4()),
+            "source_id": src,
+        }
+        details = [{"id": src, "name": "Owner KB"}]
+        resp = _run(
+            _patches(
+                "bob", repo, "editor",
+                prompt_name="Owner Prompt", source_details=details,
+            ),
+            client,
+            aid,
+        )
+        assert resp.status_code == 200
+        data = json.loads(resp.data)
+        assert data["ownership"] == "team"
+        assert data["prompt_name"] == "Owner Prompt"
+        assert data["source_details"] == details
+
+    def test_owner_payload_also_has_resolved_names(self, client):
+        aid = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = {
+            "id": aid,
+            "name": "Mine",
+            "status": "published",
+            "prompt_id": "default",
+        }
+        resp = _run(
+            _patches("alice", repo, None, prompt_name="Default", source_details=[]),
+            client,
+            aid,
+        )
+        assert resp.status_code == 200
+        data = json.loads(resp.data)
+        assert data["ownership"] == "user"
+        assert data["prompt_name"] == "Default"
+        assert data["source_details"] == []
+
+
+def _update_patches(sub, repo, team_access, can_access_mock):
+    return [
+        patch("application.app.handle_auth", return_value={"sub": sub}),
+        patch("application.app.resolve_roles", return_value=["user"]),
+        patch("application.api.user.agents.routes.db_session", lambda: _cm(Mock())),
+        patch("application.api.user.agents.routes.AgentsRepository", return_value=repo),
+        patch("application.api.user.agents.routes.team_access_for", return_value=team_access),
+        patch("application.api.user.agents.routes.can_access", can_access_mock),
+    ]
+
+
+def _run_update(patches, client, agent_id, body):
+    for p in patches:
+        p.start()
+    try:
+        return client.put(f"/api/update_agent/{agent_id}", json=body)
+    finally:
+        for p in reversed(patches):
+            p.stop()
+
+
+def _shared_agent_row(aid, owner_src, owner_prompt):
+    return {
+        "id": aid,
+        "name": "Shared",
+        "description": "desc",
+        "status": "draft",
+        "source_id": owner_src,
+        "extra_source_ids": [],
+        "prompt_id": owner_prompt,
+        "tools": [],
+        "chunks": 2,
+        "agent_type": "classic",
+        "image": "",
+    }
+
+
+@pytest.mark.unit
+class TestUpdateAgentEditorSaveGate:
+    """PUT /api/update_agent: a team EDITOR may save an agent while keeping the
+    owner's existing source/prompt (which aren't independently shared with
+    them), but must not ATTACH a new source/prompt they can't access."""
+
+    def test_editor_keeps_owner_source_and_prompt(self, client):
+        aid, owner_src, owner_prompt = (
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+        )
+        repo = Mock()
+        repo.get_any.return_value = None  # not the owner
+        repo.get_by_id.return_value = _shared_agent_row(aid, owner_src, owner_prompt)
+        repo.update_by_id.return_value = {"id": aid}
+        # Editor holds NO independent grant on the source/prompt themselves.
+        can_access = Mock(return_value=False)
+        resp = _run_update(
+            _update_patches("editor-bob", repo, "editor", can_access),
+            client,
+            aid,
+            {
+                "name": "Shared",
+                "description": "desc",
+                "source": owner_src,
+                "prompt_id": owner_prompt,
+            },
+        )
+        assert resp.status_code == 200
+        # Unchanged refs are exempt — the access check is never consulted.
+        can_access.assert_not_called()
+        repo.update_by_id.assert_called_once()
+
+    def test_editor_cannot_attach_new_unshared_source(self, client):
+        aid, owner_src, owner_prompt = (
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+        )
+        new_src = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = None
+        repo.get_by_id.return_value = _shared_agent_row(aid, owner_src, owner_prompt)
+        repo.update_by_id.return_value = {"id": aid}
+        can_access = Mock(return_value=False)
+        resp = _run_update(
+            _update_patches("editor-bob", repo, "editor", can_access),
+            client,
+            aid,
+            {"name": "Shared", "description": "desc", "sources": [new_src]},
+        )
+        assert resp.status_code == 403
+        assert "Source not accessible" in json.loads(resp.data)["message"]
+        repo.update_by_id.assert_not_called()
+
+    def test_editor_cannot_attach_new_unshared_prompt(self, client):
+        aid, owner_src, owner_prompt = (
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+            str(uuid.uuid4()),
+        )
+        new_prompt = str(uuid.uuid4())
+        repo = Mock()
+        repo.get_any.return_value = None
+        repo.get_by_id.return_value = _shared_agent_row(aid, owner_src, owner_prompt)
+        repo.update_by_id.return_value = {"id": aid}
+        can_access = Mock(return_value=False)
+        resp = _run_update(
+            _update_patches("editor-bob", repo, "editor", can_access),
+            client,
+            aid,
+            {
+                "name": "Shared",
+                "description": "desc",
+                "source": owner_src,
+                "prompt_id": new_prompt,
+            },
+        )
+        assert resp.status_code == 403
+        assert "Prompt not accessible" in json.loads(resp.data)["message"]
+        repo.update_by_id.assert_not_called()

tests/api/test_teams_endpoints.py

@@ -0,0 +1,397 @@
+"""Endpoint authorization tests for the teams API.
+
+These drive the real app.py chokepoint + team_authz plane (only handle_auth /
+resolve_roles and the leaf repos are mocked), so they catch regressions in the
+authorization wiring: who can read/manage a team, that team_id comes from the
+URL path (never the body), and that sharing requires resource ownership. Data
+correctness lives in tests/storage/db/repositories/test_teams.py.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+@pytest.fixture
+def client():
+    from application.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app.test_client()
+
+
+@contextmanager
+def _cm(value):
+    yield value
+
+
+def _auth(sub="u1", roles=("user",), team_role=None):
+    """Patch the chokepoint auth + the team_authz membership resolution.
+
+    ``team_role`` is what TeamMembersRepository.role_for returns inside
+    team_authz (None = not a member).
+    """
+    members = Mock()
+    members.role_for.return_value = team_role
+    return [
+        patch("application.app.handle_auth", return_value={"sub": sub}),
+        patch("application.app.resolve_roles", return_value=list(roles)),
+        patch("application.api.user.team_authz.db_readonly", lambda: _cm(Mock())),
+        patch("application.api.user.team_authz.TeamMembersRepository", return_value=members),
+    ]
+
+
+def _apply(patches):
+    for p in patches:
+        p.start()
+
+
+def _stop(patches):
+    for p in reversed(patches):
+        p.stop()
+
+
+@pytest.mark.unit
+class TestTeamCreation:
+    def test_unauthenticated_401(self, client):
+        with patch("application.app.handle_auth", return_value=None):
+            resp = client.post("/api/teams", json={"name": "Acme"})
+        assert resp.status_code == 401
+
+    def test_self_serve_create_returns_201(self, client):
+        teams_repo = Mock()
+        teams_repo.slug_exists.return_value = False
+        teams_repo.create.return_value = {"id": "t1", "name": "Acme", "slug": "acme"}
+        members_repo = Mock()
+        patches = _auth(sub="alice") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post("/api/teams", json={"name": "Acme"})
+        finally:
+            _stop(patches)
+        assert resp.status_code == 201
+        # Creator is added as team_admin in the same flow.
+        args, kwargs = members_repo.add_member.call_args
+        assert kwargs.get("role") == "team_admin" or "team_admin" in args
+
+    def test_missing_name_400(self, client):
+        patches = _auth(sub="alice")
+        _apply(patches)
+        try:
+            resp = client.post("/api/teams", json={})
+        finally:
+            _stop(patches)
+        assert resp.status_code == 400
+
+
+@pytest.mark.unit
+class TestTeamAccessControl:
+    def test_detail_requires_membership(self, client):
+        patches = _auth(sub="stranger", team_role=None)
+        _apply(patches)
+        try:
+            resp = client.get("/api/teams/team-1")
+        finally:
+            _stop(patches)
+        assert resp.status_code == 403
+
+    def test_member_can_read_detail(self, client):
+        teams_repo = Mock()
+        teams_repo.get.return_value = {"id": "team-1", "name": "Acme", "owner_id": "alice"}
+        members_repo = Mock()
+        members_repo.list_members.return_value = []
+        members_repo.role_for.return_value = "team_member"
+        patches = _auth(sub="bob", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.get("/api/teams/team-1")
+        finally:
+            _stop(patches)
+        assert resp.status_code == 200
+
+    def test_team_id_taken_from_path_not_body(self, client):
+        # Caller is NOT a member of team-1; a forged team_id in the body must be
+        # ignored — the decorator reads team_id from the URL path only.
+        patches = _auth(sub="stranger", team_role=None)
+        _apply(patches)
+        try:
+            resp = client.get("/api/teams/team-1", json={"team_id": "team-i-own"})
+        finally:
+            _stop(patches)
+        assert resp.status_code == 403
+
+    def test_add_member_requires_team_admin(self, client):
+        patches = _auth(sub="bob", team_role="team_member")
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/members", json={"user_id": "x", "role": "team_member"}
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 403
+
+    def test_team_admin_can_add_member(self, client):
+        members_repo = Mock()
+        patches = _auth(sub="alice", team_role="team_admin") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/members", json={"user_id": "carol", "role": "team_member"}
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 200
+        members_repo.set_manual_role.assert_called_once()
+
+    def test_add_member_by_email_resolves_to_sub(self, client):
+        members_repo = Mock()
+        users_repo = Mock()
+        users_repo.find_by_email.return_value = {"user_id": "resolved-sub"}
+        patches = _auth(sub="alice", team_role="team_admin") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+            patch(
+                "application.api.user.teams.routes.UsersRepository",
+                return_value=users_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/members",
+                json={"email": "bob@example.com", "role": "team_member"},
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 200
+        args, kwargs = members_repo.set_manual_role.call_args
+        assert "resolved-sub" in args  # resolved sub used, not the email
+
+    def test_add_member_email_not_found(self, client):
+        users_repo = Mock()
+        users_repo.find_by_email.return_value = None
+        patches = _auth(sub="alice", team_role="team_admin") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch(
+                "application.api.user.teams.routes.UsersRepository",
+                return_value=users_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/members",
+                json={"email": "ghost@example.com", "role": "team_member"},
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 404
+
+    def test_delete_team_non_owner_forbidden(self, client):
+        teams_repo = Mock()
+        teams_repo.get.return_value = {"id": "team-1", "owner_id": "alice"}
+        patches = _auth(sub="bob", roles=("user",), team_role="team_admin") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
+        ]
+        _apply(patches)
+        try:
+            resp = client.delete("/api/teams/team-1")
+        finally:
+            _stop(patches)
+        # team_admin is not the owner — only the owner (or a global admin) deletes.
+        assert resp.status_code == 403
+        teams_repo.delete.assert_not_called()
+
+
+@pytest.mark.unit
+class TestSharingAuthz:
+    def test_share_requires_ownership(self, client):
+        patches = _auth(sub="bob", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.owns_resource", return_value=False),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={
+                    "resource_type": "agent",
+                    "resource_id": "11111111-1111-1111-1111-111111111111",
+                },
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 403
+
+    def test_owner_can_share(self, client):
+        grants_repo = Mock()
+        grants_repo.grant.return_value = {"id": "g1", "access_level": "viewer"}
+        patches = _auth(sub="alice", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.owns_resource", return_value=True),
+            patch(
+                "application.api.user.teams.routes.TeamResourceGrantsRepository",
+                return_value=grants_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={
+                    "resource_type": "agent",
+                    "resource_id": "22222222-2222-2222-2222-222222222222",
+                    "access_level": "editor",
+                },
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 201
+        _, kwargs = grants_repo.grant.call_args
+        assert kwargs.get("access_level") == "editor"
+
+    def test_invalid_resource_type_rejected(self, client):
+        patches = _auth(sub="alice", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={"resource_type": "secret_thing", "resource_id": "a1"},
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 400
+
+    def test_share_with_member_validates_membership(self, client):
+        # target_user_id must be a member of the team.
+        members_repo = Mock()
+        members_repo.is_member.return_value = False
+        patches = _auth(sub="alice", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.owns_resource", return_value=True),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={
+                    "resource_type": "agent",
+                    "resource_id": "22222222-2222-2222-2222-222222222222",
+                    "target_user_id": "stranger",
+                },
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 400
+
+    def test_share_with_valid_member(self, client):
+        members_repo = Mock()
+        members_repo.is_member.return_value = True
+        grants_repo = Mock()
+        grants_repo.grant.return_value = {"id": "g1", "target_user_id": "bob"}
+        patches = _auth(sub="alice", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.owns_resource", return_value=True),
+            patch(
+                "application.api.user.teams.routes.TeamMembersRepository",
+                return_value=members_repo,
+            ),
+            patch(
+                "application.api.user.teams.routes.TeamResourceGrantsRepository",
+                return_value=grants_repo,
+            ),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={
+                    "resource_type": "agent",
+                    "resource_id": "22222222-2222-2222-2222-222222222222",
+                    "target_user_id": "bob",
+                },
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 201
+        _, kwargs = grants_repo.grant.call_args
+        assert kwargs.get("target_user_id") == "bob"
+
+    def test_non_uuid_resource_id_rejected(self, client):
+        # Team grants are UUID-only (post-cutover); a legacy/non-UUID id must be
+        # rejected cleanly, not cast-and-poison the txn into a generic error.
+        patches = _auth(sub="alice", team_role="team_member") + [
+            patch("application.api.user.teams.routes.db_session", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.owns_resource", return_value=True),
+        ]
+        _apply(patches)
+        try:
+            resp = client.post(
+                "/api/teams/team-1/grants",
+                json={"resource_type": "agent", "resource_id": "legacy-mongo-id"},
+            )
+        finally:
+            _stop(patches)
+        assert resp.status_code == 400
+
+
+@pytest.mark.unit
+class TestAdminOversight:
+    def test_non_admin_forbidden(self, client):
+        patches = _auth(sub="bob", roles=("user",))
+        _apply(patches)
+        try:
+            resp = client.get("/api/admin/teams")
+        finally:
+            _stop(patches)
+        assert resp.status_code == 403
+
+    def test_global_admin_lists_all(self, client):
+        teams_repo = Mock()
+        teams_repo.list_all.return_value = [{"id": "t1", "member_count": 3}]
+        patches = _auth(sub="root", roles=("admin", "user")) + [
+            patch("application.api.user.teams.routes.db_readonly", lambda: _cm(Mock())),
+            patch("application.api.user.teams.routes.TeamsRepository", return_value=teams_repo),
+        ]
+        _apply(patches)
+        try:
+            resp = client.get("/api/admin/teams")
+        finally:
+            _stop(patches)
+        assert resp.status_code == 200
+        assert json.loads(resp.data)["teams"][0]["member_count"] == 3

tests/api/user/sources/test_chunks.py

@@ -210,7 +210,9 @@ class TestAddChunk:
             response = AddChunk().post()
         assert response.status_code == 400
 
-    def test_returns_404_source_not_found(self, app, pg_conn):
+    def test_returns_403_inaccessible_source(self, app, pg_conn):
+        # No ownership and no team editor grant resolves to None, which the
+        # owner-or-editor gate answers as 403 "Source not accessible".
         from application.api.user.sources.chunks import AddChunk
 
         with _patch_db(pg_conn), app.test_request_context(
@@ -223,7 +225,7 @@ class TestAddChunk:
             from flask import request
             request.decoded_token = {"sub": "u"}
             response = AddChunk().post()
-        assert response.status_code == 404
+        assert response.status_code == 403
 
     def test_adds_chunk(self, app, pg_conn):
         from application.api.user.sources.chunks import AddChunk
@@ -282,7 +284,9 @@ class TestDeleteChunk:
             response = DeleteChunk().delete()
         assert response.status_code == 401
 
-    def test_returns_404_source_not_found(self, app, pg_conn):
+    def test_returns_403_inaccessible_source(self, app, pg_conn):
+        # No ownership and no team editor grant resolves to None, which the
+        # owner-or-editor gate answers as 403 "Source not accessible".
         from application.api.user.sources.chunks import DeleteChunk
 
         with _patch_db(pg_conn), app.test_request_context(
@@ -292,7 +296,7 @@ class TestDeleteChunk:
             from flask import request
             request.decoded_token = {"sub": "u"}
             response = DeleteChunk().delete()
-        assert response.status_code == 404
+        assert response.status_code == 403
 
     def test_deletes_chunk(self, app, pg_conn):
         from application.api.user.sources.chunks import DeleteChunk
@@ -359,7 +363,9 @@ class TestUpdateChunk:
             response = UpdateChunk().put()
         assert response.status_code == 400
 
-    def test_returns_404_source_not_found(self, app, pg_conn):
+    def test_returns_403_inaccessible_source(self, app, pg_conn):
+        # No ownership and no team editor grant resolves to None, which the
+        # owner-or-editor gate answers as 403 "Source not accessible".
         from application.api.user.sources.chunks import UpdateChunk
 
         with _patch_db(pg_conn), app.test_request_context(
@@ -372,7 +378,7 @@ class TestUpdateChunk:
             from flask import request
             request.decoded_token = {"sub": "u"}
             response = UpdateChunk().put()
-        assert response.status_code == 404
+        assert response.status_code == 403
 
     def test_returns_404_chunk_not_found(self, app, pg_conn):
         from application.api.user.sources.chunks import UpdateChunk

tests/api/user/sources/test_routes.py

@@ -512,7 +512,9 @@ class TestSyncSource:
             response = SyncSource().post()
         assert response.status_code == 400
 
-    def test_returns_404_missing_source(self, app, pg_conn):
+    def test_returns_403_inaccessible_source(self, app, pg_conn):
+        # No ownership and no team editor grant resolves to None, which the
+        # owner-or-editor gate answers as 403 "Source not accessible".
         from application.api.user.sources.routes import SyncSource
 
         with _patch_db(pg_conn), app.test_request_context(
@@ -523,7 +525,7 @@ class TestSyncSource:
             from flask import request
             request.decoded_token = {"sub": "u"}
             response = SyncSource().post()
-        assert response.status_code == 404
+        assert response.status_code == 403
 
     def test_returns_400_for_connector_type(self, app, pg_conn):
         from application.api.user.sources.routes import SyncSource
@@ -659,7 +661,9 @@ class TestReingestSource:
             response = ReingestSource().post()
         assert response.status_code == 400
 
-    def test_returns_404_missing_source(self, app, pg_conn):
+    def test_returns_403_inaccessible_source(self, app, pg_conn):
+        # No ownership and no team editor grant resolves to None, which the
+        # owner-or-editor gate answers as 403 "Source not accessible".
         from application.api.user.sources.routes import ReingestSource
 
         with _patch_db(pg_conn), app.test_request_context(
@@ -670,7 +674,7 @@ class TestReingestSource:
             from flask import request
             request.decoded_token = {"sub": "u"}
             response = ReingestSource().post()
-        assert response.status_code == 404
+        assert response.status_code == 403
 
     def test_triggers_reingest_task(self, app, pg_conn):
         from application.api.user.sources.routes import ReingestSource
@@ -701,6 +705,58 @@ class TestReingestSource:
             f"reingest-source:{user}:{src['id']}"
         )
 
+    def test_team_editor_reingests_as_owner(self, app, pg_conn):
+        """A team editor (not the owner) can reingest a shared source; the
+        task dispatches AS the real owner so the owner-scoped pipeline and
+        the owner-agnostic vector partition stay consistent.
+        """
+        import uuid
+
+        from application.api.user.sources.routes import ReingestSource
+        from application.storage.db.repositories.team_members import (
+            TeamMembersRepository,
+        )
+        from application.storage.db.repositories.team_resource_grants import (
+            TeamResourceGrantsRepository,
+        )
+        from application.storage.db.repositories.teams import TeamsRepository
+
+        owner = "alice-reingest"
+        editor = "bob-reingest"
+        src = _seed_source(pg_conn, owner, name="shared-src", type="file")
+        sid = str(src["id"])
+        team = TeamsRepository(pg_conn).create(
+            "Acme", f"acme-{uuid.uuid4().hex[:8]}", owner
+        )
+        TeamMembersRepository(pg_conn).add_member(
+            team["id"], editor, role="team_member"
+        )
+        TeamResourceGrantsRepository(pg_conn).grant(
+            team["id"], "source", sid, owner_id=owner, granted_by=owner,
+            access_level="editor",
+        )
+
+        fake_task = MagicMock(id="reingest-task-editor")
+        with _patch_db(pg_conn), patch(
+            "application.api.user.sources.routes.reingest_source_task.delay",
+            return_value=fake_task,
+        ) as mock_delay, app.test_request_context(
+            "/api/sources/reingest",
+            method="POST",
+            json={"source_id": sid},
+        ):
+            from flask import request
+            request.decoded_token = {"sub": editor}
+            response = ReingestSource().post()
+
+        assert response.status_code == 200
+        assert mock_delay.call_args.kwargs["source_id"] == sid
+        # Dispatched AS the owner, not the editor caller.
+        assert mock_delay.call_args.kwargs["user"] == owner
+        assert mock_delay.call_args.kwargs["idempotency_key"] == (
+            f"reingest-source:{owner}:{sid}"
+        )
+
     def test_clears_stalled_ingest_progress_row(self, app, pg_conn):
         """Reingest drops the stale chunk-progress row so the sources
         list stops deriving a 'failed' ingest status for the source.