feat(analyzer): Phase 2 — full stage persistence, fingerprints, aggregate reports, cost guards
Eight sub-phases per docs/ticket-analyzer-phase2-spec.md:
2.1 Schema (migration 070): analyzer_stage_executions table; source_snapshot,
aggregate_fingerprint, fingerprint_generated_at columns on analyzer_analyses.
model_traces marked LEGACY (kept for back-compat).
2.2 Every pipeline stage records a row to analyzer_stage_executions, success
or failure. Worker persists a status='failed' analyzer_analyses row when
the pipeline throws so partial stage records have a parent. Pipeline
exposes raw triage/sonnet/opus responses for downstream stages.
2.3 Stage 3 prompt updated with markdown formatting rules + banned filler
phrases. Added react-markdown + remark-gfm + @tailwindcss/typography.
New <AnalysisMarkdown> component replaces <ProseText>; coerces stray
headers to bold paragraphs.
2.4 Stage 6 fingerprint (Haiku) runs after persistence, failure-tolerant.
scripts/backfill-fingerprints.ts reconstructs Stage 6 input from the
legacy model_traces blob.
2.5 Browse UI rebuild at /analyzer/tickets: multi-select for client/issue/
queue/status/priority/assignee, sticky filter bar, active-filter chips,
bulk selection persisted via localStorage, "Analyze N selected" +
"Generate aggregate report" actions. New <MultiSelect> primitive.
Staleness uses last_activity_date > completed_at heuristic per spec C.1.
2.6 Aggregate reports (migration 071): runner is fire-and-forget, persists
SQL distributions immediately so UI shows partial state during the
Sonnet reduce call. Three endpoints, three pages (/analyzer/reports[/new
/:id]). IT Glue context fetcher capped at 200 doc titles.
2.7 Cost guards (migration 072): per-request $5 confirmation, soft-warn at
$20/day, hard-block at $50/day with ANALYZER_DAILY_COST_OVERRIDE_USERS
override. Every gating decision audited.
2.8 Runbook + build notes updated.
128 vitest tests passing, tsc clean. Migrations 070/071/072 idempotent
(IF NOT EXISTS). model_traces double-write retained — drop in a future
migration once aggregate reports have soaked.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
b20c94ea1a
commit
bd3401df1c
33 changed files with 7132 additions and 554 deletions
506
lib/services/analyzer/aggregate-persistence.ts
Normal file
506
lib/services/analyzer/aggregate-persistence.ts
Normal file
|
|
@ -0,0 +1,506 @@
|
|||
/**
|
||||
* 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.getFlexibleAssets({ organizationId: 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(() => {});
|
||||
}
|
||||
}
|
||||
}
|
||||
155
lib/services/analyzer/cost-guard.ts
Normal file
155
lib/services/analyzer/cost-guard.ts
Normal file
|
|
@ -0,0 +1,155 @@
|
|||
/**
|
||||
* Cost guards for analyzer LLM operations.
|
||||
*
|
||||
* Spec: docs/ticket-analyzer-phase2-spec.md → Section D.8.
|
||||
*
|
||||
* Three thresholds:
|
||||
* • $5 per request → require explicit confirmation in UI
|
||||
* • $20 per user/day → soft warn (returned in preflight; UI can surface)
|
||||
* • $50 per user/day → hard block (unless user is in override list)
|
||||
*
|
||||
* Override env var: ANALYZER_DAILY_COST_OVERRIDE_USERS (comma-separated user ids)
|
||||
*/
|
||||
|
||||
import postgresClient from '@/lib/services/postgres-client';
|
||||
|
||||
export const REQUIRES_CONFIRMATION_USD = 5.0;
|
||||
export const SOFT_WARN_DAILY_USD = 20.0;
|
||||
export const HARD_BLOCK_DAILY_USD = 50.0;
|
||||
|
||||
export type CostDecision =
|
||||
| 'approved'
|
||||
| 'requires_confirmation'
|
||||
| 'blocked'
|
||||
| 'overridden';
|
||||
|
||||
export interface CostEvaluation {
|
||||
estimatedCost: number;
|
||||
dailySpendBefore: number;
|
||||
decision: CostDecision;
|
||||
decisionReason: string | null;
|
||||
softWarn: boolean;
|
||||
hardBlocked: boolean;
|
||||
requiresConfirmation: boolean;
|
||||
isOverride: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pessimistic cost estimate for an aggregate report. Assumes Sonnet pricing
|
||||
* and a payload size proportional to the number of fingerprints (each ~2KB
|
||||
* after compaction). Conservative — actual cost is usually lower.
|
||||
*/
|
||||
export function estimateAggregateReportCost(input: {
|
||||
ticketCount: number;
|
||||
includeItglueContext: boolean;
|
||||
}): number {
|
||||
// Inputs:
|
||||
// per-fingerprint input tokens ≈ 700 (compacted JSON) → 700 * N
|
||||
// distributions + system prompt ≈ 2000 tokens
|
||||
// IT Glue context (if included): up to 200 doc titles * 50 tokens = 10K tokens
|
||||
// Outputs: cap ≈ 6K tokens (we set max_tokens 16K but real outputs are smaller).
|
||||
const inputTokens =
|
||||
700 * input.ticketCount +
|
||||
2000 +
|
||||
(input.includeItglueContext ? 10_000 : 0);
|
||||
const outputTokens = 6_000;
|
||||
// Sonnet pricing per Phase 1 pricing table: $3/$15 per 1M tokens.
|
||||
const cost = (inputTokens / 1_000_000) * 3 + (outputTokens / 1_000_000) * 15;
|
||||
return Math.round(cost * 10_000) / 10_000;
|
||||
}
|
||||
|
||||
function getOverrideUsers(): Set<string> {
|
||||
return new Set(
|
||||
(process.env.ANALYZER_DAILY_COST_OVERRIDE_USERS ?? '')
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean)
|
||||
);
|
||||
}
|
||||
|
||||
export async function getUserDailySpend(userId: string | null): Promise<number> {
|
||||
if (!userId) return 0;
|
||||
// Sum from analyzer_aggregate_reports + analyzer_analyses for the trailing 24h.
|
||||
const res = await postgresClient.query<{ total: string }>(
|
||||
`SELECT COALESCE(SUM(amt), 0)::text AS total FROM (
|
||||
SELECT estimated_cost_usd AS amt
|
||||
FROM analyzer_aggregate_reports
|
||||
WHERE generated_by_user_id = $1
|
||||
AND generated_at >= NOW() - INTERVAL '24 hours'
|
||||
AND estimated_cost_usd IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT estimated_cost_usd AS amt
|
||||
FROM analyzer_analyses
|
||||
WHERE triggered_by_user_id = $1
|
||||
AND triggered_at >= NOW() - INTERVAL '24 hours'
|
||||
AND status = 'complete'
|
||||
) x`,
|
||||
[userId]
|
||||
);
|
||||
return Number(res.rows[0]?.total ?? 0);
|
||||
}
|
||||
|
||||
export async function evaluateCost(input: {
|
||||
userId: string | null;
|
||||
estimatedCost: number;
|
||||
confirmedCost: boolean;
|
||||
}): Promise<CostEvaluation> {
|
||||
const dailySpendBefore = await getUserDailySpend(input.userId);
|
||||
const projectedDailySpend = dailySpendBefore + input.estimatedCost;
|
||||
const overrideUsers = getOverrideUsers();
|
||||
const isOverride = input.userId !== null && overrideUsers.has(input.userId);
|
||||
|
||||
const softWarn = projectedDailySpend >= SOFT_WARN_DAILY_USD;
|
||||
const hardBlocked = projectedDailySpend >= HARD_BLOCK_DAILY_USD;
|
||||
const requiresConfirmation =
|
||||
input.estimatedCost > REQUIRES_CONFIRMATION_USD && !input.confirmedCost;
|
||||
|
||||
let decision: CostDecision;
|
||||
let decisionReason: string | null = null;
|
||||
if (hardBlocked && !isOverride) {
|
||||
decision = 'blocked';
|
||||
decisionReason = `Projected daily spend $${projectedDailySpend.toFixed(2)} would exceed hard limit $${HARD_BLOCK_DAILY_USD.toFixed(2)}`;
|
||||
} else if (hardBlocked && isOverride) {
|
||||
decision = 'overridden';
|
||||
decisionReason = `Override allowed (user in ANALYZER_DAILY_COST_OVERRIDE_USERS); projected $${projectedDailySpend.toFixed(2)}`;
|
||||
} else if (requiresConfirmation) {
|
||||
decision = 'requires_confirmation';
|
||||
decisionReason = `Per-request cost $${input.estimatedCost.toFixed(2)} > confirmation threshold $${REQUIRES_CONFIRMATION_USD.toFixed(2)}`;
|
||||
} else {
|
||||
decision = 'approved';
|
||||
}
|
||||
|
||||
return {
|
||||
estimatedCost: input.estimatedCost,
|
||||
dailySpendBefore,
|
||||
decision,
|
||||
decisionReason,
|
||||
softWarn,
|
||||
hardBlocked,
|
||||
requiresConfirmation,
|
||||
isOverride,
|
||||
};
|
||||
}
|
||||
|
||||
export async function recordCostAuditDecision(input: {
|
||||
userId: string | null;
|
||||
action: string;
|
||||
evaluation: CostEvaluation;
|
||||
context?: Record<string, unknown>;
|
||||
}): Promise<void> {
|
||||
await postgresClient.query(
|
||||
`INSERT INTO analyzer_cost_audit
|
||||
(user_id, action, estimated_cost, daily_spend_before,
|
||||
decision, decision_reason, context)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb)`,
|
||||
[
|
||||
input.userId,
|
||||
input.action,
|
||||
input.evaluation.estimatedCost,
|
||||
input.evaluation.dailySpendBefore,
|
||||
input.evaluation.decision,
|
||||
input.evaluation.decisionReason,
|
||||
JSON.stringify(input.context ?? {}),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
|
@ -9,10 +9,12 @@
|
|||
|
||||
import postgresClient from '@/lib/services/postgres-client';
|
||||
import {
|
||||
type AggregateFingerprint,
|
||||
type AnalyzerJob,
|
||||
type DeepAnalysisResponse,
|
||||
type JobStatus,
|
||||
type PersistedAnalysis,
|
||||
type StageExecutionRecord,
|
||||
type TaggedEvent,
|
||||
} from '@/lib/types/analyzer';
|
||||
import type { ITGlueDocReference } from '@/lib/types/analyzer';
|
||||
|
|
@ -36,8 +38,13 @@ export interface InsertAnalysisInput {
|
|||
/** Final analysis content (after any Opus updates). null on failure. */
|
||||
analysis: DeepAnalysisResponse | null;
|
||||
filtered_noise_count: number;
|
||||
/** Per-stage trace dump for debugging — raw model responses, attempts, etc. */
|
||||
/**
|
||||
* LEGACY (phase 1). Per-stage trace dump. Retained for back-compat until
|
||||
* analyzer_stage_executions has full coverage and we drop the column.
|
||||
*/
|
||||
model_traces: Record<string, unknown>;
|
||||
/** Phase 2: Stage 0 preprocessed event list at analysis time. */
|
||||
source_snapshot?: TaggedEvent[] | null;
|
||||
error_message?: string | null;
|
||||
}
|
||||
|
||||
|
|
@ -108,7 +115,8 @@ export async function insertAnalysis(input: InsertAnalysisInput): Promise<{
|
|||
summary, timeline, what_was_done, what_should_have_been_done,
|
||||
gaps, next_step, next_step_rationale, post_resolution_analysis,
|
||||
confidence_score, needs_human_review, human_review_reasons,
|
||||
itglue_docs_referenced, model_traces, filtered_noise_count, error_message
|
||||
itglue_docs_referenced, model_traces, filtered_noise_count, error_message,
|
||||
source_snapshot
|
||||
)
|
||||
VALUES (
|
||||
$1, $2, $3,
|
||||
|
|
@ -119,7 +127,8 @@ export async function insertAnalysis(input: InsertAnalysisInput): Promise<{
|
|||
$14, $15::jsonb, $16::jsonb, $17::jsonb,
|
||||
$18::jsonb, $19, $20, $21,
|
||||
$22, $23, $24::jsonb,
|
||||
$25::jsonb, $26::jsonb, $27, $28
|
||||
$25::jsonb, $26::jsonb, $27, $28,
|
||||
$29::jsonb
|
||||
)
|
||||
RETURNING id::text AS id
|
||||
`,
|
||||
|
|
@ -154,12 +163,113 @@ export async function insertAnalysis(input: InsertAnalysisInput): Promise<{
|
|||
JSON.stringify(input.model_traces),
|
||||
input.filtered_noise_count,
|
||||
input.error_message ?? null,
|
||||
input.source_snapshot ? JSON.stringify(input.source_snapshot) : null,
|
||||
]
|
||||
);
|
||||
|
||||
return { id: res.rows[0].id, analysis_version: version };
|
||||
}
|
||||
|
||||
/**
|
||||
* Bulk-insert one analyzer_stage_executions row per record. No-op when records
|
||||
* is empty. Single multi-VALUES INSERT — fast enough for the few rows produced
|
||||
* per pipeline run that we don't need COPY.
|
||||
*/
|
||||
export async function bulkInsertStageExecutions(
|
||||
analysisId: string,
|
||||
records: StageExecutionRecord[]
|
||||
): Promise<void> {
|
||||
if (records.length === 0) return;
|
||||
const values: unknown[] = [analysisId];
|
||||
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 (
|
||||
analysis_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
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Phase 2: write the Stage 6 fingerprint to an existing analysis row.
|
||||
*/
|
||||
export async function updateAnalysisFingerprint(
|
||||
analysisId: string,
|
||||
fingerprint: AggregateFingerprint
|
||||
): Promise<void> {
|
||||
await postgresClient.query(
|
||||
`UPDATE analyzer_analyses
|
||||
SET aggregate_fingerprint = $2::jsonb,
|
||||
fingerprint_generated_at = NOW()
|
||||
WHERE id = $1`,
|
||||
[analysisId, JSON.stringify(fingerprint)]
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Phase 2: persist a 'failed' analyzer_analyses row when the pipeline throws.
|
||||
* Carries content_hash + source_snapshot so partial-run forensics work, plus
|
||||
* any stage records the pipeline managed to record before throwing.
|
||||
*/
|
||||
export async function insertFailedAnalysis(input: {
|
||||
ticket_number: string;
|
||||
autotask_ticket_id: number;
|
||||
content_hash: string;
|
||||
triggered_by_user_id: string | null;
|
||||
source_snapshot: TaggedEvent[];
|
||||
filtered_noise_count: number;
|
||||
error_message: string;
|
||||
partial_input_tokens: number;
|
||||
partial_output_tokens: number;
|
||||
partial_cost_usd: number;
|
||||
haiku_used: boolean;
|
||||
sonnet_used: boolean;
|
||||
opus_used: boolean;
|
||||
}): Promise<{ id: string; analysis_version: number }> {
|
||||
return await insertAnalysis({
|
||||
ticket_number: input.ticket_number,
|
||||
autotask_ticket_id: input.autotask_ticket_id,
|
||||
content_hash: input.content_hash,
|
||||
triggered_by_user_id: input.triggered_by_user_id,
|
||||
status: 'failed',
|
||||
haiku_used: input.haiku_used,
|
||||
sonnet_used: input.sonnet_used,
|
||||
opus_used: input.opus_used,
|
||||
total_input_tokens: input.partial_input_tokens,
|
||||
total_output_tokens: input.partial_output_tokens,
|
||||
estimated_cost_usd: input.partial_cost_usd,
|
||||
analysis: null,
|
||||
filtered_noise_count: input.filtered_noise_count,
|
||||
model_traces: {},
|
||||
source_snapshot: input.source_snapshot,
|
||||
error_message: input.error_message,
|
||||
});
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Job table operations
|
||||
// =============================================================================
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ import {
|
|||
type DeepAnalysisResponse,
|
||||
type OpusResponse,
|
||||
type PreprocessedTicket,
|
||||
type StageExecutionRecord,
|
||||
type StageName,
|
||||
type TriageResponse,
|
||||
} from '@/lib/types/analyzer';
|
||||
import { preprocessTicket, type RawTicketBundle } from './preprocessor';
|
||||
|
|
@ -96,7 +98,11 @@ export interface PipelineSuccess {
|
|||
filtered_noise_count: number;
|
||||
itglue_search_used: boolean;
|
||||
itglue_org_resolved: boolean;
|
||||
/** Per-stage debug payload — goes into analyzer_analyses.model_traces. */
|
||||
/** Phase 2: the raw triage / sonnet / opus responses fed to Stage 6 fingerprint. */
|
||||
triage_response: TriageResponse;
|
||||
sonnet_response: DeepAnalysisResponse;
|
||||
opus_response: OpusResponse | null;
|
||||
/** LEGACY: per-stage debug payload — goes into analyzer_analyses.model_traces. */
|
||||
model_traces: {
|
||||
triage?: StageTrace;
|
||||
deep_analysis?: StageTrace;
|
||||
|
|
@ -139,6 +145,61 @@ export interface PipelineProgressCallbacks {
|
|||
| 'analyzing'
|
||||
| 'deep_review'
|
||||
) => Promise<void> | void;
|
||||
/**
|
||||
* Phase 2 — emitted once per stage attempt, on success OR failure. Worker
|
||||
* collects these into an array; on pipeline failure the array still has
|
||||
* everything that ran. Pipeline pushes the record before re-throwing.
|
||||
*/
|
||||
onStageRecord?: (record: StageExecutionRecord) => void;
|
||||
/**
|
||||
* Phase 2 — emitted right after Stage 0 succeeds. Lets the worker capture
|
||||
* the preprocessed ticket so it can persist a failed analyzer_analyses row
|
||||
* (with source_snapshot + content_hash) when a later stage throws.
|
||||
*/
|
||||
onPreprocessed?: (pre: PreprocessedTicket) => void;
|
||||
}
|
||||
|
||||
/** Internal helper: run a stage, time it, push a record, propagate errors. */
|
||||
async function recordedStage<T extends { usage?: { input_tokens: number; output_tokens: number } }>(
|
||||
meta: {
|
||||
stage: StageName;
|
||||
stage_order: number;
|
||||
model_id: string | null;
|
||||
input_payload: unknown;
|
||||
},
|
||||
fn: () => Promise<T>,
|
||||
callbacks: PipelineProgressCallbacks,
|
||||
outputSelector: (result: T) => unknown
|
||||
): Promise<T> {
|
||||
const startedAt = new Date();
|
||||
try {
|
||||
const result = await fn();
|
||||
const completedAt = new Date();
|
||||
callbacks.onStageRecord?.({
|
||||
...meta,
|
||||
output_payload: outputSelector(result),
|
||||
input_tokens: result.usage?.input_tokens ?? null,
|
||||
output_tokens: result.usage?.output_tokens ?? null,
|
||||
latency_ms: completedAt.getTime() - startedAt.getTime(),
|
||||
started_at: startedAt,
|
||||
completed_at: completedAt,
|
||||
error_message: null,
|
||||
});
|
||||
return result;
|
||||
} catch (err) {
|
||||
const completedAt = new Date();
|
||||
callbacks.onStageRecord?.({
|
||||
...meta,
|
||||
output_payload: {},
|
||||
input_tokens: null,
|
||||
output_tokens: null,
|
||||
latency_ms: completedAt.getTime() - startedAt.getTime(),
|
||||
started_at: startedAt,
|
||||
completed_at: completedAt,
|
||||
error_message: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
export async function runPipeline(
|
||||
|
|
@ -151,7 +212,31 @@ export async function runPipeline(
|
|||
|
||||
// ── Stage 0: preprocess ──────────────────────────────────────────────────
|
||||
await callbacks.onStage?.('fetching');
|
||||
const preStart = new Date();
|
||||
const pre = preprocessTicket(input.bundle);
|
||||
const preEnd = new Date();
|
||||
callbacks.onPreprocessed?.(pre);
|
||||
callbacks.onStageRecord?.({
|
||||
stage: 'preprocess',
|
||||
stage_order: 1,
|
||||
model_id: null,
|
||||
input_payload: {
|
||||
ticket_number: input.bundle.ticket.ticket_number,
|
||||
notes_count: input.bundle.notes?.length ?? 0,
|
||||
time_entries_count: input.bundle.time_entries?.length ?? 0,
|
||||
},
|
||||
output_payload: {
|
||||
events_count: pre.events.length,
|
||||
counts: pre.counts,
|
||||
content_hash: pre.content_hash,
|
||||
},
|
||||
input_tokens: null,
|
||||
output_tokens: null,
|
||||
latency_ms: preEnd.getTime() - preStart.getTime(),
|
||||
started_at: preStart,
|
||||
completed_at: preEnd,
|
||||
error_message: null,
|
||||
});
|
||||
|
||||
// ── Idempotency: short-circuit if force=false and we have a complete row ─
|
||||
if (!input.force) {
|
||||
|
|
@ -181,7 +266,21 @@ export async function runPipeline(
|
|||
|
||||
// ── Stage 1: Haiku triage ────────────────────────────────────────────────
|
||||
await callbacks.onStage?.('triaging');
|
||||
const triageResult = await runTriageStage(pre, anthropic);
|
||||
const triageResult = await recordedStage(
|
||||
{
|
||||
stage: 'triage',
|
||||
stage_order: 2,
|
||||
model_id: 'claude-haiku-4-5',
|
||||
input_payload: {
|
||||
ticket_number: pre.header.ticket_number,
|
||||
events_count: pre.events.length,
|
||||
filtered_noise_count: pre.counts.filtered_noise,
|
||||
},
|
||||
},
|
||||
() => runTriageStage(pre, anthropic),
|
||||
callbacks,
|
||||
(r) => r.data
|
||||
);
|
||||
usage = addUsage(usage, triageResult.usage);
|
||||
estimatedCostUsd += triageResult.estimated_cost_usd;
|
||||
traces.triage = {
|
||||
|
|
@ -204,6 +303,8 @@ export async function runPipeline(
|
|||
pre.header.account_name
|
||||
) {
|
||||
await callbacks.onStage?.('itglue');
|
||||
const itglueStart = new Date();
|
||||
let itglueErr: Error | null = null;
|
||||
try {
|
||||
itglueResult = await itglueSearchFn({
|
||||
org_name: pre.header.account_name,
|
||||
|
|
@ -212,10 +313,35 @@ export async function runPipeline(
|
|||
itglueDocs = itglueResult.docs;
|
||||
} catch (err) {
|
||||
// Tolerate IT Glue failures — analysis continues without context.
|
||||
itglueErr = err instanceof Error ? err : new Error(String(err));
|
||||
console.warn(
|
||||
`[pipeline] IT Glue search failed for ${pre.header.ticket_number}: ${err instanceof Error ? err.message : String(err)}`
|
||||
`[pipeline] IT Glue search failed for ${pre.header.ticket_number}: ${itglueErr.message}`
|
||||
);
|
||||
}
|
||||
const itglueEnd = new Date();
|
||||
callbacks.onStageRecord?.({
|
||||
stage: 'itglue',
|
||||
stage_order: 3,
|
||||
model_id: null,
|
||||
input_payload: {
|
||||
org_name: pre.header.account_name,
|
||||
hints: triageResult.data.itglue_search_hints,
|
||||
},
|
||||
// Redacted-only payload (itglue-search applies redact() internally).
|
||||
output_payload: itglueErr
|
||||
? {}
|
||||
: {
|
||||
org_id: itglueResult?.org_id ?? null,
|
||||
alias_used: itglueResult?.alias_used ?? false,
|
||||
docs: itglueDocs,
|
||||
},
|
||||
input_tokens: null,
|
||||
output_tokens: null,
|
||||
latency_ms: itglueEnd.getTime() - itglueStart.getTime(),
|
||||
started_at: itglueStart,
|
||||
completed_at: itglueEnd,
|
||||
error_message: itglueErr ? itglueErr.message : null,
|
||||
});
|
||||
traces.itglue = {
|
||||
org_id: itglueResult?.org_id ?? null,
|
||||
alias_used: itglueResult?.alias_used ?? false,
|
||||
|
|
@ -225,13 +351,25 @@ export async function runPipeline(
|
|||
|
||||
// ── Stage 3: Sonnet deep analysis ────────────────────────────────────────
|
||||
await callbacks.onStage?.('analyzing');
|
||||
const sonnetResult = await runDeepAnalysisStage(
|
||||
const sonnetResult = await recordedStage(
|
||||
{
|
||||
pre,
|
||||
triage: triageResult.data,
|
||||
itglue_docs: itglueDocs,
|
||||
stage: 'analyze',
|
||||
stage_order: 4,
|
||||
model_id: 'claude-sonnet-4-6',
|
||||
input_payload: {
|
||||
ticket_number: pre.header.ticket_number,
|
||||
events_count: pre.events.length,
|
||||
triage: triageResult.data,
|
||||
itglue_doc_count: itglueDocs.length,
|
||||
},
|
||||
},
|
||||
anthropic
|
||||
() =>
|
||||
runDeepAnalysisStage(
|
||||
{ pre, triage: triageResult.data, itglue_docs: itglueDocs },
|
||||
anthropic
|
||||
),
|
||||
callbacks,
|
||||
(r) => r.data
|
||||
);
|
||||
usage = addUsage(usage, sonnetResult.usage);
|
||||
estimatedCostUsd += sonnetResult.estimated_cost_usd;
|
||||
|
|
@ -248,6 +386,7 @@ export async function runPipeline(
|
|||
traces.sonnet_response = sonnetResult.data;
|
||||
|
||||
let analysis: DeepAnalysisResponse = sonnetResult.data;
|
||||
let opusResponseForResult: OpusResponse | null = null;
|
||||
let opusUsed = false;
|
||||
let costCircuitBreakerTripped = false;
|
||||
|
||||
|
|
@ -271,9 +410,27 @@ export async function runPipeline(
|
|||
};
|
||||
} else {
|
||||
await callbacks.onStage?.('deep_review');
|
||||
const opusResult = await runDeepReasoningStage(
|
||||
{ pre, triage: triageResult.data, sonnet: sonnetResult.data },
|
||||
anthropic
|
||||
const opusResult = await recordedStage(
|
||||
{
|
||||
stage: 'deep_review',
|
||||
stage_order: 5,
|
||||
model_id: 'claude-opus-4-7',
|
||||
input_payload: {
|
||||
ticket_number: pre.header.ticket_number,
|
||||
events_count: pre.events.length,
|
||||
triage: triageResult.data,
|
||||
sonnet_summary: sonnetResult.data.summary,
|
||||
},
|
||||
},
|
||||
() =>
|
||||
runDeepReasoningStage(
|
||||
{ pre, triage: triageResult.data, sonnet: sonnetResult.data },
|
||||
anthropic
|
||||
),
|
||||
callbacks,
|
||||
// Per spec: store the FULL Opus response including opus_notes, not
|
||||
// just the merged updates that previously overwrote everything.
|
||||
(r) => r.data
|
||||
);
|
||||
usage = addUsage(usage, opusResult.usage);
|
||||
estimatedCostUsd += opusResult.estimated_cost_usd;
|
||||
|
|
@ -289,6 +446,7 @@ export async function runPipeline(
|
|||
events_dropped: opusResult.events_dropped,
|
||||
};
|
||||
traces.opus_response = opusResult.data;
|
||||
opusResponseForResult = opusResult.data;
|
||||
analysis = applyOpusUpdates(analysis, opusResult.data.updates);
|
||||
}
|
||||
}
|
||||
|
|
@ -300,6 +458,9 @@ export async function runPipeline(
|
|||
filtered_noise_count: pre.counts.filtered_noise,
|
||||
itglue_search_used: itglueResult !== null,
|
||||
itglue_org_resolved: !!itglueResult?.org_id,
|
||||
triage_response: triageResult.data,
|
||||
sonnet_response: sonnetResult.data,
|
||||
opus_response: opusResponseForResult,
|
||||
meta: {
|
||||
haiku_used: true,
|
||||
sonnet_used: true,
|
||||
|
|
|
|||
170
lib/services/analyzer/stages/aggregate-reduce.ts
Normal file
170
lib/services/analyzer/stages/aggregate-reduce.ts
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
/**
|
||||
* Aggregate report reduce stage.
|
||||
*
|
||||
* Takes structured fingerprints + SQL distributions + (optional) IT Glue doc
|
||||
* titles, runs a single LLM call (Sonnet by default, Opus opt-in), and
|
||||
* produces the report's LLM-derived fields.
|
||||
*
|
||||
* Spec: docs/ticket-analyzer-phase2-spec.md → Section D.6
|
||||
*/
|
||||
|
||||
import type Anthropic from '@anthropic-ai/sdk';
|
||||
import {
|
||||
AggregateReduceResponse,
|
||||
type AggregateFingerprint,
|
||||
} from '@/lib/types/analyzer';
|
||||
import { callLLMStage, type LLMCallResult } from '@/lib/services/llm/call';
|
||||
import { OPUS, SONNET, type ModelId } from '@/lib/services/llm/models';
|
||||
|
||||
const REDUCE_MAX_TOKENS = 16_000;
|
||||
|
||||
const SYSTEM_PROMPT = `You are a senior MSP analyst identifying patterns across multiple ticket analyses to surface systemic issues.
|
||||
|
||||
You will receive:
|
||||
1. SQL-derived distributions (categories, clients, resolution paths, root causes)
|
||||
2. An array of structured fingerprints, one per analyzed ticket
|
||||
3. Optionally, a list of IT Glue documentation titles for the affected clients
|
||||
|
||||
Your job is to identify patterns the SQL aggregations cannot see — patterns that emerge from the gap descriptions, vendor involvement, recurrence signals, and cross-ticket clustering.
|
||||
|
||||
Be specific. Cite ticket numbers as evidence for every claim. Distinguish between "documentation gap exists" (no doc on this topic per the IT Glue title list) and "documentation not referenced" (doc may exist but wasn't used in resolution).
|
||||
|
||||
Respond ONLY with JSON matching this schema:
|
||||
|
||||
{
|
||||
"documentation_gaps": [
|
||||
{
|
||||
"gap": string,
|
||||
"frequency": number,
|
||||
"example_ticket_numbers": string[],
|
||||
"evidence": string,
|
||||
"itglue_check": "no_doc_exists" | "doc_exists_but_unused" | "unable_to_verify"
|
||||
}
|
||||
],
|
||||
"process_gaps": [
|
||||
{
|
||||
"gap": string,
|
||||
"frequency": number,
|
||||
"severity": "low" | "medium" | "high",
|
||||
"example_ticket_numbers": string[],
|
||||
"evidence": string
|
||||
}
|
||||
],
|
||||
"client_patterns": [
|
||||
{
|
||||
"client": string,
|
||||
"pattern": string,
|
||||
"frequency": number,
|
||||
"example_ticket_numbers": string[]
|
||||
}
|
||||
],
|
||||
"recurrence_clusters": [
|
||||
{
|
||||
"theme": string,
|
||||
"ticket_numbers": string[],
|
||||
"summary": string
|
||||
}
|
||||
],
|
||||
"systemic_observations": [
|
||||
{
|
||||
"observation": string,
|
||||
"evidence": string,
|
||||
"severity": "low" | "medium" | "high"
|
||||
}
|
||||
],
|
||||
"recommended_actions": [
|
||||
{
|
||||
"action": string,
|
||||
"rationale": string,
|
||||
"priority": "low" | "medium" | "high",
|
||||
"type": "documentation" | "process" | "training" | "tooling"
|
||||
}
|
||||
],
|
||||
"executive_summary": string,
|
||||
"narrative_summary": string
|
||||
}
|
||||
|
||||
Quality bars:
|
||||
- Do not list a documentation_gap unless it appears in 2+ tickets.
|
||||
- Do not list a process_gap unless it appears in 2+ tickets OR has severity=high in at least one.
|
||||
- Recurrence clusters require at least 2 tickets.
|
||||
- Every recommended_action must be specific enough that a person could pick it up tomorrow. "Improve documentation" is rejected; "Create a runbook for AMS360 App Access Key location and integration permission verification" is acceptable.
|
||||
- The narrative_summary follows the same prose formatting rules as individual analyses: markdown, no headers, no filler phrases.
|
||||
- The executive_summary is 3-5 sentences, plain prose.`;
|
||||
|
||||
export interface AggregateReduceInput {
|
||||
distributions: {
|
||||
category_distribution: Record<string, number>;
|
||||
client_distribution: Record<string, number>;
|
||||
resolution_path_distribution: Record<string, number>;
|
||||
root_cause_distribution: Record<string, number>;
|
||||
date_range_actual: { earliest: string | null; latest: string | null };
|
||||
};
|
||||
fingerprints: Array<{
|
||||
ticket_number: string;
|
||||
fingerprint: AggregateFingerprint;
|
||||
}>;
|
||||
itglue_doc_titles?: Array<{
|
||||
client_name: string;
|
||||
doc_titles: string[];
|
||||
}>;
|
||||
}
|
||||
|
||||
export function selectReduceModel(
|
||||
fingerprintCount: number,
|
||||
forceOpus = false
|
||||
): ModelId {
|
||||
if (forceOpus) return OPUS;
|
||||
// Per spec D.6: Sonnet up to 100; Opus is opt-in. Above 25, send only the
|
||||
// structured fingerprints (no narrative excerpts) — handled at payload-build time.
|
||||
return SONNET;
|
||||
}
|
||||
|
||||
function buildUserPayload(input: AggregateReduceInput): string {
|
||||
const parts = [
|
||||
`=== SQL DISTRIBUTIONS ===`,
|
||||
JSON.stringify(input.distributions, null, 2),
|
||||
``,
|
||||
`=== FINGERPRINTS (${input.fingerprints.length} tickets) ===`,
|
||||
JSON.stringify(
|
||||
input.fingerprints.map((f) => ({
|
||||
ticket_number: f.ticket_number,
|
||||
...f.fingerprint,
|
||||
})),
|
||||
null,
|
||||
2
|
||||
),
|
||||
];
|
||||
if (input.itglue_doc_titles && input.itglue_doc_titles.length > 0) {
|
||||
parts.push(
|
||||
``,
|
||||
`=== IT GLUE DOC TITLES BY CLIENT ===`,
|
||||
JSON.stringify(input.itglue_doc_titles, null, 2)
|
||||
);
|
||||
} else {
|
||||
parts.push(
|
||||
``,
|
||||
`=== IT GLUE DOC TITLES BY CLIENT ===`,
|
||||
`Not included for this report. Mark itglue_check as "unable_to_verify" for documentation_gaps.`
|
||||
);
|
||||
}
|
||||
return parts.join('\n');
|
||||
}
|
||||
|
||||
export async function runAggregateReduceStage(
|
||||
input: AggregateReduceInput,
|
||||
options: { forceOpus?: boolean; injectedClient?: Anthropic } = {}
|
||||
): Promise<LLMCallResult<AggregateReduceResponse> & { model_used: ModelId }> {
|
||||
const model = selectReduceModel(input.fingerprints.length, options.forceOpus);
|
||||
const result = await callLLMStage({
|
||||
model,
|
||||
system: SYSTEM_PROMPT,
|
||||
user: buildUserPayload(input),
|
||||
schema: AggregateReduceResponse,
|
||||
maxTokens: REDUCE_MAX_TOKENS,
|
||||
client: options.injectedClient,
|
||||
});
|
||||
return { ...result, model_used: model };
|
||||
}
|
||||
|
||||
export const _AGGREGATE_REDUCE_INTERNALS = { SYSTEM_PROMPT, REDUCE_MAX_TOKENS };
|
||||
|
|
@ -41,7 +41,7 @@ Tag actor_type by the email domain we already classified for you (provided in th
|
|||
|
||||
Respond ONLY with JSON. No prose, no code fences. Schema:
|
||||
{
|
||||
"summary": string,
|
||||
"summary": string, // markdown, 2-4 sentences, neutral status briefing
|
||||
"timeline": [
|
||||
{
|
||||
"timestamp": string,
|
||||
|
|
@ -57,9 +57,12 @@ Respond ONLY with JSON. No prose, no code fences. Schema:
|
|||
"gaps": [
|
||||
{ "description": string, "severity": "low" | "medium" | "high", "evidence_timestamps": string[] }
|
||||
],
|
||||
"next_step": string,
|
||||
"next_step_rationale": string,
|
||||
"post_resolution_analysis": string | null,
|
||||
"next_step": string, // markdown, single concrete action; may use
|
||||
// **bold** for the action verb and bullet
|
||||
// sub-steps if multi-part
|
||||
"next_step_rationale": string, // markdown, 1-2 short paragraphs; if there are
|
||||
// 3+ competing considerations, use a bulleted list
|
||||
"post_resolution_analysis": string | null, // markdown, same conventions as above
|
||||
"confidence_score": number,
|
||||
"needs_human_review": boolean,
|
||||
"human_review_reasons": string[],
|
||||
|
|
@ -69,6 +72,32 @@ Respond ONLY with JSON. No prose, no code fences. Schema:
|
|||
]
|
||||
}
|
||||
|
||||
Formatting rules for prose fields (summary, next_step, next_step_rationale,
|
||||
post_resolution_analysis):
|
||||
|
||||
- Output is markdown and will be rendered with a markdown renderer. Use **bold**
|
||||
for emphasis on key terms or actions. Use *italics* sparingly for client-facing
|
||||
language being quoted. Use bullet lists for enumerable items.
|
||||
|
||||
- Do NOT use markdown headers (#, ##, ###). These fields render inside cards that
|
||||
already have their own headings.
|
||||
|
||||
- For summary: write as a neutral status briefing a manager could read in 10
|
||||
seconds. Lead with the most important fact. Plain language. No hedging.
|
||||
|
||||
- For next_step: state the concrete action in the first sentence, with the action
|
||||
verb in **bold**. If there are sub-steps, follow with a bulleted list. Address
|
||||
the action to the technician, not the customer.
|
||||
|
||||
- For next_step_rationale: open with one short sentence stating the core reason.
|
||||
Follow with a short paragraph elaborating, OR a bulleted list if there are 3+
|
||||
distinct considerations. If there's a competing alternative worth noting, name
|
||||
it explicitly: "Considered X but rejected because..."
|
||||
|
||||
- Avoid filler phrases. Banned openings: "It's worth noting that...", "It's
|
||||
important to understand...", "Based on the available information...",
|
||||
"After reviewing the ticket...". Get to the point.
|
||||
|
||||
Set needs_human_review = true if any of:
|
||||
- confidence_score < 0.6
|
||||
- gaps contain any "high" severity item
|
||||
|
|
|
|||
135
lib/services/analyzer/stages/stage6-fingerprint.ts
Normal file
135
lib/services/analyzer/stages/stage6-fingerprint.ts
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
/**
|
||||
* Stage 6 — Aggregate fingerprint (Haiku).
|
||||
*
|
||||
* Runs after persistence (Stage 5) on every analysis. Extracts a structured
|
||||
* fingerprint used by the cross-ticket aggregate report. Failure-tolerant:
|
||||
* the worker logs and moves on; the analysis row stays usable, just won't
|
||||
* appear in aggregate reports until re-fingerprinted via the backfill script.
|
||||
*
|
||||
* Spec: docs/ticket-analyzer-phase2-spec.md → Section D.3
|
||||
*/
|
||||
|
||||
import {
|
||||
AggregateFingerprint,
|
||||
type DeepAnalysisResponse,
|
||||
type OpusResponse,
|
||||
type TriageResponse,
|
||||
} from '@/lib/types/analyzer';
|
||||
import { callLLMStage, type LLMCallResult } from '@/lib/services/llm/call';
|
||||
import { HAIKU } from '@/lib/services/llm/models';
|
||||
import type Anthropic from '@anthropic-ai/sdk';
|
||||
|
||||
const STAGE6_MAX_TOKENS = 4_000;
|
||||
|
||||
const SYSTEM_PROMPT = `You are extracting a structured fingerprint from a completed ticket analysis to enable cross-ticket aggregation. Your job is precision and consistency, not creativity.
|
||||
|
||||
You will receive:
|
||||
- The Sonnet-tier analysis output (summary, gaps, what_was_done, etc.)
|
||||
- The triage output from Stage 1 (entities, category)
|
||||
- Optionally the Opus-tier updates if deep review ran
|
||||
|
||||
Produce a fingerprint matching this exact schema. Respond ONLY with JSON, no prose, no fences:
|
||||
|
||||
{
|
||||
"category": string,
|
||||
"subcategories": string[],
|
||||
"ticket_type_inferred": string,
|
||||
"root_cause_class": "configuration_drift" | "user_error" | "vendor_issue" | "hardware_failure" | "documentation_gap" | "process_gap" | "unknown" | "other",
|
||||
|
||||
"client_name": string,
|
||||
"vendors_involved": string[],
|
||||
"applications_involved": string[],
|
||||
"device_classes": string[],
|
||||
|
||||
"wulf_actions_taken": string[],
|
||||
"vendor_cases_opened": number,
|
||||
"resolution_path": "resolved_by_wulf" | "resolved_by_vendor" | "resolved_by_client" | "unresolved" | "self_resolved_before_wulf_action",
|
||||
|
||||
"documentation_gaps_observed": [
|
||||
{ "description": string, "confidence": "low" | "medium" | "high" }
|
||||
],
|
||||
"process_gaps_observed": [
|
||||
{ "description": string, "severity": "low" | "medium" | "high" }
|
||||
],
|
||||
|
||||
"similar_to_signals": string[],
|
||||
"tags": string[],
|
||||
|
||||
"generated_by_model": string,
|
||||
"generated_at": string
|
||||
}
|
||||
|
||||
Strict rules:
|
||||
- Use only the listed enum values for root_cause_class and resolution_path.
|
||||
- documentation_gaps_observed and process_gaps_observed should each have at most 5 entries. Quality over quantity. Each entry's description must be specific enough that two different analyses describing the same underlying gap would produce similar text.
|
||||
- Tags should be lowercase, hyphenated, and stable. Prefer reusing common tags (vertafore, ams360, m365-licensing, backup-veeam, etc.) over inventing new ones.
|
||||
- For similar_to_signals, write short observations like "vendor case opened before checking documentation portal" or "status not advanced after customer self-resolution" — patterns the reduce step can cluster.
|
||||
- generated_by_model should be the literal string "claude-haiku-4-5".
|
||||
- generated_at should be the current ISO-8601 timestamp with timezone offset.`;
|
||||
|
||||
export interface FingerprintInput {
|
||||
triage: TriageResponse;
|
||||
sonnet: DeepAnalysisResponse;
|
||||
opus?: OpusResponse | null;
|
||||
}
|
||||
|
||||
export function buildFingerprintUserPayload(input: FingerprintInput): string {
|
||||
const compactSonnet = {
|
||||
summary: input.sonnet.summary,
|
||||
what_was_done: input.sonnet.what_was_done,
|
||||
what_should_have_been_done: input.sonnet.what_should_have_been_done,
|
||||
gaps: input.sonnet.gaps,
|
||||
next_step: input.sonnet.next_step,
|
||||
next_step_rationale: input.sonnet.next_step_rationale,
|
||||
post_resolution_analysis: input.sonnet.post_resolution_analysis,
|
||||
confidence_score: input.sonnet.confidence_score,
|
||||
needs_human_review: input.sonnet.needs_human_review,
|
||||
human_review_reasons: input.sonnet.human_review_reasons,
|
||||
itglue_docs_referenced: input.sonnet.itglue_docs_referenced.map((d) => ({
|
||||
name: d.name,
|
||||
doc_type: d.doc_type,
|
||||
relevance_reason: d.relevance_reason,
|
||||
})),
|
||||
};
|
||||
|
||||
const parts = [
|
||||
`=== TRIAGE METADATA (Stage 1) ===`,
|
||||
JSON.stringify(input.triage, null, 2),
|
||||
``,
|
||||
`=== ANALYSIS (Sonnet) ===`,
|
||||
JSON.stringify(compactSonnet, null, 2),
|
||||
];
|
||||
|
||||
if (input.opus) {
|
||||
parts.push(``, `=== DEEP-REVIEW UPDATES (Opus) ===`, JSON.stringify(input.opus, null, 2));
|
||||
}
|
||||
|
||||
return parts.join('\n');
|
||||
}
|
||||
|
||||
export async function runFingerprintStage(
|
||||
input: FingerprintInput,
|
||||
injectedClient?: Anthropic
|
||||
): Promise<LLMCallResult<AggregateFingerprint>> {
|
||||
const result = await callLLMStage({
|
||||
model: HAIKU,
|
||||
system: SYSTEM_PROMPT,
|
||||
user: buildFingerprintUserPayload(input),
|
||||
schema: AggregateFingerprint,
|
||||
maxTokens: STAGE6_MAX_TOKENS,
|
||||
client: injectedClient,
|
||||
});
|
||||
|
||||
// Server-authoritative model and timestamp — model output for these is
|
||||
// advisory; we always overwrite with truth.
|
||||
return {
|
||||
...result,
|
||||
data: {
|
||||
...result.data,
|
||||
generated_by_model: HAIKU,
|
||||
generated_at: new Date().toISOString(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export const _STAGE6_INTERNALS = { SYSTEM_PROMPT, STAGE6_MAX_TOKENS };
|
||||
|
|
@ -5,6 +5,7 @@ import { analyzerWorker } from './worker';
|
|||
import * as dataAccess from './data-access';
|
||||
import * as persistence from './persistence';
|
||||
import * as pipelineModule from './pipeline';
|
||||
import * as stage6Module from './stages/stage6-fingerprint';
|
||||
import type { RawTicketBundle } from './preprocessor';
|
||||
|
||||
const FIXTURE = JSON.parse(
|
||||
|
|
@ -20,6 +21,8 @@ let insertSpy: ReturnType<typeof vi.spyOn>;
|
|||
let completeSpy: ReturnType<typeof vi.spyOn>;
|
||||
let failSpy: ReturnType<typeof vi.spyOn>;
|
||||
let updateStatusSpy: ReturnType<typeof vi.spyOn>;
|
||||
let bulkStageSpy: ReturnType<typeof vi.spyOn>;
|
||||
let insertFailedSpy: ReturnType<typeof vi.spyOn>;
|
||||
|
||||
beforeEach(() => {
|
||||
loadSpy = vi.spyOn(dataAccess, 'loadTicketBundle');
|
||||
|
|
@ -30,6 +33,18 @@ beforeEach(() => {
|
|||
completeSpy = vi.spyOn(persistence, 'completeJob').mockResolvedValue();
|
||||
failSpy = vi.spyOn(persistence, 'failJob').mockResolvedValue();
|
||||
updateStatusSpy = vi.spyOn(persistence, 'updateJobStatus').mockResolvedValue();
|
||||
bulkStageSpy = vi
|
||||
.spyOn(persistence, 'bulkInsertStageExecutions')
|
||||
.mockResolvedValue();
|
||||
insertFailedSpy = vi
|
||||
.spyOn(persistence, 'insertFailedAnalysis')
|
||||
.mockResolvedValue({ id: 'failed_uuid', analysis_version: 1 });
|
||||
vi.spyOn(persistence, 'updateAnalysisFingerprint').mockResolvedValue();
|
||||
// Default Stage 6 to a no-op success in tests; worker treats failures as
|
||||
// non-fatal anyway, so we just need it not to make real HTTP calls.
|
||||
vi.spyOn(stage6Module, 'runFingerprintStage').mockRejectedValue(
|
||||
new Error('fingerprint stub: tests do not exercise stage 6')
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
|
|
@ -94,6 +109,9 @@ describe('analyzerWorker.runJob', () => {
|
|||
filtered_noise_count: 8,
|
||||
itglue_search_used: false,
|
||||
itglue_org_resolved: false,
|
||||
triage_response: {} as never,
|
||||
sonnet_response: {} as never,
|
||||
opus_response: null,
|
||||
model_traces: {},
|
||||
});
|
||||
|
||||
|
|
@ -252,6 +270,9 @@ describe('analyzerWorker.runJob', () => {
|
|||
filtered_noise_count: 0,
|
||||
itglue_search_used: false,
|
||||
itglue_org_resolved: false,
|
||||
triage_response: {} as never,
|
||||
sonnet_response: {} as never,
|
||||
opus_response: null,
|
||||
model_traces: {},
|
||||
};
|
||||
}) as never);
|
||||
|
|
|
|||
|
|
@ -14,14 +14,23 @@
|
|||
*/
|
||||
|
||||
import {
|
||||
bulkInsertStageExecutions,
|
||||
claimQueuedJob,
|
||||
completeJob,
|
||||
failJob,
|
||||
insertAnalysis,
|
||||
insertFailedAnalysis,
|
||||
updateAnalysisFingerprint,
|
||||
updateJobStatus,
|
||||
} from './persistence';
|
||||
import { loadTicketBundle, TicketNotFoundError } from './data-access';
|
||||
import { runPipeline, type PipelineResult } from './pipeline';
|
||||
import { runFingerprintStage } from './stages/stage6-fingerprint';
|
||||
import { HAIKU } from '@/lib/services/llm/models';
|
||||
import type {
|
||||
PreprocessedTicket,
|
||||
StageExecutionRecord,
|
||||
} from '@/lib/types/analyzer';
|
||||
|
||||
const POLL_INTERVAL_MS = 2_000;
|
||||
|
||||
|
|
@ -82,6 +91,12 @@ class AnalyzerWorker {
|
|||
ticketNumber: string,
|
||||
triggeredByUserId: string | null
|
||||
): Promise<{ analysis_id: string | null; outcome: PipelineResult['outcome'] | 'failed' }> {
|
||||
// Phase 2: collect per-stage records as the pipeline runs, plus the
|
||||
// preprocessed bundle, so we can persist a failed analyzer_analyses row
|
||||
// (with source_snapshot) when a stage throws.
|
||||
const stageRecords: StageExecutionRecord[] = [];
|
||||
let capturedPre: PreprocessedTicket | null = null;
|
||||
|
||||
try {
|
||||
const bundle = await loadTicketBundle(ticketNumber);
|
||||
|
||||
|
|
@ -90,6 +105,12 @@ class AnalyzerWorker {
|
|||
{},
|
||||
{
|
||||
onStage: (stage) => updateJobStatus(jobId, stage),
|
||||
onStageRecord: (rec) => {
|
||||
stageRecords.push(rec);
|
||||
},
|
||||
onPreprocessed: (pre) => {
|
||||
capturedPre = pre;
|
||||
},
|
||||
}
|
||||
);
|
||||
|
||||
|
|
@ -117,8 +138,52 @@ class AnalyzerWorker {
|
|||
analysis: result.analysis,
|
||||
filtered_noise_count: result.filtered_noise_count,
|
||||
model_traces: result.model_traces,
|
||||
source_snapshot: result.pre.events,
|
||||
});
|
||||
|
||||
// Stage 6 — fingerprint. Failure-tolerant: log and continue.
|
||||
const fpStart = new Date();
|
||||
let fpInputTokens: number | null = null;
|
||||
let fpOutputTokens: number | null = null;
|
||||
let fpOutput: unknown = {};
|
||||
let fpErr: Error | null = null;
|
||||
try {
|
||||
const fp = await runFingerprintStage({
|
||||
triage: result.triage_response,
|
||||
sonnet: result.sonnet_response,
|
||||
opus: result.opus_response,
|
||||
});
|
||||
fpInputTokens = fp.usage.input_tokens;
|
||||
fpOutputTokens = fp.usage.output_tokens;
|
||||
fpOutput = fp.data;
|
||||
await updateAnalysisFingerprint(inserted.id, fp.data);
|
||||
} catch (err) {
|
||||
fpErr = err instanceof Error ? err : new Error(String(err));
|
||||
console.warn(
|
||||
`[ANALYZER-WORKER] fingerprint failed for analysis ${inserted.id}: ${fpErr.message}`
|
||||
);
|
||||
}
|
||||
const fpEnd = new Date();
|
||||
stageRecords.push({
|
||||
stage: 'fingerprint',
|
||||
stage_order: 6,
|
||||
model_id: HAIKU,
|
||||
input_payload: {
|
||||
triage_category: result.triage_response.category,
|
||||
ticket_number: result.pre.header.ticket_number,
|
||||
opus_used: result.opus_response !== null,
|
||||
},
|
||||
output_payload: fpErr ? {} : fpOutput,
|
||||
input_tokens: fpInputTokens,
|
||||
output_tokens: fpOutputTokens,
|
||||
latency_ms: fpEnd.getTime() - fpStart.getTime(),
|
||||
started_at: fpStart,
|
||||
completed_at: fpEnd,
|
||||
error_message: fpErr ? fpErr.message : null,
|
||||
});
|
||||
|
||||
await bulkInsertStageExecutions(inserted.id, stageRecords);
|
||||
|
||||
await completeJob(jobId, inserted.id);
|
||||
return { analysis_id: inserted.id, outcome: 'complete' };
|
||||
} catch (err) {
|
||||
|
|
@ -131,6 +196,37 @@ class AnalyzerWorker {
|
|||
err instanceof TicketNotFoundError
|
||||
? `Ticket ${ticketNumber} not found in local mirror — confirm sync is current.`
|
||||
: message;
|
||||
|
||||
// Phase 2: when we have a preprocessed bundle, persist a 'failed'
|
||||
// analyzer_analyses row with source_snapshot + accumulated stage rows.
|
||||
// Best-effort — if this fails we still fail the job below.
|
||||
if (capturedPre !== null) {
|
||||
const pre: PreprocessedTicket = capturedPre;
|
||||
try {
|
||||
const failedAnalysis = await insertFailedAnalysis({
|
||||
ticket_number: pre.header.ticket_number,
|
||||
autotask_ticket_id: pre.header.autotask_ticket_id,
|
||||
content_hash: pre.content_hash,
|
||||
triggered_by_user_id: triggeredByUserId,
|
||||
source_snapshot: pre.events,
|
||||
filtered_noise_count: pre.counts.filtered_noise,
|
||||
error_message: reason,
|
||||
partial_input_tokens: 0,
|
||||
partial_output_tokens: 0,
|
||||
partial_cost_usd: 0,
|
||||
haiku_used: stageRecords.some((r) => r.stage === 'triage'),
|
||||
sonnet_used: stageRecords.some((r) => r.stage === 'analyze'),
|
||||
opus_used: stageRecords.some((r) => r.stage === 'deep_review'),
|
||||
});
|
||||
await bulkInsertStageExecutions(failedAnalysis.id, stageRecords);
|
||||
} catch (persistErr) {
|
||||
console.error(
|
||||
`[ANALYZER-WORKER] failed to persist failed-analysis row for job ${jobId}:`,
|
||||
persistErr
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
await failJob(jobId, reason);
|
||||
return { analysis_id: null, outcome: 'failed' };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -246,6 +246,206 @@ export const AnalyzerJob = z.object({
|
|||
});
|
||||
export type AnalyzerJob = z.infer<typeof AnalyzerJob>;
|
||||
|
||||
// =============================================================================
|
||||
// Phase 2 — Stage 6 fingerprint schema (cross-ticket aggregation).
|
||||
// =============================================================================
|
||||
|
||||
export const RootCauseClass = z.enum([
|
||||
'configuration_drift',
|
||||
'user_error',
|
||||
'vendor_issue',
|
||||
'hardware_failure',
|
||||
'documentation_gap',
|
||||
'process_gap',
|
||||
'unknown',
|
||||
'other',
|
||||
]);
|
||||
export type RootCauseClass = z.infer<typeof RootCauseClass>;
|
||||
|
||||
export const ResolutionPath = z.enum([
|
||||
'resolved_by_wulf',
|
||||
'resolved_by_vendor',
|
||||
'resolved_by_client',
|
||||
'unresolved',
|
||||
'self_resolved_before_wulf_action',
|
||||
]);
|
||||
export type ResolutionPath = z.infer<typeof ResolutionPath>;
|
||||
|
||||
export const FingerprintConfidence = z.enum(['low', 'medium', 'high']);
|
||||
export type FingerprintConfidence = z.infer<typeof FingerprintConfidence>;
|
||||
|
||||
export const AggregateFingerprint = z.object({
|
||||
category: z.string(),
|
||||
subcategories: z.array(z.string()),
|
||||
ticket_type_inferred: z.string(),
|
||||
root_cause_class: RootCauseClass,
|
||||
|
||||
client_name: z.string(),
|
||||
vendors_involved: z.array(z.string()),
|
||||
applications_involved: z.array(z.string()),
|
||||
device_classes: z.array(z.string()),
|
||||
|
||||
wulf_actions_taken: z.array(z.string()),
|
||||
vendor_cases_opened: z.number().int().nonnegative(),
|
||||
resolution_path: ResolutionPath,
|
||||
|
||||
documentation_gaps_observed: z
|
||||
.array(
|
||||
z.object({
|
||||
description: z.string(),
|
||||
confidence: FingerprintConfidence,
|
||||
})
|
||||
)
|
||||
.max(5),
|
||||
process_gaps_observed: z
|
||||
.array(
|
||||
z.object({
|
||||
description: z.string(),
|
||||
severity: Severity,
|
||||
})
|
||||
)
|
||||
.max(5),
|
||||
|
||||
similar_to_signals: z.array(z.string()),
|
||||
tags: z.array(z.string()),
|
||||
|
||||
generated_by_model: z.string(),
|
||||
generated_at: z.string().datetime({ offset: true }),
|
||||
});
|
||||
export type AggregateFingerprint = z.infer<typeof AggregateFingerprint>;
|
||||
|
||||
// =============================================================================
|
||||
// Phase 2 — Stage execution row (per-stage I/O persistence).
|
||||
// =============================================================================
|
||||
|
||||
export const StageName = z.enum([
|
||||
'preprocess',
|
||||
'triage',
|
||||
'itglue',
|
||||
'analyze',
|
||||
'deep_review',
|
||||
'fingerprint',
|
||||
]);
|
||||
export type StageName = z.infer<typeof StageName>;
|
||||
|
||||
/**
|
||||
* In-memory shape used by the pipeline to record per-stage I/O. The pipeline
|
||||
* pushes one of these into the worker's array via the onStageRecord callback,
|
||||
* and the worker bulk-inserts them after the analyzer_analyses row exists
|
||||
* (success OR failure).
|
||||
*/
|
||||
export interface StageExecutionRecord {
|
||||
stage: StageName;
|
||||
stage_order: number;
|
||||
model_id: string | null;
|
||||
input_payload: unknown;
|
||||
output_payload: unknown;
|
||||
input_tokens: number | null;
|
||||
output_tokens: number | null;
|
||||
latency_ms: number | null;
|
||||
started_at: Date;
|
||||
completed_at: Date;
|
||||
error_message: string | null;
|
||||
}
|
||||
|
||||
export const StageExecution = z.object({
|
||||
id: z.string().uuid(),
|
||||
analysisId: z.string().uuid(),
|
||||
stage: StageName,
|
||||
stageOrder: z.number().int().positive(),
|
||||
modelId: z.string().nullable(),
|
||||
inputPayload: z.unknown(),
|
||||
outputPayload: z.unknown(),
|
||||
inputTokens: z.number().int().nullable(),
|
||||
outputTokens: z.number().int().nullable(),
|
||||
latencyMs: z.number().int().nullable(),
|
||||
startedAt: z.string().datetime({ offset: true }),
|
||||
completedAt: z.string().datetime({ offset: true }),
|
||||
errorMessage: z.string().nullable(),
|
||||
});
|
||||
export type StageExecution = z.infer<typeof StageExecution>;
|
||||
|
||||
// =============================================================================
|
||||
// Phase 2.6 — Aggregate report (reduce step) schemas.
|
||||
// =============================================================================
|
||||
|
||||
export const ITGlueCheck = z.enum([
|
||||
'no_doc_exists',
|
||||
'doc_exists_but_unused',
|
||||
'unable_to_verify',
|
||||
]);
|
||||
export type ITGlueCheck = z.infer<typeof ITGlueCheck>;
|
||||
|
||||
export const RecommendedActionType = z.enum([
|
||||
'documentation',
|
||||
'process',
|
||||
'training',
|
||||
'tooling',
|
||||
]);
|
||||
export type RecommendedActionType = z.infer<typeof RecommendedActionType>;
|
||||
|
||||
export const AggregateReduceResponse = z.object({
|
||||
documentation_gaps: z.array(
|
||||
z.object({
|
||||
gap: z.string(),
|
||||
frequency: z.number().int().nonnegative(),
|
||||
example_ticket_numbers: z.array(z.string()),
|
||||
evidence: z.string(),
|
||||
itglue_check: ITGlueCheck,
|
||||
})
|
||||
),
|
||||
process_gaps: z.array(
|
||||
z.object({
|
||||
gap: z.string(),
|
||||
frequency: z.number().int().nonnegative(),
|
||||
severity: Severity,
|
||||
example_ticket_numbers: z.array(z.string()),
|
||||
evidence: z.string(),
|
||||
})
|
||||
),
|
||||
client_patterns: z.array(
|
||||
z.object({
|
||||
client: z.string(),
|
||||
pattern: z.string(),
|
||||
frequency: z.number().int().nonnegative(),
|
||||
example_ticket_numbers: z.array(z.string()),
|
||||
})
|
||||
),
|
||||
recurrence_clusters: z.array(
|
||||
z.object({
|
||||
theme: z.string(),
|
||||
ticket_numbers: z.array(z.string()),
|
||||
summary: z.string(),
|
||||
})
|
||||
),
|
||||
systemic_observations: z.array(
|
||||
z.object({
|
||||
observation: z.string(),
|
||||
evidence: z.string(),
|
||||
severity: Severity,
|
||||
})
|
||||
),
|
||||
recommended_actions: z.array(
|
||||
z.object({
|
||||
action: z.string(),
|
||||
rationale: z.string(),
|
||||
priority: Severity,
|
||||
type: RecommendedActionType,
|
||||
})
|
||||
),
|
||||
executive_summary: z.string(),
|
||||
narrative_summary: z.string(),
|
||||
});
|
||||
export type AggregateReduceResponse = z.infer<typeof AggregateReduceResponse>;
|
||||
|
||||
export const AggregateReportStatus = z.enum([
|
||||
'pending',
|
||||
'running',
|
||||
'complete',
|
||||
'failed',
|
||||
]);
|
||||
export type AggregateReportStatus = z.infer<typeof AggregateReportStatus>;
|
||||
|
||||
// =============================================================================
|
||||
// API request bodies.
|
||||
// =============================================================================
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue