Function test_checkpoint_and_rehydrate

tests/test_composition.py:247–282

(tmp_path)

@pytest.mark.asyncio
async def test_checkpoint_and_rehydrate(tmp_path) -> None:
    store = ArtifactStore(root=tmp_path / "artifacts")
    artifact = store.store_text(
        "database table users has columns id, email, created_at",
        kind="tool-result",
        role="tool",
        tool_name="read_schema",
    )
    messages = [{"role": "system", "content": "You are coding assistant."}]
    for i in range(18):
        messages.append({"role": "user", "content": f"turn {i} explain the migration status in detail"})
        messages.append({"role": "assistant", "content": f"assistant response {i} with a lot of detail " * 20})
    messages.append({"role": "user", "content": f"use {artifact.id} and artifact://{artifact.id} to answer"})

    result = await compose_messages_semantic(
        messages,
        store,
        CompositionPolicy(
            checkpoint_threshold_tokens=300,
            checkpoint_keep_last_messages=4,
            checkpoint_min_messages=6,
        ),
        semantic_compressor=FakeSemanticCompressor(),
        session_id="sess-2",
        request=object(),
    )

    assert result.checkpoint_created is True
    assert result.rehydrated_artifacts == 1
    assert result.semantic_calls >= 2
    assert any(
        isinstance(msg, dict) and msg.get("role") == "system" and "checkpoint artifact://" in str(msg.get("content"))
        for msg in result.messages
    )
    latest_user = next(msg for msg in reversed(result.messages) if msg.get("role") == "user")
    assert "Rehydrated artifact://" in latest_user["content"]


@pytest.mark.asyncio