From 2a7b17a47a5c3bf7e7544819f22f3347ab271a09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 03:25:39 -0400 Subject: [PATCH] Q8: one writing run per workspace directory Write-granted runs whose directories overlap (same folder, symlink or junction, or nested) wait for each other, at claim time and for manual runs. Co-Authored-By: Claude Opus 5.5 --- src/main/app/task-queue-worker.ts | 13 ++-- src/main/services/task-execution-service.ts | 3 + src/main/services/workspace-write-lock.ts | 39 +++++++++++ tests/integration/task-context-scope.test.ts | 46 +++++++++++++ tests/unit/task-queue-worker.test.ts | 72 ++++++++++++++++++++ 5 files changed, 168 insertions(+), 5 deletions(-) create mode 100644 src/main/services/workspace-write-lock.ts diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index 5c2b003d..895a0859 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -5,6 +5,7 @@ import { expireStaleChatWakes } from "../services/chat-agent-wake-service"; import { OrganizationService } from "../services/organization-service"; import { spawnGoalReviews } from "../services/goal-review-service"; import { TaskExecutionService } from "../services/task-execution-service"; +import { runningWorkspaceWriter } from "../services/workspace-write-lock"; const DEFAULT_INTERVAL_MS = 5_000; @@ -48,6 +49,7 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { * autonomous work; within each lane, first come first served. Non-chat runs * carry no sourceSenderType and fall into the autonomous lane. An agent * runs one session at a time: its queued runs wait while one is running. + * A workspace takes one writing run at a time: write-granted runs wait. */ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = [], lane: ClaimLane = "any"): QueuedRun | null { return connection.transaction(() => { @@ -57,17 +59,18 @@ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProvi const laneFilter = lane === "interactive" ? "and json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board')" : ""; - const row = connection.db + const candidates = connection.db .prepare(` - select id, org_id, provider_id from task_runs where state = 'queued' ${exclusion} ${laneFilter} + select id, org_id, provider_id, task_id from task_runs where state = 'queued' ${exclusion} ${laneFilter} and not exists (select 1 from task_runs busy where busy.state = 'running' and busy.agent_id = task_runs.agent_id) order by case when json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc, created_at asc - limit 1 `) - .get(...excludedProviderIds) as QueuedRun | undefined; - if (!row) return null; + .all(...excludedProviderIds) as unknown as Array; + const candidate = candidates.find((run) => runningWorkspaceWriter(connection.db, run.task_id) === null); + if (!candidate) return null; + const row: QueuedRun = { id: candidate.id, org_id: candidate.org_id, provider_id: candidate.provider_id }; const result = connection.db .prepare("update task_runs set state = 'running', started_at = coalesce(started_at, ?) where id = ? and state = 'queued'") .run(new Date().toISOString(), row.id); diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 75fa6d75..7e4a8f0e 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -25,6 +25,7 @@ import { } from "./task-prompts"; import { renderTaskHandoff } from "./agent-task-handoff"; import { createRunMessagePoller, loadChatWakeContext } from "./chat-wake-context"; +import { runningWorkspaceWriter } from "./workspace-write-lock"; import { loadComposioTools } from "./composio/composio-tool-bridge"; import { connectorDisplayName } from "./composio/connector-brands"; import { ConnectorGrantsService } from "./composio/connector-grants"; @@ -321,6 +322,8 @@ export class TaskExecutionService { const busy = this.db.prepare("select t.title from task_runs r join tasks t on t.id = r.task_id where r.agent_id = ? and r.state = 'running' limit 1") .get(agentId) as { title: string } | undefined; if (busy) throw new Error(`This agent is already working on "${busy.title}"; run again when that session ends.`); + const writer = runningWorkspaceWriter(this.db, taskId); + if (writer !== null) throw new Error(`"${writer.title}" is changing this workspace right now; run again when it finishes.`); const runId = createId("run"); const now = new Date().toISOString(); this.db.prepare(` diff --git a/src/main/services/workspace-write-lock.ts b/src/main/services/workspace-write-lock.ts new file mode 100644 index 00000000..205c11d9 --- /dev/null +++ b/src/main/services/workspace-write-lock.ts @@ -0,0 +1,39 @@ +import type { DatabaseSync } from "node:sqlite"; +import { existsSync, realpathSync } from "node:fs"; +import { basename, dirname, join, resolve } from "node:path"; + +/** A running write task whose directory overlaps this task's, or null. */ +export function runningWorkspaceWriter(db: DatabaseSync, taskId: string): { title: string } | null { + const task = db.prepare(` + select t.requires_filesystem, w.local_path + from tasks t left join workspaces w on w.id = t.workspace_id + where t.id = ? + `).get(taskId) as { requires_filesystem: number; local_path: string | null } | undefined; + if (!task) throw new Error(`Task ${taskId} was not found.`); + if (task.requires_filesystem !== 1 || !task.local_path) return null; + const mine = normalizeDirectory(task.local_path); + const writers = db.prepare(` + select t.title, w.local_path + from task_runs r + join tasks t on t.id = r.task_id + join workspaces w on w.id = t.workspace_id + where r.state = 'running' and t.requires_filesystem = 1 + `).all() as Array<{ title: string; local_path: string }>; + for (const writer of writers) { + const other = normalizeDirectory(writer.local_path); + if (mine === other || mine.startsWith(`${other}/`) || other.startsWith(`${mine}/`)) return { title: writer.title }; + } + return null; +} + +function normalizeDirectory(localPath: string): string { + let existing = resolve(localPath); + const missing: string[] = []; + while (!existsSync(existing) && dirname(existing) !== existing) { + missing.unshift(basename(existing)); + existing = dirname(existing); + } + const real = existsSync(existing) ? realpathSync.native(existing) : existing; + const path = join(real, ...missing).replace(/\\/g, "/").replace(/\/+$/, ""); + return process.platform === "win32" ? path.toLowerCase() : path; +} diff --git a/tests/integration/task-context-scope.test.ts b/tests/integration/task-context-scope.test.ts index 70379683..2c6e56bd 100644 --- a/tests/integration/task-context-scope.test.ts +++ b/tests/integration/task-context-scope.test.ts @@ -10,6 +10,7 @@ import { TaskExecutionService } from "../../src/main/services/task-execution-ser import { OrganizationService } from "../../src/main/services/organization-service"; import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; import { canonicalMentionHandle } from "../../src/shared/mention-handles"; +import { RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; import type { AgentTurnResult, ProviderToolTurnInput, ToolCall } from "../../src/main/services/provider-tool-turn"; const testVault = { @@ -114,4 +115,49 @@ describe("task prompt scope", () => { expect(calls).toBe(0); connection.close(); }); + + test("a manual write run waits while another run is changing the same workspace", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-wsplock-${randomUUID()}.db`)); + const turn = async (): Promise => ({ text: "done", toolCalls: [], rawResponseJson: "{}" }); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const organization = new OrganizationService(connection.db); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + let ceoId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "OrbitOps AI", companyDescription: "Ops.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Stand up a service business.", goalDescription: "Win customers.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }); + orgId = summary.orgId; + ceoId = summary.ceoAgentId; + }); + const providerId = (connection.db.prepare("select model_provider_id as id from agents where id = ?").get(ceoId) as { id: string }).id; + let busyTask = ""; + let waitingTask = ""; + connection.transaction(() => { + const workspaceId = organization.createWorkspace({ orgId, name: "repo", localPath: tmpdir(), projectId: null }); + const engineerId = organization.createAgent({ + orgId, name: "Builder", role: "Engineer", department: "Engineering", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId, + }); + const make = (title: string, assignee: string) => organization.createTask({ + orgId, title, description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: assignee, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: true, + }); + busyTask = make("Refactor storage", engineerId); + waitingTask = make("Bump deps", ceoId); + }); + const kickoff = claimNextQueuedRun(connection)!; + connection.db.prepare("update task_runs set task_id = ?, agent_id = (select assignee_agent_id from tasks where id = ?) where id = ?") + .run(busyTask, busyTask, kickoff.id); + + await expect(execution.runTask({ orgId, taskId: waitingTask })).rejects.toThrow('"Refactor storage" is changing this workspace right now'); + connection.db.prepare("update tasks set title = '' where id = ?").run(busyTask); + await expect(execution.runTask({ orgId, taskId: waitingTask })).rejects.toThrow("is changing this workspace right now"); + connection.close(); + }); }); diff --git a/tests/unit/task-queue-worker.test.ts b/tests/unit/task-queue-worker.test.ts index 6b8a850d..e5f1667d 100644 --- a/tests/unit/task-queue-worker.test.ts +++ b/tests/unit/task-queue-worker.test.ts @@ -1,5 +1,6 @@ import { randomUUID } from "node:crypto"; import { tmpdir } from "node:os"; +import { mkdirSync, symlinkSync } from "node:fs"; import { join } from "node:path"; import { describe, expect, test } from "vitest"; import { RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; @@ -79,3 +80,74 @@ describe("task queue worker claim", () => { connection.close(); }); }); + +describe("one writing run per workspace", () => { + test("a second write-granted run in the same workspace waits; read-only work proceeds", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let providerId = ""; + const agents: string[] = []; + const tasks: string[] = []; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "Acme", employeeName: "Dana", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + const workspaceId = organization.createWorkspace({ orgId, name: "repo", localPath: tmpdir(), projectId: null }); + for (const [name, writes] of [["Ana", true], ["Ben", true], ["Cy", false]] as const) { + const agentId = organization.createAgent({ orgId, name, role: "Engineer", department: "Engineering", modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: null }); + agents.push(agentId); + tasks.push(organization.createTask({ + orgId, title: `${name} work`, description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: agentId, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: writes, + })); + } + }); + const runs = tasks.map((taskId, index) => seedQueuedRun(connection, orgId, taskId, agents[index], providerId)); + connection.db.prepare("update task_runs set created_at = ? where id = ?").run("2026-01-01T00:00:01Z", runs[0]); + connection.db.prepare("update task_runs set created_at = ? where id = ?").run("2026-01-01T00:00:02Z", runs[1]); + connection.db.prepare("update task_runs set created_at = ? where id = ?").run("2026-01-01T00:00:03Z", runs[2]); + + expect(claimNextQueuedRun(connection)?.id).toBe(runs[0]); + expect(claimNextQueuedRun(connection)?.id).toBe(runs[2]); + expect(claimNextQueuedRun(connection)).toBeNull(); + connection.db.prepare("update task_runs set state = 'completed' where id = ?").run(runs[0]); + expect(claimNextQueuedRun(connection)?.id).toBe(runs[1]); + connection.close(); + }); +}); + +describe("workspace write lock follows the directory", () => { + test("one folder, a junction to it, and nested folders (even not yet created) share the lock", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + const root = join(tmpdir(), `ac-repo-${randomUUID()}`); + mkdirSync(root, { recursive: true }); + const linked = join(tmpdir(), `ac-link-${randomUUID()}`); + symlinkSync(root, linked, "junction"); + let orgId = ""; + let providerId = ""; + const runs: string[] = []; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "Acme", employeeName: "Dana", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + const paths = [root, linked, `${root}/`, join(root, "packages", "api"), join(linked, "packages", "api", "src")]; + paths.forEach((localPath, index) => { + const workspaceId = organization.createWorkspace({ orgId, name: `w${index}`, localPath, projectId: null }); + const agentId = organization.createAgent({ orgId, name: `Dev${index}`, role: "Engineer", department: "Engineering", modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: null }); + const taskId = organization.createTask({ + orgId, title: `t${index}`, description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: agentId, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: true, + }); + runs.push(seedQueuedRun(connection, orgId, taskId, agentId, providerId)); + }); + }); + runs.forEach((runId, index) => connection.db.prepare("update task_runs set created_at = ? where id = ?").run(`2026-01-01T00:00:0${index}Z`, runId)); + + for (const runId of runs) { + expect(claimNextQueuedRun(connection)?.id).toBe(runId); + expect(claimNextQueuedRun(connection)).toBeNull(); + connection.db.prepare("update task_runs set state = 'completed' where id = ?").run(runId); + } + connection.close(); + }); +});