Pi Agent

Use piAgentHooks() with Pi's maintained Agent or AgentHarness. Neatlogs listens to Pi lifecycle events and produces nested Agent, Chain, LLM, and Tool spans. The wrapper is per-instance, so there is no import-order requirement.

Install

npm install neatlogs@latest @earendil-works/pi-agent-core @earendil-works/pi-ai

Maintained Pi 0.83 requires Node.js 22.19 or newer. The wrapper remains compatible with legacy @mariozechner/pi-agent-core projects through the same event interface. Pi packages are ESM-only; import them from an ESM project, .mts, or .mjs file.

Agent

import { Agent } from '@earendil-works/pi-agent-core';
import { createModels } from '@earendil-works/pi-ai';
import { openaiProvider } from '@earendil-works/pi-ai/providers/openai';
import { init, flush, shutdown } from 'neatlogs';
import { piAgentHooks } from 'neatlogs/pi-agent';

await init({
  apiKey: process.env.NEATLOGS_API_KEY ?? '',
  workflowName: 'pi-agent-demo',
});

const models = createModels();
models.setProvider(openaiProvider());
const model = models.getModel('openai', 'gpt-4o-mini');
if (!model) throw new Error('Model is not in the Pi catalog');

const agent = piAgentHooks(
  new Agent({
    initialState: {
      systemPrompt: 'Be concise.',
      model,
      tools: [],
      messages: [],
    },
    streamFn: models.streamSimple.bind(models),
  }),
);

await agent.prompt('In one sentence, what is Pi Agent?');
await flush();
await shutdown();

piAgentHooks() patches in place and returns the same object. Calling it twice on one object is a no-op.

AgentHarness

The maintained package adds sessions, resources, branching, and compaction through AgentHarness. Pass the harness itself to the same helper:

import { AgentHarness } from '@earendil-works/pi-agent-core';

const harness = piAgentHooks(
  new AgentHarness({ session, models, model, tools, resources }),
);

await harness.prompt('Investigate incident INC-42.');
await harness.navigateTree(targetEntryId, { summarize: true });
await harness.compact();

Ordinary harness prompts use Pi's Agent lifecycle events. compact() and summarizing navigateTree() make model calls outside that event loop, so the wrapper captures them explicitly as WORKFLOW → CHAIN → LLM. Non-summarizing navigation produces no fake LLM span. Pure configuration, session persistence, labels, bookmarks, branches, and repository operations do not produce spans because they perform no model/tool work.

Trace shape

AGENT pi_agent.run
└─ CHAIN pi_agent.turn.N
   ├─ LLM pi_agent.llm.<model>
   └─ TOOL pi_agent.tool.<name>

Captured data includes run and turn I/O, tool arguments/results, provider and model, tool calls, token counts, Pi's exact usage.cost.total, stop reason, streaming state, and time to first token. Parallel tool calls remain separate through Pi's toolCallId. Aborted and failed calls close every open span and mark the affected tree as ERROR.

Steering and follow-up messages become further Chain turns in the same run. A later prompt() is a new trace; use a stable session ID to group those conversational turns.

Sessions and end-users

Bind identity around the operation that creates the root span:

import { identify } from 'neatlogs';

await identify(
  { sessionId: conversationId, endUserId: user.id },
  () => harness.prompt(message),
);

Reuse the same sessionId for each turn of one conversation. For a standalone job, omit sessionId and provide only the end-user identity.

Functional loops

Functional loop APIs have no object to wrap. Feed every event into a fresh tracePiAgentEvents() listener:

import { agentLoop } from '@earendil-works/pi-agent-core';
import { tracePiAgentEvents } from 'neatlogs/pi-agent';

const listener = tracePiAgentEvents(() => context.messages);
const stream = agentLoop(prompts, context, config, undefined, streamFn);
for await (const event of stream) listener(event);
context.messages.push(...(await stream.result()));

The same listener works in the sink passed to runAgentLoop. The prompt-taking forms run on a copy of context.messages, so append returned messages before a continuation. Pass () => context.messages so continuation input can be recovered.

Standalone stream functions

Use tracePiStream() only for a model stream called outside an Agent or functional loop:

import { tracePiStream } from 'neatlogs/pi-agent';

const traced = tracePiStream(
  async (...args: Parameters<typeof streamFn>) => streamFn(...args),
);
const stream = await traced(model, modelContext);
for await (const event of stream) consume(event);
const result = await stream.result();

It supports both EventStream and Promise<EventStream>, observes iteration without consuming it, records streaming/TTFT and usage, and closes spans on synchronous or asynchronous failure. Standalone calls receive a pi_agent.stream Workflow root.

Await every active run before the final flush()/shutdown(). waitForIdle() is useful when work was started elsewhere or queue operations may still be active.