Building Google's Actus Execution Guardrail Pattern on AWS — Step-by-Step with Lambda, DynamoDB, and CloudWatch

import boto3
import json
from agentsre.pre_action_gate import PreActionSREGate
from agentsre.aro import AgentReliabilityOwnership
from agentsre.slo_burn import AgentSLOBurnTracker
from datetime import datetime, timezone, timedelta

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
cw = boto3.client('cloudwatch', region_name='us-east-1')
sns = boto3.client('sns', region_name='us-east-1')

ARO_TABLE = 'AgentReliabilityOwnership'
AUDIT_LOG_GROUP = '/agentsre/actus-audit'

def get_aro_registration(agent_id: str) -> dict:
"""
Retrieve agent's ARO record from DynamoDB.
This is the blast radius and ownership lookup.
"""
table = dynamodb.Table(ARO_TABLE)
response = table.get_item(Key={'agent_id': agent_id})
if 'Item' not in response:
raise ValueError(
f"Agent {agent_id} not registered in ARO. "
"Register before allowing production access."
)
return response['Item']

def check_blast_radius(aro: dict, target_resource: str) -> bool:
"""
Verify action target is within agent's registered blast radius.
This is the Actus boundary check.
"""
import json as _json
can_touch = _json.loads(aro.get('blast_radius', '[]'))
cannot_touch = _json.loads(aro.get('cannot_touch', '[]'))

if target_resource in cannot_touch:
return False
return target_resource in can_touch

def get_live_sre_signals(agent_id: str,
service_scope: list) -> dict:
"""Pull live CloudWatch SRE signals for gate evaluation."""
end = datetime.now(timezone.utc)
start = end - timedelta(minutes=60)
period = 3600

def cw_avg(metric: str, namespace: str,
dim_name: str, dim_val: str) -> float:
r = cw.get_metric_statistics(
Namespace=namespace, MetricName=metric,
Dimensions=[{'Name': dim_name, 'Value': dim_val}],
StartTime=start, EndTime=end,
Period=period, Statistics=['Average']
)
pts = r.get('Datapoints', [])
return round(pts[0]['Average'], 2) if pts else 0.0

budgets = [
cw_avg('ErrorBudgetRemaining', 'AgentSRE/ErrorBudget',
'ServiceName', svc)
for svc in service_scope
]

return {
'error_budget_pct': min(budgets) if budgets else 0.0,
'aqdd_depth': int(cw_avg(
'ApprovalQueueDepth', 'AgentSRE/Reliability',
'AgentId', agent_id
)),
'her_trend_pct': cw_avg(
'HumanEscalationRate', 'AgentSRE/Reliability',
'AgentId', agent_id
)
}

def lambda_handler(event: dict, context) -> dict:
"""
Actus Lambda — Google's execution guardrail pattern on AWS.

Every autonomous agent action passes through this Lambda before
execution. Three checks in order:
1. ARO registration exists
2. Target resource within blast radius
3. Live SRE signals within safe thresholds

All three must pass. One failure → escalate to ARO owner.

Event:
{
"agent_id": "incident-triage-agent-v2",
"intended_action": "scale payments-service to 8 replicas",
"target_resource": "ecs-service-desired-count",
"service_scope": ["payments-service"]
}
"""
agent_id = event['agent_id']
action = event['intended_action']
target = event.get('target_resource', '')
scope = event.get('service_scope', [])

audit = {
'agent_id': agent_id,
'intended_action': action,
'target_resource': target,
'timestamp': datetime.now(timezone.utc).isoformat(),
'checks': {}
}

# Check 1: ARO registration
try:
aro = get_aro_registration(agent_id)
audit['checks']['aro_registered'] = True
except ValueError as e:
audit['checks']['aro_registered'] = False
audit['blocked_reason'] = str(e)
_emit_audit(audit)
return _blocked_response(str(e), audit)

# Check 2: Blast radius
if target and not check_blast_radius(aro, target):
reason = (
f"Target '{target}' outside registered blast radius "
f"for agent '{agent_id}'. "
f"Cannot_touch: {aro.get('cannot_touch', '[]')}"
)
audit['checks']['blast_radius_ok'] = False
audit['blocked_reason'] = reason
_emit_audit(audit)
_alert_owner(aro, agent_id, action, reason)
return _blocked_response(reason, audit)

audit['checks']['blast_radius_ok'] = True

# Check 3: Live SRE gate
signals = get_live_sre_signals(agent_id, scope)
gate = PreActionSREGate(
error_budget_min_pct=20.0,
aqdd_max_depth=3,
her_max_trend_pct=15.0
)
result = gate.check(
agent_id=agent_id,
intended_action=action,
error_budget_pct=signals['error_budget_pct'],
aqdd_depth=signals['aqdd_depth'],
her_trend_pct=signals['her_trend_pct']
)

audit['checks']['sre_gate'] = result.approved
audit['sre_signals'] = signals
audit['gate_result'] = result.recommendation

if not result.approved:
audit['blocked_reason'] = result.blocking_check
_emit_audit(audit)
_alert_owner(aro, agent_id, action, result.recommendation)
return _blocked_response(result.recommendation, audit)

# All checks passed
audit['approved'] = True
_emit_audit(audit)
return {
'statusCode': 200,
'body': json.dumps({
'approved': True,
'recommendation': 'Proceed within registered blast radius.',
'audit_id': audit['timestamp']
})
}

def _blocked_response(reason: str, audit: dict) -> dict:
return {
'statusCode': 403,
'body': json.dumps({
'approved': False,
'blocking_reason': reason,
'audit_id': audit['timestamp'],
'escalation': 'Routed to ARO identity owner via SNS'
})
}

def _emit_audit(audit: dict) -> None:
"""Every gate check — approved or blocked — goes to CloudWatch Logs."""
import boto3 as _boto3
logs = _boto3.client('logs', region_name='us-east-1')
try:
logs.put_log_events(
logGroupName=AUDIT_LOG_GROUP,
logStreamName=audit['agent_id'],
logEvents=[{
'timestamp': int(
datetime.now(timezone.utc).timestamp() * 1000
),
'message': json.dumps(audit)
}]
)
except Exception:
pass # Never block agent path on logging failure

def _alert_owner(aro: dict, agent_id: str,
action: str, reason: str) -> None:
"""Route blocked action alert to ARO identity owner."""
owner_team = aro.get('identity_owner', 'platform-reliability')
sns.publish(
TopicArn=(
f"arn:aws:sns:us-east-1:ACCOUNT_ID:"
f"{owner_team}-alerts"
),
Subject=f"[Actus] Blocked: {agent_id}",
Message=json.dumps({
'agent_id': agent_id,
'blocked_action': action,
'reason': reason,
'timestamp': datetime.now(timezone.utc).isoformat()
})
)