Moderation and Privacy

from dataclasses import dataclass, field from typing import Protocol import httpx from bot.providers.base import ProviderTimeout, ProviderUnavailable, error_from_status, parse_retry_after @dataclass(frozen=True) class ModerationResult: flagged: bool categories: tuple[str, ...] = () # names of the categories that were flagged scores: dict[str, float] = field(default_factory=dict) class Moderator(Protocol): async def check(self, text: str) -> ModerationResult: ... class NoModeration: async def check(self, text: str) -> ModerationResult: return ModerationResult(False) class KeywordModerator: """A deliberately crude moderator: case-insensitive substring lists per category. Easy to evade; a stop-gap.""" def __init__(self, rules: dict[str, tuple[str, ...]]) -> None: self.rules = rules async def check(self, text: str) -> ModerationResult: lowered = text.lower() hit = tuple(cat for cat, phrases in self.rules.items() if any(p.lower() in lowered for p in phrases)) return ModerationResult(bool(hit), hit) class OpenAIModerator: """OpenAI's Moderation endpoint (POST /v1/moderations). It is a separate service from chat completions.""" def __init__(self, api_key: str, *, model: str = "omni-moderation-latest", base_url: str = "https://api.openai.com/v1", timeout: float = 10.0, transport: httpx.AsyncBaseTransport | None = None) -> None: self.model = model self._client = httpx.AsyncClient(base_url=base_url.rstrip("/"), timeout=timeout, transport=transport, headers={"authorization": f"Bearer {api_key}", "content-type": "application/json"}) async def check(self, text: str) -> ModerationResult: try: response = await self._client.post("/moderations", json={"model": self.model, "input": text}) except httpx.ConnectTimeout as e: raise ProviderUnavailable(f"could not connect: {e!r}") from e except httpx.TimeoutException as e: raise ProviderTimeout(f"no answer in time: {e!r}") from e except httpx.TransportError as e: raise ProviderUnavailable(f"could not connect: {e!r}") from e if response.status_code != 200: raise error_from_status(response.status_code, response.text, retry_after=parse_retry_after(response.headers.get("retry-after"))) result = response.json()["results"][0] flagged = tuple(name for name, hit in result.get("categories", {}).items() if hit) return ModerationResult(bool(result["flagged"]), flagged, dict(result.get("category_scores", {}))) async def aclose(self) -> None: await self._client.aclose() @dataclass(frozen=True) class SafetyPolicy: """What the bot does with moderation results, abuse and privacy. All wording is the operator's to write.""" crisis_categories: tuple[str, ...] = ("self-harm", "self-harm/intent", "self-harm/instructions") crisis_reply: str = ("I'm really sorry you're going through this. I'm an AI and can't give the help you deserve, but people can. " "If you might be in danger, please contact your local emergency services now. " "In the UK and Ireland you can call Samaritans for free, any time, on 116 123.") block_reply: str = "I can't help with that." muted_reply: str = "I'm pausing replies to you for a little while. Please try again later." moderation_down_reply: str = "I can't check that message right now. Please try again in a minute." strike_limit: int = 3 # strikes within strike_window_s ... strike_window_s: float = 3600.0 mute_s: float = 900.0 # ... mute for this long on_moderation_error: str = "closed" # "closed": don't answer if the check fails; "open": answer anyway moderate_output: bool = True disclosure: str | None = "I'm an AI assistant, not a person. Type /privacy to see what I store." privacy_notice: str = ("Privacy: I keep your messages in this chat so I can follow the conversation, and any notes you ask me " "to remember. Messages are deleted after {days} days. Type /export to see everything stored, " "/forget to remove notes, /reset to clear this chat, or /deleteme to erase your chats and notes. " "Messages that break the rules are not stored, and a short record of warnings is kept until it expires.") retention_days: int = 30

# --- who owns which session, strikes, retention, erasure ------------------------------- def link_session(self, session_id: str, user_id: str) -> None: with self.db: self.db.execute("INSERT INTO session_users (session_id, user_id) VALUES (?, ?) " "ON CONFLICT(session_id) DO UPDATE SET user_id = excluded.user_id", (session_id, user_id)) def add_strike(self, user_id: str, category: str, now: float) -> None: with self.db: self.db.execute("INSERT INTO strikes (user_id, category, at) VALUES (?, ?, ?)", (user_id, category, now)) def strikes_since(self, user_id: str, since: float) -> int: return self.db.execute("SELECT COUNT(*) FROM strikes WHERE user_id = ? AND at >= ?", (user_id, since)).fetchone()[0] def latest_strike(self, user_id: str) -> float | None: return self.db.execute("SELECT MAX(at) FROM strikes WHERE user_id = ?", (user_id,)).fetchone()[0] def purge_older_than(self, cutoff: float) -> dict[str, int]: """Retention: delete messages and strikes recorded before `cutoff` (a Unix time). Facts are kept until forgotten.""" with self.db: messages = self.db.execute("DELETE FROM messages WHERE created_at < ?", (cutoff,)).rowcount strikes = self.db.execute("DELETE FROM strikes WHERE at < ?", (cutoff,)).rowcount # a summary whose messages are all gone describes data that no longer exists: drop it too summaries = self.db.execute( "DELETE FROM summaries WHERE session_id NOT IN (SELECT DISTINCT session_id FROM messages)").rowcount self.db.execute("DELETE FROM session_users WHERE session_id NOT IN " "(SELECT DISTINCT session_id FROM messages) AND session_id NOT IN (SELECT session_id FROM summaries)") return {"messages": messages, "strikes": strikes, "summaries": summaries} def delete_user(self, user_id: str) -> dict[str, int]: """Erase what the user can ask us to erase: their sessions, summaries and facts. Strikes stay until retention expires.""" with self.db: sessions = [r[0] for r in self.db.execute("SELECT session_id FROM session_users WHERE user_id = ?", (user_id,))] messages = summaries = 0 for sid in sessions: messages += self.db.execute("DELETE FROM messages WHERE session_id = ?", (sid,)).rowcount summaries += self.db.execute("DELETE FROM summaries WHERE session_id = ?", (sid,)).rowcount self.db.execute("DELETE FROM session_users WHERE user_id = ?", (user_id,)) facts = self.db.execute("DELETE FROM facts WHERE user_id = ?", (user_id,)).rowcount return {"sessions": len(sessions), "messages": messages, "summaries": summaries, "facts": facts}

import re EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+") LONG_DIGITS = re.compile(r"(?<!\d)(?:\d[ -]?){13,19}(?!\d)") # card-number-shaped runs PHONE = re.compile(r"(?<![\w])(?:\+?\d[\d ()-]{7,}\d)(?![\w])") # 9+ digits with optional separators def redact(text: str) -> str: """Hide the most obvious personal identifiers before text goes into a log. This is a best-effort filter, not a guarantee: names, addresses and anything written in an unusual form will pass straight through. Log as little text as you can; redaction is the second line of defence. """ text = EMAIL.sub("[email]", text) text = LONG_DIGITS.sub("[number]", text) text = PHONE.sub("[number]", text) return text

import asyncio import time from bot.context import build_system, fit_window, message_tokens from bot.guardrails import GuardConfig, check_input, check_output, sanitise_fact from bot.moderation import ModerationResult, Moderator, SafetyPolicy from bot.persona import Persona from bot.providers import Message, Provider, ProviderError from bot.store import Store SUMMARISE_SYSTEM = ( "You condense conversations. Write a short, factual summary of the conversation you are given, " "keeping names, decisions and open questions. Reply with the summary only." ) HELP = """Commands: /remember <text> ask me to remember something about you /facts list what I remember /forget <number> forget one thing (or /forget all) /reset clear this conversation /export show everything I have stored for you as JSON /privacy what I store and for how long /deleteme erase your chats and notes""" async def one_shot(provider: Provider, text: str) -> str: """A bare model call: no memory, every question stands alone.""" return (await provider.complete([{"role": "user", "content": text}])).text class Bot: """A provider plus a memory: stored history per session, a token budget, summaries and user facts.""" def __init__(self, provider: Provider, system: str | None = None, *, persona: Persona | None = None, guard: GuardConfig | None = None, store: Store | None = None, moderator: Moderator | None = None, safety: SafetyPolicy | None = None, clock=time.time, history_budget: int = 2000, summarise: bool = False, max_tokens: int = 1024) -> None: self.provider = provider self.moderator = moderator self.safety = safety or (SafetyPolicy() if moderator else None) # no moderator and no policy: no safety layer self.clock = clock self.persona = persona self.system = persona.render() if persona else system # sent with EVERY call, outside the history window self.guard = guard or GuardConfig() self.events: list[tuple[str, str]] = [] # (session_id, flag): chapter 11 turns these into logs self.store = store or Store(clock=clock) self.history_budget = history_budget self.summarise = summarise self.max_tokens = max_tokens self._locks: dict[str, asyncio.Lock] = {} async def handle(self, session_id: str, text: str, *, user_id: str | None = None) -> str: user_id = user_id or session_id verdict = check_input(text, self.guard) if not verdict.ok: self.events.append((session_id, verdict.flag)) return verdict.text text = verdict.text command = await self._command(session_id, user_id, text) if command is not None: return command if self.safety: refusal = await self._screen_input(session_id, user_id, text) if refusal is not None: return refusal lock = self._locks.setdefault(session_id, asyncio.Lock()) async with lock: # one message at a time per session, or two quick messages would see the same history return await self._answer(session_id, user_id, text) async def _answer(self, session_id: str, user_id: str, text: str) -> str: rows = self.store.messages(session_id) summary = self.store.get_summary(session_id) if summary: rows = [r for r in rows if r[0] > summary[1]] new: Message = {"role": "user", "content": text} if self.summarise: rows, summary = await self._maybe_summarise(session_id, rows, summary, new) window = fit_window([m for _, m in rows] + [new], self.history_budget) system = build_system(self.system, [f for _, f in self.store.facts(user_id)], summary[0] if summary else None) reply = await self.provider.complete(window, system=system, max_tokens=self.max_tokens) answer, flags = check_output(reply.text, self.persona, self.guard) self.events += [(session_id, flag) for flag in flags] if self.safety and self.safety.moderate_output and self.moderator and not flags: try: if (await self.moderator.check(answer)).flagged: answer = self.safety.block_reply self.events.append((session_id, "output_moderated")) except ProviderError: if self.safety.on_moderation_error == "closed": self.events.append((session_id, "moderation_unavailable")) return self.safety.moderation_down_reply # not stored: we could not vouch for it first_message = not rows and summary is None self.store.add_exchange(session_id, text, answer) # only now, and the CHECKED text: a leak never re-enters the history self.store.link_session(session_id, user_id) if first_message and self.safety and self.safety.disclosure: return f"{self.safety.disclosure}\n\n{answer}" return answer async def _screen_input(self, session_id: str, user_id: str, text: str) -> str | None: """Mute check, then moderation. A returned string is the reply to send instead of calling the model.""" policy, now = self.safety, self.clock() count = self.store.strikes_since(user_id, now - policy.strike_window_s) latest = self.store.latest_strike(user_id) if count >= policy.strike_limit and latest is not None and now < latest + policy.mute_s: self.events.append((session_id, "muted")) return policy.muted_reply if not self.moderator: return None try: result: ModerationResult = await self.moderator.check(text) except ProviderError: self.events.append((session_id, "moderation_unavailable")) return policy.moderation_down_reply if policy.on_moderation_error == "closed" else None if not result.flagged: return None if any(c in policy.crisis_categories for c in result.categories): self.events.append((session_id, "crisis_input")) # no strike, and nothing from this message is stored return policy.crisis_reply self.store.add_strike(user_id, ",".join(result.categories) or "flagged", now) self.events.append((session_id, "moderated:" + ",".join(result.categories))) return policy.block_reply async def purge_expired(self) -> dict[str, int]: """Apply the retention period. Call it on a schedule (chapter 12).""" days = self.safety.retention_days if self.safety else None return self.store.purge_older_than(self.clock() - days * 86400) if days else {} async def _maybe_summarise(self, session_id, rows, summary, new): """If the unsummarised history is over budget, fold its oldest exchanges into the summary.""" total = sum(message_tokens(m) for _, m in rows) + message_tokens(new) if total <= self.history_budget: return rows, summary split, remaining = 0, total while remaining > self.history_budget // 2 and split + 2 <= len(rows): remaining -= message_tokens(rows[split][1]) + message_tokens(rows[split + 1][1]) split += 2 if split == 0: return rows, summary transcript = "\n".join(f"{m['role']}: {m['content']}" for _, m in rows[:split]) prompt = (f"Earlier summary:\n{summary[0]}\n\n" if summary else "") + f"Conversation to summarise:\n{transcript}" try: reply = await self.provider.complete([{"role": "user", "content": prompt}], system=SUMMARISE_SYSTEM, max_tokens=300) except ProviderError: return rows, summary # could not summarise: fall back to plain truncation, never fail the user's message upto = rows[split - 1][0] self.store.set_summary(session_id, reply.text, upto) return rows[split:], (reply.text, upto) async def _command(self, session_id: str, user_id: str, text: str) -> str | None: word, _, rest = text.strip().partition(" ") rest = rest.strip() match word.lower(): case "/help": return HELP case "/reset": return f"Conversation cleared ({self.store.clear_session(session_id)} messages)." case "/remember": if not rest: return "Usage: /remember <something to remember>" fact = sanitise_fact(rest, self.guard) if fact is None: return f"I can only remember short notes (up to {self.guard.max_fact_chars} characters)." return f"Remembered as #{self.store.add_fact(user_id, fact)}." case "/facts": facts = self.store.facts(user_id) return "\n".join(f"#{i} {t}" for i, t in facts) if facts else "I am not remembering anything about you." case "/forget": if rest.lower() == "all": return f"Forgot {self.store.forget(user_id)} thing(s)." if rest.isdigit(): n = self.store.forget(user_id, int(rest)) return "Forgotten." if n else f"I have no fact #{rest}." return "Usage: /forget <number> or /forget all" case "/export": return self.store.export(session_id, user_id) case "/privacy": policy = self.safety or SafetyPolicy() return policy.privacy_notice.format(days=policy.retention_days) case "/deleteme": gone = self.store.delete_user(user_id) self.store.clear_session(session_id) return (f"Done. I erased {gone['messages']} message(s) and {gone['facts']} note(s) from {gone['sessions']} chat(s) " "(a short record of rule-breaking warnings, if any, stays until it expires).") return None