""" working.py — Working Memory (short-term, in-RAM) Ultime N conversazioni per contesto immediato. """ from collections import deque from dataclasses import dataclass, field from datetime import datetime @dataclass class WorkingEntry: role: str content: str ts: float = field(default_factory=lambda: datetime.now().timestamp()) class WorkingMemory: def __init__(self, max_entries: int = 40): self._buf: deque[WorkingEntry] = deque(maxlen=max_entries) def add(self, role: str, content: str): self._buf.append(WorkingEntry(role=role, content=content)) def get_recent(self, n: int = 10) -> list[dict]: import time as _t _ttl = _t.time() - 3600 # D7: 1h TTL — entry più vecchie non distorcono il contesto entries = [e for e in list(self._buf)[-n:] if e.ts >= _ttl] return [{"role": e.role, "content": e.content} for e in entries] def get_context_string(self, n: int = 6) -> str: recent = self.get_recent(n) if not recent: return "" lines = [] for m in recent: prefix = "Utente" if m["role"] == "user" else "AI" # S571: 200→400 — evita di tagliare risposte con codice corto # S600: 400→600 — working memory snippet più lungo per risposte con codice lines.append(f"{prefix}: {m['content'][:600]}") return "Conversazione recente:\n" + "\n".join(lines) def clear(self): self._buf.clear() def stats(self) -> dict: return {"entries": len(self._buf), "max": self._buf.maxlen}