477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""Phase-5 target lease, timeout and ambiguous-result safety tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from maf_test_helpers import make_maf_service
|
|
from sqlalchemy import update
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.app import (
|
|
_scan_maf_approval_expiry,
|
|
_start_maf_approval_scanner,
|
|
_stop_maf_approval_scanner,
|
|
)
|
|
from app.maf_roundtable.action_coordinator import (
|
|
ActionCoordinator,
|
|
MafActionSecretKeyError,
|
|
derive_target_key,
|
|
encrypt_params,
|
|
)
|
|
from app.maf_roundtable.service import MafConflictError
|
|
from app.maf_roundtable.tool_executor import ToolExecutor
|
|
from app.maf_roundtable.workflow_executors import RoundtableState
|
|
from app.maf_roundtable.workflow_runner import WorkflowRunner
|
|
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.actions import action_to_public_dict
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
Base,
|
|
MafRoundtableActionLeaseRow,
|
|
MafRoundtableApprovalRow,
|
|
)
|
|
|
|
TEST_TOOL = "controlled_write"
|
|
|
|
|
|
def _snapshot() -> dict:
|
|
tool = {
|
|
"name": TEST_TOOL,
|
|
"side_effect": "write",
|
|
"risk_level": "R3",
|
|
"target_key_fields": ["tenant_id", "object_id"],
|
|
"target_key_namespace": "lease-test",
|
|
}
|
|
return {
|
|
"tools": [tool],
|
|
"agents": [{"agent_id": "seat-1", "tools": [tool]}],
|
|
}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
# A file-backed SQLite database gives concurrent sessions the same
|
|
# transaction/constraint semantics used by deployed SQLite installations.
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_action_leases.db'}", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
result = make_maf_roundtable_store(sf)
|
|
assert result is not None
|
|
yield result
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def svc(store):
|
|
return make_maf_service(
|
|
store,
|
|
enabled=True,
|
|
config=MafRoundtableConfig(
|
|
enabled=True,
|
|
approval_timeout_seconds=60,
|
|
target_lease_seconds=30,
|
|
),
|
|
)
|
|
|
|
|
|
async def _run(store, svc) -> tuple[dict, dict]:
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "test"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
return session, run
|
|
|
|
|
|
async def _approved_action(
|
|
store, session: dict, run: dict, *, key: str, target_key: str, agent_id: str = "seat-1"
|
|
) -> dict:
|
|
return await store.actions.create_action({
|
|
"session_id": session["id"],
|
|
"run_id": run["id"],
|
|
"agent_id": agent_id,
|
|
"tool_name": TEST_TOOL,
|
|
"idempotency_key": key,
|
|
"target_key": target_key,
|
|
"risk_level": "R3",
|
|
"status": "approved",
|
|
})
|
|
|
|
|
|
def test_target_key_is_server_derived_and_opaque(monkeypatch) -> None:
|
|
monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "explicit-hmac-test-secret")
|
|
snapshot = _snapshot()
|
|
original = {
|
|
"tenant_id": "tenant-a",
|
|
"object_id": "task-7",
|
|
"target_key": "browser-forged-value",
|
|
"secret": "do-not-persist",
|
|
}
|
|
same_target_different_noise = {
|
|
"tenant_id": "tenant-a",
|
|
"object_id": "task-7",
|
|
"target_key": "another-forged-value",
|
|
"secret": "changed",
|
|
}
|
|
target = derive_target_key(TEST_TOOL, original, snapshot, "seat-1")
|
|
assert target == derive_target_key(TEST_TOOL, same_target_different_noise, snapshot, "seat-1")
|
|
assert target.startswith("maf-target:v1:")
|
|
assert "tenant-a" not in target
|
|
assert "task-7" not in target
|
|
assert "browser-forged" not in target
|
|
|
|
nested = derive_target_key(
|
|
TEST_TOOL,
|
|
{"tenant_id": "tenant-a", "object_id": {"id": "task-7"}},
|
|
snapshot,
|
|
"seat-1",
|
|
)
|
|
nested_other = derive_target_key(
|
|
TEST_TOOL,
|
|
{"tenant_id": "tenant-a", "object_id": ["task-7"]},
|
|
snapshot,
|
|
"seat-1",
|
|
)
|
|
assert nested == nested_other
|
|
|
|
conservative_inputs = [
|
|
{"tenant_id": "tenant-a"},
|
|
{"tenant_id": "tenant-a", "object_id": None},
|
|
{"tenant_id": "tenant-a", "object_id": ""},
|
|
{"tenant_id": "tenant-a", "object_id": {"id": "task-7"}},
|
|
{"tenant_id": "tenant-a", "object_id": ["task-7"]},
|
|
]
|
|
conservative_keys = {
|
|
derive_target_key(TEST_TOOL, args, snapshot, "seat-1")
|
|
for args in conservative_inputs
|
|
}
|
|
assert len(conservative_keys) == 1
|
|
|
|
monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False)
|
|
with pytest.raises(MafActionSecretKeyError):
|
|
derive_target_key(TEST_TOOL, original, snapshot, "seat-1")
|
|
|
|
monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "another-test-domain-secret")
|
|
assert target != derive_target_key(TEST_TOOL, original, snapshot, "seat-1")
|
|
|
|
|
|
def test_action_public_dto_is_an_explicit_allowlist() -> None:
|
|
internal = {
|
|
"id": "action-1",
|
|
"session_id": "session-1",
|
|
"run_id": "run-1",
|
|
"agent_id": "seat-1",
|
|
"tool_name": TEST_TOOL,
|
|
"logical_action_id": "tool-call-1",
|
|
"turn_index": 1,
|
|
"risk_level": "R3",
|
|
"status": "waiting_approval",
|
|
"arguments": {"safe": "redacted"},
|
|
"result": None,
|
|
"external_object_id": None,
|
|
"approval_id": "approval-1",
|
|
"attempt_count": 0,
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"created_at": "2026-01-01T00:00:00+00:00",
|
|
"updated_at": "2026-01-01T00:00:00+00:00",
|
|
"target_key": "private-target",
|
|
"idempotency_key": "private-idempotency",
|
|
"internal_canary": "must-not-leak",
|
|
}
|
|
public = action_to_public_dict(internal)
|
|
assert set(public) == {
|
|
"id", "session_id", "run_id", "agent_id", "tool_name",
|
|
"logical_action_id", "turn_index", "risk_level", "status",
|
|
"arguments", "result", "external_object_id", "approval_id",
|
|
"attempt_count", "started_at", "finished_at", "created_at", "updated_at",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_agents_same_target_are_serialized(store, svc):
|
|
"""The DB lease, not a process lock, blocks the second approved write."""
|
|
session, run = await _run(store, svc)
|
|
target_key = derive_target_key(
|
|
TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
|
|
)
|
|
first = await _approved_action(store, session, run, key="lease-first", target_key=target_key)
|
|
second = await _approved_action(
|
|
store, session, run, key="lease-second", target_key=target_key, agent_id="seat-2"
|
|
)
|
|
|
|
first_claim, second_claim = await asyncio.gather(
|
|
store.claim_action_for_execution(first["id"], lease_seconds=30),
|
|
store.claim_action_for_execution(second["id"], lease_seconds=30),
|
|
)
|
|
assert sorted([first_claim, second_claim]) == ["claimed", "target_busy"]
|
|
winner = first if first_claim == "claimed" else second
|
|
waiting = second if winner["id"] == first["id"] else first
|
|
assert (await store.actions.get_action(waiting["id"]))["status"] == "approved"
|
|
|
|
assert await store.record_action_terminal_result(
|
|
winner["id"], status="succeeded", result={"safe": True}
|
|
)
|
|
assert await store.claim_action_for_execution(waiting["id"], lease_seconds=30) == "claimed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restart_recovers_expired_lease_as_unknown_without_replay(store, svc):
|
|
session, run = await _run(store, svc)
|
|
target_key = derive_target_key(
|
|
TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
|
|
)
|
|
action = await _approved_action(store, session, run, key="lease-restart", target_key=target_key)
|
|
assert await store.claim_action_for_execution(action["id"], lease_seconds=1) == "claimed"
|
|
|
|
recovered = await store.recover_expired_action_leases(
|
|
run_id=run["id"], now=datetime.now(UTC) + timedelta(seconds=5)
|
|
)
|
|
assert [item["id"] for item in recovered] == [action["id"]]
|
|
stored = await store.actions.get_action(action["id"])
|
|
assert stored["status"] == "unknown"
|
|
assert stored["result"] == {"error_code": "target_lease_expired_during_execution"}
|
|
assert await store.claim_action_for_execution(action["id"], lease_seconds=30) == "unknown"
|
|
blocked = await _approved_action(
|
|
store,
|
|
session,
|
|
run,
|
|
key="lease-restart-blocked",
|
|
target_key=target_key,
|
|
agent_id="seat-2",
|
|
)
|
|
assert await store.claim_action_for_execution(blocked["id"], lease_seconds=30) == "target_reconciliation_required"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_timeout_rejects_action_and_writes_audit_event(monkeypatch, store, svc):
|
|
monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "test-only-lease-secret")
|
|
session, run = await _run(store, svc)
|
|
coordinator = ActionCoordinator(svc, user_id="u1")
|
|
actions = await coordinator.intercept_write_tool_calls(
|
|
run_id=run["id"],
|
|
session_id=session["id"],
|
|
agent_id="seat-1",
|
|
turn_index=1,
|
|
tool_calls=[{
|
|
"name": TEST_TOOL,
|
|
"id": "timeout-call",
|
|
"args": {"tenant_id": "tenant-a", "object_id": "task-7"},
|
|
}],
|
|
capability_snapshot=_snapshot(),
|
|
write_policy="confirm_all_writes",
|
|
)
|
|
approval_id = actions[0]["approval_id"]
|
|
|
|
expired = await store.expire_due_approvals(
|
|
run_id=run["id"], now=datetime.now(UTC) + timedelta(minutes=2)
|
|
)
|
|
assert expired == [{"id": approval_id, "run_id": run["id"], "action_id": actions[0]["id"]}]
|
|
assert (await store.approvals.get_approval(approval_id))["status"] == "expired"
|
|
assert (await store.actions.get_action(actions[0]["id"]))["status"] == "rejected"
|
|
with pytest.raises(MafConflictError):
|
|
await svc.decide_approval(approval_id=approval_id, decision="approved", user_id="u1")
|
|
events = await store.events.list_events(run["id"])
|
|
assert any(
|
|
event["event_type"] == "action.rejected"
|
|
and event["payload"].get("reason") == "approval_expired"
|
|
for event in events
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_external_success_with_local_result_failure_never_replays(monkeypatch, store, svc):
|
|
"""After external success, an unavailable DB result write becomes unknown."""
|
|
monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "test-only-lease-secret")
|
|
session, run = await _run(store, svc)
|
|
target_key = derive_target_key(
|
|
TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
|
|
)
|
|
action = await _approved_action(store, session, run, key="persist-failure", target_key=target_key)
|
|
await store.action_secrets.create(
|
|
action_id=action["id"], arguments_encrypted=encrypt_params({"tenant_id": "tenant-a", "object_id": "task-7"})
|
|
)
|
|
|
|
calls = 0
|
|
|
|
class Client:
|
|
async def execute_tool_directly(self, **_kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"status": "succeeded", "result": {"executed": True}}
|
|
|
|
adapter = SimpleNamespace(_client=Client())
|
|
coordinator = ActionCoordinator(svc, user_id="u1")
|
|
|
|
async def fail_terminal_persistence(*_args, **_kwargs):
|
|
raise RuntimeError("simulated local persistence outage")
|
|
|
|
monkeypatch.setattr(coordinator, "record_result", fail_terminal_persistence)
|
|
executor = ToolExecutor(adapter=adapter, action_coordinator=coordinator, user_id="u1")
|
|
first = await executor.execute(
|
|
action_id=action["id"],
|
|
session_id=session["id"],
|
|
run_id=run["id"],
|
|
agent_id="seat-1",
|
|
thread_id="thread-1",
|
|
approved_tool=TEST_TOOL,
|
|
capability_snapshot=_snapshot(),
|
|
write_policy="confirm_all_writes",
|
|
)
|
|
second = await executor.execute(
|
|
action_id=action["id"],
|
|
session_id=session["id"],
|
|
run_id=run["id"],
|
|
agent_id="seat-1",
|
|
thread_id="thread-1",
|
|
approved_tool=TEST_TOOL,
|
|
capability_snapshot=_snapshot(),
|
|
write_policy="confirm_all_writes",
|
|
)
|
|
|
|
assert first["status"] == "unknown"
|
|
assert second["status"] == "in_progress"
|
|
assert calls == 1
|
|
recovered = await store.recover_expired_action_leases(
|
|
run_id=run["id"], now=datetime.now(UTC) + timedelta(minutes=20)
|
|
)
|
|
assert [item["id"] for item in recovered] == [action["id"]]
|
|
assert (await store.actions.get_action(action["id"]))["status"] == "unknown"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_workflow_recovery_stops_on_expired_lease_without_tool_replay(store, svc):
|
|
"""The restart entry point converts a stale running lease to waiting_input."""
|
|
session, run = await _run(store, svc)
|
|
target_key = derive_target_key(
|
|
TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
|
|
)
|
|
action = await _approved_action(store, session, run, key="runner-recovery", target_key=target_key)
|
|
assert await store.claim_action_for_execution(action["id"], lease_seconds=30) == "claimed"
|
|
async with store.session_factory() as db:
|
|
await db.execute(
|
|
update(MafRoundtableActionLeaseRow)
|
|
.where(MafRoundtableActionLeaseRow.action_id == action["id"])
|
|
.values(lease_expires_at=datetime.now(UTC) - timedelta(seconds=1))
|
|
)
|
|
await db.commit()
|
|
|
|
state = RoundtableState(
|
|
session_id=session["id"],
|
|
run_id=run["id"],
|
|
task_text="test",
|
|
mode="structured",
|
|
model_name=None,
|
|
reasoning_effort=None,
|
|
max_rounds=1,
|
|
write_policy="confirm_all_writes",
|
|
manager_agent_id=None,
|
|
seat_agent_ids=[],
|
|
thread_map={},
|
|
capability_snapshot=_snapshot(),
|
|
)
|
|
state.current_round = 1
|
|
state.status = "running"
|
|
state.pending_approvals = [{
|
|
"action_id": action["id"],
|
|
"approval_id": "already-approved",
|
|
"agent_id": "seat-1",
|
|
"tool_name": TEST_TOOL,
|
|
}]
|
|
await store.runs.update_run(run["id"], {"status": "running", "maf_checkpoint": state.to_dict()})
|
|
await store.sessions.update_session(session["id"], user_id="u1", fields={"status": "running"})
|
|
|
|
runner = WorkflowRunner(SimpleNamespace(), service=svc, user_id="u1")
|
|
recovered = await runner.recover(run["id"])
|
|
assert recovered is not None and recovered["status"] == "waiting_input"
|
|
stored = await store.actions.get_action(action["id"])
|
|
assert stored["status"] == "unknown"
|
|
assert (await store.sessions.get_session(session["id"], user_id="u1"))["status"] == "waiting_input"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_expiry_scan_is_repeatable_and_get_is_observational(monkeypatch, store, svc):
|
|
"""A scan expires once; ordinary approval GETs never perform that mutation."""
|
|
|
|
session, run = await _run(store, svc)
|
|
coordinator = ActionCoordinator(svc, user_id="u1")
|
|
action = (
|
|
await coordinator.intercept_write_tool_calls(
|
|
run_id=run["id"],
|
|
session_id=session["id"],
|
|
agent_id="seat-1",
|
|
turn_index=1,
|
|
tool_calls=[{"name": TEST_TOOL, "id": "expiry-scan", "args": {
|
|
"tenant_id": "tenant-a", "object_id": "task-7", "value": "x"
|
|
}}],
|
|
capability_snapshot=_snapshot(),
|
|
write_policy="confirm_all_writes",
|
|
)
|
|
)[0]
|
|
|
|
async with store.session_factory() as db:
|
|
await db.execute(
|
|
update(MafRoundtableApprovalRow)
|
|
.where(MafRoundtableApprovalRow.id == action["approval_id"])
|
|
.values(expires_at=datetime.now(UTC) - timedelta(seconds=1))
|
|
)
|
|
await db.commit()
|
|
|
|
# The normal read path must not mutate an expired pending row.
|
|
listed = await svc.list_approvals(run_id=run["id"], user_id="u1")
|
|
assert listed[0]["status"] == "pending"
|
|
|
|
class FakeDispatcher:
|
|
def __init__(self):
|
|
self.calls: list[tuple[str, str]] = []
|
|
|
|
async def resume(self, run_id: str, *, user_id: str):
|
|
self.calls.append((run_id, user_id))
|
|
return await store.runs.get_run(run_id)
|
|
|
|
fake = FakeDispatcher()
|
|
monkeypatch.setattr(
|
|
"app.maf_roundtable.dispatcher.get_dispatcher",
|
|
lambda _app: fake,
|
|
)
|
|
app = SimpleNamespace(
|
|
state=SimpleNamespace(maf_roundtable_service=svc),
|
|
)
|
|
|
|
concurrent = await asyncio.gather(
|
|
_scan_maf_approval_expiry(app),
|
|
_scan_maf_approval_expiry(app),
|
|
)
|
|
repeat_after_restart = await _scan_maf_approval_expiry(app)
|
|
assert sorted(concurrent, key=len) == [[], [run["id"]]]
|
|
assert repeat_after_restart == []
|
|
assert fake.calls == [(run["id"], "u1")]
|
|
assert (await store.approvals.get_approval(action["approval_id"]))["status"] == "expired"
|
|
assert (await store.actions.get_action(action["id"]))["status"] == "rejected"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_maf_approval_scanner_lifecycle_is_feature_gated():
|
|
config = MafRoundtableConfig(enabled=True, approval_scan_interval_seconds=5)
|
|
enabled_app = SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
maf_roundtable_service=SimpleNamespace(enabled=True, config=config),
|
|
)
|
|
)
|
|
task = _start_maf_approval_scanner(enabled_app)
|
|
assert task is not None
|
|
assert enabled_app.state.maf_approval_scanner_task is task
|
|
await _stop_maf_approval_scanner(enabled_app)
|
|
assert task.done()
|
|
|
|
disabled_app = SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
maf_roundtable_service=SimpleNamespace(enabled=False, config=config),
|
|
)
|
|
)
|
|
assert _start_maf_approval_scanner(disabled_app) is None
|
|
await _stop_maf_approval_scanner(disabled_app)
|