feat(agent-bff): audit reads and action executions

packages/agent-bff/src/action/action-routes-middleware.ts

@@ -4,6 +4,7 @@ import type {
   AgentActionClient,
   AgentActionClientOptions,
 } from './agent-action-client';
+import type { ActivityLogWriter } from '../activity-log/activity-log-writer';
 import type { AgentTransport } from '../agent/agent-transport';
 import type { Logger } from '../ports/logger-port';
 import type ReadModelStore from '../read-model/read-model-store';
@@ -28,7 +29,9 @@ import {
   requireAgentToken,
   resolveReadModel,
 } from '../http/agent-route-helpers';
+import { BffHttpError } from '../http/bff-http-error';
 import {
+  ACTION_REQUIRES_APPROVAL_TYPE,
   actionError,
   actionRequiresApproval,
   invalidRequest,
@@ -37,6 +40,18 @@ import {
 
 const ACTION_ROUTE = /^\/agent\/v1\/([^/]+)\/actions\/([^/]+)\/(form|execute)$/;
 
+const EXECUTE_VERB = 'execute';
+
+/**
+ * An approval request is a business outcome, not a failure: the action was routed for review. The
+ * BFF answers it with a 403, but recording the entry as `failed` would make the same event count
+ * differently here and in mcp-server, which records it as a success — and action-failure statistics
+ * would mix refusals with runs that never happened.
+ */
+function isApprovalRequest(error: unknown): boolean {
+  return error instanceof BffHttpError && error.type === ACTION_REQUIRES_APPROVAL_TYPE;
+}
+
 interface ActionRequestBody {
   recordIds?: unknown;
   values?: unknown;
@@ -94,6 +109,7 @@ export interface ActionRoutesMiddlewareOptions {
   store: ReadModelStore;
   transport: AgentTransport;
   logger: Logger;
+  activityLogs: ActivityLogWriter;
   createClient?: (options: AgentActionClientOptions) => AgentActionClient;
 }
 
@@ -190,6 +206,7 @@ export default function createActionRoutesMiddleware({
   store,
   transport,
   logger,
+  activityLogs,
   createClient = defaultCreateAgentActionClient,
 }: ActionRoutesMiddlewareOptions): Middleware {
   return async function actionRoutesMiddleware(ctx, next) {
@@ -209,12 +226,16 @@ export default function createActionRoutesMiddleware({
 
     // The read-model's action map IS the allow-list, so an absent action cannot be told from a
     // known-but-disallowed one — every non-exposed action maps to 404 here; `action_not_allowed`
-    // (403) has no local trigger, mirroring `collection_not_allowed`/`relation_not_allowed`. The
-    // URL identity is resolved before the body, so a bad action 404s before its payload is read.
+    // (403) has no local trigger, mirroring `collection_not_allowed`/`relation_not_allowed`.
     // TODO(PRD-673): distinguish disallowed from unknown when a separate exposure source exists.
-    if (!readModel.isActionAllowed(collection, actionName)) {
+    const allowed = readModel.isActionAllowed(collection, actionName);
+
+    const refuseUnknownAction = () => {
       throw unknownAction(`Unknown action: ${collection}.${actionName}`);
-    }
+    };
+
+    // The form is unaudited, so it keeps refusing before the body is read.
+    if (!allowed && verb !== EXECUTE_VERB) refuseUnknownAction();
 
     const body = (ctx.request.body ?? {}) as ActionRequestBody;
     assertKnownBodyKeys(body as Record<string, unknown>);
@@ -227,23 +248,49 @@ export default function createActionRoutesMiddleware({
       actionEndpoints: readModel.getActionEndpoints(),
     });
 
-    const action = await callAgent(
-      () =>
-        client.loadAction({
-          collection,
-          actionName,
-          recordIds,
-          timezone: ctx.state.timezone as string,
-        }),
-      logger,
-    );
-
-    const handlerArgs = { ctx, action, values, logger };
-
-    if (verb === 'execute') {
-      await handleExecute(handlerArgs);
-    } else {
-      await handleForm(handlerArgs);
+    const loadAction = () =>
+      callAgent(
+        () =>
+          client.loadAction({
+            collection,
+            actionName,
+            recordIds,
+            timezone: ctx.state.timezone as string,
+          }),
+        logger,
+      );
+
+    // The form is not audited, mirroring mcp-server, whose get-action-form tool writes no log
+    // either: the record-touching event the trail records is the execution.
+    if (verb !== EXECUTE_VERB) {
+      const action = await loadAction();
+
+      await handleForm({ ctx, action, values, logger });
+
+      return;
     }
+
+    // The whole sequence is audited, loadAction and setFields included, so the intent is recorded
+    // even when the attempt never reaches the agent's execute. The allow-list refusal is inside
+    // too, mirroring mcp-server, whose execute-action tool resolves the action within its own
+    // wrapper: an attempt on an action the caller may not trigger is exactly what the trail is
+    // for, and the record ids it names are only known once the body is read.
+    await activityLogs.record({
+      ctx,
+      action: 'action',
+      context: {
+        collectionName: collection,
+        recordIds,
+        label: `triggered the action "${actionName}"`,
+      },
+      isCompletedDespite: isApprovalRequest,
+      operation: async () => {
+        if (!allowed) refuseUnknownAction();
+
+        const action = await loadAction();
+
+        await handleExecute({ ctx, action, values, logger });
+      },
+    });
   };
 }

packages/agent-bff/src/activity-log/activity-log-drainer.ts

@@ -0,0 +1,73 @@
+interface InFlightOperation {
+  promise: Promise<unknown>;
+  /** What the drain names when a deadline leaves this one unfinished. Carries no record payload. */
+  description: string;
+}
+
+/**
+ * Holds the audited requests and the status transitions they fire without `await`. Nothing else
+ * keeps the transitions alive: `server.close()` waits for connections, and one sent after the
+ * response is attached to none — without this, every deploy would leave entries stuck in `pending`.
+ *
+ * The requests are tracked too, and not only their transitions, for the embedded deployment: there
+ * the host owns the connections, so `stop()` returns while requests are still running and their
+ * transitions are not registered yet.
+ */
+export default class ActivityLogDrainer {
+  private readonly inFlight = new Set<InFlightOperation>();
+
+  track<T>(operation: () => Promise<T>, description: string): Promise<T> {
+    const promise = operation();
+    const entry: InFlightOperation = { promise, description };
+    this.inFlight.add(entry);
+    promise.finally(() => this.inFlight.delete(entry)).catch(() => {});
+
+    return promise;
+  }
+
+  /**
+   * Loops rather than settling one snapshot: a transition is registered only once the request it
+   * audits has finished, so a single pass would return before the work that outlives it.
+   *
+   * `timeoutMs` is the shutdown deadline the caller shares: a stalled audit store would otherwise
+   * hold the process past the grace its orchestrator gives it, and be SIGKILLed mid-drain. Returns
+   * what the deadline left unfinished, empty when everything settled.
+   */
+  async drain(timeoutMs?: number): Promise<string[]> {
+    const deadline = timeoutMs === undefined ? undefined : Date.now() + timeoutMs;
+
+    while (this.inFlight.size > 0) {
+      const remainingMs = deadline === undefined ? undefined : deadline - Date.now();
+
+      if (remainingMs !== undefined && remainingMs <= 0) break;
+
+      // eslint-disable-next-line no-await-in-loop
+      await this.settle(remainingMs);
+    }
+
+    return [...this.inFlight].map(entry => entry.description);
+  }
+
+  private async settle(timeoutMs?: number): Promise<void> {
+    const settled = Promise.allSettled([...this.inFlight].map(entry => entry.promise));
+
+    if (timeoutMs === undefined) {
+      await settled;
+
+      return;
+    }
+
+    let timer: NodeJS.Timeout | undefined;
+
+    try {
+      await Promise.race([
+        settled,
+        new Promise<void>(resolve => {
+          timer = setTimeout(resolve, timeoutMs);
+        }),
+      ]);
+    } finally {
+      clearTimeout(timer);
+    }
+  }
+}

packages/agent-bff/src/activity-log/activity-log-writer.ts

@@ -0,0 +1,54 @@
+import type { ActivityLogContext, BffActivityLogAction } from './activity-logs-creator';
+import type { ActivityLogsWriter } from './activity-logs-service';
+import type { Logger } from '../ports/logger-port';
+import type { Context } from 'koa';
+
+import ActivityLogDrainer from './activity-log-drainer';
+import withActivityLog from './with-activity-log';
+
+export interface RecordActivityLogOptions<T> {
+  ctx: Context;
+  action: BffActivityLogAction;
+  context?: ActivityLogContext;
+  operation: () => Promise<T>;
+  isCompletedDespite?: (error: unknown) => boolean;
+}
+
+export interface ActivityLogWriter {
+  record<T>(options: RecordActivityLogOptions<T>): Promise<T>;
+  /**
+   * Waits for the audited requests still running and for the status transitions no connection
+   * holds. Called when the server stops, which shares its deadline through `timeoutMs`; returns
+   * the operations that deadline left unfinished.
+   */
+  drain(timeoutMs?: number): Promise<string[]>;
+}
+
+export interface ActivityLogWriterOptions {
+  service: ActivityLogsWriter;
+  logger: Logger;
+}
+
+function describeRequest(action: BffActivityLogAction, collectionName?: string): string {
+  return collectionName ? `'${action}' request on '${collectionName}'` : `'${action}' request`;
+}
+
+export default function createActivityLogWriter({
+  service,
+  logger,
+}: ActivityLogWriterOptions): ActivityLogWriter {
+  const drainer = new ActivityLogDrainer();
+
+  return {
+    record<T>(options: RecordActivityLogOptions<T>): Promise<T> {
+      return drainer.track(
+        () => withActivityLog({ ...options, service, drainer, logger }),
+        describeRequest(options.action, options.context?.collectionName),
+      );
+    },
+
+    drain(timeoutMs?: number): Promise<string[]> {
+      return drainer.drain(timeoutMs);
+    },
+  };
+}