durable-harness/fx: Agent Runs from libfx hooks (#11), with handed-off and fenced turns kept open

.changeset/fx-agent-runs-hooks.md

@@ -0,0 +1,5 @@
+---
+"durable-harness": patch
+---
+
+libfx's own durable agents show up in Vercel Agent Runs: `createFxAgent({ hooks: agentRunsHooks() })` with `agentRunsHooks` from `durable-harness/fx` records each libfx turn, its steps with their model calls, and its tool calls and the approvals they wait on, from libfx's semantic hooks (libfx 0.0.14 and later). A turn libfx yields, pauses or hands to another worker (at its deadline, or when another worker's write fences it out) stays one turn, continued by the execution that resumes it in any process, and a call the handoff cut off runs again as the same call. With `agentRuns()` registered first, libfx passes content only where the audience records it.

apps/demo/app/api/libfx/route.ts

@@ -1,4 +1,4 @@
-import { fxAgent, withFxTraces, log, refuseDisabled } from '@/lib/fx-durable';
+import { fxAgent, log, refuseDisabled } from '@/lib/fx-durable';
 
 // Interim, as the libfx durability doc says: Vercel Queues delivers libfx's
 // session messages here until world-vercel can deliver to the function
@@ -13,7 +13,7 @@ export async function POST(request: Request): Promise<Response> {
   if (disabled) return disabled;
   const started = Date.now();
   wake ??= fxAgent().then((agent) => agent.wakeHandler());
-  const response = await withFxTraces(async () => (await wake!)(request));
+  const response = await (await wake)(request);
   const body = await response.clone().text().catch(() => '');
   log('wake', { ms: Date.now() - started, status: response.status, body: body.slice(0, 120) });
   return response;

apps/demo/lib/fx-durable.test.ts

@@ -1,71 +1,55 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
-import { fxAgent, withFxTraces } from './fx-durable';
+import { fxAgent } from './fx-durable';
 
-const traced = vi.hoisted(() => ({ complete: vi.fn(async () => {}), fail: vi.fn(async () => {}), fetch: undefined as typeof fetch | undefined }));
+const captured = vi.hoisted(() => ({ fetch: undefined as typeof fetch | undefined }));
 vi.mock('libfx', () => ({ createFxAgent: async (options: { fetch: typeof fetch }) => {
-  traced.fetch = options.fetch;
+  captured.fetch = options.fetch;
   return {};
 } }));
 vi.mock('./model', () => ({ MODEL: 'test', INSTRUCTIONS: '' }));
 vi.mock('./fx-durable-world', () => ({ fxDurability: async () => undefined, fxDurableDisabledReason: () => undefined, refuseDisabled: vi.fn() }));
 vi.mock('./world', () => ({ getWorld: async () => ({ events: { list: async () => ({ data: [] }) }, streams: { write: async () => {} } }) }));
-vi.mock('harness-world/agent-runs', () => {
-  const step = { modelCall: async () => ({ complete: traced.complete, fail: traced.fail }), attributes() {}, complete: async () => {}, tool: async () => undefined };
-  const turn = { findTool: () => undefined, complete: async () => {} };
-  return { agentRunsHook: () => ({ session: () => ({ turn: async () => turn, step: async () => step, flush: async () => {} }) }) };
-});
 
 beforeEach(() => { vi.stubEnv('HARNESS_CHAOS', '0'); });
 afterEach(() => {
   vi.unstubAllEnvs();
   vi.unstubAllGlobals();
   vi.clearAllMocks();
   delete (globalThis as { __fxDurableAgent?: unknown }).__fxDurableAgent;
-  traced.fetch = undefined;
+  captured.fetch = undefined;
 });
 
 async function modelRequest() {
   await fxAgent();
-  return withFxTraces(() => traced.fetch!('https://gateway.example.test/v3/ai/language-model', {
+  return captured.fetch!('https://gateway.example.test/v3/ai/language-model', {
     method: 'POST', headers: { 'x-session-id': 'test-session' }, body: JSON.stringify({ prompt: [] }),
-  }));
+  });
 }
 
-describe('native libfx model response tracing', () => {
+describe("the libfx lane's model fetch", () => {
   it.each([
     { status: 200, body: 'data: {"type":"text-delta","delta":"hi"}\n\n' },
     { status: 204, body: null },
     { status: 429, body: '{"error":"rate limited"}' },
     { status: 500, body: null },
     { status: 302, body: null },
-  ])('preserves HTTP $status and marks non-success responses as failures', async ({ status, body }) => {
+  ])('passes HTTP $status and its body through', async ({ status, body }) => {
     vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(body, { status })));
     const response = await modelRequest();
     expect(response.status).toBe(status);
     expect(await response.text()).toBe(body ?? '');
-    if (status >= 200 && status < 300) {
-      expect(traced.complete).toHaveBeenCalledTimes(1);
-      expect(traced.fail).not.toHaveBeenCalled();
-    } else {
-      expect(traced.complete).not.toHaveBeenCalled();
-      expect(traced.fail).toHaveBeenCalledWith(new Error(`Model request failed (${status})`));
-    }
   });
-  it('preserves a fetch rejection as the trace error', async () => {
+  it('rejects with the fetch rejection', async () => {
     const error = new Error('network down');
     vi.stubGlobal('fetch', vi.fn().mockRejectedValue(error));
     await expect(modelRequest()).rejects.toBe(error);
-    expect(traced.fail).toHaveBeenCalledTimes(1);
-    expect(traced.fail).toHaveBeenCalledWith(error);
   });
-  it('preserves a response stream failure instead of masking it with the HTTP error', async () => {
+  it("fails the response's stream with the stream's own error", async () => {
     const error = new Error('stream interrupted');
     vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(new ReadableStream({
       pull(controller) { controller.error(error); },
     }), { status: 500 })));
     const response = await modelRequest();
     await expect(response.text()).rejects.toBe(error);
-    expect(traced.fail).toHaveBeenCalledTimes(1);
-    expect(traced.fail).toHaveBeenCalledWith(error);
   });
 });

apps/demo/lib/fx-durable.ts

@@ -1,10 +1,6 @@
-import { AsyncLocalStorage } from 'node:async_hooks';
-import { fxPromptTraceIdentity, fxTraceIdentity, fxTraceRequestBody } from './fx-trace';
-import { agentRunsHook, type SessionTracing } from 'harness-world/agent-runs';
-import { gatewayModelOptions, gatewayModelResult, ResponseCapture } from 'harness-world-process';
 import type { World } from '@workflow/world';
 import type { ChaosSpec } from 'harness-world-demo-ui';
-import { fxTools } from 'durable-harness/fx';
+import { agentRunsHooks, fxTools } from 'durable-harness/fx';
 import { virtual } from 'durable-harness/tools';
 import { type FxChaosLine, type FxWorkspace, LIBFX_WORKFLOW } from 'harness-world-demo-ui/fx-view';
 import { readFxWorkspace } from 'harness-world-hands/fx';
@@ -58,57 +54,6 @@ export const log = (event: string, detail: Record<string, unknown>) => console.l
 const LOGGED = new Set(['journal.open', 'checkpoint.save', 'checkpoint.error', 'checkpoint.mismatch', 'session.deadline', 'session.fenced', 'session.error']);
 
 type FxAgent = ReturnType<typeof createFxAgent>;
-type Turn = NonNullable<Awaited<ReturnType<SessionTracing['turn']>>>;
-type Step = Awaited<ReturnType<SessionTracing['step']>>;
-/** A libfx turn in Agent Runs: its operation, and the step of its latest model request (the tool calls it asks for go under it). */
-type FxTraceState = { session: SessionTracing; turn: Turn; turnId: string; step?: Step; outcome?: 'running' | 'failed' };
-type FxTraceInvocation = { traces: Map<string, FxTraceState>; turns: Map<string, { turnId: string; prompt?: string }> };
-const fxTraceInvocations = new AsyncLocalStorage<FxTraceInvocation>();
-
-/**
- * Agent Runs for libfx's own durable sessions: a turn per libfx turn, a step
- * per model request, the tool calls under it. A turn that goes on in a later
- * wake delivery continues from its checkpoint; one delivery ends only the
- * turns it worked on.
- */
-export async function withFxTraces<T>(run: () => Promise<T>): Promise<T> {
-  return fxTraceInvocations.run({ traces: new Map(), turns: new Map() }, async () => {
-    let error: unknown;
-    try { return await run(); } catch (caught) { error = caught; throw caught; }
-    finally {
-      for (const state of fxTraceInvocations.getStore()!.traces.values()) await endFxTrace(state, error).catch(() => {});
-    }
-  });
-}
-async function endFxTrace(state: FxTraceState, error?: unknown): Promise<void> {
-  const outcome = state.outcome ?? (error === undefined ? 'completed' : 'failed');
-  await state.step?.complete();
-  if (outcome === 'failed') await state.turn.complete({ outcome, failed: true, error: error ?? new Error('the libfx session failed') });
-  else if (outcome === 'completed') await state.turn.complete({ outcome });
-  await state.session.flush();
-}
-async function fxTrace(id: string, key: string, refreshTurn = false): Promise<FxTraceState | undefined> {
-  const invocation = fxTraceInvocations.getStore();
-  const hook = agentRunsHook();
-  if (!invocation || !hook) return undefined;
-  const { traces, turns } = invocation;
-  const existing = traces.get(id);
-  if (existing && !refreshTurn) return existing;
-  // libfx's accepted journal marks preserve the same turn id across resumes.
-  const current = turns.get(id);
-  const identity = current ?? fxTraceIdentity((await (await getWorld()).events.list({
-    runId: id, pagination: { sortOrder: 'desc', limit: 100 }, resolveData: 'all',
-  })).data, existing?.turnId ?? key);
-  const { turnId, prompt } = identity;
-  if (existing?.turnId === turnId) return existing;
-  if (existing) await endFxTrace(existing);
-  const session = hook.session({ world: await getWorld(), runId: id, agentName: 'fx-durable' });
-  const turn = await session.turn({ turnId, sequence: 0, ...(typeof prompt === 'string' ? { title: prompt } : {}) });
-  if (!turn) return undefined;
-  const state: FxTraceState = { session, turn, turnId };
-  traces.set(id, state);
-  return state;
-}
 const g = globalThis as { __fxDurableAgent?: Promise<FxAgent> };
 
 /** The process's agent. Every process creates the same one, so any of them can run any session. */
@@ -123,16 +68,9 @@ export function fxAgent(): Promise<FxAgent> {
       instructions: INSTRUCTIONS,
       tools: await workspaceTools(),
       fetch: timedFetch,
+      // Agent Runs, from libfx's semantic hooks: its turns, steps, model and tool calls.
+      hooks: agentRunsHooks({ world: await getWorld(), name: 'fx-durable' }),
       onEvent(event: FxAgentEvent) {
-        if (event.type === 'acp.send') {
-          const identity = fxPromptTraceIdentity(event.message);
-          if (identity) fxTraceInvocations.getStore()?.turns.set(identity.sessionId, identity);
-        }
-        if ('sessionId' in event && event.sessionId) {
-          const state = fxTraceInvocations.getStore()?.traces.get(event.sessionId);
-          if (state && (event.type === 'session.deadline' || event.type === 'session.fenced')) state.outcome = 'running';
-          if (state && event.type === 'session.error') state.outcome = 'failed';
-        }
         if (!LOGGED.has(event.type)) return;
         const { type, timestamp: _timestamp, ...detail } = event;
         log(type, detail);
@@ -256,41 +194,21 @@ async function timedFetch(input: RequestInfo | URL, init?: RequestInit): Promise
   const method = (init?.method ?? (input instanceof Request ? input.method : 'GET')).toUpperCase();
   if (!id || method === 'GET') return fetch(input, init);
   const key = `model:${crypto.randomUUID().slice(0, 12)}`;
-  let requestBody: unknown;
-  try { requestBody = fxTraceRequestBody(init?.body ?? (input instanceof Request ? await input.clone().text() : undefined)); } catch {}
-  const path = new URL(input instanceof Request ? input.url : String(input)).pathname;
-  const modelTrace = await (async () => {
-    const state = await fxTrace(id, key, true);
-    if (!state) return undefined;
-    await state.step?.complete();
-    const step = (state.step = await state.session.step(state.turn, state.turnId));
-    return { step, model: await step.modelCall(gatewayModelOptions({ path, model: headers.get('ai-language-model-id') ?? MODEL, body: requestBody })) };
-  })().catch(() => undefined);
-  const traceBody = new ResponseCapture();
-  const traceDecoder = new TextDecoder();
   mark(id, 'model.request', { key, effect: 'model' });
   // Checked beside the request, so it adds nothing to the model's time.
   const armed = crashDuring(id, key, 'model', 'model');
   let ended = false;
   let response: Response | undefined;
-  const end = (error?: unknown) => {
+  const end = () => {
     if (ended) return;
     ended = true;
-    const status = response?.status;
-    const failed = error ?? (response && !response.ok ? new Error(`Model request failed (${status})`) : undefined);
-    if (modelTrace && failed !== undefined) void modelTrace.model.fail(failed).catch(() => {});
-    else if (modelTrace) {
-      const { result, attributes } = gatewayModelResult(path, traceBody.text());
-      modelTrace.step.attributes(attributes);
-      void modelTrace.model.complete({ outcome: 'completed', result }).catch(() => {});
-    }
     void armed.then((disarm) => disarm());
     mark(id, 'model.done', { key, effect: 'model' });
   };
   try {
     response = await fetch(input, init);
   } catch (error) {
-    end(error);
+    end();
     throw error;
   }
   if (!response.body) {
@@ -306,7 +224,6 @@ async function timedFetch(input: RequestInfo | URL, init?: RequestInit): Promise
       try {
         const { value, done } = await reader.read();
         if (done) {
-          traceBody.push(traceDecoder.decode());
           end();
           controller.close();
           return;
@@ -318,10 +235,9 @@ async function timedFetch(input: RequestInfo | URL, init?: RequestInit): Promise
             mark(id, 'model.first_token', { key, effect: 'model' });
           }
         }
-        traceBody.push(traceDecoder.decode(value, { stream: true }));
         controller.enqueue(value);
       } catch (error) {
-        end(error);
+        end();
         controller.error(error);
       }
     },
@@ -357,22 +273,14 @@ async function crashDuring(id: string, effectKey: string, effectName: string, st
 }
 
 /** Run one tool call with its timing marks and an armed crash. */
-async function toolStep<T>(id: string, ctx: FxToolContext, name: string, input: unknown, run: () => Promise<T>): Promise<T> {
+async function toolStep<T>(id: string, ctx: FxToolContext, name: string, run: () => Promise<T>): Promise<T> {
   const key = `tool:${ctx.executionId}`;
   const effect = `tool.${name}`;
-  const toolTrace = await (async () => {
-    const state = await fxTrace(id, key);
-    if (!state) return undefined;
-    // `executionId` is the model's id for the call: Agent Runs joins it with the call in the model's reply.
-    return state.turn.findTool(ctx.executionId) ?? (await (state.step ??= await state.session.step(state.turn, state.turnId)).tool({ callId: ctx.executionId, name, arguments: input }));
-  })().catch(() => undefined);
   mark(id, 'tool.start', { key, effect });
   const disarm = await crashDuring(id, key, effect, 'tool');
   try {
-    const result = await run();
-    await toolTrace?.complete({ outcome: 'completed', output: result }).catch(() => {});
-    return result;
-  } catch (error) { await toolTrace?.fail(error).catch(() => {}); throw error; } finally {
+    return await run();
+  } finally {
     disarm();
     mark(id, 'tool.end', { key, effect });
   }
@@ -383,21 +291,21 @@ const descriptor = (name: Spec) => ({ name, description: TOOL_SPECS[name].descri
 
 /**
  * libfx's own tools (durable-harness/fx), each call with this lane's timing
- * marks, Agent Runs span and armed crash, and the two unsafe example tools
+ * marks and armed crash, and the two unsafe example tools
  * (not idempotent: never run twice).
  */
 export async function workspaceTools() {
   const world = await getWorld();
   const traced = fxTools(virtual(), { world }).map((tool) => ({
     ...tool,
-    execute: (input: unknown, ctx: FxToolContext) => toolStep(ctx.sessionId, ctx, tool.name, input, async () => tool.execute(input, ctx)),
+    execute: (input: unknown, ctx: FxToolContext) => toolStep(ctx.sessionId, ctx, tool.name, async () => tool.execute(input, ctx)),
   }));
   const unsafe = (name: UnsafeToolName) => ({
     ...descriptor(name),
     idempotent: false,
     writes: true,
     async execute(input: Record<string, unknown>, ctx: FxToolContext) {
-      return toolStep(ctx.sessionId, ctx, name, input, async () => {
+      return toolStep(ctx.sessionId, ctx, name, async () => {
         const ref = `${name === 'unsafe_send_email' ? 'msg' : 'pay'}_${ctx.executionId.slice(-10)}`;
         await new Promise<void>((resolve) => {
           const timer = setTimeout(resolve, UNSAFE_CONFIRM_MS);