d6a0d05fc0
When OTEL_EXPORTER_OTLP_ENDPOINT (or _TRACES_ENDPOINT) is set, register a global OTLP/proto TracerProvider and switch on the agent loop's telemetry so its invoke_agent / chat / execute_tool spans reach a collector instead of a no-op tracer. Honors the standard OTEL_* env contract (OTEL_SERVICE_NAME, OTEL_SDK_DISABLED, OTEL_TRACES_EXPORTER=none) and the OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT capture toggle; no-op when no endpoint is configured. Uses the exporter-trace-otlp-proto 2.x line, which exports cleanly under Bun (the 1.x line deadlocks). Adds a flushTelemetryExport() helper and a periodic + signal-driven flush so long-lived processes surface spans. This makes the existing telemetry usable from hosts that spawn omp as a child process (e.g. an ACP harness), where an in-process provider registered by the parent can't reach the child process.
127 lines
4.1 KiB
TypeScript
127 lines
4.1 KiB
TypeScript
/**
|
|
* Print mode (single-shot): Send prompts, output result, exit.
|
|
*
|
|
* Used for:
|
|
* - `omp -p "prompt"` - text output
|
|
* - `omp --mode json "prompt"` - JSON event stream
|
|
*/
|
|
import type { AssistantMessage, ImageContent } from "@oh-my-pi/pi-ai";
|
|
import { logger, sanitizeText } from "@oh-my-pi/pi-utils";
|
|
import type { AgentSession } from "../session/agent-session";
|
|
import { isSilentAbort } from "../session/messages";
|
|
import { flushTelemetryExport } from "../telemetry-export";
|
|
import { initializeExtensions } from "./runtime-init";
|
|
|
|
/**
|
|
* Options for print mode.
|
|
*/
|
|
export interface PrintModeOptions {
|
|
/** Output mode: "text" for final response only, "json" for all events */
|
|
mode: "text" | "json";
|
|
/** Array of additional prompts to send after initialMessage */
|
|
messages?: string[];
|
|
/** First message to send (may contain @file content) */
|
|
initialMessage?: string;
|
|
/** Images to attach to the initial message */
|
|
initialImages?: ImageContent[];
|
|
}
|
|
|
|
/**
|
|
* Run in print (single-shot) mode.
|
|
* Sends prompts to the agent and outputs the result.
|
|
*/
|
|
export async function runPrintMode(session: AgentSession, options: PrintModeOptions): Promise<void> {
|
|
const { mode, messages = [], initialMessage, initialImages } = options;
|
|
|
|
// Emit session header for JSON mode
|
|
if (mode === "json") {
|
|
const header = session.sessionManager.getHeader();
|
|
if (header) {
|
|
process.stdout.write(`${JSON.stringify(header)}\n`);
|
|
}
|
|
}
|
|
// Set up extensions for print mode (no UI, no command context)
|
|
await initializeExtensions(session, {
|
|
reportSendError: (action, err) => {
|
|
process.stderr.write(
|
|
`Extension ${action === "extension_send" ? "sendMessage" : "sendUserMessage"} failed: ${err.message}\n`,
|
|
);
|
|
},
|
|
reportRuntimeError: err => {
|
|
process.stderr.write(`Extension error (${err.extensionPath}): ${err.error}\n`);
|
|
},
|
|
});
|
|
|
|
// Always subscribe to enable session persistence via _handleAgentEvent
|
|
session.subscribe(event => {
|
|
// In JSON mode, output all events
|
|
if (mode === "json") {
|
|
process.stdout.write(`${JSON.stringify(event)}\n`);
|
|
}
|
|
});
|
|
|
|
// Send initial message with attachments
|
|
if (initialMessage !== undefined) {
|
|
await logger.time("print:prompt:initial", () => session.prompt(initialMessage, { images: initialImages }));
|
|
}
|
|
|
|
// Send remaining messages
|
|
for (const message of messages) {
|
|
await logger.time("print:prompt:next", () => session.prompt(message));
|
|
}
|
|
|
|
// In text mode, output final response
|
|
if (mode === "text") {
|
|
const state = session.state;
|
|
const lastMessage = state.messages[state.messages.length - 1];
|
|
|
|
if (lastMessage?.role === "assistant") {
|
|
const assistantMsg = lastMessage as AssistantMessage;
|
|
|
|
// Check for error/aborted — skip silent-abort (plan-mode compaction transition)
|
|
if (
|
|
(assistantMsg.stopReason === "error" || assistantMsg.stopReason === "aborted") &&
|
|
!isSilentAbort(assistantMsg.errorMessage)
|
|
) {
|
|
const errorLine = sanitizeText(assistantMsg.errorMessage || `Request ${assistantMsg.stopReason}`);
|
|
// Flush before this hard exit — it bypasses the awaited postmortem.quit()
|
|
// in main(), and the postmortem `exit` handler can't await, so the error
|
|
// spans would otherwise stay buffered in the batch processor and drop.
|
|
await flushTelemetryExport();
|
|
const flushed = process.stderr.write(`${errorLine}\n`);
|
|
if (flushed) {
|
|
process.exit(1);
|
|
} else {
|
|
process.stderr.once("drain", () => process.exit(1));
|
|
}
|
|
}
|
|
|
|
if (
|
|
assistantMsg.errorMessage &&
|
|
assistantMsg.stopReason !== "error" &&
|
|
assistantMsg.stopReason !== "aborted"
|
|
) {
|
|
process.stderr.write(`${sanitizeText(assistantMsg.errorMessage)}\n`);
|
|
}
|
|
|
|
// Output text content
|
|
for (const content of assistantMsg.content) {
|
|
if (content.type === "text") {
|
|
process.stdout.write(`${sanitizeText(content.text)}\n`);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Ensure stdout is fully flushed before returning
|
|
// This prevents race conditions where the process exits before all output is written
|
|
await new Promise<void>((resolve, reject) => {
|
|
process.stdout.write("", err => {
|
|
if (err) reject(err);
|
|
else resolve();
|
|
});
|
|
});
|
|
|
|
await session.dispose();
|
|
}
|