strukto-ai--mirage/examples/python/langfuse/demo.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. =========

import asyncio
import os
import time

from dotenv import load_dotenv

from mirage import MountMode, Workspace
from mirage.resource.langfuse import LangfuseConfig, LangfuseResource

load_dotenv(".env.development")

config = LangfuseConfig(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ["LANGFUSE_HOST"],
    default_trace_limit=20,
)
resource = LangfuseResource(config=config)


async def _run(ws, cmd):
    print(f"\n>>> {cmd}")
    r = await ws.execute(cmd)
    out = (await r.stdout_str()).strip()
    err = await r.stderr_str()
    if out:
        for line in out.splitlines()[:10]:
            print(f"  {line[:120]}")
        total = len(out.splitlines())
        if total > 10:
            print(f"  ... ({total} lines total)")
    if err:
        print(f"  [stderr] {err.strip()[:120]}")
    if not out and not err:
        print(f"  (empty, exit={r.exit_code})")
    return r


async def _timed(ws, cmd):
    start = time.perf_counter()
    out = await (await ws.execute(cmd)).stdout_str()
    return (time.perf_counter() - start) * 1000, out


async def main():
    ws = Workspace({"/langfuse": resource}, mode=MountMode.READ)

    print("=== not-found errors show the full virtual path ===")
    for cmd in ("cat /langfuse/__nf_missing__.txt",
                "head /langfuse/__nf_missing__.txt",
                "stat /langfuse/__nf_missing__.txt"):
        result = await ws.execute(cmd)
        print(f"$ {cmd}")
        print(f"  exit={result.exit_code}  "
              f"{(await result.stderr_str()).strip()}")

    print("=" * 60)
    print("LS across all directories")
    print("=" * 60)

    await _run(ws, "ls /langfuse/")
    await _run(ws, "ls /langfuse/traces/")
    await _run(ws, "ls /langfuse/sessions/")
    await _run(ws, "ls /langfuse/prompts/")
    await _run(ws, "ls /langfuse/datasets/")

    print("\n" + "=" * 60)
    print("CAT across different resource types")
    print("=" * 60)

    r = await ws.execute("ls /langfuse/traces/")
    traces_out = (await r.stdout_str()).strip()
    trace_files = [f for f in traces_out.splitlines() if f.strip()]
    if trace_files:
        tp = f"/langfuse/traces/{trace_files[0].strip()}"
        await _run(ws, f'cat "{tp}" | head -n 5')
    else:
        print("  no traces found, skipping cat trace")

    await _run(ws, 'cat "/langfuse/prompts/summarize/1.json"')
    await _run(ws, 'cat "/langfuse/datasets/qa-eval/items.jsonl"')

    print("\n" + "=" * 60)
    print("GREP across different scopes")
    print("=" * 60)

    await _run(ws, 'grep "chat" "/langfuse/traces/"')

    await _run(ws, 'grep "support" "/langfuse/traces/"')

    await _run(ws, 'grep "chat-session" "/langfuse/sessions/"')

    await _run(ws, 'grep "summarize" "/langfuse/prompts/"')

    await _run(ws, 'grep "qa" "/langfuse/datasets/"')

    if trace_files:
        tp = f"/langfuse/traces/{trace_files[0].strip()}"
        await _run(ws, f'grep "name" "{tp}"')

    print("\n" + "=" * 60)
    print("JQ across different resource types")
    print("=" * 60)

    if trace_files:
        tp = f"/langfuse/traces/{trace_files[0].strip()}"
        await _run(ws, f'jq ".name" "{tp}"')
        await _run(ws, f'jq ".session_id" "{tp}"')
        await _run(ws, f'jq ".input" "{tp}"')

    await _run(
        ws,
        'jq ".prompt" "/langfuse/prompts/summarize/1.json"',
    )
    await _run(
        ws,
        'jq ".config" "/langfuse/prompts/classify/1.json"',
    )

    await _run(
        ws,
        'jq ".[] | .input" '
        '"/langfuse/datasets/qa-eval/items.jsonl"',
    )
    await _run(
        ws,
        'jq -r ".[] | .expected_output.answer" '
        '"/langfuse/datasets/qa-eval/items.jsonl"',
    )

    print("\n" + "=" * 60)
    print("SESSION browsing")
    print("=" * 60)

    await _run(ws, "ls /langfuse/sessions/chat-session-001/")

    r = await ws.execute("ls /langfuse/sessions/chat-session-001/", )
    session_traces = (await r.stdout_str()).strip().splitlines()
    if session_traces:
        st = session_traces[0].strip()
        sp = f"/langfuse/sessions/chat-session-001/{st}"
        await _run(ws, f'cat "{sp}" | head -n 3')

    print("\n" + "=" * 60)
    print("PROMPTS and DATASETS detail")
    print("=" * 60)

    await _run(ws, "ls /langfuse/prompts/summarize/")
    await _run(ws, "ls /langfuse/datasets/qa-eval/")
    await _run(ws, "ls /langfuse/datasets/qa-eval/runs/")

    await _run(ws, 'stat "/langfuse/prompts/summarize"')
    await _run(ws, 'stat "/langfuse/datasets/qa-eval"')

    print("\n" + "=" * 60)
    print("TREE, FIND, NAVIGATION")
    print("=" * 60)

    await _run(ws, "tree -L 2 /langfuse/")
    await _run(
        ws,
        'find "/langfuse/prompts/" -name "*.json"',
    )
    # -path matches the display path; -size counts dirs and sizeless
    # rendered files as 0 (so +0c drops them, -1k keeps them).
    await _run(ws, 'find "/langfuse/" -maxdepth 1 -path "*prompts*"')
    await _run(ws, 'find "/langfuse/" -maxdepth 1 -size +0c')
    await _run(ws, 'find "/langfuse/" -maxdepth 1 -size -1k')

    await ws.execute('cd "/langfuse/prompts"')
    await _run(ws, "pwd")
    await _run(ws, "ls")

    print("\n" + "=" * 60)
    print("CACHING: warm reads served from cache (no backend fetch)")
    print("=" * 60)
    r = await ws.execute('find "/langfuse/prompts/" -name "*.json"')
    prompt_files = (await r.stdout_str()).strip().splitlines()
    if prompt_files:
        cache_file = prompt_files[0].strip()
        cold_ms, body = await _timed(ws, f'cat "{cache_file}"')
        warm_ms, _ = await _timed(ws, f'cat "{cache_file}"')
        grep_ms, _ = await _timed(ws, f'grep . "{cache_file}"')
        head_ms, _ = await _timed(ws, f'head -n 1 "{cache_file}"')
        tail_ms, _ = await _timed(ws, f'tail -n 1 "{cache_file}"')
        wc_ms, _ = await _timed(ws, f'wc -l "{cache_file}"')
        print(f"  file={cache_file} size={len(body)}B")
        print(
            f"  cold cat={cold_ms:.0f}ms  warm cat={warm_ms:.0f}ms  "
            f"grep={grep_ms:.0f}ms head={head_ms:.0f}ms tail={tail_ms:.0f}ms "
            f"wc={wc_ms:.0f}ms")
        print(f"  served_from_cache={warm_ms < cold_ms / 5} "
              f"(warm speedup {cold_ms / max(warm_ms, 0.001):.0f}x)")

    print("\n" + "=" * 60)
    print("GLOB: mid-path patterns walk segment by segment")
    print("=" * 60)
    r = await ws.execute("echo /langfuse/prom*/*")
    out = (await r.stdout_str()).strip()
    print(f"  echo /langfuse/prom*/* -> {out[:200]}")
    assert "/langfuse/prompts/" in out, "mid-path glob did not expand"

    # A glob that matches nothing stays the literal word, so the
    # command reports it like GNU coreutils.
    r = await ws.execute("cat /langfuse/zz-none-*/x.json")
    err = (await r.stderr_str()).strip()
    print(f"  cat /langfuse/zz-none-*/x.json -> exit={r.exit_code} "
          f"{err[:120]}")
    assert r.exit_code == 1 and "zz-none-*" in err


if __name__ == "__main__":
    asyncio.run(main())