.env

@@ -0,0 +1,7 @@
+MISTLE_HOST=sl.mud.at
+MISTLE_PORT=4711
+MISTLE_USER=mistle
+MISTLE_PASSWORD=sl-mudbot
+MISTLE_LOGIN_PROMPT=Wie heisst Du denn ("neu" fuer neuen Spieler) ?
+MISTLE_EXIT_COMMAND=schlaf ein
+MISTLE_AGENT_MODE=true

.gitignore

@@ -9,6 +9,3 @@ wheels/
 # Virtual environments
 .venv
 .DS_STORE
-
-# config
-.env

.vscode/settings.json

@@ -1,3 +0,0 @@
-{
-    "wolf.disableHotModeWarning": true
-}

README.md

@@ -1,89 +0,0 @@
-# Mistle Mudbot
-
-Python-based Telnet helper for connecting to MUD servers, handling login flows, and optionally running automated "agents" alongside interactive play.
-
-## Features
-
-- Lightweight wrapper (`TelnetClient`) around `telnetlib` with sane defaults and context-manager support.
-- Loads credentials and connection settings from a local `.env` file.
-- Interactive console session that mirrors server output and lets you type commands directly.
-- Optional always-on agent mode plus an on-demand `#execute <agent>` escape hatch for ad-hoc automations.
-- Built-in agents (`SimpleAgent`, `ExploreAgent`, `CommunicationAgent`, `IntelligentCommunicationAgent`) with a pluggable interface for custom behaviours.
-
-## Requirements
-
-- Python 3.10+
-- A reachable Telnet-based MUD server
-
-## Quick Start
-
-1. Create a virtual environment and install dependencies:
-
-   ```bash
-   pip install -e .
-   ```
-
-2. Copy `.env.example` (if available) or create a `.env` file in the project root, then fill in the connection details described below.
-
-3. Run the client:
-
-   ```bash
-   python app.py
-   ```
-
-4. Type commands directly into the console. Press `Ctrl-C` to exit; the client will send any configured shutdown command to the MUD.
-
-5. When *not* running in agent mode you can kick off one-off automations from the prompt:
-
-   ```text
-   #execute explore
-   ```
-
-   The command remains interactive while the agent works in the background and stops automatically a few seconds after things quiet down.
-
-## Environment Variables
-
-All variables can be placed in the `.env` file (one `KEY=value` per line) or provided through the shell environment.
-
-| Variable | Required | Description |
-| --- | --- | --- |
-| `MISTLE_HOST` | ✅ | DNS name or IP of the target MUD server. |
-| `MISTLE_PORT` | ✅ | Telnet port number (will be cast to `int`). |
-| `MISTLE_USER` | ❌ | Username or character name; sent automatically after login banner, if provided. |
-| `MISTLE_PASSWORD` | ❌ | Password sent after the username. Leave blank for manual entry. |
-| `MISTLE_LOGIN_PROMPT` | ❌ | Prompt string that signals the client to send credentials (e.g., `"Name:"`). When omitted, the client just waits for the initial banner. |
-| `MISTLE_EXIT_COMMAND` | ❌ | Command issued during graceful shutdown (after pressing `Ctrl-C`). Useful for `quit`/`save` macros. |
-| `MISTLE_AGENT_MODE` | ❌ | Enable full-time agent thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. |
-| `MISTLE_AGENT` | ❌ | Select which agent class to instantiate when agent mode is active. Accepted values: `simple` (default), `explore`, `communication`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. |
-| `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent agent (defaults to `mistral/mistral-small-2407`). |
-| `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationAgent` (via `litellm`) when calling the `mistral/mistral-small-2407` model. |
-
-## Agent Development
-
-- Implement new agents by subclassing `agent.Agent` and overriding `observe()` and `decide()`.
-- Register the agent by either:
-  - Adding the class to `agent.py` and referencing it in `MISTLE_AGENT` (e.g., `explore` for `ExploreAgent`).
-  - Placing the class elsewhere and configuring `MISTLE_AGENT` to `your_module:YourAgent`.
-- `observe(output)` receives the latest server text; `decide()` returns the next command string or `None` to stay idle.
-- Commands issued by the agent are throttled to one per second so manual commands can still interleave smoothly.
-- `ExploreAgent` showcases a richer workflow: it sends `schau`, identifies German nouns, inspects each with `untersuche`, and prints `[Agent]` progress updates like `Explored 3/7 — untersuche Tisch`.
-- `CommunicationAgent` auto-replies to every direct tell with a canned greeting, while `IntelligentCommunicationAgent` routes each tell through `litellm` (default model `mistral/mistral-small-2407`) to craft a contextual answer.
-
-## On-Demand Agents
-
-- When `MISTLE_AGENT_MODE` is **off**, you can trigger an ephemeral agent at any time with `#execute <agent_spec>`.
-- The syntax accepts the same values as `MISTLE_AGENT` and reuses the `build_agent` helper, so `#execute simple`, `#execute explore`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomAgent` are all valid.
-- On-demand runs share the current session, respect the one-command-per-second limit, and stop automatically after a few seconds of inactivity.
-
-## Danger Zone
-
-- The Telnet session runs until you interrupt it. Make sure the terminal is in a state where `Ctrl-C` is available.
-- When adding new agents, guard any long-running logic to avoid blocking the agent thread.
-
-## Contributing
-
-Feel free to open issues or submit pull requests for additional MUD-specific helpers, new agents, or quality-of-life improvements.
-
----
-
-Happy MUDding!

agent.py

@@ -1,10 +1,8 @@
 from __future__ import annotations
 
-import re
 from abc import ABC, abstractmethod
-from collections import deque
 from dataclasses import dataclass, field
-from typing import Deque, Optional, Pattern, Set, Tuple
+from typing import Optional
 
 
 class Agent(ABC):
@@ -21,7 +19,7 @@ class Agent(ABC):
 
 @dataclass
 class SimpleAgent(Agent):
-    """Minimal agent that always returns the same command."""
+    """Very small stateful agent that decides which command to run next."""
 
     default_command: str = "schau"
     last_output: str = field(default="", init=False)
@@ -32,86 +30,3 @@ class SimpleAgent(Agent):
 
     def decide(self) -> Optional[str]:
         return self.default_command
-
-
-@dataclass
-class ExploreAgent(Agent):
-    """Agent that inspects every noun it discovers in the room description."""
-
-    look_command: str = "schau"
-    inspect_command: str = "untersuche"
-    nouns_pattern: Pattern[str] = field(
-        default_factory=lambda: re.compile(r"\b[A-ZÄÖÜ][A-Za-zÄÖÜäöüß-]*\b")
-    )
-    last_output: str = field(default="", init=False)
-    look_sent: bool = field(default=False, init=False)
-    pending_targets: Deque[str] = field(default_factory=deque, init=False)
-    seen_targets: Set[str] = field(default_factory=set, init=False)
-    inspected_targets: Set[str] = field(default_factory=set, init=False)
-
-    def observe(self, output: str) -> None:
-        if not output:
-            return
-        self.last_output = output
-        if not self.look_sent:
-            return
-
-        for noun in self.nouns_pattern.findall(output):
-            target = noun.strip()
-            key = target.lower()
-            if not target or key in self.seen_targets:
-                continue
-            self.seen_targets.add(key)
-            self.pending_targets.append(target)
-
-    def decide(self) -> Optional[str]:
-        if not self.look_sent:
-            self.look_sent = True
-            print("[Agent] Exploring room, sending 'schau'")
-            return self.look_command
-
-        if self.pending_targets:
-            target = self.pending_targets.popleft()
-            key = target.lower()
-            self.inspected_targets.add(key)
-            progress = f"[Agent] Explored {len(self.inspected_targets)}/{len(self.seen_targets)} — untersuche {target}"
-            print(progress)
-            return f"{self.inspect_command} {target}"
-
-        return None
-
-
-@dataclass
-class CommunicationAgent(Agent):
-    """Agent that replies to private tells."""
-
-    reply_template: str = "teile {player} mit Hallo! Ich bin Mistle und ein Bot."
-    tell_pattern: Pattern[str] = field(
-        default_factory=lambda: re.compile(
-            r"^(?P<player>[^\s]+) teilt (d|D)ir mit: (?P<message>.+)$",
-            re.MULTILINE,
-        )
-    )
-    last_output: str = field(default="", init=False)
-    pending_replies: Deque[Tuple[str, str]] = field(default_factory=deque, init=False)
-
-    def observe(self, output: str) -> None:
-        if not output:
-            return
-        self.last_output = output
-        for match in self.tell_pattern.finditer(output):
-            player = match.group("player").strip()
-            message = match.group("message").strip()
-            if not player:
-                continue
-            self.pending_replies.append((player, message))
-            print(f"[Agent] Received message from {player}: {message}")
-
-    def decide(self) -> Optional[str]:
-        if not self.pending_replies:
-            return None
-        player, _ = self.pending_replies.popleft()
-        reply = self.reply_template.format(player=player)
-        print(f"[Agent] Replying to {player}")
-        return reply
-

app.py

@@ -2,10 +2,9 @@ import os
 import select
 import sys
 import time
-from importlib import import_module
 from pathlib import Path
 from threading import Event, Lock, Thread
-from typing import Callable, Optional, Type
+from typing import Optional
 
 from agent import Agent, SimpleAgent
 from telnetclient import TelnetClient
@@ -75,26 +74,14 @@ def run_agent_loop(
     *,
     idle_delay: float = 0.5,
     min_send_interval: float = 1.0,
-    auto_stop: bool = False,
-    auto_stop_idle: float = 2.0,
 ) -> None:
     """Invoke *agent* whenever new output arrives and send its response."""
-    idle_started: Optional[float] = None
-
     while not stop_event.is_set():
         triggered = state.wait_for_output(timeout=idle_delay)
         if stop_event.is_set():
             break
         if not triggered:
-            if auto_stop:
-                now = time.time()
-                if idle_started is None:
-                    idle_started = now
-                elif now - idle_started >= auto_stop_idle:
-                    break
             continue
-
-        idle_started = None
         state.clear_output_event()
         last_output = state.snapshot_output()
         if not last_output:
@@ -142,67 +129,6 @@ def require_env(key: str) -> str:
     return value
 
 
-def build_agent(agent_spec: str) -> Agent:
-    """Instantiate an agent based on ``MISTLE_AGENT`` contents."""
-    normalized = agent_spec.strip()
-    if not normalized:
-        return SimpleAgent()
-
-    key = normalized.lower()
-    if key == "simple":
-        return SimpleAgent()
-
-    builtin_agents = {
-        "explore": ("agent", "ExploreAgent", {}),
-        "communication": ("agent", "CommunicationAgent", {}),
-        "intelligent": (
-            "intelligent_agent",
-            "IntelligentCommunicationAgent",
-            {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")},
-        ),
-        "intelligentcommunication": (
-            "intelligent_agent",
-            "IntelligentCommunicationAgent",
-            {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")},
-        ),
-    }
-
-    if key in builtin_agents:
-        module_name, class_name, kwargs = builtin_agents[key]
-        try:
-            module = import_module(module_name)
-            agent_cls = getattr(module, class_name)
-        except AttributeError as exc:  # pragma: no cover - optional dependency
-            raise RuntimeError(f"{class_name} is not available in agent module") from exc
-        agent = _instantiate_agent(agent_cls, normalized, kwargs)
-        model_name = kwargs.get("model") if kwargs else None
-        if model_name:
-            print(f"[Agent] Using LLM model: {model_name}")
-        return agent
-
-    if ":" in normalized:
-        module_name, class_name = normalized.split(":", 1)
-        if not module_name or not class_name:
-            raise RuntimeError("MISTLE_AGENT must be in 'module:ClassName' format")
-        module = import_module(module_name)
-        agent_cls = getattr(module, class_name)
-        return _instantiate_agent(agent_cls, normalized)
-
-    raise RuntimeError(f"Unknown agent spec '{agent_spec}'.")
-
-
-def _instantiate_agent(
-    agent_cls: Type[Agent], agent_spec: str, kwargs: Optional[dict] = None
-) -> Agent:
-    if not issubclass(agent_cls, Agent):
-        raise RuntimeError(f"{agent_spec} is not an Agent subclass")
-    try:
-        kwargs = kwargs or {}
-        return agent_cls(**kwargs)
-    except TypeError as exc:
-        raise RuntimeError(f"Failed to instantiate {agent_spec}: {exc}") from exc
-
-
 def login(
     client: TelnetClient,
     *,
@@ -244,7 +170,6 @@ def interactive_session(
     poll_interval: float = 0.2,
     receive_timeout: float = 0.2,
     exit_command: str,
-    agent_command: Optional[Callable[[str], None]] = None,
 ) -> None:
     """Keep the Telnet session running, proxying input/output until interrupted."""
     if exit_command:
@@ -267,13 +192,6 @@ def interactive_session(
             line = line.rstrip("\r\n")
             if not line:
                 continue
-            if agent_command and line.lower().startswith("#execute"):
-                parts = line.split(maxsplit=1)
-                if len(parts) == 1:
-                    print("[Agent] Usage: #execute <agent_spec>")
-                else:
-                    agent_command(parts[1])
-                continue
             state.send(client, line)
 
 
@@ -315,13 +233,11 @@ def main() -> int:
     login_prompt = os.environ.get("MISTLE_LOGIN_PROMPT", "")
     exit_command = os.environ.get("MISTLE_EXIT_COMMAND", "")
     agent_mode = os.environ.get("MISTLE_AGENT_MODE", "").lower() in {"1", "true", "yes", "on"}
-    agent_spec = os.environ.get("MISTLE_AGENT", "")
 
     state = SessionState()
     stop_event = Event()
     agent_thread: Optional[Thread] = None
     agent: Optional[Agent] = None
-    ephemeral_agents: list[Thread] = []
 
     with TelnetClient(host=host, port=port, timeout=10.0) as client:
         login(
@@ -333,7 +249,7 @@ def main() -> int:
         )
 
         if agent_mode:
-            agent = build_agent(agent_spec)
+            agent = SimpleAgent()
             agent_thread = Thread(
                 target=run_agent_loop,
                 args=(client, state, agent, stop_event),
@@ -342,30 +258,6 @@ def main() -> int:
             )
             agent_thread.start()
 
-        def run_ephemeral_agent(spec: str) -> None:
-            spec = spec.strip()
-            if not spec:
-                print("[Agent] Usage: #execute <agent_spec>")
-                return
-            try:
-                temp_agent = build_agent(spec)
-            except RuntimeError as exc:
-                print(f"[Agent] Failed to load '{spec}': {exc}", file=sys.stderr)
-                return
-            thread = Thread(
-                target=run_agent_loop,
-                args=(client, state, temp_agent, stop_event),
-                kwargs={
-                    "min_send_interval": 1.0,
-                    "auto_stop": True,
-                    "auto_stop_idle": 2.0,
-                },
-                daemon=True,
-            )
-            ephemeral_agents.append(thread)
-            print(f"[Agent] Executing {spec!r} once")
-            thread.start()
-
         interrupted = False
         try:
             interactive_session(
@@ -373,7 +265,6 @@ def main() -> int:
                 state=state,
                 stop_event=stop_event,
                 exit_command=exit_command,
-                agent_command=None if agent_mode else run_ephemeral_agent,
             )
         except KeyboardInterrupt:
             print()
@@ -382,8 +273,6 @@ def main() -> int:
             stop_event.set()
             if agent_thread:
                 agent_thread.join(timeout=1.0)
-            for thread in ephemeral_agents:
-                thread.join(timeout=1.0)
 
         if interrupted:
             graceful_shutdown(client, exit_command, state=state)

intelligent_agent.py

@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-import re
-import sys
-from collections import deque
-from dataclasses import dataclass, field
-from typing import Deque, Dict, Optional, Pattern, Tuple
-
-try:
-    from litellm import completion
-except ImportError:  # pragma: no cover - optional dependency
-    completion = None  # type: ignore[assignment]
-
-from agent import Agent
-
-
-@dataclass
-class IntelligentCommunicationAgent(Agent):
-    """Agent that uses a language model to answer private tells."""
-
-    model: str = "mistral/mistral-tiny"
-    system_prompt: str = (
-        "Du bist Mistle, ein hilfsbereiter MUD-Bot. "
-        "Antworte freundlich und knapp in deutscher Sprache."
-    )
-    temperature: float = 0.7
-    max_output_tokens: int = 120
-    fallback_reply: str = "Hallo! Ich bin Mistle und ein Bot."
-    tell_pattern: Pattern[str] = field(
-        default_factory=lambda: re.compile(
-            r"^(?P<player>[^\s]+) teilt (d|D)ir mit: (?P<message>.+)$",
-            re.MULTILINE,
-        )
-    )
-    last_output: str = field(default="", init=False)
-    max_history_chars: int = 16_000
-    pending_replies: Deque[Tuple[str, str]] = field(default_factory=deque, init=False)
-    conversation_history: Dict[str, Deque[Tuple[str, str]]] = field(default_factory=dict, init=False)
-
-    def observe(self, output: str) -> None:
-        if not output:
-            return
-        self.last_output = output
-        for match in self.tell_pattern.finditer(output):
-            player = match.group("player").strip()
-            message = match.group("message").strip()
-            if not player:
-                continue
-            self.pending_replies.append((player, message))
-            print(f"[Agent] Received message from {player}: {message}")
-            self._append_history(player, "user", message)
-
-    def decide(self) -> Optional[str]:
-        if not self.pending_replies:
-            return None
-        player, _ = self.pending_replies.popleft()
-        reply_text = self._sanitize_reply(self._generate_reply(player))
-        self._append_history(player, "assistant", reply_text)
-        reply = f"teile {player} mit {reply_text}"
-        print(f"[Agent] Replying to {player} with model output")
-        return reply
-
-    def _generate_reply(self, player: str) -> str:
-        if completion is None:
-            print(
-                "[Agent] litellm is not installed; falling back to default reply",
-                file=sys.stderr,
-            )
-            return self.fallback_reply
-
-        try:
-            response = completion(
-                model=self.model,
-                messages=self._build_messages(player),
-                temperature=self.temperature,
-                max_tokens=self.max_output_tokens,
-            )
-        except Exception as exc:  # pragma: no cover - network/runtime errors
-            print(f"[Agent] Model call failed: {exc}", file=sys.stderr)
-            return self.fallback_reply
-
-        try:
-            content = response["choices"][0]["message"]["content"].strip()
-        except (KeyError, IndexError, TypeError):  # pragma: no cover - defensive
-            return self.fallback_reply
-
-        return content or self.fallback_reply
-
-    def _sanitize_reply(self, text: str) -> str:
-        if not text:
-            return self.fallback_reply
-        collapsed = " ".join(text.split())
-        return collapsed or self.fallback_reply
-
-    def _build_messages(self, player: str) -> list[dict[str, str]]:
-        history = self.conversation_history.get(player)
-        messages: list[dict[str, str]] = [{"role": "system", "content": self.system_prompt}]
-        if not history:
-            return messages
-        for role, content in history:
-            messages.append({"role": role, "content": content})
-        return messages
-
-    def _append_history(self, player: str, role: str, content: str) -> None:
-        if not content:
-            return
-        history = self.conversation_history.setdefault(player, deque())
-        history.append((role, content))
-        self._trim_history(history)
-
-    def _trim_history(self, history: Deque[Tuple[str, str]]) -> None:
-        total_chars = sum(len(content) for _, content in history)
-        while len(history) > 1 and total_chars > self.max_history_chars:
-            _, removed_content = history.popleft()
-            total_chars -= len(removed_content)
-        if history and total_chars > self.max_history_chars:
-            role, content = history.pop()
-            trimmed = content[-self.max_history_chars :]
-            history.append((role, trimmed))

pyproject.toml

@@ -4,6 +4,4 @@ version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.9"
-dependencies = [
-    "litellm>=1.77.4",
-]
+dependencies = []