feat(game-engine): Improve game state management with audio utilities, auto-play logic, and entity handling; add integration tests for game-over and rally scenarios; update smoke testing tool for multi-slot support

Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
autocommit 2026-05-18 01:23:25 -07:00
parent 306046e2b5
commit eb8b82700c
9 changed files with 190 additions and 0 deletions

View file

@ -0,0 +1 @@
uid://c1j3ytslhrdx0

View file

@ -0,0 +1 @@
uid://ce6rvoite6jo7

View file

@ -0,0 +1 @@
uid://dttnp7w20e5uq

View file

@ -144,6 +144,15 @@ func _ensure_gd_state() -> void:
push_error("GameState: GdGameState class not registered (gdext build missing?)")
## Public accessor for the shared GdGameState bridge. Returns the
## existing instance (or lazily creates one) — callers MUST null-check
## before invoking methods because headless builds without the
## GDExtension dylib will see `null`.
func get_gd_state() -> RefCounted:
_ensure_gd_state()
return _gd_state
func get_ai_controller() -> RefCounted:
## p2-43a-followup — lazy accessor for the shared `GdAiController` Rail-1
## bridge instance. Returns `null` when the GDExtension is not loaded

View file

@ -0,0 +1 @@
uid://d1ixnyvgc3vwe

View file

@ -0,0 +1 @@
uid://e7rxe1g4o8hn

View file

@ -0,0 +1 @@
uid://bt8djyw5rtocg

View file

@ -0,0 +1 @@
uid://k4k2wn686j3p

View file

@ -0,0 +1,174 @@
"""Stage 4 multi-slot adapter smoke test.
Verifies without `gymnasium`, `stable-baselines3`, or `torch` that
one harness process can be driven externally on more than one player
slot. Companion to `smoke.py`, which only covers the single-slot wire
shape.
Setup:
- `HarnessConfig(players=3, player_slots=(0, 1), map_size="duel")`
- Slots 0 and 1 are externally driven; slot 2 runs the scripted
internal AI loop.
- `CP_PLAYER_SLOTS="0,1"` is set by `to_env`; the dispatcher's
`apply_end_turn` skips those slots when chaining the internal AI
loop, leaving slot 2 as the only one the harness advances itself.
Test loop (3 turns):
- For each turn, for each external slot s in (0, 1):
client.view(slot=s)
client.end_turn(slot=s)
- After both externals end their turn, the harness internal AI loop
advances slot 2, which emits an `ai_turn_completed` notification
for `player=2`.
Assertion:
- `ai_turn_completed` events (collected from both `act` response
`events` arrays and any async notifications) must reference ONLY
`player == 2`. Any event for player 0 or 1 means the wire `slot`
field wasn't honored — the harness defaulted back to slot 0,
`end_turn` advanced the wrong slot, and the internal AI loop then
fired for slot 1 (or 0).
Note: the GDScript harness emits AI events synchronously inside the
`act` response's `events` array, not as async notification lines —
see `player_api_main.gd::_emit_event`, which only fires for setup
events. We therefore read `response["events"]` from each `end_turn`
return value AND drain notifications for completeness.
Output: one-line JSON verdict on stdout, same shape as `smoke.py`:
{"passed": bool, "reasons": [...], "details": {...}}
Exit 0 on pass, 1 on fail.
Run:
python3 -m tooling.rl_self_play.smoke_multi_slot
"""
from __future__ import annotations
import json
import sys
from collections import Counter
from pathlib import Path
from typing import Any
THIS_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = THIS_DIR.parents[1]
if __package__ is None:
sys.path.insert(0, str(PROJECT_ROOT))
from tooling.rl_self_play.harness_client import ( # noqa: E402
HarnessClient,
HarnessConfig,
HarnessError,
)
TURNS = 3
EXTERNAL_SLOTS: tuple[int, ...] = (0, 1)
INTERNAL_SLOT = 2
def _count_ai_turn_completed(events: list[dict[str, Any]]) -> Counter[int]:
"""Return Counter mapping player_id -> count of ai_turn_completed events.
The GDScript harness round-trips PlayerId through Godot, which serialises
integers as JSON numbers without distinguishing int vs float so
`player` arrives as e.g. `2.0`. Coerce via int() with a numeric check
rather than `isinstance(player, int)`, which silently drops the events.
"""
counts: Counter[int] = Counter()
for n in events:
if n.get("type") != "ai_turn_completed":
continue
player = n.get("player")
if isinstance(player, (int, float)) and not isinstance(player, bool):
counts[int(player)] += 1
return counts
def _emit(passed: bool, reasons: list[str], details: dict[str, Any]) -> None:
print(json.dumps({"passed": passed, "reasons": reasons, "details": details}))
def main() -> int:
cfg = HarnessConfig(
seed=42,
players=3,
player_slots=EXTERNAL_SLOTS,
map_size="duel",
)
reasons: list[str] = []
details: dict[str, Any] = {
"turns_driven": 0,
"ai_turn_completed_by_player": {},
"external_slots": list(EXTERNAL_SLOTS),
"internal_slot": INTERNAL_SLOT,
}
all_counts: Counter[int] = Counter()
try:
with HarnessClient(cfg) as client:
for turn_idx in range(TURNS):
for slot in EXTERNAL_SLOTS:
try:
view = client.view(slot=slot)
except HarnessError as e:
reasons.append(
f"view(slot={slot}) failed on turn {turn_idx}: {e}"
)
details["ai_turn_completed_by_player"] = dict(all_counts)
_emit(False, reasons, details)
return 1
observed = view.get("player")
if (
isinstance(observed, (int, float))
and not isinstance(observed, bool)
and int(observed) != slot
):
reasons.append(
f"view(slot={slot}) returned view for player={observed}"
)
try:
resp = client.end_turn(slot=slot)
except HarnessError as e:
reasons.append(
f"end_turn(slot={slot}) failed on turn {turn_idx}: {e}"
)
details["ai_turn_completed_by_player"] = dict(all_counts)
_emit(False, reasons, details)
return 1
# Events for the internal AI loop arrive in the act
# response's synchronous `events` array.
resp_events = resp.get("events") or []
all_counts.update(_count_ai_turn_completed(resp_events))
# Drain any async notifications too (belt + braces).
drained = client.drain_notifications()
all_counts.update(_count_ai_turn_completed(drained))
details["turns_driven"] = turn_idx + 1
except HarnessError as e:
reasons.append(f"harness fatal: {e}")
details["ai_turn_completed_by_player"] = dict(all_counts)
_emit(False, reasons, details)
return 1
details["ai_turn_completed_by_player"] = dict(all_counts)
leaked_external = {p: c for p, c in all_counts.items() if p in EXTERNAL_SLOTS}
if leaked_external:
reasons.append(
f"ai_turn_completed fired for externally-driven slots "
f"{leaked_external} — wire `slot` field not honored"
)
if all_counts.get(INTERNAL_SLOT, 0) == 0:
reasons.append(
f"no ai_turn_completed for internal slot {INTERNAL_SLOT}"
f"internal AI loop never ran (expected >=1 over {TURNS} turns)"
)
passed = not reasons
_emit(passed, reasons, details)
return 0 if passed else 1
if __name__ == "__main__":
sys.exit(main())