diff --git a/db/migrations/080_colliculi.sql b/db/migrations/080_colliculi.sql new file mode 100644 index 0000000..0762771 --- /dev/null +++ b/db/migrations/080_colliculi.sql @@ -0,0 +1,77 @@ +-- Migration 080: superior + inferior colliculi — Phase 1 schema +-- +-- Avenue 9 from research/autonomous-research-avenues-2026-05-20.md. +-- Subcortical orienting reflex: superior colliculus (SC) = visual / +-- attention orienting; inferior colliculus (IC) = auditory orienting. +-- They fire BEFORE cortical processing and bias attention rapidly. +-- +-- brainctl analog: pre-cortical orienting on novel-pattern signals +-- (new entity sightings, unfamiliar query shapes, unusual content +-- types). Fires a fast ARAS drive pulse + thalamic mode adjustment +-- before the full retrieval pipeline gets going. +-- +-- Phase 1 ships: +-- colliculi_orienting_events — log of pre-cortical orient events +-- colliculi_state — single row tracking SC/IC tonic activity +-- colliculi_trigger_patterns — pattern catalog (which novel shapes +-- fire which sub-nucleus) +-- +-- Phase 2 wires into MCP dispatch as a sub-millisecond early-fire +-- before BG/cerebellum consults. Phase 3 modulates ARAS + thalamus +-- in response. +-- +-- Rollback: +-- DROP TABLE IF EXISTS colliculi_trigger_patterns; +-- DROP TABLE IF EXISTS colliculi_orienting_events; +-- DROP TABLE IF EXISTS colliculi_state; +-- DELETE FROM schema_version WHERE version = 80; +-- +-- IDEMPOTENT. + +CREATE TABLE IF NOT EXISTS colliculi_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + sc_tonic REAL NOT NULL DEFAULT 0.3 CHECK(sc_tonic BETWEEN 0.0 AND 1.0), + ic_tonic REAL NOT NULL DEFAULT 0.3 CHECK(ic_tonic BETWEEN 0.0 AND 1.0), + total_orienting_events INTEGER NOT NULL DEFAULT 0, + last_orient_at TEXT, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); +INSERT OR IGNORE INTO colliculi_state (id) VALUES (1); + +CREATE TABLE IF NOT EXISTS colliculi_trigger_patterns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + sub_nucleus TEXT NOT NULL CHECK(sub_nucleus IN ('sc', 'ic')), + pattern_kind TEXT NOT NULL CHECK(pattern_kind IN ( + 'novel_entity_shape', 'unfamiliar_query_form', 'unusual_content_type', + 'sudden_volume_change', 'cross_modal_mismatch', 'other' + )), + default_strength REAL NOT NULL DEFAULT 0.4 CHECK(default_strength BETWEEN 0.0 AND 1.0), + description TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); + +INSERT OR IGNORE INTO colliculi_trigger_patterns (name, sub_nucleus, pattern_kind, default_strength, description) VALUES + ('new_entity_seen', 'sc', 'novel_entity_shape', 0.5, 'previously-unseen entity name pattern'), + ('unusual_query_structure', 'sc', 'unfamiliar_query_form', 0.4, 'query token sequence doesn''t match recent distribution'), + ('content_type_shift', 'sc', 'unusual_content_type', 0.3, 'incoming content uses category not seen in last 7d'), + ('audio_burst', 'ic', 'sudden_volume_change', 0.6, 'audio input event with sharp amplitude'), + ('cross_modal_disagree', 'ic', 'cross_modal_mismatch', 0.5, 'auditory + visual signals disagree about same target'); + +CREATE TABLE IF NOT EXISTS colliculi_orienting_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + oriented_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + agent_id TEXT, + sub_nucleus TEXT NOT NULL CHECK(sub_nucleus IN ('sc', 'ic')), + pattern_id INTEGER REFERENCES colliculi_trigger_patterns(id) ON DELETE SET NULL, + strength REAL NOT NULL CHECK(strength BETWEEN 0.0 AND 1.0), + target_description TEXT, + aras_drive_fired INTEGER NOT NULL DEFAULT 0, -- 1 if downstream ARAS was nudged + notes TEXT +); +CREATE INDEX IF NOT EXISTS idx_coe_recent ON colliculi_orienting_events(oriented_at); +CREATE INDEX IF NOT EXISTS idx_coe_subnucleus ON colliculi_orienting_events(sub_nucleus, oriented_at); + +INSERT OR IGNORE INTO schema_version (version, description, applied_at) +VALUES (80, 'colliculi Phase 1: SC/IC orienting reflex (3 tables, 5 seeded patterns)', + strftime('%Y-%m-%dT%H:%M:%S', 'now')); diff --git a/src/agentmemory/mcp_server.py b/src/agentmemory/mcp_server.py index 95e5f45..6afbd7e 100755 --- a/src/agentmemory/mcp_server.py +++ b/src/agentmemory/mcp_server.py @@ -48,6 +48,7 @@ mcp_tools_belief_merge, mcp_tools_beliefs, mcp_tools_cerebellum, + mcp_tools_colliculi, mcp_tools_consolidation, mcp_tools_dmn, mcp_tools_drives, @@ -91,6 +92,7 @@ mcp_tools_belief_merge, mcp_tools_beliefs, mcp_tools_cerebellum, + mcp_tools_colliculi, mcp_tools_consolidation, mcp_tools_dmn, mcp_tools_drives, diff --git a/src/agentmemory/mcp_tools_colliculi.py b/src/agentmemory/mcp_tools_colliculi.py new file mode 100644 index 0000000..8176871 --- /dev/null +++ b/src/agentmemory/mcp_tools_colliculi.py @@ -0,0 +1,273 @@ +"""brainctl MCP tools — superior + inferior colliculi (orienting reflex). + +Phase 1 per research-avenues memo Avenue 9. Pre-cortical orienting: +SC (visual / attention orienting) + IC (auditory / cross-modal). Phase +1 = manual orient events. Phase 2 wires into dispatch as early-fire. +""" +from __future__ import annotations + +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_SUB_NUCLEI = {"sc", "ic"} +VALID_PATTERN_KINDS = { + "novel_entity_shape", "unfamiliar_query_form", "unusual_content_type", + "sudden_volume_change", "cross_modal_mismatch", "other", +} + + +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 ("colliculi_state", "colliculi_trigger_patterns", "colliculi_orienting_events"): + if not conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (t,) + ).fetchone(): + return f"colliculi schema missing: {t}. Run `brainctl migrate` (080)." + return None + + +def tool_colliculi_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 colliculi_state WHERE id = 1").fetchone() + patterns = _rows(conn.execute( + "SELECT * FROM colliculi_trigger_patterns ORDER BY id" + ).fetchall()) + last_5 = _rows(conn.execute( + "SELECT * FROM colliculi_orienting_events ORDER BY id DESC LIMIT 5" + ).fetchall()) + agg = conn.execute( + """ + SELECT COUNT(*) AS n, + COALESCE(AVG(strength), 0.0) AS mean_strength, + SUM(CASE WHEN sub_nucleus='sc' THEN 1 ELSE 0 END) AS n_sc, + SUM(CASE WHEN sub_nucleus='ic' THEN 1 ELSE 0 END) AS n_ic, + SUM(aras_drive_fired) AS n_aras_drive_fired + FROM colliculi_orienting_events + WHERE oriented_at >= datetime('now', '-1 hour') + """ + ).fetchone() + return { + "ok": True, "state": dict(state) if state else None, + "patterns": patterns, "last_5_events": last_5, + "aggregate_1h": dict(agg) if agg else {}, + } + + +def tool_colliculi_orient( + sub_nucleus: str, + pattern_name: str | None = None, + strength: float | None = None, + target_description: str | None = None, + agent_id: str | None = None, + aras_drive_fired: bool = False, + notes: str | None = None, + **_kw: Any, +) -> dict[str, Any]: + """Record an orienting event. Provide pattern_name (uses its + default_strength) OR strength explicitly.""" + if sub_nucleus not in VALID_SUB_NUCLEI: + return {"error": f"invalid sub_nucleus {sub_nucleus!r}; expected sc or ic"} + if pattern_name is None and strength is None: + return {"error": "must pass pattern_name OR strength"} + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + pattern_id: int | None = None + if pattern_name is not None: + pat = conn.execute( + "SELECT id, default_strength, sub_nucleus FROM colliculi_trigger_patterns WHERE name = ?", + (pattern_name,), + ).fetchone() + if not pat: + return {"error": f"pattern {pattern_name!r} not registered"} + if pat["sub_nucleus"] != sub_nucleus: + return {"error": f"pattern {pattern_name!r} belongs to {pat['sub_nucleus']!r}, not {sub_nucleus!r}"} + pattern_id = int(pat["id"]) + if strength is None: + strength = float(pat["default_strength"]) + if not 0.0 <= strength <= 1.0: + return {"error": "strength must be in [0, 1]"} + cur = conn.execute( + """ + INSERT INTO colliculi_orienting_events + (agent_id, sub_nucleus, pattern_id, strength, target_description, aras_drive_fired, notes) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + (agent_id, sub_nucleus, pattern_id, float(strength), target_description, + 1 if aras_drive_fired else 0, notes), + ) + event_id = cur.lastrowid + conn.execute( + """ + UPDATE colliculi_state SET + total_orienting_events = total_orienting_events + 1, + last_orient_at = strftime('%Y-%m-%dT%H:%M:%S', 'now'), + updated_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') + WHERE id = 1 + """ + ) + conn.commit() + return { + "ok": True, "event_id": event_id, "sub_nucleus": sub_nucleus, + "strength": float(strength), "pattern_id": pattern_id, + } + + +def tool_colliculi_register_pattern( + name: str, sub_nucleus: str, pattern_kind: str, + default_strength: float = 0.4, description: str | None = None, + **_kw: Any, +) -> dict[str, Any]: + if sub_nucleus not in VALID_SUB_NUCLEI: + return {"error": f"invalid sub_nucleus; expected {sorted(VALID_SUB_NUCLEI)}"} + if pattern_kind not in VALID_PATTERN_KINDS: + return {"error": f"invalid pattern_kind; expected {sorted(VALID_PATTERN_KINDS)}"} + if not 0.0 <= default_strength <= 1.0: + return {"error": "default_strength must be in [0, 1]"} + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + conn.execute( + """ + INSERT INTO colliculi_trigger_patterns (name, sub_nucleus, pattern_kind, default_strength, description) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(name) DO UPDATE SET + default_strength = excluded.default_strength, + description = COALESCE(excluded.description, colliculi_trigger_patterns.description) + """, + (name, sub_nucleus, pattern_kind, float(default_strength), description), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM colliculi_trigger_patterns WHERE name = ?", (name,) + ).fetchone() + return {"ok": True, "pattern": dict(row) if row else None} + + +def tool_colliculi_history( + limit: int = 20, since: str | None = None, + sub_nucleus: str | None = None, min_strength: float | None = None, + **_kw: Any, +) -> dict[str, Any]: + limit = max(1, min(int(limit), 200)) + if sub_nucleus is not None and sub_nucleus not in VALID_SUB_NUCLEI: + return {"error": "invalid sub_nucleus"} + if min_strength is not None and not 0.0 <= min_strength <= 1.0: + return {"error": "min_strength must be in [0, 1]"} + with _db() as conn: + conn.row_factory = sqlite3.Row + err = _require_schema(conn) + if err: + return {"error": err} + clauses, params = [], [] + if since: + clauses.append("oriented_at >= ?"); params.append(since) + if sub_nucleus: + clauses.append("sub_nucleus = ?"); params.append(sub_nucleus) + if min_strength is not None: + clauses.append("strength >= ?"); params.append(float(min_strength)) + where = "WHERE " + " AND ".join(clauses) if clauses else "" + rows = conn.execute( + f"SELECT * FROM colliculi_orienting_events {where} ORDER BY id DESC LIMIT ?", + (*params, limit), + ).fetchall() + return {"ok": True, "history": _rows(rows)} + + +TOOLS: list[Tool] = [ + Tool( + name="colliculi_status", + description="SC + IC state + pattern catalog + last 5 events + 1h aggregate.", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="colliculi_orient", + description=( + "Record one orienting event. sub_nucleus ∈ {sc, ic}. Pass pattern_name " + "(uses its default_strength + validates sub_nucleus match) OR pass strength " + "explicitly. Set aras_drive_fired=true if a downstream ARAS pulse was fired." + ), + inputSchema={ + "type": "object", + "properties": { + "sub_nucleus": {"type": "string", "enum": sorted(VALID_SUB_NUCLEI)}, + "pattern_name": {"type": "string"}, + "strength": {"type": "number"}, + "target_description": {"type": "string"}, + "agent_id": {"type": "string"}, + "aras_drive_fired": {"type": "boolean", "default": False}, + "notes": {"type": "string"}, + }, + "required": ["sub_nucleus"], + }, + ), + Tool( + name="colliculi_register_pattern", + description="Idempotent UPSERT on colliculi_trigger_patterns. sub_nucleus + pattern_kind validated.", + inputSchema={ + "type": "object", + "properties": { + "name": {"type": "string"}, + "sub_nucleus": {"type": "string", "enum": sorted(VALID_SUB_NUCLEI)}, + "pattern_kind": {"type": "string", "enum": sorted(VALID_PATTERN_KINDS)}, + "default_strength": {"type": "number", "default": 0.4}, + "description": {"type": "string"}, + }, + "required": ["name", "sub_nucleus", "pattern_kind"], + }, + ), + Tool( + name="colliculi_history", + description="Paginated orienting-event history. Filters: since, sub_nucleus, min_strength. limit clamped to [1, 200].", + inputSchema={ + "type": "object", + "properties": { + "limit": {"type": "integer", "default": 20}, + "since": {"type": "string"}, + "sub_nucleus": {"type": "string", "enum": sorted(VALID_SUB_NUCLEI)}, + "min_strength": {"type": "number"}, + }, + }, + ), +] + + +_COLL_TOOLS = { + "colliculi_status": tool_colliculi_status, + "colliculi_orient": tool_colliculi_orient, + "colliculi_register_pattern": tool_colliculi_register_pattern, + "colliculi_history": tool_colliculi_history, +} + +DISPATCH: dict[str, Any] = { + name: (lambda _func=func, **kw: _func(**kw)) + for name, func in _COLL_TOOLS.items() +} + + +def register_tools() -> tuple[list[Tool], dict[str, Any]]: + return TOOLS, DISPATCH diff --git a/tests/test_mcp_tools_colliculi.py b/tests/test_mcp_tools_colliculi.py new file mode 100644 index 0000000..20f14d5 --- /dev/null +++ b/tests/test_mcp_tools_colliculi.py @@ -0,0 +1,119 @@ +"""Tests for mcp_tools_colliculi — Phase 1.""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +MIGRATION_080 = REPO_ROOT / "db" / "migrations" / "080_colliculi.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_080.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_colliculi as mod + monkeypatch.setattr(mod, "DB_PATH", db) + return mod + + +def test_migration_seeds_patterns(tmp_path): + db = tmp_path / "brain.db" + _apply(db) + conn = sqlite3.connect(str(db)) + try: + n = conn.execute("SELECT COUNT(*) FROM colliculi_trigger_patterns").fetchone()[0] + assert n == 5 + state = conn.execute("SELECT sc_tonic, ic_tonic FROM colliculi_state").fetchone() + assert state == (0.3, 0.3) + finally: + conn.close() + + +def test_orient_via_pattern(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + out = mod.tool_colliculi_orient( + sub_nucleus="sc", pattern_name="new_entity_seen", agent_id="a1", + target_description="entity 'Olive'", + ) + assert out["ok"] is True + # default_strength for new_entity_seen = 0.5 + assert out["strength"] == 0.5 + + +def test_orient_explicit_strength(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + out = mod.tool_colliculi_orient( + sub_nucleus="ic", strength=0.7, target_description="loud crash", + ) + assert out["ok"] is True + assert out["strength"] == 0.7 + assert out["pattern_id"] is None + + +def test_orient_validates(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + assert "error" in mod.tool_colliculi_orient(sub_nucleus="bogus") + assert "error" in mod.tool_colliculi_orient(sub_nucleus="sc") + assert "error" in mod.tool_colliculi_orient(sub_nucleus="sc", strength=1.5) + # Cross-nucleus mismatch + assert "error" in mod.tool_colliculi_orient(sub_nucleus="ic", pattern_name="new_entity_seen") + + +def test_register_pattern_idempotent(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + first = mod.tool_colliculi_register_pattern( + name="custom_audio", sub_nucleus="ic", pattern_kind="sudden_volume_change", + default_strength=0.4, + ) + second = mod.tool_colliculi_register_pattern( + name="custom_audio", sub_nucleus="ic", pattern_kind="sudden_volume_change", + default_strength=0.6, + ) + assert first["ok"] and second["ok"] + assert second["pattern"]["default_strength"] == 0.6 + + +def test_register_pattern_validates(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + assert "error" in mod.tool_colliculi_register_pattern(name="x", sub_nucleus="bogus", pattern_kind="other") + assert "error" in mod.tool_colliculi_register_pattern(name="x", sub_nucleus="sc", pattern_kind="bogus") + + +def test_status_aggregates(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + mod.tool_colliculi_orient(sub_nucleus="sc", strength=0.5) + mod.tool_colliculi_orient(sub_nucleus="ic", strength=0.4, aras_drive_fired=True) + out = mod.tool_colliculi_status() + assert out["aggregate_1h"]["n"] == 2 + assert out["aggregate_1h"]["n_sc"] == 1 + assert out["aggregate_1h"]["n_ic"] == 1 + assert out["aggregate_1h"]["n_aras_drive_fired"] == 1 + + +def test_history_filters(tmp_path, monkeypatch): + mod = _make_db(tmp_path, monkeypatch) + mod.tool_colliculi_orient(sub_nucleus="sc", strength=0.3) + mod.tool_colliculi_orient(sub_nucleus="sc", strength=0.7) + mod.tool_colliculi_orient(sub_nucleus="ic", strength=0.5) + all_h = mod.tool_colliculi_history(limit=10) + assert len(all_h["history"]) == 3 + sc_only = mod.tool_colliculi_history(limit=10, sub_nucleus="sc") + assert len(sc_only["history"]) == 2 + strong = mod.tool_colliculi_history(limit=10, min_strength=0.6) + assert len(strong["history"]) == 1