Audit pipeline PR #9: determinism + fact extraction + verifier gate + KB stats + context injection (PR #9)
Bundles PR #9's work for the audit pipeline: - N=3 consensus on cloud inference (gpt-oss:120b parallel) with qwen3-coder:480b tie-breaker - audit_discrepancies.jsonl logs N-run disagreements - scrum_master reviews route through llm_team fact extraction; source="scrum_review" - Verifier-gated persistence: drops INCORRECT, keeps UNVERIFIABLE/UNCHECKED; schema_version:2 - scrum_master_reviewed flag on accepted reviews - auditor/kb_stats.ts: on-demand observability script - claim_parser history/proof pattern class (verified-on-PR, was-flipping, the-proven-X) - claim_parser quoted-string guard (mirrors static.ts fix) - fact_extractor project context injection via docs/AUDITOR_CONTEXT.md - Fixed verifier-verdict parser to handle multiple gemma2 output formats Empirical: 3-run determinism test on unchanged PR #9 SHA showed 7/7 warn findings stable; block count oscillation eliminated; llm_team quality scores 8-9 on context-injected extract runs. See PR #9 for full run-by-run commit history.
This commit was merged in pull request #9.
This commit is contained in:
+409
-78
@@ -14,19 +14,54 @@
|
||||
|
||||
import type { Claim, Finding } from "../types.ts";
|
||||
import { Glob } from "bun";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { readFile, mkdir, appendFile } from "node:fs/promises";
|
||||
import { extractFacts } from "../fact_extractor.ts";
|
||||
|
||||
const GATEWAY = process.env.LH_GATEWAY_URL ?? "http://localhost:3100";
|
||||
const MODEL = process.env.LH_AUDITOR_REVIEW_MODEL ?? "gpt-oss:120b";
|
||||
// Tie-breaker for claims where the N=3 consensus produces a 1-1-1
|
||||
// split (genuinely borderline). Different architecture from the
|
||||
// primary reviewer (gpt-oss) so the tie-break isn't correlated with
|
||||
// the original disagreement. qwen3-coder:480b is a newer coding
|
||||
// specialist at 480B params, well-suited to PR-diff claim verification
|
||||
// and distinct in training lineage from gpt-oss.
|
||||
const TIEBREAKER_MODEL = process.env.LH_AUDITOR_TIEBREAKER_MODEL ?? "qwen3-coder:480b";
|
||||
const N_CONSENSUS = Number(process.env.LH_AUDITOR_CONSENSUS_N ?? 3);
|
||||
const AUDIT_DISCREPANCIES_JSONL = "/home/profit/lakehouse/data/_kb/audit_discrepancies.jsonl";
|
||||
// 40KB comfortably fits gpt-oss:120b's context. PR #1 (~39KB) was
|
||||
// previously truncated at 15KB causing the reviewer to miss later
|
||||
// files (gitea.ts, policy.ts) and flag "no Gitea client present" as a
|
||||
// block finding when the file was simply outside the truncation window.
|
||||
//
|
||||
// Above this threshold we curate via tree-split rather than truncate,
|
||||
// following the scrum_master pattern: shard the diff, summarize each
|
||||
// shard against the claim-verification task, merge into a compact
|
||||
// scratchpad, then ask the cloud to verify claims against the
|
||||
// scratchpad. This gives the cloud full-PR fidelity without bursting
|
||||
// its context window (observed failure mode: empty response or
|
||||
// unparseable output when prompt exceeds model's comfortable range).
|
||||
const MAX_DIFF_CHARS = 40000;
|
||||
// Tree-split kicks in above this. 30KB is below MAX_DIFF_CHARS so we
|
||||
// curate BEFORE truncation would happen — never lose signal to a hard
|
||||
// cut. Shard size is chosen so ~10 shards cover PR #8-size diffs in a
|
||||
// reasonable round-trip budget.
|
||||
const CURATION_THRESHOLD = 30000;
|
||||
const DIFF_SHARD_SIZE = 4500;
|
||||
const CALL_TIMEOUT_MS = 120_000;
|
||||
const REPO_ROOT = "/home/profit/lakehouse";
|
||||
|
||||
export async function runInferenceCheck(claims: Claim[], diff: string): Promise<Finding[]> {
|
||||
export interface InferenceContext {
|
||||
pr_number: number;
|
||||
head_sha: string;
|
||||
}
|
||||
|
||||
const AUDIT_FACTS_JSONL = "/home/profit/lakehouse/data/_kb/audit_facts.jsonl";
|
||||
|
||||
export async function runInferenceCheck(
|
||||
claims: Claim[],
|
||||
diff: string,
|
||||
ctx?: InferenceContext,
|
||||
): Promise<Finding[]> {
|
||||
if (claims.length === 0) {
|
||||
return [{
|
||||
check: "inference",
|
||||
@@ -51,9 +86,26 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
}];
|
||||
}
|
||||
|
||||
const truncated = diff.length > MAX_DIFF_CHARS
|
||||
? diff.slice(0, MAX_DIFF_CHARS) + `\n...[${diff.length - MAX_DIFF_CHARS} more chars truncated]`
|
||||
: diff;
|
||||
// Diff source for the cloud prompt — either the raw diff (small
|
||||
// enough to fit), or a tree-split scratchpad (curation layer). We
|
||||
// prefer curation to truncation: truncation silently drops files
|
||||
// past the window; curation summarizes them so the cloud still sees
|
||||
// what changed, just densified.
|
||||
let diffForPrompt: string;
|
||||
let curationNote = "";
|
||||
if (diff.length > CURATION_THRESHOLD) {
|
||||
const ts = await treeSplitDiff(diff, verifiable);
|
||||
diffForPrompt = ts.scratchpad;
|
||||
curationNote = ` (curated: ${diff.length} chars → ${ts.shards} shards → scratchpad ${ts.scratchpad.length} chars)`;
|
||||
} else {
|
||||
diffForPrompt = diff;
|
||||
}
|
||||
// Belt-and-suspenders truncation — even a tree-split scratchpad
|
||||
// shouldn't exceed MAX_DIFF_CHARS in practice, but guard anyway so
|
||||
// pathological inputs can't burst the prompt.
|
||||
const truncated = diffForPrompt.length > MAX_DIFF_CHARS
|
||||
? diffForPrompt.slice(0, MAX_DIFF_CHARS) + `\n...[${diffForPrompt.length - MAX_DIFF_CHARS} more chars truncated]`
|
||||
: diffForPrompt;
|
||||
|
||||
// Build the reviewer prompt in the same shape as run_codereview's
|
||||
// review stage (llm_team_ui.py:10950), adapted for claim verification:
|
||||
@@ -61,6 +113,30 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
// "Code: ..."
|
||||
// "Review: bugs/security/perf/style/edge. Provide corrected code."
|
||||
// We add: claim list upfront + ask for structured JSON verdict.
|
||||
//
|
||||
// When the diff was curated (tree-split scratchpad), we add an
|
||||
// explicit anti-false-positive instruction: the scratchpad is a
|
||||
// distillation, not the full source, so absence-from-scratchpad is
|
||||
// NOT evidence of absence-from-diff. Mirrors the fix we made in
|
||||
// scrum_master's review prompt for the same class of error.
|
||||
const isCurated = curationNote.length > 0;
|
||||
const curationGuard = isCurated
|
||||
? [
|
||||
"",
|
||||
"CRITICAL: the 'Diff' below is a curated multi-shard scratchpad,",
|
||||
"NOT the full raw diff. The scratchpad distills each shard down",
|
||||
"to facts useful for claim verification and drops the rest.",
|
||||
"DO NOT flag a function/field/feature as 'missing' or 'not",
|
||||
"implemented' based solely on its absence from the scratchpad —",
|
||||
"absence in a distillation is NOT evidence of absence in the",
|
||||
"actual diff. Only judge a claim NOT BACKED when the scratchpad",
|
||||
"DIRECTLY contradicts it (e.g. scratchpad shows the function was",
|
||||
"added empty, or shows the claimed code path is a stub).",
|
||||
"Skip the unflagged_gaps section entirely when operating on a",
|
||||
"curated scratchpad — you can't reliably detect gaps from a",
|
||||
"distillation, and false positives there are worse than misses.",
|
||||
].join("\n")
|
||||
: "";
|
||||
const systemMsg = [
|
||||
"You review pull-request diffs against the author's own ship-claims.",
|
||||
"For each claim, decide: is it backed by actual code in the diff, or is",
|
||||
@@ -74,6 +150,7 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
" - the claim claims integration but the integration point is a stub",
|
||||
" - the diff contains unimplemented!() / todo!() / TODO comments",
|
||||
" - the claim says 'works end-to-end' but the diff has no end-to-end test",
|
||||
curationGuard,
|
||||
"",
|
||||
"Respond with strict JSON only. No prose before or after. Shape:",
|
||||
"{",
|
||||
@@ -100,94 +177,131 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
`Strict JSON only, matching the shape described. No prose outside JSON.`,
|
||||
].join("\n");
|
||||
|
||||
let resp: Response;
|
||||
try {
|
||||
resp = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
provider: "ollama_cloud",
|
||||
model: MODEL,
|
||||
messages: [
|
||||
{ role: "system", content: systemMsg },
|
||||
{ role: "user", content: userMsg },
|
||||
],
|
||||
// Deterministic classification — temp=0 is greedy-sample, so
|
||||
// identical input yields identical output on the same model
|
||||
// version. This kills the signature creep we observed in the
|
||||
// 9-run empirical test (sig_count 16→27 from cloud phrasing
|
||||
// variance at temp=0.2).
|
||||
//
|
||||
// IMPORTANT: keep think=true. gpt-oss:120b is a reasoning
|
||||
// model; setting think=false caused it to return empty content
|
||||
// on large prompts (observed during Level 1 validation: 13421
|
||||
// tokens used, empty content returned). The reasoning trace is
|
||||
// variable prose, but at temp=0 the FINAL classification is
|
||||
// still deterministic because greedy sampling converges to
|
||||
// the same conclusion from the same starting state.
|
||||
max_tokens: 3000,
|
||||
temperature: 0,
|
||||
think: true,
|
||||
}),
|
||||
signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
|
||||
});
|
||||
} catch (e) {
|
||||
// Cloud unreachable → soft-fail. Don't block a PR because the
|
||||
// reviewer model is down. Static + dynamic + kb still run.
|
||||
// N=3 consensus — run the primary reviewer in parallel, collect
|
||||
// all three parsed responses, majority-vote per claim. Parallel
|
||||
// (Promise.all) because each call is ~20-30s and they're independent;
|
||||
// wall-clock stays ~same as single call, cost 3x tokens. Empirical
|
||||
// justification: in 3-run determinism tests, 7/8 findings were
|
||||
// stable but 1 flipped across runs — majority vote stabilizes the
|
||||
// flipping class without losing the stable signal.
|
||||
const primaryRuns = await Promise.all(
|
||||
Array.from({ length: N_CONSENSUS }, () =>
|
||||
runCloudInference(systemMsg, userMsg, MODEL)),
|
||||
);
|
||||
|
||||
const parsedRuns = primaryRuns.filter(r => r.parsed !== null);
|
||||
if (parsedRuns.length === 0) {
|
||||
// All N calls failed. Surface the first-run diagnostic so the
|
||||
// operator sees *why* (unreachable / non-200 / unparseable).
|
||||
const first = primaryRuns[0];
|
||||
return [{
|
||||
check: "inference",
|
||||
severity: "info",
|
||||
summary: "cloud inference unreachable — skipped",
|
||||
evidence: [`fetch failed: ${(e as Error).message.slice(0, 180)}`],
|
||||
}];
|
||||
}
|
||||
|
||||
if (!resp.ok) {
|
||||
return [{
|
||||
check: "inference",
|
||||
severity: "info",
|
||||
summary: `cloud inference returned ${resp.status} — skipped`,
|
||||
evidence: [`body: ${(await resp.text()).slice(0, 200)}`],
|
||||
}];
|
||||
}
|
||||
|
||||
const body: any = await resp.json();
|
||||
const content: string = body?.choices?.[0]?.message?.content ?? "";
|
||||
const usage = body?.usage ?? {};
|
||||
|
||||
const parsed = extractJson(content);
|
||||
if (!parsed) {
|
||||
return [{
|
||||
check: "inference",
|
||||
severity: "info",
|
||||
summary: "cloud returned unparseable output — skipped",
|
||||
summary: `cloud inference all ${N_CONSENSUS} consensus runs failed — ${first.error ?? "unknown"}`,
|
||||
evidence: [
|
||||
`head: ${content.slice(0, 200)}`,
|
||||
`tokens: ${usage.total_tokens ?? "?"}`,
|
||||
`first-run diagnostic: ${first.diagnostic ?? "(none)"}`,
|
||||
`successful runs: 0 / ${N_CONSENSUS}`,
|
||||
],
|
||||
}];
|
||||
}
|
||||
|
||||
// Aggregate votes per claim_idx.
|
||||
interface Votes { trues: number; falses: number; evidences: string[] }
|
||||
const votesByClaim = new Map<number, Votes>();
|
||||
const unflaggedByRun: any[][] = [];
|
||||
let totalTokens = 0;
|
||||
for (const run of parsedRuns) {
|
||||
totalTokens += run.tokens;
|
||||
unflaggedByRun.push(Array.isArray(run.parsed?.unflagged_gaps) ? run.parsed.unflagged_gaps : []);
|
||||
for (const v of run.parsed?.claim_verdicts ?? []) {
|
||||
const idx = Number(v?.claim_idx);
|
||||
if (!Number.isFinite(idx)) continue;
|
||||
const rec = votesByClaim.get(idx) ?? { trues: 0, falses: 0, evidences: [] };
|
||||
if (v.backed === false) {
|
||||
rec.falses++;
|
||||
rec.evidences.push(String(v.evidence ?? ""));
|
||||
} else if (v.backed === true) {
|
||||
rec.trues++;
|
||||
}
|
||||
votesByClaim.set(idx, rec);
|
||||
}
|
||||
}
|
||||
|
||||
const findings: Finding[] = [];
|
||||
|
||||
// One summary info finding so the verdict layer knows the check ran.
|
||||
// Summary finding so the verdict layer knows the check ran.
|
||||
findings.push({
|
||||
check: "inference",
|
||||
severity: "info",
|
||||
summary: `cloud review completed (model=${MODEL}, tokens=${usage.total_tokens ?? "?"})`,
|
||||
summary: `cloud review completed (model=${MODEL}, consensus=${parsedRuns.length}/${N_CONSENSUS}, tokens=${totalTokens})${curationNote}`,
|
||||
evidence: [
|
||||
`claim_verdicts: ${parsed.claim_verdicts?.length ?? 0}, unflagged_gaps: ${parsed.unflagged_gaps?.length ?? 0}`,
|
||||
`claims voted: ${votesByClaim.size}`,
|
||||
`parsed runs: ${parsedRuns.length} / ${N_CONSENSUS}`,
|
||||
],
|
||||
});
|
||||
|
||||
for (const v of parsed.claim_verdicts ?? []) {
|
||||
if (v?.backed === false) {
|
||||
const idx = typeof v.claim_idx === "number" ? v.claim_idx : -1;
|
||||
// Indices point at the verifiable[] list we sent the cloud,
|
||||
// not the full claims[] list. Translate back.
|
||||
const claim = verifiable[idx];
|
||||
if (!claim) continue;
|
||||
// Strong+unbacked = BLOCK. That's the whole point of the auditor.
|
||||
// Per-claim majority vote; tie-break if no majority.
|
||||
const discrepancies: Array<{
|
||||
claim_idx: number;
|
||||
claim_text: string;
|
||||
votes: { trues: number; falses: number };
|
||||
resolution: "majority_backed" | "majority_not_backed" | "tiebreaker_backed" | "tiebreaker_not_backed" | "unresolved";
|
||||
tiebreaker_model?: string;
|
||||
}> = [];
|
||||
|
||||
for (const [idx, votes] of votesByClaim) {
|
||||
const claim = verifiable[idx];
|
||||
if (!claim) continue;
|
||||
const totalVotes = votes.trues + votes.falses;
|
||||
let notBacked: boolean | null = null;
|
||||
let resolution: typeof discrepancies[number]["resolution"] = "majority_backed";
|
||||
let evidenceText = "";
|
||||
let tbModel: string | undefined;
|
||||
|
||||
if (votes.falses > votes.trues) {
|
||||
notBacked = true;
|
||||
resolution = "majority_not_backed";
|
||||
evidenceText = votes.evidences[0] ?? "(no reason given)";
|
||||
} else if (votes.trues > votes.falses) {
|
||||
notBacked = false;
|
||||
resolution = "majority_backed";
|
||||
} else {
|
||||
// Tie. Run tie-breaker with a different-architecture model.
|
||||
const tb = await runCloudInference(systemMsg, userMsg, TIEBREAKER_MODEL);
|
||||
if (tb.parsed) {
|
||||
const tv = (tb.parsed.claim_verdicts ?? []).find((v: any) => Number(v?.claim_idx) === idx);
|
||||
if (tv?.backed === false) {
|
||||
notBacked = true;
|
||||
resolution = "tiebreaker_not_backed";
|
||||
evidenceText = `(tie-breaker ${TIEBREAKER_MODEL}) ${String(tv.evidence ?? "")}`;
|
||||
tbModel = TIEBREAKER_MODEL;
|
||||
} else if (tv?.backed === true) {
|
||||
notBacked = false;
|
||||
resolution = "tiebreaker_backed";
|
||||
tbModel = TIEBREAKER_MODEL;
|
||||
} else {
|
||||
resolution = "unresolved";
|
||||
}
|
||||
} else {
|
||||
resolution = "unresolved";
|
||||
}
|
||||
}
|
||||
|
||||
// Log every case where the N runs disagreed — discrepancies are
|
||||
// signal, not noise. Separate from audit_lessons.jsonl because
|
||||
// they're about the *auditor's* quality, not the PR's quality.
|
||||
const disagreed = totalVotes >= 2 && votes.trues > 0 && votes.falses > 0;
|
||||
if (disagreed || resolution.startsWith("tiebreaker") || resolution === "unresolved") {
|
||||
discrepancies.push({
|
||||
claim_idx: idx,
|
||||
claim_text: claim.text,
|
||||
votes: { trues: votes.trues, falses: votes.falses },
|
||||
resolution,
|
||||
tiebreaker_model: tbModel,
|
||||
});
|
||||
}
|
||||
|
||||
if (notBacked === true) {
|
||||
const sev: Finding["severity"] = claim.strength === "strong" ? "block"
|
||||
: claim.strength === "moderate" ? "warn"
|
||||
: "info";
|
||||
@@ -198,13 +312,45 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
summary: `cloud: claim not backed — "${claim.text.slice(0, 100)}"`,
|
||||
evidence: [
|
||||
`at ${claim.location}`,
|
||||
`cloud reason: ${String(v.evidence ?? "no reason given").slice(0, 200)}`,
|
||||
`consensus: ${votes.falses}/${totalVotes} not-backed (resolution: ${resolution})`,
|
||||
`cloud reason: ${evidenceText.slice(0, 200)}`,
|
||||
],
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
for (const g of parsed.unflagged_gaps ?? []) {
|
||||
// Persist discrepancies so we can measure consensus drift over time.
|
||||
if (discrepancies.length > 0 && ctx) {
|
||||
persistDiscrepancies(ctx, discrepancies).catch(e =>
|
||||
console.error(`[inference] discrepancy log failed: ${(e as Error).message}`));
|
||||
}
|
||||
|
||||
// Use first run's parsed for downstream unflagged_gaps processing.
|
||||
const parsed = parsedRuns[0].parsed;
|
||||
|
||||
// Route the curated scratchpad through llm_team's extract-facts
|
||||
// pipeline when we have (a) a curated scratchpad (best signal about
|
||||
// what the PR actually changed) and (b) PR context to scope facts.
|
||||
// AWAITED (not fire-and-forget) so CLI callers like audit_one.ts
|
||||
// don't exit before extraction lands; the systemd poller has plenty
|
||||
// of headroom (90s cycle vs ~15s extraction). A failure inside
|
||||
// extractAndPersistFacts is caught + logged but never throws.
|
||||
if (isCurated && ctx && process.env.LH_AUDITOR_SKIP_EXTRACT !== "1") {
|
||||
try {
|
||||
await extractAndPersistFacts(diffForPrompt, ctx);
|
||||
} catch (e) {
|
||||
console.error(`[inference] fact extraction failed: ${(e as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Belt-and-suspenders: when operating on a curated scratchpad, drop
|
||||
// the unflagged_gaps section entirely. The distillation can't
|
||||
// reliably ground gap-detection, and false positives are worse than
|
||||
// misses for this signal class. The systemMsg already asks the
|
||||
// cloud to skip this section when curated — but the model may still
|
||||
// emit it, so we filter here too.
|
||||
const gapsToEmit = isCurated ? [] : (parsed.unflagged_gaps ?? []);
|
||||
for (const g of gapsToEmit) {
|
||||
const summary = String(g?.summary ?? "?");
|
||||
const location = String(g?.location ?? "?");
|
||||
// False-positive guard — when the cloud says "X not defined in this
|
||||
@@ -248,6 +394,191 @@ export async function runInferenceCheck(claims: Claim[], diff: string): Promise<
|
||||
return findings;
|
||||
}
|
||||
|
||||
// Single cloud call — the consensus loop calls this N times in
|
||||
// parallel. Returns the parsed JSON shape + token usage + any error
|
||||
// diagnostic. NEVER throws; the consensus aggregator handles partial
|
||||
// failures by dropping non-parsed runs from the vote.
|
||||
interface CloudRunResult {
|
||||
parsed: any | null;
|
||||
tokens: number;
|
||||
error?: string; // "unreachable" | "non_200" | "unparseable"
|
||||
diagnostic?: string; // first 200 chars for debugging
|
||||
model: string;
|
||||
}
|
||||
|
||||
async function runCloudInference(systemMsg: string, userMsg: string, model: string): Promise<CloudRunResult> {
|
||||
let resp: Response;
|
||||
try {
|
||||
resp = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
provider: "ollama_cloud",
|
||||
model,
|
||||
messages: [
|
||||
{ role: "system", content: systemMsg },
|
||||
{ role: "user", content: userMsg },
|
||||
],
|
||||
// temp=0 (greedy) + think=true. think=true is required for
|
||||
// gpt-oss:120b — without it the model returns empty content
|
||||
// on large prompts. Variance from the think trace is observed
|
||||
// in practice, which is why we use N=3 consensus, not single-
|
||||
// call determinism.
|
||||
max_tokens: 3000,
|
||||
temperature: 0,
|
||||
think: true,
|
||||
}),
|
||||
signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
|
||||
});
|
||||
} catch (e) {
|
||||
return { parsed: null, tokens: 0, error: "unreachable", diagnostic: (e as Error).message.slice(0, 200), model };
|
||||
}
|
||||
if (!resp.ok) {
|
||||
return { parsed: null, tokens: 0, error: "non_200", diagnostic: `${resp.status}: ${(await resp.text()).slice(0, 160)}`, model };
|
||||
}
|
||||
let body: any;
|
||||
try { body = await resp.json(); }
|
||||
catch (e) { return { parsed: null, tokens: 0, error: "unparseable", diagnostic: (e as Error).message, model }; }
|
||||
const content: string = body?.choices?.[0]?.message?.content ?? "";
|
||||
const tokens: number = body?.usage?.total_tokens ?? 0;
|
||||
const parsed = extractJson(content);
|
||||
if (!parsed) {
|
||||
return { parsed: null, tokens, error: "unparseable", diagnostic: content.slice(0, 200), model };
|
||||
}
|
||||
return { parsed, tokens, model };
|
||||
}
|
||||
|
||||
async function persistDiscrepancies(ctx: InferenceContext, discrepancies: any[]): Promise<void> {
|
||||
await mkdir("/home/profit/lakehouse/data/_kb", { recursive: true });
|
||||
const rows = discrepancies.map(d => JSON.stringify({
|
||||
pr_number: ctx.pr_number,
|
||||
head_sha: ctx.head_sha,
|
||||
logged_at: new Date().toISOString(),
|
||||
...d,
|
||||
}));
|
||||
await appendFile(AUDIT_DISCREPANCIES_JSONL, rows.join("\n") + "\n");
|
||||
}
|
||||
|
||||
// Extract structured knowledge from the curated scratchpad and append
|
||||
// to data/_kb/audit_facts.jsonl — one row per extract run, keyed by
|
||||
// PR number + head SHA for scope tracking. kb_query tails this next
|
||||
// audit to surface recurring entities/relationships across PRs.
|
||||
async function extractAndPersistFacts(scratchpad: string, ctx: InferenceContext): Promise<void> {
|
||||
const ex = await extractFacts(scratchpad);
|
||||
if (ex.error && ex.entities.length === 0 && ex.facts.length === 0) {
|
||||
// Full failure — log but don't write an empty row.
|
||||
console.error(`[inference] extractFacts skipped row: ${ex.error}`);
|
||||
return;
|
||||
}
|
||||
const row = {
|
||||
pr_number: ctx.pr_number,
|
||||
head_sha: ctx.head_sha,
|
||||
extracted_at: ex.extracted_at,
|
||||
extractor: ex.extractor_model,
|
||||
verifier: ex.verifier_model,
|
||||
llm_team_run_id: ex.llm_team_run_id ?? null,
|
||||
facts: ex.facts,
|
||||
entities: ex.entities,
|
||||
relationships: ex.relationships,
|
||||
verification_preview: ex.verification.slice(0, 400),
|
||||
verifier_verdicts: ex.verifier_verdicts,
|
||||
facts_dropped_by_verifier: ex.facts_dropped_by_verifier ?? 0,
|
||||
schema_version: 2,
|
||||
source: "audit_inference",
|
||||
};
|
||||
await mkdir("/home/profit/lakehouse/data/_kb", { recursive: true });
|
||||
await appendFile(AUDIT_FACTS_JSONL, JSON.stringify(row) + "\n");
|
||||
}
|
||||
|
||||
// Curation via tree-split — ports the scrum_master pattern into the
|
||||
// inference check. Shards the raw diff into DIFF_SHARD_SIZE chunks,
|
||||
// summarizes each shard *against the claim-verification task* so the
|
||||
// summary preserves exactly what the cloud needs to judge claims
|
||||
// (function signatures, struct fields, deletions, new files), drops
|
||||
// everything else. Merges into a compact scratchpad.
|
||||
//
|
||||
// Cost: N cloud calls for the shard summaries + 1 cloud call for the
|
||||
// final verification = N+1 calls instead of 1. Mitigation: shards run
|
||||
// serially (not parallel) to keep gateway load bounded; summary calls
|
||||
// use max_tokens=400 so they're fast (~2s each on gpt-oss:120b).
|
||||
//
|
||||
// Determinism: each shard summary call uses temp=0 + think=true (same
|
||||
// as the top-level inference call), so identical input yields
|
||||
// identical scratchpad. The final verification call then sees a
|
||||
// stable scratchpad, giving stable verdicts.
|
||||
async function treeSplitDiff(
|
||||
fullDiff: string,
|
||||
claims: Claim[],
|
||||
): Promise<{ scratchpad: string; shards: number }> {
|
||||
const shards: Array<{ from: number; to: number; text: string }> = [];
|
||||
for (let i = 0; i < fullDiff.length; i += DIFF_SHARD_SIZE) {
|
||||
const end = Math.min(i + DIFF_SHARD_SIZE, fullDiff.length);
|
||||
shards.push({ from: i, to: end, text: fullDiff.slice(i, end) });
|
||||
}
|
||||
// Curate the claim list into a short form the summary prompt can
|
||||
// use to bias extraction toward relevant facts.
|
||||
const claimDigest = claims.map((c, i) =>
|
||||
`${i}. [${c.strength}] "${c.text.slice(0, 100)}"`
|
||||
).join("\n");
|
||||
|
||||
let scratchpad = "";
|
||||
for (const [si, shard] of shards.entries()) {
|
||||
const prompt = [
|
||||
`You are summarizing shard ${si + 1}/${shards.length} (chars ${shard.from}..${shard.to}) of a PR diff.`,
|
||||
`The downstream task will verify these ship-claims against the full-PR summary. Extract ONLY facts that could confirm or refute these claims:`,
|
||||
"",
|
||||
claimDigest,
|
||||
"",
|
||||
"Extract: new function/method signatures, struct fields, deletions, new files, wiring (function X calls Y), absence-of-implementation markers, TODO comments on added lines.",
|
||||
"Skip: comment-only edits, whitespace, import reordering, unrelated cosmetic changes.",
|
||||
"",
|
||||
"─────── shard diff ───────",
|
||||
shard.text,
|
||||
"─────── end shard ───────",
|
||||
"",
|
||||
"Output: up to 180 words of facts in bullet form. No prose preamble, no claim verdicts (that's for the downstream step).",
|
||||
].join("\n");
|
||||
|
||||
const r = await callCloud(prompt, 400);
|
||||
if (r.content) {
|
||||
scratchpad += `\n--- shard ${si + 1} (chars ${shard.from}..${shard.to}) ---\n${r.content.trim()}\n`;
|
||||
}
|
||||
}
|
||||
return { scratchpad: scratchpad.trim(), shards: shards.length };
|
||||
}
|
||||
|
||||
// Minimal cloud caller used only by treeSplitDiff — same gateway +
|
||||
// model as the top-level call, but think=false. Shards are small
|
||||
// (≤DIFF_SHARD_SIZE ~4500 chars) and the task is pure fact
|
||||
// extraction, not reasoning. think=true on the shards introduced
|
||||
// variance in reasoning traces that compounded across 23 calls into
|
||||
// a non-deterministic scratchpad (observed during curation
|
||||
// validation: same-SHA runs produced 5/7/8 final findings).
|
||||
// think=false on small prompts is stable — only breaks at the main
|
||||
// call's 10K+ prompt size, which keeps think=true.
|
||||
async function callCloud(prompt: string, maxTokens: number): Promise<{ content: string }> {
|
||||
try {
|
||||
const r = await fetch(`${GATEWAY}/v1/chat`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
provider: "ollama_cloud",
|
||||
model: MODEL,
|
||||
messages: [{ role: "user", content: prompt }],
|
||||
max_tokens: maxTokens,
|
||||
temperature: 0,
|
||||
think: false,
|
||||
}),
|
||||
signal: AbortSignal.timeout(CALL_TIMEOUT_MS),
|
||||
});
|
||||
if (!r.ok) return { content: "" };
|
||||
const j: any = await r.json();
|
||||
return { content: j?.choices?.[0]?.message?.content ?? "" };
|
||||
} catch {
|
||||
return { content: "" };
|
||||
}
|
||||
}
|
||||
|
||||
// Pull out plausible code-symbol names from a summary string.
|
||||
// Matches:
|
||||
// - identifier with backticks: `foo_bar`
|
||||
|
||||
@@ -25,6 +25,7 @@ const OBSERVER_OPS = "/home/profit/lakehouse/data/_observer/ops.jsonl";
|
||||
const BOT_CYCLES_DIR = "/home/profit/lakehouse/data/_bot/cycles";
|
||||
const SCRUM_REVIEWS_JSONL = "/home/profit/lakehouse/data/_kb/scrum_reviews.jsonl";
|
||||
const AUDIT_LESSONS_JSONL = "/home/profit/lakehouse/data/_kb/audit_lessons.jsonl";
|
||||
const AUDIT_FACTS_JSONL = "/home/profit/lakehouse/data/_kb/audit_facts.jsonl";
|
||||
const TAIL_LINES = 500;
|
||||
const MAX_BOT_CYCLE_FILES = 30;
|
||||
|
||||
@@ -61,6 +62,14 @@ export async function runKbCheck(claims: Claim[], prFiles: string[] = []): Promi
|
||||
findings.push(...scrumFindings);
|
||||
}
|
||||
|
||||
// 6b. Audit-facts (llm_team extract pipeline output) — surface
|
||||
// entities that recur across multiple PRs. These are the
|
||||
// "core system entities" accumulating in the knowledge base;
|
||||
// showing them as info on future audits gives reviewers
|
||||
// architectural context the raw diff doesn't convey.
|
||||
const factFindings = await checkAuditFacts();
|
||||
findings.push(...factFindings);
|
||||
|
||||
// 6. Audit-lessons feedback loop — summarize the top recurring
|
||||
// patterns from prior audits' block/warn findings. If the same
|
||||
// pattern signature has fired 3+ times across prior audits,
|
||||
@@ -207,6 +216,99 @@ function observerBySource(ops: any[]): string {
|
||||
return Object.entries(c).sort((a, b) => b[1] - a[1]).map(([k, v]) => `${k}=${v}`).join(", ") || "empty";
|
||||
}
|
||||
|
||||
// Audit-facts — reads data/_kb/audit_facts.jsonl (populated by every
|
||||
// curated inference run via llm_team's extract pipeline). Each row
|
||||
// has arrays: facts, entities, relationships. We explode entities and
|
||||
// aggregate them across PRs using kb_index. An entity seen in 3+ PRs
|
||||
// is a "core system entity" — we surface the top N as info context.
|
||||
//
|
||||
// Filters out short names (<3 chars, likely qwen2.5 truncation
|
||||
// artifacts) and generic types ("string", "number") that would
|
||||
// otherwise dominate the ranking.
|
||||
const ENTITY_NAME_MIN_LEN = 3;
|
||||
const GENERIC_ENTITY_NAMES = new Set([
|
||||
"string", "number", "boolean", "any", "void", "unknown", "never",
|
||||
"object", "array", "function", "const", "let", "var", "true", "false",
|
||||
"null", "undefined", "promise", "map", "set", "record",
|
||||
]);
|
||||
|
||||
async function checkAuditFacts(): Promise<Finding[]> {
|
||||
// Read raw rows — each row has multiple entities, so we can't just
|
||||
// use aggregate() directly (it's one-signature-per-row). Explode
|
||||
// entities into (row, entity) pairs, then aggregate by entity name.
|
||||
let raw: string;
|
||||
try { raw = await (await import("node:fs/promises")).readFile(AUDIT_FACTS_JSONL, "utf8"); }
|
||||
catch { return []; }
|
||||
const lines = raw.split("\n").filter(l => l.length > 0);
|
||||
if (lines.length === 0) return [];
|
||||
|
||||
interface EntityRow { entity_key: string; pr_number: number; type: string; name: string; description: string }
|
||||
const entityRows: EntityRow[] = [];
|
||||
for (const line of lines.slice(-TAIL_LINES * 2)) {
|
||||
let row: any;
|
||||
try { row = JSON.parse(line); } catch { continue; }
|
||||
const prNum = Number(row?.pr_number);
|
||||
if (!Number.isFinite(prNum)) continue;
|
||||
for (const e of Array.isArray(row?.entities) ? row.entities : []) {
|
||||
const name = String(e?.name ?? "").trim();
|
||||
if (name.length < ENTITY_NAME_MIN_LEN) continue;
|
||||
if (GENERIC_ENTITY_NAMES.has(name.toLowerCase())) continue;
|
||||
entityRows.push({
|
||||
entity_key: name.toLowerCase(),
|
||||
pr_number: prNum,
|
||||
type: String(e?.type ?? "?"),
|
||||
name,
|
||||
description: String(e?.description ?? "").slice(0, 160),
|
||||
});
|
||||
}
|
||||
}
|
||||
if (entityRows.length === 0) return [];
|
||||
|
||||
// Aggregate manually — one key per entity name, distinct_scopes by PR.
|
||||
type Agg = { count: number; scopes: Set<number>; types: Set<string>; last_name: string; last_desc: string };
|
||||
const byEntity = new Map<string, Agg>();
|
||||
for (const r of entityRows) {
|
||||
const a = byEntity.get(r.entity_key) ?? {
|
||||
count: 0, scopes: new Set<number>(), types: new Set<string>(), last_name: "", last_desc: "",
|
||||
};
|
||||
a.count += 1;
|
||||
a.scopes.add(r.pr_number);
|
||||
a.types.add(r.type);
|
||||
a.last_name = r.name;
|
||||
a.last_desc = r.description;
|
||||
byEntity.set(r.entity_key, a);
|
||||
}
|
||||
|
||||
// Rank: require 2+ distinct PRs (same-PR entity-repeats don't count
|
||||
// as "cross-cutting"). Take the top 5 to avoid flooding the verdict.
|
||||
const ranked = Array.from(byEntity.entries())
|
||||
.filter(([_, a]) => a.scopes.size >= 2)
|
||||
.sort((a, b) => b[1].scopes.size - a[1].scopes.size || b[1].count - a[1].count)
|
||||
.slice(0, 5);
|
||||
|
||||
if (ranked.length === 0) {
|
||||
// Useful to know the KB is being populated — emit a single
|
||||
// summary so operators see fact extraction is alive.
|
||||
return [{
|
||||
check: "kb_query",
|
||||
severity: "info",
|
||||
summary: `audit_facts KB has ${entityRows.length} entity-observations across ${new Set(entityRows.map(r => r.pr_number)).size} PRs (no cross-PR recurrences yet)`,
|
||||
evidence: [`source: ${AUDIT_FACTS_JSONL}`],
|
||||
}];
|
||||
}
|
||||
|
||||
return ranked.map(([_, a]) => ({
|
||||
check: "kb_query" as const,
|
||||
severity: "info" as const,
|
||||
summary: `core entity \`${a.last_name}\` recurs in ${a.scopes.size} PRs (types: ${Array.from(a.types).join(",")})`,
|
||||
evidence: [
|
||||
`count=${a.count} distinct_PRs=${a.scopes.size}`,
|
||||
`description: ${a.last_desc.slice(0, 200)}`,
|
||||
`PRs: ${Array.from(a.scopes).sort((x, y) => x - y).join(",")}`,
|
||||
],
|
||||
}));
|
||||
}
|
||||
|
||||
// Audit-lessons — reads data/_kb/audit_lessons.jsonl (populated by
|
||||
// every audit's appendAuditLessons). Uses the shared kb_index
|
||||
// aggregator: groups by `signature`, distinct-scopes keyed by PR
|
||||
|
||||
Reference in New Issue
Block a user