feat(quick-260717-a19): add idempotency guard + retry-parse on ticket.update

- parseAndStoreMessage (Defect 3): short-circuit with
  { stored: false, reason: 'already-parsed' } when a messages row already
  exists for the report, before any Autotask attachment fetch
- webhook-service (Defect 2): new retryPhishingParseOnUpdate wired into
  ticket.update fire-and-forget path; retries the missing-EML parse for a
  flagged, unparsed, auto_parse-gated report — no new cron/polling, reuses
  existing update traffic, safe to fire repeatedly thanks to the new
  idempotency guard
- Adjust eml-service test mock default so the new leading existence-check
  query doesn't short-circuit existing happy-path tests; add new test for
  the already-parsed short-circuit

lib/services/phishing-eml-service.test.ts

@@ -76,8 +76,16 @@ describe('parseAndStoreMessage', () => {
     isB2ConfiguredMock.mockReset();
     presignUploadMock.mockReset();
 
-    // Default: any INSERT ... RETURNING resolves with a single row.
-    queryMock.mockResolvedValue({ rows: [{ id: 'message-uuid-1' }], rowCount: 1 });
+    // Default: any INSERT ... RETURNING resolves with a single row. The new
+    // leading `SELECT id FROM messages WHERE report_id` idempotency check
+    // (Defect 3, 260717-a19) must resolve empty by default so existing
+    // happy-path tests aren't short-circuited as 'already-parsed'.
+    queryMock.mockImplementation((sql: string) => {
+      if (String(sql).includes('FROM messages WHERE report_id')) {
+        return Promise.resolve({ rows: [], rowCount: 0 });
+      }
+      return Promise.resolve({ rows: [{ id: 'message-uuid-1' }], rowCount: 1 });
+    });
 
     global.fetch = vi.fn().mockResolvedValue(new Response('ok', { status: 200 }));
   });
@@ -192,4 +200,20 @@ describe('parseAndStoreMessage', () => {
       })
     );
   });
+
+  it('returns { stored: false, reason: "already-parsed" } and does no work when a messages row already exists for the report (Defect 3, 260717-a19)', async () => {
+    queryMock.mockImplementation((sql: string) => {
+      if (String(sql).includes('FROM messages WHERE report_id')) {
+        return Promise.resolve({ rows: [{ id: 'existing-message-uuid' }], rowCount: 1 });
+      }
+      return Promise.resolve({ rows: [{ id: 'message-uuid-1' }], rowCount: 1 });
+    });
+
+    const result = await parseAndStoreMessage({ reportId: REPORT_ID, ticketId: TICKET_ID });
+
+    expect(result).toEqual({ stored: false, reason: 'already-parsed' });
+    expect(getAttachmentsMock).not.toHaveBeenCalled();
+    expect(getAttachmentContentMock).not.toHaveBeenCalled();
+    expect(callsContaining('INSERT INTO messages')).toHaveLength(0);
+  });
 });

lib/services/phishing-eml-service.ts

@@ -52,6 +52,20 @@ export async function parseAndStoreMessage(
   const { reportId, ticketId } = input;
 
   try {
+    // Idempotency guard (Defect 3, quick task 260717-a19): a ticket.update
+    // webhook may retry parsing for a report that already has a messages
+    // row (e.g. the CREATE-time attempt succeeded after all, or a prior
+    // UPDATE retry already parsed it). Short-circuit before any Autotask
+    // attachment fetch so a repeated call is a cheap no-op, never a
+    // duplicate messages row.
+    const existing = await postgresClient.query<{ id: string }>(
+      `SELECT id FROM messages WHERE report_id = $1 LIMIT 1`,
+      [reportId]
+    );
+    if (existing.rows.length > 0) {
+      return { stored: false, reason: 'already-parsed' };
+    }
+
     // reports.evidence only stores fullPath/title/contentType (no attachment
     // id) — list live so we have the attachment id needed for the
     // per-attachment content fetch below.

lib/services/webhook-service.ts

@@ -125,6 +125,19 @@ export class WebhookService {
         );
       }
 
+      // Defect 2 (quick task 260717-a19): the CREATE-time .eml parse attempt
+      // can race the Autotask attachment being available, permanently
+      // starving the classifier of evidence with no retry. Reuse the
+      // ticket.update traffic every flagged ticket already receives to
+      // retry the missing-EML parse only — never re-run detect/group/
+      // classify/report here. Safe to fire on every update because
+      // parseAndStoreMessage is now idempotent (Defect 3).
+      if (payload.entityType === WebhookEntityType.TICKETS && payload.eventType === WebhookEventType.UPDATE) {
+        this.retryPhishingParseOnUpdate(payload).catch(err =>
+          console.error('[WEBHOOK] Phishing retry-parse error:', err)
+        );
+      }
+
 
       return {
         success: true,
@@ -506,6 +519,55 @@ export class WebhookService {
     }
   }
 
+  /**
+   * Defect 2 fix (quick task 260717-a19): retries the missing-EML parse for
+   * an already-flagged phishing report on ticket.update webhook traffic.
+   *
+   * Autotask's attachment-available timing can lag the ticket.created
+   * webhook by more than the CREATE-time parse attempt allows for, leaving
+   * a flagged report permanently without a `messages` row (no retry existed
+   * before this fix). Rather than add new polling/cron, this reuses the
+   * ticket.update events a flagged ticket already receives (5+ observed on
+   * Seubert ticket 699456) to attempt the parse again — bounded to exactly
+   * one `parseAndStoreMessage` call per update, gated on auto_parse, and
+   * only when there is still no messages row (parseAndStoreMessage's own
+   * Defect 3 guard makes repeat calls safe regardless).
+   *
+   * Deliberately narrow: no detection, grouping, classify, or report here —
+   * only the missing-EML retry-parse.
+   */
+  private async retryPhishingParseOnUpdate(payload: AutotaskWebhookPayload): Promise<void> {
+    const reportRow = await postgresClient.query<{ id: string; company_id: number | null }>(
+      `SELECT id::text AS id, company_id FROM reports WHERE ticket_id = $1`,
+      [payload.entityId]
+    );
+    const report = reportRow.rows[0];
+    if (!report) {
+      // Not a flagged phishing ticket — nothing to retry.
+      return;
+    }
+
+    const messageRow = await postgresClient.query<{ id: string }>(
+      `SELECT id FROM messages WHERE report_id = $1 LIMIT 1`,
+      [report.id]
+    );
+    if (messageRow.rows.length > 0) {
+      // Already parsed — nothing to retry.
+      return;
+    }
+
+    const gate = await getCompanyAutomationGate(report.company_id);
+    if (!gate.autoParse) {
+      return;
+    }
+
+    try {
+      await parseAndStoreMessage({ reportId: report.id, ticketId: Number(payload.entityId) });
+    } catch (err) {
+      console.error('[WEBHOOK] retryPhishingParseOnUpdate parse error', err);
+    }
+  }
+
   /**
    * Phase 23 D-04/D-06/D-07: runs the opted-in parse -> classify -> report
    * chain for a company after detection + grouping have already run