From 4622c9bfa7a88b9aa822070a6becff0b225febc4 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 10:02:07 +0100 Subject: [PATCH] Add Kotlin catalog watch ergonomics --- docs/cultnet-transport-parity.md | 6 +- packages/cultmesh-kotlin/README.md | 13 +++ .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 109 ++++++++++++------ 3 files changed, 91 insertions(+), 37 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 985397a3..9e0b83a1 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -287,7 +287,8 @@ Current progress: - Kotlin now mirrors the CultMesh stream catalog shape for Android-adjacent media, sensor, tensor, and byte streams: stream declaration, clock metadata, preferred body transports, consumer negotiation, copy-budget classification, - max in-flight frame pressure, and latest-frame cursors. + max in-flight frame pressure, latest-frame cursors, and watch callbacks for + stream declarations plus latest-frame publication. - Kotlin now has a branded `CultMesh` facade matching the C#/TypeScript/Python entrypoint pattern for local nodes, Verse catalogs, peer catalogs, authority lease catalogs, stream catalogs, and RUDP client/server construction. Kotlin @@ -295,6 +296,9 @@ Current progress: client directly from a peer card contact hint. The facade delegates to the same owners as the top-level helpers instead of creating a second authority path. +- Kotlin authority lease catalogs now also expose watch callbacks, so JVM and + Android clients can react to lease changes without treating peer-card contact + hints as trust. - Rust now has a small `CultMesh` facade for its substrate role: peer cards, sorted peer catalogs, authority lease catalogs, `rudp://host:port` endpoint parsing, and RUDP client/server construction from endpoint or peer-card diff --git a/packages/cultmesh-kotlin/README.md b/packages/cultmesh-kotlin/README.md index 88536d75..98015d31 100644 --- a/packages/cultmesh-kotlin/README.md +++ b/packages/cultmesh-kotlin/README.md @@ -278,6 +278,9 @@ val peer = CultMeshPeerCard( ) val leases = CultMesh.createAuthorityLeaseCatalog() +val unsubscribeLeaseWatch = leases.watch { lease -> + println("authority lease changed: ${lease.leaseId}") +} leases.upsert( CultMeshAuthorityLease( leaseId = "lease:kotlin-peer", @@ -292,6 +295,7 @@ leases.upsert( ) check(leases.isAuthorized(peer, "shard-primary", "players")) +unsubscribeLeaseWatch() ``` ## Streaming Catalog @@ -303,6 +307,12 @@ cursors; it does not own the frame bytes themselves. ```kotlin val streams = CultMesh.createStreamCatalog() +val unsubscribeStreamWatch = streams.watch { stream -> + println("stream declared: ${stream.streamId}") +} +val unsubscribeFrameWatch = streams.watchFrames { frame -> + println("latest frame ${frame.sequence} for ${frame.streamId}") +} streams.declare( CultMeshStreamDescriptor( streamId = "mimir:camera", @@ -334,6 +344,9 @@ val lane = streams.negotiate( maxInFlightFrames = 2, ), ) + +unsubscribeStreamWatch() +unsubscribeFrameWatch() ``` `EveMediaObservationDocument` carries byte-backed device streams such as camera diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index fa1d7d0a..63389c28 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -1063,12 +1063,19 @@ data class CultMeshAuthorityLease( class CultMeshAuthorityLeaseCatalog { private val knownLeases = linkedMapOf() + private val subscribers = mutableListOf<(CultMeshAuthorityLease) -> Unit>() val leases: List get() = knownLeases.toSortedMap().values.toList() + fun watch(callback: (CultMeshAuthorityLease) -> Unit): () -> Unit { + subscribers.add(callback) + return { subscribers.remove(callback) } + } + fun upsert(lease: CultMeshAuthorityLease) { knownLeases[lease.leaseId] = lease + subscribers.toList().forEach { it(lease) } } fun get(leaseId: String): CultMeshAuthorityLease? { @@ -1222,12 +1229,25 @@ data class CultMeshStreamFrameHandle( class CultMeshStreamCatalog { private val knownStreams = linkedMapOf() private val latestFrames = linkedMapOf() + private val streamSubscribers = mutableListOf<(CultMeshStreamDescriptor) -> Unit>() + private val frameSubscribers = mutableListOf<(CultMeshStreamFrameHandle) -> Unit>() val streams: List get() = knownStreams.toSortedMap().values.toList() + fun watch(callback: (CultMeshStreamDescriptor) -> Unit): () -> Unit { + streamSubscribers.add(callback) + return { streamSubscribers.remove(callback) } + } + + fun watchFrames(callback: (CultMeshStreamFrameHandle) -> Unit): () -> Unit { + frameSubscribers.add(callback) + return { frameSubscribers.remove(callback) } + } + fun declare(stream: CultMeshStreamDescriptor): CultMeshStreamDescriptor { knownStreams[stream.streamId] = stream + streamSubscribers.toList().forEach { it(stream) } return stream } @@ -1265,6 +1285,7 @@ class CultMeshStreamCatalog { fun publishFrame(frame: CultMeshStreamFrameHandle): CultMeshStreamFrameHandle { if (frame.streamId !in knownStreams) throw IOException("Unknown CultMesh stream '${frame.streamId}'") latestFrames[frame.streamId] = frame + frameSubscribers.toList().forEach { it(frame) } return frame } @@ -3048,6 +3069,8 @@ private fun cultMeshAuthorityLeasesGatePeerTrust() { val validFrom = Instant.parse("2026-06-15T00:00:00Z") val expiresAt = Instant.parse("2026-06-15T01:00:00Z") val duringLease = Instant.parse("2026-06-15T00:30:00Z") + var watchedLease: CultMeshAuthorityLease? = null + val unsubscribeLease = leases.watch { watchedLease = it } check(!leases.isAuthorized(peer, "shard-primary", "players", duringLease)) leases.upsert( @@ -3063,6 +3086,10 @@ private fun cultMeshAuthorityLeasesGatePeerTrust() { ), ) + check(watchedLease?.leaseId == "lease:peer-authority") + unsubscribeLease() + leases.upsert(watchedLease!!.copy(signature = "after-unsubscribe")) + check(watchedLease?.roles == listOf("shard-primary")) check(leases.get("lease:peer-authority")?.issuerRuntimeId == "kotlin-authority") check(leases.leases.map { it.leaseId } == listOf("lease:peer-authority")) check(leases.isAuthorized(peer, "shard-primary", "players", duringLease)) @@ -3073,32 +3100,36 @@ private fun cultMeshAuthorityLeasesGatePeerTrust() { private fun cultMeshStreamCatalogNegotiatesBodyTransports() { val streams = createStreamCatalog() - streams.declare( - CultMeshStreamDescriptor( - streamId = "mimir:kiyo-pro", - verseId = "studio", - ownerPeerId = "starfire", - kind = CultMeshStreamKinds.Video, - label = "Kiyo Pro", - clock = CultMeshStreamClock( - clockDomainId = "starfire-qpc", - confidence = 0.25, - evidenceKind = "provisional-clock-domain-edge-fit", - ), - video = CultMeshVideoStreamFormat( - width = 1920, - height = 1080, - pixelFormat = "YUY2", - framesPerSecond = 30.0, - ), - preferredTransports = listOf( - CultMeshStreamBodyTransports.SharedD3D12Texture, - CultMeshStreamBodyTransports.SharedMemory, - CultMeshStreamBodyTransports.CultCachePage, - ), - maxInFlightFrames = 3, + var watchedStream: CultMeshStreamDescriptor? = null + var watchedFrame: CultMeshStreamFrameHandle? = null + val unsubscribeStream = streams.watch { watchedStream = it } + val unsubscribeFrame = streams.watchFrames { watchedFrame = it } + val stream = CultMeshStreamDescriptor( + streamId = "mimir:kiyo-pro", + verseId = "studio", + ownerPeerId = "starfire", + kind = CultMeshStreamKinds.Video, + label = "Kiyo Pro", + clock = CultMeshStreamClock( + clockDomainId = "starfire-qpc", + confidence = 0.25, + evidenceKind = "provisional-clock-domain-edge-fit", ), + video = CultMeshVideoStreamFormat( + width = 1920, + height = 1080, + pixelFormat = "YUY2", + framesPerSecond = 30.0, + ), + preferredTransports = listOf( + CultMeshStreamBodyTransports.SharedD3D12Texture, + CultMeshStreamBodyTransports.SharedMemory, + CultMeshStreamBodyTransports.CultCachePage, + ), + maxInFlightFrames = 3, ) + streams.declare(stream) + check(watchedStream?.streamId == "mimir:kiyo-pro") val negotiation = streams.negotiate( "mimir:kiyo-pro", @@ -3124,20 +3155,26 @@ private fun cultMeshStreamCatalogNegotiatesBodyTransports() { check(negotiation.copyBudget == "zero-copy-target") check(streams.find("studio", CultMeshStreamKinds.Video).single().streamId == "mimir:kiyo-pro") - streams.publishFrame( - CultMeshStreamFrameHandle( - streamId = "mimir:kiyo-pro", - sequence = 42, - timestampNs = 1_000_000_000, - durationNs = 33_333_334, - transport = negotiation.transport, - nativeHandle = "0xfeed", - fenceHandle = "0xbeef", - fenceValue = 7, - unavoidableCopyCount = 0, - ), + val frame = CultMeshStreamFrameHandle( + streamId = "mimir:kiyo-pro", + sequence = 42, + timestampNs = 1_000_000_000, + durationNs = 33_333_334, + transport = negotiation.transport, + nativeHandle = "0xfeed", + fenceHandle = "0xbeef", + fenceValue = 7, + unavoidableCopyCount = 0, ) + streams.publishFrame(frame) + check(watchedFrame?.sequence == 42L) check(streams.latestFrame("mimir:kiyo-pro")?.sequence == 42L) + unsubscribeStream() + unsubscribeFrame() + streams.declare(stream.copy(streamId = "mimir:kiyo-pro-alt")) + streams.publishFrame(frame.copy(sequence = 43)) + check(watchedStream?.streamId == "mimir:kiyo-pro") + check(watchedFrame?.sequence == 42L) } private fun cultNetShardCatalogsAndLogsRoundTrip() {