extends SceneTree
## Layer 10 — Full E2E with keet-daemon
##
## Tests the complete production flow:
## - Phase 1: Connect to keet-daemon via UDS
## - Phase 2: Peer 0 creates room + invite, peers 1+2 join
## - Phase 3: Mesh discovery via swarm_join (room-scoped topic)
## - Phase 4: Role exchange + authority election (room creator = admin = authority)
## - Phase 5: Durable state round-trip via KeetGameState
##
## Usage: godot4 --headless --path project --script res://layer10_e2e.gd -- --peer-index 0 --socket-path /tmp/keet-daemon.sock
const TIMEOUT := 60.0
const POLL_HZ := 0.05
const PEER_COUNT := 3
var _peer_index := 0
var _socket_path := "/tmp/keet-daemon.sock"
var _peer: Object # HyperswarmMultiplayerPeer
var _uds: Node # KeetUdsClient (loaded dynamically)
var _game_state: Node # KeetGameState (loaded dynamically)
var _dht_kp: Dictionary
var _dht_pk: PackedByteArray
var _my_pk_hex := ""
var _my_role := ""
var _bootstrap: PackedStringArray
var _connected_peers: Dictionary = {} # peer_id -> true
var _peer_registry: Dictionary = {} # pk_hex -> {peer_id, role, role_priority}
var _roles_received: Dictionary = {} # pk_hex -> role
var _authority_pk := ""
var _room_id := ""
func _init() -> void:
var args := OS.get_cmdline_user_args()
for i in args.size():
if args[i] == "--peer-index" and i + 1 < args.size():
_peer_index = args[i + 1].to_int()
elif args[i] == "--socket-path" and i + 1 < args.size():
_socket_path = args[i + 1]
_write_status("STARTING")
_start.call_deferred()
func _start() -> void:
_log("starting peer index %d, socket=%s" % [_peer_index, _socket_path])
# Read bootstrap
var f := FileAccess.open("/tmp/layer_bootstrap.txt", FileAccess.READ)
if f == null:
_fail("Cannot read bootstrap")
return
_bootstrap = PackedStringArray(f.get_as_text().strip_edges().split(","))
f.close()
# ---- Phase 1: Daemon Connect ----
_write_status("PHASE1")
_log("Phase 1: connecting to daemon...")
# Load scripts dynamically (class_name not available in headless)
var uds_script = load("res://scripts/keet_uds_client.gd")
if uds_script == null:
uds_script = load("res://../scripts/keet_uds_client.gd")
if uds_script == null:
_fail("Phase 1: cannot load keet_uds_client.gd")
return
var gs_script = load("res://scripts/keet_game_state.gd")
if gs_script == null:
gs_script = load("res://../scripts/keet_game_state.gd")
if gs_script == null:
_fail("Phase 1: cannot load keet_game_state.gd")
return
# Create UDS client as child of root
_uds = Node.new()
_uds.set_script(uds_script)
_uds.set("socket_path", _socket_path)
_uds.set("auto_connect", false)
root.add_child(_uds)
# Create game state node
_game_state = Node.new()
_game_state.set_script(gs_script)
_game_state.set("client", _uds)
_game_state.set("auto_poll", false) # we'll poll manually
root.add_child(_game_state)
# Wait for _ready() to fire
await create_timer(0.1).timeout
# Connect to daemon
_uds.try_connect()
var elapsed := 0.0
while elapsed < 15.0 and not _uds.is_connected:
await create_timer(0.1).timeout
elapsed += 0.1
if not _uds.is_connected:
_fail("Phase 1: daemon not connected after 15s")
return
# Verify daemon status
var status_resp = await _uds.rpc_async("status")
if status_resp.error != null:
_fail("Phase 1: status RPC failed: %s" % str(status_resp.error))
return
_log("Phase 1 PASS — daemon connected (v=%s)" % str(status_resp.result.get("daemonVersion", "?")))
_write_status("PHASE1_OK")
# ---- Phase 2: Room Setup ----
_write_status("PHASE2")
_log("Phase 2: room setup...")
if _peer_index == 0:
# Creator: make a new room
var create_resp = await _uds.rpc_async("room.create", {"title": "L10 Test Room"})
if create_resp.error != null:
_fail("Phase 2: room.create failed: %s" % str(create_resp.error))
return
# room.create returns a string (room ID), not a Dictionary
if create_resp.result is Dictionary:
_room_id = str(create_resp.result.get("roomId", create_resp.result.get("id", "")))
else:
_room_id = str(create_resp.result)
if _room_id.is_empty():
_fail("Phase 2: room.create returned no roomId")
return
_log("room created: %s" % _room_id.left(16))
_write_file("/tmp/layer10_room_id.txt", _room_id)
# Generate invite for other peers
var invite_resp = await _uds.rpc_async("room.invite", {"roomId": _room_id})
if invite_resp.error != null:
_fail("Phase 2: room.invite failed: %s" % str(invite_resp.error))
return
# room.invite returns {token: "..."}, not {invite: "..."}
var invite: String = ""
if invite_resp.result is Dictionary:
invite = str(invite_resp.result.get("token", invite_resp.result.get("invite", "")))
else:
invite = str(invite_resp.result)
if invite.is_empty():
_fail("Phase 2: room.invite returned no invite")
return
_write_file("/tmp/layer10_invite.txt", invite)
_log("invite written")
# Room creator is admin
_my_role = "admin"
else:
# Joiner: wait for room_id and invite
_log("waiting for room_id and invite...")
elapsed = 0.0
var invite: String = ""
while elapsed < 30.0:
var rf := FileAccess.open("/tmp/layer10_room_id.txt", FileAccess.READ)
var invf := FileAccess.open("/tmp/layer10_invite.txt", FileAccess.READ)
if rf != null and invf != null:
_room_id = rf.get_as_text().strip_edges()
invite = invf.get_as_text().strip_edges()
rf.close()
invf.close()
break
if rf != null:
rf.close()
if invf != null:
invf.close()
await create_timer(0.5).timeout
elapsed += 0.5
if _room_id.is_empty() or invite.is_empty():
_fail("Phase 2: timeout waiting for room_id/invite")
return
# Join the room
var join_resp = await _uds.rpc_async("room.join", {"invite": invite})
if join_resp.error != null:
# room.join might fail if already joined — check if we have the room
var rooms_resp = await _uds.rpc_async("rooms")
var found := false
if rooms_resp.error == null and rooms_resp.result is Array:
for room in rooms_resp.result:
if room is Dictionary and str(room.get("roomId", "")) == _room_id:
found = true
break
if not found:
_fail("Phase 2: room.join failed: %s" % str(join_resp.error))
return
_log("room already joined (ok)")
else:
_log("room joined: %s" % _room_id.left(16))
# Joiners are "peer" role
_my_role = "peer"
_write_file("/tmp/layer10_peer_%d_role.txt" % _peer_index, _my_role)
_log("Phase 2 PASS — room ready, role=%s" % _my_role)
_write_status("PHASE2_OK")
# ---- Phase 3: Mesh Discovery ----
_write_status("PHASE3")
_log("Phase 3: mesh discovery via swarm_join...")
# Create HyperswarmMultiplayerPeer
_peer = ClassDB.instantiate(&"HyperswarmMultiplayerPeer")
# Generate DHT keypair
_dht_kp = _peer.generate_dht_keypair()
_dht_pk = _dht_kp["public_key"]
_my_pk_hex = _dht_pk.hex_encode()
_log("dht pk: %s" % _my_pk_hex.left(12))
_write_file("/tmp/layer10_peer_%d_pk.txt" % _peer_index, _my_pk_hex)
# Connect signals
_peer.peer_connected.connect(_on_peer_connected)
_peer.peer_disconnected.connect(_on_peer_disconnected)
# Join DHT
var err = _peer.join_dht(_dht_pk, _dht_kp["secret_key"], _bootstrap)
if err != OK:
_fail("Phase 3: join_dht: %d" % err)
return
# Wait for DHT active
elapsed = 0.0
while elapsed < 10.0:
_peer.poll()
if _peer.get_dht_status().get("active", false):
break
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if not _peer.get_dht_status().get("active", false):
_fail("Phase 3: DHT not active after 10s")
return
_log("DHT active")
# Sync point: wait for all peers ready before swarm_join
_write_file("/tmp/layer10_peer_%d_ready.txt" % _peer_index, "ready")
elapsed = 0.0
while elapsed < 15.0:
var all_ready := true
for pi in range(PEER_COUNT):
var readyf := FileAccess.open("/tmp/layer10_peer_%d_ready.txt" % pi, FileAccess.READ)
if readyf == null:
all_ready = false
break
readyf.close()
if all_ready:
break
_peer.poll()
await create_timer(0.1).timeout
elapsed += 0.1
# Topic derived from room_id (same as hyper-tps does)
var topic_str := "keet-game:%s" % _room_id
var topic := topic_str.sha256_buffer()
_log("room_id='%s' topic_hex=%s" % [_room_id, topic.hex_encode().left(16)])
err = _peer.swarm_join(topic, true, true)
if err != OK:
_fail("Phase 3: swarm_join: %d" % err)
return
_log("swarm_join called, bootstrap=%s" % str(_bootstrap))
# Wait for full mesh
elapsed = 0.0
var _last_log := 0.0
while elapsed < TIMEOUT and _connected_peers.size() < PEER_COUNT - 1:
_peer.poll()
while _peer.get_available_packet_count() > 0:
_peer.get_packet()
if elapsed - _last_log > 5.0:
var dht_st: Dictionary = _peer.get_dht_status()
var stats: Dictionary = _peer.get_stats()
_log("mesh wait %.0fs: connected=%d/%d dht=%s stats=%s" % [
elapsed, _connected_peers.size(), PEER_COUNT - 1,
str(dht_st), str(stats)])
_last_log = elapsed
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if _connected_peers.size() < PEER_COUNT - 1:
_fail("Phase 3: only %d/%d peers connected" % [_connected_peers.size(), PEER_COUNT - 1])
return
# Verify mesh mode (is_server == false)
var mesh_stats: Dictionary = _peer.get_stats()
if mesh_stats.get("is_server", false):
_fail("Phase 3: is_server = true (mesh mode broken)")
return
_log("Phase 3 PASS — full mesh (%d connections), is_server=false" % _connected_peers.size())
_write_status("PHASE3_OK")
# ---- Phase 4: Role Exchange + Authority Election ----
_write_status("PHASE4")
_log("Phase 4: role exchange + authority election...")
# Register self
_peer_registry[_my_pk_hex] = {
"peer_id": 1,
"role": _my_role,
"role_priority": _role_priority(_my_role)
}
_roles_received[_my_pk_hex] = _my_role
# Broadcast role
var role_msg := ("role:%s:%s" % [_my_pk_hex, _my_role]).to_utf8_buffer()
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(role_msg)
# Wait for all roles
elapsed = 0.0
while elapsed < TIMEOUT and _roles_received.size() < PEER_COUNT:
_peer.poll()
if int(elapsed * 20) % 10 == 0 and elapsed > 0.1:
_peer.set_target_peer(0)
_peer.set_transfer_mode(MultiplayerPeer.TRANSFER_MODE_RELIABLE)
_peer.put_packet(role_msg)
while _peer.get_available_packet_count() > 0:
var pkt: PackedByteArray = _peer.get_packet()
var text: String = pkt.get_string_from_utf8()
if text.begins_with("role:"):
_handle_role_announcement(text)
await create_timer(POLL_HZ).timeout
elapsed += POLL_HZ
if _roles_received.size() < PEER_COUNT:
_fail("Phase 4: only received roles from %d/%d peers" % [_roles_received.size(), PEER_COUNT])
return
# Elect authority
_elect_authority()
if _authority_pk.is_empty():
_fail("Phase 4: authority election returned empty")
return
_write_file("/tmp/layer10_peer_%d_authority.txt" % _peer_index, _authority_pk)
# Verify admin (peer 0) is authority
var pk0 := ""
for _attempt in range(20):
var pk0f := FileAccess.open("/tmp/layer10_peer_0_pk.txt", FileAccess.READ)
if pk0f != null:
pk0 = pk0f.get_as_text().strip_edges()
pk0f.close()
break
await create_timer(0.1).timeout
if _authority_pk != pk0:
_fail("Phase 4: authority=%s, expected admin pk=%s" % [_authority_pk.left(12), pk0.left(12)])
return
_log("Phase 4 PASS — authority is admin (peer 0): %s" % _authority_pk.left(12))
_write_status("PHASE4_OK")
# ---- Phase 5: Durable State Round-Trip ----
_write_status("PHASE5")
_log("Phase 5: durable state round-trip...")
# Start watching room state
_game_state.watch(_room_id)
if _peer_index == 0:
# Authority writes game state — direct RPC to bypass KeetGameState layer
_log("writing durable state as authority (direct RPC)...")
var meta_payload := {
"game:map": "arena",
"game:mode": "deathmatch",
"game:max_players": "8"
}
_log("meta.set payload: roomId=%s metadata=%s" % [_room_id.left(16), str(meta_payload)])
var set_resp = await _uds.rpc_async("meta.set", {
"roomId": _room_id,
"metadata": meta_payload
})
if set_resp.error != null:
_fail("Phase 5: meta.set failed: %s" % str(set_resp.error))
return
_log("meta.set response: %s" % str(set_resp.result).left(300))
# Verify what was written
await create_timer(0.5).timeout
var verify_resp = await _uds.rpc_async("meta.list", {})
_log("post-write meta.list: %s" % str(verify_resp.result).left(500))
# Signal that state is written
_write_file("/tmp/layer10_peer_0_state_written.txt", "done")
_log("durable state written")
else:
# Wait for authority to write state
_log("waiting for authority to write state...")
elapsed = 0.0
while elapsed < 20.0:
var sf := FileAccess.open("/tmp/layer10_peer_0_state_written.txt", FileAccess.READ)
if sf != null:
sf.close()
break
await create_timer(0.5).timeout
elapsed += 0.5
# Poll for state via direct RPC
_log("polling for durable state (direct RPC)...")
var map_val := ""
var mode_val := ""
elapsed = 0.0
while elapsed < 20.0:
var poll_resp = await _uds.rpc_async("meta.list", {})
if poll_resp.error == null and poll_resp.result is Dictionary:
var rmeta: Variant = poll_resp.result.get(_room_id, null)
if rmeta is Dictionary:
map_val = str(rmeta.get("game:map", ""))
mode_val = str(rmeta.get("game:mode", ""))
_log("poll %.0fs: keys=%s map='%s' mode='%s'" % [elapsed, str(rmeta.keys()), map_val, mode_val])
else:
_log("poll %.0fs: room not in meta.list (keys=%s)" % [elapsed, str(poll_resp.result.keys())])
else:
_log("poll %.0fs: error=%s" % [elapsed, str(poll_resp.error)])
if map_val == "arena" and mode_val == "deathmatch":
break
await create_timer(1.0).timeout
elapsed += 1.0
if map_val != "arena" or mode_val != "deathmatch":
_fail("Phase 5: state mismatch — map=%s mode=%s (expected arena, deathmatch)" % [map_val, mode_val])
return
var max_val := ""
var rmeta2 = await _uds.rpc_async("meta.list", {})
if rmeta2.error == null and rmeta2.result is Dictionary:
var rm: Variant = rmeta2.result.get(_room_id, {})
if rm is Dictionary:
max_val = str(rm.get("game:max_players", "?"))
var state_summary := "map=%s,mode=%s,max_players=%s" % [map_val, mode_val, max_val]
_write_file("/tmp/layer10_peer_%d_durable.txt" % _peer_index, state_summary)
_log("durable state verified: %s" % state_summary)
_log("Phase 5 PASS — durable state round-trip verified")
_write_status("PHASE5_OK")
# All done
_game_state.unwatch()
_write_status("PASS")
_log("ALL PHASES PASSED")
_peer.close()
quit()
# ---- Role Handling ----
func _handle_role_announcement(text: String) -> void:
var parts := text.split(":")
if parts.size() < 3:
return
var pk_hex := parts[1]
var role := parts[2]
if pk_hex == _my_pk_hex:
return
if _roles_received.has(pk_hex):
return
_roles_received[pk_hex] = role
_log("role received: %s=%s (total: %d/%d)" % [pk_hex.left(12), role, _roles_received.size(), PEER_COUNT])
var peer_id := 0
for pid in _connected_peers:
peer_id = pid
if not _peer_id_in_registry(pid):
break
_peer_registry[pk_hex] = {
"peer_id": peer_id,
"role": role,
"role_priority": _role_priority(role)
}
func _peer_id_in_registry(pid: int) -> bool:
for pk_hex in _peer_registry:
if _peer_registry[pk_hex]["peer_id"] == pid:
return true
return false
# ---- Authority Election (matches hyper-tps menu_main.gd) ----
func _elect_authority() -> void:
var candidates: Array = []
if _my_role != "read-only" and not _my_pk_hex.is_empty():
candidates.append({
"pk": _my_pk_hex,
"role_priority": _role_priority(_my_role)
})
for pk_hex in _peer_registry:
if pk_hex == _my_pk_hex:
continue
var entry: Dictionary = _peer_registry[pk_hex]
if entry["role"] != "read-only":
candidates.append({
"pk": pk_hex,
"role_priority": entry["role_priority"]
})
candidates.sort_custom(func(a, b):
if a["role_priority"] != b["role_priority"]:
return a["role_priority"] < b["role_priority"]
return a["pk"] < b["pk"])
if candidates.is_empty():
_authority_pk = ""
return
_authority_pk = candidates[0]["pk"]
_log("authority elected: %s (priority=%d)" % [_authority_pk.left(12), candidates[0]["role_priority"]])
static func _role_priority(role: String) -> int:
match role:
"admin": return 1
"mod": return 2
"peer": return 3
_: return 99
# ---- Signal Handlers ----
func _on_peer_connected(id: int) -> void:
_log("peer_connected: %d (total: %d)" % [id, _connected_peers.size() + 1])
_connected_peers[id] = true
func _on_peer_disconnected(id: int) -> void:
_log("peer_disconnected: %d (total: %d)" % [id, _connected_peers.size() - 1])
_connected_peers.erase(id)
# ---- File I/O ----
func _write_file(path: String, content: String) -> void:
var wf := FileAccess.open(path, FileAccess.WRITE)
wf.store_string(content)
wf.close()
func _write_status(status: String) -> void:
_write_file("/tmp/layer10_peer_%d_status.txt" % _peer_index, status)
func _log(msg: String) -> void:
print("[P%d] %s" % [_peer_index, msg])
func _fail(msg: String) -> void:
_log("FAIL: %s" % msg)
_write_status("FAIL: %s" % msg)
if _peer:
_peer.close()
quit()