sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

provider-events: raw provider stream recorder (LH_PROVIDER_EVENTS=1)

Browse Source 
pi-ai 1.0's onProviderStreamEvent, at the layer our incident class lives
(GLM arg truncation, DeepSeek id resets) — env-gated (owner choice:
no app surface, off by default; raw events carry full prompts), one
JSONL per chat under data/debug/provider-events/ with a 20MB cap and a
run-start marker for record correlation.

Wired at every agent construction: teacher/prober turns (AgentBuildOptions.
chatId), jobs via LoopContext (researcher, demo-builder through
runSubagent; visual-maker + pdf-reader through their streamOnce), and
the extraction pipeline (extractor + auditor keyed by source id).
practice deliberately unwired (sunk-cost module).
This commit is contained in:
amosblomqvist
2026-10-04 20:21:18 +03:00
parent d2d159d8ae
commit a2c1287415
12 changed files with 330 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/subagents/demo-builder.ts
+3
View File
@@ -40,6 +40,8 @@ export interface DemoBuilderOptions {
abortSignal?: AbortSignal;
/** Money channel: one call per agent step, folded into JobRecord.usage. */
onUsage?: (usage: Usage) => void;
/** Debug: raw provider-event recording key (LH_PROVIDER_EVENTS=1). */
chatId?: string;
}
const INSTRUCTIONS = `You are a demo builder — a specialist sub-agent inside a learning application. You
@@ -209,6 +211,7 @@ export async function runDemoBuilder(input: DemoBuilderInput, opts: DemoBuilderO
abortSignal: opts.abortSignal,
onUsage: (usage) => opts.onUsage?.(usage),
onToolStart: (toolName) => opts.onActivity?.(toolName),
chatId: opts.chatId,
});
finalText = result.text;
} catch (err) {
src/agent/subagents/extraction.ts
+16 -3
View File
@@ -372,6 +372,7 @@ export function sourceDigest(graph: DerivedGraph, sourceId: string): string {
/** The extractor over ONE session transcript (the chat front door). */
function runExtractor(
engine: PiEngine | undefined,
sourceId: string,
prompt: string,
existingIds: ReadonlySet<string>,
signal: AbortSignal | undefined,
@@ -382,6 +383,9 @@ function runExtractor(
// over the MIXED namespace: local ids + existing graph ids.
return runSubmitAgent<FindingsSubmit>({
role: "extractor",
// Debug recorder key: the source id (the run's ledger key — the
// chat's chatId for chat runs, the src id for notes).
chatId: sourceId,
toolName: "submit_findings",
toolDescription: "Submit the session findings.",
parameters: findingsSchema,
@@ -401,6 +405,7 @@ function runExtractor(
* front door): same single-submit shape, notes-mode prompt + schema. */
function runNotesExtractor(
engine: PiEngine | undefined,
sourceId: string,
prompt: string,
existingIds: ReadonlySet<string>,
signal: AbortSignal | undefined,
@@ -408,6 +413,8 @@ function runNotesExtractor(
): Promise<NotesFindings> {
return runSubmitAgent<FindingsSubmit>({
role: "extractor",
// Debug recorder key: the notes' source id (the ledger key).
chatId: sourceId,
toolName: "submit_findings",
toolDescription: "Submit the notes findings.",
parameters: findingsSchema,
@@ -467,6 +474,7 @@ Include the "links" field even when empty — "links": [] when nothing links. NE
* tools plus the one submit call. */
function runAuditor(
engine: PiEngine | undefined,
sourceId: string,
prompt: string,
signal: AbortSignal | undefined,
onStep: (usage: Usage, model: string, step: number, finishReason: string, tools: string) => void,
@@ -474,6 +482,8 @@ function runAuditor(
): Promise<AuditVerdicts> {
return runSubmitAgent({
role: "auditor",
// Debug recorder key: same source id as the extractor run.
chatId: sourceId,
toolName: "submit_review",
toolDescription: "Submit the identity verdicts ONLY (links) — never the propositions themselves.",
parameters: auditSchema,
@@ -804,6 +814,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
const findings = await runExtractor(
stageEngine("extractor", deps.extractor),
sourceId,
prompt,
new Set(graph.nodes.map((n) => n.id)),
signal,
@@ -816,7 +827,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
return findings;
}
async function audit(jobKey: string, findings: SessionFindings): Promise<AuditVerdicts> {
async function audit(jobKey: string, sourceId: string, findings: SessionFindings): Promise<AuditVerdicts> {
const signal = runningExtractions.get(jobKey)?.controller.signal;
signal?.throwIfAborted(); // an abort during the extract→audit gap ends here
const graph = getGraph(graphStore);
@@ -831,6 +842,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// exploration tools instead of an injected dump.
const auditResult = await runAuditor(
stageEngine("auditor", deps.auditor),
sourceId,
`Proposed findings from the extractor (the source itself is not available to you — the extractor is fully trusted; the propositions only — the reasoning chains are the merger's business, owner 2026-09-30):\n\n` +
`${JSON.stringify({ propositions: findings.propositions }, null, 2)}\n\n` +
`The current knowledge graph has ${graph.nodes.length} node(s). Explore it with your tools and link what links.`,
@@ -1017,7 +1029,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// stands AFTER every earlier job's merge.
return withAuditLock(jobKey, "chat", undefined, async () => {
setPhase(jobKey, "auditing", "chat", undefined, part);
const auditResult = await audit(jobKey, findings);
const auditResult = await audit(jobKey, chatId, findings);
setPhase(jobKey, "merging", "chat", undefined, part);
// SESSION-DELETION GUARD (wave 6): a chat deleted mid-run must never
// merge. The deletion route aborts this run first and runs its purge
@@ -1131,6 +1143,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
const t0 = performance.now();
const findings = await runNotesExtractor(
stageEngine("extractor", deps.extractor),
input.id,
prompt,
new Set(graph.nodes.map((n) => n.id)),
signal,
@@ -1191,7 +1204,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// per chunk so cross-source runs interleave fairly.
return withAuditLock(jobKey, "notes", input.filePath, async () => {
setPhase(jobKey, "auditing", "notes", input.filePath, part);
const auditResult = await audit(jobKey, findings);
const auditResult = await audit(jobKey, input.id, findings);
setPhase(jobKey, "merging", "notes", input.filePath, part);
const merged = mergeFindings(
id,
src/agent/subagents/pdf-reader.ts
+14 -1
View File
@@ -2,6 +2,7 @@ import type { AssistantMessage, Message, ToolResultMessage, Usage, Tool } from "
import { normalizeContext } from "@earendil-works/pi-ai";
import { Type } from "typebox";
import { piEngineFor, type PiEngine, type ThinkingLevel } from "../models.ts";
import { providerEventObserver, type ProviderEventObserver } from "../provider-events.ts";
import type { PdfResult, PdfToolInput } from "../../core/contracts.ts";
import { getHeadless, type HeadlessApi } from "./headless.ts";
import { pdfExtract, pdfInfo, pdfProbe, pdfSearch, type PdfSearchHit } from "./pdf-ops.ts";
@@ -66,6 +67,8 @@ export interface PdfReaderOptions {
onActivity?: (line: string) => void;
headless?: HeadlessApi;
maxIterations?: number;
/** Debug: raw provider-event recording key (LH_PROVIDER_EVENTS=1). */
chatId?: string;
}
export interface PdfReaderInput extends PdfToolInput {}
@@ -199,7 +202,15 @@ export async function runPdfReaderAgent(
try {
for (let iteration = 0; iteration < maxIterations; iteration++) {
const result = await streamOnce(engine, INSTRUCTIONS, messages, toolDeclarations, reasoning, opts.abortSignal);
const result = await streamOnce(
engine,
INSTRUCTIONS,
messages,
toolDeclarations,
reasoning,
opts.abortSignal,
opts.chatId ? providerEventObserver(opts.chatId, "job:pdfReader") : undefined,
);
// Money channel: each iteration is one model call.
opts.onUsage?.(result.usage);
messages = [...messages, result];
@@ -392,6 +403,7 @@ async function streamOnce(
tools: Tool[],
reasoning: ThinkingLevel | undefined,
abortSignal?: AbortSignal,
onProviderStreamEvent?: ProviderEventObserver,
): Promise<AssistantMessage> {
// 0.87.1: provider-facing contexts are TranscriptContexts — the prompt
// and tool declarations fold into a leading system message.
@@ -400,6 +412,7 @@ async function streamOnce(
reasoning: reasoning && reasoning !== "off" ? reasoning : undefined,
apiKey: process.env.OPENROUTER_API_KEY || undefined,
signal: abortSignal,
onProviderStreamEvent: onProviderStreamEvent as never,
} as never);
for await (const event of stream) {
if (event.type === "done") return event.message;
src/agent/subagents/pi-run.ts
+8
View File
@@ -1,6 +1,7 @@
import { Agent, type AgentEvent, type AgentMessage, type AgentTool } from "@earendil-works/pi-agent-core";
import type { AssistantMessage, ToolResultMessage, Usage } from "@earendil-works/pi-ai";
import { piEngineFor, type PiEngine } from "../models.ts";
import { providerEventObserver } from "../provider-events.ts";
import type { ModelRole } from "../../../config.ts";
/**
@@ -37,6 +38,10 @@ export interface SubagentRunOptions {
onAssistantMessage?: (message: AssistantMessage, step: number) => void;
/** Extra stop rule (extractor: stop after a clean submit). */
shouldStop?: (message: AssistantMessage, toolResults: ToolResultMessage[]) => boolean;
/** Debug: when set AND LH_PROVIDER_EVENTS=1, raw provider stream
* events append to data/debug/provider-events/<chatId>.jsonl (context
* "job:<role>"). Absent or env-off → no observer. */
chatId?: string;
}
export interface SubagentRunResult {
@@ -56,6 +61,8 @@ export function subagentEngine(role: ModelRole, override?: PiEngine): PiEngine {
export async function runSubagent(tools: AgentTool<any, any>[], opts: SubagentRunOptions): Promise<SubagentRunResult> {
const engine = subagentEngine(opts.role, opts.engine);
let turns = 0;
// Debug recorder (owner, 2026-10-04): raw provider events per job run.
const providerEvents = opts.chatId ? providerEventObserver(opts.chatId, `job:${opts.role}`) : undefined;
const agent = new Agent({
initialState: {
systemPrompt: opts.systemPrompt,
@@ -65,6 +72,7 @@ export async function runSubagent(tools: AgentTool<any, any>[], opts: SubagentRu
messages: opts.messages ?? [],
},
streamFn: engine.streamFn,
...(providerEvents ? { onProviderStreamEvent: providerEvents } : {}),
// 0.87.1: shouldStopAfterTurn → finishTurn (same call point; "end"
// exits before the steering poll, exactly the old boolean-true).
finishTurn: (turn) => {
src/agent/subagents/researcher.ts
+3
View File
@@ -71,6 +71,8 @@ export interface ResearcherRunOptions {
abortSignal?: AbortSignal;
onUsage?: (usage: Usage) => void;
onToolStart?: (toolName: string, args: unknown) => void;
/** Debug: raw provider-event recording key (LH_PROVIDER_EVENTS=1). */
chatId?: string;
}
/**
@@ -88,6 +90,7 @@ export async function runResearcherAgent(prompt: string, opts: ResearcherRunOpti
abortSignal: opts.abortSignal,
onUsage: opts.onUsage,
onToolStart: opts.onToolStart,
chatId: opts.chatId,
});
return text;
}
src/agent/subagents/runner.ts
+9
View File
@@ -106,6 +106,9 @@ interface LoopContext {
/** Fires when abortJob kills the run; abortSignal is passed into every
* model call so the spend stops mid-call, not at the next loop check. */
abortSignal: AbortSignal;
/** The job's chat (debug: raw provider-event recording under
* LH_PROVIDER_EVENTS=1 — forwarded into every subagent run). */
chatId?: string;
/** Fold one model call's usage into the job record (progress write). */
onUsage: (step: UsageStepLike) => void;
}
@@ -127,6 +130,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
const usageTotals = new Map<string, JobUsage>();
function makeLoopContext(jobId: string, engine: PiEngine): LoopContext {
const chatId = jobStore.getJob(jobId)?.chatId;
// Provider-qualified stamp: the engine's actual model — the only
// thing that ran (BYOK: no config-pin fallback stamps exist).
const model = `${engine.model.provider}:${engine.model.id}`;
@@ -137,6 +141,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
registerJobAbort(jobId, controller);
return {
abortSignal: controller.signal,
chatId,
onUsage: (step) => {
const next = accumulateJobUsage(usageTotals.get(jobId), model, step);
usageTotals.set(jobId, next);
@@ -355,6 +360,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
engine: model,
search: deps.searchProvider,
abortSignal: loop.abortSignal,
chatId: loop.chatId,
// Money channel: each agent step is one model call — fold its
// usage into the record as it completes (progress write).
onUsage: (usage) => loop.onUsage(piUsageToStep(usage)),
@@ -403,6 +409,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
{
engine: model,
abortSignal: loop.abortSignal,
chatId: loop.chatId,
onUsage: (usage) => loop.onUsage(piUsageToStep(usage)),
onActivity: (line) => jobStore.updateJob(jobId, { activity: line }),
},
@@ -438,6 +445,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
{
engine: model,
abortSignal: loop.abortSignal,
chatId: loop.chatId,
onUsage: (usage) => loop.onUsage(piUsageToStep(usage)),
onActivity: (line) => jobStore.updateJob(jobId, { activity: line }),
},
@@ -476,6 +484,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
{
engine: model,
abortSignal: loop.abortSignal,
chatId: loop.chatId,
onUsage: (usage) => loop.onUsage(piUsageToStep(usage)),
onActivity: (line) => jobStore.updateJob(jobId, { activity: line }),
},
src/agent/subagents/visual-maker.ts
+14 -1
View File
@@ -2,6 +2,7 @@ import type { AssistantMessage, Message, Tool, ToolResultMessage, Usage } from "
import { normalizeContext } from "@earendil-works/pi-ai";
import { Type } from "typebox";
import { piEngineFor, type PiEngine, type ThinkingLevel } from "../models.ts";
import { providerEventObserver, type ProviderEventObserver } from "../provider-events.ts";
import type { VisualResult } from "../../core/contracts.ts";
import { getHeadless, type HeadlessApi } from "./headless.ts";
@@ -44,6 +45,8 @@ export interface VisualMakerOptions {
/** Money channel: one call per loop iteration (streamSimple), folded
* into JobRecord.usage. */
onUsage?: (usage: Usage) => void;
/** Debug: raw provider-event recording key (LH_PROVIDER_EVENTS=1). */
chatId?: string;
}
const INSTRUCTIONS = `You are a visual maker — a diagram author + renderer for precise, spatial figures. You
@@ -155,7 +158,15 @@ export async function runVisualMaker(input: VisualMakerInput, opts: VisualMakerO
for (let iteration = 0; iteration < maxIterations; iteration++) {
// One model call per iteration (streamSimple, no agent loop — this
// loop owns tool execution and the image-in-the-loop injection).
const result = await streamOnce(engine, INSTRUCTIONS, messages, toolDeclarations, reasoning, opts.abortSignal);
const result = await streamOnce(
engine,
INSTRUCTIONS,
messages,
toolDeclarations,
reasoning,
opts.abortSignal,
opts.chatId ? providerEventObserver(opts.chatId, "job:visualMaker") : undefined,
);
// Money channel: each iteration is one model call.
opts.onUsage?.(result.usage);
messages = [...messages, result];
@@ -307,6 +318,7 @@ async function streamOnce(
tools: Tool[],
reasoning: ThinkingLevel | undefined,
abortSignal?: AbortSignal,
onProviderStreamEvent?: ProviderEventObserver,
): Promise<AssistantMessage> {
// 0.87.1: provider-facing contexts are TranscriptContexts — the prompt
// and tool declarations fold into a leading system message.
@@ -315,6 +327,7 @@ async function streamOnce(
reasoning: reasoning && reasoning !== "off" ? reasoning : undefined,
apiKey: process.env.OPENROUTER_API_KEY || undefined,
signal: abortSignal,
onProviderStreamEvent: onProviderStreamEvent as never,
} as never);
for await (const event of stream) {
if (event.type === "done") return event.message;