Phase 4: write-action safety & guardrails (#22)

Add writes.py: the single gate every write tool must pass. A write happens
only when confirm=true AND writes are enabled (off by default); otherwise a
dry-run preview is returned. Every executed write is appended to an audit
log. New `familiar writes [enable|disable|status]` CLI and a read-only
write_status MCP tool.

81 tests, coverage 92%. Closes #22.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/familiar/__main__.py

@@ -4,17 +4,20 @@ With no arguments it runs the MCP server over stdio (how Claude Desktop /
 Claude Code / Cowork launch it). The auth subcommands are meant to be run by
 hand in a terminal:
 
-    familiar login        # add a character via EVE SSO (run once per character)
+    familiar login         # add a character via EVE SSO (run once per character)
-    familiar status       # list logged-in characters
+    familiar status        # list logged-in characters
-    familiar logout       # forget ALL characters
+    familiar logout        # forget ALL characters
-    familiar logout NAME  # forget one character (by name or id)
+    familiar logout NAME   # forget one character (by name or id)
+    familiar writes        # show whether write actions are enabled
+    familiar writes enable # allow write actions (mail, waypoints, ...)
+    familiar writes disable
 """
 
 from __future__ import annotations
 
 import sys
 
-from . import auth
+from . import auth, writes
 
 
 def _login() -> int:
@@ -48,9 +51,22 @@ def _logout() -> int:
     return 0
 
 
+def _writes() -> int:
+    sub = sys.argv[2] if len(sys.argv) > 2 else "status"
+    if sub == "enable":
+        writes.set_writes(True)
+    elif sub == "disable":
+        writes.set_writes(False)
+    elif sub != "status":
+        print("Usage: familiar writes [enable|disable|status]")
+        return 1
+    print(f"Writes are {'ENABLED' if writes.writes_enabled() else 'disabled'}.")
+    return 0
+
+
 def main() -> None:
     command = sys.argv[1] if len(sys.argv) > 1 else None
-    handlers = {"login": _login, "status": _status, "logout": _logout}
+    handlers = {"login": _login, "status": _status, "logout": _logout, "writes": _writes}
     if command in handlers:
         raise SystemExit(handlers[command]())
     if command in {"-h", "--help", "help"}:

src/familiar/server.py

@@ -11,7 +11,7 @@ import datetime as _dt
 
 from fastmcp import FastMCP
 
-from . import __version__, account, industry, market, structures, trade, universe
+from . import __version__, account, industry, market, structures, trade, universe, writes
 
 mcp: FastMCP = FastMCP(
     name="familiar",
@@ -51,3 +51,6 @@ account.register(mcp)
 industry.register(mcp)
 trade.register(mcp)
 structures.register(mcp)
+
+# Phase 4: write-action safety (the gate every write tool goes through).
+writes.register(mcp)

src/familiar/writes.py

@@ -0,0 +1,96 @@
+"""Guardrails for write/command actions.
+
+familiar is safe by default: a write only happens when **all** of these hold:
+
+1. the caller passes ``confirm=True`` (otherwise it gets a dry-run preview), and
+2. writes are enabled globally (`familiar writes enable`, default off).
+
+Every executed write is appended to an audit log on disk. This module is the
+single choke point every write tool must go through.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from . import config
+
+
+class WriteDisabled(RuntimeError):
+    """A write was confirmed but writes are globally disabled."""
+
+
+def _flag_file():
+    return config.data_dir() / "writes_enabled"
+
+
+def writes_enabled() -> bool:
+    """True if writes are allowed (env override or the on-disk toggle)."""
+    env = os.environ.get("FAMILIAR_ENABLE_WRITES")
+    if env is not None:
+        return env.strip().lower() in {"1", "true", "yes", "on"}
+    return _flag_file().exists()
+
+
+def set_writes(enabled: bool) -> None:
+    """Turn the on-disk write toggle on or off."""
+    flag = _flag_file()
+    if enabled:
+        flag.write_text("enabled\n", encoding="utf-8")
+    elif flag.exists():
+        flag.unlink()
+
+
+def audit_log_path():
+    return config.data_dir() / "audit.log"
+
+
+def _audit(entry: dict[str, Any]) -> None:
+    line = json.dumps({"ts": time.time(), **entry})
+    with audit_log_path().open("a", encoding="utf-8") as fh:
+        fh.write(line + "\n")
+
+
+async def guard(
+    action: str,
+    params: dict[str, Any],
+    confirm: bool,
+    execute: Callable[[], Awaitable[Any]],
+) -> dict[str, Any]:
+    """Run a write through the safety gate.
+
+    * ``confirm=False`` -> return a dry-run preview, execute nothing.
+    * ``confirm=True`` but writes disabled -> raise :class:`WriteDisabled`.
+    * ``confirm=True`` and writes enabled -> execute, audit, return the result.
+    """
+    if not confirm:
+        return {
+            "dry_run": True,
+            "action": action,
+            "params": params,
+            "note": "Nothing was changed. Pass confirm=true to execute (writes must be enabled).",
+        }
+    if not writes_enabled():
+        raise WriteDisabled(
+            f"Writes are disabled, so '{action}' was not performed. "
+            "Run `familiar writes enable` to allow write actions."
+        )
+    result = await execute()
+    _audit({"action": action, "params": params})
+    return {"executed": True, "action": action, "result": result}
+
+
+def register(mcp) -> None:
+    """Register a read-only tool reporting whether writes are enabled."""
+
+    @mcp.tool
+    async def write_status() -> dict[str, Any]:
+        """Whether familiar is currently allowed to perform write actions."""
+        return {
+            "writes_enabled": writes_enabled(),
+            "note": "Enable with `familiar writes enable`; write tools also need confirm=true.",
+        }

tests/test_writes.py

@@ -0,0 +1,65 @@
+"""Write-safety guardrails: toggle, dry-run, audit log."""
+
+import asyncio
+import json
+
+import pytest
+
+from familiar import writes
+
+
+@pytest.fixture
+def tmp_data(tmp_path, monkeypatch):
+    monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
+    monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False)
+    return tmp_path
+
+
+async def _exec():
+    return {"ok": True}
+
+
+def test_disabled_by_default(tmp_data):
+    assert writes.writes_enabled() is False
+
+
+def test_toggle_on_disk(tmp_data):
+    writes.set_writes(True)
+    assert writes.writes_enabled() is True
+    writes.set_writes(False)
+    assert writes.writes_enabled() is False
+
+
+def test_env_override(tmp_data, monkeypatch):
+    monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "true")
+    assert writes.writes_enabled() is True
+    monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "0")
+    assert writes.writes_enabled() is False
+
+
+def test_dry_run_does_not_execute(tmp_data):
+    ran = []
+
+    async def execute():
+        ran.append(1)
+        return "done"
+
+    result = asyncio.run(writes.guard("send_mail", {"to": 1}, confirm=False, execute=execute))
+    assert result["dry_run"] is True
+    assert not ran  # never executed
+
+
+def test_confirm_but_disabled_raises(tmp_data):
+    with pytest.raises(writes.WriteDisabled):
+        asyncio.run(writes.guard("send_mail", {}, confirm=True, execute=_exec))
+
+
+def test_confirm_enabled_executes_and_audits(tmp_data):
+    writes.set_writes(True)
+    result = asyncio.run(writes.guard("send_mail", {"subject": "hi"}, confirm=True, execute=_exec))
+    assert result["executed"] is True
+    assert result["result"] == {"ok": True}
+    lines = writes.audit_log_path().read_text(encoding="utf-8").strip().splitlines()
+    entry = json.loads(lines[-1])
+    assert entry["action"] == "send_mail"
+    assert "ts" in entry