mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(agent): reserve JSONL session destinations
Coordinate create and fork operations by cwd and session ID so only the first caller can publish a destination. Split destination-reservation conformance from active-source snapshot coordination.
This commit is contained in:
David Brailovsky
parent
71026970a0
commit
95e3765ade
4 changed files with 123 additions and 70 deletions
|
|
@ -51,6 +51,7 @@ export class JsonlSessionRepo
|
|||
private readonly fileSystem: FileSystem;
|
||||
private readonly sessionsRootInput: string;
|
||||
private readonly now: () => number;
|
||||
private readonly activeCreateDestinations = new Set<string>();
|
||||
private rootPromise: Promise<string> | undefined;
|
||||
private closed = false;
|
||||
private closePromise: Promise<void> | undefined;
|
||||
|
|
@ -64,45 +65,43 @@ export class JsonlSessionRepo
|
|||
async create(options: JsonlSessionCreateOptions): Promise<Session<JsonlSessionMetadata>> {
|
||||
this.assertOpen();
|
||||
const createdAt = this.now();
|
||||
const id = options.id ?? uuidv7(createdAt);
|
||||
const cwd = fileValue(
|
||||
await this.fileSystem.absolutePath(options.cwd),
|
||||
`Failed to resolve session cwd ${options.cwd}`,
|
||||
);
|
||||
// TODO: Prevent concurrent create calls from passing this asynchronous check for the same cwd and id.
|
||||
if (await this.sessionIdExists(cwd, id)) throw new Error(`Session already exists: ${id}`);
|
||||
const path = await this.createPath(cwd, createdAt, id);
|
||||
const header: JsonlStorageHeader = {
|
||||
v: JSONL_FORMAT_VERSION,
|
||||
kind: "header",
|
||||
id,
|
||||
storageVersion: JSONL_STORAGE_VERSION,
|
||||
createdAt,
|
||||
cwd,
|
||||
...(options.parentSessionId === undefined ? {} : { parentSessionId: options.parentSessionId }),
|
||||
};
|
||||
// TODO: do we want to make session creation atomic?
|
||||
const storage = await JsonlStorage.create({ fileSystem: this.fileSystem, path, now: this.now }, header);
|
||||
try {
|
||||
await storage.commit({
|
||||
writes: [
|
||||
{ kind: "register", op: "set", namespace: "lane.leaf", key: "main", value: null },
|
||||
{
|
||||
kind: "register",
|
||||
op: "set",
|
||||
namespace: "lane.state",
|
||||
key: "main",
|
||||
value: { currentOperationId: null, pendingNextRun: [] },
|
||||
},
|
||||
],
|
||||
});
|
||||
const info = fileValue(await this.fileSystem.fileInfo(path), `Failed to read session ${path}`);
|
||||
return new StorageBackedSession(metadataFromHeader(header, path, info.mtimeMs), storage);
|
||||
} catch (error) {
|
||||
await storage.close().catch(() => undefined);
|
||||
await this.fileSystem.remove(path, { force: true });
|
||||
throw error;
|
||||
}
|
||||
const destination = await this.resolveCreateDestination(options.cwd, options.id, createdAt);
|
||||
return this.claimCreateDestination(destination, async () => {
|
||||
const { cwd, id } = destination;
|
||||
if (await this.sessionIdExists(cwd, id)) throw new Error(`Session already exists: ${id}`);
|
||||
const path = await this.createPath(cwd, createdAt, id);
|
||||
const header: JsonlStorageHeader = {
|
||||
v: JSONL_FORMAT_VERSION,
|
||||
kind: "header",
|
||||
id,
|
||||
storageVersion: JSONL_STORAGE_VERSION,
|
||||
createdAt,
|
||||
cwd,
|
||||
...(options.parentSessionId === undefined ? {} : { parentSessionId: options.parentSessionId }),
|
||||
};
|
||||
// TODO: do we want to make session creation atomic?
|
||||
const storage = await JsonlStorage.create({ fileSystem: this.fileSystem, path, now: this.now }, header);
|
||||
try {
|
||||
await storage.commit({
|
||||
writes: [
|
||||
{ kind: "register", op: "set", namespace: "lane.leaf", key: "main", value: null },
|
||||
{
|
||||
kind: "register",
|
||||
op: "set",
|
||||
namespace: "lane.state",
|
||||
key: "main",
|
||||
value: { currentOperationId: null, pendingNextRun: [] },
|
||||
},
|
||||
],
|
||||
});
|
||||
const info = fileValue(await this.fileSystem.fileInfo(path), `Failed to read session ${path}`);
|
||||
return new StorageBackedSession(metadataFromHeader(header, path, info.mtimeMs), storage);
|
||||
} catch (error) {
|
||||
await storage.close().catch(() => undefined);
|
||||
await this.fileSystem.remove(path, { force: true });
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async open(metadata: JsonlSessionMetadata): Promise<Session<JsonlSessionMetadata>> {
|
||||
|
|
@ -142,36 +141,39 @@ export class JsonlSessionRepo
|
|||
async fork(source: JsonlSessionMetadata, options: ForkOptions): Promise<Session<JsonlSessionMetadata>> {
|
||||
this.assertOpen();
|
||||
const createdAt = this.now();
|
||||
const id = options.id ?? uuidv7(createdAt);
|
||||
if (await this.sessionIdExists(source.cwd, id)) throw new Error(`Session already exists: ${id}`);
|
||||
const destination = await this.resolveCreateDestination(source.cwd, options.id, createdAt);
|
||||
return this.claimCreateDestination(destination, async () => {
|
||||
const { cwd, id } = destination;
|
||||
if (await this.sessionIdExists(cwd, id)) throw new Error(`Session already exists: ${id}`);
|
||||
|
||||
const sourceStorage = await this.loadStorage(source);
|
||||
const sourceSnapshot = await sourceStorage.snapshot().finally(() => sourceStorage.close());
|
||||
const snapshot = createForkSnapshot(sourceSnapshot, options);
|
||||
const path = await this.createPath(source.cwd, createdAt, id);
|
||||
const header: JsonlStorageHeader = {
|
||||
v: JSONL_FORMAT_VERSION,
|
||||
kind: "header",
|
||||
id,
|
||||
storageVersion: JSONL_STORAGE_VERSION,
|
||||
createdAt,
|
||||
cwd: source.cwd,
|
||||
parentSessionId: source.id,
|
||||
};
|
||||
let storage: JsonlStorage | undefined;
|
||||
try {
|
||||
storage = await JsonlStorage.createFromSnapshot(
|
||||
{ fileSystem: this.fileSystem, path, now: this.now },
|
||||
header,
|
||||
snapshot,
|
||||
);
|
||||
const info = fileValue(await this.fileSystem.fileInfo(path), `Failed to read session ${path}`);
|
||||
return new StorageBackedSession(metadataFromHeader(header, path, info.mtimeMs), storage);
|
||||
} catch (error) {
|
||||
await storage?.close().catch(() => undefined);
|
||||
await this.fileSystem.remove(path, { force: true });
|
||||
throw error;
|
||||
}
|
||||
const sourceStorage = await this.loadStorage(source);
|
||||
const sourceSnapshot = await sourceStorage.snapshot().finally(() => sourceStorage.close());
|
||||
const snapshot = createForkSnapshot(sourceSnapshot, options);
|
||||
const path = await this.createPath(cwd, createdAt, id);
|
||||
const header: JsonlStorageHeader = {
|
||||
v: JSONL_FORMAT_VERSION,
|
||||
kind: "header",
|
||||
id,
|
||||
storageVersion: JSONL_STORAGE_VERSION,
|
||||
createdAt,
|
||||
cwd,
|
||||
parentSessionId: source.id,
|
||||
};
|
||||
let storage: JsonlStorage | undefined;
|
||||
try {
|
||||
storage = await JsonlStorage.createFromSnapshot(
|
||||
{ fileSystem: this.fileSystem, path, now: this.now },
|
||||
header,
|
||||
snapshot,
|
||||
);
|
||||
const info = fileValue(await this.fileSystem.fileInfo(path), `Failed to read session ${path}`);
|
||||
return new StorageBackedSession(metadataFromHeader(header, path, info.mtimeMs), storage);
|
||||
} catch (error) {
|
||||
await storage?.close().catch(() => undefined);
|
||||
await this.fileSystem.remove(path, { force: true });
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
close(): Promise<void> {
|
||||
|
|
@ -182,6 +184,30 @@ export class JsonlSessionRepo
|
|||
return this.closePromise;
|
||||
}
|
||||
|
||||
private async resolveCreateDestination(
|
||||
cwdInput: string,
|
||||
id: string | undefined,
|
||||
createdAt: number,
|
||||
): Promise<{ cwd: string; id: string }> {
|
||||
const destinationId = id ?? uuidv7(createdAt);
|
||||
const cwd = fileValue(await this.fileSystem.absolutePath(cwdInput), `Failed to resolve session cwd ${cwdInput}`);
|
||||
return { cwd, id: destinationId };
|
||||
}
|
||||
|
||||
private async claimCreateDestination<T>(
|
||||
destination: { cwd: string; id: string },
|
||||
operation: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const key = `${destination.cwd}\0${destination.id}`;
|
||||
if (this.activeCreateDestinations.has(key)) throw new Error(`Session already exists: ${destination.id}`);
|
||||
this.activeCreateDestinations.add(key);
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
this.activeCreateDestinations.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
private async listDirectory(directory: string, cwd?: string): Promise<JsonlSessionMetadata[]> {
|
||||
if (!fileValue(await this.fileSystem.exists(directory), `Failed to check sessions directory ${directory}`)) {
|
||||
return [];
|
||||
|
|
|
|||
|
|
@ -476,8 +476,8 @@ export function createSessionRepoForkBehaviorConformance<TMetadata extends Sessi
|
|||
];
|
||||
}
|
||||
|
||||
/** Creates fork cases that require destination reservation or active-source snapshot coordination. */
|
||||
export function createSessionRepoForkCoordinationConformance<TMetadata extends SessionMetadata>(
|
||||
/** Creates fork cases that require destination reservation across create and fork. */
|
||||
export function createSessionRepoForkDestinationReservationConformance<TMetadata extends SessionMetadata>(
|
||||
backendFactory: () => Promise<Pick<SessionRepo<TMetadata>, "create" | "fork">>,
|
||||
onClose?: () => void | Promise<void>,
|
||||
): readonly ConformanceCase[] {
|
||||
|
|
@ -515,6 +515,16 @@ export function createSessionRepoForkCoordinationConformance<TMetadata extends S
|
|||
await source.close();
|
||||
},
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
/** Creates fork cases that require a snapshot boundary on an active source storage queue. */
|
||||
export function createSessionRepoForkSourceSnapshotConformance<TMetadata extends SessionMetadata>(
|
||||
backendFactory: () => Promise<Pick<SessionRepo<TMetadata>, "create" | "fork">>,
|
||||
onClose?: () => void | Promise<void>,
|
||||
): readonly ConformanceCase[] {
|
||||
const factory = prepareRepoCaseFactory(backendFactory, onClose);
|
||||
return [
|
||||
createCase(
|
||||
factory,
|
||||
"fork coordination",
|
||||
|
|
@ -555,6 +565,17 @@ export function createSessionRepoForkCoordinationConformance<TMetadata extends S
|
|||
];
|
||||
}
|
||||
|
||||
/** Creates every fork coordination case. */
|
||||
export function createSessionRepoForkCoordinationConformance<TMetadata extends SessionMetadata>(
|
||||
backendFactory: () => Promise<Pick<SessionRepo<TMetadata>, "create" | "fork">>,
|
||||
onClose?: () => void | Promise<void>,
|
||||
): readonly ConformanceCase[] {
|
||||
return [
|
||||
...createSessionRepoForkDestinationReservationConformance(backendFactory, onClose),
|
||||
...createSessionRepoForkSourceSnapshotConformance(backendFactory, onClose),
|
||||
];
|
||||
}
|
||||
|
||||
/** Creates every fork conformance case. */
|
||||
export function createSessionRepoForkConformance<TMetadata extends SessionMetadata>(
|
||||
backendFactory: () => Promise<Pick<SessionRepo<TMetadata>, "create" | "list" | "fork">>,
|
||||
|
|
|
|||
|
|
@ -21,6 +21,8 @@ export {
|
|||
createSessionRepoForkBehaviorConformance,
|
||||
createSessionRepoForkConformance,
|
||||
createSessionRepoForkCoordinationConformance,
|
||||
createSessionRepoForkDestinationReservationConformance,
|
||||
createSessionRepoForkSourceSnapshotConformance,
|
||||
createSessionRepoLifecycleConformance,
|
||||
createSessionRepoMessageConformance,
|
||||
createSessionRepoOwnershipConformance,
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { type JsonlSessionMetadata, JsonlSessionRepo } from "../../src/harness/s
|
|||
import {
|
||||
type ConformanceCase,
|
||||
createSessionRepoForkBehaviorConformance,
|
||||
createSessionRepoForkDestinationReservationConformance,
|
||||
createSessionRepoLifecycleConformance,
|
||||
createSessionRepoMessageConformance,
|
||||
} from "../../src/harness/session/testing/index.ts";
|
||||
|
|
@ -43,4 +44,7 @@ registerConformance("JsonlSessionRepo conformance", [
|
|||
...createSessionRepoLifecycleConformance<JsonlSessionMetadata>(createConformanceRepo, () => jsonlRepo.close()),
|
||||
...createSessionRepoMessageConformance<JsonlSessionMetadata>(createConformanceRepo, () => jsonlRepo.close()),
|
||||
...createSessionRepoForkBehaviorConformance<JsonlSessionMetadata>(createConformanceRepo, () => jsonlRepo.close()),
|
||||
...createSessionRepoForkDestinationReservationConformance<JsonlSessionMetadata>(createConformanceRepo, () =>
|
||||
jsonlRepo.close(),
|
||||
),
|
||||
]);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue