from __future__ import annotations from typing import Any from .config import BACKEND_MODE from .csharp_sidecar import call_sidecar_action def _require_hybrid_mode() -> None: if BACKEND_MODE != "csharp-hybrid": raise RuntimeError("Fragments backend requires JOURNAL_BACKEND_MODE=csharp-hybrid.") def list_fragments() -> list[dict[str, Any]]: _require_hybrid_mode() result = call_sidecar_action("fragments.list") if not isinstance(result, list): return [] return [item for item in result if isinstance(item, dict)] def create_fragment( type_: str, description: str, tags: list[str] | None = None, time: str | None = None, ) -> dict[str, Any]: _require_hybrid_mode() payload: dict[str, Any] = { "type": type_, "description": description, } if time: payload["time"] = time if tags: payload["tags"] = tags result = call_sidecar_action("fragments.create", payload=payload) return result if isinstance(result, dict) else {"result": result} def get_fragment(fragment_id: str) -> dict[str, Any] | None: _require_hybrid_mode() result = call_sidecar_action( "fragments.get", command_fields={"id": fragment_id}, ) return result if isinstance(result, dict) else None def update_fragment( fragment_id: str, *, type_: str | None = None, description: str | None = None, time: str | None = None, tags: list[str] | None = None, clear_tags: bool = False, ) -> bool: _require_hybrid_mode() payload: dict[str, Any] = {} if type_ is not None: payload["type"] = type_ if description is not None: payload["description"] = description if time is not None: payload["time"] = time if clear_tags: payload["tags"] = [] elif tags is not None: payload["tags"] = tags if not payload: raise ValueError( "No update fields provided. Use --type/--description/--time/--tag/--clear-tags." ) result = call_sidecar_action( "fragments.update", payload=payload, command_fields={"id": fragment_id}, ) return bool(result) def delete_fragment(fragment_id: str) -> bool: _require_hybrid_mode() result = call_sidecar_action( "fragments.delete", command_fields={"id": fragment_id}, ) return bool(result) def search_fragments(type_: str | None = None, tag: str | None = None) -> list[dict[str, Any]]: _require_hybrid_mode() result = call_sidecar_action( "fragments.search", command_fields={ "type": type_, "tag": tag, }, ) if not isinstance(result, list): return [] return [item for item in result if isinstance(item, dict)]