fix: CLI ErrorCode reuse + result_type→renderer parity contract test

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(

models/error_codes.py

@@ -14,6 +14,8 @@ class ErrorCode(StrEnum):
     SEARCH_INDEX_UNAVAILABLE = "SEARCH_INDEX_UNAVAILABLE"
     INVALID_PATH = "INVALID_PATH"
     SESSION_NOT_FOUND = "SESSION_NOT_FOUND"
+    SESSION_AMBIGUOUS_PREFIX = "SESSION_AMBIGUOUS_PREFIX"
+    PROJECTS_DIR_NOT_FOUND = "PROJECTS_DIR_NOT_FOUND"
     INVALID_REQUEST_BODY = "INVALID_REQUEST_BODY"
     INVALID_SINCE_MODE = "INVALID_SINCE_MODE"
     PARSE_ERROR = "PARSE_ERROR"

pyproject.toml

@@ -5,6 +5,7 @@ packages = ["api", "utils", "models"]
 exclude = ["tests/"]
 
 [tool.pytest.ini_options]
+pythonpath = ["."]
 addopts = "--cov=api --cov=utils --cov-report=term-missing --cov-report=xml:coverage.xml --benchmark-skip"
 testpaths = ["tests"]
 markers = [

scripts/export.py

@@ -24,20 +24,26 @@
 import sys
 import zipfile
 from datetime import datetime
-from typing import cast
+from typing import NoReturn, cast
 
 # Allow running from repo root or scripts/ directory
 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 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_code_for_exception,
+    failure_message_for_code,
     run_bulk_export,
     serialize_manifest_jsonl,
 )
@@ -176,8 +182,7 @@ def cmd_list(args):
     base_dir = getattr(args, "base_dir", None) or get_claude_projects_dir()
     project_filter = getattr(args, "project", None)
 
-    if not os.path.isdir(base_dir):
-        _die(f"Claude Code projects directory not found: {base_dir}")
+    _require_projects_dir(base_dir)
 
     projects = list_projects(base_dir)
     if project_filter:
@@ -223,8 +228,13 @@ def _list_sessions(project: dict):
             tokens = meta["total_input_tokens"] + meta["total_output_tokens"]
             tools = meta["total_tool_calls"]
             print(f"  {ts:<12} {title:<50} {sid:>10} {tokens:>10,} {tools:>6}")
-        except Exception as e:
-            print(f"  Warning: failed to parse {s['id'][:10]}: {e}", file=sys.stderr)
+        except Exception as exc:
+            code = failure_code_for_exception(exc)
+            print(
+                f"  Warning: failed to parse {s['id'][:10]}: "
+                f"{code.value} — {failure_message_for_code(code)}",
+                file=sys.stderr,
+            )
             continue
 
 
@@ -235,8 +245,7 @@ def cmd_stats(args):
     session_id = getattr(args, "session", None)
     fmt = getattr(args, "format", "text") or "text"
 
-    if not os.path.isdir(base_dir):
-        _die(f"Claude Code projects directory not found: {base_dir}")
+    _require_projects_dir(base_dir)
 
     if session_id:
         _session_stats(session_id, base_dir, fmt)
@@ -248,7 +257,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)
@@ -361,9 +370,11 @@ def _aggregate_stats(base_dir: str, project_filter: str, fmt: str):
                 if cost is not None:
                     totals["total_cost"] += cost
                     totals["has_cost"] = True
-            except Exception as e:
+            except Exception as exc:
+                code = failure_code_for_exception(exc)
                 print(
-                    f"  Warning: failed to parse {s['id'][:10]} in {project['name']}: {e}",
+                    f"  Warning: failed to parse {s['id'][:10]} in {project['name']}: "
+                    f"{code.value} — {failure_message_for_code(code)}",
                     file=sys.stderr,
                 )
                 continue
@@ -406,22 +417,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):
@@ -435,8 +451,7 @@ def cmd_export(args):
     session_filter = getattr(args, "session", None)
     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}")
+    _require_projects_dir(base_dir)
 
     rules = load_rules(resolve_exclusion_rules_path(exclusion_rules_path))
 
@@ -447,7 +462,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 +480,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 +713,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_AMBIGUOUS_PREFIX,
+            detail=(
+                f"Ambiguous prefix '{session_id}' matches {len(matches)} sessions:\n"
+                + "\n".join(f"  {m['id']}" for m in matches)
+            ),
         )
     return None
 
@@ -740,8 +762,25 @@ 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 _require_projects_dir(base_dir: str) -> None:
+    if not os.path.isdir(base_dir):
+        _die(
+            ErrorCode.PROJECTS_DIR_NOT_FOUND,
+            detail=f"Claude Code projects directory not found: {base_dir}",
+        )
+
+
+def _cli_die_message(code: ErrorCode) -> str:
+    """Stable CLI stderr label (not export-scoped)."""
+    if code == ErrorCode.INTERNAL_ERROR:
+        return "Command failed"
+    return failure_message_for_code(code)
+
+
+def _die(code: ErrorCode, *, detail: str | None = None) -> NoReturn:
+    print(f"Error: {code.value} — {_cli_die_message(code)}", file=sys.stderr)
+    if detail:
+        print(detail, file=sys.stderr)
     sys.exit(1)
 
 

tests/test_cli_export_exit_codes.py

@@ -3,19 +3,20 @@
 from __future__ import annotations
 
 import re
-import sys
 import types
 from pathlib import Path
 
 import pytest
 
-REPO_ROOT = Path(__file__).resolve().parent.parent
-sys.path.insert(0, str(REPO_ROOT))
-
 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 +94,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 +169,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 +218,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,15 @@
 from __future__ import annotations
 
 import json
+import types
 
 import pytest
 
+import scripts.export as export_cli
 from api.error_codes import ErrorCode
 from api.search import _IndexSearchOutcome
 from tests.conftest import assert_error_response
+from tests.test_cli_e2e import FIXTURES, _run_cli
 
 
 @pytest.mark.parametrize(
@@ -92,3 +95,58 @@ 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"], "PROJECTS_DIR_NOT_FOUND"),
+        (["stats", "--base-dir"], "PROJECTS_DIR_NOT_FOUND"),
+        (["list", "--base-dir"], "PROJECTS_DIR_NOT_FOUND"),
+    ],
+)
+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 "Claude projects directory not found" in proc.stderr
+    assert "Failed to export session" not in proc.stderr
+    assert "Command failed" not 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
+
+
+def test_cli_ambiguous_session_prefix_surfaces_code(tmp_path) -> None:
+    base = tmp_path / "projects"
+    project = base / "test-project"
+    project.mkdir(parents=True)
+    content = (FIXTURES / "session_minimal.jsonl").read_text(encoding="utf-8")
+    content = content.replace("demo-project", "test-project")
+    (project / "session_aaa111.jsonl").write_text(content, encoding="utf-8")
+    (project / "session_aaa222.jsonl").write_text(content, encoding="utf-8")
+
+    proc = _run_cli(["stats", "--base-dir", str(base), "--session", "session_aaa"])
+    assert proc.returncode == 1
+    assert "SESSION_AMBIGUOUS_PREFIX" in proc.stderr
+    assert "SESSION_NOT_FOUND" not in proc.stderr