WFDASH-WEBAPP/src/lib/server/quotes-ai-processor.ts
SJCoetz 63048e51c9 feat: add AI quote processing pipeline and admin settings UI
Add complete end-to-end AI-powered quote extraction workflow:
- Disable PB SDK auto-cancellation to support parallel batch requests
- Add required runtime dependencies for AI processing and document parsing
- Implement AI settings backend, operator UI, and REST API routes
- Add quote ingestion, insight population, and manual re-run AI endpoints
- Build service layers for quote items, AI settings, and structured item insights
- Fix attachment upload handling and quote item history blob parsing
- Add supporting migrations, smoke tests, and shared utility functions
- Export previously private helper functions for cross-service reuse
2026-08-09 16:17:29 +02:00

719 lines
27 KiB
TypeScript

/**
* AI processor for the email → quote pipeline (the "slow" half).
*
* Picks quotes with `ai_status = "pending" | "failed"` from PB, runs
* Claude against the email body + extracted attachment text, writes
* `quote_items` rows, and flips the lifecycle flag. Designed to be
* driven by:
*
* - the cron endpoint `/api/v1/quotes/process-pending` (background)
* - the manual retry endpoint `/api/v1/quotes/[id]/re-run-ai` (UI)
*
* Why we own AI here (vs. n8n)
* ─────────────────────────────
* - Single source of truth for the extraction prompt. A prompt
* tweak ships as one commit, no workflow redeploy.
* - Deterministic attachment text extraction lives next to the
* LLM call, so the same extraction logic runs in tests.
* - PB has direct row locking via `ai_status = "processing"` —
* no need for the n8n-level claim pattern.
*
* Concurrency / claim safety
* ──────────────────────────
* The processor's main query is
*
* ai_status = "pending" || ai_status = "failed"
*
* Two concurrent runs (cron + manual retry, or two cron ticks) could
* pick the same row. We defend against that with an atomic claim:
*
* UPDATE quotes
* SET ai_status = "processing"
* WHERE id = :id AND ai_status IN ("pending", "failed")
*
* If the UPDATE doesn't take (PB's `update` returns 404 when the
* filter matches zero rows), another worker won the race and we
* skip. The next cron tick will find another quote (or re-try this
* one if it never flips back).
*
* Failure handling
* ────────────────
* - Claude API error / network blip → ai_status="failed", write
* ai_error, return. The cron will retry next tick.
* - Claude returns no items → ai_status="complete" with zero items.
* Legitimate outcome ("thanks for your quote" replies).
* - Unexpected error → ai_status="failed", ai_error=stack/message.
*
* What this does NOT do
* ─────────────────────
* - Pricing. Claude is told to NEVER invent cost / selling prices.
* All items are written with `cost_price = 0`, `selling_price = 0`.
* - Supplier matching. Supplier assignment is the office's job,
* via the existing sprint page.
* - Quote status flips. `quotes.status` stays at `"new"` so the
* existing UI workflow (claim → process → approve → email)
* continues unchanged.
*/
import Anthropic from "@anthropic-ai/sdk";
import type { ContentBlockParam } from "@anthropic-ai/sdk/resources/messages/messages";
import { getAdminPocketBase } from "./pocketbase-admin";
import { extractAttachmentText } from "./attachment-extractor";
import type { RawAttachment } from "./attachment-extractor";
import type { Quote, QuoteItem, Customer } from "@/lib/types";
import { findOrCreateCompany } from "./quotes-service";
import { getActiveAiConfig, AiDisabledError, getAiSettings } from "./ai-settings-service";
import { DEFAULT_QUOTES_AI_SYSTEM_PROMPT } from "./ai-system-prompt";
const QUOTES_COLLECTION = "quotes";
const QUOTE_ITEMS_COLLECTION = "quote_items";
const QUOTE_ATTACHMENTS_COLLECTION = "quote_attachments";
/** Model + max_tokens now come from the `ai_settings` collection
* (operator-controlled via /settings/ai). See ai-settings-service.ts
* for the resolution + cache layer. We no longer hard-code the model
* here so the operator can switch providers (Anthropic ↔ MiniMax)
* without redeploying. */
/** Hard cap on attachments pulled per quote. Real emails rarely have
* more than 10; the cap exists so a 50-attachment junk mail doesn't
* blow the context window. Surplus attachments are skipped with a
* console.warn. */
const MAX_ATTACHMENTS = 10;
/** Token budget for the body + attachment text concatenated into the
* prompt. ~200k is Sonnet 4.5's input limit; we cap well below to
* leave room for the system prompt + tool schema + output. ~120k
* chars ≈ 30k tokens. Conservative for body + 10 docs of ~12k each. */
const MAX_TEXT_CHARS = 120_000;
interface ProcessOneInput {
/** Pass an existing PB instance (the cron batch passes one shared
* authed instance; manual retry creates its own). */
pb?: Awaited<ReturnType<typeof getAdminPocketBase>>;
}
export interface ProcessOneResult {
quoteId: string;
/** "processed" = items written. "skipped" = claim lost to a
* concurrent worker. "failed" = error captured to ai_error. */
outcome: "processed" | "skipped" | "failed";
itemsWritten: number;
/** Truncated error message; empty on success / skip. */
error: string;
}
/**
* Process one specific quote (manual retry path) or the next pending
* one (cron path). When `quoteId` is omitted, picks the oldest pending
* quote. Returns a structured result for the caller to log / surface.
*/
export async function processOneQuote(
input: ProcessOneInput & { quoteId?: string }
): Promise<ProcessOneResult> {
const pb = input.pb ?? (await getAdminPocketBase());
let quote: Quote | null = null;
if (input.quoteId) {
try {
quote = await pb.collection<Quote>(QUOTES_COLLECTION).getOne(input.quoteId);
} catch {
return {
quoteId: input.quoteId,
outcome: "failed",
itemsWritten: 0,
error: "quote not found",
};
}
} else {
quote = await pickNextPendingQuote(pb);
}
if (!quote) {
return { quoteId: "", outcome: "skipped", itemsWritten: 0, error: "" };
}
// Already-processed quotes can be re-claimed (manual retry); guard
// is on the claim UPDATE below, not here.
return processClaimedQuote(pb, quote);
}
/**
* Batch path: claim + process up to `limit` quotes sequentially.
* Each is independent — one failure doesn't block the next.
*
* Returns one entry per attempted quote, in the order they were
* processed. Cron surfaces the count + error summary in its response.
*/
export async function processPendingQuotes(
input: { limit?: number } = {}
): Promise<ProcessOneResult[]> {
const limit = Math.max(1, Math.min(20, input.limit ?? 5));
const pb = await getAdminPocketBase();
const results: ProcessOneResult[] = [];
for (let i = 0; i < limit; i++) {
const quote = await pickNextPendingQuote(pb);
if (!quote) break;
results.push(await processClaimedQuote(pb, quote));
// Stop early if we hit a hard failure (other quotes are unlikely
// to succeed if Claude is throwing). The cron will retry next tick.
const last = results[results.length - 1];
if (last.outcome === "failed" && last.error.includes("Anthropic")) break;
}
return results;
}
// ──────────────────────────────────────────────────────────────────────────────
// Claim + lock
// ──────────────────────────────────────────────────────────────────────────────
async function pickNextPendingQuote(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>
): Promise<Quote | null> {
try {
return await pb.collection<Quote>(QUOTES_COLLECTION).getFirstListItem(
`ai_status = "pending" || ai_status = "failed"`,
{ sort: "ai_processed_at" }, // nulls first in PB
);
} catch {
return null;
}
}
/**
* Atomic claim: set ai_status="processing" iff still pending/failed.
* PB's update() applies the filter to the WHERE clause, so a losing
* race returns 404 (which we treat as "skipped"). The claim also
* stamps ai_processed_at=now so the operator UI shows a "started at"
* badge during long-running calls.
*/
async function claimQuote(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>,
quote: Quote
): Promise<Quote | null> {
try {
return await pb.collection(QUOTES_COLLECTION).update(quote.id, {
ai_status: "processing",
ai_processed_at: new Date().toISOString(),
ai_error: "",
});
} catch {
// Lost the race or quote deleted; either way, skip.
return null;
}
}
/**
* Run the extraction + write items + flip status. Always flips the
* status back to either `complete` or `failed` on exit (no way to
* leave a quote stuck in `processing`).
*/
async function processClaimedQuote(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>,
quote: Quote
): Promise<ProcessOneResult> {
const claim = await claimQuote(pb, quote);
if (!claim) {
return { quoteId: quote.id, outcome: "skipped", itemsWritten: 0, error: "" };
}
try {
const result = await runExtraction(pb, claim);
await pb.collection(QUOTES_COLLECTION).update(quote.id, {
ai_status: "complete",
ai_error: "",
ai_processed_at: new Date().toISOString(),
});
return {
quoteId: quote.id,
outcome: "processed",
itemsWritten: result.itemsWritten,
error: "",
};
} catch (e) {
// AiDisabledError means the operator has flipped the killswitch
// in /settings/ai. We still want ai_status to land on a useful
// value so the operator UI surfaces the "AI off" state, not the
// raw stack. Map to ai_error="ai_disabled" + ai_status="failed"
// (failed, not complete — we did NOT run extraction).
const isDisabled = e instanceof AiDisabledError;
const msg = (
isDisabled ? "ai_disabled" : e instanceof Error ? e.message : String(e)
).slice(0, 4900);
const stack = e instanceof Error ? (e.stack || "").split("\n").slice(0, 5).join(" | ") : "";
console.error(`[quotes-ai-processor] failed for ${quote.id}: ${msg}`);
if (stack) {
console.error(`[quotes-ai-processor] stack: ${stack}`);
}
try {
await pb.collection(QUOTES_COLLECTION).update(quote.id, {
ai_status: "failed",
ai_error: msg,
ai_processed_at: new Date().toISOString(),
});
} catch (innerErr) {
console.error(
`[quotes-ai-processor] couldn't even write ai_error back:`,
innerErr,
);
}
return { quoteId: quote.id, outcome: "failed", itemsWritten: 0, error: msg };
}
}
// ──────────────────────────────────────────────────────────────────────────────
// Extraction
// ──────────────────────────────────────────────────────────────────────────────
interface ExtractionResult {
itemsWritten: number;
}
/**
* Pulls the email body + attachment text from PB, calls Claude, and
* writes the parsed items. Throws on any unrecoverable error so the
* outer try/catch can flip ai_status="failed".
*/
async function runExtraction(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>,
quote: Quote
): Promise<ExtractionResult> {
// ── 1. Pull attachments (file bytes + metadata). ─────────────────
const atts = await loadQuoteAttachments(pb, quote.id);
// ── 2. Extract deterministic text from each. ────────────────────
// Two parallel arrays: text attachments go into the prompt body,
// image attachments are sent as inline image content blocks.
const textParts: string[] = [];
const imageParts: Array<{
filename: string;
contentType: string;
base64: string;
}> = [];
for (const att of atts) {
const extracted = await extractAttachmentText(att);
if (extracted.text !== null) {
const body = `[Attachment: ${extracted.filename}]\n${extracted.text}`;
textParts.push(body);
} else if (
extracted.kind === "image" &&
isSupportedImageMime(extracted.contentType)
) {
imageParts.push({
filename: extracted.filename,
contentType: extracted.contentType,
base64: att.bytes.toString("base64"),
});
} else if (extracted.kind === "pdf-thin") {
// Render path failed (extractor couldn't render the PDF to an
// image). We still have the raw bytes; the office can manually
// re-process this quote once the underlying issue is fixed.
console.warn(
`[quotes-ai-processor] ${quote.id}: scanned PDF "${extracted.filename}" skipped (render failed)`,
);
} else {
console.warn(
`[quotes-ai-processor] ${quote.id}: skipping unsupported attachment ${extracted.filename} (kind=${extracted.kind})`,
);
}
}
// Trim text to fit the context budget. Body is always included;
// attachments get the remainder.
const subject = quote.email_subject || quote.description || "";
const emailBody = quote.email_body || "";
const header =
`Subject: ${subject}\n` +
`From: ${quote.email_sender || "(unknown sender)"}\n\n`;
const fullText = header + emailBody + "\n\n" + textParts.join("\n\n---\n\n");
// Bare-photo hint: when the email body is empty/whitespace AND the
// only content is attached images (no text attachments either), tell
// Claude to apply rule 4a and produce a placeholder item. Without
// this nudge, Claude obeys rule 4 and returns 0 items, which leaves
// the operator with nothing to edit.
const isBarePhoto =
emailBody.trim() === "" &&
imageParts.length > 0 &&
textParts.length === 0;
const barePhotoHint = isBarePhoto
? "The email body is empty — the only content is the attached image(s) below. Apply rule 4a and return ONE placeholder quote_item beginning with 'what I see:'.\n\n"
: "";
const truncated = fullText.length > MAX_TEXT_CHARS;
const textForPrompt = (barePhotoHint + fullText).length > MAX_TEXT_CHARS
? fullText.slice(0, MAX_TEXT_CHARS) + "\n\n[... truncated for context budget ...]"
: barePhotoHint + fullText;
if (isBarePhoto) {
console.log(
`[quotes-ai-processor] ${quote.id}: bare-photo case (body empty, ${imageParts.length} image(s)) — applying rule 4a hint`,
);
}
if (truncated) {
console.warn(
`[quotes-ai-processor] ${quote.id}: text truncated to ${MAX_TEXT_CHARS} chars`,
);
}
// ── 3. Build the Claude call. ───────────────────────────────────
// Resolve provider/model/api_key/baseURL from the ai_settings row.
// Throws AiDisabledError if the killswitch is off, or a plain Error
// if the config is incomplete — both are caught by the outer
// try/catch and translated to ai_error.
const aiConfig = await getActiveAiConfig();
const client = newAnthropic(aiConfig);
const userContent = buildUserContent(textForPrompt, imageParts);
const response = await client.messages.create({
model: aiConfig.modelId,
max_tokens: aiConfig.maxTokens,
tools: [SUBMIT_QUOTE_ITEMS_TOOL],
// Tool_choice "any" forces the model to call our tool — no free-
// form prose to parse. Equivalent to JSON mode but with stronger
// schema guarantees.
tool_choice: { type: "tool", name: "submit_quote_items" },
system: await resolveSystemPrompt(),
messages: [{ role: "user", content: userContent }],
});
// ── 4. Pull the tool call out of the response. ──────────────────
const toolBlock = response.content.find(
(b) => b.type === "tool_use" && b.name === "submit_quote_items",
);
if (!toolBlock || toolBlock.type !== "tool_use") {
console.warn(
`[quotes-ai-processor] ${quote.id}: no submit_quote_items tool call — full response content:`,
JSON.stringify(response.content, null, 2).slice(0, 2000),
);
throw new Error(
`Claude returned no submit_quote_items tool call (stop_reason=${response.stop_reason})`,
);
}
// Diagnostic: log how many items Claude returned and the raw
// input shape so we can debug "why did Claude extract 0 items?"
// questions without re-running.
const toolInput = toolBlock.input as Record<string, unknown>;
const rawItems = Array.isArray(toolInput.quote_items)
? (toolInput.quote_items as unknown[])
: [];
console.log(
`[quotes-ai-processor] ${quote.id}: Claude returned ${rawItems.length} items (stop_reason=${response.stop_reason}, usage=${JSON.stringify(response.usage ?? {})})`,
);
if (rawItems.length === 0) {
console.log(
`[quotes-ai-processor] ${quote.id}: raw tool input:`,
JSON.stringify(toolInput, null, 2).slice(0, 2000),
);
}
// ── 5. Defensive parse. Claude's tool input is JSON-shaped but
// arrives as `unknown`. We validate the shape we need and
// discard anything else.
const parsed = parseToolInput(toolBlock.input);
// ── 6. Optionally update the company hint if Claude is more
// confident than the email's explicit hint. The hint is just
// metadata; the office can override.
if (parsed.company_hint && parsed.company_hint !== quote.description) {
const { company } = await findOrCreateCompany({ name: parsed.company_hint });
if (company) {
try {
await pb.collection(QUOTES_COLLECTION).update(quote.id, { company: company.id });
} catch (e) {
console.warn(
`[quotes-ai-processor] couldn't update company for ${quote.id}:`,
e,
);
}
}
}
// ── 7. Write the items. Existing `quote_items.create` mirrors
// the human flow (cost_price=0, selling_price=0, is_processed=false).
const itemsWritten = await writeItems(pb, quote, parsed.items);
return { itemsWritten };
}
// ──────────────────────────────────────────────────────────────────────────────
// Claude plumbing
// ──────────────────────────────────────────────────────────────────────────────
/**
* Resolve the system prompt for an extraction. Operator overrides
* come from `ai_settings.system_prompt`; when that's empty we fall
* back to the hardcoded default. Reads are cached (60s TTL) via
* `getAiSettings`, so the per-quote hot-path only hits PB on first
* call after a settings update.
*/
async function resolveSystemPrompt(): Promise<string> {
try {
const settings = await getAiSettings();
const override = settings.system_prompt?.trim();
if (override) return override;
} catch {
// Settings read failed — fall through to default rather than
// fail the whole extraction.
}
return DEFAULT_QUOTES_AI_SYSTEM_PROMPT;
}
const SUBMIT_QUOTE_ITEMS_TOOL = {
name: "submit_quote_items",
description:
"Submit the extracted quote line items from this email. Call this tool exactly once with the full result.",
input_schema: {
type: "object" as const,
properties: {
quote_items: {
type: "array",
description:
"Line items the customer wants quoted. Empty array is a valid answer (the office will follow up).",
items: {
type: "object",
properties: {
description: {
type: "string",
description:
"Self-contained line description (part name, spec, brand). No quantity or unit. Non-empty.",
},
quantity: {
type: "number",
description: "Quantity requested. Defaults to 1.",
},
unit: {
type: "string",
description:
"Unit of measure (each, box, metre, kg, pack, …). Empty if not stated.",
},
notes: {
type: "string",
description:
"Caveats / follow-up hints for the office. Not for part description.",
},
},
required: ["description"],
},
},
company_hint: {
type: "string",
description:
"Best-guess company name on the customer's side. Empty if ambiguous.",
},
overall_notes: {
type: "string",
description:
"Free-text notes for the office (e.g. quote_in_reply_to Q2607XXX, customer urgency, special instructions).",
},
},
required: ["quote_items"],
},
};
interface ParsedToolInput {
items: Array<{
description: string;
quantity: number;
unit: string;
notes: string;
}>;
company_hint: string;
overall_notes: string;
}
function parseToolInput(raw: unknown): ParsedToolInput {
// Loose runtime validation — we trust Claude's tool_use guarantee
// for the shape, but defensively coerce numbers and trim strings
// so a stray `null` from the model doesn't blow up the write loop.
const obj = (raw ?? {}) as Record<string, unknown>;
const rawItems = Array.isArray(obj.quote_items) ? obj.quote_items : [];
const items = rawItems
.map((it): ParsedToolInput["items"][number] | null => {
if (!it || typeof it !== "object") return null;
const o = it as Record<string, unknown>;
const desc = typeof o.description === "string" ? o.description.trim() : "";
if (!desc) return null;
const qty = Number(o.quantity);
const quantity = Number.isFinite(qty) && qty > 0 ? qty : 1;
return {
description: desc,
quantity,
unit: typeof o.unit === "string" ? o.unit.trim() : "",
notes: typeof o.notes === "string" ? o.notes.trim() : "",
};
})
.filter((x): x is ParsedToolInput["items"][number] => x !== null);
return {
items,
company_hint:
typeof obj.company_hint === "string" ? obj.company_hint.trim() : "",
overall_notes:
typeof obj.overall_notes === "string" ? obj.overall_notes.trim() : "",
};
}
function buildUserContent(
textForPrompt: string,
images: Array<{ filename: string; contentType: string; base64: string }>
): ContentBlockParam[] {
const content: ContentBlockParam[] = [];
// Inline images first so the model can reference them when reading
// the text. Each one is tagged with its filename for traceability.
for (const img of images) {
content.push({
type: "image",
source: {
type: "base64",
// Already filtered through isSupportedImageMime at the call
// site, so the cast is safe.
media_type: img.contentType as "image/jpeg" | "image/png" | "image/gif" | "image/webp",
data: img.base64,
},
});
content.push({
type: "text",
text: `(image: ${img.filename})`,
});
}
content.push({
type: "text",
text:
textForPrompt +
"\n\n---\nExtract the line items the customer wants quoted and call submit_quote_items.",
});
return content;
}
function newAnthropic(aiConfig: {
apiKey: string;
baseUrl: string;
}): Anthropic {
return new Anthropic({
apiKey: aiConfig.apiKey,
// Anthropic SDK takes a full URL including "/v1" — `getActiveAiConfig`
// returns the bare base URL (e.g. "https://api.minimax.io/anthropic")
// which is exactly what the provider expects. We don't add /v1
// ourselves because that would break the Anthropic-direct path
// (which appends /v1 internally based on the SDK default).
...(aiConfig.baseUrl ? { baseURL: aiConfig.baseUrl } : {}),
});
}
/** Anthropic's image API accepts jpeg/png/gif/webp only. Anything
* else (BMP, TIFF, SVG, etc.) would 4xx at the API boundary; we
* filter them out at extraction time so the call site stays simple. */
function isSupportedImageMime(ct: string): boolean {
const c = (ct || "").toLowerCase();
return (
c === "image/jpeg" ||
c === "image/jpg" ||
c === "image/png" ||
c === "image/gif" ||
c === "image/webp"
);
}
// ──────────────────────────────────────────────────────────────────────────────
// PB helpers
// ──────────────────────────────────────────────────────────────────────────────
async function loadQuoteAttachments(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>,
quoteId: string
): Promise<RawAttachment[]> {
const rows = await pb.collection(QUOTE_ATTACHMENTS_COLLECTION).getFullList<{
id: string;
filename: string;
content_type?: string;
file?: string;
}>({
filter: `quote = "${quoteId}"`,
fields: "id,filename,content_type,file",
sort: "uploaded_at",
});
if (rows.length > MAX_ATTACHMENTS) {
console.warn(
`[quotes-ai-processor] ${quoteId}: ${rows.length} attachments, capping to ${MAX_ATTACHMENTS}`,
);
}
const limited = rows.slice(0, MAX_ATTACHMENTS);
// Each row's `file` field is a PB file-array (`["filename.pdf"]`).
// Fetch the bytes via the PB `/api/files/...` URL.
const result: RawAttachment[] = [];
for (const row of limited) {
const first = row.file?.[0];
if (!first) continue;
try {
const token = pb.authStore.token;
const url = pb.buildURL(
`/api/files/${QUOTE_ATTACHMENTS_COLLECTION}/${row.id}/${first}`,
);
const resp = await fetch(url, {
headers: token ? { Authorization: token } : {},
});
if (!resp.ok) {
console.warn(
`[quotes-ai-processor] failed to download ${row.filename} (${resp.status})`,
);
continue;
}
const ab = await resp.arrayBuffer();
result.push({
filename: row.filename,
contentType: row.content_type || "application/octet-stream",
bytes: Buffer.from(ab),
});
} catch (e) {
console.warn(
`[quotes-ai-processor] error fetching ${row.filename}:`,
e,
);
}
}
return result;
}
async function writeItems(
pb: Awaited<ReturnType<typeof getAdminPocketBase>>,
quote: Quote,
items: ParsedToolInput["items"]
): Promise<number> {
if (items.length === 0) return 0;
// Same write shape as the legacy `createQuoteFromEmail` path:
// human opt-in only, all prices zero, no supplier.
let written = 0;
for (const it of items) {
try {
await pb.collection(QUOTE_ITEMS_COLLECTION).create({
quote: quote.id,
quote_number: quote.quote_number,
description: it.description,
quantity: it.quantity,
unit: it.unit,
cost_price: 0,
selling_price: 0,
markup: 0,
notes: it.notes,
is_processed: false,
pricing_source: "ai_email",
});
written += 1;
} catch (e) {
// Don't let one bad item kill the whole batch — log and keep
// going. The ai_error field will note the partial failure.
console.warn(
`[quotes-ai-processor] failed to write item "${it.description.slice(0, 60)}":`,
e,
);
}
}
return written;
}
// Suppress unused-import warning for Customer / QuoteItem — they're
// imported as types for the IDE / docs and may be referenced later.
export type _Refs = Customer | QuoteItem;