"""Executable completeness contract for the tenant-partitioned Hermes fabric.""" from __future__ import annotations from concurrent.futures import ThreadPoolExecutor from dataclasses import replace from pathlib import Path import sqlite3 import pytest from aios_v2r1.hermes_fabric import HermesFabric from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError def authenticated_command(event_id: str, **changes: object) -> CanonicalMessage: value = CanonicalMessage( tenant_id="tenant-a", cell_id="cell-a", principal_id="user-a", principal_type="authenticated_user", origin_type="channel_user", channel="weixin", channel_account_id="account-a", peer_id="peer-a", conversation_id="conversation-a", session_id="session-a", task_id=None, event_id=event_id, message_id=f"message-{event_id}", reply_to_message_id=None, direction="inbound", command_capable=True, occurred_at="2026-07-13T00:00:00+00:00", received_at="2026-07-13T00:00:01+00:00", body="/run", quoted_body=None, adapter_name="weixin", adapter_version="v2") return replace(value, **changes) @pytest.fixture def fabric(tmp_path: Path) -> HermesFabric: return HermesFabric(tmp_path / "fabric.sqlite3") def test_duplicate_event_returns_sanitized_conflict(fabric: HermesFabric) -> None: command = authenticated_command("event-a") assert fabric.enqueue(message=command, profile_id="profile-a").event_id == "event-a" with pytest.raises(ConflictError) as raised: fabric.enqueue(message=command, profile_id="profile-a") assert not isinstance(raised.value, sqlite3.IntegrityError) assert "sqlite" not in str(raised.value).lower() assert raised.value.__cause__ is None assert raised.value.__context__ is None @pytest.mark.parametrize("changes", [ {"command_capable": False}, {"origin_type": "heartbeat"}, {"principal_type": "service"}, {"direction": "internal"}]) def test_non_claimable_provenance_cannot_enqueue( fabric: HermesFabric, changes: dict[str, object] ) -> None: with pytest.raises(ScopeError): fabric.enqueue(message=authenticated_command("event-denied", **changes), profile_id="profile-a") def test_claim_returns_running_and_matches_durable_state(fabric: HermesFabric) -> None: fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a") claimed = fabric.claim("tenant-a", "cell-a", "profile-a") assert claimed is not None assert claimed.status == "running" with sqlite3.connect(fabric.database) as db: assert db.execute("SELECT status FROM fabric_jobs WHERE job_id=?", (claimed.job_id,)).fetchone()[0] == "running" @pytest.mark.parametrize(("tenant_id", "cell_id", "profile_id"), [ ("tenant-b", "cell-a", "profile-a"), ("tenant-a", "cell-b", "profile-a"), ("tenant-a", "cell-a", "profile-b")]) def test_claim_scope_is_exact( fabric: HermesFabric, tenant_id: str, cell_id: str, profile_id: str ) -> None: fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a") assert fabric.claim(tenant_id, cell_id, profile_id) is None assert fabric.claim("tenant-a", "cell-a", "profile-a") is not None def test_only_one_concurrent_claimant_wins(fabric: HermesFabric) -> None: fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a") with ThreadPoolExecutor(max_workers=16) as executor: claimed = list(executor.map( lambda _: fabric.claim("tenant-a", "cell-a", "profile-a"), range(16))) winners = [job for job in claimed if job is not None] assert len(winners) == 1 assert winners[0].status == "running" @pytest.mark.parametrize("status", ["running", "completed", "failed", "cancelled"]) def test_nonqueued_states_are_not_claimable( fabric: HermesFabric, status: str ) -> None: job = fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a") with sqlite3.connect(fabric.database) as db: db.execute("UPDATE fabric_jobs SET status=? WHERE job_id=?", (status, job.job_id)) assert fabric.claim("tenant-a", "cell-a", "profile-a") is None def test_restart_preserves_queue_and_claims_once(tmp_path: Path) -> None: database = tmp_path / "fabric.sqlite3" HermesFabric(database).enqueue(message=authenticated_command("event-a"), profile_id="profile-a") reopened = HermesFabric(database) claimed = reopened.claim("tenant-a", "cell-a", "profile-a") assert claimed is not None and claimed.status == "running" assert reopened.claim("tenant-a", "cell-a", "profile-a") is None def test_concurrent_duplicate_enqueue_has_one_winner(fabric: HermesFabric) -> None: command = authenticated_command("event-a") def enqueue(_: int) -> str: try: fabric.enqueue(message=command, profile_id="profile-a") return "queued" except ConflictError: return "conflict" with ThreadPoolExecutor(max_workers=16) as executor: results = list(executor.map(enqueue, range(16))) assert results.count("queued") == 1 assert results.count("conflict") == 15 def test_sqlite_failures_are_sanitized(fabric: HermesFabric) -> None: Path(fabric.database).write_bytes(b"not-a-sqlite-database") with pytest.raises(ConflictError, match="fabric storage unavailable") as raised: fabric.health() assert not isinstance(raised.value, sqlite3.Error) assert "sqlite" not in str(raised.value).lower() assert raised.value.__cause__ is None assert raised.value.__context__ is None