diff --git a/infra/migrations/versions/0045_e2_campaign_approval_operations.py b/infra/migrations/versions/0045_e2_campaign_approval_operations.py new file mode 100644 index 00000000..da5540d3 --- /dev/null +++ b/infra/migrations/versions/0045_e2_campaign_approval_operations.py @@ -0,0 +1,36 @@ +"""Allow bounded E2 campaign approvals in the durable operation ledger.""" + +from __future__ import annotations + +from alembic import op + +revision: str = "0045" +down_revision: str | None = "0044" +branch_labels: str | tuple[str, ...] | None = None +depends_on: str | tuple[str, ...] | None = None + + +def upgrade() -> None: + """Permit both E1 runs and E2 campaigns in the shared approval-operation ledger.""" + op.execute( + """SET LOCAL lock_timeout = '10s'; +ALTER TABLE evolution_approval_operations +DROP CONSTRAINT evolution_approval_operations_tool_name_check; +ALTER TABLE evolution_approval_operations +ADD CONSTRAINT evolution_approval_operations_tool_name_check +CHECK (tool_name IN ('evolver.run_evolution','evolver.run_event_campaign'));""" + ) + + +def downgrade() -> None: + """Remove E2 recovery rows before restoring the E1-only ledger constraint.""" + op.execute( + """SET LOCAL lock_timeout = '10s'; +DELETE FROM evolution_approval_operations +WHERE tool_name='evolver.run_event_campaign'; +ALTER TABLE evolution_approval_operations +DROP CONSTRAINT evolution_approval_operations_tool_name_check; +ALTER TABLE evolution_approval_operations +ADD CONSTRAINT evolution_approval_operations_tool_name_check +CHECK (tool_name='evolver.run_evolution');""" + ) diff --git a/packages/orchestration/config/permissions.default.yaml b/packages/orchestration/config/permissions.default.yaml index 4e33f114..7dbc42a6 100644 --- a/packages/orchestration/config/permissions.default.yaml +++ b/packages/orchestration/config/permissions.default.yaml @@ -43,9 +43,8 @@ allow: # 演化状态与候选详情只读 - "evolver.get_evolution" - "evolver.get_candidate" - # 事件 campaign 仅在独立 sandbox 中自动迭代；不会 promote、启动或下单 + # 事件 campaign 查询只读；启动 campaign 会产生 LLM 费用，必须先显式审批 - "evolver.get_event_campaign" - - "evolver.run_event_campaign" # Swarm 批量回测（ADR-0025）：只读，无下单路径 - "swarm.*" @@ -93,8 +92,9 @@ ask: - "paper.deposit_cash" - "paper.reset_account" - # LLM 变异会产生费用；取消会改变运行状态，均需明确确认 + # LLM 演化会产生费用；E2 一次审批覆盖整个五代 campaign，不逐代重复审批 - "evolver.run_evolution" + - "evolver.run_event_campaign" - "evolver.abort_evolution" deny: diff --git a/packages/orchestration/src/clients/evolver.ts b/packages/orchestration/src/clients/evolver.ts index 9e61fba5..3647ca90 100644 --- a/packages/orchestration/src/clients/evolver.ts +++ b/packages/orchestration/src/clients/evolver.ts @@ -263,10 +263,21 @@ export class EvolverClient { options.request, headers, ); - return await this.http.post( - `/api/v1/campaigns/${created.campaign_id}/start`, - {}, - ); + if (created.status !== "draft") return created; + + try { + return await this.http.post( + `/api/v1/campaigns/${created.campaign_id}/start`, + {}, + ); + } catch (error) { + if (!(error instanceof HttpClientError) || error.code !== "CAMPAIGN_STATE_CONFLICT") { + throw error; + } + const current = await this.getEventCampaign(created.campaign_id); + if (current.status === "draft") throw error; + return current; + } } async getEventCampaign(campaignId: string): Promise { diff --git a/packages/orchestration/src/hooks/with-hooks.ts b/packages/orchestration/src/hooks/with-hooks.ts index e26d2e99..ca71d470 100644 --- a/packages/orchestration/src/hooks/with-hooks.ts +++ b/packages/orchestration/src/hooks/with-hooks.ts @@ -54,6 +54,12 @@ type GenericTool = { [key: string]: unknown; }; +const DURABLE_EVOLUTION_APPROVAL_TOOLS = new Set([ + "evolver.run_evolution", + "evolver.run_event_campaign", +]); +const E2_CAMPAIGN_RETRY_WINDOW_MS = 2 * 60 * 1_000; + /** * mastra ``server.middleware`` 从 Bearer JWT 解出的已认证主体（sub）写进 RequestContext * 的 key（#91）。getSessionId 最高优先读它 → askCache 按已认证主体 scope（替代 __global__）。 @@ -201,14 +207,14 @@ export function withHooks(tool: T, opts: WithHooksOptions if (permDecision === "ask") { const store = opts.pendingApprovals ?? defaultPendingApprovals; const projectedInput = projectApprovalInput(toolName, effectiveInput); - const llmSnapshot = - toolName === "evolver.run_evolution" - ? getRequestContextValue(ctx, USER_LLM_SNAPSHOT_KEY) - : undefined; + const durableEvolutionApproval = DURABLE_EVOLUTION_APPROVAL_TOOLS.has(toolName); + const llmSnapshot = durableEvolutionApproval + ? getRequestContextValue(ctx, USER_LLM_SNAPSHOT_KEY) + : undefined; const approvalInput = llmSnapshot ? { request: projectedInput, llm_snapshot: llmSnapshot } : projectedInput; - if (!authSub || !sessionId || (toolName === "evolver.run_evolution" && !llmSnapshot)) { + if (!authSub || !sessionId || (durableEvolutionApproval && !llmSnapshot)) { return { isError: true, deniedBy: "permission-ask", @@ -228,6 +234,10 @@ export function withHooks(tool: T, opts: WithHooksOptions toolName, approvalInput, reuseAfterConsume: toolName === "evolver.run_evolution", + reuseOnceAfterConsumeMs: + toolName === "evolver.run_event_campaign" + ? E2_CAMPAIGN_RETRY_WINDOW_MS + : undefined, }); if (!operationId) { const approvalViewInput = llmSnapshot @@ -242,7 +252,7 @@ export function withHooks(tool: T, opts: WithHooksOptions timeoutMs: opts.askTimeoutMs && opts.askTimeoutMs > 0 ? opts.askTimeoutMs - : toolName === "evolver.run_evolution" + : durableEvolutionApproval ? 300_000 : undefined, }); diff --git a/packages/orchestration/src/permissions/approval-identity.ts b/packages/orchestration/src/permissions/approval-identity.ts index 2d187455..ae006759 100644 --- a/packages/orchestration/src/permissions/approval-identity.ts +++ b/packages/orchestration/src/permissions/approval-identity.ts @@ -58,6 +58,8 @@ export const APPROVAL_IDENTITY_FIELDS: Readonly; + oneShot?: boolean; } const DEFAULT_TIMEOUT_MS = 30_000; @@ -64,13 +67,19 @@ export interface ApprovalPersistence { decision: PendingDecision, via: "user" | "timeout", ): Promise; - rememberEvolutionOperation(args: EvolutionOperationScope & { operationId: string }): Promise<{ + rememberEvolutionOperation( + args: EvolutionOperationScope & { operationId: string; retentionMs?: number }, + ): Promise<{ expiresAt: string; } | undefined>; findEvolutionOperation(args: EvolutionOperationScope): Promise<{ operationId: string; expiresAt: string; } | undefined>; + claimEvolutionOperation?(args: EvolutionOperationScope): Promise<{ + operationId: string; + expiresAt: string; + } | undefined>; } export interface EvolutionOperationScope { @@ -90,6 +99,7 @@ export class PendingApprovalsStore { private readonly records = new Map(); private readonly identityIndex = new Map(); private readonly consumedByIdentity = new Map(); + private readonly consumingByIdentity = new Map>(); private readonly telemetry: PendingTelemetrySink; private readonly persistence?: ApprovalPersistence; @@ -170,36 +180,97 @@ export class PendingApprovalsStore { /** Atomically consumes one approved decision and returns its restart-stable operation ID. */ async consumeApproved(args: PendingConsumeArgs): Promise { const identity = this.identityFor(args); - const consumed = args.reuseAfterConsume ? this.consumedByIdentity.get(identity) : undefined; - if (consumed) { - if (Date.now() >= Date.parse(consumed.expiresAt)) { - this.removeConsumed(identity); - } else { - this.telemetry({ - event: "ask_approval_operation_reused", - requestId: consumed.operationId, - toolName: args.toolName, - sessionId: args.sessionId, - authSub: args.authSub, - ts: new Date().toISOString(), - }); - return consumed.operationId; + const inFlight = this.consumingByIdentity.get(identity); + if (inFlight) { + await inFlight; + return await this.consumeApproved(args); + } + + const run = this.consumeApprovedUnlocked(args, identity); + this.consumingByIdentity.set(identity, run); + try { + return await run; + } finally { + if (this.consumingByIdentity.get(identity) === run) { + this.consumingByIdentity.delete(identity); } } + } + + private async consumeApprovedUnlocked( + args: PendingConsumeArgs, + identity: string, + ): Promise { const scope = this.operationScope(args); - if (args.reuseAfterConsume && this.persistence) { - const persisted = await this.persistence.findEvolutionOperation(scope); - if (persisted && Date.now() < Date.parse(persisted.expiresAt)) { - this.cacheConsumed(identity, persisted); - this.telemetry({ - event: "ask_approval_operation_recovered", - requestId: persisted.operationId, - toolName: args.toolName, - sessionId: args.sessionId, - authSub: args.authSub, - ts: new Date().toISOString(), - }); - return persisted.operationId; + const boundedRetryMs = + args.reuseOnceAfterConsumeMs && args.reuseOnceAfterConsumeMs > 0 + ? args.reuseOnceAfterConsumeMs + : undefined; + + if (boundedRetryMs !== undefined) { + const consumed = this.consumedByIdentity.get(identity); + if (consumed) { + if (Date.now() >= Date.parse(consumed.expiresAt)) { + this.removeConsumed(identity); + } else if (consumed.oneShot) { + const operationId = consumed.operationId; + this.removeConsumed(identity); + this.telemetry({ + event: "ask_approval_operation_reused", + requestId: operationId, + toolName: args.toolName, + sessionId: args.sessionId, + authSub: args.authSub, + ts: new Date().toISOString(), + }); + return operationId; + } + } + if (this.persistence?.claimEvolutionOperation) { + const claimed = await this.persistence.claimEvolutionOperation(scope); + if (claimed && Date.now() < Date.parse(claimed.expiresAt)) { + this.telemetry({ + event: "ask_approval_operation_recovered", + requestId: claimed.operationId, + toolName: args.toolName, + sessionId: args.sessionId, + authSub: args.authSub, + ts: new Date().toISOString(), + }); + return claimed.operationId; + } + } + } else if (args.reuseAfterConsume) { + const consumed = this.consumedByIdentity.get(identity); + if (consumed) { + if (Date.now() >= Date.parse(consumed.expiresAt)) { + this.removeConsumed(identity); + } else { + this.telemetry({ + event: "ask_approval_operation_reused", + requestId: consumed.operationId, + toolName: args.toolName, + sessionId: args.sessionId, + authSub: args.authSub, + ts: new Date().toISOString(), + }); + return consumed.operationId; + } + } + if (this.persistence) { + const persisted = await this.persistence.findEvolutionOperation(scope); + if (persisted && Date.now() < Date.parse(persisted.expiresAt)) { + this.cacheConsumed(identity, persisted); + this.telemetry({ + event: "ask_approval_operation_recovered", + requestId: persisted.operationId, + toolName: args.toolName, + sessionId: args.sessionId, + authSub: args.authSub, + ts: new Date().toISOString(), + }); + return persisted.operationId; + } } } const requestId = this.identityIndex.get(identity); @@ -209,19 +280,29 @@ export class PendingApprovalsStore { this.expire(record.requestId); return undefined; } - let reusable: { operationId: string; expiresAt: string } | undefined; - if (args.reuseAfterConsume) { + let reusable: { operationId: string; expiresAt: string; oneShot?: boolean } | undefined; + const shouldReuse = args.reuseAfterConsume || boundedRetryMs !== undefined; + let persistedReusable = false; + if (shouldReuse) { + const retentionMs = boundedRetryMs ?? EVOLUTION_OPERATION_RETENTION_MS; const fallback = { operationId: record.requestId, - expiresAt: new Date(Date.now() + EVOLUTION_OPERATION_RETENTION_MS).toISOString(), + expiresAt: new Date(Date.now() + retentionMs).toISOString(), + oneShot: boundedRetryMs !== undefined, }; try { const persisted = await this.persistence?.rememberEvolutionOperation({ ...scope, operationId: record.requestId, + retentionMs, }); + persistedReusable = Boolean(persisted); reusable = persisted - ? { operationId: record.requestId, expiresAt: persisted.expiresAt } + ? { + operationId: record.requestId, + expiresAt: persisted.expiresAt, + oneShot: boundedRetryMs !== undefined, + } : fallback; } catch (error) { this.telemetry({ @@ -237,7 +318,9 @@ export class PendingApprovalsStore { } } this.remove(record); - if (reusable) this.cacheConsumed(identity, reusable); + if (reusable && !(boundedRetryMs !== undefined && persistedReusable)) { + this.cacheConsumed(identity, reusable); + } this.telemetry({ event: "ask_approval_consumed", requestId: record.requestId, @@ -303,7 +386,7 @@ export class PendingApprovalsStore { private cacheConsumed( identity: string, - record: { operationId: string; expiresAt: string }, + record: { operationId: string; expiresAt: string; oneShot?: boolean }, ): void { const timer = setTimeout( () => this.removeConsumed(identity), @@ -373,4 +456,5 @@ export const pendingApprovals = new PendingApprovalsStore(undefined, { markResolved, rememberEvolutionOperation, findEvolutionOperation, + claimEvolutionOperation, }); diff --git a/packages/orchestration/src/permissions/repo.ts b/packages/orchestration/src/permissions/repo.ts index 53de7565..82c70305 100644 --- a/packages/orchestration/src/permissions/repo.ts +++ b/packages/orchestration/src/permissions/repo.ts @@ -141,11 +141,14 @@ export async function markResolved( /** Persist one approved evolution identity before the costful tool is allowed to execute. */ export async function rememberEvolutionOperation( - args: EvolutionOperationScope & { operationId: string }, + args: EvolutionOperationScope & { operationId: string; retentionMs?: number }, ): Promise<{ expiresAt: string } | undefined> { const pool = getPoolOrNull(); if (!pool) return undefined; - const expiresAt = new Date(Date.now() + 24 * 60 * 60 * 1_000); + const retentionMs = args.retentionMs && args.retentionMs > 0 + ? args.retentionMs + : 24 * 60 * 60 * 1_000; + const expiresAt = new Date(Date.now() + retentionMs); const result = await pool.query( `INSERT INTO evolution_approval_operations (operation_id,auth_sub,session_id,tool_name,input_digest,approved_at,expires_at) @@ -186,6 +189,25 @@ export async function findEvolutionOperation( : undefined; } +/** Atomically consume one unexpired recovery entitlement. */ +export async function claimEvolutionOperation( + args: EvolutionOperationScope, +): Promise<{ operationId: string; expiresAt: string } | undefined> { + const pool = getPoolOrNull(); + if (!pool) return undefined; + const result = await pool.query( + `DELETE FROM evolution_approval_operations + WHERE auth_sub=$1 AND session_id=$2 AND tool_name=$3 AND input_digest=$4 + AND expires_at>NOW() + RETURNING operation_id,expires_at`, + [args.authSub, args.sessionId, args.toolName, args.inputDigest], + ); + const row = result.rows[0]; + return row + ? { operationId: String(row.operation_id), expiresAt: toIso(row.expires_at) } + : undefined; +} + /** * 启动扫尾：上一进程遗留的 pending 行批量置 expired_restart。 * 返回扫掉的行数（log / 测试用）；DB 不可用返回 0。 diff --git a/packages/orchestration/src/tools/evolver-shared.ts b/packages/orchestration/src/tools/evolver-shared.ts index 76032b5c..2d83141c 100644 --- a/packages/orchestration/src/tools/evolver-shared.ts +++ b/packages/orchestration/src/tools/evolver-shared.ts @@ -1,7 +1,5 @@ /** Evolver Mastra tools 的共享 schema 与客户端解析。 */ import { z } from "zod"; -import { randomUUID } from "node:crypto"; - import { resolveRequestToken } from "../auth.js"; import { buildEvolutionStartRequest, @@ -71,8 +69,8 @@ export async function getApprovedEvolutionRunContext( }; } -/** Build a credential-bound automatic campaign without a per-generation approval pause. */ -export async function getAutomaticEventCampaignContext( +/** Build one approved campaign context; its internal five generations remain automatic. */ +export async function getApprovedEventCampaignContext( input: { eventSnapshotId: string; sourceRunId?: string; @@ -80,15 +78,18 @@ export async function getAutomaticEventCampaignContext( }, ctx?: ToolRequestContext, ) { + const operationId = getRequestContextValue( + { requestContext: ctx }, + APPROVAL_OPERATION_ID_KEY, + ); const llmSnapshot = getRequestContextValue( { requestContext: ctx }, USER_LLM_SNAPSHOT_KEY, ); const authSub = ctx?.get?.(AUTH_SUB_KEY); - if (!llmSnapshot || typeof authSub !== "string" || !authSub) { - throw new Error("event campaign requires a verified owner and frozen LLM configuration"); + if (!operationId || !llmSnapshot || typeof authSub !== "string" || !authSub) { + throw new Error("explicit event campaign approval context is missing"); } - const operationId = randomUUID(); const request = buildEventCampaignRequest({ eventSnapshotId: input.eventSnapshotId, sourceRunId: input.sourceRunId, diff --git a/packages/orchestration/src/tools/evolver.ts b/packages/orchestration/src/tools/evolver.ts index 932df56f..0c5355ce 100644 --- a/packages/orchestration/src/tools/evolver.ts +++ b/packages/orchestration/src/tools/evolver.ts @@ -5,7 +5,7 @@ import { z } from "zod"; import { evolutionConfigSchema, eventCampaignConfigSchema, - getAutomaticEventCampaignContext, + getApprovedEventCampaignContext, getApprovedEvolutionRunContext, getEvolverClient, type ToolRequestContext, @@ -17,7 +17,7 @@ export const evolverRunEventCampaignTool = createTool({ 基于冻结事件事实与模拟盘反馈启动五代双层自动演化；每代两个 Agent proposer 产生八个假设，每个确定性展开三条实现，并锁定唯一冠军等待独立 Forward。 何时用：用户要求从事件机制发散新策略方向，且已有 point-in-time event snapshot 与 15m/1h/4h 冻结行情。 何时不用：只优化既有代码用 evolver.run_evolution；没有事件快照、需要实盘或希望自动下单时不要用。 -坑：模拟研究阶段自动迭代不逐代审批，但只产出 sandbox 候选；不会 promote、启动 Runner 或下单，Forward 与一次性 holdout 仍是硬门禁。 +坑：启动整个五代 campaign 需要一次显式审批；审批后内部自动迭代不逐代审批，只产出 sandbox 候选；不会 promote、启动 Runner 或下单，Forward 与一次性 holdout 仍是硬门禁。 `.trim(), inputSchema: z.object({ eventSnapshotId: z.string().uuid(), @@ -25,14 +25,14 @@ export const evolverRunEventCampaignTool = createTool({ config: eventCampaignConfigSchema, }), execute: async (inputData, ctx) => { - const automatic = await getAutomaticEventCampaignContext( + const approved = await getApprovedEventCampaignContext( inputData, ctx?.requestContext as ToolRequestContext | undefined, ); - return await automatic.client.startEventCampaign({ - request: automatic.request, - idempotencyKey: automatic.operationId, - credentialGrant: automatic.credentialGrant, + return await approved.client.startEventCampaign({ + request: approved.request, + idempotencyKey: approved.operationId, + credentialGrant: approved.credentialGrant, }); }, }); diff --git a/packages/orchestration/tests/e2-campaign-authorization.test.ts b/packages/orchestration/tests/e2-campaign-authorization.test.ts new file mode 100644 index 00000000..22258933 --- /dev/null +++ b/packages/orchestration/tests/e2-campaign-authorization.test.ts @@ -0,0 +1,220 @@ +import { describe, expect, it, vi } from "vitest"; + +import { AUTH_SUB_KEY, HookRunner, withHooks } from "../src/hooks/index.js"; +import { + APPROVAL_OPERATION_ID_KEY, + buildEvolutionLLMSnapshot, + USER_LLM_SNAPSHOT_KEY, +} from "../src/mastra/llm/evolution-snapshot.js"; +import { DEFAULT_PERMISSIONS, PermissionEngine } from "../src/permissions/index.js"; +import { PendingApprovalsStore } from "../src/permissions/pending.js"; +import { getApprovedEventCampaignContext } from "../src/tools/evolver-shared.js"; + +const snapshotA = buildEvolutionLLMSnapshot({ + id: "config-a", + provider: "deepseek", + model: "deepseek-v4-pro", + api_key: "not-forwarded-a", +}); + +const snapshotB = buildEvolutionLLMSnapshot({ + id: "config-b", + provider: "deepseek", + model: "deepseek-v4-pro", + api_key: "not-forwarded-b", +}); + +const campaignInput = { + eventSnapshotId: "11111111-1111-4111-8111-111111111111", + sourceRunId: "22222222-2222-4222-8222-222222222222", + config: { + venue: "binance", + symbol: "BTCUSDT", + timeframe: "1h" as const, + from_ts: "2026-08-01T00:00:00Z", + as_of: "2026-08-02T00:00:00Z", + initial_cash: 10_000, + fee_rate: 0.001, + trading_mode: "perp" as const, + leverage: 1, + random_seed: 7, + }, +}; + +type ToolCtx = { requestContext: Map }; + +function context(snapshot = snapshotA): ToolCtx { + return { + requestContext: new Map([[USER_LLM_SNAPSHOT_KEY, snapshot]]), + }; +} + +function makeApprovedTool(options?: { + store?: PendingApprovalsStore; + owner?: () => string | undefined; +}) { + const store = options?.store ?? new PendingApprovalsStore(); + const execute = vi.fn(async (_input: unknown, ctx?: unknown) => { + const requestContext = (ctx as ToolCtx).requestContext; + return { operationId: requestContext.get(APPROVAL_OPERATION_ID_KEY) }; + }); + const wrapped = withHooks( + { id: "evolver.run_event_campaign", execute }, + { + runner: new HookRunner(), + permissionResolver: () => "ask", + pendingApprovals: store, + getAuthSub: options?.owner ?? (() => "user:alice"), + getSessionId: () => "thread-e2", + }, + ); + return { store, execute, wrapped }; +} + +describe("E2 campaign authorization", () => { + it("requires ask permission instead of the automatic allow path", () => { + const engine = new PermissionEngine(DEFAULT_PERMISSIONS); + expect(engine.authorize("evolver.run_event_campaign", campaignInput).decision).toBe("ask"); + expect(engine.authorize("evolver.get_event_campaign", {}).decision).toBe("allow"); + }); + + it("rejects direct campaign-context construction without a trusted approval operation", async () => { + const requestContext = new Map([ + [AUTH_SUB_KEY, "user:alice"], + [USER_LLM_SNAPSHOT_KEY, snapshotA], + ]); + + await expect( + getApprovedEventCampaignContext(campaignInput, requestContext), + ).rejects.toThrow("explicit event campaign approval context is missing"); + }); + + it("fails closed when the frozen LLM approval context is missing", async () => { + const { store, execute, wrapped } = makeApprovedTool(); + const result = (await wrapped.execute!(campaignInput, { + requestContext: new Map(), + })) as { requiresApproval: boolean; message: string }; + + expect(result.requiresApproval).toBe(true); + expect(result.message).toContain("APPROVAL_UNAVAILABLE"); + expect(execute).not.toHaveBeenCalled(); + expect(store.list("user:alice")).toHaveLength(0); + store.clearAll(); + }); + + it("allows exactly one matching compensation retry for the approved E2 operation", async () => { + const { store, execute, wrapped } = makeApprovedTool(); + const ctx = context(); + + const pending = (await wrapped.execute!(campaignInput, ctx)) as { + requiresApproval: boolean; + requestId: string; + }; + expect(pending.requiresApproval).toBe(true); + expect(execute).not.toHaveBeenCalled(); + expect(store.respond(pending.requestId, "allow", "user:alice")).toBe(true); + + const first = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + const retry = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + const exhausted = (await wrapped.execute!(campaignInput, ctx)) as { + requiresApproval: boolean; + requestId: string; + }; + + expect(first.operationId).toBe(pending.requestId); + expect(retry.operationId).toBe(pending.requestId); + expect(exhausted.requiresApproval).toBe(true); + expect(exhausted.requestId).not.toBe(pending.requestId); + expect(execute).toHaveBeenCalledTimes(2); + store.clearAll(); + }); + + it("expires the E2 compensation retry after two minutes", async () => { + vi.useFakeTimers(); + const { store, execute, wrapped } = makeApprovedTool(); + const ctx = context(); + + const pending = (await wrapped.execute!(campaignInput, ctx)) as { requestId: string }; + expect(store.respond(pending.requestId, "allow", "user:alice")).toBe(true); + const first = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + expect(first.operationId).toBe(pending.requestId); + + vi.advanceTimersByTime(2 * 60 * 1_000 + 1); + const expired = (await wrapped.execute!(campaignInput, ctx)) as { + requiresApproval: boolean; + requestId: string; + }; + + expect(expired.requiresApproval).toBe(true); + expect(expired.requestId).not.toBe(pending.requestId); + expect(execute).toHaveBeenCalledTimes(1); + store.clearAll(); + vi.useRealTimers(); + }); + + it("does not consume an approval after material campaign input is changed", async () => { + const { store, execute, wrapped } = makeApprovedTool(); + const ctx = context(); + + const pending = (await wrapped.execute!(campaignInput, ctx)) as { requestId: string }; + expect(store.respond(pending.requestId, "allow", "user:alice")).toBe(true); + + const changed = { + ...campaignInput, + config: { ...campaignInput.config, symbol: "ETHUSDT" }, + }; + const blocked = (await wrapped.execute!(changed, ctx)) as { requiresApproval: boolean }; + + expect(blocked.requiresApproval).toBe(true); + expect(execute).not.toHaveBeenCalled(); + + const original = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + expect(original.operationId).toBe(pending.requestId); + store.clearAll(); + }); + + it("does not consume an approval after the frozen LLM snapshot is substituted", async () => { + const { store, execute, wrapped } = makeApprovedTool(); + const ctx = context(snapshotA); + + const pending = (await wrapped.execute!(campaignInput, ctx)) as { requestId: string }; + expect(store.respond(pending.requestId, "allow", "user:alice")).toBe(true); + + ctx.requestContext.set(USER_LLM_SNAPSHOT_KEY, snapshotB); + const blocked = (await wrapped.execute!(campaignInput, ctx)) as { + requiresApproval: boolean; + }; + expect(blocked.requiresApproval).toBe(true); + expect(execute).not.toHaveBeenCalled(); + + ctx.requestContext.set(USER_LLM_SNAPSHOT_KEY, snapshotA); + const original = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + expect(original.operationId).toBe(pending.requestId); + store.clearAll(); + }); + + it("does not let another owner consume the approved operation", async () => { + let owner = "user:alice"; + const store = new PendingApprovalsStore(); + const { execute, wrapped } = makeApprovedTool({ + store, + owner: () => owner, + }); + const ctx = context(); + + const pending = (await wrapped.execute!(campaignInput, ctx)) as { requestId: string }; + expect(store.respond(pending.requestId, "allow", "user:alice")).toBe(true); + + owner = "user:bob"; + const blocked = (await wrapped.execute!(campaignInput, ctx)) as { + requiresApproval: boolean; + }; + expect(blocked.requiresApproval).toBe(true); + expect(execute).not.toHaveBeenCalled(); + + owner = "user:alice"; + const original = (await wrapped.execute!(campaignInput, ctx)) as { operationId: string }; + expect(original.operationId).toBe(pending.requestId); + store.clearAll(); + }); +}); diff --git a/packages/orchestration/tests/evolver-client.test.ts b/packages/orchestration/tests/evolver-client.test.ts index 04809e43..4eff3642 100644 --- a/packages/orchestration/tests/evolver-client.test.ts +++ b/packages/orchestration/tests/evolver-client.test.ts @@ -6,6 +6,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { buildEvolutionStartRequest, buildEventCampaignRequest, + eventCampaignRequestDigest, evolutionRequestDigest, EvolverClient, } from "../src/clients/evolver.js"; @@ -17,7 +18,7 @@ import { } from "../src/mastra/llm/evolution-snapshot.js"; import { getApprovedEvolutionRunContext, - getAutomaticEventCampaignContext, + getApprovedEventCampaignContext, } from "../src/tools/evolver-shared.js"; const snapshot = buildEvolutionLLMSnapshot({ @@ -38,6 +39,33 @@ function response(status: number): Response { ); } +function campaignResponse( + status = 200, + campaignStatus = "replaying", +): Response { + return new Response( + JSON.stringify({ + campaign_id: "33333333-3333-4333-8333-333333333333", + status: campaignStatus, + active_generation: campaignStatus === "draft" ? 0 : 1, + max_generations: 5, + event_snapshot_id: "11111111-1111-4111-8111-111111111111", + llm_cost_usd: 0, + }), + { status, headers: { "Content-Type": "application/json" } }, + ); +} + +function campaignConflict(): Response { + return new Response( + JSON.stringify({ + code: "CAMPAIGN_STATE_CONFLICT", + message: "campaign cannot start", + }), + { status: 409, headers: { "Content-Type": "application/json" } }, + ); +} + function options() { const request = buildEvolutionStartRequest({ budget: 1, @@ -100,7 +128,7 @@ describe("EvolverClient", () => { expect(Number(credential.exp) - Number(credential.iat)).toBe(108_000); }); - it("marks automatic campaign grants for bounded restart recovery", async () => { + it("binds an approved E2 campaign grant to the shared durable operation identity", async () => { const keys = generateKeyPairSync("ed25519"); vi.stubEnv( "EVOLUTION_CREDENTIAL_PRIVATE_KEY_B64", @@ -108,6 +136,7 @@ describe("EvolverClient", () => { ); const requestContext = new Map([ [AUTH_SUB_KEY, "user:alice"], + [APPROVAL_OPERATION_ID_KEY, "approval-operation-e2"], [USER_LLM_SNAPSHOT_KEY, snapshot], ]); const config = { @@ -118,7 +147,7 @@ describe("EvolverClient", () => { as_of: "2026-08-02T00:00:00Z", }; - const campaign = await getAutomaticEventCampaignContext( + const campaign = await getApprovedEventCampaignContext( { eventSnapshotId: "11111111-1111-4111-8111-111111111111", config }, requestContext, ); @@ -127,7 +156,14 @@ describe("EvolverClient", () => { audience: "inalpha-dashboard-credential", }); - expect(payload.grant_purpose).toBe("event_campaign"); + expect(payload).toMatchObject({ + sub: "user:alice", + grant_purpose: "event_campaign", + operation_id: "approval-operation-e2", + llm_config_digest: snapshot.config_digest, + request_digest: eventCampaignRequestDigest(campaign.request), + }); + expect(campaign.operationId).toBe("approval-operation-e2"); expect(campaign.request).toEqual( buildEventCampaignRequest({ eventSnapshotId: "11111111-1111-4111-8111-111111111111", @@ -137,6 +173,128 @@ describe("EvolverClient", () => { ); }); + it("recovers a lost start response by returning the already-started campaign on whole-operation retry", async () => { + const request = buildEventCampaignRequest({ + eventSnapshotId: "11111111-1111-4111-8111-111111111111", + config: { + venue: "binance", + symbol: "BTCUSDT", + timeframe: "1h", + from_ts: "2026-08-01T00:00:00Z", + as_of: "2026-08-02T00:00:00Z", + }, + llmSnapshot: snapshot, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(campaignResponse(201, "draft")) + .mockRejectedValueOnce(new TypeError("start response lost")) + .mockResolvedValueOnce(campaignResponse(201, "replaying")); + vi.stubGlobal("fetch", fetchMock); + + const client = new EvolverClient({ + baseUrl: "http://evolver.test", + token: "owner-token", + }); + const options = { + request, + idempotencyKey: "approval-operation-e2", + credentialGrant: "event-campaign-grant", + }; + + await expect(client.startEventCampaign(options)).rejects.toMatchObject({ + code: "UPSTREAM_UNREACHABLE", + }); + await expect(client.startEventCampaign(options)).resolves.toMatchObject({ + campaign_id: "33333333-3333-4333-8333-333333333333", + status: "replaying", + }); + + expect(fetchMock).toHaveBeenCalledTimes(3); + for (const index of [0, 2]) { + const [url, init] = fetchMock.mock.calls[index] as [string, RequestInit]; + expect(url).toContain("/api/v1/campaigns"); + expect(url).not.toContain("/start"); + expect((init.headers as Record)["Idempotency-Key"]).toBe( + "approval-operation-e2", + ); + expect((init.headers as Record)["X-Evolution-Credential"]).toBe( + "event-campaign-grant", + ); + expect(init.body).not.toContain("not-forwarded"); + } + }); + + it("reconciles a concurrent E2 start conflict when the campaign already advanced", async () => { + const request = buildEventCampaignRequest({ + eventSnapshotId: "11111111-1111-4111-8111-111111111111", + config: { + venue: "binance", + symbol: "BTCUSDT", + timeframe: "1h", + from_ts: "2026-08-01T00:00:00Z", + as_of: "2026-08-02T00:00:00Z", + }, + llmSnapshot: snapshot, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(campaignResponse(201, "draft")) + .mockResolvedValueOnce(campaignConflict()) + .mockResolvedValueOnce(campaignResponse(200, "replaying")); + vi.stubGlobal("fetch", fetchMock); + + await expect( + new EvolverClient({ + baseUrl: "http://evolver.test", + token: "owner-token", + }).startEventCampaign({ + request, + idempotencyKey: "approval-operation-e2", + credentialGrant: "event-campaign-grant", + }), + ).resolves.toMatchObject({ status: "replaying" }); + + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(String(fetchMock.mock.calls[1]?.[0])).toContain("/start"); + expect(String(fetchMock.mock.calls[2]?.[0])).not.toContain("/start"); + }); + + it("preserves E2 start conflicts when reconciliation still finds a draft campaign", async () => { + const request = buildEventCampaignRequest({ + eventSnapshotId: "11111111-1111-4111-8111-111111111111", + config: { + venue: "binance", + symbol: "BTCUSDT", + timeframe: "1h", + from_ts: "2026-08-01T00:00:00Z", + as_of: "2026-08-02T00:00:00Z", + }, + llmSnapshot: snapshot, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(campaignResponse(201, "draft")) + .mockResolvedValueOnce(campaignConflict()) + .mockResolvedValueOnce(campaignResponse(200, "draft")); + vi.stubGlobal("fetch", fetchMock); + + await expect( + new EvolverClient({ + baseUrl: "http://evolver.test", + token: "owner-token", + }).startEventCampaign({ + request, + idempotencyKey: "approval-operation-e2", + credentialGrant: "event-campaign-grant", + }), + ).rejects.toMatchObject({ + code: "CAMPAIGN_STATE_CONFLICT", + status: 409, + }); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + it("retries 502/504 with the same approval-derived operation ID", async () => { const fetchMock = vi .fn() diff --git a/packages/orchestration/tests/permissions-pending.test.ts b/packages/orchestration/tests/permissions-pending.test.ts index 46613fa1..8390dca8 100644 --- a/packages/orchestration/tests/permissions-pending.test.ts +++ b/packages/orchestration/tests/permissions-pending.test.ts @@ -136,6 +136,196 @@ describe("PendingApprovalsStore", () => { store.clearAll(); }); + it("allows one bounded recovery retry and then exhausts it", async () => { + vi.useFakeTimers(); + const store = new PendingApprovalsStore(() => {}); + const args = { + authSub: "user:alice", + sessionId: "thread-E2", + toolName: "evolver.run_event_campaign", + toolInput: { eventSnapshotId: "event-1" }, + approvalInput: { request: { eventSnapshotId: "event-1" }, llm_snapshot: { config_digest: "digest" } }, + timeoutMs: 5_000, + }; + const view = store.request(args); + expect(store.respond(view.requestId, "allow", args.authSub)).toBe(true); + const consume = { + authSub: args.authSub, + sessionId: args.sessionId, + toolName: args.toolName, + approvalInput: args.approvalInput, + reuseOnceAfterConsumeMs: 120_000, + }; + + expect(await store.consumeApproved(consume)).toBe(view.requestId); + expect(await store.consumeApproved(consume)).toBe(view.requestId); + expect(await store.consumeApproved(consume)).toBeUndefined(); + store.clearAll(); + }); + + it("does not allow the bounded recovery retry after two minutes", async () => { + vi.useFakeTimers(); + const store = new PendingApprovalsStore(() => {}); + const args = { + authSub: "user:alice", + sessionId: "thread-E2", + toolName: "evolver.run_event_campaign", + toolInput: { eventSnapshotId: "event-1" }, + approvalInput: { request: { eventSnapshotId: "event-1" }, llm_snapshot: { config_digest: "digest" } }, + timeoutMs: 5_000, + }; + const view = store.request(args); + expect(store.respond(view.requestId, "allow", args.authSub)).toBe(true); + const consume = { + authSub: args.authSub, + sessionId: args.sessionId, + toolName: args.toolName, + approvalInput: args.approvalInput, + reuseOnceAfterConsumeMs: 120_000, + }; + expect(await store.consumeApproved(consume)).toBe(view.requestId); + vi.advanceTimersByTime(120_001); + expect(await store.consumeApproved(consume)).toBeUndefined(); + store.clearAll(); + }); + + it("serializes concurrent initial E2 consumption into initial plus one retry", async () => { + let releasePersist!: () => void; + const persistGate = new Promise((resolve) => { + releasePersist = resolve; + }); + const operations = new Map(); + const persistence = { + insertPending: vi.fn(async () => {}), + markResolved: vi.fn(async () => {}), + rememberEvolutionOperation: vi.fn(async (scope: { + inputDigest: string; + operationId: string; + retentionMs?: number; + }) => { + await persistGate; + const value = { + operationId: scope.operationId, + expiresAt: new Date(Date.now() + (scope.retentionMs ?? 86_400_000)).toISOString(), + }; + operations.set(scope.inputDigest, value); + return { expiresAt: value.expiresAt }; + }), + findEvolutionOperation: vi.fn(async (scope: { inputDigest: string }) => + operations.get(scope.inputDigest), + ), + claimEvolutionOperation: vi.fn(async (scope: { inputDigest: string }) => { + const value = operations.get(scope.inputDigest); + if (!value || Date.now() >= Date.parse(value.expiresAt)) return undefined; + operations.delete(scope.inputDigest); + return value; + }), + }; + const args = { + authSub: "user:alice", + sessionId: "thread-E2", + toolName: "evolver.run_event_campaign", + toolInput: { eventSnapshotId: "event-1" }, + approvalInput: { + request: { eventSnapshotId: "event-1" }, + llm_snapshot: { config_digest: "digest" }, + }, + timeoutMs: 5_000, + }; + const store = new PendingApprovalsStore(() => {}, persistence); + const view = store.request(args); + expect(store.respond(view.requestId, "allow", args.authSub)).toBe(true); + const consume = { + authSub: args.authSub, + sessionId: args.sessionId, + toolName: args.toolName, + approvalInput: args.approvalInput, + reuseOnceAfterConsumeMs: 120_000, + }; + + const initial = store.consumeApproved(consume); + const concurrent = store.consumeApproved(consume); + await Promise.resolve(); + expect(persistence.rememberEvolutionOperation).toHaveBeenCalledTimes(1); + + releasePersist(); + expect(await initial).toBe(view.requestId); + expect(await concurrent).toBe(view.requestId); + expect(persistence.rememberEvolutionOperation).toHaveBeenCalledTimes(1); + // Initial consumption probes for a prior durable retry before persisting this approval; + // the serialized concurrent call performs the second claim and consumes that retry. + expect(persistence.claimEvolutionOperation).toHaveBeenCalledTimes(2); + expect(await store.consumeApproved(consume)).toBeUndefined(); + store.clearAll(); + }); + + it("atomically allows only one bounded recovery across fresh stores", async () => { + const operations = new Map(); + const persistence = { + insertPending: vi.fn(async () => {}), + markResolved: vi.fn(async () => {}), + rememberEvolutionOperation: vi.fn(async (scope: { + inputDigest: string; + operationId: string; + retentionMs?: number; + }) => { + const value = { + operationId: scope.operationId, + expiresAt: new Date(Date.now() + (scope.retentionMs ?? 86_400_000)).toISOString(), + }; + operations.set(scope.inputDigest, value); + return { expiresAt: value.expiresAt }; + }), + findEvolutionOperation: vi.fn(async (scope: { inputDigest: string }) => + operations.get(scope.inputDigest), + ), + claimEvolutionOperation: vi.fn(async (scope: { inputDigest: string }) => { + const value = operations.get(scope.inputDigest); + if (!value || Date.now() >= Date.parse(value.expiresAt)) return undefined; + operations.delete(scope.inputDigest); + return value; + }), + }; + const args = { + authSub: "user:alice", + sessionId: "thread-E2", + toolName: "evolver.run_event_campaign", + toolInput: { eventSnapshotId: "event-1" }, + approvalInput: { + request: { eventSnapshotId: "event-1" }, + llm_snapshot: { config_digest: "digest" }, + }, + timeoutMs: 5_000, + }; + const first = new PendingApprovalsStore(() => {}, persistence); + const view = first.request(args); + expect(first.respond(view.requestId, "allow", args.authSub)).toBe(true); + const consume = { + authSub: args.authSub, + sessionId: args.sessionId, + toolName: args.toolName, + approvalInput: args.approvalInput, + reuseOnceAfterConsumeMs: 120_000, + }; + expect(await first.consumeApproved(consume)).toBe(view.requestId); + + const retryA = new PendingApprovalsStore(() => {}, persistence); + const retryB = new PendingApprovalsStore(() => {}, persistence); + const claims = await Promise.all([ + retryA.consumeApproved(consume), + retryB.consumeApproved(consume), + ]); + + expect(claims.filter((value) => value === view.requestId)).toHaveLength(1); + expect(claims.filter((value) => value === undefined)).toHaveLength(1); + expect(await new PendingApprovalsStore(() => {}, persistence).consumeApproved(consume)) + .toBeUndefined(); + + first.clearAll(); + retryA.clearAll(); + retryB.clearAll(); + }); + it("does not reuse a consumed evolution operation after its retention deadline", async () => { vi.useFakeTimers(); const store = new PendingApprovalsStore(() => {});