Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion src/cache.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import type { ExtractedContent } from "./extract/pipeline.ts";
import type { ExtractedContent } from "./extract/types.ts";

interface CacheEntry {
content: ExtractedContent;
Expand Down
101 changes: 101 additions & 0 deletions src/extract/fallbacks.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
import { executeWithFallback, type ExecutionHooks } from "../providers/execute.ts";
import type { FetchProvider } from "../providers/types.ts";
import { sanitizeError } from "../utils/errors.ts";
import type { ExtractedContent } from "./types.ts";

export interface ExtractionFallback {
name: string;
run(): Promise<ExtractedContent | null>;
}

interface CreateFetchProviderFallbackOptions {
url: string;
providers: readonly FetchProvider[];
signal?: AbortSignal;
hooks?: ExecutionHooks;
}

function isAbortError(error: unknown): boolean {
if (!error || typeof error !== "object") return false;
const name = (error as { name?: string }).name;
return name === "AbortError";
}

export function createFetchProviderFallback(
options: CreateFetchProviderFallbackOptions,
): ExtractionFallback {
const { url, providers, signal, hooks } = options;

return {
name: "fetch-provider",
async run(): Promise<ExtractedContent> {
if (providers.length === 0) {
throw new Error("No fetch providers configured");
}

try {
const { result, providerName } = await executeWithFallback({
candidates: providers.map((provider) => ({
name: provider.name,
execute: async () => {
const fetched = await provider.fetch(url, signal);
const text = fetched.text.trim();
if (!text) throw new Error(`${provider.name} returned empty content`);
return { ...fetched, text };
},
})),
operation: "fetch",
signal,
onSuccess: hooks?.onSuccess,
onFailure: hooks?.onFailure,
});

return {
text: result.text,
title: result.title,
url,
extractionChain: [`fetch-provider:${providerName}`],
chars: result.text.length,
truncated: false,
};
} catch (error) {
signal?.throwIfAborted();
if (isAbortError(error)) throw error;
// Rethrow ordinary errors so runExtractionFallbacks records :error.
// Surface a sanitized summary, never raw provider messages with credentials.
throw new Error(sanitizeError(error));
}
},
};
}

export async function runExtractionFallbacks(
fallbacks: readonly ExtractionFallback[],
chain: string[],
signal?: AbortSignal,
): Promise<ExtractedContent | null> {
for (const fallback of fallbacks) {
signal?.throwIfAborted();

let result: ExtractedContent | null;
try {
result = await fallback.run();
} catch (error) {
signal?.throwIfAborted();
if (isAbortError(error)) throw error;
chain.push(`${fallback.name}:error`);
continue;
}

signal?.throwIfAborted();
if (result == null) {
chain.push(`${fallback.name}:fail`);
continue;
}

chain.push(result.extractionChain[0]);
return result;
}

return null;
}
2 changes: 1 addition & 1 deletion src/extract/frames.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { execFileSync } from "node:child_process";
import type { VideoFrame } from "./pipeline.ts";
import type { VideoFrame } from "./types.ts";

// ---------------------------------------------------------------------------
// Constants
Expand Down
2 changes: 1 addition & 1 deletion src/extract/gemini-url-context.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { getApiKey, getVersionedApiBase } from "./gemini-api.ts";
import { isGeminiWebAvailable, queryWithCookies } from "./gemini-web.ts";
import type { ExtractedContent } from "./pipeline.ts";
import type { ExtractedContent } from "./types.ts";

const MIN_CONTENT_LENGTH = 100;

Expand Down
2 changes: 1 addition & 1 deletion src/extract/github.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import * as os from "node:os";
import * as nodePath from "node:path";
import { promisify } from "node:util";
import { DEFAULT_GITHUB_CONFIG, type GitHubConfig } from "../config.ts";
import type { ExtractedContent } from "./pipeline.ts";
import type { ExtractedContent } from "./types.ts";

const execFileAsync = promisify(execFile);

Expand Down
31 changes: 0 additions & 31 deletions src/extract/jina-reader.ts

This file was deleted.

141 changes: 73 additions & 68 deletions src/extract/pipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,17 @@ import {
rasterizePdfPages,
modelSupportsImages,
extractTextWithGeminiVision,
type PdfPageImage,
} from "./pdf-ocr.ts";
import { collectImageBlocks, type ExtractedContent, type ImageBlock } from "./types.ts";
import { createFetchProviderFallback, runExtractionFallbacks } from "./fallbacks.ts";
import type { FetchProvider } from "../providers/types.ts";
import type { ExecutionHooks } from "../providers/execute.ts";
import { sanitizeError } from "../utils/errors.ts";

export type { ExtractedContent, ImageBlock, VideoFrame } from "./types.ts";
export { collectImageBlocks } from "./types.ts";
import { getApiKey as getGeminiApiKey } from "./gemini-api.ts";
import { extractRsc } from "./rsc.ts";
import { extractViaJinaReader } from "./jina-reader.ts";
import { isYouTubeURL, extractYouTube, isYouTubeEnabled } from "./youtube.ts";
import { isVideoFile, extractVideo, isVideoEnabled } from "./video.ts";
import {
Expand All @@ -25,7 +31,6 @@ import {
} from "./frames.ts";
import { extractWithUrlContext, extractWithGeminiWeb } from "./gemini-url-context.ts";
import { activityMonitor } from "../monitor/activity-monitor.ts";

/**
* Error thrown when the HTTP fetch fails in a way that a different fetch
* provider might succeed (network errors, 5xx, 429).
Expand All @@ -36,51 +41,6 @@ export class RetryableExtractionError extends Error {
this.name = "RetryableExtractionError";
}
}

export interface VideoFrame {
data: string;
mimeType: string;
timestamp: string;
}

export interface ExtractedContent {
text: string;
title?: string;
url: string;
extractionChain: string[];
chars: number;
truncated: boolean;
contentId?: string;
thumbnail?: { data: string; mimeType: string };
frames?: VideoFrame[];
images?: PdfPageImage[];
duration?: number;
}

export type ImageBlock = { type: "image"; data: string; mimeType: string };

export function collectImageBlocks(extracted: ExtractedContent): ImageBlock[] {
const blocks: ImageBlock[] = [];
if (extracted.thumbnail) {
blocks.push({
type: "image",
data: extracted.thumbnail.data,
mimeType: extracted.thumbnail.mimeType,
});
}
if (extracted.frames) {
for (const frame of extracted.frames) {
blocks.push({ type: "image", data: frame.data, mimeType: frame.mimeType });
}
}
if (extracted.images) {
for (const img of extracted.images) {
blocks.push({ type: "image", data: img.data, mimeType: img.mimeType });
}
}
return blocks;
}

const BINARY_CONTENT_TYPES = [
"image/",
"audio/",
Expand All @@ -106,6 +66,20 @@ const MAX_REDIRECTS = 10;

const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]);

async function tryRegisteredFallback(
adapter: ReturnType<typeof createFetchProviderFallback>[],
chain: string[],
signal: AbortSignal | undefined,
): Promise<ExtractedContent | null> {
if (adapter.length === 0) return null;
signal?.throwIfAborted();
return runExtractionFallbacks(adapter, chain, signal);
}

function throwRetryableFromTransport(originalError: unknown): never {
throw new RetryableExtractionError(sanitizeError(originalError).slice(0, 120));
}

async function fetchValidated(
url: string,
init: RequestInit,
Expand Down Expand Up @@ -159,6 +133,7 @@ export async function probeUrl(
);
} catch (error) {
if (error instanceof SSRFError) throw error;
signal?.throwIfAborted();
return { skip: false };
}

Expand Down Expand Up @@ -197,13 +172,16 @@ export interface ExtractOptions {
frames?: number;
model?: string;
ctx?: import("@earendil-works/pi-coding-agent").ExtensionContext;
fetchProviders?: readonly FetchProvider[];
executionHooks?: ExecutionHooks;
}

export async function extractContent(
url: string,
signal?: AbortSignal,
options?: ExtractOptions,
): Promise<ExtractedContent> {
signal?.throwIfAborted();
const { github, ssrf, pdf, gemini } = loadMergedConfig(process.cwd());

// --- Frame extraction mode (timestamp/frames params present) ---
Expand Down Expand Up @@ -276,6 +254,20 @@ export async function extractContent(
// --- SSRF validation (after video/YouTube routing, before HTTP fetch) ---
validateUrl(url, { allowRanges: ssrf.allowRanges });

// Built after SSRF validation so transport and HTML paths share the same
// registry-wrapped candidates.
const fetchAdapter =
options?.fetchProviders && options.fetchProviders.length > 0
? [
createFetchProviderFallback({
url,
providers: options.fetchProviders,
signal,
hooks: options.executionHooks,
}),
]
: [];

// GitHub interception: try structured extraction before HTML scraping.
// Only fires for content URLs (blob, tree, root, raw).
// Returns null for non-content URLs (issues, PRs, etc.) -> falls through.
Expand Down Expand Up @@ -310,7 +302,10 @@ export async function extractContent(
} catch (err) {
activityMonitor.logError(fetchEntryId, err instanceof Error ? err.message : String(err));
if (err instanceof SSRFError) throw err;
throw new RetryableExtractionError(err instanceof Error ? err.message : String(err));
signal?.throwIfAborted();
const fallback = await tryRegisteredFallback(fetchAdapter, [], signal);
if (fallback) return fallback;
throwRetryableFromTransport(err);
}

// Cloudflare bot challenge: retry once with honest User-Agent
Expand All @@ -330,7 +325,10 @@ export async function extractContent(
} catch (err) {
activityMonitor.logError(retryEntryId, err instanceof Error ? err.message : String(err));
if (err instanceof SSRFError) throw err;
throw new RetryableExtractionError(err instanceof Error ? err.message : String(err));
signal?.throwIfAborted();
const fallback = await tryRegisteredFallback(fetchAdapter, [], signal);
if (fallback) return fallback;
throwRetryableFromTransport(err);
}
}

Expand All @@ -340,6 +338,8 @@ export async function extractContent(
const status = response.status;
// 429 and 5xx are retryable — a different provider might succeed
if (status === 429 || status >= 500) {
const fallback = await tryRegisteredFallback(fetchAdapter, [], signal);
if (fallback) return fallback;
throw new RetryableExtractionError(`HTTP ${status}: ${response.statusText}`);
}
throw new Error(`HTTP ${status}: ${response.statusText}`);
Expand Down Expand Up @@ -500,32 +500,37 @@ export async function extractContent(
}
chain.push("rsc:no-match");

// Tier 3: Jina Reader
const jinaText = await extractViaJinaReader(url, signal);
if (jinaText) {
chain.push("jina-reader");
return {
text: jinaText,
title: undefined,
url,
extractionChain: chain,
chars: jinaText.length,
truncated: false,
};
// Tier 3: registered fetch providers (skipped when no providers configured).
if (fetchAdapter.length === 0) {
chain.push("fetch-provider:skipped");
} else {
const providerResult = await runExtractionFallbacks(fetchAdapter, chain, signal);
if (providerResult) {
return { ...providerResult, extractionChain: chain };
}
}
chain.push("jina-reader:fail");

// Tier 4: Gemini HTML fallback
const geminiResult =
(await extractWithUrlContext(url, signal)) ?? (await extractWithGeminiWeb(url, signal));
// Tier 4: combined Gemini HTML adapter (URL context, then Web).
const geminiResult = await runExtractionFallbacks(
[
{
name: "gemini-html",
run: async () => {
const urlContextResult = await extractWithUrlContext(url, signal);
signal?.throwIfAborted();
return urlContextResult ?? (await extractWithGeminiWeb(url, signal));
},
},
],
chain,
signal,
);
if (geminiResult) {
chain.push(geminiResult.extractionChain[0]);
return {
...geminiResult,
extractionChain: chain,
};
}
chain.push("gemini-html:fail");

// Final fallback: raw text stripped of HTML
const rawText = body
Expand Down
Loading