fix(event-governance): close enforcement silent-failure gaps from review

Multi-agent review surfaced that the branch hardened three DB read/persist
paths against throttles but left the actual enforcement writes swallowing
them — defeating the ceiling/approval guarantees the branch existed to fix.

- cancel_task and async require_approval now rethrow retryable infra errors
  (throttle/5xx) so the FanOut record retries instead of silently letting a
  task run past its cost ceiling / skip its approval gate; benign conditional
  failures on the approval path are classified and swallowed as INFO
- lift isRetryableInfraError into shared retryable-error.ts (approval module
  consumes it without a circular import)
- release the idempotency marker when an enforce action rethrows, so the
  retried record can re-claim and re-run rather than being deduped away
- observe_only mode no longer fires notify/escalate side effects — gate the
  whole action block on enforce (was only guarding nudge/approval/cancel)
- inject_nudge rethrows retryable errors too
- parseEventRules (TS + Python) logs dropped malformed rules instead of
  silently vanishing a ceiling rule; Python now requires id AND on (no more
  KeyError on missing on)
- add turn_count_gte to the inline BlueprintEventRule.when.aggregate type
- drop severity 'critical' from schema + both type mirrors (the agent
  PolicyEngine only knows low/medium/high)
- remove dead evaluation-filter branch in bgagent rules eval

Tests: async require_approval (enforce + observe), inject_nudge truncation,
idempotency dedup skip, cancel terminal-status guard, retryable-release-and-
rethrow, unknown-pack 422, and a new event-rule-pack-resolver suite covering
inline/pack merge + override. cdk 2297 pass, agent + cli green.

Author: bgagent

agent/event-rules/schema.json

@@ -59,7 +59,7 @@
         "reason": { "type": "string" },
         "severity": {
           "type": "string",
-          "enum": ["low", "medium", "high", "critical"]
+          "enum": ["low", "medium", "high"]
         },
         "timeout_s": { "type": "integer", "minimum": 30, "maximum": 3600 },
         "notify_channels": {

agent/src/event_governance/evaluator.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass, field
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class EventRule:
@@ -144,8 +147,15 @@ def parse_rules(
     if not raw:
         return []
     out: list[EventRule] = []
-    for item in raw:
-        if not isinstance(item, dict) or not item.get("id"):
+    for index, item in enumerate(raw):
+        # Require both id and on, matching the CDK parser — a rule missing either
+        # is dropped with a warning rather than silently vanishing or raising a
+        # KeyError mid-loop (a dropped ceiling/approval rule = zero enforcement).
+        if not isinstance(item, dict) or not item.get("id") or not item.get("on"):
+            logger.warning(
+                "[event-governance] dropped malformed rule — missing id/on (index=%s)",
+                index,
+            )
             continue
         out.append(
             EventRule(

cdk/src/constructs/blueprint.ts

@@ -51,7 +51,7 @@ export interface BlueprintEventRule {
   readonly on: string;
   readonly when?: {
     readonly fields?: Readonly<Record<string, unknown>>;
-    readonly aggregate?: { readonly cost_usd_gte?: number };
+    readonly aggregate?: { readonly cost_usd_gte?: number; readonly turn_count_gte?: number };
   };
   readonly action: string;
   readonly mode: string;

cdk/src/handlers/shared/event-governance-approval.ts

@@ -30,6 +30,7 @@ import {
 import { ulid } from 'ulid';
 import type { EventRule } from './event-governance-types';
 import { logger } from './logger';
+import { isRetryableInfraError } from './retryable-error';
 import type { TaskRecord } from './types';
 import { TaskStatus } from '../../constructs/task-status';
 
@@ -131,12 +132,29 @@ export async function createAsyncEventApproval(options: {
     }
     return requestId;
   } catch (err) {
-    logger.warn('[event-governance] async require_approval skipped', {
+    const name = (err as { name?: string })?.name;
+    // Benign: the task already left RUNNING (TransactWrite condition), or this
+    // request_id already exists (a settled duplicate). The gate did not need to
+    // fire again — swallow without a retry.
+    if (name === 'ConditionalCheckFailedException' || name === 'TransactionCanceledException') {
+      logger.info('[event-governance] async require_approval already settled', {
+        task_id: options.task.task_id,
+        rule_id: options.rule.id,
+        status: options.task.status,
+      });
+      return undefined;
+    }
+    // A throttle/5xx must retry the record — otherwise the approval row is never
+    // written, the task never transitions to AWAITING_APPROVAL, and the gate is
+    // silently skipped while the caller reports it fired (#230).
+    if (isRetryableInfraError(err)) throw err;
+    logger.error('[event-governance] async require_approval failed (non-retryable) — enforcement gap', {
       task_id: options.task.task_id,
       rule_id: options.rule.id,
       status: options.task.status,
+      error_id: 'EVENT_GOV_APPROVAL_FAILED',
       error: err instanceof Error ? err.message : String(err),
     });
-    return undefined;
+    throw err;
   }
 }

cdk/src/handlers/shared/event-governance-async.ts

@@ -22,7 +22,7 @@
  */
 
 import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
-import { DynamoDBDocumentClient, GetCommand, PutCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb';
+import { DeleteCommand, DynamoDBDocumentClient, GetCommand, PutCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb';
 import { ulid } from 'ulid';
 import { createAsyncEventApproval } from './event-governance-approval';
 import {
@@ -34,6 +34,7 @@ import {
 } from './event-rule-evaluator';
 import { logger } from './logger';
 import { coerceNumericOrNull } from './numeric';
+import { isRetryableInfraError } from './retryable-error';
 import { NUDGE_MAX_MESSAGE_LENGTH, type EventRule, type TaskRecord } from './types';
 import { computeTtlEpoch } from './validation';
 import { TaskStatus, TERMINAL_STATUSES } from '../../constructs/task-status';
@@ -67,21 +68,10 @@ export interface AsyncGovernanceResult {
   readonly forceFanOut: boolean;
 }
 
-/** Names of AWS SDK errors that a stream-record retry can plausibly clear
- *  (throttles, transient 5xx). Distinguishes "retry the record" from a
- *  poison-pill that would stall the shard forever. */
-const RETRYABLE_AWS_ERROR = /Throttling|ProvisionedThroughputExceeded|RequestLimitExceeded|ServiceUnavailable|InternalServerError|InternalFailure|TransactionInProgress|5\d\d/i;
-
-export function isRetryableInfraError(err: unknown): boolean {
-  if (err && typeof err === 'object') {
-    const e = err as { name?: string; $retryable?: unknown; $metadata?: { httpStatusCode?: number } };
-    if (e.$retryable) return true;
-    const status = e.$metadata?.httpStatusCode;
-    if (typeof status === 'number' && status >= 500) return true;
-    if (e.name && RETRYABLE_AWS_ERROR.test(e.name)) return true;
-  }
-  return false;
-}
+// Re-exported so existing importers/tests keep resolving it from this module;
+// the canonical definition lives in ./retryable-error so the approval path can
+// consume it without a circular import back into this module.
+export { isRetryableInfraError };
 
 export async function loadTaskForGovernance(taskId: string): Promise<TaskRecord | undefined> {
   if (!TASK_TABLE) return undefined;
@@ -158,11 +148,17 @@ async function cancelTaskByRule(taskId: string, rule: EventRule, reason: string)
       }));
     }
   } catch (err) {
-    logger.warn('[event-governance] cancel_task skipped', {
+    // cancel_task IS the enforcement action ("cancel if cost exceeds $25"). A
+    // throttle/5xx here must retry the record, not silently let the task run on
+    // past its ceiling. The Update's ConditionExpression makes a re-run safe.
+    if (isRetryableInfraError(err)) throw err;
+    logger.error('[event-governance] cancel_task failed (non-retryable) — enforcement gap', {
       task_id: taskId,
       rule_id: rule.id,
+      error_id: 'EVENT_GOV_CANCEL_FAILED',
       error: err instanceof Error ? err.message : String(err),
     });
+    throw err;
   }
 }
 
@@ -195,7 +191,12 @@ async function injectNudgeByRule(
       ConditionExpression: 'attribute_not_exists(nudge_id)',
     }));
   } catch (err) {
-    logger.warn('[event-governance] inject_nudge skipped', {
+    // The nudge row is keyed by a fresh ULID, so the attribute_not_exists guard
+    // only ever fires on a genuine ULID collision (never) — any CCF here is
+    // effectively an infra fault. Rethrow retryable errors so steering isn't
+    // silently dropped on a transient blip.
+    if (isRetryableInfraError(err)) throw err;
+    logger.warn('[event-governance] inject_nudge skipped (non-retryable)', {
       task_id: task.task_id,
       rule_id: rule.id,
       error: err instanceof Error ? err.message : String(err),
@@ -366,6 +367,27 @@ async function claimIdempotency(taskId: string, ruleId: string, corr: string): P
   }
 }
 
+/** Delete the idempotency marker so a rethrown (retried) record can re-claim
+ *  and re-run the enforcement action. Best-effort: if this delete is itself
+ *  throttled the marker survives and the retry is a no-op (the action is lost),
+ *  but that is strictly better than double-firing, and the enforce actions
+ *  (cancel/approval) are themselves conditional so re-running is safe. */
+async function releaseIdempotency(taskId: string, ruleId: string, corr: string): Promise<void> {
+  if (!EVENTS_TABLE) return;
+  try {
+    await ddb.send(new DeleteCommand({
+      TableName: EVENTS_TABLE,
+      Key: { task_id: taskId, event_id: idempotencyEventId(taskId, ruleId, corr) },
+    }));
+  } catch (err) {
+    logger.warn('[event-governance] idempotency release failed', {
+      task_id: taskId,
+      rule_id: ruleId,
+      error: err instanceof Error ? err.message : String(err),
+    });
+  }
+}
+
 /** Channels for a notify/escalate rule; falls back to a default so a rule with
  *  no ``notify_channels`` still delivers rather than silently doing nothing. */
 function resolveChannels(rule: EventRule, fallback: string[]): string[] {
@@ -406,42 +428,48 @@ export async function evaluateAsyncEventRules(
     const meta = buildPolicyDecisionMetadata(rule, event, enforce, corr);
     await emitPolicyDecision(event.task_id, { ...meta, correlation_id: corr });
 
-    if (rule.action === 'observe_only' || (rule.action === 'require_approval' && !enforce)) {
-      continue;
-    }
-
-    if (rule.action === 'notify') {
-      notifyChannels.push(...resolveChannels(rule, ['slack']));
-      forceFanOut = true;
-    }
+    // observe_only mode records the policy_decision above but must NOT fire the
+    // action — that is the whole "would have fired" contract (design §8).
+    if (!enforce) continue;
+
+    // An enforce action that throws a retryable infra error must retry the whole
+    // record; release the idempotency marker first so the retry can re-claim and
+    // re-run, rather than being suppressed as a duplicate.
+    try {
+      if (rule.action === 'notify') {
+        notifyChannels.push(...resolveChannels(rule, ['slack']));
+        forceFanOut = true;
+      }
 
-    if (rule.action === 'escalate') {
-      notifyChannels.push(...resolveChannels(rule, ['email', 'slack']));
-      forceFanOut = true;
-    }
+      if (rule.action === 'escalate') {
+        notifyChannels.push(...resolveChannels(rule, ['email', 'slack']));
+        forceFanOut = true;
+      }
 
-    if (rule.action === 'inject_nudge' && enforce && task) {
-      await injectNudgeByRule(task, rule, event.metadata);
-    }
+      if (rule.action === 'inject_nudge' && task) {
+        await injectNudgeByRule(task, rule, event.metadata);
+      }
 
-    if (rule.action === 'require_approval' && enforce && task) {
-      await createAsyncEventApproval({
-        task,
-        rule,
-        eventType: event.event_type,
-        metadata: event.metadata,
-      });
-      forceFanOut = true;
-    }
+      if (rule.action === 'require_approval' && task) {
+        await createAsyncEventApproval({
+          task,
+          rule,
+          eventType: event.event_type,
+          metadata: event.metadata,
+        });
+        forceFanOut = true;
+      }
 
-    if (rule.action === 'cancel_task' && enforce && task) {
-      if (!TERMINAL_STATUSES.includes(task.status)) {
+      if (rule.action === 'cancel_task' && task && !TERMINAL_STATUSES.includes(task.status)) {
         await cancelTaskByRule(
           event.task_id,
           rule,
           rule.reason ?? `Event rule ${rule.id} triggered cancel_task`,
         );
       }
+    } catch (err) {
+      await releaseIdempotency(event.task_id, rule.id, corr);
+      throw err;
     }
   }
 

cdk/src/handlers/shared/event-governance-types.ts

@@ -51,7 +51,7 @@ export interface EventRule {
   readonly mode: EventRuleMode;
   readonly evaluation: EventRuleEvaluation;
   readonly reason?: string;
-  readonly severity?: 'low' | 'medium' | 'high' | 'critical';
+  readonly severity?: 'low' | 'medium' | 'high';
   readonly timeout_s?: number;
   readonly notify_channels?: readonly NotificationChannelWithWebhook[];
   readonly rule_pack_id?: string;

cdk/src/handlers/shared/event-rule-evaluator.ts

@@ -154,10 +154,21 @@ export function buildPolicyDecisionMetadata(
 export function parseEventRules(raw: unknown): EventRule[] {
   if (!Array.isArray(raw)) return [];
   const out: EventRule[] = [];
-  for (const item of raw) {
-    if (!item || typeof item !== 'object') continue;
+  for (const [index, item] of raw.entries()) {
+    if (!item || typeof item !== 'object') {
+      logger.warn('[event-governance] dropped malformed rule — not an object', { index });
+      continue;
+    }
     const r = item as Record<string, unknown>;
-    if (typeof r.id !== 'string' || typeof r.on !== 'string') continue;
+    if (typeof r.id !== 'string' || typeof r.on !== 'string') {
+      // Fail loud: a dropped ceiling/approval rule means zero enforcement with no
+      // other signal. Mirror the fail-loud stance the pack resolver takes (#230).
+      logger.warn('[event-governance] dropped malformed rule — missing id/on', {
+        index,
+        id: typeof r.id === 'string' ? r.id : undefined,
+      });
+      continue;
+    }
     out.push({
       id: r.id,
       on: r.on,

cdk/src/handlers/shared/retryable-error.ts

@@ -0,0 +1,41 @@
+/**
+ *  MIT No Attribution
+ *
+ *  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining a copy of
+ *  the Software without restriction, including without limitation the rights to
+ *  use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ *  the Software, and to permit persons to whom the Software is furnished to do so.
+ *
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ *  SOFTWARE.
+ */
+
+/** Names of AWS SDK errors that a stream-record retry can plausibly clear
+ *  (throttles, transient 5xx). Distinguishes "retry the record" from a
+ *  poison-pill that would stall the shard forever. */
+const RETRYABLE_AWS_ERROR = /Throttling|ProvisionedThroughputExceeded|RequestLimitExceeded|ServiceUnavailable|InternalServerError|InternalFailure|TransactionInProgress|5\d\d/i;
+
+/**
+ * True when an error is a transient infra fault (throttle / 5xx) that a stream
+ * retry can clear — as opposed to a benign conditional failure or a
+ * deterministic client error that would just fail again. Governance
+ * enforcement paths rethrow these so the FanOut record enters
+ * ``batchItemFailures`` instead of silently skipping a cancel/approval (#230).
+ */
+export function isRetryableInfraError(err: unknown): boolean {
+  if (err && typeof err === 'object') {
+    const e = err as { name?: string; $retryable?: unknown; $metadata?: { httpStatusCode?: number } };
+    if (e.$retryable) return true;
+    const status = e.$metadata?.httpStatusCode;
+    if (typeof status === 'number' && status >= 500) return true;
+    if (e.name && RETRYABLE_AWS_ERROR.test(e.name)) return true;
+  }
+  return false;
+}

cdk/test/handlers/get-event-rules.test.ts

@@ -143,4 +143,22 @@ describe('get-event-rules', () => {
     const ruleIds = body.data.rules.map((r: { rule_id: string }) => r.rule_id);
     expect(ruleIds).toContain('observe-repo-setup');
   });
+
+  test('422 VALIDATION_ERROR when repo pins an unknown event-rule-pack', async () => {
+    mockSend.mockResolvedValue({});
+    mockLoadRepoConfig.mockResolvedValue({
+      repo: 'owner/repo',
+      status: 'active',
+      onboarded_at: '',
+      updated_at: '',
+      event_rule_pack: { id: 'does-not-exist', version: '9.9.9' },
+    });
+
+    const res = await handler(makeEvent());
+    // Fail loud rather than silently applying zero governance rules (#230).
+    expect(res.statusCode).toBe(422);
+    const body = JSON.parse(res.body);
+    expect(body.error.code).toBe('VALIDATION_ERROR');
+    expect(body.error.message).toContain('does-not-exist@9.9.9');
+  });
 });