# Moduli — tour ragionato

Un modulo per sezione. Le funzioni e le classi elencate sono quelle
dell'API *pubblica* del modulo — leaky implementation details sono
omesse (sono `_underscore_prefixed` e cambiano spesso).

---

## `ansi.py` — stripper ANSI

**Scopo.** Funzione pura per togliere gli escape ANSI/VT100 prima che il
testo venga dato in pasto a euristiche di parsing (stanze, login prompt,
mapper UI). È una foglia del grafo: non importa nulla e non sa nulla
del MUD.

**API pubblica.**

```python
def strip_ansi(text: str) -> str: ...
def strip_ansi_bytes(data: bytes) -> bytes: ...
def visible_len(text: str) -> int: ...
```

**Internals.** Una regex (`_ANSI_RE`) compilata a import‑time che
copre CSI, OSC, charset designation e DEC private mode. Sostituzione
`re.sub` lineare. I single ESC byte senza sequenza sono lasciati
intatti (non sono errori, sono escape non validi che la stampante
gestirà).

---

## `deadsouls.py` — euristiche Dead Souls / LP‑MUD

**Scopo.** Riconoscere stanze dal testo ricevuto dal server e
canonicalizzare le direzioni delle uscite. *Solo* testo e solo
euristica — nessuna conoscenza di IAC, GMCP, asyncio.

**API pubblica.**

```python
EXIT_ALIASES: dict[str, str]
DIRECTION_OFFSETS: dict[str, tuple[int, int]]
OPPOSITE: dict[str, str]    # importato da mapper

def parse_exits(line: str) -> list[str]

@dataclass
class RoomBlock:
    name: str
    body: str
    exits: list[str]

def parse_room_block(block: str) -> RoomBlock | None

class RollingRoomParser:
    def __init__(self) -> None
    def feed_text(self, text: str) -> list[RoomBlock]
    def flush(self) -> list[RoomBlock]
```

**Strategia di parsing blocco stanza.** Basata su linee (non regex
multi‑linea):

1. Accumula testo in un buffer interno.
2. Aggiungi il nuovo testo.
3. Split per `\n`; conserva l'ultimo pezzo parziale come buffer.
4. Per ogni riga completa: se contiene `Exits:` → chiudi il blocco
   corrente (nome = prima riga non vuota, body = resto) ed emetti un
   `RoomBlock`. Altrimenti appendi al blocco corrente.
5. Il blocco corrente rimane bufferizzato per la prossima chiamata.

**Tolleranze.** ANSI già strippato a monte. Linee con la parola
"exits" in mezzo a dialoghi NPC sono al sicuro: `parse_exits`
controlla il *formato* (bracketed/unbracketed, ":" o "-") prima di
popolare la lista uscite.

---

## `gmcp.py` — parser GMCP

**Scopo.** Decodificare il payload GMCP (`IAC SB GMCP <module> [<json>]
IAC SE`) in `GmcpEvent(module, data)`. Validare la *forma* del module
name. Estrarre i campi tolleranti (sinonimi) di `Char.Vitals` /
`Room.Info`.

**API pubblica.**

```python
@dataclass(frozen=True)
class GmcpEvent:
    module: str
    data: Any

# moduli standard
MODULE_CHAR_LOGIN: str
MODULE_CHAR_LOGOUT: str
MODULE_CHAR_VITALS: str
MODULE_CHAR_STATUS: str
MODULE_CHAR_INFO: str
MODULE_ROOM_INFO: str
MODULE_ROOM_WALK_DIR: str

def parse_gmcp(data: bytes) -> GmcpEvent | None
def extract_vitals(payload: Any) -> dict[str, int]
def extract_room_info(payload: Any) -> tuple[str, str, str, str | None] | None
```

**Sinonimi tollerati (vitale).**

| Chiave canonica | Sinonimi accettati (case‑sensitive) |
| --- | --- |
| `hp` / `max_hp` | `hp`, `health`, `life` |
| `sp` / `max_sp` | `sp`, `mana`, `magic` |
| `mv` / `max_mv` | `mv`, `move`, `stam` |

**Room.Info.** Il primo argomento è l'`id` stabile della stanza: se
`num`/`vnum`/`id` è presente e numerico, uso `vnum{int(num)}`;
altrimenti `name.strip().lower()`. Questo è ciò che permette al mapper
di ri‑identificare la stessa stanza tra sessioni.

---

## `mapper.py` — grafo stanze + render ASCII

**Scopo.** Persistenza in‑memory delle stanze visitate e rendering
ASCII del grafo. Le coordinate vengono assegnate **solo** al momento
del render (lazy layout).

**API pubblica.**

```python
CELL_W: int = 22

@dataclass
class Room:
    id: str
    name: str
    body: str
    exits: dict[str, str]   # direzione -> id vicino

class Mapper:
    def __init__(self) -> None
    def visit(self, name, body, exits, *, id_override=None) -> str
    def link_rooms(self, from_id, direction, to_id) -> None
    def move(self, direction: str) -> bool
    @property
    def current_room(self) -> Room | None
    @property
    def rooms(self) -> dict[str, Room]
    def render(self, *, max_cols=12, max_rows=16) -> str
```

**Algoritmo di layout (3 fasi).**

1. **BFS dal current.** Anchor `(0, 0)` ed esplora i forward edges
   noti. Su collisione, nudge `+1` su `y` (fino a 16 tentativi).
2. **Orphan placement.** Per ogni stanza non ancora piazzata, prova
   un edge in *qualsiasi direzione* verso un placed‑room (forward o
   reverse). Itera finché non ci sono più avanzamenti. Questo gestisce
   il backtracking `n → s` dove la `s`‑room non era collegata via
   `exits[e]` ma tramite il movimento utente.
3. **Stack in basso a destra.** Le stanze veramente isolate finiscono
   qui. Se ne trovi mai una, hai probabilmente un bug di parsing: ogni
   stanza raggiungibile ha almeno un edge.

**Snapshot in `_place_adjacent`.** `placed = list(coords.items())` viene
copiato *prima* di mutare `coords` (altrimenti l'iterazione successiva
vedrebbe il nuovo elemento appena inserito e potrebbe creare cicli).

**Render.** Coordinate y ordinate top‑down (y grande = nord). Celle di
larghezza fissa `CELL_W = 22`. Marker `»` per la stanza corrente, `·`
per le altre. I link `e → w` sono tratteggi `───`, le verticali `│`.

---

## `logger.py` — `EventLogger`

**Scopo.** Sink append‑only per eventi di sessione. Due formati
paralleli: JSON Lines (machine‑readable) e Markdown (human‑readable).
File aperti in write‑mode (non append): ogni sessione riparte pulito.

**API pubblica.**

```python
class EventLogger:
    def __init__(
        self,
        json_path: str | None = None,
        md_path: str | None = None,
        *,
        host: str = "",
        port: int = 0,
        user: str = "",
    ) -> None

    def close(self) -> None
    def event(self, kind: str, data: dict[str, Any] | None = None) -> None

    # shortcut tipizzati
    def session_start(self, host: str, port: int, user: str) -> None
    def session_end(self, reason: str = "") -> None
    def disconnect(self, reason: str = "") -> None
    def text(self, raw_ansi: str) -> None
    def room(self, rid: str, name: str, exits: list[str]) -> None
    def gmcp(self, module: str, payload: Any) -> None
    def vitals(
        self, *,
        hp=None, sp=None, mv=None,
        max_hp=None, max_sp=None, max_mv=None,
    ) -> None
    def line_sent(self, line: str, *, redacted: bool = False) -> None
    def local_cmd(self, cmd, args=None, *, redacted: bool = False) -> None
    def login(self, phase: str, value: str = "", *, redact: bool = True) -> None
```

**Thread safety.** `threading.Lock` interno. Oggi è uncontended
(asyncio è single‑threaded) ma difende da futuri worker thread. Tutte
le scritture passano per `_write_json_unlocked` + `_append_md_unlocked`
+ `_flush_unlocked` sotto il lock.

**Sicurezza password.** Tre punti di redazione:

- `line_sent(line, redacted=True)` → testo registrato come `"***"`.
- `local_cmd(cmd, args, redacted=True)` → args list collassato a `["***"]`.
- `login(phase, value, redact=True)` (default) → value registrato come `"***"`.

**Idempotenza.** `close()` è idempotente; `__init__` solleva `OSError`
in caso di errore di I/O e chiude gli handle aperti prima di rilanciare
（nessun file descriptor leak).

---

## `telnet.py` — trasporto telnet

**Scopo.** Implementare la state machine IAC (RFC 854) sopra un
`asyncio.StreamReader`/`StreamWriter` e gestire la negoziazione
opzioni (ECHO/NAWS/TTYPE/GMCP). **Nessuna** conoscenza del MUD o della
UI.

**API pubblica.**

```python
# costanti di protocollo
IAC, DO, DONT, WILL, WONT, SB, SE  = 255, 253, 254, 251, 252, 250, 240
EOR, GA                            = 239, 249
OPT_ECHO, OPT_TTYPE, OPT_NAWS, OPT_GMCP = 1, 24, 31, 201

# eventi
@dataclass
class TextEvent:     text: str
@dataclass
class NegotiateEvent: cmd: int; opt: int
@dataclass
class SubnegEvent:    opt: int; data: bytes
@dataclass
class PromptEvent:    ...

TelnetEvent = TextEvent | NegotiateEvent | SubnegEvent | PromptEvent

class IACParser:
    """Push parser: feed bytes, get events."""
    def feed(self, buf: bytes) -> list[TelnetEvent]

class TelnetConnection:
    def __init__(self, host, port, *,
                 naws_size=None, ttype="XTERM-256COLOR", want_gmcp=True) -> None
    async def connect(self) -> None
    async def close(self) -> None
    async def read_loop(self, handler: Callable[[TelnetEvent], Awaitable[None]]) -> None
    async def send(self, data: bytes) -> None
    async def send_line(self, line: str) -> None
    async def report_window_size(self, cols: int, rows: int) -> None
```

**State machine (`S` enum).**

```
NORMAL → byte normale               → NORMAL
NORMAL → IAC                        → IAC
IAC   → IAC                         → NORMAL (literal)
IAC   → DO|DONT|WILL|WONT           → NEGOTIATE
IAC   → SB                          → SB_OPTION
IAC   → SE                          → NORMAL (stray)
IAC   → EOR|GA                      → NORMAL (PromptEvent emesso)
IAC   → altro (NOP/DM/AO)           → NORMAL
NEGOTIATE → <opt>                    → NORMAL (NegotiateEvent emesso)
SB_OPTION → <opt>                    → SB_DATA
SB_DATA → IAC                        → SB_IAC
SB_IAC  → SE                         → NORMAL (SubnegEvent emesso)
SB_IAC  → IAC                        → SB_DATA (literal)
SB_IAC  → altro                      → NORMAL (abort)
```

**Negoziazione opzioni.**

| cmd | opt | Risposta client |
| --- | --- | --- |
| `DO` | `ECHO`/`NAWS`/`TTYPE`/`GMCP` se supportato | `WILL` (+ side‑effect: invio dimensione finestra per NAWS, set `expects_ttype_send` per TTYPE) |
| `DO` | altro | `WONT` |
| `DONT` | qualsiasi | `WONT` |
| `WILL` | qualsiasi | `DONT` (server offers non ci servono in v0.2) |
| `WONT` | qualsiasi | `DONT` |

**Side‑channel GMCP.** Se `want_gmcp=True`, `connect()` invia
proattivamente `IAC WILL GMCP` prima di leggere. Alcuni MUD preferiscono
l'offerta client piuttosto che `DO` server.

---

## `cli.py` — orchestratore

**Scopo.** Compostare tutti i moduli sopra in un'unica sessione
interattiva. Possiede il prompt, la status bar, il dispatcher telnet,
la macchina a stati di auto‑login e i comandi slash locali.

**API pubblica.**

```python
@dataclass
class ClientState:
    mapper: Mapper
    parser: RollingRoomParser
    vitals: dict[str, int]
    gmcp_enabled: bool
    auto_user_sent: bool
    auto_password_prompted: bool
    auto_password_sent: bool
    logger: EventLogger | None

def main(argv: list[str] | None = None) -> int
```

**Componenti.**

- `_format_prompt(state, user)` → `FormattedText` con `[room] <user> > `.
- `_format_status_bar(state)` → `FormattedText` con `room`, GMCP
  on/off, hp/sp/mv.
- `_handle_gmcp(data, state)` → dispatch modulo → vitale/room/mapper.
- `_maybe_autologin(text, conn, state, args)` → matching regex su
  `What name do you wish?` / `Password:` e invio automatico.
- `_on_telnet_event(ev, conn, state, args)` → dispatcher principale.
- `_input_loop(session, conn, state, user)` → prompt loop async.
- `_handle_local_command(line, conn, state) -> bool` → comandi slash.
- `_run(args)` → orchestrata con `asyncio.wait(FIRST_COMPLETED)`,
  `conn.close()` *prima* del drain delle pending task.

**Comandi slash.**

| Slash | Implementazione | Note |
| --- | --- | --- |
| `/quit` `/exit` `/q` | `conn.close()` + `SystemExit(0)` | non loggato |
| `/help` `/?` | `_print_help()` | loggato |
| `/map` | `mapper.render()` stampato | loggato |
| `/where` | room corrente + uscite | loggato |
| `/look` | `conn.send_line("look")` | loggato |
| `/stats` | stanze, GMCP, vitali | loggato |
| `/vitals` | `state.vitals` | loggato |
| `/clear` | `prompt_toolkit.clear()` | loggato |
| `/password PWD...` | redacted + `conn.send_line(pwd)` | loggato (redatto) |
| altro `/...` | `return False` → forward al MUD | loggato come `line_sent` da `_input_loop` |

**Auto‑login state machine.**

```
                 text: "What name do you wish?"
                       ↓
        args.user exists & not auto_user_sent
                       ↓
   send_line(args.user); auto_user_sent = True
                       ↓
                text: "Password:"
                       ↓
    auto_password_prompted = True
    ├── args.password exists → send_line, auto_password_sent = True
    └── args.password ""     → stampa suggerimento /password PWD
```