markitect-tool/tests/test_agent_working_memory.py

from pathlib import Path

import yaml
from click.testing import CliRunner

from markitect_tool.cli import main
from markitect_tool.memory import (
    ContextBudget,
    LocalContextPackageRegistry,
    MemoryNamespace,
    activate_context_package,
    create_context_package_from_index,
    create_context_package_from_manifest,
    create_context_package_from_sources,
    refresh_context_package,
)
from markitect_tool.policy import LocalLabelPolicyGateway


def test_create_context_package_from_sources_preserves_spans_and_summary(tmp_path: Path):
    source = tmp_path / "doc.md"
    source.write_text("# Doc\n\n## Decision\n\nUse local memory packages.\n", encoding="utf-8")

    package = create_context_package_from_sources(
        "sections[heading=Decision]",
        [source],
        root=tmp_path,
        namespace=MemoryNamespace(project="markitect-tool", task="MKTT-WP-0008"),
        budget=ContextBudget(max_items=1),
    )

    assert package.id.startswith("memory:package:")
    assert package.namespace.project == "markitect-tool"
    assert package.items[0].source.path == "doc.md"
    assert package.items[0].source.unit_kind == "section"
    assert "local memory packages" in package.items[0].text
    assert package.summaries[0].name == "overview"


def test_local_context_registry_saves_loads_and_activates(tmp_path: Path):
    source = tmp_path / "doc.md"
    source.write_text("# Doc\n\n## Context\n\nReusable project facts.\n", encoding="utf-8")
    package = create_context_package_from_sources("sections[heading=Context]", [source], root=tmp_path)
    registry = LocalContextPackageRegistry(tmp_path)

    saved_path = registry.save(package)
    loaded = registry.load(package.id)
    activation_id = registry.activate(package.id, "thread:test")
    activation = registry.load_activation(activation_id)

    assert saved_path.exists()
    assert loaded.id == package.id
    assert activation.status == "active"
    assert "Reusable project facts" in activation.content


def test_activation_can_recheck_policy_and_drop_denied_items(tmp_path: Path):
    public = tmp_path / "public.md"
    private = tmp_path / "private.md"
    public.write_text("---\npolicy:\n  labels: [public]\n---\n# Public\n\nVisible.\n", encoding="utf-8")
    private.write_text("---\npolicy:\n  labels: [internal]\n---\n# Private\n\nHidden.\n", encoding="utf-8")
    package = create_context_package_from_sources("document", [public, private], root=tmp_path)
    gateway = LocalLabelPolicyGateway(
        {
            "id": "memory-test-policy",
            "subjects": {
                "reader": {
                    "allowed_labels": ["public"],
                    "allowed_actions": ["read"],
                }
            },
            "default_subject": "reader",
        }
    )

    activation = activate_context_package(package, policy_gateway=gateway, subject="reader")

    assert len(activation.items) == 1
    assert "Visible" in activation.content
    assert "Hidden" not in activation.content
    assert activation.policy["summary"]["denied"] == 1


def test_create_context_package_from_index_search_and_refresh(tmp_path: Path):
    source = tmp_path / "doc.md"
    source.write_text("# Doc\n\nSearchable memory fact.\n", encoding="utf-8")
    runner = CliRunner()
    indexed = runner.invoke(main, ["cache", "index", str(tmp_path), "--root", str(tmp_path)])
    assert indexed.exit_code == 0

    package = create_context_package_from_index("Searchable", root=tmp_path, search=True)
    source.write_text("# Doc\n\nSearchable memory fact changed.\n", encoding="utf-8")
    reindexed = runner.invoke(main, ["cache", "index", str(tmp_path), "--root", str(tmp_path)])
    refreshed = refresh_context_package(package)

    assert reindexed.exit_code == 0
    assert package.items
    assert "changed" in refreshed.items[0].text
    assert refreshed.id == package.id


def test_create_context_package_from_manifest(tmp_path: Path):
    source = tmp_path / "doc.md"
    source.write_text("# Doc\n\n## Purpose\n\nManifest selected context.\n", encoding="utf-8")
    manifest = tmp_path / "context.yaml"
    manifest.write_text(
        yaml.safe_dump(
            {
                "title": "Manifest package",
                "namespace": {"project": "markitect-tool"},
                "retrieval_recipes": [
                    {
                        "kind": "selector",
                        "query": "sections[heading=Purpose]",
                        "sources": ["doc.md"],
                    }
                ],
            },
            sort_keys=False,
        ),
        encoding="utf-8",
    )

    package = create_context_package_from_manifest(manifest, root=tmp_path)

    assert package.title == "Manifest package"
    assert package.namespace.project == "markitect-tool"
    assert "Manifest selected context" in package.items[0].text


def test_mkt_context_pack_activate_explain_and_deactivate(tmp_path: Path):
    source = tmp_path / "doc.md"
    source.write_text("# Doc\n\n## Decision\n\nUse explicit context packages.\n", encoding="utf-8")
    runner = CliRunner()

    packed = runner.invoke(
        main,
        [
            "context",
            "pack",
            "sections[heading=Decision]",
            "--source",
            str(source),
            "--root",
            str(tmp_path),
            "--format",
            "json",
        ],
    )
    assert packed.exit_code == 0
    package_id = yaml.safe_load(packed.output)["id"]

    explained = runner.invoke(main, ["context", "explain", package_id, "--root", str(tmp_path)])
    activated = runner.invoke(
        main,
        [
            "context",
            "activate",
            package_id,
            "--root",
            str(tmp_path),
            "--format",
            "json",
        ],
    )
    activation_id = yaml.safe_load(activated.output)["id"]
    deactivated = runner.invoke(main, ["context", "deactivate", activation_id, "--root", str(tmp_path)])

    assert explained.exit_code == 0
    assert "Use explicit context packages" in activated.output
    assert deactivated.exit_code == 0
    assert "inactive" in deactivated.output