diff --git a/backend/app/dao/chat_session_dao.py b/backend/app/dao/chat_session_dao.py
index 3049a5daf..f6d45d3d4 100644
--- a/backend/app/dao/chat_session_dao.py
+++ b/backend/app/dao/chat_session_dao.py
@@ -9,6 +9,8 @@
 
 from app.dao.base import TenantScopedBaseDAO
 from app.models.chat_session import ChatSession
+from app.models.group import Group, GroupMember
+from app.models.participant import Participant
 
 
 class ChatSessionDAO(TenantScopedBaseDAO[ChatSession]):
@@ -47,6 +49,52 @@ async def get_active_for_agent(
             )
             return (await session_db.execute(stmt)).scalar_one_or_none()
 
+    async def get_active_for_sandbox_agent(
+        self,
+        *,
+        tenant_id: uuid.UUID,
+        agent_id: uuid.UUID,
+        session_id: uuid.UUID,
+        db: Any = None,
+    ) -> ChatSession | None:
+        """Authorize a Session for one Agent's local sandbox execution.
+
+        Direct and external-channel group Sessions retain exact Agent ownership.
+        Native group Sessions are shared, so they require an active Agent
+        participant membership in the active tenant-owned Group instead.
+        """
+        async with self.session(db=db, readonly=True) as session_db:
+            session_stmt = select(ChatSession).where(
+                ChatSession.tenant_id == tenant_id,
+                ChatSession.id == session_id,
+                ChatSession.deleted_at.is_(None),
+            )
+            chat_session = (await session_db.execute(session_stmt)).scalar_one_or_none()
+            if chat_session is None:
+                return None
+
+            if chat_session.group_id is None:
+                return chat_session if chat_session.agent_id == agent_id else None
+
+            if chat_session.session_type != "group" or chat_session.agent_id is not None:
+                return None
+
+            membership_stmt = (
+                select(GroupMember.id)
+                .join(Group, Group.id == GroupMember.group_id)
+                .join(Participant, Participant.id == GroupMember.participant_id)
+                .where(
+                    Group.id == chat_session.group_id,
+                    Group.tenant_id == tenant_id,
+                    Group.deleted_at.is_(None),
+                    GroupMember.removed_at.is_(None),
+                    Participant.type == "agent",
+                    Participant.ref_id == agent_id,
+                )
+            )
+            membership_id = (await session_db.execute(membership_stmt)).scalar_one_or_none()
+            return chat_session if membership_id is not None else None
+
     async def get_including_deleted(self, session_id: uuid.UUID, db: Any = None) -> ChatSession | None:
         """Fetch a session by ID including soft-deleted records."""
         tenant_id = self._require_tenant_id()
diff --git a/backend/app/services/agent_tools.py b/backend/app/services/agent_tools.py
index 9599fda4f..392a9b557 100644
--- a/backend/app/services/agent_tools.py
+++ b/backend/app/services/agent_tools.py
@@ -1975,7 +1975,7 @@ async def _resolve_sandbox_execution_scope(
         raise ValueError("Session sandbox execution requires a tenant")
     tenant_uuid = parse_canonical_uuid(tenant_id, label="tenant_id")
     session_uuid = parse_canonical_uuid(session_id, label="session_id")
-    chat_session = await chat_session_dao.get_active_for_agent(
+    chat_session = await chat_session_dao.get_active_for_sandbox_agent(
         tenant_id=tenant_uuid,
         agent_id=agent_id,
         session_id=session_uuid,
diff --git a/backend/tests/test_chat_session_dao.py b/backend/tests/test_chat_session_dao.py
new file mode 100644
index 000000000..63ce02c7c
--- /dev/null
+++ b/backend/tests/test_chat_session_dao.py
@@ -0,0 +1,197 @@
+"""Sandbox authorization contracts for ChatSessionDAO."""
+
+from collections import deque
+from types import SimpleNamespace
+import uuid
+
+import pytest
+from sqlalchemy.dialects import postgresql
+
+from app.dao.chat_session_dao import chat_session_dao
+
+
+class _Result:
+    def __init__(self, values=None) -> None:
+        self.values = list(values or [])
+
+    def scalar_one_or_none(self):
+        return self.values[0] if self.values else None
+
+
+class _RecordingDB:
+    def __init__(self, *results: _Result) -> None:
+        self.results = deque(results)
+        self.statements = []
+
+    async def execute(self, statement):
+        self.statements.append(statement)
+        if not self.results:
+            raise AssertionError("unexpected database query")
+        return self.results.popleft()
+
+
+def _sql(statement) -> str:
+    return str(
+        statement.compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+
+def _session(
+    *,
+    tenant_id: uuid.UUID,
+    agent_id: uuid.UUID | None,
+    session_type: str,
+    group_id: uuid.UUID | None = None,
+):
+    return SimpleNamespace(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_type=session_type,
+        group_id=group_id,
+        deleted_at=None,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("session_type", ["direct", "group"])
+async def test_sandbox_scope_preserves_exact_agent_ownership(session_type: str) -> None:
+    tenant_id = uuid.uuid4()
+    agent_id = uuid.uuid4()
+    chat_session = _session(
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_type=session_type,
+    )
+    db = _RecordingDB(_Result([chat_session]))
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_id=chat_session.id,
+        db=db,
+    )
+
+    assert result is chat_session
+    assert len(db.statements) == 1
+
+
+@pytest.mark.asyncio
+async def test_sandbox_scope_rejects_session_owned_by_another_agent() -> None:
+    tenant_id = uuid.uuid4()
+    chat_session = _session(
+        tenant_id=tenant_id,
+        agent_id=uuid.uuid4(),
+        session_type="direct",
+    )
+    db = _RecordingDB(_Result([chat_session]))
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=uuid.uuid4(),
+        session_id=chat_session.id,
+        db=db,
+    )
+
+    assert result is None
+    assert len(db.statements) == 1
+
+
+@pytest.mark.asyncio
+async def test_sandbox_scope_allows_active_native_group_agent_member() -> None:
+    tenant_id = uuid.uuid4()
+    agent_id = uuid.uuid4()
+    chat_session = _session(
+        tenant_id=tenant_id,
+        agent_id=None,
+        session_type="group",
+        group_id=uuid.uuid4(),
+    )
+    db = _RecordingDB(_Result([chat_session]), _Result([uuid.uuid4()]))
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_id=chat_session.id,
+        db=db,
+    )
+
+    assert result is chat_session
+    assert len(db.statements) == 2
+    membership_sql = _sql(db.statements[1])
+    assert "JOIN groups ON groups.id = group_members.group_id" in membership_sql
+    assert "JOIN participants ON participants.id = group_members.participant_id" in membership_sql
+    assert f"groups.tenant_id = '{tenant_id}'" in membership_sql
+    assert "groups.deleted_at IS NULL" in membership_sql
+    assert "group_members.removed_at IS NULL" in membership_sql
+    assert "participants.type = 'agent'" in membership_sql
+    assert f"participants.ref_id = '{agent_id}'" in membership_sql
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("reason", ["removed member", "deleted group", "cross-tenant group"])
+async def test_sandbox_scope_rejects_inactive_native_group_membership(reason: str) -> None:
+    tenant_id = uuid.uuid4()
+    chat_session = _session(
+        tenant_id=tenant_id,
+        agent_id=None,
+        session_type="group",
+        group_id=uuid.uuid4(),
+    )
+    db = _RecordingDB(_Result([chat_session]), _Result())
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=uuid.uuid4(),
+        session_id=chat_session.id,
+        db=db,
+    )
+
+    assert result is None, reason
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("reason", ["deleted session", "cross-tenant session"])
+async def test_sandbox_scope_rejects_inaccessible_session(reason: str) -> None:
+    tenant_id = uuid.uuid4()
+    session_id = uuid.uuid4()
+    db = _RecordingDB(_Result())
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=uuid.uuid4(),
+        session_id=session_id,
+        db=db,
+    )
+
+    assert result is None, reason
+    session_sql = _sql(db.statements[0])
+    assert f"chat_sessions.tenant_id = '{tenant_id}'" in session_sql
+    assert f"chat_sessions.id = '{session_id}'" in session_sql
+    assert "chat_sessions.deleted_at IS NULL" in session_sql
+
+
+@pytest.mark.asyncio
+async def test_sandbox_scope_rejects_malformed_owned_native_group_session() -> None:
+    tenant_id = uuid.uuid4()
+    agent_id = uuid.uuid4()
+    chat_session = _session(
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_type="group",
+        group_id=uuid.uuid4(),
+    )
+    db = _RecordingDB(_Result([chat_session]))
+
+    result = await chat_session_dao.get_active_for_sandbox_agent(
+        tenant_id=tenant_id,
+        agent_id=agent_id,
+        session_id=chat_session.id,
+        db=db,
+    )
+
+    assert result is None
+    assert len(db.statements) == 1
diff --git a/backend/tests/test_sandbox_execution_policy.py b/backend/tests/test_sandbox_execution_policy.py
index 717dd2b00..ecb459bf0 100644
--- a/backend/tests/test_sandbox_execution_policy.py
+++ b/backend/tests/test_sandbox_execution_policy.py
@@ -1,6 +1,7 @@
 """Contracts for Session-scoped sandbox policy and Redis execution leases."""
 
 import uuid
+from types import SimpleNamespace
 
 import pytest
 
@@ -154,6 +155,164 @@ async def fake_get_redis():
     assert redis.values[first.key] == "foreign-owner"
 
 
+@pytest.mark.asyncio
+async def test_same_group_session_uses_distinct_agent_leases(monkeypatch) -> None:
+    redis = FakeRedis()
+
+    async def fake_get_redis():
+        return redis
+
+    monkeypatch.setattr(execution_lease, "get_redis", fake_get_redis)
+    tenant_id = uuid.uuid4()
+    session_id = uuid.uuid4()
+    first_scope = SandboxExecutionScope(tenant_id, uuid.uuid4(), session_id)
+    second_scope = SandboxExecutionScope(tenant_id, uuid.uuid4(), session_id)
+    store = SandboxExecutionLeaseStore()
+
+    first = await store.acquire(first_scope)
+    second = await store.acquire(second_scope)
+
+    assert first is not None
+    assert second is not None
+    assert first.key != second.key
+    await first.release()
+    await second.release()
+
+
+def test_same_group_session_artifacts_remain_agent_scoped() -> None:
+    session_id = uuid.uuid4()
+    path = f"workspace/output/{session_id}/result.txt"
+    first_agent = uuid.uuid4()
+    second_agent = uuid.uuid4()
+
+    first_ref = agent_tools._workspace_artifact_ref(first_agent, path)
+    second_ref = agent_tools._workspace_artifact_ref(second_agent, path)
+
+    assert first_ref == f"workspace://{first_agent}/{path}"
+    assert second_ref == f"workspace://{second_agent}/{path}"
+    assert first_ref != second_ref
+
+
+@pytest.mark.asyncio
+async def test_authorized_native_group_scope_executes_with_isolated_output(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    tenant_id = uuid.uuid4()
+    agent_id = uuid.uuid4()
+    session_id = uuid.uuid4()
+    output_path = f"workspace/output/{session_id}/result.txt"
+    calls = []
+
+    class _Lease:
+        ownership_lost = False
+
+        async def start_heartbeat(self):
+            return None
+
+        async def ensure_publication_window(self, _seconds):
+            return True
+
+        async def release(self):
+            return None
+
+    async def tool_config(*_args):
+        return {"workspace_mode": "isolated_output"}
+
+    async def authorize(**kwargs):
+        calls.append(("authorize", kwargs))
+        return object()
+
+    async def acquire(_self, scope, **_kwargs):
+        calls.append(("lease", scope))
+        return _Lease()
+
+    async def prepare(*_args, **kwargs):
+        calls.append(("materialize", kwargs))
+        return SimpleNamespace(root=tmp_path, cleanup=lambda: None)
+
+    async def execute(_agent_id, _root, _arguments, **kwargs):
+        calls.append(("execute", kwargs))
+        return ToolExecutionOutcome("succeeded", "ok", None)
+
+    async def flush(*_args, **_kwargs):
+        return {
+            "updated": [output_path],
+            "deleted": [],
+            "conflicted": [],
+            "skipped": [],
+        }
+
+    monkeypatch.setattr(agent_tools, "_get_tool_config", tool_config)
+    monkeypatch.setattr(
+        agent_tools.chat_session_dao,
+        "get_active_for_sandbox_agent",
+        authorize,
+    )
+    monkeypatch.setattr(SandboxExecutionLeaseStore, "acquire", acquire)
+    monkeypatch.setattr(agent_tools, "_prepare_temp_workspace", prepare)
+    monkeypatch.setattr(agent_tools, "_execute_code_outcome", execute)
+    monkeypatch.setattr(agent_tools, "flush_temp_workspace", flush)
+    monkeypatch.setattr(
+        "app.config.get_sandbox_config",
+        lambda: SandboxConfig(workspace_mode="merge"),
+    )
+
+    outcome = await agent_tools._execute_code_with_workspace_outcome(
+        agent_id=agent_id,
+        tenant_id=str(tenant_id),
+        session_id=str(session_id),
+        arguments={"language": "python", "code": "print(1)"},
+        tool_name="execute_code",
+    )
+
+    assert outcome.status == "succeeded"
+    assert outcome.artifact_refs == (f"workspace://{agent_id}/{output_path}",)
+    assert [call[0] for call in calls] == ["authorize", "lease", "materialize", "execute"]
+    assert calls[0][1] == {
+        "tenant_id": tenant_id,
+        "agent_id": agent_id,
+        "session_id": session_id,
+    }
+    assert calls[1][1] == SandboxExecutionScope(tenant_id, agent_id, session_id)
+    assert calls[2][1]["publish_paths"] == [f"workspace/output/{session_id}"]
+    assert calls[3][1]["session_id"] == str(session_id)
+    assert calls[3][1]["publish_paths"] == [f"workspace/output/{session_id}"]
+
+
+@pytest.mark.asyncio
+async def test_scope_resolver_uses_sandbox_session_authorization(monkeypatch) -> None:
+    tenant_id = uuid.uuid4()
+    agent_id = uuid.uuid4()
+    session_id = uuid.uuid4()
+    calls = []
+
+    async def authorize(**kwargs):
+        calls.append(kwargs)
+        return object()
+
+    monkeypatch.setattr(
+        agent_tools.chat_session_dao,
+        "get_active_for_sandbox_agent",
+        authorize,
+    )
+
+    scope = await agent_tools._resolve_sandbox_execution_scope(
+        tenant_id=str(tenant_id),
+        agent_id=agent_id,
+        session_id=str(session_id),
+    )
+
+    assert scope == SandboxExecutionScope(tenant_id, agent_id, session_id)
+    assert calls == [
+        {
+            "tenant_id": tenant_id,
+            "agent_id": agent_id,
+            "session_id": session_id,
+        }
+    ]
+
+
 @pytest.mark.asyncio
 async def test_local_session_busy_fails_before_code(monkeypatch) -> None:
     tenant_id = uuid.uuid4()
@@ -201,6 +360,8 @@ async def forbidden_execute(*_args, **_kwargs):
 @pytest.mark.asyncio
 async def test_invalid_session_scope_fails_before_lease(monkeypatch) -> None:
     acquired = False
+    materialized = False
+    executed = False
 
     async def tool_config(*_args):
         return {"workspace_mode": "isolated_output"}
@@ -212,9 +373,19 @@ async def forbidden_acquire(*_args, **_kwargs):
         nonlocal acquired
         acquired = True
 
+    async def forbidden_materialize(*_args, **_kwargs):
+        nonlocal materialized
+        materialized = True
+
+    async def forbidden_execute(*_args, **_kwargs):
+        nonlocal executed
+        executed = True
+
     monkeypatch.setattr(agent_tools, "_get_tool_config", tool_config)
     monkeypatch.setattr(agent_tools, "_resolve_sandbox_execution_scope", invalid_scope)
     monkeypatch.setattr(SandboxExecutionLeaseStore, "acquire", forbidden_acquire)
+    monkeypatch.setattr(agent_tools, "_prepare_temp_workspace", forbidden_materialize)
+    monkeypatch.setattr(agent_tools, "_execute_code_outcome", forbidden_execute)
     monkeypatch.setattr("app.config.get_sandbox_config", lambda: SandboxConfig())
 
     outcome = await agent_tools._execute_code_with_workspace_outcome(
@@ -227,3 +398,5 @@ async def forbidden_acquire(*_args, **_kwargs):
 
     assert outcome.error_code == "sandbox_execution_scope_invalid"
     assert acquired is False
+    assert materialized is False
+    assert executed is False