Function resolve_org_token

agent/src/opentrace_agent/cli/auth.py:464–528

Get an org-scoped ``otoat_`` token, exchanging the ``otuat_`` if needed. 1. Check cache in ``~/.opentrace/org_tokens/{org_id}.json`` 2. If missing/expired, exchange ``otuat_`` + org via API 3. Cache and return Raises :class:`RuntimeError` if not logged in or exchange fails.

(org_id: str)

def resolve_org_token(org_id: str) -> dict[str, Any]:
    """Get an org-scoped ``otoat_`` token, exchanging the ``otuat_`` if needed.

    1. Check cache in ``~/.opentrace/org_tokens/{org_id}.json``
    2. If missing/expired, exchange ``otuat_`` + org via API
    3. Cache and return

    Raises :class:`RuntimeError` if not logged in or exchange fails.
    """
    from opentrace_agent.cli.credentials import load_org_token, save_org_token

    # Check cache
    cached = load_org_token(org_id)
    if cached is not None:
        return cached

    # Load user token
    user_tokens = load_tokens()
    if not user_tokens or not user_tokens.get("access_token"):
        raise RuntimeError("Not logged in. Run 'opentraceai login' first.")

    user_token = user_tokens["access_token"]

    # Handle legacy: if user has an old otoat_ token from before this change
    if user_token.startswith("otoat_"):
        raise RuntimeError(
            "Your login token is org-scoped (legacy format). "
            "Run 'opentraceai logout' then 'opentraceai login' to get a user token."
        )

    # Exchange otuat_ for org-scoped otoat_
    disco = discover()
    issuer = disco.get("issuer", ISSUER)
    exchange_url = f"{issuer}/oauth/exchange-org-token"

    body = json.dumps({"org": org_id}).encode()
    req = urllib.request.Request(
        exchange_url,
        data=body,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {user_token}",
        },
    )

    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            result = json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", errors="replace")
        if exc.code == 404:
            raise RuntimeError(
                f"Organization '{org_id}' not found. Check your .opentrace/config.yaml org value."
            ) from exc
        if exc.code == 401:
            raise RuntimeError("User token expired or invalid. Run 'opentraceai login' to re-authenticate.") from exc
        raise RuntimeError(f"Org token exchange failed ({exc.code}): {error_body}") from exc