feat(cli): group similar indexed code chunks

README.md

@@ -295,6 +295,7 @@ The background daemon starts automatically on first use.
 | `ccc search <query>` | Semantic search across the codebase |
 | `ccc grep <pattern> [path]` | Structural code search by example (no index needed) |
 | `ccc status` | Show index stats (chunk count, file count, language breakdown) |
+| `ccc duplicates` | Group similar chunks from the existing index |
 | `ccc mcp` | Run as MCP server in stdio mode |
 | `ccc doctor` | Run diagnostics — checks settings, daemon, model, file matching, and index health |
 | `ccc reset` | Delete index databases. `--all` also removes settings. `-f` skips confirmation. |
@@ -315,6 +316,22 @@ ccc search --refresh database schema                 # update index first, then
 
 By default, `ccc search` scopes results to your current working directory (relative to the project root). Use `--path` to override.
 
+### Find similar code across a repository
+
+Build the index with `ccc index` before running `ccc duplicates`.
+
+```bash
+ccc duplicates --threshold 0.88 --min-lines 5 --lang python --path 'src/*'
+ccc duplicates --threshold 0.9 --json
+```
+
+The command reads stored embeddings without running the model or refreshing
+the index. Each group is connected by threshold-matching pairs. Two chunks in
+the same group can be connected through a third chunk even if their own score
+is below the threshold. Treat similarity groups as candidates for manual
+review. The exact comparison takes quadratic time in the number of
+selected chunks, so use file and language filters on large indexes.
+
 ### Structural Search (`ccc grep`)
 
 `ccc grep` finds code by **structure**, not text — you write a by-example pattern

src/cocoindex_code/_duplicates.py

@@ -0,0 +1,108 @@
+"""Threshold-based duplicate detection over an existing code index."""
+
+from __future__ import annotations
+
+import sqlite3
+from contextlib import closing
+from pathlib import Path
+
+import msgspec
+import numpy as np
+from cocoindex.connectors import sqlite
+
+_BLOCK_SIZE = 256
+
+
+class ChunkLocation(msgspec.Struct, frozen=True):
+    file_path: str
+    language: str
+    start_line: int
+    end_line: int
+
+
+def find_duplicate_groups(
+    database: Path,
+    *,
+    threshold: float,
+    min_lines: int,
+    languages: list[str],
+    path: str | None,
+) -> list[list[ChunkLocation]]:
+    """Read a snapshot and group chunks connected by cosine-similarity matches."""
+    conditions = ["end_line - start_line + 1 >= ?"]
+    params: list[str | int] = [min_lines]
+    if languages:
+        conditions.append(f"language IN ({', '.join('?' for _ in languages)})")
+        params.extend(languages)
+    if path is not None:
+        conditions.append("file_path GLOB ?")
+        params.append(path)
+
+    locations: list[ChunkLocation] = []
+    embeddings = []
+    try:
+        with (
+            closing(
+                sqlite.connect(f"{database.resolve().as_uri()}?mode=ro", uri=True, load_vec=True)
+            ) as db,
+            db.readonly() as conn,
+        ):
+            for file_path, language, start, end, embedding in conn.execute(
+                f"""
+                SELECT file_path, language, start_line, end_line, embedding
+                FROM code_chunks_vec
+                WHERE {" AND ".join(conditions)}
+                ORDER BY file_path, start_line, end_line, id
+                """,
+                params,
+            ):
+                vector = np.frombuffer(embedding, dtype=np.float32).astype(np.float64)
+                norm = np.linalg.norm(vector)
+                if np.isfinite(norm) and norm > 0:
+                    locations.append(ChunkLocation(file_path, language, start, end))
+                    embeddings.append(vector / norm)
+    except sqlite3.OperationalError as exc:
+        if not database.is_file() or "no such table: code_chunks_vec" in str(exc):
+            raise ValueError("Index not created yet. Run `ccc index` first.") from exc
+        raise
+
+    if len(embeddings) < 2:
+        return []
+    vectors = np.stack(embeddings)
+    del embeddings
+
+    # Union by size with path compression. Each qualifying pair is visited once.
+    parents = list(range(len(vectors)))
+    sizes = [1] * len(vectors)
+
+    def root(node: int) -> int:
+        while parents[node] != node:
+            parents[node] = parents[parents[node]]
+            node = parents[node]
+        return node
+
+    # Roundoff in a dot product of normalized vectors can put an exact match
+    # slightly below 1. Leave every other threshold unchanged.
+    cutoff = 1.0 - 1e-12 if threshold == 1.0 else threshold
+    for left_start in range(0, len(vectors), _BLOCK_SIZE):
+        left = vectors[left_start : left_start + _BLOCK_SIZE]
+        for right_start in range(left_start, len(vectors), _BLOCK_SIZE):
+            right = vectors[right_start : right_start + _BLOCK_SIZE]
+            matches = left @ right.T >= cutoff
+            if left_start == right_start:
+                matches = np.triu(matches, k=1)
+            for left_offset, right_offset in zip(*np.nonzero(matches), strict=True):
+                a = root(left_start + int(left_offset))
+                b = root(right_start + int(right_offset))
+                if a != b:
+                    if sizes[a] < sizes[b]:
+                        a, b = b, a
+                    parents[b] = a
+                    sizes[a] += sizes[b]
+
+    groups: dict[int, list[ChunkLocation]] = {}
+    for idx, location in enumerate(locations):
+        leader = root(idx)
+        if sizes[leader] > 1:
+            groups.setdefault(leader, []).append(location)
+    return list(groups.values())

src/cocoindex_code/cli.py

@@ -720,6 +720,49 @@ def search(
         print_search_results(resp)
 
 
+@app.command()
+def duplicates(
+    threshold: float = _typer.Option(0.88, min=0.0, max=1.0, help="Minimum cosine similarity"),
+    min_lines: int = _typer.Option(5, min=1, help="Minimum lines per chunk"),
+    lang: list[str] = _typer.Option([], "--lang", help="Filter by language"),
+    path: str | None = _typer.Option(None, "--path", help="Filter by file path glob"),
+    json_output: bool = _typer.Option(False, "--json", help="Output groups as JSON"),
+) -> None:
+    """Find groups of similar chunks in the existing index."""
+    import math
+    import sqlite3
+
+    from ._duplicates import find_duplicate_groups
+
+    if not math.isfinite(threshold):
+        raise _typer.BadParameter("must be finite", param_hint="--threshold")
+    project_root = require_project_root()
+    try:
+        groups = find_duplicate_groups(
+            target_sqlite_db_path(project_root),
+            threshold=threshold,
+            min_lines=min_lines,
+            languages=lang,
+            path=path,
+        )
+    except (sqlite3.Error, ValueError) as exc:
+        _typer.echo(f"Duplicate scan failed: {exc}", err=True)
+        raise _typer.Exit(code=1) from exc
+
+    if json_output:
+        sys.stdout.buffer.write(_msgspec.json.encode(groups) + b"\n")
+        sys.stdout.buffer.flush()
+    elif not groups:
+        _typer.echo("No similar chunks found.")
+    else:
+        for index, group in enumerate(groups, 1):
+            _typer.echo(f"Group {index} ({len(group)} chunks)")
+            for chunk in group:
+                _echo_search_text(
+                    f"  {chunk.file_path}:{chunk.start_line}-{chunk.end_line} [{chunk.language}]"
+                )
+
+
 @app.command()
 def grep(
     pattern: str = _typer.Argument(

tests/test_duplicates.py

@@ -0,0 +1,81 @@
+"""CLI duplicate detection against a real, on-disk SQLite vector index."""
+
+import json
+import struct
+from contextlib import closing
+from pathlib import Path
+
+import pytest
+from cocoindex.connectors import sqlite
+from typer.testing import CliRunner
+
+from cocoindex_code import _duplicates, cli, settings
+
+
+@pytest.fixture
+def index(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("COCOINDEX_CODE_DIR", str(tmp_path / "user"))
+    settings.save_user_settings(settings.default_user_settings())
+    settings.save_project_settings(tmp_path, settings.default_project_settings())
+    path: Path = settings.target_sqlite_db_path(tmp_path)
+    with closing(sqlite.connect(path, load_vec=True)) as db, db.transaction() as conn:
+        conn.execute(
+            "CREATE VIRTUAL TABLE code_chunks_vec USING vec0("
+            "id INTEGER PRIMARY KEY, +file_path TEXT, language TEXT partition key, "
+            "+content TEXT, +start_line INTEGER, +end_line INTEGER, embedding float[2])"
+        )
+        rows = [
+            ("src/a.py", "python", 5, (2.0, 0.0)),
+            ("src/b.py", "python", 5, (0.8, 0.6)),
+            ("src/c.py", "python", 5, (0.28, 0.96)),
+            ("src/short.py", "python", 2, (2.0, 0.0)),
+            ("src/main.rs", "rust", 5, (2.0, 0.0)),
+            ("test/z.py", "python", 5, (2.0, 0.0)),
+            ("src/zero.py", "python", 5, (0.0, 0.0)),
+        ]
+        conn.executemany(
+            "INSERT INTO code_chunks_vec VALUES (?, ?, ?, '', 1, ?, ?)",
+            [
+                (idx, file, language, end, struct.pack("ff", *vector))
+                for idx, (file, language, end, vector) in enumerate(rows)
+            ],
+        )
+    return path
+
+
+def test_duplicate_groups_follow_transitive_edges_across_blocks(
+    index: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(_duplicates, "_BLOCK_SIZE", 2)
+    result = CliRunner().invoke(
+        cli.app,
+        ["duplicates", "--lang", "python", "--path", "src/*", "--threshold", "0.75", "--json"],
+    )
+    assert result.exit_code == 0, (result.output, result.exception)
+    assert [[chunk["file_path"] for chunk in group] for group in json.loads(result.output)] == [
+        ["src/a.py", "src/b.py", "src/c.py"]
+    ]
+
+
+def test_exact_match_and_text_output(index: Path) -> None:
+    result = CliRunner().invoke(cli.app, ["duplicates", "--threshold", "1", "--min-lines", "3"])
+    assert result.exit_code == 0, (result.output, result.exception)
+    assert "Group 1 (3 chunks)" in result.output
+    assert "src/a.py:1-5" in result.output
+    assert "src/main.rs:1-5" in result.output
+    assert "test/z.py:1-5" in result.output
+
+
+def test_missing_index_does_not_create_database(index: Path) -> None:
+    index.unlink()
+    result = CliRunner().invoke(cli.app, ["duplicates"])
+    assert result.exit_code == 1
+    assert "Run `ccc index` first" in result.output
+    assert not index.exists()
+
+
+def test_nonfinite_threshold_is_rejected(index: Path) -> None:
+    result = CliRunner().invoke(cli.app, ["duplicates", "--threshold", "nan"])
+    assert result.exit_code == 2
+    assert "must be finite" in result.output