diff --git a/crates/pi-natives/src/html.rs b/crates/pi-natives/src/html.rs index 397716503..b6e916233 100644 --- a/crates/pi-natives/src/html.rs +++ b/crates/pi-natives/src/html.rs @@ -1,6 +1,6 @@ //! HTML to Markdown conversion. -use html_to_markdown_rs::{convert, ConversionOptions, PreprocessingOptions, PreprocessingPreset}; +use html_to_markdown_rs::{ConversionOptions, PreprocessingOptions, PreprocessingPreset, convert}; use serde::Deserialize; use wasm_bindgen::prelude::*; @@ -13,7 +13,7 @@ pub struct HtmlToMarkdownOptions { pub clean_content: bool, /// Skip images during conversion. #[serde(default)] - pub skip_images: bool, + pub skip_images: bool, } /// Convert HTML to Markdown. @@ -29,13 +29,14 @@ pub fn html_to_markdown(html: &str, options: JsValue) -> Result let conversion_opts = ConversionOptions { skip_images: opts.skip_images, preprocessing: PreprocessingOptions { - enabled: opts.clean_content, - preset: PreprocessingPreset::Aggressive, + enabled: opts.clean_content, + preset: PreprocessingPreset::Aggressive, remove_navigation: true, - remove_forms: true, + remove_forms: true, }, ..Default::default() }; - convert(html, Some(conversion_opts)).map_err(|e| JsValue::from_str(&format!("Conversion error: {e}"))) + convert(html, Some(conversion_opts)) + .map_err(|e| JsValue::from_str(&format!("Conversion error: {e}"))) } diff --git a/crates/pi-natives/src/lib.rs b/crates/pi-natives/src/lib.rs index e0801d629..484340119 100644 --- a/crates/pi-natives/src/lib.rs +++ b/crates/pi-natives/src/lib.rs @@ -4,8 +4,8 @@ //! The JS side handles directory walking and file reading, then passes //! content here for matching. -pub mod html; pub mod highlight; +pub mod html; pub mod image; pub mod text; diff --git a/packages/coding-agent/src/tools/fetch.ts b/packages/coding-agent/src/tools/fetch.ts index f33b27109..aeff395f7 100644 --- a/packages/coding-agent/src/tools/fetch.ts +++ b/packages/coding-agent/src/tools/fetch.ts @@ -1,4 +1,3 @@ -import * as fs from "node:fs/promises"; import * as path from "node:path"; import type { AgentTool, AgentToolContext, AgentToolResult, AgentToolUpdateCallback } from "@oh-my-pi/pi-agent-core"; import { htmlToMarkdown } from "@oh-my-pi/pi-natives"; @@ -6,7 +5,6 @@ import type { Component } from "@oh-my-pi/pi-tui"; import { Text } from "@oh-my-pi/pi-tui"; import { ptree } from "@oh-my-pi/pi-utils"; import { type Static, Type } from "@sinclair/typebox"; -import { nanoid } from "nanoid"; import { parse as parseHtml } from "node-html-parser"; import { renderPromptTemplate } from "../config/prompt-templates"; import type { RenderResultOptions } from "../extensibility/custom-tools/types"; @@ -387,13 +385,12 @@ function parseFeedToMarkdown(content: string, maxItems = 10): string { } /** - * Render HTML to markdown using native WASM, jina, trafilatura, lynx, or html2text (in order of preference) + * Render HTML to markdown using native WASM, jina, trafilatura, lynx, or html-to-markdown (in order of preference) */ async function renderHtmlToText( url: string, html: string, timeout: number, - scratchDir: string, userSignal?: AbortSignal, ): Promise<{ content: string; ok: boolean; method: string }> { const signal = ptree.combineSignals(userSignal, timeout * 1000); @@ -405,17 +402,6 @@ async function renderHtmlToText( signal, }; - // Try native WASM converter first (fastest, no network/subprocess) - try { - const content = await htmlToMarkdown(html, { cleanContent: true }); - if (content.trim().length > 100 && !isLowQualityOutput(content)) { - return { content, ok: true, method: "native" }; - } - } catch { - // Native converter failed, continue to next method - signal?.throwIfAborted(); - } - // Try jina first (reader API) try { const jinaUrl = `https://r.jina.ai/${url}`; @@ -452,19 +438,15 @@ async function renderHtmlToText( } } - // Fall back to html2text (auto-install via uv/pip) - const html2text = await ensureTool("html2text", { signal, silent: true }); - if (html2text) { - const tmpFile = path.join(scratchDir, `omp-${nanoid()}.html`); - try { - await Bun.write(tmpFile, html); - const result = await ptree.exec([html2text, tmpFile], execOptions); - if (result.ok) { - return { content: result.stdout, ok: true, method: "html2text" }; - } - } finally { - void fs.rm(tmpFile, { force: true }).catch(() => {}); + // Fall back to native WASM converter (fastest, no network/subprocess) + try { + const content = await htmlToMarkdown(html, { cleanContent: true }); + if (content.trim().length > 100 && !isLowQualityOutput(content)) { + return { content, ok: true, method: "native" }; } + } catch { + // Native converter failed, continue to next method + signal?.throwIfAborted(); } return { content: "", ok: false, method: "none" }; } @@ -533,13 +515,7 @@ async function handleSpecialUrls(url: string, timeout: number, signal?: AbortSig /** * Main render function implementing the full pipeline */ -async function renderUrl( - url: string, - timeout: number, - raw: boolean, - scratchDir: string, - signal?: AbortSignal, -): Promise { +async function renderUrl(url: string, timeout: number, raw: boolean, signal?: AbortSignal): Promise { const notes: string[] = []; const fetchedAt = new Date().toISOString(); if (signal?.aborted) { @@ -778,7 +754,7 @@ async function renderUrl( } // Step 6: Render HTML with lynx or html2text - const htmlResult = await renderHtmlToText(finalUrl, rawContent, timeout, scratchDir, signal); + const htmlResult = await renderHtmlToText(finalUrl, rawContent, timeout, signal); if (!htmlResult.ok) { notes.push("html rendering failed (lynx/html2text unavailable)"); const output = finalizeOutput(rawContent); @@ -902,8 +878,7 @@ export class FetchTool implements AgentTool = { package: "markitdown", binaryName: "markitdown", }, - html2text: { - name: "html2text", - package: "html2text", - binaryName: "html2text", - }, trafilatura: { name: "trafilatura", package: "trafilatura", @@ -100,7 +95,7 @@ const PYTHON_TOOLS: Record = { }, }; -export type ToolName = "sd" | "sg" | "yt-dlp" | "markitdown" | "html2text" | "trafilatura"; +export type ToolName = "sd" | "sg" | "yt-dlp" | "markitdown" | "trafilatura"; // Get the path to a tool (system-wide or in our tools dir) export async function getToolPath(tool: ToolName): Promise { diff --git a/packages/natives/src/html/index.ts b/packages/natives/src/html/index.ts index d261d8a22..fe810976a 100644 --- a/packages/natives/src/html/index.ts +++ b/packages/natives/src/html/index.ts @@ -4,7 +4,7 @@ * Conversion happens in a worker thread to avoid blocking the main thread. */ -import { WorkerPool } from "../pool"; +import { type RequestOptions, WorkerPool } from "../pool"; import type { HtmlRequest, HtmlResponse, HtmlToMarkdownOptions } from "./types"; export type { HtmlToMarkdownOptions } from "./types"; @@ -22,12 +22,19 @@ const pool = new WorkerPool({ * @param options - Conversion options * @returns Markdown text */ -export async function htmlToMarkdown(html: string, options?: HtmlToMarkdownOptions): Promise { - const response = await pool.request>({ - type: "convert", - html, - options, - }); +export async function htmlToMarkdown( + html: string, + options?: HtmlToMarkdownOptions, + req?: RequestOptions, +): Promise { + const response = await pool.request>( + { + type: "convert", + html, + options, + }, + req, + ); return response.markdown; } diff --git a/packages/natives/src/image/index.ts b/packages/natives/src/image/index.ts index 9e662130c..e2f6a8e6e 100644 --- a/packages/natives/src/image/index.ts +++ b/packages/natives/src/image/index.ts @@ -49,9 +49,12 @@ export class PhotonImage { * The bytes are transferred to the worker (zero-copy). */ static async new_from_byteslice(bytes: Uint8Array): Promise { - const response = await pool.request>({ type: "load", bytes }, [ - bytes.buffer, - ]); + const response = await pool.request>( + { type: "load", bytes }, + { + transfer: [bytes.buffer], + }, + ); return new PhotonImage(response.handle, response.width, response.height); } diff --git a/packages/natives/src/index.ts b/packages/natives/src/index.ts index 23956910f..f024038be 100644 --- a/packages/natives/src/index.ts +++ b/packages/natives/src/index.ts @@ -171,4 +171,4 @@ export { // Worker Pool (shared infrastructure) // ============================================================================= -export { type BaseRequest, type BaseResponse, WorkerPool, type WorkerPoolOptions } from "./pool"; +export { type BaseRequest, type BaseResponse, type RequestOptions, WorkerPool, type WorkerPoolOptions } from "./pool"; diff --git a/packages/natives/src/pool.ts b/packages/natives/src/pool.ts index fef7c53cc..041ec59f5 100644 --- a/packages/natives/src/pool.ts +++ b/packages/natives/src/pool.ts @@ -27,6 +27,17 @@ export interface WorkerPoolOptions { idleTimeoutMs?: number; /** Timeout for worker initialization in ms (default: 10000). */ initTimeoutMs?: number; + /** Grace period after request timeout before force-terminating stuck workers (default: 5000). */ + stuckGracePeriodMs?: number; +} + +export interface RequestOptions { + /** Timeout for this request in ms. After this, the promise rejects but worker gets a grace period. */ + timeoutMs?: number; + /** Abort signal for this request. */ + signal?: AbortSignal; + /** Transfer list for postMessage. */ + transfer?: ArrayBufferLike[]; } interface PooledWorker { @@ -39,7 +50,8 @@ interface PooledWorker { interface PendingRequest { resolve: (result: T) => void; reject: (error: Error) => void; - timeout?: ReturnType; + worker?: PooledWorker; + dispose?: () => void; } /** @@ -62,26 +74,51 @@ export class WorkerPool { maxWorkers: options.maxWorkers ?? 4, idleTimeoutMs: options.idleTimeoutMs ?? 30_000, initTimeoutMs: options.initTimeoutMs ?? 10_000, + stuckGracePeriodMs: options.stuckGracePeriodMs ?? 5_000, }; } /** * Send a request to a worker and wait for the response. * Workers are acquired from the pool (or created if under limit). + * + * @param msg - Request message + * @param options - Request options (timeout, transfer) */ async request( msg: TReq | (Omit & { id?: number }), - transfer?: ArrayBufferLike[], + options?: RequestOptions, ): Promise { + const { timeoutMs, signal, transfer } = options ?? {}; + signal?.throwIfAborted(); + const worker = await this.#acquireWorker(); const id = msg.id ?? this.#nextRequestId++; const fullMsg = { ...msg, id } as TReq; const { promise, resolve, reject } = Promise.withResolvers(); - this.#pending.set(id, { + const pending: PendingRequest = { resolve: resolve as (result: TRes) => void, reject, - }); + worker, + }; + this.#pending.set(id, pending as PendingRequest); + + const onAbort = () => { + this.#handleRequestAbort(id, worker); + }; + + if (timeoutMs && timeoutMs > 0 && signal) { + const combined = AbortSignal.any([signal, AbortSignal.timeout(timeoutMs)]); + combined.addEventListener("abort", onAbort, { once: true }); + pending.dispose = () => combined.removeEventListener("abort", onAbort); + } else if (timeoutMs && timeoutMs > 0) { + const timer = setTimeout(onAbort, timeoutMs); + pending.dispose = () => clearTimeout(timer); + } else if (signal) { + signal.addEventListener("abort", onAbort, { once: true }); + pending.dispose = () => signal.removeEventListener("abort", onAbort); + } worker.currentRequestId = id; if (transfer) { @@ -109,7 +146,7 @@ export class WorkerPool { for (const pending of this.#pending.values()) { pending.reject(new Error("Worker pool terminated")); - if (pending.timeout) clearTimeout(pending.timeout); + void pending.dispose?.(); } this.#pending.clear(); } @@ -144,7 +181,7 @@ export class WorkerPool { if (!pending) return; this.#pending.delete(msg.id); - if (pending.timeout) clearTimeout(pending.timeout); + void pending.dispose?.(); if (msg.type === "error" && "error" in msg) { pending.reject(new Error(msg.error ?? "Unknown error")); @@ -163,11 +200,58 @@ export class WorkerPool { const pending = this.#pending.get(id); if (pending) { this.#pending.delete(id); - if (pending.timeout) clearTimeout(pending.timeout); + void pending.dispose?.(); pending.reject(error); } } + #handleRequestAbort(id: number, worker: PooledWorker): void { + const pending = this.#pending.get(id); + if (!pending) return; + + pending.dispose = undefined; + pending.reject(new Error("Request timeout")); + + if (this.#options.stuckGracePeriodMs > 0) { + const timer = setTimeout(() => { + this.#terminateStuckWorker(id, worker); + }, this.#options.stuckGracePeriodMs); + + pending.dispose = () => { + clearTimeout(timer); + }; + } + } + + #terminateStuckWorker(id: number, worker: PooledWorker): void { + const pending = this.#pending.get(id); + if (pending) { + this.#pending.delete(id); + void pending.dispose?.(); + } + + if (worker.currentRequestId !== id) return; + if (!this.#pool.includes(worker)) return; + + this.#removeWorker(worker); + + if (this.#pool.length === 0 && this.#waiters.length > 0) { + this.#replenishPool(); + } + } + + async #replenishPool(): Promise { + const worker = this.#createWorker(); + worker.busy = true; + this.#pool.push(worker); + try { + await this.#initializeWorker(worker); + this.#releaseWorker(worker); + } catch { + this.#removeWorker(worker); + } + } + #removeWorker(pooledWorker: PooledWorker): void { const idx = this.#pool.indexOf(pooledWorker); if (idx !== -1) { @@ -227,7 +311,7 @@ export class WorkerPool { this.#pending.set(id, { resolve: () => resolve(), reject, - timeout, + dispose: () => clearTimeout(timeout), } as PendingRequest); pooledWorker.worker.postMessage({ type: "init", id } satisfies BaseRequest);