fix: fail stalled jobs and pause invalid credentials

.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=

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.

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

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 |

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> = {}): 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))),

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<RepoRow>('SELECT * FROM repos WHERE user_id = ? AND lower(owner) = lower(?) AND lower(name) = lower(?)', userId, owner, name)
   }
 
-  updateRepo(id: string, patch: Partial<Pick<RepoRow, 'goal' | 'enabled' | 'auto_iterate' | 'interval_minutes' | 'last_run_at' | 'last_status' | 'default_branch' | 'repo_role' | 'auto_merge' | 'role_checked_at' | 'plan_required' | 'allow_paths' | 'deny_paths' | 'merge_method' | 'require_protection' | 'stage_mode' | 'models_dirty' | 'monthly_budget_cny' | 'job_budget_cny' | 'protection_state' | 'protection_checked_at' | 'permission' | 'can_review' | 'can_create_issue' | 'reviewer_enabled' | 'review_interval_minutes' | 'last_review_at' | 'last_review_status' | 'target_branch' | 'is_fork' | 'fork_created' | 'upstream_owner' | 'upstream_name' | 'fork_owner' | 'fork_name' | 'upstream_default_branch' | 'fork_default_branch' | 'upstream_html_url' | 'fork_html_url'>>): void {
+  updateRepo(id: string, patch: Partial<Pick<RepoRow, 'goal' | 'enabled' | 'auto_iterate' | 'interval_minutes' | 'last_run_at' | 'last_status' | 'default_branch' | 'repo_role' | 'auto_merge' | 'role_checked_at' | 'plan_required' | 'allow_paths' | 'deny_paths' | 'merge_method' | 'require_protection' | 'stage_mode' | 'models_dirty' | 'credential_state' | 'consecutive_failures' | 'monthly_budget_cny' | 'job_budget_cny' | 'protection_state' | 'protection_checked_at' | 'permission' | 'can_review' | 'can_create_issue' | 'reviewer_enabled' | 'review_interval_minutes' | 'last_review_at' | 'last_review_status' | 'target_branch' | 'is_fork' | 'fork_created' | 'upstream_owner' | 'upstream_name' | 'fork_owner' | 'fork_name' | 'upstream_default_branch' | 'fork_default_branch' | 'upstream_html_url' | 'fork_html_url'>>): 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<JobRow>("SELECT * FROM jobs WHERE status = 'queued' ORDER BY priority DESC, created_at ASC")
   }
 
-  updateJob(id: string, patch: Partial<Pick<JobRow, 'status' | 'branch' | 'pr_url' | 'issue_url' | 'review_score' | 'attribution_json' | 'evidence_json' | 'checkpoint' | 'team_stats_json' | 'summary' | 'error' | 'cost_usd' | 'started_at' | 'finished_at' | 'base_branch' | 'work_item_id' | 'priority' | 'plan_approved_at' | 'plan_approved_by' | 'plan_rejected_at' | 'plan_allow_paths' | 'plan_deny_paths' | 'base_sha'>>): void {
+  updateJob(id: string, patch: Partial<Pick<JobRow, 'status' | 'branch' | 'pr_url' | 'issue_url' | 'review_score' | 'attribution_json' | 'evidence_json' | 'checkpoint' | 'team_stats_json' | 'summary' | 'error' | 'cost_usd' | 'started_at' | 'last_callback_at' | 'finished_at' | 'base_branch' | 'work_item_id' | 'priority' | 'plan_approved_at' | 'plan_approved_by' | 'plan_rejected_at' | 'plan_allow_paths' | 'plan_deny_paths' | 'base_sha'>>): 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<JobRow>("SELECT * FROM jobs WHERE status = 'running' AND COALESCE(started_at, created_at) < ?", beforeIso)
   }
 
+  stalledRunningJobs(beforeIso: string): JobRow[] {
+    return this.all<JobRow>("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<JobRow>("SELECT * FROM jobs WHERE status = 'queued' AND created_at < ?", beforeIso)
   }

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<string>()
   private ticking = false
   private lastPruneAt = 0
+  private readonly credentialCache = new Map<string, { at: number }>()
   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<EnqueueResult> {
     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<RepoRow | undefined> {
@@ -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<void> {
+    const handled = new Set<string>()
+    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)