fix(hitl): emit log_event() on approval grant and denial — Art. 14 audit gap (closes #893)

The @requires_approval decorator and request_approval() call executed the
approval gate correctly but never wrote the outcome to the activity log.
EU AI Act Article 14 requires documented evidence that HITL measures were
exercised — the missing log_event() calls meant GET /workspaces/:id/activity
could not surface HITL gate outcomes.

Add log_event() at both resolution points in the requires_approval wrapper:
- Denial: event_type="hitl", action="approve", outcome="denied", actor=decided_by
- Grant:  event_type="hitl", action="approve", outcome="granted", actor=decided_by

Both calls follow the existing try/except pattern used for audit calls elsewhere
in hitl.py so a missing audit module never blocks the approval flow.

Tests: TestRequiresApproval.test_logs_hitl_denied_event and
test_logs_hitl_approved_event verify log_event is called with the correct
outcome on each resolution path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Molecule AI Backend Engineer 2026-04-18 03:10:26 +00:00
parent bcd256946f
commit f9973fda77
2 changed files with 113 additions and 0 deletions

View File

@ -385,6 +385,21 @@ def requires_approval(
}
if not approval_result.get("approved"):
# Art. 14 audit: log the denial outcome so the activity log
# contains evidence that the human oversight gate was exercised.
try:
from builtin_tools.audit import log_event
log_event(
event_type="hitl",
action="approve",
resource=action,
outcome="denied",
actor=approval_result.get("decided_by"),
approval_id=approval_result.get("approval_id"),
reason=reason,
)
except Exception:
pass
return {
"success": False,
"error": (
@ -394,6 +409,21 @@ def requires_approval(
"approval_id": approval_result.get("approval_id"),
}
# Art. 14 audit: log the approval grant before running the function.
try:
from builtin_tools.audit import log_event
log_event(
event_type="hitl",
action="approve",
resource=action,
outcome="granted",
actor=approval_result.get("decided_by"),
approval_id=approval_result.get("approval_id"),
reason=reason,
)
except Exception:
pass
# --- Approved — run the original function ------------------------
return await fn(*args, **kwargs)

View File

@ -352,6 +352,89 @@ class TestRequiresApproval:
assert result["success"] is False
assert "error" in result
@pytest.mark.asyncio
async def test_logs_hitl_denied_event(self, monkeypatch):
"""Art. 14 audit: denial outcome must be logged to activity_logs (#893)."""
mod = _load_hitl(monkeypatch)
audit_mock = MagicMock()
audit_mock.log_event = MagicMock(return_value="trace-id")
monkeypatch.setitem(sys.modules, "builtin_tools.audit", audit_mock)
approval_mock = MagicMock()
approval_mock.ainvoke = AsyncMock(return_value={
"approved": False,
"approval_id": "appr-deny-123",
"decided_by": "human-reviewer",
"message": "Denied by human",
})
monkeypatch.setitem(sys.modules, "builtin_tools.approval",
MagicMock(request_approval=approval_mock))
@mod.requires_approval("Delete production DB")
async def delete_db():
return {"done": True}
result = await delete_db()
assert result["success"] is False
# log_event must have been called with the denial outcome.
log_calls = audit_mock.log_event.call_args_list
denial_calls = [
c for c in log_calls
if c.kwargs.get("outcome") == "denied"
or (c.args and len(c.args) >= 3 and c.args[2] == "denied")
]
assert denial_calls, (
"log_event(outcome='denied') was not called — Art. 14 audit gap (issue #893)"
)
# Verify the call carries the expected resource / actor.
dc = denial_calls[0]
assert dc.kwargs.get("event_type") == "hitl" or "hitl" in str(dc)
assert dc.kwargs.get("outcome") == "denied"
@pytest.mark.asyncio
async def test_logs_hitl_approved_event(self, monkeypatch):
"""Art. 14 audit: approval grant outcome must be logged to activity_logs (#893)."""
mod = _load_hitl(monkeypatch)
audit_mock = MagicMock()
audit_mock.log_event = MagicMock(return_value="trace-id")
monkeypatch.setitem(sys.modules, "builtin_tools.audit", audit_mock)
approval_mock = MagicMock()
approval_mock.ainvoke = AsyncMock(return_value={
"approved": True,
"approval_id": "appr-ok-456",
"decided_by": "human-reviewer",
})
monkeypatch.setitem(sys.modules, "builtin_tools.approval",
MagicMock(request_approval=approval_mock))
executed = []
@mod.requires_approval("Run migration")
async def run_migration(table: str):
executed.append(table)
return {"done": True}
result = await run_migration(table="users")
assert result == {"done": True}
assert executed == ["users"]
# log_event must have been called with the granted outcome.
log_calls = audit_mock.log_event.call_args_list
granted_calls = [
c for c in log_calls
if c.kwargs.get("outcome") == "granted"
]
assert granted_calls, (
"log_event(outcome='granted') was not called — Art. 14 audit gap (issue #893)"
)
gc = granted_calls[0]
assert gc.kwargs.get("event_type") == "hitl"
assert gc.kwargs.get("outcome") == "granted"
# ============================================================================
# HITLConfig loading