OSVauco/agents/core-logic/agent.py

#!/usr/bin/env python3
"""
agent.py — OSVauco OPAX agent.

Modes:
  light  — gemini-2.5-flash, $1/task
  heavy  — gemini-2.5-pro,   $3/task

Legacy 'A' / 'A+' normaliseres automatisk.
Autoriserte brukere for heavy: opax, admin

Gjeldende modell-tilgjengelighet (mai 2026):
  gemini-2.0-flash-001 kun for eksisterende kunder — bruk IKKE.
  gemini-2.5-flash / gemini-2.5-pro krever GOOGLE_CLOUD_LOCATION=global.

NOTE: RAG er midlertidig deaktivert (CI6) — blokkerte AFC når kombinert med
FunctionTools. Legges tilbake som dedikert sub-agent når RAG-korpuset er
oppdatert (OQ-RAG backlog).

CG4-budget (ny):
  - Flash-first routing: heavy-kall under FLASH_THRESHOLD rutes til Flash
  - Context-trimming: maks OPAX_CONTEXT_TURNS turns beholdes i historikk
  - Output-tak: MAX_OUTPUT_TOKENS satt på alle kall
  - TokenBudgetExceeded kastes ved for stor input
"""

import asyncio
import os
import logging
import uuid
from typing import Literal

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types

logger = logging.getLogger(__name__)

PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
LOCATION   = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")

ORCHESTRATOR_MODEL = os.environ.get("ORCHESTRATOR_MODEL", "gemini-2.5-flash")
SUBAGENT_MODEL     = os.environ.get("SUBAGENT_MODEL",     "gemini-2.5-flash")
REASONING_MODEL    = os.environ.get("REASONING_MODEL",    "gemini-2.5-flash")

HEAVY_ORCHESTRATOR = os.environ.get("HEAVY_ORCHESTRATOR_MODEL", "gemini-2.5-pro")
HEAVY_SUBAGENT     = os.environ.get("HEAVY_SUBAGENT_MODEL",     "gemini-2.5-pro")
HEAVY_REASONING    = os.environ.get("HEAVY_REASONING_MODEL",    "gemini-2.5-flash")

BUDGET_LIGHT = float(os.environ.get("BUDGET_A_USD_PER_TASK",         "1.0"))
BUDGET_HEAVY = float(os.environ.get("HEAVY_MODE_BUDGET_USD_PER_DAY", "3.0"))

HEAVY_MODE_ALLOWED_USERS = ["opax", "admin"]

Mode = Literal["light", "heavy"]
APP_NAME = "opax"

try:
    from token_logger import log_token_usage, create_bq_table_if_not_exists
except ImportError:
    def log_token_usage(*args, **kwargs):
        pass
    def create_bq_table_if_not_exists():
        pass

try:
    import sys as _sys, pathlib as _pathlib
    _sys.path.insert(0, str(_pathlib.Path(__file__).parent.parent.parent / "ml"))
    from token_budget import trim_context, route_model, TokenBudgetExceeded, budget_summary
    logger.info(f"[agent] token_budget lastet: {budget_summary()}")
except ImportError as _e:
    logger.warning(f"[agent] token_budget ikke tilgjengelig: {_e} — kjører uten trimming")
    def trim_context(history, system_prompt="", max_tokens=32000):
        return history[-10:] if len(history) > 10 else history
    def route_model(message, mode, flash_model, pro_model):
        return flash_model if mode != "heavy" else pro_model
    class TokenBudgetExceeded(Exception):
        pass

create_bq_table_if_not_exists()


def _normalize_mode(mode: str) -> str:
    mapping = {"A": "light", "A+": "heavy", "light": "light", "heavy": "heavy"}
    if mode not in mapping:
        raise ValueError(
            f"Invalid mode '{mode}'. Must be 'light' or 'heavy' "
            f"(also accepts legacy 'A' / 'A+')."
        )
    return mapping[mode]


def authorize_mode(user_id: str, mode: str) -> None:
    mode = _normalize_mode(mode)
    if mode == "heavy" and user_id not in HEAVY_MODE_ALLOWED_USERS:
        raise PermissionError(
            f"User '{user_id}' is not authorized for heavy mode. "
            f"Authorized: {HEAVY_MODE_ALLOWED_USERS}"
        )


# —— OPAX-MCP FunctionTools (feiler stille) ——
opax_tools = []
try:
    import sys, pathlib
    sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "tools"))
    from mcp_tools import get_opax_tools
    opax_tools = get_opax_tools()
    logger.info(f"OPAX-MCP tools lastet: {[t.name for t in opax_tools]}")
except Exception as e:
    logger.warning(f"OPAX-MCP tools ikke tilgjengelig (CI3): {e}")


def get_models_for_mode(mode: Mode) -> dict:
    if mode == "heavy":
        return {
            "orchestrator": HEAVY_ORCHESTRATOR,
            "subagent":     HEAVY_SUBAGENT,
            "reasoning":    HEAVY_REASONING,
            "budget_usd":   BUDGET_HEAVY,
            "multi_agent":  True,
        }
    return {
        "orchestrator": ORCHESTRATOR_MODEL,
        "subagent":     SUBAGENT_MODEL,
        "reasoning":    REASONING_MODEL,
        "budget_usd":   BUDGET_LIGHT,
        "multi_agent":  False,
    }


# Kompakt system-prompt (redusert fra ~800 til ~350 tokens)
_INSTRUCTION_TEMPLATE = (
    "Du er OPAX — OSVauco AI-agent (Jason). Modus: {mode}. "
    "Modell: {model}. Budsjett: ${budget}/oppgave. "
    "GCP-kostnader: /billing/summary, /billing/forecast, /billing/anomalies. "
    "Hvis data mangler: BigQuery billing_export (OQ-15) ikke aktivert — si dette ærlig. "
    "HITL: ikke kjør terraform/onboard/deploy uten Chris. "
    "MCP-tools: get_build_status, get_logs, push_static, deploy_service. "
    "Svar på norsk (bokmål) med mindre annet er bedt om."
)


def build_agent(mode: str = "light", message: str = "") -> tuple:
    """
    Returnerer (agent, actual_model, instruction) der actual_model er
    Flash-first-rutet basert på meldingslængde.
    """
    mode = _normalize_mode(mode)
    models = get_models_for_mode(mode)

    # CG4-budget: Flash-first routing
    actual_model = route_model(
        message=message,
        mode=mode,
        flash_model=ORCHESTRATOR_MODEL,
        pro_model=models["orchestrator"],
    )

    instruction = _INSTRUCTION_TEMPLATE.format(
        mode=mode,
        model=actual_model,
        budget=models["budget_usd"],
    )

    agent = Agent(
        model=actual_model,
        name="opax_agent",
        description=f"OPAX — OSVauco enterprise agent [{mode}]",
        instruction=instruction,
        tools=opax_tools,
    )
    return agent, actual_model, instruction


root_agent, _, _ = build_agent(mode="light")


async def _run_async(
    message: str,
    user_id: str,
    session_id: str,
    mode: str,
    caller_type: str = "agent",
) -> str:
    mode = _normalize_mode(mode)
    agent, actual_model, instruction = build_agent(mode=mode, message=message)
    models = get_models_for_mode(mode)
    module_name = f"jason/{mode}"

    # CG4-budget: trim melding hvis enkeltmelding er for lang
    try:
        trim_context([types.Content(role="user", parts=[types.Part(text=message)])], system_prompt=instruction)
    except TokenBudgetExceeded as e:
        return f"⚠️ Token-budsjett overskredet: {e}"

    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name=APP_NAME, user_id=user_id, session_id=session_id,
    )
    runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)
    new_message = types.Content(role="user", parts=[types.Part(text=message)])
    final_text = ""
    input_tokens = 0
    output_tokens = 0
    request_id = str(uuid.uuid4())

    async for event in runner.run_async(
        user_id=user_id, session_id=session.id, new_message=new_message,
    ):
        if event.is_final_response() and event.content and event.content.parts:
            final_text = event.content.parts[0].text or ""
        if hasattr(event, "usage_metadata") and event.usage_metadata:
            um = event.usage_metadata
            input_tokens  += getattr(um, "prompt_token_count", 0) or 0
            output_tokens += getattr(um, "candidates_token_count", 0) or 0

    # Logg faktisk modell brukt (ikke nødvendigvis Pro selv i heavy-mode)
    if input_tokens > 0 or output_tokens > 0:
        log_token_usage(
            agent_name=module_name,
            model_name=actual_model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            request_id=request_id,
            module_name=module_name,
            caller_type=caller_type,
            session_id=session_id,
        )

    logger.info(
        f"[agent] {mode}/{actual_model} in={input_tokens} out={output_tokens} "
        f"caller={caller_type} session={session_id}"
    )

    return final_text


def run(
    message: str,
    user_id: str = "opax",
    session_id: str = "default",
    mode: str = "light",
    caller_type: str = "agent",
) -> str:
    mode = _normalize_mode(mode)
    authorize_mode(user_id, mode)
    return asyncio.run(_run_async(
        message=message,
        user_id=user_id,
        session_id=session_id,
        mode=mode,
        caller_type=caller_type,
    ))


if __name__ == "__main__":
    import sys
    logging.basicConfig(level=logging.WARNING)
    query       = sys.argv[1] if len(sys.argv) > 1 else "Hva er OPAX?"
    mode        = sys.argv[2] if len(sys.argv) > 2 else "light"
    caller_type = sys.argv[3] if len(sys.argv) > 3 else "cli"
    print(f"Mode: {mode} | Caller: {caller_type} | Query: {query}")
    print("-" * 60)
    print(run(message=query, user_id="opax", mode=mode, caller_type=caller_type))