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 _result_identity(doc, source: str) -> dict:
    """Identifying fields shared by the UI snippet and the model's full-text view."""
    from blueprints.rag.retrieval import citation_url

    meta = doc.metadata or {}
    result = {}
    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
    return result


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 = [
        {"snippet": snippet(doc.page_content), **_result_identity(doc, source)}
        for doc in docs
    ]
    return make_artifact(
        "search.documents",
        f"Search: {query}",
        {"query": query, "source": source, "results": results},
    )


def search_results_model_context(docs, source: str) -> dict:
    """Full chunk text for the model, keyed like `search_documents_artifact`'s results.

    Set as a tool's `artifact["model_context"]` so that under compaction (see
    compact_tool_result in blueprints/conversation/context.py) the model still
    sees the full chunk text it retrieved, not the UI's 400-char snippets.
    """
    return {
        "results": [
            {**_result_identity(doc, source), "content": doc.page_content}
            for doc in docs
        ]
    }