smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhanced worker pool with timeout/abort handling and replaced html2text with WASM converter

Browse Source 
- Added `RequestOptions` interface to worker pool with support for `timeoutMs`, `signal`, and `transfer` properties.
- Added `stuckGracePeriodMs` configuration option to `WorkerPoolOptions` for graceful handling of stuck workers after timeout.
- Enhanced worker pool request method to accept `RequestOptions` object parameter instead of simple transfer array, enabling timeout and abort signal handling.
- Implemented comprehensive timeout and abort signal handling in worker pool with support for combined signals and grace period before force-terminating stuck workers.
- Removed html2text tool from coding agent in favor of native WASM html-to-markdown converter.
- Exported `RequestOptions` type from natives package public API.
This commit is contained in:
can1357
2026-01-29 23:35:02 +01:00
parent 0eaa18c15b
commit ddb617e3f5
8 changed files with 134 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pi-natives/src/html.rs
+7 -6
View File
@@ -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<String, JsValue>
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}")))
}
crates/pi-natives/src/lib.rs
+1 -1
View File
@@ -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;
packages/coding-agent/src/tools/fetch.ts
+12 -37
View File
@@ -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<RenderResult> {
async function renderUrl(url: string, timeout: number, raw: boolean, signal?: AbortSignal): Promise<RenderResult> {
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<typeof fetchSchema, FetchToolDetails
throw new ToolAbortError();
}
const scratchDir = this.session.getArtifactsDir?.() ?? this.session.cwd;
const result = await renderUrl(url, effectiveTimeout, raw, scratchDir, signal);
const result = await renderUrl(url, effectiveTimeout, raw, signal);
const truncation = truncateHead(result.content, { maxBytes: DEFAULT_MAX_BYTES, maxLines: DEFAULT_MAX_LINES });
const needsArtifact = truncation.truncated;
let artifactId: string | undefined;
packages/coding-agent/src/utils/tools-manager.ts
+1 -6
View File
@@ -88,11 +88,6 @@ const PYTHON_TOOLS: Record<string, PythonToolConfig> = {
package: "markitdown",
binaryName: "markitdown",
},
html2text: {
name: "html2text",
package: "html2text",
binaryName: "html2text",
},
trafilatura: {
name: "trafilatura",
package: "trafilatura",
@@ -100,7 +95,7 @@ const PYTHON_TOOLS: Record<string, PythonToolConfig> = {
},
};
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<string | null> {
packages/natives/src/html/index.ts
+14 -7
View File
@@ -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<HtmlRequest, HtmlResponse>({
* @param options - Conversion options
* @returns Markdown text
*/
export async function htmlToMarkdown(html: string, options?: HtmlToMarkdownOptions): Promise<string> {
const response = await pool.request<Extract<HtmlResponse, { type: "converted" }>>({
type: "convert",
html,
options,
});
export async function htmlToMarkdown(
html: string,
options?: HtmlToMarkdownOptions,
req?: RequestOptions,
): Promise<string> {
const response = await pool.request<Extract<HtmlResponse, { type: "converted" }>>(
{
type: "convert",
html,
options,
},
req,
);
return response.markdown;
}
packages/natives/src/image/index.ts
+6 -3
View File
@@ -49,9 +49,12 @@ export class PhotonImage {
* The bytes are transferred to the worker (zero-copy).
*/
static async new_from_byteslice(bytes: Uint8Array): Promise<PhotonImage> {
const response = await pool.request<Extract<ImageResponse, { type: "loaded" }>>({ type: "load", bytes }, [
bytes.buffer,
]);
const response = await pool.request<Extract<ImageResponse, { type: "loaded" }>>(
{ type: "load", bytes },
{
transfer: [bytes.buffer],
},
);
return new PhotonImage(response.handle, response.width, response.height);
}
packages/natives/src/index.ts
+1 -1
View File
@@ -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";
packages/natives/src/pool.ts
+92 -8
View File
@@ -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<T> {
resolve: (result: T) => void;
reject: (error: Error) => void;
timeout?: ReturnType<typeof setTimeout>;
worker?: PooledWorker;
dispose?: () => void;
}
/**
@@ -62,26 +74,51 @@ export class WorkerPool<TReq extends BaseRequest, TRes extends BaseResponse> {
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<T extends TRes = TRes>(
msg: TReq | (Omit<TReq, "id"> & { id?: number }),
transfer?: ArrayBufferLike[],
options?: RequestOptions,
): Promise<T> {
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<T>();
this.#pending.set(id, {
const pending: PendingRequest<T> = {
resolve: resolve as (result: TRes) => void,
reject,
});
worker,
};
this.#pending.set(id, pending as PendingRequest<TRes>);
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<TReq extends BaseRequest, TRes extends BaseResponse> {
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<TReq extends BaseRequest, TRes extends BaseResponse> {
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<TReq extends BaseRequest, TRes extends BaseResponse> {
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<void> {
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<TReq extends BaseRequest, TRes extends BaseResponse> {
this.#pending.set(id, {
resolve: () => resolve(),
reject,
timeout,
dispose: () => clearTimeout(timeout),
} as PendingRequest<TRes>);
pooledWorker.worker.postMessage({ type: "init", id } satisfies BaseRequest);