Knowledge layer v2: notes, update-in-place, stats, tags

CLAUDE.md

@@ -25,11 +25,11 @@ CI runs all three (ruff, mypy, pytest) on push/PR to main via `.github/workflows
 
 ```
 src/mcp_awareness/
-├── schema.py      # Entry types (status/alert/pattern/suppression/context/preference),
+├── schema.py      # Entry types (status/alert/pattern/suppression/context/preference/note),
 │                  # common envelope, validation, TTL/expiry logic, severity ranking
 ├── store.py       # Store protocol + SQLiteStore implementation (WAL mode), CRUD, soft delete, TTL cleanup
 ├── collator.py    # Briefing generation: applies suppressions + patterns, composes summary/mention
-└── server.py      # FastMCP server wiring — resources (read) + tools (write) + secret path middleware
+└── server.py      # FastMCP server wiring — resources (read) + tools (write/update) + secret path middleware
 ```
 
 **Data flow**: Edge processes → tools (`report_status`, `report_alert`) → `store` → `collator.generate_briefing()` → `awareness://briefing` resource
@@ -38,7 +38,10 @@ src/mcp_awareness/
 
 **Key design decisions**:
 - Briefing is computed on-demand per read (not background task) — fine for SQLite with WAL
+- Seven entry types: status, alert, pattern, suppression, context, preference, note
 - One status entry per source (upsert), alerts keyed by source + alert_id, preferences upsert by key + scope
+- Notes support optional content payload with MIME content_type
+- update_entry works on knowledge types only (note/pattern/context/preference); status/alert/suppression are immutable. Changes tracked in _changelog array
 - Suppressions use expiry timestamps + escalation override (critical breaks through warning-level suppression)
 - Pattern matching uses word-overlap between effect string and alert fields (hyphens/dashes normalized); hour ranges handle overnight wraparound
 - Soft delete: `delete_entry` moves to trash (30-day retention), `restore_entry` recovers, `get_deleted` lists trash. Bulk deletes require `confirm=True` (dry-run by default). Auto-purged by existing `_cleanup_expired`.

README.md

@@ -169,7 +169,7 @@ docker compose --profile quick up -d mcp-awareness tunnel-quick
 
 ## Tools
 
-The server exposes 14 MCP tools. Clients that support MCP resources also get 6 read-only resources, but since many clients (including Claude.ai) only surface tools, every resource has a tool mirror.
+The server exposes 18 MCP tools. Clients that support MCP resources also get 6 read-only resources, but since many clients (including Claude.ai) only surface tools, every resource has a tool mirror.
 
 ### Read tools
 
@@ -178,16 +178,19 @@ The server exposes 14 MCP tools. Clients that support MCP resources also get 6 r
 | `get_briefing` | Compact awareness summary (~200 tokens all-clear, ~500 with issues). Call at conversation start. Pre-filtered through patterns and suppressions. |
 | `get_alerts` | Active alerts, optionally filtered by source. Drill-down from briefing. |
 | `get_status` | Full status for a specific source including metrics and inventory. |
-| `get_knowledge` | All knowledge entries: learned patterns, historical context, preferences. |
+| `get_knowledge` | Knowledge entries (patterns, context, preferences, notes). Filter by source, tags, entry_type. `include_history` controls changelog visibility. |
 | `get_suppressions` | Active alert suppressions with expiry times and escalation settings. |
+| `get_stats` | Entry counts by type, list of sources, total count. Call before `get_knowledge` to decide whether to filter. |
+| `get_tags` | All tags in use with usage counts. Use to discover existing tags and prevent drift. |
 
 ### Write tools
 
 | Tool | Description |
 |------|-------------|
 | `report_status` | Report system status. Called periodically by edge processes. Upserts one entry per source; stale if TTL expires without refresh. |
 | `report_alert` | Report or resolve an alert. Captures diagnostics at detection time. Levels: `warning`, `critical`. Types: `threshold`, `structural`, `baseline`. |
-| `learn_pattern` | Record permanent knowledge from conversation. Tagged and searchable. Any agent writes; any agent reads. Set `learned_from` to your platform. |
+| `learn_pattern` | Record an operational pattern with conditions/effects for alert matching. Set `learned_from` to your platform. |
+| `remember` | Store a general-purpose note. Optional `content` payload with MIME `content_type`. For anything that isn't an operational pattern or time-limited context. |
 | `add_context` | Record time-limited knowledge (default 30 days). Use for events, temporary situations, or facts that lose relevance. |
 | `set_preference` | Set a portable presentation preference (e.g., `alert_verbosity`, `check_frequency`). Upserts by key + scope. |
 | `suppress_alert` | Suppress alerts by source/tags/metric. Time-limited with escalation override — critical alerts can break through. |
@@ -196,6 +199,7 @@ The server exposes 14 MCP tools. Clients that support MCP resources also get 6 r
 
 | Tool | Description |
 |------|-------------|
+| `update_entry` | Update a knowledge entry in place (note, pattern, context, preference). Tracks changes in `_changelog`. Status/alert/suppression are immutable. |
 | `delete_entry` | Soft-delete entries (30-day trash). By ID, by source + type, or by source. Bulk deletes require `confirm=True` (dry-run by default). |
 | `restore_entry` | Restore a soft-deleted entry from trash. |
 | `get_deleted` | List all entries in trash with IDs for restore. |
@@ -219,20 +223,22 @@ See [Security considerations](docs/deployment-guide.md#security-considerations)
 - Portable knowledge store: agents read/write tagged knowledge via MCP tools
 - Ambient awareness: status reporting, alert detection, suppression, briefing generation
 - Storage abstraction: `Store` protocol with `SQLiteStore` default — designed for future Postgres/vector backends
-- Full MCP API: 6 resources + 14 tools (read mirrors for tools-only clients like Claude.ai)
+- Full MCP API: 6 resources + 18 tools (read mirrors for tools-only clients like Claude.ai)
+- General-purpose notes with optional content payload and MIME type
+- In-place updates with changelog tracking for knowledge entries
+- Tag registry and stats for store introspection
 - Soft delete with 30-day trash, dry-run confirmation for bulk operations
 - Streamable HTTP + stdio transports
 - Secret path auth + Cloudflare WAF for edge-level access control
 - Docker Compose with named Cloudflare Tunnel or ephemeral quick tunnel
 - Three-layer detection model (threshold + knowledge implemented; baseline planned)
 - Suppression system with time-based expiry and escalation overrides
-- 124 tests, strict type checking, CI pipeline
+- 148 tests, strict type checking, CI pipeline
 
 **Not yet implemented:**
 - Layer 2 (baseline) detection — rolling averages and deviation calculation
 - Edge processes — no automated producers yet ([example script](examples/simulate_edge.py) demonstrates the write path)
 - Semantic search — current knowledge retrieval is tag/keyword-based; vector similarity is planned
-- First-class knowledge entry type — currently uses `learn_pattern`; a dedicated knowledge tool with `remember`/`recall` semantics is planned
 - OAuth / API key authentication — current auth is secret-path-based; proper token auth requires MCP client support for auth flows
 
 ## Design docs

src/mcp_awareness/schema.py

@@ -16,6 +16,7 @@ class EntryType(str, Enum):
     SUPPRESSION = "suppression"
     CONTEXT = "context"
     PREFERENCE = "preference"
+    NOTE = "note"
 
 
 SEVERITY_RANK = {

src/mcp_awareness/server.py

@@ -157,20 +157,23 @@ async def get_knowledge(
     source: str | None = None,
     tags: list[str] | None = None,
     entry_type: str | None = None,
+    include_history: str | None = None,
 ) -> str:
-    """Get knowledge entries: learned patterns, historical context, preferences.
+    """Get knowledge entries: learned patterns, historical context, preferences, notes.
     Knowledge belongs to the system, not any specific agent. Call when you need
-    context about a system's normal behavior or operational patterns.
+    context about a system's normal behavior, operational patterns, or stored notes.
     Filter by source, tags, and/or entry_type to reduce response size.
-    Valid entry_type values: 'pattern', 'context', 'preference'.
+    Valid entry_type values: 'pattern', 'context', 'preference', 'note'.
+    include_history: omit or 'false' to strip change history, 'true' to include,
+    'only' to return only entries with change history.
     This tool always returns JSON with a status field or an entry list.
     If you receive an unstructured error, the failure is in the transport
     or platform layer, not in awareness."""
     if entry_type:
         et = EntryType(entry_type)
         entries = store.get_entries(entry_type=et, source=source, tags=tags)
     else:
-        entries = store.get_knowledge(tags=tags)
+        entries = store.get_knowledge(tags=tags, include_history=include_history)
         if source:
             entries = [e for e in entries if e.source == source]
     return json.dumps([e.to_dict() for e in entries], indent=2)
@@ -276,6 +279,104 @@ async def learn_pattern(
     return json.dumps({"status": "ok", "id": entry.id, "description": description})
 
 
+@mcp.tool()
+async def remember(
+    source: str,
+    tags: list[str],
+    description: str,
+    content: str | None = None,
+    content_type: str = "text/plain",
+    learned_from: str = "conversation",
+) -> str:
+    """Store a general-purpose note. Use this for any knowledge that doesn't fit
+    operational patterns (learn_pattern) or time-limited context (add_context).
+    Examples: personal facts, project notes, skill backups, config snapshots.
+    description is a short summary; content is the optional payload (text, JSON, etc.).
+    content_type is a MIME type (default text/plain). Set learned_from to your platform.
+    Returns JSON with status and entry id. If you receive an unstructured
+    error, the failure is in the transport or platform layer, not in awareness."""
+    now = now_iso()
+    data: dict[str, Any] = {
+        "description": description,
+        "learned_from": learned_from,
+    }
+    if content is not None:
+        data["content"] = content
+        data["content_type"] = content_type
+    entry = Entry(
+        id=make_id(),
+        type=EntryType.NOTE,
+        source=source,
+        tags=tags,
+        created=now,
+        updated=now,
+        expires=None,
+        data=data,
+    )
+    store.add(entry)
+    return json.dumps({"status": "ok", "id": entry.id, "description": description})
+
+
+@mcp.tool()
+async def update_entry(
+    entry_id: str,
+    description: str | None = None,
+    tags: list[str] | None = None,
+    source: str | None = None,
+    content: str | None = None,
+    content_type: str | None = None,
+) -> str:
+    """Update an existing entry in place, preserving its ID and creation timestamp.
+    Only works on knowledge types: note, pattern, context, preference.
+    Status, alert, and suppression entries are immutable.
+    Only provided fields are updated — omit fields to leave them unchanged.
+    Changes are tracked in a _changelog array within the entry data.
+    Use get_knowledge(include_history='true') to see change history.
+    Returns JSON with status. If you receive an unstructured error, the failure
+    is in the transport or platform layer, not in awareness."""
+    updates: dict[str, Any] = {}
+    if description is not None:
+        updates["description"] = description
+    if tags is not None:
+        updates["tags"] = tags
+    if source is not None:
+        updates["source"] = source
+    if content is not None:
+        updates["content"] = content
+    if content_type is not None:
+        updates["content_type"] = content_type
+    if not updates:
+        return json.dumps({"status": "error", "message": "No fields to update"})
+    result = store.update_entry(entry_id, updates)
+    if result is None:
+        return json.dumps(
+            {
+                "status": "error",
+                "message": "Entry not found or type is immutable (status/alert/suppression)",
+            }
+        )
+    return json.dumps({"status": "ok", "id": result.id, "updated": result.updated})
+
+
+@mcp.tool()
+async def get_stats() -> str:
+    """Get summary statistics: entry counts by type, list of sources, total count.
+    Call before get_knowledge to decide whether to pull everything or filter.
+    This tool always returns structured JSON. If you receive an unstructured
+    error, the failure is in the transport or platform layer, not in awareness."""
+    return json.dumps(store.get_stats(), indent=2)
+
+
+@mcp.tool()
+async def get_tags() -> str:
+    """Get all tags in use with usage counts, sorted by count descending.
+    Use this to discover existing tags before creating new ones — prevents
+    tag drift (e.g., 'infrastructure' vs 'infra'). This tool always returns
+    structured JSON. If you receive an unstructured error, the failure is in
+    the transport or platform layer, not in awareness."""
+    return json.dumps(store.get_tags(), indent=2)
+
+
 @mcp.tool()
 async def suppress_alert(
     source: str | None = None,

src/mcp_awareness/store.py

@@ -54,7 +54,17 @@ def get_patterns(self, source: str | None = None) -> list[Entry]: ...
 
     def count_active_suppressions(self) -> int: ...
 
-    def get_knowledge(self, tags: list[str] | None = None) -> list[Entry]: ...
+    def get_knowledge(
+        self, tags: list[str] | None = None, include_history: str | None = None
+    ) -> list[Entry]: ...
+
+    def get_entry_by_id(self, entry_id: str) -> Entry | None: ...
+
+    def update_entry(self, entry_id: str, updates: dict[str, Any]) -> Entry | None: ...
+
+    def get_stats(self) -> dict[str, Any]: ...
+
+    def get_tags(self) -> list[dict[str, Any]]: ...
 
     def soft_delete_by_id(self, entry_id: str) -> bool: ...
 
@@ -350,15 +360,107 @@ def count_active_suppressions(self) -> int:
         result: int = cur.fetchone()[0]
         return result
 
-    def get_knowledge(self, tags: list[str] | None = None) -> list[Entry]:
-        """Get knowledge entries (patterns, context, preferences)."""
-        types = (EntryType.PATTERN.value, EntryType.CONTEXT.value, EntryType.PREFERENCE.value)
+    def get_knowledge(
+        self, tags: list[str] | None = None, include_history: str | None = None
+    ) -> list[Entry]:
+        """Get knowledge entries (patterns, context, preferences, notes).
+
+        include_history: None/false = strip _changelog from results,
+                         "true" = include _changelog, "only" = only entries with changelog.
+        """
+        types = (
+            EntryType.PATTERN.value,
+            EntryType.CONTEXT.value,
+            EntryType.PREFERENCE.value,
+            EntryType.NOTE.value,
+        )
         placeholders = ",".join("?" * len(types))
         entries = self._query_entries(f"type IN ({placeholders})", types)
         if tags:
             entries = [e for e in entries if any(t in e.tags for t in tags)]
+        if include_history == "only":
+            entries = [e for e in entries if e.data.get("_changelog")]
+        elif include_history != "true":
+            # Strip _changelog from results by default
+            for e in entries:
+                e.data.pop("_changelog", None)
         return entries
 
+    def get_entry_by_id(self, entry_id: str) -> Entry | None:
+        """Get a single entry by ID (active only)."""
+        results = self._query_entries("id = ?", (entry_id,))
+        return results[0] if results else None
+
+    def update_entry(self, entry_id: str, updates: dict[str, Any]) -> Entry | None:
+        """Update an entry in place, appending previous values to _changelog.
+
+        Only works on knowledge types (note, pattern, context, preference).
+        Returns the updated entry, or None if not found or type is immutable.
+        """
+        entry = self.get_entry_by_id(entry_id)
+        if entry is None:
+            return None
+        immutable = {EntryType.STATUS, EntryType.ALERT, EntryType.SUPPRESSION}
+        if entry.type in immutable:
+            return None
+
+        with self._write_lock:
+            self._cleanup_expired()
+            now = now_iso()
+            # Build changelog record of changed fields
+            changed: dict[str, Any] = {}
+            # Envelope fields
+            for field in ("source", "tags"):
+                if field in updates and updates[field] != getattr(entry, field):
+                    changed[field] = getattr(entry, field)
+                    setattr(entry, field, updates[field])
+            # Data fields
+            for field in ("description", "content", "content_type"):
+                if field in updates and updates[field] != entry.data.get(field):
+                    old_val = entry.data.get(field)
+                    if old_val is not None:
+                        changed[field] = old_val
+                    entry.data[field] = updates[field]
+
+            if not changed:
+                return entry  # nothing actually changed
+
+            # Append to changelog
+            changelog = entry.data.setdefault("_changelog", [])
+            changelog.append({"updated": now, "changed": changed})
+            entry.updated = now
+
+            self._conn.execute(
+                "UPDATE entries SET updated = ?, source = ?, tags = ?, data = ? WHERE id = ?",
+                (now, entry.source, json.dumps(entry.tags), json.dumps(entry.data), entry.id),
+            )
+            self._conn.commit()
+            return entry
+
+    def get_stats(self) -> dict[str, Any]:
+        """Get entry counts by type, list of sources, and total count."""
+        cur = self._conn.execute(
+            f"SELECT type, COUNT(*) FROM entries WHERE {self._ACTIVE} GROUP BY type"
+        )
+        counts = {row[0]: row[1] for row in cur.fetchall()}
+        cur2 = self._conn.execute(
+            f"SELECT DISTINCT source FROM entries WHERE {self._ACTIVE} ORDER BY source"
+        )
+        sources = [row[0] for row in cur2.fetchall()]
+        return {
+            "entries": {t.value: counts.get(t.value, 0) for t in EntryType},
+            "sources": sources,
+            "total": sum(counts.values()),
+        }
+
+    def get_tags(self) -> list[dict[str, Any]]:
+        """Get all tags in use with usage counts."""
+        cur = self._conn.execute(
+            f"SELECT value, COUNT(*) as cnt FROM entries, json_each(entries.tags) "
+            f"WHERE {self._ACTIVE} GROUP BY value ORDER BY cnt DESC"
+        )
+        return [{"tag": row[0], "count": row[1]} for row in cur.fetchall()]
+
     # ------------------------------------------------------------------
     # Soft delete / trash
     # ------------------------------------------------------------------