Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 55 additions & 17 deletions packages/agent-bff/src/action/action-routes-middleware.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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,
Expand All @@ -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;
Expand Down Expand Up @@ -94,6 +109,7 @@ export interface ActionRoutesMiddlewareOptions {
store: ReadModelStore;
transport: AgentTransport;
logger: Logger;
activityLogs: ActivityLogWriter;
createClient?: (options: AgentActionClientOptions) => AgentActionClient;
}

Expand Down Expand Up @@ -181,6 +197,7 @@ export default function createActionRoutesMiddleware({
store,
transport,
logger,
activityLogs,
createClient = defaultCreateAgentActionClient,
}: ActionRoutesMiddlewareOptions): Middleware {
return async function actionRoutesMiddleware(ctx, next) {
Expand Down Expand Up @@ -218,23 +235,44 @@ 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.
await activityLogs.record({
ctx,
action: 'action',
context: {
collectionName: collection,
recordIds,
label: `triggered the action "${actionName}"`,
},
isCompletedDespite: isApprovalRequest,
operation: async () => {
const action = await loadAction();

await handleExecute({ ctx, action, values, logger });
},
});
};
}
73 changes: 73 additions & 0 deletions packages/agent-bff/src/activity-log/activity-log-drainer.ts
Original file line number Diff line number Diff line change
@@ -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);
}
}
}
54 changes: 54 additions & 0 deletions packages/agent-bff/src/activity-log/activity-log-writer.ts
Original file line number Diff line number Diff line change
@@ -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);
},
};
}
Loading
Loading