test: cover user-owned session security boundaries

harness/tests/test_postgres_session_repository.py

@@ -1,5 +1,7 @@
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime
+from threading import Barrier
 
 import pytest
 from sqlalchemy import create_engine, text
@@ -92,6 +94,74 @@ def test_admin_can_read_another_owners_session(database_url):
     assert admin.get(session_id).manifest.id == session_id
 
 
+def test_owner_admin_and_foreign_principals_have_distinct_mutation_boundaries(database_url):
+    session_id = str(uuid.uuid4())
+    alice = _repository(database_url, f"alice-{session_id}")
+    bob = _repository(database_url, f"bob-{session_id}")
+    admin = _repository(database_url, f"admin-{session_id}", is_admin=True)
+    manifest = _manifest(session_id)
+    alice.create(manifest)
+    alice.write_artifact(session_id, "question", "Alice's question")
+
+    assert [snapshot.manifest.id for snapshot in alice.list()] == [session_id]
+    assert bob.list() == []
+    for mutation in (
+        lambda: bob.write_artifact(session_id, "question", "Bob's overwrite"),
+        lambda: bob.append_decisions(
+            session_id, [DecisionInput(type="concept_clarified", subject="bob")]
+        ),
+        lambda: bob.save_manifest(manifest.model_copy(update={"name": "Bob's rename"})),
+        lambda: bob.delete(session_id),
+    ):
+        with pytest.raises(SessionError, match="Sessione non trovata"):
+            mutation()
+
+    # Admin access is an explicit, audited exception to normal owner isolation.
+    admin.write_artifact(session_id, "question", "Reviewed by admin")
+    admin.append_decisions(
+        session_id, [DecisionInput(type="concept_clarified", subject="admin-review")]
+    )
+
+    owner_snapshot = alice.get(session_id)
+    assert owner_snapshot.artifacts["question"] == "Reviewed by admin"
+    assert [record.subject for record in owner_snapshot.decisions] == ["admin-review"]
+    assert admin.get(session_id).manifest.id == session_id
+
+
+def test_concurrent_postgres_ledger_mutations_keep_every_decision_in_sequence(database_url, monkeypatch):
+    from tht.session.postgres_repository import PostgresSessionRepository
+
+    repository = _repository(database_url, "alice")
+    session_id = str(uuid.uuid4())
+    repository.create(_manifest(session_id))
+    barrier = Barrier(2)
+    original_lock = PostgresSessionRepository._lock_session
+
+    def enter_lock_together(connection, locked_session_id):
+        if locked_session_id == session_id:
+            barrier.wait(timeout=5)
+        original_lock(connection, locked_session_id)
+
+    monkeypatch.setattr(
+        PostgresSessionRepository,
+        "_lock_session",
+        staticmethod(enter_lock_together),
+    )
+
+    def append(subject: str):
+        return repository.append_decisions(
+            session_id, [DecisionInput(type="concept_clarified", subject=subject)]
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first, second = pool.map(append, ("first", "second"))
+
+    snapshot = repository.get(session_id)
+    assert {first[0].seq, second[0].seq} == {1, 2}
+    assert [record.seq for record in snapshot.decisions] == [1, 2]
+    assert {record.subject for record in snapshot.decisions} == {"first", "second"}
+
+
 def test_non_superuser_runtime_login_can_assume_the_restricted_runtime_role(database_url):
     admin = create_engine(database_url)
     runtime_url = admin.url.set(