infospace-bench/tests/test_engine_boundary.py

import json
import os
import subprocess
import sys
from pathlib import Path

from infospace_bench import add_artifact, create_infospace, load_infospace, register_artifact
from infospace_bench.engine import (
    LocalAssetRepository,
    engine_capability_contract,
    plan_asset_sync,
    sync_assets,
)


def cli_env() -> dict[str, str]:
    env = os.environ.copy()
    env["PYTHONPATH"] = "src:/home/worsch/markitect-tool/src"
    return env


def make_engine_infospace(tmp_path: Path) -> Path:
    infospace = create_infospace(tmp_path, "pilot", name="Pilot")
    source = tmp_path / "chapter.md"
    source.write_text("# Chapter\n\nSource text.\n", encoding="utf-8")
    add_artifact(infospace.root, source, kind="source", title="Chapter")

    generated = infospace.root / "artifacts" / "generated" / "chapter-summary.md"
    generated.parent.mkdir(parents=True, exist_ok=True)
    generated.write_text("# Chapter Summary\n\nGenerated text.\n", encoding="utf-8")
    register_artifact(
        infospace.root,
        artifact_id="generated/chapter-summary.md",
        path="artifacts/generated/chapter-summary.md",
        kind="generated",
        title="Chapter Summary",
        provenance={
            "workflow_id": "source-summary",
            "stage_id": "render-summary",
            "input_artifact_id": "source/chapter.md",
        },
        relationships=[
            {"type": "generated_from", "target": "source/chapter.md"},
        ],
    )
    return infospace.root


def test_engine_capability_contract_keeps_application_boundary_explicit() -> None:
    contract = engine_capability_contract()

    assert "asset_identity" in contract.engine_backed
    assert "artifact_manifest" in contract.file_backed
    assert "markdown_parsing" not in contract.engine_backed
    assert contract.assumptions["source"] == "kontextual-engine/INTENT.md"
    assert contract.review_required == [
        "apply_sync",
        "engine_client_configuration",
        "permission_changes",
    ]


def test_plan_asset_sync_is_dry_run_and_maps_artifacts_to_assets(
    tmp_path: Path,
) -> None:
    root = make_engine_infospace(tmp_path)
    repository_path = root / "output" / "engine" / "assets.yaml"

    plan = plan_asset_sync(root)

    assert plan.dry_run is True
    assert repository_path.exists() is False
    assert [action.action for action in plan.actions] == ["create", "create"]
    assert plan.actions[0].artifact_id == "source/chapter.md"
    assert plan.actions[0].asset_id == "infospace:pilot:artifact:source/chapter.md"
    assert len(plan.actions[0].digest) == 64
    assert plan.actions[1].workflow_id == "source-summary"


def test_sync_assets_apply_persists_to_local_repository_and_detects_updates(
    tmp_path: Path,
) -> None:
    root = make_engine_infospace(tmp_path)

    applied = sync_assets(root, dry_run=False)
    repository = LocalAssetRepository(root / "output" / "engine" / "assets.yaml")
    assets = repository.list_assets()
    unchanged = plan_asset_sync(root)

    source_path = root / "artifacts" / "sources" / "chapter.md"
    source_path.write_text("# Chapter\n\nChanged text.\n", encoding="utf-8")
    updated = plan_asset_sync(root)

    assert applied.status == "completed"
    assert len(assets) == 2
    assert repository.get_asset("infospace:pilot:artifact:source/chapter.md") is not None
    assert [action.action for action in unchanged.actions] == [
        "unchanged",
        "unchanged",
    ]
    assert updated.actions[0].action == "update"
    assert updated.actions[0].previous_digest != updated.actions[0].digest


def test_cli_engine_inspect_plan_sync_and_apply(tmp_path: Path) -> None:
    root = make_engine_infospace(tmp_path)

    inspected = subprocess.run(
        [sys.executable, "-m", "infospace_bench", "engine", "inspect", str(root)],
        check=False,
        env=cli_env(),
        text=True,
        capture_output=True,
    )
    planned = subprocess.run(
        [sys.executable, "-m", "infospace_bench", "engine", "plan-sync", str(root)],
        check=False,
        env=cli_env(),
        text=True,
        capture_output=True,
    )
    dry_run_sync = subprocess.run(
        [sys.executable, "-m", "infospace_bench", "engine", "sync", str(root)],
        check=False,
        env=cli_env(),
        text=True,
        capture_output=True,
    )
    applied = subprocess.run(
        [
            sys.executable,
            "-m",
            "infospace_bench",
            "engine",
            "sync",
            str(root),
            "--apply",
        ],
        check=False,
        env=cli_env(),
        text=True,
        capture_output=True,
    )

    assert inspected.returncode == 0, inspected.stderr
    assert planned.returncode == 0, planned.stderr
    assert dry_run_sync.returncode == 0, dry_run_sync.stderr
    assert applied.returncode == 0, applied.stderr
    assert json.loads(inspected.stdout)["contract"]["integration_posture"] == "optional"
    assert json.loads(planned.stdout)["dry_run"] is True
    assert json.loads(dry_run_sync.stdout)["dry_run"] is True
    assert json.loads(applied.stdout)["status"] == "completed"
    assert (root / "output" / "engine" / "assets.yaml").is_file()


def test_file_backed_infospace_remains_authoritative_after_engine_sync(
    tmp_path: Path,
) -> None:
    root = make_engine_infospace(tmp_path)

    sync_assets(root, dry_run=False)
    loaded = load_infospace(root)

    assert [artifact.id for artifact in loaded.artifacts] == [
        "source/chapter.md",
        "generated/chapter-summary.md",
    ]