diff --git a/apps/site/src/generated/repo-catalog.ts b/apps/site/src/generated/repo-catalog.ts index a426454b..f5c7e6ae 100644 --- a/apps/site/src/generated/repo-catalog.ts +++ b/apps/site/src/generated/repo-catalog.ts @@ -1835,6 +1835,42 @@ export const repoCatalog = { "headless" ] }, + { + "path": "labs/extensions/causal-patch-inbox", + "name": "@interactive-os/json-document-causal-patch-inbox", + "status": "lab-extension", + "private": true, + "publishable": false, + "version": "0.1.0", + "description": "Lab causal JSON Patch inbox for json-document documents.", + "license": "MIT", + "summary": "Lab module for admitting dependency-declared JSON Patch envelopes, holding\nout-of-order work, and applying each ready envelope to one `JSONDocument`.", + "guidance": { + "useFor": "queue and deduplicate dependency-declared envelopes, applying each as one core JSON Patch batch", + "notFor": "CRDT/OT convergence, transport, persistence, ready-time rebase, or conflict resolution" + }, + "publicExports": [ + "CausalPatchEnvelope", + "CausalPatchInbox", + "CausalPatchInboxSnapshot", + "CausalPatchIngestError", + "CausalPatchIngestErrorCode", + "CausalPatchIngestOk", + "CausalPatchIngestResult", + "FailedCausalPatch", + "FaultedCausalPatch", + "QueuedCausalPatch", + "createCausalPatchInbox" + ], + "publicExportCount": 11, + "keywords": [ + "@interactive-os/json-document", + "causal", + "headless", + "inbox", + "patch" + ] + }, { "path": "labs/extensions/change-case", "name": "@interactive-os/json-document-change-case", @@ -3315,7 +3351,7 @@ export const repoCatalog = { "totals": { "packages": 20, "officialExtensions": 19, - "labExtensions": 40, + "labExtensions": 41, "apps": 12 } } as const; diff --git a/docs/generated/extensions-catalog.md b/docs/generated/extensions-catalog.md index 3def2348..144d3ba7 100644 --- a/docs/generated/extensions-catalog.md +++ b/docs/generated/extensions-catalog.md @@ -28,7 +28,7 @@ Official extensions: 19 | `@interactive-os/json-document-search-replace` | 23 | find and replace text across document string fields | regex engines, rendered text extraction, or search UI | Official headless search and replace extension for text fields in `@interactive-os/json-document` documents. | | `@interactive-os/json-document-snippets` | 12 | insert reusable JSON payloads with schema-safe paste checks | slash palette UI or snippet storage | Official headless snippet insertion extension for `@interactive-os/json-document` documents. | -Lab extensions: 40 +Lab extensions: 41 Lab extensions are private candidates. They are listed to show product pressure, not as shipped packages. @@ -39,6 +39,7 @@ Lab extensions are private candidates. They are listed to show product pressure, | `@interactive-os/json-document-batch-update` | lab-only | 10 | set a field across a list of selected item pointers to a constant or computed value | selecting which items to edit, or JSONPath query-driven replacement | Lab batch-update extension for `@interactive-os/json-document` documents. | | `@interactive-os/json-document-bookmarks` | lab-only | 8 | keep named JSON Pointer locations stable across edits | browser bookmarks or route state | Headless bookmark tracking extension for `@interactive-os/json-document` documents. | | `@interactive-os/json-document-calculated-fields` | lab-only | 12 | sync host-computed derived JSON fields | formula languages or dependency runtimes | Lab calculated field extension for `@interactive-os/json-document` documents. | +| `@interactive-os/json-document-causal-patch-inbox` | lab-only | 11 | queue and deduplicate dependency-declared envelopes, applying each as one core JSON Patch batch | CRDT/OT convergence, transport, persistence, ready-time rebase, or conflict resolution | Lab module for admitting dependency-declared JSON Patch envelopes, holding out-of-order work, and applying each ready envelope to one `JSONDocument`. | | `@interactive-os/json-document-change-case` | lab-only | 9 | apply case/whitespace transforms (upper, lower, trim, title) to a string field | locale-aware casing, rich text formatting toolbars, or find/replace | Lab change-case extension for `@interactive-os/json-document` documents. | | `@interactive-os/json-document-checkpoints` | lab-only | 13 | name and restore document snapshots | durable version graphs or cloud backup | Lab checkpoint extension for `@interactive-os/json-document` documents. | | `@interactive-os/json-document-clear-contents` | lab-only | 10 | reset selected fields to schema-derived empty values, keeping structure | structural delete, caller-supplied bulk replace, or enum/object default policy | Lab clear-contents extension for `@interactive-os/json-document` documents. | diff --git a/docs/generated/repo-catalog.json b/docs/generated/repo-catalog.json index d9d78eb6..95bd2449 100644 --- a/docs/generated/repo-catalog.json +++ b/docs/generated/repo-catalog.json @@ -1834,6 +1834,42 @@ "headless" ] }, + { + "path": "labs/extensions/causal-patch-inbox", + "name": "@interactive-os/json-document-causal-patch-inbox", + "status": "lab-extension", + "private": true, + "publishable": false, + "version": "0.1.0", + "description": "Lab causal JSON Patch inbox for json-document documents.", + "license": "MIT", + "summary": "Lab module for admitting dependency-declared JSON Patch envelopes, holding\nout-of-order work, and applying each ready envelope to one `JSONDocument`.", + "guidance": { + "useFor": "queue and deduplicate dependency-declared envelopes, applying each as one core JSON Patch batch", + "notFor": "CRDT/OT convergence, transport, persistence, ready-time rebase, or conflict resolution" + }, + "publicExports": [ + "CausalPatchEnvelope", + "CausalPatchInbox", + "CausalPatchInboxSnapshot", + "CausalPatchIngestError", + "CausalPatchIngestErrorCode", + "CausalPatchIngestOk", + "CausalPatchIngestResult", + "FailedCausalPatch", + "FaultedCausalPatch", + "QueuedCausalPatch", + "createCausalPatchInbox" + ], + "publicExportCount": 11, + "keywords": [ + "@interactive-os/json-document", + "causal", + "headless", + "inbox", + "patch" + ] + }, { "path": "labs/extensions/change-case", "name": "@interactive-os/json-document-change-case", @@ -3314,7 +3350,7 @@ "totals": { "packages": 20, "officialExtensions": 19, - "labExtensions": 40, + "labExtensions": 41, "apps": 12 } } diff --git a/labs/extensions/causal-patch-inbox/README.md b/labs/extensions/causal-patch-inbox/README.md new file mode 100644 index 00000000..7a8afc11 --- /dev/null +++ b/labs/extensions/causal-patch-inbox/README.md @@ -0,0 +1,151 @@ +# @interactive-os/json-document-causal-patch-inbox + +Lab module for admitting dependency-declared JSON Patch envelopes, holding +out-of-order work, and applying each ready envelope to one `JSONDocument`. + +```ts +const inbox = createCausalPatchInbox(doc); + +inbox.ingest({ + id: "change-2", + dependsOn: ["change-1"], + operations: [ + { op: "test", path: "/title", value: "Ready" }, + { op: "replace", path: "/title", value: "Reviewed" }, + ], +}); +// queued: change-2 is waiting for change-1 + +const result = inbox.ingest({ + id: "change-1", + dependsOn: [], + operations: [ + { op: "test", path: "/status", value: "draft" }, + { op: "replace", path: "/status", value: "ready" }, + ], +}); + +if (result.ok) { + result.applied; // ["change-1", "change-2"] + result.pending; // ids from this input that remain queued +} +``` + +The envelope operations are already-materialized JSON Patch. They can be raw +protocol output or guarded operations produced by another planner. This module +owns causal admission and scheduling, not target recovery or merge policy. + +An `ingest` batch is preflighted before any new envelope is admitted. An exact +duplicate is a no-op; the same id with different dependencies or operations is +`duplicate_mismatch`. Dependency order has set semantics, while operation and +array order remain significant. A known graph cycle rejects the closing batch. + +Ready envelopes currently known to one drain are applied by locale-independent +id order. Each envelope is one atomic `doc.patch`; causal frontier state moves +only after that patch succeeds. A batch is not one transaction: earlier +envelopes remain applied when a later envelope fails. Patch failure globally +blocks this first tracer, preserving the failed envelope and its descendants +for inspection. Recovery, skipping, and conflict resolution need a later +policy rather than an implicit retry. + +`current().frontier` is the maximal set in the declared dependency DAG. It is +not an actor clock or version vector. `queued[].missing` contains dependencies +that have not been applied by this inbox. Snapshot status `active` describes a +usable inbox, including one that has pending dependencies; it does not mean +that a ready envelope currently exists. `blocked` preserves a structured patch +failure, while `faulted` preserves an unexpected thrown execution failure that +has no `JSONResult`. Full queue snapshots are produced only by `current()`. +Successful `ingest` results report only this call's `applied`, `pending`, and +`duplicates` ids. + +## Scope + +- Defensively validate and copy one envelope or a batch before admission. +- Queue envelopes whose declared dependencies have not been applied. +- Distinguish exact duplicate delivery from an id/payload mismatch. +- Reject empty ids, invalid dependency ids, self/duplicate dependencies, + accessor/non-enumerable data, non-JSON payloads, and cycles closed by new + dependency edges. +- Choose the smallest currently ready id with a private min-heap. +- Wake dependents through a reverse dependency index rather than rescanning the + whole queue after every successful patch. +- Apply one envelope through one public `doc.patch` call and advance causal + state only after success. +- Permit an empty projection patch to advance causal state. +- Stop with structured `patch_failed` state for non-strict documents; preserve + the core `JSONDocumentError` throw policy for strict documents. +- Move to terminal `faulted` state when patch execution throws without changing + the projection and without supplying a structured `JSONDocumentError`. +- Reject reentrant ingestion as `busy` while a document patch is publishing. +- Detect direct document mutation and move the inbox to `diverged` rather than + inferring a causal envelope from an observed patch. +- Dispose the document subscription explicitly. + +## Performance + +- Envelope validation, defensive copying, and exact comparison are linear in + envelope payload size; dependency normalization also sorts the dependency + ids. +- Known-id lookup is average `O(1)`. Ready selection is `O(log V)`, and each + reverse dependency edge is released once. +- Cycle preflight first consults the reverse dependency index. A new id with no + incoming pending edge cannot close a cycle and needs no graph walk. Other new + edges walk only reachable dependency paths with an iterative stack; the + implementation neither rebuilds unrelated graph state nor overflows the + JavaScript call stack on deep chains. +- `current()` sorts and defensively copies frontier and queue observations, so + its cost grows with the exposed state. `ingest()` does not call `current()`; + callers pay that observation cost only when they request a snapshot. +- The module does not clone, diff, or stringify the whole document. Core patch + application and schema/refinement validation can still dominate runtime. +- Exact dedup retains accepted envelope payloads, and the causal ledger has no + compaction in this lab. Memory therefore grows with accepted work. + +## Non-goals + +- No CRDT, OT, LWW policy, convergence claim, total-order protocol, sequencer, + watermark, actor clock, tombstone, or replicated undo. +- No arrival-order-independent result across separate `ingest` calls. Only the + ready set known to the current drain is ordered deterministically. +- No local submit/envelope generation, acknowledgement, transport, reconnect, + offline queue, persistence, checkpoint, or compaction format. +- No generic protocol or materialization Adapter interface yet. +- No ready-time positional rebase, stable-id lookup, target repair, automatic + retry, conflict resolution, or independent-branch skipping after failure. +- No selection, presence, DOM input, render scheduling, or focus policy. +- No support for multiple causal owners mutating the same document projection. +- No change to the `JSONDocument` public interface or core state. + +## Friction report + +The public document subscription, active envelope id, metadata marker, and +publication count distinguish this module's synchronous patch from ordinary +host publications. Reusing a previously observed marker outside that active +publication is rejected, and a nested patch that copies current metadata adds +a second publication and causes divergence. The marker remains a cooperative +in-process convention rather than a security boundary. Running another inbox +against the same document makes the first one diverge instead of trying to +merge two causal ledgers. + +Only mutations published through the public `JSONDocument` interface are +observable. In-place mutation of an object reachable through `doc.value` emits +no subscription event and cannot be detected here; callers must treat document +values as owned projection state rather than a side channel for mutation. + +Typed patch failure is a clean seam: core keeps the JSON projection atomic, so +the inbox can leave its frontier unchanged. The first strict +`JSONDocumentError` still propagates but its structured result is retained as +the global blocker, so later delivery does not retry implicitly. An error +callback or schema path can instead throw before returning a `JSONResult`; when +the projection is unchanged, the original exception propagates once and the +inbox becomes `faulted` rather than retrying. A host `onChange` or subscriber +can throw after core has published, however. The public interfaces cannot roll +that publication back across modules. This lab detects the changed projection +and halts as `diverged`; it does not claim crash-atomicity across arbitrary host +callbacks. + +Already-materialized guarded operations can become stale while waiting for +dependencies. The inbox then blocks without rewriting their intent. The next +useful experiment is ready-time materialization with at least two concrete +policies, such as positional patch rebase and stable-id target recovery. That +evidence should come before exposing a generic Adapter seam or choosing OT/CRDT. diff --git a/labs/extensions/causal-patch-inbox/package.json b/labs/extensions/causal-patch-inbox/package.json new file mode 100644 index 00000000..bf19d950 --- /dev/null +++ b/labs/extensions/causal-patch-inbox/package.json @@ -0,0 +1,49 @@ +{ + "name": "@interactive-os/json-document-causal-patch-inbox", + "version": "0.1.0", + "description": "Lab causal JSON Patch inbox for json-document documents.", + "private": true, + "type": "module", + "license": "MIT", + "sideEffects": false, + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "repository": { + "type": "git", + "url": "git+https://github.com/developer-1px/json-document.git", + "directory": "labs/extensions/causal-patch-inbox" + }, + "keywords": [ + "@interactive-os/json-document", + "causal", + "inbox", + "patch", + "headless" + ], + "files": [ + "README.md", + "dist" + ], + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "scripts": { + "clean": "rm -rf dist", + "build": "npm run clean && tsc -p tsconfig.json", + "test": "vitest run --config vitest.config.ts", + "typecheck": "tsc -p tsconfig.test.json --noEmit", + "verify": "npm run typecheck && npm test && npm run build" + }, + "peerDependencies": { + "@interactive-os/json-document": "^1.0.0" + }, + "devDependencies": { + "typescript": "^5.0.0", + "vitest": "^4.1.7", + "zod": "^4.0.0", + "@interactive-os/json-document": "*" + } +} diff --git a/labs/extensions/causal-patch-inbox/src/create.ts b/labs/extensions/causal-patch-inbox/src/create.ts new file mode 100644 index 00000000..8fb97157 --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/create.ts @@ -0,0 +1,360 @@ +import { + JSONDocumentError, + type JSONDocument, +} from "@interactive-os/json-document"; + +import type { + CausalPatchInbox, + CausalPatchInboxSnapshot, + CausalPatchIngestResult, + FailedCausalPatch, + FaultedCausalPatch, +} from "./types.js"; +import { + copyJson, + envelopesEqual, + prepareEnvelope, + type StoredEnvelope, +} from "./envelope.js"; +import { + findIntroducedDependencyCycle, +} from "./graph.js"; +import { + ReadyIdHeap, + compareIds, +} from "./ready.js"; + +interface PendingEnvelope { + readonly envelope: StoredEnvelope; + readonly missing: Set; +} + +let nextInboxInstance = 0; + +export function createCausalPatchInbox( + doc: JSONDocument, +): CausalPatchInbox { + const known = new Map(); + const queued = new Map(); + const dependents = new Map>(); + const readyIds = new ReadyIdHeap(); + const applied = new Set(); + const frontier = new Set(); + const origin = `causal-patch-inbox:${nextInboxInstance += 1}`; + let failure: FailedCausalPatch | undefined; + let fault: FaultedCausalPatch | undefined; + let diverged = false; + let disposed = false; + let busy = false; + let applyingId: string | undefined; + let publicationCount = 0; + let observing = true; + const unsubscribeDocument = doc.subscribe((_applied, metadata) => { + publicationCount += 1; + if ( + publicationCount > 1 + || !busy + || applyingId === undefined + || metadata?.origin !== origin + || metadata.mergeKey !== applyingId + ) { + diverged = true; + } + }); + const stopObserving = (): void => { + if (!observing) return; + observing = false; + unsubscribeDocument(); + }; + + const current = (): CausalPatchInboxSnapshot => ({ + status: disposed + ? "disposed" + : diverged + ? "diverged" + : fault !== undefined + ? "faulted" + : failure === undefined + ? "active" + : "blocked", + frontier: [...frontier].sort(compareIds), + queued: [...queued.values()] + .sort((left, right) => compareIds(left.envelope.id, right.envelope.id)) + .map((pending) => ({ + id: pending.envelope.id, + missing: [...pending.missing].sort(compareIds), + })), + ...(failure === undefined + ? {} + : { failure: copyFailure(failure) }), + ...(fault === undefined + ? {} + : { fault: { ...fault } }), + }); + + const ingestWhileBusy = ( + input: Parameters[0], + ): CausalPatchIngestResult => { + const envelopes: StoredEnvelope[] = []; + for (const candidate of Array.isArray(input) ? input : [input]) { + const prepared = prepareEnvelope(candidate); + if (!prepared.ok) { + return { + ok: false, + code: "invalid_envelope", + reason: prepared.reason, + ...(prepared.id === undefined ? {} : { id: prepared.id }), + applied: [], + }; + } + envelopes.push(prepared.envelope); + } + if (disposed) { + return { + ok: false, + code: "disposed", + reason: "causal inbox was disposed while reading envelopes", + applied: [], + }; + } + if (diverged) { + return { + ok: false, + code: "projection_diverged", + reason: "document projection changed while reading causal envelopes", + applied: [], + }; + } + + const duplicates = new Set(); + const additions = new Map(); + for (const envelope of envelopes) { + const existing = known.get(envelope.id) ?? additions.get(envelope.id); + if (existing !== undefined) { + if (!envelopesEqual(existing, envelope)) { + return { + ok: false, + code: "duplicate_mismatch", + reason: `causal envelope id has a different payload: ${envelope.id}`, + id: envelope.id, + applied: [], + }; + } + duplicates.add(envelope.id); + continue; + } + additions.set(envelope.id, envelope); + } + + const cycle = findIntroducedDependencyCycle( + additions, + (id) => queued.get(id)?.envelope, + (id) => dependents.has(id), + ); + if (cycle !== null) { + return { + ok: false, + code: "dependency_cycle", + reason: `causal dependency cycle: ${cycle.cycle.join(" -> ")}`, + id: cycle.id, + cycle: cycle.cycle, + applied: [], + }; + } + + for (const envelope of additions.values()) { + known.set(envelope.id, envelope); + const missing = new Set( + envelope.dependsOn.filter((dependency) => !applied.has(dependency)), + ); + queued.set(envelope.id, { envelope, missing }); + if (missing.size === 0) { + readyIds.push(envelope.id); + continue; + } + for (const dependency of missing) { + const waiting = dependents.get(dependency) ?? new Set(); + waiting.add(envelope.id); + dependents.set(dependency, waiting); + } + } + + const integrated: string[] = []; + while (true) { + const readyId = readyIds.pop(); + if (readyId === undefined) break; + const pending = queued.get(readyId); + if (pending === undefined) continue; + const ready = pending.envelope; + + const beforeProjection = doc.value; + let result: ReturnType; + publicationCount = 0; + applyingId = ready.id; + try { + result = doc.patch(ready.operations, { + origin, + mergeKey: ready.id, + }); + } catch (error) { + if (doc.value !== beforeProjection) diverged = true; + if (!diverged && error instanceof JSONDocumentError) { + failure = { + id: ready.id, + result: copyJson(error.result), + }; + } else if (!diverged) { + fault = { + id: ready.id, + reason: error instanceof Error + ? error.message + : "document patch threw an unknown value", + }; + } + throw error; + } finally { + applyingId = undefined; + } + + if (!result.ok) { + if (doc.value !== beforeProjection) diverged = true; + if (disposed) { + return { + ok: false, + code: "disposed", + reason: "causal inbox was disposed while applying an envelope", + applied: integrated, + }; + } + if (diverged) { + return { + ok: false, + code: "projection_diverged", + reason: "document projection changed while applying a causal envelope", + applied: integrated, + }; + } + failure = { id: ready.id, result: copyJson(result) }; + return { + ok: false, + code: "patch_failed", + reason: result.reason ?? `causal patch failed: ${ready.id}`, + id: ready.id, + result: copyJson(result), + applied: integrated, + }; + } + + queued.delete(ready.id); + applied.add(ready.id); + for (const dependency of ready.dependsOn) frontier.delete(dependency); + frontier.add(ready.id); + integrated.push(ready.id); + const waiting = dependents.get(ready.id); + if (waiting !== undefined) { + for (const dependentId of waiting) { + const dependent = queued.get(dependentId); + if (dependent === undefined) continue; + dependent.missing.delete(ready.id); + if (dependent.missing.size === 0) readyIds.push(dependentId); + } + dependents.delete(ready.id); + } + if (disposed) { + return { + ok: false, + code: "disposed", + reason: "causal inbox was disposed while applying an envelope", + applied: integrated, + }; + } + if (diverged) { + return { + ok: false, + code: "projection_diverged", + reason: "document projection changed while applying a causal envelope", + applied: integrated, + }; + } + } + + const inputIds = new Set(envelopes.map(({ id }) => id)); + return { + ok: true, + applied: integrated, + pending: [...inputIds].filter((id) => queued.has(id)).sort(compareIds), + duplicates: [...duplicates], + }; + }; + + const ingest: CausalPatchInbox["ingest"] = (input) => { + if (disposed) { + return { + ok: false, + code: "disposed", + reason: "causal inbox is disposed", + applied: [], + }; + } + if (busy) { + return { + ok: false, + code: "busy", + reason: "causal inbox is processing another ingestion", + applied: [], + }; + } + if (diverged) { + return { + ok: false, + code: "projection_diverged", + reason: "document projection changed outside the causal inbox", + applied: [], + }; + } + if (failure !== undefined) { + return { + ok: false, + code: "blocked", + reason: `causal inbox is blocked by failed envelope: ${failure.id}`, + id: failure.id, + result: copyJson(failure.result), + applied: [], + }; + } + if (fault !== undefined) { + return { + ok: false, + code: "faulted", + reason: `causal inbox faulted while applying envelope: ${fault.id}`, + id: fault.id, + applied: [], + }; + } + + busy = true; + try { + return ingestWhileBusy(input); + } finally { + busy = false; + if (disposed) stopObserving(); + } + }; + + return { + ingest, + current, + dispose() { + if (disposed) return; + disposed = true; + if (!busy) stopObserving(); + }, + }; +} + +function copyFailure(failure: FailedCausalPatch): FailedCausalPatch { + return { + id: failure.id, + result: copyJson(failure.result), + }; +} diff --git a/labs/extensions/causal-patch-inbox/src/envelope.ts b/labs/extensions/causal-patch-inbox/src/envelope.ts new file mode 100644 index 00000000..390a2d1f --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/envelope.ts @@ -0,0 +1,268 @@ +import type { + JSONPatchOperation, +} from "@interactive-os/json-document"; + +import { + compareIds, +} from "./ready.js"; + +export interface StoredEnvelope { + readonly id: string; + readonly dependsOn: ReadonlyArray; + readonly operations: ReadonlyArray; +} + +export type PreparedEnvelope = + | { readonly ok: true; readonly envelope: StoredEnvelope } + | { readonly ok: false; readonly reason: string; readonly id?: string }; + +const ENVELOPE_FIELDS = new Set(["id", "dependsOn", "operations"]); + +export function prepareEnvelope(input: unknown): PreparedEnvelope { + const fields = readEnvelopeFields(input); + if (!fields.ok) return fields; + + const { id } = fields; + if (typeof id !== "string" || id.trim().length === 0) { + return { ok: false, reason: "causal envelope id must be a non-empty string" }; + } + + let dependsOn: unknown; + let operations: unknown; + try { + dependsOn = copyJson(fields.dependsOn); + operations = copyJson(fields.operations); + } catch { + return { + ok: false, + reason: `causal envelope must contain plain JSON data: ${id}`, + id, + }; + } + + if (!Array.isArray(dependsOn)) { + return { + ok: false, + reason: `causal envelope dependencies must be an array: ${id}`, + id, + }; + } + const dependencies = new Set(); + for (const dependency of dependsOn) { + if (typeof dependency !== "string" || dependency.trim().length === 0) { + return { + ok: false, + reason: `causal dependency id must be a non-empty string: ${id}`, + id, + }; + } + if (dependency === id) { + return { + ok: false, + reason: `causal envelope cannot depend on itself: ${id}`, + id, + }; + } + if (dependencies.has(dependency)) { + return { + ok: false, + reason: `causal envelope dependency is duplicated: ${dependency}`, + id, + }; + } + dependencies.add(dependency); + } + if (!Array.isArray(operations)) { + return { + ok: false, + reason: `causal envelope operations must be an array: ${id}`, + id, + }; + } + + return { + ok: true, + envelope: { + id, + dependsOn: [...dependencies].sort(compareIds), + operations: operations as ReadonlyArray, + }, + }; +} + +export function envelopesEqual( + left: StoredEnvelope, + right: StoredEnvelope, +): boolean { + return jsonEqual(left.dependsOn, right.dependsOn) + && jsonEqual(left.operations, right.operations); +} + +export function copyJson(value: T): T { + return cloneJsonValue(value, new Set()) as T; +} + +type EnvelopeFields = + | { + readonly ok: true; + readonly id: unknown; + readonly dependsOn: unknown; + readonly operations: unknown; + } + | { readonly ok: false; readonly reason: string }; + +function readEnvelopeFields(input: unknown): EnvelopeFields { + if (input === null || typeof input !== "object" || Array.isArray(input)) { + return { ok: false, reason: "causal envelope must be an object" }; + } + const prototype = Object.getPrototypeOf(input) as unknown; + if (prototype !== Object.prototype && prototype !== null) { + return { ok: false, reason: "causal envelope must be a plain object" }; + } + + const descriptors = new Map(); + for (const key of Reflect.ownKeys(input)) { + if (typeof key !== "string" || !ENVELOPE_FIELDS.has(key)) { + return { ok: false, reason: "causal envelope has an unsupported field" }; + } + const descriptor = Object.getOwnPropertyDescriptor(input, key); + if ( + descriptor === undefined + || !descriptor.enumerable + || !("value" in descriptor) + ) { + return { ok: false, reason: `causal envelope field must be plain data: ${key}` }; + } + descriptors.set(key, descriptor); + } + + const id = descriptors.get("id"); + const dependsOn = descriptors.get("dependsOn"); + const operations = descriptors.get("operations"); + if (id === undefined || dependsOn === undefined || operations === undefined) { + return { ok: false, reason: "causal envelope requires id, dependsOn, and operations" }; + } + return { + ok: true, + id: id.value, + dependsOn: dependsOn.value, + operations: operations.value, + }; +} + +function cloneJsonValue(value: unknown, ancestors: Set): unknown { + if ( + value === null + || typeof value === "string" + || typeof value === "boolean" + ) { + return value; + } + if (typeof value === "number") { + if (!Number.isFinite(value)) throw new TypeError("JSON number must be finite"); + return value; + } + if (typeof value !== "object") throw new TypeError("value is not JSON"); + if (ancestors.has(value)) throw new TypeError("JSON value must be acyclic"); + + ancestors.add(value); + try { + return Array.isArray(value) + ? cloneJsonArray(value, ancestors) + : cloneJsonObject(value, ancestors); + } finally { + ancestors.delete(value); + } +} + +function cloneJsonArray( + value: ReadonlyArray, + ancestors: Set, +): unknown[] { + for (const key of Reflect.ownKeys(value)) { + if (key === "length") continue; + if (typeof key !== "string" || !isArrayIndex(key, value.length)) { + throw new TypeError("JSON arrays cannot have custom properties"); + } + } + + const copy: unknown[] = []; + for (let index = 0; index < value.length; index += 1) { + const descriptor = Object.getOwnPropertyDescriptor(value, String(index)); + if ( + descriptor === undefined + || !descriptor.enumerable + || !("value" in descriptor) + ) { + throw new TypeError("JSON array entries must be plain data"); + } + copy.push(cloneJsonValue(descriptor.value, ancestors)); + } + return copy; +} + +function cloneJsonObject( + value: object, + ancestors: Set, +): Record { + const prototype = Object.getPrototypeOf(value) as unknown; + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError("JSON objects must be plain objects"); + } + + const copy: Record = {}; + for (const key of Reflect.ownKeys(value)) { + if (typeof key !== "string") { + throw new TypeError("JSON objects cannot have symbol keys"); + } + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if ( + descriptor === undefined + || !descriptor.enumerable + || !("value" in descriptor) + ) { + throw new TypeError("JSON object fields must be plain data"); + } + Object.defineProperty(copy, key, { + value: cloneJsonValue(descriptor.value, ancestors), + enumerable: true, + configurable: true, + writable: true, + }); + } + return copy; +} + +function isArrayIndex(key: string, length: number): boolean { + if (!/^(?:0|[1-9]\d*)$/.test(key)) return false; + const index = Number(key); + return Number.isSafeInteger(index) && index >= 0 && index < length; +} + +function jsonEqual(left: unknown, right: unknown): boolean { + if (left === right) return true; + if ( + left === null + || right === null + || typeof left !== "object" + || typeof right !== "object" + || Array.isArray(left) !== Array.isArray(right) + ) { + return false; + } + if (Array.isArray(left)) { + if (left.length !== (right as ReadonlyArray).length) return false; + return left.every((item, index) => { + return jsonEqual(item, (right as ReadonlyArray)[index]); + }); + } + + const leftObject = left as Readonly>; + const rightObject = right as Readonly>; + const keys = Object.keys(leftObject); + if (keys.length !== Object.keys(rightObject).length) return false; + return keys.every((key) => { + return Object.prototype.hasOwnProperty.call(rightObject, key) + && jsonEqual(leftObject[key], rightObject[key]); + }); +} diff --git a/labs/extensions/causal-patch-inbox/src/graph.ts b/labs/extensions/causal-patch-inbox/src/graph.ts new file mode 100644 index 00000000..7c181a62 --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/graph.ts @@ -0,0 +1,87 @@ +import type { + StoredEnvelope, +} from "./envelope.js"; + +export interface IntroducedDependencyCycle { + readonly id: string; + readonly cycle: ReadonlyArray; +} + +export function findIntroducedDependencyCycle( + additions: ReadonlyMap, + existing: (id: string) => StoredEnvelope | undefined, + hasExistingDependent: (id: string) => boolean, +): IntroducedDependencyCycle | null { + const admitted = new Map(); + const admittedDependencyTargets = new Set(); + const lookup = (id: string): StoredEnvelope | undefined => { + return admitted.get(id) ?? existing(id); + }; + + for (const envelope of additions.values()) { + if ( + hasExistingDependent(envelope.id) + || admittedDependencyTargets.has(envelope.id) + ) { + for (const dependency of envelope.dependsOn) { + const path = findDependencyPath(dependency, envelope.id, lookup); + if (path !== null) { + return { + id: envelope.id, + cycle: [...path, dependency], + }; + } + } + } + admitted.set(envelope.id, envelope); + for (const dependency of envelope.dependsOn) { + admittedDependencyTargets.add(dependency); + } + } + return null; +} + +function findDependencyPath( + start: string, + target: string, + lookup: (id: string) => StoredEnvelope | undefined, +): string[] | null { + if (start === target) return [start]; + + interface Frame { + readonly dependencies: ReadonlyArray; + next: number; + } + + const first = lookup(start); + if (first === undefined) return null; + const visited = new Set([start]); + const path = [start]; + const stack: Frame[] = [{ + dependencies: first.dependsOn, + next: 0, + }]; + + while (stack.length > 0) { + const frame = stack[stack.length - 1]!; + const dependency = frame.dependencies[frame.next]; + if (dependency === undefined) { + stack.pop(); + path.pop(); + continue; + } + frame.next += 1; + if (dependency === target) return [...path, dependency]; + if (visited.has(dependency)) continue; + visited.add(dependency); + + const envelope = lookup(dependency); + if (envelope === undefined) continue; + path.push(dependency); + stack.push({ + dependencies: envelope.dependsOn, + next: 0, + }); + } + return null; +} diff --git a/labs/extensions/causal-patch-inbox/src/index.ts b/labs/extensions/causal-patch-inbox/src/index.ts new file mode 100644 index 00000000..937a45d5 --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/index.ts @@ -0,0 +1,13 @@ +export { createCausalPatchInbox } from "./create.js"; +export type { + CausalPatchEnvelope, + CausalPatchInbox, + CausalPatchInboxSnapshot, + CausalPatchIngestError, + CausalPatchIngestErrorCode, + CausalPatchIngestOk, + CausalPatchIngestResult, + FailedCausalPatch, + FaultedCausalPatch, + QueuedCausalPatch, +} from "./types.js"; diff --git a/labs/extensions/causal-patch-inbox/src/ready.ts b/labs/extensions/causal-patch-inbox/src/ready.ts new file mode 100644 index 00000000..9aa2ba2e --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/ready.ts @@ -0,0 +1,43 @@ +export class ReadyIdHeap { + readonly #values: string[] = []; + + push(id: string): void { + this.#values.push(id); + let index = this.#values.length - 1; + while (index > 0) { + const parent = Math.floor((index - 1) / 2); + if (compareIds(this.#values[parent]!, id) <= 0) break; + this.#values[index] = this.#values[parent]!; + index = parent; + } + this.#values[index] = id; + } + + pop(): string | undefined { + const first = this.#values[0]; + const last = this.#values.pop(); + if (first === undefined || last === undefined || this.#values.length === 0) { + return first; + } + + let index = 0; + while (true) { + const left = index * 2 + 1; + if (left >= this.#values.length) break; + const right = left + 1; + const child = right < this.#values.length + && compareIds(this.#values[right]!, this.#values[left]!) < 0 + ? right + : left; + if (compareIds(this.#values[child]!, last) >= 0) break; + this.#values[index] = this.#values[child]!; + index = child; + } + this.#values[index] = last; + return first; + } +} + +export function compareIds(left: string, right: string): number { + return left < right ? -1 : left > right ? 1 : 0; +} diff --git a/labs/extensions/causal-patch-inbox/src/types.ts b/labs/extensions/causal-patch-inbox/src/types.ts new file mode 100644 index 00000000..3e92659d --- /dev/null +++ b/labs/extensions/causal-patch-inbox/src/types.ts @@ -0,0 +1,108 @@ +import type { + JSONPatchOperation, + JSONResult, +} from "@interactive-os/json-document"; + +export interface CausalPatchEnvelope { + readonly id: string; + readonly dependsOn: ReadonlyArray; + readonly operations: ReadonlyArray; +} + +export interface QueuedCausalPatch { + readonly id: string; + readonly missing: ReadonlyArray; +} + +export interface FailedCausalPatch { + readonly id: string; + readonly result: Extract; +} + +export interface FaultedCausalPatch { + readonly id: string; + readonly reason: string; +} + +export interface CausalPatchInboxSnapshot { + readonly status: "active" | "blocked" | "diverged" | "faulted" | "disposed"; + readonly frontier: ReadonlyArray; + readonly queued: ReadonlyArray; + readonly failure?: FailedCausalPatch; + readonly fault?: FaultedCausalPatch; +} + +export interface CausalPatchIngestOk { + readonly ok: true; + readonly applied: ReadonlyArray; + readonly pending: ReadonlyArray; + readonly duplicates: ReadonlyArray; +} + +export type CausalPatchIngestErrorCode = + | "invalid_envelope" + | "duplicate_mismatch" + | "dependency_cycle" + | "patch_failed" + | "blocked" + | "projection_diverged" + | "faulted" + | "busy" + | "disposed"; + +export type CausalPatchIngestError = + | { + readonly ok: false; + readonly code: "invalid_envelope"; + readonly reason: string; + readonly applied: ReadonlyArray; + readonly id?: string; + } + | { + readonly ok: false; + readonly code: "duplicate_mismatch"; + readonly reason: string; + readonly applied: ReadonlyArray; + readonly id: string; + } + | { + readonly ok: false; + readonly code: "dependency_cycle"; + readonly reason: string; + readonly applied: ReadonlyArray; + readonly id: string; + readonly cycle: ReadonlyArray; + } + | { + readonly ok: false; + readonly code: "patch_failed" | "blocked"; + readonly reason: string; + readonly applied: ReadonlyArray; + readonly id: string; + readonly result: Extract; + } + | { + readonly ok: false; + readonly code: "projection_diverged" | "busy" | "disposed"; + readonly reason: string; + readonly applied: ReadonlyArray; + } + | { + readonly ok: false; + readonly code: "faulted"; + readonly reason: string; + readonly applied: ReadonlyArray; + readonly id: string; + }; + +export type CausalPatchIngestResult = + | CausalPatchIngestOk + | CausalPatchIngestError; + +export interface CausalPatchInbox { + ingest( + input: CausalPatchEnvelope | ReadonlyArray, + ): CausalPatchIngestResult; + current(): CausalPatchInboxSnapshot; + dispose(): void; +} diff --git a/labs/extensions/causal-patch-inbox/tests/causal-patch-inbox.test.ts b/labs/extensions/causal-patch-inbox/tests/causal-patch-inbox.test.ts new file mode 100644 index 00000000..b47ef1aa --- /dev/null +++ b/labs/extensions/causal-patch-inbox/tests/causal-patch-inbox.test.ts @@ -0,0 +1,847 @@ +import { describe, expect, test } from "vitest"; +import * as z from "zod"; + +import { + JSONDocumentError, + createJSONDocument, + type JSONChangeMetadata, +} from "@interactive-os/json-document"; +import { + createCausalPatchInbox, +} from "../src/index.js"; + +const LogSchema = z.object({ + log: z.array(z.string()), +}); + +describe("@interactive-os/json-document-causal-patch-inbox", () => { + test("queues a child until its parent is applied, then drains in causal order", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + const child = inbox.ingest({ + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add", path: "/log/-", value: "child" }], + }); + + expect(child).toEqual({ + ok: true, + applied: [], + pending: ["child"], + duplicates: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [{ id: "child", missing: ["parent"] }], + }); + expect(doc.value.log).toEqual([]); + + const parent = inbox.ingest({ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + }); + + expect(parent).toEqual({ + ok: true, + applied: ["parent", "child"], + pending: [], + duplicates: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: ["child"], + queued: [], + }); + expect(doc.value.log).toEqual(["parent", "child"]); + }); + + test("deduplicates the same envelope before and after it is applied", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const child = { + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add" as const, path: "/log/-" as const, value: "child" }], + }; + const parent = { + id: "parent", + dependsOn: [], + operations: [{ op: "add" as const, path: "/log/-" as const, value: "parent" }], + }; + + expect(inbox.ingest(child)).toMatchObject({ + ok: true, + applied: [], + duplicates: [], + }); + expect(inbox.ingest(child)).toMatchObject({ + ok: true, + applied: [], + pending: ["child"], + duplicates: ["child"], + }); + expect(doc.value.log).toEqual([]); + + expect(inbox.ingest(parent)).toMatchObject({ + ok: true, + applied: ["parent", "child"], + }); + expect(inbox.ingest(parent)).toMatchObject({ + ok: true, + applied: [], + pending: [], + duplicates: ["parent"], + }); + expect(doc.value.log).toEqual(["parent", "child"]); + }); + + test("treats dependency order as set order for exact duplicates", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const operations = [{ + op: "add" as const, + path: "/log/-" as const, + value: "child", + }]; + + expect(inbox.ingest({ + id: "child", + dependsOn: ["parent-b", "parent-a"], + operations, + })).toMatchObject({ + ok: true, + pending: ["child"], + duplicates: [], + }); + expect(inbox.ingest({ + id: "child", + dependsOn: ["parent-a", "parent-b"], + operations, + })).toMatchObject({ + ok: true, + pending: ["child"], + duplicates: ["child"], + }); + expect(inbox.current().queued).toEqual([{ + id: "child", + missing: ["parent-a", "parent-b"], + }]); + }); + + test("owns a defensive copy of admitted dependencies and operations", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const dependsOn = ["parent"]; + const operation = { + op: "add" as const, + path: "/log/-" as const, + value: "original", + }; + + expect(inbox.ingest({ + id: "child", + dependsOn, + operations: [operation], + })).toMatchObject({ ok: true, pending: ["child"] }); + dependsOn.push("never-arrives"); + operation.value = "mutated"; + + expect(inbox.ingest({ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + })).toMatchObject({ + ok: true, + applied: ["parent", "child"], + }); + expect(doc.value.log).toEqual(["parent", "original"]); + }); + + test("rejects a different payload for a known id before admitting the batch", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + expect(inbox.ingest({ + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add", path: "/log/-", value: "original" }], + })).toMatchObject({ ok: true, pending: ["child"] }); + + const result = inbox.ingest([{ + id: "independent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "independent" }], + }, { + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add", path: "/log/-", value: "replacement" }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "duplicate_mismatch", + id: "child", + applied: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [{ id: "child", missing: ["parent"] }], + }); + expect(doc.value.log).toEqual([]); + + expect(inbox.ingest({ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + })).toMatchObject({ + ok: true, + applied: ["parent", "child"], + }); + expect(doc.value.log).toEqual(["parent", "original"]); + }); + + test("applies the currently known ready batch in id order", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + const result = inbox.ingest([{ + id: "change-c", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "C" }], + }, { + id: "change-a", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "A" }], + }, { + id: "change-b", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "B" }], + }]); + + expect(result).toMatchObject({ + ok: true, + applied: ["change-a", "change-b", "change-c"], + }); + expect(doc.value.log).toEqual(["A", "B", "C"]); + expect(inbox.current().frontier).toEqual([ + "change-a", + "change-b", + "change-c", + ]); + }); + + test("blocks without advancing causal state when one envelope patch fails", () => { + const doc = createJSONDocument(LogSchema, { log: [] }, { history: 10 }); + const inbox = createCausalPatchInbox(doc); + + const result = inbox.ingest([{ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + }, { + id: "child", + dependsOn: ["parent"], + operations: [ + { op: "add", path: "/log/-", value: "must-not-leak" }, + { op: "replace", path: "/log", value: 42 }, + ], + }, { + id: "grandchild", + dependsOn: ["child"], + operations: [{ op: "add", path: "/log/-", value: "grandchild" }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "patch_failed", + id: "child", + applied: ["parent"], + result: { ok: false, code: "schema_violation" }, + }); + expect(inbox.current()).toMatchObject({ + status: "blocked", + frontier: ["parent"], + queued: [ + { id: "child", missing: [] }, + { id: "grandchild", missing: ["child"] }, + ], + failure: { + id: "child", + result: { ok: false, code: "schema_violation" }, + }, + }); + expect(doc.value.log).toEqual(["parent"]); + expect(doc.history.undoDepth).toBe(1); + + const blocked = inbox.ingest({ + id: "independent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "independent" }], + }); + expect(blocked).toMatchObject({ + ok: false, + code: "blocked", + applied: [], + }); + expect(inbox.current().status).toBe("blocked"); + expect(doc.value.log).toEqual(["parent"]); + }); + + test("halts when the document projection changes outside the inbox", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + expect(doc.patch({ + op: "add", + path: "/log/-", + value: "external", + })).toEqual({ ok: true }); + expect(inbox.current()).toEqual({ + status: "diverged", + frontier: [], + queued: [], + }); + + const result = inbox.ingest({ + id: "remote", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "remote" }], + }); + expect(result).toMatchObject({ + ok: false, + code: "projection_diverged", + applied: [], + }); + expect(doc.value.log).toEqual(["external"]); + }); + + test("rejects reentrant ingestion while an envelope is being published", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + let attempted = false; + let reentrant: ReturnType | undefined; + const unsubscribe = doc.subscribe(() => { + if (attempted) return; + attempted = true; + reentrant = inbox.ingest({ + id: "nested", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "nested" }], + }); + }); + + const result = inbox.ingest({ + id: "outer", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "outer" }], + }); + + expect(reentrant).toMatchObject({ + ok: false, + code: "busy", + applied: [], + }); + expect(result).toMatchObject({ + ok: true, + applied: ["outer"], + }); + expect(doc.value.log).toEqual(["outer"]); + unsubscribe(); + }); + + test("rejects a dependency cycle before admitting the closing envelope", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + expect(inbox.ingest({ + id: "a", + dependsOn: ["b"], + operations: [{ op: "add", path: "/log/-", value: "A" }], + })).toMatchObject({ ok: true, pending: ["a"] }); + + const result = inbox.ingest({ + id: "b", + dependsOn: ["a"], + operations: [{ op: "add", path: "/log/-", value: "B" }], + }); + + expect(result).toMatchObject({ + ok: false, + code: "dependency_cycle", + id: "b", + cycle: ["a", "b", "a"], + applied: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [{ id: "a", missing: ["b"] }], + }); + expect(doc.value.log).toEqual([]); + }); + + test("rejects a deep dependency cycle without recursive stack growth", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const size = 6_000; + const envelopes = Array.from({ length: size }, (_, index) => ({ + id: `change-${index.toString().padStart(4, "0")}`, + dependsOn: [ + `change-${((index + 1) % size).toString().padStart(4, "0")}`, + ], + operations: [], + })); + + expect(inbox.ingest(envelopes)).toMatchObject({ + ok: false, + code: "dependency_cycle", + applied: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("admits a deep acyclic pending chain without retraversing its prefix", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const size = 10_000; + let lastResult: ReturnType | undefined; + + for (let index = 0; index < size; index += 1) { + const id = `change-${index.toString().padStart(5, "0")}`; + const dependency = index === 0 + ? "missing-root" + : `change-${(index - 1).toString().padStart(5, "0")}`; + lastResult = inbox.ingest({ + id, + dependsOn: [dependency], + operations: [], + }); + if (!lastResult.ok) throw new Error(lastResult.code); + } + + expect(lastResult).toMatchObject({ + ok: true, + pending: ["change-09999"], + }); + expect(inbox.current().queued).toHaveLength(size); + }, 5_000); + + test("rejects invalid dependency identities before admitting the batch", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + const result = inbox.ingest([{ + id: "valid", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "valid" }], + }, { + id: "self", + dependsOn: ["self"], + operations: [{ op: "add", path: "/log/-", value: "self" }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "invalid_envelope", + id: "self", + applied: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("stops after the current envelope when publication reentry diverges the projection", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + let mutated = false; + const unsubscribe = doc.subscribe(() => { + if (mutated) return; + mutated = true; + expect(doc.patch({ + op: "add", + path: "/log/-", + value: "external", + })).toEqual({ ok: true }); + }); + + const result = inbox.ingest([{ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + }, { + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add", path: "/log/-", value: "child" }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "projection_diverged", + applied: ["parent"], + }); + expect(inbox.current()).toEqual({ + status: "diverged", + frontier: ["parent"], + queued: [{ id: "child", missing: [] }], + }); + expect(doc.value.log).toEqual(["parent", "external"]); + unsubscribe(); + }); + + test("marks divergence when document publication throws after changing state", () => { + const doc = createJSONDocument(LogSchema, { log: [] }, { + onChange() { + throw new Error("host publication failed"); + }, + }); + const inbox = createCausalPatchInbox(doc); + + expect(() => inbox.ingest({ + id: "remote", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "remote" }], + })).toThrow("host publication failed"); + + expect(doc.value.log).toEqual(["remote"]); + expect(inbox.current()).toEqual({ + status: "diverged", + frontier: [], + queued: [{ id: "remote", missing: [] }], + }); + }); + + test("integrates a causal envelope whose projection patch is empty", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + expect(inbox.ingest({ + id: "projection-noop", + dependsOn: [], + operations: [], + })).toEqual({ + ok: true, + applied: ["projection-noop"], + pending: [], + duplicates: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: ["projection-noop"], + queued: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("stops observing and ingesting after disposal", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + inbox.dispose(); + expect(inbox.current()).toEqual({ + status: "disposed", + frontier: [], + queued: [], + }); + expect(inbox.ingest({ + id: "late", + dependsOn: [], + operations: [], + })).toMatchObject({ + ok: false, + code: "disposed", + applied: [], + }); + + expect(doc.patch({ + op: "add", + path: "/log/-", + value: "external-after-dispose", + })).toEqual({ ok: true }); + expect(inbox.current().status).toBe("disposed"); + }); + + test("rejects non-JSON envelope values before admitting the batch", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + + const result = inbox.ingest([{ + id: "valid", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "valid" }], + }, { + id: "not-json", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: Number.NaN }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "invalid_envelope", + id: "not-json", + applied: [], + }); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("rejects accessor envelope fields without invoking them", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + let reads = 0; + const envelope = {} as Record; + Object.defineProperties(envelope, { + id: { + enumerable: true, + get() { + reads += 1; + return "accessor"; + }, + }, + dependsOn: { + enumerable: true, + value: [], + }, + operations: { + enumerable: true, + value: [], + }, + }); + + const result = inbox.ingest(envelope as never); + + expect(result).toMatchObject({ + ok: false, + code: "invalid_envelope", + applied: [], + }); + expect(reads).toBe(0); + expect(inbox.current()).toEqual({ + status: "active", + frontier: [], + queued: [], + }); + }); + + test("rejects reentrant ingestion while envelope admission is reading descriptors", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + let attempted = false; + let reentrant: ReturnType | undefined; + const envelope = new Proxy({ + id: "outer", + dependsOn: [], + operations: [{ op: "add" as const, path: "/log/-" as const, value: "outer" }], + }, { + ownKeys(target) { + if (!attempted) { + attempted = true; + reentrant = inbox.ingest({ + id: "nested", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "nested" }], + }); + } + return Reflect.ownKeys(target); + }, + }); + + const result = inbox.ingest(envelope); + + expect(reentrant).toMatchObject({ + ok: false, + code: "busy", + applied: [], + }); + expect(result).toMatchObject({ + ok: true, + applied: ["outer"], + }); + expect(doc.value.log).toEqual(["outer"]); + }); + + test("preserves an unapplied envelope when a strict document throws", () => { + const doc = createJSONDocument(LogSchema, { log: [] }, { strict: true }); + const inbox = createCausalPatchInbox(doc); + const envelope = { + id: "invalid-for-schema", + dependsOn: [], + operations: [{ op: "replace" as const, path: "/log" as const, value: 42 }], + }; + + expect(() => inbox.ingest(envelope)).toThrow(JSONDocumentError); + expect(inbox.current()).toEqual({ + status: "blocked", + frontier: [], + queued: [{ id: "invalid-for-schema", missing: [] }], + failure: { + id: "invalid-for-schema", + result: expect.objectContaining({ + ok: false, + code: "schema_violation", + }), + }, + }); + expect(doc.value.log).toEqual([]); + + expect(inbox.ingest(envelope)).toMatchObject({ + ok: false, + code: "blocked", + id: "invalid-for-schema", + result: { ok: false, code: "schema_violation" }, + applied: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("faults without retry when a document error callback throws", () => { + const doc = createJSONDocument(LogSchema, { log: [] }, { + onError() { + throw new Error("host error callback failed"); + }, + }); + const inbox = createCausalPatchInbox(doc); + + expect(() => inbox.ingest({ + id: "invalid-for-schema", + dependsOn: [], + operations: [{ op: "replace", path: "/log", value: 42 }], + })).toThrow("host error callback failed"); + expect(inbox.current()).toEqual({ + status: "faulted", + frontier: [], + queued: [{ id: "invalid-for-schema", missing: [] }], + fault: { + id: "invalid-for-schema", + reason: "host error callback failed", + }, + }); + + expect(inbox.ingest({ + id: "independent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "must-not-apply" }], + })).toMatchObject({ + ok: false, + code: "faulted", + id: "invalid-for-schema", + applied: [], + }); + expect(doc.value.log).toEqual([]); + }); + + test("does not trust a previously observed internal origin outside publication", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + let observed: JSONChangeMetadata | undefined; + const unsubscribe = doc.subscribe((_applied, metadata) => { + observed = metadata; + }); + const inbox = createCausalPatchInbox(doc); + + expect(inbox.ingest({ + id: "remote", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "remote" }], + })).toMatchObject({ ok: true, applied: ["remote"] }); + if (observed?.origin === undefined || observed.mergeKey === undefined) { + throw new Error("expected inbox origin metadata"); + } + const observedOrigin = observed.origin; + const observedMergeKey = observed.mergeKey; + + expect(doc.patch({ + op: "add", + path: "/log/-", + value: "spoofed-external", + }, { + origin: observedOrigin, + mergeKey: observedMergeKey, + })).toEqual({ ok: true }); + + expect(inbox.current().status).toBe("diverged"); + unsubscribe(); + }); + + test("detects a nested external patch that copies current publication metadata", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + let nested = false; + const unsubscribe = doc.subscribe((_applied, metadata) => { + if (nested || metadata?.origin === undefined || metadata.mergeKey === undefined) return; + nested = true; + expect(doc.patch({ + op: "add", + path: "/log/-", + value: "copied-metadata", + }, { + origin: metadata.origin, + mergeKey: metadata.mergeKey, + })).toEqual({ ok: true }); + }); + const inbox = createCausalPatchInbox(doc); + + const result = inbox.ingest({ + id: "remote", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "remote" }], + }); + + expect(result).toMatchObject({ + ok: false, + code: "projection_diverged", + applied: ["remote"], + }); + expect(inbox.current()).toMatchObject({ + status: "diverged", + frontier: ["remote"], + }); + expect(doc.value.log).toEqual(["remote", "copied-metadata"]); + unsubscribe(); + }); + + test("stops after the current envelope when disposed during publication", () => { + const doc = createJSONDocument(LogSchema, { log: [] }); + const inbox = createCausalPatchInbox(doc); + const unsubscribe = doc.subscribe(() => { + inbox.dispose(); + }); + + const result = inbox.ingest([{ + id: "parent", + dependsOn: [], + operations: [{ op: "add", path: "/log/-", value: "parent" }], + }, { + id: "child", + dependsOn: ["parent"], + operations: [{ op: "add", path: "/log/-", value: "child" }], + }]); + + expect(result).toMatchObject({ + ok: false, + code: "disposed", + applied: ["parent"], + }); + expect(inbox.current()).toEqual({ + status: "disposed", + frontier: ["parent"], + queued: [{ id: "child", missing: [] }], + }); + expect(doc.value.log).toEqual(["parent"]); + unsubscribe(); + }); +}); diff --git a/labs/extensions/causal-patch-inbox/tsconfig.json b/labs/extensions/causal-patch-inbox/tsconfig.json new file mode 100644 index 00000000..615be15f --- /dev/null +++ b/labs/extensions/causal-patch-inbox/tsconfig.json @@ -0,0 +1,22 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "declaration": true, + "declarationMap": false, + "sourceMap": false, + "strict": true, + "exactOptionalPropertyTypes": true, + "noUncheckedIndexedAccess": true, + "rootDir": "src", + "outDir": "dist", + "lib": [ + "ES2022", + "DOM" + ] + }, + "include": [ + "src/**/*.ts" + ] +} diff --git a/labs/extensions/causal-patch-inbox/tsconfig.test.json b/labs/extensions/causal-patch-inbox/tsconfig.test.json new file mode 100644 index 00000000..49ef16fc --- /dev/null +++ b/labs/extensions/causal-patch-inbox/tsconfig.test.json @@ -0,0 +1,15 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "rootDir": ".", + "noEmit": true, + "types": [ + "vitest" + ] + }, + "include": [ + "src/**/*.ts", + "tests/**/*.ts", + "vitest.config.ts" + ] +} diff --git a/labs/extensions/causal-patch-inbox/vitest.config.ts b/labs/extensions/causal-patch-inbox/vitest.config.ts new file mode 100644 index 00000000..b470175d --- /dev/null +++ b/labs/extensions/causal-patch-inbox/vitest.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + resolve: { + alias: { + "@interactive-os/json-document": new URL("../../../packages/json-document/src/application/document/index.ts", import.meta.url).pathname, + }, + }, + test: { + include: ["tests/**/*.test.ts"], + }, +}); diff --git a/scripts/generate-docs.mjs b/scripts/generate-docs.mjs index 6258c738..dbdf22f4 100644 --- a/scripts/generate-docs.mjs +++ b/scripts/generate-docs.mjs @@ -106,6 +106,10 @@ const extensionGuidance = { useFor: "materialize one delayed field replacement against a current stable-id target", notFor: "id policy, structural patches, CRDT/OT protocols, transport, or merge UI", }, + "@interactive-os/json-document-causal-patch-inbox": { + useFor: "queue and deduplicate dependency-declared envelopes, applying each as one core JSON Patch batch", + notFor: "CRDT/OT convergence, transport, persistence, ready-time rebase, or conflict resolution", + }, "@interactive-os/json-document-drag-drop": { useFor: "turn drag/drop intent into move or paste operations", notFor: "DOM drag/drop events, hit testing, or hover UI",