fix(review): run claude-cli models directly
This commit is contained in:
+26
-2
@@ -5,20 +5,44 @@ export class ReviewQueue {
|
|||||||
constructor() {
|
constructor() {
|
||||||
this.repoChains = new Map();
|
this.repoChains = new Map();
|
||||||
this.globalChain = Promise.resolve();
|
this.globalChain = Promise.resolve();
|
||||||
|
this.dedupeGenerations = new Map();
|
||||||
this.pending = 0;
|
this.pending = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Serialize fn per repo key. fn receives a `withGlobal` helper that
|
// Serialize fn per repo key. fn receives a `withGlobal` helper that
|
||||||
// serializes the wrapped section across ALL repos (use around LLM calls).
|
// serializes the wrapped section across ALL repos (use around LLM calls).
|
||||||
enqueue(repoKey, fn) {
|
enqueue(repoKey, fn, { dedupeKey = null, dedupeKeys = [], onSuperseded = null } = {}) {
|
||||||
const prev = this.repoChains.get(repoKey) || Promise.resolve();
|
const prev = this.repoChains.get(repoKey) || Promise.resolve();
|
||||||
|
const generationKeys = [...new Set([dedupeKey, ...dedupeKeys].filter(Boolean))]
|
||||||
|
.map((key) => `${repoKey}:${key}`);
|
||||||
|
const generations = new Map(generationKeys.map((key) => [
|
||||||
|
key,
|
||||||
|
(this.dedupeGenerations.get(key) || 0) + 1
|
||||||
|
]));
|
||||||
|
for (const [key, generation] of generations) {
|
||||||
|
this.dedupeGenerations.set(key, generation);
|
||||||
|
}
|
||||||
this.pending++;
|
this.pending++;
|
||||||
const next = prev
|
const next = prev
|
||||||
.catch(() => {})
|
.catch(() => {})
|
||||||
.then(() => fn(this.withGlobal.bind(this)))
|
.then(() => {
|
||||||
|
const superseded = [...generations].some(
|
||||||
|
([key, generation]) => this.dedupeGenerations.get(key) !== generation
|
||||||
|
);
|
||||||
|
if (superseded) {
|
||||||
|
onSuperseded?.();
|
||||||
|
return { superseded: true };
|
||||||
|
}
|
||||||
|
return fn(this.withGlobal.bind(this));
|
||||||
|
})
|
||||||
.finally(() => {
|
.finally(() => {
|
||||||
this.pending--;
|
this.pending--;
|
||||||
if (this.repoChains.get(repoKey) === next) this.repoChains.delete(repoKey);
|
if (this.repoChains.get(repoKey) === next) this.repoChains.delete(repoKey);
|
||||||
|
for (const [key, generation] of generations) {
|
||||||
|
if (this.dedupeGenerations.get(key) === generation) {
|
||||||
|
this.dedupeGenerations.delete(key);
|
||||||
|
}
|
||||||
|
}
|
||||||
});
|
});
|
||||||
this.repoChains.set(repoKey, next);
|
this.repoChains.set(repoKey, next);
|
||||||
return next;
|
return next;
|
||||||
|
|||||||
+50
-1
@@ -2,7 +2,56 @@
|
|||||||
// Real model answer lives in .message.content; top-level .content can carry an
|
// Real model answer lives in .message.content; top-level .content can carry an
|
||||||
// upstream-down apology — always prefer .message.content.
|
// upstream-down apology — always prefer .message.content.
|
||||||
|
|
||||||
export async function chat({ routerUrl, model, messages, agentId, sessionId, tenantId, timeoutMs }) {
|
import { execFile } from 'node:child_process';
|
||||||
|
import { promisify } from 'node:util';
|
||||||
|
|
||||||
|
const execFileAsync = promisify(execFile);
|
||||||
|
|
||||||
|
async function chatWithClaudeCli({ model, messages, timeoutMs, claudeBin = 'claude' }) {
|
||||||
|
const modelId = model.slice('claude-cli/'.length);
|
||||||
|
if (!modelId) throw new Error('claude-cli model id is required');
|
||||||
|
|
||||||
|
const prompt = messages
|
||||||
|
.map(({ role, content }) => `${String(role || 'user').toUpperCase()}:\n${String(content || '')}`)
|
||||||
|
.join('\n\n');
|
||||||
|
const env = { ...process.env };
|
||||||
|
delete env.ANTHROPIC_API_KEY;
|
||||||
|
delete env.ANTHROPIC_TOKEN;
|
||||||
|
delete env.ANTHROPIC_AUTH_TOKEN;
|
||||||
|
|
||||||
|
let stdout;
|
||||||
|
try {
|
||||||
|
({ stdout } = await execFileAsync(claudeBin, [
|
||||||
|
'-p', prompt,
|
||||||
|
'--model', modelId,
|
||||||
|
'--output-format', 'json',
|
||||||
|
'--tools', '',
|
||||||
|
'--permission-mode', 'dontAsk',
|
||||||
|
'--no-session-persistence'
|
||||||
|
], {
|
||||||
|
env,
|
||||||
|
timeout: timeoutMs,
|
||||||
|
killSignal: 'SIGTERM',
|
||||||
|
maxBuffer: 4 * 1024 * 1024
|
||||||
|
}));
|
||||||
|
} catch (error) {
|
||||||
|
const timedOut = error?.killed || error?.signal === 'SIGTERM' || error?.code === 'ETIMEDOUT';
|
||||||
|
const exitCode = Number.isInteger(error?.code) ? ` (exit ${error.code})` : '';
|
||||||
|
throw new Error(`claude-cli ${timedOut ? 'timed out' : 'failed'}${exitCode}`);
|
||||||
|
}
|
||||||
|
const data = JSON.parse(stdout);
|
||||||
|
const content = data?.result ?? data?.content;
|
||||||
|
if (typeof content !== 'string' || !content.trim()) {
|
||||||
|
throw new Error('claude-cli returned empty content');
|
||||||
|
}
|
||||||
|
return { content, servedModel: model };
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function chat(options) {
|
||||||
|
const { routerUrl, model, messages, agentId, sessionId, tenantId, timeoutMs } = options;
|
||||||
|
if (model.startsWith('claude-cli/')) {
|
||||||
|
return chatWithClaudeCli(options);
|
||||||
|
}
|
||||||
const res = await fetch(`${routerUrl.replace(/\/$/, '')}/v1/chat`, {
|
const res = await fetch(`${routerUrl.replace(/\/$/, '')}/v1/chat`, {
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers: { 'Content-Type': 'application/json' },
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
|||||||
+10
-1
@@ -37,9 +37,18 @@ function dispatchJob(queueRowId, job) {
|
|||||||
{
|
{
|
||||||
startJob(db, queueRowId);
|
startJob(db, queueRowId);
|
||||||
return runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log });
|
return runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log });
|
||||||
|
},
|
||||||
|
{
|
||||||
|
dedupeKeys: [
|
||||||
|
job.prIndex != null ? `pr:${job.prIndex}` : null,
|
||||||
|
`sha:${job.sha}`
|
||||||
|
].filter(Boolean)
|
||||||
}
|
}
|
||||||
).then(
|
).then(
|
||||||
(r) => finishJob(db, queueRowId, r?.ok ? (r.verdict || 'ok') : `error:${String(r?.error || '').slice(0, 200)}`),
|
(r) => {
|
||||||
|
if (r?.superseded) return finishJob(db, queueRowId, 'superseded');
|
||||||
|
return finishJob(db, queueRowId, r?.ok ? (r.verdict || 'ok') : `error:${String(r?.error || '').slice(0, 200)}`);
|
||||||
|
},
|
||||||
(e) => {
|
(e) => {
|
||||||
log(`review pipeline error for ${job.owner}/${job.repo}@${job.sha}: ${e.stack || e.message}`);
|
log(`review pipeline error for ${job.owner}/${job.repo}@${job.sha}: ${e.stack || e.message}`);
|
||||||
finishJob(db, queueRowId, `crash:${String(e.message || e).slice(0, 200)}`);
|
finishJob(db, queueRowId, `crash:${String(e.message || e).slice(0, 200)}`);
|
||||||
|
|||||||
@@ -0,0 +1,57 @@
|
|||||||
|
import { test } from 'node:test';
|
||||||
|
import assert from 'node:assert/strict';
|
||||||
|
|
||||||
|
import { ReviewQueue } from '../src/lib/queue.js';
|
||||||
|
|
||||||
|
test('newer jobs supersede queued work for the same pull request', async () => {
|
||||||
|
const queue = new ReviewQueue();
|
||||||
|
const ran = [];
|
||||||
|
const superseded = [];
|
||||||
|
|
||||||
|
const first = queue.enqueue(
|
||||||
|
'nirpa/hermes-agent',
|
||||||
|
async () => ran.push('old'),
|
||||||
|
{ dedupeKey: 'pr:86', onSuperseded: () => superseded.push('old') }
|
||||||
|
);
|
||||||
|
const second = queue.enqueue(
|
||||||
|
'nirpa/hermes-agent',
|
||||||
|
async () => ran.push('new'),
|
||||||
|
{ dedupeKey: 'pr:86', onSuperseded: () => superseded.push('new') }
|
||||||
|
);
|
||||||
|
|
||||||
|
await Promise.all([first, second]);
|
||||||
|
assert.deepEqual(ran, ['new']);
|
||||||
|
assert.deepEqual(superseded, ['old']);
|
||||||
|
assert.equal(queue.pending, 0);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('different pull requests are not deduplicated', async () => {
|
||||||
|
const queue = new ReviewQueue();
|
||||||
|
const ran = [];
|
||||||
|
|
||||||
|
await Promise.all([
|
||||||
|
queue.enqueue('nirpa/hermes-agent', async () => ran.push('pr-86'), { dedupeKey: 'pr:86' }),
|
||||||
|
queue.enqueue('nirpa/hermes-agent', async () => ran.push('pr-87'), { dedupeKey: 'pr:87' })
|
||||||
|
]);
|
||||||
|
|
||||||
|
assert.deepEqual(ran, ['pr-86', 'pr-87']);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('shared sha key deduplicates push and pull-request webhooks', async () => {
|
||||||
|
const queue = new ReviewQueue();
|
||||||
|
const ran = [];
|
||||||
|
|
||||||
|
const push = queue.enqueue(
|
||||||
|
'nirpa/hermes-agent',
|
||||||
|
async () => ran.push('push'),
|
||||||
|
{ dedupeKeys: ['sha:final'] }
|
||||||
|
);
|
||||||
|
const pullRequest = queue.enqueue(
|
||||||
|
'nirpa/hermes-agent',
|
||||||
|
async () => ran.push('pull-request'),
|
||||||
|
{ dedupeKeys: ['pr:86', 'sha:final'] }
|
||||||
|
);
|
||||||
|
|
||||||
|
await Promise.all([push, pullRequest]);
|
||||||
|
assert.deepEqual(ran, ['pull-request']);
|
||||||
|
});
|
||||||
@@ -1,5 +1,8 @@
|
|||||||
import { test } from 'node:test';
|
import { test } from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
|
import { chmod, mkdtemp, writeFile } from 'node:fs/promises';
|
||||||
|
import { tmpdir } from 'node:os';
|
||||||
|
import { join } from 'node:path';
|
||||||
import { chat } from '../src/lib/router.js';
|
import { chat } from '../src/lib/router.js';
|
||||||
|
|
||||||
test('review router client disables model fallbacks explicitly', async () => {
|
test('review router client disables model fallbacks explicitly', async () => {
|
||||||
@@ -32,3 +35,72 @@ test('review router client disables model fallbacks explicitly', async () => {
|
|||||||
assert.equal(body.raw, true);
|
assert.equal(body.raw, true);
|
||||||
assert.equal(body.stream, false);
|
assert.equal(body.stream, false);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('claude-cli models bypass the router and use subscription credentials', async () => {
|
||||||
|
const dir = await mkdtemp(join(tmpdir(), 'granthi-review-claude-'));
|
||||||
|
const fakeClaude = join(dir, 'claude');
|
||||||
|
await writeFile(fakeClaude, `#!/bin/sh
|
||||||
|
case " $* " in
|
||||||
|
*" --model claude-sonnet-4-6 "*) ;;
|
||||||
|
*) exit 21 ;;
|
||||||
|
esac
|
||||||
|
case " $* " in
|
||||||
|
*" --tools --permission-mode dontAsk --no-session-persistence "*) ;;
|
||||||
|
*) exit 22 ;;
|
||||||
|
esac
|
||||||
|
[ -z "\${ANTHROPIC_API_KEY:-}" ] || exit 23
|
||||||
|
[ -z "\${ANTHROPIC_TOKEN:-}" ] || exit 24
|
||||||
|
printf '%s' '{"result":"{\\"axes\\":{},\\"overall\\":90,\\"grade\\":\\"A\\",\\"findings\\":[]}"}'
|
||||||
|
`);
|
||||||
|
await chmod(fakeClaude, 0o700);
|
||||||
|
|
||||||
|
const originalFetch = globalThis.fetch;
|
||||||
|
const originalApiKey = process.env.ANTHROPIC_API_KEY;
|
||||||
|
const originalToken = process.env.ANTHROPIC_TOKEN;
|
||||||
|
globalThis.fetch = async () => { throw new Error('router must not be called'); };
|
||||||
|
process.env.ANTHROPIC_API_KEY = 'test-api-key';
|
||||||
|
process.env.ANTHROPIC_TOKEN = 'test-token';
|
||||||
|
|
||||||
|
try {
|
||||||
|
const result = await chat({
|
||||||
|
routerUrl: 'http://router.test',
|
||||||
|
model: 'claude-cli/claude-sonnet-4-6',
|
||||||
|
messages: [{ role: 'user', content: 'review this diff' }],
|
||||||
|
agentId: 'code-reviewer',
|
||||||
|
sessionId: 'review-test',
|
||||||
|
tenantId: 'nirlab',
|
||||||
|
timeoutMs: 1000,
|
||||||
|
claudeBin: fakeClaude
|
||||||
|
});
|
||||||
|
assert.match(result.content, /"overall":90/);
|
||||||
|
assert.equal(result.servedModel, 'claude-cli/claude-sonnet-4-6');
|
||||||
|
} finally {
|
||||||
|
globalThis.fetch = originalFetch;
|
||||||
|
if (originalApiKey === undefined) delete process.env.ANTHROPIC_API_KEY;
|
||||||
|
else process.env.ANTHROPIC_API_KEY = originalApiKey;
|
||||||
|
if (originalToken === undefined) delete process.env.ANTHROPIC_TOKEN;
|
||||||
|
else process.env.ANTHROPIC_TOKEN = originalToken;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test('claude-cli failures do not leak the review prompt', async () => {
|
||||||
|
const dir = await mkdtemp(join(tmpdir(), 'granthi-review-claude-error-'));
|
||||||
|
const fakeClaude = join(dir, 'claude');
|
||||||
|
await writeFile(fakeClaude, '#!/bin/sh\nprintf "%s\\n" "$*" >&2\nexit 25\n');
|
||||||
|
await chmod(fakeClaude, 0o700);
|
||||||
|
|
||||||
|
await assert.rejects(
|
||||||
|
chat({
|
||||||
|
routerUrl: 'http://router.test',
|
||||||
|
model: 'claude-cli/claude-sonnet-4-6',
|
||||||
|
messages: [{ role: 'user', content: 'SECRET_REVIEW_DIFF_MARKER' }],
|
||||||
|
timeoutMs: 1000,
|
||||||
|
claudeBin: fakeClaude
|
||||||
|
}),
|
||||||
|
(error) => {
|
||||||
|
assert.match(error.message, /claude-cli failed/);
|
||||||
|
assert.doesNotMatch(error.message, /SECRET_REVIEW_DIFF_MARKER/);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user