fix(cli): reuse structured ErrorCode in export CLI

Author: clean6378-max-it

api/export_api.py

@@ -146,8 +146,13 @@ def bulk_export() -> FlaskReturn:
 
     buf = io.BytesIO()
 
-    def _on_export_error(sid: str, exc: Exception) -> None:
-        current_app.logger.warning("Failed to export %s: %s", sid[:10], exc)
+    def _on_export_error(failure: ExportFailure) -> None:
+        current_app.logger.warning(
+            "Failed to export %s: %s — %s",
+            failure.session_id[:10],
+            failure.code.value,
+            failure.message,
+        )
 
     with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
         result = run_bulk_export(

scripts/export.py

@@ -31,13 +31,18 @@
 REPO_ROOT = os.path.dirname(SCRIPT_DIR)
 sys.path.insert(0, REPO_ROOT)
 
+from models.error_codes import ErrorCode
 from utils.exclusion_rules import load_rules, resolve_exclusion_rules_path
 from utils.export_engine import (
     BulkExportResult,
+    ExportFailure,
     ExportFormat,
     NoopSink,
     SinceMode,
     ZipSink,
+    bulk_export_exit_code,
+    dominant_failure_code,
+    failure_message_for_code,
     run_bulk_export,
     serialize_manifest_jsonl,
 )
@@ -177,7 +182,7 @@ def cmd_list(args):
     project_filter = getattr(args, "project", None)
 
     if not os.path.isdir(base_dir):
-        _die(f"Claude Code projects directory not found: {base_dir}")
+        _die(ErrorCode.INTERNAL_ERROR, detail=f"Claude Code projects directory not found: {base_dir}")
 
     projects = list_projects(base_dir)
     if project_filter:
@@ -236,7 +241,7 @@ def cmd_stats(args):
     fmt = getattr(args, "format", "text") or "text"
 
     if not os.path.isdir(base_dir):
-        _die(f"Claude Code projects directory not found: {base_dir}")
+        _die(ErrorCode.INTERNAL_ERROR, detail=f"Claude Code projects directory not found: {base_dir}")
 
     if session_id:
         _session_stats(session_id, base_dir, fmt)
@@ -248,7 +253,7 @@ def _session_stats(session_id: str, base_dir: str, fmt: str):
     """Detailed breakdown for one session: tokens, files, commands, cost."""
     filepath = _find_session(session_id, base_dir)
     if not filepath:
-        _die(f"Session not found: {session_id}")
+        _die(ErrorCode.SESSION_NOT_FOUND, detail=session_id)
 
     session = parse_session(filepath)
     stats = compute_stats(session)
@@ -406,22 +411,27 @@ def _aggregate_stats(base_dir: str, project_filter: str, fmt: str):
 
 
 def _exit_bulk_export(result: BulkExportResult) -> None:
-    """Map bulk-export counts to process exit code (CLI wrapper only).
+    """Map structured bulk-export failures to process exit code (CLI wrapper only).
 
     Prints a summary to stderr on any failure, stdout on clean success.
     Raises SystemExit(1) for total failure, SystemExit(2) for partial.
     """
     n = result.exported_session_count
-    k = result.failure_count
+    failures = result.failures
+    k = len(failures)
     # "attempted" = exported + failed; excludes untitled/excluded/mtime-skipped
     m = n + k
     if n > 0 or k > 0:
         dest = sys.stderr if k > 0 else sys.stdout
         print(f"Exported {n} of {m} sessions ({k} failed)", file=dest)
-    if n == 0 and k > 0:  # total failure
-        sys.exit(1)
-    elif k > 0:  # partial failure
-        sys.exit(2)
+    exit_code = bulk_export_exit_code(result)
+    if exit_code != 0:
+        code = dominant_failure_code(failures)
+        print(
+            f"  {code.value}: {failure_message_for_code(code)}",
+            file=sys.stderr,
+        )
+        sys.exit(exit_code)
 
 
 def cmd_export(args):
@@ -436,7 +446,7 @@ def cmd_export(args):
     exclusion_rules_path = getattr(args, "exclude_rules", None)
 
     if not os.path.isdir(base_dir):
-        _die(f"Claude Code projects directory not found: {base_dir}")
+        _die(ErrorCode.INTERNAL_ERROR, detail=f"Claude Code projects directory not found: {base_dir}")
 
     rules = load_rules(resolve_exclusion_rules_path(exclusion_rules_path))
 
@@ -447,7 +457,7 @@ def cmd_export(args):
     if session_filter:
         filepath = _find_session(session_filter, base_dir)
         if not filepath:
-            _die(f"Session not found: {session_filter}")
+            _die(ErrorCode.SESSION_NOT_FOUND, detail=session_filter)
         session = parse_session(filepath)
         stats = compute_stats(session)
         _export_single(session, stats, fmt, out_dir)
@@ -465,8 +475,12 @@ def cmd_export(args):
 
     skipped_mtime_unchanged = 0
 
-    def _on_export_error(sid: str, exc: Exception) -> None:
-        print(f"  Warning: failed to export {sid}: {exc}", file=sys.stderr)
+    def _on_export_error(failure: ExportFailure) -> None:
+        print(
+            f"  Warning: failed to export {failure.session_id}: "
+            f"{failure.code.value} — {failure.message}",
+            file=sys.stderr,
+        )
 
     collect_sink = NoopSink()
     export_result = run_bulk_export(
@@ -694,8 +708,11 @@ def _find_session(session_id: str, base_dir: str) -> str | None:
         return matches[0]["path"]
     if len(matches) > 1:
         _die(
-            f"Ambiguous prefix '{session_id}' matches {len(matches)} sessions:\n"
-            + "\n".join(f"  {m['id']}" for m in matches)
+            ErrorCode.SESSION_NOT_FOUND,
+            detail=(
+                f"Ambiguous prefix '{session_id}' matches {len(matches)} sessions:\n"
+                + "\n".join(f"  {m['id']}" for m in matches)
+            ),
         )
     return None
 
@@ -740,8 +757,10 @@ def _save_state(sessions: dict, count: int, out_dir: str):
         atomic_write_export_state(disk, STATE_FILE)
 
 
-def _die(msg: str):
-    print(f"Error: {msg}", file=sys.stderr)
+def _die(code: ErrorCode, *, detail: str | None = None) -> None:
+    print(f"Error: {code.value} — {failure_message_for_code(code)}", file=sys.stderr)
+    if detail:
+        print(detail, file=sys.stderr)
     sys.exit(1)
 
 

tests/test_cli_export_exit_codes.py

@@ -15,7 +15,7 @@
 import scripts.export as export
 from models.error_codes import ErrorCode
 from tests.test_cli_e2e import _run_cli, _seed_base_dir
-from utils.export_engine import BulkExportResult, ExportFailure
+from utils.export_engine import BulkExportResult, ExportFailure, bulk_export_exit_code, dominant_failure_code
 from utils.jsonl_parser import parse_session
 
 _SUMMARY_RE = re.compile(
@@ -93,6 +93,8 @@ def _parse(path: str):
     captured = capsys.readouterr()
     assert _SUMMARY_RE.search(captured.err), captured.err
     assert "Exported 1 of 2 sessions (1 failed)" in captured.err
+    assert "PARSE_ERROR" in captured.err
+    assert "simulated corrupt jsonl" not in captured.err
     assert len(list(out_dir.rglob("*.md"))) == 1
 
 
@@ -166,6 +168,7 @@ def test_since_last_early_return_exits_one_on_failure(tmp_path, monkeypatch, cap
     assert exc_info.value.code == 1
     captured = capsys.readouterr()
     assert "Exported 0 of 1 sessions (1 failed)" in captured.err
+    assert "PARSE_ERROR" in captured.err
 
 
 def test_cli_export_incremental_noop_no_stderr_summary(tmp_path):
@@ -214,5 +217,44 @@ def _parse(_path: str):
     assert exc_info.value.code == 1
     captured = capsys.readouterr()
     assert "Exported 0 of 2 sessions (2 failed)" in captured.err
+    assert "PARSE_ERROR" in captured.err
+    assert "simulated corrupt jsonl" not in captured.err
     assert "Nothing to export." in captured.out
     assert list(out_dir.rglob("*.md")) == []
+
+
+@pytest.mark.parametrize(
+    "result,expected_exit",
+    [
+        (BulkExportResult(), 0),
+        (
+            BulkExportResult(
+                exported_session_count=2,
+                failures=[
+                    ExportFailure("a", "Failed to parse session", ErrorCode.PARSE_ERROR),
+                ],
+            ),
+            2,
+        ),
+        (
+            BulkExportResult(
+                failures=[
+                    ExportFailure("a", "Failed to parse session", ErrorCode.PARSE_ERROR),
+                    ExportFailure("b", "Failed to export session", ErrorCode.INTERNAL_ERROR),
+                ],
+            ),
+            1,
+        ),
+    ],
+)
+def test_bulk_export_exit_code_mapping(result: BulkExportResult, expected_exit: int) -> None:
+    assert bulk_export_exit_code(result) == expected_exit
+
+
+def test_dominant_failure_code_picks_most_common() -> None:
+    failures = [
+        ExportFailure("a", "Failed to parse session", ErrorCode.PARSE_ERROR),
+        ExportFailure("b", "Failed to parse session", ErrorCode.PARSE_ERROR),
+        ExportFailure("c", "Failed to export session", ErrorCode.INTERNAL_ERROR),
+    ]
+    assert dominant_failure_code(failures) == ErrorCode.PARSE_ERROR

tests/test_error_codes.py

@@ -3,12 +3,21 @@
 from __future__ import annotations
 
 import json
+import sys
+import types
+from pathlib import Path
 
 import pytest
 
 from api.error_codes import ErrorCode
 from api.search import _IndexSearchOutcome
 from tests.conftest import assert_error_response
+from tests.test_cli_e2e import _run_cli
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+import scripts.export as export_cli
 
 
 @pytest.mark.parametrize(
@@ -92,3 +101,39 @@ def _raise_live_scan_failure(*_args, **_kwargs):
     body_text = json.dumps(resp.get_json())
     assert_error_response(resp, expected_code=ErrorCode.SEARCH_INDEX_UNAVAILABLE)
     assert "live scan failed" not in body_text
+
+
+@pytest.mark.parametrize(
+    "argv,code",
+    [
+        (["export", "--base-dir"], "INTERNAL_ERROR"),
+        (["stats", "--base-dir"], "INTERNAL_ERROR"),
+    ],
+)
+def test_cli_missing_projects_dir_surfaces_error_code(tmp_path, argv: list[str], code: str) -> None:
+    missing = tmp_path / "missing-claude-dir"
+    proc = _run_cli([*argv, str(missing)])
+    assert proc.returncode == 1
+    assert code in proc.stderr
+    assert "Traceback" not in proc.stderr
+
+
+def test_cli_export_session_not_found_surfaces_code(tmp_path, capsys) -> None:
+    base = tmp_path / "projects"
+    base.mkdir()
+    with pytest.raises(SystemExit) as exc_info:
+        export_cli.cmd_export(
+            types.SimpleNamespace(
+                base_dir=str(base),
+                out=str(tmp_path / "out"),
+                since="all",
+                no_zip=True,
+                project=None,
+                format="md",
+                session="missing-session-id",
+                exclude_rules=None,
+            )
+        )
+    assert exc_info.value.code == 1
+    captured = capsys.readouterr()
+    assert "SESSION_NOT_FOUND" in captured.err

tests/test_tool_dispatch_sync.py

@@ -5,6 +5,10 @@
 - ``utils/md_exporter.py`` — ``_render_tool_use`` if/elif branches (parsed)
 - ``models/tool_results.py`` — ``ToolNameLiteral``
 - ``static/js/render/registry.js`` — ``TOOL_USE_RENDERERS`` keys (parsed)
+
+``result_type`` values from ``_TOOL_RESULT_DISPATCH`` builders must match:
+- ``static/js/render/registry.js`` — ``TOOL_RESULT_RENDERERS`` keys (parsed)
+- ``utils/md_exporter.py`` — ``_render_tool_result`` if/elif branches (parsed)
 """
 
 from __future__ import annotations
@@ -23,6 +27,7 @@
 _REPO_ROOT = Path(__file__).resolve().parents[1]
 _FRONTEND_REGISTRY = _REPO_ROOT / "static" / "js" / "render" / "registry.js"
 _MD_EXPORTER = _REPO_ROOT / "utils" / "md_exporter.py"
+_TOOL_DISPATCH = _REPO_ROOT / "utils" / "tool_dispatch.py"
 _TOOL_TYPES_MANIFEST = _REPO_ROOT / "static" / "tool_types.json"
 
 
@@ -67,6 +72,55 @@ def _parse_frontend_tool_use_renderers(path: Path) -> frozenset[str]:
     return frozenset(keys)
 
 
+def _parse_frontend_tool_result_renderers(path: Path) -> frozenset[str]:
+    """Extract ``TOOL_RESULT_RENDERERS`` keys (brace-depth safe)."""
+    text = path.read_text(encoding="utf-8")
+    marker = "export const TOOL_RESULT_RENDERERS = {"
+    start = text.find(marker)
+    if start == -1:
+        msg = f"Could not find TOOL_RESULT_RENDERERS in {path}"
+        raise ValueError(msg)
+    i = start + len(marker)
+    depth = 1
+    body_start = i
+    while i < len(text) and depth > 0:
+        ch = text[i]
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+        i += 1
+    if depth != 0:
+        msg = f"Unbalanced braces in TOOL_RESULT_RENDERERS in {path}"
+        raise ValueError(msg)
+    body = text[body_start : i - 1]
+    keys = re.findall(r"^\s*(\w+)\s*:", body, re.MULTILINE)
+    return frozenset(keys)
+
+
+def _parse_dispatch_builder_result_types(path: Path) -> frozenset[str]:
+    """Extract ``result_type`` literals from tool-result dispatch builders."""
+    text = path.read_text(encoding="utf-8")
+    types = set(re.findall(r'result\["result_type"\]\s*=\s*"([^"]+)"', text))
+    types.discard("unknown")
+    return frozenset(types)
+
+
+def _parse_md_exporter_tool_result_handlers(path: Path) -> frozenset[str]:
+    """Extract ``result_type`` values handled by ``_render_tool_result`` branches."""
+    text = path.read_text(encoding="utf-8")
+    match = re.search(
+        r"def _render_tool_result\(.*?(?=\ndef _render_system)",
+        text,
+        re.DOTALL,
+    )
+    if not match:
+        msg = f"Could not find _render_tool_result in {path}"
+        raise ValueError(msg)
+    body = match.group(0)
+    return frozenset(re.findall(r'(?:if|elif) rt == "([^"]+)"', body))
+
+
 def _parse_md_exporter_tool_use_handlers(path: Path) -> frozenset[str]:
     """Extract tool names handled by ``_render_tool_use`` if/elif branches."""
     text = path.read_text(encoding="utf-8")
@@ -149,3 +203,32 @@ def test_frontend_registry_matches_known_tool_types() -> None:
 
 def test_known_tool_types_nonempty() -> None:
     assert KNOWN_TOOL_TYPES
+
+
+def test_dispatch_builder_result_types_match_frontend_registry() -> None:
+    """``TOOL_RESULT_RENDERERS`` keys must match dispatch builder ``result_type`` values."""
+    site = "static/js/render/registry.js (TOOL_RESULT_RENDERERS)"
+    try:
+        expected = _parse_dispatch_builder_result_types(_TOOL_DISPATCH)
+        actual = _parse_frontend_tool_result_renderers(_FRONTEND_REGISTRY)
+    except ValueError as exc:
+        pytest.fail(f"{site}: {exc}")
+    if actual != expected:
+        pytest.fail(_format_set_diff(expected, actual, site))
+
+
+def test_dispatch_builder_result_types_match_md_exporter() -> None:
+    """``_render_tool_result`` branches must match dispatch builder ``result_type`` values."""
+    site = "utils/md_exporter.py (_render_tool_result branches)"
+    try:
+        expected = _parse_dispatch_builder_result_types(_TOOL_DISPATCH)
+        actual = _parse_md_exporter_tool_result_handlers(_MD_EXPORTER)
+    except ValueError as exc:
+        pytest.fail(f"{site}: {exc}")
+    if actual != expected:
+        pytest.fail(_format_set_diff(expected, actual, site))
+
+
+def test_dispatch_builder_result_types_nonempty() -> None:
+    expected = _parse_dispatch_builder_result_types(_TOOL_DISPATCH)
+    assert expected