IterationBudget (agent/budget.py)

迭代预算：给一次对话设「最多调多少次 LLM」的上限，防止失控烧钱。 是什么：一个线程安全的计数器。每调一次 LLM API 就扣 1 个名额； 名额用完之前，通过 grace_call（宽限发言）给模型最后一次机会把话说完 收尾，而不是戛然而止。 在项目里的位置：由 AIAgent 主循环消费——每轮先 consume() 问「还有名额吗」， 没有就收尾。程序化的工具调用（如 execute_code）不算模型主动决策， 可以用 refund() 把名额退回来。

# Source: https://github.com/Harvil1/codeAgent | License: MIT | Copyright 2026 Harvil1
# Provided via justclaw.it.com; keep this notice with the code.
"""迭代预算：给一次对话设「最多调多少次 LLM」的上限，防止失控烧钱。

是什么：一个线程安全的计数器。每调一次 LLM API 就扣 1 个名额；
名额用完之前，通过 grace_call（宽限发言）给模型最后一次机会把话说完
收尾，而不是戛然而止。

在项目里的位置：由 AIAgent 主循环消费——每轮先 consume() 问「还有名额吗」，
没有就收尾。程序化的工具调用（如 execute_code）不算模型主动决策，
可以用 refund() 把名额退回来。
"""

import threading


class IterationBudget:
    """线程安全的迭代预算计数器（加锁是因为主循环和后台线程都会碰它）。"""

    def __init__(self, total: int):
        """初始化预算。

        参数：
            total —— 总名额数（负数会被压到 0，即「一步都不许走」）
        """
        self._total = max(0, int(total))
        self._consumed = 0
        self._lock = threading.Lock()

    @property
    def total(self) -> int:
        """总名额数。"""
        return self._tot

... (truncated -- full source via MCP)