Tracking Automation Complacency on AWS CloudWatch Metrics, Lambda Verification Hooks, and the ACR Signal for AWS DevOps Agent Deployments

import boto3
import json
from datetime import datetime, timezone, timedelta
from agentsre.automation_complacency import ACRTracker, VerificationRecord

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

DECISION_TABLE = 'AgentDecisionVerification'
LOG_GROUP = '/agentsre/acr-tracking'

def log_agent_decision(agent_id: str, decision_id: str,
action_taken: str, severity: str,
rtd_trace_id: str) -> dict:
"""
Log an agent decision to DynamoDB, opening the verification window.

Called immediately after any autonomous agent action.
Starts a TTL-backed verification window per severity:
P1: 15 minutes
P2: 60 minutes
P3: 240 minutes

If verification is not recorded before TTL expires,
the decision is marked 'accepted_without_verification'
and contributes to ACR.
"""
windows = {"P1": 15, "P2": 60, "P3": 240, "P4": 480}
window_mins = windows.get(severity, 120)
expiry = datetime.now(timezone.utc) + timedelta(minutes=window_mins)

table = dynamodb.Table(DECISION_TABLE)
record = {
'decision_id': decision_id,
'agent_id': agent_id,
'action_taken': action_taken,
'severity': severity,
'rtd_trace_id': rtd_trace_id,
'verification_status': 'PENDING',
'verifier': None,
'decision_at': datetime.now(timezone.utc).isoformat(),
'verification_deadline': expiry.isoformat(),
'ttl': int(expiry.timestamp()) + 86400 # DynamoDB TTL 24h after window
}
table.put_item(Item=record)
return record

def record_verification(decision_id: str, verifier: str,
reasoning_valid: bool,
rca_confirmed: bool,
notes: str = "") -> dict:
"""
Record that a human verified an agent's reasoning.

Verification means:
- Engineer read the RTD trace
- Confirmed the RCA was correct
- Confirmed the remediation was appropriate

NOT the same as accepting the outcome. Outcome acceptance
happens automatically when an incident resolves. Reasoning
verification requires active human engagement.

Args:
decision_id: The agent decision being verified
verifier: Name/ID of the engineer who verified
reasoning_valid: Was the agent's reasoning correct?
rca_confirmed: Did the RCA match what happened?
notes: Optional notes on verification
"""
table = dynamodb.Table(DECISION_TABLE)
response = table.update_item(
Key={'decision_id': decision_id},
UpdateExpression=(
"SET verification_status = :vs, verifier = :v, "
"reasoning_valid = :rv, rca_confirmed = :rc, "
"verified_at = :va, verification_notes = :n"
),
ExpressionAttributeValues={
':vs': 'VERIFIED',
':v': verifier,
':rv': reasoning_valid,
':rc': rca_confirmed,
':va': datetime.now(timezone.utc).isoformat(),
':n': notes
},
ReturnValues='ALL_NEW'
)
return response['Attributes']

def calculate_and_publish_acr(agent_id: str,
window_days: int = 30) -> dict:
"""
Calculate ACR for an agent over the rolling window
and publish to CloudWatch.

Called by EventBridge on a daily schedule.

Returns full ACR breakdown by severity — overall ACR
hides P1/P2 exposure when those incidents are rare.
"""
table = dynamodb.Table(DECISION_TABLE)
cutoff = (datetime.now(timezone.utc)
- timedelta(days=window_days)).isoformat()

# Scan decisions in window for this agent
response = table.scan(
FilterExpression='agent_id = :aid AND decision_at > :cutoff',
ExpressionAttributeValues={
':aid': agent_id,
':cutoff': cutoff
}
)
decisions = response.get('Items', [])

if not decisions:
return {'agent_id': agent_id, 'acr_pct': None,
'message': 'No decisions in window'}

# Calculate overall and per-severity ACR
by_severity = {}
for d in decisions:
sev = d.get('severity', 'Unknown')
if sev not in by_severity:
by_severity[sev] = {'total': 0, 'unverified': 0}
by_severity[sev]['total'] += 1
if d.get('verification_status') != 'VERIFIED':
by_severity[sev]['unverified'] += 1

total = len(decisions)
unverified = sum(
d.get('verification_status') != 'VERIFIED'
for d in decisions
)
overall_acr = (unverified / total * 100) if total > 0 else 0.0

# Publish overall ACR
cw.put_metric_data(
Namespace='AgentSRE/HumanGovernance',
MetricData=[{
'MetricName': 'AutomationComplacencyRate',
'Dimensions': [{'Name': 'AgentId', 'Value': agent_id}],
'Value': overall_acr,
'Unit': 'Percent'
}]
)

# Publish per-severity ACR — P1/P2 are the critical ones
for sev, counts in by_severity.items():
acr = (counts['unverified'] / counts['total'] * 100
if counts['total'] > 0 else 0.0)
cw.put_metric_data(
Namespace='AgentSRE/HumanGovernance',
MetricData=[{
'MetricName': 'AutomationComplacencyRate',
'Dimensions': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'Severity', 'Value': sev}
],
'Value': acr,
'Unit': 'Percent'
}]
)

result = {
'agent_id': agent_id,
'window_days': window_days,
'total_decisions': total,
'unverified_decisions': unverified,
'overall_acr_pct': round(overall_acr, 1),
'by_severity': {
sev: round(c['unverified']/c['total']*100, 1)
for sev, c in by_severity.items()
if c['total'] > 0
},
'calculated_at': datetime.now(timezone.utc).isoformat()
}
return result

def create_acr_alarms(agent_id: str, sns_arn: str) -> None:
"""
Create CloudWatch alarms for ACR thresholds.

P1 ACR > 5%: any P1 incident accepted without verification is critical
P2 ACR > 15%: most P2s should be verified
Overall ACR rising trend: team is drifting toward passive acceptance
"""
alarm_configs = [
{
'name': f'ACR-P1-Critical-{agent_id}',
'desc': (f'P1 automation complacency > 5% for {agent_id}. '
'Team accepting P1 agent decisions without verification.'),
'dims': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'Severity', 'Value': 'P1'}
],
'threshold': 5.0
},
{
'name': f'ACR-P2-Warning-{agent_id}',
'desc': (f'P2 automation complacency > 15% for {agent_id}. '
'Review team verification practices.'),
'dims': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'Severity', 'Value': 'P2'}
],
'threshold': 15.0
},
{
'name': f'ACR-Overall-Trending-{agent_id}',
'desc': (f'Overall ACR above 30% for {agent_id}. '
'Team may have drifted into passive acceptance pattern.'),
'dims': [{'Name': 'AgentId', 'Value': agent_id}],
'threshold': 30.0
}
]

for cfg in alarm_configs:
cw.put_metric_alarm(
AlarmName=cfg['name'],
AlarmDescription=cfg['desc'],
MetricName='AutomationComplacencyRate',
Namespace='AgentSRE/HumanGovernance',
Dimensions=cfg['dims'],
Period=86400, # Daily evaluation
EvaluationPeriods=3, # 3 consecutive days above threshold
Threshold=cfg['threshold'],
ComparisonOperator='GreaterThanThreshold',
Statistic='Average',
AlarmActions=[sns_arn],
TreatMissingData='notBreaching'
)

def lambda_handler(event: dict, context) -> dict:
"""
Daily ACR calculation Lambda.
Trigger: EventBridge cron(0 8 * * ? *) — 8am UTC daily.

Calculates ACR for all active agents, publishes to CloudWatch,
and alerts SRE lead if P1/P2 thresholds are breached.
"""
# Get all active agent IDs from ARO registry
aro_table = dynamodb.Table('AgentReliabilityOwnership')
agents_response = aro_table.scan(
FilterExpression='#s = :active',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':active': 'active'}
)
agent_ids = [
item['agent_id']
for item in agents_response.get('Items', [])
]

results = []
for agent_id in agent_ids:
result = calculate_and_publish_acr(agent_id, window_days=30)
results.append(result)

# Alert if P1 ACR is above zero
p1_acr = result.get('by_severity', {}).get('P1', 0.0)
if p1_acr > 5.0:
sns.publish(
TopicArn=event.get('sns_arn', ''),
Subject=f"[ACR CRITICAL] P1 complacency {p1_acr:.1f}% — {agent_id}",
Message=json.dumps(result, indent=2)
)

return {'statusCode': 200, 'body': json.dumps(results)}