strukto-ai--mirage/python/mirage/core/langfuse/readdir.py

# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========

from mirage.accessor.langfuse import LangfuseAccessor
from mirage.cache.index import IndexCacheStore, IndexEntry
from mirage.core.langfuse._client import (fetch_dataset_runs, fetch_datasets,
                                          fetch_prompts, fetch_sessions,
                                          fetch_traces)
from mirage.types import PathSpec
from mirage.utils.errors import enoent
from mirage.utils.key_prefix import mount_prefix_of

TOP_LEVEL_DIRS = ["traces", "sessions", "prompts", "datasets"]


async def readdir(
    accessor: LangfuseAccessor,
    path: PathSpec,
    index: IndexCacheStore = None,
) -> list[str]:
    """List directory contents.

    Args:
        accessor (LangfuseAccessor): langfuse accessor.
        path (PathSpec | str): resource-relative path.
        index (IndexCacheStore | None): index cache.
        prefix (str): mount prefix for virtual index keys.
    """
    if isinstance(path, str):
        path = PathSpec(virtual=path,
                        directory=path,
                        resource_path=path.strip("/"))
    virtual = path.virtual
    prefix = mount_prefix_of(path.virtual, path.resource_path)
    path = (path.dir if path.pattern else path).mount_path
    key = path.strip("/")

    if key and any(p.startswith(".") for p in key.split("/")):
        raise enoent(virtual)

    virtual_key = prefix + "/" + key if key else prefix or "/"

    if not key:
        return [f"{prefix}/{d}" for d in TOP_LEVEL_DIRS]

    parts = key.split("/")

    if parts[0] == "traces" and len(parts) == 1:
        return await _readdir_traces(accessor, virtual_key, index, prefix)

    if parts[0] == "sessions" and len(parts) == 1:
        return await _readdir_sessions(accessor, virtual_key, index, prefix)

    if parts[0] == "sessions" and len(parts) == 2:
        return await _readdir_session_traces(
            accessor,
            parts[1],
            virtual_key,
            index,
            prefix,
        )

    if parts[0] == "prompts" and len(parts) == 1:
        return await _readdir_prompts(accessor, virtual_key, index, prefix)

    if parts[0] == "prompts" and len(parts) == 2:
        return await _readdir_prompt_versions(
            accessor,
            parts[1],
            virtual_key,
            index,
            prefix,
        )

    if parts[0] == "datasets" and len(parts) == 1:
        return await _readdir_datasets(accessor, virtual_key, index, prefix)

    if parts[0] == "datasets" and len(parts) == 2:
        return [
            f"{prefix}/datasets/{parts[1]}/items.jsonl",
            f"{prefix}/datasets/{parts[1]}/runs",
        ]

    if (parts[0] == "datasets" and len(parts) == 3 and parts[2] == "runs"):
        return await _readdir_dataset_runs(
            accessor,
            parts[1],
            virtual_key,
            index,
            prefix,
        )

    raise enoent(virtual)


async def _readdir_traces(
    accessor: LangfuseAccessor,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    limit = accessor.config.default_trace_limit
    traces = await fetch_traces(accessor.api, limit=limit)
    entries = []
    names = []
    for t in traces:
        trace_id = t.get("id", "")
        filename = f"{trace_id}.json"
        entry = IndexEntry(
            id=trace_id,
            name=trace_id,
            resource_type="langfuse/trace",
            vfs_name=filename,
        )
        entries.append((filename, entry))
        names.append(f"{prefix}/traces/{filename}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_sessions(
    accessor: LangfuseAccessor,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    sessions = await fetch_sessions(accessor.api)
    entries = []
    names = []
    for s in sessions:
        session_id = s.get("id", "")
        entry = IndexEntry(
            id=session_id,
            name=session_id,
            resource_type="langfuse/session",
            vfs_name=session_id,
        )
        entries.append((session_id, entry))
        names.append(f"{prefix}/sessions/{session_id}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_session_traces(
    accessor: LangfuseAccessor,
    session_id: str,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    limit = accessor.config.default_trace_limit
    traces = await fetch_traces(
        accessor.api,
        session_id=session_id,
        limit=limit,
    )
    entries = []
    names = []
    for t in traces:
        trace_id = t.get("id", "")
        filename = f"{trace_id}.json"
        entry = IndexEntry(
            id=trace_id,
            name=trace_id,
            resource_type="langfuse/trace",
            vfs_name=filename,
        )
        entries.append((filename, entry))
        names.append(f"{prefix}/sessions/{session_id}/{filename}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_prompts(
    accessor: LangfuseAccessor,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    prompts = await fetch_prompts(accessor.api)
    seen: set[str] = set()
    entries = []
    names = []
    for p in prompts:
        prompt_name = p.get("name", "")
        if prompt_name in seen:
            continue
        seen.add(prompt_name)
        entry = IndexEntry(
            id=prompt_name,
            name=prompt_name,
            resource_type="langfuse/prompt",
            vfs_name=prompt_name,
        )
        entries.append((prompt_name, entry))
        names.append(f"{prefix}/prompts/{prompt_name}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_prompt_versions(
    accessor: LangfuseAccessor,
    prompt_name: str,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    prompts = await fetch_prompts(accessor.api)
    entries = []
    names = []
    for p in prompts:
        if p.get("name") != prompt_name:
            continue
        version = p.get("version", 0)
        filename = f"{version}.json"
        entry = IndexEntry(
            id=f"{prompt_name}/{version}",
            name=str(version),
            resource_type="langfuse/prompt_version",
            vfs_name=filename,
        )
        entries.append((filename, entry))
        names.append(f"{prefix}/prompts/{prompt_name}/{filename}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_datasets(
    accessor: LangfuseAccessor,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    datasets = await fetch_datasets(accessor.api)
    entries = []
    names = []
    for d in datasets:
        dataset_name = d.get("name", "")
        entry = IndexEntry(
            id=dataset_name,
            name=dataset_name,
            resource_type="langfuse/dataset",
            vfs_name=dataset_name,
        )
        entries.append((dataset_name, entry))
        names.append(f"{prefix}/datasets/{dataset_name}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


async def _readdir_dataset_runs(
    accessor: LangfuseAccessor,
    dataset_name: str,
    virtual_key: str,
    index: IndexCacheStore | None,
    prefix: str,
) -> list[str]:
    if index is not None:
        listing = await index.list_dir(virtual_key)
        if listing.entries is not None:
            return listing.entries
    runs = await fetch_dataset_runs(accessor.api, dataset_name)
    entries = []
    names = []
    for r in runs:
        run_name = r.get("name", "")
        filename = f"{run_name}.jsonl"
        entry = IndexEntry(
            id=run_name,
            name=run_name,
            resource_type="langfuse/dataset_run",
            vfs_name=filename,
        )
        entries.append((filename, entry))
        names.append(f"{prefix}/datasets/{dataset_name}/runs/{filename}")
    if index is not None:
        await index.set_dir(virtual_key, entries)
    return names


def is_dir_name(child: str) -> bool:
    # Entries are recognized by extension, so classification never needs
    # the stat fallback.
    name = child.rsplit("/", 1)[-1]
    return not (name.endswith(".json") or name.endswith(".jsonl"))