diff --git a/services/workspace_tabs.py b/services/workspace_tabs.py index ab2588c..4e41693 100644 --- a/services/workspace_tabs.py +++ b/services/workspace_tabs.py @@ -5,9 +5,9 @@ import logging import os import sqlite3 -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from datetime import datetime -from typing import Any, cast +from typing import Any _logger = logging.getLogger(__name__) @@ -19,6 +19,7 @@ ) from utils.exclusion_rules import RuleTokens, build_searchable_text, is_excluded_by_rules from utils.display_bubble import ( + annotate_response_times, bubble_display_timestamp_ms, build_storage_bubble_metadata, ) @@ -29,9 +30,8 @@ Composer, DisplayBubble, ParseWarningCollector, - SchemaError, ) -from models.bubble_display import BubbleMetadata, BubbleRole +from models.bubble_display import BubbleRole from models.raw_access import ( conversation_header_bubble_id, message_request_context_project_layouts, @@ -74,6 +74,24 @@ ) +ERR_CONVERSATION_NOT_FOUND = "Conversation not found" +ERR_GLOBAL_STORAGE_NOT_FOUND = "Global storage not found" +_COMPOSER_ROW_ERRORS = (KeyError, TypeError, ValueError, json.JSONDecodeError) + + +def _record_composer_processing_failure( + composer_id: str, + exc: BaseException, + *, + parse_warnings: ParseWarningCollector, +) -> None: + _logger.warning( + "Failed to process Composer from composerData:%s: %s", + composer_id, + exc, + ) + parse_warnings.record_composer_processing_failure() + def _loads_kv_value_logged(key: str, raw: object | None) -> Any | None: """Parse a cursorDiskKV ``value``; log and return ``None`` on decode failure.""" @@ -122,38 +140,63 @@ def _kv_payload_log_meta(value: object | None) -> tuple[int, str | None]: return len(payload), hashlib.sha256(payload).hexdigest()[:12] -def _assemble_tab_from_composer_data( +def _context_text_for_bubble( + bubble_id: str, + contexts: list[dict[str, Any]], +) -> str: + """Append message-request context blocks for one bubble.""" + parts: list[str] = [] + for ctx in contexts: + if ctx.get("bubbleId") != bubble_id: + continue + if ctx.get("gitStatusRaw"): + parts.append(f"\n\n**Git Status:**\n```\n{ctx['gitStatusRaw']}\n```") + tf = ctx.get("terminalFiles") + if isinstance(tf, list) and tf: + parts.append("\n\n**Terminal Files:**") + for f in tf: + if not isinstance(f, dict): + continue + parts.append(f"\n- {f.get('path', '')}") + af = ctx.get("attachedFoldersListDirResults") + if isinstance(af, list) and af: + parts.append("\n\n**Attached Folders:**") + for fld in af: + if not isinstance(fld, dict): + continue + files = fld.get("files") + if isinstance(files, list) and files: + parts.append(f"\n\n**Folder:** {fld.get('path', 'Unknown')}") + for fi in files: + if not isinstance(fi, dict): + continue + parts.append(f"\n- {fi.get('name', '')} ({fi.get('type', '')})") + cr = ctx.get("cursorRules") + if isinstance(cr, list) and cr: + parts.append("\n\n**Cursor Rules:**") + for rule in cr: + if not isinstance(rule, dict): + continue + parts.append(f"\n- {rule.get('name') or rule.get('description') or 'Rule'}") + sc = ctx.get("summarizedComposers") + if isinstance(sc, list) and sc: + parts.append("\n\n**Related Conversations:**") + for comp in sc: + if not isinstance(comp, dict): + continue + parts.append(f"\n- {comp.get('name') or comp.get('composerId') or 'Conversation'}") + return "".join(parts) + + +def _build_tab_display_bubbles( composer_id: str, composer: Composer, bubble_map: Mapping[str, Bubble], contexts: list[dict[str, Any]], - code_block_diffs: list[dict[str, Any]], - workspace_display_name: str, - rules: list[RuleTokens], - parse_warnings: ParseWarningCollector, -) -> dict[str, Any] | None: - """Assemble a single tab dict from a validated :class:`Composer`. - - Args: - composer_id: Composer UUID. - composer: Validated composer model (typed field access, not raw dict reads). - bubble_map: ``{bubble_id: Bubble}`` — global or scoped. - contexts: ``messageRequestContext`` entries for *this* composer - (list of dicts, each with an injected ``contextId`` key and a - ``bubbleId`` field from the JSON value). - code_block_diffs: ``codeBlockDiff`` entries for *this* composer. - workspace_display_name: Human-readable workspace name for rule matching. - rules: Exclusion rule token lists. - parse_warnings: Collector for skipped-bubble warnings. - - Returns: - A tab dict on success, or ``None`` when the tab should be omitted - (no renderable bubbles or excluded by rules). - """ - headers = composer.full_conversation_headers_only - +) -> list[DisplayBubble]: + """Build renderable display bubbles from composer headers and storage.""" bubbles: list[DisplayBubble] = [] - for header in headers: + for header in composer.full_conversation_headers_only: if not isinstance(header, dict): continue bubble_id = conversation_header_bubble_id(header, composer_id=composer_id) @@ -166,48 +209,7 @@ def _assemble_tab_from_composer_data( is_user = header.get("type") == 1 msg_type: BubbleRole = "user" if is_user else "ai" text = extract_text_from_bubble(bubble) - - context_text = "" - for ctx in contexts: - if ctx.get("bubbleId") == bubble_id: - if ctx.get("gitStatusRaw"): - context_text += f"\n\n**Git Status:**\n```\n{ctx['gitStatusRaw']}\n```" - tf = ctx.get("terminalFiles") - if isinstance(tf, list) and tf: - context_text += "\n\n**Terminal Files:**" - for f in tf: - if not isinstance(f, dict): - continue - context_text += f"\n- {f.get('path', '')}" - af = ctx.get("attachedFoldersListDirResults") - if isinstance(af, list) and af: - context_text += "\n\n**Attached Folders:**" - for fld in af: - if not isinstance(fld, dict): - continue - files = fld.get("files") - if isinstance(files, list) and files: - context_text += f"\n\n**Folder:** {fld.get('path', 'Unknown')}" - for fi in files: - if not isinstance(fi, dict): - continue - context_text += f"\n- {fi.get('name', '')} ({fi.get('type', '')})" - cr = ctx.get("cursorRules") - if isinstance(cr, list) and cr: - context_text += "\n\n**Cursor Rules:**" - for rule in cr: - if not isinstance(rule, dict): - continue - context_text += f"\n- {rule.get('name') or rule.get('description') or 'Rule'}" - sc = ctx.get("summarizedComposers") - if isinstance(sc, list) and sc: - context_text += "\n\n**Related Conversations:**" - for comp in sc: - if not isinstance(comp, dict): - continue - context_text += f"\n- {comp.get('name') or comp.get('composerId') or 'Conversation'}" - - full_text = text + context_text + full_text = text + _context_text_for_bubble(bubble_id, contexts) metadata = build_storage_bubble_metadata(bubble, msg_type) tool_calls = (metadata or {}).get("toolCalls") thinking = (metadata or {}).get("thinking") @@ -231,12 +233,17 @@ def _assemble_tab_from_composer_data( "timestamp": bubble_display_timestamp_ms(bubble), } if metadata: - b_entry["metadata"] = cast(BubbleMetadata, metadata) + b_entry["metadata"] = metadata bubbles.append(b_entry) + return bubbles - if not bubbles: - return None +def _derive_composer_tab_title( + composer_id: str, + composer: Composer, + bubbles: list[DisplayBubble], +) -> str: + """Derive sidebar title from composer name or first user message.""" title = composer.name or f"Conversation {composer_id[:8]}" if not composer.name and bubbles: first_msg = bubbles[0].get("text", "") @@ -246,28 +253,14 @@ def _assemble_tab_from_composer_data( title = first_lines[0][:100] if len(title) == 100: title += "..." + return title - _early_model_name = composer.model_name_from_config() - _early_model_names = [_early_model_name] if _early_model_name and _early_model_name != "default" else None - if is_excluded_by_rules(rules, build_searchable_text( - project_name=workspace_display_name, - chat_title=title, - model_names=_early_model_names, - )): - return None - - bubbles.sort(key=lambda b: b.get("timestamp") or 0) - - last_user_ts = None - for b in bubbles: - if b["type"] == "user": - last_user_ts = b.get("timestamp") - elif b["type"] == "ai" and last_user_ts is not None: - ts = b.get("timestamp") - if ts and ts > last_user_ts: - meta = b.setdefault("metadata", {}) - meta["responseTimeMs"] = ts - last_user_ts +def _aggregate_tab_metadata( + bubbles: list[DisplayBubble], + composer: Composer, +) -> dict[str, Any] | None: + """Aggregate bubble metadata into tab-level usage statistics.""" total_input = 0 total_output = 0 total_cached = 0 @@ -276,6 +269,8 @@ def _assemble_tab_from_composer_data( total_tool_calls = 0 total_thinking_ms = 0.0 models_set: set[str] = set() + max_ctx_tokens = 0 + ctx_token_limit = 0 for b in bubbles: m = b.get("metadata") or {} if m.get("inputTokens"): @@ -294,6 +289,10 @@ def _assemble_tab_from_composer_data( total_tool_calls += len(m["toolCalls"]) if m.get("thinkingDurationMs"): total_thinking_ms += m["thinkingDurationMs"] + if m.get("contextTokensUsed", 0) > max_ctx_tokens: + max_ctx_tokens = m["contextTokensUsed"] + if m.get("contextTokenLimit", 0) > ctx_token_limit: + ctx_token_limit = m["contextTokenLimit"] usage = composer.usage_data composer_cost = usage.get("cost") or usage.get("estimatedCost") @@ -305,21 +304,15 @@ def _assemble_tab_from_composer_data( files_added = composer.added_files files_removed = composer.removed_files - max_ctx_tokens = 0 - ctx_token_limit = 0 - for b in bubbles: - m = b.get("metadata") or {} - if m.get("contextTokensUsed", 0) > max_ctx_tokens: - max_ctx_tokens = m["contextTokensUsed"] - if m.get("contextTokenLimit", 0) > ctx_token_limit: - ctx_token_limit = m["contextTokenLimit"] - - tab_meta = None - has_any = any([total_input, total_output, total_cached, total_response_ms, - total_cost, models_set, total_tool_calls, total_thinking_ms, - lines_added, lines_removed, files_added, files_removed, - max_ctx_tokens]) - if has_any: + has_any = any([ + total_input, total_output, total_cached, total_response_ms, + total_cost, models_set, total_tool_calls, total_thinking_ms, + lines_added, lines_removed, files_added, files_removed, + max_ctx_tokens, ctx_token_limit, + ]) + if not has_any: + tab_meta: dict[str, Any] | None = None + else: tab_meta_raw = { "totalInputTokens": total_input or None, "totalOutputTokens": total_output or None, @@ -348,6 +341,18 @@ def _assemble_tab_from_composer_data( elif model_name_from_config not in models_used: models_used.insert(0, model_name_from_config) + return tab_meta + + +def _build_tab_dict( + composer_id: str, + title: str, + composer: Composer, + bubbles: list[DisplayBubble], + code_block_diffs: list[dict[str, Any]], + tab_meta: dict[str, Any] | None, +) -> dict[str, Any]: + """Assemble the final tab payload dict.""" tab: dict[str, Any] = { "id": composer_id, "title": title, @@ -365,6 +370,130 @@ def _assemble_tab_from_composer_data( return tab +def _iter_assigned_workspace_composers( + composer_rows: list[sqlite3.Row], + *, + parse_warnings: ParseWarningCollector, + project_layouts_map: dict[str, list[str]], + project_name_map: dict[str, str], + workspace_path_map: dict[str, str], + workspace_entries: list[dict[str, Any]], + composer_id_to_ws: dict[str, str], + invalid_workspace_ids: set[str], + invalid_workspace_aliases: dict[str, str], + matching_ws_ids: set[str], + bubble_map: Mapping[str, Bubble], +) -> Iterator[tuple[Composer, str]]: + """Yield composers assigned to the requested workspace, skipping parse failures.""" + for row in composer_rows: + composer = parse_composer_data_row( + row["key"], row["value"], parse_warnings=parse_warnings, + ) + if composer is None: + continue + composer_id = composer.composer_id + try: + assigned = assign_composer_workspace( + composer, + project_layouts_map=project_layouts_map, + project_name_map=project_name_map, + workspace_path_map=workspace_path_map, + workspace_entries=workspace_entries, + bubble_map=bubble_map, + composer_id_to_ws=composer_id_to_ws, + invalid_workspace_ids=invalid_workspace_ids, + invalid_workspace_aliases=invalid_workspace_aliases, + ) + if assigned not in matching_ws_ids: + continue + yield composer, composer_id + except _COMPOSER_ROW_ERRORS as e: + _record_composer_processing_failure( + composer_id, e, parse_warnings=parse_warnings, + ) + + +def _assemble_tab_from_composer_data( + composer_id: str, + composer: Composer, + bubble_map: Mapping[str, Bubble], + contexts: list[dict[str, Any]], + code_block_diffs: list[dict[str, Any]], + workspace_display_name: str, + rules: list[RuleTokens], + parse_warnings: ParseWarningCollector, +) -> dict[str, Any] | None: + """Assemble a single tab dict from a validated :class:`Composer`. + + Args: + composer_id: Composer UUID. + composer: Validated composer model (typed field access, not raw dict reads). + bubble_map: ``{bubble_id: Bubble}`` — global or scoped. + contexts: ``messageRequestContext`` entries for *this* composer + (list of dicts, each with an injected ``contextId`` key and a + ``bubbleId`` field from the JSON value). + code_block_diffs: ``codeBlockDiff`` entries for *this* composer. + workspace_display_name: Human-readable workspace name for rule matching. + rules: Exclusion rule token lists. + parse_warnings: Collector for skipped-bubble warnings. + + Returns: + A tab dict on success, or ``None`` when the tab should be omitted + (no renderable bubbles or excluded by rules). + """ + bubbles = _build_tab_display_bubbles( + composer_id, composer, bubble_map, contexts, + ) + if not bubbles: + return None + + title = _derive_composer_tab_title(composer_id, composer, bubbles) + + if is_excluded_by_rules(rules, build_searchable_text( + project_name=workspace_display_name, + chat_title=title, + model_names=composer_model_names(composer), + )): + return None + + bubbles.sort(key=lambda b: b.get("timestamp") or 0) + annotate_response_times(bubbles) + + tab_meta = _aggregate_tab_metadata(bubbles, composer) + return _build_tab_dict( + composer_id, title, composer, bubbles, code_block_diffs, tab_meta, + ) + + +def _safe_assemble_tab_from_composer_data( + composer_id: str, + composer: Composer, + bubble_map: Mapping[str, Bubble], + contexts: list[dict[str, Any]], + code_block_diffs: list[dict[str, Any]], + workspace_display_name: str, + rules: list[RuleTokens], + parse_warnings: ParseWarningCollector, +) -> dict[str, Any] | None: + """Like :func:`_assemble_tab_from_composer_data`, but skip+log on drift errors.""" + try: + return _assemble_tab_from_composer_data( + composer_id=composer_id, + composer=composer, + bubble_map=bubble_map, + contexts=contexts, + code_block_diffs=code_block_diffs, + workspace_display_name=workspace_display_name, + rules=rules, + parse_warnings=parse_warnings, + ) + except Exception as e: + _record_composer_processing_failure( + composer_id, e, parse_warnings=parse_warnings, + ) + return None + + def _build_matching_ws_ids( workspace_id: str, workspace_path: str, @@ -446,14 +575,12 @@ def _build_workspace_tab_summaries_uncached( with open_global_db(workspace_path) as (global_db, _): if global_db is None: - return {"error": "Global storage not found"}, 404 + return {"error": ERR_GLOBAL_STORAGE_NOT_FOUND}, 404 workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id) project_layouts_map = load_project_layouts_map(global_db) - # Full composerData roster still required for the summary tab loop; - # alias inference alone is fingerprint-cached across requests. composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL) ctx = with_invalid_workspace_aliases( @@ -466,61 +593,43 @@ def _build_workspace_tab_summaries_uncached( ) invalid_workspace_aliases = ctx.invalid_workspace_aliases or {} - for row in composer_rows: - composer = parse_composer_data_row( - row["key"], row["value"], parse_warnings=parse_warnings, - ) - if composer is None: + for composer, composer_id in _iter_assigned_workspace_composers( + composer_rows, + parse_warnings=parse_warnings, + project_layouts_map=project_layouts_map, + project_name_map=project_name_map, + workspace_path_map=workspace_path_map, + workspace_entries=workspace_entries, + composer_id_to_ws=composer_id_to_ws, + invalid_workspace_ids=invalid_workspace_ids, + invalid_workspace_aliases=invalid_workspace_aliases, + matching_ws_ids=matching_ws_ids, + bubble_map={}, + ): + if is_composer_excluded( + rules, + project_name=workspace_display_name, + composer=composer, + ): continue - composer_id = composer.composer_id - try: - assigned = assign_composer_workspace( - composer, - project_layouts_map=project_layouts_map, - project_name_map=project_name_map, - workspace_path_map=workspace_path_map, - workspace_entries=workspace_entries, - bubble_map={}, - composer_id_to_ws=composer_id_to_ws, - invalid_workspace_ids=invalid_workspace_ids, - invalid_workspace_aliases=invalid_workspace_aliases, - ) - - if assigned not in matching_ws_ids: - continue - if is_composer_excluded( - rules, - project_name=workspace_display_name, - composer=composer, - ): - continue + headers = composer.full_conversation_headers_only + title = composer_chat_title(composer) + + _early_model_names = composer_model_names(composer) + tab_meta: dict[str, Any] | None = None + if _early_model_names: + tab_meta = {"modelsUsed": _early_model_names} - headers = composer.full_conversation_headers_only - title = composer_chat_title(composer) - - _early_model_names = composer_model_names(composer) - tab_meta: dict[str, Any] | None = None - if _early_model_names: - tab_meta = {"modelsUsed": _early_model_names} - - tab_entry: dict[str, Any] = { - "id": composer_id, - "title": title, - "timestamp": _composer_tab_timestamp_ms(composer), - "messageCount": len(headers), - } - if tab_meta: - tab_entry["metadata"] = tab_meta - response["tabs"].append(tab_entry) - - except Exception as e: - _logger.warning( - "Failed to process Composer from composerData:%s: %s", - composer_id, - e, - ) - parse_warnings.record_composer_processing_failure() + tab_entry: dict[str, Any] = { + "id": composer_id, + "title": title, + "timestamp": _composer_tab_timestamp_ms(composer), + "messageCount": len(headers), + } + if tab_meta: + tab_entry["metadata"] = tab_meta + response["tabs"].append(tab_entry) response["tabs"].sort(key=lambda t: t.get("timestamp") or 0, reverse=True) return parse_warnings.attach_to(response), 200 @@ -567,7 +676,7 @@ def assemble_single_tab( with open_global_db(workspace_path) as (global_db, _): if global_db is None: - return {"error": "Global storage not found"}, 404 + return {"error": ERR_GLOBAL_STORAGE_NOT_FOUND}, 404 workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id) @@ -577,14 +686,14 @@ def assemble_single_tab( (f"composerData:{composer_id}",), ) if not rows: - return {"error": "Conversation not found"}, 404 + return {"error": ERR_CONVERSATION_NOT_FOUND}, 404 row = rows[0] composer = parse_composer_data_row( row["key"], row["value"], parse_warnings=parse_warnings, ) if composer is None: - return {"error": "Conversation not found"}, 404 + return {"error": ERR_CONVERSATION_NOT_FOUND}, 404 project_layouts_map: dict[str, list[str]] = {} project_layouts_map[composer_id] = load_project_layouts_for_composer( @@ -611,7 +720,7 @@ def assemble_single_tab( ) if assigned not in matching_ws_ids: - return {"error": "Conversation not found"}, 404 + return {"error": ERR_CONVERSATION_NOT_FOUND}, 404 contexts = load_message_request_context_for_composer(global_db, composer_id) code_block_diffs = load_code_block_diffs_for_composer(global_db, composer_id) @@ -628,7 +737,7 @@ def assemble_single_tab( ) if tab is None: - return {"error": "Conversation not found"}, 404 + return {"error": ERR_CONVERSATION_NOT_FOUND}, 404 response: dict[str, Any] = {"tab": tab} return parse_warnings.attach_to(response), 200 @@ -674,17 +783,14 @@ def assemble_workspace_tabs( with open_global_db(workspace_path) as (global_db, _): if global_db is None: - return {"error": "Global storage not found"}, 404 + return {"error": ERR_GLOBAL_STORAGE_NOT_FOUND}, 404 workspace_display_name = lookup_workspace_display_name(workspace_path, workspace_id) bubble_map = load_bubble_map(global_db, parse_warnings=parse_warnings) - # Load codeBlockDiffs code_block_diff_map = load_code_block_diff_map(global_db) - # Load messageRequestContext rows once; build both - # message_request_context_map and project_layouts_map from the same pass. project_layouts_map: dict[str, list[str]] = {} for row in safe_fetchall( global_db, @@ -721,7 +827,6 @@ def assemble_workspace_tabs( if isinstance(layout, dict) and layout.get("rootPath"): project_layouts_map[chat_id].append(layout["rootPath"]) - # Get composer data entries with conversations composer_rows = safe_fetchall(global_db, COMPOSER_ROWS_WITH_HEADERS_SQL) ctx = with_invalid_workspace_aliases( @@ -734,52 +839,32 @@ def assemble_workspace_tabs( ) invalid_workspace_aliases = ctx.invalid_workspace_aliases or {} - for row in composer_rows: - composer = parse_composer_data_row( - row["key"], row["value"], parse_warnings=parse_warnings, + for composer, composer_id in _iter_assigned_workspace_composers( + composer_rows, + parse_warnings=parse_warnings, + project_layouts_map=project_layouts_map, + project_name_map=project_name_map, + workspace_path_map=workspace_path_map, + workspace_entries=workspace_entries, + composer_id_to_ws=composer_id_to_ws, + invalid_workspace_ids=invalid_workspace_ids, + invalid_workspace_aliases=invalid_workspace_aliases, + matching_ws_ids=matching_ws_ids, + bubble_map={}, + ): + tab = _safe_assemble_tab_from_composer_data( + composer_id=composer_id, + composer=composer, + bubble_map=bubble_map, + contexts=message_request_context_map.get(composer_id, []), + code_block_diffs=code_block_diff_map.get(composer_id, []), + workspace_display_name=workspace_display_name, + rules=rules, + parse_warnings=parse_warnings, ) - if composer is None: - continue - composer_id = composer.composer_id - try: - assigned = assign_composer_workspace( - composer, - project_layouts_map=project_layouts_map, - project_name_map=project_name_map, - workspace_path_map=workspace_path_map, - workspace_entries=workspace_entries, - # Assignment matches summary path; bubble_map used only for tab body. - bubble_map={}, - composer_id_to_ws=composer_id_to_ws, - invalid_workspace_ids=invalid_workspace_ids, - invalid_workspace_aliases=invalid_workspace_aliases, - ) - - if assigned not in matching_ws_ids: - continue + if tab is not None: + response["tabs"].append(tab) - tab = _assemble_tab_from_composer_data( - composer_id=composer_id, - composer=composer, - bubble_map=bubble_map, - contexts=message_request_context_map.get(composer_id, []), - code_block_diffs=code_block_diff_map.get(composer_id, []), - workspace_display_name=workspace_display_name, - rules=rules, - parse_warnings=parse_warnings, - ) - if tab is not None: - response["tabs"].append(tab) - - except Exception as e: - _logger.warning( - "Failed to process Composer from composerData:%s: %s", - composer_id, - e, - ) - parse_warnings.record_composer_processing_failure() - - # Sort tabs by timestamp descending (newest first) response["tabs"].sort(key=lambda t: t.get("timestamp") or 0, reverse=True) return parse_warnings.attach_to(response), 200 diff --git a/tests/test_cursor_md_exporter.py b/tests/test_cursor_md_exporter.py index b7812e5..5186f33 100644 --- a/tests/test_cursor_md_exporter.py +++ b/tests/test_cursor_md_exporter.py @@ -18,7 +18,10 @@ if str(_root) not in sys.path: sys.path.insert(0, str(_root)) -from utils.cursor_md_exporter import cursor_cli_session_to_markdown +from utils.cursor_md_exporter import ( + _build_ide_session_summary, + cursor_cli_session_to_markdown, +) def _make_meta_hex(meta: dict) -> str: @@ -241,6 +244,43 @@ def test_title_quoting_with_special_chars(self): # Frontmatter title must have embedded quotes escaped as \" self.assertIn('title: "He said \\"hello\\""', result) + def test_non_dict_meta_json_falls_back_to_empty_mapping(self): + """List/scalar meta JSON must not crash; agentId falls back to parent dir name.""" + db_path = self._db_path("scalar-meta") + os.makedirs(os.path.dirname(db_path), exist_ok=True) + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)") + conn.execute("CREATE TABLE blobs (id TEXT PRIMARY KEY, data BLOB)") + conn.execute( + "INSERT INTO meta VALUES ('0', ?)", + (json.dumps(["not", "a", "dict"]).encode("utf-8").hex(),), + ) + conn.commit() + conn.close() + + result = cursor_cli_session_to_markdown(db_path) + self.assertIn(f'log_id: "{Path(db_path).parent.name}"', result) + self.assertIn("log_type: cli_agent", result) + + def test_web_only_tool_stats_produce_session_summary(self): + """Search/web-only sessions still get a Tool Results summary block.""" + summary = _build_ide_session_summary( + [], + [], + [], + { + "terminal_success": 0, + "terminal_error": 0, + "file_reads": 0, + "file_edits": 0, + "searches": 0, + "web": 2, + }, + ) + self.assertIn("## Session Summary", summary) + self.assertIn("### Tool Results", summary) + self.assertIn("Web Fetches: 2", summary) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_workspace_tabs_aggregate_metadata.py b/tests/test_workspace_tabs_aggregate_metadata.py new file mode 100644 index 0000000..e216f18 --- /dev/null +++ b/tests/test_workspace_tabs_aggregate_metadata.py @@ -0,0 +1,32 @@ +"""Regression tests for tab-level metadata aggregation edge cases.""" + +from __future__ import annotations + +import unittest + +from models import Composer +from services.workspace_tabs import _aggregate_tab_metadata + + +class TestAggregateTabMetadata(unittest.TestCase): + def test_context_token_limit_only_produces_tab_metadata(self): + """A positive contextTokenLimit alone must not be dropped by has_any.""" + bubbles = [{ + "type": "user", + "text": "hello", + "timestamp": 1, + "metadata": {"contextTokenLimit": 128_000}, + }] + composer = Composer( + composer_id="composer-limit-only", + full_conversation_headers_only=[], + created_at=1, + ) + + tab_meta = _aggregate_tab_metadata(bubbles, composer) + + self.assertEqual(tab_meta, {"contextTokenLimit": 128_000}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workspace_tabs_malformed_nested.py b/tests/test_workspace_tabs_malformed_nested.py index ef0ff3a..1f12021 100644 --- a/tests/test_workspace_tabs_malformed_nested.py +++ b/tests/test_workspace_tabs_malformed_nested.py @@ -213,5 +213,84 @@ def test_non_dict_parse_result_does_not_drop_composer(self) -> None: self.assertIn("cmp-t", ids) +def _seed_workspace_with_string_token_counts(parent: str) -> str: + ws_root = os.path.join(parent, "workspaceStorage") + global_root = os.path.join(parent, "globalStorage") + os.makedirs(ws_root, exist_ok=True) + os.makedirs(global_root, exist_ok=True) + ws_dir = os.path.join(ws_root, "ws-a") + os.makedirs(ws_dir, exist_ok=True) + with open(os.path.join(ws_dir, "workspace.json"), "w") as f: + json.dump({"folder": "/tmp/proj"}, f) + sqlite3.connect(os.path.join(ws_dir, "state.vscdb")).close() + + conn = sqlite3.connect(os.path.join(global_root, "state.vscdb")) + conn.execute("CREATE TABLE cursorDiskKV ([key] TEXT PRIMARY KEY, value TEXT)") + conn.execute( + "INSERT INTO cursorDiskKV VALUES (?, ?)", + ( + "composerData:cmp-bad-tokens", + json.dumps({ + "name": "Tab with string token counts", + "createdAt": 1_715_000_000_000, + "lastUpdatedAt": 1_715_000_500_000, + "fullConversationHeadersOnly": [{"bubbleId": "b-bad", "type": 2}], + }), + ), + ) + conn.execute( + "INSERT INTO cursorDiskKV VALUES (?, ?)", + ( + "bubbleId:cmp-bad-tokens:b-bad", + json.dumps({ + "text": "assistant with drifted tokens", + "tokenCount": {"inputTokens": "1500", "outputTokens": "500"}, + }), + ), + ) + conn.execute( + "INSERT INTO cursorDiskKV VALUES (?, ?)", + ( + "composerData:cmp-good", + json.dumps({ + "name": "Healthy tab", + "createdAt": 1_715_000_000_000, + "lastUpdatedAt": 1_715_000_600_000, + "fullConversationHeadersOnly": [{"bubbleId": "b-good", "type": 1}], + }), + ), + ) + conn.execute( + "INSERT INTO cursorDiskKV VALUES (?, ?)", + ("bubbleId:cmp-good:b-good", json.dumps({"text": "hello"})), + ) + conn.commit() + conn.close() + return ws_root + + +class TestStringTokenCountsDoNot500Listing(unittest.TestCase): + def test_drifted_token_metadata_skips_composer_not_whole_listing(self) -> None: + app = Flask(__name__) + app.config["TESTING"] = True + app.config["EXCLUSION_RULES"] = [] + + with tempfile.TemporaryDirectory() as tmp: + ws_root = _seed_workspace_with_string_token_counts(tmp) + with self.assertLogs("services.workspace_tabs", level="WARNING") as cm: + with app.test_request_context("/api/workspaces/global/tabs"): + payload, status = assemble_workspace_tabs("global", ws_root, rules=[]) + + self.assertEqual(status, 200) + ids = [t["id"] for t in payload.get("tabs", [])] + self.assertNotIn("cmp-bad-tokens", ids) + self.assertIn("cmp-good", ids) + messages = [r.getMessage() for r in cm.records] + self.assertTrue( + any("cmp-bad-tokens" in m for m in messages), + f"expected composer processing warning for cmp-bad-tokens, got: {messages}", + ) + + if __name__ == "__main__": unittest.main() diff --git a/utils/cursor_md_exporter.py b/utils/cursor_md_exporter.py index d85eb38..ca9faa7 100644 --- a/utils/cursor_md_exporter.py +++ b/utils/cursor_md_exporter.py @@ -17,6 +17,7 @@ from __future__ import annotations import json +import sqlite3 from datetime import datetime from pathlib import Path from typing import Any @@ -34,6 +35,54 @@ from utils.text_extract import slug +_CLI_META_READ_ERRORS = ( + sqlite3.Error, + json.JSONDecodeError, + IndexError, + TypeError, + ValueError, + UnicodeDecodeError, +) + + +def _render_tool_call_md(tc: dict[str, Any], *, include_status: bool = False) -> str: + """Render one tool-call block as markdown blockquote lines.""" + summary = tc.get("summary") or tc.get("name") or "unknown" + status_str = "" + if include_status: + tool_status = tc.get("status") or "" + if tool_status: + status_str = f" ({tool_status})" + lines = [f"> **Tool: {summary}**{status_str}"] + if tc.get("input"): + lines.append("> **INPUT:**\n> ```") + for iline in str(tc["input"]).split("\n"): + lines.append(f"> {iline}") + lines.append("> ```") + if tc.get("output"): + lines.append("> **OUTPUT:**\n> ```") + for oline in str(tc["output"]).split("\n"): + lines.append(f"> {oline}") + lines.append("> ```") + lines.append("") + return "\n".join(lines) + "\n" + + +def _render_cli_export_body(bubbles: list[DisplayBubble]) -> str: + """Render conversation body for CLI session export.""" + body_parts: list[str] = [] + for b in bubbles: + role_label = "User" if b["type"] == "user" else "Assistant" + body_parts.append(f"### {role_label}\n\n") + body_parts.append(b.get("text", "") + "\n\n") + tool_calls = (b.get("metadata") or {}).get("toolCalls") or [] + for tc in tool_calls: + if isinstance(tc, dict): + body_parts.append(_render_tool_call_md(tc)) + body_parts.append("---\n\n") + return "".join(body_parts) + + # ── CLI session exporter ───────────────────────────────────────────────────── @@ -81,7 +130,6 @@ def cursor_cli_session_to_markdown( # Read metadata from the database if not provided. if session_meta is None: - import sqlite3 from contextlib import closing try: # `closing(...)` guarantees .close() on scope exit (including on @@ -89,8 +137,9 @@ def cursor_cli_session_to_markdown( # commit/rollback, not close. See issue #17. with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as conn: row = conn.execute("SELECT value FROM meta WHERE key = '0'").fetchone() - session_meta = json.loads(bytes.fromhex(row[0]).decode()) if row else {} - except Exception: + decoded = json.loads(bytes.fromhex(row[0]).decode()) if row else {} + session_meta = decoded if isinstance(decoded, dict) else {} + except _CLI_META_READ_ERRORS: session_meta = {} session_id: str = session_meta.get("agentId", db_path.parent.name) @@ -169,80 +218,19 @@ def cursor_cli_session_to_markdown( header_meta_parts.append(f"Tool calls: {total_tool_calls}") header = f"# {title}\n\n_{' | '.join(header_meta_parts)}_\n\n---\n\n" - # Body. - body = "" - for b in bubbles: - role_label = "User" if b["type"] == "user" else "Assistant" - body += f"### {role_label}\n\n" - body += b.get("text", "") + "\n\n" - tool_calls = (b.get("metadata") or {}).get("toolCalls") or [] - for tc in tool_calls: - summary = tc.get("summary") or tc.get("name") or "unknown" - body += f"> **Tool: {summary}**\n" - if tc.get("input"): - body += "> **INPUT:**\n> ```\n" - for iline in str(tc["input"]).split("\n"): - body += f"> {iline}\n" - body += "> ```\n" - if tc.get("output"): - body += "> **OUTPUT:**\n> ```\n" - for oline in str(tc["output"]).split("\n"): - body += f"> {oline}\n" - body += "> ```\n" - body += "\n" - body += "---\n\n" - - return fm_str + header + body + return fm_str + header + _render_cli_export_body(bubbles) # ── IDE chat exporter ──────────────────────────────────────────────────────── -def cursor_ide_chat_to_markdown( +def _build_ide_export_bubbles( composer_data: dict[str, Any], - composer_id: str, bubble_map: dict[str, Bubble], - code_block_diff_map: dict[str, Any] | None = None, - workspace_info: dict[str, Any] | None = None, -) -> str: - """Generate a complete Markdown document from a Cursor IDE composer session. - - Parameters - ---------- - composer_data: - Parsed value of a ``composerData:`` KV entry from global storage. - composer_id: - The composer UUID — used as ``log_id`` in frontmatter and as the key - into ``code_block_diff_map``. - bubble_map: - Global ``{bubble_id: Bubble}`` map from - :func:`services.workspace_db.load_bubble_map`. - code_block_diff_map: - Optional ``{composer_id: [diff_dict]}`` map. When ``None`` no code - edit bubbles are appended. - workspace_info: - Optional dict with workspace display fields. Recognised keys: - ``ws_slug`` (str), ``ws_display_name`` (str). - - Returns - ------- - str - Full Markdown text including YAML frontmatter and conversation body. - """ - cd = composer_data - ws_info = workspace_info or {} - ws_slug = ws_info.get("ws_slug", "other-chats") - ws_display_name = ws_info.get("ws_display_name", "Other chats") - diffs = (code_block_diff_map or {}).get(composer_id, []) - - title = cd.get("name") or f"Chat {composer_id[:8]}" - model_config = cd.get("modelConfig") or {} - model_name = model_config.get("modelName") - updated_at = to_epoch_ms(cd.get("lastUpdatedAt")) or to_epoch_ms(cd.get("createdAt")) or 0 - created_ms = to_epoch_ms(cd.get("createdAt")) or updated_at or int(datetime.now().timestamp() * 1000) - headers = cd.get("fullConversationHeadersOnly") or [] - - # ── Build bubble list ───────────────────────────────────────────────────── + diffs: list[Any], +) -> list[DisplayBubble]: + """Build sorted display bubbles for IDE markdown export.""" + headers = composer_data.get("fullConversationHeadersOnly") or [] bubbles: list[DisplayBubble] = [] for h in headers: storage = bubble_map.get(h.get("bubbleId")) @@ -253,8 +241,11 @@ def cursor_ide_chat_to_markdown( if entry is not None: bubbles.append(entry) - # Append code-block diffs as synthetic AI bubbles. - diff_ts = to_epoch_ms(cd.get("lastUpdatedAt")) or to_epoch_ms(cd.get("createdAt")) or int(datetime.now().timestamp() * 1000) + diff_ts = ( + to_epoch_ms(composer_data.get("lastUpdatedAt")) + or to_epoch_ms(composer_data.get("createdAt")) + or int(datetime.now().timestamp() * 1000) + ) for d in diffs: bubbles.append({ "type": "ai", @@ -264,8 +255,14 @@ def cursor_ide_chat_to_markdown( bubbles.sort(key=lambda bub: bub.get("timestamp") or 0) annotate_response_times(bubbles) + return bubbles + - # ── Session-level aggregates ────────────────────────────────────────────── +def _compute_ide_session_aggregates( + bubbles: list[DisplayBubble], + composer_data: dict[str, Any], +) -> dict[str, Any]: + """Session-level stats for IDE export frontmatter and header.""" total_response_ms = sum( display_bubble_metadata(bub).get("responseTimeMs") or 0 for bub in bubbles ) @@ -279,8 +276,6 @@ def cursor_ide_chat_to_markdown( ctx_limit = max( (display_bubble_metadata(bub).get("contextTokenLimit") or 0) for bub in bubbles ) if bubbles else 0 - lines_added = cd.get("totalLinesAdded", 0) - lines_removed = cd.get("totalLinesRemoved", 0) tool_breakdown: dict[str, int] = {} for bub in bubbles: @@ -291,7 +286,26 @@ def cursor_ide_chat_to_markdown( ts_vals = [bub["timestamp"] for bub in bubbles if bub.get("timestamp")] wall_clock_sec = int((max(ts_vals) - min(ts_vals)) / 1000) if len(ts_vals) >= 2 else None - # ── File / command activity ─────────────────────────────────────────────── + return { + "total_response_ms": total_response_ms, + "total_thinking_ms": total_thinking_ms, + "total_tool_calls": total_tool_calls, + "max_ctx_used": max_ctx_used, + "ctx_limit": ctx_limit, + "lines_added": composer_data.get("totalLinesAdded", 0), + "lines_removed": composer_data.get("totalLinesRemoved", 0), + "tool_breakdown": tool_breakdown, + "wall_clock_sec": wall_clock_sec, + "thinking_count": sum( + 1 for bub in bubbles if display_bubble_metadata(bub).get("thinking") + ), + } + + +def _scan_ide_tool_activity( + bubbles: list[DisplayBubble], +) -> tuple[list[str], list[str], list[str], dict[str, int]]: + """Collect file/command activity lists and tool-result counters.""" files_read_list: list[str] = [] files_written_list: list[str] = [] commands_run_list: list[str] = [] @@ -322,8 +336,25 @@ def cursor_ide_chat_to_markdown( tool_result_stats["searches"] += 1 elif tn in ("web_search", "web_fetch"): tool_result_stats["web"] += 1 + return files_read_list, files_written_list, commands_run_list, tool_result_stats + - # ── Frontmatter ─────────────────────────────────────────────────────────── +def _build_ide_frontmatter( + *, + composer_id: str, + title: str, + created_ms: int, + updated_at: int, + ws_slug: str, + ws_display_name: str, + model_name: str | None, + bubbles: list[DisplayBubble], + aggregates: dict[str, Any], + files_read_list: list[str], + files_written_list: list[str], + commands_run_list: list[str], +) -> str: + """YAML frontmatter block for IDE export.""" fm_lines = ["---"] fm_lines.append(f"log_id: {json.dumps(composer_id, ensure_ascii=False)}") fm_lines.append("log_type: chat") @@ -337,26 +368,30 @@ def cursor_ide_chat_to_markdown( if model_name and model_name != "default": fm_lines.append(f"model: {json.dumps(model_name, ensure_ascii=False)}") fm_lines.append(f"message_count: {len(bubbles)}") + total_tool_calls = aggregates["total_tool_calls"] if total_tool_calls: fm_lines.append(f"total_tool_calls: {total_tool_calls}") + tool_breakdown = aggregates["tool_breakdown"] if tool_breakdown: fm_lines.append("tool_call_breakdown:") for tn, cnt in sorted(tool_breakdown.items(), key=lambda x: -x[1]): fm_lines.append(f" {json.dumps(tn, ensure_ascii=False)}: {cnt}") - total_think = sum( - 1 for bub in bubbles if display_bubble_metadata(bub).get("thinking") - ) - if total_think: - fm_lines.append(f"thinking_count: {total_think}") + if aggregates["thinking_count"]: + fm_lines.append(f"thinking_count: {aggregates['thinking_count']}") + wall_clock_sec = aggregates["wall_clock_sec"] if wall_clock_sec is not None: fm_lines.append(f"wall_clock_seconds: {wall_clock_sec}") - if total_response_ms: - fm_lines.append(f"total_response_time_sec: {total_response_ms / 1000:.1f}") - if total_thinking_ms: - fm_lines.append(f"total_thinking_time_sec: {total_thinking_ms / 1000:.1f}") + if aggregates["total_response_ms"]: + fm_lines.append(f"total_response_time_sec: {aggregates['total_response_ms'] / 1000:.1f}") + if aggregates["total_thinking_ms"]: + fm_lines.append(f"total_thinking_time_sec: {aggregates['total_thinking_ms'] / 1000:.1f}") + max_ctx_used = aggregates["max_ctx_used"] + ctx_limit = aggregates["ctx_limit"] if max_ctx_used and ctx_limit: fm_lines.append(f"max_context_tokens_used: {max_ctx_used}") fm_lines.append(f"context_token_limit: {ctx_limit}") + lines_added = aggregates["lines_added"] + lines_removed = aggregates["lines_removed"] if lines_added or lines_removed: fm_lines.append(f"lines_added: {lines_added}") fm_lines.append(f"lines_removed: {lines_removed}") @@ -366,62 +401,86 @@ def cursor_ide_chat_to_markdown( if commands_run_list: fm_lines.append(f"commands_run: {len(commands_run_list)}") fm_lines.append("---") - fm_str = "\n".join(fm_lines) + "\n\n" + return "\n".join(fm_lines) + "\n\n" - # ── Document header ─────────────────────────────────────────────────────── + +def _build_ide_document_header( + title: str, + *, + created_ms: int, + model_name: str | None, + aggregates: dict[str, Any], +) -> str: + """Title and metadata line for IDE export.""" header = f"# {title}\n\n" meta_parts: list[str] = [] if created_ms: - meta_parts.append(f"Created: {datetime.fromtimestamp(created_ms / 1000).strftime('%Y-%m-%d %H:%M:%S')}") + meta_parts.append( + f"Created: {datetime.fromtimestamp(created_ms / 1000).strftime('%Y-%m-%d %H:%M:%S')}" + ) if model_name and model_name != "default": meta_parts.append(f"Model: {model_name}") + total_tool_calls = aggregates["total_tool_calls"] if total_tool_calls: meta_parts.append(f"Tool calls: {total_tool_calls}") + wall_clock_sec = aggregates["wall_clock_sec"] if wall_clock_sec is not None: hrs, rem = divmod(wall_clock_sec, 3600) mins, secs = divmod(rem, 60) dur = f"{hrs}h {mins}m" if hrs else (f"{mins}m {secs}s" if mins else f"{secs}s") meta_parts.append(f"Duration: {dur}") header += f"_{' | '.join(meta_parts)}_\n\n---\n\n" if meta_parts else "---\n\n" + return header - # ── Session summary block ───────────────────────────────────────────────── - summary = "" - if files_read_list or files_written_list or commands_run_list: - summary += "## Session Summary\n\n" - if files_written_list or files_read_list: - summary += "### Files Touched\n\n" - summary += "| Action | File |\n|--------|------|\n" - for fp in files_written_list: - summary += f"| Edit | `{fp}` |\n" - for fp in files_read_list: - summary += f"| Read | `{fp}` |\n" - summary += "\n" - if commands_run_list: - summary += "### Commands Run\n\n" - for i, cmd in enumerate(commands_run_list, 1): - summary += f"{i}. `{cmd}`\n" - summary += "\n" - non_zero = {k: v for k, v in tool_result_stats.items() if v > 0} - if non_zero: - summary += "### Tool Results\n\n" - labels = { - "terminal_success": "Terminal Success", - "terminal_error": "Terminal Error", - "file_reads": "File Reads", - "file_edits": "File Edits", - "searches": "Searches", - "web": "Web Fetches", - } - for k, v in non_zero.items(): - summary += f"- {labels.get(k, k)}: {v}\n" - summary += "\n" - summary += "---\n\n" - - # ── Body ────────────────────────────────────────────────────────────────── - body = "" + +def _build_ide_session_summary( + files_read_list: list[str], + files_written_list: list[str], + commands_run_list: list[str], + tool_result_stats: dict[str, int], +) -> str: + """Optional session summary section for IDE export.""" + has_tool_stats = any(v > 0 for v in tool_result_stats.values()) + if not (files_read_list or files_written_list or commands_run_list or has_tool_stats): + return "" + parts: list[str] = ["## Session Summary\n\n"] + if files_written_list or files_read_list: + parts.append("### Files Touched\n\n") + parts.append("| Action | File |\n|--------|------|\n") + for fp in files_written_list: + parts.append(f"| Edit | `{fp}` |\n") + for fp in files_read_list: + parts.append(f"| Read | `{fp}` |\n") + parts.append("\n") + if commands_run_list: + parts.append("### Commands Run\n\n") + for i, cmd in enumerate(commands_run_list, 1): + parts.append(f"{i}. `{cmd}`\n") + parts.append("\n") + non_zero = {k: v for k, v in tool_result_stats.items() if v > 0} + if non_zero: + parts.append("### Tool Results\n\n") + labels = { + "terminal_success": "Terminal Success", + "terminal_error": "Terminal Error", + "file_reads": "File Reads", + "file_edits": "File Edits", + "searches": "Searches", + "web": "Web Fetches", + } + for k, v in non_zero.items(): + parts.append(f"- {labels.get(k, k)}: {v}\n") + parts.append("\n") + parts.append("---\n\n") + return "".join(parts) + + +def _render_ide_export_body(bubbles: list[DisplayBubble]) -> str: + """Render conversation body for IDE chat export.""" + body_parts: list[str] = [] for bub in bubbles: role_label = "User" if bub["type"] == "user" else "Assistant" - body += f"### {role_label}\n\n" + body_parts.append(f"### {role_label}\n\n") meta = display_bubble_metadata(bub) bub_meta: list[str] = [] if meta.get("modelName"): @@ -444,30 +503,96 @@ def cursor_ide_chat_to_markdown( remaining = meta["contextWindowPercent"] bub_meta.append(f"Context: {remaining}% remaining") if bub_meta: - body += f"_{' | '.join(bub_meta)}_\n\n" + body_parts.append(f"_{' | '.join(bub_meta)}_\n\n") if bub.get("timestamp"): - body += f"_{datetime.fromtimestamp(bub['timestamp'] / 1000).isoformat()}_\n\n" + body_parts.append( + f"_{datetime.fromtimestamp(bub['timestamp'] / 1000).isoformat()}_\n\n" + ) thinking_text = meta.get("thinking") if thinking_text: dur_str = f" ({thinking_ms / 1000:.1f}s)" if thinking_ms else "" - body += f"
Thinking{dur_str}\n\n{thinking_text}\n\n
\n\n" - body += bub["text"] + "\n\n" + body_parts.append( + f"
Thinking{dur_str}\n\n" + f"{thinking_text}\n\n
\n\n" + ) + body_parts.append(bub["text"] + "\n\n") for t in display_bubble_tool_calls(bub): - tool_summary = t.get("summary") or t.get("name") or "unknown" - tool_status = t.get("status") or "" - status_str = f" ({tool_status})" if tool_status else "" - body += f"> **Tool: {tool_summary}**{status_str}\n" - if t.get("input"): - body += "> **INPUT:**\n> ```\n" - for iline in str(t["input"]).split("\n"): - body += f"> {iline}\n" - body += "> ```\n" - if t.get("output"): - body += "> **OUTPUT:**\n> ```\n" - for oline in str(t["output"]).split("\n"): - body += f"> {oline}\n" - body += "> ```\n" - body += "\n" - body += "---\n\n" + body_parts.append(_render_tool_call_md(t, include_status=True)) + body_parts.append("---\n\n") + return "".join(body_parts) + + +def cursor_ide_chat_to_markdown( + composer_data: dict[str, Any], + composer_id: str, + bubble_map: dict[str, Bubble], + code_block_diff_map: dict[str, Any] | None = None, + workspace_info: dict[str, Any] | None = None, +) -> str: + """Generate a complete Markdown document from a Cursor IDE composer session. + + Parameters + ---------- + composer_data: + Parsed value of a ``composerData:`` KV entry from global storage. + composer_id: + The composer UUID — used as ``log_id`` in frontmatter and as the key + into ``code_block_diff_map``. + bubble_map: + Global ``{bubble_id: Bubble}`` map from + :func:`services.workspace_db.load_bubble_map`. + code_block_diff_map: + Optional ``{composer_id: [diff_dict]}`` map. When ``None`` no code + edit bubbles are appended. + workspace_info: + Optional dict with workspace display fields. Recognised keys: + ``ws_slug`` (str), ``ws_display_name`` (str). + + Returns + ------- + str + Full Markdown text including YAML frontmatter and conversation body. + """ + cd = composer_data + ws_info = workspace_info or {} + ws_slug = ws_info.get("ws_slug", "other-chats") + ws_display_name = ws_info.get("ws_display_name", "Other chats") + diffs = (code_block_diff_map or {}).get(composer_id, []) + title = cd.get("name") or f"Chat {composer_id[:8]}" + model_config = cd.get("modelConfig") or {} + model_name = model_config.get("modelName") + updated_at = to_epoch_ms(cd.get("lastUpdatedAt")) or to_epoch_ms(cd.get("createdAt")) or 0 + created_ms = to_epoch_ms(cd.get("createdAt")) or updated_at or int(datetime.now().timestamp() * 1000) + + bubbles = _build_ide_export_bubbles(cd, bubble_map, diffs) + aggregates = _compute_ide_session_aggregates(bubbles, cd) + files_read_list, files_written_list, commands_run_list, tool_result_stats = ( + _scan_ide_tool_activity(bubbles) + ) + + fm_str = _build_ide_frontmatter( + composer_id=composer_id, + title=title, + created_ms=created_ms, + updated_at=updated_at, + ws_slug=ws_slug, + ws_display_name=ws_display_name, + model_name=model_name, + bubbles=bubbles, + aggregates=aggregates, + files_read_list=files_read_list, + files_written_list=files_written_list, + commands_run_list=commands_run_list, + ) + header = _build_ide_document_header( + title, + created_ms=created_ms, + model_name=model_name, + aggregates=aggregates, + ) + summary = _build_ide_session_summary( + files_read_list, files_written_list, commands_run_list, tool_result_stats, + ) + body = _render_ide_export_body(bubbles) return fm_str + header + summary + body diff --git a/utils/display_bubble.py b/utils/display_bubble.py index 9c70a50..d79fd53 100644 --- a/utils/display_bubble.py +++ b/utils/display_bubble.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, cast +from typing import Any from models import Bubble, DisplayBubble from models.bubble_display import BubbleMetadata, BubbleRole @@ -37,7 +37,7 @@ def extract_thinking_text( def build_storage_bubble_metadata( bubble: Bubble, role: BubbleRole, -) -> dict[str, Any] | None: +) -> BubbleMetadata | None: """Metadata dict for tabs/export — tool calls, tokens, thinking, context.""" model_info = bubble.model_info model_name = model_info.get("modelName") @@ -52,7 +52,7 @@ def build_storage_bubble_metadata( elif ctx_window.get("percentageRemaining") is not None: ctx_pct = ctx_window.get("percentageRemaining") - meta: dict[str, Any] = {} + meta: BubbleMetadata = {} if model_name: meta["modelName"] = model_name if ctx_pct is not None: @@ -126,7 +126,7 @@ def build_display_bubble_from_storage( } metadata = build_storage_bubble_metadata(bubble, role) if metadata: - entry["metadata"] = cast(BubbleMetadata, metadata) + entry["metadata"] = metadata return entry @@ -155,6 +155,6 @@ def annotate_response_times(bubbles: list[DisplayBubble]) -> None: continue bts = bub.get("timestamp") if bts and bts > last_user_ts: - meta = dict(display_bubble_metadata(bub)) + meta: BubbleMetadata = {**display_bubble_metadata(bub)} meta["responseTimeMs"] = bts - last_user_ts - bub["metadata"] = cast(BubbleMetadata, meta) + bub["metadata"] = meta