fix(event-governance): durable aggregate high-water marks; resolve rule-language question (#230)

Aggregate ceiling rules were unreliable:
- turn_count_gte rules never fired — the agent emitted `turn` but every
  evaluator reads `turn_count` (NaN → no match). Agent now also emits the
  cumulative aliases `turn_count` / `cumulative_cost_usd` that the rules read.
- cost/turn aggregates were reconstructed from the current event only, so the
  per-session SDK reset (PR-fix retries, container restart) undercounted them.
  evaluateAsyncEventRules now persists a monotonic high-water mark on the
  TaskRecord (climb-only conditional UpdateItem) and evaluates against it, so
  ceilings survive restarts. Fanout's ad-hoc reconstruction is removed.

Resolves the §10 rule-language open question: declarative field matchers are
permanent; Cedar-on-events is rejected (Cedar is authorization, has no
aggregation, and would add a third Cedar runtime for no gain).

ponytail: high-water is max(session totals), not a cross-session SUM — upgrade
path noted on the TaskRecord fields if PR-fix retries must accrue.

Author: bgagent

agent/src/progress_writer.py

@@ -612,6 +612,10 @@ 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),
@@ -686,9 +690,14 @@ 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

@@ -1271,24 +1271,12 @@ export const handler = async (
       try {
         if (EVENT_GOVERNANCE_ENABLED) {
           const taskForGov = await loadTaskForGovernance(ev.task_id);
-          const rawCost = ev.metadata?.cumulative_cost_usd ?? ev.metadata?.cost_usd;
-          const parsedCost = typeof rawCost === 'number' ? rawCost : Number(rawCost);
-          const rawTurns = ev.metadata?.turn_count;
-          const parsedTurns = typeof rawTurns === 'number' ? rawTurns : Number(rawTurns);
-          const aggregateState = ev.event_type === 'agent_cost_update'
-            || ev.event_type === 'agent_turn'
-            ? {
-              ...(ev.event_type === 'agent_cost_update' && {
-                cumulative_cost_usd: Number.isFinite(parsedCost) ? parsedCost : taskForGov?.cost_usd,
-              }),
-              ...(ev.event_type === 'agent_turn' && Number.isFinite(parsedTurns) && {
-                turn_count: parsedTurns,
-              }),
-            }
-            : undefined;
+          // Aggregate high-water tracking is owned by evaluateAsyncEventRules,
+          // which persists and resolves the durable marks (#230); the handler
+          // just forwards the event and task record.
           const govResult = await evaluateAsyncEventRules(
             { ...ev, metadata: ev.metadata ?? {} },
-            { task: taskForGov, aggregateState },
+            { task: taskForGov },
           );
           governanceForcedChannels = govResult.notifyChannels.filter(
             (ch): ch is NotificationChannel => ch in DISPATCHERS,

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

@@ -29,6 +29,7 @@ import {
   buildPolicyDecisionMetadata,
   matchEventRules,
   parseEventRules,
+  type AggregateState,
   type EvaluableEvent,
 } from './event-rule-evaluator';
 import { logger } from './logger';
@@ -186,6 +187,88 @@ function correlationId(event: EvaluableEvent & { event_id?: string }, rule: Even
   return `${base}:${rule.id}`;
 }
 
+/** Extract the incoming cumulative cost/turn from a cost/turn event's metadata. */
+function incomingAggregate(event: EvaluableEvent): { cost?: number; turns?: number } {
+  const meta = event.metadata;
+  const out: { cost?: number; turns?: number } = {};
+  if (event.event_type === 'agent_cost_update') {
+    const raw = meta.cumulative_cost_usd ?? meta.cost_usd;
+    const cost = typeof raw === 'number' ? raw : Number(raw);
+    if (Number.isFinite(cost)) out.cost = cost;
+  }
+  if (event.event_type === 'agent_turn' || event.event_type === 'agent_cost_update') {
+    const raw = meta.turn_count ?? meta.turn;
+    const turns = typeof raw === 'number' ? raw : Number(raw);
+    if (Number.isFinite(turns)) out.turns = turns;
+  }
+  return out;
+}
+
+/**
+ * Bump the durable high-water marks on the TaskRecord and return the effective
+ * aggregate state for rule evaluation (#230). Uses ``if_not_exists`` + a max
+ * ConditionExpression so the value only ever climbs — a stream retry or an
+ * out-of-order event can't lower it, and it survives the per-session SDK reset
+ * across container restarts / PR-fix retries. On any DDB error we fall back to
+ * the just-observed value so evaluation still proceeds (fail-open to the live
+ * reading, never below it).
+ */
+async function persistAndResolveAggregate(
+  taskId: string,
+  event: EvaluableEvent,
+  task: TaskRecord | undefined,
+): Promise<AggregateState | undefined> {
+  const incoming = incomingAggregate(event);
+  if (incoming.cost === undefined && incoming.turns === undefined) {
+    // Not a cost/turn event — use whatever is already persisted.
+    if (!task) return undefined;
+    return { cumulative_cost_usd: task.gov_cumulative_cost_usd, turn_count: task.gov_cumulative_turn_count };
+  }
+
+  const priorCost = task?.gov_cumulative_cost_usd ?? 0;
+  const priorTurns = task?.gov_cumulative_turn_count ?? 0;
+  const resolved: AggregateState = {
+    cumulative_cost_usd: incoming.cost !== undefined ? Math.max(priorCost, incoming.cost) : task?.gov_cumulative_cost_usd,
+    turn_count: incoming.turns !== undefined ? Math.max(priorTurns, incoming.turns) : task?.gov_cumulative_turn_count,
+  };
+
+  if (!TASK_TABLE) return resolved;
+
+  const sets: string[] = [];
+  const conds: string[] = [];
+  const values: Record<string, unknown> = {};
+  if (incoming.cost !== undefined) {
+    sets.push('gov_cumulative_cost_usd = :c');
+    conds.push('(attribute_not_exists(gov_cumulative_cost_usd) OR gov_cumulative_cost_usd < :c)');
+    values[':c'] = incoming.cost;
+  }
+  if (incoming.turns !== undefined) {
+    sets.push('gov_cumulative_turn_count = :t');
+    conds.push('(attribute_not_exists(gov_cumulative_turn_count) OR gov_cumulative_turn_count < :t)');
+    values[':t'] = incoming.turns;
+  }
+  try {
+    await ddb.send(new UpdateCommand({
+      TableName: TASK_TABLE,
+      Key: { task_id: taskId },
+      UpdateExpression: `SET ${sets.join(', ')}`,
+      // Only write when at least one mark climbs; a no-op (retry with equal or
+      // lower value) fails the condition and is silently skipped.
+      ConditionExpression: `attribute_exists(task_id) AND (${conds.join(' OR ')})`,
+      ExpressionAttributeValues: values,
+    }));
+  } catch (err) {
+    const name = (err as { name?: string })?.name;
+    if (name !== 'ConditionalCheckFailedException') {
+      logger.warn('[event-governance] aggregate high-water update failed', {
+        task_id: taskId,
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
+  }
+  return resolved;
+}
+
 function idempotencyEventId(taskId: string, ruleId: string, corr: string): string {
   return `gov-idem#${taskId}#${ruleId}#${corr}`;
 }
@@ -237,9 +320,15 @@ export async function evaluateAsyncEventRules(
   const rules = parseEventRules(task?.event_rules);
   if (rules.length === 0) return { notifyChannels: [], forceFanOut: false };
 
+  // Resolve aggregates against the durable high-water mark so ceiling rules
+  // survive the per-session SDK cost/turn reset (#230). ``ctx.aggregateState``
+  // (the caller's live reading) is the fallback when no durable value applies.
+  const aggregateState = await persistAndResolveAggregate(event.task_id, event, task)
+    ?? ctx.aggregateState;
+
   const matched = matchEventRules(rules, event, {
     evaluation: 'async',
-    aggregateState: ctx.aggregateState,
+    aggregateState,
   });
   const notifyChannels: string[] = [];
   let forceFanOut = false;

cdk/src/handlers/shared/types.ts

@@ -121,6 +121,18 @@ export interface TaskRecord {
   readonly started_at?: string;
   readonly completed_at?: string;
   readonly cost_usd?: number;
+  /**
+   * Durable high-water marks for event-governance aggregate rules (#230).
+   * The FanOut stream consumer bumps these monotonically as ``agent_cost_update``
+   * / ``agent_turn`` events flow, so ``cost_usd_gte`` / ``turn_count_gte`` rules
+   * survive container restarts (the per-session SDK total resets; these do not).
+   * ponytail: monotonic max of session totals, not a cross-session SUM — a
+   * multi-session task's ceiling reflects its largest single session. Sum
+   * accounting (seed baseline from here on the agent restart path) is the
+   * upgrade if PR-fix retries must accrue.
+   */
+  readonly gov_cumulative_cost_usd?: number;
+  readonly gov_cumulative_turn_count?: number;
   readonly duration_s?: number;
   readonly build_passed?: boolean;
   /** Whether the post-run lint gate passed (#515). Written with `build_passed`

cdk/test/handlers/shared/event-governance-async-module.test.ts

@@ -84,7 +84,84 @@ describe('event-governance-async module', () => {
     expect(result1.forceFanOut).toBe(true);
   });
 
-  test('evaluateAsyncEventRules executes cancel_task action in enforce mode', async () => {
+  test('bumps durable cost high-water mark and evaluates against it', async () => {
+    // No cumulative value on the event itself; the ceiling must be met from
+    // the persisted high-water mark alone (the cross-restart case).
+    mockSend
+      .mockResolvedValueOnce({}) // high-water UpdateCommand
+      .mockResolvedValueOnce({}) // idempotency claim
+      .mockResolvedValueOnce({}) // Put policy_decision
+      .mockResolvedValueOnce({}) // Update cancel
+      .mockResolvedValueOnce({}); // Put task_cancelled
+    const mod = await import('../../../src/handlers/shared/event-governance-async');
+    await mod.evaluateAsyncEventRules(
+      {
+        task_id: 't3',
+        event_id: 'e3',
+        event_type: 'agent_cost_update',
+        metadata: { cost_usd: 5 }, // this session is cheap...
+      },
+      {
+        task: {
+          task_id: 't3',
+          status: 'RUNNING',
+          gov_cumulative_cost_usd: 40, // ...but a prior session already crossed
+          event_rules: [
+            {
+              id: 'cap',
+              on: 'agent_cost_update',
+              when: { aggregate: { cost_usd_gte: 25 } },
+              action: 'cancel_task',
+              mode: 'enforce',
+              evaluation: 'async',
+            },
+          ],
+        } as any,
+      },
+    );
+    // First send is the high-water UpdateCommand with a climb-only condition.
+    const firstCall = mockSend.mock.calls[0][0];
+    expect(firstCall._type).toBe('Update');
+    expect(firstCall.input.ConditionExpression).toContain('gov_cumulative_cost_usd < :c');
+    // The cancel fired — resolved aggregate (max(40, 5)=40) still ≥ 25.
+    const cancelUpdate = mockSend.mock.calls.find(
+      (c) => c[0]._type === 'Update' && String(c[0].input.UpdateExpression).includes('#status'),
+    );
+    expect(cancelUpdate).toBeDefined();
+  });
+
+  test('turn_count_gte rule fires on turn_count metadata', async () => {
+    mockSend.mockResolvedValue({});
+    const mod = await import('../../../src/handlers/shared/event-governance-async');
+    const result = await mod.evaluateAsyncEventRules(
+      {
+        task_id: 't4',
+        event_id: 'e4',
+        event_type: 'agent_turn',
+        metadata: { turn_count: 35 },
+      },
+      {
+        task: {
+          task_id: 't4',
+          status: 'RUNNING',
+          event_rules: [
+            {
+              id: 'turns',
+              on: 'agent_turn',
+              when: { aggregate: { turn_count_gte: 30 } },
+              action: 'escalate',
+              mode: 'enforce',
+              evaluation: 'async',
+              notify_channels: ['slack'],
+            },
+          ],
+        } as any,
+      },
+    );
+    expect(result.notifyChannels).toContain('slack');
+  });
+
+  test('executes cancel_task action in enforce mode', async () => {
     mockSend
       .mockResolvedValueOnce({}) // idempotency
       .mockResolvedValueOnce({}) // Put policy_decision

docs/design/EVENT_GOVERNANCE.md

@@ -152,6 +152,7 @@ Async `require_approval` after `pr_created` must state that the PR already exist
 ## 9. Out of scope
 
 - Replacing tool Cedar with event rules
+- Cedar-on-events (rejected — see §10; declarative matchers are permanent)
 - Stream-only HITL (race with fast agents)
 - EventBridge as primary internal bus
 - Separate approve commands for event vs tool gates
@@ -162,6 +163,6 @@ Async `require_approval` after `pr_created` must state that the PR already exist
 
 | Question | Resolution |
 |----------|------------|
-| Rule language | Declarative field matchers (Phase 0–1); Cedar-on-events deferred to ADR |
+| Rule language | **Resolved: declarative field matchers are the permanent language; Cedar-on-events is rejected.** Cedar is an *authorization* language (`principal-action-resource-context` → permit/forbid) with no aggregation — it cannot express `cost_usd_gte` / `turn_count_gte`, and event actions (`notify`/`escalate`/`cancel_task`/`inject_nudge`) are ECA automation, not authorization decisions. A Cedar port would still need a bespoke action layer, and it would add a third Cedar runtime alongside the two we already must bump in lockstep (`cedar-wasm`, `cedarpy`) for zero gain. The two-plane split stands: Cedar governs tool execution (fail-closed); the declarative matcher governs events. The matcher already has cross-language parity, schema validation, and fixture parity tests. |
 | Checkpoint trust | Pipeline-emitted only |
 | Scope algebra (tool + event overlap) | Idempotency key; document in runbooks |

docs/src/content/docs/architecture/Event-governance.md

@@ -156,6 +156,7 @@ Async `require_approval` after `pr_created` must state that the PR already exist
 ## 9. Out of scope
 
 - Replacing tool Cedar with event rules
+- Cedar-on-events (rejected — see §10; declarative matchers are permanent)
 - Stream-only HITL (race with fast agents)
 - EventBridge as primary internal bus
 - Separate approve commands for event vs tool gates
@@ -166,6 +167,6 @@ Async `require_approval` after `pr_created` must state that the PR already exist
 
 | Question | Resolution |
 |----------|------------|
-| Rule language | Declarative field matchers (Phase 0–1); Cedar-on-events deferred to ADR |
+| Rule language | **Resolved: declarative field matchers are the permanent language; Cedar-on-events is rejected.** Cedar is an *authorization* language (`principal-action-resource-context` → permit/forbid) with no aggregation — it cannot express `cost_usd_gte` / `turn_count_gte`, and event actions (`notify`/`escalate`/`cancel_task`/`inject_nudge`) are ECA automation, not authorization decisions. A Cedar port would still need a bespoke action layer, and it would add a third Cedar runtime alongside the two we already must bump in lockstep (`cedar-wasm`, `cedarpy`) for zero gain. The two-plane split stands: Cedar governs tool execution (fail-closed); the declarative matcher governs events. The matcher already has cross-language parity, schema validation, and fixture parity tests. |
 | Checkpoint trust | Pipeline-emitted only |
 | Scope algebra (tool + event overlap) | Idempotency key; document in runbooks |