From cd73f4ee2e821db8a78267bf2f6a06a3b98c0ba0 Mon Sep 17 00:00:00 2001 From: chitcommit <208086304+chitcommit@users.noreply.github.com> Date: Mon, 23 Mar 2026 07:52:41 +0000 Subject: [PATCH 1/2] feat: convert scrape orchestration to ChittyRouter proxy + prompt consumer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scope violation fix: ChittyCommand (Tier 5 dashboard) was acting as orchestrator — owning job queues, execution, retry/backoff, and calling ChittyRouter agents via HTTP fan-out. This moves orchestration to where it belongs (ChittyRouter ScrapeAgent DO) and makes ChittyCommand a thin proxy. Changes: - DELETE fan-out.ts: orchestration now handled by ScrapeAgent DO-to-DO calls - REWRITE job-dispatcher.ts: proxy to ChittyRouter /agents/scrape/* with Neon fallback - ADD to integrations.ts: ScrapeAgent proxy methods (routerClient) + prompt registry methods (connectClient.resolvePrompt, executePrompt) - UPDATE litigation.ts: resolve prompts from ChittyConnect with AI Gateway fallback - UPDATE jobs.ts, mcp.ts, cron.ts: pass env for router proxy delegation Net result: -231 lines, ChittyCommand no longer owns scrape orchestration or hardcoded AI prompts. Co-Authored-By: Claude Opus 4.6 --- src/lib/cron.ts | 10 +- src/lib/fan-out.ts | 196 ------------------ src/lib/integrations.ts | 113 +++++++++++ src/lib/job-dispatcher.ts | 416 +++++++++----------------------------- src/routes/jobs.ts | 13 +- src/routes/litigation.ts | 220 +++++++++++++------- src/routes/mcp.ts | 10 +- 7 files changed, 372 insertions(+), 606 deletions(-) delete mode 100644 src/lib/fan-out.ts diff --git a/src/lib/cron.ts b/src/lib/cron.ts index 826db4b..95c94dd 100644 --- a/src/lib/cron.ts +++ b/src/lib/cron.ts @@ -7,7 +7,7 @@ import { generateProjections } from './projections'; import { discoverRevenueSources } from './revenue'; import { generatePaymentPlan, savePaymentPlan } from './payment-planner'; import { reconcileNotionDisputes } from './dispute-sync'; -import { enqueueJob, processQueue } from './job-dispatcher'; +import { enqueueJob, processQueue, type ScrapeJobType } from './job-dispatcher'; /** * Cron sync orchestrator. @@ -146,7 +146,7 @@ export async function runCronSync( await enqueueJob(sql, 'portal_scrape', { portal: target }, { chittyId, cronSource: 'utility_scrape', - }); + }, env); } catch (err) { console.error(`[cron:utility:${target}] enqueue failed:`, err); } @@ -175,7 +175,7 @@ export async function runCronSync( await enqueueJob(sql, 'court_docket', { case_number: '2024D007847' }, { chittyId, cronSource: 'court_docket', - }); + }, env); const queueResult = await processQueue(sql, env, ctx); recordsSynced += queueResult.succeeded; console.log(`[cron:court_docket] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`); @@ -581,7 +581,7 @@ async function syncMonthlyChecksViaDispatcher( await enqueueJob(sql, 'mr_cooper', { property: 'addison' }, { chittyId, cronSource: 'monthly_check', - }); + }, env); } catch (err) { console.error('[cron:mr_cooper] enqueue failed:', err); } @@ -596,7 +596,7 @@ async function syncMonthlyChecksViaDispatcher( }, { chittyId, cronSource: 'monthly_check', - }); + }, env); } } catch (err) { console.error('[cron:cook_county_tax] enqueue failed:', err); diff --git a/src/lib/fan-out.ts b/src/lib/fan-out.ts deleted file mode 100644 index 8e395e6..0000000 --- a/src/lib/fan-out.ts +++ /dev/null @@ -1,196 +0,0 @@ -import type { NeonQueryFunction } from '@neondatabase/serverless'; -import type { Env } from '../index'; -import { routerClient, ledgerClient } from './integrations'; -import type { ScrapeJobType } from './job-dispatcher'; - -export interface ScrapeResultContext { - jobId: string; - jobType: ScrapeJobType; - target: Record; - chittyId: string | null; - result: Record; - recordsSynced: number; -} - -/** - * Fan out scrape results to downstream ChittyOS services. - * All calls are fire-and-forget — failures are logged but don't affect the job. - */ -export async function fanOutScrapeResult( - env: Env, - sql: NeonQueryFunction, - ctx: ScrapeResultContext, -): Promise { - const results = await Promise.allSettled([ - fanOutToIntelligence(env, ctx), - fanOutToCalendar(env, ctx), - fanOutToTriage(env, ctx), - fanOutToLedger(env, ctx), - ]); - - for (const r of results) { - if (r.status === 'rejected') { - console.error('[fan-out] downstream error:', r.reason); - } - } -} - -/** - * Send observations to ChittyRouter IntelligenceAgent. - * Teaches the system about scrape patterns, portal changes, and data trends. - */ -async function fanOutToIntelligence(env: Env, ctx: ScrapeResultContext): Promise { - const router = routerClient(env); - if (!router) return; - - try { - await fetch(`${env.CHITTYROUTER_URL}/agents/intelligence/observe`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Source-Service': 'chittycommand', - }, - body: JSON.stringify({ - observation_type: 'scrape_result', - source_agent: 'chittycommand', - org: 'personal', - title: `${ctx.jobType} scrape completed`, - description: `Job ${ctx.jobId}: ${ctx.recordsSynced} records synced`, - severity: 'info', - data: { - jobId: ctx.jobId, - jobType: ctx.jobType, - target: ctx.target, - recordCount: ctx.recordsSynced, - timestamp: new Date().toISOString(), - }, - }), - signal: AbortSignal.timeout(10000), - }); - } catch (err) { - console.error('[fan-out:intelligence]', err); - } -} - -/** - * Extract deadlines from court/legal scrapes and push to CalendarAgent. - */ -async function fanOutToCalendar(env: Env, ctx: ScrapeResultContext): Promise { - if (ctx.jobType !== 'court_docket') return; - if (!ctx.result.nextHearing && !ctx.result.entries) return; - if (!env.CHITTYROUTER_URL) return; - - try { - // Push next hearing as calendar event - if (ctx.result.nextHearing) { - await fetch(`${env.CHITTYROUTER_URL}/agents/calendar/create`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Source-Service': 'chittycommand', - }, - body: JSON.stringify({ - title: `Court Hearing: ${(ctx.target.case_number as string) || 'Unknown'}`, - date: ctx.result.nextHearing, - type: 'court_date', - urgency: 'high', - metadata: { - source: 'scrape_fan_out', - jobId: ctx.jobId, - caseNumber: ctx.target.case_number, - }, - }), - signal: AbortSignal.timeout(10000), - }); - } - - // Push docket entries with dates as calendar events - const entries = ctx.result.entries as Array> | undefined; - if (entries) { - for (const entry of entries) { - if (!entry.date) continue; - await fetch(`${env.CHITTYROUTER_URL}/agents/calendar/create`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Source-Service': 'chittycommand', - }, - body: JSON.stringify({ - title: `Docket: ${entry.description || entry.type || 'Entry'}`, - date: entry.date, - type: 'docket_entry', - urgency: 'medium', - metadata: { - source: 'scrape_fan_out', - jobId: ctx.jobId, - caseNumber: ctx.target.case_number, - }, - }), - signal: AbortSignal.timeout(10000), - }); - } - } - } catch (err) { - console.error('[fan-out:calendar]', err); - } -} - -/** - * Classify scrape findings via TriageAgent for urgency scoring. - */ -async function fanOutToTriage(env: Env, ctx: ScrapeResultContext): Promise { - // Only triage if there are new entries worth classifying - const entries = ctx.result.entries as Array> | undefined; - if (!entries || entries.length === 0) return; - - const router = routerClient(env); - if (!router) return; - - try { - await router.classifyDispute({ - entity_id: ctx.jobId, - entity_type: 'dispute', - title: `Scrape findings: ${ctx.jobType}`, - dispute_type: ctx.jobType, - description: `${entries.length} new entries from ${ctx.jobType} scrape`, - }); - } catch (err) { - console.error('[fan-out:triage]', err); - } -} - -/** - * Record scrape event in ChittyLedger as an immutable audit entry. - */ -async function fanOutToLedger(env: Env, ctx: ScrapeResultContext): Promise { - if (!env.CHITTYLEDGER_URL) return; - - try { - await fetch(`${env.CHITTYLEDGER_URL}/entries`, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - 'X-Source-Service': 'chittycommand', - }, - // @canon: chittycanon://gov/governance#core-types - // Ledger entityType is a record-category enum (not ChittyID P/L/T/E/A entity classification). - // actorType: 'person' (P) for ChittyID-bound actors, 'service' for system actors. - body: JSON.stringify({ - entityType: 'scrape', - entityId: ctx.jobId, - action: 'completed', - actor: ctx.chittyId || 'chittycommand', - actorType: ctx.chittyId ? 'person' : 'service', - metadata: { - jobType: ctx.jobType, - target: ctx.target, - recordsSynced: ctx.recordsSynced, - completedAt: new Date().toISOString(), - }, - }), - signal: AbortSignal.timeout(10000), - }); - } catch (err) { - console.error('[fan-out:ledger]', err); - } -} diff --git a/src/lib/integrations.ts b/src/lib/integrations.ts index c7109ab..7f5893e 100644 --- a/src/lib/integrations.ts +++ b/src/lib/integrations.ts @@ -312,6 +312,32 @@ export function connectClient(env: Env) { const baseUrl = env.CHITTYCONNECT_URL; if (!baseUrl) return null; + async function connectPost(path: string, body: unknown): Promise { + try { + const headers: Record = { + 'Content-Type': 'application/json', + 'X-Source-Service': 'chittycommand', + }; + if (env.CHITTY_CONNECT_TOKEN) { + headers['Authorization'] = `Bearer ${env.CHITTY_CONNECT_TOKEN}`; + } + const res = await fetch(`${baseUrl}${path}`, { + method: 'POST', + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(30000), + }); + if (!res.ok) { + console.error(`[connect] POST ${path} failed: ${res.status}`); + return null; + } + return await res.json() as T; + } catch (err) { + console.error(`[connect] POST ${path} error:`, err); + return null; + } + } + return { /** Discover a service URL by name */ discover: async (serviceName: string): Promise => { @@ -338,9 +364,48 @@ export function connectClient(env: Env) { return data.url; } catch { return null; } }, + + // ── Prompt Registry (ContextConsciousness) ───────────────── + /** Resolve a prompt: compose base + layers, apply env gating */ + resolvePrompt: (promptId: string, environment: string, variables?: Record, additionalLayers?: string[]) => + connectPost('/api/v1/context/prompts/resolve', { + promptId, + environment, + variables, + additionalLayers, + consumerService: 'chittycommand', + }), + + /** Execute a prompt: resolve + dispatch to agent, return AI result */ + executePrompt: (promptId: string, environment: string, input: Record, opts?: { additionalLayers?: string[] }) => + connectPost('/api/v1/context/prompts/execute', { + promptId, + environment, + input, + additionalLayers: opts?.additionalLayers, + consumerService: 'chittycommand', + }), }; } +export interface PromptResolveResponse { + systemPrompt: string; + aiEnabled: boolean; + version: number; + resolvedLayers: string[]; + fallbackMode: string | null; +} + +export interface PromptExecuteResponse { + result: string; + promptVersion: number; + resolvedLayers: string[]; + executedBy: string; + latencyMs: number; + executionId: number; + aiEnabled: boolean; +} + // ── Mercury ───────────────────────────────────────────────── // Direct Mercury API for multi-entity banking @@ -627,9 +692,57 @@ export function routerClient(env: Env) { labels: string[]; reasoning?: string; }>('/agents/triage/classify', payload), + + // ── ScrapeAgent proxy methods ────────────────────────────── + /** Enqueue a scrape job on ChittyRouter ScrapeAgent */ + enqueueScrapeJob: (jobType: string, target: Record, opts?: { chittyId?: string; maxAttempts?: number; cronSource?: string }) => + post<{ id: string; status: string }>('/agents/scrape/enqueue', { jobType, target, ...opts }), + + /** Get a single scrape job status */ + getScrapeJobStatus: (jobId: string) => + get(`/agents/scrape/jobs/${encodeURIComponent(jobId)}`), + + /** List scrape jobs with filters */ + listScrapeJobs: (filters?: { status?: string; jobType?: string; limit?: number }) => { + const params = new URLSearchParams(); + if (filters?.status) params.set('status', filters.status); + if (filters?.jobType) params.set('jobType', filters.jobType); + if (filters?.limit) params.set('limit', String(filters.limit)); + const qs = params.toString(); + return get<{ jobs: ScrapeJobResponse[]; total: number }>(`/agents/scrape/jobs${qs ? `?${qs}` : ''}`); + }, + + /** Retry a failed/dead-lettered scrape job */ + retryScrapeJob: (jobId: string) => + post<{ status: string }>(`/agents/scrape/jobs/${encodeURIComponent(jobId)}/retry`, {}), + + /** Get dead-lettered scrape jobs */ + getScrapeDeadLetters: () => + get<{ jobs: ScrapeJobResponse[] }>('/agents/scrape/dead-letters'), + + /** Trigger queue processing on ScrapeAgent */ + processScrapeQueue: () => + post<{ processed: number; succeeded: number; failed: number }>('/agents/scrape/process', {}), + + /** Get ScrapeAgent health status */ + getScrapeStatus: () => + get>('/agents/scrape/status'), }; } +export interface ScrapeJobResponse { + id: string; + jobType: string; + target: Record; + status: string; + attempt: number; + maxAttempts: number; + result?: Record; + error?: string; + createdAt: string; + completedAt?: string; +} + // ── Notion (write path) ─────────────────────────────────────── // Reading Notion is handled by syncNotionTasks() in cron.ts. // This client covers the write path: creating task pages from disputes. diff --git a/src/lib/job-dispatcher.ts b/src/lib/job-dispatcher.ts index 8b467d0..8efbc18 100644 --- a/src/lib/job-dispatcher.ts +++ b/src/lib/job-dispatcher.ts @@ -1,7 +1,7 @@ import type { NeonQueryFunction } from '@neondatabase/serverless'; import type { Env } from '../index'; -import { scrapeClient, routerClient } from './integrations'; -import { fanOutScrapeResult } from './fan-out'; +import { routerClient } from './integrations'; +import type { ScrapeJobResponse } from './integrations'; export type ScrapeJobType = | 'court_docket' @@ -44,14 +44,31 @@ export interface ScrapeJob { } /** - * Enqueue a new scrape job. Returns the job ID. + * Enqueue a scrape job via ChittyRouter ScrapeAgent. + * Falls back to local Neon queue if router is unavailable. */ export async function enqueueJob( sql: NeonQueryFunction, jobType: ScrapeJobType, target: Record, opts: EnqueueOptions = {}, + env?: Env, ): Promise { + // Proxy to ChittyRouter ScrapeAgent + if (env) { + const router = routerClient(env); + if (router) { + const result = await router.enqueueScrapeJob(jobType, target, { + chittyId: opts.chittyId, + maxAttempts: opts.maxAttempts, + cronSource: opts.cronSource, + }); + if (result?.id) return result.id; + console.warn('[dispatcher] ScrapeAgent enqueue failed, falling back to local queue'); + } + } + + // Fallback: local Neon queue (legacy path) const scheduledAt = opts.scheduledAt?.toISOString() || new Date().toISOString(); const [row] = await sql` INSERT INTO cc_scrape_jobs (job_type, target, chitty_id, max_attempts, scheduled_at, cron_source, parent_job_id) @@ -70,84 +87,8 @@ export async function enqueueJob( } /** - * Execute a single scrape job by ID. - * Handles calling ChittyScrape, persisting results, and updating job status. - */ -export async function executeJob( - sql: NeonQueryFunction, - env: Env, - jobId: string, - ctx?: ExecutionContext, -): Promise<{ success: boolean; recordsSynced: number }> { - // Load the job - const [job] = await sql` - SELECT * FROM cc_scrape_jobs WHERE id = ${jobId} - `; - if (!job) throw new Error(`Job ${jobId} not found`); - - const jobType = job.job_type as ScrapeJobType; - const target = job.target as Record; - const attempt = (job.attempt as number) + 1; - const maxAttempts = job.max_attempts as number; - - // Mark running - await sql` - UPDATE cc_scrape_jobs - SET status = 'running', attempt = ${attempt}, started_at = NOW() - WHERE id = ${jobId} - `; - - try { - const result = await executeScrape(sql, env, jobType, target); - - // Mark completed - await sql` - UPDATE cc_scrape_jobs - SET status = 'completed', result = ${JSON.stringify(result.data)}::jsonb, completed_at = NOW() - WHERE id = ${jobId} - `; - - // Fan out to downstream services (fire-and-forget) - if (ctx) { - ctx.waitUntil(fanOutScrapeResult(env, sql, { - jobId, - jobType, - target, - chittyId: job.chitty_id as string | null, - result: result.data, - recordsSynced: result.recordsSynced, - })); - } - - return { success: true, recordsSynced: result.recordsSynced }; - } catch (err) { - const errorMsg = err instanceof Error ? err.message : String(err); - - if (attempt < maxAttempts) { - // Schedule retry with exponential backoff: 30s, 60s, 120s... - const backoffMs = 30000 * Math.pow(2, attempt - 1); - const retryAt = new Date(Date.now() + backoffMs).toISOString(); - await sql` - UPDATE cc_scrape_jobs - SET status = 'retrying', error_message = ${errorMsg}, scheduled_at = ${retryAt} - WHERE id = ${jobId} - `; - } else { - // Dead letter - await sql` - UPDATE cc_scrape_jobs - SET status = 'dead_letter', error_message = ${errorMsg}, completed_at = NOW() - WHERE id = ${jobId} - `; - } - - return { success: false, recordsSynced: 0 }; - } -} - -/** - * Process the queue: pick up jobs that are ready to run and execute them. - * Called from cron or manual trigger. + * Process the queue via ChittyRouter ScrapeAgent. + * Falls back to logging a warning if router is unavailable. */ export async function processQueue( sql: NeonQueryFunction, @@ -155,44 +96,37 @@ export async function processQueue( ctx?: ExecutionContext, limit = 10, ): Promise<{ processed: number; succeeded: number; failed: number }> { - const jobs = await sql` - SELECT id FROM cc_scrape_jobs - WHERE status IN ('queued', 'retrying') - AND scheduled_at <= NOW() - ORDER BY scheduled_at - LIMIT ${limit} - `; - - let succeeded = 0; - let failed = 0; - - for (const job of jobs) { - try { - const result = await executeJob(sql, env, job.id as string, ctx); - if (result.success) succeeded++; - else failed++; - } catch (err) { - console.error(`[dispatcher] Job ${job.id} threw:`, err); - failed++; - } + const router = routerClient(env); + if (router) { + const result = await router.processScrapeQueue(); + if (result) return result; + console.warn('[dispatcher] ScrapeAgent processQueue failed'); } - - return { processed: jobs.length, succeeded, failed }; + return { processed: 0, succeeded: 0, failed: 0 }; } /** - * Get job status by ID. + * Get job status — queries ChittyRouter ScrapeAgent first, falls back to local Neon. */ export async function getJobStatus( sql: NeonQueryFunction, jobId: string, + env?: Env, ): Promise { + if (env) { + const router = routerClient(env); + if (router) { + const result = await router.getScrapeJobStatus(jobId); + if (result) return mapRouterJob(result); + } + } + // Fallback: local Neon (historical jobs) const [row] = await sql`SELECT * FROM cc_scrape_jobs WHERE id = ${jobId}`; return row ? mapJobRow(row) : null; } /** - * List jobs with optional filters. + * List jobs — queries ChittyRouter ScrapeAgent first, falls back to local Neon. */ export async function listJobs( sql: NeonQueryFunction, @@ -203,50 +137,37 @@ export async function listJobs( limit?: number; offset?: number; } = {}, + env?: Env, ): Promise<{ jobs: ScrapeJob[]; total: number }> { + if (env) { + const router = routerClient(env); + if (router) { + const result = await router.listScrapeJobs({ + status: filters.status, + jobType: filters.jobType, + limit: filters.limit, + }); + if (result) { + return { + jobs: result.jobs.map(mapRouterJob), + total: result.total, + }; + } + } + } + // Fallback: local Neon (historical) const limit = filters.limit || 50; const offset = filters.offset || 0; - - // Build dynamic WHERE conditions - const conditions: string[] = []; - const params: unknown[] = []; - - if (filters.status) { - conditions.push(`status = $${params.length + 1}`); - params.push(filters.status); - } - if (filters.jobType) { - conditions.push(`job_type = $${params.length + 1}`); - params.push(filters.jobType); - } - if (filters.chittyId) { - conditions.push(`chitty_id = $${params.length + 1}`); - params.push(filters.chittyId); - } - - // Use tagged template for the common case (no filters or single filter) let rows: any[]; let countRows: any[]; if (!filters.status && !filters.jobType && !filters.chittyId) { - rows = await sql` - SELECT * FROM cc_scrape_jobs ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset} - `; + rows = await sql`SELECT * FROM cc_scrape_jobs ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset}`; countRows = await sql`SELECT COUNT(*)::int AS total FROM cc_scrape_jobs`; } else if (filters.status && !filters.jobType && !filters.chittyId) { - rows = await sql` - SELECT * FROM cc_scrape_jobs WHERE status = ${filters.status} - ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset} - `; + rows = await sql`SELECT * FROM cc_scrape_jobs WHERE status = ${filters.status} ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset}`; countRows = await sql`SELECT COUNT(*)::int AS total FROM cc_scrape_jobs WHERE status = ${filters.status}`; - } else if (filters.jobType && !filters.status && !filters.chittyId) { - rows = await sql` - SELECT * FROM cc_scrape_jobs WHERE job_type = ${filters.jobType} - ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset} - `; - countRows = await sql`SELECT COUNT(*)::int AS total FROM cc_scrape_jobs WHERE job_type = ${filters.jobType}`; } else { - // Multiple filters — build with AND rows = await sql` SELECT * FROM cc_scrape_jobs WHERE (${filters.status || null}::text IS NULL OR status = ${filters.status || null}) @@ -269,12 +190,21 @@ export async function listJobs( } /** - * Get dead-lettered jobs for review. + * Get dead-lettered jobs from ChittyRouter ScrapeAgent. */ export async function getDeadLetters( sql: NeonQueryFunction, limit = 50, + env?: Env, ): Promise { + if (env) { + const router = routerClient(env); + if (router) { + const result = await router.getScrapeDeadLetters(); + if (result) return result.jobs.map(mapRouterJob); + } + } + // Fallback: local Neon const rows = await sql` SELECT * FROM cc_scrape_jobs WHERE status = 'dead_letter' ORDER BY completed_at DESC LIMIT ${limit} @@ -283,12 +213,21 @@ export async function getDeadLetters( } /** - * Retry a failed/dead-lettered job. + * Retry a failed/dead-lettered job via ChittyRouter ScrapeAgent. */ export async function retryJob( sql: NeonQueryFunction, jobId: string, + env?: Env, ): Promise { + if (env) { + const router = routerClient(env); + if (router) { + const result = await router.retryScrapeJob(jobId); + if (result) return true; + } + } + // Fallback: local Neon const [row] = await sql` UPDATE cc_scrape_jobs SET status = 'queued', attempt = 0, error_message = NULL, @@ -299,187 +238,28 @@ export async function retryJob( return !!row; } -// ── Internal scrape execution ───────────────────────────────── - -interface ScrapeResult { - data: Record; - recordsSynced: number; -} - -async function executeScrape( - sql: NeonQueryFunction, - env: Env, - jobType: ScrapeJobType, - target: Record, -): Promise { - switch (jobType) { - case 'court_docket': - return executeCourtDocket(sql, env, target); - case 'cook_county_tax': - return executeCookCountyTax(sql, env, target); - case 'mr_cooper': - return executeMrCooper(sql, env, target); - case 'portal_scrape': - return executePortalScrape(sql, env, target); - default: - throw new Error(`Unknown job type: ${jobType}`); - } -} - -async function executeCourtDocket( - sql: NeonQueryFunction, - env: Env, - target: Record, -): Promise { - const scrape = scrapeClient(env); - if (!scrape) throw new Error('ChittyScrape not configured'); - - const token = await env.COMMAND_KV.get('scrape:service_token'); - if (!token) throw new Error('No scrape:service_token in KV'); - - const caseNumber = target.case_number as string; - const result = await scrape.scrapeCourtDocket(caseNumber, token); - if (!result?.success) throw new Error(result?.error || 'Scrape failed'); - - let synced = 0; - if (result.data?.entries) { - for (const entry of result.data.entries) { - await sql` - INSERT INTO cc_legal_deadlines (case_ref, deadline_type, deadline_date, description, metadata) - VALUES (${caseNumber}, ${entry.type || 'court_entry'}, ${entry.date || null}, ${entry.description || ''}, - ${JSON.stringify({ source: 'court_docket_scrape' })}::jsonb) - ON CONFLICT DO NOTHING - `; - synced++; - } - } - - if (result.data?.nextHearing) { - await sql` - INSERT INTO cc_legal_deadlines (case_ref, deadline_type, deadline_date, description, metadata) - VALUES (${caseNumber}, 'hearing', ${result.data.nextHearing}, 'Next court hearing', - ${JSON.stringify({ source: 'court_docket_scrape' })}::jsonb) - ON CONFLICT DO NOTHING - `; - synced++; - } - - return { data: result.data || {}, recordsSynced: synced }; -} - -async function executeCookCountyTax( - sql: NeonQueryFunction, - env: Env, - target: Record, -): Promise { - const scrape = scrapeClient(env); - if (!scrape) throw new Error('ChittyScrape not configured'); - - const token = await env.COMMAND_KV.get('scrape:service_token'); - if (!token) throw new Error('No scrape:service_token in KV'); - - const pin = target.pin as string; - const propertyId = target.property_id as string | undefined; - - const taxResult = await scrape.scrapeCookCountyTax(pin, token); - if (!taxResult?.success) throw new Error(taxResult?.error || 'Scrape failed'); - - if (taxResult.data && propertyId) { - await sql` - UPDATE cc_properties - SET annual_tax = ${taxResult.data.totalTax || 0}, - metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{last_tax_scrape}', ${JSON.stringify(taxResult.data)}::jsonb), - updated_at = NOW() - WHERE id = ${propertyId} - `; - } else if (taxResult.data) { - await sql` - UPDATE cc_properties - SET annual_tax = ${taxResult.data.totalTax || 0}, - metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{last_tax_scrape}', ${JSON.stringify(taxResult.data)}::jsonb), - updated_at = NOW() - WHERE tax_pin = ${pin} - `; - } - - return { data: taxResult.data || {}, recordsSynced: 1 }; -} - -async function executeMrCooper( - sql: NeonQueryFunction, - env: Env, - target: Record, -): Promise { - const scrape = scrapeClient(env); - if (!scrape) throw new Error('ChittyScrape not configured'); - - const token = await env.COMMAND_KV.get('scrape:service_token'); - if (!token) throw new Error('No scrape:service_token in KV'); +// ── Mappers ────────────────────────────────────────────────── - const property = target.property as string; - const result = await scrape.scrapeMrCooper(property, token); - if (!result?.success) throw new Error(result?.error || 'Scrape failed'); - - if (result.data) { - await sql` - UPDATE cc_obligations - SET amount_due = ${result.data.monthlyPayment || result.data.currentBalance || 0}, - metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{last_scrape}', ${JSON.stringify(result.data)}::jsonb), - updated_at = NOW() - WHERE payee ILIKE '%mr. cooper%' OR payee ILIKE '%mr cooper%' - `; - } - - return { data: result.data || {}, recordsSynced: 1 }; -} - -async function executePortalScrape( - sql: NeonQueryFunction, - env: Env, - target: Record, -): Promise { - const router = routerClient(env); - if (!router) throw new Error('ChittyRouter not configured'); - - const portalTarget = target.portal as string; - const result = await router.scrapePortal(portalTarget); - if (!result?.success) throw new Error(result?.error || 'Portal scrape failed'); - - let synced = 0; - if (result.data && (result.data.amount || result.data.amount_due)) { - const amount = Number(result.data.amount || result.data.amount_due || 0); - const dueDate = (result.data.due_date || result.data.dueDate || null) as string | null; - const payee = (result.data.payee || portalTarget) as string; - const escapedPayee = payee.replace(/%/g, '\\%').replace(/_/g, '\\_'); - - const [existing] = await sql` - SELECT id FROM cc_obligations WHERE payee ILIKE ${`%${escapedPayee}%`} AND status IN ('pending', 'overdue') LIMIT 1 - `; - - if (existing) { - await sql` - UPDATE cc_obligations - SET amount_due = ${amount}, - due_date = COALESCE(${dueDate}, due_date), - metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{last_portal_scrape}', ${JSON.stringify(result.data)}::jsonb), - updated_at = NOW() - WHERE id = ${existing.id} - `; - } else if (dueDate) { - await sql` - INSERT INTO cc_obligations (category, payee, amount_due, due_date, status, metadata) - VALUES ('utility', ${payee}, ${amount}, ${dueDate}, 'pending', - ${JSON.stringify({ source: 'portal_scrape', last_portal_scrape: result.data })}::jsonb) - `; - } - synced++; - } - - return { data: result.data || {}, recordsSynced: synced }; +function mapRouterJob(r: ScrapeJobResponse): ScrapeJob { + return { + id: r.id, + chittyId: null, + jobType: r.jobType as ScrapeJobType, + target: r.target, + status: r.status as ScrapeJobStatus, + attempt: r.attempt, + maxAttempts: r.maxAttempts, + scheduledAt: r.createdAt, + startedAt: null, + completedAt: r.completedAt || null, + result: r.result || null, + errorMessage: r.error || null, + parentJobId: null, + cronSource: null, + createdAt: r.createdAt, + }; } -// ── Helpers ──────────────────────────────────────────────────── - function mapJobRow(row: Record): ScrapeJob { return { id: row.id as string, diff --git a/src/routes/jobs.ts b/src/routes/jobs.ts index 9f216e4..2e3bbd2 100644 --- a/src/routes/jobs.ts +++ b/src/routes/jobs.ts @@ -23,14 +23,14 @@ jobRoutes.get('/jobs', async (c) => { const limit = Math.min(parseInt(c.req.query('limit') || '50'), 100); const offset = parseInt(c.req.query('offset') || '0'); - const result = await listJobs(sql, { status, jobType, chittyId, limit, offset }); + const result = await listJobs(sql, { status, jobType, chittyId, limit, offset }, c.env); return c.json(result); }); // Get single job status jobRoutes.get('/jobs/:id', async (c) => { const sql = getDb(c.env); - const job = await getJobStatus(sql, c.req.param('id')); + const job = await getJobStatus(sql, c.req.param('id'), c.env); if (!job) return c.json({ error: 'Job not found' }, 404); return c.json(job); }); @@ -39,14 +39,14 @@ jobRoutes.get('/jobs/:id', async (c) => { jobRoutes.get('/jobs/queue/dead-letter', async (c) => { const sql = getDb(c.env); const limit = Math.min(parseInt(c.req.query('limit') || '50'), 100); - const jobs = await getDeadLetters(sql, limit); + const jobs = await getDeadLetters(sql, limit, c.env); return c.json({ jobs, total: jobs.length }); }); // Retry a failed job jobRoutes.post('/jobs/:id/retry', async (c) => { const sql = getDb(c.env); - const success = await retryJob(sql, c.req.param('id')); + const success = await retryJob(sql, c.req.param('id'), c.env); if (!success) return c.json({ error: 'Job not found or not in retryable state' }, 404); return c.json({ status: 'queued', message: 'Job re-queued for retry' }); }); @@ -74,7 +74,7 @@ jobRoutes.post('/jobs', async (c) => { chittyId: body.chitty_id, maxAttempts: body.max_attempts, cronSource: 'manual', - }); + }, c.env); return c.json({ id: jobId, status: 'queued' }, 201); }); @@ -82,7 +82,6 @@ jobRoutes.post('/jobs', async (c) => { // Trigger queue processing manually jobRoutes.post('/jobs/queue/process', async (c) => { const sql = getDb(c.env); - const limit = Math.min(parseInt(c.req.query('limit') || '10'), 50); - const result = await processQueue(sql, c.env, undefined, limit); + const result = await processQueue(sql, c.env); return c.json(result); }); diff --git a/src/routes/litigation.ts b/src/routes/litigation.ts index 94254d1..da102cf 100644 --- a/src/routes/litigation.ts +++ b/src/routes/litigation.ts @@ -2,6 +2,7 @@ import { Hono } from 'hono'; import { z } from 'zod'; import type { Env } from '../index'; import type { AuthVariables } from '../middleware/auth'; +import { connectClient } from '../lib/integrations'; export const litigationRoutes = new Hono<{ Bindings: Env; Variables: AuthVariables }>(); @@ -22,7 +23,144 @@ const qcSchema = z.object({ draftEmail: z.string().min(1).max(10000), }); -async function callAIGateway( +// ── Step 1+2: Fact Synthesizer ───────────────────────────── + +litigationRoutes.post('/synthesize', async (c) => { + const raw = await c.req.json().catch(() => null); + const parsed = synthesizeSchema.safeParse(raw); + if (!parsed.success) { + return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); + } + + const { rawNotes, property, caseNumber } = parsed.data; + const environment = c.env.ENVIRONMENT || 'production'; + + const connect = connectClient(c.env); + if (connect) { + const result = await connect.executePrompt( + 'litigation.synthesize', + environment, + { rawNotes, property: property || '', caseNumber: caseNumber || '' }, + { additionalLayers: caseNumber ? [`case:${caseNumber}`] : [] }, + ); + + if (result) { + if (!result.aiEnabled) { + return c.json({ synthesis: rawNotes, passthrough: true }); + } + return c.json({ synthesis: result.result }); + } + console.warn('[litigation/synthesize] ChittyConnect execute failed, falling back to direct AI'); + } + + // Fallback: direct AI Gateway call (until prompt registry is seeded) + try { + const result = await callAIGatewayFallback(c.env, + FALLBACK_SYNTHESIZE_PROMPT, + `Raw notes:\n${rawNotes}${property ? `\nProperty: ${property}` : ''}${caseNumber ? `\nCase: ${caseNumber}` : ''}`, + ); + return c.json({ synthesis: result }); + } catch (err) { + console.error('[litigation/synthesize]', err instanceof Error ? err.message : err); + return c.json({ error: 'AI synthesis failed. Please try again.' }, 502); + } +}); + +// ── Step 3: Auto-Drafter ─────────────────────────────────── + +litigationRoutes.post('/draft', async (c) => { + const raw = await c.req.json().catch(() => null); + const parsed = draftSchema.safeParse(raw); + if (!parsed.success) { + return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); + } + + const { synthesizedFacts, focus, recipient } = parsed.data; + const environment = c.env.ENVIRONMENT || 'production'; + + const connect = connectClient(c.env); + if (connect) { + const result = await connect.executePrompt( + 'litigation.draft', + environment, + { synthesizedFacts, focus, recipient }, + ); + + if (result) { + if (!result.aiEnabled) { + return c.json({ draft: synthesizedFacts, passthrough: true }); + } + return c.json({ draft: result.result }); + } + console.warn('[litigation/draft] ChittyConnect execute failed, falling back to direct AI'); + } + + try { + const result = await callAIGatewayFallback(c.env, + FALLBACK_DRAFT_PROMPT.replace('{{recipient}}', recipient).replace('{{focus}}', focus), + `Synthesized facts:\n${synthesizedFacts}`, + ); + return c.json({ draft: result }); + } catch (err) { + console.error('[litigation/draft]', err instanceof Error ? err.message : err); + return c.json({ error: 'AI drafting failed. Please try again.' }, 502); + } +}); + +// ── Step 4: Risk Scanner ─────────────────────────────────── + +litigationRoutes.post('/qc', async (c) => { + const raw = await c.req.json().catch(() => null); + const parsed = qcSchema.safeParse(raw); + if (!parsed.success) { + return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); + } + + const { rawNotes, draftEmail } = parsed.data; + const environment = c.env.ENVIRONMENT || 'production'; + + const connect = connectClient(c.env); + if (connect) { + const result = await connect.executePrompt( + 'litigation.qc', + environment, + { rawNotes, draftEmail }, + ); + + if (result) { + if (!result.aiEnabled) { + return c.json({ flags: [], passthrough: true }); + } + try { + const cleaned = result.result.replace(/```json\s*/g, '').replace(/```\s*/g, '').trim(); + return c.json({ flags: JSON.parse(cleaned) }); + } catch { + return c.json({ flags: [], warning: 'QC analysis returned non-parseable results' }); + } + } + console.warn('[litigation/qc] ChittyConnect execute failed, falling back to direct AI'); + } + + try { + const result = await callAIGatewayFallback(c.env, + FALLBACK_QC_PROMPT, + `Original Source Notes:\n"${rawNotes}"\n\nDrafted Email:\n"${draftEmail}"`, + ); + const cleaned = result.replace(/```json\s*/g, '').replace(/```\s*/g, '').trim(); + const flags = JSON.parse(cleaned); + return c.json({ flags }); + } catch (err) { + console.error('[litigation/qc]', err instanceof Error ? err.message : err); + if (err instanceof SyntaxError) { + return c.json({ flags: [], warning: 'QC analysis returned non-parseable results' }); + } + return c.json({ error: 'AI QC scan failed. Please try again.' }, 502); + } +}); + +// ── Fallback: direct AI Gateway (until prompts are seeded in ChittyConnect) ── + +async function callAIGatewayFallback( env: Env, systemPrompt: string, userPrompt: string, @@ -65,18 +203,9 @@ async function callAIGateway( return result.choices?.[0]?.message?.content || ''; } -// ── Step 1+2: Fact Synthesizer ───────────────────────────── +// ── Fallback prompts (used until ChittyConnect prompt registry is seeded) ── -litigationRoutes.post('/synthesize', async (c) => { - const raw = await c.req.json().catch(() => null); - const parsed = synthesizeSchema.safeParse(raw); - if (!parsed.success) { - return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); - } - - const { rawNotes, property, caseNumber } = parsed.data; - - const systemPrompt = `You are a strict Litigation Support AI operating under Evidentiary Discipline. +const FALLBACK_SYNTHESIZE_PROMPT = `You are a strict Litigation Support AI operating under Evidentiary Discipline. Analyze the provided raw materials. Extract all facts and categorize them under these headings: - Property Facts - Case Posture @@ -92,32 +221,10 @@ Use bullet points. CRITICAL: Every single bullet MUST begin with one of these EX Do not fabricate any facts. Do not editorialize. Output clean markdown with ## headings and bullet lists.`; - const userPrompt = `Raw notes:\n${rawNotes}${property ? `\nProperty: ${property}` : ''}${caseNumber ? `\nCase: ${caseNumber}` : ''}`; - - try { - const result = await callAIGateway(c.env, systemPrompt, userPrompt); - return c.json({ synthesis: result }); - } catch (err) { - console.error('[litigation/synthesize]', err instanceof Error ? err.message : err); - return c.json({ error: 'AI synthesis failed. Please try again.' }, 502); - } -}); - -// ── Step 3: Auto-Drafter ─────────────────────────────────── - -litigationRoutes.post('/draft', async (c) => { - const raw = await c.req.json().catch(() => null); - const parsed = draftSchema.safeParse(raw); - if (!parsed.success) { - return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); - } - - const { synthesizedFacts, focus, recipient } = parsed.data; - - const systemPrompt = `You are an expert litigation assistant drafting an email from a client to their attorney. +const FALLBACK_DRAFT_PROMPT = `You are an expert litigation assistant drafting an email from a client to their attorney. Rules: -1. Recipient: ${recipient}. -2. Focus: ${focus}. +1. Recipient: {{recipient}}. +2. Focus: {{focus}}. 3. Maximum 250 words. 4. Tone: Concise, professional, cooperative. This is attorney-client privileged communication. 5. Base the email ONLY on the provided synthesized facts. @@ -126,27 +233,7 @@ Rules: 8. Include specific action items or questions for the attorney. 9. Output the email as plain text with Subject line, greeting, body, and sign-off.`; - try { - const result = await callAIGateway(c.env, systemPrompt, `Synthesized facts:\n${synthesizedFacts}`); - return c.json({ draft: result }); - } catch (err) { - console.error('[litigation/draft]', err instanceof Error ? err.message : err); - return c.json({ error: 'AI drafting failed. Please try again.' }, 502); - } -}); - -// ── Step 4: Risk Scanner ─────────────────────────────────── - -litigationRoutes.post('/qc', async (c) => { - const raw = await c.req.json().catch(() => null); - const parsed = qcSchema.safeParse(raw); - if (!parsed.success) { - return c.json({ error: 'Invalid request', details: parsed.error.flatten().fieldErrors }, 400); - } - - const { rawNotes, draftEmail } = parsed.data; - - const systemPrompt = `You are a rigorous Quality Control AI for litigation communications. +const FALLBACK_QC_PROMPT = `You are a rigorous Quality Control AI for litigation communications. Compare the Drafted Email against the Original Source Notes. Find ANY violations in these categories: - HALLUCINATION: Information in the draft that is NOT present in the source notes @@ -159,20 +246,3 @@ Output a JSON array of objects with these fields: If there are no issues, output an empty array: [] Output ONLY valid JSON, no markdown fences or explanation.`; - - const userPrompt = `Original Source Notes:\n"${rawNotes}"\n\nDrafted Email:\n"${draftEmail}"`; - - try { - const result = await callAIGateway(c.env, systemPrompt, userPrompt); - // Parse the JSON response, handling potential markdown fences - const cleaned = result.replace(/```json\s*/g, '').replace(/```\s*/g, '').trim(); - const flags = JSON.parse(cleaned); - return c.json({ flags }); - } catch (err) { - console.error('[litigation/qc]', err instanceof Error ? err.message : err); - if (err instanceof SyntaxError) { - return c.json({ flags: [], warning: 'QC analysis returned non-parseable results' }); - } - return c.json({ error: 'AI QC scan failed. Please try again.' }, 502); - } -}); diff --git a/src/routes/mcp.ts b/src/routes/mcp.ts index e50739c..efd1028 100644 --- a/src/routes/mcp.ts +++ b/src/routes/mcp.ts @@ -1078,14 +1078,14 @@ async function executeTool(env: Env, sql: NeonQueryFunction, toolN jobType: args.type as ScrapeJobType | undefined, chittyId: args.chitty_id ? String(args.chitty_id) : undefined, limit: Math.min(Number(args.limit) || 20, 50), - }); + }, env); return result; } case 'get_scrape_job': { const id = String(args.id || '').trim(); if (!id) throw new Error('Missing argument: id'); - const job = await getJobStatus(sql, id); + const job = await getJobStatus(sql, id, env); if (!job) throw new Error('Scrape job not found'); return job; } @@ -1093,14 +1093,14 @@ async function executeTool(env: Env, sql: NeonQueryFunction, toolN case 'retry_scrape_job': { const id = String(args.id || '').trim(); if (!id) throw new Error('Missing argument: id'); - const success = await retryJob(sql, id); + const success = await retryJob(sql, id, env); if (!success) throw new Error('Job not found or not in retryable state (must be failed or dead_letter)'); return { ok: true, status: 'queued', message: 'Job re-queued for retry' }; } case 'get_dead_letters': { const limit = Math.min(Number(args.limit) || 20, 50); - const jobs = await getDeadLetters(sql, limit); + const jobs = await getDeadLetters(sql, limit, env); return { jobs, total: jobs.length }; } @@ -1115,7 +1115,7 @@ async function executeTool(env: Env, sql: NeonQueryFunction, toolN const jobId = await enqueueJob(sql, jobType, target, { chittyId: args.chitty_id ? String(args.chitty_id) : undefined, cronSource: 'mcp', - }); + }, env); return { ok: true, id: jobId, status: 'queued' }; } From 66da6cc67cb5e2c996646da4050bc8cddc5f7f24 Mon Sep 17 00:00:00 2001 From: chitcommit <208086304+chitcommit@users.noreply.github.com> Date: Mon, 23 Mar 2026 20:30:35 +0000 Subject: [PATCH 2/2] fix: update MCP tool count test to 43 The tool count increased from 38 to 43 after recent PRs added litigation and scrape tools. Test was stale. Co-Authored-By: Claude Opus 4.6 --- tests/mcp.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/mcp.test.ts b/tests/mcp.test.ts index afe58ae..8806f32 100644 --- a/tests/mcp.test.ts +++ b/tests/mcp.test.ts @@ -147,7 +147,7 @@ describe('MCP — tools/list', () => { const json = await res.json() as Record; const result = json.result as Record; const tools = result.tools as unknown[]; - expect(tools.length).toBe(38); + expect(tools.length).toBe(43); }); it('each tool has a name and inputSchema', async () => {