From 466db7aed285422b0b9451cce3206d3f49f457b8 Mon Sep 17 00:00:00 2001 From: hermes Date: Sun, 15 Mar 2026 10:22:41 -0400 Subject: [PATCH] =?UTF-8?q?[loop-cycle-44]=20refactor:=20remove=20dead=20c?= =?UTF-8?q?ode=20batch=202=20=E2=80=94=20agent=5Fcore=20+=20test=5Fagent?= =?UTF-8?q?=5Fcore=20(#147)=20(#150)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/timmy/agent_core/__init__.py | 1 - src/timmy/agent_core/interface.py | 381 ------------------------------ tests/timmy/test_agent_core.py | 343 --------------------------- 3 files changed, 725 deletions(-) delete mode 100644 src/timmy/agent_core/__init__.py delete mode 100644 src/timmy/agent_core/interface.py delete mode 100644 tests/timmy/test_agent_core.py diff --git a/src/timmy/agent_core/__init__.py b/src/timmy/agent_core/__init__.py deleted file mode 100644 index 1ed3f086..00000000 --- a/src/timmy/agent_core/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Agent Core — Substrate-agnostic agent interface and base classes.""" diff --git a/src/timmy/agent_core/interface.py b/src/timmy/agent_core/interface.py deleted file mode 100644 index a54811db..00000000 --- a/src/timmy/agent_core/interface.py +++ /dev/null @@ -1,381 +0,0 @@ -"""TimAgent Interface — The substrate-agnostic agent contract. - -This is the foundation for embodiment. Whether Timmy runs on: -- A server with Ollama (today) -- A Raspberry Pi with sensors -- A Boston Dynamics Spot robot -- A VR avatar - -The interface remains constant. Implementation varies. - -Architecture: - perceive() → reason → act() - ↑ ↓ - ←←← remember() ←←←←←←┘ - -All methods return effects that can be logged, audited, and replayed. -""" - -import uuid -from abc import ABC, abstractmethod -from dataclasses import dataclass, field -from datetime import UTC, datetime -from enum import Enum, auto -from typing import Any - - -class PerceptionType(Enum): - """Types of sensory input an agent can receive.""" - - TEXT = auto() # Natural language - IMAGE = auto() # Visual input - AUDIO = auto() # Sound/speech - SENSOR = auto() # Temperature, distance, etc. - MOTION = auto() # Accelerometer, gyroscope - NETWORK = auto() # API calls, messages - INTERNAL = auto() # Self-monitoring (battery, temp) - - -class ActionType(Enum): - """Types of actions an agent can perform.""" - - TEXT = auto() # Generate text response - SPEAK = auto() # Text-to-speech - MOVE = auto() # Physical movement - GRIP = auto() # Manipulate objects - CALL = auto() # API/network call - EMIT = auto() # Signal/light/sound - SLEEP = auto() # Power management - - -class AgentCapability(Enum): - """High-level capabilities a TimAgent may possess.""" - - REASONING = "reasoning" - CODING = "coding" - WRITING = "writing" - ANALYSIS = "analysis" - VISION = "vision" - SPEECH = "speech" - NAVIGATION = "navigation" - MANIPULATION = "manipulation" - LEARNING = "learning" - COMMUNICATION = "communication" - - -@dataclass(frozen=True) -class AgentIdentity: - """Immutable identity for an agent instance. - - This persists across sessions and substrates. If Timmy moves - from cloud to robot, the identity follows. - """ - - id: str - name: str - version: str - created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) - - @classmethod - def generate(cls, name: str, version: str = "1.0.0") -> "AgentIdentity": - """Generate a new unique identity.""" - return cls( - id=str(uuid.uuid4()), - name=name, - version=version, - ) - - -@dataclass -class Perception: - """A sensory input to the agent. - - Substrate-agnostic representation. A camera image and a - LiDAR point cloud are both Perception instances. - """ - - type: PerceptionType - data: Any # Content depends on type - timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) - source: str = "unknown" # e.g., "camera_1", "microphone", "user_input" - metadata: dict = field(default_factory=dict) - - @classmethod - def text(cls, content: str, source: str = "user") -> "Perception": - """Factory for text perception.""" - return cls( - type=PerceptionType.TEXT, - data=content, - source=source, - ) - - @classmethod - def sensor(cls, kind: str, value: float, unit: str = "") -> "Perception": - """Factory for sensor readings.""" - return cls( - type=PerceptionType.SENSOR, - data={"kind": kind, "value": value, "unit": unit}, - source=f"sensor_{kind}", - ) - - -@dataclass -class Action: - """An action the agent intends to perform. - - Actions are effects — they describe what should happen, - not how. The substrate implements the "how." - """ - - type: ActionType - payload: Any # Action-specific data - timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) - confidence: float = 1.0 # 0-1, agent's certainty - deadline: str | None = None # When action must complete - - @classmethod - def respond(cls, text: str, confidence: float = 1.0) -> "Action": - """Factory for text response action.""" - return cls( - type=ActionType.TEXT, - payload=text, - confidence=confidence, - ) - - @classmethod - def move(cls, vector: tuple[float, float, float], speed: float = 1.0) -> "Action": - """Factory for movement action (x, y, z meters).""" - return cls( - type=ActionType.MOVE, - payload={"vector": vector, "speed": speed}, - ) - - -@dataclass -class Memory: - """A stored experience or fact. - - Memories are substrate-agnostic. A conversation history - and a video recording are both Memory instances. - """ - - id: str - content: Any - created_at: str - access_count: int = 0 - last_accessed: str | None = None - importance: float = 0.5 # 0-1, for pruning decisions - tags: list[str] = field(default_factory=list) - - def touch(self) -> None: - """Mark memory as accessed.""" - self.access_count += 1 - self.last_accessed = datetime.now(UTC).isoformat() - - -@dataclass -class Communication: - """A message to/from another agent or human.""" - - sender: str - recipient: str - content: Any - timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) - protocol: str = "direct" # e.g., "http", "websocket", "speech" - encrypted: bool = False - - -class TimAgent(ABC): - """Abstract base class for all Timmy agent implementations. - - This is the substrate-agnostic interface. Implementations: - - OllamaAgent: LLM-based reasoning (today) - - RobotAgent: Physical embodiment (future) - - SimulationAgent: Virtual environment (future) - - Usage: - agent = OllamaAgent(identity) # Today's implementation - - perception = Perception.text("Hello Timmy") - memory = agent.perceive(perception) - - action = agent.reason("How should I respond?") - result = agent.act(action) - - agent.remember(memory) # Store for future - """ - - def __init__(self, identity: AgentIdentity) -> None: - self._identity = identity - self._capabilities: set[AgentCapability] = set() - self._state: dict[str, Any] = {} - - @property - def identity(self) -> AgentIdentity: - """Return this agent's immutable identity.""" - return self._identity - - @property - def capabilities(self) -> set[AgentCapability]: - """Return set of supported capabilities.""" - return self._capabilities.copy() - - def has_capability(self, capability: AgentCapability) -> bool: - """Check if agent supports a capability.""" - return capability in self._capabilities - - @abstractmethod - def perceive(self, perception: Perception) -> Memory: - """Process sensory input and create a memory. - - This is the entry point for all agent interaction. - A text message, camera frame, or temperature reading - all enter through perceive(). - - Args: - perception: Sensory input - - Returns: - Memory: Stored representation of the perception - """ - pass - - @abstractmethod - def reason(self, query: str, context: list[Memory]) -> Action: - """Reason about a situation and decide on action. - - This is where "thinking" happens. The agent uses its - substrate-appropriate reasoning (LLM, neural net, rules) - to decide what to do. - - Args: - query: What to reason about - context: Relevant memories for context - - Returns: - Action: What the agent decides to do - """ - pass - - @abstractmethod - def act(self, action: Action) -> Any: - """Execute an action in the substrate. - - This is where the abstract action becomes concrete: - - TEXT → Generate LLM response - - MOVE → Send motor commands - - SPEAK → Call TTS engine - - Args: - action: The action to execute - - Returns: - Result of the action (substrate-specific) - """ - pass - - @abstractmethod - def remember(self, memory: Memory) -> None: - """Store a memory for future retrieval. - - The storage mechanism depends on substrate: - - Cloud: SQLite, vector DB - - Robot: Local flash storage - - Hybrid: Synced with conflict resolution - - Args: - memory: Experience to store - """ - pass - - @abstractmethod - def recall(self, query: str, limit: int = 5) -> list[Memory]: - """Retrieve relevant memories. - - Args: - query: What to search for - limit: Maximum memories to return - - Returns: - List of relevant memories, sorted by relevance - """ - pass - - @abstractmethod - def communicate(self, message: Communication) -> bool: - """Send/receive communication with another agent. - - Args: - message: Message to send - - Returns: - True if communication succeeded - """ - pass - - def get_state(self) -> dict[str, Any]: - """Get current agent state for monitoring/debugging.""" - return { - "identity": self._identity, - "capabilities": list(self._capabilities), - "state": self._state.copy(), - } - - def shutdown(self) -> None: # noqa: B027 - """Graceful shutdown. Persist state, close connections.""" - # Override in subclass for cleanup - - -class AgentEffect: - """Log entry for agent actions — for audit and replay. - - The complete history of an agent's life can be captured - as a sequence of AgentEffects. This enables: - - Debugging: What did the agent see and do? - - Audit: Why did it make that decision? - - Replay: Reconstruct agent state from log - - Training: Learn from agent experiences - """ - - def __init__(self, log_path: str | None = None) -> None: - self._effects: list[dict] = [] - self._log_path = log_path - - def log_perceive(self, perception: Perception, memory_id: str) -> None: - """Log a perception event.""" - self._effects.append( - { - "type": "perceive", - "perception_type": perception.type.name, - "source": perception.source, - "memory_id": memory_id, - "timestamp": datetime.now(UTC).isoformat(), - } - ) - - def log_reason(self, query: str, action_type: ActionType) -> None: - """Log a reasoning event.""" - self._effects.append( - { - "type": "reason", - "query": query, - "action_type": action_type.name, - "timestamp": datetime.now(UTC).isoformat(), - } - ) - - def log_act(self, action: Action, result: Any) -> None: - """Log an action event.""" - self._effects.append( - { - "type": "act", - "action_type": action.type.name, - "confidence": action.confidence, - "result_type": type(result).__name__, - "timestamp": datetime.now(UTC).isoformat(), - } - ) - - def export(self) -> list[dict]: - """Export effect log for analysis.""" - return self._effects.copy() diff --git a/tests/timmy/test_agent_core.py b/tests/timmy/test_agent_core.py deleted file mode 100644 index 92f9d58a..00000000 --- a/tests/timmy/test_agent_core.py +++ /dev/null @@ -1,343 +0,0 @@ -"""Functional tests for agent_core — interface. - -Covers the substrate-agnostic agent contract (data classes, enums, -factory methods, abstract enforcement). -""" - -import uuid - -import pytest - -from timmy.agent_core.interface import ( - Action, - ActionType, - AgentCapability, - AgentEffect, - AgentIdentity, - Communication, - Memory, - Perception, - PerceptionType, - TimAgent, -) - -# ── AgentIdentity ───────────────────────────────────────────────────────────── - - -class TestAgentIdentity: - def test_generate_creates_uuid(self): - identity = AgentIdentity.generate("Timmy") - assert identity.name == "Timmy" - uuid.UUID(identity.id) # raises on invalid - - def test_generate_default_version(self): - identity = AgentIdentity.generate("Timmy") - assert identity.version == "1.0.0" - - def test_generate_custom_version(self): - identity = AgentIdentity.generate("Timmy", version="2.0.0") - assert identity.version == "2.0.0" - - def test_frozen_identity(self): - identity = AgentIdentity.generate("Timmy") - with pytest.raises(AttributeError): - identity.name = "Other" - - def test_created_at_populated(self): - identity = AgentIdentity.generate("Timmy") - assert identity.created_at # not empty - assert "T" in identity.created_at # ISO format - - def test_two_identities_differ(self): - a = AgentIdentity.generate("A") - b = AgentIdentity.generate("B") - assert a.id != b.id - - -# ── Perception ──────────────────────────────────────────────────────────────── - - -class TestPerception: - def test_text_factory(self): - p = Perception.text("hello") - assert p.type == PerceptionType.TEXT - assert p.data == "hello" - assert p.source == "user" - - def test_text_factory_custom_source(self): - p = Perception.text("hello", source="api") - assert p.source == "api" - - def test_sensor_factory(self): - p = Perception.sensor("temperature", 22.5, "°C") - assert p.type == PerceptionType.SENSOR - assert p.data["kind"] == "temperature" - assert p.data["value"] == 22.5 - assert p.data["unit"] == "°C" - assert p.source == "sensor_temperature" - - def test_timestamp_auto_populated(self): - p = Perception.text("hi") - assert p.timestamp - assert "T" in p.timestamp - - def test_metadata_defaults_empty(self): - p = Perception.text("hi") - assert p.metadata == {} - - -# ── Action ──────────────────────────────────────────────────────────────────── - - -class TestAction: - def test_respond_factory(self): - a = Action.respond("Hello!") - assert a.type == ActionType.TEXT - assert a.payload == "Hello!" - assert a.confidence == 1.0 - - def test_respond_with_confidence(self): - a = Action.respond("Maybe", confidence=0.5) - assert a.confidence == 0.5 - - def test_move_factory(self): - a = Action.move((1.0, 2.0, 3.0), speed=0.5) - assert a.type == ActionType.MOVE - assert a.payload["vector"] == (1.0, 2.0, 3.0) - assert a.payload["speed"] == 0.5 - - def test_move_default_speed(self): - a = Action.move((0, 0, 0)) - assert a.payload["speed"] == 1.0 - - def test_deadline_defaults_none(self): - a = Action.respond("test") - assert a.deadline is None - - -# ── Memory ──────────────────────────────────────────────────────────────────── - - -class TestMemory: - def test_touch_increments(self): - m = Memory(id="m1", content="hello", created_at="2025-01-01T00:00:00Z") - assert m.access_count == 0 - m.touch() - assert m.access_count == 1 - m.touch() - assert m.access_count == 2 - - def test_touch_sets_last_accessed(self): - m = Memory(id="m1", content="hello", created_at="2025-01-01T00:00:00Z") - assert m.last_accessed is None - m.touch() - assert m.last_accessed is not None - - def test_default_importance(self): - m = Memory(id="m1", content="x", created_at="now") - assert m.importance == 0.5 - - def test_tags_default_empty(self): - m = Memory(id="m1", content="x", created_at="now") - assert m.tags == [] - - -# ── Communication ───────────────────────────────────────────────────────────── - - -class TestCommunication: - def test_defaults(self): - c = Communication(sender="A", recipient="B", content="hi") - assert c.protocol == "direct" - assert c.encrypted is False - assert c.timestamp # auto-populated - - -# ── TimAgent abstract enforcement ───────────────────────────────────────────── - - -class TestTimAgentABC: - def test_cannot_instantiate_abstract(self): - with pytest.raises(TypeError): - TimAgent(AgentIdentity.generate("X")) - - def test_concrete_subclass_works(self): - class Dummy(TimAgent): - def perceive(self, p): - return Memory(id="1", content=p.data, created_at="") - - def reason(self, q, c): - return Action.respond(q) - - def act(self, a): - return a.payload - - def remember(self, m): - pass - - def recall(self, q, limit=5): - return [] - - def communicate(self, m): - return True - - d = Dummy(AgentIdentity.generate("Dummy")) - assert d.identity.name == "Dummy" - assert d.capabilities == set() - - def test_has_capability(self): - class Dummy(TimAgent): - def perceive(self, p): - pass - - def reason(self, q, c): - pass - - def act(self, a): - pass - - def remember(self, m): - pass - - def recall(self, q, limit=5): - return [] - - def communicate(self, m): - return True - - d = Dummy(AgentIdentity.generate("D")) - d._capabilities.add(AgentCapability.REASONING) - assert d.has_capability(AgentCapability.REASONING) - assert not d.has_capability(AgentCapability.VISION) - - def test_capabilities_returns_copy(self): - class Dummy(TimAgent): - def perceive(self, p): - pass - - def reason(self, q, c): - pass - - def act(self, a): - pass - - def remember(self, m): - pass - - def recall(self, q, limit=5): - return [] - - def communicate(self, m): - return True - - d = Dummy(AgentIdentity.generate("D")) - caps = d.capabilities - caps.add(AgentCapability.VISION) - assert AgentCapability.VISION not in d.capabilities - - def test_get_state(self): - class Dummy(TimAgent): - def perceive(self, p): - pass - - def reason(self, q, c): - pass - - def act(self, a): - pass - - def remember(self, m): - pass - - def recall(self, q, limit=5): - return [] - - def communicate(self, m): - return True - - d = Dummy(AgentIdentity.generate("D")) - state = d.get_state() - assert "identity" in state - assert "capabilities" in state - assert "state" in state - - def test_shutdown_does_not_raise(self): - class Dummy(TimAgent): - def perceive(self, p): - pass - - def reason(self, q, c): - pass - - def act(self, a): - pass - - def remember(self, m): - pass - - def recall(self, q, limit=5): - return [] - - def communicate(self, m): - return True - - d = Dummy(AgentIdentity.generate("D")) - d.shutdown() # should not raise - - -# ── AgentEffect ─────────────────────────────────────────────────────────────── - - -class TestAgentEffect: - def test_empty_export(self): - effect = AgentEffect() - assert effect.export() == [] - - def test_log_perceive(self): - effect = AgentEffect() - p = Perception.text("test input") - effect.log_perceive(p, "mem_0") - log = effect.export() - assert len(log) == 1 - assert log[0]["type"] == "perceive" - assert log[0]["perception_type"] == "TEXT" - assert log[0]["memory_id"] == "mem_0" - assert "timestamp" in log[0] - - def test_log_reason(self): - effect = AgentEffect() - effect.log_reason("How to help?", ActionType.TEXT) - log = effect.export() - assert len(log) == 1 - assert log[0]["type"] == "reason" - assert log[0]["query"] == "How to help?" - assert log[0]["action_type"] == "TEXT" - - def test_log_act(self): - effect = AgentEffect() - action = Action.respond("Hello!") - effect.log_act(action, "Hello!") - log = effect.export() - assert len(log) == 1 - assert log[0]["type"] == "act" - assert log[0]["confidence"] == 1.0 - assert log[0]["result_type"] == "str" - - def test_export_returns_copy(self): - effect = AgentEffect() - effect.log_reason("q", ActionType.TEXT) - exported = effect.export() - exported.clear() - assert len(effect.export()) == 1 - - def test_full_audit_trail(self): - effect = AgentEffect() - p = Perception.text("input") - effect.log_perceive(p, "m0") - effect.log_reason("what now?", ActionType.TEXT) - action = Action.respond("response") - effect.log_act(action, "response") - log = effect.export() - assert len(log) == 3 - types = [e["type"] for e in log] - assert types == ["perceive", "reason", "act"]