Implement durable typed journals, subscriptions, and Reader document connections

Add checked native journal/subscription bindings, managed-reference retention, bounded history and backpressure, atomic delivery receipts, and fenced hosting. Preserve source publication prediction and replica reconciliation; subscription attachment, controls, and delivery remain authoritative in this initial version. Connect independent manual/Gutenberg document producers to Reader bookshelves through typed adapters. Add connection inspection, shared provider selection, authoring documentation, and restart/rejection/fan-out integration coverage. Verified compiler suite (287 passed, two existing skips), Nix runtime and PostgreSQL qualification, five example assemblies, headless Reader integration, Chromium Reader integration, resource pins, formatting, and lint. No pinned workspace runtime was deployed.
2026-09-28 19:35:51 -07:00
parent 0ace71eae6
commit 1e72713ae2
30 changed files with 2949 additions and 2033 deletions
@@ -0,0 +1,186 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { compileCapabilityResourceSource, compileCapabilitySource } from "../src/capability-language/parser.js";
+import { formatQx } from "../src/capability-language/source.js";
+import { compileExecutionPlan, compileKernelProgram } from "../src/execution-world/index.js";
+import type { InterfaceRevision } from "../src/capability-model/types.js";
+import { readFileSync } from "node:fs";
+
+const source = { repository: "https://example.test/journals", commit: "a".repeat(40) };
+
+test("standard event contracts instantiate generic native conformance rules", () => {
+  const interfaces = new Map<string, InterfaceRevision>();
+  for (const [name, slug] of [
+    ["EventStream", "event-stream"],
+    ["EventHandler", "event-handler"],
+    ["EventSubscription", "event-subscription"],
+    ["EventStreamRecord", "event-stream-record"],
+    ["EventSubscriptionRecord", "event-subscription-record"],
+  ]) {
+    const text = readFileSync(new URL(`../../test/fixtures/events/${slug}.qx`, import.meta.url), "utf8");
+    const result = compileCapabilityResourceSource(text, {
+      source,
+      environment: { interfaces, interfaceClosure: [...interfaces.values()] },
+    });
+    assert.ok(result.ok && result.resource.kind === "interface", JSON.stringify(result.diagnostics));
+    interfaces.set(name!, result.resource.revision);
+  }
+  const rules = readFileSync(new URL("../../test/fixtures/events/rules.qx", import.meta.url), "utf8")
+    .replace(/^fragment\s*\{/, "")
+    .replace(/}\s*$/, "");
+  const result = compileCapabilitySource(
+    `workspace Events id "events" revision "events@1" commit "${source.commit}" {
+    ${rules}
+    atom TextStream id "text-stream";
+    conform TextStream as EventStreamRecord<string> id "text-marker" {}
+    atom TextSubscription id "text-subscription";
+    conform TextSubscription as EventSubscriptionRecord<string,unit> id "text-subscription-marker" {}
+    atom NumberStream id "number-stream";
+    conform NumberStream as EventStreamRecord<uint32> id "number-marker" {}
+  }`,
+    "events.qx",
+    { interfaces, interfaceClosure: [...interfaces.values()] },
+  );
+  assert.ok(result.ok, JSON.stringify(result.diagnostics));
+  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
+    model: { owners: { slots: { policy: string }[] }[] };
+  };
+  assert.equal(
+    kernel.model.owners.flatMap((owner) => owner.slots).filter((slot) => slot.policy === "journal").length,
+    2,
+  );
+  assert.equal(
+    kernel.model.owners.flatMap((owner) => owner.slots).filter((slot) => slot.policy === "subscription").length,
+    1,
+  );
+});
+const contract = `external atom Document id "document";
+type Event = record { document: atom-ref<Document>; title: string; };
+interface Stream id "stream" revision "stream@1" {
+  operation publish id "publish" : Event -> record { publication: string; } {
+    call id "publish:call" execution transaction targets [server];
+  }
+  queryable value head id "head" : optional<uint64> { get id "head:get" execution snapshot targets [server]; }
+  operation readPage id "page" : record { after: optional<string>; limit: uint32; }
+    -> record { entries: list<record { publication: string; position: uint64; payload: Event; }>; next: optional<string>; asOfHead: optional<uint64>; } {
+    call id "page:call" execution snapshot targets [server];
+  }
+}`;
+const assembly = `workspace Journals id "journals" revision "journals@1" commit "${source.commit}" {
+  import interface Stream;
+  atom Document id "document";
+  atom Journal id "journal-object";
+  conform Journal as Stream id "journal-owner" {
+    private state Entries id "entries" on Journal : record { document: atom-ref<Document>; title: string; }
+      policy journal { "maxEntries": 32, "maxBytes": 65536, "maxPublicationBytes": 8192, "maxProtectedSubscriptions": 8 };
+    bind publish.call to journal Entries.publish;
+    bind head.get to journal Entries.head;
+    bind readPage.call to journal Entries.readPage;
+  }
+}`;
+function compile(text = assembly, iface = contract) {
+  const resource = compileCapabilityResourceSource(iface, { source });
+  assert.ok(resource.ok && resource.resource.kind === "interface", JSON.stringify(resource.diagnostics));
+  return compileCapabilitySource(formatQx(text), "journal.qx", {
+    interfaces: new Map([["Stream", resource.resource.revision]]),
+  });
+}
+test("journal bindings typecheck and lower managed reference pins", () => {
+  const result = compile();
+  assert.ok(result.ok, JSON.stringify(result.diagnostics));
+  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
+    model: { edges: unknown[]; owners: { slots: { policy: string; journal: { pins: unknown[] } }[] }[] };
+    operations: { binding: { kind: string; primitive: string } }[];
+  };
+  assert.equal(kernel.model.owners[0]!.slots[0]!.policy, "journal");
+  assert.equal(kernel.model.owners[0]!.slots[0]!.journal.pins.length, 1);
+  assert.equal(kernel.model.edges.length, 1);
+  assert.deepEqual(kernel.operations.map((op) => op.binding.primitive).sort(), ["head", "publish", "readPage"]);
+  assert.ok(kernel.operations.every((op) => op.binding.kind === "journal"));
+});
+test("journal declarations reject state access, bad bounds, seeding and signature substitution", () => {
+  for (const [ws, iface] of [
+    [assembly.replace("journal Entries.head", "state Entries.read"), contract],
+    [assembly.replace('"maxEntries": 32', '"maxEntries": 0'), contract],
+    [assembly.replace("bind publish.call", "initialize { seed to state Entries; }; bind publish.call"), contract],
+    [assembly, contract.replace("publication: string", "publication: uint32")],
+    [assembly, contract.replace("execution transaction", "execution snapshot")],
+  ]) {
+    const result = compile(ws, iface);
+    assert.equal(result.ok, false, `${ws}\n${iface}`);
+  }
+});
+
+test("subscription storage infers its typed handler and retains configuration references", () => {
+  const interfaces = new Map<string, InterfaceRevision>();
+  const resource = (name: string, text: string) => {
+    const result = compileCapabilityResourceSource(text, {
+      source,
+      environment: { interfaces, interfaceClosure: [...interfaces.values()] },
+    });
+    assert.ok(result.ok && result.resource.kind === "interface", JSON.stringify(result.diagnostics));
+    interfaces.set(name, result.resource.revision);
+  };
+  resource("Stream", contract);
+  resource(
+    "Receiver",
+    `external atom Document id "document";
+interface Receiver id "receiver" revision "receiver@1" {
+  operation receive id "receive" : record { event: record { document: atom-ref<Document>; title: string; }; parameters: string; } -> unit {
+    call id "receive:call" execution transaction targets [server];
+  }
+}`,
+  );
+  resource(
+    "Subscription",
+    `external atom Document id "document";
+import interface Stream;
+import interface Receiver;
+type Start = variant id "start" { future id "future" : unit; retained id "retained" : unit; afterCursor id "after" : string; };
+interface Subscription id "subscription" revision "subscription@1" {
+  operation attach id "attach" : record { source: interface-ref<Stream>; handler: interface-ref<Receiver>; parameters: string; start: Start; } -> unit {
+    call id "attach:call" execution transaction targets [server];
+  }
+  queryable value status id "status" : record { status: string; generation: uint64; cursor: uint64; failure: optional<record { publication: string; code: string; message: string; }>; } {
+    get id "status:get" execution snapshot targets [server];
+  }
+  operation retire id "retire" : unit -> unit { call id "retire:call" execution transaction targets [server]; }
+}`,
+  );
+  const ws = assembly
+    .replace(
+      "import interface Stream;",
+      "import interface Stream; import interface Receiver; import interface Subscription;",
+    )
+    .replace(
+      /\n}$/,
+      `
+  atom Delivery id "delivery";
+  conform Delivery as Subscription id "delivery-owner" {
+    private state DeliveryState id "delivery-state" on Delivery : string policy subscription(record { document: atom-ref<Document>; title: string; });
+    bind attach.call to subscription DeliveryState.attach;
+    bind status.get to subscription DeliveryState.status;
+    bind retire.call to subscription DeliveryState.retire;
+  }
+}`,
+    );
+  const result = compileCapabilitySource(formatQx(ws), "subscriptions.qx", {
+    interfaces,
+    interfaceClosure: [...interfaces.values()],
+  });
+  assert.ok(result.ok, JSON.stringify(result.diagnostics));
+  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
+    model: {
+      owners: { slots: { policy: string; subscription?: { operation: string; pins: unknown[]; active: unknown } }[] }[];
+    };
+  };
+  const slot = kernel.model.owners.flatMap((owner) => owner.slots).find((slot) => slot.policy === "subscription")!;
+  assert.equal(slot.subscription!.operation, "receive:call");
+  assert.equal(slot.subscription!.pins.length, 2);
+  assert.ok(slot.subscription!.active);
+  const bad = compileCapabilitySource(ws.replace("on Delivery : string", "on Delivery : uint32"), "bad.qx", {
+    interfaces,
+    interfaceClosure: [...interfaces.values()],
+  });
+  assert.equal(bad.ok, false, "remembered parameter type must agree with attachment and handler");
+});