diff --git a/src/adversary/agent-read.ts b/src/adversary/agent-read.ts new file mode 100644 index 0000000..063f90b --- /dev/null +++ b/src/adversary/agent-read.ts @@ -0,0 +1,75 @@ +import type { AgentReply } from '@flue/runtime'; +import * as v from 'valibot'; + +// Keep each long-poll well below Workflows' recommended 30-minute step limit. +const OBSERVATION_WINDOW_MS = 5 * 60 * 1_000; +// Ten windows cover the agent's 45-minute durability timeout with one final check. +const OBSERVATION_WINDOWS = 10; +// Leave one minute for cancellation and result persistence before the step expires. +const OBSERVATION_STEP = { + retries: { limit: 3, delay: '5 seconds', backoff: 'exponential' }, + timeout: '6 minutes', +} as const; + +interface AgentReadWorkflowStep { + do( + name: string, + config: typeof OBSERVATION_STEP, + callback: (context: { attempt: number }) => Promise, + ): Promise; +} + +export async function readAdversaryAgentResult( + step: AgentReadWorkflowStep, + team: 'blue' | 'purple', + submissionId: string, + read: (signal: AbortSignal) => Promise, + schema: S, +): Promise> { + for (let index = 0; index < OBSERVATION_WINDOWS; index++) { + const result = await step.do( + `read ${team} result`, + OBSERVATION_STEP, + async (context) => { + const controller = new AbortController(); + const timeout = setTimeout( + () => + controller.abort(new Error('Agent observation window elapsed.')), + OBSERVATION_WINDOW_MS, + ); + console.info( + `[adversary] team=${team} submission=${submissionId} observation=${index + 1}/${OBSERVATION_WINDOWS} attempt=${context.attempt} state=started`, + ); + + try { + const reply = await read(controller.signal); + const writes = reply.data.result; + if (!writes?.length) + throw new Error('The adversary agent produced no result.'); + const serialized = JSON.stringify(writes.at(-1)); + if (serialized === undefined) + throw new Error('The adversary agent produced no result.'); + console.info( + `[adversary] team=${team} submission=${submissionId} observation=${index + 1}/${OBSERVATION_WINDOWS} state=settled`, + ); + return serialized; + } catch (error) { + if (controller.signal.aborted && error === controller.signal.reason) { + console.info( + `[adversary] team=${team} submission=${submissionId} observation=${index + 1}/${OBSERVATION_WINDOWS} state=pending`, + ); + return null; + } + throw error; + } finally { + clearTimeout(timeout); + } + }, + ); + if (result !== null) return v.parse(schema, JSON.parse(result)); + } + + throw new Error( + `The ${team} agent did not settle after ${OBSERVATION_WINDOWS} observation windows.`, + ); +} diff --git a/src/adversary/workflow.ts b/src/adversary/workflow.ts index d8fface..766795d 100644 --- a/src/adversary/workflow.ts +++ b/src/adversary/workflow.ts @@ -12,6 +12,7 @@ import { credentialsFromWorkerEnv, } from '../github/client.ts'; import { removeLabelIfPresent } from '../github/issues.ts'; +import { readAdversaryAgentResult } from './agent-read.ts'; import { BlueTeam } from './agents/blue-team.ts'; import { PurpleTeam } from './agents/purple-team.ts'; import { @@ -306,11 +307,12 @@ export class AdversaryWorkflow extends WorkflowEntrypoint< }, }), ); - return step.do( - 'read blue result', - { ...RETRIES, timeout: '50 minutes' }, - async () => - extractResult((await agent.read(receipt)).data, blueTeamResultSchema), + return readAdversaryAgentResult( + step, + 'blue', + receipt.submissionId, + (signal) => agent.read(receipt, { signal }), + blueTeamResultSchema, ); } @@ -361,14 +363,12 @@ export class AdversaryWorkflow extends WorkflowEntrypoint< }, }), ); - return await step.do( - 'read purple result', - { ...RETRIES, timeout: '50 minutes' }, - async () => - extractResult( - (await agent.read(receipt)).data, - purpleTeamResultSchema, - ), + return await readAdversaryAgentResult( + step, + 'purple', + receipt.submissionId, + (signal) => agent.read(receipt, { signal }), + purpleTeamResultSchema, ); } finally { await destroyInStep(step, 'purple', sandbox); @@ -436,13 +436,3 @@ async function destroyInStep( () => destroyAdversarySandbox(sandbox), ); } - -function extractResult( - data: Record, - schema: S, -): v.InferOutput { - const writes = data.result; - if (!writes?.length) - throw new Error('The adversary agent produced no result.'); - return v.parse(schema, writes.at(-1)); -} diff --git a/src/app.ts b/src/app.ts index 0b03ba1..c9dcbe9 100644 --- a/src/app.ts +++ b/src/app.ts @@ -10,7 +10,14 @@ instrument(createCloudflareTracing({ content: false })); const flueEventLoggers = new Map(); observe((event, context) => { if (context.id.startsWith('release-security:')) return; - const logger = flueEventLoggers.get(context.id) ?? createFlueEventLogger(); + const logger = + flueEventLoggers.get(context.id) ?? + createFlueEventLogger((message) => + console.info(message, { + agentName: context.agentName, + contextId: context.id, + }), + ); flueEventLoggers.set(context.id, logger); logger.present(event); if (event.type === 'submission_settled') flueEventLoggers.delete(context.id); diff --git a/src/flue-logging.ts b/src/flue-logging.ts index 685aec8..f63e96c 100644 --- a/src/flue-logging.ts +++ b/src/flue-logging.ts @@ -42,6 +42,18 @@ export function createFlueEventLogger( flush, present(event) { switch (event.type) { + case 'agent_start': + flush(); + write('[flue] agent:start'); + break; + case 'agent_end': + flush(); + write('[flue] agent:done'); + break; + case 'turn_start': + flush(); + write(`[flue] turn:start purpose=${event.purpose}`); + break; case 'text_delta': flushThinking(); beginText(); @@ -93,11 +105,38 @@ export function createFlueEventLogger( `[flue] compaction:done messages ${event.messagesBefore} -> ${event.messagesAfter}`, ); break; - case 'agent_end': case 'turn': + flush(); + write( + `[flue] turn:${event.isError ? 'error' : 'done'} purpose=${event.purpose} (${event.durationMs}ms)`, + ); + break; + case 'submission_queued': + flush(); + write( + `[flue] submission:queued id=${event.submissionId} kind=${event.kind}`, + ); + break; + case 'submission_running': + flush(); + write( + `[flue] submission:running id=${event.submissionId} attempt=${event.attemptCount}/${event.maxAttempts}`, + ); + break; + case 'submission_recovery': + flush(); + write( + `[flue] submission:recovery${event.submissionId ? ` id=${event.submissionId}` : ''} operation=${event.operation} outcome=${event.outcome}${event.error ? ` error=${truncate(redact(event.error.message))}` : ''}`, + ); + break; case 'idle': + flush(); + break; case 'submission_settled': flush(); + write( + `[flue] submission:settled id=${event.submissionId} outcome=${event.outcome}${event.error ? ` error=${truncate(redact(event.error.message))}` : ''}`, + ); break; } }, diff --git a/tests/adversary-agent-read.test.ts b/tests/adversary-agent-read.test.ts new file mode 100644 index 0000000..14b9c51 --- /dev/null +++ b/tests/adversary-agent-read.test.ts @@ -0,0 +1,94 @@ +import type { AgentReply } from '@flue/runtime'; +import * as v from 'valibot'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { readAdversaryAgentResult } from '../src/adversary/agent-read.ts'; + +const resultSchema = v.object({ solved: v.boolean() }); +type StepCallback = (context: { attempt: number }) => Promise; + +function reply(solved: boolean): AgentReply { + return { + text: '', + data: { result: [{ solved }] }, + submissionId: 'sub-1', + }; +} + +describe('readAdversaryAgentResult', () => { + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('returns a settled result from the first observation', async () => { + vi.spyOn(console, 'info').mockImplementation(() => {}); + const names: string[] = []; + const step = { + do: async (name: string, _config: unknown, callback: StepCallback) => { + names.push(name); + return callback({ attempt: 1 }); + }, + }; + + await expect( + readAdversaryAgentResult( + step, + 'blue', + 'sub-1', + async () => reply(true), + resultSchema, + ), + ).resolves.toEqual({ solved: true }); + expect(names).toEqual(['read blue result']); + }); + + it('reattaches after a bounded observation elapses', async () => { + vi.useFakeTimers(); + vi.spyOn(console, 'info').mockImplementation(() => {}); + let reads = 0; + const step = { + do: async (_name: string, _config: unknown, callback: StepCallback) => + callback({ attempt: 1 }), + }; + const result = readAdversaryAgentResult( + step, + 'purple', + 'sub-1', + (signal) => { + reads++; + if (reads > 1) return Promise.resolve(reply(true)); + return new Promise((_, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { + once: true, + }); + }); + }, + resultSchema, + ); + + await vi.advanceTimersByTimeAsync(5 * 60 * 1_000); + await expect(result).resolves.toEqual({ solved: true }); + expect(reads).toBe(2); + }); + + it('does not hide agent failures', async () => { + vi.spyOn(console, 'info').mockImplementation(() => {}); + const failure = new Error('agent failed'); + const step = { + do: async (_name: string, _config: unknown, callback: StepCallback) => + callback({ attempt: 1 }), + }; + + await expect( + readAdversaryAgentResult( + step, + 'blue', + 'sub-1', + async () => { + throw failure; + }, + resultSchema, + ), + ).rejects.toBe(failure); + }); +}); diff --git a/tests/flue-logging.test.ts b/tests/flue-logging.test.ts index 5202dfe..f260b2a 100644 --- a/tests/flue-logging.test.ts +++ b/tests/flue-logging.test.ts @@ -67,7 +67,7 @@ describe('createFlueEventLogger', () => { text: 'first line\npartial', } as FlueEvent); logger.present({ type: 'text_delta', text: ' line\n' } as FlueEvent); - logger.present({ type: 'turn' } as FlueEvent); + logger.present({ type: 'idle' } as FlueEvent); expect(lines).toEqual([ '[flue] assistant', @@ -76,6 +76,37 @@ describe('createFlueEventLogger', () => { ]); }); + it('logs agent and submission lifecycle without payload content', () => { + const lines: string[] = []; + const logger = createFlueEventLogger((line) => lines.push(line)); + + logger.present({ type: 'agent_start' } as FlueEvent); + logger.present({ + type: 'submission_queued', + submissionId: 'sub-1', + kind: 'dispatch', + } as FlueEvent); + logger.present({ + type: 'submission_running', + submissionId: 'sub-1', + kind: 'dispatch', + attemptCount: 1, + maxAttempts: 3, + } as FlueEvent); + logger.present({ + type: 'submission_settled', + submissionId: 'sub-1', + outcome: 'completed', + } as FlueEvent); + + expect(lines).toEqual([ + '[flue] agent:start', + '[flue] submission:queued id=sub-1 kind=dispatch', + '[flue] submission:running id=sub-1 attempt=1/3', + '[flue] submission:settled id=sub-1 outcome=completed', + ]); + }); + it('flushes a tokenized thought as one readable line', () => { const lines: string[] = []; const logger = createFlueEventLogger((line) => lines.push(line));