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 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 5.5 committed 2026-10-08 03:25:39 -04:00
1 parent 72eb645067
commit 2a7b17a47a
5 files changed
+168 -5

No files matched your search

+8 -5
View File
@@ -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<QueuedRun & { task_id: string }>;
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);
@@ -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(`
+39
View File
@@ -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;
}