Newer
Older
gnexus-tasks / backend / app / actor.py
"""Актор действия: кто и через что сделал (ТЗ 3.20).

Токен MCP — принципал владельца, поэтому по умолчанию действие с Bearer-токеном
считается **агентским**: иначе отличить агента от браузерного расширения нечем.
Объявить действие «от имени владельца» можно флагом `is_user` (MCP-инструменты)
или заголовком `X-Actor: user` (REST): актор станет `user`, но канал (`via`)
сохранится — в журнале видно, что действие заявлено владельцем и пришло от агента.

Это не разграничение прав: держатель токена и так действует как владелец. Актор
нужен, чтобы запись в журнале отвечала на вопрос «кто это сделал» (ТЗ 3.20, 5).
"""

from dataclasses import dataclass
from typing import Any

ACTOR_USER = "user"
ACTOR_AGENT = "agent"

VIA_UI = "ui"
VIA_API = "api"
VIA_MCP = "mcp"

# Заголовок REST, которым клиент объявляет действие от имени владельца
# (браузерное расширение — руки владельца, его захваты не агентские, ТЗ 3.18)
ACTOR_HEADER = "x-actor"
_DECLARED_VALUES = {"user", "me"}


@dataclass(frozen=True)
class Actor:
    """Кто выполняет действие: владелец или ИИ-агент, и через какой канал."""

    kind: str = ACTOR_USER
    # Подпись для журнала: имя владельца или label токена (label может быть пустым —
    # тогда интерфейс подставит своё «агент», локализованное)
    name: str = ""
    via: str = VIA_UI
    # Токен, которым пришёл агент: сравнивать аренду можно только по id — label не уникален
    token_id: int | None = None
    # Имя владельца: чем подписать действие, если агент объявил его своим (is_user)
    owner_name: str = ""

    @property
    def is_agent(self) -> bool:
        return self.kind == ACTOR_AGENT

    def declaring_user(self) -> "Actor":
        """Действие, объявленное от имени владельца: подпись — его имя, канал тот же."""
        if self.kind == ACTOR_USER:
            return self
        return Actor(
            kind=ACTOR_USER,
            name=self.owner_name,
            via=self.via,
            token_id=self.token_id,
            owner_name=self.owner_name,
        )


def agent_actor(label: str, token_id: int | None, owner_name: str, via: str) -> Actor:
    """Актор по токену: подпись — label токена («Нави»), а не имя владельца."""
    return Actor(
        kind=ACTOR_AGENT,
        name=label.strip(),
        via=via,
        token_id=token_id,
        owner_name=owner_name,
    )


def actor_from_request(request: Any) -> Actor:
    """Актор, положенный авторизацией в `request.state.gntodo_actor` (ТЗ 3.20).

    Сведений нет — считаем действие владельца: так ведут себя тесты, подменяющие
    авторизацию через dependency_overrides (и любой запрос мимо обеих схем входа,
    а их всего две: сессия SSO и Bearer).
    """
    actor = getattr(getattr(request, "state", None), "gntodo_actor", None)
    return actor if isinstance(actor, Actor) else Actor()


def declares_user(headers: Any) -> bool:
    """Объявлено ли действие владельца заголовком REST (`X-Actor: user`)."""
    try:
        value = headers.get(ACTOR_HEADER, "")
    except AttributeError:  # нет заголовков — например, синтетический запрос в тестах
        return False
    return str(value).strip().lower() in _DECLARED_VALUES