{"version":3,"file":"file-assignment-store.d.ts","sourceRoot":"","sources":["../../../src/core/assignment/file-assignment-store.ts"],"names":[],"mappings":"AAAA;;;;;;;;;;;;;;;;;;GAkBG;AAQH,OAAO,EACN,KAAK,oBAAoB,EACzB,KAAK,sBAAsB,EAC3B,KAAK,kBAAkB,EACvB,KAAK,eAAe,EACpB,KAAK,sBAAsB,EAE3B,MAAM,uBAAuB,CAAC;AAC/B,OAAO,EAAmB,KAAK,gBAAgB,EAAsB,MAAM,uBAAuB,CAAC;AAUnG,MAAM,WAAW,0BAA0B;IAC1C,IAAI,EAAE,MAAM,CAAC;IACb,OAAO,CAAC,EAAE,MAAM,CAAC;IACjB,WAAW,CAAC,EAAE,MAAM,CAAC;IACrB,WAAW,CAAC,EAAE,MAAM,CAAC;IACrB,gBAAgB,CAAC,EAAE,MAAM,CAAC;IAC1B,gBAAgB,CAAC,EAAE,MAAM,CAAC;CAC1B;AAED,wBAAgB,qBAAqB,IAAI,MAAM,CAI9C;AAED,qBAAa,mBAAoB,YAAW,eAAe;IAC1D,QAAQ,CAAC,OAAO,EAAE,MAAM,CAAC;IACzB,OAAO,CAAC,QAAQ,CAAC,IAAI,CAAS;IAC9B,OAAO,CAAC,QAAQ,CAAC,WAAW,CAAS;IACrC,OAAO,CAAC,QAAQ,CAAC,WAAW,CAAS;IACrC,OAAO,CAAC,QAAQ,CAAC,gBAAgB,CAAS;IAC1C,OAAO,CAAC,QAAQ,CAAC,gBAAgB,CAAS;IAE1C,YAAY,OAAO,EAAE,0BAA0B,EAO9C;IAED,OAAO,CAAC,OAAO;IAIf,OAAO,CAAC,kBAAkB;IAM1B,OAAO,CAAC,eAAe;IAMvB,OAAO,CAAC,mBAAmB;YAMb,WAAW;YAaX,OAAO;YAQP,eAAe;YAgEf,UAAU;YAgBV,mBAAmB;YAmBnB,YAAY;IAgBpB,IAAI,CAAC,YAAY,EAAE,MAAM,GAAG,OAAO,CAAC,oBAAoB,CAAC,CAgB9D;IAEK,eAAe,IAAI,OAAO,CAAC,MAAM,EAAE,CAAC,CAezC;IAEK,WAAW,IAAI,OAAO,CAAC;QAC5B,OAAO,EAAE,gBAAgB,EAAE,CAAC;QAC5B,OAAO,EAAE;YAAE,YAAY,EAAE,MAAM,CAAC;YAAC,UAAU,EAAE,MAAM,CAAA;SAAE,EAAE,CAAC;KACxD,CAAC,CAUD;IAEK,MAAM,CAAC,CAAC,EACb,SAAS,EAAE,MAAM,EACjB,QAAQ,EAAE,CAAC,KAAK,EAAE,sBAAsB,KAAK,kBAAkB,CAAC,CAAC,CAAC,GAChE,OAAO,CAAC,sBAAsB,CAAC,CAAC,CAAC,CAAC,CAiCpC;CACD;AAED,kFAAkF;AAClF,wBAAgB,yBAAyB,CAAC,IAAI,GAAE,MAAgC,GAAG,mBAAmB,CAErG","sourcesContent":["/**\n * File Assignment Store (2.11.0).\n *\n * Local durable implementation of the AssignmentStore port. One small record\n * file per assignment, schema-validated on load, written atomically\n * (unique temp + fsync + rename). The \"current assignment\" flag lives on the\n * record itself, so there is no separate mutable pointer file to drift.\n *\n * Cross-process model:\n *   - All mutations for a mission serialize on a per-mission proper-lockfile\n *     lock. Two processes racing `assign M→A` vs `assign M→B` therefore have\n *     exactly one winner; the loser sees the winner's current assignment and\n *     returns a structured conflict instead of writing a second current.\n *   - Unrelated missions lock different paths, so they never globally\n *     serialize.\n *   - Record writes are atomic per file. The service transitions old→non-current\n *     BEFORE writing new→current, so a crash can leave zero current records for\n *     a mission, but never two.\n */\n\nimport { randomUUID } from \"node:crypto\";\nimport * as fsp from \"node:fs/promises\";\nimport * as os from \"node:os\";\nimport * as path from \"node:path\";\nimport lockfile from \"proper-lockfile\";\nimport { isSafeMissionId } from \"../mission-domain/durable-store.js\";\nimport {\n\ttype AssignmentLoadResult,\n\ttype AssignmentMutateResult,\n\ttype AssignmentMutation,\n\ttype AssignmentStore,\n\ttype MissionAssignmentIndex,\n\tparseAssignmentRecord,\n} from \"./assignment-store.js\";\nimport { AssignmentError, type AssignmentRecord, isSafeAssignmentId } from \"./assignment-types.js\";\n\nconst RECORD_SUFFIX = \".assignment.json\";\nconst ATOMIC_SUFFIX = \".tmp\";\n\nconst DEFAULT_LOCK_STALE_MS = 30_000;\nconst DEFAULT_LOCK_RETRIES = 8;\nconst DEFAULT_LOCK_MIN_TIMEOUT_MS = 20;\nconst DEFAULT_LOCK_MAX_TIMEOUT_MS = 250;\n\nexport interface FileAssignmentStoreOptions {\n\troot: string;\n\tstoreId?: string;\n\tlockStaleMs?: number;\n\tlockRetries?: number;\n\tlockMinTimeoutMs?: number;\n\tlockMaxTimeoutMs?: number;\n}\n\nexport function defaultAssignmentRoot(): string {\n\tconst env = process.env.JENSEN_ASSIGNMENT_REGISTRY_DIR;\n\tif (env?.trim()) return env.trim();\n\treturn path.join(os.homedir(), \".jensen\", \"agent\", \"assignment-registry\");\n}\n\nexport class FileAssignmentStore implements AssignmentStore {\n\treadonly storeId: string;\n\tprivate readonly root: string;\n\tprivate readonly lockStaleMs: number;\n\tprivate readonly lockRetries: number;\n\tprivate readonly lockMinTimeoutMs: number;\n\tprivate readonly lockMaxTimeoutMs: number;\n\n\tconstructor(options: FileAssignmentStoreOptions) {\n\t\tthis.root = path.resolve(options.root);\n\t\tthis.storeId = options.storeId ?? \"file\";\n\t\tthis.lockStaleMs = options.lockStaleMs ?? DEFAULT_LOCK_STALE_MS;\n\t\tthis.lockRetries = options.lockRetries ?? DEFAULT_LOCK_RETRIES;\n\t\tthis.lockMinTimeoutMs = options.lockMinTimeoutMs ?? DEFAULT_LOCK_MIN_TIMEOUT_MS;\n\t\tthis.lockMaxTimeoutMs = options.lockMaxTimeoutMs ?? DEFAULT_LOCK_MAX_TIMEOUT_MS;\n\t}\n\n\tprivate resolve(assignmentId: string): string {\n\t\treturn path.join(this.root, `${assignmentId}${RECORD_SUFFIX}`);\n\t}\n\n\tprivate assertAssignmentId(assignmentId: string): void {\n\t\tif (!isSafeAssignmentId(assignmentId)) {\n\t\t\tthrow new AssignmentError(\"ASSIGNMENT_CORRUPT\", `Unsafe assignment id: ${assignmentId}`, { assignmentId });\n\t\t}\n\t}\n\n\tprivate assertMissionId(missionId: string): void {\n\t\tif (!isSafeMissionId(missionId)) {\n\t\t\tthrow new AssignmentError(\"MISSION_NOT_ASSIGNABLE\", `Unsafe mission id: ${missionId}`, { missionId });\n\t\t}\n\t}\n\n\tprivate assertRecordMission(record: AssignmentRecord, missionId: string): void {\n\t\tif (record.missionId !== missionId) {\n\t\t\tthrow new Error(`Assignment ${record.assignmentId} belongs to mission ${record.missionId}, not ${missionId}`);\n\t\t}\n\t}\n\n\tprivate async writeAtomic(target: string, content: string): Promise<void> {\n\t\tawait fsp.mkdir(this.root, { recursive: true });\n\t\tconst tmp = `${target}.${randomUUID()}${ATOMIC_SUFFIX}`;\n\t\tawait fsp.writeFile(tmp, content, \"utf8\");\n\t\tconst fh = await fsp.open(tmp, \"r\");\n\t\ttry {\n\t\t\tawait fh.sync();\n\t\t} finally {\n\t\t\tawait fh.close();\n\t\t}\n\t\tawait fsp.rename(tmp, target);\n\t}\n\n\tprivate async readRaw(assignmentId: string): Promise<string | undefined> {\n\t\ttry {\n\t\t\treturn await fsp.readFile(this.resolve(assignmentId), \"utf8\");\n\t\t} catch {\n\t\t\treturn undefined;\n\t\t}\n\t}\n\n\tprivate async withMissionLock<T>(missionId: string, fn: () => Promise<T>): Promise<T> {\n\t\tthis.assertMissionId(missionId);\n\t\tawait fsp.mkdir(this.root, { recursive: true });\n\t\tconst target = path.join(this.root, missionId);\n\n\t\tlet compromised = false;\n\t\tlet release: (() => Promise<void>) | undefined;\n\t\ttry {\n\t\t\trelease = await lockfile.lock(target, {\n\t\t\t\trealpath: false,\n\t\t\t\tstale: this.lockStaleMs,\n\t\t\t\tretries: {\n\t\t\t\t\tretries: this.lockRetries,\n\t\t\t\t\tfactor: 2,\n\t\t\t\t\tminTimeout: this.lockMinTimeoutMs,\n\t\t\t\t\tmaxTimeout: this.lockMaxTimeoutMs,\n\t\t\t\t\trandomize: true,\n\t\t\t\t},\n\t\t\t\tonCompromised: () => {\n\t\t\t\t\tcompromised = true;\n\t\t\t\t},\n\t\t\t});\n\t\t} catch (error) {\n\t\t\tconst code =\n\t\t\t\ttypeof error === \"object\" && error !== null && \"code\" in error\n\t\t\t\t\t? (error as { code?: unknown }).code\n\t\t\t\t\t: undefined;\n\t\t\tif (code === \"ELOCKED\") {\n\t\t\t\tthrow new AssignmentError(\n\t\t\t\t\t\"ASSIGNMENT_LOCK_TIMEOUT\",\n\t\t\t\t\t`Timed out acquiring assignment mutation lock for mission ${missionId}`,\n\t\t\t\t\t{ missionId },\n\t\t\t\t);\n\t\t\t}\n\t\t\tif (code === \"ENOTEMPTY\" || code === \"ENOTDIR\") {\n\t\t\t\tthrow new AssignmentError(\n\t\t\t\t\t\"ASSIGNMENT_CORRUPT\",\n\t\t\t\t\t`Corrupt assignment mutation lock metadata for mission ${missionId}`,\n\t\t\t\t\t{ missionId },\n\t\t\t\t);\n\t\t\t}\n\t\t\tthrow error;\n\t\t}\n\n\t\ttry {\n\t\t\tif (compromised) {\n\t\t\t\tthrow new AssignmentError(\n\t\t\t\t\t\"ASSIGNMENT_CORRUPT\",\n\t\t\t\t\t`Assignment mutation lock for mission ${missionId} was compromised`,\n\t\t\t\t\t{ missionId },\n\t\t\t\t);\n\t\t\t}\n\t\t\treturn await fn();\n\t\t} finally {\n\t\t\tif (release) {\n\t\t\t\ttry {\n\t\t\t\t\tawait release();\n\t\t\t\t} catch {\n\t\t\t\t\t// Best-effort release; a successful mutation must never become an error.\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t}\n\n\tprivate async readParsed(\n\t\tassignmentId: string,\n\t): Promise<{ status: \"ok\"; record: AssignmentRecord } | { status: \"missing\" } | { status: \"corrupt\" }> {\n\t\tconst raw = await this.readRaw(assignmentId);\n\t\tif (raw === undefined) return { status: \"missing\" };\n\t\tlet parsed: unknown;\n\t\ttry {\n\t\t\tparsed = JSON.parse(raw);\n\t\t} catch {\n\t\t\treturn { status: \"corrupt\" };\n\t\t}\n\t\tconst result = parseAssignmentRecord(parsed);\n\t\tif (!result.ok) return { status: \"corrupt\" };\n\t\treturn { status: \"ok\", record: result.record };\n\t}\n\n\tprivate async readIndexForMission(\n\t\tmissionId: string,\n\t): Promise<{ status: \"ok\"; index: MissionAssignmentIndex } | { status: \"corrupt\"; diagnostic: string }> {\n\t\tconst ids = await this.listAssignments();\n\t\tconst records: AssignmentRecord[] = [];\n\t\tfor (const id of ids) {\n\t\t\tconst loaded = await this.readParsed(id);\n\t\t\tif (loaded.status === \"corrupt\") {\n\t\t\t\treturn { status: \"corrupt\", diagnostic: `assignment ${id} is corrupt` };\n\t\t\t}\n\t\t\tif (loaded.status === \"ok\" && loaded.record.missionId === missionId) {\n\t\t\t\trecords.push(loaded.record);\n\t\t\t}\n\t\t}\n\t\trecords.sort((a, b) => a.createdAtMs - b.createdAtMs || (a.assignmentId < b.assignmentId ? -1 : 1));\n\t\tconst current = records.find((record) => record.current);\n\t\treturn { status: \"ok\", index: { missionId, records, current } };\n\t}\n\n\tprivate async writeRecords(missionId: string, records: AssignmentRecord[]): Promise<void> {\n\t\tfor (const record of records) {\n\t\t\tthis.assertRecordMission(record, missionId);\n\t\t\tthis.assertAssignmentId(record.assignmentId);\n\t\t\tconst validation = parseAssignmentRecord(record);\n\t\t\tif (!validation.ok) {\n\t\t\t\tthrow new Error(`Mutation produced an invalid record for ${record.assignmentId}: ${validation.diagnostic}`);\n\t\t\t}\n\t\t\tawait this.writeAtomic(this.resolve(record.assignmentId), JSON.stringify(record, null, 2));\n\t\t}\n\t}\n\n\t// =========================================================================\n\t// AssignmentStore\n\t// =========================================================================\n\n\tasync load(assignmentId: string): Promise<AssignmentLoadResult> {\n\t\tthis.assertAssignmentId(assignmentId);\n\t\tconst raw = await this.readRaw(assignmentId);\n\t\tif (raw === undefined) return { status: \"missing\" };\n\n\t\tlet parsed: unknown;\n\t\ttry {\n\t\t\tparsed = JSON.parse(raw);\n\t\t} catch {\n\t\t\treturn { status: \"corrupt\", assignmentId, diagnostic: \"record is not valid JSON\" };\n\t\t}\n\t\tconst result = parseAssignmentRecord(parsed);\n\t\tif (!result.ok) {\n\t\t\treturn { status: \"corrupt\", assignmentId, diagnostic: result.diagnostic };\n\t\t}\n\t\treturn { status: \"ok\", record: result.record };\n\t}\n\n\tasync listAssignments(): Promise<string[]> {\n\t\tlet entries: string[];\n\t\ttry {\n\t\t\tentries = await fsp.readdir(this.root);\n\t\t} catch {\n\t\t\treturn [];\n\t\t}\n\t\tconst ids: string[] = [];\n\t\tfor (const entry of entries) {\n\t\t\tif (entry.endsWith(ATOMIC_SUFFIX)) continue;\n\t\t\tif (!entry.endsWith(RECORD_SUFFIX)) continue;\n\t\t\tconst id = entry.slice(0, -RECORD_SUFFIX.length);\n\t\t\tif (isSafeAssignmentId(id)) ids.push(id);\n\t\t}\n\t\treturn ids.sort();\n\t}\n\n\tasync listRecords(): Promise<{\n\t\trecords: AssignmentRecord[];\n\t\tcorrupt: { assignmentId: string; diagnostic: string }[];\n\t}> {\n\t\tconst ids = await this.listAssignments();\n\t\tconst records: AssignmentRecord[] = [];\n\t\tconst corrupt: { assignmentId: string; diagnostic: string }[] = [];\n\t\tfor (const id of ids) {\n\t\t\tconst loaded = await this.load(id);\n\t\t\tif (loaded.status === \"ok\") records.push(loaded.record);\n\t\t\telse if (loaded.status === \"corrupt\") corrupt.push({ assignmentId: id, diagnostic: loaded.diagnostic });\n\t\t}\n\t\treturn { records, corrupt };\n\t}\n\n\tasync mutate<T>(\n\t\tmissionId: string,\n\t\tmutation: (index: MissionAssignmentIndex) => AssignmentMutation<T>,\n\t): Promise<AssignmentMutateResult<T>> {\n\t\tthis.assertMissionId(missionId);\n\n\t\treturn this.withMissionLock(missionId, async () => {\n\t\t\tconst read = await this.readIndexForMission(missionId);\n\t\t\tif (read.status === \"corrupt\") {\n\t\t\t\treturn { status: \"corrupt\", diagnostic: read.diagnostic };\n\t\t\t}\n\n\t\t\tconst output = mutation(read.index);\n\t\t\tif (output.kind === \"noop\") return { status: \"ok\", value: output.value };\n\n\t\t\t// History is never deleted: every previously-persisted record must still\n\t\t\t// be present in the replacement set. New records may be appended.\n\t\t\tconst existingIds = new Set(read.index.records.map((r) => r.assignmentId));\n\t\t\tfor (const record of output.records) {\n\t\t\t\texistingIds.delete(record.assignmentId);\n\t\t\t}\n\t\t\tif (existingIds.size > 0) {\n\t\t\t\tthrow new Error(\n\t\t\t\t\t`Mutation for mission ${missionId} dropped historical assignments: ${[...existingIds].join(\", \")}`,\n\t\t\t\t);\n\t\t\t}\n\n\t\t\t// Defense in depth: exactly one current assignment for the mission.\n\t\t\tconst currentRecords = output.records.filter((record) => record.current);\n\t\t\tif (currentRecords.length > 1) {\n\t\t\t\tthrow new Error(`Mutation for mission ${missionId} produced ${currentRecords.length} current assignments`);\n\t\t\t}\n\n\t\t\tawait this.writeRecords(missionId, output.records);\n\t\t\treturn { status: \"ok\", value: output.value };\n\t\t});\n\t}\n}\n\n/** Convenience factory using the default Jensen assignment registry directory. */\nexport function createFileAssignmentStore(root: string = defaultAssignmentRoot()): FileAssignmentStore {\n\treturn new FileAssignmentStore({ root });\n}\n"]}