fix(event-governance): close idempotency/enforcement ordering gaps from 2nd review

bgagent · bgagent · commit ecf9fc5970a7 · 2026-07-09T13:50:46.000-05:00
- policy_decision emit is now inside the release-guarded try alongside the
  enforce action: a retryable throttle on the audit emit (which rethrows)
  used to leave the idempotency marker claimed but the action un-run, so the
  stream retry skipped the rule entirely — dropping a cost ceiling / approval
  gate silently. Emit + action now share one release-on-failure path.
- cancelTaskByRule swallows a benign ConditionalCheckFailedException (task
  raced to terminal between load and the cancel Update) as INFO instead of
  rethrowing, mirroring the approval path — a CCF no longer parks the record
  in batchItemFailures forever.
- claimIdempotency rethrows retryable infra errors instead of proceeding on an
  unclaimed marker, which risked a double notify/escalate on the eventual retry.

Nits: comment get-event-rules cache as best-effort/per-instance; note that the
CLI rules-eval matcher is a dry-run copy with the parity suites as source of truth.

Tests: emit-failure-release, cancel CCF swallow (no release), claim-throttle
rethrow. cdk 2300 pass, cli green.
diff --git a/cdk/src/handlers/get-event-rules.ts b/cdk/src/handlers/get-event-rules.ts
@@ -36,6 +36,10 @@ const POLICIES_RATE_LIMIT_PER_MINUTE = Number(process.env.POLICIES_RATE_LIMIT_PE
 
 const CACHE_TTL_MINUTES = 5;
 const CACHE_TTL_MS = CACHE_TTL_MINUTES * 60 * 1000;
+// Best-effort, per-Lambda-instance cache — not shared across concurrent
+// instances, and a blueprint edit takes up to CACHE_TTL_MINUTES to surface.
+// Harmless at current scale (read-only rule listing); revisit if staleness or
+// cross-instance consistency ever matters.
 const cache = new Map<string, { response: GetEventRulesResponse; expiresAt: number }>();
 
 function summarizeRule(rule: EventRule): {
diff --git a/cdk/src/handlers/shared/event-governance-async.ts b/cdk/src/handlers/shared/event-governance-async.ts
@@ -148,9 +148,21 @@ async function cancelTaskByRule(taskId: string, rule: EventRule, reason: string)
       }));
     }
   } catch (err) {
+    const name = (err as { name?: string })?.name;
+    // Benign: the task raced to a terminal status between loadTaskForGovernance
+    // and this Update, so the climb-only ConditionExpression rejected it. The
+    // task is already done/cancelled — nothing to enforce. Swallow (mirrors the
+    // approval path) rather than parking the record in batchItemFailures forever.
+    if (name === 'ConditionalCheckFailedException') {
+      logger.info('[event-governance] cancel_task skipped — task already terminal', {
+        task_id: taskId,
+        rule_id: rule.id,
+      });
+      return;
+    }
     // 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.
+    // past its ceiling.
     if (isRetryableInfraError(err)) throw err;
     logger.error('[event-governance] cancel_task failed (non-retryable) — enforcement gap', {
       task_id: taskId,
@@ -357,8 +369,13 @@ async function claimIdempotency(taskId: string, ruleId: string, corr: string): P
     return true;
   } catch (err) {
     const name = (err as { name?: string })?.name;
+    // Marker already exists — a genuine duplicate delivery. Skip the rule.
     if (name === 'ConditionalCheckFailedException') return false;
-    logger.warn('[event-governance] idempotency claim failed — proceeding', {
+    // A throttle/5xx on the claim itself must retry the whole record rather than
+    // proceed: proceeding on an unclaimed marker risks double-firing a notify /
+    // escalate (which are not conditional-safe) on the eventual retry (#230).
+    if (isRetryableInfraError(err)) throw err;
+    logger.warn('[event-governance] idempotency claim failed (non-retryable) — proceeding', {
       task_id: taskId,
       rule_id: ruleId,
       error: err instanceof Error ? err.message : String(err),
@@ -426,46 +443,49 @@ export async function evaluateAsyncEventRules(
 
     const enforce = rule.mode === 'enforce';
     const meta = buildPolicyDecisionMetadata(rule, event, enforce, corr);
-    await emitPolicyDecision(event.task_id, { ...meta, correlation_id: corr });
 
-    // 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.
+    // The audit emit AND the enforce action share one release-guarded try: the
+    // idempotency marker is already claimed, so if EITHER throws a retryable
+    // error we must release the marker before the record retries — otherwise the
+    // retry re-claims false and the rule is skipped entirely, silently dropping
+    // the audit record (observe_only) or the enforcement action (#230).
     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 === 'inject_nudge' && task) {
-        await injectNudgeByRule(task, rule, event.metadata);
-      }
-
-      if (rule.action === 'require_approval' && task) {
-        await createAsyncEventApproval({
-          task,
-          rule,
-          eventType: event.event_type,
-          metadata: event.metadata,
-        });
-        forceFanOut = true;
-      }
-
-      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`,
-        );
+      await emitPolicyDecision(event.task_id, { ...meta, correlation_id: corr });
+
+      // observe_only records the policy_decision above but must NOT fire the
+      // action — the "would have fired" contract (design §8).
+      if (enforce) {
+        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 === 'inject_nudge' && task) {
+          await injectNudgeByRule(task, rule, event.metadata);
+        }
+
+        if (rule.action === 'require_approval' && task) {
+          await createAsyncEventApproval({
+            task,
+            rule,
+            eventType: event.event_type,
+            metadata: event.metadata,
+          });
+          forceFanOut = true;
+        }
+
+        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);
diff --git a/cdk/test/handlers/shared/event-governance-async-module.test.ts b/cdk/test/handlers/shared/event-governance-async-module.test.ts
@@ -557,5 +557,133 @@ describe('event-governance-async module', () => {
     // The idempotency marker was released (Delete) so the retried record re-runs.
     expect(mockSend.mock.calls.some((c) => c[0]._type === 'Delete')).toBe(true);
   });
+
+  const isPolicyDecisionPut = (c: any) =>
+    c[0]._type === 'Put' && c[0].input?.Item?.event_type === 'policy_decision';
+  const isIdemClaim = (c: any) =>
+    c[0]._type === 'Put' && c[0].input?.Item?.event_type === 'governance_idempotency';
+
+  test('retryable error in policy_decision emit releases idempotency and rethrows', async () => {
+    // The audit emit runs AFTER the claim but must be release-guarded too: if it
+    // throttles, the marker has to be released so the retry re-claims and re-runs
+    // rather than being deduped away with the enforcement action never fired.
+    const throttle = Object.assign(new Error('throttled'), { name: 'ThrottlingException' });
+    mockSend.mockImplementation((cmd: any) =>
+      isPolicyDecisionPut([cmd]) ? Promise.reject(throttle) : Promise.resolve({}),
+    );
+    const mod = await import('../../../src/handlers/shared/event-governance-async');
+    mod._resetGovernanceIdempotencyCache();
+    await expect(
+      mod.evaluateAsyncEventRules(
+        {
+          task_id: 't-emit',
+          event_id: 'e-emit',
+          event_type: 'agent_cost_update',
+          metadata: { cumulative_cost_usd: 30 },
+        },
+        {
+          aggregateState: { cumulative_cost_usd: 30 },
+          task: {
+            task_id: 't-emit',
+            status: 'RUNNING',
+            event_rules: [
+              {
+                id: 'cap',
+                on: 'agent_cost_update',
+                when: { aggregate: { cost_usd_gte: 25 } },
+                action: 'cancel_task',
+                mode: 'enforce',
+                evaluation: 'async',
+              },
+            ],
+          } as any,
+        },
+      ),
+    ).rejects.toBe(throttle);
+    expect(mockSend.mock.calls.some((c) => c[0]._type === 'Delete')).toBe(true);
+    // The enforcement action never ran (emit failed first).
+    expect(mockSend.mock.calls.some(isCancelUpdate)).toBe(false);
+  });
+
+  test('cancel_task swallows a benign ConditionalCheckFailedException (terminal race)', async () => {
+    // Task passed the loaded-snapshot terminal check but raced to terminal before
+    // the Update — the climb-only condition rejects it. That must NOT park the
+    // record in batchItemFailures; it should be swallowed and NOT released.
+    const ccf = Object.assign(new Error('stale'), { name: 'ConditionalCheckFailedException' });
+    mockSend.mockImplementation((cmd: any) =>
+      isCancelUpdate([cmd]) ? Promise.reject(ccf) : Promise.resolve({}),
+    );
+    const mod = await import('../../../src/handlers/shared/event-governance-async');
+    mod._resetGovernanceIdempotencyCache();
+    // Resolves (no throw) — the record is considered handled.
+    await expect(
+      mod.evaluateAsyncEventRules(
+        {
+          task_id: 't-race',
+          event_id: 'e-race',
+          event_type: 'agent_cost_update',
+          metadata: { cumulative_cost_usd: 30 },
+        },
+        {
+          aggregateState: { cumulative_cost_usd: 30 },
+          task: {
+            task_id: 't-race',
+            status: 'RUNNING',
+            event_rules: [
+              {
+                id: 'cap',
+                on: 'agent_cost_update',
+                when: { aggregate: { cost_usd_gte: 25 } },
+                action: 'cancel_task',
+                mode: 'enforce',
+                evaluation: 'async',
+              },
+            ],
+          } as any,
+        },
+      ),
+    ).resolves.toBeDefined();
+    // Benign CCF is not a failure — marker stays claimed, no release.
+    expect(mockSend.mock.calls.some((c) => c[0]._type === 'Delete')).toBe(false);
+  });
+
+  test('retryable error in idempotency claim rethrows instead of double-firing', async () => {
+    // A throttle on the claim Put must retry the whole record, not proceed on an
+    // unclaimed marker (which would risk a double notify/escalate on retry).
+    const throttle = Object.assign(new Error('throttled'), { name: 'ProvisionedThroughputExceededException' });
+    mockSend.mockImplementation((cmd: any) =>
+      isIdemClaim([cmd]) ? Promise.reject(throttle) : Promise.resolve({}),
+    );
+    const mod = await import('../../../src/handlers/shared/event-governance-async');
+    mod._resetGovernanceIdempotencyCache();
+    await expect(
+      mod.evaluateAsyncEventRules(
+        {
+          task_id: 't-claim',
+          event_id: 'e-claim',
+          event_type: 'agent_milestone',
+          metadata: { milestone: 'pr_created' },
+        },
+        {
+          task: {
+            task_id: 't-claim',
+            status: 'RUNNING',
+            event_rules: [
+              {
+                id: 'notify',
+                on: 'pr_created',
+                action: 'notify',
+                mode: 'enforce',
+                evaluation: 'async',
+                notify_channels: ['slack'],
+              },
+            ],
+          } as any,
+        },
+      ),
+    ).rejects.toBe(throttle);
+    // No audit emit — we bailed at the claim, before any side effect.
+    expect(mockSend.mock.calls.some(isPolicyDecisionPut)).toBe(false);
+  });
 });
 
diff --git a/cli/src/commands/rules.ts b/cli/src/commands/rules.ts
@@ -46,6 +46,13 @@ function eventName(eventType: string, metadata: Record<string, unknown>): string
   return eventType;
 }
 
+// Local, deliberately-minimal matcher for the `rules eval` dry-run. It does NOT
+// filter by `evaluation` (fixtures are single-mode) and is a third copy of the
+// matching logic — the authoritative implementations are
+// agent/src/event_governance/evaluator.py and cdk .../event-rule-evaluator.ts,
+// kept in lockstep by the shared-fixture parity suites. If this drifts, those
+// parity tests are the source of truth; keep this in sync by hand or route
+// through a shared package when one exists.
 function matchRules(fixture: FixtureFile): string[] {
   const name = eventName(fixture.event.event_type, fixture.event.metadata ?? {});
   const meta = fixture.event.metadata ?? {};
