coreason-meta-engineering 0.1.0

# Copyright (c) 2026 CoReason, Inc
#
# This software is proprietary and dual-licensed
# Licensed under the Prosperity Public License 3.0 (the "License")
# A copy of the license is available at <https://prosperitylicense.com/versions/3.0.0>
# For details, see the LICENSE file
# Commercial use beyond a 30-day trial requires a separate license
#
# Source Code: <https://github.com/CoReason-AI/coreason-meta-engineering>

"""Tests for the PVV Pipeline (pvv.py)."""

import pytest
from coreason_manifest.spec import CognitiveDeliberativeEnvelopeState, OracleExecutionReceipt
from pydantic import BaseModel

from coreason_meta_engineering.pvv import (
    _epistemic_strip,
    _generate_receipt,
    _native_validation,
    compute_canonical_hash,
    execute_pvv_pipeline,
)

# ═════════════════════════════════════════════════════════════════════════════
# Phase 1: Epistemic Strip
# ═════════════════════════════════════════════════════════════════════════════


class TestEpistemicStrip:
    """The Epistemic Strip must discard the deliberation_trace entirely."""

    def test_extracts_payload(self) -> None:
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="I am thinking very hard about this problem...",
            payload="x = 1\n",
        )
        result = _epistemic_strip(envelope)
        assert result == "x = 1\n"

    def test_discards_deliberation_trace(self) -> None:
        trace = "SECRET INTERNAL REASONING" * 100
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace=trace,
            payload="y = 2\n",
        )
        result = _epistemic_strip(envelope)
        assert "SECRET" not in result
        assert result == "y = 2\n"


# ═════════════════════════════════════════════════════════════════════════════
# Phase 2 & 3: Native Validation (Syntax & Schema)
# ═════════════════════════════════════════════════════════════════════════════


class TestNativeValidation:
    """Native validation must catch syntax errors and missing schemas."""

    def test_valid_python_parses(self) -> None:
        payload = "def greet(name: str) -> str:\n    return f'Hello, {name}'\n"
        _native_validation(payload, target_schema=None)  # Should not raise

    def test_invalid_python_raises(self) -> None:
        with pytest.raises(SyntaxError):
            _native_validation("def broken(:\n", target_schema=None)

    def test_empty_module_parses(self) -> None:
        _native_validation("", target_schema=None)

    def test_subprocess_env_preserves_pythonpath_and_virtualenv(self, monkeypatch: pytest.MonkeyPatch) -> None:
        from unittest.mock import patch

        monkeypatch.setenv("PYTHONPATH", "/workspace/path/override")
        monkeypatch.setenv("VIRTUAL_ENV", "/workspace/.venv")
        monkeypatch.setenv("PYTHONHOME", "/discarded/path")

        captured_env = None

        from typing import Any

        def mock_run(*_args: Any, **kwargs: Any) -> Any:
            nonlocal captured_env
            captured_env = kwargs.get("env")

            class MockResult:
                returncode = 0
                stdout = "Driver: main done"
                stderr = ""

            return MockResult()

        with patch("subprocess.run", side_effect=mock_run):
            _native_validation("x = 1\n", target_schema=None)

        assert captured_env is not None
        assert captured_env.get("PYTHONPATH") == "/workspace/path/override"
        assert captured_env.get("VIRTUAL_ENV") == "/workspace/.venv"
        assert "PYTHONHOME" not in captured_env


# ═════════════════════════════════════════════════════════════════════════════
# Phase 4: Receipt Generation
# ═════════════════════════════════════════════════════════════════════════════


class TestReceiptGeneration:
    """The receipt must contain a valid SHA-256 hash of the payload."""

    def test_receipt_hash_matches(self) -> None:
        payload = "x = 42\n"
        receipt = _generate_receipt(
            payload=payload,
            solver_urn="urn:coreason:solver:test:v1",
            tokens_burned=100,
        )
        expected_hash = compute_canonical_hash(payload)
        assert receipt.execution_hash == expected_hash
        assert receipt.solver_urn == "urn:coreason:solver:test:v1"
        assert receipt.tokens_burned == 100

    def test_receipt_is_oracle_execution_receipt(self) -> None:
        receipt = _generate_receipt(
            payload="y = 1\n",
            solver_urn="urn:coreason:solver:test:v1",
            tokens_burned=0,
        )
        assert isinstance(receipt, OracleExecutionReceipt)


# ═════════════════════════════════════════════════════════════════════════════
# Full Pipeline Integration
# ═════════════════════════════════════════════════════════════════════════════


class TestFullPipeline:
    """End-to-end tests for the complete PVV pipeline."""

    def test_valid_code_produces_receipt(self) -> None:
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="Let me think about this carefully...",
            payload="def greet(name: str) -> str:\n    return f'Hello, {name}'\n",
        )
        receipt = execute_pvv_pipeline(
            envelope=envelope,
            solver_urn="urn:coreason:solver:claw_developer:v1",
            tokens_burned=1500,
        )
        assert isinstance(receipt, OracleExecutionReceipt)
        assert receipt.tokens_burned == 1500
        assert len(receipt.execution_hash) == 64

    def test_syntax_error_halts_pipeline(self) -> None:
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="Generating code...",
            payload="def broken(:\n",
        )
        with pytest.raises(SyntaxError):
            execute_pvv_pipeline(
                envelope=envelope,
                solver_urn="urn:coreason:solver:claw_developer:v1",
                tokens_burned=100,
            )

    def test_timeout_halts_pipeline(self) -> None:
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="Generating infinite loop code...",
            payload="import time\ntime.sleep(15)\n",
        )
        with pytest.raises(RuntimeError, match="Epistemic Verification Timeout"):
            execute_pvv_pipeline(
                envelope=envelope,
                solver_urn="urn:coreason:solver:claw_developer:v1",
                tokens_burned=100,
            )


# ═════════════════════════════════════════════════════════════════════════════
# Schema Validation
# ═════════════════════════════════════════════════════════════════════════════


class TestSchemaValidation:
    """Tests for schema-related branches in the PVV pipeline."""

    def test_valid_code_with_matching_schema(self) -> None:
        """Schema is provided and a generated Pydantic model matches it."""
        code = "from pydantic import BaseModel\n\nclass Patient(BaseModel):\n    name: str\n    age: int\n"
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="thinking...",
            payload=code,
        )
        schema = {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}}
        receipt = execute_pvv_pipeline(
            envelope=envelope,
            solver_urn="urn:coreason:solver:test:v1",
            tokens_burned=50,
            target_schema=schema,
        )
        assert isinstance(receipt, OracleExecutionReceipt)

    def test_schema_mismatch_raises_value_error(self) -> None:
        """Schema is provided but the generated model is missing required properties."""
        code = "from pydantic import BaseModel\n\nclass Patient(BaseModel):\n    name: str\n"
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="thinking...",
            payload=code,
        )
        schema = {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}}
        with pytest.raises(ValueError, match="missing property"):
            execute_pvv_pipeline(
                envelope=envelope,
                solver_urn="urn:coreason:solver:test:v1",
                tokens_burned=50,
                target_schema=schema,
            )

    def test_schema_with_no_pydantic_models_raises(self) -> None:
        """Schema provided but the payload has no Pydantic models."""
        code = "x = 42\n"
        envelope = CognitiveDeliberativeEnvelopeState[str](
            deliberation_trace="thinking...",
            payload=code,
        )
        schema = {"properties": {"name": {"type": "string"}}}
        with pytest.raises(ValueError, match="No Pydantic models found"):
            execute_pvv_pipeline(
                envelope=envelope,
                solver_urn="urn:coreason:solver:test:v1",
                tokens_burned=50,
                target_schema=schema,
            )


# ═════════════════════════════════════════════════════════════════════════════
# _compare_schema Unit Tests
# ═════════════════════════════════════════════════════════════════════════════


class TestCompareSchema:
    """Direct unit tests for the _compare_schema function."""

    def test_no_models_with_empty_schema_passes(self) -> None:
        from types import ModuleType

        from coreason_meta_engineering.pvv import _compare_schema

        mod = ModuleType("empty_module")
        _compare_schema(mod, {})  # Should not raise

    def test_no_models_with_populated_schema_raises(self) -> None:
        from types import ModuleType

        from coreason_meta_engineering.pvv import _compare_schema

        mod = ModuleType("empty_module")
        with pytest.raises(ValueError, match="No Pydantic models found"):
            _compare_schema(mod, {"properties": {"name": {"type": "string"}}})

    def test_matching_model_passes(self) -> None:
        from types import ModuleType

        from coreason_meta_engineering.pvv import _compare_schema

        mod = ModuleType("model_module")

        class TestModel(BaseModel):
            name: str
            age: int

        mod.TestModel = TestModel  # type: ignore[attr-defined]
        _compare_schema(mod, {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}})

    def test_missing_property_raises(self) -> None:
        from types import ModuleType

        from coreason_meta_engineering.pvv import _compare_schema

        mod = ModuleType("model_module")

        class PartialModel(BaseModel):
            name: str

        mod.PartialModel = PartialModel  # type: ignore[attr-defined]
        with pytest.raises(ValueError, match="missing property"):
            _compare_schema(mod, {"properties": {"name": {"type": "string"}, "age": {"type": "integer"}}})


# ═════════════════════════════════════════════════════════════════════════════
# Spec Creation Failure
# ═════════════════════════════════════════════════════════════════════════════


class TestNativeValidationEdgeCases:
    """Edge cases in _native_validation for spec creation failures."""

    def test_spec_creation_failure_raises_runtime_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
        import importlib.util as iu

        monkeypatch.setattr(iu, "spec_from_file_location", lambda *_a, **_kw: None)
        with pytest.raises(RuntimeError, match="Failed to create module spec"):
            _native_validation("x = 1\n", target_schema=None)