diff --git a/changelog.d/job-control.added.md b/changelog.d/job-control.added.md new file mode 100644 index 0000000..f8d7a47 --- /dev/null +++ b/changelog.d/job-control.added.md @@ -0,0 +1 @@ +Added per-job pause, resume and cancel controls. Paused jobs release the repository slot and can be resumed from their checkpoint; cancellation stops active team sessions, revokes the job token and closes the linked work item. Jobs now record priority, and queued jobs show their queue position. diff --git a/src/app.ts b/src/app.ts index 893c87a..870ad41 100644 --- a/src/app.ts +++ b/src/app.ts @@ -1249,17 +1249,45 @@ export async function buildApp(options: BuildOptions): Promise { const { viewer, repo } = ownedRepo(request, String((request.params as Record).id)) const job = store.getJob(String((request.params as Record).jobId)) if (job === undefined || job.repo_id !== repo.id) throw httpError(404, '任务不存在。', 'JOB_NOT_FOUND') + const queued = store.listQueuedJobs() + const queueIndex = queued.findIndex((entry) => entry.id === job.id) return reply.type('text/html').send(jobPage({ config, locale: request.locale ?? 'zh', viewer: viewOf(viewer), repo, job, + queuePosition: queueIndex === -1 ? undefined : queueIndex + 1, notice: queryText(request, 'notice'), error: queryText(request, 'error'), })) }) + app.post('/repos/:id/jobs/:jobId/cancel', async (request, reply) => { + const { repo } = ownedRepo(request, String((request.params as Record).id), true) + const job = store.getJob(String((request.params as Record).jobId)) + if (job === undefined || job.repo_id !== repo.id) throw httpError(404, '任务不存在。', 'JOB_NOT_FOUND') + const result = await jobs.cancelJob(job.id, '用户取消任务') + return reply.redirect(`/repos/${repo.id}/jobs/${job.id}?${result.ok ? 'notice' : 'error'}=${encodeURIComponent(result.ok ? '任务已取消。' : result.error ?? '取消失败。')}`) + }) + + app.post('/repos/:id/jobs/:jobId/pause', async (request, reply) => { + const { repo } = ownedRepo(request, String((request.params as Record).id), true) + const job = store.getJob(String((request.params as Record).jobId)) + if (job === undefined || job.repo_id !== repo.id) throw httpError(404, '任务不存在。', 'JOB_NOT_FOUND') + const result = await jobs.pauseJob(job.id) + return reply.redirect(`/repos/${repo.id}/jobs/${job.id}?${result.ok ? 'notice' : 'error'}=${encodeURIComponent(result.ok ? '任务已暂停。' : result.error ?? '暂停失败。')}`) + }) + + app.post('/repos/:id/jobs/:jobId/resume', async (request, reply) => { + const { repo } = ownedRepo(request, String((request.params as Record).id), true) + const job = store.getJob(String((request.params as Record).jobId)) + if (job === undefined || job.repo_id !== repo.id) throw httpError(404, '任务不存在。', 'JOB_NOT_FOUND') + const result = await jobs.resumeJob(job.id) + if (!result.ok || result.job === undefined) return reply.redirect(`/repos/${repo.id}/jobs/${job.id}?error=${encodeURIComponent(result.error ?? '恢复失败。')}`) + return reply.redirect(`/repos/${repo.id}/jobs/${result.job.id}?notice=${encodeURIComponent('已恢复任务。')}`) + }) + app.post('/repos/:id/jobs/:jobId/retry', async (request, reply) => { const { viewer, repo } = ownedRepo(request, String((request.params as Record).id), true) const job = store.getJob(String((request.params as Record).jobId)) @@ -1329,7 +1357,7 @@ export async function buildApp(options: BuildOptions): Promise { if (title === '') return reply.redirect(`/repos/${repo.id}?error=${encodeURIComponent('请填写工作项标题。')}#work`) const detail = (body.detail ?? '').trim().slice(0, 4000) const rawIssue = (body.issueUrl ?? '').trim() - const issueUrl = /^https:\/\/github\.com\//.test(rawIssue) ? rawIssue.slice(0, 500) : null + const issueUrl = rawIssue.startsWith('https://github.com/') ? rawIssue.slice(0, 500) : null const priority = Math.max(0, Math.min(100, Math.floor(Number(body.priority) || 0))) const item = store.createWorkItem({ id: randomId('work'), repoId: repo.id, userId: viewer.user.id, title, detail, issueUrl, priority, state: 'proposed' }) store.audit(viewer.user.id, 'work_item.created', `${item.id} ${repo.owner}/${repo.name} priority=${priority}`) diff --git a/src/db.ts b/src/db.ts index 4318090..65d6021 100644 --- a/src/db.ts +++ b/src/db.ts @@ -23,7 +23,7 @@ const { DatabaseSync } = require('node:sqlite') as { DatabaseSync: new (path: st export type UserStatus = 'onboarding' | 'pending' | 'approved' | 'rejected' export type UserRole = 'user' | 'admin' | 'superadmin' -export type JobStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled' | 'quota_exhausted' +export type JobStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled' | 'paused' | 'quota_exhausted' export interface UserRow { id: string @@ -1308,7 +1308,11 @@ export class Store { } listActiveJobs(): JobRow[] { - return this.all("SELECT * FROM jobs WHERE status IN ('queued','running') ORDER BY created_at ASC") + return this.all("SELECT * FROM jobs WHERE status IN ('queued','running') ORDER BY priority DESC, created_at ASC") + } + + listQueuedJobs(): JobRow[] { + return this.all("SELECT * FROM jobs WHERE status = 'queued' ORDER BY priority DESC, created_at ASC") } updateJob(id: string, patch: Partial>): void { diff --git a/src/jobs/service.ts b/src/jobs/service.ts index f053320..e40c088 100644 --- a/src/jobs/service.ts +++ b/src/jobs/service.ts @@ -109,6 +109,77 @@ export class Jobs { } } + async cancelJob(jobId: string, reason: string): Promise<{ ok: boolean; error?: string }> { + const job = this.store.getJob(jobId) + if (job === undefined) return { ok: false, error: '任务不存在。' } + if (job.status !== 'queued' && job.status !== 'running') return { ok: false, error: '这个任务已经结束。' } + if (job.status === 'running' && this.instances !== undefined) { + try { + const state = await this.instances.teamState(job.repo_id, 5_000) + const sessions = state.bots.filter((bot) => bot.activity === 'thinking' || bot.activity === 'working').map((bot) => bot.sessionId) + if (sessions.length > 0) await this.instances.stopSessions(job.repo_id, sessions) + } catch (error) { + log.warn('cancel job stop failed', { job: jobId, error: String(error) }) + } + } + 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) + await this.instances?.clearJobToken(job.repo_id, jobId).catch((error) => log.warn('clear job token failed', { job: jobId, error: String(error) })) + this.store.updateJob(jobId, { status: 'cancelled', error: reason.slice(0, 1000), finished_at: nowIso() }) + if (job.kind === 'review') this.store.updateRepo(job.repo_id, { last_review_status: 'cancelled' }) + else this.store.updateRepo(job.repo_id, { last_status: 'cancelled' }) + if (job.work_item_id !== null) { + const item = this.store.getWorkItem(job.work_item_id) + if (item !== undefined) this.store.updateWorkItem(item.id, { state: 'closed' }) + } + this.store.audit(job.user_id, 'job.cancelled', `${jobId}: ${reason.slice(0, 300)}`) + this.store.event('job_cancelled', { userId: job.user_id, repoId: job.repo_id, jobId }) + return { ok: true } + } + + async pauseJob(jobId: string, reason = '任务已暂停'): Promise<{ ok: boolean; error?: string }> { + const job = this.store.getJob(jobId) + if (job === undefined) return { ok: false, error: '任务不存在。' } + if (job.status !== 'queued' && job.status !== 'running') return { ok: false, error: '这个任务已经结束。' } + if (job.status === 'running' && this.instances !== undefined) { + try { + const state = await this.instances.teamState(job.repo_id, 5_000) + const sessions = state.bots.filter((bot) => bot.activity === 'thinking' || bot.activity === 'working').map((bot) => bot.sessionId) + if (sessions.length > 0) await this.instances.stopSessions(job.repo_id, sessions) + } catch (error) { + log.warn('pause job stop failed', { job: jobId, error: String(error) }) + } + } + 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) + await this.instances?.clearJobToken(job.repo_id, jobId).catch((error) => log.warn('clear job token failed', { job: jobId, error: String(error) })) + this.store.updateJob(jobId, { status: 'paused', error: reason.slice(0, 1000) }) + this.store.updateRepo(job.repo_id, { last_status: 'paused', ...(job.kind === 'review' ? { last_review_status: 'paused' } : {}) }) + this.store.audit(job.user_id, 'job.paused', jobId) + this.store.event('job_paused', { userId: job.user_id, repoId: job.repo_id, jobId }) + return { ok: true } + } + + async resumeJob(jobId: string): Promise<{ ok: boolean; job?: JobRow; error?: string }> { + const job = this.store.getJob(jobId) + if (job === undefined) return { ok: false, error: '任务不存在。' } + if (job.status !== 'paused') return { ok: false, error: '只有已暂停的任务可以恢复。' } + const repo = this.store.getRepo(job.repo_id) + if (repo === undefined) return { ok: false, error: '仓库不存在。' } + const goal = job.checkpoint === null || job.checkpoint === '' ? job.goal : `${job.goal}\n\nResume from checkpoint:\n${job.checkpoint}` + const result = await this.enqueue(repo, 'manual', goal, job.kind === 'review' ? 'review' : job.kind === 'plan' ? 'plan' : 'iterate', { + ...(job.branch === null ? {} : { branch: job.branch }), + ...(job.pr_url === null ? {} : { prUrl: job.pr_url }), + ...(job.base_branch === null ? {} : { baseBranch: job.base_branch }), + ...(job.work_item_id === null ? {} : { workItemId: job.work_item_id }), + priority: job.priority, + }) + if (!result.ok || result.job === undefined) return { ok: false, error: result.error ?? '恢复失败。' } + this.store.updateJob(jobId, { status: 'cancelled', error: `resumed as ${result.job.id}`, finished_at: nowIso() }) + this.store.audit(job.user_id, 'job.resumed', `${jobId} -> ${result.job.id}`) + return { ok: true, job: result.job } + } + async cancelRepoJobs(repoId: string, reason: string): Promise { for (const job of this.store.listActiveJobs().filter((entry) => entry.repo_id === repoId)) { const revoke = (): void => { diff --git a/src/views/pages.ts b/src/views/pages.ts index 5862b85..216d57a 100644 --- a/src/views/pages.ts +++ b/src/views/pages.ts @@ -58,6 +58,8 @@ function statusLabel(t: Translate, status: string): string { onboarding: t('待申请', 'Onboarding'), idle: t('空闲', 'Idle'), quota_exhausted: t('额度用完', 'Quota exhausted'), + paused: t('已暂停', 'Paused'), + cancelled: t('已取消', 'Cancelled'), ready: t('就绪', 'Ready'), starting: t('启动中', 'Starting'), provisioning: t('初始化中', 'Provisioning'), @@ -158,6 +160,8 @@ function jobSummary(locale: Locale, t: Translate, job: JobRow): string { if (raw === '' || raw === 'probe' || raw.length < 8) { if (job.status === 'succeeded') return `${job.kind === 'research' ? t('分析完成', 'Research completed') : t('任务完成', 'Job completed')}${cost}` if (job.status === 'failed') return `${job.kind === 'research' ? t('分析失败', 'Research failed') : t('任务失败', 'Job failed')}${cost}` + if (job.status === 'paused') return `${t('已暂停', 'Paused')}${cost}` + if (job.status === 'cancelled') return `${t('已取消', 'Cancelled')}${cost}` return '—' } return `${raw}${cost}` @@ -1347,10 +1351,11 @@ export function jobPage(input: { viewer: Viewer repo: RepoRow job: JobRow + queuePosition?: number notice?: string error?: string }): string { - const { config, locale, viewer, repo, job, notice, error } = input + const { config, locale, viewer, repo, job, queuePosition, notice, error } = input const t = translator(locale) const advanced = advancedViewer(viewer) const evidence = parseEvidenceJson(job.evidence_json) @@ -1376,6 +1381,11 @@ export function jobPage(input: { const execute = job.kind === 'plan' && job.status === 'succeeded' && job.plan_approved_at !== null ? `
${csrf(viewer)}
` : '' + const control = job.status === 'queued' || job.status === 'running' + ? `
${csrf(viewer)}
${csrf(viewer)}
` + : job.status === 'paused' + ? `
${csrf(viewer)}
` + : '' const planCard = job.kind === 'plan' && job.status === 'succeeded' ? `

${escapeHtml(t('计划审批', 'Plan approval'))}

${job.plan_rejected_at !== null ? escapeHtml(t('计划已被拒绝。', 'This plan was rejected.')) : job.plan_approved_at !== null ? escapeHtml(t('计划已批准;可以执行。', 'This plan is approved and can be executed.')) : escapeHtml(t('等待 maintainer 批准计划。', 'Waiting for a maintainer to approve the plan.'))}

${job.plan_approved_at !== null ? escapeHtml(t('已批准', 'Approved')) : job.plan_rejected_at !== null ? escapeHtml(t('已拒绝', 'Rejected')) : escapeHtml(t('待批准', 'Pending'))}
@@ -1390,12 +1400,13 @@ 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))}`}

+

${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}`}

diff --git a/test/jobs.test.ts b/test/jobs.test.ts index e6e0f03..4e070e4 100644 --- a/test/jobs.test.ts +++ b/test/jobs.test.ts @@ -726,3 +726,36 @@ test('flybot preflight records the base SHA and skips scheduled work while a wor await github.close() } }) + +test('a running job can be cancelled and a paused job can be resumed', async () => { + const config = testConfig() + const store = new Store(':memory:') + const contexts: RunnerContext[] = [] + const runner: Runner = { name: 'capture', async run(context) { contexts.push(context); return { status: 'running' } } } + const jobs = new Jobs(store, config, runner, new MemoryMailer()) + try { + const user = makeUser(store, { login: 'job-control-user', status: 'approved' }) + const repo = store.createRepo({ id: 'repo_job_control', userId: user.id, owner: 'o', name: 'jc', defaultBranch: 'main', tokenEnc: 'x', permission: 'admin', repoRole: 'maintainer', goal: '', autoIterate: false, intervalMinutes: 1440 }) + const item = store.createWorkItem({ id: 'work_cancel', repoId: repo.id, userId: user.id, title: 'Cancel me', state: 'in_progress' }) + store.createJob({ id: 'job_cancel', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g', callbackHash: 'cb', platformTokenHash: '', kind: 'iterate', workItemId: item.id }) + store.updateJob('job_cancel', { status: 'running', started_at: new Date().toISOString() }) + const cancelled = await jobs.cancelJob('job_cancel', 'user cancelled') + assert.equal(cancelled.ok, true) + assert.equal(store.getJob('job_cancel')?.status, 'cancelled') + assert.equal(store.getWorkItem(item.id)?.state, 'closed') + + store.createJob({ id: 'job_pause', repoId: repo.id, userId: user.id, trigger: 'manual', goal: 'g2', callbackHash: 'cb2', platformTokenHash: '', kind: 'iterate', baseBranch: 'main' }) + store.updateJob('job_pause', { status: 'running', started_at: new Date().toISOString() }) + const paused = await jobs.pauseJob('job_pause') + assert.equal(paused.ok, true) + assert.equal(store.getJob('job_pause')?.status, 'paused') + const resumed = await jobs.resumeJob('job_pause') + assert.equal(resumed.ok, true) + assert.ok(resumed.job) + assert.equal(store.getJob('job_pause')?.status, 'cancelled') + assert.equal(contexts.length, 1) + assert.equal(resumed.job?.status, 'running') + } finally { + store.close() + } +})