OpenAI Assistants API budget control: cap spend per run, thread, and user

• GPT-4o input: $2.50/M tokens; output: $10.00/M tokens
• Turn 1 (system 500 + user 100 tokens): 600 input + ~200 output = $0.00350
• Turn 10 (10×300 avg history + 100 tokens): 3,100 input + ~200 output = $0.00975
• Turn 30 (30×300 history + 100 tokens): 9,100 input + ~200 output = $0.02475
• 30-turn conversation total (rough): ~$0.35 for moderate-length messages

from openai import OpenAI
from openai.types.beta.threads import Run

client = OpenAI()

def create_run_with_token_cap(
    thread_id: str,
    assistant_id: str,
    max_prompt_tokens: int = 8_000,  # cap thread context
    max_completion_tokens: int = 1_024,
) -> Run:
    """
    Create an Assistants Run with explicit token caps.
    max_prompt_tokens: total input token budget for this run (thread + system + tools)
    max_completion_tokens: output token cap for this run
    The API will truncate old messages from the thread to stay within max_prompt_tokens
    using the 'auto' truncation strategy (drops oldest messages first).
    """
    return client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant_id,
        max_prompt_tokens=max_prompt_tokens,
        max_completion_tokens=max_completion_tokens,
        truncation_strategy={
            "type": "auto",
            # Alternatively: "last_messages" with last_messages=N to keep
            # only the N most recent messages regardless of token count
        },
    )

def estimate_run_cost_usd(run: Run) -> float:
    """Calculate USD cost from run.usage field."""
    if not run.usage:
        return 0.0
    GPT4O_IN_PER_TOK  = 2.50  / 1_000_000
    GPT4O_OUT_PER_TOK = 10.00 / 1_000_000
    return (run.usage.prompt_tokens * GPT4O_IN_PER_TOK +
            run.usage.completion_tokens * GPT4O_OUT_PER_TOK)

import sqlite3
from datetime import datetime, timezone
from openai import OpenAI

client = OpenAI()

GPT4O_IN  = 2.50  / 1_000_000
GPT4O_OUT = 10.00 / 1_000_000

def init_budget_db(db_path: str) -> None:
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS run_costs (
                run_id TEXT PRIMARY KEY,
                thread_id TEXT NOT NULL,
                user_id TEXT NOT NULL,
                prompt_tokens INTEGER,
                completion_tokens INTEGER,
                cost_usd REAL,
                created_at TEXT
            )
        """)

def record_run_cost(
    db_path: str,
    run,  # openai.types.beta.threads.Run
    user_id: str,
) -> float:
    """Record a completed run's cost and return the cost in USD."""
    if not run.usage:
        return 0.0
    cost = (run.usage.prompt_tokens * GPT4O_IN +
            run.usage.completion_tokens * GPT4O_OUT)
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            "INSERT OR REPLACE INTO run_costs "
            "(run_id, thread_id, user_id, prompt_tokens, completion_tokens, cost_usd, created_at) "
            "VALUES (?,?,?,?,?,?,?)",
            (run.id, run.thread_id, user_id,
             run.usage.prompt_tokens, run.usage.completion_tokens,
             cost, datetime.now(timezone.utc).isoformat()),
        )
    return cost

def get_user_spend_today(db_path: str, user_id: str) -> float:
    """Return total USD spent by user_id today (UTC)."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            "SELECT COALESCE(SUM(cost_usd),0) FROM run_costs "
            "WHERE user_id=? AND created_at LIKE ?",
            (user_id, f"{today}%"),
        ).fetchone()
        return row[0] if row else 0.0

def run_with_budget_check(
    thread_id: str,
    assistant_id: str,
    user_id: str,
    db_path: str,
    daily_cap_usd: float = 1.00,
) -> str:
    """
    Run the assistant only if the user is within their daily budget cap.
    Returns the assistant's text response or a budget-exceeded message.
    """
    # Pre-check: is user within daily budget?
    spent = get_user_spend_today(db_path, user_id)
    if spent >= daily_cap_usd:
        return (
            f"[BUDGET] Daily limit ${daily_cap_usd:.2f} reached "
            f"(used ${spent:.4f}). Try again tomorrow."
        )

    run = create_run_with_token_cap(thread_id, assistant_id)
    # Poll until terminal state
    while run.status in ("queued", "in_progress", "requires_action"):
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id, run_id=run.id
        )
    cost = record_run_cost(db_path, run, user_id)

    # Post-check: did this run push user over cap?
    if spent + cost > daily_cap_usd:
        return (
            f"[BUDGET] Daily limit ${daily_cap_usd:.2f} reached after this run "
            f"(used ${spent + cost:.4f}). Usage recorded."
        )

    messages = client.beta.threads.messages.list(thread_id=thread_id)
    return messages.data[0].content[0].text.value

from openai import OpenAI
from runguard import LoopDetector, LoopDetectedError

client = OpenAI()
detector = LoopDetector(repeats=3, max_cycle_len=3)

def handle_run_with_loop_detection(
    thread_id: str,
    run_id: str,
) -> str:
    """
    Poll a Run to completion, using RunGuard to detect tool-call loops.
    Returns the final assistant message or a loop-detected error.
    """
    while True:
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id, run_id=run_id
        )
        if run.status == "completed":
            messages = client.beta.threads.messages.list(thread_id=thread_id)
            return messages.data[0].content[0].text.value

        if run.status == "requires_action":
            tool_calls = run.required_action.submit_tool_outputs.tool_calls
            tool_outputs = []

            for tc in tool_calls:
                # Build a canonical signature for loop detection
                sig = f"{tc.function.name}:{tc.function.arguments[:100]}"
                match = detector.record(sig)
                if match:
                    # Cancel the run and return a safe error
                    client.beta.threads.runs.cancel(
                        thread_id=thread_id, run_id=run_id
                    )
                    return (
                        f"[LOOP] Tool '{tc.function.name}' called in a repeating "
                        f"pattern ({match.repeats}x). Run cancelled to prevent cost overrun."
                    )

                # Execute tool and collect output
                output = dispatch_tool(tc.function.name, tc.function.arguments)
                tool_outputs.append({"tool_call_id": tc.id, "output": output})

            # Submit all tool outputs to continue the run
            client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run_id,
                tool_outputs=tool_outputs,
            )

        elif run.status in ("failed", "cancelled", "expired"):
            return f"[ERROR] Run ended with status: {run.status}"

        # Brief polling delay
        import time; time.sleep(0.5)

def dispatch_tool(name: str, arguments_json: str) -> str:
    """Replace with real tool dispatch."""
    import json
    args = json.loads(arguments_json)
    return f"Tool {name} result: {args}"