Significant-Gravitas/AutoGPT
1
0 Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-11-11 04:31:34 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

feat(backend/copilot): the chat's approval gate answers the review pauses of workflows it started (#15564)

Browse Source 
When a workflow Otto started reaches a step that can't be undone (a
block marked irreversible: sending mail, posting, paying), the chat that
started the run now decides it with its own approval gate, the same way
it decides a direct call of that block; and when the gate says yes to a
direct block call, the block no longer asks a second time.

Resolves #15563

### Why / What / How

**Why:** two approval systems ask the user today. The gate decides
Otto's own tool calls by the chat's mode (Ask first, Auto, Unsupervised)
and the rules the user set from earlier cards. The older human review
pauses any irreversible step of a workflow and waits on the run page or
Home, under its own "approve all future" toggle and none of the chat's
rules. So a workflow Otto runs from a chat asks somewhere else, under
different rules, and a direct block call the gate already allowed (by a
rule, by Unsupervised, or by the supervisor) is asked about again.

**What:** for a run started from a chat whose gate is active, an
irreversible step's pause goes to that chat's gate:
- a rule or the mode says run: the step runs at once, and the run page
records an approved review "Allowed by Otto: …";
- anything says ask: the gate's ordinary card appears in that chat;
approving or declining it answers the step and resumes the run. As on
the run page, the card takes its answer once the run has paused; while
other steps of the run are still running, an answer is refused and the
card stays open. The card offers the same rules as an action card
("Approve … from now on", "Let Otto judge … from now on", for this chat,
the Expert's every chat or the whole team), set on the block's own key,
so one rule covers later steps, later runs and direct calls of that
block. Home shows the card, not the step's row as well. Answering the
row on the run page instead closes the card, and stopping the run
deletes it.

For direct calls, any verdict an active gate reaches (a card, an allow
rule, a mode that says run, a supervisor allow) now satisfies the
block's own review. Runs started from the run page, a schedule, a
webhook, the API, an automation chat, or a chat whose gate is inactive
keep today's human review unchanged, as does any pause in a run whose
owner has no active plan (its resume meets the paywall like any other).
The Human-in-the-loop block's questions and expert spend approvals are
untouched.

**How:** the step still parks exactly as today, and the gate decides
about it the moment it parks.
- `_add_graph_execution` stamps `ExecutionContext.approval_chat_id`: a
COPILOT trigger's chat on create, the parent's stamp for a sub-graph,
and on every resume the chat at the end of the trigger chain
(`starting_chat`, which `_started_from_attended_chat` now shares). Dry
runs are never stamped. It also keeps the workflow's own
`sensitive_action_safe_mode` apart from the pause the chat forces.
- `HITLReviewHelper._handle_review_request` hands the pause to
`copilot/gate/workflow.py` after `check_approval` and only when the
block passed itself, so only an irreversible block's own pause can
route. `review_route` checks stamp → flag → the owner's plan
(`is_user_paywalled`, before any rule or supervisor call; a failed
lookup also leaves a human review) → chat → active gate; `judge_step`
judges the step through `rule_on`, the rule → mode → supervisor section
extracted from `check_action` so both paths share one ordering, and
hands the supervisor the block's tier and name as a direct call does.
The step's subject is the block's own `block:<id>`, so a rule set from
either kind of card covers both.
- A run verdict writes an approved, processed row
(`record_gate_approval`); a decision already on the row stands. An ask
writes the waiting row without its alert and opens the card (id derived
from the step, `step_card_id`); if the card cannot open, the row alerts
as an ordinary human review.
- The review route's chat branch refuses a workflow card whose run has
not paused, answers each workflow card's row, consumes the card once the
row has the answer (a failed update returns the card to waiting and the
request fails, so answering again finishes it), sets the chat's ask rule
on a decline, and resumes the run through `resume_reviewed_execution`
(extracted from the graph branch). Workflow cards are left out of the
held-call wake, because a step is not a model's tool call and its run
has its own resume.
- Direct path: `Decision.decided` → `GATE_DECIDED` (stripped from the
model's arguments like `GATE_APPROVED`) → `RunBlockTool` skips
`check_hitl_review` when the flag is on. `run_agent` ignores it: only a
card approval lifts a workflow run's forced pause, and its steps' pauses
now go to the gate on their own.
- `view_agent_output` and `run_agent` say a parked run "is waiting for
the user's approval in this chat" when its pause has a card.

### Implementation decisions

Built as the design recommended; each is a one-line switch:
1. **Unsupervised runs an irreversible block without asking**, in
workflows and in direct calls, because the mode's table is now final.
`UNSUPERVISED_RUNS_IRREVERSIBLE_BLOCKS = False` in
`copilot/gate/__init__.py` makes Unsupervised ask with a card instead.
2. **When a workflow has its own "Ask before irreversible actions"
setting on, running it from a chat still asks before those steps, even
in Unsupervised.** Switched off, the chat's mode decides instead, so
Unsupervised would run them without asking. Either way, an "always
allow" rule the user set in that chat for the block still lets it run,
and where the mode hands a step to Otto, Otto still judges it; only the
mode's plain "run" becomes an ask. Switch: `WORKFLOW_SETTING_IS_A_FLOOR`
in `copilot/gate/workflow.py`.

### Rollout and undo

Behind `copilot-unified-approvals`, read with `default=True`: a flag
PostHog does not know is ON, and a read that fails is ON. It only
reaches users the gate already reaches, the `copilot-auto-mode` cohort.
- **Undo:** create `copilot-unified-approvals` in PostHog with value
false, or deploy with `FORCE_FLAG_COPILOT_UNIFIED_APPROVALS=false`. New
pauses are human reviews again and direct calls go back to the second
ask after a rule, mode or supervisor yes. Cards already open still
finish their runs: nothing on the answer side reads the flag. Nothing is
migrated.
- **Runs parked on a human review at deploy stay human reviews**,
answered on the run page or Home as today; nothing backfills cards. A
run between pauses at deploy has no stamp in its queued payload, so its
next pause in that pass is a human review too. Every later resume stamps
it.
- **Rolling-deploy window:** an API pod still on the previous deploy
that answers a new workflow card treats it as a held call whose details
were lost: the chat says "Nothing ran: the approved action's details
were lost…" and the step's row stays waiting on the run page. It fails
closed, and only during the deploy. An executor on this deploy reaching
a database manager on the previous one gets a 404 for
`record_gate_approval`, which the RPC client does not retry, so that
step becomes a human review; the older manager also ignores the
silent-write keyword, so an asked step's row alerts beside its card.

### Verified

I ran the gate, review-route, review-helper, human-review, executor and
touched copilot tool suites with `architecture_test.py` and
`test_block.py` on the merged tree (4,872 passed, 88 skipped), then all
of `backend/copilot` (9,495 passed; the two local failures are this
machine's database, and both pass in CI), including nine integration
tests that park a real step against Postgres, Redis and the database
manager and answer its card through the real `/api/review/action` route.
Every new test was run against a mutation of the code it guards: 44 of
45 went red, and the one that did not is an equivalent mutant (a guard
the route's catch-all already covers). The frontend approval-queue tests
render the real step card. I only reasoned about the rolling-deploy
behaviour; nothing here was run on a deployed environment. Full tables
and screenshots are in the PR comments.

### Changes 🏗️

- `util/feature_flag`: `Flag.COPILOT_UNIFIED_APPROVALS`.
- `data/execution.py`: `ExecutionContext.approval_chat_id`,
`graph_sensitive_action_safe_mode`.
- `executor/utils.py`: the stamp, and `starting_chat`.
- `copilot/gate/workflow.py` (new): `review_route`, `judge_step`,
`open_step_card`, `step_card_payload`, `finish_step_cards`,
`close_step_cards`, `card_chat_for_run`; `copilot/constants.py`:
`step_card_id`.
- `copilot/gate/__init__.py`: `rule_on` extracted from `check_action`,
`Decision.decided`, `unified_approvals`; `copilot/gate/review.py`: the
card's `workflow` field.
- `blocks/helpers/review.py`, `blocks/_base.py`: the seam.
- `data/human_review.py`, `data/db_manager.py`: `record_gate_approval`,
a row written without its alert, Home's one-item filter, the stop path
deleting cards.
- Review route: workflow-card rule keys from the card,
`finish_step_cards`, `resume_reviewed_execution`, the run page closing
cards.
- `copilot/tools/`: `GATE_DECIDED` through `base.py`,
`run_capability.py`, `run_block.py`; `run_agent.py` ignores it; the
waiting wording in `agent_output.py` and `run_agent.py`.
- Frontend: no code change. `realCards.json` and `referenceCards.json`
are regenerated by their builders (each payload gains `workflow: null`,
and the real cards gain a "Workflow Step" card built by
`step_card_payload`), and `ApprovalQueue.stories.tsx` gains
`RealWorkflowStep`.

### Agents and large language models used

- Claude Code with Claude Opus 5.5

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
- [x] A step of an Otto-started run parks, opens its card in the
starting chat, and Home lists the card, not the row (integration, real
route)
- [x] Approving the card answers the row and resumes the run; declining
rejects the step and sets the chat's ask rule; the card never becomes a
held call
- [x] Answering the row on the run page or stopping the run closes the
card
- [x] An allow rule runs the step with "Allowed by Otto: …" on its row;
a missing stamp, a deleted chat or a card that cannot open leaves an
alerted human review
- [x] A card opened before the flag went off still resumes its run; the
next pause is a human review
- [x] Direct calls: a rule or Unsupervised yes skips the block's own
review with the flag on, not with it off or with the gate inactive; a
forged `_gate_decided` changes nothing; `run_agent` keeps its forced
pause
- [x] The executor stamps the starting chat on create, sub-graph and
resume, never on a dry run or a broken chain

#### For configuration changes:

- [x] `.env.default` is updated or already compatible with my changes
- [x] `docker-compose.yml` is updated or already compatible with my
changes
- [x] I have included a list of my configuration changes in the PR
description (under **Changes**): one feature flag, on by default, no new
environment variable.

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

---------

Co-authored-by: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Reinier van der LeerandClaude Opus 5.5 authored and GitHub committed 2026-10-19 18:55:17 +00:00
1 parent 6c89b1607f
commit ede2df2aec
31 files changed
+2894 -191
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

autogpt_platform/backend/backend/api/conftest.py
+1
View File
@@ -21,6 +21,7 @@ _PAYWALL_CHECK_SITES = (
"backend.copilot.rate_limit.is_user_paywalled",
"backend.executor.utils.is_user_paywalled",
"backend.api.features.integrations.router.is_user_paywalled",
"backend.copilot.gate.workflow.is_user_paywalled",
)
autogpt_platform/backend/backend/api/features/graph_executions/review/routes.py
+91 -39
View File
@@ -10,6 +10,13 @@ from backend.copilot.constants import legacy_chat_session_id, parse_node_id_from
from backend.copilot.gate.chat_rules import set_answer_rules as set_chat_rules
from backend.copilot.gate.held import subject_keys as held_subject_keys
from backend.copilot.gate.held import wake as wake_for_held_calls
from backend.copilot.gate.workflow import (
close_step_cards,
finish_step_cards,
is_step_card,
step_of,
step_rule_keys,
)
from backend.copilot.rate_limit import enforce_payment_paywall
from backend.data.execution import (
ExecutionContext,
@@ -230,6 +237,16 @@ async def process_review_action(
graph_exec_id, chat_session_id = next(iter(scopes))
# A run still running would finish its pass with nothing waiting, the step
# unrun; a queued or ended run is safe to answer.
for review in reviews_map.values():
step = step_of(review)
if step is not None and await _running(user_id, step.graph_exec_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="This workflow is still running other steps. Answer once it pauses.",
)
# Validate execution status for graph executions; a chat has none
if graph_exec_id is not None:
graph_exec_meta = await get_graph_execution_meta(
@@ -268,10 +285,10 @@ async def process_review_action(
auto_approve_requests[review.node_exec_id] = review.auto_approve_future
# Read before the answer lands: a turn it wakes claims the held calls.
ruled = [r.node_exec_id for r in request.reviews if r.chat_rule]
chat_rule_keys = (
await held_subject_keys(
chat_session_id, [r.node_exec_id for r in request.reviews if r.chat_rule]
)
await held_subject_keys(chat_session_id, ruled)
| step_rule_keys(reviews_map[node_exec_id] for node_exec_id in ruled)
if chat_session_id is not None
else {}
)
@@ -371,49 +388,48 @@ async def process_review_action(
{review.node_exec_id: review.chat_rule_scope for review in request.reviews},
)
# A held call finishes on its own: the answer starts the chat's next turn.
if chat_session_id is not None and updated_reviews:
await wake_for_held_calls(user_id, chat_session_id, updated_reviews.values())
# A workflow step's card answers the step's own row; its run resumes
# below, so the card never wakes the chat.
finished = await finish_step_cards(user_id, updated_reviews.values())
for run in {step.graph_exec_id: step for step in finished.answered}.values():
meta = await get_graph_execution_meta(
user_id=user_id, execution_id=run.graph_exec_id
)
if (
meta is not None
and meta.status in (ExecutionStatus.REVIEW, ExecutionStatus.INCOMPLETE)
and not await has_pending_reviews_for_graph_exec(run.graph_exec_id)
):
await resume_reviewed_execution(
user_id, run.graph_exec_id, run.graph_id, run.graph_version
)
# A held call finishes on its own: the answer starts the chat's next turn.
await wake_for_held_calls(
user_id,
chat_session_id,
[r for r in updated_reviews.values() if not is_step_card(r)],
)
if finished.failed:
# Each card is kept, so answering it again finishes the job.
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="The answer could not reach its workflow step. Try again.",
)
# Resume graph execution only for real graph executions; a chat is resumed
# by the LLM calling resume_capability with the review_id
if graph_exec_id is not None and updated_reviews:
still_has_pending = await has_pending_reviews_for_graph_exec(graph_exec_id)
if not still_has_pending:
await close_step_cards(user_id, updated_reviews)
if not await has_pending_reviews_for_graph_exec(graph_exec_id):
first_review = next(iter(updated_reviews.values()))
assert first_review.graph_id and first_review.graph_version is not None
try:
user = await get_user_by_id(user_id)
settings = await get_graph_settings(
user_id=user_id,
graph_id=first_review.graph_id,
graph_version=first_review.graph_version,
)
user_timezone = (
user.timezone if user.timezone != USER_TIMEZONE_NOT_SET else "UTC"
)
workspace = await get_or_create_workspace(user_id)
execution_context = ExecutionContext(
human_in_the_loop_safe_mode=settings.human_in_the_loop_safe_mode,
sensitive_action_safe_mode=settings.sensitive_action_safe_mode,
user_timezone=user_timezone,
workspace_id=workspace.id,
)
await add_graph_execution(
graph_id=first_review.graph_id,
user_id=user_id,
graph_exec_id=graph_exec_id,
execution_context=execution_context,
)
logger.info(f"Resumed execution {graph_exec_id}")
except Exception as e:
logger.error(f"Failed to resume execution {graph_exec_id}: {str(e)}")
await resume_reviewed_execution(
user_id,
graph_exec_id,
first_review.graph_id,
first_review.graph_version,
)
# Build error message if auto-approvals failed
error_message = None
@@ -429,3 +445,39 @@ async def process_review_action(
failed_count=auto_approval_failed_count,
error=error_message,
)
async def _running(user_id: str, graph_exec_id: str) -> bool:
meta = await get_graph_execution_meta(user_id=user_id, execution_id=graph_exec_id)
return meta is not None and meta.status == ExecutionStatus.RUNNING
async def resume_reviewed_execution(
user_id: str, graph_exec_id: str, graph_id: str, graph_version: int
) -> None:
"""Re-queue a run whose reviews are all answered; its paused steps re-run
and find their answers."""
try:
user = await get_user_by_id(user_id)
settings = await get_graph_settings(
user_id=user_id, graph_id=graph_id, graph_version=graph_version
)
user_timezone = (
user.timezone if user.timezone != USER_TIMEZONE_NOT_SET else "UTC"
)
workspace = await get_or_create_workspace(user_id)
execution_context = ExecutionContext(
human_in_the_loop_safe_mode=settings.human_in_the_loop_safe_mode,
sensitive_action_safe_mode=settings.sensitive_action_safe_mode,
user_timezone=user_timezone,
workspace_id=workspace.id,
)
await add_graph_execution(
graph_id=graph_id,
user_id=user_id,
graph_exec_id=graph_exec_id,
execution_context=execution_context,
)
logger.info(f"Resumed execution {graph_exec_id}")
except Exception as e:
logger.error(f"Failed to resume execution {graph_exec_id}: {str(e)}")
autogpt_platform/backend/backend/api/features/graph_executions/review/workflow_card_db_test.py
+603
View File
@@ -0,0 +1,603 @@
"""A workflow step Otto started, parked on its card in the starting chat, against
Postgres, Redis and the real review route.
The step pauses through the block's own review; only the queue publish of the
resume is captured, so everything a card answer touches is what ships.
"""
import uuid
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, patch
import httpx
import pytest
import pytest_asyncio
from prisma.enums import AgentExecutionStatus, ReviewStatus
from prisma.models import (
AgentGraph,
AgentGraphExecution,
AgentNode,
AgentNodeExecution,
User,
)
from backend.api.rest_api import app
from backend.blocks.discord.bot_blocks import SendDiscordMessageBlock
from backend.blocks.helpers.review import HITLReviewHelper
from backend.copilot.constants import step_card_id
from backend.copilot.gate import RUNS_KEY, chat_rules, check_action, held
from backend.copilot.gate.classifier import Judgement
from backend.copilot.model import (
ChatSession,
get_chat_session,
update_session_autopilot_mode,
upsert_chat_session,
)
from backend.copilot.tools.run_capability import RunCapabilityTool
from backend.data.execution import ExecutionContext
from backend.data.human_review import (
cancel_pending_reviews_for_execution,
get_pending_reviews_for_user,
get_reviews_by_node_exec_ids,
has_pending_reviews_for_graph_exec,
)
from backend.data.redis_client import get_redis_async
from backend.util.exceptions import BlockExecutionError
from backend.util.feature_flag import Flag
_ROUTES = "backend.api.features.graph_executions.review.routes"
_BLOCK = SendDiscordMessageBlock()
_INPUT = {"channel_name": "general", "message_content": "The digest is out"}
_KEY = f"block:{_BLOCK.id}"
pytestmark = pytest.mark.asyncio(loop_scope="session")
@pytest_asyncio.fixture(loop_scope="session")
async def client(server, mock_jwt_user) -> AsyncGenerator[httpx.AsyncClient, None]:
from autogpt_libs.auth.jwt_utils import get_jwt_payload
app.dependency_overrides[get_jwt_payload] = mock_jwt_user["get_jwt_payload"]
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="http://test"
) as http_client:
yield http_client
app.dependency_overrides.pop(get_jwt_payload, None)
@pytest.fixture
def flags():
"""Auto mode on for this user; unified approvals as each test sets it."""
values = {Flag.COPILOT_AUTO_MODE: True, Flag.COPILOT_UNIFIED_APPROVALS: True}
async def read(flag, user_id, default=False):
return values[flag]
with patch("backend.copilot.gate.is_feature_enabled", AsyncMock(side_effect=read)):
yield values
@pytest.fixture
def resumed():
with patch(f"{_ROUTES}.add_graph_execution", AsyncMock()) as add:
yield add
class _Run:
def __init__(self, user_id: str, chat_id: str | None) -> None:
self.user_id = user_id
self.graph_id = str(uuid.uuid4())
self.node_id = str(uuid.uuid4())
self.graph_exec_id = str(uuid.uuid4())
self.node_exec_id = str(uuid.uuid4())
self.chat_id = chat_id
@property
def card_id(self) -> str:
return step_card_id(self.user_id, self.node_exec_id)
async def create(
self, status: AgentExecutionStatus = AgentExecutionStatus.REVIEW
) -> "_Run":
await AgentGraph.prisma().create(
data={
"id": self.graph_id,
"version": 1,
"name": "Morning digest",
"description": "",
"userId": self.user_id,
"isActive": True,
}
)
await AgentNode.prisma().create(
data={
"id": self.node_id,
"agentBlockId": _BLOCK.id,
"agentGraphId": self.graph_id,
"agentGraphVersion": 1,
}
)
await AgentGraphExecution.prisma().create(
data={
"id": self.graph_exec_id,
"agentGraphId": self.graph_id,
"agentGraphVersion": 1,
"userId": self.user_id,
"executionStatus": status,
"triggerSource": "copilot",
"triggerRef": self.chat_id,
}
)
await AgentNodeExecution.prisma().create(
data={
"id": self.node_exec_id,
"agentGraphExecutionId": self.graph_exec_id,
"agentNodeId": self.node_id,
"executionStatus": AgentExecutionStatus.RUNNING,
}
)
return self
async def next_step(self) -> "_Run":
"""A later pause of the same block in this run."""
self.node_exec_id = str(uuid.uuid4())
await AgentNodeExecution.prisma().create(
data={
"id": self.node_exec_id,
"agentGraphExecutionId": self.graph_exec_id,
"agentNodeId": self.node_id,
"executionStatus": AgentExecutionStatus.RUNNING,
}
)
return self
async def step(self, stamped: bool = True) -> tuple[bool, dict]:
"""The block's own review pause, as the executor reaches it."""
ctx = ExecutionContext(
user_id=self.user_id,
graph_id=self.graph_id,
graph_exec_id=self.graph_exec_id,
graph_version=1,
sensitive_action_safe_mode=True,
graph_sensitive_action_safe_mode=False,
approval_chat_id=self.chat_id if stamped else None,
)
return await _BLOCK.is_block_exec_need_review(
_INPUT,
user_id=self.user_id,
node_id=self.node_id,
node_exec_id=self.node_exec_id,
graph_exec_id=self.graph_exec_id,
graph_id=self.graph_id,
graph_version=1,
execution_context=ctx,
)
async def row(self, node_exec_id: str | None = None):
node_exec_id = node_exec_id or self.node_exec_id
rows = await get_reviews_by_node_exec_ids([node_exec_id], self.user_id)
return rows.get(node_exec_id)
async def card(self):
return await self.row(self.card_id)
async def _chat(user_id: str) -> ChatSession:
session = await upsert_chat_session(ChatSession.new(user_id=user_id, dry_run=False))
await update_session_autopilot_mode(session.session_id, user_id, "ask_first")
reloaded = await get_chat_session(session.session_id, user_id)
assert reloaded is not None
return reloaded
async def _parked(user_id: str) -> tuple[ChatSession, _Run]:
chat = await _chat(user_id)
run = await _Run(user_id, chat.session_id).create()
paused, _ = await run.step()
assert paused
return chat, run
async def _answer(client: httpx.AsyncClient, node_exec_id: str, approved: bool):
response = await client.post(
"/api/review/action",
json={"reviews": [{"node_exec_id": node_exec_id, "approved": approved}]},
)
assert response.status_code == 200, response.text
async def test_a_step_parks_on_its_card_and_the_card_resumes_the_run(
setup_test_user, test_user_id, flags, client, resumed
):
chat, run = await _parked(test_user_id)
row, card = await run.row(), await run.card()
assert row is not None and row.status == ReviewStatus.WAITING
assert card is not None and card.status == ReviewStatus.WAITING
assert card.session_id == chat.session_id
assert card.payload["workflow"]["node_exec_id"] == run.node_exec_id
assert card.payload["tool_call_id"] == ""
node = await AgentNodeExecution.prisma().find_unique(where={"id": run.node_exec_id})
assert node is not None and node.executionStatus == AgentExecutionStatus.REVIEW
# One decision, one item on Home: the card.
home = {r.node_exec_id for r in await get_pending_reviews_for_user(test_user_id)}
assert run.card_id in home and run.node_exec_id not in home
await _answer(client, run.card_id, approved=True)
row = await run.row()
assert row is not None and row.status == ReviewStatus.APPROVED
assert await run.card() is None
assert resumed.await_args.kwargs["graph_exec_id"] == run.graph_exec_id
# The card finished through the run's resume, not as a held call.
assert await held._held(chat.session_id) == {}
assert await run.step() == (False, _INPUT)
async def test_a_card_declined_in_chat_rejects_the_step_and_asks_from_now_on(
setup_test_user, test_user_id, flags, client, resumed
):
chat, run = await _parked(test_user_id)
await _answer(client, run.card_id, approved=False)
row = await run.row()
assert row is not None and row.status == ReviewStatus.REJECTED
resumed.assert_awaited_once()
rule = await chat_rules.rule_for(
chat.session_id, f"block:{_BLOCK.id}", test_user_id, None
)
assert rule is not None and rule.rule == "ask"
with pytest.raises(BlockExecutionError):
await run.step()
async def test_a_page_of_home_stays_full_while_steps_wait_on_their_cards(server, flags):
"""A client that stops at a short page would never see the later cards."""
user_id = str(uuid.uuid4())
await User.prisma().create(data={"id": user_id, "email": f"{user_id}@example.com"})
chat = await _chat(user_id)
run = await _Run(user_id, chat.session_id).create()
cards = []
for _ in range(3):
assert (await run.step())[0]
cards.append(run.card_id)
await run.next_step()
pages = [
[
review.node_exec_id
for review in await get_pending_reviews_for_user(user_id, page, 2)
]
for page in (1, 2, 3)
]
assert [len(page) for page in pages] == [2, 1, 0]
assert sorted(sum(pages, [])) == sorted(cards)
async def test_a_step_answered_on_the_run_page_closes_its_card(
setup_test_user, test_user_id, flags, client, resumed
):
_, run = await _parked(test_user_id)
await _answer(client, run.node_exec_id, approved=True)
assert await run.card() is None
resumed.assert_awaited_once()
async def test_a_stopped_run_closes_its_steps_cards(
setup_test_user, test_user_id, flags
):
_, run = await _parked(test_user_id)
await cancel_pending_reviews_for_execution(run.graph_exec_id, test_user_id)
assert await run.card() is None
async def test_an_allow_rule_runs_the_step_and_records_why(
setup_test_user, test_user_id, flags
):
chat = await _chat(test_user_id)
run = await _Run(test_user_id, chat.session_id).create()
await chat_rules.set_rule(chat.session_id, f"block:{_BLOCK.id}", "allow")
assert await run.step() == (False, _INPUT)
row = await run.row()
assert row is not None and row.status == ReviewStatus.APPROVED
assert row.review_message == (
"Allowed by Otto: you set Send Discord Message to run without asking."
)
assert await run.card() is None
async def test_a_cheap_outward_step_reaches_the_supervisor_as_its_direct_call_does(
setup_test_user, test_user_id, flags
):
"""Auto judges a cheap outward block by the lighter rubric, naming the block."""
chat = await _chat(test_user_id)
await update_session_autopilot_mode(chat.session_id, test_user_id, "auto")
chat = await get_chat_session(chat.session_id, test_user_id)
assert chat is not None
run = await _Run(test_user_id, chat.session_id).create()
args = {"id": _BLOCK.id, "input": _INPUT}
async def subject():
return await RunCapabilityTool().gate_subject(test_user_id, chat, args)
supervise = AsyncMock(return_value=Judgement(allowed=True, reason="ok"))
with (
patch("backend.copilot.gate.workflow.supervise", supervise),
patch("backend.copilot.gate.supervise", supervise),
):
step = await run.step()
direct = await check_action(
"run_capability", args, test_user_id, chat, subject_of=subject
)
assert step == (False, _INPUT) and direct.allowed
for call in supervise.await_args_list:
assert call.kwargs.get("outward_judged") is True
assert call.kwargs["args"].get(RUNS_KEY) == "Send Discord Message"
assert supervise.await_count == 2
@pytest.mark.parametrize("failure", ["no stamp", "chat gone", "card fails"])
async def test_a_step_the_gate_cannot_answer_is_an_alerted_human_review(
setup_test_user, test_user_id, flags, failure
):
chat = await _chat(test_user_id)
run = await _Run(
test_user_id, str(uuid.uuid4()) if failure == "chat gone" else chat.session_id
).create()
spy = AsyncMock(wraps=HITLReviewHelper.get_or_create_human_review)
open_row = AsyncMock(return_value=failure != "card fails")
with (
patch.object(HITLReviewHelper, "get_or_create_human_review", spy),
patch("backend.copilot.gate.review.open_review_row", open_row),
):
paused, _ = await run.step(stamped=failure != "no stamp")
assert paused
row = await run.row()
assert row is not None and row.status == ReviewStatus.WAITING
assert await run.card() is None
# The last write of the row is the one that alerts.
assert spy.await_args.kwargs.get("alert", True) is True
@pytest.mark.parametrize("runs_it", ["allow rule", "unsupervised", "supervisor allow"])
async def test_a_paywalled_owners_step_parks_however_the_gate_would_run_it(
setup_test_user, test_user_id, flags, paywalled_user, runs_it
):
"""Without the gate this step parks and its resume meets the paywall, so the
gate must not run it instead."""
chat = await _chat(test_user_id)
if runs_it == "unsupervised":
await update_session_autopilot_mode(
chat.session_id, test_user_id, "unsupervised"
)
else:
rule = "allow" if runs_it == "allow rule" else "judge"
await chat_rules.set_rule(chat.session_id, _KEY, rule)
run = await _Run(test_user_id, chat.session_id).create()
supervise = AsyncMock(return_value=Judgement(allowed=True, reason="ok"))
spy = AsyncMock(wraps=HITLReviewHelper.get_or_create_human_review)
with (
patch("backend.copilot.gate.workflow.supervise", supervise),
patch.object(HITLReviewHelper, "get_or_create_human_review", spy),
):
paused, _ = await run.step()
assert paused
row = await run.row()
assert row is not None and row.status == ReviewStatus.WAITING
assert await run.card() is None
assert spy.await_args.kwargs.get("alert", True) is True
supervise.assert_not_awaited()
async def test_a_card_opened_before_the_flag_went_off_still_resumes_its_run(
setup_test_user, test_user_id, flags, client, resumed
):
chat, run = await _parked(test_user_id)
flags[Flag.COPILOT_UNIFIED_APPROVALS] = False
await _answer(client, run.card_id, approved=True)
resumed.assert_awaited_once()
later = await _Run(test_user_id, chat.session_id).create()
paused, _ = await later.step()
assert paused and await later.card() is None
@pytest_asyncio.fixture(loop_scope="session")
async def wider_rules_cleared(test_user_id):
"""A wider rule outlives its test and would let every later step run."""
keys = [
chat_rules._scoped_key(scope, test_user_id, None, _KEY)
for scope in ("expert", "team")
]
redis = await get_redis_async()
await redis.delete(*keys)
yield
await redis.delete(*keys)
@pytest.mark.parametrize("scope", ["chat", "expert", "team"])
@pytest.mark.parametrize("rule", ["allow", "judge"])
async def test_a_rule_from_a_step_card_covers_every_later_call_of_its_block(
setup_test_user,
test_user_id,
flags,
client,
resumed,
wider_rules_cleared,
rule,
scope,
):
"""Answered like an action card: the rule lands on the block itself, so a
later step of the run, a later run and a direct call of the block follow it."""
chat, run = await _parked(test_user_id)
response = await client.post(
"/api/review/action",
json={
"reviews": [
{
"node_exec_id": run.card_id,
"approved": True,
"chat_rule": rule,
"chat_rule_scope": scope,
}
]
},
)
assert response.status_code == 200, response.text
later_chat = chat if scope == "chat" else await _chat(test_user_id)
args = {"id": _BLOCK.id, "input": _INPUT}
async def subject():
return await RunCapabilityTool().gate_subject(test_user_id, later_chat, args)
supervise = AsyncMock(return_value=Judgement(allowed=True, reason="ok"))
with (
patch("backend.copilot.gate.workflow.supervise", supervise),
patch("backend.copilot.gate.supervise", supervise),
):
same_run = await (await run.next_step()).step()
later = await _Run(test_user_id, later_chat.session_id).create()
later_run = await later.step()
direct = await check_action(
"run_capability", args, test_user_id, later_chat, subject_of=subject
)
assert same_run == (False, _INPUT)
assert later_run == (False, _INPUT)
assert direct.allowed and direct.decided
assert supervise.await_count == (3 if rule == "judge" else 0)
if scope == "chat":
elsewhere = await _Run(
test_user_id, (await _chat(test_user_id)).session_id
).create()
assert (await elsewhere.step())[0]
async def test_a_card_answered_before_its_run_pauses_waits_until_it_has(
setup_test_user, test_user_id, flags, client, resumed
):
"""While other steps still run, the run cannot be resumed: an answer the
step took then would let the executor finish the pass with nothing waiting
and mark the run complete, the step never run."""
chat = await _chat(test_user_id)
run = await _Run(test_user_id, chat.session_id).create(
status=AgentExecutionStatus.RUNNING
)
paused, _ = await run.step()
assert paused
early = await client.post(
"/api/review/action",
json={"reviews": [{"node_exec_id": run.card_id, "approved": True}]},
)
assert early.status_code == 409, early.text
row, card = await run.row(), await run.card()
assert row is not None and row.status == ReviewStatus.WAITING
assert card is not None and card.status == ReviewStatus.WAITING
# The executor ends its pass in REVIEW only while a row still waits.
assert await has_pending_reviews_for_graph_exec(run.graph_exec_id)
resumed.assert_not_awaited()
await AgentGraphExecution.prisma().update(
where={"id": run.graph_exec_id},
data={"executionStatus": AgentExecutionStatus.REVIEW},
)
await _answer(client, run.card_id, approved=True)
row = await run.row()
assert row is not None and row.status == ReviewStatus.APPROVED
assert resumed.await_args.kwargs["graph_exec_id"] == run.graph_exec_id
async def test_an_answer_the_step_could_not_take_keeps_its_card_for_a_retry(
setup_test_user, test_user_id, flags, client, resumed
):
_, run = await _parked(test_user_id)
with patch(
"backend.data.human_review.process_all_reviews_for_execution",
AsyncMock(side_effect=OSError("database unavailable")),
):
failed = await client.post(
"/api/review/action",
json={"reviews": [{"node_exec_id": run.card_id, "approved": True}]},
)
assert failed.status_code == 503, failed.text
# Back in the chat's queue, and still the one item Home shows.
card = await run.card()
assert card is not None and card.status == ReviewStatus.WAITING
home = {r.node_exec_id for r in await get_pending_reviews_for_user(test_user_id)}
assert run.card_id in home and run.node_exec_id not in home
row = await run.row()
assert row is not None and row.status == ReviewStatus.WAITING
resumed.assert_not_awaited()
await _answer(client, run.card_id, approved=True)
row = await run.row()
assert row is not None and row.status == ReviewStatus.APPROVED
assert await run.card() is None
resumed.assert_awaited_once()
@pytest.mark.parametrize(
"ended", [AgentExecutionStatus.FAILED, AgentExecutionStatus.TERMINATED]
)
async def test_a_card_whose_run_has_ended_can_still_be_answered_and_goes(
setup_test_user, test_user_id, flags, client, resumed, ended
):
_, run = await _parked(test_user_id)
await AgentGraphExecution.prisma().update(
where={"id": run.graph_exec_id}, data={"executionStatus": ended}
)
await _answer(client, run.card_id, approved=False)
assert await run.card() is None
resumed.assert_not_awaited()
async def test_a_card_that_could_not_be_reopened_leaves_the_step_to_the_run_page(
setup_test_user, test_user_id, flags, client, resumed
):
"""With both writes failing, the step's own row is the way back: Home lists
it again, and answering it there closes the card."""
_, run = await _parked(test_user_id)
with (
patch(
"backend.data.human_review.process_all_reviews_for_execution",
AsyncMock(side_effect=OSError("database unavailable")),
),
patch(
"backend.data.human_review.reopen_review",
AsyncMock(side_effect=OSError("database unavailable")),
),
):
failed = await client.post(
"/api/review/action",
json={"reviews": [{"node_exec_id": run.card_id, "approved": True}]},
)
assert failed.status_code == 503, failed.text
home = {r.node_exec_id for r in await get_pending_reviews_for_user(test_user_id)}
assert run.node_exec_id in home and run.card_id not in home
await _answer(client, run.node_exec_id, approved=True)
assert await run.card() is None
resumed.assert_awaited_once()
autogpt_platform/backend/backend/blocks/_base.py
+2
View File
@@ -882,6 +882,8 @@ class Block(ABC, Generic[BlockSchemaInputType, BlockSchemaOutputType]):
chat_session_id=(
None if is_graph_execution else execution_context.session_id
),
block=self,
execution_context=execution_context,
)
if decision is None:
autogpt_platform/backend/backend/blocks/helpers/review.py
+116 -13
View File
@@ -4,7 +4,7 @@ Used by both the dedicated HumanInTheLoopBlock and blocks that require human rev
"""
import logging
from typing import Any, Optional
from typing import TYPE_CHECKING, Any, Literal, Optional
from prisma.enums import ReviewStatus
from pydantic import BaseModel
@@ -12,6 +12,10 @@ from pydantic import BaseModel
from backend.data.human_review import ReviewResult
from backend.util.clients import get_database_manager_async_client
if TYPE_CHECKING:
from backend.blocks._base import Block
from backend.data.execution import ExecutionContext
logger = logging.getLogger(__name__)
@@ -47,6 +51,11 @@ class HITLReviewHelper:
db_client=get_database_manager_async_client(), **kwargs
)
@staticmethod
async def record_gate_approval(**kwargs) -> Optional[ReviewResult]:
"""Record the chat gate's yes on a workflow step."""
return await get_database_manager_async_client().record_gate_approval(**kwargs)
@staticmethod
async def update_review_processed_status(
node_exec_id: str, processed: bool
@@ -71,6 +80,8 @@ class HITLReviewHelper:
organization_id: Optional[str] = None,
team_id: Optional[str] = None,
chat_session_id: Optional[str] = None,
block: "Block | None" = None,
execution_context: "ExecutionContext | None" = None,
) -> Optional[ReviewResult]:
"""
Handle a review request for a block that requires human review.
@@ -86,6 +97,8 @@ class HITLReviewHelper:
chat_session_id: The chat asking, when not a graph execution
block_name: Name of the block requesting review
editable: Whether the reviewer can edit the data
block, execution_context: an irreversible block's own pause, which
the approval gate of the chat that started the run may answer
Returns:
ReviewResult if review is complete, None if waiting for human input
@@ -134,19 +147,32 @@ class HITLReviewHelper:
node_exec_id=node_exec_id,
)
result = await HITLReviewHelper.get_or_create_human_review(
user_id=user_id,
node_exec_id=node_exec_id,
graph_exec_id=graph_exec_id,
graph_id=graph_id,
graph_version=graph_version,
chat_session_id=chat_session_id,
input_data=input_data,
message=block_name, # Use block_name directly as the message
editable=editable,
organization_id=organization_id,
team_id=team_id,
review = {
"user_id": user_id,
"node_exec_id": node_exec_id,
"graph_exec_id": graph_exec_id,
"graph_id": graph_id,
"graph_version": graph_version,
"chat_session_id": chat_session_id,
"input_data": input_data,
"message": block_name, # Use block_name directly as the message
"editable": editable,
"organization_id": organization_id,
"team_id": team_id,
}
gated = (
await HITLReviewHelper._gate_review(block, execution_context, review)
if block is not None
and execution_context is not None
and is_graph_execution
else None
)
if isinstance(gated, ReviewResult):
result: Optional[ReviewResult] = gated
elif gated == "parked":
result = None
else:
result = await HITLReviewHelper.get_or_create_human_review(**review)
if result is None:
logger.info(
@@ -167,6 +193,79 @@ class HITLReviewHelper:
return result
@staticmethod
async def _gate_review(
block: "Block", ctx: "ExecutionContext", review: dict[str, Any]
) -> ReviewResult | Literal["parked"] | None:
"""The chat's approval gate answers the pause: a decision to act on,
``"parked"`` once its card waits in the chat, or None for a human review.
"""
# Deferred: the gate imports the block registry, which imports this module.
from backend.copilot.gate import workflow
session = await workflow.review_route(ctx, review["user_id"])
step = (
await workflow.workflow_step(
review["graph_exec_id"],
review["graph_id"],
review["graph_version"],
review["node_exec_id"],
)
if session is not None
else None
)
if session is None or step is None:
return None
try:
ruling = await workflow.judge_step(
session, block, review["input_data"], ctx, step
)
if ruling.run:
return (
await HITLReviewHelper.record_gate_approval(
user_id=review["user_id"],
node_exec_id=step.node_exec_id,
graph_exec_id=step.graph_exec_id,
graph_id=step.graph_id,
graph_version=step.graph_version,
input_data=review["input_data"],
block_name=review["message"],
message=ruling.reason,
organization_id=review["organization_id"],
team_id=review["team_id"],
)
or "parked"
)
waiting = await HITLReviewHelper.get_or_create_human_review(
**review, alert=False
)
except Exception:
logger.warning(
f"Gate could not answer step {review['node_exec_id']}; "
"it waits for a human review",
exc_info=True,
)
return None
if waiting is not None:
return waiting
try:
opened = await workflow.open_step_card(
session, block, review["input_data"], step, ruling
)
except Exception:
logger.warning(
f"Gate could not open the card of step {step.node_exec_id}",
exc_info=True,
)
opened = False
if not opened:
# No card, so the row is an ordinary human review and alerts as one.
try:
await HITLReviewHelper.get_or_create_human_review(**review)
except Exception:
logger.warning(f"Could not alert review {step.node_exec_id}")
return "parked"
@staticmethod
async def handle_review_decision(
input_data: Any,
@@ -182,6 +281,8 @@ class HITLReviewHelper:
organization_id: Optional[str] = None,
team_id: Optional[str] = None,
chat_session_id: Optional[str] = None,
block: "Block | None" = None,
execution_context: "ExecutionContext | None" = None,
) -> Optional[ReviewDecision]:
"""
Handle a review request and return the decision in a single call.
@@ -216,6 +317,8 @@ class HITLReviewHelper:
organization_id=organization_id,
team_id=team_id,
chat_session_id=chat_session_id,
block=block,
execution_context=execution_context,
)
if review_result is None:
autogpt_platform/backend/backend/blocks/helpers/review_test.py
+216
View File
@@ -1,11 +1,16 @@
"""Unit tests for the HITL review helper chain."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from backend.blocks.discord.bot_blocks import SendDiscordMessageBlock
from backend.blocks.helpers.review import HITLReviewHelper
from backend.copilot.gate import Ruling
from backend.data.execution import ExecutionContext, ExecutionStatus
from backend.data.human_review import ReviewResult, ReviewStatus
from backend.util.exceptions import BlockExecutionError
@pytest.mark.asyncio
@@ -44,3 +49,214 @@ async def test_handle_review_decision_threads_org_to_review_row():
kwargs = get_or_create.await_args.kwargs
assert kwargs["organization_id"] == "org-exec"
assert kwargs["team_id"] == "team-exec"
# ---- the chat's approval gate answers an irreversible step's pause --------
_WORKFLOW = "backend.copilot.gate.workflow"
_STEP = SimpleNamespace(
graph_exec_id="ge-1",
node_exec_id="ne-1",
graph_id="g-1",
graph_version=1,
name="Morning digest",
)
_INPUT = {"channel_name": "general", "message_content": "hi"}
def _approved(message: str = "ok") -> ReviewResult:
return ReviewResult(
data=_INPUT,
status=ReviewStatus.APPROVED,
message=message,
processed=True,
node_exec_id="ne-1",
)
def _pause(**helpers):
"""The block's own pause, with the review helper's DB calls stubbed."""
stubs = {
"check_approval": AsyncMock(return_value=None),
"get_or_create_human_review": AsyncMock(return_value=None),
"record_gate_approval": AsyncMock(return_value=_approved()),
"update_node_execution_status": AsyncMock(),
"update_review_processed_status": AsyncMock(),
**helpers,
}
patches = [patch.object(HITLReviewHelper, k, v) for k, v in stubs.items()]
return stubs, patches
async def _run(
stubs,
patches,
ruling=None,
routed=True,
card_opens=True,
judge_error=None,
card_error=None,
):
ctx = ExecutionContext(
user_id="u1",
graph_id="g-1",
graph_exec_id="ge-1",
graph_version=1,
sensitive_action_safe_mode=True,
approval_chat_id="chat-1",
)
judge = AsyncMock(
return_value=ruling or Ruling(run=False, reason="asks"),
side_effect=judge_error,
)
open_card = AsyncMock(return_value=card_opens, side_effect=card_error)
chat = SimpleNamespace(session_id="chat-1") if routed else None
with (
patch(f"{_WORKFLOW}.review_route", AsyncMock(return_value=chat)),
patch(f"{_WORKFLOW}.workflow_step", AsyncMock(return_value=_STEP)),
patch(f"{_WORKFLOW}.judge_step", judge),
patch(f"{_WORKFLOW}.open_step_card", open_card),
):
for p in patches:
p.start()
try:
result = await SendDiscordMessageBlock().is_block_exec_need_review(
_INPUT,
user_id="u1",
node_id="n-1",
node_exec_id="ne-1",
graph_exec_id="ge-1",
graph_id="g-1",
graph_version=1,
execution_context=ctx,
)
finally:
for p in patches:
p.stop()
return result, judge, open_card
async def test_a_step_the_gate_does_not_answer_is_a_human_review():
stubs, patches = _pause()
(paused, _), judge, _ = await _run(stubs, patches, routed=False)
assert paused
judge.assert_not_awaited()
assert "alert" not in stubs["get_or_create_human_review"].await_args.kwargs
async def test_a_gate_yes_records_the_approval_and_the_step_runs():
stubs, patches = _pause()
ruling = Ruling(run=True, reason="Allowed by Otto: x.", by="mode")
(paused, data), _, open_card = await _run(stubs, patches, ruling=ruling)
assert not paused and data == _INPUT
assert stubs["record_gate_approval"].await_args.kwargs["message"] == ruling.reason
stubs["get_or_create_human_review"].assert_not_awaited()
open_card.assert_not_awaited()
async def test_a_gate_yes_cannot_overturn_a_decline_already_on_the_row():
declined = _approved().model_copy(update={"status": ReviewStatus.REJECTED})
stubs, patches = _pause(record_gate_approval=AsyncMock(return_value=declined))
with pytest.raises(BlockExecutionError):
await _run(stubs, patches, ruling=Ruling(run=True, by="mode"))
async def test_a_gate_ask_parks_the_step_silently_behind_its_card():
stubs, patches = _pause()
(paused, _), _, open_card = await _run(stubs, patches)
assert paused
open_card.assert_awaited_once()
assert stubs["get_or_create_human_review"].await_count == 1
assert stubs["get_or_create_human_review"].await_args.kwargs["alert"] is False
status = stubs["update_node_execution_status"].await_args.kwargs["status"]
assert status == ExecutionStatus.REVIEW
@pytest.mark.parametrize("card_error", [None, OSError])
async def test_a_card_that_cannot_open_leaves_an_alerted_human_review(card_error):
stubs, patches = _pause()
(paused, _), _, _ = await _run(
stubs, patches, card_opens=False, card_error=card_error
)
assert paused
calls = stubs["get_or_create_human_review"].await_args_list
assert [c.kwargs.get("alert", True) for c in calls] == [False, True]
async def test_a_gate_that_fails_leaves_a_human_review():
stubs, patches = _pause()
(paused, _), _, open_card = await _run(stubs, patches, judge_error=OSError)
assert paused
open_card.assert_not_awaited()
assert "alert" not in stubs["get_or_create_human_review"].await_args.kwargs
async def test_a_human_in_the_loop_pause_never_reaches_the_gate():
"""Only an irreversible block's own pause passes the block; a question
the workflow's author put to a person stays theirs."""
route = AsyncMock()
_, patches = _pause()
with patch(f"{_WORKFLOW}.review_route", route):
for p in patches:
p.start()
try:
await HITLReviewHelper.handle_review_decision(
input_data=_INPUT,
user_id="u1",
node_id="n-1",
node_exec_id="ne-1",
graph_exec_id="ge-1",
graph_id="g-1",
graph_version=1,
)
finally:
for p in patches:
p.stop()
route.assert_not_awaited()
async def test_a_sub_graphs_step_names_its_own_run_not_its_parents():
"""A sub-graph's context is copied from its parent run, so the step's own
ids come from the pause, never from the context."""
stubs, patches = _pause()
ctx = ExecutionContext(
user_id="u1",
graph_id="parent-graph",
graph_exec_id="parent-run",
graph_version=7,
sensitive_action_safe_mode=True,
approval_chat_id="chat-1",
)
judge = AsyncMock(return_value=Ruling(run=True, reason="ok", by="mode"))
graphs = SimpleNamespace(
get_graph_metadata=AsyncMock(return_value=SimpleNamespace(name="Child"))
)
with (
patch(
f"{_WORKFLOW}.review_route",
AsyncMock(return_value=SimpleNamespace(session_id="chat-1")),
),
patch(f"{_WORKFLOW}.judge_step", judge),
patch(f"{_WORKFLOW}.graph_db", return_value=graphs),
):
for p in patches:
p.start()
try:
await SendDiscordMessageBlock().is_block_exec_need_review(
_INPUT,
user_id="u1",
node_id="n-1",
node_exec_id="ne-1",
graph_exec_id="ge-1",
graph_id="g-1",
graph_version=1,
execution_context=ctx,
)
finally:
for p in patches:
p.stop()
step = judge.await_args.args[4]
assert (step.graph_exec_id, step.graph_id, step.graph_version) == ("ge-1", "g-1", 1)
graphs.get_graph_metadata.assert_awaited_once_with("g-1", 1)
recorded = stubs["record_gate_approval"].await_args.kwargs
assert (recorded["graph_exec_id"], recorded["graph_id"]) == ("ge-1", "g-1")
autogpt_platform/backend/backend/copilot/constants.py
+12
View File
@@ -1,5 +1,7 @@
"""Shared constants for the CoPilot module."""
import hashlib
# Otto is the account's built-in helper — the identity every session
# without an expert speaks as, and the fixed author of the morning briefing.
AUTOPILOT_NAME = "Otto"
@@ -65,6 +67,9 @@ SPEND_REVIEW_MARKER = "expert-spend:"
# Format: "{node_id}:{random_hex}" — extract node_id via rsplit(":", 1)[0]
COPILOT_NODE_EXEC_ID_SEPARATOR = ":"
# The approval gate's card for a workflow step, in the chat that started the run.
COPILOT_STEP_CARD_PREFIX = f"{COPILOT_NODE_PREFIX}gate-workflow_step"
# Compaction notice messages shown to users.
COMPACTION_DONE_MSG = "Earlier messages were summarized to fit within context limits."
COMPACTION_DROPPED_MSG = "Earlier messages were dropped to fit within context limits."
@@ -98,6 +103,13 @@ HUNG_TOOL_CAP_SECONDS = 2 * 60 * 60
STREAM_LOCK_PREFIX = "copilot:stream:lock:"
def step_card_id(user_id: str, node_exec_id: str) -> str:
"""Derived from the step alone, so the run page, Home and a stopped run
can find the card the gate opened for it."""
digest = hashlib.sha256(f"{user_id}\x00{node_exec_id}".encode()).hexdigest()[:16]
return f"{COPILOT_STEP_CARD_PREFIX}{COPILOT_NODE_EXEC_ID_SEPARATOR}{digest}"
def is_copilot_synthetic_id(id_value: str) -> bool:
"""Check if an ID is a CoPilot synthetic ID (not from a real graph execution)."""
return id_value.startswith(COPILOT_SYNTHETIC_ID_PREFIX)
autogpt_platform/backend/backend/copilot/gate/__init__.py
+166 -65
View File
@@ -14,7 +14,7 @@ a run into a question, never the reverse.
"""
import logging
from typing import Any, Awaitable, Callable
from typing import Any, Awaitable, Callable, Literal
from prisma.enums import ReviewStatus
from pydantic import BaseModel, ConfigDict
@@ -28,7 +28,7 @@ from backend.util.feature_flag import Flag, is_feature_enabled
from . import chat_rules, held
from . import review as review_store
from .classifier import DecidedBy, supervise
from .classifier import DecidedBy, Judgement, supervise
from .headline import Headline
from .policy import (
DEFAULT_MODE,
@@ -83,6 +83,9 @@ _RULE_VERDICTS = {
"ask": Verdict.ASK,
"unreadable": Verdict.ASK,
}
# Decision 1 (OPEN-3592): the mode's table is final, so Unsupervised runs an
# irreversible block without asking. False makes Unsupervised ask before one.
UNSUPERVISED_RUNS_IRREVERSIBLE_BLOCKS = True
class Decision(BaseModel):
@@ -97,9 +100,27 @@ class Decision(BaseModel):
headline: Headline | None = None
# The user approved this exact call on a card, so nothing downstream asks again.
approved: bool = False
# An active gate reached a verdict on this call: a card, a rule, the mode
# or the supervisor. With unified approvals on, nothing downstream asks again.
decided: bool = False
class Ruling(BaseModel):
"""The user's rule, the mode or the supervisor on one subject."""
model_config = ConfigDict(frozen=True)
run: bool
# Why it asks, or what let it run: the audit line on an approved step.
reason: str = ""
reason_kind: review_store.ReasonKind = "mode"
by: Literal["rule", "mode", "supervisor"] = "mode"
decided_by: DecidedBy | None = None
spend: dict[str, int] | None = None
ALLOW = Decision(allowed=True)
DECIDED = Decision(allowed=True, decided=True)
async def gate_active(user_id: str | None, session: ChatSession) -> bool:
@@ -112,6 +133,17 @@ async def gate_active(user_id: str | None, session: ChatSession) -> bool:
return await is_feature_enabled(Flag.COPILOT_AUTO_MODE, user_id, default=False)
async def unified_approvals(user_id: str) -> bool:
"""On by decision, so a read that fails means on."""
try:
return await is_feature_enabled(
Flag.COPILOT_UNIFIED_APPROVALS, user_id, default=True
)
except Exception:
logger.warning("Unified approvals flag unreadable; on", exc_info=True)
return True
def resolve_mode(session: ChatSession) -> AutopilotMode:
if session.metadata.source_platform in LINKED_CHAT_PLATFORMS:
return "auto"
@@ -165,7 +197,7 @@ async def check_action(
if await review_store.consume(review_id, user_id):
if review_store.is_spend_card(review):
await raise_ceiling(CEILING_UNIT_MICRODOLLARS)
return Decision(allowed=True, approved=True)
return Decision(allowed=True, approved=True, decided=True)
return Decision(allowed=False, reason=_CONSUMED)
if review is not None and review.status == ReviewStatus.REJECTED:
await review_store.consume(review_id, user_id)
@@ -186,69 +218,57 @@ async def check_action(
if effect is Effect.UNGATED:
return ALLOW
rule_key = subject.key if subject is not None else tool_name
mode = resolve_mode(session)
# Only a subject that can be parked can carry a rule, so reads and
# workspace work skip the Redis round trip.
hit = (
await chat_rules.rule_for(session_id, rule_key, user_id, session.expert_id)
if effect in PARKABLE
else None
)
rule = hit.rule if hit else None
# No supervisor reads a scheduled run, so a judge rule falls back to the mode.
if rule == "judge" and subject is not None and subject.schedules:
rule = None
# A judge rule covers irreversible subjects too: the user chose the supervisor.
verdict = _RULE_VERDICTS[rule] if rule else verdict_for_effect(mode, effect)
estimate = subject.estimate if subject is not None else estimate_for(tool_name)
spend = spend_shown = None
if effect in METERED and estimate > 0 and mode != "unsupervised":
unsupervised = resolve_mode(session) == "unsupervised"
async def over_ceiling() -> Ruling | None:
if effect not in METERED or estimate <= 0 or unsupervised:
return None
spend = await spent_past_ceiling(user_id)
reason_kind: review_store.ReasonKind
decided_by: DecidedBy | None = None
if hit and rule in ("ask", "unreadable"):
reason, reason_kind = hit.reason, "rule"
elif spend is not None:
spend_shown = _spend_shown(estimate, *spend)
reason = (
f"costs about {_dollars(estimate)}, and this chat has spent "
f"{_dollars(spend[0])} of its {_dollars(spend[1])} ceiling; approving "
f"adds {_dollars(CEILING_UNIT_MICRODOLLARS)} to it"
if spend is None:
return None
return Ruling(
run=False,
reason=(
f"costs about {_dollars(estimate)}, and this chat has spent "
f"{_dollars(spend[0])} of its {_dollars(spend[1])} ceiling; "
f"approving adds {_dollars(CEILING_UNIT_MICRODOLLARS)} to it"
),
reason_kind="spend",
spend=_spend_shown(estimate, *spend),
)
reason_kind = "spend"
elif verdict is Verdict.RUN:
return ALLOW
elif verdict is Verdict.ASK and subject is not None and subject.reason:
reason, reason_kind = subject.reason, "subject"
elif verdict is Verdict.ASK:
reason = _ASK_FIRST if mode == "ask_first" else _OUTWARD
reason_kind = "mode"
else:
reason_kind = "supervisor"
async def judge() -> Judgement:
files = await context_of() if context_of is not None else None
if unread := [path for path, text in (files or {}).items() if text is None]:
reason = (
f"Could not read {', '.join(unread)}, so this could not be checked."
return Judgement(
allowed=False,
reason=(
f"Could not read {', '.join(unread)}, so this could not be "
"checked."
),
)
else:
outward_judged = effect is Effect.EXTERNAL_JUDGED
# A workflow's name is its author's free text; only a block's is vouched for.
runs = (
subject.name
if outward_judged
and subject is not None
and subject.key.startswith("block:")
else None
)
judgement = await supervise(
tool_name=tool_name,
args=_judged_args(args, files, runs),
user_messages=_typed_messages(session),
outward_judged=outward_judged,
)
if judgement.allowed:
return ALLOW
reason, decided_by = judgement.reason, judgement.decided_by
outward_judged = effect is Effect.EXTERNAL_JUDGED
# A workflow's name is its author's free text; only a block's is vouched for.
runs = (
subject.name
if outward_judged
and subject is not None
and subject.key.startswith("block:")
else None
)
return await supervise(
tool_name=tool_name,
args=_judged_args(args, files, runs),
user_messages=_typed_messages(session),
outward_judged=outward_judged,
)
ruling = await rule_on(
session, user_id, rule_key, effect, subject, judge, over_ceiling
)
if ruling.run:
return DECIDED
call = held.HeldCall(
review_id=review_id,
tool_name=tool_name,
@@ -259,7 +279,84 @@ async def check_action(
cost_context=await capture_cost_context(user_id),
)
return await _park(
call, user_id, session, reason, reason_kind, subject, decided_by, spend_shown
call,
user_id,
session,
ruling.reason,
ruling.reason_kind,
subject,
ruling.decided_by,
ruling.spend,
)
async def rule_on(
session: ChatSession,
user_id: str,
rule_key: str,
effect: Effect,
subject: Subject | None,
judge: Callable[[], Awaitable[Judgement]],
over_ceiling: Callable[[], Awaitable[Ruling | None]] | None = None,
*,
floor: bool = False,
) -> Ruling:
"""The user's rule on the subject in this chat, else the mode's verdict for
its effect, the supervisor deciding where either says judge. ``floor``
turns the mode's run into an ask; a rule still decides."""
mode = resolve_mode(session)
# Only a subject that can be parked can carry a rule, so reads and
# workspace work skip the Redis round trip.
hit = (
await chat_rules.rule_for(
session.session_id, rule_key, user_id, session.expert_id
)
if effect in PARKABLE
else None
)
rule = hit.rule if hit else None
if hit and rule in ("ask", "unreadable"):
return Ruling(run=False, reason=hit.reason, reason_kind="rule", by="rule")
if over_ceiling is not None and (held_for_spend := await over_ceiling()):
return held_for_spend
# No supervisor reads a scheduled run, so a judge rule falls back to the mode.
if rule == "judge" and subject is not None and subject.schedules:
rule = None
# A judge rule covers irreversible subjects too: the user chose the supervisor.
verdict = _RULE_VERDICTS[rule] if rule else verdict_for_effect(mode, effect)
if rule is None and verdict is Verdict.RUN and _mode_asks(mode, subject, floor):
verdict = Verdict.ASK
by = "rule" if rule else "mode"
if verdict is Verdict.RUN:
return Ruling(run=True, by=by)
if verdict is Verdict.ASK and subject is not None and subject.reason:
return Ruling(run=False, reason=subject.reason, reason_kind="subject", by=by)
if verdict is Verdict.ASK:
reason = _ASK_FIRST if mode == "ask_first" else _OUTWARD
return Ruling(run=False, reason=reason, by=by)
judgement = await judge()
# Only an ask carries the supervisor's words onto a card.
if judgement.allowed:
return Ruling(run=True, by="supervisor")
return Ruling(
run=False,
reason=judgement.reason,
reason_kind="supervisor",
by="supervisor",
decided_by=judgement.decided_by,
)
def _mode_asks(mode: AutopilotMode, subject: Subject | None, floor: bool) -> bool:
irreversible_block = (
subject is not None
and subject.irreversible
and subject.key.startswith("block:")
)
return floor or (
mode == "unsupervised"
and irreversible_block
and not UNSUPERVISED_RUNS_IRREVERSIBLE_BLOCKS
)
@@ -342,15 +439,16 @@ def _judged_args(
return judged
def _typed_messages(session: ChatSession) -> list[str]:
"""What the user typed, oldest first; an answered card's rows and a row of
nothing but the server's context are not the user's words."""
def _typed_messages(session: ChatSession, upto: int | None = None) -> list[str]:
"""What the user typed, oldest first, among the first ``upto`` messages when
given; an answered card's rows and a row of nothing but the server's context
are not the user's words."""
# Deferred: copilot.service imports the tool registry, which imports this gate.
from backend.copilot.service import strip_injected_context_for_display
typed = [
strip_injected_context_for_display(message.content)
for message in session.messages
for message in session.messages[:upto]
if message.role == "user"
and message.content
and not held.is_answer_row(message)
@@ -362,6 +460,7 @@ __all__ = [
"RUNS_KEY",
"RUN_FILES_KEY",
"Decision",
"Ruling",
"LINKED_CHAT_PLATFORMS",
"active_mode",
"check_action",
@@ -369,4 +468,6 @@ __all__ = [
"gate_active",
"refusal_message",
"resolve_mode",
"rule_on",
"unified_approvals",
]
autogpt_platform/backend/backend/copilot/gate/card_fixture_test.py
+50 -1
View File
@@ -15,9 +15,17 @@ from backend.blocks.code_executor import ExecuteCodeStepBlock
from backend.blocks.google.gmail import GmailSendBlock
from backend.blocks.google.sheets import GoogleSheetsUpdateRowBlock
from backend.blocks.http import SendWebRequestBlock
from backend.copilot.constants import step_card_id
from backend.copilot.gate import Ruling
from backend.copilot.gate.policy import Effect
from backend.copilot.gate.review import payload_headline, review_id_for, review_payload
from backend.copilot.gate.review import (
WorkflowStep,
payload_headline,
review_id_for,
review_payload,
)
from backend.copilot.gate.subject import block_subject, workflow_subject
from backend.copilot.gate.workflow import step_card_payload, step_subject
from backend.data.graph import GraphModel, NodeModel
FIXTURE = (
@@ -126,6 +134,7 @@ def build_cards() -> list[dict[str, Any]]:
cards = [_block_card(*entry) for entry in _BLOCKS]
cards.append(_workflow_card())
cards.append(_sandbox_command_card())
cards.append(_workflow_step_card())
return json.loads(json.dumps(cards, default=str))
@@ -190,6 +199,46 @@ def _workflow_card() -> dict[str, Any]:
}
def _workflow_step_card() -> dict[str, Any]:
"""A Gmail Send step of a workflow Otto started, parked on its own review."""
send = GmailSendBlock()
inputs = dict(_BLOCKS[0][2])
step = WorkflowStep(
graph_exec_id="run-digest",
node_exec_id="node-exec-send",
graph_id="g-digest",
graph_version=1,
name="Morning digest",
)
subject = step_subject(send, inputs, step.name)
payload = step_card_payload(
send,
inputs,
step,
Ruling(run=False, reason=subject.reason, reason_kind="subject"),
mode="auto",
turn=1,
)
review_id = step_card_id("user-1", step.node_exec_id)
return {
"story": "Workflow Step",
"review": {
"node_exec_id": review_id,
"node_id": review_id.split(":")[0],
"user_id": "user-1",
"session_id": "s1",
"graph_exec_id": None,
"graph_id": None,
"graph_version": None,
"payload": payload,
"instructions": payload_headline(payload),
"editable": False,
"status": "WAITING",
},
"schema": send.input_schema.jsonschema(),
}
def _sandbox_command_card() -> dict[str, Any]:
args = {"command": _COMMAND, "timeout": 60}
row = _row("bash_exec", args, None, _SUPERVISOR_WRITE, "supervisor")
autogpt_platform/backend/backend/copilot/gate/reads_test.py
+9 -3
View File
@@ -17,7 +17,7 @@ import pytest
from prisma.enums import ReviewStatus
from backend.copilot.context import set_turn_unattended
from backend.copilot.gate import held, reads
from backend.copilot.gate import ALLOW, Decision, held, reads
from backend.copilot.gate import review as review_store
from backend.copilot.gate.content import ContentVerdict
from backend.copilot.gate.reads import read_review_id
@@ -868,7 +868,7 @@ async def test_a_blocks_own_output_is_still_judged(rows):
def _no_action_gate():
return patch.object(BaseTool, "_gate", AsyncMock(return_value=(None, False)))
return patch.object(BaseTool, "_gate", AsyncMock(return_value=(None, ALLOW)))
# What a producer declares came from outside AutoGPT is all the judge reads.
@@ -984,7 +984,13 @@ async def test_a_blocks_output_is_judged_without_the_platforms_message(rows):
with (
patch(f"{_READS}.judge_content", judge),
# Approved, so the block runs without its own review pause.
patch.object(BaseTool, "_gate", AsyncMock(return_value=(None, True))),
patch.object(
BaseTool,
"_gate",
AsyncMock(
return_value=(None, Decision(allowed=True, approved=True, decided=True))
),
),
_resolves_to(_capability("block", _STORE_VALUE)),
patch("backend.copilot.tools.helpers.workspace_db", lambda: workspace),
patch("backend.copilot.tools.helpers.user_db", lambda: users),
autogpt_platform/backend/backend/copilot/gate/review.py
+13
View File
@@ -64,6 +64,16 @@ class FieldLabel(BaseModel):
format: str | None = None
class WorkflowStep(BaseModel):
"""The workflow step a card answers: its row is the step's own review."""
graph_exec_id: str
node_exec_id: str
graph_id: str
graph_version: int
name: str
class GateReviewPayload(BaseModel):
"""What the approval card, Home and the channels render: never the raw call."""
@@ -88,6 +98,7 @@ class GateReviewPayload(BaseModel):
headline: Headline
# A money card's estimate, spend so far and ceiling.
spend: dict[str, int] | None = None
workflow: WorkflowStep | None = None
# An approval must not run a call long after the user gave it; the answered
@@ -131,6 +142,7 @@ def review_payload(
tool_call_id: str = "",
turn: int = 0,
references: list[Reference] | None = None,
workflow: WorkflowStep | None = None,
) -> dict[str, Any]:
"""The subject is kept as decided when the card opened: what the user saw,
not a recomputation over a tree that may have moved since."""
@@ -164,6 +176,7 @@ def review_payload(
else headline_for(tool_name, args, references)
),
spend=spend,
workflow=workflow,
).model_dump()
autogpt_platform/backend/backend/copilot/gate/workflow.py
+371
View File
@@ -0,0 +1,371 @@
"""The gate answers the review pause of a workflow step AutoPilot started.
The step parks as a human review always does; the gate then judges it as it
would a direct call of the same block in the chat that started the run. A run
verdict approves the row at once; an ask opens the gate's card in that chat,
pointing at the row, and answering either closes the other. Every failure
leaves an ordinary human review.
"""
import logging
from collections.abc import Iterable
from typing import Any
from prisma.enums import ReviewStatus
from pydantic import BaseModel
from backend.api.features.graph_executions.review.model import PendingHumanReviewModel
from backend.blocks._base import Block
from backend.copilot.constants import (
AUTOPILOT_NAME,
COPILOT_STEP_CARD_PREFIX,
step_card_id,
)
from backend.copilot.model import (
ChatSession,
get_chat_session,
get_chat_session_metadata,
)
from backend.copilot.rate_limit import is_user_paywalled
from backend.data.db_accessors import graph_db, review_db
from backend.data.execution import ExecutionContext
from . import Ruling, _judged_args, _typed_messages, chat_rules, gate_active
from . import review as review_store
from . import rule_on, unified_approvals
from .classifier import Judgement, supervise
from .policy import DEFAULT_MODE, Effect
from .subject import Subject, block_subject
logger = logging.getLogger(__name__)
STEP_TOOL = "workflow_step"
# Decision 2 (OPEN-3592): a workflow's own "ask before irreversible actions"
# is a floor, so the mode cannot run its steps; a rule on the block still can.
WORKFLOW_SETTING_IS_A_FLOOR = True
APPROVED_IN_CHAT = "Approved in chat"
DECLINED_IN_CHAT = "Declined in chat"
async def review_route(ctx: ExecutionContext, user_id: str) -> ChatSession | None:
"""The chat whose gate answers this pause, or None for a human review."""
if not ctx.approval_chat_id:
return None
try:
if not await unified_approvals(user_id):
return None
# Without a plan the step must wait for a resume, which the paywall answers.
if await is_user_paywalled(user_id):
return None
session = await get_chat_session(ctx.approval_chat_id, user_id)
if session is None or not await gate_active(user_id, session):
return None
return session
except Exception:
logger.warning(
f"Gate could not route the review of run {ctx.graph_exec_id}",
exc_info=True,
)
return None
async def judge_step(
session: ChatSession,
block: Block,
inputs: dict[str, Any],
ctx: ExecutionContext,
step: review_store.WorkflowStep,
) -> Ruling:
"""What the chat's gate says about this step, as about a direct call of its
block: the user's rule on the block, else the mode, else the supervisor."""
subject = step_subject(block, inputs, step.name)
async def judge() -> Judgement:
# Judged as a direct call of the block would be: same rubric, vouched name.
outward_judged = subject.effect is Effect.EXTERNAL_JUDGED
args = {"workflow": step.name, "step": subject.name, "input": inputs}
return await supervise(
tool_name=STEP_TOOL,
args=_judged_args(args, None, subject.name if outward_judged else None),
user_messages=request_that_started(
session, ctx.root_execution_id or step.graph_exec_id
),
outward_judged=outward_judged,
)
ruling = await rule_on(
session,
session.user_id,
subject.key,
subject.effect,
subject,
judge,
# A context from before the setting was carried counts it as on.
floor=WORKFLOW_SETTING_IS_A_FLOOR
and ctx.graph_sensitive_action_safe_mode is not False,
)
if ruling.run:
return ruling.model_copy(update={"reason": _allowed_by(ruling, subject)})
return ruling
async def open_step_card(
session: ChatSession,
block: Block,
inputs: dict[str, Any],
step: review_store.WorkflowStep,
ruling: Ruling,
) -> bool:
"""The gate's card in the chat that started the run, pointing at the
step's row; False when it could not be opened."""
user_id = session.user_id
try:
payload = step_card_payload(
block,
inputs,
step,
ruling,
mode=session.metadata.autopilot_mode or DEFAULT_MODE,
turn=review_store.turn_of(session),
)
except Exception:
logger.warning(
f"Gate could not build the card for step {step.node_exec_id}",
exc_info=True,
)
return False
return await review_store.open_review_row(
step_card_id(user_id, step.node_exec_id),
user_id,
session,
payload,
review_store.payload_headline(payload),
)
def step_card_payload(
block: Block,
inputs: dict[str, Any],
step: review_store.WorkflowStep,
ruling: Ruling,
*,
mode: str,
turn: int,
) -> dict[str, Any]:
"""The card a direct call of the block would show, naming its workflow."""
return review_store.review_payload(
STEP_TOOL,
{"input": inputs},
step_subject(block, inputs, step.name),
reason=ruling.reason,
reason_kind=ruling.reason_kind,
decided_by=ruling.decided_by,
mode=mode,
turn=turn,
workflow=step,
)
class StepAnswers(BaseModel):
# The steps whose rows took their card's answer, for the route to resume.
answered: list[review_store.WorkflowStep] = []
# Cards whose step could not take the answer; each stays for a retry.
failed: list[str] = []
async def finish_step_cards(
user_id: str, cards: Iterable[PendingHumanReviewModel]
) -> StepAnswers:
"""Answer each answered workflow card's row the way the card was answered.
A card is consumed once its row has the answer, or already had another."""
finished = StepAnswers()
for card in cards:
step = step_of(card)
if step is None or card.status == ReviewStatus.WAITING:
continue
approved = card.status == ReviewStatus.APPROVED
try:
# No reviewed data: the step runs on the input its card showed.
await review_db().process_all_reviews_for_execution(
user_id=user_id,
review_decisions={
step.node_exec_id: (
card.status,
None,
APPROVED_IN_CHAT if approved else DECLINED_IN_CHAT,
)
},
)
finished.answered.append(step)
except ValueError:
# Answered differently on the run page first: that answer stands.
logger.warning(
f"Step {step.node_exec_id} kept its answer from the run page",
exc_info=True,
)
except Exception:
logger.warning(
f"Step {step.node_exec_id} could not take its card's answer",
exc_info=True,
)
finished.failed.append(card.node_exec_id)
await _reopen(user_id, card.node_exec_id)
continue
await review_store.consume(card.node_exec_id, user_id)
if not approved and card.session_id:
await _ask_from_now_on(user_id, card)
return finished
async def close_step_cards(user_id: str, node_exec_ids: Iterable[str]) -> None:
"""The steps were answered on the run page, so their cards leave the chat."""
for node_exec_id in node_exec_ids:
await review_store.consume(step_card_id(user_id, node_exec_id), user_id)
async def card_chat_for_run(user_id: str, graph_exec_id: str) -> str | None:
"""The chat holding a card for one of this run's waiting steps, if any."""
try:
rows = await review_db().get_pending_reviews_for_execution(
graph_exec_id, user_id
)
cards = await review_db().get_reviews_by_node_exec_ids(
[step_card_id(user_id, row.node_exec_id) for row in rows], user_id
)
except Exception:
logger.warning(f"Could not read step cards of run {graph_exec_id}")
return None
return next(
(
card.session_id
for card in cards.values()
if card.status == ReviewStatus.WAITING and card.session_id
),
None,
)
def awaiting_whom(card_chat: str | None, session_id: str | None) -> str:
"""How a parked run's state reads to the model that asks about it."""
if card_chat is None:
return "is awaiting human review"
if card_chat == session_id:
return "is waiting for the user's approval in this chat"
return "is waiting for the user's approval in the chat that started it"
async def workflow_step(
graph_exec_id: str | None,
graph_id: str | None,
graph_version: int | None,
node_exec_id: str,
) -> review_store.WorkflowStep | None:
"""What the card says about the step, from the pause's own ids: a
sub-graph's context still names its parent's run. None outside a run."""
if not (graph_exec_id and graph_id and graph_version is not None):
return None
try:
graph = await graph_db().get_graph_metadata(graph_id, graph_version)
name = graph.name if graph and graph.name else ""
except Exception:
logger.warning(f"Could not name workflow {graph_id}", exc_info=True)
name = ""
return review_store.WorkflowStep(
graph_exec_id=graph_exec_id,
node_exec_id=node_exec_id,
graph_id=graph_id,
graph_version=graph_version,
name=name or "Untitled workflow",
)
def step_subject(block: Block, inputs: dict[str, Any], workflow: str) -> Subject:
"""The block's own subject, so a rule set from either kind of card covers
both, with the card's reason naming the workflow."""
subject = block_subject(block, inputs, priced=False)
return subject.model_copy(
update={"reason": f"A step of {workflow}, which reaches outside the platform."}
)
def step_of(card: PendingHumanReviewModel) -> review_store.WorkflowStep | None:
if not is_step_card(card):
return None
payload = card.payload if isinstance(card.payload, dict) else {}
try:
return review_store.WorkflowStep.model_validate(payload.get("workflow"))
except Exception:
logger.warning(f"Workflow card {card.node_exec_id} names no step")
return None
def is_step_card(card: PendingHumanReviewModel) -> bool:
return card.node_exec_id.startswith(COPILOT_STEP_CARD_PREFIX)
def step_rule_keys(cards: Iterable[PendingHumanReviewModel]) -> dict[str, str]:
"""The subject each workflow card can set a rule on, from the card the
server wrote: a step has no held call to read it from."""
keys: dict[str, str] = {}
for card in cards:
payload = card.payload if isinstance(card.payload, dict) else {}
subject = payload.get("subject")
if is_step_card(card) and isinstance(subject, dict) and subject.get("key"):
keys[card.node_exec_id] = str(subject["key"])
return keys
def request_that_started(session: ChatSession, run_id: str | None) -> list[str]:
"""What the user typed up to the call that started the run: a run parked
for days must not be judged against whatever they said since."""
started = next(
(
index
for index, message in enumerate(session.messages)
if run_id
and message.role == "tool"
and message.content
and run_id in message.content
),
None,
)
if started is None:
return []
return _typed_messages(session, upto=started)
async def _reopen(user_id: str, card_id: str) -> None:
# Back in the chat's queue, so the user can answer it again.
try:
await review_db().reopen_review(card_id, user_id)
except Exception:
logger.warning(f"Could not reopen card {card_id}", exc_info=True)
async def _ask_from_now_on(user_id: str, card: PendingHumanReviewModel) -> None:
# As a declined direct call does: the block asks for the rest of the chat.
key = step_rule_keys([card]).get(card.node_exec_id)
if not key or not card.session_id:
return
try:
chat = await get_chat_session_metadata(card.session_id, user_id)
await chat_rules.set_ask(
card.session_id, key, user_id, chat.expert_id if chat else None
)
except Exception:
# The step already has its answer; a lost rule must not strand its run.
logger.warning(
f"Could not set an ask rule from card {card.node_exec_id}",
exc_info=True,
)
def _allowed_by(ruling: Ruling, subject: Subject) -> str:
"""The audit line on the step's approved row, read on the run page."""
if ruling.by == "rule":
why = f"you set {subject.name} to run without asking"
elif ruling.by == "supervisor":
why = "its check found this step matches your request"
else:
why = "Unsupervised is on in the chat that started this run"
return f"Allowed by {AUTOPILOT_NAME}: {why}."
autogpt_platform/backend/backend/copilot/gate/workflow_test.py
+409
View File
@@ -0,0 +1,409 @@
"""The gate's answer to a workflow step's own review pause.
A step is judged as a direct call of its block in the chat that started the
run, so these pin that the two share their subject, their rules and their
modes, and that every missing piece leaves the step to a human review.
"""
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from prisma.enums import ReviewStatus
from backend.api.features.graph_executions.review.model import PendingHumanReviewModel
from backend.blocks.discord.bot_blocks import SendDiscordMessageBlock
from backend.blocks.email_block import SendEmailBlock
from backend.copilot.constants import COPILOT_STEP_CARD_PREFIX, step_card_id
from backend.copilot.gate import RUNS_KEY, chat_rules
from backend.copilot.gate import review as review_store
from backend.copilot.gate import unified_approvals, workflow
from backend.copilot.gate.chat_rules import DECLINED
from backend.copilot.gate.chat_rules_test import _Redis
from backend.copilot.gate.classifier import Judgement
from backend.copilot.model import (
AutopilotMode,
ChatMessage,
ChatSession,
ChatSessionMetadata,
)
from backend.copilot.tools.run_capability import RunCapabilityTool
from backend.data.execution import ExecutionContext
from backend.util.feature_flag import Flag
_GATE = "backend.copilot.gate"
_WORKFLOW = "backend.copilot.gate.workflow"
_BLOCK = SendDiscordMessageBlock()
_INPUTS = {"channel_name": "general", "message_content": "Digest is out"}
_STEP = review_store.WorkflowStep(
graph_exec_id="run-1",
node_exec_id="node-exec-1",
graph_id="graph-1",
graph_version=1,
name="Morning digest",
)
_FALLBACK = "Could not verify this action automatically."
def _session(mode: AutopilotMode = "auto", messages=()) -> ChatSession:
return ChatSession(
session_id="chat-1",
user_id="user-1",
usage=[],
started_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
metadata=ChatSessionMetadata(origin="interactive", autopilot_mode=mode),
messages=list(messages),
)
def _ctx(**update) -> ExecutionContext:
return ExecutionContext(
user_id="user-1",
graph_id="graph-1",
graph_exec_id="run-1",
graph_version=1,
sensitive_action_safe_mode=True,
approval_chat_id="chat-1",
graph_sensitive_action_safe_mode=False,
).model_copy(update=update)
def _flags(unified: bool = True, auto_mode: bool = True):
values = {
Flag.COPILOT_UNIFIED_APPROVALS: unified,
Flag.COPILOT_AUTO_MODE: auto_mode,
}
async def read(flag, user_id, default=False):
return values[flag]
return patch(f"{_GATE}.is_feature_enabled", AsyncMock(side_effect=read))
@pytest.fixture
def redis():
store = _Redis()
with patch.object(chat_rules, "get_redis_async", AsyncMock(return_value=store)):
yield store
@pytest.fixture(autouse=True)
def paywalled():
with patch(
f"{_WORKFLOW}.is_user_paywalled", AsyncMock(return_value=False)
) as check:
yield check
# ---- which pauses go to the gate -----------------------------------------
@pytest.mark.parametrize(
"missing", ["nothing", "stamp", "flag", "chat", "gate", "plan"]
)
async def test_a_pause_goes_to_the_gate_only_with_every_piece(missing, paywalled):
paywalled.return_value = missing == "plan"
session = _session()
ctx = _ctx(approval_chat_id=None) if missing == "stamp" else _ctx()
with (
_flags(unified=missing != "flag", auto_mode=missing != "gate"),
patch(
f"{_WORKFLOW}.get_chat_session",
AsyncMock(return_value=None if missing == "chat" else session),
),
):
routed = await workflow.review_route(ctx, "user-1")
assert routed is (session if missing == "nothing" else None)
@pytest.mark.parametrize("failing", ["chat", "plan"])
async def test_a_lookup_that_fails_leaves_a_human_review(failing, paywalled):
if failing == "plan":
paywalled.side_effect = OSError
load = AsyncMock(side_effect=OSError if failing == "chat" else None)
load.return_value = _session()
with _flags(), patch(f"{_WORKFLOW}.get_chat_session", load):
assert await workflow.review_route(_ctx(), "user-1") is None
async def test_a_flag_nobody_created_is_on():
"""PostHog answers an unknown flag with the default, so the default decides."""
async def unknown(flag, user_id, default=False):
return default
with patch(f"{_GATE}.is_feature_enabled", AsyncMock(side_effect=unknown)):
assert await unified_approvals("user-1")
async def test_a_flag_read_that_fails_is_on():
with patch(f"{_GATE}.is_feature_enabled", AsyncMock(side_effect=OSError)):
assert await unified_approvals("user-1")
# ---- what the gate says about a step -------------------------------------
@pytest.mark.parametrize(
"mode, verdict", [("ask_first", "ask"), ("auto", "judge"), ("unsupervised", "run")]
)
async def test_a_step_follows_the_modes_table_for_a_cheap_outward_block(
redis, mode, verdict
):
supervise = AsyncMock(return_value=Judgement(allowed=False, reason=_FALLBACK))
with patch(f"{_WORKFLOW}.supervise", supervise):
ruling = await workflow.judge_step(
_session(mode), _BLOCK, _INPUTS, _ctx(), _STEP
)
assert ruling.run is (verdict == "run")
assert supervise.await_count == (verdict == "judge")
if verdict == "ask":
assert ruling.reason == (
"A step of Morning digest, which reaches outside the platform."
)
async def test_auto_asks_before_a_step_of_a_block_hard_to_take_back(redis):
block = SendEmailBlock()
supervise = AsyncMock()
with patch(f"{_WORKFLOW}.supervise", supervise):
ruling = await workflow.judge_step(
_session("auto"), block, {"to_email": "[email protected]"}, _ctx(), _STEP
)
assert not ruling.run
supervise.assert_not_awaited()
async def test_an_allow_rule_runs_the_step_and_says_so_on_the_run_page(redis):
await chat_rules.set_rule("chat-1", f"block:{_BLOCK.id}", "allow")
ruling = await workflow.judge_step(
_session("ask_first"), _BLOCK, _INPUTS, _ctx(), _STEP
)
assert ruling.run
assert ruling.reason == (
"Allowed by Otto: you set Send Discord Message to run without asking."
)
async def test_a_declined_block_asks_even_in_unsupervised(redis):
await chat_rules.set_ask("chat-1", f"block:{_BLOCK.id}", "user-1", None)
ruling = await workflow.judge_step(
_session("unsupervised"), _BLOCK, _INPUTS, _ctx(), _STEP
)
assert not ruling.run
assert (ruling.reason, ruling.reason_kind) == (DECLINED, "rule")
@pytest.mark.parametrize(
"judgement, runs, reason",
[
(Judgement(allowed=True, reason="Allowed.", decided_by="llm"), True, None),
(Judgement(allowed=False, reason=_FALLBACK), False, _FALLBACK),
],
ids=["supervisor allows", "supervisor down"],
)
async def test_a_judge_rule_hands_the_step_to_the_supervisor(
redis, judgement, runs, reason
):
await chat_rules.set_rule("chat-1", f"block:{_BLOCK.id}", "judge")
supervise = AsyncMock(return_value=judgement)
with patch(f"{_WORKFLOW}.supervise", supervise):
ruling = await workflow.judge_step(
_session("ask_first"), _BLOCK, _INPUTS, _ctx(), _STEP
)
assert ruling.run is runs
assert supervise.await_args.kwargs["tool_name"] == "workflow_step"
assert supervise.await_args.kwargs["outward_judged"] is True
assert supervise.await_args.kwargs["args"] == {
"workflow": "Morning digest",
"step": "Send Discord Message",
"input": _INPUTS,
RUNS_KEY: "Send Discord Message",
}
if reason:
assert (ruling.reason, ruling.reason_kind) == (reason, "supervisor")
@pytest.mark.parametrize(
"own_setting, rule, runs",
[
(True, None, False),
# A context from before the setting was carried counts it as on.
(None, None, False),
(False, None, True),
(True, "allow", True),
],
)
async def test_the_workflows_own_pause_is_a_floor_the_mode_cannot_lift(
redis, own_setting, rule, runs
):
if rule:
await chat_rules.set_rule("chat-1", f"block:{_BLOCK.id}", rule)
ctx = _ctx(graph_sensitive_action_safe_mode=own_setting)
ruling = await workflow.judge_step(
_session("unsupervised"), _BLOCK, _INPUTS, ctx, _STEP
)
assert ruling.run is runs
async def test_a_step_and_a_direct_call_of_its_block_share_one_rule_key():
session = _session()
direct = await RunCapabilityTool().gate_subject(
"user-1", session, {"id": _BLOCK.id, "input": _INPUTS}
)
step = workflow.step_subject(_BLOCK, _INPUTS, "Morning digest")
assert direct is not None and direct.key == step.key == f"block:{_BLOCK.id}"
assert step.irreversible
async def test_the_supervisor_reads_the_request_that_started_the_run(redis):
await chat_rules.set_rule("chat-1", f"block:{_BLOCK.id}", "judge")
session = _session(
"ask_first",
[
ChatMessage(role="user", content="Post the digest to #general"),
ChatMessage(role="assistant", content="Starting it."),
ChatMessage(role="tool", content='{"execution_id": "run-1"}'),
ChatMessage(role="user", content="Also delete every channel"),
],
)
supervise = AsyncMock(return_value=Judgement(allowed=True, reason="ok"))
with patch(f"{_WORKFLOW}.supervise", supervise):
await workflow.judge_step(session, _BLOCK, _INPUTS, _ctx(), _STEP)
assert supervise.await_args.kwargs["user_messages"] == [
"Post the digest to #general"
]
# ---- the card and its answer ---------------------------------------------
def test_a_step_card_is_a_gate_card_the_run_page_can_find():
card_id = step_card_id("user-1", "node-exec-1")
assert card_id.startswith(review_store.node_id_for(workflow.STEP_TOOL))
assert COPILOT_STEP_CARD_PREFIX == review_store.node_id_for(workflow.STEP_TOOL)
assert card_id == step_card_id("user-1", "node-exec-1")
assert card_id != step_card_id("user-2", "node-exec-1")
def _card(status: ReviewStatus) -> PendingHumanReviewModel:
payload = workflow.step_card_payload(
_BLOCK,
_INPUTS,
_STEP,
workflow.Ruling(run=False, reason="r", reason_kind="subject"),
mode="auto",
turn=1,
)
return PendingHumanReviewModel(
node_exec_id=step_card_id("user-1", _STEP.node_exec_id),
node_id=COPILOT_STEP_CARD_PREFIX,
user_id="user-1",
session_id="chat-1",
payload=payload,
instructions="Run Send Discord Message",
editable=False,
status=status,
created_at=datetime.now(UTC),
)
@pytest.fixture
def rows():
db = SimpleNamespace(
process_all_reviews_for_execution=AsyncMock(return_value={}),
delete_review_by_node_exec_id=AsyncMock(return_value=1),
reopen_review=AsyncMock(return_value=1),
)
with (
patch(f"{_WORKFLOW}.review_db", return_value=db),
patch.object(review_store, "review_db", return_value=db),
):
yield db
@pytest.mark.parametrize(
"status, message",
[
(ReviewStatus.APPROVED, "Approved in chat"),
(ReviewStatus.REJECTED, "Declined in chat"),
],
)
async def test_an_answered_card_answers_its_steps_row_and_is_consumed(
rows, redis, status, message
):
card = _card(status)
with patch(
f"{_WORKFLOW}.get_chat_session_metadata",
AsyncMock(return_value=SimpleNamespace(expert_id=None)),
):
finished = await workflow.finish_step_cards("user-1", [card])
assert finished.answered == [_STEP] and finished.failed == []
rows.process_all_reviews_for_execution.assert_awaited_once_with(
user_id="user-1",
review_decisions={_STEP.node_exec_id: (status, None, message)},
)
rows.delete_review_by_node_exec_id.assert_awaited_once_with(
card.node_exec_id, "user-1"
)
asked = await chat_rules.rule_for("chat-1", f"block:{_BLOCK.id}", "user-1", None)
assert (asked is not None) is (status == ReviewStatus.REJECTED)
async def test_a_step_answered_on_the_run_page_first_keeps_that_answer(rows, redis):
rows.process_all_reviews_for_execution.side_effect = ValueError("already done")
card = _card(ReviewStatus.APPROVED)
finished = await workflow.finish_step_cards("user-1", [card])
assert finished.answered == [] and finished.failed == []
rows.delete_review_by_node_exec_id.assert_awaited_once()
async def test_a_step_update_that_fails_keeps_the_card_and_says_so(rows, redis):
rows.process_all_reviews_for_execution.side_effect = OSError("unavailable")
card = _card(ReviewStatus.REJECTED)
finished = await workflow.finish_step_cards("user-1", [card])
rows.delete_review_by_node_exec_id.assert_not_awaited()
rows.reopen_review.assert_awaited_once_with(card.node_exec_id, "user-1")
assert finished.answered == [] and finished.failed == [card.node_exec_id]
# Nothing was declined yet, so the block keeps no new rule.
assert (
await chat_rules.rule_for("chat-1", f"block:{_BLOCK.id}", "user-1", None)
is None
)
async def test_a_decline_whose_rule_cannot_be_set_still_answers_its_step(rows, redis):
card = _card(ReviewStatus.REJECTED)
with patch(
f"{_WORKFLOW}.get_chat_session_metadata", AsyncMock(side_effect=OSError)
):
finished = await workflow.finish_step_cards("user-1", [card])
assert finished.answered == [_STEP]
async def test_a_held_call_card_is_no_step(rows):
card = _card(ReviewStatus.APPROVED).model_copy(
update={"node_exec_id": "copilot-node-gate-run_capability:abc"}
)
assert (await workflow.finish_step_cards("user-1", [card])).answered == []
rows.process_all_reviews_for_execution.assert_not_awaited()
assert workflow.step_rule_keys([card]) == {}
assert workflow.step_rule_keys([_card(ReviewStatus.APPROVED)]) == {
step_card_id("user-1", _STEP.node_exec_id): f"block:{_BLOCK.id}"
}
@pytest.mark.parametrize(
"card_chat, says",
[
(None, "is awaiting human review"),
("chat-1", "is waiting for the user's approval in this chat"),
("chat-2", "is waiting for the user's approval in the chat that started it"),
],
)
def test_a_parked_run_says_where_it_waits(card_chat, says):
assert workflow.awaiting_whom(card_chat, "chat-1") == says
autogpt_platform/backend/backend/copilot/tools/agent_output.py
+28 -4
View File
@@ -9,6 +9,7 @@ from pydantic import BaseModel, Field, field_validator
from backend.api.features.library.model import LibraryAgent
from backend.copilot.constants import MAX_TOOL_WAIT_SECONDS
from backend.copilot.gate.workflow import awaiting_whom, card_chat_for_run
from backend.copilot.model import ChatSession
from backend.data.db_accessors import execution_db, library_db
from backend.data.execution import (
@@ -334,8 +335,10 @@ class AgentOutputTool(BaseTool):
execution: GraphExecution | GraphExecutionWithNodes | None,
available_executions: list[GraphExecutionMeta],
session_id: str | None,
card_chat: str | None = None,
) -> AgentOutputResponse:
"""Build the response based on execution data."""
"""Build the response based on execution data. ``card_chat``: the chat
whose approval card a parked run waits on."""
library_agent_link = f"/library/agents/{agent.id}"
if not execution:
@@ -411,7 +414,8 @@ class AgentOutputTool(BaseTool):
message = f"Execution for workflow '{agent.name}' was terminated"
elif execution.status == ExecutionStatus.REVIEW:
message = (
f"Execution for workflow '{agent.name}' is awaiting human review. "
f"Execution for workflow '{agent.name}' "
f"{awaiting_whom(card_chat, session_id)}. "
"The user needs to approve it before it can continue."
)
elif execution.status in (
@@ -527,7 +531,13 @@ class AgentOutputTool(BaseTool):
],
).from_outside()
return self._build_response(agent, execution, [], session_id)
return self._build_response(
agent,
execution,
[],
session_id,
await _card_chat(user_id, execution),
)
# Resolve agent from identifiers
agent, error = await self._resolve_agent(
@@ -584,4 +594,18 @@ class AgentOutputTool(BaseTool):
timeout_seconds=wait_timeout,
)
return self._build_response(agent, execution, available_executions, session_id)
return self._build_response(
agent,
execution,
available_executions,
session_id,
await _card_chat(user_id, execution),
)
async def _card_chat(
user_id: str, execution: GraphExecution | GraphExecutionWithNodes | None
) -> str | None:
if execution is None or execution.status != ExecutionStatus.REVIEW:
return None
return await card_chat_for_run(user_id, execution.id)
autogpt_platform/backend/backend/copilot/tools/base.py
+16 -8
View File
@@ -26,6 +26,7 @@ from .models import (
)
if TYPE_CHECKING:
from backend.copilot.gate import Decision
from backend.copilot.gate.headline import Headline
from backend.copilot.gate.subject import Subject
@@ -314,6 +315,9 @@ async def _record_activity(
# Passed to ``_execute`` of a tool with a gate subject when the user approved
# this exact call; stripped from the model's own arguments so it cannot be forged.
GATE_APPROVED = "_gate_approved"
# Passed beside it when an active gate reached any verdict on the call: a card,
# a rule, the mode or the supervisor. Stripped the same way.
GATE_DECIDED = "_gate_decided"
class BaseTool:
@@ -439,6 +443,7 @@ class BaseTool:
**kwargs,
) -> StreamToolOutputAvailable:
kwargs.pop(GATE_APPROVED, None)
kwargs.pop(GATE_DECIDED, None)
if self.requires_auth and not user_id:
logger.warning(
"Attempted tool call for %s but user not authenticated",
@@ -501,12 +506,15 @@ class BaseTool:
# ``_execute_tool_sync`` — so there is one place to add, not two.
# Must stay AFTER the envelope and name gates: a call the envelope refuses can
# never run, so approving it would spend a user's decision on nothing.
gated, approved = await self._gate(user_id, session, tool_call_id, kwargs)
gated, decision = await self._gate(user_id, session, tool_call_id, kwargs)
if gated is not None:
return gated
run_kwargs = kwargs
if approved and self.has_gate_subject:
run_kwargs = {**kwargs, GATE_APPROVED: True}
if decision is not None and self.has_gate_subject:
if decision.approved:
run_kwargs = {**run_kwargs, GATE_APPROVED: True}
if decision.decided:
run_kwargs = {**run_kwargs, GATE_DECIDED: True}
# After the gates, so a refused call never looks to a turn-scoped gate
# like the tool having run, and before the await, because the gates ask
@@ -558,9 +566,9 @@ class BaseTool:
session: ChatSession,
tool_call_id: str,
kwargs: dict[str, Any],
) -> tuple[StreamToolOutputAvailable | None, bool]:
) -> tuple[StreamToolOutputAvailable | None, "Decision | None"]:
"""A refusal to return instead of running (or None to proceed), and
whether the user approved this exact call.
the gate's decision on a call that proceeds.
A gate that crashes must not become a gate that passes, so an
unexpected failure here refuses the call rather than falling through.
@@ -590,11 +598,11 @@ class BaseTool:
"settings, so nothing ran. Tell the user and stop.",
args=kwargs,
),
False,
None,
)
if decision.allowed:
return None, decision.approved
return None, decision
return (
self._refusal(
tool_call_id,
@@ -604,7 +612,7 @@ class BaseTool:
args=kwargs,
headline=decision.headline,
),
False,
None,
)
async def _released_read(
autogpt_platform/backend/backend/copilot/tools/expert_capabilities_test.py
+2 -1
View File
@@ -15,6 +15,7 @@ from backend.copilot.capabilities.dispatch import resolve_tool_dispatch
from backend.copilot.capabilities.ranking import ConnectionState
from backend.copilot.capabilities.registry import get_registry
from backend.copilot.capabilities.sources import expert_entries
from backend.copilot.gate import ALLOW
from backend.copilot.gate import review as review_store
from backend.copilot.gate.content import ContentVerdict
from backend.copilot.model import ChatSessionMetadata
@@ -201,7 +202,7 @@ async def test_an_expert_ids_validate_only_answer_is_not_held_by_the_judge(team)
with (
patch("backend.copilot.gate.is_feature_enabled", AsyncMock(return_value=True)),
patch.object(review_store, "find_review", AsyncMock(return_value=None)),
patch.object(BaseTool, "_gate", AsyncMock(return_value=(None, False))),
patch.object(BaseTool, "_gate", AsyncMock(return_value=(None, ALLOW))),
patch("backend.copilot.gate.reads.judge_content", judge),
):
result = await RunCapabilityTool().execute(
autogpt_platform/backend/backend/copilot/tools/gate_subject_test.py
+114
View File
@@ -8,6 +8,7 @@ import json
import os
import shlex
import subprocess
from contextlib import ExitStack
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any
@@ -33,6 +34,7 @@ from backend.blocks.search import GetWikipediaSummaryBlock
from backend.blocks.sql_query_block import SQLQueryBlock
from backend.copilot import usage_activation_fixtures
from backend.copilot.gate import RUN_FILES_KEY, RUNS_KEY, chat_rules
from backend.copilot.gate.chat_rules import RuleHit
from backend.copilot.gate.chat_rules_test import _Redis
from backend.copilot.gate.classifier import Judgement
from backend.copilot.gate.content import ContentVerdict
@@ -53,6 +55,7 @@ from backend.copilot.tools.run_agent import RunAgentTool
from backend.copilot.tools.run_capability import RunCapabilityTool
from backend.data.graph import BaseGraph, GraphModel, Link, Node, NodeModel
from backend.util import sandbox_login
from backend.util.feature_flag import Flag
from backend.util.sandbox_login import READ_CAP
from backend.util.sandbox_login_test import FakeRedis, FakeSandbox, stock_files
@@ -351,6 +354,117 @@ async def test_an_approved_irreversible_block_asks_once(gate):
gate.open_review.assert_not_awaited()
def _block_run(hitl_answer):
"""``run_block`` past its preparation: what its own review does, and the run."""
prep = MagicMock(
synthetic_node_id="node",
input_data={"channel_name": "general", "message_content": "hi"},
required_non_credential_keys=set(),
provided_input_keys=set(),
)
hitl = AsyncMock(return_value=hitl_answer or ("exec-id", prep.input_data))
execute = AsyncMock(return_value=MagicMock(type="block_output"))
patches = ExitStack()
for target, value in (
("prepare_block_for_execution", AsyncMock(return_value=prep)),
("check_spend_approval", AsyncMock(return_value=None)),
("check_hitl_review", hitl),
("execute_block", execute),
):
patches.enter_context(patch(f"backend.copilot.tools.run_block.{target}", value))
return prep, hitl, execute, patches
def _flags(unified: bool, auto_mode: bool = True):
values = {
Flag.COPILOT_UNIFIED_APPROVALS: unified,
Flag.COPILOT_AUTO_MODE: auto_mode,
}
async def read(flag, user_id, default=False):
return values[flag]
return patch(f"{_GATE}.is_feature_enabled", AsyncMock(side_effect=read))
@pytest.mark.parametrize(
"mode, rule, unified, auto_mode, asks_again",
[
("auto", "allow", True, True, False),
# Decision 1: Unsupervised's run is final for an irreversible block.
("unsupervised", None, True, True, False),
("auto", "allow", False, True, True),
("unsupervised", None, False, True, True),
# Nothing decided: an inactive gate's yes is no review.
("auto", None, True, False, True),
],
ids=[
"rule, unified on",
"unsupervised, unified on",
"rule, unified off",
"unsupervised, unified off",
"gate inactive",
],
)
async def test_a_gate_yes_is_the_irreversible_blocks_review(
gate, mode, rule, unified, auto_mode, asks_again
):
"""One layer: whatever the gate decided, the block's own review does not
ask after it, and with the flag off it does, as before."""
_, hitl, execute, patches = _block_run(None)
hit = RuleHit(rule=rule) if rule else None
with (
_flags(unified, auto_mode),
patch(f"{_GATE}.chat_rules.rule_for", AsyncMock(return_value=hit)),
patches,
):
await _run_capability(
_session(mode),
SendDiscordMessageBlock().id,
{"channel_name": "general", "message_content": "hi"},
)
assert hitl.await_count == int(asks_again)
execute.assert_awaited_once()
gate.open_review.assert_not_awaited()
async def test_a_card_approval_still_skips_the_blocks_review_with_unified_off(gate):
gate.find_review.return_value = SimpleNamespace(
status=ReviewStatus.APPROVED, payload={}
)
_, hitl, execute, patches = _block_run(None)
with _flags(unified=False), patches:
await _run_capability(
_session(),
SendDiscordMessageBlock().id,
{"channel_name": "general", "message_content": "hi"},
)
hitl.assert_not_awaited()
execute.assert_awaited_once()
async def test_the_model_cannot_forge_a_decision(gate):
_, hitl, _, patches = _block_run(None)
with _flags(unified=True, auto_mode=False), patches:
await _run_capability(
_session(),
SendDiscordMessageBlock().id,
{"channel_name": "general", "message_content": "hi"},
_gate_decided=True,
)
hitl.assert_awaited_once()
async def test_a_workflow_run_keeps_its_forced_pause_when_the_gate_decided_without_a_card():
"""Its steps' pauses go to the gate on their own; only a card lifts the pause."""
handle = AsyncMock(return_value=_answer())
with patch.object(RunAgentTool, "_handle_preset_run", handle):
await RunAgentTool()._execute(
"user-1", _session(), preset_id="preset-1", _gate_decided=True
)
assert handle.await_args.args[3] is False
async def test_the_model_cannot_forge_an_approval(gate, ran):
await _run_capability(
_session("unsupervised"),
autogpt_platform/backend/backend/copilot/tools/run_agent.py
+8 -2
View File
@@ -20,6 +20,7 @@ from backend.copilot.gate.subject import (
graph_cost_credits,
workflow_subject,
)
from backend.copilot.gate.workflow import awaiting_whom, card_chat_for_run
from backend.copilot.model import ChatSession
from backend.copilot.tool_display import emit_tool_display_name
from backend.copilot.tracking import track_chat_outcome
@@ -54,7 +55,7 @@ from backend.util.timezone_utils import (
validate_timezone,
)
from .base import GATE_APPROVED, BaseTool
from .base import GATE_APPROVED, GATE_DECIDED, BaseTool
from .execution_utils import (
NodeFailureSummary,
build_run_health_warning,
@@ -335,6 +336,9 @@ class RunAgentTool(BaseTool):
validators defined in the Pydantic model.
"""
approved = bool(kwargs.pop(GATE_APPROVED, False))
# Only a card approval lifts a run's forced pause; its steps' pauses go
# to the gate on their own.
kwargs.pop(GATE_DECIDED, None)
params = RunAgentInput(**kwargs)
# Session-level dry_run forces all runs to be dry. In normal sessions
# the LLM may still request dry_run=True on individual calls.
@@ -1230,9 +1234,11 @@ class RunAgentTool(BaseTool):
await _safe_link_to_chat_share(
session_id=session_id, execution_id=execution.id
)
card_chat = await card_chat_for_run(user_id, execution.id)
return ExecutionStartedResponse(
message=(
f"Workflow '{library_agent.name}' is awaiting human review. "
f"Workflow '{library_agent.name}' "
f"{awaiting_whom(card_chat, session_id)}. "
f"The user can approve or reject inline. After approval, "
f"the execution resumes automatically. Use tool:view_agent_output "
f"with execution_id='{execution.id}' to check the result."
autogpt_platform/backend/backend/copilot/tools/run_block.py
+6 -1
View File
@@ -6,6 +6,7 @@ from typing import Any
from backend.blocks.llm import LLM_PROVIDER_NAMES
from backend.copilot.constants import COPILOT_NODE_EXEC_ID_SEPARATOR
from backend.copilot.gate import unified_approvals
from backend.copilot.model import ChatSession
from backend.data.activity_event import ActivityEventDraft
@@ -114,6 +115,7 @@ class RunBlockTool(BaseTool):
input_data: dict | None = None,
validate_only: bool = False,
gate_approved: bool = False,
gate_decided: bool = False,
**kwargs, # dry_run is intentionally not accepted; read from session.dry_run
) -> ToolResponseBase:
"""Execute a block with the given input data.
@@ -125,6 +127,8 @@ class RunBlockTool(BaseTool):
input_data: Input values for the block
gate_approved: The user approved this exact call on a card, so the
irreversible-action pause would ask the same question twice.
gate_decided: The chat's approval gate reached a verdict on this
call; with unified approvals on, its yes is the block's review.
Returns:
BlockOutputResponse: Block execution outputs
@@ -264,7 +268,8 @@ class RunBlockTool(BaseTool):
if spend_gate is not None:
return spend_gate
if gate_approved:
# Named args from run_capability._run_block, never gate markers in kwargs.
if gate_approved or (gate_decided and await unified_approvals(user_id)):
synthetic_node_exec_id = (
f"{prep.synthetic_node_id}"
f"{COPILOT_NODE_EXEC_ID_SEPARATOR}"
autogpt_platform/backend/backend/copilot/tools/run_capability.py
+5 -2
View File
@@ -34,7 +34,7 @@ from backend.data.activity_event import ActivityEventDraft
from .ayrshare_managed import connect_for_block as ayrshare_connect_for_block
from .ayrshare_managed import is_ayrshare
from .base import GATE_APPROVED, BaseTool
from .base import GATE_APPROVED, GATE_DECIDED, BaseTool
from .capability_gates import gate_denied, gate_denied_error
from .describe_capability import (
MCP_RUN_PARAMETERS,
@@ -172,6 +172,7 @@ class RunCapabilityTool(BaseTool):
**kwargs,
) -> ToolResponseBase:
approved = bool(kwargs.pop(GATE_APPROVED, False))
decided = bool(kwargs.pop(GATE_DECIDED, False))
session_id = session.session_id
if not user_id:
return ErrorResponse(
@@ -206,7 +207,7 @@ class RunCapabilityTool(BaseTool):
).from_outside()
if entry.kind == "block":
return await _run_block(
entry, user_id, session, payload, validate_only, approved
entry, user_id, session, payload, validate_only, approved, decided
)
if entry.kind == "tool":
return await _describe_tool(entry, session)
@@ -239,6 +240,7 @@ async def _run_block(
payload: dict[str, Any],
validate_only: bool,
approved: bool,
decided: bool = False,
) -> ToolResponseBase:
if gate_denied(BLOCK_GATE):
return gate_denied_error("blocks", session.session_id)
@@ -308,6 +310,7 @@ async def _run_block(
input_data=payload,
validate_only=validate_only,
gate_approved=approved,
gate_decided=decided,
)
autogpt_platform/backend/backend/data/db_manager.py
+6
View File
@@ -155,6 +155,8 @@ from backend.data.human_review import (
get_reviews_by_node_exec_ids,
has_pending_reviews_for_graph_exec,
process_all_reviews_for_execution,
record_gate_approval,
reopen_review,
update_review_processed_status,
)
from backend.data.onboarding import increment_onboarding_runs
@@ -435,6 +437,8 @@ class DatabaseManager(AppService):
get_reviews_by_node_exec_ids = _(get_reviews_by_node_exec_ids)
has_pending_reviews_for_graph_exec = _(has_pending_reviews_for_graph_exec)
process_all_reviews_for_execution = _(process_all_reviews_for_execution)
record_gate_approval = _(record_gate_approval)
reopen_review = _(reopen_review)
update_review_processed_status = _(update_review_processed_status)
# ============ Library ============ #
@@ -944,6 +948,8 @@ class DatabaseManagerAsyncClient(AppServiceClient):
get_pending_reviews_for_user = d.get_pending_reviews_for_user
get_reviews_by_node_exec_ids = d.get_reviews_by_node_exec_ids
process_all_reviews_for_execution = d.process_all_reviews_for_execution
record_gate_approval = d.record_gate_approval
reopen_review = d.reopen_review
update_review_processed_status = d.update_review_processed_status
# ============ User Comms ============ #
autogpt_platform/backend/backend/data/execution.py
+7
View File
@@ -131,6 +131,13 @@ class ExecutionContext(BaseModel):
# spend and the executor can post run results into the expert's thread.
expert_id: Optional[str] = None
# The chat that started this run, whose approval gate answers its review
# pauses; None for any run no chat started.
approval_chat_id: Optional[str] = None
# The workflow's own sensitive_action_safe_mode, before a chat forced the
# pause on; None on a context that predates it.
graph_sensitive_action_safe_mode: Optional[bool] = None
# -------------------------- Models -------------------------- #
autogpt_platform/backend/backend/data/human_review.py
+126 -4
View File
@@ -16,8 +16,8 @@ from prisma.models import (
LibraryAgent,
PendingHumanReview,
)
from prisma.types import PendingHumanReviewUpdateInput
from pydantic import BaseModel
from prisma.types import PendingHumanReviewUpdateInput, PendingHumanReviewWhereInput
from pydantic import BaseModel, ValidationError
from backend.api.features.graph_executions.review.model import (
PendingHumanReviewModel,
@@ -25,9 +25,11 @@ from backend.api.features.graph_executions.review.model import (
)
from backend.copilot.constants import (
COPILOT_SESSION_PREFIX,
COPILOT_STEP_CARD_PREFIX,
is_copilot_synthetic_id,
legacy_chat_session_id,
parse_node_id_from_exec_id,
step_card_id,
)
from backend.data.execution import get_graph_execution_meta
from backend.notifications.review_alerts import sync_awaiting_review
@@ -214,6 +216,7 @@ async def get_or_create_human_review(
chat_session_id: str | None = None,
organization_id: str | None = None,
team_id: str | None = None,
alert: bool = True,
) -> Optional[ReviewResult]:
"""
Get existing review or create a new pending review entry.
@@ -228,6 +231,7 @@ async def get_or_create_human_review(
editable: Whether the data can be edited
graph_exec_id, graph_id, graph_version: the graph execution asking, or
chat_session_id: the chat asking — exactly one of the two
alert: False when the chat's approval card asks in its place
Returns:
ReviewResult if the review is complete, None if waiting for human input
@@ -276,7 +280,8 @@ async def get_or_create_human_review(
if review.status == ReviewStatus.WAITING:
# Nothing sends until a human acts, which is exactly the shape of an
# Alert. The engine debounces and coalesces from here.
await _sync_awaiting_review_safely(user_id, graph_id, chat_session_id)
if alert:
await _sync_awaiting_review_safely(user_id, graph_id, chat_session_id)
return None
else:
return ReviewResult(
@@ -288,6 +293,54 @@ async def get_or_create_human_review(
)
async def record_gate_approval(
user_id: str,
node_exec_id: str,
graph_exec_id: str,
graph_id: str,
graph_version: int,
input_data: SafeJsonData,
block_name: str,
message: str,
organization_id: str | None = None,
team_id: str | None = None,
) -> Optional[ReviewResult]:
"""The chat's approval gate let this workflow step run: an approved row is
its record on the run page, and a re-run of the step finds it.
A decision already on the row stands; None means it is still waiting.
"""
review = await PendingHumanReview.prisma().upsert(
where={"nodeExecId": node_exec_id},
data={
"create": {
"userId": user_id,
"nodeExecId": node_exec_id,
**_scope_columns(graph_exec_id, graph_id, graph_version, None),
"payload": SafeJson(input_data),
"instructions": block_name,
"editable": False,
"status": ReviewStatus.APPROVED,
"reviewMessage": message,
"processed": True,
"reviewedAt": datetime.now(timezone.utc),
**({"organizationId": organization_id} if organization_id else {}),
**({"teamId": team_id} if team_id else {}),
},
"update": {},
},
)
if review.status == ReviewStatus.WAITING:
return None
return ReviewResult(
data=review.payload,
status=review.status,
message=review.reviewMessage or "",
processed=review.processed,
node_exec_id=review.nodeExecId,
)
def _review_scope(
graph_exec_id: str | None, chat_session_id: str | None
) -> tuple[str | None, str | None]:
@@ -473,8 +526,15 @@ async def get_pending_reviews_for_user(
"""
offset = (page - 1) * page_size
where: PendingHumanReviewWhereInput = {
"userId": user_id,
"status": ReviewStatus.WAITING,
}
# Left out before paging, so a page is never short of its size.
if in_chat := await _steps_waiting_in_chat(user_id):
where["nodeExecId"] = {"not_in": in_chat}
reviews = await PendingHumanReview.prisma().find_many(
where={"userId": user_id, "status": ReviewStatus.WAITING},
where=where,
order={"createdAt": "desc"},
skip=offset,
take=page_size,
@@ -484,6 +544,38 @@ async def get_pending_reviews_for_user(
return await _enrich_pending_reviews(user_id, models)
async def _steps_waiting_in_chat(user_id: str) -> list[str]:
"""A workflow step whose card waits in the chat is one decision: Home shows
the card, not the step's row as well."""
cards = await PendingHumanReview.prisma().find_many(
where={
"userId": user_id,
"status": ReviewStatus.WAITING,
"nodeExecId": {"startswith": COPILOT_STEP_CARD_PREFIX},
}
)
return [step for card in cards if (step := _step_of(user_id, card))]
def _step_of(user_id: str, card: PendingHumanReview) -> str | None:
try:
step = _StepCard.model_validate(card.payload).workflow.node_exec_id
except ValidationError:
return None
# The id is derived from the step, so a payload naming another step is ignored.
return step if step_card_id(user_id, step) == card.nodeExecId else None
class _Step(BaseModel):
node_exec_id: str
class _StepCard(BaseModel):
"""The part of a step card's payload that names its step."""
workflow: _Step
async def _enrich_pending_reviews(
user_id: str, reviews: list[PendingHumanReviewModel]
) -> list[PendingHumanReviewModel]:
@@ -821,6 +913,13 @@ async def cancel_pending_reviews_for_execution(graph_exec_id: str, user_id: str)
f"Graph execution {graph_exec_id} not found or doesn't belong to user {user_id}"
)
waiting = await PendingHumanReview.prisma().find_many(
where={
"graphExecId": graph_exec_id,
"userId": user_id,
"status": ReviewStatus.WAITING,
}
)
result = await PendingHumanReview.prisma().update_many(
where={
"graphExecId": graph_exec_id,
@@ -834,10 +933,33 @@ async def cancel_pending_reviews_for_execution(graph_exec_id: str, user_id: str)
"reviewedAt": datetime.now(timezone.utc),
},
)
# The gate's cards for those steps would otherwise wait in the chat for nothing.
if waiting:
await PendingHumanReview.prisma().delete_many(
where={
"nodeExecId": {
"in": [step_card_id(user_id, r.nodeExecId) for r in waiting]
},
"userId": user_id,
}
)
await sync_awaiting_review(user_id, graph_exec.graph_id)
return result
async def reopen_review(node_exec_id: str, user_id: str) -> int:
"""Put an answered review back to waiting, for an answer that never took effect."""
return await PendingHumanReview.prisma().update_many(
where={"nodeExecId": node_exec_id, "userId": user_id},
data={
"status": ReviewStatus.WAITING,
"reviewMessage": None,
"reviewedAt": None,
"processed": False,
},
)
async def delete_review_by_node_exec_id(node_exec_id: str, user_id: str) -> int:
"""Delete a review record by node execution ID after it has been consumed.
autogpt_platform/backend/backend/data/human_review_test.py
+26 -2
View File
@@ -5,6 +5,7 @@ import pytest
import pytest_mock
from prisma.enums import ReviewStatus
from backend.copilot.constants import step_card_id
from backend.data.human_review import (
get_or_create_human_review,
get_pending_reviews_for_execution,
@@ -130,7 +131,10 @@ async def test_get_pending_reviews_for_user(
):
"""Test getting pending reviews for a user with pagination"""
mock_find_many = mocker.patch("backend.data.human_review.PendingHumanReview.prisma")
mock_find_many.return_value.find_many = AsyncMock(return_value=[sample_db_review])
# The chat cards answering steps (none), then the page.
mock_find_many.return_value.find_many = AsyncMock(
side_effect=[[], [sample_db_review]]
)
# Mock the batched node_id resolution (AgentNodeExecution.prisma().find_many)
mock_node_exec = Mock()
@@ -157,11 +161,31 @@ async def test_get_pending_reviews_for_user(
assert result[0].node_id == "test_node_def_789"
# Verify pagination parameters
call_args = mock_find_many.return_value.find_many.call_args
call_args = mock_find_many.return_value.find_many.call_args_list[1]
assert call_args.kwargs["skip"] == 10 # (page-1) * page_size = (2-1) * 10
assert call_args.kwargs["take"] == 10
@pytest.mark.parametrize("named", ["its own step", "another step"])
@pytest.mark.asyncio(loop_scope="function")
async def test_a_step_card_hides_only_the_step_it_was_opened_for(
mocker: pytest_mock.MockFixture, named
):
card = Mock()
card.nodeExecId = step_card_id("user-1", "step-1")
card.payload = {
"workflow": {"node_exec_id": "step-1" if named == "its own step" else "step-2"}
}
prisma = mocker.patch("backend.data.human_review.PendingHumanReview.prisma")
prisma.return_value.find_many = AsyncMock(side_effect=[[card], []])
await get_pending_reviews_for_user("user-1")
page = prisma.return_value.find_many.call_args_list[1].kwargs["where"]
hidden = {"not_in": ["step-1"]} if named == "its own step" else None
assert page.get("nodeExecId") == hidden
@pytest.mark.asyncio(loop_scope="function")
async def test_get_pending_reviews_for_execution(
mocker: pytest_mock.MockFixture,
autogpt_platform/backend/backend/executor/utils.py
+45 -15
View File
@@ -5,7 +5,7 @@ import threading
import time
from collections import defaultdict
from concurrent.futures import Future
from typing import Literal, Mapping, Optional, cast, get_args
from typing import Any, Literal, Mapping, Optional, cast, get_args
from prisma.enums import ReviewStatus
from pydantic import BaseModel, Field, JsonValue, ValidationError
@@ -1441,11 +1441,12 @@ async def _add_graph_execution(
if not graph_exec:
raise NotFoundError(f"Workflow run #{graph_exec_id} not found.")
approval_chat_id = await starting_chat(graph_exec, user_id, edb)
# A resume rebuilds its context from the graph settings, which would
# drop the pause the chat that started this run asked for.
pause_irreversible_actions = (
pause_irreversible_actions
or await _started_from_attended_chat(graph_exec, user_id, edb)
or await _started_from_attended_chat(approval_chat_id)
)
# The persisted row is authoritative on resume. A caller cannot turn
@@ -1519,6 +1520,13 @@ async def _add_graph_execution(
parent_exec_id = (
execution_context.parent_execution_id if execution_context else None
)
# A sub-graph asks in the chat that started its parent run.
if trigger == ExecutionTrigger.COPILOT:
approval_chat_id = trigger_ref
elif trigger == ExecutionTrigger.SUBGRAPH and execution_context:
approval_chat_id = execution_context.approval_chat_id
else:
approval_chat_id = None
# When execution_context is provided (e.g. from AgentExecutorBlock),
# inherit dry_run so child-graph validation skips credential checks.
@@ -1684,10 +1692,21 @@ async def _add_graph_execution(
}
)
if pause_irreversible_actions:
execution_context = execution_context.model_copy(
update={"sensitive_action_safe_mode": True}
stamp: dict[str, Any] = {
"approval_chat_id": (
None
if dry_run or execution_context.dry_run or graph_exec.is_dry_run
else approval_chat_id or None
)
}
if execution_context.graph_sensitive_action_safe_mode is None:
# Kept apart from the pause forced below: the gate holds it as a floor.
stamp["graph_sensitive_action_safe_mode"] = (
execution_context.sensitive_action_safe_mode
)
if pause_irreversible_actions:
stamp["sensitive_action_safe_mode"] = True
execution_context = execution_context.model_copy(update=stamp)
try:
graph_exec_entry = graph_exec.to_graph_execution_entry(
@@ -1783,11 +1802,15 @@ async def _add_graph_execution(
return graph_exec
async def _started_from_attended_chat(
# A sub-graph run whose parent run is gone: no chat can be named for it.
PARENT_GONE = ""
async def starting_chat(
graph_exec: GraphExecutionMeta, user_id: str, edb
) -> bool:
# A sub-graph run follows its parent run's chat. A parent or chat that is
# gone cannot prove nobody is watching, so the run pauses.
) -> str | None:
"""The chat that started this run, through its sub-graph parents: None when
no chat did, ``PARENT_GONE`` when a parent run no longer exists."""
while (
graph_exec.trigger_source == ExecutionTrigger.SUBGRAPH
and graph_exec.trigger_ref
@@ -1796,14 +1819,21 @@ async def _started_from_attended_chat(
user_id=user_id, execution_id=graph_exec.trigger_ref
)
if parent is None:
return True
return PARENT_GONE
graph_exec = parent
if (
graph_exec.trigger_source != ExecutionTrigger.COPILOT
or not graph_exec.trigger_ref
):
if graph_exec.trigger_source != ExecutionTrigger.COPILOT:
return None
return graph_exec.trigger_ref or None
async def _started_from_attended_chat(chat_id: str | None) -> bool:
# A parent or chat that is gone cannot prove nobody is watching, so the
# run pauses.
if chat_id is None:
return False
session = await chat_db().get_chat_session_metadata(graph_exec.trigger_ref)
if chat_id == PARENT_GONE:
return True
session = await chat_db().get_chat_session_metadata(chat_id)
return session is None or session.metadata.pauses_irreversible_actions
autogpt_platform/backend/backend/executor/utils_test.py
+146
View File
@@ -377,6 +377,7 @@ async def test_add_graph_execution_is_repeatable(mocker: MockerFixture):
# Mock the graph execution object
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.organization_id = None
mock_graph_exec.expert_id = None
mock_graph_exec.team_id = None
@@ -481,6 +482,7 @@ async def test_add_graph_execution_is_repeatable(mocker: MockerFixture):
# Create a second mock execution for the sanity check
mock_graph_exec_2 = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec_2.is_dry_run = False
mock_graph_exec_2.organization_id = None
mock_graph_exec_2.expert_id = None
mock_graph_exec_2.team_id = None
@@ -541,6 +543,8 @@ async def test_add_graph_execution_via_rpc_returns_typed_user(
mock_graph.version = 1
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.organization_id = None
mock_graph_exec.expert_id = None
mock_graph_exec.team_id = None
@@ -632,6 +636,8 @@ async def test_add_graph_execution_born_tenanted_via_rpc_when_prisma_disconnecte
mock_graph.version = 1
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.organization_id = "org-rpc"
mock_graph_exec.expert_id = None
mock_graph_exec.team_id = "team-rpc"
@@ -859,6 +865,7 @@ async def test_add_graph_execution_with_nodes_to_skip(mocker: MockerFixture):
# Mock the graph execution object
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.organization_id = None
mock_graph_exec.expert_id = None
mock_graph_exec.team_id = None
@@ -952,6 +959,7 @@ async def test_add_graph_execution_resume_backfills_org_from_row(mocker: MockerF
# Existing row carries org/team; the resume caller's context does not.
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.id = "exec-resume-1"
mock_graph_exec.node_executions = []
mock_graph_exec.status = ExecutionStatus.QUEUED
@@ -2138,6 +2146,8 @@ def _mock_add_graph_execution_create_path(
mock_graph.version = 1
mock_graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
mock_graph_exec.is_dry_run = False
mock_graph_exec.organization_id = org_id
mock_graph_exec.expert_id = None
mock_graph_exec.team_id = team_id
@@ -2229,6 +2239,8 @@ def _mock_add_graph_execution_requeue_path(
from backend.data.execution import GraphExecutionWithNodes
graph_exec = mocker.MagicMock(spec=GraphExecutionWithNodes)
graph_exec.is_dry_run = False
graph_exec.id = "existing-execution"
graph_exec.node_executions = []
graph_exec.status = ExecutionStatus.QUEUED
@@ -3219,3 +3231,137 @@ async def test_subgraph_inherits_the_pause_through_its_context(
child_context = add.await_args.kwargs["execution_context"]
assert child_context.sensitive_action_safe_mode is True
# ============================================================================
# A run carries the chat that started it, whose approval gate answers its
# steps' review pauses (OPEN-3592).
# ============================================================================
@pytest.mark.asyncio
@pytest.mark.parametrize(
"trigger, dry_run, stamped",
[
(ExecutionTrigger.COPILOT, False, "chat-1"),
(ExecutionTrigger.COPILOT, True, None),
(ExecutionTrigger.MANUAL, False, None),
],
)
async def test_a_new_run_is_stamped_with_the_chat_that_started_it(
mocker: MockerFixture, trigger, dry_run: bool, stamped
):
mock_edb, _ = _mock_add_graph_execution_create_path(mocker)
graph_exec = mock_edb.create_graph_execution.return_value
graph_exec.is_dry_run = dry_run
await add_graph_execution(
graph_id="g",
user_id="u",
trigger=trigger,
trigger_ref="chat-1",
dry_run=dry_run,
pause_irreversible_actions=True,
)
context = graph_exec.to_graph_execution_entry.call_args.kwargs["execution_context"]
assert context.approval_chat_id == stamped
# The workflow's own setting is kept apart from the pause the chat forced.
assert context.sensitive_action_safe_mode is True
assert context.graph_sensitive_action_safe_mode is False
@pytest.mark.asyncio
@pytest.mark.parametrize(
"trigger, stamped",
[(ExecutionTrigger.SUBGRAPH, "chat-1"), (ExecutionTrigger.MANUAL, None)],
)
async def test_only_a_sub_graph_inherits_its_parents_chat(
mocker: MockerFixture, trigger, stamped
):
from backend.data.execution import ExecutionContext
mock_edb, _ = _mock_add_graph_execution_create_path(mocker)
graph_exec = mock_edb.create_graph_execution.return_value
graph_exec.is_dry_run = False
await add_graph_execution(
graph_id="g",
user_id="u",
trigger=trigger,
trigger_ref="parent-exec",
execution_context=ExecutionContext(
approval_chat_id="chat-1",
parent_execution_id="parent-exec",
sensitive_action_safe_mode=True,
graph_sensitive_action_safe_mode=False,
),
)
context = graph_exec.to_graph_execution_entry.call_args.kwargs["execution_context"]
assert context.approval_chat_id == stamped
assert context.graph_sensitive_action_safe_mode is False
@pytest.mark.asyncio
@pytest.mark.parametrize("nested", [False, True])
async def test_a_resumed_run_is_stamped_with_the_chat_at_the_end_of_its_chain(
mocker: MockerFixture, nested: bool
):
from backend.copilot.model import ChatSessionMetadata
from backend.data.execution import ExecutionContext
graph_exec, execution_store, _, captured = _mock_add_graph_execution_requeue_path(
mocker, expert_id=None, organization_id="org", team_id="team"
)
graph_exec.is_dry_run = False
if nested:
graph_exec.trigger_source = ExecutionTrigger.SUBGRAPH.value
graph_exec.trigger_ref = "parent-exec"
execution_store.get_graph_execution_meta = mocker.AsyncMock(
return_value=mocker.MagicMock(
trigger_source=ExecutionTrigger.COPILOT.value, trigger_ref="session-1"
)
)
else:
graph_exec.trigger_source = ExecutionTrigger.COPILOT.value
graph_exec.trigger_ref = "session-1"
chat_store = mocker.MagicMock()
chat_store.get_chat_session_metadata = mocker.AsyncMock(
return_value=mocker.MagicMock(
metadata=ChatSessionMetadata(origin="interactive")
)
)
mocker.patch("backend.executor.utils.chat_db", return_value=chat_store)
# As the review route resumes it: a context rebuilt from the graph settings.
await add_graph_execution(
graph_id="g",
user_id="u",
graph_exec_id="existing-execution",
execution_context=ExecutionContext(sensitive_action_safe_mode=False),
)
assert captured["execution_context"].approval_chat_id == "session-1"
assert captured["execution_context"].graph_sensitive_action_safe_mode is False
@pytest.mark.asyncio
async def test_a_resumed_sub_graph_whose_parent_is_gone_names_no_chat(
mocker: MockerFixture,
):
graph_exec, execution_store, _, captured = _mock_add_graph_execution_requeue_path(
mocker, expert_id=None, organization_id="org", team_id="team"
)
graph_exec.is_dry_run = False
graph_exec.trigger_source = ExecutionTrigger.SUBGRAPH.value
graph_exec.trigger_ref = "deleted-parent"
execution_store.get_graph_execution_meta = mocker.AsyncMock(return_value=None)
await add_graph_execution(
graph_id="g", user_id="u", graph_exec_id="existing-execution"
)
context = captured["execution_context"]
assert context.approval_chat_id is None
assert context.sensitive_action_safe_mode is True
autogpt_platform/backend/backend/util/feature_flag/__init__.py
+4
View File
@@ -73,6 +73,10 @@ class Flag(str, Enum):
# ungated behaviour, which is why it defaults False rather than
# fail-closed. Failures *inside* an enabled gate fail to ASK.
COPILOT_AUTO_MODE = "copilot-auto-mode"
# The gate answers the review pauses of workflows AutoPilot started, and
# its yes satisfies a direct block call's own review. A rollout control
# that defaults ON by decision; off restores the two approval layers.
COPILOT_UNIFIED_APPROVALS = "copilot-unified-approvals"
# Self-distilled skills registry (store_skill / read_skill /
# delete_skill / list_skills + the per-turn <available_skills>
# context block). Default-on — flip off in LaunchDarkly to disable
autogpt_platform/frontend/src/app/(platform)/copilot/components/ApprovalQueue/ApprovalQueue.stories.tsx
+2
View File
@@ -462,3 +462,5 @@ export const RealPostToX = realStory("Post To X");
export const RealWorkflow = realStory("Workflow");
export const RealSandboxCommand = realStory("Sandbox Command");
export const RealGmailNewsletter = realStory("Gmail Newsletter");
// A step of a workflow Otto started, parked on its own review: its card in the chat.
export const RealWorkflowStep = realStory("Workflow Step");
autogpt_platform/frontend/src/app/(platform)/copilot/components/ApprovalQueue/__tests__/chatRule.test.tsx
+46 -1
View File
@@ -5,7 +5,14 @@ import { server } from "@/mocks/mock-server";
import { render, screen, waitFor } from "@/tests/integrations/test-utils";
import { CopilotChatActionsProvider } from "../../CopilotChatActionsProvider/CopilotChatActionsProvider";
import { CopilotPendingReviews } from "../../CopilotPendingReviews/CopilotPendingReviews";
import { createAgent, mail, mcpTool, CHAT_SESSION } from "./fixtures";
import {
createAgent,
mail,
mcpTool,
realCardSchemaHandler,
realCards,
CHAT_SESSION,
} from "./fixtures";
function serve(
review: ReturnType<typeof mail> = mail("m1", ["allow", "judge"]),
@@ -154,3 +161,41 @@ test.each([
]);
},
);
// The card the backend builds for a step of a workflow Otto started.
const STEP = realCards().find((card) => card.story === "Workflow Step")!.review;
test.each([
["This chat", "chat", `Approve ${MAIL} from now on`, "allow"],
["Otto, every chat", "expert", `Let Otto judge ${MAIL} from now on`, "judge"],
["Every Expert on my team", "team", `Approve ${MAIL} from now on`, "allow"],
[
"Every Expert on my team",
"team",
`Let Otto judge ${MAIL} from now on`,
"judge",
],
])(
"a workflow step's card sets a rule as an action's does: %s, %s",
async (label, scope, item, rule) => {
const sent = serve(STEP);
server.use(realCardSchemaHandler(realCards()));
renderQueue(null);
await openMenu();
await userEvent.click(
await screen.findByRole("menuitemradio", { name: label }),
);
await userEvent.click(await screen.findByText(item));
await waitFor(() => expect(sent).toHaveLength(1));
expect(sent[0].reviews).toEqual([
expect.objectContaining({
node_exec_id: STEP.node_exec_id,
approved: true,
chat_rule: rule,
chat_rule_scope: scope,
}),
]);
},
);
autogpt_platform/frontend/src/app/(platform)/copilot/components/ApprovalQueue/__tests__/realCards.json
+208 -8
View File
@@ -62,7 +62,8 @@
"object": "Gmail Send",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Gmail Send",
"editable": false,
@@ -252,7 +253,8 @@
"object": "Google Sheets Update Row",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Google Sheets Update Row",
"editable": false,
@@ -475,7 +477,8 @@
"object": "Execute Code Step",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Execute Code Step",
"editable": false,
@@ -628,7 +631,8 @@
"object": "Send Web Request",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Send Web Request",
"editable": false,
@@ -775,7 +779,8 @@
"object": "Post To X",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Post To X",
"editable": false,
@@ -1124,7 +1129,8 @@
"object": "Gmail Send",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Gmail Send",
"editable": false,
@@ -1299,7 +1305,8 @@
"object": "Morning digest",
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Morning digest",
"editable": false,
@@ -1358,12 +1365,205 @@
"object": null,
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "Run a command in the sandbox",
"editable": false,
"status": "WAITING"
},
"schema": null
},
{
"story": "Workflow Step",
"review": {
"node_exec_id": "copilot-node-gate-workflow_step:8381086bb449da65",
"node_id": "copilot-node-gate-workflow_step",
"user_id": "user-1",
"session_id": "s1",
"graph_exec_id": null,
"graph_id": null,
"graph_version": null,
"payload": {
"tool": "workflow_step",
"arguments": {
"to": ["[email protected]"],
"cc": ["[email protected]"],
"subject": "Q3 invoices missing PO numbers",
"body": "Hi Dana,\n\nThe Q3 invoice pack is in the shared Invoices folder. Three of them are still missing a PO number:\n\n- INV-2041\n- INV-2044\n- INV-2051\n\nCould you look before Friday?\n\nThanks,\nOtto"
},
"clipped": [],
"fields": [
{
"key": "to",
"label": "To",
"format": null
},
{
"key": "cc",
"label": "Cc",
"format": null
},
{
"key": "subject",
"label": "Subject",
"format": null
},
{
"key": "body",
"label": "Body",
"format": null
}
],
"references": [],
"reference_totals": {},
"tool_call_id": "",
"turn": 1,
"mode": "auto",
"subject": {
"kind": "block",
"key": "block:6c27abc2-e51d-499e-a85f-5a0041ba94f0",
"name": "Gmail Send",
"effect": "external",
"irreversible": true,
"block_id": "6c27abc2-e51d-499e-a85f-5a0041ba94f0"
},
"reason": "A step of Morning digest, which reaches outside the platform.",
"reason_kind": "subject",
"decided_by": null,
"chat_rules_allowed": ["allow", "judge"],
"headline": {
"ask": "Run",
"object": "Gmail Send",
"object_key": null
},
"spend": null,
"workflow": {
"graph_exec_id": "run-digest",
"node_exec_id": "node-exec-send",
"graph_id": "g-digest",
"graph_version": 1,
"name": "Morning digest"
}
},
"instructions": "Run “Gmail Send”",
"editable": false,
"status": "WAITING"
},
"schema": {
"properties": {
"credentials": {
"credentials_provider": ["google"],
"credentials_types": ["oauth2"],
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"title": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Title"
},
"provider": {
"const": "google",
"title": "Provider",
"type": "string"
},
"type": {
"const": "oauth2",
"title": "Type",
"type": "string"
}
},
"required": ["id", "provider", "type"],
"title": "CredentialsMetaInput[Literal[<ProviderName.GOOGLE: 'google'>], Literal['oauth2']]",
"type": "object",
"credentials_scopes": ["https://www.googleapis.com/auth/gmail.send"],
"description": "The Google integration requires OAuth2 authentication."
},
"to": {
"advanced": false,
"description": "Recipient email addresses",
"items": {
"type": "string"
},
"secret": false,
"title": "To",
"type": "array"
},
"subject": {
"advanced": false,
"description": "Email subject",
"secret": false,
"title": "Subject",
"type": "string"
},
"body": {
"advanced": false,
"description": "Email body (plain text or HTML)",
"secret": false,
"title": "Body",
"type": "string"
},
"cc": {
"advanced": true,
"description": "CC recipients",
"items": {
"type": "string"
},
"secret": false,
"title": "Cc",
"type": "array"
},
"bcc": {
"advanced": true,
"description": "BCC recipients",
"items": {
"type": "string"
},
"secret": false,
"title": "Bcc",
"type": "array"
},
"content_type": {
"advanced": true,
"anyOf": [
{
"enum": ["auto", "plain", "html"],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Content type: 'auto' (default - detects HTML), 'plain', or 'html'",
"secret": false,
"title": "Content Type"
},
"attachments": {
"advanced": true,
"description": "Files to attach",
"items": {
"format": "file",
"type": "string"
},
"secret": false,
"title": "Attachments",
"type": "array"
}
},
"required": ["credentials", "to", "subject", "body"],
"title": "Input",
"type": "object"
}
}
]
autogpt_platform/frontend/src/app/(platform)/copilot/components/ApprovalQueue/__tests__/referenceCards.json
+40 -22
View File
@@ -80,7 +80,8 @@
"object": "Q3 reports",
"object_key": "folder_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "delete_folder",
"editable": false,
@@ -300,8 +301,8 @@
}
],
"reference_totals": {
"folder_id": 1,
"agent_ids": 7
"agent_ids": 7,
"folder_id": 1
},
"tool_call_id": "call-move_agents_to_folder",
"turn": 1,
@@ -323,7 +324,8 @@
"object": "Archive",
"object_key": "folder_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "move_agents_to_folder",
"editable": false,
@@ -411,7 +413,8 @@
"object": "Daily digest",
"object_key": "schedule_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "pause_schedule",
"editable": false,
@@ -498,7 +501,8 @@
"object": "Ada",
"object_key": "name"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "hire_expert",
"editable": false,
@@ -586,7 +590,8 @@
"object": "Q3 planning",
"object_key": "session_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "message_session",
"editable": false,
@@ -696,7 +701,8 @@
"object": "GitHub (work)",
"object_key": "credential_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "grant_expert_credential",
"editable": false,
@@ -784,7 +790,8 @@
"object": "Morning digest",
"object_key": "agent_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "edit_agent",
"editable": false,
@@ -860,7 +867,8 @@
"object": "Weekly digest",
"object_key": "preset_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "update_preset",
"editable": false,
@@ -936,7 +944,8 @@
"object": "New invoice email",
"object_key": "preset_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "delete_preset",
"editable": false,
@@ -1012,7 +1021,8 @@
"object": "Month-end close",
"object_key": "routine_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "schedule_routine",
"editable": false,
@@ -1105,8 +1115,8 @@
}
],
"reference_totals": {
"store_listing_version_id": 1,
"expert_id": 1
"expert_id": 1,
"store_listing_version_id": 1
},
"tool_call_id": "call-install_expert_workflow",
"turn": 1,
@@ -1128,7 +1138,8 @@
"object": null,
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "install_expert_workflow",
"editable": false,
@@ -1197,7 +1208,8 @@
"object": "Receipt matcher",
"object_key": "workflow_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "remove_expert_workflow",
"editable": false,
@@ -1279,7 +1291,8 @@
"object": "Grace",
"object_key": "confirmation_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "confirm_expert_change",
"editable": false,
@@ -1379,7 +1392,8 @@
"object": "/reports/q3-report.pdf",
"object_key": "path"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "delete_workspace_file",
"editable": false,
@@ -1448,7 +1462,8 @@
"object": null,
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "delete_preset",
"editable": false,
@@ -1542,7 +1557,8 @@
"object": null,
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "schedule_followup",
"editable": false,
@@ -1600,7 +1616,8 @@
"object": null,
"object_key": null
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "schedule_followup",
"editable": false,
@@ -1688,7 +1705,8 @@
"object": "Ada",
"object_key": "expert_id"
},
"spend": null
"spend": null,
"workflow": null
},
"instructions": "delegate_to_expert",
"editable": false,