Add CultMesh streaming mode surface

This commit is contained in:
Meta Vens 2026-06-02 00:57:07 +01:00
commit 67d3bf3dcf
4 changed files with 345 additions and 0 deletions

View file

@ -22,6 +22,7 @@ and pretending the label makes it furniture.
| Procedural geometry state | `GameCult.Geometry` | CultCache-native domain trees, LOD build requests, selected-cut diagnostics, and chunk artifact payloads | Rust, Unity, or remote workers need to share CSG/LOD geometry and graph metadata as typed state | Transport policy, peer discovery, or gameplay authority |
| Network transport and database plumbing | `GameCult.Networking` / CultNet | LiteNetLib transport, authentication, schema-v0 wire contracts, shard authority, raw document replication, snapshots, and subscriptions | You need a client/server pipe, login/session flow, schema discovery, or a low-level distributed CultCache lane | Gameplay-facing mesh ergonomics, Verse policy, mod branches, or simulation consensus composition |
| Distributed realtime gameplay state | `GameCult.Mesh` / CultMesh | Public mesh entrypoints, Verse discovery, peer exchange, shard replication defaults, authority leases, client prediction, and witness consensus | You want the game to treat clients and servers as one reactive database for persistent state, input state, and simulation facts | A tiny local-only tool, a bare transport client, or a storage format contract |
| Realtime media/frame streams | CultMesh streaming mode | Stream identity, authority, clock metadata, body transport negotiation, frame cursors, and backpressure state | Audio/video/tensor frames need to move between runtimes through shared memory, GPU handles, platform buffers, or CultCache page refs | Durable document mutation, mesh consensus facts, or pretending inline bytes are zero-copy |
Quick rule:
@ -32,6 +33,8 @@ Quick rule:
schema-aware database messages?"
- Choose CultMesh when the problem is "how does a game join a Verse and share
realtime state across a mesh?"
- Choose CultMesh streaming mode when the problem is "how do these runtimes move
audio/video frames while the mesh owns identity, clocks, cursors, and pressure?"
CultMesh sits on CultNet, and CultNet distributes CultCache documents. That is
the stack. If a design needs another peer-to-peer category, first check whether

View file

@ -0,0 +1,75 @@
# CultMesh TS
`cultmesh-ts` is the TypeScript surface for local CultMesh nodes: typed
CultCache documents, peer catalogs, authority leases, and realtime stream
contracts for runtimes that need to share media without turning the mesh into a
base64 soup kitchen.
## Streaming Mode
CultMesh streaming mode is for audio, video, tensor, and opaque byte frames that
need to move between diverse runtimes with as few copies as the platform allows.
It is not document replication with a bigger packet.
The split is deliberate:
- CultMesh owns stream identity, authority, clock metadata, negotiated body
transport, frame cursors, and health/backpressure state.
- The body lane owns frame bytes or native handles: shared memory, D3D11/D3D12
textures, DMA-BUF, IOSurface, AHardwareBuffer, CultCache page refs, or inline
bytes as a last resort.
- Producers report `unavoidableCopyCount`; zero-copy is a target and a measured
property, not a vibe in a lab coat.
The first public API is `CultMeshStreamCatalog`:
```ts
import { CultMesh } from "cultmesh-ts";
const streams = CultMesh.createStreamCatalog();
streams.declare({
streamId: "mimir:kiyo-pro",
verseId: "studio",
ownerPeerId: "starfire",
kind: "video",
clock: { clockDomainId: "starfire-qpc" },
video: {
width: 1920,
height: 1080,
pixelFormat: "YUY2",
framesPerSecond: 30,
},
preferredTransports: [
"shared-d3d12-texture",
"shared-memory",
"cultcache-page",
],
});
const lane = streams.negotiate("mimir:kiyo-pro", {
peerId: "fensalir",
verseId: "studio",
supportedTransports: ["shared-d3d12-texture", "cultcache-page"],
acceptedKinds: ["video"],
canImportGpuHandles: true,
});
streams.publishFrame({
streamId: lane.streamId,
sequence: 42n,
timestampNs: 1_000_000_000n,
durationNs: 33_333_334n,
transport: lane.transport,
nativeHandle: "0xfeed",
fenceHandle: "0xbeef",
fenceValue: 7n,
unavoidableCopyCount: 0,
});
```
Current implementation status: this package owns the typed stream declaration,
negotiation, and latest-frame cursor surface. Runtime-specific transports still
need platform adapters: named shared memory on local processes, D3D/Metal/Vulkan
handle import for GPU frames, and CultCache page append/readback for durable
recording or nonlocal fallbacks.

View file

@ -166,6 +166,190 @@ export class CultMeshAuthorityLeaseCatalog {
}
}
export type CultMeshStreamKind = "audio" | "video" | "tensor" | "bytes";
export type CultMeshStreamBodyTransport =
| "shared-memory"
| "shared-d3d12-texture"
| "shared-d3d11-texture"
| "dma-buf"
| "iosurface"
| "ahardwarebuffer"
| "cultcache-page"
| "inline-bytes";
export type CultMeshStreamAccess = "read" | "write" | "read-write";
export interface CultMeshStreamClock {
clockDomainId: string;
sourceId?: string;
sampleRate?: number;
offsetToVerseTimeNs?: number;
confidence?: number;
evidenceKind?: string;
}
export interface CultMeshAudioStreamFormat {
sampleRate: number;
channels: number;
sampleFormat: "float32" | "int16" | "int24" | "int32";
framesPerPacket?: number;
}
export interface CultMeshVideoStreamFormat {
width: number;
height: number;
pixelFormat: string;
framesPerSecond?: number;
planeCount?: number;
}
export interface CultMeshStreamDescriptor {
streamId: string;
verseId: string;
ownerPeerId: string;
kind: CultMeshStreamKind;
label?: string;
clock: CultMeshStreamClock;
audio?: CultMeshAudioStreamFormat;
video?: CultMeshVideoStreamFormat;
preferredTransports: readonly CultMeshStreamBodyTransport[];
requiredAccess?: CultMeshStreamAccess;
maxInFlightFrames?: number;
metadataSchemaId?: string;
}
export interface CultMeshStreamConsumerProfile {
peerId: string;
verseId: string;
supportedTransports: readonly CultMeshStreamBodyTransport[];
acceptedKinds?: readonly CultMeshStreamKind[];
canImportGpuHandles?: boolean;
canMapSharedMemory?: boolean;
maxInFlightFrames?: number;
}
export interface CultMeshStreamNegotiation {
streamId: string;
producerPeerId: string;
consumerPeerId: string;
transport: CultMeshStreamBodyTransport;
access: CultMeshStreamAccess;
maxInFlightFrames: number;
copyBudget: "zero-copy-target" | "one-copy-fallback" | "opaque-runtime";
}
export interface CultMeshStreamFrameHandle {
streamId: string;
sequence: bigint;
timestampNs: bigint;
durationNs?: bigint;
transport: CultMeshStreamBodyTransport;
byteLength?: number;
nativeHandle?: string;
resourceKey?: string;
pageRef?: string;
fenceHandle?: string;
fenceValue?: bigint;
unavoidableCopyCount?: number;
metadata?: Record<string, unknown>;
}
export class CultMeshStreamCatalog {
readonly #streams = new Map<string, CultMeshStreamDescriptor>();
readonly #latestFrames = new Map<string, CultMeshStreamFrameHandle>();
public get streams(): readonly CultMeshStreamDescriptor[] {
return [...this.#streams.values()].sort((left, right) =>
left.streamId.localeCompare(right.streamId),
);
}
public declare(stream: CultMeshStreamDescriptor): CultMeshStreamDescriptor {
requireNonEmpty(stream.streamId, "stream.streamId");
requireNonEmpty(stream.verseId, "stream.verseId");
requireNonEmpty(stream.ownerPeerId, "stream.ownerPeerId");
requireNonEmpty(stream.clock.clockDomainId, "stream.clock.clockDomainId");
if (stream.preferredTransports.length === 0) {
throw new Error("stream.preferredTransports must not be empty.");
}
this.#streams.set(stream.streamId, stream);
return stream;
}
public get(streamId: string): CultMeshStreamDescriptor | undefined {
requireNonEmpty(streamId, "streamId");
return this.#streams.get(streamId);
}
public find(
verseId: string,
kind?: CultMeshStreamKind,
): readonly CultMeshStreamDescriptor[] {
requireNonEmpty(verseId, "verseId");
return this.streams.filter(
(stream) => stream.verseId === verseId && (!kind || stream.kind === kind),
);
}
public negotiate(
streamId: string,
consumer: CultMeshStreamConsumerProfile,
): CultMeshStreamNegotiation {
const stream = this.get(streamId);
if (!stream) {
throw new Error(`Unknown CultMesh stream '${streamId}'.`);
}
if (consumer.verseId !== stream.verseId) {
throw new Error("stream and consumer must belong to the same Verse.");
}
if (
consumer.acceptedKinds &&
!consumer.acceptedKinds.includes(stream.kind)
) {
throw new Error(`consumer does not accept ${stream.kind} streams.`);
}
const transport = stream.preferredTransports.find((candidate) =>
consumer.supportedTransports.includes(candidate),
);
if (!transport) {
throw new Error("stream and consumer have no compatible body transport.");
}
return {
streamId: stream.streamId,
producerPeerId: stream.ownerPeerId,
consumerPeerId: consumer.peerId,
transport,
access: stream.requiredAccess ?? "read",
maxInFlightFrames: Math.min(
stream.maxInFlightFrames ?? Number.MAX_SAFE_INTEGER,
consumer.maxInFlightFrames ?? Number.MAX_SAFE_INTEGER,
),
copyBudget: copyBudgetFor(transport),
};
}
public publishFrame(frame: CultMeshStreamFrameHandle): CultMeshStreamFrameHandle {
requireNonEmpty(frame.streamId, "frame.streamId");
if (!this.#streams.has(frame.streamId)) {
throw new Error(`Unknown CultMesh stream '${frame.streamId}'.`);
}
this.#latestFrames.set(frame.streamId, frame);
return frame;
}
public latestFrame(streamId: string): CultMeshStreamFrameHandle | undefined {
requireNonEmpty(streamId, "streamId");
return this.#latestFrames.get(streamId);
}
}
export class CultMesh {
public static async createNode(
cachePath: string,
@ -209,6 +393,10 @@ export class CultMesh {
public static createAuthorityLeaseCatalog(): CultMeshAuthorityLeaseCatalog {
return new CultMeshAuthorityLeaseCatalog();
}
public static createStreamCatalog(): CultMeshStreamCatalog {
return new CultMeshStreamCatalog();
}
}
function requireNonEmpty(value: string, name: string): void {
@ -216,3 +404,21 @@ function requireNonEmpty(value: string, name: string): void {
throw new Error(`${name} must be non-empty.`);
}
}
function copyBudgetFor(
transport: CultMeshStreamBodyTransport,
): CultMeshStreamNegotiation["copyBudget"] {
switch (transport) {
case "shared-memory":
case "shared-d3d12-texture":
case "shared-d3d11-texture":
case "dma-buf":
case "iosurface":
case "ahardwarebuffer":
return "zero-copy-target";
case "cultcache-page":
return "one-copy-fallback";
case "inline-bytes":
return "opaque-runtime";
}
}

View file

@ -67,3 +67,64 @@ test("CultMesh TS local authority leases do not trust peer cards by contact alon
assert.equal(leases.isAuthorized(peer, "shard-primary"), true);
});
test("CultMesh TS negotiates streaming frame body transports explicitly", () => {
const streams = CultMesh.createStreamCatalog();
streams.declare({
streamId: "mimir:kiyo-pro",
verseId: "studio",
ownerPeerId: "starfire",
kind: "video",
label: "Kiyo Pro",
clock: {
clockDomainId: "starfire-qpc",
confidence: 0.25,
evidenceKind: "provisional-clock-domain-edge-fit",
},
video: {
width: 1920,
height: 1080,
pixelFormat: "YUY2",
framesPerSecond: 30,
},
preferredTransports: [
"shared-d3d12-texture",
"shared-memory",
"cultcache-page",
],
maxInFlightFrames: 3,
});
const negotiation = streams.negotiate("mimir:kiyo-pro", {
peerId: "fensalir",
verseId: "studio",
supportedTransports: ["shared-d3d12-texture", "cultcache-page"],
acceptedKinds: ["video"],
canImportGpuHandles: true,
maxInFlightFrames: 2,
});
assert.deepEqual(negotiation, {
streamId: "mimir:kiyo-pro",
producerPeerId: "starfire",
consumerPeerId: "fensalir",
transport: "shared-d3d12-texture",
access: "read",
maxInFlightFrames: 2,
copyBudget: "zero-copy-target",
});
streams.publishFrame({
streamId: "mimir:kiyo-pro",
sequence: 42n,
timestampNs: 1_000_000_000n,
durationNs: 33_333_334n,
transport: "shared-d3d12-texture",
nativeHandle: "0xfeed",
fenceHandle: "0xbeef",
fenceValue: 7n,
unavoidableCopyCount: 0,
});
assert.equal(streams.latestFrame("mimir:kiyo-pro")?.sequence, 42n);
});