cleveragents-core/robot/helper_temporal_data_model.py

"""Robot Framework helper for Temporal Data Model smoke tests.

Provides a CLI-style interface for Robot to invoke temporal model
creation, revision chain operations, tier queries, and service-level
operations.  Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_temporal_data_model.py <command>
"""

from __future__ import annotations

import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path

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

from cleveragents.application.services.temporal_service import (  # noqa: E402
    TemporalService,
)
from cleveragents.domain.models.acms.temporal import (  # noqa: E402
    RevisionChain,
    TemporalBackend,
    TemporalMetadata,
    TemporalNode,
    TierQueryResult,
    TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import (  # noqa: E402
    InMemoryTemporalBackend,
)
from cleveragents.domain.models.acms.tiers import ContextTier  # noqa: E402
from cleveragents.domain.models.core.project import (  # noqa: E402
    TemporalScope,
)


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print("Usage: helper_temporal_data_model.py <command>")
        return 1

    command: str = sys.argv[1]

    if command == "protocol-compliance":
        return _test_protocol_compliance()
    if command == "revision-chain":
        return _test_revision_chain()
    if command == "tier-queries":
        return _test_tier_queries()
    if command == "service-lifecycle":
        return _test_service_lifecycle()
    if command == "model-creation":
        return _test_model_creation()
    if command == "validation":
        return _test_validation()
    if command == "retention-defaults":
        return _test_retention_defaults()
    if command == "integration-modify-resource":
        return _test_integration_modify_resource()

    print(f"Unknown command: {command}")
    return 1


def _test_protocol_compliance() -> int:
    """Verify InMemoryTemporalBackend satisfies TemporalBackend."""
    try:
        backend = InMemoryTemporalBackend()
        assert isinstance(backend, TemporalBackend)
        print("temporal-protocol-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_revision_chain() -> int:
    """Test revision chain creation and traversal."""
    try:
        backend = InMemoryTemporalBackend()
        now = datetime.now(tz=UTC)

        # Store v1
        v1 = TemporalNode(
            node_uri="uko-py:class/Auth_v1",
            source_resource="RES001",
            source_path="src/auth.py",
            temporal=TemporalMetadata(
                valid_from=now - timedelta(days=5),
                is_current=True,
            ),
        )
        backend.store_node(v1)

        # Create v2
        v2 = TemporalNode(
            node_uri="uko-py:class/Auth_v2",
            source_resource="RES001",
            source_path="src/auth.py",
            temporal=TemporalMetadata(
                valid_from=now - timedelta(days=2),
                is_current=True,
                is_revision_of="uko-py:class/Auth_v1",
            ),
        )
        backend.create_revision("uko-py:class/Auth_v1", v2, now - timedelta(days=2))

        # Create v3
        v3 = TemporalNode(
            node_uri="uko-py:class/Auth_v3",
            source_resource="RES001",
            source_path="src/auth.py",
            temporal=TemporalMetadata(
                valid_from=now,
                is_current=True,
                is_revision_of="uko-py:class/Auth_v2",
            ),
        )
        backend.create_revision("uko-py:class/Auth_v2", v3, now)

        # Verify chain
        chain = backend.get_revision_chain("uko-py:class/Auth_v3")
        assert chain.depth == 3
        assert chain.current_uri == "uko-py:class/Auth_v3"
        assert "uko-py:class/Auth_v1" in chain.predecessors
        assert "uko-py:class/Auth_v2" in chain.predecessors

        print("temporal-revision-chain-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_tier_queries() -> int:
    """Test tier-aware temporal queries."""
    try:
        backend = InMemoryTemporalBackend()
        now = datetime.now(tz=UTC)
        retention = TierRetentionConfig()

        # Store a current node and a historical node
        v1 = TemporalNode(
            node_uri="uko-py:class/Foo_v1",
            source_resource="RES001",
            source_path="src/foo.py",
            temporal=TemporalMetadata(
                valid_from=now - timedelta(days=5),
                is_current=True,
            ),
        )
        backend.store_node(v1)
        v2 = TemporalNode(
            node_uri="uko-py:class/Foo_v2",
            source_resource="RES001",
            source_path="src/foo.py",
            temporal=TemporalMetadata(
                valid_from=now,
                is_current=True,
                is_revision_of="uko-py:class/Foo_v1",
            ),
        )
        backend.create_revision("uko-py:class/Foo_v1", v2, now)

        # Hot tier: only current
        hot = backend.query_by_tier(ContextTier.HOT, TemporalScope.CURRENT, retention)
        assert len(hot.nodes) == 1
        assert hot.nodes[0].temporal.is_current is True

        # Cold tier: all
        cold = backend.query_by_tier(ContextTier.COLD, TemporalScope.ALL, retention)
        assert len(cold.nodes) == 2

        print("temporal-tier-queries-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_service_lifecycle() -> int:
    """Test TemporalService lifecycle."""
    try:
        backend = InMemoryTemporalBackend()
        service = TemporalService(backend=backend)

        # Store initial
        node = service.store_initial_node(
            node_uri="uko-py:class/Bar_v1",
            source_resource="RES002",
            source_path="src/bar.py",
        )
        assert node.temporal.is_current is True

        # Create revision
        v2 = service.create_revision(
            current_uri="uko-py:class/Bar_v1",
            new_node_uri="uko-py:class/Bar_v2",
            source_resource="RES002",
            source_path="src/bar.py",
        )
        assert v2.temporal.is_revision_of == "uko-py:class/Bar_v1"

        # Get current
        current = service.get_current("uko-py:class/Bar")
        assert current is not None
        assert current.node_uri == "uko-py:class/Bar_v2"

        # Get chain
        chain = service.get_revision_chain("uko-py:class/Bar_v2")
        assert chain.depth == 2

        print("temporal-service-lifecycle-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_model_creation() -> int:
    """Test domain model creation."""
    try:
        now = datetime.now(tz=UTC)
        meta = TemporalMetadata(valid_from=now)
        assert meta.is_current is True
        assert meta.valid_until is None

        node = TemporalNode(
            node_uri="uko-py:class/Test_v1",
            source_resource="RES",
            source_path="src/test.py",
            temporal=meta,
        )
        assert node.node_uri == "uko-py:class/Test_v1"

        chain = RevisionChain(current_uri="uko:test")
        assert chain.depth == 1

        result = TierQueryResult(
            tier=ContextTier.HOT,
            temporal_scope=TemporalScope.CURRENT,
        )
        assert result.nodes == ()

        retention = TierRetentionConfig()
        assert retention.warm_retention_hours == 24

        print("temporal-model-creation-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_validation() -> int:
    """Test input validation on models and backend."""
    try:
        backend = InMemoryTemporalBackend()

        # Blank URI should raise ValueError
        try:
            backend.get_current("")
            print("FAIL: expected ValueError")
            return 1
        except ValueError:
            pass

        # Nonexistent node should raise KeyError
        try:
            backend.mark_historical("uko:nonexistent", datetime.now(tz=UTC))
            print("FAIL: expected KeyError")
            return 1
        except KeyError:
            pass

        print("temporal-validation-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_retention_defaults() -> int:
    """Test retention config defaults match spec."""
    try:
        retention = TierRetentionConfig()
        assert retention.warm_retention_hours == 24
        assert retention.cold_retention_days == 90
        print("temporal-retention-defaults-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


def _test_integration_modify_resource() -> int:
    """Integration test: modify resource, verify temporal chain.

    Acceptance criteria #10: Modify a resource, verify old node is
    marked historical, new node is current, revision chain is intact.
    """
    try:
        backend = InMemoryTemporalBackend()
        service = TemporalService(backend=backend)

        # Step 1: Initial analysis creates first node
        v1 = service.store_initial_node(
            node_uri="uko-py:class/AuthManager_v1",
            source_resource="RES_AUTH",
            source_path="src/auth/manager.py",
            source_range="15:1-87:0",
        )
        assert v1.temporal.is_current is True
        assert v1.temporal.is_revision_of is None

        # Step 2: Code change triggers re-analysis → new revision
        v2 = service.create_revision(
            current_uri="uko-py:class/AuthManager_v1",
            new_node_uri="uko-py:class/AuthManager_v2",
            source_resource="RES_AUTH",
            source_path="src/auth/manager.py",
            source_range="15:1-95:0",
        )

        # Step 3: Verify old node is marked historical
        old = backend._nodes["uko-py:class/AuthManager_v1"]
        assert old.temporal.is_current is False
        assert old.temporal.valid_until is not None

        # Step 4: Verify new node is current
        assert v2.temporal.is_current is True
        assert v2.temporal.is_revision_of == "uko-py:class/AuthManager_v1"

        # Step 5: Verify revision chain is intact
        chain = service.get_revision_chain("uko-py:class/AuthManager_v2")
        assert chain.depth == 2
        assert chain.current_uri == "uko-py:class/AuthManager_v2"
        assert "uko-py:class/AuthManager_v1" in chain.predecessors

        # Step 6: Hot tier shows only current
        hot = service.query_by_tier(ContextTier.HOT)
        assert len(hot.nodes) == 1
        assert hot.nodes[0].node_uri == "uko-py:class/AuthManager_v2"

        # Step 7: Cold tier shows full history
        cold = service.query_by_tier(ContextTier.COLD)
        assert len(cold.nodes) == 2

        print("temporal-integration-modify-resource-ok")
        return 0
    except Exception as exc:
        print(f"FAIL: {exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())