cleveragents-core/robot/helper_skeleton_compressor.py

"""Helper script for skeleton compressor Robot Framework tests.

Usage:
    python helper_skeleton_compressor.py compress <ratio>
    python helper_skeleton_compressor.py validate-ratio-bounds
    python helper_skeleton_compressor.py metadata-fields
    python helper_skeleton_compressor.py stable-ordering
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.services.skeleton_compressor import (  # noqa: E402
    SkeletonCompressorService,
)
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextFragment,
    FragmentProvenance,
)

# Default provenance for skeleton test fragments.
_SKEL_PROV = FragmentProvenance(resource_uri="skeleton://robot-test")


def _make_skel_fragment(
    fragment_id: str,
    content: str,
    token_count: int,
    relevance_score: float,
    source_decision_id: str | None = None,
) -> ContextFragment:
    """Create a ContextFragment suitable for skeleton compression tests."""
    metadata: dict[str, str] = {}
    if source_decision_id is not None:
        metadata["source_decision_id"] = source_decision_id
    return ContextFragment(
        fragment_id=fragment_id,
        uko_node=f"skeleton://{fragment_id}",
        content=content,
        token_count=token_count,
        relevance_score=relevance_score,
        provenance=_SKEL_PROV,
        metadata=metadata,
    )


def _sample_fragments() -> list[ContextFragment]:
    """Build a repeatable set of sample fragments."""
    return [
        _make_skel_fragment(
            fragment_id="frag-001",
            content="High relevance content " * 20,
            token_count=400,
            relevance_score=0.9,
            source_decision_id="01HXDECISION00000000000001",
        ),
        _make_skel_fragment(
            fragment_id="frag-002",
            content="Medium relevance content " * 15,
            token_count=300,
            relevance_score=0.6,
            source_decision_id="01HXDECISION00000000000002",
        ),
        _make_skel_fragment(
            fragment_id="frag-003",
            content="Low relevance content " * 15,
            token_count=300,
            relevance_score=0.3,
            source_decision_id="01HXDECISION00000000000003",
        ),
    ]


def cmd_compress(ratio_str: str) -> None:
    """Compress sample fragments and print summary."""
    ratio = float(ratio_str)
    svc = SkeletonCompressorService()
    result = svc.compress(_sample_fragments(), skeleton_ratio=ratio)
    meta = result.metadata
    print(f"skeleton-compress-ok ratio={meta.ratio}")
    print(f"original_tokens={meta.original_tokens}")
    print(f"compressed_tokens={meta.compressed_tokens}")
    print(f"fragment_count={len(result.fragments)}")
    print(f"decision_ids={len(meta.source_decision_ids)}")


def cmd_validate_ratio_bounds() -> None:
    """Verify that out-of-range ratios raise ValueError."""
    svc = SkeletonCompressorService()
    frags = _sample_fragments()

    for bad in (-0.1, 1.5, 2.0):
        try:
            svc.compress(frags, skeleton_ratio=bad)
            print(f"FAIL: ratio {bad} did not raise")
            sys.exit(1)
        except ValueError:
            pass

    # Valid bounds
    for good in (0.0, 0.5, 1.0):
        svc.compress(frags, skeleton_ratio=good)

    print("skeleton-ratio-bounds-ok")


def cmd_metadata_fields() -> None:
    """Verify metadata fields are populated correctly."""
    svc = SkeletonCompressorService()
    result = svc.compress(_sample_fragments(), skeleton_ratio=0.5)
    meta = result.metadata

    checks_passed = True

    if meta.ratio != 0.5:
        print(f"FAIL: ratio={meta.ratio}")
        checks_passed = False
    if meta.original_tokens != 1000:
        print(f"FAIL: original_tokens={meta.original_tokens}")
        checks_passed = False
    if meta.compressed_tokens > 500:
        print(f"FAIL: compressed_tokens={meta.compressed_tokens} > 500")
        checks_passed = False
    if not meta.source_decision_ids:
        print("FAIL: no decision IDs")
        checks_passed = False

    if checks_passed:
        print("skeleton-metadata-ok")
    else:
        sys.exit(1)


def cmd_stable_ordering() -> None:
    """Verify fragments come out in deterministic order."""
    svc = SkeletonCompressorService()
    frags = [
        _make_skel_fragment(
            fragment_id=f"frag-{chr(ord('c') - i)}",
            content="x",
            token_count=10,
            relevance_score=0.5,
        )
        for i in range(3)
    ]

    r1 = svc.compress(frags, skeleton_ratio=0.0)
    r2 = svc.compress(list(reversed(frags)), skeleton_ratio=0.0)

    ids1 = [f.fragment_id for f in r1.fragments]
    ids2 = [f.fragment_id for f in r2.fragments]

    if ids1 == ids2:
        print("skeleton-stable-ordering-ok")
    else:
        print(f"FAIL: {ids1} != {ids2}")
        sys.exit(1)


def main() -> None:
    """Dispatch subcommand."""
    if len(sys.argv) < 2:
        print("Usage: helper_skeleton_compressor.py <command> [args]")
        sys.exit(1)

    cmd = sys.argv[1]
    if cmd == "compress":
        if len(sys.argv) < 3:
            print("Usage: compress <ratio>")
            sys.exit(1)
        cmd_compress(sys.argv[2])
    elif cmd == "validate-ratio-bounds":
        cmd_validate_ratio_bounds()
    elif cmd == "metadata-fields":
        cmd_metadata_fields()
    elif cmd == "stable-ordering":
        cmd_stable_ordering()
    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)


if __name__ == "__main__":
    main()