diff --git a/README.md b/README.md
index c25ede3f09..b169bc2310 100644
--- a/README.md
+++ b/README.md
@@ -1643,6 +1643,13 @@ Users can explicitly activate an enabled skill for a single turn by starting the
 
 Data-analysis Markdown exports escape pipes and backslashes in column names and values, and render cell line breaks as `<br>` so multiline data stays within its table cell.
 
+Data-analysis JSON exports require unique result column names. For joins that
+return repeated names such as `id`, use SQL aliases (`AS left_id`, `AS right_id`)
+before exporting. Duplicate names cause an error before the output file is opened,
+preserving any existing file.
+The CLI names the repeated columns in a short stderr error and exits nonzero,
+so callers can correct their aliases without reading a traceback.
+
 The **data-analysis** skill preserves each CSV file or Excel sheet when names collide across inputs. [Prepare its Python environment](skills/public/data-analysis/SKILL.md#environment-setup) explicitly with DuckDB and openpyxl; missing packages report setup instructions instead of installing themselves during analysis. Use the SQL names printed by `inspect` (such as `report`, `report_1`) for queries and summaries. Its cache follows ordered input basenames and contents, so renaming or reordering inputs rebuilds their table map instead of silently selecting a different source. Quoted original names also resolve without rewriting query text; apostrophes in upload paths and double quotes in column names are preserved through loading, inspection, queries, and summaries.
 
 After an answer loads skills, its toolbar includes **Skills used**. Hover over or click the icon to see the skills and their sources; hovering a skill name underlines it. Select a skill to inspect its `SKILL.md` in the resizable side panel (a drawer on mobile). The view uses snapshots captured during successful configured read-tool loads or explicit slash activation, so later edits or removal of a skill do not rewrite its history. Repeated loads appear once per run, in first-load order. Range reads and bounded snapshots are labeled as partial; older conversations without captured evidence do not show this menu. Copy returns the captured Markdown, including YAML frontmatter. Package-relative links and images remain readable references rather than navigating away from the conversation. Skill loads performed through other tools, such as shell commands, are not inferred from answer text.
diff --git a/skills/public/data-analysis/AGENTS.md b/skills/public/data-analysis/AGENTS.md
index d06b6344a9..dde75f7be5 100644
--- a/skills/public/data-analysis/AGENTS.md
+++ b/skills/public/data-analysis/AGENTS.md
@@ -27,3 +27,12 @@ import-failure regressions offline in the root skill tests.
 Markdown exports use the same cell escaping for headers and values. Escape
 backslashes before pipes and normalize CRLF, CR, and LF to inline `<br>` breaks;
 raw newlines must not split one query-result row across Markdown table rows.
+
+JSON exports require unique result column names under case-sensitive JSON key
+semantics. Reject duplicate names before opening the output file and direct the
+caller to explicit SQL aliases; never silently overwrite one column with another.
+CSV and Markdown exports continue to preserve duplicate column headers.
+List each duplicate name once in result-column order, using `repr` so embedded
+line breaks remain escaped. At the CLI action-dispatch boundary, report
+`ValueError` as a short stderr error with a nonzero exit; close the connection
+in `finally` for both successful and failed actions. Other exceptions propagate.
diff --git a/skills/public/data-analysis/SKILL.md b/skills/public/data-analysis/SKILL.md
index 7f20f133ef..777ea1fb24 100644
--- a/skills/public/data-analysis/SKILL.md
+++ b/skills/public/data-analysis/SKILL.md
@@ -95,6 +95,12 @@ Supported output formats (auto-detected from extension):
 - `.json` — JSON array of records
 - `.md` — Markdown table
 
+JSON result column names must be unique. Use explicit SQL aliases for repeated
+names, for example `SELECT a.id AS left_id, b.id AS right_id FROM ...`.
+Duplicate names cause an error and leave any existing output file untouched.
+The error lists the repeated names so you can alias them directly. The CLI prints
+a short `Error: ...` message to stderr and exits nonzero for this rejection.
+
 ### Parameters
 
 | Parameter | Required | Description |
diff --git a/skills/public/data-analysis/scripts/analyze.py b/skills/public/data-analysis/scripts/analyze.py
index 2d550d0aa6..82bd478af7 100644
--- a/skills/public/data-analysis/scripts/analyze.py
+++ b/skills/public/data-analysis/scripts/analyze.py
@@ -13,6 +13,7 @@ import os
 import re
 import sys
 import tempfile
+from collections import Counter
 
 logging.basicConfig(level=logging.INFO, format="%(message)s")
 logger = logging.getLogger(__name__)
@@ -363,6 +364,13 @@ def _export_results(columns: list[str], rows: list[tuple], output_file: str) ->
             writer.writerows(rows)
 
     elif ext == ".json":
+        duplicates = [name for name, count in Counter(columns).items() if count > 1]
+        if duplicates:
+            raise ValueError(
+                "JSON export requires unique column names; duplicates: "
+                + ", ".join(repr(name) for name in duplicates)
+                + ". Use SQL aliases (AS) to give each result column a distinct name."
+            )
         records = []
         for row in rows:
             record = {}
@@ -585,15 +593,17 @@ def main():
         )
         logger.info(f"Cached database saved to: {db_path}")
 
-    # Perform action
-    if args.action == "inspect":
-        action_inspect(con, table_map)
-    elif args.action == "query":
-        action_query(con, args.sql, table_map, args.output_file)
-    elif args.action == "summary":
-        action_summary(con, args.table, table_map)
-
-    con.close()
+    try:
+        if args.action == "inspect":
+            action_inspect(con, table_map)
+        elif args.action == "query":
+            action_query(con, args.sql, table_map, args.output_file)
+        elif args.action == "summary":
+            action_summary(con, args.table, table_map)
+    except ValueError as exc:
+        sys.exit(f"Error: {exc}")
+    finally:
+        con.close()
 
 
 if __name__ == "__main__":
diff --git a/tests/skills/test_data_analysis_json_export.py b/tests/skills/test_data_analysis_json_export.py
new file mode 100644
index 0000000000..5f0aaf0861
--- /dev/null
+++ b/tests/skills/test_data_analysis_json_export.py
@@ -0,0 +1,290 @@
+"""JSON record exports must not silently discard duplicate SQL columns."""
+
+import csv
+import json
+import os
+import runpy
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+duckdb = pytest.importorskip("duckdb")
+pytest.importorskip("openpyxl")
+SCRIPT = (
+    Path(__file__).resolve().parents[2]
+    / "skills/public/data-analysis/scripts/analyze.py"
+)
+
+
+@pytest.fixture
+def analysis():
+    return runpy.run_path(str(SCRIPT))
+
+
+@pytest.fixture
+def cached_source(analysis, tmp_path):
+    # A real matching cache keeps CLI tests offline without a spatial download.
+    source = tmp_path / "input.csv"
+    source.write_text("id\n1\n", encoding="utf-8")
+    temp_root = tmp_path / "temp"
+    analysis["get_cache_db_path"].__globals__["CACHE_DIR"] = str(
+        temp_root / ".data-analysis-cache"
+    )
+    key = analysis["compute_files_hash"]([str(source)])
+    table_map = {}
+    with duckdb.connect(analysis["get_cache_db_path"](key)) as con:
+        analysis["_load_csv"](con, str(source), table_map)
+    analysis["save_table_map"](key, table_map)
+    return source, temp_root
+
+
+@pytest.fixture
+def opened_connections(analysis, monkeypatch):
+    connections = []
+
+    def connect(*args, **kwargs):
+        con = duckdb.connect(*args, **kwargs)
+        connections.append(con)
+        return con
+
+    monkeypatch.setitem(
+        analysis["main"].__globals__, "duckdb", SimpleNamespace(connect=connect)
+    )
+    yield connections
+    for con in connections:
+        con.close()
+
+
+@pytest.mark.parametrize("columns", [["id", "id"], ["数量", "数量"]])
+@pytest.mark.parametrize("rows", [[(1, 2)], []])
+@pytest.mark.parametrize("existing", [False, True])
+def test_json_export_rejects_duplicate_keys_before_writing(
+    analysis, tmp_path, columns, rows, existing
+):
+    output = tmp_path / "result.json"
+    original = b'[{"previous": "keep me"}]\n'
+    if existing:
+        output.write_bytes(original)
+
+    with pytest.raises(ValueError, match="unique column names.*SQL aliases"):
+        analysis["_export_results"](columns, rows, str(output))
+
+    if existing:
+        assert output.read_bytes() == original
+    else:
+        assert not output.exists()
+
+
+@pytest.mark.parametrize(
+    "columns, duplicates",
+    [
+        (["z", "id", "id", "z", "z", "ID", "other"], ["z", "id"]),
+        (["数量", "name", "数量", "name", "Name"], ["数量", "name"]),
+        (["line\nbreak", "", "line\nbreak", ""], ["line\nbreak", ""]),
+    ],
+)
+def test_json_error_names_exact_duplicates_once_in_column_order(
+    analysis, tmp_path, columns, duplicates
+):
+    with pytest.raises(ValueError) as error:
+        analysis["_export_results"](columns, [], str(tmp_path / "result.json"))
+
+    assert str(error.value) == (
+        "JSON export requires unique column names; duplicates: "
+        + ", ".join(repr(name) for name in duplicates)
+        + ". Use SQL aliases (AS) to give each result column a distinct name."
+    )
+    assert "\n" not in str(error.value)
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT 1 AS id, 2 AS id",
+        "SELECT a.id, b.id FROM (VALUES (1)) a(id) CROSS JOIN (VALUES (2)) b(id)",
+        "SELECT 1 AS id, 2 AS id WHERE FALSE",
+    ],
+)
+def test_query_json_export_rejects_duplicate_result_columns(analysis, tmp_path, sql):
+    output = tmp_path / "result.JSON"
+    with duckdb.connect() as con:
+        assert [column[0] for column in con.execute(sql).description] == ["id", "id"]
+        with pytest.raises(ValueError, match="unique column names.*SQL aliases"):
+            analysis["action_query"](con, sql, {}, str(output))
+    assert not output.exists()
+
+
+def test_json_export_preserves_unique_and_case_distinct_keys(analysis, tmp_path):
+    output = tmp_path / "result.json"
+    with duckdb.connect() as con:
+        analysis["action_query"](
+            con, 'SELECT 1 AS id, 2 AS ID, 3 AS "数量"', {}, str(output)
+        )
+    assert json.loads(output.read_text(encoding="utf-8")) == [
+        {"id": 1, "ID": 2, "数量": 3}
+    ]
+
+
+@pytest.mark.parametrize("extension", ["csv", "md"])
+def test_non_json_exports_keep_duplicate_columns(analysis, tmp_path, extension):
+    output = tmp_path / f"result.{extension}"
+    with duckdb.connect() as con:
+        analysis["action_query"](con, "SELECT 1 AS id, 2 AS id", {}, str(output))
+    if extension == "csv":
+        with output.open(encoding="utf-8", newline="") as exported:
+            assert list(csv.reader(exported)) == [["id", "id"], ["1", "2"]]
+    else:
+        assert output.read_text(encoding="utf-8").splitlines() == [
+            "| id | id |",
+            "| --- | --- |",
+            "| 1 | 2 |",
+        ]
+
+
+@pytest.mark.parametrize("existing", [False, True])
+def test_cli_rejects_duplicate_json_columns_without_overwriting(
+    cached_source, tmp_path, existing
+):
+    source, temp_root = cached_source
+    output = tmp_path / "result.json"
+    original = b'[{"previous": "keep me"}]\n'
+    if existing:
+        output.write_bytes(original)
+    completed = subprocess.run(
+        [
+            sys.executable,
+            str(SCRIPT),
+            "--files",
+            str(source),
+            "--action",
+            "query",
+            "--sql",
+            "SELECT id, id + 1 AS id FROM input",
+            "--output-file",
+            str(output),
+        ],
+        env={**os.environ, "TMPDIR": str(temp_root)},
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        timeout=30,
+    )
+
+    assert completed.returncode == 1
+    assert "Error: JSON export requires unique column names" in completed.stderr
+    assert "duplicates: 'id'" in completed.stderr
+    assert "SQL aliases" in completed.stderr
+    assert "Traceback" not in completed.stderr
+    assert "Results exported" not in completed.stdout
+    if existing:
+        assert output.read_bytes() == original
+    else:
+        assert not output.exists()
+
+
+def test_main_closes_connection_after_duplicate_export(
+    analysis, cached_source, opened_connections, tmp_path, monkeypatch
+):
+    source, _ = cached_source
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            str(SCRIPT),
+            "--files",
+            str(source),
+            "--action",
+            "query",
+            "--sql",
+            "SELECT id, id + 1 AS id FROM input",
+            "--output-file",
+            str(tmp_path / "result.json"),
+        ],
+    )
+
+    # The subprocess test pins the CLI message; this independently pins cleanup
+    # even on the pre-fix version, where the export error escapes as ValueError.
+    with pytest.raises((ValueError, SystemExit)):
+        analysis["main"]()
+    assert len(opened_connections) == 1
+    with pytest.raises(duckdb.ConnectionException, match="closed"):
+        opened_connections[0].execute("SELECT 1")
+
+
+@pytest.mark.parametrize(
+    "action, extra_args",
+    [
+        ("inspect", []),
+        ("query", ["--sql", "SELECT id FROM input"]),
+        ("summary", ["--table", "input"]),
+    ],
+)
+def test_main_closes_connection_after_success(
+    analysis, cached_source, opened_connections, monkeypatch, action, extra_args
+):
+    source, _ = cached_source
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [str(SCRIPT), "--files", str(source), "--action", action, *extra_args],
+    )
+    analysis["main"]()
+    assert len(opened_connections) == 1
+    with pytest.raises(duckdb.ConnectionException, match="closed"):
+        opened_connections[0].execute("SELECT 1")
+
+
+def test_main_closes_connection_without_swallowing_unexpected_errors(
+    analysis, cached_source, opened_connections, monkeypatch
+):
+    source, _ = cached_source
+    failure = RuntimeError("unexpected action failure")
+
+    def fail_inspection(con, table_map):
+        raise failure
+
+    monkeypatch.setitem(analysis["main"].__globals__, "action_inspect", fail_inspection)
+    monkeypatch.setattr(
+        sys, "argv", [str(SCRIPT), "--files", str(source), "--action", "inspect"]
+    )
+    with pytest.raises(RuntimeError) as error:
+        analysis["main"]()
+    assert error.value is failure
+    assert len(opened_connections) == 1
+    with pytest.raises(duckdb.ConnectionException, match="closed"):
+        opened_connections[0].execute("SELECT 1")
+
+
+def test_empty_cache_is_closed_before_removal(
+    analysis, opened_connections, tmp_path, monkeypatch
+):
+    namespace = analysis["main"].__globals__
+    monkeypatch.setitem(namespace, "CACHE_DIR", str(tmp_path / "cache"))
+    monkeypatch.setitem(namespace, "load_files", lambda con, files: {})
+    removed = []
+
+    def remove_closed_database(path):
+        assert len(opened_connections) == 1
+        with pytest.raises(duckdb.ConnectionException, match="closed"):
+            opened_connections[0].execute("SELECT 1")
+        removed.append(path)
+        os.remove(path)
+
+    monkeypatch.setitem(
+        namespace,
+        "os",
+        SimpleNamespace(**{**vars(os), "remove": remove_closed_database}),
+    )
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [str(SCRIPT), "--files", str(tmp_path / "missing.csv"), "--action", "inspect"],
+    )
+    with pytest.raises(SystemExit) as error:
+        analysis["main"]()
+    assert error.value.code == 1
+    assert len(removed) == 1
+    assert not Path(removed[0]).exists()