feat: iteration 2 presence and session takeover control

package.json

@@ -14,7 +14,8 @@
   "main": "index.js",
   "scripts": {
     "test": "./scripts/check-proto-compabitility.sh",
-    "gen:test": "node protoc-gen-bitwise/test/generator.test.js"
+    "gen:test": "node protoc-gen-bitwise/test/generator.test.js",
+    "test:session-control": "node test/session-control.test.js"
   },
   "devDependencies": {
     "@protobuf-ts/protoc": "^2.8.1",

proto/decentraland/kernel/comms/v3/archipelago.proto

@@ -32,13 +32,29 @@ message JoinIslandMessage {
 }
 
 enum KickedReason {
+  // Authoritative logical-session supersession, not same-session socket replacement.
   KR_NEW_SESSION = 0;
+  KR_BANNED = 1;
 }
 
 message KickedMessage {
   KickedReason reason = 1;
 }
 
+enum SessionStatus {
+  UNSPECIFIED = 0;
+  TAKEOVER_PENDING = 1;
+  TAKEOVER_FAILED = 2;
+}
+
+// Session-addressed control. Success is still IslandChangedMessage with usable credentials.
+message SessionStatusMessage {
+  SessionStatus state = 1;
+  // Minimum recovery suppression interval, not a countdown to a new handshake.
+  // Pending waits on the existing authenticated connection; failed requires explicit retry.
+  uint32 retry_after_ms = 2;
+}
+
 message ServerPacket {
   oneof message {
     ChallengeResponseMessage challenge_response = 1;
@@ -47,6 +63,7 @@ message ServerPacket {
     LeftIslandMessage  left_island = 4;
     JoinIslandMessage join_island = 5;
     KickedMessage kicked = 6;
+    SessionStatusMessage session_status = 7;
   }
 }
 

proto/decentraland/pulse/pulse_presence.proto

@@ -0,0 +1,41 @@
+syntax = "proto3";
+
+package decentraland.pulse;
+
+// Published by Pulse to NATS on `engine.parcel_changes`: the platform's single presence feed.
+// One entry per (peer, realm) state change: the first non-null placement in a realm is the
+// connect signal, an entry with `parcel` absent is the exit signal for that realm/instance,
+// and every entry carries the realm and the parcel the peer now stands on.
+//
+// Guarantees the producer (Pulse) holds — see archipelago-workers/docs/contracts/iteration-2/README.md:
+//   1. A peer's first placement in a realm emits one non-null entry, even if it never moves afterwards.
+//   2. Every exit path emits exactly one parcel-absent entry (clean disconnect, PENDING_AUTH / idle
+//      timeout, duplicate-session kick, BanEnforcer eviction, PeerDefense kick). A realm change
+//      emits a non-null entry for the new realm only; the old realm is implied.
+//   3. Within one batch a wallet appears at most once, with its latest state.
+//   4. Cadence: every Presence:BatchIntervalMs (default 2000) when non-empty; a snapshot=true batch
+//      on publisher start, every Presence:SnapshotIntervalMs (default 60000), and immediately after
+//      any outbox eviction, so consumers never run on a known-lossy delta stream.
+//   5. `realm` and `address` are lowercase; `server_name` is stable for the process lifetime.
+//
+// Consumer rule: keep lastSeq[server_name]; on a gap keep serving stale state and wait for the next
+// snapshot (<= 60 s) — never drop the map. On snapshot=true replace every entry whose server_name matches.
+
+message Parcel {
+  int32 x = 1;
+  int32 y = 2;
+}
+
+message ParcelChange {
+  string address = 1;          // lowercase 0x wallet
+  string realm   = 2;          // canonical lowercase
+  optional Parcel parcel = 3;  // absent => peer left this realm/instance
+}
+
+message ParcelChangesBatch {
+  string server_name = 1;      // NatsOptions.ServerName; consumers key seq per server_name
+  uint64 seq         = 2;      // monotonic per server_name, resets on restart (snapshot=true follows)
+  bool   snapshot    = 3;      // true => full state of this server, replaces everything known for server_name
+  uint64 server_time = 4;      // unix ms
+  repeated ParcelChange changes = 5;
+}

public/comms.proto

@@ -4,3 +4,4 @@ import public "decentraland/kernel/comms/v3/archipelago.proto";
 import public "decentraland/kernel/comms/rfc4/comms.proto";
 import public "decentraland/kernel/comms/rfc5/ws_comms.proto";
 import public "decentraland/pulse/pulse_clusters.proto";
+import public "decentraland/pulse/pulse_presence.proto";

scripts/build-session-artifacts.ps1

@@ -0,0 +1,40 @@
+$ErrorActionPreference = 'Stop'
+Set-Location (Split-Path $PSScriptRoot -Parent)
+$mutex = [Threading.Mutex]::new($false, 'Local\DCL_It2_HeavyVerification')
+$acquired = $false
+try {
+  try { $acquired = $mutex.WaitOne(0) } catch [Threading.AbandonedMutexException] { $acquired = $true }
+  if (-not $acquired) { Write-Output 'Heavy verification occupied; retry later.'; exit 75 }
+  $freeGiB = (Get-CimInstance Win32_OperatingSystem).FreePhysicalMemory / 1MB
+  Write-Output "Free physical RAM: $freeGiB GiB"
+  if ($freeGiB -lt 7) { throw 'Need 5 GiB reserve plus 2 GiB Node heap headroom.' }
+  $env:NODE_OPTIONS = '--max-old-space-size=2048'
+  function Check-Exit { if ($LASTEXITCODE -ne 0) { throw "Command failed: $LASTEXITCODE" } }
+  New-Item -ItemType Directory -Force out-ts,out-js,out-session-takeover | Out-Null
+  Copy-Item node_modules/.bin/buf node_modules/.bin/buf.exe -Force
+  & node_modules/.bin/buf.exe lint proto/
+  Check-Exit
+  & node_modules/.bin/buf.exe build proto/
+  Check-Exit
+  & node_modules/.bin/buf.exe breaking proto/ --against 'https://github.com/decentraland/protocol.git#subdir=proto'
+  Check-Exit
+  foreach ($file in Get-ChildItem public/*.proto) {
+    $plugin = (Resolve-Path node_modules/.bin/protoc-gen-dcl_ts_proto.cmd).Path
+    & node_modules/.bin/protoc.cmd ("--plugin=protoc-gen-dcl_ts_proto=" + $plugin) --dcl_ts_proto_opt=esModuleInterop=true,returnObservable=false,outputServices=generic-definitions,fileSuffix=.gen,oneof=unions --dcl_ts_proto_out=out-ts -I=proto -I=public ("public/" + $file.Name)
+    Check-Exit
+  }
+  & node_modules/.bin/tsc.cmd -p tsconfig.json
+  Check-Exit
+  & node_modules/.bin/protoc.cmd --csharp_out=out-session-takeover --csharp_opt=file_extension=.gen.cs -I=proto proto/decentraland/kernel/comms/v3/archipelago.proto
+  Check-Exit
+  node test/session-control.test.js
+  Check-Exit
+  npm run gen:test
+  Check-Exit
+  npm pack --pack-destination out-session-takeover
+  Check-Exit
+  Get-FileHash out-session-takeover/Archipelago.gen.cs,out-session-takeover/dcl-protocol-1.0.0.tgz -Algorithm SHA256
+} finally {
+  if ($acquired) { $mutex.ReleaseMutex() }
+  $mutex.Dispose()
+}

scripts/test.sh

@@ -26,4 +26,6 @@ echo "> Compiling TS project"
 
 node_modules/.bin/tsc -p tsconfig.json
 
+node test/session-control.test.js
+
 rm -rf out-cs || true

session-takeover-integration.md

@@ -0,0 +1,88 @@
+# Session takeover protocol integration
+
+## Contract
+
+Package: `decentraland.kernel.comms.v3`, source `proto/decentraland/kernel/comms/v3/archipelago.proto`.
+
+| Declaration | Wire numbers |
+| --- | --- |
+| KickedReason | KR_NEW_SESSION=0 (unchanged), KR_BANNED=1 |
+| SessionStatus | UNSPECIFIED=0, TAKEOVER_PENDING=1, TAKEOVER_FAILED=2 |
+| SessionStatusMessage | state=1 (SessionStatus), retry_after_ms=2 (uint32) |
+| ServerPacket.message | session_status=7; existing tags 1–6 unchanged |
+
+TypeScript imports:
+
+```ts
+import { ServerPacket, SessionStatus, SessionStatusMessage, KickedMessage, KickedReason }
+  from '@dcl/protocol/out-js/decentraland/kernel/comms/v3/archipelago.gen'
+
+const status = { state: SessionStatus.TAKEOVER_PENDING, retryAfterMs: 1500 }
+const natsBytes = SessionStatusMessage.encode(status).finish()
+const socketBytes = ServerPacket.encode({
+  message: { $case: 'sessionStatus', sessionStatus: status }
+}).finish()
+```
+
+Gatekeeper publishes bare status to `engine.peer.<wallet>.session_status.<session>` and bare
+KickedMessage to `engine.peer.<wallet>.kicked.<session>`. Connector authenticates and targets
+exact wallet/session, supplies the envelope, and keeps pending sockets open. Success is the existing
+island_changed. Control events bypass assignment deduplication. No wallet-only control fallback.
+
+Pending suppresses recovery for at least retryAfterMs; expiration is not permission to handshake.
+Failure withholds credentials and requires user retry. KR_NEW_SESSION means authoritative logical
+session supersession, not replacing another socket for the same session. Same-session socket
+replacement must close only the old transport or await a separately approved reason. Unknown kick
+reasons stop automatic retries but are not confirmed supersession. Legacy empty KickedMessage still
+means KR_NEW_SESSION. Unknown binary enum numbers remain numbers in the generated JS decoder;
+JSON conversions may map them to UNRECOGNIZED. Consumers must use an explicit switch/default.
+
+## Reproduce artifacts on Windows
+
+From this worktree, install dependencies with the shared heavy-verification mutex and RAM guard.
+Use Node 24, protoc 22.2 (package config), @dcl/ts-proto 1.154.0, protobufjs 7.2.4, and Buf 1.8.0.
+Provide the Windows Buf 1.8.0 binary at `node_modules/.bin/buf` (the existing local install has it).
+Then run:
+
+```powershell
+powershell -NoProfile -ExecutionPolicy Bypass -File scripts/build-session-artifacts.ps1
+```
+
+The script acquires the mutex nonblocking (exit 75 means retry later), runs Buf checks, generates all
+public TS entrypoints, compiles JS, generates Archipelago.gen.cs directly with protoc, runs the wire
+and bitwise goldens, and packs a local archive. Outputs:
+
+- `out-session-takeover/dcl-protocol-1.0.0.tgz`
+- `out-session-takeover/Archipelago.gen.cs`
+- Reusable goldens: `test/fixtures/session-control.json` (hex bytes and canonical JSON).
+
+No registry version is reserved: the archive's 1.0.0 is the repository's development version, not a
+new published stable release. Identify it by source commit and SHA256. Consumer owners may install
+the absolute archive locally with `npm install --no-save --package-lock=false <archive>` (under the
+same mutex) in their isolated candidate. Keep temporary local paths out of committed manifests and
+locks. Verify the installed generated exports before testing. Do not broaden unrelated dependency
+pins or social-service's isolated alias.
+
+For Unity, copy the generated Archipelago.gen.cs to the existing generated protocol location after
+checking local changes; keep its existing .meta file. It uses the existing Google.Protobuf runtime and
+Vectors generated definitions. No handwritten parallel protocol classes are needed.
+
+## CI and release gates
+
+Linux reproduction after installing dependencies: `make buf-breaking`, `make test`,
+`npm run gen:test`, `npm pack`; direct C# regeneration uses protoc 22.2 with
+`--csharp_out=<output> --csharp_opt=file_extension=.gen.cs -I=proto
+proto/decentraland/kernel/comms/v3/archipelago.proto`.
+
+Before publication, CI needs either an approved immutable artifact upload keyed to the reviewed
+protocol commit and SHA256, or a checkout/build of that exact protocol commit followed by a local
+archive install. Local filesystem paths cannot serve as remote CI pins. This pass does not upload or
+publish. Stable publication and coordinated consumer lock updates require separate approval.
+
+Required integration gates: all ban producers use KR_BANNED; same-session socket replacement does
+not latch logical supersession; pending/failure reach only the winning session; authoritative kick
+latches suppression before transport close; stale asynchronous callbacks cannot issue credentials or
+late control events; old readers ignore status and retain parking compatibility; two-client tests
+exercise lost status, reconnect replay, timeout/watchdog, and late assignment after supersession.
+LiveKit Cloud revocation/removal and next-second cutoff behavior require actual Cloud evidence;
+protocol goldens do not establish that gate.

session-takeover-report.md

@@ -0,0 +1,65 @@
+# Session takeover amendment — Protocol report
+
+Base: `a1f2f3db4147b3689748af205c0923304ccc5c6a`, preserved on `codex/it2-sep15-protocol`.
+Binding plan: `Claude outputs/session-takeover-client-plan.md`, September 15.
+
+## Implementation
+
+Added KR_BANNED=1, SessionStatus values UNSPECIFIED=0 / TAKEOVER_PENDING=1 /
+TAKEOVER_FAILED=2, SessionStatusMessage (state=1, retry_after_ms=2), and
+ServerPacket.session_status=7. Existing fields and enum values retain their numbers.
+C0/C1 Pulse schemas and exports have no diff against the base.
+
+Committed regression goldens cover legacy welcome/assignment/supersession bytes, bare NATS and
+wrapped WS pending/failure/ban, unspecified status, defaults, unknown binary enum values, and
+old-reader tolerance of field 7. Normal scripts/test.sh now executes the generated-JS wire tests.
+scripts/build-session-artifacts.ps1 supplies Windows reproduction and holds the shared mutex for
+generation, compilation, tests and packing. Integration instructions are in session-takeover-integration.md.
+
+## RED/GREEN
+
+RED: `node test/session-control.test.js` against the pre-amendment generated module exited 1 at
+`bare ban: decode`: numeric reason 1 decoded but the old named KR_BANNED mapped to -1. The four
+preceding legacy cases passed. The test and fixtures were written before the schema amendment.
+
+Initial Windows build-script attempts exposed executable extension, protoc include-path, and npm
+plugin path issues. These were corrected in the build helper; they are not schema regressions.
+Mutex contention caused a nonblocking skip, followed by useful documentation work before retry.
+
+GREEN: `powershell -NoProfile -ExecutionPolicy Bypass -File scripts/build-session-artifacts.ps1`
+exited 0: Buf lint/build/breaking against current main; all public TS codegen and tsc compilation;
+direct protoc C# generation; 11 wire goldens/defaults/unknown values/old-reader checks;
+two existing bitwise goldens; local npm pack (536 files). About 39 GiB RAM was free at start.
+
+Artifacts in `out-session-takeover/`:
+
+- `Archipelago.gen.cs`: SHA256 `9896bebec52112e53212b4368b08d97af80d16adc1b9777ea3b57589263e7b17`
+- `dcl-protocol-1.0.0.tgz`: SHA256 `fa3f2c51c019d729d81a0c3993b384243ac615cc33def51479f797969c29b287`
+
+No runtime application or
+LiveKit Cloud tests are claimed here. C# is generated by protoc; consumer Unity compilation remains
+an integration gate.
+
+## Changed files
+
+- proto/decentraland/kernel/comms/v3/archipelago.proto
+- package.json
+- scripts/test.sh
+- scripts/build-session-artifacts.ps1
+- test/session-control.test.js
+- test/fixtures/session-control.json
+- session-takeover-integration.md
+- session-takeover-report.md
+
+## Gates and ownership
+
+No protocol collision was found. Connector identified existing KR_NEW_SESSION use for same-session
+socket replacement: that must become close-only transport replacement or receive a separately
+approved wire reason, since logical supersession is terminal. This decision was communicated to
+connector and Unity owners; no extra enum was silently added. Protocol contains no ban producers.
+
+Consumer gate: audit bans, authoritative-only supersession, exact session subjects, replay and stale
+operation fencing, client suppression before close, watchdog behavior, old-client compatibility,
+and a real two-client LiveKit Cloud revocation scenario. The local archive is not a stable release.
+Remote CI must consume an immutable reviewed commit/artifact, followed by approved stable
+publication and coordinated pins. No push, publication, remote PR edit, or consumer re-pin occurred.

test/fixtures/session-control.json

@@ -0,0 +1,13 @@
+[
+  { "name": "legacy welcome", "type": "ServerPacket", "hex": "12030a0170", "json": { "welcome": { "peerId": "p" } } },
+  { "name": "legacy assignment", "type": "ServerPacket", "hex": "1a060a0169120163", "json": { "islandChanged": { "islandId": "i", "connStr": "c" } } },
+  { "name": "legacy supersession envelope", "type": "ServerPacket", "hex": "3200", "json": { "kicked": { "reason": "KR_NEW_SESSION" } } },
+  { "name": "legacy bare supersession", "type": "KickedMessage", "hex": "", "json": { "reason": "KR_NEW_SESSION" } },
+  { "name": "bare ban", "type": "KickedMessage", "hex": "0801", "json": { "reason": "KR_BANNED" } },
+  { "name": "ban envelope", "type": "ServerPacket", "hex": "32020801", "json": { "kicked": { "reason": "KR_BANNED" } } },
+  { "name": "bare pending 1500ms", "type": "SessionStatusMessage", "hex": "080110dc0b", "json": { "state": "TAKEOVER_PENDING", "retryAfterMs": 1500 } },
+  { "name": "pending envelope", "type": "ServerPacket", "hex": "3a05080110dc0b", "json": { "sessionStatus": { "state": "TAKEOVER_PENDING", "retryAfterMs": 1500 } } },
+  { "name": "bare failure", "type": "SessionStatusMessage", "hex": "0802", "json": { "state": "TAKEOVER_FAILED" } },
+  { "name": "failure envelope", "type": "ServerPacket", "hex": "3a020802", "json": { "sessionStatus": { "state": "TAKEOVER_FAILED" } } },
+  { "name": "unspecified envelope", "type": "ServerPacket", "hex": "3a00", "json": { "sessionStatus": {} } }
+]

test/session-control.test.js

@@ -0,0 +1,25 @@
+const assert = require('node:assert/strict')
+const { ServerPacket, KickedMessage, SessionStatusMessage, SessionStatus, KickedReason } = require('../out-js/decentraland/kernel/comms/v3/archipelago.gen')
+const fixtures = require('./fixtures/session-control.json')
+for (const fixture of fixtures) {
+  const type = { ServerPacket, KickedMessage, SessionStatusMessage }[fixture.type]
+  assert.ok(type, fixture.type + ' must be generated')
+  const bytes = Buffer.from(fixture.hex, 'hex')
+  const expected = type.fromJSON(fixture.json)
+  assert.deepEqual(type.decode(bytes), expected, fixture.name + ': decode')
+  assert.deepEqual(Buffer.from(type.encode(expected).finish()), bytes, fixture.name + ': encode')
+}
+assert.equal(KickedReason.KR_NEW_SESSION, 0)
+assert.equal(KickedReason.KR_BANNED, 1)
+assert.equal(SessionStatus.UNSPECIFIED, 0)
+assert.equal(SessionStatusMessage.decode(Buffer.alloc(0)).state, SessionStatus.UNSPECIFIED)
+assert.equal(KickedMessage.decode(Buffer.alloc(0)).reason, KickedReason.KR_NEW_SESSION)
+// Unknown binary enums must not be interpreted as confirmed supersession.
+assert.equal(KickedMessage.decode(Buffer.from('0863', 'hex')).reason, 99)
+assert.equal(SessionStatusMessage.decode(Buffer.from('0863', 'hex')).state, 99)
+// A pre-amendment reader ignores the new oneof tag rather than interpreting it as an assignment.
+const protobuf = require('protobufjs')
+const oldPacket = protobuf.parse('syntax="proto3"; message Kicked { int32 reason=1; } message OldPacket { oneof message { Kicked kicked=6; } }').root.lookupType('OldPacket')
+assert.equal(oldPacket.decode(Buffer.from('3a05080110dc0b', 'hex')).message, undefined)
+assert.equal(oldPacket.decode(Buffer.from('3200', 'hex')).kicked.reason, 0)
+console.log('PASS: ' + fixtures.length + ' wire goldens, defaults and unknown enums')