diff --git a/db/migrations/082_olfactory.sql b/db/migrations/082_olfactory.sql new file mode 100644 index 0000000..644ac1b --- /dev/null +++ b/db/migrations/082_olfactory.sql @@ -0,0 +1,61 @@ +-- Migration 082: olfactory cortex — Phase 1 schema +-- +-- Olfactory cortex is the ONE sensory modality that bypasses thalamus. +-- Olfactory bulb projects directly to piriform cortex + amygdala + +-- entorhinal cortex. This direct route is why smells produce such +-- strong emotional/memory recall (Proust effect). +-- +-- brainctl analog: a "direct binding" channel that, by-passing the +-- normal thalamus → cortex → amygdala flow, immediately binds an +-- incoming content type to a stored valence + an episodic memory +-- pointer. Useful for input modalities where the brain decides this +-- pattern is too primal for the standard W(m) gate. +-- +-- Phase 1 ships: +-- olfactory_imprints — direct (content_hash, valence, memory_id) +-- bindings that bypass standard write gates +-- olfactory_state — single row tracking total imprints + rate +-- +-- Phase 2 wires olfactory_imprint into amygdala_tag for the bypass +-- path. Phase 3 lets olfactory_query return bound memories directly +-- (Proust-style fast emotional recall). +-- +-- Rollback: +-- DROP TABLE IF EXISTS olfactory_imprints; +-- DROP TABLE IF EXISTS olfactory_state; +-- DELETE FROM schema_version WHERE version = 82; +-- +-- IDEMPOTENT. + +CREATE TABLE IF NOT EXISTS olfactory_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + total_imprints INTEGER NOT NULL DEFAULT 0, + enforcement_mode TEXT NOT NULL DEFAULT 'shadow' CHECK(enforcement_mode IN ( + 'shadow', 'enforce', 'disabled' + )), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); +INSERT OR IGNORE INTO olfactory_state (id) VALUES (1); + +CREATE TABLE IF NOT EXISTS olfactory_imprints ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + imprinted_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + content_hash TEXT NOT NULL, + content_kind TEXT, -- e.g. 'text_pattern', 'entity_name', 'phrase' + valence REAL NOT NULL CHECK(valence BETWEEN -1.0 AND 1.0), + arousal REAL NOT NULL DEFAULT 0.5 CHECK(arousal BETWEEN 0.0 AND 1.0), + bound_memory_id INTEGER, -- optional memory pointer this imprint resurrects + bound_entity_id INTEGER, -- optional entity pointer + agent_id TEXT, + times_recalled INTEGER NOT NULL DEFAULT 0, + last_recalled_at TEXT, + notes TEXT, + UNIQUE (content_hash, agent_id) +); +CREATE INDEX IF NOT EXISTS idx_oi_recent ON olfactory_imprints(imprinted_at); +CREATE INDEX IF NOT EXISTS idx_oi_content ON olfactory_imprints(content_hash); +CREATE INDEX IF NOT EXISTS idx_oi_valence ON olfactory_imprints(valence); + +INSERT OR IGNORE INTO schema_version (version, description, applied_at) +VALUES (82, 'olfactory cortex Phase 1: direct sensory-emotional binding (2 tables)', + strftime('%Y-%m-%dT%H:%M:%S', 'now')); diff --git a/src/agentmemory/mcp_server.py b/src/agentmemory/mcp_server.py index 95e5f45..2bf7030 100755 --- a/src/agentmemory/mcp_server.py +++ b/src/agentmemory/mcp_server.py @@ -63,6 +63,7 @@ mcp_tools_meb, mcp_tools_merge, mcp_tools_neuro, + mcp_tools_olfactory, mcp_tools_pfc, mcp_tools_policy, mcp_tools_procedural, @@ -106,6 +107,7 @@ mcp_tools_meb, mcp_tools_merge, mcp_tools_neuro, + mcp_tools_olfactory, mcp_tools_pfc, mcp_tools_policy, mcp_tools_procedural, diff --git a/src/agentmemory/mcp_tools_olfactory.py b/src/agentmemory/mcp_tools_olfactory.py new file mode 100644 index 0000000..930e99f --- /dev/null +++ b/src/agentmemory/mcp_tools_olfactory.py @@ -0,0 +1,276 @@ +"""brainctl MCP tools — olfactory cortex (direct sensory-emotional binding). + +Phase 1: direct imprint of (content_hash, valence, memory pointer) +that bypasses the standard W(m) write gate + thalamus routing — +just like olfaction in biology, which is the only sense that doesn't +relay through thalamus before reaching amygdala. + +Use for input patterns the operator wants to flag as primally +significant (Proust-effect smell-equivalents): a specific phrase +that should always resurface a memory, an entity name that always +fires a particular valence, etc. + +Default enforcement_mode = 'shadow' — imprints recorded but not +automatically routed into amygdala / retrieval. Phase 2 wires the +bypass. +""" +from __future__ import annotations + +import hashlib +import sqlite3 +from collections.abc import Iterable +from pathlib import Path +from typing import Any + +from mcp.types import Tool + +from agentmemory.lib.mcp_helpers import open_db +from agentmemory.paths import get_db_path + +DB_PATH: Path = get_db_path() + +VALID_ENFORCEMENT_MODES = {"shadow", "enforce", "disabled"} + + +def _db() -> sqlite3.Connection: + return open_db(str(DB_PATH)) + + +def _rows(rows: Iterable[sqlite3.Row]) -> list[dict[str, Any]]: + return [dict(r) for r in rows] + + +def _require_schema(conn: sqlite3.Connection) -> str | None: + for t in ("olfactory_state", "olfactory_imprints"): + if not conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (t,) + ).fetchone(): + return f"olfactory schema missing: {t}. Run `brainctl migrate` (082)." + return None + + +def _hash_content(content: str) -> str: + return hashlib.blake2b(content.encode("utf-8"), digest_size=16).hexdigest() + + +def tool_olfactory_status(**_kw: Any) -> dict[str, Any]: + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + state = conn.execute("SELECT * FROM olfactory_state WHERE id = 1").fetchone() + last_5 = _rows(conn.execute( + "SELECT * FROM olfactory_imprints ORDER BY id DESC LIMIT 5" + ).fetchall()) + valence_dist = _rows(conn.execute( + """ + SELECT + CASE WHEN valence >= 0.3 THEN 'positive' + WHEN valence <= -0.3 THEN 'negative' + ELSE 'neutral' END AS bucket, + COUNT(*) AS n + FROM olfactory_imprints GROUP BY bucket + """ + ).fetchall()) + return { + "ok": True, + "state": dict(state) if state else None, + "last_5_imprints": last_5, + "valence_distribution": valence_dist, + } + + +def tool_olfactory_imprint( + content: str, valence: float, + arousal: float = 0.5, + content_kind: str | None = None, + bound_memory_id: int | None = None, + bound_entity_id: int | None = None, + agent_id: str | None = None, + notes: str | None = None, + **_kw: Any, +) -> dict[str, Any]: + """Create or update an olfactory imprint. UPSERT keyed by + (content_hash, agent_id) — repeated imprints with the same content + + agent update the valence/arousal/pointers.""" + if not -1.0 <= valence <= 1.0: + return {"error": "valence must be in [-1, 1]"} + if not 0.0 <= arousal <= 1.0: + return {"error": "arousal must be in [0, 1]"} + if not content: + return {"error": "content must be non-empty"} + content_hash = _hash_content(content) + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + existing = conn.execute( + "SELECT id FROM olfactory_imprints WHERE content_hash = ? AND agent_id IS ?", + (content_hash, agent_id), + ).fetchone() + if existing: + conn.execute( + """ + UPDATE olfactory_imprints SET + valence = ?, arousal = ?, content_kind = ?, + bound_memory_id = COALESCE(?, bound_memory_id), + bound_entity_id = COALESCE(?, bound_entity_id), + notes = COALESCE(?, notes) + WHERE id = ? + """, + (float(valence), float(arousal), content_kind, + bound_memory_id, bound_entity_id, notes, existing["id"]), + ) + imprint_id = int(existing["id"]) + preexisting = True + else: + cur = conn.execute( + """ + INSERT INTO olfactory_imprints + (content_hash, content_kind, valence, arousal, + bound_memory_id, bound_entity_id, agent_id, notes) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (content_hash, content_kind, float(valence), float(arousal), + bound_memory_id, bound_entity_id, agent_id, notes), + ) + imprint_id = cur.lastrowid + preexisting = False + conn.execute( + "UPDATE olfactory_state SET total_imprints = total_imprints + 1, updated_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') WHERE id = 1" + ) + conn.commit() + return { + "ok": True, "imprint_id": imprint_id, + "content_hash": content_hash, + "preexisting": preexisting, + "valence": float(valence), "arousal": float(arousal), + } + + +def tool_olfactory_recall(content: str, agent_id: str | None = None, **_kw: Any) -> dict[str, Any]: + """Look up an olfactory imprint by content. If found, increments + times_recalled + returns the bound memory_id/entity_id + valence. + Equivalent to the Proust-effect lookup.""" + if not content: + return {"error": "content must be non-empty"} + content_hash = _hash_content(content) + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + row = conn.execute( + "SELECT * FROM olfactory_imprints WHERE content_hash = ? AND agent_id IS ?", + (content_hash, agent_id), + ).fetchone() + if not row: + return {"ok": True, "matched": False, "imprint": None} + conn.execute( + """ + UPDATE olfactory_imprints SET + times_recalled = times_recalled + 1, + last_recalled_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') + WHERE id = ? + """, + (row["id"],), + ) + conn.commit() + # Re-read to get updated times_recalled + row = conn.execute("SELECT * FROM olfactory_imprints WHERE id = ?", (row["id"],)).fetchone() + return {"ok": True, "matched": True, "imprint": dict(row)} + + +def tool_olfactory_set(enforcement_mode: str, **_kw: Any) -> dict[str, Any]: + if enforcement_mode not in VALID_ENFORCEMENT_MODES: + return {"error": f"invalid enforcement_mode; expected {sorted(VALID_ENFORCEMENT_MODES)}"} + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + conn.execute( + "UPDATE olfactory_state SET enforcement_mode = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') WHERE id = 1", + (enforcement_mode,), + ) + conn.commit() + state = conn.execute("SELECT * FROM olfactory_state WHERE id = 1").fetchone() + return {"ok": True, "state": dict(state) if state else None} + + +TOOLS: list[Tool] = [ + Tool( + name="olfactory_status", + description="Olfactory state + last 5 imprints + valence distribution (positive/neutral/negative).", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="olfactory_imprint", + description=( + "Create or update an olfactory imprint — a direct (content_hash, valence, " + "memory/entity pointer) binding that bypasses the standard W(m) gate. " + "Idempotent: UPSERT keyed by (content_hash, agent_id). valence in [-1, 1], " + "arousal in [0, 1]." + ), + inputSchema={ + "type": "object", + "properties": { + "content": {"type": "string"}, + "valence": {"type": "number"}, + "arousal": {"type": "number", "default": 0.5}, + "content_kind": {"type": "string"}, + "bound_memory_id": {"type": "integer"}, + "bound_entity_id": {"type": "integer"}, + "agent_id": {"type": "string"}, + "notes": {"type": "string"}, + }, + "required": ["content", "valence"], + }, + ), + Tool( + name="olfactory_recall", + description=( + "Look up an olfactory imprint by content (Proust-style fast emotional recall). " + "Returns matched=True + the imprint if found and increments times_recalled. " + "Returns matched=False if no imprint exists." + ), + inputSchema={ + "type": "object", + "properties": { + "content": {"type": "string"}, + "agent_id": {"type": "string"}, + }, + "required": ["content"], + }, + ), + Tool( + name="olfactory_set", + description="Set enforcement_mode ∈ {shadow, enforce, disabled}. Default shadow.", + inputSchema={ + "type": "object", + "properties": { + "enforcement_mode": {"type": "string", "enum": sorted(VALID_ENFORCEMENT_MODES)}, + }, + "required": ["enforcement_mode"], + }, + ), +] + + +_OLF_TOOLS = { + "olfactory_status": tool_olfactory_status, + "olfactory_imprint": tool_olfactory_imprint, + "olfactory_recall": tool_olfactory_recall, + "olfactory_set": tool_olfactory_set, +} + +DISPATCH: dict[str, Any] = { + name: (lambda _func=func, **kw: _func(**kw)) + for name, func in _OLF_TOOLS.items() +} + + +def register_tools() -> tuple[list[Tool], dict[str, Any]]: + return TOOLS, DISPATCH diff --git a/tests/test_mcp_tools_olfactory.py b/tests/test_mcp_tools_olfactory.py new file mode 100644 index 0000000..f0d8fa3 --- /dev/null +++ b/tests/test_mcp_tools_olfactory.py @@ -0,0 +1,112 @@ +"""Tests for mcp_tools_olfactory — Phase 1.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +MIGRATION_082 = REPO_ROOT / "db" / "migrations" / "082_olfactory.sql" + + +def _bootstrap(conn): + conn.executescript( + "CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY, description TEXT, applied_at TEXT);" + ) + + +def _apply(db_path): + conn = sqlite3.connect(str(db_path)) + try: + _bootstrap(conn) + conn.executescript(MIGRATION_082.read_text()) + conn.commit() + finally: + conn.close() + + +def _make_db(tmp_path, monkeypatch): + db = tmp_path / "brain.db" + _apply(db) + from agentmemory import mcp_tools_olfactory as mod + monkeypatch.setattr(mod, "DB_PATH", db) + return mod + + +def test_migration_with_defaults(tmp_path): + db = tmp_path / "brain.db" + _apply(db) + conn = sqlite3.connect(str(db)) + try: + state = conn.execute( + "SELECT total_imprints, enforcement_mode FROM olfactory_state" + ).fetchone() + assert state == (0, "shadow") + finally: + conn.close() + + +def test_imprint_creates(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + out = mod.tool_olfactory_imprint( + content="madeleine in tea", valence=0.8, arousal=0.7, + content_kind="phrase", bound_memory_id=42, + ) + assert out["ok"] is True + assert out["preexisting"] is False + + +def test_imprint_idempotent(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + first = mod.tool_olfactory_imprint(content="x", valence=0.5) + second = mod.tool_olfactory_imprint(content="x", valence=-0.5) + assert first["ok"] and second["ok"] + assert first["imprint_id"] == second["imprint_id"] + assert second["preexisting"] is True + # Recall should show updated valence + recall = mod.tool_olfactory_recall(content="x") + assert recall["imprint"]["valence"] == -0.5 + + +def test_imprint_validates(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + assert "error" in mod.tool_olfactory_imprint(content="x", valence=2.0) + assert "error" in mod.tool_olfactory_imprint(content="x", valence=0.5, arousal=1.5) + assert "error" in mod.tool_olfactory_imprint(content="", valence=0.5) + + +def test_recall_increments_count(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + mod.tool_olfactory_imprint(content="lemon", valence=0.3) + out1 = mod.tool_olfactory_recall(content="lemon") + assert out1["matched"] is True + assert out1["imprint"]["times_recalled"] == 1 + out2 = mod.tool_olfactory_recall(content="lemon") + assert out2["imprint"]["times_recalled"] == 2 + + +def test_recall_unmatched(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + out = mod.tool_olfactory_recall(content="never-seen") + assert out["ok"] is True + assert out["matched"] is False + assert out["imprint"] is None + + +def test_status_valence_distribution(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + mod.tool_olfactory_imprint(content="a", valence=0.8) + mod.tool_olfactory_imprint(content="b", valence=-0.7) + mod.tool_olfactory_imprint(content="c", valence=0.0) + out = mod.tool_olfactory_status() + dist = {row["bucket"]: row["n"] for row in out["valence_distribution"]} + assert dist.get("positive") == 1 + assert dist.get("negative") == 1 + assert dist.get("neutral") == 1 + + +def test_set_enforcement(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + out = mod.tool_olfactory_set(enforcement_mode="enforce") + assert out["ok"] is True + assert out["state"]["enforcement_mode"] == "enforce" + assert "error" in mod.tool_olfactory_set(enforcement_mode="bogus")