Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6b131e1427 | ||
|
|
a76221578c | ||
|
|
0dec11b5ed | ||
|
|
6478ce2bfa | ||
|
|
c648a5a6da | ||
|
|
10bbd598f3 |
@@ -35,6 +35,7 @@ const DDL = [
|
||||
state TEXT NOT NULL DEFAULT 'queued',
|
||||
outcome TEXT,
|
||||
created_ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||
started_ts TEXT,
|
||||
finished_ts TEXT
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_queue_jobs_state ON queue_jobs(state)`
|
||||
@@ -44,9 +45,17 @@ export function openDb(dbPath) {
|
||||
mkdirSync(dirname(dbPath), { recursive: true });
|
||||
const db = new DatabaseSync(dbPath);
|
||||
for (const stmt of DDL) db.prepare(stmt).run();
|
||||
ensureColumn(db, 'queue_jobs', 'started_ts', 'TEXT');
|
||||
return db;
|
||||
}
|
||||
|
||||
function ensureColumn(db, table, column, type) {
|
||||
const columns = db.prepare(`PRAGMA table_info(${table})`).all();
|
||||
if (!columns.some((c) => c.name === column)) {
|
||||
db.prepare(`ALTER TABLE ${table} ADD COLUMN ${column} ${type}`).run();
|
||||
}
|
||||
}
|
||||
|
||||
export function insertReview(db, r) {
|
||||
const stmt = db.prepare(`
|
||||
INSERT INTO reviews (repo, sha, pr, event, pusher, authors, trailers, axes,
|
||||
@@ -76,6 +85,13 @@ export function enqueueJob(db, job) {
|
||||
return res.lastInsertRowid;
|
||||
}
|
||||
|
||||
export function startJob(db, id) {
|
||||
db.prepare(
|
||||
`UPDATE queue_jobs SET state = 'running',
|
||||
started_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?`
|
||||
).run(id);
|
||||
}
|
||||
|
||||
export function finishJob(db, id, outcome = null) {
|
||||
db.prepare(
|
||||
`UPDATE queue_jobs SET state = 'done', outcome = ?,
|
||||
|
||||
@@ -19,12 +19,14 @@ Output schema (every key required):
|
||||
"findings": [
|
||||
{"title": "...", "file": "path", "line": 123, "severity": "critical"|"high"|"medium"|"low",
|
||||
"confidence": "confirmed"|"plausible", "axis": "correctness"|"security"|"tests_coverage"|"design_simplicity"|"performance"|"style"|"breaking_changes"|"docs",
|
||||
"resolved_by_this_change": true|false,
|
||||
"detail": "what is wrong, concrete failure scenario, and the fix"}
|
||||
],
|
||||
"verdict": "pass"|"warn"|"fail"
|
||||
}
|
||||
|
||||
Rules:
|
||||
- A finding must describe a defect that EXISTS IN THE RESULTING CODE after this diff is applied. The pre-existing bug that this diff FIXES is not a finding — acknowledge it in the correctness rationale instead. If you still list it (e.g. for the record), you MUST set "resolved_by_this_change": true on that finding; findings the diff introduces or leaves unfixed get "resolved_by_this_change": false.
|
||||
- "confirmed" means you can trace a concrete failure from the diff alone; anything needing unseen context is "plausible".
|
||||
- severity critical/high is reserved for real correctness bugs, security holes, data loss, or breaking API changes.
|
||||
- verdict "fail" ONLY when a confirmed critical/high correctness or security finding exists; "warn" for notable but non-blocking issues; otherwise "pass".
|
||||
|
||||
@@ -105,6 +105,7 @@ export async function runReview({ cfg, gitea, adminGitea, db, job, withGlobal, l
|
||||
tenantId: cfg.tenantId, timeoutMs: cfg.routerTimeoutMs
|
||||
});
|
||||
servedModel = out.servedModel;
|
||||
assertServedModelAllowed(cfg.model, servedModel);
|
||||
const raw = extractJson(out.content);
|
||||
if (!raw) {
|
||||
log(`[${fullRepo}@${sha.slice(0, 8)}] unparseable model output (served=${servedModel}): ${JSON.stringify(out.content.slice(0, 400))}`);
|
||||
@@ -313,6 +314,17 @@ export async function fetchDiff({ gitea, job, log = () => {}, maxBytes = Infinit
|
||||
throw allDiffSourcesFailed(tried);
|
||||
}
|
||||
|
||||
export function assertServedModelAllowed(requestedModel, servedModel) {
|
||||
if (!isLocalModel(requestedModel)) return;
|
||||
if (!servedModel) return;
|
||||
if (servedModel === requestedModel) return;
|
||||
throw new Error(`local model contract violated: requested ${requestedModel}, served ${servedModel}`);
|
||||
}
|
||||
|
||||
function isLocalModel(model) {
|
||||
return /^ollama(?:-remote)?\//.test(String(model || ''));
|
||||
}
|
||||
|
||||
function allDiffSourcesFailed(tried) {
|
||||
let msg = tried.join(' | ');
|
||||
if (msg.length > 1500) msg = msg.slice(0, 1500) + ` …(+${tried.length} sources)`;
|
||||
|
||||
+4
-1
@@ -13,7 +13,10 @@ export async function chat({ routerUrl, model, messages, agentId, sessionId, ten
|
||||
// and return non-review text. The domain preflight (agent_mismatch 400)
|
||||
// still applies per agent tier; the pipeline's attempt-2 fallback agent
|
||||
// covers that.
|
||||
body: JSON.stringify({ model, messages, agentId, sessionId, tenantId, stream: false, tools: false, raw: true }),
|
||||
body: JSON.stringify({
|
||||
model, messages, agentId, sessionId, tenantId,
|
||||
stream: false, tools: false, raw: true, fallbackModels: []
|
||||
}),
|
||||
signal: AbortSignal.timeout(timeoutMs)
|
||||
});
|
||||
if (!res.ok) {
|
||||
|
||||
+11
-2
@@ -7,6 +7,10 @@ const BLOCKING_AXES = new Set(['correctness', 'security']);
|
||||
|
||||
export function isBlockingFinding(f) {
|
||||
if (!f || typeof f !== 'object') return false;
|
||||
// A finding the diff itself resolves describes the PRE-change state; it
|
||||
// cannot block the change that fixes it (models routinely narrate the fixed
|
||||
// bug as a critical/confirmed finding, which failed correct fixes).
|
||||
if (f.resolved_by_this_change === true) return false;
|
||||
const sev = String(f.severity || '').toLowerCase();
|
||||
const conf = String(f.confidence || '').toLowerCase();
|
||||
if (!BLOCKING_SEVERITIES.has(sev)) return false;
|
||||
@@ -21,12 +25,16 @@ export function deriveVerdict(review) {
|
||||
const findings = Array.isArray(review?.findings) ? review.findings : [];
|
||||
if (findings.some(isBlockingFinding)) return 'fail';
|
||||
const hasWarn = findings.some(f => {
|
||||
if (f?.resolved_by_this_change === true) return false;
|
||||
const sev = String(f?.severity || '').toLowerCase();
|
||||
return sev === 'critical' || sev === 'high' || sev === 'medium';
|
||||
});
|
||||
if (hasWarn) return 'warn';
|
||||
// trust the model's verdict if it said warn without matching findings
|
||||
if (review?.verdict === 'warn') return 'warn';
|
||||
// Trust the model's verdict if it said warn without matching findings —
|
||||
// unless every finding it gave is resolved_by_this_change, in which case its
|
||||
// warn is about the pre-change state and the resulting code is clean.
|
||||
const allResolved = findings.length > 0 && findings.every(f => f?.resolved_by_this_change === true);
|
||||
if (review?.verdict === 'warn' && !allResolved) return 'warn';
|
||||
return 'pass';
|
||||
}
|
||||
|
||||
@@ -49,6 +57,7 @@ export function normalizeReview(raw) {
|
||||
severity: pick(String(f?.severity || '').toLowerCase(), ['critical', 'high', 'medium', 'low'], 'low'),
|
||||
confidence: pick(String(f?.confidence || '').toLowerCase(), ['confirmed', 'plausible'], 'plausible'),
|
||||
axis: f?.axis ? String(f.axis).toLowerCase() : null,
|
||||
resolved_by_this_change: f?.resolved_by_this_change === true,
|
||||
detail: String(f?.detail ?? '').slice(0, 2000)
|
||||
}));
|
||||
const review = {
|
||||
|
||||
+5
-2
@@ -1,6 +1,6 @@
|
||||
import { createServer } from 'node:http';
|
||||
import { loadConfig } from './config.js';
|
||||
import { openDb, listReviews, enqueueJob, finishJob, recoverPendingJobs } from './lib/db.js';
|
||||
import { openDb, listReviews, enqueueJob, startJob, finishJob, recoverPendingJobs } from './lib/db.js';
|
||||
import { GiteaClient } from './lib/gitea.js';
|
||||
import { ReviewQueue } from './lib/queue.js';
|
||||
import { verifySignature, jobFromWebhook } from './lib/webhook.js';
|
||||
@@ -34,7 +34,10 @@ const queue = new ReviewQueue();
|
||||
// whatever the outcome (runReview itself posts a final or fail-open status).
|
||||
function dispatchJob(queueRowId, job) {
|
||||
return queue.enqueue(`${job.owner}/${job.repo}`, (withGlobal) =>
|
||||
runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log })
|
||||
{
|
||||
startJob(db, queueRowId);
|
||||
return runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log });
|
||||
}
|
||||
).then(
|
||||
(r) => finishJob(db, queueRowId, r?.ok ? (r.verdict || 'ok') : `error:${String(r?.error || '').slice(0, 200)}`),
|
||||
(e) => {
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { assertServedModelAllowed } from '../src/lib/review.js';
|
||||
|
||||
test('local model reviews accept the exact served local model', () => {
|
||||
assert.doesNotThrow(() =>
|
||||
assertServedModelAllowed('ollama-remote/qwen2.5-coder:7b', 'ollama-remote/qwen2.5-coder:7b'));
|
||||
});
|
||||
|
||||
test('local model reviews fail on cloud fallback drift', () => {
|
||||
assert.throws(
|
||||
() => assertServedModelAllowed('ollama-remote/qwen2.5-coder:7b', 'google/gemini-2.5-flash'),
|
||||
/local model contract violated/
|
||||
);
|
||||
});
|
||||
|
||||
test('cloud model reviews allow provider fallback reporting', () => {
|
||||
assert.doesNotThrow(() =>
|
||||
assertServedModelAllowed('anthropic/claude-sonnet-4-6', 'google/gemini-2.5-flash'));
|
||||
});
|
||||
|
||||
test('missing served model metadata does not fail local requests', () => {
|
||||
assert.doesNotThrow(() =>
|
||||
assertServedModelAllowed('ollama-remote/qwen2.5-coder:7b', null));
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { openDb, enqueueJob, finishJob, recoverPendingJobs } from '../src/lib/db.js';
|
||||
import { openDb, enqueueJob, startJob, finishJob, recoverPendingJobs } from '../src/lib/db.js';
|
||||
|
||||
const JOB = (over = {}) => ({
|
||||
event: 'push', owner: 'nirpa', repo: 'demo', sha: 'a'.repeat(40),
|
||||
@@ -32,6 +32,19 @@ test('finished jobs are not recovered again', () => {
|
||||
assert.equal(recoverPendingJobs(db).length, 0);
|
||||
});
|
||||
|
||||
test('running jobs are visible and recovered on startup', () => {
|
||||
const db = openDb(':memory:');
|
||||
const id = enqueueJob(db, JOB({ sha: '4'.repeat(40) }));
|
||||
startJob(db, id);
|
||||
|
||||
const row = db.prepare(`SELECT state, started_ts FROM queue_jobs WHERE id = ?`).get(id);
|
||||
assert.equal(row.state, 'running');
|
||||
assert.match(row.started_ts, /^\d{4}-\d{2}-\d{2}T/);
|
||||
|
||||
const pending = recoverPendingJobs(db);
|
||||
assert.deepEqual(pending.map(p => p.id), [id]);
|
||||
});
|
||||
|
||||
test('unparseable persisted job is marked done, not re-run forever', () => {
|
||||
const db = openDb(':memory:');
|
||||
db.prepare(`INSERT INTO queue_jobs (repo, sha, job) VALUES ('a/b', 'c', '{broken')`).run();
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { chat } from '../src/lib/router.js';
|
||||
|
||||
test('review router client disables model fallbacks explicitly', async () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
let body;
|
||||
globalThis.fetch = async (_url, init) => {
|
||||
body = JSON.parse(init.body);
|
||||
return Response.json({
|
||||
message: { content: '{"axes":{},"overall":90,"grade":"A","findings":[]}' },
|
||||
_shre: { model: 'ollama-remote/qwen2.5-coder:7b' }
|
||||
});
|
||||
};
|
||||
|
||||
try {
|
||||
await chat({
|
||||
routerUrl: 'http://router.test',
|
||||
model: 'ollama-remote/qwen2.5-coder:7b',
|
||||
messages: [{ role: 'user', content: 'review' }],
|
||||
agentId: 'code-reviewer',
|
||||
sessionId: 'review-test',
|
||||
tenantId: 'nirlab',
|
||||
timeoutMs: 1000
|
||||
});
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
}
|
||||
|
||||
assert.deepEqual(body.fallbackModels, []);
|
||||
assert.equal(body.tools, false);
|
||||
assert.equal(body.raw, true);
|
||||
assert.equal(body.stream, false);
|
||||
});
|
||||
Reference in New Issue
Block a user