From 585f2b3bcf717a789a008e8097a70bae9f86844a Mon Sep 17 00:00:00 2001
From: Y1fe1Zh0u <zhouyifei210@gmail.com>
Date: Fri, 14 Aug 2026 15:00:07 +0800
Subject: [PATCH] Permit authorized native-group Agents to execute isolated
 code

Native group Sessions are intentionally shared and therefore have no owning Agent. Authorize their sandbox scope through the active tenant Group and active Agent participant membership while preserving exact ownership for Direct and external-channel Sessions.

Constraint: Native group Sessions intentionally have chat_sessions.agent_id NULL

Rejected: Backfill a group Session Agent | shared Sessions can host multiple Agents

Rejected: Allow every group Session | would bypass tenant Group and active membership checks

Confidence: high

Scope-risk: narrow

Reversibility: clean

Directive: Keep Direct and external-group exact Agent ownership checks intact

Tested: 155 scoped DAO, group chat, Runtime, Sandbox, subprocess, and workspace tests

Tested: Ruff passed for changed DAO and policy test files

Not-tested: 3010 live group execution before deployment
---
 backend/app/dao/chat_session_dao.py           |  48 +++++
 backend/app/services/agent_tools.py           |   2 +-
 backend/tests/test_chat_session_dao.py        | 197 ++++++++++++++++++
 .../tests/test_sandbox_execution_policy.py    | 173 +++++++++++++++
 4 files changed, 419 insertions(+), 1 deletion(-)
 create mode 100644 backend/tests/test_chat_session_dao.py

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