diff --git a/architecture.md b/architecture.md index dff3ad0..7b1376d 100644 --- a/architecture.md +++ b/architecture.md @@ -19,7 +19,7 @@ The supported product surface is the packaged Electron application. The frontend | Context Pack and delivery stores | Capture immutable, content-addressed Task inputs, attempt-specific approved prompts, exact outbound fingerprints, source hashes, target scope, and native Session provenance | `desktop/src/collaboration/context-pack-store.cjs`, `desktop/src/workspace/local-workspace-context-packs.cjs` | | Outcome store | Persists typed Artifacts, Evidence, Findings, and Adoption records separately from diagnostic reasoning and tool summaries | `desktop/src/collaboration/outcome-store.cjs`, `desktop/src/collaboration/outcome-records.cjs` | | Human Gate store | Persists permission, budget, and decision requests and their terminal user/system decisions without exposing raw request payloads to the renderer | `desktop/src/gates/human-gate-store.cjs`, `desktop/src/gates/human-gate-coordinator.cjs` | -| Scheduling and failure policy | Enforces run budgets, per-Agent controls, and one writer per folder; classifies failures (sign-in, rate limit, network, provider, context overflow) and decides retries with `Retry-After`; stops a silent Agent through the invocation watchdog | `desktop/src/runs/run-scheduler.cjs`, `desktop/src/runs/failure-policy.cjs`, `desktop/src/runs/failure-signals.cjs`, `desktop/src/runs/invocation-watchdog.cjs` | +| Scheduling and failure policy | Enforces run budgets, per-Agent controls, and one writer at a time within a conversation (conversations never wait for each other); classifies failures (sign-in, rate limit, network, provider, context overflow) and decides retries with `Retry-After`; stops a silent Agent through the invocation watchdog | `desktop/src/runs/run-scheduler.cjs`, `desktop/src/runs/failure-policy.cjs`, `desktop/src/runs/failure-signals.cjs`, `desktop/src/runs/invocation-watchdog.cjs` | | Agent Connector registry/runtime | Validates approved manifests and instances, separates CredentialRefs, records Connector/upstream versions, and reduces versioned idempotent Run Events | `desktop/src/agents/connectors/agent-connector-manifest.cjs`, `desktop/src/agents/connectors/agent-connector-registry.cjs`, `desktop/src/agents/connectors/agent-connector-runtime.cjs` | | Attachment store | Imports validated local images, media, documents, code/configuration files, Office files, and archives into private app-owned storage, verifies metadata/checksums on read, and removes unreferenced entries | `desktop/src/attachments/attachment-store.cjs`, `desktop/src/attachments/image-dimensions.cjs` | | Local Skill catalog | Scans bounded known roots per Agent and returns sanitized Skill coordinates without exposing paths or contents | `desktop/src/skills/local-skill-catalog.cjs` | @@ -68,6 +68,7 @@ Splitting by file did not change any method; the class still shares one `this`. ## Reliability Rules +- Conversations are independent. A running chat, direct or group, never makes an Agent in another conversation wait, even when both use the same folder and the same Agent. The writer lease is scoped to the conversation; inside one conversation writers still take turns. - A failure pauses an Agent for the current run only. Group membership is never rewritten because an Agent failed; a group whose members were all removed by an older build is rebuilt from the Agents that replied in it. - Only a definitive sign-in failure marks credentials missing. A refusal that could be a proxy or a tool's own output is retried once and reported as a provider failure. - An Agent that was available stays available through an inconclusive readiness probe for ten minutes. A definitive answer applies immediately. @@ -127,6 +128,7 @@ There is no Meldwork-hosted remote data store or application account namespace. ## Known Risks And Assumptions +- Writable conversations that share a folder are not coordinated. Two Agents in different conversations can edit the same files at the same time, as two terminals in one directory can. Output capture compares `.meldwork-output/` before and after a run, so when two conversations produce deliverables in the same folder at the same time, one may attach or record a file the other wrote. - Installer recipes use mutable upstream packages or scripts and currently do not verify a pinned digest or signature (`desktop/src/agents/installer/agent-installer.cjs`). - Conversation history is local but not application-encrypted (`desktop/src/workspace/local-workspace.cjs`). - Sanitized Run Ledger checkpoints are local and private but not application-encrypted. The store assumes one Electron main-process writer and recovers only the last persisted bounded checkpoint, not every in-memory event (`desktop/src/runs/run-ledger.cjs`). diff --git a/desktop/src/runs/run-scheduler.cjs b/desktop/src/runs/run-scheduler.cjs index e787893..e435f74 100644 --- a/desktop/src/runs/run-scheduler.cjs +++ b/desktop/src/runs/run-scheduler.cjs @@ -89,9 +89,6 @@ class RunScheduler { this.changeCount(this.activeWorkspaces, entry.workspaceKey, 1) if (entry.permissionMode === 'workspace-write') this.activeWorkspaceWriters.add(entry.workspaceKey) entry.signal?.removeEventListener('abort', entry.abortHandler) - if (entry.waited) { - try { entry.onGranted?.() } catch { /* queue notices are best effort */ } - } let released = false const lease = Object.freeze({ leaseId: entry.requestId, @@ -164,8 +161,6 @@ class RunScheduler { cancelled: false, abortHandler: null, pauseToken, - waited: false, - onGranted: typeof input.onGranted === 'function' ? input.onGranted : null, } entry.abortHandler = () => { if (entry.cancelled) return @@ -178,13 +173,6 @@ class RunScheduler { signal?.addEventListener('abort', entry.abortHandler, { once: true }) this.queue.push(entry) 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 */ } - } }) } diff --git a/desktop/src/workspace/local-workspace-agent-invocation.cjs b/desktop/src/workspace/local-workspace-agent-invocation.cjs index e641adc..2bb20a4 100644 --- a/desktop/src/workspace/local-workspace-agent-invocation.cjs +++ b/desktop/src/workspace/local-workspace-agent-invocation.cjs @@ -1,6 +1,4 @@ const { createHash, randomUUID } = require('node:crypto') -const fs = require('node:fs') -const path = require('node:path') const { assertResumedPermissionBinding, automaticPermissionResponse, @@ -241,13 +239,13 @@ class LocalWorkspaceAgentInvocation { const activeRun = this.activeRuns.get(group.id) const taskId = cleanText(activeRun?.taskId || context.taskId || threadRootId, 120) if (!taskId) throw new Error('LOCAL_RUN_TASK_INVALID') - let workspacePath = group.workdir ? path.resolve(group.workdir) : group.id - try { workspacePath = fs.realpathSync(workspacePath) } catch { /* runtime validates unavailable directories */ } - // One writer per real folder, across group and direct chats, so two Agents - // never edit the same files at once. A waiting writer is reported to the - // UI (see onQueued below) instead of looking stuck. + // Every conversation, group or direct, is its own execution surface. The + // writer lease is scoped to the conversation, so a writable chat never + // makes an Agent in another conversation wait, including when both use the + // same folder (most chats keep the default one). Writers inside one + // conversation still take turns. const workspaceKey = createHash('sha256') - .update(String(workspacePath)) + .update(`conversation\u0000${group.id}`) .digest('hex') const writerKind = cleanText(context.singleWriterKind || context.writerKind, 80) const permissionMode = group.allowWrite === true @@ -267,24 +265,11 @@ class LocalWorkspaceAgentInvocation { else signal.addEventListener('abort', queuedAbortHandler, { once: true }) } try { - const waiting = () => { - if (!activeRun) return null - if (!(activeRun.waitingForWorkspace instanceof Set)) activeRun.waitingForWorkspace = new Set() - return activeRun.waitingForWorkspace - } return await this.runScheduler.withLease({ taskId, workspaceKey, permissionMode, signal: agentController.signal, - onQueued: () => { - waiting()?.add(kind) - this.emitChanged?.() - }, - onGranted: () => { - waiting()?.delete(kind) - this.emitChanged?.() - }, }, (lease) => { if (signal && queuedAbortHandler) { signal.removeEventListener('abort', queuedAbortHandler) @@ -301,11 +286,6 @@ class LocalWorkspaceAgentInvocation { ) }) } 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) { signal.removeEventListener('abort', queuedAbortHandler) } diff --git a/desktop/src/workspace/local-workspace-state.cjs b/desktop/src/workspace/local-workspace-state.cjs index fd9180a..655a222 100644 --- a/desktop/src/workspace/local-workspace-state.cjs +++ b/desktop/src/workspace/local-workspace-state.cjs @@ -391,9 +391,6 @@ function workspaceSnapshot({ responseVersionRootId: run.responseVersionRootId || '', startedAt: run.startedAt || Date.now(), agentRuns: run.harness?.snapshot?.() || [], - waitingForWorkspaceKinds: run.waitingForWorkspace instanceof Set - ? [...run.waitingForWorkspace] - : [], waitingGateIds: [...(run.waitingGateIds || [])], budget: run.budget?.snapshot?.() || null, ...(publicOrchestration ? { orchestration: publicOrchestration } : {}), diff --git a/desktop/test/runs/run-scheduler.test.cjs b/desktop/test/runs/run-scheduler.test.cjs index f311ac5..3cd8abf 100644 --- a/desktop/test/runs/run-scheduler.test.cjs +++ b/desktop/test/runs/run-scheduler.test.cjs @@ -559,21 +559,3 @@ test('legacy suspension still dispatches older queued same-task work while waiti assert.equal(dispatchedWhileWaiting, true) 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() -}) diff --git a/desktop/test/workspace/local-workspace-run-coordinator.test.cjs b/desktop/test/workspace/local-workspace-run-coordinator.test.cjs index 63f15d6..41f82e2 100644 --- a/desktop/test/workspace/local-workspace-run-coordinator.test.cjs +++ b/desktop/test/workspace/local-workspace-run-coordinator.test.cjs @@ -240,7 +240,7 @@ test('workspace snapshots keep a safe controllable run shell for an unknown V4 o 'agentRuns', 'budget', 'completedKinds', 'currentKind', 'currentRound', 'failedKinds', 'groupId', 'maxRounds', 'mode', 'orchestration', 'phase', 'progress', 'responseVersionRootId', 'runId', 'startedAt', 'targetKinds', 'threadRootId', - 'unlimitedRounds', 'waitingForWorkspaceKinds', 'waitingGateIds', + 'unlimitedRounds', 'waitingGateIds', ]) }) diff --git a/desktop/test/workspace/local-workspace-write-scheduling.test.cjs b/desktop/test/workspace/local-workspace-write-scheduling.test.cjs index c5c5348..6192a4b 100644 --- a/desktop/test/workspace/local-workspace-write-scheduling.test.cjs +++ b/desktop/test/workspace/local-workspace-write-scheduling.test.cjs @@ -6,13 +6,13 @@ const { LocalWorkspace } = require('../../src/workspace/local-workspace.cjs') const { RunScheduler } = require('../../src/runs/run-scheduler.cjs') const { deferred, fixture } = require('../support/local-workspace-test-helpers.cjs') -test('workspace invocations serialize writes across groups and symlink aliases', { timeout: 10000 }, async (t) => { +test('group chats sharing a folder, even through a symlink alias, run side by side', { timeout: 10000 }, async (t) => { const { directory, options } = fixture() const workdir = path.join(directory, 'work') const alias = path.join(directory, 'alias') fs.mkdirSync(workdir) fs.symlinkSync(workdir, alias, process.platform === 'win32' ? 'junction' : 'dir') - const started = deferred() + const bothStarted = deferred() const release = deferred() t.after(() => { release.resolve() @@ -23,10 +23,8 @@ test('workspace invocations serialize writes across groups and symlink aliases', const calls = [] options.runAgent = async (agent, _prompt, cwd, runOptions) => { calls.push({ kind: agent.kind, sandbox: runOptions.sandbox }) - if (calls.length === 1) { - started.resolve() - await release.promise - } + if (calls.length === 2) bothStarted.resolve() + await release.promise fs.appendFileSync(path.join(cwd, 'result.txt'), `${agent.kind}\n`) return { text: `Written by ${agent.kind}.`, sessionRef: `${agent.kind}-session` } } @@ -39,21 +37,54 @@ test('workspace invocations serialize writes across groups and symlink aliases', name: 'second writer', agentKinds: ['hermes'], workdir: alias, allowWrite: true, }) const firstRun = workspace.sendMessage({ groupId: first.id, text: 'Write the requested result.' }) - await started.promise const secondRun = workspace.sendMessage({ groupId: second.id, text: 'Append the requested result.' }) - await new Promise(resolve => setImmediate(resolve)) - assert.equal(calls.length, 1) - assert.equal(scheduler.snapshot().queued.length, 1) - release.resolve() - await Promise.all([firstRun, secondRun]) + // The second group must not wait for the first: both Agents are running + // while neither has finished. + await bothStarted.promise assert.deepEqual(calls, [ { kind: 'codex', sandbox: 'workspace-write' }, { kind: 'hermes', sandbox: 'workspace-write' }, ]) - assert.equal(fs.readFileSync(path.join(workdir, 'result.txt'), 'utf8'), 'codex\nhermes\n') + assert.equal(scheduler.snapshot().queued.length, 0) + assert.equal(scheduler.snapshot().active.workspaces.length, 2) + release.resolve() + await Promise.all([firstRun, secondRun]) + assert.deepEqual( + fs.readFileSync(path.join(workdir, 'result.txt'), 'utf8').trim().split('\n').sort(), + ['codex', 'hermes'], + ) assert.equal(scheduler.snapshot().active.global, 0) }) +test('the writer lease belongs to the conversation, not to the folder', { timeout: 10000 }, async (t) => { + const { directory, options } = fixture() + t.after(() => fs.rmSync(directory, { recursive: true, force: true })) + const scheduler = new RunScheduler() + const leases = [] + options.runScheduler = { + withLease: (input, operation) => { + leases.push({ key: input.workspaceKey, permissionMode: input.permissionMode }) + return scheduler.withLease(input, operation) + }, + snapshot: () => scheduler.snapshot(), + } + options.runAgent = async agent => ({ text: `${agent.kind} done.` }) + const workspace = new LocalWorkspace(options) + await workspace.refreshAgents() + const [one, two] = ['one', 'two'].map(name => workspace.createGroup({ + name, agentKinds: ['codex'], workdir: directory, allowWrite: true, + })) + + await workspace.sendMessage({ groupId: one.id, text: 'First turn.' }) + await workspace.sendMessage({ groupId: one.id, text: 'Second turn.' }) + await workspace.sendMessage({ groupId: two.id, text: 'Other conversation.' }) + + assert.equal(leases.length, 3) + assert.equal(leases.every(lease => lease.permissionMode === 'workspace-write'), true) + assert.equal(leases[0].key, leases[1].key, 'turns of one conversation share a lease scope') + assert.notEqual(leases[0].key, leases[2].key, 'another conversation in the same folder does not') +}) + test('read-only group invocations retain parallel execution in the same directory', { timeout: 10000 }, async (t) => { const { directory, options } = fixture() const release = deferred() @@ -82,10 +113,10 @@ test('read-only group invocations retain parallel execution in the same director await Promise.all(runs) }) -test('writable chats in one folder take turns and show who is waiting', { timeout: 10000 }, async (t) => { +test('writable direct chats do not block another Agent in the same directory', { timeout: 10000 }, async (t) => { const { directory, options } = fixture() const release = deferred() - const firstStarted = deferred() + const bothStarted = deferred() t.after(() => { release.resolve() fs.rmSync(directory, { recursive: true, force: true }) @@ -94,10 +125,8 @@ test('writable chats in one folder take turns and show who is waiting', { timeou const calls = [] options.runAgent = async (agent, _prompt, _cwd, runOptions) => { calls.push({ kind: agent.kind, sandbox: runOptions.sandbox }) - if (agent.kind === 'codex') { - firstStarted.resolve() - await release.promise - } + if (calls.length === 2) bothStarted.resolve() + await release.promise return { text: `${agent.kind} completed.` } } const workspace = new LocalWorkspace(options) @@ -106,24 +135,55 @@ test('writable chats in one folder take turns and show who is waiting', { timeou name: `${kind} direct`, agentKinds: [kind], directAgentKind: kind, conversationType: 'direct', workdir: directory, allowWrite: true, })) - const first = workspace.sendMessage({ groupId: groups[0].id, text: 'Write first.' }) - await firstStarted.promise - 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']) + const runs = groups.map(group => workspace.sendMessage({ groupId: group.id, text: 'Run independently.' })) + await bothStarted.promise + assert.deepEqual(calls.map(call => call.sandbox), ['workspace-write', 'workspace-write']) release.resolve() - 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) + await Promise.all(runs) +}) + +test('the same Agent answers in two conversations at once', { timeout: 10000 }, async (t) => { + const { directory, options } = fixture() + const release = deferred() + const allStarted = deferred() + t.after(() => { + release.resolve() + fs.rmSync(directory, { recursive: true, force: true }) + }) + const scheduler = new RunScheduler() + options.runScheduler = scheduler + const calls = [] + options.runAgent = async (agent, _prompt, _cwd, runOptions) => { + calls.push({ kind: agent.kind, sandbox: runOptions.sandbox }) + if (calls.length === 3) allStarted.resolve() + await release.promise + return { text: `${agent.kind} completed.` } + } + const workspace = new LocalWorkspace(options) + await workspace.refreshAgents() + const direct = [1, 2].map(number => workspace.createGroup({ + name: `Hermes chat ${number}`, agentKinds: ['hermes'], directAgentKind: 'hermes', + conversationType: 'direct', workdir: directory, allowWrite: true, + })) + const group = workspace.createGroup({ + name: 'Team', agentKinds: ['hermes', 'codex'], workdir: directory, allowWrite: true, + }) + + const runs = [ + workspace.sendMessage({ groupId: direct[0].id, text: 'First chat.' }), + workspace.sendMessage({ groupId: direct[1].id, text: 'Second chat.' }), + workspace.sendMessage({ groupId: group.id, text: 'Ask the group.', targetKinds: ['hermes'] }), + ] + // Hermes is busy in three conversations and none of them waits for another. + await allStarted.promise + assert.deepEqual(calls.map(call => call.kind), ['hermes', 'hermes', 'hermes']) + assert.equal(scheduler.snapshot().queued.length, 0) + for (const run of workspace.snapshot().runs) { + assert.equal('waitingForWorkspaceKinds' in run, false) + } + release.resolve() + await Promise.all(runs) + assert.equal(scheduler.snapshot().active.global, 0) }) test('Manual V4 retains parallel readers alongside its single writer and stable message order', { timeout: 10000 }, async (t) => { diff --git a/frontend/src/components/ConversationTimelineView.vue b/frontend/src/components/ConversationTimelineView.vue index b0c8024..1154628 100644 --- a/frontend/src/components/ConversationTimelineView.vue +++ b/frontend/src/components/ConversationTimelineView.vue @@ -726,11 +726,6 @@ {{ t(runRoundProgress.unlimited ? 'run.roundProgressUnlimited' : 'run.roundProgress', runRoundProgress) }} - - {{ t('run.waitingForWorkspace', { - agents: activeRun.waitingForWorkspaceKinds.map(kind => agentLabel(kind)).join(', '), - }) }} -
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) ? input.waitingGateIds.slice(0, 32) : []).filter(gateId => typeof gateId === 'string' && HUMAN_GATE_ID.test(gateId)))] diff --git a/frontend/src/locales/en.js b/frontend/src/locales/en.js index 644550c..f231b9c 100644 --- a/frontend/src/locales/en.js +++ b/frontend/src/locales/en.js @@ -201,7 +201,6 @@ export default { 'run.roundProgressUnlimited': 'Round {current} / No round limit', 'run.retainedEvents': '{count} retained events', '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.queued': 'Queued', 'run.status.preparing': 'Preparing', diff --git a/frontend/src/locales/zh.js b/frontend/src/locales/zh.js index 6adb6bf..36e8e01 100644 --- a/frontend/src/locales/zh.js +++ b/frontend/src/locales/zh.js @@ -201,7 +201,6 @@ export default { 'run.roundProgressUnlimited': '第 {current} 轮 / 不限轮数', 'run.retainedEvents': '保留 {count} 条过程事件', 'run.openDetails': '打开运行详情', - 'run.waitingForWorkspace': '{agents} 正在等待同一文件夹中另一个正在写入的 Agent 完成。', 'run.status.pending': '待处理', 'run.status.queued': '排队中', 'run.status.preparing': '准备中',