hebbian-mind-enterprise

Instructions

Implementation Reference

• src/hebbian_mind/server.py:1242-1288 (handler)

  The query_mind tool implementation in the MCP call_tool handler.

  elif name == "query_mind":
      nodes = arguments.get("nodes", [])
      if not isinstance(nodes, list):
          raise ValueError("nodes must be a list of strings")
      if len(nodes) > 100:
          raise ValueError("nodes list exceeds maximum of 100 items")
      for n in nodes:
          if not isinstance(n, str):
              raise ValueError("each node must be a string")
      limit = int(_validate_number(arguments.get("limit", 20), "limit", 1, 500))
      include_decayed = bool(arguments.get("include_decayed", False))
  
      if not nodes:
          return [
              types.TextContent(
                  type="text",
                  text=json.dumps(
                      {"success": False, "message": "No nodes specified"}, indent=2
                  ),
              )
          ]
  
      memories = db.query_by_nodes(nodes, limit, include_decayed=include_decayed)
  
      return [
          types.TextContent(
              type="text",
              text=json.dumps(
                  {
                      "success": True,
                      "queried_nodes": nodes,
                      "memories_found": len(memories),
                      "memories": [
                          {
                              "memory_id": m["memory_id"],
                              "summary": m["summary"],
                              "source": m["source"],
                              "activations": m.get("activations", ""),
                              "created_at": m["created_at"],
                          }
                          for m in memories
                      ],
                  },
                  indent=2,
              ),
          )
      ]

• src/hebbian_mind/server.py:1032-1050 (registration)

  The query_mind tool definition in the MCP server list_tools function.

  types.Tool(
      name="query_mind",
      description="Query memories by concept nodes. Returns memories that activated specified concepts. Decayed memories are hidden by default.",
      inputSchema={
          "type": "object",
          "properties": {
              "nodes": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "List of node names to query",
              },
              "limit": {"type": "number", "description": "Max results (default: 20)"},
              "include_decayed": {
                  "type": "boolean",
                  "description": "Include decayed memories below threshold (default: false)",
              },
          },
      },
  ),

• src/hebbian_mind/server.py:786-854 (helper)

  The database logic for query_mind (query_by_nodes method).

  def query_by_nodes(
      self,
      node_names: List[str],
      limit: int = 20,
      include_decayed: bool = False,
  ) -> List[Dict]:
      """Query memories that activated specific nodes.
  
      Thread-safe: runs under _lock.
  
      Args:
          node_names: List of node names to query
          limit: Maximum results (clamped to 1-500)
          include_decayed: If True, include memories below decay threshold
      """
      # Clamp limit to safe range (H1 fix)
      limit = max(1, min(500, limit))
  
      with self._lock:
          cursor = self.read_conn.cursor()
  
          node_ids = []
          for name in node_names:
              node = self.get_node_by_name(name)
              if node:
                  node_ids.append(node["id"])
  
          if not node_ids:
              return []
  
          # Parameterized IN clause: placeholders are '?,?,?' with values bound safely
          placeholders = ",".join("?" * len(node_ids))
  
          # Build decay filter (static SQL fragment, threshold bound via params)
          decay_filter = ""
          decay_params: tuple = ()
          if not include_decayed and Config.DECAY_ENABLED:
              decay_filter = (
                  " AND (m.effective_importance IS NULL OR m.effective_importance >= ?)"
              )
              decay_params = (Config.DECAY_THRESHOLD,)
  
          cursor.execute(
              f"""
              SELECT DISTINCT m.*,
                     GROUP_CONCAT(n.name || ':' || ma.activation_score) as activations
              FROM memories m
              JOIN memory_activations ma ON m.memory_id = ma.memory_id
              JOIN nodes n ON ma.node_id = n.id
              WHERE ma.node_id IN ({placeholders})
              {decay_filter}
              GROUP BY m.id
              ORDER BY m.created_at DESC
              LIMIT ?
          """,
              (*node_ids, *decay_params, limit),
          )
  
          results = [dict(row) for row in cursor.fetchall()]
  
      # Touch accessed memories (fire-and-forget, outside lock)
      if results and hasattr(self, "_decay_engine") and self._decay_engine:
          memory_ids = [r["memory_id"] for r in results]
          try:
              self._decay_engine.touch_memories(memory_ids)
          except Exception:
              pass  # Non-critical
  
      return results