Python: add AgentBudget for autonomous loop resource limits

python/packages/core/AGENTS.md

@@ -309,6 +309,7 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
 - **`todos_remaining(*, looping_modes=None)`** / **`todos_remaining_message`** - Helper factories for todo-driven loops (the Python counterpart of .NET's `TodoCompletionLoopEvaluator`), designed for `create_harness_agent` but usable with any agent that registers a `TodoProvider` via `context_providers`. They resolve the `TodoProvider`/`AgentModeProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `loop_should_continue`/`loop_next_message`. `todos_remaining` returns a `should_continue` predicate that loops while any todo is open; pass `looping_modes=[...]` to gate looping to specific operating modes (case-insensitive; honors the `AgentModeProvider`'s `source_id`/`available_modes`), `looping_modes=None` (default) applies in every mode, and an empty sequence raises `ValueError`. `todos_remaining_message` is a `next_message` callable that lists the still-open todo titles and tells the agent to finish them, returning `None` when the session/agent/provider is unavailable or nothing is open (in which case the middleware's default `None` handling applies: reuse the previous iteration's messages verbatim under the default `fresh_context=False`, or `DEFAULT_NEXT_MESSAGE` only when `fresh_context=True`).
 - **`background_tasks_running()`** / **`background_tasks_running_message`** - Helper factories for background-agent-driven loops, mirroring the `todos_remaining` pair. They resolve the `BackgroundAgentsProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `create_harness_agent`'s `loop_should_continue`/`loop_next_message`. `background_tasks_running` returns a `should_continue` predicate that loops while the provider's persisted state shows any task with `status == RUNNING` (pair it with `max_iterations` so the loop is bounded even if a task's persisted status is never refreshed). `background_tasks_running_message` is a `next_message` callable that lists the still-running tasks (`#<id> (<agent_name>): <description>`) and tells the agent to wait for them to finish and retrieve their results, returning `None` when the session/agent/provider is unavailable or no task is running.
   - **Approval escape hatch** - `_has_pending_approval_request(result)` checks whether an iteration's response carries a pending tool-approval request (any content with `type == "function_approval_request"`). Both the streaming and non-streaming loops stop and return that response to the caller *before* evaluating `should_continue`/`max_iterations` or injecting `next_message`, so the loop is HITL-safe even when wrapped outermost around a `ToolApprovalMiddleware` (mirrors the C# `LoopAgent`'s `HasPendingApprovalRequests`).
+  - **Budget** - `budget=AgentBudget` (a `total=False` `TypedDict`; plain dicts work) adds `max_tokens` / `max_duration` limits, validated in `AgentLoopMiddleware.__init__` and checked after each iteration before `should_continue`. `max_tokens` charges each iteration the wrapped agent response's `total_token_count`, falling back to `input_token_count + output_token_count`; the `with_judge` judge call is never charged. Sub-agent usage counts: `as_tool` reports the sub-agent's `usage_details` through a private `FunctionInvocationContext.metadata` key (`_TOOL_USAGE_CONTEXT_KEY`), which the function invocation loop moves onto the function_result and folds into the parent response exactly once (non-streaming `aggregated_usage`; streaming via a `usage` content on the tool update), so nested A->B->C counts each layer once. `BackgroundAgentsProvider` stores each finished run's usage on `BackgroundTaskInfo` (`usage_details`/`usage_reported`) and charges it once, to the parent run that first observes the completion (via the observing tool's context, or a run-scoped chat middleware when `before_run` observes it). `invoke_agent` telemetry spans still report each agent's own usage. Gaps: failed/cancelled sub-agent runs produce no response (tokens lost), background tasks never observed as finished or abandoned at `release_session` are not counted, and custom tools that call `agent.run` directly are not counted. The accounting is not configurable.
   - **Harness integration** - `create_harness_agent` enables the loop when a `loop_should_continue` callable is passed; it prepends `AgentLoopMiddleware(loop_should_continue, max_iterations=loop_max_iterations, next_message=loop_next_message)` ahead of `ToolApprovalMiddleware` so the loop is the outermost middleware (each iteration is a full agent run including tool approval, and the escape hatch hands pending approvals back to the caller). `loop_next_message` and `loop_max_iterations` only take effect together with `loop_should_continue` (with no `loop_should_continue` there is no loop, so they are ignored); `loop_max_iterations` defaults to the loop's default cap (`None` → unbounded).
 
 ### Workflows (`_workflows/`)

python/packages/core/agent_framework/__init__.py

@@ -132,6 +132,7 @@
         "FileMemoryProvider",
     ),
     "._harness._loop": (
+        "AgentBudget",
         "AgentLoopMiddleware",
         "JudgeVerdict",
         "background_tasks_running",
@@ -432,6 +433,7 @@
     "USER_AGENT_KEY",
     "USER_AGENT_TELEMETRY_DISABLED_ENV_VAR",
     "Agent",
+    "AgentBudget",
     "AgentContext",
     "AgentEvalConverter",
     "AgentExecutor",

python/packages/core/agent_framework/__init__.pyi

@@ -90,6 +90,7 @@ from ._harness._file_access import (
 )
 from ._harness._file_memory import DEFAULT_FILE_MEMORY_INSTRUCTIONS, DEFAULT_FILE_MEMORY_SOURCE_ID, FileMemoryProvider
 from ._harness._loop import (
+    AgentBudget,
     AgentLoopMiddleware,
     JudgeVerdict,
     background_tasks_running,
@@ -387,6 +388,7 @@ __all__ = [
     "USER_AGENT_KEY",
     "USER_AGENT_TELEMETRY_DISABLED_ENV_VAR",
     "Agent",
+    "AgentBudget",
     "AgentContext",
     "AgentEvalConverter",
     "AgentExecutor",

python/packages/core/agent_framework/_agents.py

@@ -687,6 +687,11 @@ def as_tool(
             A FunctionTool that can be used as a tool by other agents.
 
         Note:
+            The sub-agent's response ``usage_details`` (including its own sub-agents) is rolled into
+            the calling agent's response ``usage_details`` when the tool runs inside the function
+            invocation loop, also when the sub-agent stops on an approval or user-input request. A
+            sub-agent run that fails produces no response, so its usage is not counted.
+
             Child function approvals are not propagated into the calling agent.
             Configure ToolApprovalMiddleware with runtime auto-approval rules on
             the child for immediate policy decisions. Use a workflow when approval
@@ -747,6 +752,8 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str:
                 ctx: the function invocation context used
                 **kwargs: only used to dynamically load the argument that is defined for this tool.
             """
+            from ._tools import _record_tool_usage  # pyright: ignore[reportPrivateUsage]
+
             parent_session = ctx.session
             session = AgentSession()
             child_approval_source_ids = _tool_approval_source_ids(self.middleware)
@@ -833,6 +840,10 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str:
                         if isawaitable(callback_result):
                             await callback_result
                 final_response = await stream.get_final_response()
+                # Report the sub-agent's token usage (including its own sub-agents) to the calling loop,
+                # which folds it into the parent response's usage_details exactly once. Recorded before
+                # the approval/user-input raises below so that usage is not lost on those paths.
+                _record_tool_usage(ctx, final_response.usage_details)
                 approval_requests = [
                     request
                     for request in final_response.user_input_requests