fix(export): address CodeRabbit review (engine hardening + manifest parity)

- Resolve first_timestamp from mtime without mutating SessionDict metadata
- Apply mtime fallback for API and CLI export paths
- Guard invalid since in run_bulk_export; use Literal types for since/fmt
- Export MANIFEST_SHARED_KEYS; unify manifest JSONL via serialize_manifest_jsonl
- CLI zip uses ZipSink.finalize; --no-zip uses same serializer
- Public format_duration; ListSink no longer duplicates BulkExportResult.exports

Author: clean6378-max-it

scripts/export.py

@@ -18,6 +18,7 @@
 import sys
 import zipfile
 from datetime import datetime
+from typing import cast
 
 # Allow running from repo root or scripts/ directory
 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -26,12 +27,19 @@
 
 from utils.session_path import get_claude_projects_dir, list_projects, list_sessions
 from utils.jsonl_parser import parse_session
-from utils.session_stats import compute_stats, _format_duration
+from utils.session_stats import compute_stats, format_duration
 from utils.md_exporter import session_to_markdown
 from utils.json_exporter import session_to_json
 from utils.exclusion_rules import resolve_exclusion_rules_path, load_rules
 from utils.slugify import slugify
-from utils.export_engine import ListSink, run_bulk_export
+from utils.export_engine import (
+    ExportFormat,
+    ListSink,
+    SinceMode,
+    ZipSink,
+    run_bulk_export,
+    serialize_manifest_jsonl,
+)
 from utils.export_state_store import (
     atomic_write_export_state,
     export_state_lock,
@@ -247,7 +255,7 @@ def _session_stats(session_id: str, base_dir: str, fmt: str):
     print(f"  Title:      {session['title']}")
     if meta["first_timestamp"]:
         print(f"  Created:    {meta['first_timestamp'][:19]}")
-    dur = _format_duration(meta.get("session_wall_time_seconds"))
+    dur = format_duration(meta.get("session_wall_time_seconds"))
     if dur:
         print(f"  Duration:   {dur}")
     print(f"  Models:     {', '.join(meta['models_used']) or 'unknown'}")
@@ -432,11 +440,11 @@ def _on_export_error(sid: str, exc: Exception) -> None:
     collect_sink = ListSink()
     export_result = run_bulk_export(
         projects=projects,
-        since=since,
+        since=cast(SinceMode, since),
         rules=rules,
         last_export_sessions=last_export,
         sink=collect_sink,
-        fmt=fmt,
+        fmt=cast(ExportFormat, fmt),
         path_layout="cli",
         manifest_style="cli",
         on_export_error=_on_export_error,
@@ -502,8 +510,7 @@ def _on_export_error(sid: str, exc: Exception) -> None:
                 f.write(content)
         manifest_path = os.path.join(out_dir, "manifest.jsonl")
         with open(manifest_path, "w", encoding="utf-8") as f:
-            for entry in manifest:
-                f.write(json.dumps(entry, default=str) + "\n")
+            f.write(serialize_manifest_jsonl(manifest))
         print(f"Exported {exported} file(s) to {out_dir}")
     else:
         date_tag = datetime.now().strftime("%Y-%m-%d")
@@ -518,10 +525,7 @@ def _on_export_error(sid: str, exc: Exception) -> None:
         with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
             for rel_path, content in all_exports:
                 zf.writestr(rel_path, content)
-            manifest_str = "\n".join(
-                json.dumps(e, default=str) for e in manifest
-            )
-            zf.writestr("manifest.jsonl", manifest_str)
+            ZipSink(zf).finalize(manifest)
         print(f"Exported {exported} file(s) to {zip_path}")
 
     _save_state(last_export, count=len(manifest), out_dir=out_dir)

tests/test_export_engine_parity.py

@@ -18,20 +18,13 @@
 from api.export_api import export_bp  # noqa: E402
 from tests.test_cli_e2e import _run_cli, _seed_base_dir  # noqa: E402
 from utils.export_engine import (  # noqa: E402
+    MANIFEST_SHARED_KEYS,
     ListSink,
     manifest_shared_subset,
     run_bulk_export,
 )
 from utils.session_path import list_projects  # noqa: E402
 
-_MANIFEST_KEYS = (
-    "session_id",
-    "title",
-    "project",
-    "tokens",
-    "tool_calls",
-)
-
 
 def _markdown_from_exports(exports: list[tuple[str, str]]) -> str:
     md_paths = [p for p, _ in exports if p.endswith(".md")]
@@ -75,7 +68,7 @@ def test_engine_api_vs_cli_layout_same_markdown_and_manifest(tmp_path: Path) ->
 
     api_core = manifest_shared_subset(api_result.manifest[0])
     cli_core = manifest_shared_subset(cli_result.manifest[0])
-    for key in _MANIFEST_KEYS:
+    for key in MANIFEST_SHARED_KEYS:
         assert api_core[key] == cli_core[key]
 
 
@@ -126,6 +119,6 @@ def test_http_post_export_matches_cli_no_zip(tmp_path: Path, monkeypatch) -> Non
     ]
 
     assert http_md == cli_md
-    http_core = {k: http_manifest[0][k] for k in _MANIFEST_KEYS}
-    cli_core = {k: cli_manifest[0][k] for k in _MANIFEST_KEYS}
+    http_core = {k: http_manifest[0][k] for k in MANIFEST_SHARED_KEYS}
+    cli_core = {k: cli_manifest[0][k] for k in MANIFEST_SHARED_KEYS}
     assert http_core == cli_core

utils/export_engine.py

@@ -31,8 +31,10 @@
 
 PathLayout = Literal["api", "cli"]
 ManifestStyle = Literal["api", "cli"]
+SinceMode = Literal["all", "last", "incremental"]
+ExportFormat = Literal["md", "json", "both"]
 
-_MANIFEST_SHARED_KEYS = (
+MANIFEST_SHARED_KEYS: tuple[str, ...] = (
     "session_id",
     "title",
     "project",
@@ -41,10 +43,18 @@
 )
 
 
+def serialize_manifest_jsonl(manifest: list[dict[str, Any]]) -> str:
+    """JSONL manifest body with a trailing newline (empty string if no rows)."""
+    if not manifest:
+        return ""
+    return "\n".join(json.dumps(e, default=str) for e in manifest) + "\n"
+
+
 @dataclass
 class BulkExportResult:
     """Outcome of a bulk export run."""
 
+    # Canonical list of (rel_path, content); sinks do not duplicate this.
     exports: list[tuple[str, str]] = field(default_factory=list)
     manifest: list[dict[str, Any]] = field(default_factory=list)
     new_sessions_map: dict[str, float] = field(default_factory=dict)
@@ -64,24 +74,24 @@ def add_session(
         self,
         files: list[tuple[str, str]],
         manifest_entry: dict[str, Any],
-    ) -> None: ...
+    ) -> None:
+        """Write one session's export file(s) to the sink target."""
 
-    def finalize(self, manifest: list[dict[str, Any]]) -> None: ...
+    def finalize(self, manifest: list[dict[str, Any]]) -> None:
+        """Flush manifest and any sink-specific completion (e.g. manifest.jsonl)."""
 
 
 @dataclass
 class ListSink:
-    """In-memory sink for tests and parity checks."""
+    """In-memory sink for tests; use :attr:`BulkExportResult.exports` for file pairs."""
 
-    exports: list[tuple[str, str]] = field(default_factory=list)
     manifest: list[dict[str, Any]] = field(default_factory=list)
 
     def add_session(
         self,
         files: list[tuple[str, str]],
         manifest_entry: dict[str, Any],
     ) -> None:
-        self.exports.extend(files)
         self.manifest.append(manifest_entry)
 
     def finalize(self, manifest: list[dict[str, Any]]) -> None:
@@ -105,21 +115,20 @@ def add_session(
         self._manifest.append(manifest_entry)
 
     def finalize(self, manifest: list[dict[str, Any]]) -> None:
-        if manifest:
-            manifest_str = "\n".join(json.dumps(e, default=str) for e in manifest) + "\n"
-            self._zf.writestr("manifest.jsonl", manifest_str)
+        body = serialize_manifest_jsonl(manifest)
+        if body:
+            self._zf.writestr("manifest.jsonl", body)
 
 
-def _ensure_first_timestamp(
+def _resolve_first_timestamp(
     meta: SessionMetadataDict, sess_info: SessionListItemDict
 ) -> str:
-    ts_raw = meta.get("first_timestamp") or ""
-    ts = ts_raw if ts_raw else ""
+    """Return first_timestamp from metadata, or synthesise from mtime without mutating *meta*."""
+    ts = (meta.get("first_timestamp") or "").strip()
     if not ts:
         ts = datetime.fromtimestamp(sess_info["modified"]).strftime(
             "%Y-%m-%dT%H:%M:%S"
         )
-        meta["first_timestamp"] = ts
     return ts
 
 
@@ -138,10 +147,8 @@ def build_export_rel_path(
     """Build zip/disk-relative path for one exported session file."""
     sid = sess_info["id"]
     meta = session["metadata"]
-    ts = meta.get("first_timestamp") or ""
-    if layout == "cli" and not ts:
-        ts = _ensure_first_timestamp(meta, sess_info)
-    date_str = ts[:10] if ts else "0000-00-00"
+    ts = _resolve_first_timestamp(meta, sess_info)
+    date_str = ts[:10]
     ts_file = _ts_file_slug(ts)
     title_slug = slugify(session["title"], default="session")
     short_id = sid[:8]
@@ -185,15 +192,15 @@ def build_manifest_entry(
 
 def manifest_shared_subset(entry: dict[str, Any]) -> dict[str, Any]:
     """Core manifest fields compared in HTTP vs CLI parity tests."""
-    return {k: entry[k] for k in _MANIFEST_SHARED_KEYS if k in entry}
+    return {k: entry[k] for k in MANIFEST_SHARED_KEYS if k in entry}
 
 
 def _session_files(
     session: SessionDict,
     stats: SessionStatsDict,
     project: ProjectDict,
     sess_info: SessionListItemDict,
-    fmt: str,
+    fmt: ExportFormat,
     layout: PathLayout,
 ) -> list[tuple[str, str]]:
     files: list[tuple[str, str]] = []
@@ -223,11 +230,11 @@ def _session_files(
 def run_bulk_export(
     *,
     projects: list[ProjectDict],
-    since: str,
+    since: SinceMode,
     rules: list[Any],
     last_export_sessions: dict[str, float],
     sink: ExportSink,
-    fmt: str = "md",
+    fmt: ExportFormat = "md",
     path_layout: PathLayout = "api",
     manifest_style: ManifestStyle | None = None,
     on_export_error: Callable[[str, Exception], None] | None = None,
@@ -237,6 +244,9 @@ def run_bulk_export(
     *since* must be one of ``all``, ``last``, or ``incremental``.
     Per-session failures are caught, counted, and skipped (batch continues).
     """
+    if since not in ("all", "last", "incremental"):
+        raise ValueError(f"Invalid since mode: {since!r}")
+
     if manifest_style is None:
         manifest_style = path_layout
 

utils/session_stats.py

@@ -197,7 +197,7 @@ def _summarize_tool_results(messages: list[MessageDict]) -> dict[str, int]:
     return summary
 
 
-def _format_duration(seconds: float | int | None) -> str | None:
+def format_duration(seconds: float | int | None) -> str | None:
     """Turn seconds into something like '2h 15m' or '45s'."""
     if seconds is None:
         return None
@@ -211,3 +211,6 @@ def _format_duration(seconds: float | int | None) -> str | None:
     hours = minutes // 60
     mins = minutes % 60
     return f"{hours}h {mins}m"
+
+
+_format_duration = format_duration  # backward compat for internal callers