import { getAgentToolExecutionContext } from "../../../../packages/agent-core/src/tool-execution-context.js"; /** * Per-file mutation queue. * * Serializes reads and mutations targeting the same real file while allowing independent files to run in parallel. */ import { resolveIdentityPathViaExistingAncestorSync } from "../../../infra/boundary-path.js"; import { resolveGlobalMap, resolveGlobalSingleton } from "../../../shared/global-singleton.js"; const fileMutationTails = resolveGlobalMap>( Symbol.for("openclaw.fileMutationTails"), "close-only", ); const keyAdmissions = resolveGlobalSingleton( Symbol.for("openclaw.fileMutationKeyAdmissions"), () => ({ fallbackScope: {}, tails: new WeakMap>() }), ); function resolveLocalFileMutationQueueKey(filePath: string): string { return resolveIdentityPathViaExistingAncestorSync(filePath); } export async function resolveFileMutationQueueKey( filePath: string, resolveQueueKey?: (absolutePath: string, signal?: AbortSignal) => string | Promise, signal?: AbortSignal, ): Promise { return await (resolveQueueKey?.(filePath, signal) ?? resolveLocalFileMutationQueueKey(filePath)); } /** * Preserve source-call admission while backend-owned physical identities resolve concurrently. * Registration is ordered per assistant message; file operations still use only fileMutationTails. */ export async function withFileMutationQueueKeyResolution( keyResolution: Promise, fn: () => Promise, ): Promise { return await withFileMutationQueueKeysResolution( keyResolution.then((key) => [key]), fn, ); } export async function withFileMutationQueueKeysResolution( keysResolution: Promise, fn: () => Promise, ): Promise { const scope = getAgentToolExecutionContext()?.assistantMessage ?? keyAdmissions.fallbackScope; const previousAdmission = keyAdmissions.tails.get(scope) ?? Promise.resolve(); void keysResolution.catch(() => undefined); let operation!: Promise; const admission = previousAdmission.then(async () => { const keys = await keysResolution; operation = enqueueFileMutationQueueKeys(keys, fn); }); const tail = admission.then( () => undefined, () => undefined, ); keyAdmissions.tails.set(scope, tail); const cleanup = () => { if (keyAdmissions.tails.get(scope) === tail) { keyAdmissions.tails.delete(scope); } }; tail.then(cleanup, cleanup); await admission; return await operation; } /** * Serialize file mutation operations targeting the same file. * Operations for different files still run in parallel. */ export async function withFileMutationQueue(filePath: string, fn: () => Promise): Promise { return await withFileMutationQueues([filePath], fn); } async function withFileMutationQueues( filePaths: readonly string[], fn: () => Promise, ): Promise { return await enqueueFileMutationQueueKeys(filePaths.map(resolveLocalFileMutationQueueKey), fn); } function enqueueFileMutationQueueKeys( queueKeys: readonly string[], fn: () => Promise, ): Promise { const keys = [...new Set(queueKeys)].toSorted(); const current = Promise.all( keys.map((key) => (fileMutationTails.get(key) ?? Promise.resolve()).catch(() => undefined)), ).then(fn); const tail = current.then( () => undefined, () => undefined, ); for (const key of keys) { fileMutationTails.set(key, tail); } const cleanup = () => { for (const key of keys) { if (fileMutationTails.get(key) === tail) { fileMutationTails.delete(key); } } }; tail.then(cleanup, cleanup); return current; }