diff --git a/migrations/0013_scrape_jobs.sql b/migrations/0013_scrape_jobs.sql new file mode 100644 index 0000000..fd4fa74 --- /dev/null +++ b/migrations/0013_scrape_jobs.sql @@ -0,0 +1,31 @@ +-- 0013_scrape_jobs.sql — Scrape job orchestration + identity binding + +-- Scrape job queue with retry, status tracking, and ChittyID binding +CREATE TABLE IF NOT EXISTS cc_scrape_jobs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + chitty_id VARCHAR(64), + job_type VARCHAR(50) NOT NULL, + target JSONB NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'queued', + attempt INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 3, + scheduled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + result JSONB, + error_message TEXT, + parent_job_id UUID REFERENCES cc_scrape_jobs(id), + cron_source VARCHAR(30), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_cc_scrape_jobs_status ON cc_scrape_jobs(status, scheduled_at); +CREATE INDEX idx_cc_scrape_jobs_type ON cc_scrape_jobs(job_type); +CREATE INDEX idx_cc_scrape_jobs_chitty ON cc_scrape_jobs(chitty_id); + +-- Add chitty_id to existing tables for identity binding +ALTER TABLE cc_sync_log ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64); +ALTER TABLE cc_legal_deadlines ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64); +ALTER TABLE cc_properties ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64); +ALTER TABLE cc_documents ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64); +ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64); diff --git a/src/db/schema.ts b/src/db/schema.ts index 24baea8..5d5dce7 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,4 +1,4 @@ -import { pgTable, uuid, text, numeric, boolean, integer, date, timestamp, jsonb, index } from 'drizzle-orm/pg-core'; +import { pgTable, uuid, varchar, text, numeric, boolean, integer, date, timestamp, jsonb, index } from 'drizzle-orm/pg-core'; import { sql } from 'drizzle-orm'; // ── Accounts ────────────────────────────────────────────────── @@ -21,6 +21,7 @@ export const ccAccounts = pgTable('cc_accounts', { // ── Obligations ─────────────────────────────────────────────── export const ccObligations = pgTable('cc_obligations', { id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), accountId: uuid('account_id').references(() => ccAccounts.id), category: text('category').notNull(), subcategory: text('subcategory'), @@ -73,6 +74,7 @@ export const ccTransactions = pgTable('cc_transactions', { // ── Properties ──────────────────────────────────────────────── export const ccProperties = pgTable('cc_properties', { id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), propertyName: text('property_name'), address: text('address').notNull(), unit: text('unit'), @@ -93,6 +95,7 @@ export const ccProperties = pgTable('cc_properties', { // ── Legal Deadlines ─────────────────────────────────────────── export const ccLegalDeadlines = pgTable('cc_legal_deadlines', { id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), caseRef: text('case_ref').notNull(), caseSystem: text('case_system'), deadlineType: text('deadline_type').notNull(), @@ -148,6 +151,7 @@ export const ccDisputeCorrespondence = pgTable('cc_dispute_correspondence', { // ── Documents ───────────────────────────────────────────────── export const ccDocuments = pgTable('cc_documents', { id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), docType: text('doc_type').notNull(), source: text('source').notNull(), filename: text('filename'), @@ -284,6 +288,7 @@ export const ccPaymentPlans = pgTable('cc_payment_plans', { // ── Sync Log ────────────────────────────────────────────────── export const ccSyncLog = pgTable('cc_sync_log', { id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), source: text('source').notNull(), syncType: text('sync_type').notNull(), status: text('status').notNull(), @@ -323,3 +328,26 @@ export const ccTasks = pgTable('cc_tasks', { priorityIdx: index('idx_cc_tasks_priority').on(table.priority), typeIdx: index('idx_cc_tasks_type').on(table.taskType), })); + +// ── Scrape Jobs ───────────────────────────────────────────── +export const ccScrapeJobs = pgTable('cc_scrape_jobs', { + id: uuid('id').primaryKey().defaultRandom(), + chittyId: varchar('chitty_id', { length: 64 }), + jobType: varchar('job_type', { length: 50 }).notNull(), + target: jsonb('target').notNull(), + status: varchar('status', { length: 20 }).notNull().default('queued'), + attempt: integer('attempt').notNull().default(0), + maxAttempts: integer('max_attempts').notNull().default(3), + scheduledAt: timestamp('scheduled_at', { withTimezone: true }).defaultNow(), + startedAt: timestamp('started_at', { withTimezone: true }), + completedAt: timestamp('completed_at', { withTimezone: true }), + result: jsonb('result'), + errorMessage: text('error_message'), + parentJobId: uuid('parent_job_id'), + cronSource: varchar('cron_source', { length: 30 }), + createdAt: timestamp('created_at', { withTimezone: true }).defaultNow(), +}, (table) => ({ + statusIdx: index('idx_cc_scrape_jobs_status').on(table.status, table.scheduledAt), + typeIdx: index('idx_cc_scrape_jobs_type').on(table.jobType), + chittyIdx: index('idx_cc_scrape_jobs_chitty').on(table.chittyId), +})); diff --git a/src/index.ts b/src/index.ts index 75abe67..95412c1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -22,12 +22,14 @@ import { paymentPlanRoutes } from './routes/payment-plan'; import { revenueRoutes } from './routes/revenue'; import { emailConnectionRoutes } from './routes/email-connections'; import { chatRoutes } from './routes/chat'; +import { litigationRoutes } from './routes/litigation'; import { taskRoutes } from './routes/tasks'; import { sendBeacon } from './lib/beacon'; import { contextRoutes } from './routes/context'; import { connectRoutes } from './routes/connect'; import { ledgerRoutes } from './routes/ledger'; import { tokenManagementRoutes } from './routes/token-management'; +import { jobRoutes } from './routes/jobs'; export type Env = { AI: Ai; @@ -121,6 +123,7 @@ app.route('/api/payment-plan', paymentPlanRoutes); app.route('/api/revenue', revenueRoutes); app.route('/api/email-connections', emailConnectionRoutes); app.route('/api/chat', chatRoutes); +app.route('/api/litigation', litigationRoutes); app.route('/api/tasks', taskRoutes); // Identity (authenticated) app.route('/api/v1', metaRoutes); @@ -132,6 +135,8 @@ app.route('/api/v1', connectRoutes); app.route('/api/v1', ledgerRoutes); // Token management (authenticated admin) app.route('/api/v1', tokenManagementRoutes); +// Scrape job management (authenticated) +app.route('/api/v1', jobRoutes); // MCP server — authenticated via shared token in KV app.use('/mcp/*', mcpAuthMiddleware); @@ -141,7 +146,7 @@ export default { fetch: app.fetch, async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) { const sql = getDb(env); - ctx.waitUntil(runCronSync(event, env, sql)); + ctx.waitUntil(runCronSync(event, env, sql, ctx)); ctx.waitUntil(sendBeacon(env)); }, }; diff --git a/src/lib/cron.ts b/src/lib/cron.ts index 7940f63..826db4b 100644 --- a/src/lib/cron.ts +++ b/src/lib/cron.ts @@ -7,6 +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'; /** * Cron sync orchestrator. @@ -18,6 +19,7 @@ export async function runCronSync( event: ScheduledEvent, env: Env, sql: NeonQueryFunction, + ctx?: ExecutionContext, ): Promise { const cronSources: Record = { '0 12 * * *': 'daily_api', @@ -136,15 +138,27 @@ export async function runCronSync( } if (source === 'utility_scrape') { - // Weekly utility portal scrapes via ChittyRouter + // Weekly utility portal scrapes via dispatcher + const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined; const utilityTargets = ['comed', 'peoples_gas', 'xfinity']; for (const target of utilityTargets) { try { - recordsSynced += await syncPortal(env, sql, target); + await enqueueJob(sql, 'portal_scrape', { portal: target }, { + chittyId, + cronSource: 'utility_scrape', + }); } catch (err) { - console.error(`[cron:utility:${target}] failed:`, err); + console.error(`[cron:utility:${target}] enqueue failed:`, err); } } + // Process all queued utility jobs + try { + const queueResult = await processQueue(sql, env, ctx); + recordsSynced += queueResult.succeeded; + console.log(`[cron:utility] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`); + } catch (err) { + console.error('[cron:utility] processQueue failed:', err); + } // Also pull email-parsed bills (also called in daily_api — upsert prevents duplicates) try { @@ -156,7 +170,15 @@ export async function runCronSync( if (source === 'court_docket') { try { - recordsSynced += await syncCourtDocket(env, sql); + // Enqueue via dispatcher for retry + fan-out + const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined; + await enqueueJob(sql, 'court_docket', { case_number: '2024D007847' }, { + chittyId, + cronSource: 'court_docket', + }); + const queueResult = await processQueue(sql, env, ctx); + recordsSynced += queueResult.succeeded; + console.log(`[cron:court_docket] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`); } catch (err) { console.error('[cron:court_docket] failed:', err); } @@ -164,7 +186,7 @@ export async function runCronSync( if (source === 'monthly_check') { try { - recordsSynced += await syncMonthlyChecks(env, sql); + recordsSynced += await syncMonthlyChecksViaDispatcher(env, sql, ctx); } catch (err) { console.error('[cron:monthly_check] failed:', err); } @@ -544,6 +566,48 @@ async function syncMonthlyChecks(env: Env, sql: NeonQueryFunction) return synced; } +/** + * Monthly scrapers via dispatcher — enqueues Mr. Cooper + all property tax PINs as jobs. + */ +async function syncMonthlyChecksViaDispatcher( + env: Env, + sql: NeonQueryFunction, + ctx?: ExecutionContext, +): Promise { + const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined; + + // Enqueue Mr. Cooper + try { + await enqueueJob(sql, 'mr_cooper', { property: 'addison' }, { + chittyId, + cronSource: 'monthly_check', + }); + } catch (err) { + console.error('[cron:mr_cooper] enqueue failed:', err); + } + + // Enqueue Cook County tax for each property with a PIN + try { + const properties = await sql`SELECT id, tax_pin FROM cc_properties WHERE tax_pin IS NOT NULL`; + for (const prop of properties) { + await enqueueJob(sql, 'cook_county_tax', { + pin: prop.tax_pin as string, + property_id: prop.id as string, + }, { + chittyId, + cronSource: 'monthly_check', + }); + } + } catch (err) { + console.error('[cron:cook_county_tax] enqueue failed:', err); + } + + // Process all enqueued monthly jobs + const queueResult = await processQueue(sql, env, ctx); + console.log(`[cron:monthly] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`); + return queueResult.succeeded; +} + /** * Sync a bill portal via ChittyRouter gateway. * ChittyRouter fetches credentials from ChittyConnect, dispatches to ChittyScrape, diff --git a/src/lib/fan-out.ts b/src/lib/fan-out.ts new file mode 100644 index 0000000..cb9df5d --- /dev/null +++ b/src/lib/fan-out.ts @@ -0,0 +1,193 @@ +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', + }, + body: JSON.stringify({ + entityType: 'scrape', + entityId: ctx.jobId, + action: 'completed', + actor: ctx.chittyId || 'chittycommand', + actorType: ctx.chittyId ? 'entity' : '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/job-dispatcher.ts b/src/lib/job-dispatcher.ts new file mode 100644 index 0000000..8b467d0 --- /dev/null +++ b/src/lib/job-dispatcher.ts @@ -0,0 +1,501 @@ +import type { NeonQueryFunction } from '@neondatabase/serverless'; +import type { Env } from '../index'; +import { scrapeClient, routerClient } from './integrations'; +import { fanOutScrapeResult } from './fan-out'; + +export type ScrapeJobType = + | 'court_docket' + | 'cook_county_tax' + | 'mr_cooper' + | 'portal_scrape'; + +export type ScrapeJobStatus = + | 'queued' + | 'running' + | 'completed' + | 'failed' + | 'retrying' + | 'dead_letter'; + +export interface EnqueueOptions { + chittyId?: string; + maxAttempts?: number; + scheduledAt?: Date; + cronSource?: string; + parentJobId?: string; +} + +export interface ScrapeJob { + id: string; + chittyId: string | null; + jobType: ScrapeJobType; + target: Record; + status: ScrapeJobStatus; + attempt: number; + maxAttempts: number; + scheduledAt: string; + startedAt: string | null; + completedAt: string | null; + result: Record | null; + errorMessage: string | null; + parentJobId: string | null; + cronSource: string | null; + createdAt: string; +} + +/** + * Enqueue a new scrape job. Returns the job ID. + */ +export async function enqueueJob( + sql: NeonQueryFunction, + jobType: ScrapeJobType, + target: Record, + opts: EnqueueOptions = {}, +): Promise { + 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) + VALUES ( + ${jobType}, + ${JSON.stringify(target)}::jsonb, + ${opts.chittyId || null}, + ${opts.maxAttempts || 3}, + ${scheduledAt}, + ${opts.cronSource || null}, + ${opts.parentJobId || null} + ) + RETURNING id + `; + return row.id as string; +} + +/** + * 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. + */ +export async function processQueue( + sql: NeonQueryFunction, + env: Env, + 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++; + } + } + + return { processed: jobs.length, succeeded, failed }; +} + +/** + * Get job status by ID. + */ +export async function getJobStatus( + sql: NeonQueryFunction, + jobId: string, +): Promise { + const [row] = await sql`SELECT * FROM cc_scrape_jobs WHERE id = ${jobId}`; + return row ? mapJobRow(row) : null; +} + +/** + * List jobs with optional filters. + */ +export async function listJobs( + sql: NeonQueryFunction, + filters: { + status?: ScrapeJobStatus; + jobType?: ScrapeJobType; + chittyId?: string; + limit?: number; + offset?: number; + } = {}, +): Promise<{ jobs: ScrapeJob[]; total: number }> { + 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} + `; + 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} + `; + 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}) + AND (${filters.jobType || null}::text IS NULL OR job_type = ${filters.jobType || null}) + AND (${filters.chittyId || null}::text IS NULL OR chitty_id = ${filters.chittyId || null}) + ORDER BY created_at DESC LIMIT ${limit} OFFSET ${offset} + `; + countRows = await sql` + SELECT COUNT(*)::int AS total FROM cc_scrape_jobs + WHERE (${filters.status || null}::text IS NULL OR status = ${filters.status || null}) + AND (${filters.jobType || null}::text IS NULL OR job_type = ${filters.jobType || null}) + AND (${filters.chittyId || null}::text IS NULL OR chitty_id = ${filters.chittyId || null}) + `; + } + + return { + jobs: rows.map(mapJobRow), + total: countRows[0]?.total || 0, + }; +} + +/** + * Get dead-lettered jobs for review. + */ +export async function getDeadLetters( + sql: NeonQueryFunction, + limit = 50, +): Promise { + const rows = await sql` + SELECT * FROM cc_scrape_jobs WHERE status = 'dead_letter' + ORDER BY completed_at DESC LIMIT ${limit} + `; + return rows.map(mapJobRow); +} + +/** + * Retry a failed/dead-lettered job. + */ +export async function retryJob( + sql: NeonQueryFunction, + jobId: string, +): Promise { + const [row] = await sql` + UPDATE cc_scrape_jobs + SET status = 'queued', attempt = 0, error_message = NULL, + scheduled_at = NOW(), started_at = NULL, completed_at = NULL, result = NULL + WHERE id = ${jobId} AND status IN ('failed', 'dead_letter') + RETURNING id + `; + 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'); + + 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 }; +} + +// ── Helpers ──────────────────────────────────────────────────── + +function mapJobRow(row: Record): ScrapeJob { + return { + id: row.id as string, + chittyId: row.chitty_id as string | null, + jobType: row.job_type as ScrapeJobType, + target: row.target as Record, + status: row.status as ScrapeJobStatus, + attempt: row.attempt as number, + maxAttempts: row.max_attempts as number, + scheduledAt: row.scheduled_at as string, + startedAt: row.started_at as string | null, + completedAt: row.completed_at as string | null, + result: row.result as Record | null, + errorMessage: row.error_message as string | null, + parentJobId: row.parent_job_id as string | null, + cronSource: row.cron_source as string | null, + createdAt: row.created_at as string, + }; +} diff --git a/src/routes/jobs.ts b/src/routes/jobs.ts new file mode 100644 index 0000000..9f216e4 --- /dev/null +++ b/src/routes/jobs.ts @@ -0,0 +1,88 @@ +import { Hono } from 'hono'; +import type { Env } from '../index'; +import type { AuthVariables } from '../middleware/auth'; +import { getDb } from '../lib/db'; +import { + listJobs, + getJobStatus, + retryJob, + getDeadLetters, + enqueueJob, + processQueue, +} from '../lib/job-dispatcher'; +import type { ScrapeJobType, ScrapeJobStatus } from '../lib/job-dispatcher'; + +export const jobRoutes = new Hono<{ Bindings: Env; Variables: AuthVariables }>(); + +// List jobs with optional filters +jobRoutes.get('/jobs', async (c) => { + const sql = getDb(c.env); + const status = c.req.query('status') as ScrapeJobStatus | undefined; + const jobType = c.req.query('type') as ScrapeJobType | undefined; + const chittyId = c.req.query('chitty_id'); + 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 }); + 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')); + if (!job) return c.json({ error: 'Job not found' }, 404); + return c.json(job); +}); + +// Get dead-lettered jobs +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); + 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')); + 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' }); +}); + +// Manually enqueue a new scrape job +jobRoutes.post('/jobs', async (c) => { + const sql = getDb(c.env); + const body = await c.req.json<{ + job_type: ScrapeJobType; + target: Record; + chitty_id?: string; + max_attempts?: number; + }>(); + + if (!body.job_type || !body.target) { + return c.json({ error: 'job_type and target are required' }, 400); + } + + const validTypes: ScrapeJobType[] = ['court_docket', 'cook_county_tax', 'mr_cooper', 'portal_scrape']; + if (!validTypes.includes(body.job_type)) { + return c.json({ error: `Invalid job_type. Must be one of: ${validTypes.join(', ')}` }, 400); + } + + const jobId = await enqueueJob(sql, body.job_type, body.target, { + chittyId: body.chitty_id, + maxAttempts: body.max_attempts, + cronSource: 'manual', + }); + + return c.json({ id: jobId, status: 'queued' }, 201); +}); + +// 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); + return c.json(result); +}); diff --git a/src/routes/litigation.ts b/src/routes/litigation.ts new file mode 100644 index 0000000..94254d1 --- /dev/null +++ b/src/routes/litigation.ts @@ -0,0 +1,178 @@ +import { Hono } from 'hono'; +import { z } from 'zod'; +import type { Env } from '../index'; +import type { AuthVariables } from '../middleware/auth'; + +export const litigationRoutes = new Hono<{ Bindings: Env; Variables: AuthVariables }>(); + +const synthesizeSchema = z.object({ + rawNotes: z.string().min(1).max(50000), + property: z.string().max(500).optional(), + caseNumber: z.string().max(100).optional(), +}); + +const draftSchema = z.object({ + synthesizedFacts: z.string().min(1).max(50000), + focus: z.string().max(200), + recipient: z.string().max(200), +}); + +const qcSchema = z.object({ + rawNotes: z.string().min(1).max(50000), + draftEmail: z.string().min(1).max(10000), +}); + +async function callAIGateway( + env: Env, + systemPrompt: string, + userPrompt: string, + maxTokens = 4096, +): Promise { + const gateway = env.AI.gateway('chittygateway'); + const gatewayUrl = await gateway.getUrl(); + + const chatModel = await env.COMMAND_KV.get('chat:model').catch(() => null) + || 'dynamic/chittycommand'; + + const response = await fetch( + new URL('compat/chat/completions', gatewayUrl).toString(), + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + model: chatModel, + messages: [ + { role: 'system', content: systemPrompt }, + { role: 'user', content: userPrompt }, + ], + stream: false, + max_tokens: maxTokens, + }), + signal: AbortSignal.timeout(60_000), + }, + ); + + if (!response.ok) { + const errText = await response.text().catch(() => ''); + console.error('[litigation] AI gateway error:', response.status, errText); + throw new Error(`AI gateway error (${response.status})`); + } + + const result = await response.json() as { + choices?: { message?: { content?: string } }[]; + }; + + return result.choices?.[0]?.message?.content || ''; +} + +// ── 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 systemPrompt = `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 +- Sale / Listing Status +- Prior Communications +- Financial / Fee Issues +- Sanctions / Motions + +Use bullet points. CRITICAL: Every single bullet MUST begin with one of these EXACT tags: +[GIVEN] — if explicitly stated in the source material +[DERIVED] — if a logical inference from the material +[UNKNOWN] — if context requires it but the information is missing + +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. +Rules: +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. +6. Do NOT include facts marked [UNKNOWN] in the email body. +7. Facts marked [DERIVED] must be hedged with language like "Based on...", "It appears...", "My understanding is...". +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. +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 +- MISSING: Crucial context from the source left out of the draft +- OVER-DISCLOSURE: Draft reveals unnecessary sensitive or strategic information +- AMBIGUOUS: Requests or statements that are unclear or could be misinterpreted + +Output a JSON array of objects with these fields: +{ "flagType": "HALLUCINATION|MISSING|OVER-DISCLOSURE|AMBIGUOUS", "location": "where in the draft", "issue": "description", "suggestedFix": "how to fix it" } + +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 ee3cb1c..e50739c 100644 --- a/src/routes/mcp.ts +++ b/src/routes/mcp.ts @@ -3,6 +3,8 @@ import type { Env } from '../index'; import type { AuthVariables } from '../middleware/auth'; import { getDb, typedRows } from '../lib/db'; import type { NeonQueryFunction } from '@neondatabase/serverless'; +import { listJobs, getJobStatus, retryJob, getDeadLetters, enqueueJob } from '../lib/job-dispatcher'; +import type { ScrapeJobType, ScrapeJobStatus } from '../lib/job-dispatcher'; /** * MCP (Model Context Protocol) server for ChittyCommand. @@ -291,6 +293,67 @@ const TOOLS = [ required: ['id', 'verification_artifact'], }, }, + // ── Scrape Jobs ──────────────────────────────────────────── + { + name: 'query_scrape_jobs', + description: 'List scrape jobs with optional filters (status, type, chitty_id). Returns paginated results.', + inputSchema: { + type: 'object' as const, + properties: { + status: { type: 'string', description: 'Filter by status', enum: ['queued', 'running', 'completed', 'failed', 'retrying', 'dead_letter'] }, + type: { type: 'string', description: 'Filter by job type', enum: ['court_docket', 'cook_county_tax', 'mr_cooper', 'portal_scrape'] }, + chitty_id: { type: 'string', description: 'Filter by ChittyID' }, + limit: { type: 'number', description: 'Max results (default 20)' }, + }, + required: [] as string[], + }, + }, + { + name: 'get_scrape_job', + description: 'Get detailed status of a specific scrape job by ID.', + inputSchema: { + type: 'object' as const, + properties: { + id: { type: 'string', description: 'Scrape job UUID' }, + }, + required: ['id'], + }, + }, + { + name: 'retry_scrape_job', + description: 'Re-queue a failed or dead-lettered scrape job for retry.', + inputSchema: { + type: 'object' as const, + properties: { + id: { type: 'string', description: 'Scrape job UUID' }, + }, + required: ['id'], + }, + }, + { + name: 'get_dead_letters', + description: 'List scrape jobs that exhausted all retry attempts (dead-lettered).', + inputSchema: { + type: 'object' as const, + properties: { + limit: { type: 'number', description: 'Max results (default 20)' }, + }, + required: [] as string[], + }, + }, + { + name: 'enqueue_scrape_job', + description: 'Manually enqueue a new scrape job for execution.', + inputSchema: { + type: 'object' as const, + properties: { + job_type: { type: 'string', description: 'Type of scrape', enum: ['court_docket', 'cook_county_tax', 'mr_cooper', 'portal_scrape'] }, + target: { type: 'object', description: 'Target parameters (e.g. {case_number}, {pin}, {property}, {portal})' }, + chitty_id: { type: 'string', description: 'Optional ChittyID to bind results to' }, + }, + required: ['job_type', 'target'], + }, + }, ]; // MCP endpoint — handles JSON-RPC 2.0 requests @@ -1008,6 +1071,54 @@ async function executeTool(env: Env, sql: NeonQueryFunction, toolN return { ok: true, task: updated[0] }; } + // ── Scrape Job Tools ──────────────────────────────────── + case 'query_scrape_jobs': { + const result = await listJobs(sql, { + status: args.status as ScrapeJobStatus | undefined, + jobType: args.type as ScrapeJobType | undefined, + chittyId: args.chitty_id ? String(args.chitty_id) : undefined, + limit: Math.min(Number(args.limit) || 20, 50), + }); + 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); + if (!job) throw new Error('Scrape job not found'); + return job; + } + + case 'retry_scrape_job': { + const id = String(args.id || '').trim(); + if (!id) throw new Error('Missing argument: id'); + const success = await retryJob(sql, id); + 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); + return { jobs, total: jobs.length }; + } + + case 'enqueue_scrape_job': { + const jobType = String(args.job_type || '').trim() as ScrapeJobType; + const target = args.target as Record; + if (!jobType || !target) throw new Error('Missing arguments: job_type, target'); + + const validTypes: ScrapeJobType[] = ['court_docket', 'cook_county_tax', 'mr_cooper', 'portal_scrape']; + if (!validTypes.includes(jobType)) throw new Error(`Invalid job_type. Must be one of: ${validTypes.join(', ')}`); + + const jobId = await enqueueJob(sql, jobType, target, { + chittyId: args.chitty_id ? String(args.chitty_id) : undefined, + cronSource: 'mcp', + }); + return { ok: true, id: jobId, status: 'queued' }; + } + default: throw new Error(`Unknown tool: ${toolName}`); } diff --git a/ui/src/components/Sidebar.tsx b/ui/src/components/Sidebar.tsx index 92cbd56..a544c1a 100644 --- a/ui/src/components/Sidebar.tsx +++ b/ui/src/components/Sidebar.tsx @@ -1,22 +1,52 @@ import { NavLink } from 'react-router-dom'; import { cn } from '../lib/utils'; import { - LayoutDashboard, Zap, Receipt, ShieldAlert, Wallet, Scale, + LayoutDashboard, Zap, Receipt, ShieldAlert, Wallet, Scale, Gavel, Lightbulb, TrendingUp, Upload, Settings, LogOut, } from 'lucide-react'; import { logout, getUser } from '../lib/auth'; -const navItems = [ - { path: '/', label: 'Dashboard', icon: LayoutDashboard }, - { path: '/queue', label: 'Action Queue', icon: Zap }, - { path: '/bills', label: 'Bills', icon: Receipt }, - { path: '/disputes', label: 'Disputes', icon: ShieldAlert }, - { path: '/accounts', label: 'Accounts', icon: Wallet }, - { path: '/legal', label: 'Legal', icon: Scale }, - { path: '/recommendations', label: 'AI Recs', icon: Lightbulb }, - { path: '/cashflow', label: 'Cash Flow', icon: TrendingUp }, - { path: '/upload', label: 'Upload', icon: Upload }, - { path: '/settings', label: 'Settings', icon: Settings }, +interface NavGroup { + label?: string; + items: { path: string; label: string; icon: typeof LayoutDashboard }[]; +} + +const navGroups: NavGroup[] = [ + { + items: [ + { path: '/', label: 'Command Center', icon: LayoutDashboard }, + ], + }, + { + label: 'Money', + items: [ + { path: '/bills', label: 'Bills', icon: Receipt }, + { path: '/accounts', label: 'Accounts', icon: Wallet }, + { path: '/cashflow', label: 'Cash Flow', icon: TrendingUp }, + ], + }, + { + label: 'Disputes', + items: [ + { path: '/disputes', label: 'Active Disputes', icon: ShieldAlert }, + { path: '/legal', label: 'Legal Deadlines', icon: Scale }, + { path: '/litigation', label: 'Litigation AI', icon: Gavel }, + ], + }, + { + label: 'Intelligence', + items: [ + { path: '/queue', label: 'Action Queue', icon: Zap }, + { path: '/recommendations', label: 'AI Recs', icon: Lightbulb }, + ], + }, + { + label: 'System', + items: [ + { path: '/upload', label: 'Documents', icon: Upload }, + { path: '/settings', label: 'Settings', icon: Settings }, + ], + }, ]; export function Sidebar() { @@ -28,39 +58,48 @@ export function Sidebar() {

ChittyCommand

-

Control Plane

+

Control Plane

-