fix(adversary): bound agent result reads

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<string | null>,
+	): Promise<string | null>;
+}
+
+export async function readAdversaryAgentResult<S extends v.GenericSchema>(
+	step: AgentReadWorkflowStep,
+	team: 'blue' | 'purple',
+	submissionId: string,
+	read: (signal: AbortSignal) => Promise<AgentReply>,
+	schema: S,
+): Promise<v.InferOutput<S>> {
+	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.`,
+	);
+}

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<S extends v.GenericSchema>(
-	data: Record<string, unknown[]>,
-	schema: S,
-): v.InferOutput<S> {
-	const writes = data.result;
-	if (!writes?.length)
-		throw new Error('The adversary agent produced no result.');
-	return v.parse(schema, writes.at(-1));
-}

src/app.ts

@@ -10,7 +10,14 @@ instrument(createCloudflareTracing({ content: false }));
 const flueEventLoggers = new Map<string, FlueEventLogger>();
 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);

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;
 			}
 		},

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<string | null>;
+
+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);
+	});
+});

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));