Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions backend/app/core/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from sqlalchemy import false, or_, select, exists

from app.models.agent import Agent, AgentPermission
from app.models.org import AgentAgentRelationship, AgentRelationship, OrgMember
from app.models.org import OrgMember
from app.models.user import User


Expand Down Expand Up @@ -178,10 +178,9 @@ def evaluate_roster_agent_visibility(
visible = False

if source_mode == "private":
visible = (
target_mode == "private"
and getattr(source_agent, "creator_id", None) == getattr(target_agent, "creator_id", None)
)
# Private Agents may initiate contact with company-visible Agents, but
# never expose another private Agent (including one with the same owner).
visible = target_mode == "company"
else:
visible = target_mode == "company" or (target_mode == "custom" and authorized_custom_target)

Expand Down
7 changes: 3 additions & 4 deletions backend/app/services/agent_directory.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,10 +247,9 @@ def _agent_directory_conditions(
AgentModel.id != source.id,
]
if source_mode == "private":
conditions.extend([
AgentModel.access_mode == "private",
AgentModel.creator_id == source.creator_id,
])
# Keep the query in sync with evaluate_roster_agent_visibility:
# a private Agent can only initiate contact with company-visible Agents.
conditions.append(AgentModel.access_mode == "company")
else:
conditions.append(or_(
AgentModel.access_mode == "company",
Expand Down
26 changes: 22 additions & 4 deletions backend/tests/test_agent_visibility.py
Original file line number Diff line number Diff line change
Expand Up @@ -254,17 +254,35 @@ def test_evaluate_roster_human_visibility_limits_custom_to_authorized_members():
assert authorized_custom_visibility.can_contact is True


def test_evaluate_roster_agent_visibility_allows_same_creator_private_only():
def test_evaluate_roster_agent_visibility_enforces_private_to_company_only():
tenant_id = uuid.uuid4()
creator_id = uuid.uuid4()
source = make_agent(tenant_id=tenant_id, creator_id=creator_id, access_mode="private")
same_creator_private = make_agent(tenant_id=tenant_id, creator_id=creator_id, access_mode="private")
other_private = make_agent(tenant_id=tenant_id, creator_id=uuid.uuid4(), access_mode="private")
company_agent = make_agent(tenant_id=tenant_id, creator_id=creator_id, access_mode="company")

assert permissions.evaluate_roster_agent_visibility(source, same_creator_private).visible is True
assert permissions.evaluate_roster_agent_visibility(source, other_private).visible is False
assert permissions.evaluate_roster_agent_visibility(source, company_agent).visible is False
same_owner_private = permissions.evaluate_roster_agent_visibility(source, same_creator_private)
other_owner_private = permissions.evaluate_roster_agent_visibility(source, other_private)
company_visibility = permissions.evaluate_roster_agent_visibility(source, company_agent)

assert same_owner_private.visible is False
assert same_owner_private.can_contact is False
assert other_owner_private.visible is False
assert other_owner_private.can_contact is False
assert company_visibility.visible is True
assert company_visibility.can_contact is True


def test_evaluate_roster_agent_visibility_blocks_company_to_private():
tenant_id = uuid.uuid4()
source = make_agent(tenant_id=tenant_id, access_mode="company")
private_target = make_agent(tenant_id=tenant_id, access_mode="private")

visibility = permissions.evaluate_roster_agent_visibility(source, private_target)

assert visibility.visible is False
assert visibility.can_contact is False


def test_evaluate_roster_agent_visibility_reports_uncontactable_reason():
Expand Down
23 changes: 23 additions & 0 deletions backend/tests/test_query_directory_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,29 @@ async def test_query_directory_agent_list_uses_sql_offset_and_limit():
assert "OFFSET" in statement


@pytest.mark.asyncio
async def test_private_agent_directory_queries_company_agents_only():
tenant_id = uuid.uuid4()
source = _make_agent(tenant_id=tenant_id, access_mode="private")
db = RecordingDB(
responses=[
DummyResult(scalar_value=source),
DummyResult(values=[]),
]
)

result = await agent_directory.query_agent_directory(
db,
source_agent_id=source.id,
member_type="agent",
)

assert result["members"] == []
compiled = db.statements[1].compile()
assert "agents.access_mode = :access_mode_1" in str(compiled)
assert compiled.params["access_mode_1"] == "company"


def test_format_roster_agent_returns_stable_id_and_contact_tool():
tenant_id = uuid.uuid4()
source = _make_agent(tenant_id=tenant_id)
Expand Down