simbarag/blueprints/conversation/artifacts_util.py

"""Helpers for building ToolArtifact dicts returned by agent tools.

See docs/cowork-ui-plan.md ("Artifacts") for the contract. Artifacts must be
plain JSON-serialisable dicts: they ride along in ToolMessage.artifact and are
stored in the LangGraph checkpoint.
"""

import os
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from urllib.parse import quote, urlencode

SNIPPET_MAX_CHARS = 400
NOTE_MAX_CHARS = 50_000
ELLIPSIS = "…"


def make_artifact(
    kind: str, title: str, data: dict, source_ref: dict | None = None
) -> dict:
    artifact = {"kind": kind, "title": title, "data": data}
    if source_ref:
        artifact["source_ref"] = source_ref
    return artifact


def truncate(text: str, limit: int) -> str:
    """Cap text at `limit` chars, ending in an ellipsis when cut."""
    if len(text) <= limit:
        return text
    return text[: limit - len(ELLIPSIS)].rstrip() + ELLIPSIS


def snippet(text: str, limit: int = SNIPPET_MAX_CHARS) -> str:
    """Collapse whitespace and cap at the snippet limit."""
    return truncate(" ".join((text or "").split()), limit)


def json_safe(value):
    """Convert a value (e.g. parsed YAML frontmatter) to JSON-safe types."""
    if isinstance(value, (str, int, float, bool, type(None))):
        return value
    if isinstance(value, Decimal):
        return float(value)
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, (list, tuple, set)):
        return [json_safe(v) for v in value]
    if isinstance(value, dict):
        return {str(k): json_safe(v) for k, v in value.items()}
    return str(value)


def unique(items) -> list:
    """Deduplicate while keeping first-seen order."""
    return list(dict.fromkeys(items))


def _parse_date(value: str) -> date | None:
    try:
        return datetime.strptime(value, "%Y-%m-%d").date()
    except (TypeError, ValueError):
        return None


def format_day(value: str) -> str:
    """'2026-10-12' -> 'Oct 12'. Unparseable input is returned unchanged."""
    d = _parse_date(value)
    return f"{d:%b} {d.day}" if d else str(value)


def format_month(value: str) -> str:
    """'2026-10' or '2026-10-01' -> 'October 2026'. Unparseable input is returned unchanged."""
    d = _parse_date(str(value)[:7] + "-01")
    return f"{d:%B %Y}" if d else str(value)


def format_date_range(start: str, end: str) -> str:
    """Human date range: 'Oct 12–19', 'Oct 28–Nov 3', 'Dec 28, 2026–Jan 3, 2027'."""
    s, e = _parse_date(start), _parse_date(end)
    if not s or not e:
        return f"{start}–{end}"
    if s == e:
        return f"{s:%b} {s.day}"
    if s.year != e.year:
        return f"{s:%b} {s.day}, {s.year}–{e:%b} {e.day}, {e.year}"
    if s.month == e.month:
        return f"{s:%b} {s.day}–{e.day}"
    return f"{s:%b} {s.day}–{e:%b} {e.day}"


def epoch_to_date(value) -> str | None:
    """Epoch seconds -> 'YYYY-MM-DD' (None if not a number)."""
    try:
        return datetime.fromtimestamp(float(value)).strftime("%Y-%m-%d")
    except (TypeError, ValueError, OverflowError, OSError):
        return None


def mealie_recipe_url(slug: str) -> str | None:
    """Recipe page URL for Mealie v2 (`/g/<group-slug>/r/<slug>`)."""
    base_url = os.getenv("MEALIE_BASE_URL", "").rstrip("/")
    if not base_url:
        return None
    group = os.getenv("MEALIE_GROUP_SLUG", "") or "home"
    return f"{base_url}/g/{quote(group)}/r/{quote(slug)}"


def mealie_source_ref(slug: str) -> dict:
    ref = {"system": "mealie", "id": slug}
    url = mealie_recipe_url(slug)
    if url:
        ref["url"] = url
    return ref


def obsidian_source_ref(path: str) -> dict:
    """Obsidian ref; the obsidian:// URL needs OBSIDIAN_VAULT_NAME."""
    ref = {"system": "obsidian", "path": path}
    vault = os.getenv("OBSIDIAN_VAULT_NAME", "")
    if vault:
        ref["url"] = "obsidian://open?" + urlencode(
            {"vault": vault, "file": path}, quote_via=quote
        )
    return ref


def note_title(path: str, frontmatter: dict | None = None) -> str:
    """Frontmatter `title` if present, else the filename without extension."""
    title = (frontmatter or {}).get("title")
    if isinstance(title, str) and title.strip():
        return title.strip()
    return Path(path).stem or path


def vault_relative_path(filepath: str) -> str:
    """Strip OBSIDIAN_VAULT_PATH from an absolute indexed filepath."""
    vault = os.getenv("OBSIDIAN_VAULT_PATH", "/app/data/obsidian")
    try:
        return str(Path(filepath).relative_to(vault))
    except ValueError:
        return filepath


def search_documents_artifact(query: str, source: str, docs) -> dict | None:
    """Build a `search.documents` artifact from vector-store Documents."""
    if not docs:
        return None
    results = []
    from blueprints.rag.retrieval import citation_url

    for doc in docs:
        meta = doc.metadata or {}
        result = {"snippet": snippet(doc.page_content)}
        if url := citation_url(doc):
            result["url"] = url
            result["source_id"] = str(doc.id)
        if headings := [meta[k] for k in ("h1", "h2", "h3") if meta.get(k)]:
            result["headings"] = headings
        if source == "obsidian":
            filepath = meta.get("filepath") or ""
            path = vault_relative_path(filepath) if filepath else ""
            result["title"] = note_title(path, meta) if path else "Untitled note"
            if path:
                result["path"] = path
            created = meta.get("created_at")
            if isinstance(created, str) and created:
                result["date"] = created[:10]
        else:
            filename = meta.get("filename") or ""
            result["title"] = Path(filename).stem or filename or "Document"
            if filename:
                result["filename"] = filename
            if meta.get("document_type"):
                result["document_type"] = str(meta["document_type"])
            created = epoch_to_date(meta.get("created_date"))
            if created:
                result["date"] = created
        results.append(result)
    return make_artifact(
        "search.documents",
        f"Search: {query}",
        {"query": query, "source": source, "results": results},
    )