Q126: a file write and its record succeed or fail together
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
11cdc4af20
commit
5799a3561f
7 files changed
+119
-48
No files matched your search
@@ -3,9 +3,8 @@ import { createId } from "../../shared/ids";
|
||||
import { withSavepoint } from "../database/connection";
|
||||
import { holdDeliverable, reviewsDeliverables, type HeldTarget } from "./held-deliverables";
|
||||
import { ApprovalService } from "./approval-service";
|
||||
import { writeDeliverable } from "./deliverable-writer";
|
||||
import { DocumentService } from "./document-service";
|
||||
import { recordWorkspaceFile } from "./files/file-object-store";
|
||||
import { writeRecordedFile } from "./files/file-object-store";
|
||||
import { requireString } from "./agent-tool-args";
|
||||
import type { AgentToolContext, AgentToolDeps } from "./agent-tools";
|
||||
import type { JsonRecord } from "./provider-http";
|
||||
@@ -24,13 +23,13 @@ export function writeDeliverableTool(deps: AgentToolDeps, ctx: AgentToolContext,
|
||||
holdForReview(deps, ctx, { kind: "file", workspaceRoot: ctx.workspace.localPath, relativePath: path }, content, `Review deliverable: ${path}`);
|
||||
return `Deliverable ${path} (${content.length} chars) is held for board review in Approvals. It is not written until approved, so no one can use it yet.`;
|
||||
}
|
||||
const written = writeDeliverable({ workspaceRoot: ctx.workspace.localPath, relativePath: path, content });
|
||||
const fileId = recordWorkspaceFile(deps.db, {
|
||||
const { fileId, written } = writeRecordedFile(deps.db, {
|
||||
orgId: ctx.orgId,
|
||||
ownerType: "agent",
|
||||
ownerId: ctx.agent.id,
|
||||
workspaceRoot: ctx.workspace.localPath,
|
||||
relativePath: written,
|
||||
relativePath: path,
|
||||
content,
|
||||
});
|
||||
return `Wrote deliverable ${written} (file_id=${fileId}, chars=${content.length}).`;
|
||||
}
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { dirname } from "node:path";
|
||||
import { resolveWorkspacePath } from "./files/workspace-path";
|
||||
|
||||
/**
|
||||
* Writes an agent deliverable into the run's bound workspace folder.
|
||||
*
|
||||
* The target path is resolved against the workspace root and rejected if it
|
||||
* escapes that root (path traversal / absolute paths), so an agent can only
|
||||
* write inside the folder the operator bound to the company.
|
||||
*/
|
||||
export function writeDeliverable(input: { workspaceRoot: string; relativePath: string; content: string }): string {
|
||||
const { absolute, relativePosix } = resolveWorkspacePath(input.workspaceRoot, input.relativePath);
|
||||
mkdirSync(dirname(absolute), { recursive: true });
|
||||
writeFileSync(absolute, input.content, "utf8");
|
||||
return relativePosix;
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
import { mkdirSync, statSync } from "node:fs";
|
||||
import { mkdirSync, rmSync, statSync } from "node:fs";
|
||||
import { copyFile } from "node:fs/promises";
|
||||
import { basename, join } from "node:path";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
@@ -7,7 +7,7 @@ import type { ImportFileInput, OpenFileInput } from "../../shared/commands";
|
||||
import { FILE_SCAN_STATE } from "../../shared/domain";
|
||||
import { createId } from "../../shared/ids";
|
||||
import { AuditService } from "./audit-service";
|
||||
import { fileKind, readVerifiedFile, scanFileAt, upsertFileObject } from "./files/file-object-store";
|
||||
import { fileKind, readVerifiedFile, scanFileAt, upsertFileObject, type FileObjectInput } from "./files/file-object-store";
|
||||
|
||||
const MAX_IMPORT_BYTES = 1024 * 1024 * 1024;
|
||||
|
||||
@@ -49,6 +49,17 @@ export class FileService {
|
||||
const target = join(orgDir, `${id}-${name}`);
|
||||
// The stored copy is what gets hashed and scanned, so the record describes exactly the bytes kept; nothing is held in memory whole.
|
||||
await copyFile(sourcePath, target);
|
||||
try {
|
||||
return await this.recordImport({ id, orgId, name, target, access, now });
|
||||
} catch (error) {
|
||||
// An import that is not recorded leaves no stored copy behind.
|
||||
rmSync(target, { force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async recordImport(input: { id: string; orgId: string; name: string; target: string; access: FileObjectInput["access"]; now: string }): Promise<string> {
|
||||
const { id, orgId, name, target, access, now } = input;
|
||||
const { scan, sha256: hash, size } = await scanFileAt(target, name);
|
||||
console.info(`[files] imported ${name} (${formatBytes(size)})`);
|
||||
this.db.exec("begin immediate");
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { open } from "node:fs/promises";
|
||||
import { basename, dirname, extname, isAbsolute, join, relative } from "node:path";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
@@ -226,9 +226,10 @@ export function recordWorkspaceFile(db: DatabaseSync, input: {
|
||||
const now = new Date().toISOString();
|
||||
const sha256 = createHash("sha256").update(bytes).digest("hex");
|
||||
// The id keeps exactly these bytes: a content-addressed copy the workspace file can be rewritten without touching.
|
||||
const copy = keepCopy(input.workspaceRoot, absolute, sha256, bytes);
|
||||
// A savepoint, not a transaction: an approved deliverable is recorded inside the approval's own transaction.
|
||||
return withSavepoint(db, "record_workspace_file", () => upsertFileObject(db, {
|
||||
const { copy, created } = keepCopy(input.workspaceRoot, absolute, sha256, bytes);
|
||||
try {
|
||||
// A savepoint, not a transaction: an approved deliverable is recorded inside the approval's own transaction.
|
||||
return withSavepoint(db, "record_workspace_file", () => upsertFileObject(db, {
|
||||
orgId: input.orgId,
|
||||
name,
|
||||
kind: fileKind(name),
|
||||
@@ -242,6 +243,27 @@ export function recordWorkspaceFile(db: DatabaseSync, input: {
|
||||
scan,
|
||||
now,
|
||||
}));
|
||||
} catch (error) {
|
||||
// A copy made for a record that failed goes with it; one that already backed another record stays.
|
||||
if (created) rmSync(copy, { force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/** Writes a workspace file and records it as one result: if recording fails, the old file comes back, or the new one goes. */
|
||||
export function writeRecordedFile(db: DatabaseSync, input: { orgId: string; ownerType: string; ownerId: string | null; workspaceRoot: string; relativePath: string; content: string }): { fileId: string; written: string } {
|
||||
const { absolute, relativePosix } = resolveWorkspacePath(input.workspaceRoot, input.relativePath);
|
||||
const previous = existsSync(absolute) ? readFileSync(absolute) : null;
|
||||
try {
|
||||
mkdirSync(dirname(absolute), { recursive: true });
|
||||
writeFileSync(absolute, input.content, "utf8");
|
||||
const fileId = recordWorkspaceFile(db, { orgId: input.orgId, ownerType: input.ownerType, ownerId: input.ownerId, workspaceRoot: input.workspaceRoot, relativePath: relativePosix });
|
||||
return { fileId, written: relativePosix };
|
||||
} catch (error) {
|
||||
if (previous) writeFileSync(absolute, previous);
|
||||
else rmSync(absolute, { force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/** A recorded file's bytes, refused when they are gone or no longer match the hash it was recorded with. */
|
||||
@@ -256,13 +278,21 @@ export function readVerifiedFile(file: { name: string; storagePath: string; sha2
|
||||
}
|
||||
|
||||
/** The content-addressed copy of a workspace file, written once; it keeps the extension so the file still opens with its app. */
|
||||
function keepCopy(workspaceRoot: string, source: string, sha256: string, bytes: Buffer): string {
|
||||
function keepCopy(workspaceRoot: string, source: string, sha256: string, bytes: Buffer): { copy: string; created: boolean } {
|
||||
const copy = join(workspaceRoot, ".agentcompany", "files", `${sha256}${extname(source)}`);
|
||||
if (!existsSync(copy)) {
|
||||
mkdirSync(dirname(copy), { recursive: true });
|
||||
writeFileSync(copy, bytes);
|
||||
// An existing copy is reused only if its bytes are the ones its name promises.
|
||||
if (existsSync(copy) && createHash("sha256").update(readFileSync(copy)).digest("hex") === sha256) return { copy, created: false };
|
||||
mkdirSync(dirname(copy), { recursive: true });
|
||||
// Written aside and renamed into place, so a write that fails never leaves a partial copy under the final name.
|
||||
const pending = `${copy}.${createId("tmp")}`;
|
||||
try {
|
||||
writeFileSync(pending, bytes);
|
||||
renameSync(pending, copy);
|
||||
} catch (error) {
|
||||
rmSync(pending, { force: true });
|
||||
throw error;
|
||||
}
|
||||
return copy;
|
||||
return { copy, created: true };
|
||||
}
|
||||
|
||||
/** Files scanned before whole-file scanning are scanned again when their bytes are still the recorded ones; the rest are marked as sampled. */
|
||||
@@ -294,7 +324,7 @@ export function preserveRecordedFiles(db: DatabaseSync): void {
|
||||
console.warn(`[files] ${row.id} was rewritten before its bytes could be kept; it opens the current file`);
|
||||
continue;
|
||||
}
|
||||
db.prepare("update file_objects set storage_path = ? where id = ?").run(keepCopy(workspace.local_path, row.storage_path, row.sha256, bytes), row.id);
|
||||
db.prepare("update file_objects set storage_path = ? where id = ?").run(keepCopy(workspace.local_path, row.storage_path, row.sha256, bytes).copy, row.id);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { ApprovalPin } from "../../shared/approval-pin";
|
||||
import { writeDeliverable } from "./deliverable-writer";
|
||||
import { DocumentService } from "./document-service";
|
||||
import { recordWorkspaceFile } from "./files/file-object-store";
|
||||
import { writeRecordedFile } from "./files/file-object-store";
|
||||
import { resolveWorkspacePath } from "./files/workspace-path";
|
||||
|
||||
/** What a held deliverable will become once approved. */
|
||||
@@ -84,16 +82,7 @@ function publish(db: DatabaseSync, row: HeldRow): void {
|
||||
const { absolute } = resolveWorkspacePath(row.workspace_root!, row.relative_path!);
|
||||
// The file's pin is known before it is written, so it is recorded first and the disk write stays last.
|
||||
pinApproval(db, row, { kind: "file", storagePath: absolute, sha256: createHash("sha256").update(row.content, "utf8").digest("hex") });
|
||||
const previous = existsSync(absolute) ? readFileSync(absolute) : null;
|
||||
try {
|
||||
const written = writeDeliverable({ workspaceRoot: row.workspace_root!, relativePath: row.relative_path!, content: row.content });
|
||||
recordWorkspaceFile(db, { orgId: row.org_id, ownerType: "agent", ownerId: row.agent_id, workspaceRoot: row.workspace_root!, relativePath: written });
|
||||
} catch (error) {
|
||||
// A publication that fails leaves the workspace as it was: the old file back, or none.
|
||||
if (previous) writeFileSync(absolute, previous);
|
||||
else rmSync(absolute, { force: true });
|
||||
throw error;
|
||||
}
|
||||
writeRecordedFile(db, { orgId: row.org_id, ownerType: "agent", ownerId: row.agent_id, workspaceRoot: row.workspace_root!, relativePath: row.relative_path!, content: row.content });
|
||||
return;
|
||||
}
|
||||
const documents = new DocumentService(db);
|
||||
|
||||
@@ -66,4 +66,19 @@ describe("files shared in a private conversation", () => {
|
||||
rmSync(huge);
|
||||
connection.close();
|
||||
});
|
||||
|
||||
test("an import whose record fails leaves no stored copy behind", async () => {
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-import-fail-${randomUUID()}.db`));
|
||||
let orgId = "";
|
||||
connection.transaction(() => {
|
||||
orgId = new OrganizationService(connection.db).createOrganization({ organizationName: "Acme", employeeName: "Yang", mission: "m", goal: "g" });
|
||||
});
|
||||
const source = join(tmpdir(), `ac-import-fail-${randomUUID()}.md`);
|
||||
writeFileSync(source, "notes");
|
||||
connection.db.exec("create trigger registry_down before insert on file_objects begin select raise(abort, 'registry down'); end;");
|
||||
const artifacts = join(tmpdir(), `ac-import-fail-art-${randomUUID()}`);
|
||||
await expect(new FileService(connection.db, artifacts).importPath(orgId, source, null)).rejects.toThrow("registry down");
|
||||
expect(readdirSync(artifacts, { recursive: true }).filter((entry) => String(entry).endsWith(".md"))).toEqual([]);
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { describe, expect, test } from "vitest";
|
||||
@@ -8,6 +8,7 @@ import { OrganizationService } from "../../src/main/services/organization-servic
|
||||
import { preserveRecordedFiles, recordWorkspaceFile, rescanLegacyFiles } from "../../src/main/services/files/file-object-store";
|
||||
import { ChatService } from "../../src/main/services/chat-service";
|
||||
import { readAttachmentTool } from "../../src/main/services/agent-attachment-tools";
|
||||
import { dispatchAgentTool } from "../../src/main/services/agent-tools";
|
||||
|
||||
describe("recorded workspace files", () => {
|
||||
test("a file id keeps the bytes it was recorded with after the workspace file is rewritten", () => {
|
||||
@@ -106,3 +107,46 @@ describe("files scanned before whole-file scanning", () => {
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
|
||||
describe("a file write whose record fails", () => {
|
||||
test("leaves the workspace and the artifact store as they were", async () => {
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-partial-${randomUUID()}.db`));
|
||||
const root = join(tmpdir(), `ac-partial-ws-${randomUUID()}`);
|
||||
mkdirSync(root, { recursive: true });
|
||||
let orgId = "";
|
||||
let ceoId = "";
|
||||
connection.transaction(() => {
|
||||
const organization = new OrganizationService(connection.db);
|
||||
orgId = organization.createOrganization({ organizationName: "Acme", employeeName: "Dana", mission: "m", goal: "g" });
|
||||
ceoId = organization.createAgent({ orgId, name: "CEO", role: "CEO", department: "Executive", modelProviderId: null, modelId: null, budgetCents: 1000, managerAgentId: null, roleLevel: "c_suite" });
|
||||
});
|
||||
writeFileSync(join(root, "kept.md"), "old");
|
||||
connection.db.exec("create trigger registry_down before insert on file_objects begin select raise(abort, 'registry down'); end;");
|
||||
const deps = { db: connection.db, organization: new OrganizationService(connection.db), chat: new ChatService(connection.db) };
|
||||
const ctx = { orgId, agent: { id: ceoId, modelProviderId: null, modelId: null }, taskProjectId: null, taskGoalId: null, taskId: null, sourceThreadId: null, workspace: { id: "wsp", localPath: root, writeAccess: true } };
|
||||
expect(dispatchAgentTool(deps, ctx, { id: "c1", name: "write_deliverable", arguments: { path: "kept.md", content: "new" } })).toBe("ERROR: registry down");
|
||||
expect(dispatchAgentTool(deps, ctx, { id: "c2", name: "write_deliverable", arguments: { path: "fresh.md", content: "new" } })).toBe("ERROR: registry down");
|
||||
expect(readFileSync(join(root, "kept.md"), "utf8")).toBe("old");
|
||||
expect(existsSync(join(root, "fresh.md"))).toBe(false);
|
||||
// No content copy outlives the record that failed.
|
||||
expect(existsSync(join(root, ".agentcompany", "files")) ? readdirSync(join(root, ".agentcompany", "files")) : []).toEqual([]);
|
||||
connection.close();
|
||||
});
|
||||
|
||||
test("a damaged content copy is replaced, never reused", () => {
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-damaged-${randomUUID()}.db`));
|
||||
const root = join(tmpdir(), `ac-damaged-ws-${randomUUID()}`);
|
||||
mkdirSync(join(root, ".agentcompany", "files"), { recursive: true });
|
||||
let orgId = "";
|
||||
connection.transaction(() => {
|
||||
orgId = new OrganizationService(connection.db).createOrganization({ organizationName: "Acme", employeeName: "Dana", mission: "m", goal: "g" });
|
||||
});
|
||||
writeFileSync(join(root, "plan.md"), "new");
|
||||
const sha = createHash("sha256").update("new").digest("hex");
|
||||
writeFileSync(join(root, ".agentcompany", "files", `${sha}.md`), "n");
|
||||
const fileId = recordWorkspaceFile(connection.db, { orgId, ownerType: "agent", ownerId: null, workspaceRoot: root, relativePath: "plan.md" });
|
||||
const stored = (connection.db.prepare("select storage_path from file_objects where id = ?").get(fileId) as { storage_path: string }).storage_path;
|
||||
expect(readFileSync(stored, "utf8")).toBe("new");
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user