fix: keep one writer per folder and show when an Agent waits for it

9198bf6 gave every direct chat its own scheduler key so that one writable
chat would not block another. That broke the one-writer-per-folder rule:
two Agents could edit the same files at once, and a direct chat and a
group could write the same folder concurrently. The underlying problem
was that a writer waiting for the folder lease showed nothing and looked
like a dead Agent (no queued event, no watchdog, no output).

- Scheduler requests report onQueued/onGranted when they actually wait.
- Direct chats use the real-folder key again; while an Agent waits for
  the folder, the run lists it in waitingForWorkspaceKinds and the run
  status card says it is waiting for another Agent writing in the same
  folder (zh/en). A stopped wait clears the mark.

Checks: run-scheduler (new queue notice test), write-scheduling (the
concurrent-direct-chat test now asserts the second writer waits, is shown
as waiting, then runs), run-coordinator public run keys. Full desktop
suite: 1537 of 1538 before adding the new public key to that list, which
now passes; npm --prefix frontend test (330).
This commit is contained in:
Ryder Sun committed 2026-10-15 19:17:03 +08:00
1 parent 8d8f9e65f6
commit 17cbd32387
10 files changed
+88 -15

No files matched your search

+12
View File
@@ -89,6 +89,9 @@ class RunScheduler {
this.changeCount(this.activeWorkspaces, entry.workspaceKey, 1) this.changeCount(this.activeWorkspaces, entry.workspaceKey, 1)
if (entry.permissionMode === 'workspace-write') this.activeWorkspaceWriters.add(entry.workspaceKey) if (entry.permissionMode === 'workspace-write') this.activeWorkspaceWriters.add(entry.workspaceKey)
entry.signal?.removeEventListener('abort', entry.abortHandler) entry.signal?.removeEventListener('abort', entry.abortHandler)
if (entry.waited) {
try { entry.onGranted?.() } catch { /* queue notices are best effort */ }
}
let released = false let released = false
const lease = Object.freeze({ const lease = Object.freeze({
leaseId: entry.requestId, leaseId: entry.requestId,
@@ -161,6 +164,8 @@ class RunScheduler {
cancelled: false, cancelled: false,
abortHandler: null, abortHandler: null,
pauseToken, pauseToken,
waited: false,
onGranted: typeof input.onGranted === 'function' ? input.onGranted : null,
} }
entry.abortHandler = () => { entry.abortHandler = () => {
if (entry.cancelled) return if (entry.cancelled) return
@@ -173,6 +178,13 @@ class RunScheduler {
signal?.addEventListener('abort', entry.abortHandler, { once: true }) signal?.addEventListener('abort', entry.abortHandler, { once: true })
this.queue.push(entry) this.queue.push(entry)
this.drain() this.drain()
// Tell the caller when it has to wait for another holder (for example
// the single writer of a folder), so the UI can say so instead of
// showing an Agent that looks stuck.
if (!entry.cancelled && this.queue.includes(entry)) {
entry.waited = true
try { input.onQueued?.() } catch { /* queue notices are best effort */ }
}
}) })
} }
@@ -625,13 +625,11 @@ class LocalWorkspaceAgentInvocation {
if (!taskId) throw new Error('LOCAL_RUN_TASK_INVALID') if (!taskId) throw new Error('LOCAL_RUN_TASK_INVALID')
let workspacePath = group.workdir ? path.resolve(group.workdir) : group.id let workspacePath = group.workdir ? path.resolve(group.workdir) : group.id
try { workspacePath = fs.realpathSync(workspacePath) } catch { /* runtime validates unavailable directories */ } try { workspacePath = fs.realpathSync(workspacePath) } catch { /* runtime validates unavailable directories */ }
// Direct chats are independent execution surfaces. Scope their scheduler // One writer per real folder, across group and direct chats, so two Agents
// resource to the chat so one writable chat cannot block another Agent. // never edit the same files at once. A waiting writer is reported to the
const schedulerWorkspaceKey = group.conversationType === 'direct' // UI (see onQueued below) instead of looking stuck.
? `${workspacePath}\u0000direct:${group.id}`
: workspacePath
const workspaceKey = createHash('sha256') const workspaceKey = createHash('sha256')
.update(String(schedulerWorkspaceKey)) .update(String(workspacePath))
.digest('hex') .digest('hex')
const writerKind = cleanText(context.singleWriterKind || context.writerKind, 80) const writerKind = cleanText(context.singleWriterKind || context.writerKind, 80)
const permissionMode = group.allowWrite === true const permissionMode = group.allowWrite === true
@@ -651,11 +649,24 @@ class LocalWorkspaceAgentInvocation {
else signal.addEventListener('abort', queuedAbortHandler, { once: true }) else signal.addEventListener('abort', queuedAbortHandler, { once: true })
} }
try { try {
const waiting = () => {
if (!activeRun) return null
if (!(activeRun.waitingForWorkspace instanceof Set)) activeRun.waitingForWorkspace = new Set()
return activeRun.waitingForWorkspace
}
return await this.runScheduler.withLease({ return await this.runScheduler.withLease({
taskId, taskId,
workspaceKey, workspaceKey,
permissionMode, permissionMode,
signal: agentController.signal, signal: agentController.signal,
onQueued: () => {
waiting()?.add(kind)
this.emitChanged?.()
},
onGranted: () => {
waiting()?.delete(kind)
this.emitChanged?.()
},
}, (lease) => { }, (lease) => {
if (signal && queuedAbortHandler) { if (signal && queuedAbortHandler) {
signal.removeEventListener('abort', queuedAbortHandler) signal.removeEventListener('abort', queuedAbortHandler)
@@ -672,6 +683,11 @@ class LocalWorkspaceAgentInvocation {
) )
}) })
} finally { } finally {
// A stopped or failed wait must not leave the Agent marked as queued.
if (activeRun?.waitingForWorkspace instanceof Set
&& activeRun.waitingForWorkspace.delete(kind)) {
this.emitChanged?.()
}
if (signal && queuedAbortHandler) { if (signal && queuedAbortHandler) {
signal.removeEventListener('abort', queuedAbortHandler) signal.removeEventListener('abort', queuedAbortHandler)
} }
@@ -389,6 +389,9 @@ function workspaceSnapshot({
responseVersionRootId: run.responseVersionRootId || '', responseVersionRootId: run.responseVersionRootId || '',
startedAt: run.startedAt || Date.now(), startedAt: run.startedAt || Date.now(),
agentRuns: run.harness?.snapshot?.() || [], agentRuns: run.harness?.snapshot?.() || [],
waitingForWorkspaceKinds: run.waitingForWorkspace instanceof Set
? [...run.waitingForWorkspace]
: [],
waitingGateIds: [...(run.waitingGateIds || [])], waitingGateIds: [...(run.waitingGateIds || [])],
budget: run.budget?.snapshot?.() || null, budget: run.budget?.snapshot?.() || null,
...(publicOrchestration ? { orchestration: publicOrchestration } : {}), ...(publicOrchestration ? { orchestration: publicOrchestration } : {}),
+18
View File
@@ -559,3 +559,21 @@ test('legacy suspension still dispatches older queued same-task work while waiti
assert.equal(dispatchedWhileWaiting, true) assert.equal(dispatchedWhileWaiting, true)
assert.equal(scheduler.snapshot().active.global, 0) assert.equal(scheduler.snapshot().active.global, 0)
}) })
test('queued requests report waiting and granting; immediate grants stay silent', async () => {
const scheduler = new RunScheduler()
const events = []
const first = await scheduler.acquire({
taskId: 'task-1', workspaceKey: 'folder', permissionMode: 'workspace-write',
onQueued: () => events.push('first-queued'), onGranted: () => events.push('first-granted'),
})
const second = scheduler.acquire({
taskId: 'task-2', workspaceKey: 'folder', permissionMode: 'workspace-write',
onQueued: () => events.push('second-queued'), onGranted: () => events.push('second-granted'),
})
assert.deepEqual(events, ['second-queued'])
first.release()
const lease = await second
assert.deepEqual(events, ['second-queued', 'second-granted'])
lease.release()
})
@@ -240,7 +240,7 @@ test('workspace snapshots keep a safe controllable run shell for an unknown V4 o
'agentRuns', 'budget', 'completedKinds', 'currentKind', 'currentRound', 'failedKinds', 'agentRuns', 'budget', 'completedKinds', 'currentKind', 'currentRound', 'failedKinds',
'groupId', 'maxRounds', 'mode', 'orchestration', 'phase', 'progress', 'groupId', 'maxRounds', 'mode', 'orchestration', 'phase', 'progress',
'responseVersionRootId', 'runId', 'startedAt', 'targetKinds', 'threadRootId', 'responseVersionRootId', 'runId', 'startedAt', 'targetKinds', 'threadRootId',
'unlimitedRounds', 'waitingGateIds', 'unlimitedRounds', 'waitingForWorkspaceKinds', 'waitingGateIds',
]) ])
}) })
@@ -82,10 +82,10 @@ test('read-only group invocations retain parallel execution in the same director
await Promise.all(runs) await Promise.all(runs)
}) })
test('writable direct chats do not block another Agent in the same directory', { timeout: 10000 }, async (t) => { test('writable chats in one folder take turns and show who is waiting', { timeout: 10000 }, async (t) => {
const { directory, options } = fixture() const { directory, options } = fixture()
const release = deferred() const release = deferred()
const bothStarted = deferred() const firstStarted = deferred()
t.after(() => { t.after(() => {
release.resolve() release.resolve()
fs.rmSync(directory, { recursive: true, force: true }) fs.rmSync(directory, { recursive: true, force: true })
@@ -94,8 +94,10 @@ test('writable direct chats do not block another Agent in the same directory', {
const calls = [] const calls = []
options.runAgent = async (agent, _prompt, _cwd, runOptions) => { options.runAgent = async (agent, _prompt, _cwd, runOptions) => {
calls.push({ kind: agent.kind, sandbox: runOptions.sandbox }) calls.push({ kind: agent.kind, sandbox: runOptions.sandbox })
if (calls.length === 2) bothStarted.resolve() if (agent.kind === 'codex') {
await release.promise firstStarted.resolve()
await release.promise
}
return { text: `${agent.kind} completed.` } return { text: `${agent.kind} completed.` }
} }
const workspace = new LocalWorkspace(options) const workspace = new LocalWorkspace(options)
@@ -104,11 +106,24 @@ test('writable direct chats do not block another Agent in the same directory', {
name: `${kind} direct`, agentKinds: [kind], directAgentKind: kind, name: `${kind} direct`, agentKinds: [kind], directAgentKind: kind,
conversationType: 'direct', workdir: directory, allowWrite: true, conversationType: 'direct', workdir: directory, allowWrite: true,
})) }))
const runs = groups.map(group => workspace.sendMessage({ groupId: group.id, text: 'Run independently.' })) const first = workspace.sendMessage({ groupId: groups[0].id, text: 'Write first.' })
await bothStarted.promise await firstStarted.promise
assert.deepEqual(calls.map(call => call.sandbox), ['workspace-write', 'workspace-write']) const second = workspace.sendMessage({ groupId: groups[1].id, text: 'Write second.' })
// The second writer waits for the folder and the snapshot says so.
for (let attempt = 0; attempt < 50; attempt += 1) {
const run = workspace.snapshot().runs.find(item => item.groupId === groups[1].id)
if (run?.waitingForWorkspaceKinds?.includes('hermes')) break
await new Promise(resolve => setTimeout(resolve, 10))
}
const waitingRun = workspace.snapshot().runs.find(item => item.groupId === groups[1].id)
assert.deepEqual(waitingRun.waitingForWorkspaceKinds, ['hermes'])
assert.deepEqual(calls.map(call => call.kind), ['codex'])
release.resolve() release.resolve()
await Promise.all(runs) await Promise.all([first, second])
assert.deepEqual(calls.map(call => [call.kind, call.sandbox]), [
['codex', 'workspace-write'], ['hermes', 'workspace-write'],
])
assert.equal(workspace.snapshot().runs.some(run => run.waitingForWorkspaceKinds?.length), false)
}) })
test('Manual V4 retains parallel readers alongside its single writer and stable message order', { timeout: 10000 }, async (t) => { test('Manual V4 retains parallel readers alongside its single writer and stable message order', { timeout: 10000 }, async (t) => {
@@ -726,6 +726,11 @@
<span v-if="runRoundProgress" class="run-round-progress"> <span v-if="runRoundProgress" class="run-round-progress">
{{ t(runRoundProgress.unlimited ? 'run.roundProgressUnlimited' : 'run.roundProgress', runRoundProgress) }} {{ t(runRoundProgress.unlimited ? 'run.roundProgressUnlimited' : 'run.roundProgress', runRoundProgress) }}
</span> </span>
<span v-if="activeRun?.waitingForWorkspaceKinds?.length" class="run-workspace-wait">
{{ t('run.waitingForWorkspace', {
agents: activeRun.waitingForWorkspaceKinds.map(kind => agentLabel(kind)).join(', '),
}) }}
</span>
</div> </div>
<div <div
v-if="isDisplayedCoordinatedRun && displayedRunPhaseLabel" v-if="isDisplayedCoordinatedRun && displayedRunPhaseLabel"
+2
View File
@@ -247,6 +247,8 @@ function normalizeRun(value) {
run.agentRuns = Array.isArray(rawAgentRuns) run.agentRuns = Array.isArray(rawAgentRuns)
? rawAgentRuns.slice(0, MAX_RUN_AGENTS).map(agent => normalizeRunAgent(agent, run)).filter(Boolean) ? rawAgentRuns.slice(0, MAX_RUN_AGENTS).map(agent => normalizeRunAgent(agent, run)).filter(Boolean)
: [] : []
// Agents waiting for another writer in the same folder.
run.waitingForWorkspaceKinds = normalizedAgentKinds(input.waitingForWorkspaceKinds)
run.waitingGateIds = [...new Set((Array.isArray(input.waitingGateIds) run.waitingGateIds = [...new Set((Array.isArray(input.waitingGateIds)
? input.waitingGateIds.slice(0, 32) ? input.waitingGateIds.slice(0, 32)
: []).filter(gateId => typeof gateId === 'string' && HUMAN_GATE_ID.test(gateId)))] : []).filter(gateId => typeof gateId === 'string' && HUMAN_GATE_ID.test(gateId)))]
+1
View File
@@ -201,6 +201,7 @@ export default {
'run.roundProgressUnlimited': 'Round {current} / No round limit', 'run.roundProgressUnlimited': 'Round {current} / No round limit',
'run.retainedEvents': '{count} retained events', 'run.retainedEvents': '{count} retained events',
'run.openDetails': 'Open run details', 'run.openDetails': 'Open run details',
'run.waitingForWorkspace': '{agents} is waiting for another Agent that is writing in the same folder.',
'run.status.pending': 'Pending', 'run.status.pending': 'Pending',
'run.status.queued': 'Queued', 'run.status.queued': 'Queued',
'run.status.preparing': 'Preparing', 'run.status.preparing': 'Preparing',
+1
View File
@@ -201,6 +201,7 @@ export default {
'run.roundProgressUnlimited': '第 {current} 轮 / 不限轮数', 'run.roundProgressUnlimited': '第 {current} 轮 / 不限轮数',
'run.retainedEvents': '保留 {count} 条过程事件', 'run.retainedEvents': '保留 {count} 条过程事件',
'run.openDetails': '打开运行详情', 'run.openDetails': '打开运行详情',
'run.waitingForWorkspace': '{agents} 正在等待同一文件夹中另一个正在写入的 Agent 完成。',
'run.status.pending': '待处理', 'run.status.pending': '待处理',
'run.status.queued': '排队中', 'run.status.queued': '排队中',
'run.status.preparing': '准备中', 'run.status.preparing': '准备中',