open-swe/agent/dashboard/team_settings.py

"""Team-wide Open SWE Review (Bugbot) settings stored in LangGraph Store.

A single record keyed ``"default"`` keeps all instance-wide reviewer
configuration in one place. Per-repo style prompts live in
:mod:`agent.dashboard.review_styles`.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any, Literal

from langgraph_sdk import get_client
from pydantic import BaseModel, model_validator

from .options import SUPPORTED_MODEL_IDS, default_model_pair, model_supports_effort

logger = logging.getLogger(__name__)

TEAM_SETTINGS_NAMESPACE: list[str] = ["team_settings"]
TEAM_SETTINGS_KEY = "default"

TriggerMode = Literal["every_push", "once_per_pr", "manual"]
AutofixMode = Literal["off", "low", "medium", "high"]


class TeamSettingsUpdate(BaseModel):
    trigger_mode: TriggerMode = "every_push"
    review_draft_prs: bool = False
    pr_summaries: bool = True
    autofix_mode: AutofixMode = "off"
    autofix_severity_threshold: AutofixMode = "medium"
    default_agent_model: str | None = None
    default_agent_reasoning_effort: str | None = None
    default_reviewer_model: str | None = None
    default_reviewer_reasoning_effort: str | None = None

    @model_validator(mode="after")
    def _validate_model_pairs(self) -> TeamSettingsUpdate:
        _validate_model_effort_pair(
            self.default_agent_model, self.default_agent_reasoning_effort, "agent"
        )
        _validate_model_effort_pair(
            self.default_reviewer_model, self.default_reviewer_reasoning_effort, "reviewer"
        )
        return self


def _validate_model_effort_pair(model: str | None, effort: str | None, role: str) -> None:
    if model is None and effort is None:
        return
    if model is None:
        raise ValueError(f"{role} reasoning effort set without a model")
    if model not in SUPPORTED_MODEL_IDS:
        raise ValueError(f"unsupported {role} model: {model}")
    if effort is None or not model_supports_effort(model, effort):
        raise ValueError(f"effort {effort!r} not supported by {role} model {model!r}")


def _client():
    return get_client()


def _default_settings() -> dict[str, Any]:
    fallback_model, fallback_effort = default_model_pair()
    return {
        "trigger_mode": "every_push",
        "review_draft_prs": False,
        "pr_summaries": True,
        "autofix_mode": "off",
        "autofix_severity_threshold": "medium",
        "default_agent_model": fallback_model,
        "default_agent_reasoning_effort": fallback_effort,
        "default_reviewer_model": fallback_model,
        "default_reviewer_reasoning_effort": fallback_effort,
        "updated_at": None,
    }


async def get_team_settings() -> dict[str, Any]:
    defaults = _default_settings()
    try:
        item = await _client().store.get_item(TEAM_SETTINGS_NAMESPACE, TEAM_SETTINGS_KEY)
    except Exception as e:
        logger.debug("team settings lookup failed: %s", e)
        return defaults
    if item is None:
        return defaults
    value = item.get("value") if isinstance(item, dict) else getattr(item, "value", None)
    if not isinstance(value, dict):
        return defaults
    # Skip None-valued model fields so legacy records (or PUTs that cleared the
    # selection) still surface the hardcoded default instead of a null.
    overlay = {k: v for k, v in value.items() if v is not None}
    merged = {**defaults, **overlay}
    # Drop obsolete trigger mode values so a legacy record doesn't surface a
    # value the new TriggerMode literal would reject on the next PUT.
    if merged.get("trigger_mode") not in {"every_push", "once_per_pr", "manual"}:
        merged["trigger_mode"] = defaults["trigger_mode"]
    return merged


async def upsert_team_settings(update: TeamSettingsUpdate) -> dict[str, Any]:
    value: dict[str, Any] = {
        "trigger_mode": update.trigger_mode,
        "review_draft_prs": update.review_draft_prs,
        "pr_summaries": update.pr_summaries,
        "autofix_mode": update.autofix_mode,
        "autofix_severity_threshold": update.autofix_severity_threshold,
        "default_agent_model": update.default_agent_model,
        "default_agent_reasoning_effort": update.default_agent_reasoning_effort,
        "default_reviewer_model": update.default_reviewer_model,
        "default_reviewer_reasoning_effort": update.default_reviewer_reasoning_effort,
        "updated_at": datetime.now(UTC).isoformat(),
    }
    await _client().store.put_item(TEAM_SETTINGS_NAMESPACE, TEAM_SETTINGS_KEY, value)
    return value


async def get_team_default_model(
    role: Literal["agent", "reviewer"],
) -> tuple[str, str]:
    """Return the team-wide default ``(model_id, reasoning_effort)`` for ``role``.

    Always returns a valid pair: the admin-configured pair if set, otherwise the
    hardcoded fallback from :func:`agent.dashboard.options.default_model_pair`.
    Invalid stored pairs (unsupported model or mismatched effort) fall back to
    the hardcoded default rather than propagating bad data.
    """
    settings = await get_team_settings()
    if role == "agent":
        model = settings.get("default_agent_model")
        effort = settings.get("default_agent_reasoning_effort")
    else:
        model = settings.get("default_reviewer_model")
        effort = settings.get("default_reviewer_reasoning_effort")
    if (
        isinstance(model, str)
        and model in SUPPORTED_MODEL_IDS
        and isinstance(effort, str)
        and model_supports_effort(model, effort)
    ):
        return model, effort
    return default_model_pair()