From a2c12874152316450c80174cb11d7539c4b73820 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sun, 4 Oct 2026 20:21:18 +0300 Subject: [PATCH] provider-events: raw provider stream recorder (LH_PROVIDER_EVENTS=1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- config.ts | 3 + src/agent/provider-events.ts | 119 ++++++++++++++++++++++++++++ src/agent/subagents/demo-builder.ts | 3 + src/agent/subagents/extraction.ts | 19 ++++- src/agent/subagents/pdf-reader.ts | 15 +++- src/agent/subagents/pi-run.ts | 8 ++ src/agent/subagents/researcher.ts | 3 + src/agent/subagents/runner.ts | 9 +++ src/agent/subagents/visual-maker.ts | 15 +++- src/agent/teacher-agent.ts | 21 +++-- src/server/chat-handler.ts | 8 ++ tests/provider-events.test.ts | 117 +++++++++++++++++++++++++++ 12 files changed, 330 insertions(+), 10 deletions(-) create mode 100644 src/agent/provider-events.ts create mode 100644 tests/provider-events.test.ts diff --git a/config.ts b/config.ts index 12baa83..1fdc5c1 100644 --- a/config.ts +++ b/config.ts @@ -30,6 +30,9 @@ export const PATHS = { * Bounded self-contained snapshots (propositions + chains frozen at * creation), not append-only ledgers. */ practice: path.join(DATA_ROOT, "practice"), + /** Debug artifacts (env-gated): data/debug/provider-events/.jsonl + * raw provider stream recordings (src/agent/provider-events.ts). */ + debug: path.join(DATA_ROOT, "debug"), } as const; // ─── Models ──────────────────────────────────────────────────────────────── diff --git a/src/agent/provider-events.ts b/src/agent/provider-events.ts new file mode 100644 index 0000000..8add5f6 --- /dev/null +++ b/src/agent/provider-events.ts @@ -0,0 +1,119 @@ +/** + * provider-events — the raw-wire recorder (owner, 2026-10-04). + * + * pi-ai 1.0's onProviderStreamEvent observer fires for each PARSED + * provider stream event BEFORE pi normalizes it — the exact layer where + * our provider-quirk incident class lives and where nothing else in the + * app can see: GLM-flash truncating tool-call arguments mid-JSON + * (2026-09-25), DeepSeek resetting tool-call ids every response + * (2026-10-04, the Lacan session). This module turns that observer into + * an inspectable artifact. + * + * Gating (owner choice, 2026-10-04): env var only — + * LH_PROVIDER_EVENTS=1 npm run dev + * Default off: raw events carry full prompts (privacy) and unbounded + * volume (disk). No app surface; a debugging feature, not a feature. + * + * Artifact: one JSONL file per chat at + * data/debug/provider-events/.jsonl + * (data/ is disposable scratch — no retention system). Each line: + * {"t":,"ctx":"teacher|prober|job:","provider":…,"model":…,"event":} + * plus {"note":…} run markers. A per-file byte cap (CAPPED at ~20 MB) + * writes one truncation marker and stops — a runaway turn must not fill + * the disk. Event payloads are adapter-owned; a non-serializable event + * degrades to a type note, never throws (an observer exception must not + * kill a stream). + */ + +import * as fs from "node:fs"; +import path from "node:path"; +import { PATHS } from "../../config.ts"; + +/** Max bytes per chat's JSONL before recording stops (with a marker). */ +const CAP_BYTES = 20 * 1024 * 1024; + +/** Test seam (the toolTimeoutMs idiom): shrink the cap to exercise it. */ +export const providerEventsCap = { current: CAP_BYTES }; + +/** Test seam (the toolTimeoutMs idiom): redirect the output directory. */ +export const providerEventsDir = { current: path.join(PATHS.debug, "provider-events") }; + +/** The model shape the observer receives (structural — pi passes Model). */ +export interface ProviderEventModel { + provider?: string; + id?: string; +} + +export type ProviderEventObserver = (data: unknown, model: ProviderEventModel) => void; + +/** Per-file accounting so the cap marker writes exactly once. */ +const files = new Map(); + +function fileFor(chatId: string): string { + // chat ids are app-minted (chat_) — sanitize against any future + // id scheme putting path-hostile characters in here. + const safe = chatId.replace(/[^a-zA-Z0-9_-]/g, "_"); + return path.join(providerEventsDir.current, `${safe}.jsonl`); +} + +function writeLine(file: string, line: Record): void { + const state = files.get(file); + if (state?.capped) return; + const text = `${JSON.stringify({ t: Date.now(), ...line })}\n`; + try { + const before = state?.bytes ?? (fs.existsSync(file) ? fs.statSync(file).size : 0); + fs.mkdirSync(path.dirname(file), { recursive: true }); + fs.appendFileSync(file, text, "utf-8"); + const bytes = before + Buffer.byteLength(text); + if (bytes > providerEventsCap.current) { + files.set(file, { bytes, capped: true }); + fs.appendFileSync(file, `${JSON.stringify({ t: Date.now(), note: `cap ${providerEventsCap.current} bytes reached — recording stopped` })}\n`, "utf-8"); + } else { + files.set(file, { bytes, capped: false }); + } + } catch { + // The debug recorder must never take a run down with it. + } +} + +/** Encode an adapter-owned event safely: JSON when it cooperates, a type + * note when it does not (cyclic/BigInt payloads). */ +function encodeEvent(data: unknown): unknown { + try { + JSON.stringify(data); + return data; + } catch { + return { __nonSerializable: typeof data }; + } +} + +/** True when recording is on (read per call — toggling needs no restart + * beyond the next turn's agent construction). */ +export function providerEventsEnabled(): boolean { + return !!process.env.LH_PROVIDER_EVENTS; +} + +/** + * The observer for one agent run, or undefined when disabled. Pass the + * result as the Agent's onProviderStreamEvent (both construction sites — + * teacher-agent's baseAgentOptions and pi-run's runSubagent — build it + * from a chatId in hand). + */ +export function providerEventObserver(chatId: string, context: string): ProviderEventObserver | undefined { + if (!providerEventsEnabled()) return undefined; + return (data, model) => { + writeLine(fileFor(chatId), { + ctx: context, + provider: model?.provider, + model: model?.id, + event: encodeEvent(data), + }); + }; +} + +/** A run marker (start/end/abort notes) so events correlate with the + * persisted record; no-op when disabled. */ +export function providerEventNote(chatId: string, context: string, text: string): void { + if (!providerEventsEnabled()) return; + writeLine(fileFor(chatId), { ctx: context, note: text }); +} diff --git a/src/agent/subagents/demo-builder.ts b/src/agent/subagents/demo-builder.ts index 1f434d2..8788cfb 100644 --- a/src/agent/subagents/demo-builder.ts +++ b/src/agent/subagents/demo-builder.ts @@ -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) { diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index d0c3653..3f3a907 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -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, signal: AbortSignal | undefined, @@ -382,6 +383,9 @@ function runExtractor( // over the MIXED namespace: local ids + existing graph ids. return runSubmitAgent({ 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, signal: AbortSignal | undefined, @@ -408,6 +413,8 @@ function runNotesExtractor( ): Promise { return runSubmitAgent({ 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 { 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 { + async function audit(jobKey: string, sourceId: string, findings: SessionFindings): Promise { 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, diff --git a/src/agent/subagents/pdf-reader.ts b/src/agent/subagents/pdf-reader.ts index 4648f3e..5aad0ec 100644 --- a/src/agent/subagents/pdf-reader.ts +++ b/src/agent/subagents/pdf-reader.ts @@ -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 { // 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; diff --git a/src/agent/subagents/pi-run.ts b/src/agent/subagents/pi-run.ts index 8236aae..796ed17 100644 --- a/src/agent/subagents/pi-run.ts +++ b/src/agent/subagents/pi-run.ts @@ -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/.jsonl (context + * "job:"). 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[], opts: SubagentRunOptions): Promise { 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[], 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) => { diff --git a/src/agent/subagents/researcher.ts b/src/agent/subagents/researcher.ts index 7557aba..9d60f39 100644 --- a/src/agent/subagents/researcher.ts +++ b/src/agent/subagents/researcher.ts @@ -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; } diff --git a/src/agent/subagents/runner.ts b/src/agent/subagents/runner.ts index 9621091..5beb6f1 100644 --- a/src/agent/subagents/runner.ts +++ b/src/agent/subagents/runner.ts @@ -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(); 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 }), }, diff --git a/src/agent/subagents/visual-maker.ts b/src/agent/subagents/visual-maker.ts index dd926e9..6aede8b 100644 --- a/src/agent/subagents/visual-maker.ts +++ b/src/agent/subagents/visual-maker.ts @@ -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 { // 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; diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index aa651e8..0b4001e 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -1,6 +1,7 @@ import { Agent } from "@earendil-works/pi-agent-core"; import type { AgentEvent, AgentMessage, AgentTool } from "@earendil-works/pi-agent-core"; -import { questionResult, ToolInputError } from "./tool-execute.ts"; +import { invalidInputResult, questionResult } from "./tool-execute.ts"; +import { providerEventObserver } from "./provider-events.ts"; import type { Usage } from "@earendil-works/pi-ai"; import { PATHS } from "../../config.ts"; import { piEngineFor, type PiEngine } from "./models.ts"; @@ -111,6 +112,11 @@ export interface AgentBuildOptions { * fresh per turn, honoring enabled overrides (switch on + file * exists); absent → the cached pure-source composition. */ promptSwitches?: Record; + /** Debug: when set AND LH_PROVIDER_EVENTS=1, raw provider stream events + * (pre-normalization) append to data/debug/provider-events/ + * .jsonl via providerEventObserver — the GLM-truncation / + * DeepSeek-id incident layer. Absent or env-off → no observer. */ + chatId?: string; } /** @@ -158,6 +164,10 @@ function baseAgentOptions(kind: AgentKind, opts: AgentBuildOptions, tools: Agent const systemPrompt = opts.promptSwitches ? composeInstructions(kind, opts.promptSwitches) : composedInstructions[kind]!; + // Debug recorder (owner, 2026-10-04): raw provider events when + // LH_PROVIDER_EVENTS=1 and the caller handed a chatId; undefined is + // the normal no-observer path (tests, env-off). + const providerEvents = opts.chatId ? providerEventObserver(opts.chatId, kind) : undefined; const agent = new Agent({ initialState: { systemPrompt, @@ -167,6 +177,7 @@ function baseAgentOptions(kind: AgentKind, opts: AgentBuildOptions, tools: Agent messages: opts.messages ?? [], }, streamFn: engine.streamFn, + ...(providerEvents ? { onProviderStreamEvent: providerEvents } : {}), // 0.87.1: shouldStopAfterTurn → finishTurn. Same call point (after // tool results attach, before turn_end) and `{ action: "end" }` still // exits BEFORE the steering poll — the parked-marker stranding our @@ -232,9 +243,9 @@ export function createTeacherAgent( } /** The bare teach question set (tests, bare agents): quiz with the shared - * authoring refinement (quizInputProblems, run in execute — throws - * ToolInputError, which pi's loop converts to the error tool result the - * model retries on) but NO per-session nodeId check (no record to read), + * authoring refinement (quizInputProblems, run in execute — returns an + * invalidInputResult isError result the model retries on, pi 1.0 + * convention) but NO per-session nodeId check (no record to read), * plus ask. */ export function bareQuestionTools(): Record> { // A typed const (not an inline literal): the specific AgentTool typing @@ -248,7 +259,7 @@ export function bareQuestionTools(): Record> { executionMode: "sequential", execute: async (_toolCallId, input) => { const problems = quizInputProblems(input); - if (problems.length > 0) throw new ToolInputError("quiz", problems); + if (problems.length > 0) return invalidInputResult("quiz", problems); return questionResult(); }, }; diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index d8f2938..5e562f4 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -18,6 +18,7 @@ import { } from "./jobs.ts"; import { piTurnResponse, runPiTurn } from "./pi/ui-stream.ts"; import { collectToolCallIds } from "./pi/call-ids.ts"; +import { providerEventNote } from "../agent/provider-events.ts"; import { effectiveSettings, stores } from "./stores.ts"; import { searchAvailable, searchProvider } from "./search.ts"; import { providerKeyPresent } from "./keys.ts"; @@ -321,6 +322,9 @@ export async function handleChatInput( messages: asAgentMessages(merged), onUsage: collectUsage, toolToggles: effective.agent.tools, + // Debug (owner, 2026-10-04): teacher/prober raw provider events under + // LH_PROVIDER_EVENTS=1 (provider-events.ts); no-op otherwise. + chatId, // The prompt-override switches (P2, 2026-09-19): same resolution // point as the tool delta — the agent layer composes the system // prompt fresh per turn, swapping in any enabled override segments. @@ -378,6 +382,10 @@ export async function handleChatInput( // coalescing). Settlement still cannot beat the flow below: it needs // a streamed response, which never happens inside the constructor. registerLiveAgent(chatId, constructed); + // Debug recorder run marker (owner, 2026-10-04): correlates the + // JSONL event stream with the record's turns; no-op unless + // LH_PROVIDER_EVENTS=1. + providerEventNote(chatId, "turn", `run start (base: ${merged.length} messages)`); const handle = runPiTurn({ agent: constructed, // THE LACAN FIX (2026-10-04): seed the uniquifier with every diff --git a/tests/provider-events.test.ts b/tests/provider-events.test.ts new file mode 100644 index 0000000..960b06b --- /dev/null +++ b/tests/provider-events.test.ts @@ -0,0 +1,117 @@ +/** + * provider-events — the raw-wire recorder pins (owner, 2026-10-04). + * + * The recorder is env-gated debug tooling (LH_PROVIDER_EVENTS=1); these + * pins hold its mechanics: off = zero surface (no observer, no file), + * on = one JSONL per chat under data/debug/provider-events/, safe + * degradation on non-serializable adapter payloads, and the byte cap + * that stops a runaway turn from filling the disk. + */ + +import test from "node:test"; +import assert from "node:assert/strict"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { PATHS } from "../config.ts"; +import { + providerEventNote, + providerEventObserver, + providerEventsCap, + providerEventsDir, + providerEventsEnabled, +} from "../src/agent/provider-events.ts"; + +const ENV = "LH_PROVIDER_EVENTS"; +const REAL_DIR = path.join(PATHS.debug, "provider-events"); +const REAL_CAP = 20 * 1024 * 1024; + +function tmpDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "lh-provider-events-")); +} + +function readLines(dir: string, chatId: string): Array> { + const file = path.join(dir, `${chatId}.jsonl`); + return fs + .readFileSync(file, "utf-8") + .trim() + .split("\n") + .map((l) => JSON.parse(l) as Record); +} + +/** Common harness: redirect the dir, enable the env, restore both. */ +function withRecorder(t: { after: (fn: () => void) => void }): string { + const dir = tmpDir(); + providerEventsDir.current = dir; + process.env[ENV] = "1"; + t.after(() => { + delete process.env[ENV]; + providerEventsDir.current = REAL_DIR; + }); + return dir; +} + +test("env off: no observer, no note, no file — zero surface", (t) => { + const dir = tmpDir(); + providerEventsDir.current = dir; + delete process.env[ENV]; + t.after(() => { + delete process.env[ENV]; + providerEventsDir.current = REAL_DIR; + }); + + assert.equal(providerEventsEnabled(), false); + assert.equal(providerEventObserver("chat_x", "teacher"), undefined); + providerEventNote("chat_x", "teacher", "run start"); // must be a silent no-op + assert.equal(fs.existsSync(path.join(dir, "chat_x.jsonl")), false, "nothing written while off"); +}); + +test("env on: observer writes pinned-shape JSONL lines", (t) => { + const dir = withRecorder(t); + + const observe = providerEventObserver("chat_y", "teacher"); + assert.ok(observe, "enabled → observer exists"); + observe!({ type: "raw", delta: "hello" }, { provider: "openrouter", id: "z-ai/glm-5.3-flash" }); + providerEventNote("chat_y", "teacher", "run end"); + + const lines = readLines(dir, "chat_y"); + assert.equal(lines.length, 2); + assert.deepEqual( + { ...lines[0], t: 0 }, + { t: 0, ctx: "teacher", provider: "openrouter", model: "z-ai/glm-5.3-flash", event: { type: "raw", delta: "hello" } }, + "event line: {t, ctx, provider, model, event}", + ); + assert.match(String(lines[1]!.note), /run end/); +}); + +test("non-serializable adapter payloads degrade to a type note, never throw", (t) => { + const dir = withRecorder(t); + + const cyclic: Record = { type: "chunk" }; + cyclic.self = cyclic; + const observe = providerEventObserver("chat_c", "prober"); + observe!(cyclic, { provider: "anthropic", id: "claude-fable-5" }); + + const [line] = readLines(dir, "chat_c"); + assert.deepEqual(line!.event, { __nonSerializable: "object" }); +}); + +test("the byte cap writes one marker and stops recording", (t) => { + const dir = withRecorder(t); + // Deterministic sizing: cap 250, each event line ≈ 85B base + 200B pad + // ≈ 285B → the FIRST line already crosses; every later line is dropped. + providerEventsCap.current = 250; + t.after(() => { + providerEventsCap.current = REAL_CAP; + }); + + const observe = providerEventObserver("chat_cap", "teacher"); + assert.ok(observe); + observe!({ type: "a", pad: "x".repeat(200) }, { provider: "p", id: "m" }); // crosses → written + marker + observe!({ type: "b" }, { provider: "p", id: "m" }); // capped: dropped + + const lines = readLines(dir, "chat_cap"); + assert.equal(lines.length, 2, "one event + one cap marker"); + assert.match(String(lines.at(-1)!.note), /cap 250 bytes reached/); + assert.equal(lines.some((l) => (l.event as { type?: string })?.type === "b"), false, "post-cap event dropped"); +});