diff --git a/secator/cli.py b/secator/cli.py
index b31e2ff40..99bb82866 100644
--- a/secator/cli.py
+++ b/secator/cli.py
@@ -1123,9 +1123,10 @@ def list_aliases(silent):
 @click.option('--driver', type=click.Choice(['local', 'mongodb', 'api', 'sqlite']), default=None, help='Query backend driver')  # noqa: E501
 @click.option('--dedupe/--no-dedupe', default=None, help='Deduplicate findings (defaults to config value)')
 @click.option('-l', '--limit', type=int, default=0, help='Limit number of results (0 = no limit)')
+@click.option('--group', is_flag=False, flag_value='', default=None, help='Group findings by field(s) (comma-separated) with auto-aggregation. Bare --group uses per-type defaults.')  # noqa: E501
 @click.option('--save', 'save', type=str, default=None, help='Save the query expression ARG under this name for later reuse (e.g. --save vuln_high)')  # noqa: E501
 @click.pass_context
-def query(ctx, arg, output, output_folder, time_delta, fmt, workspace, report_filter, driver, dedupe, limit, save):
+def query(ctx, arg, output, output_folder, time_delta, fmt, workspace, report_filter, driver, dedupe, limit, group, save):  # noqa: E501
 	"""Query"""
 
 	# 0. Save the expression under a name, then exit (reuse later with `secator q <name>`).
@@ -1146,12 +1147,12 @@ def query(ctx, arg, output, output_folder, time_delta, fmt, workspace, report_fi
 
 	# 1. Saved query name
 	if arg in CONFIG.queries:
-		run_report_show(report_filter, output, time_delta, CONFIG.queries[arg], fmt, workspace, driver, dedupe, limit, output_folder)  # noqa: E501
+		run_report_show(report_filter, output, time_delta, CONFIG.queries[arg], fmt, workspace, driver, dedupe, limit, output_folder, group)  # noqa: E501
 		return
 
 	# 2. Raw filter expression
 	if _looks_like_query_expr(arg):
-		run_report_show(report_filter, output, time_delta, arg, fmt, workspace, driver, dedupe, limit, output_folder)
+		run_report_show(report_filter, output, time_delta, arg, fmt, workspace, driver, dedupe, limit, output_folder, group)  # noqa: E501
 		return
 
 	# 3. Natural language -> AI chat
@@ -1364,7 +1365,7 @@ def _apply_format(results, fmt):
 	return new_results
 
 
-def run_report_show(report_query, output, time_delta, query, fmt, workspace, driver, dedupe, limit, output_folder=None):
+def run_report_show(report_query, output, time_delta, query, fmt, workspace, driver, dedupe, limit, output_folder=None, group=None):  # noqa: E501
 	"""Build and send a consolidated report. Shared by `report show` and `query`.
 
 	REPORT_QUERY: comma-separated runner paths (e.g. scans/5,tasks/3).
@@ -1477,6 +1478,30 @@ def run_report_show(report_query, output, time_delta, query, fmt, workspace, dri
 	dedupe_effective = CONFIG.runners.remove_duplicates if dedupe is None else dedupe
 	report = Report(runner, title=f'Consolidated report - {current}', exporters=exporters)
 	report.build(query=full_query, dedupe=dedupe_effective, limit=limit)
+
+	# Group findings by field(s) with auto-aggregation (processing-side, post-query).
+	# `group is None` => disabled; `group == ''` => per-type defaults; else explicit field(s).
+	grouped_types = []
+	if group is not None and not fmt:
+		from secator.query.utils import group_findings
+		user_group_by = [f.strip() for f in group.split(',') if f.strip()]
+		type_map = {cls.get_name(): cls for cls in FINDING_TYPES}
+		for type_name, items in report.data['results'].items():
+			cls = type_map.get(type_name)
+			if not items or cls is None:
+				continue
+			group_by = user_group_by or list(getattr(cls, '_group_by', ()) or ())
+			if not group_by:
+				continue
+			aggregate_field = getattr(cls, '_group_aggregate', None)
+			report.data['results'][type_name] = group_findings(items, group_by, aggregate_field)
+			grouped_types.append((type_name, ', '.join(group_by)))
+		if not grouped_types:
+			group_desc = f' "{group}"' if group else ''
+			console.print(Warning(message=f'--group{group_desc}: no groupable finding types in results'))
+	elif group is not None and fmt:
+		console.print(Warning(message='--group is ignored when --format is used'))
+
 	if fmt:
 		report.data['results'] = _apply_format(report.data['results'], fmt)
 	report.send()
@@ -1488,6 +1513,8 @@ def run_report_show(report_query, output, time_delta, query, fmt, workspace, dri
 		if searched:
 			info_msg += f' (searched: [bold cyan]{searched}[/])'
 	console.print(Info(message=info_msg))
+	for type_name, field_str in grouped_types:
+		console.print(Info(message=f'{type_name} grouped by {field_str}. To show complete results, remove the --group option.'))  # noqa: E501
 
 
 def run_ai_chat(ctx, prompt, workspace):
@@ -1514,10 +1541,11 @@ def run_ai_chat(ctx, prompt, workspace):
 @click.option('--driver', type=click.Choice(['local', 'mongodb', 'api', 'sqlite']), default=None, help='Query backend driver')  # noqa: E501
 @click.option('--dedupe/--no-dedupe', default=None, help='Deduplicate findings (defaults to config value)')
 @click.option('-l', '--limit', type=int, default=0, help='Limit number of results (0 = no limit)')
+@click.option('--group', is_flag=False, flag_value='', default=None, help='Group findings by field(s) (comma-separated) with auto-aggregation. Bare --group uses per-type defaults.')  # noqa: E501
 @click.pass_context
-def report_show(ctx, report_query, output, output_folder, time_delta, query, fmt, workspace, driver, dedupe, limit):
+def report_show(ctx, report_query, output, output_folder, time_delta, query, fmt, workspace, driver, dedupe, limit, group):  # noqa: E501
 	"""Show report results. REPORT_QUERY: comma-separated runner paths (e.g. scans/5,tasks/3)."""
-	run_report_show(report_query, output, time_delta, query, fmt, workspace, driver, dedupe, limit, output_folder)
+	run_report_show(report_query, output, time_delta, query, fmt, workspace, driver, dedupe, limit, output_folder, group)
 
 
 def _load_report_data(path):
diff --git a/secator/exporters/console.py b/secator/exporters/console.py
index 0a129f456..4dc7e4ed7 100644
--- a/secator/exporters/console.py
+++ b/secator/exporters/console.py
@@ -31,4 +31,9 @@ def send(self):
 								item = cls.load(item)
 							except (TypeError, Exception):
 								pass
-					console_stdout.print(item, highlight=False)
+					# Grouped findings carry a _group_count; show it next to the line.
+					count = getattr(item, '_group_count', 0)
+					if count:
+						console_stdout.print(item, f'[dim](count: {count})[/]', highlight=False)
+					else:
+						console_stdout.print(item, highlight=False)
diff --git a/secator/output_types/_base.py b/secator/output_types/_base.py
index c7e1febda..fd4816306 100644
--- a/secator/output_types/_base.py
+++ b/secator/output_types/_base.py
@@ -12,6 +12,10 @@
 class OutputType:
 	_table_fields = []
 	_sort_by = ()
+	# Fields to group by (tuple) and the field whose values to aggregate/collect
+	# when the CLI `--group` option is used without an explicit field.
+	_group_by = ()
+	_group_aggregate = None
 
 	@classmethod
 	def fields(cls):
diff --git a/secator/output_types/tag.py b/secator/output_types/tag.py
index 8f5f177eb..8ac577931 100644
--- a/secator/output_types/tag.py
+++ b/secator/output_types/tag.py
@@ -27,6 +27,8 @@ class Tag(OutputType):
 
 	_table_fields = ['match', 'category', 'name', 'extra_data']
 	_sort_by = ('match', 'name')
+	_group_by = ('category', 'name')
+	_group_aggregate = 'match'
 
 	def __post_init__(self):
 		super().__post_init__()
diff --git a/secator/output_types/technology.py b/secator/output_types/technology.py
index 52dcceac4..896b0bf67 100644
--- a/secator/output_types/technology.py
+++ b/secator/output_types/technology.py
@@ -24,6 +24,8 @@ class Technology(OutputType):
 
 	_table_fields = ['match', 'product', 'version', 'extra_data']
 	_sort_by = ('match', 'product', 'version')
+	_group_by = ('product',)
+	_group_aggregate = 'match'
 
 	def __str__(self) -> str:
 		return self.match
diff --git a/secator/output_types/vulnerability.py b/secator/output_types/vulnerability.py
index a2ac29dc4..3e0a960fa 100644
--- a/secator/output_types/vulnerability.py
+++ b/secator/output_types/vulnerability.py
@@ -45,6 +45,8 @@ class Vulnerability(OutputType):
 
 	_table_fields = [MATCHED_AT, SEVERITY, CONFIDENCE, NAME, ID, CVSS_SCORE, STATUS, TAGS, EXTRA_DATA, REFERENCE]
 	_sort_by = ('confidence_nb', 'severity_nb', 'matched_at', 'cvss_score')
+	_group_by = (NAME,)
+	_group_aggregate = MATCHED_AT
 
 	@staticmethod
 	def cvss_to_severity(cvss):
diff --git a/secator/query/utils.py b/secator/query/utils.py
index a05752813..c45d68d09 100644
--- a/secator/query/utils.py
+++ b/secator/query/utils.py
@@ -552,6 +552,81 @@ def emit_query_warnings(warnings):
         _warn_unknown_field(field_name, type_name, valid_fields)
 
 
+def _finding_value(item, key):
+    """Read a field from a finding, which may be a dict or an OutputType object."""
+    return item.get(key) if isinstance(item, dict) else getattr(item, key, None)
+
+
+def _finding_ts(item):
+    return _finding_value(item, '_timestamp') or 0
+
+
+def _truncate_aggregate(values, max_display=5):
+    """Join aggregated values, truncating with '.. and X more' past max_display."""
+    values = [str(v) for v in values]
+    if len(values) <= max_display:
+        return ', '.join(values)
+    shown = ', '.join(values[:max_display])
+    return f'{shown} .. and {len(values) - max_display} more'
+
+
+def group_findings(items, group_by, aggregate_field=None, max_display=5):
+    """Group findings by one or more fields, collapsing each group to a single
+    representative (the newest finding) that carries the aggregated field values
+    and a `_group_count` attribute for display.
+
+    Args:
+        items (list): Findings (dicts or OutputType objects) of a single type.
+        group_by (list[str]): Field names to group by.
+        aggregate_field (str): Field whose distinct values are collected onto the
+            representative (truncated for display). None disables aggregation.
+        max_display (int): Max aggregated values shown before '.. and X more'.
+
+    Returns:
+        list: One representative finding per group (OutputType objects when the
+            input could be loaded, else the raw items), in first-seen order.
+    """
+    from secator.output_types import OUTPUT_TYPES
+    type_map = {cls.get_name(): cls for cls in OUTPUT_TYPES}
+
+    groups = {}
+    order = []
+    for item in items:
+        key = tuple(str(_finding_value(item, f)) for f in group_by)
+        if key not in groups:
+            groups[key] = {'rep': item, 'agg': [], 'count': 0}
+            order.append(key)
+        group = groups[key]
+        group['count'] += 1
+        if _finding_ts(item) >= _finding_ts(group['rep']):
+            group['rep'] = item
+        if aggregate_field:
+            value = _finding_value(item, aggregate_field)
+            if value and value not in group['agg']:
+                group['agg'].append(value)
+
+    out = []
+    for key in order:
+        group = groups[key]
+        rep = group['rep']
+        if isinstance(rep, dict):
+            cls = type_map.get(rep.get('_type'))
+            if cls:
+                try:
+                    rep = cls.load(rep)
+                except Exception as e:
+                    debug(f'group_findings: failed to load {rep.get("_type")} representative: {e}', sub='query')
+        if aggregate_field and not isinstance(rep, dict):
+            # When grouping, other fields (extra_data, etc.) keep the newest
+            # finding's values; only the aggregate field is replaced by the
+            # truncated collected values.
+            setattr(rep, aggregate_field, _truncate_aggregate(group['agg'], max_display))
+        if not isinstance(rep, dict):
+            rep._group_count = group['count']
+        out.append(rep)
+    return out
+
+
 def query_has_type_constraint(query):
     """Check whether a MongoDB-style query contains a '_type' constraint anywhere (recursively).
 
diff --git a/tests/unit/test_query_utils.py b/tests/unit/test_query_utils.py
index 6c438927a..29877a597 100644
--- a/tests/unit/test_query_utils.py
+++ b/tests/unit/test_query_utils.py
@@ -1,5 +1,6 @@
 from secator.query.utils import (
     expand_runner_paths,
+    group_findings,
     parse_report_paths,
     python_expr_to_mongo,
     validate_query_fields,
@@ -7,6 +8,36 @@
 )
 
 
+class TestGroupFindings:
+
+    def _vuln(self, name, matched_at, ts):
+        return {'_type': 'vulnerability', 'name': name, 'matched_at': matched_at, '_timestamp': ts}
+
+    def test_groups_by_name_with_count_and_aggregate(self):
+        findings = [
+            self._vuln('CVE-1', 'http://a.com', ts=1),
+            self._vuln('CVE-1', 'http://b.com', ts=2),
+            self._vuln('CVE-2', 'http://c.com', ts=3),
+        ]
+        grouped = group_findings(findings, group_by=['name'], aggregate_field='matched_at')
+
+        assert len(grouped) == 2
+        by_name = {g.name: g for g in grouped}
+        assert by_name['CVE-1']._group_count == 2
+        # Both matched_at values collected into the representative
+        assert 'http://a.com' in by_name['CVE-1'].matched_at
+        assert 'http://b.com' in by_name['CVE-1'].matched_at
+        assert by_name['CVE-2']._group_count == 1
+
+    def test_aggregate_truncates_past_max_display(self):
+        findings = [self._vuln('CVE-1', f'http://h{i}.com', ts=i) for i in range(8)]
+        grouped = group_findings(findings, group_by=['name'], aggregate_field='matched_at', max_display=5)
+
+        assert len(grouped) == 1
+        assert grouped[0]._group_count == 8
+        assert '.. and 3 more' in grouped[0].matched_at
+
+
 class TestResolveLocalReportPaths:
     """Local runs keep a sequential folder number (tasks/0) AND a UUID {type}_id (what findings carry).
     `report show tasks/0` must resolve the folder number to that UUID before it becomes a query filter."""