fix(event-governance): address review findings

- rethrow retryable infra errors (throttle/5xx) from task load and
  high-water update so the stream retries instead of silently
  under-counting a cost/turn ceiling
- seed the monotonic cost mark from authoritative task.cost_usd so a
  ceiling can trip on a non-cost event and the first eval of an
  already-costed task
- normalize aggregate metadata aliases on both evaluators; drop the
  duplicate cumulative_* / turn_count aliases the progress writer emitted
- default gate_on_event_async ts_module to the real task_state so a
  missing module pauses for approval instead of crashing
- key get-event-rules cache on repo + workflow_ref (rules depend on the
  workflow's eventRulePack)
- add coverage for the new retry/seed paths

Author: bgagent

agent/src/event_governance/evaluator.py

@@ -48,6 +48,31 @@ def _fields_match(rule_when: dict[str, Any], metadata: dict[str, Any]) -> bool:
     return True
 
 
+# Canonical aggregate metadata fields ↔ accepted aliases (base field a producer
+# emits ↔ cumulative field the rule reads). Parity with the CDK evaluator's
+# AGGREGATE_FIELDS: the evaluator normalizes so producers emit ONE name. See #230.
+_AGGREGATE_ALIASES = {
+    "cost": ("cumulative_cost_usd", "cost_usd"),
+    "turns": ("turn_count", "turn"),
+}
+
+
+def _read_aggregate(
+    metadata: dict[str, Any],
+    aliases: tuple[str, ...],
+) -> float | None:
+    """First value among ``aliases`` that coerces to a finite float."""
+    for key in aliases:
+        raw = metadata.get(key)
+        if raw is None:
+            continue
+        try:
+            return float(raw)
+        except (TypeError, ValueError):
+            continue
+    return None
+
+
 def _aggregate_match(
     rule_when: dict[str, Any],
     metadata: dict[str, Any],
@@ -62,7 +87,7 @@ def _aggregate_match(
     if cost_gte is not None:
         cumulative = aggregate_state.get("cumulative_cost_usd") if aggregate_state else None
         if cumulative is None:
-            cumulative = metadata.get("cumulative_cost_usd")
+            cumulative = _read_aggregate(metadata, _AGGREGATE_ALIASES["cost"])
         if cumulative is None:
             return False
         try:
@@ -74,7 +99,7 @@ def _aggregate_match(
     if turn_gte is not None:
         turns = aggregate_state.get("turn_count") if aggregate_state else None
         if turns is None:
-            turns = metadata.get("turn_count")
+            turns = _read_aggregate(metadata, _AGGREGATE_ALIASES["turns"])
         if turns is None:
             return False
         try:

agent/src/event_governance/gate.py

@@ -8,6 +8,7 @@
 if TYPE_CHECKING:
     from event_governance.evaluator import EventRule
 
+import task_state
 from hooks import _handle_require_approval
 from policy import PolicyDecision
 
@@ -32,6 +33,11 @@ async def gate_on_event_async(
     ts_module: Any = None,
 ) -> EventGateResult:
     """Block on human approval for an event rule (enforce + require_approval)."""
+    # Default to the real task_state module. _handle_require_approval dereferences
+    # ``ts`` unconditionally (increment_approval_gate_count_in_ddb,
+    # transact_write_approval_request); a None here crashes the gate instead of
+    # pausing for approval. The parameter stays overridable for tests. See #230.
+    ts_module = ts_module if ts_module is not None else task_state
     checkpoint = metadata.get("checkpoint") or rule.on
     tool_input = {"checkpoint": checkpoint, "event_type": event_type, **metadata}
     decision = PolicyDecision.require_approval(

agent/src/progress_writer.py

@@ -612,10 +612,6 @@ def write_agent_turn(
             "agent_turn",
             {
                 "turn": turn,
-                # Cumulative alias read by event-governance aggregate rules
-                # (turn_count_gte). ``turn`` is the SDK's running turn index, so
-                # it is already the per-session cumulative count. See #230.
-                "turn_count": turn,
                 "model": model,
                 "thinking_preview": self._preview(thinking),
                 "text_preview": self._preview(text),
@@ -690,14 +686,9 @@ def write_agent_cost_update(
             "agent_cost_update",
             {
                 "cost_usd": cost_usd,
-                # Cumulative alias read by event-governance aggregate rules
-                # (cost_usd_gte). SDK ``total_cost_usd`` is the running
-                # session total, so it is already cumulative. See #230.
-                "cumulative_cost_usd": cost_usd,
                 "input_tokens": input_tokens,
                 "output_tokens": output_tokens,
                 "turn": turn,
-                "turn_count": turn,
             },
         )
 

cdk/src/handlers/fanout-task-events.ts

@@ -47,7 +47,7 @@ import type {
 } from 'aws-lambda';
 import { clearTokenCache, resolveGitHubToken } from './shared/context-hydration';
 import { classifyError } from './shared/error-classifier';
-import { evaluateAsyncEventRules, loadTaskForGovernance } from './shared/event-governance-async';
+import { evaluateAsyncEventRules, isRetryableInfraError, loadTaskForGovernance } from './shared/event-governance-async';
 import { renderCommentBody, upsertTaskComment } from './shared/github-comment';
 import { postIssueComment } from './shared/linear-feedback';
 import { logger } from './shared/logger';
@@ -1284,6 +1284,11 @@ export const handler = async (
           governanceForceFanOut = govResult.forceFanOut || governanceForcedChannels.length > 0;
         }
       } catch (govErr) {
+        // Infra errors (DDB throttle / 5xx) must retry the record — otherwise an
+        // enforce cancel_task / require_approval is silently skipped and the
+        // cost/turn high-water mark under-counts (#230). Benign failures degrade
+        // to fanout-only, as before, to preserve poison-pill isolation.
+        if (isRetryableInfraError(govErr)) throw govErr;
         logger.warn('[fanout] event governance evaluation failed — continuing fanout', {
           task_id: ev.task_id,
           event_id: ev.event_id,

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

@@ -21,7 +21,7 @@ import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
 import { DynamoDBDocumentClient, UpdateCommand } from '@aws-sdk/lib-dynamodb';
 import type { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda';
 import { ulid } from 'ulid';
-import { listBuiltinEventRulePacks, resolveEventRules } from './shared/event-rule-pack-resolver';
+import { listBuiltinEventRulePacks, resolveEventRules, UnknownEventRulePackError } from './shared/event-rule-pack-resolver';
 import { extractUserId } from './shared/gateway';
 import { logger } from './shared/logger';
 import { formatMinuteBucket, RATE_LIMIT_ROW_TTL_SECONDS } from './shared/rate-limit';
@@ -107,7 +107,12 @@ export async function handler(event: APIGatewayProxyEvent): Promise<APIGatewayPr
       }
     }
 
-    const cached = cache.get(repoId);
+    // Cache key includes workflow_ref: resolved rules depend on the workflow's
+    // eventRulePack, so keying on repo alone would serve one workflow's rules
+    // for another within the TTL window (#230).
+    const workflowRef = event.queryStringParameters?.workflow_ref;
+    const cacheKey = `${repoId}\n${workflowRef ?? ''}`;
+    const cached = cache.get(cacheKey);
     if (cached && cached.expiresAt > Date.now()) {
       return successResponse(200, cached.response, requestId);
     }
@@ -123,13 +128,20 @@ export async function handler(event: APIGatewayProxyEvent): Promise<APIGatewayPr
     }
 
     const repoConfig = await loadRepoConfig(repoId);
-    const workflowRef = event.queryStringParameters?.workflow_ref;
     const workflow = workflowRef ? getWorkflowDescriptor(workflowRef) : undefined;
     const packRef = repoConfig?.event_rule_pack ?? workflow?.eventRulePack;
-    const resolved = resolveEventRules({
-      inlineRules: repoConfig?.event_rules,
-      packRef,
-    });
+    let resolved: EventRule[];
+    try {
+      resolved = resolveEventRules({
+        inlineRules: repoConfig?.event_rules,
+        packRef,
+      });
+    } catch (err) {
+      if (err instanceof UnknownEventRulePackError) {
+        return errorResponse(422, ErrorCode.VALIDATION_ERROR, err.message, requestId);
+      }
+      throw err;
+    }
 
     const response: GetEventRulesResponse = {
       repo_id: repoId,
@@ -138,7 +150,7 @@ export async function handler(event: APIGatewayProxyEvent): Promise<APIGatewayPr
       registry_packs: listBuiltinEventRulePacks(),
     };
 
-    cache.set(repoId, { response, expiresAt: Date.now() + CACHE_TTL_MS });
+    cache.set(cacheKey, { response, expiresAt: Date.now() + CACHE_TTL_MS });
     return successResponse(200, response, requestId);
   } catch (err) {
     logger.error('get-event-rules failed', {

cdk/src/handlers/shared/create-task-core.ts

@@ -31,7 +31,7 @@ import type { APIGatewayProxyResult } from 'aws-lambda';
 import { ulid } from 'ulid';
 import { isDegeneratePattern, parseApprovalScope } from './approval-scope';
 import { screenImage, screenTextFile, AttachmentScreeningError, type ScreeningConfig } from './attachment-screening';
-import { resolveEventRules } from './event-rule-pack-resolver';
+import { resolveEventRules, UnknownEventRulePackError } from './event-rule-pack-resolver';
 import { generateBranchName } from './gateway';
 import { estimateImageTokensFromBuffer } from './image-tokens';
 import { logger } from './logger';
@@ -96,6 +96,24 @@ function describeRequiredInputs(requiredInputs: { allOf?: readonly string[]; one
   return parts.length > 0 ? parts.join(', plus ') : 'a task specification';
 }
 
+/** 503 for a blueprint/workflow that pins an unresolvable event-rule-pack —
+ *  a platform misconfiguration, same class as an out-of-bounds approval cap. */
+function unknownPackResponse(repo: string | undefined, err: UnknownEventRulePackError, requestId: string): APIGatewayProxyResult {
+  logger.error('Blueprint misconfiguration — unknown event-rule-pack pin', {
+    repo,
+    pack_id: err.packRef.id,
+    pack_version: err.packRef.version,
+    request_id: requestId,
+  });
+  return errorResponse(
+    503,
+    ErrorCode.SERVICE_UNAVAILABLE,
+    `Blueprint misconfiguration: event-rule-pack '${err.packRef.id}@${err.packRef.version}' `
+      + `for '${repo}' does not resolve. Ask the platform admin to re-deploy the blueprint with a valid pack pin.`,
+    requestId,
+  );
+}
+
 /**
  * Core task creation logic shared by the Cognito create-task handler
  * and the webhook create-task handler.
@@ -212,19 +230,33 @@ export async function createTaskCore(
       resolvedApprovalGateCap = blueprintCap;
       capFromBlueprint = true;
     }
-    resolvedEventRules = resolveEventRules({
-      inlineRules: repoConfig?.event_rules,
-      packRef: repoConfig?.event_rule_pack ?? workflow.eventRulePack,
-    });
     const packRef = repoConfig?.event_rule_pack ?? workflow.eventRulePack;
+    try {
+      resolvedEventRules = resolveEventRules({
+        inlineRules: repoConfig?.event_rules,
+        packRef,
+      });
+    } catch (err) {
+      if (err instanceof UnknownEventRulePackError) {
+        return unknownPackResponse(body.repo, err, requestId);
+      }
+      throw err;
+    }
     if (packRef) {
       eventRulePackId = packRef.id;
       eventRulePackVersion = packRef.version;
     }
   } else {
-    resolvedEventRules = resolveEventRules({
-      packRef: workflow.eventRulePack,
-    });
+    try {
+      resolvedEventRules = resolveEventRules({
+        packRef: workflow.eventRulePack,
+      });
+    } catch (err) {
+      if (err instanceof UnknownEventRulePackError) {
+        return unknownPackResponse(body.repo, err, requestId);
+      }
+      throw err;
+    }
     if (workflow.eventRulePack) {
       eventRulePackId = workflow.eventRulePack.id;
       eventRulePackVersion = workflow.eventRulePack.version;