From 138b41e6314be34b3c679616e4b82197946db896 Mon Sep 17 00:00:00 2001 From: zy95-12 <135135297+zy95-12@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:43:35 +0800 Subject: [PATCH 1/4] fix: queue confirmed-idle continuations through project capacity (#2328) --- core/src/host/project-chief-control.js | 1 + core/src/host/project-delivery.js | 1 + core/src/host/project-execution-capacity.js | 24 +++++ core/src/host/project-pools.js | 7 +- core/src/host/project-scheduler.js | 74 +++++++++++++- core/src/host/workbench-state.js | 4 +- core/test/project-scheduler.test.js | 107 ++++++++++++++++++++ 7 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 core/src/host/project-execution-capacity.js diff --git a/core/src/host/project-chief-control.js b/core/src/host/project-chief-control.js index b3e62aed6..bbb952347 100644 --- a/core/src/host/project-chief-control.js +++ b/core/src/host/project-chief-control.js @@ -165,6 +165,7 @@ export function createProjectChiefControl(rt) { if (task.collaboration) { task.collaboration.status = 'blocked'; task.collaboration.block = { reason, chiefDecisionId: audit.id, at: Date.now() } } if (task.collaboration?.claim?.chiefRunId === run.id) task.collaboration.claim.status = 'waiting-human' task.chiefQuestion = { decisionId: audit.id, reason, at: Date.now() } + await rt.projectScheduler?.captureWait?.(task) task.chiefQuestion.recipient = rt.projectAgents.contactUser?.(task, { id: audit.id, reason, requiresHuman: decision.requiresHuman === true }, actor) ?? { kind: 'human', memberId: task.collaboration?.initiatedBy ?? actor.memberId } } else if (decision.action === 'close') { rt.projectTaskLifecycle?.requireCompletedDependencies(task) diff --git a/core/src/host/project-delivery.js b/core/src/host/project-delivery.js index 329e0e390..b2f26f008 100644 --- a/core/src/host/project-delivery.js +++ b/core/src/host/project-delivery.js @@ -1042,6 +1042,7 @@ export function createProjectDelivery(rt, { publisher, releaseProvider } = {}) { if (['running', 'waiting-acceptance'].includes(previous.status)) previous.status = 'returned' previous.returned = { at: Date.now(), by: claim.botId, reason } const next = { id, protocol: previous.protocol, projectId: previous.projectId, taskId: task.id, requestKey: hash(`${task.projectId}\n${task.id}\n${claim.id}`), workflowVersion: previous.protocol === 'project-handoff/v1' ? 'project-delivery/pools-v4' : previous.workflowVersion, status: 'running', initiatedBy: previous.initiatedBy, executionOwnerId: previous.executionOwnerId, baseSha: previous.requirement.repository && !previous.candidateSha ? await git(settingsOf(projectOf(task.projectId, previous)).path, ['rev-parse', 'HEAD']) : previous.baseSha, seedSha: previous.candidateSha ?? previous.seedSha, ...(publicationBinding ? { publicationBinding } : {}), ...(previous.integrationInputs ? { integrationInputs: structuredClone(previous.integrationInputs), integrationAdmissions: structuredClone(previous.integrationAdmissions ?? []) } : {}), repositoryBinding: previous.repositoryBinding, requirement: requirementsOf(task, previous.requirement.repository), feedback: { previousAttemptId: previous.id, kind: 'bot-return', reason, candidateSha: previous.candidateSha }, agentDefinitions: rt.projectAgents.deliveryDefinitions(task.projectId), roleSessions: {}, rolePolicies: {}, resourceLeases: {}, workspaces: {}, evidence: {}, createdAt: Date.now(), updatedAt: Date.now(), runId: `pool-run-${randomUUID()}` } + if (previous.capacityWait && previous.capacityWait.status !== 'admitted') next.capacityWait = { ...structuredClone(previous.capacityWait), attemptId: id, previousAttemptId: previous.id } previous.supersededBy = id setEntry(attempts(), id, next); task.deliveryAttemptId = id rt.bindProject({ kind: 'run', id: previous.runId, projectId: task.projectId, taskId: task.id, attemptId: previous.id, status: 'failed' }) diff --git a/core/src/host/project-execution-capacity.js b/core/src/host/project-execution-capacity.js new file mode 100644 index 000000000..bd526a69e --- /dev/null +++ b/core/src/host/project-execution-capacity.js @@ -0,0 +1,24 @@ +// Waiting is a Chief/Task decision. Capacity suspension additionally requires host exit facts. +import { confirmedWorkerExit } from './project-task-sessions.js' +import { nativeExitConfirmed } from './project-native-sessions.js' +export function attemptIdle(state, attemptId) { + const attempt = state.projectAttempts?.[attemptId], task = state.tasks?.[attempt?.taskId] + if (!attempt || !task || task.projectId !== attempt.projectId) return false + if (attempt.workflowDefinition && Object.values(state.projectExecutions ?? {}).some(row => row.attemptId === attemptId && ['queued', 'running', 'working', 'paused', 'legacy'].includes(row.status))) return false + const leases = Object.values(state.projectResourceLeases ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + const sdk = Object.values(state.projectHarnessWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + const native = Object.values(state.projectNativeWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + if (Object.values(state.bots ?? {}).some(row => row.ephemeral && row.attemptId === attemptId && row.role !== 'chief')) return false + if (sdk.some(row => !confirmedWorkerExit(state, row) || !leases.some(lease => lease.id === row.leaseId && lease.status === 'released' && lease.generation === row.generation && lease.worker?.handle === row.id))) return false + if (native.some(row => !nativeExitConfirmed(state, row))) return false + if (leases.some(row => row.status !== 'released' || row.worker && !['sdk', 'native'].includes(row.worker.kind) || row.worker?.kind === 'sdk' && !sdk.some(worker => worker.id === row.worker.handle) || row.worker?.kind === 'native' && !native.some(worker => worker.id === row.worker.botId) || !row.worker && row.releaseProof?.kind !== 'unbound-reservation-release')) return false + const claim = task.collaboration?.claim + if (claim?.workerId && claim.role !== 'chief' && !claim.chiefRunId && !sdk.some(row => row.id === claim.workerId) && !native.some(row => row.id === claim.workerId)) return false + if (claim?.status === 'active' && !claim.workerId && !claim.chiefRunId) return false + // All adapters register before external execution; no records means no worker started. + return true +} +export function capacitySuspended(state, attempt) { + const task = state.tasks?.[attempt.taskId] + return Boolean(attempt.capacityWait && attempt.capacityWait.status !== 'admitted' && task?.deliveryAttemptId === attempt.id && attemptIdle(state, attempt.id)) +} diff --git a/core/src/host/project-pools.js b/core/src/host/project-pools.js index 51aa605d9..2d20659b9 100644 --- a/core/src/host/project-pools.js +++ b/core/src/host/project-pools.js @@ -152,6 +152,7 @@ export function createProjectPools(rt) { } } row.contextRefs = { source: contextReference(rt.state, task, fromRole), target: contextReference(rt.state, task, target.role) } + await rt.projectScheduler?.captureWait?.(task) const before = rt.projectEffects ? { collaboration: clone(c), handoff: task.handoff, question: task.chiefQuestion } : null let input, intent try { @@ -238,7 +239,10 @@ export function createProjectPools(rt) { definition = definitions(task.projectId).roles.find(row => row.id === botId) if (!definition) { c.status = 'blocked'; c.block = { reason: 'Target role is retired; Chief must explicitly select an active role', at: Date.now() }; await rt.save(); return null } if (Object.values(rt.state.tasks).some(row => row.collaboration?.claim?.botId === botId && row.collaboration.claim.status === 'active')) return null - if (adapter.ready && !adapter.ready({ task, definition })) return null + if (adapter.ready && !await adapter.ready({ task, definition })) return null + if (rt.projectScheduler?.admitPool && !await rt.projectScheduler.admitPool(task, definition)) return null + await authority(task) + if (c.generation !== expectedGeneration || c.status !== 'queued' || c.botId !== botId) return null const receipt = c.history.find(row => row.id === c.currentHandoffId) if (!receipt || receipt.status !== 'pending') throw refuse('Task has no pending handoff receipt') const limit = rt.config.projectHandoff?.quota?.steps @@ -523,6 +527,7 @@ export function createProjectPools(rt) { if (existing && existing.id !== c.attemptId && existing.feedback?.previousAttemptId === c.attemptId && rt.state.projectAttempts[c.attemptId]?.supersededBy === existing.id) c.attemptId = existing.id else c.attemptId = await adapter.rework?.({ task, claim: c.claim, reason: `Integrate adjudicated child results: ${wait.resumePoint}` }) ?? c.attemptId } + await rt.projectScheduler?.captureWait?.(task, { kind: 'children', id: wait.id }) const previous = wait.status wait.status = 'queued' try { diff --git a/core/src/host/project-scheduler.js b/core/src/host/project-scheduler.js index 5a48bec91..2ec00f651 100644 --- a/core/src/host/project-scheduler.js +++ b/core/src/host/project-scheduler.js @@ -3,6 +3,7 @@ import { createHash, randomUUID } from 'node:crypto' import { ownEntry, setEntry } from './store.js' import { taskRepository } from './project-task-repository.js' import { initialRoleTarget } from './project-role-admission.js' +import { attemptIdle, capacitySuspended } from './project-execution-capacity.js' import { preflightRepository } from './project-repository-policy.js' const ACTIVE = new Set(['queued', 'running', 'working', 'paused', 'legacy']) const error = message => Object.assign(new Error(message), { code: 'bot/invalid' }) @@ -17,6 +18,7 @@ export function releasedParentWait(state, attemptId) { export function createProjectScheduler(rt, { preflight = preflightRepository } = {}) { const { state } = rt const entries = () => (state.projectQueue ??= {}) + const savingAdmissions = new Set() let recovered = false let tail = Promise.resolve() let waking = null, wakeAgain = false @@ -28,11 +30,49 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = for (const bot of Object.values(state.bots)) if (bot.projectId === projectId && bot.ephemeral && bot.attemptId) keys.add(bot.attemptId) for (const lease of Object.values(state.projectResourceLeases ?? {})) if (lease.projectId === projectId && lease.role !== 'chief' && lease.status !== 'released' && lease.attemptId) keys.add(lease.attemptId) for (const key of [...keys]) { - if (releasedParentWait(state, key)) keys.delete(key) + if (releasedParentWait(state, key) || capacitySuspended(state, state.projectAttempts?.[key] ?? {})) keys.delete(key) } return keys.size } const quota = project => project.quota?.maxConcurrent ?? 2 + async function captureWait(task, source) { + const attempt = state.projectAttempts?.[task.deliveryAttemptId] + const wait = source ?? (task.chiefQuestion ? { kind: 'chief-input', id: task.chiefQuestion.decisionId } : task.collaboration?.status === 'waiting-children' ? { kind: 'children', id: task.projectTask?.wait?.id ?? task.collaboration.currentHandoffId } : null) + if (!attempt || !wait || attempt.taskId !== task.id) return + if (attempt.capacityWait?.source.id === wait.id && attempt.capacityWait.status !== 'admitted') return + const previous = attempt.capacityWait + attempt.capacityWait = { version: 'project-capacity-wait/v1', source: structuredClone(wait), status: 'suspended', attemptId: attempt.id, at: Date.now() } + try { await rt.save() } catch (error) { if (previous === undefined) delete attempt.capacityWait; else attempt.capacityWait = previous; throw error } + } + function capacity(projectId) { + const result = { activeSlots: slots(projectId), waitingInput: 0, waitingResume: 0, exitUnknown: 0 } + for (const attempt of Object.values(state.projectAttempts ?? {})) { + const task = state.tasks?.[attempt.taskId] + if (attempt.projectId !== projectId || task?.deliveryAttemptId !== attempt.id || !attempt.capacityWait || ['accepted', 'cancelled'].includes(task.status)) continue + if (attempt.capacityWait.status === 'admitted') continue + if (!attemptIdle(state, attempt.id)) ++result.exitUnknown + else if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') ++result.waitingInput + else ++result.waitingResume + } + return result + } + async function admitPool(task, definition) { + if (definition.role === 'chief') return true + const attempt = state.projectAttempts?.[task.deliveryAttemptId], wait = attempt?.capacityWait + if (!wait) return true + if (wait.status === 'admitted') return !savingAdmissions.has(attempt.id) + if (task.chiefQuestion || task.collaboration?.status !== 'queued' || !attemptIdle(state, attempt.id)) return false + const c = task.collaboration + let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) + if (!entry) { + entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } + setEntry(entries(), entry.id, entry) + try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } + } + // Queue admission runs outside the pool lock: drain can itself admit a new pool Task. + void wake() + return false + } function revalidate(projectId) { let changed = false for (const entry of Object.values(entries())) { @@ -81,7 +121,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = if (Object.values(state.projectAttempts ?? {}).some(attempt => attempt.taskId === task.id && ['preparing', 'running', 'waiting-acceptance'].includes(attempt.status))) throw error('Task already has an active Attempt') const initialTarget = initialRoleTarget(rt, task, body) const resourceReason = resourceBlock(project.id, initialTarget) - if (slots(project.id) < quota(project) && !resourceReason) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) + if (slots(project.id) < quota(project) && !resourceReason && !Object.values(entries()).some(row => row.projectId === project.id && row.status === 'waiting')) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) const entry = { id: `queue-${randomUUID()}`, projectId: project.id, taskId: task.id, requestId: body.requestId, requestKey, memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry) @@ -91,12 +131,33 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = async function drain() { if (rt.readOnly?.() || !rt.sharedProjectMode?.() || !rt.flybotIsRead?.('workflow/state')) return let changed = false + for (const task of Object.values(state.tasks)) if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') await captureWait(task) for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) for (const entry of pending) { const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue + if (entry.kind === 'resume') { + const task = state.tasks[entry.taskId], attempt = state.projectAttempts?.[entry.attemptId], c = task?.collaboration + if (!attempt || task.deliveryAttemptId !== attempt.id || c?.generation !== entry.generation || c.currentHandoffId !== entry.handoffId || c.botId !== entry.targetBotId || c.status !== 'queued' || task.chiefQuestion) { entry.status = 'cancelled'; entry.reason = 'Continuation changed'; changed = true; continue } + if (!attemptIdle(state, attempt.id)) { entry.reason = 'Execution exit unconfirmed'; changed = true; continue } + try { + await rt.projectDelegations?.background(entry.memberId) + rt.requireProjectAccess({ memberId: entry.memberId }, project.id, 'task-write', task) + rt.projectTaskLifecycle?.requireExecutable(task) + const catalog = rt.projectAgents?.deliveryDefinitions(project.id) + if (catalog && !catalog.roles.some(row => row.id === entry.targetBotId && row.status !== 'retired')) throw error('Continuation role was retired') + const blocked = resourceBlock(project.id, entry.initialTarget) + if (blocked) { entry.reason = blocked; changed = true; continue } + const previous = structuredClone(attempt.capacityWait) + savingAdmissions.add(attempt.id) + attempt.capacityWait.status = 'admitted'; attempt.capacityWait.queueId = entry.id; attempt.capacityWait.admittedAt = Date.now() + entry.status = 'started'; entry.updatedAt = Date.now(); delete entry.reason + try { await rt.save() } catch (failure) { attempt.capacityWait = previous; entry.status = 'waiting'; throw failure } finally { savingAdmissions.delete(attempt.id) } + } catch (failure) { entry.status = 'blocked'; entry.reason = String(failure.message).slice(0, 300) } + changed = true; continue + } let execution try { execution = taskRepository(rt, project, state.tasks[entry.taskId], { repositoryBinding: entry.repositoryBinding }); await preflight(rt, execution.project) } catch (failure) { entry.status = 'blocked'; entry.reason = failure.message; entry.updatedAt = Date.now(); changed = true; continue } @@ -126,6 +187,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = if (!entry) throw error('Unknown queued request') const task = ownEntry(state.tasks, entry.taskId) rt.requireProjectAccess(actor, entry.projectId, 'task-write', task) + if (entry.kind === 'resume') throw error('Cancel or change the Task through Chief; a continuation cannot become a new request') if (!['waiting', 'blocked'].includes(entry.status)) throw error('Queued request has already started or ended') entry.status = 'cancelled'; entry.updatedAt = Date.now(); entry.reason = 'Cancelled by an authorized member' await rt.save() @@ -137,6 +199,12 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const task = ownEntry(state.tasks, entry.taskId) rt.requireProjectAccess(actor, entry.projectId, 'run-control') if (!task || task.projectId !== entry.projectId || ['accepted', 'cancelled'].includes(task.status)) throw error('Task has ended or is unavailable') + if (entry.kind === 'resume') { + if (entry.status !== 'blocked') throw error('Only a blocked continuation can be retried') + await rt.projectDelegations?.background(entry.memberId) + if (!rt.projectCan({ memberId: entry.memberId }, entry.projectId, 'task-write', task)) throw error('Original requester no longer has Project authority') + entry.status = 'waiting'; delete entry.reason; await rt.save(); return { queue: structuredClone(entry) } + } if (entry.status !== 'blocked') throw error('Only a blocked queue request can be retried') // Recovery must not replace the initiating human with the operator's authority. const requester = { memberId: entry.memberId } @@ -191,7 +259,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = } return { request: (body, actor) => serial(() => request(body, actor)), cancel: (body, actor) => serial(() => cancel(body, actor)), retry: (body, actor) => serial(() => retry(body, actor)).then(result => { void wake(); return result }), - slots, revalidate, recover, drain: () => serial(drain), wake } + captureWait, capacity, admitPool, slots, revalidate, recover, drain: () => serial(drain), wake } } export function install(rt) { rt.projectScheduler = createProjectScheduler(rt) diff --git a/core/src/host/workbench-state.js b/core/src/host/workbench-state.js index c78fac471..55d21d050 100644 --- a/core/src/host/workbench-state.js +++ b/core/src/host/workbench-state.js @@ -39,7 +39,7 @@ export function install(rt) { const repository = await repositoryReadiness(rt, project) const activity = rt.activityView?.() ?? {} if (body.view === 'summary') return { - projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), repository, + projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), capacity: rt.projectScheduler.capacity?.(id) ?? null, repository, queue: Object.values(rt.state.projectQueue ?? {}).filter(row => row.projectId === id && ['waiting', 'blocked'].includes(row.status)).map(({ id, taskId, status, reason }) => ({ id, taskId, status, reason })), taskExecution: Object.values(rt.state.tasks).filter(task => task.projectId === id).map(task => taskExecutionStatus(rt.state, task, { activity })), metrics: deliveryMetrics(rt.state, id), @@ -51,7 +51,7 @@ export function install(rt) { const host = { scope: 'shared-host-process', rssBytes: process.memoryUsage().rss, cpuPercent: elapsedMicros >= 100_000 ? percent : null, cpuSampleMs: elapsedMicros / 1000, cpuLimit: null, memoryLimitBytes: null, gpu: null, reason: 'Per-Project CPU/memory limits and GPU pools are not reported by the current platform API' } cpu = nowCpu; measured = now - return { projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), + return { projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), capacity: rt.projectScheduler.capacity?.(id) ?? null, queue: Object.values(rt.state.projectQueue ?? {}).filter(entry => entry.projectId === id).map(entry => structuredClone(entry)), workers, workspaces: Object.values(rt.state.projectWorkspaces ?? {}).filter(workspace => workspace.projectId === id), harnessWorkers: Object.values(rt.state.projectHarnessWorkers ?? {}).filter(row => row.projectId === id && (row.visibility !== 'private' || row.memberId === actor.memberId) && (!row.roomId || (() => { const room = ownEntry(rt.state.projectRooms ?? {}, row.roomId); return room?.projectId === id && (!room.memberIds.length || room.memberIds.includes(actor.memberId)) })())).map(({ eventLog: _log, sealedLog: _sealedLog, cwd: _cwd, ...row }) => structuredClone(row)), diff --git a/core/test/project-scheduler.test.js b/core/test/project-scheduler.test.js index 67de95476..bcff61232 100644 --- a/core/test/project-scheduler.test.js +++ b/core/test/project-scheduler.test.js @@ -239,3 +239,110 @@ test('capacity queue retains the selected repository across restart and refuses assert.match(f.state.projectQueue[queued.queue.id].reason, /not currently associated/) assert.equal(Object.keys(f.state.projectAttempts).length, 0) }) + +test('confirmed input waits release both Run and Attempt slots; resumed work waits behind admitted work', async () => { + const { rt, state, scheduler } = fixture() + rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1'] + task.deliveryAttemptId = attempt.id + task.collaboration = { status: 'blocked', generation: 4, initiatedBy: 'bob', attemptId: attempt.id, history: [] } + task.chiefQuestion = { decisionId: 'chief-1', reason: 'Need project input' } + await scheduler.captureWait(task) + assert.equal(scheduler.slots('p1'), 0) + assert.equal(attempt.status, 'running'); assert.equal(state.projectExecutions[attempt.id].status, 'running') + await scheduler.request({ taskId: 'two', requestId: 'two' }, { memberId: 'bob' }) + assert.equal(scheduler.slots('p1'), 1) + delete task.chiefQuestion; task.collaboration.status = 'queued'; task.collaboration.currentHandoffId = 'resume-one'; task.collaboration.botId = 'p1:developer'; ++task.collaboration.generation + const definition = { id: 'p1:developer', role: 'developer' } + assert.equal(await scheduler.admitPool(task, definition), false) + assert.equal(Object.values(state.projectQueue).filter(row => row.kind === 'resume').length, 1) + assert.equal(await scheduler.admitPool(task, definition), false) + assert.equal(Object.values(state.projectQueue).length, 1) + state.projectAttempts['attempt-2'].status = 'waiting-acceptance'; state.projectExecutions['attempt-2'].status = 'done' + await scheduler.drain() + assert.equal(await scheduler.admitPool(task, definition), true) + assert.equal(scheduler.slots('p1'), 1) + assert.equal(Object.keys(state.projectAttempts).length, 2) + assert.equal(attempt.capacityWait.status, 'admitted') +}) + +test('input wait never releases an unknown SDK, native Bot or unreleased lease', async () => { + for (const kind of ['sdk', 'bot', 'lease', 'unknown-claim']) { + const { rt, state, scheduler } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one; task.deliveryAttemptId = 'attempt-1'; task.chiefQuestion = { decisionId: 'chief', reason: 'input' }; task.collaboration = { status: 'blocked', history: [] } + if (kind === 'sdk') state.projectHarnessWorkers = { w: { id: 'w', attemptId: 'attempt-1', status: 'released', generation: 1 } } + if (kind === 'bot') state.bots.w = { id: 'w', ephemeral: true, attemptId: 'attempt-1', projectId: 'p1' } + if (kind === 'lease') state.projectResourceLeases = { l: { id: 'l', attemptId: 'attempt-1', projectId: 'p1', status: 'unreachable' } } + if (kind === 'unknown-claim') task.collaboration.claim = { workerId: 'lost', role: 'developer', status: 'active' } + await scheduler.captureWait(task) + assert.equal(scheduler.slots('p1'), 1, kind) + assert.equal(scheduler.capacity('p1').exitUnknown, 1, kind) + } +}) + +test('simultaneous continuations and new requests share FIFO quota across restart and duplicate calls', async () => { + const f = fixture(), { rt, state, project, scheduler } = f; rt.projectScheduler = scheduler + project.quota.maxConcurrent = 2 + for (const id of ['one', 'two']) { + await scheduler.request({ taskId: id, requestId: id }, { memberId: 'bob' }) + const task = state.tasks[id]; task.deliveryAttemptId = id === 'one' ? 'attempt-1' : 'attempt-2' + task.chiefQuestion = { decisionId: `chief-${id}` }; task.collaboration = { status: 'blocked', generation: 1, history: [], initiatedBy: 'bob', botId: 'p1:developer', currentHandoffId: id } + await scheduler.captureWait(task) + } + assert.equal(scheduler.capacity('p1').waitingInput, 2) + project.quota.maxConcurrent = 1 + for (const task of Object.values(state.tasks)) { delete task.chiefQuestion; task.collaboration.status = 'queued' } + const restarted = createProjectScheduler(rt, { preflight: async () => repositorySettings(rt, project) }); rt.projectScheduler = restarted + const definition = { id: 'p1:developer', role: 'developer' } + const result = await Promise.all([restarted.admitPool(state.tasks.one, definition), restarted.admitPool(state.tasks.two, definition), restarted.admitPool(state.tasks.one, definition)]) + await restarted.drain() + assert.ok(result.every(value => typeof value === 'boolean')); assert.equal(restarted.slots('p1'), 1) + assert.equal(await restarted.admitPool(state.tasks.one, definition), true) + assert.equal(await restarted.admitPool(state.tasks.two, definition), false) + assert.equal(Object.values(state.projectQueue).length, 2) + assert.equal(restarted.capacity('p1').waitingResume, 1) + state.tasks.three = { id: 'three', projectId: 'p1', status: 'open', requesterId: 'bob' } + const { queue } = await restarted.request({ taskId: 'three', requestId: 'three' }, { memberId: 'bob' }) + state.projectAttempts['attempt-1'].status = 'accepted'; state.projectExecutions['attempt-1'].status = 'done' + await restarted.drain() + assert.equal(state.projectAttempts['attempt-2'].capacityWait.status, 'admitted') + assert.equal(state.projectQueue[queue.id].status, 'waiting'); assert.equal(Object.keys(state.projectAttempts).length, 2) +}) + +test('uncommitted re-admission, revoked principals and changed handoffs never start execution', async () => { + for (const kind of ['save', 'revoked', 'stale']) { + const { rt, state, scheduler, revoke } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1']; task.deliveryAttemptId = attempt.id + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', history: [], generation: 2, currentHandoffId: 'h', initiatedBy: 'bob', botId: 'p1:developer' } + await scheduler.captureWait(task); delete task.chiefQuestion; task.collaboration.status = 'queued' + // Hold a real project slot so queue creation and admission can be inspected separately. + state.projectAttempts.busy = { id: 'busy', projectId: 'p1', taskId: 'two', status: 'running' } + assert.equal(await scheduler.admitPool(task, { id: 'p1:developer', role: 'developer' }), false) + state.projectAttempts.busy.status = 'accepted' + if (kind === 'save') { let once = true; rt.save = async () => { if (once) { once = false; throw new Error('disk failure') } } } + if (kind === 'revoked') revoke() + if (kind === 'stale') ++task.collaboration.generation + await scheduler.drain() + const entry = Object.values(state.projectQueue)[0] + assert.ok(['blocked', 'cancelled'].includes(entry.status), kind) + assert.equal(attempt.capacityWait.status, 'suspended', kind) + assert.equal(scheduler.slots('p1'), 0, kind) + assert.equal(attempt.id, task.deliveryAttemptId) + } +}) + +test('a matching SDK exit proof releases an input wait while legacy active workflows remain held', async () => { + const { rt, state, scheduler } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1']; task.deliveryAttemptId = attempt.id + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', history: [], claim: { workerId: 'w', role: 'developer', status: 'completed' } } + state.projectResourceLeases = { l: { id: 'l', projectId: 'p1', attemptId: attempt.id, role: 'developer', resourceId: 'docker', generation: 2, status: 'released', worker: { kind: 'sdk', handle: 'w' } } } + state.projectHarnessWorkers = { w: { id: 'w', projectId: 'p1', attemptId: attempt.id, role: 'developer', resourceId: 'docker', leaseId: 'l', generation: 2, status: 'released', releaseProof: { kind: 'sdk-stop-ack-and-process-exit', generation: 2 } } } + await scheduler.captureWait(task); assert.equal(scheduler.slots('p1'), 0) + state.projectResourceLeases.l.generation = 1; assert.equal(scheduler.slots('p1'), 1) + state.projectResourceLeases.l.generation = 2; attempt.workflowDefinition = { nodes: [] }; assert.equal(scheduler.slots('p1'), 1) + state.projectExecutions[attempt.id].status = 'done'; assert.equal(scheduler.slots('p1'), 0) +}) From e2e1e70d1fd04f93ba0b5762b54f469121219720 Mon Sep 17 00:00:00 2001 From: zy95-12 <135135297+zy95-12@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:46:04 +0800 Subject: [PATCH 2/4] test: cover wait exit proofs, durable resume quota and service contract --- LESSONS.md | 8 + PROJECT.md | 4 +- ROADMAP.md | 3 +- changelog.d/2328.fixed.md | 1 + core/src/host/project-execution-capacity.js | 4 +- core/test/project-harness.test.js | 27 +++ core/test/project-pools.test.js | 14 ++ docs/backend-operations.md | 3 + docs/openapi/flybot.openapi.json | 211 ++++++++++++++---- docs/subsystems/backend-service.md | 3 + docs/subsystems/project-chief-control.md | 3 + docs/subsystems/project-delivery.md | 3 + docs/subsystems/project-harness.md | 3 + docs/subsystems/project-pools.md | 3 + docs/subsystems/project-workbench.md | 24 ++ .../files/src/host/project-chief-control.js | 1 + .../ds-bot/files/src/host/project-delivery.js | 1 + .../src/host/project-execution-capacity.js | 26 +++ .../ds-bot/files/src/host/project-pools.js | 7 +- .../files/src/host/project-scheduler.js | 74 +++++- .../ds-bot/files/src/host/workbench-state.js | 4 +- .../ds-bot/files/test/project-harness.test.js | 27 +++ .../ds-bot/files/test/project-pools.test.js | 14 ++ .../files/test/project-scheduler.test.js | 107 +++++++++ scripts/api-contracts.mjs | 8 +- 25 files changed, 533 insertions(+), 50 deletions(-) create mode 100644 changelog.d/2328.fixed.md create mode 100644 patches/ds-bot/files/src/host/project-execution-capacity.js diff --git a/LESSONS.md b/LESSONS.md index 61bb778da..4f6beb8dd 100644 --- a/LESSONS.md +++ b/LESSONS.md @@ -13,6 +13,14 @@ Dated, short lessons from real FlyBot work. Agents read this **before coding**. ## Open lessons +### 2026-11-04 — Releasing a slot requires re-admission and lock ordering (Seen 1×) + +A suspended Attempt still has a Run; filter both using the same trusted exit facts. +Resume through the original quota queue before claiming. Never await queue drain +inside the pool lock: new-task drain can itself need that lock. Save acknowledgement +must precede dispatch; input receipt alone is neither resumption nor exit proof. + + ### 2026-11-04 — Protocol errors are feedback, not business decisions (Seen 1×) A missing routing field stalled an otherwise configured pool. Validate the whole diff --git a/PROJECT.md b/PROJECT.md index 368b86d12..c4397bbc2 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -14,7 +14,9 @@ Current stage and priorities: [ROADMAP.md](ROADMAP.md). How decisions are made: preserves drafts and admits explicit inspect/integrate inputs through existing Effects. Chief selects only approved repository sources. #2327 adds bounded, durable Chief protocol correction in the resident Session; exhaustion retains incomplete judgement without automatic escalation. Follow-ups: -#2328 idle input-wait slots, #2330 inspection recovery. See +#2328 adds confirmed-idle wait suspension and FIFO continuation re-admission +through the original queue; exit-unknown capacity remains held. Remaining: #2330 +inspection recovery. See [delivery contract](docs/subsystems/project-delivery.md). Live Task mutation and deployment remain separate from source/test delivery. diff --git a/ROADMAP.md b/ROADMAP.md index b11de106f..acc81a51e 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -7,7 +7,8 @@ Implement and merge #2325 → #2327 → #2328 → #2330 sequentially into workspace materialization and explicit source admission; Chief owns recovery choices, the host preserves authority, evidence and execution-exit gates. #2327 implements same-Session correction with structured feedback and a configured -budget; no implicit role routing or human escalation. Remaining: #2328 and #2330. +budget; no implicit role routing or human escalation. #2328 adds exit-proven input waits and shared new/resume admission. +Remaining: #2330 scheduled inspection restart recovery. ## Agent connection authorization — 2026-11-03 diff --git a/changelog.d/2328.fixed.md b/changelog.d/2328.fixed.md new file mode 100644 index 000000000..5d3118ee6 --- /dev/null +++ b/changelog.d/2328.fixed.md @@ -0,0 +1 @@ +Confirmed-idle input waits release project execution capacity while retaining Task, Attempt and Session state. Continuations share the original FIFO queue and regain project/resource capacity before role-pool execution; exit-unknown work stays held. diff --git a/core/src/host/project-execution-capacity.js b/core/src/host/project-execution-capacity.js index bd526a69e..b4dbb2bce 100644 --- a/core/src/host/project-execution-capacity.js +++ b/core/src/host/project-execution-capacity.js @@ -9,9 +9,11 @@ export function attemptIdle(state, attemptId) { const sdk = Object.values(state.projectHarnessWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') const native = Object.values(state.projectNativeWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') if (Object.values(state.bots ?? {}).some(row => row.ephemeral && row.attemptId === attemptId && row.role !== 'chief')) return false - if (sdk.some(row => !confirmedWorkerExit(state, row) || !leases.some(lease => lease.id === row.leaseId && lease.status === 'released' && lease.generation === row.generation && lease.worker?.handle === row.id))) return false + if (sdk.some(row => !confirmedWorkerExit(state, row) || !leases.some(lease => lease.id === row.leaseId && lease.status === 'released' && lease.generation === row.generation && lease.projectId === row.projectId && lease.resourceId === row.resourceId && lease.role === row.role && lease.worker?.handle === row.id))) return false if (native.some(row => !nativeExitConfirmed(state, row))) return false if (leases.some(row => row.status !== 'released' || row.worker && !['sdk', 'native'].includes(row.worker.kind) || row.worker?.kind === 'sdk' && !sdk.some(worker => worker.id === row.worker.handle) || row.worker?.kind === 'native' && !native.some(worker => worker.id === row.worker.botId) || !row.worker && row.releaseProof?.kind !== 'unbound-reservation-release')) return false + const sessions = [...Object.values(attempt.roleSessions ?? {}), ...(attempt.roleSessionHistory ?? [])] + if (sessions.some(row => row.role !== 'chief' && row.botId && !sdk.some(worker => worker.id === row.botId) && !native.some(worker => worker.id === row.botId))) return false const claim = task.collaboration?.claim if (claim?.workerId && claim.role !== 'chief' && !claim.chiefRunId && !sdk.some(row => row.id === claim.workerId) && !native.some(row => row.id === claim.workerId)) return false if (claim?.status === 'active' && !claim.workerId && !claim.chiefRunId) return false diff --git a/core/test/project-harness.test.js b/core/test/project-harness.test.js index 28956eec2..8289af9b5 100644 --- a/core/test/project-harness.test.js +++ b/core/test/project-harness.test.js @@ -3,6 +3,7 @@ import { test } from 'node:test' import { mkdtemp, mkdir, writeFile, readFile, rm, stat } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { createProjectScheduler } from '../src/host/project-scheduler.js' import { createProjectEffects } from '../src/host/project-effects.js' import { install as installProjectInputs } from '../src/host/project-inputs.js' import { createProjectResources } from '../src/host/project-resources.js' @@ -526,3 +527,29 @@ if(m.type==='start')write({type:'completed',sessionId:'input-actor-thread',text: await assert.rejects(harness.input({ workerId: id, generation: rt.state.projectHarnessWorkers[id].generation, text: 'Late unauthorized input' }, { memberId: 'sender' }), /Sending member was removed/) assert.equal(rt.state.projectHarnessWorkers[id].status, 'waiting-input') }) + + +test('an actual SDK protocol subprocess must exit before its Chief input wait frees project capacity', async t => { + const { rt, harness, cwd } = await fixture(t) + rt.state.bots = {}; rt.state.projectAttempts = { a: { id: 'a', taskId: 't', projectId: 'p', status: 'running' } } + rt.state.projectExecutions = { run: { id: 'run', attemptId: 'a', taskId: 't', projectId: 'p', status: 'running' } } + rt.state.tasks.t.deliveryAttemptId = 'a' + const scheduler = createProjectScheduler(rt) + const id = await harness.spawn({ executionProjectId: 'p', executionWorkDir: cwd, prompt: 'Test execution exit and waiting capacity' }) + const result = new Promise(resolve => harness.onResult(resolve)) + await harness.dispatch({ botId: id, taskId: 't', brief: 'Capacity checkpoint', outputs: [{ name: 'implementation' }] }) + assert.equal((await result).ok, true) + rt.state.tasks.t.chiefQuestion = { decisionId: 'chief-input', reason: 'Chief requests a scoped project input' } + rt.state.tasks.t.collaboration = { status: 'blocked', claim: { workerId: id, role: 'developer', status: 'completed' }, history: [] } + await scheduler.captureWait(rt.state.tasks.t) + assert.equal(scheduler.slots('p'), 1); assert.equal(rt.projectResources.occupied('r'), 1) + assert.equal(scheduler.capacity('p').exitUnknown, 1) + await harness.release(id) + // The minimal Harness fixture omits role on its pre-existing lease; real reservations carry it. + rt.state.projectResourceLeases.l.role = 'developer' + assert.equal(rt.state.projectHarnessWorkers[id].releaseProof.kind, 'sdk-stop-ack-and-process-exit') + assert.equal(scheduler.slots('p'), 0); assert.equal(rt.projectResources.occupied('r'), 0) + assert.equal(scheduler.capacity('p').waitingInput, 1) + assert.equal(rt.state.projectAttempts.a.status, 'running'); assert.equal(rt.state.projectExecutions.run.status, 'running') + assert.equal(rt.state.projectHarnessWorkers[id].backendSessionId, 'bound-thread') +}) diff --git a/core/test/project-pools.test.js b/core/test/project-pools.test.js index 6c7d19d9b..9ca157a82 100644 --- a/core/test/project-pools.test.js +++ b/core/test/project-pools.test.js @@ -619,3 +619,17 @@ test('ambiguous developer capabilities require an explicit initial role rather t await f.pool.admit({ id: 'attempt', taskId: 'task' }, { memberId: 'owner' }, { targetRole: 'builder' }) assert.equal(f.task.collaboration.botId, 'p:builder') }) + +test('resumed pool claims cannot bypass project admission or spend a claim generation while queued', async () => { + const f = setup(); await f.admit() + let admitted = false, checks = 0 + f.rt.projectScheduler = { async admitPool(task, definition) { ++checks; assert.equal(task.id, 'task'); assert.equal(definition.role, 'developer'); return admitted } } + const generation = f.task.collaboration.generation + assert.equal(await f.claim(), null) + assert.equal(f.task.collaboration.generation, generation) + assert.equal(f.task.collaboration.history[0].status, 'pending') + admitted = true + const claim = await f.claim() + assert.equal(claim.status, 'active'); assert.equal(checks, 2) + await assert.rejects(f.pool.decide('task', { ...claim, generation: generation - 1 }, { action: 'complete', reason: 'late callback' }), /stale/) +}) diff --git a/docs/backend-operations.md b/docs/backend-operations.md index ceb6be1f6..04ab8538e 100644 --- a/docs/backend-operations.md +++ b/docs/backend-operations.md @@ -110,3 +110,6 @@ JSON/SQLite回退保留新代码格式守卫、ID和已完成Effects,**不是 [测试分层](testing-layers.md)区分单元、真实DSH/本机HTTP持久化、真实SDK与完整模型开发闭环。确定性测试供应商只出现在测试里,启动工具不创建mock模型或样例仓库。 项目自动化可选择 gh-cli 或 App installation,个人操作保留独立用户授权;配置步骤、最小权限和恢复方法见 [GitHub自动化](github-automation.md)。 + + +#2328:`project-resources` 的 `capacity` 区分 `activeSlots`、`waitingInput`、`waitingResume`、`exitUnknown`。`occupied` 是资源租约占用;已确认退出的输入等待可释放项目槽位,退出未知仍占槽。输入到达后由 Chief 决定恢复,原队列按项目配额、资源和权限重新准入;不要通过新建重复 Task 绕过等待。 diff --git a/docs/openapi/flybot.openapi.json b/docs/openapi/flybot.openapi.json index c26c01980..6842fda7a 100644 --- a/docs/openapi/flybot.openapi.json +++ b/docs/openapi/flybot.openapi.json @@ -1437,6 +1437,107 @@ "nextActions" ] }, + "CapacityState": { + "type": "object", + "properties": { + "activeSlots": { + "type": "integer", + "minimum": 0 + }, + "waitingInput": { + "type": "integer", + "minimum": 0 + }, + "waitingResume": { + "type": "integer", + "minimum": 0 + }, + "exitUnknown": { + "type": "integer", + "minimum": 0 + } + }, + "additionalProperties": true, + "required": [ + "activeSlots", + "waitingInput", + "waitingResume", + "exitUnknown" + ] + }, + "CapacityWait": { + "type": "object", + "properties": { + "version": { + "const": "project-capacity-wait/v1", + "type": "string" + }, + "source": { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": [ + "chief-input", + "children" + ] + }, + "id": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + } + }, + "additionalProperties": true, + "required": [ + "kind", + "id" + ] + }, + "status": { + "type": "string", + "enum": [ + "suspended", + "admitted" + ] + }, + "attemptId": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + }, + "previousAttemptId": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + }, + "at": { + "type": "integer", + "minimum": 0 + }, + "queueId": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + }, + "admittedAt": { + "type": "integer", + "minimum": 0 + } + }, + "additionalProperties": true, + "required": [ + "version", + "source", + "status", + "attemptId", + "at" + ] + }, "Attempt": { "type": "object", "properties": { @@ -1502,6 +1603,9 @@ "requirement": { "$ref": "#/components/schemas/JsonValue" }, + "capacityWait": { + "$ref": "#/components/schemas/CapacityWait" + }, "workspaces": { "type": "object", "additionalProperties": { @@ -1568,6 +1672,26 @@ "maxLength": 200, "description": "Opaque server-issued ID; never infer authority from it." }, + "kind": { + "const": "resume", + "type": "string" + }, + "handoffId": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + }, + "generation": { + "type": "integer", + "minimum": 0 + }, + "targetBotId": { + "type": "string", + "minLength": 1, + "maxLength": 200, + "description": "Opaque server-issued ID; never infer authority from it." + }, "projectId": { "type": "string", "minLength": 1, @@ -18974,6 +19098,16 @@ "type": "integer", "minimum": 0 }, + "capacity": { + "anyOf": [ + { + "$ref": "#/components/schemas/CapacityState" + }, + { + "type": "null" + } + ] + }, "repository": { "$ref": "#/components/schemas/JsonValue" }, @@ -29235,7 +29369,7 @@ "notes": "Read-only transport/initiative history; it is not role claim acceptance.", "implementation": { "path": "core/src/host/project-pools.js", - "sha256": "afab16b3df4d18a5bfab69c975e8d61346a7838438a634668f2b06425545f237" + "sha256": "17cce40f78b034db3b069a87019b0a28a72a43a8449c69bc937bcc6e440061bd" } }, { @@ -29313,7 +29447,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-pools.js", - "sha256": "afab16b3df4d18a5bfab69c975e8d61346a7838438a634668f2b06425545f237" + "sha256": "17cce40f78b034db3b069a87019b0a28a72a43a8449c69bc937bcc6e440061bd" } }, { @@ -29795,7 +29929,7 @@ "notes": "Although a query, the downstream integration checkpoint gate requires manager authority and confirmed release.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -29873,7 +30007,7 @@ "notes": "UTF-8 byte offsets, not character offsets; nextOffset preserves character boundaries. Rechecks rights after source I/O. No caller path.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -30034,7 +30168,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" } }, { @@ -31812,7 +31946,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" } }, { @@ -31922,7 +32056,7 @@ "notes": "New Project Task contracts require actual Chief adjudication; third-party clients cannot forge Chief credentials.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32030,7 +32164,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32137,7 +32271,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32231,7 +32365,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32325,7 +32459,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32434,7 +32568,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" } }, { @@ -32784,7 +32918,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" } }, { @@ -32910,7 +33044,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" } }, { @@ -33009,7 +33143,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" } }, { @@ -33109,7 +33243,7 @@ "notes": "30-minute inspection is a configurable schedule default, not a framework deadline. correctionBudget defaults to 2 (0–4); protocol correction reuses the Chief Session. Exhaustion is incomplete judgement, never automatic consultation or default routing.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" } }, { @@ -34143,7 +34277,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/workbench-state.js", - "sha256": "be02e9b796f315d529e09b3c53a1804a7a5d92c389a34c8f1017fcc03b114d0b" + "sha256": "b95b83fe47a6a155e16fba92284f9df2929ebd95bec635804f186c47b9eb8391" } }, { @@ -37381,7 +37515,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/workbench-state.js", - "sha256": "be02e9b796f315d529e09b3c53a1804a7a5d92c389a34c8f1017fcc03b114d0b" + "sha256": "b95b83fe47a6a155e16fba92284f9df2929ebd95bec635804f186c47b9eb8391" } }, { @@ -37456,7 +37590,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/workbench-state.js", - "sha256": "be02e9b796f315d529e09b3c53a1804a7a5d92c389a34c8f1017fcc03b114d0b" + "sha256": "b95b83fe47a6a155e16fba92284f9df2929ebd95bec635804f186c47b9eb8391" } }, { @@ -42398,7 +42532,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-pools.js", - "sha256": "afab16b3df4d18a5bfab69c975e8d61346a7838438a634668f2b06425545f237" + "sha256": "17cce40f78b034db3b069a87019b0a28a72a43a8449c69bc937bcc6e440061bd" }, "aliasOf": "development.query.handoff.get" }, @@ -42474,7 +42608,7 @@ "notes": "Read-only transport/initiative history; it is not role claim acceptance.", "implementation": { "path": "core/src/host/project-pools.js", - "sha256": "afab16b3df4d18a5bfab69c975e8d61346a7838438a634668f2b06425545f237" + "sha256": "17cce40f78b034db3b069a87019b0a28a72a43a8449c69bc937bcc6e440061bd" }, "aliasOf": "development.query.handoff.list" }, @@ -42678,7 +42812,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" }, "aliasOf": "development.command.task.run" }, @@ -42851,7 +42985,7 @@ "notes": "New Project Task contracts require actual Chief adjudication; third-party clients cannot forge Chief credentials.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.command.task.accept" }, @@ -42944,7 +43078,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.command.task.reject" }, @@ -43036,7 +43170,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.command.attempt.stop" }, @@ -43128,7 +43262,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.command.attempt.cleanup" }, @@ -43507,7 +43641,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.command.attempt.publication-retry" }, @@ -43583,7 +43717,7 @@ "notes": "UTF-8 byte offsets, not character offsets; nextOffset preserves character boundaries. Rechecks rights after source I/O. No caller path.", "implementation": { "path": "core/src/host/project-delivery.js", - "sha256": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b" + "sha256": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8" }, "aliasOf": "development.query.attempt.log" }, @@ -43681,7 +43815,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" }, "aliasOf": "development.command.queue.cancel" }, @@ -43794,7 +43928,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee" + "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" }, "aliasOf": "development.command.queue.retry" }, @@ -43891,7 +44025,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" }, "aliasOf": "development.command.chief.inspect" }, @@ -43989,7 +44123,7 @@ "notes": "30-minute inspection is a configurable schedule default, not a framework deadline. correctionBudget defaults to 2 (0–4); protocol correction reuses the Chief Session. Exhaustion is incomplete judgement, never automatic consultation or default routing.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" }, "aliasOf": "development.command.chief.configure" }, @@ -44065,7 +44199,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-chief-control.js", - "sha256": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f" + "sha256": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a" }, "aliasOf": "development.query.chief.state" }, @@ -44929,7 +45063,7 @@ "core/src/host/project-agents.js": "73970119c983c9c20a05fbe15f9342b7837a431d68d8c57a93a356830b75fada", "core/src/host/project-assistant.js": "4fa4309d4aa7abf78b6ba4e55ba754f320384f931e0e4a5c6620c859e1b50676", "core/src/host/project-board.js": "dfb0b4bc05ae42335632e3edbba883a03c5da2fb17899b23e8226320ea08aec4", - "core/src/host/project-chief-control.js": "ecfbd22b926bb52ea9929472a2e718be48fa9624aaf2ac0ec1fe833edad3d45f", + "core/src/host/project-chief-control.js": "2f325b1b6043fd39b1c3603889f5458f4583732cfa5a42cec18d6d915fd2e67a", "core/src/host/project-chief-protocol.js": "a09d85e04171b22625d3eed131dd6a5227dc1a946ad39d8a59f017bb75f71936", "core/src/host/project-chief.js": "a3f01a053177fd8f10d4b778d79ac98c25aee70aa3c4d967dc7e1da2aea2cf6e", "core/src/host/project-child-inputs.js": "db1f534c4222dba1b500d308b1cd07f0df50a02eb228f520b150a3b79d6f126d", @@ -44941,12 +45075,13 @@ "core/src/host/project-credentials.js": "18fe61057fb920664fc5077d81eee0957f7eaf64821dd10d2d2a3cd463f5fdea", "core/src/host/project-delegations.js": "d4d86192551994a3419e0d2ca0f64bc4d50a75960970d2b79ed9f97b82940903", "core/src/host/project-delivery-template.js": "daa5cc8185abc1ec84421ae3fcd5574a50573cfd1a7c4cb10941158e29c79a13", - "core/src/host/project-delivery.js": "31adb25fe196913628d67ee18b87a18015f07bc5746e824f15732a72d4443f2b", + "core/src/host/project-delivery.js": "aa71f5d8a9f2390c5814b4e7ec7f56db8a1f32472beb7f53791b7ce6477299c8", "core/src/host/project-development-service.js": "cb1e3b1cda8656f2add087f43be839776d9f6d2e93965d360bf3c8c2b1462178", "core/src/host/project-domain.js": "6d6a20727905005f7a8de0247a429a10ffbd1436773ee96a5d7615b2fb763e82", "core/src/host/project-effects.js": "5557782333a0b3437dacde2b6924fe12ea54225098571fa8e404d79a432be549", "core/src/host/project-execution-authority.js": "9c5ffb4bc9153d8f9d9e7901f7aef2aee90cacc19f19d8a0726ac0adb93e3e36", "core/src/host/project-execution-boundary.js": "9e7afe568024e19517a3d5430a0483a6ea5e2c910a53de056d83db3896d3eb16", + "core/src/host/project-execution-capacity.js": "751b48e26f6ed1ad4c000337174b39ab4c113c494cbb7bd5a2a8369cbae10aa4", "core/src/host/project-execution-exit.js": "50f94881a40c2d0c6288fd081c885c7cabb71f8abec3a8c918e1c0bc852a04f7", "core/src/host/project-execution-profile.js": "7106593979d6e21b9e1dab567c8628a7d0252976ed47d42dace337944725da59", "core/src/host/project-git-sandbox.js": "b579b01dcce0708275e9f735d08932481369803382b8f7d24af69639608d724f", @@ -44962,7 +45097,7 @@ "core/src/host/project-merge-heads.js": "9e5e39b8aaa191c2175feaafcf6524b8a16dbafdfa573f94222b9f71d8482074", "core/src/host/project-metrics.js": "80f209d76955e5a3058fa66757543059fded1030625e5e155be5254494ce5264", "core/src/host/project-native-sessions.js": "d8b796fce7a03252f7e6f10c2725ff97501ab4ba81492d8bdb8da1807a2db581", - "core/src/host/project-pools.js": "afab16b3df4d18a5bfab69c975e8d61346a7838438a634668f2b06425545f237", + "core/src/host/project-pools.js": "17cce40f78b034db3b069a87019b0a28a72a43a8449c69bc937bcc6e440061bd", "core/src/host/project-private-file.js": "e8b51d8da7bcd72658eb6eb953dbf9d0054ea0ee519b0229785e8ea0aacef6df", "core/src/host/project-process-proof.js": "85f4ed03f5191b842ca899b712c99177c059658d43a66e8ddeedc1c06eafcdea", "core/src/host/project-release-github.js": "47c52d0057dd39f1ee627c619970efad36e0134bc0c7bc218d43187230d51d64", @@ -44977,7 +45112,7 @@ "core/src/host/project-rooms.js": "148ba6f39086dc74e0c9c92ab368361b256b5185490d7983e957dd28890cfbe9", "core/src/host/project-runtime-setup.js": "7d4ff3f84c0b08540145b40c150262e8d85be957def7f14a06496064f20d51e0", "core/src/host/project-scheduled-tasks.js": "f99a7e28a95c4525672c88059878ede4b88c45b5b668dfc25f3a1dd519bab3c0", - "core/src/host/project-scheduler.js": "40ca98a131d8aad4e2b354c8a0608a07e42cc24b2755ca13159be491d021f9ee", + "core/src/host/project-scheduler.js": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326", "core/src/host/project-service-events.js": "71917d1ab367a9db8bef8c1d1003228255a57528e84a57e76de3a39d08ac28ec", "core/src/host/project-service-sse.js": "df97d6b11156da885a6a47d50a4a77c8c99b6af5eb0066de14312cc512762297", "core/src/host/project-service-views.js": "e34b36f78f4dcbb7e1f507203ba843959fa76a119fdecdc5b9902c39bbbef3fa", @@ -45020,7 +45155,7 @@ "core/src/host/work-dispatch.js": "bef48285a381a6bbc7e1c162656316c4769cca76fd44da0e4fcfed54fe38d641", "core/src/host/workbench-gateway.js": "6c2beeec490557816d5a82c8992048e43b1ec39e3d40d85bcfc027684f8a65b4", "core/src/host/workbench-personal.js": "4282e3983f8f5cba0e61f288bddcba7f8e44eab92fbffbc604acac55defdc301", - "core/src/host/workbench-state.js": "be02e9b796f315d529e09b3c53a1804a7a5d92c389a34c8f1017fcc03b114d0b", + "core/src/host/workbench-state.js": "b95b83fe47a6a155e16fba92284f9df2929ebd95bec635804f186c47b9eb8391", "core/src/host/workflow-facts.js": "f2ff8392df1d0f927dd3bc2fb6ed1fde0bb31b5a8d590725963134b73f09fe76", "core/src/host/workflow-store.js": "917662ab929311d4fbe7937aec42589478a03f4baed3cb010255d8a05ecb1261", "core/src/shared/project-policy.js": "7c9e69378c7f8a52b1a2cb538858d670411930d7439ca4d0589f9ce41d6146ed", diff --git a/docs/subsystems/backend-service.md b/docs/subsystems/backend-service.md index 95592b0f9..fef34e23c 100644 --- a/docs/subsystems/backend-service.md +++ b/docs/subsystems/backend-service.md @@ -120,3 +120,6 @@ transport. Source API changes require deliberate catalog and guide review. #2327 documents correctionBudget and typed Chief correction/run history in OpenAPI. The actual production-entry HTTP/runtime regression injects malformed admission in two flows, verifies feedback-driven correction and retains restart/auth/private-consultation coverage. Its model is deterministic; it is not a paid SDK acceptance run. + + +#2328 documents CapacityState/CapacityWait and durable resume QueueEntry metadata in the existing OpenAPI. project-resources exposes the same capacity predicates used for admission; continuation cancellation remains a Chief/Task operation. diff --git a/docs/subsystems/project-chief-control.md b/docs/subsystems/project-chief-control.md index 0ee3937bd..f2b954eb0 100644 --- a/docs/subsystems/project-chief-control.md +++ b/docs/subsystems/project-chief-control.md @@ -215,3 +215,6 @@ escalation. Chief can later independently choose consultation, repair or another route through the original inspection interface. A valid decision clears only its Task's current issue marker, retaining failed invocation/audit history. Separate per-Task execution failure isolation and scheduled restart recovery remain #2330. + + +#2328 captures an explicit Chief input-wait checkpoint after needs-human. The host releases project execution capacity only with current exit evidence; input receipt does not resume work until Chief routes and the original scheduler re-admits it. diff --git a/docs/subsystems/project-delivery.md b/docs/subsystems/project-delivery.md index efa854c68..0c6f4e1fc 100644 --- a/docs/subsystems/project-delivery.md +++ b/docs/subsystems/project-delivery.md @@ -566,3 +566,6 @@ Publication now selects either credential provider through the existing broker, ## Workspace baseline and explicit inputs — #2325 #2325 validates the actual checkout of every newly materialized/reused workspace against candidate/seed/base precedence, records materialization facts and transfers missing pinned base/upstream objects to resource clones. A changed baseline cannot overwrite an unsealed draft: explicit committed input admission permits preserved-draft integration. Rework retains same-Task approved input admissions. Inputs may be purpose=inspect (object availability only) or integrate (default, must be proven merged before sealing). Chief route/return can select exact refs/SHAs from its scoped operator-source snapshot through the original admission/Effects boundary; this does not fetch arbitrary remote URLs or grant credentials. + + +#2328 carries an unadmitted capacityWait into a successor Attempt during controlled rework. The previous Attempt remains historical; the new claim must regain project capacity before execution. In-place continuation retains the same Attempt and workspace. diff --git a/docs/subsystems/project-harness.md b/docs/subsystems/project-harness.md index e27676541..158c2a56b 100644 --- a/docs/subsystems/project-harness.md +++ b/docs/subsystems/project-harness.md @@ -372,3 +372,6 @@ Current operator disabling of a Resource or removal of its Harness support also Native input budgeting fences the actual `claim.workerId` against the temporary carrier; `claim.botId` identifies the logical role and is not a physical worker. This distinction is covered by both the scoped input regression and actual DSH native return probe. #2292 current execution authority additionally validates an Attempt repository association and its operator capability. Removing/replacing the association or withdrawing provisioning cannot keep authorizing an old worker through a changed Project default. Unknown stop/exit still retains capacity. + + +#2328 extends the actual SDK protocol subprocess test: result completion alone retains project/resource capacity; only stop acknowledgement plus process exit releases the input wait. The fixture bridge is deterministic and does not perform vendor inference. diff --git a/docs/subsystems/project-pools.md b/docs/subsystems/project-pools.md index b7cf58d0e..a0cf97c3a 100644 --- a/docs/subsystems/project-pools.md +++ b/docs/subsystems/project-pools.md @@ -184,3 +184,6 @@ Canonical claims retain the admitted requirement revision. Decision admission re ## Initial routing (#2292) `initialRoleTarget` pins role/id/capability, catalog revision and Task requirements revision before first dispatch. An explicit active non-Chief `targetRole` wins; otherwise prefer the active developer-capability role named developer, or the sole developer-capability role. Ambiguous/missing candidates refuse rather than choose by array order. Admission revalidates the binding after asynchronous authority/evidence checks. This selection is not a new Session or a DAG edge; subsequent handoffs retain existing Bot autonomy. + + +#2328 preserves wait checkpoints before handoff/child resume. Claim acceptance waits for original projectQueue resume admission outside the pool lock, rechecks authority/generation and does not spend a claim generation while queued. No extra workflow or default next role is added. diff --git a/docs/subsystems/project-workbench.md b/docs/subsystems/project-workbench.md index 2f93e2112..65ec2b873 100644 --- a/docs/subsystems/project-workbench.md +++ b/docs/subsystems/project-workbench.md @@ -3,6 +3,7 @@ subsystem: project-workbench owned_paths: - core/src/host/project-task-repository.js - core/test/project-task-repository.test.js + - core/src/host/project-execution-capacity.js - core/src/host/project-scheduler.js - core/src/host/project-repository-policy.js - core/src/host/workbench-gateway.js @@ -489,3 +490,26 @@ The source backend launcher reuses this original gateway/UI and account/resource #2292 repository scope: scheduler preflight/queued retries use the persisted Task repositoryBinding, not mutable Project defaults. Bindings identify the current Project association; removal/replacement blocks further execution. Legacy queues pin a choice before dispatch. Only repository-free decision Tasks bypass Git/test provisioning; code/release Tasks still require an associated operator-provisioned repository. #2292 exposes scoped release request/reconciliation/status on the original gateway. Release commands require manager authority; status remains a Project read, and only status is admitted on a read-only host. No new service listener or UI is introduced. + +## Confirmed idle waits and continuation admission — #2328 + +Chief explicitly requests input; the scheduler persists Attempt.capacityWait and +excludes that Attempt and its pool Run together only when every SDK/native worker +has trusted matching exit proof, resource leases are released and no ephemeral Bot +or unidentified claim remains. A legacy active Workflow retains its slot. Merely +blocked, an empty resource count, a timeout or a missing PID never proves idleness. +Run/Attempt status, Task identity, candidates, workspace and Session history remain. + +Chief routing and child continuation preserve the wait checkpoint before clearing +its source. A successor inherits that checkpoint. Role-pool acceptance defers to a +`kind=resume` entry in the original projectQueue before spending a claim generation. +New and resumed work share FIFO project quota and resource/authority checks; there +is no new executor or queue. Admission runs outside the pool lock; a save in progress +cannot authorize dispatch. Uncommitted save, revoked principal, retired role, changed +handoff and stale generation cannot start continuation. Replaying inputs/claims is +idempotent. Continuation cancellation belongs to Task/Chief lifecycle; blocked queue +retry preserves the initiating principal and exact Attempt. + +`project-resources.capacity` reports activeSlots, waitingInput, waitingResume and +exitUnknown using the actual admission predicates. occupied measures resource leases; +activeSlots measures project execution. They have independent configured limits. diff --git a/patches/ds-bot/files/src/host/project-chief-control.js b/patches/ds-bot/files/src/host/project-chief-control.js index b3e62aed6..bbb952347 100644 --- a/patches/ds-bot/files/src/host/project-chief-control.js +++ b/patches/ds-bot/files/src/host/project-chief-control.js @@ -165,6 +165,7 @@ export function createProjectChiefControl(rt) { if (task.collaboration) { task.collaboration.status = 'blocked'; task.collaboration.block = { reason, chiefDecisionId: audit.id, at: Date.now() } } if (task.collaboration?.claim?.chiefRunId === run.id) task.collaboration.claim.status = 'waiting-human' task.chiefQuestion = { decisionId: audit.id, reason, at: Date.now() } + await rt.projectScheduler?.captureWait?.(task) task.chiefQuestion.recipient = rt.projectAgents.contactUser?.(task, { id: audit.id, reason, requiresHuman: decision.requiresHuman === true }, actor) ?? { kind: 'human', memberId: task.collaboration?.initiatedBy ?? actor.memberId } } else if (decision.action === 'close') { rt.projectTaskLifecycle?.requireCompletedDependencies(task) diff --git a/patches/ds-bot/files/src/host/project-delivery.js b/patches/ds-bot/files/src/host/project-delivery.js index 329e0e390..b2f26f008 100644 --- a/patches/ds-bot/files/src/host/project-delivery.js +++ b/patches/ds-bot/files/src/host/project-delivery.js @@ -1042,6 +1042,7 @@ export function createProjectDelivery(rt, { publisher, releaseProvider } = {}) { if (['running', 'waiting-acceptance'].includes(previous.status)) previous.status = 'returned' previous.returned = { at: Date.now(), by: claim.botId, reason } const next = { id, protocol: previous.protocol, projectId: previous.projectId, taskId: task.id, requestKey: hash(`${task.projectId}\n${task.id}\n${claim.id}`), workflowVersion: previous.protocol === 'project-handoff/v1' ? 'project-delivery/pools-v4' : previous.workflowVersion, status: 'running', initiatedBy: previous.initiatedBy, executionOwnerId: previous.executionOwnerId, baseSha: previous.requirement.repository && !previous.candidateSha ? await git(settingsOf(projectOf(task.projectId, previous)).path, ['rev-parse', 'HEAD']) : previous.baseSha, seedSha: previous.candidateSha ?? previous.seedSha, ...(publicationBinding ? { publicationBinding } : {}), ...(previous.integrationInputs ? { integrationInputs: structuredClone(previous.integrationInputs), integrationAdmissions: structuredClone(previous.integrationAdmissions ?? []) } : {}), repositoryBinding: previous.repositoryBinding, requirement: requirementsOf(task, previous.requirement.repository), feedback: { previousAttemptId: previous.id, kind: 'bot-return', reason, candidateSha: previous.candidateSha }, agentDefinitions: rt.projectAgents.deliveryDefinitions(task.projectId), roleSessions: {}, rolePolicies: {}, resourceLeases: {}, workspaces: {}, evidence: {}, createdAt: Date.now(), updatedAt: Date.now(), runId: `pool-run-${randomUUID()}` } + if (previous.capacityWait && previous.capacityWait.status !== 'admitted') next.capacityWait = { ...structuredClone(previous.capacityWait), attemptId: id, previousAttemptId: previous.id } previous.supersededBy = id setEntry(attempts(), id, next); task.deliveryAttemptId = id rt.bindProject({ kind: 'run', id: previous.runId, projectId: task.projectId, taskId: task.id, attemptId: previous.id, status: 'failed' }) diff --git a/patches/ds-bot/files/src/host/project-execution-capacity.js b/patches/ds-bot/files/src/host/project-execution-capacity.js new file mode 100644 index 000000000..b4dbb2bce --- /dev/null +++ b/patches/ds-bot/files/src/host/project-execution-capacity.js @@ -0,0 +1,26 @@ +// Waiting is a Chief/Task decision. Capacity suspension additionally requires host exit facts. +import { confirmedWorkerExit } from './project-task-sessions.js' +import { nativeExitConfirmed } from './project-native-sessions.js' +export function attemptIdle(state, attemptId) { + const attempt = state.projectAttempts?.[attemptId], task = state.tasks?.[attempt?.taskId] + if (!attempt || !task || task.projectId !== attempt.projectId) return false + if (attempt.workflowDefinition && Object.values(state.projectExecutions ?? {}).some(row => row.attemptId === attemptId && ['queued', 'running', 'working', 'paused', 'legacy'].includes(row.status))) return false + const leases = Object.values(state.projectResourceLeases ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + const sdk = Object.values(state.projectHarnessWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + const native = Object.values(state.projectNativeWorkers ?? {}).filter(row => row.attemptId === attemptId && row.role !== 'chief') + if (Object.values(state.bots ?? {}).some(row => row.ephemeral && row.attemptId === attemptId && row.role !== 'chief')) return false + if (sdk.some(row => !confirmedWorkerExit(state, row) || !leases.some(lease => lease.id === row.leaseId && lease.status === 'released' && lease.generation === row.generation && lease.projectId === row.projectId && lease.resourceId === row.resourceId && lease.role === row.role && lease.worker?.handle === row.id))) return false + if (native.some(row => !nativeExitConfirmed(state, row))) return false + if (leases.some(row => row.status !== 'released' || row.worker && !['sdk', 'native'].includes(row.worker.kind) || row.worker?.kind === 'sdk' && !sdk.some(worker => worker.id === row.worker.handle) || row.worker?.kind === 'native' && !native.some(worker => worker.id === row.worker.botId) || !row.worker && row.releaseProof?.kind !== 'unbound-reservation-release')) return false + const sessions = [...Object.values(attempt.roleSessions ?? {}), ...(attempt.roleSessionHistory ?? [])] + if (sessions.some(row => row.role !== 'chief' && row.botId && !sdk.some(worker => worker.id === row.botId) && !native.some(worker => worker.id === row.botId))) return false + const claim = task.collaboration?.claim + if (claim?.workerId && claim.role !== 'chief' && !claim.chiefRunId && !sdk.some(row => row.id === claim.workerId) && !native.some(row => row.id === claim.workerId)) return false + if (claim?.status === 'active' && !claim.workerId && !claim.chiefRunId) return false + // All adapters register before external execution; no records means no worker started. + return true +} +export function capacitySuspended(state, attempt) { + const task = state.tasks?.[attempt.taskId] + return Boolean(attempt.capacityWait && attempt.capacityWait.status !== 'admitted' && task?.deliveryAttemptId === attempt.id && attemptIdle(state, attempt.id)) +} diff --git a/patches/ds-bot/files/src/host/project-pools.js b/patches/ds-bot/files/src/host/project-pools.js index 51aa605d9..2d20659b9 100644 --- a/patches/ds-bot/files/src/host/project-pools.js +++ b/patches/ds-bot/files/src/host/project-pools.js @@ -152,6 +152,7 @@ export function createProjectPools(rt) { } } row.contextRefs = { source: contextReference(rt.state, task, fromRole), target: contextReference(rt.state, task, target.role) } + await rt.projectScheduler?.captureWait?.(task) const before = rt.projectEffects ? { collaboration: clone(c), handoff: task.handoff, question: task.chiefQuestion } : null let input, intent try { @@ -238,7 +239,10 @@ export function createProjectPools(rt) { definition = definitions(task.projectId).roles.find(row => row.id === botId) if (!definition) { c.status = 'blocked'; c.block = { reason: 'Target role is retired; Chief must explicitly select an active role', at: Date.now() }; await rt.save(); return null } if (Object.values(rt.state.tasks).some(row => row.collaboration?.claim?.botId === botId && row.collaboration.claim.status === 'active')) return null - if (adapter.ready && !adapter.ready({ task, definition })) return null + if (adapter.ready && !await adapter.ready({ task, definition })) return null + if (rt.projectScheduler?.admitPool && !await rt.projectScheduler.admitPool(task, definition)) return null + await authority(task) + if (c.generation !== expectedGeneration || c.status !== 'queued' || c.botId !== botId) return null const receipt = c.history.find(row => row.id === c.currentHandoffId) if (!receipt || receipt.status !== 'pending') throw refuse('Task has no pending handoff receipt') const limit = rt.config.projectHandoff?.quota?.steps @@ -523,6 +527,7 @@ export function createProjectPools(rt) { if (existing && existing.id !== c.attemptId && existing.feedback?.previousAttemptId === c.attemptId && rt.state.projectAttempts[c.attemptId]?.supersededBy === existing.id) c.attemptId = existing.id else c.attemptId = await adapter.rework?.({ task, claim: c.claim, reason: `Integrate adjudicated child results: ${wait.resumePoint}` }) ?? c.attemptId } + await rt.projectScheduler?.captureWait?.(task, { kind: 'children', id: wait.id }) const previous = wait.status wait.status = 'queued' try { diff --git a/patches/ds-bot/files/src/host/project-scheduler.js b/patches/ds-bot/files/src/host/project-scheduler.js index 5a48bec91..2ec00f651 100644 --- a/patches/ds-bot/files/src/host/project-scheduler.js +++ b/patches/ds-bot/files/src/host/project-scheduler.js @@ -3,6 +3,7 @@ import { createHash, randomUUID } from 'node:crypto' import { ownEntry, setEntry } from './store.js' import { taskRepository } from './project-task-repository.js' import { initialRoleTarget } from './project-role-admission.js' +import { attemptIdle, capacitySuspended } from './project-execution-capacity.js' import { preflightRepository } from './project-repository-policy.js' const ACTIVE = new Set(['queued', 'running', 'working', 'paused', 'legacy']) const error = message => Object.assign(new Error(message), { code: 'bot/invalid' }) @@ -17,6 +18,7 @@ export function releasedParentWait(state, attemptId) { export function createProjectScheduler(rt, { preflight = preflightRepository } = {}) { const { state } = rt const entries = () => (state.projectQueue ??= {}) + const savingAdmissions = new Set() let recovered = false let tail = Promise.resolve() let waking = null, wakeAgain = false @@ -28,11 +30,49 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = for (const bot of Object.values(state.bots)) if (bot.projectId === projectId && bot.ephemeral && bot.attemptId) keys.add(bot.attemptId) for (const lease of Object.values(state.projectResourceLeases ?? {})) if (lease.projectId === projectId && lease.role !== 'chief' && lease.status !== 'released' && lease.attemptId) keys.add(lease.attemptId) for (const key of [...keys]) { - if (releasedParentWait(state, key)) keys.delete(key) + if (releasedParentWait(state, key) || capacitySuspended(state, state.projectAttempts?.[key] ?? {})) keys.delete(key) } return keys.size } const quota = project => project.quota?.maxConcurrent ?? 2 + async function captureWait(task, source) { + const attempt = state.projectAttempts?.[task.deliveryAttemptId] + const wait = source ?? (task.chiefQuestion ? { kind: 'chief-input', id: task.chiefQuestion.decisionId } : task.collaboration?.status === 'waiting-children' ? { kind: 'children', id: task.projectTask?.wait?.id ?? task.collaboration.currentHandoffId } : null) + if (!attempt || !wait || attempt.taskId !== task.id) return + if (attempt.capacityWait?.source.id === wait.id && attempt.capacityWait.status !== 'admitted') return + const previous = attempt.capacityWait + attempt.capacityWait = { version: 'project-capacity-wait/v1', source: structuredClone(wait), status: 'suspended', attemptId: attempt.id, at: Date.now() } + try { await rt.save() } catch (error) { if (previous === undefined) delete attempt.capacityWait; else attempt.capacityWait = previous; throw error } + } + function capacity(projectId) { + const result = { activeSlots: slots(projectId), waitingInput: 0, waitingResume: 0, exitUnknown: 0 } + for (const attempt of Object.values(state.projectAttempts ?? {})) { + const task = state.tasks?.[attempt.taskId] + if (attempt.projectId !== projectId || task?.deliveryAttemptId !== attempt.id || !attempt.capacityWait || ['accepted', 'cancelled'].includes(task.status)) continue + if (attempt.capacityWait.status === 'admitted') continue + if (!attemptIdle(state, attempt.id)) ++result.exitUnknown + else if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') ++result.waitingInput + else ++result.waitingResume + } + return result + } + async function admitPool(task, definition) { + if (definition.role === 'chief') return true + const attempt = state.projectAttempts?.[task.deliveryAttemptId], wait = attempt?.capacityWait + if (!wait) return true + if (wait.status === 'admitted') return !savingAdmissions.has(attempt.id) + if (task.chiefQuestion || task.collaboration?.status !== 'queued' || !attemptIdle(state, attempt.id)) return false + const c = task.collaboration + let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) + if (!entry) { + entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } + setEntry(entries(), entry.id, entry) + try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } + } + // Queue admission runs outside the pool lock: drain can itself admit a new pool Task. + void wake() + return false + } function revalidate(projectId) { let changed = false for (const entry of Object.values(entries())) { @@ -81,7 +121,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = if (Object.values(state.projectAttempts ?? {}).some(attempt => attempt.taskId === task.id && ['preparing', 'running', 'waiting-acceptance'].includes(attempt.status))) throw error('Task already has an active Attempt') const initialTarget = initialRoleTarget(rt, task, body) const resourceReason = resourceBlock(project.id, initialTarget) - if (slots(project.id) < quota(project) && !resourceReason) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) + if (slots(project.id) < quota(project) && !resourceReason && !Object.values(entries()).some(row => row.projectId === project.id && row.status === 'waiting')) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) const entry = { id: `queue-${randomUUID()}`, projectId: project.id, taskId: task.id, requestId: body.requestId, requestKey, memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry) @@ -91,12 +131,33 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = async function drain() { if (rt.readOnly?.() || !rt.sharedProjectMode?.() || !rt.flybotIsRead?.('workflow/state')) return let changed = false + for (const task of Object.values(state.tasks)) if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') await captureWait(task) for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) for (const entry of pending) { const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue + if (entry.kind === 'resume') { + const task = state.tasks[entry.taskId], attempt = state.projectAttempts?.[entry.attemptId], c = task?.collaboration + if (!attempt || task.deliveryAttemptId !== attempt.id || c?.generation !== entry.generation || c.currentHandoffId !== entry.handoffId || c.botId !== entry.targetBotId || c.status !== 'queued' || task.chiefQuestion) { entry.status = 'cancelled'; entry.reason = 'Continuation changed'; changed = true; continue } + if (!attemptIdle(state, attempt.id)) { entry.reason = 'Execution exit unconfirmed'; changed = true; continue } + try { + await rt.projectDelegations?.background(entry.memberId) + rt.requireProjectAccess({ memberId: entry.memberId }, project.id, 'task-write', task) + rt.projectTaskLifecycle?.requireExecutable(task) + const catalog = rt.projectAgents?.deliveryDefinitions(project.id) + if (catalog && !catalog.roles.some(row => row.id === entry.targetBotId && row.status !== 'retired')) throw error('Continuation role was retired') + const blocked = resourceBlock(project.id, entry.initialTarget) + if (blocked) { entry.reason = blocked; changed = true; continue } + const previous = structuredClone(attempt.capacityWait) + savingAdmissions.add(attempt.id) + attempt.capacityWait.status = 'admitted'; attempt.capacityWait.queueId = entry.id; attempt.capacityWait.admittedAt = Date.now() + entry.status = 'started'; entry.updatedAt = Date.now(); delete entry.reason + try { await rt.save() } catch (failure) { attempt.capacityWait = previous; entry.status = 'waiting'; throw failure } finally { savingAdmissions.delete(attempt.id) } + } catch (failure) { entry.status = 'blocked'; entry.reason = String(failure.message).slice(0, 300) } + changed = true; continue + } let execution try { execution = taskRepository(rt, project, state.tasks[entry.taskId], { repositoryBinding: entry.repositoryBinding }); await preflight(rt, execution.project) } catch (failure) { entry.status = 'blocked'; entry.reason = failure.message; entry.updatedAt = Date.now(); changed = true; continue } @@ -126,6 +187,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = if (!entry) throw error('Unknown queued request') const task = ownEntry(state.tasks, entry.taskId) rt.requireProjectAccess(actor, entry.projectId, 'task-write', task) + if (entry.kind === 'resume') throw error('Cancel or change the Task through Chief; a continuation cannot become a new request') if (!['waiting', 'blocked'].includes(entry.status)) throw error('Queued request has already started or ended') entry.status = 'cancelled'; entry.updatedAt = Date.now(); entry.reason = 'Cancelled by an authorized member' await rt.save() @@ -137,6 +199,12 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const task = ownEntry(state.tasks, entry.taskId) rt.requireProjectAccess(actor, entry.projectId, 'run-control') if (!task || task.projectId !== entry.projectId || ['accepted', 'cancelled'].includes(task.status)) throw error('Task has ended or is unavailable') + if (entry.kind === 'resume') { + if (entry.status !== 'blocked') throw error('Only a blocked continuation can be retried') + await rt.projectDelegations?.background(entry.memberId) + if (!rt.projectCan({ memberId: entry.memberId }, entry.projectId, 'task-write', task)) throw error('Original requester no longer has Project authority') + entry.status = 'waiting'; delete entry.reason; await rt.save(); return { queue: structuredClone(entry) } + } if (entry.status !== 'blocked') throw error('Only a blocked queue request can be retried') // Recovery must not replace the initiating human with the operator's authority. const requester = { memberId: entry.memberId } @@ -191,7 +259,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = } return { request: (body, actor) => serial(() => request(body, actor)), cancel: (body, actor) => serial(() => cancel(body, actor)), retry: (body, actor) => serial(() => retry(body, actor)).then(result => { void wake(); return result }), - slots, revalidate, recover, drain: () => serial(drain), wake } + captureWait, capacity, admitPool, slots, revalidate, recover, drain: () => serial(drain), wake } } export function install(rt) { rt.projectScheduler = createProjectScheduler(rt) diff --git a/patches/ds-bot/files/src/host/workbench-state.js b/patches/ds-bot/files/src/host/workbench-state.js index c78fac471..55d21d050 100644 --- a/patches/ds-bot/files/src/host/workbench-state.js +++ b/patches/ds-bot/files/src/host/workbench-state.js @@ -39,7 +39,7 @@ export function install(rt) { const repository = await repositoryReadiness(rt, project) const activity = rt.activityView?.() ?? {} if (body.view === 'summary') return { - projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), repository, + projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), capacity: rt.projectScheduler.capacity?.(id) ?? null, repository, queue: Object.values(rt.state.projectQueue ?? {}).filter(row => row.projectId === id && ['waiting', 'blocked'].includes(row.status)).map(({ id, taskId, status, reason }) => ({ id, taskId, status, reason })), taskExecution: Object.values(rt.state.tasks).filter(task => task.projectId === id).map(task => taskExecutionStatus(rt.state, task, { activity })), metrics: deliveryMetrics(rt.state, id), @@ -51,7 +51,7 @@ export function install(rt) { const host = { scope: 'shared-host-process', rssBytes: process.memoryUsage().rss, cpuPercent: elapsedMicros >= 100_000 ? percent : null, cpuSampleMs: elapsedMicros / 1000, cpuLimit: null, memoryLimitBytes: null, gpu: null, reason: 'Per-Project CPU/memory limits and GPU pools are not reported by the current platform API' } cpu = nowCpu; measured = now - return { projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), + return { projectId: id, quota: project.quota ?? { maxConcurrent: 2 }, activeSlots: rt.projectScheduler.slots(id), capacity: rt.projectScheduler.capacity?.(id) ?? null, queue: Object.values(rt.state.projectQueue ?? {}).filter(entry => entry.projectId === id).map(entry => structuredClone(entry)), workers, workspaces: Object.values(rt.state.projectWorkspaces ?? {}).filter(workspace => workspace.projectId === id), harnessWorkers: Object.values(rt.state.projectHarnessWorkers ?? {}).filter(row => row.projectId === id && (row.visibility !== 'private' || row.memberId === actor.memberId) && (!row.roomId || (() => { const room = ownEntry(rt.state.projectRooms ?? {}, row.roomId); return room?.projectId === id && (!room.memberIds.length || room.memberIds.includes(actor.memberId)) })())).map(({ eventLog: _log, sealedLog: _sealedLog, cwd: _cwd, ...row }) => structuredClone(row)), diff --git a/patches/ds-bot/files/test/project-harness.test.js b/patches/ds-bot/files/test/project-harness.test.js index 28956eec2..8289af9b5 100644 --- a/patches/ds-bot/files/test/project-harness.test.js +++ b/patches/ds-bot/files/test/project-harness.test.js @@ -3,6 +3,7 @@ import { test } from 'node:test' import { mkdtemp, mkdir, writeFile, readFile, rm, stat } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { createProjectScheduler } from '../src/host/project-scheduler.js' import { createProjectEffects } from '../src/host/project-effects.js' import { install as installProjectInputs } from '../src/host/project-inputs.js' import { createProjectResources } from '../src/host/project-resources.js' @@ -526,3 +527,29 @@ if(m.type==='start')write({type:'completed',sessionId:'input-actor-thread',text: await assert.rejects(harness.input({ workerId: id, generation: rt.state.projectHarnessWorkers[id].generation, text: 'Late unauthorized input' }, { memberId: 'sender' }), /Sending member was removed/) assert.equal(rt.state.projectHarnessWorkers[id].status, 'waiting-input') }) + + +test('an actual SDK protocol subprocess must exit before its Chief input wait frees project capacity', async t => { + const { rt, harness, cwd } = await fixture(t) + rt.state.bots = {}; rt.state.projectAttempts = { a: { id: 'a', taskId: 't', projectId: 'p', status: 'running' } } + rt.state.projectExecutions = { run: { id: 'run', attemptId: 'a', taskId: 't', projectId: 'p', status: 'running' } } + rt.state.tasks.t.deliveryAttemptId = 'a' + const scheduler = createProjectScheduler(rt) + const id = await harness.spawn({ executionProjectId: 'p', executionWorkDir: cwd, prompt: 'Test execution exit and waiting capacity' }) + const result = new Promise(resolve => harness.onResult(resolve)) + await harness.dispatch({ botId: id, taskId: 't', brief: 'Capacity checkpoint', outputs: [{ name: 'implementation' }] }) + assert.equal((await result).ok, true) + rt.state.tasks.t.chiefQuestion = { decisionId: 'chief-input', reason: 'Chief requests a scoped project input' } + rt.state.tasks.t.collaboration = { status: 'blocked', claim: { workerId: id, role: 'developer', status: 'completed' }, history: [] } + await scheduler.captureWait(rt.state.tasks.t) + assert.equal(scheduler.slots('p'), 1); assert.equal(rt.projectResources.occupied('r'), 1) + assert.equal(scheduler.capacity('p').exitUnknown, 1) + await harness.release(id) + // The minimal Harness fixture omits role on its pre-existing lease; real reservations carry it. + rt.state.projectResourceLeases.l.role = 'developer' + assert.equal(rt.state.projectHarnessWorkers[id].releaseProof.kind, 'sdk-stop-ack-and-process-exit') + assert.equal(scheduler.slots('p'), 0); assert.equal(rt.projectResources.occupied('r'), 0) + assert.equal(scheduler.capacity('p').waitingInput, 1) + assert.equal(rt.state.projectAttempts.a.status, 'running'); assert.equal(rt.state.projectExecutions.run.status, 'running') + assert.equal(rt.state.projectHarnessWorkers[id].backendSessionId, 'bound-thread') +}) diff --git a/patches/ds-bot/files/test/project-pools.test.js b/patches/ds-bot/files/test/project-pools.test.js index 6c7d19d9b..9ca157a82 100644 --- a/patches/ds-bot/files/test/project-pools.test.js +++ b/patches/ds-bot/files/test/project-pools.test.js @@ -619,3 +619,17 @@ test('ambiguous developer capabilities require an explicit initial role rather t await f.pool.admit({ id: 'attempt', taskId: 'task' }, { memberId: 'owner' }, { targetRole: 'builder' }) assert.equal(f.task.collaboration.botId, 'p:builder') }) + +test('resumed pool claims cannot bypass project admission or spend a claim generation while queued', async () => { + const f = setup(); await f.admit() + let admitted = false, checks = 0 + f.rt.projectScheduler = { async admitPool(task, definition) { ++checks; assert.equal(task.id, 'task'); assert.equal(definition.role, 'developer'); return admitted } } + const generation = f.task.collaboration.generation + assert.equal(await f.claim(), null) + assert.equal(f.task.collaboration.generation, generation) + assert.equal(f.task.collaboration.history[0].status, 'pending') + admitted = true + const claim = await f.claim() + assert.equal(claim.status, 'active'); assert.equal(checks, 2) + await assert.rejects(f.pool.decide('task', { ...claim, generation: generation - 1 }, { action: 'complete', reason: 'late callback' }), /stale/) +}) diff --git a/patches/ds-bot/files/test/project-scheduler.test.js b/patches/ds-bot/files/test/project-scheduler.test.js index 67de95476..bcff61232 100644 --- a/patches/ds-bot/files/test/project-scheduler.test.js +++ b/patches/ds-bot/files/test/project-scheduler.test.js @@ -239,3 +239,110 @@ test('capacity queue retains the selected repository across restart and refuses assert.match(f.state.projectQueue[queued.queue.id].reason, /not currently associated/) assert.equal(Object.keys(f.state.projectAttempts).length, 0) }) + +test('confirmed input waits release both Run and Attempt slots; resumed work waits behind admitted work', async () => { + const { rt, state, scheduler } = fixture() + rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1'] + task.deliveryAttemptId = attempt.id + task.collaboration = { status: 'blocked', generation: 4, initiatedBy: 'bob', attemptId: attempt.id, history: [] } + task.chiefQuestion = { decisionId: 'chief-1', reason: 'Need project input' } + await scheduler.captureWait(task) + assert.equal(scheduler.slots('p1'), 0) + assert.equal(attempt.status, 'running'); assert.equal(state.projectExecutions[attempt.id].status, 'running') + await scheduler.request({ taskId: 'two', requestId: 'two' }, { memberId: 'bob' }) + assert.equal(scheduler.slots('p1'), 1) + delete task.chiefQuestion; task.collaboration.status = 'queued'; task.collaboration.currentHandoffId = 'resume-one'; task.collaboration.botId = 'p1:developer'; ++task.collaboration.generation + const definition = { id: 'p1:developer', role: 'developer' } + assert.equal(await scheduler.admitPool(task, definition), false) + assert.equal(Object.values(state.projectQueue).filter(row => row.kind === 'resume').length, 1) + assert.equal(await scheduler.admitPool(task, definition), false) + assert.equal(Object.values(state.projectQueue).length, 1) + state.projectAttempts['attempt-2'].status = 'waiting-acceptance'; state.projectExecutions['attempt-2'].status = 'done' + await scheduler.drain() + assert.equal(await scheduler.admitPool(task, definition), true) + assert.equal(scheduler.slots('p1'), 1) + assert.equal(Object.keys(state.projectAttempts).length, 2) + assert.equal(attempt.capacityWait.status, 'admitted') +}) + +test('input wait never releases an unknown SDK, native Bot or unreleased lease', async () => { + for (const kind of ['sdk', 'bot', 'lease', 'unknown-claim']) { + const { rt, state, scheduler } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one; task.deliveryAttemptId = 'attempt-1'; task.chiefQuestion = { decisionId: 'chief', reason: 'input' }; task.collaboration = { status: 'blocked', history: [] } + if (kind === 'sdk') state.projectHarnessWorkers = { w: { id: 'w', attemptId: 'attempt-1', status: 'released', generation: 1 } } + if (kind === 'bot') state.bots.w = { id: 'w', ephemeral: true, attemptId: 'attempt-1', projectId: 'p1' } + if (kind === 'lease') state.projectResourceLeases = { l: { id: 'l', attemptId: 'attempt-1', projectId: 'p1', status: 'unreachable' } } + if (kind === 'unknown-claim') task.collaboration.claim = { workerId: 'lost', role: 'developer', status: 'active' } + await scheduler.captureWait(task) + assert.equal(scheduler.slots('p1'), 1, kind) + assert.equal(scheduler.capacity('p1').exitUnknown, 1, kind) + } +}) + +test('simultaneous continuations and new requests share FIFO quota across restart and duplicate calls', async () => { + const f = fixture(), { rt, state, project, scheduler } = f; rt.projectScheduler = scheduler + project.quota.maxConcurrent = 2 + for (const id of ['one', 'two']) { + await scheduler.request({ taskId: id, requestId: id }, { memberId: 'bob' }) + const task = state.tasks[id]; task.deliveryAttemptId = id === 'one' ? 'attempt-1' : 'attempt-2' + task.chiefQuestion = { decisionId: `chief-${id}` }; task.collaboration = { status: 'blocked', generation: 1, history: [], initiatedBy: 'bob', botId: 'p1:developer', currentHandoffId: id } + await scheduler.captureWait(task) + } + assert.equal(scheduler.capacity('p1').waitingInput, 2) + project.quota.maxConcurrent = 1 + for (const task of Object.values(state.tasks)) { delete task.chiefQuestion; task.collaboration.status = 'queued' } + const restarted = createProjectScheduler(rt, { preflight: async () => repositorySettings(rt, project) }); rt.projectScheduler = restarted + const definition = { id: 'p1:developer', role: 'developer' } + const result = await Promise.all([restarted.admitPool(state.tasks.one, definition), restarted.admitPool(state.tasks.two, definition), restarted.admitPool(state.tasks.one, definition)]) + await restarted.drain() + assert.ok(result.every(value => typeof value === 'boolean')); assert.equal(restarted.slots('p1'), 1) + assert.equal(await restarted.admitPool(state.tasks.one, definition), true) + assert.equal(await restarted.admitPool(state.tasks.two, definition), false) + assert.equal(Object.values(state.projectQueue).length, 2) + assert.equal(restarted.capacity('p1').waitingResume, 1) + state.tasks.three = { id: 'three', projectId: 'p1', status: 'open', requesterId: 'bob' } + const { queue } = await restarted.request({ taskId: 'three', requestId: 'three' }, { memberId: 'bob' }) + state.projectAttempts['attempt-1'].status = 'accepted'; state.projectExecutions['attempt-1'].status = 'done' + await restarted.drain() + assert.equal(state.projectAttempts['attempt-2'].capacityWait.status, 'admitted') + assert.equal(state.projectQueue[queue.id].status, 'waiting'); assert.equal(Object.keys(state.projectAttempts).length, 2) +}) + +test('uncommitted re-admission, revoked principals and changed handoffs never start execution', async () => { + for (const kind of ['save', 'revoked', 'stale']) { + const { rt, state, scheduler, revoke } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1']; task.deliveryAttemptId = attempt.id + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', history: [], generation: 2, currentHandoffId: 'h', initiatedBy: 'bob', botId: 'p1:developer' } + await scheduler.captureWait(task); delete task.chiefQuestion; task.collaboration.status = 'queued' + // Hold a real project slot so queue creation and admission can be inspected separately. + state.projectAttempts.busy = { id: 'busy', projectId: 'p1', taskId: 'two', status: 'running' } + assert.equal(await scheduler.admitPool(task, { id: 'p1:developer', role: 'developer' }), false) + state.projectAttempts.busy.status = 'accepted' + if (kind === 'save') { let once = true; rt.save = async () => { if (once) { once = false; throw new Error('disk failure') } } } + if (kind === 'revoked') revoke() + if (kind === 'stale') ++task.collaboration.generation + await scheduler.drain() + const entry = Object.values(state.projectQueue)[0] + assert.ok(['blocked', 'cancelled'].includes(entry.status), kind) + assert.equal(attempt.capacityWait.status, 'suspended', kind) + assert.equal(scheduler.slots('p1'), 0, kind) + assert.equal(attempt.id, task.deliveryAttemptId) + } +}) + +test('a matching SDK exit proof releases an input wait while legacy active workflows remain held', async () => { + const { rt, state, scheduler } = fixture(); rt.projectScheduler = scheduler + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one, attempt = state.projectAttempts['attempt-1']; task.deliveryAttemptId = attempt.id + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', history: [], claim: { workerId: 'w', role: 'developer', status: 'completed' } } + state.projectResourceLeases = { l: { id: 'l', projectId: 'p1', attemptId: attempt.id, role: 'developer', resourceId: 'docker', generation: 2, status: 'released', worker: { kind: 'sdk', handle: 'w' } } } + state.projectHarnessWorkers = { w: { id: 'w', projectId: 'p1', attemptId: attempt.id, role: 'developer', resourceId: 'docker', leaseId: 'l', generation: 2, status: 'released', releaseProof: { kind: 'sdk-stop-ack-and-process-exit', generation: 2 } } } + await scheduler.captureWait(task); assert.equal(scheduler.slots('p1'), 0) + state.projectResourceLeases.l.generation = 1; assert.equal(scheduler.slots('p1'), 1) + state.projectResourceLeases.l.generation = 2; attempt.workflowDefinition = { nodes: [] }; assert.equal(scheduler.slots('p1'), 1) + state.projectExecutions[attempt.id].status = 'done'; assert.equal(scheduler.slots('p1'), 0) +}) diff --git a/scripts/api-contracts.mjs b/scripts/api-contracts.mjs index 3cb82e315..b29a94903 100644 --- a/scripts/api-contracts.mjs +++ b/scripts/api-contracts.mjs @@ -38,8 +38,10 @@ export const schemas = { ChiefCorrection: record({ version: lit('chief-correction/v1'), round: int(1, 5), limit: int(0, 4), status: choice('pending', 'resolved', 'exhausted', 'commit-failed'), feedback: array(record({ code: lit('bot/chief-protocol'), taskId: nullable(id), fields: array(record({ field: str(), expected: str() }, ['field', 'expected'])), message: str(500), at: int(), sideEffects: str(), appliedTaskIds: array(id) }, ['code', 'fields', 'message', 'at']), 5) }, ['version', 'round', 'limit', 'status', 'feedback']), TaskRecord: record({ id, projectId: nullable(id), title: str(), criteria: str(), requesterId: nullable(id), projectTaskKind: choice('code', 'decision', 'release'), status: str(), priority: str(), progress: int(0, 100), note: str(), deliveryAttemptId: id, projectTask: json, collaboration: json, result: json, createdAt: int(), updatedAt: int() }, ['id', 'title', 'status']), TaskView: record({ version: lit('project-development/v1'), projectId: id, taskId: id, title: str(1000), publication: lit('published'), phase: choice('published', 'accepted', 'cancelled', 'stopping', 'blocked', 'awaiting-owner', 'awaiting-chief', 'queued', 'starting', 'running', 'awaiting-input', 'awaiting-acceptance', 'failed', 'paused', 'publishing'), waitingFor: nullable(str()), reason: nullable(str()), botId: nullable(id), currentRole: nullable(role), generation: nullable(int()), attemptId: nullable(id), queueId: nullable(id), workerId: nullable(id), candidateSha: nullable(sha), nextActions: array(str()), evidence: json, contract: nullable(json), source: json, inputs: array(json), effects: array(ref('Effect')), execution: nullable(json), outcome: nullable(json) }, ['projectId', 'taskId', 'phase', 'nextActions']), - Attempt: record({ id, projectId: id, taskId: id, status: choice('preparing', 'running', 'waiting-acceptance', 'accepted', 'failed', 'interrupted', 'returned', 'cancelled'), protocol: str(), workflowVersion: str(), baseSha: sha, candidateSha: sha, requirement: json, workspaces: map(json), evidence: json, pr: nullable(ref('GithubPublication')), release: json, roleSessions: map(json), createdAt: int(), updatedAt: int(), failure: json, sourceRefs: map(json), preview: bool }, ['id', 'projectId', 'taskId', 'status']), - QueueEntry: record({ id, projectId: id, taskId: id, requestId, status: choice('waiting', 'starting', 'started', 'blocked', 'failed', 'cancelled'), memberId: id, reason: str(), initialTarget: json, repositoryBinding: json, attemptId: nullable(id) }, ['id', 'projectId', 'taskId', 'status']), + CapacityState: result({ activeSlots: int(), waitingInput: int(), waitingResume: int(), exitUnknown: int() }), + CapacityWait: record({ version: lit('project-capacity-wait/v1'), source: record({ kind: choice('chief-input', 'children'), id }, ['kind', 'id']), status: choice('suspended', 'admitted'), attemptId: id, previousAttemptId: id, at: int(), queueId: id, admittedAt: int() }, ['version', 'source', 'status', 'attemptId', 'at']), + Attempt: record({ id, projectId: id, taskId: id, status: choice('preparing', 'running', 'waiting-acceptance', 'accepted', 'failed', 'interrupted', 'returned', 'cancelled'), protocol: str(), workflowVersion: str(), baseSha: sha, candidateSha: sha, requirement: json, capacityWait: ref('CapacityWait'), workspaces: map(json), evidence: json, pr: nullable(ref('GithubPublication')), release: json, roleSessions: map(json), createdAt: int(), updatedAt: int(), failure: json, sourceRefs: map(json), preview: bool }, ['id', 'projectId', 'taskId', 'status']), + QueueEntry: record({ id, kind: lit('resume'), handoffId: id, generation: int(), targetBotId: id, projectId: id, taskId: id, requestId, status: choice('waiting', 'starting', 'started', 'blocked', 'failed', 'cancelled'), memberId: id, reason: str(), initialTarget: json, repositoryBinding: json, attemptId: nullable(id) }, ['id', 'projectId', 'taskId', 'status']), Effect: record({ id, kind: str(), projectId: id, taskId: nullable(id), status: choice('pending', 'running', 'done', 'failed', 'cancelled'), failures: int(), dueAt: int(), error: nullable(str()), waitingFor: nullable(str()), receipt: json }, ['id', 'kind', 'status']), CommandReceipt: record({ id, name: str(), requestId, status: str(), result: json, resultRef: json, committed: bool, replayed: bool, error: json }, [],), Proposal: record({ id, projectId: id, memberId: id, fields: record({ title: str(), criteria: str(), impact: str(), scope: str(), projectTaskKind: choice('code', 'decision', 'release') }), createdAt: int(), publishedTaskId: id, personalAgentSource: json }, ['id', 'projectId']), @@ -316,7 +318,7 @@ native.get('project-role').permission = 'manage' native.get('project-role').notes = 'The shared Project access layer requires Owner manage permission. The underlying legacy role policy also supports self-demotion, but that does not bypass this public shared gateway gate. Owner cannot be assigned, removed or demoted.' native.get('project-board').response = result({ tasks: array(json), done: array(json), decisions: result({ dir: nullable(str()), files: array(json), reason: str() }, ['files']), meetings: result({ decided: array(json), pending: array(json), reason: str() }, ['decided', 'pending']), waiting: array(json), inflight: array(json), blocked: array(json), counts: map(nullable(int())), metrics: nullable(json) }) native.get('workbench-state').response = result({ demoMode: nullable(str()), self: result({ memberId: id, displayName: str() }), projects: array(ref('Project')), models: json, repositories: array(repository) }) -native.get('project-resources').response = result({ projectId: id, quota: result({ maxConcurrent: int(1, 32) }), activeSlots: int(), repository: json, queue: array(record({ id, taskId: id, status: str(), reason: str() }, ['id', 'taskId', 'status'])), taskExecution: array(json), metrics: json, workers: array(json), workspaces: array(json), harnessWorkers: array(json), executionResources: array(ref('Resource')), resourceLeases: array(json), chief: queries.get('chief.state').response, taskViews: array(ref('TaskView')), host: json, platformInstance: nullable(json) }, ['projectId', 'quota', 'activeSlots', 'repository', 'queue', 'taskExecution', 'metrics']) +native.get('project-resources').response = result({ projectId: id, quota: result({ maxConcurrent: int(1, 32) }), activeSlots: int(), capacity: nullable(ref('CapacityState')), repository: json, queue: array(record({ id, taskId: id, status: str(), reason: str() }, ['id', 'taskId', 'status'])), taskExecution: array(json), metrics: json, workers: array(json), workspaces: array(json), harnessWorkers: array(json), executionResources: array(ref('Resource')), resourceLeases: array(json), chief: queries.get('chief.state').response, taskViews: array(ref('TaskView')), host: json, platformInstance: nullable(json) }, ['projectId', 'quota', 'activeSlots', 'repository', 'queue', 'taskExecution', 'metrics']) queries.get('compatibility').response = result({ version: str(), cutoverId: nullable(id), committed: bool, baseline: nullable(result({ count: int(), sha256: hash })), totals: result({ contract: int(), legacy: int(), held: int(), active: int() }), tasks: array(record({ id, projectId: id, parentId: nullable(id), status: str(), mode: choice('contract', 'legacy', 'held'), attemptId: nullable(id), parentPresent: bool }, ['id', 'projectId', 'status', 'mode'])), page: result({ offset: int(), limit: int(1, 100), total: int(), nextOffset: nullable(int()) }) }) const sourceRecord = record({ id, version: lit('project-source/v1'), projectId: id, repository, kind: str() }, ['id', 'projectId']) const subscription = record({ id, version: lit('project-source/v1'), projectId: id, taskId: id, sourceId: id, threadId: id, createdBy: id, enabled: bool, mode: choice('active', 'context'), includeHistory: bool, createdAt: int() }, ['id', 'projectId', 'taskId', 'threadId', 'enabled']) From d04bae629efa05c274d9cb6d0138f24d3aa13934 Mon Sep 17 00:00:00 2001 From: zy95-12 <135135297+zy95-12@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:51:46 +0800 Subject: [PATCH 3/4] fix: fence continuation queue creation until durable save completes --- core/src/host/project-scheduler.js | 7 ++++--- core/test/project-scheduler.test.js | 18 ++++++++++++++++++ docs/openapi/flybot.openapi.json | 14 +++++++------- .../ds-bot/files/src/host/project-scheduler.js | 7 ++++--- .../files/test/project-scheduler.test.js | 18 ++++++++++++++++++ 5 files changed, 51 insertions(+), 13 deletions(-) diff --git a/core/src/host/project-scheduler.js b/core/src/host/project-scheduler.js index 2ec00f651..34b18cff8 100644 --- a/core/src/host/project-scheduler.js +++ b/core/src/host/project-scheduler.js @@ -18,7 +18,7 @@ export function releasedParentWait(state, attemptId) { export function createProjectScheduler(rt, { preflight = preflightRepository } = {}) { const { state } = rt const entries = () => (state.projectQueue ??= {}) - const savingAdmissions = new Set() + const savingAdmissions = new Set(), savingQueueEntries = new Set() let recovered = false let tail = Promise.resolve() let waking = null, wakeAgain = false @@ -66,8 +66,8 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) if (!entry) { entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } - setEntry(entries(), entry.id, entry) - try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } + setEntry(entries(), entry.id, entry); savingQueueEntries.add(entry.id) + try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } finally { savingQueueEntries.delete(entry.id) } } // Queue admission runs outside the pool lock: drain can itself admit a new pool Task. void wake() @@ -135,6 +135,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) for (const entry of pending) { + if (savingQueueEntries.has(entry.id)) continue const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue diff --git a/core/test/project-scheduler.test.js b/core/test/project-scheduler.test.js index bcff61232..a78038212 100644 --- a/core/test/project-scheduler.test.js +++ b/core/test/project-scheduler.test.js @@ -346,3 +346,21 @@ test('a matching SDK exit proof releases an input wait while legacy active workf state.projectResourceLeases.l.generation = 2; attempt.workflowDefinition = { nodes: [] }; assert.equal(scheduler.slots('p1'), 1) state.projectExecutions[attempt.id].status = 'done'; assert.equal(scheduler.slots('p1'), 0) }) + + +test('queue creation must commit before a concurrent drain can admit continuation', async () => { + const { rt, state, scheduler } = fixture() + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one; task.deliveryAttemptId = 'attempt-1' + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', generation: 1, history: [], initiatedBy: 'bob', currentHandoffId: 'h', botId: 'p1:developer' } + await scheduler.captureWait(task); delete task.chiefQuestion; task.collaboration.status = 'queued' + let rejectSave, count = 0 + rt.save = async () => { if (++count === 1) await new Promise((_, reject) => { rejectSave = reject }) } + const queued = scheduler.admitPool(task, { id: 'p1:developer', role: 'developer' }) + const rejected = assert.rejects(queued, /initial save failed/) + await new Promise(resolve => setImmediate(resolve)); await scheduler.drain() + assert.equal(state.projectAttempts['attempt-1'].capacityWait.status, 'suspended') + rejectSave(new Error('initial save failed')); await rejected + assert.equal(Object.keys(state.projectQueue).length, 0) + assert.equal(state.projectAttempts['attempt-1'].capacityWait.status, 'suspended') +}) diff --git a/docs/openapi/flybot.openapi.json b/docs/openapi/flybot.openapi.json index 6842fda7a..f56db04dc 100644 --- a/docs/openapi/flybot.openapi.json +++ b/docs/openapi/flybot.openapi.json @@ -31946,7 +31946,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" } }, { @@ -32918,7 +32918,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" } }, { @@ -33044,7 +33044,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" } }, { @@ -42812,7 +42812,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" }, "aliasOf": "development.command.task.run" }, @@ -43815,7 +43815,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" }, "aliasOf": "development.command.queue.cancel" }, @@ -43928,7 +43928,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326" + "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" }, "aliasOf": "development.command.queue.retry" }, @@ -45112,7 +45112,7 @@ "core/src/host/project-rooms.js": "148ba6f39086dc74e0c9c92ab368361b256b5185490d7983e957dd28890cfbe9", "core/src/host/project-runtime-setup.js": "7d4ff3f84c0b08540145b40c150262e8d85be957def7f14a06496064f20d51e0", "core/src/host/project-scheduled-tasks.js": "f99a7e28a95c4525672c88059878ede4b88c45b5b668dfc25f3a1dd519bab3c0", - "core/src/host/project-scheduler.js": "63d1e530980bf5d297d72f19f25bf9cc780c4bfb926f03fe8e205eeed0ab3326", + "core/src/host/project-scheduler.js": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c", "core/src/host/project-service-events.js": "71917d1ab367a9db8bef8c1d1003228255a57528e84a57e76de3a39d08ac28ec", "core/src/host/project-service-sse.js": "df97d6b11156da885a6a47d50a4a77c8c99b6af5eb0066de14312cc512762297", "core/src/host/project-service-views.js": "e34b36f78f4dcbb7e1f507203ba843959fa76a119fdecdc5b9902c39bbbef3fa", diff --git a/patches/ds-bot/files/src/host/project-scheduler.js b/patches/ds-bot/files/src/host/project-scheduler.js index 2ec00f651..34b18cff8 100644 --- a/patches/ds-bot/files/src/host/project-scheduler.js +++ b/patches/ds-bot/files/src/host/project-scheduler.js @@ -18,7 +18,7 @@ export function releasedParentWait(state, attemptId) { export function createProjectScheduler(rt, { preflight = preflightRepository } = {}) { const { state } = rt const entries = () => (state.projectQueue ??= {}) - const savingAdmissions = new Set() + const savingAdmissions = new Set(), savingQueueEntries = new Set() let recovered = false let tail = Promise.resolve() let waking = null, wakeAgain = false @@ -66,8 +66,8 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) if (!entry) { entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } - setEntry(entries(), entry.id, entry) - try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } + setEntry(entries(), entry.id, entry); savingQueueEntries.add(entry.id) + try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } finally { savingQueueEntries.delete(entry.id) } } // Queue admission runs outside the pool lock: drain can itself admit a new pool Task. void wake() @@ -135,6 +135,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) for (const entry of pending) { + if (savingQueueEntries.has(entry.id)) continue const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue diff --git a/patches/ds-bot/files/test/project-scheduler.test.js b/patches/ds-bot/files/test/project-scheduler.test.js index bcff61232..a78038212 100644 --- a/patches/ds-bot/files/test/project-scheduler.test.js +++ b/patches/ds-bot/files/test/project-scheduler.test.js @@ -346,3 +346,21 @@ test('a matching SDK exit proof releases an input wait while legacy active workf state.projectResourceLeases.l.generation = 2; attempt.workflowDefinition = { nodes: [] }; assert.equal(scheduler.slots('p1'), 1) state.projectExecutions[attempt.id].status = 'done'; assert.equal(scheduler.slots('p1'), 0) }) + + +test('queue creation must commit before a concurrent drain can admit continuation', async () => { + const { rt, state, scheduler } = fixture() + await scheduler.request({ taskId: 'one', requestId: 'one' }, { memberId: 'bob' }) + const task = state.tasks.one; task.deliveryAttemptId = 'attempt-1' + task.chiefQuestion = { decisionId: 'chief' }; task.collaboration = { status: 'blocked', generation: 1, history: [], initiatedBy: 'bob', currentHandoffId: 'h', botId: 'p1:developer' } + await scheduler.captureWait(task); delete task.chiefQuestion; task.collaboration.status = 'queued' + let rejectSave, count = 0 + rt.save = async () => { if (++count === 1) await new Promise((_, reject) => { rejectSave = reject }) } + const queued = scheduler.admitPool(task, { id: 'p1:developer', role: 'developer' }) + const rejected = assert.rejects(queued, /initial save failed/) + await new Promise(resolve => setImmediate(resolve)); await scheduler.drain() + assert.equal(state.projectAttempts['attempt-1'].capacityWait.status, 'suspended') + rejectSave(new Error('initial save failed')); await rejected + assert.equal(Object.keys(state.projectQueue).length, 0) + assert.equal(state.projectAttempts['attempt-1'].capacityWait.status, 'suspended') +}) From 0f8adc239677e5b43f3ee7058c184429d11f3f4d Mon Sep 17 00:00:00 2001 From: zy95-12 <135135297+zy95-12@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:54:25 +0800 Subject: [PATCH 4/4] fix: preserve FIFO order across simultaneous continuation saves --- core/src/host/project-scheduler.js | 11 +++++++---- docs/openapi/flybot.openapi.json | 14 +++++++------- patches/ds-bot/files/src/host/project-scheduler.js | 11 +++++++---- 3 files changed, 21 insertions(+), 15 deletions(-) diff --git a/core/src/host/project-scheduler.js b/core/src/host/project-scheduler.js index 34b18cff8..7d0ecff1d 100644 --- a/core/src/host/project-scheduler.js +++ b/core/src/host/project-scheduler.js @@ -34,6 +34,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = } return keys.size } + const queueOrder = () => Math.max(0, ...Object.values(entries()).map(row => row.order ?? 0)) + 1 const quota = project => project.quota?.maxConcurrent ?? 2 async function captureWait(task, source) { const attempt = state.projectAttempts?.[task.deliveryAttemptId] @@ -65,7 +66,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const c = task.collaboration let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) if (!entry) { - entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } + entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', order: queueOrder(), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry); savingQueueEntries.add(entry.id) try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } finally { savingQueueEntries.delete(entry.id) } } @@ -123,7 +124,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const resourceReason = resourceBlock(project.id, initialTarget) if (slots(project.id) < quota(project) && !resourceReason && !Object.values(entries()).some(row => row.projectId === project.id && row.status === 'waiting')) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) const entry = { id: `queue-${randomUUID()}`, projectId: project.id, taskId: task.id, requestId: body.requestId, requestKey, - memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } + memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', order: queueOrder(), ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry) await rt.save() return { queue: structuredClone(entry) } @@ -133,9 +134,11 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = let changed = false for (const task of Object.values(state.tasks)) if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') await captureWait(task) for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true - const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) + const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || (a.order ?? 0) - (b.order ?? 0) || a.id.localeCompare(b.id)) + const savingProjects = new Set() for (const entry of pending) { - if (savingQueueEntries.has(entry.id)) continue + if (savingQueueEntries.has(entry.id)) { savingProjects.add(entry.projectId); continue } + if (savingProjects.has(entry.projectId)) continue const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue diff --git a/docs/openapi/flybot.openapi.json b/docs/openapi/flybot.openapi.json index f56db04dc..1cf973217 100644 --- a/docs/openapi/flybot.openapi.json +++ b/docs/openapi/flybot.openapi.json @@ -31946,7 +31946,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" } }, { @@ -32918,7 +32918,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" } }, { @@ -33044,7 +33044,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" } }, { @@ -42812,7 +42812,7 @@ "notes": "Never bypasses Chief. Capacity busy means queue, not permission to create another Task. Cross-repository auto-split remains deferred.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" }, "aliasOf": "development.command.task.run" }, @@ -43815,7 +43815,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" }, "aliasOf": "development.command.queue.cancel" }, @@ -43928,7 +43928,7 @@ "notes": "All current authorization, Project/Task bindings and domain fences are rechecked server-side.", "implementation": { "path": "core/src/host/project-scheduler.js", - "sha256": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c" + "sha256": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6" }, "aliasOf": "development.command.queue.retry" }, @@ -45112,7 +45112,7 @@ "core/src/host/project-rooms.js": "148ba6f39086dc74e0c9c92ab368361b256b5185490d7983e957dd28890cfbe9", "core/src/host/project-runtime-setup.js": "7d4ff3f84c0b08540145b40c150262e8d85be957def7f14a06496064f20d51e0", "core/src/host/project-scheduled-tasks.js": "f99a7e28a95c4525672c88059878ede4b88c45b5b668dfc25f3a1dd519bab3c0", - "core/src/host/project-scheduler.js": "367f23d94a7171d54b32360d8b75153c87b0df6ac5c0dcabccea098182edb49c", + "core/src/host/project-scheduler.js": "9419010ed18fcfe801cab23562ea2ad53cb4dc500078ff1ec2af4438cc6726a6", "core/src/host/project-service-events.js": "71917d1ab367a9db8bef8c1d1003228255a57528e84a57e76de3a39d08ac28ec", "core/src/host/project-service-sse.js": "df97d6b11156da885a6a47d50a4a77c8c99b6af5eb0066de14312cc512762297", "core/src/host/project-service-views.js": "e34b36f78f4dcbb7e1f507203ba843959fa76a119fdecdc5b9902c39bbbef3fa", diff --git a/patches/ds-bot/files/src/host/project-scheduler.js b/patches/ds-bot/files/src/host/project-scheduler.js index 34b18cff8..7d0ecff1d 100644 --- a/patches/ds-bot/files/src/host/project-scheduler.js +++ b/patches/ds-bot/files/src/host/project-scheduler.js @@ -34,6 +34,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = } return keys.size } + const queueOrder = () => Math.max(0, ...Object.values(entries()).map(row => row.order ?? 0)) + 1 const quota = project => project.quota?.maxConcurrent ?? 2 async function captureWait(task, source) { const attempt = state.projectAttempts?.[task.deliveryAttemptId] @@ -65,7 +66,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const c = task.collaboration let entry = Object.values(entries()).find(row => row.kind === 'resume' && row.attemptId === attempt.id && row.handoffId === c.currentHandoffId && row.generation === c.generation) if (!entry) { - entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', enqueuedAt: Date.now(), updatedAt: Date.now() } + entry = { id: `queue-${randomUUID()}`, kind: 'resume', projectId: task.projectId, taskId: task.id, attemptId: attempt.id, handoffId: c.currentHandoffId, generation: c.generation, targetBotId: definition.id, initialTarget: { role: definition.role }, memberId: c.initiatedBy ?? attempt.initiatedBy, status: 'waiting', order: queueOrder(), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry); savingQueueEntries.add(entry.id) try { await rt.save() } catch (error) { delete entries()[entry.id]; throw error } finally { savingQueueEntries.delete(entry.id) } } @@ -123,7 +124,7 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = const resourceReason = resourceBlock(project.id, initialTarget) if (slots(project.id) < quota(project) && !resourceReason && !Object.values(entries()).some(row => row.projectId === project.id && row.status === 'waiting')) return rt.projectDelivery.endpoint('project-task-run', { ...body, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}) }, actor) const entry = { id: `queue-${randomUUID()}`, projectId: project.id, taskId: task.id, requestId: body.requestId, requestKey, - memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } + memberId: actor.memberId, repositoryBinding: execution.binding, ...(initialTarget ? { initialTarget } : {}), status: 'waiting', order: queueOrder(), ...(resourceReason ? { reason: resourceReason } : {}), enqueuedAt: Date.now(), updatedAt: Date.now() } setEntry(entries(), entry.id, entry) await rt.save() return { queue: structuredClone(entry) } @@ -133,9 +134,11 @@ export function createProjectScheduler(rt, { preflight = preflightRepository } = let changed = false for (const task of Object.values(state.tasks)) if (task.chiefQuestion || task.collaboration?.status === 'waiting-children') await captureWait(task) for (const project of rt.listProjects()) if (revalidate(project.id)) changed = true - const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || a.id.localeCompare(b.id)) + const pending = Object.values(entries()).filter(entry => entry.status === 'waiting').sort((a, b) => a.enqueuedAt - b.enqueuedAt || (a.order ?? 0) - (b.order ?? 0) || a.id.localeCompare(b.id)) + const savingProjects = new Set() for (const entry of pending) { - if (savingQueueEntries.has(entry.id)) continue + if (savingQueueEntries.has(entry.id)) { savingProjects.add(entry.projectId); continue } + if (savingProjects.has(entry.projectId)) continue const project = rt.readProject(entry.projectId) if (!project || project.status !== 'active') { entry.status = 'blocked'; entry.reason = 'Project is archived or unavailable'; changed = true; continue } if (slots(project.id) >= quota(project)) continue