feat(log): add self-host audit PR correlation

src/selfhost/audit.ts

@@ -16,12 +16,19 @@ export interface AuditEvent {
   ts: number;             // Unix timestamp (ms)
   job_id: number | string;
   payload_type?: string | undefined;  // top-level `type` field from the job payload, if present
+  repo?: string | undefined;
+  pr_number?: number | undefined;
   latency_ms: number;     // wall time from claim to completion/failure
   attempts: number;       // total attempts consumed (1 = first-try success)
   error?: string;         // last error message, present for job_dead / job_error
   retry_after_ms?: number; // next retry delay for job_rate_limited
 }
 
+export interface AuditPayloadContext {
+  repo?: string | undefined;
+  pr_number?: number | undefined;
+}
+
 /** Emit a single audit event as a JSON line on stdout. */
 export function logAudit(ev: AuditEvent, traceParent?: string): void {
   process.stdout.write(JSON.stringify({ level: "audit", ...ev, ...otelTraceLogFields(traceParent) }) + "\n");
@@ -37,3 +44,96 @@ export function extractPayloadType(payload: string): string | undefined {
     return undefined;
   }
 }
+
+/** Extract repo / PR correlation labels from a raw job payload. Only safe scalar fields are returned. */
+export function extractPayloadContext(payload: string): AuditPayloadContext | undefined {
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(payload) as unknown;
+  } catch {
+    return undefined;
+  }
+  if (!isRecord(parsed)) return undefined;
+
+  const repo = extractRepo(parsed);
+  const prNumber = extractPrNumber(parsed);
+  if (repo === undefined && prNumber === undefined) return undefined;
+
+  const context: AuditPayloadContext = {};
+  if (repo !== undefined) context.repo = repo;
+  if (prNumber !== undefined) context.pr_number = prNumber;
+  return context;
+}
+
+function extractRepo(payload: Record<string, unknown>): string | undefined {
+  const repoFullName = stringField(payload, "repoFullName");
+  if (repoFullName !== undefined) return repoFullName;
+
+  const webhookPayload = recordField(payload, "payload");
+  if (webhookPayload === undefined) return undefined;
+
+  const repository = recordField(webhookPayload, "repository");
+  if (repository === undefined) return undefined;
+  return stringField(repository, "full_name");
+}
+
+function extractPrNumber(payload: Record<string, unknown>): number | undefined {
+  const prNumber = numberField(payload, "prNumber");
+  if (prNumber !== undefined) return prNumber;
+
+  const webhookPayload = recordField(payload, "payload");
+  if (webhookPayload === undefined) return undefined;
+
+  const pullRequest = recordField(webhookPayload, "pull_request");
+  if (pullRequest !== undefined) {
+    const pullRequestNumber = numberField(pullRequest, "number");
+    if (pullRequestNumber !== undefined) return pullRequestNumber;
+  }
+
+  const issue = recordField(webhookPayload, "issue");
+  if (issue !== undefined) {
+    const issueNumber = numberField(issue, "number");
+    if (issueNumber !== undefined && recordField(issue, "pull_request") !== undefined) return issueNumber;
+  }
+
+  const checkRun = recordField(webhookPayload, "check_run");
+  if (checkRun !== undefined) {
+    const checkRunNumber = firstPullRequestNumber(checkRun);
+    if (checkRunNumber !== undefined) return checkRunNumber;
+  }
+
+  const checkSuite = recordField(webhookPayload, "check_suite");
+  if (checkSuite !== undefined) return firstPullRequestNumber(checkSuite);
+
+  return undefined;
+}
+
+function firstPullRequestNumber(record: Record<string, unknown>): number | undefined {
+  const pullRequests = record.pull_requests;
+  if (!Array.isArray(pullRequests)) return undefined;
+  for (const pullRequest of pullRequests) {
+    if (!isRecord(pullRequest)) continue;
+    const number = numberField(pullRequest, "number");
+    if (number !== undefined) return number;
+  }
+  return undefined;
+}
+
+function recordField(record: Record<string, unknown>, field: string): Record<string, unknown> | undefined {
+  const value = record[field];
+  return isRecord(value) ? value : undefined;
+}
+
+function stringField(record: Record<string, unknown>, field: string): string | undefined {
+  const value = record[field];
+  return typeof value === "string" && value.length > 0 ? value : undefined;
+}
+
+function numberField(record: Record<string, unknown>, field: string): number | undefined {
+  const value = record[field];
+  return typeof value === "number" && Number.isFinite(value) ? value : undefined;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}

src/selfhost/pg-queue.ts

@@ -3,7 +3,7 @@
 // app instances sharing one Postgres can claim jobs concurrently without double-processing. size()/deadCount()
 // are async (the metrics gauges accept async samplers).
 import type { Pool } from "pg";
-import { logAudit, extractPayloadType } from "./audit";
+import { logAudit, extractPayloadType, extractPayloadContext } from "./audit";
 import { incr } from "./metrics";
 import { withReviewSpan } from "./tracing";
 import { withOtelSpan } from "./otel";
@@ -385,6 +385,7 @@ export function createPgQueue(
         return true;
       }
       const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined;
+      const payloadContext = extractPayloadContext(job.payload);
       const rateLimitAdmission = await rateLimitAdmissionDelayMs(message);
       if (rateLimitAdmission !== null) {
         const rateLimitMetric = githubRateLimitMetricContext(message, rateLimitAdmission);
@@ -437,6 +438,7 @@ export function createPgQueue(
           ts: Date.now(),
           job_id: job.id,
           payload_type: extractPayloadType(job.payload),
+          ...payloadContext,
           latency_ms: Date.now() - claimedAt,
           attempts: Number(job.attempts) + 1,
         }, jobTraceParent);
@@ -477,6 +479,7 @@ export function createPgQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             retry_after_ms: Math.max(0, retryAfter - Date.now()),
@@ -505,6 +508,7 @@ export function createPgQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             error: errMsg,
@@ -527,6 +531,7 @@ export function createPgQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             error: errMsg,

src/selfhost/sqlite-queue.ts

@@ -4,7 +4,7 @@
 // backing store differs. Single-process model: node:sqlite is synchronous + serial, so claim (SELECT→UPDATE)
 // is atomic with no row-lock dance.
 import type { SqliteDriver } from "./d1-adapter";
-import { logAudit, extractPayloadType } from "./audit";
+import { logAudit, extractPayloadType, extractPayloadContext } from "./audit";
 import { incr } from "./metrics";
 import { withReviewSpan } from "./tracing";
 import { withOtelSpan } from "./otel";
@@ -325,6 +325,7 @@ export function createSqliteQueue(
         return true;
       }
       const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined;
+      const payloadContext = extractPayloadContext(job.payload);
       const rateLimitAdmission = rateLimitAdmissionDelayMs(driver, message);
       if (rateLimitAdmission !== null) {
         const rateLimitMetric = githubRateLimitMetricContext(message, rateLimitAdmission);
@@ -377,6 +378,7 @@ export function createSqliteQueue(
           ts: Date.now(),
           job_id: job.id,
           payload_type: extractPayloadType(job.payload),
+          ...payloadContext,
           latency_ms: Date.now() - claimedAt,
           attempts: job.attempts + 1,
         }, jobTraceParent);
@@ -417,6 +419,7 @@ export function createSqliteQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             retry_after_ms: Math.max(0, retryAfter - Date.now()),
@@ -445,6 +448,7 @@ export function createSqliteQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             error: errMsg,
@@ -467,6 +471,7 @@ export function createSqliteQueue(
             ts: Date.now(),
             job_id: job.id,
             payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
             latency_ms: Date.now() - claimedAt,
             attempts,
             error: errMsg,

test/unit/selfhost-audit.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { logAudit, extractPayloadType } from "../../src/selfhost/audit";
+import { logAudit, extractPayloadContext, extractPayloadType } from "../../src/selfhost/audit";
 
 describe("logAudit", () => {
   const written: string[] = [];
@@ -78,3 +78,118 @@ describe("extractPayloadType", () => {
     expect(extractPayloadType("{}")).toBeUndefined();
   });
 });
+
+describe("extractPayloadContext", () => {
+  it("returns top-level repo and PR fields from internal PR jobs", () => {
+    expect(extractPayloadContext(JSON.stringify({ type: "agent-regate-pr", repoFullName: "JSONbored/gittensory", prNumber: 2087 }))).toEqual({
+      repo: "JSONbored/gittensory",
+      pr_number: 2087,
+    });
+  });
+
+  it("returns repo-only context for repo-scoped background jobs", () => {
+    expect(extractPayloadContext(JSON.stringify({ type: "rag-index-repo", repoFullName: "JSONbored/gittensory" }))).toEqual({
+      repo: "JSONbored/gittensory",
+    });
+  });
+
+  it("returns PR-only context when a webhook payload lacks repository data", () => {
+    expect(extractPayloadContext(JSON.stringify({ type: "github-webhook", payload: { pull_request: { number: 87 } } }))).toEqual({
+      pr_number: 87,
+    });
+  });
+
+  it("returns nested repository and pull request fields from GitHub webhook jobs", () => {
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          pull_request: { number: 2087 },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 });
+  });
+
+  it("falls back from wrong top-level fields to nested webhook context", () => {
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        repoFullName: 123,
+        prNumber: "2087",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          pull_request: { number: 2087 },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 });
+  });
+
+  it("extracts PR numbers from PR issue-comment payloads without treating regular issues as PRs", () => {
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          issue: { number: 2087, pull_request: {} },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 });
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          issue: { number: 2087 },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory" });
+  });
+
+  it("extracts PR numbers from check-run and check-suite pull request arrays", () => {
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          check_run: { pull_requests: [null, { number: "bad" }, { number: 2087 }] },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 });
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        payload: {
+          repository: { full_name: "JSONbored/gittensory" },
+          check_suite: { pull_requests: [{ number: 2088 }] },
+        },
+      })),
+    ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2088 });
+  });
+
+  it("returns undefined for missing, malformed, and non-object payloads", () => {
+    expect(extractPayloadContext(JSON.stringify({ type: "refresh-registry" }))).toBeUndefined();
+    expect(extractPayloadContext("not-json")).toBeUndefined();
+    expect(extractPayloadContext("null")).toBeUndefined();
+    expect(extractPayloadContext("[]")).toBeUndefined();
+    expect(extractPayloadContext('"string"')).toBeUndefined();
+  });
+
+  it("returns undefined for wrong typed and non-finite context fields", () => {
+    expect(
+      extractPayloadContext(JSON.stringify({
+        type: "github-webhook",
+        repoFullName: "",
+        prNumber: "2087",
+        payload: {
+          repository: { full_name: "" },
+          pull_request: { number: "2087" },
+          issue: { number: "2087", pull_request: {} },
+          check_run: { pull_requests: "bad" },
+          check_suite: { pull_requests: [] },
+        },
+      })),
+    ).toBeUndefined();
+    expect(extractPayloadContext('{"prNumber":1e999}')).toBeUndefined();
+  });
+});

test/unit/selfhost-pg-queue.test.ts

@@ -508,6 +508,10 @@ describe("createPgQueue (durable #977)", () => {
     m.enqueueJob("1", {
       type: "github-webhook",
       traceParent: "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01",
+      payload: {
+        repository: { full_name: "JSONbored/gittensory" },
+        pull_request: { number: 1629 },
+      },
     });
     const q = createPgQueue(m.pool, async () => undefined);
 
@@ -516,6 +520,8 @@ describe("createPgQueue (durable #977)", () => {
 
     const audit = writes.find((line) => line.includes('"event":"job_complete"'));
     expect(JSON.parse(audit!) as Record<string, unknown>).toMatchObject({
+      repo: "JSONbored/gittensory",
+      pr_number: 1629,
       trace_id: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
     });
   });

test/unit/selfhost-sqlite-queue.test.ts

@@ -98,6 +98,8 @@ describe("createSqliteQueue (durable #980)", () => {
 
     const audit = writes.find((line) => line.includes('"event":"job_complete"'));
     expect(JSON.parse(audit!) as Record<string, unknown>).toMatchObject({
+      repo: "JSONbored/gittensory",
+      pr_number: 1629,
       trace_id: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
     });
   });