feat: WorldInterface abstraction + Heartbeat v2 loop (#871, #872)
Some checks failed
Tests / lint (pull_request) Has been cancelled
Tests / test (pull_request) Has been cancelled

Add a Gymnasium-style WorldInterface ABC at src/infrastructure/world/
with adapter registry, MockWorldAdapter for testing, and TES3MP stub.

Wire the interface into a new Heartbeat v2 loop (src/loop/heartbeat.py)
that drives observe→reason→act→reflect cycles through whatever world
adapter is connected, with graceful fallback to passive thinking when
no adapter is present. Enrich Phase 1 (Gather) with world observations.

Includes 63 passing tests covering interface contracts, all adapters,
registry, heartbeat cycles (embodied + passive), and WS broadcast.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
perplexity
2026-03-22 11:59:19 +00:00
parent 77a8fc8b96
commit c5a92b6fe1
15 changed files with 1183 additions and 4 deletions

View File

@@ -0,0 +1,29 @@
"""World interface — engine-agnostic adapter pattern for embodied agents.
Provides the ``WorldInterface`` ABC and an adapter registry so Timmy can
observe, act, and speak in any game world (Morrowind, Luanti, Godot, …)
through a single contract.
Quick start::
from infrastructure.world import get_adapter, register_adapter
from infrastructure.world.interface import WorldInterface
register_adapter("mock", MockWorldAdapter)
world = get_adapter("mock")
perception = world.observe()
"""
from infrastructure.world.registry import AdapterRegistry
_registry = AdapterRegistry()
register_adapter = _registry.register
get_adapter = _registry.get
list_adapters = _registry.list_adapters
__all__ = [
"register_adapter",
"get_adapter",
"list_adapters",
]

View File

@@ -0,0 +1 @@
"""Built-in world adapters."""

View File

@@ -0,0 +1,99 @@
"""Mock world adapter — returns canned perception and logs commands.
Useful for testing the heartbeat loop and WorldInterface contract
without a running game server.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from infrastructure.world.interface import WorldInterface
from infrastructure.world.types import (
ActionResult,
ActionStatus,
CommandInput,
PerceptionOutput,
)
logger = logging.getLogger(__name__)
@dataclass
class _ActionLog:
"""Record of an action dispatched to the mock world."""
command: CommandInput
timestamp: datetime
class MockWorldAdapter(WorldInterface):
"""In-memory mock adapter for testing.
* ``observe()`` returns configurable canned perception.
* ``act()`` logs the command and returns success.
* ``speak()`` logs the message.
Inspect ``action_log`` and ``speech_log`` to verify behaviour in tests.
"""
def __init__(
self,
*,
location: str = "Test Chamber",
entities: list[str] | None = None,
events: list[str] | None = None,
) -> None:
self._location = location
self._entities = entities or ["TestNPC"]
self._events = events or []
self._connected = False
self.action_log: list[_ActionLog] = []
self.speech_log: list[dict] = []
# -- lifecycle ---------------------------------------------------------
def connect(self) -> None:
self._connected = True
logger.info("MockWorldAdapter connected")
def disconnect(self) -> None:
self._connected = False
logger.info("MockWorldAdapter disconnected")
@property
def is_connected(self) -> bool:
return self._connected
# -- core contract -----------------------------------------------------
def observe(self) -> PerceptionOutput:
logger.debug("MockWorldAdapter.observe()")
return PerceptionOutput(
timestamp=datetime.now(UTC),
location=self._location,
entities=list(self._entities),
events=list(self._events),
raw={"adapter": "mock"},
)
def act(self, command: CommandInput) -> ActionResult:
logger.debug("MockWorldAdapter.act(%s)", command.action)
self.action_log.append(
_ActionLog(command=command, timestamp=datetime.now(UTC))
)
return ActionResult(
status=ActionStatus.SUCCESS,
message=f"Mock executed: {command.action}",
data={"adapter": "mock"},
)
def speak(self, message: str, target: str | None = None) -> None:
logger.debug("MockWorldAdapter.speak(%r, target=%r)", message, target)
self.speech_log.append({
"message": message,
"target": target,
"timestamp": datetime.now(UTC).isoformat(),
})

View File

@@ -0,0 +1,66 @@
"""TES3MP world adapter — stub for Morrowind multiplayer via TES3MP.
This adapter will eventually connect to a TES3MP server and translate
the WorldInterface contract into TES3MP commands. For now every method
raises ``NotImplementedError`` with guidance on what needs wiring up.
Once PR #864 merges, import PerceptionOutput and CommandInput directly
from ``infrastructure.morrowind.schemas`` if their shapes differ from
the canonical types in ``infrastructure.world.types``.
"""
from __future__ import annotations
import logging
from infrastructure.world.interface import WorldInterface
from infrastructure.world.types import ActionResult, CommandInput, PerceptionOutput
logger = logging.getLogger(__name__)
class TES3MPWorldAdapter(WorldInterface):
"""Stub adapter for TES3MP (Morrowind multiplayer).
All core methods raise ``NotImplementedError``.
Implement ``connect()`` first — it should open a socket to the
TES3MP server and authenticate.
"""
def __init__(self, *, host: str = "localhost", port: int = 25565) -> None:
self._host = host
self._port = port
self._connected = False
# -- lifecycle ---------------------------------------------------------
def connect(self) -> None:
raise NotImplementedError(
"TES3MPWorldAdapter.connect() — wire up TES3MP server socket"
)
def disconnect(self) -> None:
raise NotImplementedError(
"TES3MPWorldAdapter.disconnect() — close TES3MP server socket"
)
@property
def is_connected(self) -> bool:
return self._connected
# -- core contract (stubs) ---------------------------------------------
def observe(self) -> PerceptionOutput:
raise NotImplementedError(
"TES3MPWorldAdapter.observe() — poll TES3MP for player/NPC state"
)
def act(self, command: CommandInput) -> ActionResult:
raise NotImplementedError(
"TES3MPWorldAdapter.act() — translate CommandInput to TES3MP packet"
)
def speak(self, message: str, target: str | None = None) -> None:
raise NotImplementedError(
"TES3MPWorldAdapter.speak() — send chat message via TES3MP"
)

View File

@@ -0,0 +1,64 @@
"""Abstract WorldInterface — the contract every game-world adapter must fulfil.
Follows a Gymnasium-inspired pattern: observe → act → speak, with each
method returning strongly-typed data structures.
Any future engine (TES3MP, Luanti, Godot, …) plugs in by subclassing
``WorldInterface`` and implementing the three methods.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from infrastructure.world.types import ActionResult, CommandInput, PerceptionOutput
class WorldInterface(ABC):
"""Engine-agnostic base class for world adapters.
Subclasses must implement:
- ``observe()`` — gather structured perception from the world
- ``act()`` — dispatch a command and return the outcome
- ``speak()`` — send a message to an NPC / player / broadcast
Lifecycle hooks ``connect()`` and ``disconnect()`` are optional.
"""
# -- lifecycle (optional overrides) ------------------------------------
def connect(self) -> None:
"""Establish connection to the game world.
Default implementation is a no-op. Override to open sockets,
authenticate, etc.
"""
def disconnect(self) -> None:
"""Tear down the connection.
Default implementation is a no-op.
"""
@property
def is_connected(self) -> bool:
"""Return ``True`` if the adapter has an active connection.
Default returns ``True``. Override for adapters that maintain
persistent connections.
"""
return True
# -- core contract (must implement) ------------------------------------
@abstractmethod
def observe(self) -> PerceptionOutput:
"""Return a structured snapshot of the current world state."""
@abstractmethod
def act(self, command: CommandInput) -> ActionResult:
"""Execute *command* in the world and return the result."""
@abstractmethod
def speak(self, message: str, target: str | None = None) -> None:
"""Send *message* in the world, optionally directed at *target*."""

View File

@@ -0,0 +1,54 @@
"""Adapter registry — register and instantiate world adapters by name.
Usage::
registry = AdapterRegistry()
registry.register("mock", MockWorldAdapter)
adapter = registry.get("mock", some_kwarg="value")
"""
from __future__ import annotations
import logging
from typing import Any
from infrastructure.world.interface import WorldInterface
logger = logging.getLogger(__name__)
class AdapterRegistry:
"""Name → WorldInterface class registry with instantiation."""
def __init__(self) -> None:
self._adapters: dict[str, type[WorldInterface]] = {}
def register(self, name: str, cls: type[WorldInterface]) -> None:
"""Register an adapter class under *name*.
Raises ``TypeError`` if *cls* is not a ``WorldInterface`` subclass.
"""
if not (isinstance(cls, type) and issubclass(cls, WorldInterface)):
raise TypeError(f"{cls!r} is not a WorldInterface subclass")
if name in self._adapters:
logger.warning("Overwriting adapter %r (was %r)", name, self._adapters[name])
self._adapters[name] = cls
logger.info("Registered world adapter: %s%s", name, cls.__name__)
def get(self, name: str, **kwargs: Any) -> WorldInterface:
"""Instantiate and return the adapter registered as *name*.
Raises ``KeyError`` if *name* is not registered.
"""
cls = self._adapters[name]
return cls(**kwargs)
def list_adapters(self) -> list[str]:
"""Return sorted list of registered adapter names."""
return sorted(self._adapters)
def __contains__(self, name: str) -> bool:
return name in self._adapters
def __len__(self) -> int:
return len(self._adapters)

View File

@@ -0,0 +1,71 @@
"""Canonical data types for world interaction.
These mirror the PerceptionOutput / CommandInput types from PR #864's
``morrowind/schemas.py``. When that PR merges, these can be replaced
with re-exports — but until then they serve as the stable contract for
every WorldInterface adapter.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
class ActionStatus(str, Enum):
"""Outcome of an action dispatched to the world."""
SUCCESS = "success"
FAILURE = "failure"
PENDING = "pending"
NOOP = "noop"
@dataclass
class PerceptionOutput:
"""Structured world state returned by ``WorldInterface.observe()``.
Attributes:
timestamp: When the observation was captured.
location: Free-form location descriptor (e.g. "Balmora, Fighters Guild").
entities: List of nearby entity descriptions.
events: Recent game events since last observation.
raw: Optional raw / engine-specific payload for advanced consumers.
"""
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
location: str = ""
entities: list[str] = field(default_factory=list)
events: list[str] = field(default_factory=list)
raw: dict = field(default_factory=dict)
@dataclass
class CommandInput:
"""Action command sent via ``WorldInterface.act()``.
Attributes:
action: Verb / action name (e.g. "move", "attack", "use_item").
target: Optional target identifier.
parameters: Arbitrary key-value payload for engine-specific params.
"""
action: str
target: str | None = None
parameters: dict = field(default_factory=dict)
@dataclass
class ActionResult:
"""Outcome returned by ``WorldInterface.act()``.
Attributes:
status: Whether the action succeeded, failed, etc.
message: Human-readable description of the outcome.
data: Arbitrary engine-specific result payload.
"""
status: ActionStatus = ActionStatus.SUCCESS
message: str = ""
data: dict = field(default_factory=dict)

281
src/loop/heartbeat.py Normal file
View File

@@ -0,0 +1,281 @@
"""Heartbeat v2 — WorldInterface-driven cognitive loop.
Drives real observe → reason → act → reflect cycles through whatever
``WorldInterface`` adapter is connected. When no adapter is present,
gracefully falls back to the existing ``run_cycle()`` behaviour.
Usage::
heartbeat = Heartbeat(world=adapter, interval=30.0)
await heartbeat.run_once() # single cycle
await heartbeat.start() # background loop
heartbeat.stop() # graceful shutdown
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from loop.phase1_gather import gather
from loop.phase2_reason import reason
from loop.phase3_act import act
from loop.schema import ContextPayload
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Cycle log entry
# ---------------------------------------------------------------------------
@dataclass
class CycleRecord:
"""One observe → reason → act → reflect cycle."""
cycle_id: int
timestamp: str
observation: dict = field(default_factory=dict)
reasoning_summary: str = ""
action_taken: str = ""
action_status: str = ""
reflect_notes: str = ""
duration_ms: int = 0
# ---------------------------------------------------------------------------
# Heartbeat
# ---------------------------------------------------------------------------
class Heartbeat:
"""Manages the recurring cognitive loop with optional world adapter.
Parameters
----------
world:
A ``WorldInterface`` instance (or ``None`` for passive mode).
interval:
Seconds between heartbeat ticks. 30 s for embodied mode,
300 s (5 min) for passive thinking.
on_cycle:
Optional async callback invoked after each cycle with the
``CycleRecord``.
"""
def __init__(
self,
*,
world=None, # WorldInterface | None
interval: float = 30.0,
on_cycle=None, # Callable[[CycleRecord], Awaitable[None]] | None
) -> None:
self._world = world
self._interval = interval
self._on_cycle = on_cycle
self._cycle_count: int = 0
self._running = False
self._task: asyncio.Task | None = None
self.history: list[CycleRecord] = []
# -- properties --------------------------------------------------------
@property
def world(self):
return self._world
@world.setter
def world(self, adapter) -> None:
self._world = adapter
@property
def interval(self) -> float:
return self._interval
@interval.setter
def interval(self, value: float) -> None:
self._interval = max(1.0, value)
@property
def is_running(self) -> bool:
return self._running
@property
def cycle_count(self) -> int:
return self._cycle_count
# -- single cycle ------------------------------------------------------
async def run_once(self) -> CycleRecord:
"""Execute one full heartbeat cycle.
If a world adapter is present:
1. Observe — ``world.observe()``
2. Gather + Reason + Act via the three-phase loop, with the
observation injected into the payload
3. Dispatch the decided action back to ``world.act()``
4. Reflect — log the cycle
Without an adapter the existing loop runs on a timer-sourced
payload (passive thinking).
"""
self._cycle_count += 1
start = time.monotonic()
record = CycleRecord(
cycle_id=self._cycle_count,
timestamp=datetime.now(UTC).isoformat(),
)
if self._world is not None:
record = await self._embodied_cycle(record)
else:
record = await self._passive_cycle(record)
record.duration_ms = int((time.monotonic() - start) * 1000)
self.history.append(record)
# Broadcast via WebSocket (best-effort)
await self._broadcast(record)
if self._on_cycle:
await self._on_cycle(record)
logger.info(
"Heartbeat cycle #%d complete (%d ms) — action=%s status=%s",
record.cycle_id,
record.duration_ms,
record.action_taken or "(passive)",
record.action_status or "n/a",
)
return record
# -- background loop ---------------------------------------------------
async def start(self) -> None:
"""Start the recurring heartbeat loop as a background task."""
if self._running:
logger.warning("Heartbeat already running")
return
self._running = True
self._task = asyncio.current_task() or asyncio.ensure_future(self._loop())
if self._task is not asyncio.current_task():
return
await self._loop()
async def _loop(self) -> None:
logger.info(
"Heartbeat loop started (interval=%.1fs, adapter=%s)",
self._interval,
type(self._world).__name__ if self._world else "None",
)
while self._running:
try:
await self.run_once()
except Exception:
logger.exception("Heartbeat cycle failed")
await asyncio.sleep(self._interval)
def stop(self) -> None:
"""Signal the heartbeat loop to stop after the current cycle."""
self._running = False
logger.info("Heartbeat stop requested")
# -- internal: embodied cycle ------------------------------------------
async def _embodied_cycle(self, record: CycleRecord) -> CycleRecord:
"""Cycle with a live world adapter: observe → reason → act → reflect."""
from infrastructure.world.types import ActionStatus, CommandInput
# 1. Observe
perception = self._world.observe()
record.observation = {
"location": perception.location,
"entities": perception.entities,
"events": perception.events,
}
# 2. Feed observation into the three-phase loop
obs_content = (
f"Location: {perception.location}\n"
f"Entities: {', '.join(perception.entities)}\n"
f"Events: {', '.join(perception.events)}"
)
payload = ContextPayload(
source="world",
content=obs_content,
metadata={"perception": record.observation},
)
gathered = gather(payload)
reasoned = reason(gathered)
acted = act(reasoned)
# Extract action decision from the acted payload
action_name = acted.metadata.get("action", "idle")
action_target = acted.metadata.get("action_target")
action_params = acted.metadata.get("action_params", {})
record.reasoning_summary = acted.metadata.get("reasoning", acted.content[:200])
# 3. Dispatch action to world
if action_name != "idle":
cmd = CommandInput(
action=action_name,
target=action_target,
parameters=action_params,
)
result = self._world.act(cmd)
record.action_taken = action_name
record.action_status = result.status.value
else:
record.action_taken = "idle"
record.action_status = ActionStatus.NOOP.value
# 4. Reflect
record.reflect_notes = (
f"Observed {len(perception.entities)} entities at {perception.location}. "
f"Action: {record.action_taken}{record.action_status}."
)
return record
# -- internal: passive cycle -------------------------------------------
async def _passive_cycle(self, record: CycleRecord) -> CycleRecord:
"""Cycle without a world adapter — existing think_once() behaviour."""
payload = ContextPayload(
source="timer",
content="heartbeat",
metadata={"mode": "passive"},
)
gathered = gather(payload)
reasoned = reason(gathered)
acted = act(reasoned)
record.reasoning_summary = acted.content[:200]
record.action_taken = "think"
record.action_status = "noop"
record.reflect_notes = "Passive thinking cycle — no world adapter connected."
return record
# -- broadcast ---------------------------------------------------------
async def _broadcast(self, record: CycleRecord) -> None:
"""Emit heartbeat cycle data via WebSocket (best-effort)."""
try:
from infrastructure.ws_manager.handler import ws_manager
await ws_manager.broadcast("heartbeat.cycle", {
"cycle_id": record.cycle_id,
"timestamp": record.timestamp,
"action": record.action_taken,
"action_status": record.action_status,
"reasoning_summary": record.reasoning_summary[:300],
"observation": record.observation,
"duration_ms": record.duration_ms,
})
except (ImportError, AttributeError, ConnectionError, RuntimeError) as exc:
logger.debug("Heartbeat broadcast skipped: %s", exc)

View File

@@ -17,9 +17,9 @@ logger = logging.getLogger(__name__)
def gather(payload: ContextPayload) -> ContextPayload:
"""Accept raw input and return structured context for reasoning.
Stub: tags the payload with phase=gather and logs transit.
Timmy will flesh this out with context selection, memory lookup,
adapter polling, and attention-residual weighting.
When the payload carries a ``perception`` dict in metadata (injected by
the heartbeat loop from a WorldInterface adapter), that observation is
folded into the gathered context. Otherwise behaves as before.
"""
logger.info(
"Phase 1 (Gather) received: source=%s content_len=%d tokens=%d",
@@ -28,7 +28,20 @@ def gather(payload: ContextPayload) -> ContextPayload:
payload.token_count,
)
result = payload.with_metadata(phase="gather", gathered=True)
extra: dict = {"phase": "gather", "gathered": True}
# Enrich with world observation when present
perception = payload.metadata.get("perception")
if perception:
extra["world_observation"] = perception
logger.info(
"Phase 1 (Gather) world observation: location=%s entities=%d events=%d",
perception.get("location", "?"),
len(perception.get("entities", [])),
len(perception.get("events", [])),
)
result = payload.with_metadata(**extra)
logger.info(
"Phase 1 (Gather) produced: metadata_keys=%s",

View File

View File

@@ -0,0 +1,130 @@
"""Tests for the WorldInterface contract and type system."""
import pytest
from infrastructure.world.interface import WorldInterface
from infrastructure.world.types import (
ActionResult,
ActionStatus,
CommandInput,
PerceptionOutput,
)
# ---------------------------------------------------------------------------
# Type construction
# ---------------------------------------------------------------------------
class TestPerceptionOutput:
def test_defaults(self):
p = PerceptionOutput()
assert p.location == ""
assert p.entities == []
assert p.events == []
assert p.raw == {}
assert p.timestamp is not None
def test_custom_values(self):
p = PerceptionOutput(
location="Balmora",
entities=["Guard", "Merchant"],
events=["door_opened"],
)
assert p.location == "Balmora"
assert len(p.entities) == 2
assert "door_opened" in p.events
class TestCommandInput:
def test_minimal(self):
c = CommandInput(action="move")
assert c.action == "move"
assert c.target is None
assert c.parameters == {}
def test_with_target_and_params(self):
c = CommandInput(action="attack", target="Rat", parameters={"weapon": "sword"})
assert c.target == "Rat"
assert c.parameters["weapon"] == "sword"
class TestActionResult:
def test_defaults(self):
r = ActionResult()
assert r.status == ActionStatus.SUCCESS
assert r.message == ""
def test_failure(self):
r = ActionResult(status=ActionStatus.FAILURE, message="blocked")
assert r.status == ActionStatus.FAILURE
class TestActionStatus:
def test_values(self):
assert ActionStatus.SUCCESS.value == "success"
assert ActionStatus.FAILURE.value == "failure"
assert ActionStatus.PENDING.value == "pending"
assert ActionStatus.NOOP.value == "noop"
# ---------------------------------------------------------------------------
# Abstract contract
# ---------------------------------------------------------------------------
class TestWorldInterfaceContract:
"""Verify the ABC cannot be instantiated directly."""
def test_cannot_instantiate(self):
with pytest.raises(TypeError):
WorldInterface()
def test_subclass_must_implement_observe(self):
class Incomplete(WorldInterface):
def act(self, command):
pass
def speak(self, message, target=None):
pass
with pytest.raises(TypeError):
Incomplete()
def test_subclass_must_implement_act(self):
class Incomplete(WorldInterface):
def observe(self):
return PerceptionOutput()
def speak(self, message, target=None):
pass
with pytest.raises(TypeError):
Incomplete()
def test_subclass_must_implement_speak(self):
class Incomplete(WorldInterface):
def observe(self):
return PerceptionOutput()
def act(self, command):
return ActionResult()
with pytest.raises(TypeError):
Incomplete()
def test_complete_subclass_instantiates(self):
class Complete(WorldInterface):
def observe(self):
return PerceptionOutput()
def act(self, command):
return ActionResult()
def speak(self, message, target=None):
pass
adapter = Complete()
assert adapter.is_connected is True # default
assert isinstance(adapter.observe(), PerceptionOutput)
assert isinstance(adapter.act(CommandInput(action="test")), ActionResult)

View File

@@ -0,0 +1,80 @@
"""Tests for the MockWorldAdapter — full observe/act/speak cycle."""
from infrastructure.world.adapters.mock import MockWorldAdapter
from infrastructure.world.types import ActionStatus, CommandInput, PerceptionOutput
class TestMockWorldAdapter:
def test_observe_returns_perception(self):
adapter = MockWorldAdapter(location="Vivec")
perception = adapter.observe()
assert isinstance(perception, PerceptionOutput)
assert perception.location == "Vivec"
assert perception.raw == {"adapter": "mock"}
def test_observe_entities(self):
adapter = MockWorldAdapter(entities=["Jiub", "Silt Strider"])
perception = adapter.observe()
assert perception.entities == ["Jiub", "Silt Strider"]
def test_act_logs_command(self):
adapter = MockWorldAdapter()
cmd = CommandInput(action="move", target="north")
result = adapter.act(cmd)
assert result.status == ActionStatus.SUCCESS
assert "move" in result.message
assert len(adapter.action_log) == 1
assert adapter.action_log[0].command.action == "move"
def test_act_multiple_commands(self):
adapter = MockWorldAdapter()
adapter.act(CommandInput(action="attack"))
adapter.act(CommandInput(action="defend"))
adapter.act(CommandInput(action="retreat"))
assert len(adapter.action_log) == 3
def test_speak_logs_message(self):
adapter = MockWorldAdapter()
adapter.speak("Hello, traveler!")
assert len(adapter.speech_log) == 1
assert adapter.speech_log[0]["message"] == "Hello, traveler!"
assert adapter.speech_log[0]["target"] is None
def test_speak_with_target(self):
adapter = MockWorldAdapter()
adapter.speak("Die, scum!", target="Cliff Racer")
assert adapter.speech_log[0]["target"] == "Cliff Racer"
def test_lifecycle(self):
adapter = MockWorldAdapter()
assert adapter.is_connected is False
adapter.connect()
assert adapter.is_connected is True
adapter.disconnect()
assert adapter.is_connected is False
def test_full_observe_act_speak_cycle(self):
"""Acceptance criterion: full observe/act/speak cycle passes."""
adapter = MockWorldAdapter(
location="Seyda Neen",
entities=["Fargoth", "Hrisskar"],
events=["quest_started"],
)
adapter.connect()
# Observe
perception = adapter.observe()
assert perception.location == "Seyda Neen"
assert len(perception.entities) == 2
assert "quest_started" in perception.events
# Act
result = adapter.act(CommandInput(action="talk", target="Fargoth"))
assert result.status == ActionStatus.SUCCESS
# Speak
adapter.speak("Where is your ring, Fargoth?", target="Fargoth")
assert len(adapter.speech_log) == 1
adapter.disconnect()
assert adapter.is_connected is False

View File

@@ -0,0 +1,69 @@
"""Tests for the adapter registry."""
import pytest
from infrastructure.world.adapters.mock import MockWorldAdapter
from infrastructure.world.interface import WorldInterface
from infrastructure.world.registry import AdapterRegistry
class TestAdapterRegistry:
def test_register_and_get(self):
reg = AdapterRegistry()
reg.register("mock", MockWorldAdapter)
adapter = reg.get("mock")
assert isinstance(adapter, MockWorldAdapter)
def test_register_with_kwargs(self):
reg = AdapterRegistry()
reg.register("mock", MockWorldAdapter)
adapter = reg.get("mock", location="Custom Room")
assert adapter._location == "Custom Room"
def test_get_unknown_raises(self):
reg = AdapterRegistry()
with pytest.raises(KeyError):
reg.get("nonexistent")
def test_register_non_subclass_raises(self):
reg = AdapterRegistry()
with pytest.raises(TypeError):
reg.register("bad", dict)
def test_list_adapters(self):
reg = AdapterRegistry()
reg.register("beta", MockWorldAdapter)
reg.register("alpha", MockWorldAdapter)
assert reg.list_adapters() == ["alpha", "beta"]
def test_contains(self):
reg = AdapterRegistry()
reg.register("mock", MockWorldAdapter)
assert "mock" in reg
assert "other" not in reg
def test_len(self):
reg = AdapterRegistry()
assert len(reg) == 0
reg.register("mock", MockWorldAdapter)
assert len(reg) == 1
def test_overwrite_warns(self, caplog):
import logging
reg = AdapterRegistry()
reg.register("mock", MockWorldAdapter)
with caplog.at_level(logging.WARNING):
reg.register("mock", MockWorldAdapter)
assert "Overwriting" in caplog.text
class TestModuleLevelRegistry:
"""Test the convenience functions in infrastructure.world.__init__."""
def test_register_and_get(self):
from infrastructure.world import get_adapter, register_adapter
register_adapter("test_mock", MockWorldAdapter)
adapter = get_adapter("test_mock")
assert isinstance(adapter, MockWorldAdapter)

View File

@@ -0,0 +1,44 @@
"""Tests for the TES3MP stub adapter."""
import pytest
from infrastructure.world.adapters.tes3mp import TES3MPWorldAdapter
from infrastructure.world.types import CommandInput
class TestTES3MPStub:
"""Acceptance criterion: stub imports cleanly and raises NotImplementedError."""
def test_instantiates(self):
adapter = TES3MPWorldAdapter(host="127.0.0.1", port=25565)
assert adapter._host == "127.0.0.1"
assert adapter._port == 25565
def test_is_connected_default_false(self):
adapter = TES3MPWorldAdapter()
assert adapter.is_connected is False
def test_connect_raises(self):
adapter = TES3MPWorldAdapter()
with pytest.raises(NotImplementedError, match="connect"):
adapter.connect()
def test_disconnect_raises(self):
adapter = TES3MPWorldAdapter()
with pytest.raises(NotImplementedError, match="disconnect"):
adapter.disconnect()
def test_observe_raises(self):
adapter = TES3MPWorldAdapter()
with pytest.raises(NotImplementedError, match="observe"):
adapter.observe()
def test_act_raises(self):
adapter = TES3MPWorldAdapter()
with pytest.raises(NotImplementedError, match="act"):
adapter.act(CommandInput(action="move"))
def test_speak_raises(self):
adapter = TES3MPWorldAdapter()
with pytest.raises(NotImplementedError, match="speak"):
adapter.speak("Hello")

View File

@@ -0,0 +1,178 @@
"""Tests for Heartbeat v2 — WorldInterface-driven cognitive loop.
Acceptance criteria:
- With MockWorldAdapter: heartbeat runs, logs show observe→reason→act→reflect
- Without adapter: existing think_once() behaviour unchanged
- WebSocket broadcasts include current action and reasoning summary
"""
import asyncio
from unittest.mock import AsyncMock, patch
import pytest
from infrastructure.world.adapters.mock import MockWorldAdapter
from infrastructure.world.types import ActionStatus
from loop.heartbeat import CycleRecord, Heartbeat
@pytest.fixture
def mock_adapter():
adapter = MockWorldAdapter(
location="Balmora",
entities=["Guard", "Merchant"],
events=["player_entered"],
)
adapter.connect()
return adapter
class TestHeartbeatWithAdapter:
"""With MockWorldAdapter: heartbeat runs full embodied cycle."""
@pytest.mark.asyncio
async def test_run_once_returns_cycle_record(self, mock_adapter):
hb = Heartbeat(world=mock_adapter)
record = await hb.run_once()
assert isinstance(record, CycleRecord)
assert record.cycle_id == 1
@pytest.mark.asyncio
async def test_observation_populated(self, mock_adapter):
hb = Heartbeat(world=mock_adapter)
record = await hb.run_once()
assert record.observation["location"] == "Balmora"
assert "Guard" in record.observation["entities"]
assert "player_entered" in record.observation["events"]
@pytest.mark.asyncio
async def test_action_dispatched_to_world(self, mock_adapter):
"""Act phase should dispatch to world.act() for non-idle actions."""
hb = Heartbeat(world=mock_adapter)
record = await hb.run_once()
# The default loop phases don't set an explicit action, so it
# falls through to "idle" → NOOP. That's correct behaviour —
# the real LLM-powered reason phase will set action metadata.
assert record.action_status in (
ActionStatus.NOOP.value,
ActionStatus.SUCCESS.value,
)
@pytest.mark.asyncio
async def test_reflect_notes_present(self, mock_adapter):
hb = Heartbeat(world=mock_adapter)
record = await hb.run_once()
assert "Balmora" in record.reflect_notes
@pytest.mark.asyncio
async def test_cycle_count_increments(self, mock_adapter):
hb = Heartbeat(world=mock_adapter)
await hb.run_once()
await hb.run_once()
assert hb.cycle_count == 2
assert len(hb.history) == 2
@pytest.mark.asyncio
async def test_duration_recorded(self, mock_adapter):
hb = Heartbeat(world=mock_adapter)
record = await hb.run_once()
assert record.duration_ms >= 0
@pytest.mark.asyncio
async def test_on_cycle_callback(self, mock_adapter):
received = []
async def callback(record):
received.append(record)
hb = Heartbeat(world=mock_adapter, on_cycle=callback)
await hb.run_once()
assert len(received) == 1
assert received[0].cycle_id == 1
class TestHeartbeatWithoutAdapter:
"""Without adapter: existing think_once() behaviour unchanged."""
@pytest.mark.asyncio
async def test_passive_cycle(self):
hb = Heartbeat(world=None)
record = await hb.run_once()
assert record.action_taken == "think"
assert record.action_status == "noop"
assert "Passive" in record.reflect_notes
@pytest.mark.asyncio
async def test_passive_no_observation(self):
hb = Heartbeat(world=None)
record = await hb.run_once()
assert record.observation == {}
class TestHeartbeatLifecycle:
def test_interval_property(self):
hb = Heartbeat(interval=60.0)
assert hb.interval == 60.0
hb.interval = 10.0
assert hb.interval == 10.0
def test_interval_minimum(self):
hb = Heartbeat()
hb.interval = 0.1
assert hb.interval == 1.0
def test_world_property(self):
hb = Heartbeat()
assert hb.world is None
adapter = MockWorldAdapter()
hb.world = adapter
assert hb.world is adapter
def test_stop_sets_flag(self):
hb = Heartbeat()
assert not hb.is_running
hb.stop()
assert not hb.is_running
class TestHeartbeatBroadcast:
"""WebSocket broadcasts include action and reasoning summary."""
@pytest.mark.asyncio
async def test_broadcast_called(self, mock_adapter):
with patch(
"loop.heartbeat.ws_manager",
create=True,
) as mock_ws:
mock_ws.broadcast = AsyncMock()
# Patch the import inside heartbeat
with patch(
"infrastructure.ws_manager.handler.ws_manager"
) as ws_mod:
ws_mod.broadcast = AsyncMock()
hb = Heartbeat(world=mock_adapter)
await hb.run_once()
ws_mod.broadcast.assert_called_once()
call_args = ws_mod.broadcast.call_args
assert call_args[0][0] == "heartbeat.cycle"
data = call_args[0][1]
assert "action" in data
assert "reasoning_summary" in data
assert "observation" in data
class TestHeartbeatLog:
"""Verify logging of observe→reason→act→reflect cycle."""
@pytest.mark.asyncio
async def test_embodied_cycle_logs(self, mock_adapter, caplog):
import logging
with caplog.at_level(logging.INFO):
hb = Heartbeat(world=mock_adapter)
await hb.run_once()
messages = caplog.text
assert "Phase 1 (Gather)" in messages
assert "Phase 2 (Reason)" in messages
assert "Phase 3 (Act)" in messages
assert "Heartbeat cycle #1 complete" in messages