agno-agi/agno
1
0
Fork 0
mirror of https://github.com/agno-agi/agno.git synced 2026-10-11 22:09:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: create a separate approval record for each approval-required tool call in a run (#10812)

Browse Source 
## Summary

Each approval-required (`@approval`) tool call in a run now gets its own
approval record. This applies to agents and teams.

### The issue

When a run paused a second time for a new `@approval` tool call, no
approval record was created for that call. The next continue then ran it
using the approval resolved for the first call.

Example with `pay_invoice` decorated with `@approval` +
`@tool(requires_confirmation=True)`:

1. The run pauses on `pay_invoice(INV-1)`, and one pending record is
created.
2. The admin approves it, and the run is continued by `run_id` with no
requirements (the documented database-resolution flow).
3. INV-1 is paid. The model calls `pay_invoice(INV-2)`, and the run
pauses again.
4. No pending record exists for INV-2, so there is nothing for the admin
to approve.
5. Continuing the run pays INV-2 anyway. The run has one approval
record, and it names INV-1 only.

There were two causes, both in `agno/run/approval.py`:

- **No record for the second call.** `create_approval_from_pause`
skipped creating a record if any required tool in `run_response.tools`
already had an `approval_id`. That list grows across every pause in a
run, so the earlier call (already approved and run) made the check
return the old id.
- **The old record was reused.** The new call had no `approval_id`, so
`_group_tools_by_approval` used the run-level fallback. That fallback
returned the run's latest record, which was the approved INV-1 record,
and the new call was confirmed with it.

### The fix

- **`create_approval_from_pause` / `acreate_approval_from_pause`:** a
record is skipped only when every required tool already has an
`approval_id`. Tools without one get a new record.
`_build_approval_dict` builds the record from the current pause only:
`tool_name`, `tool_args` and `context.tool_names` come from the new
approval calls, and `requirements` lists those calls plus any plain HITL
calls paused in the same turn. The second record names INV-2, not INV-1.
A repeated pause hook on the same pause still creates no duplicate.
- **`_group_tools_by_approval` / `_agroup_tools_by_approval`:** a record
found by a run-level or member-run lookup is not paired with a tool call
it doesn't name in its `requirements`. Such a call is left without a
record, so the continue is blocked. Records that don't list their tool
calls behave as before.
- **`check_and_apply_approval_resolution` /
`acheck_and_apply_approval_resolution`:** only
`approval_type="required"` tools are gated. Audit tools never get a
required record, so they no longer block the continue.
`metadata["approval"]` is now the record of the call this continue runs,
not the oldest record in the run.
- **Agent and team continue with requirements:** a requirement no longer
replaces a tool call that already has a result. After a pause resolved
from the approvals table, the requirement can hold an out-of-date copy
of the call (confirmed, no result), and resending `run.requirements` ran
the call a second time.

### Agent and team coverage

| Scenario | Before | After |
|---|---|---|
| Agent with an `@approval` tool | Second call ran under the first
call's approval | Separate record and admin decision per call |
| Team with an `@approval` tool on the team leader | Same bug (the team
pause hook uses the same `create_approval_from_pause`) | Separate record
and admin decision per call |
| Team with an `@approval` tool on a member agent | Each pause got its
own record, but the second also named the first call and showed its
arguments. If the second record was deleted, the first call's approval
ran the second call | Second record names only the second call; a
deleted record blocks the continue |

The sync, async and streaming paths all go through the shared functions
above.

Related: #10600 handles a different route to the approval gate
(continues that send tool or approval fields from the client). This PR
covers the continue with no requirements, where the call picked up the
old record inside `approval.py`.

## Reproduction

<details>
<summary>Reproduction code (AgentOS app)</summary>

```python
"""
Approval Reuse Across Pauses (AgentOS repro)
============================================

A second @approval tool call in the same run must get its own approval record.
Before the fix, the second call had no record, and continuing the run executed
it under the approval an admin resolved for the first call.

Components served:
- invoice-agent:          agent on OpenAIResponses (needs OPENAI_API_KEY)
- invoice-team:           team with the tool on the team leader, OpenAIResponses
- invoice-agent-scripted: agent, deterministic model (no API key needed)
- invoice-team-scripted:  team-level tool, deterministic model

The real-model components run with parallel tool calls off, so every
pay_invoice call is its own pause. The scripted model pays every invoice named
in the user message (e.g. INV-1, INV-2), one tool call per model turn.

Run the server:  python <file>.py
Run the client:  python <file>.py --demo [--scripted]
"""

import argparse
import json
import os
import re
from typing import Any, AsyncIterator, Iterator, List

import httpx
from agno.agent import Agent
from agno.approval import approval
from agno.db.sqlite import SqliteDb
from agno.metrics import MessageMetrics
from agno.models.base import Model
from agno.models.openai import OpenAIResponses
from agno.models.response import ModelResponse, ModelResponseEvent
from agno.os import AgentOS
from agno.team import Team
from agno.tools import tool

BASE_URL = "http://localhost:7777"
DB_FILE = "tmp/approval_repro.db"
PAYMENTS: List[str] = []


# ---------------------------------------------------------------------------
# Scripted Model: one pay_invoice call per turn for each invoice in the request
# ---------------------------------------------------------------------------
class InvoiceScriptedModel(Model):
    """Stateless, so it is safe to share across runs and sessions.

    Reads the latest user message for invoice numbers and counts the tool
    results that followed it: with n results, it pays invoice n + 1, or answers
    once every invoice is paid.
    """

    def __init__(self):
        super().__init__(id="invoice-scripted", name="invoice-scripted", provider="test")

    def _next(self, messages: List[Any]) -> ModelResponse:
        last_user = max((i for i, m in enumerate(messages) if m.role == "user"), default=-1)
        user_text = str(messages[last_user].content) if last_user >= 0 else ""
        invoices = re.findall(r"\b[A-Z]+-\d+\b", user_text) or ["INV-1", "INV-2"]
        answered = sum(1 for m in messages[last_user + 1 :] if m.role == "tool")

        if answered < len(invoices):
            r = ModelResponse(role="assistant")
            args = {"invoice": invoices[answered], "payee": "ACME", "amount": "100.00"}
            r.tool_calls = [
                {
                    "id": f"call_{answered + 1}",
                    "type": "function",
                    "function": {"name": "pay_invoice", "arguments": json.dumps(args)},
                }
            ]
        else:
            r = ModelResponse(content=f"Paid {', '.join(invoices)}.", role="assistant")
            r.event = ModelResponseEvent.assistant_response.value
        r.response_usage = MessageMetrics(input_tokens=10, output_tokens=5, total_tokens=15)
        return r

    def invoke(self, *a, **k):
        return self._next(k["messages"])

    async def ainvoke(self, *a, **k):
        return self._next(k["messages"])

    def invoke_stream(self, *a, **k) -> Iterator[ModelResponse]:
        yield self._next(k["messages"])

    async def ainvoke_stream(self, *a, **k) -> AsyncIterator[ModelResponse]:
        yield self._next(k["messages"])

    def _parse_provider_response(self, response: Any, **k) -> ModelResponse:
        return response if isinstance(response, ModelResponse) else ModelResponse()

    def _parse_provider_response_delta(self, response: Any) -> ModelResponse:
        return response if isinstance(response, ModelResponse) else ModelResponse()


# ---------------------------------------------------------------------------
# Approval-gated tool
# ---------------------------------------------------------------------------
@approval
@tool(requires_confirmation=True)
def pay_invoice(invoice: str, payee: str, amount: str) -> str:
    """Pay a vendor invoice. This moves money and cannot be undone.

    Args:
        invoice (str): Invoice number
        payee (str): Payee account
        amount (str): Amount to pay
    """
    PAYMENTS.append(invoice)
    print(f"[ledger] PAID {invoice} to {payee} ({amount}). All payments: {PAYMENTS}")
    return f"PAID {invoice}"


# ---------------------------------------------------------------------------
# Create AgentOS
# ---------------------------------------------------------------------------
os.makedirs("tmp", exist_ok=True)
db = SqliteDb(id="approval-repro-db", db_file=DB_FILE, approvals_table="approvals")

LLM_INSTRUCTIONS = [
    "You pay vendor invoices with the pay_invoice tool.",
    "Pay EVERY invoice the user lists. Call pay_invoice for one invoice, and as soon as its result comes back, "
    "call pay_invoice for the next unpaid invoice.",
    "Do not reply to the user or ask for confirmation until every listed invoice has been paid.",
    "Use payee ACME and amount 100.00 unless the user says otherwise.",
]

agents = [
    Agent(
        id="invoice-agent",
        name="Invoice Agent",
        model=OpenAIResponses(id="gpt-5.6-luna", parallel_tool_calls=False),
        tools=[pay_invoice],
        instructions=LLM_INSTRUCTIONS,
        db=db,
    ),
    Agent(
        id="invoice-agent-scripted",
        name="Invoice Agent (scripted)",
        model=InvoiceScriptedModel(),
        tools=[pay_invoice],
        db=db,
    ),
]
teams = [
    Team(
        id="invoice-team",
        name="Invoice Team",
        model=OpenAIResponses(id="gpt-5.6-luna", parallel_tool_calls=False),
        members=[
            Agent(
                id="helper",
                name="Helper",
                role="Answers general questions",
                model=OpenAIResponses(id="gpt-5.6-luna"),
            )
        ],
        tools=[pay_invoice],
        instructions=LLM_INSTRUCTIONS + ["Call pay_invoice yourself; do not delegate payments."],
        db=db,
    ),
    Team(
        id="invoice-team-scripted",
        name="Invoice Team (scripted)",
        model=InvoiceScriptedModel(),
        members=[Agent(id="helper-scripted", name="Helper", model=InvoiceScriptedModel())],
        tools=[pay_invoice],
        db=db,
    ),
]

agent_os = AgentOS(id="approval-reuse-os", db=db, agents=agents, teams=teams)
app = agent_os.get_app()


# ---------------------------------------------------------------------------
# Demo client: the documented "database resolution" flow over HTTP
# ---------------------------------------------------------------------------
def describe_approvals(client: httpx.Client, run_id: str) -> List[str]:
    out = []
    resp = client.get("/approvals", params={"run_id": run_id, "approval_type": "required"})
    for a in resp.json().get("data", []):
        calls = [
            (
                (r.get("tool_execution") or {}).get("tool_call_id"),
                ((r.get("tool_execution") or {}).get("tool_args") or {}).get("invoice"),
            )
            for r in a.get("requirements") or []
        ]
        out.append(f"{a['id'][:8]} {a['status']} (by {a.get('resolved_by')}) for {calls}")
    return out


def describe_tools(run: dict) -> list:
    return [
        (t.get("tool_call_id"), (t.get("tool_args") or {}).get("invoice"), t.get("confirmed"))
        for t in run.get("tools") or []
        if t.get("tool_name") == "pay_invoice"
    ]


def run_demo(kind: str, component_id: str) -> None:
    print(f"\n=== {kind}/{component_id} ===")
    with httpx.Client(base_url=BASE_URL, timeout=120.0) as client:
        resp = client.post(
            f"/{kind}/{component_id}/runs",
            data={"message": "Pay invoices INV-1 and INV-2 to ACME.", "stream": "false"},
        )
        resp.raise_for_status()
        run = resp.json()
        run_id, session_id = run["run_id"], run["session_id"]
        print(f"run: {run['status']} tools={describe_tools(run)}")

        for step in range(1, 5):
            print(f"  approvals: {describe_approvals(client, run_id)}")
            if run.get("status") != "PAUSED":
                break
            pending = client.get(
                "/approvals", params={"run_id": run_id, "status": "pending", "approval_type": "required"}
            ).json()["data"]
            for a in pending:
                client.post(f"/approvals/{a['id']}/resolve", json={"status": "approved", "resolved_by": "admin"})
            print(f"  resolved {len(pending)} pending approval(s)")
            resp = client.post(
                f"/{kind}/{component_id}/runs/{run_id}/continue",
                data={"tools": "", "session_id": session_id, "stream": "false"},
            )
            if resp.status_code >= 400:
                print(f"continue {step}: HTTP {resp.status_code} {resp.text[:160]}")
                break
            run = resp.json()
            print(f"continue {step}: {run['status']} tools={describe_tools(run)}")

        records = client.get("/approvals", params={"run_id": run_id, "approval_type": "required"}).json()
        executed = [t for t in describe_tools(run) if t[2]]
        print(
            f"pay_invoice calls executed: {len(executed)}; required approval records: {records['meta']['total_count']}"
        )


# ---------------------------------------------------------------------------
# Run AgentOS
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--demo", action="store_true", help="Run the HTTP client against a server on port 7777.")
    parser.add_argument("--scripted", action="store_true", help="With --demo, drive the scripted components.")
    args = parser.parse_args()

    if args.demo:
        suffix = "-scripted" if args.scripted else ""
        run_demo("agents", f"invoice-agent{suffix}")
        run_demo("teams", f"invoice-team{suffix}")
    else:
        agent_os.serve(app=app, port=7777)
```

</details>

## Type of change

- [x] Bug fix
- [ ] New feature
- [ ] Breaking change
- [ ] Improvement
- [ ] Model update
- [ ] Other:

---

## Checklist

- [x] Code complies with style guidelines
- [x] Ran format/validation scripts (`./scripts/format.sh` and
`./scripts/validate.sh`)
- [x] Self-review completed
- [x] Documentation updated (comments, docstrings)
- [ ] Examples and guides: Relevant cookbook examples have been included
or updated (if applicable)
- [ ] Tested in clean environment
- [x] Tests added/updated (if applicable)

### Duplicate and AI-Generated PR Check

- [x] I have searched existing [open pull
requests](https://github.com/agno-agi/agno/pulls) and confirmed that no
other PR already addresses this issue
- [ ] If a similar PR exists, I have explained below why this PR is a
better approach
- [ ] Check if this PR was entirely AI-generated (by Copilot, Claude
Code, Cursor, etc.)

---

## Additional Notes

**Tests added**

- `tests/unit/run/test_approval.py`: record creation on a later pause
(new record for the new call only, with that call's requirements and
args), no duplicate record from a repeated pause hook, and the fallback
refusing a record that names a different call, including when the tool's
own record is missing. Sync and async.
- `tests/unit/run/test_approval_multi_pause.py`: two calls one after the
other in one run produce two records and two admin decisions, and the
second call doesn't run while its record is pending. Covers an agent and
a team-leader tool, sync and async, with a scripted model.
- `tests/integration/os/test_approval_multi_pause.py`: the AgentOS HTTP
flow (`GET /approvals`, `POST /approvals/{id}/resolve`, `POST
/{agents|teams}/{id}/runs/{run_id}/continue` with empty tools) for the
agent and the team. It checks that the second call gets its own pending
record, isn't run while that record is pending, and doesn't run if
rejected. Uses a scripted model, so no API key is needed.
-
`tests/integration/agent/human_in_the_loop/test_approval_multi_pause.py`
and
`tests/integration/teams/human_in_the_loop/test_team_approval_multi_pause.py`:
the same checks with a real model (`OpenAIResponses`,
`parallel_tool_calls=False`). Skipped without `OPENAI_API_KEY`.

All new tests fail on `main` and pass with this change. The existing
run, agent, team and os unit suites pass. `validate.sh` reports mypy
errors only in files this PR doesn't touch, and the same errors appear
on `main`.

**Behaviour change:** a continue with no requirements now blocks if a
required tool call is paired with a record that names other tool calls.
Before, that call ran under the other record.

**Not in this PR** (same behaviour on `main`, to be fixed separately):
one approval can still pay a call twice when a run is forked and the
original is continued too, or when two continues for the same run arrive
at once. A workflow continue with no payload doesn't read the approval
record (see #9494).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Harsh <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Kaustubh
2026-10-06 11:30:49 +05:30
committed by GitHub
co-authored by Harsh Claude Opus 5.5
parent c44b082010
commit f3aa7be6af
8 changed files with 1737 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/agno/agno/agent/_run.py
+26 -25
View File
@@ -3379,6 +3379,29 @@ def _sync_requirements_with_tools(run_response: RunOutput, updated_tools: List[A
req.tool_execution = updated_tools_map[req.tool_execution.tool_call_id]
def _apply_requirement_tools(run_response: RunOutput, requirements: List[Any]) -> None:
"""Set the continue requirements on the run and merge their tool executions into run_response.tools.
A call that already ran keeps the run's own copy. A requirement can carry an
out-of-date copy of that call (confirmed, no result), for example after an
earlier pause was resolved from the approvals table; swapping it in would
execute the call a second time.
"""
run_response.requirements = requirements
updated_tools = [req.tool_execution for req in requirements if req.tool_execution is not None]
if updated_tools and run_response.tools:
# Checked per tool, not per tool_call_id: ids can repeat across turns (some
# providers send none and a fallback like call_{i} is used), and a new call
# sharing an executed call's id must still take its requirement.
updated_tools_map = {tool.tool_call_id: tool for tool in updated_tools}
run_response.tools = [
updated_tools_map.get(tool.tool_call_id, tool) if tool.result is None else tool
for tool in run_response.tools
]
else:
run_response.tools = updated_tools
def continue_run_dispatch(
agent: Agent,
run_response: Optional[RunOutput] = None,
@@ -3622,13 +3645,7 @@ def continue_run_dispatch(
# If we have requirements, get the updated tools and set them in the run_response
if requirements is not None:
run_response.requirements = requirements
updated_tools = [req.tool_execution for req in requirements if req.tool_execution is not None]
if updated_tools and run_response.tools:
updated_tools_map = {tool.tool_call_id: tool for tool in updated_tools}
run_response.tools = [updated_tools_map.get(tool.tool_call_id, tool) for tool in run_response.tools]
else:
run_response.tools = updated_tools
_apply_requirement_tools(run_response, requirements)
else:
# No tools / requirements in the body. Two cases:
@@ -4958,15 +4975,7 @@ async def _acontinue_run(
# If we have requirements, get the updated tools and set them in the run_response
if requirements is not None:
run_response.requirements = requirements
updated_tools = [req.tool_execution for req in requirements if req.tool_execution is not None]
if updated_tools and run_response.tools:
updated_tools_map = {tool.tool_call_id: tool for tool in updated_tools}
run_response.tools = [
updated_tools_map.get(tool.tool_call_id, tool) for tool in run_response.tools
]
else:
run_response.tools = updated_tools
_apply_requirement_tools(run_response, requirements)
else:
# No tools / requirements in the body. Two cases:
@@ -5482,15 +5491,7 @@ async def _acontinue_run_stream(
# If we have requirements, get the updated tools and set them in the run_response
if requirements is not None:
run_response.requirements = requirements
updated_tools = [req.tool_execution for req in requirements if req.tool_execution is not None]
if updated_tools and run_response.tools:
updated_tools_map = {tool.tool_call_id: tool for tool in updated_tools}
run_response.tools = [
updated_tools_map.get(tool.tool_call_id, tool) for tool in run_response.tools
]
else:
run_response.tools = updated_tools
_apply_requirement_tools(run_response, requirements)
else:
# No tools / requirements in the body. Two cases:
libs/agno/agno/run/approval.py
+197 -30
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, Set
from uuid import uuid4
from agno.run.base import RunStatus
@@ -34,30 +34,86 @@ def _get_first_approval_tool(tools: Optional[List[Any]], requirements: Optional[
def _has_approval_requirement(tools: Optional[List[Any]], requirements: Optional[List[Any]] = None) -> bool:
"""Check if any paused tool execution has approval_type set.
"""Check if any tool execution has approval_type='required'.
Checks both run_response.tools (agent-level) and run_response.requirements
(team-level, where member tools are propagated via requirements).
"""
tool = _get_first_approval_tool(tools, requirements)
return tool is not None and getattr(tool, "approval_type", None) == "required"
candidates = list(tools or []) + [getattr(r, "tool_execution", None) for r in requirements or []]
# Any required tool counts, not just the first approval tool: an audit tool
# called earlier in the run must not hide a later required one.
return any(getattr(t, "approval_type", None) == "required" for t in candidates if t is not None)
def _stamp_approval_id_on_tools(
tools: Optional[List[Any]], requirements: Optional[List[Any]], approval_id: str
) -> None:
"""Stamp approval_id on every tool that has approval_type set."""
"""Stamp approval_id on every required tool that has none yet.
Audit tools are left alone: they never resolve through a required record, and
a shared id would let a lookup by approval_id pick the audit tool instead.
"""
if tools:
for tool in tools:
if getattr(tool, "approval_type", None) is not None and getattr(tool, "approval_id", None) is None:
if getattr(tool, "approval_type", None) == "required" and getattr(tool, "approval_id", None) is None:
tool.approval_id = approval_id
if requirements:
for req in requirements:
te = getattr(req, "tool_execution", None)
if te and getattr(te, "approval_type", None) is not None and getattr(te, "approval_id", None) is None:
if te and getattr(te, "approval_type", None) == "required" and getattr(te, "approval_id", None) is None:
te.approval_id = approval_id
def _unstamped_approval_tools(tools: Optional[List[Any]], requirements: Optional[List[Any]]) -> List[Any]:
"""Required-approval tools that carry no approval_id yet: the tools raised by the current pause.
run_response.tools and run_response.requirements accumulate across every pause
of a run, so a tool stamped at an earlier pause (and possibly already approved
and executed) must not stand in for a tool raised by a later pause.
"""
result: List[Any] = []
seen_ids: Set[str] = set()
candidates = list(tools or []) + [getattr(r, "tool_execution", None) for r in requirements or []]
for t in candidates:
if t is None or getattr(t, "approval_type", None) != "required" or getattr(t, "approval_id", None) is not None:
continue
tcid = getattr(t, "tool_call_id", None)
if any(r is t for r in result) or (tcid is not None and tcid in seen_ids):
continue
result.append(t)
if tcid is not None:
seen_ids.add(tcid)
return result
def _is_pending_tool(tool_execution: Any, pending_tools: List[Any]) -> bool:
"""Whether the tool execution is one of pending_tools, by identity or by tool_call_id.
After a DB reload, run_response.tools and the requirements hold distinct
copies of the same tool call, so identity alone misses the requirement's copy.
"""
tcid = getattr(tool_execution, "tool_call_id", None)
return any(
p is tool_execution or (tcid is not None and getattr(p, "tool_call_id", None) == tcid) for p in pending_tools
)
def _belongs_to_current_pause(req: Any, pending_tools: List[Any]) -> bool:
"""Whether a requirement belongs in the record created for the current pause.
That is the pending approval tools, plus HITL tools without approval paused
in the same turn (still unresolved). Requirements resolved at earlier pauses
of the run are left out.
"""
te = getattr(req, "tool_execution", None)
if te is None:
return False
if _is_pending_tool(te, pending_tools):
return True
is_resolved = getattr(req, "is_resolved", None)
return getattr(te, "approval_type", None) is None and callable(is_resolved) and not is_resolved()
def _build_approval_dict(
run_response: Any,
agent_id: Optional[str] = None,
@@ -69,8 +125,15 @@ def _build_approval_dict(
user_id: Optional[str] = None,
schedule_id: Optional[str] = None,
schedule_run_id: Optional[str] = None,
pending_tools: Optional[List[Any]] = None,
) -> Dict[str, Any]:
"""Build the approval record dict from run response and context."""
"""Build the approval record dict from run response and context.
When pending_tools is given, the record covers only the current pause: the
pending approval tools and any unresolved HITL tools paused alongside them,
so a record created at a later pause never describes a tool resolved at an
earlier one.
"""
# Determine source type
source_type = "agent"
source_name = agent_name
@@ -86,6 +149,8 @@ def _build_approval_dict(
if hasattr(run_response, "requirements") and run_response.requirements:
requirements_data = []
for req in run_response.requirements:
if pending_tools is not None and not _belongs_to_current_pause(req, pending_tools):
continue
if hasattr(req, "to_dict"):
requirements_data.append(req.to_dict())
elif isinstance(req, dict):
@@ -94,7 +159,7 @@ def _build_approval_dict(
# Find the first approval tool to extract pause_type, tool_name, tool_args
tools = getattr(run_response, "tools", None)
requirements = getattr(run_response, "requirements", None)
first_tool = _get_first_approval_tool(tools, requirements)
first_tool = pending_tools[0] if pending_tools else _get_first_approval_tool(tools, requirements)
pause_type = _get_pause_type(first_tool) if first_tool else "confirmation"
tool_name = getattr(first_tool, "tool_name", None) if first_tool else None
@@ -102,7 +167,9 @@ def _build_approval_dict(
# Build context with tool names for UI display.
tool_names: List[str] = []
if hasattr(run_response, "requirements") and run_response.requirements:
if pending_tools:
tool_names = [t.tool_name for t in pending_tools if getattr(t, "tool_name", None)]
elif hasattr(run_response, "requirements") and run_response.requirements:
for req in run_response.requirements:
te = getattr(req, "tool_execution", None)
if te and getattr(te, "approval_type", None) is not None:
@@ -175,10 +242,15 @@ def create_approval_from_pause(
if not _has_approval_requirement(tools, requirements):
return None
# Skip if an approval_id is already stamped (avoids duplicates when pause hook fires twice)
for t in tools or []:
if getattr(t, "approval_type", None) == "required" and getattr(t, "approval_id", None) is not None:
return getattr(t, "approval_id", None)
# Every required tool already carries a record (the pause hook fired twice):
# nothing new to approve. A tool raised by a later pause of the same run has
# no approval_id yet and gets a record of its own.
pending_tools = _unstamped_approval_tools(tools, requirements)
if not pending_tools:
for t in reversed(tools or []):
if getattr(t, "approval_type", None) == "required" and getattr(t, "approval_id", None) is not None:
return getattr(t, "approval_id", None)
return None
try:
approval_data = _build_approval_dict(
@@ -192,6 +264,7 @@ def create_approval_from_pause(
user_id=user_id,
schedule_id=schedule_id,
schedule_run_id=schedule_run_id,
pending_tools=pending_tools,
)
db.create_approval(approval_data)
approval_id: str = approval_data["id"]
@@ -231,10 +304,15 @@ async def acreate_approval_from_pause(
if not _has_approval_requirement(tools, requirements):
return None
# Skip if an approval_id is already stamped (avoids duplicates when pause hook fires twice)
for t in tools or []:
if getattr(t, "approval_type", None) == "required" and getattr(t, "approval_id", None) is not None:
return getattr(t, "approval_id", None)
# Every required tool already carries a record (the pause hook fired twice):
# nothing new to approve. A tool raised by a later pause of the same run has
# no approval_id yet and gets a record of its own.
pending_tools = _unstamped_approval_tools(tools, requirements)
if not pending_tools:
for t in reversed(tools or []):
if getattr(t, "approval_type", None) == "required" and getattr(t, "approval_id", None) is not None:
return getattr(t, "approval_id", None)
return None
try:
approval_data = _build_approval_dict(
@@ -248,6 +326,7 @@ async def acreate_approval_from_pause(
user_id=user_id,
schedule_id=schedule_id,
schedule_run_id=schedule_run_id,
pending_tools=pending_tools,
)
# Try async first, fall back to sync
create_fn = getattr(db, "create_approval", None)
@@ -474,6 +553,20 @@ def _member_run_id_for_tool(run_response: Any, tool: Any) -> Optional[str]:
return None
def _record_names_other_tools(record: Optional[Dict[str, Any]], tool: Any) -> bool:
"""True when the record lists the tool calls it covers and this tool is not among them."""
tcid = getattr(tool, "tool_call_id", None)
if record is None or tcid is None:
return False
named = [
(r.get("tool_execution") or {}).get("tool_call_id")
for r in record.get("requirements") or []
if isinstance(r, dict) and isinstance(r.get("tool_execution"), dict)
]
named = [n for n in named if n is not None]
return bool(named) and tcid not in named
def _group_tools_by_approval(db: Any, run_id: str, run_response: Any, tools: List[Any]) -> List[tuple]:
"""Pair every approval tool with ITS OWN approval record.
@@ -509,7 +602,10 @@ def _group_tools_by_approval(db: Any, run_id: str, run_response: Any, tools: Lis
key = f"run:{mid}"
if key not in cache:
cache[key] = _get_approval_for_run(db, mid)
record = cache[key]
# The member run's latest record covers only the calls it names:
# once a later call's own record is gone, an earlier call's
# approval must not execute it.
record = None if _record_names_other_tools(cache[key], tool) else cache[key]
if record is None and aid and not mid:
# A tool whose stamped record is gone and that owns no member run
# resolves within its own run only: a record re-issued under this
@@ -517,7 +613,7 @@ def _group_tools_by_approval(db: Any, run_id: str, run_response: Any, tools: Lis
key = f"run:{run_id}"
if key not in cache:
cache[key] = _get_approval_for_run(db, run_id)
record = cache[key]
record = None if _record_names_other_tools(cache[key], tool) else cache[key]
if record is None and not aid and not mid:
# The run-level lookup serves only tools with no scoped identity.
# A tool whose own record is gone stays unresolved: pairing it with
@@ -528,7 +624,9 @@ def _group_tools_by_approval(db: Any, run_id: str, run_response: Any, tools: Lis
fallback = _get_approval_for_run(db, rid)
if fallback is not None:
break
record = fallback
# A record that names its tool calls covers only those calls: an
# approval for one call never executes a different call.
record = None if _record_names_other_tools(fallback, tool) else fallback
pairs.append((tool, record))
return pairs
@@ -563,7 +661,10 @@ async def _agroup_tools_by_approval(db: Any, run_id: str, run_response: Any, too
key = f"run:{mid}"
if key not in cache:
cache[key] = await _aget_approval_for_run(db, mid)
record = cache[key]
# The member run's latest record covers only the calls it names:
# once a later call's own record is gone, an earlier call's
# approval must not execute it.
record = None if _record_names_other_tools(cache[key], tool) else cache[key]
if record is None and aid and not mid:
# A tool whose stamped record is gone and that owns no member run
# resolves within its own run only: a record re-issued under this
@@ -571,7 +672,7 @@ async def _agroup_tools_by_approval(db: Any, run_id: str, run_response: Any, too
key = f"run:{run_id}"
if key not in cache:
cache[key] = await _aget_approval_for_run(db, run_id)
record = cache[key]
record = None if _record_names_other_tools(cache[key], tool) else cache[key]
if record is None and not aid and not mid:
# The run-level lookup serves only tools with no scoped identity.
# A tool whose own record is gone stays unresolved: pairing it with
@@ -582,7 +683,9 @@ async def _agroup_tools_by_approval(db: Any, run_id: str, run_response: Any, too
fallback = await _aget_approval_for_run(db, rid)
if fallback is not None:
break
record = fallback
# A record that names its tool calls covers only those calls: an
# approval for one call never executes a different call.
record = None if _record_names_other_tools(fallback, tool) else fallback
pairs.append((tool, record))
return pairs
@@ -630,6 +733,68 @@ def _sync_requirements_after_approval(run_response: Any, approval_status: str, o
req.confirmation = False
def _calls_that_already_ran(run_response: Any) -> List[Any]:
"""The run's own tool calls that ran at an earlier pause.
A call counts only when it has a result AND the run's stored messages hold a
tool message for it. A continue payload can put a result on a tool execution,
but it cannot add a tool message, so a call cannot skip the gate by claiming
it ran. External-execution calls never count: a client-supplied result is how
they run, so it must still pass the gate.
"""
tool_message_ids = {
getattr(m, "tool_call_id", None)
for m in getattr(run_response, "messages", None) or []
if getattr(m, "role", None) == "tool" and getattr(m, "tool_call_id", None)
}
return [
t
for t in getattr(run_response, "tools", None) or []
if getattr(t, "result", None) is not None
and not getattr(t, "external_execution_required", False)
and getattr(t, "tool_call_id", None) in tool_message_ids
]
def _is_copy_of(tool: Any, other: Any) -> bool:
"""Same tool call by id, name and arguments: ids alone can repeat across turns."""
return (
getattr(tool, "tool_call_id", None) is not None
and getattr(tool, "tool_call_id", None) == getattr(other, "tool_call_id", None)
and getattr(tool, "tool_name", None) == getattr(other, "tool_name", None)
and getattr(tool, "tool_args", None) == getattr(other, "tool_args", None)
)
def _tools_to_gate(run_response: Any) -> List[Any]:
"""Required tools this continue still has to resolve.
Audit tools never get a required record: their record is written after the
HITL interaction resolves. Calls that already ran are skipped: there is
nothing left to approve, and their record may since have been deleted. That
covers the requirement's out-of-date copy of such a call (no result), matched
by id, name and arguments. A member's call stays gated, as provider-local
tool_call_ids can collide across members.
"""
ran = _calls_that_already_ran(run_response)
run_tools = getattr(run_response, "tools", None) or []
gated: List[Any] = []
for t in _collect_all_approval_tools(run_response):
if getattr(t, "approval_type", None) != "required":
continue
if any(t is r for r in ran):
continue
is_requirement_copy = not any(t is r for r in run_tools)
if (
is_requirement_copy
and _member_run_id_for_tool(run_response, t) is None
and any(_is_copy_of(t, r) for r in ran)
):
continue
gated.append(t)
return gated
def _attach_resolved_approval(run_response: Any, approval: Dict[str, Any]) -> None:
"""Expose the resolved approval record to post-hooks via run_response.metadata["approval"]."""
if run_response.metadata is None:
@@ -649,11 +814,11 @@ def check_and_apply_approval_resolution(db: Any, run_id: str, run_response: Any)
if db is None:
return
all_approval_tools = _collect_all_approval_tools(run_response)
if not any(getattr(t, "approval_type", None) == "required" for t in all_approval_tools):
required_tools = _tools_to_gate(run_response)
if not required_tools:
return
pairs = _group_tools_by_approval(db, run_id, run_response, all_approval_tools)
pairs = _group_tools_by_approval(db, run_id, run_response, required_tools)
if any(record is None for _, record in pairs):
raise RuntimeError(
"No approval record found for this run. Cannot continue a run that requires external approval."
@@ -666,6 +831,7 @@ def check_and_apply_approval_resolution(db: Any, run_id: str, run_response: Any)
_apply_approval_to_tools([tool], record.get("status", "pending"), record.get("resolution_data"))
_sync_requirements_after_approval(run_response, record.get("status", "pending"), only_tool=tool)
# pairs holds only calls this continue runs, so pairs[0] is the current call's record.
_attach_resolved_approval(run_response, pairs[0][1])
@@ -674,11 +840,11 @@ async def acheck_and_apply_approval_resolution(db: Any, run_id: str, run_respons
if db is None:
return
all_approval_tools = _collect_all_approval_tools(run_response)
if not any(getattr(t, "approval_type", None) == "required" for t in all_approval_tools):
required_tools = _tools_to_gate(run_response)
if not required_tools:
return
pairs = await _agroup_tools_by_approval(db, run_id, run_response, all_approval_tools)
pairs = await _agroup_tools_by_approval(db, run_id, run_response, required_tools)
if any(record is None for _, record in pairs):
raise RuntimeError(
"No approval record found for this run. Cannot continue a run that requires external approval."
@@ -691,6 +857,7 @@ async def acheck_and_apply_approval_resolution(db: Any, run_id: str, run_respons
_apply_approval_to_tools([tool], record.get("status", "pending"), record.get("resolution_data"))
_sync_requirements_after_approval(run_response, record.get("status", "pending"), only_tool=tool)
# pairs holds only calls this continue runs, so pairs[0] is the current call's record.
_attach_resolved_approval(run_response, pairs[0][1])
libs/agno/agno/team/_run.py
+8 -1
View File
@@ -5865,11 +5865,18 @@ def _merge_tools_preserving_approval(
This function preserves approval_type and approval_id from the session originals
whenever the incoming tool does not carry them.
A call that already ran keeps the session original. The payload can carry an
out-of-date copy of that call (confirmed, no result), for example after an
earlier pause was resolved from the approvals table; swapping it in would
execute the call a second time.
"""
merged: List[Any] = []
for orig in original_tools:
updated = updated_tools_map.get(orig.tool_call_id)
if updated is not None:
if updated is not None and getattr(orig, "result", None) is not None:
merged.append(orig)
elif updated is not None:
for attr in ("approval_type", "approval_id"):
if getattr(updated, attr, None) is None and getattr(orig, attr, None) is not None:
setattr(updated, attr, getattr(orig, attr))
libs/agno/tests/integration/agent/human_in_the_loop/test_approval_multi_pause.py
+136
View File
@@ -0,0 +1,136 @@
"""Each approval-required tool call in an agent run gets its own approval record.
With parallel tool calls off, a request to pay two invoices pauses the run once
per call. Continuing by run id after resolving the run's pending approvals must
never execute a call that no approved record names.
"""
import os
import time
from typing import List, Set
import pytest
from agno.agent import Agent
from agno.approval import approval
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
pytestmark = pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
PAID: List[str] = []
INSTRUCTIONS = [
"You pay vendor invoices with the pay_invoice tool.",
"Pay every invoice the user lists: after each pay_invoice result, immediately pay the next unpaid invoice.",
"Do not reply to the user until every listed invoice has been paid.",
]
@approval
@tool(requires_confirmation=True)
def pay_invoice(invoice: str) -> str:
"""Pay a vendor invoice. This moves money and cannot be undone.
Args:
invoice (str): Invoice number, e.g. INV-1
"""
PAID.append(invoice)
return f"PAID {invoice}"
def _make_agent(db: SqliteDb) -> Agent:
return Agent(
model=OpenAIResponses(id="gpt-5.6-luna", parallel_tool_calls=False),
tools=[pay_invoice],
instructions=INSTRUCTIONS,
db=db,
telemetry=False,
)
def _named_calls(db: SqliteDb, run_id: str, status: str) -> Set[str]:
records, _ = db.get_approvals(run_id=run_id, approval_type="required", status=status)
return {
(r.get("tool_execution") or {}).get("tool_call_id") for rec in records for r in rec.get("requirements") or []
}
def _approve_pending(db: SqliteDb, run_id: str) -> int:
pending, _ = db.get_approvals(run_id=run_id, approval_type="required", status="pending")
for record in pending:
db.update_approval(
record["id"],
expected_status="pending",
status="approved",
resolved_by="admin",
resolved_at=int(time.time()),
)
return len(pending)
def _assert_paused_calls_have_pending_records(db: SqliteDb, run) -> None:
waiting = {t.tool_call_id for t in run.tools or [] if t.tool_name == "pay_invoice" and t.confirmed is None}
assert waiting, "run paused without a pay_invoice call waiting for approval"
assert waiting <= _named_calls(db, run.run_id, "pending")
def _assert_every_executed_call_was_approved(db: SqliteDb, run) -> None:
executed = {t.tool_call_id for t in run.tools or [] if t.tool_name == "pay_invoice" and t.result is not None}
assert executed <= _named_calls(db, run.run_id, "approved")
records, total = db.get_approvals(run_id=run.run_id, approval_type="required")
assert total == len(executed)
@pytest.fixture
def db(shared_db):
PAID.clear()
return shared_db
def test_each_call_needs_its_own_approval(db):
agent = _make_agent(db)
run = agent.run("Pay invoices INV-1 and INV-2.", session_id="multi_pause_sync")
pauses = 0
while run.is_paused and pauses < 4:
pauses += 1
_assert_paused_calls_have_pending_records(db, run)
assert _approve_pending(db, run.run_id) >= 1
run = agent.continue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert pauses == 2
assert sorted(PAID) == ["INV-1", "INV-2"]
_assert_every_executed_call_was_approved(db, run)
@pytest.mark.asyncio
async def test_each_call_needs_its_own_approval_async(db):
agent = _make_agent(db)
run = await agent.arun("Pay invoices INV-1 and INV-2.", session_id="multi_pause_async")
pauses = 0
while run.is_paused and pauses < 4:
pauses += 1
_assert_paused_calls_have_pending_records(db, run)
assert _approve_pending(db, run.run_id) >= 1
run = await agent.acontinue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert pauses == 2
assert sorted(PAID) == ["INV-1", "INV-2"]
_assert_every_executed_call_was_approved(db, run)
def test_pending_second_approval_blocks_continue(db):
agent = _make_agent(db)
run = agent.run("Pay invoices INV-1 and INV-2.", session_id="multi_pause_blocked")
assert run.is_paused
_approve_pending(db, run.run_id)
run = agent.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
assert len(PAID) == 1
# The second call's record is still pending: continuing must not pay it.
with pytest.raises(ValueError, match="unresolved HITL requirements"):
agent.continue_run(run_id=run.run_id, session_id=run.session_id)
assert len(PAID) == 1
libs/agno/tests/integration/os/test_approval_multi_pause.py
+198
View File
@@ -0,0 +1,198 @@
"""AgentOS: each approval-required tool call in a run gets its own approval record.
Drives the documented database-resolution flow over HTTP: list the run's
pending approvals, resolve them, continue the run by id with empty tools. A
run that pauses a second time for a new call must surface a new pending
record for that call, and the call must not execute until it is approved.
The model is scripted so the flow is deterministic and needs no API key.
"""
import json
from typing import Any, AsyncIterator, Iterator, List
import pytest
from fastapi.testclient import TestClient
from agno.agent import Agent
from agno.approval import approval
from agno.metrics import MessageMetrics
from agno.models.base import Model
from agno.models.response import ModelResponse, ModelResponseEvent
from agno.os import AgentOS
from agno.team import Team
from agno.tools import tool
PAID: List[str] = []
class InvoiceScriptedModel(Model):
"""Pays INV-1, then INV-2 (one call per turn), then answers. Stateless:
the next turn is derived from the tool results since the last user message."""
def __init__(self):
super().__init__(id="invoice-scripted", name="invoice-scripted", provider="test")
def _next(self, messages: List[Any]) -> ModelResponse:
last_user = max((i for i, m in enumerate(messages) if m.role == "user"), default=-1)
answered = sum(1 for m in messages[last_user + 1 :] if m.role == "tool")
if answered < 2:
r = ModelResponse(role="assistant")
args = {"invoice": f"INV-{answered + 1}"}
r.tool_calls = [
{
"id": f"call_{answered + 1}",
"type": "function",
"function": {"name": "pay_invoice", "arguments": json.dumps(args)},
}
]
else:
r = ModelResponse(content="Paid.", role="assistant")
r.event = ModelResponseEvent.assistant_response.value
r.response_usage = MessageMetrics(input_tokens=10, output_tokens=5, total_tokens=15)
return r
def invoke(self, *a, **k):
return self._next(k["messages"])
async def ainvoke(self, *a, **k):
return self._next(k["messages"])
def invoke_stream(self, *a, **k) -> Iterator[ModelResponse]:
yield self._next(k["messages"])
async def ainvoke_stream(self, *a, **k) -> AsyncIterator[ModelResponse]:
yield self._next(k["messages"])
def _parse_provider_response(self, response: Any, **k) -> ModelResponse:
return response if isinstance(response, ModelResponse) else ModelResponse()
def _parse_provider_response_delta(self, response: Any) -> ModelResponse:
return response if isinstance(response, ModelResponse) else ModelResponse()
@approval
@tool(requires_confirmation=True)
def pay_invoice(invoice: str) -> str:
"""Pay a vendor invoice.
Args:
invoice (str): Invoice number
"""
PAID.append(invoice)
return f"PAID {invoice}"
@pytest.fixture
def client(shared_db):
PAID.clear()
agent = Agent(id="invoice-agent", model=InvoiceScriptedModel(), tools=[pay_invoice], db=shared_db, telemetry=False)
team = Team(
id="invoice-team",
model=InvoiceScriptedModel(),
members=[Agent(id="helper", model=InvoiceScriptedModel(), telemetry=False)],
tools=[pay_invoice],
db=shared_db,
telemetry=False,
)
app = AgentOS(agents=[agent], teams=[team], db=shared_db, telemetry=False).get_app()
return TestClient(app)
COMPONENTS = [pytest.param(("agents", "invoice-agent"), id="agent"), pytest.param(("teams", "invoice-team"), id="team")]
def _start(client: TestClient, kind: str, component_id: str) -> dict:
resp = client.post(f"/{kind}/{component_id}/runs", data={"message": "Pay INV-1 and INV-2.", "stream": "false"})
assert resp.status_code == 200, resp.text
run = resp.json()
assert run["status"] == "PAUSED"
return run
def _continue(client: TestClient, kind: str, component_id: str, run: dict):
return client.post(
f"/{kind}/{component_id}/runs/{run['run_id']}/continue",
data={"tools": "", "session_id": run["session_id"], "stream": "false"},
)
def _records(client: TestClient, run_id: str, status: str = "") -> List[dict]:
params = {"run_id": run_id, "approval_type": "required"}
if status:
params["status"] = status
resp = client.get("/approvals", params=params)
assert resp.status_code == 200, resp.text
return sorted(resp.json()["data"], key=lambda r: r["created_at"])
def _named_calls(record: dict) -> List[str]:
return [(r.get("tool_execution") or {}).get("tool_call_id") for r in record.get("requirements") or []]
def _resolve(client: TestClient, approval_id: str, status: str) -> None:
resp = client.post(f"/approvals/{approval_id}/resolve", json={"status": status, "resolved_by": "admin"})
assert resp.status_code == 200, resp.text
def _pause_on_second_call(client: TestClient, kind: str, component_id: str) -> dict:
"""Start a run, approve INV-1, continue: the run pauses again on INV-2."""
run = _start(client, kind, component_id)
(first,) = _records(client, run["run_id"], "pending")
assert _named_calls(first) == ["call_1"]
_resolve(client, first["id"], "approved")
resp = _continue(client, kind, component_id, run)
assert resp.status_code == 200, resp.text
run = resp.json()
assert run["status"] == "PAUSED"
assert PAID == ["INV-1"]
return run
@pytest.mark.parametrize("component", COMPONENTS)
def test_second_call_surfaces_its_own_pending_approval(client, component):
kind, component_id = component
run = _pause_on_second_call(client, kind, component_id)
pending = _records(client, run["run_id"], "pending")
assert len(pending) == 1
assert _named_calls(pending[0]) == ["call_2"]
assert pending[0]["tool_args"] == {"invoice": "INV-2"}
_resolve(client, pending[0]["id"], "approved")
resp = _continue(client, kind, component_id, run)
assert resp.status_code == 200, resp.text
assert resp.json()["status"] == "COMPLETED"
assert PAID == ["INV-1", "INV-2"]
records = _records(client, run["run_id"])
assert [_named_calls(r) for r in records] == [["call_1"], ["call_2"]]
assert [(r["status"], r["resolved_by"]) for r in records] == [("approved", "admin"), ("approved", "admin")]
@pytest.mark.parametrize("component", COMPONENTS)
def test_second_call_does_not_run_while_its_approval_is_pending(client, component):
kind, component_id = component
run = _pause_on_second_call(client, kind, component_id)
resp = _continue(client, kind, component_id, run)
if resp.status_code == 200:
assert resp.json()["status"] == "PAUSED"
else:
assert 400 <= resp.status_code < 500, resp.text
assert PAID == ["INV-1"]
assert len(_records(client, run["run_id"], "pending")) == 1
@pytest.mark.parametrize("component", COMPONENTS)
def test_rejecting_second_call_keeps_it_from_running(client, component):
kind, component_id = component
run = _pause_on_second_call(client, kind, component_id)
(pending,) = _records(client, run["run_id"], "pending")
_resolve(client, pending["id"], "rejected")
resp = _continue(client, kind, component_id, run)
assert resp.status_code == 200, resp.text
assert PAID == ["INV-1"]
assert [r["status"] for r in _records(client, run["run_id"])] == ["approved", "rejected"]
libs/agno/tests/integration/teams/human_in_the_loop/test_team_approval_multi_pause.py
+148
View File
@@ -0,0 +1,148 @@
"""Each approval-required call to a team-level tool gets its own approval record.
The tool sits on the Team itself. With parallel tool calls off, a request to
pay two invoices pauses the team run once per call. Continuing by run id after
resolving the run's pending approvals must never execute a call that no
approved record names.
"""
import os
import time
from typing import List, Set
import pytest
from agno.agent import Agent
from agno.approval import approval
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.team.team import Team
from agno.tools import tool
pytestmark = pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
PAID: List[str] = []
@approval
@tool(requires_confirmation=True)
def pay_invoice(invoice: str) -> str:
"""Pay a vendor invoice. This moves money and cannot be undone.
Args:
invoice (str): Invoice number, e.g. INV-1
"""
PAID.append(invoice)
return f"PAID {invoice}"
def _make_team(db: SqliteDb) -> Team:
helper = Agent(
name="Helper Agent",
role="Answers general questions",
model=OpenAIResponses(id="gpt-5.6-luna"),
telemetry=False,
)
return Team(
name="Finance Team",
model=OpenAIResponses(id="gpt-5.6-luna", parallel_tool_calls=False),
members=[helper],
tools=[pay_invoice],
db=db,
telemetry=False,
instructions=[
"You pay vendor invoices yourself with the pay_invoice tool. Never delegate payments.",
"Pay every invoice the user lists: after each pay_invoice result, immediately pay the next unpaid invoice.",
"Do not reply to the user until every listed invoice has been paid.",
],
)
def _named_calls(db: SqliteDb, run_id: str, status: str) -> Set[str]:
records, _ = db.get_approvals(run_id=run_id, approval_type="required", status=status)
return {
(r.get("tool_execution") or {}).get("tool_call_id") for rec in records for r in rec.get("requirements") or []
}
def _approve_pending(db: SqliteDb, run_id: str) -> int:
pending, _ = db.get_approvals(run_id=run_id, approval_type="required", status="pending")
for record in pending:
db.update_approval(
record["id"],
expected_status="pending",
status="approved",
resolved_by="admin",
resolved_at=int(time.time()),
)
return len(pending)
def _assert_paused_calls_have_pending_records(db: SqliteDb, run) -> None:
waiting = {t.tool_call_id for t in run.tools or [] if t.tool_name == "pay_invoice" and t.confirmed is None}
assert waiting, "team paused without a pay_invoice call waiting for approval"
assert waiting <= _named_calls(db, run.run_id, "pending")
def _assert_every_executed_call_was_approved(db: SqliteDb, run) -> None:
executed = {t.tool_call_id for t in run.tools or [] if t.tool_name == "pay_invoice" and t.result is not None}
assert executed <= _named_calls(db, run.run_id, "approved")
_, total = db.get_approvals(run_id=run.run_id, approval_type="required")
assert total == len(executed)
@pytest.fixture
def db(shared_db):
PAID.clear()
return shared_db
def test_team_tool_each_call_needs_its_own_approval(db):
team = _make_team(db)
run = team.run("Pay invoices INV-1 and INV-2.", session_id="team_multi_pause_sync")
pauses = 0
while run.is_paused and pauses < 4:
pauses += 1
_assert_paused_calls_have_pending_records(db, run)
assert _approve_pending(db, run.run_id) >= 1
run = team.continue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert pauses == 2
assert sorted(PAID) == ["INV-1", "INV-2"]
_assert_every_executed_call_was_approved(db, run)
@pytest.mark.asyncio
async def test_team_tool_each_call_needs_its_own_approval_async(db):
team = _make_team(db)
run = await team.arun("Pay invoices INV-1 and INV-2.", session_id="team_multi_pause_async")
pauses = 0
while run.is_paused and pauses < 4:
pauses += 1
_assert_paused_calls_have_pending_records(db, run)
assert _approve_pending(db, run.run_id) >= 1
run = await team.acontinue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert pauses == 2
assert sorted(PAID) == ["INV-1", "INV-2"]
_assert_every_executed_call_was_approved(db, run)
def test_team_tool_pending_second_approval_does_not_execute(db):
team = _make_team(db)
run = team.run("Pay invoices INV-1 and INV-2.", session_id="team_multi_pause_blocked")
assert run.is_paused
_approve_pending(db, run.run_id)
run = team.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
assert len(PAID) == 1
# The second call's record is still pending: continuing must not pay it.
try:
run = team.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
except ValueError:
pass
assert len(PAID) == 1
libs/agno/tests/unit/run/test_approval.py
+471
View File
@@ -28,6 +28,7 @@ from agno.run.approval import (
@dataclass
class FakeToolExecution:
tool_name: Optional[str] = None
tool_call_id: Optional[str] = None
tool_args: Optional[Dict[str, Any]] = None
approval_type: Optional[str] = None
approval_id: Optional[str] = None
@@ -54,6 +55,7 @@ class FakeRunResponse:
tools: Optional[list] = None
requirements: Optional[list] = None
metadata: Optional[Dict[str, Any]] = None
messages: Optional[list] = None
@dataclass
@@ -598,3 +600,472 @@ class TestAsyncCheckAndApplyApprovalResolution:
await acheck_and_apply_approval_resolution(db=db, run_id="r1", run_response=rr)
assert rr.metadata is not None
assert rr.metadata["approval"] == approval
# =============================================================================
# Multiple pauses in one run: each approval-required call gets its own record
# =============================================================================
@dataclass
class FakeCallRequirement:
tool_execution: Optional[FakeToolExecution] = None
def to_dict(self) -> Dict[str, Any]:
te = self.tool_execution
return {"tool_execution": {"tool_call_id": te.tool_call_id, "tool_args": te.tool_args} if te else None}
@dataclass
class FakeMemberRequirement:
tool_execution: Optional[FakeToolExecution] = None
member_run_id: Optional[str] = None
confirmation: Optional[bool] = None
def _second_pause_run_response():
"""A run at its second pause: call_1 was approved and executed under appr-1,
call_2 was just raised. tools and requirements accumulate across pauses."""
first = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_1",
tool_args={"invoice": "INV-1"},
approval_type="required",
approval_id="appr-1",
requires_confirmation=True,
confirmed=True,
result="PAID",
)
second = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_2",
tool_args={"invoice": "INV-2"},
approval_type="required",
requires_confirmation=True,
)
rr = FakeRunResponse(
tools=[first, second],
requirements=[FakeCallRequirement(tool_execution=first), FakeCallRequirement(tool_execution=second)],
)
return rr, first, second
def _record_for(*tool_call_ids: str, status: str = "approved", record_id: str = "appr-1") -> Dict[str, Any]:
return {
"id": record_id,
"status": status,
"resolution_data": None,
"requirements": [{"tool_execution": {"tool_call_id": tcid}} for tcid in tool_call_ids],
}
class TestCreateApprovalOnLaterPause:
def test_later_pause_creates_record_for_new_call_only(self):
db = MagicMock()
rr, first, second = _second_pause_run_response()
result = create_approval_from_pause(db=db, run_response=rr, agent_id="a1")
db.create_approval.assert_called_once()
data = db.create_approval.call_args[0][0]
assert result == data["id"] != "appr-1"
assert second.approval_id == result
assert first.approval_id == "appr-1"
assert data["tool_args"] == {"invoice": "INV-2"}
assert [r["tool_execution"]["tool_call_id"] for r in data["requirements"]] == ["call_2"]
assert data["context"]["tool_names"] == ["pay_invoice"]
def test_repeated_hook_on_same_pause_creates_no_duplicate(self):
db = MagicMock()
rr, _, second = _second_pause_run_response()
first_id = create_approval_from_pause(db=db, run_response=rr)
again = create_approval_from_pause(db=db, run_response=rr)
assert again == first_id == second.approval_id
db.create_approval.assert_called_once()
@pytest.mark.asyncio
async def test_later_pause_creates_record_for_new_call_only_async(self):
db = MagicMock()
db.create_approval = AsyncMock()
rr, first, second = _second_pause_run_response()
result = await acreate_approval_from_pause(db=db, run_response=rr, agent_id="a1")
db.create_approval.assert_awaited_once()
data = db.create_approval.call_args[0][0]
assert result == data["id"] != "appr-1"
assert second.approval_id == result
assert first.approval_id == "appr-1"
assert data["tool_args"] == {"invoice": "INV-2"}
assert [r["tool_execution"]["tool_call_id"] for r in data["requirements"]] == ["call_2"]
@pytest.mark.asyncio
async def test_repeated_hook_on_same_pause_creates_no_duplicate_async(self):
db = MagicMock()
db.create_approval = AsyncMock()
rr, _, second = _second_pause_run_response()
first_id = await acreate_approval_from_pause(db=db, run_response=rr)
again = await acreate_approval_from_pause(db=db, run_response=rr)
assert again == first_id == second.approval_id
db.create_approval.assert_awaited_once()
class TestRunLevelFallbackNeverCrossesToolCalls:
def _unstamped_second_call(self):
rr, first, second = _second_pause_run_response()
db = MagicMock()
db.get_approval.side_effect = lambda aid: _record_for("call_1") if aid == "appr-1" else None
db.get_approvals.return_value = ([_record_for("call_1")], 1)
return db, rr, second
def test_record_for_another_call_does_not_resolve_unstamped_call(self):
db, rr, second = self._unstamped_second_call()
with pytest.raises(RuntimeError, match="No approval record found"):
check_and_apply_approval_resolution(db=db, run_id="run-123", run_response=rr)
assert second.confirmed is None
@pytest.mark.asyncio
async def test_record_for_another_call_does_not_resolve_unstamped_call_async(self):
db, rr, second = self._unstamped_second_call()
db.get_approvals = AsyncMock(return_value=([_record_for("call_1")], 1))
with pytest.raises(RuntimeError, match="No approval record found"):
await acheck_and_apply_approval_resolution(db=db, run_id="run-123", run_response=rr)
assert second.confirmed is None
def test_record_naming_the_call_still_resolves_it(self):
db = MagicMock()
db.get_approvals.return_value = ([_record_for("call_2")], 1)
t = FakeToolExecution(tool_call_id="call_2", approval_type="required", requires_confirmation=True)
check_and_apply_approval_resolution(db=db, run_id="r1", run_response=FakeRunResponse(tools=[t]))
assert t.confirmed is True
def test_stamped_record_gone_does_not_borrow_another_calls_record(self):
db = MagicMock()
db.get_approval.return_value = None
db.get_approvals.return_value = ([_record_for("call_1")], 1)
t = FakeToolExecution(
tool_call_id="call_2", approval_type="required", approval_id="appr-gone", requires_confirmation=True
)
with pytest.raises(RuntimeError, match="No approval record found"):
check_and_apply_approval_resolution(db=db, run_id="r1", run_response=FakeRunResponse(tools=[t]))
assert t.confirmed is None
@pytest.mark.asyncio
async def test_stamped_record_gone_does_not_borrow_another_calls_record_async(self):
db = MagicMock()
db.get_approval = AsyncMock(return_value=None)
db.get_approvals = AsyncMock(return_value=([_record_for("call_1")], 1))
t = FakeToolExecution(
tool_call_id="call_2", approval_type="required", approval_id="appr-gone", requires_confirmation=True
)
with pytest.raises(RuntimeError, match="No approval record found"):
await acheck_and_apply_approval_resolution(db=db, run_id="r1", run_response=FakeRunResponse(tools=[t]))
assert t.confirmed is None
class TestGateReviewRegressions:
def _required_and_audit(self):
required = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_1",
approval_type="required",
approval_id="appr-1",
requires_confirmation=True,
)
audit = FakeToolExecution(
tool_name="log_action", tool_call_id="call_2", approval_type="audit", requires_confirmation=True
)
return required, audit
def test_audit_tool_does_not_need_a_required_record(self):
db = MagicMock()
db.get_approval.return_value = _record_for("call_1")
db.get_approvals.return_value = ([_record_for("call_1")], 1)
required, audit = self._required_and_audit()
audit.confirmed = True
check_and_apply_approval_resolution(db=db, run_id="r1", run_response=FakeRunResponse(tools=[required, audit]))
assert required.confirmed is True
assert audit.confirmed is True
@pytest.mark.asyncio
async def test_audit_tool_does_not_need_a_required_record_async(self):
db = MagicMock()
db.get_approval = AsyncMock(return_value=_record_for("call_1"))
db.get_approvals = AsyncMock(return_value=([_record_for("call_1")], 1))
required, audit = self._required_and_audit()
audit.confirmed = True
await acheck_and_apply_approval_resolution(
db=db, run_id="r1", run_response=FakeRunResponse(tools=[required, audit])
)
assert required.confirmed is True
assert audit.confirmed is True
def _member_call_with_deleted_record(self):
te = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_2",
approval_type="required",
approval_id="appr-deleted",
requires_confirmation=True,
)
rr = FakeRunResponse(
tools=[], requirements=[FakeMemberRequirement(tool_execution=te, member_run_id="member-run")]
)
return te, rr
def test_member_record_for_another_call_does_not_resolve_deleted_record(self):
te, rr = self._member_call_with_deleted_record()
db = MagicMock()
db.get_approval.return_value = None
db.get_approvals.return_value = ([_record_for("call_1")], 1)
with pytest.raises(RuntimeError, match="No approval record found"):
check_and_apply_approval_resolution(db=db, run_id="team-run", run_response=rr)
assert te.confirmed is None
@pytest.mark.asyncio
async def test_member_record_for_another_call_does_not_resolve_deleted_record_async(self):
te, rr = self._member_call_with_deleted_record()
db = MagicMock()
db.get_approval = AsyncMock(return_value=None)
db.get_approvals = AsyncMock(return_value=([_record_for("call_1")], 1))
with pytest.raises(RuntimeError, match="No approval record found"):
await acheck_and_apply_approval_resolution(db=db, run_id="team-run", run_response=rr)
assert te.confirmed is None
def test_member_record_naming_the_call_still_resolves_it(self):
te, rr = self._member_call_with_deleted_record()
db = MagicMock()
db.get_approval.return_value = None
db.get_approvals.return_value = ([_record_for("call_2", record_id="appr-reissued")], 1)
check_and_apply_approval_resolution(db=db, run_id="team-run", run_response=rr)
assert te.confirmed is True
@dataclass
class FakeResolvableRequirement(FakeCallRequirement):
resolved: bool = False
def is_resolved(self) -> bool:
return self.resolved
class TestRecordCoversCurrentPause:
def test_plain_hitl_tool_paused_alongside_is_listed(self):
earlier_plain = FakeToolExecution(tool_name="notify", tool_call_id="call_0", requires_confirmation=True)
gated = FakeToolExecution(
tool_name="pay_invoice", tool_call_id="call_1", approval_type="required", requires_confirmation=True
)
plain = FakeToolExecution(tool_name="notify", tool_call_id="call_2", requires_confirmation=True)
rr = FakeRunResponse(
tools=[earlier_plain, gated, plain],
requirements=[
FakeResolvableRequirement(tool_execution=earlier_plain, resolved=True),
FakeResolvableRequirement(tool_execution=gated),
FakeResolvableRequirement(tool_execution=plain),
],
)
db = MagicMock()
approval_id = create_approval_from_pause(db=db, run_response=rr)
data = db.create_approval.call_args[0][0]
assert [r["tool_execution"]["tool_call_id"] for r in data["requirements"]] == ["call_1", "call_2"]
assert data["tool_name"] == "pay_invoice"
assert data["context"]["tool_names"] == ["pay_invoice"]
assert gated.approval_id == approval_id
assert plain.approval_id is None
class TestAuditToolBeforeRequiredTool:
def _run_response(self):
audit = FakeToolExecution(
tool_name="log_action",
tool_call_id="call_1",
approval_type="audit",
requires_confirmation=True,
confirmed=True,
result="logged",
)
gated = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_2",
tool_args={"invoice": "INV-1"},
approval_type="required",
requires_confirmation=True,
)
rr = FakeRunResponse(
tools=[audit, gated],
requirements=[FakeCallRequirement(tool_execution=audit), FakeCallRequirement(tool_execution=gated)],
)
return rr, audit, gated
def test_has_approval_requirement_sees_required_tool_after_audit_tool(self):
rr, _, _ = self._run_response()
assert _has_approval_requirement(rr.tools, rr.requirements) is True
def test_required_tool_after_audit_tool_gets_a_record(self):
db = MagicMock()
rr, audit, gated = self._run_response()
approval_id = create_approval_from_pause(db=db, run_response=rr)
db.create_approval.assert_called_once()
data = db.create_approval.call_args[0][0]
assert data["tool_name"] == "pay_invoice"
assert data["tool_args"] == {"invoice": "INV-1"}
assert gated.approval_id == approval_id
assert audit.approval_id is None
@pytest.mark.asyncio
async def test_required_tool_after_audit_tool_gets_a_record_async(self):
db = MagicMock()
db.create_approval = AsyncMock()
rr, audit, gated = self._run_response()
approval_id = await acreate_approval_from_pause(db=db, run_response=rr)
db.create_approval.assert_awaited_once()
assert db.create_approval.call_args[0][0]["tool_name"] == "pay_invoice"
assert gated.approval_id == approval_id
assert audit.approval_id is None
def test_audit_tool_paused_in_same_turn_is_not_stamped(self):
db = MagicMock()
audit = FakeToolExecution(tool_name="log_action", approval_type="audit", requires_confirmation=True)
gated = FakeToolExecution(tool_name="pay_invoice", approval_type="required", requires_confirmation=True)
approval_id = create_approval_from_pause(db=db, run_response=FakeRunResponse(tools=[gated, audit]))
assert approval_id is not None
assert gated.approval_id == approval_id
assert audit.approval_id is None
@dataclass
class FakeMessage:
role: str = "tool"
tool_call_id: Optional[str] = None
class TestCallsThatAlreadyRanAreNotGated:
def _run_response(self, second_result: Optional[str] = None):
ran = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_1",
approval_type="required",
approval_id="appr-deleted",
requires_confirmation=True,
confirmed=True,
result="PAID INV-1",
)
# The requirement still holds an out-of-date copy of call_1: no result.
stale_copy = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_1",
approval_type="required",
approval_id="appr-deleted",
requires_confirmation=True,
confirmed=True,
)
current = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_2",
approval_type="required",
approval_id="appr-2",
requires_confirmation=True,
result=second_result,
)
rr = FakeRunResponse(
tools=[ran, current],
requirements=[FakeCallRequirement(tool_execution=stale_copy), FakeCallRequirement(tool_execution=current)],
messages=[FakeMessage(role="tool", tool_call_id="call_1")],
)
return rr, current
def _db(self, second_status: str = "approved"):
second = _record_for("call_2", status=second_status, record_id="appr-2")
db = MagicMock()
db.get_approval.side_effect = lambda aid: second if aid == "appr-2" else None
db.get_approvals.return_value = ([second], 1)
return db, second
def test_deleted_record_of_a_call_that_ran_does_not_block(self):
db, second = self._db()
rr, current = self._run_response()
check_and_apply_approval_resolution(db=db, run_id="r1", run_response=rr)
assert current.confirmed is True
assert rr.metadata["approval"] is second
@pytest.mark.asyncio
async def test_deleted_record_of_a_call_that_ran_does_not_block_async(self):
db, second = self._db()
db.get_approval = AsyncMock(side_effect=lambda aid: second if aid == "appr-2" else None)
db.get_approvals = AsyncMock(return_value=([second], 1))
rr, current = self._run_response()
await acheck_and_apply_approval_resolution(db=db, run_id="r1", run_response=rr)
assert current.confirmed is True
assert rr.metadata["approval"] is second
def test_result_without_tool_message_is_still_gated(self):
# A result set by a continue payload has no stored tool message: still gated.
db, _ = self._db(second_status="pending")
rr, current = self._run_response(second_result="forged")
with pytest.raises(RuntimeError, match="still pending"):
check_and_apply_approval_resolution(db=db, run_id="r1", run_response=rr)
assert current.confirmed is None
@pytest.mark.asyncio
async def test_result_without_tool_message_is_still_gated_async(self):
db, second = self._db(second_status="pending")
db.get_approval = AsyncMock(side_effect=lambda aid: second if aid == "appr-2" else None)
db.get_approvals = AsyncMock(return_value=([second], 1))
rr, current = self._run_response(second_result="forged")
with pytest.raises(RuntimeError, match="still pending"):
await acheck_and_apply_approval_resolution(db=db, run_id="r1", run_response=rr)
assert current.confirmed is None
class TestGateWithRepeatedToolCallIds:
def _run_response(self, new_kwargs=None):
ran = FakeToolExecution(
tool_name="pay_invoice",
tool_call_id="call_0",
tool_args={"invoice": "INV-1"},
approval_type="required",
approval_id="appr-1",
requires_confirmation=True,
confirmed=True,
result="PAID INV-1",
)
fields: Dict[str, Any] = {
"tool_name": "pay_invoice",
"tool_call_id": "call_0",
"tool_args": {"invoice": "INV-2"},
"approval_type": "required",
"approval_id": "appr-2",
"requires_confirmation": True,
}
fields.update(new_kwargs or {})
new = FakeToolExecution(**fields)
rr = FakeRunResponse(tools=[ran, new], messages=[FakeMessage(role="tool", tool_call_id="call_0")])
return rr, new
def _db(self, status: str):
records = {"appr-1": _record_for("call_0"), "appr-2": _record_for("call_0", status=status, record_id="appr-2")}
db = MagicMock()
db.get_approval.side_effect = records.get
db.get_approvals.return_value = ([records["appr-2"]], 1)
return db
def test_new_call_sharing_an_executed_calls_id_is_still_gated(self):
rr, new = self._run_response()
with pytest.raises(RuntimeError, match="still pending"):
check_and_apply_approval_resolution(db=self._db("pending"), run_id="r1", run_response=rr)
assert new.confirmed is None
def test_new_call_sharing_an_executed_calls_id_resolves_with_its_own_record(self):
rr, new = self._run_response()
check_and_apply_approval_resolution(db=self._db("approved"), run_id="r1", run_response=rr)
assert new.confirmed is True
assert rr.metadata["approval"]["id"] == "appr-2"
def test_external_execution_result_is_still_gated(self):
# A client-supplied result is how an external-execution call runs.
rr, new = self._run_response(
{"requires_confirmation": None, "external_execution_required": True, "result": "client result"}
)
with pytest.raises(RuntimeError, match="still pending"):
check_and_apply_approval_resolution(db=self._db("pending"), run_id="r1", run_response=rr)
libs/agno/tests/unit/run/test_approval_multi_pause.py
+553
View File
@@ -0,0 +1,553 @@
"""Two approval-required tool calls in one run need two approval records.
A run that pauses for an @approval tool, is approved, and then pauses again for
a second call (different arguments) must create a second record naming the
second call. Continuing by run id without requirements must not execute the
second call under the first call's approval.
"""
import json
import time
from typing import Any, AsyncIterator, Iterator, List, Optional, Union
import pytest
from agno.agent import Agent
from agno.approval import approval
from agno.db.sqlite import SqliteDb
from agno.metrics import MessageMetrics
from agno.models.base import Model
from agno.models.response import ModelResponse, ModelResponseEvent
from agno.team import Team
from agno.tools import tool
PAID: List[str] = []
class _ScriptedModel(Model):
"""Emits scripted turns offline: ('tool', name, args, id) or ('content', text)."""
def __init__(self, script: List[tuple]):
super().__init__(id="scripted", name="scripted", provider="test")
self._script = list(script)
self._i = 0
def _next(self) -> ModelResponse:
turn = self._script[min(self._i, len(self._script) - 1)]
self._i += 1
if turn[0] == "tool":
_, name, args, tcid = turn
r = ModelResponse(role="assistant")
r.tool_calls = [{"id": tcid, "type": "function", "function": {"name": name, "arguments": json.dumps(args)}}]
else:
r = ModelResponse(content=turn[1], role="assistant")
r.event = ModelResponseEvent.assistant_response.value
r.response_usage = MessageMetrics(input_tokens=10, output_tokens=5, total_tokens=15)
return r
def invoke(self, *a, **k):
return self._next()
async def ainvoke(self, *a, **k):
return self._next()
def invoke_stream(self, *a, **k) -> Iterator[ModelResponse]:
yield self._next()
async def ainvoke_stream(self, *a, **k) -> AsyncIterator[ModelResponse]:
yield self._next()
def _parse_provider_response(self, response: Any, **k) -> ModelResponse:
return response if isinstance(response, ModelResponse) else ModelResponse()
def _parse_provider_response_delta(self, response: Any) -> ModelResponse:
return response if isinstance(response, ModelResponse) else ModelResponse()
@approval
@tool(requires_confirmation=True)
def pay_invoice(invoice: str) -> str:
"""Pay a vendor invoice.
Args:
invoice (str): Invoice number
"""
PAID.append(invoice)
return f"PAID {invoice}"
def _script() -> List[tuple]:
return [
("tool", "pay_invoice", {"invoice": "INV-1"}, "call_1"),
("tool", "pay_invoice", {"invoice": "INV-2"}, "call_2"),
("content", "Paid."),
]
def _build_agent(db: SqliteDb) -> Agent:
return Agent(id="payer", model=_ScriptedModel(_script()), tools=[pay_invoice], db=db, telemetry=False)
def _build_team(db: SqliteDb) -> Team:
helper = Agent(id="helper", model=_ScriptedModel([("content", "ok")]), telemetry=False)
return Team(
id="payer-team",
model=_ScriptedModel(_script()),
members=[helper],
tools=[pay_invoice],
db=db,
telemetry=False,
)
BUILDERS = [pytest.param(_build_agent, id="agent"), pytest.param(_build_team, id="team_level_tool")]
@pytest.fixture
def db(tmp_path):
PAID.clear()
return SqliteDb(db_file=str(tmp_path / "approvals.db"))
def _required_records(db: SqliteDb, run_id: str) -> List[dict]:
records, _ = db.get_approvals(run_id=run_id, approval_type="required")
return sorted(records, key=lambda r: r["created_at"])
def _named_calls(record: dict) -> List[Optional[str]]:
return [(r.get("tool_execution") or {}).get("tool_call_id") for r in record.get("requirements") or []]
def _approve_pending(db: SqliteDb, run_id: str) -> int:
pending, _ = db.get_approvals(run_id=run_id, approval_type="required", status="pending")
for record in pending:
db.update_approval(
record["id"],
expected_status="pending",
status="approved",
resolved_by="admin",
resolved_at=int(time.time()),
)
return len(pending)
def _assert_second_pause_has_own_record(db: SqliteDb, run_id: str) -> None:
records = _required_records(db, run_id)
assert len(records) == 2
assert records[0]["status"] == "approved"
assert _named_calls(records[0]) == ["call_1"]
assert records[1]["status"] == "pending"
assert _named_calls(records[1]) == ["call_2"]
assert records[1]["tool_args"] == {"invoice": "INV-2"}
def _assert_two_admin_decisions(db: SqliteDb, run_id: str) -> None:
records = _required_records(db, run_id)
assert [r["status"] for r in records] == ["approved", "approved"]
assert [r["resolved_by"] for r in records] == ["admin", "admin"]
@pytest.mark.parametrize("build", BUILDERS)
def test_each_call_needs_its_own_admin_decision(db, build):
entity: Union[Agent, Team] = build(db)
run = entity.run("Pay INV-1 and INV-2.")
assert run.is_paused
run_id, session_id = run.run_id, run.session_id
assert _approve_pending(db, run_id) == 1
run = entity.continue_run(run_id=run_id, session_id=session_id)
assert run.is_paused
assert PAID == ["INV-1"]
_assert_second_pause_has_own_record(db, run_id)
assert _approve_pending(db, run_id) == 1
run = entity.continue_run(run_id=run_id, session_id=session_id)
assert not run.is_paused
assert PAID == ["INV-1", "INV-2"]
_assert_two_admin_decisions(db, run_id)
@pytest.mark.parametrize("build", BUILDERS)
@pytest.mark.asyncio
async def test_each_call_needs_its_own_admin_decision_async(db, build):
entity: Union[Agent, Team] = build(db)
run = await entity.arun("Pay INV-1 and INV-2.")
assert run.is_paused
run_id, session_id = run.run_id, run.session_id
assert _approve_pending(db, run_id) == 1
run = await entity.acontinue_run(run_id=run_id, session_id=session_id)
assert run.is_paused
assert PAID == ["INV-1"]
_assert_second_pause_has_own_record(db, run_id)
assert _approve_pending(db, run_id) == 1
run = await entity.acontinue_run(run_id=run_id, session_id=session_id)
assert not run.is_paused
assert PAID == ["INV-1", "INV-2"]
_assert_two_admin_decisions(db, run_id)
@pytest.mark.parametrize("build", BUILDERS)
def test_second_call_does_not_run_while_its_approval_is_pending(db, build):
entity: Union[Agent, Team] = build(db)
run = entity.run("Pay INV-1 and INV-2.")
run_id, session_id = run.run_id, run.session_id
_approve_pending(db, run_id)
entity.continue_run(run_id=run_id, session_id=session_id)
# The admin has not decided on INV-2: continuing must not pay it.
try:
run = entity.continue_run(run_id=run_id, session_id=session_id)
assert run.is_paused
except ValueError:
pass
assert PAID == ["INV-1"]
assert _required_records(db, run_id)[1]["status"] == "pending"
# ---------------------------------------------------------------------------
# A required call followed by an audit call, and a member's deleted record
# ---------------------------------------------------------------------------
LOGGED: List[str] = []
@approval(type="audit")
@tool(requires_confirmation=True)
def log_action(note: str) -> str:
"""Log an action for the audit trail.
Args:
note (str): What happened
"""
LOGGED.append(note)
return "logged"
def _audit_script() -> List[tuple]:
return [
("tool", "pay_invoice", {"invoice": "INV-1"}, "call_1"),
("tool", "log_action", {"note": "paid INV-1"}, "call_2"),
("content", "Done."),
]
def _build_agent_with_audit(db: SqliteDb) -> Agent:
return Agent(
id="payer", model=_ScriptedModel(_audit_script()), tools=[pay_invoice, log_action], db=db, telemetry=False
)
def _build_team_with_audit(db: SqliteDb) -> Team:
helper = Agent(id="helper", model=_ScriptedModel([("content", "ok")]), telemetry=False)
return Team(
id="payer-team",
model=_ScriptedModel(_audit_script()),
members=[helper],
tools=[pay_invoice, log_action],
db=db,
telemetry=False,
)
@pytest.mark.parametrize(
"build",
[pytest.param(_build_agent_with_audit, id="agent"), pytest.param(_build_team_with_audit, id="team_level_tool")],
)
def test_audit_call_after_approved_call_can_be_continued(db, build):
LOGGED.clear()
entity: Union[Agent, Team] = build(db)
run = entity.run("Pay INV-1 and log it.")
assert _approve_pending(db, run.run_id) == 1
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
assert PAID == ["INV-1"]
# The audit call never gets a required record; the client confirms it directly.
decisions = [r for r in run.requirements if r.tool_execution.tool_call_id == "call_2"]
decisions[0].confirm()
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id, requirements=decisions)
assert not run.is_paused
assert LOGGED == ["paid INV-1"]
assert PAID == ["INV-1"]
def test_member_call_does_not_run_when_its_record_is_deleted(db):
payer = Agent(id="payer", model=_ScriptedModel(_script()), tools=[pay_invoice], db=db, telemetry=False)
team = Team(
id="finance-team",
model=_ScriptedModel(
[("tool", "delegate_task_to_member", {"member_id": "payer", "task": "Pay"}, "deleg_1"), ("content", "ok")]
),
members=[payer],
db=db,
telemetry=False,
)
run = team.run("Pay INV-1 and INV-2.")
pending, _ = db.get_approvals(approval_type="required", status="pending")
assert [p["tool_args"] for p in pending] == [{"invoice": "INV-1"}]
_approve_pending(db, pending[0]["run_id"])
run = team.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
assert PAID == ["INV-1"]
# Delete the second call's pending record: the first call's approval must not run it.
(second,) = db.get_approvals(approval_type="required", status="pending")[0]
assert second["tool_args"] == {"invoice": "INV-2"}
db.delete_approval(second["id"])
try:
run = team.continue_run(run_id=run.run_id, session_id=run.session_id)
assert run.is_paused
except ValueError:
pass
assert PAID == ["INV-1"]
# ---------------------------------------------------------------------------
# Sending every requirement back must not re-run a call that already ran
# ---------------------------------------------------------------------------
AUDIT_BUILDERS = [
pytest.param(_build_agent_with_audit, id="agent"),
pytest.param(_build_team_with_audit, id="team_level_tool"),
]
def _confirm_open(run) -> None:
for req in run.active_requirements:
if req.needs_confirmation:
req.confirm()
@pytest.mark.parametrize("build", AUDIT_BUILDERS)
def test_resending_all_requirements_does_not_rerun_executed_call(db, build):
LOGGED.clear()
entity: Union[Agent, Team] = build(db)
run = entity.run("Pay INV-1 and log it.")
_approve_pending(db, run.run_id)
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
assert PAID == ["INV-1"]
# The client confirms the open call and sends back the run's whole requirements
# list, including the requirement of the INV-1 call that already ran.
_confirm_open(run)
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id, requirements=run.requirements)
assert not run.is_paused
assert PAID == ["INV-1"]
assert LOGGED == ["paid INV-1"]
@pytest.mark.parametrize("build", AUDIT_BUILDERS)
@pytest.mark.asyncio
async def test_resending_all_requirements_does_not_rerun_executed_call_async(db, build):
LOGGED.clear()
entity: Union[Agent, Team] = build(db)
run = await entity.arun("Pay INV-1 and log it.")
_approve_pending(db, run.run_id)
run = await entity.acontinue_run(run_id=run.run_id, session_id=run.session_id)
assert PAID == ["INV-1"]
_confirm_open(run)
run = await entity.acontinue_run(run_id=run.run_id, session_id=run.session_id, requirements=run.requirements)
assert not run.is_paused
assert PAID == ["INV-1"]
assert LOGGED == ["paid INV-1"]
def _stale_and_executed():
from agno.models.response import ToolExecution
executed = ToolExecution(
tool_call_id="call_1", tool_name="pay_invoice", requires_confirmation=True, confirmed=True, result="PAID INV-1"
)
stale = ToolExecution(tool_call_id="call_1", tool_name="pay_invoice", requires_confirmation=True, confirmed=True)
open_call = ToolExecution(tool_call_id="call_2", tool_name="log_action", requires_confirmation=True)
decided = ToolExecution(tool_call_id="call_2", tool_name="log_action", requires_confirmation=True, confirmed=True)
return executed, stale, open_call, decided
def test_agent_merge_keeps_executed_call():
from agno.agent._run import _apply_requirement_tools
from agno.run.agent import RunOutput
from agno.run.requirement import RunRequirement
executed, stale, open_call, decided = _stale_and_executed()
run = RunOutput(run_id="r", tools=[executed, open_call])
_apply_requirement_tools(run, [RunRequirement(tool_execution=stale), RunRequirement(tool_execution=decided)])
assert run.tools[0] is executed
assert run.tools[1] is decided
def test_team_merge_keeps_executed_call():
from agno.team._run import _merge_tools_preserving_approval
executed, stale, open_call, decided = _stale_and_executed()
merged = _merge_tools_preserving_approval([executed, open_call], {"call_1": stale, "call_2": decided})
assert merged[0] is executed
assert merged[1] is decided
# ---------------------------------------------------------------------------
# Post-hooks see the record of the call the continue runs
# ---------------------------------------------------------------------------
SEEN_BY_HOOK: List[Optional[str]] = []
def _record_seen_by_post_hook(run_output) -> None:
approval_record = (run_output.metadata or {}).get("approval") or {}
SEEN_BY_HOOK.append((approval_record.get("tool_args") or {}).get("invoice"))
@pytest.mark.parametrize("build", BUILDERS)
def test_post_hook_sees_the_record_for_the_call_being_run(db, build):
SEEN_BY_HOOK.clear()
entity: Union[Agent, Team] = build(db)
entity.post_hooks = [_record_seen_by_post_hook]
run = entity.run("Pay INV-1 and INV-2.")
for _ in range(2):
_approve_pending(db, run.run_id)
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
# The hook ran with the continued run's output, which carries the record attached by the gate.
assert ((run.metadata or {}).get("approval") or {}).get("tool_args") == {"invoice": PAID[-1]}
assert not run.is_paused
assert PAID == ["INV-1", "INV-2"]
assert SEEN_BY_HOOK[-1] == "INV-2"
# ---------------------------------------------------------------------------
# An audit call earlier in the run must not hide a later approval-required call
# ---------------------------------------------------------------------------
def _audit_first_script() -> List[tuple]:
return [
("tool", "log_action", {"note": "start"}, "call_1"),
("tool", "pay_invoice", {"invoice": "INV-1"}, "call_2"),
("content", "Done."),
]
def _build_agent_audit_first(db: SqliteDb) -> Agent:
return Agent(
id="payer", model=_ScriptedModel(_audit_first_script()), tools=[log_action, pay_invoice], db=db, telemetry=False
)
def _build_team_audit_first(db: SqliteDb) -> Team:
helper = Agent(id="helper", model=_ScriptedModel([("content", "ok")]), telemetry=False)
return Team(
id="payer-team",
model=_ScriptedModel(_audit_first_script()),
members=[helper],
tools=[log_action, pay_invoice],
db=db,
telemetry=False,
)
@pytest.mark.parametrize(
"build",
[pytest.param(_build_agent_audit_first, id="agent"), pytest.param(_build_team_audit_first, id="team_level_tool")],
)
def test_approval_call_after_audit_call_gets_its_own_record(db, build):
LOGGED.clear()
entity: Union[Agent, Team] = build(db)
run = entity.run("Log, then pay INV-1.")
decisions = [r for r in run.requirements if not r.is_resolved()]
for req in decisions:
req.confirm()
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id, requirements=decisions)
assert run.is_paused
assert LOGGED == ["start"]
assert PAID == []
(record,) = _required_records(db, run.run_id)
assert record["status"] == "pending"
assert _named_calls(record) == ["call_2"]
assert _approve_pending(db, run.run_id) == 1
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert PAID == ["INV-1"]
# ---------------------------------------------------------------------------
# Deleting the record of a call that already ran must not block the next call
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("build", BUILDERS)
def test_deleted_record_of_executed_call_does_not_block_next_call(db, build):
entity: Union[Agent, Team] = build(db)
run = entity.run("Pay INV-1 and INV-2.")
(first,) = _required_records(db, run.run_id)
_approve_pending(db, run.run_id)
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
assert PAID == ["INV-1"]
# An admin deletes INV-1's record after it ran, then approves INV-2.
assert db.delete_approval(first["id"])
assert _approve_pending(db, run.run_id) == 1
run = entity.continue_run(run_id=run.run_id, session_id=run.session_id)
assert not run.is_paused
assert PAID == ["INV-1", "INV-2"]
assert run.metadata["approval"]["tool_args"] == {"invoice": "INV-2"}
def test_agent_merge_with_repeated_tool_call_id():
# Some providers send no id and a fallback like call_{i} repeats across turns.
from agno.agent._run import _apply_requirement_tools
from agno.models.response import ToolExecution
from agno.run.agent import RunOutput
from agno.run.requirement import RunRequirement
paid = ToolExecution(
tool_call_id="call_0",
tool_name="pay_invoice",
tool_args={"invoice": "INV-1"},
requires_confirmation=True,
confirmed=True,
result="PAID INV-1",
)
new = ToolExecution(
tool_call_id="call_0", tool_name="pay_invoice", tool_args={"invoice": "INV-2"}, requires_confirmation=True
)
decided = ToolExecution(
tool_call_id="call_0",
tool_name="pay_invoice",
tool_args={"invoice": "INV-2"},
requires_confirmation=True,
confirmed=True,
)
run = RunOutput(run_id="r", tools=[paid, new])
_apply_requirement_tools(run, [RunRequirement(tool_execution=decided)])
assert run.tools[0] is paid
assert run.tools[1] is decided
assert run.tools[1].confirmed is True
def test_team_merge_with_repeated_tool_call_id():
from agno.models.response import ToolExecution
from agno.team._run import _merge_tools_preserving_approval
paid = ToolExecution(
tool_call_id="call_0",
tool_name="pay_invoice",
tool_args={"invoice": "INV-1"},
requires_confirmation=True,
confirmed=True,
result="PAID INV-1",
)
new = ToolExecution(
tool_call_id="call_0", tool_name="pay_invoice", tool_args={"invoice": "INV-2"}, requires_confirmation=True
)
decided = ToolExecution(
tool_call_id="call_0",
tool_name="pay_invoice",
tool_args={"invoice": "INV-2"},
requires_confirmation=True,
confirmed=True,
)
merged = _merge_tools_preserving_approval([paid, new], {"call_0": decided})
assert merged[0] is paid
assert merged[1] is decided