diff --git a/.env.example b/.env.example index 9099404..098b2a9 100644 --- a/.env.example +++ b/.env.example @@ -104,6 +104,9 @@ GITHUB_APP_CALLBACK_URL=https://fledge.example/auth/github/app/callback # Inbound email replies (daily digest Reply-To); shared secret with the Email Worker INBOUND_EMAIL_SECRET= +# Fail a running job after this many minutes without a callback/heartbeat. +JOB_STALL_MINUTES=30 + # Optional outbound notification webhook (Slack-compatible text plus event fields). NOTIFY_WEBHOOK_URL= NOTIFY_WEBHOOK_SECRET= diff --git a/changelog.d/job-health.fixed.md b/changelog.d/job-health.fixed.md new file mode 100644 index 0000000..f41cab0 --- /dev/null +++ b/changelog.d/job-health.fixed.md @@ -0,0 +1 @@ +Running jobs now fail after `JOB_STALL_MINUTES` (default 30) without a callback instead of blocking the repository for the full 90-minute timeout. A repository token that returns 401/403 marks the connection invalid, pauses auto-iterate and shows a re-authorize banner; three consecutive failures pause auto-iterate until the owner re-enables it. diff --git a/docs/architecture.md b/docs/architecture.md index 3ceb990..cad575e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -280,6 +280,20 @@ job is retried, so a new job can continue from the last known point instead of starting from scratch. The add/settings goal forms also offer four task templates (bug, tests, docs, dependency) that fill the goal text. +### Job health and credential circuit breaker + +Every callback records `jobs.last_callback_at`. A running job with no callback for +`JOB_STALL_MINUTES` (default 30) fails with a `stalled` message instead of holding +the repository slot until the 90-minute hard timeout. A successful job resets the +repository failure counter; three consecutive failures pause auto-iterate and +notify the owner. + +An iterate preflight (and a cached check for plan/review jobs) reads the target +branch with the repository token. A 401/403 marks `repos.credential_state = +'invalid'`, pauses auto-iterate, and notifies the owner with a re-authorize link. +The repository page shows the credential banner until a refresh or a successful +preflight clears it. + ### Model connections Settings owns a per-user catalog of OpenAI-compatible model connections diff --git a/docs/deploy.md b/docs/deploy.md index 4a48aeb..b9f49af 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -40,6 +40,7 @@ Key settings: | `INITIAL_QUOTA_CNY` | Default 1 | | `USD_CNY_RATE` | Default 7.2 | | `RUNNER_MODE` | `mock` or `flybot` | +| `JOB_TIMEOUT_MINUTES` / `JOB_STALL_MINUTES` | Hard timeout (default 90) and no-heartbeat stall timeout (default 30) | | `FLYBOT_BRIDGE_URL` / `FLYBOT_BRIDGE_TOKEN` | Bridge endpoint and shared secret | | `DATA_DIR` / `DB_PATH` | Defaults `./data`, `./data/fledge.db` | | `ALLOW_PRIVATE_MODEL_URLS` | Keep `false` for untrusted users | diff --git a/src/config.ts b/src/config.ts index 709c23e..c00a88e 100644 --- a/src/config.ts +++ b/src/config.ts @@ -53,6 +53,7 @@ export interface Config { bridgeUrl: string bridgeToken: string timeoutMinutes: number + stallMinutes: number } digest: { hour: number @@ -190,6 +191,7 @@ export function loadConfig(overrides: Partial = {}): Config { bridgeUrl: text('FLYBOT_BRIDGE_URL', 'http://127.0.0.1:8787/run'), bridgeToken: text('FLYBOT_BRIDGE_TOKEN'), timeoutMinutes: number('JOB_TIMEOUT_MINUTES', 90), + stallMinutes: number('JOB_STALL_MINUTES', 30), }, digest: { hour: Math.min(23, Math.max(0, number('DIGEST_HOUR', 9))), diff --git a/src/db.ts b/src/db.ts index ebf52f7..695c790 100644 --- a/src/db.ts +++ b/src/db.ts @@ -83,6 +83,8 @@ export interface RepoRow { require_protection: number stage_mode: string models_dirty: number + credential_state: string + consecutive_failures: number monthly_budget_cny: number job_budget_cny: number protection_state: string | null @@ -187,6 +189,7 @@ export interface JobRow { base_sha: string | null created_at: string started_at: string | null + last_callback_at: string | null finished_at: string | null } @@ -353,6 +356,8 @@ CREATE TABLE IF NOT EXISTS repos ( require_protection INTEGER NOT NULL DEFAULT 1, stage_mode TEXT NOT NULL DEFAULT 'off', models_dirty INTEGER NOT NULL DEFAULT 0, + credential_state TEXT NOT NULL DEFAULT 'ok', + consecutive_failures INTEGER NOT NULL DEFAULT 0, monthly_budget_cny REAL NOT NULL DEFAULT 0, job_budget_cny REAL NOT NULL DEFAULT 0, protection_state TEXT, @@ -441,6 +446,7 @@ CREATE TABLE IF NOT EXISTS jobs ( base_sha TEXT, created_at TEXT NOT NULL, started_at TEXT, + last_callback_at TEXT, finished_at TEXT ); CREATE INDEX IF NOT EXISTS jobs_repo_idx ON jobs(repo_id, created_at DESC); @@ -586,9 +592,9 @@ export class Store { this.db.exec('PRAGMA foreign_keys = ON;') this.db.exec('PRAGMA busy_timeout = 5000;') const previousVersion = Number(this.get<{ user_version: number }>('PRAGMA user_version')?.user_version ?? 0) - if (previousVersion > 22) { + if (previousVersion > 23) { this.db.close() - throw new Error(`database schema version ${previousVersion} is newer than this build supports (22)`) + throw new Error(`database schema version ${previousVersion} is newer than this build supports (23)`) } this.db.exec(SCHEMA) this.migrateColumns() @@ -600,7 +606,7 @@ export class Store { if (previousVersion < 16) { this.run('UPDATE repos SET role_checked_at = ? WHERE repo_role IS NOT NULL AND role_checked_at IS NULL', nowIso()) } - this.db.exec(`PRAGMA user_version = ${Math.max(previousVersion, 22)};`) + this.db.exec(`PRAGMA user_version = ${Math.max(previousVersion, 23)};`) } private migrateColumns(): void { @@ -642,6 +648,8 @@ export class Store { ['require_protection', 'INTEGER NOT NULL DEFAULT 1'], ['stage_mode', "TEXT NOT NULL DEFAULT 'off'"], ['models_dirty', 'INTEGER NOT NULL DEFAULT 0'], + ['credential_state', "TEXT NOT NULL DEFAULT 'ok'"], + ['consecutive_failures', 'INTEGER NOT NULL DEFAULT 0'], ['monthly_budget_cny', 'REAL NOT NULL DEFAULT 0'], ['job_budget_cny', 'REAL NOT NULL DEFAULT 0'], ['protection_state', 'TEXT'], @@ -691,6 +699,7 @@ export class Store { if (!jobColumns.has('plan_allow_paths')) this.run('ALTER TABLE jobs ADD COLUMN plan_allow_paths TEXT') if (!jobColumns.has('plan_deny_paths')) this.run('ALTER TABLE jobs ADD COLUMN plan_deny_paths TEXT') if (!jobColumns.has('base_sha')) this.run('ALTER TABLE jobs ADD COLUMN base_sha TEXT') + if (!jobColumns.has('last_callback_at')) this.run('ALTER TABLE jobs ADD COLUMN last_callback_at TEXT') const tokenColumns = new Set(this.all<{ name: string }>('PRAGMA table_info(api_tokens)').map((row) => row.name)) if (!tokenColumns.has('repo_id')) this.run('ALTER TABLE api_tokens ADD COLUMN repo_id TEXT') const ledgerColumns = new Set(this.all<{ name: string }>('PRAGMA table_info(quota_ledger)').map((row) => row.name)) @@ -1236,7 +1245,7 @@ export class Store { return this.get('SELECT * FROM repos WHERE user_id = ? AND lower(owner) = lower(?) AND lower(name) = lower(?)', userId, owner, name) } - updateRepo(id: string, patch: Partial>): void { + updateRepo(id: string, patch: Partial>): void { const entries = Object.entries(patch).filter(([, value]) => value !== undefined) if (entries.length === 0) return const sets = entries.map(([key]) => `${key} = ?`).join(', ') @@ -1435,7 +1444,7 @@ export class Store { return this.all("SELECT * FROM jobs WHERE status = 'queued' ORDER BY priority DESC, created_at ASC") } - updateJob(id: string, patch: Partial>): void { + updateJob(id: string, patch: Partial>): void { const entries = Object.entries(patch).filter(([, value]) => value !== undefined) if (entries.length === 0) return const sets = entries.map(([key]) => `${key} = ?`).join(', ') @@ -1451,6 +1460,29 @@ export class Store { return this.all("SELECT * FROM jobs WHERE status = 'running' AND COALESCE(started_at, created_at) < ?", beforeIso) } + stalledRunningJobs(beforeIso: string): JobRow[] { + return this.all("SELECT * FROM jobs WHERE status = 'running' AND COALESCE(last_callback_at, started_at, created_at) < ?", beforeIso) + } + + touchJobCallback(jobId: string): void { + this.run('UPDATE jobs SET last_callback_at = ? WHERE id = ?', nowIso(), jobId) + } + + markRepoCredential(repoId: string, state: 'ok' | 'invalid'): void { + this.run('UPDATE repos SET credential_state = ? WHERE id = ?', state, repoId) + } + + incrementRepoFailures(repoId: string): number { + return this.transaction(() => { + this.run('UPDATE repos SET consecutive_failures = consecutive_failures + 1 WHERE id = ?', repoId) + return Number(this.get<{ n: number }>('SELECT consecutive_failures AS n FROM repos WHERE id = ?', repoId)?.n ?? 0) + }) + } + + resetRepoFailures(repoId: string): void { + this.run('UPDATE repos SET consecutive_failures = 0 WHERE id = ?', repoId) + } + staleQueuedJobs(beforeIso: string): JobRow[] { return this.all("SELECT * FROM jobs WHERE status = 'queued' AND created_at < ?", beforeIso) } diff --git a/src/jobs/service.ts b/src/jobs/service.ts index 082ef9a..3fea997 100644 --- a/src/jobs/service.ts +++ b/src/jobs/service.ts @@ -13,7 +13,7 @@ import { parseEvidence, type Evidence } from '../lib/evidence.js' import { checkStage, parseStageMode } from '../lib/stages.js' import { decryptSecret, randomId, randomToken, safeEqual, sha256Hex } from '../lib/secrets.js' import { addMinutes, nowIso } from '../lib/time.js' -import { GithubClient, prepareRepo, prepareRepoWithInstallation, repoPermissionWithToken } from '../repos/github.js' +import { GithubClient, GithubError, prepareRepo, prepareRepoWithInstallation, repoPermissionWithToken } from '../repos/github.js' import { GithubApp } from '../repos/app.js' import { repoGithubToken } from '../repos/token.js' import { parseTeam } from '../flybot/instances.js' @@ -73,6 +73,7 @@ export class Jobs { private readonly followUpLocks = new Set() private ticking = false private lastPruneAt = 0 + private readonly credentialCache = new Map() private readonly quotaNotifier: QuotaNotifier constructor( @@ -203,6 +204,8 @@ export class Jobs { const upstreamOwner = repo.upstream_owner || repo.owner const upstreamName = repo.upstream_name || repo.name const branch = await client.getBranch(upstreamOwner, upstreamName, baseBranch) + this.store.markRepoCredential(repo.id, 'ok') + this.credentialCache.set(repo.id, { at: Date.now() }) if (branch === null) return { ok: false, error: `目标分支 ${baseBranch} 不存在或不可访问。`, code: 'BRANCH_MISSING' } let warning = '' if (repo.is_fork === 1) { @@ -216,10 +219,46 @@ export class Jobs { } return { ok: true, baseSha: branch.sha, warning } } catch (error) { + if (error instanceof GithubError && (error.status === 401 || error.status === 403) && !/rate limit/i.test(error.message)) { + this.markCredentialInvalid(repo, error) + return { ok: false, error: 'GitHub token 权限不足，请重新授权这个仓库。', code: 'REPO_CREDENTIAL_INVALID' } + } return { ok: false, error: String(error instanceof Error ? error.message : error), code: 'PREFLIGHT_FAILED' } } } + private async checkCredential(repo: RepoRow): Promise<{ ok: boolean; code?: string; error?: string }> { + const cached = this.credentialCache.get(repo.id) + if (cached !== undefined && Date.now() - cached.at < 10 * 60_000) return { ok: true } + try { + const token = await repoGithubToken(repo, this.config, this.githubApp, this.store) + const client = new GithubClient(token, this.config.github.apiBase) + const owner = repo.upstream_owner || repo.owner + const name = repo.upstream_name || repo.name + const branch = await client.getBranch(owner, name, repo.upstream_default_branch || repo.default_branch) + if (branch !== null) this.credentialCache.set(repo.id, { at: Date.now() }) + this.store.markRepoCredential(repo.id, 'ok') + return { ok: true } + } catch (error) { + if (error instanceof GithubError && (error.status === 401 || error.status === 403) && !/rate limit/i.test(error.message)) { + this.markCredentialInvalid(repo, error) + return { ok: false, code: 'REPO_CREDENTIAL_INVALID', error: 'GitHub token 权限不足，请重新授权这个仓库。' } + } + return { ok: true } + } + } + + private markCredentialInvalid(repo: RepoRow, error: unknown): void { + this.credentialCache.delete(repo.id) + this.store.markRepoCredential(repo.id, 'invalid') + const current = this.store.getRepo(repo.id) + if (current !== undefined && current.auto_iterate === 1) this.store.updateRepo(repo.id, { auto_iterate: 0 }) + const message = error instanceof Error ? error.message : String(error) + this.store.audit(repo.user_id, 'repo.credential_invalid', `${repo.owner}/${repo.name}: ${message.slice(0, 200)}`) + this.store.notify(repo.user_id, 'job', `${repo.owner}/${repo.name} 的 GitHub token 权限不足，自动迭代已暂停；请重新授权后手动开启。`) + void sendNotificationWebhook(this.config, 'repo.credential_invalid', { text: `${repo.owner}/${repo.name}: GitHub token is invalid`, repoId: repo.id, error: message.slice(0, 300) }) + } + async enqueue(repo: RepoRow, trigger: 'manual' | 'schedule' | 'followup', goalOverride?: string, kind: JobKind = 'iterate', options: { branch?: string; prUrl?: string; baseBranch?: string; workItemId?: string; priority?: number } = {}): Promise { if (repo.enabled !== 1) return { ok: false, error: '这个仓库已停用', code: 'REPO_DISABLED' } if (repo.repo_role !== 'maintainer' && repo.repo_role !== 'developer') { @@ -240,6 +279,10 @@ export class Jobs { if (active !== undefined) return { ok: false, error: '这个仓库已有任务在运行', code: 'JOB_ACTIVE' } const user = this.store.userById(repo.user_id) if (user === undefined || user.status !== 'approved') return { ok: false, error: '用户不可用', code: 'USER_UNAVAILABLE' } + if (kind !== 'iterate') { + const credential = await this.checkCredential(repo) + if (!credential.ok) return { ok: false, error: credential.error ?? 'GitHub token 权限不足。', code: credential.code ?? 'REPO_CREDENTIAL_INVALID' } + } if (repo.monthly_budget_cny > 0) { const since = new Date(Date.now() - 30 * 86_400_000).toISOString() const spentCny = this.store.repoUsageSince(repo.id, since) * this.config.quota.usdCny @@ -479,8 +522,14 @@ export class Jobs { this.store.notify(job.user_id, 'quota', `任务 ${jobId} 成本 ${jobCostCny.toFixed(2)} 元，超过单任务预算 ${repo.job_budget_cny.toFixed(2)} 元。`) } } - if (result.status === 'succeeded') this.store.deleteSetting(`ops:repo_failing:${job.repo_id}`) - if (result.status === 'failed') this.checkRepeatedFailures(job) + if (result.status === 'succeeded') { + this.store.deleteSetting(`ops:repo_failing:${job.repo_id}`) + this.store.resetRepoFailures(job.repo_id) + } + if (result.status === 'failed') { + this.store.incrementRepoFailures(job.repo_id) + this.checkRepeatedFailures(job) + } void this.captureTeamStats(jobId) } @@ -492,9 +541,13 @@ export class Jobs { this.store.setSetting(key, nowIso()) const repo = this.store.getRepo(job.repo_id) const label = repo === undefined ? job.repo_id : `${repo.upstream_owner || repo.owner}/${repo.upstream_name || repo.name}` - this.store.notify(job.user_id, 'job', `${label} 连续 3 个任务失败，请检查 bridge、GitHub 凭据和模型额度。`) + if (repo !== undefined && repo.auto_iterate === 1) { + this.store.updateRepo(repo.id, { auto_iterate: 0 }) + this.store.audit(job.user_id, 'repo.auto_paused', `${label}: 3 consecutive failures`) + } + this.store.notify(job.user_id, 'job', `${label} 连续 3 个任务失败，自动迭代已暂停；请检查 bridge、GitHub 凭据和模型额度后手动重新开启。`) this.store.audit(job.user_id, 'ops.repeated_failures', `${label}: 3 consecutive failures`) - void sendNotificationWebhook(this.config, 'repo.repeated_failures', { text: `${label}: 3 consecutive job failures`, repoId: job.repo_id, jobId: job.id }) + void sendNotificationWebhook(this.config, 'repo.repeated_failures', { text: `${label}: 3 consecutive job failures; auto-iterate paused`, repoId: job.repo_id, jobId: job.id }) } private async refreshRepoAccess(repo: RepoRow): Promise { @@ -987,6 +1040,7 @@ export class Jobs { if (job.status !== 'queued' && job.status !== 'running') return { ok: false, status: 409, error: 'job already finished' } const status = payload.status if (status !== 'succeeded' && status !== 'failed' && status !== 'running') return { ok: false, status: 400, error: 'invalid status' } + this.store.touchJobCallback(jobId) const branch = typeof payload.branch === 'string' ? payload.branch.slice(0, 300) : undefined const prUrl = typeof payload.prUrl === 'string' && payload.prUrl.startsWith('https://') ? payload.prUrl.slice(0, 500) : undefined const issueUrl = typeof payload.issueUrl === 'string' && payload.issueUrl.startsWith('https://') ? payload.issueUrl.slice(0, 500) : undefined @@ -1153,8 +1207,18 @@ export class Jobs { } private async tickOnce(): Promise { + const handled = new Set() + const stalledBefore = addMinutes(-this.config.runner.stallMinutes) + for (const job of this.store.stalledRunningJobs(stalledBefore)) { + handled.add(job.id) + const revoke = (): void => { + const row = job.platform_token_hash === null ? undefined : this.store.findApiToken(job.platform_token_hash) + if (row !== undefined) this.store.revokeApiToken(row.id, row.user_id) + } + this.finish(job.id, { status: 'failed', error: `任务停滞：超过 ${this.config.runner.stallMinutes} 分钟没有心跳。` }, revoke) + } const staleBefore = addMinutes(-this.config.runner.timeoutMinutes) - const staleJobs = [...this.store.staleQueuedJobs(addMinutes(-10)), ...this.store.staleRunningJobs(staleBefore)] + const staleJobs = [...this.store.staleQueuedJobs(addMinutes(-10)), ...this.store.staleRunningJobs(staleBefore)].filter((job) => !handled.has(job.id)) for (const job of staleJobs) { const revoke = (): void => { const row = job.platform_token_hash === null ? undefined : this.store.findApiToken(job.platform_token_hash) diff --git a/src/views/pages.ts b/src/views/pages.ts index 4a84e9c..5ed7ff2 100644 --- a/src/views/pages.ts +++ b/src/views/pages.ts @@ -652,7 +652,10 @@ export function repoPage(input: { config: Config; locale: Locale; viewer: Viewer
${workRows}
${escapeHtml(t('工作项', 'Work item'))}${escapeHtml(t('状态', 'State'))}${escapeHtml(t('优先级', 'Priority'))}Issue / PR${escapeHtml(t('任务', 'Job'))}
`; - const body = `${repoHeader}${roleCard}${appAttribution ? `` : ''}${questionsBlock(locale, viewer, questions, `/repos/${repo.id}`)}${chiefCard}${instanceNotice}${instanceCard}${workCard} + const credentialBanner = repo.credential_state === 'invalid' || repo.consecutive_failures >= 3 + ? `` + : '' + const body = `${repoHeader}${roleCard}${credentialBanner}${appAttribution ? `` : ''}${questionsBlock(locale, viewer, questions, `/repos/${repo.id}`)}${chiefCard}${instanceNotice}${instanceCard}${workCard}

${escapeHtml(t('迭代目标', 'Iteration goal'))}

@@ -711,7 +714,7 @@ export function repoPage(input: { config: Config; locale: Locale; viewer: Viewer ${repo.models_dirty === 1 ? `
${csrf(viewer)}

${escapeHtml(t('有运行中的任务时会拒绝重启。', 'Restart is refused while a job is running.'))}

` : ''}
-
+

${escapeHtml(t('连接', 'Connection'))}

${escapeHtml(t('GitHub token 已加密保存；页面不会显示明文。如果仓库或 token 变了，删除后重新添加。', 'The GitHub token is encrypted and never rendered in plain text. If the repository or token changes, delete and add it again.'))}

${escapeHtml(t('审核最近运行', 'Last review'))} ${escapeHtml(repo.last_review_at === null ? t('尚未运行', 'Not run yet') : humanTimeShort(repo.last_review_at))}${reviewStatus}

@@ -1457,7 +1460,7 @@ export function jobPage(input: {
${escapeHtml(t('任务', 'JOB'))}

${escapeHtml(job.id)}

-

${escapeHtml(repo.upstream_owner || repo.owner)}/${escapeHtml(repo.upstream_name || repo.name)} · ${badge(t, job.status)} · ${escapeHtml(jobKindLabel(t, job))} · ${escapeHtml(triggerLabel(t, job.trigger))} · ${escapeHtml(t('目标分支', 'Base branch'))} ${escapeHtml(job.base_branch ?? (repo.target_branch !== '' ? repo.target_branch : (repo.upstream_default_branch || repo.default_branch)))}${job.base_sha === null ? '' : ` · ${escapeHtml(t('基于', 'Based on'))} ${escapeHtml(job.base_sha.slice(0, 12))}`}${queuePosition === undefined ? '' : ` · ${escapeHtml(t('队列位置', 'Queue position'))} ${queuePosition}`}

+

${escapeHtml(repo.upstream_owner || repo.owner)}/${escapeHtml(repo.upstream_name || repo.name)} · ${badge(t, job.status)} · ${escapeHtml(jobKindLabel(t, job))} · ${escapeHtml(triggerLabel(t, job.trigger))} · ${escapeHtml(t('目标分支', 'Base branch'))} ${escapeHtml(job.base_branch ?? (repo.target_branch !== '' ? repo.target_branch : (repo.upstream_default_branch || repo.default_branch)))}${job.base_sha === null ? '' : ` · ${escapeHtml(t('基于', 'Based on'))} ${escapeHtml(job.base_sha.slice(0, 12))}`}${job.last_callback_at === null ? '' : ` · ${escapeHtml(t('最后心跳', 'Last heartbeat'))} ${escapeHtml(humanTimeShort(job.last_callback_at))}`}${queuePosition === undefined ? '' : ` · ${escapeHtml(t('队列位置', 'Queue position'))} ${queuePosition}`}

${escapeHtml(t('返回仓库', 'Back to repository'))} diff --git a/test/db.test.ts b/test/db.test.ts index 6b22f22..ec6a6f5 100644 --- a/test/db.test.ts +++ b/test/db.test.ts @@ -150,7 +150,7 @@ test('the role backfill runs once and future schema versions are refused', () => const reopened = new Store(path) assert.equal(reopened.getRepo(repo.id)?.repo_role, 'maintainer') assert.notEqual(reopened.getRepo(repo.id)?.role_checked_at, null) - reopened.db.exec('PRAGMA user_version = 23;') + reopened.db.exec('PRAGMA user_version = 24;') reopened.close() assert.throws(() => new Store(path), /newer than this build/) @@ -255,3 +255,24 @@ test('user model connections and repository selections persist', () => { store.close() } }) + +test('stalled jobs and repository failure counters are persisted', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'stall-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_stall', userId: user.id, owner: 'o', name: 's', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: true, intervalMinutes: 1440 }) + store.createJob({ id: 'job_stall', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb', platformTokenHash: '', kind: 'iterate' }) + store.updateJob('job_stall', { status: 'running', started_at: '2000-01-01T00:00:00.000Z' }) + assert.equal(store.stalledRunningJobs(new Date().toISOString()).length, 1) + store.touchJobCallback('job_stall') + assert.notEqual(store.getJob('job_stall')?.last_callback_at, null) + assert.equal(store.incrementRepoFailures(repo.id), 1) + assert.equal(store.incrementRepoFailures(repo.id), 2) + store.resetRepoFailures(repo.id) + assert.equal(store.getRepo(repo.id)?.consecutive_failures, 0) + store.markRepoCredential(repo.id, 'invalid') + assert.equal(store.getRepo(repo.id)?.credential_state, 'invalid') + } finally { + store.close() + } +}) diff --git a/test/db.test.ts.bak b/test/db.test.ts.bak new file mode 100644 index 0000000..6b22f22 --- /dev/null +++ b/test/db.test.ts.bak @@ -0,0 +1,257 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Store } from '../src/db.js' +import { addDays } from '../src/lib/time.js' +import { makeUser } from './helpers.js' + +test('users, repos, jobs, tokens and usage persist in SQLite', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'octo', status: 'pending' }) + assert.equal(user.status, 'pending') + assert.equal(user.quota_cny, 1) + assert.equal(user.email_job_updates, 1) + assert.equal(user.email_daily_digest, 1) + assert.equal(user.notify_email, null) + store.setNotifyEmail(user.id, 'me@example.com') + store.setEmailDailyDigest(user.id, false) + assert.equal(store.userById(user.id)?.notify_email, 'me@example.com') + assert.equal(store.userById(user.id)?.email_daily_digest, 0) + store.setEmailDailyDigest(user.id, true) + store.setEmailJobUpdates(user.id, false) + assert.equal(store.userById(user.id)?.email_job_updates, 0) + store.setEmailJobUpdates(user.id, true) + store.setUserStatus(user.id, 'approved', 'test') + assert.equal(store.userById(user.id)?.status, 'approved') + store.markApprovalSent(user.id) + assert.notEqual(store.userById(user.id)?.approval_sent_at, null) + + const repo = store.createRepo({ + id: 'repo_1', + userId: user.id, + owner: 'octo', + name: 'demo', + defaultBranch: 'main', + tokenEnc: 'v1:test', + goal: 'fix bugs', + autoIterate: true, + intervalMinutes: 60, + }) + assert.equal(store.listRepos(user.id).length, 1) + assert.equal(store.getRepoByOwnerName(user.id, 'OCTO', 'DEMO')?.id, repo.id) + + const job = store.createJob({ + id: 'job_1', + repoId: repo.id, + userId: user.id, + trigger: 'manual', + goal: 'fix bugs', + callbackHash: 'hash', + platformTokenHash: 'tokenhash', + }) + assert.equal(job.status, 'queued') + assert.equal(store.activeJobForRepo(repo.id)?.id, job.id) + store.updateJob(job.id, { status: 'running', started_at: new Date().toISOString() }) + assert.equal(store.countActiveJobs(), 1) + store.updateJob(job.id, { status: 'succeeded', finished_at: new Date().toISOString() }) + assert.equal(store.countActiveJobs(), 0) + + store.createApiToken({ + id: 'tok_1', + userId: user.id, + kind: 'pat', + name: 'cli', + tokenHash: 'abc', + prefix: 'flg_abcd', + expiresAt: null, + }) + assert.equal(store.findApiToken('abc')?.name, 'cli') + assert.equal(store.listApiTokens(user.id).length, 1) + + store.recordUsage(user.id, job.id, 0.02) + assert.equal(store.userById(user.id)?.quota_used_usd, 0.02) + assert.equal(store.usageTotal(user.id), 0.02) + + store.createApprovalToken('approvehash', user.id, 'approve', addDays(1)) + assert.equal(store.findApprovalToken('approvehash')?.action, 'approve') + store.useApprovalToken('approvehash') + assert.equal(store.findApprovalToken('approvehash'), undefined) + + store.notify(user.id, 'test', 'hello') + assert.equal(store.listNotifications(user.id).length, 1) + store.markNotificationsRead(user.id) + assert.equal(store.listNotifications(user.id)[0]?.read_at !== null, true) + } finally { + store.close() + } +}) + +test('events and product metrics are recorded', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'metrics-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_metrics', userId: user.id, owner: 'o', name: 'm', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + const job = store.createJob({ id: 'job_metrics', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb', platformTokenHash: '' }) + store.updateJob(job.id, { status: 'succeeded', pr_url: 'https://github.com/o/m/pull/1', finished_at: new Date().toISOString() }) + store.event('job_enqueued', { userId: user.id, repoId: repo.id, jobId: job.id }) + store.event('pr_open', { userId: user.id, repoId: repo.id, jobId: job.id }) + assert.equal(store.countEvents('job_enqueued'), 1) + const metrics = store.productMetrics(new Date(0).toISOString()) + assert.equal(metrics.users, 1) + assert.equal(metrics.approved, 1) + assert.equal(metrics.repos, 1) + assert.equal(metrics.jobs, 1) + assert.equal(metrics.succeeded, 1) + assert.equal(metrics.usersWithPr, 1) + assert.equal(metrics.events.find((event) => event.name === 'pr_open')?.n, 1) + store.updateJob(job.id, { team_stats_json: JSON.stringify([{ role: 'lead', calls: 12 }, { role: 'test', calls: 3 }]) }) + const team = store.teamMetrics(new Date(0).toISOString()) + assert.equal(team.jobs, 1) + assert.equal(team.avgRoles, 2) + assert.equal(team.leadOnly, 0) + assert.equal(team.roles.find((entry) => entry.role === 'lead')?.calls, 12) + } finally { + store.close() + } +}) + +test('a running job without started_at is still eligible for timeout cleanup', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'stale-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_stale', userId: user.id, owner: 'o', name: 'stale', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + store.createJob({ id: 'job_stale', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb', platformTokenHash: '' }) + store.updateJob('job_stale', { status: 'running' }) + assert.equal(store.getJob('job_stale')?.started_at, null) + const stale = store.staleRunningJobs(new Date(Date.now() + 60_000).toISOString()) + assert.equal(stale.some((job) => job.id === 'job_stale'), true) + store.createJob({ id: 'job_stale_queued', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb2', platformTokenHash: '' }) + const staleQueued = store.staleQueuedJobs(new Date(Date.now() + 60_000).toISOString()) + assert.equal(staleQueued.some((job) => job.id === 'job_stale_queued'), true) + } finally { + store.close() + } +}) + + +test('the role backfill runs once and future schema versions are refused', () => { + const dir = mkdtempSync(join(tmpdir(), 'fledge-migrate-')) + const path = join(dir, 'fledge.db') + try { + const store = new Store(path) + const user = makeUser(store, { login: 'migrate-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_migrate', userId: user.id, owner: 'o', name: 'migrate', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: null, goal: '', autoIterate: false, intervalMinutes: 1440 }) + store.db.exec('PRAGMA user_version = 14;') + store.close() + + const reopened = new Store(path) + assert.equal(reopened.getRepo(repo.id)?.repo_role, 'maintainer') + assert.notEqual(reopened.getRepo(repo.id)?.role_checked_at, null) + reopened.db.exec('PRAGMA user_version = 23;') + reopened.close() + + assert.throws(() => new Store(path), /newer than this build/) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('pruneOldData removes old jobs, notifications and settings keys', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'prune-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_prune', userId: user.id, owner: 'o', name: 'prune', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + store.createJob({ id: 'job_old', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb1', platformTokenHash: '' }) + store.updateJob('job_old', { status: 'succeeded', finished_at: '2000-01-01T00:00:00.000Z' }) + store.createJob({ id: 'job_new', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb2', platformTokenHash: '' }) + store.updateJob('job_new', { status: 'succeeded', finished_at: new Date().toISOString() }) + store.notify(user.id, 'test', 'old') + store.db.exec("UPDATE notifications SET created_at = '2000-01-01T00:00:00.000Z'") + store.setSetting('followup:old', '2000-01-01T00:00:00.000Z') + store.setSetting('followup:recent', new Date().toISOString()) + store.pruneOldData() + assert.equal(store.getJob('job_old'), undefined) + assert.ok(store.getJob('job_new')) + assert.equal(store.listNotifications(user.id).length, 0) + assert.equal(store.getSetting('followup:old'), undefined) + assert.notEqual(store.getSetting('followup:recent'), undefined) + } finally { + store.close() + } +}) + +test('work items persist, update and are found by issue or PR', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'work-item-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_work_items', userId: user.id, owner: 'o', name: 'wi', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + const item = store.createWorkItem({ id: 'work_1', repoId: repo.id, userId: user.id, title: 'Fix parser', detail: 'Details', issueUrl: 'https://github.com/o/wi/issues/1', priority: 20 }) + assert.equal(item.state, 'proposed') + assert.equal(store.findWorkItemByIssueUrl(repo.id, 'https://github.com/o/wi/issues/1')?.id, 'work_1') + store.updateWorkItem(item.id, { state: 'pr_ready', pr_url: 'https://github.com/o/wi/pull/2', pr_state: 'open' }) + assert.equal(store.findWorkItemByPrUrl(repo.id, 'https://github.com/o/wi/pull/2')?.state, 'pr_ready') + assert.equal(store.listWorkItems(repo.id).length, 1) + assert.equal(store.workItemMetrics('2000-01-01T00:00:00.000Z').open, 1) + store.updateWorkItem(item.id, { state: 'merged', merged_at: new Date().toISOString() }) + assert.equal(store.workItemMetrics('2000-01-01T00:00:00.000Z').merged, 1) + } finally { + store.close() + } +}) + +test('usage and job tokens carry the repository for cost attribution', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'attribution-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_attribution', userId: user.id, owner: 'o', name: 'a', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + store.recordUsage(user.id, null, 0.2, null, repo.id) + const usage = store.get<{ repo_id: string | null }>('SELECT repo_id FROM quota_ledger ORDER BY id DESC LIMIT 1') + assert.equal(usage?.repo_id, repo.id) + const token = store.createApiToken({ id: 'tok_attr', userId: user.id, kind: 'job', name: 'job:attr', tokenHash: 'hash_attr', prefix: 'flg', repoId: repo.id, expiresAt: null }) + assert.equal(token.repo_id, repo.id) + } finally { + store.close() + } +}) + +test('answering a team question marks only question notifications read', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'inbox-user', status: 'approved' }) + store.notify(user.id, 'job', 'o/r 团队提问：continue?') + store.notify(user.id, 'job', 'unrelated job failure') + store.markQuestionNotificationsRead(user.id) + const rows = store.listNotifications(user.id) + assert.notEqual(rows.find((row) => row.message.includes('团队提问'))?.read_at, null) + assert.equal(rows.find((row) => row.message.includes('unrelated'))?.read_at, null) + } finally { + store.close() + } +}) + +test('user model connections and repository selections persist', () => { + const store = new Store(':memory:') + try { + const user = makeUser(store, { login: 'model-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_models', userId: user.id, owner: 'o', name: 'm', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + const first = store.createUserModel({ id: 'mdl_a', userId: user.id, label: 'A', baseUrl: 'https://a.example/v1', keyEnc: 'enc', model: 'model-a', allowPrivate: false }) + const second = store.createUserModel({ id: 'mdl_b', userId: user.id, label: 'B', baseUrl: 'http://192.168.1.2:11434/v1', keyEnc: 'enc', model: 'model-b', allowPrivate: true }) + assert.equal(store.listUserModels(user.id).length, 2) + assert.equal(store.userModelById(first.id, user.id)?.label, 'A') + store.setRepoModels(repo.id, [first.id, second.id], second.id) + const selected = store.listRepoModels(repo.id) + assert.deepEqual(selected.map((model) => model.id), [first.id, second.id]) + assert.equal(selected.find((model) => model.is_default === 1)?.id, second.id) + assert.equal(store.getRepo(repo.id)?.models_dirty, 1) + assert.equal(store.repoModelForRun(repo.id, second.id)?.model, 'model-b') + store.markRepoModelsApplied(repo.id) + assert.equal(store.getRepo(repo.id)?.models_dirty, 0) + store.deleteUserModel(first.id, user.id) + assert.equal(store.listRepoModels(repo.id).length, 1) + } finally { + store.close() + } +}) diff --git a/test/helpers.ts b/test/helpers.ts index ba8feef..d077af9 100644 --- a/test/helpers.ts +++ b/test/helpers.ts @@ -22,7 +22,7 @@ export function testConfig(overrides: Partial = {}): Config { tokenEncryptionKey: randomBytes(32).toString('base64'), dataDir: '/tmp/fledge-test', dbPath: ':memory:', - runner: { mode: 'mock', bridgeUrl: '', bridgeToken: '', timeoutMinutes: 45 }, + runner: { mode: 'mock', bridgeUrl: '', bridgeToken: '', timeoutMinutes: 45, stallMinutes: 30 }, allowTokenLoginOnly: true, mail: { transport: 'console', diff --git a/test/jobs.test.ts b/test/jobs.test.ts index 70ca4da..cb5763c 100644 --- a/test/jobs.test.ts +++ b/test/jobs.test.ts @@ -943,3 +943,67 @@ test('three consecutive job failures raise one ops notice', async () => { store.close() } }) + +test('a running job with no heartbeat is failed as stalled', async () => { + const store = new Store(':memory:') + const jobs = new Jobs(store, testConfig(), { name: 'noop', async run() { return { status: 'running' } } }, new MemoryMailer()) + try { + const user = makeUser(store, { login: 'heartbeat-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_heartbeat', userId: user.id, owner: 'o', name: 'h', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + store.createJob({ id: 'job_heartbeat', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb', platformTokenHash: '', kind: 'iterate' }) + store.updateJob('job_heartbeat', { status: 'running', started_at: '2000-01-01T00:00:00.000Z' }) + await jobs.tick() + const job = store.getJob('job_heartbeat') + assert.equal(job?.status, 'failed') + assert.match(job?.error ?? '', /停滞/) + } finally { + store.close() + } +}) + +test('three consecutive failures pause auto-iterate', async () => { + const store = new Store(':memory:') + const jobs = new Jobs(store, testConfig(), { name: 'noop', async run() { return { status: 'running' } } }, new MemoryMailer()) + try { + const user = makeUser(store, { login: 'pause-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_pause', userId: user.id, owner: 'o', name: 'p', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: true, intervalMinutes: 1440 }) + for (let i = 1; i <= 3; i += 1) { + const secret = `pause-secret-${i}` + const id = `job_pause_${i}` + store.createJob({ id, repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: sha256Hex(secret), platformTokenHash: '', kind: 'iterate', baseBranch: 'main' }) + store.updateJob(id, { status: 'running', started_at: new Date().toISOString() }) + await jobs.handleCallback(id, secret, { status: 'failed', error: 'boom' }) + } + assert.equal(store.getRepo(repo.id)?.auto_iterate, 0) + assert.equal(store.getRepo(repo.id)?.consecutive_failures, 3) + assert.equal(store.listNotifications(user.id).some((entry) => entry.message.includes('自动迭代已暂停')), true) + } finally { + store.close() + } +}) + +test('a credential 403 pauses the repository before enqueue', async () => { + const github = await startFakeServer((request, response) => { + if (request.url?.startsWith('/repos/o/c/branches/')) { + json(response, 403, { message: 'Resource not accessible by personal access token' }) + return + } + json(response, 404, { message: 'not found' }) + }) + const base = testConfig() + const config = testConfig({ github: { ...base.github, apiBase: github.url } }) + const store = new Store(':memory:') + const jobs = new Jobs(store, config, { name: 'noop', async run() { return { status: 'running' } } }, new MemoryMailer()) + try { + const user = makeUser(store, { login: 'credential-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_credential', userId: user.id, owner: 'o', name: 'c', defaultBranch: 'main', tokenEnc: encryptSecret('bad-token', config.tokenEncryptionKey), permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: true, intervalMinutes: 1440 }) + const result = await jobs.enqueue(repo, 'manual', 'plan this', 'plan') + assert.equal(result.ok, false) + assert.equal(result.code, 'REPO_CREDENTIAL_INVALID') + assert.equal(store.getRepo(repo.id)?.credential_state, 'invalid') + assert.equal(store.getRepo(repo.id)?.auto_iterate, 0) + } finally { + store.close() + await github.close() + } +})