lorentz
a0a6e7f192
IT Glue's /flexible_assets endpoint refuses requests without a filter[flexible-asset-type-id] (returns 422 "Cannot index flexible assets without providing a flexible asset type ID filter"). The analyzer's Stage 2 search was caught and tolerated, but never returned docs. Added getFlexibleAssetsForOrganization(orgId) on ITGlueClient. It fetches the type list once per process (memoized), then fans out per-type fetches with Promise.allSettled so a permission-restricted type doesn't poison the whole org. Wired into itglue-search and aggregate-persistence. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
506 lines
17 KiB
TypeScript
506 lines
17 KiB
TypeScript
/**
|
||
* Persistence + runner for aggregate reports.
|
||
*
|
||
* Spec: docs/ticket-analyzer-phase2-spec.md → Sections D.4–D.6
|
||
*/
|
||
|
||
import postgresClient from '@/lib/services/postgres-client';
|
||
import {
|
||
type AggregateFingerprint,
|
||
type AggregateReduceResponse,
|
||
type AggregateReportStatus,
|
||
type StageExecutionRecord,
|
||
} from '@/lib/types/analyzer';
|
||
import { getITGlueClient } from '@/lib/services/itglue-client';
|
||
import { runAggregateReduceStage } from './stages/aggregate-reduce';
|
||
|
||
interface AggregateReportRow {
|
||
id: string;
|
||
generated_by_user_id: string | null;
|
||
generated_at: Date;
|
||
filter_criteria: unknown;
|
||
analysis_ids: string[];
|
||
ticket_count: number;
|
||
include_itglue_context: boolean;
|
||
report_title: string | null;
|
||
category_distribution: Record<string, number> | null;
|
||
client_distribution: Record<string, number> | null;
|
||
resolution_path_distribution: Record<string, number> | null;
|
||
root_cause_distribution: Record<string, number> | null;
|
||
date_range_actual: { earliest: string | null; latest: string | null } | null;
|
||
documentation_gaps: unknown;
|
||
process_gaps: unknown;
|
||
client_patterns: unknown;
|
||
recurrence_clusters: unknown;
|
||
systemic_observations: unknown;
|
||
recommended_actions: unknown;
|
||
narrative_summary: string | null;
|
||
executive_summary: string | null;
|
||
total_input_tokens: number | null;
|
||
total_output_tokens: number | null;
|
||
estimated_cost_usd: string | null;
|
||
model_used: string | null;
|
||
itglue_context_included: boolean | null;
|
||
status: AggregateReportStatus;
|
||
error_message: string | null;
|
||
}
|
||
|
||
export interface AggregateReportSummary {
|
||
id: string;
|
||
generatedByUserId: string | null;
|
||
generatedAt: string;
|
||
filterCriteria: unknown;
|
||
analysisIds: string[];
|
||
ticketCount: number;
|
||
includeItglueContext: boolean;
|
||
reportTitle: string | null;
|
||
status: AggregateReportStatus;
|
||
errorMessage: string | null;
|
||
// SQL outputs
|
||
categoryDistribution: Record<string, number> | null;
|
||
clientDistribution: Record<string, number> | null;
|
||
resolutionPathDistribution: Record<string, number> | null;
|
||
rootCauseDistribution: Record<string, number> | null;
|
||
dateRangeActual: { earliest: string | null; latest: string | null } | null;
|
||
// LLM outputs
|
||
documentationGaps: unknown;
|
||
processGaps: unknown;
|
||
clientPatterns: unknown;
|
||
recurrenceClusters: unknown;
|
||
systemicObservations: unknown;
|
||
recommendedActions: unknown;
|
||
narrativeSummary: string | null;
|
||
executiveSummary: string | null;
|
||
// Cost
|
||
totalInputTokens: number | null;
|
||
totalOutputTokens: number | null;
|
||
estimatedCostUsd: number | null;
|
||
modelUsed: string | null;
|
||
}
|
||
|
||
function rowToSummary(r: AggregateReportRow): AggregateReportSummary {
|
||
return {
|
||
id: r.id,
|
||
generatedByUserId: r.generated_by_user_id,
|
||
generatedAt: r.generated_at.toISOString(),
|
||
filterCriteria: r.filter_criteria,
|
||
analysisIds: r.analysis_ids,
|
||
ticketCount: r.ticket_count,
|
||
includeItglueContext: r.include_itglue_context,
|
||
reportTitle: r.report_title,
|
||
status: r.status,
|
||
errorMessage: r.error_message,
|
||
categoryDistribution: r.category_distribution,
|
||
clientDistribution: r.client_distribution,
|
||
resolutionPathDistribution: r.resolution_path_distribution,
|
||
rootCauseDistribution: r.root_cause_distribution,
|
||
dateRangeActual: r.date_range_actual,
|
||
documentationGaps: r.documentation_gaps,
|
||
processGaps: r.process_gaps,
|
||
clientPatterns: r.client_patterns,
|
||
recurrenceClusters: r.recurrence_clusters,
|
||
systemicObservations: r.systemic_observations,
|
||
recommendedActions: r.recommended_actions,
|
||
narrativeSummary: r.narrative_summary,
|
||
executiveSummary: r.executive_summary,
|
||
totalInputTokens: r.total_input_tokens,
|
||
totalOutputTokens: r.total_output_tokens,
|
||
estimatedCostUsd: r.estimated_cost_usd === null ? null : Number(r.estimated_cost_usd),
|
||
modelUsed: r.model_used,
|
||
};
|
||
}
|
||
|
||
const REPORT_SELECT = `
|
||
id::text AS id,
|
||
generated_by_user_id, generated_at,
|
||
filter_criteria, analysis_ids::text[] AS analysis_ids,
|
||
ticket_count, include_itglue_context, report_title,
|
||
category_distribution, client_distribution, resolution_path_distribution,
|
||
root_cause_distribution, date_range_actual,
|
||
documentation_gaps, process_gaps, client_patterns, recurrence_clusters,
|
||
systemic_observations, recommended_actions,
|
||
narrative_summary, executive_summary,
|
||
total_input_tokens, total_output_tokens,
|
||
estimated_cost_usd::text AS estimated_cost_usd,
|
||
model_used, itglue_context_included,
|
||
status, error_message
|
||
`;
|
||
|
||
export interface CreateAggregateReportInput {
|
||
generatedByUserId: string | null;
|
||
filterCriteria: unknown;
|
||
analysisIds: string[];
|
||
ticketCount: number;
|
||
includeItglueContext: boolean;
|
||
reportTitle: string | null;
|
||
}
|
||
|
||
export async function createAggregateReport(
|
||
input: CreateAggregateReportInput
|
||
): Promise<{ id: string }> {
|
||
const res = await postgresClient.query<{ id: string }>(
|
||
`INSERT INTO analyzer_aggregate_reports
|
||
(generated_by_user_id, filter_criteria, analysis_ids,
|
||
ticket_count, include_itglue_context, report_title, status)
|
||
VALUES ($1, $2::jsonb, $3::uuid[], $4, $5, $6, 'pending')
|
||
RETURNING id::text AS id`,
|
||
[
|
||
input.generatedByUserId,
|
||
JSON.stringify(input.filterCriteria),
|
||
input.analysisIds,
|
||
input.ticketCount,
|
||
input.includeItglueContext,
|
||
input.reportTitle,
|
||
]
|
||
);
|
||
return { id: res.rows[0].id };
|
||
}
|
||
|
||
export async function getAggregateReport(
|
||
id: string
|
||
): Promise<AggregateReportSummary | null> {
|
||
const res = await postgresClient.query<AggregateReportRow>(
|
||
`SELECT ${REPORT_SELECT} FROM analyzer_aggregate_reports WHERE id = $1`,
|
||
[id]
|
||
);
|
||
if (res.rowCount === 0) return null;
|
||
return rowToSummary(res.rows[0]);
|
||
}
|
||
|
||
export async function listAggregateReports(opts: {
|
||
limit?: number;
|
||
offset?: number;
|
||
generatedByUserId?: string;
|
||
}): Promise<AggregateReportSummary[]> {
|
||
const limit = Math.min(opts.limit ?? 50, 200);
|
||
const offset = opts.offset ?? 0;
|
||
const params: unknown[] = [limit, offset];
|
||
let userClause = '';
|
||
if (opts.generatedByUserId) {
|
||
params.push(opts.generatedByUserId);
|
||
userClause = `WHERE generated_by_user_id = $${params.length}`;
|
||
}
|
||
const res = await postgresClient.query<AggregateReportRow>(
|
||
`SELECT ${REPORT_SELECT}
|
||
FROM analyzer_aggregate_reports
|
||
${userClause}
|
||
ORDER BY generated_at DESC
|
||
LIMIT $1 OFFSET $2`,
|
||
params
|
||
);
|
||
return res.rows.map(rowToSummary);
|
||
}
|
||
|
||
interface FingerprintRow {
|
||
id: string;
|
||
ticket_number: string;
|
||
aggregate_fingerprint: AggregateFingerprint;
|
||
triggered_at: Date;
|
||
}
|
||
|
||
async function loadFingerprints(
|
||
analysisIds: string[]
|
||
): Promise<{ ticket_number: string; fingerprint: AggregateFingerprint; triggered_at: Date }[]> {
|
||
if (analysisIds.length === 0) return [];
|
||
const res = await postgresClient.query<FingerprintRow>(
|
||
`SELECT id::text AS id, ticket_number, aggregate_fingerprint, triggered_at
|
||
FROM analyzer_analyses
|
||
WHERE id = ANY($1::uuid[])
|
||
AND aggregate_fingerprint IS NOT NULL
|
||
ORDER BY ticket_number, analysis_version DESC`,
|
||
[analysisIds]
|
||
);
|
||
return res.rows.map((r) => ({
|
||
ticket_number: r.ticket_number,
|
||
fingerprint: r.aggregate_fingerprint,
|
||
triggered_at: r.triggered_at,
|
||
}));
|
||
}
|
||
|
||
function bucketCount(items: string[]): Record<string, number> {
|
||
const out: Record<string, number> = {};
|
||
for (const i of items) out[i] = (out[i] ?? 0) + 1;
|
||
return out;
|
||
}
|
||
|
||
async function fetchITGlueDocTitles(
|
||
clientNames: string[]
|
||
): Promise<{ client_name: string; doc_titles: string[] }[]> {
|
||
let client;
|
||
try {
|
||
client = getITGlueClient();
|
||
} catch {
|
||
return []; // not configured — caller should fall back gracefully
|
||
}
|
||
const result: { client_name: string; doc_titles: string[] }[] = [];
|
||
for (const name of clientNames) {
|
||
try {
|
||
const org = await client.findOrganizationByName(name);
|
||
if (!org) continue;
|
||
const docs = await client.getFlexibleAssetsForOrganization(org.id);
|
||
const titles = docs
|
||
.map((d) => (d as { name?: string }).name)
|
||
.filter((t): t is string => typeof t === 'string')
|
||
.slice(0, 50);
|
||
result.push({ client_name: name, doc_titles: titles });
|
||
} catch {
|
||
// Tolerate per-client failures.
|
||
}
|
||
}
|
||
return result;
|
||
}
|
||
|
||
const ITGLUE_DOC_TITLE_CAP = 200;
|
||
|
||
export async function bulkInsertReportStageExecutions(
|
||
reportId: string,
|
||
records: StageExecutionRecord[]
|
||
): Promise<void> {
|
||
if (records.length === 0) return;
|
||
const values: unknown[] = [reportId];
|
||
const tuples: string[] = [];
|
||
for (const r of records) {
|
||
const base = values.length;
|
||
values.push(
|
||
r.stage,
|
||
r.stage_order,
|
||
r.model_id,
|
||
JSON.stringify(r.input_payload ?? {}),
|
||
JSON.stringify(r.output_payload ?? {}),
|
||
r.input_tokens,
|
||
r.output_tokens,
|
||
r.latency_ms,
|
||
r.started_at,
|
||
r.completed_at,
|
||
r.error_message
|
||
);
|
||
tuples.push(
|
||
`($1, $${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}::jsonb, ` +
|
||
`$${base + 5}::jsonb, $${base + 6}, $${base + 7}, $${base + 8}, ` +
|
||
`$${base + 9}, $${base + 10}, $${base + 11})`
|
||
);
|
||
}
|
||
await postgresClient.query(
|
||
`INSERT INTO analyzer_stage_executions
|
||
(aggregate_report_id, stage, stage_order, model_id,
|
||
input_payload, output_payload,
|
||
input_tokens, output_tokens, latency_ms,
|
||
started_at, completed_at, error_message)
|
||
VALUES ${tuples.join(', ')}`,
|
||
values
|
||
);
|
||
}
|
||
|
||
/**
|
||
* Fire-and-forget runner. Intended to be invoked from the POST endpoint with
|
||
* `void runAggregateReport(id)` — the route returns immediately, this updates
|
||
* the row when work completes (or fails).
|
||
*/
|
||
export async function runAggregateReport(reportId: string): Promise<void> {
|
||
const stageRecords: StageExecutionRecord[] = [];
|
||
try {
|
||
await postgresClient.query(
|
||
`UPDATE analyzer_aggregate_reports SET status = 'running' WHERE id = $1`,
|
||
[reportId]
|
||
);
|
||
|
||
const report = await getAggregateReport(reportId);
|
||
if (!report) throw new Error('report row vanished');
|
||
|
||
// ── Step 1: load fingerprints + compute SQL distributions ──
|
||
const sqlStart = new Date();
|
||
const fingerprints = await loadFingerprints(report.analysisIds);
|
||
if (fingerprints.length === 0) {
|
||
throw new Error('no analyses with fingerprints found for the given IDs');
|
||
}
|
||
const categories = bucketCount(fingerprints.map((f) => f.fingerprint.category));
|
||
const clients = bucketCount(fingerprints.map((f) => f.fingerprint.client_name));
|
||
const resolutionPaths = bucketCount(
|
||
fingerprints.map((f) => f.fingerprint.resolution_path)
|
||
);
|
||
const rootCauses = bucketCount(
|
||
fingerprints.map((f) => f.fingerprint.root_cause_class)
|
||
);
|
||
const dates = fingerprints.map((f) => f.triggered_at.getTime());
|
||
const dateRange = {
|
||
earliest: new Date(Math.min(...dates)).toISOString(),
|
||
latest: new Date(Math.max(...dates)).toISOString(),
|
||
};
|
||
const sqlEnd = new Date();
|
||
stageRecords.push({
|
||
stage: 'analyze', // Reusing 'analyze' since CHECK constraint enumerates only stage names; a future migration could add 'aggregate_sql' / 'aggregate_reduce'.
|
||
stage_order: 1,
|
||
model_id: null,
|
||
input_payload: { analysis_ids: report.analysisIds },
|
||
output_payload: {
|
||
category_distribution: categories,
|
||
client_distribution: clients,
|
||
resolution_path_distribution: resolutionPaths,
|
||
root_cause_distribution: rootCauses,
|
||
date_range_actual: dateRange,
|
||
fingerprint_count: fingerprints.length,
|
||
},
|
||
input_tokens: null,
|
||
output_tokens: null,
|
||
latency_ms: sqlEnd.getTime() - sqlStart.getTime(),
|
||
started_at: sqlStart,
|
||
completed_at: sqlEnd,
|
||
error_message: null,
|
||
});
|
||
|
||
// Persist partial results immediately so UI can show distributions.
|
||
await postgresClient.query(
|
||
`UPDATE analyzer_aggregate_reports
|
||
SET category_distribution = $2::jsonb,
|
||
client_distribution = $3::jsonb,
|
||
resolution_path_distribution = $4::jsonb,
|
||
root_cause_distribution = $5::jsonb,
|
||
date_range_actual = $6::jsonb
|
||
WHERE id = $1`,
|
||
[
|
||
reportId,
|
||
JSON.stringify(categories),
|
||
JSON.stringify(clients),
|
||
JSON.stringify(resolutionPaths),
|
||
JSON.stringify(rootCauses),
|
||
JSON.stringify(dateRange),
|
||
]
|
||
);
|
||
|
||
// ── Step 2: IT Glue context (optional) ──
|
||
let itglueDocTitles: { client_name: string; doc_titles: string[] }[] | undefined;
|
||
let itglueIncluded = false;
|
||
if (report.includeItglueContext) {
|
||
const uniqueClients = Array.from(
|
||
new Set(fingerprints.map((f) => f.fingerprint.client_name))
|
||
);
|
||
const itglueStart = new Date();
|
||
itglueDocTitles = await fetchITGlueDocTitles(uniqueClients);
|
||
// Cap to spec total (200 doc titles across all clients).
|
||
let remaining = ITGLUE_DOC_TITLE_CAP;
|
||
itglueDocTitles = itglueDocTitles.map((c) => {
|
||
if (remaining <= 0) return { client_name: c.client_name, doc_titles: [] };
|
||
const titles = c.doc_titles.slice(0, remaining);
|
||
remaining -= titles.length;
|
||
return { client_name: c.client_name, doc_titles: titles };
|
||
});
|
||
itglueIncluded = itglueDocTitles.some((c) => c.doc_titles.length > 0);
|
||
const itglueEnd = new Date();
|
||
stageRecords.push({
|
||
stage: 'itglue',
|
||
stage_order: 2,
|
||
model_id: null,
|
||
input_payload: { client_count: uniqueClients.length },
|
||
output_payload: { doc_count: itglueDocTitles.reduce((a, c) => a + c.doc_titles.length, 0) },
|
||
input_tokens: null,
|
||
output_tokens: null,
|
||
latency_ms: itglueEnd.getTime() - itglueStart.getTime(),
|
||
started_at: itglueStart,
|
||
completed_at: itglueEnd,
|
||
error_message: null,
|
||
});
|
||
}
|
||
|
||
// ── Step 3: reduce LLM call ──
|
||
const reduceStart = new Date();
|
||
let reduceResult;
|
||
try {
|
||
reduceResult = await runAggregateReduceStage({
|
||
distributions: {
|
||
category_distribution: categories,
|
||
client_distribution: clients,
|
||
resolution_path_distribution: resolutionPaths,
|
||
root_cause_distribution: rootCauses,
|
||
date_range_actual: dateRange,
|
||
},
|
||
fingerprints: fingerprints.map((f) => ({
|
||
ticket_number: f.ticket_number,
|
||
fingerprint: f.fingerprint,
|
||
})),
|
||
itglue_doc_titles: itglueDocTitles,
|
||
});
|
||
} catch (err) {
|
||
const reduceEnd = new Date();
|
||
stageRecords.push({
|
||
stage: 'analyze',
|
||
stage_order: 3,
|
||
model_id: null,
|
||
input_payload: { fingerprint_count: fingerprints.length },
|
||
output_payload: {},
|
||
input_tokens: null,
|
||
output_tokens: null,
|
||
latency_ms: reduceEnd.getTime() - reduceStart.getTime(),
|
||
started_at: reduceStart,
|
||
completed_at: reduceEnd,
|
||
error_message: err instanceof Error ? err.message : String(err),
|
||
});
|
||
throw err;
|
||
}
|
||
const reduceEnd = new Date();
|
||
stageRecords.push({
|
||
stage: 'analyze',
|
||
stage_order: 3,
|
||
model_id: reduceResult.model_used,
|
||
input_payload: { fingerprint_count: fingerprints.length },
|
||
output_payload: reduceResult.data,
|
||
input_tokens: reduceResult.usage.input_tokens,
|
||
output_tokens: reduceResult.usage.output_tokens,
|
||
latency_ms: reduceEnd.getTime() - reduceStart.getTime(),
|
||
started_at: reduceStart,
|
||
completed_at: reduceEnd,
|
||
error_message: null,
|
||
});
|
||
|
||
// ── Step 4: persist outputs ──
|
||
await postgresClient.query(
|
||
`UPDATE analyzer_aggregate_reports
|
||
SET documentation_gaps = $2::jsonb,
|
||
process_gaps = $3::jsonb,
|
||
client_patterns = $4::jsonb,
|
||
recurrence_clusters = $5::jsonb,
|
||
systemic_observations = $6::jsonb,
|
||
recommended_actions = $7::jsonb,
|
||
narrative_summary = $8,
|
||
executive_summary = $9,
|
||
total_input_tokens = $10,
|
||
total_output_tokens = $11,
|
||
estimated_cost_usd = $12,
|
||
model_used = $13,
|
||
itglue_context_included = $14,
|
||
status = 'complete'
|
||
WHERE id = $1`,
|
||
[
|
||
reportId,
|
||
JSON.stringify(reduceResult.data.documentation_gaps),
|
||
JSON.stringify(reduceResult.data.process_gaps),
|
||
JSON.stringify(reduceResult.data.client_patterns),
|
||
JSON.stringify(reduceResult.data.recurrence_clusters),
|
||
JSON.stringify(reduceResult.data.systemic_observations),
|
||
JSON.stringify(reduceResult.data.recommended_actions),
|
||
reduceResult.data.narrative_summary,
|
||
reduceResult.data.executive_summary,
|
||
reduceResult.usage.input_tokens,
|
||
reduceResult.usage.output_tokens,
|
||
reduceResult.estimated_cost_usd,
|
||
reduceResult.model_used,
|
||
itglueIncluded,
|
||
]
|
||
);
|
||
|
||
await bulkInsertReportStageExecutions(reportId, stageRecords);
|
||
} catch (err) {
|
||
const message = err instanceof Error ? err.message : String(err);
|
||
console.error(`[ANALYZER-REPORT] runAggregateReport ${reportId} failed:`, message);
|
||
await postgresClient
|
||
.query(
|
||
`UPDATE analyzer_aggregate_reports
|
||
SET status = 'failed', error_message = $2
|
||
WHERE id = $1`,
|
||
[reportId, message]
|
||
)
|
||
.catch(() => {});
|
||
if (stageRecords.length > 0) {
|
||
await bulkInsertReportStageExecutions(reportId, stageRecords).catch(() => {});
|
||
}
|
||
}
|
||
}
|