Refactor runtime around isolated bot instances
This commit is contained in:
zenord
@@ -1,53 +1,72 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
|
||||
export interface ChatState {
|
||||
selectedRole: string;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
export interface StoreIdentity {
|
||||
botId: string;
|
||||
platform: string;
|
||||
}
|
||||
|
||||
export interface SessionBinding {
|
||||
chatKey: string;
|
||||
roleId: string;
|
||||
backendId: string;
|
||||
agentId: string;
|
||||
nativeSessionId: string;
|
||||
workspace: string;
|
||||
roleFingerprint: string;
|
||||
botFingerprint: string;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
}
|
||||
|
||||
interface StoreData {
|
||||
version: 1;
|
||||
chats: Record<string, ChatState>;
|
||||
version: 2;
|
||||
botId: string;
|
||||
platform: string;
|
||||
bindings: Record<string, SessionBinding>;
|
||||
}
|
||||
|
||||
const EMPTY: StoreData = { version: 1, chats: {}, bindings: {} };
|
||||
|
||||
export class DurableSessionStore {
|
||||
private data: StoreData = structuredClone(EMPTY);
|
||||
private data: StoreData;
|
||||
private queue: Promise<void> = Promise.resolve();
|
||||
private lockFd?: fs.promises.FileHandle;
|
||||
private closed = false;
|
||||
|
||||
constructor(readonly file: string) {}
|
||||
constructor(readonly file: string, readonly identity: StoreIdentity) {
|
||||
this.data = emptyData(identity);
|
||||
}
|
||||
|
||||
async open(): Promise<void> {
|
||||
await fs.promises.mkdir(path.dirname(this.file), { recursive: true });
|
||||
const directory = path.dirname(this.file);
|
||||
await fs.promises.mkdir(directory, { recursive: true, mode: 0o700 });
|
||||
const directoryStat = await fs.promises.lstat(directory);
|
||||
if (!directoryStat.isDirectory() || directoryStat.isSymbolicLink()) throw new Error(`Refusing unsafe state directory: ${directory}`);
|
||||
const directoryMode = directoryStat.mode & 0o777;
|
||||
if (directoryMode !== 0o700) throw new Error(`State directory mode must be 0700, got 0${directoryMode.toString(8)}`);
|
||||
if (typeof process.getuid === "function" && directoryStat.uid !== process.getuid()) throw new Error("State directory is not owned by the current user");
|
||||
const lockFile = `${this.file}.lock`;
|
||||
try {
|
||||
this.lockFd = await acquireLock(lockFile);
|
||||
await this.lockFd.writeFile(`${process.pid}\n`);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "EEXIST") throw new Error(`Session state is locked by another gori-agent instance: ${lockFile}`);
|
||||
if ((error as NodeJS.ErrnoException).code !== "EEXIST" || !await removeStaleLock(lockFile)) {
|
||||
if ((error as NodeJS.ErrnoException).code === "EEXIST") throw new Error(`Session state is locked by another gori-agent instance: ${lockFile}`);
|
||||
throw error;
|
||||
}
|
||||
this.lockFd = await acquireLock(lockFile).catch((retryError) => {
|
||||
if ((retryError as NodeJS.ErrnoException).code === "EEXIST") throw new Error(`Session state is locked by another gori-agent instance: ${lockFile}`);
|
||||
throw retryError;
|
||||
});
|
||||
}
|
||||
try {
|
||||
await this.lockFd.writeFile(`${process.pid}\n`);
|
||||
await this.lockFd.sync();
|
||||
} catch (error) {
|
||||
await this.releaseLock();
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
const raw = await fs.promises.readFile(this.file, "utf8");
|
||||
const parsed = JSON.parse(raw) as StoreData;
|
||||
if (parsed.version !== 1 || !parsed.chats || !parsed.bindings) throw new Error("unsupported or malformed state data");
|
||||
const parsed = parseStoreData(JSON.parse(raw) as unknown);
|
||||
if (parsed.botId !== this.identity.botId || parsed.platform !== this.identity.platform) {
|
||||
throw new Error(`state identity mismatch: expected bot '${this.identity.botId}' platform '${this.identity.platform}'`);
|
||||
}
|
||||
this.data = parsed;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
@@ -57,44 +76,32 @@ export class DurableSessionStore {
|
||||
}
|
||||
}
|
||||
|
||||
getSelectedRole(chatKey: string, defaultRole: string): string {
|
||||
return this.data.chats[chatKey]?.selectedRole || defaultRole;
|
||||
}
|
||||
|
||||
async setSelectedRole(chatKey: string, roleId: string): Promise<void> {
|
||||
const now = Date.now();
|
||||
const current = this.data.chats[chatKey];
|
||||
this.data.chats[chatKey] = { selectedRole: roleId, createdAt: current?.createdAt || now, updatedAt: now };
|
||||
await this.persist();
|
||||
}
|
||||
|
||||
getBinding(chatKey: string, roleId: string): SessionBinding | undefined {
|
||||
const value = this.data.bindings[bindingKey(chatKey, roleId)];
|
||||
getBinding(chatKey: string): SessionBinding | undefined {
|
||||
const value = this.data.bindings[chatKey];
|
||||
return value ? structuredClone(value) : undefined;
|
||||
}
|
||||
|
||||
async setBinding(binding: SessionBinding): Promise<void> {
|
||||
this.data.bindings[bindingKey(binding.chatKey, binding.roleId)] = structuredClone(binding);
|
||||
this.data.bindings[binding.chatKey] = structuredClone(binding);
|
||||
await this.persist();
|
||||
}
|
||||
|
||||
async touchBinding(chatKey: string, roleId: string): Promise<void> {
|
||||
const binding = this.data.bindings[bindingKey(chatKey, roleId)];
|
||||
async touchBinding(chatKey: string): Promise<void> {
|
||||
const binding = this.data.bindings[chatKey];
|
||||
if (!binding) return;
|
||||
binding.updatedAt = Date.now();
|
||||
await this.persist();
|
||||
}
|
||||
|
||||
async deleteBinding(chatKey: string, roleId: string): Promise<SessionBinding | undefined> {
|
||||
const key = bindingKey(chatKey, roleId);
|
||||
const existing = this.data.bindings[key];
|
||||
delete this.data.bindings[key];
|
||||
async deleteBinding(chatKey: string): Promise<SessionBinding | undefined> {
|
||||
const existing = this.data.bindings[chatKey];
|
||||
delete this.data.bindings[chatKey];
|
||||
if (existing) await this.persist();
|
||||
return existing;
|
||||
return existing ? structuredClone(existing) : undefined;
|
||||
}
|
||||
|
||||
stats(): { chats: number; bindings: number } {
|
||||
return { chats: Object.keys(this.data.chats).length, bindings: Object.keys(this.data.bindings).length };
|
||||
stats(): { bindings: number } {
|
||||
return { bindings: Object.keys(this.data.bindings).length };
|
||||
}
|
||||
|
||||
async flush(): Promise<void> { await this.queue; }
|
||||
@@ -102,15 +109,15 @@ export class DurableSessionStore {
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
await this.flush();
|
||||
await this.releaseLock();
|
||||
try { await this.flush(); } finally { await this.releaseLock(); }
|
||||
}
|
||||
|
||||
private persist(): Promise<void> {
|
||||
if (this.closed) return Promise.reject(new Error("Session store is closed"));
|
||||
const snapshot = JSON.stringify(this.data, null, 2) + "\n";
|
||||
this.queue = this.queue.then(() => atomicWrite(this.file, snapshot));
|
||||
return this.queue;
|
||||
const operation = this.queue.then(() => atomicWrite(this.file, snapshot));
|
||||
this.queue = operation.catch(() => undefined);
|
||||
return operation;
|
||||
}
|
||||
|
||||
private async releaseLock(): Promise<void> {
|
||||
@@ -124,26 +131,73 @@ export class DurableSessionStore {
|
||||
}
|
||||
|
||||
export function chatKeyFor(platform: string, chatId: string): string { return `${platform}:${chatId}`; }
|
||||
export function bindingKey(chatKey: string, roleId: string): string { return `${chatKey}\u0000${roleId}`; }
|
||||
|
||||
function emptyData(identity: StoreIdentity): StoreData {
|
||||
return { version: 2, botId: identity.botId, platform: identity.platform, bindings: {} };
|
||||
}
|
||||
|
||||
function parseStoreData(value: unknown): StoreData {
|
||||
if (!isRecord(value) || value.version !== 2 || typeof value.botId !== "string" || typeof value.platform !== "string" || !isRecord(value.bindings)) {
|
||||
throw new Error("unsupported state version; Config v3 requires state v2 in a new GORI_AGENT_HOME");
|
||||
}
|
||||
for (const [chatKey, binding] of Object.entries(value.bindings)) {
|
||||
if (!isRecord(binding)
|
||||
|| binding.chatKey !== chatKey
|
||||
|| typeof binding.agentId !== "string"
|
||||
|| typeof binding.nativeSessionId !== "string"
|
||||
|| typeof binding.workspace !== "string"
|
||||
|| typeof binding.botFingerprint !== "string"
|
||||
|| typeof binding.createdAt !== "number"
|
||||
|| typeof binding.updatedAt !== "number") {
|
||||
throw new Error(`invalid state v2 binding '${chatKey}'`);
|
||||
}
|
||||
}
|
||||
return value as unknown as StoreData;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
async function acquireLock(lockFile: string): Promise<fs.promises.FileHandle> {
|
||||
return fs.promises.open(lockFile, "wx", 0o600);
|
||||
}
|
||||
|
||||
async function removeStaleLock(lockFile: string): Promise<boolean> {
|
||||
let handle: fs.promises.FileHandle | undefined;
|
||||
try {
|
||||
handle = await fs.promises.open(lockFile, "r");
|
||||
const pid = Number((await handle.readFile("utf8")).trim());
|
||||
if (!Number.isSafeInteger(pid) || pid <= 1 || processExists(pid)) return false;
|
||||
const opened = await handle.stat();
|
||||
const current = await fs.promises.lstat(lockFile);
|
||||
if (opened.dev !== current.dev || opened.ino !== current.ino) return false;
|
||||
await fs.promises.unlink(lockFile);
|
||||
return true;
|
||||
} catch { return false; }
|
||||
finally { await handle?.close().catch(() => undefined); }
|
||||
}
|
||||
|
||||
function processExists(pid: number): boolean {
|
||||
try { process.kill(pid, 0); return true; }
|
||||
catch (error) { return (error as NodeJS.ErrnoException).code === "EPERM"; }
|
||||
}
|
||||
|
||||
async function atomicWrite(file: string, content: string): Promise<void> {
|
||||
const temp = `${file}.${process.pid}.${Date.now()}.tmp`;
|
||||
const handle = await fs.promises.open(temp, "wx", 0o600);
|
||||
let handle: fs.promises.FileHandle | undefined;
|
||||
try {
|
||||
handle = await fs.promises.open(temp, "wx", 0o600);
|
||||
await handle.writeFile(content, "utf8");
|
||||
await handle.sync();
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
try {
|
||||
handle = undefined;
|
||||
await fs.promises.rename(temp, file);
|
||||
await fs.promises.chmod(file, 0o600);
|
||||
const dir = await fs.promises.open(path.dirname(file), "r");
|
||||
try { await dir.sync(); } finally { await dir.close(); }
|
||||
} catch (error) {
|
||||
if (handle) await handle.close().catch(() => undefined);
|
||||
await fs.promises.unlink(temp).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user