- {job.files.length === 0 ? (
-
No files processed.
- ) : (
- job.files.map((file) => (
-
-
{file.filePath}
-
-
-
- Prompt / diff
-
-
{file.diffInput ?? 'No prompt saved.'}
-
-
-
- Raw model output
-
-
{file.rawAiOutput ?? 'No raw output saved.'}
-
+ {/* Summary counts */}
+ {counts.total > 0 && (
+
+ {[
+ { label: 'Files', val: counts.total, cls: 'text-foreground' },
+ { label: 'Reviewed', val: counts.done, cls: 'text-success' },
+ { label: 'Skipped', val: counts.skipped, cls: 'text-muted-foreground' },
+ { label: 'Failed', val: counts.failed, cls: counts.failed > 0 ? 'text-danger' : 'text-muted-foreground' },
+ ].map(({ label, val, cls }) => (
+
+ {val}
+ {label}
-
- ))
+ ))}
+
)}
+
+ {error &&
{error}}
+
+ {/* File list */}
+ {job.files.length === 0 ? (
+
+
+
+
No files processed yet
+ {(job.status === 'running' || job.status === 'queued') && (
+
+ Logs appear here once files are reviewed
+
+ )}
+
+
+ ) : (
+
+ {job.files.map(file => (
+
+ ))}
+
+ )}
);
}
diff --git a/src/client/pages/repos.tsx b/src/client/pages/repos.tsx
index 0b70032..d45b9c8 100644
--- a/src/client/pages/repos.tsx
+++ b/src/client/pages/repos.tsx
@@ -238,7 +238,7 @@ function RepoModelModal({
if (!repo || !dirty) return;
setSaving('apply');
setError(null);
- const tid = toast.loading(`Saving model strategy for ${repo.owner}/${repo.repo}...`);
+ const tid = toast.loading('Applying model strategy…');
try {
await api.updateRepoConfig(repo.owner, repo.repo, {
model: {
@@ -249,11 +249,11 @@ function RepoModelModal({
});
setInitialRoute(route);
onModelApplied(repo, route);
- toast.success('Model strategy saved', { id: tid });
+ toast.success('Strategy saved', { id: tid, description: `${repo.owner}/${repo.repo} now uses a custom model chain.` });
} catch (err) {
const msg = err instanceof Error ? err.message : 'Failed to save model strategy.';
setError(msg);
- toast.error('Save failed', { id: tid, description: msg });
+ toast.error('Could not save strategy', { id: tid, description: 'Your changes were not applied. Please try again.' });
} finally {
setSaving(null);
}
@@ -263,7 +263,7 @@ function RepoModelModal({
if (!repo) return;
setSaving('reset');
setError(null);
- const tid = toast.loading(`Using global strategy for ${repo.owner}/${repo.repo}...`);
+ const tid = toast.loading('Resetting to global defaults…');
try {
await api.updateRepoConfig(repo.owner, repo.repo, {
model: {
@@ -276,11 +276,11 @@ function RepoModelModal({
setRoute(globalRoute);
setInitialRoute(globalRoute);
onModelReset(repo);
- toast.success('Repo now uses global strategy', { id: tid });
+ toast.success('Reset to global strategy', { id: tid, description: `${repo.owner}/${repo.repo} will inherit account defaults.` });
} catch (err) {
const msg = err instanceof Error ? err.message : 'Failed to reset model strategy.';
setError(msg);
- toast.error('Reset failed', { id: tid, description: msg });
+ toast.error('Reset failed', { id: tid, description: 'Could not remove the custom strategy. Try again.' });
} finally {
setSaving(null);
}
@@ -379,14 +379,19 @@ export function ReposPage() {
const handleToggleEnabled = async (repo: RepoConfigRecord, nextEnabled: boolean) => {
const targetId = repoId(repo);
setPendingToggles(current => new Set(current).add(targetId));
- const tid = toast.loading(`${nextEnabled ? 'Enabling' : 'Pausing'} ${targetId}...`);
+ const tid = toast.loading(nextEnabled ? 'Enabling code reviews…' : 'Pausing code reviews…');
try {
await api.updateRepoConfig(repo.owner, repo.repo, { enabled: nextEnabled });
mergeRepo(targetId, { enabled: nextEnabled });
- toast.success(nextEnabled ? 'Reviews enabled' : 'Reviews paused', { id: tid, description: targetId });
+ toast.success(
+ nextEnabled ? 'Reviews active' : 'Reviews paused',
+ { id: tid, description: nextEnabled
+ ? `${targetId} will receive automated review comments.`
+ : `${targetId} is now quiet — no new reviews will be posted.`
+ },
+ );
} catch (err) {
- const msg = err instanceof Error ? err.message : 'Failed to update repository.';
- toast.error('Update failed', { id: tid, description: msg });
+ toast.error('Could not update repository', { id: tid, description: 'The change did not go through. Please try again.' });
} finally {
setPendingToggles(current => {
const next = new Set(current);
@@ -416,19 +421,21 @@ export function ReposPage() {
if (syncing) return;
setSyncing(true);
setError(null);
- const tid = toast.loading('Syncing repositories from GitHub...');
+ const tid = toast.loading('Syncing with GitHub…');
try {
const result = await api.syncRepos();
const syncedCount = result?.synced?.length ?? 0;
- toast.success('Repositories synced', {
+ toast.success('Repositories up to date', {
id: tid,
- description: `${syncedCount} ${syncedCount === 1 ? 'repository' : 'repositories'} synced successfully`,
+ description: syncedCount > 0
+ ? `${syncedCount} ${syncedCount === 1 ? 'repository' : 'repositories'} refreshed from GitHub.`
+ : 'Everything is already in sync.',
});
loadRepos();
} catch (e) {
const msg = e instanceof Error ? e.message : 'Sync failed.';
setError(msg);
- toast.error('Sync failed', { id: tid, description: msg });
+ toast.error('Sync failed', { id: tid, description: 'Could not reach GitHub. Check your connection and try again.' });
} finally {
setSyncing(false);
}
diff --git a/src/client/pages/settings.tsx b/src/client/pages/settings.tsx
index cc75810..f09f86e 100644
--- a/src/client/pages/settings.tsx
+++ b/src/client/pages/settings.tsx
@@ -98,7 +98,7 @@ export function SettingsPage() {
} catch (e) {
const msg = e instanceof Error ? e.message : 'Failed to load settings';
setError(msg);
- toast.error('Failed to load settings', { description: msg });
+ toast.error('Could not load settings', { description: 'Something went wrong fetching your configuration.' });
} finally {
setLoading(false);
}
@@ -110,18 +110,18 @@ export function SettingsPage() {
if (!globalConfig || !globalDirty) return;
setSaving('global');
setError(null);
- const tid = toast.loading('Saving global strategy...');
+ const tid = toast.loading('Saving model strategy…');
try {
await api.updateGlobalConfig(globalConfig);
setSavedGlobalConfig(globalConfig);
toast.success('Global strategy saved', {
id: tid,
- description: `Primary model: ${getModelLabel(globalConfig.main)}`,
+ description: 'All repositories without a custom strategy will use these settings.',
});
} catch (e) {
const msg = e instanceof Error ? e.message : 'Update failed';
setError(msg);
- toast.error('Failed to save strategy', { id: tid, description: msg });
+ toast.error('Could not save strategy', { id: tid, description: 'Your changes were not applied. Please try again.' });
} finally {
setSaving(null);
}
@@ -137,16 +137,16 @@ export function SettingsPage() {
if (!current) return;
setSaving(id);
setError(null);
- const tid = toast.loading(`Updating ${id}...`);
+ const tid = toast.loading('Updating quota…');
try {
await api.updateModelConfig(id, quotaPayload(current));
const saved = { ...current, updatedAt: new Date().toISOString() };
markConfigSaved(id, saved);
- toast.success('Model quota updated', { id: tid, description: id });
+ toast.success('Quota updated', { id: tid, description: 'Rate limits have been applied to this model.' });
} catch (e) {
const msg = e instanceof Error ? e.message : 'Update failed';
setError(msg);
- toast.error('Failed to update quota', { id: tid, description: msg });
+ toast.error('Quota update failed', { id: tid, description: 'The limit change did not save. Please try again.' });
} finally {
setSaving(null);
}
@@ -156,7 +156,11 @@ export function SettingsPage() {
if (dirtyConfigs.length === 0) return;
setSaving('quotas');
setError(null);
- const tid = toast.loading(`Saving ${dirtyConfigs.length} quota ${dirtyConfigs.length === 1 ? 'change' : 'changes'}...`);
+ const tid = toast.loading(
+ dirtyConfigs.length === 1
+ ? 'Saving 1 quota change…'
+ : `Saving ${dirtyConfigs.length} quota changes…`,
+ );
try {
await Promise.all(dirtyConfigs.map(cfg => api.updateModelConfig(cfg.modelId, quotaPayload(cfg))));
const now = new Date().toISOString();
@@ -165,11 +169,11 @@ export function SettingsPage() {
setSavedConfigs(current =>
configs.map(cfg => (dirtyIds.has(cfg.modelId) ? { ...cfg, updatedAt: now } : current.find(saved => saved.modelId === cfg.modelId) ?? cfg)),
);
- toast.success('Quotas saved', { id: tid });
+ toast.success('All quotas saved', { id: tid, description: `${dirtyConfigs.length} ${dirtyConfigs.length === 1 ? 'model' : 'models'} updated successfully.` });
} catch (e) {
const msg = e instanceof Error ? e.message : 'Update failed';
setError(msg);
- toast.error('Failed to save quotas', { id: tid, description: msg });
+ toast.error('Could not save quotas', { id: tid, description: 'One or more limits failed to save. Try again.' });
} finally {
setSaving(null);
}
diff --git a/src/server/core/job-recovery.ts b/src/server/core/job-recovery.ts
index 9b3a80a..c00c193 100644
--- a/src/server/core/job-recovery.ts
+++ b/src/server/core/job-recovery.ts
@@ -59,3 +59,13 @@ export async function runBestEffortJobMaintenance(env: AppBindings) {
logger.error('Opportunistic job maintenance failed', error instanceof Error ? error : new Error(String(error)));
}
}
+
+export function scheduleBestEffortJobMaintenance(
+ env: AppBindings,
+ executionCtx?: Pick
,
+) {
+ const task = runBestEffortJobMaintenance(env);
+ if (executionCtx) {
+ executionCtx.waitUntil(task);
+ }
+}
diff --git a/src/server/core/model-output.ts b/src/server/core/model-output.ts
index 856912e..3c14633 100644
--- a/src/server/core/model-output.ts
+++ b/src/server/core/model-output.ts
@@ -5,11 +5,11 @@ import { findClosestValidLine, findPositionForLine, getValidNewLines, getValidPo
import type { FileDiff } from './diff';
import { jsonrepair } from 'jsonrepair';
-const MAX_LOGGED_MODEL_OUTPUT_CHARS = 2_000;
+const MAX_LOGGED_JSON_CHARS = 2_000;
-function truncateForLog(value: string) {
- if (value.length <= MAX_LOGGED_MODEL_OUTPUT_CHARS) return value;
- return `${value.slice(0, MAX_LOGGED_MODEL_OUTPUT_CHARS)}... [truncated ${value.length - MAX_LOGGED_MODEL_OUTPUT_CHARS} chars]`;
+function truncateJsonForLog(value: string) {
+ if (value.length <= MAX_LOGGED_JSON_CHARS) return value;
+ return `${value.slice(0, MAX_LOGGED_JSON_CHARS)}... [truncated ${value.length - MAX_LOGGED_JSON_CHARS} chars]`;
}
function hasReviewKeys(input: string) {
@@ -260,7 +260,10 @@ export function parseFileReviewResponse(raw: string, file: FileDiff): {
throw new Error('Model response did not contain review JSON keys.');
}
} catch (e) {
- logger.error('Failed to extract JSON from model response', { raw: truncateForLog(raw), error: e });
+ logger.error('Failed to extract JSON from model response', {
+ rawLength: raw.length,
+ error: e instanceof Error ? e.message : String(e),
+ });
throw new Error('Could not find JSON root in model response.');
}
@@ -276,14 +279,14 @@ export function parseFileReviewResponse(raw: string, file: FileDiff): {
try {
repaired = jsonrepair(preprocessed);
} catch (e) {
- logger.warn('jsonrepair failed to fix model output, using preprocessed text', { preprocessed: truncateForLog(preprocessed), error: e });
+ logger.warn('jsonrepair failed to fix model output, using preprocessed text', { preprocessed: truncateJsonForLog(preprocessed), error: e });
}
let parsedJson: any;
try {
parsedJson = JSON.parse(repaired);
} catch (e) {
- logger.error('Critical JSON parse error after extraction and repair', { repaired: truncateForLog(repaired), error: e });
+ logger.error('Critical JSON parse error after extraction and repair', { repaired: truncateJsonForLog(repaired), error: e });
throw new Error(`Invalid JSON format: ${e instanceof Error ? e.message : 'Unknown error'}`);
}
diff --git a/src/server/core/review.ts b/src/server/core/review.ts
index 57187fa..c4d4923 100644
--- a/src/server/core/review.ts
+++ b/src/server/core/review.ts
@@ -18,11 +18,11 @@ type PersistedReviewJob = ReturnType;
export type ReviewJobRunResult = { action: 'ack' } | { action: 'retry'; delaySeconds: number };
-const REVIEW_CHUNK_FILE_LIMIT = 2;
-const REVIEW_CHUNK_WALL_CLOCK_MS = 8 * 60 * 1000;
-const JOB_LEASE_SECONDS = 10 * 60;
+const REVIEW_CHUNK_FILE_LIMIT = 3;
+const REVIEW_CHUNK_WALL_CLOCK_MS = 12 * 60 * 1000;
+const JOB_LEASE_SECONDS = 15 * 60;
const BUSY_RETRY_SECONDS = 60;
-const RETRYABLE_MODEL_FAILURE_RETRY_SECONDS = 60;
+const RETRYABLE_MODEL_FAILURE_RETRY_DELAYS_SECONDS = [60, 5 * 60, 15 * 60];
const MAX_RETRYABLE_FILE_REVIEW_FAILURES = 3;
function isRetryableFileReviewErrorMessage(message: string | null | undefined) {
@@ -45,6 +45,21 @@ function isRetryableFileReviewErrorMessage(message: string | null | undefined) {
);
}
+function retryableModelFailureDelaySeconds(failureCount: number | null | undefined) {
+ if (!failureCount || failureCount < 1) return RETRYABLE_MODEL_FAILURE_RETRY_DELAYS_SECONDS[0];
+ const index = Math.min(failureCount - 1, RETRYABLE_MODEL_FAILURE_RETRY_DELAYS_SECONDS.length - 1);
+ return RETRYABLE_MODEL_FAILURE_RETRY_DELAYS_SECONDS[index];
+}
+
+function getRetryableModelFailureDelaySeconds(error: unknown) {
+ const record = error && typeof error === 'object' ? error as { retryAfterSeconds?: unknown } : null;
+ const retryAfterSeconds =
+ typeof record?.retryAfterSeconds === 'number'
+ ? record.retryAfterSeconds
+ : null;
+ return retryAfterSeconds ?? RETRYABLE_MODEL_FAILURE_RETRY_DELAYS_SECONDS[0];
+}
+
function shouldRetryExistingFileReview(review: { file_status: string; error_msg: string | null }) {
return review.file_status === 'failed' && isRetryableFileReviewErrorMessage(review.error_msg);
}
@@ -187,7 +202,7 @@ export async function runReviewJob(env: AppBindings, message: ReviewJobMessage):
if (phase === 'prepare') {
await runPreparePhase(env, job, leaseOwner, github);
} else if (phase === 'finalize') {
- await runFinalizePhase(env, job, leaseOwner, github, model, formatter);
+ await runFinalizePhase(env, job, leaseOwner, github, formatter);
} else {
await runReviewPhase(env, job, leaseOwner, github, model);
}
@@ -203,12 +218,13 @@ export async function runReviewJob(env: AppBindings, message: ReviewJobMessage):
}
if (isRetryableModelError(error)) {
+ const delaySeconds = getRetryableModelFailureDelaySeconds(error);
logger.warn(`Review job hit transient model/provider failure; scheduling delayed continuation: ${job.owner}/${job.repo} PR #${job.prNumber}`, {
error: messageText,
phase,
- delaySeconds: RETRYABLE_MODEL_FAILURE_RETRY_SECONDS,
+ delaySeconds,
});
- await enqueueJobPhase(env, job.id, phase, RETRYABLE_MODEL_FAILURE_RETRY_SECONDS);
+ await enqueueJobPhase(env, job.id, phase, delaySeconds);
await releaseJobLease(env, job.id, leaseOwner);
return { action: 'ack' };
}
@@ -417,6 +433,8 @@ async function runReviewPhase(
const currentReviews = new Map(allExistingReviews.filter((review) => review.job_id === job.id).map((review) => [review.file_path, review]));
const parentReviews = new Map(allExistingReviews.filter((review) => review.job_id !== job.id && review.file_status === 'done').map((review) => [review.file_path, review]));
+ const reviewTasks: Array> = [];
+
for (const file of files) {
const existingReview = currentReviews.get(file.path);
if (existingReview && countsAsHandledFileReview(existingReview)) {
@@ -424,12 +442,15 @@ async function runReviewPhase(
}
const inherited = parentReviews.get(file.path);
- if (inherited) {
+ const reviewTask = async () => {
+ if (!inherited) {
+ await reviewAndPersistFile(env, job, file, pr, config, totalLineCount, model, existingReview);
+ return;
+ }
+
if (!canInheritParentFileReview(config, inherited)) {
logger.info(`Ignoring inherited review for ${file.path}; parent model ${inherited.model_used} is not in the current model strategy`);
await reviewAndPersistFile(env, job, file, pr, config, totalLineCount, model, existingReview);
- processedThisChunk += 1;
- await heartbeatAndCheckSuperseded(env, job.id, leaseOwner);
} else {
await upsertFileReview(env, job.id, {
filePath: file.path,
@@ -450,20 +471,30 @@ async function runReviewPhase(
errorMessage: null,
});
currentReviews.set(file.path, inherited);
- processedThisChunk += 1;
- await heartbeatAndCheckSuperseded(env, job.id, leaseOwner);
}
- } else {
- await reviewAndPersistFile(env, job, file, pr, config, totalLineCount, model, existingReview);
- processedThisChunk += 1;
- await heartbeatAndCheckSuperseded(env, job.id, leaseOwner);
- }
+ };
+
+ reviewTasks.push(reviewTask());
+ processedThisChunk += 1;
if (processedThisChunk >= REVIEW_CHUNK_FILE_LIMIT || Date.now() - startedAt >= REVIEW_CHUNK_WALL_CLOCK_MS) {
break;
}
}
+ const results = await Promise.allSettled(reviewTasks);
+ await heartbeatAndCheckSuperseded(env, job.id, leaseOwner);
+
+ const rejected = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected');
+ if (rejected.length > 0) {
+ rejected.forEach((result, index) => {
+ logger.error(`Review chunk task ${index + 1}/${rejected.length} failed`, result.reason);
+ });
+ throw rejected.length === 1
+ ? rejected[0].reason
+ : new AggregateError(rejected.map((result) => result.reason), `${rejected.length} review chunk tasks failed`);
+ }
+
const latestReviews = await getFileReviewsForJobs(env, [job.id]);
const reviewedPaths = new Set(latestReviews.filter(countsAsHandledFileReview).map((review) => review.file_path));
const completedCount = files.filter((file) => reviewedPaths.has(file.path)).length;
@@ -567,6 +598,10 @@ async function reviewAndPersistFile(
error: errorMessage,
attempts: failureCount,
});
+ Object.defineProperty(error, 'retryAfterSeconds', {
+ value: retryableModelFailureDelaySeconds(failureCount),
+ configurable: true,
+ });
throw error;
}
@@ -606,7 +641,6 @@ async function runFinalizePhase(
job: PersistedReviewJob,
leaseOwner: string,
github: GitHubService,
- model: ModelService,
formatter: FormatterService,
) {
await updateJobStep(env, job.id, 'Generating Summary', { status: 'running' });
@@ -640,12 +674,6 @@ async function runFinalizePhase(
const hasFailures = fileSummaries.some((file) => file.verdict === 'failed');
const failedFileCount = fileSummaries.filter((file) => file.verdict === 'failed').length;
const verdictSummary = formatter.summarizeVerdict(reviewedComments, hasFailures);
- const summaryResponse = await model.generateSummary({
- prTitle: pr.title ?? null,
- verdict: verdictSummary.verdict,
- fileSummaries,
- config,
- });
await updateJobStep(env, job.id, 'Generating Summary', { status: 'done' });
await heartbeatAndCheckSuperseded(env, job.id, leaseOwner);
@@ -699,11 +727,11 @@ async function runFinalizePhase(
verdict: verdictSummary.verdict,
fileCount: files.length,
commentCount: reviewedComments.length,
- totalInputTokens: fileInputTokens + (summaryResponse.inputTokens ?? 0),
- totalOutputTokens: fileOutputTokens + (summaryResponse.outputTokens ?? 0),
+ totalInputTokens: fileInputTokens,
+ totalOutputTokens: fileOutputTokens,
summaryMarkdown: formattedSummary,
reviewId: review.id,
- summaryModel: summaryResponse.modelUsed,
+ summaryModel: null,
errorMessage: partialErrorMessage,
});
await updateJobStep(env, job.id, 'Completing', { status: 'done' });
@@ -724,7 +752,7 @@ async function enqueueJobPhase(
phase: 'prepare' | 'review' | 'finalize',
delaySeconds = 0,
) {
- await markJobContinuationQueued(env, jobId);
+ await markJobContinuationQueued(env, jobId, delaySeconds);
await env.REVIEW_QUEUE.send(
{
jobId,
diff --git a/src/server/db/jobs.ts b/src/server/db/jobs.ts
index ffe3cbd..b97b83a 100644
--- a/src/server/db/jobs.ts
+++ b/src/server/db/jobs.ts
@@ -82,7 +82,34 @@ export function bytesToHex(value: ByteaValue) {
return Array.from(bytes, (byte) => byte.toString(16).padStart(2, '0')).join('');
}
+function latestTimestamp(...values: Array) {
+ const now = Date.now();
+ return values.reduce((latest, value) => {
+ if (!value) return latest;
+ if (new Date(value).getTime() > now) return latest;
+ if (!latest) return value;
+ return new Date(value).getTime() > new Date(latest).getTime() ? value : latest;
+ }, null);
+}
+
export function mapJob(row: JobRow) {
+ const lastQueueMessageAt = row.last_queue_message_at ? new Date(row.last_queue_message_at).getTime() : null;
+ const nextRetryAt =
+ row.status === 'running' &&
+ row.lease_owner === null &&
+ lastQueueMessageAt !== null &&
+ Number.isFinite(lastQueueMessageAt) &&
+ lastQueueMessageAt > Date.now()
+ ? row.last_queue_message_at
+ : null;
+ const updatedAt = latestTimestamp(
+ row.created_at,
+ row.started_at,
+ row.finished_at,
+ row.heartbeat_at,
+ row.last_queue_message_at,
+ ) ?? row.created_at;
+
return jobSummarySchema.parse({
id: row.id,
owner: row.owner,
@@ -100,6 +127,8 @@ export function mapJob(row: JobRow) {
totalInputTokens: row.total_input_tokens ?? 0,
totalOutputTokens: row.total_output_tokens ?? 0,
createdAt: row.created_at,
+ updatedAt,
+ nextRetryAt,
startedAt: row.started_at,
finishedAt: row.finished_at,
errorMessage: row.error_msg,
@@ -402,6 +431,12 @@ export async function claimJobLease(
OR lease_expires_at < now()
OR lease_owner = $2
)
+ AND NOT (
+ status = 'running'
+ AND lease_owner IS NULL
+ AND last_queue_message_at IS NOT NULL
+ AND last_queue_message_at > now()
+ )
RETURNING *
)
SELECT c.*, r.owner, r.repo, r.installation_id
@@ -424,8 +459,10 @@ export async function claimJobLease(
return { status: 'terminal', row };
}
- const expiresAt = row.lease_expires_at ? new Date(row.lease_expires_at).getTime() : 0;
- const secondsUntilExpiry = Math.ceil((expiresAt - Date.now()) / 1000);
+ const leaseExpiresAt = row.lease_expires_at ? new Date(row.lease_expires_at).getTime() : 0;
+ const delayedUntil = row.lease_owner === null && row.last_queue_message_at ? new Date(row.last_queue_message_at).getTime() : 0;
+ const retryAt = Math.max(leaseExpiresAt, delayedUntil);
+ const secondsUntilExpiry = Math.ceil((retryAt - Date.now()) / 1000);
return {
status: 'busy',
row,
@@ -467,16 +504,20 @@ export async function releaseJobLease(env: Pick, jobI
);
}
-export async function markJobContinuationQueued(env: Pick, jobId: string) {
+export async function markJobContinuationQueued(env: Pick, jobId: string, delaySeconds = 0) {
await queryRows(
env,
`
UPDATE jobs
- SET last_queue_message_at = now()
+ SET heartbeat_at = now(),
+ last_queue_message_at = CASE
+ WHEN $2::int > 0 THEN now() + ($2::text || ' seconds')::interval
+ ELSE now()
+ END
WHERE id = $1
AND status = 'running'
`,
- [jobId],
+ [jobId, delaySeconds],
);
}
@@ -664,7 +705,8 @@ export async function updateJobStep(
env,
`
UPDATE jobs
- SET steps = CASE
+ SET heartbeat_at = now(),
+ steps = CASE
WHEN EXISTS (SELECT 1 FROM jsonb_array_elements(COALESCE(steps, '[]'::jsonb)) s WHERE s->>'name' = $2)
THEN (
SELECT jsonb_agg(
diff --git a/src/server/models/cloudflare.ts b/src/server/models/cloudflare.ts
index 4814f25..36d4bc2 100644
--- a/src/server/models/cloudflare.ts
+++ b/src/server/models/cloudflare.ts
@@ -4,8 +4,56 @@ import { TimeoutError } from '@server/core/timeout';
import type { ModelResponse } from './types';
/** Max wall-clock time allowed for a single Workers-AI call. */
-const CLOUDFLARE_TIMEOUT_MS = 45_000;
-const CLOUDFLARE_MAX_RETRIES = 1;
+const CLOUDFLARE_TIMEOUT_MS = 180_000;
+const CLOUDFLARE_MAX_RETRIES = 0;
+const CLOUDFLARE_MAX_OUTPUT_TOKENS = 8192;
+const REVIEW_RESPONSE_SCHEMA = {
+ type: 'object',
+ additionalProperties: false,
+ required: ['findings', 'overall_explanation', 'overall_correctness', 'overall_confidence_score'],
+ properties: {
+ findings: {
+ type: 'array',
+ maxItems: 10,
+ items: {
+ type: 'object',
+ additionalProperties: false,
+ required: ['title', 'body', 'priority', 'code_location'],
+ properties: {
+ title: { type: 'string', maxLength: 100 },
+ body: { type: 'string' },
+ confidence_score: { type: 'number', minimum: 0, maximum: 1 },
+ priority: { type: 'integer', minimum: 0, maximum: 3 },
+ code_location: {
+ type: 'object',
+ additionalProperties: false,
+ properties: {
+ absolute_file_path: { type: 'string' },
+ line: { type: 'integer', minimum: 1 },
+ line_range: {
+ type: 'object',
+ additionalProperties: false,
+ required: ['start', 'end'],
+ properties: {
+ start: { type: 'integer', minimum: 1 },
+ end: { type: 'integer', minimum: 1 },
+ },
+ },
+ },
+ anyOf: [
+ { required: ['line'] },
+ { required: ['line_range'] },
+ ],
+ },
+ code_suggestion: { type: 'string' },
+ },
+ },
+ },
+ overall_explanation: { type: 'string' },
+ overall_correctness: { type: 'string', enum: ['patch is correct', 'patch is incorrect'] },
+ overall_confidence_score: { type: 'number', minimum: 0, maximum: 1 },
+ },
+} as const;
type UnknownRecord = Record;
@@ -35,6 +83,18 @@ function getNumber(value: unknown, key: string) {
return typeof child === 'number' ? child : null;
}
+function synthesizeInconclusiveReview(model: string, reason: string): string {
+ logger.warn(`Cloudflare model ${model} returned no parseable review content; synthesizing inconclusive review JSON`, {
+ reason,
+ });
+ return JSON.stringify({
+ findings: [],
+ overall_correctness: 'patch is incorrect',
+ overall_explanation: `Cloudflare model ${model} returned no parseable review content (${reason}). The file review is inconclusive.`,
+ overall_confidence_score: 0,
+ });
+}
+
function extractMessageContent(content: unknown): string | null {
if (isText(content)) return content.trim();
@@ -69,14 +129,16 @@ function extractCloudflareText(result: unknown, model: string): string {
if (content) return content;
const finishReason = isRecord(choice) ? choice.finish_reason ?? choice.stop_reason : null;
- if (finishReason) {
- throw new Error(`Cloudflare model ${model} returned no review content (finish_reason=${finishReason}).`);
+ const reasoning = isText(message?.reasoning) ? message.reasoning : isText(message?.reasoning_content) ? message.reasoning_content : null;
+ if (reasoning) {
+ return synthesizeInconclusiveReview(model, `reasoning-only response${finishReason ? `, finish_reason=${String(finishReason)}` : ''}`);
}
- if (isText(message?.reasoning) || isText(message?.reasoning_content)) {
- throw new Error(`Cloudflare model ${model} returned reasoning without review content.`);
+
+ if (finishReason) {
+ return synthesizeInconclusiveReview(model, `finish_reason=${String(finishReason)}`);
}
- throw new Error(`Cloudflare model ${model} returned an empty response.`);
+ return synthesizeInconclusiveReview(model, 'empty response');
}
function extractCloudflareUsage(result: unknown) {
@@ -116,11 +178,23 @@ export async function reviewWithCloudflare(
const result = await Promise.race([
env.AI.run(model as any, {
messages: [
- { role: 'system', content: input.systemPrompt },
- { role: 'user', content: input.userPrompt },
+ {
+ role: 'system',
+ content: `${input.systemPrompt}\n\nReturn only the JSON object. Do not include chain-of-thought, analysis, markdown, code fences, or explanatory prose.`,
+ },
+ { role: 'user', content: `${input.userPrompt}\n\nRespond with the required JSON object only.` },
],
- max_completion_tokens: 4096,
+ max_completion_tokens: CLOUDFLARE_MAX_OUTPUT_TOKENS,
+ response_format: {
+ type: 'json_schema',
+ json_schema: {
+ name: 'codra_file_review',
+ strict: true,
+ schema: REVIEW_RESPONSE_SCHEMA,
+ },
+ },
temperature: 0,
+ top_p: 0.1,
}),
timeoutPromise,
]);
diff --git a/src/server/models/google.ts b/src/server/models/google.ts
index b742768..ac31761 100644
--- a/src/server/models/google.ts
+++ b/src/server/models/google.ts
@@ -4,9 +4,13 @@ import { withTimeout } from '@server/core/timeout';
import type { ModelResponse } from './types';
/** Max wall-clock time allowed for a single Google AI Studio call. */
-const GOOGLE_TIMEOUT_MS = 45_000;
+const GOOGLE_TIMEOUT_MS = 180_000;
const GOOGLE_MAX_RETRIES = 1;
-const GOOGLE_MAX_OUTPUT_TOKENS = 3072;
+const GOOGLE_MAX_OUTPUT_TOKENS = 4096;
+
+function isRetryableGoogleStatus(status: number) {
+ return status === 408 || status === 503 || status === 524;
+}
export async function reviewWithGoogle(
env: Pick,
@@ -57,13 +61,18 @@ export async function reviewWithGoogle(
if (!response.ok) {
const errorText = await response.text();
const isRateLimit = response.status === 429;
- const isRetryable = !isRateLimit && response.status >= 500;
+ const isRetryable = isRetryableGoogleStatus(response.status);
- logger.error(`Google request failed with ${response.status}`, {
+ const logData = {
error: errorText,
attempt,
- willRetry: isRetryable && attempt < maxRetries
- });
+ willRetry: isRetryable && attempt < maxRetries,
+ };
+ if (isRetryable && attempt < maxRetries) {
+ logger.warn(`Google request failed with ${response.status}; retrying`, logData);
+ } else {
+ logger.error(`Google request failed with ${response.status}`, logData);
+ }
if (isRateLimit) {
throw new Error(`Google request failed with ${response.status}: ${errorText}`);
diff --git a/src/server/routes/api/jobs.ts b/src/server/routes/api/jobs.ts
index 88c2c2d..28503c6 100644
--- a/src/server/routes/api/jobs.ts
+++ b/src/server/routes/api/jobs.ts
@@ -1,16 +1,29 @@
import { Hono } from 'hono';
+import type { Context } from 'hono';
import { jobsQuerySchema } from '@shared/schema';
import type { AppEnv } from '@server/env';
import { bytesToHex, getJobDetail, getJobForProcessing, insertJob, listJobs, mapJob, supersedeOlderJobs } from '@server/db/jobs';
import { jsonError } from '@server/core/http';
-import { runBestEffortJobMaintenance } from '@server/core/job-recovery';
+import { scheduleBestEffortJobMaintenance } from '@server/core/job-recovery';
import { loadRepoConfig } from '@server/core/config';
+function jobEtag(input: { id: string; status: string; updatedAt: string; fileCount: number; commentCount: number }) {
+ return `"job-${input.id}-${input.status}-${input.fileCount}-${input.commentCount}-${new Date(input.updatedAt).getTime()}"`;
+}
+
+function getExecutionContext(c: Context) {
+ try {
+ return c.executionCtx;
+ } catch {
+ return undefined;
+ }
+}
+
export function createJobsRouter() {
const app = new Hono();
app.get('/', async (c) => {
- await runBestEffortJobMaintenance(c.env);
+ scheduleBestEffortJobMaintenance(c.env, getExecutionContext(c));
const rawQuery = c.req.query();
const query = jobsQuerySchema.parse(rawQuery);
@@ -20,14 +33,36 @@ export function createJobsRouter() {
});
app.get('/:id', async (c) => {
- await runBestEffortJobMaintenance(c.env);
+ scheduleBestEffortJobMaintenance(c.env, getExecutionContext(c));
+
+ const rawJob = await getJobForProcessing(c.env, c.req.param('id'));
+ if (!rawJob) {
+ return jsonError('Job not found.', 404);
+ }
+
+ const summary = mapJob(rawJob);
+ const etag = jobEtag(summary);
+ const lastModified = new Date(summary.updatedAt).toUTCString();
+ if (c.req.header('if-none-match') === etag) {
+ return new Response(null, {
+ status: 304,
+ headers: {
+ ETag: etag,
+ 'Last-Modified': lastModified,
+ },
+ });
+ }
const job = await getJobDetail(c.env, c.req.param('id'));
if (!job) {
return jsonError('Job not found.', 404);
}
- return c.json({ job });
+ const response = c.json({ job });
+ response.headers.set('ETag', etag);
+ response.headers.set('Last-Modified', lastModified);
+ response.headers.set('Cache-Control', 'private, no-cache');
+ return response;
});
app.post('/:id/retry', async (c) => {
diff --git a/src/server/routes/webhook.ts b/src/server/routes/webhook.ts
index 3241b73..2fbaa1a 100644
--- a/src/server/routes/webhook.ts
+++ b/src/server/routes/webhook.ts
@@ -1,4 +1,5 @@
import { Hono } from 'hono';
+import type { Context } from 'hono';
import { isSupportedGitHubWebhookEvent, type GitHubWebhookPayload } from '@shared/github';
import type { AppEnv } from '@server/env';
import { loadRepoConfig } from '@server/core/config';
@@ -8,10 +9,7 @@ import { jsonError } from '@server/core/http';
import { findExistingJobForHead, insertJob, supersedeOlderJobs } from '@server/db/jobs';
import { recordWebhookDelivery } from '@server/db/webhook-deliveries';
-export function createWebhookRouter() {
- const app = new Hono();
-
- app.post('/', async (c) => {
+export async function handleGitHubWebhook(c: Context) {
const eventName = c.req.header('x-github-event');
const deliveryId = c.req.header('x-github-delivery');
const signature = c.req.header('x-hub-signature-256');
@@ -131,7 +129,12 @@ export function createWebhookRouter() {
});
return c.json({ ok: true, message: 'queued' }, 202);
- });
+}
+
+export function createWebhookRouter() {
+ const app = new Hono();
+
+ app.post('/', handleGitHubWebhook);
return app;
}
diff --git a/src/server/services/model.ts b/src/server/services/model.ts
index 649a744..a06b604 100644
--- a/src/server/services/model.ts
+++ b/src/server/services/model.ts
@@ -18,7 +18,7 @@ const MODEL_ALIASES: Record = {
'gemma-4-31b': 'gemma-4-31b-it',
'gemma-4-26b': 'gemma-4-26b-a4b-it',
};
-type ModelProvider = 'cloudflare';
+type ModelProvider = 'cloudflare' | 'google';
export class RetryableModelError extends Error {
readonly retryable = true;
@@ -44,6 +44,10 @@ function isCloudflareModel(model: string) {
return model.startsWith('@cf/');
}
+function getModelProvider(model: string): ModelProvider {
+ return isCloudflareModel(model) ? 'cloudflare' : 'google';
+}
+
function normalizeModel(model: string) {
return normalizeModelId(MODEL_ALIASES[model] ?? model);
}
@@ -209,7 +213,8 @@ export class ModelService {
let lastTransientError: unknown;
let sawTransientFailure = false;
for (const currentModel of modelsToTry) {
- if (isCloudflareModel(currentModel) && await this.isProviderUnavailable('cloudflare')) {
+ const provider = getModelProvider(currentModel);
+ if (provider === 'cloudflare' && await this.isProviderUnavailable('cloudflare')) {
logger.warn(`Skipping Cloudflare model ${currentModel} because Cloudflare AI allocation is unavailable for job ${this.options.jobId ?? 'unknown'}`);
continue;
}
diff --git a/src/server/worker-env.d.ts b/src/server/worker-env.d.ts
index e5014d6..664f4e6 100644
--- a/src/server/worker-env.d.ts
+++ b/src/server/worker-env.d.ts
@@ -1,34 +1,35 @@
/* eslint-disable */
-// Generated by Wrangler by running `wrangler types ./src/server/worker-env.d.ts` (hash: c82d96387e983e77c06e253e7ea4c4b1)
-// Runtime types generated with workerd@1.20260430.1 2026-04-16 nodejs_compat
+// Generated by Wrangler by running `wrangler types ./src/server/worker-env.d.ts` (hash: 2cf95a373086a6483897fb06140fae41)
+// Runtime types generated with workerd@1.20260521.1 2026-04-16 nodejs_compat
+interface __BaseEnv_Env {
+ APP_KV: KVNamespace;
+ HYPERDRIVE: Hyperdrive;
+ REVIEW_QUEUE: Queue;
+ AI: Ai;
+ ASSETS: Fetcher;
+ APP_URL: "https://app.codra.devarshi.dev";
+ AUTH_CALLBACK_URL: "https://app.codra.devarshi.dev/auth/github/callback";
+ BOT_USERNAME: "codra-app";
+ GITHUB_APP_SLUG: "codra-app-personal";
+ DASHBOARD_ALLOWED_USERS: "devarshishimpi";
+ ENVIRONMENT: "production";
+ CF_DLQ_ID: "ed6f5472cbd146f49ce94d3004eddb0f";
+ APP_PRIVATE_KEY: string;
+ GITHUB_APP_ID: string;
+ GITHUB_APP_WEBHOOK_SECRET: string;
+ GITHUB_CLIENT_ID: string;
+ GITHUB_CLIENT_SECRET: string;
+ GEMINI_API_KEY: string;
+ CF_API_TOKEN: string;
+ CF_ACCOUNT_ID: string;
+}
declare namespace Cloudflare {
interface GlobalProps {
mainModule: typeof import("./index");
}
- interface Env {
- APP_KV: KVNamespace;
- HYPERDRIVE: Hyperdrive;
- REVIEW_QUEUE: Queue;
- AI: Ai;
- ASSETS: Fetcher;
- APP_URL: "https://app.codra.devarshi.dev";
- AUTH_CALLBACK_URL: "https://app.codra.devarshi.dev/auth/github/callback";
- BOT_USERNAME: "codra-app";
- GITHUB_APP_SLUG: "codra-app-personal";
- DASHBOARD_ALLOWED_USERS: "devarshishimpi";
- ENVIRONMENT: "production";
- CF_DLQ_ID: "ed6f5472cbd146f49ce94d3004eddb0f";
- APP_PRIVATE_KEY: string;
- GITHUB_APP_ID: string;
- GITHUB_APP_WEBHOOK_SECRET: string;
- GITHUB_CLIENT_ID: string;
- GITHUB_CLIENT_SECRET: string;
- GEMINI_API_KEY: string;
- CF_API_TOKEN: string;
- CF_ACCOUNT_ID: string;
- }
+ interface Env extends __BaseEnv_Env {}
}
-interface Env extends Cloudflare.Env {}
+interface Env extends __BaseEnv_Env {}
type StringifyValues> = {
[Binding in keyof EnvType]: EnvType[Binding] extends string ? EnvType[Binding] : string;
};
@@ -941,7 +942,7 @@ interface CustomEventCustomEventInit {
* [MDN Reference](https://developer.mozilla.org/docs/Web/API/Blob)
*/
declare class Blob {
- constructor(type?: ((ArrayBuffer | ArrayBufferView) | string | Blob)[], options?: BlobOptions);
+ constructor(bits?: ((ArrayBuffer | ArrayBufferView) | string | Blob)[], options?: BlobOptions);
/**
* The **`size`** read-only property of the Blob interface returns the size of the Blob or File in bytes.
*
@@ -10155,12 +10156,17 @@ interface ArtifactsTokenListResult {
/** Total number of tokens for the repository. */
total: number;
}
-/** Handle for a single repository. Returned by Artifacts.get(). */
+/**
+ * Handle for a single repository. Returned by Artifacts.get().
+ *
+ * Methods may throw `ArtifactsError` with code `INTERNAL_ERROR` if an unexpected service error occurs.
+ */
interface ArtifactsRepo extends ArtifactsRepoInfo {
/**
* Create an access token for this repo.
* @param scope Token scope: "write" (default) or "read".
* @param ttl Time-to-live in seconds (default 86400, min 60, max 31536000).
+ * @throws {ArtifactsError} with code `INVALID_TTL` if ttl is out of range.
*/
createToken(scope?: 'write' | 'read', ttl?: number): Promise;
/** List tokens for this repo (metadata only, no plaintext). */
@@ -10169,6 +10175,7 @@ interface ArtifactsRepo extends ArtifactsRepoInfo {
* Revoke a token by plaintext or ID.
* @param tokenOrId Plaintext token or token ID.
* @returns true if revoked, false if not found.
+ * @throws {ArtifactsError} with code `INVALID_INPUT` if tokenOrId is empty.
*/
revokeToken(tokenOrId: string): Promise;
// ── Fork ──
@@ -10176,6 +10183,9 @@ interface ArtifactsRepo extends ArtifactsRepoInfo {
* Fork this repo to a new repo.
* @param name Target repository name.
* @param opts Optional: description, readOnly flag, defaultBranchOnly (default true).
+ * @throws {ArtifactsError} with code `INVALID_REPO_NAME` if name is invalid.
+ * @throws {ArtifactsError} with code `ALREADY_EXISTS` if the target repo already exists.
+ * @throws {ArtifactsError} with code `FORK_IN_PROGRESS` if a fork is already running.
*/
fork(name: string, opts?: {
description?: string;
@@ -10183,13 +10193,41 @@ interface ArtifactsRepo extends ArtifactsRepoInfo {
defaultBranchOnly?: boolean;
}): Promise;
}
-/** Artifacts binding — namespace-level operations. */
+// ── Error types ──────────────────────────────────────────────────────────────
+/**
+ * Error codes returned by Artifacts binding operations.
+ *
+ * Each code maps to a numeric code available on `ArtifactsError.numericCode`.
+ */
+type ArtifactsErrorCode = 'ALREADY_EXISTS' | 'NOT_FOUND' | 'IMPORT_IN_PROGRESS' | 'FORK_IN_PROGRESS' | 'INVALID_INPUT' | 'INVALID_REPO_NAME' | 'INVALID_TTL' | 'INVALID_URL' | 'REMOTE_AUTH_REQUIRED' | 'UPSTREAM_UNAVAILABLE' | 'MEMORY_LIMIT' | 'INTERNAL_ERROR';
+/**
+ * Error thrown by Artifacts binding operations.
+ *
+ * Uses a string `.code` discriminator following the Cloudflare platform
+ * convention (StreamError, ImagesError, etc.). The `.numericCode` matches
+ * the REST API `errors[].code` values.
+ */
+interface ArtifactsError extends Error {
+ readonly name: 'ArtifactsError';
+ /** String error code for programmatic matching. */
+ readonly code: ArtifactsErrorCode;
+ /** Numeric error code matching the REST API. */
+ readonly numericCode: number;
+}
+// ── Binding ──────────────────────────────────────────────────────────────────
+/**
+ * Artifacts binding — namespace-level operations.
+ *
+ * Methods may throw `ArtifactsError` with code `INTERNAL_ERROR` if an unexpected service error occurs.
+ */
interface Artifacts {
/**
* Create a new repository with an initial access token.
* @param name Repository name (alphanumeric, dots, hyphens, underscores).
* @param opts Optional: readOnly flag, description, default branch name.
* @returns Repo metadata with initial token.
+ * @throws {ArtifactsError} with code `INVALID_REPO_NAME` if name is invalid.
+ * @throws {ArtifactsError} with code `ALREADY_EXISTS` if the repo already exists.
*/
create(name: string, opts?: {
readOnly?: boolean;
@@ -10200,12 +10238,23 @@ interface Artifacts {
* Get a handle to an existing repository.
* @param name Repository name.
* @returns Repo handle.
+ * @throws {ArtifactsError} with code `NOT_FOUND` if the repo does not exist.
+ * @throws {ArtifactsError} with code `IMPORT_IN_PROGRESS` if the repo is still importing.
+ * @throws {ArtifactsError} with code `FORK_IN_PROGRESS` if the repo is still forking.
*/
get(name: string): Promise;
/**
* Import a repository from an external git remote.
* @param params Source URL and optional branch/depth, plus target name and options.
* @returns Repo metadata with initial token.
+ * @throws {ArtifactsError} with code `INVALID_REPO_NAME` if the target name is invalid.
+ * @throws {ArtifactsError} with code `INVALID_INPUT` if the source URL is not valid HTTPS.
+ * @throws {ArtifactsError} with code `INVALID_URL` if the source URL does not point to a git repository.
+ * @throws {ArtifactsError} with code `REMOTE_AUTH_REQUIRED` if the remote requires authentication.
+ * @throws {ArtifactsError} with code `NOT_FOUND` if the remote repository does not exist.
+ * @throws {ArtifactsError} with code `UPSTREAM_UNAVAILABLE` if the remote cannot be reached.
+ * @throws {ArtifactsError} with code `MEMORY_LIMIT` if the import exceeds service memory limits.
+ * @throws {ArtifactsError} with code `ALREADY_EXISTS` if the target repo already exists.
*/
import(params: {
source: {
@@ -10233,6 +10282,7 @@ interface Artifacts {
* Delete a repository and all associated tokens.
* @param name Repository name.
* @returns true if deleted, false if not found.
+ * @throws {ArtifactsError} with code `INVALID_REPO_NAME` if name is invalid.
*/
delete(name: string): Promise;
}
@@ -11332,11 +11382,11 @@ interface SendEmail {
send(message: EmailMessage): Promise;
send(builder: {
from: string | EmailAddress;
- to: string | string[];
+ to: string | EmailAddress | (string | EmailAddress)[];
subject: string;
replyTo?: string | EmailAddress;
- cc?: string | string[];
- bcc?: string | string[];
+ cc?: string | EmailAddress | (string | EmailAddress)[];
+ bcc?: string | EmailAddress | (string | EmailAddress)[];
headers?: Record;
text?: string;
html?: string;
@@ -12101,13 +12151,13 @@ declare namespace Cloudflare {
type GlobalProp = K extends keyof GlobalProps ? GlobalProps[K] : Default;
// The type of the program's main module exports, if known. Requires `GlobalProps` to declare the
// `mainModule` property.
- type MainModule = GlobalProp<'mainModule', {}>;
+ type MainModule = GlobalProp<"mainModule", {}>;
// The type of ctx.exports, which contains loopback bindings for all top-level exports.
type Exports = {
- [K in keyof MainModule]: LoopbackForExport &
+ [K in keyof MainModule]: LoopbackForExport
// If the export is listed in `durableNamespaces`, then it is also a
// DurableObjectNamespace.
- (K extends GlobalProp<'durableNamespaces', never> ? MainModule[K] extends new (...args: any[]) => infer DoInstance ? DoInstance extends Rpc.DurableObjectBranded ? DurableObjectNamespace : DurableObjectNamespace : DurableObjectNamespace : {});
+ & (K extends GlobalProp<"durableNamespaces", never> ? MainModule[K] extends new (...args: any[]) => infer DoInstance ? DoInstance extends Rpc.DurableObjectBranded ? DurableObjectNamespace : DurableObjectNamespace : DurableObjectNamespace : {});
};
}
declare namespace CloudflareWorkersModule {
@@ -12178,24 +12228,15 @@ declare namespace CloudflareWorkersModule {
attempt: number;
config: WorkflowStepConfig;
};
- export interface RollbackContext {
- error: Error;
- output: NonNullable | undefined;
- stepName: string;
- }
- export interface StepPromise extends Promise {
- rollback(fn: (ctx: RollbackContext) => Promise): StepPromise;
- rollback(config: WorkflowStepConfig, fn: (ctx: RollbackContext) => Promise): StepPromise;
- }
export abstract class WorkflowStep {
- do>(name: string, callback: (ctx: WorkflowStepContext) => Promise): StepPromise;
- do>(name: string, config: WorkflowStepConfig, callback: (ctx: WorkflowStepContext) => Promise): StepPromise;
+ do>(name: string, callback: (ctx: WorkflowStepContext) => Promise): Promise;
+ do>(name: string, config: WorkflowStepConfig, callback: (ctx: WorkflowStepContext) => Promise): Promise;
sleep: (name: string, duration: WorkflowSleepDuration) => Promise;
sleepUntil: (name: string, timestamp: Date | number) => Promise;
waitForEvent>(name: string, options: {
type: string;
timeout?: WorkflowTimeoutDuration | number;
- }): StepPromise>;
+ }): Promise>;
}
export type WorkflowInstanceStatus = 'queued' | 'running' | 'paused' | 'errored' | 'terminated' | 'complete' | 'waiting' | 'waitingForPause' | 'unknown';
export abstract class WorkflowEntrypoint | unknown = unknown> implements Rpc.WorkflowEntrypointBranded {
@@ -13179,6 +13220,9 @@ declare namespace TailStream {
// 1. This is an Onset event
// 2. We are not inheriting any SpanContext. (e.g. this is a cross-account service binding or a new top-level invocation)
readonly spanId?: string;
+ // W3C trace flags from an upstream traceparent. Absent when no upstream
+ // sampling decision was made.
+ readonly traceFlags?: number;
}
interface TailEvent {
// invocation id of the currently invoked worker stage.
@@ -13553,6 +13597,27 @@ interface WorkflowError {
code?: number;
message: string;
}
+interface WorkflowInstanceRestartOptions {
+ /**
+ * Restart from a specific step. If omitted, the instance restarts from the beginning.
+ * The step must exist in the instance's execution history.
+ */
+ from?: {
+ /**
+ * The step name as defined in your workflow code.
+ */
+ name: string;
+ /**
+ * 1-indexed occurrence of this step name. Use when the same step name appears multiple times (e.g. in a loop).
+ * @default 1
+ */
+ count?: number;
+ /**
+ * Step type filter. Use when different step types share the same name.
+ */
+ type?: 'do' | 'sleep' | 'waitForEvent';
+ };
+}
declare abstract class WorkflowInstance {
public id: string;
/**
@@ -13568,9 +13633,11 @@ declare abstract class WorkflowInstance {
*/
public terminate(): Promise;
/**
- * Restart the instance.
+ * Restart the instance. Optionally restart from a specific step, preserving
+ * cached results for all steps before it.
+ * @param options Options for the restart, including an optional step to restart from.
*/
- public restart(): Promise;
+ public restart(options?: WorkflowInstanceRestartOptions): Promise;
/**
* Returns the current status of the instance.
*/
diff --git a/src/shared/schema.ts b/src/shared/schema.ts
index 63bc503..e9cf80d 100644
--- a/src/shared/schema.ts
+++ b/src/shared/schema.ts
@@ -184,6 +184,8 @@ export const jobSummarySchema = z.object({
totalInputTokens: z.number().int(),
totalOutputTokens: z.number().int(),
createdAt: dateStringSchema,
+ updatedAt: dateStringSchema,
+ nextRetryAt: dateStringSchema.nullable().optional(),
startedAt: dateStringSchema.nullable(),
finishedAt: dateStringSchema.nullable(),
errorMessage: z.string().nullable(),
diff --git a/test/model-service.spec.ts b/test/model-service.spec.ts
index dc5aed9..72630e9 100644
--- a/test/model-service.spec.ts
+++ b/test/model-service.spec.ts
@@ -1,6 +1,7 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { isRetryableModelError, ModelService } from '@server/services/model';
import { reviewWithCloudflare } from '@server/models/cloudflare';
+import { reviewWithGoogle } from '@server/models/google';
import { createTestEnv } from './helpers';
import { defaultRepoConfig } from '@shared/schema';
@@ -50,7 +51,7 @@ describe('ModelService', () => {
});
});
- it('rejects Cloudflare reasoning-only responses instead of trying to parse the response envelope', async () => {
+ it('turns Cloudflare reasoning-only responses into inconclusive review JSON', async () => {
const env = createTestEnv({
AI: {
async run() {
@@ -70,15 +71,114 @@ describe('ModelService', () => {
} as any,
});
- await expect(
- reviewWithCloudflare(env, '@cf/moonshotai/kimi-k2.6', {
- systemPrompt: 'system',
- userPrompt: 'user',
- }),
- ).rejects.toThrow('returned no review content');
+ const response = await reviewWithCloudflare(env, '@cf/moonshotai/kimi-k2.6', {
+ systemPrompt: 'system',
+ userPrompt: 'user',
+ });
+ const parsed = JSON.parse(response.rawText);
+
+ expect(parsed.findings).toEqual([]);
+ expect(parsed.overall_correctness).toBe('patch is incorrect');
+ expect(parsed.overall_explanation).toContain('inconclusive');
+ });
+
+ it('does not parse Cloudflare reasoning as review JSON when final content is missing', async () => {
+ const env = createTestEnv({
+ AI: {
+ async run() {
+ return {
+ choices: [
+ {
+ message: {
+ content: null,
+ reasoning: 'Reasoning mentioned an object like {"foo":"bar"} but never produced final JSON.',
+ },
+ finish_reason: 'length',
+ },
+ ],
+ usage: { prompt_tokens: 1, completion_tokens: 8192 },
+ };
+ },
+ } as any,
+ });
+
+ const response = await reviewWithCloudflare(env, '@cf/zai-org/glm-4.7-flash', {
+ systemPrompt: 'system',
+ userPrompt: 'user',
+ });
+ const parsed = JSON.parse(response.rawText);
+
+ expect(parsed.findings).toEqual([]);
+ expect(parsed.overall_explanation).toContain('reasoning-only response');
+ });
+
+ it('asks Cloudflare chat models for strict review JSON', async () => {
+ let inputs: any;
+ const env = createTestEnv({
+ AI: {
+ async run(_model: string, request: any) {
+ inputs = request;
+ return {
+ choices: [
+ {
+ message: {
+ content: '{"findings":[],"overall_correctness":"patch is correct","overall_explanation":"ok","overall_confidence_score":0.9}',
+ },
+ },
+ ],
+ usage: { prompt_tokens: 1, completion_tokens: 1 },
+ };
+ },
+ } as any,
+ });
+
+ await reviewWithCloudflare(env, '@cf/zai-org/glm-4.7-flash', {
+ systemPrompt: 'system',
+ userPrompt: 'user',
+ });
+
+ expect(inputs.response_format).toMatchObject({
+ type: 'json_schema',
+ json_schema: {
+ name: 'codra_file_review',
+ strict: true,
+ },
+ });
+ expect(inputs.messages[0].content).toContain('Return only the JSON object');
+ expect(inputs.max_completion_tokens).toBe(8192);
+ expect(inputs.chat_template_kwargs).toBeUndefined();
+ expect(inputs.reasoning_effort).toBeUndefined();
});
- it('retries the same Cloudflare model once before failing it', async () => {
+ it('retries Google once for transient 524 edge timeouts', async () => {
+ const fetchMock = vi.spyOn(globalThis, 'fetch')
+ .mockResolvedValueOnce(
+ new Response(
+ JSON.stringify({ error: { code: 524, message: 'A timeout occurred.' } }),
+ { status: 524, headers: { 'content-type': 'application/json' } },
+ ),
+ )
+ .mockResolvedValueOnce(
+ new Response(
+ JSON.stringify({
+ candidates: [{ content: { parts: [{ text: '{"findings":[],"overall_correctness":"patch is correct","overall_explanation":"ok","overall_confidence_score":0.9}' }] } }],
+ usageMetadata: { promptTokenCount: 1, candidatesTokenCount: 1 },
+ }),
+ { status: 200, headers: { 'content-type': 'application/json' } },
+ ),
+ );
+
+ const response = await reviewWithGoogle(
+ { GEMINI_API_KEY: 'test-key' },
+ 'gemma-4-31b-it',
+ { systemPrompt: 'system', userPrompt: 'user' },
+ );
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(response.rawText).toContain('"findings"');
+ });
+
+ it('does not spend an extra queue slice retrying the same Cloudflare model inline', async () => {
let attempts = 0;
const env = createTestEnv({
AI: {
@@ -95,7 +195,80 @@ describe('ModelService', () => {
userPrompt: 'user',
}),
).rejects.toThrow('temporary provider error');
- expect(attempts).toBe(2);
+ expect(attempts).toBe(1);
+ });
+
+ it('tries the smaller Google fallback after the primary Google model fails', async () => {
+ let cloudflareCalls = 0;
+ const fetchMock = vi.spyOn(globalThis, 'fetch')
+ .mockResolvedValueOnce(
+ new Response(
+ JSON.stringify({
+ error: {
+ code: 500,
+ message: 'Internal error encountered.',
+ status: 'INTERNAL',
+ },
+ }),
+ { status: 500, headers: { 'content-type': 'application/json' } },
+ ),
+ )
+ .mockResolvedValueOnce(
+ new Response(
+ JSON.stringify({
+ candidates: [{ content: { parts: [{ text: '{"findings":[],"overall_correctness":"patch is correct","overall_explanation":"ok","overall_confidence_score":0.9}' }] } }],
+ usageMetadata: { promptTokenCount: 1, candidatesTokenCount: 1 },
+ }),
+ { status: 200, headers: { 'content-type': 'application/json' } },
+ ),
+ );
+ const env = createTestEnv({
+ AI: {
+ async run() {
+ cloudflareCalls++;
+ return {
+ response: JSON.stringify({
+ findings: [],
+ overall_correctness: 'patch is correct',
+ overall_explanation: 'ok',
+ overall_confidence_score: 0.9,
+ }),
+ usage: { prompt_tokens: 1, completion_tokens: 1 },
+ };
+ },
+ } as any,
+ GEMINI_API_KEY: 'test-key',
+ });
+ const service = new ModelService(env);
+
+ const response = await service.reviewFile({
+ file: {
+ path: 'src/app.ts',
+ lineCount: 1,
+ hunks: [],
+ isDeleted: false,
+ isBinary: false,
+ isNew: false,
+ previousPath: null,
+ },
+ prTitle: 'Test',
+ prDescription: null,
+ config: {
+ ...defaultRepoConfig,
+ model: {
+ main: 'gemma-4-31b-it',
+ fallbacks: ['gemma-4-26b-a4b-it', '@cf/zai-org/glm-4.7-flash'],
+ size_overrides: [],
+ },
+ },
+ totalLineCount: 1,
+ });
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(String(fetchMock.mock.calls[0][0])).toContain('/models/gemma-4-31b-it:generateContent');
+ expect(String(fetchMock.mock.calls[1][0])).toContain('/models/gemma-4-26b-a4b-it:generateContent');
+ expect(cloudflareCalls).toBe(0);
+ expect(response.modelUsed).toBe('gemma-4-26b-a4b-it');
});
it('marks exhausted transient provider failures as retryable for the queue', async () => {
@@ -260,7 +433,7 @@ describe('ModelService', () => {
const userPrompt = requestBody.contents[0].parts[0].text as string;
expect(fetchMock).toHaveBeenCalledOnce();
- expect(requestBody.generationConfig.maxOutputTokens).toBe(3072);
+ expect(requestBody.generationConfig.maxOutputTokens).toBe(4096);
expect(userPrompt).toContain('[NOTE: This diff has been truncated from 900 lines to 800 lines for brevity.]');
expect(userPrompt).toContain('const value799 = 799;');
expect(userPrompt).not.toContain('const value800 = 800;');
diff --git a/test/review-flow.spec.ts b/test/review-flow.spec.ts
index 7da7c31..dece307 100644
--- a/test/review-flow.spec.ts
+++ b/test/review-flow.spec.ts
@@ -63,7 +63,10 @@ vi.mock('@server/services/model', () => {
async generateSummary() {
return {
modelUsed: 'sum-model',
+ provider: 'google',
rawText: '{"summary": "test"}',
+ inputTokens: 3,
+ outputTokens: 2,
};
}
}
@@ -399,8 +402,83 @@ dbDescribe('Review Flow Lifecycle', () => {
reviewSpy.mockRestore();
}, REVIEW_FLOW_TIMEOUT_MS);
+ it('reviews files in a chunk concurrently', async () => {
+ const { GitHubService } = await import('@server/services/github');
+ const { ModelService } = await import('@server/services/model');
+ const repo = `test-repo-${Date.now()}-concurrent`;
+ const headSha = sha('8');
+ const baseSha = sha('9');
+ const getDiffSpy = vi.spyOn(GitHubService.prototype, 'getPullRequestDiff').mockResolvedValue(
+ generateMockDiff([
+ { path: 'src/one.ts', content: 'console.log(1);' },
+ { path: 'src/two.ts', content: 'console.log(2);' },
+ { path: 'src/three.ts', content: 'console.log(3);' },
+ ]),
+ );
+ let active = 0;
+ let maxActive = 0;
+ const reviewSpy = vi.spyOn(ModelService.prototype as any, 'reviewFile').mockImplementation(async (params: any) => {
+ active += 1;
+ maxActive = Math.max(maxActive, active);
+ await new Promise((resolve) => setTimeout(resolve, 25));
+ active -= 1;
+ return {
+ parsed: {
+ comments: [],
+ verdict: 'approve',
+ fileSummary: `Reviewed ${params.file.path}`,
+ overallCorrectness: 'no issues',
+ confidenceScore: 0.9,
+ },
+ modelUsed: 'test-model',
+ provider: 'test-provider',
+ inputTokens: 10,
+ outputTokens: 5,
+ rawText: '{}',
+ userPrompt: '',
+ };
+ });
+
+ const job = await insertJob(env, {
+ installationId: '123',
+ owner: 'test-owner',
+ repo,
+ prNumber: 6,
+ prTitle: 'Concurrent Test',
+ prAuthor: 'author',
+ commitSha: headSha,
+ baseSha,
+ trigger: 'auto',
+ headRef: 'feature',
+ baseRef: 'main',
+ configSnapshot: defaultRepoConfig,
+ });
+ await updateJobFileCount(env, job.id, 3);
+ await updateJobStep(env, job.id, 'Preparation', { status: 'done' });
+
+ await runWithDb(env, async () => {
+ (env.REVIEW_QUEUE as any).sent.length = 0;
+ const result = await runReviewJob(env, {
+ jobId: job.id,
+ deliveryId: 'delivery-concurrent',
+ phase: 'review',
+ });
+
+ expect(result).toEqual({ action: 'ack' });
+ expect(maxActive).toBe(3);
+ expect((env.REVIEW_QUEUE as any).sent[0]).toMatchObject({ jobId: job.id, phase: 'finalize' });
+ });
+
+ const reviews = await getFileReviewsForJobs(env, [job.id]);
+ expect(reviews.filter((review) => review.file_status === 'done')).toHaveLength(3);
+
+ reviewSpy.mockRestore();
+ getDiffSpy.mockRestore();
+ }, REVIEW_FLOW_TIMEOUT_MS);
+
it('marks completed jobs with skipped files as partial reviews', async () => {
const { GitHubService } = await import('@server/services/github');
+ const { ModelService } = await import('@server/services/model');
const repo = `test-repo-${Date.now()}-partial`;
const headSha = sha('e');
const baseSha = sha('f');
@@ -425,6 +503,7 @@ dbDescribe('Review Flow Lifecycle', () => {
baseRef: 'main',
configSnapshot: defaultRepoConfig,
});
+ const summarySpy = vi.spyOn(ModelService.prototype as any, 'generateSummary');
await updateJobFileCount(env, job.id, 2);
await updateJobStep(env, job.id, 'Preparation', { status: 'done' });
await updateJobStep(env, job.id, 'Reviewing Files', { status: 'done' });
@@ -474,6 +553,10 @@ dbDescribe('Review Flow Lifecycle', () => {
const finalJob = await getJobForProcessing(env, job.id);
expect(finalJob?.status).toBe('done');
expect(finalJob?.error_msg).toContain('Partial review: 1 of 2 files');
+ expect(finalJob?.summary_markdown).toMatch(/^### Codra Review/);
+ expect(finalJob?.summary_model).toBeNull();
+ expect(summarySpy).not.toHaveBeenCalled();
+ summarySpy.mockRestore();
getDiffSpy.mockRestore();
}, REVIEW_FLOW_TIMEOUT_MS);
});
diff --git a/test/webhook-handling.spec.ts b/test/webhook-handling.spec.ts
index 66eb5ff..965e603 100644
--- a/test/webhook-handling.spec.ts
+++ b/test/webhook-handling.spec.ts
@@ -119,6 +119,38 @@ describe('Webhook Handling Suite', () => {
expect(queue.sent[0].payload).toBeUndefined();
});
+ it('rejects GitHub webhooks posted to the site root', async () => {
+ const repoName = `root-repo-${Date.now()}`;
+ const rawPayload = createMockPRWebhook({
+ action: 'opened',
+ repository: { name: repoName, owner: { login: 'test-owner' } }
+ });
+ rawPayload.pull_request.head.sha = 'c'.repeat(40);
+ rawPayload.pull_request.base.sha = 'd'.repeat(40);
+ const body = JSON.stringify(rawPayload);
+ const signature = await signPayload(env.GITHUB_APP_WEBHOOK_SECRET, body);
+
+ const response = await app.request(
+ 'http://codra.test/',
+ {
+ method: 'POST',
+ headers: {
+ 'x-github-event': 'pull_request',
+ 'x-github-delivery': `root-delivery-${Date.now()}`,
+ 'x-hub-signature-256': signature,
+ 'content-type': 'application/json',
+ },
+ body,
+ },
+ env,
+ );
+
+ expect(response.status).toBe(404);
+
+ const queue = env.REVIEW_QUEUE as any;
+ expect(queue.sent).toHaveLength(0);
+ });
+
it('acknowledges unsupported GitHub events without queueing review work', async () => {
const rawPayload = createMockPRWebhook({
action: 'opened',